[feat](fluss) Support reading Apache Fluss tables through a fluss catalog

be/src/exec/scan/file_scanner_v2.cpp

@@ -50,10 +50,12 @@
 #include "format/format_common.h"
 #include "format/table/iceberg_scan_semantics.h"
 #include "format_v2/column_mapper.h"
+#include "format_v2/jni/fluss_jni_reader.h"
 #include "format_v2/jni/iceberg_sys_table_reader.h"
 #include "format_v2/jni/jdbc_reader.h"
 #include "format_v2/jni/max_compute_jni_reader.h"
 #include "format_v2/jni/trino_connector_jni_reader.h"
+#include "format_v2/table/fluss_union_lake_reader.h"
 #include "format_v2/table/hive_reader.h"
 #include "format_v2/table/hudi_reader.h"
 #include "format_v2/table/iceberg_position_delete_sys_table_reader.h"
@@ -99,7 +101,11 @@ bool is_supported_table_format(const TFileRangeDesc& range) {
         return false;
     }
     return table_format == "NotSet" || table_format == "tvf" || table_format == "hive" ||
-           table_format == "iceberg" || table_format == "paimon" || table_format == "hudi";
+           table_format == "iceberg" || table_format == "paimon" || table_format == "hudi" ||
+           // A lake split of a fluss primary-key table read as its lake plus its log tail. It is the
+           // paimon sibling's own split, wrapped, so it arrives in whichever form the sibling planned
+           // it - native Parquet/ORC here, a serialized JNI split below.
+           table_format == "fluss_union";
 }
 
 bool is_supported_arrow_table_format(const TFileRangeDesc& range) {
@@ -108,7 +114,9 @@ bool is_supported_arrow_table_format(const TFileRangeDesc& range) {
 
 bool is_supported_jni_table_format(const TFileRangeDesc& range) {
     const auto table_format = table_format_name(range);
-    if (table_format == "paimon") {
+    // A wrapped lake split is a paimon split in every respect that decides how it is read, so
+    // whether it can be read here is the paimon answer, asked of the paimon payload it carries.
+    if (table_format == "paimon" || table_format == "fluss_union") {
         if (!range.__isset.table_format_params ||
             !range.table_format_params.__isset.paimon_params) {
             return false;
@@ -131,7 +139,8 @@ bool is_supported_jni_table_format(const TFileRangeDesc& range) {
                (params.file_format == "parquet" || params.file_format == "orc");
     }
     return table_format == "jdbc" || table_format == "iceberg" || table_format == "hudi" ||
-           table_format == "max_compute" || table_format == "trino_connector";
+           table_format == "max_compute" || table_format == "trino_connector" ||
+           table_format == "fluss";
 }
 
 bool is_iceberg_position_deletes_sys_table(const TFileRangeDesc& range) {
@@ -403,6 +412,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 +512,24 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
         DORIS_CHECK(_table_reader != nullptr);
         _current_range_path = _current_range.path;
 
+        // The reader is chosen by the range's table format, not the node's, because one node can be
+        // given both: a fluss union read plans its lake half through the paimon connector and its log
+        // half itself, and both land here as ranges of the same scan. Built once from the first range,
+        // the reader would then be 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 fluss range that has no paimon parameters at all. 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.
+        auto table_format = table_format_name(_current_range);
+        if (table_format != _table_reader_format) {
+            RETURN_IF_ERROR(_create_table_reader_for_format(_current_range, &_table_reader));
+            DORIS_CHECK(_table_reader != nullptr);
+            _table_reader_format = std::move(table_format);
+            // 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) {
@@ -608,6 +636,10 @@ Status FileScannerV2::_create_table_reader_for_format(
         *reader = std::make_unique<format::paimon::PaimonHybridReader>();
     } else if (table_format == "hudi") {
         *reader = std::make_unique<format::hudi::HudiHybridReader>();
+    } else if (table_format == "fluss") {
+        *reader = std::make_unique<format::fluss::FlussJniReader>();
+    } else if (table_format == "fluss_union") {
+        *reader = std::make_unique<format::fluss::FlussUnionLakeReader>();
     } else if (table_format == "jdbc") {
         *reader = std::make_unique<format::jdbc::JdbcJniReader>();
     } else if (table_format == "max_compute") {

be/src/exec/scan/file_scanner_v2.h

@@ -181,6 +181,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/format_v2/jni/fluss_jni_reader.cpp

@@ -0,0 +1,124 @@
+// 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.
+
+#include "format_v2/jni/fluss_jni_reader.h"
+
+#include "core/block/block.h"
+#include "exprs/vexpr_context.h"
+#include "format_v2/column_mapper.h"
+
+namespace doris::format::fluss {
+
+Status FlussJniReader::validate_scan_range(const TFileRangeDesc& range) const {
+    if (!range.__isset.table_format_params) {
+        return Status::InternalError("missing table_format_params for fluss jni reader");
+    }
+    if (!range.table_format_params.__isset.fluss_params ||
+        range.table_format_params.fluss_params.empty()) {
+        return Status::InternalError(
+                "missing fluss_params for fluss jni reader, possibly caused by FE/BE protocol "
+                "mismatch");
+    }
+    if (_scan_params == nullptr || !_scan_params->__isset.fluss_properties ||
+        _scan_params->fluss_properties.empty()) {
+        return Status::InternalError(
+                "missing fluss_properties for fluss jni reader, possibly caused by FE/BE protocol "
+                "mismatch");
+    }
+    return Status::OK();
+}
+
+std::string FlussJniReader::connector_class() const {
+    return "org/apache/doris/fluss/FlussJniScanner";
+}
+
+Status FlussJniReader::build_scanner_params(std::map<std::string, std::string>* params) const {
+    DORIS_CHECK(params != nullptr);
+    DORIS_CHECK(_scan_params != nullptr);
+    // Scan level first, then the range: the range is the more specific of the two, and it is the one
+    // that says which bucket and offsets to read. No key is transcribed - both maps are written by FE
+    // and read by the Java scanner, so naming them here would only add a place to drift.
+    *params = _scan_params->fluss_properties;
+    for (const auto& kv : _current_range.table_format_params.fluss_params) {
+        (*params)[kv.first] = kv.second;
+    }
+    return Status::OK();
+}
+
+Status FlussJniReader::build_jni_columns(
+        std::vector<format::JniTableReader::JniColumn>* columns) const {
+    DORIS_CHECK(columns != nullptr);
+    columns->clear();
+    columns->reserve(_projected_columns.size());
+    for (size_t i = 0; i < _projected_columns.size(); ++i) {
+        const auto& table_column = _projected_columns[i];
+        // A fluss row physically carries its partition column values, but FE declares the partition
+        // keys as path_partition_keys and ships them per range, so they are constants here. Asking
+        // the scanner for them would read per row what the split already states once - and would
+        // read them differently from the legacy JNI path, whose file slots exclude them outright.
+        if (table_column.is_partition_key &&
+            find_partition_value(table_column, _partition_values) != nullptr) {
+            continue;
+        }
+        columns->push_back({
+                .java_name = table_column.name,
+                .output_index = i,
+                .output_type = table_column.type,
+                .transfer_type = table_column.type,
+                .replace_type = "not_replace",
+        });
+    }
+    return Status::OK();
+}
+
+Status FlussJniReader::finalize_jni_block(Block* jni_block, Block* output_block, size_t* rows) {
+    DORIS_CHECK(jni_block != nullptr);
+    DORIS_CHECK(output_block != nullptr);
+    DORIS_CHECK(rows != nullptr);
+    const auto original_rows = *rows;
+
+    const auto& columns = jni_columns();
+    DORIS_CHECK(columns.size() == jni_block->columns());
+    for (size_t i = 0; i < columns.size(); ++i) {
+        const auto& column = columns[i];
+        DORIS_CHECK(column.output_index < output_block->columns());
+        output_block->get_by_position(column.output_index).type = column.output_type;
+        output_block->replace_by_position(column.output_index,
+                                          jni_block->get_by_position(i).column);
+    }
+
+    // The columns build_jni_columns() left out: materialized from the range instead of read.
+    for (size_t i = 0; i < _projected_columns.size(); ++i) {
+        const auto& table_column = _projected_columns[i];
+        const auto* partition_value = find_partition_value(table_column, _partition_values);
+        if (!table_column.is_partition_key || partition_value == nullptr) {
+            continue;
+        }
+        output_block->get_by_position(i).type = table_column.type;
+        output_block->replace_by_position(
+                i, table_column.type->create_column_const(original_rows, *partition_value));
+    }
+    DORIS_CHECK(output_block->rows() == original_rows);
+    if (!_conjuncts.empty()) {
+        RETURN_IF_ERROR(
+                VExprContext::filter_block(_conjuncts, output_block, output_block->columns()));
+    }
+    *rows = output_block->rows();
+    return Status::OK();
+}
+
+} // namespace doris::format::fluss

be/src/format_v2/jni/fluss_jni_reader.h

@@ -0,0 +1,51 @@
+// 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.
+
+#pragma once
+
+#include <map>
+#include <string>
+#include <vector>
+
+#include "common/status.h"
+#include "format_v2/jni/jni_table_reader.h"
+#include "gen_cpp/PlanNodes_types.h"
+
+namespace doris::format::fluss {
+
+/**
+ * Reads one fluss scan range through org.apache.doris.fluss.FlussJniScanner.
+ *
+ * This layer holds no fluss logic. FE writes two untyped string maps - the scan-level
+ * `fluss_properties` (connection, table identity, client options) and the per-range `fluss_params`
+ * (which bucket, which offsets) - and the only decision made here is that the range wins where both
+ * set a key.
+ */
+class FlussJniReader final : public format::JniTableReader {
+public:
+    ~FlussJniReader() override = default;
+
+protected:
+    std::string connector_class() const override;
+    Status validate_scan_range(const TFileRangeDesc& range) const override;
+    Status build_scanner_params(std::map<std::string, std::string>* params) const override;
+    Status build_jni_columns(
+            std::vector<format::JniTableReader::JniColumn>* columns) const override;
+    Status finalize_jni_block(Block* jni_block, Block* output_block, size_t* rows) override;
+};
+
+} // namespace doris::format::fluss