[GitHub] [incubator-gobblin] autumnust commented on a change in pull request #3079: [GOBBLIN-1231]Make re-compaction be able to write to a new folder based on the executCount

[GitBox <gi...@apache.org>]

autumnust commented on a change in pull request #3079:
URL: https://github.com/apache/incubator-gobblin/pull/3079#discussion_r471761431



##########
File path: gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java
##########
@@ -993,6 +993,13 @@
   public static final String COMPACTION_PRIORITIZER_ALIAS = COMPACTION_PRIORITIZATION_PREFIX + "prioritizerAlias";
   public static final String COMPACTION_ESTIMATOR = COMPACTION_PRIORITIZATION_PREFIX + "estimator";
 
+  /***
+   * Configuration properties related to Re-compaction
+   */
+  public static String COMPACTION_DIRECTORY_FORMAT = "compaction_%s";

Review comment:
       RECOMPACTION_DIRECTORY_FORMAT to be more accurate? 

##########
File path: gobblin-compaction/src/main/java/org/apache/gobblin/compaction/action/CompactionCompleteFileOperationAction.java
##########
@@ -125,9 +122,15 @@ public void onCompactionJobComplete(FileSystemDataset dataset) throws IOExceptio
         // (all previous run + current run) is possible.
         newTotalRecords = this.configurator.getFileNameRecordCount();
       } else {
-        this.configurator.getOldFiles()
-            .add(this.fs.makeQualified(dstPath).toString());
-        this.fs.delete(dstPath, true);
+        if (state.getPropAsBoolean(ConfigurationKeys.RECOMPACTION_WRITE_TO_NEW_FOLDER, false)) {
+          Path oldFilePath = new Path(dstPath, String.format(ConfigurationKeys.COMPACTION_DIRECTORY_FORMAT, executeCount));
+          dstPath = new Path(dstPath, String.format(ConfigurationKeys.COMPACTION_DIRECTORY_FORMAT, executeCount + 1));

Review comment:
       I would recommend using PathUtils.mergePath(), Path constructor will drop the first parameter if the second parameter started with "/" and for here that's the configuration from users.

##########
File path: gobblin-compaction/src/main/java/org/apache/gobblin/compaction/action/CompactionHiveRegistrationAction.java
##########
@@ -88,10 +91,16 @@ public void onCompactionJobComplete(FileSystemDataset dataset) throws IOExceptio
       HiveRegistrationPolicy hiveRegistrationPolicy = HiveRegistrationPolicyBase.getPolicy(state);
 
       List<String> paths = new ArrayList<>();
-      for (HiveSpec spec : hiveRegistrationPolicy.getHiveSpecs(new Path(result.getDstAbsoluteDir()))) {
+      long executeCount = helper.readExecutionCount(new Path(result.getDstAbsoluteDir()));
+      Path dstPath = new Path(result.getDstAbsoluteDir());
+      if (state.getPropAsBoolean(ConfigurationKeys.RECOMPACTION_WRITE_TO_NEW_FOLDER, false)) {
+        // Use new output path to do registration
+        dstPath = new Path(dstPath, String.format(ConfigurationKeys.COMPACTION_DIRECTORY_FORMAT, executeCount));

Review comment:
       Same as above

##########
File path: gobblin-compaction/src/main/java/org/apache/gobblin/compaction/action/CompactionHiveRegistrationAction.java
##########
@@ -57,12 +58,14 @@
 
   private final State state;
   private EventSubmitter eventSubmitter;
+  private InputRecordCountHelper helper;
 
   public CompactionHiveRegistrationAction (State state) {
     if (!(state instanceof WorkUnitState)) {
       throw new UnsupportedOperationException(this.getClass().getName() + " only supports workunit state");
     }
     this.state = state;
+    this.helper = new InputRecordCountHelper(state);

Review comment:
       Do you need to instantiate this object here and call readExecutionCount or this can be passed down through State object ?  If you prefer doing this, we could lazily init this object whenever required to be used.




----------------------------------------------------------------
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:
users@infra.apache.org