/*
 * 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.iceberg.flink;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assumptions.assumeThat;

import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema.UnresolvedPrimaryKey;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.CatalogMaterializedTable;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.IntervalFreshness;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.ContentFile;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.DataOperations;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;

public class TestFlinkCatalogTable extends CatalogTestBase {

  @Override
  @BeforeEach
  public void before() {
    super.before();
    sql("CREATE DATABASE %s", flinkDatabase);
    sql("USE CATALOG %s", catalogName);
    sql("USE %s", DATABASE);
  }

  @AfterEach
  public void cleanNamespaces() {
    sql("DROP TABLE IF EXISTS %s.tl", flinkDatabase);
    sql("DROP TABLE IF EXISTS %s.tl2", flinkDatabase);
    dropDatabase(flinkDatabase, true);
    super.clean();
  }

  @TestTemplate
  public void testGetTable() {
    sql("CREATE TABLE tl(id BIGINT, strV STRING)");

    Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, "tl"));
    Schema iSchema =
        new Schema(
            Types.NestedField.optional(1, "id", Types.LongType.get()),
            Types.NestedField.optional(2, "strV", Types.StringType.get()));
    assertThat(table.schema().toString())
        .as("Should load the expected iceberg schema")
        .isEqualTo(iSchema.toString());
  }

  @TestTemplate
  public void testRenameTable() {
    assumeThat(isHadoopCatalog).as("HadoopCatalog does not support rename table").isFalse();
    final Schema tableSchema =
        new Schema(Types.NestedField.optional(0, "id", Types.LongType.get()));
    validationCatalog.createTable(TableIdentifier.of(icebergNamespace, "tl"), tableSchema);
    sql("ALTER TABLE tl RENAME TO tl2");

    assertThatThrownBy(() -> getTableEnv().from("tl"))
        .isInstanceOf(ValidationException.class)
        .hasMessage("Table `tl` was not found.");

    Schema actualSchema = FlinkSchemaUtil.convert(getTableEnv().from("tl2").getResolvedSchema());
    assertThat(tableSchema.asStruct()).isEqualTo(actualSchema.asStruct());
  }

  @TestTemplate
  public void testCreateTable() throws TableNotExistException {
    sql("CREATE TABLE tl(id BIGINT)");

    Table table = table("tl");
    assertThat(table.schema().asStruct())
        .isEqualTo(
            new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct());
    CatalogTable catalogTable = catalogTable("tl");
    assertThat(catalogTable.getUnresolvedSchema())
        .isEqualTo(
            org.apache.flink.table.api.Schema.newBuilder()
                .column("id", DataTypes.BIGINT())
                .build());
  }

  @TestTemplate
  public void testCreateTableWithPrimaryKey() throws Exception {
    sql("CREATE TABLE tl(id BIGINT, data STRING, key STRING PRIMARY KEY NOT ENFORCED)");

    Table table = table("tl");
    assertThat(table.schema().identifierFieldIds())
        .as("Should have the expected row key.")
        .isEqualTo(Sets.newHashSet(table.schema().findField("key").fieldId()));
    CatalogTable catalogTable = catalogTable("tl");
    Optional<UnresolvedPrimaryKey> uniqueConstraintOptional =
        catalogTable.getUnresolvedSchema().getPrimaryKey();
    assertThat(uniqueConstraintOptional).isPresent();
    assertThat(uniqueConstraintOptional.get().getColumnNames()).containsExactly("key");
  }

  @TestTemplate
  public void testCreateTableWithMultiColumnsInPrimaryKey() throws Exception {
    sql(
        "CREATE TABLE tl(id BIGINT, data STRING, CONSTRAINT pk_constraint PRIMARY KEY(data, id) NOT ENFORCED)");

    Table table = table("tl");
    assertThat(table.schema().identifierFieldIds())
        .as("Should have the expected RowKey")
        .isEqualTo(
            Sets.newHashSet(
                table.schema().findField("id").fieldId(),
                table.schema().findField("data").fieldId()));
    CatalogTable catalogTable = catalogTable("tl");
    Optional<UnresolvedPrimaryKey> uniqueConstraintOptional =
        catalogTable.getUnresolvedSchema().getPrimaryKey();
    assertThat(uniqueConstraintOptional).isPresent();
    assertThat(uniqueConstraintOptional.get().getColumnNames()).containsExactly("id", "data");
  }

  @TestTemplate
  public void testCreateTableIfNotExists() {
    sql("CREATE TABLE tl(id BIGINT)");

    // Assert that table does exist.
    assertThat(table("tl")).isNotNull();

    sql("DROP TABLE tl");
    assertThatThrownBy(() -> table("tl"))
        .isInstanceOf(NoSuchTableException.class)
        .hasMessage("Table does not exist: " + getFullQualifiedTableName("tl"));

    sql("CREATE TABLE IF NOT EXISTS tl(id BIGINT)");
    assertThat(table("tl").properties()).doesNotContainKey("key");

    table("tl").updateProperties().set("key", "value").commit();
    assertThat(table("tl").properties()).containsEntry("key", "value");

    sql("CREATE TABLE IF NOT EXISTS tl(id BIGINT)");
    assertThat(table("tl").properties()).containsEntry("key", "value");
  }

  @TestTemplate
  public void testCreateTableLike() throws TableNotExistException {
    sql("CREATE TABLE tl(id BIGINT)");
    sql("CREATE TABLE tl2 LIKE tl");

    Table table = table("tl2");
    assertThat(table.schema().asStruct())
        .isEqualTo(
            new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct());
    CatalogTable catalogTable = catalogTable("tl2");
    assertThat(catalogTable.getUnresolvedSchema())
        .isEqualTo(
            org.apache.flink.table.api.Schema.newBuilder()
                .column("id", DataTypes.BIGINT())
                .build());
  }

  @TestTemplate
  public void testCreateTableLikeInDiffIcebergCatalog() throws TableNotExistException {
    sql("CREATE TABLE tl(id BIGINT)");

    String catalog2 = catalogName + "2";
    sql("CREATE CATALOG %s WITH %s", catalog2, toWithClause(config));
    sql("CREATE DATABASE %s", catalog2 + ".testdb");
    sql("CREATE TABLE %s LIKE tl", catalog2 + ".testdb.tl2");

    CatalogTable catalogTable = catalogTable(catalog2, "testdb", "tl2");
    assertThat(catalogTable.getUnresolvedSchema())
        .isEqualTo(
            org.apache.flink.table.api.Schema.newBuilder()
                .column("id", DataTypes.BIGINT())
                .build());

    dropCatalog(catalog2, true);
  }

  @TestTemplate
  public void testCreateTableLikeInFlinkCatalog() throws TableNotExistException {
    sql("CREATE TABLE tl(id BIGINT)");

    sql("CREATE TABLE `default_catalog`.`default_database`.tl2 LIKE tl");

    CatalogTable catalogTable = catalogTable("default_catalog", "default_database", "tl2");
    assertThat(catalogTable.getUnresolvedSchema())
        .isEqualTo(
            org.apache.flink.table.api.Schema.newBuilder()
                .column("id", DataTypes.BIGINT())
                .build());

    String srcCatalogProps = FlinkCreateTableOptions.toJson(catalogName, DATABASE, "tl");
    Map<String, String> options = catalogTable.getOptions();
    assertThat(options)
        .containsEntry(
            FlinkCreateTableOptions.CONNECTOR_PROPS_KEY,
            FlinkDynamicTableFactory.FACTORY_IDENTIFIER)
        .containsEntry(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY, srcCatalogProps);
    assertThat(options.get(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY))
        .doesNotContain("extra-catalog-prop", "extra-value");
  }

  @TestTemplate
  public void testCreateTableWithTableComment() {
    // create table with comment
    sql("CREATE TABLE tl(id BIGINT) COMMENT 'table comment'");
    assertThat(table("tl").properties()).containsEntry(TableProperties.COMMENT, "table comment");
  }

  @TestTemplate
  public void testAlterTableModifyTableComment() {
    // create table with comment
    sql("CREATE TABLE tl(id BIGINT) COMMENT 'table comment'");
    assertThat(table("tl").properties()).containsEntry(TableProperties.COMMENT, "table comment");

    // alter table comment
    sql("ALTER TABLE tl SET('comment' = 'new comment')");
    assertThat(table("tl").properties()).containsEntry(TableProperties.COMMENT, "new comment");
  }

  @TestTemplate
  public void testCreateTableLocation() {
    assumeThat(isHadoopCatalog)
        .as("HadoopCatalog does not support creating table with location")
        .isFalse();
    sql("CREATE TABLE tl(id BIGINT) WITH ('location'='file:///tmp/location')");

    Table table = table("tl");
    assertThat(table.schema().asStruct())
        .isEqualTo(
            new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct());
    assertThat(table.location()).isEqualTo("file:///tmp/location");
  }

  @TestTemplate
  public void testCreatePartitionTable() throws TableNotExistException {
    sql("CREATE TABLE tl(id BIGINT, dt STRING) PARTITIONED BY(dt)");

    Table table = table("tl");
    assertThat(table.schema().asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()))
                .asStruct());
    assertThat(table.spec())
        .isEqualTo(PartitionSpec.builderFor(table.schema()).identity("dt").build());
    CatalogTable catalogTable = catalogTable("tl");
    assertThat(catalogTable.getUnresolvedSchema())
        .isEqualTo(
            org.apache.flink.table.api.Schema.newBuilder()
                .column("id", DataTypes.BIGINT())
                .column("dt", DataTypes.STRING())
                .build());
    assertThat(catalogTable.getPartitionKeys()).isEqualTo(Collections.singletonList("dt"));
  }

  @TestTemplate
  public void testCreateTableWithColumnComment() {
    sql("CREATE TABLE tl(id BIGINT COMMENT 'comment - id', data STRING COMMENT 'comment - data')");

    Table table = table("tl");
    assertThat(table.schema().asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get(), "comment - id"),
                    Types.NestedField.optional(2, "data", Types.StringType.get(), "comment - data"))
                .asStruct());
  }

  @TestTemplate
  public void testCreateTableWithFormatV2ThroughTableProperty() throws Exception {
    sql("CREATE TABLE tl(id BIGINT) WITH ('format-version'='2')");

    Table table = table("tl");
    assertThat(((BaseTable) table).operations().current().formatVersion()).isEqualTo(2);
  }

  @TestTemplate
  public void testUpgradeTableWithFormatV2ThroughTableProperty() throws Exception {
    sql("CREATE TABLE tl(id BIGINT) WITH ('format-version'='1')");

    Table table = table("tl");
    TableOperations ops = ((BaseTable) table).operations();
    assertThat(ops.refresh().formatVersion())
        .as("should create table using format v1")
        .isEqualTo(1);
    sql("ALTER TABLE tl SET('format-version'='2')");
    assertThat(ops.refresh().formatVersion())
        .as("should update table to use format v2")
        .isEqualTo(2);
  }

  @TestTemplate
  public void testDowngradeTableToFormatV1ThroughTablePropertyFails() throws Exception {
    sql("CREATE TABLE tl(id BIGINT) WITH ('format-version'='2')");

    Table table = table("tl");
    TableOperations ops = ((BaseTable) table).operations();
    assertThat(ops.refresh().formatVersion())
        .as("should create table using format v2")
        .isEqualTo(2);
    assertThatThrownBy(() -> sql("ALTER TABLE tl SET('format-version'='1')"))
        .rootCause()
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessage("Cannot downgrade v2 table to v1");
  }

  @TestTemplate
  public void testLoadTransformPartitionTable() throws TableNotExistException {
    Schema schema = new Schema(Types.NestedField.optional(0, "id", Types.LongType.get()));
    validationCatalog.createTable(
        TableIdentifier.of(icebergNamespace, "tl"),
        schema,
        PartitionSpec.builderFor(schema).bucket("id", 100).build());

    CatalogTable catalogTable = catalogTable("tl");
    assertThat(catalogTable.getUnresolvedSchema())
        .isEqualTo(
            org.apache.flink.table.api.Schema.newBuilder()
                .column("id", DataTypes.BIGINT())
                .build());
    assertThat(catalogTable.getPartitionKeys()).isEmpty();
  }

  @TestTemplate
  public void testAlterTableProperties() throws TableNotExistException {
    sql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
    Map<String, String> properties = Maps.newHashMap();
    properties.put("oldK", "oldV");

    // new
    sql("ALTER TABLE tl SET('newK'='newV')");
    properties.put("newK", "newV");
    assertThat(table("tl").properties()).containsAllEntriesOf(properties);

    // update old
    sql("ALTER TABLE tl SET('oldK'='oldV2')");
    properties.put("oldK", "oldV2");
    assertThat(table("tl").properties()).containsAllEntriesOf(properties);

    // remove property
    sql("ALTER TABLE tl RESET('oldK')");
    properties.remove("oldK");
    assertThat(table("tl").properties()).containsAllEntriesOf(properties);
  }

  @TestTemplate
  public void testAlterTableAddColumn() {
    sql("CREATE TABLE tl(id BIGINT)");
    Schema schemaBefore = table("tl").schema();
    assertThat(schemaBefore.asStruct())
        .isEqualTo(
            new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct());
    sql("ALTER TABLE tl ADD (dt STRING)");
    Schema schemaAfter1 = table("tl").schema();
    assertThat(schemaAfter1.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()))
                .asStruct());
    // Add multiple columns
    sql("ALTER TABLE tl ADD (col1 STRING COMMENT 'comment for col1', col2 BIGINT)");
    Schema schemaAfter2 = table("tl").schema();
    assertThat(schemaAfter2.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()),
                    Types.NestedField.optional(
                        3, "col1", Types.StringType.get(), "comment for col1"),
                    Types.NestedField.optional(4, "col2", Types.LongType.get()))
                .asStruct());

    // Adding an existing field should fail due to Flink's internal validation.
    assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
        .isInstanceOf(ValidationException.class)
        .hasMessageContaining("Try to add a column `id` which already exists in the table.");
  }

  @TestTemplate
  public void testAlterTableAddColumnPosition() {
    sql("CREATE TABLE tl(id BIGINT, name STRING)");
    Schema schemaBefore = table("tl").schema();
    assertThat(schemaBefore.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "name", Types.StringType.get()))
                .asStruct());

    sql("ALTER TABLE tl ADD (col1 STRING FIRST)");
    sql("ALTER TABLE tl ADD (col2 INT AFTER id)");

    Schema schemaAfter = table("tl").schema();
    assertThat(schemaAfter.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(3, "col1", Types.StringType.get()),
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(4, "col2", Types.IntegerType.get()),
                    Types.NestedField.optional(2, "name", Types.StringType.get()))
                .asStruct());
  }

  @TestTemplate
  public void testAlterTableDropColumn() {
    sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
    Schema schemaBefore = table("tl").schema();
    assertThat(schemaBefore.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()),
                    Types.NestedField.optional(3, "col1", Types.StringType.get()),
                    Types.NestedField.optional(4, "col2", Types.LongType.get()))
                .asStruct());
    sql("ALTER TABLE tl DROP (dt)");
    Schema schemaAfter1 = table("tl").schema();
    assertThat(schemaAfter1.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(3, "col1", Types.StringType.get()),
                    Types.NestedField.optional(4, "col2", Types.LongType.get()))
                .asStruct());
    // Drop multiple columns
    sql("ALTER TABLE tl DROP (col1, col2)");
    Schema schemaAfter2 = table("tl").schema();
    assertThat(schemaAfter2.asStruct())
        .isEqualTo(
            new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct());
    // Dropping an non-existing field should fail due to Flink's internal validation.
    assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
        .isInstanceOf(ValidationException.class)
        .hasMessageContaining("The column `foo` does not exist in the base table.");

    // Dropping an already-deleted field should fail due to Flink's internal validation.
    assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (dt)"))
        .isInstanceOf(ValidationException.class)
        .hasMessageContaining("The column `dt` does not exist in the base table.");
  }

  @TestTemplate
  public void testAlterTableModifyColumnName() {
    sql("CREATE TABLE tl(id BIGINT, dt STRING)");
    Schema schemaBefore = table("tl").schema();
    assertThat(schemaBefore.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()))
                .asStruct());
    sql("ALTER TABLE tl RENAME dt TO data");
    Schema schemaAfter = table("tl").schema();
    assertThat(schemaAfter.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "data", Types.StringType.get()))
                .asStruct());
  }

  @TestTemplate
  public void testAlterTableModifyColumnType() {
    sql("CREATE TABLE tl(id INTEGER, dt STRING)");
    Schema schemaBefore = table("tl").schema();
    assertThat(schemaBefore.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.IntegerType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()))
                .asStruct());
    // Promote type from Integer to Long
    sql("ALTER TABLE tl MODIFY (id BIGINT)");
    Schema schemaAfter = table("tl").schema();
    assertThat(schemaAfter.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()))
                .asStruct());
    // Type change that doesn't follow the type-promotion rule should fail due to Iceberg's
    // validation.
    assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt INTEGER)"))
        .isInstanceOf(TableException.class)
        .hasMessageContaining("Could not execute AlterTable")
        .hasRootCauseInstanceOf(IllegalArgumentException.class)
        .hasRootCauseMessage("Cannot change column type: dt: string -> int");
  }

  @TestTemplate
  public void testAlterTableModifyColumnNullability() {
    sql("CREATE TABLE tl(id INTEGER NOT NULL, dt STRING)");
    Schema schemaBefore = table("tl").schema();
    assertThat(schemaBefore.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.required(1, "id", Types.IntegerType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()))
                .asStruct());

    // Set nullability from required to optional
    sql("ALTER TABLE tl MODIFY (id INTEGER)");
    Schema schemaAfter = table("tl").schema();
    assertThat(schemaAfter.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.IntegerType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()))
                .asStruct());
  }

  @TestTemplate
  public void testAlterTableModifyColumnPosition() {
    sql("CREATE TABLE tl(id BIGINT, dt STRING)");
    Schema schemaBefore = table("tl").schema();
    assertThat(schemaBefore.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()))
                .asStruct());

    sql("ALTER TABLE tl MODIFY (dt STRING FIRST)");
    Schema schemaAfter = table("tl").schema();
    assertThat(schemaAfter.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(2, "dt", Types.StringType.get()),
                    Types.NestedField.optional(1, "id", Types.LongType.get()))
                .asStruct());

    sql("ALTER TABLE tl MODIFY (dt STRING AFTER id)");
    Schema schemaAfterAfter = table("tl").schema();
    assertThat(schemaAfterAfter.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()))
                .asStruct());
    // Modifying the position of a non-existing column should fail due to Flink's internal
    // validation.
    assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (non_existing STRING FIRST)"))
        .isInstanceOf(ValidationException.class)
        .hasMessageContaining(
            "Try to modify a column `non_existing` which does not exist in the table.");

    // Moving a column after a non-existing column should fail due to Flink's internal validation.
    assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt STRING AFTER non_existing)"))
        .isInstanceOf(ValidationException.class)
        .hasMessageContaining(
            "Referenced column `non_existing` by 'AFTER' does not exist in the table.");
  }

  @TestTemplate
  public void testAlterTableModifyColumnComment() {
    sql("CREATE TABLE tl(id BIGINT, dt STRING)");
    Schema schemaBefore = table("tl").schema();
    assertThat(schemaBefore.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(2, "dt", Types.StringType.get()))
                .asStruct());

    sql("ALTER TABLE tl MODIFY (dt STRING COMMENT 'comment for dt field')");
    Schema schemaAfter = table("tl").schema();
    assertThat(schemaAfter.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.optional(1, "id", Types.LongType.get()),
                    Types.NestedField.optional(
                        2, "dt", Types.StringType.get(), "comment for dt field"))
                .asStruct());
  }

  @TestTemplate
  public void testAlterTableConstraint() {
    sql("CREATE TABLE tl(id BIGINT NOT NULL, dt STRING NOT NULL, col1 STRING)");
    Schema schemaBefore = table("tl").schema();
    assertThat(schemaBefore.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.required(1, "id", Types.LongType.get()),
                    Types.NestedField.required(2, "dt", Types.StringType.get()),
                    Types.NestedField.optional(3, "col1", Types.StringType.get()))
                .asStruct());
    assertThat(schemaBefore.identifierFieldNames()).isEmpty();
    sql("ALTER TABLE tl ADD (PRIMARY KEY (id) NOT ENFORCED)");
    Schema schemaAfterAdd = table("tl").schema();
    assertThat(schemaAfterAdd.identifierFieldNames()).containsExactly("id");
    sql("ALTER TABLE tl MODIFY (PRIMARY KEY (dt) NOT ENFORCED)");
    Schema schemaAfterModify = table("tl").schema();
    assertThat(schemaAfterModify.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.required(1, "id", Types.LongType.get()),
                    Types.NestedField.required(2, "dt", Types.StringType.get()),
                    Types.NestedField.optional(3, "col1", Types.StringType.get()))
                .asStruct());
    assertThat(schemaAfterModify.identifierFieldNames()).containsExactly("dt");
    // Composite primary key
    sql("ALTER TABLE tl MODIFY (PRIMARY KEY (id, dt) NOT ENFORCED)");
    Schema schemaAfterComposite = table("tl").schema();
    assertThat(schemaAfterComposite.asStruct())
        .isEqualTo(
            new Schema(
                    Types.NestedField.required(1, "id", Types.LongType.get()),
                    Types.NestedField.required(2, "dt", Types.StringType.get()),
                    Types.NestedField.optional(3, "col1", Types.StringType.get()))
                .asStruct());
    assertThat(schemaAfterComposite.identifierFieldNames()).containsExactlyInAnyOrder("id", "dt");
    // Setting an optional field as primary key should fail
    // because Iceberg's SchemaUpdate does not allow incompatible changes.
    assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (PRIMARY KEY (col1) NOT ENFORCED)"))
        .isInstanceOf(TableException.class)
        .hasMessageContaining("Could not execute AlterTable")
        .hasRootCauseInstanceOf(IllegalArgumentException.class)
        .hasRootCauseMessage("Cannot add field col1 as an identifier field: not a required field");

    // Setting a composite key containing an optional field should fail
    // because Iceberg's SchemaUpdate does not allow incompatible changes.
    assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (PRIMARY KEY (id, col1) NOT ENFORCED)"))
        .isInstanceOf(TableException.class)
        .hasMessageContaining("Could not execute AlterTable")
        .hasRootCauseInstanceOf(IllegalArgumentException.class)
        .hasRootCauseMessage("Cannot add field col1 as an identifier field: not a required field");

    // Dropping constraints is not supported yet
    assertThatThrownBy(() -> sql("ALTER TABLE tl DROP PRIMARY KEY"))
        .isInstanceOf(TableException.class)
        .hasMessageContaining("Could not execute AlterTable")
        .hasRootCauseInstanceOf(UnsupportedOperationException.class)
        .hasRootCauseMessage("Unsupported table change: DropConstraint.");
  }

  @TestTemplate
  public void testRelocateTable() {
    assumeThat(isHadoopCatalog).as("HadoopCatalog does not support relocate table").isFalse();
    sql("CREATE TABLE tl(id BIGINT)");
    sql("ALTER TABLE tl SET('location'='file:///tmp/location')");
    assertThat(table("tl").location()).isEqualTo("file:///tmp/location");
  }

  @TestTemplate
  public void testSetCurrentAndCherryPickSnapshotId() {
    sql("CREATE TABLE tl(c1 INT, c2 STRING, c3 STRING) PARTITIONED BY (c1)");

    Table table = table("tl");

    DataFile fileA =
        DataFiles.builder(table.spec())
            .withPath("/path/to/data-a.parquet")
            .withFileSizeInBytes(10)
            .withPartitionPath("c1=0") // easy way to set partition data for now
            .withRecordCount(1)
            .build();
    DataFile fileB =
        DataFiles.builder(table.spec())
            .withPath("/path/to/data-b.parquet")
            .withFileSizeInBytes(10)
            .withPartitionPath("c1=1") // easy way to set partition data for now
            .withRecordCount(1)
            .build();
    DataFile replacementFile =
        DataFiles.builder(table.spec())
            .withPath("/path/to/data-a-replacement.parquet")
            .withFileSizeInBytes(10)
            .withPartitionPath("c1=0") // easy way to set partition data for now
            .withRecordCount(1)
            .build();

    table.newAppend().appendFile(fileA).commit();
    long snapshotId = table.currentSnapshot().snapshotId();

    // stage an overwrite that replaces FILE_A
    table.newReplacePartitions().addFile(replacementFile).stageOnly().commit();

    Snapshot staged = Iterables.getLast(table.snapshots());
    assertThat(staged.operation())
        .as("Should find the staged overwrite snapshot")
        .isEqualTo(DataOperations.OVERWRITE);
    // add another append so that the original commit can't be fast-forwarded
    table.newAppend().appendFile(fileB).commit();

    // test cherry pick
    sql("ALTER TABLE tl SET('cherry-pick-snapshot-id'='%s')", staged.snapshotId());
    validateTableFiles(table, fileB, replacementFile);

    // test set current snapshot
    sql("ALTER TABLE tl SET('current-snapshot-id'='%s')", snapshotId);
    validateTableFiles(table, fileA);
  }

  private void validateTableFiles(Table tbl, DataFile... expectedFiles) {
    tbl.refresh();
    Set<CharSequence> expectedFilePaths =
        Arrays.stream(expectedFiles).map(DataFile::location).collect(Collectors.toSet());
    Set<CharSequence> actualFilePaths =
        StreamSupport.stream(tbl.newScan().planFiles().spliterator(), false)
            .map(FileScanTask::file)
            .map(ContentFile::location)
            .collect(Collectors.toSet());
    assertThat(actualFilePaths).as("Files should match").isEqualTo(expectedFilePaths);
  }

  @TestTemplate
  public void testCreateMaterializedTableIsUnsupported() {
    CatalogMaterializedTable materializedTable =
        CatalogMaterializedTable.newBuilder()
            .schema(
                org.apache.flink.table.api.Schema.newBuilder()
                    .column("id", DataTypes.BIGINT())
                    .build())
            .definitionQuery("SELECT id FROM tl")
            .freshness(IntervalFreshness.ofMinute("5"))
            .logicalRefreshMode(CatalogMaterializedTable.LogicalRefreshMode.AUTOMATIC)
            .refreshMode(CatalogMaterializedTable.RefreshMode.CONTINUOUS)
            .refreshStatus(CatalogMaterializedTable.RefreshStatus.INITIALIZING)
            .build();

    assertThatThrownBy(
            () ->
                getTableEnv()
                    .getCatalog(catalogName)
                    .get()
                    .createTable(new ObjectPath(DATABASE, "mt_table"), materializedTable, false))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessageContaining("Materialized tables and other table kinds are not supported");
  }

  private Table table(String name) {
    return validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, name));
  }

  private CatalogTable catalogTable(String name) throws TableNotExistException {
    return catalogTable(getTableEnv().getCurrentCatalog(), DATABASE, name);
  }

  private CatalogTable catalogTable(String catalog, String database, String table)
      throws TableNotExistException {
    return (CatalogTable)
        getTableEnv().getCatalog(catalog).get().getTable(new ObjectPath(database, table));
  }
}
