[PR] Feature/support paimon insert 4.1 [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt opened a new pull request, #62689:
URL: https://github.com/apache/doris/pull/62689

   ### What problem does this PR solve? 
    
   Issue Number: close #56005 
    
   Related PR: #xxx 
    
   Problem Summary: 
    
   This PR implements Doris write-back support for Paimon external tables (`INSERT INTO ...`), completing the end-to-end write path across FE and BE, including planning, sink serialization, execution, and FE-side commit. 
    
   #### 1) Core capability 
   - Support writing data from Doris into Paimon external tables. 
   - Support both writer paths: 
     - **Native Writer (paimon-cpp)**: High-performance C++ native writer path (Default).
     - **JNI Writer**: Java-based writer path for complex scenarios and compatibility.
    
   #### 2) FE/BE write pipeline integration 
   - **FE**: 
     - Add Paimon insert executor flow. 
     - Build and send Paimon sink metadata (table identity/location/options/column names/shuffle info). 
     - Collect commit payload from BE and commit on FE side. 
   - **BE**: 
     - Execute Paimon sink writing. 
     - Build commit payload and report back to FE. 
    
   #### 3) Bucket/partition write behavior 
   - Add distributed bucket shuffle support for bucketed Paimon tables. 
   - Add compatible fallback behavior (e.g. gather/random) when required bucket keys are not resolvable. 
   - Keep partition-related write behavior aligned with existing sink framework. 
    
   #### 4) Limitations of Native Writer (paimon-cpp)
   Currently, the high-performance Native Writer is the default path, but it has some functional limitations compared to the JNI Writer. When these scenarios are encountered, users must fallback to the JNI Writer (`set enable_paimon_jni_writer = true`):
   - **Dynamic Bucket (`bucket = -1`)**: Not supported. Native Writer currently only supports fixed bucket tables.
   - **Inline Compaction**: Not supported. Native Writer operates in write-only mode without compaction capabilities during the write process.
   - **Memory Spill**: Not supported. Native Writer lacks disk spill logic when memory buffers are exhausted.
   - **File Formats**: Strictly limited to `parquet` for both data files and manifest files.
   
   #### 5) Session variables and write tunables 
   This PR introduces/uses Paimon-related session controls for write behavior and performance tuning. 
   
   **Execution Path & Shuffle Control**
   - `enable_paimon_jni_writer` (Default: `false`): Switch to JNI writer path instead of native paimon-cpp.
   - `enable_paimon_distributed_bucket_shuffle` (Default: `true`): Enable distributed bucket shuffle for Paimon tables.
   
   **Memory & Buffer Tuning**
   - `paimon_global_memory_pool_size` (Default: `1073741824L` / 1GB): Global memory pool size for Paimon writer.
   - `paimon_write_buffer_size` (Default: `268435456L` / 256MB): Memory buffer size per writer.
   - `enable_paimon_adaptive_buffer_size` (Default: `false`): Enable adaptive adjustment of buffer sizes.
   - `paimon_writer_queue_size` (Default: `3`): Queue size for writer tasks.
   - `paimon_target_file_size` (Default: `268435456L` / 256MB): Target size for generated data files.
   
   **JNI Spill & Compaction Control (Only applicable when `enable_paimon_jni_writer=true`)**
   - `enable_paimon_jni_compact` (Default: `true`): Enable inline compaction during JNI writing.
   - `enable_paimon_jni_spill` (Default: `false`): Enable memory spilling to disk.
   - `paimon_spill_max_disk_size` (Default: `53687091200L` / 50GB): Maximum disk space allowed for spilling.
   - `paimon_spill_sort_buffer_size` (Default: `67108864L` / 64MB): Buffer size for sorting during spill.
   - `paimon_spill_sort_threshold` (Default: `10`): Threshold triggering sort operation.
   - `paimon_spill_compression` (Default: `"zstd"`): Compression codec used for spilled files.
    
   #### 6) Compatibility/alignment work 
   - Align merged code with current community branch conventions and dependencies. 
   - Resolve compile/checkstyle conflicts introduced by branch differences. 
   - Ensure FE build passes under Maven >= 3.9 environment.
   
   
   #### 7) Current recommendation for production usage
   - **Prefer JNI Writer in production for now**: it is currently the safer path for large-scale write workloads.
   - **Enable spill for large data volume**: for big batches, it is recommended to set `enable_paimon_jni_spill = true` to reduce memory pressure and avoid excessive memory usage.
   - **Native Writer roadmap dependency**: Native Writer can be the default high-performance path later, but full production readiness for heavy workloads still depends on paimon-cpp upstream support for **compaction** and **spill**.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4327474333

   # BE UT Coverage Report
   Increment line coverage `25.25% (531/2103)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/a2b279283bda7583cb2c741263107f2611c38b5e_a2b279283bda7583cb2c741263107f2611c38b5e/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/a2b279283bda7583cb2c741263107f2611c38b5e_a2b279283bda7583cb2c741263107f2611c38b5e/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 53.23% (20097/37752) |
   | Line Coverage     | 36.68% (189493/516668) |
   | Region Coverage   | 33.02% (147294/446039) |
   | Branch Coverage   | 34.09% (64450/189073) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3202183253


##########
be/src/vec/sink/vpaimon_table_writer.cpp:
##########
@@ -0,0 +1,740 @@
+// 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 "vec/sink/vpaimon_table_writer.h"
+
+#include <gen_cpp/DataSinks_types.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "common/metrics/doris_metrics.h"
+#include "core/block/block.h"
+#include "core/column/column.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "runtime/query_context.h"
+#include "runtime/runtime_profile.h"
+#include "runtime/runtime_state.h"
+#include "util/defer_op.h"
+#include "vec/sink/paimon_writer_utils.h"
+#include "vec/sink/writer/paimon/paimon_doris_hdfs_file_system.h"
+#include "vec/sink/writer/paimon/vpaimon_partition_writer.h"
+
+#ifdef WITH_PAIMON_CPP
+#include <arrow/array.h>
+#include <arrow/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/type.h>
+
+#include <cstdint>
+
+#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
+#include "format/parquet/arrow_memory_pool.h"
+#include "io/fs/hdfs_file_system.h"
+#include "io/hdfs_builder.h"
+#include "paimon/commit_message.h"
+#include "paimon/factories/factory_creator.h"
+#include "paimon/file_store_write.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/metrics.h"
+#include "paimon/utils/bucket_id_calculator.h"
+#include "paimon/write_context.h"
+#include "vec/sink/writer/paimon/paimon_doris_memory_pool.h"
+
+// Force link paimon file format factories
+namespace paimon {
+namespace parquet {}
+} // namespace paimon
+
+#endif
+
+namespace doris {
+namespace vectorized {
+
+#ifdef WITH_PAIMON_CPP
+namespace {
+bool is_paimon_cpp_time_metric(std::string_view name) {
+    return name.size() > 3 && name.substr(name.size() - 3) == "_ns";
+}
+
+void attach_paimon_cpp_metrics_to_profile(RuntimeProfile* profile,
+                                          const std::shared_ptr<::paimon::Metrics>& metrics) {
+    if (profile == nullptr || !metrics) {
+        return;
+    }
+    auto all = metrics->GetAllCounters();
+    if (all.empty()) {
+        return;
+    }
+    for (const auto& kv : all) {
+        std::string counter_name = "PaimonCpp_" + kv.first;
+        std::replace(counter_name.begin(), counter_name.end(), '.', '_');
+        RuntimeProfile::Counter* counter = nullptr;
+        if (is_paimon_cpp_time_metric(kv.first)) {
+            counter = ADD_COUNTER(profile, counter_name, TUnit::TIME_NS);
+        } else {
+            counter = ADD_COUNTER(profile, counter_name, TUnit::UNIT);
+        }
+        COUNTER_UPDATE(counter, kv.second);
+    }
+}
+} // namespace
+#endif
+
+VPaimonTableWriter::~VPaimonTableWriter() = default;
+
+VPaimonTableWriter::VPaimonTableWriter(const TDataSink& t_sink,
+                                       const VExprContextSPtrs& output_exprs)
+        : VPaimonTableWriter(t_sink, output_exprs, nullptr, nullptr) {}
+
+VPaimonTableWriter::VPaimonTableWriter(const TDataSink& t_sink,
+                                       const VExprContextSPtrs& output_exprs,
+                                       std::shared_ptr<Dependency> dep,
+                                       std::shared_ptr<Dependency> fin_dep)
+        : AsyncResultWriter(output_exprs, std::move(dep), std::move(fin_dep)), _t_sink(t_sink) {
+    DCHECK(_t_sink.__isset.paimon_table_sink);
+}
+
+Status VPaimonTableWriter::init_properties(ObjectPool* /*pool*/) {
+    // Currently there is no extra property to initialize. Kept for symmetry
+    // with VIcebergTableWriter and future paimon-cpp wiring.
+    return Status::OK();
+}
+
+Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
+    _state = state;
+    _profile = profile;
+#ifndef WITH_PAIMON_CPP
+    return Status::NotSupported("paimon-cpp is not enabled");
+#else
+    _written_rows_counter = ADD_COUNTER(_profile, "WrittenRows", TUnit::UNIT);
+    _written_bytes_counter = ADD_COUNTER(_profile, "WrittenBytes", TUnit::BYTES);
+    _send_data_timer = ADD_TIMER(_profile, "SendDataTime");
+    _project_timer = ADD_CHILD_TIMER(_profile, "ProjectTime", "SendDataTime");
+    _bucket_calc_timer = ADD_CHILD_TIMER(_profile, "BucketCalcTime", "SendDataTime");
+    _partition_writers_dispatch_timer =
+            ADD_CHILD_TIMER(_profile, "PartitionsDispatchTime", "SendDataTime");
+    _partition_writers_write_timer =
+            ADD_CHILD_TIMER(_profile, "PartitionsWriteTime", "SendDataTime");
+    _partition_writers_count = ADD_COUNTER(_profile, "PartitionsWriteCount", TUnit::UNIT);
+    _partition_writer_created = ADD_COUNTER(_profile, "PartitionWriterCreated", TUnit::UNIT);
+    _open_timer = ADD_TIMER(_profile, "OpenTime");
+    _close_timer = ADD_TIMER(_profile, "CloseTime");
+    _prepare_commit_timer = ADD_TIMER(_profile, "PrepareCommitTime");
+    _serialize_commit_messages_timer = ADD_TIMER(_profile, "SerializeCommitMessagesTime");
+    _commit_payload_bytes_counter = ADD_COUNTER(_profile, "CommitPayloadBytes", TUnit::BYTES);
+
+    SCOPED_TIMER(_open_timer);
+
+    ensure_paimon_doris_hdfs_file_system_registered();
+
+    auto registered_types = paimon::FactoryCreator::GetInstance()->GetRegisteredType();
+    std::string types_str;
+    bool has_parquet = false;
+    for (const auto& t : registered_types) {
+        types_str += t + ", ";
+        has_parquet |= (t == "parquet");
+    }
+    if (!has_parquet) {
+        return Status::InternalError(
+                "paimon-cpp parquet file format factory is not registered (missing 'parquet' in "
+                "FactoryCreator). Please ensure BE is built with WITH_PAIMON_CPP=ON and linked "
+                "with libpaimon_parquet_file_format.a (whole-archive). Registered factories: {}",
+                types_str);
+    }
+
+    _pool = std::make_shared<PaimonDorisMemoryPool>(_state->query_mem_tracker());
+    const auto& paimon_sink = _t_sink.paimon_table_sink;
+    if (!paimon_sink.__isset.table_location || paimon_sink.table_location.empty()) {
+        return Status::InvalidArgument("paimon table location is empty");
+    }
+    std::string commit_user;
+    if (paimon_sink.__isset.options) {
+        auto it = paimon_sink.options.find("doris.commit_user");
+        if (it != paimon_sink.options.end()) {
+            commit_user = it->second;
+        }
+    }
+    if (commit_user.empty()) {
+        commit_user = _state->user();
+    }
+
+    std::map<std::string, std::string> options;
+    if (paimon_sink.__isset.options) {
+        for (const auto& kv : paimon_sink.options) {
+            if (kv.first.rfind("doris.", 0) == 0) {
+                continue;
+            }
+            options.emplace(kv.first, kv.second);
+        }
+    }
+
+    // Workaround for paimon-cpp issue where it defaults to LocalFileSystem if path has no scheme.
+    // If table_location is missing scheme (common in HDFS setup without full URI),
+    // and fs.defaultFS is provided in options, we prepend it.
+    std::string table_location = paimon_sink.table_location;
+    if (table_location.find("://") == std::string::npos) {
+        auto it = options.find("fs.defaultFS");
+        if (it != options.end()) {
+            std::string default_fs = it->second;
+            // Remove trailing slash from default_fs if present
+            while (!default_fs.empty() && default_fs.back() == '/') {
+                default_fs.pop_back();
+            }
+            // Remove leading slash from table_location if present
+            if (!table_location.empty() && table_location.front() == '/') {
+                table_location = default_fs + table_location;
+            } else {
+                table_location = default_fs + "/" + table_location;
+            }
+        }
+    }
+
+    int64_t buffer_size = 256 * 1024 * 1024L; // Default 256MB
+
+    if (_state->query_options().__isset.paimon_write_buffer_size &&
+        _state->query_options().paimon_write_buffer_size > 0) {
+        buffer_size = _state->query_options().paimon_write_buffer_size;
+    }
+
+    bool enable_adaptive = true;
+    if (_state->query_options().__isset.enable_paimon_adaptive_buffer_size) {
+        enable_adaptive = _state->query_options().enable_paimon_adaptive_buffer_size;
+    }
+
+    if (enable_adaptive && paimon_sink.__isset.bucket_num && paimon_sink.bucket_num > 0) {
+        int bucket_num = paimon_sink.bucket_num;
+        buffer_size = get_paimon_write_buffer_size(buffer_size, true, bucket_num);
+        LOG(INFO) << "Adaptive Paimon Buffer Size: bucket_num=" << bucket_num
+                  << ", adjusted_buffer_size=" << buffer_size;
+    }
+    LOG(INFO) << "Paimon Native Writer Final Buffer Size: " << buffer_size
+              << " (enable_adaptive=" << enable_adaptive << ")";
+    options["write-buffer-size"] = std::to_string(buffer_size);
+
+    if (_state->query_options().__isset.paimon_target_file_size &&
+        _state->query_options().paimon_target_file_size > 0) {
+        options["target-file-size"] =
+                std::to_string(_state->query_options().paimon_target_file_size);
+        LOG(INFO) << "Paimon Native Writer Target File Size: "
+                  << _state->query_options().paimon_target_file_size;
+    }
+
+    options["file.format"] = "parquet";
+    options["manifest.format"] = "parquet";
+    if ((!paimon_sink.__isset.bucket_num || paimon_sink.bucket_num <= 0) &&
+        paimon_sink.__isset.bucket_keys && !paimon_sink.bucket_keys.empty()) {
+        return Status::NotSupported(
+                "paimon-cpp native writer does not support primary-key table with dynamic bucket "
+                "(bucket=-1) yet; enable_paimon_jni_writer=true is required");
+    }
+
+    ::paimon::WriteContextBuilder builder(table_location, commit_user);
+    builder.SetOptions(options);
+    builder.WithIgnorePreviousFiles(true);
+    builder.WithMemoryPool(_pool);
+    builder.WithFileSystemSchemeToIdentifierMap(
+            {{"hdfs", kPaimonDorisHdfsFsIdentifier}, {"dfs", kPaimonDorisHdfsFsIdentifier}});
+    auto ctx_result = builder.Finish();
+    if (!ctx_result.ok()) {
+        return Status::InternalError("failed to build paimon write context: {}",
+                                     ctx_result.status().ToString());
+    }
+    auto write_result = ::paimon::FileStoreWrite::Create(std::move(ctx_result).value());
+    if (!write_result.ok()) {
+        auto message = write_result.status().ToString();
+        if (message.find("format 'orc'") != std::string::npos ||
+            message.find("format \"orc\"") != std::string::npos) {
+            return Status::InternalError(
+                    "failed to create paimon file store write: {}. This Doris build does not "
+                    "support ORC in paimon-cpp; create paimon table with options "
+                    "file.format=parquet and manifest.format=parquet.",
+                    message);
+        }
+        return Status::InternalError("failed to create paimon file store write: {}", message);
+    }
+    _file_store_write = std::move(write_result).value();
+    return Status::OK();
+#endif
+}
+
+Status VPaimonTableWriter::_init_partition_column_indices(const ::doris::Block& block) const {
+    const TPaimonTableSink& paimon_sink = _t_sink.paimon_table_sink;
+    if (!paimon_sink.__isset.partition_keys || paimon_sink.partition_keys.empty() ||
+        _partition_indices_inited) {
+        return Status::OK();
+    }
+
+    std::unordered_map<std::string, int> name_to_idx;
+    for (int i = 0; i < block.columns(); ++i) {
+        std::string col_name = block.get_by_position(i).name;
+        if (col_name.empty()) {
+            if (paimon_sink.__isset.column_names && i < paimon_sink.column_names.size()) {
+                col_name = paimon_sink.column_names[i];
+            }
+        }
+        name_to_idx.emplace(col_name, i);
+    }
+    _partition_column_indices.clear();
+    _partition_column_indices.reserve(paimon_sink.partition_keys.size());
+    for (const auto& key_name : paimon_sink.partition_keys) {
+        auto it = name_to_idx.find(key_name);
+        if (it == name_to_idx.end()) {
+            return Status::InvalidArgument("paimon partition key {} not found in output block",
+                                           key_name);
+        }
+        _partition_column_indices.push_back(it->second);
+    }
+    _partition_indices_inited = true;
+    return Status::OK();
+}
+
+std::string VPaimonTableWriter::_default_partition_name() const {
+    std::string default_part = "__DEFAULT_PARTITION__";
+    const TPaimonTableSink& paimon_sink = _t_sink.paimon_table_sink;
+    if (paimon_sink.__isset.options) {
+        auto it = paimon_sink.options.find("partition.default-name");
+        if (it != paimon_sink.options.end() && !it->second.empty()) {
+            default_part = it->second;
+        }
+    }
+    return default_part;
+}
+
+Status VPaimonTableWriter::_collect_partition_value_columns(
+        const ::doris::Block& block,
+        std::vector<std::vector<std::string>>* partition_value_columns) const {
+    partition_value_columns->clear();
+    const TPaimonTableSink& paimon_sink = _t_sink.paimon_table_sink;
+    if (!paimon_sink.__isset.partition_keys || paimon_sink.partition_keys.empty()) {
+        return Status::OK();
+    }
+
+    RETURN_IF_ERROR(_init_partition_column_indices(block));
+    const size_t rows = block.rows();
+    partition_value_columns->resize(_partition_column_indices.size());
+    const std::string default_part = _default_partition_name();
+    DataTypeSerDe::FormatOptions options;
+    for (size_t part_idx = 0; part_idx < _partition_column_indices.size(); ++part_idx) {
+        auto& values = (*partition_value_columns)[part_idx];
+        values.resize(rows);
+        const auto& col_with_type = block.get_by_position(_partition_column_indices[part_idx]);
+        const auto& type = col_with_type.type;
+        const auto& col = col_with_type.column;
+        for (size_t row = 0; row < rows; ++row) {
+            if (col->is_null_at(row)) {
+                values[row] = default_part;
+            } else {
+                values[row] = type->to_string(*col, row, options);
+            }
+        }
+    }
+    return Status::OK();
+}
+
+Status VPaimonTableWriter::_get_or_create_writer(const WriteKey& key,
+                                                 std::shared_ptr<VPaimonPartitionWriter>* writer) {
+    auto it = _writers.find(key);
+    if (it != _writers.end()) {
+        *writer = it->second;
+        return Status::OK();
+    }
+
+    auto new_writer =
+            std::make_shared<VPaimonPartitionWriter>(_t_sink, key.partition_values, key.bucket_id
+#ifdef WITH_PAIMON_CPP
+                                                     ,
+                                                     _file_store_write.get(), _pool
+#endif
+            );
+    RETURN_IF_ERROR(new_writer->init_properties(nullptr));
+    RETURN_IF_ERROR(new_writer->open(_state, _profile));
+    COUNTER_UPDATE(_partition_writer_created, 1);
+    DorisMetrics::instance()->paimon_partition_writer_created->increment(1);
+    _writers.emplace(key, new_writer);
+    *writer = std::move(new_writer);
+    return Status::OK();
+}
+
+Status VPaimonTableWriter::write(RuntimeState* state, ::doris::Block& block) {
+#ifndef WITH_PAIMON_CPP
+    return Status::NotSupported("paimon-cpp is not enabled");
+#else
+    if (block.rows() == 0) {
+        return Status::OK();
+    }
+    SCOPED_TIMER(_send_data_timer);
+    int64_t send_data_ns = 0;
+    auto to_ms_ceil = [](int64_t ns) -> uint64_t {
+        if (ns <= 0) {
+            return 0;
+        }
+        return static_cast<uint64_t>((ns + 999999) / 1000000);
+    };
+    Defer record_send_data_latency {[&]() {
+        DorisMetrics::instance()->paimon_write_send_data_latency_ms->add(to_ms_ceil(send_data_ns));
+    }};
+    SCOPED_RAW_TIMER(&send_data_ns);
+
+    Block output_block;
+    int64_t project_ns = 0;
+    {
+        SCOPED_TIMER(_project_timer);
+        SCOPED_RAW_TIMER(&project_ns);
+        RETURN_IF_ERROR(_projection_block(block, &output_block));
+    }
+    _row_count += output_block.rows();

Review Comment:
   Fixed. The native paimon-cpp writer now updates RuntimeState::update_num_rows_load_total() and update_num_bytes_load_total() after the projected block is accepted, matching the JNI writer path so FE can report loaded rows/bytes correctly.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4438838184

   # Cloud UT Coverage Report
   Increment line coverage ` ` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/f1152d99acade5af865fc3f092e94845e4789508_f1152d99acade5af865fc3f092e94845e4789508_cloud/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/f1152d99acade5af865fc3f092e94845e4789508_f1152d99acade5af865fc3f092e94845e4789508_cloud/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 78.11% (1848/2366) |
   | Line Coverage     | 64.85% (33225/51232) |
   | Region Coverage   | 65.34% (16445/25167) |
   | Branch Coverage   | 55.88% (8779/15710) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["yiguolei (via GitHub)" <gi...@apache.org>]

yiguolei commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3216735484


##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2238,7 +2289,7 @@ Status PipelineFragmentContext::send_report(bool done) {
     // a internal cancellation being processed
     // When limit is reached the fragment is also cancelled, but _is_report_on_cancel will
     // be set to false, to avoid sending fault report to FE.
-    if (!_is_report_success && !_is_report_on_cancel) {

Review Comment:
   这里不要动，这里应该不感知什么writer 类型的



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4304507339

   # BE Regression && UT Coverage Report
   Increment line coverage `6.54% (136/2081)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_b82d48ce0c47b786660f4d4a313c04094ec00076_merge/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_b82d48ce0c47b786660f4d4a313c04094ec00076_merge/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 71.25% (26299/36912) |
   | Line Coverage     | 54.24% (278859/514120) |
   | Region Coverage   | 51.58% (231697/449183) |
   | Branch Coverage   | 52.90% (100101/189218) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4326386532

   # FE Regression Coverage Report
   Increment line coverage `  53.29% (308/578)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_a2b279283bda7583cb2c741263107f2611c38b5e_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_a2b279283bda7583cb2c741263107f2611c38b5e_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4440143187

   # FE Regression Coverage Report
   Increment line coverage `  7.04% (39/554)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_f1152d99acade5af865fc3f092e94845e4789508_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_f1152d99acade5af865fc3f092e94845e4789508_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4431241982

   run feut


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4437287409

   # Cloud UT Coverage Report
   Increment line coverage ` ` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/48970b4f14dc51e6915bbf349ac223e96b9ce48c_48970b4f14dc51e6915bbf349ac223e96b9ce48c_cloud/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/48970b4f14dc51e6915bbf349ac223e96b9ce48c_48970b4f14dc51e6915bbf349ac223e96b9ce48c_cloud/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 78.11% (1848/2366) |
   | Line Coverage     | 64.85% (33223/51232) |
   | Region Coverage   | 65.32% (16438/25167) |
   | Branch Coverage   | 55.95% (8790/15710) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] Feature/support paimon insert 4.1 [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4293044726

   
   Thank you for your contribution to Apache Doris.
   Don't know what should be done next? See [How to process your PR](https://cwiki.apache.org/confluence/display/DORIS/How+to+process+your+PR).
   
   Please clearly describe your PR:
   1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
   2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
   3. What features were added. Why was this function added?
   4. Which code was refactored and why was this part of the code refactored?
   5. Which functions were optimized and what is the difference before and after the optimization?
       


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4441544972

   # FE UT Coverage Report
   Increment line coverage `  50.27% (278/553)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_ebc64c58aeed7e11f3b7e6cba70d061461b1bf63/fe_increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_ebc64c58aeed7e11f3b7e6cba70d061461b1bf63/fe_report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["github-actions[bot] (via GitHub)" <gi...@apache.org>]

github-actions[bot] commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4417730173

   PR approved by at least one committer and no changes requested.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4448238558

   # FE Regression Coverage Report
   Increment line coverage `  8.32% (46/553)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_217a176440b5baddb646aef46d8f763aac735867_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_217a176440b5baddb646aef46d8f763aac735867_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4411983576

   # BE UT Coverage Report
   Increment line coverage `24.01% (531/2212)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/6a002645fe579dfc2e1cb2731413ad70493a1981_6a002645fe579dfc2e1cb2731413ad70493a1981/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/6a002645fe579dfc2e1cb2731413ad70493a1981_6a002645fe579dfc2e1cb2731413ad70493a1981/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 53.27% (20154/37835) |
   | Line Coverage     | 36.74% (189939/516961) |
   | Region Coverage   | 33.05% (147602/446599) |
   | Branch Coverage   | 34.12% (64593/189297) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4429075391

   # Cloud UT Coverage Report
   Increment line coverage ` ` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/48ebfc1ef324868ae6ae453e0a0b236c5b123f9a_48ebfc1ef324868ae6ae453e0a0b236c5b123f9a_cloud/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/48ebfc1ef324868ae6ae453e0a0b236c5b123f9a_48ebfc1ef324868ae6ae453e0a0b236c5b123f9a_cloud/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 78.11% (1848/2366) |
   | Line Coverage     | 64.85% (33223/51232) |
   | Region Coverage   | 65.36% (16449/25167) |
   | Branch Coverage   | 55.91% (8783/15710) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4327515557

   # BE Regression && UT Coverage Report
   Increment line coverage `26.11% (549/2103)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_a2b279283bda7583cb2c741263107f2611c38b5e_merge/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_a2b279283bda7583cb2c741263107f2611c38b5e_merge/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 71.46% (26412/36958) |
   | Line Coverage     | 54.44% (280314/514928) |
   | Region Coverage   | 51.68% (232539/449943) |
   | Branch Coverage   | 53.13% (100722/189579) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4385037951

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4406686474

   # BE Regression && UT Coverage Report
   Increment line coverage `25.36% (561/2212)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_09b7591b25385c938209a19f4825ba70b7f2a20f_merge/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_09b7591b25385c938209a19f4825ba70b7f2a20f_merge/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 71.57% (26502/37029) |
   | Line Coverage     | 54.56% (281023/515112) |
   | Region Coverage   | 51.68% (232821/450497) |
   | Branch Coverage   | 53.14% (100856/189807) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3226062916


##########
be/cmake/thirdparty.cmake:
##########
@@ -117,11 +117,12 @@ add_thirdparty(rdkafka_cpp LIBNAME "lib/librdkafka++.a")
 add_thirdparty(rdkafka)
 add_thirdparty(s2)
 add_thirdparty(bitshuffle)
-add_thirdparty(roaring)

Review Comment:
   感谢指出。我又检查了一下，还是分支checkout产生的问题；
   这里确实没有必要在这个 PR 里调整，所以已经把这部分改动去掉了，当前 be/cmake/thirdparty.cmake 不再有变更。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4437078446

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3226091191


##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -1877,6 +1886,16 @@ bool PipelineFragmentContext::_close_fragment_instance() {
     Defer defer_op {[&]() { _is_fragment_instance_closed = true; }};
     _fragment_level_profile->total_time_counter()->update(_fragment_watcher.elapsed_time());
     if (!_need_notify_close) {
+        if (_runtime_state) {
+            for (auto& tasks : _tasks) {
+                for (auto& task : tasks) {
+                    if (!task.first->runtime_state()->paimon_commit_messages().empty()) {
+                        _runtime_state->add_paimon_commit_messages(

Review Comment:
   感谢提醒，我对齐了一下已有 writer 的处理方式。这里不再遍历 _tasks 主动收集 Paimon commit message，避免在 close 流程里做特殊处理。现在 Paimon commit message 和 Iceberg/MaxCompute 一样，在 _coordinator_callback() 里从 RuntimeState 填充到 report status 参数中上报。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4430172796

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4430786308

   # FE UT Coverage Report
   Increment line coverage `  50.27% (278/553)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608/fe_increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608/fe_report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4448661534

   # Cloud UT Coverage Report
   Increment line coverage ` ` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/94845913004923f36a1e28983b2c859453552e6e_94845913004923f36a1e28983b2c859453552e6e_cloud/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/94845913004923f36a1e28983b2c859453552e6e_94845913004923f36a1e28983b2c859453552e6e_cloud/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 78.11% (1848/2366) |
   | Line Coverage     | 64.86% (33229/51232) |
   | Region Coverage   | 65.36% (16449/25167) |
   | Branch Coverage   | 55.94% (8788/15710) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4405975949

   run cloud_p0


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4415087587

   run p0


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4405078834

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4398126075

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4438598281

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3226079850


##########
be/src/pipeline/exec/paimon_table_sink_operator.cpp:
##########
@@ -0,0 +1,73 @@
+// 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 "pipeline/exec/paimon_table_sink_operator.h"
+
+#include "common/status.h"
+#include "vec/sink/vpaimon_jni_table_writer.h"
+#include "vec/sink/vpaimon_table_writer.h"
+
+namespace doris {
+
+bool should_use_paimon_jni_writer(const TDataSink& tsink) {
+    if (!tsink.__isset.paimon_table_sink) {
+        return false;
+    }
+    const auto& options = tsink.paimon_table_sink.options;
+    auto it = options.find("paimon_use_jni");
+    if (it == options.end()) {
+        return false;
+    }
+    return it->second == "true" || it->second == "1";

Review Comment:
   已经处理，这里现在会先 trim 配置值，并使用大小写不敏感的比较判断 true，同时保留对 1 的支持。当前代码已按 4.1 路径调整到 be/src/exec/operator/paimon_table_sink_operator.cpp。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4431243542

   run p0


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4431253540

   run NonConcurrent


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4398012569

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4403012259

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4411815276

   # Cloud UT Coverage Report
   Increment line coverage ` ` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/6a002645fe579dfc2e1cb2731413ad70493a1981_6a002645fe579dfc2e1cb2731413ad70493a1981_cloud/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/6a002645fe579dfc2e1cb2731413ad70493a1981_6a002645fe579dfc2e1cb2731413ad70493a1981_cloud/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 78.06% (1843/2361) |
   | Line Coverage     | 64.77% (33001/50952) |
   | Region Coverage   | 65.26% (16367/25078) |
   | Branch Coverage   | 55.86% (8740/15646) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["github-actions[bot] (via GitHub)" <gi...@apache.org>]

github-actions[bot] commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3198506451


##########
be/src/service/CMakeLists.txt:
##########
@@ -43,6 +43,7 @@ if (${MAKE_TEST} STREQUAL "OFF" AND ${BUILD_BENCHMARK} STREQUAL "OFF")
 
     target_link_libraries(doris_be
         ${DORIS_LINK_LIBS}

Review Comment:
   This links `paimon_rawlog_compat` even when `ENABLE_PAIMON_CPP` is off, while `be/CMakeLists.txt` already adds the same library to `DORIS_LINK_LIBS` when Paimon C++ is enabled. The compat object defines `google::RawLog__`, so linking it unconditionally bypasses the intended feature gate and can create duplicate/incorrect symbol resolution depending on whether glog or paimon-cpp also provides that symbol. Please keep this behind the same gate and avoid adding it twice.



##########
be/src/vec/sink/writer/paimon/paimon_doris_hdfs_file_system.cpp:
##########
@@ -0,0 +1,439 @@
+// 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 "vec/sink/writer/paimon/paimon_doris_hdfs_file_system.h"
+
+#ifdef WITH_PAIMON_CPP
+
+#include <gen_cpp/PlanNodes_types.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "common/status.h"
+#include "io/fs/file_reader.h"
+#include "io/fs/file_system.h"
+#include "io/fs/file_writer.h"
+#include "io/fs/hdfs_file_system.h"
+#include "io/fs/path.h"
+#include "io/hdfs_builder.h"
+#include "paimon/factories/factory_creator.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/fs/file_system_factory.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "util/slice.h"
+
+namespace {
+
+std::string _extract_hdfs_fs_name(const std::string& uri) {
+    auto starts_with = [&](const char* prefix) { return uri.rfind(prefix, 0) == 0; };
+    std::string_view scheme;
+    if (starts_with("hdfs://")) {
+        scheme = "hdfs://";
+    } else if (starts_with("dfs://")) {
+        scheme = "dfs://";
+    } else {
+        return {};
+    }
+    size_t authority_start = scheme.size();
+    size_t first_slash = uri.find('/', authority_start);
+    if (first_slash == std::string::npos) {
+        return uri;
+    }
+    if (first_slash == authority_start) {
+        return {};
+    }
+    return uri.substr(0, first_slash);
+}
+
+paimon::Status _to_paimon_status(const doris::Status& st) {
+    if (st.ok()) {
+        return paimon::Status::OK();
+    }
+    return paimon::Status::IOError(st.to_string());
+}
+
+class DorisPaimonBasicFileStatus final : public paimon::BasicFileStatus {
+public:
+    DorisPaimonBasicFileStatus(std::string path, bool is_dir)
+            : _path(std::move(path)), _is_dir(is_dir) {}
+
+    bool IsDir() const override { return _is_dir; }
+
+    std::string GetPath() const override { return _path; }
+
+private:
+    std::string _path;
+    bool _is_dir;
+};
+
+class DorisPaimonFileStatus final : public paimon::FileStatus {
+public:
+    DorisPaimonFileStatus(std::string path, bool is_dir, uint64_t len, int64_t mtime_ms)
+            : _path(std::move(path)), _is_dir(is_dir), _len(len), _mtime_ms(mtime_ms) {}
+
+    uint64_t GetLen() const override { return _len; }
+
+    bool IsDir() const override { return _is_dir; }
+
+    std::string GetPath() const override { return _path; }
+
+    int64_t GetModificationTime() const override { return _mtime_ms; }
+
+private:
+    std::string _path;
+    bool _is_dir;
+    uint64_t _len;
+    int64_t _mtime_ms;
+};
+
+class DorisPaimonInputStream final : public paimon::InputStream {
+public:
+    DorisPaimonInputStream(doris::io::FileReaderSPtr reader, std::string uri)
+            : _reader(std::move(reader)), _uri(std::move(uri)) {}
+
+    paimon::Status Close() override { return _to_paimon_status(_reader->close()); }
+
+    paimon::Status Seek(int64_t offset, paimon::SeekOrigin origin) override {
+        int64_t base = 0;
+        switch (origin) {
+        case paimon::FS_SEEK_SET:
+            base = 0;
+            break;
+        case paimon::FS_SEEK_CUR:
+            base = _pos;
+            break;
+        case paimon::FS_SEEK_END: {
+            auto len = Length();
+            if (!len.ok()) {
+                return len.status();
+            }
+            base = static_cast<int64_t>(len.value());
+            break;
+        }
+        default:
+            return paimon::Status::Invalid("invalid seek origin");
+        }
+        int64_t next = base + offset;
+        if (next < 0) {
+            return paimon::Status::Invalid("negative seek position: ", next);
+        }
+        _pos = next;
+        return paimon::Status::OK();
+    }
+
+    paimon::Result<int64_t> GetPos() const override { return _pos; }
+
+    paimon::Result<int32_t> Read(char* buffer, uint32_t size) override {
+        size_t bytes_read = 0;
+        doris::Slice slice(buffer, size);
+        doris::Status st = _reader->read_at(_pos, slice, &bytes_read, nullptr);
+        if (!st.ok()) {
+            return paimon::Status::IOError(st.to_string());
+        }
+        _pos += static_cast<int64_t>(bytes_read);
+        return static_cast<int32_t>(bytes_read);
+    }
+
+    paimon::Result<int32_t> Read(char* buffer, uint32_t size, uint64_t offset) override {
+        size_t bytes_read = 0;
+        doris::Slice slice(buffer, size);
+        doris::Status st = _reader->read_at(offset, slice, &bytes_read, nullptr);
+        if (!st.ok()) {
+            return paimon::Status::IOError(st.to_string());
+        }
+        return static_cast<int32_t>(bytes_read);
+    }
+
+    void ReadAsync(char* buffer, uint32_t size, uint64_t offset,
+                   std::function<void(paimon::Status)>&& callback) override {
+        auto res = Read(buffer, size, offset);
+        if (res.ok()) {
+            callback(paimon::Status::OK());
+        } else {
+            callback(res.status());
+        }
+    }
+
+    paimon::Result<std::string> GetUri() const override { return _uri; }
+
+    paimon::Result<uint64_t> Length() const override {
+        return static_cast<uint64_t>(_reader->size());
+    }
+
+private:
+    doris::io::FileReaderSPtr _reader;
+    std::string _uri;
+    int64_t _pos = 0;
+};
+
+class DorisPaimonOutputStream final : public paimon::OutputStream {
+public:
+    DorisPaimonOutputStream(doris::io::FileWriterPtr writer, std::string uri)
+            : _writer(std::move(writer)), _uri(std::move(uri)) {}
+
+    paimon::Status Close() override { return _to_paimon_status(_writer->close()); }
+
+    paimon::Result<int32_t> Write(const char* buffer, uint32_t size) override {
+        doris::Slice slice(buffer, size);
+        doris::Status st = _writer->append(slice);
+        if (!st.ok()) {
+            return paimon::Status::IOError(st.to_string());
+        }
+        _pos += size;
+        return static_cast<int32_t>(size);
+    }
+
+    paimon::Status Flush() override { return paimon::Status::OK(); }
+
+    paimon::Result<int64_t> GetPos() const override { return _pos; }
+
+    paimon::Result<std::string> GetUri() const override { return _uri; }
+
+private:
+    doris::io::FileWriterPtr _writer;
+    std::string _uri;
+    int64_t _pos = 0;
+};
+
+class DorisPaimonFileSystem final : public paimon::FileSystem {
+public:
+    explicit DorisPaimonFileSystem(std::shared_ptr<doris::io::FileSystem> fs)
+            : _fs(std::move(fs)) {}
+
+    paimon::Result<std::unique_ptr<paimon::InputStream>> Open(
+            const std::string& path) const override {
+        doris::io::FileReaderSPtr reader;
+        doris::Status st = _fs->open_file(doris::io::Path(path), &reader, nullptr);
+        if (!st.ok()) {
+            return paimon::Status::IOError(st.to_string());
+        }
+        return std::make_unique<DorisPaimonInputStream>(std::move(reader), path);
+    }
+
+    paimon::Result<std::unique_ptr<paimon::OutputStream>> Create(const std::string& path,
+                                                                 bool overwrite) const override {
+        bool exists = false;
+        doris::Status exists_st = _fs->exists(doris::io::Path(path), &exists);
+        if (!exists_st.ok()) {
+            return paimon::Status::IOError(exists_st.to_string());
+        }
+        if (exists) {
+            if (!overwrite) {
+                return paimon::Status::Exist("path already exists: ", path);
+            }
+            doris::Status del_st = _fs->delete_directory(doris::io::Path(path));

Review Comment:
   For `Create(path, overwrite=true)`, this deletes `path` as a directory before trying to delete it as a file. On HDFS `delete_directory` maps to recursive `hdfsDelete(..., 1)`, so if the target path is unexpectedly a directory this will remove the whole subtree. Paimon create/overwrite is for file paths; please delete files as files first and reject directories instead of recursively deleting them.



##########
be/src/exec/CMakeLists.txt:
##########
@@ -23,6 +23,17 @@ set(EXECUTABLE_OUTPUT_PATH "${BUILD_DIR}/src/exec")
 
 file(GLOB_RECURSE EXEC_FILES CONFIGURE_DEPENDS *.cpp)
 
+if (ENABLE_PAIMON_CPP)

Review Comment:
   This gate makes `ENABLE_PAIMON_CPP=OFF` builds link-fail. `operator.cpp` and `pipeline_fragment_context.cpp` now unconditionally include and instantiate `PaimonTableSinkOperatorX` / `AsyncWriterSink<VPaimonTableWriter>`, but the definitions for `PaimonTableSinkLocalState::{init,close}` and `VPaimonTableWriter` are only added to `Exec` inside this block. Since `ENABLE_PAIMON_CPP` remains an overridable build option, either the unconditional references need to be guarded too, or the non-paimon-cpp stubs need to be built so this configuration still links.



##########
be/src/vec/sink/writer/paimon/vpaimon_partition_writer.cpp:
##########
@@ -0,0 +1,269 @@
+// 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 "vec/sink/writer/paimon/vpaimon_partition_writer.h"
+
+#include <gen_cpp/DataSinks_types.h>
+
+#include <map>
+#include <utility>
+
+#include "common/metrics/doris_metrics.h"
+#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
+#include "runtime/runtime_state.h"
+
+#ifdef WITH_PAIMON_CPP
+#include <arrow/array.h>
+#include <arrow/c/bridge.h>
+#include <arrow/memory_pool.h>
+#include <arrow/record_batch.h>
+#include <arrow/status.h>
+#include <arrow/type.h>
+
+#include <string>
+
+#include "paimon/file_store_write.h"
+#include "paimon/record_batch.h"
+#endif
+
+namespace doris {
+namespace vectorized {
+
+#ifdef WITH_PAIMON_CPP
+namespace {
+class PaimonArrowMemPoolAdaptor : public arrow::MemoryPool {
+public:
+    explicit PaimonArrowMemPoolAdaptor(std::shared_ptr<::paimon::MemoryPool> pool)
+            : pool_(std::move(pool)) {}
+
+    arrow::Status Allocate(int64_t size, int64_t alignment, uint8_t** out) override {
+        *out = reinterpret_cast<uint8_t*>(pool_->Malloc(size, alignment));

Review Comment:
   `pool_->Malloc()` can return `nullptr` under memory pressure, but this Arrow `MemoryPool` reports `Status::OK()` and even records the allocation. Arrow callers treat OK as a valid non-null buffer and can then dereference a null pointer or corrupt memory. The adapter should return an out-of-memory `arrow::Status` when `Malloc`/`Realloc` returns null and only update stats after a successful allocation.



##########
be/src/vec/sink/vpaimon_table_writer.cpp:
##########
@@ -0,0 +1,740 @@
+// 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 "vec/sink/vpaimon_table_writer.h"
+
+#include <gen_cpp/DataSinks_types.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "common/metrics/doris_metrics.h"
+#include "core/block/block.h"
+#include "core/column/column.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "runtime/query_context.h"
+#include "runtime/runtime_profile.h"
+#include "runtime/runtime_state.h"
+#include "util/defer_op.h"
+#include "vec/sink/paimon_writer_utils.h"
+#include "vec/sink/writer/paimon/paimon_doris_hdfs_file_system.h"
+#include "vec/sink/writer/paimon/vpaimon_partition_writer.h"
+
+#ifdef WITH_PAIMON_CPP
+#include <arrow/array.h>
+#include <arrow/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/type.h>
+
+#include <cstdint>
+
+#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
+#include "format/parquet/arrow_memory_pool.h"
+#include "io/fs/hdfs_file_system.h"
+#include "io/hdfs_builder.h"
+#include "paimon/commit_message.h"
+#include "paimon/factories/factory_creator.h"
+#include "paimon/file_store_write.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/metrics.h"
+#include "paimon/utils/bucket_id_calculator.h"
+#include "paimon/write_context.h"
+#include "vec/sink/writer/paimon/paimon_doris_memory_pool.h"
+
+// Force link paimon file format factories
+namespace paimon {
+namespace parquet {}
+} // namespace paimon
+
+#endif
+
+namespace doris {
+namespace vectorized {
+
+#ifdef WITH_PAIMON_CPP
+namespace {
+bool is_paimon_cpp_time_metric(std::string_view name) {
+    return name.size() > 3 && name.substr(name.size() - 3) == "_ns";
+}
+
+void attach_paimon_cpp_metrics_to_profile(RuntimeProfile* profile,
+                                          const std::shared_ptr<::paimon::Metrics>& metrics) {
+    if (profile == nullptr || !metrics) {
+        return;
+    }
+    auto all = metrics->GetAllCounters();
+    if (all.empty()) {
+        return;
+    }
+    for (const auto& kv : all) {
+        std::string counter_name = "PaimonCpp_" + kv.first;
+        std::replace(counter_name.begin(), counter_name.end(), '.', '_');
+        RuntimeProfile::Counter* counter = nullptr;
+        if (is_paimon_cpp_time_metric(kv.first)) {
+            counter = ADD_COUNTER(profile, counter_name, TUnit::TIME_NS);
+        } else {
+            counter = ADD_COUNTER(profile, counter_name, TUnit::UNIT);
+        }
+        COUNTER_UPDATE(counter, kv.second);
+    }
+}
+} // namespace
+#endif
+
+VPaimonTableWriter::~VPaimonTableWriter() = default;
+
+VPaimonTableWriter::VPaimonTableWriter(const TDataSink& t_sink,
+                                       const VExprContextSPtrs& output_exprs)
+        : VPaimonTableWriter(t_sink, output_exprs, nullptr, nullptr) {}
+
+VPaimonTableWriter::VPaimonTableWriter(const TDataSink& t_sink,
+                                       const VExprContextSPtrs& output_exprs,
+                                       std::shared_ptr<Dependency> dep,
+                                       std::shared_ptr<Dependency> fin_dep)
+        : AsyncResultWriter(output_exprs, std::move(dep), std::move(fin_dep)), _t_sink(t_sink) {
+    DCHECK(_t_sink.__isset.paimon_table_sink);
+}
+
+Status VPaimonTableWriter::init_properties(ObjectPool* /*pool*/) {
+    // Currently there is no extra property to initialize. Kept for symmetry
+    // with VIcebergTableWriter and future paimon-cpp wiring.
+    return Status::OK();
+}
+
+Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
+    _state = state;
+    _profile = profile;
+#ifndef WITH_PAIMON_CPP
+    return Status::NotSupported("paimon-cpp is not enabled");
+#else
+    _written_rows_counter = ADD_COUNTER(_profile, "WrittenRows", TUnit::UNIT);
+    _written_bytes_counter = ADD_COUNTER(_profile, "WrittenBytes", TUnit::BYTES);
+    _send_data_timer = ADD_TIMER(_profile, "SendDataTime");
+    _project_timer = ADD_CHILD_TIMER(_profile, "ProjectTime", "SendDataTime");
+    _bucket_calc_timer = ADD_CHILD_TIMER(_profile, "BucketCalcTime", "SendDataTime");
+    _partition_writers_dispatch_timer =
+            ADD_CHILD_TIMER(_profile, "PartitionsDispatchTime", "SendDataTime");
+    _partition_writers_write_timer =
+            ADD_CHILD_TIMER(_profile, "PartitionsWriteTime", "SendDataTime");
+    _partition_writers_count = ADD_COUNTER(_profile, "PartitionsWriteCount", TUnit::UNIT);
+    _partition_writer_created = ADD_COUNTER(_profile, "PartitionWriterCreated", TUnit::UNIT);
+    _open_timer = ADD_TIMER(_profile, "OpenTime");
+    _close_timer = ADD_TIMER(_profile, "CloseTime");
+    _prepare_commit_timer = ADD_TIMER(_profile, "PrepareCommitTime");
+    _serialize_commit_messages_timer = ADD_TIMER(_profile, "SerializeCommitMessagesTime");
+    _commit_payload_bytes_counter = ADD_COUNTER(_profile, "CommitPayloadBytes", TUnit::BYTES);
+
+    SCOPED_TIMER(_open_timer);
+
+    ensure_paimon_doris_hdfs_file_system_registered();
+
+    auto registered_types = paimon::FactoryCreator::GetInstance()->GetRegisteredType();
+    std::string types_str;
+    bool has_parquet = false;
+    for (const auto& t : registered_types) {
+        types_str += t + ", ";
+        has_parquet |= (t == "parquet");
+    }
+    if (!has_parquet) {
+        return Status::InternalError(
+                "paimon-cpp parquet file format factory is not registered (missing 'parquet' in "
+                "FactoryCreator). Please ensure BE is built with WITH_PAIMON_CPP=ON and linked "
+                "with libpaimon_parquet_file_format.a (whole-archive). Registered factories: {}",
+                types_str);
+    }
+
+    _pool = std::make_shared<PaimonDorisMemoryPool>(_state->query_mem_tracker());
+    const auto& paimon_sink = _t_sink.paimon_table_sink;
+    if (!paimon_sink.__isset.table_location || paimon_sink.table_location.empty()) {
+        return Status::InvalidArgument("paimon table location is empty");
+    }
+    std::string commit_user;
+    if (paimon_sink.__isset.options) {
+        auto it = paimon_sink.options.find("doris.commit_user");
+        if (it != paimon_sink.options.end()) {
+            commit_user = it->second;
+        }
+    }
+    if (commit_user.empty()) {
+        commit_user = _state->user();
+    }
+
+    std::map<std::string, std::string> options;
+    if (paimon_sink.__isset.options) {
+        for (const auto& kv : paimon_sink.options) {
+            if (kv.first.rfind("doris.", 0) == 0) {
+                continue;
+            }
+            options.emplace(kv.first, kv.second);
+        }
+    }
+
+    // Workaround for paimon-cpp issue where it defaults to LocalFileSystem if path has no scheme.
+    // If table_location is missing scheme (common in HDFS setup without full URI),
+    // and fs.defaultFS is provided in options, we prepend it.
+    std::string table_location = paimon_sink.table_location;
+    if (table_location.find("://") == std::string::npos) {
+        auto it = options.find("fs.defaultFS");
+        if (it != options.end()) {
+            std::string default_fs = it->second;
+            // Remove trailing slash from default_fs if present
+            while (!default_fs.empty() && default_fs.back() == '/') {
+                default_fs.pop_back();
+            }
+            // Remove leading slash from table_location if present
+            if (!table_location.empty() && table_location.front() == '/') {
+                table_location = default_fs + table_location;
+            } else {
+                table_location = default_fs + "/" + table_location;
+            }
+        }
+    }
+
+    int64_t buffer_size = 256 * 1024 * 1024L; // Default 256MB
+
+    if (_state->query_options().__isset.paimon_write_buffer_size &&
+        _state->query_options().paimon_write_buffer_size > 0) {
+        buffer_size = _state->query_options().paimon_write_buffer_size;
+    }
+
+    bool enable_adaptive = true;
+    if (_state->query_options().__isset.enable_paimon_adaptive_buffer_size) {
+        enable_adaptive = _state->query_options().enable_paimon_adaptive_buffer_size;
+    }
+
+    if (enable_adaptive && paimon_sink.__isset.bucket_num && paimon_sink.bucket_num > 0) {
+        int bucket_num = paimon_sink.bucket_num;
+        buffer_size = get_paimon_write_buffer_size(buffer_size, true, bucket_num);
+        LOG(INFO) << "Adaptive Paimon Buffer Size: bucket_num=" << bucket_num
+                  << ", adjusted_buffer_size=" << buffer_size;
+    }
+    LOG(INFO) << "Paimon Native Writer Final Buffer Size: " << buffer_size
+              << " (enable_adaptive=" << enable_adaptive << ")";
+    options["write-buffer-size"] = std::to_string(buffer_size);
+
+    if (_state->query_options().__isset.paimon_target_file_size &&
+        _state->query_options().paimon_target_file_size > 0) {
+        options["target-file-size"] =
+                std::to_string(_state->query_options().paimon_target_file_size);
+        LOG(INFO) << "Paimon Native Writer Target File Size: "
+                  << _state->query_options().paimon_target_file_size;
+    }
+
+    options["file.format"] = "parquet";

Review Comment:
   The native writer unconditionally overrides the table options to parquet. For an existing Paimon table configured with ORC (which FE currently advertises as supported), this either writes data/manifest files in the wrong format or fails only after BE starts execution. The writer should honor the table's configured format, or FE should reject ORC for the native path and require the JNI writer before planning the sink.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3202185682


##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -0,0 +1,709 @@
+// 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.paimon;
+
+import org.apache.doris.common.security.authentication.PreExecutionAuthenticator;
+import org.apache.doris.common.security.authentication.PreExecutionAuthenticatorCache;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.DateDayVector;
+import org.apache.arrow.vector.DecimalVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.Float4Vector;
+import org.apache.arrow.vector.Float8Vector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.SmallIntVector;
+import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.TinyIntVector;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.holders.NullableTimeStampMicroHolder;
+import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.IOManagerImpl;
+import org.apache.paimon.io.DataOutputSerializer;
+import org.apache.paimon.memory.HeapMemorySegmentPool;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageSerializer;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BinaryType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VarBinaryType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.InputStream;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class PaimonJniWriter {
+    private static final Logger LOG = LoggerFactory.getLogger(PaimonJniWriter.class);
+
+    private BatchTableWrite writer;
+    private final BufferAllocator allocator;
+    private final CommitMessageSerializer serializer = new CommitMessageSerializer();
+    private String tableLocation;
+    private static final String PAIMON_OPTION_PREFIX = "paimon.";
+    private static final String HADOOP_OPTION_PREFIX = "hadoop.";
+
+    private PreExecutionAuthenticator preExecutionAuthenticator;
+    private final ClassLoader classLoader;
+
+    private Map<String, String> paimonOptionParams;
+    private Map<String, String> hadoopOptionParams;
+
+    private List<DataField> paimonFields;
+    private Map<String, DataField> paimonFieldMap;
+    private DataType[] targetTypes;
+
+    private IOManager ioManager;
+    private HeapMemorySegmentPool memorySegmentPool;
+    private boolean isDynamicBucketMode;
+
+    public PaimonJniWriter() {
+        this.allocator = new RootAllocator(Long.MAX_VALUE);
+        this.classLoader = this.getClass().getClassLoader();
+        try {
+            setWarnLevel("org.apache.paimon.shade.org.apache.parquet");
+            setWarnLevel("org.apache.paimon");
+        } catch (Throwable t) {
+            LOG.warn(t.getMessage(), t);
+        }
+    }
+
+    private void setWarnLevel(String loggerName) {
+        org.slf4j.Logger targetLogger = org.slf4j.LoggerFactory.getLogger(loggerName);
+        try {
+            Class<?> logbackLoggerClass = Class.forName("ch.qos.logback.classic.Logger");
+            Class<?> levelClass = Class.forName("ch.qos.logback.classic.Level");
+            if (!logbackLoggerClass.isInstance(targetLogger)) {
+                return;
+            }
+            Object warnLevel = levelClass.getField("WARN").get(null);
+            logbackLoggerClass.getMethod("setLevel", levelClass).invoke(targetLogger, warnLevel);
+        } catch (Throwable t) {
+            LOG.debug("set logger level skipped for {}", loggerName, t);
+        }
+    }
+
+    /**
+     * Opens the table and initializes the Writer
+     */
+    public void open(String tableLocation, Map<String, String> options, String[] columnNames) throws Exception {
+        this.tableLocation = tableLocation;
+        Thread.currentThread().setContextClassLoader(classLoader);
+        paimonOptionParams = options.entrySet().stream()
+                .filter(kv -> kv.getKey().startsWith(PAIMON_OPTION_PREFIX))
+                .collect(Collectors
+                        .toMap(kv1 -> kv1.getKey().substring(PAIMON_OPTION_PREFIX.length()),
+                                kv1 -> kv1.getValue()));
+        hadoopOptionParams = options.entrySet().stream()
+                .filter(kv -> kv.getKey().startsWith(HADOOP_OPTION_PREFIX))
+                .collect(Collectors
+                        .toMap(kv1 -> kv1.getKey().substring(HADOOP_OPTION_PREFIX.length()),
+                                kv1 -> kv1.getValue()));
+        if (!paimonOptionParams.containsKey("warehouse") && tableLocation != null) {
+            paimonOptionParams.put("warehouse", tableLocation);
+        }
+        this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(options);
+        preExecutionAuthenticator.execute(() -> {
+            try {
+                LOG.info("paimon: opening writer, location={}, opts_size={}", tableLocation, options.size());
+                Table table;
+                if (options.containsKey("serialized_table")) {
+                    table = PaimonUtils.deserialize(options.get("serialized_table"));
+                } else {
+                    Catalog catalog = createCatalog(paimonOptionParams, hadoopOptionParams);
+                    String dbName = options.getOrDefault("db_name", "default");
+                    String tblName = options.getOrDefault("table_name", "paimon_table");
+                    table = catalog.getTable(Identifier.create(dbName, tblName));
+                }
+                Map<String, String> dynamicOptions = new HashMap<>();
+                if (options.containsKey("write-buffer-size")) {
+                    dynamicOptions.put("write-buffer-size", options.get("write-buffer-size"));
+                }
+                copyIfPresent(options, dynamicOptions, "write-buffer-spillable");
+                copyIfPresent(options, dynamicOptions, "write-buffer-spill.max-disk-size");
+                copyIfPresent(options, dynamicOptions, "sort-spill-buffer-size");
+                copyIfPresent(options, dynamicOptions, "sort-spill-threshold");
+                copyIfPresent(options, dynamicOptions, "spill-compression");
+                boolean enableJniCompact = Boolean.parseBoolean(
+                        options.getOrDefault("paimon_use_jni_compact", "false"));
+                boolean enableInlineCompact = enableJniCompact;
+                if (enableJniCompact && table instanceof org.apache.paimon.table.FileStoreTable) {
+                    int tableBuckets = ((org.apache.paimon.table.FileStoreTable) table).schema().numBuckets();
+                    enableInlineCompact = tableBuckets > 0;
+                }
+                if (enableInlineCompact) {
+                    LOG.info("paimon: enabling inline compaction for JNI writer");
+                    dynamicOptions.put("write-only", "false");
+                    dynamicOptions.put("num-sorted-run.compaction-trigger", "10");
+                    dynamicOptions.put("compaction.max.worker-num", "8");
+                } else {
+                    LOG.info("paimon: write-only mode, compaction disabled");
+                    dynamicOptions.put("write-only", "true");
+                }
+                table = table.copy(dynamicOptions);
+                LOG.info("paimon: applied dynamic options to table: {}", dynamicOptions);
+                this.paimonFields = table.rowType().getFields();
+                this.paimonFieldMap = new HashMap<>();
+                for (DataField f : this.paimonFields) {
+                    this.paimonFieldMap.put(f.name(), f);
+                }
+                this.targetTypes = buildTargetTypes(columnNames);
+                this.writer = table.newBatchWriteBuilder().newWrite();
+                if (table instanceof org.apache.paimon.table.FileStoreTable) {
+                    this.isDynamicBucketMode =
+                            ((org.apache.paimon.table.FileStoreTable) table).schema().numBuckets() == -1;
+                }
+                boolean spillEnabled = Boolean.parseBoolean(options.getOrDefault("write-buffer-spillable", "false"));
+                if (spillEnabled) {
+                    String spillDir = options.get("paimon_jni_spill_dir");
+                    if (spillDir != null && !spillDir.isEmpty()) {
+                        File spillFile = new File(spillDir);
+                        if (!spillFile.exists()) {
+                            spillFile.mkdirs();
+                        }
+                        this.ioManager = new IOManagerImpl(spillDir);
+                    } else {
+                        this.ioManager = new IOManagerImpl(System.getProperty("java.io.tmpdir"));
+                    }
+                    long globalPoolSize = Long.parseLong(options.getOrDefault(
+                            "paimon_global_memory_pool_size", "1073741824"));
+                    this.memorySegmentPool = new HeapMemorySegmentPool(globalPoolSize, 32 * 1024);
+                    this.writer.withIOManager(ioManager).withMemoryPool(memorySegmentPool);
+                    LOG.info("paimon: spill enabled, spill_dir={}", spillDir);
+                }
+                return null;
+            } catch (Throwable t) {
+                throw contextException("open", "options_size=" + options.size(), t);
+            }
+        });
+    }
+
+    private void copyIfPresent(Map<String, String> options, Map<String, String> dynamicOptions, String key) {
+        String value = options.get(key);
+        if (value != null) {
+            dynamicOptions.put(key, value);
+        }
+    }
+
+    /**
+     * Receives Arrow IPC memory address from C++, deserializes and writes to Paimon
+     */
+    public void write(long address, int length) throws Exception {
+        preExecutionAuthenticator.execute(() -> {
+            try {
+                ByteBuffer directBuffer = getDirectBuffer(address, length);
+                try (ArrowStreamReader reader = new ArrowStreamReader(
+                        new DirectBufInputStream(directBuffer), allocator)) {
+                    VectorSchemaRoot root = reader.getVectorSchemaRoot();
+                    while (reader.loadNextBatch()) {
+                        writeBatch(root);
+                    }
+                }
+                return null;
+            } catch (Throwable t) {
+                throw contextException("write", "address=" + address + ", length=" + length, t);
+            }
+        });
+    }
+
+    private void writeBatch(VectorSchemaRoot root) throws Exception {
+        int rowCount = root.getRowCount();
+        if (rowCount == 0) {
+            return;
+        }
+        List<Field> fields = root.getSchema().getFields();
+        List<FieldVector> vectors = root.getFieldVectors();
+        int colCount = fields.size();
+        DataType[] currentTargetTypes = resolveTargetTypes(fields);
+        GenericRow reusedRow = new GenericRow(colCount);
+        for (int i = 0; i < rowCount; i++) {
+            for (int col = 0; col < colCount; col++) {
+                Field arrowField = fields.get(col);
+                try {
+                    reusedRow.setField(col, readArrowValue(vectors.get(col), i, arrowField, currentTargetTypes[col]));
+                } catch (Throwable t) {
+                    String fieldName = arrowField == null ? "null" : arrowField.getName();
+                    String currentTargetType = currentTargetTypes[col] == null
+                            ? "null"
+                            : currentTargetTypes[col].asSQLString();
+                    throw contextException("writeBatch.convert",
+                            "row=" + i + ", col=" + col + ", field=" + fieldName + ", targetType="
+                                    + currentTargetType + ", rowCount=" + rowCount + ", colCount=" + colCount,
+                            t);
+                }
+            }
+            try {
+                if (isDynamicBucketMode) {
+                    writer.write(reusedRow, 0);
+                } else {
+                    writer.write(reusedRow);
+                }
+            } catch (Throwable t) {
+                throw contextException("writeBatch.write",
+                        "row=" + i + ", rowCount=" + rowCount + ", colCount=" + colCount,
+                        t);
+            }
+        }
+    }
+
+    private Object readArrowValue(FieldVector vector, int row, Field arrowField, DataType targetType) {
+        if (vector == null || vector.isNull(row)) {
+            return null;
+        }
+        if (targetType instanceof BinaryType || targetType instanceof VarBinaryType) {
+            if (vector instanceof VarBinaryVector) {
+                return ((VarBinaryVector) vector).get(row);
+            }
+            if (vector instanceof VarCharVector) {
+                return ((VarCharVector) vector).get(row);
+            }
+        } else {
+            if (vector instanceof VarCharVector) {
+                return BinaryString.fromBytes(((VarCharVector) vector).get(row));
+            }
+        }
+        if (vector instanceof BitVector) {
+            return ((BitVector) vector).get(row) == 1;
+        }
+        if (vector instanceof TinyIntVector) {
+            return ((TinyIntVector) vector).get(row);
+        }
+        if (vector instanceof SmallIntVector) {
+            return ((SmallIntVector) vector).get(row);
+        }
+        if (vector instanceof IntVector) {
+            return ((IntVector) vector).get(row);
+        }
+        if (vector instanceof BigIntVector) {
+            return ((BigIntVector) vector).get(row);
+        }
+        if (vector instanceof Float4Vector) {
+            return ((Float4Vector) vector).get(row);
+        }
+        if (vector instanceof Float8Vector) {
+            return ((Float8Vector) vector).get(row);
+        }
+        if (vector instanceof DateDayVector) {
+            return ((DateDayVector) vector).get(row);
+        }
+        if (vector instanceof TimeStampMicroVector) {
+            NullableTimeStampMicroHolder holder = new NullableTimeStampMicroHolder();
+            ((TimeStampMicroVector) vector).get(row, holder);
+            return Timestamp.fromMicros(holder.value);
+        }
+        if (vector instanceof DecimalVector) {
+            return readDecimal((DecimalVector) vector, row);
+        }
+        Object val = vector.getObject(row);
+        return convertToPaimonType(val, arrowField, targetType);
+    }
+
+    private Decimal readDecimal(DecimalVector vector, int row) {
+        BigDecimal bd = getBigDecimalFromArrowBuf(vector.getDataBuffer(), row,
+                vector.getScale(), DecimalVector.TYPE_WIDTH);
+        return Decimal.fromBigDecimal(bd, vector.getPrecision(), vector.getScale());
+    }
+
+    private Object convertToPaimonType(Object val, Field arrowField, DataType targetType) {
+        if (val == null) {
+            return null;
+        }
+
+        if (targetType instanceof BinaryType || targetType instanceof VarBinaryType) {
+            if (val instanceof byte[]) {
+                return val;
+            } else if (val instanceof BinaryString) {
+                return ((BinaryString) val).toBytes();
+            } else if (val instanceof org.apache.arrow.vector.util.Text) {
+                return ((org.apache.arrow.vector.util.Text) val).copyBytes();
+            } else if (val instanceof org.apache.hadoop.io.Text) {
+                org.apache.hadoop.io.Text t = (org.apache.hadoop.io.Text) val;
+                byte[] bytes = new byte[t.getLength()];
+                System.arraycopy(t.getBytes(), 0, bytes, 0, t.getLength());
+                return bytes;
+            } else if (val instanceof String) {
+                return ((String) val).getBytes(java.nio.charset.StandardCharsets.UTF_8);
+            } else {
+                return val.toString().getBytes(java.nio.charset.StandardCharsets.UTF_8);
+            }
+        }
+
+        if (val instanceof BinaryString) {
+            return val;
+        }
+        if (val instanceof byte[]) {
+            return BinaryString.fromBytes((byte[]) val);
+        }
+        if (val instanceof org.apache.arrow.vector.util.Text) {
+            return BinaryString.fromBytes(((org.apache.arrow.vector.util.Text) val).copyBytes());
+        }
+        if (val instanceof org.apache.hadoop.io.Text) {
+            org.apache.hadoop.io.Text t = (org.apache.hadoop.io.Text) val;
+            return BinaryString.fromBytes(t.getBytes(), 0, t.getLength());
+        }
+        if (val instanceof CharSequence) {
+            return BinaryString.fromString(val.toString());
+        }
+
+
+        ArrowType arrowType = arrowField != null ? arrowField.getType() : null;
+        ArrowType.ArrowTypeID typeID = arrowType != null ? arrowType.getTypeID() : null;
+
+        if (val instanceof LocalDateTime) {
+            return Timestamp.fromLocalDateTime((LocalDateTime) val);
+        }
+        if (val instanceof Long && typeID == ArrowType.ArrowTypeID.Timestamp) {
+            return Timestamp.fromMicros((Long) val);
+        }
+        if (val instanceof java.time.LocalDate) {
+            return (int) ((java.time.LocalDate) val).toEpochDay();
+        }
+        if (val instanceof Integer && typeID == ArrowType.ArrowTypeID.Date) {
+            return val;
+        }
+        if (val instanceof BigDecimal) {
+            BigDecimal bd = (BigDecimal) val;
+            return Decimal.fromBigDecimal(bd, bd.precision(), bd.scale());
+        }
+        if (targetType instanceof RowType && val instanceof Map) {
+            RowType rowType = (RowType) targetType;
+            Map<?, ?> mapVal = (Map<?, ?>) val;
+            List<DataField> childFields = rowType.getFields();
+            GenericRow structRow = new GenericRow(childFields.size());
+
+            for (int i = 0; i < childFields.size(); i++) {
+                DataField childField = childFields.get(i);
+                Object childVal = mapVal.get(childField.name());
+
+                Field childArrowField = null;
+                if (arrowField != null && arrowField.getChildren() != null) {
+                    for (Field f : arrowField.getChildren()) {
+                        if (f.getName().equals(childField.name())) {
+                            childArrowField = f;
+                            break;
+                        }
+                    }
+                }
+                structRow.setField(i, convertToPaimonType(childVal, childArrowField, childField.type()));
+            }
+            return structRow;
+        }
+        if (targetType instanceof MapType && val instanceof List) {
+            MapType mapType = (MapType) targetType;
+            List<?> list = (List<?>) val;
+            Map<Object, Object> convertedMap = new HashMap<>();
+
+            Field keyArrowField = null;
+            Field valueArrowField = null;
+            if (arrowField != null && !arrowField.getChildren().isEmpty()) {
+                Field entriesStruct = arrowField.getChildren().get(0);
+                if (entriesStruct.getChildren().size() >= 2) {
+                    keyArrowField = entriesStruct.getChildren().get(0);
+                    valueArrowField = entriesStruct.getChildren().get(1);
+                }
+            }
+
+            String keyName = keyArrowField != null ? keyArrowField.getName() : "key";
+            String valueName = valueArrowField != null ? valueArrowField.getName() : "value";
+
+            for (Object element : list) {
+                if (element instanceof Map) { // Arrow 的内部 Struct 会被转成 Java Map
+                    Map<?, ?> kvStruct = (Map<?, ?>) element;
+                    Object k = convertToPaimonType(kvStruct.get(keyName), keyArrowField, mapType.getKeyType());
+                    Object v = convertToPaimonType(kvStruct.get(valueName), valueArrowField, mapType.getValueType());
+                    convertedMap.put(k, v);
+                }
+            }
+            return new GenericMap(convertedMap);
+        }
+        if (targetType instanceof ArrayType && val instanceof List) {
+            ArrayType arrayType = (ArrayType) targetType;
+            List<?> list = (List<?>) val;
+            Object[] convertedArray = new Object[list.size()];
+            Field childArrowField = (arrowField != null && !arrowField.getChildren().isEmpty())
+                    ? arrowField.getChildren().get(0) : null;
+
+            for (int i = 0; i < list.size(); i++) {
+                convertedArray[i] = convertToPaimonType(list.get(i), childArrowField, arrayType.getElementType());
+            }
+            return new GenericArray(convertedArray);
+        }
+
+        if (val instanceof byte[]) {
+            return val;
+        }
+
+        return val;
+    }
+
+    public byte[][] prepareCommit() throws Exception {
+        if (writer == null) {
+            return new byte[0][];
+        }
+        return preExecutionAuthenticator.execute(() -> {
+            try {
+                List<CommitMessage> messages = writer.prepareCommit();
+
+                if (messages == null || messages.isEmpty()) {
+                    LOG.info("paimon: prepareCommit returns empty, location={}", tableLocation);
+                    return new byte[0][];
+                }
+                LOG.info("paimon: prepareCommit returns {} messages", messages.size());
+                final int maxPayloadBytes = 8 * 1024 * 1024;
+                int chunkSize = 512;
+                java.util.ArrayList<byte[]> payloads = new java.util.ArrayList<>();
+                int i = 0;
+                while (i < messages.size()) {
+                    int end = Math.min(i + chunkSize, messages.size());
+                    DataOutputSerializer outputView = new DataOutputSerializer(1024);
+                    serializer.serializeList(messages.subList(i, end), outputView);
+                    byte[] data = outputView.getCopyOfBuffer();
+                    int len = data.length;
+                    int version = serializer.getVersion();
+                    byte[] payload = new byte[12 + len];
+                    payload[0] = 'D';
+                    payload[1] = 'P';
+                    payload[2] = 'C';
+                    payload[3] = 'M';
+                    payload[4] = (byte) ((version >>> 24) & 0xFF);
+                    payload[5] = (byte) ((version >>> 16) & 0xFF);
+                    payload[6] = (byte) ((version >>> 8) & 0xFF);
+                    payload[7] = (byte) (version & 0xFF);
+                    payload[8] = (byte) ((len >>> 24) & 0xFF);
+                    payload[9] = (byte) ((len >>> 16) & 0xFF);
+                    payload[10] = (byte) ((len >>> 8) & 0xFF);
+                    payload[11] = (byte) (len & 0xFF);
+                    System.arraycopy(data, 0, payload, 12, len);
+                    if (payload.length > maxPayloadBytes && chunkSize > 1) {
+                        chunkSize = Math.max(1, chunkSize / 2);
+                        continue;
+                    }
+                    payloads.add(payload);
+                    i = end;
+                }
+                return payloads.toArray(new byte[0][]);
+            } catch (Throwable t) {
+                throw contextException("prepareCommit", "tableLocation=" + tableLocation, t);
+            }
+        });
+    }
+
+    public void abort() {
+        try {
+            if (preExecutionAuthenticator != null) {
+                preExecutionAuthenticator.execute(() -> {
+                    closeWriterResources(false);
+                    return null;
+                });
+            }
+        } catch (Exception e) {
+            LOG.error("Failed to abort paimon writer", e);
+        }
+    }
+
+    public void close() {
+        try {
+            if (preExecutionAuthenticator != null) {
+                preExecutionAuthenticator.execute(() -> {
+                    closeWriterResources(true);
+                    return null;
+                });
+            }
+        } catch (Exception e) {
+            LOG.warn("Error while closing PaimonJniWriter", e);

Review Comment:
   Fixed. PaimonJniWriter.close() now propagates close failures back through JNI/C++, and BE only collects commit messages after Java close succeeds. This prevents FE from committing a transaction when the writer did not close cleanly.
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3202192214


##########
be/src/vec/sink/writer/paimon/vpaimon_partition_writer.cpp:
##########
@@ -0,0 +1,269 @@
+// 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 "vec/sink/writer/paimon/vpaimon_partition_writer.h"
+
+#include <gen_cpp/DataSinks_types.h>
+
+#include <map>
+#include <utility>
+
+#include "common/metrics/doris_metrics.h"
+#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
+#include "runtime/runtime_state.h"
+
+#ifdef WITH_PAIMON_CPP
+#include <arrow/array.h>
+#include <arrow/c/bridge.h>
+#include <arrow/memory_pool.h>
+#include <arrow/record_batch.h>
+#include <arrow/status.h>
+#include <arrow/type.h>
+
+#include <string>
+
+#include "paimon/file_store_write.h"
+#include "paimon/record_batch.h"
+#endif
+
+namespace doris {
+namespace vectorized {
+
+#ifdef WITH_PAIMON_CPP
+namespace {
+class PaimonArrowMemPoolAdaptor : public arrow::MemoryPool {
+public:
+    explicit PaimonArrowMemPoolAdaptor(std::shared_ptr<::paimon::MemoryPool> pool)
+            : pool_(std::move(pool)) {}
+
+    arrow::Status Allocate(int64_t size, int64_t alignment, uint8_t** out) override {
+        *out = reinterpret_cast<uint8_t*>(pool_->Malloc(size, alignment));

Review Comment:
   Fixed. The Arrow memory pool adapter now returns arrow::Status::OutOfMemory when Malloc/Realloc returns null, preserves the previous pointer on failed realloc, and updates allocation stats only after successful allocation.
   
   



##########
be/src/vec/sink/writer/paimon/paimon_doris_hdfs_file_system.cpp:
##########
@@ -0,0 +1,439 @@
+// 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 "vec/sink/writer/paimon/paimon_doris_hdfs_file_system.h"
+
+#ifdef WITH_PAIMON_CPP
+
+#include <gen_cpp/PlanNodes_types.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "common/status.h"
+#include "io/fs/file_reader.h"
+#include "io/fs/file_system.h"
+#include "io/fs/file_writer.h"
+#include "io/fs/hdfs_file_system.h"
+#include "io/fs/path.h"
+#include "io/hdfs_builder.h"
+#include "paimon/factories/factory_creator.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/fs/file_system_factory.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "util/slice.h"
+
+namespace {
+
+std::string _extract_hdfs_fs_name(const std::string& uri) {
+    auto starts_with = [&](const char* prefix) { return uri.rfind(prefix, 0) == 0; };
+    std::string_view scheme;
+    if (starts_with("hdfs://")) {
+        scheme = "hdfs://";
+    } else if (starts_with("dfs://")) {
+        scheme = "dfs://";
+    } else {
+        return {};
+    }
+    size_t authority_start = scheme.size();
+    size_t first_slash = uri.find('/', authority_start);
+    if (first_slash == std::string::npos) {
+        return uri;
+    }
+    if (first_slash == authority_start) {
+        return {};
+    }
+    return uri.substr(0, first_slash);
+}
+
+paimon::Status _to_paimon_status(const doris::Status& st) {
+    if (st.ok()) {
+        return paimon::Status::OK();
+    }
+    return paimon::Status::IOError(st.to_string());
+}
+
+class DorisPaimonBasicFileStatus final : public paimon::BasicFileStatus {
+public:
+    DorisPaimonBasicFileStatus(std::string path, bool is_dir)
+            : _path(std::move(path)), _is_dir(is_dir) {}
+
+    bool IsDir() const override { return _is_dir; }
+
+    std::string GetPath() const override { return _path; }
+
+private:
+    std::string _path;
+    bool _is_dir;
+};
+
+class DorisPaimonFileStatus final : public paimon::FileStatus {
+public:
+    DorisPaimonFileStatus(std::string path, bool is_dir, uint64_t len, int64_t mtime_ms)
+            : _path(std::move(path)), _is_dir(is_dir), _len(len), _mtime_ms(mtime_ms) {}
+
+    uint64_t GetLen() const override { return _len; }
+
+    bool IsDir() const override { return _is_dir; }
+
+    std::string GetPath() const override { return _path; }
+
+    int64_t GetModificationTime() const override { return _mtime_ms; }
+
+private:
+    std::string _path;
+    bool _is_dir;
+    uint64_t _len;
+    int64_t _mtime_ms;
+};
+
+class DorisPaimonInputStream final : public paimon::InputStream {
+public:
+    DorisPaimonInputStream(doris::io::FileReaderSPtr reader, std::string uri)
+            : _reader(std::move(reader)), _uri(std::move(uri)) {}
+
+    paimon::Status Close() override { return _to_paimon_status(_reader->close()); }
+
+    paimon::Status Seek(int64_t offset, paimon::SeekOrigin origin) override {
+        int64_t base = 0;
+        switch (origin) {
+        case paimon::FS_SEEK_SET:
+            base = 0;
+            break;
+        case paimon::FS_SEEK_CUR:
+            base = _pos;
+            break;
+        case paimon::FS_SEEK_END: {
+            auto len = Length();
+            if (!len.ok()) {
+                return len.status();
+            }
+            base = static_cast<int64_t>(len.value());
+            break;
+        }
+        default:
+            return paimon::Status::Invalid("invalid seek origin");
+        }
+        int64_t next = base + offset;
+        if (next < 0) {
+            return paimon::Status::Invalid("negative seek position: ", next);
+        }
+        _pos = next;
+        return paimon::Status::OK();
+    }
+
+    paimon::Result<int64_t> GetPos() const override { return _pos; }
+
+    paimon::Result<int32_t> Read(char* buffer, uint32_t size) override {
+        size_t bytes_read = 0;
+        doris::Slice slice(buffer, size);
+        doris::Status st = _reader->read_at(_pos, slice, &bytes_read, nullptr);
+        if (!st.ok()) {
+            return paimon::Status::IOError(st.to_string());
+        }
+        _pos += static_cast<int64_t>(bytes_read);
+        return static_cast<int32_t>(bytes_read);
+    }
+
+    paimon::Result<int32_t> Read(char* buffer, uint32_t size, uint64_t offset) override {
+        size_t bytes_read = 0;
+        doris::Slice slice(buffer, size);
+        doris::Status st = _reader->read_at(offset, slice, &bytes_read, nullptr);
+        if (!st.ok()) {
+            return paimon::Status::IOError(st.to_string());
+        }
+        return static_cast<int32_t>(bytes_read);
+    }
+
+    void ReadAsync(char* buffer, uint32_t size, uint64_t offset,
+                   std::function<void(paimon::Status)>&& callback) override {
+        auto res = Read(buffer, size, offset);
+        if (res.ok()) {
+            callback(paimon::Status::OK());
+        } else {
+            callback(res.status());
+        }
+    }
+
+    paimon::Result<std::string> GetUri() const override { return _uri; }
+
+    paimon::Result<uint64_t> Length() const override {
+        return static_cast<uint64_t>(_reader->size());
+    }
+
+private:
+    doris::io::FileReaderSPtr _reader;
+    std::string _uri;
+    int64_t _pos = 0;
+};
+
+class DorisPaimonOutputStream final : public paimon::OutputStream {
+public:
+    DorisPaimonOutputStream(doris::io::FileWriterPtr writer, std::string uri)
+            : _writer(std::move(writer)), _uri(std::move(uri)) {}
+
+    paimon::Status Close() override { return _to_paimon_status(_writer->close()); }
+
+    paimon::Result<int32_t> Write(const char* buffer, uint32_t size) override {
+        doris::Slice slice(buffer, size);
+        doris::Status st = _writer->append(slice);
+        if (!st.ok()) {
+            return paimon::Status::IOError(st.to_string());
+        }
+        _pos += size;
+        return static_cast<int32_t>(size);
+    }
+
+    paimon::Status Flush() override { return paimon::Status::OK(); }
+
+    paimon::Result<int64_t> GetPos() const override { return _pos; }
+
+    paimon::Result<std::string> GetUri() const override { return _uri; }
+
+private:
+    doris::io::FileWriterPtr _writer;
+    std::string _uri;
+    int64_t _pos = 0;
+};
+
+class DorisPaimonFileSystem final : public paimon::FileSystem {
+public:
+    explicit DorisPaimonFileSystem(std::shared_ptr<doris::io::FileSystem> fs)
+            : _fs(std::move(fs)) {}
+
+    paimon::Result<std::unique_ptr<paimon::InputStream>> Open(
+            const std::string& path) const override {
+        doris::io::FileReaderSPtr reader;
+        doris::Status st = _fs->open_file(doris::io::Path(path), &reader, nullptr);
+        if (!st.ok()) {
+            return paimon::Status::IOError(st.to_string());
+        }
+        return std::make_unique<DorisPaimonInputStream>(std::move(reader), path);
+    }
+
+    paimon::Result<std::unique_ptr<paimon::OutputStream>> Create(const std::string& path,
+                                                                 bool overwrite) const override {
+        bool exists = false;
+        doris::Status exists_st = _fs->exists(doris::io::Path(path), &exists);
+        if (!exists_st.ok()) {
+            return paimon::Status::IOError(exists_st.to_string());
+        }
+        if (exists) {
+            if (!overwrite) {
+                return paimon::Status::Exist("path already exists: ", path);
+            }
+            doris::Status del_st = _fs->delete_directory(doris::io::Path(path));

Review Comment:
   Fixed. Create(path, overwrite=true) now verifies the existing target is a file and deletes it with delete_file. If the target is a directory, it rejects the overwrite instead of recursively deleting the subtree.
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4415306847

   # FE Regression Coverage Report
   Increment line coverage `  7.36% (52/707)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_6a002645fe579dfc2e1cb2731413ad70493a1981_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_6a002645fe579dfc2e1cb2731413ad70493a1981_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4406587386

   # FE Regression Coverage Report
   Increment line coverage `  7.67% (45/587)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_09b7591b25385c938209a19f4825ba70b7f2a20f_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_09b7591b25385c938209a19f4825ba70b7f2a20f_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["yiguolei (via GitHub)" <gi...@apache.org>]

yiguolei commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3216715815


##########
be/src/exprs/function/simple_function_factory.h:
##########
@@ -117,6 +117,9 @@ void register_function_bit_test(SimpleFunctionFactory& factory);
 void register_function_dict_get(SimpleFunctionFactory& factory);
 void register_function_dict_get_many(SimpleFunctionFactory& factory);
 void register_function_ai(SimpleFunctionFactory& factory);
+namespace vectorized {

Review Comment:
   不要乱加namespace



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["yiguolei (via GitHub)" <gi...@apache.org>]

yiguolei commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3216730858


##########
be/cmake/thirdparty.cmake:
##########
@@ -117,11 +117,12 @@ add_thirdparty(rdkafka_cpp LIBNAME "lib/librdkafka++.a")
 add_thirdparty(rdkafka)
 add_thirdparty(s2)
 add_thirdparty(bitshuffle)
-add_thirdparty(roaring)

Review Comment:
   为什么要改这个？



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4432094915

   # BE Regression && UT Coverage Report
   Increment line coverage `28.02% (542/1934)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_merge/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_merge/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 71.45% (26494/37079) |
   | Line Coverage     | 54.50% (280946/515523) |
   | Region Coverage   | 51.88% (233888/450863) |
   | Branch Coverage   | 53.13% (100956/190014) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4320044010

   # FE Regression Coverage Report
   Increment line coverage `  7.76% (45/580)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_8ff21d356c47ba3ef5454bc7288dc67dcd2fb031_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_8ff21d356c47ba3ef5454bc7288dc67dcd2fb031_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4319625517

   # Cloud UT Coverage Report
   Increment line coverage ` ` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/8ff21d356c47ba3ef5454bc7288dc67dcd2fb031_8ff21d356c47ba3ef5454bc7288dc67dcd2fb031_cloud/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/8ff21d356c47ba3ef5454bc7288dc67dcd2fb031_8ff21d356c47ba3ef5454bc7288dc67dcd2fb031_cloud/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 78.05% (1842/2360) |
   | Line Coverage     | 64.80% (33007/50936) |
   | Region Coverage   | 65.30% (16370/25070) |
   | Branch Coverage   | 55.90% (8743/15640) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4349213381

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3202194760


##########
be/src/vec/sink/vpaimon_table_writer.cpp:
##########
@@ -0,0 +1,740 @@
+// 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 "vec/sink/vpaimon_table_writer.h"
+
+#include <gen_cpp/DataSinks_types.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "common/metrics/doris_metrics.h"
+#include "core/block/block.h"
+#include "core/column/column.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "runtime/query_context.h"
+#include "runtime/runtime_profile.h"
+#include "runtime/runtime_state.h"
+#include "util/defer_op.h"
+#include "vec/sink/paimon_writer_utils.h"
+#include "vec/sink/writer/paimon/paimon_doris_hdfs_file_system.h"
+#include "vec/sink/writer/paimon/vpaimon_partition_writer.h"
+
+#ifdef WITH_PAIMON_CPP
+#include <arrow/array.h>
+#include <arrow/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/type.h>
+
+#include <cstdint>
+
+#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
+#include "format/parquet/arrow_memory_pool.h"
+#include "io/fs/hdfs_file_system.h"
+#include "io/hdfs_builder.h"
+#include "paimon/commit_message.h"
+#include "paimon/factories/factory_creator.h"
+#include "paimon/file_store_write.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/metrics.h"
+#include "paimon/utils/bucket_id_calculator.h"
+#include "paimon/write_context.h"
+#include "vec/sink/writer/paimon/paimon_doris_memory_pool.h"
+
+// Force link paimon file format factories
+namespace paimon {
+namespace parquet {}
+} // namespace paimon
+
+#endif
+
+namespace doris {
+namespace vectorized {
+
+#ifdef WITH_PAIMON_CPP
+namespace {
+bool is_paimon_cpp_time_metric(std::string_view name) {
+    return name.size() > 3 && name.substr(name.size() - 3) == "_ns";
+}
+
+void attach_paimon_cpp_metrics_to_profile(RuntimeProfile* profile,
+                                          const std::shared_ptr<::paimon::Metrics>& metrics) {
+    if (profile == nullptr || !metrics) {
+        return;
+    }
+    auto all = metrics->GetAllCounters();
+    if (all.empty()) {
+        return;
+    }
+    for (const auto& kv : all) {
+        std::string counter_name = "PaimonCpp_" + kv.first;
+        std::replace(counter_name.begin(), counter_name.end(), '.', '_');
+        RuntimeProfile::Counter* counter = nullptr;
+        if (is_paimon_cpp_time_metric(kv.first)) {
+            counter = ADD_COUNTER(profile, counter_name, TUnit::TIME_NS);
+        } else {
+            counter = ADD_COUNTER(profile, counter_name, TUnit::UNIT);
+        }
+        COUNTER_UPDATE(counter, kv.second);
+    }
+}
+} // namespace
+#endif
+
+VPaimonTableWriter::~VPaimonTableWriter() = default;
+
+VPaimonTableWriter::VPaimonTableWriter(const TDataSink& t_sink,
+                                       const VExprContextSPtrs& output_exprs)
+        : VPaimonTableWriter(t_sink, output_exprs, nullptr, nullptr) {}
+
+VPaimonTableWriter::VPaimonTableWriter(const TDataSink& t_sink,
+                                       const VExprContextSPtrs& output_exprs,
+                                       std::shared_ptr<Dependency> dep,
+                                       std::shared_ptr<Dependency> fin_dep)
+        : AsyncResultWriter(output_exprs, std::move(dep), std::move(fin_dep)), _t_sink(t_sink) {
+    DCHECK(_t_sink.__isset.paimon_table_sink);
+}
+
+Status VPaimonTableWriter::init_properties(ObjectPool* /*pool*/) {
+    // Currently there is no extra property to initialize. Kept for symmetry
+    // with VIcebergTableWriter and future paimon-cpp wiring.
+    return Status::OK();
+}
+
+Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
+    _state = state;
+    _profile = profile;
+#ifndef WITH_PAIMON_CPP
+    return Status::NotSupported("paimon-cpp is not enabled");
+#else
+    _written_rows_counter = ADD_COUNTER(_profile, "WrittenRows", TUnit::UNIT);
+    _written_bytes_counter = ADD_COUNTER(_profile, "WrittenBytes", TUnit::BYTES);
+    _send_data_timer = ADD_TIMER(_profile, "SendDataTime");
+    _project_timer = ADD_CHILD_TIMER(_profile, "ProjectTime", "SendDataTime");
+    _bucket_calc_timer = ADD_CHILD_TIMER(_profile, "BucketCalcTime", "SendDataTime");
+    _partition_writers_dispatch_timer =
+            ADD_CHILD_TIMER(_profile, "PartitionsDispatchTime", "SendDataTime");
+    _partition_writers_write_timer =
+            ADD_CHILD_TIMER(_profile, "PartitionsWriteTime", "SendDataTime");
+    _partition_writers_count = ADD_COUNTER(_profile, "PartitionsWriteCount", TUnit::UNIT);
+    _partition_writer_created = ADD_COUNTER(_profile, "PartitionWriterCreated", TUnit::UNIT);
+    _open_timer = ADD_TIMER(_profile, "OpenTime");
+    _close_timer = ADD_TIMER(_profile, "CloseTime");
+    _prepare_commit_timer = ADD_TIMER(_profile, "PrepareCommitTime");
+    _serialize_commit_messages_timer = ADD_TIMER(_profile, "SerializeCommitMessagesTime");
+    _commit_payload_bytes_counter = ADD_COUNTER(_profile, "CommitPayloadBytes", TUnit::BYTES);
+
+    SCOPED_TIMER(_open_timer);
+
+    ensure_paimon_doris_hdfs_file_system_registered();
+
+    auto registered_types = paimon::FactoryCreator::GetInstance()->GetRegisteredType();
+    std::string types_str;
+    bool has_parquet = false;
+    for (const auto& t : registered_types) {
+        types_str += t + ", ";
+        has_parquet |= (t == "parquet");
+    }
+    if (!has_parquet) {
+        return Status::InternalError(
+                "paimon-cpp parquet file format factory is not registered (missing 'parquet' in "
+                "FactoryCreator). Please ensure BE is built with WITH_PAIMON_CPP=ON and linked "
+                "with libpaimon_parquet_file_format.a (whole-archive). Registered factories: {}",
+                types_str);
+    }
+
+    _pool = std::make_shared<PaimonDorisMemoryPool>(_state->query_mem_tracker());
+    const auto& paimon_sink = _t_sink.paimon_table_sink;
+    if (!paimon_sink.__isset.table_location || paimon_sink.table_location.empty()) {
+        return Status::InvalidArgument("paimon table location is empty");
+    }
+    std::string commit_user;
+    if (paimon_sink.__isset.options) {
+        auto it = paimon_sink.options.find("doris.commit_user");
+        if (it != paimon_sink.options.end()) {
+            commit_user = it->second;
+        }
+    }
+    if (commit_user.empty()) {
+        commit_user = _state->user();
+    }
+
+    std::map<std::string, std::string> options;
+    if (paimon_sink.__isset.options) {
+        for (const auto& kv : paimon_sink.options) {
+            if (kv.first.rfind("doris.", 0) == 0) {
+                continue;
+            }
+            options.emplace(kv.first, kv.second);
+        }
+    }
+
+    // Workaround for paimon-cpp issue where it defaults to LocalFileSystem if path has no scheme.
+    // If table_location is missing scheme (common in HDFS setup without full URI),
+    // and fs.defaultFS is provided in options, we prepend it.
+    std::string table_location = paimon_sink.table_location;
+    if (table_location.find("://") == std::string::npos) {
+        auto it = options.find("fs.defaultFS");
+        if (it != options.end()) {
+            std::string default_fs = it->second;
+            // Remove trailing slash from default_fs if present
+            while (!default_fs.empty() && default_fs.back() == '/') {
+                default_fs.pop_back();
+            }
+            // Remove leading slash from table_location if present
+            if (!table_location.empty() && table_location.front() == '/') {
+                table_location = default_fs + table_location;
+            } else {
+                table_location = default_fs + "/" + table_location;
+            }
+        }
+    }
+
+    int64_t buffer_size = 256 * 1024 * 1024L; // Default 256MB
+
+    if (_state->query_options().__isset.paimon_write_buffer_size &&
+        _state->query_options().paimon_write_buffer_size > 0) {
+        buffer_size = _state->query_options().paimon_write_buffer_size;
+    }
+
+    bool enable_adaptive = true;
+    if (_state->query_options().__isset.enable_paimon_adaptive_buffer_size) {
+        enable_adaptive = _state->query_options().enable_paimon_adaptive_buffer_size;
+    }
+
+    if (enable_adaptive && paimon_sink.__isset.bucket_num && paimon_sink.bucket_num > 0) {
+        int bucket_num = paimon_sink.bucket_num;
+        buffer_size = get_paimon_write_buffer_size(buffer_size, true, bucket_num);
+        LOG(INFO) << "Adaptive Paimon Buffer Size: bucket_num=" << bucket_num
+                  << ", adjusted_buffer_size=" << buffer_size;
+    }
+    LOG(INFO) << "Paimon Native Writer Final Buffer Size: " << buffer_size
+              << " (enable_adaptive=" << enable_adaptive << ")";
+    options["write-buffer-size"] = std::to_string(buffer_size);
+
+    if (_state->query_options().__isset.paimon_target_file_size &&
+        _state->query_options().paimon_target_file_size > 0) {
+        options["target-file-size"] =
+                std::to_string(_state->query_options().paimon_target_file_size);
+        LOG(INFO) << "Paimon Native Writer Target File Size: "
+                  << _state->query_options().paimon_target_file_size;
+    }
+
+    options["file.format"] = "parquet";

Review Comment:
   Fixed by rejecting ORC on the native paimon-cpp path. FE now forwards the table file.format/manifest.format options and raises an AnalysisException when enable_paimon_jni_writer=false targets an ORC table; BE also keeps a native-writer fallback check. ORC inserts should use the JNI writer for now.
   
   
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4406506708

   # BE Regression && UT Coverage Report
   Increment line coverage `25.36% (561/2212)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_09b7591b25385c938209a19f4825ba70b7f2a20f_merge/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_09b7591b25385c938209a19f4825ba70b7f2a20f_merge/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 71.57% (26502/37029) |
   | Line Coverage     | 54.55% (280990/515112) |
   | Region Coverage   | 51.68% (232835/450497) |
   | Branch Coverage   | 53.14% (100856/189807) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4430423788

   # Cloud UT Coverage Report
   Increment line coverage ` ` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_cloud/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_cloud/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 78.11% (1848/2366) |
   | Line Coverage     | 64.83% (33212/51232) |
   | Region Coverage   | 65.30% (16433/25167) |
   | Branch Coverage   | 55.84% (8773/15710) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4411735292

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["github-actions[bot] (via GitHub)" <gi...@apache.org>]

github-actions[bot] commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4417730337

   PR approved by anyone and no changes requested.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["yiguolei (via GitHub)" <gi...@apache.org>]

yiguolei commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3216747415


##########
be/src/vec/runtime/paimon_bucket_partitioner.cpp:
##########
@@ -0,0 +1,193 @@
+// 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 "vec/runtime/paimon_bucket_partitioner.h"

Review Comment:
   现在没有vec 这个目录了



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4437512641

   # FE UT Coverage Report
   Increment line coverage `  50.27% (278/553)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_48970b4f14dc51e6915bbf349ac223e96b9ce48c/fe_increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_48970b4f14dc51e6915bbf349ac223e96b9ce48c/fe_report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4412182542

   # FE Regression Coverage Report
   Increment line coverage `  8.58% (52/606)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_6a002645fe579dfc2e1cb2731413ad70493a1981_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_6a002645fe579dfc2e1cb2731413ad70493a1981_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["yiguolei (via GitHub)" <gi...@apache.org>]

yiguolei commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3216742852


##########
be/src/runtime/fragment_mgr.cpp:
##########
@@ -346,6 +346,289 @@ void FragmentMgr::stop() {
     }
 }
 
+std::string FragmentMgr::to_http_path(const std::string& file_name) {
+    std::stringstream url;
+    url << "http://" << BackendOptions::get_localhost() << ":" << config::webserver_port
+        << "/api/_download_load?"
+        << "token=" << _exec_env->token() << "&file=" << file_name;
+    return url.str();
+}
+
+Status FragmentMgr::trigger_pipeline_context_report(
+        const ReportStatusRequest req, std::shared_ptr<PipelineFragmentContext>&& ctx) {
+    return _thread_pool->submit_func([this, req, ctx]() {
+        SCOPED_ATTACH_TASK(ctx->get_query_ctx()->query_mem_tracker());
+        coordinator_callback(req);
+        if (!req.done) {
+            ctx->refresh_next_report_time();
+        }
+    });
+}
+
+// There can only be one of these callbacks in-flight at any moment, because
+// it is only invoked from the executor's reporting thread.
+// Also, the reported status will always reflect the most recent execution status,
+// including the final status when execution finishes.
+void FragmentMgr::coordinator_callback(const ReportStatusRequest& req) {

Review Comment:
   这里也不应该改



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4447418635

   # FE UT Coverage Report
   Increment line coverage `  50.27% (278/553)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_217a176440b5baddb646aef46d8f763aac735867/fe_increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_217a176440b5baddb646aef46d8f763aac735867/fe_report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3226095174


##########
be/src/vec/runtime/paimon_bucket_partitioner.cpp:
##########
@@ -0,0 +1,193 @@
+// 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 "vec/runtime/paimon_bucket_partitioner.h"

Review Comment:
   感谢指出，这个是分支适配时路径没有对齐导致的。现在已经按 branch-4.1 的目录结构调整，相关代码移动到了 be/src/exec/sink/writer/paimon/paimon_bucket_partitioner.*，不再使用 be/src/vec/runtime 路径。
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4432466482

   # BE Regression && UT Coverage Report
   Increment line coverage `28.02% (542/1934)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_merge/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_merge/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 71.44% (26491/37079) |
   | Line Coverage     | 54.49% (280931/515523) |
   | Region Coverage   | 51.87% (233865/450863) |
   | Branch Coverage   | 53.13% (100953/190014) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4443315199

   # FE Regression Coverage Report
   Increment line coverage `  8.32% (46/553)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_ebc64c58aeed7e11f3b7e6cba70d061461b1bf63_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_ebc64c58aeed7e11f3b7e6cba70d061461b1bf63_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4447079179

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4440941921

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["yiguolei (via GitHub)" <gi...@apache.org>]

yiguolei commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3216767298


##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -1877,6 +1886,16 @@ bool PipelineFragmentContext::_close_fragment_instance() {
     Defer defer_op {[&]() { _is_fragment_instance_closed = true; }};
     _fragment_level_profile->total_time_counter()->update(_fragment_watcher.elapsed_time());
     if (!_need_notify_close) {
+        if (_runtime_state) {
+            for (auto& tasks : _tasks) {
+                for (auto& task : tasks) {
+                    if (!task.first->runtime_state()->paimon_commit_messages().empty()) {
+                        _runtime_state->add_paimon_commit_messages(

Review Comment:
   之前别的writer 写入，应该也有写commit message，但是不是这么写的，最好你看看要统一的一致。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4294114137

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["suxiaogang223 (via GitHub)" <gi...@apache.org>]

suxiaogang223 commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3199759452


##########
be/src/vec/sink/vpaimon_table_writer.cpp:
##########
@@ -0,0 +1,740 @@
+// 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 "vec/sink/vpaimon_table_writer.h"
+
+#include <gen_cpp/DataSinks_types.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "common/metrics/doris_metrics.h"
+#include "core/block/block.h"
+#include "core/column/column.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "runtime/query_context.h"
+#include "runtime/runtime_profile.h"
+#include "runtime/runtime_state.h"
+#include "util/defer_op.h"
+#include "vec/sink/paimon_writer_utils.h"
+#include "vec/sink/writer/paimon/paimon_doris_hdfs_file_system.h"
+#include "vec/sink/writer/paimon/vpaimon_partition_writer.h"
+
+#ifdef WITH_PAIMON_CPP
+#include <arrow/array.h>
+#include <arrow/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/type.h>
+
+#include <cstdint>
+
+#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
+#include "format/parquet/arrow_memory_pool.h"
+#include "io/fs/hdfs_file_system.h"
+#include "io/hdfs_builder.h"
+#include "paimon/commit_message.h"
+#include "paimon/factories/factory_creator.h"
+#include "paimon/file_store_write.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/metrics.h"
+#include "paimon/utils/bucket_id_calculator.h"
+#include "paimon/write_context.h"
+#include "vec/sink/writer/paimon/paimon_doris_memory_pool.h"
+
+// Force link paimon file format factories
+namespace paimon {
+namespace parquet {}
+} // namespace paimon
+
+#endif
+
+namespace doris {
+namespace vectorized {
+
+#ifdef WITH_PAIMON_CPP
+namespace {
+bool is_paimon_cpp_time_metric(std::string_view name) {
+    return name.size() > 3 && name.substr(name.size() - 3) == "_ns";
+}
+
+void attach_paimon_cpp_metrics_to_profile(RuntimeProfile* profile,
+                                          const std::shared_ptr<::paimon::Metrics>& metrics) {
+    if (profile == nullptr || !metrics) {
+        return;
+    }
+    auto all = metrics->GetAllCounters();
+    if (all.empty()) {
+        return;
+    }
+    for (const auto& kv : all) {
+        std::string counter_name = "PaimonCpp_" + kv.first;
+        std::replace(counter_name.begin(), counter_name.end(), '.', '_');
+        RuntimeProfile::Counter* counter = nullptr;
+        if (is_paimon_cpp_time_metric(kv.first)) {
+            counter = ADD_COUNTER(profile, counter_name, TUnit::TIME_NS);
+        } else {
+            counter = ADD_COUNTER(profile, counter_name, TUnit::UNIT);
+        }
+        COUNTER_UPDATE(counter, kv.second);
+    }
+}
+} // namespace
+#endif
+
+VPaimonTableWriter::~VPaimonTableWriter() = default;
+
+VPaimonTableWriter::VPaimonTableWriter(const TDataSink& t_sink,
+                                       const VExprContextSPtrs& output_exprs)
+        : VPaimonTableWriter(t_sink, output_exprs, nullptr, nullptr) {}
+
+VPaimonTableWriter::VPaimonTableWriter(const TDataSink& t_sink,
+                                       const VExprContextSPtrs& output_exprs,
+                                       std::shared_ptr<Dependency> dep,
+                                       std::shared_ptr<Dependency> fin_dep)
+        : AsyncResultWriter(output_exprs, std::move(dep), std::move(fin_dep)), _t_sink(t_sink) {
+    DCHECK(_t_sink.__isset.paimon_table_sink);
+}
+
+Status VPaimonTableWriter::init_properties(ObjectPool* /*pool*/) {
+    // Currently there is no extra property to initialize. Kept for symmetry
+    // with VIcebergTableWriter and future paimon-cpp wiring.
+    return Status::OK();
+}
+
+Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
+    _state = state;
+    _profile = profile;
+#ifndef WITH_PAIMON_CPP
+    return Status::NotSupported("paimon-cpp is not enabled");
+#else
+    _written_rows_counter = ADD_COUNTER(_profile, "WrittenRows", TUnit::UNIT);
+    _written_bytes_counter = ADD_COUNTER(_profile, "WrittenBytes", TUnit::BYTES);
+    _send_data_timer = ADD_TIMER(_profile, "SendDataTime");
+    _project_timer = ADD_CHILD_TIMER(_profile, "ProjectTime", "SendDataTime");
+    _bucket_calc_timer = ADD_CHILD_TIMER(_profile, "BucketCalcTime", "SendDataTime");
+    _partition_writers_dispatch_timer =
+            ADD_CHILD_TIMER(_profile, "PartitionsDispatchTime", "SendDataTime");
+    _partition_writers_write_timer =
+            ADD_CHILD_TIMER(_profile, "PartitionsWriteTime", "SendDataTime");
+    _partition_writers_count = ADD_COUNTER(_profile, "PartitionsWriteCount", TUnit::UNIT);
+    _partition_writer_created = ADD_COUNTER(_profile, "PartitionWriterCreated", TUnit::UNIT);
+    _open_timer = ADD_TIMER(_profile, "OpenTime");
+    _close_timer = ADD_TIMER(_profile, "CloseTime");
+    _prepare_commit_timer = ADD_TIMER(_profile, "PrepareCommitTime");
+    _serialize_commit_messages_timer = ADD_TIMER(_profile, "SerializeCommitMessagesTime");
+    _commit_payload_bytes_counter = ADD_COUNTER(_profile, "CommitPayloadBytes", TUnit::BYTES);
+
+    SCOPED_TIMER(_open_timer);
+
+    ensure_paimon_doris_hdfs_file_system_registered();
+
+    auto registered_types = paimon::FactoryCreator::GetInstance()->GetRegisteredType();
+    std::string types_str;
+    bool has_parquet = false;
+    for (const auto& t : registered_types) {
+        types_str += t + ", ";
+        has_parquet |= (t == "parquet");
+    }
+    if (!has_parquet) {
+        return Status::InternalError(
+                "paimon-cpp parquet file format factory is not registered (missing 'parquet' in "
+                "FactoryCreator). Please ensure BE is built with WITH_PAIMON_CPP=ON and linked "
+                "with libpaimon_parquet_file_format.a (whole-archive). Registered factories: {}",
+                types_str);
+    }
+
+    _pool = std::make_shared<PaimonDorisMemoryPool>(_state->query_mem_tracker());
+    const auto& paimon_sink = _t_sink.paimon_table_sink;
+    if (!paimon_sink.__isset.table_location || paimon_sink.table_location.empty()) {
+        return Status::InvalidArgument("paimon table location is empty");
+    }
+    std::string commit_user;
+    if (paimon_sink.__isset.options) {
+        auto it = paimon_sink.options.find("doris.commit_user");
+        if (it != paimon_sink.options.end()) {
+            commit_user = it->second;
+        }
+    }
+    if (commit_user.empty()) {
+        commit_user = _state->user();
+    }
+
+    std::map<std::string, std::string> options;
+    if (paimon_sink.__isset.options) {
+        for (const auto& kv : paimon_sink.options) {
+            if (kv.first.rfind("doris.", 0) == 0) {
+                continue;
+            }
+            options.emplace(kv.first, kv.second);
+        }
+    }
+
+    // Workaround for paimon-cpp issue where it defaults to LocalFileSystem if path has no scheme.
+    // If table_location is missing scheme (common in HDFS setup without full URI),
+    // and fs.defaultFS is provided in options, we prepend it.
+    std::string table_location = paimon_sink.table_location;
+    if (table_location.find("://") == std::string::npos) {
+        auto it = options.find("fs.defaultFS");
+        if (it != options.end()) {
+            std::string default_fs = it->second;
+            // Remove trailing slash from default_fs if present
+            while (!default_fs.empty() && default_fs.back() == '/') {
+                default_fs.pop_back();
+            }
+            // Remove leading slash from table_location if present
+            if (!table_location.empty() && table_location.front() == '/') {
+                table_location = default_fs + table_location;
+            } else {
+                table_location = default_fs + "/" + table_location;
+            }
+        }
+    }
+
+    int64_t buffer_size = 256 * 1024 * 1024L; // Default 256MB
+
+    if (_state->query_options().__isset.paimon_write_buffer_size &&
+        _state->query_options().paimon_write_buffer_size > 0) {
+        buffer_size = _state->query_options().paimon_write_buffer_size;
+    }
+
+    bool enable_adaptive = true;
+    if (_state->query_options().__isset.enable_paimon_adaptive_buffer_size) {
+        enable_adaptive = _state->query_options().enable_paimon_adaptive_buffer_size;
+    }
+
+    if (enable_adaptive && paimon_sink.__isset.bucket_num && paimon_sink.bucket_num > 0) {
+        int bucket_num = paimon_sink.bucket_num;
+        buffer_size = get_paimon_write_buffer_size(buffer_size, true, bucket_num);
+        LOG(INFO) << "Adaptive Paimon Buffer Size: bucket_num=" << bucket_num
+                  << ", adjusted_buffer_size=" << buffer_size;
+    }
+    LOG(INFO) << "Paimon Native Writer Final Buffer Size: " << buffer_size
+              << " (enable_adaptive=" << enable_adaptive << ")";
+    options["write-buffer-size"] = std::to_string(buffer_size);
+
+    if (_state->query_options().__isset.paimon_target_file_size &&
+        _state->query_options().paimon_target_file_size > 0) {
+        options["target-file-size"] =
+                std::to_string(_state->query_options().paimon_target_file_size);
+        LOG(INFO) << "Paimon Native Writer Target File Size: "
+                  << _state->query_options().paimon_target_file_size;
+    }
+
+    options["file.format"] = "parquet";
+    options["manifest.format"] = "parquet";
+    if ((!paimon_sink.__isset.bucket_num || paimon_sink.bucket_num <= 0) &&
+        paimon_sink.__isset.bucket_keys && !paimon_sink.bucket_keys.empty()) {
+        return Status::NotSupported(
+                "paimon-cpp native writer does not support primary-key table with dynamic bucket "
+                "(bucket=-1) yet; enable_paimon_jni_writer=true is required");
+    }
+
+    ::paimon::WriteContextBuilder builder(table_location, commit_user);
+    builder.SetOptions(options);
+    builder.WithIgnorePreviousFiles(true);
+    builder.WithMemoryPool(_pool);
+    builder.WithFileSystemSchemeToIdentifierMap(
+            {{"hdfs", kPaimonDorisHdfsFsIdentifier}, {"dfs", kPaimonDorisHdfsFsIdentifier}});
+    auto ctx_result = builder.Finish();
+    if (!ctx_result.ok()) {
+        return Status::InternalError("failed to build paimon write context: {}",
+                                     ctx_result.status().ToString());
+    }
+    auto write_result = ::paimon::FileStoreWrite::Create(std::move(ctx_result).value());
+    if (!write_result.ok()) {
+        auto message = write_result.status().ToString();
+        if (message.find("format 'orc'") != std::string::npos ||
+            message.find("format \"orc\"") != std::string::npos) {
+            return Status::InternalError(
+                    "failed to create paimon file store write: {}. This Doris build does not "
+                    "support ORC in paimon-cpp; create paimon table with options "
+                    "file.format=parquet and manifest.format=parquet.",
+                    message);
+        }
+        return Status::InternalError("failed to create paimon file store write: {}", message);
+    }
+    _file_store_write = std::move(write_result).value();
+    return Status::OK();
+#endif
+}
+
+Status VPaimonTableWriter::_init_partition_column_indices(const ::doris::Block& block) const {
+    const TPaimonTableSink& paimon_sink = _t_sink.paimon_table_sink;
+    if (!paimon_sink.__isset.partition_keys || paimon_sink.partition_keys.empty() ||
+        _partition_indices_inited) {
+        return Status::OK();
+    }
+
+    std::unordered_map<std::string, int> name_to_idx;
+    for (int i = 0; i < block.columns(); ++i) {
+        std::string col_name = block.get_by_position(i).name;
+        if (col_name.empty()) {
+            if (paimon_sink.__isset.column_names && i < paimon_sink.column_names.size()) {
+                col_name = paimon_sink.column_names[i];
+            }
+        }
+        name_to_idx.emplace(col_name, i);
+    }
+    _partition_column_indices.clear();
+    _partition_column_indices.reserve(paimon_sink.partition_keys.size());
+    for (const auto& key_name : paimon_sink.partition_keys) {
+        auto it = name_to_idx.find(key_name);
+        if (it == name_to_idx.end()) {
+            return Status::InvalidArgument("paimon partition key {} not found in output block",
+                                           key_name);
+        }
+        _partition_column_indices.push_back(it->second);
+    }
+    _partition_indices_inited = true;
+    return Status::OK();
+}
+
+std::string VPaimonTableWriter::_default_partition_name() const {
+    std::string default_part = "__DEFAULT_PARTITION__";
+    const TPaimonTableSink& paimon_sink = _t_sink.paimon_table_sink;
+    if (paimon_sink.__isset.options) {
+        auto it = paimon_sink.options.find("partition.default-name");
+        if (it != paimon_sink.options.end() && !it->second.empty()) {
+            default_part = it->second;
+        }
+    }
+    return default_part;
+}
+
+Status VPaimonTableWriter::_collect_partition_value_columns(
+        const ::doris::Block& block,
+        std::vector<std::vector<std::string>>* partition_value_columns) const {
+    partition_value_columns->clear();
+    const TPaimonTableSink& paimon_sink = _t_sink.paimon_table_sink;
+    if (!paimon_sink.__isset.partition_keys || paimon_sink.partition_keys.empty()) {
+        return Status::OK();
+    }
+
+    RETURN_IF_ERROR(_init_partition_column_indices(block));
+    const size_t rows = block.rows();
+    partition_value_columns->resize(_partition_column_indices.size());
+    const std::string default_part = _default_partition_name();
+    DataTypeSerDe::FormatOptions options;
+    for (size_t part_idx = 0; part_idx < _partition_column_indices.size(); ++part_idx) {
+        auto& values = (*partition_value_columns)[part_idx];
+        values.resize(rows);
+        const auto& col_with_type = block.get_by_position(_partition_column_indices[part_idx]);
+        const auto& type = col_with_type.type;
+        const auto& col = col_with_type.column;
+        for (size_t row = 0; row < rows; ++row) {
+            if (col->is_null_at(row)) {
+                values[row] = default_part;
+            } else {
+                values[row] = type->to_string(*col, row, options);
+            }
+        }
+    }
+    return Status::OK();
+}
+
+Status VPaimonTableWriter::_get_or_create_writer(const WriteKey& key,
+                                                 std::shared_ptr<VPaimonPartitionWriter>* writer) {
+    auto it = _writers.find(key);
+    if (it != _writers.end()) {
+        *writer = it->second;
+        return Status::OK();
+    }
+
+    auto new_writer =
+            std::make_shared<VPaimonPartitionWriter>(_t_sink, key.partition_values, key.bucket_id
+#ifdef WITH_PAIMON_CPP
+                                                     ,
+                                                     _file_store_write.get(), _pool
+#endif
+            );
+    RETURN_IF_ERROR(new_writer->init_properties(nullptr));
+    RETURN_IF_ERROR(new_writer->open(_state, _profile));
+    COUNTER_UPDATE(_partition_writer_created, 1);
+    DorisMetrics::instance()->paimon_partition_writer_created->increment(1);
+    _writers.emplace(key, new_writer);
+    *writer = std::move(new_writer);
+    return Status::OK();
+}
+
+Status VPaimonTableWriter::write(RuntimeState* state, ::doris::Block& block) {
+#ifndef WITH_PAIMON_CPP
+    return Status::NotSupported("paimon-cpp is not enabled");
+#else
+    if (block.rows() == 0) {
+        return Status::OK();
+    }
+    SCOPED_TIMER(_send_data_timer);
+    int64_t send_data_ns = 0;
+    auto to_ms_ceil = [](int64_t ns) -> uint64_t {
+        if (ns <= 0) {
+            return 0;
+        }
+        return static_cast<uint64_t>((ns + 999999) / 1000000);
+    };
+    Defer record_send_data_latency {[&]() {
+        DorisMetrics::instance()->paimon_write_send_data_latency_ms->add(to_ms_ceil(send_data_ns));
+    }};
+    SCOPED_RAW_TIMER(&send_data_ns);
+
+    Block output_block;
+    int64_t project_ns = 0;
+    {
+        SCOPED_TIMER(_project_timer);
+        SCOPED_RAW_TIMER(&project_ns);
+        RETURN_IF_ERROR(_projection_block(block, &output_block));
+    }
+    _row_count += output_block.rows();

Review Comment:
   The JNI writer updates `RuntimeState` load counters after accepting rows, but this native paimon-cpp path only updates the local profile counters and metrics. FE derives INSERT OK rows, audit return rows, and `loadedRows` from BE load counters, so inserts with `enable_paimon_jni_writer=false` can succeed while reporting 0 loaded rows. Please update `update_num_rows_load_total()` and `update_num_bytes_load_total()` on this path as well after the projected block is accepted.



##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -0,0 +1,709 @@
+// 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.paimon;
+
+import org.apache.doris.common.security.authentication.PreExecutionAuthenticator;
+import org.apache.doris.common.security.authentication.PreExecutionAuthenticatorCache;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.DateDayVector;
+import org.apache.arrow.vector.DecimalVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.Float4Vector;
+import org.apache.arrow.vector.Float8Vector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.SmallIntVector;
+import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.TinyIntVector;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.holders.NullableTimeStampMicroHolder;
+import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.IOManagerImpl;
+import org.apache.paimon.io.DataOutputSerializer;
+import org.apache.paimon.memory.HeapMemorySegmentPool;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageSerializer;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BinaryType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VarBinaryType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.InputStream;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class PaimonJniWriter {
+    private static final Logger LOG = LoggerFactory.getLogger(PaimonJniWriter.class);
+
+    private BatchTableWrite writer;
+    private final BufferAllocator allocator;
+    private final CommitMessageSerializer serializer = new CommitMessageSerializer();
+    private String tableLocation;
+    private static final String PAIMON_OPTION_PREFIX = "paimon.";
+    private static final String HADOOP_OPTION_PREFIX = "hadoop.";
+
+    private PreExecutionAuthenticator preExecutionAuthenticator;
+    private final ClassLoader classLoader;
+
+    private Map<String, String> paimonOptionParams;
+    private Map<String, String> hadoopOptionParams;
+
+    private List<DataField> paimonFields;
+    private Map<String, DataField> paimonFieldMap;
+    private DataType[] targetTypes;
+
+    private IOManager ioManager;
+    private HeapMemorySegmentPool memorySegmentPool;
+    private boolean isDynamicBucketMode;
+
+    public PaimonJniWriter() {
+        this.allocator = new RootAllocator(Long.MAX_VALUE);
+        this.classLoader = this.getClass().getClassLoader();
+        try {
+            setWarnLevel("org.apache.paimon.shade.org.apache.parquet");
+            setWarnLevel("org.apache.paimon");
+        } catch (Throwable t) {
+            LOG.warn(t.getMessage(), t);
+        }
+    }
+
+    private void setWarnLevel(String loggerName) {
+        org.slf4j.Logger targetLogger = org.slf4j.LoggerFactory.getLogger(loggerName);
+        try {
+            Class<?> logbackLoggerClass = Class.forName("ch.qos.logback.classic.Logger");
+            Class<?> levelClass = Class.forName("ch.qos.logback.classic.Level");
+            if (!logbackLoggerClass.isInstance(targetLogger)) {
+                return;
+            }
+            Object warnLevel = levelClass.getField("WARN").get(null);
+            logbackLoggerClass.getMethod("setLevel", levelClass).invoke(targetLogger, warnLevel);
+        } catch (Throwable t) {
+            LOG.debug("set logger level skipped for {}", loggerName, t);
+        }
+    }
+
+    /**
+     * Opens the table and initializes the Writer
+     */
+    public void open(String tableLocation, Map<String, String> options, String[] columnNames) throws Exception {
+        this.tableLocation = tableLocation;
+        Thread.currentThread().setContextClassLoader(classLoader);
+        paimonOptionParams = options.entrySet().stream()
+                .filter(kv -> kv.getKey().startsWith(PAIMON_OPTION_PREFIX))
+                .collect(Collectors
+                        .toMap(kv1 -> kv1.getKey().substring(PAIMON_OPTION_PREFIX.length()),
+                                kv1 -> kv1.getValue()));
+        hadoopOptionParams = options.entrySet().stream()
+                .filter(kv -> kv.getKey().startsWith(HADOOP_OPTION_PREFIX))
+                .collect(Collectors
+                        .toMap(kv1 -> kv1.getKey().substring(HADOOP_OPTION_PREFIX.length()),
+                                kv1 -> kv1.getValue()));
+        if (!paimonOptionParams.containsKey("warehouse") && tableLocation != null) {
+            paimonOptionParams.put("warehouse", tableLocation);
+        }
+        this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(options);
+        preExecutionAuthenticator.execute(() -> {
+            try {
+                LOG.info("paimon: opening writer, location={}, opts_size={}", tableLocation, options.size());
+                Table table;
+                if (options.containsKey("serialized_table")) {
+                    table = PaimonUtils.deserialize(options.get("serialized_table"));
+                } else {
+                    Catalog catalog = createCatalog(paimonOptionParams, hadoopOptionParams);
+                    String dbName = options.getOrDefault("db_name", "default");
+                    String tblName = options.getOrDefault("table_name", "paimon_table");
+                    table = catalog.getTable(Identifier.create(dbName, tblName));
+                }
+                Map<String, String> dynamicOptions = new HashMap<>();
+                if (options.containsKey("write-buffer-size")) {
+                    dynamicOptions.put("write-buffer-size", options.get("write-buffer-size"));
+                }
+                copyIfPresent(options, dynamicOptions, "write-buffer-spillable");
+                copyIfPresent(options, dynamicOptions, "write-buffer-spill.max-disk-size");
+                copyIfPresent(options, dynamicOptions, "sort-spill-buffer-size");
+                copyIfPresent(options, dynamicOptions, "sort-spill-threshold");
+                copyIfPresent(options, dynamicOptions, "spill-compression");
+                boolean enableJniCompact = Boolean.parseBoolean(
+                        options.getOrDefault("paimon_use_jni_compact", "false"));
+                boolean enableInlineCompact = enableJniCompact;
+                if (enableJniCompact && table instanceof org.apache.paimon.table.FileStoreTable) {
+                    int tableBuckets = ((org.apache.paimon.table.FileStoreTable) table).schema().numBuckets();
+                    enableInlineCompact = tableBuckets > 0;
+                }
+                if (enableInlineCompact) {
+                    LOG.info("paimon: enabling inline compaction for JNI writer");
+                    dynamicOptions.put("write-only", "false");
+                    dynamicOptions.put("num-sorted-run.compaction-trigger", "10");
+                    dynamicOptions.put("compaction.max.worker-num", "8");
+                } else {
+                    LOG.info("paimon: write-only mode, compaction disabled");
+                    dynamicOptions.put("write-only", "true");
+                }
+                table = table.copy(dynamicOptions);
+                LOG.info("paimon: applied dynamic options to table: {}", dynamicOptions);
+                this.paimonFields = table.rowType().getFields();
+                this.paimonFieldMap = new HashMap<>();
+                for (DataField f : this.paimonFields) {
+                    this.paimonFieldMap.put(f.name(), f);
+                }
+                this.targetTypes = buildTargetTypes(columnNames);
+                this.writer = table.newBatchWriteBuilder().newWrite();
+                if (table instanceof org.apache.paimon.table.FileStoreTable) {
+                    this.isDynamicBucketMode =
+                            ((org.apache.paimon.table.FileStoreTable) table).schema().numBuckets() == -1;
+                }
+                boolean spillEnabled = Boolean.parseBoolean(options.getOrDefault("write-buffer-spillable", "false"));
+                if (spillEnabled) {
+                    String spillDir = options.get("paimon_jni_spill_dir");
+                    if (spillDir != null && !spillDir.isEmpty()) {
+                        File spillFile = new File(spillDir);
+                        if (!spillFile.exists()) {
+                            spillFile.mkdirs();
+                        }
+                        this.ioManager = new IOManagerImpl(spillDir);
+                    } else {
+                        this.ioManager = new IOManagerImpl(System.getProperty("java.io.tmpdir"));
+                    }
+                    long globalPoolSize = Long.parseLong(options.getOrDefault(
+                            "paimon_global_memory_pool_size", "1073741824"));
+                    this.memorySegmentPool = new HeapMemorySegmentPool(globalPoolSize, 32 * 1024);
+                    this.writer.withIOManager(ioManager).withMemoryPool(memorySegmentPool);
+                    LOG.info("paimon: spill enabled, spill_dir={}", spillDir);
+                }
+                return null;
+            } catch (Throwable t) {
+                throw contextException("open", "options_size=" + options.size(), t);
+            }
+        });
+    }
+
+    private void copyIfPresent(Map<String, String> options, Map<String, String> dynamicOptions, String key) {
+        String value = options.get(key);
+        if (value != null) {
+            dynamicOptions.put(key, value);
+        }
+    }
+
+    /**
+     * Receives Arrow IPC memory address from C++, deserializes and writes to Paimon
+     */
+    public void write(long address, int length) throws Exception {
+        preExecutionAuthenticator.execute(() -> {
+            try {
+                ByteBuffer directBuffer = getDirectBuffer(address, length);
+                try (ArrowStreamReader reader = new ArrowStreamReader(
+                        new DirectBufInputStream(directBuffer), allocator)) {
+                    VectorSchemaRoot root = reader.getVectorSchemaRoot();
+                    while (reader.loadNextBatch()) {
+                        writeBatch(root);
+                    }
+                }
+                return null;
+            } catch (Throwable t) {
+                throw contextException("write", "address=" + address + ", length=" + length, t);
+            }
+        });
+    }
+
+    private void writeBatch(VectorSchemaRoot root) throws Exception {
+        int rowCount = root.getRowCount();
+        if (rowCount == 0) {
+            return;
+        }
+        List<Field> fields = root.getSchema().getFields();
+        List<FieldVector> vectors = root.getFieldVectors();
+        int colCount = fields.size();
+        DataType[] currentTargetTypes = resolveTargetTypes(fields);
+        GenericRow reusedRow = new GenericRow(colCount);
+        for (int i = 0; i < rowCount; i++) {
+            for (int col = 0; col < colCount; col++) {
+                Field arrowField = fields.get(col);
+                try {
+                    reusedRow.setField(col, readArrowValue(vectors.get(col), i, arrowField, currentTargetTypes[col]));
+                } catch (Throwable t) {
+                    String fieldName = arrowField == null ? "null" : arrowField.getName();
+                    String currentTargetType = currentTargetTypes[col] == null
+                            ? "null"
+                            : currentTargetTypes[col].asSQLString();
+                    throw contextException("writeBatch.convert",
+                            "row=" + i + ", col=" + col + ", field=" + fieldName + ", targetType="
+                                    + currentTargetType + ", rowCount=" + rowCount + ", colCount=" + colCount,
+                            t);
+                }
+            }
+            try {
+                if (isDynamicBucketMode) {
+                    writer.write(reusedRow, 0);
+                } else {
+                    writer.write(reusedRow);
+                }
+            } catch (Throwable t) {
+                throw contextException("writeBatch.write",
+                        "row=" + i + ", rowCount=" + rowCount + ", colCount=" + colCount,
+                        t);
+            }
+        }
+    }
+
+    private Object readArrowValue(FieldVector vector, int row, Field arrowField, DataType targetType) {
+        if (vector == null || vector.isNull(row)) {
+            return null;
+        }
+        if (targetType instanceof BinaryType || targetType instanceof VarBinaryType) {
+            if (vector instanceof VarBinaryVector) {
+                return ((VarBinaryVector) vector).get(row);
+            }
+            if (vector instanceof VarCharVector) {
+                return ((VarCharVector) vector).get(row);
+            }
+        } else {
+            if (vector instanceof VarCharVector) {
+                return BinaryString.fromBytes(((VarCharVector) vector).get(row));
+            }
+        }
+        if (vector instanceof BitVector) {
+            return ((BitVector) vector).get(row) == 1;
+        }
+        if (vector instanceof TinyIntVector) {
+            return ((TinyIntVector) vector).get(row);
+        }
+        if (vector instanceof SmallIntVector) {
+            return ((SmallIntVector) vector).get(row);
+        }
+        if (vector instanceof IntVector) {
+            return ((IntVector) vector).get(row);
+        }
+        if (vector instanceof BigIntVector) {
+            return ((BigIntVector) vector).get(row);
+        }
+        if (vector instanceof Float4Vector) {
+            return ((Float4Vector) vector).get(row);
+        }
+        if (vector instanceof Float8Vector) {
+            return ((Float8Vector) vector).get(row);
+        }
+        if (vector instanceof DateDayVector) {
+            return ((DateDayVector) vector).get(row);
+        }
+        if (vector instanceof TimeStampMicroVector) {
+            NullableTimeStampMicroHolder holder = new NullableTimeStampMicroHolder();
+            ((TimeStampMicroVector) vector).get(row, holder);
+            return Timestamp.fromMicros(holder.value);
+        }
+        if (vector instanceof DecimalVector) {
+            return readDecimal((DecimalVector) vector, row);
+        }
+        Object val = vector.getObject(row);
+        return convertToPaimonType(val, arrowField, targetType);
+    }
+
+    private Decimal readDecimal(DecimalVector vector, int row) {
+        BigDecimal bd = getBigDecimalFromArrowBuf(vector.getDataBuffer(), row,
+                vector.getScale(), DecimalVector.TYPE_WIDTH);
+        return Decimal.fromBigDecimal(bd, vector.getPrecision(), vector.getScale());
+    }
+
+    private Object convertToPaimonType(Object val, Field arrowField, DataType targetType) {
+        if (val == null) {
+            return null;
+        }
+
+        if (targetType instanceof BinaryType || targetType instanceof VarBinaryType) {
+            if (val instanceof byte[]) {
+                return val;
+            } else if (val instanceof BinaryString) {
+                return ((BinaryString) val).toBytes();
+            } else if (val instanceof org.apache.arrow.vector.util.Text) {
+                return ((org.apache.arrow.vector.util.Text) val).copyBytes();
+            } else if (val instanceof org.apache.hadoop.io.Text) {
+                org.apache.hadoop.io.Text t = (org.apache.hadoop.io.Text) val;
+                byte[] bytes = new byte[t.getLength()];
+                System.arraycopy(t.getBytes(), 0, bytes, 0, t.getLength());
+                return bytes;
+            } else if (val instanceof String) {
+                return ((String) val).getBytes(java.nio.charset.StandardCharsets.UTF_8);
+            } else {
+                return val.toString().getBytes(java.nio.charset.StandardCharsets.UTF_8);
+            }
+        }
+
+        if (val instanceof BinaryString) {
+            return val;
+        }
+        if (val instanceof byte[]) {
+            return BinaryString.fromBytes((byte[]) val);
+        }
+        if (val instanceof org.apache.arrow.vector.util.Text) {
+            return BinaryString.fromBytes(((org.apache.arrow.vector.util.Text) val).copyBytes());
+        }
+        if (val instanceof org.apache.hadoop.io.Text) {
+            org.apache.hadoop.io.Text t = (org.apache.hadoop.io.Text) val;
+            return BinaryString.fromBytes(t.getBytes(), 0, t.getLength());
+        }
+        if (val instanceof CharSequence) {
+            return BinaryString.fromString(val.toString());
+        }
+
+
+        ArrowType arrowType = arrowField != null ? arrowField.getType() : null;
+        ArrowType.ArrowTypeID typeID = arrowType != null ? arrowType.getTypeID() : null;
+
+        if (val instanceof LocalDateTime) {
+            return Timestamp.fromLocalDateTime((LocalDateTime) val);
+        }
+        if (val instanceof Long && typeID == ArrowType.ArrowTypeID.Timestamp) {
+            return Timestamp.fromMicros((Long) val);
+        }
+        if (val instanceof java.time.LocalDate) {
+            return (int) ((java.time.LocalDate) val).toEpochDay();
+        }
+        if (val instanceof Integer && typeID == ArrowType.ArrowTypeID.Date) {
+            return val;
+        }
+        if (val instanceof BigDecimal) {
+            BigDecimal bd = (BigDecimal) val;
+            return Decimal.fromBigDecimal(bd, bd.precision(), bd.scale());
+        }
+        if (targetType instanceof RowType && val instanceof Map) {
+            RowType rowType = (RowType) targetType;
+            Map<?, ?> mapVal = (Map<?, ?>) val;
+            List<DataField> childFields = rowType.getFields();
+            GenericRow structRow = new GenericRow(childFields.size());
+
+            for (int i = 0; i < childFields.size(); i++) {
+                DataField childField = childFields.get(i);
+                Object childVal = mapVal.get(childField.name());
+
+                Field childArrowField = null;
+                if (arrowField != null && arrowField.getChildren() != null) {
+                    for (Field f : arrowField.getChildren()) {
+                        if (f.getName().equals(childField.name())) {
+                            childArrowField = f;
+                            break;
+                        }
+                    }
+                }
+                structRow.setField(i, convertToPaimonType(childVal, childArrowField, childField.type()));
+            }
+            return structRow;
+        }
+        if (targetType instanceof MapType && val instanceof List) {
+            MapType mapType = (MapType) targetType;
+            List<?> list = (List<?>) val;
+            Map<Object, Object> convertedMap = new HashMap<>();
+
+            Field keyArrowField = null;
+            Field valueArrowField = null;
+            if (arrowField != null && !arrowField.getChildren().isEmpty()) {
+                Field entriesStruct = arrowField.getChildren().get(0);
+                if (entriesStruct.getChildren().size() >= 2) {
+                    keyArrowField = entriesStruct.getChildren().get(0);
+                    valueArrowField = entriesStruct.getChildren().get(1);
+                }
+            }
+
+            String keyName = keyArrowField != null ? keyArrowField.getName() : "key";
+            String valueName = valueArrowField != null ? valueArrowField.getName() : "value";
+
+            for (Object element : list) {
+                if (element instanceof Map) { // Arrow 的内部 Struct 会被转成 Java Map
+                    Map<?, ?> kvStruct = (Map<?, ?>) element;
+                    Object k = convertToPaimonType(kvStruct.get(keyName), keyArrowField, mapType.getKeyType());
+                    Object v = convertToPaimonType(kvStruct.get(valueName), valueArrowField, mapType.getValueType());
+                    convertedMap.put(k, v);
+                }
+            }
+            return new GenericMap(convertedMap);
+        }
+        if (targetType instanceof ArrayType && val instanceof List) {
+            ArrayType arrayType = (ArrayType) targetType;
+            List<?> list = (List<?>) val;
+            Object[] convertedArray = new Object[list.size()];
+            Field childArrowField = (arrowField != null && !arrowField.getChildren().isEmpty())
+                    ? arrowField.getChildren().get(0) : null;
+
+            for (int i = 0; i < list.size(); i++) {
+                convertedArray[i] = convertToPaimonType(list.get(i), childArrowField, arrayType.getElementType());
+            }
+            return new GenericArray(convertedArray);
+        }
+
+        if (val instanceof byte[]) {
+            return val;
+        }
+
+        return val;
+    }
+
+    public byte[][] prepareCommit() throws Exception {
+        if (writer == null) {
+            return new byte[0][];
+        }
+        return preExecutionAuthenticator.execute(() -> {
+            try {
+                List<CommitMessage> messages = writer.prepareCommit();
+
+                if (messages == null || messages.isEmpty()) {
+                    LOG.info("paimon: prepareCommit returns empty, location={}", tableLocation);
+                    return new byte[0][];
+                }
+                LOG.info("paimon: prepareCommit returns {} messages", messages.size());
+                final int maxPayloadBytes = 8 * 1024 * 1024;
+                int chunkSize = 512;
+                java.util.ArrayList<byte[]> payloads = new java.util.ArrayList<>();
+                int i = 0;
+                while (i < messages.size()) {
+                    int end = Math.min(i + chunkSize, messages.size());
+                    DataOutputSerializer outputView = new DataOutputSerializer(1024);
+                    serializer.serializeList(messages.subList(i, end), outputView);
+                    byte[] data = outputView.getCopyOfBuffer();
+                    int len = data.length;
+                    int version = serializer.getVersion();
+                    byte[] payload = new byte[12 + len];
+                    payload[0] = 'D';
+                    payload[1] = 'P';
+                    payload[2] = 'C';
+                    payload[3] = 'M';
+                    payload[4] = (byte) ((version >>> 24) & 0xFF);
+                    payload[5] = (byte) ((version >>> 16) & 0xFF);
+                    payload[6] = (byte) ((version >>> 8) & 0xFF);
+                    payload[7] = (byte) (version & 0xFF);
+                    payload[8] = (byte) ((len >>> 24) & 0xFF);
+                    payload[9] = (byte) ((len >>> 16) & 0xFF);
+                    payload[10] = (byte) ((len >>> 8) & 0xFF);
+                    payload[11] = (byte) (len & 0xFF);
+                    System.arraycopy(data, 0, payload, 12, len);
+                    if (payload.length > maxPayloadBytes && chunkSize > 1) {
+                        chunkSize = Math.max(1, chunkSize / 2);
+                        continue;
+                    }
+                    payloads.add(payload);
+                    i = end;
+                }
+                return payloads.toArray(new byte[0][]);
+            } catch (Throwable t) {
+                throw contextException("prepareCommit", "tableLocation=" + tableLocation, t);
+            }
+        });
+    }
+
+    public void abort() {
+        try {
+            if (preExecutionAuthenticator != null) {
+                preExecutionAuthenticator.execute(() -> {
+                    closeWriterResources(false);
+                    return null;
+                });
+            }
+        } catch (Exception e) {
+            LOG.error("Failed to abort paimon writer", e);
+        }
+    }
+
+    public void close() {
+        try {
+            if (preExecutionAuthenticator != null) {
+                preExecutionAuthenticator.execute(() -> {
+                    closeWriterResources(true);
+                    return null;
+                });
+            }
+        } catch (Exception e) {
+            LOG.warn("Error while closing PaimonJniWriter", e);

Review Comment:
   This swallows failures from `closeWriterResources(true)`, so the JNI caller treats Java close as successful even if `BatchTableWrite.close()` fails. At this point BE may already have collected commit payloads, and FE can still commit the Paimon transaction with messages from a writer that did not close cleanly. Please propagate at least `writer.close()` failures back through JNI/C++ so the insert fails instead of committing a potentially incomplete write.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["Gabriel39 (via GitHub)" <gi...@apache.org>]

Gabriel39 commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4393482296

   /review


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3202188098


##########
be/src/exec/CMakeLists.txt:
##########
@@ -23,6 +23,17 @@ set(EXECUTABLE_OUTPUT_PATH "${BUILD_DIR}/src/exec")
 
 file(GLOB_RECURSE EXEC_FILES CONFIGURE_DEPENDS *.cpp)
 
+if (ENABLE_PAIMON_CPP)

Review Comment:
   Fixed. The common Paimon sink/operator writer sources are now built regardless of ENABLE_PAIMON_CPP, while the paimon-cpp dependent native implementation files remain behind the feature gate. This keeps ENABLE_PAIMON_CPP=OFF builds linkable.
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4448467318

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4450237137

   run p0


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3226068409


##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2238,7 +2289,7 @@ Status PipelineFragmentContext::send_report(bool done) {
     // a internal cancellation being processed
     // When limit is reached the fragment is also cancelled, but _is_report_on_cancel will
     // be set to false, to avoid sending fault report to FE.
-    if (!_is_report_success && !_is_report_on_cancel) {

Review Comment:
   感谢提醒，这里确实不应该感知具体 writer 类型。我已经把这部分改动去掉了，send_report() 的取消上报逻辑保持原有行为。Paimon commit message 的收集现在只放在正常的 report status 参数填充流程里处理。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3226051875


##########
be/src/exprs/function/simple_function_factory.h:
##########
@@ -117,6 +117,9 @@ void register_function_bit_test(SimpleFunctionFactory& factory);
 void register_function_dict_get(SimpleFunctionFactory& factory);
 void register_function_dict_get_many(SimpleFunctionFactory& factory);
 void register_function_ai(SimpleFunctionFactory& factory);
+namespace vectorized {

Review Comment:
   这个是从其他分支适配过来时误带的，已经删掉多余 namespace，并按 4.1 当前命名空间风格对齐。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4428851156

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4450121429

   # FE Regression Coverage Report
   Increment line coverage `  65.10% (360/553)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_94845913004923f36a1e28983b2c859453552e6e_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_94845913004923f36a1e28983b2c859453552e6e_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4447166568

   # Cloud UT Coverage Report
   Increment line coverage ` ` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/217a176440b5baddb646aef46d8f763aac735867_217a176440b5baddb646aef46d8f763aac735867_cloud/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/217a176440b5baddb646aef46d8f763aac735867_217a176440b5baddb646aef46d8f763aac735867_cloud/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 78.11% (1848/2366) |
   | Line Coverage     | 64.91% (33254/51232) |
   | Region Coverage   | 65.40% (16459/25167) |
   | Branch Coverage   | 55.96% (8792/15710) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4432470393

   # FE Regression Coverage Report
   Increment line coverage `  64.46% (361/560)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4405978412

   run p0


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4406179468

   run External


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4406798101

   # FE Regression Coverage Report
   Increment line coverage `  7.67% (45/587)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_09b7591b25385c938209a19f4825ba70b7f2a20f_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_09b7591b25385c938209a19f4825ba70b7f2a20f_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3202190627


##########
be/src/service/CMakeLists.txt:
##########
@@ -43,6 +43,7 @@ if (${MAKE_TEST} STREQUAL "OFF" AND ${BUILD_BENCHMARK} STREQUAL "OFF")
 
     target_link_libraries(doris_be
         ${DORIS_LINK_LIBS}

Review Comment:
   Fixed. Removed the unconditional paimon_rawlog_compat link from be/src/service/CMakeLists.txt; it is now only pulled through DORIS_LINK_LIBS under the existing ENABLE_PAIMON_CPP gate.
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4304610061

   # FE Regression Coverage Report
   Increment line coverage `  3.57% (45/1261)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_b82d48ce0c47b786660f4d4a313c04094ec00076_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_b82d48ce0c47b786660f4d4a313c04094ec00076_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4302178252

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4412158362

   # BE Regression && UT Coverage Report
   Increment line coverage `25.36% (561/2212)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_6a002645fe579dfc2e1cb2731413ad70493a1981_merge/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_6a002645fe579dfc2e1cb2731413ad70493a1981_merge/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 71.41% (26444/37029) |
   | Line Coverage     | 54.43% (280362/515112) |
   | Region Coverage   | 51.79% (233306/450497) |
   | Branch Coverage   | 53.06% (100711/189807) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4451142994

   # FE Regression Coverage Report
   Increment line coverage `  65.10% (360/553)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_94845913004923f36a1e28983b2c859453552e6e_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_94845913004923f36a1e28983b2c859453552e6e_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4439186865

   # FE UT Coverage Report
   Increment line coverage `  50.27% (278/553)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_f1152d99acade5af865fc3f092e94845e4789508/fe_increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_f1152d99acade5af865fc3f092e94845e4789508/fe_report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4334911268

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4324919645

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4325401133

   # FE UT Coverage Report
   Increment line coverage `  51.38% (297/578)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_a2b279283bda7583cb2c741263107f2611c38b5e/fe_increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_a2b279283bda7583cb2c741263107f2611c38b5e/fe_report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4319566335

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4319966822

   # BE UT Coverage Report
   Increment line coverage `5.67% (118/2081)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/8ff21d356c47ba3ef5454bc7288dc67dcd2fb031_8ff21d356c47ba3ef5454bc7288dc67dcd2fb031/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/8ff21d356c47ba3ef5454bc7288dc67dcd2fb031_8ff21d356c47ba3ef5454bc7288dc67dcd2fb031/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 53.08% (20034/37744) |
   | Line Coverage     | 36.57% (188869/516509) |
   | Region Coverage   | 32.95% (146914/445857) |
   | Branch Coverage   | 33.99% (64240/188997) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4320026943

   # BE Regression && UT Coverage Report
   Increment line coverage `6.54% (136/2081)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_8ff21d356c47ba3ef5454bc7288dc67dcd2fb031_merge/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_8ff21d356c47ba3ef5454bc7288dc67dcd2fb031_merge/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 71.30% (26347/36950) |
   | Line Coverage     | 54.31% (279582/514769) |
   | Region Coverage   | 51.71% (232570/449761) |
   | Branch Coverage   | 52.99% (100411/189503) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["morningman (via GitHub)" <gi...@apache.org>]

morningman commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4393461985

   run buildall


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["FayneBupt (via GitHub)" <gi...@apache.org>]

FayneBupt commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3226087058


##########
be/src/runtime/fragment_mgr.cpp:
##########
@@ -346,6 +346,289 @@ void FragmentMgr::stop() {
     }
 }
 
+std::string FragmentMgr::to_http_path(const std::string& file_name) {
+    std::stringstream url;
+    url << "http://" << BackendOptions::get_localhost() << ":" << config::webserver_port
+        << "/api/_download_load?"
+        << "token=" << _exec_env->token() << "&file=" << file_name;
+    return url.str();
+}
+
+Status FragmentMgr::trigger_pipeline_context_report(
+        const ReportStatusRequest req, std::shared_ptr<PipelineFragmentContext>&& ctx) {
+    return _thread_pool->submit_func([this, req, ctx]() {
+        SCOPED_ATTACH_TASK(ctx->get_query_ctx()->query_mem_tracker());
+        coordinator_callback(req);
+        if (!req.done) {
+            ctx->refresh_next_report_time();
+        }
+    });
+}
+
+// There can only be one of these callbacks in-flight at any moment, because
+// it is only invoked from the executor's reporting thread.
+// Also, the reported status will always reflect the most recent execution status,
+// including the final status when execution finishes.
+void FragmentMgr::coordinator_callback(const ReportStatusRequest& req) {

Review Comment:
   感谢提醒，这里确实不应该改 FragmentMgr。这部分是分支适配时带进来的重复逻辑，我已经去掉了。当前 PR 不再修改 be/src/runtime/fragment_mgr.cpp / fragment_mgr.h，Paimon commit message 仍然走现有 pipeline report status 流程处理。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4431608945

   # BE UT Coverage Report
   Increment line coverage `27.51% (532/1934)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 53.38% (20228/37896) |
   | Line Coverage     | 36.92% (191002/517407) |
   | Region Coverage   | 33.22% (148471/446995) |
   | Branch Coverage   | 34.30% (64997/189520) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4432181590

   # FE Regression Coverage Report
   Increment line coverage `  64.46% (361/560)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_merge_fe/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_803d3de3e9caca3b40f777c11ae2f6ed7c4cf608_merge_fe/report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4415284900

   # BE Regression && UT Coverage Report
   Increment line coverage `25.36% (561/2212)` :tada:
   
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_6a002645fe579dfc2e1cb2731413ad70493a1981_merge/increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_6a002645fe579dfc2e1cb2731413ad70493a1981_merge/report/index.html)
   | Category          | Coverage           |
   |-------------------|--------------------|
   | Function Coverage | 71.41% (26441/37029) |
   | Line Coverage     | 54.40% (280240/515112) |
   | Region Coverage   | 51.76% (233186/450497) |
   | Branch Coverage   | 53.04% (100682/189807) |


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["yiguolei (via GitHub)" <gi...@apache.org>]

yiguolei commented on code in PR #62689:
URL: https://github.com/apache/doris/pull/62689#discussion_r3216740424


##########
be/src/pipeline/exec/paimon_table_sink_operator.cpp:
##########
@@ -0,0 +1,73 @@
+// 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 "pipeline/exec/paimon_table_sink_operator.h"
+
+#include "common/status.h"
+#include "vec/sink/vpaimon_jni_table_writer.h"
+#include "vec/sink/vpaimon_table_writer.h"
+
+namespace doris {
+
+bool should_use_paimon_jni_writer(const TDataSink& tsink) {
+    if (!tsink.__isset.paimon_table_sink) {
+        return false;
+    }
+    const auto& options = tsink.paimon_table_sink.options;
+    auto it = options.find("paimon_use_jni");
+    if (it == options.end()) {
+        return false;
+    }
+    return it->second == "true" || it->second == "1";

Review Comment:
   这里 先trim 一下，另外比较的时候忽略大小写



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org

Re: [PR] [feat](paimon) support paimon insert by jni&paimon-cpp [doris]

["hello-stephen (via GitHub)" <gi...@apache.org>]

hello-stephen commented on PR #62689:
URL: https://github.com/apache/doris/pull/62689#issuecomment-4302584543

   # FE UT Coverage Report
   Increment line coverage `  35.81% (207/578)` :tada:
   [Increment coverage report](http://coverage.selectdb-in.cc/coverage/62689_b82d48ce0c47b786660f4d4a313c04094ec00076/fe_increment_report/index.html)
   [Complete coverage report](http://coverage.selectdb-in.cc/coverage/62689_b82d48ce0c47b786660f4d4a313c04094ec00076/fe_report/index.html)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org