Repository navigation
fix(flink): align bucket pruning with field-prefixed record keys - #20314
wombatu-kun wants to merge 3 commits into
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This makes Flink bucket pruning hash a field-prefixed single record key the same way the bucket writers store it, and also fixes the infinite loop in extractRecordKeysByFields; one non-blocking point is flagged inline about tables where the recorded encoding and the job's key generator disagree. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. One suggestion on keeping getDataBucketFunc readable.
| } | ||
| // 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) |
There was a problem hiding this 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?
| return Option.empty(); | ||
| } | ||
| return Option.of(PrimaryKeyPruners.getBucketIdFunc(indexKeyFilters, conf)); | ||
| if (metaClient == null) { |
There was a problem hiding this 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.
github-actions
Bot
added
size:L
| * | ||
| * @return empty when the form of the stored record keys cannot be determined, which skips the bucket pruning | ||
| */ | ||
| private Option<Boolean> resolvePrefixedSingleFieldKey() { |
There was a problem hiding this comment.
The HoodieTableConfig.COMPLEX_KEYGEN_ENCODING is already setup in table factory, can we just decode it and avoid to call KeyGenUtils.resolveComplexKeyGenEncoding which is costly.
There was a problem hiding this comment.
Done c2fb5ea: 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).
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #20314 +/- ##
============================================
+ Coverage 80.46% 80.71% +0.25%
- Complexity 35045 35162 +117
============================================
Files 2553 2553
Lines 143547 143567 +20
Branches 17478 17483 +5
============================================
+ Hits 115506 115887 +381
+ Misses 20076 19736 -340
+ Partials 7965 7944 -21
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
…rm is ambiguous and extract the check into a helper
78f2420 to
c2fb5ea
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR makes Flink bucket pruning hash the same <field>:<value> key the writers store for single-key complex keygen tables, skips pruning when the stored key form is unclear, and fixes the infinite loop in extractRecordKeysByFields. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. One small readability suggestion below; the rest looks clean.
cc @yihua
| * | ||
| * @return empty when the form of the stored record keys cannot be determined, which skips the bucket pruning | ||
| */ | ||
| private Option<Boolean> resolvePrefixedSingleFieldKey() { |
There was a problem hiding this comment.
🤖 nit: could you replace Option<Boolean> with a small enum (e.g. PREFIXED, BARE, UNKNOWN)? Here empty means "skip pruning" and false means "bare keys", which is easy to misread at the call site.
| // 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(); |
There was a problem hiding this comment.
just this check should be enough? the option HoodieTableConfig.COMPLEX_KEYGEN_ENCODING was inferred statically no matter whether the table exists.
There was a problem hiding this comment.
Not for a table keyed by Spark, which follows the recorded encoding rather than a job conf (KeyGenUtils#withComplexKeyGenEncoding). A reading job without an encoding option would hash 1 on a Spark ComplexKeyGenerator table with one partition field, which stores id:1, and id:1 on a legacy VALUE_ONLY table with two partition fields, which stores 1, so with the conf alone both key queries return no rows; the recorded value is already loaded by the meta client and keeps both correct.
This branch has not been deployed
4 participants
Describe the issue this Pull Request addresses
Closes #20313.
On a Flink bucket-index table with one primary key and two or more partition fields, an equality filter on the primary key returns no rows: the complex key generator stores the key as
uuid:id1(FIELD_PREFIXED) and the bucket writers hash that, whilePrimaryKeyPruners.getBucketIdFunchashed the bare literalid1.Summary and Changelog
HoodieTableSourcereads the record key encoding recorded in the table config and hashes<field>:<value>when both the table and the job's key generator prefix. When the table recordsFIELD_PREFIXEDbut the job would not prefix, the stored keys can take either form (Spark prefixes, Flink with an explicit complex key generator and one partition field does not), so the pruning is skipped, as it is for a table without a recorded encoding (no data files are read at planning time) and for a streaming read before the table exists.RowDataKeyGen#isFieldPrefixedSingleRecordKeyexposes the writer's prefix decision to the pruner.PrimaryKeyPruners#getBucketIdFunchashes the prefixed key throughBucketIdentifier.getBucketId(recordKey, indexKeyFields, numBuckets), as the writers do.KeyGenUtils.extractRecordKeysByFieldsgets the missingkeyValueSep2 > 0guard, so a skipped trailing value holding:(e.g.'a,b:c:d') no longer loops forever; only inputs that never returned change.ITTestHoodieDataSourcereproduces both table shapes with Flink SQL on COW and MOR,TestHoodieTableSourcechecks the pruning bucket against the writer's key generator and the skipped cases, andTestKeyGenUtilscovers the parser.Impact
Primary key equality queries on these tables return their rows again. A
FIELD_PREFIXEDtable read by a job that would not prefix, such as one Spark wrote with one partition field, now reads every bucket and finds its rows. Other key generators, compound keys andVALUE_ONLYtables keep their bucket ids. No storage format, API or configuration change.Risk Level
Low. The bucket changes only when the table and the job both prefix, and then follows the writers' hashing; a table without a recorded encoding is read without bucket pruning until its next write or upgrade records it.
Verified: the multi-partition IT fails on master and the single-partition IT on this PR's first revision
ee336b06e3e1, both withExpected: is "[+I[id1]]" but: was "[]", and each new unit-test assertion fails with its part of the fix removed. Merges cleanly with #20292.Documentation Update
None.
Contributor's checklist