github-actions[bot] commented on code in PR #69986: URL: https://github.com/apache/doris/pull/69986#discussion_r4446894210
########## 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: [P1] Support DATE to ordinary TIMESTAMP partition promotion. The adapter converts DATE only when the target is TIMESTAMP_NANO; an old identity-partitioned DATE file rebound to an ordinary Iceberg TIMESTAMP (microseconds) reaches the final unsupported-conversion exception. DorisDataTableScan rebinds historical specs by field ID, and the batch manifest iterator evaluates this adaptation lazily, so the producer fails the query instead of reading the historical partition. Add an exact days-to-micros conversion with overflow checks and a partitioned DATE-to-TIMESTAMP regression covering data and delete files. ########## 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: [P1] Normalize dropped DATE and ordinary TIMESTAMP keys consistently. In the UNKNOWN target branch, dropped DATE values become BigInteger days multiplied by NANOS_PER_DAY, while ordinary TIMESTAMP values fall through as Long microseconds; only TIMESTAMP_NANO uses the common BigInteger representation. After a partition field is promoted from DATE to ordinary TIMESTAMP and dropped, an old data file and a current equality-delete file therefore produce unequal keys for DeleteFileIndex, so the delete can be omitted. Convert ordinary TIMESTAMP to the same lossless representation and add a dropped DATE-to-TIMESTAMP equality-delete regression. ########## 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: [P2] Keep adapted Iceberg files serializable for task reuse. ManifestCacheValue.forScan returns DataView/DeleteView when a historical manifest partition type differs from the rebound spec, but FileView implements only ContentFile and is not Serializable. Non-batch manifest planning serializes each BaseFileScanTask; this throws and planFileScanTaskWithoutReuse falls back to a full SDK scan, so adapted DATE/TIMESTAMP or dropped-field tables bypass statement manifest/task reuse. Make the views serializable with serializable state, or materialize concrete serializable Iceberg file objects, and add a non-batch adapted-task reuse regression test. -- 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]
