diff --git a/be/src/exec/sink/writer/paimon/native_paimon_write_backend.cpp b/be/src/exec/sink/writer/paimon/native_paimon_write_backend.cpp
new file mode 100644
index 00000000000000..b5a9baad231e6b
--- /dev/null
+++ b/be/src/exec/sink/writer/paimon/native_paimon_write_backend.cpp
@@ -0,0 +1,276 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "exec/sink/writer/paimon/native_paimon_write_backend.h"
+
+#include <arrow/type.h>
+#include <fmt/format.h>
+
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "core/block/materialize_block.h"
+#include "format/table/paimon/arrow_schema_util.h"
+#include "format/transformer/vparquet_transformer.h"
+#include "io/file_factory.h"
+#include "io/fs/file_system.h"
+#include "runtime/runtime_state.h"
+#include "util/uid_util.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+
+namespace {
+
+Status parquet_compression(TFileCompressType::type compression,
+                           TParquetCompressionType::type* result) {
+    switch (compression) {
+    case TFileCompressType::PLAIN:
+        *result = TParquetCompressionType::UNCOMPRESSED;
+        return Status::OK();
+    case TFileCompressType::SNAPPYBLOCK:
+        *result = TParquetCompressionType::SNAPPY;
+        return Status::OK();
+    case TFileCompressType::ZSTD:
+        *result = TParquetCompressionType::ZSTD;
+        return Status::OK();
+    case TFileCompressType::LZ4BLOCK:
+        *result = TParquetCompressionType::LZ4_HADOOP;
+        return Status::OK();
+    case TFileCompressType::GZ:
+        *result = TParquetCompressionType::GZIP;
+        return Status::OK();
+    default:
+        return Status::NotSupported("Unsupported Paimon Parquet compression {}",
+                                    to_string(compression));
+    }
+}
+
+class NativePaimonWriter final : public IPaimonWriter {
+public:
+    NativePaimonWriter(TPaimonTableSink sink, RuntimeState* state)
+            : _sink(std::move(sink)),
+              _info(_sink.native_write_info),
+              _state(state),
+              _writer_uuid(generate_uuid_string()) {}
+
+    Status write(RuntimeState*, Block& block) override {
+        if (_prepared) {
+            return Status::InternalError("Cannot write after preparing a Paimon native commit");
+        }
+        if (block.rows() == 0) {
+            return Status::OK();
+        }
+        materialize_block_inplace(block);
+        if (_arrow_schema == nullptr) {
+            RETURN_IF_ERROR(paimon::ArrowSchemaUtil::convert(_info.schema, block,
+                                                             _state->timezone(), &_arrow_schema));
+        }
+        if (_transformer != nullptr &&
+            _transformer->written_len() >= _info.target_file_size_bytes) {
+            RETURN_IF_ERROR(_close_current_file());
+        }
+        if (_transformer == nullptr) {
+            RETURN_IF_ERROR(_open_next_file());
+        }
+        RETURN_IF_ERROR(_transformer->write(block));
+        _current_row_count += block.rows();
+        _next_sequence_number += block.rows();
+        return Status::OK();
+    }
+
+    Status prepare_commit(std::vector<TPaimonCommitMessage>& messages) override {
+        if (_prepared) {
+            return Status::InternalError("Paimon native commit was already prepared");
+        }
+        RETURN_IF_ERROR(_close_current_file());
+        messages.reserve(messages.size() + _commit_data.size());
+        for (const auto& data : _commit_data) {
+            TPaimonCommitMessage message;
+            message.__set_native_commit_data(data);
+            messages.emplace_back(std::move(message));
+        }
+        _prepared = true;
+        return Status::OK();
+    }
+
+    Status abort() override {
+        Status result = Status::OK();
+        _transformer.reset();
+        _file_writer.reset();
+        if (_fs != nullptr) {
+            for (const auto& path : _created_files) {
+                Status st = _fs->delete_file(path);
+                if (!st.ok()) {
+                    LOG(WARNING) << "Failed to delete uncommitted native Paimon file " << path
+                                 << ": " << st;
+                    if (result.ok()) {
+                        result = st;
+                    }
+                }
+            }
+        }
+        _created_files.clear();
+        _commit_data.clear();
+        return result;
+    }
+
+private:
+    Status _ensure_file_system() {
+        if (_fs != nullptr) {
+            return Status::OK();
+        }
+        io::FSPropertiesRef fs_properties(_info.file_type);
+        fs_properties.properties = &_sink.hadoop_config;
+        if (_info.__isset.broker_addresses && !_info.broker_addresses.empty()) {
+            fs_properties.broker_addresses = &_info.broker_addresses;
+        }
+        io::FileDescription description = {.path = _info.output_path, .fs_name = {}};
+        _fs = DORIS_TRY(FileFactory::create_fs(fs_properties, description));
+        return _fs->create_directory(_info.output_path, false);
+    }
+
+    Status _open_next_file() {
+        RETURN_IF_ERROR(_ensure_file_system());
+        _current_file_name =
+                fmt::format("{}{}-{}.parquet", _info.data_file_prefix, _writer_uuid, _file_index++);
+        _current_path = fmt::format("{}/{}", _info.output_path, _current_file_name);
+        io::FileWriterOptions writer_options = {.used_by_s3_committer = false};
+        RETURN_IF_ERROR(_fs->create_file(_current_path, &_file_writer, &writer_options));
+        _created_files.emplace_back(_current_path);
+
+        TParquetCompressionType::type compression;
+        RETURN_IF_ERROR(parquet_compression(_info.compression_type, &compression));
+        ParquetFileOptions parquet_options = {.compression_type = compression,
+                                              .parquet_version = TParquetVersion::PARQUET_1_0,
+                                              .parquet_disable_dictionary = false,
+                                              .enable_int96_timestamps = false,
+                                              .store_decimal_as_integer = true};
+        _transformer = std::make_unique<VParquetTransformer>(_state, _file_writer.get(),
+                                                             VExprContextSPtrs {}, _arrow_schema,
+                                                             false, parquet_options);
+        Status st = _transformer->open();
+        if (!st.ok()) {
+            _transformer.reset();
+            _file_writer.reset();
+            Status delete_st = _fs->delete_file(_current_path);
+            if (!delete_st.ok()) {
+                LOG(WARNING) << "Failed to clean up Paimon file after open failure: " << delete_st;
+            }
+            _created_files.pop_back();
+            return st;
+        }
+        _current_row_count = 0;
+        _current_min_sequence_number = _next_sequence_number;
+        return Status::OK();
+    }
+
+    Status _close_current_file() {
+        if (_transformer == nullptr) {
+            return Status::OK();
+        }
+        Status st = _transformer->close();
+        const int64_t file_size = _transformer->written_len();
+        _transformer.reset();
+        _file_writer.reset();
+        if (!st.ok()) {
+            Status delete_st = _fs->delete_file(_current_path);
+            if (!delete_st.ok()) {
+                LOG(WARNING) << "Failed to clean up Paimon file after close failure: " << delete_st;
+            }
+            return st;
+        }
+        if (_current_row_count == 0) {
+            RETURN_IF_ERROR(_fs->delete_file(_current_path));
+            return Status::OK();
+        }
+
+        TPaimonNativeCommitData data;
+        data.__set_file_name(_current_file_name);
+        data.__set_file_size(file_size);
+        data.__set_row_count(_current_row_count);
+        data.__set_min_sequence_number(_current_min_sequence_number);
+        data.__set_max_sequence_number(_next_sequence_number - 1);
+        data.__set_schema_id(_info.schema.schema_id);
+        data.__set_bucket(0);
+        data.__set_total_buckets(-1);
+        _commit_data.emplace_back(std::move(data));
+        return Status::OK();
+    }
+
+    TPaimonTableSink _sink;
+    const TPaimonNativeWriteInfo& _info;
+    RuntimeState* _state;
+    std::string _writer_uuid;
+    int _file_index = 0;
+    int64_t _next_sequence_number = 0;
+    int64_t _current_min_sequence_number = 0;
+    int64_t _current_row_count = 0;
+    bool _prepared = false;
+
+    std::shared_ptr<arrow::Schema> _arrow_schema;
+    std::shared_ptr<io::FileSystem> _fs;
+    std::unique_ptr<io::FileWriter> _file_writer;
+    std::unique_ptr<VParquetTransformer> _transformer;
+    std::string _current_file_name;
+    std::string _current_path;
+    std::vector<std::string> _created_files;
+    std::vector<TPaimonNativeCommitData> _commit_data;
+};
+
+} // namespace
+
+Status NativePaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* state,
+                                      RuntimeProfile*) {
+    if (!sink.__isset.native_write_info) {
+        return Status::InvalidArgument("Paimon native backend requires native_write_info");
+    }
+    const auto& info = sink.native_write_info;
+    if (!info.__isset.schema || !info.schema.__isset.schema_id || !info.__isset.output_path ||
+        info.output_path.empty() || !info.__isset.file_type || !info.__isset.file_format ||
+        info.file_format != TFileFormatType::FORMAT_PARQUET || !info.__isset.compression_type ||
+        !info.__isset.target_file_size_bytes || info.target_file_size_bytes <= 0 ||
+        !info.__isset.data_file_prefix || info.data_file_prefix.empty()) {
+        return Status::InvalidArgument("Incomplete phase-one Paimon native write configuration");
+    }
+    if (!sink.__isset.write_mode || sink.write_mode != TPaimonWriteMode::APPEND) {
+        return Status::NotSupported("Paimon native phase one supports APPEND only");
+    }
+    _sink = sink;
+    _state = state;
+    _opened = true;
+    return Status::OK();
+}
+
+Status NativePaimonWriteBackend::create_writer(std::unique_ptr<IPaimonWriter>* writer) {
+    if (!_opened || _state == nullptr) {
+        return Status::InternalError("Paimon native backend is not open");
+    }
+    *writer = std::make_unique<NativePaimonWriter>(_sink, _state);
+    return Status::OK();
+}
+
+Status NativePaimonWriteBackend::close() {
+    _opened = false;
+    _state = nullptr;
+    return Status::OK();
+}
+
+#include "common/compile_check_end.h"
+} // namespace doris
diff --git a/be/src/exec/sink/writer/paimon/native_paimon_write_backend.h b/be/src/exec/sink/writer/paimon/native_paimon_write_backend.h
new file mode 100644
index 00000000000000..148c3d46ecef89
--- /dev/null
+++ b/be/src/exec/sink/writer/paimon/native_paimon_write_backend.h
@@ -0,0 +1,40 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include "exec/sink/writer/paimon/paimon_write_backend.h"
+
+namespace doris {
+
+/// Phase-one Doris-native Paimon writer. It writes only data files; FE remains
+/// the metadata and snapshot commit coordinator through the Paimon Java SDK.
+class NativePaimonWriteBackend final : public IPaimonWriteBackend {
+public:
+    Status open(const TPaimonTableSink& sink, RuntimeState* state,
+                RuntimeProfile* profile) override;
+    Status create_writer(std::unique_ptr<IPaimonWriter>* writer) override;
+    Status close() override;
+    PaimonBackendType type() const override { return PaimonBackendType::NATIVE; }
+
+private:
+    TPaimonTableSink _sink;
+    RuntimeState* _state = nullptr;
+    bool _opened = false;
+};
+
+} // namespace doris
diff --git a/be/src/exec/sink/writer/paimon/paimon_table_writer.cpp b/be/src/exec/sink/writer/paimon/paimon_table_writer.cpp
index 3e0dd58a06fdc9..73d96a6eea0bac 100644
--- a/be/src/exec/sink/writer/paimon/paimon_table_writer.cpp
+++ b/be/src/exec/sink/writer/paimon/paimon_table_writer.cpp
@@ -48,7 +48,7 @@ Status PaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
 
     SCOPED_TIMER(_open_timer);
 
-    // Step 1: Create the backend (JNI or FFI) based on the sink configuration.
+    // Step 1: Create the JNI, Rust FFI, or Doris-native backend.
     RETURN_IF_ERROR(PaimonWriteBackendFactory::create(_t_sink.paimon_table_sink, &_backend));
     DCHECK(_backend);
     // Step 2: Open the backend — for JNI this loads the Java class and calls PaimonJniWriter.open().
@@ -83,8 +83,8 @@ Status PaimonTableWriter::write(RuntimeState* state, Block& block) {
     state->update_num_rows_load_total(block.rows());
     state->update_num_bytes_load_total(block.bytes());
 
-    // Step 2: Delegate to the backend writer (JNI or FFI). For the JNI path
-    // this converts Block → Arrow RecordBatch → Arrow C Data → Java PaimonJniWriter.
+    // Step 2: Delegate to the selected JNI, Rust FFI, or Doris-native backend.
+    // The JNI path converts Block → Arrow RecordBatch → Arrow C Data → Java.
     DCHECK(_writer);
     {
         SCOPED_TIMER(_file_store_write_timer);
@@ -123,8 +123,7 @@ Status PaimonTableWriter::close(Status status) {
         }
     }
 
-    // The adapter only owns Arrow conversion resources. Release it before closing
-    // the backend, whose Java close is the authoritative SDK shutdown boundary.
+    // Release writer-owned buffers and file handles before closing the backend.
     _writer.reset();
 
     if (_backend) {
@@ -139,13 +138,16 @@ Status PaimonTableWriter::close(Status status) {
     }
 
     // Only a fully prepared and cleanly stopped writer may contribute payloads
-    // to the FE transaction. A Java close failure therefore aborts the Doris
+    // to the FE transaction. A backend close failure therefore aborts the Doris
     // transaction instead of allowing it to commit potentially unsafe output.
     if (status.ok()) {
         COUNTER_UPDATE(_commit_payload_count, static_cast<int64_t>(messages.size()));
         for (const auto& msg : messages) {
-            DORIS_CHECK(msg.__isset.payload);
-            COUNTER_UPDATE(_commit_payload_bytes_counter, static_cast<int64_t>(msg.payload.size()));
+            DORIS_CHECK(msg.__isset.payload || msg.__isset.native_commit_data);
+            if (msg.__isset.payload) {
+                COUNTER_UPDATE(_commit_payload_bytes_counter,
+                               static_cast<int64_t>(msg.payload.size()));
+            }
         }
         if (!messages.empty()) {
             _state->add_paimon_commit_messages(messages);
diff --git a/be/src/exec/sink/writer/paimon/paimon_write_backend.h b/be/src/exec/sink/writer/paimon/paimon_write_backend.h
index 44e2667c7c103a..6262060c3b85b3 100644
--- a/be/src/exec/sink/writer/paimon/paimon_write_backend.h
+++ b/be/src/exec/sink/writer/paimon/paimon_write_backend.h
@@ -31,14 +31,14 @@ class RuntimeState;
 class RuntimeProfile;
 
 enum class PaimonBackendType {
-    JNI, // Java via JNI (PaimonJniWriter)
-    FFI, // Rust via FFI (placeholder, not yet implemented)
+    JNI,    // Java via JNI (PaimonJniWriter)
+    FFI,    // Rust via FFI (placeholder, not yet implemented)
+    NATIVE, // Doris C++ data-file writer; FE commits metadata through Paimon Java
 };
 
-/// Writer contract implemented by one SDK writer adapter. Each
+/// Writer contract implemented by one backend writer adapter. Each
 /// PaimonTableWriter owns one IPaimonWriter, which delegates to the
-/// underlying Paimon SDK (Java JNI or Rust FFI). Partition and bucket
-/// routing happens inside the selected SDK backend.
+/// selected implementation (Java JNI, Rust FFI, or Doris native C++).
 ///
 /// Lifecycle: created by IPaimonWriteBackend::create_writer() after the
 /// backend is opened; used for the duration of one pipeline instance.
@@ -50,19 +50,20 @@ class IPaimonWriter {
     /// For the JNI path: Block → Arrow RecordBatch → Arrow C Data → Java.
     virtual Status write(RuntimeState* state, Block& block) = 0;
 
-    /// Flush all buffered data, close files, and collect serialized commit
-    /// messages (DPCM-framed). Called once at EOS.
+    /// Flush all buffered data, close files, and collect either DPCM-framed
+    /// SDK payloads or structured native file metadata. Called once at EOS.
     virtual Status prepare_commit(std::vector<TPaimonCommitMessage>& messages) = 0;
 
     /// Discard written data files on error. Called when write or prepare_commit fails.
     virtual Status abort() = 0;
 };
 
-/// Backend boundary for creating writers via JNI (Java) or FFI (Rust).
+/// Backend boundary for creating writers via JNI (Java), FFI (Rust), or Doris native C++.
 ///
 /// The backend owns the connection/session to the external runtime:
 /// - JNI: owns the JVM class reference, method IDs, and the Java writer object.
 /// - FFI: (future) owns the Rust FFI handle.
+/// - NATIVE: owns Doris file-system and Arrow Parquet writer state.
 ///
 /// Each backend creates one or more IPaimonWriter adapters that share the
 /// same underlying connection. Snapshot commit is deliberately excluded from
@@ -95,6 +96,7 @@ class IPaimonWriteBackend {
 /// Backend selection is based on TPaimonTableSink.backend_type:
 /// - Default (unset or JNI): JniPaimonWriteBackend
 /// - FFI: FfiPaimonWriteBackend (placeholder for future Rust writer)
+/// - NATIVE: NativePaimonWriteBackend (phase-one append-only Parquet writer)
 class PaimonWriteBackendFactory {
 public:
     /// Create a backend instance based on the sink configuration.
diff --git a/be/src/exec/sink/writer/paimon/paimon_write_backend_factory.cpp b/be/src/exec/sink/writer/paimon/paimon_write_backend_factory.cpp
index 087228abbe5d2b..2b7ff62009b230 100644
--- a/be/src/exec/sink/writer/paimon/paimon_write_backend_factory.cpp
+++ b/be/src/exec/sink/writer/paimon/paimon_write_backend_factory.cpp
@@ -17,6 +17,7 @@
 
 #include "exec/sink/writer/paimon/ffi_paimon_write_backend.h"
 #include "exec/sink/writer/paimon/jni_paimon_write_backend.h"
+#include "exec/sink/writer/paimon/native_paimon_write_backend.h"
 #include "exec/sink/writer/paimon/paimon_write_backend.h"
 
 namespace doris {
@@ -30,6 +31,9 @@ Status PaimonWriteBackendFactory::create(const TPaimonTableSink& sink,
     case PaimonBackendType::FFI:
         *backend = std::make_unique<FfiPaimonWriteBackend>();
         return Status::OK();
+    case PaimonBackendType::NATIVE:
+        *backend = std::make_unique<NativePaimonWriteBackend>();
+        return Status::OK();
     }
     return Status::InternalError("Unknown Paimon write backend");
 }
@@ -38,6 +42,9 @@ PaimonBackendType PaimonWriteBackendFactory::select_backend_type(const TPaimonTa
     if (sink.__isset.backend_type && sink.backend_type == TPaimonWriteBackendType::FFI) {
         return PaimonBackendType::FFI;
     }
+    if (sink.__isset.backend_type && sink.backend_type == TPaimonWriteBackendType::NATIVE) {
+        return PaimonBackendType::NATIVE;
+    }
     return PaimonBackendType::JNI;
 }
 
diff --git a/be/src/format/table/paimon/arrow_schema_util.cpp b/be/src/format/table/paimon/arrow_schema_util.cpp
new file mode 100644
index 00000000000000..aa904ebb24ba1c
--- /dev/null
+++ b/be/src/format/table/paimon/arrow_schema_util.cpp
@@ -0,0 +1,210 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format/table/paimon/arrow_schema_util.h"
+
+#include <arrow/type.h>
+#include <arrow/util/key_value_metadata.h>
+
+#include <limits>
+#include <vector>
+
+#include "core/block/block.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_struct.h"
+#include "format/arrow/arrow_row_batch.h"
+
+namespace doris::paimon {
+#include "common/compile_check_begin.h"
+
+namespace {
+
+constexpr const char* PARQUET_FIELD_ID = "PARQUET:field_id";
+constexpr int32_t STRUCTURED_TYPE_FIELD_ID_BASE = std::numeric_limits<int32_t>::max() / 4;
+constexpr int32_t STRUCTURED_TYPE_FIELD_DEPTH_LIMIT = 1 << 10;
+
+const schema::external::TField* get_field(const schema::external::TFieldPtr& field_ptr) {
+    if (!field_ptr.__isset.field_ptr || field_ptr.field_ptr == nullptr) {
+        return nullptr;
+    }
+    return field_ptr.field_ptr.get();
+}
+
+std::shared_ptr<arrow::KeyValueMetadata> field_id_metadata(int32_t field_id) {
+    return arrow::KeyValueMetadata::Make({PARQUET_FIELD_ID}, {std::to_string(field_id)});
+}
+
+int32_t array_element_id(int32_t anchor_id, int32_t depth) {
+    return STRUCTURED_TYPE_FIELD_ID_BASE + anchor_id * STRUCTURED_TYPE_FIELD_DEPTH_LIMIT + depth;
+}
+
+int32_t map_key_id(int32_t anchor_id, int32_t depth) {
+    return STRUCTURED_TYPE_FIELD_ID_BASE - anchor_id * STRUCTURED_TYPE_FIELD_DEPTH_LIMIT - depth;
+}
+
+int32_t map_value_id(int32_t anchor_id, int32_t depth) {
+    return STRUCTURED_TYPE_FIELD_ID_BASE + anchor_id * STRUCTURED_TYPE_FIELD_DEPTH_LIMIT + depth;
+}
+
+Status convert_field(const DataTypePtr& doris_type, const schema::external::TField& field,
+                     const std::string& name, int32_t field_id, int32_t collection_anchor_id,
+                     int32_t collection_depth, const std::string& timezone, bool force_required,
+                     std::shared_ptr<arrow::Field>* result) {
+    if (!field.__isset.type) {
+        return Status::InvalidArgument("Paimon field '{}' is missing its type", name);
+    }
+
+    const auto nested_type = remove_nullable(doris_type);
+    std::shared_ptr<arrow::DataType> arrow_type;
+    switch (field.type.type) {
+    case TPrimitiveType::ARRAY: {
+        if (nested_type->get_primitive_type() != TYPE_ARRAY || !field.__isset.nestedField ||
+            !field.nestedField.__isset.array_field ||
+            !field.nestedField.array_field.__isset.item_field) {
+            return Status::InvalidArgument("Invalid Paimon ARRAY schema for field '{}'", name);
+        }
+        const auto* item = get_field(field.nestedField.array_field.item_field);
+        if (item == nullptr) {
+            return Status::InvalidArgument("Paimon ARRAY field '{}' has no element", name);
+        }
+        const auto* array_type = assert_cast<const DataTypeArray*>(nested_type.get());
+        const int32_t child_depth = collection_depth + 1;
+        std::shared_ptr<arrow::Field> item_field;
+        RETURN_IF_ERROR(convert_field(array_type->get_nested_type(), *item, "element",
+                                      array_element_id(collection_anchor_id, child_depth),
+                                      collection_anchor_id, child_depth, timezone, false,
+                                      &item_field));
+        arrow_type = arrow::list(std::move(item_field));
+        break;
+    }
+    case TPrimitiveType::MAP: {
+        if (nested_type->get_primitive_type() != TYPE_MAP || !field.__isset.nestedField ||
+            !field.nestedField.__isset.map_field ||
+            !field.nestedField.map_field.__isset.key_field ||
+            !field.nestedField.map_field.__isset.value_field) {
+            return Status::InvalidArgument("Invalid Paimon MAP schema for field '{}'", name);
+        }
+        const auto* key = get_field(field.nestedField.map_field.key_field);
+        const auto* value = get_field(field.nestedField.map_field.value_field);
+        if (key == nullptr || value == nullptr) {
+            return Status::InvalidArgument("Paimon MAP field '{}' has incomplete children", name);
+        }
+        const auto* map_type = assert_cast<const DataTypeMap*>(nested_type.get());
+        const int32_t child_depth = collection_depth + 1;
+        std::shared_ptr<arrow::Field> key_field;
+        std::shared_ptr<arrow::Field> value_field;
+        RETURN_IF_ERROR(convert_field(map_type->get_key_type(), *key, "key",
+                                      map_key_id(collection_anchor_id, child_depth),
+                                      collection_anchor_id, child_depth, timezone, true,
+                                      &key_field));
+        RETURN_IF_ERROR(convert_field(map_type->get_value_type(), *value, "value",
+                                      map_value_id(collection_anchor_id, child_depth),
+                                      collection_anchor_id, child_depth, timezone, false,
+                                      &value_field));
+        arrow_type = std::make_shared<arrow::MapType>(std::move(key_field), std::move(value_field));
+        break;
+    }
+    case TPrimitiveType::STRUCT: {
+        if (nested_type->get_primitive_type() != TYPE_STRUCT || !field.__isset.nestedField ||
+            !field.nestedField.__isset.struct_field ||
+            !field.nestedField.struct_field.__isset.fields) {
+            return Status::InvalidArgument("Invalid Paimon ROW schema for field '{}'", name);
+        }
+        const auto* struct_type = assert_cast<const DataTypeStruct*>(nested_type.get());
+        const auto& children = field.nestedField.struct_field.fields;
+        if (children.size() != struct_type->get_elements().size()) {
+            return Status::InvalidArgument(
+                    "Paimon ROW field '{}' child count {} does not match Doris type {}", name,
+                    children.size(), struct_type->get_elements().size());
+        }
+        std::vector<std::shared_ptr<arrow::Field>> arrow_children;
+        arrow_children.reserve(children.size());
+        for (size_t i = 0; i < children.size(); ++i) {
+            const auto* child = get_field(children[i]);
+            if (child == nullptr || !child->__isset.id || !child->__isset.name) {
+                return Status::InvalidArgument("Paimon ROW field '{}' has an invalid child", name);
+            }
+            std::shared_ptr<arrow::Field> arrow_child;
+            RETURN_IF_ERROR(convert_field(struct_type->get_element(i), *child, child->name,
+                                          child->id, child->id, 0, timezone, false, &arrow_child));
+            arrow_children.emplace_back(std::move(arrow_child));
+        }
+        arrow_type = arrow::struct_(std::move(arrow_children));
+        break;
+    }
+    case TPrimitiveType::VARIANT:
+        return Status::NotSupported("Paimon VARIANT native write is a phase-two feature");
+    case TPrimitiveType::VARBINARY:
+        // Doris represents both STRING and VARBINARY columns with a string column at runtime.
+        // Preserve Paimon's logical BINARY type explicitly instead of inferring UTF8 from the
+        // runtime block type.
+        arrow_type = arrow::binary();
+        break;
+    case TPrimitiveType::TIMESTAMPTZ:
+    case TPrimitiveType::DATETIMEV2: {
+        const int scale = field.type.__isset.scale ? field.type.scale : 0;
+        const auto unit = scale <= 3 ? arrow::TimeUnit::MILLI : arrow::TimeUnit::MICRO;
+        const std::string arrow_timezone =
+                field.type.type == TPrimitiveType::TIMESTAMPTZ ? timezone : "";
+        arrow_type = arrow::timestamp(unit, arrow_timezone);
+        break;
+    }
+    default:
+        RETURN_IF_ERROR(convert_to_arrow_type(doris_type, &arrow_type, timezone));
+        break;
+    }
+
+    const bool nullable = !force_required && (!field.__isset.is_optional || field.is_optional);
+    *result = arrow::field(name, std::move(arrow_type), nullable, field_id_metadata(field_id));
+    return Status::OK();
+}
+
+} // namespace
+
+Status ArrowSchemaUtil::convert(const schema::external::TSchema& schema, const Block& block,
+                                const std::string& timezone,
+                                std::shared_ptr<arrow::Schema>* arrow_schema) {
+    if (!schema.__isset.root_field || !schema.root_field.__isset.fields) {
+        return Status::InvalidArgument("Paimon native writer schema has no root fields");
+    }
+    if (schema.root_field.fields.size() != block.columns()) {
+        return Status::InvalidArgument(
+                "Paimon native writer schema columns {} do not match block columns {}",
+                schema.root_field.fields.size(), block.columns());
+    }
+
+    std::vector<std::shared_ptr<arrow::Field>> fields;
+    fields.reserve(block.columns());
+    for (size_t i = 0; i < block.columns(); ++i) {
+        const auto* field = get_field(schema.root_field.fields[i]);
+        if (field == nullptr || !field->__isset.id || !field->__isset.name) {
+            return Status::InvalidArgument("Paimon native writer has an invalid root field at {}",
+                                           i);
+        }
+        std::shared_ptr<arrow::Field> arrow_field;
+        RETURN_IF_ERROR(convert_field(block.get_by_position(i).type, *field, field->name, field->id,
+                                      field->id, 0, timezone, false, &arrow_field));
+        fields.emplace_back(std::move(arrow_field));
+    }
+    *arrow_schema = arrow::schema(std::move(fields));
+    return Status::OK();
+}
+
+#include "common/compile_check_end.h"
+} // namespace doris::paimon
diff --git a/be/src/format/table/paimon/arrow_schema_util.h b/be/src/format/table/paimon/arrow_schema_util.h
new file mode 100644
index 00000000000000..096888adf89628
--- /dev/null
+++ b/be/src/format/table/paimon/arrow_schema_util.h
@@ -0,0 +1,45 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <arrow/type_fwd.h>
+#include <gen_cpp/ExternalTableSchema_types.h>
+
+#include <memory>
+#include <string>
+
+#include "common/status.h"
+
+namespace doris {
+
+class Block;
+
+namespace paimon {
+
+/// Builds the exact Arrow schema consumed by Arrow's Parquet writer for a
+/// Paimon data file. In particular this preserves Paimon field ids, synthetic
+/// collection ids, nullability, timestamp semantics and short-decimal layout.
+class ArrowSchemaUtil {
+public:
+    static Status convert(const schema::external::TSchema& schema, const Block& block,
+                          const std::string& timezone,
+                          std::shared_ptr<arrow::Schema>* arrow_schema);
+};
+
+} // namespace paimon
+} // namespace doris
diff --git a/be/src/format/transformer/vparquet_transformer.cpp b/be/src/format/transformer/vparquet_transformer.cpp
index 0900b85b94c94c..88eb7d54e74f6d 100644
--- a/be/src/format/transformer/vparquet_transformer.cpp
+++ b/be/src/format/transformer/vparquet_transformer.cpp
@@ -195,6 +195,19 @@ VParquetTransformer::VParquetTransformer(RuntimeState* state, doris::io::FileWri
     _outstream = std::shared_ptr<ParquetOutputStream>(new ParquetOutputStream(file_writer));
 }
 
+VParquetTransformer::VParquetTransformer(RuntimeState* state, doris::io::FileWriter* file_writer,
+                                         const VExprContextSPtrs& output_vexpr_ctxs,
+                                         std::shared_ptr<arrow::Schema> arrow_schema,
+                                         bool output_object_data,
+                                         const ParquetFileOptions& parquet_options)
+        : VFileFormatTransformer(state, output_vexpr_ctxs, output_object_data),
+          _arrow_schema(std::move(arrow_schema)),
+          _parquet_options(parquet_options),
+          _iceberg_schema_json(nullptr),
+          _iceberg_schema(nullptr) {
+    _outstream = std::make_shared<ParquetOutputStream>(file_writer);
+}
+
 VParquetTransformer::VParquetTransformer(RuntimeState* state, doris::io::FileWriter* file_writer,
                                          const VExprContextSPtrs& output_vexpr_ctxs,
                                          std::vector<TParquetSchema> parquet_schemas,
@@ -222,6 +235,9 @@ Status VParquetTransformer::_parse_properties() {
         } else {
             builder.enable_dictionary();
         }
+        if (_parquet_options.store_decimal_as_integer) {
+            builder.enable_store_decimal_as_integer();
+        }
         builder.created_by(
                 fmt::format("{}({})", doris::get_short_version(), ::parquet::DEFAULT_CREATED_BY));
         builder.max_row_group_length(std::numeric_limits<int64_t>::max());
@@ -242,6 +258,9 @@ Status VParquetTransformer::_parse_properties() {
 }
 
 Status VParquetTransformer::_parse_schema() {
+    if (_arrow_schema != nullptr) {
+        return Status::OK();
+    }
     std::vector<std::shared_ptr<arrow::Field>> fields;
     if (_iceberg_schema != nullptr) {
         RETURN_IF_ERROR(
diff --git a/be/src/format/transformer/vparquet_transformer.h b/be/src/format/transformer/vparquet_transformer.h
index e89b9a51435b70..b8f2dd5244d43e 100644
--- a/be/src/format/transformer/vparquet_transformer.h
+++ b/be/src/format/transformer/vparquet_transformer.h
@@ -83,6 +83,8 @@ struct ParquetFileOptions {
     TParquetVersion::type parquet_version;
     bool parquet_disable_dictionary = false;
     bool enable_int96_timestamps = false;
+    // Paimon stores DECIMAL(1..9) as INT32 and DECIMAL(10..18) as INT64.
+    bool store_decimal_as_integer = false;
 };
 
 // a wrapper of parquet output stream
@@ -101,6 +103,13 @@ class VParquetTransformer final : public VFileFormatTransformer {
                         const ParquetFileOptions& parquet_options,
                         const std::string* iceberg_schema_json = nullptr);
 
+    // Table formats may provide the exact Arrow schema when field ids and nested
+    // physical types cannot be derived from Doris expressions alone.
+    VParquetTransformer(RuntimeState* state, doris::io::FileWriter* file_writer,
+                        const VExprContextSPtrs& output_vexpr_ctxs,
+                        std::shared_ptr<arrow::Schema> arrow_schema, bool output_object_data,
+                        const ParquetFileOptions& parquet_options);
+
     ~VParquetTransformer() override = default;
 
     Status open() override;
diff --git a/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp b/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
index 387e8932793338..87ccb4ff2e1cee 100644
--- a/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
+++ b/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
@@ -29,6 +29,9 @@ TEST(PaimonWriteBackendFactoryTest, SelectBackendType) {
 
     sink.__set_backend_type(TPaimonWriteBackendType::FFI);
     EXPECT_EQ(PaimonBackendType::FFI, PaimonWriteBackendFactory::select_backend_type(sink));
+
+    sink.__set_backend_type(TPaimonWriteBackendType::NATIVE);
+    EXPECT_EQ(PaimonBackendType::NATIVE, PaimonWriteBackendFactory::select_backend_type(sink));
 }
 
 TEST(JniPaimonWriteBackendTest, OpenAbiAndWriteModes) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java
index ddf58b0cbdedd6..87b52ac1170221 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java
@@ -20,16 +20,27 @@
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.UserException;
 import org.apache.doris.thrift.TPaimonCommitMessage;
+import org.apache.doris.thrift.TPaimonNativeCommitData;
 import org.apache.doris.transaction.Transaction;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.format.SimpleColStats;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
 import org.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.stats.SimpleStatsConverter;
 import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
 import org.apache.paimon.table.sink.CommitMessageSerializer;
 import org.apache.paimon.table.sink.InnerTableCommit;
+import org.apache.paimon.types.RowType;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -75,6 +86,8 @@ enum CommitState {
 
     private final List<byte[]> commitPayloads = Lists.newArrayList();
     private final Set<CommitPayloadKey> commitPayloadSet = new HashSet<>();
+    private final List<TPaimonNativeCommitData> nativeCommitData = Lists.newArrayList();
+    private final Set<String> nativeFileNames = new HashSet<>();
 
     public PaimonTransaction(PaimonMetadataOps ops, long transactionId) {
         this.ops = Preconditions.checkNotNull(ops, "Paimon metadata ops must not be null");
@@ -99,16 +112,19 @@ public synchronized void bind(PaimonWriteBinding binding) {
     public void commit() throws UserException {
         PaimonWriteBinding writeBinding = requireBinding();
         List<byte[]> rawPayloads = snapshotPayloads();
-        if (rawPayloads.isEmpty() && !writeBinding.isOverwrite()) {
+        List<TPaimonNativeCommitData> nativeFiles = snapshotNativeCommitData();
+        if (rawPayloads.isEmpty() && nativeFiles.isEmpty() && !writeBinding.isOverwrite()) {
             LOG.info("Skip empty PaimonTransaction commit, txnId={}, table={}",
                     transactionId, tableName());
             markPreparedTransactionCommitted();
             return;
         }
         try {
-            List<CommitMessage> allMessages = deserializePayloads(rawPayloads);
-            LOG.info("Commit PaimonTransaction, txnId={}, table={}, payloads={}, messages={}, overwrite={}",
-                    transactionId, tableName(), rawPayloads.size(), allMessages.size(),
+            List<CommitMessage> allMessages = collectCommitMessages(
+                    writeBinding, rawPayloads, nativeFiles);
+            LOG.info("Commit PaimonTransaction, txnId={}, table={}, payloads={}, nativeFiles={}, "
+                            + "messages={}, overwrite={}",
+                    transactionId, tableName(), rawPayloads.size(), nativeFiles.size(), allMessages.size(),
                     writeBinding.isOverwrite());
             if (allMessages.isEmpty() && !writeBinding.isOverwrite()) {
                 throw new RuntimeException(
@@ -136,14 +152,16 @@ public void rollback() {
             return;
         }
         List<byte[]> rawPayloads = snapshotPayloads();
-        if (rawPayloads.isEmpty()) {
+        List<TPaimonNativeCommitData> nativeFiles = snapshotNativeCommitData();
+        if (rawPayloads.isEmpty() && nativeFiles.isEmpty()) {
             LOG.info("Skip empty PaimonTransaction rollback, txnId={}, table={}",
                     transactionId, tableName());
             return;
         }
         try {
             PaimonWriteBinding writeBinding = requireBinding();
-            List<CommitMessage> allMessages = deserializePayloads(rawPayloads);
+            List<CommitMessage> allMessages = collectCommitMessages(
+                    writeBinding, rawPayloads, nativeFiles);
             if (allMessages.isEmpty()) {
                 LOG.info("Skip PaimonTransaction rollback with empty decoded messages, "
                         + "txnId={}, table={}", transactionId, tableName());
@@ -174,17 +192,23 @@ public void updateCommitMessages(List<TPaimonCommitMessage> messages) {
     }
 
     private void addPayload(TPaimonCommitMessage message) {
-        if (message == null || !message.isSetPayload()) {
+        if (message == null) {
             return;
         }
-        byte[] payload = message.getPayload();
-        if (payload == null || payload.length == 0) {
-            return;
+        if (message.isSetNativeCommitData()) {
+            TPaimonNativeCommitData data = message.getNativeCommitData();
+            if (data != null
+                    && (!data.isSetFileName() || nativeFileNames.add(data.getFileName()))) {
+                nativeCommitData.add(data.deepCopy());
+            }
         }
-        // Treat the Thrift payload as immutable after report handling and adopt it directly. The
-        // report is not reused, so copying it would only create a second full representation.
-        if (commitPayloadSet.add(new CommitPayloadKey(payload))) {
-            commitPayloads.add(payload);
+        if (message.isSetPayload()) {
+            byte[] payload = message.getPayload();
+            if (payload != null && payload.length > 0
+                    && commitPayloadSet.add(new CommitPayloadKey(payload))) {
+                // Treat the Thrift payload as immutable after report handling and adopt it directly.
+                commitPayloads.add(payload);
+            }
         }
     }
 
@@ -371,6 +395,83 @@ private List<byte[]> snapshotPayloads() {
         }
     }
 
+    private List<TPaimonNativeCommitData> snapshotNativeCommitData() {
+        synchronized (this) {
+            return new ArrayList<>(nativeCommitData);
+        }
+    }
+
+    private static List<CommitMessage> collectCommitMessages(PaimonWriteBinding writeBinding,
+            List<byte[]> payloads, List<TPaimonNativeCommitData> nativeFiles) throws IOException {
+        List<CommitMessage> messages = deserializePayloads(payloads);
+        if (!nativeFiles.isEmpty()) {
+            messages.addAll(buildNativeCommitMessages(writeBinding, nativeFiles));
+        }
+        return messages;
+    }
+
+    static List<CommitMessage> buildNativeCommitMessages(PaimonWriteBinding writeBinding,
+            List<TPaimonNativeCommitData> nativeFiles) throws IOException {
+        RowType rowType = writeBinding.getTable().rowType();
+        long pinnedSchemaId = writeBinding.getTable().schema().id();
+        SimpleColStats[] noColumnStats = new SimpleColStats[rowType.getFieldCount()];
+        Arrays.fill(noColumnStats, SimpleColStats.NONE);
+        SimpleStats rowStats = new SimpleStatsConverter(rowType).toBinaryAllMode(noColumnStats);
+
+        List<CommitMessage> messages = new ArrayList<>(nativeFiles.size());
+        for (TPaimonNativeCommitData file : nativeFiles) {
+            validateNativeCommitData(file, pinnedSchemaId);
+            DataFileMeta fileMeta = DataFileMeta.forAppend(
+                    file.getFileName(),
+                    file.getFileSize(),
+                    file.getRowCount(),
+                    rowStats,
+                    file.getMinSequenceNumber(),
+                    file.getMaxSequenceNumber(),
+                    file.getSchemaId(),
+                    Collections.emptyList(),
+                    null,
+                    FileSource.APPEND,
+                    null,
+                    null,
+                    null,
+                    null);
+            messages.add(new CommitMessageImpl(
+                    BinaryRow.EMPTY_ROW,
+                    file.getBucket(),
+                    file.getTotalBuckets(),
+                    new DataIncrement(Collections.singletonList(fileMeta),
+                            Collections.emptyList(), Collections.emptyList()),
+                    CompactIncrement.emptyIncrement()));
+        }
+        return messages;
+    }
+
+    private static void validateNativeCommitData(TPaimonNativeCommitData file,
+            long pinnedSchemaId) throws IOException {
+        if (!file.isSetFileName() || file.getFileName().isEmpty()
+                || file.getFileName().contains("/") || file.getFileName().contains("\\")) {
+            throw new IOException("Invalid Paimon native data file name");
+        }
+        if (!file.isSetFileSize() || file.getFileSize() <= 0
+                || !file.isSetRowCount() || file.getRowCount() <= 0
+                || !file.isSetMinSequenceNumber() || file.getMinSequenceNumber() < 0
+                || !file.isSetMaxSequenceNumber()
+                || file.getMaxSequenceNumber() < file.getMinSequenceNumber()
+                || file.getMaxSequenceNumber() - file.getMinSequenceNumber()
+                        != file.getRowCount() - 1) {
+            throw new IOException("Invalid Paimon native data file statistics for "
+                    + file.getFileName());
+        }
+        if (!file.isSetSchemaId() || file.getSchemaId() != pinnedSchemaId) {
+            throw new IOException("Paimon native data file schema id does not match pinned schema");
+        }
+        if (!file.isSetBucket() || file.getBucket() != 0
+                || !file.isSetTotalBuckets() || file.getTotalBuckets() != -1) {
+            throw new IOException("Phase-one Paimon native writer requires bucket=0 and totalBuckets=-1");
+        }
+    }
+
     private static final class CommitPayloadKey {
         private final byte[] payload;
         private final int hashCode;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java
index e20766c1b4e276..117ff790182bb9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java
@@ -21,26 +21,47 @@
 import org.apache.doris.catalog.Column;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.LocationPath;
+import org.apache.doris.datasource.credentials.CredentialUtils;
+import org.apache.doris.datasource.credentials.VendedCredentialsFactory;
+import org.apache.doris.datasource.paimon.PaimonExternalCatalog;
 import org.apache.doris.datasource.paimon.PaimonExternalTable;
 import org.apache.doris.datasource.paimon.PaimonTransaction;
+import org.apache.doris.datasource.paimon.PaimonUtil;
 import org.apache.doris.datasource.paimon.PaimonWriteBinding;
 import org.apache.doris.datasource.paimon.PaimonWriteTarget;
+import org.apache.doris.datasource.property.storage.StorageProperties;
 import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
 import org.apache.doris.nereids.trees.plans.commands.insert.InsertCommandContext;
 import org.apache.doris.nereids.trees.plans.commands.insert.PaimonInsertCommandContext;
+import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.thrift.TDataSink;
 import org.apache.doris.thrift.TDataSinkType;
 import org.apache.doris.thrift.TExplainLevel;
 import org.apache.doris.thrift.TFileFormatType;
+import org.apache.doris.thrift.TFileType;
+import org.apache.doris.thrift.TPaimonNativeWriteInfo;
 import org.apache.doris.thrift.TPaimonTableSink;
 import org.apache.doris.thrift.TPaimonWriteBackendType;
 import org.apache.doris.thrift.TPaimonWriteMode;
 
 import com.google.common.base.Preconditions;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.TimestampType;
 
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
@@ -51,8 +72,8 @@
  * metadata, Hadoop authentication config, transaction identity, write mode,
  * and sink column names.
  *
- * The upstream Exchange may establish concurrent HASH_FIXED writer ownership;
- * partition and bucket handling inside each writer remains delegated to the SDK.
+ * The upstream Exchange may establish concurrent writer ownership. Partition
+ * and fixed-bucket routing remain on the SDK path until later native phases.
  */
 public class PaimonTableSink extends BaseExternalTableDataSink {
     public static final String ROW_KIND_COLUMN = "__DORIS_PAIMON_ROW_KIND__";
@@ -130,7 +151,6 @@ public void bindDataSink(Optional<InsertCommandContext> insertCtx) throws Analys
         // exact table instance instead of loading catalog metadata independently.
         tSink.setSerializedTable(binding.getSerializedTable());
 
-        tSink.setBackendType(TPaimonWriteBackendType.JNI);
         if (isChangelogWrite()) {
             tSink.setWriteMode(TPaimonWriteMode.CHANGELOG);
         } else if (ctx.isOverwrite()) {
@@ -139,7 +159,13 @@ public void bindDataSink(Optional<InsertCommandContext> insertCtx) throws Analys
             tSink.setWriteMode(TPaimonWriteMode.APPEND);
         }
 
-        tSink.setHadoopConfig(binding.getHadoopConfig());
+        if (usePhaseOneNativeWriter(binding.getTable(), ctx)) {
+            tSink.setBackendType(TPaimonWriteBackendType.NATIVE);
+            configurePhaseOneNativeWriter(tSink, binding);
+        } else {
+            tSink.setBackendType(TPaimonWriteBackendType.JNI);
+            tSink.setHadoopConfig(binding.getHadoopConfig());
+        }
 
         tSink.setColumnNames(outputColumnNames);
 
@@ -169,4 +195,169 @@ private boolean isChangelogWrite() {
                 || dmlCommandType == DMLCommandType.MERGE;
     }
 
+    private boolean usePhaseOneNativeWriter(FileStoreTable table,
+            PaimonInsertCommandContext ctx) {
+        ConnectContext connectContext = ConnectContext.get();
+        if ((connectContext != null
+                && !connectContext.getSessionVariable().isPaimonNativeInsertMode())
+                || isChangelogWrite()
+                || ctx.isOverwrite()
+                || !table.primaryKeys().isEmpty()
+                || !table.partitionKeys().isEmpty()
+                || table.bucketMode() != BucketMode.BUCKET_UNAWARE) {
+            return false;
+        }
+
+        CoreOptions options = CoreOptions.fromMap(table.options());
+        String dataFilePrefix = options.dataFilePrefix();
+        if (!CoreOptions.FILE_FORMAT_PARQUET.equals(options.formatType())
+                || options.rowTrackingEnabled()
+                || !isPhaseOneStatsMode(table, options)
+                || options.dataFilePathDirectory() != null
+                || options.fileSuffixIncludeCompression()
+                || !isPhaseOneCompression(options.fileCompression())
+                || dataFilePrefix == null
+                || dataFilePrefix.isEmpty()
+                || dataFilePrefix.contains("/")
+                || dataFilePrefix.contains("\\")) {
+            return false;
+        }
+
+        // Defaults and partial-column inserts are still normalized by the Java writer in phase 1.
+        // Native mode is selected only when the BE receives the complete pinned table schema.
+        List<DataField> fields = table.schema().fields();
+        if (cols.size() != fields.size()) {
+            return false;
+        }
+        for (int i = 0; i < fields.size(); i++) {
+            if (!fields.get(i).name().equalsIgnoreCase(cols.get(i).getName())
+                    || !isPhaseOneNativeType(fields.get(i).type())) {
+                return false;
+            }
+        }
+
+        // Native phase 1 does not emit Paimon sidecar indexes or external data paths.
+        for (String key : table.options().keySet()) {
+            if (key.startsWith("file-index.")
+                    || key.startsWith("index.")
+                    || (key.startsWith("fields.") && key.endsWith(".file-index"))) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private static boolean isPhaseOneStatsMode(FileStoreTable table, CoreOptions options) {
+        if (!"none".equalsIgnoreCase(options.statsMode())) {
+            return false;
+        }
+        for (String levelMode : options.statsModePerLevel().values()) {
+            if (!"none".equalsIgnoreCase(levelMode)) {
+                return false;
+            }
+        }
+        for (Map.Entry<String, String> entry : table.options().entrySet()) {
+            if (entry.getKey().startsWith("fields.")
+                    && entry.getKey().endsWith(".stats-mode")
+                    && !"none".equalsIgnoreCase(entry.getValue())) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private static boolean isPhaseOneNativeType(DataType type) {
+        switch (type.getTypeRoot()) {
+            case BOOLEAN:
+            case TINYINT:
+            case SMALLINT:
+            case INTEGER:
+            case BIGINT:
+            case FLOAT:
+            case DOUBLE:
+            case CHAR:
+            case VARCHAR:
+            case BINARY:
+            case VARBINARY:
+            case DECIMAL:
+            case DATE:
+                return true;
+            case TIMESTAMP_WITHOUT_TIME_ZONE:
+                // Paimon encodes precision > 6 as INT96. Doris DATETIMEV2 is microsecond based,
+                // so keeping those tables on the SDK path avoids silent precision loss.
+                return ((TimestampType) type).getPrecision() <= 6;
+            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+                return ((LocalZonedTimestampType) type).getPrecision() <= 6;
+            case ARRAY:
+                return isPhaseOneNativeType(((ArrayType) type).getElementType());
+            case MAP:
+                MapType map = (MapType) type;
+                return isPhaseOneNativeType(map.getKeyType())
+                        && isPhaseOneNativeType(map.getValueType());
+            case ROW:
+                for (DataField field : ((RowType) type).getFields()) {
+                    if (!isPhaseOneNativeType(field.type())) {
+                        return false;
+                    }
+                }
+                return true;
+            default:
+                // VARIANT is added in phase 2. TIME, MULTISET and BLOB are intentionally
+                // kept on the SDK path until Doris has lossless native representations.
+                return false;
+        }
+    }
+
+    private static boolean isPhaseOneCompression(String compression) {
+        if (compression == null) {
+            return false;
+        }
+        switch (compression.toLowerCase(Locale.ROOT)) {
+            case "none":
+            case "uncompressed":
+            case "snappy":
+            case "zstd":
+            case "lz4":
+            case "gzip":
+                return true;
+            default:
+                return false;
+        }
+    }
+
+    private void configurePhaseOneNativeWriter(TPaimonTableSink tSink, PaimonWriteBinding binding)
+            throws AnalysisException {
+        FileStoreTable table = binding.getTable();
+        CoreOptions options = CoreOptions.fromMap(table.options());
+        PaimonExternalCatalog catalog = (PaimonExternalCatalog) targetTable.getCatalog();
+        Map<StorageProperties.Type, StorageProperties> storagePropertiesMap =
+                VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials(
+                        catalog.getCatalogProperty().getMetastoreProperties(),
+                        catalog.getCatalogProperty().getStoragePropertiesMap(), table);
+        TPaimonNativeWriteInfo nativeInfo = new TPaimonNativeWriteInfo();
+        nativeInfo.setSchema(PaimonUtil.getSchemaInfo(table.schema(), true, true));
+
+        String tableLocation = table.location().toString();
+        String bucketPath = (tableLocation.endsWith("/") ? tableLocation : tableLocation + "/")
+                + "bucket-0";
+        LocationPath locationPath = LocationPath.of(bucketPath, storagePropertiesMap);
+        nativeInfo.setOutputPath(locationPath.toStorageLocation().toString());
+        TFileType fileType = locationPath.getTFileTypeForBE();
+        nativeInfo.setFileType(fileType);
+        nativeInfo.setFileFormat(TFileFormatType.FORMAT_PARQUET);
+        nativeInfo.setCompressionType(getTFileCompressType(options.fileCompression()));
+        nativeInfo.setTargetFileSizeBytes(options.targetFileSize(false));
+        nativeInfo.setDataFilePrefix(options.dataFilePrefix());
+        if (fileType == TFileType.FILE_BROKER) {
+            nativeInfo.setBrokerAddresses(getBrokerAddresses(targetTable.getCatalog().bindBrokerName()));
+        }
+
+        Map<String, String> backendProperties = new java.util.HashMap<>(binding.getHadoopConfig());
+        backendProperties.putAll(
+                CredentialUtils.getBackendPropertiesFromStorageMap(storagePropertiesMap));
+        // Hadoop authentication settings in the transaction binding are also required by HDFS.
+        tSink.setHadoopConfig(backendProperties);
+        tSink.setNativeWriteInfo(nativeInfo);
+    }
+
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index a69a286babd6dd..4baa564518196e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -807,6 +807,10 @@ public String toString() {
 
     public static final String FORCE_JNI_SCANNER = "force_jni_scanner";
 
+    public static final String PAIMON_INSERT_MODE = "paimon_insert_mode";
+    public static final String PAIMON_INSERT_MODE_NATIVE = "native";
+    public static final String PAIMON_INSERT_MODE_JNI = "jni";
+
     public static final String ENABLE_COUNT_PUSH_DOWN_FOR_EXTERNAL_TABLE = "enable_count_push_down_for_external_table";
 
     public static final String FETCH_ALL_FE_FOR_SYSTEM_TABLE = "fetch_all_fe_for_system_table";
@@ -3050,6 +3054,17 @@ public Map<String, Action> getForceEagerAggHintMap() {
             description = {"强制使用 jni 方式读取外表", "Force the use of jni mode to read external table"})
     private boolean forceJniScanner = false;
 
+    @VariableMgr.VarAttr(name = PAIMON_INSERT_MODE,
+            needForward = true,
+            fuzzy = true,
+            checker = "checkPaimonInsertMode",
+            setter = "setPaimonInsertMode",
+            description = {"Paimon 写入实现。native 优先使用原生写入，不支持时回退 JNI；jni 强制使用 JNI",
+                    "Paimon write implementation. native prefers native writing and falls back to JNI when "
+                            + "unsupported; jni forces JNI"},
+            options = {PAIMON_INSERT_MODE_NATIVE, PAIMON_INSERT_MODE_JNI})
+    private String paimonInsertMode = PAIMON_INSERT_MODE_NATIVE;
+
     @VariableMgr.VarAttr(name = ENABLE_COUNT_PUSH_DOWN_FOR_EXTERNAL_TABLE,
             fuzzy = true,
             description = {"对外表启用 count(*) 下推优化", "enable count(*) pushdown optimization for external table"})
@@ -5444,6 +5459,21 @@ public void checkInsertVisibleTimeoutReturnMode(String mode) {
         parseInsertVisibleTimeoutReturnMode(mode);
     }
 
+    public void checkPaimonInsertMode(String mode) {
+        if (!PAIMON_INSERT_MODE_NATIVE.equalsIgnoreCase(mode)
+                && !PAIMON_INSERT_MODE_JNI.equalsIgnoreCase(mode)) {
+            UnsupportedOperationException exception = new UnsupportedOperationException(
+                    PAIMON_INSERT_MODE + " should be one of {'native', 'jni'}, but found " + mode);
+            LOG.warn("Check " + PAIMON_INSERT_MODE + " failed", exception);
+            throw exception;
+        }
+    }
+
+    public void setPaimonInsertMode(String mode) {
+        checkPaimonInsertMode(mode);
+        paimonInsertMode = mode.toLowerCase(Locale.ROOT);
+    }
+
     // Parse the stored string case-insensitively and expose the enum only to business logic.
     private InsertVisibleTimeoutReturnMode parseInsertVisibleTimeoutReturnMode(String mode) {
         if (StringUtils.isEmpty(mode)) {
@@ -6501,6 +6531,14 @@ public boolean isForceJniScanner() {
         return forceJniScanner;
     }
 
+    public String getPaimonInsertMode() {
+        return paimonInsertMode;
+    }
+
+    public boolean isPaimonNativeInsertMode() {
+        return PAIMON_INSERT_MODE_NATIVE.equalsIgnoreCase(paimonInsertMode);
+    }
+
     public String getIgnoreSplitType() {
         return ignoreSplitType;
     }
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonTransactionTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonTransactionTest.java
index d7e9e440cf5e24..cb712c6b4d7d3c 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonTransactionTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonTransactionTest.java
@@ -20,17 +20,23 @@
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
 import org.apache.doris.thrift.TPaimonCommitMessage;
+import org.apache.doris.thrift.TPaimonNativeCommitData;
 
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataIncrement;
 import org.apache.paimon.io.DataOutputViewStreamWrapper;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.table.sink.CommitMessageImpl;
 import org.apache.paimon.table.sink.CommitMessageSerializer;
 import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.SnapshotManager;
 import org.junit.Assert;
 import org.junit.Before;
@@ -42,6 +48,7 @@
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.Callable;
 
 public class PaimonTransactionTest {
@@ -159,6 +166,63 @@ public void testCommitPayloadDedupUsesExactContent() {
         Assert.assertEquals(2, transaction.getPayloadCount());
     }
 
+    @Test
+    public void testBuildNativeCommitMessage() throws Exception {
+        RowType rowType = RowType.of(DataTypes.INT(), DataTypes.STRING());
+        TableSchema schema = Mockito.mock(TableSchema.class);
+        Mockito.when(table.rowType()).thenReturn(rowType);
+        Mockito.when(table.schema()).thenReturn(schema);
+        Mockito.when(schema.id()).thenReturn(7L);
+
+        TPaimonNativeCommitData nativeFile = new TPaimonNativeCommitData()
+                .setFileName("data-writer-0.parquet")
+                .setFileSize(4096)
+                .setRowCount(10)
+                .setMinSequenceNumber(0)
+                .setMaxSequenceNumber(9)
+                .setSchemaId(7)
+                .setBucket(0)
+                .setTotalBuckets(-1);
+        List<CommitMessage> messages = PaimonTransaction.buildNativeCommitMessages(
+                binding, Collections.singletonList(nativeFile));
+
+        Assert.assertEquals(1, messages.size());
+        CommitMessageImpl message = (CommitMessageImpl) messages.get(0);
+        Assert.assertEquals(BinaryRow.EMPTY_ROW, message.partition());
+        Assert.assertEquals(0, message.bucket());
+        Assert.assertEquals(Integer.valueOf(-1), message.totalBuckets());
+        Assert.assertEquals(1, message.newFilesIncrement().newFiles().size());
+        DataFileMeta file = message.newFilesIncrement().newFiles().get(0);
+        Assert.assertEquals("data-writer-0.parquet", file.fileName());
+        Assert.assertEquals(4096, file.fileSize());
+        Assert.assertEquals(10, file.rowCount());
+        Assert.assertEquals(0, file.minSequenceNumber());
+        Assert.assertEquals(9, file.maxSequenceNumber());
+        Assert.assertEquals(7, file.schemaId());
+        Assert.assertEquals(FileSource.APPEND, file.fileSource().get());
+    }
+
+    @Test
+    public void testRejectInvalidNativeSequenceRange() {
+        TableSchema schema = Mockito.mock(TableSchema.class);
+        Mockito.when(table.rowType()).thenReturn(RowType.of(DataTypes.INT()));
+        Mockito.when(table.schema()).thenReturn(schema);
+        Mockito.when(schema.id()).thenReturn(7L);
+        TPaimonNativeCommitData nativeFile = new TPaimonNativeCommitData()
+                .setFileName("data-writer-0.parquet")
+                .setFileSize(4096)
+                .setRowCount(10)
+                .setMinSequenceNumber(0)
+                .setMaxSequenceNumber(8)
+                .setSchemaId(7)
+                .setBucket(0)
+                .setTotalBuckets(-1);
+
+        Assert.assertThrows(java.io.IOException.class,
+                () -> PaimonTransaction.buildNativeCommitMessages(
+                        binding, Collections.singletonList(nativeFile)));
+    }
+
     @Test
     public void testCommitUserIsNamespacedByDorisCluster() {
         Assert.assertEquals(
diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
index 52e7b4c1a437d7..c81bbdc2b628d2 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
@@ -169,6 +169,33 @@ public void testInsertVisibleTimeoutReturnModeDefaultsAndCheckerBranches() {
                 () -> sessionVar.checkInsertVisibleTimeoutReturnMode(""));
     }
 
+    @Test
+    public void testPaimonInsertMode() throws Exception {
+        SessionVariable sessionVar = new SessionVariable();
+        Assertions.assertEquals(SessionVariable.PAIMON_INSERT_MODE_NATIVE,
+                sessionVar.getPaimonInsertMode());
+        Assertions.assertTrue(sessionVar.isPaimonNativeInsertMode());
+
+        VariableMgr.setVar(sessionVar, new SetVar(SetType.SESSION,
+                SessionVariable.PAIMON_INSERT_MODE, new StringLiteral("JNI")));
+        Assertions.assertEquals(SessionVariable.PAIMON_INSERT_MODE_JNI,
+                sessionVar.getPaimonInsertMode());
+        Assertions.assertFalse(sessionVar.isPaimonNativeInsertMode());
+
+        Field field = SessionVariable.class.getDeclaredField("paimonInsertMode");
+        VariableMgr.VarAttr varAttr = field.getAnnotation(VariableMgr.VarAttr.class);
+        Assertions.assertTrue(varAttr.needForward());
+        Assertions.assertArrayEquals(new String[] {
+                SessionVariable.PAIMON_INSERT_MODE_NATIVE,
+                SessionVariable.PAIMON_INSERT_MODE_JNI
+        }, varAttr.options());
+
+        ExceptionChecker.expectThrowsWithMsg(DdlException.class,
+                "paimon_insert_mode should be one of",
+                () -> VariableMgr.setVar(sessionVar, new SetVar(SetType.SESSION,
+                        SessionVariable.PAIMON_INSERT_MODE, new StringLiteral("rust"))));
+    }
+
     @Test
     public void testRuntimeFilterBroadcastJoinProducerNumDescription() throws Exception {
         SessionVariable sessionVar = new SessionVariable();
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index a4e2cda7440bf3..e9c6e823191ace 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -23,6 +23,7 @@ include "Types.thrift"
 include "Descriptors.thrift"
 include "Partitions.thrift"
 include "PlanNodes.thrift"
+include "ExternalTableSchema.thrift"
 
 enum TDataSinkType {
     DATA_STREAM_SINK = 0,
@@ -629,6 +630,8 @@ struct TMaxComputeTableSink {
 enum TPaimonWriteBackendType {
     JNI = 0,
     FFI = 1,
+    // Doris native file writer. FFI remains reserved for paimon-rust.
+    NATIVE = 2,
 }
 
 enum TPaimonWriteMode {
@@ -637,8 +640,38 @@ enum TPaimonWriteMode {
     CHANGELOG = 2,
 }
 
+// One Paimon data file produced by the Doris native writer. FE converts this
+// stable, structured contract into the Paimon-version-specific DataFileMeta and
+// CommitMessage objects before committing the transaction.
+struct TPaimonNativeCommitData {
+    1: optional string file_name
+    2: optional i64 file_size
+    3: optional i64 row_count
+    4: optional i64 min_sequence_number
+    5: optional i64 max_sequence_number
+    6: optional i64 schema_id
+    7: optional i32 bucket
+    8: optional i32 total_buckets
+}
+
 struct TPaimonCommitMessage {
-    1: optional binary payload          // Paimon native CommitMessageSerializer bytes (DPCM-framed)
+    1: optional binary payload          // Paimon CommitMessageSerializer bytes (DPCM-framed)
+    2: optional TPaimonNativeCommitData native_commit_data
+}
+
+// Phase-one native writer contract: append-only, unpartitioned,
+// bucket-unaware Parquet tables with metadata stats disabled. Fields are
+// additive so later phases can add partition and fixed-bucket routing without
+// changing the backend selector.
+struct TPaimonNativeWriteInfo {
+    1: optional ExternalTableSchema.TSchema schema
+    2: optional string output_path       // normalized physical <table>/bucket-0 path
+    3: optional Types.TFileType file_type
+    4: optional PlanNodes.TFileFormatType file_format
+    5: optional PlanNodes.TFileCompressType compression_type
+    6: optional i64 target_file_size_bytes
+    7: optional string data_file_prefix
+    8: optional list<Types.TNetworkAddress> broker_addresses
 }
 
 struct TPaimonTableSink {
@@ -649,6 +682,7 @@ struct TPaimonTableSink {
     5: optional TPaimonWriteMode write_mode
     6: optional i64 transaction_id
     7: optional string commit_user
+    8: optional TPaimonNativeWriteInfo native_write_info
 }
 
 struct TDataSink {