text:
package org.apache.spark.sql.jdbc

import java.sql.{Connection, DriverManager}
import java.util.Properties

import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.sql.{AnalysisException, DataFrame, ExplainSuiteHelper, QueryTest, Row}
import org.apache.spark.sql.catalyst.analysis.CannotReplaceMissingTableException
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, GlobalLimit, LocalLimit, Offset, Sort}
import org.apache.spark.sql.connector.IntegralAverage
import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2ScanRelation, V1ScanWrapper}
import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog
import org.apache.spark.sql.functions.{abs, acos, asin, atan, atan2, avg, ceil, coalesce, cos, cosh, cot, count, count_distinct, degrees, exp, floor, lit, log => logarithm, log10, not, pow, radians, round, signum, sin, sinh, sqrt, sum, tan, tanh, udf, when}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.util.Utils

class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHelper {
  import testImplicits._

  val tempDir = Utils.createTempDir()
  val url = s"jdbc:h2:${tempDir.getCanonicalPath};user=testUser;password=testPass"
  var conn: java.sql.Connection = null

  override def sparkConf: SparkConf = super.sparkConf
    .set("spark.sql.catalog.h2", classOf[JDBCTableCatalog].getName)
    .set("spark.sql.catalog.h2.url", url)
    .set("spark.sql.catalog.h2.driver", "org.h2.Driver")
    .set("spark.sql.catalog.h2.pushDownAggregate", "true")
    .set("spark.sql.catalog.h2.pushDownLimit", "true")
    .set("spark.sql.catalog.h2.pushDownOffset", "true")

  private def withConnection[T](f: Connection => T): T = {
    val conn = DriverManager.getConnection(url, new Properties())
    try {
      f(conn)
    } finally {
      conn.close()
    }
  }

  override def beforeAll(): Unit = {
    super.beforeAll()
    Utils.classForName("org.h2.Driver")
    withConnection { conn =>
      conn.prepareStatement("CREATE SCHEMA \"test\"").executeUpdate()
      conn.prepareStatement(
        "CREATE TABLE \"test\".\"empty_table\" (name TEXT(32) NOT NULL, id INTEGER NOT NULL)")
        .executeUpdate()
      conn.prepareStatement(
        "CREATE TABLE \"test\".\"people\" (name TEXT(32) NOT NULL, id INTEGER NOT NULL)")
        .executeUpdate()
      conn.prepareStatement("INSERT INTO \"test\".\"people\" VALUES ('fred', 1)").executeUpdate()
      conn.prepareStatement("INSERT INTO \"test\".\"people\" VALUES ('mary', 2)").executeUpdate()
      conn.prepareStatement(
        "CREATE TABLE \"test\".\"employee\" (dept INTEGER, name TEXT(32), salary NUMERIC(20, 2)," +
          " bonus DOUBLE, is_manager BOOLEAN)").executeUpdate()
      conn.prepareStatement(
        "INSERT INTO \"test\".\"employee\" VALUES (1, 'amy', 10000, 1000, true)").executeUpdate()
      conn.prepareStatement(
        "INSERT INTO \"test\".\"employee\" VALUES (2, 'alex', 12000, 1200, false)").executeUpdate()
      conn.prepareStatement(
        "INSERT INTO \"test\".\"employee\" VALUES (1, 'cathy', 9000, 1200, false)").executeUpdate()
      conn.prepareStatement(
        "INSERT INTO \"test\".\"employee\" VALUES (2, 'david', 10000, 1300, true)").executeUpdate()
      conn.prepareStatement(
        "INSERT INTO \"test\".\"employee\" VALUES (6, 'jen', 12000, 1200, true)").executeUpdate()
      conn.prepareStatement(
        "CREATE TABLE \"test\".\"dept\" (\"dept id\" INTEGER NOT NULL, \"dept.id\" INTEGER)")
        .executeUpdate()
      conn.prepareStatement("INSERT INTO \"test\".\"dept\" VALUES (1, 1)").executeUpdate()
      conn.prepareStatement("INSERT INTO \"test\".\"dept\" VALUES (2, 1)").executeUpdate()

      // scalastyle:off
      conn.prepareStatement(
        "CREATE TABLE \"test\".\"person\" (\"名\" INTEGER NOT NULL)").executeUpdate()
      // scalastyle:on
      conn.prepareStatement("INSERT INTO \"test\".\"person\" VALUES (1)").executeUpdate()
      conn.prepareStatement("INSERT INTO \"test\".\"person\" VALUES (2)").executeUpdate()
      conn.prepareStatement(
        """CREATE TABLE "test"."view1" ("|col1" INTEGER, "|col2" INTEGER)""").executeUpdate()
      conn.prepareStatement(
        """CREATE TABLE "test"."view2" ("|col1" INTEGER, "|col3" INTEGER)""").executeUpdate()

      conn.prepareStatement(
        "CREATE TABLE \"test\".\"item\" (id INTEGER, name TEXT(32), price NUMERIC(23, 3))")
        .executeUpdate()
      conn.prepareStatement("INSERT INTO \"test\".\"item\" VALUES " +
        "(1, 'bottle', 11111111111111111111.123)").executeUpdate()
      conn.prepareStatement("INSERT INTO \"test\".\"item\" VALUES " +
        "(1, 'bottle', 99999999999999999999.123)").executeUpdate()
    }
    H2Dialect.registerFunction("my_avg", IntegralAverage)
  }

  override def afterAll(): Unit = {
    H2Dialect.clearFunctions()
    Utils.deleteRecursively(tempDir)
    super.afterAll()
  }

  test("simple scan") {
    checkAnswer(sql("SELECT * FROM h2.test.empty_table"), Seq())
    checkAnswer(sql("SELECT * FROM h2.test.people"), Seq(Row("fred", 1), Row("mary", 2)))
    checkAnswer(sql("SELECT name, id FROM h2.test.people"), Seq(Row("fred", 1), Row("mary", 2)))
  }

  private def checkPushedInfo(df: DataFrame, expectedPlanFragment: String*): Unit = {
    df.queryExecution.optimizedPlan.collect {
      case _: DataSourceV2ScanRelation =>
        checkKeywordsExistsInExplain(df, expectedPlanFragment: _*)
    }
  }

  // TABLESAMPLE ({integer_expression | decimal_expression} PERCENT) and
  // TABLESAMPLE (BUCKET integer_expression OUT OF integer_expression)
  // are tested in JDBC dialect tests because TABLESAMPLE is not supported by all the DBMS
  test("TABLESAMPLE (integer_expression ROWS) is the same as LIMIT") {
    val df = sql("SELECT NAME FROM h2.test.employee TABLESAMPLE (3 ROWS)")
    checkSchemaNames(df, Seq("NAME"))
    checkPushedInfo(df, "PushedFilters: [], PushedLimit: LIMIT 3, ")
    checkAnswer(df, Seq(Row("amy"), Row("alex"), Row("cathy")))
  }

  private def checkSchemaNames(df: DataFrame, names: Seq[String]): Unit = {
    val scan = df.queryExecution.optimizedPlan.collectFirst {
      case s: DataSourceV2ScanRelation => s
    }.get
    assert(scan.schema.names.sameElements(names))
  }

  private def checkLimitRemoved(df: DataFrame, removed: Boolean = true): Unit = {
    val limits = df.queryExecution.optimizedPlan.collect {
      case g: GlobalLimit => g
      case limit: LocalLimit => limit
    }
    if (removed) {
      assert(limits.isEmpty)
    } else {
      assert(limits.nonEmpty)
    }
  }

  test("simple scan with LIMIT") {
    val df1 = spark.read.table("h2.test.employee")
      .where($"dept" === 1).limit(1)
    checkLimitRemoved(df1)
    checkPushedInfo(df1,
      "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 1, ")
    checkAnswer(df1, Seq(Row(1, "amy", 10000.00, 1000.0, true)))

    val df2 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .filter($"dept" > 1)
      .limit(1)
    checkLimitRemoved(df2, false)
    checkPushedInfo(df2,
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 1, ")
    checkAnswer(df2, Seq(Row(2, "alex", 12000.00, 1200.0, false)))

    val df3 = sql("SELECT name FROM h2.test.employee WHERE dept > 1 LIMIT 1")
    checkSchemaNames(df3, Seq("NAME"))
    checkLimitRemoved(df3)
    checkPushedInfo(df3,
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 1, ")
    checkAnswer(df3, Seq(Row("alex")))

    val df4 = spark.read
      .table("h2.test.employee")
      .groupBy("DEPT").sum("SALARY")
      .limit(1)
    checkLimitRemoved(df4, false)
    checkPushedInfo(df4,
      "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
    checkAnswer(df4, Seq(Row(1, 19000.00)))

    val name = udf { (x: String) => x.matches("cat|dav|amy") }
    val sub = udf { (x: String) => x.substring(0, 3) }
    val df5 = spark.read
      .table("h2.test.employee")
      .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
      .filter(name($"shortName"))
      .limit(1)
    checkLimitRemoved(df5, false)
    // LIMIT is pushed down only if all the filters are pushed down
    checkPushedInfo(df5, "PushedFilters: [], ")
    checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
  }

  private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
    val offsets = df.queryExecution.optimizedPlan.collect {
      case offset: Offset => offset
    }
    if (removed) {
      assert(offsets.isEmpty)
    } else {
      assert(offsets.nonEmpty)
    }
  }

  test("simple scan with OFFSET") {
    val df1 = spark.read
      .table("h2.test.employee")
      .where($"dept" === 1)
      .offset(1)
    checkOffsetRemoved(df1)
    checkPushedInfo(df1,
      "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
    checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df2 = spark.read
      .option("pushDownOffset", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .offset(1)
    checkOffsetRemoved(df2, false)
    checkPushedInfo(df2,
      "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
    checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df3 = spark.read
      .table("h2.test.employee")
      .where($"dept" === 1)
      .sort($"salary")
      .offset(1)
    checkOffsetRemoved(df3, false)
    checkPushedInfo(df3,
      "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
    checkAnswer(df3, Seq(Row(1, "amy", 10000.00, 1000.0, true)))

    val df4 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .filter($"dept" > 1)
      .offset(1)
    checkOffsetRemoved(df4, false)
    checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
    checkAnswer(df4, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))

    val df5 = spark.read
      .table("h2.test.employee")
      .groupBy("DEPT").sum("SALARY")
      .offset(1)
    checkOffsetRemoved(df5, false)
    checkPushedInfo(df5,
      "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
    checkAnswer(df5, Seq(Row(2, 22000.00), Row(6, 12000.00)))

    val name = udf { (x: String) => x.matches("cat|dav|amy") }
    val sub = udf { (x: String) => x.substring(0, 3) }
    val df6 = spark.read
      .table("h2.test.employee")
      .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
      .filter(name($"shortName"))
      .offset(1)
    checkOffsetRemoved(df6, false)
    // OFFSET is pushed down only if all the filters are pushed down
    checkPushedInfo(df6, "PushedFilters: [], ")
    checkAnswer(df6, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
  }

  test("simple scan with LIMIT and OFFSET") {
    val df1 = spark.read
      .table("h2.test.employee")
      .where($"dept" === 1)
      .limit(2)
      .offset(1)
    checkLimitRemoved(df1)
    checkOffsetRemoved(df1)
    checkPushedInfo(df1,
      "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
    checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df2 = spark.read
      .option("pushDownLimit", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .limit(2)
      .offset(1)
    checkLimitRemoved(df2, false)
    checkOffsetRemoved(df2, false)
    checkPushedInfo(df2,
      "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
    checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df3 = spark.read
      .option("pushDownOffset", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .limit(2)
      .offset(1)
    checkLimitRemoved(df3)
    checkOffsetRemoved(df3, false)
    checkPushedInfo(df3,
      "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
    checkAnswer(df3, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df4 = spark.read
      .option("pushDownLimit", "false")
      .option("pushDownOffset", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .limit(2)
      .offset(1)
    checkLimitRemoved(df4, false)
    checkOffsetRemoved(df4, false)
    checkPushedInfo(df4,
      "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
    checkAnswer(df4, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df5 = spark.read
      .table("h2.test.employee")
      .where($"dept" === 1)
      .sort($"salary")
      .limit(2)
      .offset(1)
    checkLimitRemoved(df5)
    checkOffsetRemoved(df5)
    checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
      "PushedOffset: OFFSET 1, PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
    checkAnswer(df5, Seq(Row(1, "amy", 10000.00, 1000.0, true)))

    val df6 = spark.read
      .option("pushDownLimit", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .sort($"salary")
      .limit(2)
      .offset(1)
    checkLimitRemoved(df6, false)
    checkOffsetRemoved(df6, false)
    checkPushedInfo(df6, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
    checkAnswer(df6, Seq(Row(1, "amy", 10000.00, 1000.0, true)))

    val df7 = spark.read
      .option("pushDownOffset", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .sort($"salary")
      .limit(2)
      .offset(1)
    checkLimitRemoved(df7)
    checkOffsetRemoved(df7, false)
    checkPushedInfo(df7, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1]," +
      " PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
    checkAnswer(df7, Seq(Row(1, "amy", 10000.00, 1000.0, true)))

    val df8 = spark.read
      .option("pushDownLimit", "false")
      .option("pushDownOffset", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .sort($"salary")
      .limit(2)
      .offset(1)
    checkLimitRemoved(df8, false)
    checkOffsetRemoved(df8, false)
    checkPushedInfo(df8, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
    checkAnswer(df8, Seq(Row(1, "amy", 10000.00, 1000.0, true)))

    val df9 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .filter($"dept" > 1)
      .limit(2)
      .offset(1)
    checkLimitRemoved(df9, false)
    checkOffsetRemoved(df9, false)
    checkPushedInfo(df9,
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 2, ReadSchema:")
    checkAnswer(df9, Seq(Row(2, "david", 10000.00, 1300.0, true)))

    val df10 = spark.read
      .table("h2.test.employee")
      .groupBy("DEPT").sum("SALARY")
      .limit(2)
      .offset(1)
    checkLimitRemoved(df10, false)
    checkOffsetRemoved(df10, false)
    checkPushedInfo(df10,
      "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
    checkAnswer(df10, Seq(Row(2, 22000.00)))

    val name = udf { (x: String) => x.matches("cat|dav|amy") }
    val sub = udf { (x: String) => x.substring(0, 3) }
    val df11 = spark.read
      .table("h2.test.employee")
      .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
      .filter(name($"shortName"))
      .limit(2)
      .offset(1)
    checkLimitRemoved(df11, false)
    checkOffsetRemoved(df11, false)
    checkPushedInfo(df11, "PushedFilters: [], ")
    checkAnswer(df11, Seq(Row(9000.00, 1200.0, "cat")))
  }

  test("simple scan with OFFSET and LIMIT") {
    val df1 = spark.read
      .table("h2.test.employee")
      .where($"dept" === 1)
      .offset(1)
      .limit(1)
    checkLimitRemoved(df1)
    checkOffsetRemoved(df1)
    checkPushedInfo(df1,
      "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
    checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df2 = spark.read
      .option("pushDownOffset", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .offset(1)
      .limit(1)
    checkLimitRemoved(df2)
    checkOffsetRemoved(df2, false)
    checkPushedInfo(df2,
      "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
    checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df3 = spark.read
      .option("pushDownLimit", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .offset(1)
      .limit(1)
    checkLimitRemoved(df3, false)
    checkOffsetRemoved(df3)
    checkPushedInfo(df3,
      "[DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1, ReadSchema:")
    checkAnswer(df3, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df4 = spark.read
      .option("pushDownOffset", "false")
      .option("pushDownLimit", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .offset(1)
      .limit(1)
    checkLimitRemoved(df4, false)
    checkOffsetRemoved(df4, false)
    checkPushedInfo(df4,
      "[DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
    checkAnswer(df4, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df5 = spark.read
      .table("h2.test.employee")
      .where($"dept" === 1)
      .sort($"salary")
      .offset(1)
      .limit(1)
    checkLimitRemoved(df5)
    checkOffsetRemoved(df5)
    checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
      "PushedOffset: OFFSET 1, PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
    checkAnswer(df5, Seq(Row(1, "amy", 10000.00, 1000.0, true)))

    val df6 = spark.read
      .option("pushDownOffset", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .sort($"salary")
      .offset(1)
      .limit(1)
    checkLimitRemoved(df6)
    checkOffsetRemoved(df6, false)
    checkPushedInfo(df6, "[DEPT IS NOT NULL, DEPT = 1]," +
      " PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
    checkAnswer(df6, Seq(Row(1, "amy", 10000.00, 1000.0, true)))

    val df7 = spark.read
      .option("pushDownLimit", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .sort($"salary")
      .offset(1)
      .limit(1)
    checkLimitRemoved(df7, false)
    checkOffsetRemoved(df7, false)
    checkPushedInfo(df7, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
    checkAnswer(df7, Seq(Row(1, "amy", 10000.00, 1000.0, true)))

    val df8 = spark.read
      .option("pushDownOffset", "false")
      .option("pushDownLimit", "false")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .sort($"salary")
      .offset(1)
      .limit(1)
    checkLimitRemoved(df8, false)
    checkOffsetRemoved(df8, false)
    checkPushedInfo(df8, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
    checkAnswer(df8, Seq(Row(1, "amy", 10000.00, 1000.0, true)))

    val df9 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .filter($"dept" > 1)
      .offset(1)
      .limit(1)
    checkLimitRemoved(df9, false)
    checkOffsetRemoved(df9, false)
    checkPushedInfo(df9,
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 2, ReadSchema:")
    checkAnswer(df9, Seq(Row(2, "david", 10000.00, 1300.0, true)))

    val df10 = sql("SELECT dept, sum(salary) FROM h2.test.employee group by dept LIMIT 1 OFFSET 1")
    checkLimitRemoved(df10, false)
    checkOffsetRemoved(df10, false)
    checkPushedInfo(df10,
      "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
    checkAnswer(df10, Seq(Row(2, 22000.00)))

    val name = udf { (x: String) => x.matches("cat|dav|amy") }
    val sub = udf { (x: String) => x.substring(0, 3) }
    val df11 = spark.read
      .table("h2.test.employee")
      .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
      .filter(name($"shortName"))
      .offset(1)
      .limit(1)
    checkLimitRemoved(df11, false)
    checkOffsetRemoved(df11, false)
    checkPushedInfo(df11, "PushedFilters: [], ")
    checkAnswer(df11, Seq(Row(9000.00, 1200.0, "cat")))
  }

  private def checkSortRemoved(df: DataFrame, removed: Boolean = true): Unit = {
    val sorts = df.queryExecution.optimizedPlan.collect {
      case s: Sort => s
    }
    if (removed) {
      assert(sorts.isEmpty)
    } else {
      assert(sorts.nonEmpty)
    }
  }

  test("simple scan with top N") {
    val df1 = spark.read
      .table("h2.test.employee")
      .sort("salary")
      .limit(1)
    checkSortRemoved(df1)
    checkLimitRemoved(df1)
    checkPushedInfo(df1,
      "PushedFilters: [], PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 1, ")
    checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df2 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "1")
      .table("h2.test.employee")
      .where($"dept" === 1)
      .orderBy($"salary")
      .limit(1)
    checkSortRemoved(df2)
    checkLimitRemoved(df2)
    checkPushedInfo(df2, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
      "PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 1, ")
    checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))

    val df3 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .filter($"dept" > 1)
      .orderBy($"salary".desc)
      .limit(1)
    checkSortRemoved(df3, false)
    checkLimitRemoved(df3, false)
    checkPushedInfo(df3, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], " +
      "PushedTopN: ORDER BY [SALARY DESC NULLS LAST] LIMIT 1, ")
    checkAnswer(df3, Seq(Row(2, "alex", 12000.00, 1200.0, false)))

    val df4 =
      sql("SELECT name FROM h2.test.employee WHERE dept > 1 ORDER BY salary NULLS LAST LIMIT 1")
    checkSchemaNames(df4, Seq("NAME"))
    checkSortRemoved(df4)
    checkLimitRemoved(df4)
    checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], " +
      "PushedTopN: ORDER BY [SALARY ASC NULLS LAST] LIMIT 1, ")
    checkAnswer(df4, Seq(Row("david")))

    val df5 = spark.read.table("h2.test.employee")
      .where($"dept" === 1).orderBy($"salary")
    checkSortRemoved(df5, false)
    checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ")
    checkAnswer(df5,
      Seq(Row(1, "cathy", 9000.00, 1200.0, false), Row(1, "amy", 10000.00, 1000.0, true)))

    val df6 = spark.read
      .table("h2.test.employee")
      .groupBy("DEPT").sum("SALARY")
      .orderBy("DEPT")
      .limit(1)
    checkSortRemoved(df6, false)
    checkLimitRemoved(df6, false)
    checkPushedInfo(df6, "PushedAggregates: [SUM(SALARY)]," +
      " PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
    checkAnswer(df6, Seq(Row(1, 19000.00)))

    val name = udf { (x: String) => x.matches("cat|dav|amy") }
    val sub = udf { (x: String) => x.substring(0, 3) }
    val df7 = spark.read
      .table("h2.test.employee")
      .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
      .filter(name($"shortName"))
      .sort($"SALARY".desc)
      .limit(1)
    // LIMIT is pushed down only if all the filters are pushed down
    checkSortRemoved(df7, false)
    checkLimitRemoved(df7, false)
    checkPushedInfo(df7, "PushedFilters: [], ")
    checkAnswer(df7, Seq(Row(10000.00, 1000.0, "amy")))

    val df8 = spark.read
      .table("h2.test.employee")
      .sort(sub($"NAME"))
      .limit(1)
    checkSortRemoved(df8, false)
    checkLimitRemoved(df8, false)
    checkPushedInfo(df8, "PushedFilters: [], ")
    checkAnswer(df8, Seq(Row(2, "alex", 12000.00, 1200.0, false)))

    val df9 = spark.read
      .table("h2.test.employee")
      .select($"DEPT", $"name", $"SALARY",
        when(($"SALARY" > 8000).and($"SALARY" < 10000), $"salary").otherwise(0).as("key"))
      .sort("key", "dept", "SALARY")
      .limit(3)
    checkSortRemoved(df9)
    checkLimitRemoved(df9)
    checkPushedInfo(df9, "PushedFilters: [], " +
      "PushedTopN: ORDER BY [CASE WHEN (SALARY > 8000.00) AND " +
      "(SALARY < 10000.00) THEN SALARY ELSE 0.00 END ASC NULL..., ")
    checkAnswer(df9,
      Seq(Row(1, "amy", 10000, 0), Row(2, "david", 10000, 0), Row(2, "alex", 12000, 0)))

    val df10 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .select($"DEPT", $"name", $"SALARY",
        when(($"SALARY" > 8000).and($"SALARY" < 10000), $"salary").otherwise(0).as("key"))
      .orderBy($"key", $"dept", $"SALARY")
      .limit(3)
    checkSortRemoved(df10, false)
    checkLimitRemoved(df10, false)
    checkPushedInfo(df10, "PushedFilters: [], " +
      "PushedTopN: ORDER BY [CASE WHEN (SALARY > 8000.00) AND " +
      "(SALARY < 10000.00) THEN SALARY ELSE 0.00 END ASC NULL..., ")
    checkAnswer(df10,
      Seq(Row(1, "amy", 10000, 0), Row(2, "david", 10000, 0), Row(2, "alex", 12000, 0)))
  }

  test("simple scan with top N: order by with alias") {
    val df1 = spark.read
      .table("h2.test.employee")
      .select($"NAME", $"SALARY".as("mySalary"))
      .sort("mySalary")
      .limit(1)
    checkSortRemoved(df1)
    checkPushedInfo(df1,
      "PushedFilters: [], PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 1, ")
    checkAnswer(df1, Seq(Row("cathy", 9000.00)))

    val df2 = spark.read
      .table("h2.test.employee")
      .select($"DEPT", $"NAME", $"SALARY".as("mySalary"))
      .filter($"DEPT" > 1)
      .sort("mySalary")
      .limit(1)
    checkSortRemoved(df2)
    checkPushedInfo(df2,
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], " +
        "PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 1, ")
    checkAnswer(df2, Seq(Row(2, "david", 10000.00)))
  }

  test("scan with filter push-down") {
    val df = spark.table("h2.test.people").filter($"id" > 1)
    checkFiltersRemoved(df)
    checkPushedInfo(df, "PushedFilters: [ID IS NOT NULL, ID > 1], ")
    checkAnswer(df, Row("mary", 2))

    val df2 = spark.table("h2.test.employee").filter($"name".isin("amy", "cathy"))
    checkFiltersRemoved(df2)
    checkPushedInfo(df2, "PushedFilters: [NAME IN ('amy', 'cathy')]")
    checkAnswer(df2, Seq(Row(1, "amy", 10000, 1000, true), Row(1, "cathy", 9000, 1200, false)))

    val df3 = spark.table("h2.test.employee").filter($"name".startsWith("a"))
    checkFiltersRemoved(df3)
    checkPushedInfo(df3, "PushedFilters: [NAME IS NOT NULL, NAME LIKE 'a%']")
    checkAnswer(df3, Seq(Row(1, "amy", 10000, 1000, true), Row(2, "alex", 12000, 1200, false)))

    val df4 = spark.table("h2.test.employee").filter($"is_manager")
    checkFiltersRemoved(df4)
    checkPushedInfo(df4, "PushedFilters: [IS_MANAGER IS NOT NULL, IS_MANAGER = true]")
    checkAnswer(df4, Seq(Row(1, "amy", 10000, 1000, true), Row(2, "david", 10000, 1300, true),
      Row(6, "jen", 12000, 1200, true)))

    val df5 = spark.table("h2.test.employee").filter($"is_manager".and($"salary" > 10000))
    checkFiltersRemoved(df5)
    checkPushedInfo(df5, "PushedFilters: [IS_MANAGER IS NOT NULL, SALARY IS NOT NULL, " +
      "IS_MANAGER = true, SALARY > 10000.00]")
    checkAnswer(df5, Seq(Row(6, "jen", 12000, 1200, true)))

    val df6 = spark.table("h2.test.employee").filter($"is_manager".or($"salary" > 10000))
    checkFiltersRemoved(df6)
    checkPushedInfo(df6, "PushedFilters: [(IS_MANAGER = true) OR (SALARY > 10000.00)], ")
    checkAnswer(df6, Seq(Row(1, "amy", 10000, 1000, true), Row(2, "alex", 12000, 1200, false),
      Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))

    val df7 = spark.table("h2.test.employee").filter(not($"is_manager") === true)
    checkFiltersRemoved(df7)
    checkPushedInfo(df7, "PushedFilters: [IS_MANAGER IS NOT NULL, NOT (IS_MANAGER = true)], ")
    checkAnswer(df7, Seq(Row(1, "cathy", 9000, 1200, false), Row(2, "alex", 12000, 1200, false)))

    val df8 = spark.table("h2.test.employee").filter($"is_manager" === true)
    checkFiltersRemoved(df8)
    checkPushedInfo(df8, "PushedFilters: [IS_MANAGER IS NOT NULL, IS_MANAGER = true], ")
    checkAnswer(df8, Seq(Row(1, "amy", 10000, 1000, true),
      Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))

    val df9 = spark.table("h2.test.employee")
      .filter(when($"dept" > 1, true).when($"is_manager", false).otherwise($"dept" > 3))
    checkFiltersRemoved(df9)
    checkPushedInfo(df9, "PushedFilters: [CASE WHEN DEPT > 1 THEN TRUE " +
      "WHEN IS_MANAGER = true THEN FALSE ELSE DEPT > 3 END], ")
    checkAnswer(df9, Seq(Row(2, "alex", 12000, 1200, false),
      Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))

    val df10 = spark.table("h2.test.people")
      .select($"NAME".as("myName"), $"ID".as("myID"))
      .filter($"myID" > 1)
    checkFiltersRemoved(df10)
    checkPushedInfo(df10, "PushedFilters: [ID IS NOT NULL, ID > 1], ")
    checkAnswer(df10, Row("mary", 2))

    val df11 = sql(
      """
        |SELECT * FROM h2.test.employee
        |WHERE GREATEST(bonus, 1100) > 1200 AND LEAST(salary, 10000) > 9000 AND RAND(1) < 1
        |""".stripMargin)
    checkFiltersRemoved(df11)
    checkPushedInfo(df11, "PushedFilters: " +
      "[(GREATEST(BONUS, 1100.0)) > 1200.0, (LEAST(SALARY, 10000.00)) > 9000.00, RAND(1) < 1.0]")
    checkAnswer(df11, Row(2, "david", 10000, 1300, true))

    val df12 = sql(
      """
        |SELECT * FROM h2.test.employee
        |WHERE IF(SALARY > 10000, SALARY, LEAST(SALARY, 1000)) > 1200
        |""".stripMargin)
    checkFiltersRemoved(df12)
    checkPushedInfo(df12, "PushedFilters: " +
      "[(CASE WHEN SALARY > 10000.00 THEN SALARY ELSE LEAST(SALARY, 1000.00) END) > 1200.00]")
    checkAnswer(df12, Seq(Row(2, "alex", 12000, 1200, false), Row(6, "jen", 12000, 1200, true)))

    val df13 = spark.table("h2.test.employee")
      .filter(logarithm($"bonus") > 7)
      .filter(exp($"bonus") > 0)
      .filter(pow($"bonus", 2) === 1440000)
      .filter(sqrt($"bonus") > 34)
      .filter(floor($"bonus") === 1200)
      .filter(ceil($"bonus") === 1200)
    checkFiltersRemoved(df13)
    checkPushedInfo(df13, "PushedFilters: [BONUS IS NOT NULL, LN(BONUS) > 7.0, EXP(BONUS) > 0.0, " +
      "(POWER(BONUS, 2.0)) = 1440000.0, SQRT(BONU...,")
    checkAnswer(df13, Seq(Row(1, "cathy", 9000, 1200, false),
      Row(2, "alex", 12000, 1200, false), Row(6, "jen", 12000, 1200, true)))

    // H2 does not support width_bucket
    val df14 = sql(
      """
        |SELECT * FROM h2.test.employee
        |WHERE width_bucket(bonus, 1, 6, 3) > 4
        |""".stripMargin)
    checkFiltersRemoved(df14, false)
    checkPushedInfo(df14, "PushedFilters: [BONUS IS NOT NULL]")
    checkAnswer(df14, Seq.empty[Row])

    val df15 = spark.table("h2.test.employee")
      .filter(logarithm(2, $"bonus") > 10)
      .filter(log10($"bonus") > 3)
      .filter(round($"bonus") === 1200)
      .filter(degrees($"bonus") > 68754)
      .filter(radians($"bonus") > 20)
      .filter(signum($"bonus") === 1)
    checkFiltersRemoved(df15)
    checkPushedInfo(df15, "PushedFilters: [BONUS IS NOT NULL, (LOG(2.0, BONUS)) > 10.0, " +
      "LOG10(BONUS) > 3.0, (ROUND(BONUS, 0)) = 1200.0, DEG...,")
    checkAnswer(df15, Seq(Row(1, "cathy", 9000, 1200, false),
      Row(2, "alex", 12000, 1200, false), Row(6, "jen", 12000, 1200, true)))

    val df16 = spark.table("h2.test.employee")
      .filter(sin($"bonus") < -0.08)
      .filter(sinh($"bonus") > 200)
      .filter(cos($"bonus") > 0.9)
      .filter(cosh($"bonus") > 200)
      .filter(tan($"bonus") < -0.08)
      .filter(tanh($"bonus") === 1)
      .filter(cot($"bonus") < -11)
      .filter(asin($"bonus") > 0.1)
      .filter(acos($"bonus") > 1.4)
      .filter(atan($"bonus") > 1.4)
      .filter(atan2($"bonus", $"bonus") > 0.7)
    checkFiltersRemoved(df16)
    checkPushedInfo(df16, "PushedFilters: [BONUS IS NOT NULL, SIN(BONUS) < -0.08, " +
      "SINH(BONUS) > 200.0, COS(BONUS) > 0.9, COSH(BONUS) > 200....,")
    checkAnswer(df16, Seq(Row(1, "cathy", 9000, 1200, false),
      Row(2, "alex", 12000, 1200, false), Row(6, "jen", 12000, 1200, true)))

    // H2 does not support log2, asinh, acosh, atanh, cbrt
    val df17 = sql(
      """
        |SELECT * FROM h2.test.employee
        |WHERE log2(dept) > 2.5
        |AND asinh(bonus / salary) > 0.09
        |AND acosh(dept) > 2.4
        |AND atanh(bonus / salary) > 0.1
        |AND cbrt(dept) > 1.8
        |""".stripMargin)
    checkFiltersRemoved(df17, false)
    checkPushedInfo(df17,
      "PushedFilters: [DEPT IS NOT NULL, BONUS IS NOT NULL, SALARY IS NOT NULL]")
    checkAnswer(df17, Seq(Row(6, "jen", 12000, 1200, true)))
  }

  test("scan with filter push-down with ansi mode") {
    Seq(false, true).foreach { ansiMode =>
      withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiMode.toString) {
        val df = spark.table("h2.test.people").filter($"id" + 1 > 1)
        checkFiltersRemoved(df, ansiMode)
        val expectedPlanFragment = if (ansiMode) {
          "PushedFilters: [ID IS NOT NULL, (ID + 1) > 1]"
        } else {
          "PushedFilters: [ID IS NOT NULL]"
        }
        checkPushedInfo(df, expectedPlanFragment)
        checkAnswer(df, Seq(Row("fred", 1), Row("mary", 2)))

        val df2 = spark.table("h2.test.people").filter($"id" + Int.MaxValue > 1)
        checkFiltersRemoved(df2, ansiMode)
        val expectedPlanFragment2 = if (ansiMode) {
          "PushedFilters: [ID IS NOT NULL, (ID + 2147483647) > 1], "
        } else {
          "PushedFilters: [ID IS NOT NULL], "
        }
        checkPushedInfo(df2, expectedPlanFragment2)
        if (ansiMode) {
          val e = intercept[SparkException] {
            checkAnswer(df2, Seq.empty)
          }
          assert(e.getMessage.contains(
            "org.h2.jdbc.JdbcSQLDataException: Numeric value out of range: \"2147483648\""))
        } else {
          checkAnswer(df2, Seq.empty)
        }

        val df3 = sql(
          """
            |SELECT * FROM h2.test.employee
            |WHERE (CASE WHEN SALARY > 10000 THEN BONUS ELSE BONUS + 200 END) > 1200
            |""".stripMargin)

        checkFiltersRemoved(df3, ansiMode)
        val expectedPlanFragment3 = if (ansiMode) {
          "PushedFilters: [(CASE WHEN SALARY > 10000.00 THEN BONUS" +
            " ELSE BONUS + 200.0 END) > 1200.0]"
        } else {
          "PushedFilters: []"
        }
        checkPushedInfo(df3, expectedPlanFragment3)
        checkAnswer(df3,
          Seq(Row(1, "cathy", 9000, 1200, false), Row(2, "david", 10000, 1300, true)))

        val df4 = spark.table("h2.test.employee")
          .filter(($"salary" > 1000d).and($"salary" < 12000d))
        checkFiltersRemoved(df4, ansiMode)
        val expectedPlanFragment4 = if (ansiMode) {
          "PushedFilters: [SALARY IS NOT NULL, " +
            "CAST(SALARY AS double) > 1000.0, CAST(SALARY AS double) < 12000.0], "
        } else {
          "PushedFilters: [SALARY IS NOT NULL], "
        }
        checkPushedInfo(df4, expectedPlanFragment4)
        checkAnswer(df4, Seq(Row(1, "amy", 10000, 1000, true),
          Row(1, "cathy", 9000, 1200, false), Row(2, "david", 10000, 1300, true)))

        val df5 = spark.table("h2.test.employee")
          .filter(abs($"dept" - 3) > 1)
          .filter(coalesce($"salary", $"bonus") > 2000)
        checkFiltersRemoved(df5, ansiMode)
        val expectedPlanFragment5 = if (ansiMode) {
          "PushedFilters: [DEPT IS NOT NULL, ABS(DEPT - 3) > 1, " +
            "(COALESCE(CAST(SALARY AS double), BONUS)) > 2000.0]"
        } else {
          "PushedFilters: [DEPT IS NOT NULL]"
        }
        checkPushedInfo(df5, expectedPlanFragment5)
        checkAnswer(df5, Seq(Row(1, "amy", 10000, 1000, true),
          Row(1, "cathy", 9000, 1200, false), Row(6, "jen", 12000, 1200, true)))

        val df6 = sql(
          """
            |SELECT * FROM h2.test.employee
            |WHERE cast(bonus as string) like '%30%'
            |AND cast(dept as byte) > 1
            |AND cast(dept as short) > 1
            |AND cast(bonus as decimal(20, 2)) > 1200""".stripMargin)
        checkFiltersRemoved(df6, ansiMode)
        val expectedPlanFragment8 = if (ansiMode) {
          "PushedFilters: [BONUS IS NOT NULL, DEPT IS NOT NULL, " +
            "CAST(BONUS AS string) LIKE '%30%', CAST(DEPT AS byte) > 1, ...,"
        } else {
          "PushedFilters: [BONUS IS NOT NULL, DEPT IS NOT NULL],"
        }
        checkPushedInfo(df6, expectedPlanFragment8)
        checkAnswer(df6, Seq(Row(2, "david", 10000, 1300, true)))
      }
    }
  }

  test("scan with column pruning") {
    val df = spark.table("h2.test.people").select("id")
    checkSchemaNames(df, Seq("ID"))
    checkAnswer(df, Seq(Row(1), Row(2)))
  }

  test("scan with filter push-down and column pruning") {
    val df = spark.table("h2.test.people").filter($"id" > 1).select("name")
    checkFiltersRemoved(df)
    checkSchemaNames(df, Seq("NAME"))
    checkAnswer(df, Row("mary"))
  }

  test("read/write with partition info") {
    withTable("h2.test.abc") {
      sql("CREATE TABLE h2.test.abc AS SELECT * FROM h2.test.people")
      val df1 = Seq(("evan", 3), ("cathy", 4), ("alex", 5)).toDF("NAME", "ID")
      val e = intercept[IllegalArgumentException] {
        df1.write
          .option("partitionColumn", "id")
          .option("lowerBound", "0")
          .option("upperBound", "3")
          .option("numPartitions", "0")
          .insertInto("h2.test.abc")
      }.getMessage
      assert(e.contains("Invalid value `0` for parameter `numPartitions` in table writing " +
        "via JDBC. The minimum value is 1."))

      df1.write
        .option("partitionColumn", "id")
        .option("lowerBound", "0")
        .option("upperBound", "3")
        .option("numPartitions", "3")
        .insertInto("h2.test.abc")

      val df2 = spark.read
        .option("partitionColumn", "id")
        .option("lowerBound", "0")
        .option("upperBound", "3")
        .option("numPartitions", "2")
        .table("h2.test.abc")

      assert(df2.rdd.getNumPartitions === 2)
      assert(df2.count() === 5)
    }
  }

  test("show tables") {
    checkAnswer(sql("SHOW TABLES IN h2.test"),
      Seq(Row("test", "people", false), Row("test", "empty_table", false),
        Row("test", "employee", false), Row("test", "item", false), Row("test", "dept", false),
        Row("test", "person", false), Row("test", "view1", false), Row("test", "view2", false)))
  }

  test("SQL API: create table as select") {
    withTable("h2.test.abc") {
      sql("CREATE TABLE h2.test.abc AS SELECT * FROM h2.test.people")
      checkAnswer(sql("SELECT name, id FROM h2.test.abc"), Seq(Row("fred", 1), Row("mary", 2)))
    }
  }

  test("DataFrameWriterV2: create table as select") {
    withTable("h2.test.abc") {
      spark.table("h2.test.people").writeTo("h2.test.abc").create()
      checkAnswer(sql("SELECT name, id FROM h2.test.abc"), Seq(Row("fred", 1), Row("mary", 2)))
    }
  }

  test("SQL API: replace table as select") {
    withTable("h2.test.abc") {
      intercept[CannotReplaceMissingTableException] {
        sql("REPLACE TABLE h2.test.abc AS SELECT 1 as col")
      }
      sql("CREATE OR REPLACE TABLE h2.test.abc AS SELECT 1 as col")
      checkAnswer(sql("SELECT col FROM h2.test.abc"), Row(1))
      sql("REPLACE TABLE h2.test.abc AS SELECT * FROM h2.test.people")
      checkAnswer(sql("SELECT name, id FROM h2.test.abc"), Seq(Row("fred", 1), Row("mary", 2)))
    }
  }

  test("DataFrameWriterV2: replace table as select") {
    withTable("h2.test.abc") {
      intercept[CannotReplaceMissingTableException] {
        sql("SELECT 1 AS col").writeTo("h2.test.abc").replace()
      }
      sql("SELECT 1 AS col").writeTo("h2.test.abc").createOrReplace()
      checkAnswer(sql("SELECT col FROM h2.test.abc"), Row(1))
      spark.table("h2.test.people").writeTo("h2.test.abc").replace()
      checkAnswer(sql("SELECT name, id FROM h2.test.abc"), Seq(Row("fred", 1), Row("mary", 2)))
    }
  }

  test("SQL API: insert and overwrite") {
    withTable("h2.test.abc") {
      sql("CREATE TABLE h2.test.abc AS SELECT * FROM h2.test.people")

      sql("INSERT INTO h2.test.abc SELECT 'lucy', 3")
      checkAnswer(
        sql("SELECT name, id FROM h2.test.abc"),
        Seq(Row("fred", 1), Row("mary", 2), Row("lucy", 3)))

      sql("INSERT OVERWRITE h2.test.abc SELECT 'bob', 4")
      checkAnswer(sql("SELECT name, id FROM h2.test.abc"), Row("bob", 4))
    }
  }

  test("DataFrameWriterV2: insert and overwrite") {
    withTable("h2.test.abc") {
      sql("CREATE TABLE h2.test.abc AS SELECT * FROM h2.test.people")

      // `DataFrameWriterV2` is by-name.
      sql("SELECT 3 AS ID, 'lucy' AS NAME").writeTo("h2.test.abc").append()
      checkAnswer(
        sql("SELECT name, id FROM h2.test.abc"),
        Seq(Row("fred", 1), Row("mary", 2), Row("lucy", 3)))

      sql("SELECT 'bob' AS NAME, 4 AS ID").writeTo("h2.test.abc").overwrite(lit(true))
      checkAnswer(sql("SELECT name, id FROM h2.test.abc"), Row("bob", 4))
    }
  }

  private def checkAggregateRemoved(df: DataFrame, removed: Boolean = true): Unit = {
    val aggregates = df.queryExecution.optimizedPlan.collect {
      case agg: Aggregate => agg
    }
    if (removed) {
      assert(aggregates.isEmpty)
    } else {
      assert(aggregates.nonEmpty)
    }
  }

  test("scan with filter push-down with string functions") {
    val df1 = sql("SELECT * FROM h2.test.employee WHERE " +
      "substr(name, 2, 1) = 'e'" +
      " AND upper(name) = 'JEN' AND lower(name) = 'jen' ")
    checkFiltersRemoved(df1)
    val expectedPlanFragment1 =
      "PushedFilters: [NAME IS NOT NULL, (SUBSTRING(NAME, 2, 1)) = 'e', " +
      "UPPER(NAME) = 'JEN', LOWER(NAME) = 'jen']"
    checkPushedInfo(df1, expectedPlanFragment1)
    checkAnswer(df1, Seq(Row(6, "jen", 12000, 1200, true)))

    val df2 = sql("SELECT * FROM h2.test.employee WHERE " +
      "trim(name) = 'jen' AND trim('j', name) = 'en'" +
      "AND translate(name, 'e', 1) = 'j1n'")
    checkFiltersRemoved(df2)
    val expectedPlanFragment2 =
      "PushedFilters: [NAME IS NOT NULL, TRIM(BOTH FROM NAME) = 'jen', " +
      "(TRIM(BOTH 'j' FROM NAME)) = 'en', (TRANSLATE(NA..."
    checkPushedInfo(df2, expectedPlanFragment2)
    checkAnswer(df2, Seq(Row(6, "jen", 12000, 1200, true)))

    val df3 = sql("SELECT * FROM h2.test.employee WHERE " +
      "ltrim(name) = 'jen' AND ltrim('j', name) = 'en'")
    checkFiltersRemoved(df3)
    val expectedPlanFragment3 =
      "PushedFilters: [TRIM(LEADING FROM NAME) = 'jen', " +
      "(TRIM(LEADING 'j' FROM NAME)) = 'en']"
    checkPushedInfo(df3, expectedPlanFragment3)
    checkAnswer(df3, Seq(Row(6, "jen", 12000, 1200, true)))

    val df4 = sql("SELECT * FROM h2.test.employee WHERE " +
      "rtrim(name) = 'jen' AND rtrim('n', name) = 'je'")
    checkFiltersRemoved(df4)
    val expectedPlanFragment4 =
      "PushedFilters: [TRIM(TRAILING FROM NAME) = 'jen', " +
      "(TRIM(TRAILING 'n' FROM NAME)) = 'je']"
    checkPushedInfo(df4, expectedPlanFragment4)
    checkAnswer(df4, Seq(Row(6, "jen", 12000, 1200, true)))

    // H2 does not support OVERLAY
    val df5 = sql("SELECT * FROM h2.test.employee WHERE OVERLAY(NAME, '1', 2, 1) = 'j1n'")
    checkFiltersRemoved(df5, false)
    val expectedPlanFragment5 =
      "PushedFilters: [NAME IS NOT NULL]"
    checkPushedInfo(df5, expectedPlanFragment5)
    checkAnswer(df5, Seq(Row(6, "jen", 12000, 1200, true)))
  }

  test("scan with aggregate push-down: MAX AVG with filter and group by") {
    val df = sql("SELECT MAX(SaLaRY), AVG(BONUS) FROM h2.test.employee WHERE dept > 0" +
      " GROUP BY DePt")
    checkFiltersRemoved(df)
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [MAX(SALARY), AVG(BONUS)], " +
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
      "PushedGroupByExpressions: [DEPT], ")
    checkAnswer(df, Seq(Row(10000, 1100.0), Row(12000, 1250.0), Row(12000, 1200.0)))
  }

  private def checkFiltersRemoved(df: DataFrame, removed: Boolean = true): Unit = {
    val filters = df.queryExecution.optimizedPlan.collect {
      case f: Filter => f
    }
    if (removed) {
      assert(filters.isEmpty)
    } else {
      assert(filters.nonEmpty)
    }
  }

  test("scan with aggregate push-down: MAX AVG with filter without group by") {
    val df = sql("SELECT MAX(ID), AVG(ID) FROM h2.test.people WHERE id > 0")
    checkFiltersRemoved(df)
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [MAX(ID), AVG(ID)], " +
      "PushedFilters: [ID IS NOT NULL, ID > 0], " +
      "PushedGroupByExpressions: [], ")
    checkAnswer(df, Seq(Row(2, 1.5)))
  }

  test("partitioned scan with aggregate push-down: complete push-down only") {
    withTempView("v") {
      spark.read
        .option("partitionColumn", "dept")
        .option("lowerBound", "0")
        .option("upperBound", "2")
        .option("numPartitions", "2")
        .table("h2.test.employee")
        .createTempView("v")
      val df = sql("select AVG(SALARY) FROM v GROUP BY name")
      // Partitioned JDBC Scan doesn't support complete aggregate push-down, and AVG requires
      // complete push-down so aggregate is not pushed at the end.
      checkAggregateRemoved(df, removed = false)
      checkAnswer(df, Seq(Row(9000.0), Row(10000.0), Row(10000.0), Row(12000.0), Row(12000.0)))
    }
  }

  test("scan with aggregate push-down: aggregate + number") {
    val df = sql("SELECT MAX(SALARY) + 1 FROM h2.test.employee")
    checkAggregateRemoved(df)
    df.queryExecution.optimizedPlan.collect {
      case _: DataSourceV2ScanRelation =>
        val expected_plan_fragment =
          "PushedAggregates: [MAX(SALARY)]"
        checkKeywordsExistsInExplain(df, expected_plan_fragment)
    }
    checkPushedInfo(df, "PushedAggregates: [MAX(SALARY)]")
    checkAnswer(df, Seq(Row(12001)))
  }

  test("scan with aggregate push-down: COUNT(*)") {
    val df = sql("SELECT COUNT(*) FROM h2.test.employee")
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [COUNT(*)]")
    checkAnswer(df, Seq(Row(5)))
  }

  test("scan with aggregate push-down: GROUP BY without aggregate functions") {
    val df = sql("SELECT name FROM h2.test.employee GROUP BY name")
    checkAggregateRemoved(df)
    checkPushedInfo(df,
      "PushedAggregates: [], PushedFilters: [], PushedGroupByExpressions: [NAME],")
    checkAnswer(df, Seq(Row("alex"), Row("amy"), Row("cathy"), Row("david"), Row("jen")))

    val df2 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .groupBy($"name")
      .agg(Map.empty[String, String])
    checkAggregateRemoved(df2, false)
    checkPushedInfo(df2,
      "PushedAggregates: [], PushedFilters: [], PushedGroupByExpressions: [NAME],")
    checkAnswer(df2, Seq(Row("alex"), Row("amy"), Row("cathy"), Row("david"), Row("jen")))

    val df3 = sql("SELECT CASE WHEN SALARY > 8000 AND SALARY < 10000 THEN SALARY ELSE 0 END as" +
      " key FROM h2.test.employee GROUP BY key")
    checkAggregateRemoved(df3)
    checkPushedInfo(df3,
      """
        |PushedAggregates: [],
        |PushedFilters: [],
        |PushedGroupByExpressions:
        |[CASE WHEN (SALARY > 8000.00) AND (SALARY < 10000.00) THEN SALARY ELSE 0.00 END],
        |""".stripMargin.replaceAll("\n", " "))
    checkAnswer(df3, Seq(Row(0), Row(9000)))

    val df4 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .groupBy(when(($"SALARY" > 8000).and($"SALARY" < 10000), $"SALARY").otherwise(0).as("key"))
      .agg(Map.empty[String, String])
    checkAggregateRemoved(df4, false)
    checkPushedInfo(df4,
      """
        |PushedAggregates: [],
        |PushedFilters: [],
        |PushedGroupByExpressions:
        |[CASE WHEN (SALARY > 8000.00) AND (SALARY < 10000.00) THEN SALARY ELSE 0.00 END],
        |""".stripMargin.replaceAll("\n", " "))
    checkAnswer(df4, Seq(Row(0), Row(9000)))
  }

  test("scan with aggregate push-down: COUNT(col)") {
    val df = sql("SELECT COUNT(DEPT) FROM h2.test.employee")
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [COUNT(DEPT)]")
    checkAnswer(df, Seq(Row(5)))
  }

  test("scan with aggregate push-down: COUNT(DISTINCT col)") {
    val df = sql("SELECT COUNT(DISTINCT DEPT) FROM h2.test.employee")
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [COUNT(DISTINCT DEPT)]")
    checkAnswer(df, Seq(Row(3)))
  }

  test("scan with aggregate push-down: cannot partial push down COUNT(DISTINCT col)") {
    val df = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .agg(count_distinct($"DEPT"))
    checkAggregateRemoved(df, false)
    checkAnswer(df, Seq(Row(3)))
  }

  test("scan with aggregate push-down: SUM without filer and group by") {
    val df = sql("SELECT SUM(SALARY) FROM h2.test.employee")
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [SUM(SALARY)]")
    checkAnswer(df, Seq(Row(53000)))
  }

  test("scan with aggregate push-down: DISTINCT SUM without filer and group by") {
    val df = sql("SELECT SUM(DISTINCT SALARY) FROM h2.test.employee")
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [SUM(DISTINCT SALARY)]")
    checkAnswer(df, Seq(Row(31000)))
  }

  test("scan with aggregate push-down: SUM with group by") {
    val df1 = sql("SELECT SUM(SALARY) FROM h2.test.employee GROUP BY DEPT")
    checkAggregateRemoved(df1)
    checkPushedInfo(df1, "PushedAggregates: [SUM(SALARY)], " +
      "PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
    checkAnswer(df1, Seq(Row(19000), Row(22000), Row(12000)))

    val df2 = sql(
      """
        |SELECT CASE WHEN SALARY > 8000 AND SALARY < 10000 THEN SALARY ELSE 0 END as key,
        |  SUM(SALARY) FROM h2.test.employee GROUP BY key""".stripMargin)
    checkAggregateRemoved(df2)
    checkPushedInfo(df2,
      """
        |PushedAggregates: [SUM(SALARY)],
        |PushedFilters: [],
        |PushedGroupByExpressions:
        |[CASE WHEN (SALARY > 8000.00) AND (SALARY < 10000.00) THEN SALARY ELSE 0.00 END],
        |""".stripMargin.replaceAll("\n", " "))
    checkAnswer(df2, Seq(Row(0, 44000), Row(9000, 9000)))

    val df3 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .groupBy(when(($"SALARY" > 8000).and($"SALARY" < 10000), $"SALARY").otherwise(0).as("key"))
      .agg(sum($"SALARY"))
    checkAggregateRemoved(df3, false)
    checkPushedInfo(df3,
      """
        |PushedAggregates: [SUM(SALARY)],
        |PushedFilters: [],
        |PushedGroupByExpressions:
        |[CASE WHEN (SALARY > 8000.00) AND (SALARY < 10000.00) THEN SALARY ELSE 0.00 END],
        |""".stripMargin.replaceAll("\n", " "))
    checkAnswer(df3, Seq(Row(0, 44000), Row(9000, 9000)))

    val df4 = sql(
      """
        |SELECT DEPT, CASE WHEN SALARY > 8000 AND SALARY < 10000 THEN SALARY ELSE 0 END as key,
        |  SUM(SALARY) FROM h2.test.employee GROUP BY DEPT, key""".stripMargin)
    checkAggregateRemoved(df4)
    checkPushedInfo(df4,
      """
        |PushedAggregates: [SUM(SALARY)],
        |PushedFilters: [],
        |PushedGroupByExpressions:
        |[DEPT, CASE WHEN (SALARY > 8000.00) AND (SALARY < 10000.00) THEN SALARY ELSE 0.00 END],
        |""".stripMargin.replaceAll("\n", " "))
    checkAnswer(df4, Seq(Row(1, 0, 10000), Row(1, 9000, 9000), Row(2, 0, 22000), Row(6, 0, 12000)))

    val df5 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .groupBy($"DEPT",
        when(($"SALARY" > 8000).and($"SALARY" < 10000), $"SALARY").otherwise(0)
          .as("key"))
      .agg(sum($"SALARY"))
    checkAggregateRemoved(df5, false)
    checkPushedInfo(df5,
      """
        |PushedAggregates: [SUM(SALARY)],
        |PushedFilters: [],
        |PushedGroupByExpressions:
        |[DEPT, CASE WHEN (SALARY > 8000.00) AND (SALARY < 10000.00) THEN SALARY ELSE 0.00 END],
        |""".stripMargin.replaceAll("\n", " "))
    checkAnswer(df5, Seq(Row(1, 0, 10000), Row(1, 9000, 9000), Row(2, 0, 22000), Row(6, 0, 12000)))

    val df6 = sql(
      """
        |SELECT CASE WHEN SALARY > 8000 AND is_manager <> false THEN SALARY ELSE 0 END as key,
        |  SUM(SALARY) FROM h2.test.employee GROUP BY key""".stripMargin)
    checkAggregateRemoved(df6)
    checkPushedInfo(df6,
      """
        |PushedAggregates: [SUM(SALARY)],
        |PushedFilters: [],
        |PushedGroupByExpressions:
        |[CASE WHEN (SALARY > 8000.00) AND (IS_MANAGER = true) THEN SALARY ELSE 0.00 END],
        |""".stripMargin.replaceAll("\n", " "))
    checkAnswer(df6, Seq(Row(0, 21000), Row(10000, 20000), Row(12000, 12000)))

    val df7 = sql(
      """
        |SELECT CASE WHEN SALARY > 8000 OR is_manager <> false THEN SALARY ELSE 0 END as key,
        |  SUM(SALARY) FROM h2.test.employee GROUP BY key""".stripMargin)
    checkAggregateRemoved(df7)
    checkPushedInfo(df7,
      """
        |PushedAggregates: [SUM(SALARY)],
        |PushedFilters: [],
        |PushedGroupByExpressions:
        |[CASE WHEN (SALARY > 8000.00) OR (IS_MANAGER = true) THEN SALARY ELSE 0.00 END],
        |""".stripMargin.replaceAll("\n", " "))
    checkAnswer(df7, Seq(Row(10000, 20000), Row(12000, 24000), Row(9000, 9000)))

    val df8 = sql(
      """
        |SELECT CASE WHEN NOT(is_manager <> false) THEN SALARY ELSE 0 END as key,
        |  SUM(SALARY) FROM h2.test.employee GROUP BY key""".stripMargin)
    checkAggregateRemoved(df8)
    checkPushedInfo(df8,
      """
        |PushedAggregates: [SUM(SALARY)],
        |PushedFilters: [],
        |PushedGroupByExpressions:
        |[CASE WHEN NOT (IS_MANAGER = true) THEN SALARY ELSE 0.00 END],
        |""".stripMargin.replaceAll("\n", " "))
    checkAnswer(df8, Seq(Row(0, 32000), Row(12000, 12000), Row(9000, 9000)))
  }

  test("scan with aggregate push-down: DISTINCT SUM with group by") {
    val df = sql("SELECT SUM(DISTINCT SALARY) FROM h2.test.employee GROUP BY DEPT")
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [SUM(DISTINCT SALARY)], " +
      "PushedFilters: [], PushedGroupByExpressions: [DEPT]")
    checkAnswer(df, Seq(Row(19000), Row(22000), Row(12000)))
  }

  test("scan with aggregate push-down: with multiple group by columns") {
    val df = sql("SELECT MAX(SALARY), MIN(BONUS) FROM h2.test.employee WHERE dept > 0" +
      " GROUP BY DEPT, NAME")
    checkFiltersRemoved(df)
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [MAX(SALARY), MIN(BONUS)], " +
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], PushedGroupByExpressions: [DEPT, NAME]")
    checkAnswer(df, Seq(Row(9000, 1200), Row(12000, 1200), Row(10000, 1300),
      Row(10000, 1000), Row(12000, 1200)))
  }

  test("scan with aggregate push-down: with concat multiple group key in project") {
    val df1 = sql("SELECT concat_ws('#', DEPT, NAME), MAX(SALARY) FROM h2.test.employee" +
      " WHERE dept > 0 GROUP BY DEPT, NAME")
    val filters1 = df1.queryExecution.optimizedPlan.collect {
      case f: Filter => f
    }
    assert(filters1.isEmpty)
    checkAggregateRemoved(df1)
    checkPushedInfo(df1, "PushedAggregates: [MAX(SALARY)], " +
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], PushedGroupByExpressions: [DEPT, NAME]")
    checkAnswer(df1, Seq(Row("1#amy", 10000), Row("1#cathy", 9000), Row("2#alex", 12000),
      Row("2#david", 10000), Row("6#jen", 12000)))

    val df2 = sql("SELECT concat_ws('#', DEPT, NAME), MAX(SALARY) + MIN(BONUS)" +
      " FROM h2.test.employee WHERE dept > 0 GROUP BY DEPT, NAME")
    val filters2 = df2.queryExecution.optimizedPlan.collect {
      case f: Filter => f
    }
    assert(filters2.isEmpty)
    checkAggregateRemoved(df2)
    checkPushedInfo(df2, "PushedAggregates: [MAX(SALARY), MIN(BONUS)], " +
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], PushedGroupByExpressions: [DEPT, NAME]")
    checkAnswer(df2, Seq(Row("1#amy", 11000), Row("1#cathy", 10200), Row("2#alex", 13200),
      Row("2#david", 11300), Row("6#jen", 13200)))

    val df3 = sql("SELECT concat_ws('#', DEPT, NAME), MAX(SALARY) + MIN(BONUS)" +
      " FROM h2.test.employee WHERE dept > 0 GROUP BY concat_ws('#', DEPT, NAME)")
    checkFiltersRemoved(df3)
    checkAggregateRemoved(df3, false)
    checkPushedInfo(df3, "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], ")
    checkAnswer(df3, Seq(Row("1#amy", 11000), Row("1#cathy", 10200), Row("2#alex", 13200),
      Row("2#david", 11300), Row("6#jen", 13200)))
  }

  test("scan with aggregate push-down: with having clause") {
    val df = sql("SELECT MAX(SALARY), MIN(BONUS) FROM h2.test.employee WHERE dept > 0" +
      " GROUP BY DEPT having MIN(BONUS) > 1000")
    // filter over aggregate not push down
    checkFiltersRemoved(df, false)
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [MAX(SALARY), MIN(BONUS)], " +
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], PushedGroupByExpressions: [DEPT]")
    checkAnswer(df, Seq(Row(12000, 1200), Row(12000, 1200)))
  }

  test("scan with aggregate push-down: alias over aggregate") {
    val df = sql("SELECT * FROM h2.test.employee")
      .groupBy($"DEPT")
      .min("SALARY").as("total")
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [MIN(SALARY)], " +
      "PushedFilters: [], PushedGroupByExpressions: [DEPT]")
    checkAnswer(df, Seq(Row(1, 9000), Row(2, 10000), Row(6, 12000)))
  }

  test("scan with aggregate push-down: order by alias over aggregate") {
    val df = spark.table("h2.test.employee")
    val query = df.select($"DEPT", $"SALARY")
      .filter($"DEPT" > 0)
      .groupBy($"DEPT")
      .agg(sum($"SALARY").as("total"))
      .filter($"total" > 1000)
      .orderBy($"total")
    checkFiltersRemoved(query, false)// filter over aggregate not pushed down
    checkAggregateRemoved(query)
    checkPushedInfo(query, "PushedAggregates: [SUM(SALARY)], " +
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], PushedGroupByExpressions: [DEPT]")
    checkAnswer(query, Seq(Row(6, 12000), Row(1, 19000), Row(2, 22000)))
  }

  test("scan with aggregate push-down: udf over aggregate") {
    val df = spark.table("h2.test.employee")
    val decrease = udf { (x: Double, y: Double) => x - y }
    val query = df.select(decrease(sum($"SALARY"), sum($"BONUS")).as("value"))
    checkAggregateRemoved(query)
    checkPushedInfo(query, "PushedAggregates: [SUM(SALARY), SUM(BONUS)], ")
    checkAnswer(query, Seq(Row(47100.0)))
  }

  test("scan with aggregate push-down: partition columns are same as group by columns") {
    val df = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .groupBy($"dept")
      .count()
    checkAggregateRemoved(df)
    checkAnswer(df, Seq(Row(1, 2), Row(2, 2), Row(6, 1)))
  }

  test("scan with aggregate push-down: VAR_POP VAR_SAMP with filter and group by") {
    val df = sql("SELECT VAR_POP(bonus), VAR_SAMP(bonus) FROM h2.test.employee WHERE dept > 0" +
      " GROUP BY DePt")
    checkFiltersRemoved(df)
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [VAR_POP(BONUS), VAR_SAMP(BONUS)], " +
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], PushedGroupByExpressions: [DEPT]")
    checkAnswer(df, Seq(Row(10000d, 20000d), Row(2500d, 5000d), Row(0d, null)))
  }

  test("scan with aggregate push-down: STDDEV_POP STDDEV_SAMP with filter and group by") {
    val df = sql("SELECT STDDEV_POP(bonus), STDDEV_SAMP(bonus) FROM h2.test.employee" +
      " WHERE dept > 0 GROUP BY DePt")
    checkFiltersRemoved(df)
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [STDDEV_POP(BONUS), STDDEV_SAMP(BONUS)], " +
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], PushedGroupByExpressions: [DEPT]")
    checkAnswer(df, Seq(Row(100d, 141.4213562373095d), Row(50d, 70.71067811865476d), Row(0d, null)))
  }

  test("scan with aggregate push-down: COVAR_POP COVAR_SAMP with filter and group by") {
    val df = sql("SELECT COVAR_POP(bonus, bonus), COVAR_SAMP(bonus, bonus)" +
      " FROM h2.test.employee WHERE dept > 0 GROUP BY DePt")
    checkFiltersRemoved(df)
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [COVAR_POP(BONUS, BONUS), COVAR_SAMP(BONUS, BONUS)], " +
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], PushedGroupByExpressions: [DEPT]")
    checkAnswer(df, Seq(Row(10000d, 20000d), Row(2500d, 5000d), Row(0d, null)))
  }

  test("scan with aggregate push-down: CORR with filter and group by") {
    val df = sql("SELECT CORR(bonus, bonus) FROM h2.test.employee WHERE dept > 0" +
      " GROUP BY DePt")
    checkFiltersRemoved(df)
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [CORR(BONUS, BONUS)], " +
      "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], PushedGroupByExpressions: [DEPT]")
    checkAnswer(df, Seq(Row(1d), Row(1d), Row(null)))
  }

  test("scan with aggregate push-down: aggregate over alias push down") {
    val cols = Seq("a", "b", "c", "d", "e")
    val df1 = sql("SELECT * FROM h2.test.employee").toDF(cols: _*)
    val df2 = df1.groupBy().sum("c")
    checkAggregateRemoved(df2)
    df2.queryExecution.optimizedPlan.collect {
      case relation: DataSourceV2ScanRelation =>
        val expectedPlanFragment =
          "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: []"
        checkKeywordsExistsInExplain(df2, expectedPlanFragment)
        relation.scan match {
          case v1: V1ScanWrapper =>
            assert(v1.pushedDownOperators.aggregation.nonEmpty)
      }
    }
    checkAnswer(df2, Seq(Row(53000.00)))
  }

  test("scan with aggregate push-down: aggregate with partially pushed down filters" +
    "will NOT push down") {
    val df = spark.table("h2.test.employee")
    val name = udf { (x: String) => x.matches("cat|dav|amy") }
    val sub = udf { (x: String) => x.substring(0, 3) }
    val query = df.select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
      .filter("SALARY > 100")
      .filter(name($"shortName"))
      .agg(sum($"SALARY").as("sum_salary"))
    checkAggregateRemoved(query, false)
    query.queryExecution.optimizedPlan.collect {
      case relation: DataSourceV2ScanRelation => relation.scan match {
        case v1: V1ScanWrapper =>
          assert(v1.pushedDownOperators.aggregation.isEmpty)
      }
    }
    checkAnswer(query, Seq(Row(29000.0)))
  }

  test("scan with aggregate push-down: aggregate function with CASE WHEN") {
    val df = sql(
      """
        |SELECT
        |  COUNT(CASE WHEN SALARY > 8000 AND SALARY < 10000 THEN SALARY ELSE 0 END),
        |  COUNT(CASE WHEN SALARY > 8000 AND SALARY <= 13000 THEN SALARY ELSE 0 END),
        |  COUNT(CASE WHEN SALARY > 11000 OR SALARY < 10000 THEN SALARY ELSE 0 END),
        |  COUNT(CASE WHEN SALARY >= 12000 OR SALARY < 9000 THEN SALARY ELSE 0 END),
        |  COUNT(CASE WHEN SALARY >= 12000 OR NOT(SALARY >= 9000) THEN SALARY ELSE 0 END),
        |  MAX(CASE WHEN NOT(SALARY > 10000) AND SALARY >= 8000 THEN SALARY ELSE 0 END),
        |  MAX(CASE WHEN NOT(SALARY > 9000) OR SALARY > 10000 THEN SALARY ELSE 0 END),
        |  MAX(CASE WHEN NOT(SALARY > 10000) AND NOT(SALARY < 8000) THEN SALARY ELSE 0 END),
        |  MAX(CASE WHEN NOT(SALARY != 0) OR NOT(SALARY < 8000) THEN SALARY ELSE 0 END),
        |  MAX(CASE WHEN NOT(SALARY > 8000 AND SALARY < 10000) THEN 0 ELSE SALARY END),
        |  MIN(CASE WHEN NOT(SALARY > 8000 OR SALARY IS NULL) THEN SALARY ELSE 0 END),
        |  SUM(CASE WHEN SALARY > 10000 THEN 2 WHEN SALARY > 8000 THEN 1 END),
        |  AVG(CASE WHEN NOT(SALARY > 8000 OR SALARY IS NOT NULL) THEN SALARY ELSE 0 END)
        |FROM h2.test.employee GROUP BY DEPT
      """.stripMargin)
    checkAggregateRemoved(df)
    checkPushedInfo(df,
      "PushedAggregates: [COUNT(CASE WHEN (SALARY > 8000.00) AND (SALARY < 10000.00)" +
      " THEN SALARY ELSE 0.00 END), COUNT(CAS..., " +
      "PushedFilters: [], " +
      "PushedGroupByExpressions: [DEPT], ")
    checkAnswer(df, Seq(Row(1, 1, 1, 1, 1, 0d, 12000d, 0d, 12000d, 0d, 0d, 2, 0d),
      Row(2, 2, 2, 2, 2, 10000d, 12000d, 10000d, 12000d, 0d, 0d, 3, 0d),
      Row(2, 2, 2, 2, 2, 10000d, 9000d, 10000d, 10000d, 9000d, 0d, 2, 0d)))
  }

  test("scan with aggregate push-down: aggregate function with binary arithmetic") {
    Seq(false, true).foreach { ansiMode =>
      withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiMode.toString) {
        val df = sql("SELECT SUM(2147483647 + DEPT) FROM h2.test.employee")
        checkAggregateRemoved(df, ansiMode)
        val expectedPlanFragment = if (ansiMode) {
          "PushedAggregates: [SUM(2147483647 + DEPT)], " +
            "PushedFilters: [], " +
            "PushedGroupByExpressions: []"
        } else {
          "PushedFilters: []"
        }
        checkPushedInfo(df, expectedPlanFragment)
        if (ansiMode) {
          val e = intercept[SparkException] {
            checkAnswer(df, Seq(Row(-10737418233L)))
          }
          assert(e.getMessage.contains(
            "org.h2.jdbc.JdbcSQLDataException: Numeric value out of range: \"2147483648\""))
        } else {
          checkAnswer(df, Seq(Row(-10737418233L)))
        }
      }
    }
  }

  test("scan with aggregate push-down: aggregate function with UDF") {
    val df = spark.table("h2.test.employee")
    val decrease = udf { (x: Double, y: Double) => x - y }
    val query = df.select(sum(decrease($"SALARY", $"BONUS")).as("value"))
    checkAggregateRemoved(query, false)
    checkPushedInfo(query, "PushedFilters: []")
    checkAnswer(query, Seq(Row(47100.0)))
  }

  test("scan with aggregate push-down: partition columns with multi group by columns") {
    val df = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .groupBy($"dept", $"name")
      .count()
    checkAggregateRemoved(df, false)
    checkAnswer(df, Seq(Row(1, "amy", 1), Row(1, "cathy", 1),
      Row(2, "alex", 1), Row(2, "david", 1), Row(6, "jen", 1)))
  }

  test("scan with aggregate push-down: partition columns is different from group by columns") {
    val df = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .groupBy($"name")
      .count()
    checkAggregateRemoved(df, false)
    checkAnswer(df,
      Seq(Row("alex", 1), Row("amy", 1), Row("cathy", 1), Row("david", 1), Row("jen", 1)))
  }

  test("column name with composite field") {
    checkAnswer(sql("SELECT `dept id`, `dept.id` FROM h2.test.dept"), Seq(Row(1, 1), Row(2, 1)))

    val df1 = sql("SELECT COUNT(`dept id`) FROM h2.test.dept")
    checkPushedInfo(df1, "PushedAggregates: [COUNT(`dept id`)]")
    checkAnswer(df1, Seq(Row(2)))

    val df2 = sql("SELECT `dept.id`, COUNT(`dept id`) FROM h2.test.dept GROUP BY `dept.id`")
    checkPushedInfo(df2,
      "PushedGroupByExpressions: [`dept.id`]", "PushedAggregates: [COUNT(`dept id`)]")
    checkAnswer(df2, Seq(Row(1, 2)))

    val df3 = sql("SELECT `dept id`, COUNT(`dept.id`) FROM h2.test.dept GROUP BY `dept id`")
    checkPushedInfo(df3,
      "PushedGroupByExpressions: [`dept id`]", "PushedAggregates: [COUNT(`dept.id`)]")
    checkAnswer(df3, Seq(Row(1, 1), Row(2, 1)))
  }

  test("column name with non-ascii") {
    // scalastyle:off
    checkAnswer(sql("SELECT `名` FROM h2.test.person"), Seq(Row(1), Row(2)))
    val df = sql("SELECT COUNT(`名`) FROM h2.test.person")
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [COUNT(`名`)]")
    checkAnswer(df, Seq(Row(2)))
    // scalastyle:on
  }

  test("scan with aggregate push-down: complete push-down SUM, AVG, COUNT") {
    val df = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "1")
      .table("h2.test.employee")
      .agg(sum($"SALARY").as("sum"), avg($"SALARY").as("avg"), count($"SALARY").as("count"))
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [SUM(SALARY), AVG(SALARY), COUNT(SALARY)]")
    checkAnswer(df, Seq(Row(53000.00, 10600.000000, 5)))

    val df2 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "1")
      .table("h2.test.employee")
      .groupBy($"name")
      .agg(sum($"SALARY").as("sum"), avg($"SALARY").as("avg"), count($"SALARY").as("count"))
    checkAggregateRemoved(df)
    checkPushedInfo(df, "PushedAggregates: [SUM(SALARY), AVG(SALARY), COUNT(SALARY)]")
    checkAnswer(df2, Seq(
      Row("alex", 12000.00, 12000.000000, 1),
      Row("amy", 10000.00, 10000.000000, 1),
      Row("cathy", 9000.00, 9000.000000, 1),
      Row("david", 10000.00, 10000.000000, 1),
      Row("jen", 12000.00, 12000.000000, 1)))
  }

  test("scan with aggregate push-down: partial push-down SUM, AVG, COUNT") {
    val df = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .agg(sum($"SALARY").as("sum"), avg($"SALARY").as("avg"), count($"SALARY").as("count"))
    checkAggregateRemoved(df, false)
    checkPushedInfo(df, "PushedAggregates: [SUM(SALARY), COUNT(SALARY)]")
    checkAnswer(df, Seq(Row(53000.00, 10600.000000, 5)))

    val df2 = spark.read
      .option("partitionColumn", "dept")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .groupBy($"name")
      .agg(sum($"SALARY").as("sum"), avg($"SALARY").as("avg"), count($"SALARY").as("count"))
    checkAggregateRemoved(df, false)
    checkPushedInfo(df, "PushedAggregates: [SUM(SALARY), COUNT(SALARY)]")
    checkAnswer(df2, Seq(
      Row("alex", 12000.00, 12000.000000, 1),
      Row("amy", 10000.00, 10000.000000, 1),
      Row("cathy", 9000.00, 9000.000000, 1),
      Row("david", 10000.00, 10000.000000, 1),
      Row("jen", 12000.00, 12000.000000, 1)))
  }

  test("SPARK-37895: JDBC push down with delimited special identifiers") {
    val df = sql(
      """SELECT h2.test.view1.`|col1`, h2.test.view1.`|col2`, h2.test.view2.`|col3`
        |FROM h2.test.view1 LEFT JOIN h2.test.view2
        |ON h2.test.view1.`|col1` = h2.test.view2.`|col1`""".stripMargin)
    checkAnswer(df, Seq.empty[Row])
  }

  test("scan with aggregate push-down: complete push-down aggregate with alias") {
    val df = spark.table("h2.test.employee")
      .select($"DEPT", $"SALARY".as("mySalary"))
      .groupBy($"DEPT")
      .agg(sum($"mySalary").as("total"))
      .filter($"total" > 1000)
    checkAggregateRemoved(df)
    checkPushedInfo(df,
      "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT]")
    checkAnswer(df, Seq(Row(1, 19000.00), Row(2, 22000.00), Row(6, 12000.00)))

    val df2 = spark.table("h2.test.employee")
      .select($"DEPT".as("myDept"), $"SALARY".as("mySalary"))
      .groupBy($"myDept")
      .agg(sum($"mySalary").as("total"))
      .filter($"total" > 1000)
    checkAggregateRemoved(df2)
    checkPushedInfo(df2,
      "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT]")
    checkAnswer(df2, Seq(Row(1, 19000.00), Row(2, 22000.00), Row(6, 12000.00)))
  }

  test("scan with aggregate push-down: partial push-down aggregate with alias") {
    val df = spark.read
      .option("partitionColumn", "DEPT")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .select($"NAME", $"SALARY".as("mySalary"))
      .groupBy($"NAME")
      .agg(sum($"mySalary").as("total"))
      .filter($"total" > 1000)
    checkAggregateRemoved(df, false)
    checkPushedInfo(df,
      "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [NAME]")
    checkAnswer(df, Seq(Row("alex", 12000.00), Row("amy", 10000.00),
      Row("cathy", 9000.00), Row("david", 10000.00), Row("jen", 12000.00)))

    val df2 = spark.read
      .option("partitionColumn", "DEPT")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.employee")
      .select($"NAME".as("myName"), $"SALARY".as("mySalary"))
      .groupBy($"myName")
      .agg(sum($"mySalary").as("total"))
      .filter($"total" > 1000)
    checkAggregateRemoved(df2, false)
    checkPushedInfo(df2,
      "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [NAME]")
    checkAnswer(df2, Seq(Row("alex", 12000.00), Row("amy", 10000.00),
      Row("cathy", 9000.00), Row("david", 10000.00), Row("jen", 12000.00)))
  }

  test("scan with aggregate push-down: partial push-down AVG with overflow") {
    def createDataFrame: DataFrame = spark.read
      .option("partitionColumn", "id")
      .option("lowerBound", "0")
      .option("upperBound", "2")
      .option("numPartitions", "2")
      .table("h2.test.item")
      .agg(avg($"PRICE").as("avg"))

    Seq(true, false).foreach { ansiEnabled =>
      withSQLConf((SQLConf.ANSI_ENABLED.key, ansiEnabled.toString)) {
        val df = createDataFrame
        checkAggregateRemoved(df, false)
        df.queryExecution.optimizedPlan.collect {
          case _: DataSourceV2ScanRelation =>
            val expected_plan_fragment =
              "PushedAggregates: [SUM(PRICE), COUNT(PRICE)]"
            checkKeywordsExistsInExplain(df, expected_plan_fragment)
        }
        if (ansiEnabled) {
          val e = intercept[SparkException] {
            df.collect()
          }
          assert(e.getCause.isInstanceOf[ArithmeticException])
          assert(e.getCause.getMessage.contains("cannot be represented as Decimal") ||
            e.getCause.getMessage.contains("Overflow in sum of decimals"))
        } else {
          checkAnswer(df, Seq(Row(null)))
        }
      }
    }
  }

  test("register dialect specific functions") {
    val df = sql("SELECT h2.my_avg(id) FROM h2.test.people")
    checkAggregateRemoved(df, false)
    checkAnswer(df, Row(1) :: Nil)
    val e1 = intercept[AnalysisException] {
      checkAnswer(sql("SELECT h2.test.my_avg2(id) FROM h2.test.people"), Seq.empty)
    }
    assert(e1.getMessage.contains("Undefined function: h2.test.my_avg2"))
    val e2 = intercept[AnalysisException] {
      checkAnswer(sql("SELECT h2.my_avg2(id) FROM h2.test.people"), Seq.empty)
    }
    assert(e2.getMessage.contains("Undefined function: h2.my_avg2"))
  }
}


text:
.SettingsTabPaneWidgetType-tab-mid {
   background: transparent url(tab_inact_mid.png) repeat-x;
   height: 100%;float: left;
   line-height: -25px;
   cursor: pointer; margin: 10px; float: right;
}

/* testing embedded PHP */
li {
    background:url(<?php print $staticserver; ?>/images/<?php print $staticdir; ?>/bullet.gif) left <?php print $left; ?>px no-repeat; margin:0px; padding-left:10px; margin-bottom:<?php echo $marginBottom; ?>px; line-height:13px;
    filter: progid:DXImageTransform.Microsoft.gradient(GradientType=0, startColorstr=#2e62a8, endColorstr=#123363);
}

/* Document handling of comments and annotations. */
div#annotations {-webkit-tap-highlight-color:transparent;/* phpcs:disable Standard.Cat.SniffName */-webkit-touch-callout:none;/*phpcs:enable*/-webkit-user-select:none;}

div#comments {height:100%;/*comment*/width:100%;}

redpajama_set_name: RedPajamaGithub


text:
title: Announcing Istio 1.1.10
linktitle: 1.1.10
subtitle: Patch Release
description: Istio 1.1.10 patch release.
publishdate: 2019-06-28
release: 1.1.10
aliases:
    - /about/notes/1.1.10
    - /blog/2019/announcing-1.1.10
    - /news/2019/announcing-1.1.10
    - /news/announcing-1.1.10
---

We're pleased to announce the availability of Istio 1.1.10. Please see below for what's changed.

{{< relnote >}}

## Bug fixes

- Eliminate 503 errors caused by Envoy not being able to talk to the SDS Node Agent after a restart ([Issue 14853](https://github.com/istio/istio/issues/14853)).
- Fix cause of 'TLS error: Secret is not supplied by SDS' errors during upgrade ([Issue 15020](https://github.com/istio/istio/issues/15020)).
- Fix crash in Istio's JWT Envoy filter caused by malformed JWT ([Issue 15084](https://github.com/istio/istio/issues/15084)).


text:
<?php

namespace Kunstmaan\TranslatorBundle\Command;

use Kunstmaan\TranslatorBundle\Service\Translator\CacheValidator;
use Kunstmaan\TranslatorBundle\Service\Translator\ResourceCacher;
use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Serializer\Exception\InvalidArgumentException;

/**
 * @final since 5.1
 * NEXT_MAJOR extend from `Command` and remove `$this->getContainer` usages
 */
class TranslationCacheCommand extends ContainerAwareCommand
{
    /**
     * @var ResourceCacher
     */
    private $resourceCacher;

    /**
     * @var CacheValidator
     */
    private $cacheValidator;

    /**
     * @param ResourceCacher|null $resourceCacher
     * @param CacheValidator|null $cacheValidator
     */
    public function __construct(/* ResourceCacher */ $resourceCacher = null, /* CacheValidator */ $cacheValidator = null)
    {
        parent::__construct();

        if (!$resourceCacher instanceof ResourceCacher) {
            @trigger_error(sprintf('Passing a command name as the first argument of "%s" is deprecated since version symfony 3.4 and will be removed in symfony 4.0. If the command was registered by convention, make it a service instead. ', __METHOD__), E_USER_DEPRECATED);

            $this->setName(null === $resourceCacher ? 'kuma:translator:cache' : $resourceCacher);

            return;
        }

        $this->resourceCacher = $resourceCacher;
        $this->cacheValidator = $cacheValidator;
    }

    protected function configure()
    {
        $this
        ->setName('kuma:translator:cache')
        ->setDescription('Request cache status and flush cache')
        ->addOption('flush',        'f',    InputOption::VALUE_NONE,        'Flush translation cache (if any)')
        ->addOption('status',      null,    InputOption::VALUE_NONE,    'Request cache status')
        ;
    }

    protected function execute(InputInterface $input, OutputInterface $output)
    {

        if ($input->getOption('flush')) {
            return $this->flushTranslationCache($input, $output);
        } elseif ($input->getOption('status')) {
            return $this->showTranslationCacheStatus($input, $output);
        }

        throw new InvalidArgumentException('No or invalid option provided');
    }

    public function flushTranslationCache(InputInterface $input, OutputInterface $output)
    {
        if (null === $this->resourceCacher) {
            $this->resourceCacher = $this->getContainer()->get('kunstmaan_translator.service.translator.resource_cacher');
        }

        if ($this->resourceCacher->flushCache()) {
            $output->writeln('<info>Translation cache succesfully flushed</info>');
        }
    }

    public function showTranslationCacheStatus(InputInterface $input, OutputInterface $output)
    {
        if (null === $this->cacheValidator) {
            $this->cacheValidator = $this->getContainer()->get('kunstmaan_translator.service.translator.cache_validator');
        }

        $oldestFile = $this->cacheValidator->getOldestCachefileDate();
        $newestTranslation = $this->cacheValidator->getLastTranslationChangeDate();
        $isFresh = $this->cacheValidator->isCacheFresh();

        $output->writeln(sprintf('Oldest file mtime: <info>%s</info>', $oldestFile instanceof \DateTime ? $oldestFile->format('Y-m-d H:i:s') : 'none found'));
        $output->writeln(sprintf('Newest translation (in stash): <info>%s</info>', $newestTranslation instanceof \DateTime ? $newestTranslation->format('Y-m-d H:i:s') : 'none found'));
        $output->writeln(sprintf('Status: <info>%s</info>', $isFresh ? 'fresh' : 'outdated'));

        return 0;
    }
}


text:
%!TEX TS-program = xelatex
%!TEX encoding = UTF-8 Unicode
% Awesome CV LaTeX Template for Cover Letter
%
% This template has been downloaded from:
% https://github.com/posquit0/Awesome-CV
%
% Authors:
% Claud D. Park <posquit0.bj@gmail.com>
% Lars Richter <mail@ayeks.de>
%
% Template license:
% CC BY-SA 4.0 (https://creativecommons.org/licenses/by-sa/4.0/)
%

%-------------------------------------------------------------------------------
% CONFIGURATIONS
%-------------------------------------------------------------------------------
% A4 paper size by default, use 'letterpaper' for US letter
\documentclass[11pt, a4paper]{awesome-cv}

% Configure page margins with geometry
\geometry{left=1.4cm, top=.8cm, right=1.4cm, bottom=1.8cm, footskip=.5cm}

% Specify the location of the included fonts
\fontdir[fonts/]

% Color for highlights
% Awesome Colors: awesome-emerald, awesome-skyblue, awesome-red, awesome-pink, awesome-orange
%                 awesome-nephritis, awesome-concrete, awesome-darknight
\colorlet{awesome}{awesome-red}
% Uncomment if you would like to specify your own color
% \definecolor{awesome}{HTML}{CA63A8}

% Colors for text
% Uncomment if you would like to specify your own color
% \definecolor{darktext}{HTML}{414141}
% \definecolor{text}{HTML}{333333}
% \definecolor{graytext}{HTML}{5D5D5D}
% \definecolor{lighttext}{HTML}{999999}

% Set false if you don't want to highlight section with awesome color
\setbool{acvSectionColorHighlight}{true}

% If you would like to change the social information separator from a pipe (|) to something else
\renewcommand{\acvHeaderSocialSep}{\quad\textbar\quad}

%-------------------------------------------------------------------------------
%	PERSONAL INFORMATION
%	Comment any of the lines below if they are not required
%-------------------------------------------------------------------------------
% Available options: circle|rectangle,edge/noedge,left/right
\photo[circle,noedge,left]{./examples/profile}
\name{Claud D.}{Park}
\position{Software Engineer{\enskip\cdotp\enskip}Security Expert}
\address{246-1002, Gwangmyeongmayrouge Apt. 86, Cheongna lime-ro, Seo-gu, Incheon-si, 404-180, Rep. of KOREA}

\mobile{(+82) 10-9030-1843}
\email{posquit0.bj@gmail.com}
\homepage{www.posquit0.com}
\github{posquit0}
\linkedin{posquit0}
% \gitlab{gitlab-id}
% \stackoverflow{SO-id}{SO-name}
% \twitter{@twit}
% \skype{skype-id}
% \reddit{reddit-id}
% \extrainfo{extra informations}

\quote{``Be the change that you want to see in the world."}

%-------------------------------------------------------------------------------
%	LETTER INFORMATION
%	All of the below lines must be filled out
%-------------------------------------------------------------------------------
% The company being applied to
\recipient
  {Company Recruitment Team}
  {Google Inc.\\1600 Amphitheatre Parkway\\Mountain View, CA 94043}
% The date on the letter, default is the date of compilation
\letterdate{\today}
% The title of the letter
\lettertitle{Job Application for Software Engineer}
% How the letter is opened
\letteropening{Dear Mr./Ms./Dr. LastName,}
% How the letter is closed
\letterclosing{Sincerely,}
% Any enclosures with the letter
\letterenclosure[Attached]{Curriculum Vitae}

%-------------------------------------------------------------------------------
\begin{document}

% Print the header with above personal informations
% Give optional argument to change alignment(C: center, L: left, R: right)
\makecvheader[R]

% Print the footer with 3 arguments(<left>, <center>, <right>)
% Leave any of these blank if they are not needed
\makecvfooter
  {\today}
  {Claud D. Park~~~·~~~Cover Letter}
  {}

% Print the title with above letter informations
\makelettertitle

%-------------------------------------------------------------------------------
%	LETTER CONTENT
%-------------------------------------------------------------------------------
\begin{cvletter}

\lettersection{About Me}
Lorem ipsum dolor sit amet, consectetur adipiscing elit. Duis ullamcorper neque sit amet lectus facilisis sed luctus nisl iaculis. Vivamus at neque arcu, sed tempor quam. Curabitur pharetra tincidunt tincidunt. Morbi volutpat feugiat mauris, quis tempor neque vehicula volutpat. Duis tristique justo vel massa fermentum accumsan. Mauris ante elit, feugiat vestibulum tempor eget, eleifend ac ipsum. Donec scelerisque lobortis ipsum eu vestibulum. Pellentesque vel massa at felis accumsan rhoncus.

\lettersection{Why Google?}
Suspendisse commodo, massa eu congue tincidunt, elit mauris pellentesque orci, cursus tempor odio nisl euismod augue. Aliquam adipiscing nibh ut odio sodales et pulvinar tortor laoreet. Mauris a accumsan ligula. Class aptent taciti sociosqu ad litora torquent per conubia nostra, per inceptos himenaeos. Suspendisse vulputate sem vehicula ipsum varius nec tempus dui dapibus. Phasellus et est urna, ut auctor erat. Sed tincidunt odio id odio aliquam mattis. Donec sapien nulla, feugiat eget adipiscing sit amet, lacinia ut dolor. Phasellus tincidunt, leo a fringilla consectetur, felis diam aliquam urna, vitae aliquet lectus orci nec velit. Vivamus dapibus varius blandit.

\lettersection{Why Me?}
Duis sit amet magna ante, at sodales diam. Aenean consectetur porta risus et sagittis. Ut interdum, enim varius pellentesque tincidunt, magna libero sodales tortor, ut fermentum nunc metus a ante. Vivamus odio leo, tincidunt eu luctus ut, sollicitudin sit amet metus. Nunc sed orci lectus. Ut sodales magna sed velit volutpat sit amet pulvinar diam venenatis.

\end{cvletter}

%-------------------------------------------------------------------------------
% Print the signature and enclosures with above letter informations
\makeletterclosing

\end{document}


text:
namespace raytracer {

    struct PhotonMapped : public Phong {
        PhotonMapped(float ka, float kd, float ks, float ke, float kt, float ki, bool is_hollow)
            : Phong(ka, kd, ks, ke, kt, ki, is_hollow)
        { }

        glm::vec3 get_color(Kernel* kernel, HitResult& hit) override;
    };
}


text: export * from './compare';


text:
require_relative 'helper'

module ParseGemspec
  module Cli
    class TestGemspec < Test::Unit::TestCase
      test 'version' do
        assert do
          !::ParseGemspec::Cli::VERSION.nil?
        end
      end
    end
  end
end


text:
<!DOCTYPE html>

  <meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">

<title>Page not found (404) • leafcutter</title>

<link href="https://cdnjs.cloudflare.com/ajax/libs/bootswatch/3.4.0/cerulean/bootstrap.min.css" rel="stylesheet" crossorigin="anonymous" />

<link rel="stylesheet" href="bootstrap-toc.css">

<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/5.12.1/css/all.min.css" integrity="sha256-mmgLkCYLUQbXn0B1SRqzHar6dCnv9oZFPEC1g1cwlkk=" crossorigin="anonymous" />
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/5.12.1/css/v4-shims.min.css" integrity="sha256-wZjR52fzng1pJHwx4aV2AO3yyTOXrcDW7jBpJtTwVxw=" crossorigin="anonymous" />

<link href="pkgdown.css" rel="stylesheet">

<meta property="og:title" content="Page not found (404)" />

      <button type="button" class="navbar-toggle collapsed" data-toggle="collapse" data-target="#navbar" aria-expanded="false">
        Toggle navigation

      </button>

        leafcutter
        0.2.9

  Installation

  Differential splicing

  Outlier splicing

  Visualization

  SplicingQTL

  R functions

Page not found (404)

Content not found. Please use links in the navbar.

Contents

Developed by David Knowles, Yang Li, Jack Humphrey, Jonathan Pritchard, Garrett Jenkinson.

Site built with pkgdown 1.6.1.


text:
import math

def sumPerIndex(p, q):
    for i in range(len(p)):
        p[i] = p[i] + q[i]

    return p

def average(dataList):
    sum = 0.0
    for data in dataList:
        sum = sum + float(data)

    average = sum / len(dataList)
    return average

def standardDeviation(average, dataList):
    sum = 0.0
    for data in dataList:
        sum = sum + pow(float(data) - average, 2)

    standardDeviation = math.sqrt(sum / len(dataList))
    return standardDeviation

def standardizing(average, standardDeviation, dataList):
    standardizingList = []
    for data in dataList:
        z = (float(data) - average) / standardDeviation
        standardizingList.append(z)

    return standardizingList

def countFrequency(pieceCount, dataList):
    maxValue = 5.0
    minValue = -5.0
    totalLength = maxValue + abs(minValue)
    interval = totalLength / float(pieceCount)
    frequencyDistributionList = [0 for _ in range(pieceCount+2)] # 양쪽 최소, 최대값

    for data in dataList:
        if maxValue < float(data): # 왼쪽 끝 최소값
            frequencyDistributionList[pieceCount+1] = frequencyDistributionList[pieceCount+1] + 1
        elif float(data) < minValue: # 오른쪽 끝 최대값
            frequencyDistributionList[0] = frequencyDistributionList[0] + 1
        else:
            for i in range(1, pieceCount):
                if float(data) < (minValue + (i * interval)):
                    frequencyDistributionList[i] = frequencyDistributionList[i] + 1
                    break

    return frequencyDistributionList

def probabilityDistribution(pieceCount, standardizingList):
    frequencyDistributionList = countFrequency(pieceCount, standardizingList)
    probabilityDistribution = [0 for _ in range(pieceCount+2)]

    for i in range(0, pieceCount+2):
        probabilityDistribution[i] = float(frequencyDistributionList[i]) / float(len(standardizingList))

    return probabilityDistribution


text:
const fs = require("fs");
const path = require("path");
const cp = require("child_process");

if (!fs.existsSync(path.join(__dirname, "node_modules"))) {
    try {
        cp.execSync("npm install --no-optional");
    } catch (error) {
        console.error("Justcord could not perform initial setup. \nMake sure NPM is installed and try again.", error);
        module.exports = error;
    }
}


text:
# PokemonGo Map

[![Build Status](https://travis-ci.org/AHAAAAAAA/PokemonGo-Map.svg?branch=master)](https://travis-ci.org/AHAAAAAAA/PokemonGo-Map)  [![Coverage Status](https://coveralls.io/repos/github/AHAAAAAAA/PokemonGo-Map/badge.svg?branch=master)](https://coveralls.io/github/AHAAAAAAA/PokemonGo-Map?branch=master)

Live visualization of all the pokemon (with option to show gyms and pokestops) in your area. This is a proof of concept that we can load all the pokemon visible nearby given a location. Currently runs on a Flask server displaying a Google Maps with markers on it.

Building off [Mila432](https://github.com/Mila432/Pokemon_Go_API)'s PokemonGo API, [tejado's additions](https://github.com/tejado/pokemongo-api-demo), [leegao's additions](https://github.com/leegao/pokemongo-api-demo/tree/simulation) and [Flask-GoogleMaps](https://github.com/rochacbruno/Flask-GoogleMaps).

---

# Requirements
* Python 2.7.x

# Installation
`pip install -r requirements.txt`

# Usage
`python example.py -a authService -u myUsername -p myPassword -l "Boulder, CO" -st 5`

| Flag                       | Description                                            | Required |
|----------------------------|--------------------------------------------------------|----------|
| `-a` `--auth-service`      | Auth service (PTC or google)                           |          |
| `-u` `--username`          | Username                                               | ✓        |
| `-p` `--password`          | Password                                               | ✓        |
| `-l` `--location`          | Any location Google Maps can understand                | ✓        |
| `-st` `--step-limit`       | Steps to take                                          | ✓        |
| `-i` `--ignore`            | Comma-separated list of Pokémon names or IDs to ignore |          |
| `-o` `--only`              | Comma-separated list of Pokémon names or IDs to search |          |
| `-ar` `--auto-refresh`     | Number of seconds on which to autorefresh              |          |
| `-dp` `--display-pokestop` | Display pokéstops                                      |          |
| `-dg` `--display-gym`      | Display gyms                                           |          |
| `-H` `--host`              | Set web server listening host                          |          |
| `-P` `--port`              | Set web server listening port                          |          |
| `-L` `--locale`            | Locale for Pokemon names: default en, check locale folder for more options |          |
| `-c` `--china`             | Coordinates transformer for China                      |          |
| `-d` `--debug`             | Debug mode                                             |          |
| `-ol` `--onlylure`         | Display only lured pokéstop                            |          |

_Note:
5 steps is approximately a 1.2km radius. More than 10 is redundant (you usually can't walk that far before despawn anyway)_

## Working examples
### ignore Pokémon
`python example.py -a PTC -u myUsername -p myPassword -l "Central Park, New York, NY" -st 5 -i Pidgey,Weedle,Zubat`

### make server externally visible
#### IPv4
`python example.py -a PTC -u myUsername -p myPassword -l "Central Park, New York, NY" -st 5 -H 0.0.0.0 -P 5000`
#### IPv6
`python example.py -a PTC -u myUsername -p myPassword -l "Central Park, New York, NY" -st 5 -H :: -P 5000`

## Common Errors
> _`Exception, e <- Invalid syntax.`_

* You are using python 3, download python 2.7 instead.

> _`pip or python is not recognized as an internal or external command`_

* Replace pip with C:\Python27\Scripts\pip
* Replace python with C:\Python27\python

## FAQ
> _Can I sign in with Google?_

* Yes you can! Pass the flag `-a google` to use Google authentication

> _I'm on Windows, why does nothing work?_

* See if anything in https://www.reddit.com/r/pokemongodev/comments/4t80df/wip_pokemon_go_map_visualization_google_maps_view/d5feu2f helps


text:
DROP TABLE IF EXISTS BATCH_STEP_EXECUTION_CONTEXT /*CASCADE CONSTRAINTS*/;

DROP TABLE IF EXISTS BATCH_JOB_EXECUTION_CONTEXT /*CASCADE CONSTRAINTS*/;

DROP TABLE IF EXISTS BATCH_STEP_EXECUTION /*CASCADE CONSTRAINTS*/;

DROP TABLE IF EXISTS BATCH_JOB_EXECUTION_PARAMS /*CASCADE CONSTRAINTS*/;

DROP TABLE IF EXISTS BATCH_JOB_EXECUTION /*CASCADE CONSTRAINTS*/;

DROP TABLE IF EXISTS BATCH_JOB_INSTANCE /*CASCADE CONSTRAINTS*/;


text:
<!DOCTYPE html>

    <meta charset="utf-8">
    <meta http-equiv="x-ua-compatible" content="ie=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta name="generator" content="ExDoc v0.11.4">
    <title>FnDef – Function Decorating v0.0.6</title>
    <link rel="stylesheet" href="" />

<button class="sidebar-toggle">

</button>

  <button class="sidebar-toggle">

  </button>

        Function Decorating

        v0.0.6

    <input type="text" class="sidebar-searchInput" placeholder="search" autocomplete="off" />

Pages

Modules

        Function Decorating v0.0.6
        FnDef

            Summary

      Functions

    calc_arg_name(arg_ast)

Calcuate argument name from arg expresion

    calc_full_arg(arg_ast, mediator_arg_ast)

Generate AST for argument AST and its mediator

    convert_to_cols(list)

Examples

    decorate_arg(arg)

Add mediator argument to a function argument ast

    decorate_args(args_ast)

Add meidator arguments to a function arguments list ast

    is_operator_token(a)

    parse_fn_name_and_args(short_head)

Parse a function call into its parts: name and arguments

            Functions

    calc_arg_name(arg_ast)

Calcuate argument name from arg expresion.

Returns the first good arg name that can be used. Not _ if possible.

Examples

   iex> FnDef.calc_arg_name(quote do aa end)
   :aa

   iex> FnDef.calc_arg_name(quote do aa = bb end)
   :aa

   iex> FnDef.calc_arg_name(quote do _aa = bb end)
   :bb

   iex> FnDef.calc_arg_name(quote do _ = bb end)
   :bb

   iex> FnDef.calc_arg_name(quote do _ = bb \\ 6 end)
   :bb

   iex> FnDef.calc_arg_name(quote do _ = _bb \\ 6 end)
   :_bb

   iex> FnDef.calc_arg_name(quote do {aa} = bb end)
   :bb

   iex> FnDef.calc_arg_name(quote do [aa] = bb end)
   :bb

    calc_full_arg(arg_ast, mediator_arg_ast)

Specs

calc_full_arg(Macro.t, Macro.t) :: Macro.t

Generate AST for argument AST and its mediator.

Examples

  iex> FnDef.calc_full_arg(quote context: __MODULE__ do first_name end, quote context: __MODULE__ do arg0 end)
  {:=, [], [{:first_name, [], __MODULE__}, {:arg0, [], __MODULE__}]}

    convert_to_cols(list)

Examples

  iex> FnDef.convert_to_cols([{:first_name, {:arg0, [], FnDef}, {:=, [], [{:first_name, [], __MODULE__}, {:arg0, [], FnDef}]}}, {:last_name, {:arg1, [], FnDef}, {:=, [], [{:last_name, [], __MODULE__}, {:arg1, [], FnDef}]}}])
  {[:first_name, :last_name], [{:arg0, [], FnDef}, {:arg1, [], FnDef}], [{:=, [], [{:first_name, [], __MODULE__}, {:arg0, [], FnDef}]}, {:=, [], [{:last_name, [], __MODULE__}, {:arg1, [], FnDef}]}]}

    decorate_arg(arg)

Specs

decorate_arg({Macro.t, non_neg_integer}) :: {Macro.t, Macro.t}

Add mediator argument to a function argument ast.

Returns {decorated argument name, decorated argument ast}

Examples

  iex> FnDef.decorate_arg({quote context: __MODULE__ do first_name end, 0})
  {:first_name, {:arg0, [], FnDef}, {:=, [], [{:first_name, [], __MODULE__}, {:arg0, [], FnDef}]}}

    decorate_args(args_ast)

Specs

decorate_args(list) :: {list, list, list}

Add meidator arguments to a function arguments list ast.

Mainly for the availabity to print unbounded arguments
in a function call.

say(word) -> say(word = arg0)

Returns {[args names], [decorated args ast]}

Examples

  iex> FnDef.decorate_args(quote context: __MODULE__ do [a, b, _] end)
  {[:a, :b, :arg2],
    [{:arg0, [], FnDef}, {:arg1, [], FnDef}, {:arg2, [], FnDef}],
      [{:=, [], [{:a, [], __MODULE__}, {:arg0, [], FnDef}]},
        {:=, [], [{:b, [], __MODULE__}, {:arg1, [], FnDef}]},
        {:=, [], [{:_, [], __MODULE__}, {:arg2, [], FnDef}]}]}

    is_operator_token(a)

    parse_fn_name_and_args(short_head)

Parse a function call into its parts: name and arguments

          Built using
          ExDoc (v0.11.4),

          designed by
          Friedel Ziegelmayer.

      <button class="night-mode-toggle" />


text:
layout: default
---

{% include sections/me.html %}