This is an automated email from the ASF dual-hosted git repository.
dybyte pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 17eb09126e [BUG][Connector-V2][MaxCompute] Fix Maxcompute source split
stategy (#10845)
17eb09126e is described below
commit 17eb09126e73a1e744626135850ecf32722a97f7
Author: zhiliang-wu <[email protected]>
AuthorDate: Thu May 7 11:14:09 2026 +0200
[BUG][Connector-V2][MaxCompute] Fix Maxcompute source split stategy (#10845)
Co-authored-by: WU Zhiliang (External) <[email protected]>
---
.../maxcompute/source/MaxcomputeSourceSplit.java | 2 +-
.../source/MaxcomputeSourceSplitEnumerator.java | 3 +-
.../source/MaxcomputeSourceSplitTest.java | 74 ++++++++++++++++++++++
3 files changed, 76 insertions(+), 3 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplit.java
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplit.java
index 13d1225608..2a50fc68a6 100644
---
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplit.java
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplit.java
@@ -43,6 +43,6 @@ public class MaxcomputeSourceSplit implements SourceSplit {
@Override
public String splitId() {
- return tablePath.toString() + "_" + index;
+ return tablePath.toString() + "_" + index + "_" + rowStart;
}
}
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplitEnumerator.java
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplitEnumerator.java
index e6685f0f85..0f582bdd37 100644
---
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplitEnumerator.java
@@ -124,7 +124,6 @@ public class MaxcomputeSourceSplitEnumerator
if (sourceTableInfo.getSplitRow() != null &&
sourceTableInfo.getSplitRow() > 0) {
splitRow = sourceTableInfo.getSplitRow();
}
- int splitIndex = 0;
for (int i = 0; i < numReaders; i++) {
int readerStart = i * splitRowNum;
int readerEnd = (int) Math.min((i + 1) * splitRowNum,
recordCount);
@@ -134,7 +133,7 @@ public class MaxcomputeSourceSplitEnumerator
num,
Math.min(splitRow, readerEnd - num),
sourceTableInfo.getCatalogTable().getTablePath(),
- splitIndex));
+ i));
}
}
assignedSplits.forEach(splits::remove);
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/test/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplitTest.java
b/seatunnel-connectors-v2/connector-maxcompute/src/test/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplitTest.java
new file mode 100644
index 0000000000..d6dfcf6eba
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/test/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplitTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.maxcompute.source;
+
+import org.apache.seatunnel.api.table.catalog.TablePath;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class MaxcomputeSourceSplitTest {
+
+ @Test
+ public void testSplitIdUniquenessAndIndexDistribution() {
+ int numReaders = 5;
+ long recordCount = 105000;
+ int splitRow = 10000;
+
+ int splitRowNum = (int) Math.ceil((double) recordCount / numReaders);
// 21000
+ TablePath tablePath = TablePath.of("test_db", "test_schema",
"test_table");
+
+ Set<String> splitIds = new HashSet<>();
+ int[] indexCounts = new int[numReaders];
+ int totalSplits = 0;
+
+ // Simulate the inner logic of
MaxcomputeSourceSplitEnumerator.discoverySplits
+ for (int i = 0; i < numReaders; i++) {
+ int readerStart = i * splitRowNum;
+ int readerEnd = (int) Math.min((i + 1) * splitRowNum, recordCount);
+ for (int num = readerStart; num < readerEnd; num += splitRow) {
+ MaxcomputeSourceSplit split =
+ new MaxcomputeSourceSplit(
+ num, Math.min(splitRow, readerEnd - num),
tablePath, i);
+
+ // Verify splitId uniqueness
+ Assertions.assertTrue(
+ splitIds.add(split.splitId()),
+ "Duplicate splitId found: " + split.splitId());
+
+ // Verify index assigned to the split matches the reader index
+ Assertions.assertEquals(i, split.getIndex());
+
+ indexCounts[i]++;
+ totalSplits++;
+ }
+ }
+
+ Assertions.assertTrue(totalSplits == 15);
+
+ // Verify index distribution:
+ // Record count 105000 / 5 readers = 21000 splits per reader.
+ // Split row is 10000. Loop steps: 0, 10000, 20000. So 3 splits per
reader.
+ for (int i = 0; i < numReaders; i++) {
+ Assertions.assertEquals(3, indexCounts[i], "Reader " + i + " split
count mismatch!");
+ }
+ }
+}
