This is an automated email from the ASF dual-hosted git repository. suxiaogang223 pushed a commit to branch feature/paimon-jni-write-v1 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 33f709dc1918d3ba83e1c70ae7cfafbf3659f804 Author: Socrates <[email protected]> AuthorDate: Wed Jul 8 15:48:30 2026 +0800 format --- .../exec/operator/paimon_table_sink_operator.cpp | 4 +- be/src/exec/operator/paimon_table_sink_operator.h | 2 +- be/src/exec/pipeline/pipeline_fragment_context.cpp | 8 +-- .../writer/paimon/ffi_paimon_write_backend.cpp | 5 +- .../writer/paimon/jni_paimon_write_backend.cpp | 76 +++++++++------------- .../sink/writer/paimon/jni_paimon_write_backend.h | 3 +- .../exec/sink/writer/paimon/paimon_write_backend.h | 9 ++- .../writer/paimon/paimon_write_backend_factory.cpp | 11 ++-- .../sink/writer/paimon/vpaimon_table_writer.cpp | 26 +++----- .../exec/sink/writer/paimon/vpaimon_table_writer.h | 4 +- be/src/runtime/runtime_state.h | 7 +- 11 files changed, 63 insertions(+), 92 deletions(-) diff --git a/be/src/exec/operator/paimon_table_sink_operator.cpp b/be/src/exec/operator/paimon_table_sink_operator.cpp index 9c3ae8a19aa..165d89a801b 100644 --- a/be/src/exec/operator/paimon_table_sink_operator.cpp +++ b/be/src/exec/operator/paimon_table_sink_operator.cpp @@ -23,8 +23,8 @@ namespace doris { Status PaimonTableSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) { RETURN_IF_ERROR(Base::init(state, info)); - _writer = std::make_shared<VPaimonTableWriter>( - info.tsink, _output_vexpr_ctxs, _dependency, _finish_dependency); + _writer = std::make_shared<VPaimonTableWriter>(info.tsink, _output_vexpr_ctxs, _dependency, + _finish_dependency); return Status::OK(); } diff --git a/be/src/exec/operator/paimon_table_sink_operator.h b/be/src/exec/operator/paimon_table_sink_operator.h index 4de0fa74d3b..c1c19d16644 100644 --- a/be/src/exec/operator/paimon_table_sink_operator.h +++ b/be/src/exec/operator/paimon_table_sink_operator.h @@ -46,7 +46,7 @@ class PaimonTableSinkOperatorX final : public DataSinkOperatorX<PaimonTableSinkL public: using Base = DataSinkOperatorX<PaimonTableSinkLocalState>; PaimonTableSinkOperatorX(ObjectPool* pool, int operator_id, const RowDescriptor& row_desc, - const std::vector<TExpr>& t_output_expr) + const std::vector<TExpr>& t_output_expr) : Base(operator_id, 0, 0), _row_desc(row_desc), _t_output_expr(t_output_expr), diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp b/be/src/exec/pipeline/pipeline_fragment_context.cpp index 1eeed47c442..634c964ce3b 100644 --- a/be/src/exec/pipeline/pipeline_fragment_context.cpp +++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp @@ -1376,8 +1376,8 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS if (!thrift_sink.__isset.paimon_table_sink) { return Status::InternalError("Missing paimon table sink."); } - _sink = std::make_shared<PaimonTableSinkOperatorX>(pool, next_sink_operator_id(), - row_desc, output_exprs); + _sink = std::make_shared<PaimonTableSinkOperatorX>(pool, next_sink_operator_id(), row_desc, + output_exprs); break; } case TDataSinkType::JDBC_TABLE_SINK: { @@ -2527,13 +2527,13 @@ void PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r if (auto pcm = req.runtime_state->paimon_commit_messages(); !pcm.empty()) { params.__isset.paimon_commit_messages = true; params.paimon_commit_messages.insert(params.paimon_commit_messages.end(), pcm.begin(), - pcm.end()); + pcm.end()); } else if (!req.runtime_states.empty()) { for (auto* rs : req.runtime_states) { if (auto rs_pcm = rs->paimon_commit_messages(); !rs_pcm.empty()) { params.__isset.paimon_commit_messages = true; params.paimon_commit_messages.insert(params.paimon_commit_messages.end(), - rs_pcm.begin(), rs_pcm.end()); + rs_pcm.begin(), rs_pcm.end()); } } } diff --git a/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.cpp b/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.cpp index f69b0311ad9..9760cbdadda 100644 --- a/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.cpp +++ b/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.cpp @@ -27,12 +27,11 @@ Status FfiPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* s } Status FfiPaimonWriteBackend::create_writer(const std::string& partition_bytes, int32_t bucket, - std::unique_ptr<IPaimonWriter>* writer) { + std::unique_ptr<IPaimonWriter>* writer) { return Status::NotSupported("FFI backend: create_writer not yet implemented"); } -Status FfiPaimonWriteBackend::create_committer( - std::unique_ptr<IPaimonCommitter>* committer) { +Status FfiPaimonWriteBackend::create_committer(std::unique_ptr<IPaimonCommitter>* committer) { return Status::NotSupported("FFI backend: create_committer not yet implemented"); } diff --git a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp index 66ed1fc4f2f..a16fe5f2ee5 100644 --- a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp +++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp @@ -35,10 +35,8 @@ namespace doris { // JniPaimonWriteBackend // ──────────────────────────────────────────────────────────── -static constexpr const char* PAIMON_JNI_WRITER_CLASS = - "org/apache/doris/paimon/PaimonJniWriter"; -static constexpr const char* PAIMON_OUTPUT_COLUMN_NAMES_KEY = - "doris.output_column_names"; +static constexpr const char* PAIMON_JNI_WRITER_CLASS = "org/apache/doris/paimon/PaimonJniWriter"; +static constexpr const char* PAIMON_OUTPUT_COLUMN_NAMES_KEY = "doris.output_column_names"; static constexpr char PAIMON_COLUMN_NAME_SEPARATOR = '\x01'; JniPaimonWriteBackend::~JniPaimonWriteBackend() { @@ -74,11 +72,10 @@ Status JniPaimonWriteBackend::_get_jni_env(JNIEnv** env) { return Status::OK(); } -Status JniPaimonWriteBackend::_check_jni_exception(JNIEnv* env, - const std::string& method_name) { +Status JniPaimonWriteBackend::_check_jni_exception(JNIEnv* env, const std::string& method_name) { if (env->ExceptionCheck()) { - Status st = Jni::Env::GetJniExceptionMsg( - env, true, "JNI exception in " + method_name + ": "); + Status st = + Jni::Env::GetJniExceptionMsg(env, true, "JNI exception in " + method_name + ": "); LOG(WARNING) << st.to_string(); return st; } @@ -100,13 +97,11 @@ static std::vector<std::string> _split_column_names(const std::string& column_na return names; } -static jobject _to_java_options(JNIEnv* env, - const std::map<std::string, std::string>& options) { +static jobject _to_java_options(JNIEnv* env, const std::map<std::string, std::string>& options) { jclass map_cls = env->FindClass("java/util/HashMap"); jmethodID map_ctor = env->GetMethodID(map_cls, "<init>", "()V"); jmethodID put_method = env->GetMethodID( - map_cls, "put", - "(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;"); + map_cls, "put", "(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;"); jobject map_obj = env->NewObject(map_cls, map_ctor); for (const auto& kv : options) { @@ -131,15 +126,14 @@ Status JniPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* s jclass local_cls = env->FindClass(PAIMON_JNI_WRITER_CLASS); RETURN_IF_ERROR(_check_jni_exception(env, "FindClass")); if (local_cls == nullptr) { - return Status::InternalError("Failed to find Java class: {}", - PAIMON_JNI_WRITER_CLASS); + return Status::InternalError("Failed to find Java class: {}", PAIMON_JNI_WRITER_CLASS); } _jni_writer_cls = static_cast<jclass>(env->NewGlobalRef(local_cls)); env->DeleteLocalRef(local_cls); // Cache method IDs _open_id = env->GetMethodID(_jni_writer_cls, "open", - "(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;)V"); + "(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;)V"); _write_id = env->GetMethodID(_jni_writer_cls, "write", "(JI)V"); _prepare_commit_id = env->GetMethodID(_jni_writer_cls, "prepareCommit", "()[[B"); _abort_id = env->GetMethodID(_jni_writer_cls, "abort", "()V"); @@ -176,8 +170,8 @@ Status JniPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* s jclass string_cls = env->FindClass("java/lang/String"); jobjectArray j_cols = nullptr; if (sink.__isset.column_names && !sink.column_names.empty()) { - j_cols = env->NewObjectArray(static_cast<jsize>(sink.column_names.size()), - string_cls, nullptr); + j_cols = env->NewObjectArray(static_cast<jsize>(sink.column_names.size()), string_cls, + nullptr); for (size_t i = 0; i < sink.column_names.size(); ++i) { jstring str = env->NewStringUTF(sink.column_names[i].c_str()); env->SetObjectArrayElement(j_cols, static_cast<jsize>(i), str); @@ -203,7 +197,7 @@ Status JniPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* s } Status JniPaimonWriteBackend::create_writer(const std::string& partition_bytes, int32_t bucket, - std::unique_ptr<IPaimonWriter>* writer) { + std::unique_ptr<IPaimonWriter>* writer) { DCHECK(_opened) << "Backend must be opened before creating writers"; JNIEnv* env = nullptr; RETURN_IF_ERROR(_get_jni_env(&env)); @@ -218,8 +212,7 @@ Status JniPaimonWriteBackend::create_writer(const std::string& partition_bytes, return Status::OK(); } -Status JniPaimonWriteBackend::create_committer( - std::unique_ptr<IPaimonCommitter>* committer) { +Status JniPaimonWriteBackend::create_committer(std::unique_ptr<IPaimonCommitter>* committer) { auto c = std::make_unique<JniPaimonCommitter>(_sink); *committer = std::move(c); return Status::OK(); @@ -229,11 +222,10 @@ Status JniPaimonWriteBackend::create_committer( // JniPaimonWriter // ──────────────────────────────────────────────────────────── -JniPaimonWriter::JniPaimonWriter(JNIEnv* env, jobject jni_writer_obj, - jmethodID write_id, jmethodID prepare_commit_id, - jmethodID abort_id, - std::unique_ptr<ArrowMemoryPool<>> arrow_pool, - const TPaimonTableSink& sink) +JniPaimonWriter::JniPaimonWriter(JNIEnv* env, jobject jni_writer_obj, jmethodID write_id, + jmethodID prepare_commit_id, jmethodID abort_id, + std::unique_ptr<ArrowMemoryPool<>> arrow_pool, + const TPaimonTableSink& sink) : _env(env), _jni_writer_obj(jni_writer_obj), _write_id(write_id), @@ -268,8 +260,8 @@ Status JniPaimonWriter::_write_projected_block(Block& block) { RETURN_IF_ERROR(get_arrow_schema_from_block(block, &arrow_schema, "UTC")); std::shared_ptr<arrow::RecordBatch> record_batch; - RETURN_IF_ERROR(convert_to_arrow_batch(block, arrow_schema, _arrow_pool.get(), - &record_batch, cctz::utc_time_zone())); + RETURN_IF_ERROR(convert_to_arrow_batch(block, arrow_schema, _arrow_pool.get(), &record_batch, + cctz::utc_time_zone())); auto out_stream_res = arrow::io::BufferOutputStream::Create(); if (!out_stream_res.ok()) { @@ -313,8 +305,8 @@ Status JniPaimonWriter::_write_projected_block(Block& block) { auto address = reinterpret_cast<jlong>(buffer->data()); jint length = static_cast<jint>(buffer->size()); env->CallVoidMethod(_jni_writer_obj, _write_id, address, length); - RETURN_IF_ERROR(Jni::Env::GetJniExceptionMsg( - env, false, "JNI exception in JniPaimonWriter::write: ")); + RETURN_IF_ERROR( + Jni::Env::GetJniExceptionMsg(env, false, "JNI exception in JniPaimonWriter::write: ")); return Status::OK(); } @@ -335,8 +327,7 @@ Status JniPaimonWriter::prepare_commit(std::vector<TPaimonCommitMessage>& messag } jobject j_payloads_obj = env->CallObjectMethod(_jni_writer_obj, _prepare_commit_id); - Status st = Jni::Env::GetJniExceptionMsg( - env, false, "JNI exception in prepareCommit: "); + Status st = Jni::Env::GetJniExceptionMsg(env, false, "JNI exception in prepareCommit: "); if (!st.ok()) { return st; } @@ -349,8 +340,7 @@ Status JniPaimonWriter::prepare_commit(std::vector<TPaimonCommitMessage>& messag jsize num_payloads = env->GetArrayLength(j_payloads); for (jsize i = 0; i < num_payloads; ++i) { - jbyteArray j_bytes = static_cast<jbyteArray>( - env->GetObjectArrayElement(j_payloads, i)); + jbyteArray j_bytes = static_cast<jbyteArray>(env->GetObjectArrayElement(j_payloads, i)); if (j_bytes == nullptr) { continue; } @@ -358,8 +348,7 @@ Status JniPaimonWriter::prepare_commit(std::vector<TPaimonCommitMessage>& messag if (len > 0) { jbyte* bytes = env->GetByteArrayElements(j_bytes, nullptr); if (bytes != nullptr) { - std::string payload(reinterpret_cast<char*>(bytes), - static_cast<size_t>(len)); + std::string payload(reinterpret_cast<char*>(bytes), static_cast<size_t>(len)); TPaimonCommitMessage msg; msg.__set_payload(payload); messages.emplace_back(std::move(msg)); @@ -390,8 +379,7 @@ Status JniPaimonWriter::abort() { } } env->CallVoidMethod(_jni_writer_obj, _abort_id); - return Jni::Env::GetJniExceptionMsg( - env, true, "JNI exception in abort: "); + return Jni::Env::GetJniExceptionMsg(env, true, "JNI exception in abort: "); } // ──────────────────────────────────────────────────────────── @@ -419,28 +407,24 @@ Status JniPaimonCommitter::commit(const std::vector<TPaimonCommitMessage>& messa return _commit_impl(messages, false, nullptr); } -Status JniPaimonCommitter::overwrite( - const std::vector<TPaimonCommitMessage>& messages, - const std::map<std::string, std::string>& static_partition) { +Status JniPaimonCommitter::overwrite(const std::vector<TPaimonCommitMessage>& messages, + const std::map<std::string, std::string>& static_partition) { return _commit_impl(messages, true, &static_partition); } Status JniPaimonCommitter::truncate_table() { - return Status::NotSupported( - "truncate_table is handled by FE Coordinator"); + return Status::NotSupported("truncate_table is handled by FE Coordinator"); } Status JniPaimonCommitter::truncate_partitions( const std::vector<std::map<std::string, std::string>>& partitions) { - return Status::NotSupported( - "truncate_partitions is handled by FE Coordinator"); + return Status::NotSupported("truncate_partitions is handled by FE Coordinator"); } Status JniPaimonCommitter::abort(const std::vector<TPaimonCommitMessage>& messages) { // Abort is best-effort: delete the data files. // In the current architecture, FE handles abort through PaimonTransaction. - return Status::NotSupported( - "abort is handled by FE Coordinator"); + return Status::NotSupported("abort is handled by FE Coordinator"); } } // namespace doris diff --git a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h index 19944c9cb0e..f3f47ce3db6 100644 --- a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h +++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h @@ -94,8 +94,7 @@ class JniPaimonWriter final : public IPaimonWriter { public: JniPaimonWriter(JNIEnv* env, jobject jni_writer_obj, jmethodID write_id, jmethodID prepare_commit_id, jmethodID abort_id, - std::unique_ptr<ArrowMemoryPool<>> arrow_pool, - const TPaimonTableSink& sink); + std::unique_ptr<ArrowMemoryPool<>> arrow_pool, const TPaimonTableSink& sink); ~JniPaimonWriter() override; Status write(RuntimeState* state, Block& block) override; 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 f35debf0e9a..b46d4717309 100644 --- a/be/src/exec/sink/writer/paimon/paimon_write_backend.h +++ b/be/src/exec/sink/writer/paimon/paimon_write_backend.h @@ -35,8 +35,8 @@ class RuntimeState; // ──────────────────────────────────────────────────────────── enum class PaimonBackendType { - JNI, // Java SDK via JNI - FFI // Rust via C FFI + JNI, // Java SDK via JNI + FFI // Rust via C FFI }; // ──────────────────────────────────────────────────────────── @@ -77,9 +77,8 @@ public: virtual Status commit(const std::vector<TPaimonCommitMessage>& messages) = 0; /// Commit in OVERWRITE mode, optionally scoped to static partitions. - virtual Status overwrite( - const std::vector<TPaimonCommitMessage>& messages, - const std::map<std::string, std::string>& static_partition) = 0; + virtual Status overwrite(const std::vector<TPaimonCommitMessage>& messages, + const std::map<std::string, std::string>& static_partition) = 0; /// Truncate the entire table. virtual Status truncate_table() = 0; 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 d289eb71591..61cda74725a 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 @@ -15,15 +15,14 @@ // specific language governing permissions and limitations // under the License. -#include "exec/sink/writer/paimon/paimon_write_backend.h" - #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/paimon_write_backend.h" namespace doris { Status PaimonWriteBackendFactory::create(const TPaimonTableSink& sink, - std::unique_ptr<IPaimonWriteBackend>* backend) { + std::unique_ptr<IPaimonWriteBackend>* backend) { PaimonBackendType type = select_backend_type(sink); switch (type) { @@ -36,12 +35,10 @@ Status PaimonWriteBackendFactory::create(const TPaimonTableSink& sink, return Status::OK(); } } - return Status::InternalError("Unknown Paimon backend type: {}", - static_cast<int>(type)); + return Status::InternalError("Unknown Paimon backend type: {}", static_cast<int>(type)); } -PaimonBackendType PaimonWriteBackendFactory::select_backend_type( - const TPaimonTableSink& sink) { +PaimonBackendType PaimonWriteBackendFactory::select_backend_type(const TPaimonTableSink& sink) { // v1: Always use JNI backend. // // v2: When Rust FFI backend is ready, use the following logic: diff --git a/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp index d312821c9cc..8d76c484a53 100644 --- a/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp +++ b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp @@ -23,16 +23,14 @@ namespace doris { -static constexpr const char* PAIMON_OUTPUT_COLUMN_NAMES_KEY = - "doris.output_column_names"; +static constexpr const char* PAIMON_OUTPUT_COLUMN_NAMES_KEY = "doris.output_column_names"; static constexpr char PAIMON_COLUMN_NAME_SEPARATOR = '\x01'; 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) { + 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); } @@ -51,8 +49,7 @@ Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) { _open_timer = ADD_TIMER(_operator_profile, "OpenTime"); _close_timer = ADD_TIMER(_operator_profile, "CloseTime"); _prepare_commit_timer = ADD_TIMER(_operator_profile, "PrepareCommitTime"); - _serialize_commit_messages_timer = - ADD_TIMER(_operator_profile, "SerializeCommitMessagesTime"); + _serialize_commit_messages_timer = ADD_TIMER(_operator_profile, "SerializeCommitMessagesTime"); _commit_payload_count = ADD_COUNTER(_operator_profile, "CommitPayloadCount", TUnit::UNIT); _commit_payload_bytes_counter = ADD_COUNTER(_operator_profile, "CommitPayloadBytes", TUnit::BYTES); @@ -61,8 +58,7 @@ Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) { SCOPED_TIMER(_open_timer); // Create the backend via factory - RETURN_IF_ERROR( - PaimonWriteBackendFactory::create(_t_sink.paimon_table_sink, &_backend)); + RETURN_IF_ERROR(PaimonWriteBackendFactory::create(_t_sink.paimon_table_sink, &_backend)); RETURN_IF_ERROR(_backend->open(_t_sink.paimon_table_sink, state)); // Create a single writer (the actual per-partition-bucket routing is @@ -104,16 +100,14 @@ Status VPaimonTableWriter::write(RuntimeState* state, Block& block) { // If the block is already large enough, send it directly. // Otherwise, buffer it to reduce JNI call overhead. - if (output_block.rows() >= BATCH_MAX_ROWS || - output_block.bytes() >= BATCH_MAX_BYTES) { + if (output_block.rows() >= BATCH_MAX_ROWS || output_block.bytes() >= BATCH_MAX_BYTES) { RETURN_IF_ERROR(_flush_buffer()); return _write_projected_block(output_block); } // Check if appending would overflow the buffer - if (_buffered_rows > 0 && - (_buffered_rows + output_block.rows() >= BATCH_MAX_ROWS || - _buffered_bytes + output_block.bytes() >= BATCH_MAX_BYTES)) { + if (_buffered_rows > 0 && (_buffered_rows + output_block.rows() >= BATCH_MAX_ROWS || + _buffered_bytes + output_block.bytes() >= BATCH_MAX_BYTES)) { RETURN_IF_ERROR(_flush_buffer()); } diff --git a/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h index 3f169a22c6d..173ab42c0ce 100644 --- a/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h +++ b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h @@ -62,7 +62,7 @@ class RuntimeState; class VPaimonTableWriter final : public AsyncResultWriter { public: VPaimonTableWriter(const TDataSink& t_sink, const VExprContextSPtrs& output_exprs, - std::shared_ptr<Dependency> dep, std::shared_ptr<Dependency> fin_dep); + std::shared_ptr<Dependency> dep, std::shared_ptr<Dependency> fin_dep); ~VPaimonTableWriter() override = default; @@ -100,7 +100,7 @@ private: std::unique_ptr<Block> _buffer; size_t _buffered_rows = 0; size_t _buffered_bytes = 0; - static constexpr size_t BATCH_MAX_ROWS = 32768; // 32K rows + static constexpr size_t BATCH_MAX_ROWS = 32768; // 32K rows static constexpr size_t BATCH_MAX_BYTES = 4 * 1024 * 1024; // 4 MB // Statistics diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h index ac2be22ff06..22259533a79 100644 --- a/be/src/runtime/runtime_state.h +++ b/be/src/runtime/runtime_state.h @@ -548,11 +548,10 @@ public: return _paimon_commit_messages; } - void add_paimon_commit_messages( - const std::vector<TPaimonCommitMessage>& commit_messages) { + void add_paimon_commit_messages(const std::vector<TPaimonCommitMessage>& commit_messages) { std::lock_guard<std::mutex> lock(_paimon_commit_messages_mutex); - _paimon_commit_messages.insert(_paimon_commit_messages.end(), - commit_messages.begin(), commit_messages.end()); + _paimon_commit_messages.insert(_paimon_commit_messages.end(), commit_messages.begin(), + commit_messages.end()); } // local runtime filter mgr, the runtime filter do not have remote target or --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
