This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 50e6d56c642 branch-4.1: [fix](multi-catalog) Preserve external
partition metadata (#66012)
50e6d56c642 is described below
commit 50e6d56c642609d140f945913b5d2c7692b8e74c
Author: Socrates <[email protected]>
AuthorDate: Wed Aug 5 07:45:04 2026 +0800
branch-4.1: [fix](multi-catalog) Preserve external partition metadata
(#66012)
### What problem does this PR solve?
Issue Number: None
Related PR: #62821, #65581, #66010
Problem Summary: branch-4.1 already backported the Paimon and BE
partition metadata changes through #65581, but Hive-style scans, Hudi,
Iceberg, and file load paths still had incomplete FE metadata
propagation. Hive default partitions were encoded as `\N`, which lost
the distinction between NULL and a literal `\N`. Iceberg attached
partition metadata only when runtime partition pruning was enabled, used
unstable per-map ordering, and did not safely handle files written with
different partition specs.
This change:
- propagates aligned partition values and explicit NULL flags through
Hive-style scans, Hudi, and both legacy and Nereids file load paths;
- preserves a literal `\N` as data while representing Hive default
partitions as NULL;
- sends stable Iceberg identity-partition metadata for every split,
keyed by both spec ID and partition data;
- restricts Iceberg path partition keys to identity columns shared by
all partition specs, which is safe for both FileScannerV2 and the legacy
scanner's scan-level slot mapping.
### Release note
Fix partition-column materialization, NULL handling, and Iceberg
partition evolution compatibility for external table scans and file
loads on branch-4.1.
### Check List (For Author)
- Test: Unit Test
- `./run-fe-ut.sh --run
org.apache.doris.common.util.BrokerUtilTest,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
---
.../org/apache/doris/common/util/BrokerUtil.java | 71 ++++++---
.../org/apache/doris/datasource/FileGroupInfo.java | 24 +--
.../apache/doris/datasource/FileQueryScanNode.java | 8 +-
.../datasource/hive/HiveExternalMetaCache.java | 4 +-
.../doris/datasource/hudi/source/HudiScanNode.java | 6 +-
.../doris/datasource/iceberg/IcebergUtils.java | 66 +++++++-
.../datasource/iceberg/source/IcebergScanNode.java | 116 ++++++++++----
.../doris/nereids/load/NereidsFileGroupInfo.java | 24 +--
.../apache/doris/common/util/BrokerUtilTest.java | 37 ++++-
.../doris/datasource/iceberg/IcebergUtilsTest.java | 59 +++++++
.../iceberg/source/IcebergScanNodeTest.java | 171 ++++++++++++++++++++-
11 files changed, 500 insertions(+), 86 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/util/BrokerUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/common/util/BrokerUtil.java
index 76baa5fec5b..fd49d44edda 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/BrokerUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/BrokerUtil.java
@@ -69,6 +69,7 @@ import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Locale;
public class BrokerUtil {
private static final Logger LOG = LogManager.getLogger(BrokerUtil.class);
@@ -155,7 +156,7 @@ public class BrokerUtil {
public static List<String> parseColumnsFromPath(String filePath,
List<String> columnsFromPath)
throws UserException {
- return parseColumnsFromPath(filePath, columnsFromPath, true, false);
+ return parseColumnsFromPathWithNullInfo(filePath, columnsFromPath,
true, false).getValues();
}
public static List<String> parseColumnsFromPath(
@@ -164,23 +165,35 @@ public class BrokerUtil {
boolean caseSensitive,
boolean isACID)
throws UserException {
+ return parseColumnsFromPathWithNullInfo(filePath, columnsFromPath,
caseSensitive, isACID)
+ .getValues();
+ }
+
+ public static ParsedColumnsFromPath parseColumnsFromPathWithNullInfo(
+ String filePath,
+ List<String> columnsFromPath,
+ boolean caseSensitive,
+ boolean isACID)
+ throws UserException {
if (columnsFromPath == null || columnsFromPath.isEmpty()) {
- return Collections.emptyList();
+ return new ParsedColumnsFromPath(Collections.emptyList(),
Collections.emptyList());
}
// if it is ACID, the path count is 3. The hdfs path is
hdfs://xxx/table_name/par=xxx/delta(or base)_xxx/.
int pathCount = isACID ? 3 : 2;
+ List<String> expectedColumns = columnsFromPath;
if (!caseSensitive) {
- for (int i = 0; i < columnsFromPath.size(); i++) {
- String path = columnsFromPath.remove(i);
- columnsFromPath.add(i, path.toLowerCase());
+ 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: "
- + columnsFromPath + ", filePath: " + filePath);
+ + expectedColumns + ", filePath: " + filePath);
}
- String[] columns = new String[columnsFromPath.size()];
+ String[] columns = new String[expectedColumns.size()];
+ Boolean[] columnsFromPathIsNull = new Boolean[expectedColumns.size()];
int size = 0;
boolean skipOnce = true;
for (int i = strings.length - pathCount; i >= 0; i--) {
@@ -194,31 +207,52 @@ public class BrokerUtil {
continue;
}
throw new UserException("Fail to parse columnsFromPath,
expected: "
- + columnsFromPath + ", filePath: " + filePath);
+ + expectedColumns + ", filePath: " + filePath);
}
skipOnce = false;
String[] pair = str.split("=", 2);
if (pair.length != 2) {
throw new UserException("Fail to parse columnsFromPath,
expected: "
- + columnsFromPath + ", filePath: " + filePath);
+ + expectedColumns + ", filePath: " + filePath);
}
- String parsedColumnName = caseSensitive ? pair[0] :
pair[0].toLowerCase();
- int index = columnsFromPath.indexOf(parsedColumnName);
+ String parsedColumnName = caseSensitive ? pair[0] :
pair[0].toLowerCase(Locale.ROOT);
+ int index = expectedColumns.indexOf(parsedColumnName);
if (index == -1) {
continue;
}
- columns[index] =
HiveExternalMetaCache.HIVE_DEFAULT_PARTITION.equals(pair[1])
- ? FeConstants.null_string : pair[1];
+ boolean isNull =
HiveExternalMetaCache.HIVE_DEFAULT_PARTITION.equals(pair[1]);
+ columns[index] = isNull ? "" : pair[1];
+ columnsFromPathIsNull[index] = isNull;
size++;
- if (size >= columnsFromPath.size()) {
+ if (size >= expectedColumns.size()) {
break;
}
}
- if (size != columnsFromPath.size()) {
+ if (size != expectedColumns.size()) {
throw new UserException("Fail to parse columnsFromPath, expected: "
- + columnsFromPath + ", filePath: " + filePath);
+ + expectedColumns + ", filePath: " + filePath);
+ }
+ return new ParsedColumnsFromPath(
+ Lists.newArrayList(columns),
Lists.newArrayList(columnsFromPathIsNull));
+ }
+
+ public static ParsedColumnsFromPath
parseColumnsFromPathWithNullInfoForLoad(
+ String filePath,
+ List<String> columnsFromPath,
+ boolean caseSensitive,
+ boolean isACID)
+ throws UserException {
+ ParsedColumnsFromPath parsed = parseColumnsFromPathWithNullInfo(
+ filePath, columnsFromPath, caseSensitive, isACID);
+ List<String> values = new ArrayList<>(parsed.getValues());
+ List<Boolean> isNull = new ArrayList<>(parsed.getIsNull());
+ for (int i = 0; i < values.size(); i++) {
+ if (FeConstants.null_string.equals(values.get(i))) {
+ values.set(i, "");
+ isNull.set(i, true);
+ }
}
- return Lists.newArrayList(columns);
+ return new ParsedColumnsFromPath(values, isNull);
}
public static ParsedColumnsFromPath normalizeColumnsFromPath(List<String>
columnsFromPath) {
@@ -228,8 +262,7 @@ public class BrokerUtil {
List<String> values = new ArrayList<>(columnsFromPath.size());
List<Boolean> isNull = new ArrayList<>(columnsFromPath.size());
for (String value : columnsFromPath) {
- boolean nullValue = value == null ||
FeConstants.null_string.equals(value)
- ||
HiveExternalMetaCache.HIVE_DEFAULT_PARTITION.equals(value);
+ boolean nullValue = value == null ||
HiveExternalMetaCache.HIVE_DEFAULT_PARTITION.equals(value);
values.add(nullValue ? "" : value);
isNull.add(nullValue);
}
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 d81ba7daa84..d6320277ef3 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
@@ -266,11 +266,12 @@ public class FileGroupInfo {
Util.getOrInferCompressType(context.fileGroup.getFileFormatProperties().getCompressionType(),
fileStatus.path);
context.params.setCompressType(compressType);
- List<String> columnsFromPath =
BrokerUtil.parseColumnsFromPath(fileStatus.path,
- context.fileGroup.getColumnNamesFromPath());
+ BrokerUtil.ParsedColumnsFromPath columnsFromPath =
+
BrokerUtil.parseColumnsFromPathWithNullInfoForLoad(fileStatus.path,
+ context.fileGroup.getColumnNamesFromPath(),
true, false);
List<String> 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 +313,16 @@ public class FileGroupInfo {
Util.getOrInferCompressType(context.fileGroup.getFileFormatProperties().getCompressionType(),
fileStatus.path);
context.params.setCompressType(compressType);
- List<String> columnsFromPath =
BrokerUtil.parseColumnsFromPath(fileStatus.path,
- context.fileGroup.getColumnNamesFromPath());
+ BrokerUtil.ParsedColumnsFromPath columnsFromPath =
+
BrokerUtil.parseColumnsFromPathWithNullInfoForLoad(fileStatus.path,
+ context.fileGroup.getColumnNamesFromPath(), true,
false);
List<String> 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 +331,8 @@ public class FileGroupInfo {
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 +403,8 @@ public class FileGroupInfo {
}
private TFileRangeDesc createFileRangeDesc(long curFileOffset,
TBrokerFileStatus fileStatus, long rangeBytes,
- List<String> columnsFromPath, List<String> columnsFromPathKeys) {
+ List<String> columnsFromPath, List<String> columnsFromPathKeys,
+ List<Boolean> columnsFromPathIsNull) {
TFileRangeDesc rangeDesc = new TFileRangeDesc();
if (jobType == JobType.BULK_LOAD) {
rangeDesc.setPath(fileStatus.path);
@@ -410,6 +413,7 @@ public class FileGroupInfo {
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/FileQueryScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java
index d83326a1838..a7a0e7b381a 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
@@ -507,11 +507,11 @@ public abstract class FileQueryScanNode extends
FileScanNode {
HiveSplit hiveSplit = (HiveSplit) fileSplit;
isACID = hiveSplit.isACID();
}
- List<String> rawPartitionValues = fileSplit.getPartitionValues() ==
null
- ? BrokerUtil.parseColumnsFromPath(fileSplit.getPathString(),
pathPartitionKeys,
- false, isACID) : fileSplit.getPartitionValues();
BrokerUtil.ParsedColumnsFromPath partitionValues =
- BrokerUtil.normalizeColumnsFromPath(rawPartitionValues);
+ fileSplit.getPartitionValues() == null
+ ? BrokerUtil.parseColumnsFromPathWithNullInfo(
+ fileSplit.getPathString(), pathPartitionKeys,
false, isACID)
+ :
BrokerUtil.normalizeColumnsFromPath(fileSplit.getPartitionValues());
TFileRangeDesc rangeDesc = createFileRangeDesc(fileSplit,
partitionValues.getValues(),
pathPartitionKeys, partitionValues.getIsNull());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java
index b52ef1f92a2..73986138c51 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java
@@ -28,7 +28,6 @@ import org.apache.doris.catalog.TableIf;
import org.apache.doris.catalog.Type;
import org.apache.doris.common.AnalysisException;
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;
@@ -436,9 +435,10 @@ public class HiveExternalMetaCache extends
AbstractExternalMetaCache {
try {
FileCacheValue result = getFileCache(catalog, finalLocation,
key.inputFormat,
key.getPartitionValues(), directoryLister, table);
+ // 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 91c6da61a57..6c6888e3792 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
@@ -27,6 +27,7 @@ import org.apache.doris.catalog.PartitionItem;
import org.apache.doris.catalog.Type;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.BrokerUtil;
import org.apache.doris.common.util.FileFormatUtils;
import org.apache.doris.common.util.LocationPath;
import org.apache.doris.datasource.ExternalTable;
@@ -330,8 +331,11 @@ public class HudiScanNode extends HiveScanNode {
formPathKeys.add(entry.getKey());
formPathValues.add(entry.getValue());
}
+ BrokerUtil.ParsedColumnsFromPath parsedColumnsFromPath =
+ BrokerUtil.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 84eed01e42c..df081f72013 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
@@ -800,12 +800,22 @@ public class IcebergUtils {
}
public static List<String> getIdentityPartitionColumns(Table table) {
+ return getIdentityPartitionColumns(table, false, false);
+ }
+
+ public static List<String> getIdentityPartitionColumns(Table table,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
LinkedHashSet<String> partitionColumns = new LinkedHashSet<>();
for (PartitionSpec spec : table.specs().values()) {
for (PartitionField partitionField : spec.fields()) {
if (!partitionField.transform().isIdentity()) {
continue;
}
+ NestedField sourceField =
table.schema().findField(partitionField.sourceId());
+ if (sourceField == null || !isSupportedPartitionValueType(
+ sourceField.type(), enableMappingVarbinary,
enableMappingTimestampTz)) {
+ continue;
+ }
String columnName =
table.schema().findColumnName(partitionField.sourceId());
if (columnName != null) {
partitionColumns.add(columnName);
@@ -815,8 +825,47 @@ public class IcebergUtils {
return new ArrayList<>(partitionColumns);
}
+ /**
+ * Get identity partition columns that exist in all partition specs.
+ * The legacy 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<String> getCommonIdentityPartitionColumns(Table table) {
+ return getCommonIdentityPartitionColumns(table, false, false);
+ }
+
+ public static List<String> getCommonIdentityPartitionColumns(Table table,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
+ LinkedHashSet<Integer> 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(), enableMappingVarbinary,
enableMappingTimestampTz)) {
+ commonSourceIds.add(field.sourceId());
+ }
+ }
+ for (PartitionSpec spec : table.specs().values()) {
+ Set<Integer> 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(columnName -> columnName != null)
+ .collect(Collectors.toList());
+ }
+
public static Map<String, String>
getIdentityPartitionInfoMap(PartitionData partitionData,
PartitionSpec partitionSpec, Table table, String timeZone) {
+ return getIdentityPartitionInfoMap(partitionData, partitionSpec,
table, timeZone, false, false);
+ }
+
+ public static Map<String, String>
getIdentityPartitionInfoMap(PartitionData partitionData,
+ PartitionSpec partitionSpec, Table table, String timeZone,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
Map<String, String> partitionInfoMap = Maps.newLinkedHashMap();
List<NestedField> fields =
partitionData.getPartitionType().asNestedType().fields();
List<PartitionField> partitionFields = partitionSpec.fields();
@@ -829,8 +878,8 @@ public class IcebergUtils {
if (!partitionField.transform().isIdentity()) {
continue;
}
- TypeID partitionTypeId = field.type().typeId();
- if (partitionTypeId == TypeID.BINARY || partitionTypeId ==
TypeID.FIXED) {
+ if (!isSupportedPartitionValueType(
+ field.type(), enableMappingVarbinary,
enableMappingTimestampTz)) {
continue;
}
@@ -849,6 +898,19 @@ public class IcebergUtils {
return partitionInfoMap;
}
+ private static boolean
isSupportedPartitionValueType(org.apache.iceberg.types.Type type,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
+ TypeID typeId = type.typeId();
+ if (typeId == TypeID.BINARY || typeId == TypeID.FIXED) {
+ return false;
+ }
+ if (enableMappingVarbinary && typeId == TypeID.UUID) {
+ return false;
+ }
+ return !enableMappingTimestampTz || typeId != TypeID.TIMESTAMP
+ || !((TimestampType) type).shouldAdjustToUTC();
+ }
+
public static List<String> getPartitionValues(PartitionData partitionData,
PartitionSpec partitionSpec,
String timeZone) {
List<NestedField> fields =
partitionData.getPartitionType().asNestedType().fields();
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 9eea2a95736..957ab6ed55e 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.analysis.TupleDescriptor;
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;
@@ -153,8 +154,13 @@ 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<PartitionData, Map<String, String>> partitionMapInfos;
+ // Used to avoid repeatedly calculating partition info map for the same
+ // partition data and spec.
+ private Map<Pair<Integer, PartitionData>, Map<String, String>>
partitionMapInfos;
+ private List<String> orderedPathPartitionKeys;
+ private List<String> orderedPartitionMetadataKeys;
+ private boolean enableMappingVarbinaryForPartitionMetadata;
+ private boolean enableMappingTimestampTzForPartitionMetadata;
private boolean isPartitionedTable;
private int formatVersion;
private ExecutionAuthenticator preExecutionAuthenticator;
@@ -251,6 +257,7 @@ public class IcebergScanNode extends FileQueryScanNode {
icebergTable = source.getIcebergTable();
icebergTable = useFrozenTableGeneration(icebergTable);
partitionMapInfos = new HashMap<>();
+ initializePartitionMetadata();
isPartitionedTable = icebergTable.spec().isPartitioned();
// Metadata tables (system tables) are not BaseTable instances, so
we need to handle this case
if (icebergTable instanceof BaseTable) {
@@ -409,21 +416,75 @@ public class IcebergScanNode extends FileQueryScanNode {
deleteFilesDescByReferencedDataFile.put(icebergSplit.getOriginalPath(),
nonEqualityDeleteFileDesc);
}
tableFormatFileDesc.setIcebergParams(fileDesc);
- Map<String, String> partitionValues =
icebergSplit.getIcebergPartitionValues();
- if (partitionValues != null) {
- List<String> fromPathKeys = new ArrayList<>();
- List<String> fromPathValues = new ArrayList<>();
- List<Boolean> fromPathIsNull = new ArrayList<>();
- for (Map.Entry<String, String> entry : partitionValues.entrySet())
{
- fromPathKeys.add(entry.getKey());
- fromPathValues.add(entry.getValue() != null ? entry.getValue()
: "");
- fromPathIsNull.add(entry.getValue() == null);
+ setPartitionValues(rangeDesc,
icebergSplit.getIcebergPartitionValues());
+ rangeDesc.setTableFormatParams(tableFormatFileDesc);
+ }
+
+ private List<String> getOrderedPathPartitionKeys() {
+ if (orderedPathPartitionKeys == null) {
+ initializePartitionMetadata();
+ }
+ return orderedPathPartitionKeys;
+ }
+
+ private List<String> getOrderedPartitionMetadataKeys() {
+ if (orderedPartitionMetadataKeys == null) {
+ initializePartitionMetadata();
+ }
+ return orderedPartitionMetadataKeys;
+ }
+
+ private void initializePartitionMetadata() {
+ if (isSystemTable || icebergTable == null) {
+ orderedPathPartitionKeys = Collections.emptyList();
+ orderedPartitionMetadataKeys = Collections.emptyList();
+ return;
+ }
+ enableMappingVarbinaryForPartitionMetadata =
getEnableMappingVarbinary();
+ enableMappingTimestampTzForPartitionMetadata =
getEnableMappingTimestampTz();
+ orderedPathPartitionKeys = Collections.unmodifiableList(
+ IcebergUtils.getCommonIdentityPartitionColumns(icebergTable,
+ enableMappingVarbinaryForPartitionMetadata,
+ enableMappingTimestampTzForPartitionMetadata));
+ if (sessionVariable.enableFileScannerV2) {
+ orderedPartitionMetadataKeys = Collections.unmodifiableList(
+ IcebergUtils.getIdentityPartitionColumns(icebergTable,
+ enableMappingVarbinaryForPartitionMetadata,
+ enableMappingTimestampTzForPartitionMetadata));
+ } else {
+ orderedPartitionMetadataKeys = orderedPathPartitionKeys;
+ }
+ }
+
+ @VisibleForTesting
+ void setPartitionValues(TFileRangeDesc rangeDesc, Map<String, String>
partitionValues) {
+ rangeDesc.unsetColumnsFromPathKeys();
+ rangeDesc.unsetColumnsFromPath();
+ rangeDesc.unsetColumnsFromPathIsNull();
+
+ List<String> orderedPartitionKeys = getOrderedPartitionMetadataKeys();
+ if (orderedPartitionKeys.isEmpty() || partitionValues == null ||
partitionValues.isEmpty()) {
+ return;
+ }
+
+ List<String> fromPathKeys = new
ArrayList<>(orderedPartitionKeys.size());
+ List<String> fromPathValues = new
ArrayList<>(orderedPartitionKeys.size());
+ List<Boolean> fromPathIsNull = new
ArrayList<>(orderedPartitionKeys.size());
+ for (String partitionKey : orderedPartitionKeys) {
+ if (!partitionValues.containsKey(partitionKey)) {
+ continue;
}
- rangeDesc.setColumnsFromPathKeys(fromPathKeys);
- rangeDesc.setColumnsFromPath(fromPathValues);
- rangeDesc.setColumnsFromPathIsNull(fromPathIsNull);
+ String partitionValue = partitionValues.get(partitionKey);
+ fromPathKeys.add(partitionKey);
+ fromPathValues.add(partitionValue == null ? "" : partitionValue);
+ fromPathIsNull.add(partitionValue == null);
}
- rangeDesc.setTableFormatParams(tableFormatFileDesc);
+ if (fromPathKeys.isEmpty()) {
+ return;
+ }
+ rangeDesc.setColumnsFromPathKeys(fromPathKeys);
+ rangeDesc.setColumnsFromPath(fromPathValues);
+ rangeDesc.setColumnsFromPathIsNull(fromPathIsNull);
}
private void setIcebergPositionDeleteSysTableParams(TFileRangeDesc
rangeDesc, IcebergSplit icebergSplit,
@@ -1082,11 +1143,11 @@ public class IcebergScanNode extends FileQueryScanNode {
split.setPartitionSpecId(specId);
split.setPartitionDataJson(IcebergUtils.getPartitionDataJson(
partitionData, partitionSpec,
sessionVariable.getTimeZone()));
- }
- if (sessionVariable.isEnableRuntimeFilterPartitionPrune()) {
Map<String, String> partitionInfoMap =
partitionMapInfos.computeIfAbsent(
- partitionData, k ->
IcebergUtils.getIdentityPartitionInfoMap(
- partitionData, partitionSpec, icebergTable,
sessionVariable.getTimeZone()));
+ Pair.of(specId, partitionData), k ->
IcebergUtils.getIdentityPartitionInfoMap(
+ partitionData, partitionSpec, icebergTable,
sessionVariable.getTimeZone(),
+ enableMappingVarbinaryForPartitionMetadata,
+ enableMappingTimestampTzForPartitionMetadata));
// A spec may mix identity and transformed fields. Keep its
identity values so a
// runtime filter can prune that split without treating source
columns as constants
// for files written under another evolved spec.
@@ -1094,7 +1155,7 @@ public class IcebergScanNode extends FileQueryScanNode {
split.setIcebergPartitionValues(partitionInfoMap);
}
} else {
- partitionMapInfos.put(partitionData, null);
+ partitionMapInfos.put(Pair.of(specId, null),
Collections.emptyMap());
}
}
return split;
@@ -1538,20 +1599,7 @@ public class IcebergScanNode extends FileQueryScanNode {
@Override
public List<String> 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<>();
+ return getOrderedPathPartitionKeys();
}
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 b0862277a63..d8885a22399 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
@@ -279,11 +279,12 @@ public class NereidsFileGroupInfo {
context.fileGroup.getFileFormatProperties().getCompressionType(),
fileStatus.path);
context.params.setCompressType(compressType);
- List<String> columnsFromPath =
BrokerUtil.parseColumnsFromPath(fileStatus.path,
- context.fileGroup.getColumnNamesFromPath());
+ BrokerUtil.ParsedColumnsFromPath columnsFromPath =
+
BrokerUtil.parseColumnsFromPathWithNullInfoForLoad(fileStatus.path,
+ context.fileGroup.getColumnNamesFromPath(),
true, false);
List<String> 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 class NereidsFileGroupInfo {
context.fileGroup.getFileFormatProperties().getCompressionType(),
fileStatus.path);
context.params.setCompressType(compressType);
- List<String> columnsFromPath =
BrokerUtil.parseColumnsFromPath(fileStatus.path,
- context.fileGroup.getColumnNamesFromPath());
+ BrokerUtil.ParsedColumnsFromPath columnsFromPath =
+
BrokerUtil.parseColumnsFromPathWithNullInfoForLoad(fileStatus.path,
+ context.fileGroup.getColumnNamesFromPath(), true,
false);
List<String> 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 class NereidsFileGroupInfo {
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,8 @@ public class NereidsFileGroupInfo {
}
private TFileRangeDesc createFileRangeDesc(long curFileOffset,
TBrokerFileStatus fileStatus, long rangeBytes,
- List<String> columnsFromPath, List<String> columnsFromPathKeys) {
+ List<String> columnsFromPath, List<String> columnsFromPathKeys,
+ List<Boolean> columnsFromPathIsNull) {
TFileRangeDesc rangeDesc = new TFileRangeDesc();
if (jobType == FileGroupInfo.JobType.BULK_LOAD) {
rangeDesc.setPath(fileStatus.path);
@@ -429,6 +432,7 @@ public class NereidsFileGroupInfo {
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/common/util/BrokerUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/common/util/BrokerUtilTest.java
index 0c24316cdc4..aa644f941a4 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/common/util/BrokerUtilTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/common/util/BrokerUtilTest.java
@@ -158,14 +158,47 @@ public class BrokerUtilTest {
}
+ @Test
+ public void parseColumnsFromPathPreservesNullMetadataAndInputKeys() throws
Exception {
+ List<String> partitionKeys = Lists.newArrayList("region", "dt");
+ BrokerUtil.ParsedColumnsFromPath parsed =
BrokerUtil.parseColumnsFromPathWithNullInfo(
+ "hdfs://host/table/Region=cn/Dt=" +
HiveExternalMetaCache.HIVE_DEFAULT_PARTITION
+ + "/data.parquet",
+ partitionKeys, false, false);
+
+ Assert.assertEquals(Lists.newArrayList("cn", ""), parsed.getValues());
+ Assert.assertEquals(Lists.newArrayList(false, true),
parsed.getIsNull());
+ Assert.assertEquals(Lists.newArrayList("region", "dt"), partitionKeys);
+
+ parsed = BrokerUtil.parseColumnsFromPathWithNullInfo(
+ "hdfs://host/table/p=\\N/data.orc",
Collections.singletonList("p"), true, false);
+ Assert.assertEquals(Collections.singletonList("\\N"),
parsed.getValues());
+ Assert.assertEquals(Collections.singletonList(false),
parsed.getIsNull());
+
+ parsed = BrokerUtil.parseColumnsFromPathWithNullInfo(
+ "hdfs://host/table/p=value/delta_1_1/bucket_00000",
+ Collections.singletonList("p"), true, true);
+ Assert.assertEquals(Collections.singletonList("value"),
parsed.getValues());
+ Assert.assertEquals(Collections.singletonList(false),
parsed.getIsNull());
+ }
+
@Test
public void normalizeColumnsFromPathPreservesNullInfo() {
BrokerUtil.ParsedColumnsFromPath parsed =
BrokerUtil.normalizeColumnsFromPath(
Lists.newArrayList("p1", FeConstants.null_string,
HiveExternalMetaCache.HIVE_DEFAULT_PARTITION, null));
- Assert.assertEquals(Lists.newArrayList("p1", "", "", ""),
parsed.getValues());
- Assert.assertEquals(Lists.newArrayList(false, true, true, true),
parsed.getIsNull());
+ Assert.assertEquals(Lists.newArrayList("p1", "\\N", "", ""),
parsed.getValues());
+ Assert.assertEquals(Lists.newArrayList(false, false, true, true),
parsed.getIsNull());
+ }
+
+ @Test
+ public void parseColumnsFromPathForLoadKeepsLegacyNullSemantics() throws
Exception {
+ BrokerUtil.ParsedColumnsFromPath parsed =
BrokerUtil.parseColumnsFromPathWithNullInfoForLoad(
+ "hdfs://host/table/p=\\N/data.orc",
Collections.singletonList("p"), true, false);
+
+ Assert.assertEquals(Collections.singletonList(""), parsed.getValues());
+ Assert.assertEquals(Collections.singletonList(true),
parsed.getIsNull());
}
@Test
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 835959c6c0c..2dd1f300c24 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
@@ -471,6 +471,35 @@ public class IcebergUtilsTest {
Assert.assertEquals(Arrays.asList("Dt", "id"),
IcebergUtils.getIdentityPartitionColumns(table));
}
+ @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<Integer, PartitionSpec> 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(Collections.singletonList("Dt"),
+ IcebergUtils.getCommonIdentityPartitionColumns(table));
+ }
+
@Test
public void testGetIdentityPartitionInfoMapReturnsIdentityColumnsOnly() {
Schema schema = new Schema(
@@ -492,6 +521,36 @@ public class IcebergUtilsTest {
Assert.assertEquals(Collections.singletonMap("Dt", "2025-01-01"),
partitionInfoMap);
}
+ @Test
+ public void testMappedTypesAreExcludedFromPartitionMetadata() {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "Dt", Types.StringType.get()),
+ Types.NestedField.required(2, "uuid_col",
Types.UUIDType.get()),
+ Types.NestedField.required(3, "ts_tz",
Types.TimestampType.withZone()));
+ PartitionSpec partitionSpec = PartitionSpec.builderFor(schema)
+ .identity("Dt")
+ .identity("uuid_col")
+ .identity("ts_tz")
+ .build();
+ PartitionData partitionData = new
PartitionData(partitionSpec.partitionType());
+ partitionData.set(0, "2026-08-03");
+ partitionData.set(1,
UUID.fromString("123e4567-e89b-12d3-a456-426614174000"));
+ partitionData.set(2, 0L);
+
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.schema()).thenReturn(schema);
+ Mockito.when(table.spec()).thenReturn(partitionSpec);
+
Mockito.when(table.specs()).thenReturn(Collections.singletonMap(partitionSpec.specId(),
partitionSpec));
+
+ Assert.assertEquals(Collections.singletonList("Dt"),
+ IcebergUtils.getIdentityPartitionColumns(table, true, true));
+ Assert.assertEquals(Collections.singletonList("Dt"),
+ IcebergUtils.getCommonIdentityPartitionColumns(table, true,
true));
+ Assert.assertEquals(Collections.singletonMap("Dt", "2026-08-03"),
+ IcebergUtils.getIdentityPartitionInfoMap(
+ partitionData, partitionSpec, table, "Asia/Shanghai",
true, true));
+ }
+
@Test
public void
testGetIdentityPartitionInfoMapSupportsFloatingPointPartitions() {
Schema schema = new Schema(
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 bf9799afed3..670754614dd 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
@@ -90,7 +90,10 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -116,19 +119,26 @@ public class IcebergScanNodeTest {
private static class TestIcebergScanNode extends IcebergScanNode {
private final boolean enableMappingVarbinary;
private final boolean batchMode;
+ private final boolean enableMappingTimestampTz;
private TableScan tableScan;
TestIcebergScanNode(SessionVariable sv) {
- this(sv, false, false);
+ this(sv, false, false, false);
}
TestIcebergScanNode(SessionVariable sv, boolean
enableMappingVarbinary) {
- this(sv, enableMappingVarbinary, false);
+ this(sv, enableMappingVarbinary, false, false);
}
TestIcebergScanNode(SessionVariable sv, boolean
enableMappingVarbinary, boolean batchMode) {
+ this(sv, enableMappingVarbinary, false, batchMode);
+ }
+
+ TestIcebergScanNode(SessionVariable sv, boolean enableMappingVarbinary,
+ boolean enableMappingTimestampTz, boolean batchMode) {
super(new PlanNodeId(0), new TupleDescriptor(new TupleId(0)), sv,
ScanContext.EMPTY);
this.enableMappingVarbinary = enableMappingVarbinary;
+ this.enableMappingTimestampTz = enableMappingTimestampTz;
this.batchMode = batchMode;
}
@@ -155,6 +165,11 @@ public class IcebergScanNodeTest {
return enableMappingVarbinary;
}
+ @Override
+ protected boolean getEnableMappingTimestampTz() {
+ return enableMappingTimestampTz;
+ }
+
@Override
public List<String> getPathPartitionKeys() {
return Collections.emptyList();
@@ -230,6 +245,153 @@ public class IcebergScanNodeTest {
.getFields().get(1).getFieldPtr().getName());
}
+ @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<Integer, PartitionSpec> 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);
+ setIcebergTable(node, table);
+
+ Map<String, String> 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());
+ }
+
+ @Test
+ public void testSetPartitionValuesUsesPerSpecMetadataWithFileScannerV2()
throws Exception {
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.enableFileScannerV2 = true;
+ TestIcebergScanNode node = new TestIcebergScanNode(sessionVariable);
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "Region",
Types.StringType.get()),
+ Types.NestedField.required(2, "Dt", Types.StringType.get()),
+ Types.NestedField.required(3, "Category",
Types.StringType.get()));
+ PartitionSpec oldSpec = PartitionSpec.builderFor(schema)
+ .withSpecId(1)
+ .identity("Region")
+ .identity("Dt")
+ .build();
+ PartitionSpec currentSpec = PartitionSpec.builderFor(schema)
+ .withSpecId(2)
+ .identity("Dt")
+ .identity("Category")
+ .build();
+ Map<Integer, PartitionSpec> 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);
+ setIcebergTable(node, table);
+
+ Map<String, String> partitionValues = new HashMap<>();
+ partitionValues.put("Category", "books");
+ partitionValues.put("Dt", null);
+ TFileRangeDesc rangeDesc = new TFileRangeDesc();
+ node.setPartitionValues(rangeDesc, partitionValues);
+
+ Assert.assertEquals(Arrays.asList("Dt", "Category"),
rangeDesc.getColumnsFromPathKeys());
+ Assert.assertEquals(Arrays.asList("", "books"),
rangeDesc.getColumnsFromPath());
+ Assert.assertEquals(Arrays.asList(true, false),
rangeDesc.getColumnsFromPathIsNull());
+ }
+
+ @Test
+ public void
testSetPartitionValuesKeepsCommonMetadataWithLegacyFileScanner() throws
Exception {
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.enableFileScannerV2 = false;
+ TestIcebergScanNode node = new TestIcebergScanNode(sessionVariable);
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "Region",
Types.StringType.get()),
+ Types.NestedField.required(2, "Dt", Types.StringType.get()),
+ Types.NestedField.required(3, "Category",
Types.StringType.get()));
+ PartitionSpec oldSpec = PartitionSpec.builderFor(schema)
+ .withSpecId(1)
+ .identity("Region")
+ .identity("Dt")
+ .build();
+ PartitionSpec currentSpec = PartitionSpec.builderFor(schema)
+ .withSpecId(2)
+ .identity("Dt")
+ .identity("Category")
+ .build();
+ Map<Integer, PartitionSpec> 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);
+ setIcebergTable(node, table);
+
+ Map<String, String> partitionValues = new HashMap<>();
+ partitionValues.put("Region", "cn");
+ partitionValues.put("Dt", "2026-07-31");
+ TFileRangeDesc rangeDesc = new TFileRangeDesc();
+ node.setPartitionValues(rangeDesc, partitionValues);
+
+ Assert.assertEquals(Collections.singletonList("Dt"),
rangeDesc.getColumnsFromPathKeys());
+ Assert.assertEquals(Collections.singletonList("2026-07-31"),
rangeDesc.getColumnsFromPath());
+ Assert.assertEquals(Collections.singletonList(false),
rangeDesc.getColumnsFromPathIsNull());
+ }
+
+ @Test
+ public void testSetPartitionValuesSkipsValuesUnsupportedByMappedTypes()
throws Exception {
+ for (boolean enableFileScannerV2 : Arrays.asList(false, true)) {
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.enableFileScannerV2 = enableFileScannerV2;
+ sessionVariable.setTimeZone("Asia/Shanghai");
+ TestIcebergScanNode node = new
TestIcebergScanNode(sessionVariable, true, true, false);
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "Dt",
Types.StringType.get()),
+ Types.NestedField.required(2, "uuid_col",
Types.UUIDType.get()),
+ Types.NestedField.required(3, "ts_tz",
Types.TimestampType.withZone()));
+ PartitionSpec spec = PartitionSpec.builderFor(schema)
+ .identity("Dt")
+ .identity("uuid_col")
+ .identity("ts_tz")
+ .build();
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.schema()).thenReturn(schema);
+ Mockito.when(table.spec()).thenReturn(spec);
+
Mockito.when(table.specs()).thenReturn(Collections.singletonMap(spec.specId(),
spec));
+ setIcebergTable(node, table);
+
+ Map<String, String> partitionValues = new LinkedHashMap<>();
+ partitionValues.put("Dt", "2026-08-03");
+ partitionValues.put("uuid_col",
"123e4567-e89b-12d3-a456-426614174000");
+ partitionValues.put("ts_tz", "2026-08-03T16:00:00");
+ TFileRangeDesc rangeDesc = new TFileRangeDesc();
+ node.setPartitionValues(rangeDesc, partitionValues);
+
+ Assert.assertEquals(Collections.singletonList("Dt"),
rangeDesc.getColumnsFromPathKeys());
+ Assert.assertEquals(Collections.singletonList("2026-08-03"),
rangeDesc.getColumnsFromPath());
+ Assert.assertEquals(Collections.singletonList(false),
rangeDesc.getColumnsFromPathIsNull());
+
+ Mockito.clearInvocations(table);
+ node.setPartitionValues(new TFileRangeDesc(), partitionValues);
+ Mockito.verify(table, Mockito.never()).specs();
+ Mockito.verify(table, Mockito.never()).schema();
+ }
+ }
+
@Test
public void testExtractNameMappingDistinguishesAbsentAndEmpty() throws
Exception {
TestIcebergScanNode node = new TestIcebergScanNode(new
SessionVariable());
@@ -879,6 +1041,11 @@ public class IcebergScanNodeTest {
Field icebergTableField =
IcebergScanNode.class.getDeclaredField("icebergTable");
icebergTableField.setAccessible(true);
icebergTableField.set(node, table);
+ for (String fieldName : Arrays.asList("orderedPathPartitionKeys",
"orderedPartitionMetadataKeys")) {
+ Field field = IcebergScanNode.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(node, null);
+ }
}
private static void setIcebergSource(IcebergScanNode node, IcebergSource
source) throws Exception {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
