Conversation
jerry-024
changed the title
dcb4e01 to
d1c8eab
Compare
b90caee to
764df07
Compare
jerry-024
changed the title
Code Review SummaryMode: full Critical Issues (must fix before merge)None found. Major Issues (should fix)
Minor Issues
SuggestionsNone beyond the fixes above. Positive Observations
|
| } | ||
| columns.push(Arc::new(Int64Array::from_iter_values( | ||
| counts.iter().filter_map(|count| count.record_count), | ||
| ))); |
There was a problem hiding this comment.
[major][logic-1] Known-zero partitions create groups that do not exist in the table
counts_to_batch materializes every PartitionRowCount, including Some(0). If deletion vectors remove every row from a partition, a normal scan supplies no input row to GROUP BY, but this rewritten plan supplies a synthetic row and returns (partition, 0).
Please filter known-zero counts before constructing the batch while retaining None so unknown counts still trigger fallback. The ungrouped empty-input case can continue to use the existing coalesce-to-zero path.
| table.partition_row_counts_with_filter(predicate).await | ||
| }) | ||
| .await | ||
| .map_err(to_datafusion_error)?; |
There was a problem hiding this comment.
[major][logic-2] Pin the snapshot when the physical plan is created
The ordinary Paimon provider plans splits during TableProvider::scan, which fixes the snapshot represented by the physical plan. This path stores only a cloned Table and resolves latest inside execution, so a commit between physical planning and collection can change the result; re-executing the same physical plan can change it again.
Please resolve and store the selected snapshot in PartitionRowCountProvider::scan. Manifest I/O can remain lazy, but snapshot selection should not.
| Arc::clone(&self.output_schema), | ||
| Box::pin(stream::iter(streams).flatten()), | ||
| ))); | ||
| } |
There was a problem hiding this comment.
[major][perf-1] One unknown count discards every exact manifest count
After all selected manifests have been read and aggregated, any None sends the complete selected table through scan_by_reading. A single legacy deletion vector without cardinality therefore adds a full metadata pass before a full data scan, while exact counts already computed for other partitions are discarded.
Please retain known partition counts and restrict fallback to unknown full partition keys. At minimum, detect an unavoidable unknown-cardinality fallback before doing redundant manifest aggregation.
There was a problem hiding this comment.
Thanks. This is an intentional compatibility fallback for deletion vectors without cardinality. It preserves correctness and falls back to the same pinned ordinary scan used before this optimization, so it is not a performance regression from the existing path. A partial fallback would require a mixed manifest/data plan and merge semantics; we would only add that complexity if measurements show this legacy case is common.
There was a problem hiding this comment.
Rechecked on current HEAD 576441f. The minimum requested part is now addressed: read_partition_row_counts checks unknown DV cardinality and returns before loading the base/delta data manifests, so there is no redundant manifest aggregation before fallback. The remaining ordinary scan is the intentional exactness fallback; a mixed partial-fallback plan is deferred unless measurements show this legacy case is common.
| match &self.projection { | ||
| Some(projection) => Ok(batch.project(projection)?), | ||
| None => Ok(batch), | ||
| } |
There was a problem hiding this comment.
[major][perf-3] Apply projection before decoding partition columns
counts_to_batch decodes every partition field and builds every Arrow array before applying batch.project(projection). Ungrouped COUNT(*) needs only the row-count column, and subset grouping needs only selected partition fields. At the high partition cardinalities this optimization targets, the discarded arrays add substantial avoidable CPU and peak memory.
Please construct only the projected source columns and build the output-schema RecordBatch directly.
| .map(|value| Box::from(*value)) | ||
| .collect(), | ||
| embedded_index: entry.embedded_index.map(Box::from), | ||
| external_path: entry.external_path.map(Box::from), |
There was a problem hiding this comment.
[major][perf-4] The retained-entry budget is not a memory bound
The budget counts ADD entries, but each retained identity clones variable-sized extra_files, embedded_index, and external_path; DELETE identities have no count or byte budget. Delete-heavy manifests or large embedded indexes can therefore still consume hundreds of MB or more and defeat the OOM-avoidance goal.
Please account retained state by estimated heap bytes and spill exact identities after a configurable byte limit, applying the same policy to DELETE identities.
There was a problem hiding this comment.
Thanks. The limit is intentionally entry-based, not a strict byte cap. The PR description documents this trade-off, including the measured RSS increase for retained ADDs and the fact that exact DELETE/embedded-index payloads remain variable-sized. Exact matching is required; byte accounting plus spill would be a separate, substantially larger mechanism. We will keep the current design unless production measurements show this state is the actual memory bottleneck.
764df07 to
b2cbc62
Compare
Incremental Code Review SummaryMode: full incremental review Resolution of Previous Findings
Critical Issues (must fix before merge)None found. Major Issues (should fix)
Minor Issues
Positive Observations
|
b2cbc62 to
9932ceb
Compare
Match deletion vectors to live partition/bucket/file identities, skip rewrites on internal column-name collisions, and fall back before data-manifest aggregation when cardinalities are unknown. Add correctness and I/O regression coverage.
2 participants
Purpose
Linked issue: closes #859
Queries such as:
currently plan every live file and read data files even when grouping and filtering use only partition columns. On tables with millions of files, materializing full manifest entries and column statistics can also exhaust planner memory.
This PR adds an exact manifest-only path for partition-grouped
COUNT(*). It preserves correctness for data-evolution files and deletion vectors, and falls back to the ordinary scan whenever metadata cannot provide an exact count.Brief change log
[2, 32], without introducing a second buffered queue.Table::partition_row_counts{,_with_filter}andPartitionRowCount.COUNT(*)/COUNT(non-null literal)grouped by partition columns.PartitionRowCountExec;EXPLAINperforms no manifest I/O.Known trade-offs and follow-ups
BTreeMapstate and serial merge work. A sorted-vector merge or partition-sharded merge is a possible follow-up.Tests
A controlled release-mode pipeline benchmark on an 8-core host repeatedly read and slim-decoded the same cached 6.67 MB manifest 256 times. Across five fresh processes, replacing the two buffered stages with one 32-item pipeline reduced median peak RSS from 370,588 KiB to 352,196 KiB (-5.0%) and median measured execution time from 1,111 ms to 879 ms (-20.9%). This isolates pipeline buffering; it is not an object-store or full-table benchmark.
A second release-mode benchmark repeated the same cached 3.05 MB DELETE-bearing manifest as 900 inputs, with 1,000 ADDs per input. Across three fresh processes, retaining the 900,000 ADD identities reduced median execution time from 1,851 ms to 1,145 ms (-38.1%) versus a forced zero budget, while median peak RSS increased from 234,352 KiB to 321,444 KiB (+37.2%). This is a controlled cache/decode trade-off measurement, not a full-table benchmark.
Coverage includes:
IN,HAVING, ordering, empty results, and ungrouped counts;EXPLAINwithout manifest reads, and execution-time fallback for unknown metadata.API and Format
Adds these Rust APIs:
No storage-format or SQL-syntax changes.
Documentation
No documentation update is required. The optimization is transparent to existing SQL queries.