feat: Expand procedure architecture for distributed execution, and support iceberg procedure `rewrite_data_files` by hantangwangd · Pull Request #22659 · prestodb/presto

hantangwangd · 2024-05-03T11:23:36Z

Description

This PR expand the current procedure architecture in presto, support defining, registering and calling procedures which need to be executed in a distributed way. Then support distributed procedure in Iceberg connector and implement a specific procedure rewrite_data_files for it.

Referring to: prestodb/rfcs#12

The whole PR is separated into 6 parts:

Re-factor ProcedureRegistry/Procedure data structure to support the creation and register of DistributedProcedure. And make sure ProcedureRegistry be available in presto-analyzer module, so that we can recognize distributed procedures in call statement during prepare and analyze stages.
Handle call statement on distributed procedures in preparer stage. In this stage, we figure out the procedure's type in call statement, and define a new query type CALL_DISTRIBUTED_PROCEDURE for call distributed procedure in BuiltInPreparedQuery. In this way, call distributed procedure statement would be handled by SqlQueryExecutionFactory, then be created and handled as a SqlQueryExecution.
Analyze and plan the call distributed procedure statement based on the subtype of the distributed procedure. For subtype TableDataRewriteDistributedProcedure, ultimately generate a logical plan for it as follows:

OutputNode <- TableFinishNode <- CallDistributedProcedureNode <- FilterNode <- TableScanNode

Optimize, segmentation, grouped tag and local plan for the logical plan generated above. The handle logical for CallDistributedProcedureNode is similar as TableWriterNode. Besides, a new optimizer RewriteWriterTarget is added, which is placed after all optimization rules. It is used to update the TableHandle held in TableFinishNode and CallDistributedProcedureNode based on the underlying TableScanNode after the entire optimization is completed, considering the possible filter pushing down.
Re-factor Iceberg connector to support call distributed procedure. Introduce Iceberg's procedure context and expand IcebergSplitManager to support split source planned by IcebergAbstractMetadata.beginCallDistributedProcedure(...). This split source will be set to procedure context, and use procedure context to hold all the files to be rewritten as well.
Support Iceberg rewrite_data_files procedure. It build a customized split source, set the split source to procedure context in order to be used in IcebergSplitManager. And register a file scan task consumer to collector and hold all the scanned files into procedure context. Then finally in the commit stage, get all the data files and delete files that has been rewritten, and all the files that has been newly generated, change and commit their metadata through Iceberg table's RewriteFiles transaction.

Motivation and Context

N/A

Impact

N/A

Test Plan

Add test cases in each phase involving the procedure architecture expansion, including creating and registering for distributed procedures, preparing for call distributed procedure, analyzing for call distributed procedure, logical planning and optimizing for call distributed procedure, and finally add tests for the specific iceberg distributed procedure: rewrite_data_files

Contributor checklist

Please make sure your submission complies with our development, formatting, commit message, and attribution guidelines.
PR description addresses the issue accurately and concisely. If the change is non-trivial, a GitHub Issue is referenced.
Documented new properties (with its default value), SQL syntax, functions, or other functionality.
If release notes are required, they follow the release notes guidelines.
Adequate tests were added if applicable.
CI passed.

Release Notes

== NO RELEASE NOTE ==

steveburnett

Thanks for the draft doc! Some nits about punctuation, formatting, and some suggested rephrasing for readability and conciseness, but the content looks good.

presto-docs/src/main/sphinx/connector/iceberg.rst

hantangwangd · 2024-06-13T18:52:03Z

@steveburnett Thanks a lot for your suggestion, all be fixed. Please take a look when convenient!

steveburnett

LGTM! (docs)

Pull updated branch, new local doc build, looks good. Thanks!

hantangwangd · 2025-10-17T16:21:46Z

@tdcmeehan Thanks for the review. Sure, I'll add the relevant documentation as soon as possible.

tdcmeehan

Very good work. Well done! I've left some feedback, but it's mostly minor.

I would split this PR into at least 3 parts:

All of the code in core Presto to support distributed prcoedures
The C++ counterpart for this code
The Iceberg integration

tdcmeehan · 2025-10-17T20:07:47Z

presto-spi/src/main/java/com/facebook/presto/spi/procedure/DistributedProcedure.java

+    private final BeginCallDistributedProcedure beginCallDistributedProcedure;
+    private final FinishCallDistributedProcedure finishCallDistributedProcedure;
+
+    protected DistributedProcedure(DistributedProcedureType type, String schema, String name, List<Argument> arguments, BeginCallDistributedProcedure beginCallDistributedProcedure, FinishCallDistributedProcedure finishCallDistributedProcedure)


I think it would be easier to read and more idiomatic to make DistributedProcedure abstract, and make beginCallDistributedProcedure and finishCallDistributedProcedure abstract methods.

Great idea, I completely agree! Initially, I didn't declare DistributedProcedure as abstract because Procedure itself wasn't declared as abstract. Now, I've made both Procedure and DistributedProcedure abstract, and introduced a LocalProcedure to represent the original coordinator-only procedures. This makes the overall procedure architecture much clearer and easier to understand.

presto-main-base/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java

presto-main-base/src/main/java/com/facebook/presto/execution/CallTask.java

presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/SymbolMapper.java

tdcmeehan · 2025-10-17T20:41:23Z

presto-main-base/src/main/java/com/facebook/presto/sql/planner/sanity/PlanChecker.java

                        new DynamicFiltersChecker(),
-                        new WarnOnScanWithoutPartitionPredicate(featuresConfig));
+                        new WarnOnScanWithoutPartitionPredicate(featuresConfig),
+                        new CallDistributedProcedureValidator());


@hantangwangd it would be nice to have plan tests, like TestHashGenerationOptimizer, that show the type of plan that gets generated by a distributed procedure.

Thanks for the suggestion. Since the CALL DISTRIBUTED PROCEDURE statement requires a valid distributed procedure to be invoked, and currently only one has been implemented in Iceberg connector, I've added the test case to TestIcebergLogicalPlanner. Please take a look when you have time, thanks a lot.

presto-main-base/src/main/java/com/facebook/presto/testing/TestProcedureRegistry.java

presto-spi/src/main/java/com/facebook/presto/spi/procedure/IProcedureRegistry.java

tdcmeehan · 2025-10-17T21:11:16Z

...in-base/src/main/java/com/facebook/presto/sql/planner/plan/CallDistributedProcedureNode.java

+        return source;
+    }
+
+    @JsonIgnore


Is this intentionally ignored?

Yes, this is intentionally ignored. Subclasses of WriteTarget are only used during planning -- they will not be serialized.

presto-spi/src/main/java/com/facebook/presto/spi/procedure/DistributedProcedure.java

hantangwangd · 2025-10-20T01:50:52Z

@tdcmeehan thanks for your review and feedback. I've addressed all your comments except the one about adding documentation. Please take a look when you have time.

Very good work. Well done! I've left some feedback, but it's mostly minor.

I would split this PR into at least 3 parts:
1. All of the code in core Presto to support distributed prcoedures

2. The C++ counterpart for this code

3. The Iceberg integration

Are you suggesting that I split this into three separate PRs, or should I squash it into three commits within a single PR?

tdcmeehan · 2025-10-20T01:58:37Z

@hantangwangd since we now squash commits on merge, let's make 3 separate PRs.

hantangwangd · 2025-10-20T02:44:44Z

@tdcmeehan Sure, I'll do it.

…ailable in analyzer

…buted procedure

Use a subclass `TableDataRewriteDistributedProcedure` for table rewrite tasks, for example, merge small data files, sort table data, repartition table data etc.

Accordingly rename previous ProcedureRegistry to BuiltInProcedureRegistry

…dule

… abstract classes And introduce a new class `LocalProcedure` to represent the former coordinator-only procedures

…dure

…#26373) ## Description This PR is the first part of many PRs to support distributed procedure into Presto. It is a split of the original entire PR which is located here: #22659. The whole work in this PR includes the following parts: 1. Re-factor `ProcedureRegistry`/`Procedure` data structure to support the creation and register of `DistributedProcedure`. And make sure `ProcedureRegistry` be available in `presto-analyzer` module and connectors, so that we can recognize distributed procedures in call statement during prepare analyze stages. 2. Handle call statement on distributed procedures in preparer stage. In this stage, we figure out the procedure's type in call statement, and define a new query type `CALL_DISTRIBUTED_PROCEDURE` for `call distributed procedure` in `BuiltInPreparedQuery`. In this way, `call distributed procedure` statement would be handled by `SqlQueryExecutionFactory`, then be created and handled as a `SqlQueryExecution`. 3. Analyze and plan the `call distributed procedure` statement based on the subtype of the distributed procedure. For subtype `TableDataRewriteDistributedProcedure`, ultimately generate a logical plan for it as follows: ``` OutputNode <- TableFinishNode <- CallDistributedProcedureNode <- FilterNode <- TableScanNode ``` 4. Optimize, segmentation, grouped tag and local plan for the logical plan generated above. The handle logical for `CallDistributedProcedureNode` is similar as `TableWriterNode`. Besides, a new optimizer `RewriteWriterTarget` is added, which is placed after all optimization rules. It is used to update the `TableHandle` held in `TableFinishNode` and `CallDistributedProcedureNode` based on the underlying `TableScanNode` after the entire optimization is completed, considering the possible filter pushing down. ## Motivation and Context prestodb/rfcs#12 ## Impact N/A ## Test Plan - Add test cases in each phase involving the procedure architecture expansion, including creating and registering for distributed procedures, preparing for call distributed procedure, analyzing for call distributed procedure, logical planning and optimizing for call distributed procedure ## Contributor checklist - [x] Please make sure your submission complies with our [contributing guide](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md), in particular [code style](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#code-style) and [commit standards](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#commit-standards). - [x] PR description addresses the issue accurately and concisely. If the change is non-trivial, a GitHub Issue is referenced. - [ ] Documented new properties (with its default value), SQL syntax, functions, or other functionality. - [x] If release notes are required, they follow the [release notes guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines). - [x] Adequate tests were added if applicable. - [x] CI passed. - [ ] If adding new dependencies, verified they have an [OpenSSF Scorecard](https://securityscorecards.dev/#the-checks) score of 5.0 or higher (or obtained explicit TSC approval for lower scores). ## Release Notes ``` == RELEASE NOTES == General Changes * Upgrade the procedure architecture to support distributed executing procedures ```

## Description This PR is the second part of many PRs to support distributed procedure into Presto. It is a split of the original entire PR which is located here: #22659. The whole work in this PR includes the following parts: 1. Re-factor Iceberg connector to support `call distributed procedure`. Introduce Iceberg's procedure context and expand `IcebergSplitManager` to support split source planned by `IcebergAbstractMetadata.beginCallDistributedProcedure(...)`. This split source will be set to procedure context, and use procedure context to hold all the files to be rewritten as well. 2. Support Iceberg `rewrite_data_files` procedure. It build a customized split source, set the split source to procedure context in order to be used in `IcebergSplitManager`. And register a file scan task consumer to collector and hold all the scanned files into procedure context. Then finally in the commit stage, get all the data files and delete files that has been rewritten, and all the files that has been newly generated, change and commit their metadata through Iceberg table's `RewriteFiles` transaction. ## Motivation and Context prestodb/rfcs#12 ## Impact N/A ## Test Plan - Add test cases for validating the result and plan tree shape of iceberg specific distributed procedure: `rewrite_data_files` ## Contributor checklist - [ ] Please make sure your submission complies with our [contributing guide](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md), in particular [code style](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#code-style) and [commit standards](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#commit-standards). - [ ] PR description addresses the issue accurately and concisely. If the change is non-trivial, a GitHub Issue is referenced. - [ ] Documented new properties (with its default value), SQL syntax, functions, or other functionality. - [ ] If release notes are required, they follow the [release notes guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines). - [ ] Adequate tests were added if applicable. - [ ] CI passed. - [ ] If adding new dependencies, verified they have an [OpenSSF Scorecard](https://securityscorecards.dev/#the-checks) score of 5.0 or higher (or obtained explicit TSC approval for lower scores). ## Release Notes ``` == NO RELEASE NOTE == ```

hantangwangd · 2025-12-12T09:51:09Z

Closing this PR in favor of #26788.

hantangwangd requested review from a team, jaystarshot and shrinidhijoshi as code owners May 3, 2024 11:23

hantangwangd requested a review from presto-oss May 3, 2024 11:23

hantangwangd marked this pull request as draft May 3, 2024 11:24

hantangwangd force-pushed the support_call_distributed_procedure branch 4 times, most recently from 7ec819c to 9440737 Compare May 8, 2024 08:36

hantangwangd force-pushed the support_call_distributed_procedure branch 4 times, most recently from f89dc40 to e796fa2 Compare May 13, 2024 12:56

hantangwangd force-pushed the support_call_distributed_procedure branch 5 times, most recently from acb0351 to c3eaa96 Compare May 24, 2024 19:59

hantangwangd force-pushed the support_call_distributed_procedure branch 3 times, most recently from 05de3c8 to 0dc3dbb Compare June 11, 2024 11:08

hantangwangd changed the title ~~[ForTest]Expand procedure architecture for distributed execution, and support iceberg procedure rewrite_data_files~~ [WIP]Expand procedure architecture for distributed execution, and support iceberg procedure rewrite_data_files Jun 11, 2024

steveburnett requested changes Jun 13, 2024

View reviewed changes

hantangwangd force-pushed the support_call_distributed_procedure branch from 0dc3dbb to a78c41c Compare June 13, 2024 18:47

steveburnett previously approved these changes Jun 13, 2024

View reviewed changes

hantangwangd dismissed steveburnett’s stale review via 2fdbab7 July 13, 2024 03:17

hantangwangd force-pushed the support_call_distributed_procedure branch from a78c41c to 2fdbab7 Compare July 13, 2024 03:17

hantangwangd force-pushed the support_call_distributed_procedure branch from 2fdbab7 to befe9a7 Compare July 31, 2024 18:17

hantangwangd marked this pull request as ready for review July 31, 2024 19:28

tdcmeehan requested changes Oct 17, 2025

View reviewed changes

hantangwangd force-pushed the support_call_distributed_procedure branch from 8d53e4d to 106c2aa Compare October 19, 2025 15:31

hantangwangd added 16 commits October 20, 2025 21:16

Refactor ProcedureRegistry to support distributed procedure and be av…

278e5f0

…ailable in analyzer

Support and handle call distributed procedure statement in preparer

987118e

Analyze and plan for call distributed procedure statement

bc67dbd

Execute optimization, segmentation and local planning for call distri…

d7574eb

…buted procedure

Refactor Iceberg connector to support call distributed procedure

2a243dd

Support Iceberg procedure rewrite_data_files

2cff9a9

Fix mixed case test failures

cbc0edf

Fix test failure caused by concurrent operations on the same table

bef5763

Extend DistributedProcedure hierarchy for more extension types

f0729ec

Use a subclass `TableDataRewriteDistributedProcedure` for table rewrite tasks, for example, merge small data files, sort table data, repartition table data etc.

[native] Update presto protocol for distributed procedure

cf12a2f

[native] Prepare for the actual support of distributed procedures

eaa82da

[Address Comment] Rename IProcedureRegistry to ProcedureRegistry

73f4367

Accordingly rename previous ProcedureRegistry to BuiltInProcedureRegistry

[Address Comment] Move TestProcedureRegistry out of the presto-spi mo…

e24d0d8

…dule

[Address Comment] Fix little things and typos

6f1c34b

[Address Comment]Refactor Procedure and DistributedProcedure into…

9f4fae8

… abstract classes And introduce a new class `LocalProcedure` to represent the former coordinator-only procedures

[Address Comment] Add logical planner test for call distributed proce…

a192c74

…dure

hantangwangd force-pushed the support_call_distributed_procedure branch from 106c2aa to a192c74 Compare October 20, 2025 13:39

This was referenced Oct 21, 2025

feat: Distributed Procedure Support Part 1/X - core code base changes #26373

Merged

feat(plugin-iceberg): Add rewrite_data_files procedure #26374

Merged

feat: Distributed Procedure Support - prestissimo native changes #26375

Draft

hantangwangd closed this Dec 12, 2025

Conversation

hantangwangd commented May 3, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Motivation and Context

Impact

Test Plan

Contributor checklist

Release Notes

Uh oh!

steveburnett left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

hantangwangd commented Jun 13, 2024

Uh oh!

steveburnett left a comment

Choose a reason for hiding this comment

Uh oh!

hantangwangd commented Oct 17, 2025

Uh oh!

tdcmeehan left a comment

Choose a reason for hiding this comment

Uh oh!

tdcmeehan Oct 17, 2025

Choose a reason for hiding this comment

Uh oh!

hantangwangd Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

tdcmeehan Oct 17, 2025

Choose a reason for hiding this comment

Uh oh!

hantangwangd Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

tdcmeehan Oct 17, 2025

Choose a reason for hiding this comment

Uh oh!

hantangwangd Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

hantangwangd commented Oct 20, 2025

Uh oh!

tdcmeehan commented Oct 20, 2025

Uh oh!

hantangwangd commented Oct 20, 2025

Uh oh!

hantangwangd commented Dec 12, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

hantangwangd commented May 3, 2024 •

edited

Loading