From 13c403300e05480e2064535357aaca5639c31f87 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 14 Aug 2026 17:55:25 +0800 Subject: [PATCH 1/4] [fix](iceberg) Derive count pushdown from live manifests Iceberg snapshot summary fields are optional writer metadata and may contain a valid but incorrect positive row count. Derive exact unfiltered COUNT(*) results from live data-file record counts, and fall back to a normal scan for filters, deletes, invalid counts, or overflow. Tests: IcebergScanPlanProviderTest (148 tests) --- .../iceberg/IcebergConnectorMetadata.java | 15 +- .../iceberg/IcebergScanPlanProvider.java | 150 +++++------------- .../iceberg/IcebergCountFromSummaryTest.java | 114 ------------- .../iceberg/IcebergScanPlanProviderTest.java | 130 ++++++++++++--- 4 files changed, 157 insertions(+), 252 deletions(-) delete mode 100644 fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCountFromSummaryTest.java diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java index 2546fe4690e59b..fdaa9ff3704446 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java @@ -127,10 +127,9 @@ public class IcebergConnectorMetadata implements ConnectorMetadata { private static final int ICEBERG_ROW_LINEAGE_MIN_VERSION = 3; // Snapshot-summary keys for table-level row count (getTableStatistics). Local literal copies of the - // spec-stable iceberg strings — byte-identical to legacy IcebergUtils.TOTAL_* and to the COUNT(*) - // pushdown copies in IcebergScanPlanProvider (themselves deliberately NOT org.apache.iceberg - // .SnapshotSummary.* per that file's note). Duplicated rather than shared so this fix does not touch - // the unrelated scan provider. All THREE keys are read: legacy getIcebergRowCount (via + // spec-stable iceberg strings — byte-identical to legacy IcebergUtils.TOTAL_*. These remain optimizer + // estimates only; exact COUNT(*) pushdown deliberately derives its result from live data-file manifests. + // All THREE keys are read: legacy getIcebergRowCount (via // getCountFromSummary, upstream 32a2651f66b / #64648) nets out position deletes AND gates the count to // UNKNOWN on any equality delete — see computeRowCount. private static final String TOTAL_RECORDS = "total-records"; @@ -823,9 +822,8 @@ public Optional getTableStatistics( * .getIcebergRowCount} (which calls {@code getCountFromSummary(summary, true)}, upstream 32a2651f66b / * #64648): any equality delete ({@code total-equality-deletes} absent or {@code != "0"}) -> -1 (UNKNOWN), * since equality deletes re-project at read time and the summary cannot net them out; otherwise - * {@code total-records - total-position-deletes}. Shares the equality-delete gate with the COUNT(*) - * pushdown {@code IcebergScanPlanProvider.getCountFromSummary}, differing only in dangling-delete handling - * (table statistics always net out position deletes; the pushdown honors the dangling-delete session var). + * {@code total-records - total-position-deletes}. This best-effort optimizer estimate is not used as an + * exact query result; COUNT(*) pushdown independently sums required record counts from live data manifests. * Empty table (no current snapshot) -> -1, which the caller maps to UNKNOWN. */ private static long computeRowCount(Table table) { @@ -842,8 +840,7 @@ private static long computeRowCount(Snapshot snapshot) { // summary, true) (upstream 32a2651f66b, #64648): an absent total-* counter (compaction / replace / // overwrite snapshots may omit one — the pre-fix Long.parseLong(null) NPE-d), or any equality delete // (total-equality-deletes != "0"), makes the summary row count unsafe -> -1 (caller maps to UNKNOWN), - // because equality deletes re-project at read time and the summary cannot net them out. Same gate as - // the COUNT(*) pushdown IcebergScanPlanProvider.getCountFromSummary. + // because equality deletes re-project at read time and the summary cannot net them out. String equalityDeletes = summary.get(TOTAL_EQUALITY_DELETES); String totalRecords = summary.get(TOTAL_RECORDS); String positionDeletes = summary.get(TOTAL_POSITION_DELETES); 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 367828fb0ed60f..fd85cbaeb062d5 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 @@ -152,13 +152,8 @@ public class IcebergScanPlanProvider implements ConnectorScanPlanProvider { private static final String NUM_FILES_IN_BATCH_MODE = "num_files_in_batch_mode"; private static final long DEFAULT_NUM_FILES_IN_BATCH_MODE = 1024L; - // COUNT(*) pushdown (T05). The snapshot-summary keys are the stable iceberg spec strings — byte-identical - // to legacy IcebergUtils.TOTAL_* (themselves local constants, not org.apache.iceberg.SnapshotSummary.*). - private static final String TOTAL_RECORDS = "total-records"; - private static final String TOTAL_POSITION_DELETES = "total-position-deletes"; + // Equality-delete schema discovery uses this stable Iceberg snapshot-summary key as a read-avoidance hint. private static final String TOTAL_EQUALITY_DELETES = "total-equality-deletes"; - // Session var: when a table has only (dangling) position deletes, ignore them and still push count down. - private static final String IGNORE_ICEBERG_DANGLING_DELETE = "ignore_iceberg_dangling_delete"; // System-table (P6.5-T05) JNI split: a placeholder path matching legacy IcebergSplit.DUMMY_PATH. A sys split // carries no real file (BE reads the serialized FileScanTask), so the path is never opened — it only keeps @@ -448,6 +443,7 @@ public long streamingSplitEstimate(ConnectorSession session, ConnectorTableHandl Optional filter, boolean countPushdown) { IcebergTableHandle iceHandle = (IcebergTableHandle) handle; if (iceHandle.isResolvedEmptySnapshot() || iceHandle.isSystemTable() + || (countPushdown && filter.isEmpty()) || !sessionBool(session, ENABLE_EXTERNAL_TABLE_BATCH_MODE, true)) { return -1; } @@ -460,9 +456,6 @@ public long streamingSplitEstimate(ConnectorSession session, ConnectorTableHandl if (getFormatVersion(table) >= 3) { return -1; } - if (countPushdown && getCountFromSnapshot(scan, session) >= 0) { - return -1; - } long threshold = sessionLong(session, NUM_FILES_IN_BATCH_MODE, DEFAULT_NUM_FILES_IN_BATCH_MODE); long fileCount = 0; try (CloseableIterable matching = getMatchingManifest( @@ -707,16 +700,17 @@ private List planScanInternal( // per-file path normalization below, instead of rebuilding it per data/delete file (C3). UnaryOperator uriNormalizer = newUriNormalizer(vendedToken); - // COUNT(*) pushdown (T05): when the count is servable from the snapshot summary, collapse the scan to - // a single whole-file range carrying the full count (mirrors paimon's collapse + legacy's <=10000 - // case; the legacy >10000 parallel multi-split trim is a perf-only divergence, dropped). A -1 (equality - // deletes, or dangling position deletes without the ignore flag) falls through to the normal scan so - // BE reads and counts. - if (countPushdown) { - long realCount = getCountFromSnapshot(scan, session); - if (realCount >= 0) { - return planCountPushdown(table, scan, realCount, formatVersion, partitioned, - orderedPartitionKeys, zone, uriNormalizer, session, filter); + // COUNT(*) pushdown (T05): derive an exact count from the live data-file manifests and collapse the scan + // to a single whole-file range. Snapshot summary fields are optional writer-provided metadata and must + // never become a query result. If deletes or invalid record counts prevent an exact proof, fall through + // to the normal scan so BE reads and counts. + // A data-row predicate can leave partially matching files, whose file-level recordCount is only an + // upper bound. Keep those scans on the normal path even if the engine supplies the count signal. + if (countPushdown && filter.isEmpty()) { + Optional> countRanges = planCountPushdown(table, scan, formatVersion, + partitioned, orderedPartitionKeys, zone, uriNormalizer, session, filter); + if (countRanges.isPresent()) { + return countRanges.get(); } } @@ -1154,7 +1148,7 @@ private static boolean isPositionDeletesPartitionColumnRequested(List filter, ConnectorSession session) { @@ -1241,45 +1235,51 @@ private static Schema pinnedSchema(Table table, IcebergTableHandle handle) { } /** - * Emit the single collapsed COUNT(*)-pushdown range: the first whole-file {@link FileScanTask} from - * {@code scan.planFiles()} carrying the full {@code realCount} via {@code table_level_row_count} → BE's - * count reader serves it without opening the data file. Mirrors paimon's {@code buildCountRange} (one - * range bearing the summed total). Result-identical to legacy's count short-circuit even though legacy - * takes a different shape: legacy byte-splits the count file ({@code planFileScanTask} → - * {@code splitFiles} → {@code TableScanUtil.splitFiles}), keeps the first split task's byte-range for - * {@code count < 10000}, and {@code assignCountToSplits} distributes the same total — but under count - * pushdown BE's count reader never reads the file (the range's start/length are irrelevant) and sums - * {@code table_level_row_count} across ranges, so one whole-file range yields the identical total (and - * legacy's {@code >10000} parallel multi-split trim is the perf-only divergence we drop). An empty table - * (no files) yields no range, so BE gets 0 ranges and COUNT returns 0 (legacy returns empty splits too). + * Build a collapsed COUNT(*) range while enumerating every live {@link FileScanTask}. The data-file + * {@code recordCount} field is required by the Iceberg spec, unlike optional snapshot summary fields. The + * optimization is valid only when every task has no attached delete file and every record count can be + * summed exactly. Otherwise the empty optional tells the caller to perform a normal scan. */ - private List planCountPushdown(Table table, TableScan scan, long realCount, + private Optional> planCountPushdown(Table table, TableScan scan, int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, UnaryOperator uriNormalizer, ConnectorSession session, Optional filter) { + FileScanTask representative = null; + long exactCount = 0; try (CloseableIterable tasks = countPushdownFileScanTasks(scan, session, table, filter)) { for (FileScanTask task : tasks) { - // targetSplitSize = -1: the count-pushdown collapse emits a single range, so its scheduling - // weight is irrelevant → PluginDrivenSplit keeps SplitWeight.standard(). - return Collections.singletonList(buildRange(table, task.file(), task, formatVersion, - partitioned, orderedPartitionKeys, zone, uriNormalizer, realCount, -1, null)); + // A metadata count is safe only when manifests alone prove the exact visible row count. + if (!task.deletes().isEmpty() || task.file().recordCount() < 0) { + return Optional.empty(); + } + try { + exactCount = Math.addExact(exactCount, task.file().recordCount()); + } catch (ArithmeticException e) { + return Optional.empty(); + } + if (representative == null) { + representative = task; + } } } catch (IOException e) { throw new RuntimeException("Failed to plan iceberg count-pushdown file, error message is:" + e.getMessage(), e); } - return Collections.emptyList(); + if (representative == null) { + return Optional.of(Collections.emptyList()); + } + // targetSplitSize = -1: the count-pushdown collapse emits a single range, so its scheduling weight is + // irrelevant and PluginDrivenSplit keeps SplitWeight.standard(). + return Optional.of(Collections.singletonList(buildRange(table, representative.file(), representative, + formatVersion, partitioned, orderedPartitionKeys, zone, uriNormalizer, exactCount, -1, null))); } /** - * The COUNT(*)-pushdown placeholder enumeration: only the FIRST surviving file is consumed (BE serves the - * count from {@code table_level_row_count} and never reads the file). PERF-04 (C18): when the manifest cache is - * enabled, read through the lazy {@link #cacheBackedFileScanTasks} (stats overload — this runs on the single - * planning thread) so the manifest reads are cache hits and, being lazy, stop at the first file's manifest - * instead of the SDK {@code planFiles()}'s {@code ParallelIterable} eagerly submitting every manifest reader. + * The COUNT(*)-pushdown enumeration. PERF-04 (C18): when the manifest cache is enabled, read through the lazy + * {@link #cacheBackedFileScanTasks} (stats overload — this runs on the single planning thread) so manifest + * reads are cache hits without materializing the table's task list in FE memory. * An eager cache failure falls back to the SDK path (mirrors {@link #planFileScanTask}). The first surviving * (pruned) file may differ from the SDK path's first file (its {@code ParallelIterable} order is - * non-deterministic), but the count is identical (from the snapshot summary) and BE ignores the file. Cache - * disabled -> the SDK path, byte-unchanged. + * non-deterministic), but BE ignores the representative file. Cache disabled -> the SDK path, byte-unchanged. */ private CloseableIterable countPushdownFileScanTasks(TableScan scan, ConnectorSession session, Table table, Optional filter) { @@ -2758,68 +2758,6 @@ private static boolean sessionBool(ConnectorSession session, String key, boolean return Boolean.parseBoolean(raw.trim()); } - /** - * Compute the COUNT(*)-pushdown row count from the scan's snapshot summary, a faithful port of legacy - * {@code IcebergScanNode.getCountFromSnapshot}. No snapshot (empty table) → {@code 0}; otherwise - * delegates to {@link #getCountFromSummary}. Reads the scan's snapshot ({@code scan.snapshot()}) so the - * count tracks the scan automatically (the current snapshot today; the pinned snapshot once MVCC - * time-travel lands) — equivalent to legacy's {@code currentSnapshot()} for every non-time-travel query. - */ - private static long getCountFromSnapshot(TableScan scan, ConnectorSession session) { - Snapshot snapshot = scan.snapshot(); - if (snapshot == null) { - return 0; - } - return getCountFromSummary(snapshot.summary(), ignoreIcebergDanglingDelete(session)); - } - - /** - * Null-safe port of fe-core {@code IcebergUtils.getCountFromSummary} (upstream 32a2651f66b, #64648). - * Returns {@code -1} — this module's "count not pushable / unknown" sentinel; the {@code planScan} gate - * and count-collapse callers both test {@code >= 0} — in two cases: - *
    - *
  • any required {@code total-*} counter is ABSENT: compaction / replace / overwrite snapshots may - * omit {@code total-records} / {@code total-position-deletes} / {@code total-equality-deletes}, and - * the pre-fix code NPE-d on {@code summary.get(...).equals(...)} / {@code Long.parseLong(null)};
    • - *
  • any equality delete ({@code total-equality-deletes != "0"}) — not pushable, since equality - * deletes re-project at read time and the summary cannot net them out.
    • - *
- * Otherwise: no position deletes → {@code total-records}; position deletes present and - * {@code ignoreDanglingDelete} → {@code total-records - total-position-deletes}; else {@code -1}. - */ - static long getCountFromSummary(Map summary, boolean ignoreDanglingDelete) { - String equalityDeletes = summary.get(TOTAL_EQUALITY_DELETES); - String positionDeletes = summary.get(TOTAL_POSITION_DELETES); - String totalRecords = summary.get(TOTAL_RECORDS); - if (equalityDeletes == null || positionDeletes == null || totalRecords == null) { - // a summary that omits any total-* counter can't be netted safely -> fall back to a real scan - return -1; - } - if (!equalityDeletes.equals("0")) { - // has equality delete files, can not push down count - return -1; - } - long deleteCount = Long.parseLong(positionDeletes); - if (deleteCount == 0) { - // no delete files, can push down count directly - return Long.parseLong(totalRecords); - } - if (ignoreDanglingDelete) { - // has position delete files; if we ignore dangling deletes, the netted count can be pushed down - return Long.parseLong(totalRecords) - deleteCount; - } - // otherwise, can not push down count - return -1; - } - - private static boolean ignoreIcebergDanglingDelete(ConnectorSession session) { - if (session == null) { - return false; - } - String raw = session.getSessionProperties().get(IGNORE_ICEBERG_DANGLING_DELETE); - return raw != null && Boolean.parseBoolean(raw.trim()); - } - // The session time zone drives zone-adjusted (timestamptz) literal pushdown. Delegates to the shared // IcebergTimeUtils (Doris alias map, mirrors fe-core TimeUtils.getTimeZone()) so aliases like CST/PRC/EST // match legacy instead of throwing; null/blank/genuinely-invalid -> UTC. Package-private for unit testing. diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCountFromSummaryTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCountFromSummaryTest.java deleted file mode 100644 index a1e67d790194c4..00000000000000 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCountFromSummaryTest.java +++ /dev/null @@ -1,114 +0,0 @@ -// 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. - -package org.apache.doris.connector.iceberg; - -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; - -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; - -/** - * FIX-COUNT-NPE (upstream 32a2651f66b, #64648) — pins that - * {@link IcebergScanPlanProvider#getCountFromSummary} is null-safe. - * - *

WHY: the COUNT(*)-pushdown row count is read from the iceberg snapshot summary's {@code total-records} - * / {@code total-position-deletes} / {@code total-equality-deletes}. A compaction / replace / overwrite - * snapshot can OMIT one of those counters, and the pre-fix code (a faithful hand-port of the legacy, - * null-unsafe {@code IcebergScanNode.getCountFromSnapshot}) NPE-d on {@code summary.get(...).equals("0")} - * / {@code Long.parseLong(null)} — crashing the whole query instead of just declining the pushdown. The fix - * returns the {@code -1} "not pushable / unknown" sentinel (callers gate on {@code >= 0}) when any counter - * is absent. This is the connector-module analog of fe-core {@code IcebergCountPushDownTest}; the SPI - * migration copied the pre-fix logic, so fe-core carrying the fix did not protect the live path here. - */ -public class IcebergCountFromSummaryTest { - - // The three iceberg snapshot-summary counter keys (org.apache.iceberg.SnapshotSummary constants). - private static final String TOTAL_EQUALITY_DELETES = "total-equality-deletes"; - private static final String TOTAL_POSITION_DELETES = "total-position-deletes"; - private static final String TOTAL_RECORDS = "total-records"; - - /** Build a snapshot summary; a {@code null} arg OMITS that key — the exact absence the fix guards. */ - private static Map summary(String equalityDeletes, String positionDeletes, - String totalRecords) { - Map m = new HashMap<>(); - if (equalityDeletes != null) { - m.put(TOTAL_EQUALITY_DELETES, equalityDeletes); - } - if (positionDeletes != null) { - m.put(TOTAL_POSITION_DELETES, positionDeletes); - } - if (totalRecords != null) { - m.put(TOTAL_RECORDS, totalRecords); - } - return m; - } - - @Test - public void missingAnyCounterReturnsMinusOneInsteadOfNpe() { - // The regression: pre-fix each of these threw NPE (get(...).equals / parseLong(null)). Assert for - // BOTH dangling-delete flag values so the guard is proven independent of that branch. - for (boolean ignore : new boolean[] {false, true}) { - Assertions.assertEquals(-1L, - IcebergScanPlanProvider.getCountFromSummary(summary(null, "0", "100"), ignore), - "absent total-equality-deletes must decline pushdown, not NPE"); - Assertions.assertEquals(-1L, - IcebergScanPlanProvider.getCountFromSummary(summary("0", null, "100"), ignore), - "absent total-position-deletes must decline pushdown, not NPE"); - Assertions.assertEquals(-1L, - IcebergScanPlanProvider.getCountFromSummary(summary("0", "0", null), ignore), - "absent total-records must decline pushdown, not NPE"); - Assertions.assertEquals(-1L, - IcebergScanPlanProvider.getCountFromSummary(Collections.emptyMap(), ignore), - "empty summary must decline pushdown, not NPE"); - } - } - - @Test - public void noDeletesPushesTotalRecords() { - Assertions.assertEquals(100L, - IcebergScanPlanProvider.getCountFromSummary(summary("0", "0", "100"), false)); - } - - @Test - public void equalityDeletesNotPushable() { - // Equality deletes re-project at read time; the summary cannot net them out -> not pushable. - Assertions.assertEquals(-1L, - IcebergScanPlanProvider.getCountFromSummary(summary("3", "0", "100"), false)); - Assertions.assertEquals(-1L, - IcebergScanPlanProvider.getCountFromSummary(summary("3", "0", "100"), true)); - } - - @Test - public void positionDeletesHonorDanglingFlag() { - // ignore dangling deletes -> netted count (total - deletes) is pushable; otherwise not pushable. - Assertions.assertEquals(90L, - IcebergScanPlanProvider.getCountFromSummary(summary("0", "10", "100"), true)); - Assertions.assertEquals(-1L, - IcebergScanPlanProvider.getCountFromSummary(summary("0", "10", "100"), false)); - } - - @Test - public void allRowsDeletedNetsToZeroNotSentinel() { - // 100 records, 100 position deletes, ignore=true -> genuine 0. Must NOT collapse to the -1 sentinel - // (a real count of 0 is still a valid, pushable answer). - Assertions.assertEquals(0L, - IcebergScanPlanProvider.getCountFromSummary(summary("0", "100", "100"), true)); - } -} 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 9dc3c42da3311c..225dc3288f9c89 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 @@ -55,9 +55,11 @@ import org.apache.iceberg.Metrics; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; import org.apache.iceberg.StructLike; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; +import org.apache.iceberg.TableScan; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.inmemory.InMemoryCatalog; @@ -76,6 +78,8 @@ import java.io.FileNotFoundException; import java.io.IOException; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Proxy; import java.nio.ByteBuffer; import java.time.ZoneId; import java.time.ZoneOffset; @@ -143,6 +147,40 @@ private static DataFile dataFile(PartitionSpec spec, String path, long sizeBytes return builder.build(); } + private static Table tableWithSnapshotSummary(Table table, Map summaryOverrides) { + return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(), new Class[] {Table.class}, + (proxy, method, args) -> wrapSnapshotSummary(invoke(method, table, args), summaryOverrides)); + } + + private static Object wrapSnapshotSummary(Object value, Map summaryOverrides) { + if (value instanceof TableScan) { + TableScan scan = (TableScan) value; + return Proxy.newProxyInstance(TableScan.class.getClassLoader(), new Class[] {TableScan.class}, + (proxy, method, args) -> wrapSnapshotSummary(invoke(method, scan, args), summaryOverrides)); + } + if (value instanceof Snapshot) { + Snapshot snapshot = (Snapshot) value; + return Proxy.newProxyInstance(Snapshot.class.getClassLoader(), new Class[] {Snapshot.class}, + (proxy, method, args) -> { + if (method.getName().equals("summary")) { + Map summary = new HashMap<>(snapshot.summary()); + summary.putAll(summaryOverrides); + return summary; + } + return invoke(method, snapshot, args); + }); + } + return value; + } + + private static Object invoke(java.lang.reflect.Method method, Object target, Object[] args) throws Throwable { + try { + return method.invoke(target, args); + } catch (InvocationTargetException e) { + throw e.getCause(); + } + } + /** Run a range's BE-param population end-to-end (the generic node pre-sets table_format_type). */ private static TFileRangeDesc populate(ConnectorScanRange range) { TTableFormatFileDesc formatDesc = new TTableFormatFileDesc(); @@ -2114,14 +2152,23 @@ public void streamingSplitEstimateV3StaysSynchronous() { @Test public void streamingSplitEstimateServableCountPushdownStaysSynchronous() { - // A servable COUNT(*) collapses to one range (never streamed). 3 files, no deletes -> count servable from - // the snapshot summary. MUTATION: dropping the countPushdown short-circuit -> 3 -> red. + // COUNT(*) needs a complete live-file enumeration to prove an exact metadata count, so it never streams. + // MUTATION: dropping the countPushdown short-circuit -> 3 -> red. IcebergScanPlanProvider provider = providerOver(threeFileTable()); long estimate = provider.streamingSplitEstimate(batchSession(2, true), new IcebergTableHandle("db1", "t1"), Optional.empty(), true); Assertions.assertEquals(-1, estimate, "servable count pushdown must not stream"); } + @Test + public void streamingSplitEstimateFilteredCountUsesNormalScan() { + // File record counts cannot answer a filtered COUNT exactly, so retain normal streaming scan planning. + IcebergScanPlanProvider provider = providerOver(threeFileTable()); + long estimate = provider.streamingSplitEstimate(batchSession(2, true), + new IcebergTableHandle("db1", "t1"), Optional.of(eqInt("id", 1)), true); + Assertions.assertEquals(3L, estimate); + } + @Test public void streamSplitsProducesOneLazyRangePerFile() throws IOException { // The lazy source yields exactly one range per data file (3), with the raw paths preserved. This is the @@ -2531,7 +2578,7 @@ public void planScanNormalizesDataFilePathButKeepsOriginalFilePathRaw() { Assertions.assertEquals("oss://bucket/db/t1/f.parquet", fd.getOriginalFilePath()); } - // --- T05: COUNT(*) pushdown (getCountFromSnapshot + collapse-to-one count range, mirrors paimon) --- + // --- T05: COUNT(*) pushdown (live manifest count + collapse-to-one count range) --- private static List planCount(IcebergScanPlanProvider provider, ConnectorSession session, boolean countPushdown) { @@ -2570,6 +2617,27 @@ public void countPushdownCollapsesToSingleRangeWithTotalRecords() { Assertions.assertEquals(60L, populate(ranges.get(0)).getTableFormatParams().getTableLevelRowCount()); } + @Test + public void countPushdownUsesLiveDataFileRecordsWhenSnapshotSummaryIsWrong() { + Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned()); + table.newAppend() + .appendFile(dataFile(table.spec(), "s3://b/db/t1/f1.parquet", 1000, null, null)) + .appendFile(dataFile(table.spec(), "s3://b/db/t1/f2.parquet", 2000, null, null)) + .commit(); + Table tableWithWrongSummary = tableWithSnapshotSummary( + table, Collections.singletonMap("total-records", "999")); + Assertions.assertEquals("999", tableWithWrongSummary.currentSnapshot().summary().get("total-records"), + "precondition: the snapshot summary must contain a valid but incorrect positive count"); + IcebergScanPlanProvider provider = new IcebergScanPlanProvider( + IcebergCatalogProperties.of(Collections.emptyMap()), opsReturning(tableWithWrongSummary)); + + List ranges = planCount(provider, null, true); + + Assertions.assertEquals(1, ranges.size()); + Assertions.assertEquals(30L, ranges.get(0).getPushDownRowCount()); + Assertions.assertEquals(30L, populate(ranges.get(0)).getTableFormatParams().getTableLevelRowCount()); + } + @Test public void countPushdownNotAppliedWithEqualityDeletesScansAll() { Map v2 = Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"); @@ -2586,8 +2654,8 @@ public void countPushdownNotAppliedWithEqualityDeletesScansAll() { List ranges = planCount(provider, null, true); - // Equality deletes -> getCountFromSnapshot returns -1 -> fall back to the normal scan (every data file, - // each count -1 so BE reads & counts). MUTATION: pushing the count anyway -> 1 range / a count >= 0 -> red. + // Equality deletes prevent an exact manifest-only count, so fall back to the normal scan (every data + // file, each count -1 so BE reads and counts). Assertions.assertEquals(2, ranges.size()); for (ConnectorScanRange range : ranges) { Assertions.assertEquals(-1L, range.getPushDownRowCount()); @@ -2595,7 +2663,7 @@ public void countPushdownNotAppliedWithEqualityDeletesScansAll() { } @Test - public void countPushdownWithPositionDeletesNetsOutWhenIgnoringDangling() { + public void countPushdownWithPositionDeletesScansAllEvenWhenIgnoringDangling() { Map v2 = Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"); Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned(), v2); // 1000/100 = 10 data records. @@ -2617,11 +2685,10 @@ public void countPushdownWithPositionDeletesNetsOutWhenIgnoringDangling() { List ranges = planCount(provider, session, true); - // total-records(10) - total-position-deletes(3) = 7, pushable only because the session ignores dangling - // deletes. MUTATION: returning total-records (10) / not honoring the session flag -> wrong count -> red. + // A snapshot summary cannot prove whether position-delete entries are live or dangling. Even when the + // legacy compatibility flag is enabled, V2 must scan the data file rather than return an unverified count. Assertions.assertEquals(1, ranges.size()); - Assertions.assertEquals(7L, ranges.get(0).getPushDownRowCount()); - Assertions.assertEquals(7L, populate(ranges.get(0)).getTableFormatParams().getTableLevelRowCount()); + Assertions.assertEquals(-1L, ranges.get(0).getPushDownRowCount()); } @Test @@ -2654,9 +2721,7 @@ public void countPushdownWithPositionDeletesScansAllWhenNotIgnoringDangling() { @Test public void countPushdownEmptyTableProducesNoRanges() { - // Empty table (no snapshot) -> getCountFromSnapshot 0, but no representative file -> no range -> BE gets - // 0 ranges -> COUNT returns 0 (legacy returns empty splits too). MUTATION: emitting a synthetic count - // range with no path -> red (no file to build from). + // Empty table has no representative file, so BE gets 0 ranges and COUNT returns 0. Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned()); IcebergScanPlanProvider provider = new IcebergScanPlanProvider(IcebergCatalogProperties.of(Collections.emptyMap()), opsReturning(table)); @@ -2684,6 +2749,28 @@ public void countPushdownFalseDoesNormalMultiRangeScan() { } } + @Test + public void countPushdownWithRowFilterFallsBackToNormalScan() { + Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned()); + table.newAppend() + .appendFile(dataFile(table.spec(), "s3://b/db/t1/f1.parquet", 1000, null, null)) + .appendFile(dataFile(table.spec(), "s3://b/db/t1/f2.parquet", 1000, null, null)) + .commit(); + IcebergScanPlanProvider provider = new IcebergScanPlanProvider( + IcebergCatalogProperties.of(Collections.emptyMap()), opsReturning(table)); + + List ranges = provider.planScan(null, + ConnectorScanRequest.builder(new IcebergTableHandle("db1", "t1"), Collections.emptyList()) + .filter(Optional.of(eqInt("id", 1))) + .countPushdown(true) + .build()); + + Assertions.assertEquals(2, ranges.size()); + for (ConnectorScanRange range : ranges) { + Assertions.assertEquals(-1L, range.getPushDownRowCount()); + } + } + // --- T08: manifest-level scan planning (gated by meta.cache.iceberg.manifest.enable) --- private static Map manifestCacheProps() { @@ -2836,8 +2923,8 @@ public void streamSplitsManifestCacheFlatMapsAcrossDataManifests() throws IOExce } @Test - public void countPushdownManifestCacheMatchesCountAndReadsLazily() { - // Three appends -> three data manifests; record counts 10+20+30 = total-records 60 (snapshot summary). + public void countPushdownManifestCacheReadsAllLiveFilesWithoutMaterializingTasks() { + // Three appends -> three data manifests; required data-file record counts sum to 10+20+30 = 60. Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned()); table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/t1/f1.parquet", 1024, null, null)).commit(); table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/t1/f2.parquet", 2048, null, null)).commit(); @@ -2851,23 +2938,20 @@ public void countPushdownManifestCacheMatchesCountAndReadsLazily() { List cached = planCount(manifestProvider(manifestCacheProps(), table, cache), emptySession(), true); - // Same collapsed single range + same count (from the snapshot summary). The placeholder file path may + // Same collapsed single range + same manifest-derived count. The placeholder file path may // differ (SDK planFiles' ParallelIterable order is non-deterministic), so assert count + shape, not path. Assertions.assertEquals(1, sdk.size()); Assertions.assertEquals(1, cached.size()); Assertions.assertEquals(60L, cached.get(0).getPushDownRowCount()); Assertions.assertEquals(sdk.get(0).getPushDownRowCount(), cached.get(0).getPushDownRowCount()); - // Lazy early stop: COUNT needs only the first surviving file, so it must NOT read every data manifest. - // MUTATION: routing count through the materialized cache path -> reads all manifests -> size == total -> red. - Assertions.assertTrue(cache.size() >= 1 && cache.size() < totalManifests, - "count reads lazily (stops at the first file's manifest), not the whole table"); + // Exactness requires consuming every live data file, while the iterator still keeps FE memory bounded. + Assertions.assertEquals(totalManifests, cache.size()); } @Test public void countPushdownManifestCacheEmptyNullSnapshotReturnsNoRanges() { - // A never-appended table has no current snapshot; getCountFromSnapshot returns 0 (>=0) so planCountPushdown - // runs with a null-snapshot scan. cacheBackedFileScanTasks must keep the null-snapshot guard (empty - // iterable), not NPE. MUTATION: dropping the guard -> NPE on scan.snapshot() -> red. + // A never-appended table has no current snapshot. cacheBackedFileScanTasks must keep the null-snapshot + // guard (empty iterable), not NPE. MUTATION: dropping the guard -> NPE on scan.snapshot() -> red. Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned()); IcebergManifestCache cache = new IcebergManifestCache(); List cached = planCount(manifestProvider(manifestCacheProps(), table, cache), From a50ba1ff8209bddab65c6e3259926428d621ec18 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 14 Aug 2026 18:23:45 +0800 Subject: [PATCH 2/4] [optimize](iceberg) Avoid full file enumeration for count pushdown Use current manifest-list live-row aggregates for the common no-delete path and read only one representative file task. Fall back to bounded per-file counting when old manifests omit aggregates, and to a normal scan when live delete manifests are present. Tests: IcebergScanPlanProviderTest (148 tests) --- .../iceberg/IcebergConnectorMetadata.java | 4 +- .../iceberg/IcebergScanPlanProvider.java | 111 ++++++++++++++++-- .../iceberg/IcebergScanPlanProviderTest.java | 11 +- 3 files changed, 107 insertions(+), 19 deletions(-) diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java index fdaa9ff3704446..c6c327955eef5c 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java @@ -128,7 +128,7 @@ public class IcebergConnectorMetadata implements ConnectorMetadata { // Snapshot-summary keys for table-level row count (getTableStatistics). Local literal copies of the // spec-stable iceberg strings — byte-identical to legacy IcebergUtils.TOTAL_*. These remain optimizer - // estimates only; exact COUNT(*) pushdown deliberately derives its result from live data-file manifests. + // estimates only; exact COUNT(*) pushdown deliberately derives its result from live manifest-list counters. // All THREE keys are read: legacy getIcebergRowCount (via // getCountFromSummary, upstream 32a2651f66b / #64648) nets out position deletes AND gates the count to // UNKNOWN on any equality delete — see computeRowCount. @@ -823,7 +823,7 @@ public Optional getTableStatistics( * #64648): any equality delete ({@code total-equality-deletes} absent or {@code != "0"}) -> -1 (UNKNOWN), * since equality deletes re-project at read time and the summary cannot net them out; otherwise * {@code total-records - total-position-deletes}. This best-effort optimizer estimate is not used as an - * exact query result; COUNT(*) pushdown independently sums required record counts from live data manifests. + * exact query result; COUNT(*) pushdown independently sums live-row counters from the manifest list. * Empty table (no current snapshot) -> -1, which the caller maps to UNKNOWN. */ private static long computeRowCount(Table table) { 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 fd85cbaeb062d5..3c6db32a4639b6 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 @@ -700,10 +700,10 @@ private List planScanInternal( // per-file path normalization below, instead of rebuilding it per data/delete file (C3). UnaryOperator uriNormalizer = newUriNormalizer(vendedToken); - // COUNT(*) pushdown (T05): derive an exact count from the live data-file manifests and collapse the scan - // to a single whole-file range. Snapshot summary fields are optional writer-provided metadata and must - // never become a query result. If deletes or invalid record counts prevent an exact proof, fall through - // to the normal scan so BE reads and counts. + // COUNT(*) pushdown (T05): derive an exact count from the current manifest list and collapse the scan to + // a single whole-file range. Snapshot summary fields are optional writer-provided metadata and must never + // become a query result. If manifest aggregates are absent, fall back to bounded per-file enumeration; if + // deletes or invalid record counts prevent an exact proof, use the normal scan so BE reads and counts. // A data-row predicate can leave partially matching files, whose file-level recordCount is only an // upper bound. Keep those scans on the normal path even if the engine supplies the count signal. if (countPushdown && filter.isEmpty()) { @@ -1148,7 +1148,7 @@ private static boolean isPositionDeletesPartitionColumnRequested(List filter, ConnectorSession session) { @@ -1235,14 +1235,60 @@ private static Schema pinnedSchema(Table table, IcebergTableHandle handle) { } /** - * Build a collapsed COUNT(*) range while enumerating every live {@link FileScanTask}. The data-file - * {@code recordCount} field is required by the Iceberg spec, unlike optional snapshot summary fields. The - * optimization is valid only when every task has no attached delete file and every record count can be - * summed exactly. Otherwise the empty optional tells the caller to perform a normal scan. + * Build a collapsed COUNT(*) range from current manifest-list aggregates. Summing each data manifest's + * added and existing row counts is O(manifests), while only the first live {@link FileScanTask} is needed as + * the representative range. Old manifest lists that omit these aggregates use the bounded O(files) fallback. + * Any live delete file makes the optimization unsafe and tells the caller to perform a normal scan. */ private Optional> planCountPushdown(Table table, TableScan scan, int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, UnaryOperator uriNormalizer, ConnectorSession session, Optional filter) { + Snapshot snapshot = scan.snapshot(); + if (snapshot == null) { + return Optional.of(Collections.emptyList()); + } + + ManifestDeleteState deleteState = manifestDeleteState(snapshot.deleteManifests(table.io())); + if (deleteState == ManifestDeleteState.PRESENT) { + return Optional.empty(); + } + if (deleteState == ManifestDeleteState.NONE) { + OptionalLong manifestCount = liveRowCountFromManifests(snapshot.dataManifests(table.io())); + if (manifestCount.isPresent()) { + return planManifestCountRange(table, scan, manifestCount.getAsLong(), formatVersion, + partitioned, orderedPartitionKeys, zone, uriNormalizer, session, filter); + } + } + + // Older manifest lists may omit aggregate counters. Preserve correctness by falling back to the + // bounded per-file enumeration instead of trusting snapshot summary metadata. + return planCountPushdownFromFileTasks(table, scan, formatVersion, partitioned, + orderedPartitionKeys, zone, uriNormalizer, session, filter); + } + + private Optional> planManifestCountRange(Table table, TableScan scan, long exactCount, + int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, + UnaryOperator uriNormalizer, ConnectorSession session, Optional filter) { + try (CloseableIterable tasks = countPushdownFileScanTasks(scan, session, table, filter)) { + for (FileScanTask task : tasks) { + // Manifest-list delete counts are authoritative, but retain this defensive check for malformed + // metadata before exposing the aggregate as a query result. + if (!task.deletes().isEmpty() || task.file().recordCount() < 0) { + return Optional.empty(); + } + return Optional.of(Collections.singletonList(buildRange(table, task.file(), task, formatVersion, + partitioned, orderedPartitionKeys, zone, uriNormalizer, exactCount, -1, null))); + } + } catch (IOException e) { + throw new RuntimeException("Failed to plan iceberg count-pushdown file, error message is:" + + e.getMessage(), e); + } + return exactCount == 0 ? Optional.of(Collections.emptyList()) : Optional.empty(); + } + + private Optional> planCountPushdownFromFileTasks(Table table, TableScan scan, + int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, + UnaryOperator uriNormalizer, ConnectorSession session, Optional filter) { FileScanTask representative = null; long exactCount = 0; try (CloseableIterable tasks = countPushdownFileScanTasks(scan, session, table, filter)) { @@ -1273,10 +1319,51 @@ private Optional> planCountPushdown(Table table, TableS formatVersion, partitioned, orderedPartitionKeys, zone, uriNormalizer, exactCount, -1, null))); } + private enum ManifestDeleteState { + NONE, + PRESENT, + UNKNOWN + } + + private static ManifestDeleteState manifestDeleteState(List manifests) { + for (ManifestFile manifest : manifests) { + Integer addedFiles = manifest.addedFilesCount(); + Integer existingFiles = manifest.existingFilesCount(); + if (manifest.content() != ManifestContent.DELETES || addedFiles == null || existingFiles == null + || addedFiles < 0 || existingFiles < 0) { + return ManifestDeleteState.UNKNOWN; + } + if (addedFiles > 0 || existingFiles > 0) { + return ManifestDeleteState.PRESENT; + } + } + return ManifestDeleteState.NONE; + } + + private static OptionalLong liveRowCountFromManifests(List manifests) { + long exactCount = 0; + for (ManifestFile manifest : manifests) { + Long addedRows = manifest.addedRowsCount(); + Long existingRows = manifest.existingRowsCount(); + if (manifest.content() != ManifestContent.DATA || addedRows == null || existingRows == null + || addedRows < 0 || existingRows < 0) { + return OptionalLong.empty(); + } + try { + exactCount = Math.addExact(exactCount, addedRows); + exactCount = Math.addExact(exactCount, existingRows); + } catch (ArithmeticException e) { + return OptionalLong.empty(); + } + } + return OptionalLong.of(exactCount); + } + /** - * The COUNT(*)-pushdown enumeration. PERF-04 (C18): when the manifest cache is enabled, read through the lazy - * {@link #cacheBackedFileScanTasks} (stats overload — this runs on the single planning thread) so manifest - * reads are cache hits without materializing the table's task list in FE memory. + * The COUNT(*)-pushdown representative/fallback enumeration. PERF-04 (C18): when the manifest cache is + * enabled, read through the lazy {@link #cacheBackedFileScanTasks} (stats overload — this runs on the single + * planning thread) so the fast path stops after one representative file and the old-metadata fallback remains + * bounded without materializing the table's task list in FE memory. * An eager cache failure falls back to the SDK path (mirrors {@link #planFileScanTask}). The first surviving * (pruned) file may differ from the SDK path's first file (its {@code ParallelIterable} order is * non-deterministic), but BE ignores the representative file. Cache disabled -> the SDK path, byte-unchanged. 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 225dc3288f9c89..c1bff16d5f9b86 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 @@ -2923,8 +2923,8 @@ public void streamSplitsManifestCacheFlatMapsAcrossDataManifests() throws IOExce } @Test - public void countPushdownManifestCacheReadsAllLiveFilesWithoutMaterializingTasks() { - // Three appends -> three data manifests; required data-file record counts sum to 10+20+30 = 60. + public void countPushdownManifestCacheReadsOnlyRepresentativeFile() { + // Three appends -> three data manifests; manifest-list live-row aggregates sum to 10+20+30 = 60. Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned()); table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/t1/f1.parquet", 1024, null, null)).commit(); table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/t1/f2.parquet", 2048, null, null)).commit(); @@ -2938,14 +2938,15 @@ public void countPushdownManifestCacheReadsAllLiveFilesWithoutMaterializingTasks List cached = planCount(manifestProvider(manifestCacheProps(), table, cache), emptySession(), true); - // Same collapsed single range + same manifest-derived count. The placeholder file path may + // Same collapsed single range + same manifest-list-derived count. The placeholder file path may // differ (SDK planFiles' ParallelIterable order is non-deterministic), so assert count + shape, not path. Assertions.assertEquals(1, sdk.size()); Assertions.assertEquals(1, cached.size()); Assertions.assertEquals(60L, cached.get(0).getPushDownRowCount()); Assertions.assertEquals(sdk.get(0).getPushDownRowCount(), cached.get(0).getPushDownRowCount()); - // Exactness requires consuming every live data file, while the iterator still keeps FE memory bounded. - Assertions.assertEquals(totalManifests, cache.size()); + // The manifest list already carries exact live-row aggregates. Only the first manifest's entries are + // needed to obtain a representative file for the collapsed range. + Assertions.assertEquals(1, cache.size()); } @Test From 79644f90de9df6473e16b8498fe06b7061b8fe74 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 14 Aug 2026 19:59:33 +0800 Subject: [PATCH 3/4] [fix](iceberg) Address count pushdown review feedback --- be/src/format/table/iceberg_reader_mixin.h | 9 +- .../iceberg/IcebergScanPlanProvider.java | 102 +++++++++++-- .../iceberg/IcebergScanPlanProviderTest.java | 136 ++++++++++++++++-- .../org/apache/doris/qe/SessionVariable.java | 10 +- 4 files changed, 221 insertions(+), 36 deletions(-) diff --git a/be/src/format/table/iceberg_reader_mixin.h b/be/src/format/table/iceberg_reader_mixin.h index 55064c6687dbae..437b76d0e2d544 100644 --- a/be/src/format/table/iceberg_reader_mixin.h +++ b/be/src/format/table/iceberg_reader_mixin.h @@ -534,11 +534,10 @@ class IcebergReaderMixin : public BaseReader, public TableSchemaChangeHelper { template Status IcebergReaderMixin::_init_row_filters() { - // COUNT(*) short-circuit. A table-level row count of 0 (e.g. an all-deleted table read with - // ignore_iceberg_dangling_delete, where total-records == total-position-deletes) is still a - // valid pushed-down count, so accept >= 0 -- matching FileScanner and the Paimon readers. FE - // sends -1 when there is no table-level count; using > 0 here would drop a genuine 0 into the - // delete-applying path below and never produce the intended CountReader(0). + // COUNT(*) short-circuit. A table-level row count of 0 (an empty current snapshot) is still a + // valid pushed-down count, so accept >= 0 -- matching FileScanner and the Paimon readers. FE sends + // -1 when there is no table-level count; using > 0 here would drop a genuine 0 into the normal read + // path below and never produce the intended CountReader(0). if (this->_push_down_agg_type == TPushAggOp::type::COUNT && this->get_scan_range().table_format_params.__isset.table_level_row_count && this->get_scan_range().table_format_params.table_level_row_count >= 0) { 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 3c6db32a4639b6..f22b7a550dc820 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 @@ -150,6 +150,7 @@ public class IcebergScanPlanProvider implements ConnectorScanPlanProvider { // FIX-M3 streaming (file-count) batch gate — keys byte-identical to fe-core SessionVariable. private static final String ENABLE_EXTERNAL_TABLE_BATCH_MODE = "enable_external_table_batch_mode"; private static final String NUM_FILES_IN_BATCH_MODE = "num_files_in_batch_mode"; + private static final String IGNORE_ICEBERG_DANGLING_DELETE = "ignore_iceberg_dangling_delete"; private static final long DEFAULT_NUM_FILES_IN_BATCH_MODE = 1024L; // Equality-delete schema discovery uses this stable Iceberg snapshot-summary key as a read-avoidance hint. @@ -443,7 +444,6 @@ public long streamingSplitEstimate(ConnectorSession session, ConnectorTableHandl Optional filter, boolean countPushdown) { IcebergTableHandle iceHandle = (IcebergTableHandle) handle; if (iceHandle.isResolvedEmptySnapshot() || iceHandle.isSystemTable() - || (countPushdown && filter.isEmpty()) || !sessionBool(session, ENABLE_EXTERNAL_TABLE_BATCH_MODE, true)) { return -1; } @@ -453,6 +453,19 @@ public long streamingSplitEstimate(ConnectorSession session, ConnectorTableHandl if (snapshot == null) { return -1; } + if (countPushdown && filter.isEmpty()) { + boolean ignorePositionDeletes = sessionBool(session, IGNORE_ICEBERG_DANGLING_DELETE, false); + ManifestDeleteState deleteState = manifestDeleteState(snapshot.deleteManifests(table.io())); + // Keep synchronous planning only while COUNT(*) can still collapse to one range. Once live deletes + // make that impossible, normal file enumeration needs the same streaming OOM protection as SELECT. + if (deleteState == ManifestDeleteState.NONE + || (deleteState == ManifestDeleteState.PRESENT && ignorePositionDeletes + && !hasNonIgnorableDeleteFiles(table, snapshot, true)) + || (deleteState == ManifestDeleteState.UNKNOWN + && !hasNonIgnorableDeleteFiles(table, snapshot, ignorePositionDeletes))) { + return -1; + } + } if (getFormatVersion(table) >= 3) { return -1; } @@ -1238,7 +1251,8 @@ private static Schema pinnedSchema(Table table, IcebergTableHandle handle) { * Build a collapsed COUNT(*) range from current manifest-list aggregates. Summing each data manifest's * added and existing row counts is O(manifests), while only the first live {@link FileScanTask} is needed as * the representative range. Old manifest lists that omit these aggregates use the bounded O(files) fallback. - * Any live delete file makes the optimization unsafe and tells the caller to perform a normal scan. + * Equality deletes and non-ignored position deletes make the optimization unsafe and tell the caller to + * perform a normal scan. */ private Optional> planCountPushdown(Table table, TableScan scan, int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, @@ -1248,32 +1262,36 @@ private Optional> planCountPushdown(Table table, TableS return Optional.of(Collections.emptyList()); } + boolean ignorePositionDeletes = sessionBool(session, IGNORE_ICEBERG_DANGLING_DELETE, false); ManifestDeleteState deleteState = manifestDeleteState(snapshot.deleteManifests(table.io())); - if (deleteState == ManifestDeleteState.PRESENT) { + if (deleteState == ManifestDeleteState.PRESENT + && (!ignorePositionDeletes || hasNonIgnorableDeleteFiles(table, snapshot, true))) { return Optional.empty(); } - if (deleteState == ManifestDeleteState.NONE) { + if (deleteState != ManifestDeleteState.UNKNOWN) { OptionalLong manifestCount = liveRowCountFromManifests(snapshot.dataManifests(table.io())); if (manifestCount.isPresent()) { return planManifestCountRange(table, scan, manifestCount.getAsLong(), formatVersion, - partitioned, orderedPartitionKeys, zone, uriNormalizer, session, filter); + partitioned, orderedPartitionKeys, zone, uriNormalizer, session, filter, + ignorePositionDeletes); } } // Older manifest lists may omit aggregate counters. Preserve correctness by falling back to the // bounded per-file enumeration instead of trusting snapshot summary metadata. return planCountPushdownFromFileTasks(table, scan, formatVersion, partitioned, - orderedPartitionKeys, zone, uriNormalizer, session, filter); + orderedPartitionKeys, zone, uriNormalizer, session, filter, ignorePositionDeletes); } private Optional> planManifestCountRange(Table table, TableScan scan, long exactCount, int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, - UnaryOperator uriNormalizer, ConnectorSession session, Optional filter) { + UnaryOperator uriNormalizer, ConnectorSession session, Optional filter, + boolean ignorePositionDeletes) { try (CloseableIterable tasks = countPushdownFileScanTasks(scan, session, table, filter)) { for (FileScanTask task : tasks) { // Manifest-list delete counts are authoritative, but retain this defensive check for malformed // metadata before exposing the aggregate as a query result. - if (!task.deletes().isEmpty() || task.file().recordCount() < 0) { + if (hasNonIgnorableTaskDeletes(task, ignorePositionDeletes) || task.file().recordCount() < 0) { return Optional.empty(); } return Optional.of(Collections.singletonList(buildRange(table, task.file(), task, formatVersion, @@ -1288,13 +1306,38 @@ private Optional> planManifestCountRange(Table table, T private Optional> planCountPushdownFromFileTasks(Table table, TableScan scan, int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, - UnaryOperator uriNormalizer, ConnectorSession session, Optional filter) { + UnaryOperator uriNormalizer, ConnectorSession session, Optional filter, + boolean ignorePositionDeletes) { + if (!isManifestCacheEnabled()) { + return accumulateCountPushdownFileTasks(table, scan.planFiles(), formatVersion, partitioned, + orderedPartitionKeys, zone, uriNormalizer, ignorePositionDeletes); + } + String statsQueryId = session != null ? session.getQueryId() : null; + try { + return accumulateCountPushdownFileTasks(table, + cacheBackedFileScanTasks(scan, session, table, filter, statsQueryId), formatVersion, + partitioned, orderedPartitionKeys, zone, uriNormalizer, ignorePositionDeletes); + } catch (Exception e) { + LOG.warn("Iceberg count-pushdown plan with manifest cache failed, falling back to SDK scan: {}", + e.getMessage(), e); + manifestCache.recordFailure(statsQueryId); + // The retry owns a fresh accumulator so rows consumed before a lazy cache failure are never counted + // twice and the SDK fallback remains an exact restart. + return accumulateCountPushdownFileTasks(table, scan.planFiles(), formatVersion, partitioned, + orderedPartitionKeys, zone, uriNormalizer, ignorePositionDeletes); + } + } + + private Optional> accumulateCountPushdownFileTasks(Table table, + CloseableIterable tasks, int formatVersion, boolean partitioned, + List orderedPartitionKeys, ZoneId zone, UnaryOperator uriNormalizer, + boolean ignorePositionDeletes) { FileScanTask representative = null; long exactCount = 0; - try (CloseableIterable tasks = countPushdownFileScanTasks(scan, session, table, filter)) { - for (FileScanTask task : tasks) { + try (CloseableIterable closeableTasks = tasks) { + for (FileScanTask task : closeableTasks) { // A metadata count is safe only when manifests alone prove the exact visible row count. - if (!task.deletes().isEmpty() || task.file().recordCount() < 0) { + if (hasNonIgnorableTaskDeletes(task, ignorePositionDeletes) || task.file().recordCount() < 0) { return Optional.empty(); } try { @@ -1319,6 +1362,18 @@ private Optional> planCountPushdownFromFileTasks(Table formatVersion, partitioned, orderedPartitionKeys, zone, uriNormalizer, exactCount, -1, null))); } + private static boolean hasNonIgnorableTaskDeletes(FileScanTask task, boolean ignorePositionDeletes) { + if (task.deletes() == null) { + return false; + } + for (DeleteFile delete : task.deletes()) { + if (!ignorePositionDeletes || delete.content() != FileContent.POSITION_DELETES) { + return true; + } + } + return false; + } + private enum ManifestDeleteState { NONE, PRESENT, @@ -1340,6 +1395,24 @@ private static ManifestDeleteState manifestDeleteState(List manife return ManifestDeleteState.NONE; } + private static boolean hasNonIgnorableDeleteFiles(Table table, Snapshot snapshot, + boolean ignorePositionDeletes) { + for (ManifestFile manifest : snapshot.deleteManifests(table.io())) { + try (ManifestReader reader = ManifestFiles.readDeleteManifest( + manifest, table.io(), table.specs())) { + for (DeleteFile delete : reader) { + if (!ignorePositionDeletes || delete.content() != FileContent.POSITION_DELETES) { + return true; + } + } + } catch (IOException e) { + throw new DorisConnectorException( + "Failed to inspect iceberg delete manifest " + manifest.path() + ": " + e.getMessage(), e); + } + } + return false; + } + private static OptionalLong liveRowCountFromManifests(List manifests) { long exactCount = 0; for (ManifestFile manifest : manifests) { @@ -1371,12 +1444,13 @@ private static OptionalLong liveRowCountFromManifests(List manifes private CloseableIterable countPushdownFileScanTasks(TableScan scan, ConnectorSession session, Table table, Optional filter) { if (isManifestCacheEnabled()) { + String statsQueryId = session != null ? session.getQueryId() : null; try { - return cacheBackedFileScanTasks(scan, session, table, filter, session.getQueryId()); + return cacheBackedFileScanTasks(scan, session, table, filter, statsQueryId); } catch (Exception e) { LOG.warn("Iceberg count-pushdown plan with manifest cache failed, falling back to SDK scan: {}", e.getMessage(), e); - manifestCache.recordFailure(session.getQueryId()); + manifestCache.recordFailure(statsQueryId); } } return scan.planFiles(); 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 c1bff16d5f9b86..3634a26f503e65 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 @@ -51,6 +51,7 @@ import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileMetadata; import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.ManifestFile; import org.apache.iceberg.MetadataColumns; import org.apache.iceberg.Metrics; import org.apache.iceberg.PartitionSpec; @@ -93,6 +94,7 @@ import java.util.NoSuchElementException; import java.util.Optional; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.UnaryOperator; /** @@ -152,6 +154,48 @@ private static Table tableWithSnapshotSummary(Table table, Map s (proxy, method, args) -> wrapSnapshotSummary(invoke(method, table, args), summaryOverrides)); } + private static Table tableWithMissingManifestRowsAndIo(Table table, FileIO fileIO) { + return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(), new Class[] {Table.class}, + (proxy, method, args) -> { + if (method.getName().equals("io")) { + return fileIO; + } + return wrapMissingManifestRows(invoke(method, table, args)); + }); + } + + private static Object wrapMissingManifestRows(Object value) { + if (value instanceof TableScan) { + TableScan scan = (TableScan) value; + return Proxy.newProxyInstance(TableScan.class.getClassLoader(), new Class[] {TableScan.class}, + (proxy, method, args) -> wrapMissingManifestRows(invoke(method, scan, args))); + } + if (value instanceof Snapshot) { + Snapshot snapshot = (Snapshot) value; + return Proxy.newProxyInstance(Snapshot.class.getClassLoader(), new Class[] {Snapshot.class}, + (proxy, method, args) -> { + Object result = invoke(method, snapshot, args); + if (!method.getName().equals("dataManifests")) { + return result; + } + List manifests = new ArrayList<>(); + for (Object manifestValue : (List) result) { + ManifestFile manifest = (ManifestFile) manifestValue; + manifests.add((ManifestFile) Proxy.newProxyInstance(ManifestFile.class.getClassLoader(), + new Class[] {ManifestFile.class}, (manifestProxy, manifestMethod, manifestArgs) -> { + if (manifestMethod.getName().equals("addedRowsCount") + || manifestMethod.getName().equals("existingRowsCount")) { + return null; + } + return invoke(manifestMethod, manifest, manifestArgs); + })); + } + return manifests; + }); + } + return value; + } + private static Object wrapSnapshotSummary(Object value, Map summaryOverrides) { if (value instanceof TableScan) { TableScan scan = (TableScan) value; @@ -2160,6 +2204,21 @@ public void streamingSplitEstimateServableCountPushdownStaysSynchronous() { Assertions.assertEquals(-1, estimate, "servable count pushdown must not stream"); } + @Test + public void streamingSplitEstimateCountWithLiveDeleteUsesNormalStreamingScan() { + Table table = threeFileTable(Collections.singletonMap(TableProperties.FORMAT_VERSION, "2")); + table.newRowDelta() + .addDeletes(positionDeleteFile("s3://b/db/t1/pos.parquet", FileFormat.PARQUET, null, null)) + .commit(); + IcebergScanPlanProvider provider = providerOver(table); + + long estimate = provider.streamingSplitEstimate(batchSession(2, true), + new IcebergTableHandle("db1", "t1"), Optional.empty(), true); + + Assertions.assertEquals(3L, estimate, + "a live delete prevents metadata collapse but must retain the backpressured normal scan"); + } + @Test public void streamingSplitEstimateFilteredCountUsesNormalScan() { // File record counts cannot answer a filtered COUNT exactly, so retain normal streaming scan planning. @@ -2652,10 +2711,12 @@ public void countPushdownNotAppliedWithEqualityDeletesScansAll() { .commit(); IcebergScanPlanProvider provider = new IcebergScanPlanProvider(IcebergCatalogProperties.of(Collections.emptyMap()), opsReturning(table)); - List ranges = planCount(provider, null, true); + ConnectorSession session = new FakeScanSession("UTC", + Collections.singletonMap("ignore_iceberg_dangling_delete", "true")); + List ranges = planCount(provider, session, true); - // Equality deletes prevent an exact manifest-only count, so fall back to the normal scan (every data - // file, each count -1 so BE reads and counts). + // The dangling-delete compatibility flag applies only to position deletes. Equality deletes still force + // the normal scan (every data file, each count -1 so BE reads and counts). Assertions.assertEquals(2, ranges.size()); for (ConnectorScanRange range : ranges) { Assertions.assertEquals(-1L, range.getPushDownRowCount()); @@ -2663,7 +2724,7 @@ public void countPushdownNotAppliedWithEqualityDeletesScansAll() { } @Test - public void countPushdownWithPositionDeletesScansAllEvenWhenIgnoringDangling() { + public void countPushdownWithPositionDeletesUsesDataRowsWhenIgnoringDangling() { Map v2 = Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"); Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned(), v2); // 1000/100 = 10 data records. @@ -2685,10 +2746,10 @@ public void countPushdownWithPositionDeletesScansAllEvenWhenIgnoringDangling() { List ranges = planCount(provider, session, true); - // A snapshot summary cannot prove whether position-delete entries are live or dangling. Even when the - // legacy compatibility flag is enabled, V2 must scan the data file rather than return an unverified count. + // The compatibility flag explicitly ignores position deletes. Use current data-manifest rows rather than + // the optional snapshot summary, preserving the public contract without reintroducing the original bug. Assertions.assertEquals(1, ranges.size()); - Assertions.assertEquals(-1L, ranges.get(0).getPushDownRowCount()); + Assertions.assertEquals(10L, ranges.get(0).getPushDownRowCount()); } @Test @@ -2849,9 +2910,6 @@ public void planScanManifestCachePrunesPartitionLikeSdk() { } // --- PERF-04: streaming (C17) + COUNT(*) (C18) paths read through the manifest cache, LAZILY --- - // (fallback-to-SDK on a cache-read failure is not unit-tested: IcebergManifestCache is final so it cannot be - // made to throw, exactly as the pre-existing synchronous planFileScanTask fallback is untested; the streaming/ - // count catch(Exception)+recordFailure mirrors that path verbatim.) @Test public void streamSplitsManifestCacheEnabledMatchesSdkPathAndConsumesCache() throws IOException { @@ -2949,6 +3007,29 @@ public void countPushdownManifestCacheReadsOnlyRepresentativeFile() { Assertions.assertEquals(1, cache.size()); } + @Test + public void countPushdownLateManifestCacheFailureRetriesSdk() { + Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned()); + table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/t1/f1.parquet", 1024, null, null)).commit(); + table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/t1/f2.parquet", 2048, null, null)).commit(); + table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/t1/f3.parquet", 3072, null, null)).commit(); + List manifests = table.currentSnapshot().dataManifests(table.io()); + Assertions.assertTrue(manifests.size() >= 2, "precondition: failure must occur after iteration starts"); + + FailOnceFileIO fileIO = new FailOnceFileIO(table.io(), manifests.get(1).path()); + Table wrapped = tableWithMissingManifestRowsAndIo(table, fileIO); + IcebergManifestCache cache = new IcebergManifestCache(); + + // A lazy phase-two failure must discard the partial accumulator before retrying the SDK path. + List ranges = planCount( + manifestProvider(manifestCacheProps(), wrapped, cache), emptySession(), true); + + Assertions.assertEquals(1, ranges.size()); + Assertions.assertEquals(60L, ranges.get(0).getPushDownRowCount()); + Assertions.assertTrue(fileIO.failed.get()); + Assertions.assertEquals(1L, cache.takeStats("q")[2]); + } + @Test public void countPushdownManifestCacheEmptyNullSnapshotReturnsNoRanges() { // A never-appended table has no current snapshot. cacheBackedFileScanTasks must keep the null-snapshot @@ -3975,6 +4056,41 @@ public void deleteFile(String path) { } } + /** Injects one manifest read failure while leaving the SDK scan's own FileIO untouched. */ + private static final class FailOnceFileIO implements FileIO { + private final FileIO delegate; + private final String failingPath; + private final AtomicBoolean failed = new AtomicBoolean(); + + FailOnceFileIO(FileIO delegate, String failingPath) { + this.delegate = delegate; + this.failingPath = failingPath; + } + + @Override + public Map properties() { + return delegate.properties(); + } + + @Override + public InputFile newInputFile(String path) { + if (failingPath.equals(path) && failed.compareAndSet(false, true)) { + throw new RuntimeException("injected late manifest read failure"); + } + return delegate.newInputFile(path); + } + + @Override + public OutputFile newOutputFile(String path) { + return delegate.newOutputFile(path); + } + + @Override + public void deleteFile(String path) { + delegate.deleteFile(path); + } + } + /** A fake FileIO that ALSO vends StorageCredentials (a REST catalog's delegated creds). */ private static final class VendedFileIO implements FileIO, SupportsStorageCredentials { private final Map props; diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index a1592de2947489..24ebc99e8341c4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -3565,13 +3565,9 @@ public void checkAnnIndexCandidateRowsPercentThreshold(String value) { public static final String IGNORE_ICEBERG_DANGLING_DELETE = "ignore_iceberg_dangling_delete"; @VarAttrDef.VarAttr(name = IGNORE_ICEBERG_DANGLING_DELETE, - description = " Whether to ignore the impact of dangling delete files in Iceberg tables on COUNT(*) " - + "statistics. " - + "The default is true, COUNT(*) will directly obtain the number of rows from metadata, " - + "which has better performance, but if there are dangling deletes, " - + "the result may be inaccurate. " - + "When set to false, COUNT(*) will scan data files " - + "to exclude the impact of dangling delete files.") + description = "Whether Iceberg metadata COUNT(*) may ignore position delete files. " + + "This improves performance but can make the result inaccurate when live position deletes " + + "exist. Equality deletes always disable metadata COUNT(*).") public boolean ignoreIcebergDanglingDelete = false; @VarAttrDef.VarAttr(name = ENABLE_ICEBERG_MERGE_PARTITIONING, From 12476576e1d6c52de872d143bb9080fdcdb13c1d Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 14 Aug 2026 20:58:01 +0800 Subject: [PATCH 4/4] [fix](iceberg) Preserve count fallback semantics --- .../iceberg/IcebergScanPlanProvider.java | 160 +++++++++++------- .../iceberg/IcebergScanPlanProviderTest.java | 39 ++++- .../org/apache/doris/qe/SessionVariable.java | 6 +- 3 files changed, 136 insertions(+), 69 deletions(-) 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 f22b7a550dc820..11963dfcadbebb 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 @@ -454,17 +454,20 @@ public long streamingSplitEstimate(ConnectorSession session, ConnectorTableHandl return -1; } if (countPushdown && filter.isEmpty()) { - boolean ignorePositionDeletes = sessionBool(session, IGNORE_ICEBERG_DANGLING_DELETE, false); + boolean netPositionDeletes = sessionBool(session, IGNORE_ICEBERG_DANGLING_DELETE, false); ManifestDeleteState deleteState = manifestDeleteState(snapshot.deleteManifests(table.io())); // Keep synchronous planning only while COUNT(*) can still collapse to one range. Once live deletes // make that impossible, normal file enumeration needs the same streaming OOM protection as SELECT. - if (deleteState == ManifestDeleteState.NONE - || (deleteState == ManifestDeleteState.PRESENT && ignorePositionDeletes - && !hasNonIgnorableDeleteFiles(table, snapshot, true)) - || (deleteState == ManifestDeleteState.UNKNOWN - && !hasNonIgnorableDeleteFiles(table, snapshot, ignorePositionDeletes))) { + if (deleteState == ManifestDeleteState.NONE) { return -1; } + if (deleteState != ManifestDeleteState.PRESENT || netPositionDeletes) { + OptionalLong positionDeleteRows = livePositionDeleteRowCount(table, snapshot); + if (positionDeleteRows.isPresent() + && (netPositionDeletes || positionDeleteRows.getAsLong() == 0)) { + return -1; + } + } } if (getFormatVersion(table) >= 3) { return -1; @@ -1251,7 +1254,7 @@ private static Schema pinnedSchema(Table table, IcebergTableHandle handle) { * Build a collapsed COUNT(*) range from current manifest-list aggregates. Summing each data manifest's * added and existing row counts is O(manifests), while only the first live {@link FileScanTask} is needed as * the representative range. Old manifest lists that omit these aggregates use the bounded O(files) fallback. - * Equality deletes and non-ignored position deletes make the optimization unsafe and tell the caller to + * Equality deletes and non-netted position deletes make the optimization unsafe and tell the caller to * perform a normal scan. */ private Optional> planCountPushdown(Table table, TableScan scan, @@ -1262,36 +1265,72 @@ private Optional> planCountPushdown(Table table, TableS return Optional.of(Collections.emptyList()); } - boolean ignorePositionDeletes = sessionBool(session, IGNORE_ICEBERG_DANGLING_DELETE, false); + boolean netPositionDeletes = sessionBool(session, IGNORE_ICEBERG_DANGLING_DELETE, false); ManifestDeleteState deleteState = manifestDeleteState(snapshot.deleteManifests(table.io())); - if (deleteState == ManifestDeleteState.PRESENT - && (!ignorePositionDeletes || hasNonIgnorableDeleteFiles(table, snapshot, true))) { + if (deleteState == ManifestDeleteState.PRESENT && !netPositionDeletes) { return Optional.empty(); } - if (deleteState != ManifestDeleteState.UNKNOWN) { - OptionalLong manifestCount = liveRowCountFromManifests(snapshot.dataManifests(table.io())); - if (manifestCount.isPresent()) { - return planManifestCountRange(table, scan, manifestCount.getAsLong(), formatVersion, - partitioned, orderedPartitionKeys, zone, uriNormalizer, session, filter, - ignorePositionDeletes); + OptionalLong positionDeleteRows = deleteState == ManifestDeleteState.NONE + ? OptionalLong.of(0) + : livePositionDeleteRowCount(table, snapshot); + if (!positionDeleteRows.isPresent() + || (!netPositionDeletes && positionDeleteRows.getAsLong() != 0)) { + return Optional.empty(); + } + + OptionalLong manifestCount = liveRowCountFromManifests(snapshot.dataManifests(table.io())); + if (manifestCount.isPresent()) { + // Compatibility mode nets each live position-delete file's rows once without trusting summary + // counters; it intentionally cannot distinguish dangling entries, which is why the flag is opt-in. + OptionalLong visibleRows = subtractPositionDeleteRows( + manifestCount.getAsLong(), netPositionDeletes ? positionDeleteRows.getAsLong() : 0); + if (!visibleRows.isPresent()) { + return Optional.empty(); } + return planManifestCountRange(table, scan, visibleRows.getAsLong(), formatVersion, + partitioned, orderedPartitionKeys, zone, uriNormalizer, session, filter, + netPositionDeletes); } // Older manifest lists may omit aggregate counters. Preserve correctness by falling back to the // bounded per-file enumeration instead of trusting snapshot summary metadata. return planCountPushdownFromFileTasks(table, scan, formatVersion, partitioned, - orderedPartitionKeys, zone, uriNormalizer, session, filter, ignorePositionDeletes); + orderedPartitionKeys, zone, uriNormalizer, session, filter, netPositionDeletes, + netPositionDeletes ? positionDeleteRows.getAsLong() : 0); } private Optional> planManifestCountRange(Table table, TableScan scan, long exactCount, int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, UnaryOperator uriNormalizer, ConnectorSession session, Optional filter, - boolean ignorePositionDeletes) { - try (CloseableIterable tasks = countPushdownFileScanTasks(scan, session, table, filter)) { - for (FileScanTask task : tasks) { - // Manifest-list delete counts are authoritative, but retain this defensive check for malformed - // metadata before exposing the aggregate as a query result. - if (hasNonIgnorableTaskDeletes(task, ignorePositionDeletes) || task.file().recordCount() < 0) { + boolean netPositionDeletes) { + if (!isManifestCacheEnabled()) { + return buildManifestCountRange(table, scan.planFiles(), exactCount, formatVersion, partitioned, + orderedPartitionKeys, zone, uriNormalizer, netPositionDeletes); + } + String statsQueryId = session != null ? session.getQueryId() : null; + try { + return buildManifestCountRange(table, + cacheBackedFileScanTasks(scan, session, table, filter, statsQueryId), exactCount, + formatVersion, partitioned, orderedPartitionKeys, zone, uriNormalizer, netPositionDeletes); + } catch (Exception e) { + LOG.warn("Iceberg count-pushdown representative plan with manifest cache failed, " + + "falling back to SDK scan: {}", e.getMessage(), e); + manifestCache.recordFailure(statsQueryId); + // The SDK retry must own a new iterable because a lazy cache failure may leave the first one partial. + return buildManifestCountRange(table, scan.planFiles(), exactCount, formatVersion, partitioned, + orderedPartitionKeys, zone, uriNormalizer, netPositionDeletes); + } + } + + private Optional> buildManifestCountRange(Table table, + CloseableIterable tasks, long exactCount, int formatVersion, boolean partitioned, + List orderedPartitionKeys, ZoneId zone, UnaryOperator uriNormalizer, + boolean netPositionDeletes) { + try (CloseableIterable closeableTasks = tasks) { + for (FileScanTask task : closeableTasks) { + // Data-manifest rows and the separately netted live delete files are authoritative, but retain + // this defensive check for malformed metadata before exposing the aggregate as a query result. + if (hasNonIgnorableTaskDeletes(task, netPositionDeletes) || task.file().recordCount() < 0) { return Optional.empty(); } return Optional.of(Collections.singletonList(buildRange(table, task.file(), task, formatVersion, @@ -1307,16 +1346,17 @@ private Optional> planManifestCountRange(Table table, T private Optional> planCountPushdownFromFileTasks(Table table, TableScan scan, int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, UnaryOperator uriNormalizer, ConnectorSession session, Optional filter, - boolean ignorePositionDeletes) { + boolean netPositionDeletes, long positionDeleteRows) { if (!isManifestCacheEnabled()) { return accumulateCountPushdownFileTasks(table, scan.planFiles(), formatVersion, partitioned, - orderedPartitionKeys, zone, uriNormalizer, ignorePositionDeletes); + orderedPartitionKeys, zone, uriNormalizer, netPositionDeletes, positionDeleteRows); } String statsQueryId = session != null ? session.getQueryId() : null; try { return accumulateCountPushdownFileTasks(table, cacheBackedFileScanTasks(scan, session, table, filter, statsQueryId), formatVersion, - partitioned, orderedPartitionKeys, zone, uriNormalizer, ignorePositionDeletes); + partitioned, orderedPartitionKeys, zone, uriNormalizer, netPositionDeletes, + positionDeleteRows); } catch (Exception e) { LOG.warn("Iceberg count-pushdown plan with manifest cache failed, falling back to SDK scan: {}", e.getMessage(), e); @@ -1324,20 +1364,20 @@ private Optional> planCountPushdownFromFileTasks(Table // The retry owns a fresh accumulator so rows consumed before a lazy cache failure are never counted // twice and the SDK fallback remains an exact restart. return accumulateCountPushdownFileTasks(table, scan.planFiles(), formatVersion, partitioned, - orderedPartitionKeys, zone, uriNormalizer, ignorePositionDeletes); + orderedPartitionKeys, zone, uriNormalizer, netPositionDeletes, positionDeleteRows); } } private Optional> accumulateCountPushdownFileTasks(Table table, CloseableIterable tasks, int formatVersion, boolean partitioned, List orderedPartitionKeys, ZoneId zone, UnaryOperator uriNormalizer, - boolean ignorePositionDeletes) { + boolean netPositionDeletes, long positionDeleteRows) { FileScanTask representative = null; long exactCount = 0; try (CloseableIterable closeableTasks = tasks) { for (FileScanTask task : closeableTasks) { // A metadata count is safe only when manifests alone prove the exact visible row count. - if (hasNonIgnorableTaskDeletes(task, ignorePositionDeletes) || task.file().recordCount() < 0) { + if (hasNonIgnorableTaskDeletes(task, netPositionDeletes) || task.file().recordCount() < 0) { return Optional.empty(); } try { @@ -1353,21 +1393,27 @@ private Optional> accumulateCountPushdownFileTasks(Tabl throw new RuntimeException("Failed to plan iceberg count-pushdown file, error message is:" + e.getMessage(), e); } + OptionalLong visibleRows = subtractPositionDeleteRows(exactCount, positionDeleteRows); + if (!visibleRows.isPresent()) { + return Optional.empty(); + } if (representative == null) { - return Optional.of(Collections.emptyList()); + return visibleRows.getAsLong() == 0 + ? Optional.of(Collections.emptyList()) : Optional.empty(); } // targetSplitSize = -1: the count-pushdown collapse emits a single range, so its scheduling weight is // irrelevant and PluginDrivenSplit keeps SplitWeight.standard(). return Optional.of(Collections.singletonList(buildRange(table, representative.file(), representative, - formatVersion, partitioned, orderedPartitionKeys, zone, uriNormalizer, exactCount, -1, null))); + formatVersion, partitioned, orderedPartitionKeys, zone, uriNormalizer, + visibleRows.getAsLong(), -1, null))); } - private static boolean hasNonIgnorableTaskDeletes(FileScanTask task, boolean ignorePositionDeletes) { + private static boolean hasNonIgnorableTaskDeletes(FileScanTask task, boolean netPositionDeletes) { if (task.deletes() == null) { return false; } for (DeleteFile delete : task.deletes()) { - if (!ignorePositionDeletes || delete.content() != FileContent.POSITION_DELETES) { + if (!netPositionDeletes || delete.content() != FileContent.POSITION_DELETES) { return true; } } @@ -1395,14 +1441,19 @@ private static ManifestDeleteState manifestDeleteState(List manife return ManifestDeleteState.NONE; } - private static boolean hasNonIgnorableDeleteFiles(Table table, Snapshot snapshot, - boolean ignorePositionDeletes) { + private static OptionalLong livePositionDeleteRowCount(Table table, Snapshot snapshot) { + long exactCount = 0; for (ManifestFile manifest : snapshot.deleteManifests(table.io())) { try (ManifestReader reader = ManifestFiles.readDeleteManifest( manifest, table.io(), table.specs())) { for (DeleteFile delete : reader) { - if (!ignorePositionDeletes || delete.content() != FileContent.POSITION_DELETES) { - return true; + if (delete.content() != FileContent.POSITION_DELETES || delete.recordCount() < 0) { + return OptionalLong.empty(); + } + try { + exactCount = Math.addExact(exactCount, delete.recordCount()); + } catch (ArithmeticException e) { + return OptionalLong.empty(); } } } catch (IOException e) { @@ -1410,7 +1461,16 @@ private static boolean hasNonIgnorableDeleteFiles(Table table, Snapshot snapshot "Failed to inspect iceberg delete manifest " + manifest.path() + ": " + e.getMessage(), e); } } - return false; + return OptionalLong.of(exactCount); + } + + private static OptionalLong subtractPositionDeleteRows(long dataRows, long positionDeleteRows) { + try { + long visibleRows = Math.subtractExact(dataRows, positionDeleteRows); + return visibleRows >= 0 ? OptionalLong.of(visibleRows) : OptionalLong.empty(); + } catch (ArithmeticException e) { + return OptionalLong.empty(); + } } private static OptionalLong liveRowCountFromManifests(List manifests) { @@ -1432,30 +1492,6 @@ private static OptionalLong liveRowCountFromManifests(List manifes return OptionalLong.of(exactCount); } - /** - * The COUNT(*)-pushdown representative/fallback enumeration. PERF-04 (C18): when the manifest cache is - * enabled, read through the lazy {@link #cacheBackedFileScanTasks} (stats overload — this runs on the single - * planning thread) so the fast path stops after one representative file and the old-metadata fallback remains - * bounded without materializing the table's task list in FE memory. - * An eager cache failure falls back to the SDK path (mirrors {@link #planFileScanTask}). The first surviving - * (pruned) file may differ from the SDK path's first file (its {@code ParallelIterable} order is - * non-deterministic), but BE ignores the representative file. Cache disabled -> the SDK path, byte-unchanged. - */ - private CloseableIterable countPushdownFileScanTasks(TableScan scan, ConnectorSession session, - Table table, Optional filter) { - if (isManifestCacheEnabled()) { - String statsQueryId = session != null ? session.getQueryId() : null; - try { - return cacheBackedFileScanTasks(scan, session, table, filter, statsQueryId); - } catch (Exception e) { - LOG.warn("Iceberg count-pushdown plan with manifest cache failed, falling back to SDK scan: {}", - e.getMessage(), e); - manifestCache.recordFailure(statsQueryId); - } - } - return scan.planFiles(); - } - /** * Per-file scratch for {@link #buildRange}: the values identical for every byte-slice of one data file * ({@code TableScanUtil.splitFiles} cuts a file into k slices whose {@code FileScanTask}s all return the 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 3634a26f503e65..788b9f8dae5517 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 @@ -154,6 +154,11 @@ private static Table tableWithSnapshotSummary(Table table, Map s (proxy, method, args) -> wrapSnapshotSummary(invoke(method, table, args), summaryOverrides)); } + private static Table tableWithIo(Table table, FileIO fileIO) { + return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(), new Class[] {Table.class}, + (proxy, method, args) -> method.getName().equals("io") ? fileIO : invoke(method, table, args)); + } + private static Table tableWithMissingManifestRowsAndIo(Table table, FileIO fileIO) { return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(), new Class[] {Table.class}, (proxy, method, args) -> { @@ -2724,7 +2729,7 @@ public void countPushdownNotAppliedWithEqualityDeletesScansAll() { } @Test - public void countPushdownWithPositionDeletesUsesDataRowsWhenIgnoringDangling() { + public void countPushdownWithPositionDeletesNetsDeleteRowsWhenIgnoringDangling() { Map v2 = Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"); Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned(), v2); // 1000/100 = 10 data records. @@ -2746,10 +2751,10 @@ public void countPushdownWithPositionDeletesUsesDataRowsWhenIgnoringDangling() { List ranges = planCount(provider, session, true); - // The compatibility flag explicitly ignores position deletes. Use current data-manifest rows rather than - // the optional snapshot summary, preserving the public contract without reintroducing the original bug. + // Preserve the compatibility contract without trusting the optional summary: current data-manifest rows + // minus current position-delete file rows. The flag may still be inaccurate for dangling delete entries. Assertions.assertEquals(1, ranges.size()); - Assertions.assertEquals(10L, ranges.get(0).getPushDownRowCount()); + Assertions.assertEquals(7L, ranges.get(0).getPushDownRowCount()); } @Test @@ -3007,6 +3012,32 @@ public void countPushdownManifestCacheReadsOnlyRepresentativeFile() { Assertions.assertEquals(1, cache.size()); } + @Test + public void countPushdownManifestAggregateCacheFailureRetriesSdk() { + Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned()); + table.newAppend() + .appendFile(dataFile(table.spec(), "s3://b/db/t1/f1.parquet", 1024, null, null)) + .appendFile(dataFile(table.spec(), "s3://b/db/t1/f2.parquet", 2048, null, null)) + .appendFile(dataFile(table.spec(), "s3://b/db/t1/f3.parquet", 3072, null, null)) + .commit(); + List manifests = table.currentSnapshot().dataManifests(table.io()); + Assertions.assertEquals(1, manifests.size()); + + FailOnceFileIO fileIO = new FailOnceFileIO(table.io(), manifests.get(0).path()); + Table wrapped = tableWithSnapshotSummary(tableWithIo(table, fileIO), + Collections.singletonMap("total-records", "not-a-number")); + IcebergManifestCache cache = new IcebergManifestCache(); + + // Usable manifest aggregates must not make the optional cache a query-availability dependency. + List ranges = planCount( + manifestProvider(manifestCacheProps(), wrapped, cache), emptySession(), true); + + Assertions.assertEquals(1, ranges.size()); + Assertions.assertEquals(60L, ranges.get(0).getPushDownRowCount()); + Assertions.assertTrue(fileIO.failed.get()); + Assertions.assertEquals(1L, cache.takeStats("q")[2]); + } + @Test public void countPushdownLateManifestCacheFailureRetriesSdk() { Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index 24ebc99e8341c4..8a6c7c40bf3a49 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -3565,9 +3565,9 @@ public void checkAnnIndexCandidateRowsPercentThreshold(String value) { public static final String IGNORE_ICEBERG_DANGLING_DELETE = "ignore_iceberg_dangling_delete"; @VarAttrDef.VarAttr(name = IGNORE_ICEBERG_DANGLING_DELETE, - description = "Whether Iceberg metadata COUNT(*) may ignore position delete files. " - + "This improves performance but can make the result inaccurate when live position deletes " - + "exist. Equality deletes always disable metadata COUNT(*).") + description = "Whether Iceberg metadata COUNT(*) may subtract position-delete record counts from " + + "current data-manifest rows. This improves performance but can be inaccurate for dangling " + + "delete entries. Equality deletes always disable metadata COUNT(*).") public boolean ignoreIcebergDanglingDelete = false; @VarAttrDef.VarAttr(name = ENABLE_ICEBERG_MERGE_PARTITIONING,