Export partition - allow non matching partition expressions iff monotonicity is proved and data is not repartitioned - #2074
Conversation
|
@codex review |
![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: 3f3af7c4fc
ℹ️ 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".
| castColumn({std::move(values_column), source_type, column}, destination_type), destination_type, column}; | ||
|
|
||
| const auto dest_transform_with_tz = Iceberg::parseTransformAndArgument(dest_transform, partition_timezone); | ||
| if (!destinationTransformIsConstant(*dest_transform_with_tz, cast_values, context)) |
There was a problem hiding this comment.
When this dynamic check succeeds in the replicated EXPORT PARTITION path, the proof is only for the scheduling-time parts; the manifest still persists just partition_id, and commit later calls getPartitionSourceBlockForIcebergCommit, which reads the current active part for that partition. If a coarser source partition is accepted because its current rows map to one destination partition (for example toYear(ts) into Iceberg day(ts)) and, before commit, another day is inserted and merged with the exported parts, the committed manifest can tag the already exported files with the merged/current part's representative value instead of the exported rows' value. Persist/use representative values from the validated parts, or keep these cases structural-only for replicated exports.
Useful? React with 👍 / 👎.
| { | ||
| if (const auto * id = arg->as<ASTIdentifier>()) | ||
| column = id->name(); | ||
| else if (const auto * lit = arg->as<ASTLiteral>(); lit && lit->value.getType() != Field::Types::String) |
There was a problem hiding this comment.
This drops string literals from source partition terms, so toRelativeDayNum(ts, 'UTC') is treated as the same structural transform as the destination Iceberg day(ts) transform. With iceberg_partition_timezone set differently for the export (for example America/Los_Angeles), one UTC-day source partition can contain rows from two destination days; the structural fast path then skips the dynamic min/max proof and commits them as a single Iceberg partition. Carry/compare the timezone literal, or force timezone-qualified source terms through the dynamic proof.
Useful? React with 👍 / 👎.
|
Consider mentioning 'export' in the title and/or description |
arthurpassos
changed the title
Done, I'll soon add the description |
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ca43cf1f6d
ℹ️ 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".
This reverts commit 3848194.
|
@Selfeer I also wondered about this very same timezone case. As far as I could understand from apache iceberg docs and articles, the This is not crystal clear, tho. The docs are not very explicit. For this reason, I think @ianton-ru introduced the Export partition must respect it because it exists, even if it is non compliant. That's why I respect |
| } | ||
| else | ||
| { | ||
| if (range.left < bounds->first) |
There was a problem hiding this comment.
Im not sure about safety here so what about using accurateLess here https://github.com/Altinity/ClickHouse/blob/antalya-26.3/src/Common/FieldAccurateComparison.h#L8 ?
There was a problem hiding this comment.
Should be addressed
| auto range = part->minmax_idx->hyperrectangle[slot]; | ||
| range.shrinkToIncludedIfPossible(); | ||
| if (!bounds) | ||
| { | ||
| bounds.emplace(range.left, range.right); | ||
| } | ||
| else | ||
| { | ||
| if (range.left < bounds->first) | ||
| bounds->first = range.left; | ||
| if (bounds->second < range.right) | ||
| bounds->second = range.right; | ||
| } |
There was a problem hiding this comment.
There was a problem hiding this comment.
Should be addressed
| return terms; | ||
| } | ||
|
|
||
| std::optional<std::pair<Field, Field>> calculatePartitionColumnMinMax( |
There was a problem hiding this comment.
I expected (by the function name) that returing value will be MiMaxIndex. But it is a pair. Maybe we could return MiMaxIndex?
There was a problem hiding this comment.
Should be addressed
| std::vector<PartitionTerm> destination_terms; | ||
| destination_terms.reserve(actual_size); | ||
| for (UInt32 i = 0; i < actual_size; ++i) | ||
| { | ||
| const auto af = actual_fields->getObject(i); | ||
| const auto dest_transform = af->getValue<String>(Iceberg::f_transform); | ||
| const String column = source_id_to_name(af->getValue<Int32>(Iceberg::f_source_id)); | ||
| const auto transform_and_argument = Iceberg::parseTransformAndArgument(dest_transform, partition_timezone); |
There was a problem hiding this comment.
Here we parse manually, but below we use parsePartitionTerms. Should this use a separate function instead?
| } | ||
| }; | ||
|
|
||
| std::vector<PartitionTerm> parsePartitionTerms(const ASTPtr & partition_key_ast) |
There was a problem hiding this comment.
I suggest adding unit tests for parse* functions.
| struct PartitionTerm | ||
| { | ||
| String column; | ||
| String function; | ||
| std::optional<Int64> argument; | ||
| std::optional<String> time_zone; | ||
|
|
||
| bool operator==(const PartitionTerm & other) const | ||
| { | ||
| return column == other.column && function == other.function && argument == other.argument | ||
| && time_zone == other.time_zone; | ||
| } | ||
| }; |
There was a problem hiding this comment.
argument and time_zone are two independent std::optionals, but every construction site treats them as mutually exclusive. Worth reconsidering the design so the struct can't be filled in incorrectly by its user.
There was a problem hiding this comment.
You mean two different constructors? one that takes argument and another one that takes the time_zone?
There was a problem hiding this comment.
Not quite - rather than a constructor per case, a std::variant<std::monostate, Int64, String> in place of the two separate optionals would make the invalid "both set" state unrepresentable, with much less machinery than a Column/Function class hierarchy - the set of shapes here is small and closed, not open-ended.
Similarly, column being empty currently doubles as an implicit "couldn't parse" sentinel (Im not sure and I need to check - a nested function like toYYYYMM(toDate(ts)) isn't unwrapped).
|
|
||
|
|
||
| /// asserts that the source column maps to a single destination partition by checking the monotonicity on the min/max ranges | ||
| void verifyColumnMapsToSinglePartition( |
There was a problem hiding this comment.
nit: If we use PartitionTerm we could reduce params count.
|
Hey @arthurpassos - separate from the positional/name column-matching issue. I want to confirm the intended behaviour here before I pin it in a test. What happensA destination with an extra column is rejected on column count: CREATE TABLE src (id Int64, a Int32)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/shard0/src', '{replica}')
ORDER BY tuple() PARTITION BY a;
CREATE TABLE dst (id Int64, a Int32, b Int32)
ENGINE = S3(..., format='Parquet', partition_strategy='hive') PARTITION BY a;
INSERT INTO src VALUES (1, 42);
ALTER TABLE src EXPORT PARTITION ID '42' TO TABLE dst;
-- Code: 20. Number of columns doesn't match (source: 2 and result: 3).That one seems right to me. But giving the extra column a CREATE TABLE dst (id Int64, a Int32, b Int32 DEFAULT 42)
ENGINE = S3(..., format='Parquet', partition_strategy='hive') PARTITION BY a;
ALTER TABLE src EXPORT PARTITION ID '42' TO TABLE dst;
-- Code: 20. Number of columns doesn't match (source: 2 and result: 3).Same error, same counts - the What I'd expectI'd expect the second case to succeed, with SELECT id, a, b FROM dst;
-- 1 42 42which is what INSERT INTO dst (id, a)
SELECT id, a
FROM src;would produce. A Is the current behaviour intended? There's no data-loss risk either way since it's a loud rejection - I just want the test to assert the intended behaviour rather than whichever one I guess at. |
|
I suggest adding setting to turn on/off this functionality. |
| } | ||
| } | ||
| } | ||
| terms.push_back(std::move(term)); |
There was a problem hiding this comment.
Seems that exsits a case where term is empty. Is this expected behavior?
|
Can you check #2138 - this does seem like we should not be allowing the export here or have some other fix in this case. |
|
I think jump also reported the same thing and you agreed that this is okay, @arthurpassos but please check the issue and if all is right even with the Insert and Export behavior being different - I'll just update the expectations of our tests. |
Audit Review — PR #2074
Summary of findings
1. 🔴 High — Argument order lost, monotonicity proof evaluates the wrong expressionAnchor: Impact: A source partition that spans multiple destination partitions can pass validation. All rows of a part are then written into the single directory computed from the part's min row — rows are silently misplaced into a wrong destination partition, and readers relying on hive/wildcard partition pruning get wrong query results. Trigger (smallest realistic case):
Why it is a defect:
Iceberg destinations are unaffected by luck of signature: Fix direction: Store the full ordered argument list (or the sub-AST) in Regression test direction: Stateless test exporting from 2. 🟠 Medium — Iceberg commit permanently fails when no exported part remains locallyAnchor: Impact: An export that has already uploaded all data files retries the commit until Trigger: Exported parts are merged away and cleaned up (
Why it is a defect: The function now requires at least one part from Fix direction: Persist the folded min/max (or the derived partition source block) in the ZooKeeper manifest at schedule time and use it at commit — removing the dependency on local parts entirely. Regression test direction: Integration test: schedule an Iceberg export, let all source parts merge, drop old parts (or restart the node), then assert the commit still succeeds. 3. 🟡 Low —
|
| Destination key shape | What happens |
|---|---|
toStartOfInterval(ts, INTERVAL 1 DAY) |
nested-function argument is dropped; rebuilt call has wrong arity → NUMBER_OF_ARGUMENTS_DOESNT_MATCH |
| Float literal in the key | Field::safeGet<Int64> throws BAD_GET |
Nested expression, e.g. toYYYYMM(toDate(ts)) |
error message names column '' |
Iceberg void transform |
mapped to tuple, which has no monotonicity → always rejected, although void never repartitions anything |
Fix direction: Reject unparseable/nested terms explicitly with BAD_ARGUMENTS before building the function; special-case void as trivially single-valued.
Regression test direction: Negative stateless tests asserting BAD_ARGUMENTS for each shape; an accept-case for a void field in the Iceberg spec.
5. 🟡 Low — Force re-export deletes previous export state before validation
Anchor: src/Storages/StorageReplicatedMergeTree.cpp — exportPartitionToTable (tryRemoveRecursive runs before verifyPlainPartitionCompatibility)
Impact: EXPORT PARTITION ... SETTINGS export_merge_tree_partition_force_export = 1 to a plain destination with an incompatible partition key removes the existing export's ZooKeeper state (killing an in-progress export) and then throws — leaving no export scheduled at all.
Why it is a defect: Before this PR the plain-destination partition-key check ran at the top of the function, before any ZooKeeper mutation. The new check needs parts, which are collected after the destructive tryRemoveRecursive. (The Iceberg check already had this ordering, so for Iceberg destinations this is pre-existing.)
Fix direction: Collect parts and run both partition-compatibility checks before the force-removal of the previous export.
Regression test direction: Integration test: force re-export with an incompatible destination key; assert the previous export state survives.
Coverage summary
Scope reviewed — full PR diff (13 files):
- term parsing and structural matching; cast + transform monotonicity proof
- plain and Iceberg schedule-time gates in
MergeTreeData::exportPartToTable/StorageReplicatedMergeTree::exportPartitionToTable - Iceberg commit min/max derivation;
iceberg_partition_timezonemanifest plumbing - write paths (
ExportPartTask→StorageObjectStorage::import→computePartitionKey) - commit retry/failure classification; part pinning lifecycle; test changes
Categories failed: destination-term reconstruction (argument order); commit-time part availability (restart); Nullable-wrapper gating; error-contract consistency; force-export rollback ordering.
Categories passed:
- Iceberg transform mapping and width-first argument order
- structural match type gating — including
bucketbeing structural-only, sinceicebergBucketTransformexposes no monotonicity - cast monotonicity gate — non-monotonic casts such as
Int → Stringcorrectly fail closed becauseCASTreports no monotonicity for them - date transforms (
toYearNumSinceEpoch,toMonthNumSinceEpoch,toRelativeDayNum,toRelativeHourNum) have monotonicity viaIFunctionDateOrDateTime - unpartitioned-destination fast paths;
MinMaxIndexfold soundness, including partial part sets at commit - timezone consistency schedule → manifest → execution; manifest forward/backward JSON compatibility
- concurrency: schedule-time parts snapshot under
lockParts, commit underreadLockParts, immutableminmax_idx - no sensitive-data leakage in new messages
- dismissed candidate:
Field::safeGet<Int64>accepts UInt64 literals, so integer literals in partition keys parse fine
Assumptions / limits:
- Static reasoning only; no runtime execution.
- Pre-existing and unchanged by this PR:
verifyExportSchemaCastablemaps columns positionally while all partition checks and the hive write path match by name — coincidentally-named but positionally-different columns can validate against the wrong source column. The identical-AST fast path also does not compare column types/timezones. - Timezone monotonicity relies on the same DST assumptions as partition pruning.
arthurpassos
added
antalya
port-antalya
I think this deserves a separate issue |
Export partition is already experimental and back by a setting the user must opt in |
|
@Selfeer I would appreciate if you could "humanize" a bit more those AI reports. For example, the following issue is very hard to understand and doesn't explain what is happening: 1. 🔴 High — Argument order lost, monotonicity proof evaluates the wrong expression. It says "anchor" pointing to some code location. Then it says the impact. Then the trigger. It is too much to read to understand the real problem. Brain energy required to process this is very high. Instead, if you could understand the issue yourself first and then give me a few SQL instructions that repro the case and a human explanation of what's going on, that would be 100 times better. |
I agree. I'd much rather do it the way you described: perform my own investigation on the findings and raise issues as needed-and we actually do that. But even in that case, I would still have to post this exact message first. The purpose of the audit review has always been to perform a quick review of the PR without running any tests first, share the findings with the developer, and let you decide whether they are actual issues. If they are, we then raise separate issues afterward. I can update the skills we use for the audit review to make the output easier to read, but overall, the audit review has always been a separate part of the verification process, separate from our actual testing. |
5 participants
Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):
Allow export partition through different partition expressions as long as the destination expression does not repartition the data. This is asserted through a destination expression monotonicity check on the source minmax values. Destination expression columns must be a subset of the source.
Documentation entry for user-facing changes
...
CI/CD Options
Exclude tests:
Regression jobs to run: