hudi-agent commented on code in PR #20314:
URL: https://github.com/apache/hudi/pull/20314#discussion_r4266585956
##########
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:
🤖 non-blocking: Spark's `ComplexAvroKeyGenerator` prefixes keys whenever the
table records `FIELD_PREFIXED`, even with a single partition field. So a
Spark-written single-partition complex-keygen bucket table read from Flink
takes the bare-hash branch here, lands on the wrong bucket, and `uuid = 'x'`
returns nothing. Since we can't tell which writer produced the keys when the
table encoding says prefixed but `isFieldPrefixedSingleRecordKey(conf)` says
bare, would it make sense to return `Option.empty()` in that case instead of
hashing the bare value?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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:
🤖 nit: would it make sense to move the encoding resolution and the
`prefixedSingleFieldKey` decision into a small helper (e.g.
`resolvePrefixedSingleFieldKey()` returning `Option<Boolean>`)? That would keep
`getDataBucketFunc` focused on the filter check and bucket function.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
