[core] Defer file-store validation until runtime initialization

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

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;
+        }
+    }
 }

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

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

paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java

@@ -156,6 +156,10 @@ void alterDatabase(String name, List<PropertyChange> changes, boolean ignoreIfNo
      *
      * <p>System tables can be got by '$' splitter.
      *
+     * <p>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

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<String> 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<Path> manifestCache;
     @Nullable protected transient SegmentsCache<Path> manifestSidecarCache;
     @Nullable protected transient Cache<Path, Snapshot> snapshotCache;
@@ -134,9 +141,11 @@ public String currentBranch() {
     }
 
     @Override
-    public void setManifestCache(SegmentsCache<Path> manifestCache) {
+    public synchronized void setManifestCache(SegmentsCache<Path> manifestCache) {
         this.manifestCache = manifestCache;
-        store().setManifestCache(manifestCache);
+        if (lazyStore != null) {
+            lazyStore.setManifestCache(manifestCache);
+        }
     }
 
     @Nullable
@@ -146,9 +155,11 @@ public SegmentsCache<Path> getManifestCache() {
     }
 
     @Override
-    public void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache) {
+    public synchronized void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache) {
         this.manifestSidecarCache = manifestSidecarCache;
-        store().setManifestSidecarCache(manifestSidecarCache);
+        if (lazyStore != null) {
+            lazyStore.setManifestSidecarCache(manifestSidecarCache);
+        }
     }
 
     @Nullable
@@ -158,9 +169,11 @@ public SegmentsCache<Path> getManifestSidecarCache() {
     }
 
     @Override
-    public void setSnapshotCache(Cache<Path, Snapshot> cache) {
+    public synchronized void setSnapshotCache(Cache<Path, Snapshot> cache) {
         this.snapshotCache = cache;
-        store().setSnapshotCache(cache);
+        if (lazyStore != null) {
+            lazyStore.setSnapshotCache(cache);
+        }
     }
 
     @Override
@@ -333,6 +346,62 @@ public StreamDataTableScan newStreamScan() {
 
     protected abstract BiConsumer<FileStoreScan, Predicate> 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<String> 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<String, String> 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<String, String> 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

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<FileStoreScan, Predicate> nonPartitionFilterConsumer() {
     }
 
     @Override
-    public InnerTableRead newRead() {
+    protected InnerTableRead createRead() {
         List<Function<SplitReadConfig, SplitReadProvider>> 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(

paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java

@@ -116,13 +116,18 @@ default Optional<String> 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. */
     DataTableScan newScan(SnapshotReaderFactory snapshotReaderFactory);
 
     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<String, String> dynamicOptions);