| /** |
| * 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.test; |
| |
| import static org.apache.pig.newplan.logical.relational.LOTestHelper.newLOLoad; |
| import static org.junit.Assert.assertTrue; |
| |
| import java.io.IOException; |
| import java.util.HashSet; |
| import java.util.Properties; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.pig.ExecType; |
| import org.apache.pig.FuncSpec; |
| import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil; |
| import org.apache.pig.data.DataType; |
| import org.apache.pig.impl.PigContext; |
| import org.apache.pig.impl.io.FileSpec; |
| import org.apache.pig.impl.util.MultiMap; |
| import org.apache.pig.newplan.logical.expression.AndExpression; |
| import org.apache.pig.newplan.logical.expression.ConstantExpression; |
| import org.apache.pig.newplan.logical.expression.EqualExpression; |
| import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan; |
| import org.apache.pig.newplan.logical.expression.ProjectExpression; |
| import org.apache.pig.newplan.logical.optimizer.LogicalPlanOptimizer; |
| import org.apache.pig.newplan.logical.optimizer.SchemaResetter; |
| 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.LOJoin.JOINTYPE; |
| import org.apache.pig.newplan.logical.relational.LOLoad; |
| import org.apache.pig.newplan.logical.relational.LogicalPlan; |
| import org.apache.pig.newplan.logical.relational.LogicalSchema; |
| import org.apache.pig.newplan.logical.relational.LogicalSchema.LogicalFieldSchema; |
| import org.junit.Before; |
| import org.junit.Test; |
| |
| /** |
| * Test end to end logical optimizations. |
| */ |
| public class TestNewPlanLogicalOptimizer { |
| |
| Configuration conf = null; |
| |
| @Before |
| public void setUp() throws Exception { |
| PigContext pc = new PigContext(ExecType.LOCAL, new Properties()); |
| pc.connect(); |
| conf = new Configuration( |
| ConfigurationUtil.toConfiguration(pc.getFs().getConfiguration()) |
| ); |
| } |
| |
| @Test |
| public void testFilterPushDown() throws IOException { |
| // A logical plan for: |
| // A = load 'bla' as (x, y); |
| // B = load 'morebla' as (a, b); |
| // C = join A on x, B on a; |
| // D = filter C by x = a and x = 0 and b = 1 and y = b; |
| // store D into 'whatever'; |
| |
| // A = load |
| LogicalPlan lp = new LogicalPlan(); |
| { |
| LogicalSchema aschema = new LogicalSchema(); |
| aschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "x", null, DataType.BYTEARRAY)); |
| aschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "y", null, DataType.BYTEARRAY)); |
| LOLoad A = newLOLoad(new FileSpec("bla", new FuncSpec("PigStorage", "\t")), aschema, lp, conf); |
| A.setAlias("A"); |
| lp.add(A); |
| |
| // B = load |
| LogicalSchema bschema = new LogicalSchema(); |
| bschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "a", null, DataType.BYTEARRAY)); |
| bschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "b", null, DataType.BYTEARRAY)); |
| LOLoad B = newLOLoad(new FileSpec("morebla", new FuncSpec("PigStorage", "\t")), bschema, lp, conf); |
| B.setAlias("B"); |
| lp.add(B); |
| |
| // C = join |
| LogicalSchema cschema = new LogicalSchema(); |
| cschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "x", null, DataType.BYTEARRAY)); |
| cschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "y", null, DataType.BYTEARRAY)); |
| cschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "a", null, DataType.BYTEARRAY)); |
| cschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "b", null, DataType.BYTEARRAY)); |
| MultiMap<Integer, LogicalExpressionPlan> mm = |
| new MultiMap<Integer, LogicalExpressionPlan>(); |
| LogicalExpressionPlan aprojplan = new LogicalExpressionPlan(); |
| LOJoin C = new LOJoin(lp, mm, JOINTYPE.HASH, new boolean[] {true, true}); |
| new ProjectExpression(aprojplan, 0, 0, C); |
| LogicalExpressionPlan bprojplan = new LogicalExpressionPlan(); |
| new ProjectExpression(bprojplan, 1, 0, C); |
| mm.put(0, aprojplan); |
| mm.put(1, bprojplan); |
| |
| C.setAlias("C"); |
| lp.add(C); |
| lp.connect(A, C); |
| lp.connect(B, C); |
| |
| // D = filter |
| LogicalExpressionPlan filterPlan = new LogicalExpressionPlan(); |
| LOFilter D = new LOFilter(lp, filterPlan); |
| ProjectExpression fx = new ProjectExpression(filterPlan, 0, 0, D); |
| ConstantExpression fc0 = new ConstantExpression(filterPlan, new Integer(0)); |
| EqualExpression eq1 = new EqualExpression(filterPlan, fx, fc0); |
| ProjectExpression fanotherx = new ProjectExpression(filterPlan, 0, 0, D); |
| ProjectExpression fa = new ProjectExpression(filterPlan, 0, 2, D); |
| EqualExpression eq2 = new EqualExpression(filterPlan, fanotherx, fa); |
| AndExpression and1 = new AndExpression(filterPlan, eq1, eq2); |
| ProjectExpression fb = new ProjectExpression(filterPlan, 0, 3, D); |
| ConstantExpression fc1 = new ConstantExpression(filterPlan, new Integer(1)); |
| EqualExpression eq3 = new EqualExpression(filterPlan, fb, fc1); |
| AndExpression and2 = new AndExpression(filterPlan, and1, eq3); |
| ProjectExpression fanotherb = new ProjectExpression(filterPlan, 0, 3, D); |
| ProjectExpression fy = new ProjectExpression(filterPlan, 0, 1, D); |
| EqualExpression eq4 = new EqualExpression(filterPlan, fy, fanotherb); |
| new AndExpression(filterPlan, and2, eq4); |
| |
| D.setAlias("D"); |
| // Connect D to B, since the transform has happened. |
| lp.add(D); |
| lp.connect(C, D); |
| } |
| |
| System.out.println(lp); |
| LogicalPlanOptimizer optimizer = new LogicalPlanOptimizer(lp, 500, null); |
| optimizer.optimize(); |
| |
| LogicalPlan expected = new LogicalPlan(); |
| { |
| // A = load |
| LogicalSchema aschema = new LogicalSchema(); |
| aschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "x", null, DataType.BYTEARRAY)); |
| aschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "y", null, DataType.BYTEARRAY)); |
| LOLoad A = newLOLoad(new FileSpec("bla", new FuncSpec("PigStorage", "\t")), aschema, expected, conf); |
| expected.add(A); |
| |
| // DA = filter |
| LogicalExpressionPlan DAfilterPlan = new LogicalExpressionPlan(); |
| LOFilter DA = new LOFilter(expected, DAfilterPlan); |
| ProjectExpression fx = new ProjectExpression(DAfilterPlan, 0, 0, DA); |
| fx.neverUseForRealSetFieldSchema(new LogicalFieldSchema(null, null, DataType.BYTEARRAY)); |
| ConstantExpression fc0 = new ConstantExpression(DAfilterPlan, new Integer(0)); |
| new EqualExpression(DAfilterPlan, fx, fc0); |
| |
| DA.neverUseForRealSetSchema(aschema); |
| expected.add(DA); |
| expected.connect(A, DA); |
| |
| // A = foreach |
| LOForEach foreachA = org.apache.pig.newplan.logical.Util.addForEachAfter(expected, DA, 0, new HashSet<Integer>()); |
| foreachA.setAlias("A"); |
| foreachA.neverUseForRealSetSchema(aschema); |
| |
| // B = load |
| LogicalSchema bschema = new LogicalSchema(); |
| bschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "a", null, DataType.BYTEARRAY)); |
| bschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "b", null, DataType.BYTEARRAY)); |
| LOLoad B = newLOLoad(new FileSpec("morebla", new FuncSpec("PigStorage", "\t")), bschema, expected, conf); |
| expected.add(B); |
| |
| // DB = filter |
| LogicalExpressionPlan DBfilterPlan = new LogicalExpressionPlan(); |
| LOFilter DB = new LOFilter(expected, DBfilterPlan); |
| ProjectExpression fb = new ProjectExpression(DBfilterPlan, 0, 1, DB); |
| fb.neverUseForRealSetFieldSchema(new LogicalFieldSchema(null, null, DataType.BYTEARRAY)); |
| ConstantExpression fc1 = new ConstantExpression(DBfilterPlan, new Integer(1)); |
| new EqualExpression(DBfilterPlan, fb, fc1); |
| |
| DB.neverUseForRealSetSchema(bschema); |
| expected.add(DB); |
| expected.connect(B, DB); |
| |
| // B = foreach |
| LOForEach foreachB = org.apache.pig.newplan.logical.Util.addForEachAfter(expected, DB, 0, new HashSet<Integer>()); |
| foreachB.setAlias("B"); |
| foreachB.neverUseForRealSetSchema(bschema); |
| |
| // C = join |
| LogicalSchema cschema = new LogicalSchema(); |
| cschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "D::x", null, DataType.BYTEARRAY)); |
| cschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "D::y", null, DataType.BYTEARRAY)); |
| cschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "null::a", null, DataType.BYTEARRAY)); |
| cschema.addField(new LogicalSchema.LogicalFieldSchema( |
| "null::b", null, DataType.BYTEARRAY)); |
| cschema.getField(0).uid = 1; |
| cschema.getField(1).uid = 2; |
| cschema.getField(2).uid = 3; |
| cschema.getField(3).uid = 4; |
| LogicalExpressionPlan aprojplan = new LogicalExpressionPlan(); |
| MultiMap<Integer, LogicalExpressionPlan> mm = |
| new MultiMap<Integer, LogicalExpressionPlan>(); |
| LOJoin C = new LOJoin(expected, mm, JOINTYPE.HASH, new boolean[] {true, true}); |
| |
| ProjectExpression x = new ProjectExpression(aprojplan, 0, 0, C); |
| x.neverUseForRealSetFieldSchema(new LogicalFieldSchema(null, null, DataType.BYTEARRAY)); |
| LogicalExpressionPlan bprojplan = new LogicalExpressionPlan(); |
| ProjectExpression y = new ProjectExpression(bprojplan, 1, 0, C); |
| y.neverUseForRealSetFieldSchema(new LogicalFieldSchema(null, null, DataType.BYTEARRAY)); |
| mm.put(0, aprojplan); |
| mm.put(1, bprojplan); |
| C.neverUseForRealSetSchema(cschema); |
| expected.add(C); |
| expected.connect(foreachA, C); |
| expected.connect(foreachB, C); |
| |
| // D = filter |
| LogicalExpressionPlan filterPlan = new LogicalExpressionPlan(); |
| LOFilter D = new LOFilter(expected, filterPlan); |
| ProjectExpression fanotherx = new ProjectExpression(filterPlan, 0, 0, D); |
| fanotherx.neverUseForRealSetFieldSchema(new LogicalFieldSchema(null, null, DataType.BYTEARRAY)); |
| ProjectExpression fa = new ProjectExpression(filterPlan, 0, 2, D); |
| fa.neverUseForRealSetFieldSchema(new LogicalFieldSchema(null, null, DataType.BYTEARRAY)); |
| EqualExpression eq2 = new EqualExpression(filterPlan, fanotherx, fa); |
| ProjectExpression fanotherb = new ProjectExpression(filterPlan, 0, 3, D); |
| fanotherb.neverUseForRealSetFieldSchema(new LogicalFieldSchema(null, null, DataType.BYTEARRAY)); |
| ProjectExpression fy = new ProjectExpression(filterPlan, 0, 1, D); |
| fy.neverUseForRealSetFieldSchema(new LogicalFieldSchema(null, null, DataType.BYTEARRAY)); |
| EqualExpression eq4 = new EqualExpression(filterPlan, fy, fanotherb); |
| new AndExpression(filterPlan, eq2, eq4); |
| |
| D.neverUseForRealSetSchema(cschema); |
| expected.add(D); |
| expected.connect(C, D); |
| } |
| |
| SchemaResetter schemaResetter = new SchemaResetter(lp); |
| schemaResetter.visit(); |
| |
| schemaResetter = new SchemaResetter(expected); |
| schemaResetter.visit(); |
| |
| |
| assertTrue( lp.isEqual(expected) ); |
| } |
| |
| } |
