lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-3315] RFC-35 Part-1 Support bucket index in Flink writer (#4679)

Browse Source 
* Support bucket index in Flink writer
* Use record key as default index key
This commit is contained in:
Gary Li
2022-03-02 15:14:44 +08:00
committed by GitHub
parent 85f47b53df
commit 10d866f083
14 changed files with 321 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndex.java
View File

@@ -141,6 +141,6 @@ public abstract class HoodieIndex<I, O> implements Serializable {
}
public enum IndexType {
HBASE, INMEMORY, BLOOM, GLOBAL_BLOOM, SIMPLE, GLOBAL_SIMPLE, BUCKET
HBASE, INMEMORY, BLOOM, GLOBAL_BLOOM, SIMPLE, GLOBAL_SIMPLE, BUCKET, FLINK_STATE
}
}

12
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
View File

@@ -56,6 +56,10 @@ public class BucketIdentifier {
return hashKeyFields.hashCode() % numBuckets;
}
public static String partitionBucketIdStr(String partition, int bucketId) {
return String.format("%s_%s", partition, bucketIdStr(bucketId));
}
public static int bucketIdFromFileId(String fileId) {
return Integer.parseInt(fileId.substring(0, 8));
}
@@ -64,6 +68,10 @@ public class BucketIdentifier {
return String.format("%08d", n);
}
public static String newBucketFileIdPrefix(int bucketId) {
return newBucketFileIdPrefix(bucketIdStr(bucketId));
}
public static String newBucketFileIdPrefix(String bucketId) {
return FSUtils.createNewFileIdPfx().replaceFirst(".{8}", bucketId);
}
@@ -71,4 +79,8 @@ public class BucketIdentifier {
public static boolean isBucketFileName(String name) {
return BUCKET_NAME.matcher(name).matches();
}
public static int mod(int x, int y) {
return x % y;
}
}

4
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
View File

@@ -145,6 +145,10 @@ public abstract class HoodieTable<T extends HoodieRecordPayload, I, K, O> implem
return viewManager;
}
public HoodieTableMetadata getMetadata() {
return metadata;
}
/**
* Upsert a batch of new records into Hoodie table at the supplied instantTime.
* @param context HoodieEngineContext