This is an automated email from the ASF dual-hosted git repository.
lidavidm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/main by this push:
new 4c1afc90d8 fix(c/driver/sqlite): clean up dangling txn on failed
ingest (#4843)
4c1afc90d8 is described below
commit 4c1afc90d87e4016e89ab28ba269a2c4333132a7
Author: David Li <[email protected]>
AuthorDate: Wed Sep 30 16:12:17 2026 -0700
fix(c/driver/sqlite): clean up dangling txn on failed ingest (#4843)
Closes #4828.
---
c/driver/framework/status.h | 2 ++
c/driver/sqlite/sqlite.cc | 18 ++++++++++++++
c/driver/sqlite/sqlite_test.cc | 56 ++++++++++++++++++++++++++++++++++++++++++
3 files changed, 76 insertions(+)
diff --git a/c/driver/framework/status.h b/c/driver/framework/status.h
index 6cb4003149..a34f2f8cd6 100644
--- a/c/driver/framework/status.h
+++ b/c/driver/framework/status.h
@@ -81,6 +81,8 @@ class Status {
}
}
+ std::string message() const { return impl_ ? impl_->message : ""; }
+
/// \brief Export this status to an AdbcError.
AdbcStatusCode ToAdbc(AdbcError* adbc_error) const {
if (impl_ == nullptr) return ADBC_STATUS_OK;
diff --git a/c/driver/sqlite/sqlite.cc b/c/driver/sqlite/sqlite.cc
index 6aea5ee142..8d98a5f3aa 100644
--- a/c/driver/sqlite/sqlite.cc
+++ b/c/driver/sqlite/sqlite.cc
@@ -53,6 +53,7 @@ constexpr std::string_view
kConnectionOptionLoadExtensionEntrypoint =
/// The batch size for query results (and for initial type inference)
constexpr std::string_view kStatementOptionBatchRows =
"adbc.sqlite.query.batch_rows";
constexpr std::string_view kStatementOptionBindByName =
"adbc.statement.bind_by_name";
+constexpr std::string_view kOptionTxnState = "adbc.sqlite.txn_state";
constexpr int kDefaultBatchSize = 1024;
std::string_view GetColumnText(sqlite3_stmt* stmt, int index) {
@@ -739,6 +740,13 @@ class SqliteConnection : public
driver::Connection<SqliteConnection> {
if (key == kStatementOptionBatchRows) {
return driver::Option(
static_cast<int64_t>(batch_size_.value_or(kDefaultBatchSize)));
+ } else if (key == kOptionTxnState) {
+#if SQLITE_VERSION_NUMBER < 3034000
+ return status::NotImplemented(
+ "this version of SQLite does not support sqlite3_txn_state");
+#else
+ return driver::Option(static_cast<int64_t>(sqlite3_txn_state(conn_,
nullptr)));
+#endif
}
return Base::GetOption(key);
}
@@ -1016,6 +1024,15 @@ class SqliteStatement : public
driver::Statement<SqliteStatement> {
&stmt, /*pzTail=*/nullptr);
if (rc != SQLITE_OK) {
std::ignore = sqlite3_finalize(stmt);
+ InternalAdbcSqliteBinderRelease(&binder_);
+ if (is_autocommit) {
+ auto status = ::adbc::sqlite::SqliteQuery::Execute(conn_,
"ROLLBACK");
+ if (!status.ok()) {
+ return status::fmt::Internal(
+ "failed to prepare: {}\nquery was: {}\nfailed to rollback: {}",
+ sqlite3_errmsg(conn_), insert, status.message());
+ }
+ }
return status::fmt::Internal("failed to prepare: {}\nquery was: {}",
sqlite3_errmsg(conn_), insert);
}
@@ -1047,6 +1064,7 @@ class SqliteStatement : public
driver::Statement<SqliteStatement> {
row_count++;
}
std::ignore = sqlite3_finalize(stmt);
+ InternalAdbcSqliteBinderRelease(&binder_);
if (is_autocommit) {
if (status_code == ADBC_STATUS_OK) {
diff --git a/c/driver/sqlite/sqlite_test.cc b/c/driver/sqlite/sqlite_test.cc
index 77ad60e857..d1ba96533b 100644
--- a/c/driver/sqlite/sqlite_test.cc
+++ b/c/driver/sqlite/sqlite_test.cc
@@ -593,6 +593,62 @@ class SqliteStatementTest : public ::testing::Test,
};
ADBCV_TEST_STATEMENT(SqliteStatementTest)
+TEST_F(SqliteStatementTest, SqlIngestRollback) {
+ // Regression test for https://github.com/apache/arrow-adbc/issues/4828
+ // A failed ingest shouldn't leave a dangling transaction
+ ASSERT_THAT(quirks()->DropTable(&connection, "test_rollback", &error),
+ adbc_validation::IsOkStatus(&error));
+
+ std::string table = "test_rollback";
+ adbc_validation::Handle<struct ArrowSchema> schema;
+ adbc_validation::Handle<struct ArrowArray> array;
+ struct ArrowError na_error;
+ ASSERT_THAT(
+ adbc_validation::MakeSchema(&schema.value, {{"index",
NANOARROW_TYPE_INT64}}),
+ adbc_validation::IsOkErrno());
+ ASSERT_THAT((adbc_validation::MakeBatch<int64_t>(&schema.value,
&array.value, &na_error,
+ {42, -42, std::nullopt})),
+ adbc_validation::IsOkErrno(&na_error));
+
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error),
+ adbc_validation::IsOkStatus(&error));
+
+ {
+ int64_t txn_state = 0;
+ ASSERT_THAT(AdbcConnectionGetOptionInt(&connection,
"adbc.sqlite.txn_state",
+ &txn_state, &error),
+ adbc_validation::IsOkStatus(&error));
+ ASSERT_EQ(0, txn_state);
+ }
+
+ ASSERT_THAT(AdbcStatementSetOption(&statement,
ADBC_INGEST_OPTION_TARGET_TABLE,
+ table.c_str(), &error),
+ adbc_validation::IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_MODE,
+ ADBC_INGEST_OPTION_MODE_APPEND, &error),
+ adbc_validation::IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementBind(&statement, &array.value, &schema.value,
&error),
+ adbc_validation::IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ adbc_validation::IsStatus(ADBC_STATUS_INTERNAL, &error));
+
+ {
+ // Need to do something nontrivial for txn_state to be meaningful
+ ASSERT_THAT(AdbcStatementSetSqlQuery(
+ &statement, "CREATE TABLE IF NOT EXISTS foobar (a INT)",
&error),
+ adbc_validation::IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr,
&error),
+ adbc_validation::IsOkStatus(&error));
+
+ int64_t txn_state = 0;
+ ASSERT_THAT(AdbcConnectionGetOptionInt(&connection,
"adbc.sqlite.txn_state",
+ &txn_state, &error),
+ adbc_validation::IsOkStatus(&error));
+ ASSERT_EQ(0, txn_state);
+ }
+}
+
TEST_F(SqliteStatementTest, SqlIngestNameEscaping) {
ASSERT_THAT(quirks()->DropTable(&connection, "test-table", &error),
adbc_validation::IsOkStatus(&error));
