-
Notifications
You must be signed in to change notification settings - Fork 2.4k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[HUDI-6534]Support consistent hashing row writer #9199
[HUDI-6534]Support consistent hashing row writer #9199
Conversation
5fa5151
to
8f1d8ef
Compare
d7062e5
to
b8a0a16
Compare
23579f5
to
3eca97c
Compare
@hudi-bot run azure |
this.keyGeneratorOpt = Option.empty(); | ||
} | ||
this.extractor = RowRecordKeyExtractor.getRowRecordKeyExtractor(populateMetaFields, keyGeneratorOpt); | ||
ValidationUtils.checkArgument(table.getMetaClient().getTableType().equals(HoodieTableType.MERGE_ON_READ), |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
is the check by design for consistent hash index? if yes we could move the check to the parent class?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, we do dual write during consistent hashing bucket index resizing but CoW table do not support writing logs. And It's a little bit hard to move it to parent class since the closest parent class for these two ConsistentHashingPartitioner is BulkInsertPartitioner
@@ -83,7 +83,7 @@ private static void extractHashingMetadataFromClusteringPlan(String instant, Hoo | |||
ValidationUtils.checkState(p != null, "Clustering plan does not has partition info, plan: " + plan); | |||
// Skip unrelated clustering group | |||
if (!recordPartitions.contains(p)) { | |||
return; | |||
continue; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
would you please clarify why the logic changed here
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's a bug in the origin code. The set recordPartitions
contains all partitions that we care about, and we iterate over input groups in clustering plan to construct hashing metadata. If current input group is not included in recordPartitions
we need to continue to check the next input group instead of return directly.
/** | ||
* Set consistent hashing for partition, used in clustering | ||
* | ||
* @param partition partition to set Consistent Hashing nodes |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Consistent Hashing
-> consistent hashing
@@ -79,15 +94,19 @@ public HoodieData<WriteStatus> performClusteringWithRecordsRDD(HoodieData<Hoodie | |||
HoodieWriteConfig newConfig = HoodieWriteConfig.newBuilder().withProps(props).build(); | |||
|
|||
RDDConsistentBucketBulkInsertPartitioner<T> partitioner = new RDDConsistentBucketBulkInsertPartitioner<>(getHoodieTable(), strategyParams, preserveHoodieMetadata); | |||
addHashingChildNodes(partitioner, extraMetadata); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I am a little confused by the name between ConsistentHashingBucketInsertPartitioner#addHashingChildrenNodes
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Here I made a new interface call ConsistentHashingBucketInsertPartitioner
to share the common method addHashingChildrenNodes
between RDDConsistentBucketBulkInsertPartitioner
and ConsistentBucketIndexBulkInsertPartitionerWithRows
. I didn't put the method addHashingChildrenNodes
in their closest parent class BulkInsertPartitioner
because It's all about consistent hashing bucket index and I don't want to affect other partitioners. What do you think about this design here? Should I keep the current design, move this to the parent class, or duplicate the code for both partitioners?
try { | ||
return Option.of((BuiltinKeyGenerator) HoodieSparkKeyGeneratorFactory.createKeyGenerator(typedProperties)); | ||
} catch (ClassCastException cce) { | ||
throw new HoodieIOException("Only those key generators implementing BuiltInKeyGenerator interface is supported with virtual keys"); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
also throw the cce
here?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The code is moved from BulkInsertDataInternalWriterHelper
and I think it's reasonable to translate the cce
into a more readable exceptionHoodieIOException
.
* | ||
* @param partitionToIdentifier Mapping from table partition to bucket identifier | ||
*/ | ||
public static Map<String, Map<String, Integer>> generatePartitionToFileIdPfxIdxMap(Map<String, ConsistentBucketIdentifier> partitionToIdentifier) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
can also add uts for the method ?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sure, will add more tests~
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Added test in TestConsistentBucketIdIdentifier.java
return super.getWriteHandleFactory(idx).map(writeHandleFactory -> new WriteHandleFactory() { | ||
@Override | ||
public HoodieWriteHandle create(HoodieWriteConfig config, String commitTime, HoodieTable hoodieTable, String partitionPath, String fileIdPrefix, TaskContextSupplier taskContextSupplier) { | ||
// Ensure we do not create append handle for consistent hashing bulk_insert, align with `ConsistentBucketBulkInsertDataInternalWriterHelper` |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For reviewers: When we bulk insert twice into a consistent hashing bucket index table, we need to write logs to existing file groups in the second bulk insert, while for normal bloom filter index table, we will always create new base files when bulk insert. However currently bulk insert row writer path do not support writing logs, so I add a check here to prevent user from bulk insert twice into a consistent hashing bucket index table. We should use upsert after first bulk insert for consistent hashing bucket index table.
return HoodieMergeOnReadTestUtils.getRecordsUsingInputFormat(hadoopConf, | ||
Arrays.stream(partitions).map(p -> Paths.get(basePath, p).toString()).collect(Collectors.toList()), | ||
basePath, new JobConf(hadoopConf), true, false); | ||
private List<Row> readRecords() { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For reviewers: Reading the data written by bulk insert row writer will have the same issue as #8838. So I use spark instead of input format to read data here.
@leesf , is it good to land now ? We still have 2 days for the 0.14.0 code freeze. |
3eca97c
to
535bf53
Compare
abf8721
to
884a71a
Compare
.bulkInsert(inputRecords, instantTime, getHoodieTable(), newConfig, false, partitioner, true, numOutputGroups); | ||
} | ||
|
||
private void addHashingChildNodes(ConsistentHashingBucketInsertPartitioner partitioner, Map<String, String> extraMetadata) { | ||
try { | ||
List<ConsistentHashingNode> nodes = ConsistentHashingNode.fromJsonString(extraMetadata.get(BaseConsistentHashingBucketClusteringPlanStrategy.METADATA_CHILD_NODE_KEY)); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
will the nodes be null and need check in addHashingChildrenNodes?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Will check the nodes is non-emtpy in addHashingChildrenNodes
|
||
@Override | ||
public void addHashingChildrenNodes(String partition, List<ConsistentHashingNode> nodes) { | ||
ValidationUtils.checkState(nodes.stream().noneMatch(n -> n.getTag() == ConsistentHashingNode.NodeTag.NORMAL), |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
what's the reason the tag will not be NORMAL?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Normal
nodes are read from metadata in file system, while REPLACE
and DELETE
nodes are read from consistent hashing resizing plan. Here we are adding nodes read from clustering plan to the hash metadata so all nodes must not be NORMAL
@@ -65,7 +71,6 @@ public void write(InternalRow row) throws IOException { | |||
int bucketId = BucketIdentifier.getBucketId(String.valueOf(recordKey), indexKeyFields, bucketNum); | |||
Pair<UTF8String, Integer> fileId = Pair.of(partitionPath, bucketId); | |||
if (lastFileId == null || !lastFileId.equals(fileId)) { | |||
LOG.info("Creating new file for partition path " + partitionPath); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this log is useless then?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We will not always create a new file handle if we have already created one for this partition so I delete this line. I will add it back to getBucketRowCreateHandle
when we do need to create a new handle.
|
||
private static final Logger LOG = LoggerFactory.getLogger(ConsistentBucketBulkInsertDataInternalWriterHelper.class); | ||
|
||
public ConsistentBucketBulkInsertDataInternalWriterHelper(HoodieTable hoodieTable, HoodieWriteConfig writeConfig, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
the constructor has too many args, which can wrap in a class?
2a3b8ac
to
445cf75
Compare
445cf75
to
19efc8e
Compare
@stream2000 The build is failing due to a test failure due to this commit. Can you please check? This is blocking 0.14.0 release so please prioritize if possible. |
@hudi-bot run azure |
@prashantwason You can cherry pick #9401 |
String fileId = FSUtils.createNewFileId(node.getFileIdPrefix(), 0); | ||
|
||
ValidationUtils.checkArgument(node.getTag() != ConsistentHashingNode.NodeTag.NORMAL | ||
|| hoodieTable.getFileSystemView() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is this check too heavy to do when starting write to a new file group?
I think it might be simpler if we add a file existing check in the init method of class HoodieRowCreateHandle
and fail the writer if file existing.
e.g.
public HoodieRowCreateHandle(HoodieTable table,
HoodieWriteConfig writeConfig,
String partitionPath,
String fileId,
String instantTime,
int taskPartitionId,
long taskId,
long taskEpochId,
StructType structType,
boolean shouldPreserveHoodieMetadata) {
...
FileSystem fs = table.getMetaClient().getFs();
String writeToken = getWriteToken(taskPartitionId, taskId, taskEpochId);
String fileName = FSUtils.makeBaseFileName(instantTime, writeToken, this.fileId, table.getBaseFileExtension());
this.path = makeNewPath(fs, partitionPath, fileName, writeConfig);
if (fs.exists(this.path)) {
throw new HoodieException("File already exist. Can't bulk insert. Can use insert|upsert");
}
...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can one writer helper writes multiple file groups?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, one writer helper can write multiple file groups, but one object of HoodieRowCreateHandle
can only write one parquet file, which is this.path
. This write helper will switch between the different handlers.
So in the method BulkInsertDataInternalWriterHelper.write(InternalRow row)
, BucketBulkInsertDataInternalWriterHelper.write(InternalRow row)
and ConsistentBucketBulkInsertDataInternalWriterHelper.write(InternalRow row)
, they will create a new object of HoodieRowCreateHandle
if the incoming row need to be written into a new parquet file.
We only have this existing file group check in ConsistentBucketBulkInsertDataInternalWriterHelper.getBucketRowCreateHandle
, but actually, other bulk insert writer BulkInsertDataInternalWriterHelper
and BucketBulkInsertDataInternalWriterHelper
need this check as well. So I think it will be simpler if we add a check in the init method of HoodieRowCreateHandle
.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Just check the code, this writer help class is used in the method HoodieDatasetBulkInsertHelper.bulkInsert
in a mapPartitions
method, and org.apache.hudi.spark3.internal.HoodieBulkInsertDataInternalWriter
.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If it looks reasonable for improvement, you can try to raise a PR then, maybe @stream2000 can help for a reivew~
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah sure, I'm adding a few test in our internal code, will raise a PR for with the test cases and fix.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@TengHuo I think using file system view which can help filter out replaced/uncommitted file slices is a more common way when we need information about file slices. And does the write token is the same between every write for the same file id?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Oh got it, Indeed, writeToken
is different every time when a task or stage failed and retry. So, just checking fs.exists(this.path)
is not good enough, need to check file group with instantTime
. In that case, using hoodieTable.getFileSystemView()
is better then list partition.
writeConfig.populateMetaFields, | ||
arePartitionRecordsSorted, | ||
shouldPreserveHoodieMetadata) | ||
case _ => |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should we support SIMPLE bucket index as well?
It uses BucketBulkInsertDataInternalWriterHelper
.
Change Logs
Support consistent hashing row writer for bulk_insert load and clustering.
Impact
Support consistent hashing bulk_insert row writer
Risk level (write none, low medium or high below)
medium, will enabled by default since row writer is enabled by default
Documentation Update
will update document after landing
Contributor's checklist