[feature](lance) push COUNT(*) down to Lance dataset metadata

[Jay-ju]

Summary

• Answer COUNT(*)/COUNT(1) with no filter from the Lance dataset logical (post-deletion) row count instead of scanning any fragment.
• FE (LanceScanNode): add canPushDownCountStar(), stricter than the LIMIT pushdown gate (empty conjuncts AND empty Lance Substrait filter). Emit a single whole-dataset split carrying the logical row count; table_level_row_count is always set explicitly (-1 for ordinary/search scans), matching the Iceberg convention.
• BE (lance_reader): drop the hardcoded _remaining_table_level_count = -1 and short-circuit prepare_split()/get_block() when _is_table_level_count_active().

Tests

• New test_lance_optimize_count asserts EXPLAIN shows the metadata count with no filter and falls back to a normal scan (matching results) with a filter or when the switch is off.
• New multi_frag.lance fixture (3 fragments, one deleted row each: 30 physical / 27 logical) plus build/self-check; proves the count reports the logical total and that a multi-split scan applies each fragment deletion vector exactly once. Committed as binary, consistent with existing all_types.lance / iceberg preinstalled data.

[hello-stephen]

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?

[HappenLee]

/review

[github-actions]

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.

[github-actions]

[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.

[github-actions]

[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.

[Jay-ju]

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 LanceSplit.wholeDatasetCountAtVersion(uri, version, carriedRows) factory, so a fallback scan (an old BE, or a current BE that declines the shortcut and opens the whole-dataset range in _ensure_dataset_open) reads exactly the snapshot the count was computed from. wholeDatasetAtLatest is retained only for the backend-local TVF, which is correctly latest. Added an EXPLAIN assertion (lanceVersion=4 on the multi_frag fixture, built with three appends + three deletes) that fails if this ever regresses to latest.

[P2] Large-count parallelism — fixed. Since BE materializes one synthetic row per counted row, buildCountSplits() now shards a count >= COUNT_WITH_PARALLEL_SPLITS (10000, matching Iceberg) across parallelExecInstanceNum * numBackends fixed-version carriers, with per-carrier shares summing back to the exact total (remainder folded into the last carrier). This mirrors IcebergScanNode.assignCountToSplits and reuses the existing multi-split table_level_row_count contract, so BE needs no change. A small count stays on one carrier for backward compatibility.

Verified end to end on a real FE+BE+MinIO cluster: count(*) returns the logical 27 (not physical 30) for multi_frag and 12 for all_types, EXPLAIN shows pushdown agg=COUNT (27) with lanceVersion=4, and run-regression-test test_lance_optimize_count passes (All suites success, 0 failed).

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.

[Jay-ju]

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:

• Lance was first merged to master via [feature](lance) Add Rust-based Lance format reader for AI-native dat… #62182, then reverted from master by Revert "[feature](lance) Add Rust-based Lance format reader for AI-native data (#62182) #65303 (and the lance-c thirdparty in [chore](thirdparty) introduce thridparty about lance-c #65304 was likewise reverted).
• The integration was then re-established on branch-4.1 starting from [feature](lance) support lance catalog in doris #65730, and every Lance PR since (including [refactor](lance) improve lance catalog implementation in FE #66581, [improvement](lance) Balance fragment splits by row count #66597, [opt](lance) push down LIMIT into Lance fragment scanners #66608, [opt](lance) Support distributed execution for vector_search #66841, and this one) has targeted branch-4.1.

So branch-4.1 is intentionally ahead of master for Lance. This PR depends on that branch-4.1-only base (LanceScanNode / LanceSplit / lance_reader), which is why it targets branch-4.1.

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.

[zhangstar333]

run buildall

[Gabriel39]

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.

[Gabriel39]

[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.

[Jay-ju]

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.

[Gabriel39]

[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.

[Jay-ju]

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.

[Gabriel39]

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 table_level_row_count and scans that whole snapshot, so rolling upgrades can count the dataset once per carrier; a current BE that declines the metadata shortcut can take the same fallback path. Please make carrier fallback ranges disjoint (for example, group fragment IDs and attach the matching logical count), or retain a single carrier until the scalar path is compatibility-safe. Also, the lanceVersion=4 EXPLAIN assertion only checks plannedVersion; it does not verify the version serialized in TLanceFileDesc, so the fixed-version fallback needs a direct range-descriptor or forced-fallback test.

[hello-stephen]

BE Regression && UT Coverage Report

Increment line coverage 33.33% (2/6) 🎉

Increment coverage report
Complete coverage report

Category	Coverage
Function Coverage	74.06% (31030/41901)
Line Coverage	58.19% (343565/590386)
Region Coverage	55.09% (286542/520096)
Branch Coverage	55.92% (128909/230510)

[zhangstar333]

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 table_level_row_count and scans that whole snapshot, so rolling upgrades can count the dataset once per carrier; a current BE that declines the metadata shortcut can take the same fallback path. Please make carrier fallback ranges disjoint (for example, group fragment IDs and attach the matching logical count), or retain a single carrier until the scalar path is compatibility-safe. Also, the lanceVersion=4 EXPLAIN assertion only checks plannedVersion; it does not verify the version serialized in TLanceFileDesc, so the fixed-version fallback needs a direct range-descriptor or forced-fallback test.

lance catalog seems no need to consider upgrades problem now.

[Jay-ju]

/review