diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java index b933e3da1bd760..4d8d777b9504d3 100644 --- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java @@ -1494,7 +1494,11 @@ private static long lastDdlMillis(Map parameters) { public ConnectorTableSchema getTableSchema(ConnectorSession session, ConnectorTableHandle handle, ConnectorMvccSnapshot snapshot) { if (!(handle instanceof HiveTableHandle)) { - return siblingMetadata(session, handle).getTableSchema(session, handle, snapshot); + // Retained latest schemas shadow the ordinary schema, so inherit the same scan capabilities. + SiblingOwner owner = siblingOwnerResolver.apply(handle); + ConnectorTableSchema schema = memoizedSiblingMetadata(session, owner.connector(), owner.label()) + .getTableSchema(session, handle, snapshot); + return reflectSiblingCapabilities(owner.connector(), schema); } // Hive has no schema-at-snapshot; the SPI default ignores the snapshot and returns the latest schema. return getTableSchema(session, handle); diff --git a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java index a962953a2ed932..5f86273d29602f 100644 --- a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java +++ b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java @@ -403,6 +403,39 @@ public void foreignHandleSchemaReflectsSiblingScanCapabilitiesAsPerTableMarker() "capability reflection must not discard the sibling's write-generation fence"); } + @Test + public void pinnedLatestSchemaReflectsSiblingScanCapabilities() { + // Option C: fe-core's PluginDrivenExternalTable.hasCapability reads only the CATALOG (hive) connector, + // never the embedded sibling — so the hive gateway must reflect the sibling's connector-wide scan + // capabilities onto the delegated schema as a per-table marker, or an iceberg-on-HMS table silently loses + // auto-analyze / Top-N lazy / nested-column prune / storage predicate pruning (all declared connector-wide). + // MUTATION: dropping the reflection -> the returned schema carries no marker -> the embedded table drops the + // capabilities post-flip -> red here. + Set siblingCaps = EnumSet.of( + ConnectorCapability.SUPPORTS_COLUMN_AUTO_ANALYZE, + ConnectorCapability.SUPPORTS_TOPN_LAZY_MATERIALIZE, + ConnectorCapability.SUPPORTS_NESTED_COLUMN_PRUNE, + ConnectorCapability.SUPPORTS_STORAGE_PREDICATE_PRUNING); + HiveConnectorMetadata md = new HiveConnectorMetadata(null, HiveTestProperties.minimal(), new FakeConnectorContext(), + SUPPLIER_MUST_NOT_BE_USED, SUPPLIER_MUST_NOT_BE_USED, + handle -> new SiblingOwner(new CapabilityDeclaringSiblingConnector(siblingCaps), + SiblingOwner.ICEBERG_LABEL)); + + ConnectorTableSchema schema = md.getTableSchema(session, foreignHandle, + ConnectorMvccSnapshot.builder().snapshotId(1).schemaId(0).build()); + Set reflected = schema.getTableCapabilities(); + Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_COLUMN_AUTO_ANALYZE), + "auto-analyze must survive the delegation as a per-table capability"); + Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_TOPN_LAZY_MATERIALIZE), + "Top-N lazy must survive the delegation as a per-table capability"); + Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_NESTED_COLUMN_PRUNE), + "nested-column prune must survive the delegation as a per-table capability"); + Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_STORAGE_PREDICATE_PRUNING), + "storage predicate pruning must survive the delegation as a per-table capability"); + Assertions.assertEquals("sibling-generation", schema.getWriteMetadataIdentity(), + "capability reflection must not discard the sibling's write-generation fence"); + } + @Test public void foreignHandleSchemaReflectsOnlyThePerTableResolvedCapabilitySubset() { // WHY: fe-core resolves only a FIXED subset of capabilities per-table; every other one is answered diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java index 1caeae8fce7f38..85ea0caf2f9850 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java @@ -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; @@ -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; @@ -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; @@ -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 @@ -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 @@ -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); + PartitionSpec spec = specId == null ? table.spec() : table.specs().get(Integer.parseInt(specId)); + 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(), + "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 columns = parseSchema(schema); // Iceberg file metadata columns are always available for data tables, but are hidden from @@ -552,7 +589,7 @@ 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); } @@ -560,14 +597,10 @@ private ConnectorTableSchema buildTableSchema(String tableName, Table table, Sch 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 @@ -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 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()); } @@ -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 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; } @@ -765,10 +797,8 @@ public Map 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 @@ -2129,8 +2159,22 @@ public Optional beginQuerySnapshot( IcebergLatestSnapshotCache.CachedSnapshot pin = latestSnapshotCache != null ? latestSnapshotCache.getOrLoad(id, () -> loadLatestSnapshotPin(session, iceHandle)) : loadLatestSnapshotPin(session, iceHandle); + // Without a UUID, ordinary commits change the only available identity. Do not reuse these + // coordinates across statements; within a statement the frozen table still validates its pin. + if (latestSnapshotCache != null && pin.tableIdentity != null && pin.tableIdentity.startsWith("metadata:")) { + latestSnapshotCache.invalidate(id); + // A concurrent query may have observed the entry before eviction. Resolve from this + // statement's table even on that cache hit; a miss reuses the table it just froze. + pin = 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)); + } if (pin.snapshotId < 0) { snapshot.property(EMPTY_PARTITION_STYLE_PROPERTY, pin.emptyPartitionStyle.name()); } @@ -2177,7 +2221,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)); } /** diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLatestSnapshotCache.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLatestSnapshotCache.java index a20ab5fd2ecf4d..242a2d8ed8f376 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLatestSnapshotCache.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLatestSnapshotCache.java @@ -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}. * - *

Value carries snapshotId, schemaId and the resolved-empty partition style. + *

Value carries snapshotId, schemaId, specId, table identity and the resolved-empty partition style. * {@code beginQuerySnapshot} pins the snapshot id and 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 @@ -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) { @@ -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; } } diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java index b0dd3baaecaedf..1ea0770c1f3b27 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java @@ -26,17 +26,24 @@ import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot; import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.iceberg.BaseTable; import org.apache.iceberg.DataFile; import org.apache.iceberg.DataFiles; import org.apache.iceberg.FileFormat; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; +import org.apache.iceberg.StaticTableOperations; import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableMetadataParser; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.NotFoundException; import org.apache.iceberg.inmemory.InMemoryCatalog; +import org.apache.iceberg.inmemory.InMemoryFileIO; import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.JsonUtil; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -171,6 +178,208 @@ public void beginQuerySnapshotEnabledCachePinsStableAndLoadsOnce() { Assertions.assertEquals(1, loads, "an enabled cache must load the table at most once within the TTL"); } + @Test + public void latestCacheHitKeepsPartitionSchemaAfterLiveRename() { + Table table = dayPartitionedTable(); + RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps(); + ops.table = table; + IcebergLatestSnapshotCache cache = new IcebergLatestSnapshotCache(100, 1000); + IcebergConnectorMetadata firstQuery = new IcebergConnectorMetadata(ops, + IcebergCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot first = firstQuery.beginQuerySnapshot(null, handle()).get(); + table.updateSchema().renameColumn("ts", "renamed_ts").commit(); + table.updateSpec().addField("id").commit(); + IcebergConnectorMetadata nextQuery = new IcebergConnectorMetadata(ops, + IcebergCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot cached = nextQuery.beginQuerySnapshot(null, handle()).get(); + Assertions.assertEquals(first.getSchemaId(), cached.getSchemaId()); + ConnectorTableSchema schema = nextQuery.getTableSchema(null, handle(), cached); + Assertions.assertTrue(columnNames(schema).contains("ts")); + Assertions.assertEquals("ts", schema.getProperties().get(ConnectorTableSchema.PARTITION_COLUMNS_KEY)); + Assertions.assertEquals("PARTITION BY LIST (DAY(`ts`)) ()", + schema.getProperties().get(ConnectorTableSchema.SHOW_PARTITION_CLAUSE_KEY)); + } + + @Test + public void emptyTableCacheHitKeepsSchemaAndHandlesThroughFirstAppend() { + InMemoryCatalog catalog = new InMemoryCatalog(); + catalog.initialize("test", Collections.emptyMap()); + catalog.createNamespace(Namespace.of("db1")); + PartitionSpec spec = PartitionSpec.builderFor(PARTITIONED_SCHEMA).day("ts").build(); + Table table = catalog.createTable(TableIdentifier.of("db1", "t1"), PARTITIONED_SCHEMA, spec); + RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps(); + ops.table = table; + IcebergLatestSnapshotCache cache = new IcebergLatestSnapshotCache(100, 1000); + IcebergCatalogProperties properties = IcebergCatalogProperties.of(Collections.emptyMap()); + IcebergConnectorMetadata firstQuery = new IcebergConnectorMetadata( + ops, properties, new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot first = firstQuery.beginQuerySnapshot(null, handle()).get(); + Assertions.assertEquals(-1L, first.getSnapshotId()); + Assertions.assertTrue(first.getSchemaId() >= 0); + table.updateSchema().renameColumn("ts", "renamed_ts").commit(); + Assertions.assertNull(table.currentSnapshot()); + + // Keep only the latest-pin cache warm; each query reloads live metadata as vended catalogs do. + for (int stage = 0; stage < 2; stage++) { + IcebergConnectorMetadata nextQuery = new IcebergConnectorMetadata( + ops, properties, new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot cached = nextQuery.beginQuerySnapshot(null, handle()).get(); + Assertions.assertEquals(first.getSnapshotId(), cached.getSnapshotId()); + Assertions.assertEquals(first.getSchemaId(), cached.getSchemaId()); + ConnectorTableSchema schema = nextQuery.getTableSchema(null, handle(), cached); + Assertions.assertAll( + () -> Assertions.assertTrue(columnNames(schema).contains("ts")), + () -> Assertions.assertFalse(columnNames(schema).contains("renamed_ts")), + () -> Assertions.assertEquals("ts", + schema.getProperties().get(ConnectorTableSchema.PARTITION_COLUMNS_KEY)), + () -> Assertions.assertEquals("PARTITION BY LIST (DAY(`ts`)) ()", + schema.getProperties().get(ConnectorTableSchema.SHOW_PARTITION_CLAUSE_KEY)), + () -> Assertions.assertTrue(nextQuery.getColumnHandles(null, handle(), cached).containsKey("ts")), + () -> Assertions.assertFalse(nextQuery.getColumnHandles(null, handle(), cached) + .containsKey("renamed_ts"))); + if (stage == 0) { + table.newAppend().appendFile(DataFiles.builder(spec) + .withPath("s3://bucket/db1/t1/first.parquet").withFileSizeInBytes(100).withRecordCount(1) + .withPartitionPath("ts_day=1970-04-11").withFormat(FileFormat.PARQUET).build()).commit(); + } + } + } + + @Test + public void emptyTableMissingPinnedSchemaFallsBackForSchemaAndHandles() { + InMemoryCatalog catalog = new InMemoryCatalog(); + catalog.initialize("test", Collections.emptyMap()); + catalog.createNamespace(Namespace.of("db1")); + Table table = catalog.createTable( + TableIdentifier.of("db1", "t1"), SCHEMA_V0, PartitionSpec.unpartitioned()); + IcebergConnectorMetadata metadata = metadataFor(table, new RecordingIcebergCatalogOps()); + ConnectorMvccSnapshot missing = ConnectorMvccSnapshot.builder().snapshotId(-1).schemaId(999).build(); + Assertions.assertEquals(columnNames(metadata.getTableSchema(null, handle())), + columnNames(metadata.getTableSchema(null, handle(), missing))); + Assertions.assertEquals(metadata.getColumnHandles(null, handle()).keySet(), + metadata.getColumnHandles(null, handle(), missing).keySet()); + } + + @Test + public void latestCacheRejectsRecreatedTableWithMissingSpec() { + checkRecreatedTablePin(false, false); + } + + @Test + public void latestCacheRejectsRecreatedTableWithReusedSpec() { + checkRecreatedTablePin(true, false); + } + + @Test + public void latestCacheRejectsRecreatedTableWithReusedSchemaAndSpec() { + checkRecreatedTablePin(true, true); + } + + private void checkRecreatedTablePin(boolean reuseSpec, boolean reuseSchema) { + InMemoryCatalog catalog = new InMemoryCatalog(); + catalog.initialize("test", Collections.emptyMap()); + catalog.createNamespace(Namespace.of("db1")); + TableIdentifier id = TableIdentifier.of("db1", "t1"); + Table original = catalog.createTable(id, PARTITIONED_SCHEMA, + PartitionSpec.builderFor(PARTITIONED_SCHEMA).day("ts").build()); + if (!reuseSchema) { + original.updateSchema().renameColumn("ts", "old_ts").commit(); + } + original.updateSpec().addField("id").commit(); + RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps(); + ops.table = original; + IcebergLatestSnapshotCache cache = new IcebergLatestSnapshotCache(100, 1000); + IcebergCatalogProperties properties = IcebergCatalogProperties.of(Collections.emptyMap()); + ConnectorMvccSnapshot first = new IcebergConnectorMetadata( + ops, properties, new RecordingConnectorContext(), cache).beginQuerySnapshot(null, handle()).get(); + Assertions.assertEquals("1", first.getProperties().get("iceberg.partition.spec.id")); + catalog.dropTable(id, false); + Schema replacementSchema = new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "new_ts", Types.TimestampType.withoutZone())); + ops.table = catalog.createTable(id, replacementSchema, + PartitionSpec.builderFor(replacementSchema).day("new_ts").build()); + if (reuseSpec) { + ops.table.updateSpec().addField("id").commit(); + ops.table.updateSpec().removeField("new_ts_day").commit(); + Assertions.assertEquals(2, ops.table.spec().specId()); + } + IcebergConnectorMetadata nextQuery = new IcebergConnectorMetadata( + ops, properties, new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot cached = nextQuery.beginQuerySnapshot(null, handle()).get(); + Assertions.assertEquals(first.getProperties(), cached.getProperties()); + Assertions.assertEquals(reuseSchema, ops.table.schemas().containsKey((int) cached.getSchemaId())); + Assertions.assertEquals(reuseSpec, ops.table.specs().containsKey(1)); + Assertions.assertTrue(Assertions.assertThrows(DorisConnectorException.class, + () -> nextQuery.getTableSchema(null, handle(), cached)).getMessage().contains("retry the statement")); + Assertions.assertThrows(DorisConnectorException.class, () -> nextQuery.getColumnHandles(null, handle(), cached)); + // A failed statement must evict the old pin so a retry recovers without REFRESH TABLE. + ConnectorMvccSnapshot retry = nextQuery.beginQuerySnapshot(null, handle()).get(); + Assertions.assertEquals(ops.table.schema().schemaId(), retry.getSchemaId()); + Assertions.assertEquals(Integer.toString(ops.table.spec().specId()), + retry.getProperties().get("iceberg.partition.spec.id")); + ConnectorTableSchema schema = nextQuery.getTableSchema(null, handle(), retry); + Assertions.assertTrue(columnNames(schema).contains("new_ts")); + Assertions.assertEquals(reuseSpec ? "id" : "new_ts", + schema.getProperties().get(ConnectorTableSchema.PARTITION_COLUMNS_KEY)); + Assertions.assertTrue(nextQuery.getColumnHandles(null, handle(), retry).containsKey("new_ts")); + } + + @Test + public void latestCacheRejectsChangedUuidlessMetadata() throws Exception { + TableMetadata metadata = TableMetadata.newTableMetadata(SCHEMA_V0, PartitionSpec.unpartitioned(), + "s3://bucket/table", Collections.singletonMap("format-version", "1")); + ObjectNode json = (ObjectNode) JsonUtil.mapper().readTree(TableMetadataParser.toJson(metadata)); + json.remove("table-uuid"); + RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps(); + ops.table = new BaseTable(new StaticTableOperations( + TableMetadataParser.fromJson("s3://bucket/table/metadata/first.json", json), new InMemoryFileIO()), + "db1.t1"); + IcebergLatestSnapshotCache cache = new IcebergLatestSnapshotCache(100, 1000); + IcebergConnectorMetadata reader = new IcebergConnectorMetadata(ops, + IcebergCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot pin = reader.beginQuerySnapshot(null, handle()).get(); + Assertions.assertTrue(reader.getColumnHandles(null, handle(), pin).containsKey("id")); + Assertions.assertEquals(0, cache.size(), "UUID-less coordinates must not be cached across statements"); + // Without a UUID, even matching numeric IDs cannot justify crossing metadata-file identities. + ops.table = new BaseTable(new StaticTableOperations( + TableMetadataParser.fromJson("s3://bucket/table/metadata/replacement.json", json), new InMemoryFileIO()), + "db1.t1"); + Assertions.assertThrows(DorisConnectorException.class, () -> reader.getTableSchema(null, handle(), pin)); + ConnectorMvccSnapshot retry = reader.beginQuerySnapshot(null, handle()).get(); + Assertions.assertTrue(reader.getColumnHandles(null, handle(), retry).containsKey("id")); + } + + @Test + public void uuidlessOrdinaryCommitDoesNotReuseLatestCoordinates() throws Exception { + TableMetadata metadata = TableMetadata.newTableMetadata(SCHEMA_V0, PartitionSpec.unpartitioned(), + "s3://bucket/table", Collections.singletonMap("format-version", "1")); + ObjectNode json = (ObjectNode) JsonUtil.mapper().readTree(TableMetadataParser.toJson(metadata)); + json.remove("table-uuid"); + String firstLocation = "s3://bucket/table/metadata/first.json"; + RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps(); + ops.table = new BaseTable(new StaticTableOperations( + TableMetadataParser.fromJson(firstLocation, json), new InMemoryFileIO()), "db1.t1"); + IcebergLatestSnapshotCache cache = new IcebergLatestSnapshotCache(100, 1000); + IcebergCatalogProperties props = IcebergCatalogProperties.of(Collections.emptyMap()); + ConnectorMvccSnapshot first = new IcebergConnectorMetadata(ops, props, new RecordingConnectorContext(), cache) + .beginQuerySnapshot(null, handle()).get(); + // A normal property commit retains the prior metadata in its ancestry and advances its file. + json.putArray("metadata-log").addObject() + .put("timestamp-ms", json.get("last-updated-ms").asLong()).put("metadata-file", firstLocation); + json.put("last-updated-ms", json.get("last-updated-ms").asLong() + 1); + ((ObjectNode) json.get("properties")).put("read.split.target-size", "67108864"); + ops.table = new BaseTable(new StaticTableOperations( + TableMetadataParser.fromJson("s3://bucket/table/metadata/next.json", json), new InMemoryFileIO()), + "db1.t1"); + IcebergConnectorMetadata nextQuery = new IcebergConnectorMetadata( + ops, props, new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot next = nextQuery.beginQuerySnapshot(null, handle()).get(); + Assertions.assertNotEquals(first.getProperties(), next.getProperties()); + Assertions.assertTrue(nextQuery.getColumnHandles(null, handle(), next).containsKey("id")); + Assertions.assertEquals(0, cache.size()); + } + @Test public void beginQuerySnapshotDisabledCacheLoadsEveryCall() { Fixture f = fixture(); diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java index 0198d7ed5e5781..c60b10f006269c 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java @@ -205,17 +205,44 @@ public long schemaId() { * offline without faking a concrete paimon {@code TableSchema}. */ final class PaimonSchemaSnapshot { + private final long schemaId; + private final String fileDigest; private final List fields; private final List partitionKeys; private final List primaryKeys; public PaimonSchemaSnapshot(List fields, List partitionKeys, List primaryKeys) { + this(-1L, fields, partitionKeys, primaryKeys); + } + + public PaimonSchemaSnapshot(long schemaId, List fields, List partitionKeys, + List primaryKeys) { + this(schemaId, fields, partitionKeys, primaryKeys, null); + } + + private PaimonSchemaSnapshot(TableSchema schema) { + this(schema.id(), schema.fields(), schema.partitionKeys(), schema.primaryKeys(), + PaimonSchemaPin.schemaDigest(schema)); + } + + private PaimonSchemaSnapshot(long schemaId, List fields, List partitionKeys, + List primaryKeys, String fileDigest) { + this.schemaId = schemaId; + this.fileDigest = fileDigest; this.fields = fields; this.partitionKeys = partitionKeys; this.primaryKeys = primaryKeys; } + String fileDigest() { + return fileDigest; + } + + public long schemaId() { + return schemaId; + } + /** The schema's fields ({@code tableSchema.fields()}). */ public List fields() { return fields; @@ -367,8 +394,7 @@ public PaimonSchemaSnapshot schemaAt(Table table, long schemaId) { // schemaManager() is only on DataTable. schema(schemaId) is the historical TableSchema // (legacy PaimonExternalTable.initSchema(schemaId) reads the same accessors). TableSchema tableSchema = ((DataTable) table).schemaManager().schema(schemaId); - return new PaimonSchemaSnapshot( - tableSchema.fields(), tableSchema.partitionKeys(), tableSchema.primaryKeys()); + return new PaimonSchemaSnapshot(tableSchema); } @Override @@ -381,7 +407,7 @@ public Optional latestSchema(Table table) { return Optional.empty(); } return ((DataTable) table).schemaManager().latest() - .map(s -> new PaimonSchemaSnapshot(s.fields(), s.partitionKeys(), s.primaryKeys())); + .map(PaimonSchemaSnapshot::new); } @Override diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java index 62113194c76895..09b9e952f59bfb 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java @@ -65,6 +65,7 @@ import java.util.Optional; import java.util.OptionalLong; import java.util.Set; +import java.util.function.Supplier; /** * {@link ConnectorMetadata} implementation for Paimon. @@ -100,6 +101,12 @@ public class PaimonConnectorMetadata implements ConnectorMetadata { // existing direct-construction tests compile unchanged; production goes through the 5-arg ctor. private final PaimonLatestSnapshotCache latestSnapshotCache; + // Metadata is statement-scoped: aliases sharing a data fence must also share one schema generation. + private final Map> statementSchemas = + new java.util.concurrent.ConcurrentHashMap<>(); + private final Map> statementSchemaPins = + new java.util.concurrent.ConcurrentHashMap<>(); + // PERF-06: cross-query DERIVED partition-view cache A (generic ConnectorMetadataCache), injected by the // owning PaimonConnector; null = no cross-query derived layer (the convenience/test ctors used by ~15 // existing direct-construction tests pass null). Layered ABOVE the raw remote catalogOps.listPartitions @@ -251,7 +258,8 @@ public ConnectorTableSchema getTableSchema( // tables (isSystemTable()) always keep their synthetic rowType() (no schema-version history; some // are not DataTable). Sharing buildTableSchema with the at-snapshot path keeps the two from drifting. if (!paimonHandle.isSystemTable()) { - Optional latest = catalogOps.latestSchema(table); + Optional latest = + readSchemaAuthenticated(() -> catalogOps.latestSchema(table)); if (latest.isPresent()) { PaimonCatalogOps.PaimonSchemaSnapshot schema = latest.get(); return buildTableSchema( @@ -307,14 +315,8 @@ public ConnectorTableSchema getTableSchema( // resolved table -- and its schemaAt read -- is byte-for-byte unchanged. PaimonTableHandle pinned = (PaimonTableHandle) applySnapshot(session, paimonHandle, snapshot); Table table = resolveTable(pinned); - // FIX-B-MC2: memoize the schemaAt schema-file read across queries. resolveTable + buildTableSchema - // still run every query (keeping the live coreOptions/properties current); only the schemaAt - // round-trip is skipped on a repeat. The memo is keyed by (pinned-handle-identity, schemaId) -- a - // pure function -- and owned by the per-catalog PaimonConnector. Key on the PINNED handle (which - // carries branchName in equals/hashCode) so a branch@schemaId and a base@same-schemaId cannot - // collide in this long-lived memo. resolveTable runs ONCE, outside the loader. - PaimonCatalogOps.PaimonSchemaSnapshot schema = - schemaAtMemo.getOrLoad(pinned, schemaId, () -> catalogOps.schemaAt(table, schemaId)); + // Branch identity must be applied before consulting either statement or historical schemas. + PaimonCatalogOps.PaimonSchemaSnapshot schema = schemaForPin(pinned, table, schemaId); return buildTableSchema( paimonHandle.getTableName(), table, @@ -365,8 +367,12 @@ private Table resolveSystemTableAt(ConnectorSession session, PaimonTableHandle pinned = snapshot == null ? paimonHandle : (PaimonTableHandle) applySnapshot(session, paimonHandle, snapshot); - Table table = resolveTable(pinned); Map scanOptions = pinned.getScanOptions(); + if (PaimonScanParams.preservesBoundSchema(scanOptions)) { + // Schema-derived wrappers must be built over the bound source before exposing their fields. + return new PaimonScanPlanProvider(catalogProperties, catalogOps, context).resolveScanTable(pinned); + } + Table table = resolveTable(pinned); if (scanOptions != null && !scanOptions.isEmpty() && PaimonScanParams.isOptionsPin(scanOptions)) { return PaimonScanParams.applyOptions(table, scanOptions); } @@ -577,7 +583,71 @@ public Optional beginQuerySnapshot( Identifier identifier = Identifier.create(paimonHandle.getDatabaseName(), paimonHandle.getTableName()); long id = latestSnapshotCache.getOrLoad(identifier, () -> catalogOps.latestSnapshotId(resolveTable(paimonHandle)).orElse(-1L)); - return Optional.of(ConnectorMvccSnapshot.builder().snapshotId(id).build()); + Table table = resolveTable(paimonHandle); + long schemaId = statementSchemaId(paimonHandle, table); + Map coordinates = new HashMap<>(captureSchemaPin(paimonHandle, table, schemaId, id)); + if (!coordinates.isEmpty()) { + coordinates.putAll(PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), id)); + } + return Optional.of(ConnectorMvccSnapshot.builder().snapshotId(id).schemaId(schemaId) + .properties(coordinates).build()); + } + + private T readSchemaAuthenticated(Supplier read) { + // Cached tables do not cache schema files: exact/latest schema reads still need plugin UGI and TCCL. + try { + return context.executeAuthenticated(read::get); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException("Failed to read Paimon schema", e); + } + } + + private Map captureSchemaPin( + PaimonTableHandle handle, Table table, long schemaId, long snapshotId) { + // Aliases and planning-only selectors share the first physical pin, not a later reload of the same name. + Map pin = statementSchemaPins.computeIfAbsent(handle, + ignored -> readSchemaAuthenticated(() -> PaimonSchemaPin.capture(table, schemaId, snapshotId))); + // The schema read and file capture are separate I/O operations; reject recreation between them. + statementSchemas.getOrDefault(handle, Optional.empty()).ifPresent(schema -> + PaimonSchemaPin.validateSchema(schema, pin)); + return pin; + } + + private long statementSchemaId(PaimonTableHandle handle, Table table) { + return statementSchemas.computeIfAbsent(handle, + ignored -> readSchemaAuthenticated(() -> catalogOps.latestSchema(table))) + .map(PaimonCatalogOps.PaimonSchemaSnapshot::schemaId) + .orElse(-1L); + } + + private PaimonCatalogOps.PaimonSchemaSnapshot schemaForPin(PaimonTableHandle handle, Table table, long schemaId) { + // External recreation can reuse a schema ID. Latest pins must retain the actual statement + // schema instead of consulting the name/ID-keyed historical memo from an earlier table. + readSchemaAuthenticated(() -> { + PaimonSchemaPin.validate(table, handle.getScanOptions()); + return null; + }); + Optional captured = + statementSchemas.getOrDefault(handle, Optional.empty()); + if (captured.isPresent() && captured.get().schemaId() == schemaId) { + PaimonSchemaPin.validateSchema(captured.get(), handle.getScanOptions()); + return captured.get(); + } + if (PaimonScanParams.preservesBoundSchema(handle.getScanOptions())) { + // INSERT retries retain the source MVCC pin while replacing this metadata scope. + // Rehydrate its exact schema once in the new scope, never through the historical memo. + PaimonCatalogOps.PaimonSchemaSnapshot restored = statementSchemas.compute(handle, (key, previous) -> + previous != null && previous.isPresent() && previous.get().schemaId() == schemaId + ? previous + : Optional.of(readSchemaAuthenticated(() -> catalogOps.schemaAt(table, schemaId)))) + .get(); + PaimonSchemaPin.validateSchema(restored, handle.getScanOptions()); + return restored; + } + return schemaAtMemo.getOrLoad(handle, schemaId, + () -> readSchemaAuthenticated(() -> catalogOps.schemaAt(table, schemaId))); } @Override @@ -619,12 +689,12 @@ public boolean usesStatementSnapshotForOptions( * loads the branch as its OWN table (independent schema/snapshots, via the 3-arg branch * Identifier through {@link PaimonTableHandle#withBranch}) and pins its LATEST snapshot — * branches have NO in-branch time-travel (legacy {@code PaimonExternalTable} reads the - * branch's {@code latestSnapshot()} only). The branch identity is carried to + * branch's {@code latestSnapshot()} only). The current schema id is captured independently + * because it can advance without a data snapshot. The branch identity is carried to * {@link #applySnapshot} via an internal sentinel ({@code CoreOptions.BRANCH} key, NOT a - * scan-copy option); no {@code scan.snapshot-id} is pinned (the branch reads its own latest). - * An empty branch (no snapshot) pins {@code snapshotId=-1} and {@code schemaId=-1}: a benign - * divergence from legacy's {@code schemaId=0L} — the resulting schema is identical (both - * resolve to the branch's current schema), mirroring the INCREMENTAL empty-table -1 note. + * scan-copy option), together with the resolved data fence. + * An empty branch also pins {@code snapshotId=-1}; both empty and non-empty branches bind + * against the current branch schema. * * *

CONTRACT DIFFERENCE (intentional, documented): legacy {@code PaimonUtil} THREW a @@ -733,19 +803,18 @@ public Optional resolveTimeTravel( // latestSnapshot() only). Table branchTable = resolveTable(paimonHandle.withBranch(branchName)); long snapshotId = catalogOps.latestSnapshotId(branchTable).orElse(-1L); - long schemaId = snapshotId < 0 - ? -1L - : catalogOps.snapshotSchemaId(branchTable, snapshotId).orElse(-1L); + // A schema-only ALTER advances the branch schema without creating a data snapshot. + // Bind the branch's exact current schema independently of its latest data snapshot. + long schemaId = statementSchemaId(paimonHandle.withBranch(branchName), branchTable); // Carry the branch identity to applySnapshot via an internal sentinel // (CoreOptions.BRANCH key). Branch is a handle-IDENTITY change, not a scan-copy // option: applySnapshot reads this sentinel and routes it to handle.withBranch (it is - // never threaded into Table.copy). No scan.snapshot-id is pinned (the branch table - // natively reads its own latest). + // never threaded into Table.copy). The data fence is applied after changing identity. + Map coordinates = captureSchemaPin( + paimonHandle.withBranch(branchName), branchTable, schemaId, snapshotId); return Optional.of(ConnectorMvccSnapshot.builder() - .snapshotId(snapshotId) - .schemaId(schemaId) - .property(CoreOptions.BRANCH.key(), branchName) - .build()); + .snapshotId(snapshotId).schemaId(schemaId).properties(coordinates) + .property(CoreOptions.BRANCH.key(), branchName).build()); } case OPTIONS: { // @options carries paimon's OWN scan-option vocabulary. Validate the keys, then RESOLVE @@ -786,9 +855,12 @@ public Optional resolveTimeTravel( : pinnedSnapshotId(table, resolved); // The statement fence pins data visibility, not schema time travel. Planning-only // aliases must retain the latest-schema projection used by the plain relation. - long schemaId = usesStatementFence || pinnedId < 0 - ? -1L - : catalogOps.snapshotSchemaId(table, pinnedId).orElse(-1L); + long schemaId = usesStatementFence + ? statementSchemaId(paimonHandle, table) + : pinnedId < 0 ? -1L : catalogOps.snapshotSchemaId(table, pinnedId).orElse(-1L); + if (usesStatementFence) { + resolved.putAll(captureSchemaPin(paimonHandle, table, schemaId, pinnedId)); + } // resolved is never empty for a startup selector; for a selector-free @options (e.g. only // scan.manifest-parallelism) it is the user map verbatim, which applySnapshot still // threads -- those keys tune HOW the scan runs, not WHICH version it reads. @@ -905,8 +977,8 @@ private long parseTimestampMillis(ConnectorSession session, ConnectorTimeTravelS *

Threads the FULL {@code snapshot.getProperties()} map: this may be * {@code scan.snapshot-id=} (snapshot-id / timestamp time-travel) OR * {@code scan.tag-name=} (tag time-travel), whichever {@link #resolveTimeTravel} pinned. - * When {@code properties} is empty (the {@link #beginQuerySnapshot} latest-pin path, which - * carries no properties) it falls back to {@code scan.snapshot-id=} for B5a parity. + * Latest pins also carry immutable schema/snapshot file coordinates for generation validation. + * Empty properties retain the legacy latest-pin interpretation, including its bound schema. * *

BRANCH is special: when the snapshot carries the {@code CoreOptions.BRANCH} sentinel (set by * {@link #resolveTimeTravel}'s BRANCH case), it is a handle-IDENTITY change, not a scan option — @@ -932,36 +1004,34 @@ public ConnectorTableHandle applySnapshot(ConnectorSession session, return paimonHandle; } PaimonScanParams.validateSystemTableOptions(snapshot.getProperties()); - return paimonHandle.withScanOptions(snapshot.getProperties()); + return paimonHandle.withScanOptions(PaimonScanParams.withBoundSchema( + snapshot.getProperties(), snapshot.getSchemaId())); } if (snapshot != null) { String branch = snapshot.getProperties().get(CoreOptions.BRANCH.key()); if (branch != null) { - // Branch time-travel is a handle-identity change (a different table load), not a scan - // option: route to withBranch (which clears the transient base Table so resolveTable - // reloads the branch). The branch reads its own latest, so no scan.snapshot-id is - // pinned. Detected BEFORE the generic properties path so the branch sentinel never - // becomes a scan-copy option. - return paimonHandle.withBranch(branch); + // Branch identity and data visibility are independent: switching tables must not + // discard the resolved positive or empty fence when a branch commits during planning. + Map options = new HashMap<>(snapshot.getProperties()); + options.remove(CoreOptions.BRANCH.key()); + options.putAll(PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), snapshot.getSnapshotId())); + return paimonHandle.withBranch(branch).withScanOptions( + PaimonScanParams.withBoundSchema(options, snapshot.getSchemaId())); } if (!snapshot.getProperties().isEmpty()) { - // Explicit time-travel: the connector already resolved the exact scan options - // (scan.snapshot-id OR scan.tag-name etc.) in resolveTimeTravel — thread them verbatim. - return paimonHandle.withScanOptions(snapshot.getProperties()); + // Both latest and time-travel pins already carry their resolved scan options. + // Preserve the generation coordinates alongside those selectors. + return paimonHandle.withScanOptions(PaimonScanParams.withBoundSchema( + snapshot.getProperties(), snapshot.getSchemaId())); } } if (snapshot == null) { return paimonHandle; } - if (snapshot.getSnapshotId() < 0) { - // Empty latest is still a statement-scoped state. Carry only Doris' internal marker; - // Paimon's scan.snapshot-id=-1 would address a non-existent snapshot file. - return paimonHandle.withScanOptions( - PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), -1L)); - } - Map scanOptions = Collections.singletonMap( - CoreOptions.SCAN_SNAPSHOT_ID.key(), String.valueOf(snapshot.getSnapshotId())); - return paimonHandle.withScanOptions(scanOptions); + // The latest statement fence owns both axes, even if no data snapshot exists yet. + return paimonHandle.withScanOptions(PaimonScanParams.withBoundSchema( + PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), snapshot.getSnapshotId()), + snapshot.getSchemaId())); } /** @@ -1182,7 +1252,8 @@ public Map getColumnHandles( // (aborts the BE). latestSchema() is empty for a non-DataTable/schema-less backend -> fall back to // rowType(). System tables keep their synthetic rowType() (no schema-version history). if (!paimonHandle.isSystemTable()) { - Optional latest = catalogOps.latestSchema(table); + Optional latest = + readSchemaAuthenticated(() -> catalogOps.latestSchema(table)); if (latest.isPresent()) { return buildColumnHandles(latest.get().fields(), true); } @@ -1196,9 +1267,9 @@ public Map getColumnHandles( * ({@link #getColumnHandles(ConnectorSession, ConnectorTableHandle)}) when there is no pinned * schema id (null snapshot or {@code schemaId < 0}). * - *

Keys the handles by the PINNED names via the SAME memoized {@link PaimonCatalogOps#schemaAt} - * read the at-snapshot {@link #getTableSchema(ConnectorSession, ConnectorTableHandle, - * ConnectorMvccSnapshot)} uses, so the handle names equal the pinned Doris schema the query slots + *

Keys handles by the same captured latest schema or memoized historical schema used by + * {@link #getTableSchema(ConnectorSession, ConnectorTableHandle, ConnectorMvccSnapshot)}, + * so the handle names equal the pinned Doris schema the query slots * were bound to. Without this, a time-travel read across a RENAME would key the handles by the * latest names, the renamed column's pinned-name slot would miss the map and be silently dropped, * and the paimon field-id dict would omit that BE scan slot -> BE StructNode out_of_range crash.

@@ -1230,11 +1301,8 @@ public Map getColumnHandles( // version/tag/time pin only threads scan options resolveTable ignores -> table unchanged. PaimonTableHandle pinned = (PaimonTableHandle) applySnapshot(session, paimonHandle, snapshot); Table table = resolveTable(pinned); - // Key the memo on the PINNED handle (carries branchName in equals/hashCode): schemaAtMemo is - // per-catalog and long-lived, so keying on the base handle would let a branch@schemaId poison a - // later base@same-schemaId read (each has its own independently-evolved schema-). - PaimonCatalogOps.PaimonSchemaSnapshot schema = - schemaAtMemo.getOrLoad(pinned, schemaId, () -> catalogOps.schemaAt(table, schemaId)); + // Use the same captured schema as slot binding, including its branch identity. + PaimonCatalogOps.PaimonSchemaSnapshot schema = schemaForPin(pinned, table, schemaId); return buildColumnHandles(schema.fields(), true); } @@ -1352,11 +1420,9 @@ private List cachedPartitions(PaimonTableHandle paimonHa * and a new snapshot (data change, once the entry expires or REFRESH invalidates it) naturally mints a new key. * *

schemaId: pinned {@code -1} ("unversioned" for that axis, matching - * {@link ConnectorTableKey}'s documented convention). Unlike iceberg, paimon's {@link PaimonTableHandle} - * carries no schemaId — {@code applySnapshot} threads only {@code scanOptions} (an opaque properties map; - * see its javadoc) onto the handle, and {@link #beginQuerySnapshot} (the common latest-pin path) never - * resolves a schemaId either (its {@code ConnectorMvccSnapshot} keeps the builder default {@code -1}). This - * is not a loss for THIS view: {@link #collectPartitions} derives its output from {@code partitionKeys} + * {@link ConnectorTableKey}'s documented convention). Statement-fenced handles bypass this cache. + * Unversioned partition views do not need a schema generation: {@link #collectPartitions} derives + * its output from {@code partitionKeys} * (fixed at handle-build time) and paimon's raw partition specs, and paimon partition columns are immutable * post-creation, so schema evolution (e.g. ADD COLUMN) does not change what this method computes. */ @@ -1601,6 +1667,10 @@ private Table runtimeSafeSystemTable( return systemTable; } Table dataTable = PaimonTableResolver.resolveSystemSource(catalogOps, handle, context); + if (PaimonScanParams.preservesBoundSchema(scanOptions)) { + return readSchemaAuthenticated(() -> PaimonReaderOptions.runtimeSafeSystemTable( + handle.getSysTableName(), systemTable, dataTable, scanOptions)); + } return PaimonReaderOptions.runtimeSafeSystemTable( handle.getSysTableName(), systemTable, dataTable, scanOptions); } diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java index 5fb90b3beb9732..b959a88a12a027 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java @@ -38,11 +38,11 @@ import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypeRoot; import org.apache.paimon.types.RowType; +import org.apache.paimon.types.TimestampType; import java.math.BigDecimal; import java.time.LocalDate; import java.time.LocalDateTime; -import java.time.ZoneOffset; import java.util.ArrayList; import java.util.List; import java.util.stream.Collectors; @@ -329,15 +329,15 @@ private Object convertLiteralValue(ConnectorLiteral literal, DataType paimonType } return null; case TIMESTAMP_WITHOUT_TIME_ZONE: - // Zone-free type: interpret the literal's wall-clock in UTC to match paimon's - // stored min/max file/partition stats (computed by reading the wall clock as UTC). - // Mirrors legacy PaimonValueConverter#visit(TimestampType), which uses a fixed - // GMT Calendar. Using the session zone here would shift the epoch-millis vs the - // stored stats and risk false file/partition pruning = silent data loss. + // Doris truncates source nanoseconds to DATETIMEV2(6). Exact source comparisons + // would reject rows that become equal after that truncation, so keep them residual. + if (((TimestampType) paimonType).getPrecision() > 6) { + return null; + } + // Preserve the complete wall-clock value: narrowing it to epoch milliseconds can + // make Paimon prune every file matching a non-millisecond-aligned predicate. if (value instanceof LocalDateTime) { - LocalDateTime dt = (LocalDateTime) value; - long millis = dt.toInstant(ZoneOffset.UTC).toEpochMilli(); - return Timestamp.fromEpochMillis(millis); + return Timestamp.fromLocalDateTime((LocalDateTime) value); } return null; case TIMESTAMP_WITH_LOCAL_TIME_ZONE: diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java index 45b3ea75f9ac00..b474ea804330dd 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java @@ -24,6 +24,9 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.options.ConfigOption; import org.apache.paimon.options.FallbackKey; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FallbackReadFileStoreTable; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.Table; import org.apache.paimon.table.source.snapshot.FullCompactedStartingScanner; @@ -33,6 +36,7 @@ import java.util.HashMap; import java.util.Locale; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; @@ -61,6 +65,7 @@ public final class PaimonScanParams { private static final String PINNED_FILE_CREATION_TIME = "doris.internal.paimon.file-creation-time-millis"; private static final String PINNED_EMPTY_SCAN = "doris.internal.paimon.empty-scan"; + private static final String BOUND_SCHEMA_ID = "doris.internal.paimon.bound-schema-id"; private static final String PRESERVE_BOUND_SCHEMA = "doris.internal.paimon.preserve-bound-schema"; /** @@ -196,14 +201,55 @@ public static FileStoreTable applyOptionsWithoutTimeTravel( .filter(key -> !tableOptions.containsKey(key)) .forEach(key -> isolatedOptions.put(key, null)); } - // The statement fence already selected the schema generation. Preserve that generation - // while carrying only the resolved read selector and execution options into this copy. + String schemaId = options.get(BOUND_SCHEMA_ID); + PaimonSchemaPin.validate(table, options); + if (schemaId != null) { + table = restoreBoundSchema(table, Long.parseLong(schemaId), options, ""); + } FileStoreTable effectiveTable = (FileStoreTable) PaimonReaderOptions.runtimeSafeTable( table.copyWithoutTimeTravel(isolatedOptions)); PaimonReaderOptions.validateEffectiveTable(effectiveTable); return effectiveTable; } + private static FileStoreTable restoreBoundSchema( + FileStoreTable table, long schemaId, Map options, String path) { + if (table instanceof FallbackReadFileStoreTable) { + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) table; + // Each branch has its own schema history. Restore both children at their captured + // coordinates, preserving compatibility without broadcasting the main schema ID. + String fallbackPath = path + "fallback."; + long fallbackId = PaimonSchemaPin.fallbackSchemaId(options, fallbackPath, + () -> pair.fallback().schemaManager().latest().orElseThrow(IllegalStateException::new).id()); + return new FallbackReadFileStoreTable(restoreBoundSchema(pair.wrapped(), schemaId, options, path), + restoreBoundSchema(pair.fallback(), fallbackId, options, fallbackPath)); + } + if (table instanceof DelegatedFileStoreTable) { + FileStoreTable wrapped = ((DelegatedFileStoreTable) table).wrapped(); + return PaimonTableDecorators.replaceWrapped(table, restoreBoundSchema(wrapped, schemaId, options, path)); + } + if (table.schema().id() == schemaId) { + return table; + } + TableSchema bound = table.schemaManager().schema(schemaId); + Map persisted = table.schemaManager().schema(table.schema().id()).options(); + Map merged = new HashMap<>(bound.options()); + // Field-referencing options evolve with the schema (e.g. bucket-key and sequence.field + // on rename). Only replay the catalog/runtime delta, never another generation's options. + table.options().forEach((key, value) -> { + if (!Objects.equals(persisted.get(key), value)) { + merged.put(key, value); + } + }); + // Explicit catalog overrides can equal old physical values; equality is not provenance. + options.forEach((key, value) -> { + if (key.startsWith(INTERNAL_PREFIX + "catalog-option.")) { + merged.put(key.substring((INTERNAL_PREFIX + "catalog-option.").length()), value); + } + }); + return table.copy(bound.copy(merged)); + } + /** * Paimon's inherited read-state family: startup mode, startup position and incremental range, plus * every fallback key. Shared with {@link PaimonIncrementalScanParams#applyResetsIfIncremental}, which @@ -386,6 +432,20 @@ public static Map pinOptionsToSnapshot( return pinned; } + public static Map withBoundSchema(Map options, long schemaId) { + Map bound = new HashMap<>(options); + if (schemaId >= 0 && preservesBoundSchema(options)) { + bound.put(BOUND_SCHEMA_ID, Long.toString(schemaId)); + } + return bound; + } + + static Map withCatalogOptions(Map options, Map catalogOptions) { + Map result = new HashMap<>(options); + catalogOptions.forEach((key, value) -> result.put(INTERNAL_PREFIX + "catalog-option." + key, value)); + return result; + } + public static boolean preservesBoundSchema(Map options) { return options != null && Boolean.parseBoolean(options.get(PRESERVE_BOUND_SCHEMA)); } diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java index 3f40af1f56bc5b..a29300d2a166b9 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java @@ -114,6 +114,7 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Supplier; import java.util.stream.Collectors; /** @@ -343,10 +344,43 @@ Table resolveTable(PaimonTableHandle paimonHandle) { */ Table resolveScanTable(PaimonTableHandle paimonHandle) { Table table = resolveTable(paimonHandle); - Map scanOptions = paimonHandle.getScanOptions(); + return withBoundSchemaAuthentication(paimonHandle, () -> applyScanOptions(paimonHandle, table)); + } + + private T withBoundSchemaAuthentication(PaimonTableHandle handle, Supplier action) { + if (context == null || !PaimonScanParams.preservesBoundSchema(handle.getScanOptions())) { + return action.get(); + } + // Restoring a bound schema can read FileIO after table resolution has left the authenticated scope. + try { + return context.executeAuthenticated(action::get); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException("Failed to restore Paimon statement schema", e); + } + } + + private Map effectiveScanOptions(PaimonTableHandle handle) { + Map options = handle.getScanOptions(); + return PaimonScanParams.preservesBoundSchema(options) + ? PaimonScanParams.withCatalogOptions(options, PaimonTableOptions.extract(catalogProps.getRaw())) + : options; + } + + private Table applyScanOptions(PaimonTableHandle paimonHandle, Table table) { + Map scanOptions = effectiveScanOptions(paimonHandle); Table finalTable = table; - if (scanOptions != null && !scanOptions.isEmpty()) { - if (PaimonScanParams.isOptionsPin(scanOptions)) { + if (scanOptions != null && !scanOptions.isEmpty() + && !(paimonHandle.isSystemTable() && PaimonScanParams.preservesBoundSchema(scanOptions))) { + // Statement-fenced system wrappers are rebuilt below from their schema-bound source. + if (table instanceof FileStoreTable + && PaimonScanParams.preservesBoundSchema(scanOptions)) { + // A statement fence owns data visibility, not schema time travel. Reusing Table.copy + // here would roll schema-only ALTERs back to the data snapshot's older schema. + finalTable = PaimonScanParams.applyOptionsWithoutTimeTravel( + (FileStoreTable) table, scanOptions); + } else if (PaimonScanParams.isOptionsPin(scanOptions)) { // An @options pin owns the whole scan-startup state: applyOptions strips the internal // markers and nulls out the absent members of paimon's inherited read-state family, so a // scan.mode / tag persisted on the base table cannot leak into this relation's read. @@ -1061,8 +1095,7 @@ public Map getScanNodeProperties( source = paimonHandle.getSysBaseTable(); } Table effectiveSource = source == null ? null - : PaimonReaderOptions.runtimeSafeSystemSource( - source, paimonHandle.getScanOptions()); + : prepareSystemSource(paimonHandle, source); if (effectiveSource instanceof FileStoreTable) { // A system wrapper can hide its physical option map. Ship the exact catalog-less // source so a smaller BE can cap it and rebuild without reopening catalog state. @@ -1158,13 +1191,19 @@ OptionalInt backendManifestParallelism(PaimonTableHandle handle, Table scanTable if (source != null) { // System wrappers hide their manifest planner, so send its FE-safe value out of // band; a smaller BE can lower the same hidden planner after deserialization. - planningTable = PaimonReaderOptions.runtimeSafeSystemSource( - source, handle.getScanOptions()); + planningTable = prepareSystemSource(handle, source); } } return PaimonReaderOptions.backendManifestParallelismCap(planningTable); } + private Table prepareSystemSource(PaimonTableHandle handle, Table source) { + // These later property transformations can reopen schema files on the retained source, + // after tableForBackend has already left its authentication and plugin classloader scope. + return withBoundSchemaAuthentication(handle, + () -> PaimonReaderOptions.runtimeSafeSystemSource(source, effectiveScanOptions(handle))); + } + /** * Build the Paimon table object that is serialized to the BE. * @@ -1195,6 +1234,10 @@ OptionalInt backendManifestParallelism(PaimonTableHandle handle, Table scanTable */ // Package-private for direct unit testing (PaimonBackendBoundTableTest). Table tableForBackend(PaimonTableHandle handle, Table scanTable) { + return withBoundSchemaAuthentication(handle, () -> buildBackendTable(handle, scanTable)); + } + + private Table buildBackendTable(PaimonTableHandle handle, Table scanTable) { if (scanTable instanceof FileStoreTable) { // resolveScanTable's copy(...) merged the relation's dynamic options into the schema, // and the rebuild below goes through that schema, so this branch needs no re-application. @@ -1218,12 +1261,13 @@ Table tableForBackend(PaimonTableHandle handle, Table scanTable) { if (resolvesOnBackend) { preparedDataTable = pinCatalogSnapshot(preparedDataTable, dataTable); } - Map scanOptions = handle.getScanOptions(); + Map scanOptions = effectiveScanOptions(handle); boolean optionsAppliedToSource = PaimonScanParams.isOptionsPin(scanOptions); if (optionsAppliedToSource) { // Fallback snapshot translation consults each branch catalog, so options must be // resolved while both loaders are still present and only then made BE-safe. - preparedDataTable = (FileStoreTable) PaimonScanParams.applyOptions( + // Rebuild the backend wrapper with the same schema provenance used during binding and planning. + preparedDataTable = (FileStoreTable) PaimonReaderOptions.runtimeSafeSystemSource( preparedDataTable, scanOptions); } FileStoreTable baseForBackend = dropCatalogLoader(preparedDataTable); @@ -1411,9 +1455,11 @@ static FileStoreTable dropCatalogLoader(FileStoreTable dataTable) { } private static FileStoreTable rebuildWithoutCatalogLoader(FileStoreTable branch) { + // The factory evaluates scan.snapshot-id and can rewind a schema-only ALTER. Restore + // the already-bound schema after removing the loader so JNI reads the planned field ids. return FileStoreTableFactory.createWithoutFallbackBranch( branch.fileIO(), branch.location(), branch.schema(), new Options(), - CatalogEnvironment.empty()); + CatalogEnvironment.empty()).copy(branch.schema()); } /** @@ -1440,14 +1486,10 @@ private Table resolveSchemaDictTable(Table table, PaimonTableHandle handle) { } if (table instanceof ReadOptimizedTable) { FileStoreTable pinnedSource = handle.getSysBaseTable(); - if (pinnedSource != null) { - // $ro reads the field ids of its embedded source; a catalog reload here can observe - // schema generation B while the wrapper still plans generation A's files. Relation scan - // options must also select this source, or historical splits get the latest dictionary. - return reapplyScanParams( - pinnedSource, pinnedSource, false, handle.getScanOptions()); - } - return reloadBaseTable(handle); + // A reloaded source still needs the bound schema and data selector; otherwise the + // native dictionary can disagree with the wrapper after either cache or handle reload. + return withBoundSchemaAuthentication(handle, () -> PaimonReaderOptions.runtimeSafeSystemSource( + pinnedSource == null ? reloadBaseTable(handle) : pinnedSource, effectiveScanOptions(handle))); } return null; } diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaPin.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaPin.java new file mode 100644 index 00000000000000..7804899c998d75 --- /dev/null +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaPin.java @@ -0,0 +1,146 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.connector.paimon; + +import org.apache.doris.connector.spi.DorisConnectorException; + +import com.google.common.hash.Hashing; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.DataTable; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.utils.SnapshotManager; + +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; +import java.util.function.LongSupplier; + +/** Immutable file coordinates carried by the MVCC pin, including across INSERT planning attempts. */ +final class PaimonSchemaPin { + private static final String PREFIX = "doris.internal.paimon.schema-pin."; + + private PaimonSchemaPin() {} + + static Map capture(Table table, long schemaId, long snapshotId) { + Map pin = new HashMap<>(); + if (schemaId >= 0) { + capture(table, schemaId, snapshotId, "", pin); + if (!pin.isEmpty()) { + pin.put(PREFIX + "shape", shape(table)); + } + } + return pin; + } + + private static void capture(Table table, long schemaId, long snapshotId, String path, Map pin) { + if (table instanceof FallbackReadFileStoreTable) { + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) table; + capture(pair.wrapped(), schemaId, snapshotId, path, pin); + TableSchema fallback = pair.fallback().schemaManager().latest().orElseThrow(IllegalStateException::new); + capture(pair.fallback(), fallback.id(), -1, path + "fallback.", pin); + } else if (table instanceof DelegatedFileStoreTable) { + capture(((DelegatedFileStoreTable) table).wrapped(), schemaId, snapshotId, path, pin); + } else if (table instanceof DataTable) { + DataTable data = (DataTable) table; + pin.put(PREFIX + path + "schema-id", Long.toString(schemaId)); + pin.put(PREFIX + path + "schema", schemaDigest(data.schemaManager().schema(schemaId))); + if (snapshotId >= 0) { + pin.put(PREFIX + path + "snapshot-id", Long.toString(snapshotId)); + pin.put(PREFIX + path + "snapshot", snapshotDigest(data, snapshotId)); + } + } + } + + static void validate(Table table, Map pin) { + String capturedShape = pin.get(PREFIX + "shape"); + if (capturedShape != null && !capturedShape.equals(shape(table))) { + throw changed(); + } + validate(table, pin, ""); + } + + private static void validate(Table table, Map pin, String path) { + if (!pin.containsKey(PREFIX + path + "schema-id")) { + return; + } + if (table instanceof FallbackReadFileStoreTable) { + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) table; + validate(pair.wrapped(), pin, path); + validate(pair.fallback(), pin, path + "fallback."); + } else if (table instanceof DelegatedFileStoreTable) { + validate(((DelegatedFileStoreTable) table).wrapped(), pin, path); + } else if (table instanceof DataTable) { + DataTable data = (DataTable) table; + long schemaId = Long.parseLong(pin.get(PREFIX + path + "schema-id")); + // IDs are reusable after DROP/CREATE; compare the immutable schema and snapshot contents + // before either a retained schema or a newly resolved scan table can consume those IDs. + if (!pin.get(PREFIX + path + "schema").equals(schemaDigest(data.schemaManager().schema(schemaId)))) { + throw changed(); + } + String snapshotId = pin.get(PREFIX + path + "snapshot-id"); + if (snapshotId != null && !pin.get(PREFIX + path + "snapshot").equals( + snapshotDigest(data, Long.parseLong(snapshotId)))) { + throw changed(); + } + } + } + + static String schemaDigest(TableSchema schema) { + return digest(schema.toString()); + } + + static void validateSchema(PaimonCatalogOps.PaimonSchemaSnapshot schema, Map pin) { + String captured = pin.get(PREFIX + "schema"); + if (captured != null && schema.fileDigest() != null && !captured.equals(schema.fileDigest())) { + throw changed(); + } + } + + private static String snapshotDigest(DataTable table, long snapshotId) { + SnapshotManager manager = table.snapshotManager(); + // SDK snapshot caches are keyed by reusable paths, so generation checks must bypass them. + return digest(SnapshotManager.fromPath(manager.fileIO(), manager.snapshotPath(snapshotId)).toJson()); + } + + private static String shape(Table table) { + if (table instanceof FallbackReadFileStoreTable) { + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) table; + return "fallback(" + shape(pair.wrapped()) + "," + shape(pair.fallback()) + ")"; + } + if (table instanceof DelegatedFileStoreTable) { + return shape(((DelegatedFileStoreTable) table).wrapped()); + } + return table instanceof DataTable ? "data" : "other"; + } + + static long fallbackSchemaId(Map options, String path, LongSupplier defaultId) { + String pinned = options.get(PREFIX + path + "schema-id"); + return pinned == null ? defaultId.getAsLong() : Long.parseLong(pinned); + } + + private static DorisConnectorException changed() { + return new DorisConnectorException( + "Paimon table generation changed after the statement was pinned; retry the statement"); + } + + private static String digest(String value) { + return Hashing.sha256().hashString(value, StandardCharsets.UTF_8).toString(); + } +} diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java index fbebd7b1989a90..b762ff35e4fc03 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java @@ -17,10 +17,15 @@ package org.apache.doris.connector.paimon; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.privilege.PrivilegeChecker; +import org.apache.paimon.privilege.PrivilegedFileStoreTable; import org.apache.paimon.table.DelegatedFileStoreTable; import org.apache.paimon.table.FallbackReadFileStoreTable; import org.apache.paimon.table.FileStoreTable; +import java.lang.reflect.Field; + /** * The one place that knows how Paimon stacks {@link DelegatedFileStoreTable} decorators on a loaded * table, and how far they may be peeled off. @@ -30,6 +35,25 @@ final class PaimonTableDecorators { private PaimonTableDecorators() { } + static FileStoreTable replaceWrapped(FileStoreTable original, FileStoreTable replacement) { + if (!(original instanceof PrivilegedFileStoreTable)) { + throw new IllegalArgumentException("Unsupported Paimon planning table delegate: " + + original.getClass().getName()); + } + try { + // Paimon exposes no delegate-replacement API. Retain the original checker and identity + // so rebuilding a fallback pair does not discard authorization on subsequent reads. + Field checker = PrivilegedFileStoreTable.class.getDeclaredField("privilegeChecker"); + Field identifier = PrivilegedFileStoreTable.class.getDeclaredField("identifier"); + checker.setAccessible(true); + identifier.setAccessible(true); + return PrivilegedFileStoreTable.wrap(replacement, + (PrivilegeChecker) checker.get(original), (Identifier) identifier.get(original)); + } catch (ReflectiveOperationException e) { + throw new IllegalStateException("Failed to preserve Paimon privilege delegate", e); + } + } + /** * Peel the decorators Paimon may have stacked on top of the table, down to the fallback-branch * pair - the one layer that must stay on top, because that is what dispatches a read to the diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java index 61ec28c61d952b..e8d7f33ea8f4ca 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java @@ -518,7 +518,8 @@ public void resolveBranchFoundLoadsBranchTableAndPinsItsLatestSnapshot() { ConnectorMvccSnapshot snap = metadataWith(ops) .resolveTimeTravel(null, handle, ConnectorTimeTravelSpec.branch("b1")).get(); - // WHY: @branch must pin the BRANCH's LATEST snapshot + its schemaId, carry the branch identity + // WHY: @branch must pin the BRANCH's latest data snapshot while retaining its current schema, + // because a schema-only ALTER does not create a data snapshot. It must also carry the branch identity // via the CoreOptions.BRANCH sentinel (NOT scan.snapshot-id — the branch reads its own latest), // and validate the branch on the BASE table. Branches have no in-branch time-travel (legacy // reads the branch's latestSnapshot() only). MUTATION: pinning scan.snapshot-id -> the no-key @@ -526,8 +527,8 @@ public void resolveBranchFoundLoadsBranchTableAndPinsItsLatestSnapshot() { // base table instead of the branch -> the lastMvccTable assertion red. Assertions.assertEquals(7L, snap.getSnapshotId(), "@branch must pin the BRANCH's latest snapshot id"); - Assertions.assertEquals(3L, snap.getSchemaId(), - "@branch must stamp the BRANCH's latest snapshot schemaId"); + Assertions.assertEquals(-1L, snap.getSchemaId(), + "@branch must use the current-schema sentinel independently of its latest data snapshot"); Assertions.assertEquals("b1", snap.getProperties().get(CoreOptions.BRANCH.key()), "@branch must carry the branch name under the CoreOptions.BRANCH sentinel key"); Assertions.assertNull(snap.getProperties().get("scan.snapshot-id"), @@ -541,10 +542,11 @@ public void resolveBranchFoundLoadsBranchTableAndPinsItsLatestSnapshot() { "the branch table must be loaded via a 3-arg branch Identifier"); Assertions.assertNull(ops.lastGetTableId.getSystemTableName(), "a branch load must NOT carry a system-table name"); - // The latest-snapshot / schemaId lookups ran against the BRANCH table, not the base. (The last - // seam call before this assertion is snapshotSchemaId, which captured lastMvccTable.) + // The latest-snapshot lookup ran against the BRANCH table, not the base. Assertions.assertSame(branch, ops.lastMvccTable, - "latestSnapshotId/snapshotSchemaId must run against the BRANCH table"); + "latestSnapshotId must run against the BRANCH table"); + Assertions.assertFalse(ops.log.contains("snapshotSchemaId:7"), + "a branch schema-only ALTER must not be hidden by the data snapshot's old schema id"); // branchExists validation ran against the BASE table (legacy resolvePaimonBranch). Assertions.assertEquals("b1", ops.lastBranchExistsArg, "branchExists must be asked about the requested branch name"); @@ -880,7 +882,8 @@ public void getTableSchemaAtSnapshotIsMemoizedAcrossQueries() { rowType("id", "dt").getFields(), Arrays.asList("dt"), Collections.emptyList()); ops1.schemaAt = atSchema; ops2.schemaAt = atSchema; - ConnectorMvccSnapshot snapshot = ConnectorMvccSnapshot.builder().snapshotId(7L).schemaId(2L).build(); + ConnectorMvccSnapshot snapshot = ConnectorMvccSnapshot.builder().snapshotId(7L).schemaId(2L) + .property("scan.snapshot-id", "7").build(); ConnectorTableSchema schema1 = metadataWith(ops1, memo).getTableSchema(null, handle, snapshot); ConnectorTableSchema schema2 = metadataWith(ops2, memo).getTableSchema(null, handle, snapshot); @@ -1080,8 +1083,11 @@ public void applySnapshotWithBranchSentinelRoutesToWithBranch() { // assertions red. Assertions.assertEquals("b1", pinned.getBranchName(), "the branch sentinel must route to withBranch (handle identity), not a scan option"); - Assertions.assertTrue(pinned.getScanOptions().isEmpty(), + Assertions.assertFalse(pinned.getScanOptions().containsKey(CoreOptions.BRANCH.key()), "a branch pin must NOT thread the sentinel as a scan-copy option"); + Assertions.assertEquals("7", pinned.getScanOptions().get(CoreOptions.SCAN_SNAPSHOT_ID.key()), + "a branch pin must retain its resolved data fence after switching identity"); + Assertions.assertTrue(PaimonScanParams.preservesBoundSchema(pinned.getScanOptions())); Assertions.assertNull(pinned.getPaimonTable(), "withBranch must clear the transient base Table so the branch reloads"); } diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java index 6fcf5564d590c5..1b0763cfd2d311 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java @@ -88,6 +88,32 @@ public void ntzPushedWithUtcSemantics() { "NTZ literal must be the wall clock converted via fixed UTC (legacy GMT parity)"); } + @Test + public void ntzPushPreservesMicroseconds() { + RowType rowType = RowType.builder().field("ts", DataTypes.TIMESTAMP(6)).build(); + LocalDateTime literal = LocalDateTime.of(2024, 1, 1, 0, 0, 0, 123_456_000); + + List predicates = convertEq(rowType, "ts", literal); + + LeafPredicate leaf = (LeafPredicate) predicates.get(0); + Assertions.assertEquals(Timestamp.fromLocalDateTime(literal), leaf.literals().get(0), + "an NTZ predicate literal must retain precision below one millisecond"); + } + + @Test + public void nanosecondTimestampComparisonsRemainResidual() { + RowType rowType = RowType.builder().field("ts", DataTypes.TIMESTAMP(9)).build(); + LocalDateTime stored = LocalDateTime.of(2024, 1, 1, 0, 0, 0, 123_456_789); + LocalDateTime visible = stored.withNano(123_456_000); + Assertions.assertNotEquals(Timestamp.fromLocalDateTime(stored), Timestamp.fromLocalDateTime(visible)); + for (ConnectorComparison.Operator operator : ConnectorComparison.Operator.values()) { + ConnectorComparison comparison = new ConnectorComparison(operator, + new ConnectorColumnRef("ts", ANY), new ConnectorLiteral(ANY, visible)); + Assertions.assertTrue(new PaimonPredicateConverter(rowType).convert(comparison).isEmpty(), + "sub-microsecond source values must be compared after Doris truncation: " + operator); + } + } + @Test public void ltzNotPushed() { RowType rowType = RowType.builder() diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java index b7fe37f2ccc09e..edf264a577d528 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java @@ -23,6 +23,7 @@ import org.apache.doris.connector.spi.ConnectorType; import org.apache.doris.connector.spi.DorisConnectorException; import org.apache.doris.connector.spi.handle.ConnectorColumnHandle; +import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot; import org.apache.doris.connector.spi.pushdown.ConnectorColumnRef; import org.apache.doris.connector.spi.pushdown.ConnectorComparison; import org.apache.doris.connector.spi.pushdown.ConnectorExpression; @@ -896,6 +897,98 @@ public void resolveScanTableAppliesSnapshotPinViaCopy() { "the scan path must layer the handle's scanOptions via Table.copy(scanOptions)"); } + @Test + public void resolveScanTableKeepsCurrentSchemaForStatementSnapshot(@TempDir Path warehouse) + throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder() + .column("id", DataTypes.INT()) + .primaryKey("id") + .option("bucket", "1") + .build(), false); + FileStoreTable firstGeneration = (FileStoreTable) catalog.getTable(id); + BatchWriteBuilder writeBuilder = firstGeneration.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite()) { + write.write(GenericRow.of(1)); + List messages = write.prepareCommit(); + try (BatchTableCommit commit = writeBuilder.newCommit()) { + commit.commit(messages); + } + } + long dataSnapshotId = firstGeneration.latestSnapshot() + .orElseThrow(AssertionError::new).id(); + new SchemaManager(firstGeneration.fileIO(), firstGeneration.location()) + .commitChanges(SchemaChange.addColumn("added", DataTypes.INT())); + FileStoreTable latestGeneration = (FileStoreTable) catalog.getTable(id); + + RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps(); + PaimonTableHandle handle = new PaimonTableHandle( + "db", "t", Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(latestGeneration); + PaimonTableHandle pinned = (PaimonTableHandle) new PaimonConnectorMetadata( + ops, PaimonCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext()) + .applySnapshot(null, handle, ConnectorMvccSnapshot.builder() + .snapshotId(dataSnapshotId) + .build()); + + Table scanTable = new PaimonScanPlanProvider( + PaimonCatalogProperties.of(Collections.emptyMap()), ops).resolveScanTable(pinned); + + Assertions.assertTrue(scanTable.rowType().getFieldNames().contains("added"), + "pinning data visibility must not roll a normal query back to the snapshot's old schema"); + Assertions.assertEquals(String.valueOf(dataSnapshotId), + scanTable.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key())); + } + } + + @Test + public void resolveScanTableKeepsCurrentSchemaForReaderOnlyOptions(@TempDir Path warehouse) + throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder() + .column("id", DataTypes.INT()) + .primaryKey("id") + .option("bucket", "1") + .build(), false); + FileStoreTable firstGeneration = (FileStoreTable) catalog.getTable(id); + BatchWriteBuilder writeBuilder = firstGeneration.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite()) { + write.write(GenericRow.of(1)); + List messages = write.prepareCommit(); + try (BatchTableCommit commit = writeBuilder.newCommit()) { + commit.commit(messages); + } + } + long dataSnapshotId = firstGeneration.latestSnapshot() + .orElseThrow(AssertionError::new).id(); + new SchemaManager(firstGeneration.fileIO(), firstGeneration.location()) + .commitChanges(SchemaChange.addColumn("added", DataTypes.INT())); + FileStoreTable latestGeneration = (FileStoreTable) catalog.getTable(id); + + PaimonTableHandle handle = new PaimonTableHandle( + "db", "t", Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(latestGeneration); + Map scanOptions = PaimonScanParams.markAsOptions( + PaimonScanParams.pinOptionsToSnapshot( + Collections.singletonMap("scan.plan-sort-partition", "true"), + dataSnapshotId)); + + Table scanTable = new PaimonScanPlanProvider( + PaimonCatalogProperties.of(Collections.emptyMap()), new RecordingPaimonCatalogOps()) + .resolveScanTable(handle.withScanOptions(scanOptions)); + + Assertions.assertTrue(scanTable.rowType().getFieldNames().contains("added"), + "reader-only OPTIONS must retain the current bound schema while pinning data visibility"); + Assertions.assertEquals("true", scanTable.options().get("scan.plan-sort-partition")); + } + } + @Test public void resolveScanTableWithoutScanOptionsDoesNotCopy() { RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps(); diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java new file mode 100644 index 00000000000000..d4fc6cb17c4750 --- /dev/null +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java @@ -0,0 +1,695 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.connector.paimon; + +import org.apache.doris.connector.spi.ConnectorColumn; +import org.apache.doris.connector.spi.ConnectorContext; +import org.apache.doris.connector.spi.ConnectorSession; +import org.apache.doris.connector.spi.ForwardingConnectorContext; +import org.apache.doris.connector.spi.handle.ConnectorColumnHandle; +import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot; +import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec; +import org.apache.doris.connector.spi.scan.ConnectorScanRequest; +import org.apache.doris.thrift.TFileScanRangeParams; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.FileSystemCatalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.privilege.PrivilegeChecker; +import org.apache.paimon.privilege.PrivilegedFileStoreTable; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.InstantiationUtil; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.lang.reflect.Proxy; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; + +public class PaimonStatementSchemaTest { + @Test + public void latestPinSurvivesInsertScopeReset(@TempDir Path warehouse) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("old_name", DataTypes.INT()).build(), false); + append((FileStoreTable) catalog.getTable(id), GenericRow.of(1)); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonCatalogProperties props = PaimonCatalogProperties.of(Collections.emptyMap()); + PaimonSchemaAtMemo memo = new PaimonSchemaAtMemo(1000); + PaimonTableHandle old = new PaimonTableHandle("db", "t", Collections.emptyList(), Collections.emptyList()); + old.setPaimonTable(catalog.getTable(id)); + new PaimonConnectorMetadata(ops, props, new RecordingConnectorContext(), memo) + .getTableSchema(null, old, ConnectorMvccSnapshot.builder().snapshotId(1).schemaId(0) + .property("scan.snapshot-id", "1").build()); + Assertions.assertEquals(1, memo.size()); + catalog.dropTable(id, false); + catalog.createTable(id, Schema.newBuilder().column("new_name", DataTypes.INT()).build(), false); + PaimonTableHandle handle = new PaimonTableHandle("db", "t", Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(catalog.getTable(id)); + PaimonConnectorMetadata first = new PaimonConnectorMetadata(ops, props, new RecordingConnectorContext(), memo); + ConnectorMvccSnapshot pin = first.beginQuerySnapshot(null, handle).get(); + Assertions.assertEquals("new_name", first.getTableSchema(null, handle, pin).getColumns().get(0).getName()); + // INSERT replanning retains the source pin but creates a fresh connector metadata scope. + PaimonConnectorMetadata retry = new PaimonConnectorMetadata(ops, props, new RecordingConnectorContext(), memo); + Assertions.assertTrue(retry.getColumnHandles(null, handle, pin).containsKey("new_name")); + Assertions.assertEquals("new_name", retry.getTableSchema(null, handle, pin).getColumns().get(0).getName()); + } + } + + @Test + public void latestPinRejectsReplacementDuringStatement(@TempDir Path warehouse) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()).build(), false); + append((FileStoreTable) catalog.getTable(id), GenericRow.of(1)); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonCatalogProperties props = PaimonCatalogProperties.of(Collections.emptyMap()); + PaimonTableHandle first = new PaimonTableHandle("db", "t", Collections.emptyList(), Collections.emptyList()); + first.setPaimonTable(catalog.getTable(id)); + PaimonConnectorMetadata md = new PaimonConnectorMetadata(ops, props, new RecordingConnectorContext()); + ConnectorMvccSnapshot pin = md.beginQuerySnapshot(null, first).get(); + md.getTableSchema(null, first, pin); + catalog.dropTable(id, false); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()).build(), false); + append((FileStoreTable) catalog.getTable(id), GenericRow.of(2)); + PaimonTableHandle replacement = new PaimonTableHandle("db", "t", Collections.emptyList(), Collections.emptyList()); + replacement.setPaimonTable(catalog.getTable(id)); + Assertions.assertEquals(pin.getSchemaId(), ((FileStoreTable) replacement.getPaimonTable()).schema().id()); + ConnectorMvccSnapshot aliasPin = md.beginQuerySnapshot(null, replacement).get(); + Assertions.assertEquals(pin.getProperties(), aliasPin.getProperties()); + PaimonTableHandle scanHandle = (PaimonTableHandle) md.applySnapshot(null, replacement, pin); + scanHandle.setPaimonTable(null); + Assertions.assertTrue(Assertions.assertThrows(RuntimeException.class, + () -> new PaimonScanPlanProvider(props, ops).resolveScanTable(scanHandle)) + .getMessage().contains("changed")); + Assertions.assertTrue(Assertions.assertThrows(RuntimeException.class, + () -> md.getColumnHandles(null, replacement, pin)).getMessage().contains("changed")); + } + } + + @Test + public void latestPinRejectsRecreationWhileCapturingSchema(@TempDir Path warehouse) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("old_name", DataTypes.INT()).build(), false); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog) { + @Override + public Optional latestSchema(Table table) { + Optional schema = super.latestSchema(table); + try { + catalog.dropTable(id, false); + catalog.createTable(id, + Schema.newBuilder().column("new_name", DataTypes.INT()).build(), false); + } catch (Exception e) { + throw new RuntimeException(e); + } + return schema; + } + }; + PaimonTableHandle handle = new PaimonTableHandle("db", "t", Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(catalog.getTable(id)); + PaimonConnectorMetadata md = new PaimonConnectorMetadata(ops, + PaimonCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext()); + Assertions.assertTrue(Assertions.assertThrows(RuntimeException.class, + () -> md.beginQuerySnapshot(null, handle)).getMessage().contains("changed")); + } + } + + @Test + public void fallbackCapturesBothLatestSchemas(@TempDir Path warehouse) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()) + .column("v", DataTypes.INT()).primaryKey("id").option("bucket", "1").build(), false); + FileStoreTable main = (FileStoreTable) catalog.getTable(id); + append(main, GenericRow.of(1, 10)); + main.createBranch("backup"); + FileStoreTable fallback = main.switchToBranch("backup"); + append(fallback, GenericRow.of(2, 20)); + main.schemaManager().commitChanges(SchemaChange.addColumn("added", DataTypes.INT())); + fallback.schemaManager().commitChanges(SchemaChange.setOption("read.batch-size", "64")); + fallback.schemaManager().commitChanges(SchemaChange.addColumn("added", DataTypes.INT())); + PaimonTableHandle handle = new PaimonTableHandle("db", "t", Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(new FallbackReadFileStoreTable(main, fallback)); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonCatalogProperties props = PaimonCatalogProperties.of(Collections.emptyMap()); + PaimonConnectorMetadata md = new PaimonConnectorMetadata(ops, props, new RecordingConnectorContext()); + ConnectorMvccSnapshot pin = md.beginQuerySnapshot(null, handle).get(); + PaimonTableHandle pinned = (PaimonTableHandle) md.applySnapshot(null, handle, pin); + // Later branch changes must not move the already captured fallback schema. + fallback.schemaManager().commitChanges(SchemaChange.addColumn("later", DataTypes.INT())); + FileStoreTable scan = (FileStoreTable) new PaimonScanPlanProvider(props, ops).resolveScanTable(pinned); + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) scan; + Assertions.assertEquals(1, pair.wrapped().schema().id()); + Assertions.assertEquals(2, pair.fallback().schema().id()); + Assertions.assertEquals(pair.wrapped().rowType(), pair.fallback().rowType()); + Assertions.assertEquals("backup", pair.fallback().coreOptions().branch()); + Assertions.assertTrue(readIds(scan).contains(1)); + Assertions.assertThrows(RuntimeException.class, + () -> PaimonScanParams.applyOptionsWithoutTimeTravel(main, pinned.getScanOptions())); + } + } + + @Test + public void explicitCatalogOptionSurvivesEqualOldPhysicalValue(@TempDir Path warehouse) throws Exception { + checkCatalogOptionPrecedence(warehouse.resolve("positive"), "128"); + checkCatalogOptionPrecedence(warehouse.resolve("invalid"), "0"); + } + + private void checkCatalogOptionPrecedence(Path warehouse, String physicalValue) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()) + .option("read.batch-size", "64").build(), false); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog, + Collections.singletonMap("read.batch-size", "64")); + PaimonCatalogProperties props = PaimonCatalogProperties.of( + Collections.singletonMap("paimon.table-option.read.batch-size", "64")); + PaimonTableHandle handle = new PaimonTableHandle("db", "t", Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(ops.getTable(id)); + catalog.alterTable(id, Collections.singletonList(SchemaChange.setOption("read.batch-size", physicalValue)), false); + PaimonConnectorMetadata md = new PaimonConnectorMetadata(ops, props, new RecordingConnectorContext()); + PaimonTableHandle pinned = (PaimonTableHandle) md.applySnapshot(null, handle, + md.beginQuerySnapshot(null, handle).get()); + Table scan = new PaimonScanPlanProvider(props, ops).resolveScanTable(pinned); + Assertions.assertEquals("64", scan.options().get("read.batch-size")); + Map relationOptions = new java.util.HashMap<>(pinned.getScanOptions()); + relationOptions.put("read.batch-size", "32"); + Table relationScan = new PaimonScanPlanProvider(props, ops) + .resolveScanTable(pinned.withScanOptions(relationOptions)); + Assertions.assertEquals("32", relationScan.options().get("read.batch-size")); + } + } + + @Test + public void latestSchemaSurvivesExternalRecreationWithReusedId(@TempDir Path warehouse) throws Exception { + checkExternalRecreation(warehouse, false); + } + + @Test + public void latestSchemaAndDataSurviveExternalRecreationWithReusedIds(@TempDir Path warehouse) throws Exception { + checkExternalRecreation(warehouse, true); + } + + private void checkExternalRecreation(Path warehouse, boolean withData) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("old_name", DataTypes.INT()).build(), false); + if (withData) { + append((FileStoreTable) catalog.getTable(id), GenericRow.of(1)); + } + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonCatalogProperties props = PaimonCatalogProperties.of(Collections.emptyMap()); + PaimonSchemaAtMemo memo = new PaimonSchemaAtMemo(1000); + PaimonLatestSnapshotCache cache = new PaimonLatestSnapshotCache(0, 1000); + PaimonTableHandle firstHandle = new PaimonTableHandle("db", "t", + Collections.emptyList(), Collections.emptyList()); + firstHandle.setPaimonTable(catalog.getTable(id)); + PaimonConnectorMetadata first = new PaimonConnectorMetadata( + ops, props, new RecordingConnectorContext(), memo, cache); + ConnectorMvccSnapshot oldPin = first.beginQuerySnapshot(null, firstHandle).get(); + Assertions.assertEquals("old_name", first.getTableSchema(null, firstHandle, oldPin) + .getColumns().get(0).getName()); + // Warm the historical memo independently; latest reads must not reuse it after recreation. + new PaimonConnectorMetadata(ops, props, new RecordingConnectorContext(), memo, cache) + .getTableSchema(null, firstHandle, ConnectorMvccSnapshot.builder() + .snapshotId(oldPin.getSnapshotId()).schemaId(oldPin.getSchemaId()) + .property("scan.snapshot-id", "1").build()); + Assertions.assertEquals(1, memo.size()); + catalog.dropTable(id, false); + catalog.createTable(id, Schema.newBuilder().column("new_name", DataTypes.INT()).build(), false); + if (withData) { + append((FileStoreTable) catalog.getTable(id), GenericRow.of(2)); + } + PaimonTableHandle secondHandle = new PaimonTableHandle("db", "t", + Collections.emptyList(), Collections.emptyList()); + secondHandle.setPaimonTable(catalog.getTable(id)); + PaimonConnectorMetadata second = new PaimonConnectorMetadata( + ops, props, new RecordingConnectorContext(), memo, cache); + ConnectorMvccSnapshot newPin = second.beginQuerySnapshot(null, secondHandle).get(); + Assertions.assertEquals(oldPin.getSchemaId(), newPin.getSchemaId()); + Assertions.assertEquals(oldPin.getSnapshotId(), newPin.getSnapshotId()); + PaimonTableHandle pinned = (PaimonTableHandle) second.applySnapshot(null, secondHandle, newPin); + Table scan = new PaimonScanPlanProvider(props, ops).resolveScanTable(pinned); + Assertions.assertEquals("new_name", scan.rowType().getFieldNames().get(0)); + if (withData) { + Assertions.assertEquals(Collections.singletonList(2), readIds((FileStoreTable) scan)); + } + Assertions.assertTrue(second.getColumnHandles(null, secondHandle, newPin).containsKey("new_name")); + Assertions.assertFalse(second.getColumnHandles(null, secondHandle, newPin).containsKey("old_name")); + Assertions.assertEquals("new_name", second.getTableSchema(null, secondHandle, newPin) + .getColumns().get(0).getName(), "Recreated latest metadata must match the scan table"); + } + } + + @Test + public void latestCacheHitCapturesSchemaInsideAuth(@TempDir Path warehouse) throws Exception { + checkAuthenticatedSchemaRead(warehouse, "capture"); + } + + @Test + public void pinnedSchemaMaterializesInsideAuth(@TempDir Path warehouse) throws Exception { + checkAuthenticatedSchemaRead(warehouse, "materialize"); + } + + @Test + public void scanRestoresSchemaInsideAuth(@TempDir Path warehouse) throws Exception { + checkAuthenticatedSchemaRead(warehouse, "restore"); + } + + @Test + public void systemOptionsScanPropertiesRestoreSchemaInsideAuth(@TempDir Path warehouse) throws Exception { + checkAuthenticatedSchemaRead(warehouse, "system-properties"); + } + + private static final class SchemaGuardFileIO extends LocalFileIO { + // The FE-only assertion must not be serialized into the table sent to the backend. + private final transient Runnable checkSchemaRead; + + private SchemaGuardFileIO(Runnable checkSchemaRead) { + this.checkSchemaRead = checkSchemaRead; + } + + @Override + public SeekableInputStream newInputStream(org.apache.paimon.fs.Path path) throws IOException { + if (checkSchemaRead != null && path.toString().contains("/schema")) { + checkSchemaRead.run(); + } + return super.newInputStream(path); + } + } + + private void checkAuthenticatedSchemaRead(Path warehouse, String operation) throws Exception { + AtomicBoolean enforceScope = new AtomicBoolean(); + ThreadLocal authenticated = ThreadLocal.withInitial(() -> false); + AtomicInteger reads = new AtomicInteger(); + ClassLoader pluginLoader = new ClassLoader(getClass().getClassLoader()) {}; + ClassLoader callerLoader = Thread.currentThread().getContextClassLoader(); + FileIO guarded = new SchemaGuardFileIO(() -> { + if (enforceScope.get()) { + Assertions.assertTrue(authenticated.get(), "schema FileIO must run inside auth"); + Assertions.assertSame(pluginLoader, Thread.currentThread().getContextClassLoader()); + reads.incrementAndGet(); + } + }); + ConnectorContext context = new TcclPinningConnectorContext( + new ForwardingConnectorContext(new RecordingConnectorContext()) { + @Override + public T executeAuthenticated(Callable task) throws Exception { + boolean previous = authenticated.get(); + authenticated.set(true); + try { + return task.call(); + } finally { + authenticated.set(previous); + } + } + }, pluginLoader, () -> null); + try (Catalog catalog = new FileSystemCatalog(guarded, + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("old_name", DataTypes.INT()) + .option("scan.manifest.parallelism", "1").build(), false); + FileStoreTable warm = (FileStoreTable) catalog.getTable(id); + PaimonTableHandle handle = new PaimonTableHandle("db", "t", + Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(warm); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonLatestSnapshotCache cache = new PaimonLatestSnapshotCache(100, 1000); + PaimonCatalogProperties props = PaimonCatalogProperties.of(Collections.emptyMap()); + new PaimonConnectorMetadata(ops, props, context, new PaimonSchemaAtMemo(1000), cache) + .beginQuerySnapshot(null, handle); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("old_name", "bound_name")), false); + long schemaId = ((FileStoreTable) catalog.getTable(id)).schema().id(); + PaimonConnectorMetadata metadata = new PaimonConnectorMetadata( + ops, props, context, new PaimonSchemaAtMemo(1000), cache); + ConnectorMvccSnapshot snapshot = ConnectorMvccSnapshot.builder().snapshotId(-1L) + .schemaId(schemaId).build(); + if (operation.equals("system-properties")) { + snapshot = metadata.resolveTimeTravel(null, handle, ConnectorTimeTravelSpec.options( + Collections.singletonMap("scan.plan-sort-partition", "true"), -1L)).get(); + handle = (PaimonTableHandle) metadata.getSysTableHandle(null, handle, "ro").get(); + } + enforceScope.set(true); + if (operation.equals("capture")) { + Assertions.assertEquals(schemaId, metadata.beginQuerySnapshot(null, handle).get().getSchemaId()); + } else if (operation.equals("materialize")) { + Assertions.assertEquals("bound_name", + metadata.getTableSchema(null, handle, snapshot).getColumns().get(0).getName()); + } else { + PaimonTableHandle pinned = (PaimonTableHandle) metadata.applySnapshot(null, handle, snapshot); + PaimonScanPlanProvider provider = new PaimonScanPlanProvider(props, ops, context); + Assertions.assertEquals(Collections.singletonList("bound_name"), + provider.resolveScanTable(pinned).rowType().getFieldNames()); + if (operation.equals("system-properties")) { + // Exercise both source restorations without the pre-existing native history-dictionary IO. + Map scanProperties = provider.getScanNodeProperties( + session(true), pinned, Collections.emptyList(), Optional.empty()); + Assertions.assertTrue(scanProperties.get("paimon.options_json") + .contains("doris.serialized-system-source")); + } + } + Assertions.assertTrue(reads.get() > 0, "the assertion must exercise real schema-file IO"); + Assertions.assertFalse(authenticated.get()); + Assertions.assertSame(callerLoader, Thread.currentThread().getContextClassLoader()); + enforceScope.set(false); + } + } + + @Test + public void warmTableKeepsExactStatementSchema(@TempDir Path warehouse) throws Exception { + checkSchemaMutation(warehouse, false, false); + } + + @Test + public void warmOptionsTableKeepsExactStatementSchema(@TempDir Path warehouse) throws Exception { + checkSchemaMutation(warehouse, true, false); + } + + @Test + public void systemOptionsKeepsExactStatementSchema(@TempDir Path warehouse) throws Exception { + checkSchemaMutation(warehouse, true, true); + } + + @Test + public void warmKeyRenameUsesBoundSchemaOptions(@TempDir Path warehouse) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT().notNull()) + .column("old_key", DataTypes.INT().notNull()).primaryKey("id", "old_key") + .option("bucket", "1").option("bucket-key", "old_key") + .option("sequence.field", "old_key").option("file.format", "parquet").build(), false); + FileStoreTable warm = (FileStoreTable) catalog.getTable(id); + append(warm, GenericRow.of(1, 10)); + warm = warm.copyWithoutTimeTravel(Collections.singletonMap("read.batch-size", "64")); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("old_key", "bound_key")), false); + long boundId = ((FileStoreTable) catalog.getTable(id)).schema().id(); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("bound_key", "later_key")), false); + for (FileStoreTable loaded : Arrays.asList(warm, (FileStoreTable) catalog.getTable(id))) { + FileStoreTable pinned = PaimonScanParams.applyOptionsWithoutTimeTravel(loaded, + PaimonScanParams.withBoundSchema( + PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), 1), boundId)); + Assertions.assertEquals("bound_key", pinned.options().get("bucket-key")); + Assertions.assertEquals("bound_key", pinned.options().get("sequence.field")); + Assertions.assertEquals(Collections.singletonList(1), readIds(pinned)); + if (loaded == warm) { + Assertions.assertEquals("64", pinned.options().get("read.batch-size")); + } + } + } + } + + @Test + public void stalePrivilegedFallbackKeepsBranchProvenance(@TempDir Path warehouse) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()) + .column("value", DataTypes.INT()).partitionKeys("id") + .option("file.format", "parquet").option("scan.manifest.parallelism", "1").build(), false); + FileStoreTable warm = (FileStoreTable) catalog.getTable(id); + warm.createBranch("backup"); + FileStoreTable fallback = warm.switchToBranch("backup"); + append(fallback, GenericRow.of(2, 20)); + append(warm, GenericRow.of(1, 10)); + catalog.alterTable(id, Collections.singletonList(SchemaChange.setOption("read.batch-size", "32")), + false); + FileStoreTable latest = (FileStoreTable) catalog.getTable(id); + AtomicInteger selectChecks = new AtomicInteger(); + AtomicBoolean denied = new AtomicBoolean(); + PrivilegeChecker checker = (PrivilegeChecker) Proxy.newProxyInstance( + PrivilegeChecker.class.getClassLoader(), new Class[] {PrivilegeChecker.class}, + (proxy, method, args) -> { + if (method.getName().equals("assertCanSelect")) { + selectChecks.incrementAndGet(); + if (denied.get()) { + throw new SecurityException("Select denied"); + } + } + return null; + }); + FileStoreTable privileged = PrivilegedFileStoreTable.wrap( + new FallbackReadFileStoreTable(warm, fallback), checker, id); + FileStoreTable pinned = PaimonScanParams.applyOptionsWithoutTimeTravel(privileged, + PaimonScanParams.withBoundSchema( + PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), 1), latest.schema().id())); + Assertions.assertInstanceOf(PrivilegedFileStoreTable.class, pinned); + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) + ((DelegatedFileStoreTable) pinned).wrapped(); + Assertions.assertEquals("backup", pair.fallback().coreOptions().branch()); + Assertions.assertEquals(fallback.schema().id(), pair.fallback().schema().id()); + Assertions.assertEquals(latest.schema().id(), pair.wrapped().schema().id()); + List ids = readIds(pinned); + Collections.sort(ids); + Assertions.assertEquals(Arrays.asList(1, 2), ids); + Assertions.assertTrue(selectChecks.get() >= 2); + denied.set(true); + Assertions.assertThrows(SecurityException.class, pinned::newScan); + Assertions.assertThrows(SecurityException.class, pinned::newRead); + } + } + + private void checkSchemaMutation(Path warehouse, boolean options, boolean system) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()) + .column("old_name", DataTypes.INT()).option("file.format", "parquet").build(), false); + FileStoreTable warm = (FileStoreTable) catalog.getTable(id); + append(warm, GenericRow.of(1, 10)); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("old_name", "bound_name")), false); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonConnectorMetadata metadata = new PaimonConnectorMetadata(ops, + PaimonCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext()); + PaimonScanPlanProvider provider = new PaimonScanPlanProvider( + PaimonCatalogProperties.of(Collections.emptyMap()), ops); + PaimonTableHandle handle = new PaimonTableHandle("db", "t", + Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(warm); + ConnectorMvccSnapshot snapshot = metadata.beginQuerySnapshot(null, handle).get(); + if (options) { + snapshot = metadata.resolveTimeTravel(null, handle, ConnectorTimeTravelSpec.options( + Collections.singletonMap("scan.plan-sort-partition", "true"), + snapshot.getSnapshotId())).get(); + } + if (system) { + handle = (PaimonTableHandle) metadata.getSysTableHandle(null, handle, "ro").get(); + } + Assertions.assertTrue(metadata.getTableSchema(null, handle, snapshot).getColumns().stream() + .map(ConnectorColumn::getName).collect(Collectors.toList()).contains("bound_name")); + // The second ALTER must not change a statement that already bound the first rename. + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("bound_name", "later_name")), false); + PaimonTableHandle pinned = (PaimonTableHandle) metadata.applySnapshot(null, handle, snapshot); + Table scan = provider.resolveScanTable(pinned); + Assertions.assertEquals(Arrays.asList("id", "bound_name"), scan.rowType().getFieldNames()); + List columns = Arrays.asList(new PaimonColumnHandle("id", 0), + new PaimonColumnHandle("bound_name", 1)); + Assertions.assertFalse(provider.planScan(session(), + ConnectorScanRequest.builder(pinned, columns).build()).isEmpty()); + Map properties = provider.getScanNodeProperties(session(), pinned, columns, + Optional.empty()); + TFileScanRangeParams params = new TFileScanRangeParams(); + PaimonScanPlanProvider.applySchemaEvolutionParam(params, properties.get("paimon.schema_evolution")); + Assertions.assertEquals("bound_name", params.getHistorySchemaInfo().get(0) + .getRootField().getFields().get(1).getFieldPtr().getName()); + Table backend = InstantiationUtil.deserializeObject( + InstantiationUtil.serializeObject(provider.tableForBackend(pinned, scan)), + getClass().getClassLoader()); + Assertions.assertEquals(scan.rowType(), backend.rowType()); + Assertions.assertEquals(Collections.singletonList(1), readIds(backend)); + } + } + + @Test + public void branchCommitAfterResolutionIsInvisible(@TempDir Path warehouse) throws Exception { + checkBranchMutation(warehouse, false); + } + + @Test + public void emptyBranchRemainsEmptyAfterFirstCommit(@TempDir Path warehouse) throws Exception { + checkBranchMutation(warehouse, true); + } + + private void checkBranchMutation(Path warehouse, boolean empty) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()) + .option("file.format", "parquet").build(), false); + FileStoreTable base = (FileStoreTable) catalog.getTable(id); + base.branchManager().createBranch("dev"); + Identifier branchId = new Identifier("db", "t", "dev"); + FileStoreTable branch = (FileStoreTable) catalog.getTable(branchId); + if (!empty) { + append(branch, GenericRow.of(1)); + } + catalog.alterTable(branchId, Collections.singletonList( + SchemaChange.addColumn("bound_name", DataTypes.INT())), false); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonConnectorMetadata metadata = new PaimonConnectorMetadata(ops, + PaimonCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext()); + PaimonTableHandle handle = new PaimonTableHandle("db", "t", + Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(base); + ConnectorMvccSnapshot snapshot = metadata.resolveTimeTravel(null, handle, + ConnectorTimeTravelSpec.branch("dev")).get(); + catalog.alterTable(branchId, Collections.singletonList( + SchemaChange.renameColumn("bound_name", "later_name")), false); + append((FileStoreTable) catalog.getTable(branchId), GenericRow.of(2, 20)); + PaimonTableHandle pinned = (PaimonTableHandle) metadata.applySnapshot(null, handle, snapshot); + PaimonScanPlanProvider provider = new PaimonScanPlanProvider( + PaimonCatalogProperties.of(Collections.emptyMap()), ops); + Assertions.assertEquals(Arrays.asList("id", "bound_name"), + provider.resolveScanTable(pinned).rowType().getFieldNames()); + if (empty) { + Assertions.assertTrue(provider.planScan(session(), ConnectorScanRequest.builder(pinned, + Collections.singletonList(new PaimonColumnHandle("id", 0))).build()).isEmpty()); + } else { + Table scan = provider.resolveScanTable(pinned); + Assertions.assertEquals(Collections.singletonList(1), readIds(scan)); + Table backend = InstantiationUtil.deserializeObject( + InstantiationUtil.serializeObject(provider.tableForBackend(pinned, scan)), + getClass().getClassLoader()); + Assertions.assertEquals(Collections.singletonList(1), readIds(backend)); + } + } + } + + private static void append(FileStoreTable table, GenericRow row) throws Exception { + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite(); BatchTableCommit commit = builder.newCommit()) { + write.write(row); + commit.commit(write.prepareCommit()); + } + } + + private static List readIds(Table table) throws Exception { + List ids = new ArrayList<>(); + for (Split split : table.newReadBuilder().newScan().plan().splits()) { + try (RecordReader reader = table.newReadBuilder().newRead().createReader(split)) { + reader.forEachRemaining(row -> ids.add(row.getInt(0))); + } + } + Collections.sort(ids); + return ids; + } + + private static ConnectorSession session() { + return session(false); + } + + private static ConnectorSession session(boolean forceJni) { + return new ConnectorSession() { + @Override + public String getQueryId() { + return "q"; + } + + @Override + public String getUser() { + return "u"; + } + + @Override + public String getTimeZone() { + return "UTC"; + } + + @Override + public String getLocale() { + return "en_US"; + } + + @Override + public long getCatalogId() { + return 0; + } + + @Override + public String getCatalogName() { + return "c"; + } + + @Override + public T getProperty(String name, Class type) { + return null; + } + + @Override + public Map getCatalogProperties() { + return Collections.emptyMap(); + } + + @Override + public Map getSessionProperties() { + return Collections.singletonMap("force_jni_scanner", Boolean.toString(forceJni)); + } + }; + } + +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java index d4223f9f7c60d2..7b8fbc997ba4fc 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java @@ -172,6 +172,19 @@ private PluginDrivenMvccSnapshot materializeLatest( // legacy listPartitions/LIST/timestamp path below (byte-unchanged; the no-op applySnapshot for the // latest pin is side-effect-free for both paimon and iceberg). ConnectorTableHandle pinnedHandle = metadata.applySnapshot(session, handle, connectorSnapshot); + PluginDrivenSchemaCacheValue pinnedSchema = null; + if (connectorSnapshot.getSchemaId() >= 0) { + // Latest data and schema can advance independently. Keep the connector's exact schema + // on the statement pin so analysis cannot fall back to a different cached generation. + ConnectorTableSchema atSchema = metadata.getTableSchema(session, pinnedHandle, connectorSnapshot); + pinnedSchema = toSchemaCacheValue(metadata, session, + db != null ? db.getRemoteName() : "", getRemoteName(), atSchema); + // Eager pins bypass the schema cache loader, so preserve its mapped-column validation. + pinnedSchema.validateSchema(); + } + // This pin is not in StatementContext yet; ambient schema lookup can see a later generation. + List partitionColumns = pinnedSchema == null + ? getPartitionColumns() : pinnedSchema.getPartitionColumns(); // Partition counts feed COUNT(*) pushdown and SQL block rules, so even the initial latest // materialization must enumerate the same pinned generation that the data scan reads. ConnectorTableHandle partitionHandle = pinnedHandle; @@ -182,7 +195,7 @@ private PluginDrivenMvccSnapshot materializeLatest( if (connectorSnapshot.getSnapshotId() < 0) { // A negative query-begin pin is the connector's resolved-empty generation. Falling through to // a live LIST read would mix partitions committed after the data scan's empty boundary. - return buildFromRangeView(connectorSnapshot, view); + return buildFromRangeView(connectorSnapshot, view, pinnedSchema, partitionColumns); } // A non-RANGE (UNPARTITIONED) view is the connector's "not RANGE / not MTMV-range-eligible" verdict // (iceberg's range view only covers single time-transform specs). If the table nonetheless declares @@ -194,21 +207,23 @@ private PluginDrivenMvccSnapshot materializeLatest( // A RANGE view (range items) and a genuinely unpartitioned table (no partition columns) are // unaffected. Freshness matches the legacy LIST path (timestamps / 0), harmless here because the // UNPARTITIONED verdict keeps this table out of the snapshot-id MTMV path. - if (view.getStyle() != ConnectorMvccPartitionView.Style.RANGE && !getPartitionColumns().isEmpty()) { + if (view.getStyle() != ConnectorMvccPartitionView.Style.RANGE && !partitionColumns.isEmpty()) { Map listItems = Maps.newHashMap(); Map listLastModifiedMillis = Maps.newHashMap(); - listLatestPartitions(metadata, session, partitionHandle, listItems, listLastModifiedMillis); + listLatestPartitions(metadata, session, partitionHandle, partitionColumns, + listItems, listLastModifiedMillis); return new PluginDrivenMvccSnapshot(connectorSnapshot, listItems, listLastModifiedMillis, - null, PartitionType.UNPARTITIONED, false, 0L); + pinnedSchema, PartitionType.UNPARTITIONED, false, 0L); } - return buildFromRangeView(connectorSnapshot, view); + return buildFromRangeView(connectorSnapshot, view, pinnedSchema, partitionColumns); } Map nameToPartitionItem = Maps.newHashMap(); Map nameToLastModifiedMillis = Maps.newHashMap(); - listLatestPartitions(metadata, session, partitionHandle, nameToPartitionItem, nameToLastModifiedMillis); + listLatestPartitions(metadata, session, partitionHandle, partitionColumns, + nameToPartitionItem, nameToLastModifiedMillis); return new PluginDrivenMvccSnapshot(connectorSnapshot, nameToPartitionItem, - nameToLastModifiedMillis); + nameToLastModifiedMillis, pinnedSchema); } /** @@ -221,7 +236,7 @@ private PluginDrivenMvccSnapshot materializeLatest( * per-partition log-and-skip). */ private PluginDrivenMvccSnapshot buildFromRangeView(ConnectorMvccSnapshot connectorSnapshot, - ConnectorMvccPartitionView view) { + ConnectorMvccPartitionView view, PluginDrivenSchemaCacheValue pinnedSchema, List partitionColumns) { PartitionType partitionType = view.getStyle() == ConnectorMvccPartitionView.Style.RANGE ? PartitionType.RANGE : PartitionType.UNPARTITIONED; boolean snapshotIdFreshness = @@ -229,7 +244,6 @@ private PluginDrivenMvccSnapshot buildFromRangeView(ConnectorMvccSnapshot connec Map nameToPartitionItem = Maps.newHashMap(); Map nameToFreshnessValue = Maps.newHashMap(); if (view.getStyle() == ConnectorMvccPartitionView.Style.RANGE) { - List partitionColumns = getPartitionColumns(); for (ConnectorMvccPartition partition : view.getPartitions()) { try { nameToPartitionItem.put(partition.getName(), @@ -244,7 +258,7 @@ private PluginDrivenMvccSnapshot buildFromRangeView(ConnectorMvccSnapshot connec } } return new PluginDrivenMvccSnapshot(connectorSnapshot, nameToPartitionItem, nameToFreshnessValue, - null, partitionType, snapshotIdFreshness, view.getNewestUpdateMonotonicMarker(), + pinnedSchema, partitionType, snapshotIdFreshness, view.getNewestUpdateMonotonicMarker(), view.getNewestUpdateWallClockMillis()); } @@ -283,9 +297,8 @@ private static List toPartitionValues(List bound) { * rather than failing the whole query. */ private void listLatestPartitions(ConnectorMetadata metadata, ConnectorSession session, - ConnectorTableHandle handle, Map nameToPartitionItem, - Map nameToLastModifiedMillis) { - List partitionColumns = getPartitionColumns(); + ConnectorTableHandle handle, List partitionColumns, + Map nameToPartitionItem, Map nameToLastModifiedMillis) { List types = partitionColumns.stream().map(Column::getType).collect(Collectors.toList()); List parts = metadata.listPartitions(session, handle, Optional.empty()); for (ConnectorPartitionInfo part : parts) { @@ -461,7 +474,8 @@ private MvccSnapshot loadSnapshotInternal( // normal-read materializeLatest path — NOT a snapshot-pinned handle. Map nameToPartitionItem = Maps.newHashMap(); Map nameToLastModifiedMillis = Maps.newHashMap(); - listLatestPartitions(metadata, session, handle, nameToPartitionItem, nameToLastModifiedMillis); + listLatestPartitions(metadata, session, handle, getPartitionColumns(), + nameToPartitionItem, nameToLastModifiedMillis); return new PluginDrivenMvccSnapshot(connectorSnapshot, nameToPartitionItem, nameToLastModifiedMillis, null); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java index 3229689f0175d4..9bb9e73b8a6de0 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java @@ -1278,7 +1278,7 @@ private void pinMvccSnapshot() throws UserException { // loaded, checks this reference's own pin), so it catches the latest-masked / @incr / MTMV-refresh // cases the version-blind analysis-time binding cannot. Per user decision 2026-07-13: throw for now; // the per-reference version-aware schema-binding refactor is tracked as D-MVCC-VERSION-SCHEMA. A - // latest / @incr / hive scan carries a null pinnedSchema -> no-op; so does a sys-table scan on a + // reference without a pinned schema is a no-op; so is a sys-table scan on a // connector that rejects sys-table time travel (resolveSysTableSnapshotPin returns empty). A sys-table // scan on a connector that DOES honor it (iceberg) is excluded inside the guard — see there. if (snapshot.isPresent() && snapshot.get() instanceof PluginDrivenMvccSnapshot) { @@ -1307,12 +1307,13 @@ private void pinMvccSnapshot() throws UserException { * (BE matches iceberg columns by id, so a rename that keeps the id is fine, a renumber / added column is * caught); {@code uniqueId < 0} (paimon has no top-level field-id) matches by name. Reader-synthesized * row-id columns ({@link Column#GLOBAL_ROWID_COL}) are skipped — they are not table columns and are - * absent from every pinned schema by construction.

+ * absent from every pinned schema by construction. Connector-synthesized columns are likewise exempt + * on a schema miss; their classification comes from the scan provider, not connector-specific names.

* *

Two no-ops, both because the guard's precondition — {@code boundColumns} and {@code pinnedSchema} * describe the SAME table — does not hold: *

    - *
  • A {@code null} pinnedSchema (latest / {@code @incr} / hive reference): nothing to compare.
    • + *
  • A {@code null} pinnedSchema (e.g. {@code @incr} or an unversioned schema): nothing to compare.
    • *
  • A {@code table} that is a {@link PluginDrivenSysExternalTable}: a sys-table scan's pin is BY * CONSTRUCTION resolved off the SOURCE table ({@code resolveSysTableSnapshotPin}), so pinnedSchema * carries the source's columns while boundColumns carries the sys table's synthetic ones @@ -1325,7 +1326,7 @@ private void pinMvccSnapshot() throws UserException { * The guard keeps full strength for real MVCC tables: {@link PluginDrivenSysExternalTable} and * {@link PluginDrivenMvccExternalTable} are sibling subclasses, so no normal table matches.

    */ - static void assertBoundColumnsResolveInPinnedSchema(List boundColumns, + void assertBoundColumnsResolveInPinnedSchema(List boundColumns, SchemaCacheValue pinnedSchema, TableIf table) throws UserException { if (pinnedSchema == null || table instanceof PluginDrivenSysExternalTable) { return; @@ -1349,7 +1350,9 @@ static void assertBoundColumnsResolveInPinnedSchema(List boundColumns, boolean resolved = bound.getUniqueId() >= 0 ? pinnedFieldIds.contains(bound.getUniqueId()) : pinnedNames.contains(bound.getName().toLowerCase()); - if (!resolved) { + // Request-scoped synthesized columns do not belong to a physical schema generation. + // GENERATED columns still read file data and must pass the schema check. + if (!resolved && classifyColumnByConnector(bound.getName()) != ConnectorColumnCategory.SYNTHESIZED) { throw new UserException("Reading the same table at multiple versions with different schemas " + "in one statement is not supported yet: column '" + bound.getName() + "' of table '" + tableName + "' is bound at a different version than the one this reference scans. " diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java index ba82c074d541a6..b7d22864ee6808 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java @@ -567,6 +567,112 @@ public void testLoadSnapshotEmptyMaterializes() { "the latest pin must carry the materialized partition view"); } + @Test + public void testLatestSnapshotCarriesConnectorBoundSchema() { + Fixture f = Fixture.timeTravel(); + ConnectorMvccSnapshot snapshot = ConnectorMvccSnapshot.builder() + .snapshotId(PINNED_SNAPSHOT_ID).schemaId(Fixture.TT_SCHEMA_ID).build(); + Mockito.when(f.metadata.beginQuerySnapshot(f.session, f.handle)) + .thenReturn(Optional.of(snapshot)); + Mockito.when(f.metadata.applySnapshot(f.session, f.handle, snapshot)) + .thenReturn(f.pinnedHandle); + + PluginDrivenMvccSnapshot pin = (PluginDrivenMvccSnapshot) + f.table.loadSnapshot(Optional.empty(), Optional.empty()); + + Assertions.assertNotNull(pin.getPinnedSchema()); + Assertions.assertEquals("v1", pin.getPinnedSchema().getSchema().get(0).getName(), + "a connector's latest schema generation must survive the data fence materialization"); + } + + @Test + public void testLatestRangeUsesPinnedPartitionTypeAndArity() { + Fixture f = pinnedPartitionFixture(); + Mockito.when(f.metadata.getMvccPartitionView(f.session, f.pinnedHandle)) + .thenReturn(Optional.of(rangeView(rangePart("p1", "10", "20", FRESH_555)))); + PluginDrivenMvccSnapshot pin = (PluginDrivenMvccSnapshot) + f.table.loadSnapshot(Optional.empty(), Optional.empty()); + RangePartitionItem item = (RangePartitionItem) pin.getNameToPartitionItem().get("p1"); + Assertions.assertEquals(1, item.getItems().lowerEndpoint().getKeys().size()); + Assertions.assertEquals(Type.INT, item.getItems().lowerEndpoint().getKeys().get(0).getType()); + Assertions.assertEquals("10", item.getItems().lowerEndpoint().getKeys().get(0).getStringValue()); + } + + @Test + public void testLatestListUsesPinnedPartitionTypeAndArity() { + checkLatestPinnedList(false); + } + + @Test + public void testLatestUnpartitionedViewUsesPinnedListColumns() { + checkLatestPinnedList(true); + } + + private void checkLatestPinnedList(boolean unpartitionedView) { + Fixture f = pinnedPartitionFixture(); + if (unpartitionedView) { + // The ambient schema has no partition columns, but this pin still owns a LIST partition. + f.latestCacheValue.getPartitionColumns().clear(); + Mockito.when(f.metadata.getMvccPartitionView(f.session, f.pinnedHandle)) + .thenReturn(Optional.of(ConnectorMvccPartitionView.unpartitioned())); + } + Mockito.when(f.metadata.listPartitions(Mockito.eq(f.session), Mockito.eq(f.pinnedHandle), Mockito.any())) + .thenReturn(Collections.singletonList(cpi("key=10", TS_2024_01_01))); + PluginDrivenMvccSnapshot pin = (PluginDrivenMvccSnapshot) + f.table.loadSnapshot(Optional.empty(), Optional.empty()); + Assertions.assertEquals(1, pin.getNameToPartitionItem().size()); + PartitionKey key = ((ListPartitionItem) pin.getNameToPartitionItem().get("key=10")).getItems().get(0); + Assertions.assertEquals(1, key.getKeys().size()); + Assertions.assertEquals(Type.INT, key.getKeys().get(0).getType()); + Assertions.assertEquals("10", key.getKeys().get(0).getStringValue()); + } + + @Test + public void testLatestPinnedSchemaRejectsCaseInsensitiveDuplicateColumns() { + checkLatestPinnedDuplicateColumns(false); + } + + @Test + public void testLatestPinnedSchemaRejectsIdentifierMappingCollisions() { + checkLatestPinnedDuplicateColumns(true); + } + + private void checkLatestPinnedDuplicateColumns(boolean mappedCollision) { + Fixture f = pinnedPartitionFixture(); + ConnectorMvccSnapshot snapshot = ConnectorMvccSnapshot.builder() + .snapshotId(PINNED_SNAPSHOT_ID).schemaId(Fixture.TT_SCHEMA_ID).build(); + ConnectorTableSchema schema = new ConnectorTableSchema("REMOTE_TBL", Arrays.asList( + new ConnectorColumn("id", ConnectorType.of("INT"), "", true, null), + new ConnectorColumn(mappedCollision ? "remote_id" : "ID", ConnectorType.of("INT"), "", true, null)), + "", Collections.emptyMap()); + Mockito.when(f.metadata.getTableSchema(f.session, f.pinnedHandle, snapshot)).thenReturn(schema); + if (mappedCollision) { + Mockito.when(f.metadata.fromRemoteColumnName(f.session, "REMOTE_DB", "REMOTE_TBL", "remote_id")) + .thenReturn("ID"); + } + IllegalArgumentException error = Assertions.assertThrows(IllegalArgumentException.class, + () -> f.table.loadSnapshot(Optional.empty(), Optional.empty())); + Assertions.assertEquals("Duplicate column name found: ID", error.getMessage()); + Mockito.verify(f.metadata, Mockito.never()).getMvccPartitionView(Mockito.any(), Mockito.any()); + Mockito.verify(f.metadata, Mockito.never()).listPartitions(Mockito.any(), Mockito.any(), Mockito.any()); + } + + private Fixture pinnedPartitionFixture() { + Fixture f = Fixture.timeTravel(); + // The local pin predates a partition type/arity change in the ambient latest schema. + f.latestCacheValue.getPartitionColumns().clear(); + f.latestCacheValue.getPartitionColumns().add(new Column("key", Type.DATEV2)); + f.latestCacheValue.getPartitionColumns().add(new Column("added", Type.INT)); + ConnectorMvccSnapshot snapshot = ConnectorMvccSnapshot.builder() + .snapshotId(PINNED_SNAPSHOT_ID).schemaId(Fixture.TT_SCHEMA_ID).build(); + Mockito.when(f.metadata.beginQuerySnapshot(f.session, f.handle)).thenReturn(Optional.of(snapshot)); + ConnectorTableSchema schema = new ConnectorTableSchema("REMOTE_TBL", + Collections.singletonList(new ConnectorColumn("key", ConnectorType.of("INT"), "", true, null)), + "", Collections.singletonMap(ConnectorTableSchema.PARTITION_COLUMNS_KEY, "key")); + Mockito.when(f.metadata.getTableSchema(f.session, f.pinnedHandle, snapshot)).thenReturn(schema); + return f; + } + @Test public void testInitialLatestPartitionAccountingUsesPinnedHandle() { Fixture f = Fixture.partitioned(); @@ -1514,7 +1620,7 @@ private static Fixture build(List partitions, boolean ti // string-key path) — the LATEST schema. List schema = Collections.singletonList(new Column("dt", partitionColType)); PluginDrivenSchemaCacheValue latestCacheValue = new PluginDrivenSchemaCacheValue( - schema, schema, Collections.singletonList("dt")); + schema, new ArrayList<>(schema), Collections.singletonList("dt")); ConnectorMvccSnapshot resolvedSnapshot = ConnectorMvccSnapshot.builder() .snapshotId(7L).schemaId(TT_SCHEMA_ID).build(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeMvccSchemaGuardTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeMvccSchemaGuardTest.java index 6a87d1db22c80b..8d5e2333607f32 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeMvccSchemaGuardTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeMvccSchemaGuardTest.java @@ -21,6 +21,7 @@ import org.apache.doris.catalog.TableIf; import org.apache.doris.catalog.Type; import org.apache.doris.common.UserException; +import org.apache.doris.connector.spi.scan.ConnectorColumnCategory; import org.apache.doris.datasource.SchemaCacheValue; import org.apache.doris.datasource.mvcc.PluginDrivenMvccExternalTable; import org.apache.doris.datasource.plugin.PluginDrivenSysExternalTable; @@ -60,6 +61,46 @@ private static TableIf table() { return t; } + private static PluginDrivenScanNode node() { + PluginDrivenScanNode node = Mockito.mock(PluginDrivenScanNode.class, Mockito.CALLS_REAL_METHODS); + Mockito.doReturn(ConnectorColumnCategory.DEFAULT).when(node).classifyColumnByConnector(Mockito.anyString()); + return node; + } + + @Test + public void connectorSyntheticColumnIsNotPartOfPinnedPhysicalSchema() throws UserException { + PluginDrivenScanNode node = node(); + Mockito.doReturn(ConnectorColumnCategory.SYNTHESIZED).when(node) + .classifyColumnByConnector("synthetic_position"); + List bound = Arrays.asList(col("id", 1), col("synthetic_position", -1)); + SchemaCacheValue pinned = schema(col("id", 1)); + + Assertions.assertDoesNotThrow(() -> node.assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); + Mockito.verify(node, Mockito.never()).classifyColumnByConnector("id"); + } + + @Test + public void connectorSyntheticColumnDoesNotHideRealSchemaSkew() throws UserException { + PluginDrivenScanNode node = node(); + Mockito.doReturn(ConnectorColumnCategory.SYNTHESIZED).when(node) + .classifyColumnByConnector("synthetic_position"); + List bound = Arrays.asList(col("synthetic_position", -1), col("added", 9)); + UserException error = Assertions.assertThrows(UserException.class, () -> + node.assertBoundColumnsResolveInPinnedSchema(bound, schema(col("id", 1)), table())); + Assertions.assertTrue(error.getMessage().contains("'added'"), error.getMessage()); + } + + @Test + public void connectorGeneratedColumnStillRequiresPinnedSchemaMatch() throws UserException { + PluginDrivenScanNode node = node(); + Mockito.doReturn(ConnectorColumnCategory.GENERATED).when(node).classifyColumnByConnector("generated_value"); + Column generated = col("generated_value", 9); + generated.setIsVisible(false); + generated.setReservedPassthrough(true); + Assertions.assertThrows(UserException.class, () -> node.assertBoundColumnsResolveInPinnedSchema( + Collections.singletonList(generated), schema(col("id", 1)), table())); + } + @Test public void fieldIdRenumberBetweenBoundAndScannedVersionThrows() throws UserException { // The tuple was bound at LATEST where column `c` has field-id 7, but THIS reference scans a pinned @@ -69,7 +110,7 @@ public void fieldIdRenumberBetweenBoundAndScannedVersionThrows() throws UserExce List bound = Collections.singletonList(col("c", 7)); SchemaCacheValue pinned = schema(col("c", 5)); UserException e = Assertions.assertThrows(UserException.class, - () -> PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); + () -> node().assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); Assertions.assertTrue(e.getMessage().contains("multiple versions"), e.getMessage()); Assertions.assertTrue(e.getMessage().contains("'c'"), e.getMessage()); } @@ -81,7 +122,7 @@ public void columnAddedAfterScannedVersionThrows() throws UserException { List bound = Arrays.asList(col("id", 1), col("added", 9)); SchemaCacheValue pinned = schema(col("id", 1)); // pinned version predates `added` UserException e = Assertions.assertThrows(UserException.class, - () -> PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); + () -> node().assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); Assertions.assertTrue(e.getMessage().contains("'added'"), e.getMessage()); } @@ -93,7 +134,7 @@ public void nameMissWhenNoFieldIdThrows() throws UserException { List bound = Collections.singletonList(col("newname", -1)); SchemaCacheValue pinned = schema(col("oldname", -1)); Assertions.assertThrows(UserException.class, - () -> PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); + () -> node().assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); } @Test @@ -104,7 +145,7 @@ public void fieldIdStableRenameResolvesByIdNoThrow() throws UserException { List bound = Collections.singletonList(col("newname", 5)); SchemaCacheValue pinned = schema(col("oldname", 5), col("other", 6)); Assertions.assertDoesNotThrow(() -> - PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); + node().assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); } @Test @@ -113,7 +154,7 @@ public void subsetProjectionAllResolvedNoThrow() throws UserException { List bound = Arrays.asList(col("a", 1), col("c", 3)); SchemaCacheValue pinned = schema(col("a", 1), col("b", 2), col("c", 3)); Assertions.assertDoesNotThrow(() -> - PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); + node().assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); } @Test @@ -122,7 +163,7 @@ public void nameMatchWhenNoFieldIdNoThrow() throws UserException { List bound = Arrays.asList(col("a", -1), col("b", -1)); SchemaCacheValue pinned = schema(col("a", -1), col("b", -1), col("c", -1)); Assertions.assertDoesNotThrow(() -> - PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); + node().assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); } @Test @@ -132,7 +173,7 @@ public void nullPinnedSchemaIsNoOp() throws UserException { // reference is excluded by TYPE, not by a null schema -- see sysTableIsExcludedNoThrow.) List bound = Collections.singletonList(col("anything", 42)); Assertions.assertDoesNotThrow(() -> - PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, null, table())); + node().assertBoundColumnsResolveInPinnedSchema(bound, null, table())); } @Test @@ -146,7 +187,7 @@ public void rowIdColumnIsExcludedNoThrow() throws UserException { col(Column.GLOBAL_ROWID_COL + "tag_branch_table", Integer.MAX_VALUE)); SchemaCacheValue pinned = schema(col("id", 1)); Assertions.assertDoesNotThrow(() -> - PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); + node().assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); } @Test @@ -159,7 +200,7 @@ public void rowIdBeforeSkewedColumnStillThrows() throws UserException { col("added", 9)); SchemaCacheValue pinned = schema(col("id", 1)); Assertions.assertThrows(UserException.class, - () -> PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); + () -> node().assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); } @Test @@ -185,7 +226,7 @@ public void sysTableIsExcludedNoThrow() throws UserException { Mockito.when(sysTable.getName()).thenReturn("db.t$position_deletes"); Assertions.assertDoesNotThrow(() -> - PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, pinned, sysTable)); + node().assertBoundColumnsResolveInPinnedSchema(bound, pinned, sysTable)); } @Test @@ -198,6 +239,6 @@ public void normalMvccTableWithSameShapeStillThrows() throws UserException { SchemaCacheValue pinned = schema(col("id", -1), col("name", -1)); Assertions.assertThrows(UserException.class, - () -> PluginDrivenScanNode.assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); + () -> node().assertBoundColumnsResolveInPinnedSchema(bound, pinned, table())); } } diff --git a/regression-test/data/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.out b/regression-test/data/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.out new file mode 100644 index 00000000000000..9233f0f07c5937 --- /dev/null +++ b/regression-test/data/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.out @@ -0,0 +1,16 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !plain_schema -- +1 base + +-- !options_schema -- +1 base + +-- !branch_schema -- +1 base + +-- !native_precision -- +1 + +-- !jni_precision -- +1 + diff --git a/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy b/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy new file mode 100644 index 00000000000000..54e63a7e32b610 --- /dev/null +++ b/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy @@ -0,0 +1,147 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +suite("test_paimon_schema_only_snapshot_precision", "p0,external,paimon") { + String enabled = context.config.otherConfigs.get("enablePaimonTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable paimon test") + return + } + + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String catalogName = "test_paimon_schema_only_snapshot_precision" + String dbName = "paimon_schema_only_snapshot_precision_db" + String tableName = "schema_only_timeline" + String branchName = "schema_only_branch" + + def latestSnapshotId = { + List> rows = spark_paimon """ + select snapshot_id + from paimon.${dbName}.`${tableName}\$snapshots` + order by snapshot_id desc + limit 1 + """ + assertEquals(1, rows.size()) + return rows[0][0].toString() + } + + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + 'type'='paimon', + 'warehouse'='s3://warehouse/wh', + 's3.endpoint'='http://${externalEnvIp}:${minioPort}', + 's3.access_key'='admin', + 's3.secret_key'='password', + 's3.path.style.access'='true', + 'paimon.table-option.read.batch-size'='64', + 'meta.cache.paimon.table.ttl-second'='0' + ) + """ + + try { + // Explicit NTZ makes this exercise predicate pushdown instead of LTZ residual filtering. + spark_paimon_multi """ + create database if not exists paimon.${dbName}; + drop table if exists paimon.${dbName}.${tableName}; + create table paimon.${dbName}.${tableName} ( + id int, + old_name string, + event_time timestamp_ntz + ) using paimon + tblproperties ('file.format'='parquet'); + insert into paimon.${dbName}.${tableName} + values (1, 'base', timestamp_ntz '2024-01-01 00:00:00.123456'); + """ + String dataSnapshotId = latestSnapshotId() + spark_paimon_multi """ + call paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => 'schema_base' + ); + call paimon.sys.create_branch( + '${dbName}.${tableName}', + '${branchName}', + 'schema_base' + ); + alter table paimon.${dbName}.`${tableName}\$branch_${branchName}` + rename column old_name to branch_name; + alter table paimon.${dbName}.${tableName} + rename column old_name to current_name; + """ + + // A schema-only rename must leave the data snapshot unchanged; otherwise these queries + // would not exercise the split between current schema binding and snapshot-pinned data. + assertEquals(dataSnapshotId, latestSnapshotId()) + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh table ${tableName}""" + + order_qt_plain_schema """ + select id, current_name from ${tableName} order by id + """ + order_qt_options_schema """ + select id, current_name + from ${tableName}@options('scan.plan-sort-partition'='true') + order by id + """ + order_qt_branch_schema """ + select id, branch_name + from ${tableName}@branch(${branchName}) + order by id + """ + + // Sub-millisecond precision must survive FE predicate conversion or Paimon's file + // statistics can reject the only matching file before either reader sees it. + sql """set force_jni_scanner=false""" + order_qt_native_precision """ + select id from ${tableName} + where event_time = cast('2024-01-01 00:00:00.123456' as datetime(6)) + """ + sql """set force_jni_scanner=true""" + order_qt_jni_precision """ + select id from ${tableName} + where event_time = cast('2024-01-01 00:00:00.123456' as datetime(6)) + """ + + // Catalog reader policy must survive schema restoration even when it matched the old + // physical value. Relation overrides still take precedence for both reader paths. + String readerTable = "${tableName}_reader_options" + spark_paimon_multi """ + drop table if exists paimon.${dbName}.${readerTable}; + create table paimon.${dbName}.${readerTable} (id int) using paimon + tblproperties ('file.format'='parquet', 'read.batch-size'='64'); + insert into paimon.${dbName}.${readerTable} values (1); + """ + assertEquals([[1]], sql("select id from ${readerTable}")) + spark_paimon """ + alter table paimon.${dbName}.${readerTable} set tblproperties ('read.batch-size'='0') + """ + [false, true].each { forceJni -> + sql "set force_jni_scanner=${forceJni}" + assertEquals([[1]], sql("select id from ${readerTable}")) + assertEquals([[1]], sql(""" + select id from ${readerTable}@options('read.batch-size'='32') + """)) + } + } finally { + sql """set force_jni_scanner=false""" + sql """drop catalog if exists ${catalogName}""" + } +}