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 '::' 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 + ","


Reply via email to