diff --git a/duckdb b/duckdb index ef7a5a8c3..c55206e8e 160000 --- a/duckdb +++ b/duckdb @@ -1 +1 @@ -Subproject commit ef7a5a8c3090204407d4227584ae32019ae43066 +Subproject commit c55206e8e21a863233c68a87c55ec04151be7651 diff --git a/scripts/data_generators/tests/partitioned_deletion_vectors/__init__.py b/scripts/data_generators/tests/partitioned_deletion_vectors/__init__.py new file mode 100644 index 000000000..b4dc6bcd4 --- /dev/null +++ b/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) diff --git a/scripts/data_generators/tests/partitioned_deletion_vectors/q00.sql b/scripts/data_generators/tests/partitioned_deletion_vectors/q00.sql new file mode 100644 index 000000000..eeab6237f --- /dev/null +++ b/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) diff --git a/scripts/data_generators/tests/partitioned_deletion_vectors/q01.sql b/scripts/data_generators/tests/partitioned_deletion_vectors/q01.sql new file mode 100644 index 000000000..b19e2f12d --- /dev/null +++ b/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 diff --git a/src/iceberg_functions/iceberg_scan.cpp b/src/iceberg_functions/iceberg_scan.cpp index 19b98f42e..8289bca94 100644 --- a/src/iceberg_functions/iceberg_scan.cpp +++ b/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 #include +#include namespace duckdb { +struct IcebergPartitionRowGroup : public PartitionRowGroup { + //! Schema columns for mapping column_index -> field_id + const vector> &schema; + //! Reference to the data file with bounds + const IcebergDataFile &data_file; + + IcebergPartitionRowGroup(const vector> &schema, + const IcebergDataFile &data_file) + : schema(schema), data_file(data_file) { + } + + unique_ptr 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()); + } + if (pred_stats.has_upper_bounds) { + StringStats::SetMax(stats, pred_stats.upper_bound.GetValueUnsafe()); + } + 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 bind_data) { return BindInfo(*file_list.table); } +vector IcebergGetPartitionStats(ClientContext &context, GetPartitionStatsInput &input) { + vector result; + auto &bind_data = input.bind_data->Cast(); + auto &file_list = bind_data.file_list->Cast(); + + // 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 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 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(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(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"); diff --git a/test/sql/local/iceberg_scans/filtering_on_partition_bounds.test b/test/sql/local/iceberg_scans/filtering_on_partition_bounds.test index 4ff2409f6..0c6d4cbf0 100644 --- a/test/sql/local/iceberg_scans/filtering_on_partition_bounds.test +++ b/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 :.*~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 :.*~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 :.*~2,000.* + # 5 snapshots that each add 1000 rows (incremental) query I select count(*) from my_datalake.default.filtering_on_partition_bounds; diff --git a/test/sql/local/iceberg_scans/iceberg_cardinality_estimates.test b/test/sql/local/iceberg_scans/iceberg_cardinality_estimates.test index 4985ce5b8..8cf1f947b 100644 --- a/test/sql/local/iceberg_scans/iceberg_cardinality_estimates.test +++ b/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 :.*~50,000.* + +query I +select count(*) from ICEBERG_SCAN('__WORKING_DIRECTORY__/data/generated/iceberg/spark-local/default/deletion_vectors'); +---- +50000 + diff --git a/test/sql/local/iceberg_scans/iceberg_cardinality_with_deletes.test b/test/sql/local/iceberg_scans/iceberg_cardinality_with_deletes.test new file mode 100644 index 000000000..e7a573544 --- /dev/null +++ b/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 :.*~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 :.*~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 :.*~1,000.*