[feature](paimon) Backport Paimon options and read/write coverage

be/src/format/table/paimon_jni_reader.cpp

@@ -41,8 +41,8 @@ constexpr std::string_view PAIMON_JNI_SCANNER_IO_TMP_DIR = "paimon_jni_scanner_i
 
 const std::string PaimonJniReader::PAIMON_OPTION_PREFIX = "paimon.";
 const std::string PaimonJniReader::HADOOP_OPTION_PREFIX = "hadoop.";
-const std::string PaimonJniReader::DORIS_ENABLE_JNI_IO_MANAGER = "doris.enable_jni_io_manager";
-const std::string PaimonJniReader::DORIS_JNI_IO_MANAGER_TMP_DIR = "doris.jni_io_manager.tmp_dir";
+const std::string PaimonJniReader::DORIS_ENABLE_JNI_IO_MANAGER = "jni.enable_jni_io_manager";
+const std::string PaimonJniReader::DORIS_JNI_IO_MANAGER_TMP_DIR = "jni.io_manager.tmp_dir";
 
 PaimonJniReader::PaimonJniReader(const std::vector<SlotDescriptor*>& file_slot_descs,
                                  RuntimeState* state, RuntimeProfile* profile,

be/src/format_v2/jni/paimon_jni_reader.cpp

@@ -28,8 +28,8 @@ namespace {
 
 constexpr std::string_view PAIMON_OPTION_PREFIX = "paimon.";
 constexpr std::string_view HADOOP_OPTION_PREFIX = "hadoop.";
-constexpr std::string_view DORIS_ENABLE_JNI_IO_MANAGER = "doris.enable_jni_io_manager";
-constexpr std::string_view DORIS_JNI_IO_MANAGER_TMP_DIR = "doris.jni_io_manager.tmp_dir";
+constexpr std::string_view DORIS_ENABLE_JNI_IO_MANAGER = "jni.enable_jni_io_manager";
+constexpr std::string_view DORIS_JNI_IO_MANAGER_TMP_DIR = "jni.io_manager.tmp_dir";
 constexpr std::string_view PAIMON_JNI_SCANNER_IO_TMP_DIR = "paimon_jni_scanner_io_tmp";
 
 const std::string* get_paimon_predicate(const TFileScanRangeParams* scan_params,

be/test/format_v2/table/paimon_reader_test.cpp

@@ -888,17 +888,17 @@ TEST(PaimonHybridReaderTest, FirstNativeAndJniChildInitAreCountedOnce) {
 TEST(PaimonJniReaderTest, BuildScannerParamsKeepsExplicitIOManagerTempDir) {
     auto scan_params = make_paimon_jni_scan_params();
     scan_params.__set_paimon_options({
-            {"doris.enable_jni_io_manager", "true"},
-            {"doris.jni_io_manager.tmp_dir", "/tmp/explicit-paimon-spill"},
-            {"doris.jni_io_manager.impl_class", "org.example.CustomIOManager"},
+            {"jni.enable_jni_io_manager", "true"},
+            {"jni.io_manager.tmp_dir", "/tmp/explicit-paimon-spill"},
+            {"jni.io_manager.impl_class", "org.example.CustomIOManager"},
     });
     RuntimeState state {TQueryOptions(), TQueryGlobals()};
     state.set_exec_env(ExecEnv::GetInstance());
 
     auto params = build_paimon_jni_scanner_params(&scan_params, &state);
-    EXPECT_EQ(params["paimon.doris.enable_jni_io_manager"], "true");
-    EXPECT_EQ(params["paimon.doris.jni_io_manager.tmp_dir"], "/tmp/explicit-paimon-spill");
-    EXPECT_EQ(params["paimon.doris.jni_io_manager.impl_class"], "org.example.CustomIOManager");
+    EXPECT_EQ(params["paimon.jni.enable_jni_io_manager"], "true");
+    EXPECT_EQ(params["paimon.jni.io_manager.tmp_dir"], "/tmp/explicit-paimon-spill");
+    EXPECT_EQ(params["paimon.jni.io_manager.impl_class"], "org.example.CustomIOManager");
 }
 
 TEST(PaimonJniReaderTest, BuildScannerParamsInjectsStorageRootTmpDirForEnabledIOManager) {
@@ -908,14 +908,14 @@ TEST(PaimonJniReaderTest, BuildScannerParamsInjectsStorageRootTmpDirForEnabledIO
     });
     auto scan_params = make_paimon_jni_scan_params();
     scan_params.__set_paimon_options({
-            {"doris.enable_jni_io_manager", "true"},
+            {"jni.enable_jni_io_manager", "true"},
     });
     RuntimeState state {TQueryOptions(), TQueryGlobals()};
     state.set_exec_env(ExecEnv::GetInstance());
 
     auto params = build_paimon_jni_scanner_params(&scan_params, &state);
-    EXPECT_EQ(params["paimon.doris.enable_jni_io_manager"], "true");
-    EXPECT_EQ(params["paimon.doris.jni_io_manager.tmp_dir"],
+    EXPECT_EQ(params["paimon.jni.enable_jni_io_manager"], "true");
+    EXPECT_EQ(params["paimon.jni.io_manager.tmp_dir"],
               "/data1/doris/paimon_jni_scanner_io_tmp:/data2/doris/"
               "paimon_jni_scanner_io_tmp");
 }

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

@@ -24,7 +24,6 @@
 import org.apache.doris.common.security.authentication.PreExecutionAuthenticatorCache;
 
 import com.google.common.base.Preconditions;
-import org.apache.paimon.CoreOptions;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.disk.IOManager;
 import org.apache.paimon.disk.IOManagerImpl;
@@ -50,7 +49,6 @@
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
@@ -66,9 +64,9 @@ public class PaimonJniScanner extends JniScanner {
     private static final String PAIMON_OPTION_PREFIX = "paimon.";
     private static final String ASYNC_READER_THREAD_NAME_PREFIX = "paimon-reader-async-thread";
     private static final String FILE_READER_ASYNC_THRESHOLD = "file-reader-async-threshold";
-    static final String ENABLE_JNI_IO_MANAGER = "paimon.doris.enable_jni_io_manager";
-    static final String JNI_IO_MANAGER_TMP_DIR = "paimon.doris.jni_io_manager.tmp_dir";
-    static final String JNI_IO_MANAGER_IMPL_CLASS = "paimon.doris.jni_io_manager.impl_class";
+    static final String ENABLE_JNI_IO_MANAGER = "paimon.jni.enable_jni_io_manager";
+    static final String JNI_IO_MANAGER_TMP_DIR = "paimon.jni.io_manager.tmp_dir";
+    static final String JNI_IO_MANAGER_IMPL_CLASS = "paimon.jni.io_manager.impl_class";
     private static final AtomicInteger ACTIVE_SCANNERS = new AtomicInteger();
 
     private final Map<String, String> params;
@@ -570,8 +568,6 @@ static Optional<Long> parseDataSizeBytes(String value) {
     private void initTable() {
         Preconditions.checkState(params.containsKey("serialized_table"));
         table = PaimonUtils.deserialize(params.get("serialized_table"));
-        table = table.copy(Collections.singletonMap(
-                CoreOptions.READ_BATCH_SIZE.key(), String.valueOf(batchSize)));
         paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType());
         if (LOG.isDebugEnabled()) {
             LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames);
@@ -584,5 +580,4 @@ private static String[] splitParam(String value, String delimiter) {
         }
         return value.split(delimiter);
     }
-
 }

fe/fe-core/src/main/java/org/apache/doris/analysis/TableScanParams.java

@@ -17,23 +17,27 @@
 
 package org.apache.doris.analysis;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import org.bouncycastle.util.Strings;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
 
 public class TableScanParams {
     public static final String PARAMS_NAME = "name";
     public static final String INCREMENTAL_READ = "incr";
     public static final String BRANCH = "branch";
     public static final String TAG = "tag";
+    public static final String OPTIONS = "options";
     private static final ImmutableSet<String> VALID_PARAM_TYPES = ImmutableSet.of(
             INCREMENTAL_READ,
             BRANCH,
-            TAG);
-
+            TAG,
+            OPTIONS);
     private final String paramType;
     // There are two ways to pass parameters to a function.
     // - One is in map form, where the data is stored in `mapParams`.
@@ -42,18 +46,21 @@ public class TableScanParams {
     //   such as: `listParams` is used for @func_name('value1', 'value2', 'value3')
     private final Map<String, String> mapParams;
     private final List<String> listParams;
+    private volatile Map<String, String> resolvedMapParams;
 
     private void validate() {
         if (!VALID_PARAM_TYPES.contains(paramType)) {
             throw new IllegalArgumentException("Invalid param type: " + paramType);
         }
-        // TODO: validate mapParams and listParams for different param types
+        if (OPTIONS.equals(paramType) && (mapParams.isEmpty() || !listParams.isEmpty())) {
+            throw new IllegalArgumentException("OPTIONS requires a non-empty key/value map");
+        }
     }
 
     public TableScanParams(String paramType, Map<String, String> mapParams, List<String> listParams) {
         this.paramType = Strings.toLowerCase(paramType);
         this.mapParams = mapParams == null ? ImmutableMap.of() : ImmutableMap.copyOf(mapParams);
-        this.listParams = listParams;
+        this.listParams = listParams == null ? ImmutableList.of() : ImmutableList.copyOf(listParams);
         validate();
     }
 
@@ -69,6 +76,34 @@ public Map<String, String> getMapParams() {
         return mapParams;
     }
 
+    /**
+     * Resolve dynamic parameters once for the current statement execution.
+     */
+    public Map<String, String> getOrResolveMapParams(
+            Function<Map<String, String>, Map<String, String>> resolver) {
+        Map<String, String> resolved = resolvedMapParams;
+        if (resolved == null) {
+            synchronized (this) {
+                resolved = resolvedMapParams;
+                if (resolved == null) {
+                    // Binding and scan initialization share this instance. Caching the resolved
+                    // map prevents a mutable external selector from choosing two snapshots.
+                    resolved = ImmutableMap.copyOf(resolver.apply(mapParams));
+                    resolvedMapParams = resolved;
+                }
+            }
+        }
+        return resolved;
+    }
+
+    public Optional<Map<String, String>> getResolvedMapParams() {
+        return Optional.ofNullable(resolvedMapParams);
+    }
+
+    public synchronized void resetResolvedMapParams() {
+        resolvedMapParams = null;
+    }
+
     public boolean incrementalRead() {
         return INCREMENTAL_READ.equals(paramType);
     }
@@ -80,4 +115,8 @@ public boolean isBranch() {
     public boolean isTag() {
         return TAG.equals(paramType);
     }
+
+    public boolean isOptions() {
+        return OPTIONS.equals(paramType);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java

@@ -83,6 +83,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
 /**
  * FileQueryScanNode for querying the file access type of catalog, now only support
@@ -281,10 +282,10 @@ private void setColumnPositionMapping()
         }
 
         // Pre-index columns into a Map for O(1) lookup
-        List<Column> columns = desc.getTable().getFullSchema();
-        Map<String, Integer> columnNameMap = new HashMap<>(columns.size());
-        for (int i = 0; i < columns.size(); i++) {
-            columnNameMap.putIfAbsent(columns.get(i).getName(), i);
+        List<String> columnNames = getFileColumnNames();
+        Map<String, Integer> columnNameMap = new HashMap<>(columnNames.size());
+        for (int i = 0; i < columnNames.size(); i++) {
+            columnNameMap.putIfAbsent(columnNames.get(i), i);
         }
 
         for (TFileScanSlotInfo slot : params.getRequiredSlots()) {
@@ -306,6 +307,12 @@ private void setColumnPositionMapping()
         params.setColumnIdxs(columnIdxs);
     }
 
+    protected List<String> getFileColumnNames() {
+        return desc.getTable().getFullSchema().stream()
+                .map(Column::getName)
+                .collect(Collectors.toList());
+    }
+
     public TFileScanRangeParams getFileScanRangeParams() {
         return params;
     }
@@ -714,7 +721,8 @@ public void setQueryTableSnapshot(TableSnapshot tableSnapshot) {
     }
 
     public TableSnapshot getQueryTableSnapshot() {
-        TableSnapshot snapshot = desc.getRef().getTableSnapShot();
+        // Keep explicit snapshots available when a legacy or unit-built descriptor has no ref.
+        TableSnapshot snapshot = desc.getRef() == null ? null : desc.getRef().getTableSnapShot();
         if (snapshot != null) {
             return snapshot;
         }
@@ -726,7 +734,9 @@ public void setScanParams(TableScanParams scanParams) {
     }
 
     public TableScanParams getScanParams() {
-        TableScanParams scan = desc.getRef().getScanParams();
+        // Unit-built and legacy descriptors may not carry a relation reference; explicit scan
+        // parameters must remain usable for those callers.
+        TableScanParams scan = desc.getRef() == null ? null : desc.getRef().getScanParams();
         if (scan != null) {
             return scan;
         }

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java

@@ -489,11 +489,16 @@ public void createScanRangeLocations() throws UserException {
         super.createScanRangeLocations();
         enableCurrentIcebergScanSemantics();
         // Extract name mapping from Iceberg table properties
-        Optional<Map<Integer, List<String>>> nameMapping = extractNameMapping();
+        initializeIcebergSchemaInfo(extractNameMapping());
+    }
 
+    @VisibleForTesting
+    void initializeIcebergSchemaInfo(Optional<Map<Integer, List<String>>> nameMapping) throws UserException {
         // Equality-delete keys are hidden scan dependencies and need not appear in the query
         // projection. Both scanners need the complete current schema to resolve field ids,
         // historical names, types, and initial defaults when an old data file lacks such a key.
+        // An identity partition column can also be a physical field in files written by an older
+        // partition spec, so preserving the complete schema is required for partition evolution.
         ExternalUtil.initSchemaInfoForAllColumn(params, -1L, source.getTargetTable().getColumns(),
                 nameMapping.orElse(Collections.emptyMap()), nameMapping.isPresent(),
                 getBase64EncodedInitialDefaultsForScan());

fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonLatestSnapshotProjectionLoader.java

@@ -29,6 +29,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.DataTable;
+import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
 
 import java.util.Collections;
@@ -73,8 +74,11 @@ private PaimonSnapshot resolveLatestSnapshot(Table paimonTable) {
         Optional<Snapshot> optionalSnapshot = paimonTable.latestSnapshot();
         if (optionalSnapshot.isPresent()) {
             latestSnapshotId = optionalSnapshot.get().id();
-            snapshotTable = paimonTable.copy(
-                    Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), String.valueOf(latestSnapshotId)));
+            // A schema-only change can be newer than the latest data snapshot. Preserve the
+            // snapshot pin while exposing the current schema so added columns read as null.
+            snapshotTable = ((FileStoreTable) paimonTable.copy(
+                    Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), String.valueOf(latestSnapshotId))))
+                    .copyWithLatestSchema();
         }
         DataTable dataTable = (DataTable) paimonTable;
         long latestSchemaId = dataTable.schemaManager().latest().map(TableSchema::id).orElse(0L);

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java

@@ -34,6 +34,7 @@
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.partition.Partition;
+import org.apache.paimon.table.Table;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -114,12 +115,11 @@ public List<Partition> getPaimonPartitions(NameMapping nameMapping) {
         }
     }
 
-    public org.apache.paimon.table.Table getPaimonTable(NameMapping nameMapping) {
+    public Table getPaimonTable(NameMapping nameMapping) {
         return getPaimonTable(nameMapping, null, null);
     }
 
-    public org.apache.paimon.table.Table getPaimonTable(NameMapping nameMapping, String branch,
-            String queryType) {
+    public Table getPaimonTable(NameMapping nameMapping, String branch, String queryType) {
         makeSureInitialized();
         try {
             Identifier identifier;
@@ -135,7 +135,12 @@ public org.apache.paimon.table.Table getPaimonTable(NameMapping nameMapping, Str
             } else {
                 identifier = new Identifier(nameMapping.getRemoteDbName(), nameMapping.getRemoteTblName());
             }
-            return executionAuthenticator.execute(() -> catalog.getTable(identifier));
+            return executionAuthenticator.execute(() -> {
+                Table table = catalog.getTable(identifier);
+                Map<String, String> tableOptions =
+                        paimonProperties.getTableOptionsForCopy(table.options());
+                return tableOptions.isEmpty() ? table : table.copy(tableOptions);
+            });
         } catch (Exception e) {
             throw new RuntimeException("Failed to get Paimon table:" + getName() + "."
                     + nameMapping.getRemoteDbName() + "." + nameMapping.getRemoteTblName() + "$" + queryType
@@ -174,7 +179,8 @@ public void checkProperties() throws DdlException {
     public void notifyPropertiesUpdated(Map<String, String> updatedProps) {
         super.notifyPropertiesUpdated(updatedProps);
         if (updatedProps.keySet().stream()
-                .anyMatch(key -> CacheSpec.isMetaCacheKeyForEngine(key, PaimonExternalMetaCache.ENGINE))) {
+                .anyMatch(key -> CacheSpec.isMetaCacheKeyForEngine(key, PaimonExternalMetaCache.ENGINE)
+                        || AbstractPaimonProperties.isTableOptionProperty(key))) {
             Env.getCurrentEnv().getExtMetaCacheMgr().removeCatalogByEngine(getId(), PaimonExternalMetaCache.ENGINE);
         }
     }

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java

@@ -106,6 +106,32 @@ public Table getPaimonTable(Optional<MvccSnapshot> snapshot) {
         }
     }
 
+    public Table getPaimonTable(TableScanParams scanParams) {
+        if (scanParams != null && scanParams.isOptions()) {
+            Map<String, String> options = scanParams.getMapParams();
+            Table statementTable = getPaimonTable(MvccUtil.getSnapshotFromContext(this));
+            Table resolutionTable = PaimonScanParams.usesStatementSnapshot(options)
+                    ? statementTable
+                    : getBasePaimonTable();
+            Map<String, String> resolvedOptions = scanParams.getOrResolveMapParams(
+                    relationOptions -> PaimonScanParams.resolveOptions(resolutionTable, relationOptions));
+            // Startup options are normalized to an immutable snapshot before schema binding. The
+            // scan phase reuses that exact resolution instead of consulting a mutable tag or clock.
+            Table table = PaimonScanParams.selectsSchema(resolvedOptions)
+                    ? getBasePaimonTable()
+                    : statementTable;
+            return PaimonScanParams.applyOptions(table, resolvedOptions);
+        }
+        return getPaimonTable(MvccUtil.getSnapshotFromContext(this));
+    }
+
+    public List<Column> getFullSchema(TableScanParams scanParams) {
+        Table table = getPaimonTable(scanParams);
+        return PaimonUtil.parseSchema(table,
+                getCatalog().getEnableMappingVarbinary(),
+                getCatalog().getEnableMappingTimestampTz());
+    }
+
     private PaimonSnapshotCacheValue getPaimonSnapshotCacheValue(Optional<TableSnapshot> tableSnapshot,
             Optional<TableScanParams> scanParams) {
         makeSureInitialized();
@@ -423,7 +449,7 @@ public boolean isPartitionedTable() {
         return !getBasePaimonTable().partitionKeys().isEmpty();
     }
 
-    private Table getBasePaimonTable() {
+    Table getBasePaimonTable() {
         return PaimonUtils.getPaimonTable(this);
     }
 }