From c0f050c4838a0ad541d00037b383ce415e7f6acd Mon Sep 17 00:00:00 2001 From: Niclas Haderer Date: Thu, 11 Jun 2026 12:05:14 +0200 Subject: [PATCH 1/7] Fix duplicate deletion vectors --- src/core/deletes/iceberg_deletion_vector.cpp | 6 +- src/core/deletes/iceberg_equality_delete.cpp | 6 +- .../deletes/iceberg_positional_delete.cpp | 4 +- src/execution/operator/iceberg_delete.cpp | 4 +- .../planning/iceberg_multi_file_list.hpp | 11 ++-- src/planning/iceberg_multi_file_list.cpp | 16 ++++-- .../test_delete_dv_merge_runtime_filter.test | 55 +++++++++++++++++++ 7 files changed, 83 insertions(+), 19 deletions(-) create mode 100644 test/sql/local/catalog_test_config_setup/catalog_agnostic/delete/test_delete_dv_merge_runtime_filter.test diff --git a/src/core/deletes/iceberg_deletion_vector.cpp b/src/core/deletes/iceberg_deletion_vector.cpp index dd105f192..cddf17dee 100644 --- a/src/core/deletes/iceberg_deletion_vector.cpp +++ b/src/core/deletes/iceberg_deletion_vector.cpp @@ -157,8 +157,8 @@ void IcebergMultiFileList::ScanPuffinFile(const BoundIcebergManifestEntry &bound auto buf_handle = caching_file_handle->Read(data, length, offset); auto buffer_data = buf_handle.Ptr(); - auto it = positional_delete_data.find(data_file.referenced_data_file); - if (it != positional_delete_data.end()) { + auto it = positional_delete_data->find(data_file.referenced_data_file); + if (it != positional_delete_data->end()) { //! Another delete already exists for this table auto &existing_delete = *it->second; if (existing_delete.type == IcebergDeleteType::DELETION_VECTOR) { @@ -167,7 +167,7 @@ void IcebergMultiFileList::ScanPuffinFile(const BoundIcebergManifestEntry &bound } } //! NOTE: assign, don't emplace, deletion vectors take priority over any remaining positional delete files - positional_delete_data[data_file.referenced_data_file] = + (*positional_delete_data)[data_file.referenced_data_file] = IcebergDeletionVectorData::FromBlob(bound_entry, buffer_data, length); } diff --git a/src/core/deletes/iceberg_equality_delete.cpp b/src/core/deletes/iceberg_equality_delete.cpp index 595acbf94..7f225f0e7 100644 --- a/src/core/deletes/iceberg_equality_delete.cpp +++ b/src/core/deletes/iceberg_equality_delete.cpp @@ -65,9 +65,9 @@ void IcebergMultiFileList::ScanEqualityDeleteFile(const BoundIcebergManifestEntr const auto sequence_number = manifest_entry.GetSequenceNumber(manifest_file); //! Get or create the equality delete data for this sequence number - auto it = equality_delete_data.find(sequence_number); - if (it == equality_delete_data.end()) { - it = equality_delete_data.emplace(sequence_number, make_uniq(sequence_number)).first; + auto it = equality_delete_data->find(sequence_number); + if (it == equality_delete_data->end()) { + it = equality_delete_data->emplace(sequence_number, make_uniq(sequence_number)).first; } auto &deletes = *it->second; diff --git a/src/core/deletes/iceberg_positional_delete.cpp b/src/core/deletes/iceberg_positional_delete.cpp index ada47b841..8819a1813 100644 --- a/src/core/deletes/iceberg_positional_delete.cpp +++ b/src/core/deletes/iceberg_positional_delete.cpp @@ -38,7 +38,7 @@ void IcebergMultiFileList::ScanPositionalDeleteFile(const BoundIcebergManifestEn } reference current_file_path = names[0]; auto initial_key = current_file_path.get().GetString(); - auto deletes = TryGetOrCreate(positional_delete_data, bound_entry, initial_key); + auto deletes = TryGetOrCreate(*positional_delete_data, bound_entry, initial_key); for (idx_t i = 0; i < count; i++) { auto &name = names[i]; @@ -47,7 +47,7 @@ void IcebergMultiFileList::ScanPositionalDeleteFile(const BoundIcebergManifestEn if (name != current_file_path.get()) { current_file_path = name; auto key = current_file_path.get().GetString(); - deletes = TryGetOrCreate(positional_delete_data, bound_entry, key); + deletes = TryGetOrCreate(*positional_delete_data, bound_entry, key); } if (!deletes) { continue; diff --git a/src/execution/operator/iceberg_delete.cpp b/src/execution/operator/iceberg_delete.cpp index d2b6f2ef8..c4e1fecd5 100644 --- a/src/execution/operator/iceberg_delete.cpp +++ b/src/execution/operator/iceberg_delete.cpp @@ -278,8 +278,8 @@ void IcebergDelete::FlushDeletes(IcebergTransaction &transaction, ClientContext } if (write_deletion_vector) { //! Addd the existing delete we're replacing - auto it = multi_file_list->positional_delete_data.find(filename); - if (it != multi_file_list->positional_delete_data.end()) { + auto it = multi_file_list->positional_delete_data->find(filename); + if (it != multi_file_list->positional_delete_data->end()) { auto &delete_data = *it->second; PopulateAlteredManifests(*multi_file_list, global_state.altered_manifests, delete_data); delete_data.ToSet(sorted_deletes); diff --git a/src/include/planning/iceberg_multi_file_list.hpp b/src/include/planning/iceberg_multi_file_list.hpp index 60bdef006..de7647aaa 100644 --- a/src/include/planning/iceberg_multi_file_list.hpp +++ b/src/include/planning/iceberg_multi_file_list.hpp @@ -139,10 +139,13 @@ struct IcebergMultiFileList : public MultiFileList { mutable ManifestEntryReadState read_state; - //! For each file that has a delete file, the state for processing that/those delete file(s) - mutable case_insensitive_map_t> positional_delete_data; - //! All equality deletes with sequence numbers higher than that of the data_file apply to that data_file - mutable map> equality_delete_data; + //! For each file that has a delete file, the state for processing that/those delete file(s). + //! Shared with the filter pushdown copies of this list (see PushdownInternal) so that delete state + //! observed while scanning is also visible to the list referenced by DML operators. + mutable shared_ptr>> positional_delete_data; + //! All equality deletes with sequence numbers higher than that of the data_file apply to that data_file. + //! Shared with the filter pushdown copies of this list for the same reason as positional_delete_data. + mutable shared_ptr>> equality_delete_data; //! FIXME: this is only used in 'FinalizeBind', //! shouldn't this be used to protect all the variable accesses that are accessed there while the lock is held? diff --git a/src/planning/iceberg_multi_file_list.cpp b/src/planning/iceberg_multi_file_list.cpp index 63ebfd440..73f4b6e26 100644 --- a/src/planning/iceberg_multi_file_list.cpp +++ b/src/planning/iceberg_multi_file_list.cpp @@ -115,7 +115,9 @@ class ManifestReadTask : public BaseExecutorTask { IcebergMultiFileList::IcebergMultiFileList(ClientContext &context_p, shared_ptr scan_info, const string &path, const IcebergOptions &options) : context(context_p), fs(FileSystem::GetFileSystem(context)), scan_info(scan_info), path(path), table(nullptr), - options(options) { + options(options), + positional_delete_data(make_shared_ptr>>()), + equality_delete_data(make_shared_ptr>>()) { } IcebergMultiFileList::~IcebergMultiFileList() { @@ -257,6 +259,10 @@ unique_ptr IcebergMultiFileList::PushdownInternal(ClientCo filtered_list->names = names; filtered_list->types = types; filtered_list->have_bound = true; + //! Share the delete state: DML operators hold a reference to the original list of the scan's bind data, + //! while the scan itself can read through a filtered copy created by (dynamic) filter pushdown. + filtered_list->positional_delete_data = positional_delete_data; + filtered_list->equality_delete_data = equality_delete_data; return filtered_list; } @@ -859,8 +865,8 @@ IcebergMultiFileList::GetEqualityDeletesForFile(const BoundIcebergManifestEntry auto &manifest_file = data_manifests[bound_manifest_entry.manifest_file_idx].entry.file; auto &data_file = manifest_entry.data_file; auto &metadata = GetMetadata(); - auto it = equality_delete_data.upper_bound(manifest_entry.GetSequenceNumber(manifest_file)); - for (; it != equality_delete_data.end(); it++) { + auto it = equality_delete_data->upper_bound(manifest_entry.GetSequenceNumber(manifest_file)); + for (; it != equality_delete_data->end(); it++) { auto &files = it->second->files; for (auto &file : files) { auto &partition_spec = metadata.partition_specs.at(file.partition_spec_id); @@ -1208,8 +1214,8 @@ void IcebergMultiFileList::ScanDeleteFile(const BoundIcebergManifestEntry &bound } unique_ptr IcebergMultiFileList::GetPositionalDeletesForFile(const string &file_path) const { - auto it = positional_delete_data.find(file_path); - if (it != positional_delete_data.end()) { + auto it = positional_delete_data->find(file_path); + if (it != positional_delete_data->end()) { // There is delete data for this file, return it return it->second->ToFilter(); } diff --git a/test/sql/local/catalog_test_config_setup/catalog_agnostic/delete/test_delete_dv_merge_runtime_filter.test b/test/sql/local/catalog_test_config_setup/catalog_agnostic/delete/test_delete_dv_merge_runtime_filter.test new file mode 100644 index 000000000..52c52a6cd --- /dev/null +++ b/test/sql/local/catalog_test_config_setup/catalog_agnostic/delete/test_delete_dv_merge_runtime_filter.test @@ -0,0 +1,55 @@ +# name: test/sql/local/catalog_test_config_setup/catalog_agnostic/delete/test_delete_dv_merge_runtime_filter.test +# description: DML whose scan receives a runtime (dynamic) filter must still merge existing deletion vectors +# group: [delete] + +require-env CATALOG_TEST_CONFIG_SETUP + +require avro + +require parquet + +require iceberg + +require httpfs + +# Do not ignore 'HTTP' error messages! +set ignore_error_messages + +statement ok +DROP TABLE IF EXISTS my_datalake.default.test_dv_merge_runtime_filter; + +statement ok +CREATE TABLE my_datalake.default.test_dv_merge_runtime_filter (id INT, val VARCHAR) WITH ('format-version'='3'); + +statement ok +INSERT INTO my_datalake.default.test_dv_merge_runtime_filter VALUES + (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'), (5, 'e'); + +# The first delete writes a deletion vector for the data file +statement ok +DELETE FROM my_datalake.default.test_dv_merge_runtime_filter WHERE val = 'a'; + +query II +SELECT * FROM my_datalake.default.test_dv_merge_runtime_filter ORDER BY id; +---- +2 b +3 c +4 d +5 e + +# The hash join generates a dynamic min/max filter that is pushed into the probe side scan +# at runtime. +statement ok +UPDATE my_datalake.default.test_dv_merge_runtime_filter SET val = val || '_updated' +FROM (VALUES (2), (3)) t(uid) WHERE id = t.uid; + +query II +SELECT * FROM my_datalake.default.test_dv_merge_runtime_filter ORDER BY id; +---- +2 b_updated +3 c_updated +4 d +5 e + +statement ok +DROP TABLE my_datalake.default.test_dv_merge_runtime_filter; From c05091aa7a5b15419ec530238bb6507822047480 Mon Sep 17 00:00:00 2001 From: Niclas Haderer Date: Thu, 11 Jun 2026 13:18:45 +0200 Subject: [PATCH 2/7] Add other test --- .../merge/merge_into_multiple.test | 80 +++++++++++++++++++ 1 file changed, 80 insertions(+) create mode 100644 test/sql/local/catalog_test_config_setup/catalog_agnostic/merge/merge_into_multiple.test diff --git a/test/sql/local/catalog_test_config_setup/catalog_agnostic/merge/merge_into_multiple.test b/test/sql/local/catalog_test_config_setup/catalog_agnostic/merge/merge_into_multiple.test new file mode 100644 index 000000000..c63922892 --- /dev/null +++ b/test/sql/local/catalog_test_config_setup/catalog_agnostic/merge/merge_into_multiple.test @@ -0,0 +1,80 @@ +# name: test/sql/local/catalog_test_config_setup/catalog_agnostic/merge/merge_into_multiple.test +# group: [merge] + +require-env CATALOG_TEST_CONFIG_SETUP + +require avro + +require parquet + +require iceberg + +require httpfs + +# Do not ignore 'HTTP' error messages! +set ignore_error_messages + +statement ok +set enable_logging=true + +statement ok +set logging_level='debug' + +statement ok +DROP TABLE IF EXISTS my_datalake.default.tbl; + +statement ok +CREATE TABLE my_datalake.default.tbl( + id INTEGER, + col BOOLEAN, + col2 VARCHAR +) WITH ( + 'format-version'='3' +); + +statement ok +insert into my_datalake.default.tbl values + (1, true, NULL), + (2, NULL, NULL), + (3, false, NULL); + +# explicit default +query I +MERGE INTO my_datalake.default.tbl AS t USING ( + VALUES + (1, true, 'test'), + (4, true, 'hello world') +) AS s(id, col, col2) ON t.id = s.id +WHEN MATCHED THEN UPDATE SET col = s.col, col2 = s.col2 +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.col, s.col2) +---- +2 + +query III +select * from my_datalake.default.tbl order by id +---- +1 true test +2 NULL NULL +3 false NULL +4 true hello world + +# column list +query I +MERGE INTO my_datalake.default.tbl AS t USING ( + VALUES + (2, false, 'amsterdam'), + (5, false, 'rotterdam') +) AS s(id, col, col2) ON t.id = s.id +WHEN MATCHED THEN UPDATE SET col = s.col, col2 = s.col2 +WHEN NOT MATCHED THEN INSERT (id, col, col2) VALUES (s.id, s.col, s.col2) +---- +2 + +query III +select * from my_datalake.default.tbl order by id +---- +1 true test +2 false amsterdam +3 false NULL +4 true hello world +5 false rotterdam From d1226c400d173f0578e4e13ba43db302e577474f Mon Sep 17 00:00:00 2001 From: Tishj Date: Thu, 11 Jun 2026 23:56:11 +0200 Subject: [PATCH 3/7] format --- src/core/deletes/iceberg_equality_delete.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/core/deletes/iceberg_equality_delete.cpp b/src/core/deletes/iceberg_equality_delete.cpp index 7f225f0e7..5cd1a3bba 100644 --- a/src/core/deletes/iceberg_equality_delete.cpp +++ b/src/core/deletes/iceberg_equality_delete.cpp @@ -67,7 +67,8 @@ void IcebergMultiFileList::ScanEqualityDeleteFile(const BoundIcebergManifestEntr //! Get or create the equality delete data for this sequence number auto it = equality_delete_data->find(sequence_number); if (it == equality_delete_data->end()) { - it = equality_delete_data->emplace(sequence_number, make_uniq(sequence_number)).first; + it = + equality_delete_data->emplace(sequence_number, make_uniq(sequence_number)).first; } auto &deletes = *it->second; From 1df962e4ea62b2875b2fcb7173ccd3cdae295bb4 Mon Sep 17 00:00:00 2001 From: Tishj Date: Fri, 12 Jun 2026 13:38:17 +0200 Subject: [PATCH 4/7] add unrelated missing order bys --- .../catalog_agnostic/merge/merge_partition.test | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/sql/local/catalog_test_config_setup/catalog_agnostic/merge/merge_partition.test b/test/sql/local/catalog_test_config_setup/catalog_agnostic/merge/merge_partition.test index a3cfaf3d2..bceebfaeb 100644 --- a/test/sql/local/catalog_test_config_setup/catalog_agnostic/merge/merge_partition.test +++ b/test/sql/local/catalog_test_config_setup/catalog_agnostic/merge/merge_partition.test @@ -47,7 +47,7 @@ MERGE INTO my_datalake.default.my_timeseries WHEN NOT MATCHED THEN INSERT; query II -SELECT * FROM my_datalake.default.my_timeseries +SELECT * FROM my_datalake.default.my_timeseries order by all ---- 2025-09-15 00:00:00 42.0 2025-09-17 00:00:00 42.0 @@ -63,7 +63,7 @@ ON my_datalake.default.my_timeseries.ts = timeseries_updates.ts WHEN MATCHED THEN UPDATE; query II -SELECT * FROM my_datalake.default.my_timeseries +SELECT * FROM my_datalake.default.my_timeseries order by all ---- 2025-09-15 00:00:00 42.0 2025-09-17 00:00:00 43.0 From f2a075cb3e90ef36600285abca7a5e58a0176dee Mon Sep 17 00:00:00 2001 From: Tishj Date: Sat, 13 Jun 2026 13:44:33 +0200 Subject: [PATCH 5/7] bump duckdb submodule --- .github/workflows/MainDistributionPipeline.yml | 4 ++-- duckdb | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/MainDistributionPipeline.yml b/.github/workflows/MainDistributionPipeline.yml index 4313337a3..ea3aab2c3 100644 --- a/.github/workflows/MainDistributionPipeline.yml +++ b/.github/workflows/MainDistributionPipeline.yml @@ -17,7 +17,7 @@ jobs: uses: duckdb/extension-ci-tools/.github/workflows/_extension_distribution.yml@v1.5-variegata with: extension_name: iceberg - duckdb_version: 4d8c29780d60962a9f237c4c734ef2d4b4905e85 + duckdb_version: c285de922ab954d7bd5f0269e41d8d52eb5560c0 ci_tools_version: v1.5-variegata exclude_archs: 'windows_amd64_mingw' extra_toolchains: 'python3' @@ -29,7 +29,7 @@ jobs: secrets: inherit with: extension_name: iceberg - duckdb_version: 4d8c29780d60962a9f237c4c734ef2d4b4905e85 + duckdb_version: c285de922ab954d7bd5f0269e41d8d52eb5560c0 ci_tools_version: v1.5-variegata exclude_archs: 'windows_amd64_mingw' deploy_latest: ${{ startsWith(github.ref, 'refs/heads/v') || github.ref == 'refs/heads/main' }} diff --git a/duckdb b/duckdb index 4d8c29780..c285de922 160000 --- a/duckdb +++ b/duckdb @@ -1 +1 @@ -Subproject commit 4d8c29780d60962a9f237c4c734ef2d4b4905e85 +Subproject commit c285de922ab954d7bd5f0269e41d8d52eb5560c0 From 1334b88c49c0a2d193eac0f21132f35dcf4c1625 Mon Sep 17 00:00:00 2001 From: Tishj Date: Sat, 13 Jun 2026 16:08:26 +0200 Subject: [PATCH 6/7] functional grouping of shared state --- .../table/iceberg_table_entry.cpp | 2 +- src/core/deletes/iceberg_deletion_vector.cpp | 6 +- src/core/deletes/iceberg_equality_delete.cpp | 9 +- .../deletes/iceberg_positional_delete.cpp | 4 +- src/execution/operator/iceberg_delete.cpp | 6 +- src/function/scan/iceberg_scan.cpp | 5 +- .../planning/iceberg_manifest_read_state.hpp | 12 +- .../planning/iceberg_multi_file_list.hpp | 130 +++--- src/planning/iceberg_multi_file_list.cpp | 402 +++++++++++------- src/planning/iceberg_multi_file_reader.cpp | 26 +- src/planning/iceberg_optimizer.cpp | 11 +- .../test_delete_dv_merge_runtime_filter.test | 84 ++++ test/sql/local/equality_deletes.test | 11 + 13 files changed, 445 insertions(+), 263 deletions(-) diff --git a/src/catalog/rest/catalog_entry/table/iceberg_table_entry.cpp b/src/catalog/rest/catalog_entry/table/iceberg_table_entry.cpp index 9548efd1a..29f06207e 100644 --- a/src/catalog/rest/catalog_entry/table/iceberg_table_entry.cpp +++ b/src/catalog/rest/catalog_entry/table/iceberg_table_entry.cpp @@ -244,7 +244,7 @@ TableFunction IcebergTableEntry::GetScanFunction(ClientContext &context, unique_ file_bind_data.virtual_columns = GetVirtualColumns(); D_ASSERT(file_bind_data.file_list); auto &ic_file_list = file_bind_data.file_list->Cast(); - ic_file_list.table = this; + ic_file_list.SetTable(this); return iceberg_scan_function; } diff --git a/src/core/deletes/iceberg_deletion_vector.cpp b/src/core/deletes/iceberg_deletion_vector.cpp index cddf17dee..9f5e024ff 100644 --- a/src/core/deletes/iceberg_deletion_vector.cpp +++ b/src/core/deletes/iceberg_deletion_vector.cpp @@ -157,8 +157,8 @@ void IcebergMultiFileList::ScanPuffinFile(const BoundIcebergManifestEntry &bound auto buf_handle = caching_file_handle->Read(data, length, offset); auto buffer_data = buf_handle.Ptr(); - auto it = positional_delete_data->find(data_file.referenced_data_file); - if (it != positional_delete_data->end()) { + auto it = shared_state->positional_delete_data.find(data_file.referenced_data_file); + if (it != shared_state->positional_delete_data.end()) { //! Another delete already exists for this table auto &existing_delete = *it->second; if (existing_delete.type == IcebergDeleteType::DELETION_VECTOR) { @@ -167,7 +167,7 @@ void IcebergMultiFileList::ScanPuffinFile(const BoundIcebergManifestEntry &bound } } //! NOTE: assign, don't emplace, deletion vectors take priority over any remaining positional delete files - (*positional_delete_data)[data_file.referenced_data_file] = + shared_state->positional_delete_data[data_file.referenced_data_file] = IcebergDeletionVectorData::FromBlob(bound_entry, buffer_data, length); } diff --git a/src/core/deletes/iceberg_equality_delete.cpp b/src/core/deletes/iceberg_equality_delete.cpp index 5cd1a3bba..0e63ac13a 100644 --- a/src/core/deletes/iceberg_equality_delete.cpp +++ b/src/core/deletes/iceberg_equality_delete.cpp @@ -65,10 +65,11 @@ void IcebergMultiFileList::ScanEqualityDeleteFile(const BoundIcebergManifestEntr const auto sequence_number = manifest_entry.GetSequenceNumber(manifest_file); //! Get or create the equality delete data for this sequence number - auto it = equality_delete_data->find(sequence_number); - if (it == equality_delete_data->end()) { - it = - equality_delete_data->emplace(sequence_number, make_uniq(sequence_number)).first; + auto it = shared_state->equality_delete_data.find(sequence_number); + if (it == shared_state->equality_delete_data.end()) { + it = shared_state->equality_delete_data + .emplace(sequence_number, make_uniq(sequence_number)) + .first; } auto &deletes = *it->second; diff --git a/src/core/deletes/iceberg_positional_delete.cpp b/src/core/deletes/iceberg_positional_delete.cpp index 8819a1813..36471d12e 100644 --- a/src/core/deletes/iceberg_positional_delete.cpp +++ b/src/core/deletes/iceberg_positional_delete.cpp @@ -38,7 +38,7 @@ void IcebergMultiFileList::ScanPositionalDeleteFile(const BoundIcebergManifestEn } reference current_file_path = names[0]; auto initial_key = current_file_path.get().GetString(); - auto deletes = TryGetOrCreate(*positional_delete_data, bound_entry, initial_key); + auto deletes = TryGetOrCreate(shared_state->positional_delete_data, bound_entry, initial_key); for (idx_t i = 0; i < count; i++) { auto &name = names[i]; @@ -47,7 +47,7 @@ void IcebergMultiFileList::ScanPositionalDeleteFile(const BoundIcebergManifestEn if (name != current_file_path.get()) { current_file_path = name; auto key = current_file_path.get().GetString(); - deletes = TryGetOrCreate(*positional_delete_data, bound_entry, key); + deletes = TryGetOrCreate(shared_state->positional_delete_data, bound_entry, key); } if (!deletes) { continue; diff --git a/src/execution/operator/iceberg_delete.cpp b/src/execution/operator/iceberg_delete.cpp index c4e1fecd5..069ddcab5 100644 --- a/src/execution/operator/iceberg_delete.cpp +++ b/src/execution/operator/iceberg_delete.cpp @@ -278,9 +278,9 @@ void IcebergDelete::FlushDeletes(IcebergTransaction &transaction, ClientContext } if (write_deletion_vector) { //! Addd the existing delete we're replacing - auto it = multi_file_list->positional_delete_data->find(filename); - if (it != multi_file_list->positional_delete_data->end()) { - auto &delete_data = *it->second; + auto existing_delete = multi_file_list->GetExistingPositionalDeleteData(filename); + if (existing_delete) { + auto &delete_data = *existing_delete; PopulateAlteredManifests(*multi_file_list, global_state.altered_manifests, delete_data); delete_data.ToSet(sorted_deletes); } diff --git a/src/function/scan/iceberg_scan.cpp b/src/function/scan/iceberg_scan.cpp index 87b303da1..74bd5db6b 100644 --- a/src/function/scan/iceberg_scan.cpp +++ b/src/function/scan/iceberg_scan.cpp @@ -59,10 +59,11 @@ static unique_ptr IcebergScanDeserialize(Deserializer &deserialize BindInfo IcebergBindInfo(const optional_ptr bind_data) { auto &multi_file_data = bind_data->Cast(); auto &file_list = multi_file_data.file_list->Cast(); - if (!file_list.table) { + auto table = file_list.GetTable(); + if (!table) { return BindInfo(ScanType::EXTERNAL); } - return BindInfo(*file_list.table); + return BindInfo(*table); } //! FIXME: needs v1.5.1, causes a crash on v1.5.0 diff --git a/src/include/planning/iceberg_manifest_read_state.hpp b/src/include/planning/iceberg_manifest_read_state.hpp index 4866a7bd9..ebf3cbd23 100644 --- a/src/include/planning/iceberg_manifest_read_state.hpp +++ b/src/include/planning/iceberg_manifest_read_state.hpp @@ -24,18 +24,12 @@ struct ManifestReadBatch { struct ManifestEntryReadState { public: void PushBatch(ManifestReadBatch &&batch); - bool HasCurrentBatch() const; - optional_ptr GetCurrentBatch(); - void FinishBatch(); - -private: - bool has_batch = false; - ManifestReadBatch current_batch; + bool GetBatch(idx_t batch_idx, ManifestReadBatch &result) const; private: //! Lock guarding the batches against concurrent access - mutex lock; - queue batches; + mutable mutex lock; + vector batches; }; } // namespace duckdb diff --git a/src/include/planning/iceberg_multi_file_list.hpp b/src/include/planning/iceberg_multi_file_list.hpp index de7647aaa..efb599c4b 100644 --- a/src/include/planning/iceberg_multi_file_list.hpp +++ b/src/include/planning/iceberg_multi_file_list.hpp @@ -31,6 +31,9 @@ namespace duckdb { +class IcebergTableEntry; +struct IcebergMultiFileList; + struct IcebergManifestScanningState { public: IcebergManifestScanningState(ClientContext &context, unique_ptr scan, @@ -46,6 +49,49 @@ struct IcebergManifestScanningState { atomic in_progress_tasks; }; +struct IcebergMultiFileListSharedState { +public: + IcebergMultiFileListSharedState(ClientContext &context, shared_ptr scan_info, string path, + const IcebergOptions &options); + ~IcebergMultiFileListSharedState(); + +private: + friend struct IcebergMultiFileList; + + ClientContext &context; + FileSystem &fs; + shared_ptr scan_info; + string path; + IcebergTableEntry *table = nullptr; + IcebergOptions options; + + mutable mutex lock; + mutable mutex delete_lock; + mutable ManifestEntryReadState read_state; + + mutable bool initialized = false; + + //! Scanned delete manifests and their owners. + mutable vector committed_delete_manifests; + mutable vector> transaction_delete_manifests; + mutable unique_ptr delete_manifest_scan; + mutable unique_ptr delete_manifest_reader; + + //! Scanned data manifests and their owners. + mutable vector committed_data_manifests; + mutable vector> transaction_data_manifests; + mutable unique_ptr data_manifest_read_state; + mutable unique_ptr data_manifest_reader; + + //! Declared after the manifest owners so references in parsed delete data are destroyed first. + mutable case_insensitive_map_t> positional_delete_data; + mutable map> equality_delete_data; + mutable unordered_set processed_delete_files; + + //! Populated as parsed data-file entries become visible to any filtered view. + mutable case_insensitive_map_t> data_file_partition_info; +}; + struct IcebergMultiFileList : public MultiFileList { public: IcebergMultiFileList(ClientContext &context, shared_ptr scan_info, const string &path, @@ -60,32 +106,24 @@ struct IcebergMultiFileList : public MultiFileList { const IcebergTransactionData &GetTransactionData() const; const IcebergSnapshotScanInfo &GetSnapshot() const; const IcebergTableSchema &GetSchema() const; - bool FinishedScanningDeletes() const; + IcebergTableEntry *GetTable() const; + void SetTable(IcebergTableEntry *table); + void SetOptions(const IcebergOptions &options); void Bind(vector &return_types, vector &names); unique_ptr PushdownInternal(ClientContext &context, TableFilterSet &new_filters) const; - void ScanPositionalDeleteFile(const BoundIcebergManifestEntry &manifest_entry, DataChunk &result) const; - void ScanEqualityDeleteFile(const BoundIcebergManifestEntry &manifest_entry, DataChunk &result, - vector &columns, - const vector &global_columns, - const vector &global_column_ids, - const vector &projection_ids) const; - void ScanDeleteFile(const BoundIcebergManifestEntry &entry, const vector &global_columns, - const vector &global_column_ids, const vector &projection_ids) const; - void ScanPuffinFile(const BoundIcebergManifestEntry &entry) const; unique_ptr GetPositionalDeletesForFile(const string &file_path) const; - void EnumerateDeleteManifestEntries() const; void ProcessDeletes(const vector &global_columns, const vector &global_column_ids, const vector &projection_ids) const; - void ScanDeleteFiles(const vector &global_columns, - const vector &global_column_ids, const vector &projection_ids) const; vector> GetEqualityDeletesForFile(const BoundIcebergManifestEntry &manifest_entry) const; void GetStatistics(vector &result) const; - const BoundIcebergManifestEntry &GetManifestEntry(idx_t file_id) const; + BoundIcebergManifestEntry GetManifestEntry(idx_t file_id) const; vector GetPartitionInfoForDataFile(const string &file_path) const; const IcebergManifestFile &GetManifestFileForEntry(const BoundIcebergManifestEntry &entry, IcebergManifestContentType type) const; + vector GetDeleteManifestEntries() const; + shared_ptr GetExistingPositionalDeleteData(const string &file_path) const; public: //! MultiFileList API @@ -120,66 +158,58 @@ struct IcebergMultiFileList : public MultiFileList { const ColumnIndex &column_index) const; private: - optional_ptr TryGetNextBatch(lock_guard &guard) const; + IcebergMultiFileList(shared_ptr shared_state); + bool TryGetNextBatch(lock_guard &guard) const; void FinishScanTasks(lock_guard &guard) const; + void InitializeSharedState(lock_guard &guard) const; + bool FinishedScanningDeletes() const; + void EnumerateDeleteManifestEntriesInternal() const; + void ProcessDeletesInternal(const vector &global_columns, + const vector &global_column_ids, + const vector &projection_ids) const; + void ScanDeleteFiles(const vector &global_columns, + const vector &global_column_ids, const vector &projection_ids) const; + void ScanDeleteFile(const BoundIcebergManifestEntry &entry, const vector &global_columns, + const vector &global_column_ids, const vector &projection_ids) const; + void ScanPositionalDeleteFile(const BoundIcebergManifestEntry &manifest_entry, DataChunk &result) const; + void ScanEqualityDeleteFile(const BoundIcebergManifestEntry &manifest_entry, DataChunk &result, + vector &columns, + const vector &global_columns, + const vector &global_column_ids, + const vector &projection_ids) const; + void ScanPuffinFile(const BoundIcebergManifestEntry &entry) const; -public: +private: + shared_ptr shared_state; ClientContext &context; FileSystem &fs; - shared_ptr scan_info; - string path; - IcebergTableEntry *table; - - mutable mutex lock; + const IcebergOptions &options; //! ComplexFilterPushdown results bool have_bound = false; vector names; vector types; TableFilterSet table_filters; - mutable ManifestEntryReadState read_state; - - //! For each file that has a delete file, the state for processing that/those delete file(s). - //! Shared with the filter pushdown copies of this list (see PushdownInternal) so that delete state - //! observed while scanning is also visible to the list referenced by DML operators. - mutable shared_ptr>> positional_delete_data; - //! All equality deletes with sequence numbers higher than that of the data_file apply to that data_file. - //! Shared with the filter pushdown copies of this list for the same reason as positional_delete_data. - mutable shared_ptr>> equality_delete_data; - - //! FIXME: this is only used in 'FinalizeBind', - //! shouldn't this be used to protect all the variable accesses that are accessed there while the lock is held? - mutable mutex delete_lock; mutable idx_t transaction_delete_idx = 0; - // have we enumerated all delete files (which we need to do for equality delete planning) mutable bool committed_delete_entries_enumerated = false; mutable bool initialized = false; - mutable bool scanned_delete_manifests = false; - const IcebergOptions &options; + mutable idx_t next_data_batch_idx = 0; + mutable bool has_current_data_batch = false; + mutable ManifestReadBatch current_data_batch; + mutable idx_t current_data_batch_offset = 0; -public: //! References to items inside the 'manifest_entries' of the list entries in the 'delete_manifests' mutable vector delete_manifest_entries; //! Combination of committed + transaction delete manifests mutable vector delete_manifests; - //! Scanned delete manifests of the snapshot being scanned - mutable unique_ptr delete_manifest_scan; - mutable unique_ptr delete_manifest_reader; - mutable vector committed_delete_manifests; - //! Cached, uncommitted delete manifests created by earlier statements in the transaction - mutable vector> transaction_delete_manifests; + mutable vector delete_manifest_matches; private: //! References to items inside the 'manifest_entries' of the list entries in the 'data_manifests' mutable vector data_manifest_entries; //! Combination of committed + transaction data manifests mutable vector data_manifests; - //! Scanned data manifests of the snapshot being scanned - mutable unique_ptr data_manifest_read_state; - mutable unique_ptr data_manifest_reader; - mutable vector committed_data_manifests; - //! Cached, uncommitted data manifests created by earlier statements in the transaction - mutable vector> transaction_data_manifests; + mutable vector data_manifest_matches; }; } // namespace duckdb diff --git a/src/planning/iceberg_multi_file_list.cpp b/src/planning/iceberg_multi_file_list.cpp index 76ae01ad1..a242a1086 100644 --- a/src/planning/iceberg_multi_file_list.cpp +++ b/src/planning/iceberg_multi_file_list.cpp @@ -34,29 +34,16 @@ namespace duckdb { void ManifestEntryReadState::PushBatch(ManifestReadBatch &&batch) { lock_guard guard(lock); - batches.push(std::move(batch)); + batches.push_back(std::move(batch)); } -bool ManifestEntryReadState::HasCurrentBatch() const { - return has_batch; -} - -optional_ptr ManifestEntryReadState::GetCurrentBatch() { - if (!has_batch) { - lock_guard guard(lock); - if (batches.empty()) { - return nullptr; - } - auto &batch = batches.front(); - current_batch = std::move(batch); - batches.pop(); - has_batch = true; +bool ManifestEntryReadState::GetBatch(idx_t batch_idx, ManifestReadBatch &result) const { + lock_guard guard(lock); + if (batch_idx >= batches.size()) { + return false; } - return current_batch; -} - -void ManifestEntryReadState::FinishBatch() { - has_batch = false; + result = batches[batch_idx]; + return true; } namespace { @@ -117,16 +104,30 @@ class ManifestReadTask : public BaseExecutorTask { IcebergMultiFileList::IcebergMultiFileList(ClientContext &context_p, shared_ptr scan_info, const string &path, const IcebergOptions &options) - : context(context_p), fs(FileSystem::GetFileSystem(context)), scan_info(scan_info), path(path), table(nullptr), - options(options), - positional_delete_data(make_shared_ptr>>()), - equality_delete_data(make_shared_ptr>>()) { + : shared_state(make_shared_ptr(context_p, std::move(scan_info), path, options)), + context(shared_state->context), fs(shared_state->fs), options(shared_state->options) { +} + +IcebergMultiFileList::IcebergMultiFileList(shared_ptr shared_state_p) + : shared_state(std::move(shared_state_p)), context(shared_state->context), fs(shared_state->fs), + options(shared_state->options) { } IcebergMultiFileList::~IcebergMultiFileList() { - lock_guard guard(lock); - //! FIXME: this could throw, if the tasks encountered an error - FinishScanTasks(guard); +} + +IcebergMultiFileListSharedState::IcebergMultiFileListSharedState(ClientContext &context_p, + shared_ptr scan_info_p, string path_p, + const IcebergOptions &options_p) + : context(context_p), fs(FileSystem::GetFileSystem(context)), scan_info(std::move(scan_info_p)), + path(std::move(path_p)), options(options_p) { +} + +IcebergMultiFileListSharedState::~IcebergMultiFileListSharedState() { + if (data_manifest_read_state) { + //! FIXME: this could throw, if the tasks encountered an error + data_manifest_read_state->executor.WorkOnTasks(); + } } string IcebergMultiFileList::ToDuckDBPath(const string &raw_path) { @@ -134,32 +135,44 @@ string IcebergMultiFileList::ToDuckDBPath(const string &raw_path) { } string IcebergMultiFileList::GetPath() const { - return path; + return shared_state->path; } const IcebergTableMetadata &IcebergMultiFileList::GetMetadata() const { - return scan_info->metadata; + return shared_state->scan_info->metadata; } bool IcebergMultiFileList::HasTransactionData() const { - return scan_info->transaction_data; + return shared_state->scan_info->transaction_data; } const IcebergTransactionData &IcebergMultiFileList::GetTransactionData() const { D_ASSERT(HasTransactionData()); - return *scan_info->transaction_data; + return *shared_state->scan_info->transaction_data; } const IcebergSnapshotScanInfo &IcebergMultiFileList::GetSnapshot() const { - return scan_info->snapshot_info; + return shared_state->scan_info->snapshot_info; } const IcebergTableSchema &IcebergMultiFileList::GetSchema() const { - return scan_info->schema; + return shared_state->scan_info->schema; } bool IcebergMultiFileList::FinishedScanningDeletes() const { - return !delete_manifest_reader || delete_manifest_reader->Finished(); + return !shared_state->delete_manifest_reader || shared_state->delete_manifest_reader->Finished(); +} + +IcebergTableEntry *IcebergMultiFileList::GetTable() const { + return shared_state->table; +} + +void IcebergMultiFileList::SetTable(IcebergTableEntry *table) { + shared_state->table = table; +} + +void IcebergMultiFileList::SetOptions(const IcebergOptions &options) { + shared_state->options = options; } optional_ptr IcebergMultiFileList::GetFilterForColumnIndex(const TableFilterSet &filter_set, @@ -192,7 +205,7 @@ optional_ptr IcebergMultiFileList::GetFilterForColumnIndex(co } void IcebergMultiFileList::Bind(vector &return_types, vector &names) { - lock_guard guard(lock); + lock_guard guard(shared_state->lock); if (have_bound) { names = this->names; @@ -201,9 +214,9 @@ void IcebergMultiFileList::Bind(vector &return_types, vector(FileSystem::GetFileSystem(context), *context.db); - if (!scan_info) { - D_ASSERT(!path.empty()); - auto input_string = path; + if (!shared_state->scan_info) { + D_ASSERT(!shared_state->path.empty()); + auto input_string = shared_state->path; auto iceberg_path = IcebergUtils::GetStorageLocation(context, input_string); auto iceberg_meta_path = IcebergTableMetadata::GetMetaDataPath(context, iceberg_path, fs, options); auto table_metadata = @@ -216,7 +229,8 @@ void IcebergMultiFileList::Bind(vector &return_types, vector(iceberg_path, std::move(temp_data), snapshot_info, *schema); + shared_state->scan_info = + make_shared_ptr(iceberg_path, std::move(temp_data), snapshot_info, *schema); } if (!initialized) { @@ -241,7 +255,7 @@ void IcebergMultiFileList::Bind(vector &return_types, vector IcebergMultiFileList::PushdownInternal(ClientContext &context, TableFilterSet &new_filters) const { - auto filtered_list = make_uniq(context, scan_info, path, this->options); + auto filtered_list = unique_ptr(new IcebergMultiFileList(shared_state)); TableFilterSet result_filter_set; @@ -262,10 +276,6 @@ unique_ptr IcebergMultiFileList::PushdownInternal(ClientCo filtered_list->names = names; filtered_list->types = types; filtered_list->have_bound = true; - //! Share the delete state: DML operators hold a reference to the original list of the scan's bind data, - //! while the scan itself can read through a filtered copy created by (dynamic) filter pushdown. - filtered_list->positional_delete_data = positional_delete_data; - filtered_list->equality_delete_data = equality_delete_data; return filtered_list; } @@ -321,7 +331,7 @@ unique_ptr IcebergMultiFileList::ComplexFilterPushdown(ClientCont vector IcebergMultiFileList::GetAllFiles() const { vector file_list; //! Lock is required because it reads the 'manifest_entries' vector - lock_guard guard(lock); + lock_guard guard(shared_state->lock); for (idx_t i = 0; i < data_manifest_entries.size(); i++) { file_list.push_back(GetFileInternal(i, guard)); } @@ -330,7 +340,7 @@ vector IcebergMultiFileList::GetAllFiles() const { FileExpandResult IcebergMultiFileList::GetExpandResult() const { // GetFileInternal(1) will ensure files with index 0 and index 1 are expanded if they are available - lock_guard guard(lock); + lock_guard guard(shared_state->lock); GetFileInternal(1, guard); // always return multiple files, In the case there is only 1 data file, @@ -341,7 +351,7 @@ FileExpandResult IcebergMultiFileList::GetExpandResult() const { idx_t IcebergMultiFileList::GetTotalFileCount() const { // FIXME: the 'added_files_count' + the 'existing_files_count' // in the Manifest List should give us this information without scanning the manifest file(s) - lock_guard guard(lock); + lock_guard guard(shared_state->lock); idx_t i = data_manifest_entries.size(); while (!GetFileInternal(i, guard).path.empty()) { @@ -363,32 +373,32 @@ unique_ptr IcebergMultiFileList::GetCardinality(ClientContext &c idx_t cardinality = 0; for (idx_t i = 0; i < data_manifests.size(); i++) { auto &manifest = data_manifests[i].entry.file; + if (!data_manifest_matches[i]) { + continue; + } cardinality += manifest.added_rows_count; cardinality += manifest.existing_rows_count; } for (idx_t i = 0; i < delete_manifests.size(); i++) { auto &manifest = delete_manifests[i].entry.file; + if (!delete_manifest_matches[i]) { + continue; + } cardinality -= manifest.added_rows_count; } return make_uniq(cardinality, cardinality); } -const BoundIcebergManifestEntry &IcebergMultiFileList::GetManifestEntry(idx_t file_id) const { +BoundIcebergManifestEntry IcebergMultiFileList::GetManifestEntry(idx_t file_id) const { + lock_guard guard(shared_state->lock); return data_manifest_entries[file_id]; } vector IcebergMultiFileList::GetPartitionInfoForDataFile(const string &file_path) const { - lock_guard guard(lock); - auto iceberg_path = GetPath(); - for (auto &bound_entry : data_manifest_entries) { - auto &data_file = bound_entry.entry.data_file; - string entry_path = data_file.file_path; - if (options.allow_moved_paths) { - entry_path = IcebergUtils::GetFullPath(iceberg_path, entry_path, fs); - } - if (StringUtil::CIEquals(entry_path, file_path)) { - return data_file.partition_info; - } + lock_guard guard(shared_state->lock); + auto entry = shared_state->data_file_partition_info.find(file_path); + if (entry != shared_state->data_file_partition_info.end()) { + return entry->second; } throw InternalException("Could not find data file '%s' in manifest entries", file_path); } @@ -408,14 +418,19 @@ void IcebergMultiFileList::GetStatistics(vector &result) co return; } - if (!delete_manifests.empty()) { - //! if exist delete_manifests, return; - return; + for (idx_t i = 0; i < delete_manifests.size(); i++) { + if (delete_manifest_matches[i]) { + //! if a matching delete manifest exists, return; + return; + } } idx_t count = 0; for (idx_t i = 0; i < data_manifests.size(); i++) { auto &manifest = data_manifests[i].entry.file; + if (!data_manifest_matches[i]) { + continue; + } count += manifest.existing_rows_count; count += manifest.added_rows_count; } @@ -717,17 +732,21 @@ bool IcebergMultiFileList::FileMatchesFilter(const IcebergManifestFile &manifest return true; } -optional_ptr IcebergMultiFileList::TryGetNextBatch(lock_guard &guard) const { - auto batch = read_state.GetCurrentBatch(); - if (batch) { - //! We are still reading a batch, or have batches to read - return batch; +bool IcebergMultiFileList::TryGetNextBatch(lock_guard &guard) const { + if (has_current_data_batch) { + return true; } - if (!data_manifest_read_state) { - return batch; + if (shared_state->read_state.GetBatch(next_data_batch_idx, current_data_batch)) { + next_data_batch_idx++; + current_data_batch_offset = current_data_batch.start_index; + has_current_data_batch = true; + return true; + } + if (!shared_state->data_manifest_read_state) { + return false; } auto &scheduler = TaskScheduler::GetScheduler(context); - auto &scan_state = *data_manifest_read_state; + auto &scan_state = *shared_state->data_manifest_read_state; auto &executor = scan_state.executor; shared_ptr task_to_execute; while (scan_state.in_progress_tasks) { @@ -737,25 +756,33 @@ optional_ptr IcebergMultiFileList::TryGetNextBatch(lock_guard auto &token = *task_to_execute->token; scheduler.ScheduleTask(token, std::move(task_to_execute)); } - batch = read_state.GetCurrentBatch(); - if (batch) { - return batch; + if (shared_state->read_state.GetBatch(next_data_batch_idx, current_data_batch)) { + next_data_batch_idx++; + current_data_batch_offset = current_data_batch.start_index; + has_current_data_batch = true; + return true; } //! We didn't manage to populate the buffer with our scan //! But another task might be in the process of scanning //! Have to wait for everything to finish to conclusively say we're done } executor.WorkOnTasks(); - return read_state.GetCurrentBatch(); + break; } - return read_state.GetCurrentBatch(); + if (!shared_state->read_state.GetBatch(next_data_batch_idx, current_data_batch)) { + return false; + } + next_data_batch_idx++; + current_data_batch_offset = current_data_batch.start_index; + has_current_data_batch = true; + return true; } void IcebergMultiFileList::FinishScanTasks(lock_guard &guard) const { - if (!data_manifest_read_state) { + if (!shared_state->data_manifest_read_state) { return; } - auto &read_state = *data_manifest_read_state; + auto &read_state = *shared_state->data_manifest_read_state; auto &executor = read_state.executor; //! Make sure all tasks are done before shutting down executor.WorkOnTasks(); @@ -770,25 +797,34 @@ optional_ptr IcebergMultiFileList::GetDataFile( } while (file_id >= data_manifest_entries.size()) { - auto batch = TryGetNextBatch(guard); - if (!batch) { + if (!TryGetNextBatch(guard)) { FinishScanTasks(guard); return nullptr; } - auto ¤t_batch = *batch; + auto ¤t_batch = current_data_batch; auto &bound_manifest_list_entry = data_manifests[current_batch.manifest_list_entry_idx]; auto &manifest_list_entry = bound_manifest_list_entry.entry; auto &manifest_entries = manifest_list_entry.manifest_entries; auto &manifest_file = manifest_list_entry.file; - for (; current_batch.start_index < current_batch.end_index && file_id >= data_manifest_entries.size(); - current_batch.start_index++) { - auto &manifest_entry = manifest_entries[current_batch.start_index]; + if (!data_manifest_matches[current_batch.manifest_list_entry_idx]) { + current_data_batch_offset = current_batch.end_index; + } + for (; current_data_batch_offset < current_batch.end_index && file_id >= data_manifest_entries.size(); + current_data_batch_offset++) { + auto &manifest_entry = manifest_entries[current_data_batch_offset]; + auto &data_file = manifest_entry.data_file; + auto entry_path = data_file.file_path; + if (options.allow_moved_paths) { + entry_path = IcebergUtils::GetFullPath(GetPath(), entry_path, fs); + } + shared_state->data_file_partition_info[entry_path] = data_file.partition_info; + shared_state->data_file_partition_info[data_file.file_path] = data_file.partition_info; + if (manifest_entry.status == IcebergManifestEntryStatusType::DELETED) { continue; } - auto &data_file = manifest_entry.data_file; // Check whether current data file is filtered out. if (!table_filters.filters.empty() && !FileMatchesFilter(manifest_file, manifest_entry, IcebergManifestContentType::DATA)) { @@ -806,8 +842,8 @@ optional_ptr IcebergMultiFileList::GetDataFile( auto bound_entry = bound_manifest_list_entry.BindEntry(manifest_entry); data_manifest_entries.push_back(bound_entry); } - if (current_batch.start_index >= current_batch.end_index) { - read_state.FinishBatch(); + if (current_data_batch_offset >= current_batch.end_index) { + has_current_data_batch = false; } } return data_manifest_entries[file_id]; @@ -857,7 +893,7 @@ OpenFileInfo IcebergMultiFileList::GetFileInternal(idx_t file_id, lock_guard guard(lock); + lock_guard guard(shared_state->lock); return GetFileInternal(file_id, guard); } @@ -928,6 +964,7 @@ bool IcebergMultiFileList::ManifestMatchesFilter(const IcebergManifestFile &mani vector> IcebergMultiFileList::GetEqualityDeletesForFile(const BoundIcebergManifestEntry &bound_manifest_entry) const { + lock_guard guard(shared_state->delete_lock); vector> result; //! Look through all the equality delete files with a *higher* sequence number @@ -935,8 +972,8 @@ IcebergMultiFileList::GetEqualityDeletesForFile(const BoundIcebergManifestEntry auto &manifest_file = data_manifests[bound_manifest_entry.manifest_file_idx].entry.file; auto &data_file = manifest_entry.data_file; auto &metadata = GetMetadata(); - auto it = equality_delete_data->upper_bound(manifest_entry.GetSequenceNumber(manifest_file)); - for (; it != equality_delete_data->end(); it++) { + auto it = shared_state->equality_delete_data.upper_bound(manifest_entry.GetSequenceNumber(manifest_file)); + for (; it != shared_state->equality_delete_data.end(); it++) { auto &files = it->second->files; for (auto &file : files) { auto &partition_spec = metadata.partition_specs.at(file.partition_spec_id); @@ -965,9 +1002,42 @@ void IcebergMultiFileList::InitializeFiles(lock_guard &guard) const { if (initialized) { return; } + InitializeSharedState(guard); + + auto &committed_data_manifests = shared_state->committed_data_manifests; + auto &transaction_data_manifests = shared_state->transaction_data_manifests; + data_manifests.reserve(committed_data_manifests.size() + transaction_data_manifests.size()); + data_manifest_matches.reserve(committed_data_manifests.size() + transaction_data_manifests.size()); + for (auto &manifest : committed_data_manifests) { + data_manifests.emplace_back(data_manifests.size(), manifest); + data_manifest_matches.push_back(ManifestMatchesFilter(manifest.file)); + } + for (auto &manifest : transaction_data_manifests) { + data_manifests.emplace_back(data_manifests.size(), manifest); + data_manifest_matches.push_back(ManifestMatchesFilter(manifest.get().file)); + } + + auto &committed_delete_manifests = shared_state->committed_delete_manifests; + auto &transaction_delete_manifests = shared_state->transaction_delete_manifests; + delete_manifests.reserve(committed_delete_manifests.size() + transaction_delete_manifests.size()); + delete_manifest_matches.reserve(committed_delete_manifests.size() + transaction_delete_manifests.size()); + for (auto &manifest : committed_delete_manifests) { + delete_manifests.emplace_back(delete_manifests.size(), manifest); + delete_manifest_matches.push_back(ManifestMatchesFilter(manifest.file)); + } + for (auto &manifest : transaction_delete_manifests) { + delete_manifests.emplace_back(delete_manifests.size(), manifest); + delete_manifest_matches.push_back(ManifestMatchesFilter(manifest.get().file)); + } initialized = true; +} + +void IcebergMultiFileList::InitializeSharedState(lock_guard &guard) const { + if (shared_state->initialized) { + return; + } - auto &snapshot_info = scan_info->snapshot_info; + auto &snapshot_info = shared_state->scan_info->snapshot_info; if (snapshot_info.snapshot) { //! Load the snapshot auto iceberg_path = GetPath(); @@ -996,23 +1066,24 @@ void IcebergMultiFileList::InitializeFiles(lock_guard &guard) const { for (auto &manifest_list_entry : manifest_list_entries) { auto &manifest_file = manifest_list_entry.file; - if (!ManifestMatchesFilter(manifest_file)) { - //! Skip this manifest - continue; - } - if (manifest_file.content == IcebergManifestContentType::DATA) { - committed_data_manifests.push_back(std::move(manifest_list_entry)); + shared_state->committed_data_manifests.push_back(std::move(manifest_list_entry)); } else { D_ASSERT(manifest_file.content == IcebergManifestContentType::DELETE); - committed_delete_manifests.push_back(std::move(manifest_list_entry)); + shared_state->committed_delete_manifests.push_back(std::move(manifest_list_entry)); } } - if (!committed_delete_manifests.empty()) { - delete_manifest_scan = AvroScan::ScanManifest(snapshot_info, committed_delete_manifests, options, fs, - iceberg_path, metadata, context); - delete_manifest_reader = make_uniq(*delete_manifest_scan); + for (auto &manifest : shared_state->committed_data_manifests) { + auto &file = manifest.file; + idx_t reserve_size = file.existing_files_count + file.added_files_count + file.deleted_files_count; + manifest.manifest_entries.reserve(reserve_size); + } + if (!shared_state->committed_delete_manifests.empty()) { + shared_state->delete_manifest_scan = AvroScan::ScanManifest( + snapshot_info, shared_state->committed_delete_manifests, options, fs, iceberg_path, metadata, context); + shared_state->delete_manifest_reader = + make_uniq(*shared_state->delete_manifest_scan); } } @@ -1023,17 +1094,13 @@ void IcebergMultiFileList::InitializeFiles(lock_guard &guard) const { const auto &manifest_list_entries = alter.GetManifestFiles(); for (auto &manifest_list_entry : manifest_list_entries) { auto &manifest = manifest_list_entry.file; - if (!ManifestMatchesFilter(manifest)) { - //! Skip this manifest - continue; - } switch (manifest.content) { case IcebergManifestContentType::DATA: { - transaction_data_manifests.push_back(manifest_list_entry); + shared_state->transaction_data_manifests.push_back(manifest_list_entry); break; } case IcebergManifestContentType::DELETE: { - transaction_delete_manifests.push_back(manifest_list_entry); + shared_state->transaction_delete_manifests.push_back(manifest_list_entry); break; } default: @@ -1044,67 +1111,39 @@ void IcebergMultiFileList::InitializeFiles(lock_guard &guard) const { } } - idx_t total_data_manifests = 0; - total_data_manifests += committed_data_manifests.size(); - total_data_manifests += transaction_data_manifests.size(); - data_manifests.reserve(total_data_manifests); - - //! Add all data manifests - for (auto &manifest : committed_data_manifests) { - auto manifest_list_entry_idx = data_manifests.size(); - // reserve upfront → guarantees no reallocation - auto &file = manifest.file; - idx_t reserve_size = file.existing_files_count + file.added_files_count + file.deleted_files_count; - manifest.manifest_entries.reserve(reserve_size); - - data_manifests.emplace_back(manifest_list_entry_idx, manifest); - } - for (auto &manifest : transaction_data_manifests) { - auto manifest_list_entry_idx = data_manifests.size(); - data_manifests.emplace_back(manifest_list_entry_idx, manifest); - read_state.PushBatch(ManifestReadBatch {manifest_list_entry_idx, 0, manifest.get().manifest_entries.size()}); - } - - idx_t total_delete_manifests = 0; - total_delete_manifests += committed_delete_manifests.size(); - total_delete_manifests += transaction_delete_manifests.size(); - delete_manifests.reserve(total_delete_manifests); - - //! Add all delete manifests - for (auto &manifest : committed_delete_manifests) { - auto index = delete_manifests.size(); - delete_manifests.emplace_back(index, manifest); - } - for (auto &manifest : transaction_delete_manifests) { - auto index = delete_manifests.size(); - delete_manifests.emplace_back(index, manifest); + idx_t transaction_manifest_idx = shared_state->committed_data_manifests.size(); + for (auto &manifest : shared_state->transaction_data_manifests) { + shared_state->read_state.PushBatch( + ManifestReadBatch {transaction_manifest_idx++, 0, manifest.get().manifest_entries.size()}); } - if (!committed_data_manifests.empty()) { + if (!shared_state->committed_data_manifests.empty()) { auto &metadata = GetMetadata(); auto &snapshot_info = GetSnapshot(); auto iceberg_path = GetPath(); auto &fs = FileSystem::GetFileSystem(context); - auto data_scan = AvroScan::ScanManifest(snapshot_info, committed_data_manifests, options, fs, iceberg_path, - metadata, context, &read_state); - data_manifest_read_state = - make_uniq(context, std::move(data_scan), committed_data_manifests); - data_manifest_reader = make_uniq(*data_manifest_read_state->scan); + auto data_scan = AvroScan::ScanManifest(snapshot_info, shared_state->committed_data_manifests, options, fs, + iceberg_path, metadata, context, &shared_state->read_state); + shared_state->data_manifest_read_state = make_uniq( + context, std::move(data_scan), shared_state->committed_data_manifests); + shared_state->data_manifest_reader = + make_uniq(*shared_state->data_manifest_read_state->scan); - auto &executor = data_manifest_read_state->executor; + auto &executor = shared_state->data_manifest_read_state->executor; auto &scheduler = TaskScheduler::GetScheduler(context); auto worker_thread_count = scheduler.NumberOfThreads(); - auto num_threads = MinValue(worker_thread_count, data_manifests.size()); - data_manifest_read_state->in_progress_tasks = num_threads; + auto num_threads = MinValue(worker_thread_count, shared_state->committed_data_manifests.size()); + shared_state->data_manifest_read_state->in_progress_tasks = num_threads; for (idx_t i = 0; i < num_threads; i++) { - executor.ScheduleTask(make_uniq(*data_manifest_read_state)); + executor.ScheduleTask(make_uniq(*shared_state->data_manifest_read_state)); } } + shared_state->initialized = true; } -void IcebergMultiFileList::EnumerateDeleteManifestEntries() const { +void IcebergMultiFileList::EnumerateDeleteManifestEntriesInternal() const { // In <=v2 we now have to process *all* delete manifests // before we can be certain that we have all the delete data for the current file. @@ -1117,14 +1156,17 @@ void IcebergMultiFileList::EnumerateDeleteManifestEntries() const { transactional_delete_files = GetTransactionData().transactional_delete_files; } while (!FinishedScanningDeletes()) { - delete_manifest_reader->Read(); + shared_state->delete_manifest_reader->Read(); } if (!committed_delete_entries_enumerated) { - for (idx_t i = 0; i < committed_delete_manifests.size(); i++) { + for (idx_t i = 0; i < shared_state->committed_delete_manifests.size(); i++) { auto &manifest = delete_manifests[i]; auto &entries = manifest.entry.manifest_entries; auto &manifest_file = manifest.entry.file; + if (!delete_manifest_matches[i]) { + continue; + } for (auto &manifest_entry : entries) { if (manifest_entry.status == IcebergManifestEntryStatusType::DELETED) { continue; @@ -1149,9 +1191,13 @@ void IcebergMultiFileList::EnumerateDeleteManifestEntries() const { committed_delete_entries_enumerated = true; } - auto offset = committed_delete_manifests.size(); - while (transaction_delete_idx < transaction_delete_manifests.size()) { + auto offset = shared_state->committed_delete_manifests.size(); + while (transaction_delete_idx < shared_state->transaction_delete_manifests.size()) { auto &delete_manifest = delete_manifests[offset + transaction_delete_idx]; + if (!delete_manifest_matches[offset + transaction_delete_idx]) { + transaction_delete_idx++; + continue; + } for (auto &manifest_entry : delete_manifest.entry.manifest_entries) { auto &data_file = manifest_entry.data_file; auto &referenced_data_file = data_file.referenced_data_file; @@ -1178,6 +1224,9 @@ void IcebergMultiFileList::ScanDeleteFiles(const vectorprocessed_delete_files.count(data_file.file_path)) { + continue; + } if (StringUtil::CIEquals(data_file.file_format, "parquet")) { ScanDeleteFile(bound_manifest_entry, global_columns, global_column_ids, projection_ids); } else if (StringUtil::CIEquals(data_file.file_format, "puffin")) { @@ -1187,20 +1236,35 @@ void IcebergMultiFileList::ScanDeleteFiles(const vectorprocessed_delete_files.insert(data_file.file_path); } - scanned_delete_manifests = true; } void IcebergMultiFileList::ProcessDeletes(const vector &global_columns, const vector &global_column_ids, const vector &projection_ids) const { + lock_guard guard(shared_state->lock); + InitializeFiles(guard); + lock_guard delete_guard(shared_state->delete_lock); + ProcessDeletesInternal(global_columns, global_column_ids, projection_ids); +} + +void IcebergMultiFileList::ProcessDeletesInternal(const vector &global_columns, + const vector &global_column_ids, + const vector &projection_ids) const { //! Enumerate the delete manifest entries, then read the delete files they reference. - //! EnumerateDeleteManifestEntries() is idempotent, so this is safe even if the entries were + //! Delete enumeration is idempotent, so this is safe even if the entries were //! already enumerated earlier (e.g. by the optimizer). - EnumerateDeleteManifestEntries(); - if (!scanned_delete_manifests) { - ScanDeleteFiles(global_columns, global_column_ids, projection_ids); - } + EnumerateDeleteManifestEntriesInternal(); + ScanDeleteFiles(global_columns, global_column_ids, projection_ids); +} + +vector IcebergMultiFileList::GetDeleteManifestEntries() const { + lock_guard guard(shared_state->lock); + InitializeFiles(guard); + lock_guard delete_guard(shared_state->delete_lock); + EnumerateDeleteManifestEntriesInternal(); + return delete_manifest_entries; } void IcebergMultiFileList::ScanDeleteFile(const BoundIcebergManifestEntry &bound_manifest_entry, @@ -1284,12 +1348,22 @@ void IcebergMultiFileList::ScanDeleteFile(const BoundIcebergManifestEntry &bound } unique_ptr IcebergMultiFileList::GetPositionalDeletesForFile(const string &file_path) const { - auto it = positional_delete_data->find(file_path); - if (it != positional_delete_data->end()) { + lock_guard guard(shared_state->delete_lock); + auto it = shared_state->positional_delete_data.find(file_path); + if (it != shared_state->positional_delete_data.end()) { // There is delete data for this file, return it return it->second->ToFilter(); } return nullptr; } +shared_ptr IcebergMultiFileList::GetExistingPositionalDeleteData(const string &file_path) const { + lock_guard guard(shared_state->delete_lock); + auto it = shared_state->positional_delete_data.find(file_path); + if (it == shared_state->positional_delete_data.end()) { + return nullptr; + } + return it->second; +} + } // namespace duckdb diff --git a/src/planning/iceberg_multi_file_reader.cpp b/src/planning/iceberg_multi_file_reader.cpp index 8ccd88cc9..45223e2c7 100644 --- a/src/planning/iceberg_multi_file_reader.cpp +++ b/src/planning/iceberg_multi_file_reader.cpp @@ -52,6 +52,7 @@ bool IcebergMultiFileReader::Bind(MultiFileOptions &options, MultiFileList &file vector &names, MultiFileReaderBindData &bind_data) { auto &iceberg_multi_file_list = dynamic_cast(files); + iceberg_multi_file_list.SetOptions(this->options); iceberg_multi_file_list.Bind(return_types, names); // FIXME: apply final transformation for 'file_row_number' ??? auto &schema = iceberg_multi_file_list.GetSchema().columns; @@ -180,7 +181,7 @@ static void ApplyPartitionConstants(const IcebergMultiFileList &multi_file_list, // Get the metadata for this file auto &reader = *reader_data.reader; auto file_id = reader.file_list_idx.GetIndex(); - auto &bound_manifest_entry = multi_file_list.GetManifestEntry(file_id); + auto bound_manifest_entry = multi_file_list.GetManifestEntry(file_id); auto &manifest_file = multi_file_list.GetManifestFileForEntry(bound_manifest_entry, IcebergManifestContentType::DATA); auto &manifest_entry = bound_manifest_entry.entry; @@ -289,23 +290,10 @@ void IcebergMultiFileReader::FinalizeBind(MultiFileReaderData &reader_data, cons auto &reader = *reader_data.reader; auto file_id = reader.file_list_idx.GetIndex(); - { - lock_guard guard(multi_file_list.lock); - const auto &bound_manifest_entry = multi_file_list.GetManifestEntry(file_id); - const auto &manifest_entry = bound_manifest_entry.entry; - const auto &data_file = manifest_entry.data_file; - // The path of the data file where this chunk was read from - const auto &file_path = data_file.file_path; - lock_guard delete_guard(multi_file_list.delete_lock); - if (!multi_file_list.FinishedScanningDeletes() || - multi_file_list.transaction_delete_idx < multi_file_list.transaction_delete_manifests.size()) { - multi_file_list.ProcessDeletes(global_columns, global_column_ids, gstate.projection_ids); - } - if (!multi_file_list.scanned_delete_manifests) { - multi_file_list.ScanDeleteFiles(global_columns, global_column_ids, gstate.projection_ids); - } - reader.deletion_filter = multi_file_list.GetPositionalDeletesForFile(file_path); - } + auto bound_manifest_entry = multi_file_list.GetManifestEntry(file_id); + const auto &file_path = bound_manifest_entry.entry.data_file.file_path; + multi_file_list.ProcessDeletes(global_columns, global_column_ids, gstate.projection_ids); + reader.deletion_filter = multi_file_list.GetPositionalDeletesForFile(file_path); auto &local_columns = reader_data.reader->columns; auto &metadata = multi_file_list.GetMetadata(); @@ -436,7 +424,7 @@ void IcebergMultiFileReader::FinalizeChunk(ClientContext &context, const MultiFi D_ASSERT(global_state); // Get the metadata for this file auto file_id = reader.file_list_idx.GetIndex(); - auto &bound_manifest_entry = multi_file_list.GetManifestEntry(file_id); + auto bound_manifest_entry = multi_file_list.GetManifestEntry(file_id); auto &local_columns = reader.columns; ApplyEqualityDeletes(context, output_chunk, multi_file_list, bound_manifest_entry, local_columns); diff --git a/src/planning/iceberg_optimizer.cpp b/src/planning/iceberg_optimizer.cpp index 12028adac..5fb285174 100644 --- a/src/planning/iceberg_optimizer.cpp +++ b/src/planning/iceberg_optimizer.cpp @@ -36,13 +36,10 @@ void GuaranteeEqualityDeleteColumnsOptimizer::VisitOperator(unique_ptrCast(); - { - lock_guard guard(iceberg_list.delete_lock); - iceberg_list.EnumerateDeleteManifestEntries(); - } + auto delete_manifest_entries = iceberg_list.GetDeleteManifestEntries(); unordered_set required_field_ids; - for (auto &entry : iceberg_list.delete_manifest_entries) { + for (auto &entry : delete_manifest_entries) { auto &mft = entry.entry; if (mft.data_file.content != IcebergManifestEntryContentType::EQUALITY_DELETES) { continue; @@ -72,7 +69,9 @@ void GuaranteeEqualityDeleteColumnsOptimizer::VisitOperator(unique_ptrtable_info.table_metadata.FindColumnByFieldId(fid); + auto table = iceberg_list.GetTable(); + D_ASSERT(table); + auto col_info = table->table_info.table_metadata.FindColumnByFieldId(fid); if (!col_info) { throw InvalidConfigurationException( "column %d must apply equality deletes, but no schema has a column with that field id", fid); diff --git a/test/sql/local/catalog_test_config_setup/catalog_agnostic/delete/test_delete_dv_merge_runtime_filter.test b/test/sql/local/catalog_test_config_setup/catalog_agnostic/delete/test_delete_dv_merge_runtime_filter.test index 52c52a6cd..3f6b791b7 100644 --- a/test/sql/local/catalog_test_config_setup/catalog_agnostic/delete/test_delete_dv_merge_runtime_filter.test +++ b/test/sql/local/catalog_test_config_setup/catalog_agnostic/delete/test_delete_dv_merge_runtime_filter.test @@ -53,3 +53,87 @@ SELECT * FROM my_datalake.default.test_dv_merge_runtime_filter ORDER BY id; statement ok DROP TABLE my_datalake.default.test_dv_merge_runtime_filter; + +# Complex and dynamic pushdown must preserve parsed V2 positional-delete metadata. +statement ok +DROP TABLE IF EXISTS my_datalake.default.test_positional_delete_filter_pushdown; + +statement ok +CREATE TABLE my_datalake.default.test_positional_delete_filter_pushdown (id INT, val VARCHAR) +WITH ('format-version'='2'); + +statement ok +INSERT INTO my_datalake.default.test_positional_delete_filter_pushdown VALUES + (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'); + +statement ok +DELETE FROM my_datalake.default.test_positional_delete_filter_pushdown WHERE id = 1; + +# Static predicates go through complex filter pushdown. +statement ok +UPDATE my_datalake.default.test_positional_delete_filter_pushdown SET val = 'b_updated' WHERE id = 2; + +# The join adds dynamic filter pushdown on top of the complex filter state. +statement ok +UPDATE my_datalake.default.test_positional_delete_filter_pushdown SET val = val || '_updated' +FROM (VALUES (3)) t(uid) WHERE id = t.uid; + +query II +SELECT * FROM my_datalake.default.test_positional_delete_filter_pushdown ORDER BY id; +---- +2 b_updated +3 c_updated +4 d + +statement ok +DROP TABLE my_datalake.default.test_positional_delete_filter_pushdown; + +# Partition metadata and deletion vectors must remain available through several +# pushed-down views of the same bind-time list. +statement ok +SET threads=4; + +statement ok +DROP TABLE IF EXISTS my_datalake.default.test_partitioned_dv_filter_pushdown; + +statement ok +CREATE TABLE my_datalake.default.test_partitioned_dv_filter_pushdown (id INT, part INT, val VARCHAR) +PARTITIONED BY (part) WITH ('format-version'='3'); + +# Separate inserts create several manifests while ids 1 and 2 still share a data file. +statement ok +INSERT INTO my_datalake.default.test_partitioned_dv_filter_pushdown VALUES + (1, 10, 'a'), (2, 10, 'b'); + +statement ok +INSERT INTO my_datalake.default.test_partitioned_dv_filter_pushdown VALUES + (3, 20, 'c'), (4, 20, 'd'); + +statement ok +DELETE FROM my_datalake.default.test_partitioned_dv_filter_pushdown WHERE id = 1; + +statement ok +UPDATE my_datalake.default.test_partitioned_dv_filter_pushdown SET val = val || '_dynamic' +FROM (VALUES (2), (3)) t(uid) WHERE id = t.uid; + +# Revisit the first partition through complex filter pushdown and merge its +# replacement deletion vector without rescanning the old one. +statement ok +UPDATE my_datalake.default.test_partitioned_dv_filter_pushdown SET val = val || '_complex' WHERE id = 2; + +# A further MERGE creates another pushed-down generation and must reuse the +# already-processed deletion-vector metadata. +statement ok +MERGE INTO my_datalake.default.test_partitioned_dv_filter_pushdown target +USING (VALUES (3)) source(id) USING (id) +WHEN MATCHED THEN UPDATE SET val = val || '_merge'; + +query III +SELECT * FROM my_datalake.default.test_partitioned_dv_filter_pushdown ORDER BY id; +---- +2 10 b_dynamic_complex +3 20 c_dynamic_merge +4 20 d + +statement ok +DROP TABLE my_datalake.default.test_partitioned_dv_filter_pushdown; diff --git a/test/sql/local/equality_deletes.test b/test/sql/local/equality_deletes.test index 853fcb4f3..0083e13af 100644 --- a/test/sql/local/equality_deletes.test +++ b/test/sql/local/equality_deletes.test @@ -85,6 +85,17 @@ SELECT count(*) FROM ICEBERG_SCAN('__WORKING_DIRECTORY__/data/persistent/equalit ---- 2 +# Equality-delete planning enumerates delete metadata before complex and dynamic filter pushdown. +query II +SELECT scan.id, scan.name +FROM ICEBERG_SCAN('__WORKING_DIRECTORY__/data/persistent/equality_deletes/warehouse/mydb/mytable') scan +JOIN (VALUES (1), (4), (5), (6)) filter_ids(id) USING (id) +WHERE scan.bir >= DATE '2025-01-01' +ORDER BY ALL; +---- +4 d +5 e + # for partitioned table: # create table : mytable_partitioned(id int,name string,bir date ) partitioned by (name); From 3b9d38ff9ac0a8428a2e71685fca41a3fc826734 Mon Sep 17 00:00:00 2001 From: Tishj Date: Sat, 13 Jun 2026 16:50:44 +0200 Subject: [PATCH 7/7] more pushing of state into the shared object --- .../planning/iceberg_multi_file_list.hpp | 24 +-- src/planning/iceberg_multi_file_list.cpp | 137 +++++++++--------- 2 files changed, 83 insertions(+), 78 deletions(-) diff --git a/src/include/planning/iceberg_multi_file_list.hpp b/src/include/planning/iceberg_multi_file_list.hpp index efb599c4b..2e9fb11a5 100644 --- a/src/include/planning/iceberg_multi_file_list.hpp +++ b/src/include/planning/iceberg_multi_file_list.hpp @@ -76,6 +76,9 @@ struct IcebergMultiFileListSharedState { mutable vector> transaction_delete_manifests; mutable unique_ptr delete_manifest_scan; mutable unique_ptr delete_manifest_reader; + mutable bool delete_entries_enumerated = false; + mutable idx_t next_delete_entry_to_process = 0; + mutable vector delete_manifest_entries; //! Scanned data manifests and their owners. mutable vector committed_data_manifests; @@ -86,12 +89,19 @@ struct IcebergMultiFileListSharedState { //! Declared after the manifest owners so references in parsed delete data are destroyed first. mutable case_insensitive_map_t> positional_delete_data; mutable map> equality_delete_data; - mutable unordered_set processed_delete_files; //! Populated as parsed data-file entries become visible to any filtered view. mutable case_insensitive_map_t> data_file_partition_info; }; +struct IcebergDataViewCursor { +public: + idx_t next_batch_idx = 0; + bool has_current_batch = false; + ManifestReadBatch current_batch; + idx_t current_batch_offset = 0; +}; + struct IcebergMultiFileList : public MultiFileList { public: IcebergMultiFileList(ClientContext &context, shared_ptr scan_info, const string &path, @@ -190,16 +200,8 @@ struct IcebergMultiFileList : public MultiFileList { vector types; TableFilterSet table_filters; - mutable idx_t transaction_delete_idx = 0; - mutable bool committed_delete_entries_enumerated = false; - mutable bool initialized = false; - mutable idx_t next_data_batch_idx = 0; - mutable bool has_current_data_batch = false; - mutable ManifestReadBatch current_data_batch; - mutable idx_t current_data_batch_offset = 0; - - //! References to items inside the 'manifest_entries' of the list entries in the 'delete_manifests' - mutable vector delete_manifest_entries; + mutable bool view_initialized = false; + mutable IcebergDataViewCursor data_view_cursor; //! Combination of committed + transaction delete manifests mutable vector delete_manifests; mutable vector delete_manifest_matches; diff --git a/src/planning/iceberg_multi_file_list.cpp b/src/planning/iceberg_multi_file_list.cpp index a242a1086..7f3bd4ad9 100644 --- a/src/planning/iceberg_multi_file_list.cpp +++ b/src/planning/iceberg_multi_file_list.cpp @@ -233,7 +233,7 @@ void IcebergMultiFileList::Bind(vector &return_types, vector(iceberg_path, std::move(temp_data), snapshot_info, *schema); } - if (!initialized) { + if (!view_initialized) { InitializeFiles(guard); } @@ -368,7 +368,7 @@ unique_ptr IcebergMultiFileList::GetCardinality(ClientContext &c //! Make sure we have fetched all manifests (void)GetTotalFileCount(); - D_ASSERT(initialized); + D_ASSERT(view_initialized); idx_t cardinality = 0; for (idx_t i = 0; i < data_manifests.size(); i++) { @@ -733,13 +733,14 @@ bool IcebergMultiFileList::FileMatchesFilter(const IcebergManifestFile &manifest } bool IcebergMultiFileList::TryGetNextBatch(lock_guard &guard) const { - if (has_current_data_batch) { + auto &view_cursor = data_view_cursor; + if (view_cursor.has_current_batch) { return true; } - if (shared_state->read_state.GetBatch(next_data_batch_idx, current_data_batch)) { - next_data_batch_idx++; - current_data_batch_offset = current_data_batch.start_index; - has_current_data_batch = true; + if (shared_state->read_state.GetBatch(view_cursor.next_batch_idx, view_cursor.current_batch)) { + view_cursor.next_batch_idx++; + view_cursor.current_batch_offset = view_cursor.current_batch.start_index; + view_cursor.has_current_batch = true; return true; } if (!shared_state->data_manifest_read_state) { @@ -756,10 +757,10 @@ bool IcebergMultiFileList::TryGetNextBatch(lock_guard &guard) const { auto &token = *task_to_execute->token; scheduler.ScheduleTask(token, std::move(task_to_execute)); } - if (shared_state->read_state.GetBatch(next_data_batch_idx, current_data_batch)) { - next_data_batch_idx++; - current_data_batch_offset = current_data_batch.start_index; - has_current_data_batch = true; + if (shared_state->read_state.GetBatch(view_cursor.next_batch_idx, view_cursor.current_batch)) { + view_cursor.next_batch_idx++; + view_cursor.current_batch_offset = view_cursor.current_batch.start_index; + view_cursor.has_current_batch = true; return true; } //! We didn't manage to populate the buffer with our scan @@ -769,12 +770,12 @@ bool IcebergMultiFileList::TryGetNextBatch(lock_guard &guard) const { executor.WorkOnTasks(); break; } - if (!shared_state->read_state.GetBatch(next_data_batch_idx, current_data_batch)) { + if (!shared_state->read_state.GetBatch(view_cursor.next_batch_idx, view_cursor.current_batch)) { return false; } - next_data_batch_idx++; - current_data_batch_offset = current_data_batch.start_index; - has_current_data_batch = true; + view_cursor.next_batch_idx++; + view_cursor.current_batch_offset = view_cursor.current_batch.start_index; + view_cursor.has_current_batch = true; return true; } @@ -790,7 +791,7 @@ void IcebergMultiFileList::FinishScanTasks(lock_guard &guard) const { optional_ptr IcebergMultiFileList::GetDataFile(idx_t file_id, lock_guard &guard) const { - D_ASSERT(initialized); + D_ASSERT(view_initialized); if (file_id < data_manifest_entries.size()) { //! Have we already scanned this data file and returned it? If so, return it return data_manifest_entries[file_id]; @@ -802,17 +803,18 @@ optional_ptr IcebergMultiFileList::GetDataFile( return nullptr; } - auto ¤t_batch = current_data_batch; + auto &view_cursor = data_view_cursor; + auto ¤t_batch = view_cursor.current_batch; auto &bound_manifest_list_entry = data_manifests[current_batch.manifest_list_entry_idx]; auto &manifest_list_entry = bound_manifest_list_entry.entry; auto &manifest_entries = manifest_list_entry.manifest_entries; auto &manifest_file = manifest_list_entry.file; if (!data_manifest_matches[current_batch.manifest_list_entry_idx]) { - current_data_batch_offset = current_batch.end_index; + view_cursor.current_batch_offset = current_batch.end_index; } - for (; current_data_batch_offset < current_batch.end_index && file_id >= data_manifest_entries.size(); - current_data_batch_offset++) { - auto &manifest_entry = manifest_entries[current_data_batch_offset]; + for (; view_cursor.current_batch_offset < current_batch.end_index && file_id >= data_manifest_entries.size(); + view_cursor.current_batch_offset++) { + auto &manifest_entry = manifest_entries[view_cursor.current_batch_offset]; auto &data_file = manifest_entry.data_file; auto entry_path = data_file.file_path; if (options.allow_moved_paths) { @@ -842,15 +844,15 @@ optional_ptr IcebergMultiFileList::GetDataFile( auto bound_entry = bound_manifest_list_entry.BindEntry(manifest_entry); data_manifest_entries.push_back(bound_entry); } - if (current_data_batch_offset >= current_batch.end_index) { - has_current_data_batch = false; + if (view_cursor.current_batch_offset >= current_batch.end_index) { + view_cursor.has_current_batch = false; } } return data_manifest_entries[file_id]; } OpenFileInfo IcebergMultiFileList::GetFileInternal(idx_t file_id, lock_guard &guard) const { - if (!initialized) { + if (!view_initialized) { InitializeFiles(guard); } @@ -999,7 +1001,7 @@ IcebergMultiFileList::GetEqualityDeletesForFile(const BoundIcebergManifestEntry } void IcebergMultiFileList::InitializeFiles(lock_guard &guard) const { - if (initialized) { + if (view_initialized) { return; } InitializeSharedState(guard); @@ -1029,7 +1031,7 @@ void IcebergMultiFileList::InitializeFiles(lock_guard &guard) const { delete_manifests.emplace_back(delete_manifests.size(), manifest); delete_manifest_matches.push_back(ManifestMatchesFilter(manifest.get().file)); } - initialized = true; + view_initialized = true; } void IcebergMultiFileList::InitializeSharedState(lock_guard &guard) const { @@ -1144,6 +1146,10 @@ void IcebergMultiFileList::InitializeSharedState(lock_guard &guard) const } void IcebergMultiFileList::EnumerateDeleteManifestEntriesInternal() const { + if (shared_state->delete_entries_enumerated) { + return; + } + // In <=v2 we now have to process *all* delete manifests // before we can be certain that we have all the delete data for the current file. @@ -1159,46 +1165,31 @@ void IcebergMultiFileList::EnumerateDeleteManifestEntriesInternal() const { shared_state->delete_manifest_reader->Read(); } - if (!committed_delete_entries_enumerated) { - for (idx_t i = 0; i < shared_state->committed_delete_manifests.size(); i++) { - auto &manifest = delete_manifests[i]; - auto &entries = manifest.entry.manifest_entries; - auto &manifest_file = manifest.entry.file; - if (!delete_manifest_matches[i]) { + for (idx_t i = 0; i < shared_state->committed_delete_manifests.size(); i++) { + auto &manifest_list_entry = shared_state->committed_delete_manifests[i]; + auto manifest = BoundIcebergManifestListEntry(i, manifest_list_entry); + for (auto &manifest_entry : manifest_list_entry.manifest_entries) { + if (manifest_entry.status == IcebergManifestEntryStatusType::DELETED) { continue; } - for (auto &manifest_entry : entries) { - if (manifest_entry.status == IcebergManifestEntryStatusType::DELETED) { - continue; - } - auto &data_file = manifest_entry.data_file; - if (!table_filters.filters.empty() && - !FileMatchesFilter(manifest_file, manifest_entry, IcebergManifestContentType::DELETE)) { - DUCKDB_LOG(context, IcebergLogType, "Iceberg Filter Pushdown, skipped 'data_file': '%s'", - data_file.file_path); - continue; - } - auto &referenced_data_file = data_file.referenced_data_file; - if (!referenced_data_file.empty() && transactional_delete_files && - transactional_delete_files->count(referenced_data_file)) { - //! Skip this delete file, there's a transaction-local delete that makes it obsolete - continue; - } - auto bound_entry = manifest.BindEntry(manifest_entry); - delete_manifest_entries.push_back(std::move(bound_entry)); + auto &data_file = manifest_entry.data_file; + auto &referenced_data_file = data_file.referenced_data_file; + if (!referenced_data_file.empty() && transactional_delete_files && + transactional_delete_files->count(referenced_data_file)) { + //! Skip this delete file, there's a transaction-local delete that makes it obsolete + continue; } + auto bound_entry = manifest.BindEntry(manifest_entry); + shared_state->delete_manifest_entries.push_back(std::move(bound_entry)); } - committed_delete_entries_enumerated = true; } auto offset = shared_state->committed_delete_manifests.size(); - while (transaction_delete_idx < shared_state->transaction_delete_manifests.size()) { - auto &delete_manifest = delete_manifests[offset + transaction_delete_idx]; - if (!delete_manifest_matches[offset + transaction_delete_idx]) { - transaction_delete_idx++; - continue; - } - for (auto &manifest_entry : delete_manifest.entry.manifest_entries) { + for (idx_t transaction_delete_idx = 0; transaction_delete_idx < shared_state->transaction_delete_manifests.size(); + transaction_delete_idx++) { + auto &manifest_list_entry = shared_state->transaction_delete_manifests[transaction_delete_idx].get(); + auto delete_manifest = BoundIcebergManifestListEntry(offset + transaction_delete_idx, manifest_list_entry); + for (auto &manifest_entry : manifest_list_entry.manifest_entries) { auto &data_file = manifest_entry.data_file; auto &referenced_data_file = data_file.referenced_data_file; if (!referenced_data_file.empty() && transactional_delete_files) { @@ -1210,23 +1201,22 @@ void IcebergMultiFileList::EnumerateDeleteManifestEntriesInternal() const { } //! FIXME: no file pruning for uncommitted data? auto bound_manifest_entry = delete_manifest.BindEntry(manifest_entry); - delete_manifest_entries.push_back(std::move(bound_manifest_entry)); + shared_state->delete_manifest_entries.push_back(std::move(bound_manifest_entry)); } - transaction_delete_idx++; } + shared_state->delete_entries_enumerated = true; D_ASSERT(FinishedScanningDeletes()); } void IcebergMultiFileList::ScanDeleteFiles(const vector &global_columns, const vector &global_column_ids, const vector &projection_ids) const { - for (auto &bound_manifest_entry : delete_manifest_entries) { + for (; shared_state->next_delete_entry_to_process < shared_state->delete_manifest_entries.size(); + shared_state->next_delete_entry_to_process++) { + auto &bound_manifest_entry = shared_state->delete_manifest_entries[shared_state->next_delete_entry_to_process]; auto &manifest_entry = bound_manifest_entry.entry; auto &data_file = manifest_entry.data_file; - if (shared_state->processed_delete_files.count(data_file.file_path)) { - continue; - } if (StringUtil::CIEquals(data_file.file_format, "parquet")) { ScanDeleteFile(bound_manifest_entry, global_columns, global_column_ids, projection_ids); } else if (StringUtil::CIEquals(data_file.file_format, "puffin")) { @@ -1236,7 +1226,6 @@ void IcebergMultiFileList::ScanDeleteFiles(const vectorprocessed_delete_files.insert(data_file.file_path); } } @@ -1264,7 +1253,21 @@ vector IcebergMultiFileList::GetDeleteManifestEntries InitializeFiles(guard); lock_guard delete_guard(shared_state->delete_lock); EnumerateDeleteManifestEntriesInternal(); - return delete_manifest_entries; + vector result; + for (auto &entry : shared_state->delete_manifest_entries) { + auto manifest_idx = entry.manifest_file_idx; + auto &manifest = delete_manifests[manifest_idx]; + auto &manifest_file = manifest.entry.file; + if (!delete_manifest_matches[manifest_idx]) { + continue; + } + if (!table_filters.filters.empty() && + !FileMatchesFilter(manifest_file, entry.entry, IcebergManifestContentType::DELETE)) { + continue; + } + result.push_back(entry); + } + return result; } void IcebergMultiFileList::ScanDeleteFile(const BoundIcebergManifestEntry &bound_manifest_entry,