[opt](catalog) fe-connector: unify the connector cache foundation, memoize per-statement hot paths, fix iceberg case-insensitive column DDL

fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorStatementScopes.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;
+
+/**
+ * Statement-scoped resolution helper over the neutral {@link ConnectorStatementScope} (reached via
+ * {@link ConnectorSession#getStatementScope()}). A connector routes its "resolve {@code db.table} once per
+ * statement" memo through {@link #resolveInStatement}, so every read / scan / write resolver of one statement
+ * shares the single loaded value and the loader runs at most once — while a reused prepared-statement scope,
+ * reset per execution, still isolates each execution (its {@code queryId} is part of the key).
+ *
+ * <p>This standardizes the security-critical key convention once instead of letting each connector re-derive it:
+ * dropping {@code queryId} would leak a table across executions of a reused prepared statement; dropping
+ * {@code catalogId} would collide across a cross-catalog {@code MERGE}.
+ *
+ * <p><b>{@code keyNamespace} is connector-owned; this module stays source-neutral.</b> {@code keyNamespace}
+ * namespaces the value <em>type</em> stored under the shared {@code (catalogId, db, table, queryId)} coordinate,
+ * so a heterogeneous gateway statement touching two connectors cannot collide on {@code (db, table)} and hand one
+ * connector another's value (a {@link ClassCastException}). Each connector declares its own namespace constant in
+ * its own module (e.g. {@code IcebergStatementScope}, {@code HudiStatementScope}, {@code EsStatementScope},
+ * {@code JdbcConnectorMetadata}) — this API deliberately holds no source names. The convention every
+ * {@code keyNamespace} passed to {@link #resolveInStatement} MUST follow: it is a compile-time constant prefixed
+ * with the connector's connector-type name (its {@code ConnectorProvider.getType()}, e.g. {@code "iceberg"},
+ * {@code "hudi"}). Because {@code getType()} is a connector's unique identity, source-prefixing makes these
+ * namespaces distinct across connectors <em>by construction</em>; each connector's {@code *StatementScope} guards
+ * its own prefix with a unit test.
+ *
+ * <p>This convention governs only the {@code keyNamespace} passed to {@link #resolveInStatement}. Other
+ * per-statement memos that use {@link ConnectorStatementScope}'s {@code computeIfAbsent} /
+ * {@code getOrCreateMetadata} directly own their own key discipline and are out of scope here — e.g. an
+ * engine-reserved family that stores a single value type per key, discriminated by catalog id, stays collision-safe
+ * without a source prefix. A connector that instead memoizes on its own per-statement metadata instance needs no
+ * namespace at all.
+ */
+public final class ConnectorStatementScopes {
+
+    private ConnectorStatementScopes() {
+    }
+
+    /**
+     * Resolves {@code db.table} once per statement and shares the single value across every resolver of the
+     * statement. The key is {@code keyNamespace + ":" + catalogId + ":" + db + ":" + table + ":" + queryId}: the
+     * catalog id isolates a cross-catalog {@code MERGE}, the {@code queryId} isolates each execution of a reused
+     * prepared statement, and {@code keyNamespace} isolates value types across a heterogeneous gateway.
+     * {@code keyNamespace} MUST be a connector-owned, source-prefixed constant (see the class javadoc).
+     * {@code loader} runs at most once per statement; under a {@code null} session or
+     * {@link ConnectorStatementScope#NONE} (offline / no live statement) it runs on every call — byte-identical
+     * to loading every time.
+     */
+    public static <T> T resolveInStatement(ConnectorSession session, String keyNamespace,
+                                           String db, String table, Supplier<T> loader) {
+        if (session == null) {
+            return loader.get();
+        }
+        String key = keyNamespace + ":" + session.getCatalogId() + ":" + db + ":" + table
+                + ":" + session.getQueryId();
+        return session.getStatementScope().computeIfAbsent(key, loader);
+    }
+}

fe/fe-connector/fe-connector-api/src/test/java/org/apache/doris/connector/api/ConnectorStatementScopesTest.java

@@ -0,0 +1,225 @@
+// 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 org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+
+/**
+ * Tests for {@link ConnectorStatementScopes#resolveInStatement}: the shared per-statement resolver whose key is
+ * {@code keyNamespace + ":" + catalogId + ":" + db + ":" + table + ":" + queryId}. It proves the memo collapses
+ * repeat resolves within a statement, that each key axis (namespace / catalog id / db / table / queryId) isolates,
+ * that a null session or {@link ConnectorStatementScope#NONE} degrades to load-every-time, and that the emitted
+ * key is exactly the documented string (the byte contract every connector consumer depends on).
+ */
+public class ConnectorStatementScopesTest {
+
+    @Test
+    public void sameCoordinateResolvesOncePerStatement() {
+        // Same (namespace, catalog, db, table, queryId) within one scope -> loader runs once, one shared instance.
+        // MUTATION: not memoizing -> two loads / two instances -> red.
+        RecordingMemoScope scope = new RecordingMemoScope();
+        TestSession session = new TestSession(7L, "q1", scope);
+        AtomicInteger loads = new AtomicInteger();
+        Object a = ConnectorStatementScopes.resolveInStatement(session, "x.table", "db1", "t", () -> {
+            loads.incrementAndGet();
+            return new Object();
+        });
+        Object b = ConnectorStatementScopes.resolveInStatement(session, "x.table", "db1", "t", () -> {
+            loads.incrementAndGet();
+            return new Object();
+        });
+        Assertions.assertSame(a, b, "same coordinate -> one shared instance");
+        Assertions.assertEquals(1, loads.get(), "loaded once per statement");
+    }
+
+    @Test
+    public void differentQueryIdIsolatesTheLoad() {
+        // A reused prepared statement runs each execution under its own queryId; the memo never crosses executions.
+        RecordingMemoScope scope = new RecordingMemoScope();
+        Object a = ConnectorStatementScopes.resolveInStatement(
+                new TestSession(7L, "q1", scope), "x.table", "db1", "t", Object::new);
+        Object b = ConnectorStatementScopes.resolveInStatement(
+                new TestSession(7L, "q2", scope), "x.table", "db1", "t", Object::new);
+        Assertions.assertNotSame(a, b, "different queryId -> isolated load");
+    }
+
+    @Test
+    public void differentCatalogIdIsolatesTheLoad() {
+        // A cross-catalog MERGE resolves the two catalogs' tables independently (the key carries the catalog id).
+        RecordingMemoScope scope = new RecordingMemoScope();
+        Object a = ConnectorStatementScopes.resolveInStatement(
+                new TestSession(1L, "q1", scope), "x.table", "db1", "t", Object::new);
+        Object b = ConnectorStatementScopes.resolveInStatement(
+                new TestSession(2L, "q1", scope), "x.table", "db1", "t", Object::new);
+        Assertions.assertNotSame(a, b, "different catalog id -> isolated load");
+    }
+
+    @Test
+    public void differentDbOrTableIsolatesTheLoad() {
+        RecordingMemoScope scope = new RecordingMemoScope();
+        TestSession session = new TestSession(7L, "q1", scope);
+        Object t1 = ConnectorStatementScopes.resolveInStatement(session, "x.table", "db1", "t", Object::new);
+        Object t2 = ConnectorStatementScopes.resolveInStatement(session, "x.table", "db1", "u", Object::new);
+        Object t3 = ConnectorStatementScopes.resolveInStatement(session, "x.table", "db2", "t", Object::new);
+        Assertions.assertNotSame(t1, t2, "different table -> isolated load");
+        Assertions.assertNotSame(t1, t3, "different db -> isolated load");
+    }
+
+    @Test
+    public void differentNamespaceIsolatesValueTypes() {
+        // The namespace guards a heterogeneous-gateway statement: two connectors sharing (db, table, catalog,
+        // queryId) but storing different value types must not collide. Without namespacing, the second resolve
+        // would return the first's memoized value cast to the wrong type -> ClassCastException.
+        // MUTATION: dropping keyNamespace from the key -> ClassCastException here -> red.
+        RecordingMemoScope scope = new RecordingMemoScope();
+        TestSession session = new TestSession(7L, "q1", scope);
+        String asString = ConnectorStatementScopes.resolveInStatement(
+                session, "a.value", "db1", "t", () -> "stringValue");
+        Integer asInt = ConnectorStatementScopes.resolveInStatement(
+                session, "b.value", "db1", "t", () -> 42);
+        Assertions.assertEquals("stringValue", asString, "namespace a keeps its String value");
+        Assertions.assertEquals(Integer.valueOf(42), asInt, "namespace b keeps its Integer value (no collision)");
+    }
+
+    @Test
+    public void nullSessionLoadsEveryTime() {
+        // No session (offline / direct-construction tests): each call loads, byte-identical to loading every time.
+        AtomicInteger loads = new AtomicInteger();
+        ConnectorStatementScopes.resolveInStatement(null, "x.table", "db1", "t", () -> {
+            loads.incrementAndGet();
+            return new Object();
+        });
+        ConnectorStatementScopes.resolveInStatement(null, "x.table", "db1", "t", () -> {
+            loads.incrementAndGet();
+            return new Object();
+        });
+        Assertions.assertEquals(2, loads.get(), "null session -> load every time");
+    }
+
+    @Test
+    public void noneScopeLoadsEveryTime() {
+        // A live session whose scope is NONE (no per-statement context) also loads every time.
+        TestSession session = new TestSession(7L, "q1", ConnectorStatementScope.NONE);
+        AtomicInteger loads = new AtomicInteger();
+        ConnectorStatementScopes.resolveInStatement(session, "x.table", "db1", "t", () -> {
+            loads.incrementAndGet();
+            return new Object();
+        });
+        ConnectorStatementScopes.resolveInStatement(session, "x.table", "db1", "t", () -> {
+            loads.incrementAndGet();
+            return new Object();
+        });
+        Assertions.assertEquals(2, loads.get(), "NONE scope -> load every time");
+    }
+
+    @Test
+    public void keyIsNamespaceCatalogDbTableQueryId() {
+        // The exact byte contract: keyNamespace + ":" + catalogId + ":" + db + ":" + table + ":" + queryId.
+        // Every connector consumer's cross-statement isolation depends on this string; assert it verbatim.
+        RecordingMemoScope scope = new RecordingMemoScope();
+        ConnectorStatementScopes.resolveInStatement(
+                new TestSession(7L, "q1", scope), "test.ns", "db1", "t", Object::new);
+        Assertions.assertEquals("test.ns:7:db1:t:q1", scope.lastKey, "emitted key must match the documented format");
+    }
+
+    /** A statement scope that memoizes like the engine's real one and records the last key, for the assertions. */
+    private static final class RecordingMemoScope implements ConnectorStatementScope {
+        private final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>();
+        private String lastKey;
+
+        @Override
+        @SuppressWarnings("unchecked")
+        public <T> T computeIfAbsent(String key, Supplier<T> loader) {
+            lastKey = key;
+            return (T) cache.computeIfAbsent(key, k -> loader.get());
+        }
+    }
+
+    /** Minimal {@link ConnectorSession} carrying a catalog id, queryId and scope for the key + memo assertions. */
+    private static final class TestSession implements ConnectorSession {
+        private final long catalogId;
+        private final String queryId;
+        private final ConnectorStatementScope scope;
+
+        TestSession(long catalogId, String queryId, ConnectorStatementScope scope) {
+            this.catalogId = catalogId;
+            this.queryId = queryId;
+            this.scope = scope;
+        }
+
+        @Override
+        public long getCatalogId() {
+            return catalogId;
+        }
+
+        @Override
+        public String getQueryId() {
+            return queryId;
+        }
+
+        @Override
+        public String getSessionId() {
+            // Deliberately != queryId so keyIsNamespaceCatalogDbTableQueryId pins the per-execution queryId
+            // (cross-query isolation), not the stable per-connection sessionId; a queryId->sessionId swap in the
+            // helper's key would share a table across queries of one connection and MUST turn that assertion red.
+            return "session-" + queryId;
+        }
+
+        @Override
+        public ConnectorStatementScope getStatementScope() {
+            return scope;
+        }
+
+        @Override
+        public String getUser() {
+            return "u";
+        }
+
+        @Override
+        public String getTimeZone() {
+            return "UTC";
+        }
+
+        @Override
+        public String getLocale() {
+            return "en_US";
+        }
+
+        @Override
+        public String getCatalogName() {
+            return "c";
+        }
+
+        @Override
+        public <T> T getProperty(String name, Class<T> type) {
+            return null;
+        }
+
+        @Override
+        public Map<String, String> getCatalogProperties() {
+            return Collections.emptyMap();
+        }
+    }
+}

fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CacheSpec.java

@@ -67,6 +67,23 @@ public static CacheSpec of(boolean enable, long ttlSecond, long capacity) {
         return new CacheSpec(enable, ttlSecond, capacity);
     }
 
+    /**
+     * Build an ENABLED spec from a connector-resolved TTL under the "{@code <= 0} disables" contract.
+     *
+     * <p>A connector that resolves its own single {@code ttl-second} knob (iceberg's shared
+     * {@code meta.cache.iceberg.table.ttl-second}, paimon's snapshot cache) treats any non-positive TTL as
+     * "disable caching, always read live". That is NOT the raw {@link CacheSpec} contract, which reads
+     * {@code ttlSecond == -1} as {@link #CACHE_NO_TTL} ("no expiration", still ENABLED) and only
+     * {@code ttlSecond == 0} as {@link #CACHE_TTL_DISABLE_CACHE} ("disabled"). This factory folds any
+     * non-positive TTL to the disable sentinel so a negative operator value disables the cache rather than
+     * silently becoming a never-expiring one. It is exactly the
+     * {@code ttlSecond > 0 ? of(true, ttlSecond, capacity) : of(true, CACHE_TTL_DISABLE_CACHE, capacity)}
+     * expression each per-catalog cache used to inline.
+     */
+    public static CacheSpec ofConnectorTtl(long ttlSecond, long capacity) {
+        return of(true, ttlSecond > 0 ? ttlSecond : CACHE_TTL_DISABLE_CACHE, capacity);
+    }
+
     public static PropertySpec.Builder propertySpecBuilder() {
         return new PropertySpec.Builder();
     }