[perf](catalog) fe-connector-iceberg hot-path caching + fe-core per-statement metadata funnel, with session=user cache isolation

fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorSession.java

@@ -128,4 +128,18 @@ default void setCurrentTransaction(ConnectorTransaction txn) {
     default long allocateTransactionId() {
         throw new UnsupportedOperationException("transaction id allocation not supported");
     }
+
+    /**
+     * Returns the per-statement scope for this session — a memoization arena that lives exactly as long
+     * as the current SQL statement, letting a connector load a table (and derive per-statement state)
+     * once and share that single object across every read + write resolver in the statement.
+     *
+     * <p>The default is {@link ConnectorStatementScope#NONE} (no memoization = load every time), so
+     * every existing implementation is unaffected. The engine session implementation overrides it with a
+     * scope hung on the statement, captured at session construction (the request thread, where the
+     * statement context is reachable) so off-thread scan pumps that reuse the session still reach it.</p>
+     */
+    default ConnectorStatementScope getStatementScope() {
+        return ConnectorStatementScope.NONE;
+    }
 }

fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorStatementScope.java

@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.connector.api;
+
+import java.util.function.Supplier;
+
+/**
+ * A per-statement memoization arena, living exactly as long as one SQL statement (read + write).
+ *
+ * <p>It lets a connector load a table (and derive per-statement state) once and share that single
+ * object across every resolver in the statement — read metadata, scan planning, write shaping,
+ * begin-write. Reached from a connector via {@link ConnectorSession#getStatementScope()}.</p>
+ *
+ * <p>Neutral SPI: the engine owns the physical home (the scope is hung on the engine's per-statement
+ * context) and the connector stores its own connector-typed values under string keys as opaque
+ * {@code Object}s — fe-core never sees a connector type. Values are session-bound and reclaimed with
+ * the statement, never promoted into any cross-session / cross-identity structure.</p>
+ *
+ * <p>{@link #NONE} is the off-context default (no live statement: offline planning, tests, and any
+ * {@link ConnectorSession} that does not override {@link ConnectorSession#getStatementScope()}). It
+ * never memoizes, so every call runs the loader — byte-identical to loading every time.</p>
+ */
+public interface ConnectorStatementScope {
+
+    /**
+     * Returns the value cached under {@code key}, computing it with {@code loader} on first access and
+     * caching it for the rest of the statement. Within one statement the same key returns the same
+     * instance to every caller; under {@link #NONE} the loader runs on every call.
+     */
+    <T> T computeIfAbsent(String key, Supplier<T> loader);
+
+    /**
+     * Typed convenience over {@link #computeIfAbsent} for the ONE {@link ConnectorMetadata} a statement uses per
+     * {@code key}. The engine's metadata funnel builds {@code key} (as {@code "metadata:" + catalogId}, plus the
+     * owning connector's label for a heterogeneous gateway) and passes a {@code factory} that calls
+     * {@code Connector#getMetadata(session)}; every read / scan / DDL / MVCC resolver of the statement then shares
+     * the single memoized instance and the factory runs at most once per statement. Under {@link #NONE} the
+     * factory runs on every call (byte-identical to building metadata every time).
+     */
+    default ConnectorMetadata getOrCreateMetadata(String key, Supplier<ConnectorMetadata> factory) {
+        return computeIfAbsent(key, factory);
+    }
+
+    /**
+     * Deterministically closes, at statement end, every value this scope holds that is {@link AutoCloseable} (a
+     * memoized {@link ConnectorMetadata} is, via {@link java.io.Closeable}). Best-effort and log-and-continue: a
+     * failure closing one value does not abort the rest. MUST be idempotent (close-once) in implementations — the
+     * engine fires it from more than one locus (the query-finish callback, and a reused prepared statement's
+     * per-execution reset). The default is a no-op, so {@link #NONE} — which memoizes nothing — stays inert.
+     */
+    default void closeAll() {
+    }
+
+    /** The no-op scope: never caches; each call invokes the loader (offline / no-context / tests). */
+    ConnectorStatementScope NONE = new ConnectorStatementScope() {
+        @Override
+        public <T> T computeIfAbsent(String key, Supplier<T> loader) {
+            return loader.get();
+        }
+    };
+}

fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnector.java

@@ -169,16 +169,18 @@ private static String rootCauseMessage(Throwable t) {
      */
     HiveConnectorMetadata newMetadata(HmsClient client) {
         return new HiveConnectorMetadata(client, properties, context,
-                this::getOrCreateIcebergSibling, this::getOrCreateHudiSibling, this::resolveSiblingOwner,
+                this::getOrCreateIcebergSibling, this::getOrCreateHudiSibling, this::resolveSiblingOwnerLabeled,
                 fileListingCache);
     }
 
     /**
-     * Resolves which embedded sibling connector OWNS a foreign (non-hive) table handle, for the per-handle
-     * gateway seams (the connector-level {@code get*Provider(handle)} below and the ~34 guard-and-forward
-     * methods in {@link HiveConnectorMetadata}). Asks each sibling's {@link Connector#ownsHandle} — the sibling
-     * tests its OWN in-loader handle type, which is invisible to the hive loader across the plugin split, so the
-     * gateway can never {@code instanceof} the foreign handle itself.
+     * Resolves which embedded sibling connector OWNS a foreign (non-hive) table handle AND its stable owner label,
+     * for the per-handle gateway seams (the connector-level {@code get*Provider(handle)} below and the
+     * guard-and-forward methods in {@link HiveConnectorMetadata}). Asks each sibling's {@link Connector#ownsHandle}
+     * — the sibling tests its OWN in-loader handle type, which is invisible to the hive loader across the plugin
+     * split, so the gateway can never {@code instanceof} the foreign handle itself. The MATCHED ARM supplies the
+     * label ({@link SiblingOwner#ICEBERG_LABEL} / {@link SiblingOwner#HUDI_LABEL}), so a per-handle forward keys the
+     * per-statement metadata funnel by owner without a force-build or an identity comparison against the suppliers.
      *
      * <p>Consults only ALREADY-BUILT siblings (a plain field read, never {@code getOrCreate*}). The owning
      * sibling is always already built: a foreign handle can only originate from {@code getTableHandle}'s divert,
@@ -187,19 +189,28 @@ HiveConnectorMetadata newMetadata(HmsClient client) {
      * plugin must still route its hudi handles without building an iceberg sibling. Fails loud when no built
      * sibling owns the handle (an orphan handle is a bug, not a route), naming the catalog.
      */
-    private Connector resolveSiblingOwner(ConnectorTableHandle handle) {
+    SiblingOwner resolveSiblingOwnerLabeled(ConnectorTableHandle handle) {
         Connector iceberg = icebergSibling;
         if (iceberg != null && iceberg.ownsHandle(handle)) {
-            return iceberg;
+            return new SiblingOwner(iceberg, SiblingOwner.ICEBERG_LABEL);
         }
         Connector hudi = hudiSibling;
         if (hudi != null && hudi.ownsHandle(handle)) {
-            return hudi;
+            return new SiblingOwner(hudi, SiblingOwner.HUDI_LABEL);
         }
         throw new DorisConnectorException("Cannot route a foreign table handle in catalog '"
                 + context.getCatalogName() + "': no embedded sibling connector owns it");
     }
 
+    /**
+     * The owning sibling {@link Connector} alone (label dropped) for the connector-level per-handle provider seams
+     * ({@code get*Provider(handle)} below), which route by owner but never touch the metadata funnel. Delegates to
+     * {@link #resolveSiblingOwnerLabeled} so the 3-way ownsHandle dispatch has a single implementation.
+     */
+    private Connector resolveSiblingOwner(ConnectorTableHandle handle) {
+        return resolveSiblingOwnerLabeled(handle).connector();
+    }
+
     @Override
     public ConnectorScanPlanProvider getScanPlanProvider() {
         return new HiveScanPlanProvider(getOrCreateClient(), properties, context, readTxnManager, fileListingCache);
@@ -496,6 +507,21 @@ Connector getOrCreateIcebergSibling() {
                                 "Cannot serve iceberg-on-HMS tables in catalog '" + context.getCatalogName()
                                         + "': the iceberg connector plugin is not available");
                     }
+                    // Fail-loud invariant guard for the cache-isolation security track: the hive gateway FRONT
+                    // DOOR never declares SUPPORTS_USER_SESSION, so fe-core keys its per-user schema/name cache
+                    // bypass off THIS (front-door) connector's capabilities and would NOT bypass for a delegated
+                    // sibling. The iceberg sibling is forced iceberg.catalog.type=hms (IcebergSiblingProperties
+                    // .synthesize) and can never be REST session=user, so this must hold today. If a future change
+                    // ever let the sibling be session=user, the front-door-only bypass would silently leak
+                    // cross-user metadata — fail here instead.
+                    if (sibling.getCapabilities().contains(ConnectorCapability.SUPPORTS_USER_SESSION)) {
+                        throw new DorisConnectorException(
+                                "iceberg-on-HMS sibling in catalog '" + context.getCatalogName()
+                                        + "' unexpectedly declares SUPPORTS_USER_SESSION: the hive gateway front "
+                                        + "door is not session=user, so fe-core's per-user schema/name cache bypass "
+                                        + "would not trigger and cross-user metadata would leak. The sibling must "
+                                        + "stay iceberg.catalog.type=hms (never REST session=user).");
+                    }
                     icebergSibling = sibling;
                 }
             }