@@ -43,6 +43,10 @@ under the License.
         <allow pkg="org.junit.jupiter"/>
         <disallow pkg="org.junit"/>
         <disallow pkg="lombok"/>
+        <disallow pkg="org.apache.iceberg"/>
+    </subpackage>
+    <subpackage name="planner">
+        <disallow pkg="org.apache.iceberg"/>
     </subpackage>
 </import-control>
 
@@ -17,13 +17,17 @@
 
 package org.apache.doris.connector.api;
 
+import org.apache.doris.connector.api.handle.WriteOperation;
+import org.apache.doris.connector.api.procedure.ConnectorProcedureOps;
 import org.apache.doris.connector.api.scan.ConnectorScanPlanProvider;
 import org.apache.doris.connector.api.write.ConnectorWritePlanProvider;
 
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.List;
+import java.util.Map;
 import java.util.OptionalLong;
 import java.util.Set;
 
@@ -51,11 +55,81 @@ default ConnectorWritePlanProvider getWritePlanProvider() {
         return null;
     }
 
+    /**
+     * The write operations the engine may perform on this connector — the single admission source. Reads the
+     * write provider's {@link ConnectorWritePlanProvider#supportedOperations()}; no provider ⇒ empty set ⇒ all
+     * writes rejected. The engine consults this instead of {@code getWritePlanProvider() != null}.
+     */
+    default Set<WriteOperation> supportedWriteOperations() {
+        ConnectorWritePlanProvider p = getWritePlanProvider();
+        return p == null ? EnumSet.noneOf(WriteOperation.class) : p.supportedOperations();
+    }
+
+    /** Null-safe view of {@link ConnectorWritePlanProvider#supportsWriteBranch()}. No provider ⇒ false. */
+    default boolean supportsWriteBranch() {
+        ConnectorWritePlanProvider p = getWritePlanProvider();
+        return p != null && p.supportsWriteBranch();
+    }
+
+    /** Null-safe view of {@link ConnectorWritePlanProvider#requiresParallelWrite()}. No provider ⇒ false. */
+    default boolean requiresParallelWrite() {
+        ConnectorWritePlanProvider p = getWritePlanProvider();
+        return p != null && p.requiresParallelWrite();
+    }
+
+    /** Null-safe view of {@link ConnectorWritePlanProvider#requiresFullSchemaWriteOrder()}. No provider ⇒ false. */
+    default boolean requiresFullSchemaWriteOrder() {
+        ConnectorWritePlanProvider p = getWritePlanProvider();
+        return p != null && p.requiresFullSchemaWriteOrder();
+    }
+
+    /** Null-safe view of {@link ConnectorWritePlanProvider#requiresPartitionLocalSort()}. No provider ⇒ false. */
+    default boolean requiresPartitionLocalSort() {
+        ConnectorWritePlanProvider p = getWritePlanProvider();
+        return p != null && p.requiresPartitionLocalSort();
+    }
+
+    /**
+     * Null-safe view of {@link ConnectorWritePlanProvider#requiresMaterializeStaticPartitionValues()}. No
+     * provider ⇒ false.
+     */
+    default boolean requiresMaterializeStaticPartitionValues() {
+        ConnectorWritePlanProvider p = getWritePlanProvider();
+        return p != null && p.requiresMaterializeStaticPartitionValues();
+    }
+
+    /**
+     * Returns the procedure ops for {@code ALTER TABLE EXECUTE} dispatch, or {@code null} if this
+     * connector exposes no table procedures. Procedure-side analogue of {@link #getWritePlanProvider()}.
+     */
+    default ConnectorProcedureOps getProcedureOps() {
+        return null;
+    }
+
     /** Returns the set of capabilities this connector supports. */
     default Set<ConnectorCapability> getCapabilities() {
         return Collections.emptySet();
     }
 
+    /**
+     * Storage-configuration defaults this connector derives from its own catalog properties, which the raw
+     * catalog map does not already supply. Design S8: storage-property derivation is owned by the connector —
+     * fe-core does not parse metastore properties. fe-core folds the returned map into the catalog's storage
+     * properties as DEFAULTS (an explicit user key always wins via {@code putIfAbsent}), and does so BEFORE
+     * both the fe-filesystem bind ({@code ConnectorContext.getStorageProperties()}) and the BE storage map
+     * ({@code getBackendStorageProperties()}), so the FE bind and the BE scan see the same derived storage.
+     *
+     * <p>The default is empty (no derivation), so every connector that does not need it is unaffected. The
+     * iceberg connector overrides this to bridge a hadoop-catalog {@code warehouse=hdfs://<ns>/path} into
+     * {@code fs.defaultFS=hdfs://<ns>}, which the shared HDFS detection never derives from {@code warehouse}.</p>
+     *
+     * @param rawCatalogProps the catalog's current persisted properties
+     * @return extra storage-property defaults; an empty map when there is nothing to derive
+     */
+    default Map<String, String> deriveStorageProperties(Map<String, String> rawCatalogProps) {
+        return Collections.emptyMap();
+    }
+
     /** Returns the table-level property descriptors. */
     default List<ConnectorPropertyMetadata<?>> getTableProperties() {
         return Collections.emptyList();
 
@@ -18,85 +18,109 @@
 package org.apache.doris.connector.api;
 
 /**
- * Enumerates the optional capabilities a connector may declare.
- * The planner and execution engine use these to decide which
- * pushdown and write paths are available.
+ * Enumerates optional, connector-declared capability switches consumed directly by
+ * static query-planning code (pushdown/DDL/view/statistics gating, etc.).
+ *
+ * <p>This is an escape-hatch layer for capability checks that don't warrant a dedicated
+ * provider abstraction. Write operations and sink traits (parallel write, partition-local
+ * sort, full-schema write order, static-partition materialization) are NOT declared here —
+ * they live on the connector's {@link org.apache.doris.connector.api.write.ConnectorWritePlanProvider}
+ * instead, surfaced via {@link Connector#getWritePlanProvider()}.</p>
  */
 public enum ConnectorCapability {
-    SUPPORTS_FILTER_PUSHDOWN,
-    SUPPORTS_PROJECTION_PUSHDOWN,
-    SUPPORTS_LIMIT_PUSHDOWN,
-    SUPPORTS_PARTITION_PRUNING,
-    SUPPORTS_INSERT,
-    SUPPORTS_DELETE,
-    SUPPORTS_UPDATE,
-    SUPPORTS_MERGE,
-    SUPPORTS_CREATE_TABLE,
     SUPPORTS_MVCC_SNAPSHOT,
-    SUPPORTS_METASTORE_EVENTS,
-    SUPPORTS_STATISTICS,
-    SUPPORTS_VENDED_CREDENTIALS,
-    SUPPORTS_ACID_TRANSACTIONS,
-    SUPPORTS_TIME_TRAVEL,
     /**
-     * Indicates the connector supports multiple concurrent writers (sink instances).
+     * Indicates the connector supports passthrough query via the {@code query()} TVF.
      *
-     * <p>Connectors that do NOT declare this capability will use GATHER distribution
-     * (single writer), which is the safe default for transactional sinks like JDBC
-     * where each writer commits independently.</p>
+     * <p>Connectors declaring this capability must implement
+     * {@link ConnectorTableOps#getColumnsFromQuery} to provide column metadata
+     * for arbitrary SQL queries passed through to the remote data source.</p>
+     */
+    SUPPORTS_PASSTHROUGH_QUERY,
+    /**
+     * Indicates the connector exposes per-partition statistics (record count, on-disk size,
+     * file count) via {@link ConnectorTableOps#listPartitions}.
      *
-     * <p>File-based connectors (Hive, Iceberg, etc.) that can safely handle
-     * parallel writers should declare this capability.</p>
+     * <p>{@code SHOW PARTITIONS} renders a rich multi-column result (Partition / PartitionKey /
+     * RecordCount / FileSizeInBytes / FileCount) for connectors declaring this capability, instead
+     * of the single partition-name column used by connectors that only implement
+     * {@code listPartitionNames}.</p>
      */
-    SUPPORTS_PARALLEL_WRITE,
+    SUPPORTS_PARTITION_STATS,
     /**
-     * Indicates the connector requires dynamic-partition writes to be hash-distributed by
-     * partition columns and locally sorted by them before reaching the sink.
+     * Indicates the connector's tables support background per-column auto-analyze (NDV / min / max /
+     * null-count collection) through the generic {@code ExternalAnalysisTask} FULL path.
      *
-     * <p>Streaming partition writers (e.g. the MaxCompute Storage API) close the previous
-     * partition writer as soon as a new partition value appears; un-grouped (unsorted)
-     * multi-partition rows therefore cause "writer has been closed" errors. The planner uses
-     * this capability to require a hash-by-partition distribution plus a mandatory local sort
-     * on the partition columns for dynamic-partition writes.</p>
+     * <p>The statistics auto-collector admits a plugin-driven table into the background auto-analyze
+     * framework only when its connector declares this (replacing the legacy {@code instanceof
+     * IcebergExternalTable} whitelist), and then forces {@code AnalysisMethod.FULL} — sample analyze is
+     * unimplemented for external SQL-driven tables ({@code ExternalAnalysisTask.doSample} throws).
+     * Row/passthrough connectors that cannot serve per-column statistics (e.g. JDBC, ES) must NOT
+     * declare it so they stay excluded.</p>
+     */
+    SUPPORTS_COLUMN_AUTO_ANALYZE,
+    /**
+     * Indicates the connector's file-scan tables support Top-N lazy materialization: the scan first
+     * reads only the ordering/filter columns to locate the Top-N row ids, then materializes the
+     * remaining columns for just those rows (via the synthesized {@code GLOBAL_ROWID_COL}).
      *
-     * <p>A connector declaring this is expected to also declare
-     * {@link #SUPPORTS_PARALLEL_WRITE} (hash distribution is inherently parallel) and
-     * {@link #SINK_REQUIRE_FULL_SCHEMA_ORDER}: the sink distribution locates partition columns by their
-     * <b>full-schema</b> position in the child output, which only holds when the bind layer projects the
-     * write to full-schema order (the projection gated by {@code SINK_REQUIRE_FULL_SCHEMA_ORDER}). A
-     * connector declaring this without {@code SINK_REQUIRE_FULL_SCHEMA_ORDER} would shuffle/sort by the
-     * wrong column whenever cols order diverges from the full schema.</p>
+     * <p>The nereids Top-N lazy-materialize probe enables the {@code LazyMaterializeTopN} post-processor
+     * for a plugin-driven table only when its connector declares this (replacing the legacy exact-class
+     * {@code SUPPORT_RELATION_TYPES} membership of {@code IcebergExternalTable}). Row/passthrough
+     * connectors (e.g. JDBC, ES) must NOT declare it.</p>
      */
-    SINK_REQUIRE_PARTITION_LOCAL_SORT,
+    SUPPORTS_TOPN_LAZY_MATERIALIZE,
     /**
-     * Indicates the connector's write path maps data columns <b>positionally</b> against the full
-     * table schema (e.g. MaxCompute's columnar Storage API / JNI writer), rather than by column name.
+     * Indicates the connector's table/database properties are user-facing and safe to render in
+     * {@code SHOW CREATE TABLE} / {@code SHOW CREATE DATABASE}.
      *
-     * <p>For such connectors the sink's output rows must be projected to <b>full table schema order</b>
-     * with any unmentioned columns filled (NULL / default) — exactly like the legacy MaxCompute bind
-     * path — so that a reordered or partial explicit column list does not land values in the wrong
-     * remote columns. Name-mapped connectors (e.g. JDBC, which builds an {@code INSERT INTO t (cols)}
-     * statement) must NOT declare this capability: their data stays in user/cols order to match the
-     * generated column list.</p>
+     * <p>The SHOW CREATE TABLE plugin-driven arm renders LOCATION + PROPERTIES (and, when the
+     * connector pre-renders them under the {@code show.*} reserved keys, the PARTITION BY / ORDER BY
+     * clauses) only for connectors declaring this (replacing the legacy paimon-only engine-name gate).
+     * Row/passthrough connectors whose {@code getTableProperties()} returns connection properties
+     * <b>including credentials</b> (e.g. JDBC, ES) must NOT declare it, or SHOW CREATE TABLE would leak
+     * the connection password — the security control the legacy engine-name gate provided.</p>
      */
-    SINK_REQUIRE_FULL_SCHEMA_ORDER,
+    SUPPORTS_SHOW_CREATE_DDL,
     /**
-     * Indicates the connector supports passthrough query via the {@code query()} TVF.
+     * Indicates the connector exposes views as queryable objects distinct from tables.
      *
-     * <p>Connectors declaring this capability must implement
-     * {@link ConnectorTableOps#getColumnsFromQuery} to provide column metadata
-     * for arbitrary SQL queries passed through to the remote data source.</p>
+     * <p>When a connector declares this, a plugin-driven table resolves its {@code isView()} from the
+     * connector ({@link ConnectorTableOps#viewExists}) instead of the {@code false} default, the catalog
+     * merges the connector's {@link ConnectorTableOps#listViewNames} back into {@code SHOW TABLES} (iceberg
+     * subtracts views from {@code listTableNames}), and the read/DML/SHOW CREATE arms treat the object as a
+     * view. Connectors with no view concept (e.g. JDBC, ES) must NOT declare it so every table stays
+     * {@code isView()==false} and no view round-trips are issued.</p>
      */
-    SUPPORTS_PASSTHROUGH_QUERY,
+    SUPPORTS_VIEW,
     /**
-     * Indicates the connector exposes per-partition statistics (record count, on-disk size,
-     * file count) via {@link ConnectorTableOps#listPartitions}.
+     * Indicates the connector's file-scan tables support nested-column pruning: a query that reads only some
+     * sub-fields of a STRUCT/ARRAY/MAP column reads just those leaves from the data file instead of the whole
+     * complex column (read-amplification avoidance).
      *
-     * <p>{@code SHOW PARTITIONS} renders a rich multi-column result (Partition / PartitionKey /
-     * RecordCount / FileSizeInBytes / FileCount) for connectors declaring this capability, instead
-     * of the single partition-name column used by connectors that only implement
-     * {@code listPartitionNames}. This is distinct from {@link #SUPPORTS_STATISTICS}, which is
-     * table-level statistics for the optimizer.</p>
+     * <p>The nereids nested-column-prune probe ({@code LogicalFileScan.supportPruneNestedColumn}) enables it
+     * for a plugin-driven table only when its connector declares this (replacing the legacy exact-class
+     * {@code IcebergExternalTable} arm). It is only correct when the connector also carries a stable per-field
+     * id down its column tree (top-level via {@link ConnectorColumn#withUniqueId} + nested via
+     * {@link ConnectorType#withChildrenFieldIds}), because the engine rewrites the nested access path from
+     * field <em>names</em> to those ids ({@code SlotTypeReplacer}) and the BE field-id scan path matches
+     * nested leaves by id — an un-translated (name / {@code -1}) leaf is skipped and returns NULL. Row/
+     * passthrough connectors (e.g. JDBC, ES) and connectors that do not carry nested field ids must NOT
+     * declare it.</p>
+     */
+    SUPPORTS_NESTED_COLUMN_PRUNE,
+    /**
+     * Indicates the connector's external metadata (schema / partitions / snapshot) can be pre-warmed
+     * asynchronously by the planner before it takes the internal read lock, rather than loaded lazily
+     * during binding.
+     *
+     * <p>{@code PluginDrivenExternalTable.supportsExternalMetadataPreload} returns true for a plugin-driven
+     * table only when its connector declares this (replacing the legacy engine-name {@code "jdbc"} gate), so
+     * {@code StatementContext.registerExternalTableForPreload} admits the table into the async pre-load pass
+     * (itself opt-in via the {@code enable_preload_external_metadata} session variable, default off). It is a
+     * pure planning/lock-latency optimization with no correctness effect: connectors whose metadata reads are
+     * cheap or not yet validated for concurrent pre-warming (e.g. ES) simply do not declare it and fall back
+     * to synchronous load at binding time.</p>
      */
-    SUPPORTS_PARTITION_STATS
+    SUPPORTS_METADATA_PRELOAD
 }