This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a768fb25678 [fix](iceberg) Derive count pushdown from live manifests
(#66778)
a768fb25678 is described below
commit a768fb25678b652ee4cc182bd3deb8793223e492
Author: Gabriel <[email protected]>
AuthorDate: Mon Aug 17 09:55:18 2026 +0800
[fix](iceberg) Derive count pushdown from live manifests (#66778)
### What problem does this PR solve?
Issue Number: None
Problem Summary:
Iceberg V2 COUNT(*) pushdown treated the optional `total-records`
snapshot summary field as an exact query result. A writer can provide a
syntactically valid positive value that does not match the live data
files, causing Doris to return a wrong count without reading data.
This change derives an exact unfiltered count by summing `addedRowsCount
+ existingRowsCount` across the current manifest list. The common path
is O(manifests) and reads only the first live data-file task to build
the representative range. Older manifest lists that omit aggregate
counters fall back to bounded per-file `recordCount` enumeration.
Queries with row filters or live delete files use the normal scan, while
invalid counters and overflow safely fall back instead of exposing
unverified metadata.
### Release note
Fix incorrect Iceberg V2 COUNT(*) results when snapshot summary row
counts are inaccurate.
### Check List (For Author)
- Test: Unit Test
- `IcebergScanPlanProviderTest`: 148 tests passed
- FE Checkstyle: passed with 0 violations
- Behavior changed: Yes. Exact unfiltered COUNT(*) pushdown now uses
current manifest-list aggregates and safely falls back when exactness
cannot be proven.
- Does this need documentation: No
---
be/src/format/table/iceberg_reader_mixin.h | 9 +-
.../iceberg/IcebergConnectorMetadata.java | 15 +-
.../connector/iceberg/IcebergScanPlanProvider.java | 377 ++++++++++++++-------
.../iceberg/IcebergCountFromSummaryTest.java | 114 -------
.../iceberg/IcebergScanPlanProviderTest.java | 284 ++++++++++++++--
.../java/org/apache/doris/qe/SessionVariable.java | 10 +-
6 files changed, 527 insertions(+), 282 deletions(-)
diff --git a/be/src/format/table/iceberg_reader_mixin.h
b/be/src/format/table/iceberg_reader_mixin.h
index 55064c6687d..437b76d0e2d 100644
--- a/be/src/format/table/iceberg_reader_mixin.h
+++ b/be/src/format/table/iceberg_reader_mixin.h
@@ -534,11 +534,10 @@ protected:
template <typename BaseReader>
Status IcebergReaderMixin<BaseReader>::_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/IcebergConnectorMetadata.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
index 2546fe4690e..c6c327955ee 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 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.
private static final String TOTAL_RECORDS = "total-records";
@@ -823,9 +822,8 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
* .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 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) {
@@ -842,8 +840,7 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
// 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 fd1eab305db..bd2e90bd646 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,15 +150,11 @@ 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;
- // 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
@@ -457,10 +453,23 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
if (snapshot == null) {
return -1;
}
- if (getFormatVersion(table) >= 3) {
- return -1;
+ if (countPushdown && filter.isEmpty()) {
+ 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) {
+ return -1;
+ }
+ if (deleteState != ManifestDeleteState.PRESENT ||
netPositionDeletes) {
+ OptionalLong positionDeleteRows =
livePositionDeleteRowCount(table, snapshot);
+ if (positionDeleteRows.isPresent()
+ && (netPositionDeletes ||
positionDeleteRows.getAsLong() == 0)) {
+ return -1;
+ }
+ }
}
- if (countPushdown && getCountFromSnapshot(scan, session) >= 0) {
+ if (getFormatVersion(table) >= 3) {
return -1;
}
long threshold = sessionLong(session, NUM_FILES_IN_BATCH_MODE,
DEFAULT_NUM_FILES_IN_BATCH_MODE);
@@ -720,16 +729,17 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// per-file path normalization below, instead of rebuilding it per
data/delete file (C3).
UnaryOperator<String> 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 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()) {
+ Optional<List<ConnectorScanRange>> countRanges =
planCountPushdown(table, scan, formatVersion,
+ partitioned, orderedPartitionKeys, zone, uriNormalizer,
session, filter);
+ if (countRanges.isPresent()) {
+ return countRanges.get();
}
}
@@ -1167,7 +1177,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* (profile) and {@code planWith(threadPool)} are intentionally dropped —
the iceberg SDK default worker
* pool plans, and the file set is identical (see design deviations). The
MVCC / time-travel pin (T07) is
* applied here ({@code useRef} for a tag/branch, else {@code
useSnapshot}), mirroring legacy
- * {@code createTableScan}; {@code getCountFromSnapshot} reads {@code
scan.snapshot()} so the count follows.
+ * {@code createTableScan}; COUNT planning reads this pinned scan's
manifest list, so the count follows.
*/
private TableScan buildScan(Table table, IcebergTableHandle handle,
Optional<ConnectorExpression> filter,
ConnectorSession session) {
@@ -1254,58 +1264,245 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
}
/**
- * 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 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-netted position deletes make the optimization
unsafe and tell the caller to
+ * perform a normal scan.
*/
- private List<ConnectorScanRange> planCountPushdown(Table table, TableScan
scan, long realCount,
+ private Optional<List<ConnectorScanRange>> planCountPushdown(Table table,
TableScan scan,
int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer, ConnectorSession session,
Optional<ConnectorExpression> filter) {
- try (CloseableIterable<FileScanTask> 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));
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null) {
+ return Optional.of(Collections.emptyList());
+ }
+
+ boolean netPositionDeletes = sessionBool(session,
IGNORE_ICEBERG_DANGLING_DELETE, false);
+ ManifestDeleteState deleteState =
manifestDeleteState(snapshot.deleteManifests(table.io()));
+ if (deleteState == ManifestDeleteState.PRESENT && !netPositionDeletes)
{
+ return Optional.empty();
+ }
+ 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,
netPositionDeletes,
+ netPositionDeletes ? positionDeleteRows.getAsLong() : 0);
+ }
+
+ private Optional<List<ConnectorScanRange>> planManifestCountRange(Table
table, TableScan scan, long exactCount,
+ int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
+ UnaryOperator<String> uriNormalizer, ConnectorSession session,
Optional<ConnectorExpression> filter,
+ 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<List<ConnectorScanRange>> buildManifestCountRange(Table
table,
+ CloseableIterable<FileScanTask> tasks, long exactCount, int
formatVersion, boolean partitioned,
+ List<String> orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer,
+ boolean netPositionDeletes) {
+ try (CloseableIterable<FileScanTask> 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,
+ 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 Collections.emptyList();
+ return exactCount == 0 ? Optional.of(Collections.emptyList()) :
Optional.empty();
}
- /**
- * 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.
- * 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.
- */
- private CloseableIterable<FileScanTask>
countPushdownFileScanTasks(TableScan scan, ConnectorSession session,
- Table table, Optional<ConnectorExpression> filter) {
- if (isManifestCacheEnabled()) {
+ private Optional<List<ConnectorScanRange>>
planCountPushdownFromFileTasks(Table table, TableScan scan,
+ int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
+ UnaryOperator<String> uriNormalizer, ConnectorSession session,
Optional<ConnectorExpression> filter,
+ boolean netPositionDeletes, long positionDeleteRows) {
+ if (!isManifestCacheEnabled()) {
+ return accumulateCountPushdownFileTasks(table, scan.planFiles(),
formatVersion, partitioned,
+ 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,
netPositionDeletes,
+ positionDeleteRows);
+ } 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,
netPositionDeletes, positionDeleteRows);
+ }
+ }
+
+ private Optional<List<ConnectorScanRange>>
accumulateCountPushdownFileTasks(Table table,
+ CloseableIterable<FileScanTask> tasks, int formatVersion, boolean
partitioned,
+ List<String> orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer,
+ boolean netPositionDeletes, long positionDeleteRows) {
+ FileScanTask representative = null;
+ long exactCount = 0;
+ try (CloseableIterable<FileScanTask> closeableTasks = tasks) {
+ for (FileScanTask task : closeableTasks) {
+ // A metadata count is safe only when manifests alone prove
the exact visible row count.
+ if (hasNonIgnorableTaskDeletes(task, netPositionDeletes) ||
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);
+ }
+ OptionalLong visibleRows = subtractPositionDeleteRows(exactCount,
positionDeleteRows);
+ if (!visibleRows.isPresent()) {
+ return Optional.empty();
+ }
+ if (representative == null) {
+ 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,
+ visibleRows.getAsLong(), -1, null)));
+ }
+
+ private static boolean hasNonIgnorableTaskDeletes(FileScanTask task,
boolean netPositionDeletes) {
+ if (task.deletes() == null) {
+ return false;
+ }
+ for (DeleteFile delete : task.deletes()) {
+ if (!netPositionDeletes || delete.content() !=
FileContent.POSITION_DELETES) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private enum ManifestDeleteState {
+ NONE,
+ PRESENT,
+ UNKNOWN
+ }
+
+ private static ManifestDeleteState manifestDeleteState(List<ManifestFile>
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 livePositionDeleteRowCount(Table table,
Snapshot snapshot) {
+ long exactCount = 0;
+ for (ManifestFile manifest : snapshot.deleteManifests(table.io())) {
+ try (ManifestReader<DeleteFile> reader =
ManifestFiles.readDeleteManifest(
+ manifest, table.io(), table.specs())) {
+ for (DeleteFile delete : reader) {
+ 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) {
+ throw new DorisConnectorException(
+ "Failed to inspect iceberg delete manifest " +
manifest.path() + ": " + e.getMessage(), e);
+ }
+ }
+ 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<ManifestFile>
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 {
- return cacheBackedFileScanTasks(scan, session, table, filter,
session.getQueryId());
- } 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());
+ exactCount = Math.addExact(exactCount, addedRows);
+ exactCount = Math.addExact(exactCount, existingRows);
+ } catch (ArithmeticException e) {
+ return OptionalLong.empty();
}
}
- return scan.planFiles();
+ return OptionalLong.of(exactCount);
}
/**
@@ -2771,68 +2968,6 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
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:
- * <ul>
- * <li>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)};</li>
- * <li>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.</li>
- * </ul>
- * 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<String, String> 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 a1e67d79019..00000000000
---
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.
- *
- * <p>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<String, String> summary(String equalityDeletes, String
positionDeletes,
- String totalRecords) {
- Map<String, String> 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 c5ce41fde94..73c0abd20a9 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,13 +51,16 @@ import org.apache.iceberg.DeleteFile;
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;
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 +79,8 @@ import org.junit.jupiter.api.Test;
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;
@@ -89,6 +94,7 @@ import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.UnaryOperator;
/**
@@ -143,6 +149,87 @@ public class IcebergScanPlanProviderTest {
return builder.build();
}
+ private static Table tableWithSnapshotSummary(Table table, Map<String,
String> summaryOverrides) {
+ return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(),
new Class<?>[] {Table.class},
+ (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) -> {
+ 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<ManifestFile> 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<String,
String> 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<String, String> 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();
@@ -2131,14 +2218,38 @@ public class IcebergScanPlanProviderTest {
@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
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.
+ 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
@@ -2548,7 +2659,7 @@ public class IcebergScanPlanProviderTest {
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<ConnectorScanRange> planCount(IcebergScanPlanProvider
provider, ConnectorSession session,
boolean countPushdown) {
@@ -2587,6 +2698,27 @@ public class IcebergScanPlanProviderTest {
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<ConnectorScanRange> 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<String, String> v2 =
Collections.singletonMap(TableProperties.FORMAT_VERSION, "2");
@@ -2601,10 +2733,12 @@ public class IcebergScanPlanProviderTest {
.commit();
IcebergScanPlanProvider provider = new
IcebergScanPlanProvider(IcebergCatalogProperties.of(Collections.emptyMap()),
opsReturning(table));
- List<ConnectorScanRange> ranges = planCount(provider, null, true);
+ ConnectorSession session = new FakeScanSession("UTC",
+ Collections.singletonMap("ignore_iceberg_dangling_delete",
"true"));
+ List<ConnectorScanRange> ranges = planCount(provider, session, 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.
+ // 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());
@@ -2612,7 +2746,7 @@ public class IcebergScanPlanProviderTest {
}
@Test
- public void countPushdownWithPositionDeletesNetsOutWhenIgnoringDangling() {
+ public void
countPushdownWithPositionDeletesNetsDeleteRowsWhenIgnoringDangling() {
Map<String, String> v2 =
Collections.singletonMap(TableProperties.FORMAT_VERSION, "2");
Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned(),
v2);
// 1000/100 = 10 data records.
@@ -2634,11 +2768,10 @@ public class IcebergScanPlanProviderTest {
List<ConnectorScanRange> 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.
+ // 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(7L, ranges.get(0).getPushDownRowCount());
- Assertions.assertEquals(7L,
populate(ranges.get(0)).getTableFormatParams().getTableLevelRowCount());
}
@Test
@@ -2671,9 +2804,7 @@ public class IcebergScanPlanProviderTest {
@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));
@@ -2701,6 +2832,28 @@ public class IcebergScanPlanProviderTest {
}
}
+ @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<ConnectorScanRange> 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<String, String> manifestCacheProps() {
@@ -2779,9 +2932,6 @@ public class IcebergScanPlanProviderTest {
}
// --- 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 {
@@ -2853,8 +3003,8 @@ public class IcebergScanPlanProviderTest {
}
@Test
- public void countPushdownManifestCacheMatchesCountAndReadsLazily() {
- // Three appends -> three data manifests; record counts 10+20+30 =
total-records 60 (snapshot summary).
+ 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();
@@ -2868,23 +3018,70 @@ public class IcebergScanPlanProviderTest {
List<ConnectorScanRange> 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-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());
- // 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");
+ // 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
+ 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<ManifestFile> 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<ConnectorScanRange> 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());
+ 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<ManifestFile> 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<ConnectorScanRange> 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;
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<ConnectorScanRange> cached =
planCount(manifestProvider(manifestCacheProps(), table, cache),
@@ -3907,6 +4104,41 @@ public class IcebergScanPlanProviderTest {
}
}
+ /** 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<String, String> 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<String, String> 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 a1592de2947..8a6c7c40bf3 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 class SessionVariable implements Serializable,
Writable {
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 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,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
