Merge pull request #599 from pdet/merge

pdet · web-flow · commit c4f8102e61cf · 2025-12-05T17:36:11.000-03:00
V1.4 -&gt; Main
diff --git a/duckdb b/duckdb
@@ -1 +1 @@
-Subproject commit a8ded270169dc9bedf51a3d5e9e90a3be172919e
+Subproject commit 5d5d04f418e995f419ca8ca75fe9d42964d06d4b
diff --git a/src/functions/ducklake_cleanup_files.cpp b/src/functions/ducklake_cleanup_files.cpp
@@ -64,9 +64,10 @@ static unique_ptr<FunctionData> CleanupBind(ClientContext &context, TableFunctio
 	bool cleanup_all = false;
 	for (auto &entry : input.named_parameters) {
 		if (StringUtil::CIEquals(entry.first, "dry_run")) {
-			result->dry_run = true;
+			result->dry_run = entry.second.GetValue<bool>();
+			;
 		} else if (StringUtil::CIEquals(entry.first, "cleanup_all")) {
-			cleanup_all = true;
+			cleanup_all = entry.second.GetValue<bool>();
 		} else if (StringUtil::CIEquals(entry.first, "older_than")) {
 			from_timestamp = entry.second.GetValue<timestamp_tz_t>();
 			has_timestamp = true;
diff --git a/src/functions/ducklake_compaction_functions.cpp b/src/functions/ducklake_compaction_functions.cpp
@@ -141,7 +141,7 @@ class DuckLakeLogicalCompaction : public LogicalExtensionOperator {
 class DuckLakeCompactor {
 public:
 	DuckLakeCompactor(ClientContext &context, DuckLakeCatalog &catalog, DuckLakeTransaction &transaction,
-	                  Binder &binder, TableIndex table_id);
+	                  Binder &binder, TableIndex table_id, uint64_t max_files);
 	DuckLakeCompactor(ClientContext &context, DuckLakeCatalog &catalog, DuckLakeTransaction &transaction,
 	                  Binder &binder, TableIndex table_id, double delete_threshold);
 	void GenerateCompactions(DuckLakeTableEntry &table, vector<unique_ptr<LogicalOperator>> &compactions);
@@ -154,14 +154,15 @@ class DuckLakeCompactor {
 	Binder &binder;
 	TableIndex table_id;
 	double delete_threshold = 0.95;
+	optional_idx max_files;
 
 	CompactionType type;
 };
 
 DuckLakeCompactor::DuckLakeCompactor(ClientContext &context, DuckLakeCatalog &catalog, DuckLakeTransaction &transaction,
-                                     Binder &binder, TableIndex table_id)
+                                     Binder &binder, TableIndex table_id, uint64_t max_files)
     : context(context), catalog(catalog), transaction(transaction), binder(binder), table_id(table_id),
-      type(CompactionType::MERGE_ADJACENT_TABLES) {
+      max_files(max_files), type(CompactionType::MERGE_ADJACENT_TABLES) {
 }
 
 DuckLakeCompactor::DuckLakeCompactor(ClientContext &context, DuckLakeCatalog &catalog, DuckLakeTransaction &transaction,
@@ -249,6 +250,7 @@ void DuckLakeCompactor::GenerateCompactions(DuckLakeTableEntry &table,
 	}
 	// we have gathered all the candidate files per compaction group
 	// iterate over them to generate actual compaction commands
+	uint64_t compacted_files = 0;
 	for (auto &entry : candidates) {
 		auto &candidate_list = entry.second.candidate_files;
 		if (candidate_list.size() <= 1) {
@@ -285,13 +287,21 @@ void DuckLakeCompactor::GenerateCompactions(DuckLakeTableEntry &table,
 
 			if (start_idx < compaction_idx) {
 				idx_t compaction_file_count = compaction_idx - start_idx;
+				if (compaction_file_count == 1) {
+					// If we only have one file to compact, we have nothing to compact
+					continue;
+				}
 				vector<DuckLakeCompactionFileEntry> compaction_files;
 				for (idx_t i = start_idx; i < compaction_idx; i++) {
 					compaction_files.push_back(std::move(files[candidate_list[i]]));
 				}
 				compactions.push_back(GenerateCompactionCommand(std::move(compaction_files)));
 				start_idx += compaction_file_count - 1;
 			}
+			compacted_files++;
+			if (compacted_files >= max_files.GetIndex()) {
+				break;
+			}
 		}
 	}
 }
@@ -494,10 +504,11 @@ static unique_ptr<LogicalOperator> GenerateCompactionOperator(TableFunctionBindI
 static void GenerateCompaction(ClientContext &context, DuckLakeTransaction &transaction,
                                DuckLakeCatalog &ducklake_catalog, TableFunctionBindInput &input,
                                DuckLakeTableEntry &cur_table, CompactionType type, double delete_threshold,
-                               vector<unique_ptr<LogicalOperator>> &compactions) {
+                               uint64_t max_files, vector<unique_ptr<LogicalOperator>> &compactions) {
 	switch (type) {
 	case CompactionType::MERGE_ADJACENT_TABLES: {
-		DuckLakeCompactor compactor(context, ducklake_catalog, transaction, *input.binder, cur_table.GetTableId());
+		DuckLakeCompactor compactor(context, ducklake_catalog, transaction, *input.binder, cur_table.GetTableId(),
+		                            max_files);
 		compactor.GenerateCompactions(cur_table, compactions);
 		break;
 	}
@@ -532,6 +543,17 @@ unique_ptr<LogicalOperator> BindCompaction(ClientContext &context, TableFunction
 	}
 
 	vector<unique_ptr<LogicalOperator>> compactions;
+	uint64_t max_files = NumericLimits<uint64_t>::Maximum() - 1;
+	auto max_files_entry = input.named_parameters.find("max_compacted_files");
+	if (max_files_entry != input.named_parameters.end()) {
+		if (max_files_entry->second.IsNull()) {
+			throw BinderException("The max_compacted_files option must be a non-null integer.");
+		}
+		max_files = UBigIntValue::Get(max_files_entry->second);
+		if (max_files == 0) {
+			throw BinderException("The max_compacted_files option must be greater than zero.");
+		}
+	}
 	if (input.inputs.size() == 1) {
 		if (schema.empty() && table.empty()) {
 			// No default schema/table, we will perform rewrites on deletes in the whole database
@@ -541,7 +563,7 @@ unique_ptr<LogicalOperator> BindCompaction(ClientContext &context, TableFunction
 					if (entry.type == CatalogType::TABLE_ENTRY) {
 						auto &cur_table = entry.Cast<DuckLakeTableEntry>();
 						GenerateCompaction(context, transaction, ducklake_catalog, input, cur_table, type,
-						                   delete_threshold, compactions);
+						                   delete_threshold, max_files, compactions);
 					}
 				});
 			}
@@ -554,7 +576,7 @@ unique_ptr<LogicalOperator> BindCompaction(ClientContext &context, TableFunction
 				if (entry.type == CatalogType::TABLE_ENTRY) {
 					auto &cur_table = entry.Cast<DuckLakeTableEntry>();
 					GenerateCompaction(context, transaction, ducklake_catalog, input, cur_table, type, delete_threshold,
-					                   compactions);
+					                   max_files, compactions);
 				}
 			});
 			return GenerateCompactionOperator(input, bind_index, compactions);
@@ -568,10 +590,11 @@ unique_ptr<LogicalOperator> BindCompaction(ClientContext &context, TableFunction
 	if (schema_entry != input.named_parameters.end()) {
 		schema = StringValue::Get(schema_entry->second);
 	}
-	EntryLookupInfo table_lookup(CatalogType::TABLE_ENTRY, table, nullptr, QueryErrorContext());
+
+	const EntryLookupInfo table_lookup(CatalogType::TABLE_ENTRY, table, nullptr, QueryErrorContext());
 	auto table_entry = catalog.GetEntry(context, schema, table_lookup, OnEntryNotFound::THROW_EXCEPTION);
 	auto &ducklake_table = table_entry->Cast<DuckLakeTableEntry>();
-	GenerateCompaction(context, transaction, ducklake_catalog, input, ducklake_table, type, delete_threshold,
+	GenerateCompaction(context, transaction, ducklake_catalog, input, ducklake_table, type, delete_threshold, max_files,
 	                   compactions);
 
 	return GenerateCompactionOperator(input, bind_index, compactions);
@@ -591,6 +614,7 @@ TableFunctionSet DuckLakeMergeAdjacentFilesFunction::GetFunctions() {
 		function.bind_operator = MergeAdjacentFilesBind;
 		if (type.size() == 2) {
 			function.named_parameters["schema"] = LogicalType::VARCHAR;
+			function.named_parameters["max_compacted_files"] = LogicalType::UBIGINT;
 		}
 		set.AddFunction(function);
 	}
diff --git a/src/include/common/ducklake_options.hpp b/src/include/common/ducklake_options.hpp
@@ -31,7 +31,7 @@ struct DuckLakeOptions {
 	bool create_if_not_exists = true;
 	bool migrate_if_required = true;
 	unique_ptr<BoundAtClause> at_clause;
-	unordered_map<string, Value> metadata_parameters;
+	case_insensitive_map_t<Value> metadata_parameters;
 	option_map_t config_options;
 	map<SchemaIndex, option_map_t> schema_options;
 	map<TableIndex, option_map_t> table_options;
diff --git a/src/storage/ducklake_table_entry.cpp b/src/storage/ducklake_table_entry.cpp
@@ -629,7 +629,7 @@ static bool TypePromotionIsAllowedUSmallint(const LogicalType &to) {
 	}
 }
 bool TypePromotionIsAllowed(const LogicalType &source, const LogicalType &target) {
-	// FIXME: support DECIMAL, and DATE -> TIMESTAMP
+	// FIXME: Rework to use DUCKDB_API static LogicalType MaxLogicalType
 	switch (source.id()) {
 	case LogicalTypeId::TINYINT:
 		return TypePromotionIsAllowedTinyint(target);
@@ -649,6 +649,8 @@ bool TypePromotionIsAllowed(const LogicalType &source, const LogicalType &target
 		return false;
 	case LogicalTypeId::FLOAT:
 		return target.id() == LogicalTypeId::DOUBLE;
+	case LogicalTypeId::TIMESTAMP:
+		return target.id() == LogicalTypeId::TIMESTAMP_TZ;
 	default:
 		return false;
 	}
diff --git a/src/storage/ducklake_update.cpp b/src/storage/ducklake_update.cpp
@@ -67,20 +67,24 @@ unique_ptr<LocalSinkState> DuckLakeUpdate::GetLocalSinkState(ExecutionContext &c
 	delete_types.emplace_back(LogicalType::UBIGINT);
 	delete_types.emplace_back(LogicalType::BIGINT);
 
-	vector<LogicalType> insert_types;
+	vector<LogicalType> expression_types;
 	result->expression_executor = make_uniq<ExpressionExecutor>(context.client, expressions);
 	for (auto &expr : result->expression_executor->expressions) {
-		insert_types.push_back(expr->return_type);
+		expression_types.push_back(expr->return_type);
 	}
 
-	for (auto &type : insert_types) {
+	for (auto &type : expression_types) {
 		if (DuckLakeTypes::RequiresCast(type)) {
 			type = DuckLakeTypes::GetCastedType(type);
 		}
 	}
-	result->update_expression_chunk.Initialize(context.client, insert_types);
-	// updates also write the row id to the file, so the final version needs the row_id
-	insert_types.push_back(LogicalType::BIGINT);
+	result->update_expression_chunk.Initialize(context.client, expression_types);
+	// updates also write the row id to the file, so the final version needs the row_id placed
+	// right after the physical columns (before computed partition columns).
+	vector<LogicalType> insert_types = expression_types;
+	auto physical_column_count = columns.size();
+	D_ASSERT(physical_column_count <= insert_types.size());
+	insert_types.insert(insert_types.begin() + physical_column_count, LogicalType::BIGINT);
 	result->insert_chunk.Initialize(context.client, insert_types);
 
 	result->delete_chunk.Initialize(context.client, delete_types);
@@ -101,12 +105,27 @@ SinkResultType DuckLakeUpdate::Sink(ExecutionContext &context, DataChunk &chunk,
 	insert_chunk.SetCardinality(chunk.size());
 	lstate.expression_executor->Execute(chunk, update_expression_chunk);
 
-	// We reference all columns we created in our updates
-	for (idx_t i = 0; i < update_expression_chunk.ColumnCount(); i++) {
+	const idx_t physical_column_count = columns.size();
+	const idx_t expression_column_count = update_expression_chunk.ColumnCount();
+	D_ASSERT(expression_column_count >= physical_column_count);
+	const idx_t partition_column_count = expression_column_count - physical_column_count;
+	const idx_t insert_column_count = insert_chunk.ColumnCount();
+	D_ASSERT(insert_column_count >= expression_column_count);
+	// virtual columns (PlanUpdate sets WRITE_ROW_ID, so there is exactly one: the row id) must sit between the physical and partition columns
+	const idx_t virtual_column_count = insert_column_count - expression_column_count;
+	D_ASSERT(virtual_column_count == 1);
+
+	// copy the physical columns directly
+	for (idx_t i = 0; i < physical_column_count; i++) {
 		insert_chunk.data[i].Reference(update_expression_chunk.data[i]);
 	}
-
-	insert_chunk.data[insert_chunk.data.size() - 1].Reference(chunk.data[row_id_index]);
+	// reference the row id right after the physical columns
+	insert_chunk.data[physical_column_count].Reference(chunk.data[row_id_index]);
+	// place computed partition columns after the virtual columns
+	for (idx_t part_idx = 0; part_idx < partition_column_count; part_idx++) {
+		insert_chunk.data[physical_column_count + virtual_column_count + part_idx]
+		    .Reference(update_expression_chunk.data[physical_column_count + part_idx]);
+	}
 
 	OperatorSinkInput copy_input {*copy_op.sink_state, *lstate.copy_local_state, input.interrupt_state};
 	copy_op.Sink(context, insert_chunk, copy_input);
diff --git a/test/configs/postgres.json b/test/configs/postgres.json
@@ -8,7 +8,8 @@
     "ducklake",
     "httpfs",
     "postgres_scanner",
-    "icu"
+    "icu",
+    "httpfs"
   ],
     "test_env": [{
       "env_name":  "DUCKLAKE_CONNECTION",
diff --git a/test/configs/sqlite.json b/test/configs/sqlite.json
@@ -7,7 +7,8 @@
     "ducklake",
     "httpfs",
     "sqlite_scanner",
-    "icu"
+    "icu",
+    "httpfs"
   ],
   "test_env": [{
       "env_name":  "DUCKLAKE_CONNECTION",
diff --git a/test/sql/alter/alter_timestamptz_promotion.test b/test/sql/alter/alter_timestamptz_promotion.test
@@ -0,0 +1,33 @@
+# name: test/sql/alter/alter_timestamptz_promotion.test
+# description: test ducklake timestamp to timestamptz promotion
+# group: [alter]
+
+require ducklake
+
+require parquet
+
+test-env DUCKLAKE_CONNECTION __TEST_DIR__/{UUID}.db
+
+test-env DATA_PATH __TEST_DIR__
+
+
+statement ok
+ATTACH 'ducklake:${DUCKLAKE_CONNECTION}' AS ducklake (DATA_PATH '${DATA_PATH}/alter_timestamptz_promotion')
+
+statement ok
+CREATE TABLE ducklake.test(col1 TIMESTAMP);
+
+statement ok
+INSERT INTO ducklake.test VALUES ('2025-01-15 12:30:45'::TIMESTAMP)
+
+statement ok
+ALTER TABLE ducklake.test ALTER COLUMN col1 TYPE TIMESTAMPTZ;
+
+statement ok
+INSERT INTO ducklake.test VALUES ('2025-01-15 12:30:45'::TIMESTAMPTZ)
+
+query I
+FROM ducklake.test
+----
+2025-01-15 12:30:45+00
+2025-01-15 12:30:45+00
diff --git a/test/sql/cleanup/cleanup_old_files.test b/test/sql/cleanup/cleanup_old_files.test
@@ -0,0 +1,54 @@
+# name: test/sql/cleanup/cleanup_old_files.test
+# description: Cleanup files when creating and dropping a table
+# group: [cleanup]
+
+require ducklake
+
+require parquet
+
+test-env DUCKLAKE_CONNECTION __TEST_DIR__/{UUID}.db
+
+test-env DATA_PATH __TEST_DIR__
+
+statement ok
+ATTACH 'ducklake:${DUCKLAKE_CONNECTION}' AS ducklake (DATA_PATH '${DATA_PATH}/cleanup_old_files')
+
+statement ok
+USE ducklake;
+
+statement ok
+CREATE TABLE t (x INT);
+
+statement ok
+INSERT INTO t VALUES (1), (2), (3);
+
+statement ok
+INSERT INTO t VALUES (4), (5);
+
+statement ok
+DELETE FROM t WHERE x <= 2;
+
+statement ok
+INSERT INTO t VALUES (6), (7);
+
+statement ok
+CALL ducklake_rewrite_data_files('ducklake', 't');
+
+statement ok
+CALL ducklake_merge_adjacent_files('ducklake');
+
+statement ok
+CALL ducklake_expire_snapshots('ducklake', older_than => now());
+
+query I
+SELECT COUNT(*) FROM ducklake_cleanup_old_files('ducklake', dry_run => true, cleanup_all => true);
+----
+2
+
+statement ok
+CALL ducklake_cleanup_old_files('ducklake', dry_run => false, cleanup_all => true);
+
+query I
+SELECT COUNT(*) FROM ducklake_cleanup_old_files('ducklake', dry_run => true, cleanup_all => true);
+----
+0
diff --git a/test/sql/compaction/merge_adjacent_max_files.test b/test/sql/compaction/merge_adjacent_max_files.test
diff --git a/test/sql/merge/merge_into_tpch.test_slow b/test/sql/merge/merge_into_tpch.test_slow
diff --git a/test/sql/merge/merge_partition_update.test b/test/sql/merge/merge_partition_update.test
diff --git a/test/sql/time_travel/time_travel_views.test b/test/sql/time_travel/time_travel_views.test

Original file line number	Diff line number	Diff line change
`@@ -629,7 +629,7 @@ static bool TypePromotionIsAllowedUSmallint(const LogicalType &to) {`
`629`	`629`	`}`
`630`	`630`	`}`
`631`	`631`	`bool TypePromotionIsAllowed(const LogicalType &source, const LogicalType &target) {`
`632`		`- // FIXME: support DECIMAL, and DATE -> TIMESTAMP`
	`632`	`+ // FIXME: Rework to use DUCKDB_API static LogicalType MaxLogicalType`
`633`	`633`	`switch (source.id()) {`
`634`	`634`	`case LogicalTypeId::TINYINT:`
`635`	`635`	`return TypePromotionIsAllowedTinyint(target);`
`@@ -649,6 +649,8 @@ bool TypePromotionIsAllowed(const LogicalType &source, const LogicalType &target`
`649`	`649`	`return false;`
`650`	`650`	`case LogicalTypeId::FLOAT:`
`651`	`651`	`return target.id() == LogicalTypeId::DOUBLE;`
	`652`	`+ case LogicalTypeId::TIMESTAMP:`
	`653`	`+ return target.id() == LogicalTypeId::TIMESTAMP_TZ;`
`652`	`654`	`default:`
`653`	`655`	`return false;`
`654`	`656`	`}`