fix: DeltaAppender cleanup on rollback

storage/duckdb/ha_duckdb.cc

@@ -187,14 +187,20 @@ static int duckdb_close_connection(handlerton *hton, THD *thd)
 
 static int duckdb_register_trx(THD *thd)
 {
+  auto *ctx= get_duckdb_context(thd);
+  std::string error_msg;
+  if (ctx->duckdb_trans_begin(error_msg))
+  {
+    my_error(ER_GET_ERRMSG, MYF(0), HA_ERR_GENERIC, error_msg.c_str(),
+             "DuckDB");
+    return HA_ERR_GENERIC;
+  }
+
   trans_register_ha(thd, false, duckdb_hton, 0);
 
   if (thd_test_options(thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
     trans_register_ha(thd, true, duckdb_hton, 0);
 
-  auto *ctx= get_duckdb_context(thd);
-  if (!ctx->has_transaction())
-    ctx->duckdb_trans_begin();
   return 0;
 }
 
@@ -209,7 +215,8 @@ static void duckdb_drop_database(handlerton *hton, char *path)
   query.append(db.name);
   query.append("\"");
 
-  duckdb_register_trx(thd);
+  if (duckdb_register_trx(thd))
+    DBUG_VOID_RETURN;
   auto *ctx= get_duckdb_context(thd);
   auto query_result= myduck::duckdb_query(ctx->get_connection(), query);
   DBUG_VOID_RETURN;

storage/duckdb/mysql-test/duckdb/r/batch_disconnect.result

@@ -0,0 +1,9 @@
+connection con1;
+BEGIN;
+INSERT INTO batch_disconnect VALUES (1);
+disconnect con1;
+connection default;
+INSERT INTO batch_disconnect VALUES (2);
+SELECT * FROM batch_disconnect ORDER BY id;
+id
+2

storage/duckdb/mysql-test/duckdb/r/batch_rollback.result

@@ -0,0 +1,9 @@
+BEGIN;
+INSERT INTO batch_rollback VALUES (1);
+ROLLBACK;
+BEGIN;
+INSERT INTO batch_rollback VALUES (2);
+COMMIT;
+SELECT * FROM batch_rollback ORDER BY id;
+id
+2

storage/duckdb/mysql-test/duckdb/t/batch_disconnect.test

@@ -0,0 +1,23 @@
+--source ../include/have_duckdb.inc
+
+--disable_query_log
+SET @saved_duckdb_dml_in_batch = @@GLOBAL.duckdb_dml_in_batch;
+SET GLOBAL duckdb_dml_in_batch = ON;
+DROP TABLE IF EXISTS batch_disconnect;
+CREATE TABLE batch_disconnect (id INT PRIMARY KEY) ENGINE=DuckDB;
+connect (con1,localhost,root,,test);
+--enable_query_log
+
+connection con1;
+BEGIN;
+INSERT INTO batch_disconnect VALUES (1);
+disconnect con1;
+
+connection default;
+INSERT INTO batch_disconnect VALUES (2);
+SELECT * FROM batch_disconnect ORDER BY id;
+
+--disable_query_log
+DROP TABLE batch_disconnect;
+SET GLOBAL duckdb_dml_in_batch = @saved_duckdb_dml_in_batch;
+--enable_query_log

storage/duckdb/mysql-test/duckdb/t/batch_rollback.test

@@ -0,0 +1,23 @@
+--source ../include/have_duckdb.inc
+
+--disable_query_log
+SET @saved_duckdb_dml_in_batch = @@GLOBAL.duckdb_dml_in_batch;
+SET GLOBAL duckdb_dml_in_batch = ON;
+DROP TABLE IF EXISTS batch_rollback;
+CREATE TABLE batch_rollback (id INT PRIMARY KEY) ENGINE=DuckDB;
+--enable_query_log
+
+BEGIN;
+INSERT INTO batch_rollback VALUES (1);
+ROLLBACK;
+
+BEGIN;
+INSERT INTO batch_rollback VALUES (2);
+COMMIT;
+
+SELECT * FROM batch_rollback ORDER BY id;
+
+--disable_query_log
+DROP TABLE batch_rollback;
+SET GLOBAL duckdb_dml_in_batch = @saved_duckdb_dml_in_batch;
+--enable_query_log

storage/duckdb/runtime/delta_appender.cc

@@ -541,6 +541,12 @@ bool DeltaAppender::flush(bool idempotent_flag)
   return false;
 }
 
+void DeltaAppender::discard()
+{
+  if (m_appender)
+    m_appender->Clear();
+}
+
 void DeltaAppender::cleanup()
 {
   if (m_use_tmp_table)
@@ -555,7 +561,18 @@ void DeltaAppender::cleanup()
 void DeltaAppenders::delete_appender(std::string &db, std::string &tb)
 {
   auto key= std::make_pair(db, tb);
-  m_append_infos.erase(key);
+  auto it= m_append_infos.find(key);
+  if (it == m_append_infos.end())
+    return;
+  it->second->discard();
+  m_append_infos.erase(it);
+}
+
+void DeltaAppenders::discard_all()
+{
+  for (auto &pair : m_append_infos)
+    pair.second->discard();
+  m_append_infos.clear();
 }
 
 bool DeltaAppenders::flush_all(bool idempotent_flag, std::string &error_msg)

storage/duckdb/runtime/delta_appender.h

@@ -58,6 +58,8 @@ class DeltaAppender
 
   bool flush(bool idempotent_flag);
 
+  void discard();
+
   void cleanup();
 
 private:
@@ -97,6 +99,8 @@ class DeltaAppenders
 
   bool flush_all(bool idempotent_flag, std::string &error_msg);
 
+  void discard_all();
+
   bool is_empty() { return m_append_infos.empty(); }
 
   DeltaAppender *get_appender(std::string &db, std::string &tb,

storage/duckdb/runtime/duckdb_context.h

@@ -50,21 +50,29 @@ class DuckdbThdContext
 
   ~DuckdbThdContext()
   {
-    if (has_transaction())
-    {
-      std::string error_msg;
-      duckdb_trans_rollback(error_msg);
-    }
+    std::string error_msg;
+    duckdb_trans_rollback(error_msg);
   }
 
   bool has_transaction() { return m_con && m_con->HasActiveTransaction(); }
 
-  bool duckdb_trans_begin()
+  bool duckdb_trans_begin(std::string &error_msg)
   {
-    if (!m_con || m_con->HasActiveTransaction())
+    if (!m_con)
+    {
+      error_msg= "DuckDB connection is not available";
       return true;
+    }
+    if (m_con->HasActiveTransaction())
+      return false;
+
     auto result= duckdb_query(*m_con, "BEGIN");
-    return result->HasError();
+    if (result->HasError())
+    {
+      error_msg= result->GetError();
+      return true;
+    }
+    return false;
   }
 
   bool duckdb_trans_commit(std::string &error_msg)
@@ -87,6 +95,12 @@ class DuckdbThdContext
 
   bool duckdb_trans_rollback(std::string &error_msg)
   {
+    if (m_appenders)
+    {
+      m_appenders->discard_all();
+      m_appenders.reset();
+    }
+
     if (!m_con)
       return true;
 

storage/duckdb/runtime/duckdb_manager.cc

@@ -185,74 +185,103 @@ bool DuckdbManager::Initialize()
     return true;
   }
 
-  /* Enable autoloading of statically-linked extensions (core_functions etc.) */
+  try
   {
+    /* Enable autoloading of statically-linked extensions (core_functions etc.) */
     auto con= std::make_shared<duckdb::Connection>(*m_database);
-    con->Query("SET autoload_known_extensions=true");
-    con->Query("SET autoinstall_known_extensions=true");
+    auto init_query= [&con](const std::string &sql) {
+      auto result= con->Query(sql);
+      if (!result)
+        throw duckdb::InternalException(
+            "DuckDB initialization query returned no result");
+      if (result->HasError())
+        throw duckdb::InvalidInputException(
+            "DuckDB initialization query failed: %s", result->GetError());
+    };
+    init_query("SET autoload_known_extensions=true");
+    init_query("SET autoinstall_known_extensions=true");
 
     /*
       Register MariaDB-compatible SQL macros for functions that DuckDB
       lacks but MariaDB pushes down via the original query text.
     */
-    con->Query("CREATE OR REPLACE MACRO adddate(d, i) AS d + i");
+    init_query("CREATE OR REPLACE MACRO adddate(d, i) AS d + i");
     /* addtime/subtime registered as C++ UDFs */
     /* curdate/curtime — MariaDB aliases */
     /* datediff(d1, d2) — MariaDB returns days, DuckDB needs 3-arg form */
-    con->Query("CREATE OR REPLACE MACRO datediff(d1, d2) AS "
+    init_query("CREATE OR REPLACE MACRO datediff(d1, d2) AS "
                "(d1::DATE - d2::DATE)");
-    con->Query("CREATE OR REPLACE MACRO curdate() AS current_date");
-    con->Query("CREATE OR REPLACE MACRO curtime(fsp := 0) AS current_time");
+    init_query("CREATE OR REPLACE MACRO curdate() AS current_date");
+    init_query("CREATE OR REPLACE MACRO curtime(fsp := 0) AS current_time");
     /* utc_time/utc_timestamp/utc_date — UTC wall-clock; fsp ignored */
-    con->Query("CREATE OR REPLACE MACRO utc_timestamp(fsp := 0) AS "
+    init_query("CREATE OR REPLACE MACRO utc_timestamp(fsp := 0) AS "
                "timezone('UTC', now())::TIMESTAMP");
-    con->Query("CREATE OR REPLACE MACRO utc_time(fsp := 0) AS "
+    init_query("CREATE OR REPLACE MACRO utc_time(fsp := 0) AS "
                "timezone('UTC', now())::TIME");
-    con->Query("CREATE OR REPLACE MACRO utc_date() AS "
+    init_query("CREATE OR REPLACE MACRO utc_date() AS "
                "timezone('UTC', now())::DATE");
     /* unix_timestamp([ts]) — epoch seconds; no arg = now() */
-    con->Query("CREATE OR REPLACE MACRO unix_timestamp(ts := now()) AS "
+    init_query("CREATE OR REPLACE MACRO unix_timestamp(ts := now()) AS "
                "epoch(ts)::BIGINT");
     /* time_to_sec(t) — seconds since midnight */
-    con->Query("CREATE OR REPLACE MACRO time_to_sec(t) AS "
+    init_query("CREATE OR REPLACE MACRO time_to_sec(t) AS "
                "(date_part('hour', t)*3600 + date_part('minute', t)*60 "
                "+ date_part('second', t))::BIGINT");
     /* convert_tz(ts, from_tz, to_tz) */
-    con->Query("CREATE OR REPLACE MACRO convert_tz(ts, from_tz, to_tz) AS "
+    init_query("CREATE OR REPLACE MACRO convert_tz(ts, from_tz, to_tz) AS "
                "timezone(to_tz, timezone(from_tz, ts))");
-    con->Query("CREATE OR REPLACE MACRO subdate(d, i) AS d - i");
-    con->Query("CREATE OR REPLACE MACRO insert(str, pos, len, newstr) AS "
+    init_query("CREATE OR REPLACE MACRO subdate(d, i) AS d - i");
+    init_query("CREATE OR REPLACE MACRO insert(str, pos, len, newstr) AS "
                "CASE WHEN pos < 1 OR pos > length(str) THEN str "
                "ELSE substr(str, 1, pos - 1) || newstr || "
                "substr(str, pos + len) END");
     /* to_base64 / from_base64 — DuckDB uses base64()/from_base64() */
-    con->Query("CREATE OR REPLACE MACRO to_base64(x) AS "
+    init_query("CREATE OR REPLACE MACRO to_base64(x) AS "
                "base64(encode(x))");
     /* substring_index(str, delim, count) */
-    con->Query("CREATE OR REPLACE MACRO substring_index(s, d, c) AS "
+    init_query("CREATE OR REPLACE MACRO substring_index(s, d, c) AS "
                "CASE WHEN c > 0 THEN "
                "array_to_string(list_slice(string_split(s, d), 1, c), d) "
                "WHEN c < 0 THEN "
                "array_to_string(list_slice(string_split(s, d), c, NULL), d) "
                "ELSE '' END");
     /* strcmp(s1, s2) — returns 0, -1 or 1 */
-    con->Query("CREATE OR REPLACE MACRO strcmp(a, b) AS "
+    init_query("CREATE OR REPLACE MACRO strcmp(a, b) AS "
                "CASE WHEN a = b THEN 0 WHEN a < b THEN -1 ELSE 1 END");
     /* MID() registered as C++ UDF in register_mysql_compat_functions() */
     /* oct, bin, hex, locate are now registered as native C++ scalar functions
        in register_mysql_compat_functions() -- no SQL macros needed. */
-  }
-
-  /* Register MySQL-compatible function overloads */
-  register_mysql_compat_functions(*m_database->instance);
+    /* Register MySQL-compatible function overloads */
+    register_mysql_compat_functions(*m_database->instance);
 
-  /* Register cross-engine scan support (_mdb_scan + replacement scan) */
-  register_cross_engine_scan(*m_database->instance);
+    /* Register cross-engine scan support (_mdb_scan + replacement scan) */
+    register_cross_engine_scan(*m_database->instance);
 
-  sql_print_information("DuckDB: DuckdbManager::Initialize succeed, path=%s",
-                        path);
+    sql_print_information("DuckDB: DuckdbManager::Initialize succeed, path=%s",
+                          path);
+    return false;
+  }
+  catch (const std::exception &e)
+  {
+    sql_print_error("DuckDB: initialization failed at '%s': %s", path,
+                    e.what());
+  }
+  catch (...)
+  {
+    sql_print_error("DuckDB: initialization failed at '%s': unknown exception",
+                    path);
+  }
 
-  return false;
+  try
+  {
+    delete m_database;
+  }
+  catch (...)
+  {
+    sql_print_error("DuckDB: exception during failed initialization cleanup");
+  }
+  m_database= nullptr;
+  return true;
 }
 
 bool DuckdbManager::CreateInstance()