This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new fd0211d6f [flink] Fix batch fallback generating mixed split types for
primary-key tables (#3296)
fd0211d6f is described below
commit fd0211d6f926b34f69a2ee4e03e97fa55d59caea
Author: Howie Wang <[email protected]>
AuthorDate: Fri May 22 21:15:47 2026 +0800
[flink] Fix batch fallback generating mixed split types for primary-key
tables (#3296)
---
.../source/enumerator/FlinkSourceEnumerator.java | 34 +++++++++++++---------
1 file changed, 20 insertions(+), 14 deletions(-)
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
index e04989cbb..0b7e98b39 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
@@ -493,9 +493,12 @@ public class FlinkSourceEnumerator
p.getPartitionId(),
p.getPartitionName()))
.collect(Collectors.toList());
- splits =
this.initPartitionedSplits(partitions);
+ // Use log-only splits to avoid generating
mixed split
+ // types (HybridSnapshotLogSplit + LogSplit)
for
+ // primary-key tables, which is not supported.
+ splits =
this.initLogTablePartitionSplits(partitions);
} else {
- splits = this.initNonPartitionedSplits();
+ splits = this.getLogSplit(null, null);
}
}
return splits;
@@ -850,18 +853,21 @@ public class FlinkSourceEnumerator
}
if (!bucketsNeedInitOffset.isEmpty()) {
- startingOffsetsInitializer
- .getBucketOffsets(partitionName, bucketsNeedInitOffset,
bucketOffsetsRetriever)
- .forEach(
- (bucketId, startingOffset) ->
- splits.add(
- new LogSplit(
- new TableBucket(
-
tableInfo.getTableId(),
- partitionId,
- bucketId),
- partitionName,
- startingOffset)));
+ Map<Integer, Long> startingOffsets =
+ startingOffsetsInitializer.getBucketOffsets(
+ partitionName, bucketsNeedInitOffset,
bucketOffsetsRetriever);
+ Map<Integer, Long> stoppingOffsets =
+ stoppingOffsetsInitializer.getBucketOffsets(
+ partitionName, bucketsNeedInitOffset,
bucketOffsetsRetriever);
+ startingOffsets.forEach(
+ (bucketId, startingOffset) ->
+ splits.add(
+ new LogSplit(
+ new TableBucket(
+ tableInfo.getTableId(),
partitionId, bucketId),
+ partitionName,
+ startingOffset,
+ stoppingOffsets.get(bucketId))));
}
return splits;
}
