/*
 * 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.drill.exec.physical.impl.lateraljoin;

import ch.qos.logback.classic.Level;
import org.apache.drill.categories.OperatorTest;
import org.apache.drill.exec.physical.impl.ScanBatch;
import org.apache.drill.exec.physical.impl.aggregate.HashAggBatch;
import org.apache.drill.exec.physical.impl.aggregate.HashAggTemplate;
import org.apache.drill.exec.physical.impl.join.LateralJoinBatch;
import org.apache.drill.exec.physical.impl.protocol.OperatorRecordBatch;
import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.test.ClusterFixture;
import org.apache.drill.test.ClusterFixtureBuilder;
import org.apache.drill.test.ClusterTest;
import org.apache.drill.test.LogFixture;
import org.apache.drill.test.TestBuilder;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import java.nio.file.Paths;

import static junit.framework.TestCase.fail;

@Category(OperatorTest.class)
public class TestE2EUnnestAndLateral extends ClusterTest {
  private static LogFixture logFixture;
  private final static Level CURRENT_LOG_LEVEL = Level.INFO;

  private static final String regularTestFile_1 = "cust_order_10_1.json";
  private static final String regularTestFile_2 = "cust_order_10_2.json";
  private static final String schemaChangeFile_1 = "cust_order_10_2_stringNationKey.json";
  private static final String schemaChangeFile_2 = "cust_order_10_2_stringOrderShipPriority.json";
  private static final String schemaChangeFile_3 = "cust_order_10_2_stringNationKey_ShipPriority.json";

  @BeforeClass
  public static void setupTestFiles() throws Exception {
    dirTestWatcher.copyResourceToRoot(Paths.get("lateraljoin", "multipleFiles", regularTestFile_1));
    dirTestWatcher.copyResourceToRoot(Paths.get("lateraljoin", "multipleFiles", regularTestFile_2));
    ClusterFixtureBuilder builder = ClusterFixture.builder(dirTestWatcher)
        .sessionOption(PlannerSettings.ENABLE_UNNEST_LATERAL_KEY, true)
        .maxParallelization(1);
    startCluster(builder);
    logFixture = LogFixture.builder()
      .toConsole()
      .logger(HashAggBatch.class, CURRENT_LOG_LEVEL)
      .logger(HashAggTemplate.class, CURRENT_LOG_LEVEL)
      .logger(ScanBatch.class, CURRENT_LOG_LEVEL)
      .logger(OperatorRecordBatch.class, CURRENT_LOG_LEVEL)
      .logger(LateralJoinBatch.class, CURRENT_LOG_LEVEL)
      .build();
  }

  /***********************************************************************************************
   Test with single batch but using different keyword for Lateral and it has limit/filter operator
   within the subquery of Lateral and Unnest
   **********************************************************************************************/

  @Test
  public void testLateral_WithLimitInSubQuery() throws Exception {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_id, orders.o_amount " +
      "FROM cp.`lateraljoin/nested-customer.parquet` customer, LATERAL " +
      "(SELECT t.ord.o_id as o_id, t.ord.o_amount as o_amount FROM UNNEST(customer.orders) t(ord) LIMIT 1) orders";
    runAndLog(sql);
  }

  @Test
  public void testLateral_WithFilterInSubQuery() throws Exception {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_id, orders.o_amount " +
      "FROM cp.`lateraljoin/nested-customer.parquet` customer, LATERAL " +
      "(SELECT t.ord.o_id as o_id, t.ord.o_amount as o_amount FROM UNNEST(customer.orders) t(ord) WHERE t.ord.o_amount > 10) orders";
    runAndLog(sql);
  }

  @Test
  public void testLateral_WithFilterAndLimitInSubQuery() throws Exception {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_id, orders.o_amount " +
      "FROM cp.`lateraljoin/nested-customer.parquet` customer, LATERAL " +
      "(SELECT t.ord.o_id as o_id, t.ord.o_amount as o_amount FROM UNNEST(customer.orders) t(ord) WHERE t.ord.o_amount > 10 LIMIT 1) orders";
    runAndLog(sql);
  }

  @Test
  public void testLateral_WithTopNInSubQuery() throws Exception {
    runAndLog("alter session set `planner.enable_topn`=false");

    String sql = "SELECT customer.c_name, orders.o_id, orders.o_amount " +
      "FROM cp.`lateraljoin/nested-customer.parquet` customer, LATERAL " +
      "(SELECT t.ord.o_id as o_id, t.ord.o_amount as o_amount FROM UNNEST(customer.orders) t(ord) ORDER BY " +
      "o_amount DESC LIMIT 1) orders";

    try {
      testBuilder()
         .sqlQuery(sql)
         .unOrdered()
         .baselineColumns("c_name", "o_id", "o_amount")
         .baselineValues("customer1", 3.0,  294.5)
         .baselineValues("customer2", 10.0,  724.5)
         .baselineValues("customer3", 23.0,  772.2)
         .baselineValues("customer4", 32.0,  1030.1)
         .go();
    } finally {
      runAndLog("alter session set `planner.enable_topn`=true");
    }
  }

  /**
   * Test which disables the TopN operator from planner settintestLateral_WithTopNInSubQuerygs before running query using SORT and LIMIT in
   * subquery. The same query as in above test is executed and same result is expected.
   */
  @Test
  public void testLateral_WithSortAndLimitInSubQuery() throws Exception {

    runAndLog("alter session set `planner.enable_topn`=false");

    String sql = "SELECT customer.c_name, orders.o_id, orders.o_amount " +
      "FROM cp.`lateraljoin/nested-customer.parquet` customer, LATERAL " +
      "(SELECT t.ord.o_id as o_id, t.ord.o_amount as o_amount FROM UNNEST(customer.orders) t(ord) ORDER BY " +
      "o_amount DESC LIMIT 1) orders";

    try {
      testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("c_name", "o_id", "o_amount")
        .baselineValues("customer1", 3.0,  294.5)
        .baselineValues("customer2", 10.0,  724.5)
        .baselineValues("customer3", 23.0,  772.2)
        .baselineValues("customer4", 32.0,  1030.1)
        .go();
    } finally {
      runAndLog("alter session set `planner.enable_topn`=true");
    }
  }

  @Test
  public void testLateral_WithSortInSubQuery() throws Exception {
    String sql = "SELECT customer.c_name, orders.o_id, orders.o_amount " +
      "FROM cp.`lateraljoin/nested-customer.parquet` customer, LATERAL " +
      "(SELECT t.ord.o_id as o_id, t.ord.o_amount as o_amount FROM UNNEST(customer.orders) t(ord) ORDER BY " +
      "o_amount DESC) orders WHERE customer.c_id = 1.0";

    testBuilder()
      .sqlQuery(sql)
      .ordered()
      .baselineColumns("c_name", "o_id", "o_amount")
      .baselineValues("customer1", 3.0,  294.5)
      .baselineValues("customer1", 2.0,  104.5)
      .baselineValues("customer1", 1.0,  4.5)
      .go();
  }

  @Test
  public void testOuterApply_WithFilterAndLimitInSubQuery() throws Exception {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_id, orders.o_amount " +
      "FROM cp.`lateraljoin/nested-customer.parquet` customer OUTER APPLY " +
      "(SELECT t.ord.o_id as o_id , t.ord.o_amount as o_amount FROM UNNEST(customer.orders) t(ord) WHERE t.ord.o_amount > 10 LIMIT 1) orders";
    runAndLog(sql);
  }

  @Test
  public void testLeftLateral_WithFilterAndLimitInSubQuery() throws Exception {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_id, orders.o_amount " +
      "FROM cp.`lateraljoin/nested-customer.parquet` customer LEFT JOIN LATERAL " +
      "(SELECT t.ord.o_id as o_id, t.ord.o_amount as o_amount FROM UNNEST(customer.orders) t(ord) WHERE t.ord.o_amount > 10 LIMIT 1) orders ON TRUE";
    runAndLog(sql);
  }

  @Test
  public void testMultiUnnestAtSameLevel() throws Exception {
    String sql = "EXPLAIN PLAN FOR SELECT customer.c_name, customer.c_address, U1.order_id, U1.order_amt," +
            " U1.itemName, U1.itemNum" + " FROM cp.`lateraljoin/nested-customer.parquet` customer, LATERAL" +
            " (SELECT t.ord.o_id AS order_id, t.ord.o_amount AS order_amt, U2.item_name AS itemName, U2.item_num AS " +
            "itemNum FROM UNNEST(customer.orders) t(ord) , LATERAL" +
            " (SELECT t1.ord.i_name AS item_name, t1.ord.i_number AS item_num FROM UNNEST(t.ord) AS t1(ord)) AS U2) AS U1";
    runAndLog(sql);
  }

  @Test
  public void testMultiUnnestAtSameLevelExec() throws Exception {
    String sql = "SELECT customer.c_name, customer.c_address, U1.order_id, U1.order_amt," +
      " U1.itemName, U1.itemNum FROM cp.`lateraljoin/nested-customer.parquet` customer, LATERAL" +
      " (SELECT dt.order_id, dt.order_amt, U2.item_name AS itemName, U2.item_num AS itemNum from" +
            "(select t.ord.items as items, t.ord.o_id AS order_id, t.ord.o_amount AS order_amt FROM UNNEST(customer.orders) t(ord)) dt , LATERAL" +
      " (SELECT t1.items.i_name AS item_name, t1.items.i_number AS item_num FROM UNNEST(dt.items) AS t1(items)) AS U2) AS U1";
    String baseline = "SELECT customer.c_name, customer.c_address, U1.order_id, U1.order_amount as order_amt, U2.item_name as itemName, U2.item_num as itemNum" +
            " FROM cp.`lateraljoin/nested-customer.parquet` customer, LATERAL " +
            "(SELECT t.ord.items as items, t.ord.o_id as order_id, t.ord.o_amount as order_amount from UNNEST(customer.orders) t(ord)) U1, LATERAL" +
            "(SELECT t1.items.i_name as item_name, t1.items.i_number as item_num from UNNEST(U1.items) t1(items)) U2";
    testBuilder()
            .unOrdered()
            .sqlQuery(sql)
            .sqlBaselineQuery(baseline)
            .go();
  }

  @Test
  public void testMultiUnnestAtSameLevelExecExplicitResult() throws Exception {
    String sql = "SELECT customer.c_name, customer.c_address, U1.order_id, U1.order_amt," +
            " U1.itemName, U1.itemNum FROM cp.`lateraljoin/nested-customer.parquet` customer, LATERAL" +
            " (SELECT dt.order_id, dt.order_amt, U2.item_name AS itemName, U2.item_num AS itemNum from" +
            "(select t.ord.items as items, t.ord.o_id AS order_id, t.ord.o_amount AS order_amt FROM UNNEST(customer.orders) t(ord)) dt , LATERAL" +
            " (SELECT t1.items.i_name AS item_name, t1.items.i_number AS item_num FROM UNNEST(dt.items) AS t1(items)) AS U2) AS U1 order by 1,2,3,4,5,6 limit 1";
    testBuilder()
            .unOrdered()
            .sqlQuery(sql)
            .baselineColumns("c_name", "c_address", "order_id", "order_amt", "itemName", "itemNum")
            .baselineValues("customer1","bay area, CA",1.0,4.5,"cheese",9.0)
            .go();
  }

  @Test
  public void testUnnestWithItem() throws Exception {
    String sql = "select u.item from\n" +
        "cp.`lateraljoin/nested-customer.parquet` c," +
        "unnest(c.orders[0]['items']) as u(item)\n" +
        "limit 1";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("item")
        .baselineValues(
            TestBuilder.mapOf("i_name", "paper towel",
                "i_number", 2.0,
                "i_supplier", "oregan"))
        .go();
  }

  @Test
  public void testUnnestWithFunctionCall() throws Exception {
    String sql = "select u.ord.o_amount o_amount from\n" +
        "cp.`lateraljoin/nested-customer.parquet` c," +
        "unnest(convert_fromjson(convert_tojson(c.orders))) as u(ord)\n" +
        "limit 1";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("o_amount")
        .baselineValues(4.5)
        .go();
  }

  @Test
  public void testUnnestWithMap() throws Exception {
    String sql = "select u.item from\n" +
        "cp.`lateraljoin/nested-customer.parquet` c," +
        "unnest(c.orders[0].items) as u(item)\n" +
        "limit 1";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("item")
        .baselineValues(
            TestBuilder.mapOf("i_name", "paper towel",
                "i_number", 2.0,
                "i_supplier", "oregan"))
        .go();
  }

  @Test
  public void testMultiUnnestWithMap() throws Exception {
    String sql = "select u.item from\n" +
        "cp.`lateraljoin/nested-customer.parquet` c," +
        "unnest(c.orders[0].items) as u(item)," +
        "unnest(c.orders[0].items) as u1(item1)\n" +
        "limit 1";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("item")
        .baselineValues(
            TestBuilder.mapOf("i_name", "paper towel",
                "i_number", 2.0,
                "i_supplier", "oregan"))
        .go();
  }

  @Test
  public void testSingleUnnestCol() throws Exception {
    String sql =
      "select t.orders.o_id as id " +
      "from (select u.orders from\n" +
            "cp.`lateraljoin/nested-customer.parquet` c," +
            "unnest(c.orders) as u(orders)\n" +
            "limit 1) t";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("id")
        .baselineValues(1.0)
        .go();
  }

  @Test
  public void testNestedUnnest() throws Exception {
    String sql = "select * from (select customer.orders as orders from cp.`lateraljoin/nested-customer.parquet` customer ) t1," +
        " lateral ( select t.ord.items as items from unnest(t1.orders) t(ord) ) t2, unnest(t2.items) t3(item) ";
    runAndLog(sql);
  }

  /***********************************************************************************************
   Test where multiple files are used to trigger multiple batch scenario. And it has limit/filter
   within the subquery of Lateral and Unnest
   **********************************************************************************************/
  @Test
  public void testMultipleBatchesLateralQuery() throws Exception {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_orderkey, orders.o_totalprice " +
      "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
      "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice FROM UNNEST(customer.c_orders) t(ord)) orders";
    runAndLog(sql);
  }

  @Test
  public void testMultipleBatchesLateral_WithLimitInSubQuery() throws Exception {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_orderkey, orders.o_totalprice " +
      "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
      "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice FROM UNNEST(customer.c_orders) t(ord) LIMIT 10) orders";
    runAndLog(sql);
  }

  @Test
  public void testMultipleBatchesLateral_WithTopNInSubQuery() throws Exception {
    runAndLog("alter session set `planner.enable_topn`=false");

    String sql = "SELECT customer.c_name, orders.o_orderkey, orders.o_totalprice " +
      "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
      "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice FROM UNNEST(customer.c_orders) t(ord)" +
      " ORDER BY o_totalprice DESC LIMIT 1) orders";

    testBuilder()
      .sqlQuery(sql)
      .unOrdered()
      .baselineColumns("c_name", "o_orderkey", "o_totalprice")
      .baselineValues("Customer#000951313", (long)47035683, 306996.2)
      .baselineValues("Customer#000007180", (long)54646821, 367189.55)
      .go();
  }

  @Test
  public void testMultipleBatchesLateral_WithSortAndLimitInSubQuery() throws Exception {

    runAndLog("alter session set `planner.enable_topn`=false");

    String sql = "SELECT customer.c_name, orders.o_orderkey, orders.o_totalprice " +
      "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
      "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice FROM UNNEST(customer.c_orders) t(ord)" +
      " ORDER BY o_totalprice DESC LIMIT 1) orders";

    try {
      testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("c_name", "o_orderkey", "o_totalprice")
        .baselineValues("Customer#000951313", (long)47035683, 306996.2)
        .baselineValues("Customer#000007180", (long)54646821, 367189.55)
        .go();
    } finally {
      runAndLog("alter session set `planner.enable_topn`=true");
    }
  }

  @Test
  public void testMultipleBatchesLateral_WithSortInSubQuery() throws Exception {

    String sql = "SELECT customer.c_name, customer.c_custkey, orders.o_orderkey, orders.o_totalprice " +
      "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
      "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice FROM UNNEST(customer.c_orders) t(ord)" +
      " ORDER BY o_totalprice DESC) orders WHERE customer.c_custkey = '7180' LIMIT 1";
    testBuilder()
      .sqlQuery(sql)
      .ordered()
      .baselineColumns("c_name", "c_custkey", "o_orderkey", "o_totalprice")
      .baselineValues("Customer#000007180", "7180", (long) 54646821, 367189.55)
      .go();

  }

  @Test
  public void testMultipleBatchesLateral_WithLimitFilterInSubQuery() throws Exception {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_orderkey, orders.o_totalprice " +
      "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
      "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice FROM UNNEST(customer.c_orders) t(ord) WHERE t.ord.o_totalprice > 100000 LIMIT 2) " +
      "orders";
    runAndLog(sql);
  }

  /***********************************************************************************************
   Test where multiple files are used to trigger schema change for Lateral and Unnest
   **********************************************************************************************/

  @Test
  public void testSchemaChangeOnNonUnnestColumn() throws Exception {

    try {
      dirTestWatcher.copyResourceToRoot(Paths.get("lateraljoin", "multipleFiles", schemaChangeFile_1));

      String sql = "SELECT customer.c_name, customer.c_address, orders.o_orderkey, orders.o_totalprice " +
        "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
        "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice FROM UNNEST (customer.c_orders) t(ord)) orders";
      runAndLog(sql);
    } catch (Exception ex) {
      fail();
    } finally {
      dirTestWatcher.removeFileFromRoot(Paths.get("lateraljoin", "multipleFiles", schemaChangeFile_1));
    }
  }

  /**
   * This test is different than {@link TestE2EUnnestAndLateral#testSchemaChangeOnNonUnnestColumn()} because with
   * multilevel when the first Lateral see's a schema change it creates a new batch with new vector references. Hence
   * the second lateral will receive a new incoming with new vector references with OK_NEW_SCHEMA outcome. Now even
   * though there is schema change for non-unnest column the second Unnest has to again setup it's transfer pairs since
   * vector reference for unnest field has changed for second Unnest.
   * Whereas in other test since there is only 1 Lateral followed by Scan, the incoming for lateral which has
   * schema change will be handled by Scan in such a way that it only updates vector of affected column. Hence in this
   * case vector corresponding to unnest field will not be affected and it will work fine.
   * @throws Exception
   */
  @Test
  public void testSchemaChangeOnNonUnnestColumn_InMultilevelCase() throws Exception {

    try {
      dirTestWatcher.copyResourceToRoot(Paths.get("lateraljoin", "multipleFiles", schemaChangeFile_1));
      String sql = "SELECT customer.c_custkey, customer.c_name, customer.c_nationkey, orders.orderkey, " +
        "orders.totalprice, olineitems.l_partkey, olineitems.l_linenumber, olineitems.l_quantity " +
        "FROM dfs.`lateraljoin/multipleFiles` customer, " +
        "LATERAL (SELECT t1.o.o_orderkey as orderkey, t1.o.o_totalprice as totalprice, t1.o.o_lineitems as lineitems " +
        "FROM UNNEST(customer.c_orders) t1(o)) orders, " +
        "LATERAL (SELECT t2.l.l_partkey as l_partkey, t2.l.l_linenumber as l_linenumber, t2.l.l_quantity as l_quantity " +
        "FROM UNNEST(orders.lineitems) t2(l)) olineitems";
      runAndLog(sql);
    } catch (Exception ex) {
      fail();
    } finally {
      dirTestWatcher.removeFileFromRoot(Paths.get("lateraljoin", "multipleFiles", schemaChangeFile_1));
    }
  }

  @Test
  public void testSchemaChangeOnUnnestColumn() throws Exception {
    try {
      dirTestWatcher.copyResourceToRoot(Paths.get("lateraljoin", "multipleFiles", schemaChangeFile_2));

      String sql = "SELECT customer.c_name, customer.c_address, orders.o_orderkey, orders.o_totalprice " +
        "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
        "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice FROM UNNEST(customer.c_orders) t(ord)) orders";
      runAndLog(sql);
    } catch (Exception ex) {
      fail();
    } finally {
      dirTestWatcher.removeFileFromRoot(Paths.get("lateraljoin", "multipleFiles", schemaChangeFile_2));
    }
  }

  @Test
  public void testSchemaChangeOnUnnestColumn_InMultilevelCase() throws Exception {
    try {
      dirTestWatcher.copyResourceToRoot(Paths.get("lateraljoin", "multipleFiles", schemaChangeFile_2));

      String sql = "SELECT customer.c_custkey, customer.c_name, customer.c_nationkey, orders.orderkey, " +
        "orders.totalprice, orders.spriority, olineitems.l_partkey, olineitems.l_linenumber, olineitems.l_quantity " +
        "FROM dfs.`lateraljoin/multipleFiles` customer, " +
        "LATERAL (SELECT t1.o.o_orderkey as orderkey, t1.o.o_totalprice as totalprice, t1.o.o_lineitems as lineitems," +
        " t1.o.o_shippriority as spriority FROM UNNEST(customer.c_orders) t1(o)) orders, " +
        "LATERAL (SELECT t2.l.l_partkey as l_partkey, t2.l.l_linenumber as l_linenumber, t2.l.l_quantity as l_quantity " +
        "FROM UNNEST(orders.lineitems) t2(l)) olineitems";
      runAndLog(sql);
    } catch (Exception ex) {
      fail();
    } finally {
      dirTestWatcher.removeFileFromRoot(Paths.get("lateraljoin", "multipleFiles", schemaChangeFile_2));
    }
  }

  @Test
  public void testSchemaChangeOnMultipleColumns() throws Exception {
    try {
      dirTestWatcher.copyResourceToRoot(Paths.get("lateraljoin", "multipleFiles", schemaChangeFile_3));

      String sql = "SELECT customer.c_name, customer.c_address, customer.c_nationkey, orders.o_orderkey, " +
        "orders.o_totalprice FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
        "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice, t.ord.o_shippriority o_shippriority FROM UNNEST(customer.c_orders) t(ord)) orders";

      runAndLog(sql);
    } catch (Exception ex) {
      fail();
    } finally {
      dirTestWatcher.removeFileFromRoot(Paths.get("lateraljoin", "multipleFiles", schemaChangeFile_3));
    }
  }

  /*****************************************************************************************
   Test where Limit/Filter/Agg/Sort operator are used in parent query
   *****************************************************************************************/

  @Test
  public void testMultipleBatchesLateral_WithLimitInParent() {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_orderkey, orders.o_totalprice " +
      "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
      "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice  as o_totalprice FROM UNNEST(customer.c_orders) t(ord) WHERE t.ord.o_totalprice > 100000 LIMIT 2) " +
      "orders LIMIT 1";
    runAndLog(sql);
  }

  @Test
  public void testMultipleBatchesLateral_WithFilterInParent() {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_orderkey, orders.o_totalprice " +
      "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
      "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice FROM UNNEST(customer.c_orders) t(ord) WHERE t.ord.o_totalprice > 100000 LIMIT 2) " +
      "orders WHERE orders.o_totalprice > 240000";
    runAndLog(sql);
  }

  @Test
  @Ignore("Disable until SchemaChange in HashAgg fixed")
  public void testMultipleBatchesLateral_WithGroupByInParent() {
    String sql = "SELECT customer.c_name, avg(orders.o_totalprice) AS avgPrice " +
      "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
      "(SELECT t.ord.o_totalprice as o_totalprice FROM UNNEST(customer.c_orders) t(ord) WHERE t.ord.o_totalprice > 100000 LIMIT 2) " +
      "orders GROUP BY customer.c_name";
    runAndLog(sql);
  }

  @Test
  public void testMultipleBatchesLateral_WithOrderByInParent() {
    String sql = "SELECT customer.c_name, customer.c_address, orders.o_orderkey, orders.o_totalprice " +
      "FROM dfs.`lateraljoin/multipleFiles` customer, LATERAL " +
      "(SELECT t.ord.o_orderkey as o_orderkey, t.ord.o_totalprice as o_totalprice FROM UNNEST(customer.c_orders) t(ord)) orders " +
      "ORDER BY orders.o_orderkey";
    runAndLog(sql);
  }

  @Test
  public void testMultipleBatchesLateral_WithHashAgg() throws Exception {
    String sql = "SELECT t2.maxprice FROM (SELECT customer.c_orders AS c_orders FROM "
      + "dfs.`lateraljoin/multipleFiles/` customer) t1, LATERAL (SELECT CAST(MAX(t.ord.o_totalprice)"
      + " AS int) AS maxprice FROM UNNEST(t1.c_orders) t(ord) GROUP BY t.ord.o_orderstatus) t2";

    try {
    testBuilder()
      .optionSettingQueriesForTestQuery("alter session set `%s` = false",
        PlannerSettings.STREAMAGG.getOptionName())
      .sqlQuery(sql)
      .unOrdered()
      .baselineColumns("maxprice")
      .baselineValues(367190)
      .baselineValues(316347)
      .baselineValues(146610)
      .baselineValues(306996)
      .baselineValues(235695)
      .baselineValues(177819)
      .build().run();
    } finally {
      runAndLog("alter session set `" + PlannerSettings.STREAMAGG.getOptionName() + "` = true");
    }
  }

  @Test
  public void testLateral_HashAgg_with_nulls() throws Exception {
    String sql = "SELECT key, t3.dsls FROM cp.`lateraljoin/with_nulls.json` t LEFT OUTER "
    + "JOIN LATERAL (SELECT DISTINCT t2.sls AS dsls FROM UNNEST(t.sales) t2(sls)) t3 ON TRUE";

    try {
    testBuilder()
      .optionSettingQueriesForTestQuery("alter session set `%s` = false",
        PlannerSettings.STREAMAGG.getOptionName())
      .sqlQuery(sql)
      .unOrdered()
      .baselineColumns("key","dsls")
      .baselineValues("aa",null)
      .baselineValues("bb",100L)
      .baselineValues("bb",200L)
      .baselineValues("bb",300L)
      .baselineValues("bb",400L)
      .baselineValues("cc",null)
      .baselineValues("dd",111L)
      .baselineValues("dd",222L)
      .build().run();
    } finally {
      runAndLog("alter session set `" + PlannerSettings.STREAMAGG.getOptionName() + "` = true");
    }
  }

  @Test
  public void testMultipleBatchesLateral_WithStreamingAgg() throws Exception {
    String sql = "SELECT t2.maxprice FROM (SELECT customer.c_orders AS c_orders FROM "
        + "dfs.`lateraljoin/multipleFiles/` customer) t1, LATERAL (SELECT CAST(MAX(t.ord.o_totalprice)"
        + " AS int) AS maxprice FROM UNNEST(t1.c_orders) t(ord) GROUP BY t.ord.o_orderstatus) t2";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("maxprice")
        .baselineValues(367190)
        .baselineValues(316347)
        .baselineValues(146610)
        .baselineValues(306996)
        .baselineValues(235695)
        .baselineValues(177819)
        .build().run();
  }

  @Test
  public void testLateral_StreamingAgg_with_nulls() throws Exception {
    String sql = "SELECT key, t3.dsls FROM cp.`lateraljoin/with_nulls.json` t LEFT OUTER "
        + "JOIN LATERAL (SELECT DISTINCT t2.sls AS dsls FROM UNNEST(t.sales) t2(sls)) t3 ON TRUE";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("key","dsls")
        .baselineValues("aa",null)
        .baselineValues("bb",100L)
        .baselineValues("bb",200L)
        .baselineValues("bb",300L)
        .baselineValues("bb",400L)
        .baselineValues("cc",null)
        .baselineValues("dd",111L)
        .baselineValues("dd",222L)
        .build().run();
  }

  @Test
  public void testMultipleBatchesLateral_WithStreamingAggNoGroup() throws Exception {
    String sql = "SELECT t2.maxprice FROM (SELECT customer.c_orders AS c_orders FROM "
        + "dfs.`lateraljoin/multipleFiles/` customer) t1, LATERAL (SELECT CAST(MAX(t.ord.o_totalprice)"
        + " AS int) AS maxprice FROM UNNEST(t1.c_orders) t(ord) ) t2";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("maxprice")
        .baselineValues(367190)
        .baselineValues(306996)
        .build().run();
  }

  @Test
  public void testUnnestNestedStarSubquery() throws Exception {
    String sql = "select t2.o.o_id o_id\n" +
        "from (select * from cp.`lateraljoin/nested-customer.json` limit 1) t,\n" +
        "unnest(t.orders) t2(o)";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("o_id")
        .baselineValues(1L)
        .baselineValues(2L)
        .baselineValues(3L)
        .go();
  }

  @Test
  public void testLateralWithComplexProject() throws Exception {
    String sql = "select l.name from cp.`lateraljoin/nested-customer.parquet` c,\n" +
        "lateral (select u.item.i_name as name from unnest(c.orders[0].items) as u(item)) l limit 1";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("name")
        .baselineValues("paper towel")
        .go();
  }

  @Test
  public void testLateralWithAgg() throws Exception {
    String sql = "select l.name from cp.`lateraljoin/nested-customer.parquet` c,\n" +
        "lateral (select max(u.item.i_name) as name from unnest(c.orders[0].items) as u(item)) l limit 1";

    testBuilder()
        .sqlQuery(sql)
        .unOrdered()
        .baselineColumns("name")
        .baselineValues("paper towel")
        .go();
  }

  @Test
  public void testMultiLateralWithComplexProject() throws Exception {
    String sql = "select l1.name, l2.name as name2 from cp.`lateraljoin/nested-customer.parquet` c,\n" +
      "lateral (select u.item.i_name as name from unnest(c.orders[0].items) as u(item)) l1," +
      "lateral (select u.item.i_name as name from unnest(c.orders[0].items) as u(item)) l2 limit 1";

    testBuilder()
      .sqlQuery(sql)
      .unOrdered()
      .baselineColumns("name", "name2")
      .baselineValues("paper towel", "paper towel")
      .go();
  }
}
