diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index d27cf607a36b24..8ca027e6c4ec55 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1608,6 +1608,11 @@ DEFINE_mInt64(hive_sink_max_file_size, "1073741824"); // 1GB /** Iceberg sink configurations **/ DEFINE_mInt64(iceberg_sink_max_file_size, "1073741824"); // 1GB +/** Paimon sink configurations **/ +DEFINE_mInt64(paimon_jni_writer_memory_pool_limit_bytes, "536870912"); // 512MB +DEFINE_Validator(paimon_jni_writer_memory_pool_limit_bytes, + [](int64_t bytes) -> bool { return bytes > 0; }); + // URI scheme to Doris file type mappings used by paimon-cpp DorisFileSystem. // Each entry uses the format "=", and file_type must be one of: // local, hdfs, s3, http, broker. diff --git a/be/src/common/config.h b/be/src/common/config.h index 470d88819703de..1983712b4b6a56 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1701,6 +1701,10 @@ DECLARE_mInt64(hive_sink_max_file_size); /** Iceberg sink configurations **/ DECLARE_mInt64(iceberg_sink_max_file_size); +/** Paimon sink configurations **/ +// Hard upper bound for Doris-managed Paimon write-buffer memory per JNI writer. +DECLARE_mInt64(paimon_jni_writer_memory_pool_limit_bytes); + /** Paimon file system configurations **/ DECLARE_Strings(paimon_file_system_scheme_mappings); diff --git a/be/src/core/column/column_variant.cpp b/be/src/core/column/column_variant.cpp index e00de8cd75bac1..2b8b54ded0a996 100644 --- a/be/src/core/column/column_variant.cpp +++ b/be/src/core/column/column_variant.cpp @@ -188,7 +188,7 @@ DataTypeSerDeSPtr ColumnVariant::Subcolumn::generate_data_serdes(DataTypePtr typ ColumnVariant::Subcolumn::Subcolumn(MutableColumnPtr&& data_, DataTypePtr type, bool is_nullable_, bool is_root_) - : least_common_type(type), + : least_common_type(type, is_root_), is_nullable(is_nullable_), is_root(is_root_), num_rows(data_->size()) { @@ -463,6 +463,11 @@ void ColumnVariant::Subcolumn::insert_range_from(const Subcolumn& src, size_t st size_t part_end = end - processed_rows; insert_from_part(src.data[pos], src.data_types[pos], 0, part_end); } + + const size_t trailing_defaults_start = std::max(start, src.num_rows); + if (end > trailing_defaults_start) { + data.back()->insert_many_defaults(end - trailing_defaults_start); + } } bool ColumnVariant::Subcolumn::is_finalized() const { @@ -488,20 +493,29 @@ MutableColumnPtr ColumnVariant::apply_for_columns(Func&& func) const { auto& finalized_object = assert_cast(*finalized); return finalized_object.apply_for_columns(std::forward(func)); } - auto new_root = std::move(*func(get_root())).mutate(); - auto res = ColumnVariant::create(_max_subcolumns_count, _enable_doc_mode, get_root_type(), - std::move(new_root)); + Subcolumns transformed_subcolumns; for (const auto& subcolumn : subcolumns) { - if (subcolumn->data.is_root) { + auto transformed = std::move(*func(subcolumn->data.get_finalized_column_ptr())).mutate(); + Subcolumn transformed_subcolumn(std::move(transformed), + subcolumn->data.get_least_common_type(), is_nullable, + subcolumn->data.is_root); + if (subcolumn->data.is_root || subcolumn->path.empty()) { + transformed_subcolumns.create_root(std::move(transformed_subcolumn)); continue; } - auto new_subcolumn = func(subcolumn->data.get_finalized_column_ptr()); - if (!res->add_sub_column(subcolumn->path, std::move(*new_subcolumn).mutate(), - subcolumn->data.get_least_common_type())) { + if (!transformed_subcolumns.add(subcolumn->path, std::move(transformed_subcolumn))) { throw doris::Exception(ErrorCode::INTERNAL_ERROR, "add path {} is error", subcolumn->path.get_path()); } } + if (transformed_subcolumns.get_root() == nullptr) { + throw doris::Exception(ErrorCode::INTERNAL_ERROR, + "root is nullptr after transforming variant columns"); + } + auto res = ColumnVariant::create(_max_subcolumns_count, _enable_doc_mode, + std::move(transformed_subcolumns)); + res->typed_path_count = typed_path_count; + res->nested_path_count = nested_path_count; auto sparse_column = func(serialized_sparse_column); res->serialized_sparse_column = IColumn::mutate(std::move(sparse_column)); auto doc_value_column = func(serialized_doc_value_column); @@ -869,21 +883,10 @@ void ColumnVariant::insert_from(const IColumn& src, size_t n) { const auto* src_v = assert_cast(&src); ENABLE_CHECK_CONSISTENCY(src_v); ENABLE_CHECK_CONSISTENCY(this); - // Preserve the original root-only copy path for ordinary variant columns. - // Reconstructing through try_insert() loses sparse/doc_value structure for - // mixed-shape rows and nested-group data. - if (src_v->get_subcolumns().size() == 1 && get_subcolumns().size() == 1) { - DCHECK(_enable_doc_mode == src_v->_enable_doc_mode) - << "root-only variant copy requires matching doc mode"; - FieldWithDataType field; - src_v->subcolumns.get_root()->data.get(n, field); - subcolumns.get_mutable_root()->data.insert(field); - serialized_sparse_column->insert_from(*src_v->get_sparse_column(), n); - serialized_doc_value_column->insert_from(*src_v->get_doc_value_column(), n); - num_rows++; - } else { - try_insert((*src_v)[n]); - } + // Keep complex and materialized object paths in their native subcolumns. Reconstructing a + // single row as a Field can turn nested objects into TYPE_STRUCT, which is not a scalar type + // that Subcolumn::insert() can create through DataTypeFactory. + insert_range_from(*src_v, n, 1); ENABLE_CHECK_CONSISTENCY(this); } diff --git a/be/src/core/data_type_serde/data_type_variant_serde.cpp b/be/src/core/data_type_serde/data_type_variant_serde.cpp index 6cd0126291e7b1..e7fab23c251933 100644 --- a/be/src/core/data_type_serde/data_type_variant_serde.cpp +++ b/be/src/core/data_type_serde/data_type_variant_serde.cpp @@ -18,6 +18,7 @@ #include "core/data_type_serde/data_type_variant_serde.h" #include +#include #include #include @@ -35,6 +36,7 @@ #include "core/types.h" #include "core/value/jsonb_value.h" #include "exec/common/variant_util.h" +#include "exprs/function/parse/variant_jsonb_parse.h" #include "util/json/json_parser.h" #include "util/jsonb_writer.h" @@ -63,6 +65,77 @@ Status write_variant_column_to_arrow_impl(const IColumn& column, const ColumnVar return Status::OK(); } +Status write_variant_column_to_arrow_struct(const IColumn& column, const ColumnVariant& var, + const NullMap* null_map, arrow::StructBuilder& builder, + int64_t start, int64_t end, + const cctz::time_zone& ctz) { + const auto struct_type = std::dynamic_pointer_cast(builder.type()); + if (struct_type == nullptr || builder.num_fields() != 2 || + struct_type->field(0)->name() != "value" || struct_type->field(1)->name() != "metadata") { + return Status::InvalidArgument( + "Variant Arrow output requires " + "struct"); + } + auto* value_builder = dynamic_cast(builder.field_builder(0)); + auto* metadata_builder = dynamic_cast(builder.field_builder(1)); + if (value_builder == nullptr || metadata_builder == nullptr) { + return Status::InvalidArgument( + "Variant Arrow output requires binary value and metadata children"); + } + + const auto* root = var.get_subcolumn(PathInData()); + const bool string_root = + root != nullptr && is_string_type(root->get_least_common_base_type_id()); + JsonbToVariantEncoder encoder( + VariantBatchBuilder::ReserveHint {.rows = cast_set(end - start)}); + DataTypeSerDe::FormatOptions options; + options.timezone = &ctz; + for (int64_t row = start; row < end; ++row) { + if (null_map != nullptr && (*null_map)[cast_set(row)]) { + encoder.add_null(); + continue; + } + + std::string serialized_value; + var.serialize_one_row_to_string(row, &serialized_value, options); + if (string_root && !root->is_null_at(cast_set(row))) { + JsonbWriter writer; + if (!writer.writeStartString() || + (!serialized_value.empty() && + !writer.writeString(serialized_value.data(), serialized_value.size())) || + !writer.writeEndString()) { + return Status::InternalError("Failed to encode legacy Variant string as JSONB"); + } + encoder.add_jsonb({writer.getOutput()->getBuffer(), + static_cast(writer.getOutput()->getSize())}); + continue; + } + + JsonBinaryValue jsonb; + RETURN_IF_ERROR(jsonb.from_json_string(serialized_value)); + encoder.add_jsonb({jsonb.value(), jsonb.size()}); + } + + VariantBatchBuilder batch = encoder.finish_batch(); + for (size_t row = 0; row < batch.num_rows(); ++row) { + const size_t source_row = cast_set(start) + row; + if (null_map != nullptr && (*null_map)[source_row]) { + RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column, builder)); + continue; + } + const VariantRef value = batch.value_at(row); + RETURN_IF_ERROR(checkArrowStatus(builder.Append(), column, builder)); + RETURN_IF_ERROR(checkArrowStatus( + value_builder->Append(value.value.data, cast_set(value.value.size)), + column, *value_builder)); + RETURN_IF_ERROR( + checkArrowStatus(metadata_builder->Append(value.metadata.data, + cast_set(value.metadata.size)), + column, *metadata_builder)); + } + return Status::OK(); +} + } // namespace Status DataTypeVariantSerDe::write_column_to_mysql_binary(const IColumn& column, @@ -161,6 +234,10 @@ Status DataTypeVariantSerDe::write_column_to_arrow(const IColumn& column, const } else if (array_builder->type()->id() == arrow::Type::STRING) { auto& builder = assert_cast(*array_builder); return write_variant_column_to_arrow_impl(column, *var, null_map, builder, start, end, ctz); + } else if (array_builder->type()->id() == arrow::Type::STRUCT) { + auto& builder = assert_cast(*array_builder); + RETURN_IF_CATCH_EXCEPTION(return write_variant_column_to_arrow_struct( + column, *var, null_map, builder, start, end, ctz)); } else { return Status::InvalidArgument("Unsupported arrow type for variant column: {}", array_builder->type()->name()); @@ -208,7 +285,8 @@ Status DataTypeVariantSerDe::write_column_to_orc(const std::string& timezone, co size_t len = serialized_value.length(); if (offset + len > total_size) { return Status::InternalError( - "Buffer overflow when writing column data to ORC file. offset {} with len {} " + "Buffer overflow when writing column data " + "to ORC file. offset {} with len {} " "exceed total_size {} . ", offset, len, total_size); } diff --git a/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp b/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp index 4a1d8e70b7bc4c..3ef8142b2afb43 100644 --- a/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp +++ b/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp @@ -18,6 +18,7 @@ #include "core/data_type_serde/data_type_variant_v2_serde.h" #include +#include #include #include @@ -491,6 +492,50 @@ Status write_arrow(const IColumn& column, const NullMap* null_map, Builder& buil return status; } +Status write_arrow_variant(const IColumn& column, const NullMap* null_map, + arrow::StructBuilder& builder, size_t start, size_t end) { + const auto struct_type = std::dynamic_pointer_cast(builder.type()); + if (struct_type == nullptr || builder.num_fields() != 2 || + struct_type->field(0)->name() != "value" || struct_type->field(1)->name() != "metadata") { + return Status::InvalidArgument( + "Variant Arrow output requires struct"); + } + auto* value_builder = dynamic_cast(builder.field_builder(0)); + auto* metadata_builder = dynamic_cast(builder.field_builder(1)); + if (value_builder == nullptr || metadata_builder == nullptr) { + return Status::InvalidArgument( + "Variant Arrow output requires binary value and metadata children"); + } + + Status status = Status::OK(); + visit_variant_v2_values( + column, start, end, forced_nulls(null_map), + [&](size_t) { + if (status.ok()) { + status = checkArrowStatus(builder.AppendNull(), column, builder); + } + }, + [&](size_t, VariantRef value) { + if (!status.ok()) { + return; + } + status = checkArrowStatus(builder.Append(), column, builder); + if (status.ok()) { + status = checkArrowStatus( + value_builder->Append(value.value.data, + cast_set(value.value.size)), + column, *value_builder); + } + if (status.ok()) { + status = checkArrowStatus( + metadata_builder->Append(value.metadata.data, + cast_set(value.metadata.size)), + column, *metadata_builder); + } + }); + return status; +} + } // namespace void DataTypeVariantV2SerDe::to_string(const IColumn& column, size_t row_num, BufferWritable& bw, @@ -553,6 +598,11 @@ Status DataTypeVariantV2SerDe::write_column_to_arrow(const IColumn& column, cons assert_cast(*array_builder), first, last, options); } + if (array_builder->type()->id() == arrow::Type::STRUCT) { + return write_arrow_variant(column, null_map, + assert_cast(*array_builder), first, + last); + } return Status::InvalidArgument("Unsupported arrow type for variant column: {}", array_builder->type()->name()); }); diff --git a/be/src/exec/operator/operator.cpp b/be/src/exec/operator/operator.cpp index cfc9d6645119c8..9a12adaca8e07f 100644 --- a/be/src/exec/operator/operator.cpp +++ b/be/src/exec/operator/operator.cpp @@ -64,6 +64,7 @@ #include "exec/operator/olap_scan_operator.h" #include "exec/operator/olap_table_sink_operator.h" #include "exec/operator/olap_table_sink_v2_operator.h" +#include "exec/operator/paimon_table_sink_operator.h" #include "exec/operator/partition_sort_sink_operator.h" #include "exec/operator/partition_sort_source_operator.h" #include "exec/operator/partitioned_aggregation_sink_operator.h" @@ -830,6 +831,7 @@ DECLARE_OPERATOR(OlapTableSinkV2LocalState) DECLARE_OPERATOR(HiveTableSinkLocalState) DECLARE_OPERATOR(TVFTableSinkLocalState) DECLARE_OPERATOR(IcebergTableSinkLocalState) +DECLARE_OPERATOR(PaimonTableSinkLocalState) DECLARE_OPERATOR(SpillIcebergTableSinkLocalState) DECLARE_OPERATOR(IcebergDeleteSinkLocalState) DECLARE_OPERATOR(IcebergMergeSinkLocalState) @@ -955,6 +957,7 @@ template class AsyncWriterSink; template class AsyncWriterSink; template class AsyncWriterSink; template class AsyncWriterSink; +template class AsyncWriterSink; template class AsyncWriterSink; template class AsyncWriterSink; template class AsyncWriterSink; diff --git a/be/src/exec/operator/paimon_table_sink_operator.cpp b/be/src/exec/operator/paimon_table_sink_operator.cpp new file mode 100644 index 00000000000000..c1386be558ec62 --- /dev/null +++ b/be/src/exec/operator/paimon_table_sink_operator.cpp @@ -0,0 +1,39 @@ +// 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/operator/paimon_table_sink_operator.h" + +#include "common/logging.h" + +namespace doris { + +Status PaimonTableSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) { + return Base::init(state, info); +} + +Status PaimonTableSinkOperatorX::sink_impl(RuntimeState* state, Block* in_block, bool eos) { + auto& local_state = get_local_state(state); + SCOPED_TIMER(local_state.exec_time_counter()); + COUNTER_UPDATE(local_state.rows_input_counter(), static_cast(in_block->rows())); + + // Delegate to AsyncWriterSink → PaimonTableWriter for this pipeline instance. + // Each pipeline instance has its own writer session; partition and bucket + // routing is handled internally by the Paimon SDK inside IPaimonWriter::write(). + return local_state.sink(state, in_block, eos); +} + +} // namespace doris diff --git a/be/src/exec/operator/paimon_table_sink_operator.h b/be/src/exec/operator/paimon_table_sink_operator.h new file mode 100644 index 00000000000000..3a2f885720348c --- /dev/null +++ b/be/src/exec/operator/paimon_table_sink_operator.h @@ -0,0 +1,106 @@ +// 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 + +#include +#include + +#include "common/status.h" +#include "core/block/block.h" +#include "exec/operator/operator.h" +#include "exec/sink/writer/paimon/paimon_table_writer.h" +#include "runtime/runtime_state.h" + +namespace doris { + +/// Paimon table sink operator — simple pass-through to AsyncWriterSink. +/// +/// Each pipeline instance (LocalState) owns one PaimonTableWriter, which in +/// turn owns one IPaimonWriteBackend + IPaimonWriter. Pipeline parallelism +/// determines the number of concurrent Paimon writer sessions per table. FE's +/// Paimon write provider currently requires GATHER so bucket assignment sees +/// one ordered input stream. +/// +/// Partition and bucket routing is performed internally by the Paimon Java SDK +/// through JNI. Doris does not compute partition values or bucket ids; it +/// passes complete Blocks through the backend to the SDK, +/// where each row is routed via getPartition(row) + getBucket(row). +/// +/// This mirrors Iceberg's approach: IcebergTableSinkOperatorX delegates to +/// AsyncWriterSink, with partition routing inside +/// VIcebergTableWriter::write(). +class PaimonTableSinkOperatorX; + +class PaimonTableSinkLocalState final + : public AsyncWriterSink { +public: + using Base = AsyncWriterSink; + using Parent = PaimonTableSinkOperatorX; + ENABLE_FACTORY_CREATOR(PaimonTableSinkLocalState); + PaimonTableSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state) + : Base(parent, state) {} + Status init(RuntimeState* state, LocalSinkStateInfo& info) override; + Status open(RuntimeState* state) override { + SCOPED_TIMER(exec_time_counter()); + SCOPED_TIMER(_open_timer); + return Base::open(state); + } + + friend class PaimonTableSinkOperatorX; +}; + +class PaimonTableSinkOperatorX final : public DataSinkOperatorX { +public: + using Base = DataSinkOperatorX; + PaimonTableSinkOperatorX(ObjectPool* pool, int operator_id, const RowDescriptor& row_desc, + const std::vector& t_output_expr) + : Base(operator_id, 0, 0), + _row_desc(row_desc), + _t_output_expr(t_output_expr), + _pool(pool) {} + + Status init(const TDataSink& thrift_sink) override { + RETURN_IF_ERROR(Base::init(thrift_sink)); + DCHECK(thrift_sink.__isset.paimon_table_sink); + RETURN_IF_ERROR(VExpr::create_expr_trees(_t_output_expr, _output_vexpr_ctxs)); + return Status::OK(); + } + + Status prepare(RuntimeState* state) override { + RETURN_IF_ERROR(Base::prepare(state)); + RETURN_IF_ERROR(VExpr::prepare(_output_vexpr_ctxs, state, _row_desc)); + return VExpr::open(_output_vexpr_ctxs, state); + } + + Status sink_impl(RuntimeState* state, Block* in_block, bool eos) override; + +private: + friend class PaimonTableSinkLocalState; + template + requires(std::is_base_of_v) + friend class AsyncWriterSink; + + const RowDescriptor& _row_desc; + VExprContextSPtrs _output_vexpr_ctxs; + const std::vector& _t_output_expr; + ObjectPool* _pool = nullptr; +}; + +} // namespace doris diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp b/be/src/exec/pipeline/pipeline_fragment_context.cpp index 2fa064c8a68e09..abf8db9c6fa102 100644 --- a/be/src/exec/pipeline/pipeline_fragment_context.cpp +++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp @@ -88,6 +88,7 @@ #include "exec/operator/olap_scan_operator.h" #include "exec/operator/olap_table_sink_operator.h" #include "exec/operator/olap_table_sink_v2_operator.h" +#include "exec/operator/paimon_table_sink_operator.h" #include "exec/operator/partition_sort_sink_operator.h" #include "exec/operator/partition_sort_source_operator.h" #include "exec/operator/partitioned_aggregation_sink_operator.h" @@ -1352,6 +1353,14 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS } break; } + case TDataSinkType::PAIMON_TABLE_SINK: { + if (!thrift_sink.__isset.paimon_table_sink) { + return Status::InternalError("Missing paimon table sink."); + } + _sink = std::make_shared(pool, next_sink_operator_id(), row_desc, + output_exprs); + break; + } case TDataSinkType::ICEBERG_DELETE_SINK: { if (!thrift_sink.__isset.iceberg_delete_sink) { return Status::InternalError("Missing iceberg delete sink."); @@ -2585,7 +2594,8 @@ void PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r PrintThriftNetworkAddress(req.coord_addr), e.what()); } - const bool requires_external_file_ack = params.__isset.iceberg_commit_datas; + const bool requires_external_file_ack = + params.__isset.iceberg_commit_datas || params.__isset.paimon_commit_messages; if (rpc_status.ok() && requires_external_file_ack && (!res.__isset.external_file_commit_data_accepted || !res.external_file_commit_data_accepted)) { 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 new file mode 100644 index 00000000000000..149835b68c9210 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp @@ -0,0 +1,622 @@ +// 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/jni_paimon_write_backend.h" + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "common/check.h" +#include "common/logging.h" +#include "core/data_type/data_type_agg_state.h" +#include "core/data_type/data_type_array.h" +#include "core/data_type/data_type_map.h" +#include "core/data_type/data_type_struct.h" +#include "exec/sink/writer/paimon/paimon_jni_memory_manager.h" +#include "format/arrow/arrow_block_convertor.h" +#include "format/arrow/arrow_row_batch.h" +#include "runtime/exec_env.h" +#include "runtime/runtime_state.h" +#include "storage/options.h" +#include "util/jni-util.h" +#include "util/pretty_printer.h" +#include "util/string_util.h" + +namespace doris { + +namespace { +constexpr std::string_view PAIMON_JNI_WRITER_IO_TMP_DIR = "paimon_jni_writer_io_tmp"; + +std::atomic& paimon_jni_close_failed() { + static auto* failed = new std::atomic(false); + return *failed; +} + +std::mutex& retained_memory_managers_mutex() { + static auto* mutex = new std::mutex(); + return *mutex; +} + +std::vector>& retained_memory_managers() { + static auto* managers = new std::vector>(); + return *managers; +} + +void retain_memory_after_failed_close(std::unique_ptr manager) { + paimon_jni_close_failed().store(true, std::memory_order_release); + if (manager == nullptr) { + return; + } + std::lock_guard lock(retained_memory_managers_mutex()); + retained_memory_managers().emplace_back(std::move(manager)); +} + +Status convert_to_paimon_arrow_type(const DataTypePtr& origin_type, + std::shared_ptr* result, + const std::string& timezone) { + const DataTypePtr type = get_serialized_type(origin_type); + switch (type->get_primitive_type()) { + case TYPE_VARIANT: + // Paimon consumes the lossless Variant V2 representation. Keeping both children non-null + // distinguishes a SQL NULL struct from a non-null Variant value. + *result = arrow::struct_({arrow::field("value", arrow::binary(), false), + arrow::field("metadata", arrow::binary(), false)}); + return Status::OK(); + case TYPE_ARRAY: { + const auto& array_type = assert_cast(*remove_nullable(type)); + std::shared_ptr element_type; + RETURN_IF_ERROR(convert_to_paimon_arrow_type(array_type.get_nested_type(), &element_type, + timezone)); + *result = std::make_shared(element_type); + return Status::OK(); + } + case TYPE_MAP: { + const auto& map_type = assert_cast(*remove_nullable(type)); + std::shared_ptr key_type; + std::shared_ptr value_type; + RETURN_IF_ERROR(convert_to_paimon_arrow_type(map_type.get_key_type(), &key_type, timezone)); + RETURN_IF_ERROR( + convert_to_paimon_arrow_type(map_type.get_value_type(), &value_type, timezone)); + *result = std::make_shared(key_type, value_type); + return Status::OK(); + } + case TYPE_STRUCT: { + const auto& struct_type = assert_cast(*remove_nullable(type)); + std::vector> fields; + fields.reserve(struct_type.get_elements().size()); + for (size_t i = 0; i < struct_type.get_elements().size(); ++i) { + const DataTypePtr& element = struct_type.get_element(i); + std::shared_ptr field_type; + RETURN_IF_ERROR(convert_to_paimon_arrow_type(element, &field_type, timezone)); + fields.push_back(arrow::field(struct_type.get_element_name(i), field_type, + element->is_nullable())); + } + *result = arrow::struct_(std::move(fields)); + return Status::OK(); + } + default: + return convert_to_arrow_type(origin_type, result, timezone); + } +} + +Status get_paimon_arrow_schema_from_block(const Block& block, + std::shared_ptr* result) { + std::vector> fields; + fields.reserve(block.columns()); + for (const auto& type_and_name : block) { + std::shared_ptr arrow_type; + RETURN_IF_ERROR(convert_to_paimon_arrow_type(type_and_name.type, &arrow_type, "")); + fields.push_back(create_arrow_field_with_metadata( + type_and_name.name, arrow_type, type_and_name.type->is_nullable(), + type_and_name.type->get_primitive_type())); + } + *result = arrow::schema(std::move(fields)); + return Status::OK(); +} +} // namespace + +// ──────────────────────────────────────────────────────────── +// JNI helpers — JVM attachment and class loading +// ──────────────────────────────────────────────────────────── + +static constexpr const char* PAIMON_JNI_WRITER_CLASS = "org/apache/doris/paimon/PaimonJniWriter"; +static constexpr const char* SCANNER_LOADER_CLASS = + "org/apache/doris/common/classloader/ScannerLoader"; + +/// Attach the current native thread to the JVM if not already attached, +/// and return a valid JNIEnv pointer. +static Status _get_jni_env(JNIEnv** env) { + JavaVM* jvm = nullptr; + jsize n_vms = 0; + jint result = JNI_GetCreatedJavaVMs(&jvm, 1, &n_vms); + if (result != JNI_OK || n_vms == 0) { + return Status::InternalError("Failed to get created JavaVM"); + } + result = jvm->GetEnv(reinterpret_cast(env), JNI_VERSION_1_8); + if (result == JNI_EDETACHED) { + result = jvm->AttachCurrentThread(reinterpret_cast(env), nullptr); + if (result != JNI_OK) { + return Status::InternalError("Failed to attach current thread to JVM"); + } + } else if (result != JNI_OK) { + return Status::InternalError("Failed to get JNIEnv"); + } + return Status::OK(); +} + +JniPaimonWriteBackend::~JniPaimonWriteBackend() { + Status st = close(); + if (!st.ok()) { + LOG(WARNING) << "Failed to close Paimon JNI backend during destruction: " << st.to_string(); + } +} + +Status JniPaimonWriteBackend::close() { + if (_jni_writer_obj == nullptr && _jni_writer_cls == nullptr) { + _memory_manager.reset(); + _opened = false; + return Status::OK(); + } + + JNIEnv* env = nullptr; + Status env_status = _get_jni_env(&env); + if (!env_status.ok()) { + bool java_users_may_exist = _jni_writer_obj != nullptr; + // JNI global references cannot be released without an environment. + // Deliberately abandon the handles so the Java writer remains alive. + _jni_writer_obj = nullptr; + _jni_writer_cls = nullptr; + if (java_users_may_exist) { + retain_memory_after_failed_close(std::move(_memory_manager)); + } else { + _memory_manager.reset(); + } + _opened = false; + return env_status; + } + + Status close_status = Status::OK(); + if (_jni_writer_obj != nullptr) { + _refresh_memory_profile(); + if (_close_id == nullptr) { + close_status = Status::InternalError("PaimonJniWriter.close method is unavailable"); + } else { + env->CallVoidMethod(_jni_writer_obj, _close_id); + close_status = _check_jni_exception(env, "close PaimonJniWriter"); + } + env->DeleteGlobalRef(_jni_writer_obj); + _jni_writer_obj = nullptr; + } + if (_jni_writer_cls != nullptr) { + env->DeleteGlobalRef(_jni_writer_cls); + _jni_writer_cls = nullptr; + } + + if (close_status.ok()) { + _memory_manager.reset(); + } else { + if (_memory_manager != nullptr) { + LOG(WARNING) + << "Retaining Paimon JNI native memory after an unconfirmed Java close: limit=" + << PrettyPrinter::print_bytes(_memory_manager->memory_limit()) << ", peak=" + << PrettyPrinter::print_bytes(_memory_manager->native_peak_allocated_bytes()); + } + // Paimon may still have asynchronous flush or compaction tasks using + // MemorySegments backed by these pages. Retain ownership until process + // exit and reject new writers below. Retention is therefore limited to + // writers which were already open when the first close failure occurred. + retain_memory_after_failed_close(std::move(_memory_manager)); + } + _opened = false; + return close_status; +} + +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 + ": "); + LOG(WARNING) << st.to_string(); + return st; + } + return Status::OK(); +} + +Status JniPaimonWriteBackend::_load_writer_class(JNIEnv* env, jclass* writer_class) { + jclass loader_class = env->FindClass(SCANNER_LOADER_CLASS); + RETURN_IF_ERROR(_check_jni_exception(env, "find ScannerLoader")); + + jmethodID loader_constructor = env->GetMethodID(loader_class, "", "()V"); + jmethodID get_loaded_class = env->GetMethodID(loader_class, "getLoadedClass", + "(Ljava/lang/String;)Ljava/lang/Class;"); + RETURN_IF_ERROR(_check_jni_exception(env, "resolve ScannerLoader methods")); + + jobject loader = env->NewObject(loader_class, loader_constructor); + jstring class_name = env->NewStringUTF(PAIMON_JNI_WRITER_CLASS); + auto* loaded_class = + static_cast(env->CallObjectMethod(loader, get_loaded_class, class_name)); + RETURN_IF_ERROR(_check_jni_exception(env, "load PaimonJniWriter")); + + *writer_class = loaded_class; + env->DeleteLocalRef(class_name); + env->DeleteLocalRef(loader); + env->DeleteLocalRef(loader_class); + return Status::OK(); +} + +static jobject _to_java_options(JNIEnv* env, const std::map& options) { + jclass map_cls = env->FindClass("java/util/HashMap"); + jmethodID map_ctor = env->GetMethodID(map_cls, "", "()V"); + jmethodID put_method = env->GetMethodID( + 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) { + jstring key = env->NewStringUTF(kv.first.c_str()); + jstring val = env->NewStringUTF(kv.second.c_str()); + env->CallObjectMethod(map_obj, put_method, key, val); + env->DeleteLocalRef(key); + env->DeleteLocalRef(val); + } + env->DeleteLocalRef(map_cls); + return map_obj; +} + +Status JniPaimonWriteBackend::abort_prepared_commit( + const TPaimonTableSink& sink, const std::vector& commit_messages) { + if (commit_messages.empty()) { + return Status::OK(); + } + + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + + jclass writer_class = nullptr; + RETURN_IF_ERROR(_load_writer_class(env, &writer_class)); + jmethodID abort_id = + env->GetStaticMethodID(writer_class, "abortPreparedCommit", + "(Ljava/lang/String;Ljava/util/Map;Ljava/lang/String;[[B)V"); + Status method_status = _check_jni_exception(env, "resolve abortPreparedCommit"); + if (!method_status.ok()) { + env->DeleteLocalRef(writer_class); + return method_status; + } + + const std::map empty_config; + jstring serialized_table = env->NewStringUTF(sink.serialized_table.c_str()); + jobject hadoop_config = + _to_java_options(env, sink.__isset.hadoop_config ? sink.hadoop_config : empty_config); + jstring commit_user = env->NewStringUTF(sink.commit_user.c_str()); + jclass byte_array_class = env->FindClass("[B"); + auto payloads = env->NewObjectArray(static_cast(commit_messages.size()), + byte_array_class, nullptr); + Status allocation_status = _check_jni_exception(env, "allocate abortPreparedCommit arguments"); + if (allocation_status.ok()) { + for (size_t i = 0; i < commit_messages.size(); ++i) { + const auto& message = commit_messages[i]; + DORIS_CHECK(message.__isset.payload); + auto payload = env->NewByteArray(static_cast(message.payload.size())); + env->SetByteArrayRegion(payload, 0, static_cast(message.payload.size()), + reinterpret_cast(message.payload.data())); + env->SetObjectArrayElement(payloads, static_cast(i), payload); + env->DeleteLocalRef(payload); + } + allocation_status = _check_jni_exception(env, "populate abortPreparedCommit payloads"); + } + + Status abort_status = allocation_status; + if (abort_status.ok()) { + env->CallStaticVoidMethod(writer_class, abort_id, serialized_table, hadoop_config, + commit_user, payloads); + abort_status = _check_jni_exception(env, "abort prepared Paimon commit"); + } + + if (payloads != nullptr) { + env->DeleteLocalRef(payloads); + } + if (byte_array_class != nullptr) { + env->DeleteLocalRef(byte_array_class); + } + if (commit_user != nullptr) { + env->DeleteLocalRef(commit_user); + } + if (hadoop_config != nullptr) { + env->DeleteLocalRef(hadoop_config); + } + if (serialized_table != nullptr) { + env->DeleteLocalRef(serialized_table); + } + env->DeleteLocalRef(writer_class); + return abort_status; +} + +Status JniPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* state, + RuntimeProfile* profile) { + if (paimon_jni_close_failed().load(std::memory_order_acquire)) { + return Status::InternalError( + "Paimon JNI writes are disabled on this BE because a previous Java writer close " + "failed; restart the BE to reclaim retained native memory safely"); + } + _sink = sink; + DORIS_CHECK(sink.__isset.column_names); + DORIS_CHECK(sink.__isset.write_mode); + DORIS_CHECK(sink.__isset.serialized_table); + DORIS_CHECK(!sink.serialized_table.empty()); + DORIS_CHECK(sink.__isset.transaction_id); + DORIS_CHECK(sink.transaction_id > 0); + DORIS_CHECK(sink.__isset.commit_user); + DORIS_CHECK(!sink.commit_user.empty()); + DORIS_CHECK(profile != nullptr); + + RETURN_IF_ERROR(PaimonJniMemoryManager::create(state, &_memory_manager)); + RuntimeProfile* jni_profile = profile->create_child("JniPaimonWriteBackend", true, true); + _native_page_memory_limit = ADD_COUNTER(jni_profile, "NativePageMemoryLimit", TUnit::BYTES); + _native_page_memory_peak = ADD_COUNTER(jni_profile, "NativePageMemoryPeak", TUnit::BYTES); + + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + + // Step 1: Load PaimonJniWriter class through ScannerLoader (Paimon jars are + // not on the default application classpath, so FindClass won't work). + jclass local_cls = nullptr; + RETURN_IF_ERROR(_load_writer_class(env, &local_cls)); + _jni_writer_cls = static_cast(env->NewGlobalRef(local_cls)); + env->DeleteLocalRef(local_cls); + RETURN_IF_ERROR(PaimonJniMemoryManager::register_natives(env, _jni_writer_cls)); + + // Step 2: Cache JNI method IDs for write, prepareCommit, abort, close. + jmethodID open_id = env->GetMethodID( + _jni_writer_cls, "open", + "(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;JLjava/lang/String;ZLjava/lang/" + "String;Ljava/lang/String;JJ)V"); + _write_id = env->GetMethodID(_jni_writer_cls, "write", "(Ljava/nio/ByteBuffer;)V"); + _prepare_commit_id = env->GetMethodID(_jni_writer_cls, "prepareCommit", "()[[B"); + _abort_id = env->GetMethodID(_jni_writer_cls, "abort", "()V"); + _close_id = env->GetMethodID(_jni_writer_cls, "close", "()V"); + RETURN_IF_ERROR(_check_jni_exception(env, "GetMethodID")); + + // Step 3: Create the Java PaimonJniWriter instance. + jmethodID ctor_id = env->GetMethodID(_jni_writer_cls, "", "()V"); + jobject local_obj = env->NewObject(_jni_writer_cls, ctor_id); + RETURN_IF_ERROR(_check_jni_exception(env, "NewObject")); + _jni_writer_obj = env->NewGlobalRef(local_obj); + env->DeleteLocalRef(local_obj); + + // Step 4: Build Java arguments and call PaimonJniWriter.open(). + const std::map empty_config; + jstring j_serialized_table = env->NewStringUTF(sink.serialized_table.c_str()); + jobject j_hadoop_config = + _to_java_options(env, sink.__isset.hadoop_config ? sink.hadoop_config : empty_config); + jstring j_commit_user = env->NewStringUTF(sink.commit_user.c_str()); + jstring j_time_zone = env->NewStringUTF(state->timezone().c_str()); + std::vector spill_directories; + for (const auto& store_path : state->exec_env()->store_paths()) { + spill_directories.push_back(store_path.path + "/" + + std::string(PAIMON_JNI_WRITER_IO_TMP_DIR)); + } + DORIS_CHECK(!spill_directories.empty()); + jstring j_spill_directories = env->NewStringUTF(join(spill_directories, ":").c_str()); + + jclass string_cls = env->FindClass("java/lang/String"); + jobjectArray j_cols = + env->NewObjectArray(static_cast(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(i), str); + env->DeleteLocalRef(str); + } + + env->CallVoidMethod(_jni_writer_obj, open_id, j_serialized_table, j_hadoop_config, j_cols, + static_cast(sink.transaction_id), j_commit_user, + static_cast(sink.write_mode == TPaimonWriteMode::OVERWRITE), + j_time_zone, j_spill_directories, + static_cast(_memory_manager->memory_limit()), + reinterpret_cast(_memory_manager.get())); + Status st = _check_jni_exception(env, "open"); + + env->DeleteLocalRef(j_serialized_table); + env->DeleteLocalRef(j_hadoop_config); + env->DeleteLocalRef(j_commit_user); + env->DeleteLocalRef(j_time_zone); + env->DeleteLocalRef(j_spill_directories); + env->DeleteLocalRef(j_cols); + env->DeleteLocalRef(string_cls); + + if (st.ok()) { + _opened = true; + _refresh_memory_profile(); + LOG(INFO) << "Paimon JNI writer memory limit: " + << PrettyPrinter::print_bytes(_memory_manager->memory_limit()) + << ", local_sink_count=" << std::max(1, state->num_local_sink()); + } + return st; +} + +Status JniPaimonWriteBackend::create_writer( // NOLINT(readability-make-member-function-const) + std::unique_ptr* writer) { + DORIS_CHECK(_opened); + *writer = std::make_unique(_jni_writer_obj, _write_id, _prepare_commit_id, + _abort_id, std::make_unique>(), + _sink); + return Status::OK(); +} + +JniPaimonWriter::JniPaimonWriter(jobject jni_writer_obj, jmethodID write_id, + jmethodID prepare_commit_id, jmethodID abort_id, + std::unique_ptr> arrow_pool, + TPaimonTableSink sink) + : _jni_writer_obj(jni_writer_obj), + _write_id(write_id), + _prepare_commit_id(prepare_commit_id), + _abort_id(abort_id), + _arrow_pool(std::move(arrow_pool)), + _sink(std::move(sink)) {} + +Status JniPaimonWriter::_write_projected_block(RuntimeState* state, Block& block) { + if (block.rows() == 0) { + return Status::OK(); + } + + // Use Thrift column_names as the authoritative schema source for both + // Arrow schema construction and Java-side write type derivation. + DORIS_CHECK(_sink.__isset.column_names); + DORIS_CHECK_EQ(_sink.column_names.size(), block.columns()); + for (size_t i = 0; i < _sink.column_names.size(); ++i) { + block.get_by_position(i).name = _sink.column_names[i]; + } + + // Pipeline: Doris Block → Arrow Schema → Arrow RecordBatch → IPC Stream → JNI direct buffer + // + // Step 1: Build Arrow schema from the projected Block. + // Paimon write timestamps are transported as civil-time fields. The Java writer uses the + // pinned Paimon target type to preserve NTZ values or convert LTZ values with the session zone. + // Variant V2 is transported losslessly as its value/metadata pair, including nested Variant. + std::shared_ptr arrow_schema; + RETURN_IF_ERROR(get_paimon_arrow_schema_from_block(block, &arrow_schema)); + + // Step 2: Convert Doris Block columns to an Arrow RecordBatch. + std::shared_ptr record_batch; + RETURN_IF_ERROR(convert_to_arrow_batch(block, arrow_schema, _arrow_pool.get(), &record_batch, + state->timezone_obj())); + + // Step 3: Serialize the RecordBatch to Arrow IPC Stream format in memory. + auto out_stream_res = arrow::io::BufferOutputStream::Create(4096, _arrow_pool.get()); + if (!out_stream_res.ok()) { + return Status::InternalError("Arrow BufferOutputStream create failed: {}", + out_stream_res.status().ToString()); + } + auto out_stream = *out_stream_res; + + auto writer_res = arrow::ipc::MakeStreamWriter(out_stream, arrow_schema); + if (!writer_res.ok()) { + return Status::InternalError("Arrow StreamWriter create failed: {}", + writer_res.status().ToString()); + } + auto ipc_writer = *writer_res; + if (!ipc_writer->WriteRecordBatch(*record_batch).ok()) { + return Status::InternalError("Arrow WriteRecordBatch failed"); + } + if (!ipc_writer->Close().ok()) { + return Status::InternalError("Arrow StreamWriter close failed"); + } + + auto buffer_res = out_stream->Finish(); + if (!buffer_res.ok()) { + return Status::InternalError("Arrow output stream finish failed: {}", + buffer_res.status().ToString()); + } + std::shared_ptr buffer = *buffer_res; + + // Step 4: Wrap the IPC buffer in a JNI direct ByteBuffer (zero-copy) and + // call PaimonJniWriter.write(ByteBuffer). Java side reads the Arrow IPC + // stream via ArrowStreamReader. + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + + jobject direct_buffer = + env->NewDirectByteBuffer(buffer->mutable_data(), static_cast(buffer->size())); + RETURN_IF_ERROR(Jni::Env::GetJniExceptionMsg( + env, false, "JNI exception in NewDirectByteBuffer for PaimonJniWriter::write: ")); + + env->CallVoidMethod(_jni_writer_obj, _write_id, direct_buffer); + Status write_status = + Jni::Env::GetJniExceptionMsg(env, false, "JNI exception in JniPaimonWriter::write: "); + env->DeleteLocalRef(direct_buffer); + return write_status; +} + +Status JniPaimonWriter::write(RuntimeState* state, Block& block) { + return _write_projected_block(state, block); +} + +Status JniPaimonWriter::prepare_commit(std::vector& messages) { + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + + // Call PaimonJniWriter.prepareCommit() which returns byte[][] — + // each element is a DPCM-framed serialized CommitMessage chunk produced + // by PaimonCommitCodec.encode(). + jobject j_payloads_obj = env->CallObjectMethod(_jni_writer_obj, _prepare_commit_id); + Status st = Jni::Env::GetJniExceptionMsg(env, false, "JNI exception in prepareCommit: "); + if (!st.ok()) { + return st; + } + + if (j_payloads_obj == nullptr) { + return Status::InternalError("PaimonJniWriter.prepareCommit returned null"); + } + + // Unpack the byte[][] into TPaimonCommitMessage structs for FE transport. + auto* j_payloads = static_cast(j_payloads_obj); + jsize num_payloads = env->GetArrayLength(j_payloads); + + for (jsize i = 0; i < num_payloads; ++i) { + auto j_bytes = static_cast(env->GetObjectArrayElement(j_payloads, i)); + if (j_bytes == nullptr) { + env->DeleteLocalRef(j_payloads); + return Status::InternalError("PaimonJniWriter.prepareCommit returned a null payload"); + } + jsize len = env->GetArrayLength(j_bytes); + if (len == 0) { + env->DeleteLocalRef(j_bytes); + env->DeleteLocalRef(j_payloads); + return Status::InternalError("PaimonJniWriter.prepareCommit returned an empty payload"); + } + jbyte* bytes = env->GetByteArrayElements(j_bytes, nullptr); + if (bytes == nullptr) { + env->DeleteLocalRef(j_bytes); + env->DeleteLocalRef(j_payloads); + RETURN_IF_ERROR(Jni::Env::GetJniExceptionMsg( + env, false, "JNI exception while reading Paimon commit payload: ")); + return Status::InternalError("Failed to read Paimon commit payload"); + } + std::string payload(reinterpret_cast(bytes), static_cast(len)); + TPaimonCommitMessage msg; + msg.__set_payload(payload); + messages.emplace_back(std::move(msg)); + env->ReleaseByteArrayElements(j_bytes, bytes, JNI_ABORT); + env->DeleteLocalRef(j_bytes); + } + env->DeleteLocalRef(j_payloads); + return Status::OK(); +} + +Status JniPaimonWriter::abort() { + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + env->CallVoidMethod(_jni_writer_obj, _abort_id); + return Jni::Env::GetJniExceptionMsg(env, true, "JNI exception in abort: "); +} + +void JniPaimonWriteBackend::_refresh_memory_profile() { + if (_memory_manager == nullptr) { + return; + } + COUNTER_SET(_native_page_memory_limit, _memory_manager->memory_limit()); + COUNTER_SET(_native_page_memory_peak, _memory_manager->native_peak_allocated_bytes()); +} + +} // 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 new file mode 100644 index 00000000000000..46cea7da85f7a6 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h @@ -0,0 +1,110 @@ +// 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 +#include + +#include +#include + +#include "common/status.h" +#include "exec/sink/writer/paimon/paimon_jni_memory_manager.h" +#include "exec/sink/writer/paimon/paimon_write_backend.h" +#include "format/parquet/arrow_memory_pool.h" +#include "runtime/runtime_profile.h" + +namespace doris { + +class RuntimeState; + +/// JNI backend that owns the Java PaimonJniWriter object and its JNI method +/// handles. Creates lightweight JniPaimonWriter adapters that share this +/// backend's JVM connection. +/// +/// Each JniPaimonWriteBackend corresponds to one Java PaimonJniWriter +/// instance; the JniPaimonWriter adapters are thin wrappers that delegate +/// write/prepare_commit/abort calls through the cached JNI method IDs. JNI-only +/// memory ownership and Profile counters stay here and are not part of the +/// common backend contract. +class JniPaimonWriteBackend final : public IPaimonWriteBackend { +public: + ~JniPaimonWriteBackend() override; + + Status open(const TPaimonTableSink& sink, RuntimeState* state, + RuntimeProfile* profile) override; + Status create_writer(std::unique_ptr* writer) override; + Status close() override; + + /// Abort prepared files without retaining the original Java writer object. + static Status abort_prepared_commit(const TPaimonTableSink& sink, + const std::vector& commit_messages); + +private: + static Status _check_jni_exception(JNIEnv* env, const std::string& method_name); + static Status _load_writer_class(JNIEnv* env, jclass* writer_class); + void _refresh_memory_profile(); + + // JNI global references — live for the duration of this backend. + jclass _jni_writer_cls = nullptr; + jobject _jni_writer_obj = nullptr; + + // Cached JNI method IDs for the PaimonJniWriter Java methods. + jmethodID _write_id = nullptr; + jmethodID _prepare_commit_id = nullptr; + jmethodID _abort_id = nullptr; + jmethodID _close_id = nullptr; + + TPaimonTableSink _sink; + std::unique_ptr _memory_manager; + RuntimeProfile::Counter* _native_page_memory_limit = nullptr; + RuntimeProfile::Counter* _native_page_memory_peak = nullptr; + bool _opened = false; +}; + +/// Lightweight C++ adapter that delegates to the shared JNI backend. +/// +/// Owns the Arrow memory pool used for Block → Arrow IPC conversion. +/// Each JniPaimonWriter is created by JniPaimonWriteBackend::create_writer() +/// and shares the backend's JNI method IDs and Java writer object reference. +class JniPaimonWriter final : public IPaimonWriter { +public: + JniPaimonWriter(jobject jni_writer_obj, jmethodID write_id, jmethodID prepare_commit_id, + jmethodID abort_id, std::unique_ptr> arrow_pool, + TPaimonTableSink sink); + + Status write(RuntimeState* state, Block& block) override; + Status prepare_commit(std::vector& messages) override; + Status abort() override; + +private: + /// Convert Block → Arrow RecordBatch → IPC Stream, then pass to Java via JNI direct buffer. + Status _write_projected_block(RuntimeState* state, Block& block); + + // Shared JNI state (owned by JniPaimonWriteBackend, not this adapter). + jobject _jni_writer_obj; + jmethodID _write_id; + jmethodID _prepare_commit_id; + jmethodID _abort_id; + + // Arrow resources owned by this writer adapter. + std::unique_ptr> _arrow_pool; + TPaimonTableSink _sink; +}; + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.cpp b/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.cpp new file mode 100644 index 00000000000000..63eae9904e8c73 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.cpp @@ -0,0 +1,304 @@ +// 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_jni_memory_manager.h" + +#include +#include +#include +#include + +#include "common/check.h" +#include "common/config.h" +#include "common/exception.h" +#include "common/logging.h" +#include "core/allocator.h" +#include "runtime/memory/mem_tracker_limiter.h" +#include "runtime/query_context.h" +#include "runtime/runtime_state.h" +#include "runtime/thread_context.h" +#include "util/defer_op.h" +#include "util/jni-util.h" +#include "util/pretty_printer.h" + +namespace doris { + +class PaimonJniMemoryManager::Impl { +public: + Impl(std::shared_ptr resource_context, int64_t memory_limit) + : _resource_context(std::move(resource_context)), _memory_limit(memory_limit) { + DORIS_CHECK(_resource_context != nullptr); + DORIS_CHECK(_memory_limit > 0); + } + + ~Impl() { + // Java may retain direct buffers until its writer is closed. Release + // every outstanding page here as the final native ownership boundary. + try { + release_all_pages(); + } catch (const std::exception& e) { + LOG(WARNING) << "Failed to release Paimon JNI native memory: " << e.what(); + } catch (...) { + LOG(WARNING) << "Failed to release Paimon JNI native memory: unknown exception"; + } + } + + jobject allocate_page(JNIEnv* env, jint bytes) { + if (bytes <= 0) { + throw Exception(Status::InvalidArgument( + "Paimon JNI memory page size must be positive, actual={}", bytes)); + } + + // Reserve the writer-local budget before entering the allocator. This + // prevents concurrent JNI callbacks from transiently allocating past + // the configured cap and only discovering it after query accounting + // or the system allocator has already rejected the request. + { + std::lock_guard lock(_mutex); + if (bytes > _memory_limit - _native_allocated_bytes - _native_reserved_bytes) { + throw Exception(Status::Error( + "Paimon JNI write buffer exceeded its {} native memory limit", + PrettyPrinter::print_bytes(_memory_limit))); + } + _native_reserved_bytes += bytes; + } + bool reservation_committed = false; + Defer rollback_reservation {[&]() { + if (!reservation_committed) { + std::lock_guard lock(_mutex); + _native_reserved_bytes -= bytes; + } + }}; + + // Allocate and account while attached to the query's resource + // context. The callback can run on a JVM-created thread, so merely + // relying on the calling BE thread's context would bypass query + // memory accounting. + void* address = with_resource_context([&]() { + enable_thread_catch_bad_alloc++; + Defer restore_bad_alloc_catch {[&]() { enable_thread_catch_bad_alloc--; }}; + void* allocated = _allocator.alloc(static_cast(bytes)); + try { + std::lock_guard lock(_mutex); + _allocations.emplace_back(allocated, static_cast(bytes)); + _native_reserved_bytes -= bytes; + _native_allocated_bytes += bytes; + _native_peak_allocated_bytes = + std::max(_native_peak_allocated_bytes, _native_allocated_bytes); + reservation_committed = true; + } catch (...) { + _allocator.free(allocated, static_cast(bytes)); + throw; + } + return allocated; + }); + + // NewDirectByteBuffer does not copy memory; Paimon will read/write the + // page directly. If JNI rejects the address, undo the native + // allocation and its accounting entry before returning. + jobject buffer = env->NewDirectByteBuffer(address, bytes); + if (buffer == nullptr || env->ExceptionCheck()) { + remove_and_free_page(address, static_cast(bytes)); + return nullptr; + } + return buffer; + } + + int64_t memory_limit() const { return _memory_limit; } + + int64_t native_peak_allocated_bytes() const { + std::lock_guard lock(_mutex); + return _native_peak_allocated_bytes; + } + +private: + template + auto with_resource_context(Function&& function) + -> decltype(std::forward(function)()) { + // JNI normally re-enters on an attached async-writer thread. Attach + // Java-created threads explicitly too, so every allocation/free is + // charged to the query rather than to an unrelated thread context. + if (!pthread_context_ptr_init && bthread_self() == 0) { + SCOPED_ATTACH_TASK(_resource_context); + return std::forward(function)(); + } + if (thread_context()->is_attach_task()) { + SCOPED_SWITCH_RESOURCE_CONTEXT(_resource_context); + return std::forward(function)(); + } + SCOPED_ATTACH_TASK(_resource_context); + return std::forward(function)(); + } + + void release_all_pages() { + // Detach ownership from the bookkeeping vector under the lock, then + // free outside the lock. Allocator/free may invoke code that takes + // unrelated locks and must not block page accounting readers. + std::vector> allocations; + { + std::lock_guard lock(_mutex); + allocations.swap(_allocations); + _native_allocated_bytes = 0; + } + if (allocations.empty()) { + return; + } + + with_resource_context([&]() { + for (const auto& [address, bytes] : allocations) { + _allocator.free(address, bytes); + } + std::vector>().swap(allocations); + }); + } + + void remove_and_free_page(void* address, size_t bytes) { + // Roll back a page whose Java direct-buffer wrapper could not be + // created. The address is removed under the same lock used by the + // normal accounting path, while the potentially expensive free is + // performed after releasing it. + { + std::lock_guard lock(_mutex); + auto it = std::find_if( + _allocations.begin(), _allocations.end(), + [&](const auto& allocation) { return allocation.first == address; }); + if (it != _allocations.end()) { + _allocations.erase(it); + _native_allocated_bytes -= bytes; + } + } + with_resource_context([&]() { _allocator.free(address, bytes); }); + } + + // Query resource context used for all native allocator operations. + std::shared_ptr _resource_context; + // Immutable per-writer cap, calculated by PaimonJniMemoryManager::create. + const int64_t _memory_limit; + // Doris allocator used instead of JVM/Arrow allocation so native pages are + // visible to Doris' memory accounting and allocator hooks. + Allocator _allocator; + // Protects the allocation list and both usage counters. JNI callbacks and + // Java close/finalizer paths may arrive concurrently. + mutable std::mutex _mutex; + // Every entry is (native address, size) and remains here until released. + std::vector> _allocations; + // Bytes reserved by callbacks which have passed the local limit check but + // have not yet completed their allocator call. + int64_t _native_reserved_bytes = 0; + // Committed and high-water native page usage, respectively. + int64_t _native_allocated_bytes = 0; + int64_t _native_peak_allocated_bytes = 0; +}; + +namespace { + +jobject allocate_paimon_memory_page(JNIEnv* env, jclass, jlong manager_handle, jint bytes) { + // This is called from PaimonJniWriter's Java memory pool. The handle is + // the native manager address passed when the writer is opened; ownership + // stays with the C++ writer/backend, so this callback must never delete it. + auto* manager = reinterpret_cast(manager_handle); + if (manager == nullptr) { + jclass exception_class = env->FindClass("java/lang/IllegalStateException"); + env->ThrowNew(exception_class, "Paimon JNI memory manager is null"); + env->DeleteLocalRef(exception_class); + return nullptr; + } + try { + return manager->allocate_page(env, bytes); + } catch (const std::exception& e) { + jclass exception_class = env->FindClass("java/lang/OutOfMemoryError"); + env->ThrowNew(exception_class, e.what()); + env->DeleteLocalRef(exception_class); + return nullptr; + } +} + +} // namespace + +PaimonJniMemoryManager::PaimonJniMemoryManager(std::unique_ptr impl) + : _impl(std::move(impl)) {} + +PaimonJniMemoryManager::~PaimonJniMemoryManager() = default; + +Status PaimonJniMemoryManager::create(RuntimeState* state, + std::unique_ptr* manager) { + DORIS_CHECK(state != nullptr); + DORIS_CHECK(manager != nullptr); + if (state->query_mem_tracker() == nullptr) { + return Status::InternalError( + "Paimon JNI writer cannot size its write buffer without a query tracker"); + } + if (state->get_query_ctx() == nullptr) { + return Status::InternalError( + "Paimon JNI writer cannot allocate native memory without QueryContext"); + } + + // A query can create multiple local sink instances. Divide its budget + // before applying the configured cap so one writer cannot consume the + // entire query allowance. + const int64_t writer_count = std::max(1, state->num_local_sink()); + const int64_t query_limit = state->query_mem_tracker()->limit(); + const int64_t query_share = query_limit > 0 ? query_limit / writer_count : query_limit; + const int64_t configured_memory_limit = config::paimon_jni_writer_memory_pool_limit_bytes; + const int64_t memory_limit = query_share > 0 ? std::min(query_share, configured_memory_limit) + : configured_memory_limit; + if (memory_limit <= 0) { + return Status::Error( + "Paimon JNI writer has insufficient memory budget: query_limit={}, " + "local_sink_count={}, write_buffer_limit={}", + PrettyPrinter::print_bytes(query_limit), writer_count, + PrettyPrinter::print_bytes(memory_limit)); + } + + // ResourceContext is retained by Impl for the manager's whole lifetime; + // this is what keeps asynchronous JNI callbacks associated with the query. + auto impl = std::make_unique(state->get_query_ctx()->resource_ctx(), memory_limit); + *manager = std::unique_ptr(new PaimonJniMemoryManager(std::move(impl))); + return Status::OK(); +} + +Status PaimonJniMemoryManager::register_natives(JNIEnv* env, jclass writer_class) { + // Keep the JNI surface minimal: Java asks native code only for a page; + // all ownership, limits, and cleanup stay in PaimonJniMemoryManager. + static char allocate_name[] = "allocatePaimonMemoryPage"; + static char allocate_signature[] = "(JI)Ljava/nio/ByteBuffer;"; + static ::JNINativeMethod methods[] = { + {allocate_name, allocate_signature, + reinterpret_cast(&allocate_paimon_memory_page)}, + }; + if (env->RegisterNatives(writer_class, methods, + static_cast(sizeof(methods) / sizeof(methods[0]))) != JNI_OK) { + RETURN_IF_ERROR(Jni::Env::GetJniExceptionMsg( + env, true, "JNI exception registering Paimon memory native methods: ")); + return Status::JniError("Failed to register Paimon memory native methods"); + } + return Status::OK(); +} + +jobject PaimonJniMemoryManager::allocate_page(JNIEnv* env, jint bytes) { + return _impl->allocate_page(env, bytes); +} + +int64_t PaimonJniMemoryManager::memory_limit() const { + return _impl->memory_limit(); +} + +int64_t PaimonJniMemoryManager::native_peak_allocated_bytes() const { + return _impl->native_peak_allocated_bytes(); +} + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.h b/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.h new file mode 100644 index 00000000000000..0d4818cd6dce42 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.h @@ -0,0 +1,81 @@ +// 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 + +#include +#include + +#include "common/status.h" + +namespace doris { + +class RuntimeState; + +/// Owns the Doris-side native memory used by one Java Paimon writer. +/// +/// Paimon's sort/merge buffers are Java objects, but their page storage is +/// requested through a JNI callback. This manager is the bridge for that +/// callback: it allocates each page with Doris' allocator, exposes the page as +/// a direct ByteBuffer, tracks it until the writer is closed, and releases all +/// pages in its destructor. The native writer/backend therefore keeps this +/// manager alive for at least as long as the Java writer can access its +/// callback handle. +/// +/// The limit is a per-writer budget. It is derived from the query memory +/// limit and the number of local sink instances, then capped by the global +/// Paimon JNI configuration. The manager accounts only for pages allocated +/// by this callback; Java heap and other Paimon-managed memory remain under +/// their respective runtimes. +class PaimonJniMemoryManager { +public: + ~PaimonJniMemoryManager(); + + /// Construct a manager whose budget is sized from the query context. + /// + /// The query must provide both a memory tracker and QueryContext. The + /// latter supplies the ResourceContext used whenever allocation/freeing + /// crosses into a JNI-created or asynchronous thread. + static Status create(RuntimeState* state, std::unique_ptr* manager); + /// Register the static JNI callback used by PaimonJniWriter. + static Status register_natives(JNIEnv* env, jclass writer_class); + + /// Allocate one native page and return it as a direct ByteBuffer. + /// + /// On failure this method leaves no accounting entry behind and reports + /// the error through the JNI environment. The returned buffer remains + /// valid until the manager is destroyed (or allocation of that page is + /// rolled back because NewDirectByteBuffer failed). + jobject allocate_page(JNIEnv* env, jint bytes); + + /// Return the immutable per-writer native page budget in bytes. + int64_t memory_limit() const; + + /// Return the high-water mark of native pages allocated by this manager. + int64_t native_peak_allocated_bytes() const; + +private: + class Impl; + + explicit PaimonJniMemoryManager(std::unique_ptr impl); + + std::unique_ptr _impl; +}; + +} // 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 new file mode 100644 index 00000000000000..66238cc621760a --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_table_writer.cpp @@ -0,0 +1,193 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "exec/sink/writer/paimon/paimon_table_writer.h" + +#include "common/check.h" +#include "common/logging.h" +#include "core/block/block.h" +#include "exec/sink/writer/paimon/jni_paimon_write_backend.h" +#include "runtime/runtime_state.h" + +namespace doris { + +PaimonTableWriter::PaimonTableWriter(TDataSink t_sink, const VExprContextSPtrs& output_exprs, + std::shared_ptr dep, + std::shared_ptr fin_dep) + : AsyncResultWriter(output_exprs, std::move(dep), std::move(fin_dep)), + _t_sink(std::move(t_sink)) { + DCHECK(_t_sink.__isset.paimon_table_sink); +} + +Status PaimonTableWriter::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"); + _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"); + _commit_payload_count = ADD_COUNTER(_operator_profile, "CommitPayloadCount", TUnit::UNIT); + _commit_payload_bytes_counter = + ADD_COUNTER(_operator_profile, "CommitPayloadBytes", TUnit::BYTES); + + SCOPED_TIMER(_open_timer); + + // Step 1: Create the JNI backend that owns the Java Paimon SDK writer. + _backend = std::make_unique(); + // Step 2: Open the backend — for JNI this loads the Java class and calls PaimonJniWriter.open(). + RETURN_IF_ERROR(_backend->open(_t_sink.paimon_table_sink, state, profile)); + // Step 3: Create a lightweight writer adapter that delegates to the opened backend. + RETURN_IF_ERROR(_backend->create_writer(&_writer)); + DCHECK(_writer); + + LOG(INFO) << "PaimonTableWriter opened: backend=JNI, writer_scope=local_state"; + return Status::OK(); +} + +Status PaimonTableWriter::write(RuntimeState* state, Block& block) { + if (block.rows() == 0) { + return Status::OK(); + } + + SCOPED_TIMER(_send_data_timer); + + // Step 1: Apply output expressions to produce the columns selected by FE. + Block output_block; + { + SCOPED_TIMER(_project_timer); + RETURN_IF_ERROR(_projection_block(block, &output_block)); + } + + COUNTER_UPDATE(_written_rows_counter, block.rows()); + COUNTER_UPDATE(_written_bytes_counter, block.bytes()); + _state->update_num_rows_load_total(block.rows()); + _state->update_num_bytes_load_total(block.bytes()); + + // Step 2: Convert Block → Arrow IPC → direct buffer → Java PaimonJniWriter. + DCHECK(_writer); + { + SCOPED_TIMER(_file_store_write_timer); + RETURN_IF_ERROR(_writer->write(_state, output_block)); + } + _written_rows += block.rows(); + return Status::OK(); +} + +Status PaimonTableWriter::close(Status status) { + SCOPED_TIMER(_close_timer); + + // Prepare messages first, but do not publish them until the backend confirms + // that every SDK user has stopped and its native backing memory is safe to release. + std::vector messages; + if (status.ok()) { + DCHECK(_writer); + { + SCOPED_TIMER(_prepare_commit_timer); + Status prep_st = _writer->prepare_commit(messages); + if (!prep_st.ok()) { + status = prep_st; + } + } + } + + // If prepare_commit failed or the incoming status was already an error, + // abort the writer to clean up uncommitted data files. + if (!status.ok()) { + LOG(WARNING) << "Paimon writer closing with error: " << status.to_string(); + if (_writer) { + Status abort_st = _writer->abort(); + if (!abort_st.ok()) { + LOG(WARNING) << "Paimon writer abort failed: " << abort_st.to_string(); + } + } + } + + // Record message metrics before backend shutdown, but retain local ownership until + // every Java SDK user has stopped successfully. + if (status.ok() && !messages.empty()) { + messages.front().__set_row_count(_written_rows); + COUNTER_UPDATE(_commit_payload_count, static_cast(messages.size())); + for (const auto& msg : messages) { + DORIS_CHECK(msg.__isset.payload); + COUNTER_UPDATE(_commit_payload_bytes_counter, static_cast(msg.payload.size())); + } + } + + // The adapter only owns Arrow conversion resources. Release it before closing + // the backend, whose Java close is the authoritative SDK shutdown boundary. + _writer.reset(); + + if (_backend) { + Status close_st = _backend->close(); + if (!close_st.ok()) { + if (status.ok()) { + status = close_st; + } else { + LOG(WARNING) << "Paimon backend close also failed: " << close_st.to_string(); + } + } + } + + // A clean backend close is the ownership boundary. On any failure, the FE must never + // observe these messages; use an independent committer to clean prepared files because + // the original Java writer is already closed (or its close outcome is unsafe). + if (!status.ok() && !messages.empty()) { + WARN_IF_ERROR( + JniPaimonWriteBackend::abort_prepared_commit(_t_sink.paimon_table_sink, messages), + "failed to abort Paimon files after backend close failure"); + } + + _backend.reset(); + + if (!status.ok() || messages.empty()) { + return status; + } + + // Transfer payload ownership only after backend shutdown. If the report budget rejects + // the transfer, abort immediately. If FE later explicitly rejects the final report, the + // callback reads the same RuntimeState payloads and aborts them without retaining a second copy. + Status publish_status = _state->add_paimon_commit_messages(messages); + if (!publish_status.ok()) { + WARN_IF_ERROR( + JniPaimonWriteBackend::abort_prepared_commit(_t_sink.paimon_table_sink, messages), + "failed to abort Paimon files after report-budget rejection"); + return publish_status; + } + + RuntimeState* cleanup_state = _state; + TPaimonTableSink cleanup_sink = _t_sink.paimon_table_sink; + _state->add_rejected_external_file_report_cleanup([cleanup_state, + cleanup_sink = std::move(cleanup_sink)] { + std::vector rejected_messages; + cleanup_state->append_paimon_commit_messages(&rejected_messages); + WARN_IF_ERROR(JniPaimonWriteBackend::abort_prepared_commit(cleanup_sink, rejected_messages), + "failed to abort Paimon files after final report rejection"); + }); + + LOG(INFO) << "Paimon writer closed: " << messages.size() + << " commit messages, total rows=" << _written_rows; + return status; +} + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_table_writer.h b/be/src/exec/sink/writer/paimon/paimon_table_writer.h new file mode 100644 index 00000000000000..850d5dea4b001f --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_table_writer.h @@ -0,0 +1,102 @@ +// 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 + +#include +#include + +#include "common/status.h" +#include "core/block/block.h" +#include "exec/sink/writer/async_result_writer.h" +#include "exec/sink/writer/paimon/paimon_write_backend.h" +#include "exprs/vexpr_fwd.h" +#include "runtime/runtime_profile.h" + +namespace doris { + +class RuntimeState; + +/// Each PaimonTableSinkLocalState owns one PaimonTableWriter, which in turn +/// owns one IPaimonWriteBackend and one IPaimonWriter. Pipeline parallelism +/// therefore determines the number of independent Paimon writer sessions; +/// each writer session delegates partition and bucket routing to the Paimon +/// Java SDK through JNI. FE currently places Paimon writes on GATHER to keep +/// dynamic-bucket assignment single-writer correct. +/// +/// Doris does NOT compute partition values or bucket ids — it passes complete +/// Blocks through the JNI backend to the Paimon SDK, which +/// internally computes partition values, bucket ids, and routes rows to the +/// correct file writers. +/// +/// Architecture: +/// PaimonTableSinkOperatorX +/// │ sink_impl() → AsyncWriterSink::sink() (no routing) +/// ▼ +/// PaimonTableWriter (one per LocalState / pipeline instance) +/// │ owns IPaimonWriteBackend (JNI) +/// │ └─ create_writer() → IPaimonWriter +/// │ write() +/// │ → JNI backend: Block → Arrow IPC → Java Paimon SDK +/// │ → Paimon SDK owns row normalization, routing, buffering, +/// │ file writing, and compaction +/// ▼ +/// close() → prepareCommit() → CommitMessage[] +/// +/// Commit flow (BE only prepares messages; FE is the commit coordinator): +/// close() → writer->prepare_commit() +/// → collect TPaimonCommitMessage[] (DPCM-framed serialized messages) +/// → RuntimeState::add_paimon_commit_messages() +/// → RPC to FE Coordinator → PaimonTransaction +class PaimonTableWriter final : public AsyncResultWriter { +public: + PaimonTableWriter(TDataSink t_sink, const VExprContextSPtrs& output_exprs, + std::shared_ptr dep, std::shared_ptr fin_dep); + + ~PaimonTableWriter() override = default; + + Status open(RuntimeState* state, RuntimeProfile* profile) override; + + Status write(RuntimeState* state, Block& block) override; + + Status close(Status status) override; + +private: + TDataSink _t_sink; + RuntimeState* _state = nullptr; + + // Backend owns the JNI connection and creates the writer adapter. + // Both are scoped to this PaimonTableWriter (one per LocalState). + std::unique_ptr _backend; + std::unique_ptr _writer; + + // Profile counters + RuntimeProfile::Counter* _written_rows_counter = nullptr; + RuntimeProfile::Counter* _written_bytes_counter = nullptr; + RuntimeProfile::Counter* _send_data_timer = nullptr; + RuntimeProfile::Counter* _project_timer = nullptr; + RuntimeProfile::Counter* _file_store_write_timer = nullptr; + RuntimeProfile::Counter* _open_timer = nullptr; + RuntimeProfile::Counter* _close_timer = nullptr; + RuntimeProfile::Counter* _prepare_commit_timer = nullptr; + RuntimeProfile::Counter* _commit_payload_count = nullptr; + RuntimeProfile::Counter* _commit_payload_bytes_counter = nullptr; +}; + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_write_backend.h b/be/src/exec/sink/writer/paimon/paimon_write_backend.h new file mode 100644 index 00000000000000..ade863aec8cde1 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_write_backend.h @@ -0,0 +1,84 @@ +// 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 + +#include +#include + +#include "common/status.h" +#include "core/block/block.h" + +namespace doris { + +class RuntimeState; +class RuntimeProfile; + +/// Writer contract implemented by one SDK writer adapter. Each +/// PaimonTableWriter owns one IPaimonWriter, which delegates to the +/// Paimon Java SDK through JNI. Partition and bucket routing happens +/// inside the SDK. +/// +/// Lifecycle: created by IPaimonWriteBackend::create_writer() after the +/// backend is opened; used for the duration of one pipeline instance. +class IPaimonWriter { +public: + virtual ~IPaimonWriter() = default; + + /// Write a projected Block to the Paimon SDK. + /// For the JNI path: Block → Arrow IPC → direct buffer → 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. + virtual Status prepare_commit(std::vector& messages) = 0; + + /// Discard written data files on error. Called when write or prepare_commit fails. + virtual Status abort() = 0; +}; + +/// Backend boundary for creating writers through the Paimon Java SDK. +/// +/// The backend owns the JVM class reference, method IDs, and Java writer object. +/// +/// Each backend creates one or more IPaimonWriter adapters that share the +/// same underlying connection. Snapshot commit is deliberately excluded from +/// this boundary: BE only prepares commit messages (byte payloads), while FE +/// PaimonTransaction is the single commit coordinator. +class IPaimonWriteBackend { +public: + virtual ~IPaimonWriteBackend() = default; + + /// Initialize the backend connection. For JNI this loads the writer class, + /// creates the Java object, and calls PaimonJniWriter.open(). + virtual Status open(const TPaimonTableSink& sink, RuntimeState* state, + RuntimeProfile* profile) = 0; + + /// Create a lightweight writer adapter that delegates to this backend. + virtual Status create_writer(std::unique_ptr* writer) = 0; + + /// Stop all SDK users and release backend resources. + /// + /// A successful return is the ownership boundary after which native memory + /// backing SDK buffers can be reclaimed safely. Callers must not publish + /// prepared commit messages until this succeeds. + virtual Status close() = 0; +}; + +} // namespace doris diff --git a/be/src/format/jni/jni_data_bridge.cpp b/be/src/format/jni/jni_data_bridge.cpp index 9dc935e0b62da5..67d2e2159c64db 100644 --- a/be/src/format/jni/jni_data_bridge.cpp +++ b/be/src/format/jni/jni_data_bridge.cpp @@ -22,6 +22,7 @@ #include #include +#include "common/config.h" #include "core/block/block.h" #include "core/column/column_array.h" #include "core/column/column_map.h" @@ -29,6 +30,8 @@ #include "core/column/column_string.h" #include "core/column/column_struct.h" #include "core/column/column_varbinary.h" +#include "core/column/column_variant.h" +#include "core/column/variant_v2/column_variant_v2.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" @@ -38,6 +41,9 @@ #include "core/data_type/primitive_type.h" #include "core/types.h" #include "core/value/decimalv2_value.h" +#include "core/value/variant/variant_batch_builder.h" +#include "exec/common/variant_util.h" +#include "exprs/function/parse/variant_string_parse.h" #include "util/string_util.h" #include "util/url_coding.h" @@ -148,6 +154,10 @@ Status JniDataBridge::fill_column(TableMetaAddress& address, ColumnPtr& doris_co case PrimitiveType::TYPE_VARBINARY: status = _fill_varbinary_column(address, data_column, num_rows); break; + case PrimitiveType::TYPE_VARIANT: + status = _fill_variant_column(address, data_column, static_cast(null_map_ptr), + num_rows); + break; default: status = Status::InvalidArgument("Unsupported type {} in jni scanner", data_type->get_name()); @@ -179,6 +189,62 @@ Status JniDataBridge::_fill_varbinary_column(TableMetaAddress& address, return Status::OK(); } +Status JniDataBridge::_fill_variant_column(TableMetaAddress& address, + MutableColumnPtr& doris_column, const bool* null_map, + size_t num_rows) { + ColumnPtr values = ColumnVarbinary::create(); + ColumnPtr metadatas = ColumnVarbinary::create(); + const DataTypePtr binary_type = std::make_shared(); + RETURN_IF_ERROR(fill_column(address, values, binary_type, num_rows)); + RETURN_IF_ERROR(fill_column(address, metadatas, binary_type, num_rows)); + + RETURN_IF_CATCH_EXCEPTION({ + const auto& value_column = assert_cast(*values); + const auto& metadata_column = assert_cast(*metadatas); + VariantBatchBuilder builder; + for (size_t row_index = 0; row_index < num_rows; ++row_index) { + auto row = builder.begin_row(); + if (null_map[row_index]) { + row.add_null(); + } else { + const StringRef value = value_column.get_data_at(row_index); + const StringRef metadata = metadata_column.get_data_at(row_index); + row.add_value({.metadata = {.data = metadata.data, .size = metadata.size}, + .value = value}); + } + row.finish(); + } + VariantBatchBuilder batch = builder.finish_batch(); + if (auto* variant_v2 = check_and_get_column(doris_column.get())) { + variant_v2->insert_encoded_batch(batch); + } else { + auto* variant = check_and_get_column(doris_column.get()); + if (variant == nullptr) { + throw Exception( + ErrorCode::INVALID_ARGUMENT, + "JNI Variant destination requires ColumnVariant or ColumnVariantV2, got " + "{}", + doris_column->get_name()); + } + struct StringWriter { + void write(const char* data, size_t size) { value.append(data, size); } + std::string value; + }; + ParseConfig parse_config; + parse_config.check_duplicate_json_path = + config::variant_enable_duplicate_json_path_check; + for (size_t row_index = 0; row_index < num_rows; ++row_index) { + StringWriter writer; + to_json(batch.value_at(row_index), writer); + variant_util::parse_json_to_variant(*variant, + {writer.value.data(), writer.value.size()}, + nullptr, parse_config); + } + } + }); + return Status::OK(); +} + Status JniDataBridge::_fill_string_column(TableMetaAddress& address, MutableColumnPtr& doris_column, size_t num_rows) { auto& string_col = static_cast(*doris_column); @@ -358,6 +424,8 @@ std::string JniDataBridge::get_jni_type(const DataTypePtr& data_type) { } case TYPE_VARBINARY: return "varbinary"; + case TYPE_VARIANT: + return "struct"; // bitmap, hll, quantile_state, jsonb are transferred as strings via JNI case TYPE_BITMAP: [[fallthrough]]; @@ -433,6 +501,8 @@ std::string JniDataBridge::get_jni_type_with_different_string(const DataTypePtr& << assert_cast(remove_nullable(data_type).get())->len() << ")"; return buffer.str(); + case TYPE_VARIANT: + return "struct"; case TYPE_DECIMALV2: { buffer << "decimalv2(" << DecimalV2Value::PRECISION << "," << DecimalV2Value::SCALE << ")"; return buffer.str(); @@ -506,6 +576,8 @@ std::string JniDataBridge::encode_schema_values(const std::vector& std::string JniDataBridge::get_jni_type_with_encoded_struct_fields(const DataTypePtr& data_type) { switch (data_type->get_primitive_type()) { + case TYPE_VARIANT: + return "struct<$dmFsdWU=:varbinary,$bWV0YWRhdGE=:varbinary>"; case TYPE_STRUCT: { const auto* type_struct = assert_cast(remove_nullable(data_type).get()); diff --git a/be/src/format/jni/jni_data_bridge.h b/be/src/format/jni/jni_data_bridge.h index e037ffec3d4d5a..5a0eea8aa55b26 100644 --- a/be/src/format/jni/jni_data_bridge.h +++ b/be/src/format/jni/jni_data_bridge.h @@ -154,6 +154,9 @@ class JniDataBridge { static Status _fill_varbinary_column(TableMetaAddress& address, MutableColumnPtr& doris_column, size_t num_rows); + static Status _fill_variant_column(TableMetaAddress& address, MutableColumnPtr& doris_column, + const bool* null_map, size_t num_rows); + static Status _fill_array_column(TableMetaAddress& address, MutableColumnPtr& doris_column, const DataTypePtr& data_type, size_t num_rows); diff --git a/be/src/runtime/runtime_state.cpp b/be/src/runtime/runtime_state.cpp index 380bc8f8f72081..c02c7478a5a471 100644 --- a/be/src/runtime/runtime_state.cpp +++ b/be/src/runtime/runtime_state.cpp @@ -72,18 +72,52 @@ Status RuntimeState::add_iceberg_commit_datas(TIcebergCommitData iceberg_commit_ thrift_limit > report_envelope_headroom ? thrift_limit - report_envelope_headroom : 0; std::lock_guard budget_lock(_external_file_report_state->mutex); // Parallel task states share this budget because FE receives their vectors in one fragment report. - if (_external_file_report_state->iceberg_serialized_bytes + serialized_size + sizeof(uint32_t) > + if (_external_file_report_state->serialized_commit_bytes + serialized_size + sizeof(uint32_t) > commit_data_limit) { return Status::InternalError( "Iceberg commit metadata exceeds the Thrift report limit; reduce output file " "count"); } std::lock_guard data_lock(_iceberg_commit_datas_mutex); - _external_file_report_state->iceberg_serialized_bytes += serialized_size + sizeof(uint32_t); + _external_file_report_state->serialized_commit_bytes += serialized_size + sizeof(uint32_t); _iceberg_commit_datas.emplace_back(std::move(iceberg_commit_data)); return Status::OK(); } +Status RuntimeState::add_paimon_commit_messages(std::vector commit_messages) { + if (commit_messages.empty()) { + return Status::OK(); + } + + ThriftSerializer serializer(false, 256); + size_t messages_size = 0; + for (auto& message : commit_messages) { + uint32_t serialized_size = 0; + uint8_t* buffer = nullptr; + RETURN_IF_ERROR(serializer.serialize(&message, &serialized_size, &buffer)); + messages_size += serialized_size + sizeof(uint32_t); + } + + constexpr size_t report_envelope_headroom = 1024 * 1024; + const size_t thrift_limit = coordinator_thrift_message_limit(); + const size_t commit_data_limit = + thrift_limit > report_envelope_headroom ? thrift_limit - report_envelope_headroom : 0; + std::lock_guard budget_lock(_external_file_report_state->mutex); + if (messages_size > + commit_data_limit - + std::min(commit_data_limit, _external_file_report_state->serialized_commit_bytes)) { + return Status::InternalError( + "Paimon commit metadata exceeds the Thrift report limit; reduce output file " + "count"); + } + std::lock_guard data_lock(_paimon_commit_messages_mutex); + _external_file_report_state->serialized_commit_bytes += messages_size; + _paimon_commit_messages.insert(_paimon_commit_messages.end(), + std::make_move_iterator(commit_messages.begin()), + std::make_move_iterator(commit_messages.end())); + return Status::OK(); +} + size_t RuntimeState::coordinator_thrift_message_limit() const { int32_t effective_thrift_limit = std::max(config::thrift_max_message_size, 0); if (_query_options.__isset.coordinator_thrift_max_message_size && @@ -115,6 +149,10 @@ void RuntimeState::append_external_file_commit_data(TReportExecStatusParams* par params->mc_commit_datas.insert(params->mc_commit_datas.end(), commit_datas.begin(), commit_datas.end()); } + append_paimon_commit_messages(¶ms->paimon_commit_messages); + if (!params->paimon_commit_messages.empty()) { + params->__isset.paimon_commit_messages = true; + } } void RuntimeState::add_rejected_external_file_report_cleanup(std::function cleanup) { diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h index a3cfc5e4cad782..44e054a593070f 100644 --- a/be/src/runtime/runtime_state.h +++ b/be/src/runtime/runtime_state.h @@ -84,7 +84,7 @@ class ExternalFileReportState { private: std::mutex mutex; - size_t iceberg_serialized_bytes = 0; + size_t serialized_commit_bytes = 0; bool ownership_may_have_transferred = false; std::vector> rejected_report_cleanups; }; @@ -547,6 +547,14 @@ class RuntimeState { Status add_iceberg_commit_datas(TIcebergCommitData iceberg_commit_data); + void append_paimon_commit_messages(std::vector* output) const { + std::lock_guard lock(_paimon_commit_messages_mutex); + output->insert(output->end(), _paimon_commit_messages.begin(), + _paimon_commit_messages.end()); + } + + Status add_paimon_commit_messages(std::vector commit_messages); + size_t coordinator_thrift_message_limit() const; void append_external_file_commit_data(TReportExecStatusParams* params, bool final_report) const; @@ -1012,6 +1020,9 @@ class RuntimeState { mutable std::mutex _mc_commit_datas_mutex; std::vector _mc_commit_datas; + mutable std::mutex _paimon_commit_messages_mutex; + std::vector _paimon_commit_messages; + std::vector> _op_id_to_local_state; std::unique_ptr _sink_local_state; diff --git a/be/test/core/data_type_serde/data_type_serde_variant_test.cpp b/be/test/core/data_type_serde/data_type_serde_variant_test.cpp index 968062079d0094..ba9141661f07ff 100644 --- a/be/test/core/data_type_serde/data_type_serde_variant_test.cpp +++ b/be/test/core/data_type_serde/data_type_serde_variant_test.cpp @@ -15,13 +15,20 @@ // specific language governing permissions and limitations // under the License. +#include #include +#include #include #include +#include +#include +#include + #include "core/column/column_variant.h" #include "core/data_type_serde/data_type_variant_serde.h" #include "core/string_buffer.hpp" +#include "exprs/function/parse/variant_string_parse.h" #include "gen_cpp/types.pb.h" #include "util/mysql_row_buffer.h" #include "util/slice.h" @@ -64,4 +71,129 @@ TEST(VariantSerdeTest, BasicUnsupportedAndArrowPaths) { .ok()); } +TEST(VariantSerdeTest, StructArrowEncodesLegacyVariantAndSqlNull) { + DataTypeVariantSerDe serde; + auto column = ColumnVariant::create(0, false); + DataTypeSerDe::FormatOptions options; + const std::string json = R"({"id":7,"tags":["doris"]})"; + const std::vector json_rows {json, "null", json}; + for (const auto& json_row : json_rows) { + Slice slice(json_row.data(), json_row.size()); + ASSERT_TRUE(serde.deserialize_one_cell_from_json(*column, slice, options).ok()); + } + column->finalize(ColumnVariant::FinalizeMode::WRITE_MODE); + + auto value_builder = std::make_shared(); + auto metadata_builder = std::make_shared(); + auto arrow_type = arrow::struct_({arrow::field("value", arrow::binary(), false), + arrow::field("metadata", arrow::binary(), false)}); + arrow::StructBuilder builder(arrow_type, arrow::default_memory_pool(), + {value_builder, metadata_builder}); + NullMap forced_nulls {0, 0, 1}; + ASSERT_TRUE(serde.write_column_to_arrow(*column, &forced_nulls, &builder, 0, column->size(), + cctz::utc_time_zone()) + .ok()); + + std::shared_ptr output; + ASSERT_TRUE(builder.Finish(&output).ok()); + ASSERT_EQ(output->length(), 3); + EXPECT_FALSE(output->IsNull(0)); + EXPECT_FALSE(output->IsNull(1)); + EXPECT_TRUE(output->IsNull(2)); + + JsonStringToVariantEncoder expected_encoder({.max_json_key_length = 1024, + .throw_on_invalid_json = true, + .check_duplicate_json_path = false}); + expected_encoder.add_json({json.data(), json.size()}); + // Legacy VARIANT V1 materializes a root JSON null as an empty object. + constexpr std::string_view legacy_json_null = "{}"; + expected_encoder.add_json({legacy_json_null.data(), legacy_json_null.size()}); + VariantBatchBuilder expected = expected_encoder.finish_batch(); + const auto& values = assert_cast(*output->field(0)); + const auto& metadatas = assert_cast(*output->field(1)); + for (size_t row = 0; row < 2; ++row) { + const VariantRef expected_value = expected.value_at(row); + EXPECT_EQ(values.GetView(cast_set(row)), + std::string_view(expected_value.value.data, expected_value.value.size)); + EXPECT_EQ(metadatas.GetView(cast_set(row)), + std::string_view(expected_value.metadata.data, expected_value.metadata.size)); + } +} + +TEST(VariantSerdeTest, StructArrowPreservesScalarStringSemantics) { + DataTypeVariantSerDe serde; + auto column = ColumnVariant::create(0, false); + DataTypeSerDe::FormatOptions options; + const std::string json = R"("123")"; + Slice slice(json.data(), json.size()); + ASSERT_TRUE(serde.deserialize_one_cell_from_json(*column, slice, options).ok()); + column->finalize(ColumnVariant::FinalizeMode::WRITE_MODE); + + auto value_builder = std::make_shared(); + auto metadata_builder = std::make_shared(); + auto arrow_type = arrow::struct_({arrow::field("value", arrow::binary(), false), + arrow::field("metadata", arrow::binary(), false)}); + arrow::StructBuilder builder(arrow_type, arrow::default_memory_pool(), + {value_builder, metadata_builder}); + ASSERT_TRUE(serde.write_column_to_arrow(*column, nullptr, &builder, 0, column->size(), + cctz::utc_time_zone()) + .ok()); + + std::shared_ptr output; + ASSERT_TRUE(builder.Finish(&output).ok()); + JsonStringToVariantEncoder expected_encoder({.max_json_key_length = 1024, + .throw_on_invalid_json = true, + .check_duplicate_json_path = false}); + expected_encoder.add_json({json.data(), json.size()}); + VariantBatchBuilder expected = expected_encoder.finish_batch(); + const VariantRef expected_value = expected.value_at(0); + const auto& values = assert_cast(*output->field(0)); + const auto& metadatas = assert_cast(*output->field(1)); + EXPECT_EQ(values.GetView(0), + std::string_view(expected_value.value.data, expected_value.value.size)); + EXPECT_EQ(metadatas.GetView(0), + std::string_view(expected_value.metadata.data, expected_value.metadata.size)); +} + +TEST(VariantSerdeTest, StructArrowPreservesMixedObjectAndScalarStringSemantics) { + DataTypeVariantSerDe serde; + auto column = ColumnVariant::create(0, false); + DataTypeSerDe::FormatOptions options; + const std::vector json_rows {R"({"id":7})", R"("123")"}; + for (const auto& json : json_rows) { + Slice slice(json.data(), json.size()); + ASSERT_TRUE(serde.deserialize_one_cell_from_json(*column, slice, options).ok()); + } + column->finalize(ColumnVariant::FinalizeMode::WRITE_MODE); + + auto value_builder = std::make_shared(); + auto metadata_builder = std::make_shared(); + auto arrow_type = arrow::struct_({arrow::field("value", arrow::binary(), false), + arrow::field("metadata", arrow::binary(), false)}); + arrow::StructBuilder builder(arrow_type, arrow::default_memory_pool(), + {value_builder, metadata_builder}); + ASSERT_TRUE(serde.write_column_to_arrow(*column, nullptr, &builder, 0, column->size(), + cctz::utc_time_zone()) + .ok()); + + std::shared_ptr output; + ASSERT_TRUE(builder.Finish(&output).ok()); + JsonStringToVariantEncoder expected_encoder({.max_json_key_length = 1024, + .throw_on_invalid_json = true, + .check_duplicate_json_path = false}); + for (const auto& json : json_rows) { + expected_encoder.add_json({json.data(), json.size()}); + } + VariantBatchBuilder expected = expected_encoder.finish_batch(); + const auto& values = assert_cast(*output->field(0)); + const auto& metadatas = assert_cast(*output->field(1)); + for (size_t row = 0; row < json_rows.size(); ++row) { + const VariantRef expected_value = expected.value_at(row); + EXPECT_EQ(values.GetView(cast_set(row)), + std::string_view(expected_value.value.data, expected_value.value.size)); + EXPECT_EQ(metadatas.GetView(cast_set(row)), + std::string_view(expected_value.metadata.data, expected_value.metadata.size)); + } +} + } // namespace doris diff --git a/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp b/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp index 9f5e6054a1fba7..da30c0a040c1fc 100644 --- a/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp +++ b/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp @@ -15,7 +15,9 @@ // specific language governing permissions and limitations // under the License. +#include #include +#include #include #include @@ -251,6 +253,34 @@ TEST(DataTypeVariantV2SerdeOutputTest, DormantDirectClassExists) { EXPECT_EQ(serde.get_name(), "Variant"); } +TEST(DataTypeVariantV2SerdeOutputTest, VariantStructArrowPreservesBinaryEncoding) { + DataTypeVariantV2SerDe serde; + auto documents = encoded_json({R"({"id":7,"tags":["doris"]})", R"([1,true,null])"}); + auto value_builder = std::make_shared(); + auto metadata_builder = std::make_shared(); + auto arrow_type = arrow::struct_({arrow::field("value", arrow::binary(), false), + arrow::field("metadata", arrow::binary(), false)}); + arrow::StructBuilder builder(arrow_type, arrow::default_memory_pool(), + {value_builder, metadata_builder}); + NullMap forced_nulls {0, 1}; + + ASSERT_TRUE(serde.write_column_to_arrow(*documents, &forced_nulls, &builder, 0, + documents->size(), cctz::utc_time_zone()) + .ok()); + std::shared_ptr output; + ASSERT_TRUE(builder.Finish(&output).ok()); + ASSERT_EQ(output->length(), 2); + EXPECT_FALSE(output->IsNull(0)); + EXPECT_TRUE(output->IsNull(1)); + + const auto& values = assert_cast(*output->field(0)); + const auto& metadatas = assert_cast(*output->field(1)); + const VariantRef expected = documents->get_value_ref(0); + EXPECT_EQ(values.GetView(0), std::string_view(expected.value.data, expected.value.size)); + EXPECT_EQ(metadatas.GetView(0), + std::string_view(expected.metadata.data, expected.metadata.size)); +} + TEST(DataTypeVariantV2SerdeOutputTest, SqlScalarsFollowLegacyOutputAndDataFormatsUseJson) { DataTypeVariantV2SerDe serde; auto strings = typed_strings({std::string_view("a\"\n"), std::string_view(""), std::nullopt, diff --git a/be/test/format/table/paimon_jni_reader_test.cpp b/be/test/format/table/paimon_jni_reader_test.cpp index 969713429c29d6..506bb46dd6e2f6 100644 --- a/be/test/format/table/paimon_jni_reader_test.cpp +++ b/be/test/format/table/paimon_jni_reader_test.cpp @@ -19,10 +19,25 @@ #include +#include +#include #include +#include #include #include +#include "core/assert_cast.h" +#include "core/column/column_nullable.h" +#include "core/column/column_variant.h" +#include "core/column/variant_v2/column_variant_v2.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_variant.h" +#include "core/data_type/data_type_variant_v2.h" +#include "core/data_type_serde/data_type_serde.h" +#include "core/value/variant/variant_batch_builder.h" +#include "exec/common/variant_util.h" +#include "exprs/function/parse/variant_string_parse.h" +#include "format/jni/jni_data_bridge.h" #include "gen_cpp/PlanNodes_types.h" #include "runtime/runtime_state.h" @@ -40,6 +55,13 @@ TFileRangeDesc make_legacy_paimon_jni_range() { return range; } +struct JavaVarbinaryEntry { + int64_t length; + uint64_t address; +}; + +static_assert(sizeof(JavaVarbinaryEntry) == 16); + TEST(LegacyPaimonJniReaderTest, GeneratesMissingOrEmptySerializedTableCacheKey) { const auto range = make_legacy_paimon_jni_range(); TFileScanRangeParams scan_params; @@ -61,5 +83,174 @@ TEST(LegacyPaimonJniReaderTest, GeneratesMissingOrEmptySerializedTableCacheKey) EXPECT_NE(missing_key, empty_key); } +TEST(LegacyPaimonJniReaderTest, PublishesVariantV2BinaryStructSchema) { + const DataTypePtr type = std::make_shared(); + EXPECT_EQ(JniDataBridge::get_jni_type_with_different_string(type), + "struct"); + EXPECT_EQ(JniDataBridge::get_jni_type_with_encoded_struct_fields(type), + "struct<$dmFsdWU=:varbinary,$bWV0YWRhdGE=:varbinary>"); +} + +TEST(LegacyPaimonJniReaderTest, DecodesVariantV2BinaryStructFromJavaMetadata) { + const std::string json = R"({"id":7,"tags":["doris"]})"; + JsonStringToVariantEncoder encoder; + encoder.add_json({json.data(), json.size()}); + VariantBatchBuilder source = encoder.finish_batch(); + const VariantRef expected = source.value_at(0); + + std::array outer_nulls {0, 1}; + std::array child_nulls {0, 1}; + std::array value_entries { + JavaVarbinaryEntry {.length = static_cast(expected.value.size), + .address = reinterpret_cast(expected.value.data)}, + JavaVarbinaryEntry {.length = 0, .address = 0}}; + std::array metadata_entries { + JavaVarbinaryEntry {.length = static_cast(expected.metadata.size), + .address = reinterpret_cast(expected.metadata.data)}, + JavaVarbinaryEntry {.length = 0, .address = 0}}; + std::array metadata {reinterpret_cast(outer_nulls.data()), + reinterpret_cast(child_nulls.data()), + reinterpret_cast(value_entries.data()), + reinterpret_cast(child_nulls.data()), + reinterpret_cast(metadata_entries.data())}; + + const DataTypePtr type = make_nullable(std::make_shared()); + ColumnPtr result = type->create_column(); + JniDataBridge::TableMetaAddress address(reinterpret_cast(metadata.data())); + ASSERT_TRUE(JniDataBridge::fill_column(address, result, type, 2).ok()); + + const auto& nullable = assert_cast(*result); + EXPECT_EQ(nullable.get_null_map_data(), (NullMap {0, 1})); + const auto& variants = assert_cast(nullable.get_nested_column()); + ASSERT_EQ(variants.size(), 2); + const VariantRef actual = variants.get_value_ref(0); + EXPECT_EQ(actual.value, expected.value); + EXPECT_EQ(std::string_view(actual.metadata.data, actual.metadata.size), + std::string_view(expected.metadata.data, expected.metadata.size)); + EXPECT_TRUE(variants.get_value_ref(1).is_null()); +} + +TEST(LegacyPaimonJniReaderTest, // NOLINT(readability-function-cognitive-complexity) + DecodesLegacyVariantBinaryStructFromJavaMetadata) { + const std::vector json_rows { + R"({"name":"alpha","score":12.5,"tags":["dts","fluss","paimon"]})", + R"({"active":true,"name":"beta","nested":{"version":"2.0"}})", "null", "null", "123"}; + JsonStringToVariantEncoder encoder; + for (const auto& json : json_rows) { + encoder.add_json({json.data(), json.size()}); + } + VariantBatchBuilder source = encoder.finish_batch(); + + std::array outer_nulls {0, 0, 0, 1, 0}; + std::array child_nulls {0, 0, 0, 1, 0}; + std::array value_entries; + std::array metadata_entries; + for (size_t row = 0; row < json_rows.size(); ++row) { + const VariantRef value = source.value_at(row); + value_entries[row] = {.length = static_cast(value.value.size), + .address = reinterpret_cast(value.value.data)}; + metadata_entries[row] = {.length = static_cast(value.metadata.size), + .address = reinterpret_cast(value.metadata.data)}; + } + value_entries[3] = {.length = 0, .address = 0}; + metadata_entries[3] = {.length = 0, .address = 0}; + std::array metadata {reinterpret_cast(outer_nulls.data()), + reinterpret_cast(child_nulls.data()), + reinterpret_cast(value_entries.data()), + reinterpret_cast(child_nulls.data()), + reinterpret_cast(metadata_entries.data())}; + + const DataTypePtr type = make_nullable(std::make_shared(0, false)); + ColumnPtr result = type->create_column(); + JniDataBridge::TableMetaAddress address(reinterpret_cast(metadata.data())); + ASSERT_TRUE(JniDataBridge::fill_column(address, result, type, 5).ok()); + + const auto& nullable = assert_cast(*result); + EXPECT_EQ(nullable.get_null_map_data(), (NullMap {0, 0, 0, 1, 0})); + const auto& variants = assert_cast(nullable.get_nested_column()); + ASSERT_EQ(variants.rows(), 5); + auto finalized = variants.clone_finalized(); + const auto& decoded = assert_cast(*finalized); + DataTypeSerDe::FormatOptions options; + std::string object_json; + decoded.serialize_one_row_to_string(0, &object_json, options); + const std::string expected_object_json = + R"({"name":"alpha","score":12.5,"tags":["dts", "fluss", "paimon"]})"; + EXPECT_EQ(object_json, expected_object_json); + std::string nested_object_json; + decoded.serialize_one_row_to_string(1, &nested_object_json, options); + EXPECT_EQ(nested_object_json, R"({"active":1,"name":"beta","nested":{"version":"2.0"}})"); + std::string encoded_json_null; + decoded.serialize_one_row_to_string(2, &encoded_json_null, options); + // Legacy VARIANT V1 materializes a root JSON null as an empty object. + EXPECT_EQ(encoded_json_null, "{}"); + std::string scalar_number; + decoded.serialize_one_row_to_string(4, &scalar_number, options); + EXPECT_EQ(scalar_number, "123"); + + IColumn::Permutation permutation {0, 1, 2, 3, 4}; + auto permuted_base = nullable.permute(permutation, 0); + const auto& permuted_nullable = assert_cast(*permuted_base); + const auto& permuted_variants = + assert_cast(permuted_nullable.get_nested_column()); + auto finalized_permutation = permuted_variants.clone_finalized(); + const auto& permuted = assert_cast(*finalized_permutation); + std::string first_permuted; + permuted.serialize_one_row_to_string(0, &first_permuted, options); + EXPECT_EQ(first_permuted, expected_object_json); + std::string second_permuted; + permuted.serialize_one_row_to_string(1, &second_permuted, options); + EXPECT_EQ(second_permuted, R"({"active":1,"name":"beta","nested":{"version":"2.0"}})"); + std::string encoded_null_permuted; + permuted.serialize_one_row_to_string(2, &encoded_null_permuted, options); + EXPECT_EQ(encoded_null_permuted, "{}"); + std::string fifth_permuted; + permuted.serialize_one_row_to_string(4, &fifth_permuted, options); + EXPECT_EQ(fifth_permuted, "123"); + + auto copied_rows = type->create_column(); + for (size_t row = 0; row < result->size(); ++row) { + copied_rows->insert_from(*result, row); + } + const auto& copied_nullable = assert_cast(*copied_rows); + EXPECT_EQ(copied_nullable.get_null_map_data(), (NullMap {0, 0, 0, 1, 0})); + const auto& copied_variants = + assert_cast(copied_nullable.get_nested_column()); + auto finalized_copy = copied_variants.clone_finalized(); + const auto& copied = assert_cast(*finalized_copy); + std::string copied_nested_object; + copied.serialize_one_row_to_string(1, &copied_nested_object, options); + EXPECT_EQ(copied_nested_object, R"({"active":1,"name":"beta","nested":{"version":"2.0"}})"); + + auto merged = ColumnVariant::create(0, false); + ParseConfig parse_config; + for (size_t row = 0; row < json_rows.size(); ++row) { + auto source_column = ColumnVariant::create(0, false); + if (outer_nulls[row]) { + source_column->insert_default(); + } else { + variant_util::parse_json_to_variant(*source_column, + {json_rows[row].data(), json_rows[row].size()}, + nullptr, parse_config); + } + merged->insert_range_from(*source_column, 0, 1); + } + merged->finalize(); + auto merged_permuted = merged->permute(permutation, 0); + const auto& merged_result = assert_cast(*merged_permuted); + std::string merged_second_row; + merged_result.serialize_one_row_to_string(1, &merged_second_row, options); + EXPECT_EQ(merged_second_row, R"({"active":1,"name":"beta","nested":{"version":"2.0"}})"); + + auto merged_batch = ColumnVariant::create(0, false); + merged_batch->insert_range_from(variants, 0, variants.rows()); + merged_batch->finalize(); + auto merged_batch_permuted = merged_batch->permute(permutation, 0); + const auto& merged_batch_result = assert_cast(*merged_batch_permuted); + std::string merged_batch_second_row; + merged_batch_result.serialize_one_row_to_string(1, &merged_batch_second_row, options); + EXPECT_EQ(merged_batch_second_row, R"({"active":1,"name":"beta","nested":{"version":"2.0"}})"); +} + } // namespace } // namespace doris diff --git a/be/test/runtime/runtime_state_block_budget_test.cpp b/be/test/runtime/runtime_state_block_budget_test.cpp index 5a384378ec382b..cb55639b6cd120 100644 --- a/be/test/runtime/runtime_state_block_budget_test.cpp +++ b/be/test/runtime/runtime_state_block_budget_test.cpp @@ -60,6 +60,27 @@ TEST(RuntimeStateIcebergCommitDataTest, SharesTheReportBudgetAcrossParallelTasks EXPECT_FALSE(second_status.ok()); } +TEST(RuntimeStateIcebergCommitDataTest, SharesTheReportBudgetWithPaimon) { + RuntimeState iceberg_state; + RuntimeState paimon_state; + auto budget = std::make_shared(); + iceberg_state.set_external_file_report_state(budget); + paimon_state.set_external_file_report_state(budget); + const int32_t saved_limit = config::thrift_max_message_size; + config::thrift_max_message_size = 1024 * 1024 + 512; + TIcebergCommitData iceberg_data; + iceberg_data.__set_file_path(std::string(300, 'x')); + TPaimonCommitMessage paimon_data; + paimon_data.__set_payload(std::string(300, 'x')); + + Status iceberg_status = iceberg_state.add_iceberg_commit_datas(iceberg_data); + Status paimon_status = paimon_state.add_paimon_commit_messages({std::move(paimon_data)}); + + config::thrift_max_message_size = saved_limit; + EXPECT_TRUE(iceberg_status.ok()) << iceberg_status; + EXPECT_FALSE(paimon_status.ok()); +} + TEST(RuntimeStateIcebergCommitDataTest, UsesTheSmallerCoordinatorThriftLimit) { RuntimeState state; const int32_t saved_limit = config::thrift_max_message_size; @@ -91,6 +112,9 @@ TEST(RuntimeStateIcebergCommitDataTest, PeriodicReportOmitsExternalCommitData) { ASSERT_TRUE(state.add_iceberg_commit_datas(iceberg_data).ok()); TMCCommitData mc_data; state.add_mc_commit_datas(mc_data); + TPaimonCommitMessage paimon_data; + paimon_data.__set_payload("paimon-commit"); + ASSERT_TRUE(state.add_paimon_commit_messages({std::move(paimon_data)}).ok()); TReportExecStatusParams periodic_params; state.append_external_file_commit_data(&periodic_params, false); @@ -98,12 +122,14 @@ TEST(RuntimeStateIcebergCommitDataTest, PeriodicReportOmitsExternalCommitData) { EXPECT_FALSE(periodic_params.__isset.hive_partition_updates); EXPECT_FALSE(periodic_params.__isset.iceberg_commit_datas); EXPECT_FALSE(periodic_params.__isset.mc_commit_datas); + EXPECT_FALSE(periodic_params.__isset.paimon_commit_messages); TReportExecStatusParams final_params; state.append_external_file_commit_data(&final_params, true); EXPECT_TRUE(final_params.__isset.hive_partition_updates); EXPECT_TRUE(final_params.__isset.iceberg_commit_datas); EXPECT_TRUE(final_params.__isset.mc_commit_datas); + EXPECT_TRUE(final_params.__isset.paimon_commit_messages); } TEST(RuntimeStateIcebergCommitDataTest, RetainsFileCleanupUntilReportAcknowledgement) { diff --git a/fe/be-java-extensions/paimon-scanner/pom.xml b/fe/be-java-extensions/paimon-scanner/pom.xml index fa7c27e4e98319..f492c85035d2d3 100644 --- a/fe/be-java-extensions/paimon-scanner/pom.xml +++ b/fe/be-java-extensions/paimon-scanner/pom.xml @@ -61,6 +61,16 @@ under the License. paimon-format + + org.apache.paimon + paimon-vortex-format + + + + org.apache.arrow + arrow-vector + + true + + + * + * + org.apache.httpcomponents.client5 httpclient5 @@ -306,6 +314,16 @@ under the License. true ${project.basedir}/target/dependency-reduced-pom.xml + + org.apache.paimon:paimon-hive-connector-3.1 + + + org/apache/paimon/hive/** + + *:* diff --git a/fe/fe-connector/fe-connector-paimon-hive-shade/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/fe/fe-connector/fe-connector-paimon-hive-shade/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory new file mode 100644 index 00000000000000..3df3dcc0842588 --- /dev/null +++ b/fe/fe-connector/fe-connector-paimon-hive-shade/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory @@ -0,0 +1,2 @@ +org.apache.paimon.hive.HiveCatalogFactory +org.apache.paimon.hive.HiveCatalogLockFactory diff --git a/fe/fe-connector/fe-connector-paimon/pom.xml b/fe/fe-connector/fe-connector-paimon/pom.xml index 9b17c987ba3f1d..857de4161da3c1 100644 --- a/fe/fe-connector/fe-connector-paimon/pom.xml +++ b/fe/fe-connector/fe-connector-paimon/pom.xml @@ -116,6 +116,22 @@ under the License. org.apache.paimon paimon-core ${paimon.version} + + + + org.apache.paimon + paimon-shade-caffeine-2 + + + org.apache.paimon + paimon-shade-guava-30 + + + org.apache.paimon + paimon-shade-jackson-2 + + + org.apache.paimon paimon-format - test + + + + + org.apache.paimon + paimon-vortex-format - 1.3.1 + 2.0.0 3.4.4 17.0.0 @@ -1515,6 +1515,11 @@ under the License. paimon-format ${paimon.version} + + org.apache.paimon + paimon-vortex-format + ${paimon.version} + org.apache.paimon paimon-s3 diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift index d2cb4d534bb49b..97764d92193a84 100644 --- a/gensrc/thrift/DataSinks.thrift +++ b/gensrc/thrift/DataSinks.thrift @@ -46,6 +46,7 @@ enum TDataSinkType { MAXCOMPUTE_TABLE_SINK = 18, ICEBERG_DELETE_SINK = 19, ICEBERG_MERGE_SINK = 20, + PAIMON_TABLE_SINK = 21, } enum TResultSinkType { @@ -625,6 +626,25 @@ struct TMaxComputeTableSink { 18: optional i64 txn_id // FE external transaction ID for runtime block_id allocation } +enum TPaimonWriteMode { + APPEND = 0, + OVERWRITE = 1, +} + +struct TPaimonCommitMessage { + 1: optional binary payload // Paimon CommitMessageSerializer bytes (DPCM-framed) + 2: optional i64 row_count // set once per BE writer, on its first payload +} + +struct TPaimonTableSink { + 1: optional string serialized_table // serialized Paimon FileStoreTable object (base64) + 2: optional map hadoop_config + 3: optional list column_names + 4: optional TPaimonWriteMode write_mode + 5: optional i64 transaction_id + 6: optional string commit_user +} + struct TDataSink { 1: required TDataSinkType type 2: optional TDataStreamSink stream_sink @@ -645,4 +665,5 @@ struct TDataSink { 18: optional TMaxComputeTableSink max_compute_table_sink 19: optional TIcebergDeleteSink iceberg_delete_sink 20: optional TIcebergMergeSink iceberg_merge_sink + 21: optional TPaimonTableSink paimon_table_sink } diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift index 38452f0b1d78a0..106ae649ab879f 100644 --- a/gensrc/thrift/FrontendService.thrift +++ b/gensrc/thrift/FrontendService.thrift @@ -338,6 +338,8 @@ struct TReportExecStatusParams { 32: optional list mc_commit_datas 33: optional string first_error_msg + + 34: optional list paimon_commit_messages } struct TFeResult { diff --git a/regression-test/data/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.out b/regression-test/data/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.out new file mode 100644 index 00000000000000..baa7292a4f8f80 --- /dev/null +++ b/regression-test/data/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.out @@ -0,0 +1,9 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !ctas_rows -- +1 candidate + +-- !after_if_not_exists -- +1 candidate + +-- !after_existing_target_error -- +1 candidate diff --git a/regression-test/data/external_table_p0/paimon/test_paimon_write_boundary.out b/regression-test/data/external_table_p0/paimon/test_paimon_write_boundary.out index f1118d0bd7069e..067287d5d4556b 100644 --- a/regression-test/data/external_table_p0/paimon/test_paimon_write_boundary.out +++ b/regression-test/data/external_table_p0/paimon/test_paimon_write_boundary.out @@ -6,10 +6,36 @@ -- !before_snapshots -- 1 --- !after_rows -- +-- !after_insert_values -- 1 10 base-1 2 20 base-2 +3 30 insert-values + +-- !after_insert_select -- +1 10 base-1 +2 20 base-2 +3 30 insert-values +4 40 insert-select + +-- !after_overwrite -- +5 50 overwrite + +-- !after_rows -- +5 50 overwrite -- !after_snapshots -- -1 +4 + +-- !variant_rows -- +1 {"name":"alpha","score":12.5,"tags":["dts", "paimon"]} +2 {"active":1,"nested":{"version":"2.0"}} +3 \N +4 123 +5 {} +-- !variant_row_tracking -- +1 {"name":"alpha","score":12.5,"tags":["dts", "paimon"]} 0 1 +2 {"active":1,"nested":{"version":"2.0"}} 1 2 +3 \N 2 2 +4 123 3 2 +5 {} 4 2 diff --git a/regression-test/suites/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.groovy b/regression-test/suites/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.groovy index 034bb9581a322f..e359b298b3b404 100644 --- a/regression-test/suites/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.groovy +++ b/regression-test/suites/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.groovy @@ -50,32 +50,22 @@ suite("test_paimon_ctas_atomicity_negative", sql """switch ${catalogName}""" sql """use ${dbName}""" - // A failed CTAS must not leave metadata that makes a retry fail with TABLE ALREADY EXISTS. - // On the connector-SPI path the sink rejection is worded by the connector's declared write - // capabilities (the paimon connector declares none) rather than by the legacy fe-core - // "Load data to PaimonExternalCatalog is not supported"; the CTAS still fails at the same point. - test { - sql """ - create table ctas_target engine=paimon - as select cast(1 as int) as id, cast('candidate' as string) as payload - """ - exception "does not support INSERT operations" - } - assertEquals(0, (sql """show tables like 'ctas_target'""").size()) - - spark_paimon """ - create table paimon.${dbName}.ctas_target (id int, payload string) - using paimon + sql """ + create table ctas_target engine=paimon + as select cast(1 as int) as id, cast('candidate' as string) as payload """ - // IF NOT EXISTS must remain a no-op even though Paimon does not support the CTAS sink. + assertEquals(1, (sql """show tables like 'ctas_target'""").size()) + order_qt_ctas_rows """select id, payload from ctas_target""" + + // IF NOT EXISTS remains a no-op and must not append the SELECT result. sql """ create table if not exists ctas_target engine=paimon - as select cast(1 as int) as id, cast('candidate' as string) as payload + as select cast(2 as int) as id, cast('ignored' as string) as payload """ assertEquals(1, (sql """show tables like 'ctas_target'""").size()) - assertEquals(0, (sql """select * from ctas_target""").size()) + order_qt_after_if_not_exists """select id, payload from ctas_target""" - // An existing non-idempotent target must keep catalog error precedence; no sink can own it. + // An existing non-idempotent target keeps catalog error precedence and existing data. test { sql """ create table ctas_target engine=paimon @@ -83,7 +73,7 @@ suite("test_paimon_ctas_atomicity_negative", """ exception "already exists" } - assertEquals(0, (sql """select * from ctas_target""").size()) + order_qt_after_existing_target_error """select id, payload from ctas_target""" } finally { spark_paimon """drop table if exists paimon.${dbName}.ctas_target""" sql """drop catalog if exists ${catalogName}""" diff --git a/regression-test/suites/external_table_p0/paimon/test_paimon_write_boundary.groovy b/regression-test/suites/external_table_p0/paimon/test_paimon_write_boundary.groovy index 84fd84ce342699..8f445ba2e862ba 100644 --- a/regression-test/suites/external_table_p0/paimon/test_paimon_write_boundary.groovy +++ b/regression-test/suites/external_table_p0/paimon/test_paimon_write_boundary.groovy @@ -41,51 +41,41 @@ suite("test_paimon_write_boundary", """ try { - spark_paimon_multi """ - create database if not exists paimon.${dbName}; - drop table if exists paimon.${dbName}.write_boundary; - create table paimon.${dbName}.write_boundary ( - id int, + sql """switch ${catalogName}""" + sql """create database if not exists ${dbName}""" + sql """use ${dbName}""" + sql """drop table if exists write_boundary""" + sql """ + create table write_boundary ( + id int not null, score int, note string - ) using paimon tblproperties ( + ) engine=paimon properties ( 'primary-key'='id', 'bucket'='1', 'file.format'='parquet' - ); - insert into paimon.${dbName}.write_boundary values + ) + """ + sql """ + insert into write_boundary values (1, 10, 'base-1'), - (2, 20, 'base-2'); + (2, 20, 'base-2') """ - sql """switch ${catalogName}""" - sql """use ${dbName}""" - qt_before_rows """select id, score, note from write_boundary order by id""" qt_before_snapshots """select count(*) from write_boundary\$snapshots""" - // WB01-WB06 preserve the documented data-write boundary at analysis time. The source table - // and its snapshot list must stay unchanged after every rejected write shape. - // - // The INSERT-family rejections are worded by the connector-SPI path, not by the legacy fe-core - // one: a paimon catalog is a PluginDrivenExternalCatalog, so UnboundTableSinkCreator builds an - // UnboundConnectorTableSink instead of throwing "Load data to PaimonExternalCatalog is not - // supported", and the rejection lands on the connector's declared write capabilities (the paimon - // connector declares none). The boundary asserted here is identical -- every write shape is still - // rejected at analysis time and the table is untouched -- only the message differs. - test { - sql """insert into write_boundary values (3, 30, 'insert-values')""" - exception "does not support INSERT operations" - } - test { - sql """insert into write_boundary select 3, 30, 'insert-select'""" - exception "does not support INSERT operations" - } - test { - // INSERT OVERWRITE is gated earlier, by InsertOverwriteTableCommand's allowInsertOverwrite. - sql """insert overwrite table write_boundary values (3, 30, 'overwrite')""" - exception "insert into overwrite only support" - } + sql """insert into write_boundary values (3, 30, 'insert-values')""" + order_qt_after_insert_values """select id, score, note from write_boundary""" + + sql """insert into write_boundary select 4, 40, 'insert-select'""" + order_qt_after_insert_select """select id, score, note from write_boundary""" + + sql """insert overwrite table write_boundary values (5, 50, 'overwrite')""" + order_qt_after_overwrite """select id, score, note from write_boundary""" + + // Row-level mutation remains an OLAP-table-only command. Paimon upserts are performed + // through INSERT statements against primary-key tables. test { sql """update write_boundary set score = score + 1 where id = 1""" exception "target table in update command should be an olapTable" @@ -109,6 +99,59 @@ suite("test_paimon_write_boundary", sql """refresh table write_boundary""" qt_after_rows """select id, score, note from write_boundary order by id""" qt_after_snapshots """select count(*) from write_boundary\$snapshots""" + + sql """drop table if exists variant_row_tracking""" + sql """ + create table variant_row_tracking ( + id int, + doc variant + ) engine=paimon properties ( + 'bucket'='-1', + 'file.format'='parquet', + 'row-tracking.enabled'='true', + 'data-evolution.enabled'='true' + ) + """ + sql """ + insert into variant_row_tracking values + (1, parse_to_variant('{"name":"alpha","score":12.5,"tags":["dts","paimon"]}')) + """ + // Legacy VARIANT V1 materializes a root JSON null as an empty object. + sql """ + insert into variant_row_tracking values + (2, parse_to_variant('{"active":true,"nested":{"version":"2.0"}}')), + (3, null), + (4, parse_to_variant('"123"')), + (5, parse_to_variant('null')) + """ + + sql """set force_jni_scanner=false""" + sql """set enable_paimon_cpp_reader=true""" + String variantExplain = sql(""" + explain verbose select id, doc from variant_row_tracking + """).collect { row -> row[0].toString() }.join("\n") + def variantSplits = (variantExplain =~ /paimonNativeReadSplits=(\d+)\/(\d+)/) + assertTrue(variantSplits.find(), "Expected Paimon split counts for VARIANT projection") + assertTrue(Long.parseLong(variantSplits.group(2)) > 0 + && Long.parseLong(variantSplits.group(1)) == 0, + "VARIANT projection must use JNI-only splits: ${variantExplain}") + + String scalarExplain = sql(""" + explain verbose select id from variant_row_tracking + """).collect { row -> row[0].toString() }.join("\n") + def scalarSplits = (scalarExplain =~ /paimonNativeReadSplits=(\d+)\/(\d+)/) + assertTrue(scalarSplits.find(), "Expected Paimon split counts for scalar projection") + assertTrue(Long.parseLong(scalarSplits.group(2)) > 0 + && scalarSplits.group(1) == scalarSplits.group(2), + "Scalar-only projection must retain native splits: ${scalarExplain}") + + order_qt_variant_rows """select id, doc from variant_row_tracking order by id""" + order_qt_variant_row_tracking """ + select id, doc, _ROW_ID, _SEQUENCE_NUMBER + from variant_row_tracking\$row_tracking + order by _SEQUENCE_NUMBER, _ROW_ID + """ + } finally { sql """drop catalog if exists ${catalogName}""" }