You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by pr...@apache.org on 2009/12/21 20:58:06 UTC
svn commit: r892950 [1/2] - in /hadoop/pig/branches/load-store-redesign: ./
contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/
contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/ ivy/
lib/ license/ src/docs/src/docum...
Author: pradeepkth
Date: Mon Dec 21 19:58:03 2009
New Revision: 892950
URL: http://svn.apache.org/viewvc?rev=892950&view=rev
Log:
svn merge -r890596:892943 http://svn.apache.org/repos/asf/hadoop/pig/trunk
(The actual merge was from -r890596:892416 - it is recorded as -r890596:892943 because between r892416
and r892943 the commits in trunk were for PIG-1143 and PIG-1149. PIG-1143 should not be applied to the
branch since the code is not applicable and for PIG-1149 there is a separate patch for the branch which
will be applied separately)
Added:
hadoop/pig/branches/load-store-redesign/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/PigStorageSchema.java
hadoop/pig/branches/load-store-redesign/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/TestPigStorageSchema.java
hadoop/pig/branches/load-store-redesign/lib/automaton.jar
hadoop/pig/branches/load-store-redesign/license/brics-LICENSE.txt
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/experimental/
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/experimental/JsonMetadata.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/experimental/LoadMetadata.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/experimental/ResourceSchema.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/experimental/ResourceStatistics.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/experimental/StoreMetadata.java
hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestBatchAliases.java
Modified:
hadoop/pig/branches/load-store-redesign/CHANGES.txt
hadoop/pig/branches/load-store-redesign/build.xml
hadoop/pig/branches/load-store-redesign/ivy.xml
hadoop/pig/branches/load-store-redesign/ivy/libraries.properties
hadoop/pig/branches/load-store-redesign/ivy/pig.pom
hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/site.xml
hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/udf.xml
hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_mapreduce.xml
hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_overview.xml
hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_pig.xml
hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_reference.xml
hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_stream.xml
hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_users.xml
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/StoreConfig.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/executionengine/ExecJob.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/HJob.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/PhysicalOperator.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/local/executionengine/LocalJob.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/data/DataType.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/io/FileLocalizer.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/logicalLayer/LOCogroup.java
hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/logicalLayer/schema/SchemaUtil.java
hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestDataBagAccess.java
hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestJobSubmission.java
hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestLogicalPlanBuilder.java
hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMultiQuery.java
hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestNullConstant.java
hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestResourceSchema.java
hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestSchemaUtil.java
hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld
Modified: hadoop/pig/branches/load-store-redesign/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/CHANGES.txt?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/CHANGES.txt (original)
+++ hadoop/pig/branches/load-store-redesign/CHANGES.txt Mon Dec 21 19:58:03 2009
@@ -45,6 +45,20 @@
IMPROVEMENTS
+PIG-1162: Pig 0.6.0 - UDF doc (chandec via olgan)
+
+PIG-1163: Pig/Zebra 0.6.0 release (chandec via olgan)
+
+PIG-1156: Add aliases to ExecJobs and PhysicalOperators (dvryaboy via gates)
+
+PIG-1161: add missing license headers (dvryaboy via olgan)
+
+PIG-965: PERFORMANCE: optimize common case in matches (PORegex) (ankit.modi
+via olgan)
+
+PIG-760: Add a new PigStorageSchema load/store function that
+ store schemas for text files (dvryaboy via gates)
+
PIG-1106: FR join should not spill (ankit.modi via olgan)
PIG-1147: Zebra Docs for Pig 0.6.0 (chandec via olgan)
@@ -72,6 +86,11 @@
BUG FIXES
+PIG-1157: Sucessive replicated joins do not generate Map Reduce plan and fails
+due to OOM (rding via olgan)
+
+PIG-1075: Error in Cogroup when key fields types don't match (rding via olgan)
+
PIG-973: type resolution inconsistency (rding via olgan)
PIG-1135: skewed join partitioner returns negative partition index (yinghe
@@ -302,6 +321,11 @@
PIG-1142: Got NullPointerException merge join with pruning (daijy)
+PIG-1155: Need to make sure existing loaders work "as is" (daijy)
+
+PIG-1144: set default_parallelism construct does not set the number of
+reducers correctly (daijy)
+
Release 0.5.0
INCOMPATIBLE CHANGES
Modified: hadoop/pig/branches/load-store-redesign/build.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/build.xml?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/build.xml (original)
+++ hadoop/pig/branches/load-store-redesign/build.xml Mon Dec 21 19:58:03 2009
@@ -168,7 +168,10 @@
<fileset file="${lib.dir}/${hbase.jarfile}" />
<fileset file="${lib.dir}/${hbase.test.jarfile}" />
<fileset file="${lib.dir}/${zookeeper.jarfile}"/>
- <!-- <fileset file="${lib.dir}/commons-collections-3.2.jar" /> -->
+ <fileset file="${ivy.lib.dir}/jackson-mapper-asl-${jackson.version}.jar"/>
+ <fileset file="${ivy.lib.dir}/jackson-core-asl-${jackson.version}.jar"/>
+
+ <!-- <fileset file="${lib.dir}/commons-collections-3.2.jar" /> -->
</path>
<!-- javadoc-classpath -->
@@ -427,6 +430,8 @@
<zipfileset src="${ivy.lib.dir}/junit-${junit.version}.jar" />
<zipfileset src="${ivy.lib.dir}/jsch-${jsch.version}.jar" />
<zipfileset src="${ivy.lib.dir}/jline-${jline.version}.jar" />
+ <zipfileset src="${ivy.lib.dir}/jackson-mapper-asl-${jackson.version}.jar" />
+ <zipfileset src="${ivy.lib.dir}/jackson-core-asl-${jackson.version}.jar" />
</jar>
<copy file="${output.jarfile}" tofile="${output.jarfile.backcompat}"/>
</target>
@@ -480,6 +485,8 @@
<zipfileset src="${ivy.lib.dir}/junit-${junit.version}.jar" />
<zipfileset src="${ivy.lib.dir}/jsch-${jsch.version}.jar" />
<zipfileset src="${ivy.lib.dir}/jline-${jline.version}.jar" />
+ <zipfileset src="${ivy.lib.dir}/jackson-mapper-asl-${jackson.version}.jar" />
+ <zipfileset src="${ivy.lib.dir}/jackson-core-asl-${jackson.version}.jar" />
</jar>
<copy file="${output.jarfile.withouthadoop}" tofile="${output.jarfile.backcompat.withouthadoop}"/>
</target>
Added: hadoop/pig/branches/load-store-redesign/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/PigStorageSchema.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/PigStorageSchema.java?rev=892950&view=auto
==============================================================================
--- hadoop/pig/branches/load-store-redesign/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/PigStorageSchema.java (added)
+++ hadoop/pig/branches/load-store-redesign/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/PigStorageSchema.java Mon Dec 21 19:58:03 2009
@@ -0,0 +1,125 @@
+/*
+ * 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.pig.piggybank.storage;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.pig.ExecType;
+import org.apache.pig.experimental.JsonMetadata;
+import org.apache.pig.experimental.LoadMetadata;
+import org.apache.pig.experimental.StoreMetadata;
+import org.apache.pig.experimental.ResourceSchema;
+import org.apache.pig.experimental.ResourceStatistics;
+import org.apache.pig.StoreConfig;
+import org.apache.pig.backend.datastorage.DataStorage;
+import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
+import org.apache.pig.backend.hadoop.datastorage.HDataStorage;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
+import org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil;
+import org.apache.pig.builtin.PigStorage;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.io.FileLocalizer;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+
+/**
+ * This Load/Store Func reads/writes metafiles that allow the schema and
+ * aliases to be determined at load time, saving one from having to manually
+ * enter schemas for pig-generated datasets.
+ *
+ * It also creates a ".pig_headers" file that simply lists the delimited aliases.
+ * This is intended to make export to tools that can read files with header
+ * lines easier (just cat the header to your data).
+ *
+ * Due to StoreFunc limitations, you can only write the metafiles in MapReduce
+ * mode. You can read them in Local or MapReduce mode.
+ */
+public class PigStorageSchema extends PigStorage implements StoreMetadata {
+
+ private static final Log log = LogFactory.getLog(PigStorageSchema.class);
+
+ public PigStorageSchema() {
+ super();
+ }
+
+ public PigStorageSchema(String delim) {
+ super(delim);
+ }
+
+ @Override
+ public Schema determineSchema(String fileName, ExecType execType,
+ DataStorage storage) throws IOException {
+
+ // TODO fullPath should be retrieved ia relativeToAbsolutePath once PIG-966 is complete
+ String fullPath = FileLocalizer.fullPath(fileName, storage);
+ LoadMetadata metadataLoader = new JsonMetadata(fullPath, storage);
+ ResourceSchema resourceSchema = metadataLoader.getSchema(fullPath, null);
+ if (resourceSchema == null) {
+ return null;
+ }
+ Schema pigSchema = new Schema();
+ for (ResourceSchema.ResourceFieldSchema field : resourceSchema.getFields()) {
+ FieldSchema pigFieldSchema = DataType.determineFieldSchema(field);
+ // determineFieldSchema only sets the types. we also want the aliases.
+ // TODO this doesn't work properly for complex types
+ pigFieldSchema.alias = field.getName();
+ pigSchema.add(pigFieldSchema);
+ }
+ log.info("Loaded Schema: "+pigSchema);
+ return pigSchema;
+ }
+
+ @Override
+ public void finish() throws IOException {
+ super.finish();
+ JobConf jobConf = PigMapReduce.sJobConf;
+ if(jobConf != null){
+ StoreConfig storeConfig = MapRedUtil.getStoreConfig(jobConf);
+ DataStorage store = new HDataStorage(ConfigurationUtil.toProperties(jobConf));
+ Schema schema = storeConfig.getSchema();
+ ResourceSchema resourceSchema = new ResourceSchema(schema);
+ JsonMetadata metadataWriter = new JsonMetadata(storeConfig.getLocation(), store);
+ metadataWriter.setFieldDel(fieldDel);
+ metadataWriter.setRecordDel(recordDel);
+ metadataWriter.setSchema(resourceSchema, storeConfig.getLocation(), null);
+ }
+ }
+
+ /**
+ * @see org.apache.pig.experimental.StoreMetadata#setSchema(ResourceSchema)
+ * Does not do anything in this implementation. The finish() call writes the schema.
+ */
+ @Override
+ public void setSchema(ResourceSchema schema, String location, Configuration conf) throws IOException {
+ // n\a
+ }
+
+ /**
+ * @see org.apache.pig.experimental.StoreMetadata#setStatistics(ResourceStatistics)
+ * Does not do anything in this implementation.
+ */
+ @Override
+ public void setStatistics(ResourceStatistics stats, String location, Configuration conf) throws IOException {
+ // n\a
+ }
+}
Added: hadoop/pig/branches/load-store-redesign/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/TestPigStorageSchema.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/TestPigStorageSchema.java?rev=892950&view=auto
==============================================================================
--- hadoop/pig/branches/load-store-redesign/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/TestPigStorageSchema.java (added)
+++ hadoop/pig/branches/load-store-redesign/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/TestPigStorageSchema.java Mon Dec 21 19:58:03 2009
@@ -0,0 +1,111 @@
+/*
+ * 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.pig.piggybank.test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.io.FileLocalizer;
+import org.apache.pig.impl.logicalLayer.LogicalOperator;
+import org.apache.pig.impl.logicalLayer.LogicalPlan;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.test.MiniCluster;
+import org.apache.pig.test.Util;
+import org.apache.pig.test.utils.TypeCheckingTestUtil;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import junit.framework.TestCase;
+
+public class TestPigStorageSchema extends TestCase {
+
+ protected ExecType execType = ExecType.MAPREDUCE;
+
+ PigContext pigContext = new PigContext(ExecType.MAPREDUCE, new Properties());
+ Map<LogicalOperator, LogicalPlan> aliases = new HashMap<LogicalOperator, LogicalPlan>();
+ Map<OperatorKey, LogicalOperator> logicalOpTable = new HashMap<OperatorKey, LogicalOperator>();
+ Map<String, LogicalOperator> aliasOp = new HashMap<String, LogicalOperator>();
+ Map<String, String> fileNameMap = new HashMap<String, String>();
+
+ MiniCluster cluster = MiniCluster.buildCluster();
+
+ private PigServer pig;
+
+ @Before
+ @Override
+ protected void setUp() throws Exception {
+ super.setUp();
+ pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ String origPath = FileLocalizer.fullPath("originput", pig.getPigContext());
+ if (FileLocalizer.fileExists(origPath, pig.getPigContext())) {
+ FileLocalizer.delete(origPath, pig.getPigContext());
+ }
+ Util.createInputFile(cluster, "originput",
+ new String[] {"A,1", "B,2", "C,3", "D,2",
+ "A,5", "B,5", "C,8", "A,8",
+ "D,8", "A,9"});
+
+ }
+
+ @After
+ @Override
+ protected void tearDown() throws Exception {
+ Util.deleteFile(cluster, "originput");
+ String aoutPath = FileLocalizer.fullPath("aout", pig.getPigContext());
+ if (FileLocalizer.fileExists(aoutPath, pig.getPigContext())) {
+ FileLocalizer.delete(aoutPath, pig.getPigContext());
+ }
+ }
+
+ @Test
+ public void testPigStorageSchema() throws Exception {
+ pigContext.connect();
+ String query = "a = LOAD 'originput' using org.apache.pig.piggybank.storage.PigStorageSchema() as (f1:chararray, f2:int);";
+ pig.registerQuery(query);
+ Schema origSchema = pig.dumpSchema("a");
+ pig.registerQuery("STORE a into 'aout' using org.apache.pig.piggybank.storage.PigStorageSchema();");
+
+ // aout now has a schema.
+
+ // Verify that loading a-out with no given schema produces
+ // the original schema.
+
+ pig.registerQuery("b = LOAD 'aout' using org.apache.pig.piggybank.storage.PigStorageSchema();");
+ Schema genSchema = pig.dumpSchema("b");
+ assertTrue("generated schema equals original" , Schema.equals(genSchema, origSchema, true, false));
+
+ // Verify that giving our own schema works
+ String [] aliases ={"foo", "bar"};
+ byte[] types = {DataType.INTEGER, DataType.LONG};
+ Schema newSchema = TypeCheckingTestUtil.genFlatSchema(
+ aliases,types);
+ pig.registerQuery("c = LOAD 'aout' using org.apache.pig.piggybank.storage.PigStorageSchema() as (foo:int, bar:long);");
+ Schema newGenSchema = pig.dumpSchema("c");
+ assertTrue("explicit schema overrides metadata", Schema.equals(newSchema, newGenSchema, true, false));
+
+ }
+}
Modified: hadoop/pig/branches/load-store-redesign/ivy.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/ivy.xml?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/ivy.xml (original)
+++ hadoop/pig/branches/load-store-redesign/ivy.xml Mon Dec 21 19:58:03 2009
@@ -67,6 +67,10 @@
conf="test->default"/>
<dependency org="com.google.code.p.arat" name="rat-lib" rev="${rats-lib.version}"
conf="releaseaudit->default"/>
+ <dependency org="org.codehaus.jackson" name="jackson-mapper-asl" rev="${jackson.version}"
+ conf="compile->master"/>
+ <dependency org="org.codehaus.jackson" name="jackson-core-asl" rev="${jackson.version}"
+ conf="compile->master"/>
<!--ATM hbase, hbase-test.jar, hadoop.jar are resolved from the lib dir-->
</dependencies>
</ivy-module>
Modified: hadoop/pig/branches/load-store-redesign/ivy/libraries.properties
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/ivy/libraries.properties?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/ivy/libraries.properties (original)
+++ hadoop/pig/branches/load-store-redesign/ivy/libraries.properties Mon Dec 21 19:58:03 2009
@@ -33,3 +33,5 @@
rats-lib.version=0.5.1
xerces.version=1.4.4
+
+jackson.version=1.0.1
Modified: hadoop/pig/branches/load-store-redesign/ivy/pig.pom
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/ivy/pig.pom?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/ivy/pig.pom (original)
+++ hadoop/pig/branches/load-store-redesign/ivy/pig.pom Mon Dec 21 19:58:03 2009
@@ -83,5 +83,11 @@
<version>${junit.version}</version>
</dependency>
+ <dependency>
+ <groupId>jackson</groupId>
+ <artifactId>jackson</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+
</dependencies>
</project>
Added: hadoop/pig/branches/load-store-redesign/lib/automaton.jar
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/lib/automaton.jar?rev=892950&view=auto
==============================================================================
Files hadoop/pig/branches/load-store-redesign/lib/automaton.jar (added) and hadoop/pig/branches/load-store-redesign/lib/automaton.jar Mon Dec 21 19:58:03 2009 differ
Added: hadoop/pig/branches/load-store-redesign/license/brics-LICENSE.txt
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/license/brics-LICENSE.txt?rev=892950&view=auto
==============================================================================
--- hadoop/pig/branches/load-store-redesign/license/brics-LICENSE.txt (added)
+++ hadoop/pig/branches/load-store-redesign/license/brics-LICENSE.txt Mon Dec 21 19:58:03 2009
@@ -0,0 +1,2 @@
+website: http://www.brics.dk/automaton/
+link to the license from the wewsite above: http://www.opensource.org/licenses/bsd-license.php
Modified: hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/site.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/site.xml?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/site.xml (original)
+++ hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/site.xml Mon Dec 21 19:58:03 2009
@@ -45,24 +45,24 @@
<tutorial label="Tutorial" href="tutorial.html" />
</docs>
<docs label="Guides">
- <piglatin label="Pig Latin Users " href="piglatin_users.html" />
- <piglatin label="Pig Latin Reference" href="piglatin_reference.html" />
+ <plusers label="Pig Latin Users " href="piglatin_users.html" />
+ <plref label="Pig Latin Reference" href="piglatin_reference.html" />
<cookbook label="Cookbook" href="cookbook.html" />
<udf label="UDFs" href="udf.html" />
</docs>
<docs label="Zebra">
- <piglatin label="Zebra Overview " href="zebra_overview.html" />
- <piglatin label="Zebra Users " href="zebra_users.html" />
- <piglatin label="Zebra Reference " href="zebra_reference.html" />
- <piglatin label="Zebra MapReduce " href="zebra_mapreduce.html" />
- <piglatin label="Zebra Pig " href="zebra_pig.html" />
- <piglatin label="Zebra Streaming " href="zebra_stream.html" />
+ <zover label="Zebra Overview " href="zebra_overview.html" />
+ <zusers label="Zebra Users " href="zebra_users.html" />
+ <zref label="Zebra Reference " href="zebra_reference.html" />
+ <zmr label="Zebra MapReduce " href="zebra_mapreduce.html" />
+ <zpig label="Zebra Pig " href="zebra_pig.html" />
+ <zstream label="Zebra Streaming " href="zebra_stream.html" />
</docs>
<docs label="Miscellaneous">
<api label="API Docs" href="ext:api"/>
- <wiki label="Wiki" href="ext:wiki" />
- <faq label="FAQ" href="ext:faq" />
- <relnotes label="Release Notes" href="ext:relnotes" />
+ <wiki label="Wiki" href="ext:wiki" />
+ <faq label="FAQ" href="ext:faq" />
+ <relnotes label="Release Notes" href="ext:relnotes" />
</docs>
<external-refs>
Modified: hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/udf.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/udf.xml?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/udf.xml (original)
+++ hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/udf.xml Mon Dec 21 19:58:03 2009
@@ -731,6 +731,7 @@
command:
pig -cp sds.jar -Dudf.import.list=com.yahoo.yst.sds.ULT myscript.pig
</source>
+
</section>
<section>
@@ -885,8 +886,9 @@
<section>
<title>Builtin Functions and Function Repositories</title>
-<p>Pig comes with a set of built-in in functions. Two main properties differentiate builtin functions from UDFs. First, they don't need to be registered because Pig knows where they are. Second, they don't need to be qualified when used because Pig knows where to find them. </p>
-<p>In addition to builtins, Pig hosts a UDF repository called <code>piggybank</code> that allows users to share UDFs that they have written. The details are described in <a href="http://wiki.apache.org/pig/PiggyBank"> PiggyBank</a>. </p>
+<p>Pig comes with a set of builtin functions. Two main properties differentiate builtin functions from UDFs. First, they don't need to be registered because Pig knows where they are. Second, they don't need to be qualified when used because Pig knows where to find them. </p>
+
+<p>Pig also hosts a UDF repository called <code>piggybank</code> that allows users to share UDFs that they have written. The details are described in <a href="http://wiki.apache.org/pig/PiggyBank"> PiggyBank</a>. </p>
</section>
@@ -1110,7 +1112,12 @@
LOAD '27' USING RangeSlicer();
</source>
-</section></section></section>
+</section>
+
+</section>
+
+
+</section>
</body>
</document>
Modified: hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_mapreduce.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_mapreduce.xml?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_mapreduce.xml (original)
+++ hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_mapreduce.xml Mon Dec 21 19:58:03 2009
@@ -149,6 +149,7 @@
throw new RuntimeException(e);
}
}
+
}
static class ProjectionMap extends MapReduceBase implements Mapper<BytesWritable, Tuple, Text, IntWritable> {
@@ -209,21 +210,22 @@
jobConf.setOutputFormat(BasicTableOutputFormat.class);
BasicTableOutputFormat.setOutputPath(jobConf, new Path(args[1]));
- // set the output info:
- // ZebraSchema - 2 logical columns
- // ZebraStorageHint - 2 physical column groups (one column each)
- // ZebraSortInfo - unsorted table (null)
+ // set the storage info of logical schema with 2 columns;
+ // and create 2 physical column groups;
+ // unsorted table
BasicTableOutputFormat.setStorageInfo(jobConf,
ZebraSchema.createZebraSchema("word:string, count:int"),
- ZebraStorageHint.createZebraStorageHint("[word];[count]"),
- null);
+ ZebraStorageHint.createZebraStorageHint("[word];[count]"), null);
// set map-only job.
jobConf.setNumReduceTasks(0);
// Run Job
JobClient.runJob(jobConf);
+
+ // Need to close Zebra output streams
+ BasicTableOutputFormat.close(jobConf);
/*
Second MR Job for Table Projection of count column
@@ -264,7 +266,13 @@
<!-- ZEBRA OUTPUT EXAMPLE-->
<section>
<title>Table Input/Output Formats</title>
-<p>This MapReduce example demonstrates the Zebra table input/output formats. </p>
+<p>
+This MapReduce examples demonstrates how to perform a simple union.
+To run this program, we need two basic tables that contain
+the data as in the example above (word, count). In this example they are:
+/user/mapredu/t1 and /user/mapredu/t2. The resulting table is /user/mapredu2/t.
+</p>
+
<source>
package org.apache.hadoop.zebra.mapred;
@@ -286,13 +294,7 @@
import org.apache.hadoop.zebra.types.TypesUtils;
import org.apache.pig.data.Tuple;
-/**
- * This is a sample to show using zebra table to do a simple basic union in
- * map/reduce * To run this, we need have two basic tables ready. They contain
- * the data as in Sample 1, i.e., (word, count). In this example, they are at:
- * /user/mapredu/t1 /user/mapredu/t2 The resulting table is put at: /user/mapredu2/t1
- *
- */
+
public class TableMRSample2 {
static class MapClass implements
Mapper<BytesWritable, Tuple, BytesWritable, Tuple> {
Modified: hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_overview.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_overview.xml?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_overview.xml (original)
+++ hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_overview.xml Mon Dec 21 19:58:03 2009
@@ -42,8 +42,8 @@
<title>Prerequisites</title>
<p>Zebra requires:</p>
<ul>
- <li>Pig 0.6.0 </li>
- <li>Hadoop 0.20.1</li>
+ <li>Pig 0.6.0 or later</li>
+ <li>Hadoop 0.20.1 or later</li>
</ul>
<p></p>
<p>Also, make sure the following software is installed on your system:</p>
Modified: hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_pig.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_pig.xml?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_pig.xml (original)
+++ hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_pig.xml Mon Dec 21 19:58:03 2009
@@ -73,7 +73,7 @@
A = LOAD 'studenttab, votertab' USING org.apache.hadoop.zebra.pig.TableLoader();
</source>
-<p>TableLoader supports efficient column selection; projections are automatically push down to the loader. This example tells the loader to only return two columns, name and age.</p>
+<p>TableLoader supports efficient column selection; projections are automatically pushed down to the loader. This example tells the loader to only return two columns, name and age.</p>
<source>
A = LOAD 'studenttab' USING org.apache.hadoop.zebra.pig.TableLoader('name, age');
</source>
@@ -84,7 +84,7 @@
<section>
<title>Map-Side Group and Merge Join</title>
-<p>If the input data is globally sorted, map-side group or merge join can be used. Please, notice the âsortedâ argument passed to the loader. This lets the loader know that the data is expected to be globally sorted and that a single key must be given to the same map.</p>
+<p>If the input data is globally sorted, merge join and map-side group can be used. Please note the âsortedâ argument that is passed to the loader. This lets the loader know that the data is expected to be globally sorted and that a single key must be given to the same map.</p>
<p>Here is an example of the merge join. Note that the first argument to the loader is left empty to indicate that all columns are requested.</p>
<source>
@@ -93,7 +93,7 @@
G = JOIN A BY $0, B By $0 USING "merge";
</source>
-<p>Here is an example of a map-side group. Note that multiple sorted files are passed to the loader and that the loader will perform sort preserving merge to make sure that the data is globally sorted.</p>
+<p>Here is an example of a map-side group. Note that multiple sorted files are passed to the loader and that the loader will perform sort-preserving merge to make sure that the data is globally sorted.</p>
<source>
A = LOAD 'studentsortedtab, studentnullsortedtab' using org.apache.hadoop.zebra.pig.TableLoader('name, age, gpa, source_table', 'sorted');
B = GROUP A BY $0 USING "collected";
Modified: hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_reference.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_reference.xml?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_reference.xml (original)
+++ hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_reference.xml Mon Dec 21 19:58:03 2009
@@ -150,7 +150,7 @@
(<em>In a future release, the schema will also support type compatibility between Zebra/Pig-SQL and will guide the underlying serialization formats provided by Avro for projection, filtering, and so on. </em>)
</p>
- <p>The basic format for the the store schema is shown here.
+ <p>The basic format for the store schema is shown here.
The type name is optional; if not specified, the column defaults to type bytes.</p>
<p>
<code>
@@ -453,7 +453,7 @@
<p>The Zebra load schema is load or read table columns. </p>
<section>
<title>Schema</title>
- <p>The basic format for the the Zebra load (read) schema is shown here. The column name can be any valid Zebra type.
+ <p>The basic format for the Zebra load (read) schema is shown here. The column name can be any valid Zebra type.
If no columns are specified, the entire Zebra table is loaded.</p>
<p>
<code>
Modified: hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_stream.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_stream.xml?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_stream.xml (original)
+++ hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_stream.xml Mon Dec 21 19:58:03 2009
@@ -105,7 +105,7 @@
<section>
<title>Locating Frequently Visited Pages</title>
- <p>This perl script sorts the pages on number of page view counts. The script outputs space padded count
+ <p>This Perl script sorts the pages on number of page view counts. The script outputs space padded count
so that string sorting results in correct output. The first TAB separates the key and value for Hadoop streaming.</p>
<source>
Modified: hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_users.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_users.xml?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_users.xml (original)
+++ hadoop/pig/branches/load-store-redesign/src/docs/src/documentation/content/xdocs/zebra_users.xml Mon Dec 21 19:58:03 2009
@@ -32,31 +32,24 @@
<title>Column Security</title>
<p><strong>NOTE: THIS FEATURE IS EXPERIMENTAL AND SUBJECT TO CHANGE IN THE FUTURE</strong></p>
- <p>Since Zebra provides columnar storage of user data, we intend to separate secure and non-secure data into separate columns. We can then have access control based on HDFS file systesm based security. This would be achieved by an administrator setting appropriate permissions on the HDFS files contaning secure data. </p>
+ <p>Since Zebra supports reading and writing data in a column-oriented fashion, you can store secure and non-secure data in separate columns.
+ Then, using the HDFS file system, you can enable access control by setting the appropriate permissions on the columns containing secure data.
+ </p>
- <section>
- <title>Design Issues</title>
-<p>Roles:</p>
-<ul>
-<li>Publishers of secure data </li>
-<li>Consumers of secure data </li>
-<li>Administrators of secure data </li>
+ <p>About the data:</p>
+ <ul>
+<li>All the files and directories containing secure data will have the same permissions and groups within a table. </li>
+<li>If no security information is provided, then the HDFS file system default behavior applies.</li>
</ul>
-
- <p>How it will work:</p>
+<p></p>
+ <p>About the users:</p>
<ul>
+<li>The user who creates the data will set the file permissions.</li>
+<li>If a permissions-related error happens, it will be communicated to the user as a normal IO exception. </li>
+<li>A user running a client application needs to have chgrp permissions to execute the "secure by group" operations on a table. </li>
+<li>If a user running a client application does not have read permissions for a secure column group, an IO exception is issued.</li>
-<li>Before any data is written in tables, all the corresponding files and directories need to have right set of ownership and permissions.
-This is necessary because if all the data is written and then the operation to make it secure is executed, that can leave security holes and paranoid will not allow that.</li>
-
-<li>All the files and direcories containing secure data will have same permissions and groups within a table </li>
-<li>User of the MR Job/Pig Script is required to have permissions to execute chgrp operations on a table. </li>
-<li>If no security information provided, then default behaviour.</li>
-<li>If permissions related error happens, it will be communicated to user as normal IOException </li>
-<li>If reader does not have read permissions for a column (CG), an IOException will be thrown </li>
-<li>The publisher/creator of the files will set these permissions when creating data.</li>
</ul>
- </section>
<p></p>
<p>One simple Pig example:</p>
@@ -70,7 +63,7 @@
zStorageHint = ZebraStorageHint.createZebraStorageHint(â[a, b] secure by group:secure perm:640â);
zSchema = â¦;
zSortInfo = â¦;
-setStorageInfo(jobConf, zSchema, zStorageHint, zSortInf);
+setStorageInfo(jobConf, zSchema, zStorageHint, zSortInfo);
</source>
</section>
<!-- END COLUMN SECURITY -->
@@ -80,20 +73,20 @@
<title>Drop Column Groups</title>
<p><strong>NOTE: THIS FEATURE IS EXPERIMENTAL AND SUBJECT TO CHANGE IN THE FUTURE</strong></p>
- <p>Zebra allows you to delete a column group using the column group name.
+ <p>Zebra allows you to delete a column group (CG) using the column group name.
For examples, see <a href="zebra_mapreduce.html#Drop+Column+Groups">Drop Column Groups</a>. </p>
<p>Please note the following:</p>
<ul>
<li>Any failures during a drop will leave the table in consistent state (either with or with out the column group).
-While success of a CG removal guarantees a column removal, a failure does not imply CG is not removed.
+While success of a column group removal guarantees a column removal, a failure does not imply the column group is not removed.
In rare cases, you might receive an error but the column could still be deleted. </li>
<li>MapReduce jobs and other clients that are currently accessing the table might fail with exceptions.
-It is recommended that the column groups are dropped when there are no accesses to a table.
+It is recommended that column groups be dropped when there are no accesses to a table.
It might not be feasible to ensure that there are no readers for a table; in these cases the readers should handle the exception. </li>
-<li>Once a column group is dropped, the column gruop data is deleted from the underlying filesystem.
-In the case of HDFS, it may not imply that physical data is actually removed because of earlier snapshot of the filesystem; handling this is out side the scope of Zebra. Legal requirements might require an admin finalize HDFS (if it is not already finalized) before or after performing a deletion. </li>
-<li>Concurrent deletions are supported and their access is serialized. </li>
+<li>Once a column group is dropped, the column group data is deleted from the underlying file system.
+In the case of the HDFS filesystem, it may not imply that physical data is actually removed because of earlier snapshot of the file system; handling this is out side the scope of Zebra.</li>
+<li>Concurrent column group deletions are supported and their access is serialized. </li>
<li>Deleting a non-existant column group or a column group that is already deleted is not allowed.</li>
<li>If you delete all the remaining columns in a table, it logically leaves an empty null table. The difference between a non-existant table and a table with zero columns is that opening a non-existant table causes an error. </li>
</ul>
@@ -104,19 +97,13 @@
<section>
<title>Order-Preserving Sorted Table Union</title>
<p>
-This Zebra functionality is only available on underlying sorted Zebra tables.
-</p>
-
-<section>
-<title>Output Records</title>
-<p>
-This feature groups all records from all "delta tables" on some sort key to form an output set of records while preserving the sorted ordering of the records in the origional tables. For instance, if the client application wants to fetch records from a union of tables of T1, T2 on a column "c1", then all records from T1 with a particular value of column "c1" and all records from T2 with that value of column "c1" will be output. The ordering of the rows of the output set of the same value of column "c1" is undefined. As a prerequisite, both T1 and T2 must be sorted on column "c1". More specifically the input and results could be as follows:
+With Zebra you can group all records from all "delta tables" on some sort key to form an output set of records while preserving the sorted ordering of the records in the original tables. For instance, if the client application wants to fetch records from a union of tables of T1, T2 on a column "C1", then all records from T1 with a particular value of column "c1" and all records from T2 with that value of column "C1" will be output. The ordering of the rows of the output set of the same value of column "C1" is undefined. As a prerequisite, both T1 and T2 must be sorted on column "C1". More specifically the input and results could be as follows:
</p>
<p>Table T1: </p>
<source>
C1 C2
---------------
+-------------
A 11
A 12
B 21
@@ -137,7 +124,7 @@
<p>T1 Sort-Unioned with T2: </p>
<source>
source_table C1 C2
----------------------------------------
+------------------------------
0 A 11
1 A 101
0 A 12
@@ -148,27 +135,27 @@
1 D 401
0 D 41
</source>
-</section>
+
<p>
-Note that the sortness is guaranteed per mapper and among all mappers arranged with certain ordering, but not among mappers arranged in any ordering. For instance, the outputs generated by 4 mappers, m1, m2, m3 and m4, could be in total ordering between m1, m3, m2 and m4, but not in any other arrangements.
+Note that the sortness is guaranteed per mapper and among all mappers arranged with certain ordering, but not among mappers arranged in any ordering. For instance, the outputs generated by four mappers, m1, m2, m3 and m4, could be in total ordering between m1, m3, m2 and m4, but not in any other arrangements.
</p>
<section>
-<title>Grouping and Indexing of Sort-Unioned Output Rows </title>
+<title>Indexing Sort-Unioned Results</title>
<p>
-The order preserving sort-unioned results above could be further indexed by the component tables if the projection contains column(s) named "source_table". If so specified, the component table index will be output at the position(s) as specified in the projection list. If the underlying table is not a union of sorted tables, use of the special column name in projection will cause an exception thrown.
-</p>
+The order-preserving sort-unioned results above can be further indexed by the component tables if the projection contains column(s) named "source_table". If so specified, the component table index is output at the position(s) as specified in the projection list.
-<p>
-If an attempt is made to create a table of a column named "source_table", an excpetion will be thrown as the name is reserved by zebra for the virtual name.
+If the underlying table is not a union of sorted tables, the use of the special column name in a projection will cause an exception.
+
+If an attempt is made to create a table of a column named "source_table", an exception will be thrown as the name is reserved by zebra for the virtual name.
</p>
</section>
<section>
-<title>MapReduce Interface </title>
+<title>MapReduce Jobs</title>
<p>
-TableInputFormat will have a static method, requireSortedTable, that allows the caller to specify the behavior of a single sorted table or an order preserving sorted table union as described above. The method will ensure all tables in a union are sorted. For more information, see <a href="zebra_reference.html#TableInputFormat">TableInputFormat</a>.
+TableInputFormat has static method, requireSortedTable, that allows the caller to specify the behavior of a single sorted table or an order-preserving sorted table union as described above. The method ensures all tables in a union are sorted. For more information, see <a href="zebra_reference.html#TableInputFormat">TableInputFormat</a>.
</p>
<p>One simple example: A order-preserving sorted union B. A and B are sorted tables. </p>
@@ -182,8 +169,8 @@
</section>
<section>
-<title>Pig Interface </title>
-<p>Pig will take an extra string argument of "sorted" indicating the desire to load from a sorted table or an order preserving sorted table union.
+<title>Pig Scripts</title>
+<p>Pig takes an extra string argument of "sorted" indicating the desire to load from a sorted table or an order-preserving sorted table union.
For more information, see <a href="zebra_pig.html#Zebra+Pig+Examples">Zebra Pig Examples</a>.</p>
<p>One simple example:</p>
@@ -193,8 +180,7 @@
...
</source>
</section>
-
-</section>
+ </section>
<!-- END ORDER PRESERVE SORT-->
<!--MERGE JOIN-->
@@ -206,23 +192,22 @@
<p>One simple example:</p>
<source>
-Class myMapper {
-â¦
-Object keyGenerator;
-â¦
-
-public void map(â¦) {
- bytesKey = BasicTableOutputFormat.getSortKey(keyGenerator, userKey);
- â¦
- output.collect(bytesKey, valueTuple);
- â¦
+class myMapper extends Mapper<â¦> {
+ â¦
+ Object keyGenerator;
+ â¦
+ public void map(â¦) {
+ bytesKey = BasicTableOutputFormat.getSortKey(keyGenerator, userKey);
+ â¦
+ output.collect(bytesKey, valueTuple);
+ â¦
+ }
+ public void configure(JobConf job) {
+ keyGenerator = BasicTableOutputFormat.getSortKeyGenerator(job);
+ â¦
+ }
}
-public void configure(JobConf job)
-{
- keyGenerator = BasicTableOutputFormat.getSortKeyGenerator(job);
-â¦
-}
</source>
</section>
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/StoreConfig.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/StoreConfig.java?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/StoreConfig.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/StoreConfig.java Mon Dec 21 19:58:03 2009
@@ -1,6 +1,21 @@
-/**
- *
+/*
+ * 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.pig;
import java.io.Serializable;
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/executionengine/ExecJob.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/executionengine/ExecJob.java?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/executionengine/ExecJob.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/executionengine/ExecJob.java Mon Dec 21 19:58:03 2009
@@ -66,6 +66,11 @@
public Iterator<Tuple> getResults() throws ExecException;
/**
+ * Returns the alias of relation generated by this job
+ */
+ public String getAlias() throws ExecException;
+
+ /**
* Get configuration information
*
* @return configuration information for the execution engine
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java Mon Dec 21 19:58:03 2009
@@ -51,9 +51,11 @@
import org.apache.pig.impl.io.FileSpec;
import org.apache.pig.impl.logicalLayer.LogicalPlan;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.LogToPhyTranslationVisitor;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.tools.pigstats.PigStats;
@@ -238,15 +240,26 @@
MapReduceLauncher launcher = new MapReduceLauncher();
List<ExecJob> jobs = new ArrayList<ExecJob>();
+ Map<String, PhysicalOperator> leafMap = new HashMap<String, PhysicalOperator>();
+ for (PhysicalOperator physOp : plan.getLeaves()) {
+ log.info(physOp);
+ if (physOp instanceof POStore) {
+ FileSpec spec = ((POStore) physOp).getSFile();
+ if (spec != null)
+ leafMap.put(spec.toString(), physOp);
+ }
+ }
try {
PigStats stats = launcher.launchPig(plan, jobName, pigContext);
for (FileSpec spec: launcher.getSucceededFiles()) {
- jobs.add(new HJob(ExecJob.JOB_STATUS.COMPLETED, pigContext, spec, stats));
+ String alias = leafMap.containsKey(spec.toString()) ? leafMap.get(spec.toString()).getAlias() : null;
+ jobs.add(new HJob(ExecJob.JOB_STATUS.COMPLETED, pigContext, spec, alias, stats));
}
for (FileSpec spec: launcher.getFailedFiles()) {
- HJob j = new HJob(ExecJob.JOB_STATUS.FAILED, pigContext, spec, stats);
+ String alias = leafMap.containsKey(spec.toString()) ? leafMap.get(spec.toString()).getAlias() : null;
+ HJob j = new HJob(ExecJob.JOB_STATUS.FAILED, pigContext, spec, alias, stats);
j.setException(launcher.getError(spec));
jobs.add(j);
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/HJob.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/HJob.java?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/HJob.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/HJob.java Mon Dec 21 19:58:03 2009
@@ -49,23 +49,28 @@
protected PigContext pigContext;
protected FileSpec outFileSpec;
protected Exception backendException;
+ protected String alias;
private PigStats stats;
public HJob(JOB_STATUS status,
PigContext pigContext,
- FileSpec outFileSpec) {
+ FileSpec outFileSpec,
+ String alias) {
this.status = status;
this.pigContext = pigContext;
this.outFileSpec = outFileSpec;
+ this.alias = alias;
}
public HJob(JOB_STATUS status,
PigContext pigContext,
FileSpec outFileSpec,
+ String alias,
PigStats stats) {
this.status = status;
this.pigContext = pigContext;
this.outFileSpec = outFileSpec;
+ this.alias = alias;
this.stats = stats;
}
@@ -175,4 +180,9 @@
public Exception getException() {
return backendException;
}
+
+ @Override
+ public String getAlias() throws ExecException {
+ return alias;
+ }
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java Mon Dec 21 19:58:03 2009
@@ -31,6 +31,7 @@
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapred.JobConf;
+import org.apache.pig.ExecType;
import org.apache.pig.FuncSpec;
import org.apache.pig.IndexableLoadFunc;
import org.apache.pig.LoadFunc;
@@ -2054,10 +2055,13 @@
int val = rp;
if(val<=0){
ExecutionEngine eng = pigContext.getExecutionEngine();
- if(eng instanceof HExecutionEngine){
+ if(pigContext.getExecType() != ExecType.LOCAL){
try {
- val = ((JobConf)((HExecutionEngine)eng).getJobClient().getConf()).getNumReduceTasks();
if(val<=0)
+ val = pigContext.defaultParallel;
+ if (val<=0)
+ val = ((JobConf)((HExecutionEngine)eng).getJobClient().getConf()).getNumReduceTasks();
+ if (val<=0)
val = 1;
} catch (Exception e) {
int errCode = 6015;
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java Mon Dec 21 19:58:03 2009
@@ -89,6 +89,7 @@
super.reset();
}
+ @SuppressWarnings("deprecation")
@Override
public PigStats launchPig(PhysicalPlan php,
String grpName,
@@ -99,7 +100,7 @@
JobCreationException,
Exception {
long sleepTime = 500;
- long jobidDelayTime = 1000;
+ int MAXRETRY_JOBID = 20;
aggregateWarning = "true".equalsIgnoreCase(pc.getProperties().getProperty("aggregate.warning"));
MROperPlan mrp = compile(php, pc);
PigStats stats = new PigStats();
@@ -135,35 +136,39 @@
jcThread.setUncaughtExceptionHandler(jctExceptionHandler);
jcThread.start();
- Thread.sleep(jobidDelayTime);
- String jobTrackerAdd;
- String port;
-
- try{
- for (Job job : waitingJobs){
- JobConf jConf = job.getJobConf();
+ for (Job job : waitingJobs){
+ int retry = 0;
+ while (job.getState()==Job.WAITING || job.getState()==Job.READY && retry<MAXRETRY_JOBID) {
+ Thread.sleep(sleepTime);
+ retry++;
+ }
+ JobConf jConf = job.getJobConf();
+ if (job.getAssignedJobID()!=null)
+ {
+ log.info("Submitting job: "+job.getAssignedJobID()+" to execution engine.");
+ }
+ // Try to get job tracker, if we cannot get it, then most probably we are running in local mode.
+ // If it is the case, we don't print out job tracker location and hint to kill the job,
+ // because it is meaningless for local mode.
+ try {
+ String jobTrackerAdd;
+ String port;
port = jConf.get("mapred.job.tracker.http.address");
+
port = port.substring(port.indexOf(":"));
jobTrackerAdd = jConf.get(HExecutionEngine.JOB_TRACKER_LOCATION);
jobTrackerAdd = jobTrackerAdd.substring(0,jobTrackerAdd.indexOf(":"));
- if (job.getAssignedJobID()!=null)
- {
- log.info("Submitting job: "+job.getAssignedJobID()+" to execution engine.");
- log.info("More information at: http://"+ jobTrackerAdd+port+
- "/jobdetails.jsp?jobid="+job.getAssignedJobID());
- log.info("To kill this job, use: kill "+job.getAssignedJobID());
- }
- else
- log.info("Cannot get jobid for this job");
+ log.info("More information at: http://"+ jobTrackerAdd+port+
+ "/jobdetails.jsp?jobid="+job.getAssignedJobID());
+ log.info("To kill this job, use: kill "+job.getAssignedJobID());
+ }
+ catch(Exception e){
+ /* This is extra information Pig is providing to user.
+ If exception occurs here, job may still complete successfully.
+ So, pig shouldn't die or even give confusing message to the user.
+ */
}
}
- catch(Exception e){
- /* This is extra information Pig is providing to user.
- If exception occurs here, job may still complete successfully.
- So, pig shouldn't die or even give confusing message to the user.
- So we just log information and move on. */
- log.info("Cannot get jobid for this job");
- }
while(!jc.allFinished()){
try {
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java Mon Dec 21 19:58:03 2009
@@ -113,13 +113,15 @@
continue;
}
if (isMapOnly(successor)) {
- if (isSingleLoadMapperPlan(successor.mapPlan)) {
+ if (isSingleLoadMapperPlan(successor.mapPlan)
+ && isSinglePredecessor(successor)) {
mappers.add(successor);
} else {
multiLoadMROpers.add(successor);
}
} else {
- if (isSingleLoadMapperPlan(successor.mapPlan)) {
+ if (isSingleLoadMapperPlan(successor.mapPlan)
+ && isSinglePredecessor(successor)) {
mapReducers.add(successor);
} else {
multiLoadMROpers.add(successor);
@@ -1121,6 +1123,10 @@
return (pl.getRoots().size() == 1);
}
+ private boolean isSinglePredecessor(MapReduceOper mr) {
+ return (getPlan().getPredecessors(mr).size() == 1);
+ }
+
private POSplit getSplit(){
return new POSplit(new OperatorKey(scope, nig.getNextNodeId(scope)));
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java Mon Dec 21 19:58:03 2009
@@ -102,6 +102,7 @@
BinaryComparisonOperator exprOp = new GreaterThanExpr(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), op
.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setOperandType(op.getLhsOperand().getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -133,6 +134,7 @@
BinaryComparisonOperator exprOp = new LessThanExpr(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), op
.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setOperandType(op.getLhsOperand().getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -163,6 +165,7 @@
BinaryComparisonOperator exprOp = new GTOrEqualToExpr(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), op
.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setOperandType(op.getLhsOperand().getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -192,6 +195,7 @@
BinaryComparisonOperator exprOp = new LTOrEqualToExpr(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), op
.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setOperandType(op.getLhsOperand().getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -221,6 +225,7 @@
BinaryComparisonOperator exprOp = new EqualToExpr(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), op
.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setOperandType(op.getLhsOperand().getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -250,6 +255,7 @@
BinaryComparisonOperator exprOp = new NotEqualToExpr(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), op
.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setOperandType(op.getLhsOperand().getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -279,6 +285,7 @@
BinaryComparisonOperator exprOp =
new PORegexp(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),
op.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setLhs((ExpressionOperator)logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator)logToPhyMap.get(op.getRhsOperand()));
LogicalPlan lp = op.getPlan();
@@ -306,6 +313,7 @@
String scope = op.getOperatorKey().scope;
BinaryExpressionOperator exprOp = new Add(new OperatorKey(scope,
nodeGen.getNextNodeId(scope)), op.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setResultType(op.getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -334,6 +342,7 @@
String scope = op.getOperatorKey().scope;
BinaryExpressionOperator exprOp = new Subtract(new OperatorKey(scope,
nodeGen.getNextNodeId(scope)), op.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setResultType(op.getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -362,6 +371,7 @@
String scope = op.getOperatorKey().scope;
BinaryExpressionOperator exprOp = new Multiply(new OperatorKey(scope,
nodeGen.getNextNodeId(scope)), op.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setResultType(op.getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -390,6 +400,7 @@
String scope = op.getOperatorKey().scope;
BinaryExpressionOperator exprOp = new Divide(new OperatorKey(scope,
nodeGen.getNextNodeId(scope)), op.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setResultType(op.getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -418,6 +429,7 @@
String scope = op.getOperatorKey().scope;
BinaryExpressionOperator exprOp = new Mod(new OperatorKey(scope,
nodeGen.getNextNodeId(scope)), op.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setResultType(op.getType());
exprOp.setLhs((ExpressionOperator) logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator) logToPhyMap.get(op.getRhsOperand()));
@@ -445,6 +457,7 @@
public void visit(LOAnd op) throws VisitorException {
String scope = op.getOperatorKey().scope;
BinaryComparisonOperator exprOp = new POAnd(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), op.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setLhs((ExpressionOperator)logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator)logToPhyMap.get(op.getRhsOperand()));
LogicalPlan lp = op.getPlan();
@@ -470,6 +483,7 @@
public void visit(LOOr op) throws VisitorException {
String scope = op.getOperatorKey().scope;
BinaryComparisonOperator exprOp = new POOr(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), op.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setLhs((ExpressionOperator)logToPhyMap.get(op.getLhsOperand()));
exprOp.setRhs((ExpressionOperator)logToPhyMap.get(op.getRhsOperand()));
LogicalPlan lp = op.getPlan();
@@ -495,6 +509,7 @@
public void visit(LONot op) throws VisitorException {
String scope = op.getOperatorKey().scope;
UnaryComparisonOperator exprOp = new PONot(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), op.getRequestedParallelism());
+ exprOp.setAlias(op.getAlias());
exprOp.setExpr((ExpressionOperator)logToPhyMap.get(op.getOperand()));
LogicalPlan lp = op.getPlan();
@@ -521,9 +536,10 @@
POGlobalRearrange poGlobal = new POGlobalRearrange(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), cs
.getRequestedParallelism());
+ poGlobal.setAlias(cs.getAlias());
POPackage poPackage = new POPackage(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), cs.getRequestedParallelism());
-
+ poGlobal.setAlias(cs.getAlias());
currentPlan.add(poGlobal);
currentPlan.add(poPackage);
@@ -554,6 +570,7 @@
POUserFunc gfc = new POUserFunc(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),cs.getRequestedParallelism(), Arrays.asList((PhysicalOperator)ce1,(PhysicalOperator)ce2), new FuncSpec(GFCross.class.getName()));
+ gfc.setAlias(cs.getAlias());
gfc.setResultType(DataType.BAG);
fep1.addAsLeaf(gfc);
gfc.setInputs(Arrays.asList((PhysicalOperator)ce1,(PhysicalOperator)ce2));
@@ -563,6 +580,7 @@
PhysicalPlan fep2 = new PhysicalPlan();
POProject feproj = new POProject(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cs.getRequestedParallelism());
+ feproj.setAlias(cs.getAlias());
feproj.setResultType(DataType.TUPLE);
feproj.setStar(true);
feproj.setOverloaded(false);
@@ -570,16 +588,19 @@
List<PhysicalPlan> fePlans = Arrays.asList(fep1, fep2);
POForEach fe = new POForEach(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cs.getRequestedParallelism(), fePlans, flattenLst );
+ fe.setAlias(cs.getAlias());
currentPlan.add(fe);
currentPlan.connect(logToPhyMap.get(op), fe);
POLocalRearrange physOp = new POLocalRearrange(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), cs
.getRequestedParallelism());
+ physOp.setAlias(cs.getAlias());
List<PhysicalPlan> lrPlans = new ArrayList<PhysicalPlan>();
for(int i=0;i<inputs.size();i++){
PhysicalPlan lrp1 = new PhysicalPlan();
POProject lrproj1 = new POProject(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cs.getRequestedParallelism(), i);
+ lrproj1.setAlias(cs.getAlias());
lrproj1.setOverloaded(false);
lrproj1.setResultType(DataType.INTEGER);
lrp1.add(lrproj1);
@@ -620,6 +641,7 @@
for(int i=1;i<=count;i++){
PhysicalPlan fep1 = new PhysicalPlan();
POProject feproj1 = new POProject(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cs.getRequestedParallelism(), i);
+ feproj1.setAlias(cs.getAlias());
feproj1.setResultType(DataType.BAG);
feproj1.setOverloaded(false);
fep1.add(feproj1);
@@ -628,6 +650,7 @@
}
POForEach fe = new POForEach(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cs.getRequestedParallelism(), fePlans, flattenLst );
+ fe.setAlias(cs.getAlias());
currentPlan.add(fe);
try{
currentPlan.connect(poPackage, fe);
@@ -658,9 +681,10 @@
POGlobalRearrange poGlobal = new POGlobalRearrange(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), cg.getRequestedParallelism());
+ poGlobal.setAlias(cg.getAlias());
POPackage poPackage = new POPackage(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), cg.getRequestedParallelism());
-
+ poPackage.setAlias(cg.getAlias());
currentPlan.add(poGlobal);
currentPlan.add(poPackage);
@@ -677,8 +701,8 @@
for (LogicalOperator op : inputs) {
List<LogicalPlan> plans = (List<LogicalPlan>)cg.getGroupByPlans().get(op);
POLocalRearrange physOp = new POLocalRearrange(new OperatorKey(
- scope, nodeGen.getNextNodeId(scope)), cg
- .getRequestedParallelism());
+ scope, nodeGen.getNextNodeId(scope)), cg.getRequestedParallelism());
+ physOp.setAlias(cg.getAlias());
List<PhysicalPlan> exprPlans = new ArrayList<PhysicalPlan>();
currentPlans.push(currentPlan);
for (LogicalPlan lp : plans) {
@@ -742,7 +766,7 @@
List<LogicalPlan> plans = (List<LogicalPlan>) cg.getGroupByPlans().get(op);
POCollectedGroup physOp = new POCollectedGroup(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)));
-
+ physOp.setAlias(cg.getAlias());
List<PhysicalPlan> exprPlans = new ArrayList<PhysicalPlan>();
currentPlans.push(currentPlan);
for (LogicalPlan lp : plans) {
@@ -836,6 +860,7 @@
try {
skj = new POSkewedJoin(new OperatorKey(scope,nodeGen.getNextNodeId(scope)),loj.getRequestedParallelism(),
inp, loj.getInnerFlags());
+ skj.setAlias(loj.getAlias());
skj.setJoinPlans(joinPlans);
}
catch (Exception e) {
@@ -923,6 +948,7 @@
pfrj = new POFRJoin(new OperatorKey(scope,nodeGen.getNextNodeId(scope)),loj.getRequestedParallelism(),
inp, ppLists, keyTypes, null, fragment, isLeftOuter, nullTuple);
+ pfrj.setAlias(loj.getAlias());
} catch (ExecException e1) {
int errCode = 2058;
String msg = "Unable to set index on newly create POLocalRearrange.";
@@ -973,9 +999,10 @@
POGlobalRearrange poGlobal = new POGlobalRearrange(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), loj
.getRequestedParallelism());
+ poGlobal.setAlias(loj.getAlias());
POPackage poPackage = new POPackage(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), loj.getRequestedParallelism());
-
+ poPackage.setAlias(loj.getAlias());
currentPlan.add(poGlobal);
currentPlan.add(poPackage);
@@ -1060,6 +1087,7 @@
PhysicalPlan fep1 = new PhysicalPlan();
POProject feproj1 = new POProject(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),
loj.getRequestedParallelism(), i+1); //i+1 since the first column is the "group" field
+ feproj1.setAlias(loj.getAlias());
feproj1.setResultType(DataType.BAG);
feproj1.setOverloaded(false);
fep1.add(feproj1);
@@ -1080,6 +1108,7 @@
POForEach fe = new POForEach(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),
loj.getRequestedParallelism(), fePlans, flattenLst );
+ fe.setAlias(loj.getAlias());
currentPlan.add(fe);
currentPlan.connect(poPackage, fe);
logToPhyMap.put(loj, fe);
@@ -1155,6 +1184,7 @@
String scope = filter.getOperatorKey().scope;
POFilter poFilter = new POFilter(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), filter.getRequestedParallelism());
+ poFilter.setAlias(filter.getAlias());
poFilter.setResultType(filter.getType());
currentPlan.add(poFilter);
logToPhyMap.put(filter, poFilter);
@@ -1197,6 +1227,7 @@
POStream poStream = new POStream(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), stream.getExecutableManager(),
stream.getStreamingCommand(), this.pc.getProperties());
+ poStream.setAlias(stream.getAlias());
currentPlan.add(poStream);
logToPhyMap.put(stream, poStream);
@@ -1232,6 +1263,7 @@
exprOp = new POProject(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), op.getRequestedParallelism());
}
+ exprOp.setAlias(op.getAlias());
exprOp.setResultType(op.getType());
exprOp.setColumns((ArrayList)op.getProjection());
exprOp.setStar(op.isStar());
@@ -1282,6 +1314,7 @@
POForEach poFE = new POForEach(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), g.getRequestedParallelism(), innerPlans,
g.getFlatten());
+ poFE.setAlias(g.getAlias());
poFE.setResultType(g.getType());
logToPhyMap.put(g, poFE);
currentPlan.add(poFE);
@@ -1337,6 +1370,7 @@
.getNextNodeId(scope)), s.getRequestedParallelism(), null,
sortPlans, s.getAscendingCols(), comparator);
}
+ sort.setAlias(s.getAlias());
sort.setLimit(s.getLimit());
// sort.setRequestedParallelism(s.getType());
logToPhyMap.put(s, sort);
@@ -1377,6 +1411,7 @@
// push it in the current plan and make the connections
PhysicalOperator physOp = new PODistinct(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), op.getRequestedParallelism());
+ physOp.setAlias(op.getAlias());
physOp.setResultType(op.getType());
logToPhyMap.put(op, physOp);
currentPlan.add(physOp);
@@ -1406,6 +1441,7 @@
String scope = split.getOperatorKey().scope;
PhysicalOperator physOp = new POSplit(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), split.getRequestedParallelism());
+ physOp.setAlias(split.getAlias());
FileSpec splStrFile;
try {
splStrFile = new FileSpec(FileLocalizer.getTemporaryPath(null, pc).toString(),new FuncSpec(BinStorage.class.getName()));
@@ -1457,6 +1493,7 @@
String scope = split.getOperatorKey().scope;
PhysicalOperator physOp = new POFilter(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), split.getRequestedParallelism());
+ physOp.setAlias(split.getAlias());
logToPhyMap.put(split, physOp);
currentPlan.add(physOp);
@@ -1506,6 +1543,7 @@
.getNextNodeId(scope)), func.getRequestedParallelism(),
null, func.getFuncSpec(), (ComparisonFunc) f);
}
+ p.setAlias(func.getAlias());
p.setResultType(func.getType());
currentPlan.add(p);
List<org.apache.pig.impl.logicalLayer.ExpressionOperator> fromList = func.getArguments();
@@ -1530,6 +1568,7 @@
String scope = loLoad.getOperatorKey().scope;
POLoad load = new POLoad(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), loLoad.isSplittable());
+ load.setAlias(loLoad.getAlias());
load.setLFile(loLoad.getInputFile());
load.setPc(pc);
load.setResultType(loLoad.getType());
@@ -1561,6 +1600,7 @@
POStore store = new POStore(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)));
+ store.setAlias(loStore.getPlan().getPredecessors(loStore).get(0).getAlias());
store.setSFile(loStore.getOutputFile());
store.setInputSpec(loStore.getInputSpec());
try {
@@ -1618,6 +1658,7 @@
String scope = op.getOperatorKey().scope;
ConstantExpression ce = new ConstantExpression(new OperatorKey(scope,
nodeGen.getNextNodeId(scope)));
+ ce.setAlias(op.getAlias());
ce.setValue(op.getValue());
ce.setResultType(op.getType());
//this operator doesn't have any predecessors
@@ -1630,6 +1671,7 @@
String scope = op.getOperatorKey().scope;
ExpressionOperator physOp = new POBinCond(new OperatorKey(scope,
nodeGen.getNextNodeId(scope)), op.getRequestedParallelism());
+ physOp.setAlias(op.getAlias());
logToPhyMap.put(op, physOp);
POBinCond phy = (POBinCond) physOp;
ExpressionOperator cond = (ExpressionOperator) logToPhyMap.get(op
@@ -1665,6 +1707,7 @@
ExpressionOperator physOp = new PONegative(new OperatorKey(scope,
nodeGen.getNextNodeId(scope)), op.getRequestedParallelism(),
null);
+ physOp.setAlias(op.getAlias());
currentPlan.add(physOp);
logToPhyMap.put(op, physOp);
@@ -1697,7 +1740,7 @@
String scope = op.getOperatorKey().scope;
UnaryComparisonOperator physOp = new POIsNull(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), op.getRequestedParallelism(), null);
-
+ physOp.setAlias(op.getAlias());
List<LogicalOperator> inputs = op.getPlan().getPredecessors(op);
ExpressionOperator from;
@@ -1734,6 +1777,7 @@
nodeGen.getNextNodeId(scope)), op.getRequestedParallelism(), op
.getLookUpKey());
physOp.setResultType(op.getType());
+ physOp.setAlias(op.getAlias());
currentPlan.add(physOp);
logToPhyMap.put(op, physOp);
@@ -1755,6 +1799,7 @@
String scope = op.getOperatorKey().scope;
ExpressionOperator physOp = new POCast(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), op.getRequestedParallelism());
+ physOp.setAlias(op.getAlias());
currentPlan.add(physOp);
logToPhyMap.put(op, physOp);
@@ -1791,6 +1836,7 @@
POLimit poLimit = new POLimit(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), limit.getRequestedParallelism());
poLimit.setResultType(limit.getType());
poLimit.setLimit(limit.getLimit());
+ poLimit.setAlias(limit.getAlias());
currentPlan.add(poLimit);
logToPhyMap.put(limit, poLimit);
@@ -1818,6 +1864,7 @@
String scope = op.getOperatorKey().scope;
POUnion physOp = new POUnion(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), op.getRequestedParallelism());
+ physOp.setAlias(op.getAlias());
currentPlan.add(physOp);
physOp.setResultType(op.getType());
logToPhyMap.put(op, physOp);
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/PhysicalOperator.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/PhysicalOperator.java?rev=892950&r1=892949&r2=892950&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/PhysicalOperator.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/PhysicalOperator.java Mon Dec 21 19:58:03 2009
@@ -89,6 +89,10 @@
// The result of performing the operation along with the output
protected Result res = null;
+
+ // alias associated with this PhysicalOperator
+ protected String alias = null;
+
// Will be used by operators to report status or transmit heartbeat
// Should be set by the backends to appropriate implementations that
// wrap their own version of a reporter.
@@ -161,6 +165,14 @@
return resultType;
}
+ public String getAlias() {
+ return alias;
+ }
+
+ public void setAlias(String alias) {
+ this.alias = alias;
+ }
+
public void setAccumulative() {
accum = true;
}