924060929 commented on code in PR #66007:
URL: https://github.com/apache/doris/pull/66007#discussion_r3681560806
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergTableSink.java:
##########
@@ -121,13 +136,19 @@ public PhysicalProperties getRequirePhysicalProperties() {
return PhysicalProperties.GATHER;
}
- Set<String> partitionNames = targetTable.getPartitionNames();
+ Set<String> partitionNames = new
java.util.TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ for (PartitionField field : targetIcebergTable.spec().fields()) {
+ Types.NestedField sourceField =
targetIcebergTable.schema().findField(field.sourceId());
Review Comment:
**Blocking — this silently flips Iceberg write distribution, and the new key
does not match what the spec means.**
The old code read `targetTable.getPartitionNames()`. `targetTable` is
statically an `ExternalTable`, `TableIf#getPartitionNames()` defaults to
`Collections.emptySet()`, and only `HMSExternalTable` overrides it —
`IcebergExternalTable` does not. So on master this whole branch is **dead**:
every Iceberg write gets `SINK_RANDOM_PARTITIONED`.
Deriving the names from `spec()` makes it live, and `PhysicalPlanTranslator`
turns it into `DataPartition(TPartitionType.HIVE_TABLE_SINK_HASH_PARTITIONED,
...)`. That is a new exchange in front of every partitioned-Iceberg INSERT plus
a change in output file count and write parallelism — a behavior change
unrelated to metadata-generation consistency, and not mentioned in the
description.
It also does not match the semantics of that distribution spec.
`PhysicalHiveTableSink` uses `getPartitionColumnNames()`, i.e. **identity**
partition columns, so hashing on them co-locates one partition on one instance.
Here the hash key is the partition field's *source* column, so for `days(ts)` /
`bucket(n, id)` / `truncate(n, s)` rows of the same partition still scatter
across instances — you pay for a shuffle without reducing files per partition.
Minor extra: `spec().fields()` retains dropped partition fields as `void`
transforms, and their source columns end up in the hash key too.
Suggestion: take this out of this PR, and when it lands, restrict it to
`Transforms.identity()` fields (or hash the transformed value).
Separately, the index change on the loop below is a genuine improvement
worth keeping — walking `child().getOutput()` directly instead of indexing it
with a position from `getFullSchema()` removes a latent mismatch whenever the
full schema contains hidden columns.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -989,15 +990,23 @@ public void addPlannerHook(PlannerHook plannerHook) {
* @param tableSnapshot table snapshot info
* @param scanParams table scan params (e.g., branch/tag for Iceberg
tables)
*/
- public void loadSnapshots(TableIf specificTable, Optional<TableSnapshot>
tableSnapshot,
+ public Optional<MvccSnapshot> loadSnapshots(TableIf specificTable,
Optional<TableSnapshot> tableSnapshot,
Optional<TableScanParams> scanParams) {
- if (specificTable instanceof MvccTable) {
- MvccTableInfo mvccTableInfo = new MvccTableInfo(specificTable);
- if (!snapshots.containsKey(mvccTableInfo)) {
- snapshots.put(mvccTableInfo,
- ((MvccTable)
specificTable).loadSnapshot(tableSnapshot, scanParams));
- }
+ if (!(specificTable instanceof MvccTable)) {
+ return Optional.empty();
}
+ MvccTableInfo mvccTableInfo = new MvccTableInfo(specificTable);
+ MvccSnapshot snapshot;
+ if (tableSnapshot.isPresent() || scanParams.isPresent()) {
Review Comment:
**Blocking — repeated qualified references to the same table still resolve
independently.**
Only the unqualified path is memoized through `latestSnapshots`; this arm
calls `loadSnapshot` every time. And the branch/tag arm of
`IcebergUtils.getSnapshotCacheValue` does not go through the snapshot cache at
all — it does `retainTableGeneration(getIcebergTable(dorisTable))` fresh on
each call.
So `SELECT ... FROM t@branch(b) a JOIN t@branch(b) c` resolves twice, and if
the branch head advances in between, the two relations end up on different
generations — exactly the bug class this PR is fixing. Before the patch the
"first writer wins" behavior happened to avoid it, so this is a narrow
regression.
Suggestion: widen the cache key to `(MvccTableInfo, tableSnapshot,
scanParams)`. `latestSnapshots` then folds back into `snapshots`, and it also
saves a duplicate remote metadata resolution per repeated relation.
One more thing worth a comment on the field: `snapshots.put(...)` is now
unconditional, so the statement-scoped compatibility map went from
first-writer-wins to last-writer-wins. Roughly thirty call sites still read it
through `MvccUtil.getSnapshotFromContext` (MTMV partition logic,
`HiveTableSink`, `MetadataGenerator`, the no-arg `getFullSchema()` overloads).
For a multi-version statement they now observe whichever relation was bound
last, which is an arbitrary order. Worth documenting on the field even if the
remaining consumers migrate later.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java:
##########
@@ -113,14 +156,41 @@ protected LogicalFileScan(RelationId id, ExternalTable
table, List<String> quali
}
private static SelectedPartitions initialSelectedPartitions(
- ExternalTable table, Optional<TableScanParams> scanParams) {
+ ExternalTable table, Optional<TableScanParams> scanParams,
+ Optional<MvccSnapshot> relationSnapshot) {
if ((table instanceof PaimonExternalTable || table instanceof
PaimonSysExternalTable)
&& scanParams.isPresent() && scanParams.get().isOptions()) {
// A relation-scoped historical snapshot cannot reuse partitions
cached for the
// statement-level latest snapshot; Paimon will prune its selected
snapshot instead.
return SelectedPartitions.NOT_PRUNED;
}
- return
table.initSelectedPartitions(MvccUtil.getSnapshotFromContext(table));
+ return table.initSelectedPartitions(relationSnapshot);
+ }
+
+ private static Optional<List<Column>> captureRelationSchema(
+ ExternalTable table, Optional<TableScanParams> scanParams,
+ Optional<MvccSnapshot> relationSnapshot) {
+ if (scanParams.isPresent() && scanParams.get().isOptions()) {
+ if (table instanceof PaimonExternalTable) {
+ return Optional.of(ImmutableList.copyOf(
+ ((PaimonExternalTable)
table).getFullSchema(scanParams.get())));
+ }
+ if (table instanceof PaimonSysExternalTable) {
+ return Optional.of(ImmutableList.copyOf(
+ ((PaimonSysExternalTable)
table).getFullSchema(scanParams.get())));
+ }
+ }
+ return captureRelationSchema(table, relationSnapshot);
+ }
+
+ protected static Optional<List<Column>> captureRelationSchema(
+ ExternalTable table, Optional<MvccSnapshot> relationSnapshot) {
+ if (!(table instanceof MvccTable)) {
Review Comment:
**Blocking — plain Hive relations pin the logical schema, but the physical
side cannot reproduce it.**
`HMSExternalTable` implements `MvccTable`, so this guard passes even for
`DLAType.HIVE` and the columns get frozen into an `ImmutableList` that
`computeOutput()` turns into slots. But `HMSExternalTable.loadSnapshot` returns
`new EmptyMvccSnapshot()` for plain Hive, and that snapshot carries no state,
so at execution time:
`FileQueryScanNode.initSchemaParams` →
`getBaseSchema(Optional.of(EmptyMvccSnapshot), false)` →
`HMSExternalTable.getFullSchema(snapshot)` → not HUDI/ICEBERG →
`super.getFullSchema(snapshot)` → **ignores the argument** and re-reads the
live schema cache.
The logical output is therefore pinned at T0 while `numOfColumnsFromFile`
and the ORC position mapping are computed at T1. If the schema cache is
refreshed inside the planning window (REFRESH TABLE, HMS event, TTL expiry) the
two disagree. This asymmetry is introduced by the patch — before it, both sides
read the same live cache.
Suggestion: either carry `relationSchema` through to the physical scan the
way `relationSnapshot` is carried, or skip capturing when the snapshot cannot
replay a schema.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSnapshotCacheValue.java:
##########
@@ -54,4 +79,142 @@ public IcebergSnapshot getSnapshot() {
public Optional<Map<Integer, List<String>>> getNameMapping() {
return nameMapping;
}
+
+ public Optional<Table> getIcebergTable() {
+ return icebergTable;
+ }
+
+ static Table retainTableGeneration(Table table) {
+ if (!(table instanceof HasTableOperations) ||
isFrozenGeneration(table)) {
+ return table;
+ }
+ TableOperations operations = ((HasTableOperations) table).operations();
+ // Capture current() exactly once so every projection derived from the
returned table sees
+ // one metadata generation even when the shared catalog handle
refreshes concurrently.
+ TableOperations frozenOperations = new
FrozenTableOperations(operations, operations.current());
+ return tableWithOperations(table, frozenOperations);
+ }
+
+ static boolean isFrozenGeneration(Table table) {
+ return table instanceof HasTableOperations
+ && ((HasTableOperations) table).operations() instanceof
FrozenTableOperations;
+ }
+
+ static Table createWritableTable(Table retainedTable, Table liveTable) {
+ if (!isFrozenGeneration(retainedTable)) {
+ return retainedTable;
+ }
+ if (!(liveTable instanceof HasTableOperations)
+ || isFrozenGeneration(liveTable)) {
+ throw new IllegalArgumentException(
+ "Iceberg commit table must provide writable table
operations");
+ }
+ TableMetadata retainedMetadata = ((HasTableOperations)
retainedTable).operations().current();
+ TableOperations liveOperations = ((HasTableOperations)
liveTable).operations();
+ return tableWithOperations(retainedTable,
+ new WritableTableOperations(liveOperations, retainedMetadata));
+ }
+
+ private static Table tableWithOperations(Table table, TableOperations
operations) {
+ if (table instanceof BaseTable) {
+ return new BaseTable(operations, table.name(), ((BaseTable)
table).reporter());
+ }
+ return new BaseTable(operations, table.name());
+ }
+
+ private abstract static class RetainedTableOperations implements
TableOperations {
+ protected final TableOperations delegate;
+ private final TableMetadata metadata;
+
+ private RetainedTableOperations(TableOperations delegate,
TableMetadata metadata) {
+ this.delegate = delegate;
+ this.metadata = metadata;
+ }
+
+ @Override
+ public TableMetadata current() {
+ return metadata;
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ return metadata;
+ }
+
+ @Override
+ public FileIO io() {
+ return delegate.io();
+ }
+
+ @Override
+ public EncryptionManager encryption() {
+ return delegate.encryption();
+ }
+
+ @Override
+ public String metadataFileLocation(String fileName) {
+ return delegate.metadataFileLocation(fileName);
+ }
+
+ @Override
+ public LocationProvider locationProvider() {
+ return delegate.locationProvider();
+ }
+ }
+
+ private static class FrozenTableOperations extends RetainedTableOperations
{
+ private FrozenTableOperations(TableOperations delegate, TableMetadata
metadata) {
+ super(delegate, metadata);
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata newMetadata) {
+ throw new UnsupportedOperationException("Frozen Iceberg table
generation is read-only");
+ }
+ }
+
+ private static class WritableTableOperations extends
RetainedTableOperations {
+ private final TableMetadata retainedMetadata;
+ private TableMetadata currentMetadata;
+
+ private WritableTableOperations(TableOperations delegate,
TableMetadata retainedMetadata) {
+ super(delegate, retainedMetadata);
+ this.retainedMetadata = retainedMetadata;
+ this.currentMetadata = retainedMetadata;
+ }
+
+ @Override
+ public TableMetadata current() {
+ return currentMetadata;
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ TableMetadata refreshedMetadata = delegate.refresh();
+ // Data-only snapshot advances are safe to replay, but a changed
writer contract must
+ // fail instead of silently committing files produced for another
metadata generation.
+ if (!isWriterCompatible(refreshedMetadata)) {
+ throw new CommitFailedException(
Review Comment:
`CommitFailedException` is the one exception Iceberg's commit loop retries:
`SnapshotProducer.commit()` is built with
`Tasks...onlyRetryOn(CommitFailedException.class)`, and `apply()` calls
`refresh()` on every attempt.
So a condition you explicitly decided must *not* be retried gets retried
`commit.retry.num-retries` times (default 4) before it surfaces to the user.
`ValidationException` aborts immediately and reads better in the log.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -697,6 +724,33 @@ public TableScan createTableScan() throws UserException {
return icebergTableScan;
}
+ private Table useFrozenTableGeneration(Table currentTable) {
+ Optional<MvccSnapshot> snapshot = getPinnedRelationSnapshot();
+ if
(snapshot.filter(IcebergMvccSnapshot.class::isInstance).isPresent()) {
+ IcebergSnapshotCacheValue cacheValue =
+ ((IcebergMvccSnapshot)
snapshot.get()).getSnapshotCacheValue();
+ if (cacheValue.getIcebergTable().isPresent()) {
+ Table frozenBaseTable = cacheValue.getIcebergTable().get();
+ if (isSystemTable && source.getTargetTable() instanceof
IcebergSysExternalTable) {
+ IcebergSysExternalTable systemTable =
(IcebergSysExternalTable) source.getTargetTable();
+ if (systemTable.supportsSnapshotSelection()) {
+ MetadataTableType tableType =
MetadataTableType.from(systemTable.getSysTableType());
+ Preconditions.checkArgument(tableType != null,
+ "Unknown Iceberg system table type: %s",
systemTable.getSysTableType());
+ // Snapshot-selectable metadata tables must derive
their scans and schemas
+ // from the same frozen base generation as the
relation's snapshot fence.
+ return
MetadataTableUtils.createMetadataTableInstance(frozenBaseTable, tableType);
Review Comment:
The scan is built from the frozen base here, but the tuple slots are not.
`IcebergSysExternalTable` is constructed fresh per binding
(`SysTableResolver.resolveForPlan` → `createSysExternalTable`) and derives its
schema from `sourceTable.getIcebergTable()` at that moment — i.e. the live
generation.
With the shared metadata cache those are usually the same `BaseTable`, so
this is mostly benign. Under a session catalog it is not:
`loadIcebergTableWithSession` performs an independent `ops.loadTable(...)` on
every call, so the schema and the scan come from two separate remote loads. If
a partition spec evolved in between, the `partition` struct of `FILES` /
`ENTRIES` / `PARTITIONS` differs and the slots no longer line up with
`scan.project(...)`.
Deriving the system-table schema from the frozen base as well would close
this.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSnapshotCacheValue.java:
##########
@@ -54,4 +79,142 @@ public IcebergSnapshot getSnapshot() {
public Optional<Map<Integer, List<String>>> getNameMapping() {
return nameMapping;
}
+
+ public Optional<Table> getIcebergTable() {
+ return icebergTable;
+ }
+
+ static Table retainTableGeneration(Table table) {
+ if (!(table instanceof HasTableOperations) ||
isFrozenGeneration(table)) {
+ return table;
+ }
+ TableOperations operations = ((HasTableOperations) table).operations();
+ // Capture current() exactly once so every projection derived from the
returned table sees
+ // one metadata generation even when the shared catalog handle
refreshes concurrently.
+ TableOperations frozenOperations = new
FrozenTableOperations(operations, operations.current());
+ return tableWithOperations(table, frozenOperations);
+ }
+
+ static boolean isFrozenGeneration(Table table) {
+ return table instanceof HasTableOperations
+ && ((HasTableOperations) table).operations() instanceof
FrozenTableOperations;
+ }
+
+ static Table createWritableTable(Table retainedTable, Table liveTable) {
+ if (!isFrozenGeneration(retainedTable)) {
+ return retainedTable;
+ }
+ if (!(liveTable instanceof HasTableOperations)
+ || isFrozenGeneration(liveTable)) {
+ throw new IllegalArgumentException(
+ "Iceberg commit table must provide writable table
operations");
+ }
+ TableMetadata retainedMetadata = ((HasTableOperations)
retainedTable).operations().current();
+ TableOperations liveOperations = ((HasTableOperations)
liveTable).operations();
+ return tableWithOperations(retainedTable,
+ new WritableTableOperations(liveOperations, retainedMetadata));
+ }
+
+ private static Table tableWithOperations(Table table, TableOperations
operations) {
+ if (table instanceof BaseTable) {
+ return new BaseTable(operations, table.name(), ((BaseTable)
table).reporter());
+ }
+ return new BaseTable(operations, table.name());
+ }
+
+ private abstract static class RetainedTableOperations implements
TableOperations {
+ protected final TableOperations delegate;
+ private final TableMetadata metadata;
+
+ private RetainedTableOperations(TableOperations delegate,
TableMetadata metadata) {
+ this.delegate = delegate;
+ this.metadata = metadata;
+ }
+
+ @Override
+ public TableMetadata current() {
+ return metadata;
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ return metadata;
+ }
+
+ @Override
+ public FileIO io() {
+ return delegate.io();
+ }
+
+ @Override
+ public EncryptionManager encryption() {
+ return delegate.encryption();
+ }
+
+ @Override
+ public String metadataFileLocation(String fileName) {
+ return delegate.metadataFileLocation(fileName);
+ }
+
+ @Override
+ public LocationProvider locationProvider() {
Review Comment:
The wrapper is not a complete freeze: `io()`, `locationProvider()` and
`metadataFileLocation()` all delegate straight through, so they follow the
newest metadata rather than the retained one. A table-property change to
`write.data.path` / `write.metadata.path` would be observed by what is supposed
to be a frozen read view.
It also does not override `temp(TableMetadata)`. The interface default is
`return this`, while `HadoopTableOperations`, `RESTTableOperations` and
`BaseMetastoreTableOperations` all override it, and
`BaseTransaction$TransactionTableOperations` calls it — so wrapping drops the
implementation's behavior. The `isWriterCompatible()` check on `location` +
`properties` covers most of the practical impact today, but the wrapper is
worth completing.
(For the record, `requireStrictCleanup()` is fine to leave alone — nothing
in iceberg-core 1.10.1 overrides it.)
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundIcebergTableSink.java:
##########
@@ -130,17 +151,28 @@ public <R, C> R accept(PlanVisitor<R, C> visitor, C
context) {
public Plan withGroupExpression(Optional<GroupExpression> groupExpression)
{
return new UnboundIcebergTableSink<>(nameParts, colNames, hints,
partitions,
dmlCommandType, groupExpression,
Optional.of(getLogicalProperties()), child(),
- staticPartitionKeyValues, rewrite);
+ staticPartitionKeyValues, rewrite, branchName);
}
@Override
public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression>
groupExpression,
Optional<LogicalProperties> logicalProperties, List<Plan>
children) {
return new UnboundIcebergTableSink<>(nameParts, colNames, hints,
partitions,
- dmlCommandType, groupExpression, logicalProperties,
children.get(0), staticPartitionKeyValues, rewrite);
+ dmlCommandType, groupExpression, logicalProperties,
children.get(0),
+ staticPartitionKeyValues, rewrite, branchName);
}
public boolean isRewrite() {
return rewrite;
}
+
+ public Optional<String> getBranchName() {
Review Comment:
This looks like dead state. `branchName` is stored, threaded through
`withChildren` / `withGroupExpression` / `withGroupExprLogicalPropChildren`,
given a `withBranchName()` and asserted in a unit test — but nothing in
production code calls `getBranchName()`. The branch already reaches execution
through the `branchName` argument of `InsertIntoTableCommand` /
`InsertOverwriteTableCommand`, set a few lines below the new `withBranchName()`
call in `LogicalPlanBuilder#visitInsertTable`.
Unless something downstream is planned for it, suggest dropping the field
and the constructor overload.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
