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

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");
     }

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}")
 
 

docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/multi_frag.lance/_versions/latest_version_hint.json

@@ -0,0 +1 @@
+{"version":4}

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<Split> getSplits(int numBackends) throws UserException {
                     "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 List<Split> 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<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 @@ 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<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 @@ 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);
     }

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 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<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 boolean hasIndexSegmentUuids() {
         return !indexSegmentUuids.isEmpty();
     }
 
+    public long getTableLevelRowCount() {
+        return tableLevelRowCount;
+    }
+
     @Override
     public String getConsistentHashString() {
         return hasFragmentIds()