github-actions[bot] commented on code in PR #66321:
URL: https://github.com/apache/doris/pull/66321#discussion_r3710747059
##########
fe/pom.xml:
##########
@@ -386,7 +386,7 @@ under the License.
<quartz.version>2.3.2</quartz.version>
<aircompressor.version>2.0.3</aircompressor.version>
<!-- paimon -->
- <paimon.version>1.3.1</paimon.version>
+ <paimon.version>1.4.2</paimon.version>
Review Comment:
[P1] Preserve the mixed-version Paimon wire contract
This upgrades the classes used by the FE-to-BE Java-serialization protocol,
but the new SDK cannot decode every object produced by 1.3.1 during a rolling
upgrade. Doris serializes the actual Paimon `Table` and `Split`; in Paimon
1.3.1 `FallbackReadFileStoreTable` has no explicit UID and has a different
field/constructor/API shape from 1.4.2, so the computed UID changes and
deserialization fails before the new wrapper-order handling runs. `DataSplit`
keeps UID 7, but the 1.4.2 decoder explicitly throws for old non-empty
`beforeFiles`/`beforeDeletionFiles`. Please introduce a stable versioned Doris
DTO or a compatibility bridge, and exercise captured 1.3.1 fallback-table and
incremental-split fixtures with the 1.4.2 reader (plus the reverse direction
Doris supports).
##########
fe/be-java-extensions/java-udf/src/main/resources/package.xml:
##########
@@ -34,6 +34,19 @@ under the License.
<unpackOptions>
<excludes>
<exclude>**/Log4j2Plugins.dat</exclude>
+ <!--
+ TODO: Remove these Paimon exclusions after picking the
master change that
+ removes the hive-catalog-shade dependency.
+
+ hive-catalog-shade embeds Paimon 1.3 classes. This fat
jar is on the
+ JVM application classpath and therefore wins
parent-first resolution
+ over paimon-connector's isolated Paimon runtime. Java
UDF does not use
+ Paimon; keep those embedded classes out so the
connector owns its SDK.
+ -->
+ <exclude>org/apache/paimon/**</exclude>
+ <!-- ServiceLoader also consults parent resources. Do not
retain service
+ descriptors whose providers were removed with the
embedded SDK. -->
+ <exclude>META-INF/services/org.apache.paimon*</exclude>
Review Comment:
[P1] Remove descriptors for the providers stripped above
This filename filter misses standard SPI descriptors whose contents name
relocated Paimon classes. The currently resolved shade jar contains
`META-INF/services/java.time.chrono.Chronology` with nine
`org.apache.paimon.shade.org.threeten.extra.chrono.*` providers; line 46
removes every one of those classes, but this line leaves the descriptor in the
Java-UDF jar that `start_be.sh` places on the application classpath. OpenJDK's
chronology discovery then raises `ServiceConfigurationError` when it enumerates
the missing providers. Please exclude this descriptor (or retain its providers)
and add an assembly-level check that every retained service entry has a
packaged implementation.
##########
be/src/core/data_type_serde/data_type_variant_v2_serde.cpp:
##########
@@ -175,6 +176,130 @@ void preflight_json(const IColumn& column, size_t start,
size_t end,
});
}
+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: {
+ const auto primitive_id = value.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));
+ }
+ 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;
Review Comment:
[P1] Bound the cumulative Arrow binary buffers
This validates each value/metadata buffer against Paimon's 128 MiB row
limit, but both children use Arrow 17 `BinaryBuilder`, whose cumulative data is
capped at `INT32_MAX - 1`. `ColumnVariantV2` can deduplicate one large metadata
dictionary across many rows, whereas this transport appends that full metadata
for every row; sixteen individually valid near-limit rows can therefore fail
partway through Arrow conversion even though the source block is much smaller.
Please preflight cumulative child bytes and split the block into record
batches/JNI writes before either 32-bit offset buffer overflows, with an
aggregate-boundary regression.
##########
regression-test/suites/paimon_write/test_paimon_write_variant_table_modes.groovy:
##########
@@ -0,0 +1,179 @@
+// 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.
+
+suite("test_paimon_write_variant_table_modes", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_variant_modes_catalog"
+ String dbName = "test_pw_variant_modes_db"
+ String root = '$'
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_variant_pk;
+ CREATE TABLE paimon.${dbName}.t_variant_pk (
+ id INT,
+ payload VARIANT,
+ version BIGINT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '2',
+ 'bucket-key' = 'id',
+ 'file.format' = 'parquet'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_variant_dynamic_bucket;
+ CREATE TABLE paimon.${dbName}.t_variant_dynamic_bucket (
+ id INT,
+ payload VARIANT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '-1',
+ 'file.format' = 'parquet'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_variant_schema;
+ CREATE TABLE paimon.${dbName}.t_variant_schema (
+ id INT,
+ name STRING
+ ) USING paimon
+ TBLPROPERTIES ('file.format' = 'parquet');
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_non_variant;
+ CREATE TABLE paimon.${dbName}.t_non_variant (
+ id INT,
+ payload STRING
+ ) USING paimon;
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_variant_required;
+ CREATE TABLE paimon.${dbName}.t_variant_required (
+ id INT,
+ payload VARIANT NOT NULL
+ ) USING paimon
+ TBLPROPERTIES ('file.format' = 'parquet');
+ """
+
+ sql """DROP CATALOG IF EXISTS ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type' = 'paimon',
+ 'paimon.catalog.type' = 'filesystem',
+ 'warehouse' = 's3://warehouse/wh',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.path.style.access' = 'true'
+ )
+ """
+ sql """SWITCH ${catalogName}"""
+ sql """USE ${dbName}"""
+ sql """SET enable_variant_v2 = true"""
+
+ try {
+ // Fixed-bucket primary-key table: later rows replace the same key.
+ sql """
+ INSERT INTO t_variant_pk VALUES
+ (1, parse_to_variant('{"state":"v1","n":1}'), 1),
+ (2, parse_to_variant('{"state":"stable","n":2}'), 1),
+ (1, parse_to_variant('{"state":"v2","n":10}'), 2)
+ """
+ sql """
+ INSERT INTO t_variant_pk VALUES
+ (1, parse_to_variant('{"state":"v3","n":100}'), 3)
+ """
+ def pkRows = spark_paimon """
+ SELECT id,
+ variant_get(payload, '${root}.state', 'string'),
+ variant_get(payload, '${root}.n', 'int'),
+ version
+ FROM paimon.${dbName}.t_variant_pk
+ ORDER BY id
+ """
+ assertEquals([
+ ["1", "v3", "100", "3"],
+ ["2", "stable", "2", "1"]
+ ], pkRows.collect { row -> row.collect { value -> value.toString() } })
+
+ // Dynamic bucket routing with Variant values.
+ sql """
+ INSERT INTO t_variant_dynamic_bucket
+ SELECT number,
+ parse_to_variant(CONCAT('{"bucket":"dynamic","id":',
number, '}'))
+ FROM numbers("number" = "32")
+ """
+ def dynamicRows = spark_paimon """
Review Comment:
[P2] Verify each dynamic-bucket payload against its key
The insert deliberately writes the same number as the row key and
`payload.id`, but `COUNT(*) = 32` plus `SUM(payload.id) = 496` still passes if
all 32 Variant payloads are permuted among the keys. That leaves this
routing/transport case unable to detect cross-row association bugs. Please
assert zero rows where `id != variant_get(payload, '$.id', 'int')` (and, if
useful, verify the constant bucket field).
##########
docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl:
##########
@@ -35,8 +35,8 @@ services:
- ./spark-defaults.conf:/opt/spark/conf/spark-defaults.conf
-
./data/input/jars/iceberg-aws-bundle-1.10.1.jar:/opt/spark/jars/iceberg-aws-bundle-1.10.1.jar
-
./data/input/jars/iceberg-spark-runtime-4.0_2.13-1.10.1.jar:/opt/spark/jars/iceberg-spark-runtime-4.0_2.13-1.10.1.jar
- -
./data/input/jars/paimon-s3-1.3.1.jar:/opt/spark/jars/paimon-s3-1.3.1.jar
- -
./data/input/jars/paimon-spark-4.0-1.3.1.jar:/opt/spark/jars/paimon-spark-4.0-1.3.1.jar
+ -
./data/input/jars/paimon-s3-1.4.2.jar:/opt/spark/jars/paimon-s3-1.4.2.jar
Review Comment:
[P1] Ship the jars referenced by this template
`start_iceberg` still downloads only the fixed `iceberg_data_spark40.zip`
fixture and never downloads or renames Paimon jars. The current object (last
modified 2026-02-28) contains only `paimon-spark-4.0-1.3.1.jar` and
`paimon-s3-1.3.1.jar`, so neither 1.4.2 bind-mount source exists; Compose will
either fail the mount or expose a directory where Spark expects a jar. Please
publish/update a versioned fixture or download checksum-pinned 1.4.2 artifacts,
and fail fast by checking each source is a regular file before `compose up`.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/types/VariantType.java:
##########
@@ -336,6 +336,88 @@ public static boolean containsVariant(DataType dataType) {
return false;
}
+ /** Whether this is a legacy Variant leaf rather than the compute-only V2
representation. */
+ public static boolean isLegacyVariant(DataType dataType) {
+ return dataType instanceof VariantType && !((VariantType)
dataType).isComputeV2();
+ }
+
+ /**
+ * Whether the Variant V2 execution kernel can convert this source type.
+ *
+ * <p>This mirrors the BE {@code execute_to_variant} contract: encoded
JSON, nested arrays,
+ * compute V2 values, and the scalar types supported by the typed Variant
representation.
+ * MAP, STRUCT, TIMEV2 and DECIMAL256 are intentionally excluded until
their BE conversions
+ * are implemented.</p>
+ */
+ public static boolean isSupportedComputeV2CastSource(DataType dataType) {
+ if (dataType.isNullType() || dataType.isJsonType()) {
Review Comment:
[P1] Align ARRAY<NULL> admission with the BE encoder
This returns true for `NullType` and recursively for `ArrayType`, so
`CAST(array(NULL) AS VARIANT)` (and the same source written to a Paimon Variant
leaf) passes analysis. The BE array encoder only accepts an `INVALID_TYPE` leaf
when its nested column is empty; `array(NULL)` has a one-row `ColumnNothing`
and fails with `Array element type ... cannot be cast to Variant V2`. Please
either encode NULL-typed elements as Variant nulls, including nested NULL-only
arrays, or reject this non-empty shape before execution, and add
`array(NULL)`/nested regressions without losing the valid empty-array case.
--
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]
