Skip to content

fix(csharp): support standard bulk ingest options #260

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davidhcoe merged 2 commits into from
Aug 19, 2026
Merged

fix(csharp): support standard bulk ingest options #260

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9bc2715
fix(csharp): support standard bulk ingest options
CurtHagenlocher Aug 19, 2026
9ecf8d8
fix(csharp): poll update jobs without fetching results
CurtHagenlocher Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 54 additions & 0 deletions csharp/src/AdbcDrivers.BigQuery/BigQueryStatement.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,41 @@ public override void SetOption(string key, string value)

switch (key)
{
case AdbcOptions.Ingest.TargetCatalog:
_isBulkIngest = true;
_ingestTargetCatalog = value;
break;
case AdbcOptions.Ingest.TargetDbSchema:
_isBulkIngest = true;
_ingestTargetDbSchema = value;
break;
case AdbcOptions.Ingest.TargetTable:
_isBulkIngest = true;
_ingestTargetTable = value;
break;
case AdbcOptions.Ingest.Mode:
_isBulkIngest = true;
_ingestMode = value switch
{
AdbcOptions.IngestMode.Create => BulkIngestMode.Create,
AdbcOptions.IngestMode.Append => BulkIngestMode.Append,
AdbcOptions.IngestMode.Replace => BulkIngestMode.Replace,
AdbcOptions.IngestMode.CreateAppend => BulkIngestMode.CreateAppend,
_ => throw new AdbcException($"Unsupported bulk ingest mode: {value}", AdbcStatusCode.InvalidArgument),
};
break;
case AdbcOptions.Ingest.Temporary:
_isBulkIngest = true;
switch (value)
{
case AdbcOptions.Enabled:
throw AdbcException.NotImplemented("Temporary table bulk ingest is not supported for BigQuery");
case AdbcOptions.Disabled:
break;
default:
throw new AdbcException($"Unsupported value for {AdbcOptions.Ingest.Temporary}: {value}", AdbcStatusCode.InvalidArgument);
}
break;
Comment on lines +158 to +180
case AdbcOptions.Telemetry.TraceParent:
SetTraceParent(string.IsNullOrWhiteSpace(value) ? null : value);
break;
Expand Down Expand Up @@ -881,6 +916,25 @@ private async Task<UpdateResult> ExecuteUpdateInternalAsync()
};
}

if (SqlQuery?.TrimStart().StartsWith("DROP TABLE", StringComparison.OrdinalIgnoreCase) == true)
Comment thread
CurtHagenlocher marked this conversation as resolved.
Outdated
{
Task<BigQueryJob> pollJobAsyncFunc()
{
return ExecuteCancellableJobAsync(context, activity, async (context, jobActivity) =>
{
context.Job = await this.Client.CreateQueryJobAsync(SqlQuery, null, updateQueryOptions, context.CancellationToken).ConfigureAwait(false);
jobActivity?.AddEvent("polluntilcompletedasync_started", [new("job.id", context.Job.Reference.JobId)]);
context.Job = await context.Job.PollUntilCompletedAsync(cancellationToken: context.CancellationToken).ConfigureAwait(false);
context.Job.ThrowOnAnyError();
jobActivity?.AddEvent("polluntilcompletedasync_completed", GetJobStatistics(jobActivity, context.Job));

return context.Job;
}, ClassName + "." + nameof(ExecuteUpdateInternalAsync) + "." + nameof(BigQueryJob.PollUntilCompletedAsync));
}
await ExecuteWithRetriesAsync(pollJobAsyncFunc, activity, context.CancellationToken);
return new UpdateResult(-1L);
}

Task<BigQueryResults> getQueryResultsAsyncFunc()
{
return ExecuteCancellableJobAsync(context, activity, async (context, jobActivity) =>
Expand Down
7 changes: 7 additions & 0 deletions csharp/test/AdbcDrivers.BigQuery.MockServer/BigQueryMockServer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ public sealed class BigQueryMockServer : IDisposable
private readonly ConcurrentDictionary<string, Table> _tables = new();
private readonly ConcurrentDictionary<string, bool> _sessions = new();
private readonly ConcurrentQueue<string> _executedQueries = new();
private int _queryResultsRequestCount;

/// <summary>
/// The REST API endpoint as host:port (e.g., "127.0.0.1:12345").
Expand All @@ -64,6 +65,11 @@ public sealed class BigQueryMockServer : IDisposable
/// </summary>
public IReadOnlyList<string> ExecutedQueries => _executedQueries.ToArray();

/// <summary>
/// The number of requests made to the query-results endpoint.
/// </summary>
public int QueryResultsRequestCount => _queryResultsRequestCount;

/// <summary>
/// The mock gRPC service for configuring Storage Read API responses.
/// </summary>
Expand Down Expand Up @@ -232,6 +238,7 @@ private void MapRestRoutes(WebApplication app)
// GET /bigquery/v2/projects/{projectId}/queries/{jobId} - Get query results
app.MapGet("/bigquery/v2/projects/{projectId}/queries/{jobId}", async (HttpContext ctx, string projectId, string jobId) =>
{
Interlocked.Increment(ref _queryResultsRequestCount);
if (!_jobs.TryGetValue(jobId, out var mockJob))
{
ctx.Response.StatusCode = 404;
Expand Down
116 changes: 116 additions & 0 deletions csharp/test/AdbcDrivers.BigQuery.Tests/MockServer/MockServerTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
#if NET8_0_OR_GREATER

using System.Collections.Generic;
using System.Linq;
using Apache.Arrow;
using Apache.Arrow.Adbc;
using Apache.Arrow.Types;
Expand Down Expand Up @@ -158,6 +159,121 @@ public async System.Threading.Tasks.Task CanBulkIngestAppendToTable()
// Verify the table was created in the REST API
// (CreateAppend mode should create it since it didn't exist)
}

[Theory]
[InlineData(AdbcOptions.IngestMode.Create, false)]
[InlineData(AdbcOptions.IngestMode.Append, true)]
[InlineData(AdbcOptions.IngestMode.Replace, true)]
[InlineData(AdbcOptions.IngestMode.CreateAppend, false)]
public void CanBulkIngestThroughStatementOptions(string mode, bool createTableFirst)
{
using var mockServer = new BigQueryMockServer();

const string projectId = "mock-project";
const string datasetId = "test_dataset";
string tableId = $"option_ingest_{mode.Substring(mode.LastIndexOf('.') + 1)}";
var parameters = new Dictionary<string, string>
{
{ BigQueryParameters.ProjectId, projectId },
{ BigQueryParameters.AuthenticationType, BigQueryConstants.MockAuthenticationType },
{ BigQueryParameters.TestRestEndpoint, mockServer.RestEndpoint },
{ BigQueryParameters.TestStorageEndpoint, mockServer.GrpcEndpoint },
};

using var driver = new BigQueryDriver();
using AdbcDatabase database = driver.Open(parameters);
using AdbcConnection connection = database.Connect(new Dictionary<string, string>());
using RecordBatch batch = CreateBatch();

if (createTableFirst)
{
using AdbcStatement create = connection.BulkIngest(projectId, datasetId, tableId, BulkIngestMode.Create, false);
create.Bind(batch, batch.Schema);
create.ExecuteUpdate();
}

using AdbcStatement statement = connection.CreateStatement();
statement.SetOption(AdbcOptions.Ingest.TargetCatalog, projectId);
statement.SetOption(AdbcOptions.Ingest.TargetDbSchema, datasetId);
statement.SetOption(AdbcOptions.Ingest.TargetTable, tableId);
statement.SetOption(AdbcOptions.Ingest.Temporary, AdbcOptions.Disabled);
statement.SetOption(AdbcOptions.Ingest.Mode, mode);
statement.Bind(batch, batch.Schema);

UpdateResult result = statement.ExecuteUpdate();

Assert.Equal(3, result.AffectedRows);
Assert.Empty(mockServer.ExecutedQueries);
var writeStream = mockServer.WriteService.Streams.Values.Last();
Assert.True(writeStream.Finalized);
Assert.Single(writeStream.RecordBatches);
}
Comment on lines +205 to +213

[Fact]
public void BulkIngestThroughStatementOptionsRejectsTemporaryTable()
{
using var mockServer = new BigQueryMockServer();
var parameters = new Dictionary<string, string>
{
{ BigQueryParameters.ProjectId, "mock-project" },
{ BigQueryParameters.AuthenticationType, BigQueryConstants.MockAuthenticationType },
{ BigQueryParameters.TestRestEndpoint, mockServer.RestEndpoint },
{ BigQueryParameters.TestStorageEndpoint, mockServer.GrpcEndpoint },
};

using var driver = new BigQueryDriver();
using AdbcDatabase database = driver.Open(parameters);
using AdbcConnection connection = database.Connect(new Dictionary<string, string>());
using AdbcStatement statement = connection.CreateStatement();

AdbcException exception = Assert.Throws<AdbcException>(
() => statement.SetOption(AdbcOptions.Ingest.Temporary, AdbcOptions.Enabled));

Assert.Equal(AdbcStatusCode.NotImplemented, exception.Status);
}

[Fact]
public void DropTableExecuteUpdateDoesNotRequestQueryResults()
{
using var mockServer = new BigQueryMockServer();
var parameters = new Dictionary<string, string>
{
{ BigQueryParameters.ProjectId, "mock-project" },
{ BigQueryParameters.AuthenticationType, BigQueryConstants.MockAuthenticationType },
{ BigQueryParameters.TestRestEndpoint, mockServer.RestEndpoint },
{ BigQueryParameters.TestStorageEndpoint, mockServer.GrpcEndpoint },
};

using var driver = new BigQueryDriver();
using AdbcDatabase database = driver.Open(parameters);
using AdbcConnection connection = database.Connect(new Dictionary<string, string>());
using AdbcStatement statement = connection.CreateStatement();
statement.SqlQuery = "DROP TABLE IF EXISTS `mock-project.test_dataset.test_table`";

UpdateResult result = statement.ExecuteUpdate();

Assert.Equal(-1, result.AffectedRows);
Assert.Equal(0, mockServer.QueryResultsRequestCount);
Assert.Single(mockServer.ExecutedQueries);
}

private static RecordBatch CreateBatch()
{
var schema = new Schema(new[]
{
new Field("id", Int64Type.Default, nullable: false),
new Field("name", StringType.Default, nullable: true),
}, null);

return new RecordBatch(
schema,
new IArrowArray[]
{
new Int64Array.Builder().Append(1).Append(2).Append(3).Build(),
new StringArray.Builder().Append("Alice").Append("Bob").Append("Charlie").Build(),
},
3);
}
}
}

Expand Down
Loading