/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.beam.sdk.extensions.sql.zetasql;

import org.apache.beam.sdk.extensions.sql.impl.rel.BeamRelNode;
import org.apache.beam.sdk.extensions.sql.impl.rel.BeamSqlRelUtils;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.Row;
import org.joda.time.Duration;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

/** Tests for ZetaSQL Math functions (on INT64, DOUBLE, NUMERIC types). */
@RunWith(JUnit4.class)
@SuppressWarnings({
  "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
})
public class ZetaSqlMathFunctionsTest extends ZetaSqlTestBase {

  @Rule public transient TestPipeline pipeline = TestPipeline.create();
  @Rule public ExpectedException thrown = ExpectedException.none();

  @Before
  public void setUp() {
    initialize();
  }

  /////////////////////////////////////////////////////////////////////////////
  // INT64 type tests
  /////////////////////////////////////////////////////////////////////////////

  @Test
  public void testArithmeticOperatorsInt64() {
    String sql = "SELECT -1, 1 + 2, 1 - 2, 1 * 2, 1 / 2";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addInt64Field("f_int64_1")
                        .addInt64Field("f_int64_2")
                        .addInt64Field("f_int64_3")
                        .addInt64Field("f_int64_4")
                        .addDoubleField("f_double")
                        .build())
                .addValues(-1L, 3L, -1L, 2L, 0.5)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testAbsInt64() {
    String sql = "SELECT ABS(1), ABS(-1)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder().addInt64Field("f_int64_1").addInt64Field("f_int64_2").build())
                .addValues(1L, 1L)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testSignInt64() {
    String sql = "SELECT SIGN(0), SIGN(5), SIGN(-5)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addInt64Field("f_int64_1")
                        .addInt64Field("f_int64_2")
                        .addInt64Field("f_int64_3")
                        .build())
                .addValues(0L, 1L, -1L)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testModInt64() {
    String sql = "SELECT MOD(4, 2)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addInt64Field("f_int64").build())
                .addValues(0L)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testDivInt64() {
    String sql = "SELECT DIV(1, 2), DIV(2, 1)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder().addInt64Field("f_int64_1").addInt64Field("f_int64_2").build())
                .addValues(0L, 2L)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testSafeArithmeticFunctionsInt64() {
    String sql =
        "SELECT SAFE_ADD(9223372036854775807, 1), "
            + "SAFE_SUBTRACT(-9223372036854775808, 1), "
            + "SAFE_MULTIPLY(9223372036854775807, 2), "
            + "SAFE_DIVIDE(1, 0), "
            + "SAFE_NEGATE(-9223372036854775808)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addNullableField("f_int64_1", Schema.FieldType.INT64)
                        .addNullableField("f_int64_2", Schema.FieldType.INT64)
                        .addNullableField("f_int64_3", Schema.FieldType.INT64)
                        .addNullableField("f_int64_4", Schema.FieldType.INT64)
                        .addNullableField("f_int64_5", Schema.FieldType.INT64)
                        .build())
                .addValues(null, null, null, null, null)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  /////////////////////////////////////////////////////////////////////////////
  // DOUBLE (FLOAT64) type tests
  /////////////////////////////////////////////////////////////////////////////

  @Test
  public void testDoubleLiteral() {
    String sql =
        "SELECT 3.0, CAST('+inf' AS FLOAT64), CAST('-inf' AS FLOAT64), CAST('NaN' AS FLOAT64)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDoubleField("f_double1")
                        .addDoubleField("f_double2")
                        .addDoubleField("f_double3")
                        .addDoubleField("f_double4")
                        .build())
                .addValues(3.0, Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY, Double.NaN)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testArithmeticOperatorsDouble() {
    String sql = "SELECT -1.5, 1.5 + 2.5, 1.5 - 2.5, 1.5 * 2.5, 1.5 / 2.5";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDoubleField("f_double1")
                        .addDoubleField("f_double2")
                        .addDoubleField("f_double3")
                        .addDoubleField("f_double4")
                        .addDoubleField("f_double5")
                        .build())
                .addValues(-1.5, 4.0, -1.0, 3.75, 0.6)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testEqualsInf() {
    String sql =
        "SELECT CAST('+inf' AS FLOAT64) = CAST('+inf' AS FLOAT64), "
            + "CAST('+inf' AS FLOAT64) = CAST('-inf' AS FLOAT64)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addBooleanField("f_boolean1")
                        .addBooleanField("f_boolean2")
                        .build())
                .addValues(true, false)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testEqualsNaN() {
    String sql = "SELECT CAST('NaN' AS FLOAT64) = CAST('NaN' AS FLOAT64)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addBooleanField("f_boolean").build())
                .addValues(false)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testAbsDouble() {
    String sql = "SELECT ABS(1.5), ABS(-1.0), ABS(CAST('NaN' AS FLOAT64))";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDoubleField("f_double1")
                        .addDoubleField("f_double2")
                        .addDoubleField("f_double3")
                        .build())
                .addValues(1.5, 1.0, Double.NaN)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testSignDouble() {
    String sql = "SELECT SIGN(-0.0), SIGN(1.5), SIGN(-1.5), SIGN(CAST('NaN' AS FLOAT64))";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDoubleField("f_double1")
                        .addDoubleField("f_double2")
                        .addDoubleField("f_double3")
                        .addDoubleField("f_double4")
                        .build())
                .addValues(0.0, 1.0, -1.0, Double.NaN)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testRoundDouble() {
    String sql = "SELECT ROUND(1.23), ROUND(-1.27, 1)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDoubleField("f_double1")
                        .addDoubleField("f_double2")
                        .build())
                .addValues(1.0, -1.3)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testTruncDouble() {
    String sql = "SELECT TRUNC(1.23), TRUNC(-1.27, 1)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDoubleField("f_double1")
                        .addDoubleField("f_double2")
                        .build())
                .addValues(1.0, -1.2)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testCeilDouble() {
    String sql = "SELECT CEIL(1.2), CEIL(-1.2)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDoubleField("f_double1")
                        .addDoubleField("f_double2")
                        .build())
                .addValues(2.0, -1.0)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testFloorDouble() {
    String sql = "SELECT FLOOR(1.2), FLOOR(-1.2)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDoubleField("f_double1")
                        .addDoubleField("f_double2")
                        .build())
                .addValues(1.0, -2.0)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testIsInf() {
    String sql =
        "SELECT IS_INF(CAST('+inf' AS FLOAT64)), IS_INF(CAST('-inf' AS FLOAT64)), IS_INF(3.0)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addBooleanField("f_boolean1")
                        .addBooleanField("f_boolean2")
                        .addBooleanField("f_boolean3")
                        .build())
                .addValues(true, true, false)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testIsNaN() {
    String sql = "SELECT IS_NAN(CAST('NaN' AS FLOAT64)), IS_NAN(3.0)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addBooleanField("f_boolean1")
                        .addBooleanField("f_boolean2")
                        .build())
                .addValues(true, false)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testIeeeDivide() {
    String sql = "SELECT IEEE_DIVIDE(1.0, 0.0)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDoubleField("f_double").build())
                .addValues(Double.POSITIVE_INFINITY)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testSafeDivide() {
    String sql = "SELECT SAFE_DIVIDE(1.0, 0.0)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder().addNullableField("f_double", Schema.FieldType.DOUBLE).build())
                .addValue(null)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testSqrtDouble() {
    String sql = "SELECT SQRT(4.0)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDoubleField("f_double").build())
                .addValues(2.0)
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testPowDouble() {
    String sql = "SELECT POW(2.0, 3.0)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDoubleField("f_double").build())
                .addValues(8.0)
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testExpDouble() {
    String sql = "SELECT EXP(2.0)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDoubleField("f_double").build())
                .addValues(7.38905609893065)
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testLnDouble() {
    String sql = "SELECT LN(7.38905609893065)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDoubleField("f_double").build())
                .addValues(2.0)
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testLog10Double() {
    String sql = "SELECT LOG10(100.0)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDoubleField("f_double").build())
                .addValues(2.0)
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testLogDouble() {
    String sql = "SELECT LOG(2.25, 1.5)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDoubleField("f_double").build())
                .addValues(2.0)
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testTrigonometricFunctions() {
    String sql =
        "SELECT COS(0.0), COSH(0.0), ACOS(1.0), ACOSH(1.0), "
            + "SIN(0.0), SINH(0.0), ASIN(0.0), ASINH(0.0), "
            + "TAN(0.0), TANH(0.0), ATAN(0.0), ATANH(0.0), ATAN2(0.0, 0.0)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDoubleField("f_double1")
                        .addDoubleField("f_double2")
                        .addDoubleField("f_double3")
                        .addDoubleField("f_double4")
                        .addDoubleField("f_double5")
                        .addDoubleField("f_double6")
                        .addDoubleField("f_double7")
                        .addDoubleField("f_double8")
                        .addDoubleField("f_double9")
                        .addDoubleField("f_double10")
                        .addDoubleField("f_double11")
                        .addDoubleField("f_double12")
                        .addDoubleField("f_double13")
                        .build())
                .addValues(1.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0)
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  /////////////////////////////////////////////////////////////////////////////
  // NUMERIC type tests
  /////////////////////////////////////////////////////////////////////////////

  @Test
  public void testNumericLiteral() {
    String sql =
        "SELECT NUMERIC '0', "
            + "NUMERIC '123456', "
            + "NUMERIC '-3.14', "
            + "NUMERIC '-0.54321', "
            + "NUMERIC '1.23456e05', "
            + "NUMERIC '-9.876e-3', "
            // min value for ZetaSQL NUMERIC type
            + "NUMERIC '-99999999999999999999999999999.999999999', "
            // max value for ZetaSQL NUMERIC type
            + "NUMERIC '99999999999999999999999999999.999999999'";
    ;

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDecimalField("f_numeric1")
                        .addDecimalField("f_numeric2")
                        .addDecimalField("f_numeric3")
                        .addDecimalField("f_numeric4")
                        .addDecimalField("f_numeric5")
                        .addDecimalField("f_numeric6")
                        .addDecimalField("f_numeric7")
                        .addDecimalField("f_numeric8")
                        .build())
                .addValues(
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("0"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("123456"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("-3.14"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("-0.54321"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("123456"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("-0.009876"),
                    ZetaSqlTypesUtils.NUMERIC_MIN_VALUE,
                    ZetaSqlTypesUtils.NUMERIC_MAX_VALUE)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testNumericColumn() {
    String sql = "SELECT numeric_field FROM table_with_numeric";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    final Schema schema = Schema.builder().addDecimalField("f_numeric").build();

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(schema)
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("123.4567"))
                .build(),
            Row.withSchema(schema)
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("765.4321"))
                .build(),
            Row.withSchema(schema)
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("-555.5555"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testArithmeticOperatorsNumeric() {
    String sql =
        "SELECT - NUMERIC '1.23456e05', "
            + "NUMERIC '1.23456e05' + NUMERIC '9.876e-3', "
            + "NUMERIC '1.23456e05' - NUMERIC '-9.876e-3', "
            + "NUMERIC '1.23e02' * NUMERIC '-1.001e-3', "
            + "NUMERIC '-1.23123e-1' / NUMERIC '-1.001e-3', ";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDecimalField("f_numeric1")
                        .addDecimalField("f_numeric2")
                        .addDecimalField("f_numeric3")
                        .addDecimalField("f_numeric4")
                        .addDecimalField("f_numeric5")
                        .build())
                .addValues(
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("-123456"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("123456.009876"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("123456.009876"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("-0.123123"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("123"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testAbsNumeric() {
    String sql = "SELECT ABS(NUMERIC '1.23456e04'), ABS(NUMERIC '-1.23456e04')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDecimalField("f_numeric1")
                        .addDecimalField("f_numeric2")
                        .build())
                .addValues(
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("12345.6"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("12345.6"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testSignNumeric() {
    String sql = "SELECT SIGN(NUMERIC '0'), SIGN(NUMERIC '1.23e01'), SIGN(NUMERIC '-1.23e01')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDecimalField("f_numeric1")
                        .addDecimalField("f_numeric2")
                        .addDecimalField("f_numeric3")
                        .build())
                .addValues(
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("0"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("1"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("-1"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testRoundNumeric() {
    String sql = "SELECT ROUND(NUMERIC '1.23456e04'), ROUND(NUMERIC '-1.234567e04', 1)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDecimalField("f_numeric1")
                        .addDecimalField("f_numeric2")
                        .build())
                .addValues(
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("12346"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("-12345.7"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testTruncNumeric() {
    String sql = "SELECT TRUNC(NUMERIC '1.23456e04'), TRUNC(NUMERIC '-1.234567e04', 1)";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDecimalField("f_numeric1")
                        .addDecimalField("f_numeric2")
                        .build())
                .addValues(
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("12345"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("-12345.6"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testCeilNumeric() {
    String sql = "SELECT CEIL(NUMERIC '1.23456e04'), CEIL(NUMERIC '-1.23456e04')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDecimalField("f_numeric1")
                        .addDecimalField("f_numeric2")
                        .build())
                .addValues(
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("12346"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("-12345"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testFloorNumeric() {
    String sql = "SELECT FLOOR(NUMERIC '1.23456e04'), FLOOR(NUMERIC '-1.23456e04')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addDecimalField("f_numeric1")
                        .addDecimalField("f_numeric2")
                        .build())
                .addValues(
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("12345"),
                    ZetaSqlTypesUtils.bigDecimalAsNumeric("-12346"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testModNumeric() {
    String sql = "SELECT MOD(NUMERIC '1.23456e05', NUMERIC '5')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDecimalField("f_numeric").build())
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("1"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testDivNumeric() {
    String sql = "SELECT DIV(NUMERIC '1.23456e05', NUMERIC '5')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDecimalField("f_numeric").build())
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("24691"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testSafeArithmeticFunctionsNumeric() {
    String sql =
        "SELECT SAFE_ADD(NUMERIC '99999999999999999999999999999.999999999', NUMERIC '1'), "
            + "SAFE_SUBTRACT(NUMERIC '-99999999999999999999999999999.999999999', NUMERIC '1'), "
            + "SAFE_MULTIPLY(NUMERIC '99999999999999999999999999999.999999999', NUMERIC '2'), "
            + "SAFE_DIVIDE(NUMERIC '1.23456e05', NUMERIC '0'), "
            + "SAFE_NEGATE(NUMERIC '99999999999999999999999999999.999999999')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(
                    Schema.builder()
                        .addNullableField("f_numeric1", Schema.FieldType.DECIMAL)
                        .addNullableField("f_numeric2", Schema.FieldType.DECIMAL)
                        .addNullableField("f_numeric3", Schema.FieldType.DECIMAL)
                        .addNullableField("f_numeric4", Schema.FieldType.DECIMAL)
                        .addNullableField("f_numeric5", Schema.FieldType.DECIMAL)
                        .build())
                .addValues(null, null, null, null, ZetaSqlTypesUtils.NUMERIC_MIN_VALUE)
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testSqrtNumeric() {
    String sql = "SELECT SQRT(NUMERIC '4')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDecimalField("f_numeric").build())
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("2"))
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testPowNumeric() {
    String sql = "SELECT POW(NUMERIC '2', NUMERIC '3')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDecimalField("f_numeric").build())
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("8"))
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testExpNumeric() {
    String sql = "SELECT EXP(NUMERIC '2')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDecimalField("f_numeric").build())
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("7.389056099"))
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testLnNumeric() {
    String sql = "SELECT LN(NUMERIC '7.389056099')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDecimalField("f_numeric").build())
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("2"))
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testLog10Numeric() {
    String sql = "SELECT LOG10(NUMERIC '100')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDecimalField("f_numeric").build())
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("2"))
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  public void testLogNumeric() {
    String sql = "SELECT LOG(NUMERIC '2.25', NUMERIC '1.5')";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDecimalField("f_numeric").build())
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("2"))
                .build());

    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  @Ignore("[BEAM-10459] Aggregation functions on NUMERIC is not supported yet")
  public void testSumNumeric() {
    String sql = "SELECT SUM(numeric_field) FROM table_with_numeric";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDecimalField("f_numeric").build())
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("333.3333"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }

  @Test
  @Ignore("[BEAM-10459] Aggregation functions on NUMERIC is not supported yet")
  public void testAvgNumeric() {
    String sql = "SELECT AVG(numeric_field) FROM table_with_numeric";

    ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);
    BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
    PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

    PAssert.that(stream)
        .containsInAnyOrder(
            Row.withSchema(Schema.builder().addDecimalField("f_numeric").build())
                .addValues(ZetaSqlTypesUtils.bigDecimalAsNumeric("111.1111"))
                .build());
    pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
  }
}
