github-actions[bot] commented on code in PR #66010:
URL: https://github.com/apache/doris/pull/66010#discussion_r3712631501
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -410,11 +409,14 @@ private TScanRangeLocations splitToScanRange(
HiveSplit hiveSplit = (HiveSplit) fileSplit;
isACID = hiveSplit.isACID();
}
- List<String> partitionValuesFromPath = fileSplit.getPartitionValues()
== null
- ? BrokerUtil.parseColumnsFromPath(fileSplit.getPathString(),
pathPartitionKeys,
- false, isACID) : fileSplit.getPartitionValues();
+ FilePartitionUtils.ParsedColumnsFromPath partitionValuesFromPath =
+ fileSplit.getPartitionValues() == null
+ ? FilePartitionUtils.parseColumnsFromPathWithNullInfo(
Review Comment:
[P1] Keep explicit NULLs representable for TVF path columns
This shared path now turns `__HIVE_DEFAULT_PARTITION__` into `""/true` for
static file TVFs too, but `ExternalFileTableValuedFunction.fillColumns()`
declares every `path_partition_keys` column non-nullable. BE's explicit-NULL
branch DCHECKs that the slot is nullable and then inserts defaults; for this
slot that is an assertion failure in debug builds and an empty string rather
than SQL NULL in release builds. Please make TVF path columns nullable (or keep
a clearly defined literal contract for non-nullable TVFs) and add an HDFS/S3
TVF case that selects and filters a sentinel-valued path key.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java:
##########
@@ -638,59 +639,70 @@ 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<String, String> getPartitionInfoMap(PartitionData
partitionData, PartitionSpec partitionSpec,
- String timeZone) {
- Map<String, String> partitionInfoMap = new HashMap<>();
- List<NestedField> fields =
partitionData.getPartitionType().asNestedType().fields();
+ public static List<String> getCommonIdentityPartitionColumns(Table table) {
+ 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().typeId())) {
+ 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);
Review Comment:
[P1] Keep old manifest values available after partition evolution
Intersecting every historical spec with `table.spec()` drops an identity
source as soon as a later default spec removes or transforms it—even if the
selected snapshot contains only old-spec files. For Hive-migrated files, that
manifest value may be the only copy because the physical file omits the Hive
partition column. Once the key is removed here, `setPartitionValues` emits no
metadata and branch-4.0's missing-column path fills NULL/default instead, so
current or time-travel queries return wrong values. Please separate the
scan-wide pruning key set from spec-specific materialization metadata (and
collect it by each planned file's `specId`), with a migrated-file plus
spec-evolution test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -299,23 +301,44 @@ private void setIcebergParams(TFileRangeDesc rangeDesc,
IcebergSplit icebergSpli
}
}
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);
- }
- rangeDesc.setColumnsFromPathKeys(fromPathKeys);
- rangeDesc.setColumnsFromPath(fromPathValues);
- rangeDesc.setColumnsFromPathIsNull(fromPathIsNull);
- }
+ setPartitionValues(rangeDesc,
icebergSplit.getIcebergPartitionValues());
rangeDesc.setTableFormatParams(tableFormatFileDesc);
}
+ private List<String> getOrderedPathPartitionKeys() {
+ if (icebergTable == null) {
+ return Collections.emptyList();
+ }
+ return IcebergUtils.getCommonIdentityPartitionColumns(icebergTable);
+ }
+
+ @VisibleForTesting
+ void setPartitionValues(TFileRangeDesc rangeDesc, Map<String, String>
partitionValues) {
+ rangeDesc.unsetColumnsFromPathKeys();
+ rangeDesc.unsetColumnsFromPath();
+ rangeDesc.unsetColumnsFromPathIsNull();
+
+ List<String> orderedPartitionKeys = getOrderedPathPartitionKeys();
+ if (orderedPartitionKeys.isEmpty()) {
+ return;
+ }
+ Preconditions.checkState(partitionValues != null,
+ "Missing partition values for Iceberg identity-partitioned
table");
+
+ List<String> fromPathValues = new
ArrayList<>(orderedPartitionKeys.size());
+ List<Boolean> 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);
Review Comment:
[P1] Make Iceberg missing-column fallback per range and per column
Sending these keys on every dual file/partition slot exposes branch-4.0's
scan-wide fallback bug. When any advertised identity column is absent,
`FileScanner` permanently flips `_fill_partition_from_path` to true and passes
*all* partition descriptors to the reader. A later range (or another identity
column in the same range) that physically contains the column is still read by
Parquet/ORC and then has another `rows` manifest values appended, yielding
inconsistent column sizes or wrong materialization. Reset the decision per
reader and pass only the current file's actually missing keys; please cover
partial-missing and missing-then-present Parquet/ORC ranges with pruning both
on and off.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -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);
Review Comment:
[P1] Preserve NULL before stringifying Hudi partition keys
This normalizer cannot recover a Hudi NULL on the normal snapshot path.
`TablePartitionValues` first converts the Hive sentinel to a `NullLiteral` with
`isHive=false`; `getPartitionValuesAsStringList()` then turns it into the
ordinary string `"NULL"`. Both the generic split values and this Hudi map
therefore reach `normalizeColumnsFromPath` as `"NULL"/false`, so BE
materializes a literal string (or fails to parse a typed partition) instead of
SQL NULL. The existing Hudi MV regression even documents the resulting lost
null rows. Please carry Java null/an explicit null bit out of `PartitionKey`
before stringification and cover pruning on and off.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java:
##########
@@ -638,59 +639,70 @@ 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<String, String> getPartitionInfoMap(PartitionData
partitionData, PartitionSpec partitionSpec,
- String timeZone) {
- Map<String, String> partitionInfoMap = new HashMap<>();
- List<NestedField> fields =
partitionData.getPartitionType().asNestedType().fields();
+ public static List<String> getCommonIdentityPartitionColumns(Table table) {
+ 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().typeId())) {
+ 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(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<String, String>
getIdentityPartitionInfoMap(PartitionData partitionData,
+ PartitionSpec partitionSpec, Table table, String timeZone) {
+ Map<String, String> partitionInfoMap = Maps.newLinkedHashMap();
+ List<NestedField> fields =
partitionData.getPartitionType().asNestedType().fields();
List<PartitionField> partitionFields = partitionSpec.fields();
Preconditions.checkArgument(fields.size() == partitionFields.size(),
"PartitionData fields size does not match PartitionSpec fields
size");
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());
Review Comment:
[P1] Resolve partition names from the selected snapshot schema
`useSnapshot`/`useRef` selects historical data, and Doris builds that
query's tuple from the selected snapshot schema, but both new identity helpers
resolve the stable source ID through the latest `table.schema()`. After
`old_name` is renamed to `new_name`, an old-snapshot tuple still has `old_name`
while these range keys use `new_name`; BE matches them by exact name, so a
migrated file that omits the physical column receives NULL/default instead of
its manifest value. Please derive both ordered keys and per-spec values from
the selected schema/tuple mapping, with snapshot/tag/branch coverage across an
identity-source rename.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -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<String, String> 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<String, String> partitionInfoMap =
partitionMapInfos.computeIfAbsent(
+ Pair.of(specId, partitionData), k ->
IcebergUtils.getIdentityPartitionInfoMap(
Review Comment:
[P2] Normalize negative timestamp microseconds before materializing them
This call is now unconditional, but `serializePartitionValue` still splits
timestamps with `/` and `%`. For a valid pre-epoch value such as
`timestampMicros = -1`, Java produces epoch seconds `0` and nanoseconds
`-1000`, so `LocalDateTime.ofEpochSecond` throws and the split cannot be
planned (the caller catches only `UnsupportedOperationException`). Use
`Math.floorDiv`/`Math.floorMod` and add negative fractional
TIMESTAMP/TIMESTAMPTZ cases, especially now that disabling runtime partition
pruning no longer avoids serialization.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
