diff --git a/be/src/storage/CMakeLists.txt b/be/src/storage/CMakeLists.txt index 21e82183f58dff..306b4815f208f0 100644 --- a/be/src/storage/CMakeLists.txt +++ b/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. diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java index bd2e90bd64680a..24c2a5ec0d1922 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java +++ b/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 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> 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> planManifestCountRange(Table table, TableScan scan, long exactCount, int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, UnaryOperator uriNormalizer, ConnectorSession session, Optional filter, diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java index 73c0abd20a9b15..abbcb46a8e0d39 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java +++ b/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