This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12463-6b92e9ed7826538c280765c6bed5bc0e2c0c6f88 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 82413637863aed8d245443bf3698f692373eff7a Author: sunoj <[email protected]> AuthorDate: Sun Oct 11 09:42:47 2026 +0000 [Fix][Connector-V2] Fix MySQL CDC snapshot split error with non-consecutive primary key IDs (#12463) --- .../splitter/AbstractJdbcSourceChunkSplitter.java | 62 +++- .../AbstractJdbcSourceChunkSplitterTest.java | 334 +++++++++++++++++++++ 2 files changed, 392 insertions(+), 4 deletions(-) diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java index c5cc23048b..6d1667674f 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java @@ -132,7 +132,7 @@ public abstract class AbstractJdbcSourceChunkSplitter implements JdbcSourceChunk } } - private List<ChunkRange> splitTableIntoChunks( + protected List<ChunkRange> splitTableIntoChunks( JdbcConnection jdbc, TableId tableId, Column splitColumn) throws Exception { final String splitColumnName = splitColumn.name(); final Object[] minMax = queryMinMax(jdbc, tableId, splitColumn); @@ -173,12 +173,58 @@ public abstract class AbstractJdbcSourceChunkSplitter implements JdbcSourceChunk && doubleCompare(distributionFactor, distributionFactorUpper) <= 0; if (dataIsEvenlyDistributed) { + // When distributionFactor > 1.0, primary key IDs are not consecutive + // (there are gaps in the ID space). Arithmetic-based evenly-sized chunk + // splitting assumes gaps are uniformly distributed, which fails when + // gaps are clustered (e.g., large ranges of deleted IDs). In this case, + // prefer sampling-based splitting which uses actual data points for + // chunk boundaries, producing balanced splits regardless of gap + // distribution. + int shardCount = (int) Math.ceil((double) approximateRowCnt / chunkSize); + if (doubleCompare(distributionFactor, 1.0d) > 0 + && sampleShardingAllow + && shardCount > 1) { + int inverseSamplingRate = sourceConfig.getInverseSamplingRate(); + if (inverseSamplingRate > chunkSize) { + log.warn( + "The inverseSamplingRate is {}, which is greater than chunkSize {}, so we set inverseSamplingRate to chunkSize", + inverseSamplingRate, + chunkSize); + inverseSamplingRate = chunkSize; + } + log.info( + "Distribution factor {} > 1.0 for table {}, using sampling-based " + + "splitting for better chunk balance with non-consecutive IDs, " + + "the sampling rate is {}", + distributionFactor, + tableId, + inverseSamplingRate); + Object[] sample = + sampleDataFromColumn(jdbc, tableId, splitColumn, inverseSamplingRate); + log.info( + "Sample data from table {} end, the sample size is {}", + tableId, + sample.length); + if (sample.length >= shardCount) { + return efficientShardingThroughSampling( + tableId, sample, approximateRowCnt, shardCount); + } + log.warn( + "Sampling returned insufficient data for table {} (sample size {}, required {}). " + + "This indicates highly sparse keys defeating the sampler. " + + "Falling back to splitUnevenlySizedChunks to ensure accurate chunk sizes.", + tableId, + sample.length, + shardCount); + return splitUnevenlySizedChunks( + jdbc, tableId, splitColumn, min, max, chunkSize); + } // the minimum dynamic chunk size is at least 1 final int dynamicChunkSize = Math.max((int) (distributionFactor * chunkSize), 1); return splitEvenlySizedChunks( tableId, min, max, approximateRowCnt, chunkSize, dynamicChunkSize); } else { - int shardCount = (int) (approximateRowCnt / chunkSize); + int shardCount = (int) Math.ceil((double) approximateRowCnt / chunkSize); int inverseSamplingRate = sourceConfig.getInverseSamplingRate(); if (sampleShardingAllow && sampleShardingThreshold < shardCount) { // It is necessary to ensure that the number of data rows sampled by the @@ -204,8 +250,16 @@ public abstract class AbstractJdbcSourceChunkSplitter implements JdbcSourceChunk "Sample data from table {} end, the sample size is {}", tableId, sample.length); - return efficientShardingThroughSampling( - tableId, sample, approximateRowCnt, shardCount); + if (sample.length >= shardCount) { + return efficientShardingThroughSampling( + tableId, sample, approximateRowCnt, shardCount); + } + log.warn( + "Sampling returned insufficient data for table {} (sample size {}, required {}). " + + "Falling back to splitUnevenlySizedChunks.", + tableId, + sample.length, + shardCount); } return splitUnevenlySizedChunks(jdbc, tableId, splitColumn, min, max, chunkSize); } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitterTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitterTest.java index 66f408609b..479ebd27af 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitterTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitterTest.java @@ -34,10 +34,12 @@ import io.debezium.relational.TableId; import java.sql.SQLException; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.List; import java.util.Properties; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -155,6 +157,338 @@ public class AbstractJdbcSourceChunkSplitterTest { ChunkRange.of(6, null))); } + /** + * Verifies the decision logic in splitTableIntoChunks for non-consecutive IDs. When + * distributionFactor > 1.0 and sampling is enabled, it should use sampling. If sampling fails + * (returns insufficient data), it should fallback to splitUnevenlySizedChunks. + */ + @Test + public void testSplitTableIntoChunksDecisionLogic() throws Exception { + TestJdbcSourceConfig config = + new TestJdbcSourceConfig() { + @Override + public double getDistributionFactorUpper() { + return 100.0; + } + + @Override + public double getDistributionFactorLower() { + return 0.05; + } + + @Override + public int getSplitSize() { + return 100; + } + + @Override + public boolean isSampleShardingAllow() { + return true; + } + + @Override + public int getSampleShardingThreshold() { + return 1000; // Large default threshold that should be ignored for gap fixes + } + }; + + TableId tableId = new TableId("test", "test", "test"); + Column splitColumn = Column.editor().name("id").create(); + + final boolean[] samplingCalled = {false}; + final boolean[] unevenlyCalled = {false}; + final boolean[] evenlyCalled = {false}; + + AbstractJdbcSourceChunkSplitter splitterSuccess = + new ConfiguredUtJdbcSourceChunkSplitter(config) { + @Override + public Object[] queryMinMax( + JdbcConnection jdbc, TableId tableId, Column columnName) { + return new Object[] {1, 50000}; + } + + @Override + public boolean isEvenlySplitColumn(Column splitColumn) { + return true; + } + + @Override + public Long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId) { + return 1000L; + } + + @Override + public Object[] sampleDataFromColumn( + JdbcConnection jdbc, + TableId tableId, + String columnName, + int samplingRate) { + samplingCalled[0] = true; + return new Object[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}; + } + + @Override + protected List<ChunkRange> splitEvenlySizedChunks( + TableId tableId, + Object min, + Object max, + long approximateRowCnt, + int chunkSize, + int dynamicChunkSize) { + evenlyCalled[0] = true; + return Collections.emptyList(); + } + + @Override + protected List<ChunkRange> splitUnevenlySizedChunks( + JdbcConnection jdbc, + TableId tableId, + Column splitColumn, + Object min, + Object max, + int chunkSize) { + unevenlyCalled[0] = true; + return Collections.emptyList(); + } + + @Override + protected List<ChunkRange> efficientShardingThroughSampling( + TableId tableId, + Object[] sampleData, + long approximateRowCnt, + int shardCount) { + return Collections.emptyList(); + } + }; + + splitterSuccess.splitTableIntoChunks(null, tableId, splitColumn); + assertTrue(samplingCalled[0], "Should invoke sampling for sparse IDs"); + assertFalse(evenlyCalled[0], "Should not invoke arithmetic fallback"); + assertFalse(unevenlyCalled[0], "Should not invoke uneven query fallback"); + + samplingCalled[0] = false; + evenlyCalled[0] = false; + unevenlyCalled[0] = false; + + AbstractJdbcSourceChunkSplitter splitterFallback = + new ConfiguredUtJdbcSourceChunkSplitter(config) { + @Override + public Object[] queryMinMax( + JdbcConnection jdbc, TableId tableId, Column columnName) { + return new Object[] {1, 50000}; + } + + @Override + public boolean isEvenlySplitColumn(Column splitColumn) { + return true; + } + + @Override + public Long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId) { + return 1000L; + } + + @Override + public Object[] sampleDataFromColumn( + JdbcConnection jdbc, + TableId tableId, + String columnName, + int samplingRate) { + samplingCalled[0] = true; + return new Object[] {1, 50000}; + } + + @Override + protected List<ChunkRange> splitEvenlySizedChunks( + TableId tableId, + Object min, + Object max, + long approximateRowCnt, + int chunkSize, + int dynamicChunkSize) { + evenlyCalled[0] = true; + return Collections.emptyList(); + } + + @Override + protected List<ChunkRange> splitUnevenlySizedChunks( + JdbcConnection jdbc, + TableId tableId, + Column splitColumn, + Object min, + Object max, + int chunkSize) { + unevenlyCalled[0] = true; + return Collections.emptyList(); + } + }; + + splitterFallback.splitTableIntoChunks(null, tableId, splitColumn); + assertTrue(samplingCalled[0], "Should attempt sampling for sparse IDs"); + assertFalse(evenlyCalled[0], "Should NOT fallback to arithmetic splitting if sparse"); + assertTrue( + unevenlyCalled[0], + "Should fallback to exact query splitting when sampler is defeated"); + } + + /** + * Verifies the actual chunk boundaries produced for a gap-skewed table (e.g., simulating + * #10270). This tests both the sampling-succeeds and sampling-falls-back boundary math + * end-to-end. + */ + @Test + public void testSplitTableIntoChunksBoundaryCorrectnessWithSparseKeys() throws Exception { + TestJdbcSourceConfig config = + new TestJdbcSourceConfig() { + @Override + public double getDistributionFactorUpper() { + return 100.0; + } + + @Override + public double getDistributionFactorLower() { + return 0.05; + } + + @Override + public int getSplitSize() { + return 2; + } + + @Override + public boolean isSampleShardingAllow() { + return true; + } + }; + + TableId tableId = new TableId("test", "test", "test"); + Column splitColumn = Column.editor().name("id").create(); + + // 1. Test fallback to exact-boundary query when sample is too sparse + // PKs: {1, 2, 5, 6, 100} -> min=1, max=100, approxRowCnt=5, chunkSize=2 + // shardCount = ceil(5/2) = 3 + // distributionFactor = (100 - 1 + 1)/5 = 20.0 (evenly distributed, but > 1.0 gap skew) + // inverseSamplingRate = min(1000, chunkSize=2) = 2 + // Simulated sample for MOD((id-1), 2) = 0 -> {1, 5}, length 2 < shardCount(3) -> falls back + // to uneven + AbstractJdbcSourceChunkSplitter splitterFallback = + new ConfiguredUtJdbcSourceChunkSplitter(config) { + @Override + public Object[] queryMinMax( + JdbcConnection jdbc, TableId tableId, Column columnName) { + return new Object[] {1, 100}; + } + + @Override + public boolean isEvenlySplitColumn(Column splitColumn) { + return true; + } + + @Override + public Long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId) { + return 5L; + } + + @Override + public Object[] sampleDataFromColumn( + JdbcConnection jdbc, + TableId tableId, + String columnName, + int samplingRate) { + // Return the sparse sample simulating MOD() filtering + return new Object[] {1, 5}; + } + + @Override + public Object queryNextChunkMax( + JdbcConnection jdbc, + TableId tableId, + String columnName, + int chunkSize, + Object includedLowerBound) { + // Exact boundary querying mock for {1, 2, 5, 6, 100} + int lower = includedLowerBound == null ? 0 : (Integer) includedLowerBound; + if (lower < 2) return 2; + if (lower < 6) return 6; + if (lower < 100) return 100; + return null; + } + }; + + List<ChunkRange> fallbackChunks = + splitterFallback.splitTableIntoChunks(null, tableId, splitColumn); + assertEquals(3, fallbackChunks.size()); + assertEquals(ChunkRange.of(null, 2), fallbackChunks.get(0)); + assertEquals(ChunkRange.of(2, 6), fallbackChunks.get(1)); + assertEquals(ChunkRange.of(6, null), fallbackChunks.get(2)); + + // 2. Test successful sampling-based boundary generation + // Same table, but chunkSize=1 -> shardCount=5 -> inverseSamplingRate=1 + // sample contains all rows: {1, 2, 5, 6, 100}, length 5 >= shardCount(5) -> efficient + // sharding + TestJdbcSourceConfig configSample = + new TestJdbcSourceConfig() { + @Override + public double getDistributionFactorUpper() { + return 100.0; + } + + @Override + public double getDistributionFactorLower() { + return 0.05; + } + + @Override + public int getSplitSize() { + return 1; + } + + @Override + public boolean isSampleShardingAllow() { + return true; + } + }; + + AbstractJdbcSourceChunkSplitter splitterSample = + new ConfiguredUtJdbcSourceChunkSplitter(configSample) { + @Override + public Object[] queryMinMax( + JdbcConnection jdbc, TableId tableId, Column columnName) { + return new Object[] {1, 100}; + } + + @Override + public boolean isEvenlySplitColumn(Column splitColumn) { + return true; + } + + @Override + public Long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId) { + return 5L; + } + + @Override + public Object[] sampleDataFromColumn( + JdbcConnection jdbc, + TableId tableId, + String columnName, + int samplingRate) { + // inverseSamplingRate=1 means all 5 rows are sampled + return new Object[] {1, 2, 5, 6, 100}; + } + }; + + List<ChunkRange> sampleChunks = + splitterSample.splitTableIntoChunks(null, tableId, splitColumn); + assertEquals(6, sampleChunks.size()); + assertEquals(ChunkRange.of(null, 1), sampleChunks.get(0)); + assertEquals(ChunkRange.of(1, 2), sampleChunks.get(1)); + assertEquals(ChunkRange.of(2, 5), sampleChunks.get(2)); + assertEquals(ChunkRange.of(5, 6), sampleChunks.get(3)); + assertEquals(ChunkRange.of(6, 100), sampleChunks.get(4)); + assertEquals(ChunkRange.of(100, null), sampleChunks.get(5)); + } + private void check(List<ChunkRange> a, List<ChunkRange> b) { checkRule(b); assertEquals(a, b);
