Copilot commented on code in PR #12463:
URL: https://github.com/apache/seatunnel/pull/12463#discussion_r4090951725
##########
seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java:
##########
@@ -173,6 +173,46 @@ private List<ChunkRange> splitTableIntoChunks(
&& 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) (approximateRowCnt / chunkSize);
+ if (doubleCompare(distributionFactor, 1.0d) > 0
+ && sampleShardingAllow
+ && shardCount > 0) {
+ 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 > 0) {
+ return efficientShardingThroughSampling(
+ tableId, sample, approximateRowCnt,
shardCount);
Review Comment:
Checking only `sample.length > 0` is not enough for this fix. The MySQL
sampler selects rows where `(id - MIN(id)) MOD inverseSamplingRate = 0`; with
sparse keys such as 0, 1,000,001, 2,000,002 and the default rate 1000, the
sample can contain only the minimum even though the table has many rows.
`efficientShardingThroughSampling` then creates only two ranges, leaving one
overloaded chunk and the original checkpoint-timeout failure intact. Please
make sampling representative for sparse keys (or detect an insufficient sample
and use a splitting strategy that does not under-split).
##########
seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java:
##########
@@ -173,6 +173,46 @@ private List<ChunkRange> splitTableIntoChunks(
&& 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) (approximateRowCnt / chunkSize);
Review Comment:
`shardCount` is floored here, so a table with just over one `chunkSize` of
rows (for example, 1,001 rows with a split size of 1,000) gets `shardCount ==
1`. `efficientShardingThroughSampling` then returns one unbounded range,
defeating the configured split size and leaving the entire table in a single
snapshot task. Use a ceiling when deriving the shard count (while preserving
the existing overflow/zero handling).
##########
seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java:
##########
@@ -173,6 +173,46 @@ private List<ChunkRange> splitTableIntoChunks(
&& 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) (approximateRowCnt / chunkSize);
+ if (doubleCompare(distributionFactor, 1.0d) > 0
+ && sampleShardingAllow
+ && shardCount > 0) {
Review Comment:
This condition bypasses the existing `sample-sharding.threshold` for every
evenly distributed table with a factor above 1.0. In the MySQL implementation,
`sampleDataFromColumn` reads every row with `SELECT <column> FROM <table>`
before retaining 1/N values
(`connector-cdc-mysql/.../MySqlUtils.java:167-217`), so large tables with only
modest gaps now incur an additional full-table scan during startup. Please
retain a bounded/opt-in trigger or otherwise document and account for this
extra O(row-count) planning cost.
##########
seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitterTest.java:
##########
@@ -155,6 +155,61 @@ public void testEfficientShardingThroughSampling() throws
NoSuchMethodException
ChunkRange.of(6, null)));
}
+ /**
+ * When IDs are non-consecutive (distributionFactor > 1.0) and sampling is
enabled, the
+ * splitEvenlySizedChunks path should NOT be used. Instead, sampling-based
splitting should be
+ * preferred for better chunk balance.
+ *
+ * <p>This test verifies that for a table with IDs spanning a wide range
but few actual rows
+ * (distributionFactor >> 1.0), the algorithm falls back to arithmetic
stepping only when
+ * sampling is disabled.
+ */
+ @Test
+ public void testSplitEvenlySizedChunksWithNonConsecutiveIds() {
Review Comment:
This test never exercises the new decision in `splitTableIntoChunks`: it
calls `splitEvenlySizedChunks` and `efficientShardingThroughSampling` directly
on a splitter whose source config is null. It would pass even if the added
`distributionFactor > 1.0` condition, sampling callback, or fallback were
removed, so the PR's core regression is unprotected. Please configure a test
double and invoke the actual split-generation path, asserting the sampling
callback is used for sparse IDs and the arithmetic fallback is used when
sampling is disabled or empty.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
