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 87989bf3f feat!(c/driver/postgresql): defer transaction start (#4424)
87989bf3f is described below
commit 87989bf3f9867c5dd2bc61d58b90704973eaa732
Author: David Li <[email protected]>
AuthorDate: Tue Jun 23 16:52:00 2026 -0700
feat!(c/driver/postgresql): defer transaction start (#4424)
- Use ROLLBACK and not ROLLBACK AND CHAIN, which unexpectedly preserves
transaction properties
- Lazily start transactions instead of eagerly
- For pooled connections, this is better behavior
- We do make slightly more server round-trips
Closes #4321.
Assisted-by: GPT-5.5 <[email protected]>
---
c/driver/postgresql/connection.cc | 52 +++++++++++++----
c/driver/postgresql/connection.h | 4 ++
c/driver/postgresql/postgresql_test.cc | 56 +++++++++++++++++-
c/driver/postgresql/statement.cc | 4 ++
python/adbc_driver_postgresql/tests/test_dbapi.py | 69 +++++++++++++++++++++--
5 files changed, 168 insertions(+), 17 deletions(-)
diff --git a/c/driver/postgresql/connection.cc
b/c/driver/postgresql/connection.cc
index 131b2f335..86a61fc7a 100644
--- a/c/driver/postgresql/connection.cc
+++ b/c/driver/postgresql/connection.cc
@@ -478,7 +478,7 @@ AdbcStatusCode PostgresConnection::Commit(struct AdbcError*
error) {
return ADBC_STATUS_OK;
}
- PGresult* result = PQexec(conn_, "COMMIT; BEGIN TRANSACTION");
+ PGresult* result = PQexec(conn_, "COMMIT");
if (PQresultStatus(result) != PGRES_COMMAND_OK) {
AdbcStatusCode code = SetError(error, result, "%s%s",
"[libpq] Failed to commit: ",
PQerrorMessage(conn_));
@@ -489,6 +489,31 @@ AdbcStatusCode PostgresConnection::Commit(struct
AdbcError* error) {
return ADBC_STATUS_OK;
}
+AdbcStatusCode PostgresConnection::EnsureTransaction(struct AdbcError* error) {
+ if (autocommit_) {
+ return ADBC_STATUS_OK;
+ }
+ auto txstatus = PQtransactionStatus(conn_);
+ if (txstatus == PQTRANS_ACTIVE || txstatus == PQTRANS_INTRANS) {
+ return ADBC_STATUS_OK;
+ } else if (txstatus == PQTRANS_INERROR) {
+ InternalAdbcSetError(error,
+ "[libpq] cannot start transaction: "
+ "the connection is in an error state; first
rollback");
+ return ADBC_STATUS_INVALID_STATE;
+ }
+
+ PGresult* result = PQexec(conn_, "BEGIN TRANSACTION");
+ if (PQresultStatus(result) != PGRES_COMMAND_OK) {
+ InternalAdbcSetError(error, "%s%s",
+ "[libpq] Failed to begin transaction: ",
PQerrorMessage(conn_));
+ PQclear(result);
+ return ADBC_STATUS_IO;
+ }
+ PQclear(result);
+ return ADBC_STATUS_OK;
+}
+
AdbcStatusCode PostgresConnection::GetInfo(struct AdbcConnection* connection,
const uint32_t* info_codes,
size_t info_codes_length,
@@ -507,6 +532,7 @@ AdbcStatusCode PostgresConnection::GetInfo(struct
AdbcConnection* connection,
infos.push_back({info_codes[i], std::string(VendorName())});
break;
case ADBC_INFO_VENDOR_VERSION: {
+ RAISE_ADBC(EnsureTransaction(error));
// Gives a version in the form 140000 instead of 14.0.0
const char* stmt = "SHOW server_version_num";
auto result_helper = PqResultHelper{conn_, std::string(stmt)};
@@ -587,6 +613,8 @@ AdbcStatusCode PostgresConnection::GetObjects(
.ToAdbc(error);
}
+ RAISE_ADBC(EnsureTransaction(error));
+
auto status = BuildGetObjects(&helper, depth, catalog_filter, schema_filter,
table_filter, column_filter,
table_type_filter, out);
RAISE_STATUS(error, helper.Close());
@@ -938,6 +966,8 @@ AdbcStatusCode PostgresConnection::GetStatistics(const
char* catalog,
return ADBC_STATUS_NOT_IMPLEMENTED;
}
+ RAISE_ADBC(EnsureTransaction(error));
+
struct ArrowSchema schema;
std::memset(&schema, 0, sizeof(schema));
struct ArrowArray array;
@@ -1011,6 +1041,8 @@ AdbcStatusCode PostgresConnection::GetTableSchema(const
char* catalog,
const char* table_name,
struct ArrowSchema* schema,
struct AdbcError* error) {
+ RAISE_ADBC(EnsureTransaction(error));
+
AdbcStatusCode final_status = ADBC_STATUS_OK;
char* quoted = PQescapeIdentifier(conn_, table_name, strlen(table_name));
@@ -1134,7 +1166,7 @@ AdbcStatusCode PostgresConnection::Rollback(struct
AdbcError* error) {
return ADBC_STATUS_OK;
}
- PGresult* result = PQexec(conn_, "ROLLBACK AND CHAIN");
+ PGresult* result = PQexec(conn_, "ROLLBACK");
if (PQresultStatus(result) != PGRES_COMMAND_OK) {
InternalAdbcSetError(error, "%s%s",
"[libpq] Failed to rollback: ",
PQerrorMessage(conn_));
@@ -1165,16 +1197,16 @@ AdbcStatusCode PostgresConnection::SetOption(const
char* key, const char* value,
}
if (autocommit != autocommit_) {
- const char* query = autocommit ? "COMMIT" : "BEGIN TRANSACTION";
-
- PGresult* result = PQexec(conn_, query);
- if (PQresultStatus(result) != PGRES_COMMAND_OK) {
- InternalAdbcSetError(error, "%s%s", "[libpq] Failed to update
autocommit: ",
- PQerrorMessage(conn_));
+ if (autocommit && PQtransactionStatus(conn_) != PQTRANS_IDLE) {
+ PGresult* result = PQexec(conn_, "COMMIT");
+ if (PQresultStatus(result) != PGRES_COMMAND_OK) {
+ InternalAdbcSetError(error, "%s%s", "[libpq] Failed to update
autocommit: ",
+ PQerrorMessage(conn_));
+ PQclear(result);
+ return ADBC_STATUS_IO;
+ }
PQclear(result);
- return ADBC_STATUS_IO;
}
- PQclear(result);
autocommit_ = autocommit;
}
return ADBC_STATUS_OK;
diff --git a/c/driver/postgresql/connection.h b/c/driver/postgresql/connection.h
index 02e0c4f1b..ff9d2effe 100644
--- a/c/driver/postgresql/connection.h
+++ b/c/driver/postgresql/connection.h
@@ -81,6 +81,10 @@ class PostgresConnection {
const std::array<int, 3>& VendorVersion();
private:
+ friend class PostgresStatement;
+
+ AdbcStatusCode EnsureTransaction(struct AdbcError* error);
+
std::shared_ptr<PostgresDatabase> database_;
std::shared_ptr<PostgresTypeResolver> type_resolver_;
PGconn* conn_;
diff --git a/c/driver/postgresql/postgresql_test.cc
b/c/driver/postgresql/postgresql_test.cc
index e2582299b..ef9308331 100644
--- a/c/driver/postgresql/postgresql_test.cc
+++ b/c/driver/postgresql/postgresql_test.cc
@@ -1103,7 +1103,7 @@ TEST_F(PostgresStatementTest, TransactionStatus) {
ADBC_OPTION_VALUE_DISABLED, &error),
IsOkStatus(&error));
- ASSERT_EQ("intrans", ConnectionGetOption(&connection, txn_status, &error));
+ ASSERT_EQ("idle", ConnectionGetOption(&connection, txn_status, &error));
ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error),
IsOkStatus(&error));
@@ -1119,7 +1119,7 @@ TEST_F(PostgresStatementTest, TransactionStatus) {
ASSERT_EQ("active", ConnectionGetOption(&connection, txn_status, &error));
ASSERT_THAT(AdbcConnectionRollback(&connection, &error),
IsOkStatus(&error));
- ASSERT_EQ("intrans", ConnectionGetOption(&connection, txn_status, &error));
+ ASSERT_EQ("idle", ConnectionGetOption(&connection, txn_status, &error));
}
{
adbc_validation::StreamReader reader;
@@ -1133,10 +1133,60 @@ TEST_F(PostgresStatementTest, TransactionStatus) {
ASSERT_EQ("active", ConnectionGetOption(&connection, txn_status, &error));
ASSERT_THAT(AdbcConnectionCommit(&connection, &error), IsOkStatus(&error));
- ASSERT_EQ("intrans", ConnectionGetOption(&connection, txn_status, &error));
+ ASSERT_EQ("idle", ConnectionGetOption(&connection, txn_status, &error));
}
}
+TEST_F(PostgresStatementTest, RollbackDoesNotChainTransaction) {
+ ASSERT_THAT(AdbcConnectionSetOption(&connection,
ADBC_CONNECTION_OPTION_AUTOCOMMIT,
+ ADBC_OPTION_VALUE_DISABLED, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error),
IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "SET TRANSACTION READ
ONLY", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcConnectionRollback(&connection, &error), IsOkStatus(&error));
+
+ adbc_validation::StreamReader reader;
+ ASSERT_THAT(AdbcStatementSetOption(&statement, "adbc.postgresql.use_copy",
+ ADBC_OPTION_VALUE_DISABLED, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "SHOW
transaction_read_only", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, &reader.stream.value,
+ &reader.rows_affected, &error),
+ IsOkStatus(&error));
+ ASSERT_NO_FATAL_FAILURE(reader.GetSchema());
+ ASSERT_NO_FATAL_FAILURE(reader.Next());
+
+ ASSERT_EQ(reader.array->length, 1);
+ ArrowStringView view =
ArrowArrayViewGetStringUnsafe(reader.array_view->children[0], 0);
+ ASSERT_EQ(std::string_view(view.data, static_cast<size_t>(view.size_bytes)),
"off");
+}
+
+TEST_F(PostgresConnectionTest, GetObjectsStartsTransaction) {
+ using adbc_validation::ConnectionGetOption;
+ const char* txn_status = "adbc.postgresql.transaction_status";
+
+ ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error),
IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcConnectionSetOption(&connection,
ADBC_CONNECTION_OPTION_AUTOCOMMIT,
+ ADBC_OPTION_VALUE_DISABLED, &error),
+ IsOkStatus(&error));
+ ASSERT_EQ("idle", ConnectionGetOption(&connection, txn_status, &error));
+
+ adbc_validation::StreamReader reader;
+ ASSERT_THAT(
+ AdbcConnectionGetObjects(&connection, ADBC_OBJECT_DEPTH_CATALOGS,
nullptr, nullptr,
+ nullptr, nullptr, nullptr,
&reader.stream.value, &error),
+ IsOkStatus(&error));
+ ASSERT_EQ("intrans", ConnectionGetOption(&connection, txn_status, &error));
+}
+
TEST_F(PostgresStatementTest, IsolationLevels) {
ASSERT_THAT(AdbcConnectionSetOption(&connection,
ADBC_CONNECTION_OPTION_ISOLATION_LEVEL,
ADBC_OPTION_ISOLATION_LEVEL_SERIALIZABLE, &error),
diff --git a/c/driver/postgresql/statement.cc b/c/driver/postgresql/statement.cc
index f23cb1483..b0c70dda6 100644
--- a/c/driver/postgresql/statement.cc
+++ b/c/driver/postgresql/statement.cc
@@ -526,6 +526,8 @@ AdbcStatusCode PostgresStatement::ExecuteQuery(struct
ArrowArrayStream* stream,
return ADBC_STATUS_INVALID_STATE;
}
+ RAISE_ADBC(connection_->EnsureTransaction(error));
+
// Use a dedicated path to handle parameter binding
if (bind_.release != nullptr) {
return ExecuteBind(stream, rows_affected, error);
@@ -638,6 +640,8 @@ AdbcStatusCode PostgresStatement::ExecuteSchema(struct
ArrowSchema* schema,
AdbcStatusCode PostgresStatement::ExecuteIngest(struct ArrowArrayStream*
stream,
int64_t* rows_affected,
struct AdbcError* error) {
+ RAISE_ADBC(connection_->EnsureTransaction(error));
+
if (!bind_.release) {
InternalAdbcSetError(error, "%s",
"[libpq] Must Bind() before Execute() for bulk
ingestion");
diff --git a/python/adbc_driver_postgresql/tests/test_dbapi.py
b/python/adbc_driver_postgresql/tests/test_dbapi.py
index 50e2f29ec..4f6456047 100644
--- a/python/adbc_driver_postgresql/tests/test_dbapi.py
+++ b/python/adbc_driver_postgresql/tests/test_dbapi.py
@@ -573,19 +573,19 @@ def test_txn_status(postgres: dbapi.Connection) -> None:
ConnectionOptions.TRANSACTION_STATUS.value
)
- assert status() == "intrans"
+ assert status() == "idle"
postgres.rollback()
- assert status() == "intrans"
+ assert status() == "idle"
with postgres.cursor() as cur:
cur.execute("SELECT 1")
assert status() == "active"
postgres.commit()
- assert status() == "intrans"
+ assert status() == "idle"
cur.execute("SELECT 1")
assert status() == "active"
postgres.rollback()
- assert status() == "intrans"
+ assert status() == "idle"
def test_connect_conn_kwargs_db_schema(postgres_uri: str, postgres:
dbapi.Connection):
@@ -723,3 +723,64 @@ def test_bind_null_unknown_inference(postgres:
dbapi.Connection) -> None:
result = cur.fetchone()
assert result is not None
assert result[0] is None
+
+
+def test_transaction(postgres_uri: str) -> None:
+ with dbapi.connect(postgres_uri) as conn1, dbapi.connect(postgres_uri) as
conn2:
+ with conn1.cursor() as cur1:
+ cur1.execute("DROP TABLE IF EXISTS test_transaction")
+ conn1.commit()
+
+ with conn1.cursor() as cur1:
+ cur1.execute("CREATE TABLE test_transaction (a INTEGER)")
+
+ with conn2.cursor() as cur2:
+ with pytest.raises(dbapi.ProgrammingError) as excinfo:
+ cur2.execute("INSERT INTO test_transaction VALUES (1)")
+ assert excinfo.value.sqlstate == "42P01"
+
+ assert (
+
conn2.adbc_connection.get_option(ConnectionOptions.TRANSACTION_STATUS.value)
+ == "inerror"
+ )
+
+ with conn2.cursor() as cur2:
+ with pytest.raises(
+ dbapi.ProgrammingError,
+ match="connection is in an error state; first rollback",
+ ):
+ cur2.execute("INSERT INTO test_transaction VALUES (1)")
+
+ conn2.rollback()
+
+ with conn1.cursor() as cur1:
+ cur1.execute("INSERT INTO test_transaction VALUES (1)")
+
+ conn1.rollback()
+
+ with conn1.cursor() as cur1:
+ with pytest.raises(dbapi.ProgrammingError) as excinfo:
+ cur1.execute("INSERT INTO test_transaction VALUES (1)")
+ assert excinfo.value.sqlstate == "42P01"
+
+ conn1.rollback()
+
+ with conn1.cursor() as cur1:
+ cur1.execute("CREATE TABLE test_transaction (a INTEGER)")
+ conn1.commit()
+
+ with conn2.cursor() as cur2:
+ cur2.execute("INSERT INTO test_transaction VALUES (1)")
+
+ with conn1.cursor() as cur1:
+ cur1.execute("SELECT COUNT(*) FROM test_transaction")
+ assert cur1.fetchone() == (0,)
+
+ conn2.commit()
+
+ cur1.execute("SELECT COUNT(*) FROM test_transaction")
+ assert cur1.fetchone() == (1,)
+
+ with conn2.cursor() as cur2:
+ cur2.execute("SELECT COUNT(*) FROM test_transaction")
+ assert cur2.fetchone() == (1,)
