[Dev] Create shared state for IcebergMultiFileList

.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' }}

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<IcebergMultiFileList>();
-	ic_file_list.table = this;
+	ic_file_list.SetTable(this);
 	return iceberg_scan_function;
 }
 

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);
 }
 

src/core/deletes/iceberg_equality_delete.cpp

@@ -65,9 +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<IcebergEqualityDeleteData>(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<IcebergEqualityDeleteData>(sequence_number))
+		         .first;
 	}
 	auto &deletes = *it->second;
 

src/core/deletes/iceberg_positional_delete.cpp

@@ -38,7 +38,7 @@ void IcebergMultiFileList::ScanPositionalDeleteFile(const BoundIcebergManifestEn
 	}
 	reference<string_t> 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;

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);
 			}

src/function/scan/iceberg_scan.cpp

@@ -59,10 +59,11 @@ static unique_ptr<FunctionData> IcebergScanDeserialize(Deserializer &deserialize
 BindInfo IcebergBindInfo(const optional_ptr<FunctionData> bind_data) {
 	auto &multi_file_data = bind_data->Cast<MultiFileBindData>();
 	auto &file_list = multi_file_data.file_list->Cast<IcebergMultiFileList>();
-	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

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<ManifestReadBatch> 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<ManifestReadBatch> batches;
+	mutable mutex lock;
+	vector<ManifestReadBatch> batches;
 };
 
 } // namespace duckdb

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<AvroScan> scan,
@@ -46,6 +49,59 @@ struct IcebergManifestScanningState {
 	atomic<idx_t> in_progress_tasks;
 };
 
+struct IcebergMultiFileListSharedState {
+public:
+	IcebergMultiFileListSharedState(ClientContext &context, shared_ptr<IcebergScanInfo> scan_info, string path,
+	                                const IcebergOptions &options);
+	~IcebergMultiFileListSharedState();
+
+private:
+	friend struct IcebergMultiFileList;
+
+	ClientContext &context;
+	FileSystem &fs;
+	shared_ptr<IcebergScanInfo> 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<IcebergManifestListEntry> committed_delete_manifests;
+	mutable vector<reference<const IcebergManifestListEntry>> transaction_delete_manifests;
+	mutable unique_ptr<AvroScan> delete_manifest_scan;
+	mutable unique_ptr<manifest_file::ManifestReader> delete_manifest_reader;
+	mutable bool delete_entries_enumerated = false;
+	mutable idx_t next_delete_entry_to_process = 0;
+	mutable vector<BoundIcebergManifestEntry> delete_manifest_entries;
+
+	//! Scanned data manifests and their owners.
+	mutable vector<IcebergManifestListEntry> committed_data_manifests;
+	mutable vector<reference<const IcebergManifestListEntry>> transaction_data_manifests;
+	mutable unique_ptr<IcebergManifestScanningState> data_manifest_read_state;
+	mutable unique_ptr<manifest_file::ManifestReader> data_manifest_reader;
+
+	//! Declared after the manifest owners so references in parsed delete data are destroyed first.
+	mutable case_insensitive_map_t<shared_ptr<IcebergDeleteData>> positional_delete_data;
+	mutable map<sequence_number_t, unique_ptr<IcebergEqualityDeleteData>> equality_delete_data;
+
+	//! Populated as parsed data-file entries become visible to any filtered view.
+	mutable case_insensitive_map_t<vector<IcebergPartitionInfo>> 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<IcebergScanInfo> scan_info, const string &path,
@@ -60,32 +116,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<LogicalType> &return_types, vector<string> &names);
 	unique_ptr<IcebergMultiFileList> 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<MultiFileColumnDefinition> &columns,
-	                            const vector<MultiFileColumnDefinition> &global_columns,
-	                            const vector<ColumnIndex> &global_column_ids,
-	                            const vector<idx_t> &projection_ids) const;
-	void ScanDeleteFile(const BoundIcebergManifestEntry &entry, const vector<MultiFileColumnDefinition> &global_columns,
-	                    const vector<ColumnIndex> &global_column_ids, const vector<idx_t> &projection_ids) const;
-	void ScanPuffinFile(const BoundIcebergManifestEntry &entry) const;
 	unique_ptr<DeleteFilter> GetPositionalDeletesForFile(const string &file_path) const;
-	void EnumerateDeleteManifestEntries() const;
 	void ProcessDeletes(const vector<MultiFileColumnDefinition> &global_columns,
 	                    const vector<ColumnIndex> &global_column_ids, const vector<idx_t> &projection_ids) const;
-	void ScanDeleteFiles(const vector<MultiFileColumnDefinition> &global_columns,
-	                     const vector<ColumnIndex> &global_column_ids, const vector<idx_t> &projection_ids) const;
 	vector<reference<const IcebergEqualityDeleteRow>>
 	GetEqualityDeletesForFile(const BoundIcebergManifestEntry &manifest_entry) const;
 	void GetStatistics(vector<PartitionStatistics> &result) const;
-	const BoundIcebergManifestEntry &GetManifestEntry(idx_t file_id) const;
+	BoundIcebergManifestEntry GetManifestEntry(idx_t file_id) const;
 	vector<IcebergPartitionInfo> GetPartitionInfoForDataFile(const string &file_path) const;
 	const IcebergManifestFile &GetManifestFileForEntry(const BoundIcebergManifestEntry &entry,
 	                                                   IcebergManifestContentType type) const;
+	vector<BoundIcebergManifestEntry> GetDeleteManifestEntries() const;
+	shared_ptr<IcebergDeleteData> GetExistingPositionalDeleteData(const string &file_path) const;
 
 public:
 	//! MultiFileList API
@@ -120,63 +168,50 @@ struct IcebergMultiFileList : public MultiFileList {
 	                                                        const ColumnIndex &column_index) const;
 
 private:
-	optional_ptr<ManifestReadBatch> TryGetNextBatch(lock_guard<mutex> &guard) const;
+	IcebergMultiFileList(shared_ptr<IcebergMultiFileListSharedState> shared_state);
+	bool TryGetNextBatch(lock_guard<mutex> &guard) const;
 	void FinishScanTasks(lock_guard<mutex> &guard) const;
+	void InitializeSharedState(lock_guard<mutex> &guard) const;
+	bool FinishedScanningDeletes() const;
+	void EnumerateDeleteManifestEntriesInternal() const;
+	void ProcessDeletesInternal(const vector<MultiFileColumnDefinition> &global_columns,
+	                            const vector<ColumnIndex> &global_column_ids,
+	                            const vector<idx_t> &projection_ids) const;
+	void ScanDeleteFiles(const vector<MultiFileColumnDefinition> &global_columns,
+	                     const vector<ColumnIndex> &global_column_ids, const vector<idx_t> &projection_ids) const;
+	void ScanDeleteFile(const BoundIcebergManifestEntry &entry, const vector<MultiFileColumnDefinition> &global_columns,
+	                    const vector<ColumnIndex> &global_column_ids, const vector<idx_t> &projection_ids) const;
+	void ScanPositionalDeleteFile(const BoundIcebergManifestEntry &manifest_entry, DataChunk &result) const;
+	void ScanEqualityDeleteFile(const BoundIcebergManifestEntry &manifest_entry, DataChunk &result,
+	                            vector<MultiFileColumnDefinition> &columns,
+	                            const vector<MultiFileColumnDefinition> &global_columns,
+	                            const vector<ColumnIndex> &global_column_ids,
+	                            const vector<idx_t> &projection_ids) const;
+	void ScanPuffinFile(const BoundIcebergManifestEntry &entry) const;
 
-public:
+private:
+	shared_ptr<IcebergMultiFileListSharedState> shared_state;
 	ClientContext &context;
 	FileSystem &fs;
-	shared_ptr<IcebergScanInfo> scan_info;
-	string path;
-	IcebergTableEntry *table;
-
-	mutable mutex lock;
+	const IcebergOptions &options;
 	//! ComplexFilterPushdown results
 	bool have_bound = false;
 	vector<string> names;
 	vector<LogicalType> 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)
-	mutable case_insensitive_map_t<shared_ptr<IcebergDeleteData>> positional_delete_data;
-	//! All equality deletes with sequence numbers higher than that of the data_file apply to that data_file
-	mutable map<sequence_number_t, unique_ptr<IcebergEqualityDeleteData>> 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;
-
-public:
-	//! References to items inside the 'manifest_entries' of the list entries in the 'delete_manifests'
-	mutable vector<BoundIcebergManifestEntry> delete_manifest_entries;
+	mutable bool view_initialized = false;
+	mutable IcebergDataViewCursor data_view_cursor;
 	//! Combination of committed + transaction delete manifests
 	mutable vector<BoundIcebergManifestListEntry> delete_manifests;
-	//! Scanned delete manifests of the snapshot being scanned
-	mutable unique_ptr<AvroScan> delete_manifest_scan;
-	mutable unique_ptr<manifest_file::ManifestReader> delete_manifest_reader;
-	mutable vector<IcebergManifestListEntry> committed_delete_manifests;
-	//! Cached, uncommitted delete manifests created by earlier statements in the transaction
-	mutable vector<reference<const IcebergManifestListEntry>> transaction_delete_manifests;
+	mutable vector<bool> delete_manifest_matches;
 
 private:
 	//! References to items inside the 'manifest_entries' of the list entries in the 'data_manifests'
 	mutable vector<BoundIcebergManifestEntry> data_manifest_entries;
 	//! Combination of committed + transaction data manifests
 	mutable vector<BoundIcebergManifestListEntry> data_manifests;
-	//! Scanned data manifests of the snapshot being scanned
-	mutable unique_ptr<IcebergManifestScanningState> data_manifest_read_state;
-	mutable unique_ptr<manifest_file::ManifestReader> data_manifest_reader;
-	mutable vector<IcebergManifestListEntry> committed_data_manifests;
-	//! Cached, uncommitted data manifests created by earlier statements in the transaction
-	mutable vector<reference<const IcebergManifestListEntry>> transaction_data_manifests;
+	mutable vector<bool> data_manifest_matches;
 };
 
 } // namespace duckdb