[fix](paimon) Harden JNI reader options and transport

be/src/format/jni/jni_data_bridge.cpp

@@ -38,6 +38,8 @@
 #include "core/data_type/primitive_type.h"
 #include "core/types.h"
 #include "core/value/decimalv2_value.h"
+#include "util/string_util.h"
+#include "util/url_coding.h"
 
 namespace doris {
 
@@ -490,6 +492,55 @@ std::string JniDataBridge::get_jni_type_with_different_string(const DataTypePtr&
     }
 }
 
+std::string JniDataBridge::encode_schema_values(const std::vector<std::string>& values) {
+    std::vector<std::string> encoded_values;
+    encoded_values.reserve(values.size());
+    for (const auto& value : values) {
+        std::string encoded;
+        base64_encode(value, &encoded);
+        // Prefix every element so an empty Base64 token remains distinct from an empty list.
+        encoded_values.emplace_back("$" + encoded);
+    }
+    return join(encoded_values, ",");
+}
+
+std::string JniDataBridge::get_jni_type_with_encoded_struct_fields(const DataTypePtr& data_type) {
+    switch (data_type->get_primitive_type()) {
+    case TYPE_STRUCT: {
+        const auto* type_struct =
+                assert_cast<const DataTypeStruct*>(remove_nullable(data_type).get());
+        std::ostringstream buffer;
+        buffer << "struct<";
+        for (int i = 0; i < type_struct->get_elements().size(); ++i) {
+            if (i != 0) {
+                buffer << ",";
+            }
+            std::string encoded_name;
+            base64_encode(type_struct->get_element_name(i), &encoded_name);
+            // '$' versions the nested-name token and cannot collide with Base64 or grammar
+            // delimiters; Java rejects an unversioned name in the encoded schema path.
+            buffer << "$" << encoded_name << ":"
+                   << get_jni_type_with_encoded_struct_fields(type_struct->get_element(i));
+        }
+        buffer << ">";
+        return buffer.str();
+    }
+    case TYPE_ARRAY: {
+        const auto* type_array =
+                assert_cast<const DataTypeArray*>(remove_nullable(data_type).get());
+        return "array<" + get_jni_type_with_encoded_struct_fields(type_array->get_nested_type()) +
+               ">";
+    }
+    case TYPE_MAP: {
+        const auto* type_map = assert_cast<const DataTypeMap*>(remove_nullable(data_type).get());
+        return "map<" + get_jni_type_with_encoded_struct_fields(type_map->get_key_type()) + "," +
+               get_jni_type_with_encoded_struct_fields(type_map->get_value_type()) + ">";
+    }
+    default:
+        return get_jni_type_with_different_string(data_type);
+    }
+}
+
 Status JniDataBridge::_fill_column_meta(const ColumnPtr& doris_column, const DataTypePtr& data_type,
                                         std::vector<long>& meta_data) {
     auto logical_type = data_type->get_primitive_type();

be/src/format/jni/jni_data_bridge.h

@@ -140,6 +140,12 @@ class JniDataBridge {
      */
     static std::string get_jni_type_with_different_string(const DataTypePtr& data_type);
 
+    /** Encodes every list element independently so schema delimiters remain unambiguous. */
+    static std::string encode_schema_values(const std::vector<std::string>& values);
+
+    /** Encodes STRUCT field names inside a JNI type descriptor. */
+    static std::string get_jni_type_with_encoded_struct_fields(const DataTypePtr& data_type);
+
 private:
     // Column fill helpers for various types
     static Status _fill_string_column(TableMetaAddress& address, MutableColumnPtr& doris_column,

be/src/format/table/paimon_jni_reader.cpp

@@ -54,10 +54,14 @@ PaimonJniReader::PaimonJniReader(const std::vector<SlotDescriptor*>& file_slot_d
                   [&]() {
                       std::vector<std::string> column_names;
                       std::vector<std::string> column_types;
+                      std::vector<std::string> encoded_column_types;
                       for (const auto& desc : file_slot_descs) {
                           column_names.emplace_back(desc->col_name());
                           column_types.emplace_back(
                                   JniDataBridge::get_jni_type_with_different_string(desc->type()));
+                          encoded_column_types.emplace_back(
+                                  JniDataBridge::get_jni_type_with_encoded_struct_fields(
+                                          desc->type()));
                       }
                       const auto& paimon_params = range.table_format_params.paimon_params;
                       std::map<String, String> params;
@@ -70,6 +74,12 @@ PaimonJniReader::PaimonJniReader(const std::vector<SlotDescriptor*>& file_slot_d
                       }
                       params["required_fields"] = join(column_names, ",");
                       params["columns_types"] = join(column_types, "#");
+                      // V1 and V2 must publish the same safe schema protocol because session
+                      // routing can select either producer for the same Paimon scanner.
+                      params["required_fields_base64"] =
+                              JniDataBridge::encode_schema_values(column_names);
+                      params["columns_types_base64"] =
+                              JniDataBridge::encode_schema_values(encoded_column_types);
                       params["time_zone"] = state->timezone();
                       if (range_params->__isset.serialized_table) {
                           params["serialized_table"] = range_params->serialized_table;

be/src/format_v2/jni/jni_table_reader.cpp

@@ -496,11 +496,16 @@ Status JniTableReader::_set_open_scanner_batch_size(size_t batch_size) {
 }
 
 void JniTableReader::_prepare_jni_scanner_schema() {
+    const bool publish_encoded_schema = publishes_encoded_schema();
     std::vector<std::string> required_fields;
     std::vector<std::string> column_types;
+    std::vector<std::string> encoded_column_types;
     std::vector<std::string> replace_types;
     required_fields.reserve(_jni_columns.size());
     column_types.reserve(_jni_columns.size());
+    if (publish_encoded_schema) {
+        encoded_column_types.reserve(_jni_columns.size());
+    }
     replace_types.reserve(_jni_columns.size());
     _jni_block_template.clear();
     _jni_block_template.reserve(_jni_columns.size());
@@ -511,13 +516,25 @@ void JniTableReader::_prepare_jni_scanner_schema() {
         required_fields.push_back(column.java_name);
         column_types.push_back(
                 JniDataBridge::get_jni_type_with_different_string(column.transfer_type));
+        if (publish_encoded_schema) {
+            encoded_column_types.push_back(
+                    JniDataBridge::get_jni_type_with_encoded_struct_fields(column.transfer_type));
+        }
         replace_types.push_back(column.replace_type);
         has_replace_type = has_replace_type || column.replace_type != "not_replace";
         _jni_block_template.insert(
                 {column.transfer_type->create_column(), column.transfer_type, column.java_name});
     }
     _scanner_params["required_fields"] = join(required_fields, ",");
     _scanner_params["columns_types"] = join(column_types, "#");
+    if (publish_encoded_schema) {
+        // Only Paimon consumes the paired payload. Keeping it capability-gated avoids recursively
+        // encoding nested types for every split of unrelated V2 JNI connectors.
+        _scanner_params["required_fields_base64"] =
+                JniDataBridge::encode_schema_values(required_fields);
+        _scanner_params["columns_types_base64"] =
+                JniDataBridge::encode_schema_values(encoded_column_types);
+    }
     if (has_replace_type) {
         _scanner_params["replace_string"] = join(replace_types, ",");
     }

be/src/format_v2/jni/jni_table_reader.h

@@ -81,6 +81,7 @@ class JniTableReader : public TableReader {
     virtual Status _close_jni_scanner();
     virtual Status _set_open_scanner_batch_size(size_t batch_size);
     virtual bool supports_batch_size_update_after_open() const { return true; }
+    virtual bool publishes_encoded_schema() const { return false; }
     virtual Status _open_jni_scanner();
     bool _reserve_split_profile_publication();
     const std::vector<JniColumn>& jni_columns() const { return _jni_columns; }

be/src/format_v2/jni/paimon_jni_reader.h

@@ -51,6 +51,7 @@ class PaimonJniReader final : public format::JniTableReader {
     Status validate_scan_range(const TFileRangeDesc& range) const override;
     Status build_scanner_params(std::map<std::string, std::string>* params) const override;
     bool supports_batch_size_update_after_open() const override { return false; }
+    bool publishes_encoded_schema() const override { return true; }
 
 private:
     static std::string build_default_io_manager_tmp_dirs(const std::vector<StorePath>& store_paths);

be/test/format_v2/jni/jni_table_reader_test.cpp

@@ -27,6 +27,9 @@
 #include <thread>
 #include <vector>
 
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_struct.h"
+#include "format/jni/jni_data_bridge.h"
 #include "io/io_common.h"
 
 namespace doris::format {
@@ -109,6 +112,36 @@ Status init_reader(FakeJniTableReader* reader, const std::shared_ptr<io::IOConte
     });
 }
 
+TEST(JniTableReaderTest, RequiredFieldEncodingPreservesQuotedIdentifiers) {
+    EXPECT_EQ(JniDataBridge::encode_schema_values({"region,code", "hash#name", "地区 名"}),
+              "$cmVnaW9uLGNvZGU=,$aGFzaCNuYW1l,$5Zyw5Yy6IOWQjQ==");
+    EXPECT_EQ(JniDataBridge::encode_schema_values({}), "");
+    EXPECT_EQ(JniDataBridge::encode_schema_values({""}), "$");
+}
+
+TEST(JniTableReaderTest, EncodedTypeDescriptorsPreserveNestedQuotedIdentifiers) {
+    auto type = std::make_shared<DataTypeStruct>(
+            DataTypes {std::make_shared<DataTypeString>(), std::make_shared<DataTypeString>(),
+                       std::make_shared<DataTypeString>()},
+            Strings {"hash#name", "region,code", "colon:name"});
+
+    EXPECT_EQ(JniDataBridge::get_jni_type_with_encoded_struct_fields(type),
+              "struct<$aGFzaCNuYW1l:string,$cmVnaW9uLGNvZGU:string,$Y29sb246bmFtZQ:string>");
+}
+
+TEST(JniTableReaderTest, GenericConnectorDoesNotPublishPaimonEncodedSchema) {
+    FakeJniTableReader reader;
+    reader._jni_columns = {JniTableReader::JniColumn {
+            .java_name = "region,code",
+            .transfer_type = std::make_shared<DataTypeString>(),
+    }};
+
+    reader._prepare_jni_scanner_schema();
+
+    EXPECT_FALSE(reader._scanner_params.contains("required_fields_base64"));
+    EXPECT_FALSE(reader._scanner_params.contains("columns_types_base64"));
+}
+
 TEST(JniTableReaderTest, CancellationStopsBeforeFetchingAnotherJavaBatch) {
     auto io_ctx = std::make_shared<io::IOContext>();
     FakeJniTableReader reader;

be/test/format_v2/jni/paimon_jni_reader_test.cpp

@@ -23,6 +23,7 @@
 #include <string>
 #include <utility>
 
+#include "core/data_type/data_type_string.h"
 #include "format_v2/table_reader.h"
 #include "gen_cpp/PlanNodes_types.h"
 #include "runtime/runtime_state.h"
@@ -161,6 +162,19 @@ TEST(PaimonJniReaderTest, ScanLevelOptionsOverrideLegacySplitFallbacks) {
     EXPECT_EQ(params["hadoop.source"], "scan");
 }
 
+TEST(PaimonJniReaderTest, PublishesEncodedSchemaForQuotedIdentifiers) {
+    PaimonJniReader reader;
+    reader._jni_columns = {JniTableReader::JniColumn {
+            .java_name = "region,code",
+            .transfer_type = std::make_shared<DataTypeString>(),
+    }};
+
+    reader._prepare_jni_scanner_schema();
+
+    EXPECT_EQ(reader._scanner_params.at("required_fields_base64"), "$cmVnaW9uLGNvZGU=");
+    EXPECT_TRUE(reader._scanner_params.contains("columns_types_base64"));
+}
+
 TEST(PaimonJniReaderTest, UsesStableRuntimeBatchSizeBeforeAndAfterOpen) {
     TQueryOptions query_options;
     query_options.__set_batch_size(8160);

fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ColumnType.java

@@ -17,8 +17,10 @@
 
 package org.apache.doris.common.jni.vec;
 
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Base64;
 import java.util.Collections;
 import java.util.List;
 import java.util.regex.Matcher;
@@ -277,6 +279,15 @@ private static int findNextNestedField(String commaSplitFields) {
     }
 
     public static ColumnType parseType(String columnName, String hiveType) {
+        return parseType(columnName, hiveType, false);
+    }
+
+    public static ColumnType parseTypeWithEncodedStructFields(String columnName, String hiveType) {
+        return parseType(columnName, hiveType, true);
+    }
+
+    private static ColumnType parseType(String columnName, String hiveType, boolean encodedStructFields) {
+        String originalType = hiveType.trim();
         String lowerCaseType = hiveType.toLowerCase();
         Type type = Type.UNSUPPORTED;
         int length = -1;
@@ -397,20 +408,21 @@ public static ColumnType parseType(String columnName, String hiveType) {
                     if (lowerCaseType.indexOf("<") == 5
                             && lowerCaseType.lastIndexOf(">") == lowerCaseType.length() - 1) {
                         ColumnType nestedType = parseType("element",
-                                lowerCaseType.substring(6, lowerCaseType.length() - 1));
+                                originalType.substring(6, originalType.length() - 1), encodedStructFields);
                         ColumnType arrayType = new ColumnType(columnName, Type.ARRAY);
                         arrayType.setChildTypes(Collections.singletonList(nestedType));
                         return arrayType;
                     }
                 } else if (lowerCaseType.startsWith("map")) {
                     if (lowerCaseType.indexOf("<") == 3
                             && lowerCaseType.lastIndexOf(">") == lowerCaseType.length() - 1) {
-                        String keyValue = lowerCaseType.substring(4, lowerCaseType.length() - 1);
+                        String keyValue = originalType.substring(4, originalType.length() - 1);
                         int index = findNextNestedField(keyValue);
                         if (index != keyValue.length() && index != 0) {
-                            ColumnType keyType = parseType("key", keyValue.substring(0, index).trim());
+                            ColumnType keyType = parseType(
+                                    "key", keyValue.substring(0, index).trim(), encodedStructFields);
                             ColumnType valueType =
-                                    parseType("value", keyValue.substring(index + 1).trim());
+                                    parseType("value", keyValue.substring(index + 1).trim(), encodedStructFields);
                             ColumnType mapType = new ColumnType(columnName, Type.MAP);
                             mapType.setChildTypes(Arrays.asList(keyType, valueType));
                             return mapType;
@@ -419,16 +431,31 @@ public static ColumnType parseType(String columnName, String hiveType) {
                 } else if (lowerCaseType.startsWith("struct")) {
                     if (lowerCaseType.indexOf("<") == 6
                             && lowerCaseType.lastIndexOf(">") == lowerCaseType.length() - 1) {
-                        String listFields = lowerCaseType.substring(7, lowerCaseType.length() - 1);
+                        String listFields = originalType.substring(7, originalType.length() - 1);
                         ArrayList<ColumnType> fields = new ArrayList<>();
                         ArrayList<String> names = new ArrayList<>();
                         while (listFields.length() > 0) {
                             int index = findNextNestedField(listFields);
                             int pivot = listFields.indexOf(':');
                             if (pivot > 0 && pivot < listFields.length() - 1) {
-                                fields.add(parseType(listFields.substring(0, pivot),
-                                        listFields.substring(pivot + 1, index)));
-                                names.add(listFields.substring(0, pivot));
+                                String fieldName = listFields.substring(0, pivot);
+                                if (encodedStructFields) {
+                                    if (!fieldName.startsWith("$")) {
+                                        throw new IllegalArgumentException(
+                                                "Encoded JNI struct field is missing its version marker");
+                                    }
+                                    // The encoded producer removes every grammar delimiter from
+                                    // names before parsing, then restores the exact UTF-8 spelling.
+                                    fieldName = new String(Base64.getDecoder().decode(fieldName.substring(1)),
+                                            StandardCharsets.UTF_8);
+                                } else {
+                                    // The public legacy grammar historically normalized STRUCT keys
+                                    // to lowercase; only the versioned Paimon payload preserves spelling.
+                                    fieldName = fieldName.toLowerCase();
+                                }
+                                fields.add(parseType(fieldName,
+                                        listFields.substring(pivot + 1, index), encodedStructFields));
+                                names.add(fieldName);
                                 listFields = listFields.substring(Math.min(index + 1, listFields.length()));
                             } else {
                                 break;

fe/be-java-extensions/java-common/src/test/java/org/apache/doris/common/jni/JniScannerTest.java

@@ -19,15 +19,27 @@
 
 
 import org.apache.doris.common.jni.utils.OffHeap;
+import org.apache.doris.common.jni.vec.ColumnType;
 import org.apache.doris.common.jni.vec.VectorTable;
 
 import org.junit.Assert;
 import org.junit.Test;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.HashMap;
 
 public class JniScannerTest {
+    @Test
+    public void testUnencodedStructFieldNamesRemainLowerCase() {
+        ColumnType structType = ColumnType.parseType(
+                "value", "struct<Mixed:int,UPPER:struct<Nested:string>>");
+
+        Assert.assertEquals(Arrays.asList("mixed", "upper"), structType.getChildNames());
+        Assert.assertEquals(Arrays.asList("nested"),
+                structType.getChildTypes().get(1).getChildNames());
+    }
+
     @Test
     public void testMockJniScanner() throws IOException {
         OffHeap.setTesting();