You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ch...@apache.org on 2013/10/31 20:21:29 UTC
svn commit: r1537592 [1/2] - in /pig/branches/tez: ./ conf/ ivy/ src/
src/docs/src/documentation/content/xdocs/ src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/ src...
Author: cheolsoo
Date: Thu Oct 31 19:21:28 2013
New Revision: 1537592
URL: http://svn.apache.org/r1537592
Log:
Merge the latest trunk changes
Added:
pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/DuplicateForEachColumnRewriteVisitor.java
- copied unchanged from r1537587, pig/trunk/src/org/apache/pig/newplan/logical/visitor/DuplicateForEachColumnRewriteVisitor.java
pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/ImplicitSplitInsertVisitor.java
- copied unchanged from r1537587, pig/trunk/src/org/apache/pig/newplan/logical/visitor/ImplicitSplitInsertVisitor.java
pig/branches/tez/test/org/apache/pig/test/PigStorageWithDifferentCaster.java
- copied unchanged from r1537587, pig/trunk/test/org/apache/pig/test/PigStorageWithDifferentCaster.java
pig/branches/tez/test/org/apache/pig/test/TestLineageFindRelVisitor.java
- copied unchanged from r1537587, pig/trunk/test/org/apache/pig/test/TestLineageFindRelVisitor.java
Removed:
pig/branches/tez/src/org/apache/pig/newplan/logical/rules/DuplicateForEachColumnRewrite.java
pig/branches/tez/src/org/apache/pig/newplan/logical/rules/ImplicitSplitInserter.java
Modified:
pig/branches/tez/ (props changed)
pig/branches/tez/CHANGES.txt
pig/branches/tez/build.xml
pig/branches/tez/conf/pig.properties
pig/branches/tez/ivy/libraries.properties
pig/branches/tez/ivy/pig-template.xml
pig/branches/tez/src/docs/src/documentation/content/xdocs/basic.xml
pig/branches/tez/src/docs/src/documentation/content/xdocs/perf.xml
pig/branches/tez/src/org/apache/pig/EvalFunc.java
pig/branches/tez/src/org/apache/pig/FuncSpec.java
pig/branches/tez/src/org/apache/pig/PigConfiguration.java
pig/branches/tez/src/org/apache/pig/PigServer.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/Launcher.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java
pig/branches/tez/src/org/apache/pig/builtin/CONCAT.java
pig/branches/tez/src/org/apache/pig/builtin/REGEX_EXTRACT_ALL.java
pig/branches/tez/src/org/apache/pig/builtin/StringConcat.java
pig/branches/tez/src/org/apache/pig/data/DefaultAbstractBag.java
pig/branches/tez/src/org/apache/pig/impl/util/Utils.java
pig/branches/tez/src/org/apache/pig/impl/util/avro/AvroStorageSchemaConversionUtilities.java
pig/branches/tez/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java
pig/branches/tez/src/org/apache/pig/newplan/logical/optimizer/AllSameRalationalNodesVisitor.java
pig/branches/tez/src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java
pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOGenerate.java
pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOJoin.java
pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java
pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/TypeCheckingExpVisitor.java
pig/branches/tez/src/org/apache/pig/parser/LogicalPlanBuilder.java
pig/branches/tez/src/org/apache/pig/pen/LineageTrimmingVisitor.java
pig/branches/tez/src/org/apache/pig/tools/parameters/PreprocessorContext.java
pig/branches/tez/src/pig-default.properties (props changed)
pig/branches/tez/test/e2e/pig/tests/negative.conf
pig/branches/tez/test/org/apache/pig/parser/TestQueryParser.java
pig/branches/tez/test/org/apache/pig/test/PigStorageWithStatistics.java
pig/branches/tez/test/org/apache/pig/test/TestBuiltin.java
pig/branches/tez/test/org/apache/pig/test/TestFRJoin2.java
pig/branches/tez/test/org/apache/pig/test/TestHBaseStorage.java
pig/branches/tez/test/org/apache/pig/test/TestMultiQueryCompiler.java
pig/branches/tez/test/org/apache/pig/test/TestMultiQueryLocal.java
pig/branches/tez/test/org/apache/pig/test/TestOptimizeLimit.java
pig/branches/tez/test/org/apache/pig/test/TestProjectStarRangeInUdf.java
pig/branches/tez/test/org/apache/pig/test/TestPruneColumn.java
pig/branches/tez/test/org/apache/pig/test/TestTypeCheckingValidatorNewLP.java
pig/branches/tez/test/org/apache/pig/test/TestUDF.java
pig/branches/tez/test/org/apache/pig/test/TestUnion.java
pig/branches/tez/test/org/apache/pig/test/data/bzipdir1.bz2/bzipdir2.bz2/recordLossblockHeaderEndsAt136500.txt.bz2 (props changed)
Propchange: pig/branches/tez/
------------------------------------------------------------------------------
Merged /pig/trunk:r1532966-1537587
Modified: pig/branches/tez/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/branches/tez/CHANGES.txt?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/CHANGES.txt (original)
+++ pig/branches/tez/CHANGES.txt Thu Oct 31 19:21:28 2013
@@ -22,18 +22,42 @@ Trunk (unreleased changes)
INCOMPATIBLE CHANGES
-PIG-3485: Remove CastUtils.bytesToMap(byte[] b) method from LoadCaster interface (cheolsoo)
+PIG-3485: Remove CastUtils.bytesToMap(byte[] b) method from LoadCaster interface (cheolsoo)
PIG-3419: Pluggable Execution Engine (achalsoni81 via cheolsoo)
IMPROVEMENTS
+PIG-3295: Casting from bytearray failing after Union even when each field is from a single Loader (knoguchi)
+
+PIG-3444: CONCAT with 2+ input parameters fail (lbendig via daijy)
+
+PIG-3529: Upgrade HBase dependency from 0.95-SNAPSHOT to 0.96 (jarcec via daijy)
+
+PIG-3117: A debug mode in which pig does not delete temporary files (ihadanny via cheolsoo)
+
PIG-3484: Make the size of pig.script property configurable (cheolsoo)
OPTIMIZATIONS
BUG FIXES
+PIG-3551: Minor typo on pig latin basics page (elserj via aniket486)
+
+PIG-3526: Unions with Enums do not work with AvroStorage (jadler via cheolsoo)
+
+PIG-3377: New AvroStorage throws NPE when storing untyped map/array/bag (jadler via cheolsoo)
+
+PIG-3542: Javadoc of REGEX_EXTRACT_ALL (nyigitba via daijy)
+
+PIG-3518: Need to ship jruby.jar in the release (daijy)
+
+PIG-3516: pig does not bring in joda-time as dependency in its pig-template.xml (daijy)
+
+PIG-3524: Clean up Launcher and MapReduceLauncher after PIG-3419 (cheolsoo)
+
+PIG-3515: Shell commands are limited from OS buffer (andronat via cheolsoo)
+
PIG-3520: Provide backward compatibility for PigRunner and PPNL after PIG-3419 (daijy via cheolsoo)
PIG-3519: Remove dependency on uber avro-tools jar (jarcec via cheolsoo)
@@ -52,22 +76,38 @@ PIG-3496: Propagate HBase 0.95 jars to t
Release 0.12.1 (unreleased changes)
+IMPROVEMENTS
+
+PIG-3549: Print hadoop jobids for failed, killed job (aniket486)
+
+PIG-3047: Check the size of a relation before adding it to distributed cache in Replicated join (aniket486)
+
+PIG-3480: TFile-based tmpfile compression crashes in some cases (dvryaboy via aniket486)
+
BUG FIXES
+PIG-3530: Some e2e tests is broken due to PIG-3480 (daijy)
+
+PIG-3492: ColumnPrune dropping used column due to LogicalRelationalOperator.fixDuplicateUids changes not propagating (knoguchi via daijy)
+
+PIG-3325: Adding a tuple to a bag is slow (dvryaboy via aniket486)
+
+PIG-3512: Reducer estimater is broken by PIG-3497
+
PIG-3510: New filter extractor fails with more than one filter statement (aniket486 via cheolsoo)
Release 0.12.0
INCOMPATIBLE CHANGES
+PIG-3082: outputSchema of a UDF allows two usages when describing a Tuple schema (jcoveney)
+
PIG-3191: [piggybank] MultiStorage output filenames are not sortable (Danny Antonelli via jcoveney)
PIG-3174: Remove rpm and deb artifacts from build.xml (gates)
IMPROVEMENTS
-PIG-3480: TFile-based tmpfile compression crashes in some cases (dvryaboy via aniket486)
-
PIG-3503: More document for Pig 0.12 new features (daijy)
PIG-3445: Make Parquet format available out of the box in Pig (lbendig via aniket486)
@@ -227,8 +267,6 @@ PIG-2764: Add a biginteger and bigdecima
PIG-3073: POUserFunc creating log spam for large scripts (jcoveney)
-PIG-3082: outputSchema of a UDF allows two usages when describing a Tuple schema (jcoveney)
-
PIG-3124: Push FLATTENs After FILTERs If Possible (nwhite via daijy)
PIG-3086: Allow A Prefix To Be Added To URIs In PigUnit Tests (nwhite via gates)
Modified: pig/branches/tez/build.xml
URL: http://svn.apache.org/viewvc/pig/branches/tez/build.xml?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/build.xml (original)
+++ pig/branches/tez/build.xml Thu Oct 31 19:21:28 2013
@@ -990,6 +990,7 @@
<copy todir="${tar.dist.dir}/lib">
<fileset dir="${ivy.lib.dir}" includes="jython*.jar"/>
+ <fileset dir="${ivy.lib.dir}" includes="jruby*.jar"/>
</copy>
<copy file="${output.jarfile.backcompat.withouthadoop}" tofile="${tar.dist.dir}/${final.name}-withouthadoop.jar" />
Modified: pig/branches/tez/conf/pig.properties
URL: http://svn.apache.org/viewvc/pig/branches/tez/conf/pig.properties?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/conf/pig.properties (original)
+++ pig/branches/tez/conf/pig.properties Thu Oct 31 19:21:28 2013
@@ -84,7 +84,11 @@
#using more counters than hadoop configured limit
#pig.disable.counter=true
-# Use this option to turn on UDF timers. This will cause two
+# By default, pig will allow 1GB of data to be replicated using
+# the distributed cache when doing fragment-replicated join.
+# pig.join.replicated.max.bytes=1000000000
+
+# Use this option to turn on UDF timers. This will cause two
# counters to be tracked for every UDF and LoadFunc in your script:
# approx_microsecs measures approximate time spent inside a UDF
# approx_invocations reports the approximate number of times the UDF was invoked
@@ -228,3 +232,8 @@ pig.location.check.strict=false
# By default, the size of pig script stored in job xml is limited to 10,240
# characters. This property can be used to configure it.
# pig.script.max.size=<somevalue>
+
+# Set this option to false to keep intermediate outputs files for debuggin
+# purpose. By default, it is set to true.
+# To inspect, use a = load '<path_to_tmp_file>' using org.apache.pig.impl.io.InterStorage();
+# pig.delete.temp.files=true
Modified: pig/branches/tez/ivy/libraries.properties
URL: http://svn.apache.org/viewvc/pig/branches/tez/ivy/libraries.properties?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/ivy/libraries.properties (original)
+++ pig/branches/tez/ivy/libraries.properties Thu Oct 31 19:21:28 2013
@@ -42,7 +42,7 @@ hadoop-common.version=2.2.0
hadoop-hdfs.version=2.2.0
hadoop-mapreduce.version=2.2.0
hbase94.version=0.94.1
-hbase95.version=0.95.3-hadoop1-SNAPSHOT
+hbase95.version=0.96.0-hadoop1
hsqldb.version=1.8.0.10
hive.version=0.8.0
httpcomponents.version=4.1
Modified: pig/branches/tez/ivy/pig-template.xml
URL: http://svn.apache.org/viewvc/pig/branches/tez/ivy/pig-template.xml?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/ivy/pig-template.xml (original)
+++ pig/branches/tez/ivy/pig-template.xml Thu Oct 31 19:21:28 2013
@@ -129,6 +129,11 @@
<version>4.0.4</version>
</dependency>
<dependency>
+ <groupId>joda-time</groupId>
+ <artifactId>joda-time</artifactId>
+ <version>2.1</version>
+ </dependency>
+ <dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>1.7.4</version>
Modified: pig/branches/tez/src/docs/src/documentation/content/xdocs/basic.xml
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/docs/src/documentation/content/xdocs/basic.xml?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/docs/src/documentation/content/xdocs/basic.xml (original)
+++ pig/branches/tez/src/docs/src/documentation/content/xdocs/basic.xml Thu Oct 31 19:21:28 2013
@@ -5087,7 +5087,7 @@ X = FILTER A BY (f1 matches '.*apache.*'
<p> [ ]</p>
</td>
<td>
- <p>Use to construct a bag from the specified elements. Equivalent to <a href="func.html#tomap">TOMAP</a>.</p>
+ <p>Use to construct a map from the specified elements. Equivalent to <a href="func.html#tomap">TOMAP</a>.</p>
</td>
</tr>
</table>
Modified: pig/branches/tez/src/docs/src/documentation/content/xdocs/perf.xml
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/docs/src/documentation/content/xdocs/perf.xml?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/docs/src/documentation/content/xdocs/perf.xml (original)
+++ pig/branches/tez/src/docs/src/documentation/content/xdocs/perf.xml Thu Oct 31 19:21:28 2013
@@ -1084,6 +1084,7 @@ C = JOIN big BY b1, tiny BY t1, mini BY
<p>Fragment replicate joins are experimental; we don't have a strong sense of how small the small relation must be to fit
into memory. In our tests with a simple query that involves just a JOIN, a relation of up to 100 M can be used if the process overall
gets 1 GB of memory. Please share your observations and experience with us.</p>
+<p>In order to avoid replicated joins on large relations, we fail if size of relation(s) to be replicated (in bytes) is greater than pig.join.replicated.max.bytes (default = 1GB).</p>
</section>
</section>
<!-- END FRAGMENT REPLICATE JOINS-->
Modified: pig/branches/tez/src/org/apache/pig/EvalFunc.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/EvalFunc.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/EvalFunc.java (original)
+++ pig/branches/tez/src/org/apache/pig/EvalFunc.java Thu Oct 31 19:21:28 2013
@@ -89,6 +89,15 @@ public abstract class EvalFunc<T> {
*/
protected Type returnType;
+ /**
+ * EvalFunc's schema type.
+ * @see {@link EvalFunc#getSchemaType()}
+ */
+ public static enum SchemaType {
+ NORMAL, //default field type
+ VARARG //if the last field of the (udf) schema is of type vararg
+ };
+
public EvalFunc() {
// Resolve concrete type for T of EvalFunc<T>
// 1. Build map from type param to type for class hierarchy from current class to EvalFunc
@@ -320,4 +329,16 @@ public abstract class EvalFunc<T> {
public Schema getInputSchema(){
return this.inputSchemaInternal;
}
+
+ /**
+ * Returns the {@link SchemaType} of the EvalFunc. User defined functions can override
+ * this method to return {@link SchemaType#VARARG}. In this case the last FieldSchema
+ * added to the Schema in {@link #getArgToFuncMapping()} will be considered as a vararg field.
+ *
+ * @return the schema type of the UDF
+ */
+ public SchemaType getSchemaType() {
+ return SchemaType.NORMAL;
+ }
+
}
Modified: pig/branches/tez/src/org/apache/pig/FuncSpec.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/FuncSpec.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/FuncSpec.java (original)
+++ pig/branches/tez/src/org/apache/pig/FuncSpec.java Thu Oct 31 19:21:28 2013
@@ -242,7 +242,8 @@ public class FuncSpec implements Seriali
@Override
public int hashCode() {
- return getClassName().hashCode() + ctorArgs.length;
+ return ctorArgs == null ? getClassName().hashCode() :
+ getClassName().hashCode() + ctorArgs.length;
}
@Override
Modified: pig/branches/tez/src/org/apache/pig/PigConfiguration.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/PigConfiguration.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/PigConfiguration.java (original)
+++ pig/branches/tez/src/org/apache/pig/PigConfiguration.java Thu Oct 31 19:21:28 2013
@@ -127,5 +127,16 @@ public class PigConfiguration {
* TFileStorage only support gzip and lzo.
*/
public static final String PIG_TEMP_FILE_COMPRESSION_CODEC = "pig.tmpfilecompression.codec";
+
+ /**
+ * This key is used to define whether to delete intermediate files of Hadoop jobs.
+ */
+ public static final String PIG_DELETE_TEMP_FILE = "pig.delete.temp.files";
+
+ /**
+ * This key used to control the maximum size loaded into
+ * the distributed cache when doing fragment-replicated join
+ */
+ public static final String PIG_JOIN_REPLICATED_MAX_BYTES = "pig.join.replicated.max.bytes";
}
Modified: pig/branches/tez/src/org/apache/pig/PigServer.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/PigServer.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/PigServer.java (original)
+++ pig/branches/tez/src/org/apache/pig/PigServer.java Thu Oct 31 19:21:28 2013
@@ -87,6 +87,8 @@ import org.apache.pig.newplan.logical.re
import org.apache.pig.newplan.logical.relational.LogicalSchema;
import org.apache.pig.newplan.logical.visitor.CastLineageSetter;
import org.apache.pig.newplan.logical.visitor.ColumnAliasConversionVisitor;
+import org.apache.pig.newplan.logical.visitor.DuplicateForEachColumnRewriteVisitor;
+import org.apache.pig.newplan.logical.visitor.ImplicitSplitInsertVisitor;
import org.apache.pig.newplan.logical.visitor.ScalarVariableValidator;
import org.apache.pig.newplan.logical.visitor.ScalarVisitor;
import org.apache.pig.newplan.logical.visitor.SchemaAliasVisitor;
@@ -1685,8 +1687,14 @@ public class PigServer {
new SchemaAliasVisitor(lp).visit();
new ScalarVisitor(lp, pigContext, scope).visit();
- // TODO: move optimizer here from HExecuteEngine.
- // TODO: input/output validation visitor
+ // ImplicitSplitInsertVisitor has to be called before
+ // DuplicateForEachColumnRewriteVisitor. Detail at pig-1766
+ new ImplicitSplitInsertVisitor(lp).visit();
+
+ // DuplicateForEachColumnRewriteVisitor should be before
+ // TypeCheckingRelVisitor which does resetSchema/getSchema
+ // heavily
+ new DuplicateForEachColumnRewriteVisitor(lp).visit();
CompilationMessageCollector collector = new CompilationMessageCollector() ;
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/Launcher.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/Launcher.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/Launcher.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/Launcher.java Thu Oct 31 19:21:28 2013
@@ -129,11 +129,7 @@ public abstract class Launcher {
* @param php
* @param grpName
* @param pc
- * @throws PlanException
- * @throws VisitorException
- * @throws IOException
- * @throws ExecException
- * @throws JobCreationException
+ * @throws Exception
*/
public abstract PigStats launchPig(PhysicalPlan php, String grpName,
PigContext pc) throws Exception;
@@ -315,7 +311,7 @@ public abstract class Launcher {
try {
jobControlException = getExceptionFromString(jobControlExceptionStackTrace);
} catch (Exception e) {
- String errMsg = "Could not resolve error that occured when launching map reduce job: "
+ String errMsg = "Could not resolve error that occured when launching job: "
+ jobControlExceptionStackTrace;
jobControlException = new RuntimeException(errMsg, throwable);
}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java Thu Oct 31 19:21:28 2013
@@ -17,6 +17,9 @@
*/
package org.apache.pig.backend.hadoop.executionengine.mapReduceLayer;
+import java.io.IOException;
+import java.util.List;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -29,11 +32,8 @@ import org.apache.pig.LoadMetadata;
import org.apache.pig.ResourceStatistics;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper;
+import org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil;
import org.apache.pig.impl.util.UriUtil;
-import org.apache.pig.impl.util.Utils;
-
-import java.io.IOException;
-import java.util.List;
/**
* Class that estimates the number of reducers based on input size.
@@ -116,7 +116,7 @@ public class InputSizeReducerEstimator i
FileStatus[] status = fs.globStatus(path);
if (status != null) {
for (FileStatus s : status) {
- totalInputFileSize += Utils.getPathLength(fs, s);
+ totalInputFileSize += MapRedUtil.getPathLength(fs, s);
foundSize = true;
}
}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java Thu Oct 31 19:21:28 2013
@@ -77,6 +77,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper;
import org.apache.pig.backend.hadoop.executionengine.shims.HadoopShims;
+import org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataType;
import org.apache.pig.data.SchemaTupleFrontend;
@@ -1497,24 +1498,39 @@ public class JobControlCompiler{
FileSpec[] newReplFiles = new FileSpec[replFiles.length];
// the first input is not replicated
- for (int i = 0; i < replFiles.length; i++) {
- // ignore fragmented file
- String symlink = "";
- if (i != join.getFragment()) {
- symlink = "pigrepl_" + join.getOperatorKey().toString() + "_"
- + Integer.toString(System.identityHashCode(replFiles[i].getFileName()))
- + "_" + Long.toString(System.currentTimeMillis())
- + "_" + i;
- replicatedPath.add(replFiles[i].getFileName() + "#"
- + symlink);
+ long sizeOfReplicatedInputs = 0;
+ try {
+ for (int i = 0; i < replFiles.length; i++) {
+ // ignore fragmented file
+ String symlink = "";
+ if (i != join.getFragment()) {
+ symlink = "pigrepl_" + join.getOperatorKey().toString() + "_"
+ + Integer.toString(System.identityHashCode(
+ replFiles[i].getFileName()))
+ + "_" + Long.toString(System.currentTimeMillis())
+ + "_" + i;
+ replicatedPath.add(replFiles[i].getFileName() + "#"
+ + symlink);
+
+ Path path = new Path(replFiles[i].getFileName());
+ FileSystem fs = path.getFileSystem(conf);
+ sizeOfReplicatedInputs +=
+ MapRedUtil.getPathLength(fs, fs.getFileStatus(path));
+ }
+ newReplFiles[i] = new FileSpec(symlink,
+ (replFiles[i] == null ? null : replFiles[i].getFuncSpec()));
}
- newReplFiles[i] = new FileSpec(symlink,
- (replFiles[i] == null ? null : replFiles[i].getFuncSpec()));
- }
- join.setReplFiles(newReplFiles);
+ join.setReplFiles(newReplFiles);
+
+ String maxSize = pigContext.getProperties().getProperty(
+ PigConfiguration.PIG_JOIN_REPLICATED_MAX_BYTES, "1000000000");
+ if (sizeOfReplicatedInputs > Long.parseLong(maxSize)){
+ throw new VisitorException("Replicated input files size: "
+ + sizeOfReplicatedInputs + " exceeds " +
+ PigConfiguration.PIG_JOIN_REPLICATED_MAX_BYTES + ": " + maxSize);
+ }
- try {
setupDistributedCache(pigContext, conf, replicatedPath
.toArray(new String[0]), false);
} catch (IOException e) {
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java Thu Oct 31 19:21:28 2013
@@ -103,7 +103,7 @@ public class MapReduceLauncher extends L
RunningJob runningJob = job.getJobClient().getJob(job.getAssignedJobID());
if (runningJob!=null)
runningJob.killJob();
- log.info("Job " + job.getJobID() + " killed");
+ log.info("Job " + job.getAssignedJobID().toString() + " killed");
}
}
} catch (Exception e) {
@@ -416,9 +416,11 @@ public class MapReduceLauncher extends L
failed = true;
}
- // Clean up all the intermediate data
- for (String path : intermediateVisitor.getIntermediate()) {
- FileLocalizer.delete(path, pc);
+ if (!"false".equalsIgnoreCase(pc.getProperties().getProperty(PigConfiguration.PIG_DELETE_TEMP_FILE))) {
+ // Clean up all the intermediate data
+ for (String path : intermediateVisitor.getIntermediate()) {
+ FileLocalizer.delete(path, pc);
+ }
}
// Look to see if any jobs failed. If so, we need to report that.
@@ -535,7 +537,7 @@ public class MapReduceLauncher extends L
for (int i=0; i<jc.getFailedJobs().size(); i++) {
Job j = jc.getFailedJobs().get(i);
- msg.append(j.getMessage());
+ msg.append("JobID: " + j.getAssignedJobID().toString() + " Reason: " + j.getMessage());
if (i!=jc.getFailedJobs().size()-1) {
msg.append("\n");
}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java Thu Oct 31 19:21:28 2013
@@ -61,6 +61,7 @@ import org.joda.time.DateTime;
*/
public class POCast extends ExpressionOperator {
private final static Log log = LogFactory.getLog(POCast.class);
+ private static final String unknownByteArrayErrorMessage = "Received a bytearray from the UDF or Union from two different Loaders. Cannot determine how to convert the bytearray to ";
private FuncSpec funcSpec = null;
transient private LoadCaster caster;
private boolean castNotNeeded = false;
@@ -164,7 +165,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToBigInteger(dba.get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to BigInteger.";
+ String msg = unknownByteArrayErrorMessage + "BigInteger.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -280,7 +281,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToBigDecimal(dba.get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to BigDecimal.";
+ String msg = unknownByteArrayErrorMessage + "BigDecimal.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -395,7 +396,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToBoolean(dba.get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to boolean.";
+ String msg = unknownByteArrayErrorMessage + "boolean.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -509,7 +510,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToInteger(dba.get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to int.";
+ String msg = unknownByteArrayErrorMessage + "int.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -635,7 +636,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToLong(dba.get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to long.";
+ String msg = unknownByteArrayErrorMessage + "long.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -758,7 +759,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToDouble(dba.get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to double.";
+ String msg = unknownByteArrayErrorMessage + "double.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -880,7 +881,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToFloat(dba.get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to float.";
+ String msg = unknownByteArrayErrorMessage + "float.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -1006,7 +1007,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToDateTime(dba.get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to datetime.";
+ String msg = unknownByteArrayErrorMessage + "datetime.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -1117,7 +1118,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToCharArray(dba.get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to string.";
+ String msg = unknownByteArrayErrorMessage + "string.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -1269,7 +1270,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToTuple(dba.get(), fieldSchema);
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to tuple.";
+ String msg = unknownByteArrayErrorMessage + "tuple.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -1331,7 +1332,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToBag(((DataByteArray)obj).get(), fs);
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to bag.";
+ String msg = unknownByteArrayErrorMessage + "bag.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} else {
@@ -1362,7 +1363,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToTuple(((DataByteArray)obj).get(), fs);
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to tuple.";
+ String msg = unknownByteArrayErrorMessage + "tuple.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} else {
@@ -1387,7 +1388,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToMap(((DataByteArray)obj).get(), fs);
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to tuple.";
+ String msg = unknownByteArrayErrorMessage + "tuple.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} else {
@@ -1401,7 +1402,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToBoolean(((DataByteArray) obj).get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to int.";
+ String msg = unknownByteArrayErrorMessage + "int.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
break;
@@ -1440,7 +1441,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToInteger(((DataByteArray) obj).get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to int.";
+ String msg = unknownByteArrayErrorMessage + "int.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
break;
@@ -1486,7 +1487,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToDouble(((DataByteArray) obj).get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to double.";
+ String msg = unknownByteArrayErrorMessage + "double.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
break;
@@ -1532,7 +1533,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToLong(((DataByteArray)obj).get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to long.";
+ String msg = unknownByteArrayErrorMessage + "long.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
break;
@@ -1578,7 +1579,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToFloat(((DataByteArray)obj).get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to float.";
+ String msg = unknownByteArrayErrorMessage + "float.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
break;
@@ -1624,7 +1625,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToDateTime(((DataByteArray)obj).get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to datetime.";
+ String msg = unknownByteArrayErrorMessage + "datetime.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
break;
@@ -1663,7 +1664,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToCharArray(((DataByteArray)obj).get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to float.";
+ String msg = unknownByteArrayErrorMessage + "float.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
break;
@@ -1711,7 +1712,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToBigInteger(((DataByteArray)obj).get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to BigInteger.";
+ String msg = unknownByteArrayErrorMessage + "BigInteger.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
break;
@@ -1756,7 +1757,7 @@ public class POCast extends ExpressionOp
result = caster.bytesToBigDecimal(((DataByteArray)obj).get());
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to BigDecimal.";
+ String msg = unknownByteArrayErrorMessage + "BigDecimal.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
break;
@@ -1860,7 +1861,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToBag(dba.get(), fieldSchema);
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to bag.";
+ String msg = unknownByteArrayErrorMessage + "bag.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -1951,7 +1952,7 @@ public class POCast extends ExpressionOp
res.result = caster.bytesToMap(dba.get(), fieldSchema);
} else {
int errCode = 1075;
- String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to map.";
+ String msg = unknownByteArrayErrorMessage + "map.";
throw new ExecException(msg, errCode, PigException.INPUT);
}
} catch (ExecException ee) {
@@ -2036,4 +2037,4 @@ public class POCast extends ExpressionOp
public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
return (Tuple) out;
}
-}
\ No newline at end of file
+}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java Thu Oct 31 19:21:28 2013
@@ -238,17 +238,36 @@ public class MapRedUtil {
result.add(stat);
}
}
- }
+ }
private static final PathFilter hiddenFileFilter = new PathFilter(){
public boolean accept(Path p){
- String name = p.getName();
- return !name.startsWith("_") && !name.startsWith(".");
+ String name = p.getName();
+ return !name.startsWith("_") && !name.startsWith(".");
}
- };
+ };
+
+ /**
+ * Returns the total number of bytes for this file,
+ * or if a directory all files in the directory.
+ */
+ public static long getPathLength(FileSystem fs, FileStatus status)
+ throws IOException {
+ if (!status.isDir()) {
+ return status.getLen();
+ } else {
+ FileStatus[] children = fs.listStatus(
+ status.getPath(), hiddenFileFilter);
+ long size = 0;
+ for (FileStatus child : children) {
+ size += getPathLength(fs, child);
+ }
+ return size;
+ }
+ }
/* The following codes are for split combination: see PIG-1518
- *
+ *
*/
private static Comparator<Node> nodeComparator = new Comparator<Node>() {
@Override
Modified: pig/branches/tez/src/org/apache/pig/builtin/CONCAT.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/builtin/CONCAT.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/builtin/CONCAT.java (original)
+++ pig/branches/tez/src/org/apache/pig/builtin/CONCAT.java Thu Oct 31 19:21:28 2013
@@ -82,4 +82,9 @@ public class CONCAT extends EvalFunc<Dat
funcList.add(new FuncSpec(StringConcat.class.getName(), s));
return funcList;
}
+
+ @Override
+ public SchemaType getSchemaType() {
+ return SchemaType.VARARG;
+ }
}
Modified: pig/branches/tez/src/org/apache/pig/builtin/REGEX_EXTRACT_ALL.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/builtin/REGEX_EXTRACT_ALL.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/builtin/REGEX_EXTRACT_ALL.java (original)
+++ pig/branches/tez/src/org/apache/pig/builtin/REGEX_EXTRACT_ALL.java Thu Oct 31 19:21:28 2013
@@ -43,7 +43,7 @@ import org.apache.pig.impl.logicalLayer.
* <dt><b>Matching strategy:</b></dt>
* <dd>Trying to match the entire input by using {@link Matcher#matches()} instead of
* {@link Matcher#find()} (default useMatches=true).</dd>
- * <dd><code>DEFINE GREEDY_EXTRACT REGEX_EXTRACT(false);</code></dd>
+ * <dd><code>DEFINE GREEDY_EXTRACT REGEX_EXTRACT_ALL(false);</code></dd>
* </dl>
*/
Modified: pig/branches/tez/src/org/apache/pig/builtin/StringConcat.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/builtin/StringConcat.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/builtin/StringConcat.java (original)
+++ pig/branches/tez/src/org/apache/pig/builtin/StringConcat.java Thu Oct 31 19:21:28 2013
@@ -20,6 +20,7 @@ package org.apache.pig.builtin;
import java.io.IOException;
import org.apache.pig.EvalFunc;
import org.apache.pig.PigException;
+import org.apache.pig.EvalFunc.SchemaType;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
@@ -57,5 +58,9 @@ public class StringConcat extends EvalFu
public Schema outputSchema(Schema input) {
return new Schema(new Schema.FieldSchema(null, DataType.CHARARRAY));
}
-
+
+ @Override
+ public SchemaType getSchemaType() {
+ return SchemaType.VARARG;
+ }
}
Modified: pig/branches/tez/src/org/apache/pig/data/DefaultAbstractBag.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/data/DefaultAbstractBag.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/data/DefaultAbstractBag.java (original)
+++ pig/branches/tez/src/org/apache/pig/data/DefaultAbstractBag.java Thu Oct 31 19:21:28 2013
@@ -65,9 +65,13 @@ public abstract class DefaultAbstractBag
// to run through the disk when people ask.
protected long mSize = 0;
- protected int mLastContentsSize = -1;
+ // Number of tuples to sample per bag, to get an estimate of tuple size
+ private static final int SPILL_SAMPLE_SIZE = 100;
+ private static final int SPILL_SAMPLE_FREQUENCY = 10;
- protected long avgTupleSize = 0;
+ long aggSampleTupleSize = 0;
+
+ int sampled = 0;
private boolean spillableRegistered = false;
@@ -79,6 +83,28 @@ public abstract class DefaultAbstractBag
return mSize;
}
+
+ /**
+ * Sample every SPILL_SAMPLE_FREQUENCYth tuple
+ * until we reach a max of SPILL_SAMPLE_SIZE
+ * to get an estimate of the tuple sizes.
+ */
+ protected void sampleContents() {
+ synchronized (mContents) {
+ Iterator<Tuple> iter = mContents.iterator();
+ for (int i = 0; i < sampled * SPILL_SAMPLE_FREQUENCY && iter.hasNext(); i++) {
+ iter.next();
+ }
+ for (int i = sampled; iter.hasNext() && sampled < SPILL_SAMPLE_SIZE; i++) {
+ Tuple t = iter.next();
+ if (i % SPILL_SAMPLE_FREQUENCY == 0) {
+ aggSampleTupleSize += t.getMemorySize();
+ sampled += 1;
+ }
+ }
+ }
+ }
+
/**
* Add a tuple to the bag.
* @param t tuple to add.
@@ -97,9 +123,12 @@ public abstract class DefaultAbstractBag
* should call this method after every time they add an element.
*/
protected void markSpillableIfNecessary() {
- if (!spillableRegistered && getMemorySize() >= SPILL_REGISTER_THRESHOLD) {
- SpillableMemoryManager.getInstance().registerSpillable(this);
- spillableRegistered = true;
+ if (!spillableRegistered) {
+ long estimate = getMemorySize();
+ if ( estimate >= SPILL_REGISTER_THRESHOLD) {
+ SpillableMemoryManager.getInstance().registerSpillable(this);
+ spillableRegistered = true;
+ }
}
}
@@ -130,40 +159,30 @@ public abstract class DefaultAbstractBag
*/
@Override
public long getMemorySize() {
- int j = 0;
int numInMem = 0;
synchronized (mContents) {
numInMem = mContents.size();
-
// If we've already gotten the estimate
- // and the number of tuples hasn't changed, or was above 100 and
- // is still above 100, we can
+ // and the number of tuples hasn't changed, or was above
+ // the sample size and is still above the sample size, we can
// produce a new estimate without sampling the tuples again.
- if (avgTupleSize != 0 && (mLastContentsSize == numInMem ||
- mLastContentsSize > 100 && numInMem > 100))
- return totalSizeFromAvgTupleSize(avgTupleSize, numInMem);
-
- // Measure only what's in memory, not what's on disk.
- // I can't afford to walk through all the tuples every time the
- // memory manager wants to know if it's time to dump. Just sample
- // the first 100 and see what we get. This may not be 100%
- // accurate, but it's just an estimate anyway.
- Iterator<Tuple> i = mContents.iterator();
- for (j = 0; i.hasNext() && j < 100; j++) {
- avgTupleSize += i.next().getMemorySize();
+ if (sampled != 0 && (sampled == numInMem ||
+ sampled > SPILL_SAMPLE_SIZE && numInMem > SPILL_SAMPLE_SIZE)) {
+ return totalSizeFromAvgTupleSize(aggSampleTupleSize/sampled, numInMem);
+ }
+ sampleContents();
+ int avgTupleSize;
+ if (sampled != 0) {
+ avgTupleSize = (int) (aggSampleTupleSize / sampled);
+ } else {
+ avgTupleSize = 0;
}
- }
- mLastContentsSize = numInMem;
- if (j != 0) {
- avgTupleSize /= j;
- } else {
- avgTupleSize = 0;
- }
- return totalSizeFromAvgTupleSize(avgTupleSize, numInMem);
+ return totalSizeFromAvgTupleSize(avgTupleSize, numInMem);
}
+ }
private long totalSizeFromAvgTupleSize(long avgTupleSize, int numInMem) {
long used = avgTupleSize * numInMem;
@@ -227,6 +246,9 @@ public abstract class DefaultAbstractBag
mSpillFiles.clear();
}
mSize = 0;
+ aggSampleTupleSize = 0;
+ sampled = 0;
+ // not changing spillableRegistered -- clear doesn't change that.
}
}
Modified: pig/branches/tez/src/org/apache/pig/impl/util/Utils.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/impl/util/Utils.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/impl/util/Utils.java (original)
+++ pig/branches/tez/src/org/apache/pig/impl/util/Utils.java Thu Oct 31 19:21:28 2013
@@ -73,7 +73,6 @@ import org.apache.pig.impl.logicalLayer.
import org.apache.pig.newplan.logical.relational.LogicalSchema;
import org.apache.pig.parser.ParserException;
import org.apache.pig.parser.QueryParserDriver;
-import org.xerial.snappy.SnappyCodec;
import com.google.common.collect.Lists;
@@ -255,7 +254,7 @@ public class Utils {
GZ (GzipCodec.class.getName()),
GZIP (GzipCodec.class.getName()),
LZO ("com.hadoop.compression.lzo.LzoCodec"),
- SNAPPY (SnappyCodec.class.getName()),
+ SNAPPY ("org.xerial.snappy.SnappyCodec"),
BZIP2 (BZip2Codec.class.getName());
private String hadoopCodecClassName;
@@ -479,22 +478,6 @@ public class Utils {
}
/**
- * Returns the total number of bytes for this file, or if a file all files in the directory.
- */
- public static long getPathLength(FileSystem fs, FileStatus status) throws IOException {
- if (!status.isDir()) {
- return status.getLen();
- } else {
- FileStatus[] children = fs.listStatus(status.getPath());
- long size = 0;
- for (FileStatus child : children) {
- size += getPathLength(fs, child);
- }
- return size;
- }
- }
-
- /**
* Method to set the customized SSH socket factory.
*
* @param pc Pig context that stores the SSH gateway address.
Modified: pig/branches/tez/src/org/apache/pig/impl/util/avro/AvroStorageSchemaConversionUtilities.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/impl/util/avro/AvroStorageSchemaConversionUtilities.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/impl/util/avro/AvroStorageSchemaConversionUtilities.java (original)
+++ pig/branches/tez/src/org/apache/pig/impl/util/avro/AvroStorageSchemaConversionUtilities.java Thu Oct 31 19:21:28 2013
@@ -427,6 +427,9 @@ public class AvroStorageSchemaConversion
schema.getFields()[0].getSchema(), name, null,
definedRecordNames,
doubleColonsToDoubleUnderscores);
+ if (innerBagSchema == null) {
+ throw new IOException("AvroStorage can't save bags with untyped values; please specify a value type or a schema.");
+ }
return createNullableUnion(Schema.createArray(innerBagSchema));
case DataType.BIGCHARARRAY:
return createNullableUnion(Type.STRING);
@@ -457,10 +460,15 @@ public class AvroStorageSchemaConversion
case DataType.LONG:
return createNullableUnion(Type.LONG);
case DataType.MAP:
+ if (schema == null) {
+ throw new IOException("AvroStorage can't save maps with untyped values; please specify a value type or a schema.");
+ }
byte innerType = schema.getFields()[0].getType();
String desc = schema.getFields()[0].getDescription();
- if (desc.equals("autogenerated from Pig Field Schema")) {
- desc = null;
+ if (desc != null) {
+ if (desc.equals("autogenerated from Pig Field Schema")) {
+ desc = null;
+ }
}
Schema innerSchema;
if (DataType.isComplex(innerType)) {
@@ -480,6 +488,9 @@ public class AvroStorageSchemaConversion
case DataType.NULL:
return Schema.create(Type.NULL);
case DataType.TUPLE:
+ if (schema == null) {
+ throw new IOException("AvroStorage can't save tuples with untyped values; please specify a value type or a schema.");
+ }
Schema returnSchema = createNullableUnion(
resourceSchemaToAvroSchema(schema, name, null,
definedRecordNames, doubleColonsToDoubleUnderscores));
Modified: pig/branches/tez/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java (original)
+++ pig/branches/tez/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java Thu Oct 31 19:21:28 2013
@@ -32,6 +32,7 @@ import org.apache.avro.Schema.Type;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.IndexedRecord;
+import org.apache.avro.generic.GenericEnumSymbol;
import org.apache.commons.logging.LogFactory;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.DataByteArray;
@@ -143,6 +144,8 @@ public final class AvroTupleWrapper <T e
return new DataByteArray(((GenericData.Fixed) o).bytes());
} else if (o instanceof ByteBuffer) {
return new DataByteArray(((ByteBuffer) o).array());
+ } else if (o instanceof GenericEnumSymbol) {
+ return o.toString();
}
default:
return o;
Modified: pig/branches/tez/src/org/apache/pig/newplan/logical/optimizer/AllSameRalationalNodesVisitor.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/newplan/logical/optimizer/AllSameRalationalNodesVisitor.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/newplan/logical/optimizer/AllSameRalationalNodesVisitor.java (original)
+++ pig/branches/tez/src/org/apache/pig/newplan/logical/optimizer/AllSameRalationalNodesVisitor.java Thu Oct 31 19:21:28 2013
@@ -23,11 +23,14 @@ import org.apache.pig.newplan.OperatorPl
import org.apache.pig.newplan.PlanWalker;
import org.apache.pig.newplan.logical.relational.LOCogroup;
import org.apache.pig.newplan.logical.relational.LOCross;
+import org.apache.pig.newplan.logical.relational.LOCube;
import org.apache.pig.newplan.logical.relational.LODistinct;
import org.apache.pig.newplan.logical.relational.LOFilter;
import org.apache.pig.newplan.logical.relational.LOForEach;
import org.apache.pig.newplan.logical.relational.LOJoin;
+import org.apache.pig.newplan.logical.relational.LOLimit;
import org.apache.pig.newplan.logical.relational.LOLoad;
+import org.apache.pig.newplan.logical.relational.LONative;
import org.apache.pig.newplan.logical.relational.LORank;
import org.apache.pig.newplan.logical.relational.LOSort;
import org.apache.pig.newplan.logical.relational.LOSplit;
@@ -129,4 +132,19 @@ public abstract class AllSameRalationalN
public void visit(LOStream stream) throws FrontendException {
execute(stream);
}
+
+ @Override
+ public void visit(LOLimit limit) throws FrontendException {
+ execute(limit);
+ }
+
+ @Override
+ public void visit(LONative loNative) throws FrontendException {
+ execute(loNative);
+ }
+
+ @Override
+ public void visit(LOCube cube) throws FrontendException {
+ execute(cube);
+ }
}
Modified: pig/branches/tez/src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java (original)
+++ pig/branches/tez/src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java Thu Oct 31 19:21:28 2013
@@ -31,10 +31,8 @@ import org.apache.commons.logging.LogFac
import org.apache.pig.newplan.OperatorPlan;
import org.apache.pig.newplan.logical.rules.AddForEach;
import org.apache.pig.newplan.logical.rules.ColumnMapKeyPrune;
-import org.apache.pig.newplan.logical.rules.DuplicateForEachColumnRewrite;
import org.apache.pig.newplan.logical.rules.FilterAboveForeach;
import org.apache.pig.newplan.logical.rules.GroupByConstParallelSetter;
-import org.apache.pig.newplan.logical.rules.ImplicitSplitInserter;
import org.apache.pig.newplan.logical.rules.LimitOptimizer;
import org.apache.pig.newplan.logical.rules.LoadTypeCastInserter;
import org.apache.pig.newplan.logical.rules.LogicalExpressionSimplifier;
@@ -77,27 +75,10 @@ public class LogicalPlanOptimizer extend
protected List<Set<Rule>> buildRuleSets() {
List<Set<Rule>> ls = new ArrayList<Set<Rule>>();
- // ImplicitSplitInserter set
- // This set of rules Insert Foreach dedicated for casting after load
- Set<Rule> s = new HashSet<Rule>();
- Rule r = new ImplicitSplitInserter("ImplicitSplitInserter");
- checkAndAddRule(s, r);
- if (!s.isEmpty())
- ls.add(s);
-
- // DuplicateForEachColumnRewrite set
- // This insert Identity UDF in the case foreach duplicate field.
- // This is because we need unique uid through out the plan
- s = new HashSet<Rule>();
- r = new DuplicateForEachColumnRewrite("DuplicateForEachColumnRewrite");
- checkAndAddRule(s, r);
- if (!s.isEmpty())
- ls.add(s);
-
// Logical expression simplifier
- s = new HashSet<Rule>();
+ Set <Rule> s = new HashSet<Rule>();
// add logical expression simplification rule
- r = new LogicalExpressionSimplifier("FilterLogicExpressionSimplifier");
+ Rule r = new LogicalExpressionSimplifier("FilterLogicExpressionSimplifier");
checkAndAddRule(s, r);
ls.add(s);
Modified: pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOGenerate.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOGenerate.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOGenerate.java (original)
+++ pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOGenerate.java Thu Oct 31 19:21:28 2013
@@ -193,10 +193,6 @@ public class LOGenerate extends LogicalR
outputPlanSchemas = null;
}
- if (schema != null) {
- LogicalRelationalOperator.fixDuplicateUids(schema.getFields());
- }
-
return schema;
}
Modified: pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOJoin.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOJoin.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOJoin.java (original)
+++ pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOJoin.java Thu Oct 31 19:21:28 2013
@@ -156,8 +156,6 @@ public class LOJoin extends LogicalRelat
}
}
- LogicalRelationalOperator.fixDuplicateUids(fss);
-
schema = new LogicalSchema();
for(LogicalSchema.LogicalFieldSchema fieldSchema: fss) {
schema.addField(fieldSchema);
Modified: pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java (original)
+++ pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java Thu Oct 31 19:21:28 2013
@@ -17,14 +17,20 @@
*/
package org.apache.pig.newplan.logical.visitor;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.pig.FuncSpec;
+import org.apache.pig.LoadCaster;
+import org.apache.pig.LoadFunc;
import org.apache.pig.PigException;
import org.apache.pig.data.DataType;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.StreamToPig;
import org.apache.pig.impl.builtin.IdentityColumn;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.plan.VisitorException;
@@ -86,6 +92,8 @@ public class LineageFindRelVisitor exten
// then the mapping is stored here
Map<LogicalRelationalOperator, FuncSpec> rel2InputFuncMap =
new HashMap<LogicalRelationalOperator, FuncSpec>();
+
+ Map<FuncSpec, Class> func2casterMap = new HashMap<FuncSpec, Class>();
public LineageFindRelVisitor(OperatorPlan plan) throws FrontendException {
super(plan, new DependencyOrderWalker(plan));
@@ -192,7 +200,8 @@ public class LineageFindRelVisitor exten
}
//ensure that all predecessors are mapped to same load func spec
for(int i=1; i<preds.size(); i++){
- if(!loadFuncSpec.equals(getAssociatedLoadFunc((LogicalRelationalOperator) preds.get(i)))){
+ if(!haveIdenticalCasters(loadFuncSpec,getAssociatedLoadFunc(
+ (LogicalRelationalOperator) preds.get(i)))){
return;
}
}
@@ -218,7 +227,8 @@ public class LineageFindRelVisitor exten
if(funcSpec != null) {
for(int i=1; i<schema.size(); i++){
LogicalFieldSchema fs = schema.getField(i);
- if(! funcSpec.equals(uid2LoadFuncMap.get(fs.uid))){
+ if(! haveIdenticalCasters(funcSpec,
+ uid2LoadFuncMap.get(fs.uid))){
//all uid are not associated with same func spec, there is no
// single func spec that represents all the fields
funcSpec = null;
@@ -420,8 +430,38 @@ public class LineageFindRelVisitor exten
// Since the uid changes for Union, add mappings for new uids to funcspec
LogicalSchema schema = relOp.getSchema();
if(schema != null){
+ // For each output field, checking all the fields being
+ // union-ed(bundled) together and only set the funcspec when ALL
+ // of them come from the same caster
+ //
+ // A = (i,j)
+ // B = (i,j)
+ // C = UNION A, B;
+ // Checking if A.i and B.i have the same caster.
+ // Same for A.j and B.j
+ // A.i and A.j may come from the different casters
for (LogicalFieldSchema logicalFieldSchema : schema.getFields()) {
- addUidLoadFuncToMap(logicalFieldSchema.uid,rel2InputFuncMap.get(relOp));
+ Set<Long> inputs = relOp.getInputUids(logicalFieldSchema.uid);
+ if( inputs.size() == 0 ) {
+ // uid was not changed.
+ // funcspec should be already set. skipping
+ continue;
+ }
+ FuncSpec prevLoadFuncSpec = null, curLoadFuncSpec = null;
+ boolean allSameLoader = true;
+ for(Long inputUid: inputs) {
+ curLoadFuncSpec = uid2LoadFuncMap.get(inputUid) ;
+ if( prevLoadFuncSpec != null
+ && !haveIdenticalCasters(prevLoadFuncSpec,
+ curLoadFuncSpec) ) {
+ allSameLoader = false;
+ break;
+ }
+ prevLoadFuncSpec = curLoadFuncSpec;
+ }
+ if( allSameLoader ) {
+ addUidLoadFuncToMap(logicalFieldSchema.uid,curLoadFuncSpec);
+ }
}
}
}
@@ -483,7 +523,7 @@ public class LineageFindRelVisitor exten
FuncSpec curFuncSpec = uid2LoadFuncMap.get(uid);
if(curFuncSpec == null){
uid2LoadFuncMap.put(uid, loadFuncSpec);
- }else if(! curFuncSpec.equals(loadFuncSpec)){
+ }else if(! haveIdenticalCasters(curFuncSpec,loadFuncSpec)){
String msg = "Bug: uid mapped to two different load functions : " +
curFuncSpec + " and " + loadFuncSpec;
throw new VisitorException(msg,2262, PigException.BUG) ;
@@ -524,7 +564,7 @@ public class LineageFindRelVisitor exten
for(LogicalFieldSchema fs : inputFieldSchemas){
//check if all func spec match
- if(!funcSpec1.equals(uid2LoadFuncMap.get(fs.uid))){
+ if(!haveIdenticalCasters(funcSpec1,uid2LoadFuncMap.get(fs.uid))){
allMatch = false;
}
//check if all inner schema match for use later
@@ -712,7 +752,67 @@ public class LineageFindRelVisitor exten
}
-
+
+ //Copied from POCast.instantiateFunc
+ private Class instantiateCaster(FuncSpec funcSpec) throws VisitorException {
+ if ( funcSpec == null ) {
+ return null;
+ }
+
+ if( func2casterMap.containsKey(funcSpec) ) {
+ return func2casterMap.get(funcSpec);
+ }
+
+ LoadCaster caster = null;
+ Object obj = PigContext.instantiateFuncFromSpec(funcSpec);
+ try {
+ if (obj instanceof LoadFunc) {
+ caster = ((LoadFunc)obj).getLoadCaster();
+ } else if (obj instanceof StreamToPig) {
+ caster = ((StreamToPig)obj).getLoadCaster();
+ } else {
+ throw new VisitorException("Invalid class type " + funcSpec.getClassName(),
+ 2270, PigException.BUG );
+ }
+ } catch (IOException e) {
+ throw new VisitorException("Invalid class type " + funcSpec.getClassName(),
+ 2270, e );
+ }
+
+ Class retval = (caster == null) ? null : caster.getClass();
+ func2casterMap.put(funcSpec, retval);
+ return retval;
+ }
+
+
+ private boolean haveIdenticalCasters(FuncSpec f1, FuncSpec f2) throws VisitorException{
+ if( f1 == null || f2 == null ) {
+ return false;
+ }
+ if( f1.equals(f2) ) {
+ return true;
+ }
+ Class caster1 = instantiateCaster(f1);
+ Class caster2 = instantiateCaster(f2);
+
+ if( caster1 == null || caster2 == null ) {
+ return false;
+ }
+
+ //From PIG-3295
+ //If the class name for LoadCaster are the same, and LoadCaster only has
+ //default constructor, then two LoadCasters are considered equal
+ if( caster1.getCanonicalName().equals(caster2.getCanonicalName()) &&
+ caster1.getConstructors().length == 1 &&
+ caster1.getConstructors()[0].getGenericParameterTypes().length == 0 &&
+ caster2.getConstructors().length == 1 &&
+ caster2.getConstructors()[0].getGenericParameterTypes().length == 0 ) {
+ return true;
+ }
+
+ return false;
+
+ }
}
Modified: pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/TypeCheckingExpVisitor.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/TypeCheckingExpVisitor.java?rev=1537592&r1=1537591&r2=1537592&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/TypeCheckingExpVisitor.java (original)
+++ pig/branches/tez/src/org/apache/pig/newplan/logical/visitor/TypeCheckingExpVisitor.java Thu Oct 31 19:21:28 2013
@@ -27,6 +27,7 @@ import java.util.List;
import java.util.Map;
import org.apache.pig.EvalFunc;
+import org.apache.pig.EvalFunc.SchemaType;
import org.apache.pig.FuncSpec;
import org.apache.pig.PigException;
import org.apache.pig.PigWarning;
@@ -709,6 +710,9 @@ public class TypeCheckingExpVisitor exte
String msg = "Unable to get list of overloaded methods.";
throw new TypeCheckerException(func, msg, errCode, PigException.INPUT, e);
}
+
+ // EvalFunc's schema type
+ SchemaType udfSchemaType = ef.getSchemaType();
/**
* Here is an explanation of the way the matching UDF funcspec will be chosen
@@ -751,7 +755,7 @@ public class TypeCheckingExpVisitor exte
if(funcSpecs!=null && funcSpecs.size()!=0){
//Some function mappings found. Trying to see
//if one of them fits the input schema
- if((matchingSpec = exactMatch(funcSpecs, currentArgSchema, func))==null){
+ if((matchingSpec = exactMatch(funcSpecs, currentArgSchema, func, udfSchemaType))==null){
//Oops, no exact match found. Trying to see if we
//have mappings that we can fit using casts.
notExactMatch = true;
@@ -759,9 +763,9 @@ public class TypeCheckingExpVisitor exte
// try "exact" matching all other fields except the byte array
// fields and if they all exact match and we have only one candidate
// for the byte array cast then that's the matching one!
- if((matchingSpec = exactMatchWithByteArrays(funcSpecs, currentArgSchema, func))==null){
+ if((matchingSpec = exactMatchWithByteArrays(funcSpecs, currentArgSchema, func, udfSchemaType))==null){
// "exact" match with byte arrays did not work - try best fit match
- if((matchingSpec = bestFitMatchWithByteArrays(funcSpecs, currentArgSchema, func)) == null) {
+ if((matchingSpec = bestFitMatchWithByteArrays(funcSpecs, currentArgSchema, func, udfSchemaType)) == null) {
int errCode = 1045;
String msg = "Could not infer the matching function for "
+ func.getFuncSpec()
@@ -770,7 +774,7 @@ public class TypeCheckingExpVisitor exte
throw new TypeCheckerException(func, msg, errCode, PigException.INPUT);
}
}
- } else if ((matchingSpec = bestFitMatch(funcSpecs, currentArgSchema)) == null) {
+ } else if ((matchingSpec = bestFitMatch(funcSpecs, currentArgSchema, udfSchemaType)) == null) {
// Either no byte arrays found or there are byte arrays
// but only one mapping exists.
// However, we could not find a match as there were either
@@ -801,7 +805,7 @@ public class TypeCheckingExpVisitor exte
matchingSpec.setCtorArgs(func.getFuncSpec().getCtorArgs());
}
func.setFuncSpec(matchingSpec);
- insertCastsForUDF(func, currentArgSchema, matchingSpec.getInputArgsSchema());
+ insertCastsForUDF(func, currentArgSchema, matchingSpec.getInputArgsSchema(), udfSchemaType);
}
}
@@ -816,18 +820,20 @@ public class TypeCheckingExpVisitor exte
* input schema
* @param func -
* udf expression
+ * @param udfSchemaType -
+ * schema type of the udf
* @return the funcSpec that supports the schema that is best suited to s.
* The best suited schema is one that has the lowest score as
* returned by fitPossible().
* @throws VisitorException
*/
private FuncSpec bestFitMatchWithByteArrays(List<FuncSpec> funcSpecs,
- Schema s, UserFuncExpression func) throws VisitorException {
+ Schema s, UserFuncExpression func, SchemaType udfSchemaType) throws VisitorException {
List<Pair<Long, FuncSpec>> scoreFuncSpecList = new ArrayList<Pair<Long,FuncSpec>>();
for (Iterator<FuncSpec> iterator = funcSpecs.iterator(); iterator
.hasNext();) {
FuncSpec fs = iterator.next();
- long score = fitPossible(s, fs.getInputArgsSchema());
+ long score = fitPossible(s, fs.getInputArgsSchema(), udfSchemaType);
if (score != INF) {
scoreFuncSpecList.add(new Pair<Long, FuncSpec>(score, fs));
}
@@ -926,14 +932,15 @@ public class TypeCheckingExpVisitor exte
* @param funcSpecs - mappings provided by udf
* @param s - input schema
* @param func - UserFuncExpression for which matching is requested
+ * @param udfSchemaType - schema type of the udf
* @return the matching spec if found else null
* @throws FrontendException
*/
private FuncSpec exactMatchWithByteArrays(List<FuncSpec> funcSpecs,
- Schema s, UserFuncExpression func) throws FrontendException {
+ Schema s, UserFuncExpression func, SchemaType udfSchemaType) throws FrontendException {
// exact match all fields except byte array fields
// ignore byte array fields for matching
- return exactMatchHelper(funcSpecs, s, func, true);
+ return exactMatchHelper(funcSpecs, s, func, udfSchemaType, true);
}
/**
@@ -943,13 +950,14 @@ public class TypeCheckingExpVisitor exte
* @param funcSpecs - mappings provided by udf
* @param s - input schema
* @param func - UserFuncExpression for which matching is requested
+ * @param udfSchemaType - schema type of the user defined function
* @return the matching spec if found else null
* @throws FrontendException
*/
private FuncSpec exactMatch(List<FuncSpec> funcSpecs, Schema s,
- UserFuncExpression func) throws FrontendException {
+ UserFuncExpression func, SchemaType udfSchemaType) throws FrontendException {
// exact match all fields, don't ignore byte array fields
- return exactMatchHelper(funcSpecs, s, func, false);
+ return exactMatchHelper(funcSpecs, s, func, udfSchemaType, false);
}
/**
@@ -957,18 +965,19 @@ public class TypeCheckingExpVisitor exte
* be obtained by inserting a set of casts to the input schema
* @param funcSpecs - mappings provided by udf
* @param s - input schema
+ * @param udfSchemaType - schema type of the udf
* @return the funcSpec that supports the schema that is best suited
* to s. The best suited schema is one that has the
* lowest score as returned by fitPossible().
*/
- private FuncSpec bestFitMatch(List<FuncSpec> funcSpecs, Schema s) {
+ private FuncSpec bestFitMatch(List<FuncSpec> funcSpecs, Schema s, SchemaType udfSchemaType) {
FuncSpec matchingSpec = null;
long score = INF;
long prevBestScore = Long.MAX_VALUE;
long bestScore = Long.MAX_VALUE;
for (Iterator<FuncSpec> iterator = funcSpecs.iterator(); iterator.hasNext();) {
FuncSpec fs = iterator.next();
- score = fitPossible(s,fs.getInputArgsSchema());
+ score = fitPossible(s,fs.getInputArgsSchema(), udfSchemaType);
if(score!=INF && score<=bestScore){
matchingSpec = fs;
prevBestScore = bestScore;
@@ -1037,18 +1046,19 @@ public class TypeCheckingExpVisitor exte
* @param funcSpecs - mappings provided by udf
* @param s - input schema
* @param func user defined function
+ * @param udfSchemaType - schema type of the user defined function
* @param ignoreByteArrays - flag for whether the exact match is to computed
* after ignoring bytearray (if true) or without ignoring bytearray (if false)
* @return the matching spec if found else null
* @throws FrontendException
*/
private FuncSpec exactMatchHelper(List<FuncSpec> funcSpecs, Schema s,
- UserFuncExpression func, boolean ignoreByteArrays)
+ UserFuncExpression func, SchemaType udfSchemaType, boolean ignoreByteArrays)
throws FrontendException {
List<FuncSpec> matchingSpecs = new ArrayList<FuncSpec>();
for (Iterator<FuncSpec> iterator = funcSpecs.iterator(); iterator.hasNext();) {
FuncSpec fs = iterator.next();
- if (schemaEqualsForMatching(s, fs.getInputArgsSchema(), ignoreByteArrays)) {
+ if (schemaEqualsForMatching(s, fs.getInputArgsSchema(), udfSchemaType, ignoreByteArrays)) {
matchingSpecs.add(fs);
}
@@ -1087,7 +1097,7 @@ public class TypeCheckingExpVisitor exte
* @throws FrontendException
*/
public static boolean schemaEqualsForMatching(Schema inputSchema,
- Schema udfSchema, boolean ignoreByteArrays) throws FrontendException {
+ Schema udfSchema, SchemaType udfSchemaType, boolean ignoreByteArrays) throws FrontendException {
// If both of them are null, they are equal
@@ -1108,25 +1118,28 @@ public class TypeCheckingExpVisitor exte
// fix that!
udfSchema = Util.fixSchemaAddTupleInBag(udfSchema);
-
- if (inputSchema.size() != udfSchema.size())
+ if ((udfSchemaType == SchemaType.NORMAL) && (inputSchema.size() != udfSchema.size()))
+ return false;
+ if ((udfSchemaType == SchemaType.VARARG) && inputSchema.size() < udfSchema.size())
return false;
Iterator<FieldSchema> i = inputSchema.getFields().iterator();
Iterator<FieldSchema> j = udfSchema.getFields().iterator();
+ FieldSchema udfFieldSchema = null;
while (i.hasNext()) {
FieldSchema inputFieldSchema = i.next();
- FieldSchema udfFieldSchema = j.next();
if(inputFieldSchema == null)
return false;
-
+ //if there's no more UDF field: take the last one which is the vararg field
+ udfFieldSchema = j.hasNext() ? j.next() : udfFieldSchema;
+
if(ignoreByteArrays && inputFieldSchema.type == DataType.BYTEARRAY) {
continue;
}
-
+
if (inputFieldSchema.type != udfFieldSchema.type) {
return false;
}
@@ -1293,13 +1306,17 @@ public class TypeCheckingExpVisitor exte
* Unallowed casts are returned a dist of INFINITY.
* @param s1
* @param s2
+ * @param s2Type
* @return
*/
- private long fitPossible(Schema s1, Schema s2) {
+ private long fitPossible(Schema s1, Schema s2, SchemaType s2Type) {
if(s1==null || s2==null) return INF;
List<FieldSchema> sFields = s1.getFields();
List<FieldSchema> fsFields = s2.getFields();
- if(sFields.size()!=fsFields.size())
+
+ if((s2Type == SchemaType.NORMAL) && (sFields.size()!=fsFields.size()))
+ return INF;
+ if((s2Type == SchemaType.VARARG) && (sFields.size() < fsFields.size()))
return INF;
long score = 0;
int castCnt=0;
@@ -1315,8 +1332,10 @@ public class TypeCheckingExpVisitor exte
// of this function
if (sFS.type == DataType.BYTEARRAY)
continue;
-
- FieldSchema fsFS = fsFields.get(i);
+
+ //if we get to the vararg field (if defined) : take it repeatedly
+ FieldSchema fsFS = ((s2Type == SchemaType.VARARG) && i >= s2.size()) ?
+ fsFields.get(s2.size() - 1) : fsFields.get(i);
if(DataType.isSchemaType(sFS.type)){
if(!FieldSchema.equals(sFS, fsFS, false, true))
@@ -1333,7 +1352,7 @@ public class TypeCheckingExpVisitor exte
return score * castCnt;
}
- private void insertCastsForUDF(UserFuncExpression func, Schema fromSch, Schema toSch)
+ private void insertCastsForUDF(UserFuncExpression func, Schema fromSch, Schema toSch, SchemaType toSchType)
throws FrontendException {
List<FieldSchema> fsLst = fromSch.getFields();
List<FieldSchema> tsLst = toSch.getFields();
@@ -1341,8 +1360,10 @@ public class TypeCheckingExpVisitor exte
int i=-1;
for (FieldSchema fFSch : fsLst) {
++i;
- FieldSchema tFSch = tsLst.get(i);
- if(fFSch.type==tFSch.type) {
+ //if we get to the vararg field (if defined) : take it repeatedly
+ FieldSchema tFSch = ((toSchType == SchemaType.VARARG) && i >= tsLst.size()) ?
+ tsLst.get(tsLst.size() - 1) : tsLst.get(i);
+ if (fFSch.type == tFSch.type) {
continue;
}
insertCast(func, Util.translateFieldSchema(tFSch), args.get(i));