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/03 14:24:35 UTC
[arrow-adbc] 01/04: feat(c/driver/postgresql): implement ADBC 1.1.0 features (#852)
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
commit 61fbe7618e27e0a7e7170d72038284e43b647d89
Author: David Li <li...@gmail.com>
AuthorDate: Wed Jul 26 11:16:36 2023 -0400
feat(c/driver/postgresql): implement ADBC 1.1.0 features (#852)
- ADBC_INFO_DRIVER_ADBC_VERSION
- StatementExecuteSchema (#318)
- ADBC_CONNECTION_OPTION_CURRENT_{CATALOG, DB_SCHEMA} (#319)
- Cancellation
- Get/SetOption
- New ingest modes
---
adbc.h | 12 +
c/driver/common/utils.c | 13 +
c/driver/common/utils.h | 3 +
c/driver/flightsql/dremio_flightsql_test.cc | 3 +-
c/driver/flightsql/sqlite_flightsql_test.cc | 3 +-
c/driver/postgresql/connection.cc | 111 ++++++-
c/driver/postgresql/connection.h | 17 +-
c/driver/postgresql/database.cc | 35 +++
c/driver/postgresql/database.h | 12 +
c/driver/postgresql/postgres_copy_reader.h | 8 +-
c/driver/postgresql/postgresql.cc | 398 +++++++++++++++++++++++--
c/driver/postgresql/postgresql_test.cc | 121 ++++++--
c/driver/postgresql/statement.cc | 241 +++++++++++----
c/driver/postgresql/statement.h | 22 +-
c/driver/snowflake/snowflake_test.cc | 2 +-
c/driver/sqlite/sqlite.c | 291 +++++++++++++++++-
c/driver/sqlite/sqlite_test.cc | 20 ++
c/driver_manager/CMakeLists.txt | 5 +-
c/driver_manager/adbc_driver_manager.cc | 40 +--
c/driver_manager/adbc_driver_manager_test.cc | 22 +-
c/integration/duckdb/duckdb_test.cc | 4 +-
c/validation/CMakeLists.txt | 19 +-
c/validation/adbc_validation.cc | 421 +++++++++++++++++++++++++--
c/validation/adbc_validation.h | 86 ++++--
go/adbc/drivermgr/adbc.h | 12 +
go/adbc/drivermgr/adbc_driver_manager.cc | 40 +--
26 files changed, 1752 insertions(+), 209 deletions(-)
diff --git a/adbc.h b/adbc.h
index f7ea69af..9d9fa923 100644
--- a/adbc.h
+++ b/adbc.h
@@ -1022,6 +1022,7 @@ AdbcStatusCode AdbcDatabaseNew(struct AdbcDatabase* database, struct AdbcError*
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcDatabaseGetOption(struct AdbcDatabase* database, const char* key,
char* value, size_t* length,
struct AdbcError* error);
@@ -1062,6 +1063,7 @@ AdbcStatusCode AdbcDatabaseGetOption(struct AdbcDatabase* database, const char*
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcDatabaseGetOptionBytes(struct AdbcDatabase* database, const char* key,
uint8_t* value, size_t* length,
struct AdbcError* error);
@@ -1086,6 +1088,7 @@ AdbcStatusCode AdbcDatabaseGetOptionBytes(struct AdbcDatabase* database, const c
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcDatabaseGetOptionDouble(struct AdbcDatabase* database, const char* key,
double* value, struct AdbcError* error);
@@ -1109,6 +1112,7 @@ AdbcStatusCode AdbcDatabaseGetOptionDouble(struct AdbcDatabase* database, const
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcDatabaseGetOptionInt(struct AdbcDatabase* database, const char* key,
int64_t* value, struct AdbcError* error);
@@ -1518,6 +1522,7 @@ AdbcStatusCode AdbcConnectionGetObjects(struct AdbcConnection* connection, int d
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcConnectionGetOption(struct AdbcConnection* connection, const char* key,
char* value, size_t* length,
struct AdbcError* error);
@@ -1558,6 +1563,7 @@ AdbcStatusCode AdbcConnectionGetOption(struct AdbcConnection* connection, const
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcConnectionGetOptionBytes(struct AdbcConnection* connection,
const char* key, uint8_t* value,
size_t* length, struct AdbcError* error);
@@ -1582,6 +1588,7 @@ AdbcStatusCode AdbcConnectionGetOptionBytes(struct AdbcConnection* connection,
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcConnectionGetOptionInt(struct AdbcConnection* connection,
const char* key, int64_t* value,
struct AdbcError* error);
@@ -1606,6 +1613,7 @@ AdbcStatusCode AdbcConnectionGetOptionInt(struct AdbcConnection* connection,
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcConnectionGetOptionDouble(struct AdbcConnection* connection,
const char* key, double* value,
struct AdbcError* error);
@@ -2005,6 +2013,7 @@ AdbcStatusCode AdbcStatementCancel(struct AdbcStatement* statement,
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcStatementGetOption(struct AdbcStatement* statement, const char* key,
char* value, size_t* length,
struct AdbcError* error);
@@ -2045,6 +2054,7 @@ AdbcStatusCode AdbcStatementGetOption(struct AdbcStatement* statement, const cha
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcStatementGetOptionBytes(struct AdbcStatement* statement,
const char* key, uint8_t* value,
size_t* length, struct AdbcError* error);
@@ -2069,6 +2079,7 @@ AdbcStatusCode AdbcStatementGetOptionBytes(struct AdbcStatement* statement,
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcStatementGetOptionInt(struct AdbcStatement* statement, const char* key,
int64_t* value, struct AdbcError* error);
@@ -2092,6 +2103,7 @@ AdbcStatusCode AdbcStatementGetOptionInt(struct AdbcStatement* statement, const
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcStatementGetOptionDouble(struct AdbcStatement* statement,
const char* key, double* value,
struct AdbcError* error);
diff --git a/c/driver/common/utils.c b/c/driver/common/utils.c
index dfac14f5..eb01bc18 100644
--- a/c/driver/common/utils.c
+++ b/c/driver/common/utils.c
@@ -244,6 +244,19 @@ AdbcStatusCode AdbcConnectionGetInfoAppendString(struct ArrowArray* array,
return ADBC_STATUS_OK;
}
+AdbcStatusCode AdbcConnectionGetInfoAppendInt(struct ArrowArray* array,
+ uint32_t info_code, int64_t info_value,
+ struct AdbcError* error) {
+ CHECK_NA(INTERNAL, ArrowArrayAppendUInt(array->children[0], info_code), error);
+ // Append to type variant
+ CHECK_NA(INTERNAL, ArrowArrayAppendInt(array->children[1]->children[2], info_value),
+ error);
+ // Append type code/offset
+ CHECK_NA(INTERNAL, ArrowArrayFinishUnionElement(array->children[1], /*type_id=*/2),
+ error);
+ return ADBC_STATUS_OK;
+}
+
AdbcStatusCode AdbcInitConnectionObjectsSchema(struct ArrowSchema* schema,
struct AdbcError* error) {
ArrowSchemaInit(schema);
diff --git a/c/driver/common/utils.h b/c/driver/common/utils.h
index 5735bb94..381c7b05 100644
--- a/c/driver/common/utils.h
+++ b/c/driver/common/utils.h
@@ -117,6 +117,9 @@ AdbcStatusCode AdbcConnectionGetInfoAppendString(struct ArrowArray* array,
uint32_t info_code,
const char* info_value,
struct AdbcError* error);
+AdbcStatusCode AdbcConnectionGetInfoAppendInt(struct ArrowArray* array,
+ uint32_t info_code, int64_t info_value,
+ struct AdbcError* error);
AdbcStatusCode AdbcInitConnectionObjectsSchema(struct ArrowSchema* schema,
struct AdbcError* error);
diff --git a/c/driver/flightsql/dremio_flightsql_test.cc b/c/driver/flightsql/dremio_flightsql_test.cc
index c128bd49..416b8aea 100644
--- a/c/driver/flightsql/dremio_flightsql_test.cc
+++ b/c/driver/flightsql/dremio_flightsql_test.cc
@@ -42,11 +42,11 @@ class DremioFlightSqlQuirks : public adbc_validation::DriverQuirks {
}
std::string BindParameter(int index) const override { return "?"; }
+ bool supports_bulk_ingest(const char* /*mode*/) const override { return false; }
bool supports_concurrent_statements() const override { return true; }
bool supports_transactions() const override { return false; }
bool supports_get_sql_info() const override { return false; }
bool supports_get_objects() const override { return true; }
- bool supports_bulk_ingest() const override { return false; }
bool supports_partitioned_data() const override { return true; }
bool supports_dynamic_parameter_binding() const override { return false; }
};
@@ -87,6 +87,7 @@ class DremioFlightSqlStatementTest : public ::testing::Test,
void SetUp() override { ASSERT_NO_FATAL_FAILURE(SetUpTest()); }
void TearDown() override { ASSERT_NO_FATAL_FAILURE(TearDownTest()); }
+ void TestResultInvalidation() { GTEST_SKIP() << "Dremio generates a CANCELLED"; }
void TestSqlIngestTableEscaping() { GTEST_SKIP() << "Table escaping not implemented"; }
protected:
diff --git a/c/driver/flightsql/sqlite_flightsql_test.cc b/c/driver/flightsql/sqlite_flightsql_test.cc
index 019a92c9..cd697dd9 100644
--- a/c/driver/flightsql/sqlite_flightsql_test.cc
+++ b/c/driver/flightsql/sqlite_flightsql_test.cc
@@ -90,6 +90,8 @@ class SqliteFlightSqlQuirks : public adbc_validation::DriverQuirks {
}
std::string BindParameter(int index) const override { return "?"; }
+
+ bool supports_bulk_ingest(const char* /*mode*/) const override { return false; }
bool supports_concurrent_statements() const override { return true; }
bool supports_transactions() const override { return false; }
bool supports_get_sql_info() const override { return true; }
@@ -113,7 +115,6 @@ class SqliteFlightSqlQuirks : public adbc_validation::DriverQuirks {
}
}
bool supports_get_objects() const override { return true; }
- bool supports_bulk_ingest() const override { return false; }
bool supports_partitioned_data() const override { return true; }
bool supports_dynamic_parameter_binding() const override { return true; }
};
diff --git a/c/driver/postgresql/connection.cc b/c/driver/postgresql/connection.cc
index 08ff9027..1c1b0507 100644
--- a/c/driver/postgresql/connection.cc
+++ b/c/driver/postgresql/connection.cc
@@ -36,8 +36,9 @@
namespace {
static const uint32_t kSupportedInfoCodes[] = {
- ADBC_INFO_VENDOR_NAME, ADBC_INFO_VENDOR_VERSION, ADBC_INFO_DRIVER_NAME,
- ADBC_INFO_DRIVER_VERSION, ADBC_INFO_DRIVER_ARROW_VERSION,
+ ADBC_INFO_VENDOR_NAME, ADBC_INFO_VENDOR_VERSION,
+ ADBC_INFO_DRIVER_NAME, ADBC_INFO_DRIVER_VERSION,
+ ADBC_INFO_DRIVER_ARROW_VERSION, ADBC_INFO_DRIVER_ADBC_VERSION,
};
static const std::unordered_map<std::string, std::string> kPgTableTypes = {
@@ -114,8 +115,10 @@ class PqResultHelper {
result_ = PQexecPrepared(conn_, "", param_values_.size(), param_c_strs.data(), NULL,
NULL, 0);
- if (PQresultStatus(result_) != PGRES_TUPLES_OK) {
- SetError(error_, "[libpq] Failed to execute query: %s", PQerrorMessage(conn_));
+ ExecStatusType status = PQresultStatus(result_);
+ if (status != PGRES_TUPLES_OK && status != PGRES_COMMAND_OK) {
+ SetError(error_, "[libpq] Failed to execute query '%s': %s", query_.c_str(),
+ PQerrorMessage(conn_));
return ADBC_STATUS_IO;
}
@@ -729,6 +732,20 @@ class PqGetObjectsHelper {
namespace adbcpq {
+AdbcStatusCode PostgresConnection::Cancel(struct AdbcError* error) {
+ // > errbuf must be a char array of size errbufsize (the recommended size is
+ // > 256 bytes).
+ // https://www.postgresql.org/docs/current/libpq-cancel.html
+ char errbuf[256];
+ // > The return value is 1 if the cancel request was successfully dispatched
+ // > and 0 if not.
+ if (PQcancel(cancel_, errbuf, sizeof(errbuf)) != 1) {
+ SetError(error, "[libpq] Failed to cancel operation: %s", errbuf);
+ return ADBC_STATUS_UNKNOWN;
+ }
+ return ADBC_STATUS_OK;
+}
+
AdbcStatusCode PostgresConnection::Commit(struct AdbcError* error) {
if (autocommit_) {
SetError(error, "%s", "[libpq] Cannot commit when autocommit is enabled");
@@ -776,6 +793,10 @@ AdbcStatusCode PostgresConnectionGetInfoImpl(const uint32_t* info_codes,
RAISE_ADBC(AdbcConnectionGetInfoAppendString(array, info_codes[i],
NANOARROW_VERSION, error));
break;
+ case ADBC_INFO_DRIVER_ADBC_VERSION:
+ RAISE_ADBC(AdbcConnectionGetInfoAppendInt(array, info_codes[i],
+ ADBC_VERSION_1_1_0, error));
+ break;
default:
// Ignore
continue;
@@ -840,6 +861,47 @@ AdbcStatusCode PostgresConnection::GetObjects(
return BatchToArrayStream(&array, &schema, out, error);
}
+AdbcStatusCode PostgresConnection::GetOption(const char* option, char* value,
+ size_t* length, struct AdbcError* error) {
+ std::string output;
+ if (std::strcmp(option, ADBC_CONNECTION_OPTION_CURRENT_CATALOG) == 0) {
+ output = PQdb(conn_);
+ } else if (std::strcmp(option, ADBC_CONNECTION_OPTION_CURRENT_DB_SCHEMA) == 0) {
+ PqResultHelper result_helper{conn_, "SELECT CURRENT_SCHEMA", {}, error};
+ RAISE_ADBC(result_helper.Prepare());
+ RAISE_ADBC(result_helper.Execute());
+ auto it = result_helper.begin();
+ if (it == result_helper.end()) {
+ SetError(error, "[libpq] PostgreSQL returned no rows for 'SELECT CURRENT_SCHEMA'");
+ return ADBC_STATUS_INTERNAL;
+ }
+ output = (*it)[0].data;
+ } else if (std::strcmp(option, ADBC_CONNECTION_OPTION_AUTOCOMMIT) == 0) {
+ output = autocommit_ ? ADBC_OPTION_VALUE_ENABLED : ADBC_OPTION_VALUE_DISABLED;
+ } else {
+ return ADBC_STATUS_NOT_FOUND;
+ }
+
+ if (output.size() + 1 <= *length) {
+ std::memcpy(value, output.c_str(), output.size() + 1);
+ }
+ *length = output.size() + 1;
+ return ADBC_STATUS_OK;
+}
+AdbcStatusCode PostgresConnection::GetOptionBytes(const char* option, uint8_t* value,
+ size_t* length,
+ struct AdbcError* error) {
+ return ADBC_STATUS_NOT_FOUND;
+}
+AdbcStatusCode PostgresConnection::GetOptionInt(const char* option, int64_t* value,
+ struct AdbcError* error) {
+ return ADBC_STATUS_NOT_FOUND;
+}
+AdbcStatusCode PostgresConnection::GetOptionDouble(const char* option, double* value,
+ struct AdbcError* error) {
+ return ADBC_STATUS_NOT_FOUND;
+}
+
AdbcStatusCode PostgresConnection::GetTableSchema(const char* catalog,
const char* db_schema,
const char* table_name,
@@ -964,16 +1026,26 @@ AdbcStatusCode PostgresConnection::GetTableTypes(struct AdbcConnection* connecti
AdbcStatusCode PostgresConnection::Init(struct AdbcDatabase* database,
struct AdbcError* error) {
if (!database || !database->private_data) {
- SetError(error, "%s", "[libpq] Must provide an initialized AdbcDatabase");
+ SetError(error, "[libpq] Must provide an initialized AdbcDatabase");
return ADBC_STATUS_INVALID_ARGUMENT;
}
database_ =
*reinterpret_cast<std::shared_ptr<PostgresDatabase>*>(database->private_data);
type_resolver_ = database_->type_resolver();
- return database_->Connect(&conn_, error);
+ RAISE_ADBC(database_->Connect(&conn_, error));
+ cancel_ = PQgetCancel(conn_);
+ if (!cancel_) {
+ SetError(error, "[libpq] Could not initialize PGcancel");
+ return ADBC_STATUS_UNKNOWN;
+ }
+ return ADBC_STATUS_OK;
}
AdbcStatusCode PostgresConnection::Release(struct AdbcError* error) {
+ if (cancel_) {
+ PQfreeCancel(cancel_);
+ cancel_ = nullptr;
+ }
if (conn_) {
return database_->Disconnect(&conn_, error);
}
@@ -1023,8 +1095,35 @@ AdbcStatusCode PostgresConnection::SetOption(const char* key, const char* value,
autocommit_ = autocommit;
}
return ADBC_STATUS_OK;
+ } else if (std::strcmp(key, ADBC_CONNECTION_OPTION_CURRENT_DB_SCHEMA) == 0) {
+ // PostgreSQL doesn't accept a parameter here
+ PqResultHelper result_helper{
+ conn_, std::string("SET search_path TO ") + value, {}, error};
+ RAISE_ADBC(result_helper.Prepare());
+ RAISE_ADBC(result_helper.Execute());
+ return ADBC_STATUS_OK;
}
SetError(error, "%s%s", "[libpq] Unknown option ", key);
return ADBC_STATUS_NOT_IMPLEMENTED;
}
+
+AdbcStatusCode PostgresConnection::SetOptionBytes(const char* key, const uint8_t* value,
+ size_t length,
+ struct AdbcError* error) {
+ SetError(error, "%s%s", "[libpq] Unknown option ", key);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode PostgresConnection::SetOptionDouble(const char* key, double value,
+ struct AdbcError* error) {
+ SetError(error, "%s%s", "[libpq] Unknown option ", key);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode PostgresConnection::SetOptionInt(const char* key, int64_t value,
+ struct AdbcError* error) {
+ SetError(error, "%s%s", "[libpq] Unknown option ", key);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
} // namespace adbcpq
diff --git a/c/driver/postgresql/connection.h b/c/driver/postgresql/connection.h
index 74315ee0..a9900564 100644
--- a/c/driver/postgresql/connection.h
+++ b/c/driver/postgresql/connection.h
@@ -29,8 +29,10 @@ namespace adbcpq {
class PostgresDatabase;
class PostgresConnection {
public:
- PostgresConnection() : database_(nullptr), conn_(nullptr), autocommit_(true) {}
+ PostgresConnection()
+ : database_(nullptr), conn_(nullptr), cancel_(nullptr), autocommit_(true) {}
+ AdbcStatusCode Cancel(struct AdbcError* error);
AdbcStatusCode Commit(struct AdbcError* error);
AdbcStatusCode GetInfo(struct AdbcConnection* connection, uint32_t* info_codes,
size_t info_codes_length, struct ArrowArrayStream* out,
@@ -40,6 +42,14 @@ class PostgresConnection {
const char* table_name, const char** table_types,
const char* column_name, struct ArrowArrayStream* out,
struct AdbcError* error);
+ AdbcStatusCode GetOption(const char* option, char* value, size_t* length,
+ struct AdbcError* error);
+ AdbcStatusCode GetOptionBytes(const char* option, uint8_t* value, size_t* length,
+ struct AdbcError* error);
+ AdbcStatusCode GetOptionDouble(const char* option, double* value,
+ struct AdbcError* error);
+ AdbcStatusCode GetOptionInt(const char* option, int64_t* value,
+ struct AdbcError* error);
AdbcStatusCode GetTableSchema(const char* catalog, const char* db_schema,
const char* table_name, struct ArrowSchema* schema,
struct AdbcError* error);
@@ -49,6 +59,10 @@ class PostgresConnection {
AdbcStatusCode Release(struct AdbcError* error);
AdbcStatusCode Rollback(struct AdbcError* error);
AdbcStatusCode SetOption(const char* key, const char* value, struct AdbcError* error);
+ AdbcStatusCode SetOptionBytes(const char* key, const uint8_t* value, size_t length,
+ struct AdbcError* error);
+ AdbcStatusCode SetOptionDouble(const char* key, double value, struct AdbcError* error);
+ AdbcStatusCode SetOptionInt(const char* key, int64_t value, struct AdbcError* error);
PGconn* conn() const { return conn_; }
const std::shared_ptr<PostgresTypeResolver>& type_resolver() const {
@@ -60,6 +74,7 @@ class PostgresConnection {
std::shared_ptr<PostgresDatabase> database_;
std::shared_ptr<PostgresTypeResolver> type_resolver_;
PGconn* conn_;
+ PGcancel* cancel_;
bool autocommit_;
};
} // namespace adbcpq
diff --git a/c/driver/postgresql/database.cc b/c/driver/postgresql/database.cc
index 3976c4b0..2e18b5bf 100644
--- a/c/driver/postgresql/database.cc
+++ b/c/driver/postgresql/database.cc
@@ -36,6 +36,23 @@ PostgresDatabase::PostgresDatabase() : open_connections_(0) {
}
PostgresDatabase::~PostgresDatabase() = default;
+AdbcStatusCode PostgresDatabase::GetOption(const char* option, char* value,
+ size_t* length, struct AdbcError* error) {
+ return ADBC_STATUS_NOT_FOUND;
+}
+AdbcStatusCode PostgresDatabase::GetOptionBytes(const char* option, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ return ADBC_STATUS_NOT_FOUND;
+}
+AdbcStatusCode PostgresDatabase::GetOptionInt(const char* option, int64_t* value,
+ struct AdbcError* error) {
+ return ADBC_STATUS_NOT_FOUND;
+}
+AdbcStatusCode PostgresDatabase::GetOptionDouble(const char* option, double* value,
+ struct AdbcError* error) {
+ return ADBC_STATUS_NOT_FOUND;
+}
+
AdbcStatusCode PostgresDatabase::Init(struct AdbcError* error) {
// Connect to validate the parameters.
return RebuildTypeResolver(error);
@@ -61,6 +78,24 @@ AdbcStatusCode PostgresDatabase::SetOption(const char* key, const char* value,
return ADBC_STATUS_OK;
}
+AdbcStatusCode PostgresDatabase::SetOptionBytes(const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ SetError(error, "%s%s", "[libpq] Unknown option ", key);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode PostgresDatabase::SetOptionDouble(const char* key, double value,
+ struct AdbcError* error) {
+ SetError(error, "%s%s", "[libpq] Unknown option ", key);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode PostgresDatabase::SetOptionInt(const char* key, int64_t value,
+ struct AdbcError* error) {
+ SetError(error, "%s%s", "[libpq] Unknown option ", key);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
AdbcStatusCode PostgresDatabase::Connect(PGconn** conn, struct AdbcError* error) {
if (uri_.empty()) {
SetError(error, "%s",
diff --git a/c/driver/postgresql/database.h b/c/driver/postgresql/database.h
index f1046478..9269b195 100644
--- a/c/driver/postgresql/database.h
+++ b/c/driver/postgresql/database.h
@@ -36,7 +36,19 @@ class PostgresDatabase {
AdbcStatusCode Init(struct AdbcError* error);
AdbcStatusCode Release(struct AdbcError* error);
+ AdbcStatusCode GetOption(const char* option, char* value, size_t* length,
+ struct AdbcError* error);
+ AdbcStatusCode GetOptionBytes(const char* option, uint8_t* value, size_t* length,
+ struct AdbcError* error);
+ AdbcStatusCode GetOptionDouble(const char* option, double* value,
+ struct AdbcError* error);
+ AdbcStatusCode GetOptionInt(const char* option, int64_t* value,
+ struct AdbcError* error);
AdbcStatusCode SetOption(const char* key, const char* value, struct AdbcError* error);
+ AdbcStatusCode SetOptionBytes(const char* key, const uint8_t* value, size_t length,
+ struct AdbcError* error);
+ AdbcStatusCode SetOptionDouble(const char* key, double value, struct AdbcError* error);
+ AdbcStatusCode SetOptionInt(const char* key, int64_t value, struct AdbcError* error);
// Internal implementation
diff --git a/c/driver/postgresql/postgres_copy_reader.h b/c/driver/postgresql/postgres_copy_reader.h
index 4aa5a82e..5c7214dc 100644
--- a/c/driver/postgresql/postgres_copy_reader.h
+++ b/c/driver/postgresql/postgres_copy_reader.h
@@ -893,12 +893,13 @@ static inline ArrowErrorCode MakeCopyFieldReader(const PostgresType& pg_type,
class PostgresCopyStreamReader {
public:
- ArrowErrorCode Init(const PostgresType& pg_type) {
+ ArrowErrorCode Init(PostgresType pg_type) {
if (pg_type.type_id() != PostgresTypeId::kRecord) {
return EINVAL;
}
- root_reader_.Init(pg_type);
+ pg_type_ = std::move(pg_type);
+ root_reader_.Init(pg_type_);
array_size_approx_bytes_ = 0;
return NANOARROW_OK;
}
@@ -1022,7 +1023,10 @@ class PostgresCopyStreamReader {
return NANOARROW_OK;
}
+ const PostgresType& pg_type() const { return pg_type_; }
+
private:
+ PostgresType pg_type_;
PostgresCopyFieldTupleReader root_reader_;
nanoarrow::UniqueSchema schema_;
nanoarrow::UniqueArray array_;
diff --git a/c/driver/postgresql/postgresql.cc b/c/driver/postgresql/postgresql.cc
index 95e6c8b8..9f6730eb 100644
--- a/c/driver/postgresql/postgresql.cc
+++ b/c/driver/postgresql/postgresql.cc
@@ -83,14 +83,92 @@ AdbcStatusCode PostgresDatabaseRelease(struct AdbcDatabase* database,
return status;
}
+AdbcStatusCode PostgresDatabaseGetOption(struct AdbcDatabase* database, const char* key,
+ char* value, size_t* length,
+ struct AdbcError* error) {
+ if (!database->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr = reinterpret_cast<std::shared_ptr<PostgresDatabase>*>(database->private_data);
+ return (*ptr)->GetOption(key, value, length, error);
+}
+
+AdbcStatusCode PostgresDatabaseGetOptionBytes(struct AdbcDatabase* database,
+ const char* key, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ if (!database->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr = reinterpret_cast<std::shared_ptr<PostgresDatabase>*>(database->private_data);
+ return (*ptr)->GetOptionBytes(key, value, length, error);
+}
+
+AdbcStatusCode PostgresDatabaseGetOptionDouble(struct AdbcDatabase* database,
+ const char* key, double* value,
+ struct AdbcError* error) {
+ if (!database->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr = reinterpret_cast<std::shared_ptr<PostgresDatabase>*>(database->private_data);
+ return (*ptr)->GetOptionDouble(key, value, error);
+}
+
+AdbcStatusCode PostgresDatabaseGetOptionInt(struct AdbcDatabase* database,
+ const char* key, int64_t* value,
+ struct AdbcError* error) {
+ if (!database->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr = reinterpret_cast<std::shared_ptr<PostgresDatabase>*>(database->private_data);
+ return (*ptr)->GetOptionInt(key, value, error);
+}
+
AdbcStatusCode PostgresDatabaseSetOption(struct AdbcDatabase* database, const char* key,
const char* value, struct AdbcError* error) {
if (!database || !database->private_data) return ADBC_STATUS_INVALID_STATE;
auto ptr = reinterpret_cast<std::shared_ptr<PostgresDatabase>*>(database->private_data);
return (*ptr)->SetOption(key, value, error);
}
+
+AdbcStatusCode PostgresDatabaseSetOptionBytes(struct AdbcDatabase* database,
+ const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ if (!database->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr = reinterpret_cast<std::shared_ptr<PostgresDatabase>*>(database->private_data);
+ return (*ptr)->SetOptionBytes(key, value, length, error);
+}
+
+AdbcStatusCode PostgresDatabaseSetOptionDouble(struct AdbcDatabase* database,
+ const char* key, double value,
+ struct AdbcError* error) {
+ if (!database->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr = reinterpret_cast<std::shared_ptr<PostgresDatabase>*>(database->private_data);
+ return (*ptr)->SetOptionDouble(key, value, error);
+}
+
+AdbcStatusCode PostgresDatabaseSetOptionInt(struct AdbcDatabase* database,
+ const char* key, int64_t value,
+ struct AdbcError* error) {
+ if (!database->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr = reinterpret_cast<std::shared_ptr<PostgresDatabase>*>(database->private_data);
+ return (*ptr)->SetOptionInt(key, value, error);
+}
} // namespace
+AdbcStatusCode AdbcDatabaseGetOption(struct AdbcDatabase* database, const char* key,
+ char* value, size_t* length,
+ struct AdbcError* error) {
+ return PostgresDatabaseGetOption(database, key, value, length, error);
+}
+
+AdbcStatusCode AdbcDatabaseGetOptionBytes(struct AdbcDatabase* database, const char* key,
+ uint8_t* value, size_t* length,
+ struct AdbcError* error) {
+ return PostgresDatabaseGetOptionBytes(database, key, value, length, error);
+}
+
+AdbcStatusCode AdbcDatabaseGetOptionInt(struct AdbcDatabase* database, const char* key,
+ int64_t* value, struct AdbcError* error) {
+ return PostgresDatabaseGetOptionInt(database, key, value, error);
+}
+
+AdbcStatusCode AdbcDatabaseGetOptionDouble(struct AdbcDatabase* database, const char* key,
+ double* value, struct AdbcError* error) {
+ return PostgresDatabaseGetOptionDouble(database, key, value, error);
+}
+
AdbcStatusCode AdbcDatabaseInit(struct AdbcDatabase* database, struct AdbcError* error) {
return PostgresDatabaseInit(database, error);
}
@@ -109,10 +187,34 @@ AdbcStatusCode AdbcDatabaseSetOption(struct AdbcDatabase* database, const char*
return PostgresDatabaseSetOption(database, key, value, error);
}
+AdbcStatusCode AdbcDatabaseSetOptionBytes(struct AdbcDatabase* database, const char* key,
+ const uint8_t* value, size_t length,
+ struct AdbcError* error) {
+ return PostgresDatabaseSetOptionBytes(database, key, value, length, error);
+}
+
+AdbcStatusCode AdbcDatabaseSetOptionInt(struct AdbcDatabase* database, const char* key,
+ int64_t value, struct AdbcError* error) {
+ return PostgresDatabaseSetOptionInt(database, key, value, error);
+}
+
+AdbcStatusCode AdbcDatabaseSetOptionDouble(struct AdbcDatabase* database, const char* key,
+ double value, struct AdbcError* error) {
+ return PostgresDatabaseSetOptionDouble(database, key, value, error);
+}
+
// ---------------------------------------------------------------------
// AdbcConnection
namespace {
+AdbcStatusCode PostgresConnectionCancel(struct AdbcConnection* connection,
+ struct AdbcError* error) {
+ if (!connection->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresConnection>*>(connection->private_data);
+ return (*ptr)->Cancel(error);
+}
+
AdbcStatusCode PostgresConnectionCommit(struct AdbcConnection* connection,
struct AdbcError* error) {
if (!connection->private_data) return ADBC_STATUS_INVALID_STATE;
@@ -142,6 +244,42 @@ AdbcStatusCode PostgresConnectionGetObjects(
table_types, column_name, stream, error);
}
+AdbcStatusCode PostgresConnectionGetOption(struct AdbcConnection* connection,
+ const char* key, char* value, size_t* length,
+ struct AdbcError* error) {
+ if (!connection->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresConnection>*>(connection->private_data);
+ return (*ptr)->GetOption(key, value, length, error);
+}
+
+AdbcStatusCode PostgresConnectionGetOptionBytes(struct AdbcConnection* connection,
+ const char* key, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ if (!connection->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresConnection>*>(connection->private_data);
+ return (*ptr)->GetOptionBytes(key, value, length, error);
+}
+
+AdbcStatusCode PostgresConnectionGetOptionDouble(struct AdbcConnection* connection,
+ const char* key, double* value,
+ struct AdbcError* error) {
+ if (!connection->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresConnection>*>(connection->private_data);
+ return (*ptr)->GetOptionDouble(key, value, error);
+}
+
+AdbcStatusCode PostgresConnectionGetOptionInt(struct AdbcConnection* connection,
+ const char* key, int64_t* value,
+ struct AdbcError* error) {
+ if (!connection->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresConnection>*>(connection->private_data);
+ return (*ptr)->GetOptionInt(key, value, error);
+}
+
AdbcStatusCode PostgresConnectionGetTableSchema(
struct AdbcConnection* connection, const char* catalog, const char* db_schema,
const char* table_name, struct ArrowSchema* schema, struct AdbcError* error) {
@@ -213,7 +351,40 @@ AdbcStatusCode PostgresConnectionSetOption(struct AdbcConnection* connection,
return (*ptr)->SetOption(key, value, error);
}
+AdbcStatusCode PostgresConnectionSetOptionBytes(struct AdbcConnection* connection,
+ const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ if (!connection->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresConnection>*>(connection->private_data);
+ return (*ptr)->SetOptionBytes(key, value, length, error);
+}
+
+AdbcStatusCode PostgresConnectionSetOptionDouble(struct AdbcConnection* connection,
+ const char* key, double value,
+ struct AdbcError* error) {
+ if (!connection->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresConnection>*>(connection->private_data);
+ return (*ptr)->SetOptionDouble(key, value, error);
+}
+
+AdbcStatusCode PostgresConnectionSetOptionInt(struct AdbcConnection* connection,
+ const char* key, int64_t value,
+ struct AdbcError* error) {
+ if (!connection->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresConnection>*>(connection->private_data);
+ return (*ptr)->SetOptionInt(key, value, error);
+}
+
} // namespace
+
+AdbcStatusCode AdbcConnectionCancel(struct AdbcConnection* connection,
+ struct AdbcError* error) {
+ return PostgresConnectionCancel(connection, error);
+}
+
AdbcStatusCode AdbcConnectionCommit(struct AdbcConnection* connection,
struct AdbcError* error) {
return PostgresConnectionCommit(connection, error);
@@ -237,6 +408,30 @@ AdbcStatusCode AdbcConnectionGetObjects(struct AdbcConnection* connection, int d
table_types, column_name, stream, error);
}
+AdbcStatusCode AdbcConnectionGetOption(struct AdbcConnection* connection, const char* key,
+ char* value, size_t* length,
+ struct AdbcError* error) {
+ return PostgresConnectionGetOption(connection, key, value, length, error);
+}
+
+AdbcStatusCode AdbcConnectionGetOptionBytes(struct AdbcConnection* connection,
+ const char* key, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ return PostgresConnectionGetOptionBytes(connection, key, value, length, error);
+}
+
+AdbcStatusCode AdbcConnectionGetOptionInt(struct AdbcConnection* connection,
+ const char* key, int64_t* value,
+ struct AdbcError* error) {
+ return PostgresConnectionGetOptionInt(connection, key, value, error);
+}
+
+AdbcStatusCode AdbcConnectionGetOptionDouble(struct AdbcConnection* connection,
+ const char* key, double* value,
+ struct AdbcError* error) {
+ return PostgresConnectionGetOptionDouble(connection, key, value, error);
+}
+
AdbcStatusCode AdbcConnectionGetTableSchema(struct AdbcConnection* connection,
const char* catalog, const char* db_schema,
const char* table_name,
@@ -287,6 +482,24 @@ AdbcStatusCode AdbcConnectionSetOption(struct AdbcConnection* connection, const
return PostgresConnectionSetOption(connection, key, value, error);
}
+AdbcStatusCode AdbcConnectionSetOptionBytes(struct AdbcConnection* connection,
+ const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ return PostgresConnectionSetOptionBytes(connection, key, value, length, error);
+}
+
+AdbcStatusCode AdbcConnectionSetOptionInt(struct AdbcConnection* connection,
+ const char* key, int64_t value,
+ struct AdbcError* error) {
+ return PostgresConnectionSetOptionInt(connection, key, value, error);
+}
+
+AdbcStatusCode AdbcConnectionSetOptionDouble(struct AdbcConnection* connection,
+ const char* key, double value,
+ struct AdbcError* error) {
+ return PostgresConnectionSetOptionDouble(connection, key, value, error);
+}
+
// ---------------------------------------------------------------------
// AdbcStatement
@@ -310,6 +523,14 @@ AdbcStatusCode PostgresStatementBindStream(struct AdbcStatement* statement,
return (*ptr)->Bind(stream, error);
}
+AdbcStatusCode PostgresStatementCancel(struct AdbcStatement* statement,
+ struct AdbcError* error) {
+ if (!statement->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto* ptr =
+ reinterpret_cast<std::shared_ptr<PostgresStatement>*>(statement->private_data);
+ return (*ptr)->Cancel(error);
+}
+
AdbcStatusCode PostgresStatementExecutePartitions(struct AdbcStatement* statement,
struct ArrowSchema* schema,
struct AdbcPartitions* partitions,
@@ -329,16 +550,49 @@ AdbcStatusCode PostgresStatementExecuteQuery(struct AdbcStatement* statement,
return (*ptr)->ExecuteQuery(output, rows_affected, error);
}
-AdbcStatusCode PostgresStatementGetPartitionDesc(struct AdbcStatement* statement,
- uint8_t* partition_desc,
- struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+AdbcStatusCode PostgresStatementExecuteSchema(struct AdbcStatement* statement,
+ struct ArrowSchema* schema,
+ struct AdbcError* error) {
+ if (!statement->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto* ptr =
+ reinterpret_cast<std::shared_ptr<PostgresStatement>*>(statement->private_data);
+ return (*ptr)->ExecuteSchema(schema, error);
}
-AdbcStatusCode PostgresStatementGetPartitionDescSize(struct AdbcStatement* statement,
- size_t* length,
- struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+AdbcStatusCode PostgresStatementGetOption(struct AdbcStatement* statement,
+ const char* key, char* value, size_t* length,
+ struct AdbcError* error) {
+ if (!statement->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresStatement>*>(statement->private_data);
+ return (*ptr)->GetOption(key, value, length, error);
+}
+
+AdbcStatusCode PostgresStatementGetOptionBytes(struct AdbcStatement* statement,
+ const char* key, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ if (!statement->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresStatement>*>(statement->private_data);
+ return (*ptr)->GetOptionBytes(key, value, length, error);
+}
+
+AdbcStatusCode PostgresStatementGetOptionDouble(struct AdbcStatement* statement,
+ const char* key, double* value,
+ struct AdbcError* error) {
+ if (!statement->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresStatement>*>(statement->private_data);
+ return (*ptr)->GetOptionDouble(key, value, error);
+}
+
+AdbcStatusCode PostgresStatementGetOptionInt(struct AdbcStatement* statement,
+ const char* key, int64_t* value,
+ struct AdbcError* error) {
+ if (!statement->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresStatement>*>(statement->private_data);
+ return (*ptr)->GetOptionInt(key, value, error);
}
AdbcStatusCode PostgresStatementGetParameterSchema(struct AdbcStatement* statement,
@@ -386,6 +640,33 @@ AdbcStatusCode PostgresStatementSetOption(struct AdbcStatement* statement,
return (*ptr)->SetOption(key, value, error);
}
+AdbcStatusCode PostgresStatementSetOptionBytes(struct AdbcStatement* statement,
+ const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ if (!statement->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresStatement>*>(statement->private_data);
+ return (*ptr)->SetOptionBytes(key, value, length, error);
+}
+
+AdbcStatusCode PostgresStatementSetOptionDouble(struct AdbcStatement* statement,
+ const char* key, double value,
+ struct AdbcError* error) {
+ if (!statement->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresStatement>*>(statement->private_data);
+ return (*ptr)->SetOptionDouble(key, value, error);
+}
+
+AdbcStatusCode PostgresStatementSetOptionInt(struct AdbcStatement* statement,
+ const char* key, int64_t value,
+ struct AdbcError* error) {
+ if (!statement->private_data) return ADBC_STATUS_INVALID_STATE;
+ auto ptr =
+ reinterpret_cast<std::shared_ptr<PostgresStatement>*>(statement->private_data);
+ return (*ptr)->SetOptionInt(key, value, error);
+}
+
AdbcStatusCode PostgresStatementSetSqlQuery(struct AdbcStatement* statement,
const char* query, struct AdbcError* error) {
if (!statement->private_data) return ADBC_STATUS_INVALID_STATE;
@@ -407,6 +688,11 @@ AdbcStatusCode AdbcStatementBindStream(struct AdbcStatement* statement,
return PostgresStatementBindStream(statement, stream, error);
}
+AdbcStatusCode AdbcStatementCancel(struct AdbcStatement* statement,
+ struct AdbcError* error) {
+ return PostgresStatementCancel(statement, error);
+}
+
AdbcStatusCode AdbcStatementExecutePartitions(struct AdbcStatement* statement,
ArrowSchema* schema,
struct AdbcPartitions* partitions,
@@ -423,16 +709,32 @@ AdbcStatusCode AdbcStatementExecuteQuery(struct AdbcStatement* statement,
return PostgresStatementExecuteQuery(statement, output, rows_affected, error);
}
-AdbcStatusCode AdbcStatementGetPartitionDesc(struct AdbcStatement* statement,
- uint8_t* partition_desc,
- struct AdbcError* error) {
- return PostgresStatementGetPartitionDesc(statement, partition_desc, error);
+AdbcStatusCode AdbcStatementExecuteSchema(struct AdbcStatement* statement,
+ ArrowSchema* schema, struct AdbcError* error) {
+ return PostgresStatementExecuteSchema(statement, schema, error);
}
-AdbcStatusCode AdbcStatementGetPartitionDescSize(struct AdbcStatement* statement,
- size_t* length,
- struct AdbcError* error) {
- return PostgresStatementGetPartitionDescSize(statement, length, error);
+AdbcStatusCode AdbcStatementGetOption(struct AdbcStatement* statement, const char* key,
+ char* value, size_t* length,
+ struct AdbcError* error) {
+ return PostgresStatementGetOption(statement, key, value, length, error);
+}
+
+AdbcStatusCode AdbcStatementGetOptionBytes(struct AdbcStatement* statement,
+ const char* key, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ return PostgresStatementGetOptionBytes(statement, key, value, length, error);
+}
+
+AdbcStatusCode AdbcStatementGetOptionInt(struct AdbcStatement* statement, const char* key,
+ int64_t* value, struct AdbcError* error) {
+ return PostgresStatementGetOptionInt(statement, key, value, error);
+}
+
+AdbcStatusCode AdbcStatementGetOptionDouble(struct AdbcStatement* statement,
+ const char* key, double* value,
+ struct AdbcError* error) {
+ return PostgresStatementGetOptionDouble(statement, key, value, error);
}
AdbcStatusCode AdbcStatementGetParameterSchema(struct AdbcStatement* statement,
@@ -462,6 +764,23 @@ AdbcStatusCode AdbcStatementSetOption(struct AdbcStatement* statement, const cha
return PostgresStatementSetOption(statement, key, value, error);
}
+AdbcStatusCode AdbcStatementSetOptionBytes(struct AdbcStatement* statement,
+ const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ return PostgresStatementSetOptionBytes(statement, key, value, length, error);
+}
+
+AdbcStatusCode AdbcStatementSetOptionInt(struct AdbcStatement* statement, const char* key,
+ int64_t value, struct AdbcError* error) {
+ return PostgresStatementSetOptionInt(statement, key, value, error);
+}
+
+AdbcStatusCode AdbcStatementSetOptionDouble(struct AdbcStatement* statement,
+ const char* key, double value,
+ struct AdbcError* error) {
+ return PostgresStatementSetOptionDouble(statement, key, value, error);
+}
+
AdbcStatusCode AdbcStatementSetSqlQuery(struct AdbcStatement* statement,
const char* query, struct AdbcError* error) {
return PostgresStatementSetSqlQuery(statement, query, error);
@@ -469,12 +788,47 @@ AdbcStatusCode AdbcStatementSetSqlQuery(struct AdbcStatement* statement,
extern "C" {
ADBC_EXPORT
-AdbcStatusCode AdbcDriverInit(int version, void* raw_driver, struct AdbcError* error) {
- if (version != ADBC_VERSION_1_0_0) return ADBC_STATUS_NOT_IMPLEMENTED;
+AdbcStatusCode PostgresqlDriverInit(int version, void* raw_driver,
+ struct AdbcError* error) {
+ if (version != ADBC_VERSION_1_0_0 && version != ADBC_VERSION_1_1_0) {
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+ }
if (!raw_driver) return ADBC_STATUS_INVALID_ARGUMENT;
auto* driver = reinterpret_cast<struct AdbcDriver*>(raw_driver);
- std::memset(driver, 0, ADBC_DRIVER_1_0_0_SIZE);
+ if (version >= ADBC_VERSION_1_1_0) {
+ std::memset(driver, 0, ADBC_DRIVER_1_1_0_SIZE);
+
+ driver->DatabaseGetOption = PostgresDatabaseGetOption;
+ driver->DatabaseGetOptionBytes = PostgresDatabaseGetOptionBytes;
+ driver->DatabaseGetOptionDouble = PostgresDatabaseGetOptionDouble;
+ driver->DatabaseGetOptionInt = PostgresDatabaseGetOptionInt;
+ driver->DatabaseSetOptionBytes = PostgresDatabaseSetOptionBytes;
+ driver->DatabaseSetOptionDouble = PostgresDatabaseSetOptionDouble;
+ driver->DatabaseSetOptionInt = PostgresDatabaseSetOptionInt;
+
+ driver->ConnectionCancel = PostgresConnectionCancel;
+ driver->ConnectionGetOption = PostgresConnectionGetOption;
+ driver->ConnectionGetOptionBytes = PostgresConnectionGetOptionBytes;
+ driver->ConnectionGetOptionDouble = PostgresConnectionGetOptionDouble;
+ driver->ConnectionGetOptionInt = PostgresConnectionGetOptionInt;
+ driver->ConnectionSetOptionBytes = PostgresConnectionSetOptionBytes;
+ driver->ConnectionSetOptionDouble = PostgresConnectionSetOptionDouble;
+ driver->ConnectionSetOptionInt = PostgresConnectionSetOptionInt;
+
+ driver->StatementCancel = PostgresStatementCancel;
+ driver->StatementExecuteSchema = PostgresStatementExecuteSchema;
+ driver->StatementGetOption = PostgresStatementGetOption;
+ driver->StatementGetOptionBytes = PostgresStatementGetOptionBytes;
+ driver->StatementGetOptionDouble = PostgresStatementGetOptionDouble;
+ driver->StatementGetOptionInt = PostgresStatementGetOptionInt;
+ driver->StatementSetOptionBytes = PostgresStatementSetOptionBytes;
+ driver->StatementSetOptionDouble = PostgresStatementSetOptionDouble;
+ driver->StatementSetOptionInt = PostgresStatementSetOptionInt;
+ } else {
+ std::memset(driver, 0, ADBC_DRIVER_1_0_0_SIZE);
+ }
+
driver->DatabaseInit = PostgresDatabaseInit;
driver->DatabaseNew = PostgresDatabaseNew;
driver->DatabaseRelease = PostgresDatabaseRelease;
@@ -502,6 +856,12 @@ AdbcStatusCode AdbcDriverInit(int version, void* raw_driver, struct AdbcError* e
driver->StatementRelease = PostgresStatementRelease;
driver->StatementSetOption = PostgresStatementSetOption;
driver->StatementSetSqlQuery = PostgresStatementSetSqlQuery;
+
return ADBC_STATUS_OK;
}
+
+ADBC_EXPORT
+AdbcStatusCode AdbcDriverInit(int version, void* raw_driver, struct AdbcError* error) {
+ return PostgresqlDriverInit(version, raw_driver, error);
+}
}
diff --git a/c/driver/postgresql/postgresql_test.cc b/c/driver/postgresql/postgresql_test.cc
index a826e172..a44721a0 100644
--- a/c/driver/postgresql/postgresql_test.cc
+++ b/c/driver/postgresql/postgresql_test.cc
@@ -103,6 +103,29 @@ class PostgresQuirks : public adbc_validation::DriverQuirks {
std::string catalog() const override { return "postgres"; }
std::string db_schema() const override { return "public"; }
+
+ bool supports_cancel() const override { return true; }
+ bool supports_execute_schema() const override { return true; }
+ std::optional<adbc_validation::SqlInfoValue> supports_get_sql_info(
+ uint32_t info_code) const override {
+ switch (info_code) {
+ case ADBC_INFO_DRIVER_ADBC_VERSION:
+ return ADBC_VERSION_1_1_0;
+ case ADBC_INFO_DRIVER_NAME:
+ return "ADBC PostgreSQL Driver";
+ case ADBC_INFO_DRIVER_VERSION:
+ return "(unknown)";
+ case ADBC_INFO_VENDOR_NAME:
+ return "PostgreSQL";
+ case ADBC_INFO_VENDOR_VERSION:
+ // Strings are checked via substring match
+ return "15";
+ default:
+ return std::nullopt;
+ }
+ }
+ bool supports_metadata_current_catalog() const override { return true; }
+ bool supports_metadata_current_db_schema() const override { return true; }
};
class PostgresDatabaseTest : public ::testing::Test,
@@ -134,10 +157,8 @@ TEST_F(PostgresConnectionTest, GetInfoMetadata) {
adbc_validation::StreamReader reader;
std::vector<uint32_t> info = {
- ADBC_INFO_DRIVER_NAME,
- ADBC_INFO_DRIVER_VERSION,
- ADBC_INFO_VENDOR_NAME,
- ADBC_INFO_VENDOR_VERSION,
+ ADBC_INFO_DRIVER_NAME, ADBC_INFO_DRIVER_VERSION, ADBC_INFO_DRIVER_ADBC_VERSION,
+ ADBC_INFO_VENDOR_NAME, ADBC_INFO_VENDOR_VERSION,
};
ASSERT_THAT(AdbcConnectionGetInfo(&connection, info.data(), info.size(),
&reader.stream.value, &error),
@@ -153,29 +174,30 @@ TEST_F(PostgresConnectionTest, GetInfoMetadata) {
ASSERT_FALSE(ArrowArrayViewIsNull(reader.array_view->children[0], row));
const uint32_t code =
reader.array_view->children[0]->buffer_views[1].data.as_uint32[row];
+ const uint32_t offset =
+ reader.array_view->children[1]->buffer_views[1].data.as_int32[row];
seen.push_back(code);
- int str_child_index = 0;
- struct ArrowArrayView* str_child =
- reader.array_view->children[1]->children[str_child_index];
+ struct ArrowArrayView* str_child = reader.array_view->children[1]->children[0];
+ struct ArrowArrayView* int_child = reader.array_view->children[1]->children[2];
switch (code) {
case ADBC_INFO_DRIVER_NAME: {
- ArrowStringView val = ArrowArrayViewGetStringUnsafe(str_child, 0);
+ ArrowStringView val = ArrowArrayViewGetStringUnsafe(str_child, offset);
EXPECT_EQ("ADBC PostgreSQL Driver", std::string(val.data, val.size_bytes));
break;
}
case ADBC_INFO_DRIVER_VERSION: {
- ArrowStringView val = ArrowArrayViewGetStringUnsafe(str_child, 1);
+ ArrowStringView val = ArrowArrayViewGetStringUnsafe(str_child, offset);
EXPECT_EQ("(unknown)", std::string(val.data, val.size_bytes));
break;
}
case ADBC_INFO_VENDOR_NAME: {
- ArrowStringView val = ArrowArrayViewGetStringUnsafe(str_child, 2);
+ ArrowStringView val = ArrowArrayViewGetStringUnsafe(str_child, offset);
EXPECT_EQ("PostgreSQL", std::string(val.data, val.size_bytes));
break;
}
case ADBC_INFO_VENDOR_VERSION: {
- ArrowStringView val = ArrowArrayViewGetStringUnsafe(str_child, 3);
+ ArrowStringView val = ArrowArrayViewGetStringUnsafe(str_child, offset);
#ifdef __WIN32
const char* pater = "\\d\\d\\d\\d\\d\\d";
#else
@@ -185,6 +207,10 @@ TEST_F(PostgresConnectionTest, GetInfoMetadata) {
::testing::MatchesRegex(pater));
break;
}
+ case ADBC_INFO_DRIVER_ADBC_VERSION: {
+ EXPECT_EQ(ADBC_VERSION_1_1_0, ArrowArrayViewGetIntUnsafe(int_child, offset));
+ break;
+ }
default:
// Ignored
break;
@@ -198,10 +224,6 @@ TEST_F(PostgresConnectionTest, GetObjectsGetCatalogs) {
ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error), IsOkStatus(&error));
- if (!quirks()->supports_get_objects()) {
- GTEST_SKIP();
- }
-
adbc_validation::StreamReader reader;
ASSERT_THAT(
AdbcConnectionGetObjects(&connection, ADBC_OBJECT_DEPTH_CATALOGS, nullptr, nullptr,
@@ -228,10 +250,6 @@ TEST_F(PostgresConnectionTest, GetObjectsGetDbSchemas) {
ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error), IsOkStatus(&error));
- if (!quirks()->supports_get_objects()) {
- GTEST_SKIP();
- }
-
adbc_validation::StreamReader reader;
ASSERT_THAT(AdbcConnectionGetObjects(&connection, ADBC_OBJECT_DEPTH_DB_SCHEMAS, nullptr,
nullptr, nullptr, nullptr, nullptr,
@@ -255,10 +273,6 @@ TEST_F(PostgresConnectionTest, GetObjectsGetAllFindsPrimaryKey) {
ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error), IsOkStatus(&error));
- if (!quirks()->supports_get_objects()) {
- GTEST_SKIP();
- }
-
ASSERT_THAT(quirks()->DropTable(&connection, "adbc_pkey_test", &error),
IsOkStatus(&error));
@@ -329,10 +343,6 @@ TEST_F(PostgresConnectionTest, GetObjectsGetAllFindsForeignKey) {
ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error), IsOkStatus(&error));
- if (!quirks()->supports_get_objects()) {
- GTEST_SKIP();
- }
-
ASSERT_THAT(quirks()->DropTable(&connection, "adbc_fkey_test", &error),
IsOkStatus(&error));
ASSERT_THAT(quirks()->DropTable(&connection, "adbc_fkey_test_base", &error),
@@ -450,10 +460,6 @@ TEST_F(PostgresConnectionTest, GetObjectsTableTypesFilter) {
ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error), IsOkStatus(&error));
- if (!quirks()->supports_get_objects()) {
- GTEST_SKIP();
- }
-
ASSERT_THAT(quirks()->DropView(&connection, "adbc_table_types_view_test", &error),
IsOkStatus(&error));
ASSERT_THAT(quirks()->DropTable(&connection, "adbc_table_types_table_test", &error),
@@ -516,7 +522,7 @@ TEST_F(PostgresConnectionTest, GetObjectsTableTypesFilter) {
}
TEST_F(PostgresConnectionTest, MetadataGetTableSchemaInjection) {
- if (!quirks()->supports_bulk_ingest()) {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE)) {
GTEST_SKIP();
}
ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
@@ -549,6 +555,57 @@ TEST_F(PostgresConnectionTest, MetadataGetTableSchemaInjection) {
{"strings", NANOARROW_TYPE_STRING, true}}));
}
+TEST_F(PostgresConnectionTest, MetadataSetCurrentDbSchema) {
+ ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error), IsOkStatus(&error));
+
+ {
+ adbc_validation::Handle<struct AdbcStatement> statement;
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement.value, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(
+ &statement.value, "CREATE SCHEMA IF NOT EXISTS testschema", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement.value, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(
+ AdbcStatementSetSqlQuery(
+ &statement.value,
+ "CREATE TABLE IF NOT EXISTS testschema.schematable (ints INT)", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement.value, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementRelease(&statement.value, &error), IsOkStatus(&error));
+ }
+
+ adbc_validation::Handle<struct AdbcStatement> statement;
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement.value, &error),
+ IsOkStatus(&error));
+
+ // Table does not exist in this schema
+ ASSERT_THAT(
+ AdbcStatementSetSqlQuery(&statement.value, "SELECT * FROM schematable", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement.value, nullptr, nullptr, &error),
+ IsStatus(ADBC_STATUS_IO, &error));
+
+ ASSERT_THAT(
+ AdbcConnectionSetOption(&connection, ADBC_CONNECTION_OPTION_CURRENT_DB_SCHEMA,
+ "testschema", &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(
+ AdbcStatementSetSqlQuery(&statement.value, "SELECT * FROM schematable", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement.value, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementRelease(&statement.value, &error), IsOkStatus(&error));
+}
+
ADBCV_TEST_CONNECTION(PostgresConnectionTest)
class PostgresStatementTest : public ::testing::Test,
diff --git a/c/driver/postgresql/statement.cc b/c/driver/postgresql/statement.cc
index d8c474bd..6521fce9 100644
--- a/c/driver/postgresql/statement.cc
+++ b/c/driver/postgresql/statement.cc
@@ -780,11 +780,41 @@ AdbcStatusCode PostgresStatement::Bind(struct ArrowArrayStream* stream,
return ADBC_STATUS_OK;
}
+AdbcStatusCode PostgresStatement::Cancel(struct AdbcError* error) {
+ // Ultimately the same underlying PGconn
+ return connection_->Cancel(error);
+}
+
AdbcStatusCode PostgresStatement::CreateBulkTable(
const struct ArrowSchema& source_schema,
const std::vector<struct ArrowSchemaView>& source_schema_fields,
struct AdbcError* error) {
std::string create = "CREATE TABLE ";
+ switch (ingest_.mode) {
+ case IngestMode::kCreate:
+ // Nothing to do
+ break;
+ case IngestMode::kAppend:
+ return ADBC_STATUS_OK;
+ case IngestMode::kReplace: {
+ std::string drop = "DROP TABLE IF EXISTS " + ingest_.target;
+ PGresult* result = PQexecParams(connection_->conn(), drop.c_str(), /*nParams=*/0,
+ /*paramTypes=*/nullptr, /*paramValues=*/nullptr,
+ /*paramLengths=*/nullptr, /*paramFormats=*/nullptr,
+ /*resultFormat=*/1 /*(binary)*/);
+ if (PQresultStatus(result) != PGRES_COMMAND_OK) {
+ SetError(error, "[libpq] Failed to drop table: %s\nQuery was: %s",
+ PQerrorMessage(connection_->conn()), drop.c_str());
+ PQclear(result);
+ return ADBC_STATUS_IO;
+ }
+ PQclear(result);
+ break;
+ }
+ case IngestMode::kCreateAppend:
+ create += "IF NOT EXISTS ";
+ break;
+ }
create += ingest_.target;
create += " (";
@@ -907,50 +937,12 @@ AdbcStatusCode PostgresStatement::ExecuteQuery(struct ArrowArrayStream* stream,
// 1. Prepare the query to get the schema
{
- // TODO: we should pipeline here and assume this will succeed
- PGresult* result = PQprepare(connection_->conn(), /*stmtName=*/"", query_.c_str(),
- /*nParams=*/0, nullptr);
- if (PQresultStatus(result) != PGRES_COMMAND_OK) {
- SetError(error,
- "[libpq] Failed to execute query: could not infer schema: failed to "
- "prepare query: %s\nQuery was:%s",
- PQerrorMessage(connection_->conn()), query_.c_str());
- PQclear(result);
- return ADBC_STATUS_IO;
- }
- PQclear(result);
- result = PQdescribePrepared(connection_->conn(), /*stmtName=*/"");
- if (PQresultStatus(result) != PGRES_COMMAND_OK) {
- SetError(error,
- "[libpq] Failed to execute query: could not infer schema: failed to "
- "describe prepared statement: %s\nQuery was:%s",
- PQerrorMessage(connection_->conn()), query_.c_str());
- PQclear(result);
- return ADBC_STATUS_IO;
- }
-
- // Resolve the information from the PGresult into a PostgresType
- PostgresType root_type;
- AdbcStatusCode status =
- ResolvePostgresType(*type_resolver_, result, &root_type, error);
- PQclear(result);
- if (status != ADBC_STATUS_OK) return status;
-
- // Initialize the copy reader and infer the output schema (i.e., error for
- // unsupported types before issuing the COPY query)
- reader_.copy_reader_.reset(new PostgresCopyStreamReader());
- reader_.copy_reader_->Init(root_type);
- struct ArrowError na_error;
- int na_res = reader_.copy_reader_->InferOutputSchema(&na_error);
- if (na_res != NANOARROW_OK) {
- SetError(error, "[libpq] Failed to infer output schema: %s", na_error.message);
- return na_res;
- }
+ RAISE_ADBC(SetupReader(error));
// If the caller did not request a result set or if there are no
// inferred output columns (e.g. a CREATE or UPDATE), then don't
// use COPY (which would fail anyways)
- if (!stream || root_type.n_children() == 0) {
+ if (!stream || reader_.copy_reader_->pg_type().n_children() == 0) {
RAISE_ADBC(ExecuteUpdateQuery(rows_affected, error));
if (stream) {
struct ArrowSchema schema;
@@ -964,7 +956,8 @@ AdbcStatusCode PostgresStatement::ExecuteQuery(struct ArrowArrayStream* stream,
// This resolves the reader specific to each PostgresType -> ArrowSchema
// conversion. It is unlikely that this will fail given that we have just
// inferred these conversions ourselves.
- na_res = reader_.copy_reader_->InitFieldReaders(&na_error);
+ struct ArrowError na_error;
+ int na_res = reader_.copy_reader_->InitFieldReaders(&na_error);
if (na_res != NANOARROW_OK) {
SetError(error, "[libpq] Failed to initialize field readers: %s", na_error.message);
return na_res;
@@ -993,6 +986,23 @@ AdbcStatusCode PostgresStatement::ExecuteQuery(struct ArrowArrayStream* stream,
return ADBC_STATUS_OK;
}
+AdbcStatusCode PostgresStatement::ExecuteSchema(struct ArrowSchema* schema,
+ struct AdbcError* error) {
+ ClearResult();
+ if (query_.empty()) {
+ SetError(error, "%s", "[libpq] Must SetSqlQuery before ExecuteQuery");
+ return ADBC_STATUS_INVALID_STATE;
+ } else if (bind_.release) {
+ // TODO: if we have parameters, bind them (since they can affect the output schema)
+ SetError(error, "[libpq] ExecuteSchema with parameters is not implemented");
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+ }
+
+ RAISE_ADBC(SetupReader(error));
+ CHECK_NA(INTERNAL, reader_.copy_reader_->GetSchema(schema), error);
+ return ADBC_STATUS_OK;
+}
+
AdbcStatusCode PostgresStatement::ExecuteUpdateBulk(int64_t* rows_affected,
struct AdbcError* error) {
if (!bind_.release) {
@@ -1004,12 +1014,8 @@ AdbcStatusCode PostgresStatement::ExecuteUpdateBulk(int64_t* rows_affected,
std::memset(&bind_, 0, sizeof(bind_));
RAISE_ADBC(bind_stream.Begin(
[&]() -> AdbcStatusCode {
- if (!ingest_.append) {
- // CREATE TABLE
- return CreateBulkTable(bind_stream.bind_schema.value,
- bind_stream.bind_schema_fields, error);
- }
- return ADBC_STATUS_OK;
+ return CreateBulkTable(bind_stream.bind_schema.value,
+ bind_stream.bind_schema_fields, error);
},
error));
RAISE_ADBC(bind_stream.SetParamTypes(*type_resolver_, error));
@@ -1037,7 +1043,8 @@ AdbcStatusCode PostgresStatement::ExecuteUpdateQuery(int64_t* rows_affected,
PQexecPrepared(connection_->conn(), /*stmtName=*/"", /*nParams=*/0,
/*paramValues=*/nullptr, /*paramLengths=*/nullptr,
/*paramFormats=*/nullptr, /*resultFormat=*/kPgBinaryFormat);
- if (PQresultStatus(result) != PGRES_COMMAND_OK) {
+ ExecStatusType status = PQresultStatus(result);
+ if (status != PGRES_COMMAND_OK && status != PGRES_TUPLES_OK) {
SetError(error, "[libpq] Failed to execute query: %s\nQuery was:%s",
PQerrorMessage(connection_->conn()), query_.c_str());
PQclear(result);
@@ -1048,6 +1055,64 @@ AdbcStatusCode PostgresStatement::ExecuteUpdateQuery(int64_t* rows_affected,
return ADBC_STATUS_OK;
}
+AdbcStatusCode PostgresStatement::GetOption(const char* key, char* value, size_t* length,
+ struct AdbcError* error) {
+ std::string result;
+ if (std::strcmp(key, ADBC_INGEST_OPTION_TARGET_TABLE) == 0) {
+ result = ingest_.target;
+ } else if (std::strcmp(key, ADBC_INGEST_OPTION_MODE) == 0) {
+ switch (ingest_.mode) {
+ case IngestMode::kCreate:
+ result = ADBC_INGEST_OPTION_MODE_CREATE;
+ break;
+ case IngestMode::kAppend:
+ result = ADBC_INGEST_OPTION_MODE_APPEND;
+ break;
+ case IngestMode::kReplace:
+ result = ADBC_INGEST_OPTION_MODE_REPLACE;
+ break;
+ case IngestMode::kCreateAppend:
+ result = ADBC_INGEST_OPTION_MODE_CREATE_APPEND;
+ break;
+ }
+ } else if (std::strcmp(key, ADBC_POSTGRESQL_OPTION_BATCH_SIZE_HINT_BYTES) == 0) {
+ result = std::to_string(reader_.batch_size_hint_bytes_);
+ } else {
+ SetError(error, "[libq] Unknown statement option '%s'", key);
+ return ADBC_STATUS_NOT_FOUND;
+ }
+
+ if (result.size() + 1 <= *length) {
+ std::memcpy(value, result.data(), result.size() + 1);
+ }
+ *length = static_cast<int64_t>(result.size() + 1);
+ return ADBC_STATUS_OK;
+}
+
+AdbcStatusCode PostgresStatement::GetOptionBytes(const char* key, uint8_t* value,
+ size_t* length,
+ struct AdbcError* error) {
+ SetError(error, "[libq] Unknown statement option '%s'", key);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode PostgresStatement::GetOptionDouble(const char* key, double* value,
+ struct AdbcError* error) {
+ SetError(error, "[libq] Unknown statement option '%s'", key);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode PostgresStatement::GetOptionInt(const char* key, int64_t* value,
+ struct AdbcError* error) {
+ std::string result;
+ if (std::strcmp(key, ADBC_POSTGRESQL_OPTION_BATCH_SIZE_HINT_BYTES) == 0) {
+ *value = reader_.batch_size_hint_bytes_;
+ return ADBC_STATUS_OK;
+ }
+ SetError(error, "[libq] Unknown statement option '%s'", key);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
AdbcStatusCode PostgresStatement::GetParameterSchema(struct ArrowSchema* schema,
struct AdbcError* error) {
return ADBC_STATUS_NOT_IMPLEMENTED;
@@ -1088,14 +1153,18 @@ AdbcStatusCode PostgresStatement::SetOption(const char* key, const char* value,
ingest_.target = value;
} else if (std::strcmp(key, ADBC_INGEST_OPTION_MODE) == 0) {
if (std::strcmp(value, ADBC_INGEST_OPTION_MODE_CREATE) == 0) {
- ingest_.append = false;
+ ingest_.mode = IngestMode::kCreate;
} else if (std::strcmp(value, ADBC_INGEST_OPTION_MODE_APPEND) == 0) {
- ingest_.append = true;
+ ingest_.mode = IngestMode::kAppend;
+ } else if (std::strcmp(value, ADBC_INGEST_OPTION_MODE_REPLACE) == 0) {
+ ingest_.mode = IngestMode::kReplace;
+ } else if (std::strcmp(value, ADBC_INGEST_OPTION_MODE_CREATE_APPEND) == 0) {
+ ingest_.mode = IngestMode::kCreateAppend;
} else {
SetError(error, "[libpq] Invalid value '%s' for option '%s'", value, key);
return ADBC_STATUS_INVALID_ARGUMENT;
}
- } else if (std::strcmp(value, ADBC_POSTGRESQL_OPTION_BATCH_SIZE_HINT_BYTES)) {
+ } else if (std::strcmp(key, ADBC_POSTGRESQL_OPTION_BATCH_SIZE_HINT_BYTES) == 0) {
int64_t int_value = std::atol(value);
if (int_value <= 0) {
SetError(error, "[libpq] Invalid value '%s' for option '%s'", value, key);
@@ -1110,6 +1179,76 @@ AdbcStatusCode PostgresStatement::SetOption(const char* key, const char* value,
return ADBC_STATUS_OK;
}
+AdbcStatusCode PostgresStatement::SetOptionBytes(const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ SetError(error, "%s%s", "[libpq] Unknown option ", key);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode PostgresStatement::SetOptionDouble(const char* key, double value,
+ struct AdbcError* error) {
+ SetError(error, "%s%s", "[libpq] Unknown option ", key);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode PostgresStatement::SetOptionInt(const char* key, int64_t value,
+ struct AdbcError* error) {
+ if (std::strcmp(key, ADBC_POSTGRESQL_OPTION_BATCH_SIZE_HINT_BYTES) == 0) {
+ if (value <= 0) {
+ SetError(error, "[libpq] Invalid value '%" PRIi64 "' for option '%s'", value, key);
+ return ADBC_STATUS_INVALID_ARGUMENT;
+ }
+
+ this->reader_.batch_size_hint_bytes_ = value;
+ return ADBC_STATUS_OK;
+ }
+ SetError(error, "%s%s", "[libpq] Unknown option ", key);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode PostgresStatement::SetupReader(struct AdbcError* error) {
+ // TODO: we should pipeline here and assume this will succeed
+ PGresult* result = PQprepare(connection_->conn(), /*stmtName=*/"", query_.c_str(),
+ /*nParams=*/0, nullptr);
+ if (PQresultStatus(result) != PGRES_COMMAND_OK) {
+ SetError(error,
+ "[libpq] Failed to execute query: could not infer schema: failed to "
+ "prepare query: %s\nQuery was:%s",
+ PQerrorMessage(connection_->conn()), query_.c_str());
+ PQclear(result);
+ return ADBC_STATUS_IO;
+ }
+ PQclear(result);
+ result = PQdescribePrepared(connection_->conn(), /*stmtName=*/"");
+ if (PQresultStatus(result) != PGRES_COMMAND_OK) {
+ SetError(error,
+ "[libpq] Failed to execute query: could not infer schema: failed to "
+ "describe prepared statement: %s\nQuery was:%s",
+ PQerrorMessage(connection_->conn()), query_.c_str());
+ PQclear(result);
+ return ADBC_STATUS_IO;
+ }
+
+ // Resolve the information from the PGresult into a PostgresType
+ PostgresType root_type;
+ AdbcStatusCode status = ResolvePostgresType(*type_resolver_, result, &root_type, error);
+ PQclear(result);
+ if (status != ADBC_STATUS_OK) return status;
+
+ // Initialize the copy reader and infer the output schema (i.e., error for
+ // unsupported types before issuing the COPY query)
+ reader_.copy_reader_.reset(new PostgresCopyStreamReader());
+ reader_.copy_reader_->Init(root_type);
+ struct ArrowError na_error;
+ int na_res = reader_.copy_reader_->InferOutputSchema(&na_error);
+ if (na_res != NANOARROW_OK) {
+ SetError(error, "[libpq] Failed to infer output schema: (%d) %s: %s", na_res,
+ std::strerror(na_res), na_error.message);
+ return ADBC_STATUS_INTERNAL;
+ }
+ return ADBC_STATUS_OK;
+}
+
void PostgresStatement::ClearResult() {
// TODO: we may want to synchronize here for safety
reader_.Release();
diff --git a/c/driver/postgresql/statement.h b/c/driver/postgresql/statement.h
index 62af2457..eca39971 100644
--- a/c/driver/postgresql/statement.h
+++ b/c/driver/postgresql/statement.h
@@ -100,13 +100,25 @@ class PostgresStatement {
AdbcStatusCode Bind(struct ArrowArray* values, struct ArrowSchema* schema,
struct AdbcError* error);
AdbcStatusCode Bind(struct ArrowArrayStream* stream, struct AdbcError* error);
+ AdbcStatusCode Cancel(struct AdbcError* error);
AdbcStatusCode ExecuteQuery(struct ArrowArrayStream* stream, int64_t* rows_affected,
struct AdbcError* error);
+ AdbcStatusCode ExecuteSchema(struct ArrowSchema* schema, struct AdbcError* error);
+ AdbcStatusCode GetOption(const char* key, char* value, size_t* length,
+ struct AdbcError* error);
+ AdbcStatusCode GetOptionBytes(const char* key, uint8_t* value, size_t* length,
+ struct AdbcError* error);
+ AdbcStatusCode GetOptionDouble(const char* key, double* value, struct AdbcError* error);
+ AdbcStatusCode GetOptionInt(const char* key, int64_t* value, struct AdbcError* error);
AdbcStatusCode GetParameterSchema(struct ArrowSchema* schema, struct AdbcError* error);
AdbcStatusCode New(struct AdbcConnection* connection, struct AdbcError* error);
AdbcStatusCode Prepare(struct AdbcError* error);
AdbcStatusCode Release(struct AdbcError* error);
AdbcStatusCode SetOption(const char* key, const char* value, struct AdbcError* error);
+ AdbcStatusCode SetOptionBytes(const char* key, const uint8_t* value, size_t length,
+ struct AdbcError* error);
+ AdbcStatusCode SetOptionDouble(const char* key, double value, struct AdbcError* error);
+ AdbcStatusCode SetOptionInt(const char* key, int64_t value, struct AdbcError* error);
AdbcStatusCode SetSqlQuery(const char* query, struct AdbcError* error);
// ---------------------------------------------------------------------
@@ -122,6 +134,7 @@ class PostgresStatement {
AdbcStatusCode ExecutePreparedStatement(struct ArrowArrayStream* stream,
int64_t* rows_affected,
struct AdbcError* error);
+ AdbcStatusCode SetupReader(struct AdbcError* error);
private:
std::shared_ptr<PostgresTypeResolver> type_resolver_;
@@ -133,9 +146,16 @@ class PostgresStatement {
struct ArrowArrayStream bind_;
// Bulk ingest state
+ enum class IngestMode {
+ kCreate,
+ kAppend,
+ kReplace,
+ kCreateAppend,
+ };
+
struct {
std::string target;
- bool append = false;
+ IngestMode mode = IngestMode::kCreate;
} ingest_;
TupleReader reader_;
diff --git a/c/driver/snowflake/snowflake_test.cc b/c/driver/snowflake/snowflake_test.cc
index ed2f5de0..1ca09066 100644
--- a/c/driver/snowflake/snowflake_test.cc
+++ b/c/driver/snowflake/snowflake_test.cc
@@ -106,11 +106,11 @@ class SnowflakeQuirks : public adbc_validation::DriverQuirks {
}
std::string BindParameter(int index) const override { return "?"; }
+ bool supports_bulk_ingest(const char* /*mode*/) const override { return true; }
bool supports_concurrent_statements() const override { return true; }
bool supports_transactions() const override { return true; }
bool supports_get_sql_info() const override { return false; }
bool supports_get_objects() const override { return true; }
- bool supports_bulk_ingest() const override { return true; }
bool supports_partitioned_data() const override { return false; }
bool supports_dynamic_parameter_binding() const override { return false; }
bool ddl_implicit_commit_txn() const override { return true; }
diff --git a/c/driver/sqlite/sqlite.c b/c/driver/sqlite/sqlite.c
index 51b682af..efbcd03c 100644
--- a/c/driver/sqlite/sqlite.c
+++ b/c/driver/sqlite/sqlite.c
@@ -86,6 +86,26 @@ AdbcStatusCode SqliteDatabaseSetOption(struct AdbcDatabase* database, const char
return ADBC_STATUS_NOT_IMPLEMENTED;
}
+AdbcStatusCode SqliteDatabaseSetOptionBytes(struct AdbcDatabase* database,
+ const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ CHECK_DB_INIT(database, error);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode SqliteDatabaseSetOptionDouble(struct AdbcDatabase* database,
+ const char* key, double value,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(database, error);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode SqliteDatabaseSetOptionInt(struct AdbcDatabase* database, const char* key,
+ int64_t value, struct AdbcError* error) {
+ CHECK_DB_INIT(database, error);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
int OpenDatabase(const char* maybe_uri, sqlite3** db, struct AdbcError* error) {
const char* uri = maybe_uri ? maybe_uri : kDefaultUri;
int rc = sqlite3_open_v2(uri, db,
@@ -120,6 +140,33 @@ AdbcStatusCode ExecuteQuery(struct SqliteConnection* conn, const char* query,
return ADBC_STATUS_OK;
}
+AdbcStatusCode SqliteDatabaseGetOption(struct AdbcDatabase* database, const char* key,
+ char* value, size_t* length,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(database, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode SqliteDatabaseGetOptionBytes(struct AdbcDatabase* database,
+ const char* key, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ CHECK_DB_INIT(database, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode SqliteDatabaseGetOptionDouble(struct AdbcDatabase* database,
+ const char* key, double* value,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(database, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode SqliteDatabaseGetOptionInt(struct AdbcDatabase* database, const char* key,
+ int64_t* value, struct AdbcError* error) {
+ CHECK_DB_INIT(database, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
AdbcStatusCode SqliteDatabaseInit(struct AdbcDatabase* database,
struct AdbcError* error) {
CHECK_DB_INIT(database, error);
@@ -204,6 +251,27 @@ AdbcStatusCode SqliteConnectionSetOption(struct AdbcConnection* connection,
return ADBC_STATUS_NOT_IMPLEMENTED;
}
+AdbcStatusCode SqliteConnectionSetOptionBytes(struct AdbcConnection* connection,
+ const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ CHECK_DB_INIT(connection, error);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode SqliteConnectionSetOptionDouble(struct AdbcConnection* connection,
+ const char* key, double value,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(connection, error);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode SqliteConnectionSetOptionInt(struct AdbcConnection* connection,
+ const char* key, int64_t value,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(connection, error);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
AdbcStatusCode SqliteConnectionInit(struct AdbcConnection* connection,
struct AdbcDatabase* database,
struct AdbcError* error) {
@@ -754,6 +822,34 @@ AdbcStatusCode SqliteConnectionGetObjects(struct AdbcConnection* connection, int
return BatchToArrayStream(&array, &schema, out, error);
}
+AdbcStatusCode SqliteConnectionGetOption(struct AdbcConnection* connection,
+ const char* key, char* value, size_t* length,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(connection, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode SqliteConnectionGetOptionBytes(struct AdbcConnection* connection,
+ const char* key, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ CHECK_DB_INIT(connection, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode SqliteConnectionGetOptionDouble(struct AdbcConnection* connection,
+ const char* key, double* value,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(connection, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode SqliteConnectionGetOptionInt(struct AdbcConnection* connection,
+ const char* key, int64_t* value,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(connection, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
AdbcStatusCode SqliteConnectionGetTableSchema(struct AdbcConnection* connection,
const char* catalog, const char* db_schema,
const char* table_name,
@@ -1286,6 +1382,34 @@ AdbcStatusCode SqliteStatementBindStream(struct AdbcStatement* statement,
return AdbcSqliteBinderSetArrayStream(&stmt->binder, stream, error);
}
+AdbcStatusCode SqliteStatementGetOption(struct AdbcStatement* statement, const char* key,
+ char* value, size_t* length,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(statement, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode SqliteStatementGetOptionBytes(struct AdbcStatement* statement,
+ const char* key, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ CHECK_DB_INIT(statement, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode SqliteStatementGetOptionDouble(struct AdbcStatement* statement,
+ const char* key, double* value,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(statement, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
+AdbcStatusCode SqliteStatementGetOptionInt(struct AdbcStatement* statement,
+ const char* key, int64_t* value,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(statement, error);
+ return ADBC_STATUS_NOT_FOUND;
+}
+
AdbcStatusCode SqliteStatementGetParameterSchema(struct AdbcStatement* statement,
struct ArrowSchema* schema,
struct AdbcError* error) {
@@ -1375,6 +1499,27 @@ AdbcStatusCode SqliteStatementSetOption(struct AdbcStatement* statement, const c
return ADBC_STATUS_NOT_IMPLEMENTED;
}
+AdbcStatusCode SqliteStatementSetOptionBytes(struct AdbcStatement* statement,
+ const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ CHECK_DB_INIT(statement, error);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode SqliteStatementSetOptionDouble(struct AdbcStatement* statement,
+ const char* key, double value,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(statement, error);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode SqliteStatementSetOptionInt(struct AdbcStatement* statement,
+ const char* key, int64_t value,
+ struct AdbcError* error) {
+ CHECK_DB_INIT(statement, error);
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
AdbcStatusCode SqliteStatementExecutePartitions(struct AdbcStatement* statement,
struct ArrowSchema* schema,
struct AdbcPartitions* partitions,
@@ -1425,24 +1570,91 @@ AdbcStatusCode SqliteDriverInit(int version, void* raw_driver, struct AdbcError*
// Public names
-AdbcStatusCode AdbcDatabaseNew(struct AdbcDatabase* database, struct AdbcError* error) {
- return SqliteDatabaseNew(database, error);
+AdbcStatusCode AdbcDatabaseGetOption(struct AdbcDatabase* database, const char* key,
+ char* value, size_t* length,
+ struct AdbcError* error) {
+ return SqliteDatabaseGetOption(database, key, value, length, error);
}
-AdbcStatusCode AdbcDatabaseSetOption(struct AdbcDatabase* database, const char* key,
- const char* value, struct AdbcError* error) {
- return SqliteDatabaseSetOption(database, key, value, error);
+AdbcStatusCode AdbcDatabaseGetOptionBytes(struct AdbcDatabase* database, const char* key,
+ uint8_t* value, size_t* length,
+ struct AdbcError* error) {
+ return SqliteDatabaseGetOptionBytes(database, key, value, length, error);
+}
+
+AdbcStatusCode AdbcDatabaseGetOptionInt(struct AdbcDatabase* database, const char* key,
+ int64_t* value, struct AdbcError* error) {
+ return SqliteDatabaseGetOptionInt(database, key, value, error);
+}
+
+AdbcStatusCode AdbcDatabaseGetOptionDouble(struct AdbcDatabase* database, const char* key,
+ double* value, struct AdbcError* error) {
+ return SqliteDatabaseGetOptionDouble(database, key, value, error);
}
AdbcStatusCode AdbcDatabaseInit(struct AdbcDatabase* database, struct AdbcError* error) {
return SqliteDatabaseInit(database, error);
}
+AdbcStatusCode AdbcDatabaseNew(struct AdbcDatabase* database, struct AdbcError* error) {
+ return SqliteDatabaseNew(database, error);
+}
+
AdbcStatusCode AdbcDatabaseRelease(struct AdbcDatabase* database,
struct AdbcError* error) {
return SqliteDatabaseRelease(database, error);
}
+AdbcStatusCode AdbcDatabaseSetOption(struct AdbcDatabase* database, const char* key,
+ const char* value, struct AdbcError* error) {
+ return SqliteDatabaseSetOption(database, key, value, error);
+}
+
+AdbcStatusCode AdbcDatabaseSetOptionBytes(struct AdbcDatabase* database, const char* key,
+ const uint8_t* value, size_t length,
+ struct AdbcError* error) {
+ return SqliteDatabaseSetOptionBytes(database, key, value, length, error);
+}
+
+AdbcStatusCode AdbcDatabaseSetOptionInt(struct AdbcDatabase* database, const char* key,
+ int64_t value, struct AdbcError* error) {
+ return SqliteDatabaseSetOptionInt(database, key, value, error);
+}
+
+AdbcStatusCode AdbcDatabaseSetOptionDouble(struct AdbcDatabase* database, const char* key,
+ double value, struct AdbcError* error) {
+ return SqliteDatabaseSetOptionDouble(database, key, value, error);
+}
+
+AdbcStatusCode AdbcConnectionCancel(struct AdbcConnection* connection,
+ struct AdbcError* error) {
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
+AdbcStatusCode AdbcConnectionGetOption(struct AdbcConnection* connection, const char* key,
+ char* value, size_t* length,
+ struct AdbcError* error) {
+ return SqliteConnectionGetOption(connection, key, value, length, error);
+}
+
+AdbcStatusCode AdbcConnectionGetOptionBytes(struct AdbcConnection* connection,
+ const char* key, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ return SqliteConnectionGetOptionBytes(connection, key, value, length, error);
+}
+
+AdbcStatusCode AdbcConnectionGetOptionInt(struct AdbcConnection* connection,
+ const char* key, int64_t* value,
+ struct AdbcError* error) {
+ return SqliteConnectionGetOptionInt(connection, key, value, error);
+}
+
+AdbcStatusCode AdbcConnectionGetOptionDouble(struct AdbcConnection* connection,
+ const char* key, double* value,
+ struct AdbcError* error) {
+ return SqliteConnectionGetOptionDouble(connection, key, value, error);
+}
+
AdbcStatusCode AdbcConnectionNew(struct AdbcConnection* connection,
struct AdbcError* error) {
return SqliteConnectionNew(connection, error);
@@ -1453,6 +1665,24 @@ AdbcStatusCode AdbcConnectionSetOption(struct AdbcConnection* connection, const
return SqliteConnectionSetOption(connection, key, value, error);
}
+AdbcStatusCode AdbcConnectionSetOptionBytes(struct AdbcConnection* connection,
+ const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ return SqliteConnectionSetOptionBytes(connection, key, value, length, error);
+}
+
+AdbcStatusCode AdbcConnectionSetOptionInt(struct AdbcConnection* connection,
+ const char* key, int64_t value,
+ struct AdbcError* error) {
+ return SqliteConnectionSetOptionInt(connection, key, value, error);
+}
+
+AdbcStatusCode AdbcConnectionSetOptionDouble(struct AdbcConnection* connection,
+ const char* key, double value,
+ struct AdbcError* error) {
+ return SqliteConnectionSetOptionDouble(connection, key, value, error);
+}
+
AdbcStatusCode AdbcConnectionInit(struct AdbcConnection* connection,
struct AdbcDatabase* database,
struct AdbcError* error) {
@@ -1515,6 +1745,11 @@ AdbcStatusCode AdbcConnectionRollback(struct AdbcConnection* connection,
return SqliteConnectionRollback(connection, error);
}
+AdbcStatusCode AdbcStatementCancel(struct AdbcStatement* statement,
+ struct AdbcError* error) {
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
AdbcStatusCode AdbcStatementNew(struct AdbcConnection* connection,
struct AdbcStatement* statement,
struct AdbcError* error) {
@@ -1533,6 +1768,12 @@ AdbcStatusCode AdbcStatementExecuteQuery(struct AdbcStatement* statement,
return SqliteStatementExecuteQuery(statement, out, rows_affected, error);
}
+AdbcStatusCode AdbcStatementExecuteSchema(struct AdbcStatement* statement,
+ struct ArrowSchema* schema,
+ struct AdbcError* error) {
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+}
+
AdbcStatusCode AdbcStatementPrepare(struct AdbcStatement* statement,
struct AdbcError* error) {
return SqliteStatementPrepare(statement, error);
@@ -1561,6 +1802,29 @@ AdbcStatusCode AdbcStatementBindStream(struct AdbcStatement* statement,
return SqliteStatementBindStream(statement, stream, error);
}
+AdbcStatusCode AdbcStatementGetOption(struct AdbcStatement* statement, const char* key,
+ char* value, size_t* length,
+ struct AdbcError* error) {
+ return SqliteStatementGetOption(statement, key, value, length, error);
+}
+
+AdbcStatusCode AdbcStatementGetOptionBytes(struct AdbcStatement* statement,
+ const char* key, uint8_t* value,
+ size_t* length, struct AdbcError* error) {
+ return SqliteStatementGetOptionBytes(statement, key, value, length, error);
+}
+
+AdbcStatusCode AdbcStatementGetOptionInt(struct AdbcStatement* statement, const char* key,
+ int64_t* value, struct AdbcError* error) {
+ return SqliteStatementGetOptionInt(statement, key, value, error);
+}
+
+AdbcStatusCode AdbcStatementGetOptionDouble(struct AdbcStatement* statement,
+ const char* key, double* value,
+ struct AdbcError* error) {
+ return SqliteStatementGetOptionDouble(statement, key, value, error);
+}
+
AdbcStatusCode AdbcStatementGetParameterSchema(struct AdbcStatement* statement,
struct ArrowSchema* schema,
struct AdbcError* error) {
@@ -1572,6 +1836,23 @@ AdbcStatusCode AdbcStatementSetOption(struct AdbcStatement* statement, const cha
return SqliteStatementSetOption(statement, key, value, error);
}
+AdbcStatusCode AdbcStatementSetOptionBytes(struct AdbcStatement* statement,
+ const char* key, const uint8_t* value,
+ size_t length, struct AdbcError* error) {
+ return SqliteStatementSetOptionBytes(statement, key, value, length, error);
+}
+
+AdbcStatusCode AdbcStatementSetOptionInt(struct AdbcStatement* statement, const char* key,
+ int64_t value, struct AdbcError* error) {
+ return SqliteStatementSetOptionInt(statement, key, value, error);
+}
+
+AdbcStatusCode AdbcStatementSetOptionDouble(struct AdbcStatement* statement,
+ const char* key, double value,
+ struct AdbcError* error) {
+ return SqliteStatementSetOptionDouble(statement, key, value, error);
+}
+
AdbcStatusCode AdbcStatementExecutePartitions(struct AdbcStatement* statement,
struct ArrowSchema* schema,
struct AdbcPartitions* partitions,
diff --git a/c/driver/sqlite/sqlite_test.cc b/c/driver/sqlite/sqlite_test.cc
index b03158cc..d2821a30 100644
--- a/c/driver/sqlite/sqlite_test.cc
+++ b/c/driver/sqlite/sqlite_test.cc
@@ -92,7 +92,27 @@ class SqliteQuirks : public adbc_validation::DriverQuirks {
return ddl;
}
+ bool supports_bulk_ingest(const char* mode) const override {
+ return std::strcmp(mode, ADBC_INGEST_OPTION_MODE_APPEND) == 0 ||
+ std::strcmp(mode, ADBC_INGEST_OPTION_MODE_CREATE) == 0;
+ }
bool supports_concurrent_statements() const override { return true; }
+ bool supports_get_option() const override { return false; }
+ std::optional<adbc_validation::SqlInfoValue> supports_get_sql_info(
+ uint32_t info_code) const override {
+ switch (info_code) {
+ case ADBC_INFO_DRIVER_NAME:
+ return "ADBC SQLite Driver";
+ case ADBC_INFO_DRIVER_VERSION:
+ return "(unknown)";
+ case ADBC_INFO_VENDOR_NAME:
+ return "SQLite";
+ case ADBC_INFO_VENDOR_VERSION:
+ return "3.";
+ default:
+ return std::nullopt;
+ }
+ }
std::string catalog() const override { return "main"; }
std::string db_schema() const override { return ""; }
diff --git a/c/driver_manager/CMakeLists.txt b/c/driver_manager/CMakeLists.txt
index 637843ff..6fb51d9a 100644
--- a/c/driver_manager/CMakeLists.txt
+++ b/c/driver_manager/CMakeLists.txt
@@ -55,10 +55,9 @@ if(ADBC_BUILD_TESTS)
driver-manager
SOURCES
adbc_driver_manager_test.cc
- ../validation/adbc_validation.cc
- ../validation/adbc_validation_util.cc
EXTRA_LINK_LIBS
adbc_driver_common
+ adbc_validation
nanoarrow
${TEST_LINK_LIBS})
target_compile_features(adbc-driver-manager-test PRIVATE cxx_std_17)
@@ -73,8 +72,8 @@ if(ADBC_BUILD_TESTS)
SOURCES
adbc_version_100.c
adbc_version_100_compatibility_test.cc
- ../validation/adbc_validation_util.cc
EXTRA_LINK_LIBS
+ adbc_validation_util
nanoarrow
${TEST_LINK_LIBS})
target_compile_features(adbc-version-100-compatibility-test PRIVATE cxx_std_17)
diff --git a/c/driver_manager/adbc_driver_manager.cc b/c/driver_manager/adbc_driver_manager.cc
index 1d35482a..ea4ddc4c 100644
--- a/c/driver_manager/adbc_driver_manager.cc
+++ b/c/driver_manager/adbc_driver_manager.cc
@@ -132,23 +132,23 @@ static AdbcStatusCode ReleaseDriver(struct AdbcDriver* driver, struct AdbcError*
AdbcStatusCode DatabaseGetOption(struct AdbcDatabase* database, const char* key,
char* value, size_t* length, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode DatabaseGetOptionBytes(struct AdbcDatabase* database, const char* key,
uint8_t* value, size_t* length,
struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode DatabaseGetOptionInt(struct AdbcDatabase* database, const char* key,
int64_t* value, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode DatabaseGetOptionDouble(struct AdbcDatabase* database, const char* key,
double* value, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode DatabaseSetOption(struct AdbcDatabase* database, const char* key,
@@ -195,24 +195,24 @@ AdbcStatusCode ConnectionGetObjects(struct AdbcConnection*, int, const char*, co
AdbcStatusCode ConnectionGetOption(struct AdbcConnection* connection, const char* key,
char* value, size_t* length, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode ConnectionGetOptionBytes(struct AdbcConnection* connection,
const char* key, uint8_t* value, size_t* length,
struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode ConnectionGetOptionInt(struct AdbcConnection* connection, const char* key,
int64_t* value, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode ConnectionGetOptionDouble(struct AdbcConnection* connection,
const char* key, double* value,
struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode ConnectionGetTableSchema(struct AdbcConnection*, const char*, const char*,
@@ -284,23 +284,23 @@ AdbcStatusCode StatementExecuteSchema(struct AdbcStatement* statement,
AdbcStatusCode StatementGetOption(struct AdbcStatement* statement, const char* key,
char* value, size_t* length, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode StatementGetOptionBytes(struct AdbcStatement* statement, const char* key,
uint8_t* value, size_t* length,
struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode StatementGetOptionInt(struct AdbcStatement* statement, const char* key,
int64_t* value, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode StatementGetOptionDouble(struct AdbcStatement* statement, const char* key,
double* value, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode StatementGetParameterSchema(struct AdbcStatement* statement,
@@ -535,11 +535,11 @@ AdbcStatusCode AdbcDatabaseInit(struct AdbcDatabase* database, struct AdbcError*
// So we don't confuse a driver into thinking it's initialized already
database->private_data = nullptr;
if (args->init_func) {
- status = AdbcLoadDriverFromInitFunc(args->init_func, ADBC_VERSION_1_0_0,
+ status = AdbcLoadDriverFromInitFunc(args->init_func, ADBC_VERSION_1_1_0,
database->private_driver, error);
} else {
status = AdbcLoadDriver(args->driver.c_str(), args->entrypoint.c_str(),
- ADBC_VERSION_1_0_0, database->private_driver, error);
+ ADBC_VERSION_1_1_0, database->private_driver, error);
}
if (status != ADBC_STATUS_OK) {
// Restore private_data so it will be released by AdbcDatabaseRelease
@@ -1375,30 +1375,30 @@ AdbcStatusCode AdbcLoadDriverFromInitFunc(AdbcDriverInitFunc init_func, int vers
auto* driver = reinterpret_cast<struct AdbcDriver*>(raw_driver);
FILL_DEFAULT(driver, DatabaseGetOption);
FILL_DEFAULT(driver, DatabaseGetOptionBytes);
- FILL_DEFAULT(driver, DatabaseGetOptionInt);
FILL_DEFAULT(driver, DatabaseGetOptionDouble);
+ FILL_DEFAULT(driver, DatabaseGetOptionInt);
FILL_DEFAULT(driver, DatabaseSetOptionBytes);
- FILL_DEFAULT(driver, DatabaseSetOptionInt);
FILL_DEFAULT(driver, DatabaseSetOptionDouble);
+ FILL_DEFAULT(driver, DatabaseSetOptionInt);
FILL_DEFAULT(driver, ConnectionCancel);
FILL_DEFAULT(driver, ConnectionGetOption);
FILL_DEFAULT(driver, ConnectionGetOptionBytes);
- FILL_DEFAULT(driver, ConnectionGetOptionInt);
FILL_DEFAULT(driver, ConnectionGetOptionDouble);
+ FILL_DEFAULT(driver, ConnectionGetOptionInt);
FILL_DEFAULT(driver, ConnectionSetOptionBytes);
- FILL_DEFAULT(driver, ConnectionSetOptionInt);
FILL_DEFAULT(driver, ConnectionSetOptionDouble);
+ FILL_DEFAULT(driver, ConnectionSetOptionInt);
FILL_DEFAULT(driver, StatementCancel);
FILL_DEFAULT(driver, StatementExecuteSchema);
FILL_DEFAULT(driver, StatementGetOption);
FILL_DEFAULT(driver, StatementGetOptionBytes);
- FILL_DEFAULT(driver, StatementGetOptionInt);
FILL_DEFAULT(driver, StatementGetOptionDouble);
+ FILL_DEFAULT(driver, StatementGetOptionInt);
FILL_DEFAULT(driver, StatementSetOptionBytes);
- FILL_DEFAULT(driver, StatementSetOptionInt);
FILL_DEFAULT(driver, StatementSetOptionDouble);
+ FILL_DEFAULT(driver, StatementSetOptionInt);
}
return ADBC_STATUS_OK;
diff --git a/c/driver_manager/adbc_driver_manager_test.cc b/c/driver_manager/adbc_driver_manager_test.cc
index d3ff6f58..6ffa3c55 100644
--- a/c/driver_manager/adbc_driver_manager_test.cc
+++ b/c/driver_manager/adbc_driver_manager_test.cc
@@ -40,7 +40,7 @@ class DriverManager : public ::testing::Test {
std::memset(&driver, 0, sizeof(driver));
std::memset(&error, 0, sizeof(error));
- ASSERT_THAT(AdbcLoadDriver("adbc_driver_sqlite", nullptr, ADBC_VERSION_1_0_0, &driver,
+ ASSERT_THAT(AdbcLoadDriver("adbc_driver_sqlite", nullptr, ADBC_VERSION_1_1_0, &driver,
&error),
IsOkStatus(&error));
}
@@ -191,7 +191,27 @@ class SqliteQuirks : public adbc_validation::DriverQuirks {
}
}
+ bool supports_bulk_ingest(const char* mode) const override {
+ return std::strcmp(mode, ADBC_INGEST_OPTION_MODE_APPEND) == 0 ||
+ std::strcmp(mode, ADBC_INGEST_OPTION_MODE_CREATE) == 0;
+ }
bool supports_concurrent_statements() const override { return true; }
+ bool supports_get_option() const override { return false; }
+ std::optional<adbc_validation::SqlInfoValue> supports_get_sql_info(
+ uint32_t info_code) const override {
+ switch (info_code) {
+ case ADBC_INFO_DRIVER_NAME:
+ return "ADBC SQLite Driver";
+ case ADBC_INFO_DRIVER_VERSION:
+ return "(unknown)";
+ case ADBC_INFO_VENDOR_NAME:
+ return "SQLite";
+ case ADBC_INFO_VENDOR_VERSION:
+ return "3.";
+ default:
+ return std::nullopt;
+ }
+ }
};
class SqliteDatabaseTest : public ::testing::Test, public adbc_validation::DatabaseTest {
diff --git a/c/integration/duckdb/duckdb_test.cc b/c/integration/duckdb/duckdb_test.cc
index fd6e1984..be163418 100644
--- a/c/integration/duckdb/duckdb_test.cc
+++ b/c/integration/duckdb/duckdb_test.cc
@@ -46,7 +46,7 @@ class DuckDbQuirks : public adbc_validation::DriverQuirks {
std::string BindParameter(int index) const override { return "?"; }
- bool supports_bulk_ingest() const override { return false; }
+ bool supports_bulk_ingest(const char* /*mode*/) const override { return false; }
bool supports_concurrent_statements() const override { return true; }
bool supports_dynamic_parameter_binding() const override { return false; }
bool supports_get_sql_info() const override { return false; }
@@ -96,6 +96,8 @@ class DuckDbStatementTest : public ::testing::Test,
void TestSqlIngestTableEscaping() { GTEST_SKIP() << "Table escaping not implemented"; }
+ void TestSqlQueryErrors() { GTEST_SKIP() << "DuckDB does not set AdbcError.release"; }
+
protected:
DuckDbQuirks quirks_;
};
diff --git a/c/validation/CMakeLists.txt b/c/validation/CMakeLists.txt
index 2f6549b5..bab7a63b 100644
--- a/c/validation/CMakeLists.txt
+++ b/c/validation/CMakeLists.txt
@@ -15,11 +15,24 @@
# specific language governing permissions and limitations
# under the License.
-add_library(adbc_validation OBJECT adbc_validation.cc adbc_validation_util.cc)
+add_library(adbc_validation_util STATIC adbc_validation_util.cc)
+adbc_configure_target(adbc_validation_util)
+target_compile_features(adbc_validation_util PRIVATE cxx_std_17)
+target_include_directories(adbc_validation_util SYSTEM
+ PRIVATE "${REPOSITORY_ROOT}" "${REPOSITORY_ROOT}/c/driver/"
+ "${REPOSITORY_ROOT}/c/vendor/")
+target_link_libraries(adbc_validation_util PUBLIC adbc_driver_common nanoarrow
+ GTest::gtest GTest::gmock)
+
+add_library(adbc_validation OBJECT adbc_validation.cc)
adbc_configure_target(adbc_validation)
target_compile_features(adbc_validation PRIVATE cxx_std_17)
target_include_directories(adbc_validation SYSTEM
PRIVATE "${REPOSITORY_ROOT}" "${REPOSITORY_ROOT}/c/driver/"
"${REPOSITORY_ROOT}/c/vendor/")
-target_link_libraries(adbc_validation PUBLIC adbc_driver_common nanoarrow GTest::gtest
- GTest::gmock)
+target_link_libraries(adbc_validation
+ PUBLIC adbc_driver_common
+ adbc_validation_util
+ nanoarrow
+ GTest::gtest
+ GTest::gmock)
diff --git a/c/validation/adbc_validation.cc b/c/validation/adbc_validation.cc
index c5ada836..05014b92 100644
--- a/c/validation/adbc_validation.cc
+++ b/c/validation/adbc_validation.cc
@@ -34,6 +34,7 @@
#include <gtest/gtest-matchers.h>
#include <gtest/gtest.h>
#include <nanoarrow/nanoarrow.h>
+#include <nanoarrow/nanoarrow.hpp>
#include "adbc_validation_util.h"
@@ -102,7 +103,7 @@ AdbcStatusCode DriverQuirks::EnsureSampleTable(struct AdbcConnection* connection
AdbcStatusCode DriverQuirks::CreateSampleTable(struct AdbcConnection* connection,
const std::string& name,
struct AdbcError* error) const {
- if (!supports_bulk_ingest()) {
+ if (!supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE)) {
return ADBC_STATUS_NOT_IMPLEMENTED;
}
return DoIngestSampleTable(connection, name, error);
@@ -248,6 +249,56 @@ void ConnectionTest::TestAutocommitToggle() {
//------------------------------------------------------------
// Tests of metadata
+std::optional<std::string> ConnectionGetOption(struct AdbcConnection* connection,
+ std::string_view option,
+ struct AdbcError* error) {
+ char buffer[128];
+ size_t buffer_size = sizeof(buffer);
+ AdbcStatusCode status =
+ AdbcConnectionGetOption(connection, option.data(), buffer, &buffer_size, error);
+ EXPECT_THAT(status, IsOkStatus(error));
+ if (status != ADBC_STATUS_OK) return std::nullopt;
+ EXPECT_GT(buffer_size, 0);
+ if (buffer_size == 0) return std::nullopt;
+ return std::string(buffer, buffer_size - 1);
+}
+
+void ConnectionTest::TestMetadataCurrentCatalog() {
+ ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error), IsOkStatus(&error));
+
+ if (quirks()->supports_metadata_current_catalog()) {
+ ASSERT_THAT(
+ ConnectionGetOption(&connection, ADBC_CONNECTION_OPTION_CURRENT_CATALOG, &error),
+ ::testing::Optional(quirks()->catalog()));
+ } else {
+ char buffer[128];
+ size_t buffer_size = sizeof(buffer);
+ ASSERT_THAT(
+ AdbcConnectionGetOption(&connection, ADBC_CONNECTION_OPTION_CURRENT_CATALOG,
+ buffer, &buffer_size, &error),
+ IsStatus(ADBC_STATUS_NOT_FOUND));
+ }
+}
+
+void ConnectionTest::TestMetadataCurrentDbSchema() {
+ ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error), IsOkStatus(&error));
+
+ if (quirks()->supports_metadata_current_db_schema()) {
+ ASSERT_THAT(ConnectionGetOption(&connection, ADBC_CONNECTION_OPTION_CURRENT_DB_SCHEMA,
+ &error),
+ ::testing::Optional(quirks()->db_schema()));
+ } else {
+ char buffer[128];
+ size_t buffer_size = sizeof(buffer);
+ ASSERT_THAT(
+ AdbcConnectionGetOption(&connection, ADBC_CONNECTION_OPTION_CURRENT_DB_SCHEMA,
+ buffer, &buffer_size, &error),
+ IsStatus(ADBC_STATUS_NOT_FOUND));
+ }
+}
+
void ConnectionTest::TestMetadataGetInfo() {
ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error), IsOkStatus(&error));
@@ -263,6 +314,11 @@ void ConnectionTest::TestMetadataGetInfo() {
ADBC_INFO_VENDOR_NAME,
ADBC_INFO_VENDOR_VERSION,
}) {
+ SCOPED_TRACE("info_code = " + std::to_string(info_code));
+ std::optional<SqlInfoValue> expected = quirks()->supports_get_sql_info(info_code);
+
+ if (!expected.has_value()) continue;
+
uint32_t info[] = {info_code};
StreamReader reader;
@@ -316,8 +372,10 @@ void ConnectionTest::TestMetadataGetInfo() {
const uint32_t code =
reader.array_view->children[0]->buffer_views[1].data.as_uint32[row];
seen.push_back(code);
+ if (code != info_code) {
+ continue;
+ }
- std::optional<SqlInfoValue> expected = quirks()->supports_get_sql_info(code);
ASSERT_TRUE(expected.has_value()) << "Got unexpected info code " << code;
uint8_t type_code =
@@ -329,16 +387,16 @@ void ConnectionTest::TestMetadataGetInfo() {
using T = std::decay_t<decltype(expected_value)>;
if constexpr (std::is_same_v<T, int64_t>) {
ASSERT_EQ(uint8_t(2), type_code);
- ASSERT_EQ(expected_value,
+ EXPECT_EQ(expected_value,
ArrowArrayViewGetIntUnsafe(
reader.array_view->children[1]->children[2], offset));
} else if constexpr (std::is_same_v<T, std::string>) {
ASSERT_EQ(uint8_t(0), type_code);
struct ArrowStringView view = ArrowArrayViewGetStringUnsafe(
reader.array_view->children[1]->children[0], offset);
- ASSERT_EQ(expected_value,
- std::string_view(static_cast<const char*>(view.data),
- view.size_bytes));
+ EXPECT_THAT(std::string_view(static_cast<const char*>(view.data),
+ view.size_bytes),
+ ::testing::HasSubstr(expected_value));
} else {
static_assert(!sizeof(T), "not yet implemented");
}
@@ -347,12 +405,12 @@ void ConnectionTest::TestMetadataGetInfo() {
<< "code: " << type_code;
}
}
- ASSERT_THAT(seen, ::testing::IsSupersetOf(info));
+ EXPECT_THAT(seen, ::testing::IsSupersetOf(info));
}
}
void ConnectionTest::TestMetadataGetTableSchema() {
- if (!quirks()->supports_bulk_ingest()) {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE)) {
GTEST_SKIP();
}
ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
@@ -932,6 +990,33 @@ void ConnectionTest::TestMetadataGetObjectsPrimaryKey() {
ASSERT_EQ(constraint_column_name, "id");
}
+void ConnectionTest::TestMetadataGetObjectsCancel() {
+ if (!quirks()->supports_cancel() || !quirks()->supports_get_objects()) {
+ GTEST_SKIP();
+ }
+
+ ASSERT_THAT(AdbcConnectionNew(&connection, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcConnectionInit(&connection, &database, &error), IsOkStatus(&error));
+
+ StreamReader reader;
+ ASSERT_THAT(
+ AdbcConnectionGetObjects(&connection, ADBC_OBJECT_DEPTH_CATALOGS, nullptr, nullptr,
+ nullptr, nullptr, nullptr, &reader.stream.value, &error),
+ IsOkStatus(&error));
+ ASSERT_NO_FATAL_FAILURE(reader.GetSchema());
+
+ ASSERT_THAT(AdbcConnectionCancel(&connection, &error), IsOkStatus(&error));
+
+ while (true) {
+ int err = reader.MaybeNext();
+ if (err != 0) {
+ ASSERT_THAT(err, ::testing::AnyOf(0, IsErrno(ECANCELED, &reader.stream.value,
+ /*ArrowError*/ nullptr)));
+ }
+ if (!reader.array->release) break;
+ }
+}
+
//------------------------------------------------------------
// Tests of AdbcStatement
@@ -986,7 +1071,7 @@ void StatementTest::TestRelease() {
template <typename CType>
void StatementTest::TestSqlIngestType(ArrowType type,
const std::vector<std::optional<CType>>& values) {
- if (!quirks()->supports_bulk_ingest()) {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE)) {
GTEST_SKIP();
}
@@ -1126,7 +1211,7 @@ void StatementTest::TestSqlIngestDate32() {
template <enum ArrowTimeUnit TU>
void StatementTest::TestSqlIngestTimestampType(const char* timezone) {
- if (!quirks()->supports_bulk_ingest()) {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE)) {
GTEST_SKIP();
}
@@ -1224,7 +1309,7 @@ void StatementTest::TestSqlIngestTimestampTz() {
}
void StatementTest::TestSqlIngestInterval() {
- if (!quirks()->supports_bulk_ingest()) {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE)) {
GTEST_SKIP();
}
@@ -1332,7 +1417,8 @@ void StatementTest::TestSqlIngestTableEscaping() {
}
void StatementTest::TestSqlIngestAppend() {
- if (!quirks()->supports_bulk_ingest()) {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE) ||
+ !quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_APPEND)) {
GTEST_SKIP();
}
@@ -1410,8 +1496,185 @@ void StatementTest::TestSqlIngestAppend() {
ASSERT_THAT(AdbcStatementRelease(&statement, &error), IsOkStatus(&error));
}
+void StatementTest::TestSqlIngestReplace() {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_REPLACE)) {
+ GTEST_SKIP();
+ }
+
+ // Ingest
+
+ Handle<struct ArrowSchema> schema;
+ Handle<struct ArrowArray> array;
+ struct ArrowError na_error;
+ ASSERT_THAT(MakeSchema(&schema.value, {{"int64s", NANOARROW_TYPE_INT64}}), IsOkErrno());
+ ASSERT_THAT(MakeBatch<int64_t>(&schema.value, &array.value, &na_error, {42}),
+ IsOkErrno());
+
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_TARGET_TABLE,
+ "bulk_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_MODE,
+ ADBC_INGEST_OPTION_MODE_REPLACE, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementBind(&statement, &array.value, &schema.value, &error),
+ IsOkStatus(&error));
+
+ int64_t rows_affected = 0;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, &rows_affected, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(rows_affected, ::testing::AnyOf(::testing::Eq(1), ::testing::Eq(-1)));
+
+ // Read data back
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "SELECT * FROM bulk_ingest", &error),
+ IsOkStatus(&error));
+ {
+ StreamReader reader;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, &reader.stream.value,
+ &reader.rows_affected, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(reader.rows_affected,
+ ::testing::AnyOf(::testing::Eq(1), ::testing::Eq(-1)));
+
+ ASSERT_NO_FATAL_FAILURE(reader.GetSchema());
+ ASSERT_NO_FATAL_FAILURE(CompareSchema(&reader.schema.value,
+ {{"int64s", NANOARROW_TYPE_INT64, NULLABLE}}));
+
+ ASSERT_NO_FATAL_FAILURE(reader.Next());
+ ASSERT_NE(nullptr, reader.array->release);
+ ASSERT_EQ(1, reader.array->length);
+ ASSERT_EQ(1, reader.array->n_children);
+
+ ASSERT_NO_FATAL_FAILURE(CompareArray<int64_t>(reader.array_view->children[0], {42}));
+
+ ASSERT_NO_FATAL_FAILURE(reader.Next());
+ ASSERT_EQ(nullptr, reader.array->release);
+ }
+
+ // Replace
+ // Re-initialize since Bind() should take ownership of data
+ ASSERT_THAT(MakeSchema(&schema.value, {{"int64s", NANOARROW_TYPE_INT64}}), IsOkErrno());
+ ASSERT_THAT(MakeBatch<int64_t>(&schema.value, &array.value, &na_error, {-42, -42}),
+ IsOkErrno());
+
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_TARGET_TABLE,
+ "bulk_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_MODE,
+ ADBC_INGEST_OPTION_MODE_REPLACE, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementBind(&statement, &array.value, &schema.value, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, &rows_affected, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(rows_affected, ::testing::AnyOf(::testing::Eq(2), ::testing::Eq(-1)));
+
+ // Read data back
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "SELECT * FROM bulk_ingest", &error),
+ IsOkStatus(&error));
+ {
+ StreamReader reader;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, &reader.stream.value,
+ &reader.rows_affected, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(reader.rows_affected,
+ ::testing::AnyOf(::testing::Eq(2), ::testing::Eq(-1)));
+
+ ASSERT_NO_FATAL_FAILURE(reader.GetSchema());
+ ASSERT_NO_FATAL_FAILURE(CompareSchema(&reader.schema.value,
+ {{"int64s", NANOARROW_TYPE_INT64, NULLABLE}}));
+
+ ASSERT_NO_FATAL_FAILURE(reader.Next());
+ ASSERT_NE(nullptr, reader.array->release);
+ ASSERT_EQ(2, reader.array->length);
+ ASSERT_EQ(1, reader.array->n_children);
+
+ ASSERT_NO_FATAL_FAILURE(
+ CompareArray<int64_t>(reader.array_view->children[0], {-42, -42}));
+
+ ASSERT_NO_FATAL_FAILURE(reader.Next());
+ ASSERT_EQ(nullptr, reader.array->release);
+ }
+}
+
+void StatementTest::TestSqlIngestCreateAppend() {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE_APPEND)) {
+ GTEST_SKIP();
+ }
+
+ ASSERT_THAT(quirks()->DropTable(&connection, "bulk_ingest", &error),
+ IsOkStatus(&error));
+
+ // Ingest
+
+ Handle<struct ArrowSchema> schema;
+ Handle<struct ArrowArray> array;
+ struct ArrowError na_error;
+ ASSERT_THAT(MakeSchema(&schema.value, {{"int64s", NANOARROW_TYPE_INT64}}), IsOkErrno());
+ ASSERT_THAT(MakeBatch<int64_t>(&schema.value, &array.value, &na_error, {42}),
+ IsOkErrno());
+
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_TARGET_TABLE,
+ "bulk_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_MODE,
+ ADBC_INGEST_OPTION_MODE_CREATE_APPEND, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementBind(&statement, &array.value, &schema.value, &error),
+ IsOkStatus(&error));
+
+ int64_t rows_affected = 0;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, &rows_affected, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(rows_affected, ::testing::AnyOf(::testing::Eq(1), ::testing::Eq(-1)));
+
+ // Append
+ // Re-initialize since Bind() should take ownership of data
+ ASSERT_THAT(MakeSchema(&schema.value, {{"int64s", NANOARROW_TYPE_INT64}}), IsOkErrno());
+ ASSERT_THAT(MakeBatch<int64_t>(&schema.value, &array.value, &na_error, {42, 42}),
+ IsOkErrno());
+
+ ASSERT_THAT(AdbcStatementBind(&statement, &array.value, &schema.value, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, &rows_affected, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(rows_affected, ::testing::AnyOf(::testing::Eq(2), ::testing::Eq(-1)));
+
+ // Read data back
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "SELECT * FROM bulk_ingest", &error),
+ IsOkStatus(&error));
+ {
+ StreamReader reader;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, &reader.stream.value,
+ &reader.rows_affected, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(reader.rows_affected,
+ ::testing::AnyOf(::testing::Eq(3), ::testing::Eq(-1)));
+
+ ASSERT_NO_FATAL_FAILURE(reader.GetSchema());
+ ASSERT_NO_FATAL_FAILURE(CompareSchema(&reader.schema.value,
+ {{"int64s", NANOARROW_TYPE_INT64, NULLABLE}}));
+
+ ASSERT_NO_FATAL_FAILURE(reader.Next());
+ ASSERT_NE(nullptr, reader.array->release);
+ ASSERT_EQ(3, reader.array->length);
+ ASSERT_EQ(1, reader.array->n_children);
+
+ ASSERT_NO_FATAL_FAILURE(
+ CompareArray<int64_t>(reader.array_view->children[0], {42, 42, 42}));
+
+ ASSERT_NO_FATAL_FAILURE(reader.Next());
+ ASSERT_EQ(nullptr, reader.array->release);
+ }
+
+ ASSERT_THAT(AdbcStatementRelease(&statement, &error), IsOkStatus(&error));
+}
+
void StatementTest::TestSqlIngestErrors() {
- if (!quirks()->supports_bulk_ingest()) {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE)) {
GTEST_SKIP();
}
@@ -1491,7 +1754,7 @@ void StatementTest::TestSqlIngestErrors() {
}
void StatementTest::TestSqlIngestMultipleConnections() {
- if (!quirks()->supports_bulk_ingest()) {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE)) {
GTEST_SKIP();
}
@@ -1561,7 +1824,7 @@ void StatementTest::TestSqlIngestMultipleConnections() {
}
void StatementTest::TestSqlIngestSample() {
- if (!quirks()->supports_bulk_ingest()) {
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE)) {
GTEST_SKIP();
}
@@ -1807,7 +2070,7 @@ void StatementTest::TestSqlPrepareSelectParams() {
}
void StatementTest::TestSqlPrepareUpdate() {
- if (!quirks()->supports_bulk_ingest() ||
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE) ||
!quirks()->supports_dynamic_parameter_binding()) {
GTEST_SKIP();
}
@@ -1886,7 +2149,7 @@ void StatementTest::TestSqlPrepareUpdateNoParams() {
}
void StatementTest::TestSqlPrepareUpdateStream() {
- if (!quirks()->supports_bulk_ingest() ||
+ if (!quirks()->supports_bulk_ingest(ADBC_INGEST_OPTION_MODE_CREATE) ||
!quirks()->supports_dynamic_parameter_binding()) {
GTEST_SKIP();
}
@@ -2161,6 +2424,36 @@ void StatementTest::TestSqlQueryStrings() {
ASSERT_THAT(AdbcStatementRelease(&statement, &error), IsOkStatus(&error));
}
+void StatementTest::TestSqlQueryCancel() {
+ if (!quirks()->supports_cancel()) {
+ GTEST_SKIP();
+ }
+
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "SELECT 'SaShiSuSeSo'", &error),
+ IsOkStatus(&error));
+
+ {
+ StreamReader reader;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, &reader.stream.value,
+ &reader.rows_affected, &error),
+ IsOkStatus(&error));
+ ASSERT_NO_FATAL_FAILURE(reader.GetSchema());
+
+ ASSERT_THAT(AdbcStatementCancel(&statement, &error), IsOkStatus(&error));
+ while (true) {
+ int err = reader.MaybeNext();
+ if (err != 0) {
+ ASSERT_THAT(err, ::testing::AnyOf(0, IsErrno(ECANCELED, &reader.stream.value,
+ /*ArrowError*/ nullptr)));
+ }
+ if (!reader.array->release) break;
+ }
+ }
+
+ ASSERT_THAT(AdbcStatementRelease(&statement, &error), IsOkStatus(&error));
+}
+
void StatementTest::TestSqlQueryErrors() {
// Invalid query
ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error), IsOkStatus(&error));
@@ -2180,6 +2473,13 @@ void StatementTest::TestTransactions() {
ASSERT_THAT(quirks()->DropTable(&connection, "bulk_ingest", &error),
IsOkStatus(&error));
+ if (quirks()->supports_get_option()) {
+ auto autocommit =
+ ConnectionGetOption(&connection, ADBC_CONNECTION_OPTION_AUTOCOMMIT, &error);
+ ASSERT_THAT(autocommit,
+ ::testing::Optional(::testing::StrEq(ADBC_OPTION_VALUE_ENABLED)));
+ }
+
Handle<struct AdbcConnection> connection2;
ASSERT_THAT(AdbcConnectionNew(&connection2.value, &error), IsOkStatus(&error));
ASSERT_THAT(AdbcConnectionInit(&connection2.value, &database, &error),
@@ -2189,6 +2489,13 @@ void StatementTest::TestTransactions() {
ADBC_OPTION_VALUE_DISABLED, &error),
IsOkStatus(&error));
+ if (quirks()->supports_get_option()) {
+ auto autocommit =
+ ConnectionGetOption(&connection, ADBC_CONNECTION_OPTION_AUTOCOMMIT, &error);
+ ASSERT_THAT(autocommit,
+ ::testing::Optional(::testing::StrEq(ADBC_OPTION_VALUE_DISABLED)));
+ }
+
// Uncommitted change
ASSERT_NO_FATAL_FAILURE(IngestSampleTable(&connection, &error));
@@ -2264,6 +2571,86 @@ void StatementTest::TestTransactions() {
}
}
+void StatementTest::TestSqlSchemaInts() {
+ if (!quirks()->supports_execute_schema()) {
+ GTEST_SKIP() << "Not supported";
+ }
+
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "SELECT 42", &error),
+ IsOkStatus(&error));
+
+ nanoarrow::UniqueSchema schema;
+ ASSERT_THAT(AdbcStatementExecuteSchema(&statement, schema.get(), &error),
+ IsOkStatus(&error));
+
+ ASSERT_EQ(1, schema->n_children);
+ ASSERT_THAT(schema->children[0]->format, ::testing::AnyOfArray({
+ ::testing::StrEq("i"), // int32
+ ::testing::StrEq("l"), // int64
+ }));
+
+ ASSERT_THAT(AdbcStatementRelease(&statement, &error), IsOkStatus(&error));
+}
+
+void StatementTest::TestSqlSchemaFloats() {
+ if (!quirks()->supports_execute_schema()) {
+ GTEST_SKIP() << "Not supported";
+ }
+
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "SELECT CAST(1.5 AS FLOAT)", &error),
+ IsOkStatus(&error));
+
+ nanoarrow::UniqueSchema schema;
+ ASSERT_THAT(AdbcStatementExecuteSchema(&statement, schema.get(), &error),
+ IsOkStatus(&error));
+
+ ASSERT_EQ(1, schema->n_children);
+ ASSERT_THAT(schema->children[0]->format, ::testing::AnyOfArray({
+ ::testing::StrEq("f"), // float32
+ ::testing::StrEq("g"), // float64
+ }));
+
+ ASSERT_THAT(AdbcStatementRelease(&statement, &error), IsOkStatus(&error));
+}
+
+void StatementTest::TestSqlSchemaStrings() {
+ if (!quirks()->supports_execute_schema()) {
+ GTEST_SKIP() << "Not supported";
+ }
+
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error), IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement, "SELECT 'hi'", &error),
+ IsOkStatus(&error));
+
+ nanoarrow::UniqueSchema schema;
+ ASSERT_THAT(AdbcStatementExecuteSchema(&statement, schema.get(), &error),
+ IsOkStatus(&error));
+
+ ASSERT_EQ(1, schema->n_children);
+ ASSERT_THAT(schema->children[0]->format, ::testing::AnyOfArray({
+ ::testing::StrEq("u"), // string
+ ::testing::StrEq("U"), // large_string
+ }));
+
+ ASSERT_THAT(AdbcStatementRelease(&statement, &error), IsOkStatus(&error));
+}
+
+void StatementTest::TestSqlSchemaErrors() {
+ if (!quirks()->supports_execute_schema()) {
+ GTEST_SKIP() << "Not supported";
+ }
+
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error), IsOkStatus(&error));
+
+ nanoarrow::UniqueSchema schema;
+ ASSERT_THAT(AdbcStatementExecuteSchema(&statement, schema.get(), &error),
+ IsStatus(ADBC_STATUS_INVALID_STATE, &error));
+
+ ASSERT_THAT(AdbcStatementRelease(&statement, &error), IsOkStatus(&error));
+}
+
void StatementTest::TestConcurrentStatements() {
Handle<struct AdbcStatement> statement1;
Handle<struct AdbcStatement> statement2;
diff --git a/c/validation/adbc_validation.h b/c/validation/adbc_validation.h
index d4d0265e..acca5be4 100644
--- a/c/validation/adbc_validation.h
+++ b/c/validation/adbc_validation.h
@@ -88,10 +88,22 @@ class DriverQuirks {
return ingest_type;
}
+ /// \brief Whether bulk ingest is supported
+ virtual bool supports_bulk_ingest(const char* mode) const { return true; }
+
+ /// \brief Whether we can cancel queries.
+ virtual bool supports_cancel() const { return false; }
+
/// \brief Whether two statements can be used at the same time on a
/// single connection
virtual bool supports_concurrent_statements() const { return false; }
+ /// \brief Whether AdbcStatementExecuteSchema should work
+ virtual bool supports_execute_schema() const { return false; }
+
+ /// \brief Whether GetOption* should work
+ virtual bool supports_get_option() const { return true; }
+
/// \brief Whether AdbcStatementExecutePartitions should work
virtual bool supports_partitioned_data() const { return false; }
@@ -112,8 +124,11 @@ class DriverQuirks {
/// \brief Whether GetObjects is implemented
virtual bool supports_get_objects() const { return true; }
- /// \brief Whether bulk ingest is supported
- virtual bool supports_bulk_ingest() const { return true; }
+ /// \brief Whether we can get ADBC_CONNECTION_OPTION_CURRENT_CATALOG
+ virtual bool supports_metadata_current_catalog() const { return false; }
+
+ /// \brief Whether we can get ADBC_CONNECTION_OPTION_CURRENT_DB_SCHEMA
+ virtual bool supports_metadata_current_db_schema() const { return false; }
/// \brief Whether dynamic parameter bindings are supported for prepare
virtual bool supports_dynamic_parameter_binding() const { return true; }
@@ -165,6 +180,9 @@ class ConnectionTest {
void TestAutocommitToggle();
+ void TestMetadataCurrentCatalog();
+ void TestMetadataCurrentDbSchema();
+
void TestMetadataGetInfo();
void TestMetadataGetTableSchema();
void TestMetadataGetTableTypes();
@@ -176,6 +194,7 @@ class ConnectionTest {
void TestMetadataGetObjectsColumns();
void TestMetadataGetObjectsConstraints();
void TestMetadataGetObjectsPrimaryKey();
+ void TestMetadataGetObjectsCancel();
protected:
struct AdbcError error;
@@ -183,28 +202,31 @@ class ConnectionTest {
struct AdbcConnection connection;
};
-#define ADBCV_TEST_CONNECTION(FIXTURE) \
- static_assert(std::is_base_of<adbc_validation::ConnectionTest, FIXTURE>::value, \
- ADBCV_STRINGIFY(FIXTURE) " must inherit from ConnectionTest"); \
- TEST_F(FIXTURE, NewInit) { TestNewInit(); } \
- TEST_F(FIXTURE, Release) { TestRelease(); } \
- TEST_F(FIXTURE, Concurrent) { TestConcurrent(); } \
- TEST_F(FIXTURE, AutocommitDefault) { TestAutocommitDefault(); } \
- TEST_F(FIXTURE, AutocommitToggle) { TestAutocommitToggle(); } \
- TEST_F(FIXTURE, MetadataGetInfo) { TestMetadataGetInfo(); } \
- TEST_F(FIXTURE, MetadataGetTableSchema) { TestMetadataGetTableSchema(); } \
- TEST_F(FIXTURE, MetadataGetTableTypes) { TestMetadataGetTableTypes(); } \
- TEST_F(FIXTURE, MetadataGetObjectsCatalogs) { TestMetadataGetObjectsCatalogs(); } \
- TEST_F(FIXTURE, MetadataGetObjectsDbSchemas) { TestMetadataGetObjectsDbSchemas(); } \
- TEST_F(FIXTURE, MetadataGetObjectsTables) { TestMetadataGetObjectsTables(); } \
- TEST_F(FIXTURE, MetadataGetObjectsTablesTypes) { \
- TestMetadataGetObjectsTablesTypes(); \
- } \
- TEST_F(FIXTURE, MetadataGetObjectsColumns) { TestMetadataGetObjectsColumns(); } \
- TEST_F(FIXTURE, MetadataGetObjectsConstraints) { \
- TestMetadataGetObjectsConstraints(); \
- } \
- TEST_F(FIXTURE, MetadataGetObjectsPrimaryKey) { TestMetadataGetObjectsPrimaryKey(); }
+#define ADBCV_TEST_CONNECTION(FIXTURE) \
+ static_assert(std::is_base_of<adbc_validation::ConnectionTest, FIXTURE>::value, \
+ ADBCV_STRINGIFY(FIXTURE) " must inherit from ConnectionTest"); \
+ TEST_F(FIXTURE, NewInit) { TestNewInit(); } \
+ TEST_F(FIXTURE, Release) { TestRelease(); } \
+ TEST_F(FIXTURE, Concurrent) { TestConcurrent(); } \
+ TEST_F(FIXTURE, AutocommitDefault) { TestAutocommitDefault(); } \
+ TEST_F(FIXTURE, AutocommitToggle) { TestAutocommitToggle(); } \
+ TEST_F(FIXTURE, MetadataCurrentCatalog) { TestMetadataCurrentCatalog(); } \
+ TEST_F(FIXTURE, MetadataCurrentDbSchema) { TestMetadataCurrentDbSchema(); } \
+ TEST_F(FIXTURE, MetadataGetInfo) { TestMetadataGetInfo(); } \
+ TEST_F(FIXTURE, MetadataGetTableSchema) { TestMetadataGetTableSchema(); } \
+ TEST_F(FIXTURE, MetadataGetTableTypes) { TestMetadataGetTableTypes(); } \
+ TEST_F(FIXTURE, MetadataGetObjectsCatalogs) { TestMetadataGetObjectsCatalogs(); } \
+ TEST_F(FIXTURE, MetadataGetObjectsDbSchemas) { TestMetadataGetObjectsDbSchemas(); } \
+ TEST_F(FIXTURE, MetadataGetObjectsTables) { TestMetadataGetObjectsTables(); } \
+ TEST_F(FIXTURE, MetadataGetObjectsTablesTypes) { \
+ TestMetadataGetObjectsTablesTypes(); \
+ } \
+ TEST_F(FIXTURE, MetadataGetObjectsColumns) { TestMetadataGetObjectsColumns(); } \
+ TEST_F(FIXTURE, MetadataGetObjectsConstraints) { \
+ TestMetadataGetObjectsConstraints(); \
+ } \
+ TEST_F(FIXTURE, MetadataGetObjectsPrimaryKey) { TestMetadataGetObjectsPrimaryKey(); } \
+ TEST_F(FIXTURE, MetadataGetObjectsCancel) { TestMetadataGetObjectsCancel(); }
class StatementTest {
public:
@@ -247,6 +269,8 @@ class StatementTest {
void TestSqlIngestTableEscaping();
void TestSqlIngestAppend();
+ void TestSqlIngestReplace();
+ void TestSqlIngestCreateAppend();
void TestSqlIngestErrors();
void TestSqlIngestMultipleConnections();
void TestSqlIngestSample();
@@ -266,8 +290,15 @@ class StatementTest {
void TestSqlQueryFloats();
void TestSqlQueryStrings();
+ void TestSqlQueryCancel();
void TestSqlQueryErrors();
+ void TestSqlSchemaInts();
+ void TestSqlSchemaFloats();
+ void TestSqlSchemaStrings();
+
+ void TestSqlSchemaErrors();
+
void TestTransactions();
void TestConcurrentStatements();
@@ -316,6 +347,8 @@ class StatementTest {
TEST_F(FIXTURE, SqlIngestInterval) { TestSqlIngestInterval(); } \
TEST_F(FIXTURE, SqlIngestTableEscaping) { TestSqlIngestTableEscaping(); } \
TEST_F(FIXTURE, SqlIngestAppend) { TestSqlIngestAppend(); } \
+ TEST_F(FIXTURE, SqlIngestReplace) { TestSqlIngestReplace(); } \
+ TEST_F(FIXTURE, SqlIngestCreateAppend) { TestSqlIngestCreateAppend(); } \
TEST_F(FIXTURE, SqlIngestErrors) { TestSqlIngestErrors(); } \
TEST_F(FIXTURE, SqlIngestMultipleConnections) { TestSqlIngestMultipleConnections(); } \
TEST_F(FIXTURE, SqlIngestSample) { TestSqlIngestSample(); } \
@@ -333,7 +366,12 @@ class StatementTest {
TEST_F(FIXTURE, SqlQueryInts) { TestSqlQueryInts(); } \
TEST_F(FIXTURE, SqlQueryFloats) { TestSqlQueryFloats(); } \
TEST_F(FIXTURE, SqlQueryStrings) { TestSqlQueryStrings(); } \
+ TEST_F(FIXTURE, SqlQueryCancel) { TestSqlQueryCancel(); } \
TEST_F(FIXTURE, SqlQueryErrors) { TestSqlQueryErrors(); } \
+ TEST_F(FIXTURE, SqlSchemaInts) { TestSqlSchemaInts(); } \
+ TEST_F(FIXTURE, SqlSchemaFloats) { TestSqlSchemaFloats(); } \
+ TEST_F(FIXTURE, SqlSchemaStrings) { TestSqlSchemaStrings(); } \
+ TEST_F(FIXTURE, SqlSchemaErrors) { TestSqlSchemaErrors(); } \
TEST_F(FIXTURE, Transactions) { TestTransactions(); } \
TEST_F(FIXTURE, ConcurrentStatements) { TestConcurrentStatements(); } \
TEST_F(FIXTURE, ResultInvalidation) { TestResultInvalidation(); }
diff --git a/go/adbc/drivermgr/adbc.h b/go/adbc/drivermgr/adbc.h
index f7ea69af..9d9fa923 100644
--- a/go/adbc/drivermgr/adbc.h
+++ b/go/adbc/drivermgr/adbc.h
@@ -1022,6 +1022,7 @@ AdbcStatusCode AdbcDatabaseNew(struct AdbcDatabase* database, struct AdbcError*
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcDatabaseGetOption(struct AdbcDatabase* database, const char* key,
char* value, size_t* length,
struct AdbcError* error);
@@ -1062,6 +1063,7 @@ AdbcStatusCode AdbcDatabaseGetOption(struct AdbcDatabase* database, const char*
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcDatabaseGetOptionBytes(struct AdbcDatabase* database, const char* key,
uint8_t* value, size_t* length,
struct AdbcError* error);
@@ -1086,6 +1088,7 @@ AdbcStatusCode AdbcDatabaseGetOptionBytes(struct AdbcDatabase* database, const c
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcDatabaseGetOptionDouble(struct AdbcDatabase* database, const char* key,
double* value, struct AdbcError* error);
@@ -1109,6 +1112,7 @@ AdbcStatusCode AdbcDatabaseGetOptionDouble(struct AdbcDatabase* database, const
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcDatabaseGetOptionInt(struct AdbcDatabase* database, const char* key,
int64_t* value, struct AdbcError* error);
@@ -1518,6 +1522,7 @@ AdbcStatusCode AdbcConnectionGetObjects(struct AdbcConnection* connection, int d
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcConnectionGetOption(struct AdbcConnection* connection, const char* key,
char* value, size_t* length,
struct AdbcError* error);
@@ -1558,6 +1563,7 @@ AdbcStatusCode AdbcConnectionGetOption(struct AdbcConnection* connection, const
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcConnectionGetOptionBytes(struct AdbcConnection* connection,
const char* key, uint8_t* value,
size_t* length, struct AdbcError* error);
@@ -1582,6 +1588,7 @@ AdbcStatusCode AdbcConnectionGetOptionBytes(struct AdbcConnection* connection,
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcConnectionGetOptionInt(struct AdbcConnection* connection,
const char* key, int64_t* value,
struct AdbcError* error);
@@ -1606,6 +1613,7 @@ AdbcStatusCode AdbcConnectionGetOptionInt(struct AdbcConnection* connection,
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcConnectionGetOptionDouble(struct AdbcConnection* connection,
const char* key, double* value,
struct AdbcError* error);
@@ -2005,6 +2013,7 @@ AdbcStatusCode AdbcStatementCancel(struct AdbcStatement* statement,
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcStatementGetOption(struct AdbcStatement* statement, const char* key,
char* value, size_t* length,
struct AdbcError* error);
@@ -2045,6 +2054,7 @@ AdbcStatusCode AdbcStatementGetOption(struct AdbcStatement* statement, const cha
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcStatementGetOptionBytes(struct AdbcStatement* statement,
const char* key, uint8_t* value,
size_t* length, struct AdbcError* error);
@@ -2069,6 +2079,7 @@ AdbcStatusCode AdbcStatementGetOptionBytes(struct AdbcStatement* statement,
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcStatementGetOptionInt(struct AdbcStatement* statement, const char* key,
int64_t* value, struct AdbcError* error);
@@ -2092,6 +2103,7 @@ AdbcStatusCode AdbcStatementGetOptionInt(struct AdbcStatement* statement, const
/// \param[out] error An optional location to return an error
/// message if necessary.
/// \return ADBC_STATUS_NOT_FOUND if the option is not recognized.
+ADBC_EXPORT
AdbcStatusCode AdbcStatementGetOptionDouble(struct AdbcStatement* statement,
const char* key, double* value,
struct AdbcError* error);
diff --git a/go/adbc/drivermgr/adbc_driver_manager.cc b/go/adbc/drivermgr/adbc_driver_manager.cc
index 1d35482a..ea4ddc4c 100644
--- a/go/adbc/drivermgr/adbc_driver_manager.cc
+++ b/go/adbc/drivermgr/adbc_driver_manager.cc
@@ -132,23 +132,23 @@ static AdbcStatusCode ReleaseDriver(struct AdbcDriver* driver, struct AdbcError*
AdbcStatusCode DatabaseGetOption(struct AdbcDatabase* database, const char* key,
char* value, size_t* length, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode DatabaseGetOptionBytes(struct AdbcDatabase* database, const char* key,
uint8_t* value, size_t* length,
struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode DatabaseGetOptionInt(struct AdbcDatabase* database, const char* key,
int64_t* value, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode DatabaseGetOptionDouble(struct AdbcDatabase* database, const char* key,
double* value, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode DatabaseSetOption(struct AdbcDatabase* database, const char* key,
@@ -195,24 +195,24 @@ AdbcStatusCode ConnectionGetObjects(struct AdbcConnection*, int, const char*, co
AdbcStatusCode ConnectionGetOption(struct AdbcConnection* connection, const char* key,
char* value, size_t* length, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode ConnectionGetOptionBytes(struct AdbcConnection* connection,
const char* key, uint8_t* value, size_t* length,
struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode ConnectionGetOptionInt(struct AdbcConnection* connection, const char* key,
int64_t* value, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode ConnectionGetOptionDouble(struct AdbcConnection* connection,
const char* key, double* value,
struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode ConnectionGetTableSchema(struct AdbcConnection*, const char*, const char*,
@@ -284,23 +284,23 @@ AdbcStatusCode StatementExecuteSchema(struct AdbcStatement* statement,
AdbcStatusCode StatementGetOption(struct AdbcStatement* statement, const char* key,
char* value, size_t* length, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode StatementGetOptionBytes(struct AdbcStatement* statement, const char* key,
uint8_t* value, size_t* length,
struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode StatementGetOptionInt(struct AdbcStatement* statement, const char* key,
int64_t* value, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode StatementGetOptionDouble(struct AdbcStatement* statement, const char* key,
double* value, struct AdbcError* error) {
- return ADBC_STATUS_NOT_IMPLEMENTED;
+ return ADBC_STATUS_NOT_FOUND;
}
AdbcStatusCode StatementGetParameterSchema(struct AdbcStatement* statement,
@@ -535,11 +535,11 @@ AdbcStatusCode AdbcDatabaseInit(struct AdbcDatabase* database, struct AdbcError*
// So we don't confuse a driver into thinking it's initialized already
database->private_data = nullptr;
if (args->init_func) {
- status = AdbcLoadDriverFromInitFunc(args->init_func, ADBC_VERSION_1_0_0,
+ status = AdbcLoadDriverFromInitFunc(args->init_func, ADBC_VERSION_1_1_0,
database->private_driver, error);
} else {
status = AdbcLoadDriver(args->driver.c_str(), args->entrypoint.c_str(),
- ADBC_VERSION_1_0_0, database->private_driver, error);
+ ADBC_VERSION_1_1_0, database->private_driver, error);
}
if (status != ADBC_STATUS_OK) {
// Restore private_data so it will be released by AdbcDatabaseRelease
@@ -1375,30 +1375,30 @@ AdbcStatusCode AdbcLoadDriverFromInitFunc(AdbcDriverInitFunc init_func, int vers
auto* driver = reinterpret_cast<struct AdbcDriver*>(raw_driver);
FILL_DEFAULT(driver, DatabaseGetOption);
FILL_DEFAULT(driver, DatabaseGetOptionBytes);
- FILL_DEFAULT(driver, DatabaseGetOptionInt);
FILL_DEFAULT(driver, DatabaseGetOptionDouble);
+ FILL_DEFAULT(driver, DatabaseGetOptionInt);
FILL_DEFAULT(driver, DatabaseSetOptionBytes);
- FILL_DEFAULT(driver, DatabaseSetOptionInt);
FILL_DEFAULT(driver, DatabaseSetOptionDouble);
+ FILL_DEFAULT(driver, DatabaseSetOptionInt);
FILL_DEFAULT(driver, ConnectionCancel);
FILL_DEFAULT(driver, ConnectionGetOption);
FILL_DEFAULT(driver, ConnectionGetOptionBytes);
- FILL_DEFAULT(driver, ConnectionGetOptionInt);
FILL_DEFAULT(driver, ConnectionGetOptionDouble);
+ FILL_DEFAULT(driver, ConnectionGetOptionInt);
FILL_DEFAULT(driver, ConnectionSetOptionBytes);
- FILL_DEFAULT(driver, ConnectionSetOptionInt);
FILL_DEFAULT(driver, ConnectionSetOptionDouble);
+ FILL_DEFAULT(driver, ConnectionSetOptionInt);
FILL_DEFAULT(driver, StatementCancel);
FILL_DEFAULT(driver, StatementExecuteSchema);
FILL_DEFAULT(driver, StatementGetOption);
FILL_DEFAULT(driver, StatementGetOptionBytes);
- FILL_DEFAULT(driver, StatementGetOptionInt);
FILL_DEFAULT(driver, StatementGetOptionDouble);
+ FILL_DEFAULT(driver, StatementGetOptionInt);
FILL_DEFAULT(driver, StatementSetOptionBytes);
- FILL_DEFAULT(driver, StatementSetOptionInt);
FILL_DEFAULT(driver, StatementSetOptionDouble);
+ FILL_DEFAULT(driver, StatementSetOptionInt);
}
return ADBC_STATUS_OK;