[spark] Support filter pushdown for log tables

[fresh-borzoni]

closes #3117

Adds SupportsPushDownFilters to FlussAppendScanBuilder and a SparkPredicateConverter mirroring Flink's PredicateConverter.

Record-batch pushdown uses the server-side filter from #2951, Spark re-applies every filter as a safety net, making pushdown a pure optimization.

[fresh-borzoni]

@Yohahaha @YannByron @luoyuxia PTAL 🙏

[Yohahaha]

thank you! left some comments.

[Yohahaha]

LGTM!

[fresh-borzoni]

@Yohahaha @YannByron Ty for the review 👍

Redesigned to the newer API, PTAL 🙏

[YannByron]

LGTM. thanks @fresh-borzoni

[fresh-borzoni]

@luoyuxia Can you take a look, pls?

[Copilot]

Pull request overview

This PR adds Spark-side filter pushdown for Fluss log (append) tables by converting Spark V2 predicates into Fluss predicates and applying them as server-side record-batch filters, while still letting Spark re-apply predicates for row-exact correctness.

Changes:

• Introduce SparkPredicateConverter to translate Spark V2 Predicate expressions into Fluss Predicates.
• Wire Spark V2 filter pushdown into FlussAppendScanBuilder/FlussAppendScan and apply the pushed predicate in FlussAppendPartitionReader via table.newScan().filter(...).
• Add unit/integration tests validating predicate conversion and verifying pushdown shows up in Spark scan plans.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/utils/SparkPredicateConverter.scala	New converter from Spark predicates to Fluss predicates for pushdown.
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala	Add SupportsPushDownV2Filters mixin and pushdown plumbing for append scans.
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScan.scala	Extend FlussAppendScan to carry pushed predicates and include them in scan description.
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussBatch.scala	Thread pushed predicate into append batch reader factory creation.
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussPartitionReaderFactory.scala	Extend append reader factory to accept an optional pushed predicate.
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussAppendPartitionReader.scala	Apply server-side batch filter via TableScan.filter(...) when reading.
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala	Update append micro-batch reader factory call signature (currently still passes None).
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakeAppendBatch.scala	Update factory construction signature for fallback path.
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakePartitionReaderFactory.scala	Update append partition reader construction signature for log splits.
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/utils/SparkPredicateConverterTest.scala	New unit tests covering predicate conversion semantics.
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala	New tests asserting Spark plans show pushed predicates for log-table reads.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[luoyuxia]

@fresh-borzoni Thanks for the pr. Only one minor comments. Also, will the pushdown for lake reader be done in following pr?

[fresh-borzoni]

@luoyuxia Ty for the review, addressed comments, PTAL 🙏

Yes, pushdown for the lake reader is planned as a follow-up, here it is primarly converter introduction for further use-cases.