From 9efdb521b4b55bb31b88c0c9fa32c7ba2b1f3cb1 Mon Sep 17 00:00:00 2001 From: suxiaogang Date: Fri, 24 Jul 2026 17:18:08 +0800 Subject: [PATCH 1/3] [fix](multi-catalog) Preserve external partition metadata ### What problem does this PR solve? Issue Number: None Related PR: #62821, #65583 Problem Summary: branch-4.0 only backported the Paimon side of the external partition metadata fixes. Hive-style scans, Hudi, Iceberg, and file load paths could still lose explicit NULL semantics or rely on physical file columns that are absent from migrated partitioned files. Iceberg also attached partition metadata only when runtime partition pruning was enabled and did not safely account for mixed transforms or partition evolution. This change adds common path partition parsing with aligned null flags, applies it to external scans and load paths, and sends stable Iceberg identity partition metadata for every split while respecting branch-4.0 scan-level slot constraints. ### Release note Fix partition-column materialization and NULL handling for Hive, Hudi, Iceberg, and file load paths on branch-4.0. ### Check List (For Author) - Test: Unit Test - ./run-fe-ut.sh --run org.apache.doris.datasource.FilePartitionUtilsTest,org.apache.doris.datasource.hive.source.HiveScanNodeTest,org.apache.doris.datasource.iceberg.IcebergUtilsTest,org.apache.doris.datasource.iceberg.source.IcebergScanNodeTest,org.apache.doris.datasource.paimon.source.PaimonScanNodeTest - Behavior changed: Yes. External partition metadata and NULL values are materialized consistently per split. - Does this need documentation: No --- .../doris/datasource/FileGroupInfo.java | 24 +-- .../doris/datasource/FilePartitionUtils.java | 148 ++++++++++++++++++ .../doris/datasource/FileQueryScanNode.java | 16 +- .../datasource/hive/HiveMetaStoreCache.java | 5 +- .../datasource/hudi/source/HudiScanNode.java | 6 +- .../datasource/iceberg/IcebergUtils.java | 84 ++++++---- .../iceberg/source/IcebergScanNode.java | 100 ++++++------ .../nereids/load/NereidsFileGroupInfo.java | 25 +-- .../datasource/FilePartitionUtilsTest.java | 71 +++++++++ .../datasource/iceberg/IcebergUtilsTest.java | 84 ++++++++++ .../iceberg/source/IcebergScanNodeTest.java | 43 +++++ 11 files changed, 494 insertions(+), 112 deletions(-) create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/FilePartitionUtils.java create mode 100644 fe/fe-core/src/test/java/org/apache/doris/datasource/FilePartitionUtilsTest.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileGroupInfo.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileGroupInfo.java index d81ba7daa8419c..9efeda06dbc83e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileGroupInfo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileGroupInfo.java @@ -26,7 +26,6 @@ import org.apache.doris.common.Config; import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; -import org.apache.doris.common.util.BrokerUtil; import org.apache.doris.common.util.Util; import org.apache.doris.load.BrokerFileGroup; import org.apache.doris.planner.FileLoadScanNode; @@ -266,11 +265,12 @@ public void createScanRangeLocationsUnsplittable(FileLoadScanNode.ParamCreateCon Util.getOrInferCompressType(context.fileGroup.getFileFormatProperties().getCompressionType(), fileStatus.path); context.params.setCompressType(compressType); - List columnsFromPath = BrokerUtil.parseColumnsFromPath(fileStatus.path, - context.fileGroup.getColumnNamesFromPath()); + FilePartitionUtils.ParsedColumnsFromPath columnsFromPath = + FilePartitionUtils.parseColumnsFromPathWithNullInfo(fileStatus.path, + context.fileGroup.getColumnNamesFromPath(), true, false); List columnsFromPathKeys = context.fileGroup.getColumnNamesFromPath(); - TFileRangeDesc rangeDesc = createFileRangeDesc(0, fileStatus, fileStatus.size, columnsFromPath, - columnsFromPathKeys); + TFileRangeDesc rangeDesc = createFileRangeDesc(0, fileStatus, fileStatus.size, + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); locations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); } scanRangeLocations.add(locations); @@ -312,15 +312,16 @@ public void createScanRangeLocationsSplittable(FileLoadScanNode.ParamCreateConte Util.getOrInferCompressType(context.fileGroup.getFileFormatProperties().getCompressionType(), fileStatus.path); context.params.setCompressType(compressType); - List columnsFromPath = BrokerUtil.parseColumnsFromPath(fileStatus.path, - context.fileGroup.getColumnNamesFromPath()); + FilePartitionUtils.ParsedColumnsFromPath columnsFromPath = + FilePartitionUtils.parseColumnsFromPathWithNullInfo(fileStatus.path, + context.fileGroup.getColumnNamesFromPath(), true, false); List columnsFromPathKeys = context.fileGroup.getColumnNamesFromPath(); // Assign scan range locations only for broker load. // stream load has only one file, and no need to set multi scan ranges. if (tmpBytes > bytesPerInstance && jobType != JobType.STREAM_LOAD) { long rangeBytes = bytesPerInstance - curInstanceBytes; TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, rangeBytes, - columnsFromPath, columnsFromPathKeys); + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); curFileOffset += rangeBytes; @@ -329,8 +330,8 @@ public void createScanRangeLocationsSplittable(FileLoadScanNode.ParamCreateConte curLocations = newLocations(context.params, brokerDesc, backendPolicy); curInstanceBytes = 0; } else { - TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, leftBytes, columnsFromPath, - columnsFromPathKeys); + TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, leftBytes, + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); curFileOffset = 0; curInstanceBytes += leftBytes; @@ -401,7 +402,7 @@ private TFileFormatType formatType(String fileFormat, String path) throws UserEx } private TFileRangeDesc createFileRangeDesc(long curFileOffset, TBrokerFileStatus fileStatus, long rangeBytes, - List columnsFromPath, List columnsFromPathKeys) { + List columnsFromPath, List columnsFromPathKeys, List columnsFromPathIsNull) { TFileRangeDesc rangeDesc = new TFileRangeDesc(); if (jobType == JobType.BULK_LOAD) { rangeDesc.setPath(fileStatus.path); @@ -410,6 +411,7 @@ private TFileRangeDesc createFileRangeDesc(long curFileOffset, TBrokerFileStatus rangeDesc.setFileSize(fileStatus.size); rangeDesc.setColumnsFromPath(columnsFromPath); rangeDesc.setColumnsFromPathKeys(columnsFromPathKeys); + rangeDesc.setColumnsFromPathIsNull(columnsFromPathIsNull); if (getFileType() == TFileType.FILE_HDFS) { URI fileUri = new Path(fileStatus.path).toUri(); rangeDesc.setFsName(fileUri.getScheme() + "://" + fileUri.getAuthority()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FilePartitionUtils.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FilePartitionUtils.java new file mode 100644 index 00000000000000..0a50b73d8d899e --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FilePartitionUtils.java @@ -0,0 +1,148 @@ +// 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; + +import org.apache.doris.common.UserException; +import org.apache.doris.datasource.hive.HiveMetaStoreCache; + +import com.google.common.collect.Lists; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Locale; + +/** + * Utility methods for parsing partition column values from Hive-style file paths. + */ +public final class FilePartitionUtils { + + private FilePartitionUtils() {} + + public static final class ParsedColumnsFromPath { + private final List values; + private final List isNull; + + private ParsedColumnsFromPath(List values, List isNull) { + this.values = values; + this.isNull = isNull; + } + + public List getValues() { + return values; + } + + public List getIsNull() { + return isNull; + } + } + + public static List parseColumnsFromPath(String filePath, List columnsFromPath) + throws UserException { + return parseColumnsFromPathWithNullInfo(filePath, columnsFromPath, true, false).getValues(); + } + + public static List parseColumnsFromPath( + String filePath, + List columnsFromPath, + boolean caseSensitive, + boolean isACID) + throws UserException { + return parseColumnsFromPathWithNullInfo(filePath, columnsFromPath, caseSensitive, isACID) + .getValues(); + } + + public static ParsedColumnsFromPath parseColumnsFromPathWithNullInfo( + String filePath, + List columnsFromPath, + boolean caseSensitive, + boolean isACID) + throws UserException { + if (columnsFromPath == null || columnsFromPath.isEmpty()) { + return new ParsedColumnsFromPath(Collections.emptyList(), Collections.emptyList()); + } + int pathCount = isACID ? 3 : 2; + List expectedColumns = columnsFromPath; + if (!caseSensitive) { + expectedColumns = new ArrayList<>(columnsFromPath.size()); + for (String path : columnsFromPath) { + expectedColumns.add(path.toLowerCase(Locale.ROOT)); + } + } + String[] strings = filePath.split("/"); + if (strings.length < 2) { + throw new UserException("Fail to parse columnsFromPath, expected: " + + expectedColumns + ", filePath: " + filePath); + } + String[] columns = new String[expectedColumns.size()]; + Boolean[] columnValueIsNull = new Boolean[expectedColumns.size()]; + int size = 0; + boolean skipOnce = true; + for (int i = strings.length - pathCount; i >= 0; i--) { + String str = strings[i]; + if (str != null && str.isEmpty()) { + continue; + } + if (str == null || !str.contains("=")) { + if (!isACID && skipOnce) { + skipOnce = false; + continue; + } + throw new UserException("Fail to parse columnsFromPath, expected: " + + expectedColumns + ", filePath: " + filePath); + } + skipOnce = false; + String[] pair = str.split("=", 2); + if (pair.length != 2) { + throw new UserException("Fail to parse columnsFromPath, expected: " + + expectedColumns + ", filePath: " + filePath); + } + String parsedColumnName = caseSensitive ? pair[0] : pair[0].toLowerCase(Locale.ROOT); + int index = expectedColumns.indexOf(parsedColumnName); + if (index == -1) { + continue; + } + boolean isNull = HiveMetaStoreCache.HIVE_DEFAULT_PARTITION.equals(pair[1]); + columns[index] = isNull ? "" : pair[1]; + columnValueIsNull[index] = isNull; + size++; + if (size >= expectedColumns.size()) { + break; + } + } + if (size != expectedColumns.size()) { + throw new UserException("Fail to parse columnsFromPath, expected: " + + expectedColumns + ", filePath: " + filePath); + } + return new ParsedColumnsFromPath(Lists.newArrayList(columns), Lists.newArrayList(columnValueIsNull)); + } + + public static ParsedColumnsFromPath normalizeColumnsFromPath(List columnsFromPath) { + if (columnsFromPath == null || columnsFromPath.isEmpty()) { + return new ParsedColumnsFromPath(Collections.emptyList(), Collections.emptyList()); + } + List values = new ArrayList<>(columnsFromPath.size()); + List isNull = new ArrayList<>(columnsFromPath.size()); + for (String value : columnsFromPath) { + boolean nullValue = value == null || HiveMetaStoreCache.HIVE_DEFAULT_PARTITION.equals(value); + values.add(nullValue ? "" : value); + isNull.add(nullValue); + } + return new ParsedColumnsFromPath(values, isNull); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java index 8fc003bd520562..29bfcf6a2b9e87 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java @@ -31,7 +31,6 @@ import org.apache.doris.common.AnalysisException; import org.apache.doris.common.NotImplementedException; import org.apache.doris.common.UserException; -import org.apache.doris.common.util.BrokerUtil; import org.apache.doris.common.util.Util; import org.apache.doris.datasource.hive.source.HiveSplit; import org.apache.doris.planner.PlanNodeId; @@ -410,11 +409,14 @@ private TScanRangeLocations splitToScanRange( HiveSplit hiveSplit = (HiveSplit) fileSplit; isACID = hiveSplit.isACID(); } - List partitionValuesFromPath = fileSplit.getPartitionValues() == null - ? BrokerUtil.parseColumnsFromPath(fileSplit.getPathString(), pathPartitionKeys, - false, isACID) : fileSplit.getPartitionValues(); + FilePartitionUtils.ParsedColumnsFromPath partitionValuesFromPath = + fileSplit.getPartitionValues() == null + ? FilePartitionUtils.parseColumnsFromPathWithNullInfo( + fileSplit.getPathString(), pathPartitionKeys, false, isACID) + : FilePartitionUtils.normalizeColumnsFromPath(fileSplit.getPartitionValues()); - TFileRangeDesc rangeDesc = createFileRangeDesc(fileSplit, partitionValuesFromPath, pathPartitionKeys); + TFileRangeDesc rangeDesc = createFileRangeDesc(fileSplit, partitionValuesFromPath.getValues(), + pathPartitionKeys, partitionValuesFromPath.getIsNull()); TFileCompressType fileCompressType = getFileCompressType(fileSplit); rangeDesc.setCompressType(fileCompressType); // Seed connector-specific setup with the scan-level default. A connector may then @@ -491,7 +493,8 @@ private TScanRangeLocations newLocations() { } private TFileRangeDesc createFileRangeDesc(FileSplit fileSplit, List columnsFromPath, - List columnsFromPathKeys) { + List columnsFromPathKeys, + List columnsFromPathIsNull) { TFileRangeDesc rangeDesc = new TFileRangeDesc(); rangeDesc.setStartOffset(fileSplit.getStart()); rangeDesc.setSize(fileSplit.getLength()); @@ -501,6 +504,7 @@ private TFileRangeDesc createFileRangeDesc(FileSplit fileSplit, List col if (!columnsFromPathKeys.isEmpty()) { rangeDesc.setColumnsFromPath(columnsFromPath); rangeDesc.setColumnsFromPathKeys(columnsFromPathKeys); + rangeDesc.setColumnsFromPathIsNull(columnsFromPathIsNull); } rangeDesc.setFileType(fileSplit.getLocationType()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java index 8b07f3e875a5ce..957a2f903e956f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java @@ -29,7 +29,6 @@ import org.apache.doris.common.AnalysisException; import org.apache.doris.common.CacheFactory; import org.apache.doris.common.Config; -import org.apache.doris.common.FeConstants; import org.apache.doris.common.UserException; import org.apache.doris.common.security.authentication.AuthenticationConfig; import org.apache.doris.common.security.authentication.HadoopAuthenticator; @@ -393,10 +392,10 @@ private FileCacheValue loadFiles(FileCacheKey key, DirectoryLister directoryList try { FileCacheValue result = getFileCache(finalLocation, key.inputFormat, key.getPartitionValues(), directoryLister, table); - // Replace default hive partition with a null_string. + // Replace default hive partition with null to distinguish it from a literal "\N". for (int i = 0; i < result.getValuesSize(); i++) { if (HIVE_DEFAULT_PARTITION.equals(result.getPartitionValues().get(i))) { - result.getPartitionValues().set(i, FeConstants.null_string); + result.getPartitionValues().set(i, null); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java index 0b63d4d539af43..fd6979aedc123e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java @@ -31,6 +31,7 @@ import org.apache.doris.common.util.LocationPath; import org.apache.doris.datasource.ExternalTable; import org.apache.doris.datasource.ExternalUtil; +import org.apache.doris.datasource.FilePartitionUtils; import org.apache.doris.datasource.NameMapping; import org.apache.doris.datasource.TableFormatType; import org.apache.doris.datasource.hive.HivePartition; @@ -324,8 +325,11 @@ private void setHudiParams(TFileRangeDesc rangeDesc, HudiSplit hudiSplit) { formPathKeys.add(entry.getKey()); formPathValues.add(entry.getValue()); } + FilePartitionUtils.ParsedColumnsFromPath parsedColumnsFromPath = + FilePartitionUtils.normalizeColumnsFromPath(formPathValues); rangeDesc.setColumnsFromPathKeys(formPathKeys); - rangeDesc.setColumnsFromPath(formPathValues); + rangeDesc.setColumnsFromPath(parsedColumnsFromPath.getValues()); + rangeDesc.setColumnsFromPathIsNull(parsedColumnsFromPath.getIsNull()); } rangeDesc.setTableFormatParams(tableFormatFileDesc); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java index cc8970d3d62566..2984393d0e4bf1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java @@ -131,9 +131,10 @@ import java.time.temporal.TemporalAccessor; import java.util.ArrayList; import java.util.Comparator; -import java.util.HashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.UUID; @@ -638,26 +639,36 @@ public static Type icebergTypeToDorisType(org.apache.iceberg.types.Type type, bo } /** - * Get partition info map for identity partitions only, considering partition - * evolution. - * For non-identity partitions (e.g., day, bucket, truncate), returns null to - * skip - * dynamic partition pruning. - * - * @param partitionData The partition data from the file - * @param partitionSpec The partition spec corresponding to the file's specId - * (required) - * @param timeZone The time zone for timestamp serialization - * @return Map of partition field name to partition value string, or null if - * there are non-identity partitions + * Get identity partition columns that exist in all partition specs. + * The file scanner uses partition columns in the first scan range for all ranges, + * so only common identity partition columns can be used for partition pruning. */ - public static Map getPartitionInfoMap(PartitionData partitionData, PartitionSpec partitionSpec, - String timeZone) { - Map partitionInfoMap = new HashMap<>(); - List fields = partitionData.getPartitionType().asNestedType().fields(); + public static List getCommonIdentityPartitionColumns(Table table) { + LinkedHashSet commonSourceIds = new LinkedHashSet<>(); + for (PartitionField field : table.spec().fields()) { + NestedField sourceField = table.schema().findField(field.sourceId()); + if (field.transform().isIdentity() && sourceField != null + && isSupportedPartitionValueType(sourceField.type().typeId())) { + commonSourceIds.add(field.sourceId()); + } + } + for (PartitionSpec spec : table.specs().values()) { + Set specIdentitySourceIds = spec.fields().stream() + .filter(field -> field.transform().isIdentity()) + .map(PartitionField::sourceId) + .collect(Collectors.toSet()); + commonSourceIds.retainAll(specIdentitySourceIds); + } + return commonSourceIds.stream() + .map(table.schema()::findColumnName) + .filter(Objects::nonNull) + .collect(Collectors.toList()); + } - // Check if all partition fields are identity transform - // If any field is not identity, return null to skip dynamic partition pruning + public static Map getIdentityPartitionInfoMap(PartitionData partitionData, + PartitionSpec partitionSpec, Table table, String timeZone) { + Map partitionInfoMap = Maps.newLinkedHashMap(); + List fields = partitionData.getPartitionType().asNestedType().fields(); List partitionFields = partitionSpec.fields(); Preconditions.checkArgument(fields.size() == partitionFields.size(), "PartitionData fields size does not match PartitionSpec fields size"); @@ -665,32 +676,33 @@ public static Map getPartitionInfoMap(PartitionData partitionDat for (int i = 0; i < fields.size(); i++) { NestedField field = fields.get(i); PartitionField partitionField = partitionFields.get(i); - - // Only process identity transform partitions - // For other transforms (day, bucket, truncate, etc.), skip dynamic partition - // pruning if (!partitionField.transform().isIdentity()) { - if (LOG.isDebugEnabled()) { - LOG.debug( - "Skip dynamic partition pruning for non-identity partition field: {} with transform: {}", - field.name(), partitionField.transform().toString()); - } - return null; + continue; + } + if (!isSupportedPartitionValueType(field.type().typeId())) { + continue; + } + String columnName = table.schema().findColumnName(partitionField.sourceId()); + if (columnName == null) { + continue; } Object value = partitionData.get(i); try { String partitionString = serializePartitionValue(field.type(), value, timeZone); - partitionInfoMap.put(field.name(), partitionString); + partitionInfoMap.put(columnName, partitionString); } catch (UnsupportedOperationException e) { LOG.warn("Failed to serialize Iceberg table partition value for field {}: {}", field.name(), e.getMessage()); - return null; } } return partitionInfoMap; } + private static boolean isSupportedPartitionValueType(TypeID typeId) { + return typeId != TypeID.BINARY && typeId != TypeID.FIXED; + } + private static String serializePartitionValue(org.apache.iceberg.types.Type type, Object value, String timeZone) { switch (type.typeId()) { case BOOLEAN: @@ -703,6 +715,16 @@ private static String serializePartitionValue(org.apache.iceberg.types.Type type return null; } return value.toString(); + case FLOAT: + if (value == null) { + return null; + } + return Float.toString((Float) value); + case DOUBLE: + if (value == null) { + return null; + } + return Double.toString((Double) value); // case binary, fixed should not supported, because if return string with utf8, // the data maybe be corrupted case DATE: diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java index 07423573423a6a..74a535d41078b3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java @@ -25,6 +25,7 @@ import org.apache.doris.catalog.Column; import org.apache.doris.catalog.Env; import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; import org.apache.doris.common.profile.SummaryProfile; import org.apache.doris.common.security.authentication.ExecutionAuthenticator; @@ -131,8 +132,9 @@ public class IcebergScanNode extends FileQueryScanNode { private long countFromSnapshot; private static final long COUNT_WITH_PARALLEL_SPLITS = 10000; private long targetSplitSize = 0; - // This is used to avoid repeatedly calculating partition info map for the same partition data. - private Map> partitionMapInfos; + // Used to avoid repeatedly calculating partition info map for the same + // partition data and spec. + private Map, Map> partitionMapInfos; private boolean isPartitionedTable; private int formatVersion; private ExecutionAuthenticator preExecutionAuthenticator; @@ -299,23 +301,44 @@ private void setIcebergParams(TFileRangeDesc rangeDesc, IcebergSplit icebergSpli } } tableFormatFileDesc.setIcebergParams(fileDesc); - Map partitionValues = icebergSplit.getIcebergPartitionValues(); - if (partitionValues != null) { - List fromPathKeys = new ArrayList<>(); - List fromPathValues = new ArrayList<>(); - List fromPathIsNull = new ArrayList<>(); - for (Map.Entry entry : partitionValues.entrySet()) { - fromPathKeys.add(entry.getKey()); - fromPathValues.add(entry.getValue() != null ? entry.getValue() : ""); - fromPathIsNull.add(entry.getValue() == null); - } - rangeDesc.setColumnsFromPathKeys(fromPathKeys); - rangeDesc.setColumnsFromPath(fromPathValues); - rangeDesc.setColumnsFromPathIsNull(fromPathIsNull); - } + setPartitionValues(rangeDesc, icebergSplit.getIcebergPartitionValues()); rangeDesc.setTableFormatParams(tableFormatFileDesc); } + private List getOrderedPathPartitionKeys() { + if (icebergTable == null) { + return Collections.emptyList(); + } + return IcebergUtils.getCommonIdentityPartitionColumns(icebergTable); + } + + @VisibleForTesting + void setPartitionValues(TFileRangeDesc rangeDesc, Map partitionValues) { + rangeDesc.unsetColumnsFromPathKeys(); + rangeDesc.unsetColumnsFromPath(); + rangeDesc.unsetColumnsFromPathIsNull(); + + List orderedPartitionKeys = getOrderedPathPartitionKeys(); + if (orderedPartitionKeys.isEmpty()) { + return; + } + Preconditions.checkState(partitionValues != null, + "Missing partition values for Iceberg identity-partitioned table"); + + List fromPathValues = new ArrayList<>(orderedPartitionKeys.size()); + List fromPathIsNull = new ArrayList<>(orderedPartitionKeys.size()); + for (String partitionKey : orderedPartitionKeys) { + Preconditions.checkState(partitionValues.containsKey(partitionKey), + "Missing partition value for Iceberg partition key: %s", partitionKey); + String partitionValue = partitionValues.get(partitionKey); + fromPathValues.add(partitionValue == null ? "" : partitionValue); + fromPathIsNull.add(partitionValue == null); + } + rangeDesc.setColumnsFromPathKeys(orderedPartitionKeys); + rangeDesc.setColumnsFromPath(fromPathValues); + rangeDesc.setColumnsFromPathIsNull(fromPathIsNull); + } + @Override protected List getDeleteFiles(TFileRangeDesc rangeDesc) { List deleteFiles = new ArrayList<>(); @@ -771,25 +794,15 @@ private Split createIcebergSplit(FileScanTask fileScanTask) { split.setTargetSplitSize(targetSplitSize); if (isPartitionedTable) { PartitionData partitionData = (PartitionData) fileScanTask.file().partition(); - if (sessionVariable.isEnableRuntimeFilterPartitionPrune()) { - // Get specId and corresponding PartitionSpec to handle partition evolution - int specId = fileScanTask.file().specId(); - PartitionSpec partitionSpec = icebergTable.specs().get(specId); - - Preconditions.checkNotNull(partitionSpec, "Partition spec with specId %s not found for table %s", - specId, icebergTable.name()); - Map partitionInfoMap = partitionMapInfos.computeIfAbsent( - partitionData, k -> { - return IcebergUtils.getPartitionInfoMap(partitionData, partitionSpec, - sessionVariable.getTimeZone()); - }); - // Only set partition values if all partitions are identity transform - // For non-identity partitions, getPartitionInfoMap returns null to skip dynamic partition pruning - if (partitionInfoMap != null) { - split.setIcebergPartitionValues(partitionInfoMap); - } - } else { - partitionMapInfos.put(partitionData, null); + int specId = fileScanTask.file().specId(); + PartitionSpec partitionSpec = icebergTable.specs().get(specId); + Preconditions.checkNotNull(partitionSpec, "Partition spec with specId %s not found for table %s", + specId, icebergTable.name()); + Map partitionInfoMap = partitionMapInfos.computeIfAbsent( + Pair.of(specId, partitionData), k -> IcebergUtils.getIdentityPartitionInfoMap( + partitionData, partitionSpec, icebergTable, sessionVariable.getTimeZone())); + if (!partitionInfoMap.isEmpty()) { + split.setIcebergPartitionValues(partitionInfoMap); } } return split; @@ -947,20 +960,9 @@ public TFileFormatType getFileFormatType() throws UserException { @Override public List getPathPartitionKeys() throws UserException { - // return icebergTable.spec().fields().stream().map(PartitionField::name).map(String::toLowerCase) - // .collect(Collectors.toList()); - /**First, iceberg partition columns are based on existing fields, which will be stored in the actual data file. - * Second, iceberg partition columns support Partition transforms. In this case, the path partition key is not - * equal to the column name of the partition column, so remove this code and get all the columns you want to - * read from the file. - * Related code: - * be/src/vec/exec/scan/vfile_scanner.cpp: - * VFileScanner::_init_expr_ctxes() - * if (slot_info.is_file_slot) { - * xxxx - * } - */ - return new ArrayList<>(); + // Iceberg identity partition columns are still file columns. Per-range partition values are + // only used for partition pruning and as a fallback for files written before partition evolution. + return Collections.emptyList(); } private void recordManifestCacheAccess(boolean cacheHit) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsFileGroupInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsFileGroupInfo.java index b0862277a633ed..b9a2da955a363d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsFileGroupInfo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsFileGroupInfo.java @@ -26,10 +26,10 @@ import org.apache.doris.common.Config; import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; -import org.apache.doris.common.util.BrokerUtil; import org.apache.doris.common.util.Util; import org.apache.doris.datasource.FederationBackendPolicy; import org.apache.doris.datasource.FileGroupInfo; +import org.apache.doris.datasource.FilePartitionUtils; import org.apache.doris.system.Backend; import org.apache.doris.thrift.TBrokerFileStatus; import org.apache.doris.thrift.TExternalScanRange; @@ -279,11 +279,12 @@ public void createScanRangeLocationsUnsplittable(NereidsParamCreateContext conte context.fileGroup.getFileFormatProperties().getCompressionType(), fileStatus.path); context.params.setCompressType(compressType); - List columnsFromPath = BrokerUtil.parseColumnsFromPath(fileStatus.path, - context.fileGroup.getColumnNamesFromPath()); + FilePartitionUtils.ParsedColumnsFromPath columnsFromPath = + FilePartitionUtils.parseColumnsFromPathWithNullInfo(fileStatus.path, + context.fileGroup.getColumnNamesFromPath(), true, false); List columnsFromPathKeys = context.fileGroup.getColumnNamesFromPath(); - TFileRangeDesc rangeDesc = createFileRangeDesc(0, fileStatus, fileStatus.size, columnsFromPath, - columnsFromPathKeys); + TFileRangeDesc rangeDesc = createFileRangeDesc(0, fileStatus, fileStatus.size, + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); locations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); } scanRangeLocations.add(locations); @@ -331,15 +332,16 @@ public void createScanRangeLocationsSplittable(NereidsParamCreateContext context context.fileGroup.getFileFormatProperties().getCompressionType(), fileStatus.path); context.params.setCompressType(compressType); - List columnsFromPath = BrokerUtil.parseColumnsFromPath(fileStatus.path, - context.fileGroup.getColumnNamesFromPath()); + FilePartitionUtils.ParsedColumnsFromPath columnsFromPath = + FilePartitionUtils.parseColumnsFromPathWithNullInfo(fileStatus.path, + context.fileGroup.getColumnNamesFromPath(), true, false); List columnsFromPathKeys = context.fileGroup.getColumnNamesFromPath(); // Assign scan range locations only for broker load. // stream load has only one file, and no need to set multi scan ranges. if (tmpBytes > bytesPerInstance && jobType != FileGroupInfo.JobType.STREAM_LOAD) { long rangeBytes = bytesPerInstance - curInstanceBytes; TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, rangeBytes, - columnsFromPath, columnsFromPathKeys); + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); curFileOffset += rangeBytes; @@ -348,8 +350,8 @@ public void createScanRangeLocationsSplittable(NereidsParamCreateContext context curLocations = newLocations(context.params, brokerDesc, backendPolicy); curInstanceBytes = 0; } else { - TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, leftBytes, columnsFromPath, - columnsFromPathKeys); + TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, leftBytes, + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); curFileOffset = 0; curInstanceBytes += leftBytes; @@ -420,7 +422,7 @@ private TFileFormatType formatType(String fileFormat, String path) throws UserEx } private TFileRangeDesc createFileRangeDesc(long curFileOffset, TBrokerFileStatus fileStatus, long rangeBytes, - List columnsFromPath, List columnsFromPathKeys) { + List columnsFromPath, List columnsFromPathKeys, List columnsFromPathIsNull) { TFileRangeDesc rangeDesc = new TFileRangeDesc(); if (jobType == FileGroupInfo.JobType.BULK_LOAD) { rangeDesc.setPath(fileStatus.path); @@ -429,6 +431,7 @@ private TFileRangeDesc createFileRangeDesc(long curFileOffset, TBrokerFileStatus rangeDesc.setFileSize(fileStatus.size); rangeDesc.setColumnsFromPath(columnsFromPath); rangeDesc.setColumnsFromPathKeys(columnsFromPathKeys); + rangeDesc.setColumnsFromPathIsNull(columnsFromPathIsNull); if (getFileType() == TFileType.FILE_HDFS) { URI fileUri = new Path(fileStatus.path).toUri(); rangeDesc.setFsName(fileUri.getScheme() + "://" + fileUri.getAuthority()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/FilePartitionUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/FilePartitionUtilsTest.java new file mode 100644 index 00000000000000..544431d96e723e --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/FilePartitionUtilsTest.java @@ -0,0 +1,71 @@ +// 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; + +import org.apache.doris.datasource.FilePartitionUtils.ParsedColumnsFromPath; +import org.apache.doris.datasource.hive.HiveMetaStoreCache; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +public class FilePartitionUtilsTest { + + @Test + public void testParseColumnsFromPathPreservesNullMetadataAndInputKeys() throws Exception { + List partitionKeys = new ArrayList<>(Arrays.asList("region", "dt")); + ParsedColumnsFromPath parsed = FilePartitionUtils.parseColumnsFromPathWithNullInfo( + "hdfs://host/table/Region=cn/Dt=" + HiveMetaStoreCache.HIVE_DEFAULT_PARTITION + + "/data.parquet", + partitionKeys, false, false); + + Assert.assertEquals(Arrays.asList("cn", ""), parsed.getValues()); + Assert.assertEquals(Arrays.asList(false, true), parsed.getIsNull()); + Assert.assertEquals(Arrays.asList("region", "dt"), partitionKeys); + } + + @Test + public void testParseColumnsFromPathPreservesLiteralBackslashN() throws Exception { + ParsedColumnsFromPath parsed = FilePartitionUtils.parseColumnsFromPathWithNullInfo( + "hdfs://host/table/p=\\N/data.orc", Arrays.asList("p"), true, false); + + Assert.assertEquals(Arrays.asList("\\N"), parsed.getValues()); + Assert.assertEquals(Arrays.asList(false), parsed.getIsNull()); + } + + @Test + public void testParseAcidColumnsFromPath() throws Exception { + ParsedColumnsFromPath parsed = FilePartitionUtils.parseColumnsFromPathWithNullInfo( + "hdfs://host/table/p=value/delta_1_1/bucket_00000", Arrays.asList("p"), true, true); + + Assert.assertEquals(Arrays.asList("value"), parsed.getValues()); + Assert.assertEquals(Arrays.asList(false), parsed.getIsNull()); + } + + @Test + public void testNormalizeColumnsFromPath() { + ParsedColumnsFromPath parsed = FilePartitionUtils.normalizeColumnsFromPath( + Arrays.asList(null, HiveMetaStoreCache.HIVE_DEFAULT_PARTITION, "", "\\N")); + + Assert.assertEquals(Arrays.asList("", "", "", "\\N"), parsed.getValues()); + Assert.assertEquals(Arrays.asList(true, true, false, false), parsed.getIsNull()); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java index 3330685b85a4b7..69dba111508342 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java @@ -30,6 +30,7 @@ import org.apache.iceberg.ManifestContent; import org.apache.iceberg.ManifestFile; import org.apache.iceberg.ManifestFile.PartitionFieldSummary; +import org.apache.iceberg.PartitionData; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; @@ -56,9 +57,11 @@ import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.Comparator; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -134,6 +137,87 @@ public void testParseSchemaPreservesNonLowercaseColumnNames() { Assert.assertEquals("PART", columns.get(1).getName()); } + @Test + public void testGetCommonIdentityPartitionColumnsUsesSafeIntersection() { + Schema schema = new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.required(2, "Dt", Types.StringType.get()), + Types.NestedField.required(3, "ts", Types.TimestampType.withoutZone())); + PartitionSpec oldSpec = PartitionSpec.builderFor(schema) + .withSpecId(1) + .identity("id") + .identity("Dt") + .build(); + PartitionSpec currentSpec = PartitionSpec.builderFor(schema) + .withSpecId(2) + .identity("Dt") + .day("ts") + .build(); + Map specs = new LinkedHashMap<>(); + specs.put(oldSpec.specId(), oldSpec); + specs.put(currentSpec.specId(), currentSpec); + + Table table = Mockito.mock(Table.class); + Mockito.when(table.schema()).thenReturn(schema); + Mockito.when(table.spec()).thenReturn(currentSpec); + Mockito.when(table.specs()).thenReturn(specs); + + Assert.assertEquals(Arrays.asList("Dt"), + IcebergUtils.getCommonIdentityPartitionColumns(table)); + } + + @Test + public void testGetIdentityPartitionInfoMapReturnsIdentityColumnsOnly() { + Schema schema = new Schema( + Types.NestedField.required(1, "Dt", Types.StringType.get()), + Types.NestedField.required(2, "ts", Types.TimestampType.withoutZone())); + PartitionSpec partitionSpec = PartitionSpec.builderFor(schema) + .identity("Dt") + .day("ts") + .build(); + PartitionData partitionData = new PartitionData(partitionSpec.partitionType()); + partitionData.set(0, "2025-01-01"); + partitionData.set(1, 20000); + + Table table = Mockito.mock(Table.class); + Mockito.when(table.schema()).thenReturn(schema); + + Map partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( + partitionData, partitionSpec, table, "UTC"); + Assert.assertEquals(Collections.singletonMap("Dt", "2025-01-01"), partitionInfoMap); + } + + @Test + public void testGetIdentityPartitionInfoMapSupportsFloatingPointPartitions() { + Schema schema = new Schema( + Types.NestedField.required(1, "float_partition", Types.FloatType.get()), + Types.NestedField.required(2, "double_partition", Types.DoubleType.get())); + PartitionSpec partitionSpec = PartitionSpec.builderFor(schema) + .identity("float_partition") + .identity("double_partition") + .build(); + float floatValue = Math.nextUp(0.1F); + double doubleValue = Math.nextUp(0.1D); + PartitionData partitionData = new PartitionData(partitionSpec.partitionType()); + partitionData.set(0, floatValue); + partitionData.set(1, doubleValue); + + Table table = Mockito.mock(Table.class); + Mockito.when(table.schema()).thenReturn(schema); + + Map partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( + partitionData, partitionSpec, table, "UTC"); + + String serializedFloat = partitionInfoMap.get("float_partition"); + String serializedDouble = partitionInfoMap.get("double_partition"); + Assert.assertEquals(Float.toString(floatValue), serializedFloat); + Assert.assertEquals(Double.toString(doubleValue), serializedDouble); + Assert.assertEquals(Float.floatToIntBits(floatValue), + Float.floatToIntBits(Float.parseFloat(serializedFloat))); + Assert.assertEquals(Double.doubleToLongBits(doubleValue), + Double.doubleToLongBits(Double.parseDouble(serializedDouble))); + } + @Test public void testGetMatchingManifest() { diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java index 8b42f50774cf54..b855646d533289 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java @@ -30,14 +30,23 @@ import org.apache.iceberg.DataFile; import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.types.Types; import org.apache.iceberg.util.ScanTaskUtil; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; +import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; public class IcebergScanNodeTest { private static final long MB = 1024L * 1024L; @@ -95,4 +104,38 @@ public void testSetIcebergParamsUsesSplitFileFormat() throws Exception { Assert.assertEquals(TFileFormatType.FORMAT_ORC, rangeDesc.getFormatType()); } + + @Test + public void testSetPartitionValuesBuildsStableAlignedMetadata() throws Exception { + TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable()); + Schema schema = new Schema( + Types.NestedField.required(1, "Region", Types.StringType.get()), + Types.NestedField.required(2, "Dt", Types.StringType.get())); + PartitionSpec spec = PartitionSpec.builderFor(schema) + .identity("Region") + .identity("Dt") + .build(); + Map specs = new LinkedHashMap<>(); + specs.put(spec.specId(), spec); + Table table = Mockito.mock(Table.class); + Mockito.when(table.schema()).thenReturn(schema); + Mockito.when(table.spec()).thenReturn(spec); + Mockito.when(table.specs()).thenReturn(specs); + + Field icebergTable = IcebergScanNode.class.getDeclaredField("icebergTable"); + icebergTable.setAccessible(true); + icebergTable.set(node, table); + + Assert.assertTrue(node.getPathPartitionKeys().isEmpty()); + + Map partitionValues = new HashMap<>(); + partitionValues.put("Dt", null); + partitionValues.put("Region", "cn"); + TFileRangeDesc rangeDesc = new TFileRangeDesc(); + node.setPartitionValues(rangeDesc, partitionValues); + + Assert.assertEquals(Arrays.asList("Region", "Dt"), rangeDesc.getColumnsFromPathKeys()); + Assert.assertEquals(Arrays.asList("cn", ""), rangeDesc.getColumnsFromPath()); + Assert.assertEquals(Arrays.asList(false, true), rangeDesc.getColumnsFromPathIsNull()); + } } From e890f114809a8852a38af9f3f2e1c339f5dbc670 Mon Sep 17 00:00:00 2001 From: suxiaogang Date: Fri, 7 Aug 2026 14:30:06 +0800 Subject: [PATCH 2/3] [fix](multi-catalog) Handle evolved partition metadata per range --- be/src/vec/exec/scan/file_scanner.cpp | 200 +++++++++--------- be/src/vec/exec/scan/file_scanner.h | 13 +- .../datasource/hudi/source/HudiScanNode.java | 10 +- .../datasource/iceberg/IcebergUtils.java | 36 +--- .../iceberg/source/IcebergScanNode.java | 50 ++--- .../ExternalFileTableValuedFunction.java | 6 +- .../hudi/source/HudiScanNodeTest.java | 46 ++++ .../datasource/iceberg/IcebergUtilsTest.java | 75 +++---- .../iceberg/source/IcebergScanNodeTest.java | 45 ++-- .../ExternalFileTableValuedFunctionTest.java | 8 + 10 files changed, 252 insertions(+), 237 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/datasource/hudi/source/HudiScanNodeTest.java diff --git a/be/src/vec/exec/scan/file_scanner.cpp b/be/src/vec/exec/scan/file_scanner.cpp index 01144f1e55d42e..efa1a666374dcd 100644 --- a/be/src/vec/exec/scan/file_scanner.cpp +++ b/be/src/vec/exec/scan/file_scanner.cpp @@ -210,8 +210,7 @@ Status FileScanner::init(RuntimeState* state, const VExprContextSPtrs& conjuncts bool FileScanner::_check_partition_prune_expr(const VExprSPtr& expr) { if (expr->is_slot_ref()) { auto* slot_ref = static_cast(expr.get()); - return _partition_slot_index_map.find(slot_ref->slot_id()) != - _partition_slot_index_map.end(); + return _partition_slot_ids.contains(slot_ref->slot_id()); } if (expr->is_literal()) { return true; @@ -247,12 +246,6 @@ Status FileScanner::_process_runtime_filters_partition_prune(bool& can_filter_al if (_runtime_filter_partition_prune_ctxs.empty() || _partition_col_descs.empty()) { return Status::OK(); } - if (_partition_col_descs.size() != _partition_slot_descs.size()) { - // This range only carries values for a subset of the partition slots (e.g. a file - // written under an older Iceberg partition spec). The prune conjuncts may reference - // unbound slots whose block columns would stay empty, so skip pruning this range. - return Status::OK(); - } size_t partition_value_column_size = 1; // 1. Get partition key values to string columns. @@ -367,8 +360,7 @@ Status FileScanner::_open_impl(RuntimeState* state) { if (_first_scan_range) { RETURN_IF_ERROR(_init_expr_ctxes()); if (_state->query_options().enable_runtime_filter_partition_prune && - !_partition_slot_index_map.empty()) { - _init_runtime_filter_partition_prune_ctxs(); + (!_is_load || !_partition_slot_index_map.empty())) { _init_runtime_filter_partition_prune_block(); } } else { @@ -603,11 +595,11 @@ Status FileScanner::_cast_to_input_block(Block* block) { } Status FileScanner::_fill_columns_from_path(size_t rows) { - if (!_fill_partition_from_path) { + if (_partition_col_descs_to_fill.empty()) { return Status::OK(); } DataTypeSerDe::FormatOptions _text_formatOptions; - for (auto& kv : _partition_col_descs) { + for (auto& kv : _partition_col_descs_to_fill) { auto doris_column = _src_block_ptr->get_by_position(_src_block_name_to_idx[kv.first]).column; // _src_block_ptr points to a mutable block created by this class itself, so const_cast can be used here. @@ -901,26 +893,20 @@ Status FileScanner::_get_next_reader() { const TFileRangeDesc& range = _current_range; _current_range_path = range.path; - if (!_partition_slot_descs.empty()) { - // we need get partition columns first for runtime filter partition pruning - RETURN_IF_ERROR(_generate_partition_columns()); + // Partition keys may vary between ranges when a table's partition spec evolves. + RETURN_IF_ERROR(_generate_partition_columns()); - if (_state->query_options().enable_runtime_filter_partition_prune) { - // if enable_runtime_filter_partition_prune is true, we need to check whether this range can be filtered out - // by runtime filter partition prune - if (_push_down_conjuncts.size() < _conjuncts.size()) { - // there are new runtime filters, need to re-init runtime filter partition pruning ctxs - _init_runtime_filter_partition_prune_ctxs(); - } + if (_state->query_options().enable_runtime_filter_partition_prune && + !_partition_slot_ids.empty()) { + // Rebuild the contexts because only columns provided by this range can be used + // for partition pruning. + _init_runtime_filter_partition_prune_ctxs(); - bool can_filter_all = false; - RETURN_IF_ERROR(_process_runtime_filters_partition_prune(can_filter_all)); - if (can_filter_all) { - // this range can be filtered out by runtime filter partition pruning - // so we need to skip this range - COUNTER_UPDATE(_runtime_filter_partition_pruned_range_counter, 1); - continue; - } + bool can_filter_all = false; + RETURN_IF_ERROR(_process_runtime_filters_partition_prune(can_filter_all)); + if (can_filter_all) { + COUNTER_UPDATE(_runtime_filter_partition_pruned_range_counter, 1); + continue; } } @@ -1370,6 +1356,7 @@ Status FileScanner::_init_orc_reader(std::unique_ptr&& orc_reader, Status FileScanner::_set_fill_or_truncate_columns(bool need_to_get_parsed_schema) { _missing_cols.clear(); _slot_lower_name_to_col_type.clear(); + _partition_col_descs_to_fill.clear(); std::unordered_map name_to_col_type; RETURN_IF_ERROR(_cur_reader->get_columns(&name_to_col_type, &_missing_cols)); for (const auto& [col_name, col_type] : name_to_col_type) { @@ -1390,23 +1377,28 @@ Status FileScanner::_set_fill_or_truncate_columns(bool need_to_get_parsed_schema _slot_lower_name_to_col_type.emplace(col_name_lower, col_type); } - if (!_fill_partition_from_path && config::enable_iceberg_partition_column_fallback) { - // check if the cols of _partition_col_descs are in _missing_cols - // if so, set _fill_partition_from_path to true and remove the col from _missing_cols - for (const auto& [col_name, col_type] : _partition_col_descs) { - if (_missing_cols.contains(col_name)) { - _fill_partition_from_path = true; + if (_is_load) { + if (_load_fill_partition_from_path) { + _partition_col_descs_to_fill = _partition_col_descs; + } + } else { + for (const auto& [col_name, partition_col_desc] : _partition_col_descs) { + const auto* slot_desc = std::get<1>(partition_col_desc); + if (!_is_file_slot.contains(slot_desc->id())) { + _partition_col_descs_to_fill.emplace(col_name, partition_col_desc); + } else if (config::enable_iceberg_partition_column_fallback && + _missing_cols.contains(col_name)) { + _partition_col_descs_to_fill.emplace(col_name, partition_col_desc); _missing_cols.erase(col_name); } } } RETURN_IF_ERROR(_generate_missing_columns()); - if (_fill_partition_from_path) { - RETURN_IF_ERROR(_cur_reader->set_fill_columns(_partition_col_descs, _missing_col_descs, - _partition_value_is_null)); + if (!_partition_col_descs_to_fill.empty()) { + RETURN_IF_ERROR(_cur_reader->set_fill_columns( + _partition_col_descs_to_fill, _missing_col_descs, _partition_value_is_null)); } else { - // If the partition columns are not from path, we only fill the missing columns. RETURN_IF_ERROR(_cur_reader->set_fill_columns({}, _missing_col_descs)); } if (VLOG_NOTICE_IS_ON && !_missing_cols.empty() && _is_load) { @@ -1540,68 +1532,76 @@ Status FileScanner::read_lines_from_range(const TFileRangeDesc& range, Status FileScanner::_generate_partition_columns() { _partition_col_descs.clear(); _partition_value_is_null.clear(); + _partition_slot_ids.clear(); const TFileRangeDesc& range = _current_range; - if (!range.__isset.columns_from_path || _partition_slot_descs.empty()) { + if (!range.__isset.columns_from_path) { return Status::OK(); } - if (range.__isset.columns_from_path_is_null) { - DORIS_CHECK(range.columns_from_path_is_null.size() == range.columns_from_path.size()); + if (range.__isset.columns_from_path_is_null && + range.columns_from_path_is_null.size() != range.columns_from_path.size()) { + return Status::InternalError("Partition null marker count {} does not match value count {}", + range.columns_from_path_is_null.size(), + range.columns_from_path.size()); } - if (!_is_load && range.__isset.columns_from_path_keys) { - // Ranges of one scanner may carry different columns_from_path_keys. E.g. after - // Iceberg partition evolution, a file written under an older partition spec only - // carries the identity partition values of that spec (possibly none), while - // _partition_slot_descs was built from the first range's keys. So bind values by - // this range's own key list. A partition slot absent from this range's keys is - // read from the data file instead (for such tables all partition columns are - // also file slots, see IcebergScanNode.getPathPartitionKeys). - if (range.columns_from_path.size() != range.columns_from_path_keys.size()) { - return Status::InternalError( - "columns_from_path size {} does not match columns_from_path_keys size {}", - range.columns_from_path.size(), range.columns_from_path_keys.size()); - } - std::unordered_map name_to_value_index; - for (size_t i = 0; i < range.columns_from_path_keys.size(); ++i) { - name_to_value_index.emplace(range.columns_from_path_keys[i], i); - } + + if (_is_load) { for (const auto& slot_desc : _partition_slot_descs) { - if (!slot_desc) { - continue; - } - auto it = name_to_value_index.find(slot_desc->col_name()); - if (it == name_to_value_index.end()) { - continue; - } - _partition_col_descs.emplace( - slot_desc->col_name(), - std::make_tuple(range.columns_from_path[it->second], slot_desc)); - if (range.__isset.columns_from_path_is_null) { - _partition_value_is_null.emplace(slot_desc->col_name(), - range.columns_from_path_is_null[it->second]); + if (slot_desc) { + auto it = _partition_slot_index_map.find(slot_desc->id()); + if (it == std::end(_partition_slot_index_map)) { + return Status::InternalError("Unknown source slot descriptor, slot_id={}", + slot_desc->id()); + } + if (it->second < 0 || + static_cast(it->second) >= range.columns_from_path.size()) { + return Status::InternalError( + "Invalid partition value index {}, value count {} for column {}", + it->second, range.columns_from_path.size(), slot_desc->col_name()); + } + const std::string& column_from_path = range.columns_from_path[it->second]; + _partition_col_descs.emplace(slot_desc->col_name(), + std::make_tuple(column_from_path, slot_desc)); + _partition_slot_ids.emplace(slot_desc->id()); + if (range.__isset.columns_from_path_is_null) { + _partition_value_is_null.emplace(slot_desc->col_name(), + range.columns_from_path_is_null[it->second]); + } } } return Status::OK(); } - for (const auto& slot_desc : _partition_slot_descs) { - if (slot_desc) { - auto it = _partition_slot_index_map.find(slot_desc->id()); - if (it == std::end(_partition_slot_index_map)) { - return Status::InternalError("Unknown source slot descriptor, slot_id={}", - slot_desc->id()); - } - if (it->second < 0 || - static_cast(it->second) >= range.columns_from_path.size()) { - return Status::InternalError( - "Invalid partition value index {}, value count {} for column {}", - it->second, range.columns_from_path.size(), slot_desc->col_name()); - } - const std::string& column_from_path = range.columns_from_path[it->second]; - _partition_col_descs.emplace(slot_desc->col_name(), - std::make_tuple(column_from_path, slot_desc)); - if (range.__isset.columns_from_path_is_null) { - _partition_value_is_null.emplace(slot_desc->col_name(), - range.columns_from_path_is_null[it->second]); - } + + if (!range.__isset.columns_from_path_keys) { + return Status::OK(); + } + if (range.columns_from_path_keys.size() != range.columns_from_path.size()) { + return Status::InternalError("Partition key count {} does not match value count {}", + range.columns_from_path_keys.size(), + range.columns_from_path.size()); + } + + std::unordered_map partition_name_to_index; + for (size_t i = 0; i < range.columns_from_path_keys.size(); ++i) { + partition_name_to_index.emplace(range.columns_from_path_keys[i], i); + } + for (const auto& slot_info : _params->required_slots) { + auto* slot_desc = _state->desc_tbl().get_slot_descriptor(slot_info.slot_id); + if (slot_desc == nullptr) { + return Status::InternalError("Unknown source slot descriptor, slot_id={}", + slot_info.slot_id); + } + auto index_it = partition_name_to_index.find(slot_desc->col_name()); + if (index_it == partition_name_to_index.end()) { + continue; + } + size_t value_index = index_it->second; + _partition_col_descs.emplace( + slot_desc->col_name(), + std::make_tuple(range.columns_from_path[value_index], slot_desc)); + _partition_slot_ids.emplace(slot_desc->id()); + if (range.__isset.columns_from_path_is_null) { + _partition_value_is_null.emplace(slot_desc->col_name(), + range.columns_from_path_is_null[value_index]); } } return Status::OK(); @@ -1636,13 +1636,8 @@ Status FileScanner::_init_expr_ctxes() { full_src_index_map.emplace(slot_desc->id(), index++); } - // For external table query, find the index of column in path. - // Because query doesn't always search for all columns in a table - // and the order of selected columns is random. - // All ranges in _ranges vector should have identical columns_from_path_keys - // because they are all file splits for the same external table. - // So here use the first element of _ranges to fill the partition_name_to_key_index_map - if (_current_range.__isset.columns_from_path_keys) { + // Load tasks do not always read all source columns, and the selected column order may vary. + if (_is_load && _current_range.__isset.columns_from_path_keys) { std::vector key_map = _current_range.columns_from_path_keys; if (!key_map.empty()) { for (size_t i = 0; i < key_map.size(); i++) { @@ -1675,9 +1670,8 @@ Status FileScanner::_init_expr_ctxes() { if (slot_info.is_file_slot) { // If there is slot which is both a partition column and a file column, // we should not fill the partition column from path. - _fill_partition_from_path = false; - } else if (!_fill_partition_from_path) { - // This should not happen + _load_fill_partition_from_path = false; + } else if (!_load_fill_partition_from_path) { return Status::InternalError( "Partition column {} is not a file column, but there is already a column " "which is both a partition column and a file column.", diff --git a/be/src/vec/exec/scan/file_scanner.h b/be/src/vec/exec/scan/file_scanner.h index 4ca982e0bea137..b39ba1c8fefbc8 100644 --- a/be/src/vec/exec/scan/file_scanner.h +++ b/be/src/vec/exec/scan/file_scanner.h @@ -131,10 +131,12 @@ class FileScanner : public Scanner { // col names from _file_slot_descs std::vector _file_col_names; - // Partition source slot descriptors + // Partition source slot descriptors used by load tasks. std::vector _partition_slot_descs; - // Partition slot id to index in _partition_slot_descs + // Partition slot id to value index used by load tasks. std::unordered_map _partition_slot_index_map; + // Partition slot ids provided by the current query range. + std::unordered_set _partition_slot_ids; // created from param.expr_of_dest_slot // For query, it saves default value expr of all dest columns, or nullptr for NULL. // For load, it saves conversion expr/default value of all dest columns. @@ -189,10 +191,13 @@ class FileScanner : public Scanner { std::unique_ptr _file_reader_stats; std::unique_ptr _io_ctx; - // Whether to fill partition columns from path, default is true. - bool _fill_partition_from_path = true; + // Whether load tasks should fill partition columns from the path. + bool _load_fill_partition_from_path = true; std::unordered_map> _partition_col_descs; + // Partition columns that should be filled for the current reader. + std::unordered_map> + _partition_col_descs_to_fill; std::unordered_map _partition_value_is_null; std::unordered_map _missing_col_descs; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java index fd6979aedc123e..b3b6eec5ad282d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java @@ -24,6 +24,7 @@ import org.apache.doris.catalog.Env; import org.apache.doris.catalog.ListPartitionItem; import org.apache.doris.catalog.PartitionItem; +import org.apache.doris.catalog.PartitionKey; import org.apache.doris.catalog.Type; import org.apache.doris.common.AnalysisException; import org.apache.doris.common.UserException; @@ -355,7 +356,7 @@ private List getPrunedPartitions(HoodieTableMetaClient metaClient String path = basePath + "/" + key; hivePartitions.add(new HivePartition( nameMapping, false, inputFormat, path, - ((ListPartitionItem) value).getItems().get(0).getPartitionValuesAsStringList(), + getPartitionValues((ListPartitionItem) value), Maps.newHashMap())); } ); @@ -372,6 +373,13 @@ private List getPrunedPartitions(HoodieTableMetaClient metaClient return Lists.newArrayList(dummyPartition); } + static List getPartitionValues(ListPartitionItem partitionItem) { + PartitionKey partitionKey = partitionItem.getItems().get(0); + return partitionKey.getKeys().stream() + .map(key -> key.isNullLiteral() ? null : key.getStringValue()) + .collect(Collectors.toList()); + } + private List getIncrementalSplits() { if (canUseNativeReader()) { List splits = incrementalRelation.collectSplits(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java index 2984393d0e4bf1..5d41e400c0e88a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java @@ -131,10 +131,8 @@ import java.time.temporal.TemporalAccessor; import java.util.ArrayList; import java.util.Comparator; -import java.util.LinkedHashSet; import java.util.List; import java.util.Map; -import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.UUID; @@ -638,35 +636,8 @@ public static Type icebergTypeToDorisType(org.apache.iceberg.types.Type type, bo } } - /** - * Get identity partition columns that exist in all partition specs. - * The file scanner uses partition columns in the first scan range for all ranges, - * so only common identity partition columns can be used for partition pruning. - */ - public static List getCommonIdentityPartitionColumns(Table table) { - LinkedHashSet commonSourceIds = new LinkedHashSet<>(); - for (PartitionField field : table.spec().fields()) { - NestedField sourceField = table.schema().findField(field.sourceId()); - if (field.transform().isIdentity() && sourceField != null - && isSupportedPartitionValueType(sourceField.type().typeId())) { - commonSourceIds.add(field.sourceId()); - } - } - for (PartitionSpec spec : table.specs().values()) { - Set specIdentitySourceIds = spec.fields().stream() - .filter(field -> field.transform().isIdentity()) - .map(PartitionField::sourceId) - .collect(Collectors.toSet()); - commonSourceIds.retainAll(specIdentitySourceIds); - } - return commonSourceIds.stream() - .map(table.schema()::findColumnName) - .filter(Objects::nonNull) - .collect(Collectors.toList()); - } - public static Map getIdentityPartitionInfoMap(PartitionData partitionData, - PartitionSpec partitionSpec, Table table, String timeZone) { + PartitionSpec partitionSpec, Schema querySchema, String timeZone) { Map partitionInfoMap = Maps.newLinkedHashMap(); List fields = partitionData.getPartitionType().asNestedType().fields(); List partitionFields = partitionSpec.fields(); @@ -682,7 +653,7 @@ public static Map getIdentityPartitionInfoMap(PartitionData part if (!isSupportedPartitionValueType(field.type().typeId())) { continue; } - String columnName = table.schema().findColumnName(partitionField.sourceId()); + String columnName = querySchema.findColumnName(partitionField.sourceId()); if (columnName == null) { continue; } @@ -751,7 +722,8 @@ private static String serializePartitionValue(org.apache.iceberg.types.Type type long timestampMicros = (Long) value; TimestampType timestampType = (TimestampType) type; LocalDateTime timestamp = LocalDateTime.ofEpochSecond( - timestampMicros / 1_000_000, (int) (timestampMicros % 1_000_000) * 1000, + Math.floorDiv(timestampMicros, 1_000_000L), + (int) Math.floorMod(timestampMicros, 1_000_000L) * 1000, ZoneOffset.UTC); // type is timestamptz if timestampType.shouldAdjustToUTC() is true if (timestampType.shouldAdjustToUTC()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java index 74a535d41078b3..55dd03c35f587f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java @@ -79,6 +79,7 @@ import org.apache.iceberg.PartitionData; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.PartitionSpecParser; +import org.apache.iceberg.Schema; import org.apache.iceberg.SchemaParser; import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; @@ -121,6 +122,7 @@ public class IcebergScanNode extends FileQueryScanNode { private IcebergSource source; private Table icebergTable; + private Schema querySchema; private List pushdownIcebergPredicates = Lists.newArrayList(); // If tableLevelPushDownCount is true, means we can do count push down opt at table level. // which means all splits have no position/equality delete files, @@ -135,7 +137,6 @@ public class IcebergScanNode extends FileQueryScanNode { // Used to avoid repeatedly calculating partition info map for the same // partition data and spec. private Map, Map> partitionMapInfos; - private boolean isPartitionedTable; private int formatVersion; private ExecutionAuthenticator preExecutionAuthenticator; private TableScan icebergTableScan; @@ -203,8 +204,12 @@ public IcebergScanNode(PlanNodeId id, TupleDescriptor desc, boolean needCheckCol @Override protected void doInitialize() throws UserException { icebergTable = source.getIcebergTable(); + IcebergTableQueryInfo queryInfo = getSpecifiedSnapshot(); + querySchema = queryInfo == null ? icebergTable.schema() + : Preconditions.checkNotNull(icebergTable.schemas().get(queryInfo.getSchemaId()), + "Schema with schemaId %s not found for table %s", + queryInfo.getSchemaId(), icebergTable.name()); partitionMapInfos = new HashMap<>(); - isPartitionedTable = icebergTable.spec().isPartitioned(); formatVersion = ((BaseTable) icebergTable).operations().current().formatVersion(); preExecutionAuthenticator = source.getCatalog().getExecutionAuthenticator(); storagePropertiesMap = VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials( @@ -305,36 +310,25 @@ private void setIcebergParams(TFileRangeDesc rangeDesc, IcebergSplit icebergSpli rangeDesc.setTableFormatParams(tableFormatFileDesc); } - private List getOrderedPathPartitionKeys() { - if (icebergTable == null) { - return Collections.emptyList(); - } - return IcebergUtils.getCommonIdentityPartitionColumns(icebergTable); - } - @VisibleForTesting void setPartitionValues(TFileRangeDesc rangeDesc, Map partitionValues) { rangeDesc.unsetColumnsFromPathKeys(); rangeDesc.unsetColumnsFromPath(); rangeDesc.unsetColumnsFromPathIsNull(); - List orderedPartitionKeys = getOrderedPathPartitionKeys(); - if (orderedPartitionKeys.isEmpty()) { + if (partitionValues == null || partitionValues.isEmpty()) { return; } - Preconditions.checkState(partitionValues != null, - "Missing partition values for Iceberg identity-partitioned table"); - - List fromPathValues = new ArrayList<>(orderedPartitionKeys.size()); - List fromPathIsNull = new ArrayList<>(orderedPartitionKeys.size()); - for (String partitionKey : orderedPartitionKeys) { - Preconditions.checkState(partitionValues.containsKey(partitionKey), - "Missing partition value for Iceberg partition key: %s", partitionKey); - String partitionValue = partitionValues.get(partitionKey); + + List fromPathKeys = new ArrayList<>(partitionValues.size()); + List fromPathValues = new ArrayList<>(partitionValues.size()); + List fromPathIsNull = new ArrayList<>(partitionValues.size()); + partitionValues.forEach((partitionKey, partitionValue) -> { + fromPathKeys.add(partitionKey); fromPathValues.add(partitionValue == null ? "" : partitionValue); fromPathIsNull.add(partitionValue == null); - } - rangeDesc.setColumnsFromPathKeys(orderedPartitionKeys); + }); + rangeDesc.setColumnsFromPathKeys(fromPathKeys); rangeDesc.setColumnsFromPath(fromPathValues); rangeDesc.setColumnsFromPathIsNull(fromPathIsNull); } @@ -792,15 +786,15 @@ private Split createIcebergSplit(FileScanTask fileScanTask) { } split.setTableFormatType(TableFormatType.ICEBERG); split.setTargetSplitSize(targetSplitSize); - if (isPartitionedTable) { + int specId = fileScanTask.file().specId(); + PartitionSpec partitionSpec = icebergTable.specs().get(specId); + Preconditions.checkNotNull(partitionSpec, "Partition spec with specId %s not found for table %s", + specId, icebergTable.name()); + if (partitionSpec.isPartitioned()) { PartitionData partitionData = (PartitionData) fileScanTask.file().partition(); - int specId = fileScanTask.file().specId(); - PartitionSpec partitionSpec = icebergTable.specs().get(specId); - Preconditions.checkNotNull(partitionSpec, "Partition spec with specId %s not found for table %s", - specId, icebergTable.name()); Map partitionInfoMap = partitionMapInfos.computeIfAbsent( Pair.of(specId, partitionData), k -> IcebergUtils.getIdentityPartitionInfoMap( - partitionData, partitionSpec, icebergTable, sessionVariable.getTimeZone())); + partitionData, partitionSpec, querySchema, sessionVariable.getTimeZone())); if (!partitionInfoMap.isEmpty()) { split.setIcebergPartitionValues(partitionInfoMap); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java index e08eaa2c8256fb..c71512896a1af1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java @@ -424,10 +424,14 @@ private void fillColumns(InternalService.PFetchTableSchemaResult result) { // HACK(tsy): path columns are all treated as STRING type now, after BE supports reading all columns // types by all format readers from file meta, maybe reading path columns types from BE then. for (String colName : pathPartitionKeys) { - columns.add(new Column(colName, ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH), false)); + columns.add(createPathPartitionColumn(colName)); } } + static Column createPathPartitionColumn(String colName) { + return new Column(colName, ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH), true); + } + private PFetchTableSchemaRequest getFetchTableStructureRequest() throws TException { // set TFileScanRangeParams TFileScanRangeParams fileScanRangeParams = new TFileScanRangeParams(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/hudi/source/HudiScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/hudi/source/HudiScanNodeTest.java new file mode 100644 index 00000000000000..6befa82f418cf2 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/hudi/source/HudiScanNodeTest.java @@ -0,0 +1,46 @@ +// 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.hudi.source; + +import org.apache.doris.analysis.PartitionValue; +import org.apache.doris.catalog.ListPartitionItem; +import org.apache.doris.catalog.PartitionKey; +import org.apache.doris.catalog.ScalarType; +import org.apache.doris.catalog.Type; +import org.apache.doris.common.AnalysisException; + +import com.google.common.collect.ImmutableList; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Arrays; +import java.util.List; + +public class HudiScanNodeTest { + @Test + public void testGetPartitionValuesPreservesNullLiteral() throws AnalysisException { + List values = Arrays.asList( + new PartitionValue("__HIVE_DEFAULT_PARTITION__", true), + new PartitionValue("NULL")); + List types = Arrays.asList(ScalarType.STRING, ScalarType.STRING); + PartitionKey key = PartitionKey.createListPartitionKeyWithTypes(values, types, false); + ListPartitionItem item = new ListPartitionItem(ImmutableList.of(key)); + + Assert.assertEquals(Arrays.asList(null, "NULL"), HudiScanNode.getPartitionValues(item)); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java index 69dba111508342..6e0f448aa903a1 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java @@ -137,35 +137,6 @@ public void testParseSchemaPreservesNonLowercaseColumnNames() { Assert.assertEquals("PART", columns.get(1).getName()); } - @Test - public void testGetCommonIdentityPartitionColumnsUsesSafeIntersection() { - Schema schema = new Schema( - Types.NestedField.required(1, "id", Types.IntegerType.get()), - Types.NestedField.required(2, "Dt", Types.StringType.get()), - Types.NestedField.required(3, "ts", Types.TimestampType.withoutZone())); - PartitionSpec oldSpec = PartitionSpec.builderFor(schema) - .withSpecId(1) - .identity("id") - .identity("Dt") - .build(); - PartitionSpec currentSpec = PartitionSpec.builderFor(schema) - .withSpecId(2) - .identity("Dt") - .day("ts") - .build(); - Map specs = new LinkedHashMap<>(); - specs.put(oldSpec.specId(), oldSpec); - specs.put(currentSpec.specId(), currentSpec); - - Table table = Mockito.mock(Table.class); - Mockito.when(table.schema()).thenReturn(schema); - Mockito.when(table.spec()).thenReturn(currentSpec); - Mockito.when(table.specs()).thenReturn(specs); - - Assert.assertEquals(Arrays.asList("Dt"), - IcebergUtils.getCommonIdentityPartitionColumns(table)); - } - @Test public void testGetIdentityPartitionInfoMapReturnsIdentityColumnsOnly() { Schema schema = new Schema( @@ -179,11 +150,8 @@ public void testGetIdentityPartitionInfoMapReturnsIdentityColumnsOnly() { partitionData.set(0, "2025-01-01"); partitionData.set(1, 20000); - Table table = Mockito.mock(Table.class); - Mockito.when(table.schema()).thenReturn(schema); - Map partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( - partitionData, partitionSpec, table, "UTC"); + partitionData, partitionSpec, schema, "UTC"); Assert.assertEquals(Collections.singletonMap("Dt", "2025-01-01"), partitionInfoMap); } @@ -202,11 +170,8 @@ public void testGetIdentityPartitionInfoMapSupportsFloatingPointPartitions() { partitionData.set(0, floatValue); partitionData.set(1, doubleValue); - Table table = Mockito.mock(Table.class); - Mockito.when(table.schema()).thenReturn(schema); - Map partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( - partitionData, partitionSpec, table, "UTC"); + partitionData, partitionSpec, schema, "UTC"); String serializedFloat = partitionInfoMap.get("float_partition"); String serializedDouble = partitionInfoMap.get("double_partition"); @@ -218,6 +183,42 @@ public void testGetIdentityPartitionInfoMapSupportsFloatingPointPartitions() { Double.doubleToLongBits(Double.parseDouble(serializedDouble))); } + @Test + public void testGetIdentityPartitionInfoMapUsesQuerySchemaName() { + Schema specSchema = new Schema( + Types.NestedField.required(1, "old_name", Types.StringType.get())); + PartitionSpec partitionSpec = PartitionSpec.builderFor(specSchema) + .identity("old_name") + .build(); + PartitionData partitionData = new PartitionData(partitionSpec.partitionType()); + partitionData.set(0, "value"); + + Map partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( + partitionData, partitionSpec, specSchema, "UTC"); + + Assert.assertEquals(Collections.singletonMap("old_name", "value"), partitionInfoMap); + } + + @Test + public void testGetIdentityPartitionInfoMapSupportsNegativeTimestampMicros() { + Schema schema = new Schema( + Types.NestedField.required(1, "local_ts", Types.TimestampType.withoutZone()), + Types.NestedField.required(2, "utc_ts", Types.TimestampType.withZone())); + PartitionSpec partitionSpec = PartitionSpec.builderFor(schema) + .identity("local_ts") + .identity("utc_ts") + .build(); + PartitionData partitionData = new PartitionData(partitionSpec.partitionType()); + partitionData.set(0, -1L); + partitionData.set(1, -1L); + + Map partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( + partitionData, partitionSpec, schema, "Asia/Shanghai"); + + Assert.assertEquals("1969-12-31T23:59:59.999999", partitionInfoMap.get("local_ts")); + Assert.assertEquals("1970-01-01T07:59:59.999999", partitionInfoMap.get("utc_ts")); + } + @Test public void testGetMatchingManifest() { diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java index b855646d533289..fd8e8643344aea 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java @@ -30,21 +30,15 @@ import org.apache.iceberg.DataFile; import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileScanTask; -import org.apache.iceberg.PartitionSpec; -import org.apache.iceberg.Schema; -import org.apache.iceberg.Table; -import org.apache.iceberg.types.Types; import org.apache.iceberg.util.ScanTaskUtil; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; -import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; @@ -106,36 +100,25 @@ public void testSetIcebergParamsUsesSplitFileFormat() throws Exception { } @Test - public void testSetPartitionValuesBuildsStableAlignedMetadata() throws Exception { + public void testSetPartitionValuesBuildsPerRangeAlignedMetadata() throws Exception { TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable()); - Schema schema = new Schema( - Types.NestedField.required(1, "Region", Types.StringType.get()), - Types.NestedField.required(2, "Dt", Types.StringType.get())); - PartitionSpec spec = PartitionSpec.builderFor(schema) - .identity("Region") - .identity("Dt") - .build(); - Map specs = new LinkedHashMap<>(); - specs.put(spec.specId(), spec); - Table table = Mockito.mock(Table.class); - Mockito.when(table.schema()).thenReturn(schema); - Mockito.when(table.spec()).thenReturn(spec); - Mockito.when(table.specs()).thenReturn(specs); - - Field icebergTable = IcebergScanNode.class.getDeclaredField("icebergTable"); - icebergTable.setAccessible(true); - icebergTable.set(node, table); - Assert.assertTrue(node.getPathPartitionKeys().isEmpty()); - Map partitionValues = new HashMap<>(); + Map partitionValues = new LinkedHashMap<>(); partitionValues.put("Dt", null); partitionValues.put("Region", "cn"); - TFileRangeDesc rangeDesc = new TFileRangeDesc(); - node.setPartitionValues(rangeDesc, partitionValues); + TFileRangeDesc oldSpecRange = new TFileRangeDesc(); + node.setPartitionValues(oldSpecRange, partitionValues); + + Assert.assertEquals(Arrays.asList("Dt", "Region"), oldSpecRange.getColumnsFromPathKeys()); + Assert.assertEquals(Arrays.asList("", "cn"), oldSpecRange.getColumnsFromPath()); + Assert.assertEquals(Arrays.asList(true, false), oldSpecRange.getColumnsFromPathIsNull()); + + TFileRangeDesc newSpecRange = new TFileRangeDesc(); + node.setPartitionValues(newSpecRange, Collections.singletonMap("Region", "us")); - Assert.assertEquals(Arrays.asList("Region", "Dt"), rangeDesc.getColumnsFromPathKeys()); - Assert.assertEquals(Arrays.asList("cn", ""), rangeDesc.getColumnsFromPath()); - Assert.assertEquals(Arrays.asList(false, true), rangeDesc.getColumnsFromPathIsNull()); + Assert.assertEquals(Collections.singletonList("Region"), newSpecRange.getColumnsFromPathKeys()); + Assert.assertEquals(Collections.singletonList("us"), newSpecRange.getColumnsFromPath()); + Assert.assertEquals(Collections.singletonList(false), newSpecRange.getColumnsFromPathIsNull()); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java index e5b06bd5dd4101..e0ead50ed13450 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java @@ -33,6 +33,14 @@ import java.util.Map; public class ExternalFileTableValuedFunctionTest { + @Test + public void testPathPartitionColumnIsNullable() { + Column column = ExternalFileTableValuedFunction.createPathPartitionColumn("part"); + + Assert.assertTrue(column.isAllowNull()); + Assert.assertEquals(PrimitiveType.VARCHAR, column.getType().getPrimitiveType()); + } + @Test public void testCsvSchemaParse() { Config.enable_date_conversion = true; From 4df0f8e340c3b21a2724a3248c9fd715d3c88a55 Mon Sep 17 00:00:00 2001 From: suxiaogang Date: Fri, 7 Aug 2026 15:15:26 +0800 Subject: [PATCH 3/3] [fix](multi-catalog) Remove unused Iceberg test imports --- .../org/apache/doris/datasource/iceberg/IcebergUtilsTest.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java index 6e0f448aa903a1..81e935fc7d9064 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java @@ -57,11 +57,9 @@ import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.Comparator; import java.util.HashMap; -import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional;