Skip to content

Commit 7909d96

Browse files
committed
[feature](lance) push COUNT(*) down to Lance dataset metadata
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 a single whole-dataset split carrying the logical row count (one split is enough: the metadata lookup is O(1)). 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 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.
1 parent 970d951 commit 7909d96

21 files changed

Lines changed: 325 additions & 21 deletions

be/src/format_v2/table/lance_reader.cpp

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -368,12 +368,15 @@ Status LanceTableReader::prepare_split(const SplitReadOptions& options) {
368368
_eof = false;
369369

370370
RETURN_IF_ERROR(TableReader::prepare_split(options));
371-
// Lance does not currently provide metadata aggregate pushdown. Do not let a generic
372-
// table-level count supplied by a future planner bypass fragment reads.
373-
_remaining_table_level_count = -1;
374371
if (current_split_pruned()) {
375372
return Status::OK();
376373
}
374+
// COUNT(*)/COUNT(1) with no filter is served from Lance metadata. The base class already set
375+
// _remaining_table_level_count from the split's table_level_row_count, so skip opening any
376+
// dataset scanner; get_block() synthesizes the counted rows.
377+
if (_is_table_level_count_active()) {
378+
return Status::OK();
379+
}
377380
if (_global_rowid_output_idx.has_value() && !_global_rowid_context.has_value()) {
378381
return Status::InvalidArgument(
379382
"Lance global row id requested without global row id context");
@@ -394,6 +397,11 @@ Status LanceTableReader::get_block(Block* block, bool* eos) {
394397
*eos = true;
395398
return Status::OK();
396399
}
400+
// Metadata COUNT(*) split: no scanner is opened. Emit synthetic rows for the upper COUNT
401+
// operator directly from the row count the base class parsed out of the split.
402+
if (_is_table_level_count_active()) {
403+
return _read_table_level_count(block, eos);
404+
}
397405
if (_scanner == nullptr) {
398406
return Status::InternalError("Lance scanner is not initialized for the current split");
399407
}

docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,25 @@
9999
# degradation - the regression suite pins that error too.
100100
HNSW_SEARCH_PARAMS = {"ef": 100}
101101

102+
# multi_frag.lance is the COUNT(*) metadata-pushdown fixture for test_lance_optimize_count:
103+
# MULTI_FRAG_NUM_FRAGMENTS fragments of MULTI_FRAG_FRAGMENT_ROWS physical rows each, with one
104+
# deleted row per fragment, so the dataset holds MULTI_FRAG_PHYSICAL_ROWS physical rows on disk
105+
# but only MULTI_FRAG_LOGICAL_ROWS logical rows after deletions. A COUNT(*) that reported the
106+
# physical total would be off by MULTI_FRAG_DELETED_ROWS, so this table is what proves the
107+
# pushdown reads Lance's post-deletion row count and that a multi-split scan applies every
108+
# fragment's deletion vector exactly once. It carries no index, so unlike the vector tables its
109+
# data and every derived count are deterministic (there is no IVF training to perturb them and
110+
# no golden ever shifts on regeneration), and Doris discovers it by directory listing without a
111+
# __manifest entry (verified against a live FE/BE/MinIO cluster).
112+
MULTI_FRAG_DIR = "multi_frag.lance"
113+
MULTI_FRAG_NUM_FRAGMENTS = 3
114+
MULTI_FRAG_FRAGMENT_ROWS = 10
115+
MULTI_FRAG_DELETED_ROW_IDS = (5, 15, 25)
116+
MULTI_FRAG_FILTER_ROW_ID = 15
117+
MULTI_FRAG_PHYSICAL_ROWS = MULTI_FRAG_NUM_FRAGMENTS * MULTI_FRAG_FRAGMENT_ROWS
118+
MULTI_FRAG_DELETED_ROWS = len(MULTI_FRAG_DELETED_ROW_IDS)
119+
MULTI_FRAG_LOGICAL_ROWS = MULTI_FRAG_PHYSICAL_ROWS - MULTI_FRAG_DELETED_ROWS
120+
102121
# One table per ANN algorithm and element type, identical data, exactly one index named
103122
# embedding_<the table name without its vs_ prefix>. Naming is vs_<algorithm>_<element
104123
# type>, so one table is exactly one cell of the algorithm x element type matrix and a
@@ -239,8 +258,27 @@ def compact_manifest(root: Path) -> None:
239258
hint.write_text(f'{{"version":{manifest.version}}}')
240259

241260

261+
def build_multi_frag(root: Path) -> None:
262+
# Reuse make_fragment_table so the row_id/category/label columns and their NOT NULL mapping
263+
# stay identical to the vector tables; multi_frag just drops the embedding it does not need.
264+
location = str(root / MULTI_FRAG_DIR)
265+
for index in range(MULTI_FRAG_NUM_FRAGMENTS):
266+
offset = index * MULTI_FRAG_FRAGMENT_ROWS
267+
fragment = make_fragment_table(offset, offset + MULTI_FRAG_FRAGMENT_ROWS)
268+
fragment = fragment.drop_columns(["embedding"])
269+
# Match all_types.lance (data storage version 2.2) so every committed Lance data file
270+
# shares one on-disk format and the oldest reader (lance-rs 4.0.1) can open it.
271+
lance.write_dataset(
272+
fragment, location, mode="create" if index == 0 else "append",
273+
data_storage_version="2.2",
274+
)
275+
deleted = ", ".join(str(row_id) for row_id in MULTI_FRAG_DELETED_ROW_IDS)
276+
lance.dataset(location).delete(f"row_id in ({deleted})")
277+
278+
242279
def build(root: Path, all_types_source: Path) -> None:
243280
shutil.copytree(all_types_source, root / ALL_TYPES_DIR)
281+
build_multi_frag(root)
244282
namespace = lance_namespace.connect("dir", {"root": str(root)})
245283
namespace.register_table(
246284
RegisterTableRequest(id=["all_types"], location=ALL_TYPES_DIR)
@@ -431,6 +469,44 @@ def check_ef_discriminator(name: str, dataset, assert_it: bool) -> None:
431469
f"rows {[row for row, _ in wide]} (differs={differs}, asserted={assert_it})")
432470

433471

472+
def check_multi_frag(root: Path) -> None:
473+
location = root / MULTI_FRAG_DIR
474+
assert location.is_dir(), f"multi_frag location missing: {location}"
475+
dataset = lance.dataset(str(location))
476+
fragments = dataset.get_fragments()
477+
assert len(fragments) == MULTI_FRAG_NUM_FRAGMENTS, (
478+
f"multi_frag: expected {MULTI_FRAG_NUM_FRAGMENTS} fragments, got {len(fragments)}"
479+
)
480+
for fragment in fragments:
481+
metadata = fragment.metadata
482+
assert metadata.physical_rows == MULTI_FRAG_FRAGMENT_ROWS, (
483+
f"multi_frag fragment {fragment.fragment_id}: physical_rows "
484+
f"{metadata.physical_rows} != {MULTI_FRAG_FRAGMENT_ROWS}"
485+
)
486+
assert metadata.num_deletions == 1, (
487+
f"multi_frag fragment {fragment.fragment_id}: expected exactly one deleted row, "
488+
f"got {metadata.num_deletions}"
489+
)
490+
# The whole point of this table: logical (post-deletion) count, not the physical total.
491+
assert dataset.count_rows() == MULTI_FRAG_LOGICAL_ROWS, (
492+
f"multi_frag: logical row count {dataset.count_rows()} != {MULTI_FRAG_LOGICAL_ROWS}; "
493+
"test_lance_optimize_count asserts COUNT(*) folds to exactly this number"
494+
)
495+
surviving = set(dataset.to_table(columns=["row_id"]).column("row_id").to_pylist())
496+
expected = set(range(1, MULTI_FRAG_PHYSICAL_ROWS + 1)) - set(MULTI_FRAG_DELETED_ROW_IDS)
497+
assert surviving == expected, (
498+
"multi_frag: surviving row_ids are not the expected contiguous-minus-deleted set"
499+
)
500+
# The filtered count in the suite disables the pushdown; keep its golden derivable here so a
501+
# data-shape change fails this self-check instead of only the opaque .out diff.
502+
expected_half = sum(1 for row_id in expected if row_id > MULTI_FRAG_FILTER_ROW_ID)
503+
half = dataset.count_rows(filter=f"row_id > {MULTI_FRAG_FILTER_ROW_ID}")
504+
assert half == expected_half, (
505+
f"multi_frag: COUNT(*) WHERE row_id > {MULTI_FRAG_FILTER_ROW_ID} is {half}, not "
506+
f"{expected_half}; the filtered-count golden in test_lance_optimize_count is now stale"
507+
)
508+
509+
434510
def check_catalog(root: Path) -> None:
435511
namespace = lance_namespace.connect("dir", {"root": str(root)})
436512
tables = namespace.list_tables(ListTablesRequest(id=[NAMESPACE]))
@@ -478,6 +554,7 @@ def check_catalog(root: Path) -> None:
478554
check_boundary_discriminator(table_name, dataset, search)
479555
if search.get("ef"):
480556
check_ef_discriminator(table_name, dataset, spec.get("ef_discriminator", False))
557+
check_multi_frag(root)
481558
print(f"self-check OK: {root}")
482559

483560

0 commit comments

Comments
 (0)