Skip to content

[fix](fe) Preserve external schema generations and timestamp precision #67904

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Gabriel39 wants to merge 9 commits into
base: master
Choose a base branch
from
Open

[fix](fe) Preserve external schema generations and timestamp precision #67904

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
c70ea3b
fix(paimon): preserve schema and timestamp precision
Gabriel39 Sep 13, 2026
4ae6c48
test(paimon): cover schema-only snapshot reads
Gabriel39 Sep 13, 2026
4ffdc1f
[fix](paimon) Retain bound schema and branch fences through planning
Gabriel39 Sep 13, 2026
75ceac1
[fix](fe) Preserve schema provenance and safe timestamp predicates
Gabriel39 Sep 13, 2026
56cfe11
[fix](fe) Authenticate bound schema reads and preserve partition types
Gabriel39 Sep 13, 2026
1cf784f
[fix](fe) Preserve Iceberg synthetic columns and empty-table schema pins
Gabriel39 Sep 13, 2026
a56a252
[fix](fe) Handle stale Iceberg specs and authenticate Paimon source p…
Gabriel39 Sep 13, 2026
b6f108f
[fix](fe) Validate eagerly pinned external schemas
Gabriel39 Sep 13, 2026
54a5294
[fix](fe) Preserve table identity across cached schema pins
Gabriel39 Sep 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 76 additions & 39 deletions ...or-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.MetadataTableType;
import org.apache.iceberg.MetadataTableUtils;
import org.apache.iceberg.PartitionField;
Expand All @@ -64,6 +65,7 @@
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
Expand All @@ -85,6 +87,7 @@
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
Expand Down Expand Up @@ -113,6 +116,8 @@ public class IcebergConnectorMetadata implements ConnectorMetadata {
// Internal sentinel property carrying a tag/branch ref name from resolveTimeTravel to applySnapshot (the
// typed ConnectorMvccSnapshot has snapshotId/schemaId carriers but no ref field). NOT a BE scan option.
static final String REF_PROPERTY = "iceberg.scan.ref";
private static final String TABLE_IDENTITY_PROPERTY = "iceberg.table.identity";
private static final String PARTITION_SPEC_ID_PROPERTY = "iceberg.partition.spec.id";
private static final String EMPTY_PARTITION_STYLE_PROPERTY = "iceberg.empty.partition.style";

// Iceberg v3 row-lineage hidden columns. Local literal copies of the Doris-side constants — the
Expand Down Expand Up @@ -464,8 +469,8 @@ public ConnectorTableSchema getTableSchema(
/**
* Returns the schema AS OF {@code snapshot.getSchemaId()} (the pinned schema version, for time-travel reads
* under schema evolution), or the LATEST schema when there is no pinned schema id (null snapshot or
* {@code schemaId < 0}). Mirrors legacy {@code IcebergUtils.getSchema}: {@code table.schemas().get(schemaId)}
* when the id is set and a current snapshot exists, else {@code table.schema()}. Shares
* {@code schemaId < 0}). Resolves {@code table.schemas().get(schemaId)} even before the first append,
* since schema-only changes do not create data snapshots. Shares
* {@link #buildTableSchema} with the latest path so the two cannot drift.
*/
@Override
Expand All @@ -483,33 +488,65 @@ public ConnectorTableSchema getTableSchema(
return getTableSchema(session, handle);
}
Table table = loadTable(session, iceHandle);
Schema schema;
if (table.currentSnapshot() == null) {
// Empty table: legacy getSchema falls back to the latest schema (NEWEST_SCHEMA_ID path).
schema = table.schema();
} else {
schema = table.schemas().get((int) snapshot.getSchemaId());
if (schema == null) {
// Defensive: a pinned id absent from table.schemas() (legacy would NPE) -> latest.
// INVARIANT: this SLOT-schema fallback MUST stay identical to the DICT-schema fallback in
// IcebergScanPlanProvider.pinnedSchema (same getSchemaId() lookup + same silent -> table.schema()).
// If the two diverge, the field-id dict names and the BE scan-slot names resolve DIFFERENT
// schemas -> BE children.at() std::out_of_range-SIGABRT on a schema-evolved time-travel read
// (reverify #65185 L16). Do not harden ONE side to throw without the other.
schema = table.schema();
validateSnapshotTable(iceHandle, table, snapshot);
Schema schema = resolvePinnedSchema(table, snapshot);
String specId = snapshot.getProperties().get(PARTITION_SPEC_ID_PROPERTY);
Comment thread
Gabriel39 marked this conversation as resolved.
PartitionSpec spec = specId == null ? table.spec() : table.specs().get(Integer.parseInt(specId));
Comment thread
Gabriel39 marked this conversation as resolved.
Comment thread
Gabriel39 marked this conversation as resolved.
if (spec == null) {
// Keep the legacy missing-history fallback after checking the table identity.
spec = table.spec();
}
return buildTableSchema(iceHandle.getTableName(), table, schema, spec, true);
}

private void validateSnapshotTable(IcebergTableHandle handle, Table table, ConnectorMvccSnapshot snapshot) {
String identity = snapshot.getProperties().get(TABLE_IDENTITY_PROPERTY);
if (identity != null && !identity.equals(tableIdentity(table))) {
// Numeric schema/spec IDs can be reused after recreation. Reject the entire old pin;
// replacing only its schema or spec would still mix the new table with an old data fence.
if (latestSnapshotCache != null) {
latestSnapshotCache.invalidate(TableIdentifier.of(handle.getDbName(), handle.getTableName()));
}
throw new DorisConnectorException("Iceberg table " + handle.getDbName() + "." + handle.getTableName()
+ " identity changed after its snapshot was cached; retry the statement");
}
return buildTableSchema(iceHandle.getTableName(), table, schema, true);
}

private static String tableIdentity(Table table) {
if (table instanceof HasTableOperations) {
TableMetadata metadata = ((HasTableOperations) table).operations().current();
if (metadata.uuid() != null) {
return metadata.uuid();
}
// Legacy V1 metadata may lack a UUID. Only the exact metadata file can safely reuse its IDs.
return "metadata:" + Objects.requireNonNull(metadata.metadataFileLocation(),

@github-actions github-actions Bot Sep 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Do not equate a UUID-less table's identity with its current metadata file. An ordinary Iceberg commit writes a new metadata file for the same table. With REST vended credentials, latestSnapshotCache stays enabled while tableCache is disabled, so the next statement can receive cached M1 coordinates, freshly load the same table at M2, and fail this identity check; active V1 tables then require a retry after every commit. IcebergWriteSchemaContext already handles this case by accepting a retained metadata ancestor. Please disable latest-pin caching for UUID-less tables or validate ancestry, and test a normal same-table commit under the warm-pin/fresh-table cache combination while retaining recreate protection.

"Iceberg table metadata location is unavailable");
}
return Objects.requireNonNull(table.uuid(), "Iceberg table UUID is unavailable").toString();
}

private static Schema resolvePinnedSchema(Table table, ConnectorMvccSnapshot snapshot) {
// An empty table can evolve its schema while a cached snapshot-less pin remains unchanged.
// Slots and handles must honor that schema ID both before and after the first append.
Schema schema = table.schemas().get((int) snapshot.getSchemaId());
// Keep the missing-ID fallback aligned with IcebergScanPlanProvider.pinnedSchema so the
// reader's field-ID dictionary and FE slots cannot resolve different schema generations.
return schema == null ? table.schema() : schema;
}

/**
* Assembles the {@link ConnectorTableSchema} for {@code table} from {@code schema} (the latest schema, or a
* historical schema for a time-travel read). The {@code iceberg.format-version} / {@code location} /
* {@code iceberg.partition-spec} properties are table-level (not schema-versioned). Factored out so the
* latest and at-snapshot paths share ONE assembly.
* historical schema for a time-travel read). Pinned reads also supply the partition spec from their
* metadata generation; table properties and location still come from the loaded table. Factored out
* so the latest and at-snapshot paths share one assembly.
*/
private ConnectorTableSchema buildTableSchema(String tableName, Table table, Schema schema,
boolean appendDataFileMetadataColumns) {
return buildTableSchema(tableName, table, schema, table.spec(), appendDataFileMetadataColumns);
}

private ConnectorTableSchema buildTableSchema(String tableName, Table table, Schema schema,
PartitionSpec spec, boolean appendDataFileMetadataColumns) {
List<ConnectorColumn> columns = parseSchema(schema);

// Iceberg file metadata columns are always available for data tables, but are hidden from
Expand Down Expand Up @@ -552,22 +589,18 @@ private ConnectorTableSchema buildTableSchema(String tableName, Table table, Sch
if (table.location() != null) {
tableProps.put(ConnectorTableSchema.SHOW_LOCATION_KEY, table.location());
}
String partitionClause = buildShowPartitionClause(table);
String partitionClause = buildShowPartitionClause(schema, spec);
if (!partitionClause.isEmpty()) {
tableProps.put(ConnectorTableSchema.SHOW_PARTITION_CLAUSE_KEY, partitionClause);
}
String sortClause = buildShowSortClause(table);
if (!sortClause.isEmpty()) {
tableProps.put(ConnectorTableSchema.SHOW_SORT_CLAUSE_KEY, sortClause);
}
if (!table.spec().isUnpartitioned()) {
// Generic FE partition-column contract: post-cutover, PluginDrivenExternalTable derives the
// table's partition columns SOLELY from a "partition_columns" CSV property (toSchemaCacheValue),
// the same key MaxCompute/paimon emit. Mirror legacy IcebergUtils.loadTableSchemaCacheValue:
// walk the CURRENT spec, resolve each partition field's SOURCE column name (NO identity filter),
// case-preserved to match parseSchema's case-preserved column names (#65094 read-path
// alignment; fromRemoteColumnName is identity for iceberg, so the FE consumer looks the names up
// case-sensitively).
if (!spec.isUnpartitioned()) {
// A cached latest pin can outlive schema-only renames and spec evolution while REST
// credentials require a fresh Table. Resolve partition source IDs in the pinned schema
// and spec so FE never receives historical columns paired with live partition names.
// DEDUPED per source column (LinkedHashSet, first-occurrence order): this CSV becomes a SET of
// partition COLUMNS on the FE side, not a list of spec FIELDS. fe-core maps each name to one scan
// Slot (PruneFileScanPartition) and OneListPartitionEvaluator collects Slot -> literal into an
Expand All @@ -577,8 +610,8 @@ private ConnectorTableSchema buildTableSchema(String tableName, Table table, Sch
// deduped column sequence, so the two stay index-aligned (the arity checkState in
// PluginDrivenMvccExternalTable.toListPartitionItem).
Set<String> partitionColumns = new LinkedHashSet<>();
for (PartitionField field : table.spec().fields()) {
Types.NestedField source = table.schema().findField(field.sourceId());
for (PartitionField field : spec.fields()) {
Types.NestedField source = schema.findField(field.sourceId());
if (source != null) {
partitionColumns.add(source.name());
}
Expand All @@ -601,14 +634,13 @@ private ConnectorTableSchema buildTableSchema(String tableName, Table table, Sch
* {@code bucket[N]}/{@code truncate[W]}/{@code year}/{@code month}/{@code day}/{@code hour} -> the
* matching Doris partition function. Returns "" for an unpartitioned table or no renderable field.
*/
private String buildShowPartitionClause(Table table) {
PartitionSpec spec = table.spec();
private String buildShowPartitionClause(Schema schema, PartitionSpec spec) {
if (spec == null || spec.isUnpartitioned()) {
return "";
}
List<String> fields = new ArrayList<>();
for (PartitionField field : spec.fields()) {
String colName = table.schema().findColumnName(field.sourceId());
String colName = schema.findColumnName(field.sourceId());
if (colName == null) {
continue;
}
Expand Down Expand Up @@ -765,10 +797,8 @@ public Map<String, ConnectorColumnHandle> getColumnHandles(
return getColumnHandles(session, handle);
}
Table table = loadTable(session, iceHandle);
Schema schema = table.currentSnapshot() == null
? table.schema() : table.schemas().get((int) snapshot.getSchemaId());
// Keep the handle-schema fallback identical to getTableSchema so slots and handles cannot diverge.
return buildColumnHandles(schema == null ? table.schema() : schema, true);
validateSnapshotTable(iceHandle, table, snapshot);
return buildColumnHandles(resolvePinnedSchema(table, snapshot), true);
}

@Override
Expand Down Expand Up @@ -2131,6 +2161,12 @@ public Optional<ConnectorMvccSnapshot> beginQuerySnapshot(
: loadLatestSnapshotPin(session, iceHandle);
ConnectorMvccSnapshot.Builder snapshot = ConnectorMvccSnapshot.builder()
.snapshotId(pin.snapshotId).schemaId(pin.schemaId);
if (pin.tableIdentity != null) {
snapshot.property(TABLE_IDENTITY_PROPERTY, pin.tableIdentity);
}
if (pin.specId >= 0) {
snapshot.property(PARTITION_SPEC_ID_PROPERTY, Integer.toString(pin.specId));
Comment thread
Gabriel39 marked this conversation as resolved.
}
if (pin.snapshotId < 0) {
snapshot.property(EMPTY_PARTITION_STYLE_PROPERTY, pin.emptyPartitionStyle.name());
}
Expand Down Expand Up @@ -2177,7 +2213,8 @@ private IcebergLatestSnapshotCache.CachedSnapshot latestSnapshotPin(Table table)
? ConnectorMvccPartitionView.Style.RANGE
: ConnectorMvccPartitionView.Style.UNPARTITIONED;
return new IcebergLatestSnapshotCache.CachedSnapshot(
current == null ? -1L : current.snapshotId(), table.schema().schemaId(), emptyPartitionStyle);
current == null ? -1L : current.snapshotId(), table.schema().schemaId(),
table.spec().specId(), emptyPartitionStyle, tableIdentity(table));
}

/**
Expand Down
18 changes: 16 additions & 2 deletions ...-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLatestSnapshotCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@
* query-begin pin ({@link IcebergConnectorMetadata#beginQuerySnapshot}) reads the SAME snapshot until the
* entry expires or is invalidated by {@code REFRESH TABLE}/{@code REFRESH CATALOG}.
*
* <p><b>Value carries snapshotId, schemaId and the resolved-empty partition style.</b>
* <p><b>Value carries snapshotId, schemaId, specId, table identity and the resolved-empty partition style.</b>
* {@code beginQuerySnapshot} pins the snapshot id <i>and</i> the LATEST schema id
* ({@code table.schema().schemaId()} — not {@code currentSnapshot().schemaId()}, mirroring legacy
* {@code IcebergUtils.getLatestIcebergSnapshot}). A schema-only {@code ALTER} bumps the latest schema id
Expand All @@ -55,10 +55,12 @@
*/
final class IcebergLatestSnapshotCache {

/** Immutable atomic pin for the latest snapshot/schema and its resolved-empty partition style. */
/** Immutable atomic pin for the latest snapshot/schema/spec, table identity and empty partition style. */
static final class CachedSnapshot {
final long snapshotId;
final long schemaId;
final int specId;
final String tableIdentity;
final ConnectorMvccPartitionView.Style emptyPartitionStyle;

CachedSnapshot(long snapshotId, long schemaId) {
Expand All @@ -67,8 +69,20 @@ static final class CachedSnapshot {

CachedSnapshot(long snapshotId, long schemaId,
ConnectorMvccPartitionView.Style emptyPartitionStyle) {
this(snapshotId, schemaId, -1, emptyPartitionStyle);
}

CachedSnapshot(long snapshotId, long schemaId, int specId,
ConnectorMvccPartitionView.Style emptyPartitionStyle) {
this(snapshotId, schemaId, specId, emptyPartitionStyle, null);
}

CachedSnapshot(long snapshotId, long schemaId, int specId,
ConnectorMvccPartitionView.Style emptyPartitionStyle, String tableIdentity) {
this.snapshotId = snapshotId;
this.schemaId = schemaId;
this.tableIdentity = tableIdentity;
this.specId = specId;
this.emptyPartitionStyle = emptyPartitionStyle;
}
}
Expand Down
Loading
Loading