danny0405 commented on a change in pull request #2581:
URL: https://github.com/apache/hudi/pull/2581#discussion_r579915210
##########
File path:
hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/BucketAssignFunction.java
##########
@@ -146,5 +209,69 @@ public void notifyCheckpointComplete(long l) {
// Refresh the table state when there are new commits.
this.bucketAssigner.reset();
this.bucketAssigner.refreshTable();
+ checkPartitionsLoaded();
+ }
+
+ /**
+ * Load all the indices of give partition path into the backup state.
+ *
+ * @param partitionPath The partition path
+ * @throws Exception when error occurs for state update
+ */
+ private void loadRecords(String partitionPath) throws Exception {
+ HoodieTable<?, ?, ?, ?> hoodieTable = bucketAssigner.getTable();
+ List<HoodieBaseFile> latestBaseFiles =
+ HoodieIndexUtils.getLatestBaseFilesForPartition(partitionPath,
context, hoodieTable);
+ for (HoodieBaseFile baseFile : latestBaseFiles) {
+ List<HoodieKey> hoodieKeys =
+ ParquetUtils.fetchRecordKeyPartitionPathFromParquet(hadoopConf, new
Path(baseFile.getPath()));
+ hoodieKeys.forEach(hoodieKey -> {
+ try {
+ this.indexState.put(hoodieKey, new
HoodieRecordLocation(baseFile.getCommitTime(), baseFile.getFileId()));
+ } catch (Exception e) {
+ throw new HoodieIOException("Error when load record keys from file:
" + baseFile);
+ }
+ });
+ }
+ // Mark the partition path as loaded.
+ partitionLoadState.put(partitionPath, 0);
Review comment:
> `The 0 is meaningless here`
It is meaningless anyway, because Flink does not have Set state.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
