/*
 * 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.flink.connector.jdbc.databases.postgres.catalog;

import org.apache.flink.connector.jdbc.databases.postgres.PostgresTestBase;
import org.apache.flink.connector.jdbc.testutils.DatabaseMetadata;
import org.apache.flink.connector.jdbc.testutils.JdbcITCaseBase;
import org.apache.flink.connector.jdbc.testutils.databases.postgres.PostgresDatabase;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.types.logical.DecimalType;

import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;

/** Test base for {@link PostgresCatalog}. */
class PostgresCatalogTestBase implements JdbcITCaseBase, PostgresTestBase {

    private static DatabaseMetadata getStaticMetadata() {
        return PostgresDatabase.getMetadata();
    }

    protected static final String TEST_CATALOG_NAME = "mypg";
    protected static final String TEST_USERNAME = getStaticMetadata().getUsername();
    protected static final String TEST_PWD = getStaticMetadata().getPassword();
    protected static final String TEST_DB = "test";
    protected static final String TEST_SCHEMA = "test_schema";
    protected static final String TABLE1 = "t1";
    protected static final String TABLE2 = "t2";
    protected static final String TABLE3 = "t3";
    protected static final String TABLE4 = "t4";
    protected static final String TABLE5 = "t5";
    protected static final String TABLE_PRIMITIVE_TYPE = "primitive_table";
    protected static final String TABLE_PRIMITIVE_TYPE2 = "primitive_table2";
    protected static final String TABLE_ARRAY_TYPE = "array_table";
    protected static final String TABLE_SERIAL_TYPE = "serial_table";

    protected static String baseUrl;
    protected static PostgresCatalog catalog;

    @BeforeAll
    static void init() throws SQLException {
        // jdbc:postgresql://localhost:50807/postgres?user=postgres
        String jdbcUrl = getStaticMetadata().getJdbcUrl();
        // jdbc:postgresql://localhost:50807/
        baseUrl = jdbcUrl.substring(0, jdbcUrl.lastIndexOf("/"));

        catalog =
                new PostgresCatalog(
                        Thread.currentThread().getContextClassLoader(),
                        TEST_CATALOG_NAME,
                        PostgresCatalog.DEFAULT_DATABASE,
                        TEST_USERNAME,
                        TEST_PWD,
                        baseUrl);

        // create test database and schema
        createSchema(TEST_DB, TEST_SCHEMA);

        // create test tables
        // table: postgres.public.t1
        // table: postgres.public.t4
        // table: postgres.public.t5
        createTable(PostgresTablePath.fromFlinkTableName(TABLE1), getSimpleTable().pgSchemaSql);
        createTable(PostgresTablePath.fromFlinkTableName(TABLE4), getSimpleTable().pgSchemaSql);
        createTable(PostgresTablePath.fromFlinkTableName(TABLE5), getSimpleTable().pgSchemaSql);

        // table: test.public.t2
        // table: test.test_schema.t3
        // table: postgres.public.dt
        // table: postgres.public.dt2
        createTable(
                TEST_DB,
                PostgresTablePath.fromFlinkTableName(TABLE2),
                getSimpleTable().pgSchemaSql);
        createTable(
                TEST_DB, new PostgresTablePath(TEST_SCHEMA, TABLE3), getSimpleTable().pgSchemaSql);
        createTable(
                PostgresTablePath.fromFlinkTableName(TABLE_PRIMITIVE_TYPE),
                getPrimitiveTable().pgSchemaSql);
        createTable(
                PostgresTablePath.fromFlinkTableName(TABLE_PRIMITIVE_TYPE2),
                getPrimitiveTable("test_pk2").pgSchemaSql);
        createTable(
                PostgresTablePath.fromFlinkTableName(TABLE_ARRAY_TYPE),
                getArrayTable().pgSchemaSql);
        createTable(
                PostgresTablePath.fromFlinkTableName(TABLE_SERIAL_TYPE),
                getSerialTable().pgSchemaSql);

        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format(
                        "insert into public.%s values (%s);", TABLE1, getSimpleTable().values));
        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format(
                        "insert into %s values (%s);",
                        TABLE_PRIMITIVE_TYPE, getPrimitiveTable().values));
        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format(
                        "insert into %s values (%s);", TABLE_ARRAY_TYPE, getArrayTable().values));
        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format(
                        "insert into %s values (%s);", TABLE_SERIAL_TYPE, getSerialTable().values));
    }

    @AfterAll
    static void afterAll() throws SQLException {
        executeSQL(TEST_DB, String.format("DROP SCHEMA %s CASCADE", TEST_SCHEMA));
        executeSQL(
                TEST_DB,
                String.format("DROP TABLE %s ", PostgresTablePath.fromFlinkTableName(TABLE2)));

        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format("DROP TABLE %s ", PostgresTablePath.fromFlinkTableName(TABLE1)));
        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format("DROP TABLE %s ", PostgresTablePath.fromFlinkTableName(TABLE4)));
        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format("DROP TABLE %s ", PostgresTablePath.fromFlinkTableName(TABLE5)));
        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format(
                        "DROP TABLE %s ",
                        PostgresTablePath.fromFlinkTableName(TABLE_PRIMITIVE_TYPE)));
        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format(
                        "DROP TABLE %s ",
                        PostgresTablePath.fromFlinkTableName(TABLE_PRIMITIVE_TYPE2)));
        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format(
                        "DROP TABLE %s ", PostgresTablePath.fromFlinkTableName(TABLE_ARRAY_TYPE)));
        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format(
                        "DROP TABLE %s ", PostgresTablePath.fromFlinkTableName(TABLE_SERIAL_TYPE)));
    }

    public static void createTable(PostgresTablePath tablePath, String tableSchemaSql)
            throws SQLException {
        executeSQL(
                PostgresCatalog.DEFAULT_DATABASE,
                String.format("CREATE TABLE %s(%s);", tablePath.getFullPath(), tableSchemaSql));
    }

    public static void createTable(String db, PostgresTablePath tablePath, String tableSchemaSql)
            throws SQLException {
        executeSQL(
                db, String.format("CREATE TABLE %s(%s);", tablePath.getFullPath(), tableSchemaSql));
    }

    public static void createSchema(String db, String schema) throws SQLException {
        executeSQL(db, String.format("CREATE SCHEMA %s", schema));
    }

    public static void createDatabase(String database) throws SQLException {
        executeSQL(String.format("CREATE DATABASE %s;", database));
    }

    public static void executeSQL(String sql) throws SQLException {
        executeSQL("", sql);
    }

    public static void executeSQL(String db, String sql) throws SQLException {
        try (Connection conn =
                        DriverManager.getConnection(
                                String.format("%s/%s", baseUrl, db), TEST_USERNAME, TEST_PWD);
                Statement statement = conn.createStatement()) {
            statement.executeUpdate(sql);
        } catch (SQLException e) {
            throw e;
        }
    }

    /** Object holding schema and corresponding sql. */
    public static class TestTable {
        Schema schema;
        String pgSchemaSql;
        String values;

        public TestTable(Schema schema, String pgSchemaSql, String values) {
            this.schema = schema;
            this.pgSchemaSql = pgSchemaSql;
            this.values = values;
        }
    }

    public static TestTable getSimpleTable() {
        return new TestTable(
                Schema.newBuilder().column("id", DataTypes.INT()).build(), "id integer", "1");
    }

    // posgres doesn't support to use the same primary key name across different tables,
    // make the table parameterized to resolve this problem.
    public static TestTable getPrimitiveTable() {
        return getPrimitiveTable("test_pk");
    }

    // TODO: add back timestamptz and time types.
    //  Flink currently doesn't support converting time's precision, with the following error
    //  TableException: Unsupported conversion from data type 'TIME(6)' (conversion class:
    // java.sql.Time)
    //  to type information. Only data types that originated from type information fully support a
    // reverse conversion.
    public static TestTable getPrimitiveTable(String primaryKeyName) {
        return new TestTable(
                Schema.newBuilder()
                        .column("int", DataTypes.INT().notNull())
                        .column("bytea", DataTypes.BYTES())
                        .column("short", DataTypes.SMALLINT().notNull())
                        .column("long", DataTypes.BIGINT())
                        .column("real", DataTypes.FLOAT())
                        .column("double_precision", DataTypes.DOUBLE())
                        .column("numeric", DataTypes.DECIMAL(10, 5))
                        .column("decimal", DataTypes.DECIMAL(10, 1))
                        .column("boolean", DataTypes.BOOLEAN())
                        .column("text", DataTypes.STRING())
                        .column("char", DataTypes.CHAR(1))
                        .column("character", DataTypes.CHAR(3))
                        .column("character_varying", DataTypes.VARCHAR(20))
                        .column("timestamp", DataTypes.TIMESTAMP(5))
                        //				.field("timestamptz", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(4))
                        .column("date", DataTypes.DATE())
                        .column("time", DataTypes.TIME(0))
                        .column("default_numeric", DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 18))
                        .primaryKeyNamed(primaryKeyName, "short", "int")
                        .build(),
                "int integer, "
                        + "bytea bytea, "
                        + "short smallint, "
                        + "long bigint, "
                        + "real real, "
                        + "double_precision double precision, "
                        + "numeric numeric(10, 5), "
                        + "decimal decimal(10, 1), "
                        + "boolean boolean, "
                        + "text text, "
                        + "char char, "
                        + "character character(3), "
                        + "character_varying character varying(20), "
                        + "timestamp timestamp(5), "
                        +
                        //				"timestamptz timestamptz(4), " +
                        "date date,"
                        + "time time(0), "
                        + "default_numeric numeric, "
                        + "CONSTRAINT "
                        + primaryKeyName
                        + " PRIMARY KEY (short, int)",
                "1,"
                        + "'2',"
                        + "3,"
                        + "4,"
                        + "5.5,"
                        + "6.6,"
                        + "7.7,"
                        + "8.8,"
                        + "true,"
                        + "'a',"
                        + "'b',"
                        + "'c',"
                        + "'d',"
                        + "'2016-06-22 19:10:25',"
                        +
                        //				"'2006-06-22 19:10:25'," +
                        "'2015-01-01',"
                        + "'00:51:02.746572', "
                        + "500");
    }

    // TODO: add back timestamptz once planner supports timestamp with timezone
    public static TestTable getArrayTable() {
        return new TestTable(
                Schema.newBuilder()
                        .column("int_arr", DataTypes.ARRAY(DataTypes.INT()))
                        .column("bytea_arr", DataTypes.ARRAY(DataTypes.BYTES()))
                        .column("short_arr", DataTypes.ARRAY(DataTypes.SMALLINT()))
                        .column("long_arr", DataTypes.ARRAY(DataTypes.BIGINT()))
                        .column("real_arr", DataTypes.ARRAY(DataTypes.FLOAT()))
                        .column("double_precision_arr", DataTypes.ARRAY(DataTypes.DOUBLE()))
                        .column("numeric_arr", DataTypes.ARRAY(DataTypes.DECIMAL(10, 5)))
                        .column(
                                "numeric_arr_default",
                                DataTypes.ARRAY(DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 18)))
                        .column("decimal_arr", DataTypes.ARRAY(DataTypes.DECIMAL(10, 2)))
                        .column("boolean_arr", DataTypes.ARRAY(DataTypes.BOOLEAN()))
                        .column("text_arr", DataTypes.ARRAY(DataTypes.STRING()))
                        .column("char_arr", DataTypes.ARRAY(DataTypes.CHAR(1)))
                        .column("character_arr", DataTypes.ARRAY(DataTypes.CHAR(3)))
                        .column("character_varying_arr", DataTypes.ARRAY(DataTypes.VARCHAR(20)))
                        .column("timestamp_arr", DataTypes.ARRAY(DataTypes.TIMESTAMP(5)))
                        //				.field("timestamptz_arr",
                        // DataTypes.ARRAY(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(4)))
                        .column("date_arr", DataTypes.ARRAY(DataTypes.DATE()))
                        .column("time_arr", DataTypes.ARRAY(DataTypes.TIME(0)))
                        .column("null_bytea_arr", DataTypes.ARRAY(DataTypes.BYTES()))
                        .column("null_text_arr", DataTypes.ARRAY(DataTypes.STRING()))
                        .build(),
                "int_arr integer[], "
                        + "bytea_arr bytea[], "
                        + "short_arr smallint[], "
                        + "long_arr bigint[], "
                        + "real_arr real[], "
                        + "double_precision_arr double precision[], "
                        + "numeric_arr numeric(10, 5)[], "
                        + "numeric_arr_default numeric[], "
                        + "decimal_arr decimal(10,2)[], "
                        + "boolean_arr boolean[], "
                        + "text_arr text[], "
                        + "char_arr char[], "
                        + "character_arr character(3)[], "
                        + "character_varying_arr character varying(20)[], "
                        + "timestamp_arr timestamp(5)[], "
                        +
                        //				"timestamptz_arr timestamptz(4)[], " +
                        "date_arr date[], "
                        + "time_arr time(0)[], "
                        + "null_bytea_arr bytea[], "
                        + "null_text_arr text[]",
                String.format(
                        "'{1,2,3}',"
                                + "'{2,3,4}',"
                                + "'{3,4,5}',"
                                + "'{4,5,6}',"
                                + "'{5.5,6.6,7.7}',"
                                + "'{6.6,7.7,8.8}',"
                                + "'{7.7,8.8,9.9}',"
                                + "'{8.8,9.9,10.10}',"
                                + "'{9.9,10.10,11.11}',"
                                + "'{true,false,true}',"
                                + "'{a,b,c}',"
                                + "'{b,c,d}',"
                                + "'{b,c,d}',"
                                + "'{b,c,d}',"
                                + "'{\"2016-06-22 19:10:25\", \"2019-06-22 19:10:25\"}',"
                                +
                                //				"'{\"2006-06-22 19:10:25\", \"2009-06-22 19:10:25\"}'," +
                                "'{\"2015-01-01\", \"2020-01-01\"}',"
                                + "'{\"00:51:02.746572\", \"00:59:02.746572\"}',"
                                + "NULL,"
                                + "NULL"));
    }

    public static TestTable getSerialTable() {
        return new TestTable(
                Schema.newBuilder()
                        // serial fields are returned as not null by ResultSetMetaData.columnNoNulls
                        .column("f0", DataTypes.SMALLINT().notNull())
                        .column("f1", DataTypes.INT().notNull())
                        .column("f2", DataTypes.SMALLINT().notNull())
                        .column("f3", DataTypes.INT().notNull())
                        .column("f4", DataTypes.BIGINT().notNull())
                        .column("f5", DataTypes.BIGINT().notNull())
                        .build(),
                "f0 smallserial, "
                        + "f1 serial, "
                        + "f2 serial2, "
                        + "f3 serial4, "
                        + "f4 serial8, "
                        + "f5 bigserial",
                "32767,"
                        + "2147483647,"
                        + "32767,"
                        + "2147483647,"
                        + "9223372036854775807,"
                        + "9223372036854775807");
    }
}