wombatu-kun commented on code in PR #20314:
URL: https://github.com/apache/hudi/pull/20314#discussion_r4302758988
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java:
##########
@@ -496,7 +498,37 @@ 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() {
+ if (metaClient == null) {
+ // a streaming read of a table that does not exist yet has no record key
encoding to follow
+ return Option.empty();
+ }
+ if
(!metaClient.getTableConfig().isComplexKeyGenWithSingleRecordKeyField()) {
+ return Option.of(false);
+ }
+ // a single-field complex keygen table stores `<field>:<value>` record
keys under the FIELD_PREFIXED encoding;
+ // only the recorded encoding is followed, because deducing it from the
data would read files at planning time,
+ // so a table that has not recorded it yet is not pruned by key until a
write or an upgrade records it
+ Option<ComplexKeyGenEncoding> encodingOpt =
metaClient.getTableConfig().getComplexKeyGenEncoding();
+ if (!encodingOpt.isPresent()) {
+ return Option.empty();
+ }
+ if (!encodingOpt.get().encodesFieldName()) {
+ return Option.of(false);
+ }
+ // Spark writes `<field>:<value>` keys into a FIELD_PREFIXED table, while
the Flink key generator follows the job
+ // configuration and writes bare keys for an explicit complex key
generator with a single partition field, so
+ // when this job would not prefix, the stored keys can take either form
+ return RowDataKeyGen.isFieldPrefixedSingleRecordKey(conf) ?
Option.of(true) : Option.empty();
Review Comment:
`checkKeygenGenerator` infers from the job's key and partition field counts
and never reads the table, so a job that does not declare the encoding is only
right for tables its own DDL would have keyed. A writer of a table from 1.0.x
already has to declare it (`TestComplexKeyGenEncodingWrite`), but a job that
only reads one with `PARTITIONED BY (p1, p2)` gets `FIELD_PREFIXED` for its
bare keys, so with the conf alone `uuid = 'id1'` silently returns no rows where
master returns the row, and with one partition field the bare hash misses the
`uuid:id1` keys Spark writes. If readers should be required to declare the
encoding too, I can switch to the conf alone; otherwise the recorded value
keeps both cases correct.
--
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]
