This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 3b81a1eeaa9 [fix](hudi) Decode Spark SQL timestamp partition paths
(#69157)
3b81a1eeaa9 is described below
commit 3b81a1eeaa9e5d94eae3714b353796193df83eb9
Author: Gabriel <[email protected]>
AuthorDate: Fri Oct 16 16:02:38 2026 +0800
[fix](hudi) Decode Spark SQL timestamp partition paths (#69157)
### What problem does this PR solve?
Spark SQL's Hudi `SqlKeyGenerator` always escapes formatted TIMESTAMP
partition values, independently of
`hoodie.table.url.encode.partitioning`. Doris currently unescapes
partition values only when that general flag is enabled. With the flag
absent or false, a physical component such as `event_time=2024-01-01
10%3A00%3A00` reaches the BE as an invalid timestamp instead of
`2024-01-01 10:00:00`. This breaks `test_hudi_partition_prune` and
`test_hudi_runtime_filter_partition_pruning`.
Carry timestamp partition-column names from the table schema through
Hudi handles and apply exactly one Hive-style unescape to those logical
values. Use the same interpretation for HMS/Hudi metadata listing,
partition pruning, snapshot pin filtering, and COW/MOR snapshot and
incremental scan ranges. Preserve physical file paths. Other types still
follow the persisted URL-encoding flag, so a STRING containing literal
`%3A`, `%253A`, or `+` remains distinct and unchanged when encoding is
disabled.
The writer behavior is documented in [Hudi's
SqlKeyGenerator](https://github.com/apache/hudi/blob/release-1.0.2/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/SqlKeyGenerator.scala).
### Validation
- Confirmed the new metadata regression fails before the fix with
expected `10:00:00`, actual `10%3A00%3A00`.
- `mvn -f fe/pom.xml -pl :fe-connector-hudi -am test
-Dmaven.build.cache.enabled=false`: 1,007 tests passed, including 270
Hudi tests; zero failures/errors/skips.
- FE Checkstyle/validate and Groovy syntax compilation passed.
- Coverage includes real persisted SDK configurations with
missing/false/true encoding flags, mixed timestamp/STRING partitions,
single-layer decoding, nulls, HMS and metadata pruning, historical
listing, a real local COW commit/write-stat mapping, and MOR native/JNI
Thrift values.
- Extended the existing external regression to compare native/JNI
timestamp rows under both partition-listing modes while keeping the
existing golden assertion and restoring the session setting.
The real Spark writer and full external Doris/HMS/Hudi suites were not
run locally; the new external assertion is syntax-checked and remains
for CI.
### Release note
Fix reads of Spark SQL Hudi TIMESTAMP partitions whose paths are escaped
while generic URL partition encoding is disabled.
### Check List (For Author)
- [x] Unit Test
- [x] Regression test extension
- [x] Behavior changed: timestamp partition values are decoded according
to Spark/Hudi writer semantics
- [x] No documentation needed
---
.../connector/hudi/COWIncrementalRelation.java | 7 +-
.../connector/hudi/HudiConnectorMetadata.java | 32 +++++--
.../doris/connector/hudi/HudiScanPlanProvider.java | 44 +++++++--
.../doris/connector/hudi/HudiTableHandle.java | 15 +++
.../hudi/HudiConnectorPartitionListingTest.java | 104 ++++++++++++++++++++-
.../hudi/HudiIncrementalPlanScanTest.java | 91 +++++++++++++++---
.../connector/hudi/HudiPartitionValuesTest.java | 29 ++++++
.../hudi/test_hudi_partition_prune.groovy | 12 +++
8 files changed, 303 insertions(+), 31 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/COWIncrementalRelation.java
b/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/COWIncrementalRelation.java
index 71566952746..1430a9a9288 100644
---
a/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/COWIncrementalRelation.java
+++
b/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/COWIncrementalRelation.java
@@ -42,6 +42,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.function.Consumer;
import java.util.function.UnaryOperator;
import java.util.stream.Collectors;
@@ -59,6 +60,7 @@ import java.util.stream.Collectors;
final class COWIncrementalRelation implements IncrementalRelation {
private final Map<String, String> optParams;
+ private final Set<String> timestampPartitionKeys;
private final HoodieTableMetaClient metaClient;
private final HollowCommitHandling hollowCommitHandling;
private final boolean startInstantArchived;
@@ -74,8 +76,9 @@ final class COWIncrementalRelation implements
IncrementalRelation {
COWIncrementalRelation(HoodieTableMetaClient metaClient, Configuration
configuration,
String startTs, String endTs, HollowCommitHandling
hollowCommitHandling,
- Map<String, String> optParams) throws IOException {
+ Map<String, String> optParams, Set<String> timestampPartitionKeys)
throws IOException {
this.optParams = optParams;
+ this.timestampPartitionKeys = timestampPartitionKeys;
this.metaClient = metaClient;
this.hollowCommitHandling = hollowCommitHandling;
this.startTs = startTs;
@@ -209,7 +212,7 @@ final class COWIncrementalRelation implements
IncrementalRelation {
.fileFormat(HudiScanPlanProvider.detectFileFormat(baseFile))
.partitionValues(
HudiScanPlanProvider.parsePartitionValues(
- stat.getPartitionPath(), partitionNames,
urlEncoded))
+ stat.getPartitionPath(), partitionNames,
urlEncoded, timestampPartitionKeys))
.build());
};
diff --git
a/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiConnectorMetadata.java
b/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiConnectorMetadata.java
index ec0d4bb8bd1..60f0a969b7d 100644
---
a/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiConnectorMetadata.java
@@ -224,6 +224,10 @@ public class HudiConnectorMetadata implements
ConnectorMetadata {
.inputFormat(tableInfo.getInputFormat())
.serdeLib(tableInfo.getSerializationLib())
.partitionKeyNames(partKeyNames)
+ .timestampPartitionKeys(tableInfo.getPartitionKeys().stream()
+ .filter(column ->
"DATETIMEV2".equals(column.getType().getTypeName())
+ ||
"TIMESTAMPTZ".equals(column.getType().getTypeName()))
+
.map(ConnectorColumn::getName).collect(Collectors.toSet()))
.tableParameters(tableInfo.getParameters())
.build());
}
@@ -297,7 +301,7 @@ public class HudiConnectorMetadata implements
ConnectorMetadata {
}
allPartPaths = hmsPartitionPaths(hudiHandle, allPartPaths);
matchedPartPaths = prunePartitionPaths(allPartPaths, partKeyNames,
partitionPredicates,
- urlEncodePartitioning(hudiHandle, allPartPaths));
+ urlEncodePartitioning(hudiHandle, allPartPaths),
hudiHandle.getTimestampPartitionKeys());
} else {
// non-hive-sync (Hudi default), or any pinned read: list the
RELATIVE storage paths from Hudi
// metadata -- the SAME source the unpruned scan
(resolvePartitions -> listAllPartitionPaths) uses --
@@ -315,7 +319,7 @@ public class HudiConnectorMetadata implements
ConnectorMetadata {
return Optional.empty();
}
matchedPartPaths = prunePartitionPaths(allPartPaths, partKeyNames,
partitionPredicates,
- listing.urlEncoded);
+ listing.urlEncoded,
hudiHandle.getTimestampPartitionKeys());
}
if (matchedPartPaths.size() == allPartPaths.size()) {
// No pruning effect
@@ -807,7 +811,8 @@ public class HudiConnectorMetadata implements
ConnectorMetadata {
// non-numeric pin, the day it became reachable.
List<String> paths = hmsPartitionPaths(handle, hmsNames);
PartitionState state = latestPartitionState(handle);
- return buildPartitionInfos(paths, partKeyNames,
state.lastModifiedMillis, state.urlEncoded);
+ return buildPartitionInfos(paths, partKeyNames,
state.lastModifiedMillis, state.urlEncoded,
+ handle.getTimestampPartitionKeys());
}
LOG.warn("hive-sync hudi table {}.{} has no HMS partitions; "
+ "falling back to hudi metadata partition listing",
@@ -837,7 +842,8 @@ public class HudiConnectorMetadata implements
ConnectorMetadata {
HudiScanPlanProvider.urlEncodePartitioning(metaClient));
});
List<ConnectorPartitionInfo> partitions =
- buildPartitionInfos(listing.rawPaths, partKeyNames,
listing.lastModifiedMillis, listing.urlEncoded);
+ buildPartitionInfos(listing.rawPaths, partKeyNames,
listing.lastModifiedMillis, listing.urlEncoded,
+ handle.getTimestampPartitionKeys());
return listing.pinned == null
? partitions
: restrictToPin(handle, partKeyNames, partitions,
listing.pinned, queryInstant, listing.urlEncoded);
@@ -921,7 +927,8 @@ public class HudiConnectorMetadata implements
ConnectorMetadata {
List<ConnectorPartitionInfo> partitions, PinnedListing pinned,
String queryInstant, boolean urlEncoded) {
Set<Map<String, String>> pinnedValues = new HashSet<>();
for (String rawPath : pinned.rawPaths) {
-
pinnedValues.add(HudiScanPlanProvider.parsePartitionValues(rawPath,
partKeyNames, urlEncoded));
+
pinnedValues.add(HudiScanPlanProvider.parsePartitionValues(rawPath,
partKeyNames, urlEncoded,
+ handle.getTimestampPartitionKeys()));
}
List<ConnectorPartitionInfo> atPin = new
ArrayList<>(pinnedValues.size());
for (ConnectorPartitionInfo partition : partitions) {
@@ -1002,11 +1009,17 @@ public class HudiConnectorMetadata implements
ConnectorMetadata {
*/
static List<ConnectorPartitionInfo> buildPartitionInfos(
List<String> rawPaths, List<String> partKeyNames, long
lastModifiedMillis, boolean urlEncoded) {
+ return buildPartitionInfos(rawPaths, partKeyNames, lastModifiedMillis,
urlEncoded, Collections.emptySet());
+ }
+
+ static List<ConnectorPartitionInfo> buildPartitionInfos(List<String>
rawPaths, List<String> partKeyNames,
+ long lastModifiedMillis, boolean urlEncoded, Set<String>
timestampPartitionKeys) {
List<ConnectorPartitionInfo> result = new ArrayList<>(rawPaths.size());
for (String rawPath : rawPaths) {
// Parse the unescaped values ONCE; render the hive-style name
from the SAME values so the name and
// the values map agree by construction (the name re-parses back
to these values in fe-core).
- Map<String, String> values =
HudiScanPlanProvider.parsePartitionValues(rawPath, partKeyNames, urlEncoded);
+ Map<String, String> values =
HudiScanPlanProvider.parsePartitionValues(rawPath, partKeyNames, urlEncoded,
+ timestampPartitionKeys);
String name =
HudiScanPlanProvider.renderHiveStylePartitionName(partKeyNames, values);
// Ordered values in partKeyNames (render) order; render/parse are
exact inverses, so this equals
// fe-core's legacy parse of `name`. Supplied so fe-core skips the
parse.
@@ -1477,10 +1490,15 @@ public class HudiConnectorMetadata implements
ConnectorMetadata {
*/
static List<String> prunePartitionPaths(List<String> allPartPaths,
List<String> partKeyNames, Map<String, List<String>> predicates,
boolean urlEncoded) {
+ return prunePartitionPaths(allPartPaths, partKeyNames, predicates,
urlEncoded, Collections.emptySet());
+ }
+
+ static List<String> prunePartitionPaths(List<String> allPartPaths,
List<String> partKeyNames,
+ Map<String, List<String>> predicates, boolean urlEncoded,
Set<String> timestampPartitionKeys) {
List<String> matched = new ArrayList<>();
for (String partPath : allPartPaths) {
Map<String, String> partValues =
HudiScanPlanProvider.parsePartitionValues(
- partPath, partKeyNames, urlEncoded);
+ partPath, partKeyNames, urlEncoded,
timestampPartitionKeys);
if (matchesPredicates(partValues, predicates)) {
matched.add(partPath);
}
diff --git
a/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiScanPlanProvider.java
b/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiScanPlanProvider.java
index 091d61f994a..be4e81dd629 100644
---
a/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiScanPlanProvider.java
@@ -73,6 +73,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
@@ -398,7 +399,8 @@ public class HudiScanPlanProvider implements
ConnectorScanPlanProvider {
IncrementalRelation relation =
buildIncrementalRelation(metaClient, conf, hudiHandle, isCow);
Optional<List<ConnectorScanRange>> incrementalRanges =
incrementalRanges(relation, isCow, forceJni,
basePath, inputFormat, serdeLib, columnNames, columnTypes,
partitionFieldNames(metaClient),
- this::normalizeNativeUri,
urlEncodePartitioning(metaClient));
+ this::normalizeNativeUri,
urlEncodePartitioning(metaClient),
+ hudiHandle.getTimestampPartitionKeys());
if (incrementalRanges.isPresent()) {
LOG.info("Hudi incremental scan planning: {}.{} window=({},
{}] splits={}",
hudiHandle.getDbName(), hudiHandle.getTableName(),
@@ -423,7 +425,8 @@ public class HudiScanPlanProvider implements
ConnectorScanPlanProvider {
boolean urlEncoded = urlEncodePartitioning(metaClient);
for (String partitionPath : partitionPaths) {
Map<String, String> partValues = parsePartitionValues(
- partitionPath, hudiHandle.getPartitionKeyNames(),
urlEncoded);
+ partitionPath, hudiHandle.getPartitionKeyNames(),
urlEncoded,
+ hudiHandle.getTimestampPartitionKeys());
if (useNativeCowPath) {
collectCowSplits(fsView, partitionPath, queryInstant,
@@ -768,7 +771,7 @@ public class HudiScanPlanProvider implements
ConnectorScanPlanProvider {
try {
return isCow
? new COWIncrementalRelation(metaClient, conf,
handle.getBeginInstant(),
- handle.getEndInstant(), policy, optParams)
+ handle.getEndInstant(), policy, optParams,
handle.getTimestampPartitionKeys())
: new MORIncrementalRelation(metaClient, conf,
handle.getBeginInstant(),
handle.getEndInstant(), policy, optParams);
} catch (IOException e) {
@@ -803,6 +806,14 @@ public class HudiScanPlanProvider implements
ConnectorScanPlanProvider {
boolean forceJni, String basePath, String inputFormat, String
serdeLib,
List<String> columnNames, List<String> columnTypes, List<String>
partitionFieldNames,
UnaryOperator<String> nativePathNormalizer, boolean urlEncoded) {
+ return incrementalRanges(relation, isCow, forceJni, basePath,
inputFormat, serdeLib, columnNames,
+ columnTypes, partitionFieldNames, nativePathNormalizer,
urlEncoded, Collections.emptySet());
+ }
+
+ static Optional<List<ConnectorScanRange>>
incrementalRanges(IncrementalRelation relation, boolean isCow,
+ boolean forceJni, String basePath, String inputFormat, String
serdeLib,
+ List<String> columnNames, List<String> columnTypes, List<String>
partitionFieldNames,
+ UnaryOperator<String> nativePathNormalizer, boolean urlEncoded,
Set<String> timestampPartitionKeys) {
if (relation.fallbackFullTableScan()) {
return Optional.empty();
}
@@ -816,7 +827,7 @@ public class HudiScanPlanProvider implements
ConnectorScanPlanProvider {
String endTs = relation.getEndTs();
for (FileSlice fileSlice : relation.collectFileSlices()) {
Map<String, String> partValues = parsePartitionValues(
- fileSlice.getPartitionPath(), partitionFieldNames,
urlEncoded);
+ fileSlice.getPartitionPath(), partitionFieldNames,
urlEncoded, timestampPartitionKeys);
// @incr lists the LATEST schema (no per-file schema_id dict on
the incremental path) -> null resolver.
ranges.add(buildMorRange(fileSlice, partValues, endTs, forceJni,
basePath, inputFormat, serdeLib, columnNames, columnTypes,
null, nativePathNormalizer));
@@ -1085,8 +1096,8 @@ public class HudiScanPlanProvider implements
ConnectorScanPlanProvider {
/**
* Parse a Hudi partition's relative path into a column→value map.
Handles BOTH hive-style
* ("year=2024/month=01") and Hudi's
- * DEFAULT non-hive-style POSITIONAL ("2024/01") layouts, decoding values
only when the table enables
- * URL-encoded partition paths:
+ * DEFAULT non-hive-style POSITIONAL ("2024/01") layouts, decoding values
when the table enables
+ * URL-encoded partition paths or the column is a timestamp:
* <ul>
* <li>A fragment carrying the "col=" prefix contributes the suffix; a
fragment WITHOUT it is mapped
* POSITIONALLY to the i-th partition column. The old split-on-'='
logic silently DROPPED a
@@ -1103,10 +1114,16 @@ public class HudiScanPlanProvider implements
ConnectorScanPlanProvider {
* <p>Static + package-private for direct unit testing (no live
HoodieTableMetaClient needed).
*
* <p>The encoding flag comes from hoodie.properties, not the presence of
percent characters. With
- * encoding disabled (Hudi's default), "%3A" is a literal value distinct
from ":".
+ * encoding disabled (Hudi's default), "%3A" in a STRING is a literal
value distinct from ":".
+ * Spark SQL timestamp paths are always escaped, so callers supply their
schema-derived column names.
*/
static Map<String, String> parsePartitionValues(
String partitionPath, List<String> partKeyNames, boolean
urlEncoded) {
+ return parsePartitionValues(partitionPath, partKeyNames, urlEncoded,
Collections.emptySet());
+ }
+
+ static Map<String, String> parsePartitionValues(String partitionPath,
List<String> partKeyNames,
+ boolean urlEncoded, Set<String> timestampPartitionKeys) {
if (partKeyNames == null || partKeyNames.isEmpty()) {
// Non-partitioned table (legacy returns an empty value list). The
unpartitioned scan path always
// reaches here with an empty key list, so an empty partitionPath
needs no separate guard.
@@ -1119,7 +1136,8 @@ public class HudiScanPlanProvider implements
ConnectorScanPlanProvider {
String prefix = partKeyNames.get(0) + "=";
String value = partitionPath.startsWith(prefix)
? partitionPath.substring(prefix.length()) :
partitionPath;
- values.put(partKeyNames.get(0), urlEncoded ?
unescapePathName(value) : value);
+ values.put(partKeyNames.get(0), decodePartitionValue(value,
partKeyNames.get(0),
+ urlEncoded, timestampPartitionKeys));
return values;
}
throw new DorisConnectorException(
@@ -1129,11 +1147,19 @@ public class HudiScanPlanProvider implements
ConnectorScanPlanProvider {
String prefix = partKeyNames.get(i) + "=";
String raw = fragments[i].startsWith(prefix)
? fragments[i].substring(prefix.length()) : fragments[i];
- values.put(partKeyNames.get(i), urlEncoded ? unescapePathName(raw)
: raw);
+ values.put(partKeyNames.get(i), decodePartitionValue(raw,
partKeyNames.get(i),
+ urlEncoded, timestampPartitionKeys));
}
return values;
}
+ private static String decodePartitionValue(String value, String name,
boolean urlEncoded,
+ Set<String> timestampPartitionKeys) {
+ // Spark SQL's SqlKeyGenerator always escapes formatted TIMESTAMP
values, independently of the
+ // table URL-encoding flag. Decode these once, while preserving
literal "%3A" in unencoded STRINGs.
+ return urlEncoded || timestampPartitionKeys.contains(name) ?
unescapePathName(value) : value;
+ }
+
/**
* How hudi spells a NULL partition value in a directory name.
Byte-frozen: it is what this connector has
* always SENT in columns_from_path, and it is also accepted as an INPUT
spelling.
diff --git
a/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiTableHandle.java
b/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiTableHandle.java
index c8e7621e6f5..41c7e84478c 100644
---
a/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiTableHandle.java
+++
b/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiTableHandle.java
@@ -20,8 +20,10 @@ package org.apache.doris.connector.hudi;
import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* Opaque table handle for a Hudi table, carrying database/table coordinates,
@@ -44,6 +46,7 @@ public class HudiTableHandle implements ConnectorTableHandle {
private final String inputFormat;
private final String serdeLib;
private final List<String> partitionKeyNames;
+ private final Set<String> timestampPartitionKeys;
private final Map<String, String> tableParameters;
// Set after applyFilter for partition pruning
@@ -80,6 +83,7 @@ public class HudiTableHandle implements ConnectorTableHandle {
this.partitionKeyNames = builder.partitionKeyNames != null
? Collections.unmodifiableList(builder.partitionKeyNames)
: Collections.emptyList();
+ this.timestampPartitionKeys = Collections.unmodifiableSet(new
HashSet<>(builder.timestampPartitionKeys));
this.tableParameters = builder.tableParameters != null
? Collections.unmodifiableMap(builder.tableParameters)
: Collections.emptyMap();
@@ -125,6 +129,10 @@ public class HudiTableHandle implements
ConnectorTableHandle {
return partitionKeyNames;
}
+ public Set<String> getTimestampPartitionKeys() {
+ return timestampPartitionKeys;
+ }
+
public Map<String, String> getTableParameters() {
return tableParameters;
}
@@ -165,6 +173,7 @@ public class HudiTableHandle implements
ConnectorTableHandle {
b.inputFormat = this.inputFormat;
b.serdeLib = this.serdeLib;
b.partitionKeyNames = this.partitionKeyNames;
+ b.timestampPartitionKeys = this.timestampPartitionKeys;
b.tableParameters = this.tableParameters;
b.prunedPartitionPaths = this.prunedPartitionPaths;
b.queryInstant = this.queryInstant;
@@ -191,6 +200,7 @@ public class HudiTableHandle implements
ConnectorTableHandle {
private String inputFormat;
private String serdeLib;
private List<String> partitionKeyNames;
+ private Set<String> timestampPartitionKeys = Collections.emptySet();
private Map<String, String> tableParameters;
private List<String> prunedPartitionPaths;
private String queryInstant;
@@ -220,6 +230,11 @@ public class HudiTableHandle implements
ConnectorTableHandle {
return this;
}
+ public Builder timestampPartitionKeys(Set<String> val) {
+ this.timestampPartitionKeys = val;
+ return this;
+ }
+
public Builder tableParameters(Map<String, String> val) {
this.tableParameters = val;
return this;
diff --git
a/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiConnectorPartitionListingTest.java
b/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiConnectorPartitionListingTest.java
index 67260d13a8f..8bef7f3c473 100644
---
a/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiConnectorPartitionListingTest.java
+++
b/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiConnectorPartitionListingTest.java
@@ -24,6 +24,7 @@ import org.apache.doris.connector.hms.HmsClientException;
import org.apache.doris.connector.hms.HmsDatabaseInfo;
import org.apache.doris.connector.hms.HmsPartitionInfo;
import org.apache.doris.connector.hms.HmsTableInfo;
+import org.apache.doris.connector.spi.ConnectorColumn;
import org.apache.doris.connector.spi.ConnectorPartitionInfo;
import org.apache.doris.connector.spi.ConnectorType;
import org.apache.doris.connector.spi.DorisConnectorException;
@@ -37,6 +38,7 @@ import
org.apache.doris.connector.spi.pushdown.ConnectorLiteral;
import org.apache.hadoop.conf.Configuration;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.PartitionPathEncodeUtils;
import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -77,6 +79,100 @@ public class HudiConnectorPartitionListingTest {
// ── item 1: instant string → long (pure)
──────────────────────────────────────────────────────────
+ @Test
+ public void
sparkSqlTimestampPartitionsDecodeWithoutChangingStringPartitions() {
+ String timestamp = "2024-01-01 10:00:00";
+ // SqlKeyGenerator escapes TIMESTAMP values even when URL encoding is
disabled for STRING values.
+ String physical = "event_time=" +
PartitionPathEncodeUtils.escapePathName(timestamp) + "/label=%3A";
+ RecordingHmsClient hms = new RecordingHmsClient(null);
+ hms.tableInfo = timestampTable("s3://b/t");
+ HudiConnectorMetadata md = metadata(false, hms,
+ stub(HudiConnectorMetadata.unpinnedListing(99,
Collections.singletonList(physical), false)));
+ HudiTableHandle handle = (HudiTableHandle) md.getTableHandle(null,
"db", "t").get();
+ ConnectorPartitionInfo partition = md.listPartitions(null, handle,
Optional.empty()).get(0);
+ Assertions.assertEquals(timestamp,
partition.getPartitionValues().get("event_time"));
+ Assertions.assertEquals("%3A",
partition.getPartitionValues().get("label"));
+ Assertions.assertEquals(Arrays.asList(timestamp, "%3A"),
partition.getOrderedPartitionValues());
+ Assertions.assertEquals(partition.getPartitionValues(),
+
HudiConnectorMetadata.parsePartitionName(partition.getPartitionName(),
handle.getPartitionKeyNames()));
+ }
+
+ @Test
+ public void
sparkSqlTimestampHmsListingAndPruningRespectPersistedUrlFlags(@TempDir Path
warehouse)
+ throws Exception {
+ String first = "2024-01-01 10:00:00";
+ String second = "2024-01-02 11:00:00";
+ for (Boolean encoded : Arrays.asList(null, false, true)) {
+ String basePath =
warehouse.resolve(String.valueOf(encoded)).toUri().toString();
+ HoodieTableMetaClient.TableBuilder builder =
HoodieTableMetaClient.newTableBuilder()
+
.setTableType(HoodieTableType.COPY_ON_WRITE).setTableName("t")
+ .setPartitionFields("event_time,label");
+ if (encoded != null) {
+ builder.setUrlEncodePartitioning(encoded);
+ }
+ builder.initTable(new HadoopStorageConfiguration(new
Configuration()), basePath);
+ String label = Boolean.TRUE.equals(encoded) ? "%253A" : "%3A";
+ List<String> paths = Arrays.asList(
+ "event_time=" +
PartitionPathEncodeUtils.escapePathName(first) + "/label=" + label,
+ "event_time=" +
PartitionPathEncodeUtils.escapePathName(second) + "/label=" + label);
+ List<String> names = paths.stream().map(path -> path.replace("%",
"%25")).collect(Collectors.toList());
+ RecordingHmsClient hms = new RecordingHmsClient(names);
+ hms.tableInfo = timestampTable(basePath);
+ hms.locations = Map.of(names.get(0), basePath + "/" + paths.get(0),
+ names.get(1), basePath + "/" + paths.get(1));
+ HudiConnectorMetadata md = new HudiConnectorMetadata(hms,
+
HudiTestProperties.with(HudiCatalogProperties.USE_HIVE_SYNC_PARTITION, "true"),
+ HudiMetaClientExecutor.inline());
+ HudiTableHandle handle = (HudiTableHandle) md.getTableHandle(null,
"db", "t").get();
+ List<ConnectorPartitionInfo> partitions = md.listPartitions(null,
handle, Optional.empty());
+ Assertions.assertEquals(Arrays.asList(first, second),
partitions.stream()
+ .map(partition ->
partition.getPartitionValues().get("event_time")).collect(Collectors.toList()));
+ Assertions.assertEquals(Arrays.asList("%3A", "%3A"),
partitions.stream()
+ .map(partition ->
partition.getPartitionValues().get("label")).collect(Collectors.toList()));
+ ConnectorComparison predicate = new
ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("event_time",
ConnectorType.of("DATETIMEV2", 6, -1)),
+ new ConnectorLiteral(ConnectorType.of("DATETIMEV2", 6, -1),
+ LocalDateTime.of(2024, 1, 1, 10, 0)));
+ HudiTableHandle filtered = (HudiTableHandle) md.applyFilter(null,
handle,
+ new
ConnectorFilterConstraint(predicate)).get().getHandle();
+ Assertions.assertEquals(Collections.singletonList(paths.get(0)),
filtered.getPrunedPartitionPaths());
+ Assertions.assertEquals(Collections.singleton("event_time"),
filtered.getTimestampPartitionKeys());
+ }
+ }
+
+ @Test
+ public void
sparkSqlTimestampMetadataPruningAndHistoricalListingKeepPhysicalPaths() {
+ List<String> paths = Arrays.asList("event_time=2024-01-01
10%3A00%3A00/label=%3A",
+ "event_time=2024-01-02 11%3A00%3A00/label=%3A");
+ RecordingHmsClient hms = new RecordingHmsClient(null);
+ hms.tableInfo = timestampTable("s3://b/t");
+ HudiConnectorMetadata current = metadata(false, hms,
+ stub(HudiConnectorMetadata.unpinnedListing(99, paths, false)));
+ HudiTableHandle handle = (HudiTableHandle)
current.getTableHandle(null, "db", "t").get();
+ ConnectorComparison predicate = new
ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("event_time",
ConnectorType.of("DATETIMEV2", 6, -1)),
+ new ConnectorLiteral(ConnectorType.of("DATETIMEV2", 6, -1),
LocalDateTime.of(2024, 1, 1, 10, 0)));
+ HudiTableHandle filtered = (HudiTableHandle) current.applyFilter(null,
handle,
+ new ConnectorFilterConstraint(predicate)).get().getHandle();
+ Assertions.assertEquals(Collections.singletonList(paths.get(0)),
filtered.getPrunedPartitionPaths());
+ HudiConnectorMetadata historical = metadata(false, hms,
+ stub(HudiConnectorMetadata.pinnedListing(99, paths, 7,
+ Collections.singletonList(paths.get(0)), false)));
+ List<ConnectorPartitionInfo> partitions =
historical.listPartitions(null,
+ handle.toBuilder().queryInstant("20240101120000000").build(),
Optional.empty());
+ Assertions.assertEquals(1, partitions.size());
+ Assertions.assertEquals(Arrays.asList("2024-01-01 10:00:00", "%3A"),
+ partitions.get(0).getOrderedPartitionValues());
+ }
+
+ private static HmsTableInfo timestampTable(String basePath) {
+ return
HmsTableInfo.builder().dbName("db").tableName("t").location(basePath)
+ .partitionKeys(Arrays.asList(
+ new ConnectorColumn("event_time",
ConnectorType.of("DATETIMEV2", 6, -1), null, true, null),
+ new ConnectorColumn("label",
ConnectorType.of("STRING"), null, true, null)))
+ .build();
+ }
+
@Test
public void
metadataListingAndTimeTravelKeepLiteralPercentPartitionsDistinct() {
for (boolean encoded : Arrays.asList(false, true)) {
@@ -695,6 +791,7 @@ public class HudiConnectorPartitionListingTest {
/** Records partition lookups and supplies locations; unrelated HMS
operations fail loudly. */
private static final class RecordingHmsClient implements HmsClient {
private final List<String> partitionNames;
+ private HmsTableInfo tableInfo;
private Map<String, String> locations = Collections.emptyMap();
private List<String> deletedNames = Collections.emptyList();
private int getPartitionsCalls;
@@ -730,12 +827,15 @@ public class HudiConnectorPartitionListingTest {
@Override
public boolean tableExists(String dbName, String tableName) {
- throw new UnsupportedOperationException();
+ return tableInfo != null;
}
@Override
public HmsTableInfo getTable(String dbName, String tableName) {
- throw new UnsupportedOperationException();
+ if (tableInfo == null) {
+ throw new UnsupportedOperationException("No table metadata
configured");
+ }
+ return tableInfo;
}
@Override
diff --git
a/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiIncrementalPlanScanTest.java
b/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiIncrementalPlanScanTest.java
index a9bb6b92c76..7d5fe83b534 100644
---
a/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiIncrementalPlanScanTest.java
+++
b/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiIncrementalPlanScanTest.java
@@ -18,12 +18,27 @@
package org.apache.doris.connector.hudi;
import org.apache.doris.connector.spi.scan.ConnectorScanRange;
+import org.apache.doris.thrift.TFileRangeDesc;
+import org.apache.doris.thrift.TTableFormatFileDesc;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import
org.apache.hudi.common.table.timeline.TimelineUtils.HollowCommitHandling;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -33,18 +48,13 @@ import java.util.Optional;
import java.util.function.UnaryOperator;
/**
- * Tests the OFFLINE-verifiable core of the INC-3 incremental {@code planScan}
wiring: the pure static helpers
- * {@link HudiScanPlanProvider#incrementalRanges} (COW/MOR routing + fallback
degrade) and
- * {@link HudiScanPlanProvider#buildMorRange} (file-slice → JNI range
mapping). Both are exercised without a
- * live {@code HoodieTableMetaClient} — {@code incrementalRanges} via a FAKE
{@link IncrementalRelation}, and
- * {@code buildMorRange} via a hand-built {@link FileSlice}.
+ * Tests incremental scan-range mapping through both a real local Hudi commit
and recording fake relations.
+ * Covers {@link HudiScanPlanProvider#incrementalRanges} (COW/MOR routing and
fallback) and
+ * {@link HudiScanPlanProvider#buildMorRange} (file-slice to native/JNI range
mapping).
*
- * <p><b>Coverage scope (Rule 12 — no over-claim).</b> These cover the routing
+ mapping DECISIONS only. Building
- * the real relation (which does eager timeline/metadata/filesystem I/O) and
the actual file SELECTION
- * (commit-range, write-stat mapping, {@code fs.exists} full-table-scan
probes, the completion-time END axis) are
- * e2e-only (design §5) — same boundary as {@link
HudiIncrementalRelationTest}. Row-level filtering of the read to
- * the {@code (begin, end]} window is a LATER step (an FE-side synthetic
{@code _hoodie_commit_time} predicate),
- * NOT this file-selection step.
+ * <p>The COW timestamp regression persists a local SDK commit and exercises
the real relation's write-stat
+ * mapping. Other selection scenarios use a fake relation; these tests do not
decode Parquet data or exercise
+ * a remote Hudi deployment. Incremental row-level filtering is outside the
scan-range mapping tested here.
*/
public class HudiIncrementalPlanScanTest {
@@ -56,6 +66,65 @@ public class HudiIncrementalPlanScanTest {
// ── fallback → degrade (Optional.empty), collect* NOT called
─────────────────────────────────────────────
+ @Test
+ public void
cowIncrementalTimestampDecodingPreservesCommittedFilePath(@TempDir Path
warehouse) throws Exception {
+ Configuration configuration = new Configuration();
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.newTableBuilder()
+ .setTableType(HoodieTableType.COPY_ON_WRITE).setTableName("t")
+
.setPartitionFields("event_time,label").setUrlEncodePartitioning(false)
+ .initTable(new HadoopStorageConfiguration(configuration),
warehouse.toUri().toString());
+ String partition = "event_time=2024-01-01 10%3A00%3A00/label=%3A";
+ String instantTime = "20240101120000000";
+ String relativeFile = partition + "/fileid_0_" + instantTime +
".parquet";
+ HoodieWriteStat stat = new HoodieWriteStat();
+ stat.setFileId("fileid");
+ stat.setPartitionPath(partition);
+ stat.setPath(relativeFile);
+ stat.setFileSizeInBytes(128);
+ HoodieCommitMetadata metadata = new HoodieCommitMetadata();
+ metadata.addWriteStat(partition, stat);
+ HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
+ HoodieInstant requested =
metaClient.createNewInstant(HoodieInstant.State.REQUESTED,
+ HoodieTimeline.COMMIT_ACTION, instantTime);
+ timeline.createNewInstant(requested);
+ timeline.transitionRequestedToInflight(requested, Option.empty());
+
timeline.saveAsComplete(metaClient.createNewInstant(HoodieInstant.State.INFLIGHT,
+ HoodieTimeline.COMMIT_ACTION, instantTime),
Option.of(metadata));
+ metaClient.reloadActiveTimeline();
+ COWIncrementalRelation relation = new
COWIncrementalRelation(metaClient, configuration,
+ "20240101000000000", instantTime, HollowCommitHandling.FAIL,
+ Collections.emptyMap(), Collections.singleton("event_time"));
+ List<HudiScanRange> ranges =
relation.collectSplits(UnaryOperator.identity());
+ Assertions.assertEquals(1, ranges.size());
+ HudiScanRange range = ranges.get(0);
+ Assertions.assertTrue(range.getPath().get().endsWith(relativeFile));
+ TFileRangeDesc descriptor = new TFileRangeDesc();
+ range.populateRangeParams(new TTableFormatFileDesc(), descriptor);
+ Assertions.assertEquals(Arrays.asList("event_time", "label"),
descriptor.getColumnsFromPathKeys());
+ Assertions.assertEquals(Arrays.asList("2024-01-01 10:00:00", "%3A"),
descriptor.getColumnsFromPath());
+ }
+
+ @Test
+ public void morIncrementalTimestampDecodingReachesNativeAndJniScanRanges()
{
+ String path = "event_time=2024-01-01 10%3A00%3A00/label=%3A";
+ String file = BASE_PATH + "/" + path +
"/fileid_0_20240101000000.parquet";
+ for (boolean forceJni : Arrays.asList(false, true)) {
+ FakeRelation mor = new FakeRelation();
+ mor.fileSlices = Collections.singletonList(baseOnlySlice(path,
file));
+ HudiScanRange range = (HudiScanRange)
HudiScanPlanProvider.incrementalRanges(
+ mor, false, forceJni, BASE_PATH, INPUT_FORMAT, SERDE,
+ Arrays.asList("event_time", "label"),
Arrays.asList("timestamp", "string"),
+ Arrays.asList("event_time", "label"),
UnaryOperator.identity(), false,
+ Collections.singleton("event_time")).get().get(0);
+ TFileRangeDesc descriptor = new TFileRangeDesc();
+ range.populateRangeParams(new TTableFormatFileDesc(), descriptor);
+ Assertions.assertEquals(Arrays.asList("event_time", "label"),
descriptor.getColumnsFromPathKeys());
+ Assertions.assertEquals(Arrays.asList("2024-01-01 10:00:00",
"%3A"), descriptor.getColumnsFromPath());
+ Assertions.assertEquals(Arrays.asList(false, false),
descriptor.getColumnsFromPathIsNull());
+ Assertions.assertEquals(file, range.getPath().get());
+ }
+ }
+
@Test
public void morIncrementalRespectsPhysicalPartitionEncoding() {
for (boolean encoded : Arrays.asList(false, true)) {
diff --git
a/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiPartitionValuesTest.java
b/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiPartitionValuesTest.java
index ce946dd6ab9..5ff6834c5f6 100644
---
a/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiPartitionValuesTest.java
+++
b/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiPartitionValuesTest.java
@@ -19,6 +19,7 @@ package org.apache.doris.connector.hudi;
import org.apache.doris.connector.spi.DorisConnectorException;
+import org.apache.hudi.common.util.PartitionPathEncodeUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -33,6 +34,34 @@ import java.util.Map;
*/
public class HudiPartitionValuesTest {
+ @Test
+ public void timestampDecodingIsPerColumnAndConsumesOnlyOneEscapingLayer() {
+ String timestamp = "2024-01-01 10:00:00";
+ String timestampPath =
PartitionPathEncodeUtils.escapePathName(timestamp);
+ for (boolean urlEncoded : Arrays.asList(false, true)) {
+ for (boolean hiveStyle : Arrays.asList(false, true)) {
+ for (String label : Arrays.asList("%3A", "%253A", "a+b",
"a%2Fb")) {
+ String labelPath = urlEncoded ?
PartitionPathEncodeUtils.escapePathName(label) : label;
+ String path = (hiveStyle ? "event_time=" : "") +
timestampPath + "/"
+ + (hiveStyle ? "label=" : "") + labelPath;
+ Map<String, String> values =
HudiScanPlanProvider.parsePartitionValues(path,
+ Arrays.asList("event_time", "label"), urlEncoded,
Collections.singleton("event_time"));
+ Assertions.assertEquals(timestamp,
values.get("event_time"));
+ Assertions.assertEquals(label, values.get("label"));
+ }
+ }
+ }
+ Assertions.assertEquals("2024-01-01 10%3A00%3A00",
HudiScanPlanProvider.parsePartitionValues(
+ "event_time=2024-01-01 10%253A00%253A00",
Collections.singletonList("event_time"),
+ false, Collections.singleton("event_time")).get("event_time"));
+ Assertions.assertEquals("2024-01-01 10:00:00",
HudiScanPlanProvider.parsePartitionValues(
+ "event_time=" + timestamp,
Collections.singletonList("event_time"),
+ false, Collections.singleton("event_time")).get("event_time"));
+ Assertions.assertEquals("__HIVE_DEFAULT_PARTITION__",
HudiScanPlanProvider.parsePartitionValues(
+ "event_time=__HIVE_DEFAULT_PARTITION__",
Collections.singletonList("event_time"),
+ false, Collections.singleton("event_time")).get("event_time"));
+ }
+
@Test
public void rawAndEncodedValuesHaveTheSameLogicalIdentity() {
for (String value : Arrays.asList(":", "%3A", "%253A", "a/b", "a+b",
"a%2Fb", "bad%2", "你好")) {
diff --git
a/regression-test/suites/external_table_p2/hudi/test_hudi_partition_prune.groovy
b/regression-test/suites/external_table_p2/hudi/test_hudi_partition_prune.groovy
index a74f34aca05..961aaf69c33 100644
---
a/regression-test/suites/external_table_p2/hudi/test_hudi_partition_prune.groovy
+++
b/regression-test/suites/external_table_p2/hudi/test_hudi_partition_prune.groovy
@@ -385,6 +385,18 @@ suite("test_hudi_partition_prune", "p2,external") {
contains "partition=1/2"
}
+ // Spark SQL escapes timestamp partition values independently of the
table URL-encoding flag.
+ // Keep the golden result above and also compare both readers under
each partition-listing mode.
+ def originalForceJni = sql("SHOW VARIABLES LIKE
'force_jni_scanner'")[0][1]
+ try {
+ sql """set force_jni_scanner = false"""
+ def nativeTimestampRows = sql(one_partition_timestamp)
+ sql """set force_jni_scanner = true"""
+ assertEquals(nativeTimestampRows, sql(one_partition_timestamp))
+ } finally {
+ sql """set force_jni_scanner = ${originalForceJni}"""
+ }
+
// A partition that was dropped after the instant being read.
//
// Every other table here only ever gains partitions, so "the
partitions now" and "the
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
