Gabriel39 commented on code in PR #66321: URL: https://github.com/apache/doris/pull/66321#discussion_r3734596432
########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonVariantWriteAnalyzer.java: ########## @@ -0,0 +1,155 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.datasource.paimon; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.Type; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.exceptions.UnboundException; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.types.ArrayType; +import org.apache.doris.nereids.types.DataType; +import org.apache.doris.nereids.types.MapType; +import org.apache.doris.nereids.types.StructField; +import org.apache.doris.nereids.types.StructType; +import org.apache.doris.nereids.types.VariantType; + +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** Analysis checks for the V2-only Paimon Variant write protocol. */ +public final class PaimonVariantWriteAnalyzer { + private PaimonVariantWriteAnalyzer() { + } + + /** + * Rejects a disabled V2 protocol and unsupported Variant V2 conversions before sink coercion. + */ + public static void validate( + PaimonWriteTarget writeTarget, + List<Column> writeColumns, + Map<String, NamedExpression> columnToOutput, + boolean enableVariantV2) throws AnalysisException { + for (Column column : writeColumns) { + Type targetCatalogType = writeTarget.getColumnTypes().get(column.getName()); + if (targetCatalogType == null) { + continue; + } + DataType targetType = DataType.fromCatalogType(targetCatalogType); + if (!VariantType.containsVariant(targetType)) { Review Comment: [P1] Reject Variant writes for unsupported Paimon file formats This validation enables the V2 protocol for every `FileStoreTable` regardless of `file.format`. In Paimon 1.4.2, the ORC `FieldWriterFactory.visit(VariantType)` throws `UnsupportedOperationException`; a nested Variant reaches the same visitor recursively. As a result, an INSERT into an ORC-backed Paimon table passes Doris analysis and Arrow conversion, then deterministically fails only when the Java writer is created. Please either reject non-Parquet Variant targets from the pinned table options during analysis (with top-level and nested ORC regressions), or implement the missing format support and document the supported formats. ########## be/src/core/data_type_serde/data_type_variant_v2_serde.cpp: ########## @@ -175,6 +176,133 @@ void preflight_json(const IColumn& column, size_t start, size_t end, }); } +void validate_paimon_variant_primitive(VariantPrimitiveId primitive_id) { + switch (primitive_id) { + case VariantPrimitiveId::NULL_VALUE: + case VariantPrimitiveId::TRUE_VALUE: + case VariantPrimitiveId::FALSE_VALUE: + case VariantPrimitiveId::INT8: + case VariantPrimitiveId::INT16: + case VariantPrimitiveId::INT32: + case VariantPrimitiveId::INT64: + case VariantPrimitiveId::DOUBLE: + case VariantPrimitiveId::DECIMAL4: + case VariantPrimitiveId::DECIMAL8: + case VariantPrimitiveId::DECIMAL16: + case VariantPrimitiveId::DATE: + case VariantPrimitiveId::TIMESTAMP_MICROS: + case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS: + case VariantPrimitiveId::FLOAT: + case VariantPrimitiveId::BINARY: + case VariantPrimitiveId::STRING: + case VariantPrimitiveId::UUID: + return; + case VariantPrimitiveId::TIME_NTZ_MICROS: + case VariantPrimitiveId::TIMESTAMP_NANOS: + case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS: + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Paimon does not support Variant primitive id {}", + static_cast<uint8_t>(primitive_id)); + } + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Paimon does not support unknown Variant primitive id {}", + static_cast<uint8_t>(primitive_id)); +} + +void validate_paimon_variant_value(VariantRef value, uint32_t depth = 0) { + if (depth > VARIANT_MAX_NESTING_DEPTH) { + throw Exception(ErrorCode::CORRUPTION, "Variant value exceeds maximum nesting depth {}", + VARIANT_MAX_NESTING_DEPTH); + } + const size_t encoded_size = value.value_size(); + if (encoded_size != value.value.size) { + throw Exception(ErrorCode::CORRUPTION, + "Variant value has {} trailing bytes after the encoded value", + value.value.size - encoded_size); + } + + switch (value.basic_type()) { + case VariantBasicType::PRIMITIVE: + validate_paimon_variant_primitive(value.primitive_id()); + return; + case VariantBasicType::SHORT_STRING: + return; + case VariantBasicType::OBJECT: + for (uint32_t i = 0; i < value.num_elements(); ++i) { + uint32_t field_id = 0; + VariantRef child = value.object_value_at(i, &field_id); + value.metadata.key_at(field_id); + validate_paimon_variant_value(child, depth + 1); + } + return; + case VariantBasicType::ARRAY: + for (uint32_t i = 0; i < value.num_elements(); ++i) { + validate_paimon_variant_value(value.array_at(i), depth + 1); + } + return; + } +} + +void require_variant_arrow_status(const arrow::Status& status) { + if (!status.ok()) { + throw Exception(ErrorCode::INTERNAL_ERROR, "Variant V2 Arrow append failed: {}", + status.ToString()); + } +} + +Status write_binary_variant_arrow(const IColumn& column, const NullMap* null_map, + arrow::StructBuilder& builder, size_t start, size_t end) { + // StructBuilder::type() returns a shared_ptr by value. Keep that owner alive while using the + // cast reference; otherwise the reference would dangle as soon as the temporary is destroyed. + const auto builder_type = builder.type(); + const auto& struct_type = assert_cast<const arrow::StructType&>(*builder_type); + if (struct_type.num_fields() != 2 || struct_type.field(0)->name() != "value" || + struct_type.field(1)->name() != "metadata" || + struct_type.field(0)->type()->id() != arrow::Type::BINARY || + struct_type.field(1)->type()->id() != arrow::Type::BINARY) { + return Status::InvalidArgument( + "Binary Variant V2 Arrow type must be " + "struct<value: binary, metadata: binary>, got {}", + struct_type.ToString()); + } + auto* value_builder = dynamic_cast<arrow::BinaryBuilder*>(builder.field_builder(0)); + auto* metadata_builder = dynamic_cast<arrow::BinaryBuilder*>(builder.field_builder(1)); + if (value_builder == nullptr || metadata_builder == nullptr) { + return Status::InvalidArgument("Binary Variant V2 Arrow child builders must be binary"); + } + + // GenericVariant assumes its input is valid, and Paimon's unshredded writer copies these two + // buffers without inspecting them. Validate once at the Doris-to-Paimon boundary so a write + // cannot commit bytes which Paimon is unable to read later. + const auto outer_nulls = forced_nulls(null_map); + visit_variant_v2_values( + column, start, end, outer_nulls, + [&](size_t) { require_variant_arrow_status(builder.AppendNull()); }, + [&](size_t row, VariantRef value) { + try { + constexpr size_t PAIMON_VARIANT_SIZE_LIMIT = 128 * 1024 * 1024; + if (value.value.size > PAIMON_VARIANT_SIZE_LIMIT || + value.metadata.size > PAIMON_VARIANT_SIZE_LIMIT) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "exceeds the 128 MiB value/metadata limit"); + } + value.metadata.validate(); + validate_paimon_variant_value(value); + } catch (const Exception& e) { + throw Exception(e.code(), "Paimon Variant V2 row {} is incompatible: {}", row, + e.what()); + } + require_variant_arrow_status(builder.Append()); Review Comment: [P1] Split the block before Arrow binary offsets overflow The 128 MiB checks only bound each row, while both children are still Arrow `BinaryBuilder`s with cumulative 32-bit offsets. `ColumnVariantV2` can share one metadata dictionary across many source rows, but this loop appends the complete metadata for every row. For example, 4096 rows sharing about 1 MiB of metadata expand the metadata child to about 4 GiB even though the source column remains much smaller, so a valid write fails once the Arrow child crosses the 2 GiB offset limit. The subsequent IPC stream and JNI `DirectByteBuffer` also need an aggregate bound. Please preflight cumulative value/metadata and IPC sizes and split the block into multiple record batches/JNI writes before those limits, with an aggregate-boundary regression. The earlier thread about this issue was resolved, but the current head still converts the entire block in one batch. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
