joellubi commented on code in PR #38385:
URL: https://github.com/apache/arrow/pull/38385#discussion_r1384165515
##########
go/arrow/flight/flightsql/client.go:
##########
@@ -218,6 +218,56 @@ func (c *Client) ExecuteSubstraitUpdate(ctx
context.Context, plan SubstraitPlan,
return updateResult.GetRecordCount(), nil
}
+// ExecuteIngest is for executing a bulk ingestion and only returns the number
of affected rows.
+func (c *Client) ExecuteIngest(ctx context.Context, rdr array.RecordReader,
reqOptions *ExecuteIngestOpts, opts ...grpc.CallOption) (n int64, err error) {
+ var (
+ desc *flight.FlightDescriptor
+ stream pb.FlightService_DoPutClient
+ wr *flight.Writer
+ res *pb.PutResult
+ updateResult pb.DoPutUpdateResult
+ )
+
+ cmd := (*pb.CommandStatementIngest)(reqOptions)
+ if desc, err = descForCommand(cmd); err != nil {
+ return
+ }
+
+ if stream, err = c.Client.DoPut(ctx, opts...); err != nil {
+ return
+ }
+
+ wr = flight.NewRecordWriter(stream, ipc.WithAllocator(c.Alloc),
ipc.WithSchema(rdr.Schema()))
+ defer wr.Close()
+
+ wr.SetFlightDescriptor(desc)
+
+ for rdr.Next() {
+ rec := rdr.Record()
+ wr.Write(rec)
+ }
+
+ if err = stream.CloseSend(); err != nil {
+ return
+ }
+
+ if res, err = stream.Recv(); err != nil {
+ return
+ }
+
+ // Drain the stream, ensuring only a single result was sent by the
server.
+ // Flight RPC allows multiple server results, so the check may be
removed if Flight SQL ingestion defines semanatics for this scenario.
+ if _, err = stream.Recv(); err != io.EOF {
+ return
+ }
Review Comment:
Good catch. Looking at this now it definitely is incorrect to return the nil
error. Do you think it makes more sense to stick with the same approach but
instead return an explicit error right away, or to continue draining the stream
until the end even though the client doesn't know how to handle the subsequent
payloads?
The former is nice because it fails fast upon entering an undefined state,
but the latter might make for a more well-behaved client. In your comment about
needing to drain this stream, were you referring to a convention defined for
Flight RPC, GRPC in general, or something else?
--
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]
