[flink] Add projection and filter pushdown for virtual log tables

[loserwang1024]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Currently, Fluss virtual log tables ($changelog and $binlog) only implement ScanTableSource.

Compared with normal Fluss tables, virtual log tables do not support projection pushdown or filter pushdown yet. As a result, queries on $changelog / $binlog always read and deserialize full records even when only a few columns are required, and filters cannot be used to prune partitions, log batches, or metadata ranges early.

This issue aims to align virtual log tables with normal Fluss tables by adding projection and filter pushdown support.

Solution

No response

Anything else?

No response

Willingness to contribute

• I'm willing to submit a PR!

[naivedogger]

Please assign this to me, thanks!