[feature](paimon) Forward-port Paimon table writes to master

be/src/common/config.cpp

@@ -1719,6 +1719,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

@@ -1785,6 +1785,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/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>
@@ -182,6 +183,133 @@ void preflight_json(const IColumn& column, size_t start, size_t end,
             });
 }
 
+void validate_paimon_variant_primitive(VariantPrimitiveId primitive_id) {
+    switch (primitive_id) {
+    case VariantPrimitiveId::NULL_VALUE:
+    case VariantPrimitiveId::TRUE_VALUE:
+    case VariantPrimitiveId::FALSE_VALUE:
+    case VariantPrimitiveId::INT8:
+    case VariantPrimitiveId::INT16:
+    case VariantPrimitiveId::INT32:
+    case VariantPrimitiveId::INT64:
+    case VariantPrimitiveId::DOUBLE:
+    case VariantPrimitiveId::DECIMAL4:
+    case VariantPrimitiveId::DECIMAL8:
+    case VariantPrimitiveId::DECIMAL16:
+    case VariantPrimitiveId::DATE:
+    case VariantPrimitiveId::TIMESTAMP_MICROS:
+    case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS:
+    case VariantPrimitiveId::FLOAT:
+    case VariantPrimitiveId::BINARY:
+    case VariantPrimitiveId::STRING:
+    case VariantPrimitiveId::UUID:
+        return;
+    case VariantPrimitiveId::TIME_NTZ_MICROS:
+    case VariantPrimitiveId::TIMESTAMP_NANOS:
+    case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS:
+        throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+                        "Paimon does not support Variant primitive id {}",
+                        static_cast<uint8_t>(primitive_id));
+    }
+    throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+                    "Paimon does not support unknown Variant primitive id {}",
+                    static_cast<uint8_t>(primitive_id));
+}
+
+void validate_paimon_variant_value(VariantRef value, uint32_t depth = 0) {
+    if (depth > VARIANT_MAX_NESTING_DEPTH) {
+        throw Exception(ErrorCode::CORRUPTION, "Variant value exceeds maximum nesting depth {}",
+                        VARIANT_MAX_NESTING_DEPTH);
+    }
+    const size_t encoded_size = value.value_size();
+    if (encoded_size != value.value.size) {
+        throw Exception(ErrorCode::CORRUPTION,
+                        "Variant value has {} trailing bytes after the encoded value",
+                        value.value.size - encoded_size);
+    }
+
+    switch (value.basic_type()) {
+    case VariantBasicType::PRIMITIVE:
+        validate_paimon_variant_primitive(value.primitive_id());
+        return;
+    case VariantBasicType::SHORT_STRING:
+        return;
+    case VariantBasicType::OBJECT:
+        for (uint32_t i = 0; i < value.num_elements(); ++i) {
+            uint32_t field_id = 0;
+            VariantRef child = value.object_value_at(i, &field_id);
+            value.metadata.key_at(field_id);
+            validate_paimon_variant_value(child, depth + 1);
+        }
+        return;
+    case VariantBasicType::ARRAY:
+        for (uint32_t i = 0; i < value.num_elements(); ++i) {
+            validate_paimon_variant_value(value.array_at(i), depth + 1);
+        }
+        return;
+    }
+}
+
+void require_variant_arrow_status(const arrow::Status& status) {
+    if (!status.ok()) {
+        throw Exception(ErrorCode::INTERNAL_ERROR, "Variant V2 Arrow append failed: {}",
+                        status.ToString());
+    }
+}
+
+Status write_binary_variant_arrow(const IColumn& column, const NullMap* null_map,
+                                  arrow::StructBuilder& builder, size_t start, size_t end) {
+    // StructBuilder::type() returns a shared_ptr by value. Keep that owner alive while using the
+    // cast reference; otherwise the reference would dangle as soon as the temporary is destroyed.
+    const auto builder_type = builder.type();
+    const auto& struct_type = assert_cast<const arrow::StructType&>(*builder_type);
+    if (struct_type.num_fields() != 2 || struct_type.field(0)->name() != "value" ||
+        struct_type.field(1)->name() != "metadata" ||
+        struct_type.field(0)->type()->id() != arrow::Type::BINARY ||
+        struct_type.field(1)->type()->id() != arrow::Type::BINARY) {
+        return Status::InvalidArgument(
+                "Binary Variant V2 Arrow type must be "
+                "struct<value: binary, metadata: binary>, got {}",
+                struct_type.ToString());
+    }
+    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("Binary Variant V2 Arrow child builders must be binary");
+    }
+
+    // GenericVariant assumes its input is valid, and Paimon's unshredded writer copies these two
+    // buffers without inspecting them. Validate once at the Doris-to-Paimon boundary so a write
+    // cannot commit bytes which Paimon is unable to read later.
+    const auto outer_nulls = forced_nulls(null_map);
+    visit_variant_v2_values(
+            column, start, end, outer_nulls,
+            [&](size_t) { require_variant_arrow_status(builder.AppendNull()); },
+            [&](size_t row, VariantRef value) {
+                try {
+                    constexpr size_t PAIMON_VARIANT_SIZE_LIMIT = 128 * 1024 * 1024;
+                    if (value.value.size > PAIMON_VARIANT_SIZE_LIMIT ||
+                        value.metadata.size > PAIMON_VARIANT_SIZE_LIMIT) {
+                        throw Exception(ErrorCode::INVALID_ARGUMENT,
+                                        "exceeds the 128 MiB value/metadata limit");
+                    }
+                    value.metadata.validate();
+                    validate_paimon_variant_value(value);
+                } catch (const Exception& e) {
+                    throw Exception(e.code(), "Paimon Variant V2 row {} is incompatible: {}", row,
+                                    e.what());
+                }
+                require_variant_arrow_status(builder.Append());
+                require_variant_arrow_status(
+                        value_builder->Append(reinterpret_cast<const uint8_t*>(value.value.data),
+                                              cast_set<int32_t, size_t, false>(value.value.size)));
+                require_variant_arrow_status(metadata_builder->Append(
+                        reinterpret_cast<const uint8_t*>(value.metadata.data),
+                        cast_set<int32_t, size_t, false>(value.metadata.size)));
+            });
+    return Status::OK();
+}
+
 } // namespace
 
 DataTypeVariantV2SerDe::DataTypeVariantV2SerDe(int nesting_level) : DataTypeSerDe(nesting_level) {}
@@ -553,6 +681,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_binary_variant_arrow(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/exchange_sink_operator.cpp

@@ -28,16 +28,17 @@
 #include <random>
 #include <string>
 
+#include "agent/be_exec_version_manager.h"
 #include "common/status.h"
 #include "core/column/column_const.h"
 #include "exec/exchange/exchange_writer.h"
 #include "exec/exchange/local_exchange_sink_operator.h"
 #include "exec/operator/exchange_sink_buffer.h"
 #include "exec/operator/operator.h"
 #include "exec/operator/sort_source_operator.h"
+#include "exec/partitioner/external/external_table_sink_hash_partitioner.h"
 #include "exec/pipeline/dependency.h"
 #include "exec/pipeline/pipeline_fragment_context.h"
-#include "exec/sink/scale_writer_partitioning_exchanger.hpp"
 #include "exec/sink/tablet_sink_hash_partitioner.h"
 #include "exprs/vexpr.h"
 #include "format/transformer/merge_partitioner.h"
@@ -98,7 +99,7 @@ Status ExchangeSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& inf
     _part_type = p._part_type;
     // Shuffle the channels randomly
     if (_part_type == TPartitionType::UNPARTITIONED || _part_type == TPartitionType::RANDOM ||
-        _part_type == TPartitionType::HIVE_TABLE_SINK_UNPARTITIONED) {
+        _part_type == TPartitionType::EXTERNAL_TABLE_SINK_UNPARTITIONED) {
         std::random_device rd;
         std::mt19937 g(rd());
         shuffle(channels.begin(), channels.end(), g);
@@ -153,28 +154,32 @@ Status ExchangeSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& inf
                 p._tablet_sink_partition, p._tablet_sink_location, p._tablet_sink_tuple_id, this);
         RETURN_IF_ERROR(_partitioner->init({}));
         RETURN_IF_ERROR(_partitioner->prepare(state, {}));
-    } else if (_part_type == TPartitionType::HIVE_TABLE_SINK_HASH_PARTITIONED) {
-        _partition_count =
-                channels.size() * config::table_sink_partition_write_max_partition_nums_per_writer;
-        _partitioner = std::make_unique<ScaleWriterPartitioner>(
-                channels.size(), _partition_count, channels.size(), 1,
-                config::table_sink_partition_write_min_partition_data_processed_rebalance_threshold /
-                                        state->task_num() ==
-                                0
-                        ? config::table_sink_partition_write_min_partition_data_processed_rebalance_threshold
-                        : config::table_sink_partition_write_min_partition_data_processed_rebalance_threshold /
-                                  state->task_num(),
-                config::table_sink_partition_write_min_data_processed_rebalance_threshold /
-                                        state->task_num() ==
-                                0
-                        ? config::table_sink_partition_write_min_data_processed_rebalance_threshold
-                        : config::table_sink_partition_write_min_data_processed_rebalance_threshold /
-                                  state->task_num());
-
+    } else if (_part_type == TPartitionType::EXTERNAL_TABLE_SINK_HASH_PARTITIONED) {
+        if (state->be_exec_version() < SUPPORT_EXTERNAL_TABLE_SINK_HASH_VERSION) {
+            return Status::NotSupported(
+                    "External table sink hash exchange requires BE execution version {}, actual {}",
+                    SUPPORT_EXTERNAL_TABLE_SINK_HASH_VERSION, state->be_exec_version());
+        }
+        if (!p._has_external_table_sink_hash_partition_info) {
+            return Status::InternalError("External table sink hash partition info is missing");
+        }
+        _partition_count = channels.size();
+        const bool use_crc32c = _state->query_options().__isset.enable_new_shuffle_hash_method &&
+                                _state->query_options().enable_new_shuffle_hash_method;
+        const ShuffleHashMethod hash_method =
+                use_crc32c ? ShuffleHashMethod::CRC32C : ShuffleHashMethod::CRC32;
+        _partitioner = std::make_unique<ExternalTableSinkHashPartitioner>(
+                _partition_count, hash_method, p._external_table_sink_hash_partition_info);
         RETURN_IF_ERROR(_partitioner->init(p._texprs));
         RETURN_IF_ERROR(_partitioner->prepare(state, p._row_desc));
         custom_profile()->add_info_string(
-                "Partitioner", fmt::format("ScaleWriterPartitioner({})", _partition_count));
+                "Partitioner",
+                fmt::format("ExternalTableSinkHashPartitioner({})", _partition_count));
+        custom_profile()->add_info_string(
+                "WriterAssignment", p._external_table_sink_hash_partition_info.writer_assignment ==
+                                                    TExternalTableSinkWriterAssignment::IDENTITY
+                                            ? "IDENTITY"
+                                            : "SKEWED");
     } else if (_part_type == TPartitionType::MERGE_PARTITIONED) {
         if (!p._has_merge_partition_info) {
             return Status::InternalError("Merge partition info is missing");
@@ -271,7 +276,7 @@ Status ExchangeSinkLocalState::open(RuntimeState* state) {
 
     if (_part_type == TPartitionType::HASH_PARTITIONED ||
         _part_type == TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED ||
-        _part_type == TPartitionType::HIVE_TABLE_SINK_HASH_PARTITIONED ||
+        _part_type == TPartitionType::EXTERNAL_TABLE_SINK_HASH_PARTITIONED ||
         _part_type == TPartitionType::OLAP_TABLE_SINK_HASH_PARTITIONED ||
         _part_type == TPartitionType::MERGE_PARTITIONED) {
         RETURN_IF_ERROR(_partitioner->open(state));
@@ -320,8 +325,8 @@ ExchangeSinkOperatorX::ExchangeSinkOperatorX(
            sink.output_partition.type == TPartitionType::RANGE_PARTITIONED ||
            sink.output_partition.type == TPartitionType::OLAP_TABLE_SINK_HASH_PARTITIONED ||
            sink.output_partition.type == TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED ||
-           sink.output_partition.type == TPartitionType::HIVE_TABLE_SINK_HASH_PARTITIONED ||
-           sink.output_partition.type == TPartitionType::HIVE_TABLE_SINK_UNPARTITIONED ||
+           sink.output_partition.type == TPartitionType::EXTERNAL_TABLE_SINK_HASH_PARTITIONED ||
+           sink.output_partition.type == TPartitionType::EXTERNAL_TABLE_SINK_UNPARTITIONED ||
            sink.output_partition.type == TPartitionType::MERGE_PARTITIONED);
 #endif
     _name = "ExchangeSinkOperatorX";
@@ -333,6 +338,11 @@ ExchangeSinkOperatorX::ExchangeSinkOperatorX(
         _merge_partition_info = sink.output_partition.merge_partition_info;
         _has_merge_partition_info = true;
     }
+    if (sink.output_partition.__isset.external_table_sink_hash_partition_info) {
+        _external_table_sink_hash_partition_info =
+                sink.output_partition.external_table_sink_hash_partition_info;
+        _has_external_table_sink_hash_partition_info = true;
+    }
 
     if (_part_type != TPartitionType::UNPARTITIONED) {
         // if the destinations only one dest, we need to use broadcast
@@ -533,11 +543,11 @@ Status ExchangeSinkOperatorX::sink_impl(RuntimeState* state, Block* block, bool
                 (local_state.current_channel_idx + 1) % local_state.channels.size();
     } else if (_part_type == TPartitionType::HASH_PARTITIONED ||
                _part_type == TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED ||
+               _part_type == TPartitionType::EXTERNAL_TABLE_SINK_HASH_PARTITIONED ||
                _part_type == TPartitionType::OLAP_TABLE_SINK_HASH_PARTITIONED ||
-               _part_type == TPartitionType::HIVE_TABLE_SINK_HASH_PARTITIONED ||
                _part_type == TPartitionType::MERGE_PARTITIONED) {
         RETURN_IF_ERROR(local_state._writer->write(state, block, eos));
-    } else if (_part_type == TPartitionType::HIVE_TABLE_SINK_UNPARTITIONED) {
+    } else if (_part_type == TPartitionType::EXTERNAL_TABLE_SINK_UNPARTITIONED) {
         // Control the number of channels according to the flow, thereby controlling the number of table sink writers.
         RETURN_IF_ERROR(send_to_current_channel());
         _data_processed += block->bytes();

be/src/exec/operator/exchange_sink_operator.h

@@ -243,6 +243,8 @@ class ExchangeSinkOperatorX MOCK_REMOVE(final) : public DataSinkOperatorX<Exchan
     RuntimeState* _state = nullptr;
 
     const std::vector<TExpr> _texprs;
+    TExternalTableSinkHashPartitionInfo _external_table_sink_hash_partition_info;
+    bool _has_external_table_sink_hash_partition_info = false;
     TMergePartitionInfo _merge_partition_info;
     bool _has_merge_partition_info = false;
 

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"
@@ -846,6 +847,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)