JingsongLi commented on code in PR #10297:
URL: https://github.com/apache/paimon/pull/10297#discussion_r4261988319
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWrite.java:
##########
@@ -124,6 +126,22 @@ static StoreSinkWrite.Provider createWriteProvider(
boolean isStreaming,
boolean ignorePreviousFiles,
boolean hasSinkMaterializer) {
+ return createWriteProvider(
+ fileStoreTable,
+ checkpointConfig,
+ isStreaming,
+ ignorePreviousFiles,
+ hasSinkMaterializer,
+ null);
Review Comment:
[P2] Handle the CDC sinks before falling back to a null mapping
`CdcFixedBucketSink` inherits the five-argument provider and therefore takes
this null-mapping path; CDC multi-table writes also construct the legacy writer
directly. The new core guard then rejects the first record of any partitioned
fixed-bucket CDC table with `bucket.per-partition-count-enabled=true`, even
before any rescale. I exercised the actual CDC sink provider and
`CdcRecordStoreWriteOperator` harness against a real Parquet-backed table: the
option-disabled control writes and flushes, while enabling the option fails in
`processElement` with the new `UnsupportedOperationException`. CDC table
configuration accepts this property and the documentation currently describes
Flink write support generally. Please propagate the same per-partition mapping
through CDC routing and writes, or explicitly restrict the supported scope and
reject this configuration when constructing a CDC job. Removing the core safety
guard alone would leave CDC routing on the schema default.
##########
paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java:
##########
@@ -179,6 +179,16 @@ public FileStoreScan withPartitionBucket(BinaryRow
partition, int bucket) {
return this;
}
+ @Override
+ public FileStoreScan clearPartitionAndBucketFilters() {
+ manifestsReader.withPartitionFilter((Predicate) null);
+ manifestsReader.withBucketFilter(null);
Review Comment:
[P1] Clear the manifest reader’s bucket selectors as well
This clears `ManifestsReader.bucketFilter`, but `withBucket` stores a
separate `ManifestsReader.specifiedBucket`; its `totalAwareBucketFilter` also
survives. `readPartitionEntries` uses the reader’s filtered manifests, so
resetting only the scan fields still omits manifests whose bucket bounds
exclude the previous bucket. I reproduced this with real PK files: restore
P1/bucket 0, physically overwrite P2 from 2 to 4 buckets with its key in bucket
3, then reuse the restorer for empty P2/bucket 1. The reused mapping returns
`totalBuckets=null`, while a fresh restorer returns 4 and rejects the old
count-2 route. With the reused restorer, that same stale write passes and an
ordinary commit/read returns both values 300 and 999 for the same primary key.
Please clear every bucket selector in the manifest reader, or use an
independent unfiltered layout scan. Cover actual bounded manifests; the new
mock-only clearing test does not exercise this implementation.
--
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]
