From 15b277daa6d8fe11c50d87b7e6a73712e2ba6661 Mon Sep 17 00:00:00 2001
From: Qifeng <qifeng.bai@csiro.au>
Date: Fri, 21 Jan 2022 15:18:21 +1100
Subject: [PATCH] #622 impl(Only calculated new added records)

---
 .../java/au/org/ala/utils/ALAFsUtils.java     | 38 ++++++++++++++++---
 1 file changed, 33 insertions(+), 5 deletions(-)

diff --git a/livingatlas/pipelines/src/main/java/au/org/ala/utils/ALAFsUtils.java b/livingatlas/pipelines/src/main/java/au/org/ala/utils/ALAFsUtils.java
index 8885c9a449..3da5fdfd24 100644
--- a/livingatlas/pipelines/src/main/java/au/org/ala/utils/ALAFsUtils.java
+++ b/livingatlas/pipelines/src/main/java/au/org/ala/utils/ALAFsUtils.java
@@ -72,10 +72,40 @@ public static String buildPathSamplingUsingTargetPath(AllDatasetsPipelinesOption
   }
 
   /**
-   * Build a path to outlier records. {fsPath}/pipelines-outlier/{datasetId}
+   * NOTE: It will delete the existing folder Build a path to outlier records.
+   * {fsPath}/pipelines-outlier/{datasetId} {fsPath}/pipelines-outlier/all
+   */
+  public static String buildPathOutlierUsingTargetPath(
+      AllDatasetsPipelinesOptions options, boolean delete) throws IOException {
+    // default: {fsPath}/pipelines-outlier
+    FileSystem fs =
+        FileSystemFactory.getInstance(options.getHdfsSiteConfig(), options.getCoreSiteConfig())
+            .getFs(options.getTargetPath());
+
+    String outputPath = PathBuilder.buildPath(options.getTargetPath()).toString();
+
+    // {fsPath}/pipelines-outlier/{datasetId}
+    if (options.getDatasetId() != null && !"all".equalsIgnoreCase(options.getDatasetId())) {
+      outputPath = PathBuilder.buildPath(outputPath, options.getDatasetId()).toString();
+    } else {
+      // {fsPath}/pipelines-outlier/all
+      outputPath = PathBuilder.buildPath(outputPath, "all").toString();
+    }
+    // delete previous runs
+    if (delete)
+      FsUtils.deleteIfExist(options.getHdfsSiteConfig(), options.getCoreSiteConfig(), outputPath);
+    else {
+      if (!exists(fs, outputPath)) ALAFsUtils.createDirectory(fs, outputPath);
+    }
+
+    return outputPath;
+  }
+
+  /**
+   * Get an output path to outlier records. {fsPath}/pipelines-outlier/{datasetId}
    * {fsPath}/pipelines-outlier/all
    */
-  public static String buildPathOutlierUsingTargetPath(AllDatasetsPipelinesOptions options)
+  public static String getOutlierTargetPath(AllDatasetsPipelinesOptions options)
       throws IOException {
     // default: {fsPath}/pipelines-outlier
     FileSystem fs =
@@ -91,9 +121,7 @@ public static String buildPathOutlierUsingTargetPath(AllDatasetsPipelinesOptions
       // {fsPath}/pipelines-outlier/all
       outputPath = PathBuilder.buildPath(outputPath, "all").toString();
     }
-    // delete previous runs
-    FsUtils.deleteIfExist(options.getHdfsSiteConfig(), options.getCoreSiteConfig(), outputPath);
-    ALAFsUtils.createDirectory(fs, outputPath);
+
     return outputPath;
   }