HIVE-18117: Create TestCliDriver for HDFS EC (Andrew Sherman, reviewe…

…d by Sahil Takiar)

itests/qtest/src/test/java/org/apache/hadoop/hive/cli/TestErasureCodingHDFSCliDriver.java

@@ -0,0 +1,65 @@
+/*
+ * 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.hive.cli;
+
+import java.io.File;
+import java.util.List;
+
+import org.apache.hadoop.hive.cli.control.CliAdapter;
+import org.apache.hadoop.hive.cli.control.CliConfigs;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestRule;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+/**
+ * A Test Driver that can be used to run tests over hdfs directories that employ Erasure Coding.
+ */
+@RunWith(Parameterized.class)
+public class TestErasureCodingHDFSCliDriver {
+
+  static CliAdapter adapter = new CliConfigs.ErasureCodingHDFSCliConfig().getCliAdapter();
+
+  @Parameters(name = "{0}")
+  public static List<Object[]> getParameters() throws Exception {
+    return adapter.getParameters();
+  }
+
+  // fields annotated with ClassRule must be public
+  @ClassRule
+  public static TestRule cliClassRule = adapter.buildClassRule();
+
+  @Rule
+  public TestRule cliTestRule = adapter.buildTestRule();
+
+  private String name;
+  private File qfile;
+
+  public TestErasureCodingHDFSCliDriver(String name, File qfile) {
+    this.name = name;
+    this.qfile = qfile;
+  }
+
+  @Test
+  public void testCliDriver() throws Exception {
+    adapter.runTest(name, qfile);
+  }
+}

itests/src/test/resources/testconfiguration.properties

@@ -1693,3 +1693,9 @@ druid.query.files=druidmini_test1.q,\
   druidmini_floorTime.q
 
 druid.kafka.query.files=druidkafkamini_basic.q
+
+# tests to be run by TestErasureCodingHDFSCliDriver and TestCliDriver
+erasurecoding.shared.query.files=erasure_commands.q
+
+# tests to be run only by TestErasureCodingHDFSCliDriver
+erasurecoding.only.query.files=erasure_simple.q

itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CliConfigs.java

@@ -57,6 +57,7 @@ public CliConfig() {
         excludesFrom(testConfigProps, "localSpark.only.query.files");
         excludesFrom(testConfigProps, "druid.query.files");
         excludesFrom(testConfigProps, "druid.kafka.query.files");
+        excludesFrom(testConfigProps, "erasurecoding.only.query.files");
 
         excludeQuery("fouter_join_ppr.q"); // Disabled in HIVE-19509
 
@@ -305,6 +306,7 @@ public TezPerfCliConfig() {
         excludesFrom(testConfigProps, "minimr.query.files");
         excludesFrom(testConfigProps, "minitez.query.files");
         excludesFrom(testConfigProps, "encrypted.query.files");
+        excludesFrom(testConfigProps, "erasurecoding.only.query.files");
 
         setResultsDir("ql/src/test/results/clientpositive/perf/tez");
         setLogDir("itests/qtest/target/qfile-results/clientpositive/tez");
@@ -660,4 +662,52 @@ public BlobstoreNegativeCliConfig() {
       }
     }
   }
+
+  /**
+   * Configuration for TestErasureCodingHDFSCliDriver.
+   */
+  public static class ErasureCodingHDFSCliConfig extends AbstractCliConfig {
+    public ErasureCodingHDFSCliConfig() {
+      super(CoreCliDriver.class);
+      try {
+        setQueryDir("ql/src/test/queries/clientpositive");
+
+        includesFrom(testConfigProps, "erasurecoding.shared.query.files");
+        includesFrom(testConfigProps, "erasurecoding.only.query.files");
+
+        setResultsDir("ql/src/test/results/clientpositive/erasurecoding");
+        setLogDir("itests/qtest/target/qfile-results/clientpositive");
+
+        setInitScript("q_test_init_src.sql");
+        setCleanupScript("q_test_cleanup_src.sql");
+
+        setClusterType(MiniClusterType.mr);
+        setFsType(QTestUtil.FsType.erasure_coded_hdfs);
+        setHiveConfDir(getClusterType());
+      } catch (Exception e) {
+        throw new RuntimeException("can't construct cliconfig", e);
+      }
+    }
+
+    /**
+     * Set the appropriate conf dir based on the cluster type.
+     */
+    private void setHiveConfDir(MiniClusterType clusterType) {
+      switch (clusterType) {
+      case tez:
+        setHiveConfDir("data/conf/tez");
+        break;
+      case spark:
+        setHiveConfDir("data/conf/spark/standalone");
+        break;
+      case miniSparkOnYarn:
+        setHiveConfDir("data/conf/spark/yarn-cluster");
+        break;
+      default:
+        setHiveConfDir("data/conf");
+        break;
+      }
+    }
+  }
+
 }

itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java

@@ -109,6 +109,7 @@
 import org.apache.hadoop.hive.ql.dataset.DatasetParser;
 import org.apache.hadoop.hive.ql.dataset.Dataset;
 import org.apache.hadoop.hive.shims.HadoopShims;
+import org.apache.hadoop.hive.shims.HadoopShims.HdfsErasureCodingShim;
 import org.apache.hadoop.hive.shims.ShimLoader;
 import org.apache.hive.common.util.StreamPrinter;
 import org.apache.hive.druid.MiniDruidCluster;
@@ -143,14 +144,19 @@ public class QTestUtil {
   static final Logger LOG = LoggerFactory.getLogger("QTestUtil");
   private final static String defaultInitScript = "q_test_init.sql";
   private final static String defaultCleanupScript = "q_test_cleanup.sql";
-  private final String[] testOnlyCommands = new String[]{"crypto"};
+  private final String[] testOnlyCommands = new String[]{"crypto", "erasure"};
 
   public static final String TEST_TMP_DIR_PROPERTY = "test.tmp.dir"; // typically target/tmp
   private static final String BUILD_DIR_PROPERTY = "build.dir"; // typically target
 
   public static final String TEST_SRC_TABLES_PROPERTY = "test.src.tables";
   public static final String TEST_HIVE_USER_PROPERTY = "test.hive.user";
 
+  /**
+   * The Erasure Coding Policy to use in TestErasureCodingHDFSCliDriver.
+   */
+  private static final String DEFAULT_TEST_EC_POLICY = "RS-3-2-1024k";
+
   private String testWarehouse;
   private final String testFiles;
   private final File datasetDir;
@@ -479,6 +485,7 @@ public enum FsType {
     local,
     hdfs,
     encrypted_hdfs,
+    erasure_coded_hdfs,
   }
 
   public enum MiniClusterType {
@@ -650,25 +657,47 @@ private void setupFileSystem(HadoopShims shims) throws IOException {
 
     if (fsType == FsType.local) {
       fs = FileSystem.getLocal(conf);
-    } else if (fsType == FsType.hdfs || fsType == FsType.encrypted_hdfs) {
+    } else if (fsType == FsType.hdfs || fsType == FsType.encrypted_hdfs|| fsType == FsType.erasure_coded_hdfs) {
       int numDataNodes = 4;
 
-      if (fsType == FsType.encrypted_hdfs) {
+      // Setup before getting dfs
+      switch (fsType) {
+      case encrypted_hdfs:
         // Set the security key provider so that the MiniDFS cluster is initialized
         // with encryption
         conf.set(SECURITY_KEY_PROVIDER_URI_NAME, getKeyProviderURI());
         conf.setInt("fs.trash.interval", 50);
+        break;
+      case erasure_coded_hdfs:
+        // We need more NameNodes for EC.
+        // To fully exercise hdfs code paths we need 5 NameNodes for the RS-3-2-1024k policy.
+        // With 6 NameNodes we can also run the RS-6-3-1024k policy.
+        numDataNodes = 6;
+        break;
+      default:
+        break;
+      }
 
-        dfs = shims.getMiniDfs(conf, numDataNodes, true, null);
-        fs = dfs.getFileSystem();
+      dfs = shims.getMiniDfs(conf, numDataNodes, true, null);
+      fs = dfs.getFileSystem();
 
+      // Setup after getting dfs
+      switch (fsType) {
+      case encrypted_hdfs:
         // set up the java key provider for encrypted hdfs cluster
         hes = shims.createHdfsEncryptionShim(fs, conf);
-
         LOG.info("key provider is initialized");
-      } else {
-        dfs = shims.getMiniDfs(conf, numDataNodes, true, null);
-        fs = dfs.getFileSystem();
+        break;
+      case erasure_coded_hdfs:
+        // The Erasure policy can't be set in a q_test_init script as QTestUtil runs that code in
+        // a mode that disallows test-only CommandProcessors.
+        // Set the default policy on the root of the file system here.
+        HdfsErasureCodingShim erasureCodingShim = shims.createHdfsErasureCodingShim(fs, conf);
+        erasureCodingShim.enableErasureCodingPolicy(DEFAULT_TEST_EC_POLICY);
+        erasureCodingShim.setErasureCodingPolicy(new Path("hdfs:///"), DEFAULT_TEST_EC_POLICY);
+        break;
+      default:
+        break;
       }
     } else {
       throw new IllegalArgumentException("Unknown or unhandled fsType [" + fsType + "]");
@@ -1030,7 +1059,7 @@ public void clearTablesCreatedDuringTests() throws Exception {
             LOG.warn("Trying to drop table " + e.getTableName() + ". But it does not exist.");
             continue;
           }
-          db.dropTable(dbName, tblName, true, true, fsType == FsType.encrypted_hdfs);
+          db.dropTable(dbName, tblName, true, true, fsNeedsPurge(fsType));
         }
       }
       if (!DEFAULT_DATABASE_NAME.equals(dbName)) {
@@ -2297,4 +2326,15 @@ public QOutProcessor getQOutProcessor() {
   public static void initEventNotificationPoll() throws Exception {
     NotificationEventPoll.initialize(SessionState.get().getConf());
   }
+
+  /**
+   * Should deleted test tables have their data purged.
+   * @return true if data should be purged
+   */
+  private static boolean fsNeedsPurge(FsType type) {
+    if (type == FsType.encrypted_hdfs || type == FsType.erasure_coded_hdfs) {
+      return true;
+    }
+    return false;
+  }
 }

ql/src/java/org/apache/hadoop/hive/ql/processors/CommandProcessorFactory.java

@@ -20,6 +20,7 @@
 
 import static org.apache.commons.lang.StringUtils.isBlank;
 
+import java.io.IOException;
 import java.sql.SQLException;
 import java.util.HashSet;
 import java.util.Set;
@@ -72,35 +73,41 @@ public static CommandProcessor getForHiveCommandInternal(String[] cmd, HiveConf
       return null;
     }
     switch (hiveCommand) {
-      case SET:
-        return new SetProcessor();
-      case RESET:
-        return new ResetProcessor();
-      case DFS:
-        SessionState ss = SessionState.get();
-        return new DfsProcessor(ss.getConf());
-      case ADD:
-        return new AddResourceProcessor();
-      case LIST:
-        return new ListResourceProcessor();
-      case LLAP_CLUSTER:
-        return new LlapClusterResourceProcessor();
-      case LLAP_CACHE:
-        return new LlapCacheResourceProcessor();
-      case DELETE:
-        return new DeleteResourceProcessor();
-      case COMPILE:
-        return new CompileProcessor();
-      case RELOAD:
-        return new ReloadProcessor();
-      case CRYPTO:
-        try {
-          return new CryptoProcessor(SessionState.get().getHdfsEncryptionShim(), conf);
-        } catch (HiveException e) {
-          throw new SQLException("Fail to start the command processor due to the exception: ", e);
-        }
-      default:
-        throw new AssertionError("Unknown HiveCommand " + hiveCommand);
+    case SET:
+      return new SetProcessor();
+    case RESET:
+      return new ResetProcessor();
+    case DFS:
+      SessionState ss = SessionState.get();
+      return new DfsProcessor(ss.getConf());
+    case ADD:
+      return new AddResourceProcessor();
+    case LIST:
+      return new ListResourceProcessor();
+    case LLAP_CLUSTER:
+      return new LlapClusterResourceProcessor();
+    case LLAP_CACHE:
+      return new LlapCacheResourceProcessor();
+    case DELETE:
+      return new DeleteResourceProcessor();
+    case COMPILE:
+      return new CompileProcessor();
+    case RELOAD:
+      return new ReloadProcessor();
+    case CRYPTO:
+      try {
+        return new CryptoProcessor(SessionState.get().getHdfsEncryptionShim(), conf);
+      } catch (HiveException e) {
+        throw new SQLException("Fail to start the command processor due to the exception: ", e);
+      }
+    case ERASURE:
+      try {
+        return new ErasureProcessor(conf);
+      } catch (IOException e) {
+        throw new SQLException("Fail to start the erasure command processor due to the exception: ", e);
+      }
+    default:
+      throw new AssertionError("Unknown HiveCommand " + hiveCommand);
     }
   }