fix: fix crash, add transaction support

Author: rustyconover

CLAUDE.md

@@ -18,6 +18,11 @@ The extension provides the following functions:
 - `adbc_connect(options)` - Connect to an ADBC data source. Returns a connection handle (BIGINT). Options can be passed as a STRUCT (preferred) or MAP. The `driver` option is required.
 - `adbc_disconnect(handle)` - Disconnect from an ADBC data source. Returns true on success.
 
+### Transaction Control
+- `adbc_set_autocommit(handle, enabled)` - Enable or disable autocommit mode. When disabled, changes require explicit commit.
+- `adbc_commit(handle)` - Commit the current transaction.
+- `adbc_rollback(handle)` - Rollback the current transaction, discarding all uncommitted changes.
+
 ### Query Execution
 - `adbc_scan(handle, query, [params := row(...)])` - Execute a SELECT query and return results as a table. Supports parameterized queries via the optional `params` named parameter.
 - `adbc_execute(handle, query)` - Execute DDL/DML statements (CREATE, INSERT, UPDATE, DELETE). Returns affected row count.
@@ -56,6 +61,13 @@ SELECT * FROM adbc_table_types(getvariable('conn')::BIGINT);
 SELECT * FROM adbc_columns(getvariable('conn')::BIGINT, table_name := 'test');
 SELECT * FROM adbc_schema(getvariable('conn')::BIGINT, 'test');
 
+-- Transaction control
+SELECT adbc_set_autocommit(getvariable('conn')::BIGINT, false);  -- Start transaction
+SELECT adbc_execute(getvariable('conn')::BIGINT, 'INSERT INTO test VALUES (2, ''world'')');
+SELECT adbc_commit(getvariable('conn')::BIGINT);  -- Commit changes
+-- Or: SELECT adbc_rollback(getvariable('conn')::BIGINT);  -- Discard changes
+SELECT adbc_set_autocommit(getvariable('conn')::BIGINT, true);  -- Back to autocommit
+
 -- Disconnect
 SELECT adbc_disconnect(getvariable('conn')::BIGINT);
 ```

src/adbc_functions.cpp

@@ -149,6 +149,56 @@ static void AdbcDisconnectFunction(DataChunk &args, ExpressionState &state, Vect
     });
 }
 
+// adbc_commit(connection_id BIGINT) -> BOOLEAN
+// Commits the current transaction
+static void AdbcCommitFunction(DataChunk &args, ExpressionState &state, Vector &result) {
+    auto &connection_vector = args.data[0];
+
+    UnaryExecutor::Execute<int64_t, bool>(connection_vector, result, args.size(), [&](int64_t connection_id) {
+        auto &registry = ConnectionRegistry::Get();
+        auto connection = registry.Get(connection_id);
+        if (!connection) {
+            throw InvalidInputException("adbc_commit: Invalid connection handle: " + to_string(connection_id));
+        }
+        connection->Commit();
+        return true;
+    });
+}
+
+// adbc_rollback(connection_id BIGINT) -> BOOLEAN
+// Rolls back the current transaction
+static void AdbcRollbackFunction(DataChunk &args, ExpressionState &state, Vector &result) {
+    auto &connection_vector = args.data[0];
+
+    UnaryExecutor::Execute<int64_t, bool>(connection_vector, result, args.size(), [&](int64_t connection_id) {
+        auto &registry = ConnectionRegistry::Get();
+        auto connection = registry.Get(connection_id);
+        if (!connection) {
+            throw InvalidInputException("adbc_rollback: Invalid connection handle: " + to_string(connection_id));
+        }
+        connection->Rollback();
+        return true;
+    });
+}
+
+// adbc_set_autocommit(connection_id BIGINT, enabled BOOLEAN) -> BOOLEAN
+// Sets the autocommit mode for the connection
+static void AdbcSetAutocommitFunction(DataChunk &args, ExpressionState &state, Vector &result) {
+    auto &connection_vector = args.data[0];
+    auto &enabled_vector = args.data[1];
+
+    BinaryExecutor::Execute<int64_t, bool, bool>(connection_vector, enabled_vector, result, args.size(),
+        [&](int64_t connection_id, bool enabled) {
+            auto &registry = ConnectionRegistry::Get();
+            auto connection = registry.Get(connection_id);
+            if (!connection) {
+                throw InvalidInputException("adbc_set_autocommit: Invalid connection handle: " + to_string(connection_id));
+            }
+            connection->SetAutocommit(enabled);
+            return true;
+        });
+}
+
 // Register the ADBC scalar functions using ExtensionLoader
 void RegisterAdbcScalarFunctions(DatabaseInstance &db) {
     ExtensionLoader loader(db, "adbc");
@@ -171,6 +221,33 @@ void RegisterAdbcScalarFunctions(DatabaseInstance &db) {
         AdbcDisconnectFunction
     );
     loader.RegisterFunction(adbc_disconnect_function);
+
+    // adbc_commit: Commit the current transaction
+    auto adbc_commit_function = ScalarFunction(
+        "adbc_commit",
+        {LogicalType::BIGINT},
+        LogicalType::BOOLEAN,
+        AdbcCommitFunction
+    );
+    loader.RegisterFunction(adbc_commit_function);
+
+    // adbc_rollback: Rollback the current transaction
+    auto adbc_rollback_function = ScalarFunction(
+        "adbc_rollback",
+        {LogicalType::BIGINT},
+        LogicalType::BOOLEAN,
+        AdbcRollbackFunction
+    );
+    loader.RegisterFunction(adbc_rollback_function);
+
+    // adbc_set_autocommit: Set autocommit mode
+    auto adbc_set_autocommit_function = ScalarFunction(
+        "adbc_set_autocommit",
+        {LogicalType::BIGINT, LogicalType::BOOLEAN},
+        LogicalType::BOOLEAN,
+        AdbcSetAutocommitFunction
+    );
+    loader.RegisterFunction(adbc_set_autocommit_function);
 }
 
 } // namespace adbc

src/adbc_scan.cpp

@@ -369,8 +369,7 @@ static void AdbcScanFunction(ClientContext &context, TableFunctionInput &data, D
         ArrowTableFunction::ArrowToDuckDB(local_state,
                                           bind_data.arrow_table.GetColumns(),
                                           output,
-                                          0,      // start
-                                          false); // arrow_scan_is_projected = false
+                                          false); // arrow_scan_is_projected = false (no projection pushdown)
     }
 
     local_state.chunk_offset += output.size();

src/include/adbc_connection.hpp

@@ -179,6 +179,26 @@ class AdbcConnectionWrapper {
         CheckAdbc(status, error.Get(), "Failed to get table schema", GetDriverName());
     }
 
+    // Transaction support
+    void Commit() {
+        AdbcErrorGuard error;
+        auto status = AdbcConnectionCommit(&connection, error.Get());
+        CheckAdbc(status, error.Get(), "Failed to commit transaction", GetDriverName());
+    }
+
+    void Rollback() {
+        AdbcErrorGuard error;
+        auto status = AdbcConnectionRollback(&connection, error.Get());
+        CheckAdbc(status, error.Get(), "Failed to rollback transaction", GetDriverName());
+    }
+
+    void SetAutocommit(bool enabled) {
+        AdbcErrorGuard error;
+        const char *value = enabled ? ADBC_OPTION_VALUE_ENABLED : ADBC_OPTION_VALUE_DISABLED;
+        auto status = AdbcConnectionSetOption(&connection, ADBC_CONNECTION_OPTION_AUTOCOMMIT, value, error.Get());
+        CheckAdbc(status, error.Get(), "Failed to set autocommit", GetDriverName());
+    }
+
     // Non-copyable
     AdbcConnectionWrapper(const AdbcConnectionWrapper &) = delete;
     AdbcConnectionWrapper &operator=(const AdbcConnectionWrapper &) = delete;

test/sql/adbc.test

@@ -360,6 +360,70 @@ Invalid Input Error: adbc_execute: Invalid connection handle: 12345
 statement ok
 SELECT adbc_disconnect(getvariable('exec_conn')::BIGINT);
 
+# ============================================
+# Transaction Tests
+# ============================================
+
+# Test adbc_commit with invalid connection handle
+statement error
+SELECT adbc_commit(12345);
+----
+Invalid Input Error: adbc_commit: Invalid connection handle: 12345
+
+# Test adbc_rollback with invalid connection handle
+statement error
+SELECT adbc_rollback(12345);
+----
+Invalid Input Error: adbc_rollback: Invalid connection handle: 12345
+
+# Test adbc_set_autocommit with invalid connection handle
+statement error
+SELECT adbc_set_autocommit(12345, false);
+----
+Invalid Input Error: adbc_set_autocommit: Invalid connection handle: 12345
+
+# Create a connection for transaction tests
+statement ok
+SET VARIABLE tx_conn = (SELECT adbc_connect({'driver': getvariable('sqlite_driver'), 'uri': ':memory:'}));
+
+# Create a test table
+statement ok
+SELECT adbc_execute(getvariable('tx_conn')::BIGINT, 'CREATE TABLE tx_test (id INTEGER, value TEXT)');
+
+# Test rollback - disable autocommit, insert, rollback
+query I
+SELECT adbc_set_autocommit(getvariable('tx_conn')::BIGINT, false);
+----
+true
+
+statement ok
+SELECT adbc_execute(getvariable('tx_conn')::BIGINT, 'INSERT INTO tx_test VALUES (1, ''will rollback'')');
+
+# Rollback the transaction
+query I
+SELECT adbc_rollback(getvariable('tx_conn')::BIGINT);
+----
+true
+
+# Test commit - insert data and commit
+statement ok
+SELECT adbc_execute(getvariable('tx_conn')::BIGINT, 'INSERT INTO tx_test VALUES (2, ''committed'')');
+
+query I
+SELECT adbc_commit(getvariable('tx_conn')::BIGINT);
+----
+true
+
+# Re-enable autocommit
+query I
+SELECT adbc_set_autocommit(getvariable('tx_conn')::BIGINT, true);
+----
+true
+
+# Clean up transaction connection
+statement ok
+SELECT adbc_disconnect(getvariable('tx_conn')::BIGINT);
+
 # Test disconnect
 query I
 SELECT adbc_disconnect(getvariable('conn_id')::BIGINT);