[feature](lance) push COUNT(*) down to Lance dataset metadata - #66999
[feature](lance) push COUNT(*) down to Lance dataset metadata#66999Jay-ju wants to merge 2 commits into
Conversation
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
bf80fe0 to
7909d96
Compare
|
/review |
There was a problem hiding this comment.
Request changes: the new Lance metadata-count path has one snapshot-correctness blocker and one material large-table performance regression.
Critical checkpoints:
- Snapshot correctness and mixed-version compatibility: the FE count belongs to the relation's fixed MVCC snapshot, but the replacement range serializes version 0/latest. An old BE or a current correctness fallback can therefore scan a different snapshot.
- Execution and performance: the BE count contract materializes one synthetic input row per counted row. Returning one range serializes that O(rowCount) work on one scanner instead of preserving the prior fragment parallelism.
- Layer contracts and lifecycle: FE planning, Thrift serialization, TableReader activation, Lance reader reset, zero/exact-batch EOF, cancellation, and subsequent-split behavior were traced end to end. No additional ownership or lifecycle defect was found.
- Predicates and deletion semantics: COUNT argument identity, pushed/residual predicates, runtime-filter gates, and logical post-deletion row counts remain aligned; the fixture distinguishes 27 logical rows from 30 physical rows.
- Tests: the added latest-snapshot fixture coverage is useful, but it does not cover time-travel/mixed-version fallback or representative large counts and parallel scan ranges. Those gaps correspond to the two inline findings.
- User focus: no additional user-provided focus was supplied.
- Completion: two full review rounds converged; all Round 2 agents returned NO_NEW_VALUABLE_FINDINGS after fencing these two accepted issues.
No builds or tests were run in this review runner, as required by the task instructions.
| // enough: the metadata lookup is O(1) and needs no parallelism. | ||
| long rowCount = metadata.getRowCount(); | ||
| setPushDownCount(rowCount); | ||
| LanceSplit countSplit = LanceSplit.wholeDatasetAtLatest(metadata.getDatasetUri()); |
There was a problem hiding this comment.
[P1] Keep the metadata-count fallback on the planned snapshot
metadata is the relation's fixed MVCC snapshot, but this factory hard-codes version 0 (latest). In a rolling upgrade the base-sha BE ignores table_level_row_count (the removed _remaining_table_level_count = -1 path) and opens this no-fragment split as a whole-dataset scan; the current BE can do the same whenever its metadata-count gate declines the shortcut. A COUNT(*) ... FOR VERSION AS OF v can therefore read latest instead of v (and an ordinary statement can drift after a concurrent commit). Please carry metadata.getVersion() in a fixed-version whole-dataset/count split and cover the fallback/time-travel case.
| setPushDownCount(rowCount); | ||
| LanceSplit countSplit = LanceSplit.wholeDatasetAtLatest(metadata.getDatasetUri()); | ||
| countSplit.setTableLevelRowCount(rowCount); | ||
| return Collections.singletonList(countSplit); |
There was a problem hiding this comment.
[P2] Preserve parallelism for large metadata counts
table_level_row_count is not consumed as one scalar: BE's _read_count_batch() keeps materializing default rows in runtime-sized batches until all rowCount rows have passed through the unchanged upper COUNT. Returning one range also caps scanner concurrency to one, so a large Lance table moves this O(N) work onto a single scanner even though its former fragment splits ran in parallel. Iceberg and Paimon distribute counts above 10,000 across parallelExecInstanceNum * numBackends carriers for this execution model. Please preserve parallelism with fallback-safe fixed-version carriers (not cloned unrestricted whole-dataset ranges, which a fallback BE would scan repeatedly), or change the execution contract to consume the count in O(1), and add a large-count test.
7909d96 to
674c18b
Compare
|
Thanks for the thorough review. Both findings were real; fixed in the latest revision (single commit, force-pushed). [P1] Snapshot correctness — fixed. The count carrier no longer serializes version 0/latest. It now pins the planned MVCC version via a new [P2] Large-count parallelism — fixed. Since BE materializes one synthetic row per counted row, Verified end to end on a real FE+BE+MinIO cluster: One note on the large-count test: I kept the committed fixtures lightweight rather than adding a >=10000-row binary dataset just to cross the parallel-split threshold. The sharding is pure FE arithmetic over the existing multi-split BE contract that Iceberg/Paimon already exercise at scale, and the small-table path (single carrier) is covered by the suite. Happy to add a large fixture if you'd prefer explicit end-to-end coverage of the parallel path. |
|
Re: target branch — why this sits on branch-4.1 and not master A heads-up on branch choice, since I initially assumed branch-4.1 was simply "a bit ahead" of master. After checking, the whole Lance integration currently lives only on branch-4.1, not master:
So branch-4.1 is intentionally ahead of master for Lance. This PR depends on that branch-4.1-only base ( Question for the maintainers: is Lance planned to be forward-ported to master? If so, I'm happy to prepare master PRs in dependency order for my changes (#66597 -> #66608 -> #66999) once the base integration (#65730 etc.) lands there. Please advise on the preferred branch strategy. |
|
run buildall |
Gabriel39
left a comment
There was a problem hiding this comment.
Request changes: the large-count sharding is not fallback-safe. Each carrier is an unrestricted whole-dataset range, so a rolling-upgrade BE (or any current BE that declines the metadata shortcut) scans the full dataset once per carrier and overcounts. The snapshot-version test also does not inspect the serialized split version.
| long carriedRows = (rowCount - assigned) / (carrierCount - i); | ||
| assigned += carriedRows; | ||
| LanceSplit countSplit = LanceSplit.wholeDatasetCountAtVersion( | ||
| metadata.getDatasetUri(), metadata.getVersion(), carriedRows); |
There was a problem hiding this comment.
[P1] Keep sharded carriers fallback-disjoint
For rowCount >= 10000, this loop creates multiple splits with no fragment IDs. A new BE consumes each table_level_row_count share, but the pre-change BE explicitly resets that count to -1; Lance then interprets the missing fragment IDs as an unrestricted whole-dataset scan. During a rolling upgrade, every carrier assigned to an old BE therefore scans the entire snapshot, producing up to carrierCount * rowCount rows (and a current BE can take the same path whenever its metadata-count gate declines the shortcut). Please make the fallback inputs disjoint—e.g. carriers whose fragment groups and logical counts match—or keep a single carrier until compatibility guarantees the scalar path, and cover the mixed-version/fallback case.
There was a problem hiding this comment.
Fixed in 707e2ae. Metadata-count carriers now contain disjoint fragment groups and each carrier carries that group’s exact logical row count. If a BE declines the shortcut, the fallback scans each fragment exactly once at the pinned snapshot while retaining parallel carriers for large counts.
| // "lanceVersion=0" here and let time-travel / concurrent-commit reads drift. | ||
| explain { | ||
| sql(mfCountStar) | ||
| contains "lanceVersion=4" |
There was a problem hiding this comment.
[P2] Assert the serialized split version, not the planner field
getNodeExplainString() prints plannedVersion, which is assigned directly from metadata.getVersion() before buildCountSplits() runs. Consequently this assertion still reports lanceVersion=4 if the carrier factory regresses to version 0; it does not test the TLanceFileDesc.version consumed by the fallback BE. Please inspect the generated range descriptor in an FE test or execute a fixed-version query through a forced fallback path so this regression is actually covered.
There was a problem hiding this comment.
Fixed in 707e2ae. Added LanceScanNodeTest coverage that serializes each count split into TFileRangeDesc and directly asserts TLanceFileDesc.version=42, the disjoint fragment IDs, and table_level_row_count.
|
Thanks for the update. One correctness blocker remains in the large-count path: every sharded carrier is still an unrestricted whole-dataset split. A pre-change BE ignores |
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
lance catalog seems no need to consider upgrades problem now. |
COUNT(*)/COUNT(1) with no filter can be answered from the Lance dataset's logical (post-deletion) row count instead of scanning any fragment. FE (LanceScanNode): add canPushDownCountStar(), which is stricter than the LIMIT pushdown gate -- it requires both an empty conjunct list and an empty Lance Substrait filter, since any predicate would make the dataset-wide row count larger than the real result. When it holds, emit whole-dataset count carriers holding the logical row count. Each carrier is pinned to the planned MVCC version (not latest) so a fallback scan -- an old BE, or a BE that declines the shortcut -- reads exactly the snapshot the count came from instead of drifting to latest on a time-travel or concurrent-commit read. Because BE materializes one synthetic row per counted row, a count at or above COUNT_WITH_PARALLEL_SPLITS is spread over parallelExecInstanceNum * numBackends carriers (shares summing back to the exact total) to keep the former fragment parallelism, mirroring IcebergScanNode; a small count stays on one carrier. table_level_row_count is now always set explicitly, -1 for ordinary and search scans, matching the Iceberg convention so BE never mistakes a stale value for a metadata count. BE (lance_reader): drop the hardcoded _remaining_table_level_count = -1 that unconditionally disabled the base-class count path, and short-circuit both prepare_split() and get_block() when _is_table_level_count_active() so the counted rows are synthesized without opening a scanner. Tests: add test_lance_optimize_count asserting EXPLAIN shows the metadata count with no filter (and that the carrier pins the planned dataset version), and falls back to a normal scan (with matching results) when a filter is present or the switch is off. Add the multi_frag.lance fixture (three fragments, one deleted row each: 30 physical / 27 logical rows) plus its build/self-check in the preinstalled catalog script, which proves the count reports the logical total and that a multi-split scan applies every fragment's deletion vector exactly once.
674c18b to
707e2ae
Compare
|
/review |
Summary
Tests