[draft](catalog) Master catalog spi 07 paimon

build.sh

@@ -1094,6 +1094,23 @@ if [[ "${BUILD_FE}" -eq 1 ]]; then
     done
     unset CONN_PLUGIN_DIR conn_module conn_plugin_target conn_module_dir conn_zip
 
+    # RC-4: self-contain the paimon connector plugin for OSS. The connector sets
+    # fs.oss.impl=com.aliyun.jindodata.oss.JindoOssFileSystem; that impl lives in the jindofs jars,
+    # which are packaged from thirdparty by post-build.sh into fe/lib/jindofs (NOT a maven artifact).
+    # The plugin runs child-first, so without its OWN copy JindoOssFileSystem resolves from the parent
+    # 'app' classloader and cannot be cast to the plugin's child-loaded org.apache.hadoop.fs.FileSystem.
+    # Copy the jindofs jars into the paimon plugin lib so JindoOssFileSystem loads child-first alongside
+    # the plugin's own hadoop FileSystem (same self-contained intent as the bundled hadoop-aws/S3A).
+    # Naturally gated: a no-op unless jindofs was packaged (--jindofs / DISABLE_BUILD_JINDOFS=OFF).
+    # CAVEAT (docker-gated, enablePaimonTest=true): jindo-core ships a native lib that can bind to only one
+    # classloader per JVM, so this is safe only while no concurrent non-paimon path loads jindo from
+    # fe/lib/jindofs in the same FE process.
+    PAIMON_CONN_LIB="${DORIS_OUTPUT}/fe/plugins/connector/paimon/lib"
+    if [[ -d "${PAIMON_CONN_LIB}" && -d "${DORIS_OUTPUT}/fe/lib/jindofs" ]]; then
+        cp -p "${DORIS_OUTPUT}/fe/lib/jindofs/"*.jar "${PAIMON_CONN_LIB}/" 2>/dev/null || true
+    fi
+    unset PAIMON_CONN_LIB
+
     if [ "${TARGET_SYSTEM}" = "Darwin" ] || [ "${TARGET_SYSTEM}" = "Linux" ]; then
       mkdir -p "${DORIS_OUTPUT}/fe/arthas"
       rm -rf "${DORIS_OUTPUT}/fe/arthas/*"

fe/fe-common/src/main/java/org/apache/doris/common/maxcompute/MCUtils.java → fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MCUtils.java

@@ -15,7 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package org.apache.doris.common.maxcompute;
+package org.apache.doris.maxcompute;
+
+import org.apache.doris.common.maxcompute.MCProperties;
 
 import com.aliyun.auth.credentials.Credential;
 import com.aliyun.auth.credentials.provider.EcsRamRoleCredentialProvider;

fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java

@@ -19,7 +19,6 @@
 
 import org.apache.doris.common.jni.JniScanner;
 import org.apache.doris.common.jni.vec.ColumnType;
-import org.apache.doris.common.maxcompute.MCUtils;
 
 import com.aliyun.odps.Odps;
 import com.aliyun.odps.table.configuration.CompressionCodec;

fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MaxComputeJniWriter.java

@@ -21,7 +21,6 @@
 import org.apache.doris.common.jni.vec.VectorColumn;
 import org.apache.doris.common.jni.vec.VectorTable;
 import org.apache.doris.common.maxcompute.MCProperties;
-import org.apache.doris.common.maxcompute.MCUtils;
 
 import com.aliyun.odps.Odps;
 import com.aliyun.odps.OdpsType;

fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java

@@ -37,6 +37,7 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.TimeZone;
@@ -134,6 +135,12 @@ private int[] getProjected() {
     }
 
     private List<Predicate> getPredicates() {
+        // Backstop for a missing paimon_predicate param (scan with no pushed-down filter): a null here means
+        // "no filter", not an error. Guard the unconditional deserialize so the JNI reader never NPEs on
+        // deserialize(null) ("encodedStr is null"). The FE producer also always emits an (empty) predicate now.
+        if (paimonPredicate == null) {
+            return Collections.emptyList();
+        }
         List<Predicate> predicates = PaimonUtils.deserialize(paimonPredicate);
         if (LOG.isDebugEnabled()) {
             LOG.debug("predicates:{}", predicates);

fe/be-java-extensions/preload-extensions/pom.xml

@@ -62,6 +62,18 @@ under the License.
             <artifactId>commons-io</artifactId>
             <version>${commons-io.version}</version>
         </dependency>
+        <!-- Runtime-only: Hive-using JNI scanners (paimon-hive-connector / hudi / HiveConf)
+             need org.apache.commons.lang.StringUtils (commons-lang 2.x) at runtime. It used to
+             arrive transitively via fe-common's odps-sdk-core; after P4-T09 (a53f2b17b8d) made
+             fe-common odps-free it was evicted from this shared preload classpath, breaking every
+             scanner with NoClassDefFoundError. Restore it here (version managed by fe/pom.xml) so
+             it lands in the preload-extensions runtime classpath, the shared parent of all JNI
+             scanners, while staying out of java-udf's own jar. -->
+        <dependency>
+            <groupId>commons-lang</groupId>
+            <artifactId>commons-lang</artifactId>
+            <scope>runtime</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.arrow</groupId>
             <artifactId>arrow-memory-unsafe</artifactId>

fe/fe-common/pom.xml

@@ -134,23 +134,15 @@ under the License.
             <artifactId>antlr4-runtime</artifactId>
             <version>${antlr4.version}</version>
         </dependency>
+        <!-- Used by DorisHttpException (netty) and GsonUtilsBase (protobuf); previously pulled in
+             transitively via odps-sdk-core, now declared directly so fe-common is odps-free. -->
         <dependency>
-            <groupId>com.aliyun.odps</groupId>
-            <artifactId>odps-sdk-core</artifactId>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.apache.arrow</groupId>
-                    <artifactId>arrow-vector</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.ini4j</groupId>
-                    <artifactId>ini4j</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.bouncycastle</groupId>
-                    <artifactId>bcprov-jdk18on</artifactId>
-                </exclusion>
-            </exclusions>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
         </dependency>
     </dependencies>
     <build>

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

@@ -18,11 +18,13 @@
 package org.apache.doris.connector.api;
 
 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.List;
+import java.util.OptionalLong;
 import java.util.Set;
 
 /**
@@ -41,6 +43,14 @@ default ConnectorScanPlanProvider getScanPlanProvider() {
         return null;
     }
 
+    /**
+     * Returns the write plan provider for sink ({@code TDataSink}) generation,
+     * or {@code null} if this connector does not support writes.
+     */
+    default ConnectorWritePlanProvider getWritePlanProvider() {
+        return null;
+    }
+
     /** Returns the set of capabilities this connector supports. */
     default Set<ConnectorCapability> getCapabilities() {
         return Collections.emptySet();
@@ -118,4 +128,28 @@ default void close() throws IOException {
     default String executeRestRequest(String path, String body) {
         throw new UnsupportedOperationException("REST passthrough not supported by this connector");
     }
+
+    /**
+     * Invalidates any connector-side per-table cache (e.g. a latest-snapshot/version cache) so a subsequent
+     * read reflects the latest external state. Called by the engine on {@code REFRESH TABLE}. The names are
+     * the REMOTE db/table names (as seen by the connector). Default no-op for connectors that cache nothing.
+     */
+    default void invalidateTable(String dbName, String tableName) {
+    }
+
+    /** Invalidates all connector-side per-table caches. Default no-op. */
+    default void invalidateAll() {
+    }
+
+    /**
+     * Optional per-connector override of the catalog's schema-cache TTL (in seconds), consulted generically by
+     * the engine when sizing the schema meta-cache. Semantics match {@code schema.cache.ttl-second}:
+     * {@code 0} disables schema caching (always read fresh), {@code -1} = no expiration, {@code > 0} = TTL.
+     * Lets a connector make its own cache knob also govern schema freshness (e.g. paimon's
+     * {@code meta.cache.paimon.table.ttl-second}, which legacy used for the whole table cache). An explicit
+     * user {@code schema.cache.ttl-second} always wins over this. Default: no override.
+     */
+    default OptionalLong schemaCacheTtlSecondOverride() {
+        return OptionalLong.empty();
+    }
 }

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

@@ -49,12 +49,54 @@ public enum ConnectorCapability {
      * parallel writers should declare this capability.</p>
      */
     SUPPORTS_PARALLEL_WRITE,
+    /**
+     * Indicates the connector requires dynamic-partition writes to be hash-distributed by
+     * partition columns and locally sorted by them before reaching the sink.
+     *
+     * <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>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>
+     */
+    SINK_REQUIRE_PARTITION_LOCAL_SORT,
+    /**
+     * 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.
+     *
+     * <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>
+     */
+    SINK_REQUIRE_FULL_SCHEMA_ORDER,
     /**
      * Indicates the connector supports passthrough query via the {@code query()} TVF.
      *
      * <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
+    SUPPORTS_PASSTHROUGH_QUERY,
+    /**
+     * Indicates the connector exposes per-partition statistics (record count, on-disk size,
+     * file count) via {@link ConnectorTableOps#listPartitions}.
+     *
+     * <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>
+     */
+    SUPPORTS_PARTITION_STATS
 }

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

@@ -30,6 +30,13 @@ public final class ConnectorColumn {
     private final boolean nullable;
     private final String defaultValue;
     private final boolean isKey;
+    private final boolean isAutoInc;
+    private final boolean isAggregated;
+    // Marks a "with local time zone" timestamp column. fe-core's ConnectorColumnConverter translates
+    // this into Column.setWithTZExtraInfo() so DESC shows the WITH_TIMEZONE "Extra" marker, matching
+    // legacy PaimonExternalTable/PaimonSysExternalTable/IcebergUtils which set it from the SOURCE type
+    // root regardless of the timestamp_tz mapping flag. Defaults false; set via withTimeZone().
+    private final boolean withTimeZone;
 
     public ConnectorColumn(String name, ConnectorType type, String comment,
             boolean nullable, String defaultValue) {
@@ -38,12 +45,42 @@ public ConnectorColumn(String name, ConnectorType type, String comment,
 
     public ConnectorColumn(String name, ConnectorType type, String comment,
             boolean nullable, String defaultValue, boolean isKey) {
+        this(name, type, comment, nullable, defaultValue, isKey, false);
+    }
+
+    public ConnectorColumn(String name, ConnectorType type, String comment,
+            boolean nullable, String defaultValue, boolean isKey, boolean isAutoInc) {
+        this(name, type, comment, nullable, defaultValue, isKey, isAutoInc, false);
+    }
+
+    public ConnectorColumn(String name, ConnectorType type, String comment,
+            boolean nullable, String defaultValue, boolean isKey, boolean isAutoInc,
+            boolean isAggregated) {
+        this(name, type, comment, nullable, defaultValue, isKey, isAutoInc, isAggregated, false);
+    }
+
+    private ConnectorColumn(String name, ConnectorType type, String comment,
+            boolean nullable, String defaultValue, boolean isKey, boolean isAutoInc,
+            boolean isAggregated, boolean withTimeZone) {
         this.name = Objects.requireNonNull(name, "name");
         this.type = Objects.requireNonNull(type, "type");
         this.comment = comment;
         this.nullable = nullable;
         this.defaultValue = defaultValue;
         this.isKey = isKey;
+        this.isAutoInc = isAutoInc;
+        this.isAggregated = isAggregated;
+        this.withTimeZone = withTimeZone;
+    }
+
+    /**
+     * Returns a copy of this column marked as a "with local time zone" timestamp. See
+     * {@link #isWithTimeZone()}; the marker is intentionally orthogonal to the mapped {@link #getType()}
+     * so it survives even when the column is mapped to a plain DATETIME (timestamp_tz mapping off).
+     */
+    public ConnectorColumn withTimeZone() {
+        return new ConnectorColumn(name, type, comment, nullable, defaultValue,
+                isKey, isAutoInc, isAggregated, true);
     }
 
     public String getName() {
@@ -70,6 +107,18 @@ public boolean isKey() {
         return isKey;
     }
 
+    public boolean isAutoInc() {
+        return isAutoInc;
+    }
+
+    public boolean isAggregated() {
+        return isAggregated;
+    }
+
+    public boolean isWithTimeZone() {
+        return withTimeZone;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -81,6 +130,9 @@ public boolean equals(Object o) {
         ConnectorColumn that = (ConnectorColumn) o;
         return nullable == that.nullable
                 && isKey == that.isKey
+                && isAutoInc == that.isAutoInc
+                && isAggregated == that.isAggregated
+                && withTimeZone == that.withTimeZone
                 && name.equals(that.name)
                 && type.equals(that.type)
                 && Objects.equals(comment, that.comment)
@@ -89,7 +141,8 @@ public boolean equals(Object o) {
 
     @Override
     public int hashCode() {
-        return Objects.hash(name, type, comment, nullable, defaultValue, isKey);
+        return Objects.hash(name, type, comment, nullable, defaultValue, isKey, isAutoInc, isAggregated,
+                withTimeZone);
     }
 
     @Override