diff --git a/src/duckdb/extension/core_functions/scalar/list/list_value.cpp b/src/duckdb/extension/core_functions/scalar/list/list_value.cpp
index c66d36d42..2619e6847 100644
--- a/src/duckdb/extension/core_functions/scalar/list/list_value.cpp
+++ b/src/duckdb/extension/core_functions/scalar/list/list_value.cpp
@@ -267,6 +267,12 @@ unique_ptr<FunctionData> UnpivotBind(BindScalarFunctionInput &input) {
 	}
 	child_type = LogicalType::NormalizeType(child_type);
 
+	auto &function_args = bound_function.GetArguments();
+	function_args.clear();
+	function_args.reserve(arguments.size());
+	for (idx_t i = 0; i < arguments.size(); i++) {
+		function_args.push_back(child_type);
+	}
 	bound_function.SetReturnType(LogicalType::LIST(child_type));
 	return make_uniq<VariableReturnBindData>(bound_function.GetReturnType());
 }
diff --git a/src/duckdb/src/common/hive_partitioning.cpp b/src/duckdb/src/common/hive_partitioning.cpp
index 092a98c66..bff0d0c5b 100644
--- a/src/duckdb/src/common/hive_partitioning.cpp
+++ b/src/duckdb/src/common/hive_partitioning.cpp
@@ -86,12 +86,30 @@ string HivePartitioning::Escape(const string &input) {
 	return StringUtil::URLEncode(input);
 }
 
+string HivePartitioning::EscapeValue(const string &input) {
+	auto result = Escape(input);
+	// the comparison is case-insensitive because on a case-insensitive file system a value that differs only in
+	// case still lands in the directory that is reserved for NULL values
+	if (!StringUtil::CIEquals(result, DEFAULT_PARTITION_NAME)) {
+		return result;
+	}
+	// percent-encode the first character so the value gets its own directory while still unescaping back to the
+	// original value
+	static constexpr const char *HEX_DIGIT = "0123456789ABCDEF";
+	const auto first = static_cast<unsigned char>(result[0]);
+	string escaped = "%";
+	escaped += HEX_DIGIT[first >> 4];
+	escaped += HEX_DIGIT[first & 15];
+	escaped += result.substr(1);
+	return escaped;
+}
+
 string HivePartitioning::Unescape(const string &input) {
 	return StringUtil::URLDecode(input);
 }
 
 bool HivePartitioning::IsNull(const string &input) {
-	return StringUtil::CIEquals(input, "NULL") || input == "__HIVE_DEFAULT_PARTITION__";
+	return StringUtil::CIEquals(input, "NULL") || input == DEFAULT_PARTITION_NAME;
 }
 
 // matches hive partitions in file name. For example:
@@ -132,7 +150,7 @@ std::map<string, string> HivePartitioning::Parse(const string &filename) {
 Value HivePartitioning::GetValue(ClientContext &context, const string &key, const string &str_val,
                                  const LogicalType &type) {
 	// On SQLNULL, DuckDB writes "__HIVE_DEFAULT_PARTITION__", instead of string version "NULL".
-	if (str_val == "__HIVE_DEFAULT_PARTITION__") {
+	if (str_val == DEFAULT_PARTITION_NAME) {
 		return Value(type);
 	}
 	if (type.id() == LogicalTypeId::VARCHAR) {
diff --git a/src/duckdb/src/execution/operator/persistent/physical_copy_to_file.cpp b/src/duckdb/src/execution/operator/persistent/physical_copy_to_file.cpp
index 3e00c378a..acef1199b 100644
--- a/src/duckdb/src/execution/operator/persistent/physical_copy_to_file.cpp
+++ b/src/duckdb/src/execution/operator/persistent/physical_copy_to_file.cpp
@@ -2990,9 +2990,9 @@ PartitionDirectory PartitionFileRequestBuilder::BuildDirectory(string path) cons
 			p_dir += HivePartitioning::Escape(partition_col_name.GetIdentifierName());
 			p_dir += "=";
 			if (partition_value.IsNull()) {
-				p_dir += "__HIVE_DEFAULT_PARTITION__";
+				p_dir += HivePartitioning::DEFAULT_PARTITION_NAME;
 			} else {
-				p_dir += HivePartitioning::Escape(partition_value.ToString());
+				p_dir += HivePartitioning::EscapeValue(partition_value.ToString());
 			}
 			result.path = fs.JoinPath(result.path, p_dir);
 			result.directories.push_back(result.path);
diff --git a/src/duckdb/src/function/table/system/duckdb_dialects.cpp b/src/duckdb/src/function/table/system/duckdb_dialects.cpp
new file mode 100644
index 000000000..844081ab6
--- /dev/null
+++ b/src/duckdb/src/function/table/system/duckdb_dialects.cpp
@@ -0,0 +1,44 @@
+#include "duckdb/function/table/system_functions.hpp"
+
+#include "duckdb/main/client_context.hpp"
+#include "duckdb/main/extension_callback_manager.hpp"
+#include "duckdb/parser/dialect_extension.hpp"
+
+namespace duckdb {
+
+struct DuckDBDialectsData : public GlobalTableFunctionState {
+	vector<string> dialects;
+	idx_t offset = 0;
+};
+
+static unique_ptr<FunctionData> DuckDBDialectsBind(ClientContext &context, TableFunctionBindInput &input,
+                                                   vector<LogicalType> &return_types, vector<Identifier> &names) {
+	names.emplace_back("dialect_name");
+	return_types.emplace_back(LogicalType::VARCHAR);
+	return nullptr;
+}
+
+static unique_ptr<GlobalTableFunctionState> DuckDBDialectsInit(ClientContext &context, TableFunctionInitInput &input) {
+	auto result = make_uniq<DuckDBDialectsData>();
+	for (auto &dialect : ExtensionCallbackManager::Get(context).DialectExtensions()) {
+		result->dialects.push_back(dialect.name);
+	}
+	return std::move(result);
+}
+
+static void DuckDBDialectsFunction(ClientContext &context, TableFunctionInput &data_p, DataChunk &output) {
+	auto &data = data_p.global_state->Cast<DuckDBDialectsData>();
+	auto &dialect_name = output.data[0];
+	idx_t count = 0;
+	while (data.offset < data.dialects.size() && count < STANDARD_VECTOR_SIZE) {
+		dialect_name.Append(Value(data.dialects[data.offset++]));
+		count++;
+	}
+}
+
+void DuckDBDialectsFun::RegisterFunction(BuiltinFunctions &set) {
+	set.AddFunction(
+	    TableFunction("duckdb_dialects", {}, DuckDBDialectsFunction, DuckDBDialectsBind, DuckDBDialectsInit));
+}
+
+} // namespace duckdb
diff --git a/src/duckdb/src/function/table/system_functions.cpp b/src/duckdb/src/function/table/system_functions.cpp
index f67e30fdc..63219fcb7 100644
--- a/src/duckdb/src/function/table/system_functions.cpp
+++ b/src/duckdb/src/function/table/system_functions.cpp
@@ -32,6 +32,7 @@ void BuiltinFunctions::RegisterSQLiteFunctions() {
 	DuckDBIndexesFun::RegisterFunction(*this);
 	DuckDBSchemasFun::RegisterFunction(*this);
 	DuckDBDependenciesFun::RegisterFunction(*this);
+	DuckDBDialectsFun::RegisterFunction(*this);
 	DuckDBExtensionsFun::RegisterFunction(*this);
 	RegisterExternalResourceTypeFun::RegisterFunction(*this);
 	CreateExternalResourceFun::RegisterFunction(*this);
diff --git a/src/duckdb/src/function/table/version/pragma_version.cpp b/src/duckdb/src/function/table/version/pragma_version.cpp
index c12043f0d..068cda0b9 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 "0-alpha38143"
+#define DUCKDB_PATCH_VERSION "0-alpha38195"
 #endif
 #ifndef DUCKDB_MINOR_VERSION
 #define DUCKDB_MINOR_VERSION 0
@@ -8,10 +8,10 @@
 #define DUCKDB_MAJOR_VERSION 2
 #endif
 #ifndef DUCKDB_VERSION
-#define DUCKDB_VERSION "v2.0.0-alpha38143"
+#define DUCKDB_VERSION "v2.0.0-alpha38195"
 #endif
 #ifndef DUCKDB_SOURCE_ID
-#define DUCKDB_SOURCE_ID "11dc00c898"
+#define DUCKDB_SOURCE_ID "8cbdaba6ac"
 #endif
 #include "duckdb/function/table/system_functions.hpp"
 #include "duckdb/main/database.hpp"
diff --git a/src/duckdb/src/include/duckdb/common/hive_partitioning.hpp b/src/duckdb/src/include/duckdb/common/hive_partitioning.hpp
index a217359f7..534020b92 100644
--- a/src/duckdb/src/include/duckdb/common/hive_partitioning.hpp
+++ b/src/duckdb/src/include/duckdb/common/hive_partitioning.hpp
@@ -24,6 +24,10 @@ struct HivePartitioningFilterInfo {
 };
 
 class HivePartitioning {
+public:
+	//! The directory name that is written for a NULL partition value
+	static constexpr const char *DEFAULT_PARTITION_NAME = "__HIVE_DEFAULT_PARTITION__";
+
 public:
 	//! Parse a filename that follows the hive partitioning scheme
 	DUCKDB_API static std::map<string, string> Parse(const string &filename);
@@ -39,6 +43,8 @@ class HivePartitioning {
 	                                 const LogicalType &type);
 	//! Escape a hive partition key or value using URL encoding
 	DUCKDB_API static string Escape(const string &input);
+	//! Escape a non-NULL hive partition value, avoiding a collision with the NULL partition directory
+	DUCKDB_API static string EscapeValue(const string &input);
 	//! Unescape a hive partition key or value encoded using URL encoding
 	DUCKDB_API static string Unescape(const string &input);
 	//! Whether the value is a null marker when detecting Hive partition types
diff --git a/src/duckdb/src/include/duckdb/common/multi_file/multi_file_function.hpp b/src/duckdb/src/include/duckdb/common/multi_file/multi_file_function.hpp
index 30eb0a9e1..7d7f43297 100644
--- a/src/duckdb/src/include/duckdb/common/multi_file/multi_file_function.hpp
+++ b/src/duckdb/src/include/duckdb/common/multi_file/multi_file_function.hpp
@@ -932,6 +932,14 @@ class MultiFileFunction : public TableFunction {
 		auto primary_index = column_index.GetPrimaryIndex();
 		const auto &col_name = bind_data.names[primary_index];
 
+		// a hive partitioning column overrides any file column of the same name - the statistics stored in the
+		// file describe the overridden column and can even have a different type, so they cannot be used here
+		for (auto &hive_partitioning_index : bind_data.reader_bind.hive_partitioning_indexes) {
+			if (hive_partitioning_index.index == primary_index) {
+				return nullptr;
+			}
+		}
+
 		// NOTE: we do not want to parse the file metadata for the sole purpose of getting column statistics
 		if (bind_data.file_list->GetExpandResult() == FileExpandResult::MULTIPLE_FILES) {
 			if (!bind_data.file_options.union_by_name) {
diff --git a/src/duckdb/src/include/duckdb/function/table/system_functions.hpp b/src/duckdb/src/include/duckdb/function/table/system_functions.hpp
index 05fdf6bdc..f014798e8 100644
--- a/src/duckdb/src/include/duckdb/function/table/system_functions.hpp
+++ b/src/duckdb/src/include/duckdb/function/table/system_functions.hpp
@@ -84,6 +84,10 @@ struct DuckDBDependenciesFun {
 	static void RegisterFunction(BuiltinFunctions &set);
 };
 
+struct DuckDBDialectsFun {
+	static void RegisterFunction(BuiltinFunctions &set);
+};
+
 struct DuckDBExtensionsFun {
 	static void RegisterFunction(BuiltinFunctions &set);
 };
diff --git a/src/duckdb/src/include/duckdb/main/extension_callback_manager.hpp b/src/duckdb/src/include/duckdb/main/extension_callback_manager.hpp
index 0b1c99219..4f30ac406 100644
--- a/src/duckdb/src/include/duckdb/main/extension_callback_manager.hpp
+++ b/src/duckdb/src/include/duckdb/main/extension_callback_manager.hpp
@@ -17,6 +17,7 @@ namespace duckdb {
 
 class ClientContext;
 class DatabaseInstance;
+class DialectExtension;
 class ExtensionCallback;
 class OperatorExtension;
 class OptimizerExtension;
@@ -42,6 +43,7 @@ class ExtensionCallbackManager {
 	static const ExtensionCallbackManager &Get(const ClientContext &context);
 
 	void Register(ParserExtension extension);
+	void Register(DialectExtension extension);
 	void Register(PlannerExtension extension);
 	void Register(OptimizerExtension extension);
 	void Register(shared_ptr<OperatorExtension> extension);
@@ -52,11 +54,13 @@ class ExtensionCallbackManager {
 	ExtensionCallbackIteratorHelper<shared_ptr<OperatorExtension>> OperatorExtensions() const;
 	ExtensionCallbackIteratorHelper<OptimizerExtension> OptimizerExtensions() const;
 	ExtensionCallbackIteratorHelper<ParserExtension> ParserExtensions() const;
+	ExtensionCallbackIteratorHelper<DialectExtension> DialectExtensions() const;
 	ExtensionCallbackIteratorHelper<PlannerExtension> PlannerExtensions() const;
 	ExtensionCallbackIteratorHelper<shared_ptr<ExtensionCallback>> ExtensionCallbacks() const;
 	optional_ptr<StorageExtension> FindStorageExtension(const string &name) const;
 	optional_ptr<ProfilerExtension> FindProfilerExtension(const string &name) const;
 	bool HasParserExtensions() const;
+	bool HasDialectExtension(const string &name) const;
 
 private:
 	mutex registry_lock;
diff --git a/src/duckdb/src/include/duckdb/main/settings.hpp b/src/duckdb/src/include/duckdb/main/settings.hpp
index 131c9bd45..247f53769 100644
--- a/src/duckdb/src/include/duckdb/main/settings.hpp
+++ b/src/duckdb/src/include/duckdb/main/settings.hpp
@@ -457,6 +457,17 @@ struct ConfigureProfilingSetting {
 	static Value GetSetting(const ClientContext &context);
 };
 
+struct CurrentDialectSetting {
+	using RETURN_TYPE = string;
+	static constexpr const char *Name = "current_dialect";
+	static constexpr const char *Description = "The SQL dialect used by the parser";
+	static constexpr const char *InputType = "VARCHAR";
+	static constexpr const char *DefaultValue = "duckdb";
+	static constexpr SettingScopeTarget Scope = SettingScopeTarget::GLOBAL_ONLY;
+	static constexpr idx_t SettingIndex = NEXT_SETTING_INDEX();
+	static void OnSet(SettingCallbackInfo &info, Value &input);
+};
+
 struct CurrentTransactionInvalidationPolicySetting {
 	using RETURN_TYPE = string;
 	static constexpr const char *Name = "current_transaction_invalidation_policy";
diff --git a/src/duckdb/src/include/duckdb/parser/dialect_extension.hpp b/src/duckdb/src/include/duckdb/parser/dialect_extension.hpp
new file mode 100644
index 000000000..8ae41ef2b
--- /dev/null
+++ b/src/duckdb/src/include/duckdb/parser/dialect_extension.hpp
@@ -0,0 +1,27 @@
+//===----------------------------------------------------------------------===//
+//                         DuckDB
+//
+// duckdb/parser/dialect_extension.hpp
+//
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include "duckdb/common/common.hpp"
+
+namespace duckdb {
+struct DBConfig;
+
+//! A named SQL dialect that can customize the PEG parser.
+class DialectExtension {
+public:
+	explicit DialectExtension(string name_p) : name(std::move(name_p)) {
+	}
+
+	string name;
+
+	static void Register(DBConfig &config, DialectExtension extension);
+};
+
+} // namespace duckdb
diff --git a/src/duckdb/src/include/duckdb/parser/peg/matcher.hpp b/src/duckdb/src/include/duckdb/parser/peg/matcher.hpp
index dd504bb9c..6839dcba5 100644
--- a/src/duckdb/src/include/duckdb/parser/peg/matcher.hpp
+++ b/src/duckdb/src/include/duckdb/parser/peg/matcher.hpp
@@ -230,6 +230,19 @@ class Matcher {
 	bool packrat_memoized = false;
 };
 
+class KeywordInfo {
+public:
+	KeywordInfo() {
+	}
+	explicit KeywordInfo(int32_t score_bonus, char extra_char = ' ')
+	    : score_bonus(score_bonus), extra_char(extra_char) {
+	}
+
+public:
+	int32_t score_bonus = 0;
+	char extra_char = '\0';
+};
+
 class MatcherAllocator {
 public:
 	Matcher &Allocate(unique_ptr<Matcher> matcher);
diff --git a/src/duckdb/src/include/duckdb/storage/optimistic_data_writer.hpp b/src/duckdb/src/include/duckdb/storage/optimistic_data_writer.hpp
index 42566ec72..d5ea54be8 100644
--- a/src/duckdb/src/include/duckdb/storage/optimistic_data_writer.hpp
+++ b/src/duckdb/src/include/duckdb/storage/optimistic_data_writer.hpp
@@ -52,6 +52,9 @@ class OptimisticDataWriter {
 	//! Rollback
 	void Rollback();
 
+	//! Whether this writer can write to disk at all (not temporary / in-memory / read-only)
+	bool CanWriteToDisk() const;
+
 	//! Return the client context.
 	ClientContext &GetClientContext() {
 		return context;
diff --git a/src/duckdb/src/include/duckdb/transaction/duck_transaction.hpp b/src/duckdb/src/include/duckdb/transaction/duck_transaction.hpp
index 67cbde1f3..d756b3d9b 100644
--- a/src/duckdb/src/include/duckdb/transaction/duck_transaction.hpp
+++ b/src/duckdb/src/include/duckdb/transaction/duck_transaction.hpp
@@ -64,6 +64,7 @@ class DuckTransaction : public Transaction {
 	void SetModifications(DatabaseModificationType type) override;
 
 	bool ShouldWriteToWAL(AttachedDatabase &db);
+	ErrorData PreFlushOptimisticBlocks(AttachedDatabase &db) noexcept;
 	ErrorData WriteToWAL(ClientContext &context, AttachedDatabase &db,
 	                     unique_ptr<StorageCommitState> &commit_state) noexcept;
 	//! Commit the current transaction with the given commit identifier. Returns an error message if the transaction
diff --git a/src/duckdb/src/include/duckdb/transaction/local_storage.hpp b/src/duckdb/src/include/duckdb/transaction/local_storage.hpp
index 388307fdf..67c439d6a 100644
--- a/src/duckdb/src/include/duckdb/transaction/local_storage.hpp
+++ b/src/duckdb/src/include/duckdb/transaction/local_storage.hpp
@@ -85,6 +85,14 @@ class LocalTableStorage : public enable_shared_from_this<LocalTableStorage> {
 	//! Write a new row group to disk (if possible)
 	void WriteNewRowGroup(idx_t flushed_row_group_idx);
 	void FlushBlocks();
+	//! Whether Flush() takes the bulk-append path for this storage: the append covers at least one
+	//! full row group and there are no deletes. Only depends on transaction-local state, i.e. this
+	//! can be decided before taking any locks.
+	bool IsBulkAppend() const;
+	//! Whether the optimistic writer of this storage writes to disk (not temporary / in-memory / read-only)
+	bool WritesToDisk() const;
+	//! Whether this storage holds optimistically written (flushed) row groups
+	bool HasFlushedRowGroups() const;
 	void Rollback();
 	idx_t EstimatedSize();
 
@@ -115,6 +123,7 @@ class LocalTableManager {
 public:
 	shared_ptr<LocalTableStorage> MoveEntry(DataTable &table);
 	reference_map_t<DataTable, shared_ptr<LocalTableStorage>> MoveEntries();
+	vector<shared_ptr<LocalTableStorage>> GetEntries() const;
 	optional_ptr<LocalTableStorage> GetStorage(DataTable &table) const;
 	LocalTableStorage &GetOrCreateStorage(ClientContext &context, DataTable &table);
 	idx_t EstimatedSize() const;
@@ -213,10 +222,16 @@ class LocalStorage {
 		return context;
 	}
 
+	void FlushBulkAppendBlocksAndSync(AttachedDatabase &db);
+	bool SyncedFlushedBlocks() const {
+		return synced_flushed_blocks;
+	}
+
 private:
 	ClientContext &context;
 	DuckTransaction &transaction;
 	LocalTableManager table_manager;
+	bool synced_flushed_blocks = false;
 
 private:
 	void Flush(DataTable &table, LocalTableStorage &storage, optional_ptr<StorageCommitState> commit_state);
diff --git a/src/duckdb/src/main/config.cpp b/src/duckdb/src/main/config.cpp
index 1f525f719..d6965c526 100644
--- a/src/duckdb/src/main/config.cpp
+++ b/src/duckdb/src/main/config.cpp
@@ -97,6 +97,7 @@ static const ConfigurationOption internal_options[] = {
     DUCKDB_SETTING_CALLBACK(CheckpointOnDetachSetting),
     DUCKDB_GLOBAL(CheckpointThresholdSetting),
     DUCKDB_LOCAL(ConfigureProfilingSetting),
+    DUCKDB_SETTING_CALLBACK(CurrentDialectSetting),
     DUCKDB_SETTING_CALLBACK(CurrentTransactionInvalidationPolicySetting),
     DUCKDB_SETTING(CustomExtensionRepositorySetting),
     DUCKDB_GLOBAL(CustomUserAgentSetting),
@@ -250,12 +251,12 @@ static const ConfigurationOption internal_options[] = {
 
 static const ConfigurationAlias setting_aliases[] = {DUCKDB_SETTING_ALIAS("configure_metrics", 30),
                                                      DUCKDB_SETTING_ALIAS("custom_profiling_settings", 30),
-                                                     DUCKDB_SETTING_ALIAS("memory_limit", 129),
-                                                     DUCKDB_SETTING_ALIAS("null_order", 61),
-                                                     DUCKDB_SETTING_ALIAS("profile_output", 152),
-                                                     DUCKDB_SETTING_ALIAS("user", 171),
+                                                     DUCKDB_SETTING_ALIAS("memory_limit", 130),
+                                                     DUCKDB_SETTING_ALIAS("null_order", 62),
+                                                     DUCKDB_SETTING_ALIAS("profile_output", 153),
+                                                     DUCKDB_SETTING_ALIAS("user", 172),
                                                      DUCKDB_SETTING_ALIAS("wal_autocheckpoint", 29),
-                                                     DUCKDB_SETTING_ALIAS("worker_threads", 169),
+                                                     DUCKDB_SETTING_ALIAS("worker_threads", 170),
                                                      FINAL_ALIAS};
 
 vector<ConfigurationOption> DBConfig::GetOptions() {
diff --git a/src/duckdb/src/main/database_manager.cpp b/src/duckdb/src/main/database_manager.cpp
index f52524155..4b6ffe4a7 100644
--- a/src/duckdb/src/main/database_manager.cpp
+++ b/src/duckdb/src/main/database_manager.cpp
@@ -345,6 +345,9 @@ shared_ptr<AttachedDatabase> DatabaseManager::DetachInternal(const Identifier &n
 		}
 		attached_db = std::move(entry->second);
 		databases.erase(entry);
+		if (name == default_database) {
+			default_database = databases.empty() ? Identifier() : databases.begin()->first;
+		}
 	}
 	if (attached_db && attached_db->GetCatalog().Supports(RemoteCapability::IS_REMOTE)) {
 		--remote_catalog_count;
diff --git a/src/duckdb/src/main/extension_callback_manager.cpp b/src/duckdb/src/main/extension_callback_manager.cpp
index b095cbc84..7095fb695 100644
--- a/src/duckdb/src/main/extension_callback_manager.cpp
+++ b/src/duckdb/src/main/extension_callback_manager.cpp
@@ -1,5 +1,6 @@
 #include "duckdb/main/extension_callback_manager.hpp"
 #include "duckdb/parser/parser_extension.hpp"
+#include "duckdb/parser/dialect_extension.hpp"
 #include "duckdb/optimizer/optimizer_extension.hpp"
 #include "duckdb/planner/operator_extension.hpp"
 #include "duckdb/planner/planner_extension.hpp"
@@ -11,6 +12,8 @@
 namespace duckdb {
 
 struct ExtensionCallbackRegistry {
+	//! SQL dialects made available to the PEG parser
+	vector<DialectExtension> dialect_extensions;
 	//! Extensions made to the parser
 	vector<ParserExtension> parser_extensions;
 	//! Extensions made to the planner
@@ -40,6 +43,7 @@ ExtensionCallbackManager &ExtensionCallbackManager::Get(DatabaseInstance &db) {
 }
 
 ExtensionCallbackManager::ExtensionCallbackManager() : callback_registry(make_shared_ptr<ExtensionCallbackRegistry>()) {
+	callback_registry->dialect_extensions.emplace_back("duckdb");
 }
 ExtensionCallbackManager::~ExtensionCallbackManager() {
 }
@@ -59,6 +63,21 @@ void ExtensionCallbackManager::Register(ParserExtension extension) {
 	callback_registry.atomic_store(new_registry);
 }
 
+void ExtensionCallbackManager::Register(DialectExtension extension) {
+	if (extension.name.empty()) {
+		throw InvalidInputException("Dialect name cannot be empty");
+	}
+	lock_guard<mutex> guard(registry_lock);
+	auto new_registry = make_shared_ptr<ExtensionCallbackRegistry>(*callback_registry);
+	for (auto &existing : new_registry->dialect_extensions) {
+		if (StringUtil::CIEquals(existing.name, extension.name)) {
+			throw InvalidInputException("Dialect \"%s\" is already registered", extension.name);
+		}
+	}
+	new_registry->dialect_extensions.push_back(std::move(extension));
+	callback_registry.atomic_store(new_registry);
+}
+
 void ExtensionCallbackManager::Register(PlannerExtension extension) {
 	lock_guard<mutex> guard(registry_lock);
 	auto new_registry = make_shared_ptr<ExtensionCallbackRegistry>(*callback_registry);
@@ -129,6 +148,12 @@ ExtensionCallbackIteratorHelper<ParserExtension> ExtensionCallbackManager::Parse
 	return ExtensionCallbackIteratorHelper<ParserExtension>(parser_extensions, std::move(registry));
 }
 
+ExtensionCallbackIteratorHelper<DialectExtension> ExtensionCallbackManager::DialectExtensions() const {
+	auto registry = callback_registry.atomic_load();
+	auto &dialect_extensions = registry->dialect_extensions;
+	return ExtensionCallbackIteratorHelper<DialectExtension>(dialect_extensions, std::move(registry));
+}
+
 ExtensionCallbackIteratorHelper<PlannerExtension> ExtensionCallbackManager::PlannerExtensions() const {
 	auto registry = callback_registry.atomic_load();
 	auto &planner_extensions = registry->planner_extensions;
@@ -164,6 +189,16 @@ bool ExtensionCallbackManager::HasParserExtensions() const {
 	return !registry->parser_extensions.empty();
 }
 
+bool ExtensionCallbackManager::HasDialectExtension(const string &name) const {
+	auto registry = callback_registry.atomic_load();
+	for (auto &dialect : registry->dialect_extensions) {
+		if (StringUtil::CIEquals(dialect.name, name)) {
+			return true;
+		}
+	}
+	return false;
+}
+
 void OptimizerExtension::Register(DBConfig &config, OptimizerExtension extension) {
 	config.GetCallbackManager().Register(std::move(extension));
 }
@@ -172,6 +207,10 @@ void ParserExtension::Register(DBConfig &config, ParserExtension extension) {
 	config.GetCallbackManager().Register(std::move(extension));
 }
 
+void DialectExtension::Register(DBConfig &config, DialectExtension extension) {
+	config.GetCallbackManager().Register(std::move(extension));
+}
+
 void PlannerExtension::Register(DBConfig &config, PlannerExtension extension) {
 	config.GetCallbackManager().Register(std::move(extension));
 }
@@ -205,6 +244,7 @@ template class ExtensionCallbackIteratorHelper<shared_ptr<ExtensionCallback>>;
 template class ExtensionCallbackIteratorHelper<shared_ptr<OperatorExtension>>;
 template class ExtensionCallbackIteratorHelper<OptimizerExtension>;
 template class ExtensionCallbackIteratorHelper<ParserExtension>;
+template class ExtensionCallbackIteratorHelper<DialectExtension>;
 template class ExtensionCallbackIteratorHelper<PlannerExtension>;
 
 } // namespace duckdb
diff --git a/src/duckdb/src/main/settings/custom_settings.cpp b/src/duckdb/src/main/settings/custom_settings.cpp
index 21b0b360f..b1b969799 100644
--- a/src/duckdb/src/main/settings/custom_settings.cpp
+++ b/src/duckdb/src/main/settings/custom_settings.cpp
@@ -27,10 +27,12 @@
 #include "duckdb/main/database_manager.hpp"
 #include "duckdb/common/tree_renderer.hpp"
 #include "duckdb/main/extension_helper.hpp"
+#include "duckdb/main/extension_callback_manager.hpp"
 #include "duckdb/main/query_profiler.hpp"
 #include "duckdb/main/secret/secret_manager.hpp"
 #include "duckdb/parallel/task_scheduler.hpp"
 #include "duckdb/parser/parser.hpp"
+#include "duckdb/parser/peg/matcher.hpp"
 #include "duckdb/planner/expression_binder.hpp"
 #include "duckdb/storage/external_file_cache/external_file_cache.hpp"
 #include "duckdb/storage/buffer/buffer_pool.hpp"
@@ -1648,4 +1650,17 @@ void CurrentTransactionInvalidationPolicySetting::OnSet(SettingCallbackInfo &inf
 	info.context->transaction.SetInvalidationPolicy(
 	    EnumUtil::FromString<TransactionInvalidationPolicy>(input.GetValue<string>()));
 }
+
+void CurrentDialectSetting::OnSet(SettingCallbackInfo &info, Value &input) {
+	if (input.IsNull()) {
+		throw InvalidInputException("current_dialect setting cannot be NULL");
+	}
+	auto dialect_name = input.GetValue<string>();
+	if (!info.config.GetCallbackManager().HasDialectExtension(dialect_name)) {
+		throw InvalidInputException("Dialect \"%s\" is not installed", dialect_name);
+	}
+	if (info.db) {
+		info.db->GetParserCache().Invalidate();
+	}
+}
 } // namespace duckdb
diff --git a/src/duckdb/src/optimizer/common_subplan_optimizer.cpp b/src/duckdb/src/optimizer/common_subplan_optimizer.cpp
index c842a755a..47c8a6b04 100644
--- a/src/duckdb/src/optimizer/common_subplan_optimizer.cpp
+++ b/src/duckdb/src/optimizer/common_subplan_optimizer.cpp
@@ -568,7 +568,6 @@ class PlanSignature {
 		case LogicalOperatorType::LOGICAL_TOP_N:
 		case LogicalOperatorType::LOGICAL_DISTINCT:
 		case LogicalOperatorType::LOGICAL_PIVOT:
-		case LogicalOperatorType::LOGICAL_GET:
 		case LogicalOperatorType::LOGICAL_EXPRESSION_GET:
 		case LogicalOperatorType::LOGICAL_DUMMY_SCAN:
 		case LogicalOperatorType::LOGICAL_COMPARISON_JOIN:
@@ -580,6 +579,18 @@ class PlanSignature {
 		case LogicalOperatorType::LOGICAL_EXCEPT:
 		case LogicalOperatorType::LOGICAL_INTERSECT:
 			return true;
+		case LogicalOperatorType::LOGICAL_GET: {
+			auto &get = op.Cast<LogicalGet>();
+			if (get.bind_data && !get.function.HasSerializationCallbacks() && get.parameters.empty() &&
+			    get.named_parameters.empty()) {
+				// Without serialization callbacks, the serialized form carries only the call parameters
+				// (see LogicalGet::Serialize). A parameter-less scan - e.g., one created through an
+				// attached catalog - keeps its identity solely in the bind data, so equal serialized
+				// bytes cannot prove that two scans read the same table.
+				return false;
+			}
+			return true;
+		}
 		case LogicalOperatorType::LOGICAL_CHUNK_GET:
 			// Avoid serializing massive amounts of data (this is here because of the "Test TPCH arrow roundtrip" test)
 			return op.Cast<LogicalColumnDataGet>().collection->Count() < 1000;
diff --git a/src/duckdb/src/parser/peg/matcher.cpp b/src/duckdb/src/parser/peg/matcher.cpp
index bf74ec4d1..ca427144f 100644
--- a/src/duckdb/src/parser/peg/matcher.cpp
+++ b/src/duckdb/src/parser/peg/matcher.cpp
@@ -6,6 +6,7 @@
 // #define PEG_PARSER_SOURCE_FILE "duckdb/parser/peg/inlined_grammar.gram"
 
 #include "duckdb/common/printer.hpp"
+#include "duckdb/common/optional.hpp"
 #include "duckdb/common/string_map_set.hpp"
 #include "duckdb/common/types/string_type.hpp"
 #include "duckdb/parser/peg/keyword_helper.hpp"
@@ -58,8 +59,8 @@ class KeywordMatcher : public Matcher {
 	static constexpr MatcherType TYPE = MatcherType::KEYWORD;
 
 public:
-	explicit KeywordMatcher(string keyword_p, int32_t score_bonus = 0, char extra_char = '\0')
-	    : Matcher(TYPE), keyword(std::move(keyword_p)), score_bonus(score_bonus), extra_char(extra_char) {
+	explicit KeywordMatcher(string keyword_p, const KeywordInfo &info)
+	    : Matcher(TYPE), keyword(std::move(keyword_p)), info(info) {
 	}
 
 	MatchResultType Match(MatchState &state) const override {
@@ -85,8 +86,9 @@ class KeywordMatcher : public Matcher {
 	}
 
 	SuggestionType AddSuggestionInternal(MatchState &state) const override {
-		AutoCompleteCandidate candidate(keyword, SuggestionState::SUGGEST_KEYWORD, score_bonus, CandidateType::KEYWORD);
-		candidate.extra_char = extra_char;
+		AutoCompleteCandidate candidate(keyword, SuggestionState::SUGGEST_KEYWORD, info.score_bonus,
+		                                CandidateType::KEYWORD);
+		candidate.extra_char = info.extra_char;
 		state.AddSuggestion(MatcherSuggestion(std::move(candidate)));
 		return SuggestionType::MANDATORY;
 	}
@@ -111,9 +113,8 @@ class KeywordMatcher : public Matcher {
 	}
 
 private:
-	string keyword;
-	int32_t score_bonus;
-	char extra_char;
+	const string keyword;
+	const KeywordInfo info;
 };
 
 class ListMatcher : public Matcher {
@@ -288,7 +289,7 @@ class ChoiceMatcher : public Matcher {
 public:
 	ChoiceMatcher() : Matcher(TYPE) {
 	}
-	explicit ChoiceMatcher(vector<reference<Matcher>> matchers_p) : Matcher(TYPE), matchers(std::move(matchers_p)) {
+	explicit ChoiceMatcher(vector<reference<Matcher>> &&matchers_p) : Matcher(TYPE), matchers(std::move(matchers_p)) {
 	}
 
 	MatchResultType Match(MatchState &state) const override {
@@ -581,13 +582,15 @@ class IdentifierMatcher : public Matcher {
 		}
 	}
 
-	PEGKeywordCategory GetBannedCategory() const {
+	PEGKeywordCategory GetAllowedCategory() const {
 		switch (suggestion_type) {
+		case SuggestionState::SUGGEST_TYPE_NAME:
+			return PEGKeywordCategory::KEYWORD_TYPE_NAME;
 		case SuggestionState::SUGGEST_SCALAR_FUNCTION_NAME:
 		case SuggestionState::SUGGEST_TABLE_FUNCTION_NAME:
-			return PEGKeywordCategory::KEYWORD_COL_NAME;
-		default:
 			return PEGKeywordCategory::KEYWORD_TYPE_FUNC;
+		default:
+			return PEGKeywordCategory::KEYWORD_COL_NAME;
 		}
 	}
 
@@ -628,48 +631,23 @@ class IdentifierMatcher : public Matcher {
 	}
 
 private:
+	bool IsAllowedKeyword(const string &token_text) const {
+		auto &keyword_helper = PEGKeywordHelper::Instance();
+		if (!keyword_helper.IsKeyword(token_text)) {
+			return true;
+		}
+		if (keyword_helper.KeywordCategoryType(token_text, PEGKeywordCategory::KEYWORD_UNRESERVED)) {
+			return true;
+		}
+		return keyword_helper.KeywordCategoryType(token_text, GetAllowedCategory());
+	}
+
 	bool MatchIdentifier(MatchState &state) const {
 		if (state.token_index >= state.tokens.size()) {
 			return false;
 		}
-		// variable matchers match anything except for reserved keywords
 		auto &token_text = state.tokens[state.token_index].text;
-		const auto &keyword_helper = PEGKeywordHelper::Instance();
-		switch (suggestion_type) {
-		case SuggestionState::SUGGEST_TYPE_NAME:
-			if (keyword_helper.KeywordCategoryType(token_text, PEGKeywordCategory::KEYWORD_UNRESERVED) ||
-			    keyword_helper.KeywordCategoryType(token_text, PEGKeywordCategory::KEYWORD_TYPE_NAME)) {
-				break;
-			}
-			if (keyword_helper.KeywordCategoryType(token_text, PEGKeywordCategory::KEYWORD_RESERVED) ||
-			    keyword_helper.KeywordCategoryType(token_text, PEGKeywordCategory::KEYWORD_TYPE_FUNC) ||
-			    keyword_helper.KeywordCategoryType(token_text, PEGKeywordCategory::KEYWORD_COL_NAME)) {
-				return false;
-			}
-			break;
-		default: {
-			const auto banned_category = GetBannedCategory();
-			const auto allowed_override_category = banned_category == PEGKeywordCategory::KEYWORD_COL_NAME
-			                                           ? PEGKeywordCategory::KEYWORD_TYPE_FUNC
-			                                           : PEGKeywordCategory::KEYWORD_COL_NAME;
-
-			const bool is_reserved =
-			    keyword_helper.KeywordCategoryType(token_text, PEGKeywordCategory::KEYWORD_RESERVED);
-			const bool has_extra_banned_category = keyword_helper.KeywordCategoryType(token_text, banned_category);
-			const bool has_banned_flag = is_reserved || has_extra_banned_category;
-
-			const bool is_unreserved =
-			    keyword_helper.KeywordCategoryType(token_text, PEGKeywordCategory::KEYWORD_UNRESERVED);
-			const bool has_override_flag = keyword_helper.KeywordCategoryType(token_text, allowed_override_category);
-			const bool has_allowed_flag = is_unreserved || has_override_flag;
-
-			if (has_banned_flag && !has_allowed_flag) {
-				return false;
-			}
-			break;
-		}
-		}
-		if (!IsIdentifier(token_text)) {
+		if (!IsAllowedKeyword(token_text) || !IsIdentifier(token_text)) {
 			return false;
 		}
 		state.token_index++;
@@ -1060,12 +1038,15 @@ optional_ptr<ParseResult> ParseResultAllocator::Allocate(unique_ptr<ParseResult>
 
 //! Class for building matchers
 class MatcherFactory {
+public:
 	friend struct MatcherList;
 
 public:
 	explicit MatcherFactory(MatcherAllocator &allocator) : allocator(allocator) {
 	}
+	virtual ~MatcherFactory() = default;
 
+public:
 	//! Create a matcher from a PEG grammar
 	Matcher &CreateMatcher(const char *grammar, const char *root_rule);
 	//! Look up a matcher for a rule that was already built (as a sub-rule of a previous
@@ -1074,14 +1055,20 @@ class MatcherFactory {
 
 private:
 	// Base primitives
-	Matcher &Keyword(const string &keyword) const;
-	Matcher &List() const;
-	Matcher &List(vector<reference<Matcher>> matchers) const;
-	Matcher &Choice(vector<reference<Matcher>> matchers) const;
-	Matcher &Optional(Matcher &matcher) const;
-	Matcher &Repeat(Matcher &matcher) const;
-
-	void AddKeywordOverride(const char *name, int32_t score, char extra_char = ' ');
+	KeywordMatcher &Keyword(const string &keyword) const;
+	ListMatcher &List() const;
+	ListMatcher &List(vector<reference<Matcher>> matchers) const;
+	ChoiceMatcher &Choice(vector<reference<Matcher>> &&matchers) const;
+	OptionalMatcher &Optional(Matcher &matcher) const;
+	RepeatMatcher &Repeat(Matcher &matcher) const;
+
+	virtual unique_ptr<KeywordMatcher> CreateKeyword(const string &keyword, const KeywordInfo &info) const;
+	virtual unique_ptr<ListMatcher> CreateList() const;
+	virtual unique_ptr<ChoiceMatcher> CreateChoice(vector<reference<Matcher>> &&matchers) const;
+	virtual unique_ptr<OptionalMatcher> CreateOptional(Matcher &matcher) const;
+	virtual unique_ptr<RepeatMatcher> CreateRepeat(Matcher &matcher) const;
+
+	void AddKeywordOverride(const char *name, KeywordInfo keyword_info);
 	void AddRuleOverride(const char *name, Matcher &matcher);
 	void AddPackratMemoizedRule(const char *name);
 	void SuppressSuggestions(const char *name);
@@ -1091,37 +1078,70 @@ class MatcherFactory {
 private:
 	MatcherAllocator &allocator;
 	string_map_t<reference<Matcher>> matchers;
-	case_insensitive_map_t<reference<Matcher>> keyword_overrides;
+	mutable case_insensitive_map_t<reference<KeywordMatcher>> keywords;
+	case_insensitive_map_t<KeywordInfo> keyword_overrides;
 	string_set_t no_suggestion_rules;
 	string_set_t packrat_memoized_rules;
 };
 
-Matcher &MatcherFactory::Keyword(const string &keyword) const {
+unique_ptr<KeywordMatcher> MatcherFactory::CreateKeyword(const string &keyword, const KeywordInfo &info) const {
+	return make_uniq<KeywordMatcher>(keyword, info);
+}
+
+unique_ptr<ListMatcher> MatcherFactory::CreateList() const {
+	return make_uniq<ListMatcher>();
+}
+
+unique_ptr<ChoiceMatcher> MatcherFactory::CreateChoice(vector<reference<Matcher>> &&matchers) const {
+	return make_uniq<ChoiceMatcher>(std::move(matchers));
+}
+
+unique_ptr<OptionalMatcher> MatcherFactory::CreateOptional(Matcher &matcher) const {
+	return make_uniq<OptionalMatcher>(matcher);
+}
+
+unique_ptr<RepeatMatcher> MatcherFactory::CreateRepeat(Matcher &matcher) const {
+	return make_uniq<RepeatMatcher>(matcher);
+}
+
+KeywordMatcher &MatcherFactory::Keyword(const string &keyword) const {
+	auto it = keywords.find(keyword);
+	if (it != keywords.end()) {
+		return it->second;
+	}
+
+	optional<KeywordInfo> info;
 	auto entry = keyword_overrides.find(keyword);
 	if (entry != keyword_overrides.end()) {
-		return entry->second.get();
+		info.emplace(entry->second);
+	} else {
+		info.emplace(0, ' ');
 	}
-	return allocator.Allocate(make_uniq<KeywordMatcher>(keyword, 0, ' '));
+	auto &result = allocator.Allocate(CreateKeyword(keyword, *info)).Cast<KeywordMatcher>();
+	keywords.emplace(keyword, result);
+	return result;
 }
 
-Matcher &MatcherFactory::List() const {
-	return allocator.Allocate(make_uniq<ListMatcher>());
+ListMatcher &MatcherFactory::List() const {
+	return allocator.Allocate(CreateList()).Cast<ListMatcher>();
 }
 
-Matcher &MatcherFactory::List(vector<reference<Matcher>> matchers) const {
-	return allocator.Allocate(make_uniq<ListMatcher>(std::move(matchers)));
+ListMatcher &MatcherFactory::List(vector<reference<Matcher>> matchers) const {
+	auto result = CreateList();
+	result->matchers = std::move(matchers);
+	return allocator.Allocate(std::move(result)).Cast<ListMatcher>();
 }
 
-Matcher &MatcherFactory::Choice(vector<reference<Matcher>> matchers) const {
-	return allocator.Allocate(make_uniq<ChoiceMatcher>(std::move(matchers)));
+ChoiceMatcher &MatcherFactory::Choice(vector<reference<Matcher>> &&matchers) const {
+	return allocator.Allocate(CreateChoice(std::move(matchers))).Cast<ChoiceMatcher>();
 }
 
-Matcher &MatcherFactory::Optional(Matcher &matcher) const {
-	return allocator.Allocate(make_uniq<OptionalMatcher>(matcher));
+OptionalMatcher &MatcherFactory::Optional(Matcher &matcher) const {
+	return allocator.Allocate(CreateOptional(matcher)).Cast<OptionalMatcher>();
 }
 
-Matcher &MatcherFactory::Repeat(Matcher &matcher) const {
-	return allocator.Allocate(make_uniq<RepeatMatcher>(matcher));
+RepeatMatcher &MatcherFactory::Repeat(Matcher &matcher) const {
+	return allocator.Allocate(CreateRepeat(matcher)).Cast<RepeatMatcher>();
 }
 
 Matcher &MatcherFactory::GetMatcher(const string &rule_name) {
@@ -1343,7 +1363,7 @@ Matcher &MatcherFactory::CreateMatcher(PEGParser &parser, string_t rule_name, ve
 				} else {
 					vector<reference<Matcher>> choice_options;
 					choice_options.push_back(previous_matcher);
-					auto &new_choice_matcher = Choice(choice_options);
+					auto &new_choice_matcher = Choice(std::move(choice_options));
 
 					if (!list_matcher.matchers.empty()) {
 						list_matcher.matchers.pop_back();
@@ -1395,9 +1415,8 @@ Matcher &MatcherFactory::CreateMatcher(PEGParser &parser, string_t rule_name, ve
 	return matcher;
 }
 
-void MatcherFactory::AddKeywordOverride(const char *name, int32_t score, char extra_char) {
-	auto &keyword_matcher = allocator.Allocate(make_uniq<KeywordMatcher>(name, score, extra_char));
-	keyword_overrides.insert(make_pair(name, reference<Matcher>(keyword_matcher)));
+void MatcherFactory::AddKeywordOverride(const char *name, KeywordInfo info) {
+	keyword_overrides.insert(make_pair(name, info));
 }
 
 void MatcherFactory::AddRuleOverride(const char *name, Matcher &matcher) {
@@ -1421,9 +1440,9 @@ Matcher &MatcherFactory::CreateMatcher(const char *grammar, const char *root_rul
 	parser.ParseRules(grammar);
 
 	// keyword overrides
-	AddKeywordOverride("TABLE", 1, ' ');
-	AddKeywordOverride(".", 0, '\0');
-	AddKeywordOverride("(", 0, '\0');
+	AddKeywordOverride("TABLE", KeywordInfo(1, ' '));
+	AddKeywordOverride(".", KeywordInfo(0, '\0'));
+	AddKeywordOverride("(", KeywordInfo(0, '\0'));
 	// packrat memoized rules
 	//===--------------------------------------------------------------------===//
 	// START GENERATED PACKRAT MEMOIZED RULES
diff --git a/src/duckdb/src/storage/local_storage.cpp b/src/duckdb/src/storage/local_storage.cpp
index 4a61d9aab..647513b9c 100644
--- a/src/duckdb/src/storage/local_storage.cpp
+++ b/src/duckdb/src/storage/local_storage.cpp
@@ -2,9 +2,11 @@
 #include "duckdb/transaction/commit_state.hpp"
 
 #include "duckdb/catalog/catalog_entry/duck_table_entry.hpp"
+#include "duckdb/main/attached_database.hpp"
 #include "duckdb/planner/table_filter.hpp"
 #include "duckdb/storage/data_table.hpp"
 #include "duckdb/storage/partial_block_manager.hpp"
+#include "duckdb/storage/storage_manager.hpp"
 #include "duckdb/storage/table/append_state.hpp"
 #include "duckdb/storage/table/data_table_info.hpp"
 #include "duckdb/storage/table/row_group.hpp"
@@ -157,6 +159,22 @@ void LocalTableStorage::FlushBlocks() {
 	optimistic_writer.FinalFlush();
 }
 
+bool LocalTableStorage::WritesToDisk() const {
+	return optimistic_writer.CanWriteToDisk();
+}
+
+bool LocalTableStorage::HasFlushedRowGroups() const {
+	return !row_groups->flushed_row_groups.empty();
+}
+
+bool LocalTableStorage::IsBulkAppend() const {
+	if (is_dropped || deleted_rows != 0) {
+		return false;
+	}
+	auto &collection = *row_groups->collection;
+	return collection.GetTotalRows() >= collection.GetRowGroupSize();
+}
+
 ErrorData LocalTableStorage::AppendToIndexes(DuckTransaction &transaction, RowGroupCollection &source,
                                              TableIndexList &index_list, const vector<LogicalType> &table_types,
                                              row_t &start_row) {
@@ -327,6 +345,16 @@ reference_map_t<DataTable, shared_ptr<LocalTableStorage>> LocalTableManager::Mov
 	return std::move(table_storage);
 }
 
+vector<shared_ptr<LocalTableStorage>> LocalTableManager::GetEntries() const {
+	lock_guard<mutex> l(table_storage_lock);
+	vector<shared_ptr<LocalTableStorage>> result;
+	result.reserve(table_storage.size());
+	for (auto &entry : table_storage) {
+		result.push_back(entry.second);
+	}
+	return result;
+}
+
 idx_t LocalTableManager::EstimatedSize() const {
 	lock_guard<mutex> l(table_storage_lock);
 	idx_t estimated_size = 0;
@@ -577,13 +605,16 @@ void LocalStorage::Flush(DataTable &table, LocalTableStorage &storage, optional_
 	}
 
 	auto append_count = storage.GetCollection().GetTotalRows() - storage.deleted_rows;
-	const auto row_group_size = storage.GetCollection().GetRowGroupSize();
 
 	TableAppendState append_state;
 	table.AppendLock(transaction, append_state);
-	if ((append_state.row_start == 0 || storage.GetCollection().GetTotalRows() >= row_group_size) &&
-	    storage.deleted_rows == 0) {
-		// table is currently empty OR we are bulk appending: move over the storage directly
+	if (storage.IsBulkAppend() ||
+	    (append_state.row_start == 0 && storage.deleted_rows == 0 && !storage.WritesToDisk())) {
+		// bulk append (at least one full row group, no deletes): move over the storage directly.
+		// Appends to an empty table are also merged directly if the table cannot be written to
+		// disk (temporary / in-memory / read-only, e.g. WAL replay of a read-only attach) -
+		// there are no optimistically written blocks to manage, and merging avoids re-appending
+		// row by row.
 		// first flush any outstanding blocks
 		storage.FlushBlocks();
 		// Append to the indexes.
@@ -594,6 +625,9 @@ void LocalStorage::Flush(DataTable &table, LocalTableStorage &storage, optional_
 		// check if we have written data
 		// if we have, we cannot merge to disk after all
 		// so we need to revert the data we have already written
+		// this only happens for transactions that deleted rows after bulk-appending: a pure bulk
+		// append always takes the merge path above, using its pre-flushed blocks as written
+		D_ASSERT(!storage.HasFlushedRowGroups() || storage.deleted_rows > 0);
 		storage.Rollback();
 		// append to the indexes
 		storage.AppendToIndexes(transaction, append_state);
@@ -610,6 +644,24 @@ void LocalStorage::Flush(DataTable &table, LocalTableStorage &storage, optional_
 #endif
 }
 
+void LocalStorage::FlushBulkAppendBlocksAndSync(AttachedDatabase &db) {
+	bool requires_sync = false;
+	for (auto &storage : table_manager.GetEntries()) {
+		if (storage->IsBulkAppend()) {
+			// Flush() is guaranteed to take the bulk path - the blocks will be used as written
+			storage->FlushBlocks();
+			requires_sync |= storage->HasFlushedRowGroups();
+		}
+	}
+	if (requires_sync) {
+		// the WAL will reference the flushed row groups (flushed just now or already during the
+		// statement, e.g. by batch inserts) - persist them now so that the commit does not have
+		// to FileSync while holding the WAL lock
+		db.GetStorageManager().GetBlockManager().FileSync();
+		synced_flushed_blocks = true;
+	}
+}
+
 void LocalStorage::Commit(optional_ptr<StorageCommitState> commit_state) {
 	// commit local storage
 	// iterate over all entries in the table storage map and commit them
diff --git a/src/duckdb/src/storage/optimistic_data_writer.cpp b/src/duckdb/src/storage/optimistic_data_writer.cpp
index c610ed2d1..e155b2e10 100644
--- a/src/duckdb/src/storage/optimistic_data_writer.cpp
+++ b/src/duckdb/src/storage/optimistic_data_writer.cpp
@@ -25,15 +25,19 @@ OptimisticDataWriter::OptimisticDataWriter(DataTable &table, OptimisticDataWrite
 OptimisticDataWriter::~OptimisticDataWriter() {
 }
 
+bool OptimisticDataWriter::CanWriteToDisk() const {
+	auto &attached = table.GetAttached();
+	auto &storage_manager = StorageManager::Get(attached);
+	return !table.IsTemporary() && !storage_manager.InMemory() && !attached.IsReadOnly();
+}
+
 bool OptimisticDataWriter::PrepareWrite() {
 	// check if optimistic writing is enabled
 	if (!Settings::Get<EnableOptimisticWriteSetting>(context)) {
 		return false;
 	}
 	// check if we should pre-emptively write the table to disk
-	auto &attached = table.GetAttached();
-	auto &storage_manager = StorageManager::Get(attached);
-	if (table.IsTemporary() || storage_manager.InMemory() || attached.IsReadOnly()) {
+	if (!CanWriteToDisk()) {
 		return false;
 	}
 	// we should! write the second-to-last row group to disk
@@ -118,12 +122,16 @@ void OptimisticWriteCollection::FinalizeFlush() {
 }
 
 void OptimisticDataWriter::WriteUnflushedRowGroups(OptimisticWriteCollection &row_groups) {
+	auto total_row_groups = row_groups.collection->GetRowGroupCount();
+	if (row_groups.flushed_row_groups.size() == total_row_groups && row_groups.partial_block_managers.empty()) {
+		// everything is flushed already and there is nothing to merge - a repeated call ends up here
+		return;
+	}
 	// we finished writing a complete row group
 	if (!PrepareWrite()) {
 		return;
 	}
 	// add any incomplete row groups to the set of unflushed row groups
-	auto total_row_groups = row_groups.collection->GetRowGroupCount();
 	for (idx_t i = 0; i < total_row_groups; i++) {
 		// check if this row group was flushed
 		auto entry = row_groups.flushed_row_groups.find(i);
diff --git a/src/duckdb/src/transaction/duck_transaction.cpp b/src/duckdb/src/transaction/duck_transaction.cpp
index f815357b2..7f940e510 100644
--- a/src/duckdb/src/transaction/duck_transaction.cpp
+++ b/src/duckdb/src/transaction/duck_transaction.cpp
@@ -206,6 +206,21 @@ bool DuckTransaction::ShouldWriteToWAL(AttachedDatabase &db) {
 	return true;
 }
 
+ErrorData DuckTransaction::PreFlushOptimisticBlocks(AttachedDatabase &db) noexcept {
+	ErrorData error;
+	if (!ShouldWriteToWAL(db)) {
+		return error;
+	}
+	try {
+		storage->FlushBulkAppendBlocksAndSync(db);
+	} catch (std::exception &ex) {
+		// fail the commit: the flush machinery cannot safely be re-run after an error, and a failed
+		// fsync must not be retried (the retry can succeed without the data being durable)
+		error = ErrorData(ex);
+	}
+	return error;
+}
+
 ErrorData DuckTransaction::WriteToWAL(ClientContext &context, AttachedDatabase &db,
                                       unique_ptr<StorageCommitState> &commit_state) noexcept {
 	ErrorData error_data;
@@ -222,14 +237,12 @@ ErrorData DuckTransaction::WriteToWAL(ClientContext &context, AttachedDatabase &
 
 		auto wal_timer = profiler.StartTimer<MetricStorageWriteToWALLatency>();
 		undo_buffer.WriteToWAL(*wal, commit_state.get());
-		if (commit_state->HasRowGroupData()) {
-			// if we have optimistically written any data AND we are writing to the WAL, we have written references to
-			// optimistically written blocks
-			// hence we need to ensure those optimistically written blocks are persisted
-			storage_manager.GetBlockManager().FileSync();
-		}
 		wal_timer.EndTimer();
 
+		// no FileSync is required here: any optimistically written blocks that the WAL references
+		// have already been synced by FlushBulkAppendBlocksAndSync, before the commit locks were taken
+		D_ASSERT(!commit_state->HasRowGroupData() || storage->SyncedFlushedBlocks());
+
 	} catch (std::exception &ex) {
 		// Call RevertCommit() outside this try-catch as it itself may throw
 		error_data = ErrorData(ex);
diff --git a/src/duckdb/src/transaction/duck_transaction_manager.cpp b/src/duckdb/src/transaction/duck_transaction_manager.cpp
index 25b15225d..390b9ff4a 100644
--- a/src/duckdb/src/transaction/duck_transaction_manager.cpp
+++ b/src/duckdb/src/transaction/duck_transaction_manager.cpp
@@ -297,6 +297,8 @@ void DuckTransactionManager::CleanupTransactions() {
 
 ErrorData DuckTransactionManager::CommitTransaction(ClientContext &context, Transaction &transaction_p) {
 	auto &transaction = transaction_p.Cast<DuckTransaction>();
+	// flush the transaction-local blocks of bulk appends before taking any commit locks (see PreFlushOptimisticBlocks)
+	ErrorData error = transaction.PreFlushOptimisticBlocks(db);
 	unique_lock<mutex> t_lock(transaction_lock);
 	if (!db.IsSystem() && !db.IsTemporary()) {
 		if (transaction.ChangesMade()) {
@@ -311,7 +313,6 @@ ErrorData DuckTransactionManager::CommitTransaction(ClientContext &context, Tran
 	unique_ptr<StorageLockKey> lock;
 	auto undo_properties = transaction.GetUndoProperties();
 	auto checkpoint_decision = CanCheckpoint(transaction, lock, undo_properties);
-	ErrorData error;
 	unique_lock<mutex> held_wal_lock;
 	unique_ptr<StorageCommitState> commit_state;
 	bool skip_wal_write_due_to_checkpoint = false;
@@ -329,7 +330,7 @@ ErrorData DuckTransactionManager::CommitTransaction(ClientContext &context, Tran
 			skip_wal_write_due_to_checkpoint = true;
 		}
 	}
-	bool should_write_to_wal = transaction.ShouldWriteToWAL(db);
+	bool should_write_to_wal = !error.HasError() && transaction.ShouldWriteToWAL(db);
 	if (should_write_to_wal) {
 		auto &storage_manager = db.GetStorageManager().Cast<SingleFileStorageManager>();
 		// if we are committing changes and we are not doing a "checkpoint instead of WAL write"
diff --git a/src/duckdb/ub_src_function_table_system.cpp b/src/duckdb/ub_src_function_table_system.cpp
index 46cebbc2f..14bd2a80e 100644
--- a/src/duckdb/ub_src_function_table_system.cpp
+++ b/src/duckdb/ub_src_function_table_system.cpp
@@ -14,6 +14,8 @@
 
 #include "src/function/table/system/duckdb_dependencies.cpp"
 
+#include "src/function/table/system/duckdb_dialects.cpp"
+
 #include "src/function/table/system/duckdb_eviction_queues.cpp"
 
 #include "src/function/table/system/duckdb_extensions.cpp"