[flink] Support projection and filter pushdown for $changelog and $binlog virtual tables - #3759
Open
naivedogger wants to merge 11 commits into
Open
[flink] Support projection and filter pushdown for $changelog and $binlog virtual tables#3759naivedogger wants to merge 11 commits into
naivedogger wants to merge 11 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.
Contributor
There was a problem hiding this comment.
Pull request overview
Note
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Adds projection and filter pushdown support for Flink’s $changelog and $binlog virtual-table sources to reduce scanned columns and push eligible predicates into Fluss scans, while updating converters/deserializers and expanding unit/integration coverage.
Changes:
- Implement
SupportsProjectionPushDown/SupportsFilterPushDownfor$changelogand$binlogtable sources, including record-batch (statistics) and partition pruning pathways where supported. - Update changelog/binlog row converters and deserialization schemas to honor projection via
ProjectedRowDataand produced type derivation. - Add/expand unit and integration tests covering projection derivation, filter routing/pushdown behavior, and projection-aware conversion.
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java | Implements projection + filter pushdown for $changelog, derives dataProjection and baseRowProjection, adds statistics-based batch filter handling. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java | Implements top-level projection pushdown + filter pushdown interface parity for $binlog (no pushable filters). |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/ChangelogRowConverter.java | Adds base-row projection support via ProjectedRowData and produced type derivation. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/BinlogRowConverter.java | Adds top-level projection support via ProjectedRowData and produced type derivation. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java | Propagates base-row projection into converter and produced type. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java | Propagates top-level projection into converter and produced type. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java | Extracts statistics-column availability + predicate-eligibility logic into reusable utilities. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java | Switches to using PushdownUtils for stats gating of record-batch predicates. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java | Refactors change-type coverage into parameterized test; adds projection tests. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/BinlogRowConverterTest.java | Adds projection-focused converter tests (top-level projection, update merge, delete). |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java | Adds IT coverage for changelog projection + filter pushdown and metadata-only projection runtime path. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java | Adds IT coverage for binlog top-level projection. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java | New unit tests for $changelog pushdown behavior and copy() state preservation. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java | New unit tests for $binlog projection pushdown state and filter handling. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- Use the precomputed Fluss row type for record-batch predicate conversion instead of re-converting the Flink row type per filter. - Replace the O(n^2) contains/indexOf projection derivation with a LinkedHashMap position lookup (first-appearance order preserved). - Only include predicates that pass the partition-key-only check in partitionFilters to avoid pushing down unsafe partition predicates.
…of re-deriving it in deserialization schemas
…tive assertion for cross-partition records
| public void open(InitializationContext context) throws Exception { | ||
| if (converter == null) { | ||
| this.converter = new BinlogRowConverter(context.getRowSchema()); | ||
| this.converter = new BinlogRowConverter(context.getRowSchema(), projectedTopLevel); |
Contributor
There was a problem hiding this comment.
context.getRowSchema() is already projected rows, no need projectedTopLevel here
// in FlinkTableSource
// handle normal scan
RowType flussRowType = FlinkConversions.toFlussRowType(tableOutputType);
if (projectedFields != null) {
flussRowType = flussRowType.project(projectedFields);
}
Contributor
Author
There was a problem hiding this comment.
Thanks for the comment. I think projectedTopLevel is still required here.
context.getRowSchema() comes from FlinkSource.sourceOutputType. For $binlog,
the table source passes the full physical data-row type as sourceOutputType and
passes null as the scan projection, because before/after are whole nested
ROWs and the metadata columns do not exist in the physical Fluss row.
Therefore, context.getRowSchema() is the full stored data schema, not the
projected top-level binlog schema. projectedTopLevel is applied only after
BinlogRowConverter constructs the five-field virtual row
[_change_type, _log_offset, _commit_timestamp, before, after].
Removing it would make the declared produced type projected while the emitted
RowData still contains all five fields. A reordered projection such as
SELECT after, _change_type also requires this explicit top-level projection.
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
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.