feat: Distributed Procedure Support Part 2/X - iceberg part changes #26374
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Sorry @hantangwangd, your pull request is larger than the review limit of 150000 diff characters
hantangwangd
force-pushed
the
support_call_distributed_procedure_part2
branch
2 times, most recently
from
e3f4e26 to
a6a6101
Compare
hantangwangd
force-pushed
the
support_call_distributed_procedure_part2
branch
3 times, most recently
from
7b97da0 to
7ebf2a0
Compare
hantangwangd
force-pushed
the
support_call_distributed_procedure_part2
branch
2 times, most recently
from
d258f2f to
30f74af
Compare
hantangwangd
requested review from
a team,
ClarenceThreepwood,
ZacBlanco,
elharo,
feilong-liu,
jaystarshot and
steveburnett
as code owners
Contributor
Reviewer's GuideThis PR refactors the Iceberg connector to enable distributed procedure execution by introducing a procedure context framework, adding a new serialized handle type, extending metadata to manage procedure lifecycles, implementing the rewrite_data_files procedure and its split source, wiring ProcedureRegistry into the connector, and providing extensive tests. Sequence diagram for distributed procedure lifecycle in Iceberg connectorsequenceDiagram
participant "Coordinator (Presto Engine)"
participant "IcebergAbstractMetadata"
participant "ProcedureRegistry"
participant "DistributedProcedure (RewriteDataFilesProcedure)"
participant "IcebergProcedureContext"
participant "IcebergSplitManager"
participant "CallDistributedProcedureSplitSource"
participant "IcebergPageSinkProvider"
participant "IcebergPageSink"
participant "IcebergTable"
"Coordinator (Presto Engine)"->>"IcebergAbstractMetadata": beginCallDistributedProcedure(...)
"IcebergAbstractMetadata"->>"ProcedureRegistry": resolve(procedureName)
"ProcedureRegistry"-->>"IcebergAbstractMetadata": DistributedProcedure instance
"IcebergAbstractMetadata"->>"DistributedProcedure": createContext()
"DistributedProcedure"-->>"IcebergAbstractMetadata": IcebergProcedureContext
"IcebergAbstractMetadata"->>"IcebergProcedureContext": setTable(Table)
"IcebergAbstractMetadata"->>"IcebergProcedureContext": setTransaction(Transaction)
"IcebergAbstractMetadata"->>"DistributedProcedure": begin(...)
"DistributedProcedure"->>"IcebergProcedureContext": setConnectorSplitSource(CallDistributedProcedureSplitSource)
"IcebergAbstractMetadata"-->>"Coordinator (Presto Engine)": IcebergDistributedProcedureHandle
"Coordinator (Presto Engine)"->>"IcebergSplitManager": getSplits(...)
"IcebergSplitManager"->>"IcebergAbstractMetadata": getSplitSourceInCurrentCallProcedureTransaction()
"IcebergAbstractMetadata"-->>"IcebergSplitManager": CallDistributedProcedureSplitSource
"IcebergSplitManager"-->>"Coordinator (Presto Engine)": splits
"Coordinator (Presto Engine)"->>"IcebergPageSinkProvider": createPageSink(..., IcebergDistributedProcedureHandle)
"IcebergPageSinkProvider"->>"IcebergPageSink": createPageSink(...)
"Coordinator (Presto Engine)"->>"IcebergAbstractMetadata": finishCallDistributedProcedure(...)
"IcebergAbstractMetadata"->>"ProcedureRegistry": resolve(procedureName)
"ProcedureRegistry"-->>"IcebergAbstractMetadata": DistributedProcedure instance
"IcebergAbstractMetadata"->>"DistributedProcedure": finish(...)
"DistributedProcedure"->>"IcebergProcedureContext": collect scanned files, commit new files
"IcebergAbstractMetadata"->>"IcebergTable": commitTransaction()
"IcebergAbstractMetadata"->>"IcebergProcedureContext": destroy()
"IcebergAbstractMetadata"-->>"Coordinator (Presto Engine)": procedure finished
ER diagram for new IcebergDistributedProcedureHandle data typeerDiagram
ICEBERG_DISTRIBUTED_PROCEDURE_HANDLE {
String schemaName
IcebergTableName tableName
PrestoIcebergSchema schema
PrestoIcebergPartitionSpec partitionSpec
IcebergColumnHandle inputColumns
String outputPath
FileFormat fileFormat
HiveCompressionCodec compressionCodec
Map storageProperties
}
ICEBERG_DISTRIBUTED_PROCEDURE_HANDLE ||--o| ICEBERG_TABLE_NAME : "tableName"
ICEBERG_DISTRIBUTED_PROCEDURE_HANDLE ||--o| PRESTO_ICEBERG_SCHEMA : "schema"
ICEBERG_DISTRIBUTED_PROCEDURE_HANDLE ||--o| PRESTO_ICEBERG_PARTITION_SPEC : "partitionSpec"
ICEBERG_DISTRIBUTED_PROCEDURE_HANDLE ||--|{ ICEBERG_COLUMN_HANDLE : "inputColumns"
ICEBERG_DISTRIBUTED_PROCEDURE_HANDLE ||--o| FILE_FORMAT : "fileFormat"
ICEBERG_DISTRIBUTED_PROCEDURE_HANDLE ||--o| HIVE_COMPRESSION_CODEC : "compressionCodec"
Class diagram for new and updated Iceberg distributed procedure typesclassDiagram
class IcebergProcedureContext {
+Set<DataFile> scannedDataFiles
+Set<DeleteFile> fullyAppliedDeleteFiles
+Map<String, Object> relevantData
+Optional<Table> table
+Transaction transaction
+Optional<ConnectorSplitSource> connectorSplitSource
+setTable(Table table)
+setTransaction(Transaction transaction)
+getTable()
+getTransaction()
+setConnectorSplitSource(ConnectorSplitSource splitSource)
+getConnectorSplitSource()
+getScannedDataFiles()
+getFullyAppliedDeleteFiles()
+getRelevantData()
+destroy()
}
class IcebergDistributedProcedureHandle {
+String schemaName
+IcebergTableName tableName
+PrestoIcebergSchema schema
+PrestoIcebergPartitionSpec partitionSpec
+List<IcebergColumnHandle> inputColumns
+String outputPath
+FileFormat fileFormat
+HiveCompressionCodec compressionCodec
+Map<String, String> storageProperties
+IcebergDistributedProcedureHandle(...)
}
class IcebergWritableTableHandle {
}
IcebergDistributedProcedureHandle --|> IcebergWritableTableHandle
IcebergDistributedProcedureHandle ..|> ConnectorDistributedProcedureHandle
class CallDistributedProcedureSplitSource {
-CloseableIterator<FileScanTask> fileScanTaskIterator
-Optional<Consumer<FileScanTask>> fileScanTaskConsumer
-TableScan tableScan
-Closer closer
-double minimumAssignedSplitWeight
-ConnectorSession session
+getNextBatch(...)
+isFinished()
+close()
-toIcebergSplit(FileScanTask task)
}
class RewriteDataFilesProcedure {
+TypeManager typeManager
+JsonCodec<CommitTaskData> commitTaskCodec
+RewriteDataFilesProcedure(...)
+get()
-beginCallDistributedProcedure(...)
-finishCallDistributedProcedure(...)
}
IcebergProcedureContext ..|> ConnectorProcedureContext
CallDistributedProcedureSplitSource ..|> ConnectorSplitSource
RewriteDataFilesProcedure ..|> Provider
RewriteDataFilesProcedure ..|> DistributedProcedure
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Contributor
There was a problem hiding this comment.
Hey there - I've reviewed your changes - here's some feedback:
- This PR is very large and touches many layers (protocol, metadata, split manager, planner, C++ bindings, and extensive tests); consider splitting into smaller, focused PRs to simplify review and isolate potential regressions.
- There’s a lot of duplicated test setup/assertion code in TestRewriteDataFilesProcedure (and related tests); factor out common helpers for table creation, file‐count assertions, and cleanup to reduce maintenance overhead.
- In IcebergProcedureContext.destroy, you clear splits and file sets but don’t reset the 'table' or 'transaction' fields—consider clearing those as well to fully release resources after procedure completion.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- This PR is very large and touches many layers (protocol, metadata, split manager, planner, C++ bindings, and extensive tests); consider splitting into smaller, focused PRs to simplify review and isolate potential regressions.
- There’s a lot of duplicated test setup/assertion code in TestRewriteDataFilesProcedure (and related tests); factor out common helpers for table creation, file‐count assertions, and cleanup to reduce maintenance overhead.
- In IcebergProcedureContext.destroy, you clear splits and file sets but don’t reset the 'table' or 'transaction' fields—consider clearing those as well to fully release resources after procedure completion.
## Individual Comments
### Comment 1
<location> `presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergProcedureContext.java:92-93` </location>
<code_context>
+ this.relevantData.clear();
+ this.scannedDataFiles.clear();
+ this.fullyAppliedDeleteFiles.clear();
+ this.connectorSplitSource.ifPresent(ConnectorSplitSource::close);
+ this.connectorSplitSource = null;
+ }
+}
</code_context>
<issue_to_address>
**issue (bug_risk):** Setting connectorSplitSource to null may lead to NullPointerExceptions.
Assigning null to connectorSplitSource, which is an Optional, breaks expected usage and may cause runtime errors. Use Optional.empty() instead to prevent NullPointerExceptions.
</issue_to_address>
### Comment 2
<location> `presto-iceberg/src/main/java/com/facebook/presto/iceberg/RewriteDataFilesProcedure.java:167-170` </location>
<code_context>
+ .map(slice -> commitTaskCodec.fromJson(slice.getBytes()))
+ .collect(toImmutableList());
+
+ org.apache.iceberg.types.Type[] partitionColumnTypes = icebergTable.spec().fields().stream()
+ .map(field -> field.transform().getResultType(
+ icebergTable.schema().findType(field.sourceId())))
+ .toArray(Type[]::new);
+
+ Set<DataFile> newFiles = new HashSet<>();
</code_context>
<issue_to_address>
**issue (bug_risk):** Potential mismatch between partition spec fields and schema types.
If findType returns null for a missing sourceId, this could lead to runtime errors. Please add validation or error handling for cases where the type is not found.
</issue_to_address>
### Comment 3
<location> `presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergAbstractMetadata.java:1074-1083` </location>
<code_context>
+ throw new PrestoException(NOT_SUPPORTED, "This connector do not allow table execute at specified snapshot");
+ }
+
+ transaction = icebergTable.newTransaction();
+ BaseProcedure<?> procedure = procedureRegistry.resolve(
+ new ConnectorId(procedureName.getCatalogName()),
+ new SchemaTableName(
+ procedureName.getSchemaName(),
+ procedureName.getObjectName()));
+ verify(procedure instanceof DistributedProcedure, "procedure must be DistributedProcedure");
+ procedureContext = Optional.of((IcebergProcedureContext) ((DistributedProcedure) procedure).createContext());
+ procedureContext.get().setTable(icebergTable);
+ procedureContext.get().setTransaction(transaction);
+ return ((DistributedProcedure) procedure).begin(session, procedureContext.get(), tableLayoutHandle, arguments);
+ }
</code_context>
<issue_to_address>
**issue (bug_risk):** Transaction is assigned to a field but not cleared after procedure completion.
Since the transaction field remains set after finishCallDistributedProcedure, running multiple procedures may result in stale or incorrect state. Please ensure the transaction field is cleared or properly scoped after each procedure completes.
</issue_to_address>
### Comment 4
<location> `presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergAbstractMetadata.java:1099` </location>
<code_context>
+ verify(procedureContext.isPresent(), "procedure context must be present");
+ ((DistributedProcedure) procedure).finish(procedureContext.get(), procedureHandle, fragments);
+ transaction.commitTransaction();
+ procedureContext.get().destroy();
+ }
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Destroying procedureContext does not reset the Optional field.
Reset procedureContext to Optional.empty() after destroy to prevent unexpected behavior if accessed post-completion.
</issue_to_address>
### Comment 5
<location> `presto-docs/src/main/sphinx/connector/iceberg.rst:1239-1242` </location>
<code_context>
+Rewrite Data Files
+^^^^^^^^^^^^^^^^^^
+
+Iceberg tracks all data files under different partition specs in a table. More data files requires
+more metadata to be stored in manifest files, and small data files can cause unnecessary amount metadata and
+less efficient queries from file open costs. Also, data files under different partition specs can
+prevent metadata level deletion or thorough predicate push down for Presto.
</code_context>
<issue_to_address>
**issue (typo):** Correct verb agreement and missing word in sentence.
The correct sentence is: 'More data files require more metadata to be stored in manifest files, and small data files can cause an unnecessary amount of metadata and less efficient queries due to file open costs.'
```suggestion
Iceberg tracks all data files under different partition specs in a table. More data files require
more metadata to be stored in manifest files, and small data files can cause an unnecessary amount of metadata and
less efficient queries due to file open costs. Also, data files under different partition specs can
prevent metadata level deletion or thorough predicate push down for Presto.
```
</issue_to_address>
### Comment 6
<location> `presto-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp:745` </location>
<code_context>
} // namespace facebook::presto::protocol::iceberg
namespace facebook::presto::protocol::iceberg {
+IcebergDistributedProcedureHandle::
+ IcebergDistributedProcedureHandle() noexcept {
+ _type = "hive-iceberg";
+}
</code_context>
<issue_to_address>
**issue (review_instructions):** Member variable '_type' uses leading underscore, but should use camelCase_ for private/protected members.
The member variable '_type' does not follow the required camelCase_ convention for private/protected members. Please rename it to 'type_' to comply with the coding standard.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `presto-native-execution/**/*.hpp,presto-native-execution/**/*.hpp,presto-native-execution/**/*.cpp`
**Instructions:**
Use camelCase_ for private and protected members variables.
</details>
</issue_to_address>
### Comment 7
<location> `presto-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp:746` </location>
<code_context>
namespace facebook::presto::protocol::iceberg {
+IcebergDistributedProcedureHandle::
+ IcebergDistributedProcedureHandle() noexcept {
+ _type = "hive-iceberg";
+}
+
</code_context>
<issue_to_address>
**issue (review_instructions):** Member variable '_type' should use camelCase_ (e.g., 'type_') for private/protected members.
Please update '_type' to 'type_' to match the required naming convention for private/protected member variables.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `presto-native-execution/**/*.hpp,presto-native-execution/**/*.hpp,presto-native-execution/**/*.cpp`
**Instructions:**
Use camelCase_ for private and protected members variables.
</details>
</issue_to_address>
### Comment 8
<location> `presto-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp:818` </location>
<code_context>
+}
+
+void from_json(const json& j, IcebergDistributedProcedureHandle& p) {
+ p._type = j["@type"];
+ from_json_key(
+ j,
</code_context>
<issue_to_address>
**issue (review_instructions):** Member variable '_type' should use camelCase_ (e.g., 'type_') for private/protected members.
Please update '_type' to 'type_' to match the required naming convention for private/protected member variables.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `presto-native-execution/**/*.hpp,presto-native-execution/**/*.hpp,presto-native-execution/**/*.cpp`
**Instructions:**
Use camelCase_ for private and protected members variables.
</details>
</issue_to_address>
### Comment 9
<location> `presto-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp:745` </location>
<code_context>
} // namespace facebook::presto::protocol::iceberg
namespace facebook::presto::protocol::iceberg {
+IcebergDistributedProcedureHandle::
+ IcebergDistributedProcedureHandle() noexcept {
+ _type = "hive-iceberg";
+}
</code_context>
<issue_to_address>
**issue (review_instructions):** The member variable '_type' uses a leading underscore, which is not camelCase_ as required for private/protected members.
Private/protected member variables should use camelCase_ (e.g., 'type_') rather than a leading underscore. Please rename '_type' to 'type_' for consistency with the coding standard.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `presto-native-execution/**/*.hpp,presto-native-execution/**/*.cpp`
**Instructions:**
Use camelCase_ for private and protected members variables.
</details>
</issue_to_address>
### Comment 10
<location> `presto-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp:746` </location>
<code_context>
namespace facebook::presto::protocol::iceberg {
+IcebergDistributedProcedureHandle::
+ IcebergDistributedProcedureHandle() noexcept {
+ _type = "hive-iceberg";
+}
+
</code_context>
<issue_to_address>
**issue (review_instructions):** The member variable '_type' uses a leading underscore, which is not camelCase_ as required for private/protected members.
Please rename '_type' to 'type_' to follow the camelCase_ convention for private/protected member variables.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `presto-native-execution/**/*.hpp,presto-native-execution/**/*.cpp`
**Instructions:**
Use camelCase_ for private and protected members variables.
</details>
</issue_to_address>
### Comment 11
<location> `presto-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp:818` </location>
<code_context>
+}
+
+void from_json(const json& j, IcebergDistributedProcedureHandle& p) {
+ p._type = j["@type"];
+ from_json_key(
+ j,
</code_context>
<issue_to_address>
**issue (review_instructions):** The member variable '_type' uses a leading underscore, which is not camelCase_ as required for private/protected members.
Please update '_type' to 'type_' to comply with the camelCase_ convention for private/protected member variables.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `presto-native-execution/**/*.hpp,presto-native-execution/**/*.cpp`
**Instructions:**
Use camelCase_ for private and protected members variables.
</details>
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergProcedureContext.java
Outdated
Show resolved
Hide resolved
presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergAbstractMetadata.java
Show resolved
Hide resolved
presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergAbstractMetadata.java
Show resolved
Hide resolved
...to-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp
Show resolved
Hide resolved
...to-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp
Show resolved
Hide resolved
...to-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp
Show resolved
Hide resolved
...to-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp
Show resolved
Hide resolved
steveburnett
requested changes
Nov 17, 2025
Contributor
steveburnett
left a comment
steveburnett
left a comment
There was a problem hiding this comment.
Thanks for the doc! Just a nit of formatting.
hantangwangd
force-pushed
the
support_call_distributed_procedure_part2
branch
from
30f74af to
9959e0c
Compare
Member
Author
|
@steveburnett thanks for the review, fixed! Please take a look when you have a minute. |
steveburnett
approved these changes
Nov 18, 2025
Contributor
steveburnett
left a comment
steveburnett
left a comment
There was a problem hiding this comment.
LGTM! (docs)
Pull updated branch, new local doc build. Looks good, thanks!
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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:
Re-factor Iceberg connector to support
call distributed procedure. Introduce Iceberg's procedure context and expandIcebergSplitManagerto support split source planned byIcebergAbstractMetadata.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_filesprocedure. It build a customized split source, set the split source to procedure context in order to be used inIcebergSplitManager. 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'sRewriteFilestransaction.Motivation and Context
prestodb/rfcs#12
Impact
N/A
Test Plan
rewrite_data_filesContributor checklist
Release Notes