[flink] Support projection and filter pushdown for $changelog and $binlog virtual tables#3759
Open
naivedogger wants to merge 6 commits into
Open
[flink] Support projection and filter pushdown for $changelog and $binlog virtual tables#3759naivedogger wants to merge 6 commits into
naivedogger wants to merge 6 commits into
Conversation
…irtual tables Add projection and filter pushdown to the $changelog and $binlog virtual table sources, with dedicated converter/source unit tests and end-to-end ITCases. Fixes two issues found in review: - Metadata-only changelog projection produced an empty (zero-column) scan and yielded no records; normalize the scan projection to null so the full data row is scanned and metadata is selected during deserialization. - Changelog log filter pushdown is now only enabled for the ARROW log format, avoiding an UnsupportedOperationException from TableScan#createLogScanner on INDEXED/COMPACTED tables.
Introduce an SLF4J logger in PushdownUtils and emit diagnostics when statistics collection is disabled, and when a configured statistics column is skipped due to an unsupported type or because it is absent from the table schema.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Linked issue: close #3722
The $changelog and $binlog virtual tables did not implement any pushdown
abilities, so every query read the full data row and applied all filters in
Flink. This change lets both virtual-table sources implement
SupportsProjectionPushDownandSupportsFilterPushDown, reducing scannedcolumns and, where possible, pushing filters down to the Fluss scan.
Brief change log
ChangelogFlinkTableSourcenow implementsSupportsProjectionPushDownandSupportsFilterPushDown:_change_type/_log_offset/_commit_timestamp+ data columns) into thephysical
dataProjectionscanned from Fluss and abaseRowProjectionthatreorders the base changelog row during deserialization. A metadata-only
projection normalizes the scan to a full data-row read to avoid a
zero-column scan.
filters (statistics-based record-batch skipping, ARROW log format only);
metadata-column filters are naturally non-pushable. All filters are still
returned as remaining so Flink applies them as a safety net (FLINK-38635).
BinlogFlinkTableSourcenow implements both abilities:[_change_type, _log_offset, _commit_timestamp, before, after]; the datascan stays full because
before/afterare whole nested ROWs (nestedpruning is out of scope).
non-pushable, data/partition columns are nested inside
before/after), sonothing is pushed and all filters are returned to Flink; implemented for
interface parity.
ChangelogDeserializationSchema/BinlogDeserializationSchemaand their row converters honor the projection when building the produced row
and row type (via
ProjectedRowData).Tests
ChangelogFlinkTableSourceTest,BinlogFlinkTableSourceTest— projectionderivation, partition/data-column/metadata filter routing, statistics
gating, and
copy()state preservation.ChangelogRowConverterTest,BinlogRowConverterTest— change-typeconversion, before/after nesting, update merge, and projection reordering.
ChangelogVirtualTableITCase— projection, metadata-only projection,partition-filter and data-column-filter pushdown (asserted via
explainSql), startup modes.BinlogVirtualTableITCase— top-level projection, nested field access,partitioned tables, startup modes.
API and Format
No public API or storage format changes. The affected sources are internal
Flink connector components.
Documentation
No documentation changes.