You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by ko...@apache.org on 2023/03/19 21:30:41 UTC
[arrow-adbc] branch main updated: feat(glib): add gadbc_statement_bind() and ingest related bindings (#528)
This is an automated email from the ASF dual-hosted git repository.
kou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/main by this push:
new 0b2b9d4 feat(glib): add gadbc_statement_bind() and ingest related bindings (#528)
0b2b9d4 is described below
commit 0b2b9d4e9503b54d8d1ca433927d30539f84f8b7
Author: Sutou Kouhei <ko...@clear-code.com>
AuthorDate: Mon Mar 20 06:30:36 2023 +0900
feat(glib): add gadbc_statement_bind() and ingest related bindings (#528)
Fixes #524.
---------
Co-authored-by: David Li <li...@gmail.com>
---
glib/adbc-glib/statement.c | 162 +++++++++++++++++++--
glib/adbc-glib/statement.h | 37 ++++-
glib/adbc-glib/version.h.in | 31 +++-
glib/test/test-statement.rb | 60 ++++++--
ruby/lib/adbc/connection.rb | 30 ++--
ruby/lib/adbc/loader.rb | 1 +
ruby/lib/adbc/statement.rb | 86 +++++++++++
ruby/test/test-connection.rb | 12 +-
.../test/{test-connection.rb => test-statement.rb} | 40 +++--
9 files changed, 389 insertions(+), 70 deletions(-)
diff --git a/glib/adbc-glib/statement.c b/glib/adbc-glib/statement.c
index 8b045b5..fad52ef 100644
--- a/glib/adbc-glib/statement.c
+++ b/glib/adbc-glib/statement.c
@@ -140,7 +140,7 @@ gboolean gadbc_statement_release(GADBCStatement* statement, GError** error) {
* queries expected to be executed repeatedly,
* gadbc_statement_prepare() the statement first.
*
- * Returns: %TRUE if option is set successfully, %FALSE otherwise.
+ * Returns: %TRUE if query is set successfully, %FALSE otherwise.
*
* Since: 0.1.0
*/
@@ -158,9 +158,149 @@ gboolean gadbc_statement_set_sql_query(GADBCStatement* statement, const gchar* q
return gadbc_error_check(error, status_code, &adbc_error, context);
}
+static gboolean gadbc_statement_set_option_internal(GADBCStatement* statement,
+ const gchar* key, const gchar* value,
+ const gchar* context,
+ GError** error) {
+ struct AdbcStatement* adbc_statement =
+ gadbc_statement_get_raw(statement, context, error);
+ if (!adbc_statement) {
+ return FALSE;
+ }
+ struct AdbcError adbc_error = {};
+ AdbcStatusCode status_code =
+ AdbcStatementSetOption(adbc_statement, key, value, &adbc_error);
+ return gadbc_error_check(error, status_code, &adbc_error, context);
+}
+
+/**
+ * gadbc_statement_set_option:
+ * @statement: A #GADBCStatement.
+ * @key: A option key.
+ * @value: A option value.
+ * @error: (out) (optional): Return location for a #GError or %NULL.
+ *
+ * Set a string option on a statement.
+ *
+ * Returns: %TRUE if option is set successfully, %FALSE otherwise.
+ *
+ * Since: 0.4.0
+ */
+gboolean gadbc_statement_set_option(GADBCStatement* statement, const gchar* key,
+ const char* value, GError** error) {
+ const gchar* context = "[adbc][statement][set-option]";
+ return gadbc_statement_set_option_internal(statement, key, value, context, error);
+}
+
+/**
+ * gadbc_statement_set_ingest_target_table:
+ * @statement: A #GADBCStatement.
+ * @table: A table name to be ingested.
+ * @error: (out) (optional): Return location for a #GError or %NULL.
+ *
+ * Set an ingest target table name on a statement.
+ *
+ * Returns: %TRUE if table is set successfully, %FALSE otherwise.
+ *
+ * Since: 0.4.0
+ */
+gboolean gadbc_statement_set_ingest_target_table(GADBCStatement* statement,
+ const gchar* table, GError** error) {
+ const gchar* context = "[adbc][statement][set-ingest-target-table]";
+ return gadbc_statement_set_option_internal(statement, ADBC_INGEST_OPTION_TARGET_TABLE,
+ table, context, error);
+}
+
+/**
+ * gadbc_statement_set_ingest_mode:
+ * @statement: A #GADBCStatement.
+ * @mode: A #GADBCIngestMode.
+ * @error: (out) (optional): Return location for a #GError or %NULL.
+ *
+ * Set an ingest mode on a statement.
+ *
+ * Returns: %TRUE if mode is set successfully, %FALSE otherwise.
+ *
+ * Since: 0.4.0
+ */
+gboolean gadbc_statement_set_ingest_mode(GADBCStatement* statement, GADBCIngestMode mode,
+ GError** error) {
+ const gchar* context = "[adbc][statement][set-ingest-mode]";
+ const gchar* mode_value = ADBC_INGEST_OPTION_MODE_CREATE;
+ switch (mode) {
+ case GADBC_INGEST_MODE_APPEND:
+ mode_value = ADBC_INGEST_OPTION_MODE_APPEND;
+ break;
+ default:
+ break;
+ }
+ return gadbc_statement_set_option_internal(statement, ADBC_INGEST_OPTION_MODE,
+ mode_value, context, error);
+}
+
+/**
+ * gadbc_statement_prepare:
+ * @statement: A #GADBCStatement.
+ * @error: (out) (optional): Return location for a #GError or %NULL.
+ *
+ * Turn this statement into a prepared statement to be
+ * executed multiple times.
+ *
+ * This invalidates any prior result sets.
+ *
+ * Returns: %TRUE if preparation is done successfully, %FALSE
+ * otherwise.
+ *
+ * Since: 0.4.0
+ */
+gboolean gadbc_statement_prepare(GADBCStatement* statement, GError** error) {
+ const gchar* context = "[adbc][statement][prepare]";
+ struct AdbcStatement* adbc_statement =
+ gadbc_statement_get_raw(statement, context, error);
+ if (!adbc_statement) {
+ return FALSE;
+ }
+ struct AdbcError adbc_error = {};
+ AdbcStatusCode status_code = AdbcStatementPrepare(adbc_statement, &adbc_error);
+ return gadbc_error_check(error, status_code, &adbc_error, context);
+}
+
+/**
+ * gadbc_statement_bind:
+ * @statement: A #GADBCStatement.
+ * @c_abi_array: A `struct ArrowArray *` of a record batch to
+ * bind. The driver will call the release callback itself, although
+ * it may not do this until the statement is released.
+ * @c_abi_schema: A `struct ArrowSchema *` of @c_abi_array.
+ * @error: (out) (optional): Return location for a #GError or %NULL.
+ *
+ * Bind Arrow data. This can be used for bulk inserts or prepared
+ * statements.
+ *
+ * Returns: %TRUE if binding is done successfully, %FALSE
+ * otherwise.
+ *
+ * Since: 0.4.0
+ */
+gboolean gadbc_statement_bind(GADBCStatement* statement, gpointer c_abi_array,
+ gpointer c_abi_schema, GError** error) {
+ const gchar* context = "[adbc][statement][bind]";
+ struct AdbcStatement* adbc_statement =
+ gadbc_statement_get_raw(statement, context, error);
+ if (!adbc_statement) {
+ return FALSE;
+ }
+ struct AdbcError adbc_error = {};
+ AdbcStatusCode status_code =
+ AdbcStatementBind(adbc_statement, c_abi_array, c_abi_schema, &adbc_error);
+ return gadbc_error_check(error, status_code, &adbc_error, context);
+}
+
/**
* gadbc_statement_execute:
* @statement: A #GADBCStatement.
+ * @need_result: Whether the results are received by @c_abi_arrray_stream or
+ * not.
* @c_abi_array_stream: (out) (optional): Return location for the execution as
* `struct ArrowArrayStream *`. It should be freed with the
* `ArrowArrayStream::release` callback then g_free() when no longer needed.
@@ -172,30 +312,34 @@ gboolean gadbc_statement_set_sql_query(GADBCStatement* statement, const gchar* q
*
* This invalidates any prior result sets.
*
- * Returns: %TRUE if option is set successfully, %FALSE otherwise.
+ * Returns: %TRUE if execution is done successfully, %FALSE otherwise.
*
* Since: 0.1.0
*/
-gboolean gadbc_statement_execute(GADBCStatement* statement, gpointer* c_abi_array_stream,
- gint64* n_rows_affected, GError** error) {
+gboolean gadbc_statement_execute(GADBCStatement* statement, gboolean need_result,
+ gpointer* c_abi_array_stream, gint64* n_rows_affected,
+ GError** error) {
const gchar* context = "[adbc][statement][execute]";
+ if (c_abi_array_stream) {
+ *c_abi_array_stream = NULL;
+ }
struct AdbcStatement* adbc_statement =
gadbc_statement_get_raw(statement, context, error);
if (!adbc_statement) {
return FALSE;
}
struct ArrowArrayStream* array_stream = NULL;
- if (c_abi_array_stream) {
+ if (need_result && c_abi_array_stream) {
array_stream = g_new0(struct ArrowArrayStream, 1);
}
struct AdbcError adbc_error = {};
AdbcStatusCode status_code = AdbcStatementExecuteQuery(adbc_statement, array_stream,
n_rows_affected, &adbc_error);
gboolean success = gadbc_error_check(error, status_code, &adbc_error, context);
- if (success) {
- *c_abi_array_stream = array_stream;
- } else {
- if (array_stream) {
+ if (array_stream) {
+ if (success) {
+ *c_abi_array_stream = array_stream;
+ } else {
g_free(array_stream);
}
}
diff --git a/glib/adbc-glib/statement.h b/glib/adbc-glib/statement.h
index fadcb11..0b1de36 100644
--- a/glib/adbc-glib/statement.h
+++ b/glib/adbc-glib/statement.h
@@ -23,6 +23,24 @@
G_BEGIN_DECLS
+/**
+ * GADBCIngestMode:
+ * @GADBC_INGEST_MODE_CREATE: Create the table and insert data;
+ * error if the table exists.
+ * @GADBC_INGEST_MODE_APPEND: Do not create the table, and insert
+ * data; error if the table does not exist (%GADBC_ERROR_NOT_FOUND)
+ * or does not match the schema of the data to append
+ * (%GADBC_ERROR_ALREADY_EXISTS).
+ *
+ * Whether to create (the default) or append on bulk insert.
+ *
+ * Since: 0.4.0
+ */
+typedef enum {
+ GADBC_INGEST_MODE_CREATE,
+ GADBC_INGEST_MODE_APPEND,
+} GADBCIngestMode;
+
#define GADBC_TYPE_STATEMENT (gadbc_statement_get_type())
G_DECLARE_DERIVABLE_TYPE(GADBCStatement, gadbc_statement, GADBC, STATEMENT, GObject)
struct _GADBCStatementClass {
@@ -36,8 +54,23 @@ gboolean gadbc_statement_release(GADBCStatement* statement, GError** error);
GADBC_AVAILABLE_IN_0_1
gboolean gadbc_statement_set_sql_query(GADBCStatement* statement, const gchar* query,
GError** error);
+GADBC_AVAILABLE_IN_0_4
+gboolean gadbc_statement_set_option(GADBCStatement* statement, const gchar* key,
+ const gchar* value, GError** error);
+GADBC_AVAILABLE_IN_0_4
+gboolean gadbc_statement_set_ingest_target_table(GADBCStatement* statement,
+ const gchar* table, GError** error);
+GADBC_AVAILABLE_IN_0_4
+gboolean gadbc_statement_set_ingest_mode(GADBCStatement* statement, GADBCIngestMode mode,
+ GError** error);
+GADBC_AVAILABLE_IN_0_4
+gboolean gadbc_statement_prepare(GADBCStatement* statement, GError** error);
+GADBC_AVAILABLE_IN_0_4
+gboolean gadbc_statement_bind(GADBCStatement* statement, gpointer c_abi_array,
+ gpointer c_abi_schema, GError** error);
GADBC_AVAILABLE_IN_0_1
-gboolean gadbc_statement_execute(GADBCStatement* statement, gpointer* c_abi_array_stream,
- gint64* n_rows_affected, GError** error);
+gboolean gadbc_statement_execute(GADBCStatement* statement, gboolean need_result,
+ gpointer* c_abi_array_stream, gint64* n_rows_affected,
+ GError** error);
G_END_DECLS
diff --git a/glib/adbc-glib/version.h.in b/glib/adbc-glib/version.h.in
index c67cde8..bf75cfc 100644
--- a/glib/adbc-glib/version.h.in
+++ b/glib/adbc-glib/version.h.in
@@ -100,6 +100,15 @@
# define GADBC_UNAVAILABLE(major, minor) G_UNAVAILABLE(major, minor)
#endif
+/**
+ * GADBC_VERSION_0_4:
+ *
+ * You can use this macro value for compile time API version check.
+ *
+ * Since: 0.4.0
+ */
+#define GADBC_VERSION_0_4 G_ENCODE_VERSION(0, 4)
+
/**
* GADBC_VERSION_0_1:
*
@@ -156,12 +165,26 @@
#define GADBC_AVAILABLE_IN_ALL
+#if GADBC_VERSION_MIN_REQUIRED >= GADBC_VERSION_0_4
+# define GADBC_DEPRECATED_IN_0_4 GADBC_DEPRECATED
+# define GADBC_DEPRECATED_IN_0_4_FOR(function) GADBC_DEPRECATED_FOR(function)
+#else
+# define GADBC_DEPRECATED_IN_0_4
+# define GADBC_DEPRECATED_IN_0_4_FOR(function)
+#endif
+
+#if GADBC_VERSION_MAX_ALLOWED < GADBC_VERSION_0_4
+# define GADBC_AVAILABLE_IN_0_4 GADBC_UNAVAILABLE(0, 4)
+#else
+# define GADBC_AVAILABLE_IN_0_4
+#endif
+
#if GADBC_VERSION_MIN_REQUIRED >= GADBC_VERSION_0_1
-# define GADBC_DEPRECATED_IN_1_0 GADBC_DEPRECATED
-# define GADBC_DEPRECATED_IN_1_0_FOR(function) GADBC_DEPRECATED_FOR(function)
+# define GADBC_DEPRECATED_IN_0_1 GADBC_DEPRECATED
+# define GADBC_DEPRECATED_IN_0_1_FOR(function) GADBC_DEPRECATED_FOR(function)
#else
-# define GADBC_DEPRECATED_IN_1_0
-# define GADBC_DEPRECATED_IN_1_0_FOR(function)
+# define GADBC_DEPRECATED_IN_0_1
+# define GADBC_DEPRECATED_IN_0_1_FOR(function)
#endif
#if GADBC_VERSION_MAX_ALLOWED < GADBC_VERSION_0_1
diff --git a/glib/test/test-statement.rb b/glib/test/test-statement.rb
index 2a3f11c..242b4f5 100644
--- a/glib/test/test-statement.rb
+++ b/glib/test/test-statement.rb
@@ -23,25 +23,63 @@ class StatementTest < Test::Unit::TestCase
@database.init
@connection = ADBC::Connection.new
@connection.init(@database)
+ @statement = ADBC::Statement.new(@connection)
end
- def test_release
- statement = ADBC::Statement.new(@connection)
- assert do
- statement.release
+ def teardown
+ @statement.release
+ end
+
+ def execute_statement(need_result: true)
+ _, c_abi_array_stream, n_rows_affected = @statement.execute(need_result)
+ begin
+ if need_result
+ reader = Arrow::RecordBatchReader.import(c_abi_array_stream)
+ table = reader.read_all
+ yield(table, n_rows_affected) if block_given?
+ else
+ yield(n_rows_affected) if block_given?
+ end
+ ensure
+ GLib.free(c_abi_array_stream) if need_result
end
end
def test_execute
- statement = ADBC::Statement.new(@connection)
- statement.set_sql_query("SELECT 1")
- success, c_abi_array_stream, n_rows_affected = statement.execute
- begin
- reader = Arrow::RecordBatchReader.import(c_abi_array_stream)
+ @statement.set_sql_query("SELECT 1")
+ execute_statement do |table, _n_rows_affected|
assert_equal(Arrow::Table.new("1" => Arrow::Int64Array.new([1])),
- reader.read_all)
+ table)
+ end
+ end
+
+ def test_bind
+ @statement.set_sql_query("CREATE TABLE data (number int)")
+ execute_statement
+
+ record_batch =
+ Arrow::RecordBatch.new(number: Arrow::Int64Array.new([10, 20, 30]))
+ @statement.set_sql_query("INSERT INTO data VALUES (?)")
+ @statement.ingest_target_table = "data"
+ @statement.ingest_mode = :append
+ _, c_abi_array, c_abi_schema = record_batch.export
+ begin
+ @statement.bind(c_abi_array, c_abi_schema)
+ execute_statement(need_result: false) do |n_rows_affected|
+ assert_equal(3, n_rows_affected)
+ end
ensure
- GLib.free(c_abi_array_stream)
+ begin
+ GLib.free(c_abi_array)
+ ensure
+ GLib.free(c_abi_schema)
+ end
+ end
+
+ @statement.set_sql_query("SELECT * FROM data")
+ execute_statement do |table, _n_rows_affected|
+ assert_equal(record_batch.to_table,
+ table)
end
end
end
diff --git a/ruby/lib/adbc/connection.rb b/ruby/lib/adbc/connection.rb
index d859565..eeed95c 100644
--- a/ruby/lib/adbc/connection.rb
+++ b/ruby/lib/adbc/connection.rb
@@ -17,23 +17,19 @@
module ADBC
class Connection
- def query(sql)
- statement = Statement.new(self)
- begin
- statement.set_sql_query(sql)
- _, c_abi_array_stream, n_rows_affected = statement.execute
- begin
- reader = Arrow::RecordBatchReader.import(c_abi_array_stream)
- if block_given?
- yield(reader, n_rows_affected)
- else
- reader.read_all
- end
- ensure
- GLib.free(c_abi_array_stream)
- end
- ensure
- statement.release
+ def open_statement(&block)
+ Statement.open(self, &block)
+ end
+
+ def query(sql, &block)
+ open_statement do |statement|
+ statement.query(sql, &block)
+ end
+ end
+
+ def ingest(table_name, values, mode: :create)
+ open_statment do |statement|
+ statement.ingest(table_name, values, mode: mode)
end
end
end
diff --git a/ruby/lib/adbc/loader.rb b/ruby/lib/adbc/loader.rb
index bca2190..280dae5 100644
--- a/ruby/lib/adbc/loader.rb
+++ b/ruby/lib/adbc/loader.rb
@@ -31,6 +31,7 @@ module ADBC
def require_libraries
require_relative "connection"
require_relative "database"
+ require_relative "statement"
end
end
end
diff --git a/ruby/lib/adbc/statement.rb b/ruby/lib/adbc/statement.rb
new file mode 100644
index 0000000..ca3dfcf
--- /dev/null
+++ b/ruby/lib/adbc/statement.rb
@@ -0,0 +1,86 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+module ADBC
+ class Statement
+ class << self
+ def open(connection)
+ statement = new(connection)
+ if block_given?
+ begin
+ yield(statement)
+ ensure
+ statement.release
+ end
+ else
+ statement
+ end
+ end
+ end
+
+ alias_method :execute_raw, :execute
+ def execute(need_result: true)
+ _, c_abi_array_stream, n_rows_affected = execute_raw(need_result)
+ if need_result
+ begin
+ reader = Arrow::RecordBatchReader.import(c_abi_array_stream)
+ if block_given?
+ yield(reader, n_rows_affected)
+ else
+ [reader.read_all, n_rows_affected]
+ end
+ ensure
+ GLib.free(c_abi_array_stream)
+ end
+ else
+ if block_given?
+ yield(n_rows_affected)
+ else
+ n_rows_affected
+ end
+ end
+ end
+
+ def ingest(table_name, values, mode: :create)
+ insert = "INSERT INTO #{table_name} (" # TODO escape
+ fields = values.schema.fields
+ insert << fields.collect(&:name).join(", ")
+ insert << ") VALUES ("
+ insert << (["?"] * fields.size).join(", ")
+ insert << ")"
+ self.sql_query = insert
+ self.ingest_target_table = table_name
+ self.ingest_mode = mode
+ _, c_abi_array, c_abi_schema = values.export
+ begin
+ bind(c_abi_array, c_abi_schema)
+ execute(need_result: false)
+ ensure
+ begin
+ GLib.free(c_abi_array)
+ ensure
+ GLib.free(c_abi_schema)
+ end
+ end
+ end
+
+ def query(sql, &block)
+ self.sql_query = sql
+ execute(&block)
+ end
+ end
+end
diff --git a/ruby/test/test-connection.rb b/ruby/test/test-connection.rb
index e76a897..2568035 100644
--- a/ruby/test/test-connection.rb
+++ b/ruby/test/test-connection.rb
@@ -32,22 +32,24 @@ class ConnectionTest < Test::Unit::TestCase
sub_test_case("#query") do
def test_block
- @connection.query("SELECT 1") do |reader, n_affected_rows|
+ @connection.query("SELECT 1") do |reader, n_rows_affected|
assert_equal([
Arrow::Table.new("1" => Arrow::Int64Array.new([1])),
-1,
],
[
reader.read_all,
- n_affected_rows,
+ n_rows_affected,
])
end
end
def test_no_block
- table = @connection.query("SELECT 1")
- assert_equal(Arrow::Table.new("1" => Arrow::Int64Array.new([1])),
- table)
+ assert_equal([
+ Arrow::Table.new("1" => Arrow::Int64Array.new([1])),
+ -1,
+ ],
+ @connection.query("SELECT 1"))
end
end
end
diff --git a/ruby/test/test-connection.rb b/ruby/test/test-statement.rb
similarity index 60%
copy from ruby/test/test-connection.rb
copy to ruby/test/test-statement.rb
index e76a897..d6f7075 100644
--- a/ruby/test/test-connection.rb
+++ b/ruby/test/test-statement.rb
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-class ConnectionTest < Test::Unit::TestCase
+class StatementTest < Test::Unit::TestCase
def setup
options = {
driver: "adbc_driver_sqlite",
@@ -24,30 +24,26 @@ class ConnectionTest < Test::Unit::TestCase
ADBC::Database.open(**options) do |database|
connect_options = {}
database.connect(**connect_options) do |connection|
- @connection = connection
- yield
+ connection.open_statement do |statement|
+ @statement = statement
+ yield
+ end
end
end
end
- sub_test_case("#query") do
- def test_block
- @connection.query("SELECT 1") do |reader, n_affected_rows|
- assert_equal([
- Arrow::Table.new("1" => Arrow::Int64Array.new([1])),
- -1,
- ],
- [
- reader.read_all,
- n_affected_rows,
- ])
- end
- end
-
- def test_no_block
- table = @connection.query("SELECT 1")
- assert_equal(Arrow::Table.new("1" => Arrow::Int64Array.new([1])),
- table)
- end
+ def test_ingest
+ numbers = Arrow::Int64Array.new([10, 20, 30])
+ record_batch = Arrow::RecordBatch.new(number: numbers)
+ @statement.ingest("data", record_batch)
+ table, n_rows_affected = @statement.query("SELECT * FROM data")
+ assert_equal([
+ Arrow::Table.new(number: numbers),
+ -1,
+ ],
+ [
+ table,
+ n_rows_affected,
+ ])
end
end