GH-38255: [Go][C++] Implement Flight SQL Bulk Ingestion #38385

joellubi · 2023-10-22T10:11:44Z

Rationale for this change

It was suggested in the discussion around apache/arrow-adbc#1107 for the Flight SQL ADBC driver that an "Ingest" command would be a helpful addition to the Flight SQL specification. This command would enable a Flight SQL client to provide a FlightData stream to the server without needing to know its SQL syntax, and have that stream loaded into a target table by whichever means the server deems appropriate.

What changes are included in this PR?

Format:
- Add CommandStatementIngest message type to Flight SQL proto definition
- Add FLIGHT_SQL_SERVER_BULK_INGESTION and FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED options for SqlInfo
Go:
- Generate pb
- Server-side implementation
- Client-side implementation
- Unit + integration tests
C++:
- Server-side implementation
- Client-side implementation
- Integration tests

Are these changes tested?

Yes, see server_test.go, scenario.go, and test_integration.cc.

Are there any user-facing changes?

Yes, new Flight SQL client and server functionality is being added. Changes are not expected to break existing users.

Closes: [Format][Flight SQL] Add Flight SQL Command for Bulk Ingestion #38255

kou · 2023-10-22T20:51:47Z

Could you also add a new integration test to

arrow/dev/archery/archery/integration/runner.py

Lines 572 to 630 in f489996

    
           flight_scenarios = [ 
        
               Scenario( 
        
                   "auth:basic_proto", 
        
                   description="Authenticate using the BasicAuth protobuf."), 
        
               Scenario( 
        
                   "middleware", 
        
                   description="Ensure headers are propagated via middleware.", 
        
               ), 
        
               Scenario( 
        
                   "ordered", 
        
                   description="Ensure FlightInfo.ordered is supported.", 
        
                   skip_testers={"JS", "C#", "Rust"}, 
        
               ), 
        
               Scenario( 
        
                   "expiration_time:do_get", 
        
                   description=("Ensure FlightEndpoint.expiration_time with " 
        
                                "DoGet is working as expected."), 
        
                   skip_testers={"JS", "C#", "Rust"}, 
        
               ), 
        
               Scenario( 
        
                   "expiration_time:list_actions", 
        
                   description=("Ensure FlightEndpoint.expiration_time related " 
        
                                "pre-defined actions is working with ListActions " 
        
                                "as expected."), 
        
                   skip_testers={"JS", "C#", "Rust"}, 
        
               ), 
        
               Scenario( 
        
                   "expiration_time:cancel_flight_info", 
        
                   description=("Ensure FlightEndpoint.expiration_time and " 
        
                                "CancelFlightInfo are working as expected."), 
        
                   skip_testers={"JS", "C#", "Rust"}, 
        
               ), 
        
               Scenario( 
        
                   "expiration_time:renew_flight_endpoint", 
        
                   description=("Ensure FlightEndpoint.expiration_time and " 
        
                                "RenewFlightEndpoint are working as expected."), 
        
                   skip_testers={"JS", "C#", "Rust"}, 
        
               ), 
        
               Scenario( 
        
                   "poll_flight_info", 
        
                   description="Ensure PollFlightInfo is supported.", 
        
                   skip_testers={"JS", "C#", "Rust"} 
        
               ), 
        
               Scenario( 
        
                   "app_metadata_flight_info_endpoint", 
        
                   description="Ensure support FlightInfo and Endpoint app_metadata", 
        
                   skip_testers={"JS", "C#", "Rust", "Java"} 
        
               ), 
        
               Scenario( 
        
                   "flight_sql", 
        
                   description="Ensure Flight SQL protocol is working as expected.", 
        
                   skip_testers={"Rust"} 
        
               ), 
        
               Scenario( 
        
                   "flight_sql:extension", 
        
                   description="Ensure Flight SQL extensions work as expected.", 
        
                   skip_testers={"Rust"} 
        
               ), 
        
           ]

to run the integration test in our CI?

joellubi · 2023-10-23T09:10:50Z

@kou I added my tests under the existing scenario flight_sql, so my understanding was that they are already getting run. However I do see they're failing for implementations besides Go. Do you mean to create a new scenario in which other implementations besides Go are excluded?

lidavidm

We should create a new integration test scenario so that it can be individually skipped/enabled for each implementation

lidavidm · 2023-10-23T15:44:49Z

format/FlightSql.proto

+
+  /*
+   * Retrieves a boolean value indicating whether transactions are supported for bulk ingestion. If not, invoking
+   * the method commit in the context of a bulk ingestion is a noop, and the isolation level is
+   * `arrow.flight.protocol.sql.SqlTransactionIsolationLevel.TRANSACTION_NONE`.
+   *
+   * Returns:
+   * - false: if bulk ingestion transactions are unsupported;
+   * - true: if bulk ingestion transactions are supported.
+   */
+   INGEST_TRANSACTIONS_SUPPORTED = 577;


I think this should go in the 0-500 range

Makes sense. Done.

lidavidm · 2023-10-23T15:45:01Z

format/FlightSql.proto

+  /*
+   * Represents a bulk ingestion request. Used in the command member of FlightDescriptor
+   * for the the RPC call DoPut to cause the server load the contents of the stream's
+   * FlightData into the target destination.
+   */


nit: why the extra indent?

lidavidm · 2023-10-23T15:49:15Z

go/arrow/flight/flightsql/client.go

+	if err = proto.Unmarshal(res.GetAppMetadata(), &updateResult); err != nil {
+		return
+	}
+


I believe you have to drain the read-side of the stream here, even if technically the server should only have sent one message

Updated. Let me know what you think of the approach. We don't currently ensure the stream is drained in the other DoPut methods FWIW, but I can make changes there as well if it's preferred.

kou · 2023-10-23T20:17:13Z

I added my tests under the existing scenario flight_sql, so my understanding was that they are already getting run. However I do see they're failing for implementations besides Go. Do you mean to create a new scenario in which other implementations besides Go are excluded?

Yes. Sorry for not clarifying this.

joellubi · 2023-10-24T01:41:16Z

Yes. Sorry for not clarifying this.

@kou No problem, I created a separate scenario.

zeroshade · 2023-10-24T17:24:44Z

go/arrow/internal/flight_integration/scenario.go

+func (m *flightSqlIngestionScenarioTester) flightInfoForCommand(desc *flight.FlightDescriptor, schema *arrow.Schema) *flight.FlightInfo {
+	return &flight.FlightInfo{
+		Endpoint: []*flight.FlightEndpoint{
+			{Ticket: &flight.Ticket{Ticket: desc.Cmd}},
+		},
+		Schema:           flight.SerializeSchema(schema, memory.DefaultAllocator),
+		FlightDescriptor: desc,
+		TotalRecords:     -1,
+		TotalBytes:       -1,
+	}
+}


this isn't actually being used anywhere, that's why the CI is failing

Sorry, just removed.

zeroshade · 2023-10-24T17:45:29Z

go/arrow/flight/flightsql/client.go

+	wr = flight.NewRecordWriter(stream, ipc.WithAllocator(c.Alloc), ipc.WithSchema(rdr.Schema()))
+	wr.SetFlightDescriptor(desc)
+


don't we need defer wr.Close() ?

Good catch, forgot to do so. Just added.

zeroshade

This LGTM generally, though we need a second implementation before it can be voted on in the Mailing List to officially add this to the spec.

joellubi · 2023-10-26T09:22:21Z

This LGTM generally, though we need a second implementation before it can be voted on in the Mailing List to officially add this to the spec.

Thanks @zeroshade. I'm planning to get another implementation out in the next week or so.

lidavidm · 2024-04-09T22:24:32Z

@zeroshade or others, any last comments?

zeroshade · 2024-04-11T20:04:21Z

go/arrow/internal/flight_integration/scenario.go

+}
+
+// Options to assert before/after mocked ingest call
+func getIngestOptions() flightsql.ExecuteIngestOpts {


shouldn't this return a pointer to avoid the error about copying the mutex?

zeroshade · 2024-04-11T20:05:00Z

go/arrow/flight/flightsql/client.go

+	// Servers cannot infer defaults for these parameters, so we validate the request to ensure they are set.
+	if reqOptions.TableDefinitionOptions == nil {


should we validate that reqOptions is not nil first to avoid a potential panic?

Good point. Since we cast to the protobuf command type anyway, I did so a bit earlier to let it take care of nil handling for us. I added a nil test for completeness too.

zeroshade · 2024-04-11T20:06:07Z

go/arrow/flight/flightsql/client.go

+	rdr.Retain()
+	defer rdr.Release()


since you aren't passing the reader to anything outside of this function, I don't think this is necessary at all

Good catch, updated

zeroshade · 2024-04-11T20:14:26Z

go/arrow/flight/flightsql/types.go

 type CreatePreparedStatementResult = pb.ActionCreatePreparedStatementResult
+
+type TableDefinitionOptions = pb.CommandStatementIngest_TableDefinitionOptions
+
+type TableDefinitionOptionsTableNotExistOption = pb.CommandStatementIngest_TableDefinitionOptions_TableNotExistOption
+type TableDefinitionOptionsTableExistsOption = pb.CommandStatementIngest_TableDefinitionOptions_TableExistsOption


for clarity should we gather these into a single declaration block?

Suggested change

type CreatePreparedStatementResult = pb.ActionCreatePreparedStatementResult

type TableDefinitionOptions = pb.CommandStatementIngest_TableDefinitionOptions

type TableDefinitionOptionsTableNotExistOption = pb.CommandStatementIngest_TableDefinitionOptions_TableNotExistOption

type TableDefinitionOptionsTableExistsOption = pb.CommandStatementIngest_TableDefinitionOptions_TableExistsOption

type (

CreatePreparedStatementResult = pb.ActionCreatePreparedStatementResult

TableDefinitionOptions = pb.CommandStatementIngest_TableDefinitionOptions

TableDefinitionOptionsTableNotExistOption = pb.CommandStatementIngest_TableDefinitionOptions_TableNotExistOption

TableDefinitionOptionsTableExistsOption = pb.CommandStatementIngest_TableDefinitionOptions_TableExistsOption

)

?

Sure, I gathered the ones related to TableDefinitionOptions.

The CreatePreparedStatementResult type is unrelated.

zeroshade · 2024-04-11T20:16:38Z

Outside of my nitpicks, this all LGTM

joellubi · 2024-04-12T19:15:05Z

@zeroshade @lidavidm

I addressed the remaining comments. There are 2 failing workflows but they appear unrelated. If this is fine, would one of you please merge this or let me know if I should do so?

zeroshade · 2024-04-15T19:11:13Z

You're right, the failures are unrelated to your changes. This looks good to me. If no one else objects I'll merge this later this evening

joellubi · 2024-04-17T14:03:47Z

@zeroshade It looks like you're good with this merging and there have been no objections. I'll merge this shortly.

conbench-apache-arrow · 2024-04-18T07:05:02Z

After merging your PR, Conbench analyzed the 5 benchmarking runs that have been run so far on merge-commit 0d1ea5d.

There were no benchmark performance regressions. 🎉

The full Conbench report has more details. It also includes information about 12 possible false positives for unstable benchmarks that are known to sometimes produce them.

…#38385) ### Rationale for this change It was suggested in the discussion around apache/arrow-adbc#1107 for the Flight SQL ADBC driver that an "Ingest" command would be a helpful addition to the Flight SQL specification. This command would enable a Flight SQL client to provide a FlightData stream to the server without needing to know its SQL syntax, and have that stream loaded into a target table by whichever means the server deems appropriate. ### What changes are included in this PR? - Format: - Add CommandStatementIngest message type to Flight SQL proto definition - Add FLIGHT_SQL_SERVER_BULK_INGESTION and FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED options for SqlInfo - Go: - Generate pb - Server-side implementation - Client-side implementation - Unit + integration tests - C++: - Server-side implementation - Client-side implementation - Integration tests ### Are these changes tested? Yes, see `server_test.go`, `scenario.go`, and `test_integration.cc`. ### Are there any user-facing changes? Yes, new Flight SQL client and server functionality is being added. Changes are not expected to break existing users. * Closes: apache#38255 Lead-authored-by: Joel Lubinitsky <[email protected]> Co-authored-by: Joel Lubinitsky <[email protected]> Co-authored-by: Antoine Pitrou <[email protected]> Co-authored-by: Andrew Lamb <[email protected]> Signed-off-by: Matt Topol <[email protected]>

Please look at #38255 for details on this functionality. Support for Go and C++ was added as part of #38385. This pull request is to add the required support for Java. * GitHub Issue: #38255 Lead-authored-by: Amit Mittal <[email protected]> Co-authored-by: Amit Mittal <[email protected]> Signed-off-by: David Li <[email protected]>

) Please look at apache#38255 for details on this functionality. Support for Go and C++ was added as part of apache#38385. This pull request is to add the required support for Java. * GitHub Issue: apache#38255 Lead-authored-by: Amit Mittal <[email protected]> Co-authored-by: Amit Mittal <[email protected]> Signed-off-by: David Li <[email protected]>

github-actions bot added Component: Go awaiting review Awaiting review labels Oct 22, 2023

This was referenced Oct 22, 2023

GH-38255: [Format] Add Flight SQL Command for Bulk Ingestion #38256

Closed

go/adbc/driver/flightsql: support generic ingest apache/arrow-adbc#1107

Open

joellubi marked this pull request as ready for review October 22, 2023 10:39

joellubi requested a review from zeroshade as a code owner October 22, 2023 10:39

lidavidm self-requested a review October 22, 2023 14:05

lidavidm reviewed Oct 23, 2023

View reviewed changes

github-actions bot added awaiting changes Awaiting changes and removed awaiting review Awaiting review labels Oct 23, 2023

github-actions bot added awaiting change review Awaiting change review and removed awaiting changes Awaiting changes labels Oct 23, 2023

joellubi requested review from assignUser, kou and raulcd as code owners October 24, 2023 00:28

zeroshade reviewed Oct 24, 2023

View reviewed changes

github-actions bot added awaiting changes Awaiting changes and removed awaiting change review Awaiting change review labels Oct 24, 2023

zeroshade reviewed Oct 24, 2023

View reviewed changes

github-actions bot added awaiting change review Awaiting change review and removed awaiting changes Awaiting changes labels Oct 24, 2023

zeroshade reviewed Oct 25, 2023

View reviewed changes

github-actions bot added awaiting changes Awaiting changes and removed awaiting change review Awaiting change review labels Oct 25, 2023

joellubi force-pushed the add-flight-sql-ingestion-go-impl branch from 434d537 to db70c20 Compare October 31, 2023 10:14

github-actions bot added the awaiting change review Awaiting change review label Apr 9, 2024

zeroshade reviewed Apr 11, 2024

View reviewed changes

github-actions bot added awaiting changes Awaiting changes and removed awaiting change review Awaiting change review labels Apr 11, 2024

zeroshade reviewed Apr 11, 2024

View reviewed changes

joellubi added 3 commits April 12, 2024 13:49

improve client nil handling

ca1e70a

consolidate table def types into a single block

b568efe

getIngestOptions return pointer

60a9bd4

github-actions bot added awaiting change review Awaiting change review and removed awaiting changes Awaiting changes labels Apr 12, 2024

joellubi requested a review from zeroshade April 12, 2024 17:53

zeroshade merged commit 0d1ea5d into apache:main Apr 17, 2024
56 of 58 checks passed

zeroshade removed the awaiting change review Awaiting change review label Apr 17, 2024

eramitmittal mentioned this pull request Aug 4, 2024

GH-38255: [Java] Implement Flight SQL Bulk Ingestion #43551

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

GH-38255: [Go][C++] Implement Flight SQL Bulk Ingestion #38385

GH-38255: [Go][C++] Implement Flight SQL Bulk Ingestion #38385

joellubi commented Oct 22, 2023 •

edited

Loading

kou commented Oct 22, 2023

joellubi commented Oct 23, 2023

lidavidm left a comment

lidavidm Oct 23, 2023

joellubi Oct 24, 2023

lidavidm Oct 23, 2023

joellubi Oct 23, 2023

lidavidm Oct 23, 2023

joellubi Oct 24, 2023

kou commented Oct 23, 2023

joellubi commented Oct 24, 2023

zeroshade Oct 24, 2023

joellubi Oct 24, 2023

zeroshade Oct 24, 2023

joellubi Oct 24, 2023

zeroshade left a comment

joellubi commented Oct 26, 2023

lidavidm commented Apr 9, 2024

zeroshade Apr 11, 2024

joellubi Apr 12, 2024

zeroshade Apr 11, 2024

joellubi Apr 12, 2024

zeroshade Apr 11, 2024

joellubi Apr 12, 2024

zeroshade Apr 11, 2024

joellubi Apr 12, 2024

zeroshade commented Apr 11, 2024

joellubi commented Apr 12, 2024

zeroshade commented Apr 15, 2024

joellubi commented Apr 17, 2024

conbench-apache-arrow bot commented Apr 18, 2024

		wr = flight.NewRecordWriter(stream, ipc.WithAllocator(c.Alloc), ipc.WithSchema(rdr.Schema()))
		wr.SetFlightDescriptor(desc)

		// Servers cannot infer defaults for these parameters, so we validate the request to ensure they are set.
		if reqOptions.TableDefinitionOptions == nil {

GH-38255: [Go][C++] Implement Flight SQL Bulk Ingestion #38385

GH-38255: [Go][C++] Implement Flight SQL Bulk Ingestion #38385

Conversation

joellubi commented Oct 22, 2023 • edited Loading

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

kou commented Oct 22, 2023

joellubi commented Oct 23, 2023

lidavidm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kou commented Oct 23, 2023

joellubi commented Oct 24, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zeroshade left a comment

Choose a reason for hiding this comment

joellubi commented Oct 26, 2023

lidavidm commented Apr 9, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zeroshade commented Apr 11, 2024

joellubi commented Apr 12, 2024

zeroshade commented Apr 15, 2024

joellubi commented Apr 17, 2024

conbench-apache-arrow bot commented Apr 18, 2024

joellubi commented Oct 22, 2023 •

edited

Loading