Add RFC for supporting distributed procedure

[hantangwangd]

Propose design to expand the current procedure architecture in presto, support defining, registering and calling procedures which need to be executed in a distributed way.

Besides, in order to demonstrate as an example and to figure out the functional boundaries among the different architecture levels, also describe the design for Iceberg to support distributed procedure as well as the design of a specific distributed procedure rewrite_data_files.

[rschlussel]

This is an interesting feature. do you have thoughts on integrating this for cpp workers as well?

[hantangwangd]

@rschlussel Thanks for the comment. Yes, of course we should integrate this for cpp workers if it's proved to be useful and needed.

I think we can take a two-step approach for this. Firstly, we support it on java workers, confirm the feasible of the entire architecture and figure out the best functional boundary division. Then, after that, we can support it on cpp workers following the design and implementation path on java workers. What's your opinion?

[aditi-pandit]

Thanks @hantangwangd for this RFC. Had couple of comments mainly related to Native execution related impact.

[aditi-pandit]

Native vs Java runtimes begin to diverge significantly at this point. For native runtimes its best if the distributed procedure call method is in C++. But that would mean older java procedures would need a rewrite. Do you have any particular ideas on how you will handle this for Native engine ?

[hantangwangd]

Thanks for the comment. Actually, beginCallDistributedProcedure and finishCallDistributedProcedure in ConnectorMetadata are all invoked in coordinator, so native worker do not need to handle them, that is, there is no need in native worker end to resolve the distributed procedures and invoke them.

The non-coordinator worker's responsibility in call distributed procedure is similar to that of insert into or create table as. That is, local planning CallDistributedProcedureNode to a TableWriterOperator with ExecutionWriterTarget, getting corresponding ConnectorPageSink based on this writer target, executing the data writing and finally returning the fragments page to table finish stage.

So I think there is no need to consider this issue, what do you think? Any misunderstand please let me know.

[aditi-pandit]

Presume the procedureArguments have types. There could be a difference in the types supported by native vs java execution. Would be great to clarify about that.

[hantangwangd]

As I understand, CallDistributedProcedureTarget is a subclass of WriterTarget which is used only in coordinator. That is, it's always used in java execution environment.

ExecuteProcedureHandle which is a subclass of ExecutionWriterTarget will be sent to worker. The handling of it is similar to InsertHandle or CreateHandle.

[aditi-pandit]

Would be great to give more details on what row/control messages are exchanged between the CallDistricutedProcedureNode and TableFinishNode. e.g. TableWriteNode provides protocol of PartitionUpdates and CommitContext that is shared between the two.

[hantangwangd]

Sure, it's a good suggestion. I will supplement this part of the content.

[aditi-pandit]

Iceberg in Native engine uses HiveConnector itself. The HiveConnector in Native engine handles both Hive and Iceberg splits. Does HiveConnector support distributed procedure ? How will it be reused or enhanced for Iceberg ?

[hantangwangd]

As beginCallDistributedProcedure and finishCallDistributedProcedure in ConnectorMetadata are all invoked in coordinator, I think the main change for hive to support distributed procedures is in java implementation.

Hive should implement beginCallDistributedProcedure and finishCallDistributedProcedure in ConnectorMetadata, customize it's own preparation and submission logic, and then implement and register its own distributed procedures. All these jobs could be done in java only.

So I think the main change in worker end logic which need to support native c++ is that it should generate and provide ConnectorPageSink based on the newly added ConnectorDistributedProcedureHandle. Please let me know if there are any omissions.

[jaystarshot]

Not sure I understand correctly, but if we are going to support a general distributed procedure then why does it have to be just above TableScanNode or table finish?
If its not for a general case but specific to table layouts, then why not just a custom operator implementation?

[hantangwangd]

Thanks for the comment. As described in this RFC, compared to existing coordinator-only procedures, these newly expanded kind of procedures which need to be executed distributively involve operations on table's data other than metadata. For example, these procedures can rewrite table data, merge small data files, sort table data, repartition table data etc. So the common action process of them includes reading data from the target table, doing some filtering and translation, writing the translated data to files and submitting the changes in metadata level. So I think the general distributed procedure could be planned to a plan tree like this. (Or adding some other plan nodes like JoinNode/AggregationNode etc. for functional expansion in future)

[hantangwangd]

why not just a custom operator implementation?

After optimization and plan fragmentation, we should be able to shuffle data between stages to utilize the capabilities of the entire cluster.

[jaystarshot]

Can you also define what you mean by procedure?

[hantangwangd]

Thanks for the suggestion. Strictly speaking, maybe they should be called connector specific system procedures. I'll add some explanations here.

[hantangwangd]

This RFC is ready for review, the comments are all handled, and the relevant implementation can be viewed here: prestodb/presto#22659. Please take a look when convenient, thanks! cc: @rschlussel @aditi-pandit @jaystarshot @tdcmeehan @ZacBlanco @yingsu00

[aditi-pandit]

Its not clear what is the query syntax proposed. Are you proposing a CALL statement or ALTER TABLE ? Can you give some sample examples with their statements here. It would be great to specify the statement syntax specifying what parameters will be and their behavior/limitations.

In the future sections a table and filter are used. How are they specified in the statement used by the end user ?

[hantangwangd]

Thanks for your suggestion. I will supplement this part of the content. The following is a brief explanation.

We propose the syntax of Call statement for the reason described in line 24 to line 28 (https://github.com/prestodb/rfcs/pull/12/files#diff-7655a74167c417d09fba4cc48ec39232f65d7056a449e1cb09d51664f4fe1b7eR24-R28), for example:

CALL iceberg.system.rewrite_data_files('db', 'sample');
CALL iceberg.system.rewrite_data_files(schema => 'db', table_name => 'sample');
CALL iceberg.system.rewrite_data_files('db', 'sample', 'partition_key = 1');
CALL iceberg.system.rewrite_data_files(schema => 'db', table_name => 'sample', filter => 'partition_key = 1');

For all distributed procedures, the implementation of their base class defaults to including parameters schema, table_name and filter. Among them, schema and table_name are required parameters, which specify the target table to be processed by this distributed procedure. While filter is an optional parameter, which indicates the filtering conditions for the data to be processed.

The specific procedure implementations can expand the parameter list, but they must include the required parameters schema and table_name.

[hantangwangd]

Hi @aditi-pandit, I have supplemented the proposed syntax, examples, and the explanation of parameters in several places. Referring to:

https://github.com/prestodb/rfcs/pull/12/files#diff-7655a74167c417d09fba4cc48ec39232f65d7056a449e1cb09d51664f4fe1b7eR15-R20

https://github.com/prestodb/rfcs/pull/12/files#diff-7655a74167c417d09fba4cc48ec39232f65d7056a449e1cb09d51664f4fe1b7eR97

https://github.com/prestodb/rfcs/pull/12/files#diff-7655a74167c417d09fba4cc48ec39232f65d7056a449e1cb09d51664f4fe1b7eR107-R109

Please take a look when available, thanks!

[aditi-pandit]

Its interesting that this is rooted in a TableScanNode. Is this of a regular table or something else ?

[hantangwangd]

It's a regular table representing the target table. As mentioned above, the target table is specified through the parameters schema and table_name.

[aditi-pandit]

ConnectorPageSink will need a Prestissimo implementation as well.

[hantangwangd]

I'm not so familiar with Prestissimo implementation. Are we currently reusing the implementation of ConnectorPageSink from Hive in native worker? If so, yes maybe it's better to implement an Iceberg own ConnectorPageSink.

[aditi-pandit]

What will be the fields and members of ExecutionWriterTarget ?

TableWriter partitions and bucketizes input rows to files. This is a very specfic action whereas ExecutionWriterTarget is very generic. Are there specific actions like say a rewrite action or optimize action that we will implement in code that could be specified by ExecutionWriterTarget ? If yes, then we will need to implement the same logic in Native engine as well.

[hantangwangd]

The detailed description for ExecuteProcedureHandle (which is an implementation class of interface ExecutionWriterTarget) and it's members is in line 252 to line 273 (https://github.com/prestodb/rfcs/pull/12/files#diff-7655a74167c417d09fba4cc48ec39232f65d7056a449e1cb09d51664f4fe1b7eR252-R273) and line 357 to line 372 (https://github.com/prestodb/rfcs/pull/12/files#diff-7655a74167c417d09fba4cc48ec39232f65d7056a449e1cb09d51664f4fe1b7eR357-R372). The definitions of its fields and members are similar to InsertHandle.

Currently there is no specific logic that needs to be implemented through code in ExecuteProcedureHandle, so in order to support ExecuteProcedureHandle in native, we can basically refer to the implementation of InsertHandle. The only difference is that an additional field will appear in ExecuteProcedureHandle: QualifiedObjectName procedureName.

[aditi-pandit]

Its good to clarify if beginCallDistributedProcedure and finishCallDistributedProcedure will happen at the co-ordinator.

[hantangwangd]

OK, will add this.

[hantangwangd]

Done, added in https://github.com/prestodb/rfcs/pull/12/files#diff-7655a74167c417d09fba4cc48ec39232f65d7056a449e1cb09d51664f4fe1b7eR131-R132.

[aditi-pandit]

Does this procedure need to be dynamically loaded ? It might be worth thinking how you will do it in C++. Are you planning to use dlopen call ? That can come with lots of security implications.

[hantangwangd]

Firstly, in the current design, distributed procedures serve as Connector specific built-in system procedures, therefore they do not need to be dynamically loaded, but only need to be statically pre-defined and loaded. Dynamic loading is more like what need to be done when supporting UDF or UDP, which is beyond the scope of this proposal. And we can consider how to support it if necessary in the future.

Secondly, as discussed above, the methods of a DistributedProcedure will always be called in Java environment. Even when executing the final submission, the method FinishCallDistributedProcedure.finish(...) will be executed in operator TableFinishOperator in the Java worker inside coordinator as well. So it seems that there is currently no need to consider dynamic loading on C++ environment. Please let me know if there are any omissions or misunderstandings.

[yingsu00]

@hantangwangd Is there a working prototype or a PR now?

[yingsu00]

The operation may not necessarily update the real data files. For example, We may want to run CALL system.clean_data_cache('ALL_WORKERS'); to just call AsyncDataCache::shutdown() on all Presto C++ workers. What would the plan look like?

[yingsu00]

cc @agrawalreetika

[hantangwangd]

The operation may not necessarily update the real data files. For example, We may want to run CALL system.clean_data_cache('ALL_WORKERS'); to just call AsyncDataCache::shutdown() on all Presto C++ workers. What would the plan look like?

Thanks for the question, @yingsu00. When we need to realize the functionality you mentioned, I think we can extend the subtypes of DistributedProcedure to, for example, DistributedDataProcedure and DistributedMetadataProcedure. For the type of DistributedMetadataProcedure, we can analyze and plan it to a logical plan with the following shape:

OutputNode <- ProcedureFinishNode <- CallDistributedProcedureNode

Wherein, ProcedureFinishNode is a newly added plan node. And after being segmented, the plan only needs to include two stages. Do you think it's reasonable?

[hantangwangd]

Is there a working prototype or a PR now?

Yes, there is already a PR, #22659. Will rebase it to the newest commit list as soon as possible.