diff --git a/be/src/format_v2/table/lance_reader.cpp b/be/src/format_v2/table/lance_reader.cpp index e8ad0c5229aca3..860623613fda37 100644 --- a/be/src/format_v2/table/lance_reader.cpp +++ b/be/src/format_v2/table/lance_reader.cpp @@ -431,12 +431,15 @@ Status LanceTableReader::prepare_split(const SplitReadOptions& options) { _eof = false; RETURN_IF_ERROR(TableReader::prepare_split(options)); - // Lance does not currently provide metadata aggregate pushdown. Do not let a generic - // table-level count supplied by a future planner bypass fragment reads. - _remaining_table_level_count = -1; if (current_split_pruned()) { return Status::OK(); } + // COUNT(*)/COUNT(1) with no filter is served from Lance metadata. The base class already set + // _remaining_table_level_count from the split's table_level_row_count, so skip opening any + // dataset scanner; get_block() synthesizes the counted rows. + if (_is_table_level_count_active()) { + return Status::OK(); + } if (_global_rowid_output_idx.has_value() && !_global_rowid_context.has_value()) { return Status::InvalidArgument( "Lance global row id requested without global row id context"); @@ -457,6 +460,11 @@ Status LanceTableReader::get_block(Block* block, bool* eos) { *eos = true; return Status::OK(); } + // Metadata COUNT(*) split: no scanner is opened. Emit synthetic rows for the upper COUNT + // operator directly from the row count the base class parsed out of the split. + if (_is_table_level_count_active()) { + return _read_table_level_count(block, eos); + } if (_scanner == nullptr) { return Status::InternalError("Lance scanner is not initialized for the current split"); } diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py b/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py index 9721074094f4b3..fa1b7946d6934e 100644 --- a/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py +++ b/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py @@ -117,6 +117,25 @@ # degradation - the regression suite pins that error too. HNSW_SEARCH_PARAMS = {"ef": 100} +# multi_frag.lance is the COUNT(*) metadata-pushdown fixture for test_lance_optimize_count: +# MULTI_FRAG_NUM_FRAGMENTS fragments of MULTI_FRAG_FRAGMENT_ROWS physical rows each, with one +# deleted row per fragment, so the dataset holds MULTI_FRAG_PHYSICAL_ROWS physical rows on disk +# but only MULTI_FRAG_LOGICAL_ROWS logical rows after deletions. A COUNT(*) that reported the +# physical total would be off by MULTI_FRAG_DELETED_ROWS, so this table is what proves the +# pushdown reads Lance's post-deletion row count and that a multi-split scan applies every +# fragment's deletion vector exactly once. It carries no index, so unlike the vector tables its +# data and every derived count are deterministic (there is no IVF training to perturb them and +# no golden ever shifts on regeneration), and Doris discovers it by directory listing without a +# __manifest entry (verified against a live FE/BE/MinIO cluster). +MULTI_FRAG_DIR = "multi_frag.lance" +MULTI_FRAG_NUM_FRAGMENTS = 3 +MULTI_FRAG_FRAGMENT_ROWS = 10 +MULTI_FRAG_DELETED_ROW_IDS = (5, 15, 25) +MULTI_FRAG_FILTER_ROW_ID = 15 +MULTI_FRAG_PHYSICAL_ROWS = MULTI_FRAG_NUM_FRAGMENTS * MULTI_FRAG_FRAGMENT_ROWS +MULTI_FRAG_DELETED_ROWS = len(MULTI_FRAG_DELETED_ROW_IDS) +MULTI_FRAG_LOGICAL_ROWS = MULTI_FRAG_PHYSICAL_ROWS - MULTI_FRAG_DELETED_ROWS + # The boundary query is symmetric for the ladder profiles - rows r-d and r+d are # equidistant - so a top-k that lands mid-pair would pin an arbitrary choice of tie winner # in the goldens. 9 is the last cut that ends on a complete pair. This is the regression @@ -787,8 +806,27 @@ def compact_manifest(root: Path) -> None: print(f"record: __manifest committed at version {manifest.version}") +def build_multi_frag(root: Path) -> None: + # Reuse make_fragment_table so the row_id/category/label columns and their NOT NULL mapping + # stay identical to the vector tables; multi_frag just drops the embedding it does not need. + location = str(root / MULTI_FRAG_DIR) + for index in range(MULTI_FRAG_NUM_FRAGMENTS): + offset = index * MULTI_FRAG_FRAGMENT_ROWS + fragment = make_fragment_table(offset, offset + MULTI_FRAG_FRAGMENT_ROWS) + fragment = fragment.drop_columns(["embedding"]) + # Match all_types.lance (data storage version 2.2) so every committed Lance data file + # shares one on-disk format and the oldest reader (lance-rs 4.0.1) can open it. + lance.write_dataset( + fragment, location, mode="create" if index == 0 else "append", + data_storage_version="2.2", + ) + deleted = ", ".join(str(row_id) for row_id in MULTI_FRAG_DELETED_ROW_IDS) + lance.dataset(location).delete(f"row_id in ({deleted})") + + def build(root: Path, all_types_source: Path) -> None: shutil.copytree(all_types_source, root / ALL_TYPES_DIR) + build_multi_frag(root) namespace = lance_namespace.connect("dir", {"root": str(root)}) namespace.register_table( RegisterTableRequest(id=["all_types"], location=ALL_TYPES_DIR) @@ -1431,6 +1469,44 @@ def check_nested_dataset(location: str): assert probe == [7], f"{NESTED_TABLE}: BTREE probe returned {probe}" +def check_multi_frag(root: Path) -> None: + location = root / MULTI_FRAG_DIR + assert location.is_dir(), f"multi_frag location missing: {location}" + dataset = lance.dataset(str(location)) + fragments = dataset.get_fragments() + assert len(fragments) == MULTI_FRAG_NUM_FRAGMENTS, ( + f"multi_frag: expected {MULTI_FRAG_NUM_FRAGMENTS} fragments, got {len(fragments)}" + ) + for fragment in fragments: + metadata = fragment.metadata + assert metadata.physical_rows == MULTI_FRAG_FRAGMENT_ROWS, ( + f"multi_frag fragment {fragment.fragment_id}: physical_rows " + f"{metadata.physical_rows} != {MULTI_FRAG_FRAGMENT_ROWS}" + ) + assert metadata.num_deletions == 1, ( + f"multi_frag fragment {fragment.fragment_id}: expected exactly one deleted row, " + f"got {metadata.num_deletions}" + ) + # The whole point of this table: logical (post-deletion) count, not the physical total. + assert dataset.count_rows() == MULTI_FRAG_LOGICAL_ROWS, ( + f"multi_frag: logical row count {dataset.count_rows()} != {MULTI_FRAG_LOGICAL_ROWS}; " + "test_lance_optimize_count asserts COUNT(*) folds to exactly this number" + ) + surviving = set(dataset.to_table(columns=["row_id"]).column("row_id").to_pylist()) + expected = set(range(1, MULTI_FRAG_PHYSICAL_ROWS + 1)) - set(MULTI_FRAG_DELETED_ROW_IDS) + assert surviving == expected, ( + "multi_frag: surviving row_ids are not the expected contiguous-minus-deleted set" + ) + # The filtered count in the suite disables the pushdown; keep its golden derivable here so a + # data-shape change fails this self-check instead of only the opaque .out diff. + expected_half = sum(1 for row_id in expected if row_id > MULTI_FRAG_FILTER_ROW_ID) + half = dataset.count_rows(filter=f"row_id > {MULTI_FRAG_FILTER_ROW_ID}") + assert half == expected_half, ( + f"multi_frag: COUNT(*) WHERE row_id > {MULTI_FRAG_FILTER_ROW_ID} is {half}, not " + f"{expected_half}; the filtered-count golden in test_lance_optimize_count is now stale" + ) + + def check_catalog(root: Path) -> None: check_data_shapes() namespace = lance_namespace.connect("dir", {"root": str(root)}) @@ -1490,6 +1566,7 @@ def check_catalog(root: Path) -> None: nested_path = Path(nested.location.removeprefix("file://")) assert nested_path.is_dir(), f"{NESTED_TABLE} location missing: {nested.location}" check_nested_dataset(nested.location) + check_multi_frag(root) print(f"self-check OK: {root}") diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_deletions/0-3-12903222946140852451.arrow b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_deletions/0-3-12903222946140852451.arrow new file mode 100644 index 00000000000000..badfc10c3787ef Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_deletions/0-3-12903222946140852451.arrow differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_deletions/1-3-3997032015744280171.arrow b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_deletions/1-3-3997032015744280171.arrow new file mode 100644 index 00000000000000..badfc10c3787ef Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_deletions/1-3-3997032015744280171.arrow differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_deletions/2-3-16205658664475652884.arrow b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_deletions/2-3-16205658664475652884.arrow new file mode 100644 index 00000000000000..badfc10c3787ef Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_deletions/2-3-16205658664475652884.arrow differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/0-9d78d1d6-4352-45af-81d3-11147a3165c5.txn b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/0-9d78d1d6-4352-45af-81d3-11147a3165c5.txn new file mode 100644 index 00000000000000..70d9f843807671 Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/0-9d78d1d6-4352-45af-81d3-11147a3165c5.txn differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/1-4ba9d959-a46f-4ecc-bc97-edc7030c82ef.txn b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/1-4ba9d959-a46f-4ecc-bc97-edc7030c82ef.txn new file mode 100644 index 00000000000000..8867f8e63c263c Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/1-4ba9d959-a46f-4ecc-bc97-edc7030c82ef.txn differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/2-1b03ecda-1a9c-411c-b9f0-f91af2575aa7.txn b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/2-1b03ecda-1a9c-411c-b9f0-f91af2575aa7.txn new file mode 100644 index 00000000000000..930321f052e9b0 Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/2-1b03ecda-1a9c-411c-b9f0-f91af2575aa7.txn differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/3-4b3d53f1-4352-4d80-a8d9-312412d6e5c6.txn b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/3-4b3d53f1-4352-4d80-a8d9-312412d6e5c6.txn new file mode 100644 index 00000000000000..45da071eab93e3 Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_transactions/3-4b3d53f1-4352-4d80-a8d9-312412d6e5c6.txn differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551611.manifest b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551611.manifest new file mode 100644 index 00000000000000..0bdc388ef6f63d Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551611.manifest differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551612.manifest b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551612.manifest new file mode 100644 index 00000000000000..da761c20840a50 Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551612.manifest differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551613.manifest b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551613.manifest new file mode 100644 index 00000000000000..c2920b2c6cf5f0 Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551613.manifest differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551614.manifest b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551614.manifest new file mode 100644 index 00000000000000..9bcdc43892730f Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/18446744073709551614.manifest differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/latest_version_hint.json b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/latest_version_hint.json new file mode 100644 index 00000000000000..205c7a40a84f18 --- /dev/null +++ b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/latest_version_hint.json @@ -0,0 +1 @@ +{"version":4} \ No newline at end of file diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/data/000110011100000010110011ff221e4e9d80211c9e467a875c.lance b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/data/000110011100000010110011ff221e4e9d80211c9e467a875c.lance new file mode 100644 index 00000000000000..46378d9b5e9ddc Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/data/000110011100000010110011ff221e4e9d80211c9e467a875c.lance differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/data/010101101110010111111100e282674630a37f9b31b69c3b3f.lance b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/data/010101101110010111111100e282674630a37f9b31b69c3b3f.lance new file mode 100644 index 00000000000000..a5dedbec0e5e12 Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/data/010101101110010111111100e282674630a37f9b31b69c3b3f.lance differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/data/10000001100001000001100118a7f64f1c805a2e133d8d5c21.lance b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/data/10000001100001000001100118a7f64f1c805a2e133d8d5c21.lance new file mode 100644 index 00000000000000..8e049a43590c72 Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/data/10000001100001000001100118a7f64f1c805a2e133d8d5c21.lance differ diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java index 394cb6ebf068a1..af26c973163269 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java @@ -65,6 +65,11 @@ * requested snapshot-wide result. */ public class LanceScanNode extends FileQueryScanNode { + // A metadata COUNT(*) whose result is at least this large is sharded across several + // fragment groups, because BE materializes one synthetic row per counted row and one carrier + // would serialize that O(rowCount) work on a single scanner. Matches IcebergScanNode. + private static final long COUNT_WITH_PARALLEL_SPLITS = 10000; + private LanceExternalTable lanceTable; private LanceTableMetadata plannedMetadata; private int vectorFieldId = -1; @@ -151,6 +156,14 @@ private boolean canPushDownLimit() { return hasLimit() && conjuncts.isEmpty(); } + // COUNT(*)/COUNT(1) can be answered from Lance metadata only when nothing narrows the row set: + // no residual Doris conjunct and no predicate pushed into Lance. Any filter would make the + // dataset-wide logical row count larger than the real result, so this is stricter than + // canPushDownLimit(), which still allows predicates already pushed into Lance. + private boolean canPushDownCountStar() { + return isTableLevelCountStarPushdown() && conjuncts.isEmpty() && lanceSubstraitFilter.length == 0; + } + @Override protected void convertPredicate() { if (isExternalSearch()) { @@ -194,6 +207,10 @@ public List getSplits(int numBackends) throws UserException { "Lance vector search requires a fixed positive dataset version"); } + if (canPushDownCountStar()) { + return buildCountSplits(metadata, numBackends); + } + Map visibleFragments = getVisibleFragments(metadata); if (isExternalSearch() && shouldUseIndex()) { Optional> indexSplits = createIndexSegmentSplits(metadata, visibleFragments); @@ -204,6 +221,44 @@ public List getSplits(int numBackends) throws UserException { return createFragmentSplits(metadata, visibleFragments); } + // COUNT(*)/COUNT(1) with no filter is answered from Lance metadata. Each carrier contains a + // disjoint fragment group and its logical row count, so a BE that cannot use the metadata count + // falls back to an equivalent fixed-snapshot scan. Large counts use several carriers to retain + // parallelism; small counts use one. + private List buildCountSplits(LanceTableMetadata metadata, int numBackends) { + long rowCount = metadata.getRowCount(); + setPushDownCount(rowCount); + int carrierCount = 1; + if (rowCount >= COUNT_WITH_PARALLEL_SPLITS && !metadata.getFragments().isEmpty()) { + int parallelism = sessionVariable.getParallelExecInstanceNum(scanContext.getClusterName()) + * Math.max(numBackends, 1); + carrierCount = Math.min(metadata.getFragments().size(), Math.max(1, parallelism)); + } + List> fragmentGroups = new ArrayList<>(carrierCount); + for (int i = 0; i < carrierCount; i++) { + fragmentGroups.add(new ArrayList<>()); + } + List fragments = metadata.getFragments(); + for (int i = 0; i < fragments.size(); i++) { + fragmentGroups.get(i % carrierCount).add(fragments.get(i)); + } + + List splits = new ArrayList<>(carrierCount); + for (List group : fragmentGroups) { + List fragmentIds = new ArrayList<>(group.size()); + long logicalRows = 0; + long physicalRows = 0; + for (LanceFragmentInfo fragment : group) { + fragmentIds.add(fragment.getId()); + logicalRows += fragment.getRowCount(); + physicalRows += fragment.getPhysicalRows(); + } + splits.add(LanceSplit.forCount(metadata.getDatasetUri(), metadata.getVersion(), + fragmentIds, logicalRows, physicalRows)); + } + return splits; + } + private Map getVisibleFragments(LanceTableMetadata metadata) throws UserException { Map visible = new LinkedHashMap<>(); @@ -369,25 +424,29 @@ protected void setScanParams(TFileRangeDesc rangeDesc, Split split) { TLanceFileDesc lanceParams = new TLanceFileDesc(); lanceParams.setDatasetUri(lanceSplit.getDatasetUri()); lanceParams.setVersion(lanceSplit.getVersion()); - if (lanceSplit.getFragmentIds().isEmpty()) { - throw new IllegalArgumentException("Lance scan split must contain fragments"); - } - if (!isExternalSearch() && (lanceSplit.getFragmentIds().size() != 1 - || lanceSplit.hasIndexSegmentUuids())) { - throw new IllegalArgumentException( - "Ordinary Lance scan split must contain one fragment and no index segment"); - } - lanceParams.setFragmentIds(lanceSplit.getFragmentIds()); - if (lanceSplit.hasIndexSegmentUuids()) { - List uuids = new ArrayList<>(lanceSplit.getIndexSegmentUuids().size()); - for (UUID uuid : lanceSplit.getIndexSegmentUuids()) { - ByteBuffer uuidBytes = ByteBuffer.allocate(16); - uuidBytes.putLong(uuid.getMostSignificantBits()); - uuidBytes.putLong(uuid.getLeastSignificantBits()); - uuidBytes.flip(); - uuids.add(uuidBytes); + if (lanceSplit.hasFragmentIds()) { + if (!isExternalSearch() && lanceSplit.getTableLevelRowCount() < 0 + && (lanceSplit.getFragmentIds().size() != 1 + || lanceSplit.hasIndexSegmentUuids())) { + throw new IllegalArgumentException( + "Ordinary Lance scan split must contain one fragment and no index segment"); } - lanceParams.setIndexSegmentUuids(uuids); + lanceParams.setFragmentIds(lanceSplit.getFragmentIds()); + if (lanceSplit.hasIndexSegmentUuids()) { + List uuids = new ArrayList<>(lanceSplit.getIndexSegmentUuids().size()); + for (UUID uuid : lanceSplit.getIndexSegmentUuids()) { + ByteBuffer uuidBytes = ByteBuffer.allocate(16); + uuidBytes.putLong(uuid.getMostSignificantBits()); + uuidBytes.putLong(uuid.getLeastSignificantBits()); + uuidBytes.flip(); + uuids.add(uuidBytes); + } + lanceParams.setIndexSegmentUuids(uuids); + } + } else if (lanceSplit.getTableLevelRowCount() < 0) { + // Only the metadata COUNT(*) split may omit fragment ids; it opens no BE scanner and + // BE serves the row count from table_level_row_count below, leaving fragment_ids unset. + throw new IllegalArgumentException("Lance scan split must contain fragments"); } // Push LIMIT into each ordinary fragment scanner only when it is safe to truncate that // fragment early. Vector search uses its own per-split candidate bound. @@ -397,6 +456,9 @@ protected void setScanParams(TFileRangeDesc rangeDesc, Split split) { TTableFormatFileDesc tableFormatParams = new TTableFormatFileDesc(); tableFormatParams.setTableFormatType(TableFormatType.LANCE.value()); + // Match the Iceberg convention: always set explicitly, -1 for ordinary and search scans + // so BE never mistakes a stale value for a metadata count. + tableFormatParams.setTableLevelRowCount(lanceSplit.getTableLevelRowCount()); tableFormatParams.setLanceParams(lanceParams); rangeDesc.setTableFormatParams(tableFormatParams); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceSplit.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceSplit.java index 8fbe24c8401446..5125ba6c749ce5 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceSplit.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceSplit.java @@ -36,6 +36,9 @@ public class LanceSplit extends FileSplit { private final long version; private final List fragmentIds; private final List indexSegmentUuids; + // Set to a nonnegative value only when this split carries a metadata COUNT(*) result so BE can + // synthesize that many rows instead of scanning fragments. -1 means ordinary scan. + private long tableLevelRowCount = -1; public static LanceSplit forFragment( String datasetUri, long version, long fragmentId, long physicalRows) { @@ -47,6 +50,16 @@ public static LanceSplit wholeDatasetAtLatest(String datasetUri) { return new LanceSplit(datasetUri, 0, Collections.emptyList(), Collections.emptyList(), 1); } + // A metadata COUNT(*) carrier pinned to the planned snapshot. Its fragment range remains valid + // input if BE falls back to scanning, while rowCount lets the metadata path skip that scan. + public static LanceSplit forCount(String datasetUri, long version, List fragmentIds, + long rowCount, long physicalRows) { + LanceSplit split = new LanceSplit( + datasetUri, version, fragmentIds, Collections.emptyList(), physicalRows); + split.tableLevelRowCount = rowCount; + return split; + } + public static LanceSplit forIndexSegment(String datasetUri, long version, UUID indexSegmentUuid, List fragmentIds, long physicalRows) { if (fragmentIds == null || fragmentIds.isEmpty()) { @@ -112,6 +125,10 @@ public boolean hasIndexSegmentUuids() { return !indexSegmentUuids.isEmpty(); } + public long getTableLevelRowCount() { + return tableLevelRowCount; + } + @Override public String getConsistentHashString() { return hasFragmentIds() diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/source/LanceScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/source/LanceScanNodeTest.java index 2fb3b7c390df6f..c2407be03fc848 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/source/LanceScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/source/LanceScanNodeTest.java @@ -30,6 +30,7 @@ import org.apache.doris.thrift.TExternalSearchQuery; import org.apache.doris.thrift.TExternalSearchRequest; import org.apache.doris.thrift.TFileRangeDesc; +import org.apache.doris.thrift.TPushAggOp; import org.apache.doris.thrift.TVectorMetric; import org.apache.doris.thrift.TVectorSearchOptions; import org.apache.doris.thrift.TVectorSearchParams; @@ -94,6 +95,31 @@ public void testDeletionHeavyFragmentKeepsPhysicalScanWeight() throws Exception assertSplit(splits.get(1), 11, 1000, 100); } + @Test + public void testCountSplitsPinVersionAndKeepFallbackRangesDisjoint() throws Exception { + LanceTableMetadata metadata = LanceTableMetadata.withoutIndexSegments( + "s3://bucket/table.lance", + 42, + new Schema(Collections.emptyList()), + Arrays.asList( + new LanceFragmentInfo(7, 6000, 6001), + new LanceFragmentInfo(11, 5000, 5001), + new LanceFragmentInfo(13, 4000, 4001)), + Collections.emptyMap()); + SessionVariable sessionVariable = new SessionVariable(); + sessionVariable.parallelExecInstanceNum = 1; + LanceScanNode node = newNode(sessionVariable); + setMetadata(node, metadata); + node.setPushDownAggNoGrouping(TPushAggOp.COUNT); + node.setPushDownCountSlotIds(Collections.emptyList()); + + List splits = node.getSplits(2); + + Assert.assertEquals(2, splits.size()); + assertCountRange(node, splits.get(0), Arrays.asList(7L, 13L), 10_000); + assertCountRange(node, splits.get(1), Collections.singletonList(11L), 5_000); + } + @Test public void testExternalSearchUsesFragmentSplits() throws Exception { LanceTableMetadata metadata = LanceTableMetadata.withoutIndexSegments( @@ -320,11 +346,15 @@ public void testLanceSplitRejectsInvalidRangeFieldsInFrontend() { } private static LanceScanNode newNode() { + return newNode(new SessionVariable()); + } + + private static LanceScanNode newNode(SessionVariable sessionVariable) { return new LanceScanNode( new PlanNodeId(0), new TupleDescriptor(new TupleId(0)), false, - new SessionVariable(), + sessionVariable, ScanContext.EMPTY); } @@ -360,6 +390,20 @@ private static void assertIndexSplit(Split split, UUID segmentUuid, List f Assert.assertEquals(weight, lanceSplit.getSplitWeight().getRawValue()); } + private static void assertCountRange( + LanceScanNode node, Split split, List fragmentIds, long rowCount) { + LanceSplit lanceSplit = (LanceSplit) split; + Assert.assertEquals(fragmentIds, lanceSplit.getFragmentIds()); + Assert.assertEquals(rowCount, lanceSplit.getTableLevelRowCount()); + + TFileRangeDesc range = new TFileRangeDesc(); + node.setScanParams(range, split); + Assert.assertEquals(42L, range.getTableFormatParams().getLanceParams().getVersion()); + Assert.assertEquals(fragmentIds, + range.getTableFormatParams().getLanceParams().getFragmentIds()); + Assert.assertEquals(rowCount, range.getTableFormatParams().getTableLevelRowCount()); + } + private static Schema vectorSchema() { return new Schema(Collections.singletonList( Field.nullable("vector", ArrowType.Utf8.INSTANCE))); diff --git a/regression-test/data/external_table_p0/lance/test_lance_optimize_count.out b/regression-test/data/external_table_p0/lance/test_lance_optimize_count.out new file mode 100644 index 00000000000000..5bb00ccb27c2b5 --- /dev/null +++ b/regression-test/data/external_table_p0/lance/test_lance_optimize_count.out @@ -0,0 +1,31 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !count_star_pushdown -- +12 + +-- !count_one_pushdown -- +12 + +-- !count_star_all_rows -- +12 + +-- !count_star_half -- +6 + +-- !count_star_no_pushdown -- +12 + +-- !mf_count_star_pushdown -- +27 + +-- !mf_count_one_pushdown -- +27 + +-- !mf_count_all_rows -- +27 + +-- !mf_count_half -- +14 + +-- !mf_count_star_no_pushdown -- +27 + diff --git a/regression-test/suites/external_table_p0/lance/test_lance_optimize_count.groovy b/regression-test/suites/external_table_p0/lance/test_lance_optimize_count.groovy new file mode 100644 index 00000000000000..bf0dd7d424f863 --- /dev/null +++ b/regression-test/suites/external_table_p0/lance/test_lance_optimize_count.groovy @@ -0,0 +1,160 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +suite("test_lance_optimize_count", "p0,external") { + /* + * COUNT(*)/COUNT(1) with no filter is served from Lance dataset metadata: + * FE emits a single split carrying the logical row count and EXPLAIN shows + * "pushdown agg=COUNT ()". Any of the following disables that path and + * falls back to a normal scan ("pushdown agg=NONE"), which must still return + * the same count: + * 1. enable_count_push_down_for_external_table = false; + * 2. a WHERE filter, because Lance cannot describe COUNT with a predicate, + * so the plan keeps Aggregate(Filter(FileScan)) and never folds into a + * storage-layer aggregate. + * + * all_types has exactly 12 rows in a single fragment with contiguous, unique + * row_id in [1, 12], so every count below is deterministic. + * + * multi_frag is the multi-split counterpart: 30 physical rows in three fragments + * with one deleted row per fragment (row_id 5, 15, 25), so its logical count is 27. + * It proves the metadata count reports the logical 27 rather than the physical 30, + * and that a normal multi-split scan applies every fragment's deletion vector exactly + * once (no fragment double-counted or skipped). + */ + String enabled = context.config.otherConfigs.get("enableIcebergTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable Lance count pushdown test because the Iceberg MinIO environment is disabled.") + return + } + + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String catalogName = "test_lance_optimize_count" + + sql """DROP CATALOG IF EXISTS `${catalogName}`""" + try { + sql """ + CREATE CATALOG `${catalogName}` PROPERTIES ( + "type" = "lance", + "lance.catalog.type" = "filesystem", + "warehouse" = "s3://warehouse/lance", + "s3.endpoint" = "http://${externalEnvIp}:${minioPort}", + "s3.access_key" = "admin", + "s3.secret_key" = "password", + "s3.region" = "us-east-1", + "use_path_style" = "true" + ) + """ + + sql """ USE `${catalogName}`.`default`; """ + // Lance is only served by FileScannerV2, which is where the metadata count + // short-circuit lives. + sql """ SET enable_file_scanner_v2 = true; """ + + String countStar = """ SELECT count(*) FROM all_types """ + String countOne = """ SELECT count(1) FROM all_types """ + String countStarAllRows = """ SELECT count(*) FROM all_types WHERE row_id > 0 """ + String countStarHalf = """ SELECT count(*) FROM all_types WHERE row_id > 6 """ + + // ---- Pushdown ON (the optimization) ---- + sql """ SET enable_count_push_down_for_external_table = true; """ + + // No filter: COUNT(*) and COUNT(1) both fold into the metadata count. + explain { + sql(countStar) + contains "pushdown agg=COUNT (12)" + } + explain { + sql(countOne) + contains "pushdown agg=COUNT (12)" + } + qt_count_star_pushdown """${countStar}""" + qt_count_one_pushdown """${countOne}""" + + // A filter keeps the aggregate above the scan, so no metadata count. + explain { + sql(countStarHalf) + contains "pushdown agg=NONE" + } + qt_count_star_all_rows """${countStarAllRows}""" + qt_count_star_half """${countStarHalf}""" + + // ---- Pushdown OFF (the baseline before the optimization) ---- + sql """ SET enable_count_push_down_for_external_table = false; """ + + explain { + sql(countStar) + contains "pushdown agg=NONE" + } + // Same result whether or not the metadata count is used. + qt_count_star_no_pushdown """${countStar}""" + + // ---- Multi-fragment table with deletions (three splits, logical count 27) ---- + String mfCountStar = """ SELECT count(*) FROM multi_frag """ + String mfCountOne = """ SELECT count(1) FROM multi_frag """ + // Filter keeps > 12 rows so it cannot be confused with all_types' count. + String mfCountHalf = """ SELECT count(*) FROM multi_frag WHERE row_id > 15 """ + // A whole-table filter must still equal the metadata count: it exercises the + // multi-split scan path (deletion vectors applied per fragment) instead of the + // metadata short-circuit, and the two must agree. + String mfCountAll = """ SELECT count(*) FROM multi_frag WHERE row_id > 0 """ + + sql """ SET enable_count_push_down_for_external_table = true; """ + + // No filter: the single metadata split reports the logical 27, not the physical 30. + explain { + sql(mfCountStar) + contains "pushdown agg=COUNT (27)" + } + explain { + sql(mfCountOne) + contains "pushdown agg=COUNT (27)" + } + qt_mf_count_star_pushdown """${mfCountStar}""" + qt_mf_count_one_pushdown """${mfCountOne}""" + + // The metadata count carrier must pin the planned dataset version, not latest (version 0): + // a fallback scan (an old BE, or a BE that declines the shortcut) has to read the same + // snapshot the count came from. multi_frag is built with three appends and three deletes, + // so its planned version is 4. A regression to wholeDatasetAtLatest() would print + // "lanceVersion=0" here and let time-travel / concurrent-commit reads drift. + explain { + sql(mfCountStar) + contains "lanceVersion=4" + } + + // Filtered counts fall back to the three-split scan and must still be exact. + explain { + sql(mfCountHalf) + contains "pushdown agg=NONE" + } + qt_mf_count_all_rows """${mfCountAll}""" + qt_mf_count_half """${mfCountHalf}""" + + // Off switch: the same multi-split scan must reproduce the logical 27. + sql """ SET enable_count_push_down_for_external_table = false; """ + explain { + sql(mfCountStar) + contains "pushdown agg=NONE" + } + qt_mf_count_star_no_pushdown """${mfCountStar}""" + } finally { + sql """ SET enable_count_push_down_for_external_table = true; """ + sql """DROP CATALOG IF EXISTS `${catalogName}`""" + } +}