This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 284569abc55 [feature](lance) push COUNT(*) down to Lance dataset
metadata (#66999)
284569abc55 is described below
commit 284569abc5566d8a3ff05b4c5210000a604875c4
Author: jay <[email protected]>
AuthorDate: Mon Sep 7 09:16:27 2026 +0800
[feature](lance) push COUNT(*) down to Lance dataset metadata (#66999)
## 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.
---
be/src/format_v2/table/lance_reader.cpp | 14 +-
.../scripts/lance_build_preinstalled_catalog.py | 77 ++++++++++
.../_deletions/0-3-12903222946140852451.arrow | Bin 0 -> 698 bytes
.../_deletions/1-3-3997032015744280171.arrow | Bin 0 -> 698 bytes
.../_deletions/2-3-16205658664475652884.arrow | Bin 0 -> 698 bytes
.../0-9d78d1d6-4352-45af-81d3-11147a3165c5.txn | Bin 0 -> 220 bytes
.../1-4ba9d959-a46f-4ecc-bc97-edc7030c82ef.txn | Bin 0 -> 124 bytes
.../2-1b03ecda-1a9c-411c-b9f0-f91af2575aa7.txn | Bin 0 -> 124 bytes
.../3-4b3d53f1-4352-4d80-a8d9-312412d6e5c6.txn | Bin 0 -> 364 bytes
.../_versions/18446744073709551611.manifest | Bin 0 -> 890 bytes
.../_versions/18446744073709551612.manifest | Bin 0 -> 596 bytes
.../_versions/18446744073709551613.manifest | Bin 0 -> 513 bytes
.../_versions/18446744073709551614.manifest | Bin 0 -> 526 bytes
.../_versions/latest_version_hint.json | 1 +
...100000010110011ff221e4e9d80211c9e467a875c.lance | Bin 0 -> 1191 bytes
...110010111111100e282674630a37f9b31b69c3b3f.lance | Bin 0 -> 1191 bytes
...00001000001100118a7f64f1c805a2e133d8d5c21.lance | Bin 0 -> 1191 bytes
.../datasource/lance/source/LanceScanNode.java | 98 ++++++++++---
.../doris/datasource/lance/source/LanceSplit.java | 17 +++
.../datasource/lance/source/LanceScanNodeTest.java | 46 +++++-
.../lance/test_lance_optimize_count.out | 31 ++++
.../lance/test_lance_optimize_count.groovy | 160 +++++++++++++++++++++
22 files changed, 422 insertions(+), 22 deletions(-)
diff --git a/be/src/format_v2/table/lance_reader.cpp
b/be/src/format_v2/table/lance_reader.cpp
index e8ad0c5229a..860623613fd 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 9721074094f..fa1b7946d69 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 @@ HNSW_PQ_BUILD_PARAMS = {**HNSW_BUILD_PARAMS,
"num_sub_vectors": 4, "num_bits": 8
# 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 00000000000..badfc10c378
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 00000000000..badfc10c378
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 00000000000..badfc10c378
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 00000000000..70d9f843807
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 00000000000..8867f8e63c2
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 00000000000..930321f052e
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 00000000000..45da071eab9
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 00000000000..0bdc388ef6f
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 00000000000..da761c20840
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 00000000000..c2920b2c6cf
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 00000000000..9bcdc438927
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 00000000000..205c7a40a84
--- /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 00000000000..46378d9b5e9
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 00000000000..a5dedbec0e5
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 00000000000..8e049a43590
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 394cb6ebf06..af26c973163 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 @@ import java.util.UUID;
* 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 @@ public class LanceScanNode extends FileQueryScanNode {
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 class LanceScanNode extends FileQueryScanNode {
"Lance vector search requires a fixed positive dataset
version");
}
+ if (canPushDownCountStar()) {
+ return buildCountSplits(metadata, numBackends);
+ }
+
Map<Long, LanceFragmentInfo> visibleFragments =
getVisibleFragments(metadata);
if (isExternalSearch() && shouldUseIndex()) {
Optional<List<Split>> indexSplits =
createIndexSegmentSplits(metadata, visibleFragments);
@@ -204,6 +221,44 @@ public class LanceScanNode extends FileQueryScanNode {
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<Split> 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<List<LanceFragmentInfo>> fragmentGroups = new
ArrayList<>(carrierCount);
+ for (int i = 0; i < carrierCount; i++) {
+ fragmentGroups.add(new ArrayList<>());
+ }
+ List<LanceFragmentInfo> fragments = metadata.getFragments();
+ for (int i = 0; i < fragments.size(); i++) {
+ fragmentGroups.get(i % carrierCount).add(fragments.get(i));
+ }
+
+ List<Split> splits = new ArrayList<>(carrierCount);
+ for (List<LanceFragmentInfo> group : fragmentGroups) {
+ List<Long> 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<Long, LanceFragmentInfo>
getVisibleFragments(LanceTableMetadata metadata)
throws UserException {
Map<Long, LanceFragmentInfo> visible = new LinkedHashMap<>();
@@ -369,25 +424,29 @@ public class LanceScanNode extends FileQueryScanNode {
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<ByteBuffer> 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<ByteBuffer> 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 @@ public class LanceScanNode extends FileQueryScanNode {
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 8fbe24c8401..5125ba6c749 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<Long> fragmentIds;
private final List<UUID> 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 class LanceSplit extends FileSplit {
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<Long> 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<Long> fragmentIds, long physicalRows) {
if (fragmentIds == null || fragmentIds.isEmpty()) {
@@ -112,6 +125,10 @@ public class LanceSplit extends FileSplit {
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 2fb3b7c390d..c2407be03fc 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.spi.Split;
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 class LanceScanNodeTest {
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<Split> 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 class LanceScanNodeTest {
}
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 @@ public class LanceScanNodeTest {
Assert.assertEquals(weight, lanceSplit.getSplitWeight().getRawValue());
}
+ private static void assertCountRange(
+ LanceScanNode node, Split split, List<Long> 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 00000000000..5bb00ccb27c
--- /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 00000000000..bf0dd7d424f
--- /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 (<rows>)". 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}`"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
