diff --git a/docs/docs/program-api/catalog-api.md b/docs/docs/program-api/catalog-api.md index 3afb60941380..6d7d02326311 100644 --- a/docs/docs/program-api/catalog-api.md +++ b/docs/docs/program-api/catalog-api.md @@ -28,6 +28,13 @@ Use `Catalog` to manage databases and tables, and to load a `Table` for reading Catalog configuration and table options have different scopes: configure the warehouse and metastore on the catalog; put storage and read/write behavior in the table schema's options. +For file-store tables, `getTable()` loads the table definition without initializing the runtime +store. Metadata access and `copy(options)` can therefore succeed even if stored options are invalid +for reading or writing. Runtime options are validated when a scan, reader, writer, committer, or +explicit `FileStoreTable.store()` call initializes the store. `copy(options)` still rejects immutable +option changes immediately. CREATE and ALTER validate the resulting schema before committing it, +so an invalid existing configuration can be inspected, repaired, or dropped. + ## Set up the examples Add the [Java dependency](java-api#dependency) and save the diff --git a/paimon-common/src/main/java/org/apache/paimon/table/BucketMode.java b/paimon-common/src/main/java/org/apache/paimon/table/BucketMode.java index 4740ba8a0080..23d0cd1b6366 100644 --- a/paimon-common/src/main/java/org/apache/paimon/table/BucketMode.java +++ b/paimon-common/src/main/java/org/apache/paimon/table/BucketMode.java @@ -71,4 +71,19 @@ public enum BucketMode { public static final int UNAWARE_BUCKET = 0; public static final int POSTPONE_BUCKET = -2; + + public static BucketMode fromOptions( + int bucket, boolean hasPrimaryKey, boolean crossPartitionUpdate) { + if (!hasPrimaryKey) { + return bucket == -1 ? BUCKET_UNAWARE : HASH_FIXED; + } + switch (bucket) { + case -2: + return POSTPONE_MODE; + case -1: + return crossPartitionUpdate ? KEY_DYNAMIC : HASH_DYNAMIC; + default: + return HASH_FIXED; + } + } } diff --git a/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java index b3310d537743..1fe6bd51284f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java @@ -64,7 +64,7 @@ public AppendOnlyFileStore( @Override public BucketMode bucketMode() { - return options.bucket() == -1 ? BucketMode.BUCKET_UNAWARE : BucketMode.HASH_FIXED; + return BucketMode.fromOptions(options.bucket(), false, false); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java index 7675fb752523..7aa258c04e3a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java @@ -91,15 +91,7 @@ public KeyValueFileStore( @Override public BucketMode bucketMode() { - int bucket = options.bucket(); - switch (bucket) { - case -2: - return BucketMode.POSTPONE_MODE; - case -1: - return crossPartitionUpdate ? BucketMode.KEY_DYNAMIC : BucketMode.HASH_DYNAMIC; - default: - return BucketMode.HASH_FIXED; - } + return BucketMode.fromOptions(options.bucket(), true, crossPartitionUpdate); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java index f3882b5b66bb..88390954d1ac 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java @@ -156,6 +156,10 @@ void alterDatabase(String name, List changes, boolean ignoreIfNo * *

System tables can be got by '$' splitter. * + *

File-store tables load their definition without initializing the runtime store. Runtime + * option validation happens when a read, scan, write, commit, or explicit store access first + * initializes that store. Loading a table is not a runtime configuration check. + * * @param identifier Path of the table * @return The requested table * @throws TableNotExistException if the target does not exist diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index 459713687e88..4bcb1ad91a90 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java @@ -19,6 +19,7 @@ package org.apache.paimon.table; import org.apache.paimon.CoreOptions; +import org.apache.paimon.FileStore; import org.apache.paimon.Snapshot; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.consumer.ConsumerManager; @@ -49,7 +50,9 @@ import org.apache.paimon.table.sink.RowKindGenerator; import org.apache.paimon.table.sink.TableCommitImpl; import org.apache.paimon.table.sink.WriteSelector; +import org.apache.paimon.table.source.DataTableScan; import org.apache.paimon.table.source.DataTableStreamScan; +import org.apache.paimon.table.source.InnerTableRead; import org.apache.paimon.table.source.SplitGenerator; import org.apache.paimon.table.source.StreamDataTableScan; import org.apache.paimon.table.source.snapshot.SnapshotReader; @@ -106,6 +109,10 @@ abstract class AbstractFileStoreTable implements FileStoreTable { // Track explicit copy() keys, including removals, separately from persisted schema options. @Nullable private Set appliedDynamicOptionKeys; + // Runtime option mutations, like the store, do not survive table serialization. + @Nullable private transient volatile CoreOptions lazyOptions; + private transient volatile FileStore lazyStore; + @Nullable protected transient SegmentsCache manifestCache; @Nullable protected transient SegmentsCache manifestSidecarCache; @Nullable protected transient Cache snapshotCache; @@ -134,9 +141,11 @@ public String currentBranch() { } @Override - public void setManifestCache(SegmentsCache manifestCache) { + public synchronized void setManifestCache(SegmentsCache manifestCache) { this.manifestCache = manifestCache; - store().setManifestCache(manifestCache); + if (lazyStore != null) { + lazyStore.setManifestCache(manifestCache); + } } @Nullable @@ -146,9 +155,11 @@ public SegmentsCache getManifestCache() { } @Override - public void setManifestSidecarCache(SegmentsCache manifestSidecarCache) { + public synchronized void setManifestSidecarCache(SegmentsCache manifestSidecarCache) { this.manifestSidecarCache = manifestSidecarCache; - store().setManifestSidecarCache(manifestSidecarCache); + if (lazyStore != null) { + lazyStore.setManifestSidecarCache(manifestSidecarCache); + } } @Nullable @@ -158,9 +169,11 @@ public SegmentsCache getManifestSidecarCache() { } @Override - public void setSnapshotCache(Cache cache) { + public synchronized void setSnapshotCache(Cache cache) { this.snapshotCache = cache; - store().setSnapshotCache(cache); + if (lazyStore != null) { + lazyStore.setSnapshotCache(cache); + } } @Override @@ -333,6 +346,62 @@ public StreamDataTableScan newStreamScan() { protected abstract BiConsumer nonPartitionFilterConsumer(); + /** Validate the captured schema and install caches before publishing the runtime store. */ + protected final FileStore runtimeStore() { + FileStore store = lazyStore; + if (store == null) { + synchronized (this) { + store = lazyStore; + if (store == null) { + CoreOptions options = coreOptions(); + TableSchema runtimeSchema = tableSchema.copy(options.toMap()); + Set dynamicOptionKeys = + new HashSet<>(options.toConfiguration().dynamicOptions().keySet()); + if (appliedDynamicOptionKeys != null) { + dynamicOptionKeys.addAll(appliedDynamicOptionKeys); + } + SchemaValidation.validateTableSchema( + withoutIcebergMirror(runtimeSchema), dynamicOptionKeys); + store = createStore(runtimeSchema); + if (manifestCache != null) { + store.setManifestCache(manifestCache); + } + if (manifestSidecarCache != null) { + store.setManifestSidecarCache(manifestSidecarCache); + } + if (snapshotCache != null) { + store.setSnapshotCache(snapshotCache); + } + lazyStore = store; + } + } + } + return store; + } + + protected abstract FileStore createStore(TableSchema runtimeSchema); + + @Override + public InnerTableRead newRead() { + store(); + return createRead(); + } + + protected abstract InnerTableRead createRead(); + + @Override + public DataTableScan newScan() { + return newScan(FileStoreTable::newSnapshotReader); + } + + @Override + public DataTableScan newScan(SnapshotReaderFactory snapshotReaderFactory) { + store(); + return createScan(snapshotReaderFactory); + } + + protected abstract DataTableScan createScan(SnapshotReaderFactory snapshotReaderFactory); + @Override public FileStoreTable copy(Map dynamicOptions) { checkImmutability(dynamicOptions); @@ -396,9 +465,6 @@ protected FileStoreTable copyInternal( tryTimeTravel(newOptions, mergedDynamicOptionKeys).orElse(newTableSchema); } - // validate schema with new options - SchemaValidation.validateTableSchema( - withoutIcebergMirror(newTableSchema), dynamicOptions.keySet()); if (new CoreOptions(tableSchema.options()) .toConfiguration() .get(IcebergOptions.METADATA_ICEBERG_STORAGE) @@ -424,7 +490,6 @@ public FileStoreTable copyWithLatestSchema() { Map mergedOptions = new HashMap<>(latestSchema.options()); mergedOptions.putAll(tableSchema.options()); TableSchema newTableSchema = latestSchema.copy(mergedOptions); - SchemaValidation.validateTableSchema(withoutIcebergMirror(newTableSchema)); return copy(newTableSchema); } else { return this; @@ -472,7 +537,25 @@ public SchemaManager schemaManager() { @Override public CoreOptions coreOptions() { - return store().options(); + CoreOptions options = lazyOptions; + if (options == null) { + synchronized (this) { + options = lazyOptions; + if (options == null) { + options = new CoreOptions(tableSchema.options()); + lazyOptions = options; + } + } + } + return options; + } + + @Override + public BucketMode bucketMode() { + return BucketMode.fromOptions( + coreOptions().bucket(), + !tableSchema.primaryKeys().isEmpty(), + tableSchema.crossPartitionUpdate()); } @Override @@ -492,12 +575,19 @@ public TableSchema schema() { @Override public SnapshotManager snapshotManager() { - return store().snapshotManager(); + CoreOptions options = coreOptions(); + return new SnapshotManager( + fileIO, + options.path(), + options.branch(), + catalogEnvironment.snapshotLoader(), + snapshotCache); } @Override public ChangelogManager changelogManager() { - return store().changelogManager(); + CoreOptions options = coreOptions(); + return new ChangelogManager(fileIO, options.path(), options.branch()); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java index 5c7579908c8b..553dcc7df737 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java @@ -58,8 +58,6 @@ public class AppendOnlyFileStoreTable extends AbstractFileStoreTable { private static final long serialVersionUID = 1L; - private transient AppendOnlyFileStore lazyStore; - AppendOnlyFileStoreTable(FileIO fileIO, Path path, TableSchema tableSchema) { this(fileIO, path, tableSchema, CatalogEnvironment.empty()); } @@ -74,20 +72,21 @@ public AppendOnlyFileStoreTable( @Override public AppendOnlyFileStore store() { - if (lazyStore == null) { - lazyStore = - new AppendOnlyFileStore( - fileIO, - schemaManager(), - tableSchema, - new CoreOptions(tableSchema.options()), - tableSchema.logicalPartitionType(), - tableSchema.logicalBucketKeyType(), - tableSchema.logicalRowType().notNull(), - name(), - catalogEnvironment); - } - return lazyStore; + return (AppendOnlyFileStore) runtimeStore(); + } + + @Override + protected AppendOnlyFileStore createStore(TableSchema runtimeSchema) { + return new AppendOnlyFileStore( + fileIO, + schemaManager(), + runtimeSchema, + coreOptions(), + runtimeSchema.logicalPartitionType(), + runtimeSchema.logicalBucketKeyType(), + runtimeSchema.logicalRowType().notNull(), + name(), + catalogEnvironment); } @Override @@ -113,7 +112,7 @@ protected BiConsumer nonPartitionFilterConsumer() { } @Override - public InnerTableRead newRead() { + protected InnerTableRead createRead() { List> providerFactories = new ArrayList<>(); if (coreOptions().dataEvolutionEnabled()) { // add data evolution first @@ -139,12 +138,7 @@ public InnerTableRead newRead() { } @Override - public DataTableScan newScan() { - return newScan(FileStoreTable::newSnapshotReader); - } - - @Override - public DataTableScan newScan(SnapshotReaderFactory snapshotReaderFactory) { + protected DataTableScan createScan(SnapshotReaderFactory snapshotReaderFactory) { CoreOptions options = coreOptions(); AppendBatchTableScan scan = new AppendBatchTableScan( diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java index ddaa17407624..933cdbb6974f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java @@ -116,6 +116,7 @@ default Optional comment() { TableSchema schema(); + /** Initializes the runtime store, validating this table's effective schema and options. */ FileStore store(); /** Creates a scan with a customized snapshot reader. */ @@ -123,6 +124,10 @@ default Optional comment() { CatalogEnvironment catalogEnvironment(); + /** + * Copies the table definition with dynamic options. Immutable option changes fail immediately; + * runtime configuration is validated when the copied table initializes its own store. + */ @Override FileStoreTable copy(Map dynamicOptions); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java index 97c541175bd3..ac01b165cec0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java @@ -60,8 +60,6 @@ public class PrimaryKeyFileStoreTable extends AbstractFileStoreTable { private static final long serialVersionUID = 1L; - private transient KeyValueFileStore lazyStore; - @VisibleForTesting PrimaryKeyFileStoreTable(FileIO fileIO, Path path, TableSchema tableSchema) { this(fileIO, path, tableSchema, CatalogEnvironment.empty()); @@ -77,49 +75,49 @@ public PrimaryKeyFileStoreTable( @Override public KeyValueFileStore store() { - if (lazyStore == null) { - RowType rowType = tableSchema.logicalRowType(); - CoreOptions options = CoreOptions.fromMap(tableSchema.options()); - KeyValueFieldsExtractor extractor = - PrimaryKeyTableUtils.PrimaryKeyFieldsExtractor.EXTRACTOR; + return (KeyValueFileStore) runtimeStore(); + } - RowType keyType = new RowType(extractor.keyFields(tableSchema)); - List deleteNotNullFields = deleteNotNullFields(options); - RowType valueType = - deleteNotNullFields == null - ? rowType - : rowType.copy( - rowType.getFields().stream() - .map( - field -> - deleteNotNullFields.contains( - field.name()) - ? field - : field.copy(true)) - .collect(Collectors.toList())); + @Override + protected KeyValueFileStore createStore(TableSchema runtimeSchema) { + RowType rowType = runtimeSchema.logicalRowType(); + CoreOptions options = coreOptions(); + KeyValueFieldsExtractor extractor = + PrimaryKeyTableUtils.PrimaryKeyFieldsExtractor.EXTRACTOR; - MergeFunctionFactory mfFactory = - PrimaryKeyTableUtils.createMergeFunctionFactory(tableSchema); - if (options.needLookup()) { - mfFactory = LookupMergeFunction.wrap(mfFactory, options, keyType, valueType); - } + RowType keyType = new RowType(extractor.keyFields(runtimeSchema)); + List deleteNotNullFields = deleteNotNullFields(options); + RowType valueType = + deleteNotNullFields == null + ? rowType + : rowType.copy( + rowType.getFields().stream() + .map( + field -> + deleteNotNullFields.contains(field.name()) + ? field + : field.copy(true)) + .collect(Collectors.toList())); - lazyStore = - new KeyValueFileStore( - fileIO(), - schemaManager(), - tableSchema, - tableSchema.crossPartitionUpdate(), - options, - tableSchema.logicalPartitionType(), - keyType, - valueType, - extractor, - mfFactory, - name(), - catalogEnvironment); + MergeFunctionFactory mfFactory = + PrimaryKeyTableUtils.createMergeFunctionFactory(runtimeSchema); + if (options.needLookup()) { + mfFactory = LookupMergeFunction.wrap(mfFactory, options, keyType, valueType); } - return lazyStore; + + return new KeyValueFileStore( + fileIO(), + schemaManager(), + runtimeSchema, + runtimeSchema.crossPartitionUpdate(), + options, + runtimeSchema.logicalPartitionType(), + keyType, + valueType, + extractor, + mfFactory, + name(), + catalogEnvironment); } @Override @@ -164,7 +162,7 @@ protected BiConsumer nonPartitionFilterConsumer() { } @Override - public InnerTableRead newRead() { + protected InnerTableRead createRead() { return new KeyValueTableRead( () -> store().newRead(), () -> store().newBatchRawFileRead(), @@ -174,12 +172,7 @@ public InnerTableRead newRead() { } @Override - public DataTableScan newScan() { - return newScan(FileStoreTable::newSnapshotReader); - } - - @Override - public DataTableScan newScan(SnapshotReaderFactory snapshotReaderFactory) { + protected DataTableScan createScan(SnapshotReaderFactory snapshotReaderFactory) { return new PrimaryKeyBatchScan( this, snapshotReaderFactory.create(this), diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java index c50bebd4399f..0e180938a160 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java @@ -178,7 +178,7 @@ public void testAlterTableWithInvalidOptions(boolean cacheEnabled) throws Except Identifier identifier = Identifier.create("test_db", "legacy_table"); FileSystemSchemaManager schemaManager = createLegacyTable(identifier, cacheEnabled); - assertThatThrownBy(() -> catalog.getTable(identifier)) + assertThatThrownBy(() -> ((FileStoreTable) catalog.getTable(identifier)).newScan()) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("primary key"); @@ -230,7 +230,7 @@ public void testAlterTableWithInvalidOptionsAndSnapshots(boolean cacheEnabled) assertThat(snapshotManager.latestSnapshot()).isNotNull(); long snapshotId = snapshotManager.latestSnapshot().id(); assertThat(snapshotManager.latestSnapshot().schemaId()).isZero(); - assertThatThrownBy(() -> catalog.getTable(identifier)) + assertThatThrownBy(() -> ((FileStoreTable) catalog.getTable(identifier)).newScan()) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("primary key"); @@ -285,7 +285,7 @@ public void testAlterTableWithInvalidOptionsOnBranch(boolean cacheEnabled) throw assertThat(mainSchemaManager.latest().get().id()).isZero(); assertThat(catalog.getTable(branchIdentifier).options()) .containsEntry("fields.seq.sequence-group", "v"); - assertThatThrownBy(() -> catalog.getTable(identifier)) + assertThatThrownBy(() -> ((FileStoreTable) catalog.getTable(identifier)).newScan()) .isInstanceOf(IllegalArgumentException.class); } @@ -298,7 +298,7 @@ public void testDropTableWithInvalidOptions(boolean cacheEnabled) throws Excepti ((FileSystemCatalog) DelegateCatalog.rootCatalog(catalog)) .getTableLocation(identifier); - assertThatThrownBy(() -> catalog.getTable(identifier)) + assertThatThrownBy(() -> ((FileStoreTable) catalog.getTable(identifier)).newScan()) .isInstanceOf(IllegalArgumentException.class); catalog.dropTable(identifier, false); @@ -338,7 +338,7 @@ public void testDropTableWithInvalidOptionsAndExternalPaths(boolean cacheEnabled CoreOptions.GLOBAL_INDEX_EXTERNAL_PATH.key(), newBranchPath.toUri().toString()); writeLegacySchema(branchSchemaManager, legacySchema, 1, branchOptions); - assertThatThrownBy(() -> catalog.getTable(identifier)) + assertThatThrownBy(() -> ((FileStoreTable) catalog.getTable(identifier)).newScan()) .isInstanceOf(IllegalArgumentException.class); catalog.dropTable(identifier, false); diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java index 98449676be11..f12cca3064d6 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java @@ -311,6 +311,13 @@ public String getUrl() { return server.url("").toString(); } + public void setTableSchema(Identifier identifier, TableSchema schema) { + TableMetadata metadata = tableMetadataStore.get(identifier.getFullName()); + tableMetadataStore.put( + identifier.getFullName(), + new TableMetadata(schema, metadata.isExternal(), metadata.uuid())); + } + public void shutdown() throws IOException { server.shutdown(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java b/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java index 0523d97f7c66..d88001fde647 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java @@ -241,6 +241,9 @@ void testDependencyReadContextPreservesCustomRestMetastore() { void testAppendTableUsesDependencyReadContext() { CatalogEnvironment environment = mock(CatalogEnvironment.class); when(environment.dependencyReadContext()).thenReturn(CatalogContext.create(new Options())); + Options options = new Options(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + options.set(CoreOptions.ROW_TRACKING_ENABLED, true); TableSchema schema = new TableSchema( 0, @@ -248,7 +251,7 @@ void testAppendTableUsesDependencyReadContext() { 0, Collections.emptyList(), Collections.emptyList(), - Collections.singletonMap(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"), + options.toMap(), null); AppendOnlyFileStoreTable table = new AppendOnlyFileStoreTable( diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java index 2c9ab22db2b8..4d4239e1a288 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java @@ -137,6 +137,24 @@ private static Stream filterPushdownParams() { keyRange)))); } + @Test + public void testMetadataCopyDefersRuntimeValidation() throws Exception { + createChainTable(options -> {}); + FileStoreTable table = loadTable(); + FileStoreTable invalid = + table.copy( + Collections.singletonMap( + CoreOptions.RECORD_LEVEL_TIME_FIELD.key(), "missing")); + assertThat(invalid).isInstanceOf(ChainTableFileStoreTable.class); + assertThat(invalid.rowType()).isEqualTo(table.rowType()); + assertThat(invalid.coreOptions().recordLevelTimeField()).isEqualTo("missing"); + assertThatThrownBy(invalid::newRead).hasMessageContaining("missing"); + assertThatThrownBy(invalid::newScan).hasMessageContaining("missing"); + assertThatThrownBy(invalid::newStreamScan).hasMessageContaining("missing"); + assertThatThrownBy(() -> invalid.newWrite("test")).hasMessageContaining("missing"); + assertThatThrownBy(() -> invalid.newCommit("test")).hasMessageContaining("missing"); + } + @ParameterizedTest @MethodSource("filterPushdownParams") public void testFilterPushdownAcrossChainMerge( diff --git a/paimon-core/src/test/java/org/apache/paimon/table/TableRuntimeValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/table/TableRuntimeValidationTest.java new file mode 100644 index 000000000000..7798a6f6142d --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/TableRuntimeValidationTest.java @@ -0,0 +1,497 @@ +/* + * 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.paimon.table; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.FileStore; +import org.apache.paimon.Snapshot; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.fs.Path; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestSidecar; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.ChangelogManager; +import org.apache.paimon.utils.InstantiationUtil; +import org.apache.paimon.utils.SegmentsCache; +import org.apache.paimon.utils.SnapshotManager; + +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for metadata loading without runtime initialization. */ +public class TableRuntimeValidationTest extends TableTestBase { + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testMetadataManagersUseSharedCoreOptions(boolean primaryKey) throws Exception { + Schema.Builder builder = Schema.newBuilder().column("id", DataTypes.INT()); + if (primaryKey) { + builder.primaryKey("id").option("bucket", "1"); + } + catalog.createTable(identifier(), builder.build(), false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + CoreOptions options = table.coreOptions(); + Path relocated = new Path(warehouse, "relocated"); + options.toConfiguration().set(CoreOptions.PATH, relocated.toString()); + options.toConfiguration().set(CoreOptions.BRANCH, "dev"); + options.toConfiguration().set(CoreOptions.RECORD_LEVEL_TIME_FIELD, "missing"); + + SnapshotManager snapshots = table.snapshotManager(); + ChangelogManager changelogs = table.changelogManager(); + assertThat(snapshots.tablePath()).isEqualTo(relocated); + assertThat(snapshots.branch()).isEqualTo("dev"); + assertThatThrownBy(table::store).hasMessageContaining("missing"); + + options.toConfiguration().remove(CoreOptions.RECORD_LEVEL_TIME_FIELD); + FileStore store = table.store(); + assertThat(snapshots.snapshotDirectory()) + .isEqualTo(store.snapshotManager().snapshotDirectory()); + assertThat(changelogs.changelogDirectory()) + .isEqualTo(store.changelogManager().changelogDirectory()); + + Path afterInitialization = new Path(warehouse, "after-initialization"); + options.toConfiguration().set(CoreOptions.PATH, afterInitialization.toString()); + options.toConfiguration().set(CoreOptions.BRANCH, "other"); + assertThat(table.snapshotManager().tablePath()).isEqualTo(afterInitialization); + assertThat(table.snapshotManager().branch()).isEqualTo("other"); + assertThat(table.snapshotManager().snapshotDirectory()) + .isEqualTo(store.snapshotManager().snapshotDirectory()); + assertThat(table.changelogManager().changelogDirectory()) + .isEqualTo(store.changelogManager().changelogDirectory()); + } + + @ParameterizedTest + @CsvSource({"false,false", "false,true", "true,false", "true,true"}) + public void testWriteOnlyOverrideForSnapshotOrdering( + boolean copyWriteOnly, boolean initializeFirst) throws Exception { + catalog.createTable( + identifier(), + Schema.newBuilder() + .column("id", DataTypes.INT()) + .primaryKey("id") + .option(CoreOptions.BUCKET.key(), "1") + .option(CoreOptions.SEQUENCE_SNAPSHOT_ORDERING.key(), "true") + .option(CoreOptions.WRITE_ONLY.key(), "true") + .build(), + false); + FileStoreTable original = (FileStoreTable) catalog.getTable(identifier()); + FileStoreTable table = original; + if (copyWriteOnly) { + table = table.copy(Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "false")); + } + if (initializeFirst) { + table.store(); + } + CoreOptions options = table.coreOptions(); + if (!copyWriteOnly) { + options.toConfiguration().set(CoreOptions.WRITE_ONLY, false); + } + options.toConfiguration().set(CoreOptions.READ_BATCH_SIZE, 32); + + assertThat(table.store().options()).isSameAs(options); + assertThat(table.store().options().writeOnly()).isFalse(); + assertThat(table.newRead()).isNotNull(); + assertThat(table.newScan()).isNotNull(); + + Map invalidOptions = new HashMap<>(table.options()); + invalidOptions.put(CoreOptions.WRITE_ONLY.key(), "false"); + FileStoreTable invalid = original.copy(original.schema().copy(invalidOptions)); + invalid.coreOptions().toConfiguration().set(CoreOptions.READ_BATCH_SIZE, 32); + assertThatThrownBy(invalid::store).hasMessageContaining("requires write-only = true"); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testSharedCoreOptions(boolean primaryKey) throws Exception { + Schema.Builder builder = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .option(CoreOptions.READ_BATCH_SIZE.key(), "1024"); + if (primaryKey) { + builder.primaryKey("id").option("bucket", "1"); + } + catalog.createTable(identifier(), builder.build(), false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + CoreOptions options = table.coreOptions(); + assertThat(table.coreOptions()).isSameAs(options); + + options.toConfiguration().set(CoreOptions.READ_BATCH_SIZE, 32); + assertThat(table.coreOptions().toConfiguration().get(CoreOptions.READ_BATCH_SIZE)) + .isEqualTo(32); + assertThat(table.store().options()).isSameAs(options); + assertThat(table.store().options().toConfiguration().get(CoreOptions.READ_BATCH_SIZE)) + .isEqualTo(32); + + table.store().options().toConfiguration().set(CoreOptions.READ_BATCH_SIZE, 64); + assertThat(table.coreOptions().toConfiguration().get(CoreOptions.READ_BATCH_SIZE)) + .isEqualTo(64); + assertThat(table.options()).containsEntry(CoreOptions.READ_BATCH_SIZE.key(), "1024"); + + FileStoreTable copied = + table.copy(Collections.singletonMap(CoreOptions.READ_BATCH_SIZE.key(), "128")); + assertThat(copied.coreOptions()).isNotSameAs(options); + assertThat(copied.store().options()).isSameAs(copied.coreOptions()); + assertThat(copied.coreOptions().toConfiguration().get(CoreOptions.READ_BATCH_SIZE)) + .isEqualTo(128); + assertThat(table.coreOptions().toConfiguration().get(CoreOptions.READ_BATCH_SIZE)) + .isEqualTo(64); + + FileStoreTable restored = + InstantiationUtil.deserializeObject( + InstantiationUtil.serializeObject(table), getClass().getClassLoader()); + assertThat(restored.coreOptions()).isNotSameAs(options); + assertThat(restored.coreOptions().toConfiguration().get(CoreOptions.READ_BATCH_SIZE)) + .isEqualTo(1024); + assertThat(restored.store().options()).isSameAs(restored.coreOptions()); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testMutableCoreOptionsAreValidatedBeforeStoreInitialization(boolean primaryKey) + throws Exception { + Schema.Builder builder = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("seq", DataTypes.INT()) + .column("v", DataTypes.INT()); + if (primaryKey) { + builder.primaryKey("id").option("bucket", "1").option("merge-engine", "partial-update"); + } + catalog.createTable(identifier(), builder.build(), false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + CoreOptions options = table.coreOptions(); + options.toConfiguration().set(CoreOptions.RECORD_LEVEL_TIME_FIELD, "missing"); + assertThat(table.rowType().getFieldNames()).containsExactly("id", "seq", "v"); + assertThatThrownBy(table::store).hasMessageContaining("missing"); + options.toConfiguration().remove(CoreOptions.RECORD_LEVEL_TIME_FIELD.key()); + + if (primaryKey) { + options.toConfiguration().setString("fields.v.aggregate-function", "max"); + assertThatThrownBy(table::newRead).hasMessageContaining("Must use sequence group"); + options.toConfiguration().setString("fields.seq.sequence-group", "v"); + } + assertThat(table.store().options()).isSameAs(options); + write(table, GenericRow.of(1, 10, 100)); + assertThat(read(table)).extracting(row -> row.getInt(2)).containsExactly(100); + } + + @ParameterizedTest + @CsvSource({"false,false", "true,false", "false,true"}) + public void testAllRuntimeEntrypointsValidate(boolean primaryKey, boolean dataEvolution) + throws Exception { + Schema.Builder builder = Schema.newBuilder().column("id", DataTypes.INT()); + if (primaryKey) { + builder.primaryKey("id").option("bucket", "1"); + } + if (dataEvolution) { + builder.option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") + .option("bucket", "-1"); + } + catalog.createTable(identifier(), builder.build(), false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + table.store(); + FileStoreTable invalid = + table.copy( + Collections.singletonMap( + CoreOptions.RECORD_LEVEL_TIME_FIELD.key(), "missing")); + assertThat(invalid.rowType().getFieldNames()).containsExactly("id"); + assertThat(invalid.coreOptions().recordLevelTimeField()).isEqualTo("missing"); + assertThat(invalid.snapshotManager().latestSnapshot()).isNull(); + assertThat(invalid.tagManager().allTagNames()).isEmpty(); + assertThatThrownBy(invalid::store).hasMessageContaining("missing"); + assertThatThrownBy(invalid::newRead).hasMessageContaining("missing"); + assertThatThrownBy(invalid::newScan).hasMessageContaining("missing"); + assertThatThrownBy(() -> invalid.newScan(t -> null)).hasMessageContaining("missing"); + assertThatThrownBy(invalid::newStreamScan).hasMessageContaining("missing"); + assertThatThrownBy(() -> invalid.newWrite("test")).hasMessageContaining("missing"); + assertThatThrownBy(() -> invalid.newCommit("test")).hasMessageContaining("missing"); + assertThatThrownBy( + () -> + catalog.createTable( + identifier("invalid"), + builder.option("record-level.time-field", "missing") + .build(), + false)) + .hasMessageContaining("missing"); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testCachesBeforeAndAfterStoreInitialization(boolean primaryKey) throws Exception { + Schema.Builder builder = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .option(CoreOptions.MANIFEST_SIDECAR_ENABLED.key(), "true"); + if (primaryKey) { + builder.primaryKey("id").option("bucket", "1"); + } + catalog.createTable(identifier(), builder.build(), false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + for (int i = 0; i < 2; i++) { + SegmentsCache manifests = + new SegmentsCache<>(1024, new MemorySize(1024 * 1024), 1024 * 1024); + SegmentsCache sidecars = + new SegmentsCache<>(1024, new MemorySize(1024 * 1024), 1024 * 1024); + Cache snapshots = Caffeine.newBuilder().build(); + table.setManifestCache(manifests); + table.setManifestSidecarCache(sidecars); + table.setSnapshotCache(snapshots); + write(table, GenericRow.of(i)); + assertThat(read(table)).hasSize(i + 1); + assertThat(manifests.estimatedSize()).isPositive(); + FileStore store = table.store(); + Snapshot snapshot = table.latestSnapshot().get(); + snapshots.invalidateAll(); + store.snapshotManager().snapshot(snapshot.id()); + assertThat(snapshots.getIfPresent(store.snapshotManager().snapshotPath(snapshot.id()))) + .isNotNull(); + ManifestFileMeta manifest = + store.manifestListFactory().create().readDataManifests(snapshot).get(0); + assertThat( + store.manifestFileFactory() + .create() + .selectBlocks( + manifest, null, PartitionPredicate.ALWAYS_TRUE, null)) + .isNotNull(); + assertThat( + sidecars.getIfPresents( + new Path( + table.location(), + "manifest/" + ManifestSidecar.fileName(manifest)))) + .isNotNull(); + FileStoreTable copied = table.copy(Collections.singletonMap("read.batch-size", "32")); + assertThat(copied.getManifestCache()).isSameAs(manifests); + assertThat(copied.getManifestSidecarCache()).isSameAs(sidecars); + snapshots.invalidateAll(); + copied.store().snapshotManager().snapshot(snapshot.id()); + assertThat(snapshots.getIfPresent(copied.snapshotManager().snapshotPath(snapshot.id()))) + .isNotNull(); + } + } + + @ParameterizedTest + @CsvSource({ + "false,-1,false,BUCKET_UNAWARE", + "false,2,false,HASH_FIXED", + "true,2,false,HASH_FIXED", + "true,-1,false,HASH_DYNAMIC", + "true,-1,true,KEY_DYNAMIC", + "true,-2,false,POSTPONE_MODE" + }) + public void testBucketMode( + boolean primaryKey, int buckets, boolean crossPartition, BucketMode expected) + throws Exception { + Schema.Builder builder = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("pt", DataTypes.INT()) + .option("bucket", String.valueOf(buckets)); + if (primaryKey) { + builder.primaryKey("id"); + } else if (buckets > 0) { + builder.option("bucket-key", "id"); + } + if (crossPartition) { + builder.partitionKeys("pt"); + } + catalog.createTable(identifier(), builder.build(), false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + assertThat(table.bucketMode()).isEqualTo(expected); + assertThat(table.store().bucketMode()).isEqualTo(expected); + } + + @Test + public void testCopyWithoutTimeTravelDefersValidation() throws Exception { + catalog.createTable( + identifier(), Schema.newBuilder().column("id", DataTypes.INT()).build(), false); + FileStoreTable original = (FileStoreTable) catalog.getTable(identifier()); + FileSystemSchemaManager schemas = + new FileSystemSchemaManager(original.fileIO(), original.location()); + Map options = new HashMap<>(original.options()); + options.put(CoreOptions.RECORD_LEVEL_TIME_FIELD.key(), "missing"); + original.fileIO() + .overwriteFileUtf8( + schemas.toSchemaPath(0), original.schema().copy(options).toString()); + catalog.invalidateTable(identifier()); + + FileStoreTable planning = (FileStoreTable) catalog.getTable(identifier()); + FileStoreTable copy = + planning.copyWithoutTimeTravel(Collections.singletonMap("read.batch-size", "32")); + assertThat(copy.options()).containsEntry("read.batch-size", "32"); + assertThatThrownBy(copy::newScan).hasMessageContaining("missing"); + FileStoreTable repaired = + planning.copyWithoutTimeTravel( + Collections.singletonMap(CoreOptions.RECORD_LEVEL_TIME_FIELD.key(), null)); + assertThat(read(repaired)).isEmpty(); + } + + @Test + public void testStoreInitializationFailureDoesNotBlockMetadata() throws Exception { + catalog.createTable( + identifier(), + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("v", DataTypes.INT()) + .primaryKey("id") + .option("bucket", "1") + .option("merge-engine", "partial-update") + .build(), + false); + FileStoreTable original = (FileStoreTable) catalog.getTable(identifier()); + FileSystemSchemaManager schemas = + new FileSystemSchemaManager(original.fileIO(), original.location()); + Map options = new HashMap<>(original.options()); + options.put("fields.v.aggregate-function", "max"); + original.fileIO() + .overwriteFileUtf8( + schemas.toSchemaPath(0), original.schema().copy(options).toString()); + catalog.invalidateTable(identifier()); + + FileStoreTable planning = (FileStoreTable) catalog.getTable(identifier()); + assertThat(planning.options()).containsEntry("fields.v.aggregate-function", "max"); + assertThat(planning.rowType().getFieldNames()).containsExactly("id", "v"); + assertThatThrownBy(planning::newScan).hasMessageContaining("Must use sequence group"); + assertThatThrownBy(() -> planning.newWrite("test")) + .hasMessageContaining("Must use sequence group"); + catalog.alterTable( + identifier(), SchemaChange.removeOption("fields.v.aggregate-function"), false); + FileStoreTable repaired = (FileStoreTable) catalog.getTable(identifier()); + write(repaired, GenericRow.of(1, 10)); + assertThat(read(repaired)).extracting(row -> row.getInt(1)).containsExactly(10); + } + + @ParameterizedTest + @ValueSource(strings = {"scan.fallback-branch", "scan.primary-branch"}) + public void testFallbackBranchLoading(String option) throws Exception { + catalog.createTable( + identifier(), + Schema.newBuilder() + .column("pt", DataTypes.INT()) + .column("v", DataTypes.INT()) + .partitionKeys("pt") + .build(), + false); + FileStoreTable main = (FileStoreTable) catalog.getTable(identifier()); + main.createBranch("backup"); + write(main, GenericRow.of(1, 10)); + write(main.switchToBranch("backup"), GenericRow.of(1, 20), GenericRow.of(2, 30)); + catalog.alterTable(identifier(), SchemaChange.setOption(option, "backup"), false); + + FileStoreTable planning = (FileStoreTable) catalog.getTable(identifier()); + assertThat(planning).isInstanceOf(FallbackReadFileStoreTable.class); + assertThat(read(planning)) + .extracting(row -> row.getInt(1)) + .containsExactlyInAnyOrder(option.equals("scan.fallback-branch") ? 10 : 20, 30); + FileStoreTable invalid = + planning.copy( + Collections.singletonMap( + CoreOptions.RECORD_LEVEL_TIME_FIELD.key(), "missing")); + assertThat(invalid.rowType()).isEqualTo(planning.rowType()); + assertThatThrownBy(invalid::newRead).hasMessageContaining("missing"); + assertThatThrownBy(invalid::newScan).hasMessageContaining("missing"); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testCapturedSchemaAndSerialization(boolean primaryKey) throws Exception { + Schema.Builder builder = Schema.newBuilder().column("id", DataTypes.INT()); + if (primaryKey) { + builder.primaryKey("id").option("bucket", "1"); + } + catalog.createTable(identifier(), builder.build(), false); + FileStoreTable planning = (FileStoreTable) catalog.getTable(identifier()); + FileStoreTable runtime = (FileStoreTable) catalog.getTable(identifier()); + assertThat(planning.bucketMode()).isEqualTo(runtime.bucketMode()); + write(runtime, GenericRow.of(1)); + catalog.alterTable(identifier(), SchemaChange.addColumn("v", DataTypes.INT()), false); + + FileStoreTable restored = + InstantiationUtil.deserializeObject( + InstantiationUtil.serializeObject(planning), getClass().getClassLoader()); + assertThat(restored.schema().id()).isEqualTo(0L); + assertThat(restored.rowType().getFieldNames()).containsExactly("id"); + assertThat(read(restored)).extracting(row -> row.getInt(0)).containsExactly(1); + assertThat(restored.schema()).isEqualTo(planning.schema()); + } + + @Test + public void testTimeTravelSchemaAndDynamicOptions() throws Exception { + catalog.createTable( + identifier(), Schema.newBuilder().column("id", DataTypes.INT()).build(), false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + write(table, GenericRow.of(1)); + table.createTag("before_evolution", 1); + catalog.alterTable(identifier(), SchemaChange.addColumn("v", DataTypes.INT()), false); + table = (FileStoreTable) catalog.getTable(identifier()); + write(table, GenericRow.of(2, 20)); + + FileStoreTable planning = (FileStoreTable) catalog.getTable(identifier()); + FileStoreTable historical = + planning.copy(Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1")); + historical = + historical.copy(Collections.singletonMap(CoreOptions.READ_BATCH_SIZE.key(), "32")); + assertThat(historical.rowType().getFieldNames()).containsExactly("id"); + assertThat(historical.options()).containsEntry(CoreOptions.READ_BATCH_SIZE.key(), "32"); + assertThat(read(historical)).extracting(row -> row.getInt(0)).containsExactly(1); + assertThat(historical.store().options().toMap()).isEqualTo(historical.options()); + + FileStoreTable tagged = + planning.copy( + Collections.singletonMap( + CoreOptions.SCAN_TAG_NAME.key(), "before_evolution")); + assertThat(tagged.schema()).isEqualTo(historical.schema().copy(tagged.options())); + assertThat(read(tagged)).extracting(row -> row.getInt(0)).containsExactly(1); + FileStoreTable noTimeTravel = + planning.copyWithoutTimeTravel( + Collections.singletonMap( + CoreOptions.SCAN_TAG_NAME.key(), "before_evolution")); + assertThat(noTimeTravel.rowType().getFieldNames()).containsExactly("id", "v"); + assertThat(read(noTimeTravel)).extracting(row -> row.getInt(0)).containsExactly(1); + + assertThatThrownBy(() -> planning.copy(Collections.singletonMap("bucket", "2"))) + .isInstanceOf(UnsupportedOperationException.class); + assertThatThrownBy( + () -> + ((FileStoreTable) catalog.getTable(identifier())) + .copy( + Collections.singletonMap( + CoreOptions.RECORD_LEVEL_TIME_FIELD.key(), + "missing")) + .newScan()) + .isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/TimeTravelSchemaEvolutionTest.java b/paimon-core/src/test/java/org/apache/paimon/table/TimeTravelSchemaEvolutionTest.java index fd567b6415e3..1a7d762ec351 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/TimeTravelSchemaEvolutionTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/TimeTravelSchemaEvolutionTest.java @@ -194,7 +194,7 @@ public void testExplicitVectorOptionsArePreserved() throws Exception { Map queryOptions = new HashMap<>(); queryOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1"); queryOptions.put(CoreOptions.VECTOR_FIELD.key(), "missing"); - assertThatThrownBy(() -> table.copy(queryOptions)) + assertThatThrownBy(() -> table.copy(queryOptions).newScan()) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Some of the columns specified as vector-field are unknown."); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java index 288f72518e9f..8570630a632b 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java @@ -374,7 +374,8 @@ public void testTimeTravelRead() throws Exception { time3))) .hasRootCauseInstanceOf(IllegalArgumentException.class) .hasRootCauseMessage( - "[scan.snapshot-id] must be null when you set [scan.timestamp-millis,scan.timestamp]"); + "Only one of the following parameters may be set : " + + "[scan.snapshot-id, scan.tag-name, scan.watermark, scan.timestamp, scan.timestamp-millis]"); assertThatThrownBy( () -> diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java index 5980d627d1b6..40b2c8b35391 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java @@ -320,10 +320,14 @@ public void testContinuousFromTimestamp() throws Exception { @Test public void testLackStartupTimestamp() { assertThatThrownBy( - () -> - streamSqlIter( - "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='from-timestamp') */")) - .hasCauseInstanceOf(IllegalArgumentException.class) + () -> { + try (CloseableIterator invalidIterator = + streamSqlIter( + "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='from-timestamp') */")) { + invalidIterator.hasNext(); + } + }) + .hasRootCauseInstanceOf(IllegalArgumentException.class) .hasRootCauseMessage( "must set only one key in [scan.timestamp-millis,scan.timestamp] when you use from-timestamp for scan.mode"); } @@ -344,11 +348,15 @@ public void testConfigureStartupTimestamp() throws Exception { // Configure 'scan.timestamp-millis' with 'scan.mode=latest'. assertThatThrownBy( - () -> - streamSqlIter( - "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='latest', 'scan.timestamp-millis'='%s') */", - 0)) - .hasCauseInstanceOf(IllegalArgumentException.class) + () -> { + try (CloseableIterator invalidIterator = + streamSqlIter( + "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='latest', 'scan.timestamp-millis'='%s') */", + 0)) { + invalidIterator.hasNext(); + } + }) + .hasRootCauseInstanceOf(IllegalArgumentException.class) .hasRootCauseMessage( "scan.timestamp-millis must be null when you use latest for scan.mode"); } @@ -393,11 +401,15 @@ public void testConfigureStartupSnapshot() throws Exception { // Configure 'scan.snapshot-id' with 'scan.mode=latest'. assertThatThrownBy( - () -> - streamSqlIter( - "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='latest', 'scan.snapshot-id'='%s') */", - 0)) - .hasCauseInstanceOf(IllegalArgumentException.class) + () -> { + try (CloseableIterator invalidIterator = + streamSqlIter( + "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='latest', 'scan.snapshot-id'='%s') */", + 0)) { + invalidIterator.hasNext(); + } + }) + .hasRootCauseInstanceOf(IllegalArgumentException.class) .hasRootCauseMessage( "scan.snapshot-id must be null when you use latest for scan.mode"); } diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java index bb8acd72bb38..a99f1d0dee85 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java @@ -26,6 +26,9 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.operation.Lock; import org.apache.paimon.options.Options; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.utils.BlockingIterator; import org.apache.flink.table.api.TableEnvironment; @@ -33,12 +36,16 @@ import org.apache.flink.util.CloseableIterator; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.io.File; import java.io.IOException; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; @@ -74,6 +81,102 @@ public void testWriteRead() throws Exception { innerTestWriteRead(); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testMetadataOperationsWithInvalidOptions(boolean cacheEnabled) throws Exception { + tEnv = tableEnvironmentBuilder().batchMode().parallelism(1).build(); + tEnv.executeSql( + String.format( + "CREATE CATALOG metadata WITH ('type'='paimon', 'warehouse'='%s', 'cache-enabled'='%s')", + path, cacheEnabled)); + tEnv.useCatalog("metadata"); + tEnv.executeSql( + "CREATE TABLE legacy (id INT, seq INT, v INT, PRIMARY KEY (id) NOT ENFORCED) " + + "WITH ('bucket'='1', 'merge-engine'='partial-update', " + + "'changelog-producer'='lookup', 'fields.seq.sequence-group'='v')"); + Catalog catalog = ((FlinkCatalog) tEnv.getCatalog("metadata").get()).catalog(); + Identifier identifier = Identifier.create(DB_NAME, "legacy"); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + FileSystemSchemaManager schemas = + new FileSystemSchemaManager(table.fileIO(), table.location()); + overwriteOption(table, schemas, "fields.seq.sequence-group", "id,v"); + catalog.invalidateTable(identifier); + + try (CloseableIterator rows = tEnv.executeSql("SHOW CREATE TABLE legacy").collect()) { + assertThat(rows.next().getField(0).toString()) + .contains("fields.seq.sequence-group", "id,v"); + } + assertThatThrownBy( + () -> tEnv.executeSql("SELECT * FROM legacy").await(30, TimeUnit.SECONDS)) + .hasStackTraceContaining("primary key"); + assertThatThrownBy( + () -> + tEnv.executeSql("INSERT INTO legacy VALUES (1, 10, 100)") + .await(30, TimeUnit.SECONDS)) + .hasStackTraceContaining("primary key"); + assertThatThrownBy(() -> tEnv.executeSql("ALTER TABLE legacy SET ('bucket'='1')")) + .hasStackTraceContaining("primary key"); + assertThat(schemas.listAllIds()).containsExactly(0L); + + tEnv.executeSql("ALTER TABLE legacy SET ('fields.seq.sequence-group'='v')"); + assertThat(schemas.latest().get().id()).isEqualTo(1L); + tEnv.executeSql("INSERT INTO legacy VALUES (1, 10, 100)").await(30, TimeUnit.SECONDS); + try (CloseableIterator rows = tEnv.executeSql("SELECT * FROM legacy").collect()) { + assertThat(rows.next()).isEqualTo(Row.of(1, 10, 100)); + assertThat(rows.hasNext()).isFalse(); + } + + overwriteOption(table, schemas, "fields.seq.sequence-group", null); + overwriteOption(table, schemas, "fields.v.aggregate-function", "max"); + catalog.invalidateTable(identifier); + try (CloseableIterator rows = tEnv.executeSql("SHOW CREATE TABLE legacy").collect()) { + assertThat(rows.next().getField(0).toString()).contains("fields.v.aggregate-function"); + } + assertThatThrownBy( + () -> tEnv.executeSql("SELECT * FROM legacy").await(30, TimeUnit.SECONDS)) + .hasStackTraceContaining("Must use sequence group"); + assertThatThrownBy( + () -> + tEnv.executeSql("INSERT INTO legacy VALUES (1, 10, 100)") + .await(30, TimeUnit.SECONDS)) + .hasStackTraceContaining("Must use sequence group"); + tEnv.executeSql("ALTER TABLE legacy RESET ('fields.v.aggregate-function')"); + tEnv.executeSql("ALTER TABLE legacy SET ('fields.seq.sequence-group'='v')"); + try (CloseableIterator rows = tEnv.executeSql("SELECT * FROM legacy").collect()) { + assertThat(rows.next()).isEqualTo(Row.of(1, 10, 100)); + assertThat(rows.hasNext()).isFalse(); + } + + overwriteOption(table, schemas, "full-compaction.delta-commits", "1"); + catalog.invalidateTable(identifier); + assertThatThrownBy(() -> ((FileStoreTable) catalog.getTable(identifier)).newScan()) + .isInstanceOf(RuntimeException.class); + tEnv.executeSql("ALTER TABLE legacy RESET ('full-compaction.delta-commits')"); + assertThat(catalog.getTable(identifier).options()) + .doesNotContainKey("full-compaction.delta-commits"); + + overwriteOption(table, schemas, "fields.seq.sequence-group", "id,v"); + catalog.invalidateTable(identifier); + tEnv.executeSql("DROP TABLE legacy"); + assertThat(catalog.listTables(DB_NAME)).doesNotContain("legacy"); + assertThat(table.fileIO().exists(table.location())).isFalse(); + } + + private void overwriteOption( + FileStoreTable table, FileSystemSchemaManager schemas, String key, String value) + throws IOException { + TableSchema schema = schemas.latest().get(); + Map options = new HashMap<>(schema.options()); + if (value == null) { + options.remove(key); + } else { + options.put(key, value); + } + table.fileIO() + .overwriteFileUtf8( + schemas.toSchemaPath(schema.id()), schema.copy(options).toString()); + } + @Test public void testRenameTable() throws Exception { tEnv.useCatalog("fs"); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java index 74d4b2c52057..659bc793ab8f 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java @@ -18,6 +18,7 @@ package org.apache.paimon.flink; +import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.BinaryString; import org.apache.paimon.partition.Partition; @@ -35,6 +36,8 @@ import org.apache.paimon.predicate.Transform; import org.apache.paimon.predicate.UpperTransform; import org.apache.paimon.rest.RESTToken; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.types.DataTypes; import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList; @@ -64,6 +67,39 @@ /** ITCase for REST catalog. */ class RESTCatalogITCase extends RESTCatalogITCaseBase { + @Test + void testMetadataOperationsWithInvalidOptions() throws Exception { + Identifier identifier = Identifier.create(DATABASE_NAME, TABLE_NAME); + org.apache.paimon.catalog.Catalog catalog = + ((FlinkCatalog) tEnv.getCatalog("PAIMON").get()).catalog(); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + TableSchema schema = table.schema(); + Map options = new HashMap<>(schema.options()); + options.put(CoreOptions.RECORD_LEVEL_TIME_FIELD.key(), "missing"); + restCatalogServer.setTableSchema(identifier, schema.copy(options)); + catalog.invalidateTable(identifier); + + String name = DATABASE_NAME + "." + TABLE_NAME; + assertThat(sql("SHOW CREATE TABLE " + name).toString()) + .contains("record-level.time-field", "missing"); + assertThatThrownBy(() -> batchSql("SELECT * FROM " + name)) + .hasStackTraceContaining("missing"); + assertThatThrownBy(() -> batchSql("INSERT INTO " + name + " VALUES ('1', 11)")) + .hasStackTraceContaining("missing"); + + sql("ALTER TABLE " + name + " RESET ('record-level.time-field')"); + batchSql("INSERT INTO " + name + " VALUES ('1', 11)"); + assertThat(batchSql("SELECT * FROM " + name)).containsExactly(Row.of("1", 11.0D)); + + TableSchema repaired = ((FileStoreTable) catalog.getTable(identifier)).schema(); + options = new HashMap<>(repaired.options()); + options.put(CoreOptions.RECORD_LEVEL_TIME_FIELD.key(), "missing"); + restCatalogServer.setTableSchema(identifier, repaired.copy(options)); + catalog.invalidateTable(identifier); + sql("DROP TABLE " + name); + assertThat(catalog.listTables(DATABASE_NAME)).doesNotContain(TABLE_NAME); + } + @Test void testCreateTable() { List result = sql(String.format("SHOW CREATE TABLE %s.%s", DATABASE_NAME, TABLE_NAME)); diff --git a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java index 9e22b024e483..22ae5a709e84 100644 --- a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java +++ b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java @@ -335,7 +335,7 @@ public void testAlterAndDropTableWithInvalidOptions(boolean external) throws Exc fileIO.overwriteFileUtf8( schemaManager.toSchemaPath(schema.id()), schema.copy(invalidOptions).toString()); - assertThatThrownBy(() -> catalog.getTable(identifier)) + assertThatThrownBy(() -> ((FileStoreTable) catalog.getTable(identifier)).newScan()) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("missing"); catalog.alterTable( @@ -350,7 +350,7 @@ public void testAlterAndDropTableWithInvalidOptions(boolean external) throws Exc fileIO.overwriteFileUtf8( schemaManager.toSchemaPath(repairedSchema.id()), repairedSchema.copy(invalidOptions).toString()); - assertThatThrownBy(() -> catalog.getTable(identifier)) + assertThatThrownBy(() -> ((FileStoreTable) catalog.getTable(identifier)).newScan()) .isInstanceOf(IllegalArgumentException.class); catalog.dropTable(identifier, false); diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSourceTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSourceTest.scala index d49b0dbe730a..2a5b84dc69f6 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSourceTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSourceTest.scala @@ -994,10 +994,10 @@ class PaimonSourceTest extends PaimonSparkTestBase with StreamTest { test("Paimon Source: with error options") { withTempDir { checkpointDir => - val TableSnapshotState(_, location, snapshotData, latestChanges, _) = + val TableSnapshotState(_, location, _, _, _) = prepareTableAndGetLocation(3, true) - assertThrows[IllegalArgumentException] { + withStartedQuery( spark.readStream .format("paimon") .option("scan.snapshot-id", 3) @@ -1009,7 +1009,19 @@ class PaimonSourceTest extends PaimonSparkTestBase with StreamTest { .queryName("mem_table") .outputMode("append") .trigger(Trigger.ProcessingTime(5, TimeUnit.SECONDS)) - .start() + .start()) { + query => + val error = intercept[StreamingQueryException] { + query.processAllAvailable() + } + val cause = Iterator + .iterate(error: Throwable)(_.getCause) + .takeWhile(_ != null) + .collectFirst { case cause: IllegalArgumentException => cause } + .getOrElse(fail("Expected an IllegalArgumentException in the cause chain", error)) + assert(cause.getMessage.contains("scan.snapshot-id")) + assert(cause.getMessage.contains("scan.timestamp-millis")) + assert(cause.getMessage.contains("must be null when you set")) } } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogMetadataOperationsTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogMetadataOperationsTest.scala new file mode 100644 index 000000000000..db5ca3a43901 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogMetadataOperationsTest.scala @@ -0,0 +1,138 @@ +/* + * 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.paimon.spark.sql + +import org.apache.paimon.catalog.Identifier +import org.apache.paimon.schema.FileSystemSchemaManager +import org.apache.paimon.spark.{PaimonSparkTestBase, SparkCatalog} +import org.apache.paimon.table.FileStoreTable + +import org.apache.spark.sql.Row + +import java.util.{HashMap => JHashMap} + +class CatalogMetadataOperationsTest extends PaimonSparkTestBase { + + for (cacheEnabled <- Seq(false, true); v2Write <- Seq(false, true)) { + test(s"metadata operations with invalid options, cache=$cacheEnabled, v2Write=$v2Write") { + val catalogName = s"metadata_${cacheEnabled}_$v2Write" + withSparkSQLConf( + s"spark.sql.catalog.$catalogName" -> classOf[SparkCatalog].getName, + s"spark.sql.catalog.$catalogName.warehouse" -> s"${tempDBDir.getCanonicalPath}/$catalogName", + s"spark.sql.catalog.$catalogName.cache-enabled" -> cacheEnabled.toString, + "spark.paimon.write.use-v2-write" -> v2Write.toString + ) { + sql(s"USE $catalogName") + sql("CREATE DATABASE IF NOT EXISTS default") + sql("USE default") + try { + sql("""CREATE TABLE legacy (id INT, seq INT, v INT) USING paimon + |TBLPROPERTIES ('primary-key'='id', 'bucket'='1', + |'merge-engine'='partial-update', 'changelog-producer'='lookup', + |'fields.seq.sequence-group'='v')""".stripMargin) + sql("INSERT INTO legacy VALUES (1, 10, 100)") + val catalog = paimonCatalog + val identifier = Identifier.create("default", "legacy") + val table = catalog.getTable(identifier).asInstanceOf[FileStoreTable] + val schemas = new FileSystemSchemaManager(table.fileIO(), table.location()) + + def overwriteOption(key: String, value: String): Unit = { + val schema = schemas.latest().get() + val options = new JHashMap[String, String](schema.options()) + if (value == null) { + options.remove(key) + } else { + options.put(key, value) + } + table + .fileIO() + .overwriteFileUtf8(schemas.toSchemaPath(schema.id()), schema.copy(options).toString) + catalog.invalidateTable(identifier) + } + + overwriteOption("fields.seq.sequence-group", "id,v") + val ddl = sql("SHOW CREATE TABLE legacy").head().getString(0) + assert(ddl.contains("fields.seq.sequence-group")) + assert(ddl.contains("id,v")) + assert(catalog.getTable(identifier).options().containsKey("fields.seq.sequence-group")) + assertInvalid("primary key")(sql("SELECT * FROM legacy").collect()) + assertInvalid("primary key")(sql("INSERT INTO legacy VALUES (2, 20, 200)").collect()) + assertInvalid("primary key")(sql("ALTER TABLE legacy SET TBLPROPERTIES ('bucket'='1')")) + assert(schemas.latest().get().id() == 0L) + + sql("ALTER TABLE legacy SET TBLPROPERTIES ('fields.seq.sequence-group'='v')") + checkAnswer(sql("SELECT * FROM legacy"), Seq(Row(1, 10, 100))) + sql("INSERT INTO legacy VALUES (2, 20, 200)") + checkAnswer(sql("SELECT * FROM legacy"), Seq(Row(1, 10, 100), Row(2, 20, 200))) + + overwriteOption("fields.seq.sequence-group", null) + overwriteOption("fields.v.aggregate-function", "max") + assert( + sql("SHOW CREATE TABLE legacy") + .head() + .getString(0) + .contains("fields.v.aggregate-function")) + assertInvalid("Must use sequence group")(sql("SELECT * FROM legacy").collect()) + assertInvalid("Must use sequence group")( + sql("INSERT INTO legacy VALUES (3, 30, 300)").collect()) + sql("ALTER TABLE legacy UNSET TBLPROPERTIES ('fields.v.aggregate-function')") + sql("ALTER TABLE legacy SET TBLPROPERTIES ('fields.seq.sequence-group'='v')") + checkAnswer(sql("SELECT * FROM legacy"), Seq(Row(1, 10, 100), Row(2, 20, 200))) + + overwriteOption("full-compaction.delta-commits", "1") + intercept[UnsupportedOperationException]( + catalog.getTable(identifier).asInstanceOf[FileStoreTable].newScan()) + sql("ALTER TABLE legacy UNSET TBLPROPERTIES ('full-compaction.delta-commits')") + assert( + !catalog.getTable(identifier).options().containsKey("full-compaction.delta-commits")) + + overwriteOption("fields.seq.sequence-group", "id,v") + sql("DROP TABLE legacy") + assert(!catalog.listTables("default").contains("legacy")) + assert(!table.fileIO().exists(table.location())) + + sql("CREATE TABLE legacy (id INT) USING paimon") + overwriteOption("record-level.time-field", "missing") + assert(sql("SHOW CREATE TABLE legacy").head().getString(0).contains("missing")) + assertInvalid("missing")(sql("SELECT * FROM legacy").collect()) + assertInvalid("missing")(sql("INSERT INTO legacy VALUES (1)").collect()) + sql("ALTER TABLE legacy UNSET TBLPROPERTIES ('record-level.time-field')") + sql("INSERT INTO legacy VALUES (1)") + checkAnswer(sql("SELECT * FROM legacy"), Seq(Row(1))) + sql("DELETE FROM legacy WHERE id = 1") + assert(sql("SELECT * FROM legacy").collect().isEmpty) + } finally { + sql("DROP TABLE IF EXISTS legacy") + sql("USE paimon") + sql(s"USE $dbName0") + } + } + } + } + + private def assertInvalid(message: String)(action: => Unit): Unit = { + val error = intercept[Exception](action) + assert( + Iterator + .iterate(error: Throwable)(_.getCause) + .takeWhile(_ != null) + .exists(e => Option(e.getMessage).exists(_.contains(message))), + error.toString) + } +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HiveCatalogMetadataOperationsTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HiveCatalogMetadataOperationsTest.scala new file mode 100644 index 000000000000..15899933d818 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HiveCatalogMetadataOperationsTest.scala @@ -0,0 +1,80 @@ +/* + * 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.paimon.spark.sql + +import org.apache.paimon.catalog.Identifier +import org.apache.paimon.schema.FileSystemSchemaManager +import org.apache.paimon.spark.PaimonHiveTestBase +import org.apache.paimon.table.FileStoreTable + +import org.apache.spark.sql.Row + +import java.util.{HashMap => JHashMap} + +class HiveCatalogMetadataOperationsTest extends PaimonHiveTestBase { + + Seq(sparkCatalogName, paimonHiveCatalogName).foreach { + catalogName => + test(s"metadata operations with invalid options through $catalogName") { + sql(s"USE $catalogName") + sql(s"USE $hiveDbName") + withTable("legacy_metadata") { + sql("CREATE TABLE legacy_metadata (id INT) USING paimon") + sql("INSERT INTO legacy_metadata VALUES (1)") + val catalog = paimonCatalog + val identifier = Identifier.create(hiveDbName, "legacy_metadata") + val table = catalog.getTable(identifier).asInstanceOf[FileStoreTable] + val schemas = new FileSystemSchemaManager(table.fileIO(), table.location()) + + def invalidateOptions(): Unit = { + val schema = schemas.latest().get() + val options = new JHashMap[String, String](schema.options()) + options.put("record-level.time-field", "missing") + table + .fileIO() + .overwriteFileUtf8(schemas.toSchemaPath(schema.id()), schema.copy(options).toString) + catalog.invalidateTable(identifier) + } + + invalidateOptions() + assert(sql("SHOW CREATE TABLE legacy_metadata").head().getString(0).contains("missing")) + assertInvalid(sql("SELECT * FROM legacy_metadata").collect()) + assertInvalid(sql("INSERT INTO legacy_metadata VALUES (2)").collect()) + sql("ALTER TABLE legacy_metadata UNSET TBLPROPERTIES ('record-level.time-field')") + checkAnswer(sql("SELECT * FROM legacy_metadata"), Seq(Row(1))) + sql("INSERT INTO legacy_metadata VALUES (2)") + checkAnswer(sql("SELECT * FROM legacy_metadata"), Seq(Row(1), Row(2))) + + invalidateOptions() + sql("DROP TABLE legacy_metadata") + assert(!catalog.listTables(hiveDbName).contains("legacy_metadata")) + } + } + } + + private def assertInvalid(action: => Unit): Unit = { + val error = intercept[Exception](action) + assert( + Iterator + .iterate(error: Throwable)(_.getCause) + .takeWhile(_ != null) + .exists(e => Option(e.getMessage).exists(_.contains("missing"))), + error.toString) + } +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RefreshTableTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RefreshTableTest.scala index ce0cd87a9edb..e759d2e8e1ef 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RefreshTableTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RefreshTableTest.scala @@ -19,18 +19,35 @@ package org.apache.paimon.spark.sql import org.apache.paimon.catalog.{CachingCatalog, Identifier} +import org.apache.paimon.schema.SchemaChange import org.apache.paimon.spark.PaimonSparkTestBase +import org.apache.paimon.types.DataTypes + +import org.apache.spark.sql.Row class RefreshTableTest extends PaimonSparkTestBase { - test("Refresh Table: clear cache") { + test("Refresh Table: clear cache and reload schema") { withTable("t") { - sql(s"CREATE TABLE t (f0 INT, f1 INT)") - sql(s"INSERT INTO t VALUES (1, 2)") - val tableCache = paimonCatalog.asInstanceOf[CachingCatalog].tableCache().asMap() - assert(tableCache.containsKey(Identifier.create(dbName0, "t"))) + sql("CREATE TABLE t (f0 INT, f1 INT)") + sql("INSERT INTO t VALUES (1, 2)") + val catalog = paimonCatalog.asInstanceOf[CachingCatalog] + val identifier = Identifier.create(dbName0, "t") + val cachedTable = catalog.getTable(identifier) + val tableCache = catalog.tableCache().asMap() + assert(tableCache.containsKey(identifier)) + + // Simulate an external schema change without invalidating this catalog's cache. + catalog.wrapped().alterTable(identifier, SchemaChange.addColumn("f2", DataTypes.INT()), false) + assert(sql("SELECT * FROM t").schema.fieldNames.toSeq == Seq("f0", "f1")) + assert(catalog.getTable(identifier).eq(cachedTable)) + sql("REFRESH TABLE t") - assert(!tableCache.containsKey(Identifier.create(dbName0, "t"))) + assert(!tableCache.containsKey(identifier)) + assert(!catalog.getTable(identifier).eq(cachedTable)) + assert(sql("SELECT * FROM t").schema.fieldNames.toSeq == Seq("f0", "f1", "f2")) + checkAnswer(sql("SELECT * FROM t"), Seq(Row(1, 2, null))) + assert(catalog.getTable(identifier).rowType().getFieldNames.contains("f2")) } } }