[plusplusjiajia]

Purpose

QueryAuthSplit carries the row filter and the column masks. It implements Split but is not a DataSplit, and AbstractDataTableScan#plan() wraps every split once a table has rules, so the Flink source's casts fail on any table with one. #9929 fixed this class of bug in paimon-core; this is the same fix for the Flink source.

Each site had to answer the same question: does it read the underlying split, or hand the split on to a read? The ones that only read it look through the wrappers now, while the element they route, assign or key on stays the planned split — looking through something passed downstream would drop the rules and return unmasked, unfiltered rows instead of failing.

One wrapper is not enough. toFallbackSplit produces a FallbackDataSplit, which is a DataSplit and keeps every cast working — but only when handed a DataSplit. Under authorization it is handed a QueryAuthSplit and produces a FallbackSplitImpl instead, so enabling authorization is what turns a working cast into a failing one. Splits.underlying looks through both, stopping at a DataSplit: FallbackDataSplit is one and wraps itself, so that condition ends the loop as much as it preserves what a plan with no rules resolved to.

Nine sites, three of which fail quietly rather than loudly:

• TableScanUtils#getSnapshotId and FileStoreSourceSplitReader test for DataSplit and fall through, reporting no snapshot id and no file creation time — consumer progress and watermarks stop advancing for every split of an authorized table.
• AlignedSplitAssigner's instanceof PlaceholderSplit read the wrapper, so a wrapped placeholder took the wrong branch.
• ContinuousFileSplitEnumerator, AlignedContinuousFileSplitEnumerator, PreAssignSplitAssigner, DynamicPartitionPruningAssigner, MonitorSource#shuffleOrdered and AlignedSplitAssigner's four alignment reads fail the cast outright.

Two needed more than unwrapping. MonitorSource#shuffleOrdered only ever handled a DataSplit, so an IncrementalSplit failed the same cast with or without rules; it keys on partition and bucket for either shape now. ReadOperator reports UNDEFINED for a split with no file creation time rather than failing a read over a metric it cannot compute, and no longer turns that UNDEFINED into an epoch-sized currentEmitEventTimeLag. That last guard is the one thing here without a test: the value is only observable through the operator's metric group, and intercepting it needs an API whose signature differs between Flink 1 and Flink 2.

Making these casts succeed exposed one more way to answer wrong. Dynamic partition pruning runs in the assigners, after planning, and decides from the partition the plan recorded — which a masked partition key makes the wrong value to decide on. A row stored under p=2 but read back as p=1 had its split pruned against a dynamic key of {1} and quietly went missing; previously this path crashed on the cast instead. A mask on a filtering field keeps the split now and leaves the join to filter on what the reader produces.

QueryAuthSplit.unwrap is removed: a second wrapper meant every caller had to change anyway, so Splits.underlying replaces it everywhere, including three sites in ChainGroupReadTable and ChainTableFileStoreTable that had the same exposure and had simply not met a fallback-wrapped split yet. Two entry points differing only in how far they look would be the next version of this bug.

[JingsongLi]

Requirement fit: SUPPORTED (triage: GO)

Implementation: CLEAN

This restores query authorization across the Flink source end to end: code that only inspects split shape looks through QueryAuthSplit and fallback wrappers, while readers and assigners continue carrying the original wrapped split so row filters and masks are not lost. I also checked dynamic partition pruning with masked keys, incremental split routing, alignment, snapshot/watermark metrics, serialization, and chain/fallback reads. The focused tests cover the wrapper combinations and both Flink lines. I did not find an actionable issue, and CI is green.