@@ -311,6 +311,42 @@ String, convert inside `of()`, log a warning) explicitly.
 
 `fe-connector-adbc` is the reference implementation of both classes.
 
+## Snapshot Metadata and Partition Views
+
+The engine must pin a table read before binding columns or building scan ranges.
+Derived metadata caches must describe that same read, including the selected
+snapshot/ref, schema and partition-spec generation. Never obtain a newer table
+independently to build one part of an already pinned read.
+
+Connectors can supply `getPartitionViewVersion` as a cheap generation key for
+an immutable partition view and its engine pruning index. Derive this key from
+the pinned metadata; do not enumerate or hash every partition to calculate it.
+A metadata refresh invalidates the corresponding engine cache, while in-flight
+readers retain their existing view. Query-specific conversions must not mutate
+objects shared by the connector cache.
+
+A connector declaring `supportsNativePartitionPlanning` can defer the engine
+partition view until a consumer actually requires internal partition semantics.
+For Iceberg, `enable_iceberg_internal_partition_pruning` defaults to `false`:
+ordinary scans use native manifest/file pruning. Setting it to `true` enables
+Doris partition pruning. MV planning and partition metadata rules still request
+the full view; deferred metadata must never be treated as an empty table.
+
+`getSqlCacheVersion` supplies a lightweight result-cache version for a pinned
+read without loading its partitions. The version must distinguish table
+identity and all metadata that can change the query result or bound schema.
+The engine records each table reference separately, so reading the same table
+through current, historical and branch/tag references does not collapse their
+cache dependencies. Cache validation resolves those original references again:
+a moved or removed ref, changed schema, expired snapshot or recreated table
+must not reuse an incompatible cached answer.
+
+Within a statement, reuse the frozen table and its resource lease through
+`ConnectorStatementScope`. Its read-only lookup must not load metadata on a
+miss. Relation existence and name listing can reuse this metadata through the
+SPI without source-specific checks in fe-core. This does not extend cache TTLs
+or authorize cross-user reuse of metadata.
+
 ## Testing and Verification
 
 - **Unit tests** live in each module; build/test recipes are in `AGENTS.md`.
 
@@ -53,6 +53,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -94,6 +95,13 @@ public interface IcebergCatalogOps {
     /** Lists the view names in {@code dbName}; empty when the catalog is not a (view-enabled) ViewCatalog. */
     List<String> listViewNames(String dbName);
 
+    /** Lists relations without requiring callers to subtract and then re-add view names. */
+    default List<String> listRelationNames(String dbName) {
+        LinkedHashSet<String> names = new LinkedHashSet<>(listTableNames(dbName));
+        names.addAll(listViewNames(dbName));
+        return new ArrayList<>(names);
+    }
+
     /** Returns {@code true} iff {@code dbName.tableName} exists. */
     boolean tableExists(String dbName, String tableName);
 
@@ -392,6 +400,15 @@ public List<String> listViewNames(String dbName) {
                     .collect(Collectors.toList());
         }
 
+        @Override
+        public List<String> listRelationNames(String dbName) {
+            LinkedHashSet<String> names = catalog.listTables(toNamespace(dbName)).stream()
+                    .map(TableIdentifier::name)
+                    .collect(Collectors.toCollection(LinkedHashSet::new));
+            names.addAll(listViewNames(dbName));
+            return new ArrayList<>(names);
+        }
+
         @Override
         public boolean tableExists(String dbName, String tableName) {
             return catalog.tableExists(toTableIdentifier(dbName, tableName));
 
@@ -25,6 +25,7 @@
 import org.apache.doris.connector.spi.ConnectorMetadata;
 import org.apache.doris.connector.spi.ConnectorPartitionInfo;
 import org.apache.doris.connector.spi.ConnectorSession;
+import org.apache.doris.connector.spi.ConnectorStatementScope;
 import org.apache.doris.connector.spi.ConnectorStorageContext;
 import org.apache.doris.connector.spi.ConnectorTableSchema;
 import org.apache.doris.connector.spi.ConnectorTableStatistics;
@@ -45,6 +46,7 @@
 import org.apache.doris.connector.spi.handle.WriteOperation;
 import org.apache.doris.connector.spi.mvcc.ConnectorMvccPartitionView;
 import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot;
+import org.apache.doris.connector.spi.mvcc.ConnectorSqlCacheVersion;
 import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec;
 import org.apache.doris.connector.spi.pushdown.ConnectorExpression;
 import org.apache.doris.thrift.THiveTable;
@@ -64,6 +66,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;
@@ -311,6 +314,17 @@ public List<String> listTableNames(ConnectorSession session, String dbName) {
         }
     }
 
+    @Override
+    public List<String> listRelationNames(ConnectorSession session, String dbName) {
+        try {
+            return executeAuthenticated(() -> catalogOps.listRelationNames(dbName));
+        } catch (RuntimeException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to list relation names, error message is: " + e.getMessage(), e);
+        }
+    }
+
     @Override
     public List<String> listViewNames(ConnectorSession session, String dbName) {
         // Mirror legacy IcebergMetadataOps.listViewNames: wrap in the auth context; a RuntimeException
@@ -424,26 +438,56 @@ private String loadTableComment(ConnectorSession session, String dbName, String
     @Override
     public Optional<ConnectorTableHandle> getTableHandle(
             ConnectorSession session, String dbName, String tableName) {
+        IcebergTableHandle handle = new IcebergTableHandle(dbName, tableName);
+        if (IcebergStatementScope.hasReadTable(session, dbName, tableName)
+                || (tableCache != null && tableCache.contains(TableIdentifier.of(dbName, tableName)))) {
+            return Optional.of(handle);
+        }
+        // Offline callers cannot retain a cache lease. Pure existence checks use tableExists instead.
+        if (session == null || session.getStatementScope() == ConnectorStatementScope.NONE) {
+            return tableExists(session, dbName, tableName) ? Optional.of(handle) : Optional.empty();
+        }
+        try {
+            // Loading proves existence and pins the same metadata used by schema/snapshot/scan planning.
+            // Borrow through the existing generation-aware cache so concurrent cold readers share one load.
+            executeAuthenticated(() -> resolveTableForRead(session, handle));
+            return Optional.of(handle);
+        } catch (Exception e) {
+            if (ExceptionUtils.getThrowableList(e).stream().anyMatch(NoSuchTableException.class::isInstance)) {
+                return Optional.empty();
+            }
+            throw IcebergExceptionUtils.wrapTableLoadFailure(handle, e,
+                    "Failed to load table, error message is:");
+        }
+    }
+
+    @Override
+    public boolean tableExists(
+            ConnectorSession session, String dbName, String tableName) {
+        // Reuse the statement's frozen read before consulting the shared cache: REFRESH must not
+        // make an in-flight reader re-check a different table generation. A shared cache is absent
+        // for credential-dependent catalogs; their statement-local proof never crosses identities.
+        if (IcebergStatementScope.hasReadTable(session, dbName, tableName)
+                || (tableCache != null && tableCache.contains(TableIdentifier.of(dbName, tableName)))) {
+            return true;
+        }
         // Mirror legacy IcebergMetadataOps.tableExist: wrap the remote existence check in the auth context
-        // (the handle build below is pure — no remote call).
+        // without loading and pinning a table merely to answer a DDL existence check.
         boolean exists;
         try {
             exists = executeAuthenticated(() -> catalogOps.tableExists(dbName, tableName));
         } catch (Exception e) {
             // Preserve Optional.empty even when an auth/catalog layer wraps NoSuchTableException.
             if (ExceptionUtils.getThrowableList(e).stream()
                     .anyMatch(NoSuchTableException.class::isInstance)) {
-                return Optional.empty();
+                return false;
             }
             // Existence checks in several catalogs load metadata internally, so normalize them at the shared
             // handle boundary just like explicit table loads.
             throw IcebergExceptionUtils.wrapTableLoadFailure(new IcebergTableHandle(dbName, tableName), e,
                     "Failed to check table exist, error message is:");
         }
-        if (!exists) {
-            return Optional.empty();
-        }
-        return Optional.of(new IcebergTableHandle(dbName, tableName));
+        return exists;
     }
 
     @Override
@@ -1998,6 +2042,32 @@ public boolean supportsCastPredicatePushdown(ConnectorSession session) {
         return true;
     }
 
+    @Override
+    public Optional<ConnectorSqlCacheVersion> getSqlCacheVersion(
+            ConnectorSession session, ConnectorTableHandle handle) {
+        IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+        try {
+            return executeAuthenticated(() -> {
+                Table table = resolveTableForRead(session, iceHandle);
+                String version = metadataVersion(table, iceHandle).orElseThrow(
+                        () -> new UnsupportedOperationException("Iceberg SQL cache requires published table metadata"));
+                long updateMillis = iceHandle.getSnapshotId() < 0 ? 0
+                        : table.snapshot(iceHandle.getSnapshotId()).timestampMillis();
+                return Optional.of(new ConnectorSqlCacheVersion(version, updateMillis));
+            });
+        } catch (Exception e) {
+            throw IcebergExceptionUtils.wrapTableLoadFailure(iceHandle, e,
+                    "Failed to get iceberg SQL cache version, error message is:");
+        }
+    }
+
+    @Override
+    public boolean supportsNativePartitionPlanning(ConnectorSession session, ConnectorTableHandle handle) {
+        return !((IcebergTableHandle) handle).isSystemTable()
+                && (session == null || !Boolean.parseBoolean(session.getSessionProperties()
+                        .getOrDefault("enable_iceberg_internal_partition_pruning", "false")));
+    }
+
     // ========== B-2: partition enumeration (MTMV RANGE view + SHOW PARTITIONS) ==========
 
     /**
@@ -2102,6 +2172,39 @@ public List<String> listPartitionNames(ConnectorSession session, ConnectorTableH
         }
     }
 
+    private static Optional<String> metadataVersion(Table table, IcebergTableHandle handle) {
+        // Custom catalog tables and unpublished metadata cannot supply a stable metadata generation.
+        if (!(table instanceof BaseTable)) {
+            return Optional.empty();
+        }
+        TableMetadata metadata = ((BaseTable) table).operations().current();
+        String location = metadata.metadataFileLocation();
+        if (location == null) {
+            return Optional.empty();
+        }
+        return Optional.of(metadata.uuid() + ":" + handle.getSnapshotId() + ":"
+                + handle.getSchemaId() + ":" + location);
+    }
+
+    @Override
+    public Optional<String> getPartitionViewVersion(ConnectorSession session, ConnectorTableHandle handle) {
+        IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+        if (!iceHandle.hasSnapshotPin() || session == null
+                || session.getStatementScope() == ConnectorStatementScope.NONE) {
+            // Without a statement-frozen table, metadata could change between version lookup and listing.
+            return Optional.empty();
+        }
+        try {
+            return executeAuthenticated(() -> {
+                Table table = resolveTableForRead(session, iceHandle);
+                return metadataVersion(table, iceHandle);
+            });
+        } catch (Exception e) {
+            throw IcebergExceptionUtils.wrapTableLoadFailure(iceHandle, e,
+                    "Failed to resolve iceberg partition view version, error message is:");
+        }
+    }
+
     /**
      * The physical iceberg partitions of {@code handle} with per-partition value maps, so the generic
      * {@code PluginDrivenExternalTable.getNameToPartitionItems} can populate {@code selectedPartitionNum}
@@ -2170,7 +2273,7 @@ private List<ConnectorPartitionInfo> listPartitionsUncached(
     }
 
     private List<ConnectorPartitionInfo> listPartitions(Table table, IcebergTableHandle iceHandle) {
-        return IcebergPartitionUtils.listPartitions(table,
+        return IcebergPartitionUtils.listPartitions(table, iceHandle.getSnapshotId(),
                 TableIdentifier.of(iceHandle.getDbName(), iceHandle.getTableName()), partitionCache);
     }
 
@@ -2182,7 +2285,8 @@ private List<ConnectorPartitionInfo> listPartitions(Table table, IcebergTableHan
      * snapshot id (or {@code -1} for an empty table — iceberg DOES support MVCC, so it still pins, mirroring
      * paimon), and the LATEST schema id (NOT {@code currentSnapshot().schemaId()} — a schema-only change without
      * a new snapshot advances the schema while the snapshot's id lags; legacy reads {@code table.schema()
-     * .schemaId()}). T08 serves this through the per-catalog {@link IcebergLatestSnapshotCache}: within the TTL
+     * .schemaId()}). An already frozen statement read takes precedence over the shared pin. Otherwise T08
+     * serves this through the per-catalog {@link IcebergLatestSnapshotCache}: within the TTL
      * a HIT returns the cached pin without re-loading the table (saves the load I/O and keeps the snapshot
      * STABLE across queries — the legacy with-cache catalog). The cached value carries BOTH ids atomically so a
      * schema-only ALTER between two queries cannot skew snapshotId vs schemaId. A disabled cache
@@ -2198,11 +2302,21 @@ public Optional<ConnectorMvccSnapshot> beginQuerySnapshot(
         // per-user loadTable authorization) reads live per-user every call, so authorization runs every time (no
         // stale-authz window); mirrors resolveTableForRead's tableCache null-fallback. A disabled cache (the table
         // entry turned off) is a non-null cache that reads live internally.
-        IcebergLatestSnapshotCache.CachedSnapshot pin = latestSnapshotCache != null
-                ? latestSnapshotCache.getOrLoad(id, () -> loadLatestSnapshotPin(session, iceHandle))
-                : loadLatestSnapshotPin(session, iceHandle);
+        IcebergLatestSnapshotCache.CachedSnapshot pin;
+        if (IcebergStatementScope.hasReadTable(session, iceHandle.getDbName(), iceHandle.getTableName())) {
+            // Handle/schema resolution may have frozen an older generation before another query's REFRESH.
+            // A newer shared pin must not point outside that frozen metadata, nor may this reader publish its
+            // old pin over the refreshed cache entry.
+            pin = latestSnapshotPin(loadTable(session, iceHandle));
+        } else {
+            pin = latestSnapshotCache != null
+                    ? latestSnapshotCache.getOrLoad(id, () -> loadLatestSnapshotPin(session, iceHandle))
+                    : loadLatestSnapshotPin(session, iceHandle);
+        }
+        // SQL result-cache identity and column binding must describe the same read generation. The engine
+        // must not combine this snapshot/schema pin with an independently expiring name-keyed schema cache.
         ConnectorMvccSnapshot.Builder snapshot = ConnectorMvccSnapshot.builder()
-                .snapshotId(pin.snapshotId).schemaId(pin.schemaId);
+                .snapshotId(pin.snapshotId).schemaId(pin.schemaId).retainSchema(true);
         if (pin.snapshotId < 0) {
             snapshot.property(EMPTY_PARTITION_STYLE_PROPERTY, pin.emptyPartitionStyle.name());
         }
 
@@ -777,10 +777,15 @@ static List<ConnectorPartitionInfo> listPartitions(Table table) {
     /** Cache-aware overload (PERF-02): see {@link #buildMvccPartitionView(Table, long, TableIdentifier,
      * IcebergPartitionCache)}. Keyed by the table's CURRENT snapshot id. */
     static List<ConnectorPartitionInfo> listPartitions(Table table, TableIdentifier id, IcebergPartitionCache cache) {
+        return listPartitions(table, -1L, id, cache);
+    }
+
+    static List<ConnectorPartitionInfo> listPartitions(Table table, long pinnedSnapshotId,
+            TableIdentifier id, IcebergPartitionCache cache) {
         if (table.spec().isUnpartitioned()) {
             return Collections.emptyList();
         }
-        Snapshot current = table.currentSnapshot();
+        Snapshot current = pinnedSnapshotId >= 0 ? table.snapshot(pinnedSnapshotId) : table.currentSnapshot();
         if (current == null) {
             return Collections.emptyList();
         }
 
@@ -74,6 +74,11 @@ final class IcebergStatementScope {
 
     private IcebergStatementScope() {}
 
+    /** A frozen read or its live lease proves existence even if REFRESH retires the catalog cache. */
+    static boolean hasReadTable(ConnectorSession session, String dbName, String tableName) {
+        return ConnectorStatementScopes.getIfPresent(session, TABLE_NAMESPACE, dbName, tableName) != null;
+    }
+
     /**
      * Loads and freezes the iceberg {@link Table} for {@code db.tbl} once per statement. The frozen operations
      * prevent another statement's write refresh from changing the metadata generation after slots are bound.
 
@@ -113,6 +113,11 @@ boolean isEnabled() {
         return entry.isEnabled();
     }
 
+    /** Only a retained, unexpired table proves existence; never loads or caches a boolean answer. */
+    boolean contains(TableIdentifier identifier) {
+        return entry.getIfPresent(identifier) != null;
+    }
+
     /**
      * Returns the cached table for {@code identifier} if present and unexpired, else runs {@code loader} (the
      * live remote {@code loadTable}), caches and returns it. When caching is disabled ({@link #isEnabled()} is