Gabriel39 commented on code in PR #66321: URL: https://github.com/apache/doris/pull/66321#discussion_r3691145699
########## regression-test/suites/paimon_write/test_paimon_write_variant_shredding.groovy: ########## @@ -0,0 +1,250 @@ +// 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_shredding", "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_shredding_catalog" + String dbName = "test_pw_variant_shredding_db" + String root = '$' + String shreddingSchema = + '{"type":"ROW","fields":[{"name":"payload","type":{"type":"ROW","fields":[' + + '{"name":"age","type":"INT"},' + + '{"name":"city","type":"STRING"},' + + '{"name":"active","type":"BOOLEAN"},' + + '{"name":"profile","type":{"type":"ROW","fields":[' + + '{"name":"name","type":"STRING"},' + + '{"name":"scores","type":{"type":"ARRAY","element":"INT"}}' + + ']}}]}}]}' + + spark_paimon_multi """ + CREATE DATABASE IF NOT EXISTS paimon.${dbName}; + + DROP TABLE IF EXISTS paimon.${dbName}.t_variant_shredded; + CREATE TABLE paimon.${dbName}.t_variant_shredded ( + id INT, + payload VARIANT + ) USING paimon + TBLPROPERTIES ( + 'file.format' = 'parquet', + 'write-only' = 'true', + 'parquet.variant.shreddingSchema' = '${shreddingSchema}' Review Comment: Please upgrade the bundled Paimon dependency from 1.3.1 to at least 1.4.2 and exercise automatic per-file shredding inference here. Paimon 1.4.2 provides `variant.inferShreddingSchema=true` together with the inference buffer, width, depth, and field-cardinality controls, so a table-level hard-coded `parquet.variant.shreddingSchema` should not be the primary coverage for this feature. Please write separate batches/files with different field and type distributions, inspect their raw Parquet `typed_value` layouts, and verify that Paimon reconstructs all files into the same logical Variant values. The existing type-mismatch and residual-value assertions are valuable and should be retained under the inferred-schema path. If fixed-schema compatibility is still needed, it can remain as a smaller separate case. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonArrowConverter.java: ########## @@ -249,17 +250,56 @@ private Object convertToPaimonType(Object value, Field arrowField, DataType targ } static Object convertText(byte[] value, DataType targetType) { - if (targetType instanceof VariantType) { - return toVariant(value); - } if (targetType instanceof BinaryType || targetType instanceof VarBinaryType) { return value; } return BinaryString.fromBytes(value); } - private static GenericVariant toVariant(byte[] json) { - return GenericVariant.fromJson(new String(json, StandardCharsets.UTF_8)); + private GenericVariant convertVariantVector(StructVector vector, int index) { + List<FieldVector> children = vector.getChildrenFromFields(); + if (children.size() != 2 + || !VARIANT_VALUE_FIELD.equals(children.get(0).getName()) + || !VARIANT_METADATA_FIELD.equals(children.get(1).getName()) + || !(children.get(0) instanceof VarBinaryVector) + || !(children.get(1) instanceof VarBinaryVector)) { + throw new IllegalArgumentException( + "Paimon VARIANT binary transport requires Arrow " + + "struct<value: binary, metadata: binary>, but got " + + vector.getField()); + } + VarBinaryVector valueVector = (VarBinaryVector) children.get(0); + VarBinaryVector metadataVector = (VarBinaryVector) children.get(1); + if (valueVector.isNull(index) || metadataVector.isNull(index)) { + throw new IllegalArgumentException( + "A non-null Paimon VARIANT struct requires non-null value and metadata"); + } + GenericVariant variant = new GenericVariant( + valueVector.get(index), metadataVector.get(index)); + ensurePaimonVariantCompatibility(variant); Review Comment: Please avoid, fuse, or justify the recursive compatibility walk on every row. `VarBinaryVector.get` already materializes the value and metadata byte arrays, and `ensurePaimonVariantCompatibility` then traverses the complete Variant tree and creates nested wrapper objects. With inferred or configured shredding, the Paimon writer traverses the value again, so large or deeply nested Variant workloads pay two full walks before the Parquet columns are emitted. After upgrading Paimon and tightening the FE/BE supported-type boundary, please consider removing this per-row walk or integrating validation with the shredding/inference traversal. If it must remain as a defensive boundary, please add a representative large/deep Variant write benchmark and document the measured throughput/allocation impact. -- 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]
