[fix](paimon connector) Five independent fixes a sibling-connector read depends on

be/src/exec/scan/file_scanner_v2.cpp

@@ -403,6 +403,7 @@ Status FileScannerV2::_open_impl(RuntimeState* state) {
     if (_first_scan_range) {
         RETURN_IF_ERROR(_create_table_reader_for_format(_current_range, &_table_reader));
         DORIS_CHECK(_table_reader != nullptr);
+        _table_reader_format = table_format_name(_current_range);
         RETURN_IF_ERROR(_init_expr_ctxes());
         RETURN_IF_ERROR(_init_table_reader(_current_range));
     }
@@ -502,6 +503,14 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
         DORIS_CHECK(_table_reader != nullptr);
         _current_range_path = _current_range.path;
 
+        bool reader_rebuilt = false;
+        RETURN_IF_ERROR(_rebuild_table_reader_if_format_changed(_current_range, &reader_rebuilt));
+        if (reader_rebuilt) {
+            // Same init the first reader got. The expression contexts are NOT rebuilt: they are
+            // per-scanner and format-independent, and _init_expr_ctxes is not idempotent.
+            RETURN_IF_ERROR(_init_table_reader(_current_range));
+        }
+
         const auto format_type = get_range_format_type(*_params, _current_range);
         _init_adaptive_batch_size_state(format_type);
         if (_block_size_predictor != nullptr) {
@@ -583,6 +592,31 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) {
     return Status::OK();
 }
 
+Status FileScannerV2::_rebuild_table_reader_if_format_changed(const TFileRangeDesc& range,
+                                                              bool* rebuilt) {
+    // The reader is chosen by the range's table format, not the node's, because one node can be given
+    // both: a connector that reads a table as a lake plus the log written after it plans its lake half
+    // through a sibling connector and its log half itself, and both land here as ranges of the same
+    // scan. Built once from the first range and never revisited, the reader is then handed a range of
+    // the other format -- which does not fail cleanly. It fails as whatever that reader makes of a
+    // foreign range, e.g. paimon's reporting an unsupported file format for a range that carries no
+    // paimon parameters at all. And which ranges share a scanner is up to the engine's assignment, so
+    // the same query succeeds or fails by how the ranges happened to be dealt out.
+    //
+    // Split out from _prepare_next_split so the decision can be tested on its own: re-initializing the
+    // new reader needs scan-wide state that choosing it does not, so that step stays with the caller.
+    auto table_format = table_format_name(range);
+    if (table_format == _table_reader_format) {
+        *rebuilt = false;
+        return Status::OK();
+    }
+    RETURN_IF_ERROR(_create_table_reader_for_format(range, &_table_reader));
+    DORIS_CHECK(_table_reader != nullptr);
+    _table_reader_format = std::move(table_format);
+    *rebuilt = true;
+    return Status::OK();
+}
+
 Status FileScannerV2::_create_table_reader_for_format(
         const TFileRangeDesc& range, std::unique_ptr<format::TableReader>* reader) const {
     DORIS_CHECK(reader != nullptr);

be/src/exec/scan/file_scanner_v2.h

@@ -129,6 +129,9 @@ class FileScannerV2 final : public Scanner {
     Status _init_table_reader(const TFileRangeDesc& range);
     Status _create_table_reader_for_format(const TFileRangeDesc& range,
                                            std::unique_ptr<format::TableReader>* reader) const;
+    // Replaces _table_reader when {@code range} carries a different table format than the one it was
+    // built for, reporting whether it did. See the definition for why the reader follows the range.
+    Status _rebuild_table_reader_if_format_changed(const TFileRangeDesc& range, bool* rebuilt);
     Status _prepare_table_reader_split(const TFileRangeDesc& range,
                                        std::map<std::string, Field> partition_values);
     static bool _should_skip_not_found(const Status& status, bool ignore_not_found);
@@ -181,6 +184,10 @@ class FileScannerV2 final : public Scanner {
     std::string _current_range_path;
 
     std::unique_ptr<format::TableReader> _table_reader;
+    // The table format _table_reader was built for. A scan node may mix table formats -- a fluss
+    // union read gives one node its lake half as paimon ranges and its log half as fluss ones -- and
+    // the reader is format-specific, so it is rebuilt whenever this stops matching the range.
+    std::string _table_reader_format;
     std::vector<format::ColumnDefinition> _projected_columns;
     // File formats without embedded schema, such as CSV, still need the FE slot descriptors in
     // file-column order. This mirrors old FileScanner::_file_slot_descs and is passed only to

be/src/service/CMakeLists.txt

@@ -49,6 +49,24 @@ if (${MAKE_TEST} STREQUAL "OFF" AND ${BUILD_BENCHMARK} STREQUAL "OFF")
     # This permits libraries loaded by dlopen to link to the symbols in the program.
     set_target_properties(doris_be PROPERTIES ENABLE_EXPORTS 1)
 
+    # ...but not the symbols of the RocksDB we link statically. Exporting those makes this
+    # executable the definition every later-loaded library binds to, and a JNI library that
+    # carries its own RocksDB then runs half on ours: the fluss scanner bundles frocksdbjni,
+    # whose librocksdbjni.so defines 2576 rocksdb symbols under names identical to ours but
+    # was built against the pre-C++11 libstdc++ string ABI. Objects laid out by one and used
+    # by the other yield a garbage length, an std::bad_alloc that escapes the JNI frame, and
+    # an aborted BE. Hiding this archive lets that library bind to its own copy.
+    #
+    # Scoped to the archive rather than dropping ENABLE_EXPORTS: what needs the exports is
+    # native UDFs (runtime/user_function_cache.cpp dlopens them), and those use the Doris UDF
+    # ABI, which has nothing to do with RocksDB. Crash stacks do not need it either -- they are
+    # symbolized from debug info, which is why they name even anonymous-namespace functions.
+    #
+    # The same library also duplicates zstd, lz4, snappy, bzip2 and zlib symbols. Those are C
+    # ABIs, stable across versions and layout-free, so they are left alone until something
+    # shows otherwise -- unlike RocksDB, whose C++ objects are what actually corrupt.
+    target_link_options(doris_be PRIVATE "-Wl,--exclude-libs,librocksdb.a")
+
     target_link_libraries(doris_be
         ${DORIS_LINK_LIBS}
     )

be/test/exec/scan/file_scanner_v2_test.cpp

@@ -475,6 +475,60 @@ TEST(FileScannerV2Test, JniCompatibilityShapesUseV2Scanner) {
     EXPECT_TRUE(FileScannerV2::is_supported(params, legacy_paimon_jni_range_without_reader_type()));
 }
 
+// Scenario: one scan node is given ranges of two different table formats, which is what a connector
+// reading a table as a lake plus the log written after it produces -- its lake half planned by a
+// sibling connector, its own half by itself. The reader is format-specific, so it has to follow the
+// RANGE. Built once from the first range, it is later handed a foreign one and fails as whatever that
+// reader makes of it, not as a clean error; and since which ranges share a scanner is the engine's
+// assignment, the same query then succeeds or fails by how the ranges happened to be dealt out.
+TEST(FileScannerV2Test, TheTableReaderIsRebuiltWhenARangeChangesTableFormat) {
+    RuntimeState state {TQueryOptions(), TQueryGlobals()};
+    RuntimeProfile profile("file_scanner_v2_reader_per_range");
+    TFileScanRangeParams params;
+    params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+
+    FileScannerV2 scanner(&state, &profile, nullptr);
+    scanner._params = &params;
+
+    const auto paimon_range = range_with_format("paimon", TFileFormatType::FORMAT_PARQUET);
+    const auto hive_range = range_with_format("hive", TFileFormatType::FORMAT_PARQUET);
+
+    // Nothing has been built yet, so the first range always builds.
+    bool rebuilt = false;
+    ASSERT_TRUE(scanner._rebuild_table_reader_if_format_changed(paimon_range, &rebuilt).ok());
+    EXPECT_TRUE(rebuilt);
+    EXPECT_EQ(scanner._table_reader_format, "paimon");
+    const auto* first_reader = scanner._table_reader.get();
+    ASSERT_NE(first_reader, nullptr);
+
+    // A second range of the same format reuses it. Rebuilding here would be wasteful rather than
+    // wrong, but it would also throw away per-reader state the next split expects to still be there.
+    ASSERT_TRUE(scanner._rebuild_table_reader_if_format_changed(paimon_range, &rebuilt).ok());
+    EXPECT_FALSE(rebuilt);
+    EXPECT_EQ(scanner._table_reader.get(), first_reader);
+
+    // A range of another format must not be handed to the reader built for the first one.
+    ASSERT_TRUE(scanner._rebuild_table_reader_if_format_changed(hive_range, &rebuilt).ok());
+    EXPECT_TRUE(rebuilt);
+    EXPECT_EQ(scanner._table_reader_format, "hive");
+    EXPECT_NE(scanner._table_reader.get(), first_reader);
+
+    // And back again, because the ranges of a mixed node arrive interleaved rather than grouped.
+    ASSERT_TRUE(scanner._rebuild_table_reader_if_format_changed(paimon_range, &rebuilt).ok());
+    EXPECT_TRUE(rebuilt);
+    EXPECT_EQ(scanner._table_reader_format, "paimon");
+
+    // The formats really do get different readers -- otherwise every assertion above would hold
+    // just as well for a scanner that never rebuilt anything.
+    std::unique_ptr<format::TableReader> as_paimon;
+    std::unique_ptr<format::TableReader> as_hive;
+    ASSERT_TRUE(scanner._create_table_reader_for_format(paimon_range, &as_paimon).ok());
+    ASSERT_TRUE(scanner._create_table_reader_for_format(hive_range, &as_hive).ok());
+    const format::TableReader& paimon_reader = *as_paimon;
+    const format::TableReader& hive_reader = *as_hive;
+    EXPECT_STRNE(typeid(paimon_reader).name(), typeid(hive_reader).name());
+}
+
 TEST(FileScannerV2Test, FailedTableReaderCloseCanBeRetriedThroughScanner) {
     RuntimeState state {TQueryOptions(), TQueryGlobals()};
     RuntimeProfile profile("file_scanner_v2_close_retry");

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

@@ -30,6 +30,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalLong;
+import java.util.Set;
 
 /**
  * Plans the set of scan ranges (splits) needed to read a connector table.
@@ -148,6 +149,42 @@ default TFileCompressType adjustFileCompressType(TFileCompressType inferred) {
         return inferred;
     }
 
+    /**
+     * The columns BE must READ for this scan even when the query references none of them, by Doris-side
+     * column name. The engine keeps their slots in the scan's tuple instead of pruning them away; the
+     * projection above the scan still removes them from the query's output, so the answer changes what is
+     * read, never what is returned.
+     *
+     * <p>This exists for a connector whose BE-side reader needs a column to produce CORRECT ROWS rather than
+     * to answer the query — a merge key, a suppression key, a row identity. Doris does the same thing for its
+     * own aggregate / merge-on-read unique-key tables ({@code PhysicalPlanTranslator.preserveExtraStorageKeySlots}):
+     * BE merges by key whether or not the user selected the key. Trino has no counterpart because its
+     * connectors own the page source and can add such columns privately; here the reader is BE, so the columns
+     * have to reach it through the plan.</p>
+     *
+     * <p>Answer per SCAN, not per table: a connector that only sometimes needs the column (e.g. only when it
+     * decides to combine two sources) must return it only for those scans, and must reach the SAME decision
+     * when it later plans the splits — the engine asks this during plan translation, strictly before
+     * {@link #planScan}. Memoize that decision on the provider instance (the engine keeps one per scan node)
+     * rather than deciding twice: two independent decisions can disagree, and then BE is asked to read a
+     * column the tuple does not carry.</p>
+     *
+     * <p>Every name returned must be a column of the scanned table, spelled as Doris knows it (the same
+     * identifier-mapped name {@link #classifyColumn} receives). A name that matches no slot in the scan's
+     * tuple fails the query loud: it means the connector and the engine disagree about the table, and reading
+     * on would silently produce whatever the connector's reader does without that column.</p>
+     *
+     * <p>The default returns an empty set — every connector whose reader needs nothing beyond the projection
+     * is untouched, and its scans prune exactly as before.</p>
+     *
+     * @param session the current session
+     * @param handle  the table handle being scanned
+     * @return Doris-side names of the columns to read regardless of the projection (default: empty)
+     */
+    default Set<String> getMustReadColumns(ConnectorSession session, ConnectorTableHandle handle) {
+        return Collections.emptySet();
+    }
+
     /**
      * Plans the scan described by {@code request}, returning the ranges that cover the requested data.
      *

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

@@ -0,0 +1,79 @@
+// 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.scan;
+
+import org.apache.doris.connector.api.ConnectorSession;
+import org.apache.doris.connector.api.handle.ConnectorTableHandle;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Guards the additive {@code getMustReadColumns} SPI default on {@link ConnectorScanPlanProvider}.
+ *
+ * <p>WHY: the engine consults this on EVERY plugin-table scan that has a projection above it, and widens the
+ * scan's tuple by whatever comes back. The default must therefore be empty, or every connector that never
+ * asked for anything would start reading extra columns — and, worse, would fail the query loud when a name
+ * it never returned matches no slot. This is the zero-break guard for es/jdbc/paimon/iceberg/hive/maxcompute,
+ * none of which override it.</p>
+ */
+public class ConnectorScanPlanProviderMustReadColumnsTest {
+
+    /** Bare provider: only the abstract planScan implemented; everything else inherits SPI defaults. */
+    private static final class BareProvider implements ConnectorScanPlanProvider {
+        @Override
+        public List<ConnectorScanRange> planScan(ConnectorSession session, ConnectorScanRequest request) {
+            return Collections.emptyList();
+        }
+    }
+
+    /** A connector whose BE-side reader needs a merge key the query may not have selected. */
+    private static final class KeyReadingProvider implements ConnectorScanPlanProvider {
+        @Override
+        public List<ConnectorScanRange> planScan(ConnectorSession session, ConnectorScanRequest request) {
+            return Collections.emptyList();
+        }
+
+        @Override
+        public Set<String> getMustReadColumns(ConnectorSession session, ConnectorTableHandle handle) {
+            return Collections.singleton("id");
+        }
+    }
+
+    @Test
+    public void defaultAsksForNoExtraColumns() {
+        ConnectorScanPlanProvider provider = new BareProvider();
+
+        // MUTATION: a default returning anything non-empty would widen every connector's scans and fail
+        // loud on the first name that matches no slot -> red here first.
+        Assertions.assertEquals(Collections.emptySet(), provider.getMustReadColumns(null, null),
+                "a connector that never opted in must ask for no extra columns");
+    }
+
+    @Test
+    public void connectorThatOptsInIsObeyed() {
+        ConnectorScanPlanProvider provider = new KeyReadingProvider();
+
+        Assertions.assertEquals(Collections.singleton("id"), provider.getMustReadColumns(null, null),
+                "the engine must read back exactly what the connector asked for");
+    }
+}

fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java

@@ -23,6 +23,7 @@
 import org.apache.doris.connector.api.ConnectorPartitionInfo;
 import org.apache.doris.connector.api.ConnectorSession;
 import org.apache.doris.connector.api.ConnectorValidationContext;
+import org.apache.doris.connector.api.handle.ConnectorTableHandle;
 import org.apache.doris.connector.api.scan.ConnectorScanPlanProvider;
 import org.apache.doris.connector.cache.ConnectorMetadataCache;
 import org.apache.doris.connector.metastore.HmsMetaStoreProperties;
@@ -254,6 +255,22 @@ public ConnectorMetadata getMetadata(ConnectorSession session) {
                 properties, context, schemaAtMemo, latestSnapshotCache, partitionViewCache);
     }
 
+    /**
+     * True for a handle this connector produced (a {@link PaimonTableHandle}). Tested against this connector's
+     * OWN in-loader type, so a gateway connector that embeds this one as a sibling can route a foreign paimon
+     * handle here without casting it across the plugin classloader split. Returns false for any other
+     * connector's handle, so the gateway keeps looking.
+     *
+     * <p>The default is {@code false}, which for a sibling means every one of the gateway's guards silently
+     * fails open and the first cast throws a ClassCastException instead — so this is required of any connector
+     * used as a sibling, not an optimization. Same implementation as the iceberg and hudi siblings behind the
+     * hms gateway.
+     */
+    @Override
+    public boolean ownsHandle(ConnectorTableHandle handle) {
+        return handle instanceof PaimonTableHandle;
+    }
+
     @Override
     public void invalidateTable(String dbName, String tableName) {
         // REFRESH TABLE (and, via the generic PluginDrivenExternalCatalog DDL hook, a Doris-issued