This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new fe2aa4039 [FLINK-40739][maxcompute] Fix partition handling for sink
writes (#4562)
fe2aa4039 is described below
commit fe2aa4039e23eeac9adafea8c9a9a5dcc5346f2b
Author: Kunni <[email protected]>
AuthorDate: Tue Sep 29 09:47:31 2026 +0800
[FLINK-40739][maxcompute] Fix partition handling for sink writes (#4562)
---
.../pom.xml | 28 ++++
.../coordinator/SessionManageOperator.java | 17 ++-
.../maxcompute/utils/MaxComputeUtils.java | 47 +++---
.../maxcompute/utils/TypeConvertUtils.java | 15 +-
.../maxcompute/writer/BatchAppendWriter.java | 46 ++++--
.../maxcompute/writer/BatchUpsertWriter.java | 12 +-
.../connectors/maxcompute/EmulatorTestBase.java | 29 ++--
.../coordinator/SessionManageOperatorTest.java | 145 ++++++++++++++++++
.../utils/MaxComputeUtilsPartitionTest.java | 116 +++++++++++++++
.../utils/TypeConvertUtilsPartitionTest.java | 64 ++++++++
.../maxcompute/utils/TypeConvertUtilsTest.java | 27 ++++
.../maxcompute/writer/BatchAppendWriterTest.java | 163 +++++++++++++++++++++
.../maxcompute/writer/BatchUpsertWriterTest.java | 134 +++++++++++++++++
.../values/source/ValuesDataSourceHelper.java | 63 ++++++++
.../values/source/ValuesDataSourceOptions.java | 2 +
.../pipeline/tests/stage3/MaxComputeE2eITCase.java | 24 +++
16 files changed, 867 insertions(+), 65 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/pom.xml
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/pom.xml
index a6398cad6..6c8e44d2f 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/pom.xml
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/pom.xml
@@ -26,6 +26,10 @@ limitations under the License.
<artifactId>flink-cdc-pipeline-connector-maxcompute</artifactId>
<name>flink-cdc-pipeline-connector-maxcompute</name>
+ <properties>
+ <mockito.version>3.12.4</mockito.version>
+ </properties>
+
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
@@ -81,6 +85,30 @@ limitations under the License.
<version>1.19.8</version>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-inline</artifactId>
+ <version>${mockito.version}</version>
+ <type>jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>${mockito.version}</version>
+ <type>jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <version>${mockito.version}</version>
+ <type>jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/coordinator/SessionManageOperator.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/coordinator/SessionManageOperator.java
index 5c3485b69..89386300b 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/coordinator/SessionManageOperator.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/coordinator/SessionManageOperator.java
@@ -252,8 +252,7 @@ public class SessionManageOperator extends
AbstractStreamOperatorAdapter<Event>
}
}
- /** partition column is always after data column. */
- private String extractPartition(RecordData recordData, TableId tableId) {
+ String extractPartition(RecordData recordData, TableId tableId) {
Schema schema = schemaMaps.get(tableId);
int partitionKeyCount = schema.partitionKeys().size();
if (partitionKeyCount == 0) {
@@ -263,11 +262,17 @@ public class SessionManageOperator extends
AbstractStreamOperatorAdapter<Event>
List<RecordData.FieldGetter> fieldGetters =
fieldGetterMaps.get(tableId);
PartitionSpec partitionSpec = new PartitionSpec();
- for (int i = 0; i < partitionKeyCount; i++) {
- RecordData.FieldGetter fieldGetter =
- fieldGetters.get(columnCount - partitionKeyCount - 1 + i);
+ for (String partitionKey : schema.partitionKeys()) {
+ int partitionColumnIndex =
schema.getColumnNames().indexOf(partitionKey);
+ if (partitionColumnIndex < 0 || partitionColumnIndex >=
columnCount) {
+ throw new IllegalStateException(
+ String.format(
+ "Unable to find partition column \"%s\" in
schema %s",
+ partitionKey, schema));
+ }
+ RecordData.FieldGetter fieldGetter =
fieldGetters.get(partitionColumnIndex);
Object value = fieldGetter.getFieldOrNull(recordData);
- partitionSpec.set(schema.partitionKeys().get(i),
Objects.toString(value));
+ partitionSpec.set(partitionKey, Objects.toString(value));
}
return partitionSpec.toString(true, true);
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/utils/MaxComputeUtils.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/utils/MaxComputeUtils.java
index a6832dbe5..bf5bd3a12 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/utils/MaxComputeUtils.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/utils/MaxComputeUtils.java
@@ -206,33 +206,32 @@ public class MaxComputeUtils {
MaxComputeOptions options, String schema, String table, String
partitionName)
throws OdpsException {
Odps odps = getOdps(options);
+ Table partitionedTable;
if (options.isSupportSchema()) {
- if (StringUtils.isNullOrEmpty(schema)) {
- LOG.info(
- "create partition {} in {}.default.{}",
- partitionName,
- options.getProject(),
- table);
- odps.tables()
- .get(options.getProject(), "default", table)
- .createPartition(new PartitionSpec(partitionName),
true);
- } else {
- LOG.info(
- "create partition {} in {}.{}.{}",
- partitionName,
- options.getProject(),
- schema,
- table);
- odps.tables()
- .get(options.getProject(), schema, table)
- .createPartition(new PartitionSpec(partitionName),
true);
- }
+ String resolvedSchema = StringUtils.isNullOrEmpty(schema) ?
"default" : schema;
+ partitionedTable = odps.tables().get(options.getProject(),
resolvedSchema, table);
+ LOG.info(
+ "create partition {} in {}.{}.{} if absent",
+ partitionName,
+ options.getProject(),
+ resolvedSchema,
+ table);
} else {
- LOG.info("create partition {} in {}.{}", partitionName,
options.getProject(), table);
- odps.tables()
- .get(options.getProject(), table)
- .createPartition(new PartitionSpec(partitionName), true);
+ partitionedTable = odps.tables().get(options.getProject(), table);
+ LOG.info(
+ "create partition {} in {}.{} if absent",
+ partitionName,
+ options.getProject(),
+ table);
+ }
+ PartitionSpec partitionSpec = new PartitionSpec(partitionName);
+ if (partitionedTable.hasPartition(partitionSpec)) {
+ LOG.info("partition {} already exists, skip creating",
partitionName);
+ return;
}
+ // ifNotExists guards against concurrent partition creation between the
+ // existence check and the creation attempt
+ partitionedTable.createPartition(partitionSpec, true);
}
public static String getSchema(MaxComputeOptions options, TableId tableId)
{
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtils.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtils.java
index a76bd8ef2..22b07a6e9 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtils.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtils.java
@@ -180,11 +180,12 @@ public class TypeConvertUtils {
public static void toMaxComputeRecord(Schema flinkSchema, RecordData from,
ArrayRecord to) {
Preconditions.checkNotNull(from, "flink record data");
Preconditions.checkNotNull(to, "maxcompute arrayRecord");
- int partitionKeyCount = flinkSchema.partitionKeys().size();
+ Set<String> partitionKeys = new HashSet<>(flinkSchema.partitionKeys());
+ int dataColumnCount = flinkSchema.getColumnCount() -
partitionKeys.size();
List<RecordData.FieldGetter> fieldGetters =
createFieldGetters(flinkSchema);
- if (to.getColumnCount() != (fieldGetters.size() - partitionKeyCount)) {
+ if (to.getColumnCount() != dataColumnCount) {
throw new IllegalArgumentException(
"record data count not match, odps {"
+ Arrays.stream(to.getColumns())
@@ -195,11 +196,15 @@ public class TypeConvertUtils {
+ "vs flink {"
+ flinkSchema
+ "} count "
- + (fieldGetters.size() - partitionKeyCount));
+ + dataColumnCount);
}
- for (int i = 0; i < (fieldGetters.size() - partitionKeyCount); i++) {
+ int dataColumnIndex = 0;
+ for (int i = 0; i < fieldGetters.size(); i++) {
+ if
(partitionKeys.contains(flinkSchema.getColumns().get(i).getName())) {
+ continue;
+ }
Object value = fieldGetters.get(i).getFieldOrNull(from);
- to.set(i, value);
+ to.set(dataColumnIndex++, value);
}
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchAppendWriter.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchAppendWriter.java
index 1cf6be4c0..7212727d4 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchAppendWriter.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchAppendWriter.java
@@ -74,21 +74,39 @@ public class BatchAppendWriter implements MaxComputeWriter {
try {
if (StringUtils.isNullOrWhitespaceOnly(identifier.getSessionId()))
{
- this.uploadSession =
- tunnel.createUploadSession(
- identifier.getProject(),
- identifier.getSchema(),
- identifier.getTable(),
- new PartitionSpec(partitionSpec),
- false);
+ if (StringUtils.isNullOrWhitespaceOnly(partitionSpec)) {
+ this.uploadSession =
+ tunnel.createUploadSession(
+ identifier.getProject(),
+ identifier.getSchema(),
+ identifier.getTable(),
+ false);
+ } else {
+ this.uploadSession =
+ tunnel.createUploadSession(
+ identifier.getProject(),
+ identifier.getSchema(),
+ identifier.getTable(),
+ new PartitionSpec(partitionSpec),
+ false);
+ }
} else {
- this.uploadSession =
- tunnel.getUploadSession(
- identifier.getProject(),
- identifier.getSchema(),
- identifier.getTable(),
- new PartitionSpec(partitionSpec),
- sessionId);
+ if (StringUtils.isNullOrWhitespaceOnly(partitionSpec)) {
+ this.uploadSession =
+ tunnel.getUploadSession(
+ identifier.getProject(),
+ identifier.getSchema(),
+ identifier.getTable(),
+ sessionId);
+ } else {
+ this.uploadSession =
+ tunnel.getUploadSession(
+ identifier.getProject(),
+ identifier.getSchema(),
+ identifier.getTable(),
+ new PartitionSpec(partitionSpec),
+ sessionId);
+ }
}
this.recordWriter =
uploadSession.openBufferedWriter(
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchUpsertWriter.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchUpsertWriter.java
index 731e6139c..5fd949d04 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchUpsertWriter.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchUpsertWriter.java
@@ -18,6 +18,7 @@
package org.apache.flink.cdc.connectors.maxcompute.writer;
+import org.apache.flink.cdc.common.utils.StringUtils;
import org.apache.flink.cdc.connectors.maxcompute.common.SessionIdentifier;
import
org.apache.flink.cdc.connectors.maxcompute.common.UncheckedOdpsException;
import org.apache.flink.cdc.connectors.maxcompute.options.MaxComputeOptions;
@@ -69,14 +70,17 @@ public class BatchUpsertWriter implements MaxComputeWriter {
String partitionSpec = identifier.getPartitionName();
String sessionId = identifier.getSessionId();
try {
- this.upsertSession =
+ Builder builder =
((Builder)
tunnel.buildUpsertSession(
identifier.getProject(),
identifier.getTable()))
.setConfig(tunnel.getConfig())
- .setSchemaName(identifier.getSchema())
- .setPartitionSpec(partitionSpec)
- .setUpsertId(sessionId)
+ .setSchemaName(identifier.getSchema());
+ if (!StringUtils.isNullOrWhitespaceOnly(partitionSpec)) {
+ builder.setPartitionSpec(partitionSpec);
+ }
+ this.upsertSession =
+ builder.setUpsertId(sessionId)
.setConcurrentNum(writeOptions.getFlushConcurrent())
.build();
this.upsertStream =
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/EmulatorTestBase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/EmulatorTestBase.java
index 7d930b829..cadff1e19 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/EmulatorTestBase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/EmulatorTestBase.java
@@ -46,40 +46,45 @@ public class EmulatorTestBase {
public static final DockerImageName MAXCOMPUTE_IMAGE =
DockerImageName.parse("maxcompute/maxcompute-emulator:v0.0.7");
- public static final GenericContainer<?> MAXCOMPUTE_CONTAINER =
- new GenericContainer<>(MAXCOMPUTE_IMAGE)
- .withExposedPorts(8080)
- .waitingFor(
- Wait.forLogMessage(".*Started
MaxcomputeEmulatorApplication.*\\n", 1))
- .withLogConsumer(new Slf4jLogConsumer(LOG));
+ // a fresh container per test class: a stopped GenericContainer instance
cannot be restarted
+ public static GenericContainer<?> maxcomputeContainer;
@BeforeAll
static void createContainer() {
- Startables.deepStart(MAXCOMPUTE_CONTAINER).join();
+ maxcomputeContainer =
+ new GenericContainer<>(MAXCOMPUTE_IMAGE)
+ .withExposedPorts(8080)
+ .waitingFor(
+ Wait.forLogMessage(
+ ".*Started
MaxcomputeEmulatorApplication.*\\n", 1))
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ Startables.deepStart(maxcomputeContainer).join();
}
@AfterAll
static void destroyContainer() {
- MAXCOMPUTE_CONTAINER.stop();
+ maxcomputeContainer.stop();
}
public final MaxComputeOptions testOptions =
- MaxComputeOptions.builder("ak", "sk", getEndpoint(),
"mocked_mc").build();
+ MaxComputeOptions.builder("ak", "sk", getEndpoint(), "mocked_mc")
+ .withTunnelEndpoint(getEndpoint())
+ .build();
public final Odps odpsInstance = MaxComputeUtils.getOdps(testOptions);
private String getEndpoint() {
String ip;
- if (MAXCOMPUTE_CONTAINER.getHost().equals("localhost")) {
+ if (maxcomputeContainer.getHost().equals("localhost")) {
try {
ip = InetAddress.getLocalHost().getHostAddress();
} catch (UnknownHostException e) {
ip = "127.0.0.1";
}
} else {
- ip = MAXCOMPUTE_CONTAINER.getHost();
+ ip = maxcomputeContainer.getHost();
}
- String endpoint = "http://"; + ip + ":" +
MAXCOMPUTE_CONTAINER.getFirstMappedPort();
+ String endpoint = "http://"; + ip + ":" +
maxcomputeContainer.getFirstMappedPort();
sendPOST(endpoint + "/init", endpoint);
return endpoint;
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/coordinator/SessionManageOperatorTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/coordinator/SessionManageOperatorTest.java
new file mode 100644
index 000000000..95e9c24ce
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/coordinator/SessionManageOperatorTest.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.maxcompute.coordinator;
+
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.connectors.maxcompute.options.MaxComputeOptions;
+import org.apache.flink.cdc.connectors.maxcompute.utils.TypeConvertUtils;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+
+import com.aliyun.odps.PartitionSpec;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link SessionManageOperator}. */
+class SessionManageOperatorTest {
+
+ @AfterEach
+ void tearDown() {
+ SessionManageOperator.instance = null;
+ }
+
+ @Test
+ void testExtractPartitionByColumnName() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("pt", DataTypes.STRING())
+ .physicalColumn("name", DataTypes.STRING())
+ .physicalColumn("dt", DataTypes.STRING())
+ .partitionKey("pt", "dt")
+ .build();
+ BinaryRecordDataGenerator dataGenerator =
+ new BinaryRecordDataGenerator((RowType)
schema.toRowDataType());
+ BinaryRecordData record =
+ dataGenerator.generate(
+ new Object[] {
+ 1,
+ BinaryStringData.fromString("hangzhou"),
+ BinaryStringData.fromString("Alice"),
+ BinaryStringData.fromString("20260728")
+ });
+
+ SessionManageOperator operator =
+ new SessionManageOperator(mock(MaxComputeOptions.class), new
OperatorID());
+ operator.open();
+ TableId tableId = TableId.tableId("test_table");
+ Map<TableId, Schema> schemaMaps = getField(operator, "schemaMaps");
+ Map<TableId, List<RecordData.FieldGetter>> fieldGetterMaps =
+ getField(operator, "fieldGetterMaps");
+ schemaMaps.put(tableId, schema);
+ fieldGetterMaps.put(tableId,
TypeConvertUtils.createFieldGetters(schema));
+
+ PartitionSpec expected = new PartitionSpec();
+ expected.set("pt", "hangzhou");
+ expected.set("dt", "20260728");
+
+ assertThat(operator.extractPartition(record, tableId))
+ .isEqualTo(expected.toString(true, true));
+ }
+
+ @Test
+ void testExtractPartitionFromUnpartitionedTable() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .build();
+ BinaryRecordDataGenerator dataGenerator =
+ new BinaryRecordDataGenerator((RowType)
schema.toRowDataType());
+ BinaryRecordData record =
+ dataGenerator.generate(new Object[] {1,
BinaryStringData.fromString("Alice")});
+
+ SessionManageOperator operator =
+ new SessionManageOperator(mock(MaxComputeOptions.class), new
OperatorID());
+ operator.open();
+ TableId tableId = TableId.tableId("test_table");
+ Map<TableId, Schema> schemaMaps = getField(operator, "schemaMaps");
+ schemaMaps.put(tableId, schema);
+
+ assertThat(operator.extractPartition(record, tableId)).isNull();
+ }
+
+ @Test
+ void testExtractPartitionFailsForMissingPartitionColumn() throws Exception
{
+ Schema schema = mock(Schema.class);
+
when(schema.partitionKeys()).thenReturn(Collections.singletonList("missing_pt"));
+ when(schema.getColumnCount()).thenReturn(1);
+
when(schema.getColumnNames()).thenReturn(Collections.singletonList("id"));
+
+ SessionManageOperator operator =
+ new SessionManageOperator(mock(MaxComputeOptions.class), new
OperatorID());
+ operator.open();
+ TableId tableId = TableId.tableId("test_table");
+ Map<TableId, Schema> schemaMaps = getField(operator, "schemaMaps");
+ Map<TableId, List<RecordData.FieldGetter>> fieldGetterMaps =
+ getField(operator, "fieldGetterMaps");
+ schemaMaps.put(tableId, schema);
+ fieldGetterMaps.put(tableId, Collections.emptyList());
+
+ assertThatThrownBy(() ->
operator.extractPartition(mock(RecordData.class), tableId))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("missing_pt");
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T getField(Object target, String fieldName)
+ throws ReflectiveOperationException {
+ Field field = target.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return (T) field.get(target);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/MaxComputeUtilsPartitionTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/MaxComputeUtilsPartitionTest.java
new file mode 100644
index 000000000..7227f29ec
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/MaxComputeUtilsPartitionTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.maxcompute.utils;
+
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.connectors.maxcompute.EmulatorTestBase;
+import org.apache.flink.cdc.connectors.maxcompute.common.SessionIdentifier;
+import org.apache.flink.cdc.connectors.maxcompute.options.CompressAlgorithm;
+import
org.apache.flink.cdc.connectors.maxcompute.options.MaxComputeWriteOptions;
+import org.apache.flink.cdc.connectors.maxcompute.writer.MaxComputeWriter;
+
+import com.aliyun.odps.Instance;
+import com.aliyun.odps.data.ArrayRecord;
+import com.aliyun.odps.data.Record;
+import com.aliyun.odps.task.SQLTask;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Emulator-based tests for partition handling. Note that the Emulator only
supports uppercase input
+ * (However, MaxCompute can correctly distinguish between uppercase and
lowercase).
+ */
+class MaxComputeUtilsPartitionTest extends EmulatorTestBase {
+
+ private static final String TEST_TABLE = "PARTITION_WRITE_TEST_TABLE";
+
+ @AfterEach
+ void deleteTable() throws Exception {
+ odpsInstance.tables().delete(TEST_TABLE, true);
+ }
+
+ /**
+ * Verifies the partitioned-table write path used by the pipeline job:
create the partitioned
+ * table, create the partition if absent, write records through the
session writer, and read
+ * them back. The partition key is deliberately placed as the first column
to cover partition
+ * columns that are not the last column.
+ */
+ @Test
+ void testCreatePartitionIfAbsentAndWritePartitionedTable() throws
Exception {
+ SchemaEvolutionUtils.createTable(
+ testOptions,
+ TableId.tableId(TEST_TABLE),
+ Schema.newBuilder()
+ .physicalColumn("PT", DataTypes.STRING())
+ .physicalColumn("COL1", DataTypes.STRING())
+ .physicalColumn("COL2", DataTypes.STRING())
+ .partitionKey("PT")
+ .primaryKey("COL1")
+ .build());
+
+ // must not throw, and must be idempotent
+ MaxComputeUtils.createPartitionIfAbsent(testOptions, null, TEST_TABLE,
"PT='2024-01'");
+ MaxComputeUtils.createPartitionIfAbsent(testOptions, null, TEST_TABLE,
"PT='2024-01'");
+
+ writeRow("2024-01", "1", "a");
+ writeRow("2024-01", "2", "b");
+ writeRow("2024-02", "3", "c");
+
+ Instance instance =
+ SQLTask.run(
+ odpsInstance,
+ "select PT, COL1, COL2 from " + TEST_TABLE + " order
by COL1;");
+ instance.waitForSuccess();
+ List<Record> result = SQLTask.getResult(instance);
+ assertThat(result).hasSize(3);
+ assertThat(result.get(0).get(0)).isEqualTo("2024-01");
+ assertThat(result.get(0).get(1)).isEqualTo("1");
+ assertThat(result.get(0).get(2)).isEqualTo("a");
+ assertThat(result.get(1).get(0)).isEqualTo("2024-01");
+ assertThat(result.get(1).get(1)).isEqualTo("2");
+ assertThat(result.get(1).get(2)).isEqualTo("b");
+ assertThat(result.get(2).get(0)).isEqualTo("2024-02");
+ assertThat(result.get(2).get(1)).isEqualTo("3");
+ assertThat(result.get(2).get(2)).isEqualTo("c");
+ }
+
+ private void writeRow(String pt, String col1, String col2) throws
Exception {
+ // session id is null on first creation, mirroring
SessionManageCoordinator#createWriter
+ SessionIdentifier identifier =
+ SessionIdentifier.of(testOptions.getProject(), null,
TEST_TABLE, "PT='" + pt + "'");
+ MaxComputeWriteOptions writeOptions =
+ MaxComputeWriteOptions.builder()
+ .withCompressAlgorithm(CompressAlgorithm.RAW)
+ .build();
+ MaxComputeWriter writer =
+ MaxComputeWriter.batchWriter(testOptions, writeOptions,
identifier);
+ ArrayRecord record = writer.newElement();
+ record.setString(0, col1);
+ record.setString(1, col2);
+ writer.write(record);
+ writer.flush();
+ writer.commit();
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtilsPartitionTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtilsPartitionTest.java
new file mode 100644
index 000000000..4cf56346c
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtilsPartitionTest.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.maxcompute.utils;
+
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+
+import com.aliyun.odps.Column;
+import com.aliyun.odps.TableSchema;
+import com.aliyun.odps.data.ArrayRecord;
+import com.aliyun.odps.type.TypeInfoFactory;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Partition-related tests for {@link TypeConvertUtils}. */
+class TypeConvertUtilsPartitionTest {
+
+ @Test
+ void testRecordConversionRejectsMismatchedDataColumnCount() {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("pt", DataTypes.STRING())
+ .partitionKey("pt")
+ .build();
+ BinaryRecordDataGenerator dataGenerator =
+ new BinaryRecordDataGenerator((RowType)
schema.toRowDataType());
+ BinaryRecordData record =
+ dataGenerator.generate(
+ new Object[] {7,
BinaryStringData.fromString("partition-value")});
+
+ TableSchema mismatchedSchema = new TableSchema();
+ mismatchedSchema.addColumn(new Column("id", TypeInfoFactory.INT));
+ mismatchedSchema.addColumn(new Column("extra",
TypeInfoFactory.STRING));
+ ArrayRecord arrayRecord = new ArrayRecord(mismatchedSchema);
+
+ assertThatThrownBy(() -> TypeConvertUtils.toMaxComputeRecord(schema,
record, arrayRecord))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("record data count not match")
+ .hasMessageContaining("count 2")
+ .hasMessageContaining("count 1");
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtilsTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtilsTest.java
index 26edede0d..189149851 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtilsTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtilsTest.java
@@ -185,6 +185,33 @@ class TypeConvertUtilsTest {
assertThat(arrayRecord).hasToString(expect);
}
+ @Test
+ void testRecordConvertWithInterleavedPartitionColumn() {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("pt", DataTypes.STRING())
+ .physicalColumn("name", DataTypes.STRING())
+ .partitionKey("pt")
+ .build();
+ BinaryRecordDataGenerator dataGenerator =
+ new BinaryRecordDataGenerator((RowType)
schema.toRowDataType());
+ BinaryRecordData record =
+ dataGenerator.generate(
+ new Object[] {
+ 7,
+ BinaryStringData.fromString("20260728"),
+ BinaryStringData.fromString("Alice")
+ });
+
+ ArrayRecord arrayRecord = new
ArrayRecord(TypeConvertUtils.toMaxCompute(schema));
+ TypeConvertUtils.toMaxComputeRecord(schema, record, arrayRecord);
+
+ assertThat(arrayRecord.getColumnCount()).isEqualTo(2);
+ assertThat(arrayRecord.get(0)).isEqualTo(7);
+ assertThat(arrayRecord.get(1)).isEqualTo("Alice");
+ }
+
@Test
void testColumnCommentConversion() {
// Test column with comment
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchAppendWriterTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchAppendWriterTest.java
new file mode 100644
index 000000000..8a754d7fc
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchAppendWriterTest.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.maxcompute.writer;
+
+import org.apache.flink.cdc.connectors.maxcompute.common.SessionIdentifier;
+import org.apache.flink.cdc.connectors.maxcompute.options.MaxComputeOptions;
+import
org.apache.flink.cdc.connectors.maxcompute.options.MaxComputeWriteOptions;
+import org.apache.flink.cdc.connectors.maxcompute.utils.MaxComputeUtils;
+
+import com.aliyun.odps.PartitionSpec;
+import com.aliyun.odps.data.RecordWriter;
+import com.aliyun.odps.tunnel.TableTunnel;
+import com.aliyun.odps.tunnel.io.CompressOption;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link BatchAppendWriter}. */
+class BatchAppendWriterTest {
+
+ private static final String PROJECT = "project";
+ private static final String SCHEMA = "schema";
+ private static final String TABLE = "table";
+ private static final String PARTITION = "pt='partition-value'";
+ private static final String SESSION_ID = "session-id";
+
+ private MaxComputeOptions options;
+ private MaxComputeWriteOptions writeOptions;
+ private TableTunnel tunnel;
+ private TableTunnel.UploadSession uploadSession;
+ private MockedStatic<MaxComputeUtils> mockedMaxComputeUtils;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ options = mock(MaxComputeOptions.class);
+ writeOptions = MaxComputeWriteOptions.builder().build();
+ tunnel = mock(TableTunnel.class);
+ uploadSession = mock(TableTunnel.UploadSession.class);
+ RecordWriter recordWriter = mock(RecordWriter.class);
+ CompressOption compressOption = mock(CompressOption.class);
+
+ mockedMaxComputeUtils = Mockito.mockStatic(MaxComputeUtils.class);
+ mockedMaxComputeUtils
+ .when(() -> MaxComputeUtils.getTunnel(options, writeOptions))
+ .thenReturn(tunnel);
+ mockedMaxComputeUtils
+ .when(() ->
MaxComputeUtils.compressOptionOf(writeOptions.getCompressAlgorithm()))
+ .thenReturn(compressOption);
+
when(uploadSession.openBufferedWriter(compressOption)).thenReturn(recordWriter);
+ }
+
+ @AfterEach
+ void tearDown() {
+ mockedMaxComputeUtils.close();
+ }
+
+ @Test
+ void testCreateSessionWithoutPartitionSpec() throws Exception {
+ when(tunnel.createUploadSession(PROJECT, SCHEMA, TABLE,
false)).thenReturn(uploadSession);
+
+ new BatchAppendWriter(
+ options, writeOptions, SessionIdentifier.of(PROJECT, SCHEMA,
TABLE, null));
+
+ verify(tunnel).createUploadSession(PROJECT, SCHEMA, TABLE, false);
+ verify(tunnel, never())
+ .createUploadSession(
+ eq(PROJECT), eq(SCHEMA), eq(TABLE),
any(PartitionSpec.class), eq(false));
+ }
+
+ @Test
+ void testCreateSessionWithPartitionSpec() throws Exception {
+ when(tunnel.createUploadSession(
+ eq(PROJECT), eq(SCHEMA), eq(TABLE),
any(PartitionSpec.class), eq(false)))
+ .thenReturn(uploadSession);
+
+ new BatchAppendWriter(
+ options, writeOptions, SessionIdentifier.of(PROJECT, SCHEMA,
TABLE, PARTITION));
+
+ ArgumentCaptor<PartitionSpec> partitionCaptor =
+ ArgumentCaptor.forClass(PartitionSpec.class);
+ verify(tunnel)
+ .createUploadSession(
+ eq(PROJECT), eq(SCHEMA), eq(TABLE),
partitionCaptor.capture(), eq(false));
+ assertThat(partitionCaptor.getValue().toString(true, true))
+ .isEqualTo(new PartitionSpec(PARTITION).toString(true, true));
+ verify(tunnel, never()).createUploadSession(PROJECT, SCHEMA, TABLE,
false);
+ }
+
+ @Test
+ void testReloadSessionWithoutPartitionSpec() throws Exception {
+ when(tunnel.getUploadSession(PROJECT, SCHEMA, TABLE,
SESSION_ID)).thenReturn(uploadSession);
+
+ new BatchAppendWriter(
+ options,
+ writeOptions,
+ SessionIdentifier.of(PROJECT, SCHEMA, TABLE, " ",
SESSION_ID));
+
+ verify(tunnel).getUploadSession(PROJECT, SCHEMA, TABLE, SESSION_ID);
+ verify(tunnel, never())
+ .getUploadSession(
+ eq(PROJECT),
+ eq(SCHEMA),
+ eq(TABLE),
+ any(PartitionSpec.class),
+ eq(SESSION_ID));
+ }
+
+ @Test
+ void testReloadSessionWithPartitionSpec() throws Exception {
+ when(tunnel.getUploadSession(
+ eq(PROJECT),
+ eq(SCHEMA),
+ eq(TABLE),
+ any(PartitionSpec.class),
+ eq(SESSION_ID)))
+ .thenReturn(uploadSession);
+
+ new BatchAppendWriter(
+ options,
+ writeOptions,
+ SessionIdentifier.of(PROJECT, SCHEMA, TABLE, PARTITION,
SESSION_ID));
+
+ ArgumentCaptor<PartitionSpec> partitionCaptor =
+ ArgumentCaptor.forClass(PartitionSpec.class);
+ verify(tunnel)
+ .getUploadSession(
+ eq(PROJECT),
+ eq(SCHEMA),
+ eq(TABLE),
+ partitionCaptor.capture(),
+ eq(SESSION_ID));
+ assertThat(partitionCaptor.getValue().toString(true, true))
+ .isEqualTo(new PartitionSpec(PARTITION).toString(true, true));
+ verify(tunnel, never()).getUploadSession(PROJECT, SCHEMA, TABLE,
SESSION_ID);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchUpsertWriterTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchUpsertWriterTest.java
new file mode 100644
index 000000000..e2a4acc6e
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/writer/BatchUpsertWriterTest.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.maxcompute.writer;
+
+import org.apache.flink.cdc.connectors.maxcompute.common.SessionIdentifier;
+import org.apache.flink.cdc.connectors.maxcompute.options.MaxComputeOptions;
+import
org.apache.flink.cdc.connectors.maxcompute.options.MaxComputeWriteOptions;
+import org.apache.flink.cdc.connectors.maxcompute.utils.MaxComputeUtils;
+
+import com.aliyun.odps.PartitionSpec;
+import com.aliyun.odps.tunnel.Configuration;
+import com.aliyun.odps.tunnel.TableTunnel;
+import com.aliyun.odps.tunnel.impl.UpsertSessionImpl;
+import com.aliyun.odps.tunnel.impl.UpsertSessionImpl.Builder;
+import com.aliyun.odps.tunnel.io.CompressOption;
+import com.aliyun.odps.tunnel.streams.UpsertStream;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.nullable;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link BatchUpsertWriter}. */
+class BatchUpsertWriterTest {
+
+ private static final String PROJECT = "project";
+ private static final String SCHEMA = "schema";
+ private static final String TABLE = "table";
+ private static final String PARTITION = "pt='partition-value'";
+ private static final String SESSION_ID = "session-id";
+
+ private MaxComputeOptions options;
+ private MaxComputeWriteOptions writeOptions;
+ private TableTunnel tunnel;
+ private Builder sessionBuilder;
+ private MockedStatic<MaxComputeUtils> mockedMaxComputeUtils;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ options = mock(MaxComputeOptions.class);
+ writeOptions = MaxComputeWriteOptions.builder().build();
+ tunnel = mock(TableTunnel.class);
+ sessionBuilder = mock(Builder.class);
+ UpsertSessionImpl upsertSession = mock(UpsertSessionImpl.class);
+ UpsertStream.Builder streamBuilder = mock(UpsertStream.Builder.class);
+ UpsertStream upsertStream = mock(UpsertStream.class);
+ Configuration configuration = mock(Configuration.class);
+ CompressOption compressOption = mock(CompressOption.class);
+
+ mockedMaxComputeUtils = Mockito.mockStatic(MaxComputeUtils.class);
+ mockedMaxComputeUtils
+ .when(() -> MaxComputeUtils.getTunnel(options, writeOptions))
+ .thenReturn(tunnel);
+ mockedMaxComputeUtils
+ .when(() ->
MaxComputeUtils.compressOptionOf(writeOptions.getCompressAlgorithm()))
+ .thenReturn(compressOption);
+
+ when(tunnel.buildUpsertSession(PROJECT,
TABLE)).thenReturn(sessionBuilder);
+ when(tunnel.getConfig()).thenReturn(configuration);
+
when(sessionBuilder.setConfig(configuration)).thenReturn(sessionBuilder);
+ when(sessionBuilder.setSchemaName(SCHEMA)).thenReturn(sessionBuilder);
+
when(sessionBuilder.setPartitionSpec(anyString())).thenReturn(sessionBuilder);
+
when(sessionBuilder.setUpsertId(SESSION_ID)).thenReturn(sessionBuilder);
+
when(sessionBuilder.setConcurrentNum(writeOptions.getFlushConcurrent()))
+ .thenReturn(sessionBuilder);
+ when(sessionBuilder.build()).thenReturn(upsertSession);
+
+ when(upsertSession.buildUpsertStream()).thenReturn(streamBuilder);
+
when(streamBuilder.setListener(any(UpsertStream.Listener.class))).thenReturn(streamBuilder);
+ when(streamBuilder.setMaxBufferSize(writeOptions.getMaxBufferSize()))
+ .thenReturn(streamBuilder);
+ when(streamBuilder.setSlotBufferSize(writeOptions.getSlotBufferSize()))
+ .thenReturn(streamBuilder);
+
when(streamBuilder.setCompressOption(compressOption)).thenReturn(streamBuilder);
+ when(streamBuilder.build()).thenReturn(upsertStream);
+ }
+
+ @AfterEach
+ void tearDown() {
+ mockedMaxComputeUtils.close();
+ }
+
+ @Test
+ void testDoesNotSetNullOrBlankPartitionSpec() throws Exception {
+ new BatchUpsertWriter(
+ options,
+ writeOptions,
+ SessionIdentifier.of(PROJECT, SCHEMA, TABLE, null,
SESSION_ID));
+ new BatchUpsertWriter(
+ options,
+ writeOptions,
+ SessionIdentifier.of(PROJECT, SCHEMA, TABLE, " ",
SESSION_ID));
+
+ verify(sessionBuilder,
never()).setPartitionSpec(nullable(String.class));
+ verify(sessionBuilder,
never()).setPartitionSpec(any(PartitionSpec.class));
+ verify(sessionBuilder, times(2)).build();
+ }
+
+ @Test
+ void testSetsNonBlankPartitionSpec() throws Exception {
+ new BatchUpsertWriter(
+ options,
+ writeOptions,
+ SessionIdentifier.of(PROJECT, SCHEMA, TABLE, PARTITION,
SESSION_ID));
+
+ verify(sessionBuilder).setPartitionSpec(PARTITION);
+ verify(sessionBuilder,
never()).setPartitionSpec(any(PartitionSpec.class));
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceHelper.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceHelper.java
index 2d5ffd5d7..8b777ce9f 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceHelper.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceHelper.java
@@ -52,6 +52,7 @@ public class ValuesDataSourceHelper {
public enum EventSetId {
SINGLE_SPLIT_SINGLE_TABLE,
SINGLE_SPLIT_SINGLE_TABLE_WITH_DEFAULT_VALUE,
+ SINGLE_SPLIT_SINGLE_PARTITION_TABLE,
SINGLE_SPLIT_MULTI_TABLES,
MULTI_SPLITS_SINGLE_TABLE,
CUSTOM_SOURCE_EVENTS,
@@ -122,6 +123,11 @@ public class ValuesDataSourceHelper {
sourceEvents = singleSplitSingleTableWithDefaultValue();
break;
}
+ case SINGLE_SPLIT_SINGLE_PARTITION_TABLE:
+ {
+ sourceEvents = singleSplitSinglePartitionTable();
+ break;
+ }
case SINGLE_SPLIT_MULTI_TABLES:
{
sourceEvents = singleSplitMultiTables();
@@ -342,6 +348,63 @@ public class ValuesDataSourceHelper {
return eventOfSplits;
}
+ /**
+ * A single split with a single partitioned table. The partition column is
intentionally placed
+ * before data columns to verify partition handling when partition key is
not at the end of the
+ * schema.
+ */
+ public static List<List<Event>> singleSplitSinglePartitionTable() {
+ List<List<Event>> eventOfSplits = new ArrayList<>();
+ List<Event> split1 = new ArrayList<>();
+
+ // create table with partition column before data columns
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("pt", DataTypes.STRING())
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .partitionKey("pt")
+ .primaryKey("col1")
+ .build();
+ split1.add(new CreateTableEvent(TABLE_1, schema));
+
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(
+ RowType.of(DataTypes.STRING(), DataTypes.STRING(),
DataTypes.STRING()));
+
+ // insert into different partitions
+ split1.add(
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2024-01"),
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("a")
+ })));
+ split1.add(
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2024-01"),
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("b")
+ })));
+ split1.add(
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2024-02"),
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("c")
+ })));
+
+ eventOfSplits.add(split1);
+ return eventOfSplits;
+ }
+
private static AddColumnEvent addColumnWithDefaultValue(TableId tableId) {
AddColumnEvent.ColumnWithPosition columnWithPositionWithDefault =
new AddColumnEvent.ColumnWithPosition(
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceOptions.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceOptions.java
index 42009a48a..3df6c8d33 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceOptions.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceOptions.java
@@ -42,6 +42,8 @@ public class ValuesDataSourceOptions {
"SINGLE_SPLIT_SINGLE_TABLE: Default and predetermined case. Creating schema
changes of single table and put them into one split."),
text(
"SINGLE_SPLIT_SINGLE_TABLE_WITH_DEFAULT_VALUE: A predetermined case. Creating
schema changes of single table (some columns have default value) and put them
into one split."),
+ text(
+
"SINGLE_SPLIT_SINGLE_PARTITION_TABLE: A predetermined case. Creating schema
changes of single partitioned table whose partition key is not the last column
and put them into one split."),
text(
"SINGLE_SPLIT_MULTI_TABLES: A predetermined case. Creating schema changes of
multiple tables and put them into one split."),
text(
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/MaxComputeE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/MaxComputeE2eITCase.java
index 3d31ba015..901dded37 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/MaxComputeE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/MaxComputeE2eITCase.java
@@ -111,6 +111,30 @@ class MaxComputeE2eITCase extends PipelineTestEnvironment {
assertThat(result.get(1).get(1)).isEqualTo("NULL");
}
+ @Test
+ void testSingleSplitSinglePartitionTable() throws Exception {
+ startTest("SINGLE_SPLIT_SINGLE_PARTITION_TABLE");
+ Instance instance =
+ SQLTask.run(
+ MaxComputeUtils.getOdps(testOptions),
+ "select pt, col1, col2 from table1 order by col1;");
+ instance.waitForSuccess();
+ List<Record> result = SQLTask.getResult(instance);
+ LOG.info("{}", result);
+ assertThat(result).hasSize(3);
+ // verify partition column is correctly extracted and data columns are
+ // mapped correctly when partition key is not the last column
+ assertThat(result.get(0).get(0)).isEqualTo("2024-01");
+ assertThat(result.get(0).get(1)).isEqualTo("1");
+ assertThat(result.get(0).get(2)).isEqualTo("a");
+ assertThat(result.get(1).get(0)).isEqualTo("2024-01");
+ assertThat(result.get(1).get(1)).isEqualTo("2");
+ assertThat(result.get(1).get(2)).isEqualTo("b");
+ assertThat(result.get(2).get(0)).isEqualTo("2024-02");
+ assertThat(result.get(2).get(1)).isEqualTo("3");
+ assertThat(result.get(2).get(2)).isEqualTo("c");
+ }
+
private void startTest(String testSet) throws Exception {
sendPOST(getEndpoint() + "/init", getEndpoint());
