ARROW-17688: [C++][Java][FlightRPC] Substrait, transaction, cancellation for Flight SQL

[lidavidm]

"[VOTE] Substrait for Flight SQL"
https://lists.apache.org/thread/3k3np6314dwb0n7n1hrfwony5fcy7kzl

[github-actions]

Thanks for opening a pull request!

If this is not a minor PR. Could you open an issue for this pull request on JIRA? https://issues.apache.org/jira/browse/ARROW

Opening JIRAs ahead of time contributes to the Openness of the Apache Arrow project.

Then could you also rename pull request title in the following format?

ARROW-${JIRA_ID}: [${COMPONENT}] ${SUMMARY}

or

MINOR: [${COMPONENT}] ${SUMMARY}

See also:

• Other pull requests
• Contribution Guidelines - How to contribute patches

[emkornfield]

what is the difference between transcations and save points? Are there docs someplace?

[lidavidm]

The terminology is taken from JDBC and some databases, e.g. see Postgres, JDBC. I will add some more language so this is reasonably self-contained.

[emkornfield]

I haven't been keeping up but is serialized plan sufficient, or are yaml plugins necessary?

[lidavidm]

The root plan contains extension definitions, and I believe they're intended to be self-contained as a result, but I'll seek some clarification: https://github.com/substrait-io/substrait/blob/1080f06298d8e50abcd6acfaa6c425326a7e0579/proto/substrait/plan.proto#L24-L45

[zeroshade]

If we're expecting the Protobuf serialized plan here, would it make more sense to just import the substrait proto definition and reference the object directly rather than having to serialize the plan and then stick the bytes inside another serialized protobuf?

[lidavidm]

I'd like to keep code dependencies on other Protobuf messages out because Windows/Protobuf has issues with those when they're in different DLLs (as they are with Flight/Flight SQL, and as they would be here). As seen below with the CancelQuery message I already ran into linking issues and I think they're insurmountable unless protoc itself is modified.

[lidavidm]

On the wire, the actual encoding is the same either way.

[zeroshade]

@lidavidm Fair enough, that makes sense.

[emkornfield]

should server timeout be something that the client has the option of specifying? Can it be introspected?

[lidavidm]

I chose to follow prepared statements in this regard. I'll add a SqlInfo value to retrieve the timeout.

[emkornfield]

is this misplaced should it be above name?

[emkornfield]

is transaction_id optional here for a new transaction? it seems like in most cases this should be server assigned?

[lidavidm]

Hopefully it's clearer now that it's split. The ID is always server assigned. To begin a savepoint, you must provide the ID of the transaction it falls under.

[emkornfield]

what are the semantics relative to savepoint here?

[lidavidm]

I clarified the docstring, though I'm not sure what you're referring to here.

[emkornfield]

transaction + savepoint?

[lidavidm]

Yes, updated docstring.

[emkornfield]

from an RPC optimization standpoint, it seems that maybe we want a way of specifying this should be considered the first action inside of a transaction?

[emkornfield]

It seems like maybe there should be something like:

message TransactionDetails {
   oneof transaction {
     ActionBeginTranscation begin_transaction = 1;
     bytes existing_transaction_id = 2;
   }
}

that can be included.

[lidavidm]

The intent was to mimic prepared statements: the server assigns the transaction ID and gives it to the client. So here there's not a great way to return the transaction ID to the client. It would be good if FlightInfo could gain an app_metadata field for such things (since Tickets are not meant to be client-introspected).

[lidavidm]

I suppose we could use schema metadata for that, though.

[emkornfield]

it seems app_metadata would probably be generally useful. It seems like a hack to force the info onto the schema.

[lidavidm]

I'll propose that separately when I get a chance (and I'll see if I can set aside some time to help with the small-result optimization stuff)

[emkornfield]

does there need to be a response associated with this?

[lidavidm]

What would it contain?

[emkornfield]

Could state. CANCELLED, NOT_CANCELLABLE, ALREADY_DONE, CANCELLING. Not sure this is useful, so feel free to ignore.

[lidavidm]

If we declare the command idempotent, that could be useful. I suppose the server could just synchronously block until the query is cancelled but then it'd be unclear how to recover from a transient failure.

[emkornfield]

should this be some sort of ID instead?

[lidavidm]

What would it be? Since most of the data inside FlightInfo isn't meant to be introspected, and we haven't specified what the contents of Ticket should be, there's no consistent 'query ID' concept right now. So I chose FlightInfo since presumably that has all information the server needs to identify the query.

[emkornfield]

would this be another place where app_metadata would be useful? Otherwise, it seems like the only implementation path for the server would be to introspect one of the tickets? (it would seem lfight descriptors might very commonly be non-unique). This might be fine, I just want to confirm my understanding.

[lidavidm]

Yes - if we had FlightInfo.app_metadata, we could return an explicit cancellation token. But given the server is generating the tickets, it should be OK for the server to also introspect them, so long as the client doesn't.

[emkornfield]

yeah, I think introspecting the tickets for now should be fine.

[emkornfield]

are there semantics that need to be considered relative to transcations?

[lidavidm]

I'll update it to specify that the transaction is not rolled back/this is only to terminate reading of the result set (CC @jduo is that in line with what you were thinking?)

[jduo]

This is what I was thinking in terms of read, but also think it should terminate a running write (same as in ODBC/JDBC).

[lidavidm]

Thanks - updated. The semantics around cancelling a write are a little unclear, I suppose in autocommit mode it gets committed, and otherwise the client can commit or rollback.

[emkornfield]

Generally seems OK to me.

[pitrou]

So "Flight SQL" is generic enough that it might support things other than actual SQL?

[lidavidm]

Yes, the naming is unfortunate, but there's no real reason why we need to be tied to SQL specifically, and no reason why many of the existing concepts can't map to Substrait.

[pitrou]

Er... which unit is that? Seconds? Can we make it a real/float instead, if protobuf allows that?

[lidavidm]

I made it milliseconds, unless we prefer floating point seconds?

[pitrou]

Just for my understanding, savepoints are for two-phase commits, right?

[lidavidm]

Hmm, not in the distributed systems sense, a savepoint is really just a nested transaction. I included it for parity with JDBC but it's not essential.

[pitrou]

Hmm... should there be two separate commands for starting a transaction and a savepoint, so that they can take different parameters?

[lidavidm]

Split into separate commands (though the response messasge and EndTransaction are still shared)

[pitrou]

Does the "default" happen if transaction_id is left unset?

[pitrou]

If so, perhaps replace "by default" with "if unset, " for clarity?

[lidavidm]

Should we specify that the timeout is refreshed when the handle is "used"?

[zeroshade]

I think so, we should probably provide any semantics necessary for when a timeout is refreshed.

[lidavidm]

These transaction_id fields should be explicitly marked optional

[zeroshade]

technically, by definition in proto3 all fields are optional and it's unnecessary to explicitly mark any as such.

[lidavidm]

Flight SQL uses explicit optional elsewhere though (just to make it clear what is expected), so Flight SQL actually has a higher protoc minimum version as a result

[lidavidm]

Hopefully it's clearer now that it's split. The ID is always server assigned. To begin a savepoint, you must provide the ID of the transaction it falls under.

[lidavidm]

I've been pushing various tweaks to the spec while implementing it in C++.

One thing I will need to change: transaction IDs should be supplied when creating a prepared statement, not when executing them (since generally existing APIs associate the statement with a particular connection).

[lidavidm]

There's now implementations in C++ and Java.

TODOs:

• Integration tests in C++
• https://issues.apache.org/jira/browse/ARROW-17254
• Ensure integration tests work across languages
• Fix C++ API/Proto comments (methods should accept only Transaction values)
• Add wrapper types for new types in Java (stop using the Protobuf classes directly)
• Implement transactions in the Derby example (depends on ARROW-17199: [Java][FlightRPC] Clean up Flight SQL example server #13710)

[lidavidm]

Hmm, Windows builds fail because of a similar issue to #13434 - Protobuf and DLLs don't interact well, since you can't get protoc to insert the dllimport/dllexport declarations correctly.

The easiest thing might be to just punt on CancelQuery for now. Or else, it would have to be

message ActionCancelQueryRequest {
  // XXX(ARROW-16902): A serialized FlightInfo   
  bytes info = 1;
}

and then rely on FlightInfo::Deserialize.

[pitrou]

@lidavidm Since this is a draft, are you looking for a detailed review or more for general opinions?

[lidavidm]

The Protobuf definitions deserve more scrutiny; for the code, I'm just looking for general opinions.

[lidavidm]

No worries, just want to make sure!

[zeroshade]

Added some comments for clarification

[zeroshade]

I think so, we should probably provide any semantics necessary for when a timeout is refreshed.

[zeroshade]

Is there a functional difference here? Or are we just including an "Unknown" so that it will be the default? In most cases developers are likely to treat "unknown" the same as "none" when it comes to transaction support: (ie. don't try calling them)

[lidavidm]

Not really - I guess in that case let's just fold them together since there's no point. (I suppose normally in protobuf you'd distinguish the two, but that doesn't apply here.)

[zeroshade]

yea, I agree that in protobuf you'd distinguish the two in many cases, but my typical litmus test is if a consumer of the protobuf would treat the "unknown" case differently than the "none" case. And if there isn't any functional difference, it's not necessary to distinguish them.

[lidavidm]

Yup - I ended up folding them together

[zeroshade]

If we're expecting the Protobuf serialized plan here, would it make more sense to just import the substrait proto definition and reference the object directly rather than having to serialize the plan and then stick the bytes inside another serialized protobuf?

[zeroshade]

Should this be a serialized flightinfo? or should it actually be a Ticket?

[lidavidm]

It should be a FlightInfo because I assume the server needs the information of all endpoints in order to fully cancel a query. Also in the event that we do update FlightInfo with an application metadata field, it would automatically get passed back to the server

[zeroshade]

Gotcha, that makes sense then.

[zeroshade]

If the server returns CANCEL_RESULT_CANCELLING is a client supposed to poll with subsequent cancel requests until it receives CANCELLED? Or is there a different way to determine when the cancellation is completed?

[lidavidm]

That was the intent. I'll document the variants.

[lidavidm]

Updated - is this clearer?

[zeroshade]

yup much better! thanks!

[lidavidm]

Updated again, since @jvanstraten pointed out that the server may want to know the client's Substrait release version since otherwise it may be unclear how to interpret the plan (even if it parses properly).

Also, adds some validation for SqlInfo values to the integration test + adds SqlInfo values so the server can report Substrait version support.

[pitrou]

Hmm... what is this for?

[lidavidm]

Hmm, I think I needed it originally, but maybe now that we aren't referencing Protobuf files from each other it's not needed anymore - removed

[pitrou]

Should perhaps use a URI placeholder and produce the proper URI programmatically?

Suggested change

	"uri_file": "file://FILENAME_PLACEHOLDER",
	"uri_file": "URI_PLACEHOLDER",

[pitrou]

Perhaps expose a server_->connect_uri() instead?

[lidavidm]

I was trying not to add new methods (MakeAceroServer just returns the base FlightSqlServerBase) and a server doesn't necessarily know what its 'public' address is (even here, we bind to 0.0.0.0 but assume that it's accessible on 'localhost')

[pitrou]

and a server doesn't necessarily know what its 'public' address is (even here, we bind to 0.0.0.0 but assume that it's accessible on 'localhost')

Hmm... if you mean the server might reside behind a NAT, sure, but at least from a local point of view it should know on which addresses (plural, ideally :-)) it is reachable (perhaps only localhost, or perhaps one interface, or perhaps all/many interfaces...).

[lidavidm]

Updated to use server_->location() and bind to localhost instead of 0.0.0.0 (should be OK/preferable in test code anyways)

[pitrou]

Can you perhaps make parameters more explicit?

Suggested change

	client_->GetSqlInfo({}, {
	client_->GetSqlInfo(/*abc=*/{}, /*def=*/{

[pitrou]

Same here (at least for the first one :-)).

[pitrou]

Do you plan to remove the "experimental" markers like the above?

[lidavidm]

I think not in this RFC, but we should do it in the near future (perhaps after the JDBC driver has seen some use)

[pitrou]

Is there a particular rationale for adding this option in some message definitions but not all of them?

[lidavidm]

No, I just missed it :) I'll fix that.

[lidavidm]

Thanks for the comments, I'll update this soon. (I guess these are mostly minor changes and we shouldn't need to restart the vote?)

[lidavidm]

As I understand, the backwards compatibility story is not yet worked out, so in the future, this may be less useful, but for now, it's the only way to reliably determine whether a plan can really be executed.

[lidavidm]

Maybe @jduo can chime in; this timeout existed implicitly before, but something here is necessary because unlike JDBC/ODBC which can tie these to the lifetime of an actual connection, Flight SQL makes fewer assumptions about state being tied to the gRPC connection (which are more disposable).

[lidavidm]

I think not in this RFC, but we should do it in the near future (perhaps after the JDBC driver has seen some use)

[lidavidm]

No, I just missed it :) I'll fix that.

[lidavidm]

Hmm, I think that makes sense, though current clients are going to assume SQL support.

[lidavidm]

I'll update this.

[pitrou]

(I guess these are mostly minor changes and we shouldn't need to restart the vote?)

Yes, definitely.

Also, sorry, the review contains C++ comments that I did some days/weeks ago but had forgotten to submit apparently :-S

[lidavidm]

Rebased + updated (minus the timeout since we'd have to change the result set schema for GetSqlInfo to add floating point)

[github-actions]

https://issues.apache.org/jira/browse/ARROW-17688

[pitrou]

I took a quick look at the C++ parts again.

[pitrou]

Use checked_cast here and below?

[pitrou]

Would be nice to add docstrings/comments explaining each non-trivial helper class here.

[pitrou]

Nit, but the logic to create the FlightInfo from an encoded substrait plan could perhaps be factored out in a dedicated helper method? (since GetFlightInfoPreparedStatement has the same logic inside)

[pitrou]

Nit

Suggested change

	std::string db_uri_;
	const std::string db_uri_;

[pitrou]

I see that accesses to prepared_statements_ are never mutex-protected, is it right?

[pitrou]

Should be mutex-protected?

[pitrou]

Not sure how efficient you want this to be, but you might release the lock around these lines (and call open_transactions_.erase before?).

[lidavidm]

Updated, thanks Antoine!

[lidavidm]

CI failures here are addressed/fixed elsewhere

[ursabot]

Benchmark runs are scheduled for baseline = d571e93 and contender = 3ce4014. 3ce4014 is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Finished ⬇️0.0% ⬆️0.0%] ec2-t3-xlarge-us-east-2
[Failed ⬇️0.1% ⬆️0.0%] test-mac-arm
[Failed ⬇️0.28% ⬆️0.0%] ursa-i9-9960x
[Finished ⬇️0.21% ⬆️0.0%] ursa-thinkcentre-m75q
Buildkite builds:
[Finished] 3ce40143 ec2-t3-xlarge-us-east-2
[Failed] 3ce40143 test-mac-arm
[Failed] 3ce40143 ursa-i9-9960x
[Finished] 3ce40143 ursa-thinkcentre-m75q
[Finished] d571e93a ec2-t3-xlarge-us-east-2
[Failed] d571e93a test-mac-arm
[Failed] d571e93a ursa-i9-9960x
[Finished] d571e93a ursa-thinkcentre-m75q
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java

[ursabot]

['Python', 'R'] benchmarks have high level of regressions.
test-mac-arm
ursa-i9-9960x