content: /*
   * 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.calcite.test;
  
  import org.apache.calcite.rel.RelNode;
  import org.apache.calcite.sql.SqlDialect;
  import org.apache.calcite.sql.dialect.CalciteSqlDialect;
  import org.apache.calcite.util.TestUtil;
  import org.apache.calcite.util.Util;
  
  import org.hamcrest.Matcher;
  import org.junit.jupiter.api.Test;
  
  import java.io.File;
  import java.io.FileNotFoundException;
  import java.io.FileOutputStream;
  import java.io.IOException;
  import java.io.OutputStreamWriter;
  import java.io.PrintWriter;
  import java.io.StringWriter;
  import java.nio.charset.StandardCharsets;
  import java.util.List;
  
  import static org.apache.calcite.test.Matchers.hasTree;
  
  import static org.hamcrest.CoreMatchers.is;
  import static org.hamcrest.MatcherAssert.assertThat;
  
  /**
   * Tests for {@code PigRelOpVisitor}.
   */
  class PigRelOpTest extends PigRelTestBase {
    /**
     * SQL dialect for the tests.
     */
    private static class PigRelSqlDialect extends SqlDialect {
      static final SqlDialect DEFAULT =
          new CalciteSqlDialect(SqlDialect.EMPTY_CONTEXT
              .withDatabaseProduct(DatabaseProduct.CALCITE));
  
      private PigRelSqlDialect(Context context) {
        super(context);
      }
    }
  
    /** Contains a Pig script and has various methods to translate and
     * run that script and check the results. Each method returns
     * this, so that method calls for the same script can be
     * chained. */
    class Fluent {
      private final String script;
  
      Fluent(String script) {
        this.script = script;
      }
  
      private Fluent assertRel(String pigAlias, boolean optimized,
          Matcher<RelNode> relMatcher) {
        try {
          final RelNode rel;
          final List<RelNode> relNodes =
              converter.pigQuery2Rel(script, optimized, true, optimized);
          if (pigAlias == null) {
            rel = relNodes.get(0);
          } else {
            rel = converter.getBuilder().getRel(pigAlias);
          }
          assertThat(rel, relMatcher);
        } catch (IOException e) {
          throw TestUtil.rethrow(e);
        }
        return this;
      }
  
      private Fluent assertRel(Matcher<RelNode> relMatcher) {
        return assertRel(null, false, relMatcher);
      }
  
      private Fluent assertOptimizedRel(Matcher<RelNode> relMatcher) {
        return assertRel(null, true, relMatcher);
      }
  
      private Fluent assertSql(Matcher<String> sqlMatcher) {
        try {
          final String sql =
              converter.pigToSql(script, PigRelSqlDialect.DEFAULT).get(0);
          assertThat(sql, sqlMatcher);
          return this;
        } catch (IOException e) {
          throw TestUtil.rethrow(e);
        }
      }
  
      private Fluent assertSql(Matcher<String> sqlMatcher, int pos) {
        try {
          final String sql =
              converter.pigToSql(script, PigRelSqlDialect.DEFAULT).get(pos);
          assertThat(sql, sqlMatcher);
          return this;
        } catch (IOException e) {
          throw TestUtil.rethrow(e);
        }
      }
  
      private Fluent assertResult(Matcher<String> resultMatcher) {
        final RelNode rel;
        try {
          rel = converter.pigQuery2Rel(script, false, true, false).get(0);
        } catch (IOException e) {
          throw TestUtil.rethrow(e);
        }
        final StringWriter sw = new StringWriter();
        CalciteHandler.dump(rel, new PrintWriter(sw));
        assertThat(Util.toLinux(sw.toString()), resultMatcher);
        return this;
      }
    }
  
    private static void writeToFile(File f, String[] inputData) {
      try (PrintWriter pw =
               new PrintWriter(
                   new OutputStreamWriter(new FileOutputStream(f),
                       StandardCharsets.UTF_8))) {
        for (String input : inputData) {
          pw.print(input);
          pw.print("\n");
        }
      } catch (FileNotFoundException e) {
        throw TestUtil.rethrow(e);
      }
    }
  
    /** Creates a {@link Fluent} containing a script, that can then be used to
     * translate and execute that script. */
    private Fluent pig(String script) {
      return new Fluent(script);
    }
  
    @Test void testLoadFromFile() {
      final String datadir = "/tmp/pigdata";
      final String schema = "{\"fields\":["
          + "{\"name\":\"x\",\"type\":55,\"schema\":null},"
          + "{\"name\":\"y\",\"type\":10,\"schema\":null},"
          + "{\"name\":\"z\",\"type\":25,\"schema\":null}],"
          + "\"version\":0,\"sortKeys\":[],\"sortKeyOrders\":[]}";
      final File inputDir = new File(datadir, "testTable");
      inputDir.mkdirs();
      final File inputSchemaFile = new File(inputDir, ".pig_schema");
      writeToFile(inputSchemaFile, new String[]{schema});
  
      final String script = ""
          + "A = LOAD '" + inputDir.getAbsolutePath() + "' using PigStorage();\n"
          + "B = FILTER A BY z > 5.5;\n"
          + "C = GROUP B BY x;\n";
      final String plan = ""
          + "LogicalProject(group=[$0], B=[$1])\n"
          + "  LogicalAggregate(group=[{0}], B=[COLLECT($1)])\n"
          + "    LogicalProject(x=[$0], $f1=[ROW($0, $1, $2)])\n"
          + "      LogicalFilter(condition=[>($2, 5.5E0)])\n"
          + "        LogicalTableScan(table=[[/tmp/pigdata/testTable]])\n";
      pig(script).assertRel(hasTree(plan));
    }
  
    @Test void testLoadWithoutSchema() {
      final String script = "A = LOAD 'scott.DEPT';";
      final String plan = "LogicalTableScan(table=[[scott, DEPT]])\n";
      final String result = ""
          + "(10,ACCOUNTING,NEW YORK)\n"
          + "(20,RESEARCH,DALLAS)\n"
          + "(30,SALES,CHICAGO)\n"
          + "(40,OPERATIONS,BOSTON)\n";
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result));
    }
  
    @Test void testLoadWithSchema() {
      final String script = ""
          + "A = LOAD 'testSchema.testTable' as (a:int, b:long, c:float, "
          + "d:double, e:chararray, "
          + "f:bytearray, g:boolean, "
          + "h:datetime, i:biginteger, j:bigdecimal, k1:tuple(), k2:tuple"
          + "(k21:int, k22:float), "
          + "l1:bag{}, "
          + "l2:bag{l21:(l22:int, l23:float)}, m1:map[], m2:map[int], m3:map["
          + "(m3:float)])\n;";
      final String plan = "LogicalTableScan(table=[[testSchema, testTable]])\n";
      pig(script).assertRel(hasTree(plan));
  
      final String script1 =
          "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);";
      pig(script1)
          .assertRel(hasTree("LogicalTableScan(table=[[scott, DEPT]])\n"));
    }
  
    @Test void testFilter() {
      final String script = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n"
          + "B = FILTER A BY DEPTNO == 10;\n";
      final String plan = ""
          + "LogicalFilter(condition=[=($0, 10)])\n"
          + "  LogicalTableScan(table=[[scott, DEPT]])\n";
      final String result = "(10,ACCOUNTING,NEW YORK)\n";
      final String sql = "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "WHERE DEPTNO = 10";
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testSample() {
      final String script = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n"
          + "B = SAMPLE A 0.5;\n";
      final String plan = ""
          + "LogicalFilter(condition=[<(RAND(), 0.5E0)])\n"
          + "  LogicalTableScan(table=[[scott, DEPT]])\n";
      final String sql = ""
          + "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "WHERE RAND() < 5E-1";
      pig(script).assertRel(hasTree(plan))
          .assertSql(is(sql));
    }
  
    @Test void testSplit() {
      String script = ""
          + "A = LOAD 'scott.EMP'as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "SPLIT A INTO B1 IF DEPTNO == 10, B2 IF  DEPTNO == 20;\n"
          + "B = UNION B1, B2;\n";
      final String scan = "  LogicalTableScan(table=[[scott, EMP]])\n";
      final String plan = ""
          + "LogicalUnion(all=[true])\n"
          + "  LogicalFilter(condition=[=($7, 10)])\n"
          + "    LogicalTableScan(table=[[scott, EMP]])\n"
          + "  LogicalFilter(condition=[=($7, 20)])\n"
          + "    LogicalTableScan(table=[[scott, EMP]])\n";
  
      final String result = ""
          + "(7782,CLARK,MANAGER,7839,1981-06-09,2450.00,null,10)\n"
          + "(7839,KING,PRESIDENT,null,1981-11-17,5000.00,null,10)\n"
          + "(7934,MILLER,CLERK,7782,1982-01-23,1300.00,null,10)\n"
          + "(7369,SMITH,CLERK,7902,1980-12-17,800.00,null,20)\n"
          + "(7566,JONES,MANAGER,7839,1981-02-04,2975.00,null,20)\n"
          + "(7788,SCOTT,ANALYST,7566,1987-04-19,3000.00,null,20)\n"
          + "(7876,ADAMS,CLERK,7788,1987-05-23,1100.00,null,20)\n"
          + "(7902,FORD,ANALYST,7566,1981-12-03,3000.00,null,20)\n";
  
      final String sql = ""
          + "SELECT *\n"
          + "FROM scott.EMP\n"
          + "WHERE DEPTNO = 10\n"
          + "UNION ALL\n"
          + "SELECT *\n"
          + "FROM scott.EMP\n"
          + "WHERE DEPTNO = 20";
      pig(script)
          .assertRel("B1", false,
              hasTree("LogicalFilter(condition=[=($7, 10)])\n"
                + scan))
          .assertRel("B2", false,
              hasTree("LogicalFilter(condition=[=($7, 20)])\n"
                + scan))
          .assertRel(hasTree(plan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testUdf() {
      final String script = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n"
          + "B = FILTER A BY ENDSWITH(DNAME, 'LES');\n";
      final String plan = ""
          + "LogicalFilter(condition=[ENDSWITH(PIG_TUPLE($1, 'LES'))])\n"
          + "  LogicalTableScan(table=[[scott, DEPT]])\n";
      final String result = "(30,SALES,CHICAGO)\n";
      final String sql = ""
          + "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "WHERE ENDSWITH(PIG_TUPLE(DNAME, 'LES'))";
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testSimpleForEach1() {
      String script = ""
          + "A = LOAD 'testSchema.testTable' as (a:int, b:long, c:float, "
          + "d:double, e:chararray, f:bytearray, g:boolean, "
          + "h:datetime, i:biginteger, j:bigdecimal, k1:tuple(), "
          + "k2:tuple(k21:int, k22:float), l1:bag{}, "
          + "l2:bag{l21:(l22:int, l23:float)}, "
          + "m1:map[], m2:map[int], m3:map[(m3:float)]);\n"
          + "B = FOREACH A GENERATE a, a as a2, b, c, d, e, f, g, h, i, j, k2, "
          + "l2, m2, null as n:chararray;\n";
      final String plan = ""
          + "LogicalProject(a=[$0], a2=[$0], b=[$1], c=[$2], d=[$3], e=[$4], "
          + "f=[$5], g=[$6], h=[$7], i=[$8], j=[$9], k2=[$11], l2=[$13], "
          + "m2=[$15], n=[null:VARCHAR])\n"
          + "  LogicalTableScan(table=[[testSchema, testTable]])\n";
      final String sql = ""
          + "SELECT a, a AS a2, b, c, d, e, f, g, h, i, j, k2, l2, m2, "
          + "CAST(NULL AS VARCHAR CHARACTER SET ISO-8859-1) AS n\n"
          + "FROM testSchema.testTable";
      pig(script).assertRel(hasTree(plan))
          .assertSql(is(sql));
    }
  
    @Test void testSimpleForEach2() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = FOREACH A GENERATE DEPTNO + 10 as dept, MGR;\n";
      final String plan = ""
          + "LogicalProject(dept=[+($7, 10)], MGR=[$3])\n"
          + "  LogicalTableScan(table=[[scott, EMP]])\n";
  
      final String result = ""
          + "(30,7902)\n"
          + "(40,7698)\n"
          + "(40,7698)\n"
          + "(30,7839)\n"
          + "(40,7698)\n"
          + "(40,7839)\n"
          + "(20,7839)\n"
          + "(30,7566)\n"
          + "(20,null)\n"
          + "(40,7698)\n"
          + "(30,7788)\n"
          + "(40,7698)\n"
          + "(30,7566)\n"
          + "(20,7782)\n";
      final String sql = ""
          + "SELECT DEPTNO + 10 AS dept, MGR\n"
          + "FROM scott.EMP";
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testSimpleForEach3() {
      String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = FILTER A BY JOB != 'CLERK';\n"
          + "C = GROUP B BY (DEPTNO, JOB);\n"
          + "D = FOREACH C GENERATE flatten(group) as (dept, job), flatten(B);\n"
          + "E = ORDER D BY dept, job;\n";
      final String plan = ""
          + "LogicalSort(sort0=[$0], sort1=[$1], dir0=[ASC], dir1=[ASC])\n"
          + "  LogicalProject(dept=[$0], job=[$1], EMPNO=[$3], ENAME=[$4], "
          + "JOB=[$5], MGR=[$6], HIREDATE=[$7], SAL=[$8], COMM=[$9], "
          + "DEPTNO=[$10])\n"
          + "    LogicalCorrelate(correlation=[$cor0], joinType=[inner], "
          + "requiredColumns=[{2}])\n"
          + "      LogicalProject(dept=[$0.DEPTNO], job=[$0.JOB], B=[$1])\n"
          + "        LogicalProject(group=[ROW($0, $1)], B=[$2])\n"
          + "          LogicalAggregate(group=[{0, 1}], B=[COLLECT($2)])\n"
          + "            LogicalProject(DEPTNO=[$7], JOB=[$2], $f2=[ROW($0, "
          + "$1, $2, $3, $4, $5, $6, $7)])\n"
          + "              LogicalFilter(condition=[<>($2, 'CLERK')])\n"
          + "                LogicalTableScan(table=[[scott, EMP]])\n"
          + "      Uncollect\n"
          + "        LogicalProject($f0=[$cor0.B])\n"
          + "          LogicalValues(tuples=[[{ 0 }]])\n";
  
      final String sql = ""
          + "SELECT $cor1.DEPTNO AS dept, $cor1.JOB AS job, t30.EMPNO,"
          + " t30.ENAME, t30.JOB, t30.MGR, t30.HIREDATE,"
          + " t30.SAL, t30.COMM, t30.DEPTNO\n"
          + "FROM (SELECT DEPTNO, JOB, COLLECT(ROW(EMPNO, ENAME, JOB, MGR, "
          + "HIREDATE, SAL, COMM, DEPTNO)) AS $f2\n"
          + "    FROM scott.EMP\n"
          + "    WHERE JOB <> 'CLERK'\n"
          + "    GROUP BY DEPTNO, JOB) AS $cor1,\n"
          + "  LATERAL UNNEST((SELECT $cor1.$f2 AS $f0\n"
          + "      FROM (VALUES (0)) AS t (ZERO))) AS t30 (EMPNO, ENAME, JOB,"
          + " MGR, HIREDATE, SAL, COMM, DEPTNO)\n"
          + "ORDER BY $cor1.DEPTNO, $cor1.JOB";
      pig(script).assertRel(hasTree(plan))
          .assertSql(is(sql));
  
      // TODO fix Calcite execution
      final String result = ""
          + "(10,7782,CLARK,MANAGER,7839,1981-06-09,2450.00,null,10)\n"
          + "(10,7839,KING,PRESIDENT,null,1981-11-17,5000.00,null,10)\n"
          + "(20,7566,JONES,MANAGER,7839,1981-02-04,2975.00,null,20)\n"
          + "(20,7788,SCOTT,ANALYST,7566,1987-04-19,3000.00,null,20)\n"
          + "(20,7902,FORD,ANALYST,7566,1981-12-03,3000.00,null,20)\n"
          + "(30,7499,ALLEN,SALESMAN,7698,1981-02-20,1600.00,300.00,30)\n"
          + "(30,7521,WARD,SALESMAN,7698,1981-02-22,1250.00,500.00,30)\n"
          + "(30,7654,MARTIN,SALESMAN,7698,1981-09-28,1250.00,1400.00,30)\n"
          + "(30,7698,BLAKE,MANAGER,7839,1981-01-05,2850.00,null,30)\n"
          + "(30,7844,TURNER,SALESMAN,7698,1981-09-08,1500.00,0.00,30)\n";
      if (false) {
        pig(script).assertResult(is(result));
      }
    }
  
    @Test void testForEachNested() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = GROUP A BY DEPTNO;\n"
          + "C = FOREACH B {\n"
          + "      S = FILTER A BY JOB != 'CLERK';\n"
          + "      Y = FOREACH S GENERATE ENAME, JOB, DEPTNO, SAL;\n"
          + "      X = ORDER Y BY SAL;\n"
          + "      GENERATE group, COUNT(X) as cnt, flatten(X), BigDecimalMax(X.SAL);\n"
          + "}\n"
          + "D = ORDER C BY $0;\n";
      final String plan = ""
          + "LogicalSort(sort0=[$0], dir0=[ASC])\n"
          + "  LogicalProject(group=[$0], cnt=[$1], ENAME=[$4], JOB=[$5], "
          + "DEPTNO=[$6], SAL=[$7], $f3=[$3])\n"
          + "    LogicalCorrelate(correlation=[$cor1], joinType=[inner], "
          + "requiredColumns=[{2}])\n"
          + "      LogicalProject(group=[$0], cnt=[COUNT(PIG_BAG($2))], X=[$2], "
          + "$f3=[BigDecimalMax(PIG_BAG(MULTISET_PROJECTION($2, 3)))])\n"
          + "        LogicalCorrelate(correlation=[$cor0], joinType=[inner], "
          + "requiredColumns=[{1}])\n"
          + "          LogicalProject(group=[$0], A=[$1])\n"
          + "            LogicalAggregate(group=[{0}], A=[COLLECT($1)])\n"
          + "              LogicalProject(DEPTNO=[$7], $f1=[ROW($0, $1, $2, "
          + "$3, $4, $5, $6, $7)])\n"
          + "                LogicalTableScan(table=[[scott, EMP]])\n"
          + "          LogicalProject(X=[$1])\n"
          + "            LogicalAggregate(group=[{0}], X=[COLLECT($1)])\n"
          + "              LogicalProject($f0=['all'], $f1=[ROW($0, $1, $2, $3)])\n"
          + "                LogicalSort(sort0=[$3], dir0=[ASC])\n"
          + "                  LogicalProject(ENAME=[$1], JOB=[$2], "
          + "DEPTNO=[$7], SAL=[$5])\n"
          + "                    LogicalFilter(condition=[<>($2, 'CLERK')])\n"
          + "                      Uncollect\n"
          + "                        LogicalProject($f0=[$cor0.A])\n"
          + "                          LogicalValues(tuples=[[{ 0 }]])\n"
          + "      Uncollect\n"
          + "        LogicalProject($f0=[$cor1.X])\n"
          + "          LogicalValues(tuples=[[{ 0 }]])\n";
  
      final String result = ""
          + "(10,2,CLARK,MANAGER,10,2450.00,5000.00)\n"
          + "(10,2,KING,PRESIDENT,10,5000.00,5000.00)\n"
          + "(20,3,JONES,MANAGER,20,2975.00,3000.00)\n"
          + "(20,3,SCOTT,ANALYST,20,3000.00,3000.00)\n"
          + "(20,3,FORD,ANALYST,20,3000.00,3000.00)\n"
          + "(30,5,WARD,SALESMAN,30,1250.00,2850.00)\n"
          + "(30,5,MARTIN,SALESMAN,30,1250.00,2850.00)\n"
          + "(30,5,TURNER,SALESMAN,30,1500.00,2850.00)\n"
          + "(30,5,ALLEN,SALESMAN,30,1600.00,2850.00)\n"
          + "(30,5,BLAKE,MANAGER,30,2850.00,2850.00)\n";
  
      final String sql = ""
          + "SELECT $cor5.group, $cor5.cnt, t110.ENAME, t110.JOB, "
          + "t110.DEPTNO, t110.SAL, $cor5.$f3\n"
          + "FROM (SELECT $cor4.DEPTNO AS group, "
          + "COUNT(PIG_BAG(t8.X)) AS cnt, t8.X, "
          + "BigDecimalMax(PIG_BAG(MULTISET_PROJECTION(t8.X, 3))) AS $f3\n"
          + "    FROM (SELECT DEPTNO, COLLECT(ROW(EMPNO, ENAME, JOB, MGR, "
          + "HIREDATE, SAL, COMM, DEPTNO)) AS A\n"
          + "        FROM scott.EMP\n"
          + "        GROUP BY DEPTNO) AS $cor4,\n"
          + "      LATERAL (SELECT COLLECT($f1) AS X\n"
          + "        FROM (SELECT 'all' AS $f0, ROW(ENAME, JOB, DEPTNO, SAL) AS $f1\n"
          + "            FROM UNNEST((SELECT $cor4.A AS $f0\n"
          + "                  FROM (VALUES (0)) AS t (ZERO))) "
          + "AS t2 (EMPNO, ENAME, JOB, MGR, HIREDATE, SAL, COMM, DEPTNO)\n"
          + "            WHERE JOB <> 'CLERK'\n"
          + "            ORDER BY SAL) AS t6\n"
          + "        GROUP BY $f0) AS t8) AS $cor5,\n"
          + "  LATERAL UNNEST((SELECT $cor5.X AS $f0\n"
          + "      FROM (VALUES (0)) AS t (ZERO))) AS t110 (ENAME, JOB, DEPTNO, SAL)\n"
          + "ORDER BY $cor5.group";
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testUnionSameSchema() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = FILTER A BY DEPTNO == 10;\n"
          + "C = FILTER A BY DEPTNO == 20;\n"
          + "D = UNION B, C;\n";
      final String plan = ""
          + "LogicalUnion(all=[true])\n"
          + "  LogicalFilter(condition=[=($7, 10)])\n"
          + "    LogicalTableScan(table=[[scott, EMP]])\n"
          + "  LogicalFilter(condition=[=($7, 20)])\n"
          + "    LogicalTableScan(table=[[scott, EMP]])\n";
      final String result = ""
          + "(7782,CLARK,MANAGER,7839,1981-06-09,2450.00,null,10)\n"
          + "(7839,KING,PRESIDENT,null,1981-11-17,5000.00,null,10)\n"
          + "(7934,MILLER,CLERK,7782,1982-01-23,1300.00,null,10)\n"
          + "(7369,SMITH,CLERK,7902,1980-12-17,800.00,null,20)\n"
          + "(7566,JONES,MANAGER,7839,1981-02-04,2975.00,null,20)\n"
          + "(7788,SCOTT,ANALYST,7566,1987-04-19,3000.00,null,20)\n"
          + "(7876,ADAMS,CLERK,7788,1987-05-23,1100.00,null,20)\n"
          + "(7902,FORD,ANALYST,7566,1981-12-03,3000.00,null,20)\n";
      final String sql = ""
          + "SELECT *\n"
          + "FROM scott.EMP\n"
          + "WHERE DEPTNO = 10\n"
          + "UNION ALL\n"
          + "SELECT *\n"
          + "FROM scott.EMP\n"
          + "WHERE DEPTNO = 20";
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testUnionDifferentSchemas1() {
      final String script = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n"
          + "B = FOREACH A GENERATE DEPTNO, DNAME;\n"
          + "C = UNION ONSCHEMA A, B;\n";
      final String plan = ""
          + "LogicalUnion(all=[true])\n"
          + "  LogicalTableScan(table=[[scott, DEPT]])\n"
          + "  LogicalProject(DEPTNO=[$0], DNAME=[$1], LOC=[null:VARCHAR])\n"
          + "    LogicalProject(DEPTNO=[$0], DNAME=[$1])\n"
          + "      LogicalTableScan(table=[[scott, DEPT]])\n";
      final String optimizedPlan = ""
          + "LogicalUnion(all=[true])\n"
          + "  LogicalTableScan(table=[[scott, DEPT]])\n"
          + "  LogicalProject(DEPTNO=[$0], DNAME=[$1], LOC=[null:VARCHAR])\n"
          + "    LogicalTableScan(table=[[scott, DEPT]])\n";
      final String result = ""
          + "(10,ACCOUNTING,NEW YORK)\n"
          + "(20,RESEARCH,DALLAS)\n"
          + "(30,SALES,CHICAGO)\n"
          + "(40,OPERATIONS,BOSTON)\n"
          + "(10,ACCOUNTING,null)\n"
          + "(20,RESEARCH,null)\n"
          + "(30,SALES,null)\n"
          + "(40,OPERATIONS,null)\n";
      final String sql = ""
          + "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "UNION ALL\n"
          + "SELECT DEPTNO, DNAME, "
          + "CAST(NULL AS VARCHAR CHARACTER SET ISO-8859-1) AS LOC\n"
          + "FROM scott.DEPT";
      pig(script).assertRel(hasTree(plan))
          .assertOptimizedRel(hasTree(optimizedPlan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testUnionDifferentSchemas2() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = FILTER A BY DEPTNO == 10;\n"
          + "C = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n"
          + "D = UNION ONSCHEMA B, C;\n";
      final String plan = ""
          + "LogicalUnion(all=[true])\n"
          + "  LogicalProject(EMPNO=[$0], ENAME=[$1], JOB=[$2], MGR=[$3], "
          + "HIREDATE=[$4], SAL=[$5], COMM=[$6], DEPTNO=[$7], "
          + "DNAME=[null:VARCHAR], LOC=[null:VARCHAR])\n"
          + "    LogicalFilter(condition=[=($7, 10)])\n"
          + "      LogicalTableScan(table=[[scott, EMP]])\n"
          + "  LogicalProject(EMPNO=[null:INTEGER], ENAME=[null:VARCHAR], "
          + "JOB=[null:VARCHAR], MGR=[null:INTEGER], HIREDATE=[null:DATE], "
          + "SAL=[null:DECIMAL(19, 0)], COMM=[null:DECIMAL(19, 0)], DEPTNO=[$0], "
          + "DNAME=[$1], LOC=[$2])\n"
          + "    LogicalTableScan(table=[[scott, DEPT]])\n";
      final String result = ""
          + "(7782,CLARK,MANAGER,7839,1981-06-09,2450.00,null,10,null,null)\n"
          + "(7839,KING,PRESIDENT,null,1981-11-17,5000.00,null,10,null,"
          + "null)\n"
          + "(7934,MILLER,CLERK,7782,1982-01-23,1300.00,null,10,null,null)\n"
          + "(null,null,null,null,null,null,null,10,ACCOUNTING,NEW YORK)\n"
          + "(null,null,null,null,null,null,null,20,RESEARCH,DALLAS)\n"
          + "(null,null,null,null,null,null,null,30,SALES,CHICAGO)\n"
          + "(null,null,null,null,null,null,null,40,OPERATIONS,BOSTON)\n";
      final String sql = ""
          + "SELECT EMPNO, ENAME, JOB, MGR, HIREDATE, SAL, COMM, DEPTNO, "
          + "CAST(NULL AS VARCHAR CHARACTER SET ISO-8859-1) AS DNAME, "
          + "CAST(NULL AS VARCHAR CHARACTER SET ISO-8859-1) AS LOC\n"
          + "FROM scott.EMP\n"
          + "WHERE DEPTNO = 10\n"
          + "UNION ALL\n"
          + "SELECT CAST(NULL AS INTEGER) AS EMPNO, "
          + "CAST(NULL AS VARCHAR CHARACTER SET ISO-8859-1) AS ENAME, "
          + "CAST(NULL AS VARCHAR CHARACTER SET ISO-8859-1) AS JOB, "
          + "CAST(NULL AS INTEGER) AS MGR, "
          + "CAST(NULL AS DATE) AS HIREDATE, CAST(NULL AS DECIMAL(19, 0)) AS SAL, "
          + "CAST(NULL AS DECIMAL(19, 0)) AS COMM, DEPTNO, DNAME, LOC\n"
          + "FROM scott.DEPT";
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testJoin2Rels() {
      final String scanScript = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n";
      final String scanPlan = ""
          + "  LogicalTableScan(table=[[scott, EMP]])\n"
          + "  LogicalTableScan(table=[[scott, DEPT]])\n";
  
      final String innerScript = scanScript
          + "C = JOIN A BY DEPTNO, B BY DEPTNO;\n";
      final String plan = ""
          + "LogicalJoin(condition=[=($7, $8)], joinType=[inner])\n"
          + scanPlan;
      final String innerSql = ""
          + "SELECT *\n"
          + "FROM scott.EMP\n"
          + "  INNER JOIN scott.DEPT ON EMP.DEPTNO = DEPT.DEPTNO";
      pig(innerScript).assertRel(hasTree(plan))
          .assertSql(is(innerSql));
  
      final String leftScript =
          scanScript + "C = JOIN A BY DEPTNO LEFT OUTER, B BY DEPTNO;\n";
      final String leftSql = ""
          + "SELECT *\n"
          + "FROM scott.EMP\n"
          + "  LEFT JOIN scott.DEPT ON EMP.DEPTNO = DEPT.DEPTNO";
      final String leftPlan = ""
          + "LogicalJoin(condition=[=($7, $8)], joinType=[left])\n"
          + scanPlan;
      pig(leftScript).assertRel(hasTree(leftPlan))
          .assertSql(is(leftSql));
  
      final String rightScript = scanScript
          + "C = JOIN A BY DEPTNO RIGHT OUTER, B BY DEPTNO;\n";
      final String rightSql = ""
          + "SELECT *\n"
          + "FROM scott.EMP\n"
          + "  RIGHT JOIN scott.DEPT ON EMP.DEPTNO = DEPT.DEPTNO";
      final String rightPlan =
          "LogicalJoin(condition=[=($7, $8)], joinType=[right])\n"
              + scanPlan;
      pig(rightScript)
          .assertRel(hasTree(rightPlan))
          .assertSql(is(rightSql));
  
      final String fullScript = scanScript
          + "C = JOIN A BY DEPTNO FULL, B BY DEPTNO;\n";
      final String fullPlan = ""
          + "LogicalJoin(condition=[=($7, $8)], joinType=[full])\n"
          + scanPlan;
      final String fullSql = ""
          + "SELECT *\n"
          + "FROM scott.EMP\n"
          + "  FULL JOIN scott.DEPT ON EMP.DEPTNO = DEPT.DEPTNO";
      final String fullResult = ""
          + "(7369,SMITH,CLERK,7902,1980-12-17,800.00,null,20,20,"
          + "RESEARCH,DALLAS)\n"
          + "(7499,ALLEN,SALESMAN,7698,1981-02-20,1600.00,300.00,30,30,"
          + "SALES,CHICAGO)\n"
          + "(7521,WARD,SALESMAN,7698,1981-02-22,1250.00,500.00,30,30,"
          + "SALES,CHICAGO)\n"
          + "(7566,JONES,MANAGER,7839,1981-02-04,2975.00,null,20,20,"
          + "RESEARCH,DALLAS)\n"
          + "(7654,MARTIN,SALESMAN,7698,1981-09-28,1250.00,1400.00,30,30,"
          + "SALES,CHICAGO)\n"
          + "(7698,BLAKE,MANAGER,7839,1981-01-05,2850.00,null,30,30,"
          + "SALES,CHICAGO)\n"
          + "(7782,CLARK,MANAGER,7839,1981-06-09,2450.00,null,10,10,"
          + "ACCOUNTING,NEW YORK)\n"
          + "(7788,SCOTT,ANALYST,7566,1987-04-19,3000.00,null,20,20,"
          + "RESEARCH,DALLAS)\n"
          + "(7839,KING,PRESIDENT,null,1981-11-17,5000.00,null,10,10,"
          + "ACCOUNTING,NEW YORK)\n"
          + "(7844,TURNER,SALESMAN,7698,1981-09-08,1500.00,0.00,30,30,"
          + "SALES,CHICAGO)\n"
          + "(7876,ADAMS,CLERK,7788,1987-05-23,1100.00,null,20,20,"
          + "RESEARCH,DALLAS)\n"
          + "(7900,JAMES,CLERK,7698,1981-12-03,950.00,null,30,30,SALES,"
          + "CHICAGO)\n"
          + "(7902,FORD,ANALYST,7566,1981-12-03,3000.00,null,20,20,"
          + "RESEARCH,DALLAS)\n"
          + "(7934,MILLER,CLERK,7782,1982-01-23,1300.00,null,10,10,"
          + "ACCOUNTING,NEW YORK)\n"
          + "(null,null,null,null,null,null,null,null,40,OPERATIONS,"
          + "BOSTON)\n";
      pig(fullScript)
          .assertRel(hasTree(fullPlan))
          .assertSql(is(fullSql))
          .assertResult(is(fullResult));
    }
  
    @Test void testJoin3Rels() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n"
          + "C = FILTER B BY LOC == 'CHICAGO';\n"
          + "D = JOIN A BY DEPTNO, B BY DEPTNO, C BY DEPTNO;\n";
      final String plan = ""
          + "LogicalJoin(condition=[=($7, $11)], joinType=[inner])\n"
          + "  LogicalJoin(condition=[=($7, $8)], joinType=[inner])\n"
          + "    LogicalTableScan(table=[[scott, EMP]])\n"
          + "    LogicalTableScan(table=[[scott, DEPT]])\n"
          + "  LogicalFilter(condition=[=($2, 'CHICAGO')])\n"
          + "    LogicalTableScan(table=[[scott, DEPT]])\n";
      final String sql = ""
          + "SELECT *\n"
          + "FROM scott.EMP\n"
          + "  INNER JOIN scott.DEPT ON EMP.DEPTNO = DEPT.DEPTNO\n"
          + "  INNER JOIN (SELECT *\n"
          + "    FROM scott.DEPT\n"
          + "    WHERE LOC = 'CHICAGO') AS t ON EMP.DEPTNO = t.DEPTNO";
      final String result = ""
          + "(7499,ALLEN,SALESMAN,7698,1981-02-20,1600.00,300.00,30,30,"
          + "SALES,CHICAGO,30,SALES,"
          + "CHICAGO)\n"
          + "(7521,WARD,SALESMAN,7698,1981-02-22,1250.00,500.00,30,30,"
          + "SALES,CHICAGO,30,SALES,"
          + "CHICAGO)\n"
          + "(7654,MARTIN,SALESMAN,7698,1981-09-28,1250.00,1400.00,30,30,"
          + "SALES,CHICAGO,30,"
          + "SALES,CHICAGO)\n"
          + "(7698,BLAKE,MANAGER,7839,1981-01-05,2850.00,null,30,30,SALES,"
          + "CHICAGO,30,SALES,"
          + "CHICAGO)\n"
          + "(7844,TURNER,SALESMAN,7698,1981-09-08,1500.00,0.00,30,30,"
          + "SALES,CHICAGO,30,SALES,"
          + "CHICAGO)\n"
          + "(7900,JAMES,CLERK,7698,1981-12-03,950.00,null,30,30,SALES,"
          + "CHICAGO,30,SALES,"
          + "CHICAGO)\n";
      pig(script).assertRel(hasTree(plan))
          .assertSql(is(sql))
          .assertResult(is(result));
  
      final String script2 = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray,\n"
          + "    LOC:CHARARRAY);\n"
          + "C = FILTER B BY LOC == 'CHICAGO';\n"
          + "D = JOIN A BY (DEPTNO, ENAME), B BY (DEPTNO, DNAME),\n"
          + "    C BY (DEPTNO, DNAME);\n";
      final String plan2 = ""
          + "LogicalJoin(condition=[AND(=($7, $11), =($9, $12))], "
          + "joinType=[inner])\n"
          + "  LogicalJoin(condition=[AND(=($7, $8), =($1, $9))], "
          + "joinType=[inner])\n"
          + "    LogicalTableScan(table=[[scott, EMP]])\n"
          + "    LogicalTableScan(table=[[scott, DEPT]])\n"
          + "  LogicalFilter(condition=[=($2, 'CHICAGO')])\n"
          + "    LogicalTableScan(table=[[scott, DEPT]])\n";
      final String sql2 = ""
          + "SELECT *\n"
          + "FROM scott.EMP\n"
          + "  INNER JOIN scott.DEPT ON EMP.DEPTNO = DEPT.DEPTNO "
          + "AND EMP.ENAME = DEPT.DNAME\n"
          + "  INNER JOIN (SELECT *\n"
          + "    FROM scott.DEPT\n"
          + "    WHERE LOC = 'CHICAGO') AS t ON EMP.DEPTNO = t.DEPTNO "
          + "AND DEPT.DNAME = t.DNAME";
      pig(script2).assertRel(hasTree(plan2))
          .assertSql(is(sql2));
    }
  
    @Test void testCross() {
      final String script = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray,\n"
          + "    LOC:CHARARRAY);\n"
          + "B = FOREACH A GENERATE DEPTNO;\n"
          + "C = FILTER B BY DEPTNO <= 20;\n"
          + "D = CROSS B, C;\n";
      final String plan = ""
          + "LogicalJoin(condition=[true], joinType=[inner])\n"
          + "  LogicalProject(DEPTNO=[$0])\n"
          + "    LogicalTableScan(table=[[scott, DEPT]])\n"
          + "  LogicalFilter(condition=[<=($0, 20)])\n"
          + "    LogicalProject(DEPTNO=[$0])\n"
          + "      LogicalTableScan(table=[[scott, DEPT]])\n";
      final String sql = ""
          + "SELECT *\n"
          + "FROM (SELECT DEPTNO\n"
          + "    FROM scott.DEPT) AS t,\n"
          + "    (SELECT DEPTNO\n"
          + "    FROM scott.DEPT\n"
          + "    WHERE DEPTNO <= 20) AS t1";
      final String result = ""
          + "(10,10)\n"
          + "(10,20)\n"
          + "(20,10)\n"
          + "(20,20)\n"
          + "(30,10)\n"
          + "(30,20)\n"
          + "(40,10)\n"
          + "(40,20)\n";
      pig(script).assertRel(hasTree(plan))
          .assertSql(is(sql))
          .assertResult(is(result));
  
      final String script2 = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray,"
          + "    LOC:CHARARRAY);\n"
          + "B = FOREACH A GENERATE DEPTNO;\n"
          + "C = FILTER B BY DEPTNO <= 20;\n"
          + "D = FILTER B BY DEPTNO > 20;\n"
          + "E = CROSS B, C, D;\n";
  
      final String plan2 = ""
          + "LogicalJoin(condition=[true], joinType=[inner])\n"
          + "  LogicalJoin(condition=[true], joinType=[inner])\n"
          + "    LogicalProject(DEPTNO=[$0])\n"
          + "      LogicalTableScan(table=[[scott, DEPT]])\n"
          + "    LogicalFilter(condition=[<=($0, 20)])\n"
          + "      LogicalProject(DEPTNO=[$0])\n"
          + "        LogicalTableScan(table=[[scott, DEPT]])\n"
          + "  LogicalFilter(condition=[>($0, 20)])\n"
          + "    LogicalProject(DEPTNO=[$0])\n"
          + "      LogicalTableScan(table=[[scott, DEPT]])\n";
  
      final String result2 = ""
          + "(10,10,30)\n"
          + "(10,10,40)\n"
          + "(10,20,30)\n"
          + "(10,20,40)\n"
          + "(20,10,30)\n"
          + "(20,10,40)\n"
          + "(20,20,30)\n"
          + "(20,20,40)\n"
          + "(30,10,30)\n"
          + "(30,10,40)\n"
          + "(30,20,30)\n"
          + "(30,20,40)\n"
          + "(40,10,30)\n"
          + "(40,10,40)\n"
          + "(40,20,30)\n"
          + "(40,20,40)\n";
      final String sql2 = ""
          + "SELECT *\n"
          + "FROM (SELECT DEPTNO\n"
          + "    FROM scott.DEPT) AS t,\n"
          + "    (SELECT DEPTNO\n"
          + "    FROM scott.DEPT\n"
          + "    WHERE DEPTNO <= 20) AS t1,\n"
          + "    (SELECT DEPTNO\n"
          + "    FROM scott.DEPT\n"
          + "    WHERE DEPTNO > 20) AS t3";
      pig(script2).assertRel(hasTree(plan2))
          .assertResult(is(result2))
          .assertSql(is(sql2));
    }
  
    @Test void testGroupby() {
      final String baseScript =
          "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n";
      final String basePlan = "      LogicalTableScan(table=[[scott, DEPT]])\n";
      final String script = baseScript + "B = GROUP A BY DEPTNO;\n";
      final String plan = ""
          + "LogicalProject(group=[$0], A=[$1])\n"
          + "  LogicalAggregate(group=[{0}], A=[COLLECT($1)])\n"
          + "    LogicalProject(DEPTNO=[$0], $f1=[ROW($0, $1, $2)])\n"
          + basePlan;
      final String result = ""
          + "(20,{(20,RESEARCH,DALLAS)})\n"
          + "(40,{(40,OPERATIONS,BOSTON)})\n"
          + "(10,{(10,ACCOUNTING,NEW YORK)})\n"
          + "(30,{(30,SALES,CHICAGO)})\n";
  
      final String sql = ""
          + "SELECT DEPTNO, COLLECT(ROW(DEPTNO, DNAME, LOC)) AS A\n"
          + "FROM scott.DEPT\n"
          + "GROUP BY DEPTNO";
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result))
          .assertSql(is(sql));
  
      final String script1 = baseScript + "B = GROUP A ALL;\n";
      final String plan1 = ""
          + "LogicalProject(group=[$0], A=[$1])\n"
          + "  LogicalAggregate(group=[{0}], A=[COLLECT($1)])\n"
          + "    LogicalProject($f0=['all'], $f1=[ROW($0, $1, $2)])\n"
          + basePlan;
      final String result1 = ""
          + "(all,{(10,ACCOUNTING,NEW YORK),(20,RESEARCH,DALLAS),"
          + "(30,SALES,CHICAGO),(40,OPERATIONS,BOSTON)})\n";
      pig(script1).assertResult(is(result1))
          .assertRel(hasTree(plan1));
    }
  
    @Test void testGroupby2() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = FOREACH A GENERATE EMPNO, ENAME, JOB, MGR, SAL, COMM, DEPTNO;\n"
          + "C = GROUP B BY (DEPTNO, JOB);\n";
      final String plan = ""
          + "LogicalProject(group=[ROW($0, $1)], B=[$2])\n"
          + "  LogicalAggregate(group=[{0, 1}], B=[COLLECT($2)])\n"
          + "    LogicalProject(DEPTNO=[$6], JOB=[$2], $f2=[ROW($0, $1, $2, "
          + "$3, $4, $5, $6)])\n"
          + "      LogicalProject(EMPNO=[$0], ENAME=[$1], JOB=[$2], MGR=[$3],"
          + " SAL=[$5], COMM=[$6], DEPTNO=[$7])\n"
          + "        LogicalTableScan(table=[[scott, EMP]])\n";
      final String result = ""
          + "({10, MANAGER},{(7782,CLARK,MANAGER,7839,2450.00,null,10)})\n"
          + "({10, PRESIDENT},{(7839,KING,PRESIDENT,null,5000.00,null,10)})\n"
          + "({20, CLERK},{(7369,SMITH,CLERK,7902,800.00,null,20),"
          + "(7876,ADAMS,CLERK,7788,1100.00,null,20)})\n"
          + "({30, MANAGER},{(7698,BLAKE,MANAGER,7839,2850.00,null,30)})\n"
          + "({20, ANALYST},{(7788,SCOTT,ANALYST,7566,3000.00,null,20),"
          + "(7902,FORD,ANALYST,7566,3000.00,null,20)})\n"
          + "({30, SALESMAN},{(7499,ALLEN,SALESMAN,7698,1600.00,300.00,30),"
          + "(7521,WARD,SALESMAN,7698,1250.00,500.00,30),"
          + "(7654,MARTIN,SALESMAN,7698,1250.00,1400.00,30),"
          + "(7844,TURNER,SALESMAN,7698,1500.00,0.00,30)})\n"
          + "({30, CLERK},{(7900,JAMES,CLERK,7698,950.00,null,30)})\n"
          + "({20, MANAGER},{(7566,JONES,MANAGER,7839,2975.00,null,20)})\n"
          + "({10, CLERK},{(7934,MILLER,CLERK,7782,1300.00,null,10)})\n";
  
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result));
    }
  
    @Test void testCubeCube() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = CUBE A BY CUBE(DEPTNO, JOB);\n"
          + "C = FOREACH B GENERATE group, COUNT(cube.ENAME);\n";
      final String plan = ""
          + "LogicalProject(group=[$0], $f1=[COUNT(PIG_BAG"
          + "(MULTISET_PROJECTION($1, 3)))])\n"
          + "  LogicalProject(group=[ROW($0, $1)], cube=[$2])\n"
          + "    LogicalAggregate(group=[{0, 1}], "
          + "groups=[[{0, 1}, {0}, {1}, {}]], cube=[COLLECT($2)])\n"
          + "      LogicalProject(DEPTNO=[$7], JOB=[$2], "
          + "$f2=[ROW($7, $2, $0, $1, $3, $4, $5, $6)])\n"
          + "        LogicalTableScan(table=[[scott, EMP]])\n";
      final String optimizedPlan = ""
          + "LogicalProject(group=[ROW($0, $1)], $f1=[CAST($2):BIGINT])\n"
          + "  LogicalAggregate(group=[{0, 1}], "
          + "groups=[[{0, 1}, {0}, {1}, {}]], agg#0=[COUNT($2)])\n"
          + "    LogicalProject(DEPTNO=[$7], JOB=[$2], ENAME=[$1])\n"
          + "      LogicalTableScan(table=[[scott, EMP]])\n";
      final String result = ""
          + "({30, SALESMAN},4)\n"
          + "({30, null},6)\n"
          + "({10, null},3)\n"
          + "({null, PRESIDENT},1)\n"
          + "({30, MANAGER},1)\n"
          + "({20, MANAGER},1)\n"
          + "({20, ANALYST},2)\n"
          + "({10, MANAGER},1)\n"
          + "({null, CLERK},4)\n"
          + "(null,14)\n"
          + "({20, null},5)\n"
          + "({10, PRESIDENT},1)\n"
          + "({null, ANALYST},2)\n"
          + "({null, SALESMAN},4)\n"
          + "({30, CLERK},1)\n"
          + "({10, CLERK},1)\n"
          + "({20, CLERK},2)\n"
          + "({null, MANAGER},3)\n";
  
      final String sql = ""
          + "SELECT ROW(DEPTNO, JOB) AS group,"
          + " CAST(COUNT(ENAME) AS BIGINT) AS $f1\n"
          + "FROM scott.EMP\n"
          + "GROUP BY CUBE(DEPTNO, JOB)";
      pig(script).assertRel(hasTree(plan))
          .assertOptimizedRel(hasTree(optimizedPlan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testCubeRollup() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = CUBE A BY ROLLUP(DEPTNO, JOB);\n"
          + "C = FOREACH B GENERATE group, COUNT(cube.ENAME);\n";
      final String plan = ""
          + "LogicalProject(group=[$0], $f1=[COUNT(PIG_BAG"
          + "(MULTISET_PROJECTION($1, 3)))])\n"
          + "  LogicalProject(group=[ROW($0, $1)], cube=[$2])\n"
          + "    LogicalAggregate(group=[{0, 1}], groups=[[{0, 1}, {1}, {}]],"
          + " cube=[COLLECT($2)])\n"
          + "      LogicalProject(DEPTNO=[$7], JOB=[$2], $f2=[ROW($7, $2, $0,"
          + " $1, $3, $4, $5, $6)])\n"
          + "        LogicalTableScan(table=[[scott, EMP]])\n";
      final String optimizedPlan = ""
          + "LogicalProject(group=[ROW($0, $1)], $f1=[CAST($2):BIGINT])\n"
          + "  LogicalAggregate(group=[{0, 1}], groups=[[{0, 1}, {1}, {}]], "
          + "agg#0=[COUNT($2)])\n"
          + "    LogicalProject(DEPTNO=[$7], JOB=[$2], ENAME=[$1])\n"
          + "      LogicalTableScan(table=[[scott, EMP]])\n";
      final String result = ""
          + "({30, SALESMAN},4)\n"
          + "({null, PRESIDENT},1)\n"
          + "({30, MANAGER},1)\n"
          + "({20, MANAGER},1)\n"
          + "({20, ANALYST},2)\n"
          + "({10, MANAGER},1)\n"
          + "({null, CLERK},4)\n"
          + "(null,14)\n"
          + "({10, PRESIDENT},1)\n"
          + "({null, ANALYST},2)\n"
          + "({null, SALESMAN},4)\n"
          + "({30, CLERK},1)\n"
          + "({10, CLERK},1)\n"
          + "({20, CLERK},2)\n"
          + "({null, MANAGER},3)\n";
      final String sql = ""
          + "SELECT ROW(DEPTNO, JOB) AS group, "
          + "CAST(COUNT(ENAME) AS BIGINT) AS $f1\n"
          + "FROM scott.EMP\n"
          + "GROUP BY ROLLUP(DEPTNO, JOB)";
      pig(script).assertRel(hasTree(plan))
          .assertOptimizedRel(hasTree(optimizedPlan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testMultisetProjection() {
      final String script = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray,\n"
          + "    LOC:CHARARRAY);\n"
          + "B = GROUP A BY DEPTNO;\n"
          + "C = FOREACH B GENERATE A.(DEPTNO, DNAME);\n";
      final String plan = ""
          + "LogicalProject($f0=[MULTISET_PROJECTION($1, 0, 1)])\n"
          + "  LogicalProject(group=[$0], A=[$1])\n"
          + "    LogicalAggregate(group=[{0}], A=[COLLECT($1)])\n"
          + "      LogicalProject(DEPTNO=[$0], $f1=[ROW($0, $1, $2)])\n"
          + "        LogicalTableScan(table=[[scott, DEPT]])\n";
      final String optimizedPlan = ""
          + "LogicalProject($f0=[$1])\n"
          + "  LogicalAggregate(group=[{0}], agg#0=[COLLECT($1)])\n"
          + "    LogicalProject(DEPTNO=[$0], $f2=[ROW($0, $1)])\n"
          + "      LogicalTableScan(table=[[scott, DEPT]])\n";
      final String result = ""
          + "({(20,RESEARCH)})\n"
          + "({(40,OPERATIONS)})\n"
          + "({(10,ACCOUNTING)})\n"
          + "({(30,SALES)})\n";
      final String sql = ""
          + "SELECT COLLECT(ROW(DEPTNO, DNAME)) AS $f0\n"
          + "FROM scott.DEPT\n"
          + "GROUP BY DEPTNO";
      pig(script).assertRel(hasTree(plan))
          .assertOptimizedRel(hasTree(optimizedPlan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testOrderBy() {
      final String scan = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray,\n"
          + "    LOC:CHARARRAY);\n";
      final String scanPlan =
          "  LogicalTableScan(table=[[scott, DEPT]])\n";
  
      final String plan0 = "LogicalSort(sort0=[$1], dir0=[ASC])\n"
          + scanPlan;
      final String script0 = scan + "B = ORDER A BY DNAME;\n";
      final String sql0 = "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "ORDER BY DNAME";
      final String result0 = ""
          + "(10,ACCOUNTING,NEW YORK)\n"
          + "(40,OPERATIONS,BOSTON)\n"
          + "(20,RESEARCH,DALLAS)\n"
          + "(30,SALES,CHICAGO)\n";
      pig(script0).assertRel(hasTree(plan0))
          .assertSql(is(sql0))
          .assertResult(is(result0));
  
      final String plan1 = "LogicalSort(sort0=[$1], dir0=[DESC])\n"
          + scanPlan;
      final String script1 = scan + "B = ORDER A BY DNAME DESC;\n";
      final String sql1 = "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "ORDER BY DNAME DESC";
      pig(script1).assertRel(hasTree(plan1))
          .assertSql(is(sql1));
  
      final String plan2 = ""
          + "LogicalSort(sort0=[$2], sort1=[$0], dir0=[DESC], dir1=[ASC])\n"
          + scanPlan;
      final String script2 = scan + "B = ORDER A BY LOC DESC, DEPTNO;\n";
      final String sql2 = "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "ORDER BY LOC DESC, DEPTNO";
      pig(script2).assertRel(hasTree(plan2))
          .assertSql(is(sql2));
  
      final String plan3 = ""
          + "LogicalSort(sort0=[$0], sort1=[$1], sort2=[$2], dir0=[ASC], dir1=[ASC], dir2=[ASC])\n"
          + scanPlan;
      final String script3 = scan + "B = ORDER A BY *;\n";
      final String sql3 = "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "ORDER BY DEPTNO, DNAME, LOC";
      pig(script3).assertRel(hasTree(plan3))
          .assertSql(is(sql3));
  
      final String plan4 = ""
          + "LogicalSort(sort0=[$0], sort1=[$1], sort2=[$2], dir0=[DESC], dir1=[DESC], dir2=[DESC])\n"
          + scanPlan;
      final String script4 = scan + "B = ORDER A BY * DESC;\n";
      final String result4 = ""
          + "(40,OPERATIONS,BOSTON)\n"
          + "(30,SALES,CHICAGO)\n"
          + "(20,RESEARCH,DALLAS)\n"
          + "(10,ACCOUNTING,NEW YORK)\n";
      pig(script4).assertRel(hasTree(plan4))
          .assertResult(is(result4));
    }
  
    @Test void testRank() {
      final String base = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = FOREACH A GENERATE EMPNO, JOB, DEPTNO;\n";
      final String basePlan = ""
          + "  LogicalProject(EMPNO=[$0], JOB=[$2], DEPTNO=[$7])\n"
          + "    LogicalTableScan(table=[[scott, EMP]])\n";
      final String optimizedPlan = ""
          + "LogicalProject(rank_B=[$3], EMPNO=[$0], JOB=[$1], DEPTNO=[$2])\n"
          + "  LogicalWindow(window#0=[window(order by [2, 1 DESC] "
          + "aggs [RANK()])])\n"
          + "    LogicalProject(EMPNO=[$0], JOB=[$2], DEPTNO=[$7])\n"
          + "      LogicalTableScan(table=[[scott, EMP]])\n";
  
      final String script = base + "C = RANK B BY DEPTNO ASC, JOB DESC;\n";
      final String plan = ""
          + "LogicalProject(rank_B=[RANK() OVER (ORDER BY $2, $1 DESC)], "
          + "EMPNO=[$0], JOB=[$1], DEPTNO=[$2])\n"
          + basePlan;
      final String result = ""
          + "(1,7839,PRESIDENT,10)\n"
          + "(2,7782,MANAGER,10)\n"
          + "(3,7934,CLERK,10)\n"
          + "(4,7566,MANAGER,20)\n"
          + "(5,7369,CLERK,20)\n"
          + "(5,7876,CLERK,20)\n"
          + "(7,7788,ANALYST,20)\n"
          + "(7,7902,ANALYST,20)\n"
          + "(9,7499,SALESMAN,30)\n"
          + "(9,7521,SALESMAN,30)\n"
          + "(9,7654,SALESMAN,30)\n"
          + "(9,7844,SALESMAN,30)\n"
          + "(13,7698,MANAGER,30)\n"
          + "(14,7900,CLERK,30)\n";
      final String sql = ""
          + "SELECT RANK() OVER (ORDER BY DEPTNO, JOB DESC RANGE BETWEEN "
          + "UNBOUNDED PRECEDING AND CURRENT ROW) AS rank_B, EMPNO, JOB, DEPTNO\n"
          + "FROM scott.EMP";
      pig(script).assertRel(hasTree(plan))
          .assertOptimizedRel(hasTree(optimizedPlan))
          .assertResult(is(result))
          .assertSql(is(sql));
  
      final String script2 = base + "C = RANK B BY DEPTNO ASC, JOB DESC DENSE;\n";
      final String optimizedPlan2 = ""
          + "LogicalProject(rank_B=[$3], EMPNO=[$0], JOB=[$1], DEPTNO=[$2])\n"
          + "  LogicalWindow(window#0=[window(order by [2, 1 DESC] "
          + "aggs [DENSE_RANK()])"
          + "])\n"
          + "    LogicalProject(EMPNO=[$0], JOB=[$2], DEPTNO=[$7])\n"
          + "      LogicalTableScan(table=[[scott, EMP]])\n";
      final String plan2 = ""
          + "LogicalProject(rank_B=[DENSE_RANK() OVER (ORDER BY $2, $1 DESC)], "
          + "EMPNO=[$0], JOB=[$1], DEPTNO=[$2])\n"
          + basePlan;
      final String result2 = ""
          + "(1,7839,PRESIDENT,10)\n"
          + "(2,7782,MANAGER,10)\n"
          + "(3,7934,CLERK,10)\n"
          + "(4,7566,MANAGER,20)\n"
          + "(5,7369,CLERK,20)\n"
          + "(5,7876,CLERK,20)\n"
          + "(6,7788,ANALYST,20)\n"
          + "(6,7902,ANALYST,20)\n"
          + "(7,7499,SALESMAN,30)\n"
          + "(7,7521,SALESMAN,30)\n"
          + "(7,7654,SALESMAN,30)\n"
          + "(7,7844,SALESMAN,30)\n"
          + "(8,7698,MANAGER,30)\n"
          + "(9,7900,CLERK,30)\n";
      final String sql2 = ""
          + "SELECT DENSE_RANK() OVER (ORDER BY DEPTNO, JOB DESC RANGE BETWEEN "
          + "UNBOUNDED PRECEDING AND CURRENT ROW) AS rank_B, EMPNO, JOB, DEPTNO\n"
          + "FROM scott.EMP";
      pig(script2).assertRel(hasTree(plan2))
          .assertOptimizedRel(hasTree(optimizedPlan2))
          .assertResult(is(result2))
          .assertSql(is(sql2));
    }
  
    @Test void testLimit() {
      final String scan = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray,\n"
          + "    LOC:CHARARRAY);\n";
      final String scanPlan = "  LogicalTableScan(table=[[scott, DEPT]])\n";
  
      final String plan1 = "LogicalSort(sort0=[$1], dir0=[ASC], fetch=[2])\n"
          + scanPlan;
      final String script1 = scan
          + "B = ORDER A BY DNAME;\n"
          + "C = LIMIT B 2;\n";
      final String sql1 = "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "ORDER BY DNAME\n"
          + "FETCH NEXT 2 ROWS ONLY";
      final String result1 = ""
          + "(10,ACCOUNTING,NEW YORK)\n"
          + "(40,OPERATIONS,BOSTON)\n";
      pig(script1).assertRel(hasTree(plan1))
          .assertSql(is(sql1))
          .assertResult(is(result1));
  
      final String plan2 = "LogicalSort(fetch=[2])\n"
          + scanPlan;
      final String script2 = scan + "B = LIMIT A 2;\n";
      final String sql2 = "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "FETCH NEXT 2 ROWS ONLY";
      pig(script2).assertRel(hasTree(plan2))
          .assertSql(is(sql2));
    }
  
    @Test void testDistinct() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = FOREACH A GENERATE DEPTNO;\n"
          + "C = DISTINCT B;\n";
      final String plan = ""
          + "LogicalAggregate(group=[{0}])\n"
          + "  LogicalProject(DEPTNO=[$7])\n"
          + "    LogicalTableScan(table=[[scott, EMP]])\n";
  
      final String result = ""
          + "(20)\n"
          + "(10)\n"
          + "(30)\n";
      final String sql = "SELECT DEPTNO\n"
          + "FROM scott.EMP\n"
          + "GROUP BY DEPTNO";
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testAggregate() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = GROUP A BY DEPTNO;\n"
          + "C = FOREACH B GENERATE group, COUNT(A), BigDecimalSum(A.SAL);\n";
      final String plan = ""
          + "LogicalProject(group=[$0], $f1=[COUNT(PIG_BAG($1))], "
          + "$f2=[BigDecimalSum(PIG_BAG(MULTISET_PROJECTION($1, 5)))])\n"
          + "  LogicalProject(group=[$0], A=[$1])\n"
          + "    LogicalAggregate(group=[{0}], A=[COLLECT($1)])\n"
          + "      LogicalProject(DEPTNO=[$7], $f1=[ROW($0, $1, $2, $3, $4, $5, $6, $7)])\n"
          + "        LogicalTableScan(table=[[scott, EMP]])\n";
      final String optimizedPlan = ""
          + "LogicalProject(group=[$0], $f1=[CAST($1):BIGINT], $f2=[CAST($2):DECIMAL(19, 0)])\n"
          + "  LogicalAggregate(group=[{0}], agg#0=[COUNT()], agg#1=[SUM($1)])\n"
          + "    LogicalProject(DEPTNO=[$7], SAL=[$5])\n"
          + "      LogicalTableScan(table=[[scott, EMP]])\n";
      final String result = ""
          + "(20,5,10875.00)\n"
          + "(10,3,8750.00)\n"
          + "(30,6,9400.00)\n";
      final String sql = ""
          + "SELECT DEPTNO AS group, CAST(COUNT(*) AS BIGINT) AS $f1, CAST(SUM(SAL) AS "
          + "DECIMAL(19, 0)) AS $f2\n"
          + "FROM scott.EMP\n"
          + "GROUP BY DEPTNO";
      pig(script).assertRel(hasTree(plan))
          .assertOptimizedRel(hasTree(optimizedPlan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testAggregate2() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = GROUP A BY (DEPTNO, MGR, HIREDATE);\n"
          + "C = FOREACH B GENERATE group, COUNT(A), SUM(A.SAL) as salSum;\n"
          + "D = ORDER C BY salSum;\n";
      final String plan = ""
          + "LogicalSort(sort0=[$2], dir0=[ASC])\n"
          + "  LogicalProject(group=[$0], $f1=[COUNT(PIG_BAG($1))], "
          + "salSum=[BigDecimalSum(PIG_BAG(MULTISET_PROJECTION($1, 5)))])\n"
          + "    LogicalProject(group=[ROW($0, $1, $2)], A=[$3])\n"
          + "      LogicalAggregate(group=[{0, 1, 2}], A=[COLLECT($3)])\n"
          + "        LogicalProject(DEPTNO=[$7], MGR=[$3], HIREDATE=[$4], "
          + "$f3=[ROW($0, $1, $2, $3, $4, $5, $6, $7)])\n"
          + "          LogicalTableScan(table=[[scott, EMP]])\n";
      final String optimizedPlan = ""
          + "LogicalSort(sort0=[$2], dir0=[ASC])\n"
          + "  LogicalProject(group=[ROW($0, $1, $2)], $f1=[CAST($3):BIGINT], "
          + "salSum=[CAST($4):DECIMAL(19, 0)])\n"
          + "    LogicalAggregate(group=[{0, 1, 2}], agg#0=[COUNT()], agg#1=[SUM($3)])\n"
          + "      LogicalProject(DEPTNO=[$7], MGR=[$3], HIREDATE=[$4], SAL=[$5])\n"
          + "        LogicalTableScan(table=[[scott, EMP]])\n";
      pig(script).assertRel(hasTree(plan))
          .assertOptimizedRel(hasTree(optimizedPlan));
  
      final String result = ""
          + "({20, 7902, 1980-12-17},1,800.00)\n"
          + "({30, 7698, 1981-12-03},1,950.00)\n"
          + "({20, 7788, 1987-05-23},1,1100.00)\n"
          + "({30, 7698, 1981-09-28},1,1250.00)\n"
          + "({30, 7698, 1981-02-22},1,1250.00)\n"
          + "({10, 7782, 1982-01-23},1,1300.00)\n"
          + "({30, 7698, 1981-09-08},1,1500.00)\n"
          + "({30, 7698, 1981-02-20},1,1600.00)\n"
          + "({10, 7839, 1981-06-09},1,2450.00)\n"
          + "({30, 7839, 1981-01-05},1,2850.00)\n"
          + "({20, 7839, 1981-02-04},1,2975.00)\n"
          + "({20, 7566, 1981-12-03},1,3000.00)\n"
          + "({20, 7566, 1987-04-19},1,3000.00)\n"
          + "({10, null, 1981-11-17},1,5000.00)\n";
      final String sql = ""
          + "SELECT ROW(DEPTNO, MGR, HIREDATE) AS group, CAST(COUNT(*) AS "
          + "BIGINT) AS $f1, CAST(SUM(SAL) AS DECIMAL(19, 0)) AS salSum\n"
          + "FROM scott.EMP\n"
          + "GROUP BY DEPTNO, MGR, HIREDATE\n"
          + "ORDER BY 3";
      pig(script).assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testAggregate2half() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = GROUP A BY (DEPTNO, MGR, HIREDATE);\n"
          + "C = FOREACH B GENERATE flatten(group) as (DEPTNO, MGR, HIREDATE),\n"
          + "    COUNT(A), SUM(A.SAL) as salSum, MAX(A.DEPTNO) as maxDep,\n"
          + "    MIN(A.HIREDATE) as minHire;\n"
          + "D = ORDER C BY salSum;\n";
      final String plan = ""
          + "LogicalSort(sort0=[$4], dir0=[ASC])\n"
          + "  LogicalProject(DEPTNO=[$0.DEPTNO], MGR=[$0.MGR], HIREDATE=[$0.HIREDATE], "
          + "$f3=[COUNT(PIG_BAG($1))], salSum=[BigDecimalSum(PIG_BAG(MULTISET_PROJECTION($1, 5)))]"
          + ", maxDep=[IntMax(PIG_BAG(MULTISET_PROJECTION($1, 7)))], minHire=[DateTimeMin(PIG_BAG"
          + "(MULTISET_PROJECTION($1, 4)))])\n"
          + "    LogicalProject(group=[ROW($0, $1, $2)], A=[$3])\n"
          + "      LogicalAggregate(group=[{0, 1, 2}], A=[COLLECT($3)])\n"
          + "        LogicalProject(DEPTNO=[$7], MGR=[$3], HIREDATE=[$4], "
          + "$f3=[ROW($0, $1, $2, $3, $4, $5, $6, $7)])\n"
          + "          LogicalTableScan(table=[[scott, EMP]])\n";
  
      final String optimizedPlan = ""
          + "LogicalSort(sort0=[$4], dir0=[ASC])\n"
          + "  LogicalProject(DEPTNO=[$0], MGR=[$1], HIREDATE=[$2], $f3=[CAST($3):BIGINT], "
          + "salSum=[CAST($4):DECIMAL(19, 0)], maxDep=[CAST($5):INTEGER], minHire=[$6])\n"
          + "    LogicalAggregate(group=[{0, 1, 2}], agg#0=[COUNT()"
          + "], agg#1=[SUM($3)], agg#2=[MAX($0)], agg#3=[MIN($2)])\n"
          + "      LogicalProject(DEPTNO=[$7], MGR=[$3], HIREDATE=[$4], SAL=[$5])\n"
          + "        LogicalTableScan(table=[[scott, EMP]])\n";
      pig(script).assertRel(hasTree(plan))
          .assertOptimizedRel(hasTree(optimizedPlan));
  
      final String script2 = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = GROUP A BY (DEPTNO, MGR, HIREDATE);\n"
          + "C = FOREACH B GENERATE group.DEPTNO, COUNT(A), SUM(A.SAL) as salSum, "
          + "group.MGR, MAX(A.DEPTNO) as maxDep, MIN(A.HIREDATE) as minHire;\n"
          + "D = ORDER C BY salSum;\n";
      final String plan2 = ""
          + "LogicalSort(sort0=[$2], dir0=[ASC])\n"
          + "  LogicalProject(DEPTNO=[$0.DEPTNO], $f1=[COUNT(PIG_BAG($1))], "
          + "salSum=[BigDecimalSum(PIG_BAG(MULTISET_PROJECTION($1, 5)))], "
          + "MGR=[$0.MGR], maxDep=[IntMax(PIG_BAG(MULTISET_PROJECTION($1, 7)"
          + "))], minHire=[DateTimeMin(PIG_BAG(MULTISET_PROJECTION($1, 4)))])\n"
          + "    LogicalProject(group=[ROW($0, $1, $2)], A=[$3])\n"
          + "      LogicalAggregate(group=[{0, 1, 2}], A=[COLLECT($3)])\n"
          + "        LogicalProject(DEPTNO=[$7], MGR=[$3], HIREDATE=[$4], "
          + "$f3=[ROW($0, $1, $2, $3, $4, $5, $6, $7)])\n"
          + "          LogicalTableScan(table=[[scott, EMP]])\n";
  
      final String optimizedPlan2 = ""
          + "LogicalSort(sort0=[$2], dir0=[ASC])\n"
          + "  LogicalProject(DEPTNO=[$0], $f1=[CAST($3):BIGINT], salSum=[CAST($4):DECIMAL(19, 0)]"
          + ", MGR=[$1], maxDep=[CAST($5):INTEGER], minHire=[$6])\n"
          + "    LogicalAggregate(group=[{0, 1, 2}], agg#0=[COUNT()], agg#1=[SUM($3)], "
          + "agg#2=[MAX($0)], agg#3=[MIN($2)])\n"
          + "      LogicalProject(DEPTNO=[$7], MGR=[$3], HIREDATE=[$4], SAL=[$5])\n"
          + "        LogicalTableScan(table=[[scott, EMP]])\n";
      pig(script2).assertRel(hasTree(plan2))
          .assertOptimizedRel(hasTree(optimizedPlan2));
    }
  
    @Test void testAggregate3() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = GROUP A BY (DEPTNO, MGR, HIREDATE);\n"
          + "C = FOREACH B GENERATE group, COUNT(A) + 1, BigDecimalSum(A.SAL) as "
          + "salSum, BigDecimalSum(A.SAL) / COUNT(A) as salAvg;\n"
          + "D = ORDER C BY salSum;\n";
      final String plan = ""
          + "LogicalSort(sort0=[$2], dir0=[ASC])\n"
          + "  LogicalProject(group=[$0], $f1=[+(COUNT(PIG_BAG($1)), 1)], "
          + "salSum=[BigDecimalSum(PIG_BAG(MULTISET_PROJECTION($1, 5)))], "
          + "salAvg=[/(BigDecimalSum(PIG_BAG(MULTISET_PROJECTION($1, 5))), "
          + "CAST(COUNT(PIG_BAG($1))):DECIMAL(19, 0))])\n"
          + "    LogicalProject(group=[ROW($0, $1, $2)], A=[$3])\n"
          + "      LogicalAggregate(group=[{0, 1, 2}], A=[COLLECT($3)])\n"
          + "        LogicalProject(DEPTNO=[$7], MGR=[$3], HIREDATE=[$4], "
          + "$f3=[ROW($0, $1, $2, $3, $4, $5, $6, $7)])\n"
          + "          LogicalTableScan(table=[[scott, EMP]])\n";
      final String optimizedPlan = ""
          + "LogicalSort(sort0=[$2], dir0=[ASC])\n"
          + "  LogicalProject(group=[ROW($0, $1, $2)], $f1=[+($3, 1)], salSum=[CAST($4):DECIMAL(19,"
          + " 0)], salAvg=[/(CAST($4):DECIMAL(19, 0), CAST($3):DECIMAL(19, 0))])\n"
          + "    LogicalAggregate(group=[{0, 1, 2}], agg#0=[COUNT()], agg#1=[SUM($3)])\n"
          + "      LogicalProject(DEPTNO=[$7], MGR=[$3], HIREDATE=[$4], SAL=[$5])\n"
          + "        LogicalTableScan(table=[[scott, EMP]])\n";
      final String result = ""
          + "({20, 7902, 1980-12-17},2,800.00,800.00)\n"
          + "({30, 7698, 1981-12-03},2,950.00,950.00)\n"
          + "({20, 7788, 1987-05-23},2,1100.00,1100.00)\n"
          + "({30, 7698, 1981-09-28},2,1250.00,1250.00)\n"
          + "({30, 7698, 1981-02-22},2,1250.00,1250.00)\n"
          + "({10, 7782, 1982-01-23},2,1300.00,1300.00)\n"
          + "({30, 7698, 1981-09-08},2,1500.00,1500.00)\n"
          + "({30, 7698, 1981-02-20},2,1600.00,1600.00)\n"
          + "({10, 7839, 1981-06-09},2,2450.00,2450.00)\n"
          + "({30, 7839, 1981-01-05},2,2850.00,2850.00)\n"
          + "({20, 7839, 1981-02-04},2,2975.00,2975.00)\n"
          + "({20, 7566, 1981-12-03},2,3000.00,3000.00)\n"
          + "({20, 7566, 1987-04-19},2,3000.00,3000.00)\n"
          + "({10, null, 1981-11-17},2,5000.00,5000.00)\n";
      final String sql = ""
          + "SELECT ROW(DEPTNO, MGR, HIREDATE) AS group, COUNT(*) + 1 AS $f1, CAST(SUM(SAL) AS "
          + "DECIMAL(19, 0)) AS salSum, CAST(SUM(SAL) AS DECIMAL(19, 0)) / CAST(COUNT(*) AS DECIMAL"
          + "(19, 0)) AS salAvg\n"
          + "FROM scott.EMP\n"
          + "GROUP BY DEPTNO, MGR, HIREDATE\n"
          + "ORDER BY 3";
      pig(script).assertRel(hasTree(plan))
          .assertOptimizedRel(hasTree(optimizedPlan))
          .assertResult(is(result))
          .assertSql(is(sql));
  
      final String script2 = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = GROUP A BY (DEPTNO, MGR, HIREDATE);\n"
          + "C = FOREACH B GENERATE group, COUNT(A) + 1, BigDecimalSum(A.SAL) as salSum, "
          + "BigDecimalSum(A.SAL) / COUNT(A) as salAvg, A;\n"
          + "D = ORDER C BY salSum;\n";
      final String sql2 = ""
          + "SELECT ROW(DEPTNO, MGR, HIREDATE) AS group, COUNT(*) + 1 AS $f1, CAST(SUM(SAL) AS "
          + "DECIMAL(19, 0)) AS salSum, CAST(SUM(SAL) AS DECIMAL(19, 0)) / CAST(COUNT(*) AS DECIMAL"
          + "(19, 0)) AS salAvg, COLLECT(ROW(EMPNO, ENAME, JOB, MGR, HIREDATE, SAL, COMM, DEPTNO)) "
          + "AS A\n"
          + "FROM scott.EMP\n"
          + "GROUP BY DEPTNO, MGR, HIREDATE\n"
          + "ORDER BY 3";
      pig(script2).assertSql(is(sql2));
  
      final String script3 = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = GROUP A BY (DEPTNO, MGR, HIREDATE);\n"
          + "C = FOREACH B GENERATE group, A, COUNT(A);\n";
      final String sql3 = ""
          + "SELECT ROW(DEPTNO, MGR, HIREDATE) AS group, COLLECT(ROW(EMPNO, ENAME, "
          + "JOB, MGR, HIREDATE, SAL, COMM, DEPTNO)) AS A, CAST(COUNT(*) AS BIGINT) "
          + "AS $f2\n"
          + "FROM scott.EMP\n"
          + "GROUP BY DEPTNO, MGR, HIREDATE";
      pig(script3).assertSql(is(sql3));
    }
  
    @Test void testAggregate4() {
      final String script = ""
          + "A = LOAD 'scott.EMP' as (EMPNO:int, ENAME:chararray,\n"
          + "    JOB:chararray, MGR:int, HIREDATE:datetime, SAL:bigdecimal,\n"
          + "    COMM:bigdecimal, DEPTNO:int);\n"
          + "B = GROUP A BY (DEPTNO, MGR, HIREDATE);\n"
          + "C = FOREACH B GENERATE FLATTEN(group) as (DEPTNO, MGR, HIREDATE), "
          + "COUNT(A), 1L as newCol, A.COMM as comArray, SUM(A.SAL) as salSum;\n"
          + "D = ORDER C BY salSum;\n";
      final String plan = ""
          + "LogicalSort(sort0=[$6], dir0=[ASC])\n"
          + "  LogicalProject(DEPTNO=[$0.DEPTNO], MGR=[$0.MGR], HIREDATE=[$0.HIREDATE], "
          + "$f3=[COUNT(PIG_BAG($1))], newCol=[1:BIGINT], comArray=[CAST(MULTISET_PROJECTION($1, 6)):RecordType(DECIMAL(19, 0) COMM) MULTISET], "
          + "salSum=[BigDecimalSum(PIG_BAG(MULTISET_PROJECTION($1, 5)))])\n"
          + "    LogicalProject(group=[ROW($0, $1, $2)], A=[$3])\n"
          + "      LogicalAggregate(group=[{0, 1, 2}], A=[COLLECT($3)])\n"
          + "        LogicalProject(DEPTNO=[$7], MGR=[$3], HIREDATE=[$4], "
          + "$f3=[ROW($0, $1, $2, $3, $4, $5, $6, $7)])\n"
          + "          LogicalTableScan(table=[[scott, EMP]])\n";
      final String optimizedPlan = ""
          + "LogicalSort(sort0=[$6], dir0=[ASC])\n"
          + "  LogicalProject(DEPTNO=[$0], MGR=[$1], HIREDATE=[$2], $f3=[CAST($3):BIGINT], "
          + "newCol=[1:BIGINT], comArray=[CAST($4):RecordType(DECIMAL(19, 0) COMM) MULTISET], salSum=[CAST($5):DECIMAL(19, 0)])\n"
          + "    LogicalAggregate(group=[{0, 1, 2}], agg#0=[COUNT()], agg#1=[COLLECT($3)], "
          + "agg#2=[SUM($4)])\n"
          + "      LogicalProject(DEPTNO=[$7], MGR=[$3], HIREDATE=[$4], COMM=[$6], SAL=[$5])\n"
          + "        LogicalTableScan(table=[[scott, EMP]])\n";
      final String sql = ""
          + "SELECT DEPTNO, MGR, HIREDATE, CAST(COUNT(*) AS BIGINT) AS $f3, 1 AS newCol, "
          + "CAST(COLLECT(COMM) AS ROW(COMM DECIMAL(19, 0) NULL) MULTISET) AS comArray, CAST(SUM(SAL) AS DECIMAL(19, 0)) AS salSum\n"
          + "FROM scott.EMP\n"
          + "GROUP BY DEPTNO, MGR, HIREDATE\n"
          + "ORDER BY 7";
      pig(script).assertRel(hasTree(plan))
          .assertOptimizedRel(hasTree(optimizedPlan))
          .assertSql(is(sql));
    }
  
    @Test void testCoGroup() {
      final String script = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n"
          + "B = FILTER A BY DEPTNO <= 30;\n"
          + "C = FILTER A BY DEPTNO >= 20;\n"
          + "D = GROUP A BY DEPTNO + 10, B BY (int) DEPTNO, C BY (int) DEPTNO;\n"
          + "E = ORDER D BY $0;\n";
      final String plan = ""
          + "LogicalSort(sort0=[$0], dir0=[ASC])\n"
          + "  LogicalProject(group=[$0], A=[$1], B=[$2], C=[$3])\n"
          + "    LogicalProject(DEPTNO=[CASE(IS NOT NULL($0), $0, $3)], A=[$1], B=[$2], C=[$4])\n"
          + "      LogicalJoin(condition=[=($0, $3)], joinType=[full])\n"
          + "        LogicalProject(DEPTNO=[CASE(IS NOT NULL($0), $0, $2)], A=[$1], B=[$3])\n"
          + "          LogicalJoin(condition=[=($0, $2)], joinType=[full])\n"
          + "            LogicalAggregate(group=[{0}], A=[COLLECT($1)])\n"
          + "              LogicalProject($f0=[+($0, 10)], $f1=[ROW($0, $1, $2)])\n"
          + "                LogicalTableScan(table=[[scott, DEPT]])\n"
          + "            LogicalAggregate(group=[{0}], B=[COLLECT($1)])\n"
          + "              LogicalProject(DEPTNO=[CAST($0):INTEGER], $f1=[ROW($0, $1, $2)])\n"
          + "                LogicalFilter(condition=[<=($0, 30)])\n"
          + "                  LogicalTableScan(table=[[scott, DEPT]])\n"
          + "        LogicalAggregate(group=[{0}], C=[COLLECT($1)])\n"
          + "          LogicalProject(DEPTNO=[CAST($0):INTEGER], $f1=[ROW($0, $1, $2)])\n"
          + "            LogicalFilter(condition=[>=($0, 20)])\n"
          + "              LogicalTableScan(table=[[scott, DEPT]])\n";
  
      final String result = ""
          + "(10,{},{(10,ACCOUNTING,NEW YORK)},{})\n"
          + "(20,{(10,ACCOUNTING,NEW YORK)},{(20,RESEARCH,DALLAS)},{(20,RESEARCH,DALLAS)})\n"
          + "(30,{(20,RESEARCH,DALLAS)},{(30,SALES,CHICAGO)},{(30,SALES,CHICAGO)})\n"
          + "(40,{(30,SALES,CHICAGO)},{},{(40,OPERATIONS,BOSTON)})\n"
          + "(50,{(40,OPERATIONS,BOSTON)},{},{})\n";
  
      final String sql = ""
          + "SELECT CASE WHEN t4.DEPTNO IS NOT NULL THEN t4.DEPTNO ELSE t7.DEPTNO END "
          + "AS DEPTNO, t4.A, t4.B, t7.C\n"
          + "FROM (SELECT CASE WHEN t0.$f0 IS NOT NULL THEN t0.$f0 ELSE t3.DEPTNO END "
          + "AS DEPTNO, t0.A, t3.B\n"
          + "    FROM (SELECT DEPTNO + 10 AS $f0, "
          + "COLLECT(ROW(DEPTNO, DNAME, LOC)) AS A\n"
          + "        FROM scott.DEPT\n"
          + "        GROUP BY DEPTNO + 10) AS t0\n"
          + "      FULL JOIN (SELECT CAST(DEPTNO AS INTEGER) AS DEPTNO, "
          + "COLLECT(ROW(DEPTNO, DNAME, LOC)) AS B\n"
          + "        FROM scott.DEPT\n"
          + "        WHERE DEPTNO <= 30\n"
          + "        GROUP BY CAST(DEPTNO AS INTEGER)) AS t3 "
          + "ON t0.$f0 = t3.DEPTNO) AS t4\n"
          + "  FULL JOIN (SELECT CAST(DEPTNO AS INTEGER) AS DEPTNO, COLLECT(ROW(DEPTNO, DNAME, "
          + "LOC)) AS C\n"
          + "    FROM scott.DEPT\n"
          + "    WHERE DEPTNO >= 20\n"
          + "    GROUP BY CAST(DEPTNO AS INTEGER)) AS t7 ON t4.DEPTNO = t7.DEPTNO\n"
          + "ORDER BY 1";
      pig(script).assertRel(hasTree(plan))
          .assertResult(is(result))
          .assertSql(is(sql));
    }
  
    @Test void testFlattenStrSplit() {
      final String script = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n"
          + "B = FOREACH A GENERATE FLATTEN(STRSPLIT(DNAME, ',')) as NAMES;\n";
      final String plan = ""
          + "LogicalProject(NAMES=[CAST(ITEM(STRSPLIT(PIG_TUPLE($1, ',')), 1)):BINARY(1)])\n"
          + "  LogicalTableScan(table=[[scott, DEPT]])\n";
      final String sql = ""
          + "SELECT CAST(STRSPLIT(PIG_TUPLE(DNAME, ','))[1] AS BINARY(1)) AS NAMES\n"
          + "FROM scott.DEPT";
      pig(script).assertRel(hasTree(plan))
          .assertSql(is(sql));
    }
  
    @Test void testMultipleStores() {
      final String script = ""
          + "A = LOAD 'scott.DEPT' as (DEPTNO:int, DNAME:chararray, LOC:CHARARRAY);\n"
          + "B = FILTER A BY DEPTNO <= 30;\n"
          + "STORE B into 'output.csv';\n"
          + "C = FILTER A BY DEPTNO >= 20;\n"
          + "STORE C into 'output1.csv';\n";
      final String plan = ""
          + "LogicalFilter(condition=[<=($0, 30)])\n"
          + "  LogicalTableScan(table=[[scott, DEPT]])\n";
      final String sql0 = ""
          + "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "WHERE DEPTNO <= 30";
      final String sql1 = ""
          + "SELECT *\n"
          + "FROM scott.DEPT\n"
          + "WHERE DEPTNO >= 20";
      pig(script).assertRel(hasTree(plan))
          .assertSql(is(sql0), 0)
          .assertSql(is(sql1), 1);
    }
  
    @Test void testRankAndFilter() {
      final String script = ""
          + "A = LOAD 'emp1' USING PigStorage(',')  as ("
          + "    id:int, name:chararray, age:int, city:chararray);\n"
          + "B = rank A;\n"
          + "C = FILTER B by ($0 > 1);";
  
      final String plan = ""
          + "LogicalFilter(condition=[>($0, 1)])\n"
          + "  LogicalProject(rank_A=[RANK() OVER ()], id=[$0],"
          + " name=[$1], age=[$2], city=[$3])\n"
          + "    LogicalTableScan(table=[[emp1]])\n";
  
      final String sql = "SELECT w0$o0 AS rank_A, id, name, age, city\n"
          + "FROM (SELECT id, name, age, city, RANK() OVER ()\n"
          + "    FROM emp1) AS t\n"
          + "WHERE w0$o0 > 1";
      pig(script).assertRel(hasTree(plan))
          .assertSql(is(sql));
    }
  
  }
  