924060929 commented on code in PR #69986: URL: https://github.com/apache/doris/pull/69986#discussion_r4447229753
########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/cache/IcebergPartitionValueAdapter.java: ########## @@ -0,0 +1,331 @@ +// 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.iceberg.cache; + +import org.apache.iceberg.ContentFile; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Query-local partition adaptation; all other content-file metadata is delegated unchanged. */ +final class IcebergPartitionValueAdapter { + private static final long NANOS_PER_DAY = 86_400_000_000_000L; + private final int[] sourcePositions; + private final Type[] sourceTypes; + private final Type[] targetTypes; + private final PartitionData template; + + IcebergPartitionValueAdapter(Types.StructType source, Types.StructType target) { + int size = target.fields().size(); + sourcePositions = new int[size]; + sourceTypes = new Type[size]; + targetTypes = new Type[size]; + template = new PartitionData(target); + for (int i = 0; i < size; i++) { + Types.NestedField field = target.fields().get(i); + targetTypes[i] = field.type(); + sourcePositions[i] = -1; + for (int j = 0; j < source.fields().size(); j++) { + if (source.fields().get(j).fieldId() == field.fieldId()) { + sourcePositions[i] = j; + sourceTypes[i] = source.fields().get(j).type(); + break; + } + } + if (sourcePositions[i] < 0) { + throw new IllegalArgumentException("Missing cached partition field " + field.fieldId()); + } + } + } + + DataFile adapt(DataFile file) { + return new DataView(file, adapt((PartitionData) file.partition())); + } + + DeleteFile adapt(DeleteFile file) { + return new DeleteView(file, adapt((PartitionData) file.partition())); + } + + PartitionData adapt(PartitionData source) { + // Copies reuse the Avro schema; do not reconstruct/serialize it for every file. + PartitionData original = source.copy(); + PartitionData result = template.copy(); + for (int i = 0; i < sourcePositions.length; i++) { + if (sourcePositions[i] >= 0) { + result.set(i, convert(original.get(sourcePositions[i], Object.class), sourceTypes[i], targetTypes[i])); + } + } + return result; + } + + private static Object convert(Object value, Type source, Type target) { + if (value == null) { + return null; + } + if (target.typeId() == Type.TypeID.UNKNOWN) { + // A dropped source has no target type. Different cache loaders may have decoded the same + // physical value before or after promotion, so preserve its value in a common representation. + // Iceberg's UNKNOWN comparator uses natural ordering: Integer/Long and Float/Double cannot + // compare across Java classes and would silently fail data/delete partition-key matching. + if (source.typeId() == Type.TypeID.INTEGER) { + return ((Integer) value).longValue(); + } + if (source.typeId() == Type.TypeID.FLOAT) { + return ((Float) value).doubleValue(); + } + if (source.typeId() == Type.TypeID.DATE) { + // A dropped DATE may lie outside the long-nanos range. Keep the key lossless. + return BigInteger.valueOf((Integer) value).multiply(BigInteger.valueOf(NANOS_PER_DAY)); + } + if (source.typeId() == Type.TypeID.TIMESTAMP_NANO) { + return BigInteger.valueOf((Long) value); + } + // Decimal promotion changes precision only; its Java type and scale already agree. + return value; + } + if (source.equals(target)) { + return value; + } + if (source.typeId() == Type.TypeID.DATE && target.typeId() == Type.TypeID.TIMESTAMP_NANO) { + return Math.multiplyExact(((Integer) value).longValue(), NANOS_PER_DAY); + } + if (source.typeId() == Type.TypeID.TIMESTAMP_NANO && target.typeId() == Type.TypeID.DATE) { + long nanos = (Long) value; + if (nanos % NANOS_PER_DAY != 0) { + throw new IllegalArgumentException("Partition value cannot be represented as date: " + value); + } + return Math.toIntExact(nanos / NANOS_PER_DAY); + } + if (source.typeId() == Type.TypeID.INTEGER && target.typeId() == Type.TypeID.LONG) { + return ((Integer) value).longValue(); + } + if (source.typeId() == Type.TypeID.LONG && target.typeId() == Type.TypeID.INTEGER) { + return Math.toIntExact((Long) value); + } + if (source.typeId() == Type.TypeID.FLOAT && target.typeId() == Type.TypeID.DOUBLE) { + return ((Float) value).doubleValue(); + } + if (source.typeId() == Type.TypeID.DOUBLE && target.typeId() == Type.TypeID.FLOAT) { + double original = (Double) value; + float narrowed = (float) original; + if (!Double.isNaN(original) && Double.compare(original, (double) narrowed) != 0) { + throw new IllegalArgumentException("Partition value cannot be represented as float: " + value); + } + return narrowed; + } + if (source.typeId() == Type.TypeID.DECIMAL && target.typeId() == Type.TypeID.DECIMAL) { + Types.DecimalType decimalType = (Types.DecimalType) target; + BigDecimal decimal = ((BigDecimal) value).setScale(decimalType.scale()); + if (decimal.precision() > decimalType.precision()) { + throw new IllegalArgumentException("Partition value exceeds " + target + ": " + value); + } + return decimal; + } + throw new IllegalArgumentException("Cannot adapt partition value from " + source + " to " + target); Review Comment: Fixed in 68ef8c21edda4dd13fec92dcc835b590d8c6f81e. DATE partitions now convert exactly to ordinary without-zone TIMESTAMP microseconds, with overflow checks and exact-midnight checks for reverse historical adaptation. DATE-to-zoned timestamp conversions are rejected for both precisions. Added real partitioned streaming metadata cases and unit tests for positive/negative days, nulls, overflow and lossy narrowing. This is a reader repair; SQL ALTER type support is unchanged. All 253 related FE tests and full FE packaging/Checkstyle pass. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/cache/IcebergPartitionValueAdapter.java: ########## @@ -0,0 +1,331 @@ +// 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.iceberg.cache; + +import org.apache.iceberg.ContentFile; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Query-local partition adaptation; all other content-file metadata is delegated unchanged. */ +final class IcebergPartitionValueAdapter { + private static final long NANOS_PER_DAY = 86_400_000_000_000L; + private final int[] sourcePositions; + private final Type[] sourceTypes; + private final Type[] targetTypes; + private final PartitionData template; + + IcebergPartitionValueAdapter(Types.StructType source, Types.StructType target) { + int size = target.fields().size(); + sourcePositions = new int[size]; + sourceTypes = new Type[size]; + targetTypes = new Type[size]; + template = new PartitionData(target); + for (int i = 0; i < size; i++) { + Types.NestedField field = target.fields().get(i); + targetTypes[i] = field.type(); + sourcePositions[i] = -1; + for (int j = 0; j < source.fields().size(); j++) { + if (source.fields().get(j).fieldId() == field.fieldId()) { + sourcePositions[i] = j; + sourceTypes[i] = source.fields().get(j).type(); + break; + } + } + if (sourcePositions[i] < 0) { + throw new IllegalArgumentException("Missing cached partition field " + field.fieldId()); + } + } + } + + DataFile adapt(DataFile file) { + return new DataView(file, adapt((PartitionData) file.partition())); + } + + DeleteFile adapt(DeleteFile file) { + return new DeleteView(file, adapt((PartitionData) file.partition())); + } + + PartitionData adapt(PartitionData source) { + // Copies reuse the Avro schema; do not reconstruct/serialize it for every file. + PartitionData original = source.copy(); + PartitionData result = template.copy(); + for (int i = 0; i < sourcePositions.length; i++) { + if (sourcePositions[i] >= 0) { + result.set(i, convert(original.get(sourcePositions[i], Object.class), sourceTypes[i], targetTypes[i])); + } + } + return result; + } + + private static Object convert(Object value, Type source, Type target) { + if (value == null) { + return null; + } + if (target.typeId() == Type.TypeID.UNKNOWN) { + // A dropped source has no target type. Different cache loaders may have decoded the same + // physical value before or after promotion, so preserve its value in a common representation. + // Iceberg's UNKNOWN comparator uses natural ordering: Integer/Long and Float/Double cannot + // compare across Java classes and would silently fail data/delete partition-key matching. + if (source.typeId() == Type.TypeID.INTEGER) { + return ((Integer) value).longValue(); + } + if (source.typeId() == Type.TypeID.FLOAT) { + return ((Float) value).doubleValue(); + } + if (source.typeId() == Type.TypeID.DATE) { Review Comment: Fixed in 68ef8c21edda4dd13fec92dcc835b590d8c6f81e. Dropped DATE, microsecond TIMESTAMP and nanosecond TIMESTAMP keys now use the same lossless BigInteger nanosecond representation, so comparisons and hashes agree without overflowing the source ranges. The dropped DATE-to-microsecond partition test now retains its equality delete, and unit tests compare keys across all three representations. All 253 related FE tests and full FE packaging/Checkstyle pass. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/cache/IcebergPartitionValueAdapter.java: ########## @@ -0,0 +1,331 @@ +// 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.iceberg.cache; + +import org.apache.iceberg.ContentFile; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Query-local partition adaptation; all other content-file metadata is delegated unchanged. */ +final class IcebergPartitionValueAdapter { + private static final long NANOS_PER_DAY = 86_400_000_000_000L; + private final int[] sourcePositions; + private final Type[] sourceTypes; + private final Type[] targetTypes; + private final PartitionData template; + + IcebergPartitionValueAdapter(Types.StructType source, Types.StructType target) { + int size = target.fields().size(); + sourcePositions = new int[size]; + sourceTypes = new Type[size]; + targetTypes = new Type[size]; + template = new PartitionData(target); + for (int i = 0; i < size; i++) { + Types.NestedField field = target.fields().get(i); + targetTypes[i] = field.type(); + sourcePositions[i] = -1; + for (int j = 0; j < source.fields().size(); j++) { + if (source.fields().get(j).fieldId() == field.fieldId()) { + sourcePositions[i] = j; + sourceTypes[i] = source.fields().get(j).type(); + break; + } + } + if (sourcePositions[i] < 0) { + throw new IllegalArgumentException("Missing cached partition field " + field.fieldId()); + } + } + } + + DataFile adapt(DataFile file) { + return new DataView(file, adapt((PartitionData) file.partition())); + } + + DeleteFile adapt(DeleteFile file) { + return new DeleteView(file, adapt((PartitionData) file.partition())); + } + + PartitionData adapt(PartitionData source) { + // Copies reuse the Avro schema; do not reconstruct/serialize it for every file. + PartitionData original = source.copy(); + PartitionData result = template.copy(); + for (int i = 0; i < sourcePositions.length; i++) { + if (sourcePositions[i] >= 0) { + result.set(i, convert(original.get(sourcePositions[i], Object.class), sourceTypes[i], targetTypes[i])); + } + } + return result; + } + + private static Object convert(Object value, Type source, Type target) { + if (value == null) { + return null; + } + if (target.typeId() == Type.TypeID.UNKNOWN) { + // A dropped source has no target type. Different cache loaders may have decoded the same + // physical value before or after promotion, so preserve its value in a common representation. + // Iceberg's UNKNOWN comparator uses natural ordering: Integer/Long and Float/Double cannot + // compare across Java classes and would silently fail data/delete partition-key matching. + if (source.typeId() == Type.TypeID.INTEGER) { + return ((Integer) value).longValue(); + } + if (source.typeId() == Type.TypeID.FLOAT) { + return ((Float) value).doubleValue(); + } + if (source.typeId() == Type.TypeID.DATE) { + // A dropped DATE may lie outside the long-nanos range. Keep the key lossless. + return BigInteger.valueOf((Integer) value).multiply(BigInteger.valueOf(NANOS_PER_DAY)); + } + if (source.typeId() == Type.TypeID.TIMESTAMP_NANO) { + return BigInteger.valueOf((Long) value); + } + // Decimal promotion changes precision only; its Java type and scale already agree. + return value; + } + if (source.equals(target)) { + return value; + } + if (source.typeId() == Type.TypeID.DATE && target.typeId() == Type.TypeID.TIMESTAMP_NANO) { + return Math.multiplyExact(((Integer) value).longValue(), NANOS_PER_DAY); + } + if (source.typeId() == Type.TypeID.TIMESTAMP_NANO && target.typeId() == Type.TypeID.DATE) { + long nanos = (Long) value; + if (nanos % NANOS_PER_DAY != 0) { + throw new IllegalArgumentException("Partition value cannot be represented as date: " + value); + } + return Math.toIntExact(nanos / NANOS_PER_DAY); + } + if (source.typeId() == Type.TypeID.INTEGER && target.typeId() == Type.TypeID.LONG) { + return ((Integer) value).longValue(); + } + if (source.typeId() == Type.TypeID.LONG && target.typeId() == Type.TypeID.INTEGER) { + return Math.toIntExact((Long) value); + } + if (source.typeId() == Type.TypeID.FLOAT && target.typeId() == Type.TypeID.DOUBLE) { + return ((Float) value).doubleValue(); + } + if (source.typeId() == Type.TypeID.DOUBLE && target.typeId() == Type.TypeID.FLOAT) { + double original = (Double) value; + float narrowed = (float) original; + if (!Double.isNaN(original) && Double.compare(original, (double) narrowed) != 0) { + throw new IllegalArgumentException("Partition value cannot be represented as float: " + value); + } + return narrowed; + } + if (source.typeId() == Type.TypeID.DECIMAL && target.typeId() == Type.TypeID.DECIMAL) { + Types.DecimalType decimalType = (Types.DecimalType) target; + BigDecimal decimal = ((BigDecimal) value).setScale(decimalType.scale()); + if (decimal.precision() > decimalType.precision()) { + throw new IllegalArgumentException("Partition value exceeds " + target + ": " + value); + } + return decimal; + } + throw new IllegalArgumentException("Cannot adapt partition value from " + source + " to " + target); + } + + private abstract static class FileView<F extends ContentFile<F>> implements ContentFile<F> { Review Comment: Fixed in 68ef8c21edda4dd13fec92dcc835b590d8c6f81e. The shared static file-view base and both concrete wrappers are serializable with explicit serialVersionUID; their only state is the native Iceberg content file and query-owned PartitionData. No adapter, cache loader or statement resource is captured. A real supported int-to-long partition evolution test reproduced NotSerializableException before the change and now verifies serialized data/delete partition values and one manifest load across two equivalent non-batch relations. All 253 related FE tests and full FE packaging/Checkstyle pass. -- 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]
