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,13 @@ 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));
+
   /**
    * Used by shell/rest interface to access this metadata attribute which denotes if the table
    * should be treated by region normalizer.
@@ -490,6 +497,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 +760,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
+     * @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-protocol-shaded/src/main/protobuf/server/master/MasterProcedure.proto

@@ -74,6 +74,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 {

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

@@ -0,0 +1,96 @@
+/*
+ * 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.stream.Collectors;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseIOException;
+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 and enabled.
+   */
+  public static void checkAvailable(FileSystem fs, String policy) throws HBaseIOException {
+    DistributedFileSystem dfs = getDfs(fs);
+
+    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 HBaseIOException("Cannot set Erasure Coding policy: " + policy
+            + ". The policy must be enabled, but has state " + policyInfo.getState());
+        }
+        return;
+      }
+    }
+
+    throw new HBaseIOException(
+      "Cannot set Erasure Coding policy: " + policy + ". Policy not found. Available policies are: "
+        + policies.stream().map(p -> p.getPolicy().getName()).collect(Collectors.joining(", ")));
+  }
+
+  /**
+   * 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 path) throws IOException {
+    DistributedFileSystem dfs = getDfs(fs);
+    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(FileSystem fs) throws HBaseIOException {
+    if (!(fs instanceof DistributedFileSystem)) {
+      throw new HBaseIOException(
+        "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/ModifyTableProcedure.java

@@ -27,11 +27,14 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Objects;
 import java.util.Set;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.ConcurrentTableModificationException;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
 import org.apache.hadoop.hbase.HBaseIOException;
@@ -42,12 +45,14 @@
 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
 import org.apache.hadoop.hbase.client.TableDescriptor;
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.fs.ErasureCodingUtils;
 import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
 import org.apache.hadoop.hbase.master.zksyncer.MetaLocationSyncer;
 import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer;
 import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerValidationUtils;
 import org.apache.hadoop.hbase.rsgroup.RSGroupInfo;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -114,6 +119,12 @@ protected void preflightChecks(MasterProcedureEnv env, Boolean enabled) throws H
         }
       }
     }
+
+    String policy = modifiedTableDescriptor.getErasureCodingPolicy();
+    if (policy != null) {
+      ErasureCodingUtils.checkAvailable(env.getMasterFileSystem().getFileSystem(), policy);
+    }
+
     if (!reopenRegions) {
       if (this.unmodifiedTableDescriptor == null) {
         throw new HBaseIOException(
@@ -220,7 +231,7 @@ protected Flow executeFromState(final MasterProcedureEnv env, final ModifyTableS
           if (reopenRegions) {
             setNextState(ModifyTableState.MODIFY_TABLE_REOPEN_ALL_REGIONS);
           } else {
-            return Flow.NO_MORE_STATE;
+            setNextState(ModifyTableState.MODIFY_TABLE_SYNC_ERASURE_CODING_POLICY);
           }
           break;
         case MODIFY_TABLE_REOPEN_ALL_REGIONS:
@@ -246,11 +257,15 @@ protected Flow executeFromState(final MasterProcedureEnv env, final ModifyTableS
           if (deleteColumnFamilyInModify) {
             setNextState(ModifyTableState.MODIFY_TABLE_DELETE_FS_LAYOUT);
           } else {
-            return Flow.NO_MORE_STATE;
+            setNextState(ModifyTableState.MODIFY_TABLE_SYNC_ERASURE_CODING_POLICY);
           }
           break;
         case MODIFY_TABLE_DELETE_FS_LAYOUT:
           deleteFromFs(env, unmodifiedTableDescriptor, modifiedTableDescriptor);
+          setNextState(ModifyTableState.MODIFY_TABLE_SYNC_ERASURE_CODING_POLICY);
+          break;
+        case MODIFY_TABLE_SYNC_ERASURE_CODING_POLICY:
+          syncErasureCodingPolicy(env);
           return Flow.NO_MORE_STATE;
         default:
           throw new UnsupportedOperationException("unhandled state=" + state);
@@ -512,6 +527,23 @@ private void closeExcessReplicasIfNeeded(MasterProcedureEnv env) {
       .createUnassignProceduresForClosingExcessRegionReplicas(getTableName(), newReplicaCount));
   }
 
+  private void syncErasureCodingPolicy(MasterProcedureEnv env) throws IOException {
+    String oldPolicy = unmodifiedTableDescriptor.getErasureCodingPolicy();
+    String newPolicy = modifiedTableDescriptor.getErasureCodingPolicy();
+    if (Objects.equals(oldPolicy, newPolicy)) {
+      return;
+    }
+
+    FileSystem fileSystem = env.getMasterFileSystem().getFileSystem();
+    Path tableDir = CommonFSUtils.getTableDir(env.getMasterFileSystem().getRootDir(),
+      unmodifiedTableDescriptor.getTableName());
+    if (newPolicy == null) {
+      ErasureCodingUtils.unsetPolicy(fileSystem, tableDir);
+    } else {
+      ErasureCodingUtils.setPolicy(fileSystem, tableDir, newPolicy);
+    }
+  }
+
   /**
    * Action after modifying table.
    * @param env   MasterProcedureEnv

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactedHFilesDischarger.java

@@ -70,7 +70,7 @@ public CompactedHFilesDischarger(final int period, final Stoppable stopper,
    * no-executor before you call run.
    * @return The old setting for <code>useExecutor</code>
    */
-  boolean setUseExecutor(final boolean useExecutor) {
+  public boolean setUseExecutor(final boolean useExecutor) {
     boolean oldSetting = this.useExecutor;
     this.useExecutor = useExecutor;
     return oldSetting;