You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2022/09/28 14:54:33 UTC
[GitHub] [arrow] lidavidm opened a new pull request, #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
lidavidm opened a new pull request, #14266:
URL: https://github.com/apache/arrow/pull/14266
This allows us to send multiple parameters and have them all executed, instead of having to do multiple round trips per row - this technically existed in the API but the sample server didn't implement it properly
Also contains misc cleanups on the API, and removes client_test.cc since the mocks are brittle
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on a diff in pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
pitrou commented on code in PR #14266:
URL: https://github.com/apache/arrow/pull/14266#discussion_r1009667423
##########
cpp/src/arrow/flight/sql/example/sqlite_statement.cc:
##########
@@ -160,10 +169,100 @@ arrow::Result<int> SqliteStatement::Reset() {
sqlite3_stmt* SqliteStatement::GetSqlite3Stmt() const { return stmt_; }
arrow::Result<int64_t> SqliteStatement::ExecuteUpdate() {
- ARROW_RETURN_NOT_OK(Step());
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(int rc, Step());
+ if (rc == SQLITE_DONE) break;
+ }
return sqlite3_changes(db_);
}
+Status SqliteStatement::SetParameters(
+ std::vector<std::shared_ptr<arrow::RecordBatch>> parameters) {
+ const int num_params = sqlite3_bind_parameter_count(stmt_);
+ for (const auto& batch : parameters) {
+ if (batch->num_columns() != num_params) {
+ return Status::Invalid("Expected ", num_params, " parameters, but got ",
+ batch->num_columns());
+ }
+ }
+ parameters_ = std::move(parameters);
+ auto end = std::remove_if(
+ parameters_.begin(), parameters_.end(),
+ [](const std::shared_ptr<RecordBatch>& batch) { return batch->num_rows() == 0; });
+ parameters_.erase(end, parameters_.end());
+ return Status::OK();
+}
+
+Status SqliteStatement::Bind(size_t batch_index, int64_t row_index) {
+ if (batch_index >= parameters_.size()) {
+ return Status::IndexError("Cannot bind to batch ", batch_index);
+ }
+ const RecordBatch& batch = *parameters_[batch_index];
+ if (row_index < 0 || row_index >= batch.num_rows()) {
+ return Status::IndexError("Cannot bind to row ", row_index, " in batch ",
+ batch_index);
+ }
+
+ if (sqlite3_clear_bindings(stmt_) != SQLITE_OK) {
+ return Status::Invalid("Failed to reset bindings: ", sqlite3_errmsg(db_));
+ }
+ for (int c = 0; c < batch.num_columns(); ++c) {
+ Array* column = batch.column(c).get();
+ int64_t column_index = row_index;
+ if (column->type_id() == Type::DENSE_UNION) {
+ // Allow polymorphic bindings via union
Review Comment:
(I realize this is still classified as example, though, so perhaps you don't want to maximize type coverage)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on a diff in pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
pitrou commented on code in PR #14266:
URL: https://github.com/apache/arrow/pull/14266#discussion_r1005878920
##########
cpp/src/arrow/flight/sql/server_test.cc:
##########
@@ -502,51 +489,53 @@ TEST_F(TestFlightSqlServer, TestCommandPreparedStatementQueryWithParameterBindin
auto prepared_statement,
sql_client->Prepare({}, "SELECT * FROM intTable WHERE keyName LIKE ?"));
- auto parameter_schema = prepared_statement->parameter_schema();
-
+ const std::shared_ptr<Schema>& parameter_schema =
+ prepared_statement->parameter_schema();
const std::shared_ptr<Schema>& expected_parameter_schema =
arrow::schema({arrow::field("parameter_1", example::GetUnknownColumnDataType())});
+ ASSERT_NO_FATAL_FAILURE(AssertSchemaEqual(expected_parameter_schema, parameter_schema));
- AssertSchemaEqual(expected_parameter_schema, parameter_schema);
-
- std::shared_ptr<Array> type_ids = ArrayFromJSON(int8(), R"([0])");
- std::shared_ptr<Array> offsets = ArrayFromJSON(int32(), R"([0])");
- std::shared_ptr<Array> string_array = ArrayFromJSON(utf8(), R"(["%one"])");
- std::shared_ptr<Array> bytes_array = ArrayFromJSON(binary(), R"([])");
- std::shared_ptr<Array> bigint_array = ArrayFromJSON(int64(), R"([])");
- std::shared_ptr<Array> double_array = ArrayFromJSON(float64(), R"([])");
-
- ASSERT_OK_AND_ASSIGN(
- auto parameter_1_array,
- DenseUnionArray::Make(*type_ids, *offsets,
- {string_array, bytes_array, bigint_array, double_array},
- {"string", "bytes", "bigint", "double"}, {0, 1, 2, 3}));
-
- const std::shared_ptr<RecordBatch>& record_batch =
- RecordBatch::Make(parameter_schema, 1, {parameter_1_array});
-
- ASSERT_OK(prepared_statement->SetParameters(record_batch));
+ auto record_batch = RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
ASSERT_OK_AND_ASSIGN(auto flight_info, prepared_statement->Execute());
-
ASSERT_OK_AND_ASSIGN(auto stream,
sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
-
ASSERT_OK_AND_ASSIGN(auto table, stream->ToTable());
const std::shared_ptr<Schema>& expected_schema =
arrow::schema({arrow::field("id", int64()), arrow::field("keyName", utf8()),
arrow::field("value", int64()), arrow::field("foreignId", int64())});
- const auto id_array = ArrayFromJSON(int64(), R"([1, 3])");
- const auto keyname_array = ArrayFromJSON(utf8(), R"(["one", "negative one"])");
- const auto value_array = ArrayFromJSON(int64(), R"([1, -1])");
- const auto foreignId_array = ArrayFromJSON(int64(), R"([1, 1])");
-
- const std::shared_ptr<Table>& expected_table = Table::Make(
- expected_schema, {id_array, keyname_array, value_array, foreignId_array});
-
- AssertTablesEqual(*expected_table, *table);
+ auto expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1]
+ ])"});
+ ASSERT_NO_FATAL_FAILURE(AssertTablesEqual(*expected_table, *table, /*verbose=*/true));
+
+ // Set multiple parameters at once
+ record_batch =
+ RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]], [[0, "%zero"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
+ ASSERT_OK_AND_ASSIGN(flight_info, prepared_statement->Execute());
+ ASSERT_OK_AND_ASSIGN(stream, sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
+ ASSERT_OK_AND_ASSIGN(table, stream->ToTable());
+ expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1],
+ [2, "zero", 0, 1]
Review Comment:
Not really, since it would switch to an actual resolved builder at the first non-null value encountered.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #14266:
URL: https://github.com/apache/arrow/pull/14266#issuecomment-1261213010
:warning: Ticket **has no components in JIRA**, make sure you assign one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
lidavidm commented on PR #14266:
URL: https://github.com/apache/arrow/pull/14266#issuecomment-1290642347
@jduo or @pitrou, would one of you be able to review this sometime?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] ursabot commented on pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
ursabot commented on PR #14266:
URL: https://github.com/apache/arrow/pull/14266#issuecomment-1299285759
Benchmark runs are scheduled for baseline = e1303aacc1a0e3a01ee3ccf9fab93de18a0d2d2a and contender = 85ae0d2e83a31a1a3e99c30c43e905c0fbdd0791. 85ae0d2e83a31a1a3e99c30c43e905c0fbdd0791 is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Finished :arrow_down:0.0% :arrow_up:0.0%] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/f4251639dec9487e9ee52c36560c15ce...dbc5a65fadcb4f2b9380f0e6bf439d28/)
[Failed :arrow_down:0.0% :arrow_up:0.0%] [test-mac-arm](https://conbench.ursa.dev/compare/runs/4f0f4ca60a074f4385eccf314ad5e015...497c6a64375c4e1989d8f64b9a5275b9/)
[Finished :arrow_down:0.0% :arrow_up:0.0%] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/ec2e6b7081b94766b785633d2c947ce1...f6abecd2d5a3423d95163940651628cd/)
[Finished :arrow_down:0.29% :arrow_up:0.04%] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/fd816083f3734d399c4c76cb53d28dd8...9975cf2f18b94a64b12715285fbe687b/)
Buildkite builds:
[Finished] [`85ae0d2e` ec2-t3-xlarge-us-east-2](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ec2-t3-xlarge-us-east-2/builds/1801)
[Failed] [`85ae0d2e` test-mac-arm](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-test-mac-arm/builds/1822)
[Finished] [`85ae0d2e` ursa-i9-9960x](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-i9-9960x/builds/1789)
[Finished] [`85ae0d2e` ursa-thinkcentre-m75q](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-thinkcentre-m75q/builds/1814)
[Finished] [`e1303aac` ec2-t3-xlarge-us-east-2](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ec2-t3-xlarge-us-east-2/builds/1800)
[Failed] [`e1303aac` test-mac-arm](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-test-mac-arm/builds/1821)
[Finished] [`e1303aac` ursa-i9-9960x](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-i9-9960x/builds/1788)
[Finished] [`e1303aac` ursa-thinkcentre-m75q](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-thinkcentre-m75q/builds/1813)
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #14266:
URL: https://github.com/apache/arrow/pull/14266#issuecomment-1261212978
https://issues.apache.org/jira/browse/ARROW-17867
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on a diff in pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #14266:
URL: https://github.com/apache/arrow/pull/14266#discussion_r1005873000
##########
cpp/src/arrow/flight/sql/server_test.cc:
##########
@@ -502,51 +489,53 @@ TEST_F(TestFlightSqlServer, TestCommandPreparedStatementQueryWithParameterBindin
auto prepared_statement,
sql_client->Prepare({}, "SELECT * FROM intTable WHERE keyName LIKE ?"));
- auto parameter_schema = prepared_statement->parameter_schema();
-
+ const std::shared_ptr<Schema>& parameter_schema =
+ prepared_statement->parameter_schema();
const std::shared_ptr<Schema>& expected_parameter_schema =
arrow::schema({arrow::field("parameter_1", example::GetUnknownColumnDataType())});
+ ASSERT_NO_FATAL_FAILURE(AssertSchemaEqual(expected_parameter_schema, parameter_schema));
- AssertSchemaEqual(expected_parameter_schema, parameter_schema);
-
- std::shared_ptr<Array> type_ids = ArrayFromJSON(int8(), R"([0])");
- std::shared_ptr<Array> offsets = ArrayFromJSON(int32(), R"([0])");
- std::shared_ptr<Array> string_array = ArrayFromJSON(utf8(), R"(["%one"])");
- std::shared_ptr<Array> bytes_array = ArrayFromJSON(binary(), R"([])");
- std::shared_ptr<Array> bigint_array = ArrayFromJSON(int64(), R"([])");
- std::shared_ptr<Array> double_array = ArrayFromJSON(float64(), R"([])");
-
- ASSERT_OK_AND_ASSIGN(
- auto parameter_1_array,
- DenseUnionArray::Make(*type_ids, *offsets,
- {string_array, bytes_array, bigint_array, double_array},
- {"string", "bytes", "bigint", "double"}, {0, 1, 2, 3}));
-
- const std::shared_ptr<RecordBatch>& record_batch =
- RecordBatch::Make(parameter_schema, 1, {parameter_1_array});
-
- ASSERT_OK(prepared_statement->SetParameters(record_batch));
+ auto record_batch = RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
ASSERT_OK_AND_ASSIGN(auto flight_info, prepared_statement->Execute());
-
ASSERT_OK_AND_ASSIGN(auto stream,
sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
-
ASSERT_OK_AND_ASSIGN(auto table, stream->ToTable());
const std::shared_ptr<Schema>& expected_schema =
arrow::schema({arrow::field("id", int64()), arrow::field("keyName", utf8()),
arrow::field("value", int64()), arrow::field("foreignId", int64())});
- const auto id_array = ArrayFromJSON(int64(), R"([1, 3])");
- const auto keyname_array = ArrayFromJSON(utf8(), R"(["one", "negative one"])");
- const auto value_array = ArrayFromJSON(int64(), R"([1, -1])");
- const auto foreignId_array = ArrayFromJSON(int64(), R"([1, 1])");
-
- const std::shared_ptr<Table>& expected_table = Table::Make(
- expected_schema, {id_array, keyname_array, value_array, foreignId_array});
-
- AssertTablesEqual(*expected_table, *table);
+ auto expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1]
+ ])"});
+ ASSERT_NO_FATAL_FAILURE(AssertTablesEqual(*expected_table, *table, /*verbose=*/true));
+
+ // Set multiple parameters at once
+ record_batch =
+ RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]], [[0, "%zero"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
+ ASSERT_OK_AND_ASSIGN(flight_info, prepared_statement->Execute());
+ ASSERT_OK_AND_ASSIGN(stream, sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
+ ASSERT_OK_AND_ASSIGN(table, stream->ToTable());
+ expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1],
+ [2, "zero", 0, 1]
Review Comment:
Ah, well, that would handle nulls, but you could mix values of different types within a column as well…and I can imagine some people want errors on mixed types, others want upcasting/coercion, others would be happy with nulls, and maybe one person wants a dense union :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on a diff in pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
pitrou commented on code in PR #14266:
URL: https://github.com/apache/arrow/pull/14266#discussion_r1005865068
##########
cpp/src/arrow/flight/sql/server_test.cc:
##########
@@ -502,51 +489,53 @@ TEST_F(TestFlightSqlServer, TestCommandPreparedStatementQueryWithParameterBindin
auto prepared_statement,
sql_client->Prepare({}, "SELECT * FROM intTable WHERE keyName LIKE ?"));
- auto parameter_schema = prepared_statement->parameter_schema();
-
+ const std::shared_ptr<Schema>& parameter_schema =
+ prepared_statement->parameter_schema();
const std::shared_ptr<Schema>& expected_parameter_schema =
arrow::schema({arrow::field("parameter_1", example::GetUnknownColumnDataType())});
+ ASSERT_NO_FATAL_FAILURE(AssertSchemaEqual(expected_parameter_schema, parameter_schema));
- AssertSchemaEqual(expected_parameter_schema, parameter_schema);
-
- std::shared_ptr<Array> type_ids = ArrayFromJSON(int8(), R"([0])");
- std::shared_ptr<Array> offsets = ArrayFromJSON(int32(), R"([0])");
- std::shared_ptr<Array> string_array = ArrayFromJSON(utf8(), R"(["%one"])");
- std::shared_ptr<Array> bytes_array = ArrayFromJSON(binary(), R"([])");
- std::shared_ptr<Array> bigint_array = ArrayFromJSON(int64(), R"([])");
- std::shared_ptr<Array> double_array = ArrayFromJSON(float64(), R"([])");
-
- ASSERT_OK_AND_ASSIGN(
- auto parameter_1_array,
- DenseUnionArray::Make(*type_ids, *offsets,
- {string_array, bytes_array, bigint_array, double_array},
- {"string", "bytes", "bigint", "double"}, {0, 1, 2, 3}));
-
- const std::shared_ptr<RecordBatch>& record_batch =
- RecordBatch::Make(parameter_schema, 1, {parameter_1_array});
-
- ASSERT_OK(prepared_statement->SetParameters(record_batch));
+ auto record_batch = RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
ASSERT_OK_AND_ASSIGN(auto flight_info, prepared_statement->Execute());
-
ASSERT_OK_AND_ASSIGN(auto stream,
sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
-
ASSERT_OK_AND_ASSIGN(auto table, stream->ToTable());
const std::shared_ptr<Schema>& expected_schema =
arrow::schema({arrow::field("id", int64()), arrow::field("keyName", utf8()),
arrow::field("value", int64()), arrow::field("foreignId", int64())});
- const auto id_array = ArrayFromJSON(int64(), R"([1, 3])");
- const auto keyname_array = ArrayFromJSON(utf8(), R"(["one", "negative one"])");
- const auto value_array = ArrayFromJSON(int64(), R"([1, -1])");
- const auto foreignId_array = ArrayFromJSON(int64(), R"([1, 1])");
-
- const std::shared_ptr<Table>& expected_table = Table::Make(
- expected_schema, {id_array, keyname_array, value_array, foreignId_array});
-
- AssertTablesEqual(*expected_table, *table);
+ auto expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1]
+ ])"});
+ ASSERT_NO_FATAL_FAILURE(AssertTablesEqual(*expected_table, *table, /*verbose=*/true));
+
+ // Set multiple parameters at once
+ record_batch =
+ RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]], [[0, "%zero"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
+ ASSERT_OK_AND_ASSIGN(flight_info, prepared_statement->Execute());
+ ASSERT_OK_AND_ASSIGN(stream, sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
+ ASSERT_OK_AND_ASSIGN(table, stream->ToTable());
+ expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1],
+ [2, "zero", 0, 1]
Review Comment:
I would say you mainly need some kind of "deferred builder" that just counts nulls at the start, and switches to an actual builder once a non-null value is encountered (or switches to a null array at the end). Perhaps you want to abstract that in a SQLiteColumnBuilder or something.
But, yes, more custom code to write :-)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on a diff in pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #14266:
URL: https://github.com/apache/arrow/pull/14266#discussion_r1009852379
##########
cpp/src/arrow/flight/sql/example/sqlite_statement.cc:
##########
@@ -160,10 +169,100 @@ arrow::Result<int> SqliteStatement::Reset() {
sqlite3_stmt* SqliteStatement::GetSqlite3Stmt() const { return stmt_; }
arrow::Result<int64_t> SqliteStatement::ExecuteUpdate() {
- ARROW_RETURN_NOT_OK(Step());
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(int rc, Step());
+ if (rc == SQLITE_DONE) break;
+ }
return sqlite3_changes(db_);
}
+Status SqliteStatement::SetParameters(
+ std::vector<std::shared_ptr<arrow::RecordBatch>> parameters) {
+ const int num_params = sqlite3_bind_parameter_count(stmt_);
+ for (const auto& batch : parameters) {
+ if (batch->num_columns() != num_params) {
+ return Status::Invalid("Expected ", num_params, " parameters, but got ",
+ batch->num_columns());
+ }
+ }
+ parameters_ = std::move(parameters);
+ auto end = std::remove_if(
+ parameters_.begin(), parameters_.end(),
+ [](const std::shared_ptr<RecordBatch>& batch) { return batch->num_rows() == 0; });
+ parameters_.erase(end, parameters_.end());
+ return Status::OK();
+}
+
+Status SqliteStatement::Bind(size_t batch_index, int64_t row_index) {
+ if (batch_index >= parameters_.size()) {
+ return Status::IndexError("Cannot bind to batch ", batch_index);
+ }
+ const RecordBatch& batch = *parameters_[batch_index];
+ if (row_index < 0 || row_index >= batch.num_rows()) {
+ return Status::IndexError("Cannot bind to row ", row_index, " in batch ",
+ batch_index);
+ }
+
+ if (sqlite3_clear_bindings(stmt_) != SQLITE_OK) {
+ return Status::Invalid("Failed to reset bindings: ", sqlite3_errmsg(db_));
+ }
+ for (int c = 0; c < batch.num_columns(); ++c) {
+ Array* column = batch.column(c).get();
+ int64_t column_index = row_index;
+ if (column->type_id() == Type::DENSE_UNION) {
+ // Allow polymorphic bindings via union
Review Comment:
Flight SQL returns the types of the parameters when you prepare a statement, and this server only ever returns dense union (since SQLite is untyped); in #14082 however the ADBC test suite uses ints specifically and it's easy to support both dense union and int
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on a diff in pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #14266:
URL: https://github.com/apache/arrow/pull/14266#discussion_r1005845228
##########
cpp/src/arrow/flight/sql/server_test.cc:
##########
@@ -502,51 +489,53 @@ TEST_F(TestFlightSqlServer, TestCommandPreparedStatementQueryWithParameterBindin
auto prepared_statement,
sql_client->Prepare({}, "SELECT * FROM intTable WHERE keyName LIKE ?"));
- auto parameter_schema = prepared_statement->parameter_schema();
-
+ const std::shared_ptr<Schema>& parameter_schema =
+ prepared_statement->parameter_schema();
const std::shared_ptr<Schema>& expected_parameter_schema =
arrow::schema({arrow::field("parameter_1", example::GetUnknownColumnDataType())});
+ ASSERT_NO_FATAL_FAILURE(AssertSchemaEqual(expected_parameter_schema, parameter_schema));
- AssertSchemaEqual(expected_parameter_schema, parameter_schema);
-
- std::shared_ptr<Array> type_ids = ArrayFromJSON(int8(), R"([0])");
- std::shared_ptr<Array> offsets = ArrayFromJSON(int32(), R"([0])");
- std::shared_ptr<Array> string_array = ArrayFromJSON(utf8(), R"(["%one"])");
- std::shared_ptr<Array> bytes_array = ArrayFromJSON(binary(), R"([])");
- std::shared_ptr<Array> bigint_array = ArrayFromJSON(int64(), R"([])");
- std::shared_ptr<Array> double_array = ArrayFromJSON(float64(), R"([])");
-
- ASSERT_OK_AND_ASSIGN(
- auto parameter_1_array,
- DenseUnionArray::Make(*type_ids, *offsets,
- {string_array, bytes_array, bigint_array, double_array},
- {"string", "bytes", "bigint", "double"}, {0, 1, 2, 3}));
-
- const std::shared_ptr<RecordBatch>& record_batch =
- RecordBatch::Make(parameter_schema, 1, {parameter_1_array});
-
- ASSERT_OK(prepared_statement->SetParameters(record_batch));
+ auto record_batch = RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
ASSERT_OK_AND_ASSIGN(auto flight_info, prepared_statement->Execute());
-
ASSERT_OK_AND_ASSIGN(auto stream,
sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
-
ASSERT_OK_AND_ASSIGN(auto table, stream->ToTable());
const std::shared_ptr<Schema>& expected_schema =
arrow::schema({arrow::field("id", int64()), arrow::field("keyName", utf8()),
arrow::field("value", int64()), arrow::field("foreignId", int64())});
- const auto id_array = ArrayFromJSON(int64(), R"([1, 3])");
- const auto keyname_array = ArrayFromJSON(utf8(), R"(["one", "negative one"])");
- const auto value_array = ArrayFromJSON(int64(), R"([1, -1])");
- const auto foreignId_array = ArrayFromJSON(int64(), R"([1, 1])");
-
- const std::shared_ptr<Table>& expected_table = Table::Make(
- expected_schema, {id_array, keyname_array, value_array, foreignId_array});
-
- AssertTablesEqual(*expected_table, *table);
+ auto expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1]
+ ])"});
+ ASSERT_NO_FATAL_FAILURE(AssertTablesEqual(*expected_table, *table, /*verbose=*/true));
+
+ // Set multiple parameters at once
+ record_batch =
+ RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]], [[0, "%zero"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
+ ASSERT_OK_AND_ASSIGN(flight_info, prepared_statement->Execute());
+ ASSERT_OK_AND_ASSIGN(stream, sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
+ ASSERT_OK_AND_ASSIGN(table, stream->ToTable());
+ expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1],
+ [2, "zero", 0, 1]
Review Comment:
The implementation deals badly with nulls since we have to then guess the type, and I didn't really want to implement an adaptive type-guessing reader right here. It might be worth it if we expect to turn this into a more robust example. (But then I'd want to explore reusing the work I'm already doing in the ADBC SQLite driver instead of rewriting that logic twice.)
That said it should be fine to add some nulls so long as they're never the first row of a result, I'll go update that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on a diff in pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
pitrou commented on code in PR #14266:
URL: https://github.com/apache/arrow/pull/14266#discussion_r1009671144
##########
cpp/src/arrow/flight/sql/server_test.cc:
##########
@@ -555,41 +548,34 @@ TEST_F(TestFlightSqlServer, TestCommandPreparedStatementUpdateWithParameterBindi
sql_client->Prepare(
{}, "INSERT INTO INTTABLE (keyName, value) VALUES ('new_value', ?)"));
- auto parameter_schema = prepared_statement->parameter_schema();
-
+ const std::shared_ptr<Schema>& parameter_schema =
+ prepared_statement->parameter_schema();
const std::shared_ptr<Schema>& expected_parameter_schema =
arrow::schema({arrow::field("parameter_1", example::GetUnknownColumnDataType())});
+ ASSERT_NO_FATAL_FAILURE(AssertSchemaEqual(expected_parameter_schema, parameter_schema));
- AssertSchemaEqual(expected_parameter_schema, parameter_schema);
-
- std::shared_ptr<Array> type_ids = ArrayFromJSON(int8(), R"([2])");
- std::shared_ptr<Array> offsets = ArrayFromJSON(int32(), R"([0])");
- std::shared_ptr<Array> string_array = ArrayFromJSON(utf8(), R"([])");
- std::shared_ptr<Array> bytes_array = ArrayFromJSON(binary(), R"([])");
- std::shared_ptr<Array> bigint_array = ArrayFromJSON(int64(), R"([999])");
- std::shared_ptr<Array> double_array = ArrayFromJSON(float64(), R"([])");
-
- ASSERT_OK_AND_ASSIGN(
- auto parameter_1_array,
- DenseUnionArray::Make(*type_ids, *offsets,
- {string_array, bytes_array, bigint_array, double_array},
- {"string", "bytes", "bigint", "double"}, {0, 1, 2, 3}));
-
- const std::shared_ptr<RecordBatch>& record_batch =
- RecordBatch::Make(parameter_schema, 1, {parameter_1_array});
-
- ASSERT_OK(prepared_statement->SetParameters(record_batch));
-
- ASSERT_OK_AND_EQ(4, ExecuteCountQuery("SELECT COUNT(*) FROM intTable"));
-
- ASSERT_OK_AND_EQ(1, prepared_statement->ExecuteUpdate());
+ auto record_batch = RecordBatchFromJSON(parameter_schema, R"([ [[2, 999]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
ASSERT_OK_AND_EQ(5, ExecuteCountQuery("SELECT COUNT(*) FROM intTable"));
-
+ ASSERT_OK_AND_EQ(1, prepared_statement->ExecuteUpdate());
+ ASSERT_OK_AND_EQ(6, ExecuteCountQuery("SELECT COUNT(*) FROM intTable"));
ASSERT_OK_AND_EQ(1, sql_client->ExecuteUpdate(
{}, "DELETE FROM intTable WHERE keyName = 'new_value'"));
+ ASSERT_OK_AND_EQ(5, ExecuteCountQuery("SELECT COUNT(*) FROM intTable"));
- ASSERT_OK_AND_EQ(4, ExecuteCountQuery("SELECT COUNT(*) FROM intTable"));
+ // Set multiple parameters at once
+ record_batch = RecordBatchFromJSON(parameter_schema, R"([ [[2, 999]], [[2, 42]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
+ ASSERT_OK_AND_EQ(2, prepared_statement->ExecuteUpdate());
+ ASSERT_OK_AND_EQ(7, ExecuteCountQuery("SELECT COUNT(*) FROM intTable"));
+
+ // Set a stream of parameters
+ record_batch = RecordBatchFromJSON(parameter_schema, R"([ [[2, 999]], [[2, 42]] ])");
+ ASSERT_OK_AND_ASSIGN(auto reader, RecordBatchReader::Make({std::move(record_batch)}));
Review Comment:
Perhaps instead make this reader non-trivial (at least 2 different batches)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on a diff in pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #14266:
URL: https://github.com/apache/arrow/pull/14266#discussion_r1006890482
##########
cpp/src/arrow/flight/sql/client.cc:
##########
@@ -574,26 +584,21 @@ arrow::Result<int64_t> PreparedStatement::ExecuteUpdate(
command.set_prepared_statement_handle(handle_);
ARROW_ASSIGN_OR_RAISE(FlightDescriptor descriptor,
GetFlightDescriptorForCommand(command));
- std::unique_ptr<FlightStreamWriter> writer;
- std::unique_ptr<FlightMetadataReader> reader;
-
- if (parameter_binding_ && parameter_binding_->num_rows() > 0) {
- ARROW_RETURN_NOT_OK(client_->DoPut(options, descriptor, parameter_binding_->schema(),
- &writer, &reader));
- ARROW_RETURN_NOT_OK(writer->WriteRecordBatch(*parameter_binding_));
+ std::shared_ptr<Buffer> metadata;
+ if (parameter_binding_) {
+ ARROW_ASSIGN_OR_RAISE(metadata, BindParameters(client_->impl_.get(), options,
+ descriptor, parameter_binding_.get()));
} else {
const std::shared_ptr<Schema> schema = arrow::schema({});
- ARROW_RETURN_NOT_OK(client_->DoPut(options, descriptor, schema, &writer, &reader));
- const ArrayVector columns;
- const auto& record_batch = arrow::RecordBatch::Make(schema, 0, columns);
- ARROW_RETURN_NOT_OK(writer->WriteRecordBatch(*record_batch));
+ auto record_batch = arrow::RecordBatch::Make(schema, 0, ArrayVector{});
+ ARROW_ASSIGN_OR_RAISE(auto params,
+ RecordBatchReader::Make({std::move(record_batch)}));
Review Comment:
It's not written into the spec. It also doesn't seem necessary here. I've removed it (or rather, replaced it with a 0-batch reader).
##########
cpp/src/arrow/flight/sql/example/sqlite_server.cc:
##########
@@ -77,50 +84,25 @@ std::string PrepareQueryForGetTables(const GetTables& command) {
return table_query.str();
}
-Status SetParametersOnSQLiteStatement(sqlite3_stmt* stmt, FlightMessageReader* reader) {
+template <typename Callback>
+Status SetParametersOnSQLiteStatement(SqliteStatement* statement,
+ FlightMessageReader* reader, Callback callback) {
+ sqlite3_stmt* stmt = statement->GetSqlite3Stmt();
while (true) {
ARROW_ASSIGN_OR_RAISE(FlightStreamChunk chunk, reader->Next());
- std::shared_ptr<RecordBatch>& record_batch = chunk.data;
- if (record_batch == nullptr) break;
+ if (chunk.data == nullptr) break;
- const int64_t num_rows = record_batch->num_rows();
- const int& num_columns = record_batch->num_columns();
+ const int64_t num_rows = chunk.data->num_rows();
+ if (num_rows == 0) continue;
+ ARROW_RETURN_NOT_OK(statement->SetParameters({std::move(chunk.data)}));
for (int i = 0; i < num_rows; ++i) {
- for (int c = 0; c < num_columns; ++c) {
- const std::shared_ptr<Array>& column = record_batch->column(c);
- ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Scalar> scalar, column->GetScalar(i));
-
- auto& holder = static_cast<DenseUnionScalar&>(*scalar).value;
-
- switch (holder->type->id()) {
- case Type::INT64: {
- int64_t value = static_cast<Int64Scalar&>(*holder).value;
- sqlite3_bind_int64(stmt, c + 1, value);
- break;
- }
- case Type::FLOAT: {
- double value = static_cast<FloatScalar&>(*holder).value;
- sqlite3_bind_double(stmt, c + 1, value);
- break;
- }
- case Type::STRING: {
- std::shared_ptr<Buffer> buffer = static_cast<StringScalar&>(*holder).value;
- sqlite3_bind_text(stmt, c + 1, reinterpret_cast<const char*>(buffer->data()),
- static_cast<int>(buffer->size()), SQLITE_TRANSIENT);
- break;
- }
- case Type::BINARY: {
- std::shared_ptr<Buffer> buffer = static_cast<BinaryScalar&>(*holder).value;
- sqlite3_bind_blob(stmt, c + 1, buffer->data(),
- static_cast<int>(buffer->size()), SQLITE_TRANSIENT);
- break;
- }
- default:
- return Status::Invalid("Received unsupported data type: ",
- holder->type->ToString());
- }
+ if (sqlite3_clear_bindings(stmt) != SQLITE_OK) {
+ return Status::Invalid("Failed to reset bindings on row ", i, ": ",
+ sqlite3_errmsg(statement->db()));
}
+ ARROW_RETURN_NOT_OK(statement->Bind(/*batch_index=*/0, i));
+ ARROW_RETURN_NOT_OK(callback());
Review Comment:
We can't bind multiple rows at once, so we have to execute the query after each binding.
##########
cpp/src/arrow/flight/sql/example/sqlite_statement.h:
##########
@@ -62,15 +62,25 @@ class SqliteStatement {
/// \brief Returns the underlying sqlite3_stmt.
/// \return A sqlite statement.
- sqlite3_stmt* GetSqlite3Stmt() const;
+ [[nodiscard]] sqlite3_stmt* GetSqlite3Stmt() const;
Review Comment:
I've removed them all.
##########
cpp/src/arrow/flight/sql/example/sqlite_server.cc:
##########
@@ -77,50 +84,25 @@ std::string PrepareQueryForGetTables(const GetTables& command) {
return table_query.str();
}
-Status SetParametersOnSQLiteStatement(sqlite3_stmt* stmt, FlightMessageReader* reader) {
+template <typename Callback>
+Status SetParametersOnSQLiteStatement(SqliteStatement* statement,
+ FlightMessageReader* reader, Callback callback) {
+ sqlite3_stmt* stmt = statement->GetSqlite3Stmt();
while (true) {
ARROW_ASSIGN_OR_RAISE(FlightStreamChunk chunk, reader->Next());
- std::shared_ptr<RecordBatch>& record_batch = chunk.data;
- if (record_batch == nullptr) break;
+ if (chunk.data == nullptr) break;
- const int64_t num_rows = record_batch->num_rows();
- const int& num_columns = record_batch->num_columns();
+ const int64_t num_rows = chunk.data->num_rows();
+ if (num_rows == 0) continue;
+ ARROW_RETURN_NOT_OK(statement->SetParameters({std::move(chunk.data)}));
for (int i = 0; i < num_rows; ++i) {
- for (int c = 0; c < num_columns; ++c) {
- const std::shared_ptr<Array>& column = record_batch->column(c);
- ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Scalar> scalar, column->GetScalar(i));
-
- auto& holder = static_cast<DenseUnionScalar&>(*scalar).value;
-
- switch (holder->type->id()) {
- case Type::INT64: {
- int64_t value = static_cast<Int64Scalar&>(*holder).value;
- sqlite3_bind_int64(stmt, c + 1, value);
- break;
- }
- case Type::FLOAT: {
- double value = static_cast<FloatScalar&>(*holder).value;
- sqlite3_bind_double(stmt, c + 1, value);
- break;
- }
- case Type::STRING: {
- std::shared_ptr<Buffer> buffer = static_cast<StringScalar&>(*holder).value;
- sqlite3_bind_text(stmt, c + 1, reinterpret_cast<const char*>(buffer->data()),
- static_cast<int>(buffer->size()), SQLITE_TRANSIENT);
- break;
- }
- case Type::BINARY: {
- std::shared_ptr<Buffer> buffer = static_cast<BinaryScalar&>(*holder).value;
- sqlite3_bind_blob(stmt, c + 1, buffer->data(),
- static_cast<int>(buffer->size()), SQLITE_TRANSIENT);
- break;
- }
- default:
- return Status::Invalid("Received unsupported data type: ",
- holder->type->ToString());
- }
+ if (sqlite3_clear_bindings(stmt) != SQLITE_OK) {
+ return Status::Invalid("Failed to reset bindings on row ", i, ": ",
+ sqlite3_errmsg(statement->db()));
}
+ ARROW_RETURN_NOT_OK(statement->Bind(/*batch_index=*/0, i));
Review Comment:
Yes, since there's only a single batch being fed through at a time.
##########
cpp/src/arrow/flight/sql/server_test.cc:
##########
@@ -502,51 +489,53 @@ TEST_F(TestFlightSqlServer, TestCommandPreparedStatementQueryWithParameterBindin
auto prepared_statement,
sql_client->Prepare({}, "SELECT * FROM intTable WHERE keyName LIKE ?"));
- auto parameter_schema = prepared_statement->parameter_schema();
-
+ const std::shared_ptr<Schema>& parameter_schema =
+ prepared_statement->parameter_schema();
const std::shared_ptr<Schema>& expected_parameter_schema =
arrow::schema({arrow::field("parameter_1", example::GetUnknownColumnDataType())});
+ ASSERT_NO_FATAL_FAILURE(AssertSchemaEqual(expected_parameter_schema, parameter_schema));
- AssertSchemaEqual(expected_parameter_schema, parameter_schema);
-
- std::shared_ptr<Array> type_ids = ArrayFromJSON(int8(), R"([0])");
- std::shared_ptr<Array> offsets = ArrayFromJSON(int32(), R"([0])");
- std::shared_ptr<Array> string_array = ArrayFromJSON(utf8(), R"(["%one"])");
- std::shared_ptr<Array> bytes_array = ArrayFromJSON(binary(), R"([])");
- std::shared_ptr<Array> bigint_array = ArrayFromJSON(int64(), R"([])");
- std::shared_ptr<Array> double_array = ArrayFromJSON(float64(), R"([])");
-
- ASSERT_OK_AND_ASSIGN(
- auto parameter_1_array,
- DenseUnionArray::Make(*type_ids, *offsets,
- {string_array, bytes_array, bigint_array, double_array},
- {"string", "bytes", "bigint", "double"}, {0, 1, 2, 3}));
-
- const std::shared_ptr<RecordBatch>& record_batch =
- RecordBatch::Make(parameter_schema, 1, {parameter_1_array});
-
- ASSERT_OK(prepared_statement->SetParameters(record_batch));
+ auto record_batch = RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
ASSERT_OK_AND_ASSIGN(auto flight_info, prepared_statement->Execute());
-
ASSERT_OK_AND_ASSIGN(auto stream,
sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
-
ASSERT_OK_AND_ASSIGN(auto table, stream->ToTable());
const std::shared_ptr<Schema>& expected_schema =
arrow::schema({arrow::field("id", int64()), arrow::field("keyName", utf8()),
arrow::field("value", int64()), arrow::field("foreignId", int64())});
- const auto id_array = ArrayFromJSON(int64(), R"([1, 3])");
- const auto keyname_array = ArrayFromJSON(utf8(), R"(["one", "negative one"])");
- const auto value_array = ArrayFromJSON(int64(), R"([1, -1])");
- const auto foreignId_array = ArrayFromJSON(int64(), R"([1, 1])");
-
- const std::shared_ptr<Table>& expected_table = Table::Make(
- expected_schema, {id_array, keyname_array, value_array, foreignId_array});
-
- AssertTablesEqual(*expected_table, *table);
+ auto expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1]
+ ])"});
+ ASSERT_NO_FATAL_FAILURE(AssertTablesEqual(*expected_table, *table, /*verbose=*/true));
+
+ // Set multiple parameters at once
+ record_batch =
+ RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]], [[0, "%zero"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
+ ASSERT_OK_AND_ASSIGN(flight_info, prepared_statement->Execute());
+ ASSERT_OK_AND_ASSIGN(stream, sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
+ ASSERT_OK_AND_ASSIGN(table, stream->ToTable());
+ expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1],
+ [2, "zero", 0, 1]
Review Comment:
Adjusted some tests to return nulls.
##########
cpp/src/arrow/flight/sql/example/sqlite_statement.h:
##########
@@ -62,15 +62,25 @@ class SqliteStatement {
/// \brief Returns the underlying sqlite3_stmt.
/// \return A sqlite statement.
- sqlite3_stmt* GetSqlite3Stmt() const;
+ [[nodiscard]] sqlite3_stmt* GetSqlite3Stmt() const;
Review Comment:
The nodiscard thing is a quirk of the code linter in clangd; it suggests `[[nodiscard]]` on basically every getter due to https://clang.llvm.org/extra/clang-tidy/checks/modernize/use-nodiscard.html.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm merged pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
lidavidm merged PR #14266:
URL: https://github.com/apache/arrow/pull/14266
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on a diff in pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
pitrou commented on code in PR #14266:
URL: https://github.com/apache/arrow/pull/14266#discussion_r1005675232
##########
cpp/src/arrow/flight/sql/client.h:
##########
@@ -392,17 +392,18 @@ class ARROW_FLIGHT_SQL_EXPORT PreparedStatement {
/// \brief Retrieve the parameter schema from the query.
/// \return The parameter schema from the query.
- std::shared_ptr<Schema> parameter_schema() const;
+ [[nodiscard]] const std::shared_ptr<Schema>& parameter_schema() const;
Review Comment:
Is the `nodiscard` meant to protect against some incorrect usage here?
##########
cpp/src/arrow/flight/sql/client.cc:
##########
@@ -574,26 +584,21 @@ arrow::Result<int64_t> PreparedStatement::ExecuteUpdate(
command.set_prepared_statement_handle(handle_);
ARROW_ASSIGN_OR_RAISE(FlightDescriptor descriptor,
GetFlightDescriptorForCommand(command));
- std::unique_ptr<FlightStreamWriter> writer;
- std::unique_ptr<FlightMetadataReader> reader;
-
- if (parameter_binding_ && parameter_binding_->num_rows() > 0) {
- ARROW_RETURN_NOT_OK(client_->DoPut(options, descriptor, parameter_binding_->schema(),
- &writer, &reader));
- ARROW_RETURN_NOT_OK(writer->WriteRecordBatch(*parameter_binding_));
+ std::shared_ptr<Buffer> metadata;
+ if (parameter_binding_) {
+ ARROW_ASSIGN_OR_RAISE(metadata, BindParameters(client_->impl_.get(), options,
+ descriptor, parameter_binding_.get()));
} else {
const std::shared_ptr<Schema> schema = arrow::schema({});
- ARROW_RETURN_NOT_OK(client_->DoPut(options, descriptor, schema, &writer, &reader));
- const ArrayVector columns;
- const auto& record_batch = arrow::RecordBatch::Make(schema, 0, columns);
- ARROW_RETURN_NOT_OK(writer->WriteRecordBatch(*record_batch));
+ auto record_batch = arrow::RecordBatch::Make(schema, 0, ArrayVector{});
+ ARROW_ASSIGN_OR_RAISE(auto params,
+ RecordBatchReader::Make({std::move(record_batch)}));
Review Comment:
Is it part of the spec that we have to send a single 0-row batch here?
##########
cpp/src/arrow/flight/sql/example/sqlite_server.cc:
##########
@@ -346,18 +331,17 @@ class SQLiteFlightSqlServer::Impl {
arrow::Result<std::unique_ptr<FlightDataStream>> DoGetCatalogs(
const ServerCallContext& context) {
- // As SQLite doesn't support catalogs, this will return an empty record batch.
-
+ // https://www.sqlite.org/cli.html
+ // > The ".databases" command shows a list of all databases open
+ // > in the current connection. There will always be at least
+ // > 2. The first one is "main", the original database opened.
Review Comment:
This comment is a bit confusing, because we return a single row ("main") here.
##########
cpp/src/arrow/flight/sql/example/sqlite_server.cc:
##########
@@ -77,50 +84,25 @@ std::string PrepareQueryForGetTables(const GetTables& command) {
return table_query.str();
}
-Status SetParametersOnSQLiteStatement(sqlite3_stmt* stmt, FlightMessageReader* reader) {
+template <typename Callback>
+Status SetParametersOnSQLiteStatement(SqliteStatement* statement,
+ FlightMessageReader* reader, Callback callback) {
+ sqlite3_stmt* stmt = statement->GetSqlite3Stmt();
while (true) {
ARROW_ASSIGN_OR_RAISE(FlightStreamChunk chunk, reader->Next());
- std::shared_ptr<RecordBatch>& record_batch = chunk.data;
- if (record_batch == nullptr) break;
+ if (chunk.data == nullptr) break;
- const int64_t num_rows = record_batch->num_rows();
- const int& num_columns = record_batch->num_columns();
+ const int64_t num_rows = chunk.data->num_rows();
+ if (num_rows == 0) continue;
+ ARROW_RETURN_NOT_OK(statement->SetParameters({std::move(chunk.data)}));
for (int i = 0; i < num_rows; ++i) {
- for (int c = 0; c < num_columns; ++c) {
- const std::shared_ptr<Array>& column = record_batch->column(c);
- ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Scalar> scalar, column->GetScalar(i));
-
- auto& holder = static_cast<DenseUnionScalar&>(*scalar).value;
-
- switch (holder->type->id()) {
- case Type::INT64: {
- int64_t value = static_cast<Int64Scalar&>(*holder).value;
- sqlite3_bind_int64(stmt, c + 1, value);
- break;
- }
- case Type::FLOAT: {
- double value = static_cast<FloatScalar&>(*holder).value;
- sqlite3_bind_double(stmt, c + 1, value);
- break;
- }
- case Type::STRING: {
- std::shared_ptr<Buffer> buffer = static_cast<StringScalar&>(*holder).value;
- sqlite3_bind_text(stmt, c + 1, reinterpret_cast<const char*>(buffer->data()),
- static_cast<int>(buffer->size()), SQLITE_TRANSIENT);
- break;
- }
- case Type::BINARY: {
- std::shared_ptr<Buffer> buffer = static_cast<BinaryScalar&>(*holder).value;
- sqlite3_bind_blob(stmt, c + 1, buffer->data(),
- static_cast<int>(buffer->size()), SQLITE_TRANSIENT);
- break;
- }
- default:
- return Status::Invalid("Received unsupported data type: ",
- holder->type->ToString());
- }
+ if (sqlite3_clear_bindings(stmt) != SQLITE_OK) {
+ return Status::Invalid("Failed to reset bindings on row ", i, ": ",
+ sqlite3_errmsg(statement->db()));
}
+ ARROW_RETURN_NOT_OK(statement->Bind(/*batch_index=*/0, i));
+ ARROW_RETURN_NOT_OK(callback());
Review Comment:
Hmm, why call the callback for each row?
##########
cpp/src/arrow/flight/sql/example/sqlite_statement.cc:
##########
@@ -160,10 +166,85 @@ arrow::Result<int> SqliteStatement::Reset() {
sqlite3_stmt* SqliteStatement::GetSqlite3Stmt() const { return stmt_; }
arrow::Result<int64_t> SqliteStatement::ExecuteUpdate() {
- ARROW_RETURN_NOT_OK(Step());
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(int rc, Step());
+ if (rc == SQLITE_DONE) break;
+ }
return sqlite3_changes(db_);
}
+Status SqliteStatement::SetParameters(
+ std::vector<std::shared_ptr<arrow::RecordBatch>> parameters) {
+ const int num_params = sqlite3_bind_parameter_count(stmt_);
+ for (const auto& batch : parameters) {
+ if (batch->num_columns() != num_params) {
+ return Status::Invalid("Expected ", num_params, " parameters, but got ",
+ batch->num_columns());
+ }
+ }
+ parameters_ = std::move(parameters);
+ auto end = std::remove_if(
+ parameters_.begin(), parameters_.end(),
+ [](const std::shared_ptr<RecordBatch>& batch) { return batch->num_rows() == 0; });
+ parameters_.erase(end, parameters_.end());
+ return Status::OK();
+}
+
+Status SqliteStatement::Bind(size_t batch_index, int64_t row_index) {
+ if (batch_index >= parameters_.size()) {
+ return Status::Invalid("Cannot bind to batch ", batch_index);
Review Comment:
`IndexError` perhaps?
##########
cpp/src/arrow/flight/sql/example/sqlite_statement.cc:
##########
@@ -160,10 +166,85 @@ arrow::Result<int> SqliteStatement::Reset() {
sqlite3_stmt* SqliteStatement::GetSqlite3Stmt() const { return stmt_; }
arrow::Result<int64_t> SqliteStatement::ExecuteUpdate() {
- ARROW_RETURN_NOT_OK(Step());
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(int rc, Step());
+ if (rc == SQLITE_DONE) break;
+ }
return sqlite3_changes(db_);
}
+Status SqliteStatement::SetParameters(
+ std::vector<std::shared_ptr<arrow::RecordBatch>> parameters) {
+ const int num_params = sqlite3_bind_parameter_count(stmt_);
+ for (const auto& batch : parameters) {
+ if (batch->num_columns() != num_params) {
+ return Status::Invalid("Expected ", num_params, " parameters, but got ",
+ batch->num_columns());
+ }
+ }
+ parameters_ = std::move(parameters);
+ auto end = std::remove_if(
+ parameters_.begin(), parameters_.end(),
+ [](const std::shared_ptr<RecordBatch>& batch) { return batch->num_rows() == 0; });
+ parameters_.erase(end, parameters_.end());
+ return Status::OK();
+}
+
+Status SqliteStatement::Bind(size_t batch_index, int64_t row_index) {
+ if (batch_index >= parameters_.size()) {
+ return Status::Invalid("Cannot bind to batch ", batch_index);
+ }
+ const RecordBatch& batch = *parameters_[batch_index];
+ if (row_index < 0 || row_index >= batch.num_rows()) {
+ return Status::Invalid("Cannot bind to row ", row_index, " in batch ", batch_index);
Review Comment:
Same here.
##########
cpp/src/arrow/flight/sql/server_test.cc:
##########
@@ -502,51 +489,53 @@ TEST_F(TestFlightSqlServer, TestCommandPreparedStatementQueryWithParameterBindin
auto prepared_statement,
sql_client->Prepare({}, "SELECT * FROM intTable WHERE keyName LIKE ?"));
- auto parameter_schema = prepared_statement->parameter_schema();
-
+ const std::shared_ptr<Schema>& parameter_schema =
+ prepared_statement->parameter_schema();
const std::shared_ptr<Schema>& expected_parameter_schema =
arrow::schema({arrow::field("parameter_1", example::GetUnknownColumnDataType())});
+ ASSERT_NO_FATAL_FAILURE(AssertSchemaEqual(expected_parameter_schema, parameter_schema));
- AssertSchemaEqual(expected_parameter_schema, parameter_schema);
-
- std::shared_ptr<Array> type_ids = ArrayFromJSON(int8(), R"([0])");
- std::shared_ptr<Array> offsets = ArrayFromJSON(int32(), R"([0])");
- std::shared_ptr<Array> string_array = ArrayFromJSON(utf8(), R"(["%one"])");
- std::shared_ptr<Array> bytes_array = ArrayFromJSON(binary(), R"([])");
- std::shared_ptr<Array> bigint_array = ArrayFromJSON(int64(), R"([])");
- std::shared_ptr<Array> double_array = ArrayFromJSON(float64(), R"([])");
-
- ASSERT_OK_AND_ASSIGN(
- auto parameter_1_array,
- DenseUnionArray::Make(*type_ids, *offsets,
- {string_array, bytes_array, bigint_array, double_array},
- {"string", "bytes", "bigint", "double"}, {0, 1, 2, 3}));
-
- const std::shared_ptr<RecordBatch>& record_batch =
- RecordBatch::Make(parameter_schema, 1, {parameter_1_array});
-
- ASSERT_OK(prepared_statement->SetParameters(record_batch));
+ auto record_batch = RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
ASSERT_OK_AND_ASSIGN(auto flight_info, prepared_statement->Execute());
-
ASSERT_OK_AND_ASSIGN(auto stream,
sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
-
ASSERT_OK_AND_ASSIGN(auto table, stream->ToTable());
const std::shared_ptr<Schema>& expected_schema =
arrow::schema({arrow::field("id", int64()), arrow::field("keyName", utf8()),
arrow::field("value", int64()), arrow::field("foreignId", int64())});
- const auto id_array = ArrayFromJSON(int64(), R"([1, 3])");
- const auto keyname_array = ArrayFromJSON(utf8(), R"(["one", "negative one"])");
- const auto value_array = ArrayFromJSON(int64(), R"([1, -1])");
- const auto foreignId_array = ArrayFromJSON(int64(), R"([1, 1])");
-
- const std::shared_ptr<Table>& expected_table = Table::Make(
- expected_schema, {id_array, keyname_array, value_array, foreignId_array});
-
- AssertTablesEqual(*expected_table, *table);
+ auto expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1]
+ ])"});
+ ASSERT_NO_FATAL_FAILURE(AssertTablesEqual(*expected_table, *table, /*verbose=*/true));
+
+ // Set multiple parameters at once
+ record_batch =
+ RecordBatchFromJSON(parameter_schema, R"([ [[0, "%one"]], [[0, "%zero"]] ])");
+ ASSERT_OK(prepared_statement->SetParameters(std::move(record_batch)));
+ ASSERT_OK_AND_ASSIGN(flight_info, prepared_statement->Execute());
+ ASSERT_OK_AND_ASSIGN(stream, sql_client->DoGet({}, flight_info->endpoints()[0].ticket));
+ ASSERT_OK_AND_ASSIGN(table, stream->ToTable());
+ expected_table = TableFromJSON(expected_schema, {R"([
+ [1, "one", 1, 1],
+ [3, "negative one", -1, 1],
+ [2, "zero", 0, 1]
Review Comment:
For the record, are there any tests emitting/receiving nulls?
##########
cpp/src/arrow/flight/sql/example/sqlite_server.cc:
##########
@@ -77,50 +84,25 @@ std::string PrepareQueryForGetTables(const GetTables& command) {
return table_query.str();
}
-Status SetParametersOnSQLiteStatement(sqlite3_stmt* stmt, FlightMessageReader* reader) {
+template <typename Callback>
+Status SetParametersOnSQLiteStatement(SqliteStatement* statement,
+ FlightMessageReader* reader, Callback callback) {
+ sqlite3_stmt* stmt = statement->GetSqlite3Stmt();
while (true) {
ARROW_ASSIGN_OR_RAISE(FlightStreamChunk chunk, reader->Next());
- std::shared_ptr<RecordBatch>& record_batch = chunk.data;
- if (record_batch == nullptr) break;
+ if (chunk.data == nullptr) break;
- const int64_t num_rows = record_batch->num_rows();
- const int& num_columns = record_batch->num_columns();
+ const int64_t num_rows = chunk.data->num_rows();
+ if (num_rows == 0) continue;
+ ARROW_RETURN_NOT_OK(statement->SetParameters({std::move(chunk.data)}));
for (int i = 0; i < num_rows; ++i) {
- for (int c = 0; c < num_columns; ++c) {
- const std::shared_ptr<Array>& column = record_batch->column(c);
- ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Scalar> scalar, column->GetScalar(i));
-
- auto& holder = static_cast<DenseUnionScalar&>(*scalar).value;
-
- switch (holder->type->id()) {
- case Type::INT64: {
- int64_t value = static_cast<Int64Scalar&>(*holder).value;
- sqlite3_bind_int64(stmt, c + 1, value);
- break;
- }
- case Type::FLOAT: {
- double value = static_cast<FloatScalar&>(*holder).value;
- sqlite3_bind_double(stmt, c + 1, value);
- break;
- }
- case Type::STRING: {
- std::shared_ptr<Buffer> buffer = static_cast<StringScalar&>(*holder).value;
- sqlite3_bind_text(stmt, c + 1, reinterpret_cast<const char*>(buffer->data()),
- static_cast<int>(buffer->size()), SQLITE_TRANSIENT);
- break;
- }
- case Type::BINARY: {
- std::shared_ptr<Buffer> buffer = static_cast<BinaryScalar&>(*holder).value;
- sqlite3_bind_blob(stmt, c + 1, buffer->data(),
- static_cast<int>(buffer->size()), SQLITE_TRANSIENT);
- break;
- }
- default:
- return Status::Invalid("Received unsupported data type: ",
- holder->type->ToString());
- }
+ if (sqlite3_clear_bindings(stmt) != SQLITE_OK) {
+ return Status::Invalid("Failed to reset bindings on row ", i, ": ",
+ sqlite3_errmsg(statement->db()));
}
+ ARROW_RETURN_NOT_OK(statement->Bind(/*batch_index=*/0, i));
Review Comment:
Is `/*batch_index=*/0` expected here?
##########
cpp/src/arrow/flight/sql/example/sqlite_statement.cc:
##########
@@ -160,10 +166,85 @@ arrow::Result<int> SqliteStatement::Reset() {
sqlite3_stmt* SqliteStatement::GetSqlite3Stmt() const { return stmt_; }
arrow::Result<int64_t> SqliteStatement::ExecuteUpdate() {
- ARROW_RETURN_NOT_OK(Step());
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(int rc, Step());
+ if (rc == SQLITE_DONE) break;
+ }
return sqlite3_changes(db_);
}
+Status SqliteStatement::SetParameters(
+ std::vector<std::shared_ptr<arrow::RecordBatch>> parameters) {
+ const int num_params = sqlite3_bind_parameter_count(stmt_);
+ for (const auto& batch : parameters) {
+ if (batch->num_columns() != num_params) {
+ return Status::Invalid("Expected ", num_params, " parameters, but got ",
+ batch->num_columns());
+ }
+ }
+ parameters_ = std::move(parameters);
+ auto end = std::remove_if(
+ parameters_.begin(), parameters_.end(),
+ [](const std::shared_ptr<RecordBatch>& batch) { return batch->num_rows() == 0; });
+ parameters_.erase(end, parameters_.end());
+ return Status::OK();
+}
+
+Status SqliteStatement::Bind(size_t batch_index, int64_t row_index) {
+ if (batch_index >= parameters_.size()) {
+ return Status::Invalid("Cannot bind to batch ", batch_index);
+ }
+ const RecordBatch& batch = *parameters_[batch_index];
+ if (row_index < 0 || row_index >= batch.num_rows()) {
+ return Status::Invalid("Cannot bind to row ", row_index, " in batch ", batch_index);
+ }
+
+ if (sqlite3_clear_bindings(stmt_) != SQLITE_OK) {
+ return Status::Invalid("Failed to reset bindings: ", sqlite3_errmsg(db_));
+ }
+ for (int c = 0; c < batch.num_columns(); ++c) {
+ const std::shared_ptr<Array>& column = batch.column(c);
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Scalar> scalar, column->GetScalar(row_index));
Review Comment:
Instead of using `GetScalar` and then switch on type, wouldn't it be more efficient to first switch on type and then access the concrete array directly?
##########
cpp/src/arrow/flight/sql/example/sqlite_statement.h:
##########
@@ -62,15 +62,25 @@ class SqliteStatement {
/// \brief Returns the underlying sqlite3_stmt.
/// \return A sqlite statement.
- sqlite3_stmt* GetSqlite3Stmt() const;
+ [[nodiscard]] sqlite3_stmt* GetSqlite3Stmt() const;
Review Comment:
Not sure why `nodiscard` is added on these?
##########
cpp/src/arrow/flight/sql/example/sqlite_statement.cc:
##########
@@ -160,10 +166,85 @@ arrow::Result<int> SqliteStatement::Reset() {
sqlite3_stmt* SqliteStatement::GetSqlite3Stmt() const { return stmt_; }
arrow::Result<int64_t> SqliteStatement::ExecuteUpdate() {
- ARROW_RETURN_NOT_OK(Step());
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(int rc, Step());
+ if (rc == SQLITE_DONE) break;
+ }
return sqlite3_changes(db_);
}
+Status SqliteStatement::SetParameters(
+ std::vector<std::shared_ptr<arrow::RecordBatch>> parameters) {
+ const int num_params = sqlite3_bind_parameter_count(stmt_);
+ for (const auto& batch : parameters) {
+ if (batch->num_columns() != num_params) {
+ return Status::Invalid("Expected ", num_params, " parameters, but got ",
+ batch->num_columns());
+ }
+ }
+ parameters_ = std::move(parameters);
+ auto end = std::remove_if(
+ parameters_.begin(), parameters_.end(),
+ [](const std::shared_ptr<RecordBatch>& batch) { return batch->num_rows() == 0; });
+ parameters_.erase(end, parameters_.end());
+ return Status::OK();
+}
+
+Status SqliteStatement::Bind(size_t batch_index, int64_t row_index) {
+ if (batch_index >= parameters_.size()) {
+ return Status::Invalid("Cannot bind to batch ", batch_index);
+ }
+ const RecordBatch& batch = *parameters_[batch_index];
+ if (row_index < 0 || row_index >= batch.num_rows()) {
+ return Status::Invalid("Cannot bind to row ", row_index, " in batch ", batch_index);
+ }
+
+ if (sqlite3_clear_bindings(stmt_) != SQLITE_OK) {
+ return Status::Invalid("Failed to reset bindings: ", sqlite3_errmsg(db_));
+ }
+ for (int c = 0; c < batch.num_columns(); ++c) {
+ const std::shared_ptr<Array>& column = batch.column(c);
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Scalar> scalar, column->GetScalar(row_index));
+ if (scalar->type->id() == Type::DENSE_UNION) {
+ scalar = checked_cast<DenseUnionScalar&>(*scalar).value;
+ }
+
+ int rc = 0;
+ if (!scalar->is_valid) {
+ rc = sqlite3_bind_null(stmt_, c + 1);
+ continue;
+ } else {
+ switch (scalar->type->id()) {
+ case Type::INT64: {
+ int64_t value = checked_cast<const Int64Scalar&>(*scalar).value;
+ rc = sqlite3_bind_int64(stmt_, c + 1, value);
+ break;
+ }
+ case Type::FLOAT: {
+ float value = checked_cast<const FloatScalar&>(*scalar).value;
+ rc = sqlite3_bind_double(stmt_, c + 1, value);
+ break;
+ }
+ case Type::STRING: {
+ const std::shared_ptr<Buffer>& buffer =
+ checked_cast<const StringScalar&>(*scalar).value;
+ rc = sqlite3_bind_text(stmt_, c + 1,
+ reinterpret_cast<const char*>(buffer->data()),
+ static_cast<int>(buffer->size()), SQLITE_TRANSIENT);
+ break;
+ }
+ default:
+ return Status::Invalid("Received unsupported data type: ", *scalar->type);
Review Comment:
`TypeError`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on a diff in pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
pitrou commented on code in PR #14266:
URL: https://github.com/apache/arrow/pull/14266#discussion_r1009666006
##########
cpp/src/arrow/flight/sql/example/sqlite_statement.cc:
##########
@@ -160,10 +169,100 @@ arrow::Result<int> SqliteStatement::Reset() {
sqlite3_stmt* SqliteStatement::GetSqlite3Stmt() const { return stmt_; }
arrow::Result<int64_t> SqliteStatement::ExecuteUpdate() {
- ARROW_RETURN_NOT_OK(Step());
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(int rc, Step());
+ if (rc == SQLITE_DONE) break;
+ }
return sqlite3_changes(db_);
}
+Status SqliteStatement::SetParameters(
+ std::vector<std::shared_ptr<arrow::RecordBatch>> parameters) {
+ const int num_params = sqlite3_bind_parameter_count(stmt_);
+ for (const auto& batch : parameters) {
+ if (batch->num_columns() != num_params) {
+ return Status::Invalid("Expected ", num_params, " parameters, but got ",
+ batch->num_columns());
+ }
+ }
+ parameters_ = std::move(parameters);
+ auto end = std::remove_if(
+ parameters_.begin(), parameters_.end(),
+ [](const std::shared_ptr<RecordBatch>& batch) { return batch->num_rows() == 0; });
+ parameters_.erase(end, parameters_.end());
+ return Status::OK();
+}
+
+Status SqliteStatement::Bind(size_t batch_index, int64_t row_index) {
+ if (batch_index >= parameters_.size()) {
+ return Status::IndexError("Cannot bind to batch ", batch_index);
+ }
+ const RecordBatch& batch = *parameters_[batch_index];
+ if (row_index < 0 || row_index >= batch.num_rows()) {
+ return Status::IndexError("Cannot bind to row ", row_index, " in batch ",
+ batch_index);
+ }
+
+ if (sqlite3_clear_bindings(stmt_) != SQLITE_OK) {
+ return Status::Invalid("Failed to reset bindings: ", sqlite3_errmsg(db_));
+ }
+ for (int c = 0; c < batch.num_columns(); ++c) {
+ Array* column = batch.column(c).get();
+ int64_t column_index = row_index;
+ if (column->type_id() == Type::DENSE_UNION) {
+ // Allow polymorphic bindings via union
Review Comment:
Nice, but why not also handle sparse unions?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
lidavidm commented on PR #14266:
URL: https://github.com/apache/arrow/pull/14266#issuecomment-1298604923
MacOS/Ruby failures look unrelated, restarted R build since the R setup action 404'd
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
lidavidm commented on PR #14266:
URL: https://github.com/apache/arrow/pull/14266#issuecomment-1261036746
Split from #14082
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #14266: ARROW-17867: [C++][FlightRPC] Expose bulk parameter binding in Flight SQL
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #14266:
URL: https://github.com/apache/arrow/pull/14266#issuecomment-1261213031
:warning: Ticket **has not been started in JIRA**, please click 'Start Progress'.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org