github-actions[bot] commented on code in PR #69986:
URL: https://github.com/apache/doris/pull/69986#discussion_r4446137427
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -476,11 +479,11 @@ public void createScanRangeLocations() throws
UserException {
admissionResult = fileCacheAdmissionCheck();
}
- if (isBatchMode()) {
+ if (sessionVariable.getEnableExternalTableBatchMode() &&
isBatchMode()) {
// File splits are generated lazily, and fetched by backends while
scanning.
// Only provide the unique ID of split source to backend.
- splitAssignment = new SplitAssignment(backendPolicy, this,
this::splitToScanRange,
- locationProperties, pathPartitionKeys, admissionResult);
+ getScanTaskLease();
Review Comment:
[P1] Hudi can miss cancellation before syncFsView starts. Registering the
statement lease here lets cleanup stop the SplitAssignment while Hudi's
BatchFsViewOwner has not yet published its sync thread.
SplitAssignment.addCloseable then closes the owner with syncThread still null;
Hudi subsequently calls beginSync and the potentially blocking syncFsView
unconditionally (lines 738-744), so the stop interrupt is missed and a
cancelled query can remain blocked on remote timeline I/O. Check the stop state
before entering sync or make owner publication/close atomically interrupt an
already-stopping sync.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/cache/IcebergPartitionValueAdapter.java:
##########
@@ -0,0 +1,312 @@
+// 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.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 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();
+ }
+ // 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.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] Preserve historical DATE partition values across DATE-to-TIMESTAMP_NANO
evolution. DorisDataTableScan rebinds historical manifest specs to the current
schema, so ManifestCacheValue.forScan reaches this unsupported conversion for
old date-partitioned files and throws. In batch mode this runs lazily from
IcebergScanNode's manifest iterator after the manifest-cache fallback boundary,
causing the query to fail instead of scanning the historical partition (and
dropped promoted fields can also mismatch equality-delete partition keys). Add
a correct DATE/TIMESTAMP_NANO adaptation or retain the historical spec/value
for matching, with a partitioned streaming 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]
