danny0405 commented on code in PR #12884: URL: https://github.com/apache/hudi/pull/12884#discussion_r1978417143
########## rfc/rfc-89/rfc-89.md: ########## @@ -0,0 +1,345 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +# RFC-89: Partition Level Bucket Index + +## Proposers +- @zhangyue19921010 + +## Approvers +- @danny0405 +- @codope +- @xiarixiaoyao + +## Status + +JIRA: https://issues.apache.org/jira/browse/HUDI-8990 + +## Abstract + +As we know, Hudi proposed and introduced Bucket Index in RFC-29. Bucket Index can well unify the indexes of Flink and +Spark, that is, Spark and Flink could upsert the same Hudi table using bucket index. + +However, Bucket Index has a limit of fixed number of buckets. In order to solve this problem, RFC-42 proposed the ability +of consistent hashing achieving bucket resizing by splitting or merging several local buckets dynamically. + +But from PRD experience, sometimes a Partition-Level Bucket Index and a offline way to do bucket rescale is good enough +without introducing additional efforts (multiple writes, clustering, automatic resizing,etc.). Because the more complex +the Architecture, the more error-prone it is and the greater operation and maintenance pressure. + +In this regard, we could upgrade the traditional Bucket Index to implement a Partition-Level Bucket Index, so that users +can set a specific number of buckets for different partitions through a rule engine (such as regular expression matching). +On the other hand, for a certain existing partitions, an off-line command is provided to reorganized the data using insert +overwrite(need to stop the data writing of the current partition). + +More importantly, the existing Bucket Index table can be upgraded to Partition-Level Bucket Index smoothly and seamlessly. + +## Background +The following is the core read-write process of the Flink/Spark engine based on Simple Bucket Index +### Flink Write Using Simple Bucket Index +**Step 1**: re-partition input records based on `BucketIndexPartitioner`, BucketIndexPartitioner has **a fixed bucketNumber** for all partition path. +For each record key, compute a fixed data partition number, doing re-partition works. + +```java +/** + * Bucket index input partitioner. + * The fields to hash can be a subset of the primary key fields. + * + * @param <T> The type of obj to hash + */ +public class BucketIndexPartitioner<T extends HoodieKey> implements Partitioner<T> { + + private final int bucketNum; + private final String indexKeyFields; + + private Functions.Function2<String, Integer, Integer> partitionIndexFunc; + + public BucketIndexPartitioner(int bucketNum, String indexKeyFields) { + this.bucketNum = bucketNum; + this.indexKeyFields = indexKeyFields; + } + + @Override + public int partition(HoodieKey key, int numPartitions) { + if (this.partitionIndexFunc == null) { + this.partitionIndexFunc = BucketIndexUtil.getPartitionIndexFunc(bucketNum, numPartitions); + } + int curBucket = BucketIdentifier.getBucketId(key.getRecordKey(), indexKeyFields, bucketNum); + return this.partitionIndexFunc.apply(key.getPartitionPath(), curBucket); + } +} +``` +**Step 2**: Using `BucketStreamWriteFunction` upsert records into hoodie +- Bootstrap and cache `partition_bucket -> fileID` mapping from the existing hudi table +- Tagging: compute `bucketNum` and tag `fileID` based on record key and bucketNumber config through `BucketIdentifier` +- buffer and write records + +### Flink Read Pruning Using Simple Bucket Index +**Step 1**: compute `dataBucket` +```java + private int getDataBucket(List<ResolvedExpression> dataFilters) { + if (!OptionsResolver.isBucketIndexType(conf) || dataFilters.isEmpty()) { + return PrimaryKeyPruners.BUCKET_ID_NO_PRUNING; + } + Set<String> indexKeyFields = Arrays.stream(OptionsResolver.getIndexKeys(conf)).collect(Collectors.toSet()); + List<ResolvedExpression> indexKeyFilters = dataFilters.stream().filter(expr -> ExpressionUtils.isEqualsLitExpr(expr, indexKeyFields)).collect(Collectors.toList()); + if (!ExpressionUtils.isFilteringByAllFields(indexKeyFilters, indexKeyFields)) { + return PrimaryKeyPruners.BUCKET_ID_NO_PRUNING; + } + return PrimaryKeyPruners.getBucketId(indexKeyFilters, conf); + } +``` +**Step 2**: Do partition pruning and get all files in given partitions +**Step 3**: do bucket pruning for all files from step2 +```java + /** + * Returns all the file statuses under the table base path. + */ + public List<StoragePathInfo> getFilesInPartitions() { + ... + // Partition pruning + String[] partitions = + getOrBuildPartitionPaths().stream().map(p -> fullPartitionPath(path, p)).toArray(String[]::new); + if (partitions.length < 1) { + return Collections.emptyList(); + } + List<StoragePathInfo> allFiles = ... + + // bucket pruning + if (this.dataBucket >= 0) { + String bucketIdStr = BucketIdentifier.bucketIdStr(this.dataBucket); + List<StoragePathInfo> filesAfterBucketPruning = allFiles.stream() + .filter(fileInfo -> fileInfo.getPath().getName().contains(bucketIdStr)) + .collect(Collectors.toList()); + logPruningMsg(allFiles.size(), filesAfterBucketPruning.size(), "bucket pruning"); + allFiles = filesAfterBucketPruning; + } + ... + } + +``` + +### Spark Write/Read Using Simple Bucket Index +The read-write process of Spark based on Bucket Index is also similar. +- Use `HoodieSimpleBucketIndex` to tag location. +- Use `SparkBucketIndexPartitioner` to packs incoming records to be inserted into buckets (1 bucket = 1 RDD partition). +- Use `BucketIndexSupport` to Bucket Index pruning during reading. + +## Design +### Workflow +The core process of writing to the Partition Level Bucket Index is shown in the figure below. First, the Bucket Identifier +loads the partition bucket count information from the metadata as a cache. During the Partitioner and Writer stages, when +using the Identifier for shuffling and tagging, it will first attempt to find the bucketNumber of the corresponding +partition from the cache. If the cache was missed, it will calculate the corresponding value based on the user's expression, +and then submit it to the metadata during the commit stage. + +The design details of each part are as follows. + +### Config +Add new Config `hoodie.bucket.index.partition.rule` default `regex` + +Add new config `hoodie.bucket.index.partition.expressions` default null + +Re-use existed config `hoodie.bucket.index.num.buckets` as default partition bucket number + +The above parameters are table-level parameters and need to be declared in the DDL. Users can't change this config through +runtime options like Flink /*+ OPTIONS(...) */ + +#### hoodie.bucket.index.partition.rule +`hoodie.bucket.index.partition.rule` + +It is used to set the rule parser. By default, it is a regular expression parser. +In the future, more parsers will be provided, such as range expression parsers and hybrid parsers. +Here, an interface-based capability will be offered, allowing users to customize parsers. + +#### hoodie.bucket.index.partition.expressions +`hoodie.bucket.index.partition.expressions` + +Users can use this parameter to specify expression and the corresponding bucket numbers (separated by commas). +Multiple rules are separated by semicolons like `hoodie.bucket.index.partition.expressions=expression1,bucket-number1;expression2,bucket-number2` + +#### Demo +```sql +CREATE TABLE hudi_table( + id BIGINT, + name STRING, + price DOUBLE +) +WITH ( +'connector' = 'hudi', +'path' = 'file:///tmp/hudi_table', +'table.type' = 'MERGE_ON_READ', +'index.type' = 'BUCKET', +'hoodie.bucket.index.partition.expressions' = '\\d{4}-(06-(01|17|18)|11-(01|10|11)),256', Review Comment: How do we handle the invalid chars or encoding issues, guess the SQL parser would handle that. -- 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]
