pig-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From da...@apache.org
Subject svn commit: r1201553 - in /pig/trunk: CHANGES.txt src/org/apache/pig/builtin/JsonMetadata.java test/org/apache/pig/test/TestPigStorage.java
Date Sun, 13 Nov 2011 23:46:05 GMT
Author: daijy
Date: Sun Nov 13 23:46:04 2011
New Revision: 1201553

URL: http://svn.apache.org/viewvc?rev=1201553&view=rev
Log:
PIG-2209: JsonMetadata fails to find schema for glob paths

Modified:
    pig/trunk/CHANGES.txt
    pig/trunk/src/org/apache/pig/builtin/JsonMetadata.java
    pig/trunk/test/org/apache/pig/test/TestPigStorage.java

Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1201553&r1=1201552&r2=1201553&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Sun Nov 13 23:46:04 2011
@@ -170,6 +170,8 @@ PIG-2228: support partial aggregation in
 
 BUG FIXES
 
+PIG-2209: JsonMetadata fails to find schema for glob paths (daijy)
+
 PIG-2165: Need a way to deal with params and param_file in embedded pig in python (daijy)
 
 PIG-2313: NPE in ILLUSTRATE trying to get StatusReporter in STORE (daijy)

Modified: pig/trunk/src/org/apache/pig/builtin/JsonMetadata.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/builtin/JsonMetadata.java?rev=1201553&r1=1201552&r2=1201553&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/builtin/JsonMetadata.java (original)
+++ pig/trunk/src/org/apache/pig/builtin/JsonMetadata.java Sun Nov 13 23:46:04 2011
@@ -44,6 +44,7 @@ import org.apache.pig.backend.hadoop.dat
 import org.apache.pig.backend.hadoop.datastorage.HDataStorage;
 import org.apache.pig.backend.hadoop.datastorage.HDirectory;
 import org.apache.pig.backend.hadoop.datastorage.HFile;
+import org.apache.pig.backend.hadoop.datastorage.HPath;
 import org.apache.pig.impl.io.FileLocalizer;
 import org.apache.pig.impl.logicalLayer.FrontendException;
 import org.codehaus.jackson.JsonGenerationException;
@@ -115,25 +116,29 @@ public class JsonMetadata implements Loa
             } else {
                 ElementDescriptor[] descriptors = storage.asCollection(loc);
                 for(ElementDescriptor descriptor : descriptors) {
-                    String fileName = null, parentName = null;
-                    ContainerDescriptor parentContainer = null;
+                    ContainerDescriptor container = null;
+                    
                     if (descriptor instanceof HFile) {
-                        Path descriptorPath = ((HFile) descriptor).getPath();
-                        fileName = descriptorPath.getName();
+                        Path descriptorPath = ((HPath) descriptor).getPath();
+                        String fileName = descriptorPath.getName();
                         Path parent = descriptorPath.getParent();
-                        parentName = parent.toString();
-                        parentContainer = new HDirectory((HDataStorage)storage,parent);
-                    }
-                    ElementDescriptor metaFilePath = storage.asElement(parentName, prefix+"."+fileName);
-
-                    // if the file has a custom schema, use it
-                    if (metaFilePath.exists()) {
-                        metaFileSet.add(metaFilePath);
-                        continue;
+                        String parentName = parent.toString();
+                        container = new HDirectory((HDataStorage)storage,parent);
+                        
+                        // try prefix.filename for the file
+                        ElementDescriptor metaFilePath = storage.asElement(parentName, prefix+"."+fileName);
+
+                        // if the file has a custom schema, use it
+                        if (metaFilePath.exists()) {
+                            metaFileSet.add(metaFilePath);
+                            continue;
+                        }
+                    } else { // descriptor instanceof HDirectory
+                        container = (HDirectory)descriptor;
                     }
 
                     // if no custom schema, try the parent directory
-                    metaFilePath = storage.asElement(parentContainer, prefix);
+                    ElementDescriptor metaFilePath = storage.asElement(container, prefix);
                     if (metaFilePath.exists()) {
                         metaFileSet.add(metaFilePath);
                     }

Modified: pig/trunk/test/org/apache/pig/test/TestPigStorage.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestPigStorage.java?rev=1201553&r1=1201552&r2=1201553&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestPigStorage.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestPigStorage.java Sun Nov 13 23:46:04 2011
@@ -37,18 +37,25 @@ import java.util.Properties;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.mapreduce.Job;
 import org.apache.pig.ExecType;
 import org.apache.pig.PigServer;
 import org.apache.pig.ResourceSchema;
 import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POProject;
+import org.apache.pig.builtin.PigStorage;
 import org.apache.pig.data.DataType;
 import org.apache.pig.data.Tuple;
 import org.apache.pig.impl.PigContext;
 import org.apache.pig.impl.io.FileLocalizer;
 import org.apache.pig.impl.logicalLayer.FrontendException;
 import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.util.Utils;
 import org.apache.pig.test.utils.TypeCheckingTestUtil;
+import org.codehaus.jackson.JsonGenerationException;
+import org.codehaus.jackson.map.JsonMappingException;
+import org.codehaus.jackson.map.ObjectMapper;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
@@ -395,4 +402,66 @@ public class TestPigStorage  {
         header = Util.readOutput(pig.getPigContext(), outPath);
         Assert.assertArrayEquals("Headers are not the same.", new String[] {"foo\tbar"},
header);
     }
+    
+    private void putInputFile(String filename) throws IOException {
+        Util.createLocalInputFile(filename, new String[] {});
+    }
+    
+    private void putSchemaFile(String schemaFilename, ResourceSchema testSchema) throws JsonGenerationException,
JsonMappingException, IOException {
+        new ObjectMapper().writeValue(new File(schemaFilename), testSchema);
+    }
+    
+    @Test
+    public void testPigStorageSchemaSearch() throws Exception {
+        String globtestdir = "build/test/tmpglobbingdata/";
+        ResourceSchema testSchema = new ResourceSchema(Utils.parseSchema("a0:chararray"));
+        PigStorage pigStorage = new PigStorage();
+        pigContext.connect();
+        try{
+            Util.deleteDirectory(new File(datadir));
+            pig.mkdirs(globtestdir+"a");
+            pig.mkdirs(globtestdir+"a/a0");
+            putInputFile(globtestdir+"a/a0/input");
+            pig.mkdirs(globtestdir+"a/b0");
+            putInputFile(globtestdir+"a/b0/input");
+            pig.mkdirs(globtestdir+"b");
+        } catch (IOException e) {};
+        
+        // if schema file is not found, schema is null
+        ResourceSchema schema = pigStorage.getSchema(globtestdir, new Job(ConfigurationUtil.toConfiguration(pigContext.getProperties())));
+        Assert.assertTrue(schema==null);
+        
+        // .pig_schema.input in along with input file
+        putSchemaFile(globtestdir+"a/a0/.pig_schema.input", testSchema);
+        schema = pigStorage.getSchema(globtestdir+"a/a0/*", new Job(ConfigurationUtil.toConfiguration(pigContext.getProperties())));
+        Assert.assertTrue(ResourceSchema.equals(schema, testSchema));
+        new File(globtestdir+"a/a0/.pig_schema.input").delete();
+        
+        // if .pig_schema is in the input directory
+        putSchemaFile(globtestdir+"a/a0/.pig_schema", testSchema);
+        schema = pigStorage.getSchema(globtestdir+"a/a0", new Job(ConfigurationUtil.toConfiguration(pigContext.getProperties())));
+        Assert.assertTrue(ResourceSchema.equals(schema, testSchema));
+        new File(globtestdir+"a/a0/.pig_schema").delete();
+        
+        // .pig_schema in one of globStatus returned directory
+        putSchemaFile(globtestdir+"a/.pig_schema", testSchema);
+        schema = pigStorage.getSchema(globtestdir+"*", new Job(ConfigurationUtil.toConfiguration(pigContext.getProperties())));
+        Assert.assertTrue(ResourceSchema.equals(schema, testSchema));
+        new File(globtestdir+"a/.pig_schema").delete();
+        
+        putSchemaFile(globtestdir+"b/.pig_schema", testSchema);
+        schema = pigStorage.getSchema(globtestdir+"*", new Job(ConfigurationUtil.toConfiguration(pigContext.getProperties())));
+        Assert.assertTrue(ResourceSchema.equals(schema, testSchema));
+        new File(globtestdir+"b/.pig_schema").delete();
+        
+        // if .pig_schema is deep in the globbing, it will not get used
+        putSchemaFile(globtestdir+"a/a0/.pig_schema", testSchema);
+        schema = pigStorage.getSchema(globtestdir+"*", new Job(ConfigurationUtil.toConfiguration(pigContext.getProperties())));
+        Assert.assertTrue(schema==null);
+        putSchemaFile(globtestdir+"a/.pig_schema", testSchema);
+        schema = pigStorage.getSchema(globtestdir+"*", new Job(ConfigurationUtil.toConfiguration(pigContext.getProperties())));
+        Assert.assertTrue(ResourceSchema.equals(schema, testSchema));
+        new File(globtestdir+"a/a0/.pig_schema").delete();
+        new File(globtestdir+"a/.pig_schema").delete();
+    }
 }



Mime
View raw message