HBASE-22086: Space Quota issue: Deleting snapshot doesn't update the usage of table

Signed-off-by: Duo Zhang <zhangduo@apache.org>

Author: the-sakthi

hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/QuotaTableUtil.java

@@ -21,19 +21,24 @@
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.regex.Pattern;
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellScanner;
 import org.apache.hadoop.hbase.CompareOperator;
 import org.apache.hadoop.hbase.NamespaceDescriptor;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Get;
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Result;
@@ -53,6 +58,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hbase.thirdparty.com.google.common.collect.HashMultimap;
+import org.apache.hbase.thirdparty.com.google.common.collect.Multimap;
 import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
 import org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
 import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
@@ -540,6 +547,87 @@ static Put createPutForNamespaceSnapshotSize(String namespace, long size) {
     return p;
   }
 
+  /**
+   * Returns a list of {@code Delete} to remove given table snapshot
+   * entries to remove from quota table
+   * @param snapshotEntriesToRemove the entries to remove
+   */
+  static List<Delete> createDeletesForExistingTableSnapshotSizes(
+      Multimap<TableName, String> snapshotEntriesToRemove) {
+    List<Delete> deletes = new ArrayList<>();
+    for (Map.Entry<TableName, Collection<String>> entry : snapshotEntriesToRemove.asMap()
+        .entrySet()) {
+      for (String snapshot : entry.getValue()) {
+        Delete d = new Delete(getTableRowKey(entry.getKey()));
+        d.addColumns(QUOTA_FAMILY_USAGE,
+            Bytes.add(QUOTA_SNAPSHOT_SIZE_QUALIFIER, Bytes.toBytes(snapshot)));
+        deletes.add(d);
+      }
+    }
+    return deletes;
+  }
+
+  /**
+   * Returns a list of {@code Delete} to remove all table snapshot entries from quota table.
+   * @param connection connection to re-use
+   */
+  static List<Delete> createDeletesForExistingTableSnapshotSizes(Connection connection)
+      throws IOException {
+    return createDeletesForExistingSnapshotsFromScan(connection, createScanForSpaceSnapshotSizes());
+  }
+
+  /**
+   * Returns a list of {@code Delete} to remove given namespace snapshot
+   * entries to removefrom quota table
+   * @param snapshotEntriesToRemove the entries to remove
+   */
+  static List<Delete> createDeletesForExistingNamespaceSnapshotSizes(
+      Set<String> snapshotEntriesToRemove) {
+    List<Delete> deletes = new ArrayList<>();
+    for (String snapshot : snapshotEntriesToRemove) {
+      Delete d = new Delete(getNamespaceRowKey(snapshot));
+      d.addColumns(QUOTA_FAMILY_USAGE, QUOTA_SNAPSHOT_SIZE_QUALIFIER);
+      deletes.add(d);
+    }
+    return deletes;
+  }
+
+  /**
+   * Returns a list of {@code Delete} to remove all namespace snapshot entries from quota table.
+   * @param connection connection to re-use
+   */
+  static List<Delete> createDeletesForExistingNamespaceSnapshotSizes(Connection connection)
+      throws IOException {
+    return createDeletesForExistingSnapshotsFromScan(connection,
+        createScanForNamespaceSnapshotSizes());
+  }
+
+  /**
+   * Returns a list of {@code Delete} to remove all entries returned by the passed scanner.
+   * @param connection connection to re-use
+   * @param scan the scanner to use to generate the list of deletes
+   */
+  static List<Delete> createDeletesForExistingSnapshotsFromScan(Connection connection, Scan scan)
+      throws IOException {
+    List<Delete> deletes = new ArrayList<>();
+    try (Table quotaTable = connection.getTable(QUOTA_TABLE_NAME);
+        ResultScanner rs = quotaTable.getScanner(scan)) {
+      for (Result r : rs) {
+        CellScanner cs = r.cellScanner();
+        while (cs.advance()) {
+          Cell c = cs.current();
+          byte[] family = Bytes.copy(c.getFamilyArray(), c.getFamilyOffset(), c.getFamilyLength());
+          byte[] qual =
+              Bytes.copy(c.getQualifierArray(), c.getQualifierOffset(), c.getQualifierLength());
+          Delete d = new Delete(r.getRow());
+          d.addColumns(family, qual);
+          deletes.add(d);
+        }
+      }
+      return deletes;
+    }
+  }
+
   /**
    * Fetches the computed size of all snapshots against tables in a namespace for space quotas.
    */
@@ -575,6 +663,34 @@ static long parseSnapshotSize(Cell c) throws InvalidProtocolBufferException {
     return QuotaProtos.SpaceQuotaSnapshot.parseFrom(bs).getQuotaUsage();
   }
 
+  /**
+   * Returns a scanner for all existing namespace snapshot entries.
+   */
+  static Scan createScanForNamespaceSnapshotSizes() {
+    return createScanForNamespaceSnapshotSizes(null);
+  }
+
+  /**
+   * Returns a scanner for all namespace snapshot entries of the given namespace
+   * @param namespace name of the namespace whose snapshot entries are to be scanned
+   */
+  static Scan createScanForNamespaceSnapshotSizes(String namespace) {
+    Scan s = new Scan();
+    if (namespace == null || namespace.isEmpty()) {
+      // Read all namespaces, just look at the row prefix
+      s.setRowPrefixFilter(QUOTA_NAMESPACE_ROW_KEY_PREFIX);
+    } else {
+      // Fetch the exact row for the table
+      byte[] rowkey = getNamespaceRowKey(namespace);
+      // Fetch just this one row
+      s.withStartRow(rowkey).withStopRow(rowkey, true);
+    }
+
+    // Just the usage family and only the snapshot size qualifiers
+    return s.addFamily(QUOTA_FAMILY_USAGE)
+        .setFilter(new ColumnPrefixFilter(QUOTA_SNAPSHOT_SIZE_QUALIFIER));
+  }
+
   static Scan createScanForSpaceSnapshotSizes() {
     return createScanForSpaceSnapshotSizes(null);
   }
@@ -621,6 +737,46 @@ public static Map<String,Long> getObservedSnapshotSizes(Connection conn) throws
     }
   }
 
+  /**
+   * Returns a multimap for all existing table snapshot entries.
+   * @param conn connection to re-use
+   */
+  public static Multimap<TableName, String> getTableSnapshots(Connection conn) throws IOException {
+    try (Table quotaTable = conn.getTable(QUOTA_TABLE_NAME);
+        ResultScanner rs = quotaTable.getScanner(createScanForSpaceSnapshotSizes())) {
+      Multimap<TableName, String> snapshots = HashMultimap.create();
+      for (Result r : rs) {
+        CellScanner cs = r.cellScanner();
+        while (cs.advance()) {
+          Cell c = cs.current();
+
+          final String snapshot = extractSnapshotNameFromSizeCell(c);
+          snapshots.put(getTableFromRowKey(r.getRow()), snapshot);
+        }
+      }
+      return snapshots;
+    }
+  }
+
+  /**
+   * Returns a set of the names of all namespaces containing snapshot entries.
+   * @param conn connection to re-use
+   */
+  public static Set<String> getNamespaceSnapshots(Connection conn) throws IOException {
+    try (Table quotaTable = conn.getTable(QUOTA_TABLE_NAME);
+        ResultScanner rs = quotaTable.getScanner(createScanForNamespaceSnapshotSizes())) {
+      Set<String> snapshots = new HashSet<>();
+      for (Result r : rs) {
+        CellScanner cs = r.cellScanner();
+        while (cs.advance()) {
+          cs.current();
+          snapshots.add(getNamespaceFromRowKey(r.getRow()));
+        }
+      }
+      return snapshots;
+    }
+  }
+
   /**
    * Returns the current space quota snapshot of the given {@code tableName} from
    * {@code QuotaTableUtil.QUOTA_TABLE_NAME} or null if the no quota information is available for

hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/SnapshotQuotaObserverChore.java

@@ -21,6 +21,7 @@
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
@@ -37,6 +38,7 @@
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.master.HMaster;
 import org.apache.hadoop.hbase.master.MetricsMaster;
@@ -110,6 +112,12 @@ void _chore() throws IOException {
       metrics.incrementSnapshotFetchTime((System.nanoTime() - start) / 1_000_000);
     }
 
+    // Remove old table snapshots data
+    pruneTableSnapshots(snapshotsToComputeSize);
+
+    // Remove old namespace snapshots data
+    pruneNamespaceSnapshots(snapshotsToComputeSize);
+
     // For each table, compute the size of each snapshot
     Map<String,Long> namespaceSnapshotSizes = computeSnapshotSizes(snapshotsToComputeSize);
 
@@ -118,6 +126,43 @@ void _chore() throws IOException {
     persistSnapshotSizesForNamespaces(namespaceSnapshotSizes);
   }
 
+  /**
+   * Removes the snapshot entries that are present in Quota table but not in snapshotsToComputeSize
+   *
+   * @param snapshotsToComputeSize list of snapshots to be persisted
+   */
+  void pruneTableSnapshots(Multimap<TableName, String> snapshotsToComputeSize) throws IOException {
+    Multimap<TableName, String> existingSnapshotEntries = QuotaTableUtil.getTableSnapshots(conn);
+    Multimap<TableName, String> snapshotEntriesToRemove = HashMultimap.create();
+    for (Entry<TableName, Collection<String>> entry : existingSnapshotEntries.asMap().entrySet()) {
+      TableName tn = entry.getKey();
+      Set<String> setOfSnapshots = new HashSet<>(entry.getValue());
+      for (String snapshot : snapshotsToComputeSize.get(tn)) {
+        setOfSnapshots.remove(snapshot);
+      }
+
+      for (String snapshot : setOfSnapshots) {
+        snapshotEntriesToRemove.put(tn, snapshot);
+      }
+    }
+    removeExistingTableSnapshotSizes(snapshotEntriesToRemove);
+  }
+
+  /**
+   * Removes the snapshot entries that are present in Quota table but not in snapshotsToComputeSize
+   *
+   * @param snapshotsToComputeSize list of snapshots to be persisted
+   */
+  void pruneNamespaceSnapshots(Multimap<TableName, String> snapshotsToComputeSize)
+      throws IOException {
+    Set<String> existingSnapshotEntries = QuotaTableUtil.getNamespaceSnapshots(conn);
+    for (TableName tableName : snapshotsToComputeSize.keySet()) {
+      existingSnapshotEntries.remove(tableName.getNamespaceAsString());
+    }
+    // here existingSnapshotEntries is left with the entries to be removed
+    removeExistingNamespaceSnapshotSizes(existingSnapshotEntries);
+  }
+
   /**
    * Fetches each table with a quota (table or namespace quota), and then fetch the name of each
    * snapshot which was created from that table.
@@ -220,6 +265,24 @@ void persistSnapshotSizesForNamespaces(
     }
   }
 
+  void removeExistingTableSnapshotSizes(Multimap<TableName, String> snapshotEntriesToRemove)
+      throws IOException {
+    removeExistingSnapshotSizes(
+        QuotaTableUtil.createDeletesForExistingTableSnapshotSizes(snapshotEntriesToRemove));
+  }
+
+  void removeExistingNamespaceSnapshotSizes(Set<String> snapshotEntriesToRemove)
+      throws IOException {
+    removeExistingSnapshotSizes(
+        QuotaTableUtil.createDeletesForExistingNamespaceSnapshotSizes(snapshotEntriesToRemove));
+  }
+
+  void removeExistingSnapshotSizes(List<Delete> deletes) throws IOException {
+    try (Table quotaTable = conn.getTable(QuotaUtil.QUOTA_TABLE_NAME)) {
+      quotaTable.delete(deletes);
+    }
+  }
+
   /**
    * Extracts the period for the chore from the configuration.
    *