diff --git a/src/duckdb/extension/core_functions/aggregate/holistic/approximate_quantile.cpp b/src/duckdb/extension/core_functions/aggregate/holistic/approximate_quantile.cpp
index 35336383b..9aa5e9406 100644
--- a/src/duckdb/extension/core_functions/aggregate/holistic/approximate_quantile.cpp
+++ b/src/duckdb/extension/core_functions/aggregate/holistic/approximate_quantile.cpp
@@ -355,11 +355,11 @@ AggregateFunction GetApproxQuantileListAggregateFunction(const LogicalType &type
 		return GetTypedApproxQuantileListAggregateFunction<int16_t, int16_t>(type);
 	case LogicalTypeId::INTEGER:
 	case LogicalTypeId::DATE:
-	case LogicalTypeId::TIME:
 		return GetTypedApproxQuantileListAggregateFunction<int32_t, int32_t>(type);
 	case LogicalTypeId::BIGINT:
 	case LogicalTypeId::TIMESTAMP:
 	case LogicalTypeId::TIMESTAMP_TZ:
+	case LogicalTypeId::TIME:
 		return GetTypedApproxQuantileListAggregateFunction<int64_t, int64_t>(type);
 	case LogicalTypeId::TIME_TZ:
 		//	Not binary comparable
diff --git a/src/duckdb/extension/icu/icu_extension.cpp b/src/duckdb/extension/icu/icu_extension.cpp
index b0cd3d277..c18b75efd 100644
--- a/src/duckdb/extension/icu/icu_extension.cpp
+++ b/src/duckdb/extension/icu/icu_extension.cpp
@@ -230,8 +230,16 @@ static string NormalizeTimeZone(const string &tz_str) {
 		}
 
 		idx_t pos = 3;
-		const auto sign = tz_str[pos++];
-		if (sign != '+' && sign != '-') {
+		const auto utc = tz_str[pos++];
+		// Invert the sign (UTC and Etc use opposite sign conventions)
+		// https://en.wikipedia.org/wiki/Tz_database#Area
+		auto sign = utc;
+		if (utc == '+') {
+			sign = '-';
+			;
+		} else if (utc == '-') {
+			sign = '+';
+		} else {
 			break;
 		}
 
@@ -424,12 +432,13 @@ static void LoadInternal(ExtensionLoader &loader) {
 	auto locales = icu::Collator::getAvailableLocales(count);
 	for (int32_t i = 0; i < count; i++) {
 		string collation;
-		if (string(locales[i].getCountry()).empty()) {
+		const auto &locale = locales[i]; // NOLINT
+		if (string(locale.getCountry()).empty()) {
 			// language only
-			collation = locales[i].getLanguage();
+			collation = locale.getLanguage();
 		} else {
 			// language + country
-			collation = locales[i].getLanguage() + string("_") + locales[i].getCountry();
+			collation = locale.getLanguage() + string("_") + locale.getCountry();
 		}
 		collation = StringUtil::Lower(collation);
 
diff --git a/src/duckdb/src/common/encryption_key_manager.cpp b/src/duckdb/src/common/encryption_key_manager.cpp
index 2e75a7e8b..e20f2208b 100644
--- a/src/duckdb/src/common/encryption_key_manager.cpp
+++ b/src/duckdb/src/common/encryption_key_manager.cpp
@@ -72,21 +72,25 @@ string EncryptionKeyManager::GenerateRandomKeyID() {
 }
 
 void EncryptionKeyManager::AddKey(const string &key_name, data_ptr_t key) {
+	lock_guard<mutex> guard(lock);
 	derived_keys.emplace(key_name, EncryptionKey(key));
 	// Zero-out the encryption key
 	duckdb_mbedtls::MbedTlsWrapper::AESStateMBEDTLS::SecureClearData(key, DERIVED_KEY_LENGTH);
 }
 
 bool EncryptionKeyManager::HasKey(const string &key_name) const {
+	lock_guard<mutex> guard(lock);
 	return derived_keys.find(key_name) != derived_keys.end();
 }
 
 const_data_ptr_t EncryptionKeyManager::GetKey(const string &key_name) const {
 	D_ASSERT(HasKey(key_name));
+	lock_guard<mutex> guard(lock);
 	return derived_keys.at(key_name).GetPtr();
 }
 
 void EncryptionKeyManager::DeleteKey(const string &key_name) {
+	lock_guard<mutex> guard(lock);
 	derived_keys.erase(key_name);
 }
 
diff --git a/src/duckdb/src/common/types/column/column_data_collection.cpp b/src/duckdb/src/common/types/column/column_data_collection.cpp
index b53e07d68..89890b325 100644
--- a/src/duckdb/src/common/types/column/column_data_collection.cpp
+++ b/src/duckdb/src/common/types/column/column_data_collection.cpp
@@ -1036,6 +1036,7 @@ void ColumnDataCollection::InitializeScan(ColumnDataParallelScanState &state, ve
 
 bool ColumnDataCollection::Scan(ColumnDataParallelScanState &state, ColumnDataLocalScanState &lstate,
                                 DataChunk &result) const {
+	D_ASSERT(result.GetTypes() == types);
 	result.Reset();
 
 	idx_t chunk_index;
@@ -1129,6 +1130,10 @@ void ColumnDataCollection::ScanAtIndex(ColumnDataParallelScanState &state, Colum
 }
 
 bool ColumnDataCollection::Scan(ColumnDataScanState &state, DataChunk &result) const {
+	for (idx_t i = 0; i < state.column_ids.size(); i++) {
+		D_ASSERT(result.GetTypes()[i] == types[state.column_ids[i]]);
+	}
+
 	result.Reset();
 
 	idx_t chunk_index;
@@ -1213,6 +1218,7 @@ idx_t ColumnDataCollection::ChunkCount() const {
 }
 
 void ColumnDataCollection::FetchChunk(idx_t chunk_idx, DataChunk &result) const {
+	D_ASSERT(result.GetTypes() == types);
 	D_ASSERT(chunk_idx < ChunkCount());
 	for (auto &segment : segments) {
 		if (chunk_idx >= segment->ChunkCount()) {
diff --git a/src/duckdb/src/common/types/conflict_manager.cpp b/src/duckdb/src/common/types/conflict_manager.cpp
index 49d5d1186..9348fd5c0 100644
--- a/src/duckdb/src/common/types/conflict_manager.cpp
+++ b/src/duckdb/src/common/types/conflict_manager.cpp
@@ -87,7 +87,7 @@ optional_idx ConflictManager::GetFirstInvalidIndex(const idx_t count, const bool
 	for (idx_t i = 0; i < count; i++) {
 		if (negate && !validity.RowIsValid(i)) {
 			return i;
-		} else if (validity.RowIsValid(i)) {
+		} else if (!negate && validity.RowIsValid(i)) {
 			return i;
 		}
 	}
diff --git a/src/duckdb/src/execution/operator/scan/physical_positional_scan.cpp b/src/duckdb/src/execution/operator/scan/physical_positional_scan.cpp
index bff24d785..6e4bfcaf3 100644
--- a/src/duckdb/src/execution/operator/scan/physical_positional_scan.cpp
+++ b/src/duckdb/src/execution/operator/scan/physical_positional_scan.cpp
@@ -67,10 +67,14 @@ class PositionalTableScanner {
 
 				InterruptState interrupt_state;
 				OperatorSourceInput source_input {global_state, *local_state, interrupt_state};
-				auto source_result = table.GetData(context, source, source_input);
-				if (source_result == SourceResultType::BLOCKED) {
-					throw NotImplementedException(
-					    "Unexpected interrupt from table Source in PositionalTableScanner refill");
+				auto source_result = SourceResultType::HAVE_MORE_OUTPUT;
+				while (source_result == SourceResultType::HAVE_MORE_OUTPUT && source.size() == 0) {
+					// TODO: this could as well just be propagated further, but for now iterating it is
+					source_result = table.GetData(context, source, source_input);
+					if (source_result == SourceResultType::BLOCKED) {
+						throw NotImplementedException(
+						    "Unexpected interrupt from table Source in PositionalTableScanner refill");
+					}
 				}
 			}
 			source_offset = 0;
diff --git a/src/duckdb/src/execution/physical_plan/plan_aggregate.cpp b/src/duckdb/src/execution/physical_plan/plan_aggregate.cpp
index fcb2aaef5..b22997068 100644
--- a/src/duckdb/src/execution/physical_plan/plan_aggregate.cpp
+++ b/src/duckdb/src/execution/physical_plan/plan_aggregate.cpp
@@ -236,7 +236,7 @@ PhysicalOperator &PhysicalPlanGenerator::CreatePlan(LogicalAggregate &op) {
 	D_ASSERT(op.children.size() == 1);
 
 	reference<PhysicalOperator> plan = CreatePlan(*op.children[0]);
-	plan = ExtractAggregateExpressions(plan, op.expressions, op.groups);
+	plan = ExtractAggregateExpressions(plan, op.expressions, op.groups, op.grouping_sets);
 
 	bool can_use_simple_aggregation = true;
 	for (auto &expression : op.expressions) {
@@ -305,7 +305,8 @@ PhysicalOperator &PhysicalPlanGenerator::CreatePlan(LogicalAggregate &op) {
 
 PhysicalOperator &PhysicalPlanGenerator::ExtractAggregateExpressions(PhysicalOperator &child,
                                                                      vector<unique_ptr<Expression>> &aggregates,
-                                                                     vector<unique_ptr<Expression>> &groups) {
+                                                                     vector<unique_ptr<Expression>> &groups,
+                                                                     optional_ptr<vector<GroupingSet>> grouping_sets) {
 	vector<unique_ptr<Expression>> expressions;
 	vector<LogicalType> types;
 
@@ -314,7 +315,7 @@ PhysicalOperator &PhysicalPlanGenerator::ExtractAggregateExpressions(PhysicalOpe
 		auto &bound_aggr = aggr->Cast<BoundAggregateExpression>();
 		if (bound_aggr.order_bys) {
 			// sorted aggregate!
-			FunctionBinder::BindSortedAggregate(context, bound_aggr, groups);
+			FunctionBinder::BindSortedAggregate(context, bound_aggr, groups, grouping_sets);
 		}
 	}
 	for (auto &group : groups) {
diff --git a/src/duckdb/src/execution/physical_plan/plan_distinct.cpp b/src/duckdb/src/execution/physical_plan/plan_distinct.cpp
index f1b8aec47..39b6d96e8 100644
--- a/src/duckdb/src/execution/physical_plan/plan_distinct.cpp
+++ b/src/duckdb/src/execution/physical_plan/plan_distinct.cpp
@@ -65,7 +65,8 @@ PhysicalOperator &PhysicalPlanGenerator::CreatePlan(LogicalDistinct &op) {
 
 			if (ClientConfig::GetConfig(context).enable_optimizer) {
 				bool changes_made = false;
-				auto new_expr = OrderedAggregateOptimizer::Apply(context, *first_aggregate, groups, changes_made);
+				auto new_expr =
+				    OrderedAggregateOptimizer::Apply(context, *first_aggregate, groups, nullptr, changes_made);
 				if (new_expr) {
 					D_ASSERT(new_expr->return_type == first_aggregate->return_type);
 					D_ASSERT(new_expr->GetExpressionType() == ExpressionType::BOUND_AGGREGATE);
@@ -81,7 +82,7 @@ PhysicalOperator &PhysicalPlanGenerator::CreatePlan(LogicalDistinct &op) {
 		}
 	}
 
-	child = ExtractAggregateExpressions(child, aggregates, groups);
+	child = ExtractAggregateExpressions(child, aggregates, groups, nullptr);
 
 	// we add a physical hash aggregation in the plan to select the distinct groups
 	auto &group_by = Make<PhysicalHashAggregate>(context, aggregate_types, std::move(aggregates), std::move(groups),
diff --git a/src/duckdb/src/function/aggregate/sorted_aggregate_function.cpp b/src/duckdb/src/function/aggregate/sorted_aggregate_function.cpp
index bde4c1479..c4fd191b7 100644
--- a/src/duckdb/src/function/aggregate/sorted_aggregate_function.cpp
+++ b/src/duckdb/src/function/aggregate/sorted_aggregate_function.cpp
@@ -677,14 +677,15 @@ struct SortedAggregateFunction {
 } // namespace
 
 void FunctionBinder::BindSortedAggregate(ClientContext &context, BoundAggregateExpression &expr,
-                                         const vector<unique_ptr<Expression>> &groups) {
+                                         const vector<unique_ptr<Expression>> &groups,
+                                         optional_ptr<vector<GroupingSet>> grouping_sets) {
 	if (!expr.order_bys || expr.order_bys->orders.empty() || expr.children.empty()) {
 		// not a sorted aggregate: return
 		return;
 	}
 	// Remove unnecessary ORDER BY clauses and return if nothing remains
 	if (context.config.enable_optimizer) {
-		if (expr.order_bys->Simplify(groups)) {
+		if (expr.order_bys->Simplify(groups, grouping_sets)) {
 			expr.order_bys.reset();
 			return;
 		}
@@ -741,7 +742,7 @@ void FunctionBinder::BindSortedAggregate(ClientContext &context, BoundWindowExpr
 	}
 	// Remove unnecessary ORDER BY clauses and return if nothing remains
 	if (context.config.enable_optimizer) {
-		if (BoundOrderModifier::Simplify(expr.arg_orders, expr.partitions)) {
+		if (BoundOrderModifier::Simplify(expr.arg_orders, expr.partitions, nullptr)) {
 			expr.arg_orders.clear();
 			return;
 		}
diff --git a/src/duckdb/src/function/macro_function.cpp b/src/duckdb/src/function/macro_function.cpp
index 66e36181b..8487827ec 100644
--- a/src/duckdb/src/function/macro_function.cpp
+++ b/src/duckdb/src/function/macro_function.cpp
@@ -48,13 +48,31 @@ MacroBindResult MacroFunction::BindMacroFunction(
 
 	ExpressionBinder expr_binder(binder, binder.context);
 	expr_binder.lambda_bindings = binder.lambda_bindings;
+
+	// Figure out whether we even need to bind arguments
+	bool requires_bind = false;
+	for (auto &function : functions) {
+		for (const auto &type : function->types) {
+			if (type.id() != LogicalTypeId::UNKNOWN) {
+				requires_bind = true;
+				break;
+			}
+		}
+		if (requires_bind) {
+			break;
+		}
+	}
+
 	// Find argument types and separate positional and default arguments
 	vector<LogicalType> positional_arg_types;
 	InsertionOrderPreservingMap<LogicalType> named_arg_types;
 	for (auto &arg : function_expr.children) {
 		auto arg_copy = arg->Copy();
-		const auto arg_bind_result = expr_binder.BindExpression(arg_copy, depth + 1);
-		auto arg_type = arg_bind_result.HasError() ? LogicalType::UNKNOWN : arg_bind_result.expression->return_type;
+		LogicalType arg_type = LogicalType::UNKNOWN;
+		if (requires_bind) {
+			const auto arg_bind_result = expr_binder.BindExpression(arg_copy, depth + 1);
+			arg_type = arg_bind_result.HasError() ? LogicalType::UNKNOWN : arg_bind_result.expression->return_type;
+		}
 		if (!arg->GetAlias().empty()) {
 			// Default argument
 			if (named_arguments.find(arg->GetAlias()) != named_arguments.end()) {
diff --git a/src/duckdb/src/function/table/system/test_all_types.cpp b/src/duckdb/src/function/table/system/test_all_types.cpp
index cd4ba3964..1e0c0ced3 100644
--- a/src/duckdb/src/function/table/system/test_all_types.cpp
+++ b/src/duckdb/src/function/table/system/test_all_types.cpp
@@ -19,9 +19,10 @@ struct TestAllTypesData : public GlobalTableFunctionState {
 	idx_t offset;
 };
 
-vector<TestType> TestAllTypesFun::GetTestTypes(bool use_large_enum, bool use_large_bignum) {
+vector<TestType> TestAllTypesFun::GetTestTypes(const bool use_large_enum, const bool use_large_bignum) {
 	vector<TestType> result;
-	// scalar types/numerics
+
+	// Numeric types.
 	result.emplace_back(LogicalType::BOOLEAN, "bool");
 	result.emplace_back(LogicalType::TINYINT, "tinyint");
 	result.emplace_back(LogicalType::SMALLINT, "smallint");
@@ -33,24 +34,31 @@ vector<TestType> TestAllTypesFun::GetTestTypes(bool use_large_enum, bool use_lar
 	result.emplace_back(LogicalType::USMALLINT, "usmallint");
 	result.emplace_back(LogicalType::UINTEGER, "uint");
 	result.emplace_back(LogicalType::UBIGINT, "ubigint");
+
+	// BIGNUM.
 	if (use_large_bignum) {
 		string data;
-		idx_t total_data_size = Bignum::BIGNUM_HEADER_SIZE + Bignum::MAX_DATA_SIZE;
+		constexpr idx_t total_data_size = Bignum::BIGNUM_HEADER_SIZE + Bignum::MAX_DATA_SIZE;
 		data.resize(total_data_size);
-		// Let's set our header
+
+		// Let's set the max header.
 		Bignum::SetHeader(&data[0], Bignum::MAX_DATA_SIZE, false);
-		// Set all our other bits
+		// Set all other max bits.
 		memset(&data[Bignum::BIGNUM_HEADER_SIZE], 0xFF, Bignum::MAX_DATA_SIZE);
 		auto max = Value::BIGNUM(data);
-		// Let's set our header
+
+		// Let's set the min header.
 		Bignum::SetHeader(&data[0], Bignum::MAX_DATA_SIZE, true);
-		// Set all our other bits
+		// Set all other min bits.
 		memset(&data[Bignum::BIGNUM_HEADER_SIZE], 0x00, Bignum::MAX_DATA_SIZE);
 		auto min = Value::BIGNUM(data);
 		result.emplace_back(LogicalType::BIGNUM, "bignum", min, max);
+
 	} else {
 		result.emplace_back(LogicalType::BIGNUM, "bignum");
 	}
+
+	// Time-types.
 	result.emplace_back(LogicalType::DATE, "date");
 	result.emplace_back(LogicalType::TIME, "time");
 	result.emplace_back(LogicalType::TIMESTAMP, "timestamp");
@@ -59,6 +67,8 @@ vector<TestType> TestAllTypesFun::GetTestTypes(bool use_large_enum, bool use_lar
 	result.emplace_back(LogicalType::TIMESTAMP_NS, "timestamp_ns");
 	result.emplace_back(LogicalType::TIME_TZ, "time_tz");
 	result.emplace_back(LogicalType::TIMESTAMP_TZ, "timestamp_tz");
+
+	// More complex numeric types.
 	result.emplace_back(LogicalType::FLOAT, "float");
 	result.emplace_back(LogicalType::DOUBLE, "double");
 	result.emplace_back(LogicalType::DECIMAL(4, 1), "dec_4_1");
@@ -67,7 +77,7 @@ vector<TestType> TestAllTypesFun::GetTestTypes(bool use_large_enum, bool use_lar
 	result.emplace_back(LogicalType::DECIMAL(38, 10), "dec38_10");
 	result.emplace_back(LogicalType::UUID, "uuid");
 
-	// interval
+	// Interval.
 	interval_t min_interval;
 	min_interval.months = 0;
 	min_interval.days = 0;
@@ -79,14 +89,15 @@ vector<TestType> TestAllTypesFun::GetTestTypes(bool use_large_enum, bool use_lar
 	max_interval.micros = 999999999;
 	result.emplace_back(LogicalType::INTERVAL, "interval", Value::INTERVAL(min_interval),
 	                    Value::INTERVAL(max_interval));
-	// strings/blobs/bitstrings
+
+	// VARCHAR / BLOB / Bitstrings.
 	result.emplace_back(LogicalType::VARCHAR, "varchar", Value("🦆🦆🦆🦆🦆🦆"),
 	                    Value(string("goo\x00se", 6)));
 	result.emplace_back(LogicalType::BLOB, "blob", Value::BLOB("thisisalongblob\\x00withnullbytes"),
 	                    Value::BLOB("\\x00\\x00\\x00a"));
 	result.emplace_back(LogicalType::BIT, "bit", Value::BIT("0010001001011100010101011010111"), Value::BIT("10101"));
 
-	// enums
+	// ENUMs.
 	Vector small_enum(LogicalType::VARCHAR, 2);
 	auto small_enum_ptr = FlatVector::GetData<string_t>(small_enum);
 	small_enum_ptr[0] = StringVector::AddStringOrBlob(small_enum, "DUCK_DUCK_ENUM");
@@ -116,7 +127,7 @@ vector<TestType> TestAllTypesFun::GetTestTypes(bool use_large_enum, bool use_lar
 		result.emplace_back(LogicalType::ENUM(large_enum, 2), "large_enum");
 	}
 
-	// arrays
+	// ARRAYs.
 	auto int_list_type = LogicalType::LIST(LogicalType::INTEGER);
 	auto empty_int_list = Value::LIST(LogicalType::INTEGER, vector<Value>());
 	auto int_list =
diff --git a/src/duckdb/src/function/table/table_scan.cpp b/src/duckdb/src/function/table/table_scan.cpp
index 99a9bcf79..1edd97e64 100644
--- a/src/duckdb/src/function/table/table_scan.cpp
+++ b/src/duckdb/src/function/table/table_scan.cpp
@@ -54,6 +54,7 @@ struct IndexScanLocalState : public LocalTableFunctionState {
 	TableScanState scan_state;
 	//! The column IDs of the local storage scan.
 	vector<StorageIndex> column_ids;
+	bool in_charge_of_final_stretch {false};
 };
 
 static StorageIndex TransformStorageIndex(const ColumnIndex &column_id) {
@@ -114,7 +115,7 @@ class DuckIndexScanState : public TableScanGlobalState {
 public:
 	DuckIndexScanState(ClientContext &context, const FunctionData *bind_data_p)
 	    : TableScanGlobalState(context, bind_data_p), next_batch_index(0), arena(Allocator::Get(context)),
-	      row_ids(nullptr), row_id_count(0), finished(false) {
+	      row_ids(nullptr), row_id_count(0), finished_first_phase(false), started_last_phase(false) {
 	}
 
 	//! The batch index of the next Sink.
@@ -129,7 +130,8 @@ class DuckIndexScanState : public TableScanGlobalState {
 	//! The column IDs of the to-be-scanned columns.
 	vector<StorageIndex> column_ids;
 	//! True, if no more row IDs must be scanned.
-	bool finished;
+	bool finished_first_phase;
+	bool started_last_phase;
 	//! Synchronize changes to the global index scan state.
 	mutex index_scan_lock;
 
@@ -163,44 +165,75 @@ class DuckIndexScanState : public TableScanGlobalState {
 		auto &storage = duck_table.GetStorage();
 		auto &l_state = data_p.local_state->Cast<IndexScanLocalState>();
 
-		idx_t scan_count = 0;
-		idx_t offset = 0;
-
-		{
-			// Synchronize changes to the shared global state.
-			lock_guard<mutex> l(index_scan_lock);
-			if (!finished) {
-				l_state.batch_index = next_batch_index;
-				next_batch_index++;
-
-				offset = l_state.batch_index * STANDARD_VECTOR_SIZE;
-				auto remaining = row_id_count - offset;
-				scan_count = remaining < STANDARD_VECTOR_SIZE ? remaining : STANDARD_VECTOR_SIZE;
-				finished = remaining < STANDARD_VECTOR_SIZE ? true : false;
+		enum class ExecutionPhase { NONE = 0, STORAGE = 1, LOCAL_STORAGE = 2 };
+
+		// We might need to loop back, so while (true)
+		while (true) {
+			idx_t scan_count = 0;
+			idx_t offset = 0;
+
+			// Phase selection
+			auto phase_to_be_performed = ExecutionPhase::NONE;
+			{
+				// Synchronize changes to the shared global state.
+				lock_guard<mutex> l(index_scan_lock);
+				if (!finished_first_phase) {
+					l_state.batch_index = next_batch_index;
+					next_batch_index++;
+
+					offset = l_state.batch_index * STANDARD_VECTOR_SIZE;
+					auto remaining = row_id_count - offset;
+					scan_count = remaining <= STANDARD_VECTOR_SIZE ? remaining : STANDARD_VECTOR_SIZE;
+					finished_first_phase = remaining <= STANDARD_VECTOR_SIZE ? true : false;
+					phase_to_be_performed = ExecutionPhase::STORAGE;
+				} else if (!started_last_phase) {
+					// First thread to get last phase, great, set l_state's in_charge_of_final_stretch, so same thread
+					// will be on again
+					started_last_phase = true;
+					l_state.in_charge_of_final_stretch = true;
+					phase_to_be_performed = ExecutionPhase::LOCAL_STORAGE;
+				} else if (l_state.in_charge_of_final_stretch) {
+					phase_to_be_performed = ExecutionPhase::LOCAL_STORAGE;
+				}
 			}
-		}
 
-		if (scan_count != 0) {
-			auto row_id_data = reinterpret_cast<data_ptr_t>(row_ids + offset);
-			Vector local_vector(LogicalType::ROW_TYPE, row_id_data);
-
-			if (CanRemoveFilterColumns()) {
-				l_state.all_columns.Reset();
-				storage.Fetch(tx, l_state.all_columns, column_ids, local_vector, scan_count, l_state.fetch_state);
-				output.ReferenceColumns(l_state.all_columns, projection_ids);
-			} else {
-				storage.Fetch(tx, output, column_ids, local_vector, scan_count, l_state.fetch_state);
+			switch (phase_to_be_performed) {
+			case ExecutionPhase::NONE: {
+				// No work to be picked up
+				return;
+			}
+			case ExecutionPhase::STORAGE: {
+				// Scan (in parallel) storage
+				auto row_id_data = reinterpret_cast<data_ptr_t>(row_ids + offset);
+				Vector local_vector(LogicalType::ROW_TYPE, row_id_data);
+
+				if (CanRemoveFilterColumns()) {
+					l_state.all_columns.Reset();
+					storage.Fetch(tx, l_state.all_columns, column_ids, local_vector, scan_count, l_state.fetch_state);
+					output.ReferenceColumns(l_state.all_columns, projection_ids);
+				} else {
+					storage.Fetch(tx, output, column_ids, local_vector, scan_count, l_state.fetch_state);
+				}
+				if (output.size() == 0) {
+					// output is empty, loop back, since there might be results to be picked up from LOCAL_STORAGE phase
+					continue;
+				}
+				return;
+			}
+			case ExecutionPhase::LOCAL_STORAGE: {
+				// Scan (sequentially, always same logical thread) local_storage
+				auto &local_storage = LocalStorage::Get(tx);
+				{
+					if (CanRemoveFilterColumns()) {
+						l_state.all_columns.Reset();
+						local_storage.Scan(l_state.scan_state.local_state, column_ids, l_state.all_columns);
+						output.ReferenceColumns(l_state.all_columns, projection_ids);
+					} else {
+						local_storage.Scan(l_state.scan_state.local_state, column_ids, output);
+					}
+				}
+				return;
 			}
-		}
-
-		if (output.size() == 0) {
-			auto &local_storage = LocalStorage::Get(tx);
-			if (CanRemoveFilterColumns()) {
-				l_state.all_columns.Reset();
-				local_storage.Scan(l_state.scan_state.local_state, column_ids, l_state.all_columns);
-				output.ReferenceColumns(l_state.all_columns, projection_ids);
-			} else {
-				local_storage.Scan(l_state.scan_state.local_state, column_ids, output);
 			}
 		}
 	}
@@ -350,7 +383,8 @@ unique_ptr<GlobalTableFunctionState> DuckTableScanInitGlobal(ClientContext &cont
 unique_ptr<GlobalTableFunctionState> DuckIndexScanInitGlobal(ClientContext &context, TableFunctionInitInput &input,
                                                              const TableScanBindData &bind_data, set<row_t> &row_ids) {
 	auto g_state = make_uniq<DuckIndexScanState>(context, input.bind_data.get());
-	g_state->finished = row_ids.empty() ? true : false;
+	g_state->finished_first_phase = row_ids.empty() ? true : false;
+	g_state->started_last_phase = false;
 
 	if (!row_ids.empty()) {
 		auto row_id_ptr = g_state->arena.AllocateAligned(row_ids.size() * sizeof(row_t));
diff --git a/src/duckdb/src/function/table/version/pragma_version.cpp b/src/duckdb/src/function/table/version/pragma_version.cpp
index c799882a5..640080b09 100644
--- a/src/duckdb/src/function/table/version/pragma_version.cpp
+++ b/src/duckdb/src/function/table/version/pragma_version.cpp
@@ -1,5 +1,5 @@
 #ifndef DUCKDB_PATCH_VERSION
-#define DUCKDB_PATCH_VERSION "3-dev8"
+#define DUCKDB_PATCH_VERSION "3-dev132"
 #endif
 #ifndef DUCKDB_MINOR_VERSION
 #define DUCKDB_MINOR_VERSION 4
@@ -8,10 +8,10 @@
 #define DUCKDB_MAJOR_VERSION 1
 #endif
 #ifndef DUCKDB_VERSION
-#define DUCKDB_VERSION "v1.4.3-dev8"
+#define DUCKDB_VERSION "v1.4.3-dev132"
 #endif
 #ifndef DUCKDB_SOURCE_ID
-#define DUCKDB_SOURCE_ID "5f0c38c5eb"
+#define DUCKDB_SOURCE_ID "9c1f71da3a"
 #endif
 #include "duckdb/function/table/system_functions.hpp"
 #include "duckdb/main/database.hpp"
diff --git a/src/duckdb/src/include/duckdb/common/encryption_key_manager.hpp b/src/duckdb/src/include/duckdb/common/encryption_key_manager.hpp
index 446b64758..fa256eab6 100644
--- a/src/duckdb/src/include/duckdb/common/encryption_key_manager.hpp
+++ b/src/duckdb/src/include/duckdb/common/encryption_key_manager.hpp
@@ -76,6 +76,7 @@ class EncryptionKeyManager : public ObjectCacheEntry {
 	static constexpr idx_t DERIVED_KEY_LENGTH = 32;
 
 private:
+	mutable mutex lock;
 	std::unordered_map<std::string, EncryptionKey> derived_keys;
 };
 
diff --git a/src/duckdb/src/include/duckdb/common/types/row/block_iterator.hpp b/src/duckdb/src/include/duckdb/common/types/row/block_iterator.hpp
index e18c01e53..1e0ccf305 100644
--- a/src/duckdb/src/include/duckdb/common/types/row/block_iterator.hpp
+++ b/src/duckdb/src/include/duckdb/common/types/row/block_iterator.hpp
@@ -31,6 +31,8 @@ template <class BLOCK_ITERATOR_STATE>
 class BlockIteratorStateBase {
 protected:
 	friend BLOCK_ITERATOR_STATE;
+
+private:
 	explicit BlockIteratorStateBase(const idx_t tuple_count_p) : tuple_count(tuple_count_p) {
 	}
 
diff --git a/src/duckdb/src/include/duckdb/execution/physical_plan_generator.hpp b/src/duckdb/src/include/duckdb/execution/physical_plan_generator.hpp
index 5d9e0aa46..1a1080e7b 100644
--- a/src/duckdb/src/include/duckdb/execution/physical_plan_generator.hpp
+++ b/src/duckdb/src/include/duckdb/execution/physical_plan_generator.hpp
@@ -10,6 +10,7 @@
 
 #include "duckdb/common/common.hpp"
 #include "duckdb/execution/physical_operator.hpp"
+#include "duckdb/parser/group_by_node.hpp"
 #include "duckdb/planner/logical_operator.hpp"
 #include "duckdb/planner/logical_tokens.hpp"
 #include "duckdb/planner/joinside.hpp"
@@ -152,7 +153,8 @@ class PhysicalPlanGenerator {
 	PhysicalOperator &PlanComparisonJoin(LogicalComparisonJoin &op);
 	PhysicalOperator &PlanDelimJoin(LogicalComparisonJoin &op);
 	PhysicalOperator &ExtractAggregateExpressions(PhysicalOperator &child, vector<unique_ptr<Expression>> &expressions,
-	                                              vector<unique_ptr<Expression>> &groups);
+	                                              vector<unique_ptr<Expression>> &groups,
+	                                              optional_ptr<vector<GroupingSet>> grouping_sets);
 
 private:
 	ClientContext &context;
diff --git a/src/duckdb/src/include/duckdb/function/function_binder.hpp b/src/duckdb/src/include/duckdb/function/function_binder.hpp
index 6eba740ab..6b43a0777 100644
--- a/src/duckdb/src/include/duckdb/function/function_binder.hpp
+++ b/src/duckdb/src/include/duckdb/function/function_binder.hpp
@@ -70,7 +70,8 @@ class FunctionBinder {
 	                      AggregateType aggr_type = AggregateType::NON_DISTINCT);
 
 	DUCKDB_API static void BindSortedAggregate(ClientContext &context, BoundAggregateExpression &expr,
-	                                           const vector<unique_ptr<Expression>> &groups);
+	                                           const vector<unique_ptr<Expression>> &groups,
+	                                           optional_ptr<vector<GroupingSet>> grouping_sets);
 	DUCKDB_API static void BindSortedAggregate(ClientContext &context, BoundWindowExpression &expr);
 
 	//! Cast a set of expressions to the arguments of this function
diff --git a/src/duckdb/src/include/duckdb/main/db_instance_cache.hpp b/src/duckdb/src/include/duckdb/main/db_instance_cache.hpp
index 8fa226b2d..a8a14c416 100644
--- a/src/duckdb/src/include/duckdb/main/db_instance_cache.hpp
+++ b/src/duckdb/src/include/duckdb/main/db_instance_cache.hpp
@@ -26,6 +26,8 @@ struct DatabaseCacheEntry {
 	mutex update_database_mutex;
 };
 
+enum class CacheBehavior { AUTOMATIC, ALWAYS_CACHE, NEVER_CACHE };
+
 class DBInstanceCache {
 public:
 	DBInstanceCache();
@@ -41,6 +43,9 @@ class DBInstanceCache {
 	//! Either returns an existing entry, or creates and caches a new DB Instance
 	shared_ptr<DuckDB> GetOrCreateInstance(const string &database, DBConfig &config_dict, bool cache_instance,
 	                                       const std::function<void(DuckDB &)> &on_create = nullptr);
+	shared_ptr<DuckDB> GetOrCreateInstance(const string &database, DBConfig &config_dict,
+	                                       CacheBehavior cache_behavior = CacheBehavior::AUTOMATIC,
+	                                       const std::function<void(DuckDB &)> &on_create = nullptr);
 
 private:
 	shared_ptr<DatabaseFilePathManager> path_manager;
diff --git a/src/duckdb/src/include/duckdb/main/extension_entries.hpp b/src/duckdb/src/include/duckdb/main/extension_entries.hpp
index 5ab7e72d0..e3d3d80d6 100644
--- a/src/duckdb/src/include/duckdb/main/extension_entries.hpp
+++ b/src/duckdb/src/include/duckdb/main/extension_entries.hpp
@@ -1045,6 +1045,7 @@ static constexpr ExtensionEntry EXTENSION_SETTINGS[] = {
     {"http_retry_wait_ms", "httpfs"},
     {"http_timeout", "httpfs"},
     {"httpfs_client_implementation", "httpfs"},
+    {"iceberg_via_aws_sdk_for_catalog_interactions", "iceberg"},
     {"mysql_bit1_as_boolean", "mysql_scanner"},
     {"mysql_debug_show_queries", "mysql_scanner"},
     {"mysql_experimental_filter_pushdown", "mysql_scanner"},
diff --git a/src/duckdb/src/include/duckdb/optimizer/filter_combiner.hpp b/src/duckdb/src/include/duckdb/optimizer/filter_combiner.hpp
index 890b90970..36cdbf59c 100644
--- a/src/duckdb/src/include/duckdb/optimizer/filter_combiner.hpp
+++ b/src/duckdb/src/include/duckdb/optimizer/filter_combiner.hpp
@@ -50,6 +50,7 @@ class FilterCombiner {
 	//! If this returns true - this sorts "in_list" as a side-effect
 	static bool IsDenseRange(vector<Value> &in_list);
 	static bool ContainsNull(vector<Value> &in_list);
+	static bool FindNextLegalUTF8(string &prefix_string);
 
 	void GenerateFilters(const std::function<void(unique_ptr<Expression> filter)> &callback);
 	bool HasFilters();
diff --git a/src/duckdb/src/include/duckdb/optimizer/join_order/relation_manager.hpp b/src/duckdb/src/include/duckdb/optimizer/join_order/relation_manager.hpp
index 2a687ad1b..13f037a69 100644
--- a/src/duckdb/src/include/duckdb/optimizer/join_order/relation_manager.hpp
+++ b/src/duckdb/src/include/duckdb/optimizer/join_order/relation_manager.hpp
@@ -57,10 +57,10 @@ class RelationManager {
 	bool ExtractBindings(Expression &expression, unordered_set<idx_t> &bindings);
 	void AddRelation(LogicalOperator &op, optional_ptr<LogicalOperator> parent, const RelationStats &stats);
 	//! Add an unnest relation which can come from a logical unnest or a logical get which has an unnest function
-	void AddUnnestRelation(JoinOrderOptimizer &optimizer, LogicalOperator &op, LogicalOperator &input_op,
-	                       optional_ptr<LogicalOperator> parent, RelationStats &child_stats,
-	                       optional_ptr<LogicalOperator> limit_op,
-	                       vector<reference<LogicalOperator>> &datasource_filters);
+	void AddRelationWithChildren(JoinOrderOptimizer &optimizer, LogicalOperator &op, LogicalOperator &input_op,
+	                             optional_ptr<LogicalOperator> parent, RelationStats &child_stats,
+	                             optional_ptr<LogicalOperator> limit_op,
+	                             vector<reference<LogicalOperator>> &datasource_filters);
 	void AddAggregateOrWindowRelation(LogicalOperator &op, optional_ptr<LogicalOperator> parent,
 	                                  const RelationStats &stats, LogicalOperatorType op_type);
 	vector<unique_ptr<SingleJoinRelation>> GetRelations();
diff --git a/src/duckdb/src/include/duckdb/optimizer/rule/ordered_aggregate_optimizer.hpp b/src/duckdb/src/include/duckdb/optimizer/rule/ordered_aggregate_optimizer.hpp
index 1b757e78c..6330b4144 100644
--- a/src/duckdb/src/include/duckdb/optimizer/rule/ordered_aggregate_optimizer.hpp
+++ b/src/duckdb/src/include/duckdb/optimizer/rule/ordered_aggregate_optimizer.hpp
@@ -10,6 +10,7 @@
 
 #include "duckdb/optimizer/rule.hpp"
 #include "duckdb/parser/expression_map.hpp"
+#include "duckdb/parser/group_by_node.hpp"
 
 namespace duckdb {
 
@@ -18,7 +19,8 @@ class OrderedAggregateOptimizer : public Rule {
 	explicit OrderedAggregateOptimizer(ExpressionRewriter &rewriter);
 
 	static unique_ptr<Expression> Apply(ClientContext &context, BoundAggregateExpression &aggr,
-	                                    vector<unique_ptr<Expression>> &groups, bool &changes_made);
+	                                    vector<unique_ptr<Expression>> &groups,
+	                                    optional_ptr<vector<GroupingSet>> grouping_sets, bool &changes_made);
 	unique_ptr<Expression> Apply(LogicalOperator &op, vector<reference<Expression>> &bindings, bool &changes_made,
 	                             bool is_root) override;
 };
diff --git a/src/duckdb/src/include/duckdb/planner/binder.hpp b/src/duckdb/src/include/duckdb/planner/binder.hpp
index 6ed02f8e3..5a1670bde 100644
--- a/src/duckdb/src/include/duckdb/planner/binder.hpp
+++ b/src/duckdb/src/include/duckdb/planner/binder.hpp
@@ -405,7 +405,7 @@ class Binder : public enable_shared_from_this<Binder> {
 
 	unique_ptr<QueryNode> BindTableMacro(FunctionExpression &function, TableMacroCatalogEntry &macro_func, idx_t depth);
 
-	unique_ptr<BoundCTENode> BindMaterializedCTE(CommonTableExpressionMap &cte_map);
+	unique_ptr<BoundCTENode> BindMaterializedCTE(CommonTableExpressionMap &cte_map, unique_ptr<CTENode> &cte_root);
 	unique_ptr<BoundCTENode> BindCTE(CTENode &statement);
 
 	unique_ptr<BoundQueryNode> BindNode(SelectNode &node);
diff --git a/src/duckdb/src/include/duckdb/planner/bound_result_modifier.hpp b/src/duckdb/src/include/duckdb/planner/bound_result_modifier.hpp
index 853384e0b..5f26dd8f7 100644
--- a/src/duckdb/src/include/duckdb/planner/bound_result_modifier.hpp
+++ b/src/duckdb/src/include/duckdb/planner/bound_result_modifier.hpp
@@ -9,6 +9,7 @@
 #pragma once
 
 #include "duckdb/common/limits.hpp"
+#include "duckdb/parser/group_by_node.hpp"
 #include "duckdb/parser/result_modifier.hpp"
 #include "duckdb/planner/bound_statement.hpp"
 #include "duckdb/planner/expression.hpp"
@@ -155,8 +156,9 @@ class BoundOrderModifier : public BoundResultModifier {
 
 	//! Remove unneeded/duplicate order elements.
 	//! Returns true of orders is not empty.
-	static bool Simplify(vector<BoundOrderByNode> &orders, const vector<unique_ptr<Expression>> &groups);
-	bool Simplify(const vector<unique_ptr<Expression>> &groups);
+	static bool Simplify(vector<BoundOrderByNode> &orders, const vector<unique_ptr<Expression>> &groups,
+	                     optional_ptr<vector<GroupingSet>> grouping_sets);
+	bool Simplify(const vector<unique_ptr<Expression>> &groups, optional_ptr<vector<GroupingSet>> grouping_sets);
 };
 
 enum class DistinctType : uint8_t { DISTINCT = 0, DISTINCT_ON = 1 };
diff --git a/src/duckdb/src/include/duckdb/planner/expression_binder.hpp b/src/duckdb/src/include/duckdb/planner/expression_binder.hpp
index b2712f3bf..5b4f74e6f 100644
--- a/src/duckdb/src/include/duckdb/planner/expression_binder.hpp
+++ b/src/duckdb/src/include/duckdb/planner/expression_binder.hpp
@@ -10,9 +10,8 @@
 
 #include "duckdb/common/exception.hpp"
 #include "duckdb/common/stack_checker.hpp"
-#include "duckdb/common/exception/binder_exception.hpp"
 #include "duckdb/common/error_data.hpp"
-#include "duckdb/common/unordered_map.hpp"
+#include "duckdb/common/exception/binder_exception.hpp"
 #include "duckdb/parser/expression/bound_expression.hpp"
 #include "duckdb/parser/expression/lambdaref_expression.hpp"
 #include "duckdb/parser/parsed_expression.hpp"
diff --git a/src/duckdb/src/include/duckdb/storage/table/chunk_info.hpp b/src/duckdb/src/include/duckdb/storage/table/chunk_info.hpp
index 44b92dd74..fb1267278 100644
--- a/src/duckdb/src/include/duckdb/storage/table/chunk_info.hpp
+++ b/src/duckdb/src/include/duckdb/storage/table/chunk_info.hpp
@@ -45,7 +45,7 @@ class ChunkInfo {
 	virtual bool Fetch(TransactionData transaction, row_t row) = 0;
 	virtual void CommitAppend(transaction_t commit_id, idx_t start, idx_t end) = 0;
 	virtual idx_t GetCommittedDeletedCount(idx_t max_count) = 0;
-	virtual bool Cleanup(transaction_t lowest_transaction, unique_ptr<ChunkInfo> &result) const;
+	virtual bool Cleanup(transaction_t lowest_transaction) const;
 
 	virtual bool HasDeletes() const = 0;
 
@@ -87,7 +87,7 @@ class ChunkConstantInfo : public ChunkInfo {
 	bool Fetch(TransactionData transaction, row_t row) override;
 	void CommitAppend(transaction_t commit_id, idx_t start, idx_t end) override;
 	idx_t GetCommittedDeletedCount(idx_t max_count) override;
-	bool Cleanup(transaction_t lowest_transaction, unique_ptr<ChunkInfo> &result) const override;
+	bool Cleanup(transaction_t lowest_transaction) const override;
 
 	bool HasDeletes() const override;
 
@@ -124,7 +124,7 @@ class ChunkVectorInfo : public ChunkInfo {
 	                            SelectionVector &sel_vector, idx_t max_count) override;
 	bool Fetch(TransactionData transaction, row_t row) override;
 	void CommitAppend(transaction_t commit_id, idx_t start, idx_t end) override;
-	bool Cleanup(transaction_t lowest_transaction, unique_ptr<ChunkInfo> &result) const override;
+	bool Cleanup(transaction_t lowest_transaction) const override;
 	idx_t GetCommittedDeletedCount(idx_t max_count) override;
 
 	void Append(idx_t start, idx_t end, transaction_t commit_id);
diff --git a/src/duckdb/src/include/duckdb/storage/table/row_group.hpp b/src/duckdb/src/include/duckdb/storage/table/row_group.hpp
index 10ef32a72..b5ea0b11b 100644
--- a/src/duckdb/src/include/duckdb/storage/table/row_group.hpp
+++ b/src/duckdb/src/include/duckdb/storage/table/row_group.hpp
@@ -99,10 +99,6 @@ class RowGroup : public SegmentBase<RowGroup> {
 
 	const vector<MetaBlockPointer> &GetColumnStartPointers() const;
 
-	//! Returns the list of meta block pointers used by the deletes
-	const vector<MetaBlockPointer> &GetDeletesPointers() const {
-		return deletes_pointers;
-	}
 	BlockManager &GetBlockManager();
 	DataTableInfo &GetTableInfo();
 
@@ -198,6 +194,8 @@ class RowGroup : public SegmentBase<RowGroup> {
 
 	static FilterPropagateResult CheckRowIdFilter(const TableFilter &filter, idx_t beg_row, idx_t end_row);
 
+	vector<MetaBlockPointer> CheckpointDeletes(MetadataManager &manager);
+
 private:
 	optional_ptr<RowVersionManager> GetVersionInfo();
 	shared_ptr<RowVersionManager> GetOrCreateVersionInfoPtr();
@@ -214,8 +212,6 @@ class RowGroup : public SegmentBase<RowGroup> {
 	template <TableScanType TYPE>
 	void TemplatedScan(TransactionData transaction, CollectionScanState &state, DataChunk &result);
 
-	vector<MetaBlockPointer> CheckpointDeletes(MetadataManager &manager);
-
 	bool HasUnloadedDeletes() const;
 
 private:
diff --git a/src/duckdb/src/include/duckdb/storage/table/row_version_manager.hpp b/src/duckdb/src/include/duckdb/storage/table/row_version_manager.hpp
index bb0d0056b..cc7464bde 100644
--- a/src/duckdb/src/include/duckdb/storage/table/row_version_manager.hpp
+++ b/src/duckdb/src/include/duckdb/storage/table/row_version_manager.hpp
@@ -46,11 +46,14 @@ class RowVersionManager {
 	static shared_ptr<RowVersionManager> Deserialize(MetaBlockPointer delete_pointer, MetadataManager &manager,
 	                                                 idx_t start);
 
+	bool HasUnserializedChanges();
+	vector<MetaBlockPointer> GetStoragePointers();
+
 private:
 	mutex version_lock;
 	idx_t start;
 	vector<unique_ptr<ChunkInfo>> vector_info;
-	bool has_changes;
+	bool has_unserialized_changes;
 	vector<MetaBlockPointer> storage_pointers;
 
 private:
diff --git a/src/duckdb/src/logging/log_storage.cpp b/src/duckdb/src/logging/log_storage.cpp
index c6733d968..e2596e003 100644
--- a/src/duckdb/src/logging/log_storage.cpp
+++ b/src/duckdb/src/logging/log_storage.cpp
@@ -22,26 +22,19 @@ namespace duckdb {
 
 vector<LogicalType> LogStorage::GetSchema(LoggingTargetTable table) {
 	switch (table) {
-	case LoggingTargetTable::ALL_LOGS:
-		return {
-		    LogicalType::UBIGINT,   // context_id
-		    LogicalType::VARCHAR,   // scope
-		    LogicalType::UBIGINT,   // connection_id
-		    LogicalType::UBIGINT,   // transaction_id
-		    LogicalType::UBIGINT,   // query_id
-		    LogicalType::UBIGINT,   // thread
-		    LogicalType::TIMESTAMP, // timestamp
-		    LogicalType::VARCHAR,   // log_type
-		    LogicalType::VARCHAR,   // level
-		    LogicalType::VARCHAR,   // message
-		};
+	case LoggingTargetTable::ALL_LOGS: {
+		auto all_logs = GetSchema(LoggingTargetTable::LOG_CONTEXTS);
+		auto log_entries = GetSchema(LoggingTargetTable::LOG_ENTRIES);
+		all_logs.insert(all_logs.end(), log_entries.begin() + 1, log_entries.end());
+		return all_logs;
+	}
 	case LoggingTargetTable::LOG_ENTRIES:
 		return {
-		    LogicalType::UBIGINT,   // context_id
-		    LogicalType::TIMESTAMP, // timestamp
-		    LogicalType::VARCHAR,   // log_type
-		    LogicalType::VARCHAR,   // level
-		    LogicalType::VARCHAR,   // message
+		    LogicalType::UBIGINT,      // context_id
+		    LogicalType::TIMESTAMP_TZ, // timestamp
+		    LogicalType::VARCHAR,      // log_type
+		    LogicalType::VARCHAR,      // level
+		    LogicalType::VARCHAR,      // message
 		};
 	case LoggingTargetTable::LOG_CONTEXTS:
 		return {
@@ -59,11 +52,12 @@ vector<LogicalType> LogStorage::GetSchema(LoggingTargetTable table) {
 
 vector<string> LogStorage::GetColumnNames(LoggingTargetTable table) {
 	switch (table) {
-	case LoggingTargetTable::ALL_LOGS:
-		return {
-		    "context_id", "scope",     "connection_id", "transaction_id", "query_id",
-		    "thread_id",  "timestamp", "type",          "log_level",      "message",
-		};
+	case LoggingTargetTable::ALL_LOGS: {
+		auto all_logs = GetColumnNames(LoggingTargetTable::LOG_CONTEXTS);
+		auto log_entries = GetColumnNames(LoggingTargetTable::LOG_ENTRIES);
+		all_logs.insert(all_logs.end(), log_entries.begin() + 1, log_entries.end());
+		return all_logs;
+	}
 	case LoggingTargetTable::LOG_ENTRIES:
 		return {"context_id", "timestamp", "type", "log_level", "message"};
 	case LoggingTargetTable::LOG_CONTEXTS:
diff --git a/src/duckdb/src/main/capi/duckdb-c.cpp b/src/duckdb/src/main/capi/duckdb-c.cpp
index 344fa265d..3cfedbf61 100644
--- a/src/duckdb/src/main/capi/duckdb-c.cpp
+++ b/src/duckdb/src/main/capi/duckdb-c.cpp
@@ -41,7 +41,7 @@ duckdb_state duckdb_open_internal(DBInstanceCacheWrapper *cache, const char *pat
 			if (path) {
 				path_str = path;
 			}
-			wrapper->database = cache->instance_cache->GetOrCreateInstance(path_str, *db_config, true);
+			wrapper->database = cache->instance_cache->GetOrCreateInstance(path_str, *db_config);
 		} else {
 			wrapper->database = duckdb::make_shared_ptr<DuckDB>(path, db_config);
 		}
diff --git a/src/duckdb/src/main/connection.cpp b/src/duckdb/src/main/connection.cpp
index af76cfd17..ccb2775bc 100644
--- a/src/duckdb/src/main/connection.cpp
+++ b/src/duckdb/src/main/connection.cpp
@@ -23,11 +23,6 @@ Connection::Connection(DatabaseInstance &database)
 	auto &connection_manager = ConnectionManager::Get(database);
 	connection_manager.AddConnection(*context);
 	connection_manager.AssignConnectionId(*this);
-
-#ifdef DEBUG
-	EnableProfiling();
-	context->config.emit_profiler_output = false;
-#endif
 }
 
 Connection::Connection(DuckDB &database) : Connection(*database.instance) {
diff --git a/src/duckdb/src/main/database_manager.cpp b/src/duckdb/src/main/database_manager.cpp
index 848f080fc..d9366396a 100644
--- a/src/duckdb/src/main/database_manager.cpp
+++ b/src/duckdb/src/main/database_manager.cpp
@@ -83,7 +83,16 @@ shared_ptr<AttachedDatabase> DatabaseManager::GetDatabaseInternal(const lock_gua
 
 shared_ptr<AttachedDatabase> DatabaseManager::AttachDatabase(ClientContext &context, AttachInfo &info,
                                                              AttachOptions &options) {
-	auto &config = DBConfig::GetConfig(context);
+	string extension = "";
+	if (FileSystem::IsRemoteFile(info.path, extension)) {
+		if (options.access_mode == AccessMode::AUTOMATIC) {
+			// Attaching of remote files gets bumped to READ_ONLY
+			// This is due to the fact that on most (all?) remote files writes to DB are not available
+			// and having this raised later is not super helpful
+			options.access_mode = AccessMode::READ_ONLY;
+		}
+	}
+
 	if (options.db_type.empty() || StringUtil::CIEquals(options.db_type, "duckdb")) {
 		while (InsertDatabasePath(info, options) == InsertDatabasePathResult::ALREADY_EXISTS) {
 			// database with this name and path already exists
@@ -99,6 +108,7 @@ shared_ptr<AttachedDatabase> DatabaseManager::AttachDatabase(ClientContext &cont
 			}
 		}
 	}
+	auto &config = DBConfig::GetConfig(context);
 	GetDatabaseType(context, info, config, options);
 	if (!options.db_type.empty()) {
 		// we only need to prevent duplicate opening of DuckDB files
@@ -108,18 +118,11 @@ shared_ptr<AttachedDatabase> DatabaseManager::AttachDatabase(ClientContext &cont
 	if (AttachedDatabase::NameIsReserved(info.name)) {
 		throw BinderException("Attached database name \"%s\" cannot be used because it is a reserved name", info.name);
 	}
-	string extension = "";
-	if (FileSystem::IsRemoteFile(info.path, extension)) {
+	if (!extension.empty()) {
 		if (!ExtensionHelper::TryAutoLoadExtension(context, extension)) {
 			throw MissingExtensionException("Attaching path '%s' requires extension '%s' to be loaded", info.path,
 			                                extension);
 		}
-		if (options.access_mode == AccessMode::AUTOMATIC) {
-			// Attaching of remote files gets bumped to READ_ONLY
-			// This is due to the fact that on most (all?) remote files writes to DB are not available
-			// and having this raised later is not super helpful
-			options.access_mode = AccessMode::READ_ONLY;
-		}
 	}
 
 	// now create the attached database
diff --git a/src/duckdb/src/main/db_instance_cache.cpp b/src/duckdb/src/main/db_instance_cache.cpp
index 57f4ee457..1960c5ee3 100644
--- a/src/duckdb/src/main/db_instance_cache.cpp
+++ b/src/duckdb/src/main/db_instance_cache.cpp
@@ -137,9 +137,23 @@ shared_ptr<DuckDB> DBInstanceCache::CreateInstance(const string &database, DBCon
 shared_ptr<DuckDB> DBInstanceCache::GetOrCreateInstance(const string &database, DBConfig &config_dict,
                                                         bool cache_instance,
                                                         const std::function<void(DuckDB &)> &on_create) {
+	auto cache_behavior = cache_instance ? CacheBehavior::ALWAYS_CACHE : CacheBehavior::NEVER_CACHE;
+	return GetOrCreateInstance(database, config_dict, cache_behavior, on_create);
+}
+
+shared_ptr<DuckDB> DBInstanceCache::GetOrCreateInstance(const string &database, DBConfig &config_dict,
+                                                        CacheBehavior cache_behavior,
+                                                        const std::function<void(DuckDB &)> &on_create) {
 	unique_lock<mutex> lock(cache_lock, std::defer_lock);
+	bool cache_instance = cache_behavior == CacheBehavior::ALWAYS_CACHE;
+	if (cache_behavior == CacheBehavior::AUTOMATIC) {
+		// cache all unnamed in-memory connections
+		cache_instance = true;
+		if (database == IN_MEMORY_PATH || database.empty()) {
+			cache_instance = false;
+		}
+	}
 	if (cache_instance) {
-
 		// While we do not own the lock, we cannot definitively say that the database instance does not exist.
 		while (!lock.owns_lock()) {
 			// The problem is, that we have to unlock the mutex in GetInstanceInternal, so we can non-blockingly wait
diff --git a/src/duckdb/src/optimizer/filter_combiner.cpp b/src/duckdb/src/optimizer/filter_combiner.cpp
index ddbe82ab0..f7099c9a1 100644
--- a/src/duckdb/src/optimizer/filter_combiner.cpp
+++ b/src/duckdb/src/optimizer/filter_combiner.cpp
@@ -2,6 +2,7 @@
 
 #include "duckdb/common/enums/expression_type.hpp"
 #include "duckdb/execution/expression_executor.hpp"
+#include "duckdb/function/scalar/string_common.hpp"
 #include "duckdb/optimizer/optimizer.hpp"
 #include "duckdb/planner/expression.hpp"
 #include "duckdb/planner/expression/bound_between_expression.hpp"
@@ -24,6 +25,7 @@
 #include "duckdb/optimizer/column_lifetime_analyzer.hpp"
 #include "duckdb/planner/expression_iterator.hpp"
 #include "duckdb/planner/operator/logical_get.hpp"
+#include "utf8proc_wrapper.hpp"
 
 namespace duckdb {
 
@@ -282,6 +284,35 @@ static bool SupportedFilterComparison(ExpressionType expression_type) {
 	}
 }
 
+bool FilterCombiner::FindNextLegalUTF8(string &prefix_string) {
+	// find the start of the last codepoint
+	idx_t last_codepoint_start;
+	for (last_codepoint_start = prefix_string.size(); last_codepoint_start > 0; last_codepoint_start--) {
+		if (IsCharacter(prefix_string[last_codepoint_start - 1])) {
+			break;
+		}
+	}
+	if (last_codepoint_start == 0) {
+		throw InvalidInputException("Invalid UTF8 found in string \"%s\"", prefix_string);
+	}
+	last_codepoint_start--;
+	int codepoint_size;
+	auto codepoint = Utf8Proc::UTF8ToCodepoint(prefix_string.c_str() + last_codepoint_start, codepoint_size) + 1;
+	if (codepoint >= 0xD800 && codepoint <= 0xDFFF) {
+		// next codepoint falls within surrogate range increment to next valid character
+		codepoint = 0xE000;
+	}
+	char next_codepoint_text[4];
+	int next_codepoint_size;
+	if (!Utf8Proc::CodepointToUtf8(codepoint, next_codepoint_size, next_codepoint_text)) {
+		// invalid codepoint
+		return false;
+	}
+	auto s = static_cast<idx_t>(next_codepoint_size);
+	prefix_string = prefix_string.substr(0, last_codepoint_start) + string(next_codepoint_text, s);
+	return true;
+}
+
 bool TypeSupportsConstantFilter(const LogicalType &type) {
 	if (TypeIsNumeric(type.InternalType())) {
 		return true;
@@ -397,11 +428,14 @@ FilterPushdownResult FilterCombiner::TryPushdownPrefixFilter(TableFilterSet &tab
 	auto &column_index = column_ids[column_ref.binding.column_index];
 	//! Replace prefix with a set of comparisons
 	auto lower_bound = make_uniq<ConstantFilter>(ExpressionType::COMPARE_GREATERTHANOREQUALTO, Value(prefix_string));
-	prefix_string[prefix_string.size() - 1]++;
-	auto upper_bound = make_uniq<ConstantFilter>(ExpressionType::COMPARE_LESSTHAN, Value(prefix_string));
 	table_filters.PushFilter(column_index, std::move(lower_bound));
-	table_filters.PushFilter(column_index, std::move(upper_bound));
-	return FilterPushdownResult::PUSHED_DOWN_FULLY;
+	if (FilterCombiner::FindNextLegalUTF8(prefix_string)) {
+		auto upper_bound = make_uniq<ConstantFilter>(ExpressionType::COMPARE_LESSTHAN, Value(prefix_string));
+		table_filters.PushFilter(column_index, std::move(upper_bound));
+		return FilterPushdownResult::PUSHED_DOWN_FULLY;
+	}
+	// could not find next legal utf8 string - skip upper bound
+	return FilterPushdownResult::NO_PUSHDOWN;
 }
 
 FilterPushdownResult FilterCombiner::TryPushdownLikeFilter(TableFilterSet &table_filters,
diff --git a/src/duckdb/src/optimizer/join_order/relation_manager.cpp b/src/duckdb/src/optimizer/join_order/relation_manager.cpp
index 72b584363..60f470224 100644
--- a/src/duckdb/src/optimizer/join_order/relation_manager.cpp
+++ b/src/duckdb/src/optimizer/join_order/relation_manager.cpp
@@ -54,12 +54,9 @@ void RelationManager::AddRelation(LogicalOperator &op, optional_ptr<LogicalOpera
 	auto relation_id = relations.size();
 
 	auto table_indexes = op.GetTableIndex();
-	bool is_unnest_or_get_with_unnest = op.type == LogicalOperatorType::LOGICAL_UNNEST;
+	bool get_all_child_bindings = op.type == LogicalOperatorType::LOGICAL_UNNEST;
 	if (op.type == LogicalOperatorType::LOGICAL_GET) {
-		auto &get = op.Cast<LogicalGet>();
-		if (get.function.name == "unnest") {
-			is_unnest_or_get_with_unnest = true;
-		}
+		get_all_child_bindings = !op.children.empty();
 	}
 	if (table_indexes.empty()) {
 		// relation represents a non-reorderable relation, most likely a join relation
@@ -72,9 +69,9 @@ void RelationManager::AddRelation(LogicalOperator &op, optional_ptr<LogicalOpera
 			D_ASSERT(relation_mapping.find(reference) == relation_mapping.end());
 			relation_mapping[reference] = relation_id;
 		}
-	} else if (is_unnest_or_get_with_unnest) {
-		// logical unnest has a logical_unnest index, but other bindings can refer to
-		// columns that are not unnested.
+	} else if (get_all_child_bindings) {
+		// logical get has a logical_get index, but if a function is present other bindings can refer to
+		// columns that are not unnested, and from the child of the logical get.
 		auto bindings = op.GetColumnBindings();
 		for (auto &binding : bindings) {
 			relation_mapping[binding.table_index] = relation_id;
@@ -189,10 +186,10 @@ static void ModifyStatsIfLimit(optional_ptr<LogicalOperator> limit_op, RelationS
 	}
 }
 
-void RelationManager::AddUnnestRelation(JoinOrderOptimizer &optimizer, LogicalOperator &op, LogicalOperator &input_op,
-                                        optional_ptr<LogicalOperator> parent, RelationStats &child_stats,
-                                        optional_ptr<LogicalOperator> limit_op,
-                                        vector<reference<LogicalOperator>> &datasource_filters) {
+void RelationManager::AddRelationWithChildren(JoinOrderOptimizer &optimizer, LogicalOperator &op,
+                                              LogicalOperator &input_op, optional_ptr<LogicalOperator> parent,
+                                              RelationStats &child_stats, optional_ptr<LogicalOperator> limit_op,
+                                              vector<reference<LogicalOperator>> &datasource_filters) {
 	D_ASSERT(!op.children.empty());
 	auto child_optimizer = optimizer.CreateChildOptimizer();
 	op.children[0] = child_optimizer.Optimize(std::move(op.children[0]), &child_stats);
@@ -301,7 +298,7 @@ bool RelationManager::ExtractJoinRelations(JoinOrderOptimizer &optimizer, Logica
 	case LogicalOperatorType::LOGICAL_UNNEST: {
 		// optimize children of unnest
 		RelationStats child_stats;
-		AddUnnestRelation(optimizer, *op, input_op, parent, child_stats, limit_op, datasource_filters);
+		AddRelationWithChildren(optimizer, *op, input_op, parent, child_stats, limit_op, datasource_filters);
 		return true;
 	}
 	case LogicalOperatorType::LOGICAL_COMPARISON_JOIN: {
@@ -359,9 +356,12 @@ bool RelationManager::ExtractJoinRelations(JoinOrderOptimizer &optimizer, Logica
 	case LogicalOperatorType::LOGICAL_GET: {
 		// TODO: Get stats from a logical GET
 		auto &get = op->Cast<LogicalGet>();
-		if (get.function.name == "unnest" && !op->children.empty()) {
+		// this is a get that *most likely* has a function (like unnest or json_each).
+		// there are new bindings for output of the function, but child bindings also exist, and can
+		// be used in joins
+		if (!op->children.empty()) {
 			RelationStats child_stats;
-			AddUnnestRelation(optimizer, *op, input_op, parent, child_stats, limit_op, datasource_filters);
+			AddRelationWithChildren(optimizer, *op, input_op, parent, child_stats, limit_op, datasource_filters);
 			return true;
 		}
 		auto stats = RelationStatisticsHelper::ExtractGetStats(get, context);
diff --git a/src/duckdb/src/optimizer/late_materialization.cpp b/src/duckdb/src/optimizer/late_materialization.cpp
index a144df188..689c87a9a 100644
--- a/src/duckdb/src/optimizer/late_materialization.cpp
+++ b/src/duckdb/src/optimizer/late_materialization.cpp
@@ -432,6 +432,11 @@ bool LateMaterialization::OptimizeLargeLimit(LogicalLimit &limit, idx_t limit_va
 		}
 		current_op = *current_op.get().children[0];
 	}
+	// if there are any filters we shouldn't do large limit optimization
+	auto &get = current_op.get().Cast<LogicalGet>();
+	if (!get.table_filters.filters.empty()) {
+		return false;
+	}
 	return true;
 }
 
diff --git a/src/duckdb/src/optimizer/rule/ordered_aggregate_optimizer.cpp b/src/duckdb/src/optimizer/rule/ordered_aggregate_optimizer.cpp
index 8f6435e1f..5bbbc2c48 100644
--- a/src/duckdb/src/optimizer/rule/ordered_aggregate_optimizer.cpp
+++ b/src/duckdb/src/optimizer/rule/ordered_aggregate_optimizer.cpp
@@ -17,7 +17,9 @@ OrderedAggregateOptimizer::OrderedAggregateOptimizer(ExpressionRewriter &rewrite
 }
 
 unique_ptr<Expression> OrderedAggregateOptimizer::Apply(ClientContext &context, BoundAggregateExpression &aggr,
-                                                        vector<unique_ptr<Expression>> &groups, bool &changes_made) {
+                                                        vector<unique_ptr<Expression>> &groups,
+                                                        optional_ptr<vector<GroupingSet>> grouping_sets,
+                                                        bool &changes_made) {
 	if (!aggr.order_bys) {
 		// no ORDER BYs defined
 		return nullptr;
@@ -30,7 +32,7 @@ unique_ptr<Expression> OrderedAggregateOptimizer::Apply(ClientContext &context,
 	}
 
 	// Remove unnecessary ORDER BY clauses and return if nothing remains
-	if (aggr.order_bys->Simplify(groups)) {
+	if (aggr.order_bys->Simplify(groups, grouping_sets)) {
 		aggr.order_bys.reset();
 		changes_made = true;
 		return nullptr;
@@ -90,7 +92,8 @@ unique_ptr<Expression> OrderedAggregateOptimizer::Apply(ClientContext &context,
 unique_ptr<Expression> OrderedAggregateOptimizer::Apply(LogicalOperator &op, vector<reference<Expression>> &bindings,
                                                         bool &changes_made, bool is_root) {
 	auto &aggr = bindings[0].get().Cast<BoundAggregateExpression>();
-	return Apply(rewriter.context, aggr, op.Cast<LogicalAggregate>().groups, changes_made);
+	return Apply(rewriter.context, aggr, op.Cast<LogicalAggregate>().groups, op.Cast<LogicalAggregate>().grouping_sets,
+	             changes_made);
 }
 
 } // namespace duckdb
diff --git a/src/duckdb/src/planner/binder.cpp b/src/duckdb/src/planner/binder.cpp
index 018957058..9628e463a 100644
--- a/src/duckdb/src/planner/binder.cpp
+++ b/src/duckdb/src/planner/binder.cpp
@@ -70,7 +70,7 @@ Binder::Binder(ClientContext &context, shared_ptr<Binder> parent_p, BinderType b
 	}
 }
 
-unique_ptr<BoundCTENode> Binder::BindMaterializedCTE(CommonTableExpressionMap &cte_map) {
+unique_ptr<BoundCTENode> Binder::BindMaterializedCTE(CommonTableExpressionMap &cte_map, unique_ptr<CTENode> &cte_root) {
 	// Extract materialized CTEs from cte_map
 	vector<unique_ptr<CTENode>> materialized_ctes;
 	for (auto &cte : cte_map.map) {
@@ -87,7 +87,6 @@ unique_ptr<BoundCTENode> Binder::BindMaterializedCTE(CommonTableExpressionMap &c
 		return nullptr;
 	}
 
-	unique_ptr<CTENode> cte_root = nullptr;
 	while (!materialized_ctes.empty()) {
 		unique_ptr<CTENode> node_result;
 		node_result = std::move(materialized_ctes.back());
@@ -110,7 +109,8 @@ unique_ptr<BoundCTENode> Binder::BindMaterializedCTE(CommonTableExpressionMap &c
 template <class T>
 BoundStatement Binder::BindWithCTE(T &statement) {
 	BoundStatement bound_statement;
-	auto bound_cte = BindMaterializedCTE(statement.template Cast<T>().cte_map);
+	unique_ptr<CTENode> cte_root;
+	auto bound_cte = BindMaterializedCTE(statement.template Cast<T>().cte_map, cte_root);
 	if (bound_cte) {
 		reference<BoundCTENode> tail_ref = *bound_cte;
 
diff --git a/src/duckdb/src/planner/binder/expression/bind_star_expression.cpp b/src/duckdb/src/planner/binder/expression/bind_star_expression.cpp
index f48fc14e6..84333d333 100644
--- a/src/duckdb/src/planner/binder/expression/bind_star_expression.cpp
+++ b/src/duckdb/src/planner/binder/expression/bind_star_expression.cpp
@@ -216,7 +216,7 @@ void TryTransformStarLike(unique_ptr<ParsedExpression> &root) {
 		child_expr = std::move(list_filter);
 	}
 
-	auto columns_expr = make_uniq<StarExpression>();
+	auto columns_expr = make_uniq<StarExpression>(star.relation_name);
 	columns_expr->columns = true;
 	columns_expr->expr = std::move(child_expr);
 	columns_expr->SetAlias(std::move(original_alias));
diff --git a/src/duckdb/src/planner/binder/statement/bind_copy.cpp b/src/duckdb/src/planner/binder/statement/bind_copy.cpp
index b7881a0a1..10bfbdc7b 100644
--- a/src/duckdb/src/planner/binder/statement/bind_copy.cpp
+++ b/src/duckdb/src/planner/binder/statement/bind_copy.cpp
@@ -423,7 +423,10 @@ vector<Value> BindCopyOption(ClientContext &context, TableFunctionBinder &option
 		}
 	}
 	auto bound_expr = option_binder.Bind(expr);
-	auto val = ExpressionExecutor::EvaluateScalar(context, *bound_expr);
+	if (bound_expr->HasParameter()) {
+		throw ParameterNotResolvedException();
+	}
+	auto val = ExpressionExecutor::EvaluateScalar(context, *bound_expr, true);
 	if (val.IsNull()) {
 		throw BinderException("NULL is not supported as a valid option for COPY option \"" + name + "\"");
 	}
diff --git a/src/duckdb/src/planner/bound_result_modifier.cpp b/src/duckdb/src/planner/bound_result_modifier.cpp
index edf49c4b1..4b7710bce 100644
--- a/src/duckdb/src/planner/bound_result_modifier.cpp
+++ b/src/duckdb/src/planner/bound_result_modifier.cpp
@@ -101,14 +101,17 @@ bool BoundOrderModifier::Equals(const unique_ptr<BoundOrderModifier> &left,
 	return BoundOrderModifier::Equals(*left, *right);
 }
 
-bool BoundOrderModifier::Simplify(vector<BoundOrderByNode> &orders, const vector<unique_ptr<Expression>> &groups) {
+bool BoundOrderModifier::Simplify(vector<BoundOrderByNode> &orders, const vector<unique_ptr<Expression>> &groups,
+                                  optional_ptr<vector<GroupingSet>> grouping_sets) {
 	// for each ORDER BY - check if it is actually necessary
 	// expressions that are in the groups do not need to be ORDERED BY
 	// `ORDER BY` on a group has no effect, because for each aggregate, the group is unique
 	// similarly, we only need to ORDER BY each aggregate once
+	expression_map_t<idx_t> group_expressions;
 	expression_set_t seen_expressions;
+	idx_t i = 0;
 	for (auto &target : groups) {
-		seen_expressions.insert(*target);
+		group_expressions.insert({*target, i++});
 	}
 	vector<BoundOrderByNode> new_order_nodes;
 	for (auto &order_node : orders) {
@@ -116,16 +119,30 @@ bool BoundOrderModifier::Simplify(vector<BoundOrderByNode> &orders, const vector
 			// we do not need to order by this node
 			continue;
 		}
+		auto it = group_expressions.find(*order_node.expression);
+		bool add_to_new_order = it == group_expressions.end();
+		if (!add_to_new_order && grouping_sets) {
+			idx_t group_idx = it->second;
+			for (auto &grouping_set : *grouping_sets) {
+				if (grouping_set.find(group_idx) == grouping_set.end()) {
+					add_to_new_order = true;
+					break;
+				}
+			}
+		}
 		seen_expressions.insert(*order_node.expression);
-		new_order_nodes.push_back(std::move(order_node));
+		if (add_to_new_order) {
+			new_order_nodes.push_back(std::move(order_node));
+		}
 	}
 	orders.swap(new_order_nodes);
 
 	return orders.empty(); // NOLINT
 }
 
-bool BoundOrderModifier::Simplify(const vector<unique_ptr<Expression>> &groups) {
-	return Simplify(orders, groups);
+bool BoundOrderModifier::Simplify(const vector<unique_ptr<Expression>> &groups,
+                                  optional_ptr<vector<GroupingSet>> grouping_sets) {
+	return Simplify(orders, groups, grouping_sets);
 }
 
 BoundLimitNode::BoundLimitNode(LimitNodeType type, idx_t constant_integer, double constant_percentage,
diff --git a/src/duckdb/src/planner/expression/bound_function_expression.cpp b/src/duckdb/src/planner/expression/bound_function_expression.cpp
index 5556dec21..44678762c 100644
--- a/src/duckdb/src/planner/expression/bound_function_expression.cpp
+++ b/src/duckdb/src/planner/expression/bound_function_expression.cpp
@@ -39,7 +39,10 @@ bool BoundFunctionExpression::IsFoldable() const {
 			}
 		}
 	}
-	return function.stability == FunctionStability::VOLATILE ? false : Expression::IsFoldable();
+	if (function.stability == FunctionStability::VOLATILE) {
+		return false;
+	}
+	return Expression::IsFoldable();
 }
 
 bool BoundFunctionExpression::CanThrow() const {
diff --git a/src/duckdb/src/planner/expression_binder.cpp b/src/duckdb/src/planner/expression_binder.cpp
index 220714733..eeeaeb97c 100644
--- a/src/duckdb/src/planner/expression_binder.cpp
+++ b/src/duckdb/src/planner/expression_binder.cpp
@@ -1,6 +1,5 @@
 #include "duckdb/planner/expression_binder.hpp"
 
-#include "duckdb/catalog/catalog_entry/scalar_function_catalog_entry.hpp"
 #include "duckdb/parser/expression/list.hpp"
 #include "duckdb/parser/parsed_expression_iterator.hpp"
 #include "duckdb/planner/binder.hpp"
@@ -166,7 +165,7 @@ static bool CombineMissingColumns(ErrorData &current, ErrorData new_error) {
 		}
 		auto score = StringUtil::SimilarityRating(candidate_column, column_name);
 		candidates.insert(candidate);
-		scores.emplace_back(make_pair(std::move(candidate), score));
+		scores.emplace_back(std::move(candidate), score);
 	}
 	// get a new top-n
 	auto top_candidates = StringUtil::TopNStrings(scores);
diff --git a/src/duckdb/src/planner/expression_binder/constant_binder.cpp b/src/duckdb/src/planner/expression_binder/constant_binder.cpp
index 97a65ba31..01f4ab11d 100644
--- a/src/duckdb/src/planner/expression_binder/constant_binder.cpp
+++ b/src/duckdb/src/planner/expression_binder/constant_binder.cpp
@@ -19,7 +19,7 @@ BindResult ConstantBinder::BindExpression(unique_ptr<ParsedExpression> &expr_ptr
 				return BindExpression(expr_ptr, depth, root_expression);
 			}
 		}
-		return BindUnsupportedExpression(expr, depth, clause + " cannot contain column names");
+		throw BinderException::Unsupported(expr, clause + " cannot contain column names");
 	}
 	case ExpressionClass::SUBQUERY:
 		throw BinderException(clause + " cannot contain subqueries");
diff --git a/src/duckdb/src/storage/storage_info.cpp b/src/duckdb/src/storage/storage_info.cpp
index fb1d78d59..5b5f7572c 100644
--- a/src/duckdb/src/storage/storage_info.cpp
+++ b/src/duckdb/src/storage/storage_info.cpp
@@ -85,6 +85,7 @@ static const StorageVersionInfo storage_version_info[] = {
 	{"v1.4.0", 67},
 	{"v1.4.1", 67},
 	{"v1.4.2", 67},
+	{"v1.4.3", 67},
 	{nullptr, 0}
 };
 // END OF STORAGE VERSION INFO
@@ -112,6 +113,7 @@ static const SerializationVersionInfo serialization_version_info[] = {
 	{"v1.4.0", 6},
 	{"v1.4.1", 6},
 	{"v1.4.2", 6},
+	{"v1.4.3", 6},
 	{"latest", 6},
 	{nullptr, 0}
 };
diff --git a/src/duckdb/src/storage/table/chunk_info.cpp b/src/duckdb/src/storage/table/chunk_info.cpp
index 3b7b11d7b..de6db28a0 100644
--- a/src/duckdb/src/storage/table/chunk_info.cpp
+++ b/src/duckdb/src/storage/table/chunk_info.cpp
@@ -32,7 +32,7 @@ static bool UseVersion(TransactionData transaction, transaction_t id) {
 	return TransactionVersionOperator::UseInsertedVersion(transaction.start_time, transaction.transaction_id, id);
 }
 
-bool ChunkInfo::Cleanup(transaction_t lowest_transaction, unique_ptr<ChunkInfo> &result) const {
+bool ChunkInfo::Cleanup(transaction_t lowest_transaction) const {
 	return false;
 }
 
@@ -99,7 +99,7 @@ idx_t ChunkConstantInfo::GetCommittedDeletedCount(idx_t max_count) {
 	return delete_id < TRANSACTION_ID_START ? max_count : 0;
 }
 
-bool ChunkConstantInfo::Cleanup(transaction_t lowest_transaction, unique_ptr<ChunkInfo> &result) const {
+bool ChunkConstantInfo::Cleanup(transaction_t lowest_transaction) const {
 	if (delete_id != NOT_DELETED_ID) {
 		// the chunk info is labeled as deleted - we need to keep it around
 		return false;
@@ -253,7 +253,7 @@ void ChunkVectorInfo::CommitAppend(transaction_t commit_id, idx_t start, idx_t e
 	}
 }
 
-bool ChunkVectorInfo::Cleanup(transaction_t lowest_transaction, unique_ptr<ChunkInfo> &result) const {
+bool ChunkVectorInfo::Cleanup(transaction_t lowest_transaction) const {
 	if (any_deleted) {
 		// if any rows are deleted we can't clean-up
 		return false;
diff --git a/src/duckdb/src/storage/table/column_data.cpp b/src/duckdb/src/storage/table/column_data.cpp
index aafcf58a0..a1aa2b1c9 100644
--- a/src/duckdb/src/storage/table/column_data.cpp
+++ b/src/duckdb/src/storage/table/column_data.cpp
@@ -536,6 +536,11 @@ void ColumnData::RevertAppend(row_t start_row_p) {
 	if (segment->start == start_row) {
 		// we are truncating exactly this segment - erase it entirely
 		data.EraseSegments(l, segment_index);
+		if (segment_index > 0) {
+			// if we have a previous segment, we need to update the next pointer
+			auto previous_segment = data.GetSegmentByIndex(l, UnsafeNumericCast<int64_t>(segment_index - 1));
+			previous_segment->next = nullptr;
+		}
 	} else {
 		// we need to truncate within the segment
 		// remove any segments AFTER this segment: they should be deleted entirely
diff --git a/src/duckdb/src/storage/table/column_segment.cpp b/src/duckdb/src/storage/table/column_segment.cpp
index 347463fbe..f7fcc25b3 100644
--- a/src/duckdb/src/storage/table/column_segment.cpp
+++ b/src/duckdb/src/storage/table/column_segment.cpp
@@ -242,7 +242,9 @@ void ColumnSegment::ConvertToPersistent(QueryContext context, optional_ptr<Block
 	// Thus, we set the compression function to constant and reset the block buffer.
 	D_ASSERT(stats.statistics.IsConstant());
 	auto &config = DBConfig::GetConfig(db);
-	function = *config.GetCompressionFunction(CompressionType::COMPRESSION_CONSTANT, type.InternalType());
+	if (GetCompressionFunction().type != CompressionType::COMPRESSION_EMPTY) {
+		function = *config.GetCompressionFunction(CompressionType::COMPRESSION_CONSTANT, type.InternalType());
+	}
 	block.reset();
 }
 
diff --git a/src/duckdb/src/storage/table/row_group.cpp b/src/duckdb/src/storage/table/row_group.cpp
index a79b0b4fc..3824ca85a 100644
--- a/src/duckdb/src/storage/table/row_group.cpp
+++ b/src/duckdb/src/storage/table/row_group.cpp
@@ -1053,7 +1053,7 @@ RowGroupPointer RowGroup::Checkpoint(RowGroupWriteData write_data, RowGroupWrite
 		row_group_pointer.data_pointers = column_pointers;
 		row_group_pointer.has_metadata_blocks = true;
 		row_group_pointer.extra_metadata_blocks = write_data.existing_extra_metadata_blocks;
-		row_group_pointer.deletes_pointers = deletes_pointers;
+		row_group_pointer.deletes_pointers = CheckpointDeletes(*metadata_manager);
 		vector<MetaBlockPointer> extra_metadata_block_pointers;
 		extra_metadata_block_pointers.reserve(write_data.existing_extra_metadata_blocks.size());
 		for (auto &block_pointer : write_data.existing_extra_metadata_blocks) {
@@ -1061,7 +1061,6 @@ RowGroupPointer RowGroup::Checkpoint(RowGroupWriteData write_data, RowGroupWrite
 		}
 		metadata_manager->ClearModifiedBlocks(column_pointers);
 		metadata_manager->ClearModifiedBlocks(extra_metadata_block_pointers);
-		metadata_manager->ClearModifiedBlocks(deletes_pointers);
 		// remember metadata_blocks to avoid loading them on future checkpoints
 		has_metadata_blocks = true;
 		extra_metadata_blocks = row_group_pointer.extra_metadata_blocks;
@@ -1109,14 +1108,13 @@ RowGroupPointer RowGroup::Checkpoint(RowGroupWriteData write_data, RowGroupWrite
 		row_group_pointer.extra_metadata_blocks.push_back(column_pointer.block_pointer);
 		metadata_blocks.insert(column_pointer.block_pointer);
 	}
+	if (metadata_manager) {
+		row_group_pointer.deletes_pointers = CheckpointDeletes(*metadata_manager);
+	}
 	// set up the pointers correctly within this row group for future operations
 	column_pointers = row_group_pointer.data_pointers;
 	has_metadata_blocks = true;
 	extra_metadata_blocks = row_group_pointer.extra_metadata_blocks;
-
-	if (metadata_manager) {
-		row_group_pointer.deletes_pointers = CheckpointDeletes(*metadata_manager);
-	}
 	Verify();
 	return row_group_pointer;
 }
@@ -1125,11 +1123,11 @@ bool RowGroup::HasChanges() const {
 	if (has_changes) {
 		return true;
 	}
-	if (version_info.load()) {
+	auto version_info_loaded = version_info.load();
+	if (version_info_loaded && version_info_loaded->HasUnserializedChanges()) {
 		// we have deletes
 		return true;
 	}
-	D_ASSERT(!deletes_is_loaded.load());
 	// check if any of the columns have changes
 	// avoid loading unloaded columns - unloaded columns can never have changes
 	for (idx_t c = 0; c < columns.size(); c++) {
diff --git a/src/duckdb/src/storage/table/row_group_collection.cpp b/src/duckdb/src/storage/table/row_group_collection.cpp
index e7b5aee7a..fce6605fc 100644
--- a/src/duckdb/src/storage/table/row_group_collection.cpp
+++ b/src/duckdb/src/storage/table/row_group_collection.cpp
@@ -14,6 +14,7 @@
 #include "duckdb/storage/table/column_checkpoint_state.hpp"
 #include "duckdb/storage/table/persistent_table_data.hpp"
 #include "duckdb/storage/table/row_group_segment_tree.hpp"
+#include "duckdb/storage/table/row_version_manager.hpp"
 #include "duckdb/storage/table/scan_state.hpp"
 #include "duckdb/storage/table_storage_info.hpp"
 #include "duckdb/main/settings.hpp"
@@ -505,6 +506,11 @@ void RowGroupCollection::RevertAppendInternal(idx_t start_row) {
 	if (segment.start == start_row) {
 		// we are truncating exactly this row group - erase it entirely
 		row_groups->EraseSegments(l, segment_index);
+		if (segment_index > 0) {
+			// if we have a previous segment, we need to update the next pointer
+			auto previous_segment = row_groups->GetSegmentByIndex(l, UnsafeNumericCast<int64_t>(segment_index - 1));
+			previous_segment->next = nullptr;
+		}
 	} else {
 		// we need to truncate within a row group
 		// remove any segments AFTER this segment: they should be deleted entirely
@@ -1167,7 +1173,7 @@ void RowGroupCollection::Checkpoint(TableDataWriter &writer, TableStatistics &gl
 					extra_metadata_block_pointers.emplace_back(block_pointer, 0);
 				}
 				metadata_manager.ClearModifiedBlocks(extra_metadata_block_pointers);
-				metadata_manager.ClearModifiedBlocks(row_group.GetDeletesPointers());
+				row_group.CheckpointDeletes(metadata_manager);
 				row_groups->AppendSegment(l, std::move(entry.node));
 			}
 			writer.WriteUnchangedTable(metadata_pointer, total_rows.load());
@@ -1285,6 +1291,40 @@ void RowGroupCollection::Checkpoint(TableDataWriter &writer, TableStatistics &gl
 
 				throw InternalException("Reloading blocks just written does not yield same blocks: " + oss.str());
 			}
+
+			vector<MetaBlockPointer> read_deletes_pointers;
+			if (!pointer_copy.deletes_pointers.empty()) {
+				auto root_delete = pointer_copy.deletes_pointers[0];
+				auto vm = RowVersionManager::Deserialize(root_delete, GetBlockManager().GetMetadataManager(),
+				                                         row_group.start);
+				read_deletes_pointers = vm->GetStoragePointers();
+			}
+
+			set<idx_t> all_written_deletes_block_ids;
+			for (auto &ptr : pointer_copy.deletes_pointers) {
+				all_written_deletes_block_ids.insert(ptr.block_pointer);
+			}
+			set<idx_t> all_read_deletes_block_ids;
+			for (auto &ptr : read_deletes_pointers) {
+				all_read_deletes_block_ids.insert(ptr.block_pointer);
+			}
+
+			if (all_written_deletes_block_ids != all_read_deletes_block_ids) {
+				std::stringstream oss;
+				oss << "Written: ";
+				for (auto &block : all_written_deletes_block_ids) {
+					oss << block << ", ";
+				}
+				oss << "\n";
+				oss << "Read: ";
+				for (auto &block : all_read_deletes_block_ids) {
+					oss << block << ", ";
+				}
+				oss << "\n";
+
+				throw InternalException("Reloading deletes blocks just written does not yield same blocks: " +
+				                        oss.str());
+			}
 		}
 	}
 	total_rows = new_total_rows;
diff --git a/src/duckdb/src/storage/table/row_version_manager.cpp b/src/duckdb/src/storage/table/row_version_manager.cpp
index df4e463da..7df22474f 100644
--- a/src/duckdb/src/storage/table/row_version_manager.cpp
+++ b/src/duckdb/src/storage/table/row_version_manager.cpp
@@ -7,7 +7,7 @@
 
 namespace duckdb {
 
-RowVersionManager::RowVersionManager(idx_t start) noexcept : start(start), has_changes(false) {
+RowVersionManager::RowVersionManager(idx_t start) noexcept : start(start), has_unserialized_changes(false) {
 }
 
 void RowVersionManager::SetStart(idx_t new_start) {
@@ -88,7 +88,7 @@ void RowVersionManager::FillVectorInfo(idx_t vector_idx) {
 void RowVersionManager::AppendVersionInfo(TransactionData transaction, idx_t count, idx_t row_group_start,
                                           idx_t row_group_end) {
 	lock_guard<mutex> lock(version_lock);
-	has_changes = true;
+	has_unserialized_changes = true;
 	idx_t start_vector_idx = row_group_start / STANDARD_VECTOR_SIZE;
 	idx_t end_vector_idx = (row_group_end - 1) / STANDARD_VECTOR_SIZE;
 
@@ -141,6 +141,7 @@ void RowVersionManager::CommitAppend(transaction_t commit_id, idx_t row_group_st
 		idx_t vend =
 		    vector_idx == end_vector_idx ? row_group_end - end_vector_idx * STANDARD_VECTOR_SIZE : STANDARD_VECTOR_SIZE;
 		auto &info = *vector_info[vector_idx];
+		D_ASSERT(has_unserialized_changes);
 		info.CommitAppend(commit_id, vstart, vend);
 	}
 }
@@ -167,10 +168,12 @@ void RowVersionManager::CleanupAppend(transaction_t lowest_active_transaction, i
 		}
 		auto &info = *vector_info[vector_idx];
 		// if we wrote the entire chunk info try to compress it
-		unique_ptr<ChunkInfo> new_info;
-		auto cleanup = info.Cleanup(lowest_active_transaction, new_info);
+		auto cleanup = info.Cleanup(lowest_active_transaction);
 		if (cleanup) {
-			vector_info[vector_idx] = std::move(new_info);
+			if (info.HasDeletes()) {
+				has_unserialized_changes = true;
+			}
+			vector_info[vector_idx].reset();
 		}
 	}
 }
@@ -179,6 +182,7 @@ void RowVersionManager::RevertAppend(idx_t start_row) {
 	lock_guard<mutex> lock(version_lock);
 	idx_t start_vector_idx = (start_row + (STANDARD_VECTOR_SIZE - 1)) / STANDARD_VECTOR_SIZE;
 	for (idx_t vector_idx = start_vector_idx; vector_idx < vector_info.size(); vector_idx++) {
+		D_ASSERT(has_unserialized_changes);
 		vector_info[vector_idx].reset();
 	}
 }
@@ -205,19 +209,19 @@ ChunkVectorInfo &RowVersionManager::GetVectorInfo(idx_t vector_idx) {
 
 idx_t RowVersionManager::DeleteRows(idx_t vector_idx, transaction_t transaction_id, row_t rows[], idx_t count) {
 	lock_guard<mutex> lock(version_lock);
-	has_changes = true;
+	has_unserialized_changes = true;
 	return GetVectorInfo(vector_idx).Delete(transaction_id, rows, count);
 }
 
 void RowVersionManager::CommitDelete(idx_t vector_idx, transaction_t commit_id, const DeleteInfo &info) {
 	lock_guard<mutex> lock(version_lock);
-	has_changes = true;
+	has_unserialized_changes = true;
 	GetVectorInfo(vector_idx).CommitDelete(commit_id, info);
 }
 
 vector<MetaBlockPointer> RowVersionManager::Checkpoint(MetadataManager &manager) {
-	if (!has_changes && !storage_pointers.empty()) {
-		// the row version manager already exists on disk and no changes were made
+	lock_guard<mutex> lock(version_lock);
+	if (!has_unserialized_changes) {
 		// we can write the current pointer as-is
 		// ensure the blocks we are pointing to are not marked as free
 		manager.ClearModifiedBlocks(storage_pointers);
@@ -236,24 +240,23 @@ vector<MetaBlockPointer> RowVersionManager::Checkpoint(MetadataManager &manager)
 		}
 		to_serialize.emplace_back(vector_idx, *chunk_info);
 	}
-	if (to_serialize.empty()) {
-		return vector<MetaBlockPointer>();
-	}
 
 	storage_pointers.clear();
 
-	MetadataWriter writer(manager, &storage_pointers);
-	// now serialize the actual version information
-	writer.Write<idx_t>(to_serialize.size());
-	for (auto &entry : to_serialize) {
-		auto &vector_idx = entry.first;
-		auto &chunk_info = entry.second.get();
-		writer.Write<idx_t>(vector_idx);
-		chunk_info.Write(writer);
+	if (!to_serialize.empty()) {
+		MetadataWriter writer(manager, &storage_pointers);
+		// now serialize the actual version information
+		writer.Write<idx_t>(to_serialize.size());
+		for (auto &entry : to_serialize) {
+			auto &vector_idx = entry.first;
+			auto &chunk_info = entry.second.get();
+			writer.Write<idx_t>(vector_idx);
+			chunk_info.Write(writer);
+		}
+		writer.Flush();
 	}
-	writer.Flush();
 
-	has_changes = false;
+	has_unserialized_changes = false;
 	return storage_pointers;
 }
 
@@ -277,8 +280,19 @@ shared_ptr<RowVersionManager> RowVersionManager::Deserialize(MetaBlockPointer de
 		version_info->FillVectorInfo(vector_index);
 		version_info->vector_info[vector_index] = ChunkInfo::Read(source);
 	}
-	version_info->has_changes = false;
+	version_info->has_unserialized_changes = false;
 	return version_info;
 }
 
+bool RowVersionManager::HasUnserializedChanges() {
+	lock_guard<mutex> lock(version_lock);
+	return has_unserialized_changes;
+}
+
+vector<MetaBlockPointer> RowVersionManager::GetStoragePointers() {
+	lock_guard<mutex> lock(version_lock);
+	D_ASSERT(!has_unserialized_changes);
+	return storage_pointers;
+}
+
 } // namespace duckdb