924060929 commented on code in PR #65381:
URL: https://github.com/apache/doris/pull/65381#discussion_r3549497387
##########
gensrc/thrift/FrontendService.thrift:
##########
@@ -325,14 +325,16 @@ struct TReportExecStatusParams {
28: optional list<DataSinks.TIcebergCommitData> iceberg_commit_datas
- 29: optional i64 txn_id
- 30: optional string label
+ 29: optional list<DataSinks.TPaimonCommitMessage> paimon_commit_messages
Review Comment:
๐ด ไธ่ฝๆนๅจๅทฒๆ Thrift ๅญๆฎต็ field idใ่ฟ้ๆ `paimon_commit_messages` ๆพๅฐไบ 29๏ผ็ญไบๆ
`txn_id`(29โ30)ใ`label`(30โ31)ใ`fragment_instance_reports`(31โ32)ใ`mc_commit_datas`(32โ33)ใ`first_error_msg`(33โ34)
ๆดไฝๅ็งปใ
field id ๆฏ Thrift ็็บฟไธ่บซไปฝใ`TReportExecStatusParams` ๆฏๆฌก fragment ไธๆฅ้ฝไผ็จๅฐ๏ผๆปๅจๅ็บง /
ๆทท็ๆฌๆถๆง BE ๅ็ `txn_id`(29) ไผ่ขซๆฐ FE ๅฝๆ `paimon_commit_messages` ่งฃๆ๏ผ็ฑปๅไธ็ฌฆ่ขซไธขๅผ๏ผ๏ผ็ดๆฅ็ ดๅๅ
ๆฌ
Iceberg / Hive / MaxCompute ๅจๅ
็ๆๆๅค่กจๅฏผๅ
ฅ็ถๆไธๆฅใ
่ฏทไฟๆ 29โ33 ๅๆ ทไธๅจ๏ผๆฐๅญๆฎต็จไธไธไธช็ฉบ้ฒ id๏ผ่ฟ้ๅบไธบ 34๏ผใ
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -591,6 +594,21 @@ public PlanFragment
visitPhysicalIcebergTableSink(PhysicalIcebergTableSink<? ext
return rootFragment;
}
+ @Override
+ public PlanFragment visitPhysicalPaimonTableSink(PhysicalPaimonTableSink<?
extends Plan> paimonTableSink,
+ PlanTranslatorContext
context) {
+ PlanFragment rootFragment = paimonTableSink.child().accept(this,
context);
+ rootFragment.setOutputPartition(DataPartition.UNPARTITIONED);
+ List<Expr> outputExprs = Lists.newArrayList();
+ paimonTableSink.getOutput().stream().map(Slot::getExprId)
+ .forEach(exprId ->
outputExprs.add(context.findSlotRef(exprId)));
+ PaimonTableSink sink = new PaimonTableSink(
+ (PaimonExternalTable) paimonTableSink.getTargetTable());
+ rootFragment.setSink(sink);
+ sink.setOutputExprs(outputExprs);
Review Comment:
๐ด ่ฟ้ๅช `setOutputExprs`๏ผ็ผบ `sink.setCols(paimonTableSink.getCols())`ใ
`PaimonTableSink.bindDataSink()` ้ `for (Column col : cols)` ไผๆ ๆกไปถ้ๅ `cols`๏ผ่
`cols` ไธ็ดๆฏ null โ ๆฏๆฌก Paimon insert ๅฟ
NPE๏ผๅๆถ `bucket_key_indices` ๅ ไธบ `cols !=
null` ๅค็ฉบไนๆฐธ่ฟไธไผไธๅใๅ่ Iceberg / Hive sink ็ translator ่กฅไธ setColsใ
##########
be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp:
##########
@@ -0,0 +1,299 @@
+// 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/vpaimon_table_writer.h"
+
+#include "common/logging.h"
+#include "core/block/block.h"
+#include "exec/sink/writer/paimon/paimon_bucket.h"
+#include "runtime/runtime_state.h"
+#include "vec/data_types/data_type_factory.hpp"
+
+namespace doris {
+
+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) {
+ DCHECK(_t_sink.__isset.paimon_table_sink);
+}
+
+Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
+ _state = state;
+ _operator_profile = profile;
+
+ // Register profile counters
+ _written_rows_counter = ADD_COUNTER(_operator_profile, "WrittenRows",
TUnit::UNIT);
+ _written_bytes_counter = ADD_COUNTER(_operator_profile, "WrittenBytes",
TUnit::BYTES);
+ _send_data_timer = ADD_TIMER(_operator_profile, "SendDataTime");
+ _project_timer = ADD_CHILD_TIMER(_operator_profile, "ProjectTime",
"SendDataTime");
+ _bucket_compute_timer = ADD_CHILD_TIMER(_operator_profile,
"BucketComputeTime", "SendDataTime");
+ _arrow_convert_timer = ADD_CHILD_TIMER(_operator_profile,
"ArrowConvertTime", "SendDataTime");
+ _file_store_write_timer =
+ ADD_CHILD_TIMER(_operator_profile, "FileStoreWriteTime",
"SendDataTime");
+ _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");
+ _commit_payload_count = ADD_COUNTER(_operator_profile,
"CommitPayloadCount", TUnit::UNIT);
+ _commit_payload_bytes_counter =
+ ADD_COUNTER(_operator_profile, "CommitPayloadBytes", TUnit::BYTES);
+ _bucket_writer_count = ADD_COUNTER(_operator_profile, "BucketWriterCount",
TUnit::UNIT);
+
+ SCOPED_TIMER(_open_timer);
+
+ // Create the backend via factory
+
RETURN_IF_ERROR(PaimonWriteBackendFactory::create(_t_sink.paimon_table_sink,
&_backend));
+ RETURN_IF_ERROR(_backend->open(_t_sink.paimon_table_sink, state));
+
+ // Determine bucket routing strategy
+ const auto& paimon_sink = _t_sink.paimon_table_sink;
+ _total_buckets = paimon_sink.__isset.bucket_num ? paimon_sink.bucket_num :
1;
+
+ // Bucket routing is enabled when:
+ // 1. num_buckets > 1 (HASH_FIXED mode)
+ // 2. Bucket key columns are specified
+ _bucket_routing_enabled = _total_buckets > 1;
+
+ if (_bucket_routing_enabled && paimon_sink.__isset.column_names) {
+ // The bucket key is determined by Paimon options (bucket-key).
+ // FE passes column names; we need the indices of bucket key columns.
+ // For now, we use the primary key columns as bucket key (Paimon
default).
+ // In production, FE should explicitly pass bucket_key_indices.
+ //
+ // When bucket-key is explicitly configured, FE should populate
+ // TPaimonTableSink.bucket_key_indices.
+ if (paimon_sink.__isset.bucket_key_indices &&
!paimon_sink.bucket_key_indices.empty()) {
+ _bucket_key_indices =
std::vector<int>(paimon_sink.bucket_key_indices.begin(),
+
paimon_sink.bucket_key_indices.end());
+ }
+ }
+
+ LOG(INFO) << "VPaimonTableWriter opened: table=" << paimon_sink.tb_name
+ << ", backend=" << static_cast<int>(_backend->type())
+ << ", total_buckets=" << _total_buckets
+ << ", bucket_routing=" << _bucket_routing_enabled
+ << ", bucket_key_indices=" << _bucket_key_indices.size();
+ return Status::OK();
+}
+
+Status VPaimonTableWriter::write(RuntimeState* state, Block& block) {
+ if (block.rows() == 0) {
+ return Status::OK();
+ }
+
+ SCOPED_TIMER(_send_data_timer);
+
+ // Project: apply output expressions
+ Block output_block;
+ {
+ SCOPED_TIMER(_project_timer);
+ RETURN_IF_ERROR(_projection_block(block, &output_block));
+ }
+
+ // Rename columns using the output column names from the sink config
+ const auto& paimon_sink = _t_sink.paimon_table_sink;
+ if (paimon_sink.__isset.paimon_options) {
+ auto it =
paimon_sink.paimon_options.find(PAIMON_OUTPUT_COLUMN_NAMES_KEY);
+ if (it != paimon_sink.paimon_options.end()) {
+ auto output_names = _split_column_names(it->second);
Review Comment:
๐ด ็ผ่ฏไธ่ฟ๏ผ`_split_column_names` ๆฏๆฌๆไปถ็ `static` ๅฝๆฐ๏ผๅฎไนๅจๆไปถๆซๅฐพ๏ผไฝ่ฟ้ๅจๅฎไนไนๅๅฐฑ่ฐ็จไบใไธๆฒกๆๅๅๅฃฐๆ
โ `use of undeclared identifier`ใ่ฟไนๅฐ่ฏไบ BE ็กฎๅฎ่ฟๆฒก็ผ่ฟใๅ ไธชๅๅๅฃฐๆๆๆๅฎไนๅ็งปๅณๅฏใ
##########
be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp:
##########
@@ -0,0 +1,299 @@
+// 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/vpaimon_table_writer.h"
+
+#include "common/logging.h"
+#include "core/block/block.h"
+#include "exec/sink/writer/paimon/paimon_bucket.h"
+#include "runtime/runtime_state.h"
+#include "vec/data_types/data_type_factory.hpp"
+
+namespace doris {
+
+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) {
+ DCHECK(_t_sink.__isset.paimon_table_sink);
+}
+
+Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
+ _state = state;
+ _operator_profile = profile;
+
+ // Register profile counters
+ _written_rows_counter = ADD_COUNTER(_operator_profile, "WrittenRows",
TUnit::UNIT);
+ _written_bytes_counter = ADD_COUNTER(_operator_profile, "WrittenBytes",
TUnit::BYTES);
+ _send_data_timer = ADD_TIMER(_operator_profile, "SendDataTime");
+ _project_timer = ADD_CHILD_TIMER(_operator_profile, "ProjectTime",
"SendDataTime");
+ _bucket_compute_timer = ADD_CHILD_TIMER(_operator_profile,
"BucketComputeTime", "SendDataTime");
+ _arrow_convert_timer = ADD_CHILD_TIMER(_operator_profile,
"ArrowConvertTime", "SendDataTime");
+ _file_store_write_timer =
+ ADD_CHILD_TIMER(_operator_profile, "FileStoreWriteTime",
"SendDataTime");
+ _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");
+ _commit_payload_count = ADD_COUNTER(_operator_profile,
"CommitPayloadCount", TUnit::UNIT);
+ _commit_payload_bytes_counter =
+ ADD_COUNTER(_operator_profile, "CommitPayloadBytes", TUnit::BYTES);
+ _bucket_writer_count = ADD_COUNTER(_operator_profile, "BucketWriterCount",
TUnit::UNIT);
+
+ SCOPED_TIMER(_open_timer);
+
+ // Create the backend via factory
+
RETURN_IF_ERROR(PaimonWriteBackendFactory::create(_t_sink.paimon_table_sink,
&_backend));
+ RETURN_IF_ERROR(_backend->open(_t_sink.paimon_table_sink, state));
+
+ // Determine bucket routing strategy
+ const auto& paimon_sink = _t_sink.paimon_table_sink;
+ _total_buckets = paimon_sink.__isset.bucket_num ? paimon_sink.bucket_num :
1;
+
+ // Bucket routing is enabled when:
+ // 1. num_buckets > 1 (HASH_FIXED mode)
+ // 2. Bucket key columns are specified
+ _bucket_routing_enabled = _total_buckets > 1;
+
+ if (_bucket_routing_enabled && paimon_sink.__isset.column_names) {
+ // The bucket key is determined by Paimon options (bucket-key).
+ // FE passes column names; we need the indices of bucket key columns.
+ // For now, we use the primary key columns as bucket key (Paimon
default).
+ // In production, FE should explicitly pass bucket_key_indices.
+ //
+ // When bucket-key is explicitly configured, FE should populate
+ // TPaimonTableSink.bucket_key_indices.
+ if (paimon_sink.__isset.bucket_key_indices &&
!paimon_sink.bucket_key_indices.empty()) {
+ _bucket_key_indices =
std::vector<int>(paimon_sink.bucket_key_indices.begin(),
+
paimon_sink.bucket_key_indices.end());
+ }
+ }
+
+ LOG(INFO) << "VPaimonTableWriter opened: table=" << paimon_sink.tb_name
+ << ", backend=" << static_cast<int>(_backend->type())
+ << ", total_buckets=" << _total_buckets
+ << ", bucket_routing=" << _bucket_routing_enabled
+ << ", bucket_key_indices=" << _bucket_key_indices.size();
+ return Status::OK();
+}
+
+Status VPaimonTableWriter::write(RuntimeState* state, Block& block) {
+ if (block.rows() == 0) {
+ return Status::OK();
+ }
+
+ SCOPED_TIMER(_send_data_timer);
+
+ // Project: apply output expressions
+ Block output_block;
+ {
+ SCOPED_TIMER(_project_timer);
+ RETURN_IF_ERROR(_projection_block(block, &output_block));
+ }
+
+ // Rename columns using the output column names from the sink config
+ const auto& paimon_sink = _t_sink.paimon_table_sink;
+ if (paimon_sink.__isset.paimon_options) {
+ auto it =
paimon_sink.paimon_options.find(PAIMON_OUTPUT_COLUMN_NAMES_KEY);
+ if (it != paimon_sink.paimon_options.end()) {
+ auto output_names = _split_column_names(it->second);
+ if (!output_names.empty() && output_names.size() ==
output_block.columns()) {
+ for (size_t i = 0; i < output_names.size(); ++i) {
+ output_block.get_by_position(i).name = output_names[i];
+ }
+ }
+ }
+ }
+
+ // Route by bucket and dispatch to per-bucket writers
+ return _route_by_bucket(output_block);
+}
+
+Status VPaimonTableWriter::_route_by_bucket(Block& block) {
+ // Compute bucket ids for all rows
+ std::vector<int32_t> buckets;
+ {
+ SCOPED_TIMER(_bucket_compute_timer);
+ buckets = _compute_buckets(block);
+ }
+
+ // Split block into per-bucket sub-blocks
+ auto bucket_blocks = _split_by_bucket(block, buckets);
+
+ // Write each bucket's data to the corresponding writer
+ for (auto& [bucket, sub_block] : bucket_blocks) {
+ std::unique_ptr<IPaimonWriter> writer;
+ RETURN_IF_ERROR(_get_or_create_bucket_writer(bucket, &writer));
+
+ COUNTER_UPDATE(_written_rows_counter, sub_block->rows());
+ COUNTER_UPDATE(_written_bytes_counter, sub_block->bytes());
+ _state->update_num_rows_load_total(sub_block->rows());
+ _state->update_num_bytes_load_total(sub_block->bytes());
+
+ {
+ SCOPED_TIMER(_file_store_write_timer);
+ RETURN_IF_ERROR(writer->write(_state, *sub_block));
+ }
+ _written_rows += sub_block->rows();
+ _written_bytes += sub_block->bytes();
+ }
+
+ COUNTER_SET(_bucket_writer_count,
static_cast<int64_t>(_bucket_writers.size()));
+ return Status::OK();
+}
+
+std::vector<int32_t> VPaimonTableWriter::_compute_buckets(const Block& block) {
+ if (!_bucket_routing_enabled) {
+ return std::vector<int32_t>(block.rows(), 0);
+ }
+ // Use BE-side Paimon murmur hash for bucket assignment
+ if (!_bucket_key_indices.empty()) {
+ return paimon_compute_buckets(block, _bucket_key_indices,
_total_buckets);
+ }
+ // Fallback: use hash of first column
+ std::vector<int> first_col = {0};
+ if (block.columns() > 0) {
+ return paimon_compute_buckets(block, first_col, _total_buckets);
+ }
+ return std::vector<int32_t>(block.rows(), 0);
+}
+
+std::unordered_map<BucketKey, std::unique_ptr<Block>>
VPaimonTableWriter::_split_by_bucket(
+ const Block& block, const std::vector<int32_t>& buckets) {
+ std::unordered_map<BucketKey, std::unique_ptr<Block>> result;
+
+ // Group row indices by bucket
+ std::unordered_map<BucketKey, std::vector<size_t>> bucket_rows;
+ for (size_t i = 0; i < buckets.size(); ++i) {
+ bucket_rows[buckets[i]].push_back(i);
+ }
+
+ // Build sub-blocks for each bucket
+ for (auto& [bucket, row_indices] : bucket_rows) {
+ auto sub_block = Block::create_unique(block.clone_empty());
+ auto columns = std::move(*sub_block).mutate_columns();
+ for (int col = 0; col < block.columns(); ++col) {
+ const auto& src_col = block.get_by_position(col).column;
+ for (size_t idx : row_indices) {
+ columns[col]->insert_from(*src_col, idx);
+ }
+ }
+ sub_block->set_columns(std::move(columns));
+ result[bucket] = std::move(sub_block);
+ }
+ return result;
+}
+
+Status VPaimonTableWriter::_get_or_create_bucket_writer(int32_t bucket,
+
std::unique_ptr<IPaimonWriter>* writer) {
+ auto it = _bucket_writers.find(bucket);
+ if (it != _bucket_writers.end()) {
+ // Return a raw pointer to the existing writer. The caller uses this
+ // temporarily and must not store it.
+ *writer = nullptr; // caller uses the returned pointer, not stored
Review Comment:
๐ด ่ฟไธช helper ็ๅบๅ `*writer` ไธคไธชๅๆฏ้ฝๆฟไธๅฐๅฏ็จๅฏน่ฑก๏ผๅทฒๅญๅจๅๆฏๆพๅผ็ฝฎ
`nullptr`๏ผๆฐๅปบๅๆฏ๏ผไธ้ข๏ผ`create_writer` ๅ็ๆฏ `_bucket_writers[bucket]`๏ผๅๆ ทๆฒก่ตๅผ
`*writer`ใ่ฐ็จๆน `_route_by_bucket` ้ๅ `writer->write(...)` ๅฟ
็ถๅฏน nullptr ่งฃๅผ็จ โ
ๆฏๆฌกๅๅ
ฅ้ฝ crashใ
่ไธ็จ `unique_ptr*` ๅๅบๅ่ฏญไนๆฌ่บซๅฐฑไธๅฏน๏ผwriter ๅฝ `_bucket_writers` ่ฟไธช map
ๆๆ๏ผไธ่ฝๅไบคๅบๆๆๆใๅบๆนๆ่ฟๅ่ฃธๆ้ `IPaimonWriter*`๏ผ`*out =
_bucket_writers[bucket].get();`๏ผ๏ผ่ฐ็จๆน็จ่ฃธๆ้ๅใ
##########
be/src/exec/sink/writer/paimon/paimon_bucket.cpp:
##########
@@ -0,0 +1,222 @@
+// 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/paimon_bucket.h"
+
+#include <cstring>
+
+#include "core/column/column.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
+
+namespace doris {
+
+// โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+// Murmur3-32 hash (seed=42, word-aligned input)
+// Matches Paimon's MurmurHashUtils.hashBytesByWords
+// โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+
+static constexpr uint32_t MURMUR_C1 = 0xcc9e2d51;
+static constexpr uint32_t MURMUR_C2 = 0x1b873593;
+static constexpr uint32_t MURMUR_SEED = 42;
+
+static uint32_t murmur_mix_k1(uint32_t k1) {
+ k1 *= MURMUR_C1;
+ k1 = (k1 << 15) | (k1 >> 17); // rotate_left(15)
+ k1 *= MURMUR_C2;
+ return k1;
+}
+
+static uint32_t murmur_mix_h1(uint32_t h1, uint32_t k1) {
+ h1 ^= k1;
+ h1 = (h1 << 13) | (h1 >> 19); // rotate_left(13)
+ h1 = h1 * 5 + 0xe6546b64;
+ return h1;
+}
+
+static uint32_t murmur_fmix(uint32_t h) {
+ h ^= h >> 16;
+ h *= 0x85ebca6b;
+ h ^= h >> 13;
+ h *= 0xc2b2ae35;
+ h ^= h >> 16;
+ return h;
+}
+
+int32_t paimon_murmur_hash(const uint8_t* data, size_t len) {
+ DCHECK(len % 4 == 0) << "Murmur hash requires word-aligned input, got " <<
len;
+ uint32_t h1 = MURMUR_SEED;
+ for (size_t i = 0; i < len; i += 4) {
+ uint32_t word = static_cast<uint32_t>(data[i]) |
(static_cast<uint32_t>(data[i + 1]) << 8) |
+ (static_cast<uint32_t>(data[i + 2]) << 16) |
+ (static_cast<uint32_t>(data[i + 3]) << 24);
+ uint32_t k1 = murmur_mix_k1(word);
+ h1 = murmur_mix_h1(h1, k1);
+ }
+ return static_cast<int32_t>(murmur_fmix(h1 ^ static_cast<uint32_t>(len)));
+}
+
+// โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+// Binary row construction for bucket key hashing
+// โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+
+/// Compute Paimon BinaryRow null-bit-set width.
+/// Matches BinaryRow.calBitSetWidthInBytes(arity).
+static inline int32_t cal_bit_set_width(int32_t arity) {
+ return ((arity + 63 + 8 /* HEADER_SIZE_IN_BYTES */) / 64) * 8;
+}
+
+/// Compute the fixed-size portion of a Paimon BinaryRow in bytes.
+static inline int32_t fixed_row_size(int32_t arity) {
+ return cal_bit_set_width(arity) + arity * 8;
+}
+
+/// Write a null bit at the given position.
+/// Matches BinaryRowBuilder.setNullAt(pos).
+static void set_null_bit(uint8_t* data, int32_t pos, bool is_null) {
+ int32_t bit_index = pos + 8; // HEADER_SIZE_IN_BYTES
+ int32_t byte_idx = bit_index / 8;
+ int32_t bit_off = bit_index % 8;
+ if (is_null) {
+ data[byte_idx] |= (1 << bit_off);
+ }
+}
+
+// โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+// Per-column value extraction helpers
+// โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+
+namespace {
+
+/// Write a single column value at `row_idx` into the binary row at the
+/// correct fixed-field offset. Returns the number of bytes actually
+/// written (always writes into an 8-byte slot at field_offset).
+void write_col_to_row(uint8_t* row_data, int32_t null_bytes_size, int32_t
field_pos,
+ const ColumnPtr& col, size_t row_idx) {
+ int32_t field_offset = null_bytes_size + field_pos * 8;
+
+ // Handle nullable column: unwrap the data column
+ const IColumn* data_col = col.get();
+ bool is_null = false;
+ if (col->is_nullable()) {
+ auto* nullable = assert_cast<const ColumnNullable*>(col.get());
+ is_null = nullable->is_null_at(row_idx);
+ data_col = &nullable->get_nested_column();
+ }
+
+ set_null_bit(row_data, field_pos, is_null);
+ if (is_null) {
+ return; // field already zeroed from initialization
+ }
+
+ // Write value in little-endian into the field slot
+ PrimitiveType ptype = data_col->get_data_type();
+
+ if (ptype == TYPE_TINYINT || ptype == TYPE_BOOLEAN) {
+ auto val = assert_cast<const
ColumnVector<Int8>*>(data_col)->get_element(row_idx);
+ row_data[field_offset] = static_cast<uint8_t>(val);
+ } else if (ptype == TYPE_SMALLINT) {
+ auto val = assert_cast<const
ColumnVector<Int16>*>(data_col)->get_element(row_idx);
+ std::memcpy(row_data + field_offset, &val, sizeof(val));
+ } else if (ptype == TYPE_INT || ptype == TYPE_DATE || ptype ==
TYPE_DATEV2) {
+ auto val = assert_cast<const
ColumnVector<Int32>*>(data_col)->get_element(row_idx);
+ std::memcpy(row_data + field_offset, &val, sizeof(val));
+ } else if (ptype == TYPE_BIGINT || ptype == TYPE_DATETIME || ptype ==
TYPE_DATETIMEV2) {
+ auto val = assert_cast<const
ColumnVector<Int64>*>(data_col)->get_element(row_idx);
+ std::memcpy(row_data + field_offset, &val, sizeof(val));
+ } else if (ptype == TYPE_FLOAT) {
+ float f = assert_cast<const
ColumnVector<Float32>*>(data_col)->get_element(row_idx);
+ int32_t bits;
+ std::memcpy(&bits, &f, sizeof(bits));
+ std::memcpy(row_data + field_offset, &bits, sizeof(bits));
+ } else if (ptype == TYPE_DOUBLE) {
+ double d = assert_cast<const
ColumnVector<Float64>*>(data_col)->get_element(row_idx);
+ int64_t bits;
+ std::memcpy(&bits, &d, sizeof(bits));
+ std::memcpy(row_data + field_offset, &bits, sizeof(bits));
+ } else if (ptype == TYPE_DECIMAL32) {
+ auto val = assert_cast<const
ColumnVector<Int32>*>(data_col)->get_element(row_idx);
+ std::memcpy(row_data + field_offset, &val, sizeof(val));
+ } else if (ptype == TYPE_DECIMAL64) {
+ auto val = assert_cast<const
ColumnVector<Int64>*>(data_col)->get_element(row_idx);
+ std::memcpy(row_data + field_offset, &val, sizeof(val));
+ } else if (ptype == TYPE_DECIMAL128 || ptype == TYPE_DECIMAL128I) {
+ auto val = assert_cast<const
ColumnVector<Int128>*>(data_col)->get_element(row_idx);
+ std::memcpy(row_data + field_offset, &val, sizeof(val));
+ } else {
+ // For complex types (STRING, ARRAY, MAP, etc.), we cannot compute
+ // the correct binary row in BE. Fallback to single-bucket routing.
+ // Callers should check this before calling.
+ LOG(WARNING) << "Unsupported bucket key type for BE-side computation: "
Review Comment:
๐ ่ฟๆดๅฅ BE ็ซฏ bucket ่ฎก็ฎๅจ v1 ๆขๅไฝๅไธๅฎๅ
จ๏ผๅปบ่ฎฎ็ดๆฅๅ ๆใ
**ๅไฝ**๏ผJNI ่ทฏๅพไธ `PaimonJniWriter.writeBatch` ๅทฒ็ป็จ `rowKeyExtractor.bucket()` ๅจ
Java ไพง้็ฎ bucket๏ผ่ BE ็ฎๅบๆฅ็ bucket ไปๆฒกไผ ็ป Java๏ผๆๆ per-bucket writer ้ฝๅ
ๅไธไธช Java
ๅฏน่ฑก๏ผ๏ผBE ่ฟ้็ๅๆกถ + ๆ block ๅฎๅ
จๆฏ็ฉบ่ฝฌใ
**ไธๅฎๅ
จ**๏ผๆๅทฅ้ๅปบ Paimon BinaryRow ๅนถไธ่ฝไธ Paimon ๅๅฐๅญ่็บงไธ่ด โโ
- `DATEV2` / `DATETIMEV2` ๅ็ๆฏ Doris ๅ
้จ packed ่กจ็คบ๏ผไธๆฏ Paimon ็ epoch-day int /
micros long๏ผ
- `DECIMAL32` / `DECIMAL64` ๆฒกๆ็ฌฆๅทๆฉๅฑๅฐ 8 ๅญ่ๆงฝ๏ผ่ดๆฐไผ็ฎ้๏ผ
- ่ฟ้ string / ๅคๆ็ฑปๅ็ดๆฅ `LOG(WARNING)` ไธๅไปปไฝๅญ่ โ ๆๆ่ก่ฝๅฐๅไธไธช bucketใ
ไธๆฆไปฅๅ็ๆไบบ็จๅฐ BE ็ฎ็ bucket๏ผๅฐฑไผ PK ๅป้้ไนฑใJava ไพงๅทฒ็ป็ฎๅฏนไบ๏ผv1 ๆฒกๅฟ
่ฆไฟ็่ฟๆก่ทฏๅพใ
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
