Export partition to apache iceberg#1618
Open
arthurpassos wants to merge 49 commits intoantalya-26.1from
Open
Conversation
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5b0e833565
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| { | ||
| auto * object_storage = dynamic_cast<StorageObjectStorageCluster *>(dest_storage.get()); | ||
|
|
||
| auto * iceberg_metadata = dynamic_cast<IcebergMetadata *>(object_storage->getExternalMetadata(query_context)); |
There was a problem hiding this comment.
IcebergMetadata is defined only under #if USE_AVRO in IcebergMetadata.h, but this new dynamic_cast<IcebergMetadata *> is compiled unconditionally. In non-AVRO builds (USE_AVRO=0), this translation unit (and the same pattern added in StorageReplicatedMergeTree.cpp) cannot compile, so the change breaks optional-AVRO build configurations.
Useful? React with 👍 / 👎.
Comment on lines
+1780
to
+1781
| const String sidecar_path = replaceFileExtensionWithAvro( | ||
| filename_generator.convertMetadataPathToStoragePath(path)); |
There was a problem hiding this comment.
The export path list is populated from filename.path_in_storage, but commit now treats each entry as a metadata path and calls convertMetadataPathToStoragePath before reading sidecars. With write_full_path_in_iceberg_metadata=1, table_dir is an URI prefix (for example s3://...) while these entries are plain storage paths (/...), so the conversion throws and EXPORT PARTITION cannot commit.
Useful? React with 👍 / 👎.
Collaborator
Author
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 6c6194814d
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Collaborator
Author
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 763dec1c0a
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Comment on lines
+6535
to
+6537
| auto * object_storage = dynamic_cast<StorageObjectStorageCluster *>(dest_storage.get()); | ||
|
|
||
| auto * iceberg_metadata = dynamic_cast<IcebergMetadata *>(object_storage->getExternalMetadata(query_context)); |
There was a problem hiding this comment.
When EXPORT PART targets a regular Iceberg table (engine backed by StorageObjectStorage, not StorageObjectStorageCluster), this dynamic_cast returns nullptr and object_storage->getExternalMetadata(...) dereferences it. That turns a user-facing validation path into a null-pointer crash instead of an error, and it is reachable because this branch runs whenever iceberg_metadata_json_ is not pre-supplied.
Useful? React with 👍 / 👎.
Comment on lines
+8261
to
+8263
| auto * object_storage = dynamic_cast<StorageObjectStorageCluster *>(dest_storage.get()); | ||
|
|
||
| auto * iceberg_metadata = dynamic_cast<IcebergMetadata *>(object_storage->getExternalMetadata(query_context)); |
There was a problem hiding this comment.
The replicated EXPORT PARTITION ... TO TABLE path unconditionally casts dest_storage to StorageObjectStorageCluster and immediately dereferences it. For local Iceberg destination tables, dest_storage is typically StorageObjectStorage, so this can null-deref before any graceful validation, causing a crash instead of returning a structured exception.
Useful? React with 👍 / 👎.
Comment on lines
+676
to
+678
| std::lock_guard lock(status_changes_mutex); | ||
| std::swap(status_changes, local_status_changes); | ||
| } |
There was a problem hiding this comment.
Swapping all pending status changes into a local queue means any exception during processing drops the remaining events. exportMergeTreePartitionStatusHandlingTask() catches and returns (especially on ZooKeeper/session errors), but does not requeue these keys, so some exports can miss terminal status transitions until another watch event happens (or forever if none arrives).
Useful? React with 👍 / 👎.
Comment on lines
+1695
to
+1700
| const auto latest_schema_id = metadata->getValue<Int64>(Iceberg::f_current_schema_id); | ||
| if (latest_schema_id != original_schema_id) | ||
| throw Exception(ErrorCodes::NOT_IMPLEMENTED, | ||
| "Table schema changed before export could commit (expected schema {}, got {}). " | ||
| "Restart the export operation.", | ||
| original_schema_id, latest_schema_id); |
There was a problem hiding this comment.
This code throws on schema/spec drift before running the idempotency scan. If the first commit already succeeded but the node crashed before ZooKeeper was marked completed, and the Iceberg table schema/spec changes before retry, the retry will now fail permanently instead of recognizing the existing snapshot tagged with the same transaction id.
Useful? React with 👍 / 👎.
Collaborator
Author
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 675061716f
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
|
||
| manifest.lock_inside_the_task = json->getValue<bool>("lock_inside_the_task"); | ||
|
|
||
| manifest.write_full_path_in_iceberg_metadata = json->getValue<bool>("write_full_path_in_iceberg_metadata"); |
There was a problem hiding this comment.
Treating write_full_path_in_iceberg_metadata as mandatory breaks reading manifests that were written before this field existed. During rolling upgrades, existing ZooKeeper metadata.json entries for in-flight exports won't have this key, so getValue<bool> throws and status/polling paths that call fromJsonString cannot process those tasks. Please make this field optional with a default (false) when absent.
Useful? React with 👍 / 👎.
| auto dot_pos = data_file_storage_path.rfind('.'); | ||
| auto slash_pos = data_file_storage_path.rfind('/'); | ||
| if (dot_pos != String::npos && (slash_pos == String::npos || dot_pos > slash_pos)) | ||
| return data_file_storage_path.substr(0, dot_pos) + ".avro"; |
There was a problem hiding this comment.
sidecarStoragePath rewrites any extension to .avro; if the Iceberg table writes Avro data files, the data file already ends with .avro, so the sidecar path equals the data file path. writeDataFileSidecar then writes in Rewrite mode, replacing the actual data file with sidecar metadata during onFinish, which corrupts exported data for Avro-formatted Iceberg tables.
Useful? React with 👍 / 👎.
arthurpassos
changed the title
Collaborator
Author
|
I think there is a race condition on the commit procedure, and that is why I am observing failures that indicate more data was written than expected. The normal procedure of exporting partitions is that the replica that finishes the last part export will try to commit. At the same time, it is possible that some other replica is running the cleanup routine and finds a task in the pseudo "pending commit" state. In this case, it will try to commit as well, concurrently. On plain object storage that is not a problem because the commit filename will guarantee the idempotency. On Iceberg, we don't have that control. The only control we have is the I think the simplest fix is to introduce a commit lock on zookeeper |
Collaborator
Author
Locking on zookeeper is not enough because the locks are ephemeral and there is a chance the replica briefly loses connection to zookeeper, the lock is released, some other replica b acquires it and the replica a doesn't hear about it. And it still tries to commit. The real deal is to re-check the |
Collaborator
Audit Report: PR #1618AI audit note: This review comment was generated by AI (gpt-5.3-codex). Scope: Export partition to Apache Iceberg. Confirmed defectsHigh: Iceberg write path can report success after exhausting metadata commit retries
size_t i = 0;
while (i < MAX_TRANSACTION_RETRIES)
{
if (initializeMetadata())
break;
++i;
}Medium: Export commit aggregates can overflow due 64-bit to 32-bit narrowing
Int32 total_rows = 0;
Int32 total_chunks_size = 0;
...
total_rows += static_cast<Int32>(sidecar.record_count);
total_chunks_size += static_cast<Int32>(sidecar.file_size_in_bytes);Medium: Export commit depends on mutable source partition state instead of persisted manifest state
const auto parts = storage.getDataPartsVectorInPartitionForInternalUsage(
MergeTreeDataPartState::Active, partition_id, lock);
if (parts.empty())
throw Exception(ErrorCodes::BAD_ARGUMENTS,
"Cannot find active part for partition_id '{}' to derive Iceberg partition values...",
partition_id);if (!manifest.iceberg_metadata_json.empty())
{
iceberg_args.metadata_json_string = manifest.iceberg_metadata_json;
if (source_storage.getInMemoryMetadataPtr()->hasPartitionKey())
iceberg_args.partition_values =
getPartitionValuesForIcebergCommit(source_storage, manifest.partition_id);
}Medium: Backward-incompatible manifest parsing breaks in-flight exports after upgrade
manifest.lock_inside_the_task = json->getValue<bool>("lock_inside_the_task");
manifest.write_full_path_in_iceberg_metadata = json->getValue<bool>("write_full_path_in_iceberg_metadata");Medium: Commit-phase failures are swallowed without terminal state transition
try
{
auto context = ExportPartitionUtils::getContextCopyWithTaskSettings(storage.getContext(), manifest);
ExportPartitionUtils::commit(manifest, destination_storage, zk, storage.log.load(), export_path, context, storage);
}
catch (const Exception & e)
{
const auto error_requests = getErrorRequests(export_path, storage.replica_name, zk, storage.log.load(), part_name, e);
// logs + error bookkeeping only, no terminal status transition here
}Low: KILL can be silently overwritten by an in-flight commit (
|
Collaborator
|
AI audit note: This review comment was generated by AI (gpt-5.3-codex). Follow-up audit delta for PR #1618 — only newly found issues from the latest state-model-driven fault-injection pass:
|
Collaborator
Author
|
Collaborator
But given that we rely on the Iceberg write, you think it does not make sense to fix it? |
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 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.
Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):
...
Documentation entry for user-facing changes
...
CI/CD Options
Exclude tests:
Regression jobs to run: