merge

Author: David Coe

.asf.yaml

@@ -21,16 +21,22 @@ github:
   collaborators:
     - krlmlr
     - nbenn
-    - ywc88
   enabled_merge_buttons:
     merge: false
     rebase: false
     squash: true
   features:
+    discussions: true
     issues: true
+  labels:
+    - arrow
+    - database
+  protected_branches:
+    main: {}
 
 notifications:
   commits:      commits@arrow.apache.org
+  discussions:  user@arrow.apache.org
   issues_status: issues@arrow.apache.org
   issues:       github@arrow.apache.org
   pullrequests: github@arrow.apache.org

.github/workflows/packaging.yml

@@ -342,7 +342,7 @@ jobs:
           mv apache-arrow-adbc-$VERSION.tar.gz adbc/ci/linux-packages/
 
       - name: Set up Ruby
-        uses: ruby/setup-ruby@922ebc4c5262cd14e07bb0e1db020984b6c064fe  # v1.226.0
+        uses: ruby/setup-ruby@354a1ad156761f5ee2b7b13fa8e09943a5e8d252  # v1.229.0
         with:
           ruby-version: ruby
 

.github/workflows/r-check.yml

@@ -47,7 +47,7 @@ jobs:
       - uses: actions/setup-go@v5
         with:
           go-version: "${{ env.GO_VERSION }}"
-      - uses: r-lib/actions/setup-r@14a7e741c1cb130261263aa1593718ba42cf443b  # v2
+      - uses: r-lib/actions/setup-r@bd49c52ffe281809afa6f0fecbf37483c5dd0b93  # v2
         with:
           r-version: release
           use-public-rspm: true
@@ -72,7 +72,7 @@ jobs:
           popd
         shell: bash
 
-      - uses: r-lib/actions/setup-r-dependencies@f4937e0dc26f9b99c969cd3e4ca943b576e7f991
+      - uses: r-lib/actions/setup-r-dependencies@bd49c52ffe281809afa6f0fecbf37483c5dd0b93
         with:
           extra-packages: any::rcmdcheck
           needs: check
@@ -94,7 +94,7 @@ jobs:
           ADBC_FLIGHTSQL_TEST_URI="grpc://localhost:8080"
           echo "ADBC_FLIGHTSQL_TEST_URI=${ADBC_FLIGHTSQL_TEST_URI}" >> $GITHUB_ENV
 
-      - uses: r-lib/actions/check-r-package@14a7e741c1cb130261263aa1593718ba42cf443b  # v2
+      - uses: r-lib/actions/check-r-package@bd49c52ffe281809afa6f0fecbf37483c5dd0b93  # v2
         env:
           ADBC_SNOWFLAKE_TEST_URI: ${{ secrets.SNOWFLAKE_URI }}
           R_KEEP_PKG_SOURCE: yes

.github/workflows/r-extended.yml

@@ -108,7 +108,7 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
-      - uses: r-lib/actions/setup-r@14a7e741c1cb130261263aa1593718ba42cf443b  # v2
+      - uses: r-lib/actions/setup-r@bd49c52ffe281809afa6f0fecbf37483c5dd0b93  # v2
         with:
           rversion: ${{ matrix.rversion }}
           use-public-rspm: true
@@ -131,7 +131,7 @@ jobs:
           popd
         shell: bash
 
-      - uses: r-lib/actions/setup-r-dependencies@f4937e0dc26f9b99c969cd3e4ca943b576e7f991
+      - uses: r-lib/actions/setup-r-dependencies@bd49c52ffe281809afa6f0fecbf37483c5dd0b93
         with:
           working-directory: r/${{ matrix.pkg }}
 

c/driver/postgresql/statement.cc

@@ -30,6 +30,7 @@
 #include <limits>
 #include <memory>
 #include <string>
+#include <string_view>
 #include <utility>
 #include <vector>
 
@@ -219,55 +220,89 @@ void TupleReader::Release() {
   row_id_ = -1;
 }
 
+// Instead of directly exporting the TupleReader, which is tied to the
+// lifetime of the Statement, we export a weak_ptr reference instead.  That
+// way if the user accidentally closes the Statement before the
+// ArrowArrayStream, we can avoid a crash.
+// See https://github.com/apache/arrow-adbc/issues/2629
+struct ExportedTupleReader {
+  std::weak_ptr<TupleReader> self;
+};
+
 void TupleReader::ExportTo(struct ArrowArrayStream* stream) {
   stream->get_schema = &GetSchemaTrampoline;
   stream->get_next = &GetNextTrampoline;
   stream->get_last_error = &GetLastErrorTrampoline;
   stream->release = &ReleaseTrampoline;
-  stream->private_data = this;
+  stream->private_data = new ExportedTupleReader{weak_from_this()};
 }
 
-const struct AdbcError* TupleReader::ErrorFromArrayStream(struct ArrowArrayStream* stream,
+const struct AdbcError* TupleReader::ErrorFromArrayStream(struct ArrowArrayStream* self,
                                                           AdbcStatusCode* status) {
-  if (!stream->private_data || stream->release != &ReleaseTrampoline) {
+  if (!self->private_data || self->release != &ReleaseTrampoline) {
     return nullptr;
   }
 
-  TupleReader* reader = static_cast<TupleReader*>(stream->private_data);
-  if (status) {
-    *status = reader->status_;
+  auto* wrapper = static_cast<ExportedTupleReader*>(self->private_data);
+  auto maybe_reader = wrapper->self.lock();
+  if (maybe_reader) {
+    if (status) {
+      *status = maybe_reader->status_;
+    }
+    return &maybe_reader->error_;
   }
-  return &reader->error_;
+  return nullptr;
 }
 
 int TupleReader::GetSchemaTrampoline(struct ArrowArrayStream* self,
                                      struct ArrowSchema* out) {
   if (!self || !self->private_data) return EINVAL;
 
-  TupleReader* reader = static_cast<TupleReader*>(self->private_data);
-  return reader->GetSchema(out);
+  auto* wrapper = static_cast<ExportedTupleReader*>(self->private_data);
+  auto maybe_reader = wrapper->self.lock();
+  if (maybe_reader) {
+    return maybe_reader->GetSchema(out);
+  }
+  // statement was closed or reader was otherwise invalidated
+  return EINVAL;
 }
 
 int TupleReader::GetNextTrampoline(struct ArrowArrayStream* self,
                                    struct ArrowArray* out) {
   if (!self || !self->private_data) return EINVAL;
 
-  TupleReader* reader = static_cast<TupleReader*>(self->private_data);
-  return reader->GetNext(out);
+  auto* wrapper = static_cast<ExportedTupleReader*>(self->private_data);
+  auto maybe_reader = wrapper->self.lock();
+  if (maybe_reader) {
+    return maybe_reader->GetNext(out);
+  }
+  // statement was closed or reader was otherwise invalidated
+  return EINVAL;
 }
 
 const char* TupleReader::GetLastErrorTrampoline(struct ArrowArrayStream* self) {
   if (!self || !self->private_data) return nullptr;
+  constexpr std::string_view kReaderInvalidated =
+      "[libpq] Reader invalidated (statement or reader was closed)";
 
-  TupleReader* reader = static_cast<TupleReader*>(self->private_data);
-  return reader->last_error();
+  auto* wrapper = static_cast<ExportedTupleReader*>(self->private_data);
+  auto maybe_reader = wrapper->self.lock();
+  if (maybe_reader) {
+    return maybe_reader->last_error();
+  }
+  // statement was closed or reader was otherwise invalidated
+  return kReaderInvalidated.data();
 }
 
 void TupleReader::ReleaseTrampoline(struct ArrowArrayStream* self) {
   if (!self || !self->private_data) return;
 
-  TupleReader* reader = static_cast<TupleReader*>(self->private_data);
-  reader->Release();
+  auto* wrapper = static_cast<ExportedTupleReader*>(self->private_data);
+  auto maybe_reader = wrapper->self.lock();
+  if (maybe_reader) {
+    maybe_reader->Release();
+  }
+  delete wrapper;
   self->private_data = nullptr;
   self->release = nullptr;
 }
@@ -281,7 +316,7 @@ AdbcStatusCode PostgresStatement::New(struct AdbcConnection* connection,
   connection_ =
       *reinterpret_cast<std::shared_ptr<PostgresConnection>*>(connection->private_data);
   type_resolver_ = connection_->type_resolver();
-  reader_.conn_ = connection_->conn();
+  ClearResult();
   return ADBC_STATUS_OK;
 }
 
@@ -514,24 +549,24 @@ AdbcStatusCode PostgresStatement::ExecuteQuery(struct ArrowArrayStream* stream,
   }
 
   struct ArrowError na_error;
-  reader_.copy_reader_ = std::make_unique<PostgresCopyStreamReader>();
-  CHECK_NA(INTERNAL, reader_.copy_reader_->Init(root_type), error);
+  reader_->copy_reader_ = std::make_unique<PostgresCopyStreamReader>();
+  CHECK_NA(INTERNAL, reader_->copy_reader_->Init(root_type), error);
   CHECK_NA_DETAIL(INTERNAL,
-                  reader_.copy_reader_->InferOutputSchema(
+                  reader_->copy_reader_->InferOutputSchema(
                       std::string(connection_->VendorName()), &na_error),
                   &na_error, error);
 
-  CHECK_NA_DETAIL(INTERNAL, reader_.copy_reader_->InitFieldReaders(&na_error), &na_error,
+  CHECK_NA_DETAIL(INTERNAL, reader_->copy_reader_->InitFieldReaders(&na_error), &na_error,
                   error);
 
   // Execute the COPY query
   RAISE_STATUS(error, helper.ExecuteCopy());
 
   // We need the PQresult back for the reader
-  reader_.result_ = helper.ReleaseResult();
+  reader_->result_ = helper.ReleaseResult();
 
   // Export to stream
-  reader_.ExportTo(stream);
+  reader_->ExportTo(stream);
   if (rows_affected) *rows_affected = -1;
   return ADBC_STATUS_OK;
 }
@@ -674,7 +709,7 @@ AdbcStatusCode PostgresStatement::GetOption(const char* key, char* value, size_t
         break;
     }
   } else if (std::strcmp(key, ADBC_POSTGRESQL_OPTION_BATCH_SIZE_HINT_BYTES) == 0) {
-    result = std::to_string(reader_.batch_size_hint_bytes_);
+    result = std::to_string(reader_->batch_size_hint_bytes_);
   } else if (std::strcmp(key, ADBC_POSTGRESQL_OPTION_USE_COPY) == 0) {
     if (UseCopy()) {
       result = "true";
@@ -710,7 +745,7 @@ AdbcStatusCode PostgresStatement::GetOptionInt(const char* key, int64_t* value,
                                                struct AdbcError* error) {
   std::string result;
   if (std::strcmp(key, ADBC_POSTGRESQL_OPTION_BATCH_SIZE_HINT_BYTES) == 0) {
-    *value = reader_.batch_size_hint_bytes_;
+    *value = reader_->batch_size_hint_bytes_;
     return ADBC_STATUS_OK;
   }
   SetError(error, "[libpq] Unknown statement option '%s'", key);
@@ -799,7 +834,7 @@ AdbcStatusCode PostgresStatement::SetOption(const char* key, const char* value,
       return ADBC_STATUS_INVALID_ARGUMENT;
     }
 
-    this->reader_.batch_size_hint_bytes_ = int_value;
+    this->batch_size_hint_bytes_ = this->reader_->batch_size_hint_bytes_ = int_value;
   } else if (std::strcmp(key, ADBC_POSTGRESQL_OPTION_USE_COPY) == 0) {
     if (std::strcmp(value, ADBC_OPTION_VALUE_ENABLED) == 0) {
       use_copy_ = true;
@@ -836,7 +871,7 @@ AdbcStatusCode PostgresStatement::SetOptionInt(const char* key, int64_t value,
       return ADBC_STATUS_INVALID_ARGUMENT;
     }
 
-    this->reader_.batch_size_hint_bytes_ = value;
+    this->batch_size_hint_bytes_ = this->reader_->batch_size_hint_bytes_ = value;
     return ADBC_STATUS_OK;
   }
   SetError(error, "[libpq] Unknown statement option '%s'", key);
@@ -845,7 +880,9 @@ AdbcStatusCode PostgresStatement::SetOptionInt(const char* key, int64_t value,
 
 void PostgresStatement::ClearResult() {
   // TODO: we may want to synchronize here for safety
-  reader_.Release();
+  if (reader_) reader_->Release();
+  reader_ = std::make_shared<TupleReader>(connection_->conn());
+  reader_->batch_size_hint_bytes_ = batch_size_hint_bytes_;
 }
 
 int PostgresStatement::UseCopy() {

c/driver/postgresql/statement.h

@@ -39,8 +39,10 @@ namespace adbcpq {
 class PostgresConnection;
 class PostgresStatement;
 
+constexpr static int64_t kDefaultBatchSizeHintBytes = 16777216;
+
 /// \brief An ArrowArrayStream that reads tuples from a PGresult.
-class TupleReader final {
+class TupleReader final : public std::enable_shared_from_this<TupleReader> {
  public:
   TupleReader(PGconn* conn)
       : status_(ADBC_STATUS_OK),
@@ -50,7 +52,7 @@ class TupleReader final {
         pgbuf_(nullptr),
         copy_reader_(nullptr),
         row_id_(-1),
-        batch_size_hint_bytes_(16777216),
+        batch_size_hint_bytes_(kDefaultBatchSizeHintBytes),
         is_finished_(false) {
     ArrowErrorInit(&na_error_);
     data_.data.as_char = nullptr;
@@ -98,7 +100,8 @@ class PostgresStatement {
         query_(),
         prepared_(false),
         use_copy_(-1),
-        reader_(nullptr) {
+        reader_(nullptr),
+        batch_size_hint_bytes_(kDefaultBatchSizeHintBytes) {
     std::memset(&bind_, 0, sizeof(bind_));
   }
 
@@ -170,7 +173,8 @@ class PostgresStatement {
     bool temporary = false;
   } ingest_;
 
-  TupleReader reader_;
+  std::shared_ptr<TupleReader> reader_;
+  int64_t batch_size_hint_bytes_;
 
   int UseCopy();
 };

c/integration/duckdb/CMakeLists.txt

@@ -20,7 +20,7 @@ include(FetchContent)
 if(ADBC_BUILD_TESTS)
   fetchcontent_declare(duckdb
                        GIT_REPOSITORY https://github.com/duckdb/duckdb.git
-                       GIT_TAG 0d84ccf478578278b2d1168675b8b93c60f78a5e # v0.9.0
+                       GIT_TAG 8e52ec43959ab363643d63cb78ee214577111da4 # v1.2.1
                        GIT_PROGRESS TRUE
                        USES_TERMINAL_DOWNLOAD TRUE)
   set(BUILD_JEMALLOC_EXTENSION

c/integration/duckdb/duckdb_test.cc

@@ -16,20 +16,21 @@
 // under the License.
 
 #include <cstdlib>
+#include <stdexcept>
 #include <string>
+#include "gmock/gmock.h"
 
 #include <arrow-adbc/adbc.h>
 #include <arrow-adbc/adbc_driver_manager.h>
+#define ADBC_EXPORTING  // duckdb changed the include guard...
 #include <duckdb/common/adbc/adbc-init.hpp>
 
 #include "validation/adbc_validation.h"
 #include "validation/adbc_validation_util.h"
 
 // Convert between our definitions and DuckDB's
 AdbcStatusCode DuckDbDriverInitFunc(int version, void* driver, struct AdbcError* error) {
-  return duckdb_adbc_init(static_cast<size_t>(version),
-                          reinterpret_cast<duckdb_adbc::AdbcDriver*>(driver),
-                          reinterpret_cast<duckdb_adbc::AdbcError*>(error));
+  return duckdb_adbc_init(static_cast<size_t>(version), driver, error);
 }
 
 class DuckDbQuirks : public adbc_validation::DriverQuirks {
@@ -109,10 +110,38 @@ class DuckDbStatementTest : public ::testing::Test,
     GTEST_SKIP() << "Cannot query rows affected in delete stream (not implemented)";
   }
 
+  void TestSqlQueryTrailingSemicolons() {
+    ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error),
+                adbc_validation::IsOkStatus(&error));
+
+    ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "INSTALL icu", &error),
+                adbc_validation::IsOkStatus(&error));
+
+    ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+                adbc_validation::IsOkStatus(&error));
+
+    ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "LOAD icu", &error),
+                adbc_validation::IsOkStatus(&error));
+
+    ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+                adbc_validation::IsOkStatus(&error));
+
+    ASSERT_THAT(AdbcStatementRelease(&statement, &error),
+                adbc_validation::IsOkStatus(&error));
+
+    adbc_validation::StatementTest::TestSqlQueryTrailingSemicolons();
+  }
+
   void TestErrorCompatibility() {
     GTEST_SKIP() << "DuckDB does not set AdbcError.release";
   }
 
+  void TestResultIndependence() {
+    // DuckDB detects this by throwing
+    ASSERT_THAT([this]() { adbc_validation::StatementTest::TestResultIndependence(); },
+                ::testing::Throws<std::runtime_error>());
+  }
+
  protected:
   DuckDbQuirks quirks_;
 };