Partition-aware cardinality estimation for Iceberg

scripts/data_generators/tests/partitioned_deletion_vectors/__init__.py

@@ -0,0 +1,9 @@
+from scripts.data_generators.tests.base import IcebergTest
+import pathlib
+
+
+@IcebergTest.register()
+class Test(IcebergTest):
+    def __init__(self):
+        path = pathlib.PurePath(__file__)
+        super().__init__(path.parent.name)

scripts/data_generators/tests/partitioned_deletion_vectors/q00.sql

@@ -0,0 +1,14 @@
+CREATE or REPLACE TABLE default.partitioned_deletion_vectors (
+    seq integer,
+    col integer
+)
+USING iceberg
+PARTITIONED BY (seq)
+TBLPROPERTIES (
+    'format-version' = '3',
+    'write.delete.mode' = 'merge-on-read',
+    'write.delete.format' = 'puffin'
+);
+INSERT INTO default.partitioned_deletion_vectors SELECT 1 as seq, id as col FROM range(0, 1000);
+INSERT INTO default.partitioned_deletion_vectors SELECT 2 as seq, id as col FROM range(0, 1000);
+INSERT INTO default.partitioned_deletion_vectors SELECT 3 as seq, id as col FROM range(0, 1000)

scripts/data_generators/tests/partitioned_deletion_vectors/q01.sql

@@ -0,0 +1 @@
+DELETE FROM default.partitioned_deletion_vectors WHERE seq = 1 AND col % 2 = 0

src/iceberg_functions/iceberg_scan.cpp

@@ -26,11 +26,95 @@
 #include "iceberg_functions.hpp"
 #include "storage/catalog/iceberg_table_entry.hpp"
 
+#include "duckdb/common/multi_file/multi_file_states.hpp"
+#include "duckdb/function/partition_stats.hpp"
+#include "duckdb/storage/statistics/numeric_stats.hpp"
+#include "duckdb/storage/statistics/string_stats.hpp"
+#include "metadata/iceberg_column_definition.hpp"
+#include "metadata/iceberg_predicate_stats.hpp"
+#include "iceberg_value.hpp"
+
 #include <string>
 #include <numeric>
+#include <unordered_map>
 
 namespace duckdb {
 
+struct IcebergPartitionRowGroup : public PartitionRowGroup {
+	//! Schema columns for mapping column_index -> field_id
+	const vector<unique_ptr<IcebergColumnDefinition>> &schema;
+	//! Reference to the data file with bounds
+	const IcebergDataFile &data_file;
+
+	IcebergPartitionRowGroup(const vector<unique_ptr<IcebergColumnDefinition>> &schema,
+	                         const IcebergDataFile &data_file)
+	    : schema(schema), data_file(data_file) {
+	}
+
+	unique_ptr<BaseStatistics> GetColumnStatistics(const StorageIndex &storage_index) override {
+		auto col_idx = storage_index.GetPrimaryIndex();
+		if (col_idx >= schema.size()) {
+			return nullptr;
+		}
+		auto &column = *schema[col_idx];
+		auto field_id = column.id;
+
+		auto lower_it = data_file.lower_bounds.find(field_id);
+		auto upper_it = data_file.upper_bounds.find(field_id);
+		if (lower_it == data_file.lower_bounds.end() && upper_it == data_file.upper_bounds.end()) {
+			return nullptr;
+		}
+
+		Value lower_bound, upper_bound;
+		if (lower_it != data_file.lower_bounds.end()) {
+			lower_bound = lower_it->second;
+		}
+		if (upper_it != data_file.upper_bounds.end()) {
+			upper_bound = upper_it->second;
+		}
+
+		// DeserializeBounds converts from BLOB to typed values
+		IcebergPredicateStats pred_stats;
+		try {
+			pred_stats = IcebergPredicateStats::DeserializeBounds(lower_bound, upper_bound, column.name, column.type);
+		} catch (...) {
+			return nullptr;
+		}
+
+		if (!pred_stats.has_lower_bounds && !pred_stats.has_upper_bounds) {
+			return nullptr;
+		}
+
+		auto &type = column.type;
+		if (type.IsNumeric() || type.IsTemporal()) {
+			auto stats = NumericStats::CreateEmpty(type);
+			if (pred_stats.has_lower_bounds) {
+				NumericStats::SetMin(stats, pred_stats.lower_bound);
+			}
+			if (pred_stats.has_upper_bounds) {
+				NumericStats::SetMax(stats, pred_stats.upper_bound);
+			}
+			return stats.ToUnique();
+		} else if (type.id() == LogicalTypeId::VARCHAR) {
+			auto stats = StringStats::CreateEmpty(type);
+			if (pred_stats.has_lower_bounds) {
+				StringStats::SetMin(stats, pred_stats.lower_bound.GetValueUnsafe<string_t>());
+			}
+			if (pred_stats.has_upper_bounds) {
+				StringStats::SetMax(stats, pred_stats.upper_bound.GetValueUnsafe<string_t>());
+			}
+			return stats.ToUnique();
+		}
+		return nullptr;
+	}
+
+	bool MinMaxIsExact(const BaseStatistics &stats, const StorageIndex &storage_index) override {
+		// File-level bounds are not exact (they are min/max across the file, not single values)
+		// This is fine for partition pruning but not for eager MIN/MAX aggregates
+		return false;
+	}
+};
+
 static void AddNamedParameters(TableFunction &fun) {
 	fun.named_parameters["allow_moved_paths"] = LogicalType::BOOLEAN;
 	fun.named_parameters["mode"] = LogicalType::VARCHAR;
@@ -62,6 +146,73 @@ BindInfo IcebergBindInfo(const optional_ptr<FunctionData> bind_data) {
 	return BindInfo(*file_list.table);
 }
 
+vector<PartitionStatistics> IcebergGetPartitionStats(ClientContext &context, GetPartitionStatsInput &input) {
+	vector<PartitionStatistics> result;
+	auto &bind_data = input.bind_data->Cast<MultiFileBindData>();
+	auto &file_list = bind_data.file_list->Cast<IcebergMultiFileList>();
+
+	// V1 tables lack reliable per-file row counts
+	if (file_list.GetMetadata().iceberg_version == 1) {
+		return result;
+	}
+
+	// Force all files to be loaded
+	(void)file_list.GetTotalFileCount();
+
+	// Build per-data-file delete counts from delete manifest entries.
+	// For positional deletes with a referenced_data_file (deletion vectors and optimized V2
+	// positional deletes), we know exactly which data file is targeted and how many rows are
+	// deleted, so we can compute exact net counts from manifest metadata alone.
+	std::unordered_map<string, int64_t> deletes_per_file;
+	bool has_inexact_deletes = false;
+
+	if (!file_list.delete_manifests.empty()) {
+		auto iceberg_path = file_list.GetPath();
+		auto &fs = FileSystem::GetFileSystem(context);
+		auto &options = file_list.options;
+		auto &metadata = file_list.GetMetadata();
+		auto &snapshot = *file_list.GetSnapshot();
+
+		// Re-scan delete manifests independently (don't consume the shared delete_manifest_reader)
+		auto delete_scan =
+		    AvroScan::ScanManifest(snapshot, file_list.delete_manifests, options, fs, iceberg_path, metadata, context);
+		manifest_file::ManifestReader reader(*delete_scan, true);
+
+		vector<IcebergManifestEntry> entries;
+		while (!reader.Finished()) {
+			entries.clear();
+			reader.Read(STANDARD_VECTOR_SIZE, entries);
+			for (auto &e : entries) {
+				if (e.data_file.content == IcebergManifestEntryContentType::POSITION_DELETES &&
+				    !e.data_file.referenced_data_file.empty()) {
+					deletes_per_file[e.data_file.referenced_data_file] += e.data_file.record_count;
+				} else {
+					// Equality deletes or V2 positional without a specific target — genuinely approximate
+					has_inexact_deletes = true;
+				}
+			}
+		}
+	}
+
+	auto &schema = file_list.GetSchema().columns;
+
+	for (auto &entry : file_list.manifest_entries) {
+		PartitionStatistics stats;
+		int64_t deleted = 0;
+		auto it = deletes_per_file.find(entry.data_file.file_path);
+		if (it != deletes_per_file.end()) {
+			deleted = it->second;
+		}
+		stats.count = NumericCast<idx_t>(entry.data_file.record_count - deleted);
+		stats.count_type = has_inexact_deletes ? CountType::COUNT_APPROXIMATE : CountType::COUNT_EXACT;
+		if (!entry.data_file.lower_bounds.empty() || !entry.data_file.upper_bounds.empty()) {
+			stats.partition_row_group = make_shared_ptr<IcebergPartitionRowGroup>(schema, entry.data_file);
+		}
+		result.push_back(std::move(stats));
+	}
+	return result;
+}
+
 TableFunctionSet IcebergFunctions::GetIcebergScanFunction(ExtensionLoader &loader) {
 	// The iceberg_scan function is constructed by grabbing the parquet scan from the Catalog, then injecting the
 	// IcebergMultiFileReader into it to create a Iceberg-based multi file read
@@ -83,6 +234,7 @@ TableFunctionSet IcebergFunctions::GetIcebergScanFunction(ExtensionLoader &loade
 		function.table_scan_progress = nullptr;
 		function.get_bind_info = IcebergBindInfo;
 		function.get_virtual_columns = IcebergVirtualColumns;
+		function.get_partition_stats = IcebergGetPartitionStats;
 
 		// Schema param is just confusing here
 		function.named_parameters.erase("schema");

test/sql/local/iceberg_scans/filtering_on_partition_bounds.test

@@ -21,6 +21,26 @@ create view my_datalake.default.filtering_on_partition_bounds as select * from I
 statement ok
 CALL enable_logging('Iceberg');
 
+# 5 partitions (seq=1..5), each with 1000 rows, no deletions.
+# EXPLAIN cardinality for the full scan reflects the total across all partitions.
+query II
+explain select * from my_datalake.default.filtering_on_partition_bounds;
+----
+physical_plan	<REGEX>:.*~5,000.*
+
+# Partition-pruned scan: only seq=1 survives → cardinality drops to ~1,000.
+# This demonstrates that partition pruning correctly reduces the estimated row count.
+query II
+explain select * from my_datalake.default.filtering_on_partition_bounds where seq = 1;
+----
+physical_plan	<REGEX>:.*~1,000.*
+
+# Two partitions survive (seq=2, seq=3) → ~2,000.
+query II
+explain select * from my_datalake.default.filtering_on_partition_bounds where seq > 1 and seq < 4;
+----
+physical_plan	<REGEX>:.*~2,000.*
+
 # 5 snapshots that each add 1000 rows (incremental)
 query I
 select count(*) from my_datalake.default.filtering_on_partition_bounds;

test/sql/local/iceberg_scans/iceberg_cardinality_estimates.test

@@ -67,3 +67,34 @@ select count(*) from ICEBERG_SCAN('__WORKING_DIRECTORY__/data/generated/iceberg/
 ----
 40000
 
+
+# ── deletion_vectors: V3 puffin deletion vectors ─────────────────────────────
+# 100,000 rows inserted, every even col value deleted (50,000 rows deleted).
+# Net: 50,000 rows.  Because referenced_data_file is always set for puffin DVs,
+# PartitionStatistics reports COUNT_EXACT with the net per-file counts.
+
+# 1 data file
+query I
+select count(*) from ICEBERG_METADATA('__WORKING_DIRECTORY__/data/generated/iceberg/spark-local/default/deletion_vectors')
+where manifest_content = 'DATA';
+----
+1
+
+# 1 delete file (puffin DV)
+query I
+select count(*) from ICEBERG_METADATA('__WORKING_DIRECTORY__/data/generated/iceberg/spark-local/default/deletion_vectors')
+where manifest_content = 'DELETE';
+----
+1
+
+# EXPLAIN shows the net count ~50,000 (gross 100,000 - 50,000 DV deletes)
+query II
+explain select count(*) from ICEBERG_SCAN('__WORKING_DIRECTORY__/data/generated/iceberg/spark-local/default/deletion_vectors');
+----
+physical_plan	<REGEX>:.*~50,000.*
+
+query I
+select count(*) from ICEBERG_SCAN('__WORKING_DIRECTORY__/data/generated/iceberg/spark-local/default/deletion_vectors');
+----
+50000
+

test/sql/local/iceberg_scans/iceberg_cardinality_with_deletes.test

@@ -0,0 +1,96 @@
+# name: test/sql/local/iceberg_scans/iceberg_cardinality_with_deletes.test
+# description: Test cardinality estimation when partition pruning and deletions interact.
+# group: [iceberg_scans]
+
+#
+#   partitioned_deletion_vectors:
+#     - 3 identity-partitioned partitions (seq=1, seq=2, seq=3), 1000 rows each (3000 gross)
+#     - seq=1 has 500 rows deleted via V3 puffin deletion vectors (col%2=0)
+#     - seq=2 and seq=3 have no deletions
+#
+#   Without this PR's fix:
+#     Any table with delete files fell back to COUNT_APPROXIMATE with the *gross*
+#     record_count per file.  A filter that prunes seq=2 and seq=3 would still
+#     produce EXPLAIN ~1,000 for seq=1 (gross), not the correct ~500.
+#
+#   With this PR's fix:
+#     Positional-delete files that carry referenced_data_file (deletion vectors,
+#     and optimised V2 positional deletes) are resolved to exact per-file net
+#     counts.  Only equality deletes (or V2 positional without referenced_data_file)
+#     remain COUNT_APPROXIMATE.  So a filter on seq=1 now produces EXPLAIN ~500.
+#
+
+require-env DUCKDB_ICEBERG_HAVE_GENERATED_DATA
+
+require avro
+
+require parquet
+
+require iceberg
+
+statement ok
+attach ':memory:' as my_datalake;
+
+statement ok
+create schema my_datalake.default;
+
+statement ok
+create view my_datalake.default.partitioned_deletion_vectors as
+    select * from ICEBERG_SCAN('__WORKING_DIRECTORY__/data/generated/iceberg/spark-local/default/partitioned_deletion_vectors');
+
+# ── Verify the shape of the table ───────────────────────────────────────────
+
+# 3 data files (one per partition)
+query I
+select count(*) from ICEBERG_METADATA('__WORKING_DIRECTORY__/data/generated/iceberg/spark-local/default/partitioned_deletion_vectors')
+where manifest_content = 'DATA';
+----
+3
+
+# 1 delete file (the puffin DV for seq=1)
+query I
+select count(*) from ICEBERG_METADATA('__WORKING_DIRECTORY__/data/generated/iceberg/spark-local/default/partitioned_deletion_vectors')
+where manifest_content = 'DELETE';
+----
+1
+
+# ── Actual row counts (ground truth) ────────────────────────────────────────
+
+query I
+select count(*) from my_datalake.default.partitioned_deletion_vectors;
+----
+2500
+
+query I
+select count(*) from my_datalake.default.partitioned_deletion_vectors where seq = 1;
+----
+500
+
+query I
+select count(*) from my_datalake.default.partitioned_deletion_vectors where seq = 3;
+----
+1000
+
+# ── Full-scan EXPLAIN: GetCardinality subtracts DV rows at manifest level ───
+# Expected: ~2,500 (3000 gross - 500 DV deletes = 2500 net)
+
+query II
+explain select * from my_datalake.default.partitioned_deletion_vectors;
+----
+physical_plan	<REGEX>:.*~2,500.*
+
+# ── Partition-pruned EXPLAIN: only seq=1 files survive ──────────────────────
+# With deletion-vector-aware PartitionStatistics (this PR), each file contributes
+# its NET count (gross - DV deletes).  seq=1 has 1000 - 500 = 500 net rows.
+# Without this fix the scan would report the gross ~1,000 (COUNT_APPROXIMATE).
+
+query II
+explain select * from my_datalake.default.partitioned_deletion_vectors where seq = 1;
+----
+physical_plan	<REGEX>:.*~500.*
+
+# seq=3 has no deletions, so its net count equals the gross count.
+query II
+explain select * from my_datalake.default.partitioned_deletion_vectors where seq = 3;
+----
+physical_plan	<REGEX>:.*~1,000.*