fix(spark): apply procedure filters before the limit, not after - #19974
rangareddy wants to merge 2 commits into
Conversation
Every show_* procedure that takes both limit and filter truncated to limit first and evaluated the filter on the truncated list, so `limit => 10, filter => ...` returned the matches among the first 10 rows rather than the first 10 matching rows. Every one of these procedures has a default limit, so a plain `filter => ...` call was affected too. BaseProcedure gains applyFilterAndLimit, which filters and then truncates. Where limit only truncated an already-materialised list, the early bound is simply dropped. Where it bounded real work, capping how many clean instants get their metadata read and how many records are pulled out of log files, it is lifted only when a filter is present, so an unfiltered call does exactly the work it did before. show_timeline needs one extra condition: getTimelineEntries deliberately ignores limit when both startTime and endTime are given, so reapplying the bound unconditionally would truncate a fully specified range to the default.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19974 +/- ##
=========================================
Coverage 80.32% 80.33%
+ Complexity 34745 34741 -4
=========================================
Files 2545 2545
Lines 142491 142458 -33
Branches 17306 17392 +86
=========================================
- Hits 114463 114440 -23
+ Misses 20126 20117 -9
+ Partials 7902 7901 -1
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR moves filter evaluation ahead of limit in the ten show_* procedures so that limit bounds the matching rows rather than the rows the filter gets to see, and lifts the source-side bound (DataFrame limit, rdd.take, clean/timeline/log-record scan caps) only when a filter is present. The logic traces cleanly across all ten procedures and the Int.MaxValue scan limits are safe against the loader's >=-based checks. One thing worth double-checking in the inline comments: the log-file-records procedure now materialises every record on the driver when a filter is supplied, which may be a memory concern on large file groups. 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. Logic looks correct and well-commented; one recurring duplication worth a shared helper.
cc @yihua
Review feedback on the first revision: lifting the scan bound so a filter could see every record made show_logfile_records buffer all records of every matched log file into an on-heap list, so a plain `filter => ...` on a large file group could exhaust the driver heap where it previously returned quickly. Rows are now collected into a small batch, the filter is evaluated per batch, only matches are retained, and the scan stops once `limit` matches have been found. Heap is bounded by the batch plus the matches regardless of how much the log files hold. applyFilter is a passthrough when no filter is set, so a batch of one reproduces the previous bound exactly. Also pull the repeated optional-limit resolution into BaseProcedure.resolveLimit.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR makes the show_* procedures apply filter before limit so limit bounds the matching rows, and adds an incremental batch-filter path for show_logfile_records. The remaining question in the inline comment is about the unbounded archived-timeline load that a filter now triggers in show_timeline/show_cleans. 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. A couple of small simplification/dedup suggestions below; otherwise the refactor reads well.
| applyFilter(timelineEntries, filter, outputType) | ||
| // getTimelineEntries applies `limit` while collecting, so with a filter it has to be lifted here and | ||
| // reapplied to the matching rows; otherwise the filter only ever sees the first `limit` instants. | ||
| val scanLimit = if (hasFilter(filter)) Int.MaxValue else limit |
There was a problem hiding this comment.
🤖 With showArchived => true and any filter, scanLimit = Int.MaxValue now flows into loadCompletedInstantDetailsInMemory / loadCompactionDetailsInMemory, so the whole archived timeline (with commit metadata, in METADATA load mode) is materialised on the driver before the filter runs. For a long-lived table that's a lot more than the previous limit instants. Is that acceptable here, or would the same incremental approach you used for show_logfile_records (scan newest-first in chunks until limit matches are found) be worth applying? Same shape in ShowCleansProcedure where every archived clean's metadata is now read.
| (activeResults ++ archivedResults) | ||
| .sortWith((a, b) => a.getString(0) > b.getString(0)) | ||
| .take(limit) | ||
| .take(scanLimit) |
There was a problem hiding this comment.
🤖 nit: this .take(scanLimit) looks redundant now — applyFilterAndLimit already applies limit below, and when scanLimit is Int.MaxValue this is a no-op. Could we drop it?
| val limitValue = resolveLimit(limit) | ||
| // The bound has to come after the filter; limiting the DataFrame first would show the filter only | ||
| // the first `limit` rows of the ordering. | ||
| val results = if (limit.isDefined && !hasFilter(filter)) ordered.limit(limitValue).collect() else ordered.collect() |
There was a problem hiding this comment.
🤖 nit: this collect-all-if-filtered else limit-then-collect branch plus its comment is duplicated verbatim in ShowBootstrapMappingProcedure — worth a small helper in BaseProcedure (e.g. collectForFilter(df, filter, limit)) next to applyFilterAndLimit so the reasoning lives in one place?
This branch has not been deployed
4 participants
Describe the issue this Pull Request addresses
Closes #19862.
Every
show_*procedure that accepts bothlimitandfiltertruncated its rows tolimitfirst and evaluatedfilteron the truncated list, solimit => 10, filter => ...returned the matching subset of the first 10 rows rather than the first 10 matching rows. With a selective filter the result is empty whenever the matches sit past the cut-off, and since every one of these procedures applies a defaultlimit(10, 20 or 100), a plainfilter => ...call with no explicit limit was affected too.Introduced alongside the generic filter option in #13736 / #13790.
Summary and Changelog
BaseProceduregainsapplyFilterAndLimit(results, filter, schema, limit), which filters and then truncates, plushasFilterandresolveLimithelpers (the latter turning an optionallimitargument into a bound, where "unset" means no bound rather than zero). The ten affected procedures now call it. They are not all the same shape, and the difference is the substance of the change:Truncating an already-materialised list.
ShowTablePropertiesProcedure,ShowFileSystemViewProcedure,ShowHoodieLogFileMetadataProcedureandShowMetadataTableFilesProcedurejust dropped the earlystream().limit(...); the rows were already in memory, so nothing about the work done changes.Bounding at the source.
ShowBootstrapMappingProcedureandShowFsPathDetailProcedureuseddf.orderBy(...).limit(n).collect(), andShowInvalidParquetProcedureusedrdd.take(n). The bound now moves after the filter. To keep the unfiltered path exactly as it was, the early bound is only skipped when a filter is actually present.Bounding real work. In
ShowCleansProcedure,ShowTimelineProcedureandShowHoodieLogFileRecordsProcedure,limitis not display truncation: it caps how many clean instants get their metadata read and how many records are pulled out of log files. Filtering first without care would make those unbounded, turning a correctness fix into a performance regression.ShowCleansProcedureandShowTimelineProcedureusescanLimit = if (hasFilter(filter)) Int.MaxValue else limit, so a call with no filter does exactly the work it did before and only the filtered path reads further.limitis never used to size an allocation on those paths, soInt.MaxValueis only ever atakebound or a loop comparison.ShowHoodieLogFileRecordsProcedureneeded more than that, because lifting its bound would buffer every record of every matched log file into an on-heap list and could OOM the driver on a large file group. It instead filters incrementally: rows are collected into a small batch, the filter is evaluated per batch, only the matches are retained, and the scan stops as soon aslimitmatches have been found. Heap stays bounded by the batch plus the matches no matter how much the log files hold. With no filterapplyFilteris a passthrough, so a batch of one reproduces the previous bound exactly.ShowTimelineProcedureneeds one extra condition.getTimelineEntriesdeliberately ignoreslimitwhen bothstartTimeandendTimeare given ("Apply limit only if time range is not fully specified"), so reapplying the bound unconditionally would have truncated a fully specified range to the default of 20. The reapplied bound honours that case.The six other procedures that call
applyFilter(ShowBootstrapPartitionsProcedure,ShowColumnStatsOverlapProcedure,ShowFileStatusProcedure,ShowMetadataTableColumnStatsProcedure,ShowMetadataTablePartitionsProcedure,ShowMetadataTableStatsProcedure) take nolimitat all and are untouched, so the list of ten in the issue is complete.Verification
Three regression tests, each confirmed to fail without the corresponding fix.
TestShowCleansProcedures: three cleans are created, then the oldest is requested withlimit => 1. Cleans come back newest first, so the match sits past a limit of 1.TestHoodieLogFileProcedure: with two records in the log files, each is requested withlimit => 1and a filter naming it. Asserting both keeps this independent of scan order, since whichever record is not first is invisible to a filter applied after a limit of 1. Bounding the scan by rows seen rather than rows matched fails it:TestShowTimelineTableProcedureTest Case 13, which runs across all four existing variants (V1/V2 x COW/MOR): a fully specified time range withlimit => 1must not be truncated. Reintroducing the unconditional bound fails it everywhere:Full procedure package on Spark 3.5 / Scala 2.12:
Impact
limit => n, filter => ...now returns the first n matching rows instead of the matches among the first n rows. Calls that pass no filter are unaffected, including the work they do: the early bound is only lifted when a filter is present.One behaviour change worth calling out: with a filter,
show_cleansreads the metadata of every clean rather than only the newestlimit, andshow_timelinereads the whole timeline rather than the firstlimitinstants. That is required for the filter to see the rows it is supposed to match, it is bounded by the size of the timeline, and the unfiltered path, which is the common one, is unchanged.show_logfile_recordsis deliberately not in that list. An earlier revision of this PR did lift its bound the same way, and this description said so; that was wrong, because the records it reads are unbounded by anything the table caps, so a plainfilter => ...on a large file group could have exhausted the driver heap where it previously returned quickly. It now filters incrementally and stops atlimitmatches, so it stays bounded.Risk Level
low: the change is confined to how
limitandfiltercompose inside these procedures. No API, config, or table format change, and no engine or table-type specific behaviour is involved.Documentation Update
none: this restores the documented meaning of
limitcombined withfilterrather than changing it.Contributor's checklist