fix(spark): incremental query returns no rows when projection excludes _hoodie_commit_time - #19568
fix(spark): incremental query returns no rows when projection excludes _hoodie_commit_time#19568nsivabalan wants to merge 1 commit into
Conversation
An incremental query filters on _hoodie_commit_time and that filter is pushed into
the file reader, but the reader is only given the columns the query projects. When
the projection omits the filtered column the predicate cannot be satisfied and every
row is dropped -- silently, with no error.
So on a CoW incremental query:
collect() -> 2 rows (correct)
count() -> 0 rows (wrong)
select("k").collect() -> 0 rows (wrong)
select("k").count() -> 0 rows (wrong)
collect() worked only incidentally: it happens to project every column, including
the filtered one. The general statement is that any incremental query whose
projection excludes _hoodie_commit_time returns nothing, and count() is simply its
most visible instance -- a silent wrong answer that reads as "no new data".
readBaseFile now reads any column a required filter references but the projection
omits, then projects it back out before returning so the caller's schema is exact.
Partition columns are excluded: the caller appends those from partition values, so
adding one here would duplicate the field and shift every ordinal after it.
This is a strict no-op for every non-incremental read. Snapshot, CDC and bootstrap
relations all supply Seq.empty for required filters, so there is nothing to
reference and nothing to add -- no extra column is read by the ordinary count()
path, which matters because that is the most common query in Hudi.
Reproduced on unmodified master (65cf7e8) at table version 10, and confirmed
not to be specific to a table version or to any meta-fields configuration.
Tests: TestIncrementalQueryProjection covers count(), projected reads, the actual
row values rather than just counts, and a snapshot no-op guard. Verified not
vacuous -- with the fix reverted, exactly the three incremental cases fail and the
snapshot guard still passes.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR fixes a silent data-loss bug where a CoW incremental query returns zero rows whenever its projection omits _hoodie_commit_time, by reading any filter-referenced column that the projection excludes and then projecting it back out before returning rows. I traced the projection logic across all three read branches (by-name projections in both generateUnsafeProjectionInternal and ColumnarBatchUtils.generateProjection make the append-and-drop robust), confirmed the presence check against requiredSchemaIn is equivalent for non-partition columns given how outputSchema/requestedStructType are built at the call site, and verified the change is a strict no-op for snapshot/CDC/bootstrap reads (which supply Seq.empty required filters). One completeness question on MOR coverage is worth a look in the inline comments. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. One small naming nit on the helper's candidates parameter — otherwise the fix is clear and well-commented.
| return DataTypes.createStructType(new StructField[] { | ||
| DataTypes.createStructField("record_key", DataTypes.StringType, true), | ||
| DataTypes.createStructField("partition_path", DataTypes.StringType, true), | ||
| DataTypes.createStructField("payload", DataTypes.StringType, true) |
There was a problem hiding this comment.
🤖 The fix and all four tests are CoW-only. For MOR incremental, file slices that have log files route through the HoodieFileGroupReader path (the case Some(fileSlice) if !isCount && (requiredSchema.nonEmpty || logFiles present) branch) rather than readBaseFile, and there requiredFilters are handed to SparkFileFormatInternalRowReaderContext and applied against getAppliedRequiredSchema(...). Have you checked whether a MOR incremental count()/projected read hits the same missing-_hoodie_commit_time predicate on the base-file read there? If so it may warrant either the same augmentation or at least a MOR test case; if it's already covered by the merge schema, a one-line note here would help.
| * Partition columns are excluded by the caller, which appends them separately. | ||
| */ | ||
| private def filterColumnsMissingFrom(target: StructType, | ||
| candidates: StructType, |
There was a problem hiding this comment.
🤖 nit: candidates is a bit abstract here — the Javadoc calls it "the full table schema", so could you name the parameter fullSchema (or tableSchema) to make the call-site self-documenting?
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19568 +/- ##
============================================
- Coverage 76.84% 76.84% -0.01%
- Complexity 32379 32391 +12
============================================
Files 2522 2522
Lines 139106 139125 +19
Branches 16713 16719 +6
============================================
+ Hits 106892 106906 +14
- Misses 24621 24628 +7
+ Partials 7593 7591 -2
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
4 participants
Describe the issue this Pull Request addresses
A CoW incremental query returns zero rows whenever its projection excludes
_hoodie_commit_time.count()is the most visible instance, and it fails silently — a wrong answer that reads as "no new data" rather than an error.Reproduced on unmodified
master(65cf7e8ede7d), table version 10, on a plain CoW table with no special configuration:Not specific to a table version, a table type beyond CoW, or any meta-fields configuration.
Root cause
An incremental query filters on
_hoodie_commit_time—IsNotNullplus anInover the instants in range — and those filters are pushed into the file reader asrequiredFilters. The reader evaluates a pushed predicate against the schema it was asked to read, which contains only the columns the query projects.When the projection omits
_hoodie_commit_time, the predicate cannot be satisfied and every row is dropped.collect()works only incidentally: it happens to project every column, including the filtered one. It is not a working case so much as a case that has not failed yet.count()projects nothing at all, andselect(subset)projects only what the user asked for — both hit it.Verified by instrumenting
readBaseFilerather than by inspection. The filters are identical in all three cases; only the schema differs:collect()_hoodie_commit_timeIsNotNull(_hoodie_commit_time),In(...)count()[]select("k")[k]Worth noting what is not the bug, since both look suspicious and both are correct:
isCountexcludes incremental deliberately (HoodieFileGroupReaderBasedFileFormat:254). You cannot answer an incrementalcount()from row counts, because only some rows fall in the commit range.31eea3b9a851), which only restructured an innerifinto the guard.Both mechanisms are right in isolation. The defect is the seam between them: a required filter references a column, and nothing guarantees that column is in the schema handed to the reader.
Summary and Changelog
readBaseFilenow reads any column a required filter references but the projection omits, then projects it back out before returning so the caller's schema stays exact.Two details worth review:
Impact
Behavior: incremental queries that previously returned zero rows now return the correct rows. No other query shape changes.
Strict no-op for every non-incremental read. Snapshot, CDC and bootstrap relations all supply
Seq.emptyfor required filters, so there is nothing to reference and nothing to add. No extra column is read on the ordinarycount()path — which matters, since that is the most common query in Hudi.Performance: on an incremental read whose projection omits the filtered column, one extra column is read and then projected away. That is the minimum needed to evaluate the predicate correctly.
API / storage: unchanged.
Risk Level
medium
The change is small and its no-op property is structural rather than incidental. Raised above "low" only because
readBaseFileis on the shared read path for every Spark query — snapshot, incremental, CDC, bootstrap and MoR all route through it — so the blast radius of a mistake here is wide even though this change is narrow.Tests
TestIncrementalQueryProjection, four cases:count()agrees withcollect()Verified not vacuous: with the fix reverted, exactly the three incremental cases fail and the snapshot guard still passes. That check mattered here — the first version of this fix compiled cleanly and did nothing, and a test run only against the fixed build would have looked identical.
Contributor's checklist
🤖 Generated with Claude Code