/*
 * 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 com.google.zetasql.ZetaSQLFunction.FunctionSignatureId;
import java.util.List;
import java.util.Map;
import org.apache.beam.sdk.annotations.Internal;
import org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.sql.SqlOperator;
import org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;

/** SqlStdOperatorMappingTable. */
@Internal
public class SqlStdOperatorMappingTable {
  static final List<FunctionSignatureId> ZETASQL_BUILTIN_FUNCTION_ALLOWLIST =
      ImmutableList.of(
          FunctionSignatureId.FN_AND,
          FunctionSignatureId.FN_ANY_VALUE,
          FunctionSignatureId.FN_STRING_AGG_STRING,
          FunctionSignatureId.FN_BIT_OR_INT64,
          // JIRA link: https://issues.apache.org/jira/browse/BEAM-10379
          // FunctionSignatureId.FN_BIT_AND_INT64,
          FunctionSignatureId.FN_OR,
          FunctionSignatureId.FN_NOT,
          FunctionSignatureId.FN_MULTIPLY_DOUBLE,
          FunctionSignatureId.FN_MULTIPLY_INT64,
          FunctionSignatureId.FN_MULTIPLY_NUMERIC,
          FunctionSignatureId.FN_DIVIDE_DOUBLE,
          FunctionSignatureId.FN_DIVIDE_NUMERIC,
          FunctionSignatureId.FN_ADD_DOUBLE,
          FunctionSignatureId.FN_ADD_INT64,
          FunctionSignatureId.FN_ADD_NUMERIC,
          FunctionSignatureId.FN_SUBTRACT_DOUBLE,
          FunctionSignatureId.FN_SUBTRACT_INT64,
          FunctionSignatureId.FN_SUBTRACT_NUMERIC,
          FunctionSignatureId.FN_UNARY_MINUS_INT64,
          FunctionSignatureId.FN_UNARY_MINUS_DOUBLE,
          FunctionSignatureId.FN_UNARY_MINUS_NUMERIC,
          FunctionSignatureId.FN_GREATER,
          FunctionSignatureId.FN_GREATER_OR_EQUAL,
          FunctionSignatureId.FN_LESS,
          FunctionSignatureId.FN_LESS_OR_EQUAL,
          FunctionSignatureId.FN_EQUAL,
          FunctionSignatureId.FN_NOT_EQUAL,
          FunctionSignatureId.FN_IS_NULL,
          FunctionSignatureId.FN_IS_TRUE,
          FunctionSignatureId.FN_IS_FALSE,
          FunctionSignatureId.FN_STARTS_WITH_STRING,
          FunctionSignatureId.FN_SUBSTR_STRING,
          FunctionSignatureId.FN_TRIM_STRING,
          FunctionSignatureId.FN_LTRIM_STRING,
          FunctionSignatureId.FN_RTRIM_STRING,
          FunctionSignatureId.FN_REPLACE_STRING,
          FunctionSignatureId.FN_CONCAT_STRING,
          FunctionSignatureId.FN_COUNT_STAR,
          FunctionSignatureId.FN_COUNT,
          FunctionSignatureId.FN_MAX,
          FunctionSignatureId.FN_MIN,
          FunctionSignatureId.FN_AVG_DOUBLE,
          FunctionSignatureId.FN_AVG_INT64,
          FunctionSignatureId.FN_AVG_NUMERIC,
          FunctionSignatureId.FN_SUM_DOUBLE,
          FunctionSignatureId.FN_SUM_INT64,
          FunctionSignatureId.FN_SUM_NUMERIC,
          FunctionSignatureId.FN_MOD_INT64,
          FunctionSignatureId.FN_MOD_NUMERIC,
          FunctionSignatureId.FN_CASE_NO_VALUE,
          FunctionSignatureId.FN_CASE_WITH_VALUE,
          FunctionSignatureId.FN_FLOOR_DOUBLE,
          FunctionSignatureId.FN_FLOOR_NUMERIC,
          FunctionSignatureId.FN_CEIL_DOUBLE,
          FunctionSignatureId.FN_CEIL_NUMERIC,
          FunctionSignatureId.FN_REVERSE_STRING,
          FunctionSignatureId.FN_CHAR_LENGTH_STRING,
          FunctionSignatureId.FN_ENDS_WITH_STRING,
          FunctionSignatureId.FN_STRING_LIKE,
          FunctionSignatureId.FN_BYTE_LIKE,
          FunctionSignatureId.FN_COALESCE,
          FunctionSignatureId.FN_IF,
          FunctionSignatureId.FN_IFNULL,
          FunctionSignatureId.FN_NULLIF,

          // Date functions
          FunctionSignatureId.FN_CURRENT_DATE, // current_date
          FunctionSignatureId.FN_EXTRACT_FROM_DATE, // $extract
          FunctionSignatureId.FN_DATE_FROM_YEAR_MONTH_DAY, // date
          FunctionSignatureId.FN_DATE_FROM_TIMESTAMP, // date
          // FunctionSignatureId.FN_DATE_FROM_DATETIME, // date
          FunctionSignatureId.FN_DATE_ADD_DATE, // date_add
          FunctionSignatureId.FN_DATE_SUB_DATE, // date_sub
          FunctionSignatureId.FN_DATE_DIFF_DATE, // date_diff
          FunctionSignatureId.FN_DATE_TRUNC_DATE, // date_trunc
          FunctionSignatureId.FN_FORMAT_DATE, // format_date
          FunctionSignatureId.FN_PARSE_DATE, // parse_date
          FunctionSignatureId.FN_UNIX_DATE, // unix_date
          FunctionSignatureId.FN_DATE_FROM_UNIX_DATE, // date_from_unix_date

          // Time functions
          FunctionSignatureId.FN_CURRENT_TIME, // current_time
          FunctionSignatureId.FN_EXTRACT_FROM_TIME, // $extract
          FunctionSignatureId.FN_TIME_FROM_HOUR_MINUTE_SECOND, // time
          FunctionSignatureId.FN_TIME_FROM_TIMESTAMP, // time
          // FunctionSignatureId.FN_TIME_FROM_DATETIME, // time
          FunctionSignatureId.FN_TIME_ADD, // time_add
          FunctionSignatureId.FN_TIME_SUB, // time_sub
          FunctionSignatureId.FN_TIME_DIFF, // time_diff
          FunctionSignatureId.FN_TIME_TRUNC, // time_trunc
          FunctionSignatureId.FN_FORMAT_TIME, // format_time
          FunctionSignatureId.FN_PARSE_TIME, // parse_time

          // Timestamp functions
          FunctionSignatureId.FN_CURRENT_TIMESTAMP, // current_timestamp
          FunctionSignatureId.FN_EXTRACT_FROM_TIMESTAMP, // $extract
          FunctionSignatureId.FN_STRING_FROM_TIMESTAMP, // string
          FunctionSignatureId.FN_TIMESTAMP_FROM_STRING, // timestamp
          FunctionSignatureId.FN_TIMESTAMP_FROM_DATE, // timestamp
          // FunctionSignatureId.FN_TIMESTAMP_FROM_DATETIME, // timestamp
          FunctionSignatureId.FN_TIMESTAMP_ADD, // timestamp_add
          FunctionSignatureId.FN_TIMESTAMP_SUB, // timestamp_sub
          FunctionSignatureId.FN_TIMESTAMP_DIFF, // timestamp_diff
          FunctionSignatureId.FN_TIMESTAMP_TRUNC, // timestamp_trunc
          FunctionSignatureId.FN_FORMAT_TIMESTAMP, // format_timestamp
          FunctionSignatureId.FN_PARSE_TIMESTAMP, // parse_timestamp
          FunctionSignatureId.FN_TIMESTAMP_FROM_INT64_SECONDS, // timestamp_seconds
          FunctionSignatureId.FN_TIMESTAMP_FROM_INT64_MILLIS, // timestamp_millis
          // FunctionSignatureId.FN_TIMESTAMP_FROM_INT64_MICROS, // timestamp_micros
          FunctionSignatureId.FN_UNIX_SECONDS_FROM_TIMESTAMP, // unix_seconds
          FunctionSignatureId.FN_UNIX_MILLIS_FROM_TIMESTAMP, // unix_millis
          // FunctionSignatureId.FN_UNIX_MICROS_FROM_TIMESTAMP, // unix_micros
          FunctionSignatureId.FN_TIMESTAMP_FROM_UNIX_SECONDS_INT64, // timestamp_from_unix_seconds
          FunctionSignatureId.FN_TIMESTAMP_FROM_UNIX_MILLIS_INT64, // timestamp_from_unix_millis
          // FunctionSignatureId.FN_TIMESTAMP_FROM_UNIX_MICROS_INT64, // timestamp_from_unix_micros

          // Datetime functions
          FunctionSignatureId.FN_EXTRACT_FROM_DATETIME);

  // todo: Some of operators defined here are later overridden in ZetaSQLPlannerImpl.
  // We should remove them from this table and add generic way to provide custom
  // implementation. (Ex.: timestamp_add)
  public static final Map<String, SqlOperator> ZETASQL_FUNCTION_TO_CALCITE_SQL_OPERATOR =
      ImmutableMap.<String, SqlOperator>builder()
          // grouped window function
          .put("TUMBLE", SqlStdOperatorTable.TUMBLE)
          .put("HOP", SqlStdOperatorTable.HOP)
          .put("SESSION", SqlStdOperatorTable.SESSION)

          // built-in logical operator
          .put("$and", SqlStdOperatorTable.AND)
          .put("$or", SqlStdOperatorTable.OR)
          .put("$not", SqlStdOperatorTable.NOT)

          // built-in comparison operator
          .put("$equal", SqlStdOperatorTable.EQUALS)
          .put("$not_equal", SqlStdOperatorTable.NOT_EQUALS)
          .put("$greater", SqlStdOperatorTable.GREATER_THAN)
          .put("$greater_or_equal", SqlStdOperatorTable.GREATER_THAN_OR_EQUAL)
          .put("$less", SqlStdOperatorTable.LESS_THAN)
          .put("$less_or_equal", SqlStdOperatorTable.LESS_THAN_OR_EQUAL)
          .put("$like", SqlOperators.LIKE)
          // .put("$in", SqlStdOperatorTable.IN)
          // .put("$between", SqlStdOperatorTable.BETWEEN)
          .put("$is_null", SqlStdOperatorTable.IS_NULL)
          .put("$is_true", SqlStdOperatorTable.IS_TRUE)
          .put("$is_false", SqlStdOperatorTable.IS_FALSE)

          // +, -, *, /
          .put("$add", SqlStdOperatorTable.PLUS)
          .put("$subtract", SqlStdOperatorTable.MINUS)
          .put("$multiply", SqlStdOperatorTable.MULTIPLY)
          .put("$unary_minus", SqlStdOperatorTable.UNARY_MINUS)
          .put("$divide", SqlStdOperatorTable.DIVIDE)

          // built-in string function
          .put("concat", SqlOperators.CONCAT)
          // .put("lower", SqlStdOperatorTable.LOWER)
          // .put("upper", SqlStdOperatorTable.UPPER)
          .put("substr", SqlOperators.SUBSTR)
          .put("trim", SqlOperators.TRIM)
          .put("replace", SqlOperators.REPLACE)
          .put("char_length", SqlOperators.CHAR_LENGTH)

          // string function UDFs
          // .put("strpos", )
          // .put("length", )
          // tells Calcite codegen that starts_with function is a udf.
          .put("starts_with", SqlOperators.START_WITHS)
          .put("ends_with", SqlOperators.ENDS_WITH)
          .put("ltrim", SqlOperators.LTRIM)
          .put("rtrim", SqlOperators.RTRIM)
          // .put("regexp_match",)
          // .put("regexp_extract",)
          // .put("regexp_replace",)
          // .put("regexp_extract_all",)
          // .put("byte_length",)
          // .put("format",)
          // .put("split",)
          // .put("regexp_contains", )
          // .put("normalize",)
          // .put("to_base32",)
          // .put("to_base64",)
          // .put("to_hex",)
          // .put("from_base64",)
          // .put("from_base32",)
          // .put("from_hex",)
          // .put("to_code_points")
          // .put("code_points_to_string")
          // .put("lpad", )
          // .put("rpad", )
          // .put("repeat", )
          .put("reverse", SqlOperators.REVERSE)

          // built-in aggregate function
          .put("$count_star", SqlStdOperatorTable.COUNT)
          // TODO: add support to all aggregate functions.
          .put("max", SqlStdOperatorTable.MAX)
          .put("min", SqlStdOperatorTable.MIN)
          .put("avg", SqlStdOperatorTable.AVG)
          .put("sum", SqlStdOperatorTable.SUM)
          .put("any_value", SqlStdOperatorTable.ANY_VALUE)
          .put("count", SqlStdOperatorTable.COUNT)

          // .put("bit_and", SqlStdOperatorTable.BIT_AND) //JIRA link:
          // https://issues.apache.org/jira/browse/BEAM-10379

          // aggregate UDF
          // .put("array_agg", )
          // .put("array_concat_agg")
          .put("string_agg", SqlOperators.STRING_AGG_STRING_FN) // NULL values not supported
          // .put("bit_xor")
          // .put("logical_and")
          // .put("logical_or")
          .put("bit_or", SqlStdOperatorTable.BIT_OR)
          // built-in statistical aggregate function
          // .put("covar_pop", SqlStdOperatorTable.COVAR_POP)
          // .put("covar_samp", SqlStdOperatorTable.COVAR_SAMP)
          // .put("stddev_pop", SqlStdOperatorTable.STDDEV_POP)
          // .put("stddev_samp", SqlStdOperatorTable.STDDEV_SAMP)
          // .put("var_pop", SqlStdOperatorTable.VAR_POP)
          // .put("var_samp", SqlStdOperatorTable.VAR_SAMP)

          // statistical aggregate UDF
          // .put("corr", )

          // built-in approximate aggregate function
          // .put("approx_count_distinct", SqlStdOperatorTable.APPROX_COUNT_DISTINCT)

          // approximate aggregate UDF
          // .put("approx_quantiles", )
          // .put("approx_top_sum")

          // HLL++ UDF
          // hll_count.merge
          // hll_count.extract
          // hll_count.init
          // hll_count.merge_partial

          // CAST
          // CAST operator does not go through lookup table.
          // .put("cast", SqlStdOperatorTable.CAST)

          // built-in math functions
          // .put("math", SqlStdOperatorTable.ABS)
          // .put("sign", SqlStdOperatorTable.SIGN)
          // .put("round", SqlStdOperatorTable.ROUND)
          .put("ceil", SqlStdOperatorTable.CEIL)
          .put("floor", SqlStdOperatorTable.FLOOR)
          .put("mod", SqlStdOperatorTable.MOD)
          // .put("sqrt", SqlStdOperatorTable.SQRT)
          // .put("exp", SqlStdOperatorTable.EXP)
          // .put("ln and log", SqlStdOperatorTable.LN)
          // .put("log10", SqlStdOperatorTable.LOG10)
          // .put("cos", SqlStdOperatorTable.COS)
          // .put("acos", SqlStdOperatorTable.ACOS)
          // .put("sin", SqlStdOperatorTable.SIN)
          // .put("asin", SqlStdOperatorTable.ASIN)
          // .put("tan", SqlStdOperatorTable.TAN)
          // .put("atan", SqlStdOperatorTable.ATAN)
          // .put("atan2", SqlStdOperatorTable.ATAN2)
          // .put("abs", SqlStdOperatorTable.ABS)
          // .put("pow", SqlStdOperatorTable.POWER)
          // .put("div", SqlStdOperatorTable.DIVIDE)
          // .put("trunc", SqlStdOperatorTable.TRUNCATE)

          // math UDF
          // .put("is_inf",)
          // .put("is_nan",)
          // .put("ieee_divide")
          // .put("safe_add")
          // .put("safe_divide")
          // .put("safe_subtract")
          // .put("safe_multiply")
          // .put("safe_negate")
          // .put("greatest")
          // .put("least")
          // .put("log")
          // .put("cosh")
          // .put("acosh")
          // .put("sinh")
          // .put("asinh")
          // .put("tanh")
          // .put("atanh")

          // Analytic functions
          // .put("dense_rank", SqlStdOperatorTable.DENSE_RANK)
          // .put("rank", SqlStdOperatorTable.RANK)
          // .put("row_number", SqlStdOperatorTable.ROW_NUMBER)
          // .put("percent_rank", SqlStdOperatorTable.PERCENT_RANK)
          // .put("cume_dist", SqlStdOperatorTable.CUME_DIST)
          // .put("ntile", SqlStdOperatorTable.NTILE)
          // .put("lead", SqlStdOperatorTable.LEAD)
          // .put("lag", SqlStdOperatorTable.LAG)
          // .put("first_value", SqlStdOperatorTable.FIRST_VALUE)
          // .put("last_value", SqlStdOperatorTable.LAST_VALUE)
          // .put("nth_value", SqlStdOperatorTable.NTH_VALUE)

          // .put("percentile_cont", )
          // .put("percentile_disc",)

          // misc functions
          // .put("fingerprint")
          // .put("fingerprint2011")

          // hash functions
          // .put("md5")
          // .put("sha1")
          // .put("sha256")
          // .put("sha512")

          // timestamp functions
          .put("$extract", SqlStdOperatorTable.EXTRACT)
          .put("timestamp", SqlOperators.TIMESTAMP_OP)

          // other functions
          // .put("session_user", SqlStdOperatorTable.SESSION_USER)
          // .put("bit_cast_to_int32")
          // .put("bit_cast_to_int64")
          // .put("bit_cast_to_uint32")
          // .put("bit_cast_to_uint64")
          // .put("countif", )

          // case operator
          .put("$case_no_value", SqlStdOperatorTable.CASE)

          // if operator - IF(cond, pos, neg) can actually be mapped directly to `CASE WHEN cond
          // THEN pos ELSE neg`
          .put("if", SqlStdOperatorTable.CASE)

          // $case_no_value specializations
          // all of these operators can have their operands adjusted to achieve the same thing with
          // a call to $case_with_value
          .put("$case_with_value", SqlStdOperatorTable.CASE)
          .put("coalesce", SqlStdOperatorTable.CASE)
          .put("ifnull", SqlStdOperatorTable.CASE)
          .put("nullif", SqlStdOperatorTable.CASE)
          .build();

  public static final Map<String, SqlOperatorRewriter>
      ZETASQL_FUNCTION_TO_CALCITE_SQL_OPERATOR_REWRITER =
          ImmutableMap.<String, SqlOperatorRewriter>builder()
              .put("$case_with_value", new SqlCaseWithValueOperatorRewriter())
              .put("coalesce", new SqlCoalesceOperatorRewriter())
              .put("ifnull", new SqlIfNullOperatorRewriter())
              .put("nullif", new SqlNullIfOperatorRewriter())
              .put("$extract", new SqlExtractTimestampOperatorRewriter())
              .build();
}
