Merge pull request duckdb#879 from pdet/merge

Merge

Author: pdet

duckdb

@@ -93,11 +93,17 @@ SourceResultType DuckLakeCompaction::GetDataInternal(ExecutionContext &context,
 	}
 	source_state.returned_result = true;
 
+	if (!this->sink_state) {
+		throw InternalException("DuckLakeCompaction - missing sink state while producing result");
+	}
+	auto &gstate = this->sink_state->Cast<DuckLakeInsertGlobalState>();
+	auto files_created = gstate.written_files.size();
+
 	chunk.SetCardinality(1);
 	chunk.SetValue(0, 0, Value(table.schema.name));
 	chunk.SetValue(1, 0, Value(table.name));
 	chunk.SetValue(2, 0, Value::BIGINT(static_cast<int64_t>(source_files.size())));
-	chunk.SetValue(3, 0, Value::BIGINT(1)); // Each compaction creates 1 output file
+	chunk.SetValue(3, 0, Value::BIGINT(static_cast<int64_t>(files_created)));
 	return SourceResultType::FINISHED;
 }
 
@@ -121,7 +127,10 @@ SinkFinalizeType DuckLakeCompaction::Finalize(Pipeline &pipeline, Event &event,
                                               OperatorSinkFinalizeInput &input) const {
 	auto &global_state = input.global_state.Cast<DuckLakeInsertGlobalState>();
 
-	if (global_state.written_files.size() != 1) {
+	if (global_state.written_files.size() > 1) {
+		throw InternalException("DuckLakeCompaction - expected at most a single output file");
+	}
+	if (global_state.written_files.empty() && type != CompactionType::REWRITE_DELETES) {
 		throw InternalException("DuckLakeCompaction - expected a single output file");
 	}
 	// set the partition values correctly
@@ -137,7 +146,9 @@ SinkFinalizeType DuckLakeCompaction::Finalize(Pipeline &pipeline, Event &event,
 	DuckLakeCompactionEntry compaction_entry;
 	compaction_entry.row_id_start = row_id_start;
 	compaction_entry.source_files = source_files;
-	compaction_entry.written_file = global_state.written_files[0];
+	if (!global_state.written_files.empty()) {
+		compaction_entry.written_file = global_state.written_files[0];
+	}
 	compaction_entry.type = type;
 
 	auto &transaction = DuckLakeTransaction::Get(context, global_state.table.catalog);
@@ -305,6 +316,9 @@ void DuckLakeCompactor::GenerateCompactions(DuckLakeTableEntry &table,
 				break;
 			}
 		}
+		if (compacted_files >= options.max_files) {
+			break;
+		}
 	}
 }
 
@@ -560,7 +574,7 @@ DuckLakeCompactor::GenerateCompactionCommand(vector<DuckLakeCompactionFileEntry>
 	copy->filename_pattern = std::move(copy_options.filename_pattern);
 	copy->file_extension = std::move(copy_options.file_extension);
 	copy->overwrite_mode = copy_options.overwrite_mode;
-	copy->per_thread_output = copy_options.per_thread_output;
+	copy->per_thread_output = false;
 	copy->file_size_bytes = copy_options.file_size_bytes;
 	copy->rotate = copy_options.rotate;
 	copy->return_type = copy_options.return_type;

src/functions/ducklake_compaction_functions.cpp

@@ -277,8 +277,9 @@ DuckLakeDataFlusher::DuckLakeDataFlusher(ClientContext &context, DuckLakeCatalog
 
 unique_ptr<LogicalOperator> DuckLakeDataFlusher::GenerateFlushCommand() {
 	// get the table entry at the specified snapshot
-	DuckLakeSnapshot snapshot(catalog.GetBeginSnapshotForTable(table_id, transaction), inlined_table.schema_version, 0,
-	                          0);
+	DuckLakeSnapshot snapshot(
+	    catalog.GetBeginSnapshotForSchemaVersion(table_id, inlined_table.schema_version, transaction),
+	    inlined_table.schema_version, 0, 0);
 
 	auto entry = catalog.GetEntryById(transaction, snapshot, table_id);
 	if (!entry) {

src/functions/ducklake_flush_inlined_data.cpp

@@ -15,6 +15,11 @@ namespace duckdb {
 
 struct DuckLakeConstants {
 	static constexpr const idx_t TRANSACTION_LOCAL_ID_START = 9223372036854775808ULL;
+	static constexpr const idx_t TRANSACTION_LOCAL_ROW_ID_START = 1000000000000000000ULL;
+
+	static bool IsTransactionLocalRowId(int64_t rid) {
+		return rid >= 0 && static_cast<idx_t>(rid) >= TRANSACTION_LOCAL_ROW_ID_START;
+	}
 };
 
 struct SchemaIndex {

src/include/common/index.hpp

@@ -27,6 +27,8 @@ class PostgresMetadataManager : public DuckLakeMetadataManager {
 	}
 
 	string GetColumnTypeInternal(const LogicalType &type) override;
+	shared_ptr<DuckLakeInlinedData> TransformInlinedData(QueryResult &result,
+	                                                     const vector<LogicalType> &expected_types) override;
 
 	unique_ptr<QueryResult> Execute(DuckLakeSnapshot snapshot, string &query) override;
 

src/include/metadata_manager/postgres_metadata_manager.hpp

@@ -25,6 +25,8 @@ struct DuckLakeConfigOption;
 struct DeleteFileMap;
 class LogicalGet;
 
+enum class InlinedDeletionCacheResult { EXISTS, DOES_NOT_EXIST, UNKNOWN };
+
 class DuckLakeCatalog : public Catalog {
 public:
 	// default target file size: 512MB
@@ -57,6 +59,8 @@ class DuckLakeCatalog : public Catalog {
 		return metadata_type;
 	}
 	idx_t DataInliningRowLimit(SchemaIndex schema_index, TableIndex table_index) const;
+	//! Returns the inlining limit (0 if the table is not eligible)
+	idx_t GetInliningLimit(ClientContext &context, DuckLakeTableEntry &table, const vector<LogicalType> &types);
 	string &Separator() {
 		return separator;
 	}
@@ -164,12 +168,18 @@ class DuckLakeCatalog : public Catalog {
 	optional_ptr<const DuckLakeNameMap> TryGetMappingById(DuckLakeTransaction &transaction, MappingIndex mapping_id);
 	MappingIndex TryGetCompatibleNameMap(DuckLakeTransaction &transaction, const DuckLakeNameMap &name_map);
 	idx_t GetBeginSnapshotForTable(TableIndex table_id, DuckLakeTransaction &transaction);
+	idx_t GetBeginSnapshotForSchemaVersion(TableIndex table_id, idx_t schema_version, DuckLakeTransaction &transaction);
 
 	static unique_ptr<DuckLakeStats> ConstructStatsMap(vector<DuckLakeGlobalStatsInfo> &global_stats,
 	                                                   DuckLakeCatalogSet &schema);
 	//! Return the schema for the given snapshot - loading it if it is not yet loaded
 	DuckLakeCatalogSet &GetSchemaForSnapshot(DuckLakeTransaction &transaction, DuckLakeSnapshot snapshot);
 
+	//! Check if an inlined deletion table is known to exist or not exist for the given table and snapshot
+	InlinedDeletionCacheResult CheckInlinedDeletionTableCache(TableIndex table_id, DuckLakeSnapshot snapshot);
+	//! Cache the result of an inlined deletion table existence check
+	void CacheInlinedDeletionTableResult(TableIndex table_id, DuckLakeSnapshot snapshot, bool exists);
+
 private:
 	void DropSchema(ClientContext &context, DropInfo &info) override;
 	unique_ptr<DuckLakeCatalogSet> LoadSchemaForSnapshot(DuckLakeTransaction &transaction, DuckLakeSnapshot snapshot);
@@ -200,6 +210,13 @@ class DuckLakeCatalog : public Catalog {
 	string metadata_type;
 	//! Whether or not the catalog is initialized
 	bool initialized = false;
+	//! Cache for inlined deletion table existence checks
+	mutex inlined_deletion_cache_lock;
+	//! Table IDs where the inlined deletion table is known to exist (permanent - never invalidated)
+	unordered_set<idx_t> inlined_deletion_exists;
+	//! Table IDs where the inlined deletion table is known to NOT exist, with the snapshot_id at which we checked
+	//! Valid as long as current snapshot.snapshot_id <= cached snapshot_id
+	unordered_map<idx_t, idx_t> inlined_deletion_not_exists;
 	//! The id of the last committed snapshot, set at FlushChanges on a successful commit
 	mutable mutex commit_lock;
 	optional_idx last_committed_snapshot;

src/include/storage/ducklake_catalog.hpp

@@ -17,6 +17,16 @@ namespace duckdb {
 struct DuckLakeInlinedData {
 	unique_ptr<ColumnDataCollection> data;
 	map<FieldIndex, DuckLakeColumnStats> column_stats;
+	//! Row Ids for update inlining
+	vector<int64_t> row_ids;
+
+	bool HasPreservedRowIds() const;
+	//! Get the row_id for a given position in the data collection
+	idx_t GetRowId(idx_t position) const;
+	//! Get the output row_id for a surviving (non-deleted) row
+	int64_t GetOutputRowId(idx_t position) const;
+	//! Merge preserved row_ids from update inlining
+	void MergeRowIds(const DuckLakeInlinedData &new_data, idx_t new_data_count);
 };
 
 struct DuckLakeInlinedDataDeletes {

src/include/storage/ducklake_inlined_data.hpp

@@ -145,6 +145,7 @@ class DuckLakeMetadataManager {
 	                                                                  DuckLakeSnapshot snapshot,
 	                                                                  DuckLakeFileSizeOptions options);
 	virtual idx_t GetBeginSnapshotForTable(TableIndex table_id);
+	virtual idx_t GetBeginSnapshotForSchemaVersion(TableIndex table_id, idx_t schema_version);
 	virtual idx_t GetNetDataFileRowCount(TableIndex table_id, DuckLakeSnapshot snapshot);
 	virtual idx_t GetNetInlinedRowCount(const string &inlined_table_name, DuckLakeSnapshot snapshot);
 	virtual vector<DuckLakeFileForCleanup> GetOldFilesForCleanup(const string &filter);
@@ -243,6 +244,7 @@ class DuckLakeMetadataManager {
 
 	string LoadPath(string path);
 	string StorePath(string path);
+	string GetPathSeparator(const string &path);
 
 protected:
 	virtual string GetLatestSnapshotQuery() const;
@@ -311,9 +313,10 @@ class DuckLakeMetadataManager {
 public:
 	//! Read inlined file deletions for regular table scans (no snapshot info per row)
 	map<idx_t, set<idx_t>> ReadInlinedFileDeletions(TableIndex table_id, DuckLakeSnapshot snapshot);
+	//! Clear inlined table caches (needed after rollback so retry re-creates the tables)
+	void ClearInlinedTableCaches();
 
 private:
-	unordered_map<idx_t, string> inlined_table_name_cache;
 	static unordered_map<string /* name */, create_t> metadata_managers;
 	static mutex metadata_managers_lock;
 

src/include/storage/ducklake_metadata_manager.hpp

@@ -20,7 +20,6 @@ namespace duckdb {
 //! The DuckLakeMultiFileList implements the MultiFileList API to allow injecting it into the regular DuckDB parquet
 //! scan
 class DuckLakeMultiFileList : public MultiFileList {
-	static constexpr const idx_t TRANSACTION_LOCAL_ID_START = 1000000000000000000ULL;
 	static constexpr const char *DUCKLAKE_TRANSACTION_LOCAL_INLINED_FILENAME =
 	    "__ducklake_inlined_transaction_local_data";
 
@@ -67,6 +66,8 @@ class DuckLakeMultiFileList : public MultiFileList {
 	void GetFilesForTable() const;
 	void GetTableInsertions() const;
 	void GetTableDeletions() const;
+	//! Get the row_id_start for transaction-local inlined data.
+	idx_t GetTransactionLocalRowIdStart(idx_t transaction_row_start) const;
 	void AddFilterToPushdownInfo(FilterPushdownInfo &pushdown_info, column_t column_id, unique_ptr<TableFilter> filter) const;
 
 private:

src/include/storage/ducklake_multi_file_list.hpp

@@ -15,7 +15,7 @@ class BaseStatistics;
 
 //! Returns true for types that require value-based (not lexicographic string) comparison for min/max stats
 inline bool RequiresValueComparison(const LogicalType &type) {
-	return type.IsNumeric() || type.IsTemporal();
+	return type.IsNumeric() || type.IsTemporal() || type.id() == LogicalTypeId::BOOLEAN;
 }
 
 struct DuckLakeColumnStats;