From pig-commitskellyelizabeth@example.org Mon Jan 04 19:32:55 2010 Return-Path: <pig-commitskellyelizabeth@example.org> Delivered-To: kellyelizabeth@example.org Received: (qmail 49770 invoked from network); 4 Jan 2010 19:32:55 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 4 Jan 2010 19:32:55 -0000 Received: (qmail 32294 invoked by uid 500); 4 Jan 2010 19:32:55 -0000 Delivered-To: kellyelizabeth@example.org Received: (qmail 32256 invoked by uid 500); 4 Jan 2010 19:32:54 -0000 Mailing-List: contact kellyelizabeth@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:kellyelizabeth@example.org> List-Unsubscribe: <mailto:kellyelizabeth@example.org> List-Post: <mailto:kellyelizabeth@example.org> List-Id: <pig-commits.hadoop.apache.org> Reply-To: kellyelizabeth@example.org Delivered-To: mailing list kellyelizabeth@example.org Received: (qmail 32247 invoked by uid 500); 4 Jan 2010 19:32:54 -0000 Delivered-To: kellyelizabeth@example.org Received: (qmail 32244 invoked by uid 99); 4 Jan 2010 19:32:54 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 04 Jan 2010 19:32:54 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 04 Jan 2010 19:32:51 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 863CF23888D1; Mon, 4 Jan 2010 19:32:29 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r895751 - in /hadoop/pig/branches/branch-0.6/contrib/zebra: ./ src/java/org/apache/hadoop/zebra/pig/ src/test/org/apache/hadoop/zebra/pig/ Date: Mon, 04 Jan 2010 19:32:18 -0000 To: kellyelizabeth@example.org From: kellyelizabeth@example.org X-Mailer: svnmailer-1.0.8 Message-Id: <kellyelizabeth@example.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: yanz Date: Mon Jan 4 19:32:00 2010 New Revision: 895751 URL: http://svn.apache.org/viewvc?rev=895751&view=rev Log: PIG-1167: Hadoop file glob support (yanz) Added: hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestGlobTableLoader.java hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveMultiTableGlob.java Modified: hadoop/pig/branches/branch-0.6/contrib/zebra/CHANGES.txt hadoop/pig/branches/branch-0.6/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveProjectionNegative.java Modified: hadoop/pig/branches/branch-0.6/contrib/zebra/CHANGES.txt URL: http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/contrib/zebra/CHANGES.txt?rev=895751&r1=895750&r2=895751&view=diff ============================================================================== --- hadoop/pig/branches/branch-0.6/contrib/zebra/CHANGES.txt (original) +++ hadoop/pig/branches/branch-0.6/contrib/zebra/CHANGES.txt Mon Jan 4 19:32:00 2010 @@ -39,6 +39,8 @@ BUG FIXES + PIG-1167: Hadoop file glob support (yanz) + PIG-1145: Merge Join on Large Table throws an EOF exception (yanz) PIG-1074 Zebra store function should allow '888a:dbb7:2d49:b3f6:8605:bf4e:caed:e5a4' in column names in output Modified: hadoop/pig/branches/branch-0.6/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java URL: http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java?rev=895751&r1=895750&r2=895751&view=diff ============================================================================== --- hadoop/pig/branches/branch-0.6/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java (original) +++ hadoop/pig/branches/branch-0.6/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java Mon Jan 4 19:32:00 2010 @@ -226,7 +226,7 @@ FileSystem fs = p.getFileSystem(jobConf); FileStatus[] matches = fs.globStatus(p); if (matches == null) { - LOG.warn("Input path does not exist: " + p); + throw new IOException("Input path does not exist: " + p); } else if (matches.length == 0) { LOG.warn("Input Pattern " + p + " matches 0 files"); @@ -293,33 +293,14 @@ Projection projection; - if (!fileName.contains(",")) { // one table; - org.apache.hadoop.zebra.schema.Schema tschema = BasicTable.Reader.getSchema(new Path(fileName), jobConf); - try { - projection = new org.apache.hadoop.zebra.types.Projection(tschema, TableInputFormat.getProjection(jobConf)); - projectionSchema = projection.getProjectionSchema(); - } catch (ParseException e) { - throw new IOException("Schema parsing failed : "+e.getMessage()); - } - } else { // table union; - org.apache.hadoop.zebra.schema.Schema unionSchema = new org.apache.hadoop.zebra.schema.Schema(); - for (Path p : paths) { - org.apache.hadoop.zebra.schema.Schema schema = BasicTable.Reader.getSchema(p, jobConf); - try { - unionSchema.unionSchema(schema); - } catch (ParseException e) { - throw new IOException(e.getMessage()); - } - } - - try { - projection = new org.apache.hadoop.zebra.types.Projection(unionSchema, TableInputFormat.getProjection(jobConf)); - projectionSchema = projection.getProjectionSchema(); - } catch (ParseException e) { - throw new IOException("Schema parsing failed : "+e.getMessage()); - } - } - + org.apache.hadoop.zebra.schema.Schema tschema = TableInputFormat.getSchema(jobConf); + try { + projection = new org.apache.hadoop.zebra.types.Projection(tschema, TableInputFormat.getProjection(jobConf)); + projectionSchema = projection.getProjectionSchema(); + } catch (ParseException e) { + throw new IOException("Schema parsing failed : "+e.getMessage()); + } + if (projectionSchema == null) { throw new IOException("Cannot determine table projection schema"); } Added: hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestGlobTableLoader.java URL: http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestGlobTableLoader.java?rev=895751&view=auto ============================================================================== --- hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestGlobTableLoader.java (added) +++ hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestGlobTableLoader.java Mon Jan 4 19:32:00 2010 @@ -0,0 +1,232 @@ +/** + * 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.zebra.pig; + +import java.io.File; +import java.io.IOException; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.BytesWritable; +import org.apache.hadoop.zebra.io.BasicTable; +import org.apache.hadoop.zebra.io.TableInserter; +import org.apache.hadoop.zebra.io.TableScanner; +import org.apache.hadoop.zebra.io.BasicTable.Reader.RangeSplit; +import org.apache.hadoop.zebra.parser.ParseException; +import org.apache.hadoop.zebra.schema.Schema; +import org.apache.hadoop.zebra.types.TypesUtils; +import org.apache.pig.ExecType; +import org.apache.pig.PigServer; +import org.apache.pig.backend.executionengine.ExecException; +import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil; +import org.apache.pig.data.Tuple; +import org.apache.pig.test.MiniCluster; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +/** + * Note: + * + * Make sure you add the build/pig-0.1.0-dev-core.jar to the Classpath of the + * app/debug configuration, when run this from inside the Eclipse. + * + */ +public class TestGlobTableLoader{ + protected static ExecType execType = ExecType.MAPREDUCE; + private static MiniCluster cluster; + protected static PigServer pigServer; + private static Path pathTable; + private static Configuration conf; + private static String zebraJar; + private static String whichCluster; + private static FileSystem fs; + @BeforeClass + public static void setUp() throws Exception { + if (System.getProperty("hadoop.log.dir") == null) { + String base = new File(".").getPath(); // getAbsolutePath(); + System + .setProperty("hadoop.log.dir", new Path(base).toString() + "./logs"); + } + + // if whichCluster is not defined, or defined something other than + // "realCluster" or "miniCluster", set it to "miniCluster" + if (System.getProperty("whichCluster") == null + || ((!System.getProperty("whichCluster") + .equalsIgnoreCase("realCluster")) && (!System.getProperty( + "whichCluster").equalsIgnoreCase("miniCluster")))) { + System.setProperty("whichCluster", "miniCluster"); + whichCluster = System.getProperty("whichCluster"); + } else { + whichCluster = System.getProperty("whichCluster"); + } + + System.out.println("cluster: " + whichCluster); + if (whichCluster.equalsIgnoreCase("realCluster") + && System.getenv("HADOOP_HOME") == null) { + System.out.println("Please set HADOOP_HOME"); + System.exit(0); + } + + conf = new Configuration(); + + if (whichCluster.equalsIgnoreCase("realCluster") + && System.getenv("USER") == null) { + System.out.println("Please set USER"); + System.exit(0); + } + zebraJar = System.getenv("HADOOP_HOME") + "/../jars/zebra.jar"; + File file = new File(zebraJar); + if (!file.exists() && whichCluster.equalsIgnoreCase("realCulster")) { + System.out.println("Please put zebra.jar at hadoop_home/../jars"); + System.exit(0); + } + + if (whichCluster.equalsIgnoreCase("realCluster")) { + pigServer = new PigServer(ExecType.MAPREDUCE, ConfigurationUtil + .toProperties(conf)); + pigServer.registerJar(zebraJar); + pathTable = new Path("/user/" + System.getenv("USER") + + "/TestMapTableLoader"); + removeDir(pathTable); + fs = pathTable.getFileSystem(conf); + } + + if (whichCluster.equalsIgnoreCase("miniCluster")) { + if (execType == ExecType.MAPREDUCE) { + cluster = MiniCluster.buildCluster(); + pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties()); + fs = cluster.getFileSystem(); + pathTable = new Path(fs.getWorkingDirectory() + + "/TestMapTableLoader1"); + removeDir(pathTable); + System.out.println("path1 =" + pathTable); + } else { + pigServer = new PigServer(ExecType.LOCAL); + } + } + + + BasicTable.Writer writer = new BasicTable.Writer(pathTable, + "m1:map(string)", "[m1#{a}]", conf); + Schema schema = writer.getSchema(); + Tuple tuple = TypesUtils.createTuple(schema); + + final int numsBatch = 10; + final int numsInserters = 2; + TableInserter[] inserters = new TableInserter[numsInserters]; + for (int i = 0; i < numsInserters; i++) { + inserters[i] = writer.getInserter("ins" + i, false); + } + + for (int b = 0; b < numsBatch; b++) { + for (int i = 0; i < numsInserters; i++) { + TypesUtils.resetTuple(tuple); + Map<String, String> map = new HashMap<String, String>(); + map.put("a", "x"); + map.put("b", "y"); + map.put("c", "z"); + tuple.set(0, map); + + try { + inserters[i].insert(new BytesWritable(("key" + i).getBytes()), tuple); + } catch (Exception e) { + System.out.println(e.getMessage()); + } + } + } + for (int i = 0; i < numsInserters; i++) { + inserters[i].close(); + } + } + + @AfterClass + public static void tearDown() throws Exception { + pigServer.shutdown(); + } + public static void removeDir(Path outPath) throws IOException { + String command = null; + if (whichCluster.equalsIgnoreCase("realCluster")) { + command = System.getenv("HADOOP_HOME") +"/bin/hadoop fs -rmr " + outPath.toString(); + } + else{ + command = "rm -rf " + outPath.toString(); + } + Runtime runtime = Runtime.getRuntime(); + Process proc = runtime.exec(command); + int exitVal = -1; + try { + exitVal = proc.waitFor(); + } catch (InterruptedException e) { + System.err.println(e); + } + + } + + // @Test + public void test1() throws IOException, ParseException { + String projection = new String("m1#{b}"); + BasicTable.Reader reader = new BasicTable.Reader(pathTable, conf); + reader.setProjection(projection); + // long totalBytes = reader.getStatus().getSize(); + + List<RangeSplit> splits = reader.rangeSplit(1); + reader.close(); + reader = new BasicTable.Reader(pathTable, conf); + reader.setProjection(projection); + + TableScanner scanner = reader.getScanner(splits.get(0), true); + BytesWritable key = new BytesWritable(); + Tuple value = TypesUtils.createTuple(scanner.getSchema()); + // HashMap<String, Object> mapval; + while (!scanner.atEnd()) { + scanner.getKey(key); + // Assert.assertEquals(key, new BytesWritable("key0".getBytes())); + scanner.getValue(value); + System.out.println("key = " + key + " value = " + value); + + // mapval = (HashMap<String, Object>) value.get(0); + // Assert.assertEquals("x", mapval.get("a")); + // Assert.assertEquals(null, mapval.get("b")); + // Assert.assertEquals(null, mapval.get("c")); + scanner.advance(); + } + reader.close(); + } + + @Test + public void testReader() throws ExecException, IOException { + pathTable = new Path("/user/" + System.getenv("USER") + + "/{TestMapTableLoader1}"); + String query = "records = LOAD '" + pathTable.toString() + + "' USING org.apache.hadoop.zebra.pig.TableLoader('m1#{a}');"; + System.out.println(query); + pigServer.registerQuery(query); + Iterator<Tuple> it = pigServer.openIterator("records"); + while (it.hasNext()) { + Tuple cur = it.next(); + System.out.println(cur); + } + } +} Added: hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveMultiTableGlob.java URL: http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveMultiTableGlob.java?rev=895751&view=auto ============================================================================== --- hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveMultiTableGlob.java (added) +++ hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveMultiTableGlob.java Mon Jan 4 19:32:00 2010 @@ -0,0 +1,424 @@ +/** + * 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.zebra.pig; + +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.PrintWriter; +import java.util.Iterator; +import java.util.ArrayList; +import java.util.StringTokenizer; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.BytesWritable; +import org.apache.hadoop.zebra.io.BasicTable; +import org.apache.hadoop.zebra.io.TableInserter; +import org.apache.hadoop.zebra.pig.TableStorer; +import org.apache.hadoop.zebra.schema.Schema; +import org.apache.hadoop.zebra.types.TypesUtils; +import org.apache.pig.ExecType; +import org.apache.pig.PigServer; +import org.apache.pig.backend.executionengine.ExecException; +import org.apache.pig.backend.executionengine.ExecJob; +import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil; +import org.apache.pig.data.DataByteArray; +import org.apache.pig.data.Tuple; +import org.apache.pig.test.MiniCluster; + +import junit.framework.Assert; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + + +public class TestOrderPreserveMultiTableGlob { + + final static int NUMB_TABLE = 10; // number of tables for stress test + final static int NUMB_TABLE_ROWS = 5; // number of rows for each table + + final static String TABLE_SCHEMA = "int1:int,str1:string,byte1:bytes"; + final static String TABLE_STORAGE = "[int1,str1,byte1]"; + + static int fileId = 0; + static int sortId = 0; + + protected static ExecType execType = ExecType.MAPREDUCE; + private static MiniCluster cluster; + protected static PigServer pigServer; + protected static ExecJob pigJob; + + private static ArrayList<Path> pathTables; + private static int totalTableRows =0; + + private static Configuration conf; + private static FileSystem fs; + + private static String zebraJar; + private static String whichCluster; + + @BeforeClass + public static void setUp() throws Exception { + if (System.getProperty("hadoop.log.dir") == null) { + String base = new File(".").getPath(); // getAbsolutePath(); + System.setProperty("hadoop.log.dir", new Path(base).toString() + "./logs"); + } + + // if whichCluster is not defined, or defined something other than + // "realCluster" or "miniCluster", set it to "realCluster" + if (System.getProperty("whichCluster") == null + || ((!System.getProperty("whichCluster") + .equalsIgnoreCase("realCluster")) && (!System.getProperty( + "whichCluster").equalsIgnoreCase("miniCluster")))) { + System.setProperty("whichCluster", "miniCluster"); + whichCluster = System.getProperty("whichCluster"); + } else { + whichCluster = System.getProperty("whichCluster"); + } + + System.out.println("cluster: " + whichCluster); + if (whichCluster.equalsIgnoreCase("realCluster") + && System.getenv("HADOOP_HOME") == null) { + System.out.println("Please set HADOOP_HOME"); + System.exit(0); + } + + conf = new Configuration(); + + if (whichCluster.equalsIgnoreCase("realCluster") + && System.getenv("USER") == null) { + System.out.println("Please set USER"); + System.exit(0); + } + zebraJar = System.getenv("HADOOP_HOME") + "/../jars/zebra.jar"; + File file = new File(zebraJar); + if (!file.exists() && whichCluster.equalsIgnoreCase("realCulster")) { + System.out.println("Please put zebra.jar at hadoop_home/../jars"); + System.exit(0); + } + + if (whichCluster.equalsIgnoreCase("realCluster")) { + pigServer = new PigServer(ExecType.MAPREDUCE, ConfigurationUtil + .toProperties(conf)); + pigServer.registerJar(zebraJar); + + pathTables = new ArrayList<Path>(); + for (int i=0; i<NUMB_TABLE; ++i) { + Path pathTable = new Path("/user/" + System.getenv("USER") + + "/TestOderPerserveMultiTable" + i); + pathTables.add(pathTable); + removeDir(pathTable); + } + fs = pathTables.get(0).getFileSystem(conf); + } + + if (whichCluster.equalsIgnoreCase("miniCluster")) { + if (execType == ExecType.MAPREDUCE) { + cluster = MiniCluster.buildCluster(); + pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties()); + fs = cluster.getFileSystem(); + + pathTables = new ArrayList<Path>(); + for (int i=0; i<NUMB_TABLE; ++i) { + Path pathTable = new Path(fs.getWorkingDirectory() + + "/TestOderPerserveMultiTable" + i); + pathTables.add(pathTable); + removeDir(pathTable); + } + } else { + pigServer = new PigServer(ExecType.LOCAL); + } + } + + // Create tables + for (int i=0; i<NUMB_TABLE; ++i) { + // Create table data + Object[][] table = new Object[NUMB_TABLE_ROWS][3]; // three columns + + for (int j=0; j<NUMB_TABLE_ROWS; ++j) { + table[j][0] = i; + table[j][1] = new String("string" + j); + table[j][2] = new DataByteArray("byte" + (NUMB_TABLE_ROWS - j)); + ++totalTableRows; + } + // Create table + createTable(pathTables.get(i), TABLE_SCHEMA, TABLE_STORAGE, table); + + // Load Table + String query = "table" + i + " = LOAD '" + pathTables.get(i).toString() + + "' USING org.apache.hadoop.zebra.pig.TableLoader();"; + pigServer.registerQuery(query); + } + } + + private static void createTable(Path path, String schemaString, String storageString, Object[][] tableData) + throws IOException { + // + // Create table from tableData array + // + BasicTable.Writer writer = new BasicTable.Writer(path, schemaString, storageString, conf); + + Schema schema = writer.getSchema(); + Tuple tuple = TypesUtils.createTuple(schema); + TableInserter inserter = writer.getInserter("ins", false); + + for (int i = 0; i < tableData.length; ++i) { + TypesUtils.resetTuple(tuple); + for (int k = 0; k < tableData[i].length; ++k) { + tuple.set(k, tableData[i][k]); + System.out.println("DEBUG: setting tuple k=" + k + "value= " + tableData[i][k]); + } + inserter.insert(new BytesWritable(("key" + i).getBytes()), tuple); + } + inserter.close(); + writer.close(); + } + + @AfterClass + public static void tearDown() throws Exception { + pigServer.shutdown(); + } + + public static void removeDir(Path outPath) throws IOException { + String command = null; + if (whichCluster.equalsIgnoreCase("realCluster")) { + command = System.getenv("HADOOP_HOME") +"/bin/hadoop fs -rmr " + outPath.toString(); + } + else{ + command = "rm -rf " + outPath.toString(); + } + Runtime runtime = Runtime.getRuntime(); + Process proc = runtime.exec(command); + int exitVal = -1; + try { + exitVal = proc.waitFor(); + } catch (InterruptedException e) { + System.err.println(e); + } + } + + private Iterator<Tuple> testOrderPreserveUnion(ArrayList<String> inputTables, String sortkey, String columns) + throws IOException { + // + // Test order preserve union from input tables and provided output columns + // + Assert.assertTrue("Table union requires two or more input tables", inputTables.size() >= 2); + + Path newPath = new Path(getCurrentMethodName()); + ArrayList<String> pathList = new ArrayList<String>(); + + // Load and store each of the input tables + for (int i=0; i<inputTables.size(); ++i) { + String tablename = inputTables.get(i); + String sortName = "sort" + ++sortId; + + // Sort tables + String orderby = sortName + " = ORDER " + tablename + " BY " + sortkey + " ;"; + pigServer.registerQuery(orderby); + + String sortPath = new String(newPath.toString() + ++fileId); // increment fileId suffix + + // Store sorted tables + pigJob = pigServer.store(sortName, sortPath, TableStorer.class.getCanonicalName() + + "('" + TABLE_STORAGE + "')"); + Assert.assertNull(pigJob.getException()); + + pathList.add(sortPath); // add table path to list + } + + String paths = new String(); + + paths += newPath.toString() + "{"; + fileId = 0; + for (String path:pathList) + paths += ++fileId + ","; + paths = paths.substring(0, paths.lastIndexOf(",")); // remove trailing comma + paths += "}"; + + String queryLoad = "records1 = LOAD '" + + paths + + "' USING org.apache.hadoop.zebra.pig.TableLoader('" + columns + "', 'sorted');"; + + System.out.println("queryLoad: " + queryLoad); + pigServer.registerQuery(queryLoad); + + // Return iterator + Iterator<Tuple> it1 = pigServer.openIterator("records1"); + return it1; + } + + @Test + public void test_sorted_union_multi_table() throws ExecException, IOException { + // + // Test sorted union + // + + // Create input tables for order preserve union + ArrayList<String> inputTables = new ArrayList<String>(); // Input tables + for (int i=0; i<NUMB_TABLE; ++i) { + inputTables.add("table" + i); // add input table + } + + // Test with input tables and provided output columns + testOrderPreserveUnion(inputTables, "int1", "int1, str1, byte1"); + + // Create results table for verification + ArrayList<ArrayList<Object>> resultTable = new ArrayList<ArrayList<Object>>(); + for (int i=0; i<NUMB_TABLE; ++i) { + for (int j=0; j<NUMB_TABLE_ROWS; ++j) { + ArrayList<Object> resultRow = new ArrayList<Object>(); + + resultRow.add(i); // int1 + resultRow.add(new String("string" + j)); // str1 + resultRow.add(new DataByteArray("byte" + (NUMB_TABLE_ROWS - j))); // byte1 + + resultTable.add(resultRow); + } + } + + // Verify union table + Iterator<Tuple> it = pigServer.openIterator("records1"); + int numbRows = verifyTable(resultTable, 0, it); + + Assert.assertEquals(totalTableRows, numbRows); + + // Print Table + //printTable("records1"); + } + + /** + * Verify union output table with expected results + * + */ + private int verifyTable(ArrayList<ArrayList<Object>> resultTable, int keyColumn, Iterator<Tuple> it) throws IOException { + int numbRows = 0; + int index = 0; + Object value = resultTable.get(index).get(keyColumn); // get value of primary key + + while (it.hasNext()) { + Tuple rowValues = it.next(); + + // If last primary sort key does match then search for next matching key + if (! compareObj(value, rowValues.get(keyColumn))) { + int subIndex = index + 1; + while (subIndex < resultTable.size()) { + if ( ! compareObj(value, resultTable.get(subIndex).get(keyColumn)) ) { // found new key + index = subIndex; + value = resultTable.get(index).get(keyColumn); + break; + } + ++subIndex; + } + Assert.assertEquals("Table comparison error for row : " + numbRows + " - no key found for : " + + rowValues.get(keyColumn), value, rowValues.get(keyColumn)); + } + // Search for matching row with this primary key + int subIndex = index; + + while (subIndex < resultTable.size()) { + // Compare row + ArrayList<Object> resultRow = resultTable.get(subIndex); + if ( compareRow(rowValues, resultRow) ) + break; // found matching row + ++subIndex; + Assert.assertEquals("Table comparison error for row : " + numbRows + " - no matching row found for : " + + rowValues.get(keyColumn), value, resultTable.get(subIndex).get(keyColumn)); + } + ++numbRows; + } + Assert.assertEquals(resultTable.size(), numbRows); // verify expected row count + return numbRows; + } + + /** + * Compare table rows + * + */ + private boolean compareRow(Tuple rowValues, ArrayList<Object> resultRow) throws IOException { + boolean result = true; + Assert.assertEquals(resultRow.size(), rowValues.size()); + for (int i = 0; i < rowValues.size(); ++i) { + if (! compareObj(rowValues.get(i), resultRow.get(i)) ) { + result = false; + break; + } + } + return result; + } + + /** + * Compare table values + * + */ + private boolean compareObj(Object object1, Object object2) { + if (object1 == null) { + if (object2 == null) + return true; + else + return false; + } else if (object1.equals(object2)) + return true; + else + return false; + } + + /** + * Print Pig Table (for debugging) + * + */ + private int printTable(String tablename) throws IOException { + Iterator<Tuple> it1 = pigServer.openIterator(tablename); + int numbRows = 0; + while (it1.hasNext()) { + Tuple RowValue1 = it1.next(); + ++numbRows; + System.out.println(); + for (int i = 0; i < RowValue1.size(); ++i) + System.out.println("DEBUG: " + tablename + " RowValue.get(" + i + ") = " + RowValue1.get(i)); + } + System.out.println("\nRow count : " + numbRows); + return numbRows; + } + + /** + * Return the name of the routine that called getCurrentMethodName + * + */ + private String getCurrentMethodName() { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + PrintWriter pw = new PrintWriter(baos); + (new Throwable()).printStackTrace(pw); + pw.flush(); + String stackTrace = baos.toString(); + pw.close(); + + StringTokenizer tok = new StringTokenizer(stackTrace, "\n"); + tok.nextToken(); // 'java.lang.Throwable' + tok.nextToken(); // 'at ...getCurrentMethodName' + String l = tok.nextToken(); // 'at ...<caller to getCurrentRoutine>' + // Parse line 3 + tok = new StringTokenizer(l.trim(), " <("); + String t = tok.nextToken(); // 'at' + t = tok.nextToken(); // '...<caller to getCurrentRoutine>' + return t; + } + +} Modified: hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveProjectionNegative.java URL: http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveProjectionNegative.java?rev=895751&r1=895750&r2=895751&view=diff ============================================================================== --- hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveProjectionNegative.java (original) +++ hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveProjectionNegative.java Mon Jan 4 19:32:00 2010 @@ -455,7 +455,7 @@ } finally { //System.out.println(getStackTrace(exception)); Assert.assertNotNull(exception); - Assert.assertTrue(getStackTrace(exception).contains("Schema file doesn't exist")); + Assert.assertTrue(getStackTrace(exception).contains("Input path does not exist: ")); } } @@ -465,6 +465,7 @@ // Test sorted union error handling when one of the table paths is invalid (Negative test) // IOException exception = null; + String pathSort2 = null; try { // Sort tables @@ -479,7 +480,7 @@ "('" + TABLE1_STORAGE + "')"); Assert.assertNull(pigJob.getException()); - String pathSort2 = newPath.toString() + "2"; // invalid path + pathSort2 = newPath.toString() + "2"; // invalid path String queryLoad = "records1 = LOAD '" + pathSort1 + "," From pig-commitskellyelizabeth@example.org Mon Jan 04 19:34:18 2010 Return-Path: <pig-commitskellyelizabeth@example.org> Delivered-To: kellyelizabeth@example.org Received: (qmail 50141 invoked from network); 4 Jan 2010 19:34:18 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 4 Jan 2010 19:34:18 -0000 Received: (qmail 35640 invoked by uid 500); 4 Jan 2010 19:34:18 -0000 Delivered-To: kellyelizabeth@example.org Received: (qmail 35620 invoked by uid 500); 4 Jan 2010 19:34:17 -0000 Mailing-List: contact kellyelizabeth@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:kellyelizabeth@example.org> List-Unsubscribe: <mailto:kellyelizabeth@example.org> List-Post: <mailto:kellyelizabeth@example.org> List-Id: <pig-commits.hadoop.apache.org> Reply-To: kellyelizabeth@example.org Delivered-To: mailing list kellyelizabeth@example.org Received: (qmail 35611 invoked by uid 500); 4 Jan 2010 19:34:17 -0000 Delivered-To: kellyelizabeth@example.org Received: (qmail 35608 invoked by uid 99); 4 Jan 2010 19:34:17 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 04 Jan 2010 19:34:17 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 04 Jan 2010 19:34:16 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 88B44238898B; Mon, 4 Jan 2010 19:33:56 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r895753 - in /hadoop/pig/trunk/contrib/zebra: ./ src/java/org/apache/hadoop/zebra/pig/ src/test/org/apache/hadoop/zebra/pig/ Date: Mon, 04 Jan 2010 19:33:55 -0000 To: kellyelizabeth@example.org From: kellyelizabeth@example.org X-Mailer: svnmailer-1.0.8 Message-Id: <kellyelizabeth@example.org> Author: yanz Date: Mon Jan 4 19:33:53 2010 New Revision: 895753 URL: http://svn.apache.org/viewvc?rev=895753&view=rev Log: PIG-1167: Hadoop file glob support (yanz) Added: hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestGlobTableLoader.java hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveMultiTableGlob.java Modified: hadoop/pig/trunk/contrib/zebra/CHANGES.txt hadoop/pig/trunk/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveProjectionNegative.java Modified: hadoop/pig/trunk/contrib/zebra/CHANGES.txt URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/contrib/zebra/CHANGES.txt?rev=895753&r1=895752&r2=895753&view=diff ============================================================================== --- hadoop/pig/trunk/contrib/zebra/CHANGES.txt (original) +++ hadoop/pig/trunk/contrib/zebra/CHANGES.txt Mon Jan 4 19:33:53 2010 @@ -52,6 +52,8 @@ BUG FIXES + PIG-1167: Hadoop file glob support (yanz) + PIG-1153: Record split exception fix (yanz) PIG-1145: Merge Join on Large Table throws an EOF exception (yanz) Modified: hadoop/pig/trunk/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java?rev=895753&r1=895752&r2=895753&view=diff ============================================================================== --- hadoop/pig/trunk/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java (original) +++ hadoop/pig/trunk/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java Mon Jan 4 19:33:53 2010 @@ -226,7 +226,7 @@ FileSystem fs = p.getFileSystem(jobConf); FileStatus[] matches = fs.globStatus(p); if (matches == null) { - LOG.warn("Input path does not exist: " + p); + throw new IOException("Input path does not exist: " + p); } else if (matches.length == 0) { LOG.warn("Input Pattern " + p + " matches 0 files"); @@ -293,33 +293,14 @@ Projection projection; - if (!fileName.contains(",")) { // one table; - org.apache.hadoop.zebra.schema.Schema tschema = BasicTable.Reader.getSchema(new Path(fileName), jobConf); - try { - projection = new org.apache.hadoop.zebra.types.Projection(tschema, TableInputFormat.getProjection(jobConf)); - projectionSchema = projection.getProjectionSchema(); - } catch (ParseException e) { - throw new IOException("Schema parsing failed : "+e.getMessage()); - } - } else { // table union; - org.apache.hadoop.zebra.schema.Schema unionSchema = new org.apache.hadoop.zebra.schema.Schema(); - for (Path p : paths) { - org.apache.hadoop.zebra.schema.Schema schema = BasicTable.Reader.getSchema(p, jobConf); - try { - unionSchema.unionSchema(schema); - } catch (ParseException e) { - throw new IOException(e.getMessage()); - } - } - - try { - projection = new org.apache.hadoop.zebra.types.Projection(unionSchema, TableInputFormat.getProjection(jobConf)); - projectionSchema = projection.getProjectionSchema(); - } catch (ParseException e) { - throw new IOException("Schema parsing failed : "+e.getMessage()); - } - } - + org.apache.hadoop.zebra.schema.Schema tschema = TableInputFormat.getSchema(jobConf); + try { + projection = new org.apache.hadoop.zebra.types.Projection(tschema, TableInputFormat.getProjection(jobConf)); + projectionSchema = projection.getProjectionSchema(); + } catch (ParseException e) { + throw new IOException("Schema parsing failed : "+e.getMessage()); + } + if (projectionSchema == null) { throw new IOException("Cannot determine table projection schema"); } Added: hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestGlobTableLoader.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestGlobTableLoader.java?rev=895753&view=auto ============================================================================== --- hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestGlobTableLoader.java (added) +++ hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestGlobTableLoader.java Mon Jan 4 19:33:53 2010 @@ -0,0 +1,232 @@ +/** + * 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.zebra.pig; + +import java.io.File; +import java.io.IOException; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.BytesWritable; +import org.apache.hadoop.zebra.io.BasicTable; +import org.apache.hadoop.zebra.io.TableInserter; +import org.apache.hadoop.zebra.io.TableScanner; +import org.apache.hadoop.zebra.io.BasicTable.Reader.RangeSplit; +import org.apache.hadoop.zebra.parser.ParseException; +import org.apache.hadoop.zebra.schema.Schema; +import org.apache.hadoop.zebra.types.TypesUtils; +import org.apache.pig.ExecType; +import org.apache.pig.PigServer; +import org.apache.pig.backend.executionengine.ExecException; +import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil; +import org.apache.pig.data.Tuple; +import org.apache.pig.test.MiniCluster; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +/** + * Note: + * + * Make sure you add the build/pig-0.1.0-dev-core.jar to the Classpath of the + * app/debug configuration, when run this from inside the Eclipse. + * + */ +public class TestGlobTableLoader{ + protected static ExecType execType = ExecType.MAPREDUCE; + private static MiniCluster cluster; + protected static PigServer pigServer; + private static Path pathTable; + private static Configuration conf; + private static String zebraJar; + private static String whichCluster; + private static FileSystem fs; + @BeforeClass + public static void setUp() throws Exception { + if (System.getProperty("hadoop.log.dir") == null) { + String base = new File(".").getPath(); // getAbsolutePath(); + System + .setProperty("hadoop.log.dir", new Path(base).toString() + "./logs"); + } + + // if whichCluster is not defined, or defined something other than + // "realCluster" or "miniCluster", set it to "miniCluster" + if (System.getProperty("whichCluster") == null + || ((!System.getProperty("whichCluster") + .equalsIgnoreCase("realCluster")) && (!System.getProperty( + "whichCluster").equalsIgnoreCase("miniCluster")))) { + System.setProperty("whichCluster", "miniCluster"); + whichCluster = System.getProperty("whichCluster"); + } else { + whichCluster = System.getProperty("whichCluster"); + } + + System.out.println("cluster: " + whichCluster); + if (whichCluster.equalsIgnoreCase("realCluster") + && System.getenv("HADOOP_HOME") == null) { + System.out.println("Please set HADOOP_HOME"); + System.exit(0); + } + + conf = new Configuration(); + + if (whichCluster.equalsIgnoreCase("realCluster") + && System.getenv("USER") == null) { + System.out.println("Please set USER"); + System.exit(0); + } + zebraJar = System.getenv("HADOOP_HOME") + "/../jars/zebra.jar"; + File file = new File(zebraJar); + if (!file.exists() && whichCluster.equalsIgnoreCase("realCulster")) { + System.out.println("Please put zebra.jar at hadoop_home/../jars"); + System.exit(0); + } + + if (whichCluster.equalsIgnoreCase("realCluster")) { + pigServer = new PigServer(ExecType.MAPREDUCE, ConfigurationUtil + .toProperties(conf)); + pigServer.registerJar(zebraJar); + pathTable = new Path("/user/" + System.getenv("USER") + + "/TestMapTableLoader"); + removeDir(pathTable); + fs = pathTable.getFileSystem(conf); + } + + if (whichCluster.equalsIgnoreCase("miniCluster")) { + if (execType == ExecType.MAPREDUCE) { + cluster = MiniCluster.buildCluster(); + pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties()); + fs = cluster.getFileSystem(); + pathTable = new Path(fs.getWorkingDirectory() + + "/TestMapTableLoader1"); + removeDir(pathTable); + System.out.println("path1 =" + pathTable); + } else { + pigServer = new PigServer(ExecType.LOCAL); + } + } + + + BasicTable.Writer writer = new BasicTable.Writer(pathTable, + "m1:map(string)", "[m1#{a}]", conf); + Schema schema = writer.getSchema(); + Tuple tuple = TypesUtils.createTuple(schema); + + final int numsBatch = 10; + final int numsInserters = 2; + TableInserter[] inserters = new TableInserter[numsInserters]; + for (int i = 0; i < numsInserters; i++) { + inserters[i] = writer.getInserter("ins" + i, false); + } + + for (int b = 0; b < numsBatch; b++) { + for (int i = 0; i < numsInserters; i++) { + TypesUtils.resetTuple(tuple); + Map<String, String> map = new HashMap<String, String>(); + map.put("a", "x"); + map.put("b", "y"); + map.put("c", "z"); + tuple.set(0, map); + + try { + inserters[i].insert(new BytesWritable(("key" + i).getBytes()), tuple); + } catch (Exception e) { + System.out.println(e.getMessage()); + } + } + } + for (int i = 0; i < numsInserters; i++) { + inserters[i].close(); + } + } + + @AfterClass + public static void tearDown() throws Exception { + pigServer.shutdown(); + } + public static void removeDir(Path outPath) throws IOException { + String command = null; + if (whichCluster.equalsIgnoreCase("realCluster")) { + command = System.getenv("HADOOP_HOME") +"/bin/hadoop fs -rmr " + outPath.toString(); + } + else{ + command = "rm -rf " + outPath.toString(); + } + Runtime runtime = Runtime.getRuntime(); + Process proc = runtime.exec(command); + int exitVal = -1; + try { + exitVal = proc.waitFor(); + } catch (InterruptedException e) { + System.err.println(e); + } + + } + + // @Test + public void test1() throws IOException, ParseException { + String projection = new String("m1#{b}"); + BasicTable.Reader reader = new BasicTable.Reader(pathTable, conf); + reader.setProjection(projection); + // long totalBytes = reader.getStatus().getSize(); + + List<RangeSplit> splits = reader.rangeSplit(1); + reader.close(); + reader = new BasicTable.Reader(pathTable, conf); + reader.setProjection(projection); + + TableScanner scanner = reader.getScanner(splits.get(0), true); + BytesWritable key = new BytesWritable(); + Tuple value = TypesUtils.createTuple(scanner.getSchema()); + // HashMap<String, Object> mapval; + while (!scanner.atEnd()) { + scanner.getKey(key); + // Assert.assertEquals(key, new BytesWritable("key0".getBytes())); + scanner.getValue(value); + System.out.println("key = " + key + " value = " + value); + + // mapval = (HashMap<String, Object>) value.get(0); + // Assert.assertEquals("x", mapval.get("a")); + // Assert.assertEquals(null, mapval.get("b")); + // Assert.assertEquals(null, mapval.get("c")); + scanner.advance(); + } + reader.close(); + } + + @Test + public void testReader() throws ExecException, IOException { + pathTable = new Path("/user/" + System.getenv("USER") + + "/{TestMapTableLoader1}"); + String query = "records = LOAD '" + pathTable.toString() + + "' USING org.apache.hadoop.zebra.pig.TableLoader('m1#{a}');"; + System.out.println(query); + pigServer.registerQuery(query); + Iterator<Tuple> it = pigServer.openIterator("records"); + while (it.hasNext()) { + Tuple cur = it.next(); + System.out.println(cur); + } + } +} Added: hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveMultiTableGlob.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveMultiTableGlob.java?rev=895753&view=auto ============================================================================== --- hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveMultiTableGlob.java (added) +++ hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveMultiTableGlob.java Mon Jan 4 19:33:53 2010 @@ -0,0 +1,424 @@ +/** + * 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.zebra.pig; + +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.PrintWriter; +import java.util.Iterator; +import java.util.ArrayList; +import java.util.StringTokenizer; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.BytesWritable; +import org.apache.hadoop.zebra.io.BasicTable; +import org.apache.hadoop.zebra.io.TableInserter; +import org.apache.hadoop.zebra.pig.TableStorer; +import org.apache.hadoop.zebra.schema.Schema; +import org.apache.hadoop.zebra.types.TypesUtils; +import org.apache.pig.ExecType; +import org.apache.pig.PigServer; +import org.apache.pig.backend.executionengine.ExecException; +import org.apache.pig.backend.executionengine.ExecJob; +import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil; +import org.apache.pig.data.DataByteArray; +import org.apache.pig.data.Tuple; +import org.apache.pig.test.MiniCluster; + +import junit.framework.Assert; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + + +public class TestOrderPreserveMultiTableGlob { + + final static int NUMB_TABLE = 10; // number of tables for stress test + final static int NUMB_TABLE_ROWS = 5; // number of rows for each table + + final static String TABLE_SCHEMA = "int1:int,str1:string,byte1:bytes"; + final static String TABLE_STORAGE = "[int1,str1,byte1]"; + + static int fileId = 0; + static int sortId = 0; + + protected static ExecType execType = ExecType.MAPREDUCE; + private static MiniCluster cluster; + protected static PigServer pigServer; + protected static ExecJob pigJob; + + private static ArrayList<Path> pathTables; + private static int totalTableRows =0; + + private static Configuration conf; + private static FileSystem fs; + + private static String zebraJar; + private static String whichCluster; + + @BeforeClass + public static void setUp() throws Exception { + if (System.getProperty("hadoop.log.dir") == null) { + String base = new File(".").getPath(); // getAbsolutePath(); + System.setProperty("hadoop.log.dir", new Path(base).toString() + "./logs"); + } + + // if whichCluster is not defined, or defined something other than + // "realCluster" or "miniCluster", set it to "realCluster" + if (System.getProperty("whichCluster") == null + || ((!System.getProperty("whichCluster") + .equalsIgnoreCase("realCluster")) && (!System.getProperty( + "whichCluster").equalsIgnoreCase("miniCluster")))) { + System.setProperty("whichCluster", "miniCluster"); + whichCluster = System.getProperty("whichCluster"); + } else { + whichCluster = System.getProperty("whichCluster"); + } + + System.out.println("cluster: " + whichCluster); + if (whichCluster.equalsIgnoreCase("realCluster") + && System.getenv("HADOOP_HOME") == null) { + System.out.println("Please set HADOOP_HOME"); + System.exit(0); + } + + conf = new Configuration(); + + if (whichCluster.equalsIgnoreCase("realCluster") + && System.getenv("USER") == null) { + System.out.println("Please set USER"); + System.exit(0); + } + zebraJar = System.getenv("HADOOP_HOME") + "/../jars/zebra.jar"; + File file = new File(zebraJar); + if (!file.exists() && whichCluster.equalsIgnoreCase("realCulster")) { + System.out.println("Please put zebra.jar at hadoop_home/../jars"); + System.exit(0); + } + + if (whichCluster.equalsIgnoreCase("realCluster")) { + pigServer = new PigServer(ExecType.MAPREDUCE, ConfigurationUtil + .toProperties(conf)); + pigServer.registerJar(zebraJar); + + pathTables = new ArrayList<Path>(); + for (int i=0; i<NUMB_TABLE; ++i) { + Path pathTable = new Path("/user/" + System.getenv("USER") + + "/TestOderPerserveMultiTable" + i); + pathTables.add(pathTable); + removeDir(pathTable); + } + fs = pathTables.get(0).getFileSystem(conf); + } + + if (whichCluster.equalsIgnoreCase("miniCluster")) { + if (execType == ExecType.MAPREDUCE) { + cluster = MiniCluster.buildCluster(); + pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties()); + fs = cluster.getFileSystem(); + + pathTables = new ArrayList<Path>(); + for (int i=0; i<NUMB_TABLE; ++i) { + Path pathTable = new Path(fs.getWorkingDirectory() + + "/TestOderPerserveMultiTable" + i); + pathTables.add(pathTable); + removeDir(pathTable); + } + } else { + pigServer = new PigServer(ExecType.LOCAL); + } + } + + // Create tables + for (int i=0; i<NUMB_TABLE; ++i) { + // Create table data + Object[][] table = new Object[NUMB_TABLE_ROWS][3]; // three columns + + for (int j=0; j<NUMB_TABLE_ROWS; ++j) { + table[j][0] = i; + table[j][1] = new String("string" + j); + table[j][2] = new DataByteArray("byte" + (NUMB_TABLE_ROWS - j)); + ++totalTableRows; + } + // Create table + createTable(pathTables.get(i), TABLE_SCHEMA, TABLE_STORAGE, table); + + // Load Table + String query = "table" + i + " = LOAD '" + pathTables.get(i).toString() + + "' USING org.apache.hadoop.zebra.pig.TableLoader();"; + pigServer.registerQuery(query); + } + } + + private static void createTable(Path path, String schemaString, String storageString, Object[][] tableData) + throws IOException { + // + // Create table from tableData array + // + BasicTable.Writer writer = new BasicTable.Writer(path, schemaString, storageString, conf); + + Schema schema = writer.getSchema(); + Tuple tuple = TypesUtils.createTuple(schema); + TableInserter inserter = writer.getInserter("ins", false); + + for (int i = 0; i < tableData.length; ++i) { + TypesUtils.resetTuple(tuple); + for (int k = 0; k < tableData[i].length; ++k) { + tuple.set(k, tableData[i][k]); + System.out.println("DEBUG: setting tuple k=" + k + "value= " + tableData[i][k]); + } + inserter.insert(new BytesWritable(("key" + i).getBytes()), tuple); + } + inserter.close(); + writer.close(); + } + + @AfterClass + public static void tearDown() throws Exception { + pigServer.shutdown(); + } + + public static void removeDir(Path outPath) throws IOException { + String command = null; + if (whichCluster.equalsIgnoreCase("realCluster")) { + command = System.getenv("HADOOP_HOME") +"/bin/hadoop fs -rmr " + outPath.toString(); + } + else{ + command = "rm -rf " + outPath.toString(); + } + Runtime runtime = Runtime.getRuntime(); + Process proc = runtime.exec(command); + int exitVal = -1; + try { + exitVal = proc.waitFor(); + } catch (InterruptedException e) { + System.err.println(e); + } + } + + private Iterator<Tuple> testOrderPreserveUnion(ArrayList<String> inputTables, String sortkey, String columns) + throws IOException { + // + // Test order preserve union from input tables and provided output columns + // + Assert.assertTrue("Table union requires two or more input tables", inputTables.size() >= 2); + + Path newPath = new Path(getCurrentMethodName()); + ArrayList<String> pathList = new ArrayList<String>(); + + // Load and store each of the input tables + for (int i=0; i<inputTables.size(); ++i) { + String tablename = inputTables.get(i); + String sortName = "sort" + ++sortId; + + // Sort tables + String orderby = sortName + " = ORDER " + tablename + " BY " + sortkey + " ;"; + pigServer.registerQuery(orderby); + + String sortPath = new String(newPath.toString() + ++fileId); // increment fileId suffix + + // Store sorted tables + pigJob = pigServer.store(sortName, sortPath, TableStorer.class.getCanonicalName() + + "('" + TABLE_STORAGE + "')"); + Assert.assertNull(pigJob.getException()); + + pathList.add(sortPath); // add table path to list + } + + String paths = new String(); + + paths += newPath.toString() + "{"; + fileId = 0; + for (String path:pathList) + paths += ++fileId + ","; + paths = paths.substring(0, paths.lastIndexOf(",")); // remove trailing comma + paths += "}"; + + String queryLoad = "records1 = LOAD '" + + paths + + "' USING org.apache.hadoop.zebra.pig.TableLoader('" + columns + "', 'sorted');"; + + System.out.println("queryLoad: " + queryLoad); + pigServer.registerQuery(queryLoad); + + // Return iterator + Iterator<Tuple> it1 = pigServer.openIterator("records1"); + return it1; + } + + @Test + public void test_sorted_union_multi_table() throws ExecException, IOException { + // + // Test sorted union + // + + // Create input tables for order preserve union + ArrayList<String> inputTables = new ArrayList<String>(); // Input tables + for (int i=0; i<NUMB_TABLE; ++i) { + inputTables.add("table" + i); // add input table + } + + // Test with input tables and provided output columns + testOrderPreserveUnion(inputTables, "int1", "int1, str1, byte1"); + + // Create results table for verification + ArrayList<ArrayList<Object>> resultTable = new ArrayList<ArrayList<Object>>(); + for (int i=0; i<NUMB_TABLE; ++i) { + for (int j=0; j<NUMB_TABLE_ROWS; ++j) { + ArrayList<Object> resultRow = new ArrayList<Object>(); + + resultRow.add(i); // int1 + resultRow.add(new String("string" + j)); // str1 + resultRow.add(new DataByteArray("byte" + (NUMB_TABLE_ROWS - j))); // byte1 + + resultTable.add(resultRow); + } + } + + // Verify union table + Iterator<Tuple> it = pigServer.openIterator("records1"); + int numbRows = verifyTable(resultTable, 0, it); + + Assert.assertEquals(totalTableRows, numbRows); + + // Print Table + //printTable("records1"); + } + + /** + * Verify union output table with expected results + * + */ + private int verifyTable(ArrayList<ArrayList<Object>> resultTable, int keyColumn, Iterator<Tuple> it) throws IOException { + int numbRows = 0; + int index = 0; + Object value = resultTable.get(index).get(keyColumn); // get value of primary key + + while (it.hasNext()) { + Tuple rowValues = it.next(); + + // If last primary sort key does match then search for next matching key + if (! compareObj(value, rowValues.get(keyColumn))) { + int subIndex = index + 1; + while (subIndex < resultTable.size()) { + if ( ! compareObj(value, resultTable.get(subIndex).get(keyColumn)) ) { // found new key + index = subIndex; + value = resultTable.get(index).get(keyColumn); + break; + } + ++subIndex; + } + Assert.assertEquals("Table comparison error for row : " + numbRows + " - no key found for : " + + rowValues.get(keyColumn), value, rowValues.get(keyColumn)); + } + // Search for matching row with this primary key + int subIndex = index; + + while (subIndex < resultTable.size()) { + // Compare row + ArrayList<Object> resultRow = resultTable.get(subIndex); + if ( compareRow(rowValues, resultRow) ) + break; // found matching row + ++subIndex; + Assert.assertEquals("Table comparison error for row : " + numbRows + " - no matching row found for : " + + rowValues.get(keyColumn), value, resultTable.get(subIndex).get(keyColumn)); + } + ++numbRows; + } + Assert.assertEquals(resultTable.size(), numbRows); // verify expected row count + return numbRows; + } + + /** + * Compare table rows + * + */ + private boolean compareRow(Tuple rowValues, ArrayList<Object> resultRow) throws IOException { + boolean result = true; + Assert.assertEquals(resultRow.size(), rowValues.size()); + for (int i = 0; i < rowValues.size(); ++i) { + if (! compareObj(rowValues.get(i), resultRow.get(i)) ) { + result = false; + break; + } + } + return result; + } + + /** + * Compare table values + * + */ + private boolean compareObj(Object object1, Object object2) { + if (object1 == null) { + if (object2 == null) + return true; + else + return false; + } else if (object1.equals(object2)) + return true; + else + return false; + } + + /** + * Print Pig Table (for debugging) + * + */ + private int printTable(String tablename) throws IOException { + Iterator<Tuple> it1 = pigServer.openIterator(tablename); + int numbRows = 0; + while (it1.hasNext()) { + Tuple RowValue1 = it1.next(); + ++numbRows; + System.out.println(); + for (int i = 0; i < RowValue1.size(); ++i) + System.out.println("DEBUG: " + tablename + " RowValue.get(" + i + ") = " + RowValue1.get(i)); + } + System.out.println("\nRow count : " + numbRows); + return numbRows; + } + + /** + * Return the name of the routine that called getCurrentMethodName + * + */ + private String getCurrentMethodName() { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + PrintWriter pw = new PrintWriter(baos); + (new Throwable()).printStackTrace(pw); + pw.flush(); + String stackTrace = baos.toString(); + pw.close(); + + StringTokenizer tok = new StringTokenizer(stackTrace, "\n"); + tok.nextToken(); // 'java.lang.Throwable' + tok.nextToken(); // 'at ...getCurrentMethodName' + String l = tok.nextToken(); // 'at ...<caller to getCurrentRoutine>' + // Parse line 3 + tok = new StringTokenizer(l.trim(), " <("); + String t = tok.nextToken(); // 'at' + t = tok.nextToken(); // '...<caller to getCurrentRoutine>' + return t; + } + +} Modified: hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveProjectionNegative.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveProjectionNegative.java?rev=895753&r1=895752&r2=895753&view=diff ============================================================================== --- hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveProjectionNegative.java (original) +++ hadoop/pig/trunk/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestOrderPreserveProjectionNegative.java Mon Jan 4 19:33:53 2010 @@ -455,7 +455,7 @@ } finally { //System.out.println(getStackTrace(exception)); Assert.assertNotNull(exception); - Assert.assertTrue(getStackTrace(exception).contains("Schema file doesn't exist")); + Assert.assertTrue(getStackTrace(exception).contains("Input path does not exist: ")); } } @@ -465,6 +465,7 @@ // Test sorted union error handling when one of the table paths is invalid (Negative test) // IOException exception = null; + String pathSort2 = null; try { // Sort tables @@ -479,7 +480,7 @@ "('" + TABLE1_STORAGE + "')"); Assert.assertNull(pigJob.getException()); - String pathSort2 = newPath.toString() + "2"; // invalid path + pathSort2 = newPath.toString() + "2"; // invalid path String queryLoad = "records1 = LOAD '" + pathSort1 + "," From pig-commitskellyelizabeth@example.org Mon Jan 04 19:37:32 2010 Return-Path: <pig-commitskellyelizabeth@example.org> Delivered-To: kellyelizabeth@example.org Received: (qmail 50685 invoked from network); 4 Jan 2010 19:37:32 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 4 Jan 2010 19:37:32 -0000 Received: (qmail 37485 invoked by uid 500); 4 Jan 2010 19:37:32 -0000 Delivered-To: kellyelizabeth@example.org Received: (qmail 37440 invoked by uid 500); 4 Jan 2010 19:37:32 -0000 Mailing-List: contact kellyelizabeth@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:kellyelizabeth@example.org> List-Unsubscribe: <mailto:kellyelizabeth@example.org> List-Post: <mailto:kellyelizabeth@example.org> List-Id: <pig-commits.hadoop.apache.org> Reply-To: kellyelizabeth@example.org Delivered-To: mailing list kellyelizabeth@example.org Received: (qmail 37431 invoked by uid 500); 4 Jan 2010 19:37:32 -0000 Delivered-To: kellyelizabeth@example.org Received: (qmail 37427 invoked by uid 99); 4 Jan 2010 19:37:32 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 04 Jan 2010 19:37:32 +0000 X-ASF-Spam-Status: No, hits=-1999.6 required=10.0 tests=ALL_TRUSTED,SUBJECT_FUZZY_TION X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 04 Jan 2010 19:37:30 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 6C124238898B; Mon, 4 Jan 2010 19:37:08 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r895754 - in /hadoop/pig/branches/load-store-redesign: src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/ src/org/apache/pig/impl/builtin/ test/org/apache/pig/test/ test/org/apache/pig/test/data/GoldenFiles/ Date: Mon, 04 Jan 2010 19:37:08 -0000 To: kellyelizabeth@example.org From: kellyelizabeth@example.org X-Mailer: svnmailer-1.0.8 Message-Id: <kellyelizabeth@example.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: pradeepkth Date: Mon Jan 4 19:37:05 2010 New Revision: 895754 URL: http://svn.apache.org/viewvc?rev=895754&view=rev Log: PIG-1094: Fix unit tests corresponding to source changes so far - fixes for some order by,merge join and skew join related tests (thejas via pradeepkth) Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/WeightedRangePartitioner.java hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/GetMemNumRows.java hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/PoissonSampleLoader.java hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPoissonSampleLoader.java hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/WeightedRangePartitioner.java URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/WeightedRangePartitioner.java?rev=895754&r1=895753&r2=895754&view=diff ============================================================================== --- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/WeightedRangePartitioner.java (original) +++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/WeightedRangePartitioner.java Mon Jan 4 19:37:05 2010 @@ -29,6 +29,8 @@ import org.apache.hadoop.io.RawComparator; import org.apache.hadoop.io.Writable; import org.apache.hadoop.mapreduce.Partitioner; +import org.apache.pig.FuncSpec; +import org.apache.pig.LoadFunc; import org.apache.pig.backend.executionengine.ExecException; import org.apache.pig.backend.hadoop.HDataType; import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil; @@ -37,6 +39,7 @@ import org.apache.pig.data.DataBag; import org.apache.pig.data.InternalMap; import org.apache.pig.data.Tuple; +import org.apache.pig.impl.PigContext; import org.apache.pig.impl.builtin.FindQuantiles; import org.apache.pig.impl.io.BufferedPositionedInputStream; import org.apache.pig.impl.io.FileLocalizer; @@ -93,7 +96,6 @@ throw new RuntimeException(this.getClass().getSimpleName() + " used but no quantiles found"); try{ - InputStream is = FileLocalizer.openDFSFile(quantilesFile,ConfigurationUtil.toProperties(job)); ReadToEndLoader loader = new ReadToEndLoader(new BinStorage(), job, quantilesFile, 0); DataBag quantilesList; @@ -114,17 +116,25 @@ new DiscreteProbabilitySampleGenerator(probVec)); } } - else - { - ArrayList<Pair<FileSpec, Boolean>> inp = (ArrayList<Pair<FileSpec, Boolean>>)ObjectSerializer.deserialize(job.get("pig.inputs", "")); - String inputFileName = inp.get(0).first.getFileName(); - long inputSize = FileLocalizer.getSize(inputFileName); - if (inputSize!=0) + else { + ArrayList<FileSpec> inp = + (ArrayList<FileSpec>) + ObjectSerializer.deserialize(job.get("pig.inputs", "")); + //order-by MR job will have only one input + FileSpec fileSpec = inp.get(0); + LoadFunc inpLoad = + (LoadFunc)PigContext.instantiateFuncFromSpec(fileSpec.getFuncSpec()); + + ReadToEndLoader r2eLoad = new ReadToEndLoader(inpLoad, job, + fileSpec.getFileName(), 0); + + if (r2eLoad.getNext() != null) { throw new RuntimeException("Empty samples file and non-empty input file"); } // Otherwise, we do not put anything to weightedParts } + }catch (Exception e){ throw new RuntimeException(e); } Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/GetMemNumRows.java URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/GetMemNumRows.java?rev=895754&r1=895753&r2=895754&view=diff ============================================================================== --- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/GetMemNumRows.java (original) +++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/GetMemNumRows.java Mon Jan 4 19:37:05 2010 @@ -54,8 +54,8 @@ // if this is specially marked tuple, get the number of rows int tSize = in.size(); - if(tSize >=2 && - in.get(tSize-2).equals(PoissonSampleLoader.NUMROWS_TUPLE_MARKER)){ + if(tSize >=2 && + PoissonSampleLoader.NUMROWS_TUPLE_MARKER.equals(in.get(tSize-2)) ){ numRows = (Long)in.get(tSize-1); } Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/PoissonSampleLoader.java URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/PoissonSampleLoader.java?rev=895754&r1=895753&r2=895754&view=diff ============================================================================== --- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/PoissonSampleLoader.java (original) +++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/PoissonSampleLoader.java Mon Jan 4 19:37:05 2010 @@ -34,8 +34,10 @@ */ public class PoissonSampleLoader extends SampleLoader { - // marker string for special row with total number or rows. - // this will be value of first column in the special row + // marker string to mark the last sample row, which has total number or rows + // seen by this map instance + // this string will be in the 2nd last column of the last sample row + // it is used by GetMemNumRows public static final String NUMROWS_TUPLE_MARKER = "\u4956\u3838_pig_inTeRnal-spEcial_roW_num_tuple3kt579CFLehkblah"; @@ -90,7 +92,7 @@ // were read, nothing more to read return null; } - + if(skipInterval == -1){ //select first tuple as sample and calculate @@ -101,9 +103,9 @@ long availRedMem = (long) (Runtime.getRuntime().maxMemory() * heapPerc); memToSkipPerSample = availRedMem/sampleRate; updateSkipInterval(t); - + rowNum++; - newSample = t; + newSample = t; } // skip tuples @@ -113,16 +115,16 @@ } rowNum++; } - + // skipped enough, get new sample Tuple t = loader.getNext(); if(t == null) return createNumRowTuple(newSample); updateSkipInterval(t); rowNum++; - Tuple currentSample = newSample; - newSample = t; - return currentSample; + Tuple currentSample = newSample; + newSample = t; + return currentSample; } /** @@ -141,7 +143,7 @@ // very few samples being sampled. Sampling a little extra is OK if(numRowsSampled < 5) skipInterval = skipInterval/(10-numRowsSampled); - ++numRowsSampled; + ++numRowsSampled; } Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPoissonSampleLoader.java URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPoissonSampleLoader.java?rev=895754&r1=895753&r2=895754&view=diff ============================================================================== --- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPoissonSampleLoader.java (original) +++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPoissonSampleLoader.java Mon Jan 4 19:37:05 2010 @@ -32,10 +32,10 @@ import org.apache.pig.data.Tuple; import org.apache.pig.impl.builtin.PoissonSampleLoader; import org.apache.pig.impl.util.Pair; +import org.apache.pig.impl.io.FileSpec; import org.junit.After; import org.junit.Before; import org.junit.Test; -import org.apache.pig.impl.io.FileSpec; public class TestPoissonSampleLoader extends TestCase{ @@ -86,33 +86,19 @@ } @Test - public void testComputeSamples() throws IOException{ - FileSpec fs = new FileSpec(INPUT_FILE1, new FuncSpec(PigStorage.class.getName())); - - ArrayList<Pair<FileSpec, Boolean>> inputs = new ArrayList<Pair<FileSpec, Boolean> >(); - inputs.add(new Pair<FileSpec, Boolean>(fs, true)); - - // Use 100 as a default value; - PoissonSampleLoader ps = new PoissonSampleLoader((new FuncSpec(PigStorage.class.getName())).toString(), "100"); - - // Get the number of samples for the file - ps.computeSamples(inputs, pigServer.getPigContext()); - - if (ps.getNumSamples() != 3) { - fail("Compute samples returned the wrong number of samples"); + public void testNumSamples() throws IOException { + pigServer.registerQuery("A = Load '"+INPUT_FILE1+"' Using PoissonSampleLoader('PigStorage()', '100');"); + Iterator<Tuple> iter = pigServer.openIterator("A"); + int count = 0; + while(iter.hasNext()){ + count++; + iter.next(); } + assertEquals(count, 1); } /* - * FIXME This currently tests for 5 elements because PoissonSampleLoader - * only produces a single sample for the test data, and the last sample has - * extra information appended in PoissonSampleLoader. - * - * This is incorrect. The proper number of samples should be > 1, and therefore - * the first sample should only have 3 elements. - * - * See PIG-1062 and PIG-1149 for more information. - * + * Test use of LoadFunc with parameters as argument to PoissonSampleLoader */ @Test public void testInstantiation() throws IOException { @@ -122,6 +108,4 @@ assertEquals(5, iter.next().size()); } - - } \ No newline at end of file Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld?rev=895754&r1=895753&r2=895754&view=diff ============================================================================== --- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld (original) +++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld Mon Jan 4 19:37:05 2010 @@ -1,10 +1,10 @@ MapReduce(-1,PigStorage) - scope-125: Reduce Plan Empty -| Store(file:/tmp:org.apache.pig.builtin.PigStorage) - scope-124 +| Store(file:///tmp:org.apache.pig.builtin.PigStorage) - scope-124 | | | |---MergeJoin[tuple] - scope-121 | | -| |---Load(file:/tmp/input1:org.apache.pig.builtin.PigStorage) - scope-117 +| |---Load(file:///tmp/input1:org.apache.pig.builtin.PigStorage) - scope-117 | |---MapReduce(1,PigStorage) - scope-126: | Store(file:/tmp/temp-1456742965/tmp-1456742965:org.apache.pig.builtin.BinStorage) - scope-133 @@ -20,4 +20,4 @@ | | | | | Constant(all) - scope-127 | | - | |---Load(file:/tmp/input2:org.apache.pig.impl.builtin.MergeJoinIndexer('org.apache.pig.builtin.PigStorage','kmonaaafhdhcaabdgkgbhggbcohfhegjgmcoebhchcgbhjemgjhdhehiibncbnjjmhgbjnadaaabejaaaehdgjhkgfhihaaaaaaaabhhaeaaaaaaabhdhcaaeogphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccohagmgbgohdcofagihjhdgjgdgbgmfagmgbgoaaaaaaaaaaaaaaabacaaabfkaaangfgogeepggebgmgmejgohahfhehihcaacfgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcfagmgbgohlinhnlnngaiiihoacaaagemaaakgneghcgpgnefgeghgfhdheaacdemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphfhegjgmcpenhfgmhegjengbhadlemaaafgnelgfhjhdheaaapemgkgbhggbcphfhegjgmcpengbhadlemaaahgnemgfgbhggfhdheaabaemgkgbhggbcphfhegjgmcpemgjhdhedlemaaaegnephahdhbaahoaaafemaaaggnfcgpgphehdhbaahoaaagemaaaignfegpefgeghgfhdhbaahoaaaehihahdhcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohfhegjgmcoenhfgmhegjengbhaaaaaaaaaaaaaaaacacaaabemaaaegnengbhahbaahoaaafhihahdhcaabbgkgbh ggbcohfhegjgmcoeigbhdgiengbhaafahnkmbmdbgganbadaaacegaaakgmgpgbgeeggbgdhegphcejaaajhegihcgfhdgigpgmgehihadpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhdhcaacegphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcelgfhjaaaaaaaaaaaaaaabacaaacekaaacgjgeemaaafhdgdgphagfheaabcemgkgbhggbcpgmgbgoghcpfdhehcgjgoghdlhihaaaaaaaaaaaaaaahiheaaafhdgdgphagfhdhcaafjgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcofaepfahcgpgkgfgdheaaaaaaaaaaaaaaabacaaaffkaaakgphggfhcgmgpgbgegfgefkaabfhahcgpgdgfhdhdgjgoghecgbghepggfehfhagmgfhdfkaabehcgfhdhfgmhefdgjgoghgmgffehfhagmgfecgbghfkaaaehdhegbhcemaaahgdgpgmhfgngohdheaabfemgkgbhggbcphfhegjgmcpebhchcgbhjemgjhdhedlhihcaagcgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcoefhihahcgfhdhd gjgpgoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaabemaaadgmgpghheaacaemgphcghcpgbhagbgdgigfcpgdgpgngngpgohdcpgmgpghghgjgoghcpemgpghdlhihcaaemgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofagihjhdgjgdgbgmephagfhcgbhegphcaaaaaaaaaaaaaaabacaaamfkaaafgbgdgdhfgnfkaaangjgohahfheebhehegbgdgigfgeejaabehcgfhbhfgfhdhegfgefagbhcgbgmgmgfgmgjhdgnecaaakhcgfhdhfgmhefehjhagfemaaafgbgmgjgbhdhbaahoaaaoemaaafgjgohahfheheaablemgphcghcpgbhagbgdgigfcphagjghcpgegbhegbcpfehfhagmgfdlemaaaggjgohahfhehdhbaahoaaagemaaangmgjgogfgbghgffehcgbgdgfhcheaachemgphcghcpgbhagbgdgigfcphagjghcphagfgocphfhegjgmcpemgjgogfgbghgffehcgbgdgfhcdlemaaadgmgpghhbaahoaabeemaaahgphfhehahfhehdhbaahoaaagemaaakhagbhcgfgohefagmgbgoheaafaemgphcghcpgbhagbgdgigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccphagmgbgohdcpfagihjhdgjgdgbgmfagmgbgodlemaaadhcgfhdheaaeeemgphcghcpgbhagbgdgigfcphagjghcpgcgbgdglgfgogecpgigbg egpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccpfcgfhdhfgmhedlhihcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaabemaaaegnelgfhjheaacgemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphagmgbgocpephagfhcgbhegphcelgfhjdlhihahbaahoaaapaaaappppppppdchahahahahdhcaaclgphcghcogbhagbgdgigfcogdgpgngngpgohdcogmgpghghgjgoghcogjgnhagmcoemgpghdeekemgpghghgfhccikmpnoicknfncdiacaaabemaaaegogbgngfhbaahoaaaohihaheaafjgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcofaepfahcgpgkgfgdhehahahdhcaaecgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofcgfhdhfgmheaaaaaaaaaaaaaaabacaaacecaaamhcgfhehfhcgofdhegbhehfhdemaaaghcgfhdhfgmheheaabcemgkgbhggbcpgmgbgoghcpepgcgkgfgdhedlhihaachahbaahoaaboaaaaaaaahdhbaahoaaaaaaaaaaabhhaeaaaaaaabhdhcaabbgkgbhggbcogmgbgoghcoejgohegf ghgfhcbcockakephibihdiacaaabejaaafhggbgmhfgfhihcaabagkgbhggbcogmgbgoghcoeohfgngcgfhcigkmjfbnaljeoailacaaaahihaaaaaaaaahihihdhbaahoaaaaaaaaaaabhhaeaaaaaaakhbaahoaabmhihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhbaahoaabmhbaahoaaaphihdhbaahoaaaaaaaaaaaahhaeaaaaaaakhihdhbaahoaaaihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahiaahi','')) - scope-118 \ No newline at end of file + | |---Load(file:///tmp/input2:org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MergeJoinIndexer('org.apache.pig.builtin.PigStorage','kmonaaafhdhcaabdgkgbhggbcohfhegjgmcoebhchcgbhjemgjhdhehiibncbnjjmhgbjnadaaabejaaaehdgjhkgfhihaaaaaaaabhhaeaaaaaaabhdhcaaeogphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccohagmgbgohdcofagihjhdgjgdgbgmfagmgbgoaaaaaaaaaaaaaaabacaaabfkaaangfgogeepggebgmgmejgohahfhehihcaacfgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcfagmgbgohlinhnlnngaiiihoacaaagemaaakgneghcgpgnefgeghgfhdheaacdemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphfhegjgmcpenhfgmhegjengbhadlemaaafgnelgfhjhdheaaapemgkgbhggbcphfhegjgmcpengbhadlemaaahgnemgfgbhggfhdheaabaemgkgbhggbcphfhegjgmcpemgjhdhedlemaaaegnephahdhbaahoaaafemaaaggnfcgpgphehdhbaahoaaagemaaaignfegpefgeghgfhdhbaahoaaaehihahdhcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohfhegjgmcoenhfgmhegjengbhaaaaaaaaaaaaaaaacacaaabemaaae gnengbhahbaahoaaafhihahdhcaabbgkgbhggbcohfhegjgmcoeigbhdgiengbhaafahnkmbmdbgganbadaaacegaaakgmgpgbgeeggbgdhegphcejaaajhegihcgfhdgigpgmgehihadpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhdhcaacegphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcelgfhjaaaaaaaaaaaaaaabacaaacekaaacgjgeemaaafhdgdgphagfheaabcemgkgbhggbcpgmgbgoghcpfdhehcgjgoghdlhihaaaaaaaaaaaaaaahiheaaafhdgdgphagfhdhcaafjgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcofaepfahcgpgkgfgdheaaaaaaaaaaaaaaabacaaaffkaaakgphggfhcgmgpgbgegfgefkaabfhahcgpgdgfhdhdgjgoghecgbghepggfehfhagmgfhdfkaabehcgfhdhfgmhefdgjgoghgmgffehfhagmgfecgbghfkaaaehdhegbhcemaaahgdgpgmhfgngohdheaabfemgkgbhggbcphfhegjgmcpebhchcgbhjemgjhdhedlhihcaagcgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpg oephagfhcgbhegphchdcoefhihahcgfhdhdgjgpgoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaabemaaadgmgpghheaacaemgphcghcpgbhagbgdgigfcpgdgpgngngpgohdcpgmgpghghgjgoghcpemgpghdlhihcaaemgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofagihjhdgjgdgbgmephagfhcgbhegphcaaaaaaaaaaaaaaabacaaamfkaaafgbgdgdhfgnfkaaangjgohahfheebhehegbgdgigfgeejaabehcgfhbhfgfhdhegfgefagbhcgbgmgmgfgmgjhdgnecaaakhcgfhdhfgmhefehjhagfemaaafgbgmgjgbhdhbaahoaaaoemaaafgjgohahfheheaablemgphcghcpgbhagbgdgigfcphagjghcpgegbhegbcpfehfhagmgfdlemaaaggjgohahfhehdhbaahoaaagemaaangmgjgogfgbghgffehcgbgdgfhcheaachemgphcghcpgbhagbgdgigfcphagjghcphagfgocphfhegjgmcpemgjgogfgbghgffehcgbgdgfhcdlemaaadgmgpghhbaahoaabeemaaahgphfhehahfhehdhbaahoaaagemaaakhagbhcgfgohefagmgbgoheaafaemgphcghcpgbhagbgdgigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccphagmgbgohdcpfagihjhdgjgdgbgmfagmgbgodlemaaadhcgfhdheaaeeemgphcghcpgbhagbgd gigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccpfcgfhdhfgmhedlhihcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaabemaaaegnelgfhjheaacgemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphagmgbgocpephagfhcgbhegphcelgfhjdlhihahbaahoaaapaaaappppppppdchahahahahdhcaaclgphcghcogbhagbgdgigfcogdgpgngngpgohdcogmgpghghgjgoghcogjgnhagmcoemgpghdeekemgpghghgfhccikmpnoicknfncdiacaaabemaaaegogbgngfhbaahoaaaohihaheaafjgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcofaepfahcgpgkgfgdhehahahdhcaaecgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofcgfhdhfgmheaaaaaaaaaaaaaaabacaaacecaaamhcgfhehfhcgofdhegbhehfhdemaaaghcgfhdhfgmheheaabcemgkgbhggbcpgmgbgoghcpepgcgkgfgdhedlhihaachahbaahoaaboaaaaaaaahdhbaahoaaaaaaaaaaabhhaeaaaaaaabh dhcaabbgkgbhggbcogmgbgoghcoejgohegfghgfhcbcockakephibihdiacaaabejaaafhggbgmhfgfhihcaabagkgbhggbcogmgbgoghcoeohfgngcgfhcigkmjfbnaljeoailacaaaahihaaaaaaaaahihihdhbaahoaaaaaaaaaaabhhaeaaaaaaakhbaahoaabmhihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhbaahoaabmhbaahoaaaphihdhbaahoaaaaaaaaaaaahhaeaaaaaaakhihdhbaahoaaaihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahiaahi','')) - scope-118 \ No newline at end of file From pig-commitskellyelizabeth@example.org Mon Jan 04 22:20:26 2010 Return-Path: <pig-commitskellyelizabeth@example.org> Delivered-To: kellyelizabeth@example.org Received: (qmail 5357 invoked from network); 4 Jan 2010 22:20:26 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 4 Jan 2010 22:20:26 -0000 Received: (qmail 9504 invoked by uid 500); 4 Jan 2010 22:20:25 -0000 Delivered-To: kellyelizabeth@example.org Received: (qmail 9475 invoked by uid 500); 4 Jan 2010 22:20:25 -0000 Mailing-List: contact kellyelizabeth@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:kellyelizabeth@example.org> List-Unsubscribe: <mailto:kellyelizabeth@example.org> List-Post: <mailto:kellyelizabeth@example.org> List-Id: <pig-commits.hadoop.apache.org> Reply-To: kellyelizabeth@example.org Delivered-To: mailing list
{'source': 'Apache Software Foundation Public Mail Archives'}