Merge remote-tracking branch 'origin/master'

.gitignore

@@ -34,3 +34,4 @@ maven-eclipse.xml
 #log files
 logs
 *.log
+test-output

addons/hive-bridge/pom.xml

@@ -103,6 +103,20 @@
         </dependency>
 
         <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client</artifactId>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client</artifactId>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
             <groupId>org.testng</groupId>
             <artifactId>testng</artifactId>
         </dependency>

addons/hive-bridge/src/main/java/org/apache/hadoop/metadata/hive/bridge/HiveMetaStoreBridge.java

@@ -62,7 +62,7 @@ public class HiveMetaStoreBridge {
         }
     }
 
-    public static final String DGI_URL_PROPERTY = "hive.dgi.url";
+    public static final String DGI_URL_PROPERTY = "hive.hook.dgi.url";
 
     private static final Logger LOG = LoggerFactory.getLogger(HiveMetaStoreBridge.class);
 

addons/hive-bridge/src/main/java/org/apache/hadoop/metadata/hive/hook/HiveHook.java

@@ -78,6 +78,8 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -91,6 +93,14 @@ public class HiveHook implements ExecuteWithHookContext, HiveSemanticAnalyzerHoo
     private static final int WAIT_TIME = 3;
     private static ExecutorService executor;
 
+    private static final String MIN_THREADS = "hive.hook.dgi.minThreads";
+    private static final String MAX_THREADS = "hive.hook.dgi.maxThreads";
+    private static final String KEEP_ALIVE_TIME = "hive.hook.dgi.keepAliveTime";
+
+    private static final int minThreadsDefault = 5;
+    private static final int maxThreadsDefault = 5;
+    private static final long keepAliveTimeDefault = 10;
+
     static {
         // anything shared should be initialized here and destroyed in the
         // shutdown hook The hook contract is weird in that it creates a
@@ -99,7 +109,14 @@ public class HiveHook implements ExecuteWithHookContext, HiveSemanticAnalyzerHoo
         // initialize the async facility to process hook calls. We don't
         // want to do this inline since it adds plenty of overhead for the
         // query.
-        executor = Executors.newSingleThreadExecutor(
+        HiveConf hiveConf = new HiveConf();
+        int minThreads = hiveConf.getInt(MIN_THREADS, minThreadsDefault);
+        int maxThreads = hiveConf.getInt(MAX_THREADS, maxThreadsDefault);
+        long keepAliveTime = hiveConf.getLong(KEEP_ALIVE_TIME, keepAliveTimeDefault);
+
+        executor = new ThreadPoolExecutor(minThreads, maxThreads,
+                keepAliveTime, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<Runnable>(),
                 new ThreadFactoryBuilder()
                         .setDaemon(true)
                         .setNameFormat("DGI Logger %d")
@@ -199,8 +216,16 @@ public class HiveHook implements ExecuteWithHookContext, HiveSemanticAnalyzerHoo
     }
 
     private void registerCTAS(HiveMetaStoreBridge dgiBridge, HookContext hookContext, HiveConf conf) throws Exception {
+        LOG.debug("Registering CTAS");
+
         Set<ReadEntity> inputs = hookContext.getInputs();
         Set<WriteEntity> outputs = hookContext.getOutputs();
+
+        //Even explain CTAS has operation name as CREATETABLE_AS_SELECT
+        if (inputs.isEmpty() && outputs.isEmpty()) {
+            LOG.info("Explain statement. Skipping...");
+        }
+
         String user = hookContext.getUserName();
         HiveOperation operation = HiveOperation.valueOf(hookContext.getOperationName());
         String queryId = null;
@@ -214,7 +239,6 @@ public class HiveHook implements ExecuteWithHookContext, HiveSemanticAnalyzerHoo
             queryStartTime = plan.getQueryStartTime();
         }
 
-
         Referenceable processReferenceable = new Referenceable(HiveDataTypes.HIVE_PROCESS.getName());
         processReferenceable.set("processName", operation.getOperationName());
         processReferenceable.set("startTime", queryStartTime);
@@ -311,7 +335,7 @@ public class HiveHook implements ExecuteWithHookContext, HiveSemanticAnalyzerHoo
         ExplainTask explain = new ExplainTask();
         explain.initialize(conf, queryPlan, null);
 
-        org.json.JSONObject explainPlan = explain.getJSONLogicalPlan(null, ew);
+        org.json.JSONObject explainPlan = explain.getJSONPlan(null, ew);
         return explainPlan.toString();
     }
 

addons/hive-bridge/src/site/twiki/Bridge-Hive.twiki

@@ -20,12 +20,13 @@ Hive metadata can be modelled in DGI using its Type System. The default modellin
 
 
 ---++ Importing Hive Metadata
-org.apache.hadoop.metadata.hive.bridge.HiveMetaStoreBridge imports the hive metadata into DGI using the typesystem defined in org.apache.hadoop.metadata.hive.model.HiveDataModelGenerator. import-hive.sh command can be used to facilitate this. Set-up the following configs in <dgi package>/conf/hive-site.xml:
+org.apache.hadoop.metadata.hive.bridge.HiveMetaStoreBridge imports the hive metadata into DGI using the typesystem defined in org.apache.hadoop.metadata.hive.model.HiveDataModelGenerator. import-hive.sh command can be used to facilitate this.
+Set-up the following configs in <dgi package>/conf/hive-site.xml:
    * Hive metastore configuration - Refer [[https://cwiki.apache.org/confluence/display/Hive/AdminManual+MetastoreAdmin][Hive Metastore Configuration documentation]]
    * DGI endpoint - Add the following property with the DGI endpoint for your set-up
 <verbatim>
 <property>
-  <name>hive.dgi.url</name>
+  <name>hive.hook.dgi.url</name>
   <value>http://localhost:21000/</value>
 </property>
 </verbatim>
@@ -34,13 +35,26 @@ Usage: <dgi package>/bin/import-hive.sh
 
 
 ---++ Hive Hook
-Hive supports listeners on hive command execution using hive hooks. This can be used to add/update/remove entities in DGI. Follow the following instructions in your hive set-up
-   * Add org.apache.hadoop.metadata.hive.hook.HiveHook as post execution hook in hive-ste.xml
+Hive supports listeners on hive command execution using hive hooks. This is used to add/update/remove entities in DGI using the model defined in org.apache.hadoop.metadata.hive.model.HiveDataModelGenerator.
+The hook submits the request to a thread pool executor to avoid blocking the command execution. Follow the these instructions in your hive set-up to add hive hook for DGI:
+   * Add org.apache.hadoop.metadata.hive.hook.HiveHook as post execution hook in hive-site.xml
 <verbatim>
 <property>
   <name>hive.exec.post.hooks</name>
   <value>org.apache.hadoop.metadata.hive.hook.HiveHook</value>
 </property>
 </verbatim>
+   * Add the following property in hive-ste.xml with the DGI endpoint for your set-up
+<verbatim>
+<property>
+  <name>hive.hook.dgi.url</name>
+  <value>http://localhost:21000/</value>
+</property>
+</verbatim>
    * Add 'export HIVE_AUX_JARS_PATH=<dgi package>/hook/hive' in hive-env.sh
-   * Restart hive-server2
+
+The following properties in hive-site.xml control the thread pool details:
+   * hive.hook.dgi.minThreads - core number of threads. default 5
+   * hive.hook.dgi.maxThreads - maximum number of threads. default 5
+   * hive.hook.dgi.keepAliveTime - keep alive time in msecs. default 10
+

addons/hive-bridge/src/test/java/org/apache/hadoop/metadata/hive/hook/HiveHookIT.java

@@ -69,6 +69,7 @@ public class HiveHookIT {
         hiveConf.setBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY, false);
         hiveConf.setVar(HiveConf.ConfVars.METASTOREWAREHOUSE, System.getProperty("user.dir") + "/target/metastore");
         hiveConf.set(HiveMetaStoreBridge.DGI_URL_PROPERTY, DGI_URL);
+        hiveConf.set("javax.jdo.option.ConnectionURL", "jdbc:derby:./target/metastore_db;create=true");
         hiveConf.set("debug", "true");
         return hiveConf;
     }