HBASE-29133: Implement "pitr" Command for Point-in-Time Restore (#6717)

Signed-off-by: Andor Molnar <andor@apache.org>
Signed-off-by: Tak Lon (Stephen) Wu <taklwu@apache.org>

Author: vinayakphegde

bin/hbase

@@ -102,6 +102,7 @@ show_usage() {
   echo "  version          Print the version"
   echo "  backup           Backup tables for recovery"
   echo "  restore          Restore tables from existing backup image"
+  echo "  pitr             Restore tables to a specific point in time using backup and WAL replay"
   echo "  completebulkload Run BulkLoadHFiles tool"
   echo "  regionsplitter   Run RegionSplitter tool"
   echo "  rowcounter       Run RowCounter tool"
@@ -639,6 +640,22 @@ elif [ "$COMMAND" = "restore" ] ; then
       fi
     done
   fi
+elif [ "$COMMAND" = "pitr" ] ; then
+  CLASS='org.apache.hadoop.hbase.backup.PointInTimeRestoreDriver'
+  if [ -n "${shaded_jar}" ] ; then
+    for f in "${HBASE_HOME}"/lib/hbase-backup*.jar; do
+      if [ -f "${f}" ]; then
+        CLASSPATH="${CLASSPATH}:${f}"
+        break
+      fi
+    done
+    for f in "${HBASE_HOME}"/lib/commons-lang3*.jar; do
+      if [ -f "${f}" ]; then
+        CLASSPATH="${CLASSPATH}:${f}"
+        break
+      fi
+    done
+  fi
 elif [ "$COMMAND" = "upgrade" ] ; then
   echo "This command was used to upgrade to HBase 0.96, it was removed in HBase 2.0.0."
   echo "Please follow the documentation at http://hbase.apache.org/book.html#upgrading."

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/AbstractRestoreDriver.java

@@ -0,0 +1,232 @@
+/*
+ * 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.hadoop.hbase.backup;
+
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_CHECK;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_CHECK_DESC;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_DEBUG;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_DEBUG_DESC;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_OVERWRITE;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_OVERWRITE_DESC;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_SET;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_SET_RESTORE_DESC;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_TABLE;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_TABLE_LIST_DESC;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_TABLE_MAPPING;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_TABLE_MAPPING_DESC;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_YARN_QUEUE_NAME;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_YARN_QUEUE_NAME_RESTORE_DESC;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.backup.impl.BackupManager;
+import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
+import org.apache.hadoop.hbase.backup.util.BackupUtils;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.logging.Log4jUtils;
+import org.apache.hadoop.hbase.util.AbstractHBaseTool;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine;
+import org.apache.hbase.thirdparty.org.apache.commons.cli.HelpFormatter;
+
+@InterfaceAudience.Private
+public abstract class AbstractRestoreDriver extends AbstractHBaseTool {
+  protected static final Logger LOG = LoggerFactory.getLogger(AbstractRestoreDriver.class);
+  protected CommandLine cmd;
+
+  protected static final String USAGE_FOOTER = "";
+
+  protected AbstractRestoreDriver() {
+    init();
+  }
+
+  protected void init() {
+    Log4jUtils.disableZkAndClientLoggers();
+  }
+
+  protected abstract int executeRestore(boolean check, TableName[] fromTables, TableName[] toTables,
+    boolean isOverwrite);
+
+  private int parseAndRun() throws IOException {
+    if (!BackupManager.isBackupEnabled(getConf())) {
+      System.err.println(BackupRestoreConstants.ENABLE_BACKUP);
+      return -1;
+    }
+
+    if (cmd.hasOption(OPTION_DEBUG)) {
+      Log4jUtils.setLogLevel("org.apache.hadoop.hbase.backup", "DEBUG");
+    }
+
+    boolean overwrite = cmd.hasOption(OPTION_OVERWRITE);
+    if (overwrite) {
+      LOG.debug("Found overwrite option (-{}) in restore command, "
+        + "will overwrite to existing table if any in the restore target", OPTION_OVERWRITE);
+    }
+
+    boolean check = cmd.hasOption(OPTION_CHECK);
+    if (check) {
+      LOG.debug(
+        "Found check option (-{}) in restore command, will check and verify the dependencies",
+        OPTION_CHECK);
+    }
+
+    if (cmd.hasOption(OPTION_SET) && cmd.hasOption(OPTION_TABLE)) {
+      System.err.printf(
+        "Set name (-%s) and table list (-%s) are mutually exclusive, you can not specify both "
+          + "of them.%n",
+        OPTION_SET, OPTION_TABLE);
+      printToolUsage();
+      return -1;
+    }
+
+    if (!cmd.hasOption(OPTION_SET) && !cmd.hasOption(OPTION_TABLE)) {
+      System.err.printf(
+        "You have to specify either set name (-%s) or table list (-%s) to " + "restore%n",
+        OPTION_SET, OPTION_TABLE);
+      printToolUsage();
+      return -1;
+    }
+
+    if (cmd.hasOption(OPTION_YARN_QUEUE_NAME)) {
+      String queueName = cmd.getOptionValue(OPTION_YARN_QUEUE_NAME);
+      // Set MR job queuename to configuration
+      getConf().set("mapreduce.job.queuename", queueName);
+    }
+
+    String tables;
+    TableName[] sTableArray;
+    TableName[] tTableArray;
+
+    String tableMapping = cmd.getOptionValue(OPTION_TABLE_MAPPING);
+
+    try (final Connection conn = ConnectionFactory.createConnection(conf)) {
+      // Check backup set
+      if (cmd.hasOption(OPTION_SET)) {
+        String setName = cmd.getOptionValue(OPTION_SET);
+        try {
+          tables = getTablesForSet(conn, setName);
+        } catch (IOException e) {
+          System.out.println("ERROR: " + e.getMessage() + " for setName=" + setName);
+          printToolUsage();
+          return -2;
+        }
+        if (tables == null) {
+          System.out
+            .println("ERROR: Backup set '" + setName + "' is either empty or does not exist");
+          printToolUsage();
+          return -3;
+        }
+      } else {
+        tables = cmd.getOptionValue(OPTION_TABLE);
+      }
+
+      sTableArray = BackupUtils.parseTableNames(tables);
+      tTableArray = BackupUtils.parseTableNames(tableMapping);
+
+      if (
+        sTableArray != null && tTableArray != null && (sTableArray.length != tTableArray.length)
+      ) {
+        System.out.println("ERROR: table mapping mismatch: " + tables + " : " + tableMapping);
+        printToolUsage();
+        return -4;
+      }
+    }
+
+    return executeRestore(check, sTableArray, tTableArray, overwrite);
+  }
+
+  @Override
+  protected void addOptions() {
+    addOptNoArg(OPTION_OVERWRITE, OPTION_OVERWRITE_DESC);
+    addOptNoArg(OPTION_CHECK, OPTION_CHECK_DESC);
+    addOptNoArg(OPTION_DEBUG, OPTION_DEBUG_DESC);
+    addOptWithArg(OPTION_SET, OPTION_SET_RESTORE_DESC);
+    addOptWithArg(OPTION_TABLE, OPTION_TABLE_LIST_DESC);
+    addOptWithArg(OPTION_TABLE_MAPPING, OPTION_TABLE_MAPPING_DESC);
+    addOptWithArg(OPTION_YARN_QUEUE_NAME, OPTION_YARN_QUEUE_NAME_RESTORE_DESC);
+  }
+
+  @Override
+  protected void processOptions(CommandLine cmd) {
+    this.cmd = cmd;
+  }
+
+  @Override
+  protected int doWork() throws Exception {
+    return parseAndRun();
+  }
+
+  @Override
+  public int run(String[] args) {
+    Objects.requireNonNull(conf, "Tool configuration is not initialized");
+
+    try {
+      cmd = parseArgs(args);
+    } catch (Exception e) {
+      System.out.println("Error parsing command-line arguments: " + e.getMessage());
+      printToolUsage();
+      return EXIT_FAILURE;
+    }
+
+    if (cmd.hasOption(SHORT_HELP_OPTION) || cmd.hasOption(LONG_HELP_OPTION)) {
+      printToolUsage();
+      return EXIT_FAILURE;
+    }
+
+    processOptions(cmd);
+
+    try {
+      return doWork();
+    } catch (Exception e) {
+      LOG.error("Error running restore tool", e);
+      return EXIT_FAILURE;
+    }
+  }
+
+  protected void printToolUsage() {
+    System.out.println(getUsageString());
+    HelpFormatter helpFormatter = new HelpFormatter();
+    helpFormatter.setLeftPadding(2);
+    helpFormatter.setDescPadding(8);
+    helpFormatter.setWidth(100);
+    helpFormatter.setSyntaxPrefix("Options:");
+    helpFormatter.printHelp(" ", null, options, USAGE_FOOTER);
+    System.out.println(BackupRestoreConstants.VERIFY_BACKUP);
+  }
+
+  protected abstract String getUsageString();
+
+  private String getTablesForSet(Connection conn, String name) throws IOException {
+    try (final BackupSystemTable table = new BackupSystemTable(conn)) {
+      List<TableName> tables = table.describeBackupSet(name);
+
+      if (tables == null) {
+        return null;
+      }
+
+      return StringUtils.join(tables, BackupRestoreConstants.TABLENAME_DELIMITER_IN_COMMAND);
+    }
+  }
+}

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/BackupAdmin.java

@@ -51,6 +51,13 @@ public interface BackupAdmin extends Closeable {
    */
   void restore(RestoreRequest request) throws IOException;
 
+  /**
+   * Restore the tables to specific time
+   * @param request Point in Time restore request
+   * @throws IOException exception
+   */
+  void pointInTimeRestore(PointInTimeRestoreRequest request) throws IOException;
+
   /**
    * Describe backup image command
    * @param backupId backup id

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/BackupRestoreConstants.java

@@ -109,6 +109,16 @@ public interface BackupRestoreConstants {
   String OPTION_FORCE_DELETE_DESC =
     "Flag to forcefully delete the backup, even if it may be required for Point-in-Time Restore";
 
+  String OPTION_TO_DATETIME = "td";
+  String LONG_OPTION_TO_DATETIME = "to-datetime";
+  String OPTION_TO_DATETIME_DESC = "Target date and time up to which data should be restored";
+
+  String OPTION_PITR_BACKUP_PATH = "bp";
+  String LONG_OPTION_PITR_BACKUP_PATH = "backup-path";
+  String OPTION_PITR_BACKUP_PATH_DESC =
+    "Specifies a custom backup location for Point-In-Time Recovery (PITR). "
+      + "If provided, this location will be used exclusively instead of deriving the path from the system table.";
+
   String JOB_NAME_CONF_KEY = "mapreduce.job.name";
 
   String BACKUP_CONFIG_STRING = BackupRestoreConstants.BACKUP_ENABLE_KEY + "=true\n"

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/HBackupFileSystem.java

@@ -17,13 +17,21 @@
  */
 package org.apache.hadoop.hbase.backup;
 
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.BACKUPID_PREFIX;
+
 import com.google.errorprone.annotations.RestrictedApi;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.backup.impl.BackupManifest;
+import org.apache.hadoop.hbase.backup.impl.BackupManifest.BackupImage;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -135,4 +143,36 @@ public static BackupManifest getManifest(Configuration conf, Path backupRootPath
       new BackupManifest(conf, getManifestPath(conf, backupRootPath, backupId));
     return manifest;
   }
+
+  public static List<BackupImage> getAllBackupImages(Configuration conf, Path backupRootPath)
+    throws IOException {
+    FileSystem fs = FileSystem.get(backupRootPath.toUri(), conf);
+    RemoteIterator<LocatedFileStatus> it = fs.listLocatedStatus(backupRootPath);
+
+    List<BackupImage> images = new ArrayList<>();
+
+    while (it.hasNext()) {
+      LocatedFileStatus lfs = it.next();
+      if (!lfs.isDirectory()) {
+        continue;
+      }
+
+      String backupId = lfs.getPath().getName();
+      try {
+        BackupManifest manifest = getManifest(conf, backupRootPath, backupId);
+        images.add(manifest.getBackupImage());
+      } catch (IOException e) {
+        LOG.error("Cannot load backup manifest from: " + lfs.getPath(), e);
+      }
+    }
+
+    // Sort images by timestamp in descending order
+    images.sort(Comparator.comparingLong(m -> -getTimestamp(m.getBackupId())));
+
+    return images;
+  }
+
+  private static long getTimestamp(String backupId) {
+    return Long.parseLong(backupId.substring(BACKUPID_PREFIX.length()));
+  }
 }