From pig-commitsmichael55@example.net Mon Nov 02 19:35:07 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 67358 invoked from network); 2 Nov 2009 19:35:07 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 2 Nov 2009 19:35:07 -0000 Received: (qmail 55935 invoked by uid 500); 2 Nov 2009 19:35:07 -0000 Delivered-To: michael55@example.net Received: (qmail 55900 invoked by uid 500); 2 Nov 2009 19:35:06 -0000 Mailing-List: contact michael55@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:michael55@example.net> List-Unsubscribe: <mailto:michael55@example.net> List-Post: <mailto:michael55@example.net> List-Id: <pig-commits.hadoop.apache.org> Reply-To: michael55@example.net Delivered-To: mailing list michael55@example.net Received: (qmail 55891 invoked by uid 500); 2 Nov 2009 19:35:06 -0000 Delivered-To: michael55@example.net Received: (qmail 55888 invoked by uid 99); 2 Nov 2009 19:35:06 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 19:35:06 +0000 X-ASF-Spam-Status: No, hits=-1.8 required=5.0 tests=AWL,BAYES_00,MIME_QP_LONG_LINE X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 19:35:03 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 7688016E09; Mon, 2 Nov 2009 19:34:43 +0000 (GMT) MIME-Version: 1.0 Content-Type: text/plain; charset="utf-8" Content-Transfer-Encoding: quoted-printable From: Apache Wiki <michael55@example.net> To: Apache Wiki <michael55@example.net> Date: Mon, 02 Nov 2009 19:34:43 -0000 Message-ID: <michael55@example.net> Subject: =?utf-8?q?=5BPig_Wiki=5D_Update_of_=22LoadStoreRedesignProposal=22_by_Pra?= =?utf-8?q?deepKamath?= Dear Wiki user, You have subscribed to a wiki page or wiki category on "Pig Wiki" for chang= e notification. The "LoadStoreRedesignProposal" page has been changed by PradeepKamath. http://wiki.apache.org/pig/LoadStoreRedesignProposal?action=3Ddiff&rev1=3D8= &rev2=3D9 -------------------------------------------------- '''!LoadFunc''' = {{{ + = /** * This interface is used to implement functions to parse records * from a dataset. */ - interface LoadFunc { + public interface LoadFunc { + /** + * This method is called by the Pig runtime in the front end to conve= rt the + * input location to an absolute path if the location is relative. The + * loadFunc implementation is free to choose how it converts a relati= ve = + * location to an absolute location since this may depend on what the= location + * string represent (hdfs path or some other data source) + * = + * @param location location as provided in the "load" statement of th= e script + * @param curDir the current working direction based on any "cd" stat= ements + * in the script before the "load" statement + * @return the absolute location based on the arguments passed + * @throws IOException if the conversion is not possible + */ + String relativeToAbsolutePath(String location, String curDir) throws = IOException; = /** * Communicate to the loader the load string used in Pig Latin to ref= er to the = - * object(s) being loaded. That is, if the PL script is - * <b>A =3D load 'bla'</b> - * then 'bla' is the load string. In general Pig expects these to be - * a path name, a glob, or a URI. If there is no URI scheme present, - * Pig will assume it is a file name. This will be = - * called during planning on the front end, not during execution on - * the backend. - * @param location Location indicated in load statement. + * object(s) being loaded. The location string passed to the LoadFun= c here = + * is the return value of {@link LoadFunc#relativeToAbsolutePath(Stri= ng, String)} + * = + * This method will be called in the backend multiple times. Implemen= tations + * should bear in mind that this method is called multiple times and = should + * ensure there are no inconsistent side effects due to the multiple = calls. + * = + * @param location Location as returned by = + * {@link LoadFunc#relativeToAbsolutePath(String, String)}. + * @param job the {@link Job} object * @throws IOException if the location is not valid. */ - void setLocation(String location) throws IOException; + void setLocation(String location, Job job) throws IOException; = /** + * This will be called during planning on the front end. This is the - * Return the InputFormat associated with this loader. This will be - * called during planning on the front end. The LoadFunc need not - * carry the InputFormat information to the backend, as it will - * be provided with the appropriate RecordReader there. This is the * instance of InputFormat (rather than the class name) because the = * load function may need to instantiate the InputFormat in order = * to control how it is constructed. + * @return the InputFormat associated with this loader. + * @throws IOException if there is an exception during InputFormat = + * construction */ - InputFormat getInputFormat(); + InputFormat getInputFormat() throws IOException; = /** + * This will be called on the front end during planning and not on th= e back = + * end during execution. - * Return the LoadCaster associated with this loader. Returning + * @return the {@link LoadCaster} associated with this loader. Return= ing null = - * null indicates that casts from byte array are not supported + * indicates that casts from byte array are not supported for this lo= ader. = - * for this loader. This will be called on the front end during - * planning and not on the back end during execution. + * construction + * @throws IOException if there is an exception during LoadCaster = */ - LoadCaster getLoadCaster(); + LoadCaster getLoadCaster() throws IOException; = /** * Initializes LoadFunc for reading data. This will be called during= execution * before any calls to getNext. The RecordReader needs to be passed = here because * it has been instantiated for a particular InputSplit. - * @param reader RecordReader to be used by this instance of the Load= Func + * @param reader {@link RecordReader} to be used by this instance of = the LoadFunc + * @param split The input {@link PigSplit} to process + * @throws IOException if there is an exception during initialization */ + void prepareToRead(RecordReader reader, PigSplit split) throws IOExce= ption; - void prepareToRead(RecordReader reader); - = - /** - * Called after all reading is finished. - */ - void doneReading(); = /** * Retrieves the next tuple to be processed. * @return the next tuple to be processed or null if there are no mor= e tuples * to be processed. - * @throws IOException + * @throws IOException if there is an exception while retrieving the = next + * tuple */ Tuple getNext() throws IOException; = @@ -119, +136 @@ * If a given loader does not implement this interface, it will be assume= d that it * is unable to provide metadata about the associated data. */ - interface LoadMetadata { + public interface LoadMetadata { = /** + * Get a schema for the data to be loaded. = + * @param location Location as returned by = + * {@link LoadFunc#relativeToAbsolutePath(String, String)} + * @param conf The {@link Configuration} object = - * Get a schema for the data to be loaded. This schema should repres= ent + * @return schema for the data to be loaded. This schema should repre= sent * all tuples of the returned data. If the schema is unknown or it is * not possible to return a schema that represents all returned data, * then null should be returned. + * @throws IOException if an exception occurs while determining the s= chema */ - ResourceSchema getSchema(); + ResourceSchema getSchema(String location, Configuration conf) throws = + IOException; = /** * Get statistics about the data to be loaded. If no statistics are * available, then null should be returned. + * @param location Location as returned by = + * {@link LoadFunc#relativeToAbsolutePath(String, String)} + * @param conf The {@link Configuration} object + * @return statistics about the data to be loaded. If no statistics = are + * available, then null should be returned. + * @throws IOException if an exception occurs while retrieving statis= tics */ - ResourceStatistics getStatistics(); + ResourceStatistics getStatistics(String location, Configuration conf) = + throws IOException; = /** * Find what columns are partition keys for this input. - * This function assumes that setLocation has already been called. + * @param location Location as returned by = + * {@link LoadFunc#relativeToAbsolutePath(String, String)} + * @param conf The {@link Configuration} object * @return array of field names of the partition keys. + * @throws IOException if an exception occurs while retrieving partit= ion keys */ - String>[] getPartitionKeys(); + String[] getPartitionKeys(String location, Configuration conf) = + throws IOException; = /** * Set the filter for partitioning. It is assumed that this filter @@ -484, +518 @@ Sept 29 2009, Gates * Added answer for open question 1. Added and answered open questions 2= and 3. = + Nov 2 2009, Pradeep Kamath + = + In LoadFunc: + * Added relativeToAbsolutePath() method in LoadFunc per http://issues.ap= ache.org/jira/browse/PIG-879?focusedCommentId=3D12768818&page=3Dcom.atlassi= an.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12768818 + * Changed comments in setLocation regarding the location passed - the lo= cation will now be the return value of relativeToAbsolutePath() + * setLocation() now also takes a Job argument since the main purpose of = this call is to an opportunity to the LoadFunc implementation to communicat= e the input location to underlying InputFormat. InputFormat implementations= inturn seem to be storing this information inthe Job. For example, FileInp= utFormat has the following static method to set the input location: setInpu= tPaths(JobConf conf, String commaSeparatedPaths) ; + * All methods now can throw IOException - this keeps the interface more = flexible for exception cases + = + In LoadMetadata: + * getSchema(), getStatistics() and getPartitionKeys() methods now take a= location and Configuration argument so that the implementation can use tha= t information in returning the information requested. +=20 From pig-commitsmichael55@example.net Mon Nov 02 19:46:47 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 72554 invoked from network); 2 Nov 2009 19:46:47 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 2 Nov 2009 19:46:47 -0000 Received: (qmail 76675 invoked by uid 500); 2 Nov 2009 19:46:47 -0000 Delivered-To: michael55@example.net Received: (qmail 76636 invoked by uid 500); 2 Nov 2009 19:46:47 -0000 Mailing-List: contact michael55@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:michael55@example.net> List-Unsubscribe: <mailto:michael55@example.net> List-Post: <mailto:michael55@example.net> List-Id: <pig-commits.hadoop.apache.org> Reply-To: michael55@example.net Delivered-To: mailing list michael55@example.net Received: (qmail 76627 invoked by uid 500); 2 Nov 2009 19:46:47 -0000 Delivered-To: michael55@example.net Received: (qmail 76623 invoked by uid 99); 2 Nov 2009 19:46:47 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 19:46:47 +0000 X-ASF-Spam-Status: No, hits=-2.5 required=5.0 tests=AWL,BAYES_00 X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 19:46:44 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id A894116E09 for <michael55@example.net>; Mon, 2 Nov 2009 19:46:24 +0000 (GMT) MIME-Version: 1.0 Content-Type: text/plain; charset="utf-8" Content-Transfer-Encoding: quoted-printable From: Apache Wiki <michael55@example.net> To: Apache Wiki <michael55@example.net> Date: Mon, 02 Nov 2009 19:46:24 -0000 Message-ID: <michael55@example.net> Subject: =?utf-8?q?=5BPig_Wiki=5D_Update_of_=22PigErrorHandlingFunctionalSpecifica?= =?utf-8?q?tion=22_by_ankit=2Emodi?= Dear Wiki user, You have subscribed to a wiki page or wiki category on "Pig Wiki" for chang= e notification. The "PigErrorHandlingFunctionalSpecification" page has been changed by anki= t.modi. http://wiki.apache.org/pig/PigErrorHandlingFunctionalSpecification?action= =3Ddiff&rev1=3D130&rev2=3D131 -------------------------------------------------- ||1106||Merge join is possible only for simple column or '*' join keys wh= en using <funcspec> as the loader || ||1107||Try to merge incompatible types (eg. numerical type vs non-numeir= cal type)|| ||1108||Duplicated schema|| + ||1109||Input ( <input alias> ) on which outer join is desired should hav= e a valid schema|| ||2000||Internal error. Mismatch in group by arities. Expected: <schema>.= Found: <schema>|| ||2001||Unable to clone plan before compiling|| ||2002||The output file(s): <filename> already exists|| @@ -467, +468 @@ 1. February 11, 2009: Updated "Compendium of error messages" to includ= e new error codes (2116 through 2121, 6015 and 6016) 1. February 12, 2009: Updated "Compendium of error messages" to includ= e new error code 2122 1. April 10, 2009: Updated "Compendium of error messages" to replace = error code 2110 + 1. November 2, 2009: Updated "Compendium of error messages" to include= new error code 1109 =3D=3D References =3D=3D = 1. <<Anchor(cookbook)>> "Pig Developer Cookbook" October 21, 2008, http:= //wiki.apache.org/pig/PigDeveloperCookbook From pig-commitsmichael55@example.net Mon Nov 02 20:05:10 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 82350 invoked from network); 2 Nov 2009 20:05:10 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 2 Nov 2009 20:05:10 -0000 Received: (qmail 6018 invoked by uid 500); 2 Nov 2009 20:05:10 -0000 Delivered-To: michael55@example.net Received: (qmail 5968 invoked by uid 500); 2 Nov 2009 20:05:09 -0000 Mailing-List: contact michael55@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:michael55@example.net> List-Unsubscribe: <mailto:michael55@example.net> List-Post: <mailto:michael55@example.net> List-Id: <pig-commits.hadoop.apache.org> Reply-To: michael55@example.net Delivered-To: mailing list michael55@example.net Received: (qmail 5959 invoked by uid 500); 2 Nov 2009 20:05:09 -0000 Delivered-To: michael55@example.net Received: (qmail 5956 invoked by uid 99); 2 Nov 2009 20:05:09 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 20:05:09 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 20:05:06 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id CF04C23888D8; Mon, 2 Nov 2009 20:04:45 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r832084 - in /hadoop/pig/trunk: CHANGES.txt src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java test/org/apache/pig/test/TestCombiner.java Date: Mon, 02 Nov 2009 20:04:45 -0000 To: michael55@example.net From: michael55@example.net X-Mailer: svnmailer-1.0.8 Message-Id: <michael55@example.net> X-Virus-Checked: Checked by ClamAV on apache.org Author: pradeepkth Date: Mon Nov 2 20:04:45 2009 New Revision: 832084 URL: http://svn.apache.org/viewvc?rev=832084&view=rev Log: PIG-1030: explain and dump not working with two UDFs inside inner plan of foreach (rding via pradeepkth) Modified: hadoop/pig/trunk/CHANGES.txt hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java hadoop/pig/trunk/test/org/apache/pig/test/TestCombiner.java Modified: hadoop/pig/trunk/CHANGES.txt URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=832084&r1=832083&r2=832084&view=diff ============================================================================== --- hadoop/pig/trunk/CHANGES.txt (original) +++ hadoop/pig/trunk/CHANGES.txt Mon Nov 2 20:04:45 2009 @@ -109,6 +109,9 @@ BUG FIXES +PIG-1030: explain and dump not working with two UDFs inside inner plan of +foreach (rding via pradeepkth) + PIG-1048: inner join using 'skewed' produces multiple rows for keys with single row in both input relations (sriranjan via gates) Modified: hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java?rev=832084&r1=832083&r2=832084&view=diff ============================================================================== --- hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java (original) +++ hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java Mon Nov 2 20:04:45 2009 @@ -712,6 +712,7 @@ // we apparently have seen a PODistinct before, so lets not // combine. sawNonAlgebraic = true; + return; } // check that this distinct is the only input to an agg // We could have the following two cases @@ -748,6 +749,16 @@ PhysicalOperator leaf = mPlan.getLeaves().get(0); // the leaf has to be a POUserFunc (need not be algebraic) if(leaf instanceof POUserFunc) { + + // we want to combine only in the case where there is only + // one PODistinct which is the only input to an agg. + // Do not combine if there are additional inputs. + List<PhysicalOperator> preds = mPlan.getPredecessors(leaf); + if (preds.size() > 1) { + sawNonAlgebraic = true; + return; + } + List<PhysicalOperator> immediateSuccs = mPlan.getSuccessors(distinct); if(immediateSuccs.size() == 1 && immediateSuccs.get(0) instanceof POProject) { if(checkSuccessorIsLeaf(leaf, immediateSuccs.get(0))) { // script 1 above Modified: hadoop/pig/trunk/test/org/apache/pig/test/TestCombiner.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestCombiner.java?rev=832084&r1=832083&r2=832084&view=diff ============================================================================== --- hadoop/pig/trunk/test/org/apache/pig/test/TestCombiner.java (original) +++ hadoop/pig/trunk/test/org/apache/pig/test/TestCombiner.java Mon Nov 2 20:04:45 2009 @@ -33,6 +33,7 @@ import junit.framework.Assert; import junit.framework.TestCase; +import org.apache.pig.EvalFunc; import org.apache.pig.ExecType; import org.apache.pig.PigServer; import org.apache.pig.builtin.PigStorage; @@ -341,4 +342,53 @@ } } + public static class JiraPig1030 extends EvalFunc<String> { + + public String exec(Tuple input) throws IOException { + return ""; + } + } + + @Test + public void testJiraPig1030() { + // test that combiner is NOT invoked when + // one of the elements in the foreach generate + // has a non-algebraic UDF that have multiple inputs + // (one of them is distinct). + + String input[] = { + "pig1\t18\t2.1", + "pig2\t24\t3.3", + "pig5\t45\t2.4", + "pig1\t18\t2.1", + "pig1\t19\t2.1", + "pig2\t24\t4.5", + "pig1\t20\t3.1" }; + + try { + Util.createInputFile(cluster, "forEachNoCombinerInput.txt", input); + PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties()); + pigServer.registerQuery("a = load 'forEachNoCombinerInput.txt' as (name:chararray, age:int, gpa:double);"); + pigServer.registerQuery("b = group a all;"); + pigServer.registerQuery("c = foreach b {" + + " d = distinct a.age;" + + " generate group, " + JiraPig1030.class.getName() + "(d, 0);};"); + + // make sure there isn't a combine plan in the explain output + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + PrintStream ps = new PrintStream(baos); + pigServer.explain("c", ps); + assertFalse(baos.toString().matches("(?si).*combine plan.*")); + } catch (Exception e) { + e.printStackTrace(); + Assert.fail(); + } finally { + try { + Util.deleteFile(cluster, "forEachNoCombinerInput.txt"); + } catch (IOException e) { + e.printStackTrace(); + Assert.fail(); + } + } + } } From pig-commitsmichael55@example.net Mon Nov 02 20:20:32 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 90318 invoked from network); 2 Nov 2009 20:20:32 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 2 Nov 2009 20:20:32 -0000 Received: (qmail 30222 invoked by uid 500); 2 Nov 2009 20:20:32 -0000 Delivered-To: michael55@example.net Received: (qmail 30183 invoked by uid 500); 2 Nov 2009 20:20:32 -0000 Mailing-List: contact michael55@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:michael55@example.net> List-Unsubscribe: <mailto:michael55@example.net> List-Post: <mailto:michael55@example.net> List-Id: <pig-commits.hadoop.apache.org> Reply-To: michael55@example.net Delivered-To: mailing list michael55@example.net Received: (qmail 30173 invoked by uid 500); 2 Nov 2009 20:20:32 -0000 Delivered-To: michael55@example.net Received: (qmail 30170 invoked by uid 99); 2 Nov 2009 20:20:32 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 20:20:32 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 20:20:28 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 03D1E23888D8; Mon, 2 Nov 2009 20:20:07 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r832086 - in /hadoop/pig/trunk: ./ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relational... Date: Mon, 02 Nov 2009 20:20:06 -0000 To: michael55@example.net From: michael55@example.net X-Mailer: svnmailer-1.0.8 Message-Id: <michael55@example.net> X-Virus-Checked: Checked by ClamAV on apache.org Author: pradeepkth Date: Mon Nov 2 20:20:05 2009 New Revision: 832086 URL: http://svn.apache.org/viewvc?rev=832086&view=rev Log: PIG-1035: support for skewed outer join (sriranjan via pradeepkth) Added: hadoop/pig/trunk/src/org/apache/pig/impl/util/CompilerUtils.java Modified: hadoop/pig/trunk/CHANGES.txt hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSkewedJoin.java hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt hadoop/pig/trunk/test/org/apache/pig/test/TestJoin.java hadoop/pig/trunk/test/org/apache/pig/test/TestSkewedJoin.java Modified: hadoop/pig/trunk/CHANGES.txt URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=832086&r1=832085&r2=832086&view=diff ============================================================================== --- hadoop/pig/trunk/CHANGES.txt (original) +++ hadoop/pig/trunk/CHANGES.txt Mon Nov 2 20:20:05 2009 @@ -109,6 +109,8 @@ BUG FIXES +PIG-1035: support for skewed outer join (sriranjan via pradeepkth) + PIG-1030: explain and dump not working with two UDFs inside inner plan of foreach (rding via pradeepkth) Modified: hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java?rev=832086&r1=832085&r2=832086&view=diff ============================================================================== --- hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java (original) +++ hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java Mon Nov 2 20:20:05 2009 @@ -91,6 +91,7 @@ import org.apache.pig.impl.plan.PlanException; import org.apache.pig.impl.plan.VisitorException; import org.apache.pig.impl.plan.CompilationMessageCollector.MessageType; +import org.apache.pig.impl.util.CompilerUtils; import org.apache.pig.impl.util.MultiMap; import org.apache.pig.impl.util.ObjectSerializer; import org.apache.pig.impl.util.Pair; @@ -1495,7 +1496,7 @@ pkg.setKeyType(type); pkg.setResultType(DataType.TUPLE); pkg.setNumInps(2); - boolean[] inner = {true, true}; + boolean [] inner = op.getInnerFlags(); pkg.setInner(inner); pkg.visit(this); compiledInputs = new MapReduceOper[] {curMROp}; @@ -1504,23 +1505,22 @@ List<PhysicalPlan> eps = new ArrayList<PhysicalPlan>(); List<Boolean> flat = new ArrayList<Boolean>(); - PhysicalPlan ep = new PhysicalPlan(); - POProject prj = new POProject(new OperatorKey(scope,nig.getNextNodeId(scope))); - prj.setColumn(1); - prj.setOverloaded(false); - prj.setResultType(DataType.BAG); - ep.add(prj); - eps.add(ep); - flat.add(true); - - ep = new PhysicalPlan(); - prj = new POProject(new OperatorKey(scope,nig.getNextNodeId(scope))); - prj.setColumn(2); - prj.setOverloaded(false); - prj.setResultType(DataType.BAG); - ep.add(prj); - eps.add(ep); - flat.add(true); + PhysicalPlan ep; + // Add corresponding POProjects + for (int i=0; i < 2; i++ ) { + ep = new PhysicalPlan(); + POProject prj = new POProject(new OperatorKey(scope,nig.getNextNodeId(scope))); + prj.setColumn(i+1); + prj.setOverloaded(false); + prj.setResultType(DataType.BAG); + ep.add(prj); + eps.add(ep); + if (!inner[i]) { + // Add an empty bag for outer join + CompilerUtils.addEmptyBagOuterJoin(ep, op.getSchema(i)); + } + flat.add(true); + } POForEach fe = new POForEach(new OperatorKey(scope,nig.getNextNodeId(scope)), -1, eps, flat); fe.setResultType(DataType.TUPLE); Modified: hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java?rev=832086&r1=832085&r2=832086&view=diff ============================================================================== --- hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java (original) +++ hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java Mon Nov 2 20:20:05 2009 @@ -62,6 +62,7 @@ import org.apache.pig.impl.plan.VisitorException; +import org.apache.pig.impl.util.CompilerUtils; import org.apache.pig.impl.util.LinkedMultiMap; import org.apache.pig.impl.util.MultiMap; @@ -834,7 +835,7 @@ POSkewedJoin skj; try { skj = new POSkewedJoin(new OperatorKey(scope,nodeGen.getNextNodeId(scope)),loj.getRequestedParallelism(), - inp); + inp, loj.getInnerFlags()); skj.setJoinPlans(joinPlans); } catch (Exception e) { @@ -843,6 +844,30 @@ throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e); } skj.setResultType(DataType.TUPLE); + + boolean[] innerFlags = loj.getInnerFlags(); + for (int i=0; i < inputs.size(); i++) { + LogicalOperator op = inputs.get(i); + if (!innerFlags[i]) { + try { + Schema s = op.getSchema(); + // if the schema cannot be determined + if (s == null) { + throw new FrontendException(); + } + skj.addSchema(s); + } catch (FrontendException e) { + int errCode = 2015; + String msg = "Couldn't set the schema for outer join" ; + throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e); + } + } else { + // This will never be retrieved. It just guarantees that the index will be valid when + // MRCompiler is trying to read the schema + skj.addSchema(null); + } + } + currentPlan.add(skj); for (LogicalOperator op : inputs) { @@ -1045,8 +1070,8 @@ Schema inputSchema = null; try { inputSchema = joinInput.getSchema(); - - + + if(inputSchema == null) { int errCode = 1105; String msg = "Input (" + joinInput.getAlias() + ") " + @@ -1059,71 +1084,7 @@ throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e); } - // we currently have POProject[bag] as the only operator in the plan - // If the bag is an empty bag, we should replace - // it with a bag with one tuple with null fields so that when we flatten - // we do not drop records (flatten will drop records if the bag is left - // as an empty bag) and actually project nulls for the fields in - // the empty bag - - // So we need to get to the following state: - // POProject[Bag] - // \ - // POUserFunc["IsEmpty()"] Const[Bag](bag with null fields) - // \ | POProject[Bag] - // \ | / - // POBinCond - - POProject relationProject = (POProject) fePlan.getRoots().get(0); - try { - - // condition of the bincond - POProject relationProjectForIsEmpty = relationProject.clone(); - fePlan.add(relationProjectForIsEmpty); - String scope = relationProject.getOperatorKey().scope; - FuncSpec isEmptySpec = new FuncSpec(IsEmpty.class.getName()); - Object f = PigContext.instantiateFuncFromSpec(isEmptySpec); - POUserFunc isEmpty = new POUserFunc(new OperatorKey(scope, NodeIdGenerator.getGenerator(). - getNextNodeId(scope)), -1, null, isEmptySpec, (EvalFunc) f); - isEmpty.setResultType(DataType.BOOLEAN); - fePlan.add(isEmpty); - fePlan.connect(relationProjectForIsEmpty, isEmpty); - - // lhs of bincond (const bag with null fields) - ConstantExpression ce = new ConstantExpression(new OperatorKey(scope, - NodeIdGenerator.getGenerator().getNextNodeId(scope))); - // the following should give a tuple with the - // required number of nulls - Tuple t = TupleFactory.getInstance().newTuple(inputSchema.size()); - for(int i = 0; i < inputSchema.size(); i++) { - t.set(i, null); - } - List<Tuple> bagContents = new ArrayList<Tuple>(1); - bagContents.add(t); - DataBag bg = new NonSpillableDataBag(bagContents); - ce.setValue(bg); - ce.setResultType(DataType.BAG); - //this operator doesn't have any predecessors - fePlan.add(ce); - - //rhs of bincond is the original project - // let's set up the bincond now - POBinCond bincond = new POBinCond(new OperatorKey(scope, - NodeIdGenerator.getGenerator().getNextNodeId(scope))); - bincond.setCond(isEmpty); - bincond.setLhs(ce); - bincond.setRhs(relationProject); - bincond.setResultType(DataType.BAG); - fePlan.add(bincond); - - fePlan.connect(isEmpty, bincond); - fePlan.connect(ce, bincond); - fePlan.connect(relationProject, bincond); - - } catch (Exception e) { - throw new PlanException("Error setting up outerjoin", e); - } - + CompilerUtils.addEmptyBagOuterJoin(fePlan, inputSchema); } Modified: hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSkewedJoin.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSkewedJoin.java?rev=832086&r1=832085&r2=832086&view=diff ============================================================================== --- hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSkewedJoin.java (original) +++ hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSkewedJoin.java Mon Nov 2 20:20:05 2009 @@ -18,6 +18,7 @@ package org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators; +import java.util.ArrayList; import java.util.List; import org.apache.commons.logging.Log; @@ -26,6 +27,7 @@ import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan; import org.apache.pig.data.DataType; +import org.apache.pig.impl.logicalLayer.schema.Schema; import org.apache.pig.impl.plan.OperatorKey; import org.apache.pig.impl.plan.VisitorException; import org.apache.pig.impl.util.MultiMap; @@ -42,6 +44,10 @@ private static final long serialVersionUID = 1L; + private boolean[] mInnerFlags; + + // The schema is used only by the MRCompiler to support outer join + transient private List<Schema> inputSchema = new ArrayList<Schema>(); transient private static Log log = LogFactory.getLog(POSkewedJoin.class); @@ -51,19 +57,30 @@ private MultiMap<PhysicalOperator, PhysicalPlan> mJoinPlans; public POSkewedJoin(OperatorKey k) { - this(k,-1,null); + this(k,-1,null, null); } public POSkewedJoin(OperatorKey k, int rp) { - this(k, rp, null); + this(k, rp, null, null); } - public POSkewedJoin(OperatorKey k, List<PhysicalOperator> inp) { - this(k, -1, inp); + public POSkewedJoin(OperatorKey k, List<PhysicalOperator> inp, boolean []flags) { + this(k, -1, inp, flags); } - public POSkewedJoin(OperatorKey k, int rp, List<PhysicalOperator> inp) { - super(k,rp,inp); + public POSkewedJoin(OperatorKey k, int rp, List<PhysicalOperator> inp, boolean []flags) { + super(k,rp,inp); + if (flags != null) { + // copy the inner flags + mInnerFlags = new boolean[flags.length]; + for (int i = 0; i < flags.length; i++) { + mInnerFlags[i] = flags[i]; + } + } + } + + public boolean[] getInnerFlags() { + return mInnerFlags; } public MultiMap<PhysicalOperator, PhysicalPlan> getJoinPlans() { @@ -93,5 +110,13 @@ public boolean supportsMultipleOutputs() { return false; } + + public void addSchema(Schema s) { + inputSchema.add(s); + } + + public Schema getSchema(int i) { + return inputSchema.get(i); + } } Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt?rev=832086&r1=832085&r2=832086&view=diff ============================================================================== --- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt (original) +++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt Mon Nov 2 20:20:05 2009 @@ -2019,10 +2019,7 @@ } frj=parseJoin(gis, lp, LOJoin.JOINTYPE.REPLICATED); } - |"\"skewed\"" { - if(isOuter) { - throw new ParseException("Skewed join does not support (left|right|full) outer joins"); - } + |"\"skewed\"" { skj = parseJoin(gis, lp, LOJoin.JOINTYPE.SKEWED); } |"\"merge\"" { Added: hadoop/pig/trunk/src/org/apache/pig/impl/util/CompilerUtils.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/util/CompilerUtils.java?rev=832086&view=auto ============================================================================== --- hadoop/pig/trunk/src/org/apache/pig/impl/util/CompilerUtils.java (added) +++ hadoop/pig/trunk/src/org/apache/pig/impl/util/CompilerUtils.java Mon Nov 2 20:20:05 2009 @@ -0,0 +1,117 @@ +/* + * 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.impl.util; + +import java.util.ArrayList; +import java.util.List; + + +import org.apache.pig.EvalFunc; +import org.apache.pig.FuncSpec; +import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.ConstantExpression; +import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POBinCond; +import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POProject; +import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc; +import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan; +import org.apache.pig.builtin.IsEmpty; +import org.apache.pig.data.DataBag; +import org.apache.pig.data.DataType; +import org.apache.pig.data.NonSpillableDataBag; +import org.apache.pig.data.Tuple; +import org.apache.pig.data.TupleFactory; +import org.apache.pig.impl.PigContext; +import org.apache.pig.impl.logicalLayer.schema.Schema; +import org.apache.pig.impl.plan.NodeIdGenerator; +import org.apache.pig.impl.plan.OperatorKey; +import org.apache.pig.impl.plan.PlanException; + +/* + * A class to add util functions that gets used by LogToPhyTranslator and MRCompiler + * + */ +public class CompilerUtils { + + public static void addEmptyBagOuterJoin(PhysicalPlan fePlan, Schema inputSchema) throws PlanException { + // we currently have POProject[bag] as the only operator in the plan + // If the bag is an empty bag, we should replace + // it with a bag with one tuple with null fields so that when we flatten + // we do not drop records (flatten will drop records if the bag is left + // as an empty bag) and actually project nulls for the fields in + // the empty bag + + // So we need to get to the following state: + // POProject[Bag] + // \ + // POUserFunc["IsEmpty()"] Const[Bag](bag with null fields) + // \ | POProject[Bag] + // \ | / + // POBinCond + POProject relationProject = (POProject) fePlan.getRoots().get(0); + try { + + // condition of the bincond + POProject relationProjectForIsEmpty = relationProject.clone(); + fePlan.add(relationProjectForIsEmpty); + String scope = relationProject.getOperatorKey().scope; + FuncSpec isEmptySpec = new FuncSpec(IsEmpty.class.getName()); + Object f = PigContext.instantiateFuncFromSpec(isEmptySpec); + POUserFunc isEmpty = new POUserFunc(new OperatorKey(scope, NodeIdGenerator.getGenerator(). + getNextNodeId(scope)), -1, null, isEmptySpec, (EvalFunc) f); + isEmpty.setResultType(DataType.BOOLEAN); + fePlan.add(isEmpty); + fePlan.connect(relationProjectForIsEmpty, isEmpty); + + // lhs of bincond (const bag with null fields) + ConstantExpression ce = new ConstantExpression(new OperatorKey(scope, + NodeIdGenerator.getGenerator().getNextNodeId(scope))); + // the following should give a tuple with the + // required number of nulls + Tuple t = TupleFactory.getInstance().newTuple(inputSchema.size()); + for(int i = 0; i < inputSchema.size(); i++) { + t.set(i, null); + } + List<Tuple> bagContents = new ArrayList<Tuple>(1); + bagContents.add(t); + DataBag bg = new NonSpillableDataBag(bagContents); + ce.setValue(bg); + ce.setResultType(DataType.BAG); + //this operator doesn't have any predecessors + fePlan.add(ce); + + //rhs of bincond is the original project + // let's set up the bincond now + POBinCond bincond = new POBinCond(new OperatorKey(scope, + NodeIdGenerator.getGenerator().getNextNodeId(scope))); + bincond.setCond(isEmpty); + bincond.setLhs(ce); + bincond.setRhs(relationProject); + bincond.setResultType(DataType.BAG); + fePlan.add(bincond); + + fePlan.connect(isEmpty, bincond); + fePlan.connect(ce, bincond); + fePlan.connect(relationProject, bincond); + + } catch (Exception e) { + throw new PlanException("Error setting up outerjoin", e); + } + + } + +} Modified: hadoop/pig/trunk/test/org/apache/pig/test/TestJoin.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestJoin.java?rev=832086&r1=832085&r2=832086&view=diff ============================================================================== --- hadoop/pig/trunk/test/org/apache/pig/test/TestJoin.java (original) +++ hadoop/pig/trunk/test/org/apache/pig/test/TestJoin.java Mon Nov 2 20:20:05 2009 @@ -456,7 +456,7 @@ lpt.buildPlan("a = load 'a.txt' as (n:chararray, a:int); "); lpt.buildPlan("b = load 'b.txt' as (n:chararray, m:chararray); "); String[] types = new String[] { "left", "right", "full" }; - String[] joinTypes = new String[] { "replicated", "repl", "skewed", "merge" }; + String[] joinTypes = new String[] { "replicated", "repl", "merge" }; for (int i = 0; i < types.length; i++) { for(int j = 0; j < joinTypes.length; j++) { boolean errCaught = false; Modified: hadoop/pig/trunk/test/org/apache/pig/test/TestSkewedJoin.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestSkewedJoin.java?rev=832086&r1=832085&r2=832086&view=diff ============================================================================== --- hadoop/pig/trunk/test/org/apache/pig/test/TestSkewedJoin.java (original) +++ hadoop/pig/trunk/test/org/apache/pig/test/TestSkewedJoin.java Mon Nov 2 20:20:05 2009 @@ -324,6 +324,43 @@ return; } + public void testSkewedJoinOuter() throws IOException { + pigServer.registerQuery("A = LOAD '" + INPUT_FILE5 + "' as (id,name);"); + pigServer.registerQuery("B = LOAD '" + INPUT_FILE5 + "' as (id,name);"); + try { + DataBag dbfrj = BagFactory.getInstance().newDefaultBag(); + { + pigServer.registerQuery("C = join A by id left, B by id using \"skewed\";"); + Iterator<Tuple> iter = pigServer.openIterator("C"); + + while(iter.hasNext()) { + dbfrj.add(iter.next()); + } + } + { + pigServer.registerQuery("C = join A by id right, B by id using \"skewed\";"); + Iterator<Tuple> iter = pigServer.openIterator("C"); + + while(iter.hasNext()) { + dbfrj.add(iter.next()); + } + } + { + pigServer.registerQuery("C = join A by id full, B by id using \"skewed\";"); + Iterator<Tuple> iter = pigServer.openIterator("C"); + + while(iter.hasNext()) { + dbfrj.add(iter.next()); + } + } + } catch(Exception e) { + System.out.println(e.getMessage()); + e.printStackTrace(); + fail("Should support outer join in skewed join"); + } + return; + } + // pig 1048 public void testSkewedJoinOneValue() throws IOException { pigServer.registerQuery("A = LOAD '" + INPUT_FILE3 + "' as (id,name);"); From pig-commitsmichael55@example.net Mon Nov 02 20:24:22 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 91873 invoked from network); 2 Nov 2009 20:24:21 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 2 Nov 2009 20:24:21 -0000 Received: (qmail 33896 invoked by uid 500); 2 Nov 2009 20:24:21 -0000 Delivered-To: michael55@example.net Received: (qmail 33873 invoked by uid 500); 2 Nov 2009 20:24:21 -0000 Mailing-List: contact michael55@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:michael55@example.net> List-Unsubscribe: <mailto:michael55@example.net> List-Post: <mailto:michael55@example.net> List-Id: <pig-commits.hadoop.apache.org> Reply-To: michael55@example.net Delivered-To: mailing list michael55@example.net Received: (qmail 33864 invoked by uid 500); 2 Nov 2009 20:24:21 -0000 Delivered-To: michael55@example.net Received: (qmail 33860 invoked by uid 99); 2 Nov 2009 20:24:21 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 20:24:21 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 20:24:19 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 48D2A16E09; Mon, 2 Nov 2009 20:23:58 +0000 (GMT) MIME-Version: 1.0 Content-Type: text/plain; charset="utf-8" Content-Transfer-Encoding: quoted-printable From: Apache Wiki <michael55@example.net> To: Apache Wiki <michael55@example.net> Date: Mon, 02 Nov 2009 20:23:58 -0000 Message-ID: <michael55@example.net> Subject: =?utf-8?q?=5BPig_Wiki=5D_Update_of_=22LoadStoreRedesignProposal=22_by_Pra?= =?utf-8?q?deepKamath?= X-Virus-Checked: Checked by ClamAV on apache.org Dear Wiki user, You have subscribed to a wiki page or wiki category on "Pig Wiki" for chang= e notification. The "LoadStoreRedesignProposal" page has been changed by PradeepKamath. http://wiki.apache.org/pig/LoadStoreRedesignProposal?action=3Ddiff&rev1=3D9= &rev2=3D10 -------------------------------------------------- * Added relativeToAbsolutePath() method in LoadFunc per http://issues.ap= ache.org/jira/browse/PIG-879?focusedCommentId=3D12768818&page=3Dcom.atlassi= an.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12768818 * Changed comments in setLocation regarding the location passed - the lo= cation will now be the return value of relativeToAbsolutePath() * setLocation() now also takes a Job argument since the main purpose of = this call is to an opportunity to the LoadFunc implementation to communicat= e the input location to underlying InputFormat. InputFormat implementations= inturn seem to be storing this information inthe Job. For example, FileInp= utFormat has the following static method to set the input location: setInpu= tPaths(JobConf conf, String commaSeparatedPaths) ; + * Removed doneReading() method since there is already a RecordReader.clo= se() method which will be called by Hadoop wherein all the functionality th= at needs to be done on completion of reading can be done. * All methods now can throw IOException - this keeps the interface more = flexible for exception cases = In LoadMetadata: From pig-commitsmichael55@example.net Mon Nov 02 20:40:08 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 99982 invoked from network); 2 Nov 2009 20:40:08 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 2 Nov 2009 20:40:08 -0000 Received: (qmail 62577 invoked by uid 500); 2 Nov 2009 20:40:08 -0000 Delivered-To: michael55@example.net Received: (qmail 62533 invoked by uid 500); 2 Nov 2009 20:40:07 -0000 Mailing-List: contact michael55@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:michael55@example.net> List-Unsubscribe: <mailto:michael55@example.net> List-Post: <mailto:michael55@example.net> List-Id: <pig-commits.hadoop.apache.org> Reply-To: michael55@example.net Delivered-To: mailing list michael55@example.net Received: (qmail 62519 invoked by uid 500); 2 Nov 2009 20:40:07 -0000 Delivered-To: michael55@example.net Received: (qmail 62516 invoked by uid 99); 2 Nov 2009 20:40:07 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 20:40:07 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 20:40:05 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 9D23416E09; Mon, 2 Nov 2009 20:39:44 +0000 (GMT) MIME-Version: 1.0 Content-Type: text/plain; charset="utf-8" Content-Transfer-Encoding: quoted-printable From: Apache Wiki <michael55@example.net> To: Apache Wiki <michael55@example.net> Date: Mon, 02 Nov 2009 20:39:44 -0000 Message-ID: <michael55@example.net> Subject: =?utf-8?q?=5BPig_Wiki=5D_Update_of_=22LoadStoreRedesignProposal=22_by_Pra?= =?utf-8?q?deepKamath?= X-Virus-Checked: Checked by ClamAV on apache.org Dear Wiki user, You have subscribed to a wiki page or wiki category on "Pig Wiki" for chang= e notification. The "LoadStoreRedesignProposal" page has been changed by PradeepKamath. http://wiki.apache.org/pig/LoadStoreRedesignProposal?action=3Ddiff&rev1=3D1= 0&rev2=3D11 -------------------------------------------------- * @return the absolute location based on the arguments passed * @throws IOException if the conversion is not possible */ - String relativeToAbsolutePath(String location, String curDir) throws = IOException; + String relativeToAbsolutePath(String location, Path curDir) throws IO= Exception; = /** * Communicate to the loader the load string used in Pig Latin to ref= er to the=20 From pig-commitsmichael55@example.net Mon Nov 02 21:39:34 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 27522 invoked from network); 2 Nov 2009 21:39:34 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 2 Nov 2009 21:39:34 -0000 Received: (qmail 62328 invoked by uid 500); 2 Nov 2009 21:39:34 -0000 Delivered-To: michael55@example.net Received: (qmail 62287 invoked by uid 500); 2 Nov 2009 21:39:33 -0000 Mailing-List: contact michael55@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:michael55@example.net> List-Unsubscribe: <mailto:michael55@example.net> List-Post: <mailto:michael55@example.net> List-Id: <pig-commits.hadoop.apache.org> Reply-To: michael55@example.net Delivered-To: mailing list michael55@example.net Received: (qmail 62278 invoked by uid 500); 2 Nov 2009 21:39:33 -0000 Delivered-To: michael55@example.net Received: (qmail 62275 invoked by uid 99); 2 Nov 2009 21:39:33 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 21:39:33 +0000 X-ASF-Spam-Status: No, hits=-2.5 required=5.0 tests=AWL,BAYES_00 X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 21:39:31 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 46AA216E09; Mon, 2 Nov 2009 21:39:11 +0000 (GMT) MIME-Version: 1.0 Content-Type: text/plain; charset="utf-8" Content-Transfer-Encoding: quoted-printable From: Apache Wiki <michael55@example.net> To: Apache Wiki <michael55@example.net> Date: Mon, 02 Nov 2009 21:39:11 -0000 Message-ID: <michael55@example.net> Subject: =?utf-8?q?=5BPig_Wiki=5D_Update_of_=22LoadStoreRedesignProposal=22_by_Pra?= =?utf-8?q?deepKamath?= Dear Wiki user, You have subscribed to a wiki page or wiki category on "Pig Wiki" for chang= e notification. The "LoadStoreRedesignProposal" page has been changed by PradeepKamath. http://wiki.apache.org/pig/LoadStoreRedesignProposal?action=3Ddiff&rev1=3D1= 1&rev2=3D12 -------------------------------------------------- * = * @param location location as provided in the "load" statement of th= e script * @param curDir the current working direction based on any "cd" stat= ements - * in the script before the "load" statement + * in the script before the "load" statement. If there are no "cd" st= atements + * in the script, this would be the home directory - = + * <pre>/user/<username> </pre> * @return the absolute location based on the arguments passed * @throws IOException if the conversion is not possible */ String relativeToAbsolutePath(String location, Path curDir) throws IO= Exception; + = = /** * Communicate to the loader the load string used in Pig Latin to ref= er to the=20 From pig-commitsmichael55@example.net Mon Nov 02 21:57:06 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 33119 invoked from network); 2 Nov 2009 21:57:06 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 2 Nov 2009 21:57:06 -0000 Received: (qmail 86197 invoked by uid 500); 2 Nov 2009 21:57:06 -0000 Delivered-To: michael55@example.net Received: (qmail 86154 invoked by uid 500); 2 Nov 2009 21:57:06 -0000 Mailing-List: contact michael55@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:michael55@example.net> List-Unsubscribe: <mailto:michael55@example.net> List-Post: <mailto:michael55@example.net> List-Id: <pig-commits.hadoop.apache.org> Reply-To: michael55@example.net Delivered-To: mailing list michael55@example.net Received: (qmail 86144 invoked by uid 500); 2 Nov 2009 21:57:06 -0000 Delivered-To: michael55@example.net Received: (qmail 86141 invoked by uid 99); 2 Nov 2009 21:57:06 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 21:57:06 +0000 X-ASF-Spam-Status: No, hits=-2.5 required=5.0 tests=AWL,BAYES_00 X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 21:57:03 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 4201316E09; Mon, 2 Nov 2009 21:56:43 +0000 (GMT) MIME-Version: 1.0 Content-Type: text/plain; charset="utf-8" Content-Transfer-Encoding: quoted-printable From: Apache Wiki <michael55@example.net> To: Apache Wiki <michael55@example.net> Date: Mon, 02 Nov 2009 21:56:43 -0000 Message-ID: <michael55@example.net> Subject: =?utf-8?q?=5BPig_Wiki=5D_Update_of_=22LoadStoreRedesignProposal=22_by_Pra?= =?utf-8?q?deepKamath?= Dear Wiki user, You have subscribed to a wiki page or wiki category on "Pig Wiki" for chang= e notification. The "LoadStoreRedesignProposal" page has been changed by PradeepKamath. http://wiki.apache.org/pig/LoadStoreRedesignProposal?action=3Ddiff&rev1=3D1= 2&rev2=3D13 -------------------------------------------------- '''!StoreFunc''' = {{{ + = + /** + * This interface is used to implement functions to write records + * from a dataset. + */ + = public interface StoreFunc { + = + /** + * This method is called by the Pig runtime in the front end to conve= rt the + * output location to an absolute path if the location is relative. T= he + * StoreFunc implementation is free to choose how it converts a relat= ive = + * location to an absolute location since this may depend on what the= location + * string represent (hdfs path or some other data source) + * = + * @param location location as provided in the "store" statement of t= he script + * @param curDir the current working direction based on any "cd" stat= ements + * in the script before the "store" statement. If there are no "cd" s= tatements + * in the script, this would be the home directory - = + * <pre>/user/<username> </pre> + * @return the absolute location based on the arguments passed + * @throws IOException if the conversion is not possible + */ + String relToAbsPathForStoreLocation(String location, Path curDir) thr= ows IOException; = /** * Return the OutputFormat associated with StoreFunc. This will be c= alled * on the front end during planning and not on the backend during - * execution. OutputFormat information need not be carried to the ba= ck end - * as the appropriate RecordWriter will be provided to the StoreFunc. + * execution. = + * @return the {@link OutputFormat} associated with StoreFunc + * @throws IOException if an exception occurs while constructing the = + * OutputFormat - */ + * + */ - OutputFormat getOutputFormat(); + OutputFormat getOutputFormat() throws IOException; = /** * Communicate to the store function the location used in Pig Latin t= o refer = @@ -327, +353 @@ * called during planning on the front end, not during execution on * the backend. * @param location Location indicated in store statement. + * @param job The {@link Job} object * @throws IOException if the location is not valid. */ - void setLocation(String location) throws IOException; + void setStoreLocation(String location, Job job) throws IOException; = /** * Set the schema for data to be stored. This will be called on the + * front end during planning. A Store function should implement this = function to - * front end during planning. If the store function wishes to record - * the schema it will need to carry it to the backend. - * Even if a store function cannot - * record the schema, it may need to implement this function to * check that a given schema is acceptable to it. For example, it * can check that the correct partition keys are included; * a storage function to be written directly to an OutputFormat can * make sure the schema will translate in a well defined way. = - * @param schema to be checked/set + * @param s to be checked - * @throw IOException if this schema is not acceptable. It should in= clude + * @throws IOException if this schema is not acceptable. It should i= nclude * a detailed error message indicating what is wrong with the schema. */ - void setSchema(ResourceSchema s) throws IOException; + void checkSchema(ResourceSchema s) throws IOException; = /** * Initialize StoreFunc to write data. This will be called during * execution before the call to putNext. * @param writer RecordWriter to use. + * @throws IOException if an exception occurs during initialization */ - void prepareToWrite(RecordWriter writer); + void prepareToWrite(RecordWriter writer) throws IOException; - = - /** - * Called when all writing is finished. This will be called on the b= ackend, - * once for each writing task. - */ - void doneWriting(); = /** * Write a tuple the output stream to which this instance was * previously bound. * = - * @param f the tuple to store. + * @param t the tuple to store. - * @throws IOException + * @throws IOException if an exception occurs during the write */ void putNext(Tuple t) throws IOException; - = - /** - * Called when writing all of the data is finished. This can be used - * to commit information to a metadata system, clean up tmp files, = - * close connections, etc. This call will be made on the front end - * after all back end processing is finished. - * @param conf The job configuration = - */ - void allFinished(Configuration conf); - = - = = } = @@ -529, +537 @@ * setLocation() now also takes a Job argument since the main purpose of = this call is to an opportunity to the LoadFunc implementation to communicat= e the input location to underlying InputFormat. InputFormat implementations= inturn seem to be storing this information inthe Job. For example, FileInp= utFormat has the following static method to set the input location: setInpu= tPaths(JobConf conf, String commaSeparatedPaths) ; * Removed doneReading() method since there is already a RecordReader.clo= se() method which will be called by Hadoop wherein all the functionality th= at needs to be done on completion of reading can be done. * All methods now can throw IOException - this keeps the interface more = flexible for exception cases - = In LoadMetadata: * getSchema(), getStatistics() and getPartitionKeys() methods now take a= location and Configuration argument so that the implementation can use tha= t information in returning the information requested. + In StoreFunc: + * Added relativeToAbsolutePath() method per http://issues.apache.org/jir= a/browse/PIG-879?focusedCommentId=3D12768818&page=3Dcom.atlassian.jira.plug= in.system.issuetabpanels%3Acomment-tabpanel#action_12768818 + * Methods which did not throw IOException now do so to enable exceptions= in implementations + * Removed doneWriting() - same functionality already present in RecordWr= iter.close() and OutputCommitter.commitTask() + * Changed setSchema() to checkSchema since this method is called only to= allow StoreFunc to check + * Removed allFinished() - same functionality already present in OutputCo= mmitter.cleanupJob() =20 From pig-commitsmichael55@example.net Mon Nov 02 21:58:59 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 33733 invoked from network); 2 Nov 2009 21:58:59 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 2 Nov 2009 21:58:59 -0000 Received: (qmail 88370 invoked by uid 500); 2 Nov 2009 21:58:59 -0000 Delivered-To: michael55@example.net Received: (qmail 88328 invoked by uid 500); 2 Nov 2009 21:58:59 -0000 Mailing-List: contact michael55@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:michael55@example.net> List-Unsubscribe: <mailto:michael55@example.net> List-Post: <mailto:michael55@example.net> List-Id: <pig-commits.hadoop.apache.org> Reply-To: michael55@example.net Delivered-To: mailing list michael55@example.net Received: (qmail 88319 invoked by uid 500); 2 Nov 2009 21:58:59 -0000 Delivered-To: michael55@example.net Received: (qmail 88316 invoked by uid 99); 2 Nov 2009 21:58:59 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 21:58:59 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 21:58:56 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 9EFED16E09; Mon, 2 Nov 2009 21:58:35 +0000 (GMT) MIME-Version: 1.0 Content-Type: text/plain; charset="utf-8" Content-Transfer-Encoding: quoted-printable From: Apache Wiki <michael55@example.net> To: Apache Wiki <michael55@example.net> Date: Mon, 02 Nov 2009 21:58:35 -0000 Message-ID: <michael55@example.net> Subject: =?utf-8?q?=5BPig_Wiki=5D_Update_of_=22LoadStoreRedesignProposal=22_by_Pra?= =?utf-8?q?deepKamath?= X-Virus-Checked: Checked by ClamAV on apache.org Dear Wiki user, You have subscribed to a wiki page or wiki category on "Pig Wiki" for chang= e notification. The "LoadStoreRedesignProposal" page has been changed by PradeepKamath. http://wiki.apache.org/pig/LoadStoreRedesignProposal?action=3Ddiff&rev1=3D1= 3&rev2=3D14 -------------------------------------------------- = Nov 2 2009, Pradeep Kamath = - In LoadFunc: + In !LoadFunc: - * Added relativeToAbsolutePath() method in LoadFunc per http://issues.ap= ache.org/jira/browse/PIG-879?focusedCommentId=3D12768818&page=3Dcom.atlassi= an.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12768818 + * Added relativeToAbsolutePath() method in !LoadFunc per http://issues.a= pache.org/jira/browse/PIG-879?focusedCommentId=3D12768818&page=3Dcom.atlass= ian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12768818 * Changed comments in setLocation regarding the location passed - the lo= cation will now be the return value of relativeToAbsolutePath() - * setLocation() now also takes a Job argument since the main purpose of = this call is to an opportunity to the LoadFunc implementation to communicat= e the input location to underlying InputFormat. InputFormat implementations= inturn seem to be storing this information inthe Job. For example, FileInp= utFormat has the following static method to set the input location: setInpu= tPaths(JobConf conf, String commaSeparatedPaths) ; + * setLocation() now also takes a Job argument since the main purpose of = this call is to an opportunity to the !LoadFunc implementation to communica= te the input location to underlying !InputFormat. !InputFormat implementati= ons inturn seem to be storing this information inthe Job. For example, !Fil= eInputFormat has the following static method to set the input location: set= InputPaths(JobConf conf, String commaSeparatedPaths) ; * Removed doneReading() method since there is already a RecordReader.clo= se() method which will be called by Hadoop wherein all the functionality th= at needs to be done on completion of reading can be done. * All methods now can throw IOException - this keeps the interface more = flexible for exception cases - In LoadMetadata: + In !LoadMetadata: * getSchema(), getStatistics() and getPartitionKeys() methods now take a= location and Configuration argument so that the implementation can use tha= t information in returning the information requested. - In StoreFunc: + In !StoreFunc: - * Added relativeToAbsolutePath() method per http://issues.apache.org/jir= a/browse/PIG-879?focusedCommentId=3D12768818&page=3Dcom.atlassian.jira.plug= in.system.issuetabpanels%3Acomment-tabpanel#action_12768818 + * Added relToAbsPathForStoreLocation() method per http://issues.apache.o= rg/jira/browse/PIG-879?focusedCommentId=3D12768818&page=3Dcom.atlassian.jir= a.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12768818 * Methods which did not throw IOException now do so to enable exceptions= in implementations - * Removed doneWriting() - same functionality already present in RecordWr= iter.close() and OutputCommitter.commitTask() + * Removed doneWriting() - same functionality already present in !RecordW= riter.close() and !OutputCommitter.commitTask() * Changed setSchema() to checkSchema since this method is called only to= allow StoreFunc to check - * Removed allFinished() - same functionality already present in OutputCo= mmitter.cleanupJob() + * Removed allFinished() - same functionality already present in !OutputC= ommitter.cleanupJob() =20 From pig-commitsmichael55@example.net Mon Nov 02 22:32:47 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 47011 invoked from network); 2 Nov 2009 22:32:47 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 2 Nov 2009 22:32:47 -0000 Received: (qmail 52766 invoked by uid 500); 2 Nov 2009 22:32:47 -0000 Delivered-To: michael55@example.net Received: (qmail 52725 invoked by uid 500); 2 Nov 2009 22:32:46 -0000 Mailing-List: contact michael55@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:michael55@example.net> List-Unsubscribe: <mailto:michael55@example.net> List-Post: <mailto:michael55@example.net> List-Id: <pig-commits.hadoop.apache.org> Reply-To: michael55@example.net Delivered-To: mailing list michael55@example.net Received: (qmail 52716 invoked by uid 500); 2 Nov 2009 22:32:46 -0000 Delivered-To: michael55@example.net Received: (qmail 52713 invoked by uid 99); 2 Nov 2009 22:32:46 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 22:32:46 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 02 Nov 2009 22:32:44 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 90BF916E09; Mon, 2 Nov 2009 22:32:23 +0000 (GMT) MIME-Version: 1.0 Content-Type: text/plain; charset="utf-8" Content-Transfer-Encoding: quoted-printable From: Apache Wiki <michael55@example.net> To: Apache Wiki <michael55@example.net> Date: Mon, 02 Nov 2009 22:32:23 -0000 Message-ID: <michael55@example.net> Subject: =?utf-8?q?=5BPig_Wiki=5D_Update_of_=22LoadStoreRedesignProposal=22_by_Pra?= =?utf-8?q?deepKamath?= X-Virus-Checked: Checked by ClamAV on apache.org Dear Wiki user, You have subscribed to a wiki page or wiki category on "Pig Wiki" for chang= e notification. The "LoadStoreRedesignProposal" page has been changed by PradeepKamath. http://wiki.apache.org/pig/LoadStoreRedesignProposal?action=3Ddiff&rev1=3D1= 4&rev2=3D15 -------------------------------------------------- = }}} = - Open questions for !LoadFunc: - 1. Should setLocation instead be setURI and take a URI? The advantage o= f a URI is we know exactly what users are trying to communicate with, and w= e can define what Pig does in default cases (when a scheme is not given). = The disadvantage is forcing more structure on users and their load function= s. - = The '''!LoadCaster''' interface will include bytesToInt, bytesToLong, etc= . functions currently in !LoadFunc. UTF8!StorageConverter will implement this interf= ace. + = + Open Question: Should the methods to convert to a Bag, Tuple and Map take= a Schema (ResourceSchema?) argument? = + = = '''!LoadMetadata''' {{{ From pig-commitsmichael55@example.net Mon Nov 02 22:53:36 2009 Return-Path: <pig-commitsmichael55@example.net> Delivered-To: michael55@example.net Received: (qmail 56734 invoked from network);
{'source': 'Apache Software Foundation Public Mail Archives'}