[HUDI-3283] Bootstrap support overwrite existing table (#4647)

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/BootstrapExecutor.java

@@ -97,6 +97,7 @@ public class BootstrapExecutor  implements Serializable {
 
   /**
    * Bootstrap Executor.
+   *
    * @param cfg        DeltaStreamer Config
    * @param jssc       Java Spark Context
    * @param fs         File System
@@ -168,10 +169,16 @@ public class BootstrapExecutor  implements Serializable {
   }
 
   private void initializeTable() throws IOException {
-    if (fs.exists(new Path(cfg.targetBasePath))) {
+    Path basePath = new Path(cfg.targetBasePath);
+    if (fs.exists(basePath)) {
+      if (cfg.bootstrapOverwrite) {
+        LOG.warn("Target base path already exists, overwrite it");
+        fs.delete(basePath, true);
+      } else {
         throw new HoodieException("target base path already exists at " + cfg.targetBasePath
             + ". Cannot bootstrap data on top of an existing table");
       }
+    }
     HoodieTableMetaClient.withPropertyBuilder()
         .setTableType(cfg.tableType)
         .setTableName(cfg.targetTableName)

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java

@@ -363,6 +363,9 @@ public class HoodieDeltaStreamer implements Serializable {
     @Parameter(names = {"--run-bootstrap"}, description = "Run bootstrap if bootstrap index is not found")
     public Boolean runBootstrap = false;
 
+    @Parameter(names = {"--bootstrap-overwrite"}, description = "Overwrite existing target table, default false")
+    public Boolean bootstrapOverwrite = false;
+
     @Parameter(names = {"--bootstrap-index-class"}, description = "subclass of BootstrapIndex")
     public String bootstrapIndexClass = HFileBootstrapIndex.class.getName();