Adding pinot file system command (#8659)

* Adding pinot file system command

* Move to picocli subcommand

pinot-tools/pom.xml

@@ -468,6 +468,16 @@
                 </extraArguments>
               </jvmSettings>
             </program>
+            <program>
+              <mainClass>org.apache.pinot.tools.admin.command.FileSystemCommand</mainClass>
+              <name>pinot-fs-util</name>
+              <jvmSettings>
+                <initialMemorySize>4G</initialMemorySize>
+                <extraArguments>
+                  <extraArgument>-Dlog4j2.configurationFile=conf/pinot-tools-log4j2.xml</extraArgument>
+                </extraArguments>
+              </jvmSettings>
+            </program>
           </programs>
           <binFileExtensions>
             <unix>.sh</unix>

pinot-tools/src/main/java/org/apache/pinot/tools/admin/PinotAdministrator.java

@@ -33,6 +33,7 @@
 import org.apache.pinot.tools.admin.command.ChangeTableState;
 import org.apache.pinot.tools.admin.command.CreateSegmentCommand;
 import org.apache.pinot.tools.admin.command.DeleteClusterCommand;
+import org.apache.pinot.tools.admin.command.FileSystemCommand;
 import org.apache.pinot.tools.admin.command.GenerateDataCommand;
 import org.apache.pinot.tools.admin.command.GitHubEventsQuickStartCommand;
 import org.apache.pinot.tools.admin.command.ImportDataCommand;
@@ -128,6 +129,7 @@ public class PinotAdministrator {
     SUBCOMMAND_MAP.put("StreamGitHubEvents", new StreamGitHubEventsCommand());
     SUBCOMMAND_MAP.put("BootstrapTable", new BootstrapTableCommand());
     SUBCOMMAND_MAP.put("SegmentProcessorFramework", new SegmentProcessorFrameworkCommand());
+    SUBCOMMAND_MAP.put("FileSystem", new FileSystemCommand());
   }
 
   @CommandLine.Option(names = {"-help", "-h", "--h", "--help"}, required = false,

pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/FileSystemCommand.java

@@ -0,0 +1,111 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.tools.admin.command;
+
+import java.util.Arrays;
+import org.apache.pinot.spi.plugin.PluginManager;
+import org.apache.pinot.tools.Command;
+import org.apache.pinot.tools.admin.command.filesystem.CopyFiles;
+import org.apache.pinot.tools.admin.command.filesystem.DeleteFiles;
+import org.apache.pinot.tools.admin.command.filesystem.ListFiles;
+import org.apache.pinot.tools.admin.command.filesystem.MoveFiles;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import picocli.CommandLine;
+
+
+@CommandLine.Command(name = "FileSystem", subcommands = {
+    ListFiles.class,
+    CopyFiles.class,
+    MoveFiles.class,
+    DeleteFiles.class
+})
+public class FileSystemCommand extends AbstractBaseAdminCommand implements Command {
+  private static final Logger LOGGER = LoggerFactory.getLogger(FileSystemCommand.class.getName());
+
+  @CommandLine.Option(names = {"-conf", "-config", "-configFile", "-config-file"}, scope =
+      CommandLine.ScopeType.INHERIT, description = "PinotFS Config file.")
+  private String _configFile;
+
+  @CommandLine.Option(names = {"-help", "-h", "--h", "--help"}, usageHelp = true, description = "Print this message.",
+      scope = CommandLine.ScopeType.INHERIT)
+  private boolean _help = false;
+
+  @CommandLine.Parameters
+  private String[] _parameters;
+
+  public FileSystemCommand setConfigFile(String configFile) {
+    _configFile = configFile;
+    return this;
+  }
+
+  public FileSystemCommand setHelp(boolean help) {
+    _help = help;
+    return this;
+  }
+
+  public String getConfigFile() {
+    return _configFile;
+  }
+
+  public FileSystemCommand setParameters(String[] parameters) {
+    _parameters = parameters;
+    return this;
+  }
+
+  @Override
+  public boolean getHelp() {
+    return _help;
+  }
+
+  @Override
+  public String getName() {
+    return "FileSystem";
+  }
+
+  @Override
+  public String toString() {
+    return "FileSystem -configFile " + _configFile;
+  }
+
+  @Override
+  public void cleanup() {
+  }
+
+  @Override
+  public String description() {
+    return "Pinot File system operation.";
+  }
+
+  @Override
+  public boolean execute()
+      throws Exception {
+    // All FileSystem commands should go to SubCommand class.
+    System.out.println("Unsupported subcommand: " + Arrays.toString(_parameters));
+    return false;
+  }
+
+  public static void main(String[] args) {
+    PluginManager.get().init();
+    int exitCode = new CommandLine(new FileSystemCommand()).execute(args);
+    if (exitCode != 0 && Boolean.parseBoolean(System.getProperties().getProperty("pinot.admin.system.exit", "true"))) {
+      System.exit(exitCode);
+    }
+  }
+}

pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/filesystem/BaseFileOperation.java

@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.tools.admin.command.filesystem;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.tools.Command;
+import org.apache.pinot.tools.admin.command.FileSystemCommand;
+import org.apache.pinot.tools.admin.command.QuickstartRunner;
+import org.apache.pinot.tools.utils.PinotConfigUtils;
+import picocli.CommandLine;
+
+
+/**
+ * Base class for file operation classes
+ */
+public abstract class BaseFileOperation implements Command {
+
+  @CommandLine.ParentCommand
+  protected FileSystemCommand _parent;
+
+  public BaseFileOperation setParent(FileSystemCommand parent) {
+    _parent = parent;
+    return this;
+  }
+
+  protected void initialPinotFS()
+      throws Exception {
+    String configFile = _parent.getConfigFile();
+    Map<String, Object> configs =
+        configFile == null ? new HashMap<>() : PinotConfigUtils.readConfigFromFile(configFile);
+    PinotFSFactory.init(new PinotConfiguration(configs));
+    QuickstartRunner.registerDefaultPinotFS();
+  }
+
+  @Override
+  public boolean getHelp() {
+    return _parent.getHelp();
+  }
+}

pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/filesystem/CopyFiles.java

@@ -0,0 +1,113 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.tools.admin.command.filesystem;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import org.apache.pinot.spi.filesystem.LocalPinotFS;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import picocli.CommandLine;
+
+
+@CommandLine.Command(name = "cp", aliases = {"copy"}, description = "Copy file from source to destination")
+public class CopyFiles extends BaseFileOperation {
+  @CommandLine.Parameters(index = "0", description = "Source file")
+  private String _source;
+
+  @CommandLine.Parameters(index = "1", description = "Destination file")
+  private String _destination;
+
+  public CopyFiles setSource(String source) {
+    _source = source;
+    return this;
+  }
+
+  public CopyFiles setDestination(String destination) {
+    _destination = destination;
+    return this;
+  }
+
+  @Override
+  public boolean execute()
+      throws Exception {
+    try {
+      super.initialPinotFS();
+    } catch (Exception e) {
+      System.err.println("Failed to initialize PinotFS, exception: " + e.getMessage());
+      return false;
+    }
+
+    URI sourceURI = URI.create(_source);
+    PinotFS sourcePinotFS = Utils.getPinotFS(sourceURI);
+
+    URI destinationURI = URI.create(_destination);
+    PinotFS destinationPinotFS = Utils.getPinotFS(destinationURI);
+
+    if (sourcePinotFS == destinationPinotFS) {
+      try {
+        sourcePinotFS.copy(sourceURI, destinationURI);
+        return true;
+      } catch (IOException e) {
+        System.err.println(
+            "Unable to copy files from " + _source + " to " + _destination + ", exception: " + e.getMessage());
+      }
+    } else if (sourcePinotFS instanceof LocalPinotFS) {
+      File localFile = new File(_source);
+      if (localFile.isDirectory()) {
+        try {
+          destinationPinotFS.copyFromLocalDir(localFile, destinationURI);
+          return true;
+        } catch (Exception e) {
+          System.err.println(
+              "Failed to copy local directory " + _source + " to " + _destination + ", exception: " + e.getMessage());
+        }
+      } else {
+        try {
+          destinationPinotFS.copyFromLocalFile(localFile, destinationURI);
+          return true;
+        } catch (Exception e) {
+          System.err.println(
+              "Failed to copy local file " + _source + " to " + _destination + ", exception: " + e.getMessage());
+        }
+      }
+    } else if (destinationPinotFS instanceof LocalPinotFS) {
+      try {
+        sourcePinotFS.copyToLocalFile(sourceURI, new File(_destination));
+        return true;
+      } catch (Exception e) {
+        System.err.println(
+            "Failed to copy remote " + _source + " to local file " + _destination + ", exception: " + e.getMessage());
+      }
+    } else {
+      System.err.println("Copying files between two different remote PinotFS is not supported");
+    }
+    return false;
+  }
+
+  @Override
+  public void printUsage() {
+    System.out.println("cp <source-uri> <destination-uri>");
+  }
+
+  @Override
+  public String description() {
+    return "Copy file from source to destination, copying from two different remote PinotFS is not supported.";
+  }
+}

pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/filesystem/DeleteFiles.java

@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.tools.admin.command.filesystem;
+
+import java.io.IOException;
+import java.net.URI;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import picocli.CommandLine;
+
+
+@CommandLine.Command(name = "rm", aliases = {"delete", "del"}, description = "Delete files")
+public class DeleteFiles extends BaseFileOperation {
+  @CommandLine.Option(names = {"-f", "--force"},
+      description = "Force delete if possible")
+  private boolean _force;
+
+  @CommandLine.Parameters(arity = "1..*", description = "File paths to delete")
+  private String[] _filePaths;
+
+  public DeleteFiles setForce(boolean force) {
+    _force = force;
+    return this;
+  }
+
+  public DeleteFiles setFilePaths(String[] filePaths) {
+    _filePaths = filePaths;
+    return this;
+  }
+
+  @Override
+  public boolean execute()
+      throws Exception {
+    try {
+      super.initialPinotFS();
+    } catch (Exception e) {
+      System.err.println("Failed to initialize PinotFS, exception: " + e.getMessage());
+      return false;
+    }
+    for (String filePath : _filePaths) {
+      URI pathURI = URI.create(filePath);
+      String scheme = Utils.getScheme(pathURI);
+      if (!PinotFSFactory.isSchemeSupported(scheme)) {
+        System.err.println("Scheme: " + scheme + " is not registered.");
+        return false;
+      }
+      try {
+        PinotFSFactory.create(scheme).delete(pathURI, _force);
+      } catch (IOException e) {
+        System.err.println("Unable to delete files under: " + pathURI + ", exception: " + e.getMessage());
+        return false;
+      }
+    }
+    return true;
+  }
+
+  @Override
+  public void printUsage() {
+    System.out.println("rm [-f] <path-uri>...");
+  }
+
+  @Override
+  public String description() {
+    return "Delete files";
+  }
+}