fix(streamer): Include start commit in S3/GCS IncrSource incremental query

[yihua]

Describe the issue this Pull Request addresses

S3/GCS cloud-object incremental sources can silently drop records whenever a previous batch persisted a commit#fileKey mid-commit-pagination checkpoint (i.e., the prior batch hit sourceLimit before exhausting the start commit's files). Files in the start commit after the checkpoint key become unreachable, and the persisted checkpoint advances past them as a bare instant.

Affected sources and conditions:

• S3EventsHoodieIncrSource / GcsEventsHoodieIncrSource
• Triggered when at least one upstream source-events commit exceeds hoodie.deltastreamer.read.source.limit

Common triggers: cold-start backfills against a source table with a big initial commit, bursty event writers, low sourceLimit overrides. Steady-state streams whose upstream commits fit within sourceLimit are unaffected. HoodieIncrSource (non-cloud) does not go through QueryRunner and is unaffected.

Summary and Changelog

Root cause 1: start commit dropped from the scan. QueryRunner.runIncrementalQuery passes queryInfo.getStartInstant() as the Spark START_COMMIT. The Spark V1 incremental relation's findInstantsInRange is (start, end] (start-exclusive), so the start commit is dropped from the scan. The downstream (commit_time || object_key) > 'commit#fileKey' filter in IncrSourceHelper.filterAndGenerateCheckpointBasedOnSourceLimit then matches nothing in the start commit, the empty-batch branch fires, and the new checkpoint is emitted as endInstant with no #fileKey suffix. The next batch resumes past the gap. Regression introduced by ab4c7774a6 [HUDI-8141] Support incremental query with completion time (#11947).

Root cause 2: file-group-reader incremental scans return no rows when _hoodie_commit_time is pruned. The incremental relations enforce the commit-time span filters only via parquet filter push-down inside HoodieFileGroupReaderBasedFileFormat. When Spark's column pruning drops _hoodie_commit_time from the scan schema, which happens for count(), isEmpty() (the first operation the cloud sources run on the batch), and any projection without meta fields, parquet evaluates the predicate against a missing column as all-null and filters out every row, so the batch looks empty and the checkpoint again advances as a bare instant. Sessions with HoodieSparkSessionExtension are shielded because AdaptIngestionTargetLogicalRelations injects the same filters into the logical plan; sessions without the extension relied solely on the broken push-down path. Affects both V1 and V2 incremental relations through the file group reader.

Fix.

1. Force the V1 incremental relation regardless of source table version (QueryRunner): set INCREMENTAL_READ_TABLE_VERSION = 6 unconditionally. Cloud event sources always use V1 checkpoints (commit#fileKey, requested-time) externally, so they always route through the V1 incremental relation internally.
2. Pass previousInstant (not startInstant) to START_COMMIT (QueryRunner): with start-exclusive semantics, the scan range (previousInstant, end] still includes the start commit needed for commit#fileKey pagination.
3. Read required-filter columns regardless of Spark's column pruning (HoodieFileGroupReaderBasedFileFormat): columns referenced by the relation's required filters but pruned from the scan schema are now read from the file and projected away after filtering. This restores format-level enforcement that HUDI-6658 moved to the HoodieSparkSessionExtension-gated plan-injection rule (leaving sessions without the extension unprotected) and HUDI-7567 removed the last remnants of. Snapshot/CDC reads have no required filters, so their read path is unchanged.

Tests.

• testRealQueryRunnerResumesMidCommitPagination in both TestS3EventsHoodieIncrSource and TestGcsEventsHoodieIncrSource (@ParameterizedTest over source meta-table versions 6/8 x COW/MOR) exercises a real QueryRunner against an on-disk events meta-table with the file group reader enabled, resumes from a mid-commit commit#fileKey checkpoint with sourceLimit smaller than the remaining files, and asserts both the next persisted checkpoint and the exact files passed downstream. The source meta-table is written non-partitioned with the production S3 events schema (no test-only schema changes); the MOR runs assert the second commit's records land in log files so the read exercises log merging. Supporting S3EventsHoodieIncrSourceHarness changes: non-partitioned meta-table writes, multi-record commits, and using the table's commit action type so MOR writes commit as delta commits.
• New TestIncrementalReadWithFileGroupReader covers {COW, MOR} x {source v6 + v6 read, source v8 + v6 read, source v8 + v8 read} on a session without HoodieSparkSessionExtension (no plan-injected filters; the file format alone enforces the span filters). Each run writes 3 insert commits (small-file handling keeps a single file group with base files only, validated as a precondition) and 3 update commits (one log file each on MOR, base rewrites on COW, also validated), then queries incremental ranges over base-only (multi-file and single-file), base-plus-some-logs, logs-only, and empty windows, validating the exact row set for select * plus a narrow projection, count(), and isEmpty per range. One key is updated in every update commit, so a range spanning several updates must surface it exactly once with the latest in-range value; a precondition also asserts the latest base file retains rows from multiple commits: the carried-over rows the span filters must exclude. Without the fix, the pruned-schema shapes fail: COW count() returns 0; MOR count() returns only log-merged records with base-file rows silently dropped.

Impact

Behavior change is contained to incremental reads:

• The incremental read of the source events meta-table now always uses the V1 relation (requested-time semantics), matching the V1 commit#fileKey checkpoint contract regardless of source table version.
• commit#fileKey mid-commit-pagination resumes now re-include the start commit in the scan (fixes the silent data drop).
• Incremental queries through the file group reader now return correct results for query shapes that prune _hoodie_commit_time (e.g., count(), isEmpty()), for both V1 and V2 relations.

No API, config, or on-disk format changes; HoodieIncrSource (non-cloud) is untouched.

Risk Level

low

Documentation Update

none

Contributor's checklist

• Read through contributor's guide
• Enough context is provided in the sections above
• Adequate tests were added if applicable

[codecov-commenter]

Codecov Report

❌ Patch coverage is 60.00000% with 8 lines in your changes missing coverage. Please review.
✅ Project coverage is 53.66%. Comparing base (1fd2c36) to head (bb7b1af).
⚠️ Report is 9 commits behind head on master.

Files with missing lines	Patch %	Lines
...parquet/HoodieFileGroupReaderBasedFileFormat.scala	52.94%	5 Missing and 3 partials ⚠️

❗ There is a different number of reports uploaded between BASE (1fd2c36) and HEAD (bb7b1af). Click for more details.

HEAD has 31 uploads less than BASE

Flag	BASE (1fd2c36)	HEAD (bb7b1af)
spark-java-tests	18	0
spark-scala-tests	12	0
utilities	1	0

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #18949       +/-   ##
=============================================
- Coverage     68.26%   53.66%   -14.60%     
+ Complexity    29500    21865     -7635     
=============================================
  Files          2542     2452       -90     
  Lines        142618   132467    -10151     
  Branches      17790    15525     -2265     
=============================================
- Hits          97352    71084    -26268     
- Misses        37261    55745    +18484     
+ Partials       8005     5638     -2367     

Flag	Coverage Δ
common-and-other-modules	45.32% <60.00%> (+0.53%)	⬆️
hadoop-mr-java-client	44.77% <ø> (+0.09%)	⬆️
spark-client-hadoop-common	48.08% <ø> (+0.02%)	⬆️
spark-java-tests	?
spark-scala-tests	?
utilities	?

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...he/hudi/utilities/sources/helpers/QueryRunner.java	50.00% <100.00%> (+27.50%)	⬆️
...parquet/HoodieFileGroupReaderBasedFileFormat.scala	50.99% <52.94%> (-33.03%)	⬇️

... and 914 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[hudi-bot]

CI report:

• 0f4d233 UNKNOWN
• 199e19c UNKNOWN
• b95efa8 UNKNOWN
• ae54e8f UNKNOWN
• 40dc1bf UNKNOWN
• 92d67e5 UNKNOWN
• bb7b1af Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build