HBASE-28216 HDFS erasure coding support for table data dirs

hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptor.java

@@ -266,6 +266,15 @@ public interface TableDescriptor {
    */
   boolean isReadOnly();
 
+  /**
+   * The HDFS erasure coding policy for a table. This will be set on the data dir of the table, and
+   * is an alternative to normal replication which takes less space at the cost of locality.
+   * @return the current policy, or null if undefined
+   */
+  default String getErasureCodingPolicy() {
+    return null;
+  }
+
   /**
    * Returns Name of this table and then a map of all of the column family descriptors (with only
    * the non-default column family attributes)

hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java

@@ -143,6 +143,14 @@ public class TableDescriptorBuilder {
   private static final Bytes REGION_MEMSTORE_REPLICATION_KEY =
     new Bytes(Bytes.toBytes(REGION_MEMSTORE_REPLICATION));
 
+  /**
+   * If non-null, the HDFS erasure coding policy to set on the data dir of the table
+   */
+  public static final String ERASURE_CODING_POLICY = "ERASURE_CODING_POLICY";
+  private static final Bytes ERASURE_CODING_POLICY_KEY =
+    new Bytes(Bytes.toBytes(ERASURE_CODING_POLICY));
+
+  private static final String DEFAULT_ERASURE_CODING_POLICY = null;
   /**
    * Used by shell/rest interface to access this metadata attribute which denotes if the table
    * should be treated by region normalizer.
@@ -226,6 +234,7 @@ public class TableDescriptorBuilder {
     DEFAULT_VALUES.put(DURABILITY, DEFAULT_DURABLITY.name()); // use the enum name
     DEFAULT_VALUES.put(REGION_REPLICATION, String.valueOf(DEFAULT_REGION_REPLICATION));
     DEFAULT_VALUES.put(PRIORITY, String.valueOf(DEFAULT_PRIORITY));
+    DEFAULT_VALUES.put(ERASURE_CODING_POLICY, String.valueOf(DEFAULT_ERASURE_CODING_POLICY));
     DEFAULT_VALUES.keySet().stream().map(s -> new Bytes(Bytes.toBytes(s)))
       .forEach(RESERVED_KEYWORDS::add);
     RESERVED_KEYWORDS.add(IS_META_KEY);
@@ -490,6 +499,11 @@ public TableDescriptorBuilder setReadOnly(final boolean readOnly) {
     return this;
   }
 
+  public TableDescriptorBuilder setErasureCodingPolicy(String policy) {
+    desc.setErasureCodingPolicy(policy);
+    return this;
+  }
+
   public TableDescriptorBuilder setRegionMemStoreReplication(boolean memstoreReplication) {
     desc.setRegionMemStoreReplication(memstoreReplication);
     return this;
@@ -748,6 +762,28 @@ public ModifyableTableDescriptor setReadOnly(final boolean readOnly) {
       return setValue(READONLY_KEY, Boolean.toString(readOnly));
     }
 
+    /**
+     * The HDFS erasure coding policy for a table. This will be set on the data dir of the table,
+     * and is an alternative to normal replication which takes less space at the cost of locality.
+     * @return the current policy, or null if undefined
+     */
+    @Override
+    public String getErasureCodingPolicy() {
+      return getValue(ERASURE_CODING_POLICY);
+    }
+
+    /**
+     * Sets the HDFS erasure coding policy for the table. This will be propagated to HDFS for the
+     * data dir of the table. Erasure coding is an alternative to normal replication which takes
+     * less space at the cost of locality. The policy must be available and enabled on the hdfs
+     * cluster before being set.
+     * @param policy the policy to set, or null to disable erasure coding
+     * @return the modifyable TD
+     */
+    public ModifyableTableDescriptor setErasureCodingPolicy(String policy) {
+      return setValue(ERASURE_CODING_POLICY_KEY, policy);
+    }
+
     /**
      * Check if the compaction enable flag of the table is true. If flag is false then no
      * minor/major compactions will be done in real.

hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestTableDescriptorBuilder.java

@@ -362,10 +362,10 @@ public void testStringCustomizedValues() throws HBaseException {
       htd.toStringCustomizedValues());
 
     htd = TableDescriptorBuilder.newBuilder(htd).setMaxFileSize("10737942528")
-      .setMemStoreFlushSize("256MB").build();
+      .setMemStoreFlushSize("256MB").setErasureCodingPolicy("RS-6-3-1024k").build();
     assertEquals(
       "'testStringCustomizedValues', " + "{TABLE_ATTRIBUTES => {DURABILITY => 'ASYNC_WAL', "
-        + "MAX_FILESIZE => '10737942528 B (10GB 512KB)', "
+        + "ERASURE_CODING_POLICY => 'RS-6-3-1024k', MAX_FILESIZE => '10737942528 B (10GB 512KB)', "
         + "MEMSTORE_FLUSHSIZE => '268435456 B (256MB)'}}, "
         + "{NAME => 'cf', BLOCKSIZE => '131072 B (128KB)'}",
       htd.toStringCustomizedValues());

hbase-protocol-shaded/src/main/protobuf/server/master/MasterProcedure.proto

@@ -56,6 +56,7 @@ enum CreateTableState {
   CREATE_TABLE_ASSIGN_REGIONS = 4;
   CREATE_TABLE_UPDATE_DESC_CACHE = 5;
   CREATE_TABLE_POST_OPERATION = 6;
+  CREATE_TABLE_SET_ERASURE_CODING_POLICY = 7;
 }
 
 message CreateTableStateData {
@@ -74,6 +75,7 @@ enum ModifyTableState {
   MODIFY_TABLE_REOPEN_ALL_REGIONS = 7;
   MODIFY_TABLE_CLOSE_EXCESS_REPLICAS = 8;
   MODIFY_TABLE_ASSIGN_NEW_REPLICAS = 9;
+  MODIFY_TABLE_SYNC_ERASURE_CODING_POLICY = 10;
 }
 
 message ModifyTableStateData {
@@ -267,6 +269,7 @@ enum CloneSnapshotState {
   CLONE_SNAPSHOT_UPDATE_DESC_CACHE = 5;
   CLONE_SNAPSHOT_POST_OPERATION = 6;
   CLONE_SNAPHOST_RESTORE_ACL = 7;
+  CLONE_SNAPSHOT_SET_ERASURE_CODING_POLICY = 8;
 }
 
 message CloneSnapshotStateData {
@@ -285,6 +288,7 @@ enum RestoreSnapshotState {
   RESTORE_SNAPSHOT_WRITE_FS_LAYOUT = 3;
   RESTORE_SNAPSHOT_UPDATE_META = 4;
   RESTORE_SNAPSHOT_RESTORE_ACL = 5;
+  RESTORE_SNAPSHOT_SYNC_ERASURE_CODING_POLICY = 6;
 }
 
 message RestoreSnapshotStateData {
@@ -296,6 +300,7 @@ message RestoreSnapshotStateData {
   repeated RegionInfo region_info_for_add = 6;
   repeated RestoreParentToChildRegionsPair parent_to_child_regions_pair_list = 7;
   optional bool restore_acl = 8;
+  required TableSchema old_table_schema = 9;
 }
 
 enum DispatchMergingRegionsState {

hbase-server/src/main/java/org/apache/hadoop/hbase/fs/ErasureCodingUtils.java

@@ -0,0 +1,185 @@
+/*
+ * 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.fs;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.HBaseIOException;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicyInfo;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@InterfaceAudience.Private
+public final class ErasureCodingUtils {
+
+  private ErasureCodingUtils() {
+  }
+
+  private static final Logger LOG = LoggerFactory.getLogger(ErasureCodingUtils.class);
+
+  /**
+   * Runs checks against the FileSystem, verifying that HDFS is supported and the policy is
+   * available, enabled, and works with a simple write.
+   */
+  public static void verifySupport(Configuration conf, String policy) throws HBaseIOException {
+    DistributedFileSystem dfs = getDfs(conf);
+    checkAvailable(dfs, policy);
+
+    // Enable the policy on a test directory. Try writing ot it to ensure that HDFS allows it
+    // This acts as a safeguard against topology issues (not enough nodes for policy, etc) and
+    // anything else. This is otherwise hard to validate more directly.
+    Path globalTempDir = new Path(conf.get(HConstants.HBASE_DIR), HConstants.HBASE_TEMP_DIRECTORY);
+    Path currentTempDir = createTempDir(dfs, globalTempDir);
+    try {
+      setPolicy(dfs, currentTempDir, policy);
+      try (FSDataOutputStream out = dfs.create(new Path(currentTempDir, "test.out"))) {
+        out.writeUTF("Testing " + policy);
+      }
+    } catch (IOException e) {
+      throw new DoNotRetryIOException("Failed write test for EC policy. Check cause or logs", e);
+    } finally {
+      try {
+        dfs.delete(currentTempDir, true);
+      } catch (IOException e) {
+        LOG.warn("Failed to delete temp path for ec test", e);
+      }
+    }
+  }
+
+  private static Path createTempDir(FileSystem fs, Path tempDir) throws HBaseIOException {
+    Path currentTempDir = new Path(tempDir, "ec-test-" + System.currentTimeMillis());
+    try {
+      fs.mkdirs(currentTempDir);
+      fs.deleteOnExit(currentTempDir);
+    } catch (IOException e) {
+      throw new HBaseIOException("Failed to create test dir for EC write test", e);
+    }
+    return currentTempDir;
+  }
+
+  private static void checkAvailable(DistributedFileSystem dfs, String policy)
+    throws HBaseIOException {
+    Collection<ErasureCodingPolicyInfo> policies;
+    try {
+      policies = dfs.getAllErasureCodingPolicies();
+    } catch (IOException e) {
+      throw new HBaseIOException("Failed to check for Erasure Coding policy: " + policy, e);
+    }
+    for (ErasureCodingPolicyInfo policyInfo : policies) {
+      if (policyInfo.getPolicy().getName().equals(policy)) {
+        if (!policyInfo.isEnabled()) {
+          throw new DoNotRetryIOException("Cannot set Erasure Coding policy: " + policy
+            + ". The policy must be enabled, but has state " + policyInfo.getState());
+        }
+        return;
+      }
+    }
+    throw new DoNotRetryIOException(
+      "Cannot set Erasure Coding policy: " + policy + ". Policy not found. Available policies are: "
+        + policies.stream().map(p -> p.getPolicy().getName()).collect(Collectors.joining(", ")));
+  }
+
+  /**
+   * Check if EC policy is different between two descriptors
+   * @return true if a sync is necessary
+   */
+  public static boolean needsSync(TableDescriptor oldDescriptor, TableDescriptor newDescriptor) {
+    String newPolicy = oldDescriptor.getErasureCodingPolicy();
+    String oldPolicy = newDescriptor.getErasureCodingPolicy();
+    return !Objects.equals(oldPolicy, newPolicy);
+  }
+
+  /**
+   * Sync the EC policy state from the newDescriptor onto the FS for the table dir of the provided
+   * table descriptor. If the policy is null, we will remove erasure coding from the FS for the
+   * table dir. If it's non-null, we'll set it to that policy.
+   * @param newDescriptor descriptor containing the policy and table name
+   */
+  public static void sync(FileSystem fs, Path rootDir, TableDescriptor newDescriptor)
+    throws IOException {
+    String newPolicy = newDescriptor.getErasureCodingPolicy();
+    if (newPolicy == null) {
+      unsetPolicy(fs, rootDir, newDescriptor.getTableName());
+    } else {
+      setPolicy(fs, rootDir, newDescriptor.getTableName(), newPolicy);
+    }
+  }
+
+  /**
+   * Sets the EC policy on the table directory for the specified table
+   */
+  public static void setPolicy(FileSystem fs, Path rootDir, TableName tableName, String policy)
+    throws IOException {
+    Path path = CommonFSUtils.getTableDir(rootDir, tableName);
+    setPolicy(fs, path, policy);
+  }
+
+  /**
+   * Sets the EC policy on the path
+   */
+  public static void setPolicy(FileSystem fs, Path path, String policy) throws IOException {
+    getDfs(fs).setErasureCodingPolicy(path, policy);
+  }
+
+  /**
+   * Unsets any EC policy specified on the path.
+   */
+  public static void unsetPolicy(FileSystem fs, Path rootDir, TableName tableName)
+    throws IOException {
+    DistributedFileSystem dfs = getDfs(fs);
+    Path path = CommonFSUtils.getTableDir(rootDir, tableName);
+    if (dfs.getErasureCodingPolicy(path) == null) {
+      LOG.warn("No EC policy set for path {}, nothing to unset", path);
+      return;
+    }
+    dfs.unsetErasureCodingPolicy(path);
+  }
+
+  private static DistributedFileSystem getDfs(Configuration conf) throws HBaseIOException {
+    try {
+      return getDfs(FileSystem.get(conf));
+    } catch (DoNotRetryIOException e) {
+      throw e;
+    } catch (IOException e) {
+      throw new HBaseIOException("Failed to get FileSystem from conf", e);
+    }
+
+  }
+
+  private static DistributedFileSystem getDfs(FileSystem fs) throws DoNotRetryIOException {
+    if (!(fs instanceof DistributedFileSystem)) {
+      throw new DoNotRetryIOException(
+        "Cannot manage Erasure Coding policy. Erasure Coding is only available on HDFS, but fs is "
+          + fs.getClass().getSimpleName());
+    }
+    return (DistributedFileSystem) fs;
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/CloneSnapshotProcedure.java

@@ -36,6 +36,7 @@
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.hadoop.hbase.errorhandling.ForeignException;
 import org.apache.hadoop.hbase.errorhandling.ForeignExceptionDispatcher;
+import org.apache.hadoop.hbase.fs.ErasureCodingUtils;
 import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
 import org.apache.hadoop.hbase.master.MasterFileSystem;
 import org.apache.hadoop.hbase.master.MetricsSnapshot;
@@ -155,6 +156,16 @@ protected Flow executeFromState(final MasterProcedureEnv env, final CloneSnapsho
           updateTableDescriptorWithSFT();
           newRegions = createFilesystemLayout(env, tableDescriptor, newRegions);
           env.getMasterServices().getTableDescriptors().update(tableDescriptor, true);
+          if (tableDescriptor.getErasureCodingPolicy() != null) {
+            setNextState(CloneSnapshotState.CLONE_SNAPSHOT_SET_ERASURE_CODING_POLICY);
+          } else {
+            setNextState(CloneSnapshotState.CLONE_SNAPSHOT_ADD_TO_META);
+          }
+          break;
+        case CLONE_SNAPSHOT_SET_ERASURE_CODING_POLICY:
+          ErasureCodingUtils.setPolicy(env.getMasterFileSystem().getFileSystem(),
+            env.getMasterFileSystem().getRootDir(), getTableName(),
+            tableDescriptor.getErasureCodingPolicy());
           setNextState(CloneSnapshotState.CLONE_SNAPSHOT_ADD_TO_META);
           break;
         case CLONE_SNAPSHOT_ADD_TO_META:

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/CreateTableProcedure.java

@@ -31,6 +31,7 @@
 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
 import org.apache.hadoop.hbase.client.TableDescriptor;
 import org.apache.hadoop.hbase.client.TableState;
+import org.apache.hadoop.hbase.fs.ErasureCodingUtils;
 import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
 import org.apache.hadoop.hbase.master.MasterFileSystem;
 import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer;
@@ -98,6 +99,16 @@ protected Flow executeFromState(final MasterProcedureEnv env, final CreateTableS
           DeleteTableProcedure.deleteFromFs(env, getTableName(), newRegions, true);
           newRegions = createFsLayout(env, tableDescriptor, newRegions);
           env.getMasterServices().getTableDescriptors().update(tableDescriptor, true);
+          if (tableDescriptor.getErasureCodingPolicy() != null) {
+            setNextState(CreateTableState.CREATE_TABLE_SET_ERASURE_CODING_POLICY);
+          } else {
+            setNextState(CreateTableState.CREATE_TABLE_ADD_TO_META);
+          }
+          break;
+        case CREATE_TABLE_SET_ERASURE_CODING_POLICY:
+          ErasureCodingUtils.setPolicy(env.getMasterFileSystem().getFileSystem(),
+            env.getMasterFileSystem().getRootDir(), getTableName(),
+            tableDescriptor.getErasureCodingPolicy());
           setNextState(CreateTableState.CREATE_TABLE_ADD_TO_META);
           break;
         case CREATE_TABLE_ADD_TO_META: