[feature](paimon) Add Paimon 2.0 transactional writes

be/src/common/config.cpp

@@ -1608,6 +1608,11 @@ DEFINE_mInt64(hive_sink_max_file_size, "1073741824"); // 1GB
 /** Iceberg sink configurations **/
 DEFINE_mInt64(iceberg_sink_max_file_size, "1073741824"); // 1GB
 
+/** Paimon sink configurations **/
+DEFINE_mInt64(paimon_jni_writer_memory_pool_limit_bytes, "536870912"); // 512MB
+DEFINE_Validator(paimon_jni_writer_memory_pool_limit_bytes,
+                 [](int64_t bytes) -> bool { return bytes > 0; });
+
 // URI scheme to Doris file type mappings used by paimon-cpp DorisFileSystem.
 // Each entry uses the format "<scheme>=<file_type>", and file_type must be one of:
 // local, hdfs, s3, http, broker.

be/src/common/config.h

@@ -1701,6 +1701,10 @@ DECLARE_mInt64(hive_sink_max_file_size);
 /** Iceberg sink configurations **/
 DECLARE_mInt64(iceberg_sink_max_file_size);
 
+/** Paimon sink configurations **/
+// Hard upper bound for Doris-managed Paimon write-buffer memory per JNI writer.
+DECLARE_mInt64(paimon_jni_writer_memory_pool_limit_bytes);
+
 /** Paimon file system configurations **/
 DECLARE_Strings(paimon_file_system_scheme_mappings);
 

be/src/core/column/column_variant.cpp

@@ -188,7 +188,7 @@ DataTypeSerDeSPtr ColumnVariant::Subcolumn::generate_data_serdes(DataTypePtr typ
 
 ColumnVariant::Subcolumn::Subcolumn(MutableColumnPtr&& data_, DataTypePtr type, bool is_nullable_,
                                     bool is_root_)
-        : least_common_type(type),
+        : least_common_type(type, is_root_),
           is_nullable(is_nullable_),
           is_root(is_root_),
           num_rows(data_->size()) {
@@ -463,6 +463,11 @@ void ColumnVariant::Subcolumn::insert_range_from(const Subcolumn& src, size_t st
         size_t part_end = end - processed_rows;
         insert_from_part(src.data[pos], src.data_types[pos], 0, part_end);
     }
+
+    const size_t trailing_defaults_start = std::max(start, src.num_rows);
+    if (end > trailing_defaults_start) {
+        data.back()->insert_many_defaults(end - trailing_defaults_start);
+    }
 }
 
 bool ColumnVariant::Subcolumn::is_finalized() const {
@@ -488,20 +493,29 @@ MutableColumnPtr ColumnVariant::apply_for_columns(Func&& func) const {
         auto& finalized_object = assert_cast<ColumnVariant&>(*finalized);
         return finalized_object.apply_for_columns(std::forward<Func>(func));
     }
-    auto new_root = std::move(*func(get_root())).mutate();
-    auto res = ColumnVariant::create(_max_subcolumns_count, _enable_doc_mode, get_root_type(),
-                                     std::move(new_root));
+    Subcolumns transformed_subcolumns;
     for (const auto& subcolumn : subcolumns) {
-        if (subcolumn->data.is_root) {
+        auto transformed = std::move(*func(subcolumn->data.get_finalized_column_ptr())).mutate();
+        Subcolumn transformed_subcolumn(std::move(transformed),
+                                        subcolumn->data.get_least_common_type(), is_nullable,
+                                        subcolumn->data.is_root);
+        if (subcolumn->data.is_root || subcolumn->path.empty()) {
+            transformed_subcolumns.create_root(std::move(transformed_subcolumn));
             continue;
         }
-        auto new_subcolumn = func(subcolumn->data.get_finalized_column_ptr());
-        if (!res->add_sub_column(subcolumn->path, std::move(*new_subcolumn).mutate(),
-                                 subcolumn->data.get_least_common_type())) {
+        if (!transformed_subcolumns.add(subcolumn->path, std::move(transformed_subcolumn))) {
             throw doris::Exception(ErrorCode::INTERNAL_ERROR, "add path {} is error",
                                    subcolumn->path.get_path());
         }
     }
+    if (transformed_subcolumns.get_root() == nullptr) {
+        throw doris::Exception(ErrorCode::INTERNAL_ERROR,
+                               "root is nullptr after transforming variant columns");
+    }
+    auto res = ColumnVariant::create(_max_subcolumns_count, _enable_doc_mode,
+                                     std::move(transformed_subcolumns));
+    res->typed_path_count = typed_path_count;
+    res->nested_path_count = nested_path_count;
     auto sparse_column = func(serialized_sparse_column);
     res->serialized_sparse_column = IColumn::mutate(std::move(sparse_column));
     auto doc_value_column = func(serialized_doc_value_column);
@@ -869,21 +883,10 @@ void ColumnVariant::insert_from(const IColumn& src, size_t n) {
     const auto* src_v = assert_cast<const ColumnVariant*>(&src);
     ENABLE_CHECK_CONSISTENCY(src_v);
     ENABLE_CHECK_CONSISTENCY(this);
-    // Preserve the original root-only copy path for ordinary variant columns.
-    // Reconstructing through try_insert() loses sparse/doc_value structure for
-    // mixed-shape rows and nested-group data.
-    if (src_v->get_subcolumns().size() == 1 && get_subcolumns().size() == 1) {
-        DCHECK(_enable_doc_mode == src_v->_enable_doc_mode)
-                << "root-only variant copy requires matching doc mode";
-        FieldWithDataType field;
-        src_v->subcolumns.get_root()->data.get(n, field);
-        subcolumns.get_mutable_root()->data.insert(field);
-        serialized_sparse_column->insert_from(*src_v->get_sparse_column(), n);
-        serialized_doc_value_column->insert_from(*src_v->get_doc_value_column(), n);
-        num_rows++;
-    } else {
-        try_insert((*src_v)[n]);
-    }
+    // Keep complex and materialized object paths in their native subcolumns. Reconstructing a
+    // single row as a Field can turn nested objects into TYPE_STRUCT, which is not a scalar type
+    // that Subcolumn::insert() can create through DataTypeFactory.
+    insert_range_from(*src_v, n, 1);
     ENABLE_CHECK_CONSISTENCY(this);
 }
 

be/src/core/data_type_serde/data_type_variant_serde.cpp

@@ -18,6 +18,7 @@
 #include "core/data_type_serde/data_type_variant_serde.h"
 
 #include <arrow/array/builder_binary.h>
+#include <arrow/array/builder_nested.h>
 
 #include <cstdint>
 #include <string>
@@ -35,6 +36,7 @@
 #include "core/types.h"
 #include "core/value/jsonb_value.h"
 #include "exec/common/variant_util.h"
+#include "exprs/function/parse/variant_jsonb_parse.h"
 #include "util/json/json_parser.h"
 #include "util/jsonb_writer.h"
 
@@ -63,6 +65,77 @@ Status write_variant_column_to_arrow_impl(const IColumn& column, const ColumnVar
     return Status::OK();
 }
 
+Status write_variant_column_to_arrow_struct(const IColumn& column, const ColumnVariant& var,
+                                            const NullMap* null_map, arrow::StructBuilder& builder,
+                                            int64_t start, int64_t end,
+                                            const cctz::time_zone& ctz) {
+    const auto struct_type = std::dynamic_pointer_cast<arrow::StructType>(builder.type());
+    if (struct_type == nullptr || builder.num_fields() != 2 ||
+        struct_type->field(0)->name() != "value" || struct_type->field(1)->name() != "metadata") {
+        return Status::InvalidArgument(
+                "Variant Arrow output requires "
+                "struct<value: binary, metadata: binary>");
+    }
+    auto* value_builder = dynamic_cast<arrow::BinaryBuilder*>(builder.field_builder(0));
+    auto* metadata_builder = dynamic_cast<arrow::BinaryBuilder*>(builder.field_builder(1));
+    if (value_builder == nullptr || metadata_builder == nullptr) {
+        return Status::InvalidArgument(
+                "Variant Arrow output requires binary value and metadata children");
+    }
+
+    const auto* root = var.get_subcolumn(PathInData());
+    const bool string_root =
+            root != nullptr && is_string_type(root->get_least_common_base_type_id());
+    JsonbToVariantEncoder encoder(
+            VariantBatchBuilder::ReserveHint {.rows = cast_set<size_t>(end - start)});
+    DataTypeSerDe::FormatOptions options;
+    options.timezone = &ctz;
+    for (int64_t row = start; row < end; ++row) {
+        if (null_map != nullptr && (*null_map)[cast_set<size_t>(row)]) {
+            encoder.add_null();
+            continue;
+        }
+
+        std::string serialized_value;
+        var.serialize_one_row_to_string(row, &serialized_value, options);
+        if (string_root && !root->is_null_at(cast_set<size_t>(row))) {
+            JsonbWriter writer;
+            if (!writer.writeStartString() ||
+                (!serialized_value.empty() &&
+                 !writer.writeString(serialized_value.data(), serialized_value.size())) ||
+                !writer.writeEndString()) {
+                return Status::InternalError("Failed to encode legacy Variant string as JSONB");
+            }
+            encoder.add_jsonb({writer.getOutput()->getBuffer(),
+                               static_cast<size_t>(writer.getOutput()->getSize())});
+            continue;
+        }
+
+        JsonBinaryValue jsonb;
+        RETURN_IF_ERROR(jsonb.from_json_string(serialized_value));
+        encoder.add_jsonb({jsonb.value(), jsonb.size()});
+    }
+
+    VariantBatchBuilder batch = encoder.finish_batch();
+    for (size_t row = 0; row < batch.num_rows(); ++row) {
+        const size_t source_row = cast_set<size_t>(start) + row;
+        if (null_map != nullptr && (*null_map)[source_row]) {
+            RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column, builder));
+            continue;
+        }
+        const VariantRef value = batch.value_at(row);
+        RETURN_IF_ERROR(checkArrowStatus(builder.Append(), column, builder));
+        RETURN_IF_ERROR(checkArrowStatus(
+                value_builder->Append(value.value.data, cast_set<int32_t>(value.value.size)),
+                column, *value_builder));
+        RETURN_IF_ERROR(
+                checkArrowStatus(metadata_builder->Append(value.metadata.data,
+                                                          cast_set<int32_t>(value.metadata.size)),
+                                 column, *metadata_builder));
+    }
+    return Status::OK();
+}
+
 } // namespace
 
 Status DataTypeVariantSerDe::write_column_to_mysql_binary(const IColumn& column,
@@ -161,6 +234,10 @@ Status DataTypeVariantSerDe::write_column_to_arrow(const IColumn& column, const
     } else if (array_builder->type()->id() == arrow::Type::STRING) {
         auto& builder = assert_cast<arrow::StringBuilder&>(*array_builder);
         return write_variant_column_to_arrow_impl(column, *var, null_map, builder, start, end, ctz);
+    } else if (array_builder->type()->id() == arrow::Type::STRUCT) {
+        auto& builder = assert_cast<arrow::StructBuilder&>(*array_builder);
+        RETURN_IF_CATCH_EXCEPTION(return write_variant_column_to_arrow_struct(
+                column, *var, null_map, builder, start, end, ctz));
     } else {
         return Status::InvalidArgument("Unsupported arrow type for variant column: {}",
                                        array_builder->type()->name());
@@ -208,7 +285,8 @@ Status DataTypeVariantSerDe::write_column_to_orc(const std::string& timezone, co
         size_t len = serialized_value.length();
         if (offset + len > total_size) {
             return Status::InternalError(
-                    "Buffer overflow when writing column data to ORC file. offset {} with len {} "
+                    "Buffer overflow when writing column data "
+                    "to ORC file. offset {} with len {} "
                     "exceed total_size {} . ",
                     offset, len, total_size);
         }

be/src/core/data_type_serde/data_type_variant_v2_serde.cpp

@@ -18,6 +18,7 @@
 #include "core/data_type_serde/data_type_variant_v2_serde.h"
 
 #include <arrow/array/builder_binary.h>
+#include <arrow/array/builder_nested.h>
 
 #include <algorithm>
 #include <cstring>
@@ -491,6 +492,50 @@ Status write_arrow(const IColumn& column, const NullMap* null_map, Builder& buil
     return status;
 }
 
+Status write_arrow_variant(const IColumn& column, const NullMap* null_map,
+                           arrow::StructBuilder& builder, size_t start, size_t end) {
+    const auto struct_type = std::dynamic_pointer_cast<arrow::StructType>(builder.type());
+    if (struct_type == nullptr || builder.num_fields() != 2 ||
+        struct_type->field(0)->name() != "value" || struct_type->field(1)->name() != "metadata") {
+        return Status::InvalidArgument(
+                "Variant Arrow output requires struct<value: binary, metadata: binary>");
+    }
+    auto* value_builder = dynamic_cast<arrow::BinaryBuilder*>(builder.field_builder(0));
+    auto* metadata_builder = dynamic_cast<arrow::BinaryBuilder*>(builder.field_builder(1));
+    if (value_builder == nullptr || metadata_builder == nullptr) {
+        return Status::InvalidArgument(
+                "Variant Arrow output requires binary value and metadata children");
+    }
+
+    Status status = Status::OK();
+    visit_variant_v2_values(
+            column, start, end, forced_nulls(null_map),
+            [&](size_t) {
+                if (status.ok()) {
+                    status = checkArrowStatus(builder.AppendNull(), column, builder);
+                }
+            },
+            [&](size_t, VariantRef value) {
+                if (!status.ok()) {
+                    return;
+                }
+                status = checkArrowStatus(builder.Append(), column, builder);
+                if (status.ok()) {
+                    status = checkArrowStatus(
+                            value_builder->Append(value.value.data,
+                                                  cast_set<int32_t>(value.value.size)),
+                            column, *value_builder);
+                }
+                if (status.ok()) {
+                    status = checkArrowStatus(
+                            metadata_builder->Append(value.metadata.data,
+                                                     cast_set<int32_t>(value.metadata.size)),
+                            column, *metadata_builder);
+                }
+            });
+    return status;
+}
+
 } // namespace
 
 void DataTypeVariantV2SerDe::to_string(const IColumn& column, size_t row_num, BufferWritable& bw,
@@ -553,6 +598,11 @@ Status DataTypeVariantV2SerDe::write_column_to_arrow(const IColumn& column, cons
                                assert_cast<arrow::LargeStringBuilder&>(*array_builder), first, last,
                                options);
         }
+        if (array_builder->type()->id() == arrow::Type::STRUCT) {
+            return write_arrow_variant(column, null_map,
+                                       assert_cast<arrow::StructBuilder&>(*array_builder), first,
+                                       last);
+        }
         return Status::InvalidArgument("Unsupported arrow type for variant column: {}",
                                        array_builder->type()->name());
     });

be/src/exec/operator/operator.cpp

@@ -64,6 +64,7 @@
 #include "exec/operator/olap_scan_operator.h"
 #include "exec/operator/olap_table_sink_operator.h"
 #include "exec/operator/olap_table_sink_v2_operator.h"
+#include "exec/operator/paimon_table_sink_operator.h"
 #include "exec/operator/partition_sort_sink_operator.h"
 #include "exec/operator/partition_sort_source_operator.h"
 #include "exec/operator/partitioned_aggregation_sink_operator.h"
@@ -830,6 +831,7 @@ DECLARE_OPERATOR(OlapTableSinkV2LocalState)
 DECLARE_OPERATOR(HiveTableSinkLocalState)
 DECLARE_OPERATOR(TVFTableSinkLocalState)
 DECLARE_OPERATOR(IcebergTableSinkLocalState)
+DECLARE_OPERATOR(PaimonTableSinkLocalState)
 DECLARE_OPERATOR(SpillIcebergTableSinkLocalState)
 DECLARE_OPERATOR(IcebergDeleteSinkLocalState)
 DECLARE_OPERATOR(IcebergMergeSinkLocalState)
@@ -955,6 +957,7 @@ template class AsyncWriterSink<doris::VTabletWriter, OlapTableSinkOperatorX>;
 template class AsyncWriterSink<doris::VTabletWriterV2, OlapTableSinkV2OperatorX>;
 template class AsyncWriterSink<doris::VHiveTableWriter, HiveTableSinkOperatorX>;
 template class AsyncWriterSink<doris::VIcebergTableWriter, IcebergTableSinkOperatorX>;
+template class AsyncWriterSink<doris::PaimonTableWriter, PaimonTableSinkOperatorX>;
 template class AsyncWriterSink<doris::VIcebergTableWriter, SpillIcebergTableSinkOperatorX>;
 template class AsyncWriterSink<doris::VIcebergDeleteSink, IcebergDeleteSinkOperatorX>;
 template class AsyncWriterSink<doris::VIcebergMergeSink, IcebergMergeSinkOperatorX>;

be/src/exec/operator/paimon_table_sink_operator.cpp

@@ -0,0 +1,39 @@
+// 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 "exec/operator/paimon_table_sink_operator.h"
+
+#include "common/logging.h"
+
+namespace doris {
+
+Status PaimonTableSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) {
+    return Base::init(state, info);
+}
+
+Status PaimonTableSinkOperatorX::sink_impl(RuntimeState* state, Block* in_block, bool eos) {
+    auto& local_state = get_local_state(state);
+    SCOPED_TIMER(local_state.exec_time_counter());
+    COUNTER_UPDATE(local_state.rows_input_counter(), static_cast<int64_t>(in_block->rows()));
+
+    // Delegate to AsyncWriterSink → PaimonTableWriter for this pipeline instance.
+    // Each pipeline instance has its own writer session; partition and bucket
+    // routing is handled internally by the Paimon SDK inside IPaimonWriter::write().
+    return local_state.sink(state, in_block, eos);
+}
+
+} // namespace doris