You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "HackPoint (via GitHub)" <gi...@apache.org> on 2024/11/18 17:41:16 UTC
[PR] Feat/flight sql client [arrow]
HackPoint opened a new pull request, #44783:
URL: https://github.com/apache/arrow/pull/44783
# GH-${GITHUB_ISSUE_ID}: [FlightSqlClient] New C# Implementation for FlightSqlClient and PreparedStatement
## Rationale for this Change
This pull request introduces a **new implementation of `FlightSqlClient` and `PreparedStatement` in C#**. Previously, there was no C# client for Flight SQL, leaving a significant gap for .NET developers who wished to interact with Flight SQL servers.
The implementation aligns with the existing **C++ Flight SQL client** API, ensuring consistent and familiar behavior across languages and providing a robust client for the Apache Arrow ecosystem in .NET.
---
## What's Included in this PR?
### Key Features
1. **`FlightSqlClient`**:
- Provides query execution (`ExecuteAsync`, `ExecuteUpdateAsync`) and schema retrieval (`GetCatalogsAsync`, `GetDbSchemasAsync`, etc.).
- Implements metadata operations for catalogs, schemas, tables, and more.
- Fully integrated with gRPC and Apache Arrow ecosystems.
- Supports extensibility for advanced features like transactions.
2. **`PreparedStatement`**:
- Implements parameterized query execution (`SetParameters`, `ExecuteAsync`, and `ExecuteUpdateAsync`).
- Supports lifecycle management (`CloseAsync`) for effective resource handling.
- Aligns with the prepared statement design in the C++ client.
### API Consistency
This implementation mirrors the **C++ Flight SQL client** to ensure API alignment across supported languages:
- Consistent naming conventions and parameter semantics.
- Ensures .NET developers can work seamlessly with existing Flight SQL servers.
---
## Are These Changes Tested?
### Testing Overview
1. **Unit Tests**:
- Added tests for query execution and parameter binding in `PreparedStatement`.
- Verified schema retrieval methods like `GetCatalogsAsync` and `GetDbSchemasAsync`.
2. **Integration Tests**:
- Tested against a live Flight SQL server to validate query execution, schema retrieval, and metadata operations.
3. **End-to-End Tests**:
- Covered real-world scenarios for parameterized updates and queries, ensuring robustness.
### Example Test Cases
- Verify that parameterized queries return correct results with valid input.
- Ensure schema retrieval throws appropriate exceptions for invalid descriptors.
- Validate row counts after `ExecuteUpdateAsync`.
---
## Are There Any Breaking Changes?
This PR introduces **new functionality** and does not affect any existing features. There are **no breaking changes**.
---
## Are There Any User-Facing Changes?
### New Capabilities
1. **FlightSqlClient**:
- Query execution and schema retrieval for SQL queries on Flight SQL servers.
- Metadata retrieval for catalogs, schemas, tables, and more.
2. **PreparedStatement**:
- Supports parameterized queries with proper parameter binding.
- Provides robust lifecycle management and execution.
### API Consistency
- Aligns with the C++ Flight SQL client for interoperability and familiar API design.
---
## Additional Notes
- This PR **does not include transaction support** at this stage, as it requires additional server-side capabilities.
- All methods follow idiomatic C# practices, including `async/await` for non-blocking operations.
- Extensible for future enhancements, including advanced features like savepoints.
---
## Resources
- **C++ Flight SQL Client Reference**: [Apache Arrow Flight SQL Documentation](https://arrow.apache.org/docs/)
- **Apache Arrow Contribution Guide**: [Contributing to Apache Arrow](https://arrow.apache.org/docs/dev/developers/guide/)
---
## Feedback and Suggestions
Thank you for reviewing this contribution! Suggestions and feedback are welcome to ensure the implementation meets the project's standards and requirements.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "jeremyosterhoudt (via GitHub)" <gi...@apache.org>.
jeremyosterhoudt commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2573618767
Any updates on this? We are also looking to use .NET with a Flight SQL client and are eager to get this functionality.
--
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
Re: [PR] Feat/flight sql client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2485218939
> ould you please add the official Apache header to all the source files and open an issue to link this to?
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1968063719
##########
csharp/src/Apache.Arrow.Flight.Sql/RecordBatchExtensions.cs:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Apache.Arrow.Ipc;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public static class RecordBatchExtensions
+{
+ /// <summary>
+ /// Converts a RecordBatch into an asynchronous stream of FlightData.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An asynchronous stream of FlightData objects.</returns>
+ public static async IAsyncEnumerable<FlightData> ToFlightDataStreamAsync(this RecordBatch recordBatch,
+ FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null)
+ {
+ throw new ArgumentNullException(nameof(recordBatch));
+ }
+
+ // Use a memory stream to write the Arrow RecordBatch into FlightData format
+ using var memoryStream = new MemoryStream();
+ var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+
+ // Write the RecordBatch to the stream
+ await writer.WriteRecordBatchAsync(recordBatch).ConfigureAwait(false);
+ await writer.WriteEndAsync().ConfigureAwait(false);
+
+ // Reset the memory stream position
+ memoryStream.Position = 0;
+
+ // Read back the data to create FlightData
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()),
Review Comment:
I fully removed this class from implementation.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "zgramana (via GitHub)" <gi...@apache.org>.
zgramana commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2783802768
> Thanks again for such a comprehensive change! I'm very sorry it took me this long to get to it. I've left a bunch of feedback of various degrees of seriousness. You'll also likely need to rebase given the amount of time that has passed (and the presence of a merge conflict).
@CurtHagenlocher since this PR no longer has a merge conflict, and it get your final approval and merge? Much appreciated.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1968079460
##########
csharp/src/Apache.Arrow.Flight.Sql/DoPutResult.cs:
##########
@@ -0,0 +1,54 @@
+// 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.
+
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Client;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class DoPutResult
+{
+ public FlightClientRecordBatchStreamWriter Writer { get; }
+ public IAsyncStreamReader<FlightPutResult> Reader { get; }
+
+ public DoPutResult(FlightClientRecordBatchStreamWriter writer, IAsyncStreamReader<FlightPutResult> reader)
+ {
+ Writer = writer;
+ Reader = reader;
+ }
+
+ /// <summary>
+ /// Reads the metadata asynchronously from the reader.
+ /// </summary>
+ /// <returns>A ByteString containing the metadata read from the reader.</returns>
+ public async Task<Google.Protobuf.ByteString> ReadMetadataAsync()
Review Comment:
These are; Very short, CPU-bound calculations, thus is not required to put CancelationToken.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "kou (via GitHub)" <gi...@apache.org>.
kou commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2499692202
Could you revert all `cpp/` changes?
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1947995089
##########
csharp/src/Apache.Arrow.Flight.Sql/TableRef.cs:
##########
@@ -0,0 +1,23 @@
+// 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.
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class TableRef
+{
+ public string? Catalog { get; set; }
+ public string DbSchema { get; set; } = null!;
Review Comment:
The `null!` seems dodgy. As these are required to be set, consider using a pair of constructors `TableRef(string, string)` and `TableRef(string, string, string)` instead.
##########
csharp/src/Apache.Arrow.Flight.Sql/SchemaExtensions.cs:
##########
@@ -0,0 +1,52 @@
+// 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.
+
+using System;
+using System.IO;
+using Apache.Arrow.Ipc;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public static class SchemaExtensions
+{
+ /// <summary>
+ /// Deserializes a schema from a byte array.
+ /// </summary>
+ /// <param name="serializedSchema">The byte array representing the serialized schema.</param>
+ /// <returns>The deserialized Schema object.</returns>
+ public static Schema DeserializeSchema(byte[] serializedSchema)
+ {
+ if (serializedSchema == null || serializedSchema.Length == 0)
+ {
+ throw new ArgumentException("Invalid serialized schema");
+ }
+
+ using var stream = new MemoryStream(serializedSchema);
+ var reader = new ArrowStreamReader(stream);
Review Comment:
There's a ctor for `ArrowStreamReader` which takes a `ReadOnlyMemory<byte>` as an argument, so you can just pass in the bytes. This is probably a bit more efficient.
##########
csharp/src/Apache.Arrow.Flight.Sql/DoPutResult.cs:
##########
@@ -0,0 +1,54 @@
+// 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.
+
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Client;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class DoPutResult
+{
+ public FlightClientRecordBatchStreamWriter Writer { get; }
+ public IAsyncStreamReader<FlightPutResult> Reader { get; }
+
+ public DoPutResult(FlightClientRecordBatchStreamWriter writer, IAsyncStreamReader<FlightPutResult> reader)
+ {
+ Writer = writer;
+ Reader = reader;
+ }
+
+ /// <summary>
+ /// Reads the metadata asynchronously from the reader.
+ /// </summary>
+ /// <returns>A ByteString containing the metadata read from the reader.</returns>
+ public async Task<Google.Protobuf.ByteString> ReadMetadataAsync()
Review Comment:
In general, none of the `async` methods in this change take a `CancellationToken`. Shouldn't they?
##########
csharp/test/Apache.Arrow.Flight.Sql.Tests/FlightSqlClientTests.cs:
##########
@@ -0,0 +1,855 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Client;
+using Apache.Arrow.Flight.Sql.Client;
+using Apache.Arrow.Flight.TestWeb;
+using Apache.Arrow.Types;
+using Arrow.Flight.Protocol.Sql;
+using Google.Protobuf;
+using Grpc.Core.Utils;
+using Xunit;
+
+namespace Apache.Arrow.Flight.Sql.Tests;
+
+public class FlightSqlClientTests : IDisposable
+{
+ readonly TestFlightSqlWebFactory _testWebFactory;
+ readonly FlightStore _flightStore;
+ private readonly FlightSqlClient _flightSqlClient;
+ private readonly FlightSqlTestUtils _testUtils;
+
+ public FlightSqlClientTests()
+ {
+ _flightStore = new FlightStore();
+ _testWebFactory = new TestFlightSqlWebFactory(_flightStore);
+ FlightClient flightClient = new(_testWebFactory.GetChannel());
+ _flightSqlClient = new FlightSqlClient(flightClient);
+
+ _testUtils = new FlightSqlTestUtils(_testWebFactory, _flightStore);
+ }
+
+ #region Transactions
+
+ [Fact]
+ public async Task CommitTransactionAsync()
+ {
+ // Arrange
+ string transactionId = "sample-transaction-id";
+ var transaction = new Transaction(transactionId);
+
+ // Act
+ var streamCall = _flightSqlClient.CommitAsync(transaction);
+ var result = await streamCall.ResponseStream.ToListAsync();
+
+ // Assert
+ Assert.NotNull(result);
+ Assert.Equal(transaction.TransactionId, result.FirstOrDefault()?.Body);
+ }
+
+ [Fact]
+ public async Task BeginTransactionAsync()
+ {
+ // Arrange
+ string expectedTransactionId = "sample-transaction-id";
+
+ // Act
+ var transaction = await _flightSqlClient.BeginTransactionAsync();
+
+ // Assert
+ Assert.NotNull(transaction);
+ Assert.Equal(ByteString.CopyFromUtf8(expectedTransactionId), transaction.TransactionId);
+ }
+
+ [Fact]
+ public async Task RollbackTransactionAsync()
+ {
+ // Arrange
+ string transactionId = "sample-transaction-id";
+ var transaction = new Transaction(transactionId);
+
+ // Act
+ var streamCall = _flightSqlClient.RollbackAsync(transaction);
+ var result = await streamCall.ResponseStream.ToListAsync();
+
+ // Assert
+ Assert.NotNull(transaction);
+ Assert.Equal(result.FirstOrDefault()?.Body, transaction.TransactionId);
+ }
+
+ #endregion
+
+ #region PreparedStatement
+
+ [Fact]
+ public async Task PreparedAsync()
+ {
+ // Arrange
+ string query = "INSERT INTO users (id, name) VALUES (1, 'John Doe')";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+
+ // Create a sample schema for the dataset and parameters
+ var schema = new Schema.Builder()
+ .Field(f => f.Name("id").DataType(Int32Type.Default))
+ .Field(f => f.Name("name").DataType(StringType.Default))
+ .Build();
+
+ var recordBatch = new RecordBatch(schema, new Array[]
+ {
+ new Int32Array.Builder().Append(1).Build(),
+ new StringArray.Builder().Append("John Doe").Build()
+ }, 1);
+
+ var flightHolder = new FlightHolder(flightDescriptor, schema, _testWebFactory.GetAddress());
+ flightHolder.AddBatch(new RecordBatchWithMetadata(recordBatch));
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ var datasetSchemaBytes = SchemaExtensions.SerializeSchema(schema);
+ var parameterSchemaBytes = SchemaExtensions.SerializeSchema(schema);
+
+ var preparedStatementResponse = new ActionCreatePreparedStatementResult
+ {
+ PreparedStatementHandle = ByteString.CopyFromUtf8("prepared-handle"),
+ DatasetSchema = ByteString.CopyFrom(datasetSchemaBytes),
+ ParameterSchema = ByteString.CopyFrom(parameterSchemaBytes)
+ };
+
+ // Act
+ var preparedStatement = await _flightSqlClient.PrepareAsync(query, transaction);
+ var deserializedDatasetSchema = SchemaExtensions.DeserializeSchema(preparedStatementResponse.DatasetSchema.ToByteArray());
+ var deserializedParameterSchema = SchemaExtensions.DeserializeSchema(preparedStatementResponse.ParameterSchema.ToByteArray());
+
+ // Assert
+ Assert.NotNull(preparedStatement);
+ Assert.NotNull(deserializedDatasetSchema);
+ Assert.NotNull(deserializedParameterSchema);
+ CompareSchemas(schema, deserializedDatasetSchema);
+ CompareSchemas(schema, deserializedParameterSchema);
+ }
+
+ #endregion
+
+ [Fact]
+ public async Task ExecuteUpdateAsync()
+ {
+ // Arrange
+ string query = "UPDATE test_table SET column1 = 'value' WHERE column2 = 'condition'";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+
+ var schema = new Schema.Builder()
+ .Field(f => f.Name("id").DataType(Int32Type.Default))
+ .Field(f => f.Name("name").DataType(StringType.Default))
+ .Build();
+
+ var recordBatch = new RecordBatch(schema, new Array[]
+ {
+ new Int32Array.Builder().Append(1).Build(),
+ new StringArray.Builder().Append("John Doe").Build()
+ }, 1);
+
+
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema, _testWebFactory.GetAddress());
+ flightHolder.AddBatch(new RecordBatchWithMetadata(recordBatch));
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ long affectedRows = await _flightSqlClient.ExecuteUpdateAsync(query, transaction);
+
+ // Assert
+ Assert.Equal(1, affectedRows);
+ }
+
+ [Fact]
+ public async Task ExecuteAsync()
+ {
+ // Arrange
+ string query = "SELECT * FROM test_table";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema, _testWebFactory.GetAddress());
+ flightHolder.AddBatch(new RecordBatchWithMetadata(recordBatch));
+
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ var flightInfo = await _flightSqlClient.ExecuteAsync(query, transaction);
+
+ // Assert
+ Assert.NotNull(flightInfo);
+ Assert.Single(flightInfo.Endpoints);
+ }
+
+ [Fact]
+ public async Task ExecuteAsync_ShouldReturnFlightInfo_WhenValidInputsAreProvided()
+ {
+ // Arrange
+ string query = "SELECT * FROM test_table";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ var flightInfo = await _flightSqlClient.ExecuteAsync(query, transaction);
+
+ // Assert
+ Assert.NotNull(flightInfo);
+ Assert.IsType<FlightInfo>(flightInfo);
+ }
+
+ [Fact]
+ public async Task ExecuteAsync_ShouldThrowArgumentException_WhenQueryIsEmpty()
+ {
+ // Arrange
+ string emptyQuery = string.Empty;
+ var transaction = new Transaction("sample-transaction-id");
+
+ // Act & Assert
+ await Assert.ThrowsAsync<ArgumentException>(async () =>
+ await _flightSqlClient.ExecuteAsync(emptyQuery, transaction));
+ }
+
+ [Fact]
+ public async Task ExecuteAsync_ShouldReturnFlightInfo_WhenTransactionIsNoTransaction()
+ {
+ // Arrange
+ string query = "SELECT * FROM test_table";
+ var transaction = Transaction.NoTransaction;
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ var flightInfo = await _flightSqlClient.ExecuteAsync(query, transaction);
+
+ // Assert
+ Assert.NotNull(flightInfo);
+ Assert.IsType<FlightInfo>(flightInfo);
+ }
+
+
+ [Fact]
+ public async Task GetFlightInfoAsync()
+ {
+ // Arrange
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+ // Act
+ var flightInfo = await _flightSqlClient.GetFlightInfoAsync(flightDescriptor);
+
+ // Assert
+ Assert.NotNull(flightInfo);
+ }
+
+ [Fact]
+ public async Task GetExecuteSchemaAsync()
+ {
+ // Arrange
+ string query = "SELECT * FROM test_table";
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ Schema resultSchema =
+ await _flightSqlClient.GetExecuteSchemaAsync(query, new Transaction("sample-transaction-id"));
+
+ // Assert
+ Assert.NotNull(resultSchema);
+ Assert.Equal(recordBatch.Schema.FieldsList.Count, resultSchema.FieldsList.Count);
+ CompareSchemas(resultSchema, recordBatch.Schema);
+ }
+
+ [Fact]
+ public async Task GetCatalogsAsync()
+ {
+ // Arrange
+ var options = new FlightCallOptions();
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ var result = await _flightSqlClient.GetCatalogsAsync(options);
+
+ // Assert
+ Assert.NotNull(result);
+ Assert.Equal(flightHolder.GetFlightInfo().Endpoints.Count, result.Endpoints.Count);
+ Assert.Equal(flightDescriptor, result.Descriptor);
+ }
+
+ [Fact]
+ public async Task GetSchemaAsync()
+ {
+ // Arrange
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ var result = await _flightSqlClient.GetSchemaAsync(flightDescriptor);
+
+ // Assert
+ Assert.NotNull(result);
+ Assert.Equal(recordBatch.Schema.FieldsList.Count, result.FieldsList.Count);
+ CompareSchemas(result, recordBatch.Schema);
+ }
+
+ [Fact]
+ public async Task GetDbSchemasAsync()
+ {
+ // Arrange
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+ string catalog = "test-catalog";
+ string dbSchemaFilterPattern = "test-schema-pattern";
+
+ // Act
+ var result = await _flightSqlClient.GetDbSchemasAsync(catalog, dbSchemaFilterPattern);
+
+ // Assert
+ Assert.NotNull(result);
+ var expectedFlightInfo = flightHolder.GetFlightInfo();
+ Assert.Equal(recordBatch.Schema.FieldsList.Count, result.Schema.FieldsList.Count);
+ Assert.Equal(expectedFlightInfo.Descriptor.Command, result.Descriptor.Command);
+ Assert.Equal(expectedFlightInfo.Descriptor.Type, result.Descriptor.Type);
+ Assert.Equal(expectedFlightInfo.Schema.FieldsList.Count, result.Schema.FieldsList.Count);
+ Assert.Equal(expectedFlightInfo.Endpoints.Count, result.Endpoints.Count);
+
+ for (int i = 0; i < expectedFlightInfo.Schema.FieldsList.Count; i++)
+ {
+ var expectedField = expectedFlightInfo.Schema.FieldsList[i];
+ var actualField = result.Schema.FieldsList[i];
+
+ Assert.Equal(expectedField.Name, actualField.Name);
+ Assert.Equal(expectedField.DataType, actualField.DataType);
+ Assert.Equal(expectedField.IsNullable, actualField.IsNullable);
+ Assert.Equal(expectedField.Metadata?.Count ?? 0, actualField.Metadata?.Count ?? 0);
+ }
+
+ for (int i = 0; i < expectedFlightInfo.Endpoints.Count; i++)
+ {
+ var expectedEndpoint = expectedFlightInfo.Endpoints[i];
+ var actualEndpoint = result.Endpoints[i];
+
+ Assert.Equal(expectedEndpoint.Ticket, actualEndpoint.Ticket);
+ Assert.Equal(expectedEndpoint.Locations.Count(), actualEndpoint.Locations.Count());
+ }
+ }
+
+ [Fact]
+ public async Task GetPrimaryKeysAsync()
+ {
+ // Arrange
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var tableRef = new TableRef { Catalog = "test-catalog", Table = "test-table", DbSchema = "test-schema" };
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ var result = await _flightSqlClient.GetPrimaryKeysAsync(tableRef);
+
+ // Assert
+ Assert.NotNull(result);
+ var expectedFlightInfo = flightHolder.GetFlightInfo();
+ Assert.Equal(expectedFlightInfo.Descriptor.Command, result.Descriptor.Command);
+ Assert.Equal(expectedFlightInfo.Descriptor.Type, result.Descriptor.Type);
+ Assert.Equal(expectedFlightInfo.Schema.FieldsList.Count, result.Schema.FieldsList.Count);
+
+ for (int i = 0; i < expectedFlightInfo.Schema.FieldsList.Count; i++)
+ {
+ var expectedField = expectedFlightInfo.Schema.FieldsList[i];
+ var actualField = result.Schema.FieldsList[i];
+ Assert.Equal(expectedField.Name, actualField.Name);
+ Assert.Equal(expectedField.DataType, actualField.DataType);
+ Assert.Equal(expectedField.IsNullable, actualField.IsNullable);
+ Assert.Equal(expectedField.Metadata?.Count ?? 0, actualField.Metadata?.Count ?? 0);
+ }
+
+ Assert.Equal(expectedFlightInfo.Endpoints.Count, result.Endpoints.Count);
+
+ for (int i = 0; i < expectedFlightInfo.Endpoints.Count; i++)
+ {
+ var expectedEndpoint = expectedFlightInfo.Endpoints[i];
+ var actualEndpoint = result.Endpoints[i];
+
+ Assert.Equal(expectedEndpoint.Ticket, actualEndpoint.Ticket);
+ Assert.Equal(expectedEndpoint.Locations.Count(), actualEndpoint.Locations.Count());
+ }
+ }
+
+ [Fact]
+ public async Task GetTablesAsync()
+ {
+ // Arrange
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ string catalog = "sample_catalog";
+ string dbSchemaFilterPattern = "sample_schema";
+ string tableFilterPattern = "sample_table";
+ bool includeSchema = true;
+ var tableTypes = new List<string> { "BASE TABLE" };
+
+ // Act
+ var result = await _flightSqlClient.GetTablesAsync(catalog, dbSchemaFilterPattern, tableFilterPattern,
+ includeSchema, tableTypes);
+
+ // Assert
+ Assert.NotNull(result);
+ Assert.Single(result);
+
+ var expectedFlightInfo = flightHolder.GetFlightInfo();
+ var flightInfo = result.First();
+ Assert.Equal(expectedFlightInfo.Descriptor.Command, flightInfo.Descriptor.Command);
+ Assert.Equal(expectedFlightInfo.Descriptor.Type, flightInfo.Descriptor.Type);
+ Assert.Equal(expectedFlightInfo.Schema.FieldsList.Count, flightInfo.Schema.FieldsList.Count);
+ for (int i = 0; i < expectedFlightInfo.Schema.FieldsList.Count; i++)
+ {
+ var expectedField = expectedFlightInfo.Schema.FieldsList[i];
+ var actualField = flightInfo.Schema.FieldsList[i];
+ Assert.Equal(expectedField.Name, actualField.Name);
+ Assert.Equal(expectedField.DataType, actualField.DataType);
+ Assert.Equal(expectedField.IsNullable, actualField.IsNullable);
+ }
+
+ Assert.Equal(expectedFlightInfo.Endpoints.Count, flightInfo.Endpoints.Count);
+ }
+
+
Review Comment:
nit: extra blank line
##########
csharp/src/Apache.Arrow.Flight.Sql/Transaction.cs:
##########
@@ -0,0 +1,48 @@
+// 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.
+
+using Google.Protobuf;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class Transaction
Review Comment:
Should this be a value type?
##########
csharp/examples/Examples.sln:
##########
@@ -7,6 +7,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FluentBuilderExample", "Flu
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow", "..\src\Apache.Arrow\Apache.Arrow.csproj", "{1FE1DE95-FF6E-4895-82E7-909713C53524}"
EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "FlightClientExample", "FlightClientExample\FlightClientExample.csproj", "{CBB46C39-530D-465A-9367-4E771595209A}"
+EndProject
Review Comment:
FlightClientExample was added to Examples.sln but its files are not part of this pull request.
##########
csharp/src/Apache.Arrow.Flight/FlightInfoCancelRequest.cs:
##########
@@ -0,0 +1,38 @@
+using System;
Review Comment:
Added files must have the Apache header. It's missing from this file as well as FlightInfoCancelResult.cs.
##########
csharp/src/Apache.Arrow.Flight.Sql/Transaction.cs:
##########
@@ -0,0 +1,48 @@
+// 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.
+
+using Google.Protobuf;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class Transaction
+{
+ private static readonly ByteString TransactionIdDefaultValue = ByteString.Empty;
+ private ByteString? _transactionId;
+
+ public ByteString TransactionId
+ {
+ get => _transactionId ?? TransactionIdDefaultValue;
+ set => _transactionId = ProtoPreconditions.CheckNotNull(value, nameof(value));
+ }
+
+ public static readonly Transaction NoTransaction = new(TransactionIdDefaultValue);
+
+ public Transaction(ByteString transactionId)
+ {
+ TransactionId = transactionId;
+ }
+
+ public Transaction(string transactionId)
+ {
+ _transactionId = ByteString.CopyFromUtf8(transactionId);
+ }
+
+ public bool IsValid() => TransactionId.Length > 0;
+ public void ResetTransaction()
Review Comment:
`ResetTransaction` doesnt appear to be used.
##########
csharp/src/Apache.Arrow.Flight.Sql/Transaction.cs:
##########
@@ -0,0 +1,48 @@
+// 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.
+
+using Google.Protobuf;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class Transaction
+{
+ private static readonly ByteString TransactionIdDefaultValue = ByteString.Empty;
+ private ByteString? _transactionId;
+
+ public ByteString TransactionId
+ {
+ get => _transactionId ?? TransactionIdDefaultValue;
+ set => _transactionId = ProtoPreconditions.CheckNotNull(value, nameof(value));
+ }
+
+ public static readonly Transaction NoTransaction = new(TransactionIdDefaultValue);
+
+ public Transaction(ByteString transactionId)
+ {
+ TransactionId = transactionId;
+ }
+
+ public Transaction(string transactionId)
+ {
+ _transactionId = ByteString.CopyFromUtf8(transactionId);
+ }
+
+ public bool IsValid() => TransactionId.Length > 0;
Review Comment:
Should this be a property?
##########
csharp/src/Apache.Arrow.Flight.Sql/RecordBatchExtensions.cs:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Apache.Arrow.Ipc;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public static class RecordBatchExtensions
+{
+ /// <summary>
+ /// Converts a RecordBatch into an asynchronous stream of FlightData.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An asynchronous stream of FlightData objects.</returns>
+ public static async IAsyncEnumerable<FlightData> ToFlightDataStreamAsync(this RecordBatch recordBatch,
+ FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null)
+ {
+ throw new ArgumentNullException(nameof(recordBatch));
+ }
+
+ // Use a memory stream to write the Arrow RecordBatch into FlightData format
+ using var memoryStream = new MemoryStream();
+ var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+
+ // Write the RecordBatch to the stream
+ await writer.WriteRecordBatchAsync(recordBatch).ConfigureAwait(false);
+ await writer.WriteEndAsync().ConfigureAwait(false);
+
+ // Reset the memory stream position
+ memoryStream.Position = 0;
+
+ // Read back the data to create FlightData
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()),
Review Comment:
`ByteString.CopyFrom(memoryStream.ToArray())` makes an extra copy of the data. This can be avoided by doing `ByteString.CopyFrom(memoryStream.GetBuffer(), 0, (int)memoryStream.Length)`.
##########
csharp/src/Apache.Arrow.Flight.Sql/PreparedStatement.cs:
##########
@@ -0,0 +1,383 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Sql.Client;
+using Arrow.Flight.Protocol.Sql;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class PreparedStatement : IDisposable
+{
+ private readonly FlightSqlClient _client;
+ private readonly string _handle;
+ private Schema _datasetSchema;
+ private Schema _parameterSchema;
+ private RecordBatch? _recordsBatch;
+ private bool _isClosed;
+ public bool IsClosed => _isClosed;
+ public string Handle => _handle;
+ public RecordBatch? ParametersBatch => _recordsBatch;
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PreparedStatement"/> class.
+ /// </summary>
+ /// <param name="client">The Flight SQL client used for executing SQL operations.</param>
+ /// <param name="handle">The handle representing the prepared statement.</param>
+ /// <param name="datasetSchema">The schema of the result dataset.</param>
+ /// <param name="parameterSchema">The schema of the parameters for this prepared statement.</param>
+ public PreparedStatement(FlightSqlClient client, string handle, Schema datasetSchema, Schema parameterSchema)
+ {
+ _client = client ?? throw new ArgumentNullException(nameof(client));
+ _handle = handle ?? throw new ArgumentNullException(nameof(handle));
+ _datasetSchema = datasetSchema ?? throw new ArgumentNullException(nameof(datasetSchema));
+ _parameterSchema = parameterSchema ?? throw new ArgumentNullException(nameof(parameterSchema));
+ _isClosed = false;
+ }
+
+ /// <summary>
+ /// Retrieves the schema associated with the prepared statement asynchronously.
+ /// </summary>
+ /// <param name="options">The options used to configure the Flight call.</param>
+ /// <returns>A task representing the asynchronous operation, which returns the schema of the result set.</returns>
+ /// <exception cref="InvalidOperationException">Thrown when the schema is empty or invalid.</exception>
+ public async Task<Schema> GetSchemaAsync(FlightCallOptions? options = default)
+ {
+ EnsureStatementIsNotClosed();
+
+ try
+ {
+ var command = new CommandPreparedStatementQuery
+ {
+ PreparedStatementHandle = ByteString.CopyFrom(_handle, Encoding.UTF8)
+ };
+ var descriptor = FlightDescriptor.CreateCommandDescriptor(command.PackAndSerialize());
+ var schema = await _client.GetSchemaAsync(descriptor, options).ConfigureAwait(false);
+ if (schema == null || !schema.FieldsList.Any())
+ {
+ throw new InvalidOperationException("Schema is empty or invalid.");
+ }
+ return schema;
+ }
+ catch (RpcException ex)
+ {
+ throw new InvalidOperationException("Failed to retrieve the schema for the prepared statement", ex);
+ }
+ }
+
+
Review Comment:
nit: extra blank line
##########
csharp/src/Apache.Arrow.Flight.Sql/RecordBatchExtensions.cs:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Apache.Arrow.Ipc;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public static class RecordBatchExtensions
+{
+ /// <summary>
+ /// Converts a RecordBatch into an asynchronous stream of FlightData.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An asynchronous stream of FlightData objects.</returns>
+ public static async IAsyncEnumerable<FlightData> ToFlightDataStreamAsync(this RecordBatch recordBatch,
+ FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null)
+ {
+ throw new ArgumentNullException(nameof(recordBatch));
+ }
+
+ // Use a memory stream to write the Arrow RecordBatch into FlightData format
+ using var memoryStream = new MemoryStream();
+ var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+
+ // Write the RecordBatch to the stream
+ await writer.WriteRecordBatchAsync(recordBatch).ConfigureAwait(false);
+ await writer.WriteEndAsync().ConfigureAwait(false);
+
+ // Reset the memory stream position
+ memoryStream.Position = 0;
+
+ // Read back the data to create FlightData
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()),
+ ByteString.CopyFrom(memoryStream.ToArray()));
Review Comment:
Is it correct for the `dataBody` and `dataHeader` arguments to have the same value? The sync version passes `ByteString.Empty` as the `dataHeader` parameter.
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightCallOptions.cs:
##########
@@ -0,0 +1,47 @@
+// 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.
+
+using System;
+using System.Buffers;
+using System.Threading;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class FlightCallOptions
+{
+ public FlightCallOptions()
+ {
+ Timeout = TimeSpan.FromSeconds(-1);
+ }
+ // Implement any necessary options for RPC calls
+ public Metadata Headers { get; set; } = new();
+
+ /// <summary>
+ /// Gets or sets a token to enable interactive user cancellation of long-running requests.
+ /// </summary>
+ public CancellationToken StopToken { get; set; }
+
+ /// <summary>
+ /// Gets or sets the optional timeout for this call.
+ /// Negative durations mean an implementation-defined default behavior will be used instead.
+ /// </summary>
+ public TimeSpan Timeout { get; set; }
+
+ /// <summary>
+ /// Gets or sets an optional memory manager to control where to allocate incoming data.
+ /// </summary>
+ public MemoryManager<ArrowBuffer>? MemoryManager { get; set; }
Review Comment:
Nothing references `MemoryManager`. To the extent that these are aspirational, it would be better to remove them now and add them back in a separate PR once implemented instead of putting them in unused and giving the impression that they might do something.
##########
csharp/src/Apache.Arrow.Flight.Sql/PreparedStatement.cs:
##########
@@ -0,0 +1,383 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Sql.Client;
+using Arrow.Flight.Protocol.Sql;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class PreparedStatement : IDisposable
+{
+ private readonly FlightSqlClient _client;
+ private readonly string _handle;
+ private Schema _datasetSchema;
+ private Schema _parameterSchema;
Review Comment:
Neither of these schemas appear to be used.
##########
csharp/src/Apache.Arrow.Flight.Sql/RecordBatchExtensions.cs:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Apache.Arrow.Ipc;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public static class RecordBatchExtensions
+{
+ /// <summary>
+ /// Converts a RecordBatch into an asynchronous stream of FlightData.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An asynchronous stream of FlightData objects.</returns>
+ public static async IAsyncEnumerable<FlightData> ToFlightDataStreamAsync(this RecordBatch recordBatch,
+ FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null)
+ {
+ throw new ArgumentNullException(nameof(recordBatch));
+ }
+
+ // Use a memory stream to write the Arrow RecordBatch into FlightData format
+ using var memoryStream = new MemoryStream();
+ var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+
+ // Write the RecordBatch to the stream
+ await writer.WriteRecordBatchAsync(recordBatch).ConfigureAwait(false);
+ await writer.WriteEndAsync().ConfigureAwait(false);
+
+ // Reset the memory stream position
+ memoryStream.Position = 0;
+
+ // Read back the data to create FlightData
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()),
+ ByteString.CopyFrom(memoryStream.ToArray()));
+ yield return flightData;
+ }
+
+ /// <summary>
+ /// Converts a RecordBatch into an IAsyncStreamReader<FlightData>.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An IAsyncStreamReader of FlightData.</returns>
+ public static IAsyncStreamReader<FlightData> ToFlightDataStream(this RecordBatch recordBatch, FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null) throw new ArgumentNullException(nameof(recordBatch));
+ if (flightDescriptor == null) throw new ArgumentNullException(nameof(flightDescriptor));
+
+ var channel = Channel.CreateUnbounded<FlightData>();
+
+ try
+ {
+ if (recordBatch.Schema == null || !recordBatch.Schema.FieldsList.Any())
+ {
+ throw new InvalidOperationException("The record batch has an invalid or empty schema.");
+ }
+
+ using var memoryStream = new MemoryStream();
+ using var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+ writer.WriteRecordBatch(recordBatch);
+ writer.WriteEnd();
+ memoryStream.Position = 0;
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()), ByteString.Empty, ByteString.Empty);
+ if (flightData.DataBody.IsEmpty)
+ {
+ throw new InvalidOperationException(
+ "The generated FlightData is empty. Check the RecordBatch content.");
+ }
+
+ channel.Writer.TryWrite(flightData);
Review Comment:
What would it mean if this returned `false`?
##########
csharp/src/Apache.Arrow.Flight.Sql/RecordBatchExtensions.cs:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Apache.Arrow.Ipc;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public static class RecordBatchExtensions
+{
+ /// <summary>
+ /// Converts a RecordBatch into an asynchronous stream of FlightData.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An asynchronous stream of FlightData objects.</returns>
+ public static async IAsyncEnumerable<FlightData> ToFlightDataStreamAsync(this RecordBatch recordBatch,
+ FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null)
+ {
+ throw new ArgumentNullException(nameof(recordBatch));
+ }
+
+ // Use a memory stream to write the Arrow RecordBatch into FlightData format
+ using var memoryStream = new MemoryStream();
+ var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+
+ // Write the RecordBatch to the stream
+ await writer.WriteRecordBatchAsync(recordBatch).ConfigureAwait(false);
+ await writer.WriteEndAsync().ConfigureAwait(false);
+
+ // Reset the memory stream position
+ memoryStream.Position = 0;
+
+ // Read back the data to create FlightData
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()),
+ ByteString.CopyFrom(memoryStream.ToArray()));
+ yield return flightData;
+ }
+
+ /// <summary>
+ /// Converts a RecordBatch into an IAsyncStreamReader<FlightData>.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An IAsyncStreamReader of FlightData.</returns>
+ public static IAsyncStreamReader<FlightData> ToFlightDataStream(this RecordBatch recordBatch, FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null) throw new ArgumentNullException(nameof(recordBatch));
+ if (flightDescriptor == null) throw new ArgumentNullException(nameof(flightDescriptor));
+
+ var channel = Channel.CreateUnbounded<FlightData>();
+
+ try
+ {
+ if (recordBatch.Schema == null || !recordBatch.Schema.FieldsList.Any())
+ {
+ throw new InvalidOperationException("The record batch has an invalid or empty schema.");
+ }
+
+ using var memoryStream = new MemoryStream();
+ using var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+ writer.WriteRecordBatch(recordBatch);
+ writer.WriteEnd();
+ memoryStream.Position = 0;
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()), ByteString.Empty, ByteString.Empty);
+ if (flightData.DataBody.IsEmpty)
Review Comment:
In addition to the same comment about `ByteString.CopyFrom`, it should be possible to check for an empty body before creating the `FlightData` by looking at `memoryStream.Length`. Does this length check belong in the async version too?
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightExtensions.cs:
##########
@@ -0,0 +1,81 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
+
+namespace Apache.Arrow.Flight.Sql;
+
+internal static class FlightExtensions
+{
+ public static byte[] PackAndSerialize(this IMessage command) => Any.Pack(command).ToByteArray();
+ public static T ParseAndUnpack<T>(this ByteString source) where T : IMessage<T>, new() => Any.Parser.ParseFrom(source).Unpack<T>();
+
+ public static IEnumerable<long> ExtractRowCount(this RecordBatch batch)
+ {
+ foreach (var array in batch.Arrays)
+ {
+ var values = ExtractValues(array);
+ foreach (var value in values)
+ {
+ yield return value switch
+ {
+ long l => l,
+ int i => i != 0 ? i : 0,
Review Comment:
Isn't this just `int i => i`?
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightCallOptions.cs:
##########
@@ -0,0 +1,47 @@
+// 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.
+
+using System;
+using System.Buffers;
+using System.Threading;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class FlightCallOptions
+{
+ public FlightCallOptions()
+ {
+ Timeout = TimeSpan.FromSeconds(-1);
+ }
+ // Implement any necessary options for RPC calls
+ public Metadata Headers { get; set; } = new();
+
+ /// <summary>
+ /// Gets or sets a token to enable interactive user cancellation of long-running requests.
+ /// </summary>
+ public CancellationToken StopToken { get; set; }
Review Comment:
Nothing references `StopToken`
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightCallOptions.cs:
##########
@@ -0,0 +1,47 @@
+// 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.
+
+using System;
+using System.Buffers;
+using System.Threading;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class FlightCallOptions
+{
+ public FlightCallOptions()
+ {
+ Timeout = TimeSpan.FromSeconds(-1);
+ }
Review Comment:
nit; insert blank line after closing bracket
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightExtensions.cs:
##########
@@ -0,0 +1,81 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
+
+namespace Apache.Arrow.Flight.Sql;
+
+internal static class FlightExtensions
+{
+ public static byte[] PackAndSerialize(this IMessage command) => Any.Pack(command).ToByteArray();
+ public static T ParseAndUnpack<T>(this ByteString source) where T : IMessage<T>, new() => Any.Parser.ParseFrom(source).Unpack<T>();
+
+ public static IEnumerable<long> ExtractRowCount(this RecordBatch batch)
Review Comment:
I have to admit, I can't tell what the semantics of this are supposed to be :(. I did a quick search to see what the result of `ExecuteUpdate` was supposed to look like but that didn't help. Putting the explanation in a comment might be worthwhile.
##########
csharp/test/Apache.Arrow.Flight.Sql.Tests/FlightSqlPreparedStatementTests.cs:
##########
@@ -0,0 +1,227 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Client;
+using Apache.Arrow.Flight.Sql.Client;
+using Apache.Arrow.Flight.TestWeb;
+using Apache.Arrow.Types;
+using Arrow.Flight.Protocol.Sql;
+using Google.Protobuf;
+using Grpc.Core;
+using Xunit;
+
+namespace Apache.Arrow.Flight.Sql.Tests
+{
+ public class FlightSqlPreparedStatementTests
+ {
+ readonly TestFlightSqlWebFactory _testWebFactory;
+ readonly FlightStore _flightStore;
+ readonly FlightSqlClient _flightSqlClient;
+ private readonly PreparedStatement _preparedStatement;
+ private readonly Schema _schema;
+ private readonly FlightDescriptor _flightDescriptor;
+ private readonly RecordBatch _parameterBatch;
+
+ public FlightSqlPreparedStatementTests()
+ {
+ _flightStore = new FlightStore();
+ _testWebFactory = new TestFlightSqlWebFactory(_flightStore);
+ _flightSqlClient = new FlightSqlClient(new FlightClient(_testWebFactory.GetChannel()));
+
+ _flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test-query");
+ _schema = CreateSchema();
+ _parameterBatch = CreateParameterBatch();
+ _preparedStatement = new PreparedStatement(_flightSqlClient, "test-handle-guid", _schema, _schema);
+ }
+
+ private static Schema CreateSchema()
+ {
+ return new Schema.Builder()
+ .Field(f => f.Name("DATA_TYPE_ID").DataType(Int32Type.Default).Nullable(false))
+ .Build();
+ }
+
+ private RecordBatch CreateParameterBatch()
+ {
+ return new RecordBatch(_schema,
+ new IArrowArray[]
+ {
+ new Int32Array.Builder().AppendRange(new[] { 1, 2, 3 }).Build(),
+ new StringArray.Builder().AppendRange(new[] { "INTEGER", "VARCHAR", "BOOLEAN" }).Build(),
+ new Int32Array.Builder().AppendRange(new[] { 32, 255, 1 }).Build()
+ }, 3);
+ }
+
+
Review Comment:
nit: extra blank line
##########
csharp/test/Apache.Arrow.Flight.Sql.Tests/FlightSqlTestUtils.cs:
##########
@@ -0,0 +1,64 @@
+// 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.
+
+using System.Linq;
+using Apache.Arrow.Flight.Tests;
+using Apache.Arrow.Flight.TestWeb;
+
+namespace Apache.Arrow.Flight.Sql.Tests;
+
+public class FlightSqlTestUtils
+{
+ private readonly TestFlightSqlWebFactory _testWebFactory;
+ private readonly FlightStore _flightStore;
+
+ public FlightSqlTestUtils(TestFlightSqlWebFactory testWebFactory, FlightStore flightStore)
+ {
+ _testWebFactory = testWebFactory;
+ _flightStore = flightStore;
+ }
+
+ public RecordBatch CreateTestBatch(int startValue, int length)
+ {
+ var batchBuilder = new RecordBatch.Builder();
+ Int32Array.Builder builder = new();
+ for (int i = 0; i < length; i++)
+ {
+ builder.Append(startValue + i);
+ }
+
+ batchBuilder.Append("test", true, builder.Build());
+ return batchBuilder.Build();
+ }
+
+
Review Comment:
nit: extra blank line
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightExtensions.cs:
##########
@@ -0,0 +1,81 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
+
+namespace Apache.Arrow.Flight.Sql;
+
+internal static class FlightExtensions
+{
+ public static byte[] PackAndSerialize(this IMessage command) => Any.Pack(command).ToByteArray();
+ public static T ParseAndUnpack<T>(this ByteString source) where T : IMessage<T>, new() => Any.Parser.ParseFrom(source).Unpack<T>();
+
+ public static IEnumerable<long> ExtractRowCount(this RecordBatch batch)
+ {
+ foreach (var array in batch.Arrays)
+ {
+ var values = ExtractValues(array);
+ foreach (var value in values)
+ {
+ yield return value switch
+ {
+ long l => l,
+ int i => i != 0 ? i : 0,
+ _ => 0L
+ };
+ }
+ }
+ }
+
+ private static IEnumerable<object?> ExtractValues(IArrowArray array)
+ {
+ return array switch
+ {
+ Int32Array int32Array => ExtractPrimitiveValues(int32Array),
+ Int64Array int64Array => ExtractPrimitiveValues(int64Array),
+ FloatArray floatArray => ExtractPrimitiveValues(floatArray),
+ BooleanArray booleanArray => ExtractBooleanValues(booleanArray),
+ StringArray stringArray => ExtractStringValues(stringArray),
+ _ => throw new NotSupportedException($"Array type {array.GetType().Name} is not supported.")
+ };
+ }
+
+ private static IEnumerable<object?> ExtractPrimitiveValues<T>(PrimitiveArray<T> array) where T : struct, IEquatable<T>
+ {
Review Comment:
The underlying arrays in all three cases should already be `IEnumerable<>`. Can they not be used directly?
##########
csharp/test/Apache.Arrow.Flight.Sql.Tests/FlightSqlClientTests.cs:
##########
@@ -0,0 +1,855 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Client;
+using Apache.Arrow.Flight.Sql.Client;
+using Apache.Arrow.Flight.TestWeb;
+using Apache.Arrow.Types;
+using Arrow.Flight.Protocol.Sql;
+using Google.Protobuf;
+using Grpc.Core.Utils;
+using Xunit;
+
+namespace Apache.Arrow.Flight.Sql.Tests;
+
+public class FlightSqlClientTests : IDisposable
+{
+ readonly TestFlightSqlWebFactory _testWebFactory;
+ readonly FlightStore _flightStore;
+ private readonly FlightSqlClient _flightSqlClient;
+ private readonly FlightSqlTestUtils _testUtils;
+
+ public FlightSqlClientTests()
+ {
+ _flightStore = new FlightStore();
+ _testWebFactory = new TestFlightSqlWebFactory(_flightStore);
+ FlightClient flightClient = new(_testWebFactory.GetChannel());
+ _flightSqlClient = new FlightSqlClient(flightClient);
+
+ _testUtils = new FlightSqlTestUtils(_testWebFactory, _flightStore);
+ }
+
+ #region Transactions
+
+ [Fact]
+ public async Task CommitTransactionAsync()
+ {
+ // Arrange
+ string transactionId = "sample-transaction-id";
+ var transaction = new Transaction(transactionId);
+
+ // Act
+ var streamCall = _flightSqlClient.CommitAsync(transaction);
+ var result = await streamCall.ResponseStream.ToListAsync();
+
+ // Assert
+ Assert.NotNull(result);
+ Assert.Equal(transaction.TransactionId, result.FirstOrDefault()?.Body);
+ }
+
+ [Fact]
+ public async Task BeginTransactionAsync()
+ {
+ // Arrange
+ string expectedTransactionId = "sample-transaction-id";
+
+ // Act
+ var transaction = await _flightSqlClient.BeginTransactionAsync();
+
+ // Assert
+ Assert.NotNull(transaction);
+ Assert.Equal(ByteString.CopyFromUtf8(expectedTransactionId), transaction.TransactionId);
+ }
+
+ [Fact]
+ public async Task RollbackTransactionAsync()
+ {
+ // Arrange
+ string transactionId = "sample-transaction-id";
+ var transaction = new Transaction(transactionId);
+
+ // Act
+ var streamCall = _flightSqlClient.RollbackAsync(transaction);
+ var result = await streamCall.ResponseStream.ToListAsync();
+
+ // Assert
+ Assert.NotNull(transaction);
+ Assert.Equal(result.FirstOrDefault()?.Body, transaction.TransactionId);
+ }
+
+ #endregion
+
+ #region PreparedStatement
+
+ [Fact]
+ public async Task PreparedAsync()
+ {
+ // Arrange
+ string query = "INSERT INTO users (id, name) VALUES (1, 'John Doe')";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+
+ // Create a sample schema for the dataset and parameters
+ var schema = new Schema.Builder()
+ .Field(f => f.Name("id").DataType(Int32Type.Default))
+ .Field(f => f.Name("name").DataType(StringType.Default))
+ .Build();
+
+ var recordBatch = new RecordBatch(schema, new Array[]
+ {
+ new Int32Array.Builder().Append(1).Build(),
+ new StringArray.Builder().Append("John Doe").Build()
+ }, 1);
+
+ var flightHolder = new FlightHolder(flightDescriptor, schema, _testWebFactory.GetAddress());
+ flightHolder.AddBatch(new RecordBatchWithMetadata(recordBatch));
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ var datasetSchemaBytes = SchemaExtensions.SerializeSchema(schema);
+ var parameterSchemaBytes = SchemaExtensions.SerializeSchema(schema);
+
+ var preparedStatementResponse = new ActionCreatePreparedStatementResult
+ {
+ PreparedStatementHandle = ByteString.CopyFromUtf8("prepared-handle"),
+ DatasetSchema = ByteString.CopyFrom(datasetSchemaBytes),
+ ParameterSchema = ByteString.CopyFrom(parameterSchemaBytes)
+ };
+
+ // Act
+ var preparedStatement = await _flightSqlClient.PrepareAsync(query, transaction);
+ var deserializedDatasetSchema = SchemaExtensions.DeserializeSchema(preparedStatementResponse.DatasetSchema.ToByteArray());
+ var deserializedParameterSchema = SchemaExtensions.DeserializeSchema(preparedStatementResponse.ParameterSchema.ToByteArray());
+
+ // Assert
+ Assert.NotNull(preparedStatement);
+ Assert.NotNull(deserializedDatasetSchema);
+ Assert.NotNull(deserializedParameterSchema);
+ CompareSchemas(schema, deserializedDatasetSchema);
+ CompareSchemas(schema, deserializedParameterSchema);
+ }
+
+ #endregion
+
+ [Fact]
+ public async Task ExecuteUpdateAsync()
+ {
+ // Arrange
+ string query = "UPDATE test_table SET column1 = 'value' WHERE column2 = 'condition'";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+
+ var schema = new Schema.Builder()
+ .Field(f => f.Name("id").DataType(Int32Type.Default))
+ .Field(f => f.Name("name").DataType(StringType.Default))
+ .Build();
+
+ var recordBatch = new RecordBatch(schema, new Array[]
+ {
+ new Int32Array.Builder().Append(1).Build(),
+ new StringArray.Builder().Append("John Doe").Build()
+ }, 1);
+
+
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema, _testWebFactory.GetAddress());
+ flightHolder.AddBatch(new RecordBatchWithMetadata(recordBatch));
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ long affectedRows = await _flightSqlClient.ExecuteUpdateAsync(query, transaction);
+
+ // Assert
+ Assert.Equal(1, affectedRows);
+ }
+
+ [Fact]
+ public async Task ExecuteAsync()
+ {
+ // Arrange
+ string query = "SELECT * FROM test_table";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema, _testWebFactory.GetAddress());
+ flightHolder.AddBatch(new RecordBatchWithMetadata(recordBatch));
+
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ var flightInfo = await _flightSqlClient.ExecuteAsync(query, transaction);
+
+ // Assert
+ Assert.NotNull(flightInfo);
+ Assert.Single(flightInfo.Endpoints);
+ }
+
+ [Fact]
+ public async Task ExecuteAsync_ShouldReturnFlightInfo_WhenValidInputsAreProvided()
+ {
+ // Arrange
+ string query = "SELECT * FROM test_table";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ var flightInfo = await _flightSqlClient.ExecuteAsync(query, transaction);
+
+ // Assert
+ Assert.NotNull(flightInfo);
+ Assert.IsType<FlightInfo>(flightInfo);
+ }
+
+ [Fact]
+ public async Task ExecuteAsync_ShouldThrowArgumentException_WhenQueryIsEmpty()
+ {
+ // Arrange
+ string emptyQuery = string.Empty;
+ var transaction = new Transaction("sample-transaction-id");
+
+ // Act & Assert
+ await Assert.ThrowsAsync<ArgumentException>(async () =>
+ await _flightSqlClient.ExecuteAsync(emptyQuery, transaction));
+ }
+
+ [Fact]
+ public async Task ExecuteAsync_ShouldReturnFlightInfo_WhenTransactionIsNoTransaction()
+ {
+ // Arrange
+ string query = "SELECT * FROM test_table";
+ var transaction = Transaction.NoTransaction;
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+ var recordBatch = _testUtils.CreateTestBatch(0, 100);
+ var flightHolder = new FlightHolder(flightDescriptor, recordBatch.Schema,
+ _testWebFactory.GetAddress());
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ // Act
+ var flightInfo = await _flightSqlClient.ExecuteAsync(query, transaction);
+
+ // Assert
+ Assert.NotNull(flightInfo);
+ Assert.IsType<FlightInfo>(flightInfo);
+ }
+
+
Review Comment:
nit: extra blank line
##########
csharp/test/Apache.Arrow.Flight.Sql.Tests/FlightSqlClientTests.cs:
##########
@@ -0,0 +1,855 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Client;
+using Apache.Arrow.Flight.Sql.Client;
+using Apache.Arrow.Flight.TestWeb;
+using Apache.Arrow.Types;
+using Arrow.Flight.Protocol.Sql;
+using Google.Protobuf;
+using Grpc.Core.Utils;
+using Xunit;
+
+namespace Apache.Arrow.Flight.Sql.Tests;
+
+public class FlightSqlClientTests : IDisposable
+{
+ readonly TestFlightSqlWebFactory _testWebFactory;
+ readonly FlightStore _flightStore;
+ private readonly FlightSqlClient _flightSqlClient;
+ private readonly FlightSqlTestUtils _testUtils;
+
+ public FlightSqlClientTests()
+ {
+ _flightStore = new FlightStore();
+ _testWebFactory = new TestFlightSqlWebFactory(_flightStore);
+ FlightClient flightClient = new(_testWebFactory.GetChannel());
+ _flightSqlClient = new FlightSqlClient(flightClient);
+
+ _testUtils = new FlightSqlTestUtils(_testWebFactory, _flightStore);
+ }
+
+ #region Transactions
+
+ [Fact]
+ public async Task CommitTransactionAsync()
+ {
+ // Arrange
+ string transactionId = "sample-transaction-id";
+ var transaction = new Transaction(transactionId);
+
+ // Act
+ var streamCall = _flightSqlClient.CommitAsync(transaction);
+ var result = await streamCall.ResponseStream.ToListAsync();
+
+ // Assert
+ Assert.NotNull(result);
+ Assert.Equal(transaction.TransactionId, result.FirstOrDefault()?.Body);
+ }
+
+ [Fact]
+ public async Task BeginTransactionAsync()
+ {
+ // Arrange
+ string expectedTransactionId = "sample-transaction-id";
+
+ // Act
+ var transaction = await _flightSqlClient.BeginTransactionAsync();
+
+ // Assert
+ Assert.NotNull(transaction);
+ Assert.Equal(ByteString.CopyFromUtf8(expectedTransactionId), transaction.TransactionId);
+ }
+
+ [Fact]
+ public async Task RollbackTransactionAsync()
+ {
+ // Arrange
+ string transactionId = "sample-transaction-id";
+ var transaction = new Transaction(transactionId);
+
+ // Act
+ var streamCall = _flightSqlClient.RollbackAsync(transaction);
+ var result = await streamCall.ResponseStream.ToListAsync();
+
+ // Assert
+ Assert.NotNull(transaction);
+ Assert.Equal(result.FirstOrDefault()?.Body, transaction.TransactionId);
+ }
+
+ #endregion
+
+ #region PreparedStatement
+
+ [Fact]
+ public async Task PreparedAsync()
+ {
+ // Arrange
+ string query = "INSERT INTO users (id, name) VALUES (1, 'John Doe')";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+
+ // Create a sample schema for the dataset and parameters
+ var schema = new Schema.Builder()
+ .Field(f => f.Name("id").DataType(Int32Type.Default))
+ .Field(f => f.Name("name").DataType(StringType.Default))
+ .Build();
+
+ var recordBatch = new RecordBatch(schema, new Array[]
+ {
+ new Int32Array.Builder().Append(1).Build(),
+ new StringArray.Builder().Append("John Doe").Build()
+ }, 1);
+
+ var flightHolder = new FlightHolder(flightDescriptor, schema, _testWebFactory.GetAddress());
+ flightHolder.AddBatch(new RecordBatchWithMetadata(recordBatch));
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ var datasetSchemaBytes = SchemaExtensions.SerializeSchema(schema);
+ var parameterSchemaBytes = SchemaExtensions.SerializeSchema(schema);
+
+ var preparedStatementResponse = new ActionCreatePreparedStatementResult
+ {
+ PreparedStatementHandle = ByteString.CopyFromUtf8("prepared-handle"),
+ DatasetSchema = ByteString.CopyFrom(datasetSchemaBytes),
+ ParameterSchema = ByteString.CopyFrom(parameterSchemaBytes)
+ };
+
+ // Act
+ var preparedStatement = await _flightSqlClient.PrepareAsync(query, transaction);
+ var deserializedDatasetSchema = SchemaExtensions.DeserializeSchema(preparedStatementResponse.DatasetSchema.ToByteArray());
+ var deserializedParameterSchema = SchemaExtensions.DeserializeSchema(preparedStatementResponse.ParameterSchema.ToByteArray());
+
+ // Assert
+ Assert.NotNull(preparedStatement);
+ Assert.NotNull(deserializedDatasetSchema);
+ Assert.NotNull(deserializedParameterSchema);
+ CompareSchemas(schema, deserializedDatasetSchema);
+ CompareSchemas(schema, deserializedParameterSchema);
+ }
+
+ #endregion
+
+ [Fact]
+ public async Task ExecuteUpdateAsync()
+ {
+ // Arrange
+ string query = "UPDATE test_table SET column1 = 'value' WHERE column2 = 'condition'";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+
+ var schema = new Schema.Builder()
+ .Field(f => f.Name("id").DataType(Int32Type.Default))
+ .Field(f => f.Name("name").DataType(StringType.Default))
+ .Build();
+
+ var recordBatch = new RecordBatch(schema, new Array[]
+ {
+ new Int32Array.Builder().Append(1).Build(),
+ new StringArray.Builder().Append("John Doe").Build()
+ }, 1);
+
+
Review Comment:
nit: extra blank line
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2616200400
> > Any updates on this? We are also looking to use .NET with a Flight SQL client and are eager to get this functionality.
>
> I'm sorry about the delay; I was very much wanting to get the review done before today's fork but first work, then vacation, then a lengthy internet outage at home got in the way. I will definitely make time for a review later this week.
Hi Curt,
Is it possible to review and if approved merge my PR, it's already waiting for more than 2 months.
--
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
Re: [PR] Feat/flight sql client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1847541597
##########
csharp/src/Apache.Arrow.Flight.Sql/Apache.Arrow.Flight.Sql.csproj:
##########
@@ -5,7 +5,8 @@
</PropertyGroup>
<ItemGroup>
- <PackageReference Include="Grpc.Tools" Version="2.67.0" PrivateAssets="All" />
+ <PackageReference Include="Grpc.Tools" Version="2.66.0" PrivateAssets="All" />
Review Comment:
This version number shouldn't be reverted
--
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
Re: [PR] Feat/flight sql client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2485221173
> Thanks for a very large change! Could you please add the official Apache header to all the source files and open an issue to link this to?
Sure, sir, I will do everything mentioned here. Thank you for your guidelines.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "kou (via GitHub)" <gi...@apache.org>.
kou commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2917689771
My current proposal is https://lists.apache.org/thread/y6zq26qb91b25lvs9cc7kb297lxdsjcl .
(We move `csharp/` in apache/arrow to apache/arrow-dotnet BEFORE we merge this. We continue this in apache/arrow-dotnet.)
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-3045691672
> My current proposal is https://lists.apache.org/thread/y6zq26qb91b25lvs9cc7kb297lxdsjcl . (We move `csharp/` in apache/arrow to apache/arrow-dotnet BEFORE we merge this. We continue this in apache/arrow-dotnet.)
@CurtHagenlocher Hi, still haven't been accepted yet, can you please assist with it?
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "kou (via GitHub)" <gi...@apache.org>.
kou commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2785066214
@CurtHagenlocher We should do https://github.com/apache/arrow-dotnet/issues/1 after we merge this, right?
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "zgramana (via GitHub)" <gi...@apache.org>.
zgramana commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2783805813
> Thanks again for such a comprehensive change! I'm very sorry it took me this long to get to it. I've left a bunch of feedback of various degrees of seriousness. You'll also likely need to rebase given the amount of time that has passed (and the presence of a merge conflict).
@CurtHagenlocher since this PR no longer has a merge conflict, and it get your final
--
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
Re: [PR] Feat/flight sql client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1849645866
##########
cpp/src/arrow/flight/sql/client.h:
##########
Review Comment:
What changes did I do?
##########
cpp/CMakeLists.txt:
##########
Review Comment:
What changes did I do?
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2505806254
> Could you revert all `cpp/` changes?
seems that it's done
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-3070370780
> @HackPoint, I do think it's important to add cancellation tokens to methods which themselves call other methods that already take a cancellation token -- particularly when the method in question is a public API. Even if no one ever ends up using it, the cost is quite small. By contrast, if one needs to be added later then we end up needing to override the method in order to preserve backwards binary compatibility.
>
> I've gone through and figured out which methods fall into this category and published a version of this PR that includes those changes to https://github.com/CurtHagenlocher/arrow/tree/flight-sql-client-changes. I can also submit a PR from that branch to your branch if that helps. In addition to the cancellation token changes (which are the largest part of the delta), I've also implemented `IEquatable<>` on `Transaction` and `IAsyncDisposable` on `PreparedStatement`.
>
> One thing I didn't do was to resolve the question of the unused `_dataSetSchema` and `_parameterSchema` fields on `PreparedStatement`. Work is being done to populate these, but they are never used. I started removing them but then realized that I don't really understand the difference between these and the `GetSchemaAsync` call on the same class. Can you take another look and decide what you think should be done about this?
Hi Curt,
Thanks a lot for pushing those updates—adding CancellationToken parameters and the extra interfaces makes perfect sense.
Could you please go ahead and open a PR from flight-sql-client-changes to my branch? That will let me review and integrate your changes right away.
I’ll take another pass over _dataSetSchema and _parameterSchema while I’m at it. My initial hunch is that _dataSetSchema may be redundant, but _parameterSchema is probably still necessary for optional parameter binding, so I’ll verify that and adjust accordingly.
Once I’ve incorporated any tweaks you need, I’d appreciate it if you could merge my original PR as soon as everything looks good on your end.
Thanks again for the thorough improvements!
--
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
Re: [PR] Feat/flight sql client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1849737430
##########
cpp/CMakeLists.txt:
##########
Review Comment:
reverted this, and no I haven't meant to do it.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2498316865
> I approved the workflows.
>
> I don't work on C# sharp but wanted to thank you @HackPoint for such a big first-time contribution!
Can you please review the code and allow me to close the PR please?
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2489188280
:warning: GitHub issue #44800 **has been automatically assigned in GitHub** to PR creator.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1969424656
##########
csharp/test/Apache.Arrow.Flight.Sql.Tests/FlightSqlClientTests.cs:
##########
@@ -0,0 +1,855 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Client;
+using Apache.Arrow.Flight.Sql.Client;
+using Apache.Arrow.Flight.TestWeb;
+using Apache.Arrow.Types;
+using Arrow.Flight.Protocol.Sql;
+using Google.Protobuf;
+using Grpc.Core.Utils;
+using Xunit;
+
+namespace Apache.Arrow.Flight.Sql.Tests;
+
+public class FlightSqlClientTests : IDisposable
+{
+ readonly TestFlightSqlWebFactory _testWebFactory;
+ readonly FlightStore _flightStore;
+ private readonly FlightSqlClient _flightSqlClient;
+ private readonly FlightSqlTestUtils _testUtils;
+
+ public FlightSqlClientTests()
+ {
+ _flightStore = new FlightStore();
+ _testWebFactory = new TestFlightSqlWebFactory(_flightStore);
+ FlightClient flightClient = new(_testWebFactory.GetChannel());
+ _flightSqlClient = new FlightSqlClient(flightClient);
+
+ _testUtils = new FlightSqlTestUtils(_testWebFactory, _flightStore);
+ }
+
+ #region Transactions
+
+ [Fact]
+ public async Task CommitTransactionAsync()
+ {
+ // Arrange
+ string transactionId = "sample-transaction-id";
+ var transaction = new Transaction(transactionId);
+
+ // Act
+ var streamCall = _flightSqlClient.CommitAsync(transaction);
+ var result = await streamCall.ResponseStream.ToListAsync();
+
+ // Assert
+ Assert.NotNull(result);
+ Assert.Equal(transaction.TransactionId, result.FirstOrDefault()?.Body);
+ }
+
+ [Fact]
+ public async Task BeginTransactionAsync()
+ {
+ // Arrange
+ string expectedTransactionId = "sample-transaction-id";
+
+ // Act
+ var transaction = await _flightSqlClient.BeginTransactionAsync();
+
+ // Assert
+ Assert.NotNull(transaction);
+ Assert.Equal(ByteString.CopyFromUtf8(expectedTransactionId), transaction.TransactionId);
+ }
+
+ [Fact]
+ public async Task RollbackTransactionAsync()
+ {
+ // Arrange
+ string transactionId = "sample-transaction-id";
+ var transaction = new Transaction(transactionId);
+
+ // Act
+ var streamCall = _flightSqlClient.RollbackAsync(transaction);
+ var result = await streamCall.ResponseStream.ToListAsync();
+
+ // Assert
+ Assert.NotNull(transaction);
+ Assert.Equal(result.FirstOrDefault()?.Body, transaction.TransactionId);
+ }
+
+ #endregion
+
+ #region PreparedStatement
+
+ [Fact]
+ public async Task PreparedAsync()
+ {
+ // Arrange
+ string query = "INSERT INTO users (id, name) VALUES (1, 'John Doe')";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+
+ // Create a sample schema for the dataset and parameters
+ var schema = new Schema.Builder()
+ .Field(f => f.Name("id").DataType(Int32Type.Default))
+ .Field(f => f.Name("name").DataType(StringType.Default))
+ .Build();
+
+ var recordBatch = new RecordBatch(schema, new Array[]
+ {
+ new Int32Array.Builder().Append(1).Build(),
+ new StringArray.Builder().Append("John Doe").Build()
+ }, 1);
+
+ var flightHolder = new FlightHolder(flightDescriptor, schema, _testWebFactory.GetAddress());
+ flightHolder.AddBatch(new RecordBatchWithMetadata(recordBatch));
+ _flightStore.Flights.Add(flightDescriptor, flightHolder);
+
+ var datasetSchemaBytes = SchemaExtensions.SerializeSchema(schema);
+ var parameterSchemaBytes = SchemaExtensions.SerializeSchema(schema);
+
+ var preparedStatementResponse = new ActionCreatePreparedStatementResult
+ {
+ PreparedStatementHandle = ByteString.CopyFromUtf8("prepared-handle"),
+ DatasetSchema = ByteString.CopyFrom(datasetSchemaBytes),
+ ParameterSchema = ByteString.CopyFrom(parameterSchemaBytes)
+ };
+
+ // Act
+ var preparedStatement = await _flightSqlClient.PrepareAsync(query, transaction);
+ var deserializedDatasetSchema = SchemaExtensions.DeserializeSchema(preparedStatementResponse.DatasetSchema.ToByteArray());
+ var deserializedParameterSchema = SchemaExtensions.DeserializeSchema(preparedStatementResponse.ParameterSchema.ToByteArray());
+
+ // Assert
+ Assert.NotNull(preparedStatement);
+ Assert.NotNull(deserializedDatasetSchema);
+ Assert.NotNull(deserializedParameterSchema);
+ CompareSchemas(schema, deserializedDatasetSchema);
+ CompareSchemas(schema, deserializedParameterSchema);
+ }
+
+ #endregion
+
+ [Fact]
+ public async Task ExecuteUpdateAsync()
+ {
+ // Arrange
+ string query = "UPDATE test_table SET column1 = 'value' WHERE column2 = 'condition'";
+ var transaction = new Transaction("sample-transaction-id");
+ var flightDescriptor = FlightDescriptor.CreateCommandDescriptor("test");
+
+ var schema = new Schema.Builder()
+ .Field(f => f.Name("id").DataType(Int32Type.Default))
+ .Field(f => f.Name("name").DataType(StringType.Default))
+ .Build();
+
+ var recordBatch = new RecordBatch(schema, new Array[]
+ {
+ new Int32Array.Builder().Append(1).Build(),
+ new StringArray.Builder().Append("John Doe").Build()
+ }, 1);
+
+
Review Comment:
done
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1969422951
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightExtensions.cs:
##########
@@ -0,0 +1,81 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
+
+namespace Apache.Arrow.Flight.Sql;
+
+internal static class FlightExtensions
+{
+ public static byte[] PackAndSerialize(this IMessage command) => Any.Pack(command).ToByteArray();
+ public static T ParseAndUnpack<T>(this ByteString source) where T : IMessage<T>, new() => Any.Parser.ParseFrom(source).Unpack<T>();
+
+ public static IEnumerable<long> ExtractRowCount(this RecordBatch batch)
Review Comment:
Added:
Executes an update SQL command and returns the number of affected rows.
But after looking at:
```c++
ARROW_ASSIGN_OR_RAISE(auto updated_rows,
sql_client->ExecuteUpdate({}, "UPDATE STATEMENT"));
```
I created the same method and implemented it as is.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1968061250
##########
csharp/src/Apache.Arrow.Flight.Sql/SchemaExtensions.cs:
##########
@@ -0,0 +1,52 @@
+// 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.
+
+using System;
+using System.IO;
+using Apache.Arrow.Ipc;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public static class SchemaExtensions
+{
+ /// <summary>
+ /// Deserializes a schema from a byte array.
+ /// </summary>
+ /// <param name="serializedSchema">The byte array representing the serialized schema.</param>
+ /// <returns>The deserialized Schema object.</returns>
+ public static Schema DeserializeSchema(byte[] serializedSchema)
+ {
+ if (serializedSchema == null || serializedSchema.Length == 0)
+ {
+ throw new ArgumentException("Invalid serialized schema");
+ }
+
+ using var stream = new MemoryStream(serializedSchema);
+ var reader = new ArrowStreamReader(stream);
Review Comment:
Done
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2835753558
> @CurtHagenlocher We should do [apache/arrow-dotnet#1](https://github.com/apache/arrow-dotnet/issues/1) after we merge this, right?
Hi, have you got a chance to merge this code? I have another huge PR prepared? I'm waiting for 6 or more months for it.
@jeremyosterhoudt can you assist please?
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1978568789
##########
csharp/src/Apache.Arrow.Flight/FlightInfoCancelRequest.cs:
##########
@@ -0,0 +1,38 @@
+using System;
Review Comment:
The Apache header is still missing from these two files in the latest iteration. (I haven't yet had a chance to review any or the other changes in the latest iteration, but the missing copyright notice does produce a build failure.)
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1967861118
##########
csharp/src/Apache.Arrow.Flight.Sql/Transaction.cs:
##########
@@ -0,0 +1,48 @@
+// 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.
+
+using Google.Protobuf;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class Transaction
+{
+ private static readonly ByteString TransactionIdDefaultValue = ByteString.Empty;
+ private ByteString? _transactionId;
+
+ public ByteString TransactionId
+ {
+ get => _transactionId ?? TransactionIdDefaultValue;
+ set => _transactionId = ProtoPreconditions.CheckNotNull(value, nameof(value));
+ }
+
+ public static readonly Transaction NoTransaction = new(TransactionIdDefaultValue);
+
+ public Transaction(ByteString transactionId)
+ {
+ TransactionId = transactionId;
+ }
+
+ public Transaction(string transactionId)
+ {
+ _transactionId = ByteString.CopyFromUtf8(transactionId);
+ }
+
+ public bool IsValid() => TransactionId.Length > 0;
+ public void ResetTransaction()
Review Comment:
removed it from the implementation
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-3096746066
> One thing I didn't do was to resolve the question of the unused `_dataSetSchema` and `_parameterSchema` fields on `PreparedStatement`. Work is being done to populate these, but they are never used. I started removing them but then realized that I don't really understand the difference between these and the `GetSchemaAsync` call on the same class. Can you take another look and decide what you think should be done about this?
After reviewing the C++ implementation (client.h / PreparedStatement), I confirmed that:
```cpp
std::shared_ptr<Schema> parameter_schema() const { return parameter_schema_; }
std::shared_ptr<Schema> dataset_schema() const { return dataset_schema_; }
```
source: `[client.h / PreparedStatement]`
These schemas are accessed directly by consumers after creating the prepared statement, allowing them to inspect the expected input/output without making additional calls.
In contrast, `GetSchemaAsync()` in C# makes a separate server call to retrieve the schema again — which may be redundant if we already have it.
So I made the design decision to expose both schemas as public, readonly properties:
```csharp
public Schema DatasetSchema { get; }
public Schema ParameterSchema { get; }
```
This aligns with the C++ API and gives clients immediate access to the known schemas without having to re-fetch them. If schema refresh is ever required, they can still call `GetSchemaAsync()` manually.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher merged PR #44783:
URL: https://github.com/apache/arrow/pull/44783
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "assignUser (via GitHub)" <gi...@apache.org>.
assignUser commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2489218881
I approved the workflows.
I don't work on C# sharp but wanted to thank you @HackPoint for such a big first-time contribution!
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-3074093702
@HackPoint See https://github.com/HackPoint/arrow/pull/2
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "kou (via GitHub)" <gi...@apache.org>.
kou commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-3054752366
I'll move `csharp/` in apache/arrow to apache/arrow-dotnet after apache/arrow 21.0.0 release if nobody objects it: https://lists.apache.org/thread/j3v93j2s6wpr7hh0xxtofd5167nrf4g7
--
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
Re: [PR] Feat/flight sql client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1847541222
##########
csharp/Apache.Arrow.sln:
##########
@@ -17,8 +17,6 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.Flight", "src\
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.Flight.AspNetCore", "src\Apache.Arrow.Flight.AspNetCore\Apache.Arrow.Flight.AspNetCore.csproj", "{E4F74938-E8FF-4AC1-A495-FEE95FC1EFDF}"
EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.IntegrationTest", "test\Apache.Arrow.IntegrationTest\Apache.Arrow.IntegrationTest.csproj", "{E8264B7F-B680-4A55-939B-85DB628164BB}"
-EndProject
Review Comment:
This should be added back (I suspect you forked before it was added to begin with.)
--
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
Re: [PR] Feat/flight sql client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1848026339
##########
csharp/src/Apache.Arrow.Flight.Sql/Apache.Arrow.Flight.Sql.csproj:
##########
@@ -5,7 +5,8 @@
</PropertyGroup>
<ItemGroup>
- <PackageReference Include="Grpc.Tools" Version="2.67.0" PrivateAssets="All" />
+ <PackageReference Include="Grpc.Tools" Version="2.66.0" PrivateAssets="All" />
Review Comment:
Understood. Starting fixups.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1968066965
##########
csharp/src/Apache.Arrow.Flight.Sql/RecordBatchExtensions.cs:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Apache.Arrow.Ipc;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public static class RecordBatchExtensions
+{
+ /// <summary>
+ /// Converts a RecordBatch into an asynchronous stream of FlightData.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An asynchronous stream of FlightData objects.</returns>
+ public static async IAsyncEnumerable<FlightData> ToFlightDataStreamAsync(this RecordBatch recordBatch,
+ FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null)
+ {
+ throw new ArgumentNullException(nameof(recordBatch));
+ }
+
+ // Use a memory stream to write the Arrow RecordBatch into FlightData format
+ using var memoryStream = new MemoryStream();
+ var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+
+ // Write the RecordBatch to the stream
+ await writer.WriteRecordBatchAsync(recordBatch).ConfigureAwait(false);
+ await writer.WriteEndAsync().ConfigureAwait(false);
+
+ // Reset the memory stream position
+ memoryStream.Position = 0;
+
+ // Read back the data to create FlightData
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()),
+ ByteString.CopyFrom(memoryStream.ToArray()));
+ yield return flightData;
+ }
+
+ /// <summary>
+ /// Converts a RecordBatch into an IAsyncStreamReader<FlightData>.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An IAsyncStreamReader of FlightData.</returns>
+ public static IAsyncStreamReader<FlightData> ToFlightDataStream(this RecordBatch recordBatch, FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null) throw new ArgumentNullException(nameof(recordBatch));
+ if (flightDescriptor == null) throw new ArgumentNullException(nameof(flightDescriptor));
+
+ var channel = Channel.CreateUnbounded<FlightData>();
+
+ try
+ {
+ if (recordBatch.Schema == null || !recordBatch.Schema.FieldsList.Any())
+ {
+ throw new InvalidOperationException("The record batch has an invalid or empty schema.");
+ }
+
+ using var memoryStream = new MemoryStream();
+ using var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+ writer.WriteRecordBatch(recordBatch);
+ writer.WriteEnd();
+ memoryStream.Position = 0;
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()), ByteString.Empty, ByteString.Empty);
+ if (flightData.DataBody.IsEmpty)
+ {
+ throw new InvalidOperationException(
+ "The generated FlightData is empty. Check the RecordBatch content.");
+ }
+
+ channel.Writer.TryWrite(flightData);
Review Comment:
Implementation is deleted
##########
csharp/src/Apache.Arrow.Flight.Sql/RecordBatchExtensions.cs:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Apache.Arrow.Ipc;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public static class RecordBatchExtensions
+{
+ /// <summary>
+ /// Converts a RecordBatch into an asynchronous stream of FlightData.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An asynchronous stream of FlightData objects.</returns>
+ public static async IAsyncEnumerable<FlightData> ToFlightDataStreamAsync(this RecordBatch recordBatch,
+ FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null)
+ {
+ throw new ArgumentNullException(nameof(recordBatch));
+ }
+
+ // Use a memory stream to write the Arrow RecordBatch into FlightData format
+ using var memoryStream = new MemoryStream();
+ var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+
+ // Write the RecordBatch to the stream
+ await writer.WriteRecordBatchAsync(recordBatch).ConfigureAwait(false);
+ await writer.WriteEndAsync().ConfigureAwait(false);
+
+ // Reset the memory stream position
+ memoryStream.Position = 0;
+
+ // Read back the data to create FlightData
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()),
+ ByteString.CopyFrom(memoryStream.ToArray()));
+ yield return flightData;
+ }
+
+ /// <summary>
+ /// Converts a RecordBatch into an IAsyncStreamReader<FlightData>.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An IAsyncStreamReader of FlightData.</returns>
+ public static IAsyncStreamReader<FlightData> ToFlightDataStream(this RecordBatch recordBatch, FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null) throw new ArgumentNullException(nameof(recordBatch));
+ if (flightDescriptor == null) throw new ArgumentNullException(nameof(flightDescriptor));
+
+ var channel = Channel.CreateUnbounded<FlightData>();
+
+ try
+ {
+ if (recordBatch.Schema == null || !recordBatch.Schema.FieldsList.Any())
+ {
+ throw new InvalidOperationException("The record batch has an invalid or empty schema.");
+ }
+
+ using var memoryStream = new MemoryStream();
+ using var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+ writer.WriteRecordBatch(recordBatch);
+ writer.WriteEnd();
+ memoryStream.Position = 0;
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()), ByteString.Empty, ByteString.Empty);
+ if (flightData.DataBody.IsEmpty)
Review Comment:
Implementation is deleted
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1968049070
##########
csharp/src/Apache.Arrow.Flight.Sql/Transaction.cs:
##########
@@ -0,0 +1,48 @@
+// 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.
+
+using Google.Protobuf;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class Transaction
Review Comment:
After re-evaluation, I've changed the Transaction to be value type for the moment be,
Seems that there's no too much data ( properties ) are being passed.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r2021649610
##########
csharp/src/Apache.Arrow.Flight.Sql/Transaction.cs:
##########
@@ -0,0 +1,48 @@
+// 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.
+
+namespace Apache.Arrow.Flight.Sql;
+
+using Google.Protobuf; // Ensure you have the Protobuf dependency
+
+public readonly struct Transaction
+{
+ private static readonly ByteString TransactionIdDefaultValue = ByteString.Empty;
+
+ private readonly ByteString _transactionId;
+
+ public ByteString TransactionId => _transactionId ?? TransactionIdDefaultValue;
+
+ public static readonly Transaction NoTransaction = new(TransactionIdDefaultValue);
+
+ public Transaction(ByteString transactionId)
+ {
+ _transactionId = ProtoPreconditions.CheckNotNull(transactionId, nameof(transactionId));
+ }
+
+ public Transaction(string transactionId)
+ {
+ _transactionId = ByteString.CopyFromUtf8(transactionId);
+ }
+
+ public bool IsValid() => _transactionId.Length > 0;
Review Comment:
```suggestion
public bool IsValid => _transactionId.Length > 0;
```
##########
csharp/src/Apache.Arrow.Flight.Sql/PreparedStatement.cs:
##########
@@ -0,0 +1,383 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Sql.Client;
+using Arrow.Flight.Protocol.Sql;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class PreparedStatement : IDisposable
+{
+ private readonly FlightSqlClient _client;
+ private readonly string _handle;
+ private Schema _datasetSchema;
+ private Schema _parameterSchema;
Review Comment:
These are still unused as of the latest iteration. They're set by the constructor and never referenced again.
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightExtensions.cs:
##########
@@ -0,0 +1,41 @@
+// 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.
+
+using System;
+using Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
+
+namespace Apache.Arrow.Flight.Sql;
+
+internal static class FlightExtensions
+{
+ public static byte[] PackAndSerialize(this IMessage command) => Any.Pack(command).ToByteArray();
+
+ public static T ParseAndUnpack<T>(this ByteString source) where T : IMessage<T>, new() =>
+ Any.Parser.ParseFrom(source).Unpack<T>();
+
+ public static int ExtractRowCount(this RecordBatch batch)
Review Comment:
Why isn't this just `RecordBatch.Length`?
##########
csharp/src/Apache.Arrow.Flight.Sql/DoPutResult.cs:
##########
@@ -0,0 +1,54 @@
+// 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.
+
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Client;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class DoPutResult
+{
+ public FlightClientRecordBatchStreamWriter Writer { get; }
+ public IAsyncStreamReader<FlightPutResult> Reader { get; }
+
+ public DoPutResult(FlightClientRecordBatchStreamWriter writer, IAsyncStreamReader<FlightPutResult> reader)
+ {
+ Writer = writer;
+ Reader = reader;
+ }
+
+ /// <summary>
+ /// Reads the metadata asynchronously from the reader.
+ /// </summary>
+ /// <returns>A ByteString containing the metadata read from the reader.</returns>
+ public async Task<Google.Protobuf.ByteString> ReadMetadataAsync()
Review Comment:
So at least for this method, we expect that `Reader.MoveNext()` will be working on data already in memory vs coming in over the network?
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2573692881
> Any updates on this? We are also looking to use .NET with a Flight SQL client and are eager to get this functionality.
I'm sorry about the delay; I was very much wanting to get the review done before today's fork but first work, then vacation, then a lengthy internet outage at home got in the way. I will definitely make time for a review later this week.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1968073191
##########
csharp/src/Apache.Arrow.Flight.Sql/PreparedStatement.cs:
##########
@@ -0,0 +1,383 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Sql.Client;
+using Arrow.Flight.Protocol.Sql;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class PreparedStatement : IDisposable
+{
+ private readonly FlightSqlClient _client;
+ private readonly string _handle;
+ private Schema _datasetSchema;
+ private Schema _parameterSchema;
+ private RecordBatch? _recordsBatch;
+ private bool _isClosed;
+ public bool IsClosed => _isClosed;
+ public string Handle => _handle;
+ public RecordBatch? ParametersBatch => _recordsBatch;
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PreparedStatement"/> class.
+ /// </summary>
+ /// <param name="client">The Flight SQL client used for executing SQL operations.</param>
+ /// <param name="handle">The handle representing the prepared statement.</param>
+ /// <param name="datasetSchema">The schema of the result dataset.</param>
+ /// <param name="parameterSchema">The schema of the parameters for this prepared statement.</param>
+ public PreparedStatement(FlightSqlClient client, string handle, Schema datasetSchema, Schema parameterSchema)
+ {
+ _client = client ?? throw new ArgumentNullException(nameof(client));
+ _handle = handle ?? throw new ArgumentNullException(nameof(handle));
+ _datasetSchema = datasetSchema ?? throw new ArgumentNullException(nameof(datasetSchema));
+ _parameterSchema = parameterSchema ?? throw new ArgumentNullException(nameof(parameterSchema));
+ _isClosed = false;
+ }
+
+ /// <summary>
+ /// Retrieves the schema associated with the prepared statement asynchronously.
+ /// </summary>
+ /// <param name="options">The options used to configure the Flight call.</param>
+ /// <returns>A task representing the asynchronous operation, which returns the schema of the result set.</returns>
+ /// <exception cref="InvalidOperationException">Thrown when the schema is empty or invalid.</exception>
+ public async Task<Schema> GetSchemaAsync(FlightCallOptions? options = default)
+ {
+ EnsureStatementIsNotClosed();
+
+ try
+ {
+ var command = new CommandPreparedStatementQuery
+ {
+ PreparedStatementHandle = ByteString.CopyFrom(_handle, Encoding.UTF8)
+ };
+ var descriptor = FlightDescriptor.CreateCommandDescriptor(command.PackAndSerialize());
+ var schema = await _client.GetSchemaAsync(descriptor, options).ConfigureAwait(false);
+ if (schema == null || !schema.FieldsList.Any())
+ {
+ throw new InvalidOperationException("Schema is empty or invalid.");
+ }
+ return schema;
+ }
+ catch (RpcException ex)
+ {
+ throw new InvalidOperationException("Failed to retrieve the schema for the prepared statement", ex);
+ }
+ }
+
+
Review Comment:
don't the extra line, but removed one above.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1968081819
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightCallOptions.cs:
##########
@@ -0,0 +1,47 @@
+// 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.
+
+using System;
+using System.Buffers;
+using System.Threading;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class FlightCallOptions
+{
+ public FlightCallOptions()
+ {
+ Timeout = TimeSpan.FromSeconds(-1);
+ }
+ // Implement any necessary options for RPC calls
+ public Metadata Headers { get; set; } = new();
+
+ /// <summary>
+ /// Gets or sets a token to enable interactive user cancellation of long-running requests.
+ /// </summary>
+ public CancellationToken StopToken { get; set; }
Review Comment:
removed the property
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightCallOptions.cs:
##########
@@ -0,0 +1,47 @@
+// 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.
+
+using System;
+using System.Buffers;
+using System.Threading;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class FlightCallOptions
+{
+ public FlightCallOptions()
+ {
+ Timeout = TimeSpan.FromSeconds(-1);
+ }
+ // Implement any necessary options for RPC calls
+ public Metadata Headers { get; set; } = new();
+
+ /// <summary>
+ /// Gets or sets a token to enable interactive user cancellation of long-running requests.
+ /// </summary>
+ public CancellationToken StopToken { get; set; }
+
+ /// <summary>
+ /// Gets or sets the optional timeout for this call.
+ /// Negative durations mean an implementation-defined default behavior will be used instead.
+ /// </summary>
+ public TimeSpan Timeout { get; set; }
+
+ /// <summary>
+ /// Gets or sets an optional memory manager to control where to allocate incoming data.
+ /// </summary>
+ public MemoryManager<ArrowBuffer>? MemoryManager { get; set; }
Review Comment:
removed the property
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1968080313
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightCallOptions.cs:
##########
@@ -0,0 +1,47 @@
+// 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.
+
+using System;
+using System.Buffers;
+using System.Threading;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class FlightCallOptions
+{
+ public FlightCallOptions()
+ {
+ Timeout = TimeSpan.FromSeconds(-1);
+ }
Review Comment:
done
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2498446618
> Can you please review the code and allow me to close the PR please?
I will review this tonight.
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-3069382218
@HackPoint, I do think it's important to add cancellation tokens to methods which themselves call other methods that already take a cancellation token -- particularly when the method in question is a public API. Even if no one ever ends up using it, the cost is quite small. By contrast, if one needs to be added later then we end up needing to override the method in order to preserve backwards binary compatibility.
I've gone through and figured out which methods fall into this category and published a version of this PR that includes those changes to https://github.com/CurtHagenlocher/arrow/tree/flight-sql-client-changes. I can also submit a PR from that branch to your branch if that helps. In addition to the cancellation token changes (which are the largest part of the delta), I've also implemented `IEquatable<>` on `Transaction` and `IAsyncDisposable` on `PreparedStatement`.
One thing I didn't do was to resolve the question of the unused `_dataSetSchema` and `_parameterSchema` fields on `PreparedStatement`. Work is being done to populate these, but they are never used. I started removing them but then realized that I don't really understand the difference between these and the `GetSchemaAsync` call on the same class. Can you take another look and decide what you think should be done about this?
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2499695614
I’ve haven’t made any changes in c++, but I’ll re-fork and cherry-pick my
code into a new fork and push all the changes.
On Tue, 26 Nov 2024 at 7:28 Sutou Kouhei ***@***.***> wrote:
> Could you revert all cpp/ changes?
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/arrow/pull/44783#issuecomment-2499692202>, or
> unsubscribe
> <https://github.com/notifications/unsubscribe-auth/ABTSBM7DBABI7X26GBOCS7D2CQBGVAVCNFSM6AAAAABSAGSJF6VHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDIOJZGY4TEMRQGI>
> .
> You are receiving this because you were mentioned.Message ID:
> ***@***.***>
>
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1968064767
##########
csharp/src/Apache.Arrow.Flight.Sql/RecordBatchExtensions.cs:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Apache.Arrow.Ipc;
+using Google.Protobuf;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public static class RecordBatchExtensions
+{
+ /// <summary>
+ /// Converts a RecordBatch into an asynchronous stream of FlightData.
+ /// </summary>
+ /// <param name="recordBatch">The RecordBatch to convert.</param>
+ /// <param name="flightDescriptor">The FlightDescriptor describing the Flight data.</param>
+ /// <returns>An asynchronous stream of FlightData objects.</returns>
+ public static async IAsyncEnumerable<FlightData> ToFlightDataStreamAsync(this RecordBatch recordBatch,
+ FlightDescriptor flightDescriptor)
+ {
+ if (recordBatch == null)
+ {
+ throw new ArgumentNullException(nameof(recordBatch));
+ }
+
+ // Use a memory stream to write the Arrow RecordBatch into FlightData format
+ using var memoryStream = new MemoryStream();
+ var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema);
+
+ // Write the RecordBatch to the stream
+ await writer.WriteRecordBatchAsync(recordBatch).ConfigureAwait(false);
+ await writer.WriteEndAsync().ConfigureAwait(false);
+
+ // Reset the memory stream position
+ memoryStream.Position = 0;
+
+ // Read back the data to create FlightData
+ var flightData = new FlightData(flightDescriptor, ByteString.CopyFrom(memoryStream.ToArray()),
+ ByteString.CopyFrom(memoryStream.ToArray()));
Review Comment:
Implementation is deleted
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1969398707
##########
csharp/src/Apache.Arrow.Flight.Sql/FlightExtensions.cs:
##########
@@ -0,0 +1,81 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
+
+namespace Apache.Arrow.Flight.Sql;
+
+internal static class FlightExtensions
+{
+ public static byte[] PackAndSerialize(this IMessage command) => Any.Pack(command).ToByteArray();
+ public static T ParseAndUnpack<T>(this ByteString source) where T : IMessage<T>, new() => Any.Parser.ParseFrom(source).Unpack<T>();
+
+ public static IEnumerable<long> ExtractRowCount(this RecordBatch batch)
+ {
+ foreach (var array in batch.Arrays)
+ {
+ var values = ExtractValues(array);
+ foreach (var value in values)
+ {
+ yield return value switch
+ {
+ long l => l,
+ int i => i != 0 ? i : 0,
Review Comment:
Improved the implementation and removed at all this redundant implementation
--
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
Re: [PR] GH-44800: [C#] Implement Flight SQL Client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2916887941
@CurtHagenlocher @adamreeve @kou hey guys, take a look at this is it ready, is it good to go or should I do some changes?
--
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
Re: [PR] Feat/flight sql client [arrow]
Posted by "HackPoint (via GitHub)" <gi...@apache.org>.
HackPoint closed pull request #44783: Feat/flight sql client
URL: https://github.com/apache/arrow/pull/44783
--
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
Re: [PR] Feat/flight sql client [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2483710575
<!--
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.
-->
Thanks for opening a pull request!
If this is not a [minor PR](https://github.com/apache/arrow/blob/main/CONTRIBUTING.md#Minor-Fixes). Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose
Opening GitHub issues ahead of time contributes to the [Openness](http://theapacheway.com/open/#:~:text=Openness%20allows%20new%20users%20the,must%20happen%20in%20the%20open.) of the Apache Arrow project.
Then could you also rename the pull request title in the following format?
GH-${GITHUB_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}
or
MINOR: [${COMPONENT}] ${SUMMARY}
In the case of PARQUET issues on JIRA the title also supports:
PARQUET-${JIRA_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}
See also:
* [Other pull requests](https://github.com/apache/arrow/pulls/)
* [Contribution Guidelines - Contributing Overview](https://arrow.apache.org/docs/developers/overview.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
Re: [PR] Feat/flight sql client [arrow]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on PR #44783:
URL: https://github.com/apache/arrow/pull/44783#issuecomment-2483732818
Thanks for a very large change! Could you please add the official Apache header to all the source files and open an issue to link this to?
--
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
Re: [PR] Feat/flight sql client [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #44783:
URL: https://github.com/apache/arrow/pull/44783#discussion_r1847449010
##########
cpp/CMakeLists.txt:
##########
Review Comment:
Did you mean to modify this?
##########
cpp/src/arrow/flight/sql/client.h:
##########
Review Comment:
This too.
--
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