lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-322] DeltaSteamer should pick checkpoints off only deltacommits for MOR tables

Browse Source
This commit is contained in:
harveyyue
2020-01-10 15:22:44 +08:00
committed by Balaji Varadarajan
parent ad50008a59
commit 971c7d41bd
1 changed files with 11 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
View File

@@ -195,7 +195,16 @@ public class DeltaSync implements Serializable {
if (fs.exists(new Path(cfg.targetBasePath))) { if (fs.exists(new Path(cfg.targetBasePath))) {
HoodieTableMetaClient meta = new HoodieTableMetaClient(new Configuration(fs.getConf()), cfg.targetBasePath, HoodieTableMetaClient meta = new HoodieTableMetaClient(new Configuration(fs.getConf()), cfg.targetBasePath,
cfg.payloadClassName); cfg.payloadClassName);
this.commitTimelineOpt = Option.of(meta.getActiveTimeline().getCommitsTimeline().filterCompletedInstants()); switch (meta.getTableType()) {
case COPY_ON_WRITE:
this.commitTimelineOpt = Option.of(meta.getActiveTimeline().getCommitTimeline().filterCompletedInstants());
break;
case MERGE_ON_READ:
this.commitTimelineOpt = Option.of(meta.getActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants());
break;
default:
throw new HoodieException("Unsupported table type :" + meta.getTableType());
}
} else { } else {
this.commitTimelineOpt = Option.empty(); this.commitTimelineOpt = Option.empty();
HoodieTableMetaClient.initTableType(new Configuration(jssc.hadoopConfiguration()), cfg.targetBasePath, HoodieTableMetaClient.initTableType(new Configuration(jssc.hadoopConfiguration()), cfg.targetBasePath,
@@ -378,7 +387,7 @@ public class DeltaSync implements Serializable {
// Schedule compaction if needed // Schedule compaction if needed
if (cfg.isAsyncCompactionEnabled()) { if (cfg.isAsyncCompactionEnabled()) {
scheduledCompactionInstant = writeClient.scheduleCompaction(Option.of(checkpointCommitMetadata)); scheduledCompactionInstant = writeClient.scheduleCompaction(Option.empty());
} }
if (!isEmpty) { if (!isEmpty) {