LocalExecutor works outside of MiniCluster.

Author: Harish Butani

windowing/src/main/java/com/sap/hadoop/windowing/Constants.java

@@ -47,6 +47,8 @@ public class Constants
 	public static final String WINDOWING_JOB_QUERY_FILE = "windowing.job.query.file";
 
 	public static final String WINDOWING_TEST_MODE = "windowing.test.mode";
+	public static final String WINDOWING_TEST_LOCAL_MODE = "windowing.test.local.mode";
+	public static final String WINDOWING_TEST_DATA_DIR = "windowing.test.data.dir";
 
 	public static final String WINDOWING_OUTPUT_QUERY_RESULT = "windowing.output.query.result";
 

windowing/src/main/java/com/sap/hadoop/windowing/io/IOUtils.java

@@ -8,6 +8,7 @@
 
 import java.io.IOException;
 import java.io.PrintStream;
+import java.net.URI;
 import java.util.Properties;
 
 import org.apache.commons.logging.Log;
@@ -34,6 +35,7 @@
 import org.apache.hadoop.mapred.TextInputFormat;
 
 import com.sap.hadoop.HiveUtils;
+import com.sap.hadoop.windowing.Constants;
 import com.sap.hadoop.windowing.WindowingException;
 import com.sap.hadoop.windowing.runtime2.Partition;
 
@@ -199,6 +201,19 @@ public static Path makeQualified(Path p, Configuration conf) throws IOException
 	{
 		FileSystem fs = FileSystem.get(conf);
 		p = new Path(p.toUri().getPath()).makeQualified(fs);
+		return convertToTest(p, conf);
+	}
+	
+	public static Path convertToTest(Path p, Configuration conf)
+	{
+		if (conf.getBoolean(Constants.WINDOWING_TEST_LOCAL_MODE, false))
+		{
+			String testDataDir = conf.get(Constants.WINDOWING_TEST_DATA_DIR);
+			URI uri = p.toUri();
+			String location = uri.getPath();
+			location = location.replace("/user/hive/warehouse", testDataDir);
+			return new Path(location);
+		}
 		return p;
 	}
 

windowing/src/main/java/com/sap/hadoop/windowing/runtime2/LocalExecutor.java

@@ -5,11 +5,9 @@
 import org.apache.hadoop.io.Writable;
 
 import com.sap.hadoop.windowing.WindowingException;
-import com.sap.hadoop.windowing.functions2.TableFunctionEvaluator;
 import com.sap.hadoop.windowing.io.IOUtils;
 import com.sap.hadoop.windowing.io.WindowingInput;
 import com.sap.hadoop.windowing.query2.definition.QueryDef;
-import com.sap.hadoop.windowing.query2.definition.TableFuncDef;
 import com.sap.hadoop.windowing.query2.specification.HiveTableSpec;
 import com.sap.hadoop.windowing.query2.translate.QueryTranslationInfo;
 
@@ -31,10 +29,6 @@ public void execute(QueryDef qDef, WindowingShell wShell) throws WindowingExcept
 	{
 		QueryTranslationInfo tInfo = qDef.getTranslationInfo();
 		HiveTableSpec hvTblSpec = qDef.getInput().getHiveTableSpec();
-		TableFuncDef tabDef = (TableFuncDef) qDef.getInput();
-		TableFunctionEvaluator tEval = tabDef.getFunction();
-		String partClassName = tEval.getPartitionClass();
-		int partMemSize = tEval.getPartitionMemSize();
 
 		WindowingInput wIn = IOUtils.createTableWindowingInput(hvTblSpec.getDbName(), hvTblSpec.getTableName(), tInfo.getHiveCfg());
 		//Partition p = IOUtils.createPartition(partClassName, partMemSize, wIn);

windowing/src/test/java/com/sap/hadoop/windowing/runtime2/LocalExecutorTest.java

@@ -0,0 +1,33 @@
+package com.sap.hadoop.windowing.runtime2;
+
+//import junit.framework.Assert;
+
+import org.junit.Test;
+
+import com.sap.hadoop.windowing.WindowingException;
+import com.sap.hadoop.windowing.testutils.LocalExecutorTest;
+
+public class LocalExecutorTests extends LocalExecutorTest
+{
+	@Test
+	public void test1() throws WindowingException
+	{
+		wshell.execute("select  p_mfgr,p_name, p_size,\n" +
+				"rank() as r,\n" +
+				"denserank() as dr\n" +
+				"from part_tiny\n" +
+				"partition by p_mfgr\n" +
+				"order by p_mfgr\n" +
+				"window w1 as rows between 2 preceding and 2 following\n" +
+				"into path='/tmp/wout2'\n" +
+				"serde 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'\n" +
+				"with serdeproperties('field.delim'=',')\n" +
+				"format 'org.apache.hadoop.mapred.TextOutputFormat'", outPrinter);
+		
+		String r = outStream.toString();
+		r = r.replace("\r\n", "\n");
+		System.out.println(r);
+		//String e = WindowingTestsFactory.BASIC.getResult().replaceAll("\t", ",");
+		//Assert.assertEquals(r, e);
+	}
+}

windowing/src/test/java/com/sap/hadoop/windowing/runtime2/LocalExecutorTests.java

@@ -0,0 +1,63 @@
+package com.sap.hadoop.windowing.testutils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.junit.Before;
+import org.junit.BeforeClass;
+
+import com.sap.hadoop.windowing.Constants;
+import com.sap.hadoop.windowing.query2.translate.Translator;
+import com.sap.hadoop.windowing.runtime2.LocalExecutor;
+import com.sap.hadoop.windowing.runtime2.QueryOutputPrinter;
+import com.sap.hadoop.windowing.runtime2.WindowingShell;
+
+public class LocalExecutorTest
+{
+	protected static WindowingShell wshell;
+	protected static String basedir = "src/test/java/data";
+	protected static ByteArrayOutputStream outStream;
+	protected static QueryOutputPrinter outPrinter;
+	
+	@BeforeClass
+	public static void setupClass()
+	{
+		outStream = new ByteArrayOutputStream();
+		
+		Configuration conf = HOME();
+		conf.setBoolean(Constants.WINDOWING_TEST_LOCAL_MODE, true);
+		conf.set(Constants.WINDOWING_TEST_DATA_DIR, basedir);
+		HiveConf hCfg = new HiveConf(conf, conf.getClass());
+		
+		wshell = new WindowingShell(hCfg, new Translator(), new LocalExecutor(new PrintStream(outStream)));
+		//wshell.hiveQryExec = new ThriftBasedHiveQueryExecutor(conf)
+		outPrinter = null;
+	}
+	
+	@Before
+	public void setup()
+	{
+		outStream.reset();
+	}
+	
+	public static Configuration HOME()
+	{
+		Configuration conf = new Configuration();
+		
+		//conf.set("fs.default.name", "hdfs://localhost:8020");
+		//conf.set("mapred.job.tracker", "localhost:8021");
+		
+		//conf.set("hive.metastore.uris", "thrift://localhost:9083");
+		conf.set("javax.jdo.option.ConnectionURL", 
+			"jdbc:mysql://localhost/hivems?createDatabaseIfNotExist=true");
+		conf.set("javax.jdo.option.ConnectionDriverName", "com.mysql.jdbc.Driver");
+		conf.set("javax.jdo.option.ConnectionUserName", "hiveuser");
+		conf.set("javax.jdo.option.ConnectionPassword", "hive");
+		conf.set("hive.metastore.local", "true");
+		
+		
+		return conf;
+	}
+}