github-actions[bot] commented on code in PR #66481:
URL: https://github.com/apache/doris/pull/66481#discussion_r3773190473
##########
fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiConnector.java:
##########
@@ -140,7 +140,7 @@ public boolean ownsHandle(ConnectorTableHandle handle) {
@Override
public ConnectorScanPlanProvider getScanPlanProvider() {
- return new HudiScanPlanProvider(properties, context);
+ return new HudiScanPlanProvider(properties, context,
this::getOrCreateClient);
Review Comment:
[P1] Forward partition invalidation to the Hudi HMS cache
This provider now reads Hudi partition membership and objects through the
connector's `CachingHmsClient`, but `HudiConnector` still inherits the no-op
`Connector.invalidatePartition`. HMS event sync calls the Hive connector, whose
implementation forwards partition invalidation to built siblings; the Hudi
sibling therefore keeps its partition-name/object entries for the default
86400-second TTL. After an add/drop/alter, an unpruned Hive-Sync scan can omit
or retain partitions even though the event was processed. Override
`invalidatePartition` here (targeted `invalidatePartitions` or a
correctness-safe table flush) and add a warmed-cache event regression.
##########
fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiScanPlanProvider.java:
##########
@@ -597,80 +626,324 @@ private static IncrementalRelation
buildIncrementalRelation(HoodieTableMetaClien
* <ul>
* <li>{@code relation.fallbackFullTableScan()} (an archived instant /
missing file) →
* {@link Optional#empty()} = degrade to the latest-snapshot scan
(NOT an error), legacy {@code :470}.</li>
- * <li>COW → {@link IncrementalRelation#collectSplits()} yields
native ranges directly.
+ * <li>COW → {@link IncrementalRelation#collectSplits(Function,
UnaryOperator)} yields native ranges
+ * directly.
* <b>{@code force_jni} is intentionally IGNORED for a COW
incremental read</b> (it always reads native)
* — a signed, deliberate deviation from legacy, which routes
{@code force_jni}+COW to the MOR-style
* branch and calls {@code collectFileSlices()} on a COW relation
→ {@code UnsupportedOperationException}
* (a latent legacy crash). Routing on the relation type never calls
the unsupported shape.</li>
* <li>MOR → {@link IncrementalRelation#collectFileSlices()} (a
FLAT cross-partition slice list) turned
* into JNI ranges at the resolved window END ({@code
relation.getEndTs()}), with per-slice partition
- * values parsed from the slice's own partition path against the
Hudi table-config partition fields
- * (the same non-handle source the COW relation uses). {@code
force_jni} still keeps a no-log MOR slice
- * on JNI via {@link #buildMorRange}.</li>
+ * values resolved from the slice's own partition path. For Hive
Sync tables the resolver is built from
+ * HMS locations and extractor-produced logical values; otherwise it
parses the Hudi physical layout.
+ * {@code force_jni} still keeps a no-log MOR slice on JNI via
{@link #buildMorRange}.</li>
* </ul>
* Package-private static, pure over the {@link IncrementalRelation}
contract, so file-selection routing +
* the degrade decision are unit-testable with a fake relation (no live
metaClient).
*/
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,
+ List<String> columnNames, List<String> columnTypes,
+ Supplier<Function<String, Map<String, String>>>
partitionValueResolverSupplier,
UnaryOperator<String> nativePathNormalizer) {
if (relation.fallbackFullTableScan()) {
return Optional.empty();
}
+ Function<String, Map<String, String>> partitionValueResolver =
+ new LazyPartitionValueResolver(partitionValueResolverSupplier);
List<ConnectorScanRange> ranges = new ArrayList<>();
if (isCow) {
// COW @incr yields native ranges directly; normalize their scheme
(s3a->s3) for BE's native reader
// (COWIncrementalRelation.collectSplits builds .path() from the
raw HMS base path).
- ranges.addAll(relation.collectSplits(nativePathNormalizer));
+ ranges.addAll(relation.collectSplits(partitionValueResolver,
nativePathNormalizer));
return Optional.of(ranges);
}
String endTs = relation.getEndTs();
for (FileSlice fileSlice : relation.collectFileSlices()) {
- Map<String, String> partValues =
parsePartitionValues(fileSlice.getPartitionPath(), partitionFieldNames);
+ Map<String, String> partValues =
partitionValueResolver.apply(fileSlice.getPartitionPath());
// @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));
}
return Optional.of(ranges);
}
+ private Function<String, Map<String, String>>
incrementalPartitionValueResolver(
+ HudiTableHandle handle, HoodieTableMetaClient metaClient, boolean
hiveStylePartitioning) {
+ return incrementalPartitionValueResolver(
+ partitionFieldNames(metaClient), handle.getPartitionKeyNames(),
+ useHiveSyncPartition(), hiveStylePartitioning,
+ () -> listHiveSyncPartitions(handle));
+ }
+
+ static Function<String, Map<String, String>>
incrementalPartitionValueResolver(
+ List<String> tableConfigPartitionFields, List<String>
hmsPartitionFields,
+ boolean useHiveSyncPartition, boolean hiveStylePartitioning,
+ Supplier<Optional<List<PartitionScanInfo>>>
hiveSyncPartitionsSupplier) {
+ List<String> fields = tableConfigPartitionFields.isEmpty()
+ ? hmsPartitionFields : tableConfigPartitionFields;
+ Function<String, Map<String, String>> physicalPathResolver =
+ path -> parsePartitionValues(path, fields,
hiveStylePartitioning);
+ if (!useHiveSyncPartition || hmsPartitionFields.isEmpty()) {
+ return physicalPathResolver;
+ }
+ Optional<List<PartitionScanInfo>> hiveSyncPartitions =
hiveSyncPartitionsSupplier.get();
Review Comment:
[P2] Bound incremental partition lookup to the selected window
The first selected incremental path calls this supplier, which lists every
HMS partition with `-1`, fetches every object in one `getPartitionsByNames`
request, and copies the whole result into another map. A one-partition
`(begin,end]` window is therefore O(total table partitions); beyond the
100000-entry object cache this also repeats oversized metastore requests. The
previous path parsed only selected write-stat/file-slice paths. Resolve
selected paths on demand or use bounded batching, and test that a narrow window
does not request all table partitions.
##########
be/src/exec/sink/viceberg_delete_sink.cpp:
##########
@@ -681,8 +681,8 @@ Status VIcebergDeleteSink::_write_deletion_vector_files(
commit_data.__set_content_offset(blob.content_offset);
commit_data.__set_content_size_in_bytes(blob.content_size_in_bytes);
commit_data.__set_referenced_data_file_path(blob.referenced_data_file);
- if (blob.partition_spec_id != 0 || !blob.partition_data_json.empty()) {
- commit_data.__set_partition_spec_id(blob.partition_spec_id);
+ commit_data.__set_partition_spec_id(blob.partition_spec_id);
Review Comment:
[P1] Preserve spec-id presence across rolling upgrades
This unconditional set treats numeric zero as a genuine spec, but the BE
reader still maps an absent old-FE thrift field to `0` in `$row_id`; a new BE
therefore turns absence into explicit spec 0 and can select an old evolved
spec. In the reverse direction, the old BE condition removed here drops a
genuine new-FE spec 0 when partition JSON is empty, so a new FE can reject or
mis-handle the commit when the current spec is partitioned. Preserve presence
through `$row_id`/the sink and add a compatibility fence plus old-FE/new-BE and
new-FE/old-BE evolved-spec regressions.
##########
fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiScanPlanProvider.java:
##########
@@ -597,80 +626,324 @@ private static IncrementalRelation
buildIncrementalRelation(HoodieTableMetaClien
* <ul>
* <li>{@code relation.fallbackFullTableScan()} (an archived instant /
missing file) →
* {@link Optional#empty()} = degrade to the latest-snapshot scan
(NOT an error), legacy {@code :470}.</li>
- * <li>COW → {@link IncrementalRelation#collectSplits()} yields
native ranges directly.
+ * <li>COW → {@link IncrementalRelation#collectSplits(Function,
UnaryOperator)} yields native ranges
+ * directly.
* <b>{@code force_jni} is intentionally IGNORED for a COW
incremental read</b> (it always reads native)
* — a signed, deliberate deviation from legacy, which routes
{@code force_jni}+COW to the MOR-style
* branch and calls {@code collectFileSlices()} on a COW relation
→ {@code UnsupportedOperationException}
* (a latent legacy crash). Routing on the relation type never calls
the unsupported shape.</li>
* <li>MOR → {@link IncrementalRelation#collectFileSlices()} (a
FLAT cross-partition slice list) turned
* into JNI ranges at the resolved window END ({@code
relation.getEndTs()}), with per-slice partition
- * values parsed from the slice's own partition path against the
Hudi table-config partition fields
- * (the same non-handle source the COW relation uses). {@code
force_jni} still keeps a no-log MOR slice
- * on JNI via {@link #buildMorRange}.</li>
+ * values resolved from the slice's own partition path. For Hive
Sync tables the resolver is built from
+ * HMS locations and extractor-produced logical values; otherwise it
parses the Hudi physical layout.
+ * {@code force_jni} still keeps a no-log MOR slice on JNI via
{@link #buildMorRange}.</li>
* </ul>
* Package-private static, pure over the {@link IncrementalRelation}
contract, so file-selection routing +
* the degrade decision are unit-testable with a fake relation (no live
metaClient).
*/
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,
+ List<String> columnNames, List<String> columnTypes,
+ Supplier<Function<String, Map<String, String>>>
partitionValueResolverSupplier,
UnaryOperator<String> nativePathNormalizer) {
if (relation.fallbackFullTableScan()) {
return Optional.empty();
}
+ Function<String, Map<String, String>> partitionValueResolver =
+ new LazyPartitionValueResolver(partitionValueResolverSupplier);
List<ConnectorScanRange> ranges = new ArrayList<>();
if (isCow) {
// COW @incr yields native ranges directly; normalize their scheme
(s3a->s3) for BE's native reader
// (COWIncrementalRelation.collectSplits builds .path() from the
raw HMS base path).
- ranges.addAll(relation.collectSplits(nativePathNormalizer));
+ ranges.addAll(relation.collectSplits(partitionValueResolver,
nativePathNormalizer));
return Optional.of(ranges);
}
String endTs = relation.getEndTs();
for (FileSlice fileSlice : relation.collectFileSlices()) {
- Map<String, String> partValues =
parsePartitionValues(fileSlice.getPartitionPath(), partitionFieldNames);
+ Map<String, String> partValues =
partitionValueResolver.apply(fileSlice.getPartitionPath());
// @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));
}
return Optional.of(ranges);
}
+ private Function<String, Map<String, String>>
incrementalPartitionValueResolver(
+ HudiTableHandle handle, HoodieTableMetaClient metaClient, boolean
hiveStylePartitioning) {
+ return incrementalPartitionValueResolver(
+ partitionFieldNames(metaClient), handle.getPartitionKeyNames(),
+ useHiveSyncPartition(), hiveStylePartitioning,
+ () -> listHiveSyncPartitions(handle));
+ }
+
+ static Function<String, Map<String, String>>
incrementalPartitionValueResolver(
+ List<String> tableConfigPartitionFields, List<String>
hmsPartitionFields,
+ boolean useHiveSyncPartition, boolean hiveStylePartitioning,
+ Supplier<Optional<List<PartitionScanInfo>>>
hiveSyncPartitionsSupplier) {
+ List<String> fields = tableConfigPartitionFields.isEmpty()
+ ? hmsPartitionFields : tableConfigPartitionFields;
+ Function<String, Map<String, String>> physicalPathResolver =
+ path -> parsePartitionValues(path, fields,
hiveStylePartitioning);
+ if (!useHiveSyncPartition || hmsPartitionFields.isEmpty()) {
+ return physicalPathResolver;
+ }
+ Optional<List<PartitionScanInfo>> hiveSyncPartitions =
hiveSyncPartitionsSupplier.get();
+ return hiveSyncPartitions.isPresent()
+ ? exactPartitionValueResolver(hiveSyncPartitions.get())
+ : physicalPathResolver;
+ }
+
+ /** Builds a fail-loud physical-path to logical-value resolver for Hive
Sync incremental scans. */
+ static Function<String, Map<String, String>> exactPartitionValueResolver(
+ List<PartitionScanInfo> partitions) {
+ Map<String, Map<String, String>> valuesByPath = new HashMap<>();
+ for (PartitionScanInfo partition : partitions) {
+ Map<String, String> old = valuesByPath.put(
+ partition.partitionPath, partition.partitionValues);
+ if (old != null) {
+ throw new DorisConnectorException(
+ "Multiple Hudi Hive Sync partitions point to " +
partition.partitionPath);
+ }
+ }
+ return partitionPath -> {
+ Map<String, String> values = valuesByPath.get(partitionPath);
+ if (values == null) {
+ throw new DorisConnectorException(
+ "Hudi partition path " + partitionPath + " is missing
from Hive Sync metadata");
+ }
+ return values;
+ };
+ }
+
/**
- * The Hudi table-config partition-field names (byte-faithful to legacy
{@code HudiScanNode:391-393}), the
- * source the incremental MOR path parses per-slice partition values
against — NOT the HMS-sourced
- * handle partition keys the snapshot path uses (the two coincide only for
hive-synced tables).
+ * The Hudi table-config partition-field names, canonicalized to Hudi's
lower-case Doris-column convention.
+ * Incremental scans prefer these names when present and fall back to the
HMS-backed handle keys for legacy
+ * Hudi tables whose table config predates the partition-fields property.
*/
private static List<String> partitionFieldNames(HoodieTableMetaClient
metaClient) {
Option<String[]> fields =
metaClient.getTableConfig().getPartitionFields();
- return fields.isPresent() ? Arrays.asList(fields.get()) :
Collections.emptyList();
+ return fields.isPresent()
+ ? Arrays.stream(fields.get())
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList())
+ : Collections.emptyList();
+ }
+
+ /** Whether Hudi storage paths use {@code column=value} rather than
positional fragments. */
+ static boolean hiveStylePartitioning(HoodieTableMetaClient metaClient) {
+ return
Boolean.parseBoolean(metaClient.getTableConfig().getHiveStylePartitioningEnable());
}
/**
* Resolve partition paths from handle or by listing all partitions.
*/
- private List<String> resolvePartitions(
- HudiTableHandle handle, HoodieTableMetaClient metaClient) {
- // Check if partitions were pruned via applyFilter
+ List<PartitionScanInfo> resolvePartitions(
+ HudiTableHandle handle, HoodieTableMetaClient metaClient, boolean
hiveStylePartitioning) {
+ // Hive Sync pruning carries both the exact storage location and the
extractor-produced logical values.
+ List<HmsPartitionInfo> prunedPartitions = handle.getPrunedPartitions();
+ if (prunedPartitions != null) {
+ return hmsPartitionScanInfos(
+ handle.getBasePath(), handle.getPartitionKeyNames(),
prunedPartitions);
+ }
+
+ // Non-Hive-Sync pruning carries physical relative paths from Hudi
metadata.
List<String> prunedPaths = handle.getPrunedPartitionPaths();
if (prunedPaths != null) {
- return prunedPaths;
+ return physicalPartitionScanInfos(
+ prunedPaths, handle.getPartitionKeyNames(),
hiveStylePartitioning);
}
// No pruning — list all partitions
List<String> partKeyNames = handle.getPartitionKeyNames();
if (partKeyNames == null || partKeyNames.isEmpty()) {
// Unpartitioned table
- return Collections.singletonList("");
+ return Collections.singletonList(new PartitionScanInfo("",
Collections.emptyMap()));
+ }
+
+ Optional<List<PartitionScanInfo>> hiveSyncPartitions =
listHiveSyncPartitions(handle);
Review Comment:
[P1] Keep Hudi's timeline as the partition completeness source
Returning any nonempty current HMS listing here makes it the complete
physical partition universe. A latest commit that precedes Hive Sync is
silently omitted, and a `FOR TIME AS OF` scan loses a partition later dropped
or renamed from HMS even though FileSystemView is queried at the pinned
instant. Incremental COW/MOR independently select Hudi paths and then the exact
current-HMS resolver throws on the same absence. Keep Hudi/timeline physical
selection authoritative and join HMS only for logical extractor values, with
sync-lag, historical-removal, and incremental regressions.
--
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]
