This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new 0c6bc243 fix(arrow/flight/flightsql): send schema for empty ingest
(#1433)
0c6bc243 is described below
commit 0c6bc24309135517992a5070be6c7495e2ec53ca
Author: Arnold Wakim <[email protected]>
AuthorDate: Thu Oct 29 17:05:36 2026 +0100
fix(arrow/flight/flightsql): send schema for empty ingest (#1433)
### Rationale for this change
Flight SQL bulk ingestion fails with `InvalidArgument: failed to read
input stream: EOF` when the input had a schema but no record batches.
The writer was closed after the gRPC send side, so the schema and
descriptor were never sent.
This issue is also observable through native ADBC integrations.
The ADBC Flight SQL driver exposes the standard C ABI as a shared lib,
but its implem uses the Arrow Go Flight SQL client internally.
Native consumers such as DuckDB’s `adbc_scanner` therefore reach
`flightsql.Client.ExecuteIngest` through the C-to-Go boundary. When the
supplied ArrowArrayStream contains a schema but no record batches, the
Go client previously closed the gRPC send side before emitting the IPC
schema and Flight descriptor, causing the server to receive an immediate
EOF.
To reproduce the bug you can execute the following:
```bash
docker run --name gizmosql \
--detach \
--rm \
--tty \
--init \
--publish 31337:31337 \
--env TLS_ENABLED="1" \
--env GIZMOSQL_PASSWORD="gizmosql_password" \
--env PRINT_QUERIES="1" \
--pull always \
gizmodata/gizmosql:latest
```
use duckdb v2.0.0:
```duckdb
LOAD adbc;
ATTACH 'profile://gizmosql_local' AS gizmosql_local (TYPE adbc);
CREATE TABLE IF NOT EXISTS gizmosql_local.main.users(id INTEGER, name TEXT);
IO Error:
[FlightSQL] Stream finished before first message sent (Unknown;
FlightSQL.BulkIngest.Execute)
Run Time (s): real 0.016 user 0.005183 sys 0.004606
```
### What changes are included in this PR?
- Close the Flight record writer before closing the send side.
- Preserve cleanup and server-error handling.
- Add a regression test for ingesting a non-empty schema with zero
record batches.
### Are these changes tested?
with: `go test ./arrow/flight/flightsql -count=1`
The regression test fails with `EOF` before the fix and passes afterward
with zero rows ingested.
### Are there any user-facing changes?
There are no API changes.
---
arrow/flight/flightsql/client.go | 11 +++++++++--
arrow/flight/flightsql/server_test.go | 24 ++++++++++++++++++++++++
2 files changed, 33 insertions(+), 2 deletions(-)
diff --git a/arrow/flight/flightsql/client.go b/arrow/flight/flightsql/client.go
index 6b640a10..a3b34e8e 100644
--- a/arrow/flight/flightsql/client.go
+++ b/arrow/flight/flightsql/client.go
@@ -290,7 +290,11 @@ func (c *Client) executeIngest(ctx context.Context, rdr
array.RecordReader, reqO
writerOpts = append(writerOpts, ipc.WithAllocator(c.Alloc),
ipc.WithSchema(rdr.Schema()))
writerOpts = append(writerOpts, ipcOpts...)
wr = flight.NewRecordWriter(stream, writerOpts...)
- defer wr.Close()
+ defer func() {
+ if wr != nil {
+ _ = wr.Close()
+ }
+ }()
wr.SetFlightDescriptor(desc)
@@ -312,9 +316,12 @@ func (c *Client) executeIngest(ctx context.Context, rdr
array.RecordReader, reqO
return 0, err
}
- if err = stream.CloseSend(); err != nil {
+ if err = wr.Close(); err != nil && !errors.Is(err, io.EOF) {
return 0, err
}
+ wr = nil
+
+ _ = stream.CloseSend()
if res, err = stream.Recv(); err != nil {
return 0, err
diff --git a/arrow/flight/flightsql/server_test.go
b/arrow/flight/flightsql/server_test.go
index a8526e67..51075440 100644
--- a/arrow/flight/flightsql/server_test.go
+++ b/arrow/flight/flightsql/server_test.go
@@ -536,6 +536,30 @@ func (s *FlightSqlServerSuite) TestExecuteIngest() {
s.Equal(nRowsExpected, nRowsIngested)
}
+func (s *FlightSqlServerSuite) TestExecuteIngestEmptyReader() {
+ schema := arrow.NewSchema(
+ []arrow.Field{{Name: "id", Type: arrow.PrimitiveTypes.Int64}},
+ nil,
+ )
+ rdr, err := array.NewRecordReader(schema, nil)
+ s.Require().NoError(err)
+ defer rdr.Release()
+
+ nRowsIngested, err := s.cl.ExecuteIngest(
+ context.TODO(),
+ rdr,
+ &flightsql.ExecuteIngestOpts{
+ TableDefinitionOptions:
&flightsql.TableDefinitionOptions{
+ IfNotExist:
flightsql.TableDefinitionOptionsTableNotExistOptionCreate,
+ IfExists:
flightsql.TableDefinitionOptionsTableExistsOptionReplace,
+ },
+ Table: "empty_test_table",
+ },
+ )
+ s.NoError(err)
+ s.Zero(nRowsIngested)
+}
+
func (s *FlightSqlServerSuite) TestExecuteIngestWithServerError() {
nRecords := 11 // intentionally exceed maximum number of rows the
server can ingest
nRowsPerRecord := 5
