chore: merge upstream v1.5-variegata for DuckDB v1.5.1

src/common/CMakeLists.txt

@@ -1,7 +1,7 @@
 add_library(
   ducklake_common OBJECT
-  ducklake_data_file.cpp ducklake_name_map.cpp ducklake_types.cpp
-  ducklake_util.cpp parquet_file_scanner.cpp)
+  ducklake_data_file.cpp ducklake_name_map.cpp ducklake_snapshot.cpp
+  ducklake_types.cpp ducklake_util.cpp parquet_file_scanner.cpp)
 set(ALL_OBJECT_FILES
     ${ALL_OBJECT_FILES} $<TARGET_OBJECTS:ducklake_common>
     PARENT_SCOPE)

src/common/ducklake_snapshot.cpp

@@ -0,0 +1,23 @@
+#include "common/ducklake_snapshot.hpp"
+#include "duckdb/common/serializer/serializer.hpp"
+#include "duckdb/common/serializer/deserializer.hpp"
+
+namespace duckdb {
+
+void DuckLakeSnapshot::Serialize(Serializer &serializer) const {
+	serializer.WriteProperty(100, "snapshot_id", snapshot_id);
+	serializer.WriteProperty(101, "schema_version", schema_version);
+	serializer.WriteProperty(102, "next_catalog_id", next_catalog_id);
+	serializer.WriteProperty(103, "next_file_id", next_file_id);
+}
+
+DuckLakeSnapshot DuckLakeSnapshot::Deserialize(Deserializer &deserializer) {
+	DuckLakeSnapshot result;
+	result.snapshot_id = deserializer.ReadProperty<idx_t>(100, "snapshot_id");
+	result.schema_version = deserializer.ReadProperty<idx_t>(101, "schema_version");
+	result.next_catalog_id = deserializer.ReadProperty<idx_t>(102, "next_catalog_id");
+	result.next_file_id = deserializer.ReadProperty<idx_t>(103, "next_file_id");
+	return result;
+}
+
+} // namespace duckdb

src/ducklake_extension.cpp

@@ -3,6 +3,7 @@
 #include "duckdb/common/exception.hpp"
 #include "duckdb/common/string_util.hpp"
 #include "storage/ducklake_storage.hpp"
+#include "storage/ducklake_scan.hpp"
 #include "functions/ducklake_table_functions.hpp"
 #include "storage/ducklake_secret.hpp"
 #include "duckdb/storage/storage_extension.hpp"
@@ -86,6 +87,10 @@ static void LoadInternal(ExtensionLoader &loader) {
 	DuckLakeSettingsFunction settings;
 	loader.RegisterFunction(settings);
 
+	// Register ducklake_scan so it can be found during deserialization
+	auto ducklake_scan = DuckLakeFunctions::GetDuckLakeScanFunction(loader.GetDatabaseInstance());
+	loader.RegisterFunction(ducklake_scan);
+
 	// secrets
 	auto secret_type = DuckLakeSecret::GetSecretType();
 	loader.RegisterSecretType(secret_type);

src/functions/ducklake_add_data_files.cpp

@@ -130,9 +130,8 @@ struct DuckLakeFileProcessor {
 public:
 	DuckLakeFileProcessor(DuckLakeTransaction &transaction, ClientContext &context,
 	                      const DuckLakeAddDataFilesData &bind_data)
-	    : transaction(transaction), context(context), table(bind_data.table),
-	      allow_missing(bind_data.allow_missing), ignore_extra_columns(bind_data.ignore_extra_columns),
-	      hive_partitioning(bind_data.hive_partitioning) {
+	    : transaction(transaction), context(context), table(bind_data.table), allow_missing(bind_data.allow_missing),
+	      ignore_extra_columns(bind_data.ignore_extra_columns), hive_partitioning(bind_data.hive_partitioning) {
 	}
 
 	vector<DuckLakeDataFile> AddFiles(const vector<string> &globs);
@@ -1167,7 +1166,8 @@ void DuckLakeFileProcessor::MapPartitionColumns(ParquetFileMetadata &file) {
 		// For YEAR/MONTH/DAY/HOUR transforms, the type is BIGINT, not the source column type
 		auto partition_key_type =
 		    DuckLakePartitionUtils::GetPartitionKeyType(partition_field.transform.type, field_id->Type());
-		auto hive_value = HivePartitioning::GetValue(context, partition_key_name, hive_entry->second, partition_key_type);
+		auto hive_value =
+		    HivePartitioning::GetValue(context, partition_key_name, hive_entry->second, partition_key_type);
 		file.hive_partition_values.emplace_back(
 		    HivePartition {partition_field.field_id, partition_key_type, hive_value});
 	}

src/functions/ducklake_compaction_functions.cpp

@@ -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);
@@ -222,8 +233,9 @@ void DuckLakeCompactor::GenerateCompactions(DuckLakeTableEntry &table,
 	compaction_map_t<DuckLakeCompactionCandidates> candidates;
 	for (idx_t file_idx = 0; file_idx < files.size(); file_idx++) {
 		auto &candidate = files[file_idx];
-		if (candidate.file.data.file_size_bytes >= target_file_size) {
+		if (candidate.file.data.file_size_bytes >= target_file_size && type != CompactionType::REWRITE_DELETES) {
 			// this file by itself exceeds the threshold - skip merging
+			// (does not apply to REWRITE_DELETES - delete files must be rewritten regardless of data file size)
 			continue;
 		}
 		if ((!candidate.delete_files.empty() && type == CompactionType::MERGE_ADJACENT_TABLES) ||
@@ -304,6 +316,9 @@ void DuckLakeCompactor::GenerateCompactions(DuckLakeTableEntry &table,
 				break;
 			}
 		}
+		if (compacted_files >= options.max_files) {
+			break;
+		}
 	}
 }
 
@@ -559,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_flush_inlined_data.cpp

@@ -278,8 +278,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/include/common/ducklake_snapshot.hpp

@@ -12,6 +12,9 @@
 
 namespace duckdb {
 
+class Serializer;
+class Deserializer;
+
 struct DuckLakeSnapshot {
 	DuckLakeSnapshot(idx_t snapshot_id, idx_t schema_version, idx_t next_catalog_id, idx_t next_file_id)
 	    : snapshot_id(snapshot_id), schema_version(schema_version), next_catalog_id(next_catalog_id),
@@ -27,6 +30,9 @@ struct DuckLakeSnapshot {
 	idx_t schema_version;
 	idx_t next_catalog_id;
 	idx_t next_file_id;
+
+	void Serialize(Serializer &serializer) const;
+	static DuckLakeSnapshot Deserialize(Deserializer &deserializer);
 };
 
 } // namespace duckdb

src/include/common/index.hpp

@@ -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/storage/ducklake_catalog.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_inlined_data.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_metadata_manager.hpp

@@ -159,6 +159,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);
@@ -259,6 +260,7 @@ class DuckLakeMetadataManager {
 
 	string LoadPath(string path);
 	string StorePath(string path);
+	string GetPathSeparator(const string &path);
 
 protected:
 	virtual string GetLatestSnapshotQuery() const;

src/include/storage/ducklake_multi_file_list.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";
 
@@ -63,6 +62,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;
 
 private:
 	mutable mutex file_lock;

src/include/storage/ducklake_scan.hpp

@@ -17,6 +17,8 @@ namespace duckdb {
 class DuckLakeMultiFileList;
 class DuckLakeTableEntry;
 class DuckLakeTransaction;
+class Serializer;
+class Deserializer;
 
 class DuckLakeFunctions {
 public:
@@ -28,11 +30,19 @@ class DuckLakeFunctions {
 	static CopyFunctionCatalogEntry &GetCopyFunction(ClientContext &context, const string &name);
 };
 
+//! Serialize/Deserialize callbacks for DuckLakeScan (used by table macro Copy)
+void DuckLakeScanSerialize(Serializer &serializer, const optional_ptr<FunctionData> bind_data,
+                           const TableFunction &function);
+unique_ptr<FunctionData> DuckLakeScanDeserialize(Deserializer &deserializer, TableFunction &function);
+
 enum class DuckLakeScanType { SCAN_TABLE, SCAN_INSERTIONS, SCAN_DELETIONS, SCAN_FOR_FLUSH };
 
 struct DuckLakeFunctionInfo : public TableFunctionInfo {
 	DuckLakeFunctionInfo(DuckLakeTableEntry &table, DuckLakeTransaction &transaction, DuckLakeSnapshot snapshot);
 
+	static shared_ptr<DuckLakeFunctionInfo> Create(DuckLakeTableEntry &table, DuckLakeTransaction &transaction,
+	                                               DuckLakeSnapshot snapshot);
+
 	DuckLakeTableEntry &table;
 	weak_ptr<DuckLakeTransaction> transaction;
 	string table_name;

src/include/storage/ducklake_stats.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;