[spark] Fix Fluss Spark read paths

[fkzhao]

Fix Spark reads for Iceberg sort-merge tables and append log scans.

Handle empty and progress-only log scan batches without reporting a false end of data, and preserve consumed offsets when reading full projections. Use independent rows and the correct timestamp representation in Iceberg sort-merge reads to avoid projection and timestamp failures.

Add regression coverage for the affected Iceberg and log-change reader paths.

Purpose

Fix Spark read failures when querying Fluss tables with full-column projections such as SELECT *.

The Spark append reader previously treated an empty ScanRecords result as end-of-data. However, a scan result may be empty while the scanner has already advanced its consumed offset, or may simply be an empty poll caused by a timeout. This caused Spark tasks to fail with an incorrect No more data from fluss server exception.

This change also fixes Iceberg sort-merge reading and empty log-change iterator handling.

Brief change log

• Handle progress-only and empty poll results in FlussAppendPartitionReader.
• Advance the local offset using consumedUpToOffset after the current batch is consumed.
• Fix Iceberg sorted reads for primary-key projections and timestamp types.
• Avoid sharing mutable row objects between Iceberg records.
• Handle empty input correctly in LogChangesIterator.
• Add regression test coverage for the affected read paths.

Tests

• IcebergLakeSourceTest: 5 tests passed.

• Spark common module compilation passed.

• Spark 3.5 connector packaging passed with:

  mvn -pl fluss-spark/fluss-spark-3.5 \
      -am \
      -DskipTests \
      -Dcheckstyle.skip=true \
      -Drat.skip=true \
      package

[Copilot]

🟡 Changes recommended

The new Iceberg key comparator has confirmed ordering bugs for TIMESTAMP_NANO precision handling and signed byte comparisons that can mis-order primary keys.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR fixes several Spark read-path edge cases in Fluss integrations: bounded append log scans that can return empty/progress-only batches, and Iceberg sort-merge reads that must preserve correct primary-key ordering and avoid mutable row reuse when projecting full rows.

Changes:

• Update FlussAppendPartitionReader to handle empty/progress-only polls and only advance offsets after finishing a batch.
• Make LogChangesIterator robust to empty inputs and add a regression test.
• Add an Iceberg SortedRecordReader implementation for sort-merge union reads and extend Iceberg lake-source tests.

File summaries

File	Description
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/utils/LogChangesIteratorTest.scala	Adds regression coverage for empty log-change inputs.
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/utils/LogChangesIterator.scala	Avoids .head on empty record sets by using an empty iterator.
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussAppendPartitionReader.scala	Handles empty/progress-only scan batches and defers applying consumed offsets until the batch is drained.
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergLakeSourceTest.java	Adds coverage for sorted readers with missing splits and PK projections.
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSortedRecordReader.java	Introduces a sorted Iceberg reader for primary-key sort/merge.
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordReader.java	Ensures per-record row objects are independent to prevent projection/mutation issues.
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergLakeSource.java	Returns the sorted reader when the reader context requires sorted records.

Review details

• Files reviewed: 7/7 changed files
• Comments generated: 2
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.