wombatu-kun commented on code in PR #20314:
URL: https://github.com/apache/hudi/pull/20314#discussion_r4266883143
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java:
##########
@@ -496,7 +499,27 @@ private Option<Function<Integer, Integer>>
getDataBucketFunc(List<ResolvedExpres
if (!ExpressionUtils.isFilteringByAllFields(indexKeyFilters,
indexKeyFields)) {
return Option.empty();
}
- return Option.of(PrimaryKeyPruners.getBucketIdFunc(indexKeyFilters, conf));
+ if (metaClient == null) {
+ // a streaming read of a table that does not exist yet has no record key
encoding to follow
+ return Option.empty();
+ }
+ // a single-field complex keygen table stores `<field>:<value>` record
keys under the FIELD_PREFIXED encoding,
+ // and cannot be pruned by key while its encoding is unknown
+ Option<ComplexKeyGenEncoding> encodingOpt;
+ try {
+ encodingOpt = KeyGenUtils.resolveComplexKeyGenEncoding(metaClient);
+ } catch (Exception e) {
+ log.warn("Failed to resolve the record key encoding of table {},
skipping the bucket pruning", path, e);
+ return Option.empty();
+ }
+ if (metaClient.getTableConfig().isComplexKeyGenWithSingleRecordKeyField()
&& !encodingOpt.isPresent()) {
+ return Option.empty();
+ }
+ // the Flink key generator prefixes by the job configuration, which can
disagree with the table's encoding
+ // (an explicit complex key generator with a single partition field writes
bare keys), so require both
+ boolean prefixedSingleFieldKey =
encodingOpt.map(ComplexKeyGenEncoding::encodesFieldName).orElse(false)
Review Comment:
Done d05f8afc44e3
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java:
##########
@@ -496,7 +499,27 @@ private Option<Function<Integer, Integer>>
getDataBucketFunc(List<ResolvedExpres
if (!ExpressionUtils.isFilteringByAllFields(indexKeyFilters,
indexKeyFields)) {
return Option.empty();
}
- return Option.of(PrimaryKeyPruners.getBucketIdFunc(indexKeyFilters, conf));
+ if (metaClient == null) {
Review Comment:
Done d05f8afc44e3
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java:
##########
@@ -496,7 +499,39 @@ private Option<Function<Integer, Integer>>
getDataBucketFunc(List<ResolvedExpres
if (!ExpressionUtils.isFilteringByAllFields(indexKeyFilters,
indexKeyFields)) {
return Option.empty();
}
- return Option.of(PrimaryKeyPruners.getBucketIdFunc(indexKeyFilters, conf));
+ return resolvePrefixedSingleFieldKey()
+ .map(prefixedSingleFieldKey ->
PrimaryKeyPruners.getBucketIdFunc(indexKeyFilters, conf,
prefixedSingleFieldKey));
+ }
+
+ /**
+ * Resolves whether the bucket writers hash a single record key field as
{@code <field>:<value>}.
+ *
+ * @return empty when the form of the stored record keys cannot be
determined, which skips the bucket pruning
+ */
+ private Option<Boolean> resolvePrefixedSingleFieldKey() {
Review Comment:
Done c2fb5eaee44b: the helper reads only the encoding recorded in the table
config the meta client has already loaded, and skips the pruning for a table
that has not recorded it yet instead of deducing it from the data. The conf
alone cannot replace that check: a `FIELD_PREFIXED` table keyed by Spark and
read by a single-partition job would return no rows
(`ITTestHoodieDataSource#testBucketPruningWithPrefixedKeysAndSinglePartitionField`).
--
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]
