You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by li...@apache.org on 2023/08/08 17:01:21 UTC
[arrow-adbc] branch spec-1.1.0 updated: feat(format,c,go,python): add error details to ArrowArrayStream (#963)
This is an automated email from the ASF dual-hosted git repository.
lidavidm pushed a commit to branch spec-1.1.0
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/spec-1.1.0 by this push:
new 6b008d00 feat(format,c,go,python): add error details to ArrowArrayStream (#963)
6b008d00 is described below
commit 6b008d00d1ede34005c0618420bfed3d3b5dd895
Author: David Li <li...@gmail.com>
AuthorDate: Tue Aug 8 13:01:13 2023 -0400
feat(format,c,go,python): add error details to ArrowArrayStream (#963)
---
.env | 2 +-
.github/workflows/integration.yml | 4 +-
.github/workflows/native-unix.yml | 18 +-
.github/workflows/native-windows.yml | 6 +-
.github/workflows/packaging.yml | 4 +-
.pre-commit-config.yaml | 2 +-
adbc.h | 28 +-
c/driver/common/utils.c | 41 ++-
c/driver/common/utils.h | 6 +-
c/driver/flightsql/sqlite_flightsql_test.cc | 117 +++++++++
c/driver/postgresql/postgresql.cc | 18 +-
c/driver/postgresql/postgresql_test.cc | 40 +++
c/driver/postgresql/statement.cc | 73 ++++--
c/driver/postgresql/statement.h | 19 +-
c/driver_manager/adbc_driver_manager.cc | 121 +++++++--
c/integration/duckdb/CMakeLists.txt | 1 +
c/integration/duckdb/duckdb_test.cc | 4 +
dev/release/verify-release-candidate.sh | 8 +-
docker-compose.yml | 26 +-
go/adbc/driver/flightsql/flightsql_adbc.go | 110 +++++---
go/adbc/driver/flightsql/flightsql_statement.go | 40 ++-
go/adbc/driver/flightsql/record_reader.go | 21 +-
go/adbc/driver/flightsql/utils.go | 37 ++-
go/adbc/drivermgr/adbc.h | 28 +-
go/adbc/drivermgr/adbc_driver_manager.cc | 121 +++++++--
go/adbc/go.mod | 2 +-
go/adbc/pkg/_tmpl/driver.go.tmpl | 174 ++++++++++++-
go/adbc/pkg/_tmpl/utils.c.tmpl | 33 ++-
go/adbc/pkg/_tmpl/utils.h.tmpl | 8 +-
go/adbc/pkg/flightsql/driver.go | 174 ++++++++++++-
go/adbc/pkg/flightsql/utils.c | 33 ++-
go/adbc/pkg/flightsql/utils.h | 11 +-
go/adbc/pkg/panicdummy/driver.go | 174 ++++++++++++-
go/adbc/pkg/panicdummy/utils.c | 33 ++-
go/adbc/pkg/panicdummy/utils.h | 11 +-
go/adbc/pkg/snowflake/driver.go | 174 ++++++++++++-
go/adbc/pkg/snowflake/utils.c | 33 ++-
go/adbc/pkg/snowflake/utils.h | 11 +-
.../adbc_driver_manager/_lib.pxd | 284 +++++++++++++++++++++
.../adbc_driver_manager/_lib.pyx | 257 +------------------
.../adbc_driver_manager/_reader.pyi | 36 ++-
.../adbc_driver_manager/_reader.pyx | 113 ++++++++
.../adbc_driver_manager/dbapi.py | 5 +-
python/adbc_driver_manager/setup.py | 11 +-
python/adbc_driver_manager/tests/test_reader.py | 80 ++++++
python/adbc_driver_postgresql/tests/test_dbapi.py | 35 ++-
r/adbcflightsql/tools/download-go.R | 2 +-
r/adbcsnowflake/tools/download-go.R | 2 +-
48 files changed, 2047 insertions(+), 544 deletions(-)
diff --git a/.env b/.env
index 2b73edb5..fdf41835 100644
--- a/.env
+++ b/.env
@@ -33,7 +33,7 @@ MANYLINUX=2014
MAVEN=3.5.4
PYTHON=3.10
GO=1.19.5
-ARROW_MAJOR_VERSION=12
+ARROW_MAJOR_VERSION=13
# Used through docker-compose.yml and serves as the default version for the
# ci/scripts/install_vcpkg.sh script.
diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index c293e0f0..0aa9c779 100644
--- a/.github/workflows/integration.yml
+++ b/.github/workflows/integration.yml
@@ -130,7 +130,7 @@ jobs:
--file ci/conda_env_python.txt
- uses: actions/setup-go@v3
with:
- go-version: 1.18.10
+ go-version: 1.20.7
check-latest: true
cache: true
cache-dependency-path: go/adbc/go.sum
@@ -291,7 +291,7 @@ jobs:
--file ci/conda_env_python.txt
- uses: actions/setup-go@v3
with:
- go-version: 1.18.10
+ go-version: 1.20.7
check-latest: true
cache: true
cache-dependency-path: go/adbc/go.sum
diff --git a/.github/workflows/native-unix.yml b/.github/workflows/native-unix.yml
index b12aa46d..62498baf 100644
--- a/.github/workflows/native-unix.yml
+++ b/.github/workflows/native-unix.yml
@@ -98,7 +98,7 @@ jobs:
--file ci/conda_env_cpp.txt
- uses: actions/setup-go@v3
with:
- go-version: 1.18.10
+ go-version: 1.20.7
check-latest: true
cache: true
cache-dependency-path: go/adbc/go.sum
@@ -112,14 +112,14 @@ jobs:
export ADBC_BUILD_TESTS=OFF
export ADBC_USE_ASAN=OFF
export ADBC_USE_UBSAN=OFF
- export PATH=$RUNNER_TOOL_CACHE/go/1.18.10/x64/bin:$PATH
+ export PATH=$RUNNER_TOOL_CACHE/go/1.20.7/x64/bin:$PATH
./ci/scripts/cpp_build.sh "$(pwd)" "$(pwd)/build" "$HOME/local"
- name: Go Build
shell: bash -l {0}
env:
CGO_ENABLED: "1"
run: |
- export PATH=$RUNNER_TOOL_CACHE/go/1.18.10/x64/bin:$PATH
+ export PATH=$RUNNER_TOOL_CACHE/go/1.20.7/x64/bin:$PATH
./ci/scripts/go_build.sh "$(pwd)" "$(pwd)/build" "$HOME/local"
- uses: actions/upload-artifact@v3
@@ -307,7 +307,7 @@ jobs:
cache: true
cache-dependency-path: go/adbc/go.sum
- name: Install staticcheck
- run: go install honnef.co/go/tools/cmd/staticcheck@v0.3.3
+ run: go install honnef.co/go/tools/cmd/staticcheck@v0.4.3
- name: Go Build
run: |
./ci/scripts/go_build.sh "$(pwd)" "$(pwd)/build" "$HOME/local"
@@ -361,14 +361,14 @@ jobs:
--file ci/conda_env_cpp.txt
- uses: actions/setup-go@v3
with:
- go-version: 1.18.10
+ go-version: 1.20.7
check-latest: true
cache: true
cache-dependency-path: go/adbc/go.sum
- name: Install staticcheck
shell: bash -l {0}
if: ${{ !contains('macos-latest', matrix.os) }}
- run: go install honnef.co/go/tools/cmd/staticcheck@v0.3.3
+ run: go install honnef.co/go/tools/cmd/staticcheck@v0.4.3
- uses: actions/download-artifact@v3
with:
@@ -378,7 +378,7 @@ jobs:
- name: Go Build
shell: bash -l {0}
run: |
- export PATH=$RUNNER_TOOL_CACHE/go/1.18.10/x64/bin:$PATH
+ export PATH=$RUNNER_TOOL_CACHE/go/1.20.7/x64/bin:$PATH
./ci/scripts/go_build.sh "$(pwd)" "$(pwd)/build" "$HOME/local"
- name: Run Staticcheck
if: ${{ !contains('macos-latest', matrix.os) }}
@@ -392,7 +392,7 @@ jobs:
env:
SNOWFLAKE_URI: ${{ secrets.SNOWFLAKE_URI }}
run: |
- export PATH=$RUNNER_TOOL_CACHE/go/1.18.10/x64/bin:$PATH
+ export PATH=$RUNNER_TOOL_CACHE/go/1.20.7/x64/bin:$PATH
./ci/scripts/go_test.sh "$(pwd)" "$(pwd)/build" "$HOME/local"
# ------------------------------------------------------------
@@ -441,7 +441,7 @@ jobs:
--file ci/conda_env_python.txt
- uses: actions/setup-go@v3
with:
- go-version: 1.18.10
+ go-version: 1.20.7
check-latest: true
cache: true
cache-dependency-path: go/adbc/go.sum
diff --git a/.github/workflows/native-windows.yml b/.github/workflows/native-windows.yml
index a6851374..d5089a38 100644
--- a/.github/workflows/native-windows.yml
+++ b/.github/workflows/native-windows.yml
@@ -223,7 +223,7 @@ jobs:
--file ci/conda_env_cpp.txt
- uses: actions/setup-go@v3
with:
- go-version: 1.18.10
+ go-version: 1.20.7
check-latest: true
cache: true
cache-dependency-path: go/adbc/go.sum
@@ -238,7 +238,7 @@ jobs:
env:
CGO_ENABLED: "1"
run: |
- $env:PATH="$($env:RUNNER_TOOL_CACHE)\go\1.18.10\x64\bin;" + $env:PATH
+ $env:PATH="$($env:RUNNER_TOOL_CACHE)\go\1.20.7\x64\bin;" + $env:PATH
.\ci\scripts\go_build.ps1 $pwd $pwd\build
# TODO(apache/arrow#358): enable these tests on Windows
# - name: Go Test
@@ -246,7 +246,7 @@ jobs:
# env:
# CGO_ENABLED: "1"
# run: |
- # $env:PATH="$($env:RUNNER_TOOL_CACHE)\go\1.18.10\x64\bin;" + $env:PATH
+ # $env:PATH="$($env:RUNNER_TOOL_CACHE)\go\1.20.7\x64\bin;" + $env:PATH
# .\ci\scripts\go_test.ps1 $pwd $pwd\build
# ------------------------------------------------------------
diff --git a/.github/workflows/packaging.yml b/.github/workflows/packaging.yml
index f677e740..b09100f6 100644
--- a/.github/workflows/packaging.yml
+++ b/.github/workflows/packaging.yml
@@ -605,7 +605,7 @@ jobs:
- uses: actions/setup-go@v3
with:
- go-version: 1.18.10
+ go-version: 1.20.7
check-latest: true
cache: true
cache-dependency-path: adbc/go/adbc/go.sum
@@ -727,7 +727,7 @@ jobs:
- uses: actions/setup-go@v3
with:
- go-version: 1.18.10
+ go-version: 1.20.7
check-latest: true
cache: true
cache-dependency-path: adbc/go/adbc/go.sum
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 188b069b..eef6a96e 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -52,7 +52,7 @@ repos:
- "--linelength=90"
- "--verbose=2"
- repo: https://github.com/golangci/golangci-lint
- rev: v1.49.0
+ rev: v1.53.3
hooks:
- id: golangci-lint
entry: bash -c 'cd go/adbc && golangci-lint run --fix --timeout 5m'
diff --git a/adbc.h b/adbc.h
index c38f2064..a2a33a4d 100644
--- a/adbc.h
+++ b/adbc.h
@@ -360,7 +360,7 @@ struct ADBC_EXPORT AdbcErrorDetail {
///
/// \since ADBC API revision 1.1.0
ADBC_EXPORT
-int AdbcErrorGetDetailCount(struct AdbcError* error);
+int AdbcErrorGetDetailCount(const struct AdbcError* error);
/// \brief Get a metadata value in an error by index.
///
@@ -369,7 +369,25 @@ int AdbcErrorGetDetailCount(struct AdbcError* error);
///
/// \since ADBC API revision 1.1.0
ADBC_EXPORT
-struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index);
+struct AdbcErrorDetail AdbcErrorGetDetail(const struct AdbcError* error, int index);
+
+/// \brief Get an ADBC error from an ArrowArrayStream created by a driver.
+///
+/// This allows retrieving error details and other metadata that would
+/// normally be suppressed by the Arrow C Stream Interface.
+///
+/// The caller MUST NOT release the error; it is managed by the release
+/// callback in the stream itself.
+///
+/// \param[in] stream The stream to query.
+/// \param[out] status The ADBC status code, or ADBC_STATUS_OK if there is no
+/// error. Not written to if the stream does not contain an ADBC error or
+/// if the pointer is NULL.
+/// \return NULL if not supported.
+/// \since ADBC API revision 1.1.0
+ADBC_EXPORT
+const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status);
/// @}
@@ -950,8 +968,10 @@ struct ADBC_EXPORT AdbcDriver {
///
/// @{
- int (*ErrorGetDetailCount)(struct AdbcError* error);
- struct AdbcErrorDetail (*ErrorGetDetail)(struct AdbcError* error, int index);
+ int (*ErrorGetDetailCount)(const struct AdbcError* error);
+ struct AdbcErrorDetail (*ErrorGetDetail)(const struct AdbcError* error, int index);
+ const struct AdbcError* (*ErrorFromArrayStream)(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status);
AdbcStatusCode (*DatabaseGetOption)(struct AdbcDatabase*, const char*, char*, size_t*,
struct AdbcError*);
diff --git a/c/driver/common/utils.c b/c/driver/common/utils.c
index abd41f8c..71d9e7ef 100644
--- a/c/driver/common/utils.c
+++ b/c/driver/common/utils.c
@@ -17,16 +17,49 @@
#include "utils.h"
+#include <assert.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
-#include <assert.h>
-#include "adbc.h"
+#include <adbc.h>
static size_t kErrorBufferSize = 1024;
+int AdbcStatusCodeToErrno(AdbcStatusCode code) {
+ switch (code) {
+ case ADBC_STATUS_OK:
+ return 0;
+ case ADBC_STATUS_UNKNOWN:
+ return EIO;
+ case ADBC_STATUS_NOT_IMPLEMENTED:
+ return ENOTSUP;
+ case ADBC_STATUS_NOT_FOUND:
+ return ENOENT;
+ case ADBC_STATUS_ALREADY_EXISTS:
+ return EEXIST;
+ case ADBC_STATUS_INVALID_ARGUMENT:
+ case ADBC_STATUS_INVALID_STATE:
+ return EINVAL;
+ case ADBC_STATUS_INVALID_DATA:
+ case ADBC_STATUS_INTEGRITY:
+ case ADBC_STATUS_INTERNAL:
+ case ADBC_STATUS_IO:
+ return EIO;
+ case ADBC_STATUS_CANCELLED:
+ return ECANCELED;
+ case ADBC_STATUS_TIMEOUT:
+ return ETIMEDOUT;
+ case ADBC_STATUS_UNAUTHENTICATED:
+ // FreeBSD/macOS have EAUTH, but not other platforms
+ case ADBC_STATUS_UNAUTHORIZED:
+ return EACCES;
+ default:
+ return EIO;
+ }
+}
+
/// For ADBC 1.1.0, the structure held in private_data.
struct AdbcErrorDetails {
char* message;
@@ -165,7 +198,7 @@ void AppendErrorDetail(struct AdbcError* error, const char* key, const uint8_t*
details->count++;
}
-int CommonErrorGetDetailCount(struct AdbcError* error) {
+int CommonErrorGetDetailCount(const struct AdbcError* error) {
if (error->release != ReleaseErrorWithDetails) {
return 0;
}
@@ -173,7 +206,7 @@ int CommonErrorGetDetailCount(struct AdbcError* error) {
return details->count;
}
-struct AdbcErrorDetail CommonErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail CommonErrorGetDetail(const struct AdbcError* error, int index) {
if (error->release != ReleaseErrorWithDetails) {
return (struct AdbcErrorDetail){NULL, NULL, 0};
}
diff --git a/c/driver/common/utils.h b/c/driver/common/utils.h
index 88185557..e3d81cb0 100644
--- a/c/driver/common/utils.h
+++ b/c/driver/common/utils.h
@@ -28,6 +28,8 @@
extern "C" {
#endif
+int AdbcStatusCodeToErrno(AdbcStatusCode code);
+
// The printf checking attribute doesn't work properly on gcc 4.8
// and results in spurious compiler warnings
#if defined(__clang__) || (defined(__GNUC__) && __GNUC__ >= 5)
@@ -47,8 +49,8 @@ void SetErrorVariadic(struct AdbcError* error, const char* format, va_list args)
void AppendErrorDetail(struct AdbcError* error, const char* key, const uint8_t* detail,
size_t detail_length);
-int CommonErrorGetDetailCount(struct AdbcError* error);
-struct AdbcErrorDetail CommonErrorGetDetail(struct AdbcError* error, int index);
+int CommonErrorGetDetailCount(const struct AdbcError* error);
+struct AdbcErrorDetail CommonErrorGetDetail(const struct AdbcError* error, int index);
struct StringBuilder {
char* buffer;
diff --git a/c/driver/flightsql/sqlite_flightsql_test.cc b/c/driver/flightsql/sqlite_flightsql_test.cc
index b2aea631..46ca69be 100644
--- a/c/driver/flightsql/sqlite_flightsql_test.cc
+++ b/c/driver/flightsql/sqlite_flightsql_test.cc
@@ -295,3 +295,120 @@ TEST_F(SqliteFlightSqlStatementTest, NonexistentTable) {
error.release(&error);
}
}
+
+TEST_F(SqliteFlightSqlStatementTest, CancelError) {
+ // Ensure cancellation propagates properly through the Go FFI boundary
+ adbc_validation::Handle<struct AdbcStatement> statement;
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement.value, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement.value, "SELECT 1", &error),
+ IsOkStatus(&error));
+
+ adbc_validation::StreamReader reader;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement.value, &reader.stream.value,
+ &reader.rows_affected, &error),
+ adbc_validation::IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementCancel(&statement.value, &error),
+ adbc_validation::IsOkStatus(&error));
+
+ ASSERT_NO_FATAL_FAILURE(reader.GetSchema());
+
+ int retcode = 0;
+ while (true) {
+ retcode = reader.MaybeNext();
+ if (retcode != 0 || !reader.array->release) break;
+ }
+
+ ASSERT_EQ(ECANCELED, retcode);
+ AdbcStatusCode status = ADBC_STATUS_OK;
+ const struct AdbcError* adbc_error =
+ AdbcErrorFromArrayStream(&reader.stream.value, &status);
+ ASSERT_NE(nullptr, adbc_error);
+ ASSERT_EQ(ADBC_STATUS_CANCELLED, status);
+}
+
+TEST_F(SqliteFlightSqlStatementTest, RpcError) {
+ // Ensure errors that happen at the start of the stream propagate properly
+ // through the Go FFI boundary
+ adbc_validation::Handle<struct AdbcStatement> statement;
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement.value, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement.value, "SELECT", &error),
+ IsOkStatus(&error));
+
+ adbc_validation::StreamReader reader;
+ error.vendor_code = ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement.value, &reader.stream.value,
+ &reader.rows_affected, &error),
+ adbc_validation::IsStatus(ADBC_STATUS_UNKNOWN, &error));
+
+ int count = AdbcErrorGetDetailCount(&error);
+ ASSERT_NE(0, count);
+ for (int i = 0; i < count; i++) {
+ struct AdbcErrorDetail detail = AdbcErrorGetDetail(&error, i);
+ ASSERT_NE(nullptr, detail.key);
+ ASSERT_NE(nullptr, detail.value);
+ ASSERT_NE(0, detail.value_length);
+ EXPECT_STREQ("afsql-sqlite-query", detail.key);
+ EXPECT_EQ("SELECT", std::string_view(reinterpret_cast<const char*>(detail.value),
+ detail.value_length));
+ }
+}
+
+TEST_F(SqliteFlightSqlStatementTest, StreamError) {
+ // Ensure errors that happen during the stream propagate properly through
+ // the Go FFI boundary
+ adbc_validation::Handle<struct AdbcStatement> statement;
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement.value, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement.value,
+ R"(
+DROP TABLE IF EXISTS foo;
+CREATE TABLE foo (a INT);
+WITH RECURSIVE sequence(x) AS
+ (SELECT 1 UNION ALL SELECT x+1 FROM sequence LIMIT 1024)
+INSERT INTO foo(a)
+SELECT x FROM sequence;
+INSERT INTO foo(a) VALUES ('foo');)",
+ &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement.value, nullptr, nullptr, &error),
+ adbc_validation::IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement.value, "SELECT * FROM foo", &error),
+ IsOkStatus(&error));
+
+ adbc_validation::StreamReader reader;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement.value, &reader.stream.value,
+ &reader.rows_affected, &error),
+ adbc_validation::IsOkStatus(&error));
+
+ ASSERT_NO_FATAL_FAILURE(reader.GetSchema());
+
+ int retcode = 0;
+ while (true) {
+ retcode = reader.MaybeNext();
+ if (retcode != 0 || !reader.array->release) break;
+ }
+
+ ASSERT_NE(0, retcode);
+ AdbcStatusCode status = ADBC_STATUS_OK;
+ const struct AdbcError* adbc_error =
+ AdbcErrorFromArrayStream(&reader.stream.value, &status);
+ ASSERT_NE(nullptr, adbc_error);
+ ASSERT_EQ(ADBC_STATUS_UNKNOWN, status);
+
+ int count = AdbcErrorGetDetailCount(adbc_error);
+ ASSERT_NE(0, count);
+ for (int i = 0; i < count; i++) {
+ struct AdbcErrorDetail detail = AdbcErrorGetDetail(adbc_error, i);
+ ASSERT_NE(nullptr, detail.key);
+ ASSERT_NE(nullptr, detail.value);
+ ASSERT_NE(0, detail.value_length);
+ EXPECT_STREQ("grpc-status-details-bin", detail.key);
+ }
+}
diff --git a/c/driver/postgresql/postgresql.cc b/c/driver/postgresql/postgresql.cc
index 0c51f8b8..28cc1bee 100644
--- a/c/driver/postgresql/postgresql.cc
+++ b/c/driver/postgresql/postgresql.cc
@@ -50,14 +50,27 @@ using adbcpq::PostgresStatement;
// ---------------------------------------------------------------------
// AdbcError
-int AdbcErrorGetDetailCount(struct AdbcError* error) {
+namespace {
+const struct AdbcError* PostgresErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ // Currently only valid for TupleReader
+ return adbcpq::TupleReader::ErrorFromArrayStream(stream, status);
+}
+} // namespace
+
+int AdbcErrorGetDetailCount(const struct AdbcError* error) {
return CommonErrorGetDetailCount(error);
}
-struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail AdbcErrorGetDetail(const struct AdbcError* error, int index) {
return CommonErrorGetDetail(error, index);
}
+const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ return PostgresErrorFromArrayStream(stream, status);
+}
+
// ---------------------------------------------------------------------
// AdbcDatabase
@@ -848,6 +861,7 @@ AdbcStatusCode PostgresqlDriverInit(int version, void* raw_driver,
driver->ErrorGetDetailCount = CommonErrorGetDetailCount;
driver->ErrorGetDetail = CommonErrorGetDetail;
+ driver->ErrorFromArrayStream = PostgresErrorFromArrayStream;
driver->DatabaseGetOption = PostgresDatabaseGetOption;
driver->DatabaseGetOptionBytes = PostgresDatabaseGetOptionBytes;
diff --git a/c/driver/postgresql/postgresql_test.cc b/c/driver/postgresql/postgresql_test.cc
index 96fc2e44..84a264f3 100644
--- a/c/driver/postgresql/postgresql_test.cc
+++ b/c/driver/postgresql/postgresql_test.cc
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+#include <cerrno>
#include <cmath>
#include <cstdlib>
#include <cstring>
@@ -979,6 +980,45 @@ TEST_F(PostgresStatementTest, AdbcErrorBackwardsCompatibility) {
free(error);
}
+TEST_F(PostgresStatementTest, Cancel) {
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error), IsOkStatus(&error));
+
+ for (const char* query : {
+ "DROP TABLE IF EXISTS test_cancel",
+ "CREATE TABLE test_cancel (ints INT)",
+ R"(INSERT INTO test_cancel (ints)
+ SELECT g :: INT FROM GENERATE_SERIES(1, 65536) temp(g))",
+ }) {
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, query, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+ }
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "SELECT * FROM test_cancel", &error),
+ IsOkStatus(&error));
+ adbc_validation::StreamReader reader;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, &reader.stream.value,
+ &reader.rows_affected, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementCancel(&statement, &error), IsOkStatus(&error));
+
+ int retcode = 0;
+ while (true) {
+ retcode = reader.MaybeNext();
+ if (retcode != 0 || !reader.array->release) break;
+ }
+
+ ASSERT_EQ(ECANCELED, retcode);
+ AdbcStatusCode status = ADBC_STATUS_OK;
+ const struct AdbcError* detail =
+ AdbcErrorFromArrayStream(&reader.stream.value, &status);
+ ASSERT_NE(nullptr, detail);
+ ASSERT_EQ(ADBC_STATUS_CANCELLED, status);
+ ASSERT_EQ("57014", std::string_view(detail->sqlstate, 5));
+ ASSERT_NE(0, AdbcErrorGetDetailCount(detail));
+}
+
struct TypeTestCase {
std::string name;
std::string sql_type;
diff --git a/c/driver/postgresql/statement.cc b/c/driver/postgresql/statement.cc
index 3e6f92e2..e46be0a1 100644
--- a/c/driver/postgresql/statement.cc
+++ b/c/driver/postgresql/statement.cc
@@ -523,12 +523,13 @@ struct BindStream {
int TupleReader::GetSchema(struct ArrowSchema* out) {
int na_res = copy_reader_->GetSchema(out);
if (out->release == nullptr) {
- StringBuilderAppend(&error_builder_,
- "[libpq] Result set was already consumed or freed");
- return EINVAL;
+ SetError(&error_, "[libpq] Result set was already consumed or freed");
+ status_ = ADBC_STATUS_INVALID_STATE;
+ return AdbcStatusCodeToErrno(status_);
} else if (na_res != NANOARROW_OK) {
// e.g., Can't allocate memory
- StringBuilderAppend(&error_builder_, "[libpq] Error copying schema");
+ SetError(&error_, "[libpq] Error copying schema");
+ status_ = ADBC_STATUS_INTERNAL;
}
return na_res;
@@ -543,15 +544,16 @@ int TupleReader::InitQueryAndFetchFirst(struct ArrowError* error) {
data_.data.as_char = pgbuf_;
if (get_copy_res == -2) {
- StringBuilderAppend(&error_builder_, "[libpq] Fetch header failed: %s",
- PQerrorMessage(conn_));
- return EIO;
+ SetError(&error_, "[libpq] Fetch header failed: %s", PQerrorMessage(conn_));
+ status_ = ADBC_STATUS_IO;
+ return AdbcStatusCodeToErrno(status_);
}
int na_res = copy_reader_->ReadHeader(&data_, error);
if (na_res != NANOARROW_OK) {
- StringBuilderAppend(&error_builder_, "[libpq] ReadHeader failed: %s", error->message);
- return EIO;
+ SetError(&error_, "[libpq] ReadHeader failed: %s", error->message);
+ status_ = ADBC_STATUS_IO;
+ return AdbcStatusCodeToErrno(status_);
}
return NANOARROW_OK;
@@ -562,9 +564,9 @@ int TupleReader::AppendRowAndFetchNext(struct ArrowError* error) {
// call to PQgetCopyData())
int na_res = copy_reader_->ReadRecord(&data_, error);
if (na_res != NANOARROW_OK && na_res != ENODATA) {
- StringBuilderAppend(&error_builder_,
- "[libpq] ReadRecord failed at row %" PRId64 ": %s", row_id_,
- error->message);
+ SetError(&error_, "[libpq] ReadRecord failed at row %" PRId64 ": %s", row_id_,
+ error->message);
+ status_ = ADBC_STATUS_IO;
return na_res;
}
@@ -578,10 +580,10 @@ int TupleReader::AppendRowAndFetchNext(struct ArrowError* error) {
data_.data.as_char = pgbuf_;
if (get_copy_res == -2) {
- StringBuilderAppend(&error_builder_,
- "[libpq] PQgetCopyData failed at row %" PRId64 ": %s", row_id_,
- PQerrorMessage(conn_));
- return EIO;
+ SetError(&error_, "[libpq] PQgetCopyData failed at row %" PRId64 ": %s", row_id_,
+ PQerrorMessage(conn_));
+ status_ = ADBC_STATUS_IO;
+ return AdbcStatusCodeToErrno(status_);
} else if (get_copy_res == -1) {
// Returned when COPY has finished successfully
return ENODATA;
@@ -603,8 +605,8 @@ int TupleReader::BuildOutput(struct ArrowArray* out, struct ArrowError* error) {
int na_res = copy_reader_->GetArray(out, error);
if (na_res != NANOARROW_OK) {
- StringBuilderAppend(&error_builder_, "[libpq] Failed to build result array: %s",
- error->message);
+ SetError(&error_, "[libpq] Failed to build result array: %s", error->message);
+ status_ = ADBC_STATUS_INTERNAL;
return na_res;
}
@@ -625,7 +627,11 @@ void TupleReader::ResetQuery() {
}
// Clear the error builder
- error_builder_.size = 0;
+ if (error_.release) {
+ error_.release(&error_);
+ }
+ error_ = ADBC_ERROR_INIT;
+ status_ = ADBC_STATUS_OK;
row_id_ = -1;
}
@@ -673,17 +679,19 @@ int TupleReader::GetNext(struct ArrowArray* out) {
const ExecStatusType pq_status = PQresultStatus(result_);
if (pq_status != PGRES_COMMAND_OK) {
const char* sqlstate = PQresultErrorField(result_, PG_DIAG_SQLSTATE);
- StringBuilderAppend(&error_builder_, "[libpq] Query failed [%s]: %s",
- PQresStatus(pq_status), PQresultErrorMessage(result_));
+ SetError(&error_, result_, "[libpq] Query failed [%s]: %s", PQresStatus(pq_status),
+ PQresultErrorMessage(result_));
if (tmp.release != nullptr) {
tmp.release(&tmp);
}
if (sqlstate != nullptr && std::strcmp(sqlstate, "57014") == 0) {
- return ECANCELED;
+ status_ = ADBC_STATUS_CANCELLED;
+ } else {
+ status_ = ADBC_STATUS_IO;
}
- return EIO;
+ return AdbcStatusCodeToErrno(status_);
}
ResetQuery();
@@ -692,7 +700,11 @@ int TupleReader::GetNext(struct ArrowArray* out) {
}
void TupleReader::Release() {
- StringBuilderReset(&error_builder_);
+ if (error_.release) {
+ error_.release(&error_);
+ error_ = ADBC_ERROR_INIT;
+ }
+ status_ = ADBC_STATUS_OK;
if (result_) {
PQclear(result_);
@@ -713,6 +725,19 @@ void TupleReader::ExportTo(struct ArrowArrayStream* stream) {
stream->private_data = this;
}
+const struct AdbcError* TupleReader::ErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ if (!stream->private_data || stream->release != &ReleaseTrampoline) {
+ return nullptr;
+ }
+
+ TupleReader* reader = static_cast<TupleReader*>(stream->private_data);
+ if (status) {
+ *status = reader->status_;
+ }
+ return &reader->error_;
+}
+
int TupleReader::GetSchemaTrampoline(struct ArrowArrayStream* self,
struct ArrowSchema* out) {
if (!self || !self->private_data) return EINVAL;
diff --git a/c/driver/postgresql/statement.h b/c/driver/postgresql/statement.h
index eca39971..8c6c560c 100644
--- a/c/driver/postgresql/statement.h
+++ b/c/driver/postgresql/statement.h
@@ -41,29 +41,27 @@ class PostgresStatement;
class TupleReader final {
public:
TupleReader(PGconn* conn)
- : conn_(conn),
+ : status_(ADBC_STATUS_OK),
+ error_(ADBC_ERROR_INIT),
+ conn_(conn),
result_(nullptr),
pgbuf_(nullptr),
copy_reader_(nullptr),
row_id_(-1),
batch_size_hint_bytes_(16777216) {
- StringBuilderInit(&error_builder_, 0);
data_.data.as_char = nullptr;
data_.size_bytes = 0;
}
int GetSchema(struct ArrowSchema* out);
int GetNext(struct ArrowArray* out);
- const char* last_error() const {
- if (error_builder_.size > 0) {
- return error_builder_.buffer;
- } else {
- return nullptr;
- }
- }
+ const char* last_error() const { return error_.message; }
void Release();
void ExportTo(struct ArrowArrayStream* stream);
+ static const struct AdbcError* ErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status);
+
private:
friend class PostgresStatement;
@@ -77,11 +75,12 @@ class TupleReader final {
static const char* GetLastErrorTrampoline(struct ArrowArrayStream* self);
static void ReleaseTrampoline(struct ArrowArrayStream* self);
+ AdbcStatusCode status_;
+ struct AdbcError error_;
PGconn* conn_;
PGresult* result_;
char* pgbuf_;
struct ArrowBufferView data_;
- struct StringBuilder error_builder_;
std::unique_ptr<PostgresCopyStreamReader> copy_reader_;
int64_t row_id_;
int64_t batch_size_hint_bytes_;
diff --git a/c/driver_manager/adbc_driver_manager.cc b/c/driver_manager/adbc_driver_manager.cc
index d5579303..abd06a56 100644
--- a/c/driver_manager/adbc_driver_manager.cc
+++ b/c/driver_manager/adbc_driver_manager.cc
@@ -128,14 +128,71 @@ static AdbcStatusCode ReleaseDriver(struct AdbcDriver* driver, struct AdbcError*
return status;
}
+// ArrowArrayStream wrapper to support AdbcErrorFromArrayStream
+
+struct ErrorArrayStream {
+ struct ArrowArrayStream stream;
+ struct AdbcDriver* private_driver;
+};
+
+void ErrorArrayStreamRelease(struct ArrowArrayStream* stream) {
+ if (stream->release != ErrorArrayStreamRelease || !stream->private_data) return;
+
+ auto* private_data = reinterpret_cast<struct ErrorArrayStream*>(stream->private_data);
+ private_data->stream.release(&private_data->stream);
+ delete private_data;
+ std::memset(stream, 0, sizeof(*stream));
+}
+
+const char* ErrorArrayStreamGetLastError(struct ArrowArrayStream* stream) {
+ if (stream->release != ErrorArrayStreamRelease || !stream->private_data) return nullptr;
+ auto* private_data = reinterpret_cast<struct ErrorArrayStream*>(stream->private_data);
+ return private_data->stream.get_last_error(&private_data->stream);
+}
+
+int ErrorArrayStreamGetNext(struct ArrowArrayStream* stream, struct ArrowArray* array) {
+ if (stream->release != ErrorArrayStreamRelease || !stream->private_data) return EINVAL;
+ auto* private_data = reinterpret_cast<struct ErrorArrayStream*>(stream->private_data);
+ return private_data->stream.get_next(&private_data->stream, array);
+}
+
+int ErrorArrayStreamGetSchema(struct ArrowArrayStream* stream,
+ struct ArrowSchema* schema) {
+ if (stream->release != ErrorArrayStreamRelease || !stream->private_data) return EINVAL;
+ auto* private_data = reinterpret_cast<struct ErrorArrayStream*>(stream->private_data);
+ return private_data->stream.get_schema(&private_data->stream, schema);
+}
+
// Default stubs
-int ErrorGetDetailCount(struct AdbcError* error) { return 0; }
+int ErrorGetDetailCount(const struct AdbcError* error) { return 0; }
-struct AdbcErrorDetail ErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail ErrorGetDetail(const struct AdbcError* error, int index) {
return {nullptr, nullptr, 0};
}
+const struct AdbcError* ErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ return nullptr;
+}
+
+void ErrorArrayStreamInit(struct ArrowArrayStream* out,
+ struct AdbcDriver* private_driver) {
+ if (!out || !out->release ||
+ // Don't bother wrapping if driver didn't claim support
+ private_driver->ErrorFromArrayStream == ErrorFromArrayStream) {
+ return;
+ }
+ struct ErrorArrayStream* private_data = new ErrorArrayStream;
+ private_data->stream = *out;
+ private_data->private_driver = private_driver;
+ out->get_last_error = ErrorArrayStreamGetLastError;
+ out->get_next = ErrorArrayStreamGetNext;
+ out->get_schema = ErrorArrayStreamGetSchema;
+ out->release = ErrorArrayStreamRelease;
+ out->private_data = private_data;
+}
+
AdbcStatusCode DatabaseGetOption(struct AdbcDatabase* database, const char* key,
char* value, size_t* length, struct AdbcError* error) {
return ADBC_STATUS_NOT_FOUND;
@@ -383,7 +440,7 @@ struct TempConnection {
// Direct implementations of API methods
-int AdbcErrorGetDetailCount(struct AdbcError* error) {
+int AdbcErrorGetDetailCount(const struct AdbcError* error) {
if (error->vendor_code == ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA && error->private_data &&
error->private_driver) {
return error->private_driver->ErrorGetDetailCount(error);
@@ -391,7 +448,7 @@ int AdbcErrorGetDetailCount(struct AdbcError* error) {
return 0;
}
-struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail AdbcErrorGetDetail(const struct AdbcError* error, int index) {
if (error->vendor_code == ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA && error->private_data &&
error->private_driver) {
return error->private_driver->ErrorGetDetail(error, index);
@@ -399,11 +456,30 @@ struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index) {
return {nullptr, nullptr, 0};
}
+const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ if (!stream->private_data || stream->release != ErrorArrayStreamRelease) {
+ return nullptr;
+ }
+ auto* private_data = reinterpret_cast<struct ErrorArrayStream*>(stream->private_data);
+ return private_data->private_driver->ErrorFromArrayStream(&private_data->stream,
+ status);
+}
+
#define INIT_ERROR(ERROR, SOURCE) \
if ((ERROR)->vendor_code == ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA) { \
(ERROR)->private_driver = (SOURCE)->private_driver; \
}
+#define WRAP_STREAM(EXPR, OUT, SOURCE) \
+ if (!(OUT)) { \
+ /* Happens for ExecuteQuery where out is optional */ \
+ return EXPR; \
+ } \
+ AdbcStatusCode status_code = EXPR; \
+ ErrorArrayStreamInit(OUT, (SOURCE)->private_driver); \
+ return status_code;
+
AdbcStatusCode AdbcDatabaseNew(struct AdbcDatabase* database, struct AdbcError* error) {
// Allocate a temporary structure to store options pre-Init
database->private_data = new TempDatabase();
@@ -700,8 +776,9 @@ AdbcStatusCode AdbcConnectionGetInfo(struct AdbcConnection* connection,
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionGetInfo(connection, info_codes,
- info_codes_length, out, error);
+ WRAP_STREAM(connection->private_driver->ConnectionGetInfo(
+ connection, info_codes, info_codes_length, out, error),
+ out, connection);
}
AdbcStatusCode AdbcConnectionGetObjects(struct AdbcConnection* connection, int depth,
@@ -714,9 +791,10 @@ AdbcStatusCode AdbcConnectionGetObjects(struct AdbcConnection* connection, int d
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionGetObjects(
- connection, depth, catalog, db_schema, table_name, table_types, column_name, stream,
- error);
+ WRAP_STREAM(connection->private_driver->ConnectionGetObjects(
+ connection, depth, catalog, db_schema, table_name, table_types,
+ column_name, stream, error),
+ stream, connection);
}
AdbcStatusCode AdbcConnectionGetOption(struct AdbcConnection* connection, const char* key,
@@ -822,8 +900,10 @@ AdbcStatusCode AdbcConnectionGetStatistics(struct AdbcConnection* connection,
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionGetStatistics(
- connection, catalog, db_schema, table_name, approximate == 1, out, error);
+ WRAP_STREAM(
+ connection->private_driver->ConnectionGetStatistics(
+ connection, catalog, db_schema, table_name, approximate == 1, out, error),
+ out, connection);
}
AdbcStatusCode AdbcConnectionGetStatisticNames(struct AdbcConnection* connection,
@@ -833,7 +913,9 @@ AdbcStatusCode AdbcConnectionGetStatisticNames(struct AdbcConnection* connection
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionGetStatisticNames(connection, out, error);
+ WRAP_STREAM(
+ connection->private_driver->ConnectionGetStatisticNames(connection, out, error),
+ out, connection);
}
AdbcStatusCode AdbcConnectionGetTableSchema(struct AdbcConnection* connection,
@@ -856,7 +938,9 @@ AdbcStatusCode AdbcConnectionGetTableTypes(struct AdbcConnection* connection,
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionGetTableTypes(connection, stream, error);
+ WRAP_STREAM(
+ connection->private_driver->ConnectionGetTableTypes(connection, stream, error),
+ stream, connection);
}
AdbcStatusCode AdbcConnectionInit(struct AdbcConnection* connection,
@@ -928,8 +1012,9 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionReadPartition(
- connection, serialized_partition, serialized_length, out, error);
+ WRAP_STREAM(connection->private_driver->ConnectionReadPartition(
+ connection, serialized_partition, serialized_length, out, error),
+ out, connection);
}
AdbcStatusCode AdbcConnectionRelease(struct AdbcConnection* connection,
@@ -1079,8 +1164,9 @@ AdbcStatusCode AdbcStatementExecuteQuery(struct AdbcStatement* statement,
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, statement);
- return statement->private_driver->StatementExecuteQuery(statement, out, rows_affected,
- error);
+ WRAP_STREAM(statement->private_driver->StatementExecuteQuery(statement, out,
+ rows_affected, error),
+ out, statement);
}
AdbcStatusCode AdbcStatementExecuteSchema(struct AdbcStatement* statement,
@@ -1484,6 +1570,7 @@ AdbcStatusCode AdbcLoadDriverFromInitFunc(AdbcDriverInitFunc init_func, int vers
auto* driver = reinterpret_cast<struct AdbcDriver*>(raw_driver);
FILL_DEFAULT(driver, ErrorGetDetailCount);
FILL_DEFAULT(driver, ErrorGetDetail);
+ FILL_DEFAULT(driver, ErrorFromArrayStream);
FILL_DEFAULT(driver, DatabaseGetOption);
FILL_DEFAULT(driver, DatabaseGetOptionBytes);
diff --git a/c/integration/duckdb/CMakeLists.txt b/c/integration/duckdb/CMakeLists.txt
index 52fb9d0f..8053713f 100644
--- a/c/integration/duckdb/CMakeLists.txt
+++ b/c/integration/duckdb/CMakeLists.txt
@@ -49,6 +49,7 @@ if(ADBC_BUILD_TESTS)
CACHE INTERNAL "Disable UBSAN")
# Force cmake to honor our options here in the subproject
cmake_policy(SET CMP0077 NEW)
+ message(STATUS "Fetching DuckDB")
fetchcontent_makeavailable(duckdb)
include_directories(SYSTEM ${REPOSITORY_ROOT})
diff --git a/c/integration/duckdb/duckdb_test.cc b/c/integration/duckdb/duckdb_test.cc
index be163418..c42c3813 100644
--- a/c/integration/duckdb/duckdb_test.cc
+++ b/c/integration/duckdb/duckdb_test.cc
@@ -98,6 +98,10 @@ class DuckDbStatementTest : public ::testing::Test,
void TestSqlQueryErrors() { GTEST_SKIP() << "DuckDB does not set AdbcError.release"; }
+ void TestErrorCompatibility() {
+ GTEST_SKIP() << "DuckDB does not set AdbcError.release";
+ }
+
protected:
DuckDbQuirks quirks_;
};
diff --git a/dev/release/verify-release-candidate.sh b/dev/release/verify-release-candidate.sh
index bb5dae2d..1ce64fde 100755
--- a/dev/release/verify-release-candidate.sh
+++ b/dev/release/verify-release-candidate.sh
@@ -23,7 +23,7 @@
# - Maven >= 3.3.9
# - JDK >=7
# - gcc >= 4.8
-# - Go >= 1.18
+# - Go >= 1.20
# - Docker
#
# To reuse build artifacts between runs set ARROW_TMPDIR environment variable to
@@ -254,7 +254,7 @@ install_go() {
return 0
fi
- local version=1.18.8
+ local version=1.20.7
show_info "Installing go version ${version}..."
local arch="$(uname -m)"
@@ -411,7 +411,7 @@ test_cpp() {
maybe_setup_conda \
--file ci/conda_env_cpp.txt \
compilers \
- go=1.18 || exit 1
+ go=1.20 || exit 1
if [ "${USE_CONDA}" -gt 0 ]; then
export CMAKE_PREFIX_PATH="${CONDA_BACKUP_CMAKE_PREFIX_PATH}:${CMAKE_PREFIX_PATH}"
@@ -561,7 +561,7 @@ test_go() {
# apache/arrow-adbc#517: `go build` calls git. Don't assume system
# has git; even if it's there, go_build.sh sets DYLD_LIBRARY_PATH
# which can interfere with system git.
- maybe_setup_conda compilers git go=1.18 || exit 1
+ maybe_setup_conda compilers git go=1.20 || exit 1
if [ "${USE_CONDA}" -gt 0 ]; then
# The CMake setup forces RPATH to be the Conda prefix
diff --git a/docker-compose.yml b/docker-compose.yml
index e65d9359..853aeaf5 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -30,19 +30,6 @@ services:
command: |
/bin/bash -c 'git config --global --add safe.directory /adbc && source /opt/conda/etc/profile.d/conda.sh && mamba create -y -n adbc -c conda-forge go --file /adbc/ci/conda_env_cpp.txt --file /adbc/ci/conda_env_docs.txt --file /adbc/ci/conda_env_python.txt && conda activate adbc && env ADBC_USE_ASAN=0 ADBC_USE_UBSAN=0 /adbc/ci/scripts/cpp_build.sh /adbc /adbc/build && env CGO_ENABLED=1 /adbc/ci/scripts/go_build.sh /adbc /adbc/build && /adbc/ci/scripts/python_build.sh /adbc /adbc/bui [...]
- golang-sqlite-flightsql:
- image: ${REPO}:golang-${GO}-sqlite-flightsql
- build:
- context: .
- cache_from:
- - ${REPO}:golang-${GO}-sqlite-flightsql
- dockerfile: ci/docker/golang-flightsql-sqlite.dockerfile
- args:
- GO: ${GO}
- ARROW_MAJOR_VERSION: ${ARROW_MAJOR_VERSION}
- ports:
- - 8080:8080
-
############################ Java JARs ######################################
java-dist:
@@ -162,3 +149,16 @@ services:
entrypoint: "/init/bootstrap.sh"
volumes:
- "./ci/scripts/integration/dremio:/init"
+
+ golang-sqlite-flightsql:
+ image: ${REPO}:golang-${GO}-sqlite-flightsql
+ build:
+ context: .
+ cache_from:
+ - ${REPO}:golang-${GO}-sqlite-flightsql
+ dockerfile: ci/docker/golang-flightsql-sqlite.dockerfile
+ args:
+ GO: ${GO}
+ ARROW_MAJOR_VERSION: ${ARROW_MAJOR_VERSION}
+ ports:
+ - 8080:8080
diff --git a/go/adbc/driver/flightsql/flightsql_adbc.go b/go/adbc/driver/flightsql/flightsql_adbc.go
index 13fe1062..03d24e2f 100644
--- a/go/adbc/driver/flightsql/flightsql_adbc.go
+++ b/go/adbc/driver/flightsql/flightsql_adbc.go
@@ -727,12 +727,10 @@ func getFlightClient(ctx context.Context, loc string, d *database) (*flightsql.C
cl.Alloc = d.alloc
if d.user != "" || d.pass != "" {
- ctx, err = cl.Client.AuthenticateBasicToken(ctx, d.user, d.pass)
+ var header, trailer metadata.MD
+ ctx, err = cl.Client.AuthenticateBasicToken(ctx, d.user, d.pass, grpc.Header(&header), grpc.Trailer(&trailer), d.timeout)
if err != nil {
- return nil, adbc.Error{
- Msg: err.Error(),
- Code: adbc.StatusUnauthenticated,
- }
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "AuthenticateBasicToken")
}
if md, ok := metadata.FromOutgoingContext(ctx); ok {
@@ -1129,12 +1127,14 @@ func (c *cnxn) GetInfo(ctx context.Context, infoCodes []adbc.InfoCode) (array.Re
}
ctx = metadata.NewOutgoingContext(ctx, c.hdrs)
- info, err := c.cl.GetSqlInfo(ctx, translated, c.timeouts)
+ var header, trailer metadata.MD
+ info, err := c.cl.GetSqlInfo(ctx, translated, grpc.Header(&header), grpc.Trailer(&trailer), c.timeouts)
if err == nil {
for i, endpoint := range info.Endpoint {
- rdr, err := doGet(ctx, c.cl, endpoint, c.clientCache, c.timeouts)
+ var header, trailer metadata.MD
+ rdr, err := doGet(ctx, c.cl, endpoint, c.clientCache, grpc.Header(&header), grpc.Trailer(&trailer), c.timeouts)
if err != nil {
- return nil, adbcFromFlightStatus(err, "GetInfo(DoGet): endpoint %d: %s", i, endpoint.Location)
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetInfo(DoGet): endpoint %d: %s", i, endpoint.Location)
}
for rdr.Next() {
@@ -1163,7 +1163,9 @@ func (c *cnxn) GetInfo(ctx context.Context, infoCodes []adbc.InfoCode) (array.Re
}
if rdr.Err() != nil {
- return nil, adbcFromFlightStatus(rdr.Err(), "GetInfo(DoGet): endpoint %d: %s", i, endpoint.Location)
+ return nil, adbcFromFlightStatusWithDetails(rdr.Err(), header, trailer, "GetInfo(DoGet): endpoint %d: %s", i, endpoint.Location)
+ } else if ctx.Err() != nil {
+ return nil, adbcFromFlightStatusWithDetails(context.Cause(ctx), header, trailer, "GetInfo(DoGet): endpoint %d: %s", i, endpoint.Location)
}
}
} else if grpcstatus.Code(err) != grpccodes.Unimplemented {
@@ -1270,15 +1272,18 @@ func (c *cnxn) GetObjects(ctx context.Context, depth adbc.ObjectDepth, catalog *
}
defer g.Release()
+ var header, trailer metadata.MD
// To avoid an N+1 query problem, we assume result sets here will fit in memory and build up a single response.
- info, err := c.cl.GetCatalogs(ctx)
+ info, err := c.cl.GetCatalogs(ctx, grpc.Header(&header), grpc.Trailer(&trailer), c.timeouts)
if err != nil {
- return nil, adbcFromFlightStatus(err, "GetObjects(GetCatalogs)")
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetObjects(GetCatalogs)")
}
- rdr, err := c.readInfo(ctx, schema_ref.Catalogs, info)
+ header = metadata.MD{}
+ trailer = metadata.MD{}
+ rdr, err := c.readInfo(ctx, schema_ref.Catalogs, info, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer))
if err != nil {
- return nil, adbcFromFlightStatus(err, "GetObjects(GetCatalogs)")
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetObjects(GetCatalogs)")
}
defer rdr.Release()
@@ -1299,16 +1304,18 @@ func (c *cnxn) GetObjects(ctx context.Context, depth adbc.ObjectDepth, catalog *
}
if err = rdr.Err(); err != nil {
- return nil, adbcFromFlightStatus(err, "GetObjects(GetCatalogs)")
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetObjects(GetCatalogs)")
+ } else if ctx.Err() != nil {
+ return nil, adbcFromFlightStatusWithDetails(context.Cause(ctx), header, trailer, "GetObjects(GetCatalogs)")
}
return g.Finish()
}
// Helper function to read and validate a metadata stream
-func (c *cnxn) readInfo(ctx context.Context, expectedSchema *arrow.Schema, info *flight.FlightInfo) (array.RecordReader, error) {
+func (c *cnxn) readInfo(ctx context.Context, expectedSchema *arrow.Schema, info *flight.FlightInfo, opts ...grpc.CallOption) (array.RecordReader, error) {
// use a default queueSize for the reader
- rdr, err := newRecordReader(ctx, c.db.alloc, c.cl, info, c.clientCache, 5)
+ rdr, err := newRecordReader(ctx, c.db.alloc, c.cl, info, c.clientCache, 5, opts...)
if err != nil {
return nil, adbcFromFlightStatus(err, "DoGet")
}
@@ -1329,15 +1336,18 @@ func (c *cnxn) getObjectsDbSchemas(ctx context.Context, depth adbc.ObjectDepth,
return
}
result = make(map[string][]string)
+ var header, trailer metadata.MD
// Pre-populate the map of which schemas are in which catalogs
- info, err := c.cl.GetDBSchemas(ctx, &flightsql.GetDBSchemasOpts{DbSchemaFilterPattern: dbSchema})
+ info, err := c.cl.GetDBSchemas(ctx, &flightsql.GetDBSchemasOpts{DbSchemaFilterPattern: dbSchema}, grpc.Header(&header), grpc.Trailer(&trailer), c.timeouts)
if err != nil {
- return nil, adbcFromFlightStatus(err, "GetObjects(GetDBSchemas)")
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetObjects(GetDBSchemas)")
}
- rdr, err := c.readInfo(ctx, schema_ref.DBSchemas, info)
+ header = metadata.MD{}
+ trailer = metadata.MD{}
+ rdr, err := c.readInfo(ctx, schema_ref.DBSchemas, info, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer))
if err != nil {
- return nil, adbcFromFlightStatus(err, "GetObjects(GetDBSchemas)")
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetObjects(GetDBSchemas)")
}
defer rdr.Release()
@@ -1358,7 +1368,9 @@ func (c *cnxn) getObjectsDbSchemas(ctx context.Context, depth adbc.ObjectDepth,
if rdr.Err() != nil {
result = nil
- err = adbcFromFlightStatus(rdr.Err(), "GetObjects(GetDBSchemas)")
+ err = adbcFromFlightStatusWithDetails(rdr.Err(), header, trailer, "GetObjects(GetDBSchemas)")
+ } else if ctx.Err() != nil {
+ return nil, adbcFromFlightStatusWithDetails(context.Cause(ctx), header, trailer, "GetObjects(GetCatalogs)")
}
return
}
@@ -1371,21 +1383,24 @@ func (c *cnxn) getObjectsTables(ctx context.Context, depth adbc.ObjectDepth, cat
// Pre-populate the map of which schemas are in which catalogs
includeSchema := depth == adbc.ObjectDepthAll || depth == adbc.ObjectDepthColumns
+ var header, trailer metadata.MD
info, err := c.cl.GetTables(ctx, &flightsql.GetTablesOpts{
DbSchemaFilterPattern: dbSchema,
TableNameFilterPattern: tableName,
TableTypes: tableType,
IncludeSchema: includeSchema,
- })
+ }, grpc.Header(&header), grpc.Trailer(&trailer), c.timeouts)
if err != nil {
- return nil, adbcFromFlightStatus(err, "GetObjects(GetTables)")
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetObjects(GetTables)")
}
expectedSchema := schema_ref.Tables
if includeSchema {
expectedSchema = schema_ref.TablesWithIncludedSchema
}
- rdr, err := c.readInfo(ctx, expectedSchema, info)
+ header = metadata.MD{}
+ trailer = metadata.MD{}
+ rdr, err := c.readInfo(ctx, expectedSchema, info, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer))
if err != nil {
return nil, adbcFromFlightStatus(err, "GetObjects(GetTables)")
}
@@ -1436,7 +1451,10 @@ func (c *cnxn) getObjectsTables(ctx context.Context, depth adbc.ObjectDepth, cat
if rdr.Err() != nil {
result = nil
- err = adbcFromFlightStatus(rdr.Err(), "GetObjects(GetTables)")
+ err = adbcFromFlightStatusWithDetails(rdr.Err(), header, trailer, "GetObjects(GetTables)")
+ } else if ctx.Err() != nil {
+ result = nil
+ err = adbcFromFlightStatusWithDetails(context.Cause(ctx), header, trailer, "GetObjects(GetTables)")
}
return
}
@@ -1450,14 +1468,17 @@ func (c *cnxn) GetTableSchema(ctx context.Context, catalog *string, dbSchema *st
}
ctx = metadata.NewOutgoingContext(ctx, c.hdrs)
- info, err := c.cl.GetTables(ctx, opts, c.timeouts)
+ var header, trailer metadata.MD
+ info, err := c.cl.GetTables(ctx, opts, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer))
if err != nil {
- return nil, adbcFromFlightStatus(err, "GetTableSchema(GetTables)")
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetTableSchema(GetTables)")
}
- rdr, err := doGet(ctx, c.cl, info.Endpoint[0], c.clientCache, c.timeouts)
+ header = metadata.MD{}
+ trailer = metadata.MD{}
+ rdr, err := doGet(ctx, c.cl, info.Endpoint[0], c.clientCache, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer))
if err != nil {
- return nil, adbcFromFlightStatus(err, "GetTableSchema(DoGet)")
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetTableSchema(DoGet)")
}
defer rdr.Release()
@@ -1469,7 +1490,7 @@ func (c *cnxn) GetTableSchema(ctx context.Context, catalog *string, dbSchema *st
Code: adbc.StatusNotFound,
}
}
- return nil, adbcFromFlightStatus(err, "GetTableSchema(DoGet)")
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetTableSchema(DoGet)")
}
if rec.NumRows() == 0 {
@@ -1501,9 +1522,10 @@ func (c *cnxn) GetTableSchema(ctx context.Context, catalog *string, dbSchema *st
// table_type | utf8 not null
func (c *cnxn) GetTableTypes(ctx context.Context) (array.RecordReader, error) {
ctx = metadata.NewOutgoingContext(ctx, c.hdrs)
- info, err := c.cl.GetTableTypes(ctx, c.timeouts)
+ var header, trailer metadata.MD
+ info, err := c.cl.GetTableTypes(ctx, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer))
if err != nil {
- return nil, adbcFromFlightStatus(err, "GetTableTypes")
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetTableTypes")
}
return newRecordReader(ctx, c.db.alloc, c.cl, info, c.clientCache, 5)
@@ -1528,14 +1550,17 @@ func (c *cnxn) Commit(ctx context.Context) error {
}
ctx = metadata.NewOutgoingContext(ctx, c.hdrs)
- err := c.txn.Commit(ctx, c.timeouts)
+ var header, trailer metadata.MD
+ err := c.txn.Commit(ctx, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer))
if err != nil {
- return adbcFromFlightStatus(err, "Commit")
+ return adbcFromFlightStatusWithDetails(err, header, trailer, "Commit")
}
- c.txn, err = c.cl.BeginTransaction(ctx, c.timeouts)
+ header = metadata.MD{}
+ trailer = metadata.MD{}
+ c.txn, err = c.cl.BeginTransaction(ctx, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer))
if err != nil {
- return adbcFromFlightStatus(err, "BeginTransaction")
+ return adbcFromFlightStatusWithDetails(err, header, trailer, "BeginTransaction")
}
return nil
}
@@ -1559,14 +1584,17 @@ func (c *cnxn) Rollback(ctx context.Context) error {
}
ctx = metadata.NewOutgoingContext(ctx, c.hdrs)
- err := c.txn.Rollback(ctx, c.timeouts)
+ var header, trailer metadata.MD
+ err := c.txn.Rollback(ctx, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer))
if err != nil {
- return adbcFromFlightStatus(err, "Rollback")
+ return adbcFromFlightStatusWithDetails(err, header, trailer, "Rollback")
}
- c.txn, err = c.cl.BeginTransaction(ctx, c.timeouts)
+ header = metadata.MD{}
+ trailer = metadata.MD{}
+ c.txn, err = c.cl.BeginTransaction(ctx, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer))
if err != nil {
- return adbcFromFlightStatus(err, "BeginTransaction")
+ return adbcFromFlightStatusWithDetails(err, header, trailer, "BeginTransaction")
}
return nil
}
@@ -1658,7 +1686,7 @@ func (c *cnxn) Close() error {
err := c.cl.Close()
c.cl = nil
- return err
+ return adbcFromFlightStatus(err, "Close")
}
// ReadPartition constructs a statement for a partition of a query. The
diff --git a/go/adbc/driver/flightsql/flightsql_statement.go b/go/adbc/driver/flightsql/flightsql_statement.go
index d0305320..1dc70744 100644
--- a/go/adbc/driver/flightsql/flightsql_statement.go
+++ b/go/adbc/driver/flightsql/flightsql_statement.go
@@ -135,7 +135,9 @@ type statement struct {
}
func (s *statement) closePreparedStatement() error {
- return s.prepared.Close(metadata.NewOutgoingContext(context.Background(), s.hdrs), s.timeouts)
+ var header, trailer metadata.MD
+ err := s.prepared.Close(metadata.NewOutgoingContext(context.Background(), s.hdrs), grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts)
+ return adbcFromFlightStatusWithDetails(err, header, trailer, "ClosePreparedStatement")
}
// Close releases any relevant resources associated with this statement
@@ -329,14 +331,16 @@ func (s *statement) SetSqlQuery(query string) error {
func (s *statement) ExecuteQuery(ctx context.Context) (rdr array.RecordReader, nrec int64, err error) {
ctx = metadata.NewOutgoingContext(ctx, s.hdrs)
var info *flight.FlightInfo
+ var header, trailer metadata.MD
+ opts := append([]grpc.CallOption{}, grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts)
if s.prepared != nil {
- info, err = s.prepared.Execute(ctx, s.timeouts)
+ info, err = s.prepared.Execute(ctx, opts...)
} else {
- info, err = s.query.execute(ctx, s.cnxn, s.timeouts)
+ info, err = s.query.execute(ctx, s.cnxn, opts...)
}
if err != nil {
- return nil, -1, adbcFromFlightStatus(err, "ExecuteQuery")
+ return nil, -1, adbcFromFlightStatusWithDetails(err, header, trailer, "ExecuteQuery")
}
nrec = info.TotalRecords
@@ -348,15 +352,16 @@ func (s *statement) ExecuteQuery(ctx context.Context) (rdr array.RecordReader, n
// set. It returns the number of rows affected if known, otherwise -1.
func (s *statement) ExecuteUpdate(ctx context.Context) (n int64, err error) {
ctx = metadata.NewOutgoingContext(ctx, s.hdrs)
-
+ var header, trailer metadata.MD
+ opts := append([]grpc.CallOption{}, grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts)
if s.prepared != nil {
- n, err = s.prepared.ExecuteUpdate(ctx, s.timeouts)
+ n, err = s.prepared.ExecuteUpdate(ctx, opts...)
} else {
- n, err = s.query.executeUpdate(ctx, s.cnxn, s.timeouts)
+ n, err = s.query.executeUpdate(ctx, s.cnxn, opts...)
}
if err != nil {
- err = adbcFromFlightStatus(err, "ExecuteUpdate")
+ err = adbcFromFlightStatusWithDetails(err, header, trailer, "ExecuteQuery")
}
return
@@ -366,9 +371,10 @@ func (s *statement) ExecuteUpdate(ctx context.Context) (n int64, err error) {
// multiple times. This invalidates any prior result sets.
func (s *statement) Prepare(ctx context.Context) error {
ctx = metadata.NewOutgoingContext(ctx, s.hdrs)
- prep, err := s.query.prepare(ctx, s.cnxn, s.timeouts)
+ var header, trailer metadata.MD
+ prep, err := s.query.prepare(ctx, s.cnxn, grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts)
if err != nil {
- return adbcFromFlightStatus(err, "Prepare")
+ return adbcFromFlightStatusWithDetails(err, header, trailer, "Prepare")
}
s.prepared = prep
return nil
@@ -484,14 +490,15 @@ func (s *statement) ExecutePartitions(ctx context.Context) (*arrow.Schema, adbc.
err error
)
+ var header, trailer metadata.MD
if s.prepared != nil {
- info, err = s.prepared.Execute(ctx, s.timeouts)
+ info, err = s.prepared.Execute(ctx, grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts)
} else {
- info, err = s.query.execute(ctx, s.cnxn, s.timeouts)
+ info, err = s.query.execute(ctx, s.cnxn, grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts)
}
if err != nil {
- return nil, out, -1, adbcFromFlightStatus(err, "ExecutePartitions")
+ return nil, out, -1, adbcFromFlightStatusWithDetails(err, header, trailer, "ExecutePartitions")
}
if len(info.Schema) > 0 {
@@ -535,5 +542,10 @@ func (s *statement) ExecuteSchema(ctx context.Context) (schema *arrow.Schema, er
return
}
- return s.query.executeSchema(ctx, s.cnxn, s.timeouts)
+ var header, trailer metadata.MD
+ schema, err = s.query.executeSchema(ctx, s.cnxn, grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts)
+ if err != nil {
+ err = adbcFromFlightStatusWithDetails(err, header, trailer, "ExecuteSchema")
+ }
+ return
}
diff --git a/go/adbc/driver/flightsql/record_reader.go b/go/adbc/driver/flightsql/record_reader.go
index fa62a052..9014be75 100644
--- a/go/adbc/driver/flightsql/record_reader.go
+++ b/go/adbc/driver/flightsql/record_reader.go
@@ -32,6 +32,7 @@ import (
"github.com/bluele/gcache"
"golang.org/x/sync/errgroup"
"google.golang.org/grpc"
+ "google.golang.org/grpc/metadata"
)
type reader struct {
@@ -49,6 +50,8 @@ type reader struct {
// gathers all of the records as they come in.
func newRecordReader(ctx context.Context, alloc memory.Allocator, cl *flightsql.Client, info *flight.FlightInfo, clCache gcache.Cache, bufferSize int, opts ...grpc.CallOption) (rdr array.RecordReader, err error) {
endpoints := info.Endpoint
+ var header, trailer metadata.MD
+ opts = append(append([]grpc.CallOption{}, opts...), grpc.Header(&header), grpc.Trailer(&trailer))
var schema *arrow.Schema
if len(endpoints) == 0 {
if info.Schema == nil {
@@ -91,7 +94,7 @@ func newRecordReader(ctx context.Context, alloc memory.Allocator, cl *flightsql.
firstEndpoint := endpoints[0]
rdr, err := doGet(ctx, cl, firstEndpoint, clCache, opts...)
if err != nil {
- return nil, adbcFromFlightStatus(err, "DoGet: endpoint 0: remote: %s", firstEndpoint.Location)
+ return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "DoGet: endpoint 0: remote: %s", firstEndpoint.Location)
}
schema = rdr.Schema()
group.Go(func() error {
@@ -105,7 +108,12 @@ func newRecordReader(ctx context.Context, alloc memory.Allocator, cl *flightsql.
rec.Retain()
ch <- rec
}
- return adbcFromFlightStatus(rdr.Err(), "DoGet: endpoint 0: remote: %s", firstEndpoint.Location)
+ if rdr.Err() != nil {
+ return adbcFromFlightStatusWithDetails(rdr.Err(), header, trailer, "DoGet: endpoint 0: remote: %s", firstEndpoint.Location)
+ } else if ctx.Err() != nil {
+ return adbcFromFlightStatusWithDetails(context.Cause(ctx), header, trailer, "DoGet: endpoint 0: remote: %s", firstEndpoint.Location)
+ }
+ return nil
})
endpoints = endpoints[1:]
@@ -136,7 +144,7 @@ func newRecordReader(ctx context.Context, alloc memory.Allocator, cl *flightsql.
rdr, err := doGet(ctx, cl, endpoint, clCache, opts...)
if err != nil {
- return adbcFromFlightStatus(err, "DoGet: endpoint %d: %s", endpointIndex, endpoint.Location)
+ return adbcFromFlightStatusWithDetails(err, header, trailer, "DoGet: endpoint %d: %s", endpointIndex, endpoint.Location)
}
defer rdr.Release()
@@ -151,7 +159,12 @@ func newRecordReader(ctx context.Context, alloc memory.Allocator, cl *flightsql.
chs[endpointIndex] <- rec
}
- return adbcFromFlightStatus(rdr.Err(), "DoGet: endpoint %d: %s", endpointIndex, endpoint.Location)
+ if rdr.Err() != nil {
+ return adbcFromFlightStatusWithDetails(rdr.Err(), header, trailer, "DoGet: endpoint %d: %s", endpointIndex, endpoint.Location)
+ } else if ctx.Err() != nil {
+ return adbcFromFlightStatusWithDetails(context.Cause(ctx), header, trailer, "DoGet: endpoint %d: %s", endpointIndex, endpoint.Location)
+ }
+ return nil
})
}
diff --git a/go/adbc/driver/flightsql/utils.go b/go/adbc/driver/flightsql/utils.go
index 8c1263f9..62d0b4ce 100644
--- a/go/adbc/driver/flightsql/utils.go
+++ b/go/adbc/driver/flightsql/utils.go
@@ -22,11 +22,17 @@ import (
"github.com/apache/arrow-adbc/go/adbc"
"google.golang.org/grpc/codes"
+ "google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/proto"
)
func adbcFromFlightStatus(err error, context string, args ...any) error {
+ var header, trailer metadata.MD
+ return adbcFromFlightStatusWithDetails(err, header, trailer, context, args...)
+}
+
+func adbcFromFlightStatusWithDetails(err error, header, trailer metadata.MD, context string, args ...any) error {
if _, ok := err.(adbc.Error); ok {
return err
}
@@ -73,7 +79,6 @@ func adbcFromFlightStatus(err error, context string, args ...any) error {
adbcCode = adbc.StatusUnknown
}
- // People don't read error messages, so backload the context and frontload the server error
details := []adbc.ErrorDetail{}
// slice of proto.Message or error
for _, detail := range grpcStatus.Details() {
@@ -81,12 +86,38 @@ func adbcFromFlightStatus(err error, context string, args ...any) error {
details = append(details, &adbc.TextErrorDetail{Name: "grpc-status-details-bin", Detail: err.Error()})
} else if msg, ok := detail.(proto.Message); ok {
details = append(details, &adbc.ProtobufErrorDetail{Name: "grpc-status-details-bin", Message: msg})
- } else {
- panic(fmt.Sprintf("gRPC returned non-Protobuf detail in violation of method contract: %#v", detail))
+ }
+ // else, gRPC returned non-Protobuf detail in violation of their method contract
+ }
+
+ // XXX(https://github.com/grpc/grpc-go/issues/5485): don't count on
+ // grpc-status-details-bin since Google hardcodes it to only work with
+ // Google Cloud
+ // XXX: must check both headers and trailers because some implementations
+ // (like gRPC-Java) will consolidate trailers into headers for failed RPCs
+ for key, values := range header {
+ switch key {
+ case "content-type", "grpc-status-details-bin":
+ continue
+ default:
+ for _, value := range values {
+ details = append(details, &adbc.TextErrorDetail{Name: key, Detail: value})
+ }
+ }
+ }
+ for key, values := range trailer {
+ switch key {
+ case "content-type", "grpc-status-details-bin":
+ continue
+ default:
+ for _, value := range values {
+ details = append(details, &adbc.TextErrorDetail{Name: key, Detail: value})
+ }
}
}
return adbc.Error{
+ // People don't read error messages, so backload the context and frontload the server error
Msg: fmt.Sprintf("[FlightSQL] %s (%s; %s)", grpcStatus.Message(), grpcStatus.Code(), fmt.Sprintf(context, args...)),
Code: adbcCode,
Details: details,
diff --git a/go/adbc/drivermgr/adbc.h b/go/adbc/drivermgr/adbc.h
index c38f2064..a2a33a4d 100644
--- a/go/adbc/drivermgr/adbc.h
+++ b/go/adbc/drivermgr/adbc.h
@@ -360,7 +360,7 @@ struct ADBC_EXPORT AdbcErrorDetail {
///
/// \since ADBC API revision 1.1.0
ADBC_EXPORT
-int AdbcErrorGetDetailCount(struct AdbcError* error);
+int AdbcErrorGetDetailCount(const struct AdbcError* error);
/// \brief Get a metadata value in an error by index.
///
@@ -369,7 +369,25 @@ int AdbcErrorGetDetailCount(struct AdbcError* error);
///
/// \since ADBC API revision 1.1.0
ADBC_EXPORT
-struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index);
+struct AdbcErrorDetail AdbcErrorGetDetail(const struct AdbcError* error, int index);
+
+/// \brief Get an ADBC error from an ArrowArrayStream created by a driver.
+///
+/// This allows retrieving error details and other metadata that would
+/// normally be suppressed by the Arrow C Stream Interface.
+///
+/// The caller MUST NOT release the error; it is managed by the release
+/// callback in the stream itself.
+///
+/// \param[in] stream The stream to query.
+/// \param[out] status The ADBC status code, or ADBC_STATUS_OK if there is no
+/// error. Not written to if the stream does not contain an ADBC error or
+/// if the pointer is NULL.
+/// \return NULL if not supported.
+/// \since ADBC API revision 1.1.0
+ADBC_EXPORT
+const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status);
/// @}
@@ -950,8 +968,10 @@ struct ADBC_EXPORT AdbcDriver {
///
/// @{
- int (*ErrorGetDetailCount)(struct AdbcError* error);
- struct AdbcErrorDetail (*ErrorGetDetail)(struct AdbcError* error, int index);
+ int (*ErrorGetDetailCount)(const struct AdbcError* error);
+ struct AdbcErrorDetail (*ErrorGetDetail)(const struct AdbcError* error, int index);
+ const struct AdbcError* (*ErrorFromArrayStream)(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status);
AdbcStatusCode (*DatabaseGetOption)(struct AdbcDatabase*, const char*, char*, size_t*,
struct AdbcError*);
diff --git a/go/adbc/drivermgr/adbc_driver_manager.cc b/go/adbc/drivermgr/adbc_driver_manager.cc
index d5579303..abd06a56 100644
--- a/go/adbc/drivermgr/adbc_driver_manager.cc
+++ b/go/adbc/drivermgr/adbc_driver_manager.cc
@@ -128,14 +128,71 @@ static AdbcStatusCode ReleaseDriver(struct AdbcDriver* driver, struct AdbcError*
return status;
}
+// ArrowArrayStream wrapper to support AdbcErrorFromArrayStream
+
+struct ErrorArrayStream {
+ struct ArrowArrayStream stream;
+ struct AdbcDriver* private_driver;
+};
+
+void ErrorArrayStreamRelease(struct ArrowArrayStream* stream) {
+ if (stream->release != ErrorArrayStreamRelease || !stream->private_data) return;
+
+ auto* private_data = reinterpret_cast<struct ErrorArrayStream*>(stream->private_data);
+ private_data->stream.release(&private_data->stream);
+ delete private_data;
+ std::memset(stream, 0, sizeof(*stream));
+}
+
+const char* ErrorArrayStreamGetLastError(struct ArrowArrayStream* stream) {
+ if (stream->release != ErrorArrayStreamRelease || !stream->private_data) return nullptr;
+ auto* private_data = reinterpret_cast<struct ErrorArrayStream*>(stream->private_data);
+ return private_data->stream.get_last_error(&private_data->stream);
+}
+
+int ErrorArrayStreamGetNext(struct ArrowArrayStream* stream, struct ArrowArray* array) {
+ if (stream->release != ErrorArrayStreamRelease || !stream->private_data) return EINVAL;
+ auto* private_data = reinterpret_cast<struct ErrorArrayStream*>(stream->private_data);
+ return private_data->stream.get_next(&private_data->stream, array);
+}
+
+int ErrorArrayStreamGetSchema(struct ArrowArrayStream* stream,
+ struct ArrowSchema* schema) {
+ if (stream->release != ErrorArrayStreamRelease || !stream->private_data) return EINVAL;
+ auto* private_data = reinterpret_cast<struct ErrorArrayStream*>(stream->private_data);
+ return private_data->stream.get_schema(&private_data->stream, schema);
+}
+
// Default stubs
-int ErrorGetDetailCount(struct AdbcError* error) { return 0; }
+int ErrorGetDetailCount(const struct AdbcError* error) { return 0; }
-struct AdbcErrorDetail ErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail ErrorGetDetail(const struct AdbcError* error, int index) {
return {nullptr, nullptr, 0};
}
+const struct AdbcError* ErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ return nullptr;
+}
+
+void ErrorArrayStreamInit(struct ArrowArrayStream* out,
+ struct AdbcDriver* private_driver) {
+ if (!out || !out->release ||
+ // Don't bother wrapping if driver didn't claim support
+ private_driver->ErrorFromArrayStream == ErrorFromArrayStream) {
+ return;
+ }
+ struct ErrorArrayStream* private_data = new ErrorArrayStream;
+ private_data->stream = *out;
+ private_data->private_driver = private_driver;
+ out->get_last_error = ErrorArrayStreamGetLastError;
+ out->get_next = ErrorArrayStreamGetNext;
+ out->get_schema = ErrorArrayStreamGetSchema;
+ out->release = ErrorArrayStreamRelease;
+ out->private_data = private_data;
+}
+
AdbcStatusCode DatabaseGetOption(struct AdbcDatabase* database, const char* key,
char* value, size_t* length, struct AdbcError* error) {
return ADBC_STATUS_NOT_FOUND;
@@ -383,7 +440,7 @@ struct TempConnection {
// Direct implementations of API methods
-int AdbcErrorGetDetailCount(struct AdbcError* error) {
+int AdbcErrorGetDetailCount(const struct AdbcError* error) {
if (error->vendor_code == ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA && error->private_data &&
error->private_driver) {
return error->private_driver->ErrorGetDetailCount(error);
@@ -391,7 +448,7 @@ int AdbcErrorGetDetailCount(struct AdbcError* error) {
return 0;
}
-struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail AdbcErrorGetDetail(const struct AdbcError* error, int index) {
if (error->vendor_code == ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA && error->private_data &&
error->private_driver) {
return error->private_driver->ErrorGetDetail(error, index);
@@ -399,11 +456,30 @@ struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index) {
return {nullptr, nullptr, 0};
}
+const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ if (!stream->private_data || stream->release != ErrorArrayStreamRelease) {
+ return nullptr;
+ }
+ auto* private_data = reinterpret_cast<struct ErrorArrayStream*>(stream->private_data);
+ return private_data->private_driver->ErrorFromArrayStream(&private_data->stream,
+ status);
+}
+
#define INIT_ERROR(ERROR, SOURCE) \
if ((ERROR)->vendor_code == ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA) { \
(ERROR)->private_driver = (SOURCE)->private_driver; \
}
+#define WRAP_STREAM(EXPR, OUT, SOURCE) \
+ if (!(OUT)) { \
+ /* Happens for ExecuteQuery where out is optional */ \
+ return EXPR; \
+ } \
+ AdbcStatusCode status_code = EXPR; \
+ ErrorArrayStreamInit(OUT, (SOURCE)->private_driver); \
+ return status_code;
+
AdbcStatusCode AdbcDatabaseNew(struct AdbcDatabase* database, struct AdbcError* error) {
// Allocate a temporary structure to store options pre-Init
database->private_data = new TempDatabase();
@@ -700,8 +776,9 @@ AdbcStatusCode AdbcConnectionGetInfo(struct AdbcConnection* connection,
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionGetInfo(connection, info_codes,
- info_codes_length, out, error);
+ WRAP_STREAM(connection->private_driver->ConnectionGetInfo(
+ connection, info_codes, info_codes_length, out, error),
+ out, connection);
}
AdbcStatusCode AdbcConnectionGetObjects(struct AdbcConnection* connection, int depth,
@@ -714,9 +791,10 @@ AdbcStatusCode AdbcConnectionGetObjects(struct AdbcConnection* connection, int d
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionGetObjects(
- connection, depth, catalog, db_schema, table_name, table_types, column_name, stream,
- error);
+ WRAP_STREAM(connection->private_driver->ConnectionGetObjects(
+ connection, depth, catalog, db_schema, table_name, table_types,
+ column_name, stream, error),
+ stream, connection);
}
AdbcStatusCode AdbcConnectionGetOption(struct AdbcConnection* connection, const char* key,
@@ -822,8 +900,10 @@ AdbcStatusCode AdbcConnectionGetStatistics(struct AdbcConnection* connection,
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionGetStatistics(
- connection, catalog, db_schema, table_name, approximate == 1, out, error);
+ WRAP_STREAM(
+ connection->private_driver->ConnectionGetStatistics(
+ connection, catalog, db_schema, table_name, approximate == 1, out, error),
+ out, connection);
}
AdbcStatusCode AdbcConnectionGetStatisticNames(struct AdbcConnection* connection,
@@ -833,7 +913,9 @@ AdbcStatusCode AdbcConnectionGetStatisticNames(struct AdbcConnection* connection
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionGetStatisticNames(connection, out, error);
+ WRAP_STREAM(
+ connection->private_driver->ConnectionGetStatisticNames(connection, out, error),
+ out, connection);
}
AdbcStatusCode AdbcConnectionGetTableSchema(struct AdbcConnection* connection,
@@ -856,7 +938,9 @@ AdbcStatusCode AdbcConnectionGetTableTypes(struct AdbcConnection* connection,
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionGetTableTypes(connection, stream, error);
+ WRAP_STREAM(
+ connection->private_driver->ConnectionGetTableTypes(connection, stream, error),
+ stream, connection);
}
AdbcStatusCode AdbcConnectionInit(struct AdbcConnection* connection,
@@ -928,8 +1012,9 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, connection);
- return connection->private_driver->ConnectionReadPartition(
- connection, serialized_partition, serialized_length, out, error);
+ WRAP_STREAM(connection->private_driver->ConnectionReadPartition(
+ connection, serialized_partition, serialized_length, out, error),
+ out, connection);
}
AdbcStatusCode AdbcConnectionRelease(struct AdbcConnection* connection,
@@ -1079,8 +1164,9 @@ AdbcStatusCode AdbcStatementExecuteQuery(struct AdbcStatement* statement,
return ADBC_STATUS_INVALID_STATE;
}
INIT_ERROR(error, statement);
- return statement->private_driver->StatementExecuteQuery(statement, out, rows_affected,
- error);
+ WRAP_STREAM(statement->private_driver->StatementExecuteQuery(statement, out,
+ rows_affected, error),
+ out, statement);
}
AdbcStatusCode AdbcStatementExecuteSchema(struct AdbcStatement* statement,
@@ -1484,6 +1570,7 @@ AdbcStatusCode AdbcLoadDriverFromInitFunc(AdbcDriverInitFunc init_func, int vers
auto* driver = reinterpret_cast<struct AdbcDriver*>(raw_driver);
FILL_DEFAULT(driver, ErrorGetDetailCount);
FILL_DEFAULT(driver, ErrorGetDetail);
+ FILL_DEFAULT(driver, ErrorFromArrayStream);
FILL_DEFAULT(driver, DatabaseGetOption);
FILL_DEFAULT(driver, DatabaseGetOptionBytes);
diff --git a/go/adbc/go.mod b/go/adbc/go.mod
index f3d5ce06..d46fe8b9 100644
--- a/go/adbc/go.mod
+++ b/go/adbc/go.mod
@@ -17,7 +17,7 @@
module github.com/apache/arrow-adbc/go/adbc
-go 1.18
+go 1.20
require (
github.com/apache/arrow/go/v13 v13.0.0-20230713180941-b97597765355
diff --git a/go/adbc/pkg/_tmpl/driver.go.tmpl b/go/adbc/pkg/_tmpl/driver.go.tmpl
index 75e564da..879f01d5 100644
--- a/go/adbc/pkg/_tmpl/driver.go.tmpl
+++ b/go/adbc/pkg/_tmpl/driver.go.tmpl
@@ -26,11 +26,21 @@ package main
// #cgo CXXFLAGS: -std=c++11 -DADBC_EXPORTING
// #include "../../drivermgr/adbc.h"
// #include "utils.h"
+// #include <errno.h>
// #include <stdint.h>
// #include <string.h>
//
// typedef const char cchar_t;
// typedef const uint8_t cuint8_t;
+// typedef const struct AdbcError ConstAdbcError;
+//
+// int {{.Prefix}}ArrayStreamGetSchema(struct ArrowArrayStream*, struct ArrowSchema*);
+// int {{.Prefix}}ArrayStreamGetNext(struct ArrowArrayStream*, struct ArrowArray*);
+// const char* {{.Prefix}}ArrayStreamGetLastError(struct ArrowArrayStream*);
+// void {{.Prefix}}ArrayStreamRelease(struct ArrowArrayStream*);
+//
+// int {{.Prefix}}ArrayStreamGetSchemaTrampoline(struct ArrowArrayStream*, struct ArrowSchema*);
+// int {{.Prefix}}ArrayStreamGetNextTrampoline(struct ArrowArrayStream*, struct ArrowArray*);
//
// void releasePartitions(struct AdbcPartitions* partitions);
//
@@ -191,18 +201,18 @@ func exportBytesOption(val []byte, out *C.uint8_t, length *C.size_t) C.AdbcStatu
type cancellableContext struct {
ctx context.Context
- cancel context.CancelFunc
+ cancel context.CancelCauseFunc
}
func (c *cancellableContext) newContext() context.Context {
c.cancelContext()
- c.ctx, c.cancel = context.WithCancel(context.Background())
+ c.ctx, c.cancel = context.WithCancelCause(context.Background())
return c.ctx
}
func (c *cancellableContext) cancelContext() {
if c.cancel != nil {
- c.cancel()
+ c.cancel(adbc.Error{Msg: "Cancelled by request", Code: adbc.StatusCancelled})
}
c.ctx = nil
c.cancel = nil
@@ -237,6 +247,144 @@ func checkDBInit(db *C.struct_AdbcDatabase, err *C.struct_AdbcError, fname strin
return cdb
}
+// Custom ArrowArrayStream export to support ADBC error data in ArrowArrayStream
+
+type cArrayStream struct {
+ rdr array.RecordReader
+ // Must be C-allocated
+ adbcErr *C.struct_AdbcError
+ status C.AdbcStatusCode
+}
+
+func (cStream *cArrayStream) maybeError() C.int {
+ err := cStream.rdr.Err()
+ if err != nil {
+ if cStream.adbcErr != nil {
+ C.{{.Prefix}}errRelease(cStream.adbcErr)
+ } else {
+ cStream.adbcErr = (*C.struct_AdbcError)(C.malloc(C.ADBC_ERROR_1_1_0_SIZE))
+ }
+ cStream.adbcErr.vendor_code = C.ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA
+ cStream.status = C.AdbcStatusCode(errToAdbcErr(cStream.adbcErr, err))
+ switch adbc.Status(cStream.status) {
+ case adbc.StatusUnknown:
+ return C.EIO
+ case adbc.StatusNotImplemented:
+ return C.ENOTSUP
+ case adbc.StatusNotFound:
+ return C.ENOENT
+ case adbc.StatusAlreadyExists:
+ return C.EEXIST
+ case adbc.StatusInvalidArgument:
+ return C.EINVAL
+ case adbc.StatusInvalidState:
+ return C.EINVAL
+ case adbc.StatusInvalidData:
+ return C.EIO
+ case adbc.StatusIntegrity:
+ return C.EIO
+ case adbc.StatusInternal:
+ return C.EIO
+ case adbc.StatusIO:
+ return C.EIO
+ case adbc.StatusCancelled:
+ return C.ECANCELED
+ case adbc.StatusTimeout:
+ return C.ETIMEDOUT
+ case adbc.StatusUnauthenticated:
+ return C.EACCES
+ case adbc.StatusUnauthorized:
+ return C.EACCES
+ default:
+ return C.EIO
+ }
+ }
+ return 0
+}
+
+//export {{.Prefix}}ArrayStreamGetLastError
+func {{.Prefix}}ArrayStreamGetLastError(stream *C.struct_ArrowArrayStream) *C.cchar_t {
+ if stream == nil || stream.release != (*[0]byte)(C.{{.Prefix}}ArrayStreamRelease) {
+ return nil
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if cStream.adbcErr != nil {
+ return cStream.adbcErr.message
+ }
+ return nil
+}
+
+//export {{.Prefix}}ArrayStreamGetNext
+func {{.Prefix}}ArrayStreamGetNext(stream *C.struct_ArrowArrayStream, array *C.struct_ArrowArray) C.int {
+ if stream == nil || stream.release != (*[0]byte)(C.{{.Prefix}}ArrayStreamRelease) {
+ return C.EINVAL
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if cStream.rdr.Next() {
+ cdata.ExportArrowRecordBatch(cStream.rdr.Record(), toCdataArray(array), nil)
+ return 0;
+ }
+ array.release = nil
+ array.private_data = nil
+ return cStream.maybeError()
+}
+
+//export {{.Prefix}}ArrayStreamGetSchema
+func {{.Prefix}}ArrayStreamGetSchema(stream *C.struct_ArrowArrayStream, schema *C.struct_ArrowSchema) C.int {
+ if stream == nil || stream.release != (*[0]byte)(C.{{.Prefix}}ArrayStreamRelease) {
+ return C.EINVAL
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ s := cStream.rdr.Schema()
+ if s == nil {
+ return cStream.maybeError()
+ }
+ cdata.ExportArrowSchema(s, toCdataSchema(schema))
+ return 0
+}
+
+//export {{.Prefix}}ArrayStreamRelease
+func {{.Prefix}}ArrayStreamRelease(stream *C.struct_ArrowArrayStream) {
+ if stream == nil || stream.release != (*[0]byte)(C.{{.Prefix}}ArrayStreamRelease) {
+ return
+ }
+ h := (*(*cgo.Handle)(stream.private_data))
+
+ cStream := h.Value().(*cArrayStream)
+ cStream.rdr.Release()
+ if cStream.adbcErr != nil {
+ C.{{.Prefix}}errRelease(cStream.adbcErr)
+ C.free(unsafe.Pointer(cStream.adbcErr))
+ }
+ C.free(unsafe.Pointer(stream.private_data))
+ stream.private_data = nil
+ h.Delete()
+ runtime.GC()
+}
+
+//export {{.Prefix}}ErrorFromArrayStream
+func {{.Prefix}}ErrorFromArrayStream(stream *C.struct_ArrowArrayStream, status *C.AdbcStatusCode) (*C.struct_AdbcError) {
+ if stream == nil || stream.release != (*[0]byte)(C.{{.Prefix}}ArrayStreamRelease) {
+ return nil
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if status != nil {
+ *status = cStream.status
+ }
+ return cStream.adbcErr
+}
+
+func exportRecordReader(rdr array.RecordReader, stream *C.struct_ArrowArrayStream) {
+ cStream := &cArrayStream{rdr: rdr, status: C.ADBC_STATUS_OK}
+ stream.get_last_error = (*[0]byte)(C.{{.Prefix}}ArrayStreamGetLastError)
+ stream.get_next = (*[0]byte)(C.{{.Prefix}}ArrayStreamGetNextTrampoline)
+ stream.get_schema = (*[0]byte)(C.{{.Prefix}}ArrayStreamGetSchemaTrampoline)
+ stream.release = (*[0]byte)(C.{{.Prefix}}ArrayStreamRelease)
+ hndl := cgo.NewHandle(cStream)
+ stream.private_data = createHandle(hndl)
+ rdr.Retain()
+}
+
type cDatabase struct {
opts map[string]string
db adbc.Database
@@ -907,7 +1055,7 @@ func {{.Prefix}}ConnectionGetInfo(cnxn *C.struct_AdbcConnection, codes *C.uint32
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -929,7 +1077,7 @@ func {{.Prefix}}ConnectionGetObjects(cnxn *C.struct_AdbcConnection, depth C.int,
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -957,7 +1105,7 @@ func {{.Prefix}}ConnectionGetStatistics(cnxn *C.struct_AdbcConnection, catalog,
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -984,7 +1132,7 @@ func {{.Prefix}}ConnectionGetStatisticNames(cnxn *C.struct_AdbcConnection, out *
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1025,7 +1173,7 @@ func {{.Prefix}}ConnectionGetTableTypes(cnxn *C.struct_AdbcConnection, out *C.st
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1046,7 +1194,7 @@ func {{.Prefix}}ConnectionReadPartition(cnxn *C.struct_AdbcConnection, serialize
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1282,10 +1430,10 @@ func {{.Prefix}}StatementRelease(stmt *C.struct_AdbcStatement, err *C.struct_Adb
func {{.Prefix}}StatementCancel(stmt *C.struct_AdbcStatement, err *C.struct_AdbcError) (code C.AdbcStatusCode) {
defer func() {
if e := recover(); e != nil {
- code = poison(err, "AdbcStatementPrepare", e)
+ code = poison(err, "AdbcStatementCancel", e)
}
}()
- st := checkStmtInit(stmt, err, "AdbcStatementPrepare")
+ st := checkStmtInit(stmt, err, "AdbcStatementCancel")
if st == nil {
return C.ADBC_STATUS_INVALID_STATE
}
@@ -1341,8 +1489,7 @@ func {{.Prefix}}StatementExecuteQuery(stmt *C.struct_AdbcStatement, out *C.struc
}
defer rdr.Release()
- // TODO: proxy the reader and record error details (and do this elsewhere, too)
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
}
return C.ADBC_STATUS_OK
}
@@ -1662,6 +1809,7 @@ func {{.Prefix}}DriverInit(version C.int, rawDriver *C.void, err *C.struct_AdbcE
if version == C.ADBC_VERSION_1_1_0 {
driver.ErrorGetDetailCount = (*[0]byte)(C.{{.Prefix}}ErrorGetDetailCount)
driver.ErrorGetDetail = (*[0]byte)(C.{{.Prefix}}ErrorGetDetail)
+ driver.ErrorFromArrayStream = (*[0]byte)(C.{{.Prefix}}ErrorFromArrayStream)
driver.DatabaseGetOption = (*[0]byte)(C.{{.Prefix}}DatabaseGetOption)
driver.DatabaseGetOptionBytes = (*[0]byte)(C.{{.Prefix}}DatabaseGetOptionBytes)
diff --git a/go/adbc/pkg/_tmpl/utils.c.tmpl b/go/adbc/pkg/_tmpl/utils.c.tmpl
index 382c48fa..055bba58 100644
--- a/go/adbc/pkg/_tmpl/utils.c.tmpl
+++ b/go/adbc/pkg/_tmpl/utils.c.tmpl
@@ -45,6 +45,8 @@ void {{.Prefix}}ReleaseErrWithDetails(struct AdbcError* error) {
free(details->keys[i]);
free(details->values[i]);
}
+ free(details->keys);
+ free(details->values);
free(details->lengths);
free(details);
@@ -54,7 +56,7 @@ void {{.Prefix}}ReleaseErrWithDetails(struct AdbcError* error) {
error->private_data = NULL;
}
-int {{.Prefix}}ErrorGetDetailCount(struct AdbcError* error) {
+int {{.Prefix}}ErrorGetDetailCount(const struct AdbcError* error) {
if (!error || error->release != {{.Prefix}}ReleaseErrWithDetails ||
!error->private_data) {
return 0;
@@ -63,7 +65,8 @@ int {{.Prefix}}ErrorGetDetailCount(struct AdbcError* error) {
return ((struct {{.Prefix}}Error*) error->private_data)->count;
}
-struct AdbcErrorDetail {{.Prefix}}ErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail {{.Prefix}}ErrorGetDetail(const struct AdbcError* error,
+ int index) {
if (!error || error->release != {{.Prefix}}ReleaseErrWithDetails ||
!error->private_data) {
return (struct AdbcErrorDetail){NULL, NULL, 0};
@@ -80,14 +83,19 @@ struct AdbcErrorDetail {{.Prefix}}ErrorGetDetail(struct AdbcError* error, int in
};
}
-int AdbcErrorGetDetailCount(struct AdbcError* error) {
+int AdbcErrorGetDetailCount(const struct AdbcError* error) {
return {{.Prefix}}ErrorGetDetailCount(error);
}
-struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail AdbcErrorGetDetail(const struct AdbcError* error, int index) {
return {{.Prefix}}ErrorGetDetail(error, index);
}
+const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ return {{.Prefix}}ErrorFromArrayStream(stream, status);
+}
+
AdbcStatusCode AdbcDatabaseGetOption(struct AdbcDatabase* database, const char* key,
char* value, size_t* length,
struct AdbcError* error) {
@@ -413,6 +421,23 @@ AdbcStatusCode AdbcDriverInit(int version, void* driver, struct AdbcError* error
return {{.Prefix}}DriverInit(version, driver, error);
}
+int {{.Prefix}}ArrayStreamGetSchema(struct ArrowArrayStream*, struct ArrowSchema*);
+int {{.Prefix}}ArrayStreamGetNext(struct ArrowArrayStream*, struct ArrowArray*);
+
+int {{.Prefix}}ArrayStreamGetSchemaTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out) {
+ // XXX(https://github.com/apache/arrow-adbc/issues/729)
+ memset(out, 0, sizeof(*out));
+ return {{.Prefix}}ArrayStreamGetSchema(stream, out);
+}
+
+int {{.Prefix}}ArrayStreamGetNextTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowArray* out) {
+ // XXX(https://github.com/apache/arrow-adbc/issues/729)
+ memset(out, 0, sizeof(*out));
+ return {{.Prefix}}ArrayStreamGetNext(stream, out);
+}
+
#ifdef __cplusplus
}
#endif
diff --git a/go/adbc/pkg/_tmpl/utils.h.tmpl b/go/adbc/pkg/_tmpl/utils.h.tmpl
index 1821c0fd..025635b2 100644
--- a/go/adbc/pkg/_tmpl/utils.h.tmpl
+++ b/go/adbc/pkg/_tmpl/utils.h.tmpl
@@ -24,6 +24,7 @@
#include "../../drivermgr/adbc.h"
#include <stdlib.h>
+struct AdbcError* {{.Prefix}}ErrorFromArrayStream(struct ArrowArrayStream*, AdbcStatusCode*);
AdbcStatusCode {{.Prefix}}DatabaseGetOption(struct AdbcDatabase*, const char*, char*, size_t*, struct AdbcError*);
AdbcStatusCode {{.Prefix}}DatabaseGetOptionBytes(struct AdbcDatabase*, const char*, uint8_t*, size_t*, struct AdbcError*);
AdbcStatusCode {{.Prefix}}DatabaseGetOptionDouble(struct AdbcDatabase*, const char*, double*, struct AdbcError*);
@@ -97,5 +98,8 @@ struct {{.Prefix}}Error {
void {{.Prefix}}ReleaseErrWithDetails(struct AdbcError* error);
-int {{.Prefix}}ErrorGetDetailCount(struct AdbcError* error);
-struct AdbcErrorDetail {{.Prefix}}ErrorGetDetail(struct AdbcError* error, int index);
+int {{.Prefix}}ErrorGetDetailCount(const struct AdbcError* error);
+struct AdbcErrorDetail {{.Prefix}}ErrorGetDetail(const struct AdbcError* error, int index);
+
+int {{.Prefix}}ArrayStreamGetSchemaTrampoline(struct ArrowArrayStream* stream, struct ArrowSchema* out);
+int {{.Prefix}}ArrayStreamGetNextTrampoline(struct ArrowArrayStream* stream, struct ArrowArray* out);
diff --git a/go/adbc/pkg/flightsql/driver.go b/go/adbc/pkg/flightsql/driver.go
index ed7950b0..ba2bb6ec 100644
--- a/go/adbc/pkg/flightsql/driver.go
+++ b/go/adbc/pkg/flightsql/driver.go
@@ -28,11 +28,21 @@ package main
// #cgo CXXFLAGS: -std=c++11 -DADBC_EXPORTING
// #include "../../drivermgr/adbc.h"
// #include "utils.h"
+// #include <errno.h>
// #include <stdint.h>
// #include <string.h>
//
// typedef const char cchar_t;
// typedef const uint8_t cuint8_t;
+// typedef const struct AdbcError ConstAdbcError;
+//
+// int FlightSQLArrayStreamGetSchema(struct ArrowArrayStream*, struct ArrowSchema*);
+// int FlightSQLArrayStreamGetNext(struct ArrowArrayStream*, struct ArrowArray*);
+// const char* FlightSQLArrayStreamGetLastError(struct ArrowArrayStream*);
+// void FlightSQLArrayStreamRelease(struct ArrowArrayStream*);
+//
+// int FlightSQLArrayStreamGetSchemaTrampoline(struct ArrowArrayStream*, struct ArrowSchema*);
+// int FlightSQLArrayStreamGetNextTrampoline(struct ArrowArrayStream*, struct ArrowArray*);
//
// void releasePartitions(struct AdbcPartitions* partitions);
//
@@ -195,18 +205,18 @@ func exportBytesOption(val []byte, out *C.uint8_t, length *C.size_t) C.AdbcStatu
type cancellableContext struct {
ctx context.Context
- cancel context.CancelFunc
+ cancel context.CancelCauseFunc
}
func (c *cancellableContext) newContext() context.Context {
c.cancelContext()
- c.ctx, c.cancel = context.WithCancel(context.Background())
+ c.ctx, c.cancel = context.WithCancelCause(context.Background())
return c.ctx
}
func (c *cancellableContext) cancelContext() {
if c.cancel != nil {
- c.cancel()
+ c.cancel(adbc.Error{Msg: "Cancelled by request", Code: adbc.StatusCancelled})
}
c.ctx = nil
c.cancel = nil
@@ -241,6 +251,144 @@ func checkDBInit(db *C.struct_AdbcDatabase, err *C.struct_AdbcError, fname strin
return cdb
}
+// Custom ArrowArrayStream export to support ADBC error data in ArrowArrayStream
+
+type cArrayStream struct {
+ rdr array.RecordReader
+ // Must be C-allocated
+ adbcErr *C.struct_AdbcError
+ status C.AdbcStatusCode
+}
+
+func (cStream *cArrayStream) maybeError() C.int {
+ err := cStream.rdr.Err()
+ if err != nil {
+ if cStream.adbcErr != nil {
+ C.FlightSQLerrRelease(cStream.adbcErr)
+ } else {
+ cStream.adbcErr = (*C.struct_AdbcError)(C.malloc(C.ADBC_ERROR_1_1_0_SIZE))
+ }
+ cStream.adbcErr.vendor_code = C.ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA
+ cStream.status = C.AdbcStatusCode(errToAdbcErr(cStream.adbcErr, err))
+ switch adbc.Status(cStream.status) {
+ case adbc.StatusUnknown:
+ return C.EIO
+ case adbc.StatusNotImplemented:
+ return C.ENOTSUP
+ case adbc.StatusNotFound:
+ return C.ENOENT
+ case adbc.StatusAlreadyExists:
+ return C.EEXIST
+ case adbc.StatusInvalidArgument:
+ return C.EINVAL
+ case adbc.StatusInvalidState:
+ return C.EINVAL
+ case adbc.StatusInvalidData:
+ return C.EIO
+ case adbc.StatusIntegrity:
+ return C.EIO
+ case adbc.StatusInternal:
+ return C.EIO
+ case adbc.StatusIO:
+ return C.EIO
+ case adbc.StatusCancelled:
+ return C.ECANCELED
+ case adbc.StatusTimeout:
+ return C.ETIMEDOUT
+ case adbc.StatusUnauthenticated:
+ return C.EACCES
+ case adbc.StatusUnauthorized:
+ return C.EACCES
+ default:
+ return C.EIO
+ }
+ }
+ return 0
+}
+
+//export FlightSQLArrayStreamGetLastError
+func FlightSQLArrayStreamGetLastError(stream *C.struct_ArrowArrayStream) *C.cchar_t {
+ if stream == nil || stream.release != (*[0]byte)(C.FlightSQLArrayStreamRelease) {
+ return nil
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if cStream.adbcErr != nil {
+ return cStream.adbcErr.message
+ }
+ return nil
+}
+
+//export FlightSQLArrayStreamGetNext
+func FlightSQLArrayStreamGetNext(stream *C.struct_ArrowArrayStream, array *C.struct_ArrowArray) C.int {
+ if stream == nil || stream.release != (*[0]byte)(C.FlightSQLArrayStreamRelease) {
+ return C.EINVAL
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if cStream.rdr.Next() {
+ cdata.ExportArrowRecordBatch(cStream.rdr.Record(), toCdataArray(array), nil)
+ return 0
+ }
+ array.release = nil
+ array.private_data = nil
+ return cStream.maybeError()
+}
+
+//export FlightSQLArrayStreamGetSchema
+func FlightSQLArrayStreamGetSchema(stream *C.struct_ArrowArrayStream, schema *C.struct_ArrowSchema) C.int {
+ if stream == nil || stream.release != (*[0]byte)(C.FlightSQLArrayStreamRelease) {
+ return C.EINVAL
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ s := cStream.rdr.Schema()
+ if s == nil {
+ return cStream.maybeError()
+ }
+ cdata.ExportArrowSchema(s, toCdataSchema(schema))
+ return 0
+}
+
+//export FlightSQLArrayStreamRelease
+func FlightSQLArrayStreamRelease(stream *C.struct_ArrowArrayStream) {
+ if stream == nil || stream.release != (*[0]byte)(C.FlightSQLArrayStreamRelease) {
+ return
+ }
+ h := (*(*cgo.Handle)(stream.private_data))
+
+ cStream := h.Value().(*cArrayStream)
+ cStream.rdr.Release()
+ if cStream.adbcErr != nil {
+ C.FlightSQLerrRelease(cStream.adbcErr)
+ C.free(unsafe.Pointer(cStream.adbcErr))
+ }
+ C.free(unsafe.Pointer(stream.private_data))
+ stream.private_data = nil
+ h.Delete()
+ runtime.GC()
+}
+
+//export FlightSQLErrorFromArrayStream
+func FlightSQLErrorFromArrayStream(stream *C.struct_ArrowArrayStream, status *C.AdbcStatusCode) *C.struct_AdbcError {
+ if stream == nil || stream.release != (*[0]byte)(C.FlightSQLArrayStreamRelease) {
+ return nil
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if status != nil {
+ *status = cStream.status
+ }
+ return cStream.adbcErr
+}
+
+func exportRecordReader(rdr array.RecordReader, stream *C.struct_ArrowArrayStream) {
+ cStream := &cArrayStream{rdr: rdr, status: C.ADBC_STATUS_OK}
+ stream.get_last_error = (*[0]byte)(C.FlightSQLArrayStreamGetLastError)
+ stream.get_next = (*[0]byte)(C.FlightSQLArrayStreamGetNextTrampoline)
+ stream.get_schema = (*[0]byte)(C.FlightSQLArrayStreamGetSchemaTrampoline)
+ stream.release = (*[0]byte)(C.FlightSQLArrayStreamRelease)
+ hndl := cgo.NewHandle(cStream)
+ stream.private_data = createHandle(hndl)
+ rdr.Retain()
+}
+
type cDatabase struct {
opts map[string]string
db adbc.Database
@@ -911,7 +1059,7 @@ func FlightSQLConnectionGetInfo(cnxn *C.struct_AdbcConnection, codes *C.uint32_t
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -933,7 +1081,7 @@ func FlightSQLConnectionGetObjects(cnxn *C.struct_AdbcConnection, depth C.int, c
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -961,7 +1109,7 @@ func FlightSQLConnectionGetStatistics(cnxn *C.struct_AdbcConnection, catalog, db
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -988,7 +1136,7 @@ func FlightSQLConnectionGetStatisticNames(cnxn *C.struct_AdbcConnection, out *C.
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1029,7 +1177,7 @@ func FlightSQLConnectionGetTableTypes(cnxn *C.struct_AdbcConnection, out *C.stru
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1050,7 +1198,7 @@ func FlightSQLConnectionReadPartition(cnxn *C.struct_AdbcConnection, serialized
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1286,10 +1434,10 @@ func FlightSQLStatementRelease(stmt *C.struct_AdbcStatement, err *C.struct_AdbcE
func FlightSQLStatementCancel(stmt *C.struct_AdbcStatement, err *C.struct_AdbcError) (code C.AdbcStatusCode) {
defer func() {
if e := recover(); e != nil {
- code = poison(err, "AdbcStatementPrepare", e)
+ code = poison(err, "AdbcStatementCancel", e)
}
}()
- st := checkStmtInit(stmt, err, "AdbcStatementPrepare")
+ st := checkStmtInit(stmt, err, "AdbcStatementCancel")
if st == nil {
return C.ADBC_STATUS_INVALID_STATE
}
@@ -1345,8 +1493,7 @@ func FlightSQLStatementExecuteQuery(stmt *C.struct_AdbcStatement, out *C.struct_
}
defer rdr.Release()
- // TODO: proxy the reader and record error details (and do this elsewhere, too)
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
}
return C.ADBC_STATUS_OK
}
@@ -1666,6 +1813,7 @@ func FlightSQLDriverInit(version C.int, rawDriver *C.void, err *C.struct_AdbcErr
if version == C.ADBC_VERSION_1_1_0 {
driver.ErrorGetDetailCount = (*[0]byte)(C.FlightSQLErrorGetDetailCount)
driver.ErrorGetDetail = (*[0]byte)(C.FlightSQLErrorGetDetail)
+ driver.ErrorFromArrayStream = (*[0]byte)(C.FlightSQLErrorFromArrayStream)
driver.DatabaseGetOption = (*[0]byte)(C.FlightSQLDatabaseGetOption)
driver.DatabaseGetOptionBytes = (*[0]byte)(C.FlightSQLDatabaseGetOptionBytes)
diff --git a/go/adbc/pkg/flightsql/utils.c b/go/adbc/pkg/flightsql/utils.c
index 055346c3..6c30a7dd 100644
--- a/go/adbc/pkg/flightsql/utils.c
+++ b/go/adbc/pkg/flightsql/utils.c
@@ -47,6 +47,8 @@ void FlightSQLReleaseErrWithDetails(struct AdbcError* error) {
free(details->keys[i]);
free(details->values[i]);
}
+ free(details->keys);
+ free(details->values);
free(details->lengths);
free(details);
@@ -56,7 +58,7 @@ void FlightSQLReleaseErrWithDetails(struct AdbcError* error) {
error->private_data = NULL;
}
-int FlightSQLErrorGetDetailCount(struct AdbcError* error) {
+int FlightSQLErrorGetDetailCount(const struct AdbcError* error) {
if (!error || error->release != FlightSQLReleaseErrWithDetails ||
!error->private_data) {
return 0;
@@ -65,7 +67,8 @@ int FlightSQLErrorGetDetailCount(struct AdbcError* error) {
return ((struct FlightSQLError*) error->private_data)->count;
}
-struct AdbcErrorDetail FlightSQLErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail FlightSQLErrorGetDetail(const struct AdbcError* error,
+ int index) {
if (!error || error->release != FlightSQLReleaseErrWithDetails ||
!error->private_data) {
return (struct AdbcErrorDetail){NULL, NULL, 0};
@@ -82,14 +85,19 @@ struct AdbcErrorDetail FlightSQLErrorGetDetail(struct AdbcError* error, int inde
};
}
-int AdbcErrorGetDetailCount(struct AdbcError* error) {
+int AdbcErrorGetDetailCount(const struct AdbcError* error) {
return FlightSQLErrorGetDetailCount(error);
}
-struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail AdbcErrorGetDetail(const struct AdbcError* error, int index) {
return FlightSQLErrorGetDetail(error, index);
}
+const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ return FlightSQLErrorFromArrayStream(stream, status);
+}
+
AdbcStatusCode AdbcDatabaseGetOption(struct AdbcDatabase* database, const char* key,
char* value, size_t* length,
struct AdbcError* error) {
@@ -415,6 +423,23 @@ AdbcStatusCode AdbcDriverInit(int version, void* driver, struct AdbcError* error
return FlightSQLDriverInit(version, driver, error);
}
+int FlightSQLArrayStreamGetSchema(struct ArrowArrayStream*, struct ArrowSchema*);
+int FlightSQLArrayStreamGetNext(struct ArrowArrayStream*, struct ArrowArray*);
+
+int FlightSQLArrayStreamGetSchemaTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out) {
+ // XXX(https://github.com/apache/arrow-adbc/issues/729)
+ memset(out, 0, sizeof(*out));
+ return FlightSQLArrayStreamGetSchema(stream, out);
+}
+
+int FlightSQLArrayStreamGetNextTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowArray* out) {
+ // XXX(https://github.com/apache/arrow-adbc/issues/729)
+ memset(out, 0, sizeof(*out));
+ return FlightSQLArrayStreamGetNext(stream, out);
+}
+
#ifdef __cplusplus
}
#endif
diff --git a/go/adbc/pkg/flightsql/utils.h b/go/adbc/pkg/flightsql/utils.h
index fa79d34d..2b426a25 100644
--- a/go/adbc/pkg/flightsql/utils.h
+++ b/go/adbc/pkg/flightsql/utils.h
@@ -26,6 +26,8 @@
#include <stdlib.h>
#include "../../drivermgr/adbc.h"
+struct AdbcError* FlightSQLErrorFromArrayStream(struct ArrowArrayStream*,
+ AdbcStatusCode*);
AdbcStatusCode FlightSQLDatabaseGetOption(struct AdbcDatabase*, const char*, char*,
size_t*, struct AdbcError*);
AdbcStatusCode FlightSQLDatabaseGetOptionBytes(struct AdbcDatabase*, const char*,
@@ -164,5 +166,10 @@ struct FlightSQLError {
void FlightSQLReleaseErrWithDetails(struct AdbcError* error);
-int FlightSQLErrorGetDetailCount(struct AdbcError* error);
-struct AdbcErrorDetail FlightSQLErrorGetDetail(struct AdbcError* error, int index);
+int FlightSQLErrorGetDetailCount(const struct AdbcError* error);
+struct AdbcErrorDetail FlightSQLErrorGetDetail(const struct AdbcError* error, int index);
+
+int FlightSQLArrayStreamGetSchemaTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out);
+int FlightSQLArrayStreamGetNextTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowArray* out);
diff --git a/go/adbc/pkg/panicdummy/driver.go b/go/adbc/pkg/panicdummy/driver.go
index ec9e00fb..dc3c4c30 100644
--- a/go/adbc/pkg/panicdummy/driver.go
+++ b/go/adbc/pkg/panicdummy/driver.go
@@ -28,11 +28,21 @@ package main
// #cgo CXXFLAGS: -std=c++11 -DADBC_EXPORTING
// #include "../../drivermgr/adbc.h"
// #include "utils.h"
+// #include <errno.h>
// #include <stdint.h>
// #include <string.h>
//
// typedef const char cchar_t;
// typedef const uint8_t cuint8_t;
+// typedef const struct AdbcError ConstAdbcError;
+//
+// int PanicDummyArrayStreamGetSchema(struct ArrowArrayStream*, struct ArrowSchema*);
+// int PanicDummyArrayStreamGetNext(struct ArrowArrayStream*, struct ArrowArray*);
+// const char* PanicDummyArrayStreamGetLastError(struct ArrowArrayStream*);
+// void PanicDummyArrayStreamRelease(struct ArrowArrayStream*);
+//
+// int PanicDummyArrayStreamGetSchemaTrampoline(struct ArrowArrayStream*, struct ArrowSchema*);
+// int PanicDummyArrayStreamGetNextTrampoline(struct ArrowArrayStream*, struct ArrowArray*);
//
// void releasePartitions(struct AdbcPartitions* partitions);
//
@@ -195,18 +205,18 @@ func exportBytesOption(val []byte, out *C.uint8_t, length *C.size_t) C.AdbcStatu
type cancellableContext struct {
ctx context.Context
- cancel context.CancelFunc
+ cancel context.CancelCauseFunc
}
func (c *cancellableContext) newContext() context.Context {
c.cancelContext()
- c.ctx, c.cancel = context.WithCancel(context.Background())
+ c.ctx, c.cancel = context.WithCancelCause(context.Background())
return c.ctx
}
func (c *cancellableContext) cancelContext() {
if c.cancel != nil {
- c.cancel()
+ c.cancel(adbc.Error{Msg: "Cancelled by request", Code: adbc.StatusCancelled})
}
c.ctx = nil
c.cancel = nil
@@ -241,6 +251,144 @@ func checkDBInit(db *C.struct_AdbcDatabase, err *C.struct_AdbcError, fname strin
return cdb
}
+// Custom ArrowArrayStream export to support ADBC error data in ArrowArrayStream
+
+type cArrayStream struct {
+ rdr array.RecordReader
+ // Must be C-allocated
+ adbcErr *C.struct_AdbcError
+ status C.AdbcStatusCode
+}
+
+func (cStream *cArrayStream) maybeError() C.int {
+ err := cStream.rdr.Err()
+ if err != nil {
+ if cStream.adbcErr != nil {
+ C.PanicDummyerrRelease(cStream.adbcErr)
+ } else {
+ cStream.adbcErr = (*C.struct_AdbcError)(C.malloc(C.ADBC_ERROR_1_1_0_SIZE))
+ }
+ cStream.adbcErr.vendor_code = C.ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA
+ cStream.status = C.AdbcStatusCode(errToAdbcErr(cStream.adbcErr, err))
+ switch adbc.Status(cStream.status) {
+ case adbc.StatusUnknown:
+ return C.EIO
+ case adbc.StatusNotImplemented:
+ return C.ENOTSUP
+ case adbc.StatusNotFound:
+ return C.ENOENT
+ case adbc.StatusAlreadyExists:
+ return C.EEXIST
+ case adbc.StatusInvalidArgument:
+ return C.EINVAL
+ case adbc.StatusInvalidState:
+ return C.EINVAL
+ case adbc.StatusInvalidData:
+ return C.EIO
+ case adbc.StatusIntegrity:
+ return C.EIO
+ case adbc.StatusInternal:
+ return C.EIO
+ case adbc.StatusIO:
+ return C.EIO
+ case adbc.StatusCancelled:
+ return C.ECANCELED
+ case adbc.StatusTimeout:
+ return C.ETIMEDOUT
+ case adbc.StatusUnauthenticated:
+ return C.EACCES
+ case adbc.StatusUnauthorized:
+ return C.EACCES
+ default:
+ return C.EIO
+ }
+ }
+ return 0
+}
+
+//export PanicDummyArrayStreamGetLastError
+func PanicDummyArrayStreamGetLastError(stream *C.struct_ArrowArrayStream) *C.cchar_t {
+ if stream == nil || stream.release != (*[0]byte)(C.PanicDummyArrayStreamRelease) {
+ return nil
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if cStream.adbcErr != nil {
+ return cStream.adbcErr.message
+ }
+ return nil
+}
+
+//export PanicDummyArrayStreamGetNext
+func PanicDummyArrayStreamGetNext(stream *C.struct_ArrowArrayStream, array *C.struct_ArrowArray) C.int {
+ if stream == nil || stream.release != (*[0]byte)(C.PanicDummyArrayStreamRelease) {
+ return C.EINVAL
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if cStream.rdr.Next() {
+ cdata.ExportArrowRecordBatch(cStream.rdr.Record(), toCdataArray(array), nil)
+ return 0
+ }
+ array.release = nil
+ array.private_data = nil
+ return cStream.maybeError()
+}
+
+//export PanicDummyArrayStreamGetSchema
+func PanicDummyArrayStreamGetSchema(stream *C.struct_ArrowArrayStream, schema *C.struct_ArrowSchema) C.int {
+ if stream == nil || stream.release != (*[0]byte)(C.PanicDummyArrayStreamRelease) {
+ return C.EINVAL
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ s := cStream.rdr.Schema()
+ if s == nil {
+ return cStream.maybeError()
+ }
+ cdata.ExportArrowSchema(s, toCdataSchema(schema))
+ return 0
+}
+
+//export PanicDummyArrayStreamRelease
+func PanicDummyArrayStreamRelease(stream *C.struct_ArrowArrayStream) {
+ if stream == nil || stream.release != (*[0]byte)(C.PanicDummyArrayStreamRelease) {
+ return
+ }
+ h := (*(*cgo.Handle)(stream.private_data))
+
+ cStream := h.Value().(*cArrayStream)
+ cStream.rdr.Release()
+ if cStream.adbcErr != nil {
+ C.PanicDummyerrRelease(cStream.adbcErr)
+ C.free(unsafe.Pointer(cStream.adbcErr))
+ }
+ C.free(unsafe.Pointer(stream.private_data))
+ stream.private_data = nil
+ h.Delete()
+ runtime.GC()
+}
+
+//export PanicDummyErrorFromArrayStream
+func PanicDummyErrorFromArrayStream(stream *C.struct_ArrowArrayStream, status *C.AdbcStatusCode) *C.struct_AdbcError {
+ if stream == nil || stream.release != (*[0]byte)(C.PanicDummyArrayStreamRelease) {
+ return nil
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if status != nil {
+ *status = cStream.status
+ }
+ return cStream.adbcErr
+}
+
+func exportRecordReader(rdr array.RecordReader, stream *C.struct_ArrowArrayStream) {
+ cStream := &cArrayStream{rdr: rdr, status: C.ADBC_STATUS_OK}
+ stream.get_last_error = (*[0]byte)(C.PanicDummyArrayStreamGetLastError)
+ stream.get_next = (*[0]byte)(C.PanicDummyArrayStreamGetNextTrampoline)
+ stream.get_schema = (*[0]byte)(C.PanicDummyArrayStreamGetSchemaTrampoline)
+ stream.release = (*[0]byte)(C.PanicDummyArrayStreamRelease)
+ hndl := cgo.NewHandle(cStream)
+ stream.private_data = createHandle(hndl)
+ rdr.Retain()
+}
+
type cDatabase struct {
opts map[string]string
db adbc.Database
@@ -911,7 +1059,7 @@ func PanicDummyConnectionGetInfo(cnxn *C.struct_AdbcConnection, codes *C.uint32_
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -933,7 +1081,7 @@ func PanicDummyConnectionGetObjects(cnxn *C.struct_AdbcConnection, depth C.int,
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -961,7 +1109,7 @@ func PanicDummyConnectionGetStatistics(cnxn *C.struct_AdbcConnection, catalog, d
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -988,7 +1136,7 @@ func PanicDummyConnectionGetStatisticNames(cnxn *C.struct_AdbcConnection, out *C
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1029,7 +1177,7 @@ func PanicDummyConnectionGetTableTypes(cnxn *C.struct_AdbcConnection, out *C.str
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1050,7 +1198,7 @@ func PanicDummyConnectionReadPartition(cnxn *C.struct_AdbcConnection, serialized
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1286,10 +1434,10 @@ func PanicDummyStatementRelease(stmt *C.struct_AdbcStatement, err *C.struct_Adbc
func PanicDummyStatementCancel(stmt *C.struct_AdbcStatement, err *C.struct_AdbcError) (code C.AdbcStatusCode) {
defer func() {
if e := recover(); e != nil {
- code = poison(err, "AdbcStatementPrepare", e)
+ code = poison(err, "AdbcStatementCancel", e)
}
}()
- st := checkStmtInit(stmt, err, "AdbcStatementPrepare")
+ st := checkStmtInit(stmt, err, "AdbcStatementCancel")
if st == nil {
return C.ADBC_STATUS_INVALID_STATE
}
@@ -1345,8 +1493,7 @@ func PanicDummyStatementExecuteQuery(stmt *C.struct_AdbcStatement, out *C.struct
}
defer rdr.Release()
- // TODO: proxy the reader and record error details (and do this elsewhere, too)
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
}
return C.ADBC_STATUS_OK
}
@@ -1666,6 +1813,7 @@ func PanicDummyDriverInit(version C.int, rawDriver *C.void, err *C.struct_AdbcEr
if version == C.ADBC_VERSION_1_1_0 {
driver.ErrorGetDetailCount = (*[0]byte)(C.PanicDummyErrorGetDetailCount)
driver.ErrorGetDetail = (*[0]byte)(C.PanicDummyErrorGetDetail)
+ driver.ErrorFromArrayStream = (*[0]byte)(C.PanicDummyErrorFromArrayStream)
driver.DatabaseGetOption = (*[0]byte)(C.PanicDummyDatabaseGetOption)
driver.DatabaseGetOptionBytes = (*[0]byte)(C.PanicDummyDatabaseGetOptionBytes)
diff --git a/go/adbc/pkg/panicdummy/utils.c b/go/adbc/pkg/panicdummy/utils.c
index 82f71d88..db4a7fdd 100644
--- a/go/adbc/pkg/panicdummy/utils.c
+++ b/go/adbc/pkg/panicdummy/utils.c
@@ -47,6 +47,8 @@ void PanicDummyReleaseErrWithDetails(struct AdbcError* error) {
free(details->keys[i]);
free(details->values[i]);
}
+ free(details->keys);
+ free(details->values);
free(details->lengths);
free(details);
@@ -56,7 +58,7 @@ void PanicDummyReleaseErrWithDetails(struct AdbcError* error) {
error->private_data = NULL;
}
-int PanicDummyErrorGetDetailCount(struct AdbcError* error) {
+int PanicDummyErrorGetDetailCount(const struct AdbcError* error) {
if (!error || error->release != PanicDummyReleaseErrWithDetails ||
!error->private_data) {
return 0;
@@ -65,7 +67,8 @@ int PanicDummyErrorGetDetailCount(struct AdbcError* error) {
return ((struct PanicDummyError*) error->private_data)->count;
}
-struct AdbcErrorDetail PanicDummyErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail PanicDummyErrorGetDetail(const struct AdbcError* error,
+ int index) {
if (!error || error->release != PanicDummyReleaseErrWithDetails ||
!error->private_data) {
return (struct AdbcErrorDetail){NULL, NULL, 0};
@@ -82,14 +85,19 @@ struct AdbcErrorDetail PanicDummyErrorGetDetail(struct AdbcError* error, int ind
};
}
-int AdbcErrorGetDetailCount(struct AdbcError* error) {
+int AdbcErrorGetDetailCount(const struct AdbcError* error) {
return PanicDummyErrorGetDetailCount(error);
}
-struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail AdbcErrorGetDetail(const struct AdbcError* error, int index) {
return PanicDummyErrorGetDetail(error, index);
}
+const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ return PanicDummyErrorFromArrayStream(stream, status);
+}
+
AdbcStatusCode AdbcDatabaseGetOption(struct AdbcDatabase* database, const char* key,
char* value, size_t* length,
struct AdbcError* error) {
@@ -415,6 +423,23 @@ AdbcStatusCode AdbcDriverInit(int version, void* driver, struct AdbcError* error
return PanicDummyDriverInit(version, driver, error);
}
+int PanicDummyArrayStreamGetSchema(struct ArrowArrayStream*, struct ArrowSchema*);
+int PanicDummyArrayStreamGetNext(struct ArrowArrayStream*, struct ArrowArray*);
+
+int PanicDummyArrayStreamGetSchemaTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out) {
+ // XXX(https://github.com/apache/arrow-adbc/issues/729)
+ memset(out, 0, sizeof(*out));
+ return PanicDummyArrayStreamGetSchema(stream, out);
+}
+
+int PanicDummyArrayStreamGetNextTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowArray* out) {
+ // XXX(https://github.com/apache/arrow-adbc/issues/729)
+ memset(out, 0, sizeof(*out));
+ return PanicDummyArrayStreamGetNext(stream, out);
+}
+
#ifdef __cplusplus
}
#endif
diff --git a/go/adbc/pkg/panicdummy/utils.h b/go/adbc/pkg/panicdummy/utils.h
index 47d463ca..00ec5f85 100644
--- a/go/adbc/pkg/panicdummy/utils.h
+++ b/go/adbc/pkg/panicdummy/utils.h
@@ -26,6 +26,8 @@
#include <stdlib.h>
#include "../../drivermgr/adbc.h"
+struct AdbcError* PanicDummyErrorFromArrayStream(struct ArrowArrayStream*,
+ AdbcStatusCode*);
AdbcStatusCode PanicDummyDatabaseGetOption(struct AdbcDatabase*, const char*, char*,
size_t*, struct AdbcError*);
AdbcStatusCode PanicDummyDatabaseGetOptionBytes(struct AdbcDatabase*, const char*,
@@ -168,5 +170,10 @@ struct PanicDummyError {
void PanicDummyReleaseErrWithDetails(struct AdbcError* error);
-int PanicDummyErrorGetDetailCount(struct AdbcError* error);
-struct AdbcErrorDetail PanicDummyErrorGetDetail(struct AdbcError* error, int index);
+int PanicDummyErrorGetDetailCount(const struct AdbcError* error);
+struct AdbcErrorDetail PanicDummyErrorGetDetail(const struct AdbcError* error, int index);
+
+int PanicDummyArrayStreamGetSchemaTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out);
+int PanicDummyArrayStreamGetNextTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowArray* out);
diff --git a/go/adbc/pkg/snowflake/driver.go b/go/adbc/pkg/snowflake/driver.go
index 123c9bc4..11ba205c 100644
--- a/go/adbc/pkg/snowflake/driver.go
+++ b/go/adbc/pkg/snowflake/driver.go
@@ -28,11 +28,21 @@ package main
// #cgo CXXFLAGS: -std=c++11 -DADBC_EXPORTING
// #include "../../drivermgr/adbc.h"
// #include "utils.h"
+// #include <errno.h>
// #include <stdint.h>
// #include <string.h>
//
// typedef const char cchar_t;
// typedef const uint8_t cuint8_t;
+// typedef const struct AdbcError ConstAdbcError;
+//
+// int SnowflakeArrayStreamGetSchema(struct ArrowArrayStream*, struct ArrowSchema*);
+// int SnowflakeArrayStreamGetNext(struct ArrowArrayStream*, struct ArrowArray*);
+// const char* SnowflakeArrayStreamGetLastError(struct ArrowArrayStream*);
+// void SnowflakeArrayStreamRelease(struct ArrowArrayStream*);
+//
+// int SnowflakeArrayStreamGetSchemaTrampoline(struct ArrowArrayStream*, struct ArrowSchema*);
+// int SnowflakeArrayStreamGetNextTrampoline(struct ArrowArrayStream*, struct ArrowArray*);
//
// void releasePartitions(struct AdbcPartitions* partitions);
//
@@ -195,18 +205,18 @@ func exportBytesOption(val []byte, out *C.uint8_t, length *C.size_t) C.AdbcStatu
type cancellableContext struct {
ctx context.Context
- cancel context.CancelFunc
+ cancel context.CancelCauseFunc
}
func (c *cancellableContext) newContext() context.Context {
c.cancelContext()
- c.ctx, c.cancel = context.WithCancel(context.Background())
+ c.ctx, c.cancel = context.WithCancelCause(context.Background())
return c.ctx
}
func (c *cancellableContext) cancelContext() {
if c.cancel != nil {
- c.cancel()
+ c.cancel(adbc.Error{Msg: "Cancelled by request", Code: adbc.StatusCancelled})
}
c.ctx = nil
c.cancel = nil
@@ -241,6 +251,144 @@ func checkDBInit(db *C.struct_AdbcDatabase, err *C.struct_AdbcError, fname strin
return cdb
}
+// Custom ArrowArrayStream export to support ADBC error data in ArrowArrayStream
+
+type cArrayStream struct {
+ rdr array.RecordReader
+ // Must be C-allocated
+ adbcErr *C.struct_AdbcError
+ status C.AdbcStatusCode
+}
+
+func (cStream *cArrayStream) maybeError() C.int {
+ err := cStream.rdr.Err()
+ if err != nil {
+ if cStream.adbcErr != nil {
+ C.SnowflakeerrRelease(cStream.adbcErr)
+ } else {
+ cStream.adbcErr = (*C.struct_AdbcError)(C.malloc(C.ADBC_ERROR_1_1_0_SIZE))
+ }
+ cStream.adbcErr.vendor_code = C.ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA
+ cStream.status = C.AdbcStatusCode(errToAdbcErr(cStream.adbcErr, err))
+ switch adbc.Status(cStream.status) {
+ case adbc.StatusUnknown:
+ return C.EIO
+ case adbc.StatusNotImplemented:
+ return C.ENOTSUP
+ case adbc.StatusNotFound:
+ return C.ENOENT
+ case adbc.StatusAlreadyExists:
+ return C.EEXIST
+ case adbc.StatusInvalidArgument:
+ return C.EINVAL
+ case adbc.StatusInvalidState:
+ return C.EINVAL
+ case adbc.StatusInvalidData:
+ return C.EIO
+ case adbc.StatusIntegrity:
+ return C.EIO
+ case adbc.StatusInternal:
+ return C.EIO
+ case adbc.StatusIO:
+ return C.EIO
+ case adbc.StatusCancelled:
+ return C.ECANCELED
+ case adbc.StatusTimeout:
+ return C.ETIMEDOUT
+ case adbc.StatusUnauthenticated:
+ return C.EACCES
+ case adbc.StatusUnauthorized:
+ return C.EACCES
+ default:
+ return C.EIO
+ }
+ }
+ return 0
+}
+
+//export SnowflakeArrayStreamGetLastError
+func SnowflakeArrayStreamGetLastError(stream *C.struct_ArrowArrayStream) *C.cchar_t {
+ if stream == nil || stream.release != (*[0]byte)(C.SnowflakeArrayStreamRelease) {
+ return nil
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if cStream.adbcErr != nil {
+ return cStream.adbcErr.message
+ }
+ return nil
+}
+
+//export SnowflakeArrayStreamGetNext
+func SnowflakeArrayStreamGetNext(stream *C.struct_ArrowArrayStream, array *C.struct_ArrowArray) C.int {
+ if stream == nil || stream.release != (*[0]byte)(C.SnowflakeArrayStreamRelease) {
+ return C.EINVAL
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if cStream.rdr.Next() {
+ cdata.ExportArrowRecordBatch(cStream.rdr.Record(), toCdataArray(array), nil)
+ return 0
+ }
+ array.release = nil
+ array.private_data = nil
+ return cStream.maybeError()
+}
+
+//export SnowflakeArrayStreamGetSchema
+func SnowflakeArrayStreamGetSchema(stream *C.struct_ArrowArrayStream, schema *C.struct_ArrowSchema) C.int {
+ if stream == nil || stream.release != (*[0]byte)(C.SnowflakeArrayStreamRelease) {
+ return C.EINVAL
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ s := cStream.rdr.Schema()
+ if s == nil {
+ return cStream.maybeError()
+ }
+ cdata.ExportArrowSchema(s, toCdataSchema(schema))
+ return 0
+}
+
+//export SnowflakeArrayStreamRelease
+func SnowflakeArrayStreamRelease(stream *C.struct_ArrowArrayStream) {
+ if stream == nil || stream.release != (*[0]byte)(C.SnowflakeArrayStreamRelease) {
+ return
+ }
+ h := (*(*cgo.Handle)(stream.private_data))
+
+ cStream := h.Value().(*cArrayStream)
+ cStream.rdr.Release()
+ if cStream.adbcErr != nil {
+ C.SnowflakeerrRelease(cStream.adbcErr)
+ C.free(unsafe.Pointer(cStream.adbcErr))
+ }
+ C.free(unsafe.Pointer(stream.private_data))
+ stream.private_data = nil
+ h.Delete()
+ runtime.GC()
+}
+
+//export SnowflakeErrorFromArrayStream
+func SnowflakeErrorFromArrayStream(stream *C.struct_ArrowArrayStream, status *C.AdbcStatusCode) *C.struct_AdbcError {
+ if stream == nil || stream.release != (*[0]byte)(C.SnowflakeArrayStreamRelease) {
+ return nil
+ }
+ cStream := getFromHandle[cArrayStream](stream.private_data)
+ if status != nil {
+ *status = cStream.status
+ }
+ return cStream.adbcErr
+}
+
+func exportRecordReader(rdr array.RecordReader, stream *C.struct_ArrowArrayStream) {
+ cStream := &cArrayStream{rdr: rdr, status: C.ADBC_STATUS_OK}
+ stream.get_last_error = (*[0]byte)(C.SnowflakeArrayStreamGetLastError)
+ stream.get_next = (*[0]byte)(C.SnowflakeArrayStreamGetNextTrampoline)
+ stream.get_schema = (*[0]byte)(C.SnowflakeArrayStreamGetSchemaTrampoline)
+ stream.release = (*[0]byte)(C.SnowflakeArrayStreamRelease)
+ hndl := cgo.NewHandle(cStream)
+ stream.private_data = createHandle(hndl)
+ rdr.Retain()
+}
+
type cDatabase struct {
opts map[string]string
db adbc.Database
@@ -911,7 +1059,7 @@ func SnowflakeConnectionGetInfo(cnxn *C.struct_AdbcConnection, codes *C.uint32_t
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -933,7 +1081,7 @@ func SnowflakeConnectionGetObjects(cnxn *C.struct_AdbcConnection, depth C.int, c
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -961,7 +1109,7 @@ func SnowflakeConnectionGetStatistics(cnxn *C.struct_AdbcConnection, catalog, db
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -988,7 +1136,7 @@ func SnowflakeConnectionGetStatisticNames(cnxn *C.struct_AdbcConnection, out *C.
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1029,7 +1177,7 @@ func SnowflakeConnectionGetTableTypes(cnxn *C.struct_AdbcConnection, out *C.stru
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1050,7 +1198,7 @@ func SnowflakeConnectionReadPartition(cnxn *C.struct_AdbcConnection, serialized
return C.AdbcStatusCode(errToAdbcErr(err, e))
}
defer rdr.Release()
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
return C.ADBC_STATUS_OK
}
@@ -1286,10 +1434,10 @@ func SnowflakeStatementRelease(stmt *C.struct_AdbcStatement, err *C.struct_AdbcE
func SnowflakeStatementCancel(stmt *C.struct_AdbcStatement, err *C.struct_AdbcError) (code C.AdbcStatusCode) {
defer func() {
if e := recover(); e != nil {
- code = poison(err, "AdbcStatementPrepare", e)
+ code = poison(err, "AdbcStatementCancel", e)
}
}()
- st := checkStmtInit(stmt, err, "AdbcStatementPrepare")
+ st := checkStmtInit(stmt, err, "AdbcStatementCancel")
if st == nil {
return C.ADBC_STATUS_INVALID_STATE
}
@@ -1345,8 +1493,7 @@ func SnowflakeStatementExecuteQuery(stmt *C.struct_AdbcStatement, out *C.struct_
}
defer rdr.Release()
- // TODO: proxy the reader and record error details (and do this elsewhere, too)
- cdata.ExportRecordReader(rdr, toCdataStream(out))
+ exportRecordReader(rdr, out)
}
return C.ADBC_STATUS_OK
}
@@ -1666,6 +1813,7 @@ func SnowflakeDriverInit(version C.int, rawDriver *C.void, err *C.struct_AdbcErr
if version == C.ADBC_VERSION_1_1_0 {
driver.ErrorGetDetailCount = (*[0]byte)(C.SnowflakeErrorGetDetailCount)
driver.ErrorGetDetail = (*[0]byte)(C.SnowflakeErrorGetDetail)
+ driver.ErrorFromArrayStream = (*[0]byte)(C.SnowflakeErrorFromArrayStream)
driver.DatabaseGetOption = (*[0]byte)(C.SnowflakeDatabaseGetOption)
driver.DatabaseGetOptionBytes = (*[0]byte)(C.SnowflakeDatabaseGetOptionBytes)
diff --git a/go/adbc/pkg/snowflake/utils.c b/go/adbc/pkg/snowflake/utils.c
index 9c4f08b7..f2269847 100644
--- a/go/adbc/pkg/snowflake/utils.c
+++ b/go/adbc/pkg/snowflake/utils.c
@@ -47,6 +47,8 @@ void SnowflakeReleaseErrWithDetails(struct AdbcError* error) {
free(details->keys[i]);
free(details->values[i]);
}
+ free(details->keys);
+ free(details->values);
free(details->lengths);
free(details);
@@ -56,7 +58,7 @@ void SnowflakeReleaseErrWithDetails(struct AdbcError* error) {
error->private_data = NULL;
}
-int SnowflakeErrorGetDetailCount(struct AdbcError* error) {
+int SnowflakeErrorGetDetailCount(const struct AdbcError* error) {
if (!error || error->release != SnowflakeReleaseErrWithDetails ||
!error->private_data) {
return 0;
@@ -65,7 +67,8 @@ int SnowflakeErrorGetDetailCount(struct AdbcError* error) {
return ((struct SnowflakeError*) error->private_data)->count;
}
-struct AdbcErrorDetail SnowflakeErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail SnowflakeErrorGetDetail(const struct AdbcError* error,
+ int index) {
if (!error || error->release != SnowflakeReleaseErrWithDetails ||
!error->private_data) {
return (struct AdbcErrorDetail){NULL, NULL, 0};
@@ -82,14 +85,19 @@ struct AdbcErrorDetail SnowflakeErrorGetDetail(struct AdbcError* error, int inde
};
}
-int AdbcErrorGetDetailCount(struct AdbcError* error) {
+int AdbcErrorGetDetailCount(const struct AdbcError* error) {
return SnowflakeErrorGetDetailCount(error);
}
-struct AdbcErrorDetail AdbcErrorGetDetail(struct AdbcError* error, int index) {
+struct AdbcErrorDetail AdbcErrorGetDetail(const struct AdbcError* error, int index) {
return SnowflakeErrorGetDetail(error, index);
}
+const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream,
+ AdbcStatusCode* status) {
+ return SnowflakeErrorFromArrayStream(stream, status);
+}
+
AdbcStatusCode AdbcDatabaseGetOption(struct AdbcDatabase* database, const char* key,
char* value, size_t* length,
struct AdbcError* error) {
@@ -415,6 +423,23 @@ AdbcStatusCode AdbcDriverInit(int version, void* driver, struct AdbcError* error
return SnowflakeDriverInit(version, driver, error);
}
+int SnowflakeArrayStreamGetSchema(struct ArrowArrayStream*, struct ArrowSchema*);
+int SnowflakeArrayStreamGetNext(struct ArrowArrayStream*, struct ArrowArray*);
+
+int SnowflakeArrayStreamGetSchemaTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out) {
+ // XXX(https://github.com/apache/arrow-adbc/issues/729)
+ memset(out, 0, sizeof(*out));
+ return SnowflakeArrayStreamGetSchema(stream, out);
+}
+
+int SnowflakeArrayStreamGetNextTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowArray* out) {
+ // XXX(https://github.com/apache/arrow-adbc/issues/729)
+ memset(out, 0, sizeof(*out));
+ return SnowflakeArrayStreamGetNext(stream, out);
+}
+
#ifdef __cplusplus
}
#endif
diff --git a/go/adbc/pkg/snowflake/utils.h b/go/adbc/pkg/snowflake/utils.h
index d12e91a9..c1b3c07d 100644
--- a/go/adbc/pkg/snowflake/utils.h
+++ b/go/adbc/pkg/snowflake/utils.h
@@ -26,6 +26,8 @@
#include <stdlib.h>
#include "../../drivermgr/adbc.h"
+struct AdbcError* SnowflakeErrorFromArrayStream(struct ArrowArrayStream*,
+ AdbcStatusCode*);
AdbcStatusCode SnowflakeDatabaseGetOption(struct AdbcDatabase*, const char*, char*,
size_t*, struct AdbcError*);
AdbcStatusCode SnowflakeDatabaseGetOptionBytes(struct AdbcDatabase*, const char*,
@@ -164,5 +166,10 @@ struct SnowflakeError {
void SnowflakeReleaseErrWithDetails(struct AdbcError* error);
-int SnowflakeErrorGetDetailCount(struct AdbcError* error);
-struct AdbcErrorDetail SnowflakeErrorGetDetail(struct AdbcError* error, int index);
+int SnowflakeErrorGetDetailCount(const struct AdbcError* error);
+struct AdbcErrorDetail SnowflakeErrorGetDetail(const struct AdbcError* error, int index);
+
+int SnowflakeArrayStreamGetSchemaTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out);
+int SnowflakeArrayStreamGetNextTrampoline(struct ArrowArrayStream* stream,
+ struct ArrowArray* out);
diff --git a/python/adbc_driver_manager/adbc_driver_manager/_lib.pxd b/python/adbc_driver_manager/adbc_driver_manager/_lib.pxd
new file mode 100644
index 00000000..676ee9f1
--- /dev/null
+++ b/python/adbc_driver_manager/adbc_driver_manager/_lib.pxd
@@ -0,0 +1,284 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# cython: language_level = 3
+
+from libc.stdint cimport int32_t, int64_t, uint8_t, uint32_t
+
+
+cdef extern from "adbc.h" nogil:
+ # C ABI
+ cdef struct CArrowSchema"ArrowSchema":
+ pass
+ cdef struct CArrowArray"ArrowArray":
+ pass
+
+ ctypedef int (*CArrowArrayStreamGetLastError)(void*)
+ ctypedef int (*CArrowArrayStreamGetNext)(void*, CArrowArray*)
+ ctypedef char* (*CArrowArrayStreamGetSchema)(void*, CArrowSchema*)
+ ctypedef void (*CArrowArrayStreamRelease)(void*)
+
+ cdef struct CArrowArrayStream"ArrowArrayStream":
+ CArrowArrayStreamGetLastError get_last_error
+ CArrowArrayStreamGetNext get_next
+ CArrowArrayStreamGetSchema get_schema
+ CArrowArrayStreamRelease release
+
+ # ADBC
+ ctypedef uint8_t CAdbcStatusCode"AdbcStatusCode"
+ cdef CAdbcStatusCode ADBC_STATUS_OK
+ cdef CAdbcStatusCode ADBC_STATUS_UNKNOWN
+ cdef CAdbcStatusCode ADBC_STATUS_NOT_IMPLEMENTED
+ cdef CAdbcStatusCode ADBC_STATUS_NOT_FOUND
+ cdef CAdbcStatusCode ADBC_STATUS_ALREADY_EXISTS
+ cdef CAdbcStatusCode ADBC_STATUS_INVALID_ARGUMENT
+ cdef CAdbcStatusCode ADBC_STATUS_INVALID_STATE
+ cdef CAdbcStatusCode ADBC_STATUS_INVALID_DATA
+ cdef CAdbcStatusCode ADBC_STATUS_INTEGRITY
+ cdef CAdbcStatusCode ADBC_STATUS_INTERNAL
+ cdef CAdbcStatusCode ADBC_STATUS_IO
+ cdef CAdbcStatusCode ADBC_STATUS_CANCELLED
+ cdef CAdbcStatusCode ADBC_STATUS_TIMEOUT
+ cdef CAdbcStatusCode ADBC_STATUS_UNAUTHENTICATED
+ cdef CAdbcStatusCode ADBC_STATUS_UNAUTHORIZED
+
+ cdef const char* ADBC_OPTION_VALUE_DISABLED
+ cdef const char* ADBC_OPTION_VALUE_ENABLED
+
+ cdef const char* ADBC_CONNECTION_OPTION_AUTOCOMMIT
+ cdef const char* ADBC_INGEST_OPTION_TARGET_TABLE
+ cdef const char* ADBC_INGEST_OPTION_MODE
+ cdef const char* ADBC_INGEST_OPTION_MODE_APPEND
+ cdef const char* ADBC_INGEST_OPTION_MODE_CREATE
+ cdef const char* ADBC_INGEST_OPTION_MODE_REPLACE
+ cdef const char* ADBC_INGEST_OPTION_MODE_CREATE_APPEND
+
+ cdef int ADBC_OBJECT_DEPTH_ALL
+ cdef int ADBC_OBJECT_DEPTH_CATALOGS
+ cdef int ADBC_OBJECT_DEPTH_DB_SCHEMAS
+ cdef int ADBC_OBJECT_DEPTH_TABLES
+ cdef int ADBC_OBJECT_DEPTH_COLUMNS
+
+ cdef uint32_t ADBC_INFO_VENDOR_NAME
+ cdef uint32_t ADBC_INFO_VENDOR_VERSION
+ cdef uint32_t ADBC_INFO_VENDOR_ARROW_VERSION
+ cdef uint32_t ADBC_INFO_DRIVER_NAME
+ cdef uint32_t ADBC_INFO_DRIVER_VERSION
+ cdef uint32_t ADBC_INFO_DRIVER_ARROW_VERSION
+
+ ctypedef void (*CAdbcErrorRelease)(CAdbcError*)
+ ctypedef void (*CAdbcPartitionsRelease)(CAdbcPartitions*)
+
+ cdef struct CAdbcError"AdbcError":
+ char* message
+ int32_t vendor_code
+ char[5] sqlstate
+ CAdbcErrorRelease release
+
+ cdef struct CAdbcErrorDetail"AdbcErrorDetail":
+ char* key
+ uint8_t* value
+ size_t value_length
+
+ int AdbcErrorGetDetailCount(CAdbcError* error)
+ CAdbcErrorDetail AdbcErrorGetDetail(CAdbcError* error, int index)
+ CAdbcError* AdbcErrorFromArrayStream(CArrowArrayStream*, CAdbcStatusCode*)
+
+ cdef int ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA
+
+ cdef struct CAdbcDriver"AdbcDriver":
+ pass
+
+ cdef struct CAdbcDatabase"AdbcDatabase":
+ void* private_data
+
+ cdef struct CAdbcConnection"AdbcConnection":
+ void* private_data
+
+ cdef struct CAdbcStatement"AdbcStatement":
+ void* private_data
+
+ cdef struct CAdbcPartitions"AdbcPartitions":
+ size_t num_partitions
+ const uint8_t** partitions
+ const size_t* partition_lengths
+ void* private_data
+ CAdbcPartitionsRelease release
+
+ CAdbcStatusCode AdbcDatabaseNew(CAdbcDatabase* database, CAdbcError* error)
+ CAdbcStatusCode AdbcDatabaseGetOption(
+ CAdbcDatabase*, const char*, char*, size_t*, CAdbcError*);
+ CAdbcStatusCode AdbcDatabaseGetOptionBytes(
+ CAdbcDatabase*, const char*, uint8_t*, size_t*, CAdbcError*);
+ CAdbcStatusCode AdbcDatabaseGetOptionDouble(
+ CAdbcDatabase*, const char*, double*, CAdbcError*);
+ CAdbcStatusCode AdbcDatabaseGetOptionInt(
+ CAdbcDatabase*, const char*, int64_t*, CAdbcError*);
+ CAdbcStatusCode AdbcDatabaseSetOption(
+ CAdbcDatabase*, const char*, const char*, CAdbcError*)
+ CAdbcStatusCode AdbcDatabaseSetOptionBytes(
+ CAdbcDatabase*, const char*, const uint8_t*, size_t, CAdbcError*)
+ CAdbcStatusCode AdbcDatabaseSetOptionDouble(
+ CAdbcDatabase*, const char*, double, CAdbcError*)
+ CAdbcStatusCode AdbcDatabaseSetOptionInt(
+ CAdbcDatabase*, const char*, int64_t, CAdbcError*)
+ CAdbcStatusCode AdbcDatabaseInit(CAdbcDatabase* database, CAdbcError* error)
+ CAdbcStatusCode AdbcDatabaseRelease(CAdbcDatabase* database, CAdbcError* error)
+
+ ctypedef void (*CAdbcDriverInitFunc "AdbcDriverInitFunc")(int, void*, CAdbcError*)
+ CAdbcStatusCode AdbcDriverManagerDatabaseSetInitFunc(
+ CAdbcDatabase* database,
+ CAdbcDriverInitFunc init_func,
+ CAdbcError* error)
+
+ CAdbcStatusCode AdbcConnectionCancel(CAdbcConnection*, CAdbcError*)
+ CAdbcStatusCode AdbcConnectionCommit(
+ CAdbcConnection* connection,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcConnectionRollback(
+ CAdbcConnection* connection,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcConnectionReadPartition(
+ CAdbcConnection* connection,
+ const uint8_t* serialized_partition,
+ size_t serialized_length,
+ CArrowArrayStream* out,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcConnectionGetInfo(
+ CAdbcConnection* connection,
+ uint32_t* info_codes,
+ size_t info_codes_length,
+ CArrowArrayStream* stream,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcConnectionGetObjects(
+ CAdbcConnection* connection,
+ int depth,
+ const char* catalog,
+ const char* db_schema,
+ const char* table_name,
+ const char** table_type,
+ const char* column_name,
+ CArrowArrayStream* stream,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcConnectionGetOption(
+ CAdbcConnection*, const char*, char*, size_t*, CAdbcError*);
+ CAdbcStatusCode AdbcConnectionGetOptionBytes(
+ CAdbcConnection*, const char*, uint8_t*, size_t*, CAdbcError*);
+ CAdbcStatusCode AdbcConnectionGetOptionDouble(
+ CAdbcConnection*, const char*, double*, CAdbcError*);
+ CAdbcStatusCode AdbcConnectionGetOptionInt(
+ CAdbcConnection*, const char*, int64_t*, CAdbcError*);
+ CAdbcStatusCode AdbcConnectionGetStatistics(
+ CAdbcConnection*, const char*, const char*, const char*,
+ char, CArrowArrayStream*, CAdbcError*);
+ CAdbcStatusCode AdbcConnectionGetStatisticNames(
+ CAdbcConnection*, CArrowArrayStream*, CAdbcError*);
+ CAdbcStatusCode AdbcConnectionGetTableSchema(
+ CAdbcConnection* connection,
+ const char* catalog,
+ const char* db_schema,
+ const char* table_name,
+ CArrowSchema* schema,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcConnectionGetTableTypes(
+ CAdbcConnection* connection,
+ CArrowArrayStream* stream,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcConnectionInit(
+ CAdbcConnection* connection,
+ CAdbcDatabase* database,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcConnectionNew(
+ CAdbcConnection* connection,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcConnectionRelease(
+ CAdbcConnection* connection,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcConnectionSetOption(
+ CAdbcConnection*, const char*, const char*, CAdbcError*)
+ CAdbcStatusCode AdbcConnectionSetOptionBytes(
+ CAdbcConnection*, const char*, const uint8_t*, size_t, CAdbcError*)
+ CAdbcStatusCode AdbcConnectionSetOptionDouble(
+ CAdbcConnection*, const char*, double, CAdbcError*)
+ CAdbcStatusCode AdbcConnectionSetOptionInt(
+ CAdbcConnection*, const char*, int64_t, CAdbcError*)
+ CAdbcStatusCode AdbcStatementBind(
+ CAdbcStatement* statement,
+ CArrowArray*,
+ CArrowSchema*,
+ CAdbcError* error)
+
+ CAdbcStatusCode AdbcStatementCancel(CAdbcStatement*, CAdbcError*)
+ CAdbcStatusCode AdbcStatementBindStream(
+ CAdbcStatement* statement,
+ CArrowArrayStream*,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcStatementExecutePartitions(
+ CAdbcStatement* statement,
+ CArrowSchema* schema, CAdbcPartitions* partitions,
+ int64_t* rows_affected,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcStatementExecuteQuery(
+ CAdbcStatement* statement,
+ CArrowArrayStream* out, int64_t* rows_affected,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcStatementExecuteSchema(
+ CAdbcStatement*, CArrowSchema*, CAdbcError*)
+ CAdbcStatusCode AdbcStatementGetOption(
+ CAdbcStatement*, const char*, char*, size_t*, CAdbcError*);
+ CAdbcStatusCode AdbcStatementGetOptionBytes(
+ CAdbcStatement*, const char*, uint8_t*, size_t*, CAdbcError*);
+ CAdbcStatusCode AdbcStatementGetOptionDouble(
+ CAdbcStatement*, const char*, double*, CAdbcError*);
+ CAdbcStatusCode AdbcStatementGetOptionInt(
+ CAdbcStatement*, const char*, int64_t*, CAdbcError*);
+ CAdbcStatusCode AdbcStatementGetParameterSchema(
+ CAdbcStatement* statement,
+ CArrowSchema* schema,
+ CAdbcError* error);
+ CAdbcStatusCode AdbcStatementNew(
+ CAdbcConnection* connection,
+ CAdbcStatement* statement,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcStatementPrepare(
+ CAdbcStatement* statement,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcStatementSetOption(
+ CAdbcStatement*, const char*, const char*, CAdbcError*)
+ CAdbcStatusCode AdbcStatementSetOptionBytes(
+ CAdbcStatement*, const char*, const uint8_t*, size_t, CAdbcError*)
+ CAdbcStatusCode AdbcStatementSetOptionDouble(
+ CAdbcStatement*, const char*, double, CAdbcError*)
+ CAdbcStatusCode AdbcStatementSetOptionInt(
+ CAdbcStatement*, const char*, int64_t, CAdbcError*)
+ CAdbcStatusCode AdbcStatementSetSqlQuery(
+ CAdbcStatement* statement,
+ const char* query,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcStatementSetSubstraitPlan(
+ CAdbcStatement* statement,
+ const uint8_t* plan,
+ size_t length,
+ CAdbcError* error)
+ CAdbcStatusCode AdbcStatementRelease(
+ CAdbcStatement* statement,
+ CAdbcError* error)
+
+cdef const CAdbcError* PyAdbcErrorFromArrayStream(
+ CArrowArrayStream* stream, CAdbcStatusCode* status)
+
+cdef void check_error(CAdbcStatusCode status, CAdbcError* error) except *
diff --git a/python/adbc_driver_manager/adbc_driver_manager/_lib.pyx b/python/adbc_driver_manager/adbc_driver_manager/_lib.pyx
index 2acf0083..b7e8fd0a 100644
--- a/python/adbc_driver_manager/adbc_driver_manager/_lib.pyx
+++ b/python/adbc_driver_manager/adbc_driver_manager/_lib.pyx
@@ -33,254 +33,10 @@ from libcpp.vector cimport vector as c_vector
if typing.TYPE_CHECKING:
from typing import Self
-
-cdef extern from "adbc.h" nogil:
- # C ABI
- cdef struct CArrowSchema"ArrowSchema":
- pass
- cdef struct CArrowArray"ArrowArray":
- pass
- cdef struct CArrowArrayStream"ArrowArrayStream":
- pass
-
- # ADBC
- ctypedef uint8_t CAdbcStatusCode"AdbcStatusCode"
- cdef CAdbcStatusCode ADBC_STATUS_OK
- cdef CAdbcStatusCode ADBC_STATUS_UNKNOWN
- cdef CAdbcStatusCode ADBC_STATUS_NOT_IMPLEMENTED
- cdef CAdbcStatusCode ADBC_STATUS_NOT_FOUND
- cdef CAdbcStatusCode ADBC_STATUS_ALREADY_EXISTS
- cdef CAdbcStatusCode ADBC_STATUS_INVALID_ARGUMENT
- cdef CAdbcStatusCode ADBC_STATUS_INVALID_STATE
- cdef CAdbcStatusCode ADBC_STATUS_INVALID_DATA
- cdef CAdbcStatusCode ADBC_STATUS_INTEGRITY
- cdef CAdbcStatusCode ADBC_STATUS_INTERNAL
- cdef CAdbcStatusCode ADBC_STATUS_IO
- cdef CAdbcStatusCode ADBC_STATUS_CANCELLED
- cdef CAdbcStatusCode ADBC_STATUS_TIMEOUT
- cdef CAdbcStatusCode ADBC_STATUS_UNAUTHENTICATED
- cdef CAdbcStatusCode ADBC_STATUS_UNAUTHORIZED
-
- cdef const char* ADBC_OPTION_VALUE_DISABLED
- cdef const char* ADBC_OPTION_VALUE_ENABLED
-
- cdef const char* ADBC_CONNECTION_OPTION_AUTOCOMMIT
- cdef const char* ADBC_INGEST_OPTION_TARGET_TABLE
- cdef const char* ADBC_INGEST_OPTION_MODE
- cdef const char* ADBC_INGEST_OPTION_MODE_APPEND
- cdef const char* ADBC_INGEST_OPTION_MODE_CREATE
- cdef const char* ADBC_INGEST_OPTION_MODE_REPLACE
- cdef const char* ADBC_INGEST_OPTION_MODE_CREATE_APPEND
-
- cdef int ADBC_OBJECT_DEPTH_ALL
- cdef int ADBC_OBJECT_DEPTH_CATALOGS
- cdef int ADBC_OBJECT_DEPTH_DB_SCHEMAS
- cdef int ADBC_OBJECT_DEPTH_TABLES
- cdef int ADBC_OBJECT_DEPTH_COLUMNS
-
- cdef uint32_t ADBC_INFO_VENDOR_NAME
- cdef uint32_t ADBC_INFO_VENDOR_VERSION
- cdef uint32_t ADBC_INFO_VENDOR_ARROW_VERSION
- cdef uint32_t ADBC_INFO_DRIVER_NAME
- cdef uint32_t ADBC_INFO_DRIVER_VERSION
- cdef uint32_t ADBC_INFO_DRIVER_ARROW_VERSION
-
- ctypedef void (*CAdbcErrorRelease)(CAdbcError*)
- ctypedef void (*CAdbcPartitionsRelease)(CAdbcPartitions*)
-
- cdef struct CAdbcError"AdbcError":
- char* message
- int32_t vendor_code
- char[5] sqlstate
- CAdbcErrorRelease release
-
- cdef struct CAdbcErrorDetail"AdbcErrorDetail":
- char* key
- uint8_t* value
- size_t value_length
-
- int AdbcErrorGetDetailCount(CAdbcError* error)
- CAdbcErrorDetail AdbcErrorGetDetail(CAdbcError* error, int index)
-
- cdef int ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA
-
- cdef struct CAdbcDriver"AdbcDriver":
- pass
-
- cdef struct CAdbcDatabase"AdbcDatabase":
- void* private_data
-
- cdef struct CAdbcConnection"AdbcConnection":
- void* private_data
-
- cdef struct CAdbcStatement"AdbcStatement":
- void* private_data
-
- cdef struct CAdbcPartitions"AdbcPartitions":
- size_t num_partitions
- const uint8_t** partitions
- const size_t* partition_lengths
- void* private_data
- CAdbcPartitionsRelease release
-
- CAdbcStatusCode AdbcDatabaseNew(CAdbcDatabase* database, CAdbcError* error)
- CAdbcStatusCode AdbcDatabaseGetOption(
- CAdbcDatabase*, const char*, char*, size_t*, CAdbcError*);
- CAdbcStatusCode AdbcDatabaseGetOptionBytes(
- CAdbcDatabase*, const char*, uint8_t*, size_t*, CAdbcError*);
- CAdbcStatusCode AdbcDatabaseGetOptionDouble(
- CAdbcDatabase*, const char*, double*, CAdbcError*);
- CAdbcStatusCode AdbcDatabaseGetOptionInt(
- CAdbcDatabase*, const char*, int64_t*, CAdbcError*);
- CAdbcStatusCode AdbcDatabaseSetOption(
- CAdbcDatabase*, const char*, const char*, CAdbcError*)
- CAdbcStatusCode AdbcDatabaseSetOptionBytes(
- CAdbcDatabase*, const char*, const uint8_t*, size_t, CAdbcError*)
- CAdbcStatusCode AdbcDatabaseSetOptionDouble(
- CAdbcDatabase*, const char*, double, CAdbcError*)
- CAdbcStatusCode AdbcDatabaseSetOptionInt(
- CAdbcDatabase*, const char*, int64_t, CAdbcError*)
- CAdbcStatusCode AdbcDatabaseInit(CAdbcDatabase* database, CAdbcError* error)
- CAdbcStatusCode AdbcDatabaseRelease(CAdbcDatabase* database, CAdbcError* error)
-
- ctypedef void (*CAdbcDriverInitFunc "AdbcDriverInitFunc")(int, void*, CAdbcError*)
- CAdbcStatusCode AdbcDriverManagerDatabaseSetInitFunc(
- CAdbcDatabase* database,
- CAdbcDriverInitFunc init_func,
- CAdbcError* error)
-
- CAdbcStatusCode AdbcConnectionCancel(CAdbcConnection*, CAdbcError*)
- CAdbcStatusCode AdbcConnectionCommit(
- CAdbcConnection* connection,
- CAdbcError* error)
- CAdbcStatusCode AdbcConnectionRollback(
- CAdbcConnection* connection,
- CAdbcError* error)
- CAdbcStatusCode AdbcConnectionReadPartition(
- CAdbcConnection* connection,
- const uint8_t* serialized_partition,
- size_t serialized_length,
- CArrowArrayStream* out,
- CAdbcError* error)
- CAdbcStatusCode AdbcConnectionGetInfo(
- CAdbcConnection* connection,
- uint32_t* info_codes,
- size_t info_codes_length,
- CArrowArrayStream* stream,
- CAdbcError* error)
- CAdbcStatusCode AdbcConnectionGetObjects(
- CAdbcConnection* connection,
- int depth,
- const char* catalog,
- const char* db_schema,
- const char* table_name,
- const char** table_type,
- const char* column_name,
- CArrowArrayStream* stream,
- CAdbcError* error)
- CAdbcStatusCode AdbcConnectionGetOption(
- CAdbcConnection*, const char*, char*, size_t*, CAdbcError*);
- CAdbcStatusCode AdbcConnectionGetOptionBytes(
- CAdbcConnection*, const char*, uint8_t*, size_t*, CAdbcError*);
- CAdbcStatusCode AdbcConnectionGetOptionDouble(
- CAdbcConnection*, const char*, double*, CAdbcError*);
- CAdbcStatusCode AdbcConnectionGetOptionInt(
- CAdbcConnection*, const char*, int64_t*, CAdbcError*);
- CAdbcStatusCode AdbcConnectionGetStatistics(
- CAdbcConnection*, const char*, const char*, const char*,
- char, CArrowArrayStream*, CAdbcError*);
- CAdbcStatusCode AdbcConnectionGetStatisticNames(
- CAdbcConnection*, CArrowArrayStream*, CAdbcError*);
- CAdbcStatusCode AdbcConnectionGetTableSchema(
- CAdbcConnection* connection,
- const char* catalog,
- const char* db_schema,
- const char* table_name,
- CArrowSchema* schema,
- CAdbcError* error)
- CAdbcStatusCode AdbcConnectionGetTableTypes(
- CAdbcConnection* connection,
- CArrowArrayStream* stream,
- CAdbcError* error)
- CAdbcStatusCode AdbcConnectionInit(
- CAdbcConnection* connection,
- CAdbcDatabase* database,
- CAdbcError* error)
- CAdbcStatusCode AdbcConnectionNew(
- CAdbcConnection* connection,
- CAdbcError* error)
- CAdbcStatusCode AdbcConnectionRelease(
- CAdbcConnection* connection,
- CAdbcError* error)
- CAdbcStatusCode AdbcConnectionSetOption(
- CAdbcConnection*, const char*, const char*, CAdbcError*)
- CAdbcStatusCode AdbcConnectionSetOptionBytes(
- CAdbcConnection*, const char*, const uint8_t*, size_t, CAdbcError*)
- CAdbcStatusCode AdbcConnectionSetOptionDouble(
- CAdbcConnection*, const char*, double, CAdbcError*)
- CAdbcStatusCode AdbcConnectionSetOptionInt(
- CAdbcConnection*, const char*, int64_t, CAdbcError*)
- CAdbcStatusCode AdbcStatementBind(
- CAdbcStatement* statement,
- CArrowArray*,
- CArrowSchema*,
- CAdbcError* error)
-
- CAdbcStatusCode AdbcStatementCancel(CAdbcStatement*, CAdbcError*)
- CAdbcStatusCode AdbcStatementBindStream(
- CAdbcStatement* statement,
- CArrowArrayStream*,
- CAdbcError* error)
- CAdbcStatusCode AdbcStatementExecutePartitions(
- CAdbcStatement* statement,
- CArrowSchema* schema, CAdbcPartitions* partitions,
- int64_t* rows_affected,
- CAdbcError* error)
- CAdbcStatusCode AdbcStatementExecuteQuery(
- CAdbcStatement* statement,
- CArrowArrayStream* out, int64_t* rows_affected,
- CAdbcError* error)
- CAdbcStatusCode AdbcStatementExecuteSchema(
- CAdbcStatement*, CArrowSchema*, CAdbcError*)
- CAdbcStatusCode AdbcStatementGetOption(
- CAdbcStatement*, const char*, char*, size_t*, CAdbcError*);
- CAdbcStatusCode AdbcStatementGetOptionBytes(
- CAdbcStatement*, const char*, uint8_t*, size_t*, CAdbcError*);
- CAdbcStatusCode AdbcStatementGetOptionDouble(
- CAdbcStatement*, const char*, double*, CAdbcError*);
- CAdbcStatusCode AdbcStatementGetOptionInt(
- CAdbcStatement*, const char*, int64_t*, CAdbcError*);
- CAdbcStatusCode AdbcStatementGetParameterSchema(
- CAdbcStatement* statement,
- CArrowSchema* schema,
- CAdbcError* error);
- CAdbcStatusCode AdbcStatementNew(
- CAdbcConnection* connection,
- CAdbcStatement* statement,
- CAdbcError* error)
- CAdbcStatusCode AdbcStatementPrepare(
- CAdbcStatement* statement,
- CAdbcError* error)
- CAdbcStatusCode AdbcStatementSetOption(
- CAdbcStatement*, const char*, const char*, CAdbcError*)
- CAdbcStatusCode AdbcStatementSetOptionBytes(
- CAdbcStatement*, const char*, const uint8_t*, size_t, CAdbcError*)
- CAdbcStatusCode AdbcStatementSetOptionDouble(
- CAdbcStatement*, const char*, double, CAdbcError*)
- CAdbcStatusCode AdbcStatementSetOptionInt(
- CAdbcStatement*, const char*, int64_t, CAdbcError*)
- CAdbcStatusCode AdbcStatementSetSqlQuery(
- CAdbcStatement* statement,
- const char* query,
- CAdbcError* error)
- CAdbcStatusCode AdbcStatementSetSubstraitPlan(
- CAdbcStatement* statement,
- const uint8_t* plan,
- size_t length,
- CAdbcError* error)
- CAdbcStatusCode AdbcStatementRelease(
- CAdbcStatement* statement,
- CAdbcError* error)
+try:
+ import pyarrow
+except ImportError as e:
+ pyarrow = None
cdef extern from "adbc_driver_manager.h":
@@ -1633,3 +1389,8 @@ cdef class AdbcStatement(_AdbcHandle):
status = AdbcStatementSetSubstraitPlan(
&self.statement, c_plan, length, &c_error)
check_error(status, &c_error)
+
+
+cdef const CAdbcError* PyAdbcErrorFromArrayStream(
+ CArrowArrayStream* stream, CAdbcStatusCode* status):
+ return AdbcErrorFromArrayStream(stream, status)
diff --git a/.env b/python/adbc_driver_manager/adbc_driver_manager/_reader.pyi
similarity index 55%
copy from .env
copy to python/adbc_driver_manager/adbc_driver_manager/_reader.pyi
index 2b73edb5..48fc718d 100644
--- a/.env
+++ b/python/adbc_driver_manager/adbc_driver_manager/_reader.pyi
@@ -15,26 +15,20 @@
# specific language governing permissions and limitations
# under the License.
-# All of the following environment variables are required to set default values
-# for the parameters in docker-compose.yml.
+import typing
-# Default repository to pull and push images from
-REPO=apache/arrow-dev
+import pandas
+import pyarrow
-# different architecture notations
-ARCH=amd64
-ARCH_ALIAS=x86_64
-ARCH_SHORT=amd64
-ARCH_CONDA_FORGE=linux_64_
-
-# Default versions for various dependencies
-JDK=8
-MANYLINUX=2014
-MAVEN=3.5.4
-PYTHON=3.10
-GO=1.19.5
-ARROW_MAJOR_VERSION=12
-
-# Used through docker-compose.yml and serves as the default version for the
-# ci/scripts/install_vcpkg.sh script.
-VCPKG="2871ddd918cecb9cb642bcb9c56897f397283192"
+class AdbcRecordBatchReader(pyarrow.RecordBatchReader):
+ def close(self) -> None: ...
+ def read_all(self) -> pyarrow.Table: ...
+ def read_next_batch(self) -> pyarrow.RecordBatch: ...
+ def read_pandas(self, **kwargs) -> pandas.DataFrame: ...
+ @property
+ def schema(self) -> pyarrow.Schema: ...
+ @classmethod
+ def _import_from_c(cls, address: int) -> AdbcRecordBatchReader: ...
+ def __enter__(self) -> AdbcRecordBatchReader: ...
+ def __exit__(self, type, value, traceback) -> None: ...
+ def __iter__(self) -> typing.Iterator[pyarrow.RecordBatch]: ...
diff --git a/python/adbc_driver_manager/adbc_driver_manager/_reader.pyx b/python/adbc_driver_manager/adbc_driver_manager/_reader.pyx
new file mode 100644
index 00000000..43de6740
--- /dev/null
+++ b/python/adbc_driver_manager/adbc_driver_manager/_reader.pyx
@@ -0,0 +1,113 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# cython: language_level = 3
+
+import pyarrow
+from cython.operator cimport dereference as deref
+from libc.stdint cimport uintptr_t
+
+from ._lib cimport *
+
+
+cdef class _AdbcErrorHelper:
+ cdef:
+ CArrowArrayStream c_stream
+
+ def check_error(self, exception):
+ cdef:
+ CAdbcStatusCode c_status
+ const CAdbcError* error = PyAdbcErrorFromArrayStream(&self.c_stream, &c_status)
+ CAdbcErrorDetail detail
+
+ if error != NULL:
+ check_error(c_status, <CAdbcError*> error)
+
+ raise exception
+
+
+# Can't directly inherit pyarrow.RecordBatchReader, but we want to in order to
+# pass isinstance checks
+class AdbcRecordBatchReader(pyarrow.RecordBatchReader):
+ def __init__(self, reader, helper):
+ self._reader = reader
+ self._helper = helper
+
+ def close(self):
+ self._reader.close()
+
+ @classmethod
+ def _import_from_c(cls, address) -> AdbcRecordBatchReader:
+ cdef:
+ CArrowArrayStream* c_stream = <CArrowArrayStream*> <uintptr_t> address
+ _AdbcErrorHelper helper = _AdbcErrorHelper.__new__(_AdbcErrorHelper)
+ # Save a copy of the stream to use
+ helper.c_stream = deref(c_stream)
+ try:
+ reader = pyarrow.RecordBatchReader._import_from_c(int(address))
+ except Exception as e:
+ helper.check_error(e)
+ return cls(reader, helper)
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_info, exc_val, exc_tb):
+ pass
+
+ def __iter__(self):
+ try:
+ yield from self._reader
+ except Exception as e:
+ self._helper.check_error(e)
+
+ def iter_batches_with_custom_metadata(self):
+ try:
+ return self._reader.iter_batches_with_custom_metadata()
+ except Exception as e:
+ self._helper.check_error(e)
+
+ def read_all(self):
+ try:
+ return self._reader.read_all()
+ except Exception as e:
+ self._helper.check_error(e)
+
+ def read_next_batch(self, *args, **kwargs):
+ try:
+ return self._reader.read_next_batch(*args, **kwargs)
+ except Exception as e:
+ self._helper.check_error(e)
+
+ def read_next_batch_with_custom_metadata(self):
+ try:
+ return self._reader.read_next_batch_with_custom_metadata()
+ except Exception as e:
+ self._helper.check_error(e)
+
+ def read_pandas(self, *args, **kwargs):
+ try:
+ return self._reader.read_pandas(*args, **kwargs)
+ except Exception as e:
+ self._helper.check_error(e)
+
+ @property
+ def schema(self):
+ try:
+ return self._reader.schema
+ except Exception as e:
+ self._helper.check_error(e)
diff --git a/python/adbc_driver_manager/adbc_driver_manager/dbapi.py b/python/adbc_driver_manager/adbc_driver_manager/dbapi.py
index 6c32429e..6e201b1b 100644
--- a/python/adbc_driver_manager/adbc_driver_manager/dbapi.py
+++ b/python/adbc_driver_manager/adbc_driver_manager/dbapi.py
@@ -45,7 +45,7 @@ except ImportError as e:
import adbc_driver_manager
-from . import _lib
+from . import _lib, _reader
if typing.TYPE_CHECKING:
import pandas
@@ -668,7 +668,8 @@ class Cursor(_Closeable):
self._prepare_execute(operation, parameters)
handle, self._rowcount = self._stmt.execute_query()
self._results = _RowIterator(
- pyarrow.RecordBatchReader._import_from_c(handle.address)
+ # pyarrow.RecordBatchReader._import_from_c(handle.address)
+ _reader.AdbcRecordBatchReader._import_from_c(handle.address)
)
def executemany(self, operation: Union[bytes, str], seq_of_parameters) -> None:
diff --git a/python/adbc_driver_manager/setup.py b/python/adbc_driver_manager/setup.py
index 9584b78b..075ae861 100644
--- a/python/adbc_driver_manager/setup.py
+++ b/python/adbc_driver_manager/setup.py
@@ -97,7 +97,16 @@ setup(
"adbc_driver_manager/_lib.pyx",
"adbc_driver_manager/adbc_driver_manager.cc",
],
- )
+ ),
+ Extension(
+ name="adbc_driver_manager._reader",
+ extra_compile_args=extra_compile_args,
+ include_dirs=[str(source_root.joinpath("adbc_driver_manager").resolve())],
+ language="c++",
+ sources=[
+ "adbc_driver_manager/_reader.pyx",
+ ],
+ ),
],
version=version,
)
diff --git a/python/adbc_driver_manager/tests/test_reader.py b/python/adbc_driver_manager/tests/test_reader.py
new file mode 100644
index 00000000..54d2c7d5
--- /dev/null
+++ b/python/adbc_driver_manager/tests/test_reader.py
@@ -0,0 +1,80 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pyarrow
+import pytest
+
+from adbc_driver_manager._lib import ArrowArrayStreamHandle
+from adbc_driver_manager._reader import AdbcRecordBatchReader
+
+schema = pyarrow.schema([("ints", "int32")])
+batches = [
+ pyarrow.record_batch([[1, 2, 3, 4]], schema=schema),
+]
+
+
+def _make_reader():
+ original = pyarrow.RecordBatchReader.from_batches(schema, batches)
+ exported = ArrowArrayStreamHandle()
+ original._export_to_c(exported.address)
+ return AdbcRecordBatchReader._import_from_c(exported.address)
+
+
+def test_reader():
+ wrapped = _make_reader()
+ assert wrapped.read_next_batch() == batches[0]
+
+
+def test_reader_error():
+ schema = pyarrow.schema([("ints", "int32")])
+
+ def batches():
+ yield pyarrow.record_batch([[1, 2, 3, 4]], schema=schema)
+ raise ValueError("foo")
+
+ original = pyarrow.RecordBatchReader.from_batches(schema, batches())
+ exported = ArrowArrayStreamHandle()
+ original._export_to_c(exported.address)
+ wrapped = AdbcRecordBatchReader._import_from_c(exported.address)
+
+ assert wrapped.read_next_batch() is not None
+ with pytest.raises(pyarrow.ArrowInvalid):
+ wrapped.read_next_batch()
+
+
+def test_reader_methods():
+ with _make_reader() as reader:
+ assert reader.read_all() == pyarrow.Table.from_batches(batches, schema)
+
+ with _make_reader() as reader:
+ assert reader.read_pandas() is not None
+
+ with _make_reader() as reader:
+ for batch in reader:
+ assert batch == batches[0]
+
+ with _make_reader() as reader:
+ with pytest.raises(NotImplementedError):
+ assert reader.read_next_batch_with_custom_metadata() is not None
+
+ with _make_reader() as reader:
+ with pytest.raises(NotImplementedError):
+ for batch in reader.iter_batches_with_custom_metadata():
+ assert batch == batches[0]
+
+ with _make_reader() as reader:
+ assert reader.schema == schema
diff --git a/python/adbc_driver_postgresql/tests/test_dbapi.py b/python/adbc_driver_postgresql/tests/test_dbapi.py
index ca6908b6..f0551fb8 100644
--- a/python/adbc_driver_postgresql/tests/test_dbapi.py
+++ b/python/adbc_driver_postgresql/tests/test_dbapi.py
@@ -106,30 +106,35 @@ def test_query_cancel(postgres: dbapi.Connection) -> None:
"""
INSERT INTO test_batch_size (ints)
SELECT generated :: INT
- FROM GENERATE_SERIES(1, 65536) temp(generated)
+ FROM GENERATE_SERIES(1, 1048576) temp(generated)
"""
)
+ postgres.commit()
+ # Ensure different ways of reading all raise the desired error
+ with postgres.cursor() as cur:
cur.execute("SELECT * FROM test_batch_size")
cur.adbc_cancel()
- # XXX(https://github.com/apache/arrow-adbc/issues/940):
- # PyArrow swallows the errno and doesn't set it into the
- # OSError, so we have no clue what happened here. (Though the
- # driver does properly return ECANCELED.)
- with pytest.raises(OSError, match="canceling statement"):
+ with pytest.raises(postgres.OperationalError, match="canceling statement"):
cur.fetchone()
+ with postgres.cursor() as cur:
+ cur.execute("SELECT * FROM test_batch_size")
+ cur.adbc_cancel()
+ with pytest.raises(postgres.OperationalError, match="canceling statement"):
+ cur.fetch_arrow_table()
-def test_query_execute_schema(postgres: dbapi.Connection) -> None:
with postgres.cursor() as cur:
- schema = cur.adbc_execute_schema("SELECT 1 AS foo")
- assert schema == pyarrow.schema([("foo", "int32")])
+ cur.execute("SELECT * FROM test_batch_size")
+ cur.adbc_cancel()
+ with pytest.raises(postgres.OperationalError, match="canceling statement"):
+ cur.fetch_df()
-def test_query_trivial(postgres: dbapi.Connection):
+def test_query_execute_schema(postgres: dbapi.Connection) -> None:
with postgres.cursor() as cur:
- cur.execute("SELECT 1")
- assert cur.fetchone() == (1,)
+ schema = cur.adbc_execute_schema("SELECT 1 AS foo")
+ assert schema == pyarrow.schema([("foo", "int32")])
def test_query_invalid(postgres: dbapi.Connection) -> None:
@@ -143,6 +148,12 @@ def test_query_invalid(postgres: dbapi.Connection) -> None:
assert len(excinfo.value.details) > 0
+def test_query_trivial(postgres: dbapi.Connection):
+ with postgres.cursor() as cur:
+ cur.execute("SELECT 1")
+ assert cur.fetchone() == (1,)
+
+
def test_stmt_ingest(postgres: dbapi.Connection) -> None:
table = pyarrow.table(
[
diff --git a/r/adbcflightsql/tools/download-go.R b/r/adbcflightsql/tools/download-go.R
index 3a9c40c7..5fc986d7 100644
--- a/r/adbcflightsql/tools/download-go.R
+++ b/r/adbcflightsql/tools/download-go.R
@@ -17,7 +17,7 @@
tmp_dir <- "src/go/tmp"
-go_version <- Sys.getenv("R_ADBC_GO_VERSION_DOWNLOAD", "1.19.9")
+go_version <- Sys.getenv("R_ADBC_GO_VERSION_DOWNLOAD", "1.20.7")
go_platform <- tolower(Sys.info()[["sysname"]])
if (!(go_platform %in% c("darwin", "linux", "windows"))) {
diff --git a/r/adbcsnowflake/tools/download-go.R b/r/adbcsnowflake/tools/download-go.R
index 3a9c40c7..5fc986d7 100644
--- a/r/adbcsnowflake/tools/download-go.R
+++ b/r/adbcsnowflake/tools/download-go.R
@@ -17,7 +17,7 @@
tmp_dir <- "src/go/tmp"
-go_version <- Sys.getenv("R_ADBC_GO_VERSION_DOWNLOAD", "1.19.9")
+go_version <- Sys.getenv("R_ADBC_GO_VERSION_DOWNLOAD", "1.20.7")
go_platform <- tolower(Sys.info()[["sysname"]])
if (!(go_platform %in% c("darwin", "linux", "windows"))) {