[fix](build) Unbreak master: stale unity-skip entry (BE) and dropped count probe (FE)

be/src/storage/CMakeLists.txt

@@ -55,10 +55,6 @@ pch_reuse(Storage)
 #   FORMAT_*_ADD_JSON_NODE, RETURN_IF_ERROR_) must not leak into unity siblings
 # - the three heaviest template-instantiation TUs (predicate creators) which
 #   would dominate any batch they join
-# - compaction/collection_statistics.cpp: its test compiles it a second time
-#   by #including the .cpp; the test object must shadow a never-pulled archive
-#   member, but a unity batch is pulled in for its siblings and the linker
-#   sees a duplicate definition
 set(STORAGE_UNITY_SKIP
     ${CMAKE_CURRENT_SOURCE_DIR}/index/inverted/inverted_index_compound_reader.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/index/inverted/inverted_index_fs_directory.cpp
@@ -69,8 +65,7 @@ set(STORAGE_UNITY_SKIP
     ${CMAKE_CURRENT_SOURCE_DIR}/task/engine_clone_task.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/predicate/predicate_creator_comparison.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/predicate/predicate_creator_in_list_in.cpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/predicate/predicate_creator_in_list_not_in.cpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/compaction/collection_statistics.cpp)
+    ${CMAKE_CURRENT_SOURCE_DIR}/predicate/predicate_creator_in_list_not_in.cpp)
 if (ENABLE_VARIANT_NESTED_GROUP)
     # Out-of-tree module sources swapped into this target: unity hygiene
     # unaudited, keep them individual.

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java

@@ -497,12 +497,12 @@ public boolean canServeMetadataOnlyCount(ConnectorSession session, ConnectorTabl
             Optional<ConnectorExpression> filter) {
         IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
         if (iceHandle.isSystemTable() || filter.isPresent()) {
-            // Snapshot summaries describe the whole table and cannot prove a filtered row count.
+            // A metadata count describes the whole table and cannot prove a filtered row count.
             return false;
         }
         Table table = resolveTable(session, iceHandle);
         TableScan scan = buildScan(table, iceHandle, filter, session);
-        return getCountFromSnapshot(scan, session) >= 0;
+        return canProveCountFromManifests(table, scan, session);
     }
 
     /**
@@ -1312,6 +1312,38 @@ private Optional<List<ConnectorScanRange>> planCountPushdown(Table table, TableS
                 netPositionDeletes ? positionDeleteRows.getAsLong() : 0);
     }
 
+    /**
+     * The capability probe behind {@link #canServeMetadataOnlyCount}: the delete gate of
+     * {@link #planCountPushdown} plus the requirement that the data manifests really carry aggregate row
+     * counters. Reading only the manifest list keeps this O(manifests) with no data-file enumeration, which is
+     * what a pre-planning probe can afford; the price is answering {@code false} for the older manifest lists
+     * that {@code planCountPushdown} still serves through its bounded per-file fallback. Never derives the
+     * count from snapshot summary fields — those are writer-provided hints, not a query result.
+     */
+    private static boolean canProveCountFromManifests(Table table, TableScan scan, ConnectorSession session) {
+        Snapshot snapshot = scan.snapshot();
+        if (snapshot == null) {
+            // No snapshot (empty table, or a pinned empty snapshot) is an exact count of 0 without any read.
+            return true;
+        }
+        boolean netPositionDeletes = sessionBool(session, IGNORE_ICEBERG_DANGLING_DELETE, false);
+        ManifestDeleteState deleteState = manifestDeleteState(snapshot.deleteManifests(table.io()));
+        if (deleteState == ManifestDeleteState.PRESENT && !netPositionDeletes) {
+            return false;
+        }
+        OptionalLong positionDeleteRows = deleteState == ManifestDeleteState.NONE
+                ? OptionalLong.of(0)
+                : livePositionDeleteRowCount(table, snapshot);
+        if (!positionDeleteRows.isPresent()
+                || (!netPositionDeletes && positionDeleteRows.getAsLong() != 0)) {
+            return false;
+        }
+        OptionalLong manifestCount = liveRowCountFromManifests(snapshot.dataManifests(table.io()));
+        return manifestCount.isPresent()
+                && subtractPositionDeleteRows(manifestCount.getAsLong(),
+                        netPositionDeletes ? positionDeleteRows.getAsLong() : 0).isPresent();
+    }
+
     private Optional<List<ConnectorScanRange>> planManifestCountRange(Table table, TableScan scan, long exactCount,
             int formatVersion, boolean partitioned, List<String> orderedPartitionKeys, ZoneId zone,
             UnaryOperator<String> uriNormalizer, ConnectorSession session, Optional<ConnectorExpression> filter,

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java

@@ -1871,7 +1871,7 @@ public void countPushdownFollowsTheSnapshotPin() {
     }
 
     @Test
-    public void metadataOnlyCountCapabilityUsesSnapshotSummary() {
+    public void metadataOnlyCountCapabilityUsesManifestAggregates() {
         Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned());
         table.newAppend().appendFile(dataFile(
                 table.spec(), "s3://b/db/t1/f1.parquet", 1000, null, null)).commit();
@@ -1887,6 +1887,24 @@ public void metadataOnlyCountCapabilityUsesSnapshotSummary() {
                         "db1", "t1", "snapshots", -1L, null, -1L), Optional.empty()));
     }
 
+    @Test
+    public void metadataOnlyCountCapabilityFollowsTheDeleteGate() {
+        // The capability must follow the same delete gate as count planning: live deletes leave the row count
+        // unprovable from manifests alone. MUTATION: reporting the capability from data manifests only -> red.
+        Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned(),
+                Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+        table.newAppend().appendFile(dataFile(
+                table.spec(), "s3://b/db/t1/f1.parquet", 1000, null, null)).commit();
+        table.newRowDelta().addDeletes(
+                positionDeleteFile("s3://b/db/t1/pos.parquet", FileFormat.PARQUET, null, null)).commit();
+        IcebergScanPlanProvider provider = new IcebergScanPlanProvider(
+                IcebergCatalogProperties.of(Collections.emptyMap()), opsReturning(table));
+        ConnectorSession session = new FakeScanSession("UTC", Collections.emptyMap());
+
+        Assertions.assertFalse(provider.canServeMetadataOnlyCount(
+                session, new IcebergTableHandle("db1", "t1"), Optional.empty()));
+    }
+
     @Test
     public void getScanNodePropertiesUnderPinEmitsFullPinnedSchemaDict() throws Exception {
         // T07 Option A: under a time-travel pin the field-id dict is built from the FULL pinned schema (covering