CurtHagenlocher commented on code in PR #4885:
URL: https://github.com/apache/arrow-adbc/pull/4885#discussion_r4260957176
##########
csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs:
##########
@@ -362,25 +370,47 @@ public unsafe override AdbcDatabase
Open(IReadOnlyDictionary<string, object> par
{
if (parameters == null) throw new
ArgumentNullException(nameof(parameters));
+ if (_version < AdbcVersion.Version_1_1_0)
+ {
+ return Open(AdbcOptionValue.ToOptionStrings(parameters));
+ }
Review Comment:
Fixed in 85533739f. For a 1.0 driver, the importer's
`Open(IReadOnlyDictionary<string, object>)` now calls the base implementation.
A binary value now fails with `NotImplemented` and the database is released,
the same as for `ConnectWithTypedOptions`.
`Version1_0DriverRejectsBinaryOpenParameters` covers this.
##########
csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs:
##########
@@ -0,0 +1,650 @@
+/*
+ * 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.Adbc.C;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Types;
+using Xunit;
+
+namespace Apache.Arrow.Adbc.Tests
+{
+ /// <summary>
+ /// Round-trips the ADBC 1.1.0 API surface through <see
cref="CAdbcDriverExporter"/>
+ /// and <see cref="CAdbcDriverImporter"/>: typed options, statistics,
error details
+ /// and the standard connection options.
+ /// </summary>
+ public class ExportedDriverRoundTrip11Tests
+ {
+ [Fact]
+ public void TypedOpenParametersAreMarshaledToProducer()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
object>
+ {
+ { "string", "value" },
+ { "int", 42L },
+ { "double", 2.5 },
+ { "bytes", new byte[] { 1, 2, 3 } },
+ { "bool", true },
+ });
+
+ Dictionary<string, object> options = fixture.LastDatabase!.Options;
+ Assert.Equal("value", options["string"]);
+ Assert.Equal(42L, options["int"]);
+ Assert.Equal(2.5, options["double"]);
+ Assert.Equal(new byte[] { 1, 2, 3 }, options["bytes"]);
+ Assert.Equal("true", options["bool"]);
+ }
+
+ [Fact]
+ public void DatabaseTypedOptionsRoundTrip()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+
+ db.SetOption("s", "text");
+ db.SetOptionInt64("i", long.MinValue);
+ db.SetOptionDouble("d", -0.125);
+ db.SetOptionBytes("b", new byte[] { 0, 255 });
+
+ Assert.Equal("text", db.GetOption("s"));
+ Assert.Equal(long.MinValue, db.GetOptionInt64("i"));
+ Assert.Equal(-0.125, db.GetOptionDouble("d"));
+ Assert.Equal(new byte[] { 0, 255 }, db.GetOptionBytes("b"));
+ }
+
+ [Fact]
+ public void DatabaseGetOptionReportsOptionsTheProducerDoesNotTrack()
+ {
+ // A 1.0-style producer that never overrode GetOption still answers
+ // DatabaseGetOption for options set through the C API.
+ var fixture = new Driver11 { DatabaseTracksOptions = false };
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string> { { "uri", "memory://" } });
+
+ Assert.Equal("memory://", db.GetOption("uri"));
+ AdbcException ex = Assert.ThrowsAny<AdbcException>(() =>
db.GetOption("missing"));
+ Assert.Equal(AdbcStatusCode.NotFound, ex.Status);
+ }
+
+ [Fact]
+ public void LongStringOptionUsesBufferRetry()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+
+ string longValue = new string('x', 1000) + "é";
+ stmt.SetOption("long", longValue);
+ Assert.Equal(longValue, stmt.GetOption("long"));
+
+ byte[] longBytes = new byte[1000];
+ new Random(1).NextBytes(longBytes);
+ stmt.SetOptionBytes("long_bytes", longBytes);
+ Assert.Equal(longBytes, stmt.GetOptionBytes("long_bytes"));
+ }
+
+ [Fact]
+ public void ConnectionTypedOptionsRoundTrip()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(new Dictionary<string,
string> { { "at_connect", "yes" } });
+
+ conn.SetOptionInt64("timeout", 30);
+ conn.SetOptionDouble("ratio", 0.75);
+
+ Assert.Equal(30L, fixture.LastConnection!.Options["timeout"]);
+ Assert.Equal(0.75, fixture.LastConnection!.Options["ratio"]);
+ Assert.Equal(30L, conn.GetOptionInt64("timeout"));
+ Assert.Equal(0.75, conn.GetOptionDouble("ratio"));
+ Assert.Equal("yes", conn.GetOption("at_connect"));
+ }
+
+ [Fact]
+ public void StandardConnectionOptionsComeFromProperties()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+
+ // The producer doesn't override GetOption for these keys; the base
+ // AdbcConnection implementation answers from its properties.
+ Assert.Equal(AdbcOptions.Enabled,
conn.GetOption(AdbcOptions.Connection.Autocommit));
+ Assert.True(conn.AutoCommit);
+ Assert.Equal(IsolationLevel.Default, conn.IsolationLevel);
+
+ conn.CurrentCatalog = "main";
+ Assert.Equal("main", fixture.LastConnection!.CurrentCatalog);
+ Assert.Equal("main", conn.CurrentCatalog);
+ Assert.Equal("main",
conn.GetOption(AdbcOptions.Connection.CurrentCatalog));
+
+ // The producer doesn't support a current schema at all
+ Assert.Null(conn.CurrentDbSchema);
+ }
+
+ [Fact]
+ public void TypedConnectOptionsReachTheProducerBeforeInit()
+ {
+ var fixture = new Driver11 { DatabaseHandlesTypedConnect = true };
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.ConnectWithTypedOptions(new
Dictionary<string, object>
+ {
+ { "s", "text" },
+ { "n", 5L },
+ { "d", 0.5 },
+ { "cert", new byte[] { 1, 2, 3 } },
+ });
+
+ IReadOnlyDictionary<string, object> received =
fixture.LastTypedConnectOptions!;
+ Assert.Equal("text", received["s"]);
+ Assert.Equal(5L, received["n"]);
+ Assert.Equal(0.5, received["d"]);
+ Assert.Equal(new byte[] { 1, 2, 3 }, received["cert"]);
+ }
+
+ [Fact]
+ public void TypedConnectOptionsFallBackToConnectForExistingProducers()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.ConnectWithTypedOptions(new
Dictionary<string, object>
+ {
+ { "n", 5L },
+ { "cert", new byte[] { 1, 2, 3 } },
+ });
+
+ Assert.Null(fixture.LastTypedConnectOptions);
+ Assert.Equal("5", fixture.LastConnection!.Options["n"]);
+ Assert.Equal(new byte[] { 1, 2, 3 },
fixture.LastConnection!.Options["cert"]);
+ }
+
+ [Fact]
+ public void AutoCommitSetBeforeInitIsReported()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(new Dictionary<string,
string>
+ {
+ { AdbcOptions.Connection.Autocommit, AdbcOptions.Disabled },
+ });
+
+ Assert.False(fixture.LastConnection!.AutoCommitAtConnect);
+ Assert.Equal(AdbcOptions.Disabled,
conn.GetOption(AdbcOptions.Connection.Autocommit));
+ Assert.False(conn.AutoCommit);
+ }
+
+ [Fact]
+ public void ExecuteUpdateRoundTripsWithoutAResultStream()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+
+ stmt.SqlQuery = "UPDATE";
+ Assert.Equal(42, stmt.ExecuteUpdate().AffectedRows);
+ }
+
+ [Fact]
+ public void StatisticsAreMarshaled()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+
+ using (IArrowArrayStream names = conn.GetStatisticsNames())
+ {
+ Assert.Equal("statistic_name",
names.Schema.FieldsList[0].Name);
+ }
+
+ using (IArrowArrayStream stats = conn.GetStatistics("cat", null,
"tbl", approximate: true))
+ {
+ Assert.Equal("statistic_name",
stats.Schema.FieldsList[0].Name);
+ }
+ Assert.Equal(("cat", (string?)null, "tbl", true),
fixture.LastConnection!.LastStatisticsRequest);
+ }
+
+ [Fact]
+ public void ErrorDetailsArePropagated()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "FAIL";
+
+ AdbcException ex = Assert.ThrowsAny<AdbcException>(() =>
stmt.ExecuteQuery());
+ Assert.Equal(AdbcStatusCode.InvalidData, ex.Status);
+ Assert.Equal("42000", ex.SqlState);
+ Assert.Contains("query failed", ex.Message);
+ Assert.Equal(2, ex.Details.Count);
+ Assert.Equal("query_id", ex.Details[0].Key);
+ Assert.Equal("q-123",
Encoding.UTF8.GetString(ex.Details[0].Value));
+ Assert.Equal("empty", ex.Details[1].Key);
+ Assert.Empty(ex.Details[1].Value);
+ }
+
+ [Fact]
+ public void ExecutePartitionedRoundTrips()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+
+ stmt.SqlQuery = "PARTITIONS";
+ PartitionedResult result = stmt.ExecutePartitioned();
+
+ Assert.Equal(5, result.AffectedRows);
+ Assert.Equal("x", Assert.Single(result.Schema.FieldsList).Name);
+ Assert.Equal(2, result.PartitionDescriptors.Count);
+ Assert.Equal(new byte[] { 1, 2 },
result.PartitionDescriptors[0].Descriptor.ToArray());
+ Assert.Equal(new byte[] { 3 },
result.PartitionDescriptors[1].Descriptor.ToArray());
+ }
+
+ [Fact]
+ public void ExecutePartitionedFailureDoesNotReleaseGarbage()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+
+ // The producer fails without writing the partitions struct
+ stmt.SqlQuery = "OTHER";
+ for (int i = 0; i < 20; i++)
+ {
+ Assert.Equal(AdbcStatusCode.NotImplemented,
Assert.ThrowsAny<AdbcException>(() => stmt.ExecutePartitioned()).Status);
+ }
+ }
+
+ [Fact]
+ public void ErrorWithoutDetailsReportsVendorCode()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "FAIL_PLAIN";
+
+ AdbcException ex = Assert.ThrowsAny<AdbcException>(() =>
stmt.ExecuteQuery());
+ Assert.Equal("HY000", ex.SqlState);
+ Assert.Equal(1234, ex.NativeError);
+ Assert.Empty(ex.Details);
+ }
+
+ [Fact]
+ public void UnknownOptionIsNotFound()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+
+ Assert.Equal(AdbcStatusCode.NotFound,
Assert.ThrowsAny<AdbcException>(() => conn.GetOption("nope")).Status);
+ Assert.Equal(AdbcStatusCode.NotFound,
Assert.ThrowsAny<AdbcException>(() => stmt.GetOptionInt64("nope")).Status);
+ }
+
+ [Fact]
+ public void OperationCanceledExceptionIsReportedAsCancelled()
+ {
+ using AdbcDriver imported = Load(new Driver11());
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "FAIL_CANCELED";
+
+ AdbcException ex = Assert.ThrowsAny<AdbcException>(() =>
stmt.ExecuteQuery());
+ Assert.Equal(AdbcStatusCode.Cancelled, ex.Status);
+ }
+
+ [Fact]
+ public async Task CancelReachesImportedStatement()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "BLOCK";
+
+ Task<QueryResult> pending = Task.Run(() => stmt.ExecuteQuery());
+
Assert.True(fixture.LastStatement!.Started.Wait(TimeSpan.FromSeconds(10)));
+ stmt.Cancel();
+
+ AdbcException ex = await Assert.ThrowsAnyAsync<AdbcException>(()
=> pending);
+ Assert.Equal(AdbcStatusCode.Cancelled, ex.Status);
+ }
+
+ [Fact]
+ public void Version1_0DriverFallsBackForTypedSetters()
+ {
+ var fixture = new Driver11();
+ AdbcDriverInit adapter = (int version, ref CAdbcDriver
nativeDriver, ref CAdbcError error) =>
+ {
+ if (version != AdbcVersion.Version_1_0_0) { return
AdbcStatusCode.NotImplemented; }
+ return CallExporter(version, ref nativeDriver, ref error,
fixture);
+ };
+
+ using AdbcDriver imported = CAdbcDriverImporter.Load(adapter);
+ Assert.Equal(AdbcVersion.Version_1_0_0, imported.DriverVersion);
+
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
object> { { "n", 7L } });
+ Assert.Equal("7", fixture.LastDatabase!.Options["n"]);
+
+ using AdbcConnection conn = db.Connect(null);
+ conn.SetOptionInt64("timeout", 5);
+ Assert.Equal("5", fixture.LastConnection!.Options["timeout"]);
+
+ Assert.Equal(AdbcStatusCode.NotImplemented,
Assert.ThrowsAny<AdbcException>(() => conn.GetOption("timeout")).Status);
Review Comment:
Fixed in 85533739f. For a 1.0 driver, all twelve imported getters (database,
connection and statement) now fall back to the base implementation, so they
report `NotFound`, as the C driver manager does. The test now expects
`NotFound` and checks each level.
Two related fixes:
- `AdbcConnection.GetOption` now reports `NotFound` when the autocommit,
read-only or isolation level property has no value. Before, an imported 1.0
connection reported `NotImplemented` for autocommit.
- The imported connection's `CurrentCatalog` and `CurrentDbSchema` called
`GetOption`. With the new fallback that recursed, so for a 1.0 driver they now
return null without calling it.
##########
csharp/src/Apache.Arrow.Adbc/AdbcDriver.cs:
##########
@@ -40,11 +44,48 @@ public abstract class AdbcDriver : IDisposable
/// Open a database via this driver.
/// </summary>
/// <param name="parameters">
- /// Driver-specific parameters.
+ /// Driver-specific parameters. Values may be strings, booleans,
integers,
+ /// floating-point numbers, byte arrays or <see
cref="ReadOnlyMemory{T}"/> of bytes.
/// </param>
+ /// <remarks>
+ /// The default implementation converts the non-binary values to
strings and calls
+ /// <see cref="Open(IReadOnlyDictionary{string, string})"/>, then
applies binary values
+ /// to the new database through <see
cref="AdbcDatabase.SetOptionBytes"/>. Drivers that
+ /// need binary options while opening should override this method.
+ /// </remarks>
public virtual AdbcDatabase Open(IReadOnlyDictionary<string, object>
parameters)
{
- throw AdbcException.NotImplemented("Driver does not support
non-string parameters");
+ if (parameters == null) throw new
ArgumentNullException(nameof(parameters));
+
+ Dictionary<string, string> stringParameters = new
Dictionary<string, string>(parameters.Count);
+ List<KeyValuePair<string, ReadOnlyMemory<byte>>> binaryParameters
= new List<KeyValuePair<string, ReadOnlyMemory<byte>>>();
+ foreach (KeyValuePair<string, object> pair in parameters)
+ {
+ if (AdbcOptionValue.TryGetBytes(pair.Value, out
ReadOnlyMemory<byte> bytes))
+ {
+ binaryParameters.Add(new KeyValuePair<string,
ReadOnlyMemory<byte>>(pair.Key, bytes));
+ }
+ else
+ {
+ stringParameters[pair.Key] =
AdbcOptionValue.ToOptionString(pair.Key, pair.Value);
+ }
+ }
+
+ AdbcDatabase database = Open(stringParameters);
+ try
+ {
+ foreach (KeyValuePair<string, ReadOnlyMemory<byte>> pair in
binaryParameters)
+ {
+ database.SetOptionBytes(pair.Key, pair.Value.Span);
Review Comment:
Added `DriverOpenWithObjectsAppliesBinaryValuesAfterOpening` and
`DriverOpenWithObjectsDisposesTheDatabaseIfABinaryValueFails` in 85533739f.
##########
csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs:
##########
@@ -19,14 +19,30 @@
using System.Collections.Generic;
using System.Linq;
using System.Runtime.InteropServices;
+using System.Text;
using Apache.Arrow.Adbc.Extensions;
using Apache.Arrow.C;
using Apache.Arrow.Ipc;
namespace Apache.Arrow.Adbc.C
{
+ /// <summary>
+ /// Exposes a managed <see cref="AdbcDriver"/> through the ADBC C API.
+ /// </summary>
+ /// <remarks>
+ /// When a caller requests ADBC 1.1.0, every 1.1.0 entry point is
populated.
+ /// Functionality that the managed driver doesn't override surfaces as
+ /// <see cref="AdbcStatusCode.NotImplemented"/>, which is the same status a
+ /// driver manager reports for a missing entry point.
Review Comment:
Fixed in 85533739f. The remarks now say that `ErrorFromArrayStream` isn't
provided, and that exported streams report errors through `get_last_error`
instead.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
