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

import static org.apache.spark.sql.functions.col;
import static org.apache.spark.sql.functions.current_date;
import static org.apache.spark.sql.functions.date_add;
import static org.apache.spark.sql.functions.date_format;
import static org.apache.spark.sql.functions.expr;
import static org.assertj.core.api.Assertions.assertThat;

import io.delta.standalone.DeltaLog;
import io.delta.standalone.Operation;
import io.delta.standalone.OptimisticTransaction;
import io.delta.standalone.VersionLog;
import io.delta.standalone.actions.Action;
import io.delta.standalone.actions.AddFile;
import io.delta.standalone.actions.RemoveFile;
import io.delta.standalone.exceptions.DeltaConcurrentModificationException;
import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.sql.Timestamp;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.stream.Collectors;
import org.apache.commons.codec.DecoderException;
import org.apache.commons.codec.net.URLCodec;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotChanges;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.spark.Spark3Util;
import org.apache.iceberg.spark.SparkCatalog;
import org.apache.iceberg.util.LocationUtil;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.connector.catalog.CatalogPlugin;
import org.apache.spark.sql.delta.catalog.DeltaCatalog;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

public class TestSnapshotDeltaLakeTable extends SparkDeltaLakeSnapshotTestBase {
  private static final String SNAPSHOT_SOURCE_PROP = "snapshot_source";
  private static final String DELTA_SOURCE_VALUE = "delta";
  private static final String ORIGINAL_LOCATION_PROP = "original_location";
  private static final String NAMESPACE = "delta_conversion_test";
  private static final String DEFAULT_SPARK_CATALOG = "spark_catalog";
  private static final String ICEBERG_CATALOG_NAME = "iceberg_hive";
  private static final Map<String, String> CONFIG =
      ImmutableMap.of(
          "type", "hive",
          "default-namespace", "default",
          "parquet-enabled", "true",
          "cache-enabled",
              "false" // Spark will delete tables using v1, leaving the cache out of sync
          );
  private static Dataset<Row> typeTestDataFrame;
  private static Dataset<Row> nestedDataFrame;

  @TempDir private File tempA;
  @TempDir private File tempB;

  public TestSnapshotDeltaLakeTable() {
    super(ICEBERG_CATALOG_NAME, SparkCatalog.class.getName(), CONFIG);
    spark.conf().set("spark.sql.catalog." + DEFAULT_SPARK_CATALOG, DeltaCatalog.class.getName());
  }

  @BeforeAll
  public static void beforeClass() {
    spark.sql(String.format("CREATE DATABASE IF NOT EXISTS %s", NAMESPACE));

    typeTestDataFrame =
        spark
            .range(0, 5, 1, 5)
            .withColumnRenamed("id", "longCol")
            .withColumn("intCol", expr("CAST(longCol AS INT)"))
            .withColumn("floatCol", expr("CAST(longCol AS FLOAT)"))
            .withColumn("doubleCol", expr("CAST(longCol AS DOUBLE)"))
            .withColumn("dateCol", date_add(current_date(), 1))
            .withColumn("timestampCol", expr("TO_TIMESTAMP(dateCol)"))
            .withColumn("timestampStrCol", expr("CAST(timestampCol AS STRING)"))
            .withColumn("stringCol", date_format(col("timestampCol"), "yyyy/M/d"))
            .withColumn("booleanCol", expr("longCol > 5"))
            .withColumn("binaryCol", expr("CAST(longCol AS BINARY)"))
            .withColumn("byteCol", expr("CAST(longCol AS BYTE)"))
            .withColumn("decimalCol", expr("CAST(longCol AS DECIMAL(10, 2))"))
            .withColumn("shortCol", expr("CAST(longCol AS SHORT)"))
            .withColumn("mapCol", expr("MAP(longCol, decimalCol)"))
            .withColumn("arrayCol", expr("ARRAY(longCol)"))
            .withColumn("structCol", expr("STRUCT(mapCol, arrayCol)"));
    nestedDataFrame =
        spark
            .range(0, 5, 1, 5)
            .withColumn("longCol", expr("id"))
            .withColumn("decimalCol", expr("CAST(longCol AS DECIMAL(10, 2))"))
            .withColumn("magic_number", expr("rand(5) * 100"))
            .withColumn("dateCol", date_add(current_date(), 1))
            .withColumn("dateString", expr("CAST(dateCol AS STRING)"))
            .withColumn("random1", expr("CAST(rand(5) * 100 as LONG)"))
            .withColumn("random2", expr("CAST(rand(51) * 100 as LONG)"))
            .withColumn("random3", expr("CAST(rand(511) * 100 as LONG)"))
            .withColumn("random4", expr("CAST(rand(15) * 100 as LONG)"))
            .withColumn("random5", expr("CAST(rand(115) * 100 as LONG)"))
            .withColumn("innerStruct1", expr("STRUCT(random1, random2)"))
            .withColumn("innerStruct2", expr("STRUCT(random3, random4)"))
            .withColumn("structCol1", expr("STRUCT(innerStruct1, innerStruct2)"))
            .withColumn(
                "innerStruct3",
                expr("STRUCT(SHA1(CAST(random5 AS BINARY)), SHA1(CAST(random1 AS BINARY)))"))
            .withColumn(
                "structCol2",
                expr(
                    "STRUCT(innerStruct3, STRUCT(SHA1(CAST(random2 AS BINARY)), SHA1(CAST(random3 AS BINARY))))"))
            .withColumn("arrayCol", expr("ARRAY(random1, random2, random3, random4, random5)"))
            .withColumn("arrayStructCol", expr("ARRAY(innerStruct1, innerStruct1, innerStruct1)"))
            .withColumn("mapCol1", expr("MAP(structCol1, structCol2)"))
            .withColumn("mapCol2", expr("MAP(longCol, dateString)"))
            .withColumn("mapCol3", expr("MAP(dateCol, arrayCol)"))
            .withColumn("structCol3", expr("STRUCT(structCol2, mapCol3, arrayCol)"));
  }

  @AfterAll
  public static void afterClass() {
    spark.sql(String.format("DROP DATABASE IF EXISTS %s CASCADE", NAMESPACE));
  }

  @Test
  public void testBasicSnapshotPartitioned() {
    String partitionedIdentifier = destName(DEFAULT_SPARK_CATALOG, "partitioned_table");
    String partitionedLocation = tempA.toURI().toString();

    writeDeltaTable(nestedDataFrame, partitionedIdentifier, partitionedLocation, "id");
    spark.sql("DELETE FROM " + partitionedIdentifier + " WHERE id=3");
    spark.sql("UPDATE " + partitionedIdentifier + " SET id=3 WHERE id=1");

    String newTableIdentifier = destName(ICEBERG_CATALOG_NAME, "iceberg_partitioned_table");
    SnapshotDeltaLakeTable.Result result =
        DeltaLakeToIcebergMigrationSparkIntegration.snapshotDeltaLakeTable(
                spark, newTableIdentifier, partitionedLocation)
            .execute();

    checkSnapshotIntegrity(
        partitionedLocation, partitionedIdentifier, newTableIdentifier, result, 0);
    checkTagContentAndOrder(partitionedLocation, newTableIdentifier, 0);
    checkIcebergTableLocation(newTableIdentifier, partitionedLocation);
  }

  @Test
  public void testBasicSnapshotUnpartitioned() {
    String unpartitionedIdentifier = destName(DEFAULT_SPARK_CATALOG, "unpartitioned_table");
    String unpartitionedLocation = tempA.toURI().toString();

    writeDeltaTable(nestedDataFrame, unpartitionedIdentifier, unpartitionedLocation);
    spark.sql("DELETE FROM " + unpartitionedIdentifier + " WHERE id=3");
    spark.sql("UPDATE " + unpartitionedIdentifier + " SET id=3 WHERE id=1");

    String newTableIdentifier = destName(ICEBERG_CATALOG_NAME, "iceberg_unpartitioned_table");
    SnapshotDeltaLakeTable.Result result =
        DeltaLakeToIcebergMigrationSparkIntegration.snapshotDeltaLakeTable(
                spark, newTableIdentifier, unpartitionedLocation)
            .execute();

    checkSnapshotIntegrity(
        unpartitionedLocation, unpartitionedIdentifier, newTableIdentifier, result, 0);
    checkTagContentAndOrder(unpartitionedLocation, newTableIdentifier, 0);
    checkIcebergTableLocation(newTableIdentifier, unpartitionedLocation);
  }

  @Test
  public void testSnapshotWithNewLocation() {
    String partitionedIdentifier = destName(DEFAULT_SPARK_CATALOG, "partitioned_table");
    String partitionedLocation = tempA.toURI().toString();
    String newIcebergTableLocation = tempB.toURI().toString();

    writeDeltaTable(nestedDataFrame, partitionedIdentifier, partitionedLocation, "id");
    spark.sql("DELETE FROM " + partitionedIdentifier + " WHERE id=3");
    spark.sql("UPDATE " + partitionedIdentifier + " SET id=3 WHERE id=1");

    String newTableIdentifier = destName(ICEBERG_CATALOG_NAME, "iceberg_new_table_location_table");
    SnapshotDeltaLakeTable.Result result =
        DeltaLakeToIcebergMigrationSparkIntegration.snapshotDeltaLakeTable(
                spark, newTableIdentifier, partitionedLocation)
            .tableLocation(newIcebergTableLocation)
            .execute();

    checkSnapshotIntegrity(
        partitionedLocation, partitionedIdentifier, newTableIdentifier, result, 0);
    checkTagContentAndOrder(partitionedLocation, newTableIdentifier, 0);
    checkIcebergTableLocation(newTableIdentifier, newIcebergTableLocation);
  }

  @Test
  public void testSnapshotWithAdditionalProperties() {
    String unpartitionedIdentifier = destName(DEFAULT_SPARK_CATALOG, "unpartitioned_table");
    String unpartitionedLocation = tempA.toURI().toString();

    writeDeltaTable(nestedDataFrame, unpartitionedIdentifier, unpartitionedLocation);
    spark.sql("DELETE FROM " + unpartitionedIdentifier + " WHERE id=3");
    spark.sql("UPDATE " + unpartitionedIdentifier + " SET id=3 WHERE id=1");

    // add some properties to the original delta table
    spark.sql(
        "ALTER TABLE "
            + unpartitionedIdentifier
            + " SET TBLPROPERTIES ('foo'='bar', 'test0'='test0')");

    String newTableIdentifier =
        destName(ICEBERG_CATALOG_NAME, "iceberg_additional_properties_table");
    SnapshotDeltaLakeTable.Result result =
        DeltaLakeToIcebergMigrationSparkIntegration.snapshotDeltaLakeTable(
                spark, newTableIdentifier, unpartitionedLocation)
            .tableProperty("test1", "test1")
            .tableProperties(
                ImmutableMap.of(
                    "test2", "test2", "test3", "test3", "test4",
                    "test4")) // add additional iceberg table properties
            .execute();

    checkSnapshotIntegrity(
        unpartitionedLocation, unpartitionedIdentifier, newTableIdentifier, result, 0);
    checkTagContentAndOrder(unpartitionedLocation, newTableIdentifier, 0);
    checkIcebergTableLocation(newTableIdentifier, unpartitionedLocation);
    checkIcebergTableProperties(
        newTableIdentifier,
        ImmutableMap.of(
            "foo", "bar", "test0", "test0", "test1", "test1", "test2", "test2", "test3", "test3",
            "test4", "test4"),
        unpartitionedLocation);
  }

  @Test
  public void testSnapshotTableWithExternalDataFiles() {
    String unpartitionedIdentifier = destName(DEFAULT_SPARK_CATALOG, "unpartitioned_table");
    String externalDataFilesIdentifier =
        destName(DEFAULT_SPARK_CATALOG, "external_data_files_table");
    String unpartitionedLocation = tempA.toURI().toString();
    String externalDataFilesTableLocation = tempB.toURI().toString();

    writeDeltaTable(nestedDataFrame, unpartitionedIdentifier, unpartitionedLocation);
    spark.sql("DELETE FROM " + unpartitionedIdentifier + " WHERE id=3");
    spark.sql("UPDATE " + unpartitionedIdentifier + " SET id=3 WHERE id=1");

    writeDeltaTable(nestedDataFrame, externalDataFilesIdentifier, externalDataFilesTableLocation);
    // Add parquet files to default.external_data_files_table. The newly added parquet files
    // are not at the same location as the table.
    addExternalDatafiles(externalDataFilesTableLocation, unpartitionedLocation);

    String newTableIdentifier = destName(ICEBERG_CATALOG_NAME, "iceberg_external_data_files_table");
    SnapshotDeltaLakeTable.Result result =
        DeltaLakeToIcebergMigrationSparkIntegration.snapshotDeltaLakeTable(
                spark, newTableIdentifier, externalDataFilesTableLocation)
            .execute();
    checkSnapshotIntegrity(
        externalDataFilesTableLocation, externalDataFilesIdentifier, newTableIdentifier, result, 0);
    checkTagContentAndOrder(externalDataFilesTableLocation, newTableIdentifier, 0);
    checkIcebergTableLocation(newTableIdentifier, externalDataFilesTableLocation);
    checkDataFilePathsIntegrity(newTableIdentifier, externalDataFilesTableLocation);
  }

  @Test
  public void testSnapshotSupportedTypes() {
    String typeTestIdentifier = destName(DEFAULT_SPARK_CATALOG, "type_test_table");
    String typeTestTableLocation = tempA.toURI().toString();

    writeDeltaTable(
        typeTestDataFrame,
        typeTestIdentifier,
        typeTestTableLocation,
        "stringCol",
        "timestampStrCol",
        "booleanCol",
        "longCol");
    String newTableIdentifier = destName(ICEBERG_CATALOG_NAME, "iceberg_type_test_table");
    SnapshotDeltaLakeTable.Result result =
        DeltaLakeToIcebergMigrationSparkIntegration.snapshotDeltaLakeTable(
                spark, newTableIdentifier, typeTestTableLocation)
            .execute();
    checkSnapshotIntegrity(
        typeTestTableLocation, typeTestIdentifier, newTableIdentifier, result, 0);
    checkTagContentAndOrder(typeTestTableLocation, newTableIdentifier, 0);
    checkIcebergTableLocation(newTableIdentifier, typeTestTableLocation);
    checkIcebergTableProperties(newTableIdentifier, ImmutableMap.of(), typeTestTableLocation);
  }

  @Test
  public void testSnapshotVacuumTable() throws IOException {
    String vacuumTestIdentifier = destName(DEFAULT_SPARK_CATALOG, "vacuum_test_table");
    String vacuumTestTableLocation = tempA.toURI().toString();

    writeDeltaTable(nestedDataFrame, vacuumTestIdentifier, vacuumTestTableLocation);
    Random random = new Random();
    for (int i = 0; i < 13; i++) {
      spark.sql(
          "UPDATE "
              + vacuumTestIdentifier
              + " SET magic_number = "
              + random.nextDouble()
              + " WHERE id = 1");
    }

    boolean deleteResult =
        Files.deleteIfExists(
            Paths.get(
                URI.create(
                    vacuumTestTableLocation.concat("/_delta_log/00000000000000000000.json"))));
    assertThat(deleteResult).isTrue();
    spark.sql("VACUUM " + vacuumTestIdentifier + " RETAIN 0 HOURS");

    String newTableIdentifier = destName(ICEBERG_CATALOG_NAME, "iceberg_vacuum_table");
    SnapshotDeltaLakeTable.Result result =
        DeltaLakeToIcebergMigrationSparkIntegration.snapshotDeltaLakeTable(
                spark, newTableIdentifier, vacuumTestTableLocation)
            .execute();
    checkSnapshotIntegrity(
        vacuumTestTableLocation, vacuumTestIdentifier, newTableIdentifier, result, 13);
    checkTagContentAndOrder(vacuumTestTableLocation, newTableIdentifier, 13);
    checkIcebergTableLocation(newTableIdentifier, vacuumTestTableLocation);
  }

  @Test
  public void testSnapshotLogCleanTable() throws IOException {
    String logCleanTestIdentifier = destName(DEFAULT_SPARK_CATALOG, "log_clean_test_table");
    String logCleanTestTableLocation = tempA.toURI().toString();

    writeDeltaTable(nestedDataFrame, logCleanTestIdentifier, logCleanTestTableLocation, "id");
    Random random = new Random();
    for (int i = 0; i < 25; i++) {
      spark.sql(
          "UPDATE "
              + logCleanTestIdentifier
              + " SET magic_number = "
              + random.nextDouble()
              + " WHERE id = 1");
    }

    boolean deleteResult =
        Files.deleteIfExists(
            Paths.get(
                URI.create(
                    logCleanTestTableLocation.concat("/_delta_log/00000000000000000000.json"))));
    assertThat(deleteResult).isTrue();

    String newTableIdentifier = destName(ICEBERG_CATALOG_NAME, "iceberg_log_clean_table");
    SnapshotDeltaLakeTable.Result result =
        DeltaLakeToIcebergMigrationSparkIntegration.snapshotDeltaLakeTable(
                spark, newTableIdentifier, logCleanTestTableLocation)
            .execute();
    checkSnapshotIntegrity(
        logCleanTestTableLocation, logCleanTestIdentifier, newTableIdentifier, result, 10);
    checkTagContentAndOrder(logCleanTestTableLocation, newTableIdentifier, 10);
    checkIcebergTableLocation(newTableIdentifier, logCleanTestTableLocation);
  }

  private void checkSnapshotIntegrity(
      String deltaTableLocation,
      String deltaTableIdentifier,
      String icebergTableIdentifier,
      SnapshotDeltaLakeTable.Result snapshotReport,
      long firstConstructableVersion) {
    DeltaLog deltaLog = DeltaLog.forTable(spark.sessionState().newHadoopConf(), deltaTableLocation);

    List<Row> deltaTableContents =
        spark.sql("SELECT * FROM " + deltaTableIdentifier).collectAsList();
    List<Row> icebergTableContents =
        spark.sql("SELECT * FROM " + icebergTableIdentifier).collectAsList();

    assertThat(deltaTableContents).hasSize(icebergTableContents.size());
    assertThat(snapshotReport.snapshotDataFilesCount())
        .isEqualTo(countDataFilesInDeltaLakeTable(deltaLog, firstConstructableVersion));
    assertThat(icebergTableContents).containsExactlyInAnyOrderElementsOf(deltaTableContents);
  }

  private void checkTagContentAndOrder(
      String deltaTableLocation, String icebergTableIdentifier, long firstConstructableVersion) {
    DeltaLog deltaLog = DeltaLog.forTable(spark.sessionState().newHadoopConf(), deltaTableLocation);
    long currentVersion = deltaLog.snapshot().getVersion();
    Table icebergTable = getIcebergTable(icebergTableIdentifier);
    Map<String, SnapshotRef> icebergSnapshotRefs = icebergTable.refs();
    List<Snapshot> icebergSnapshots = Lists.newArrayList(icebergTable.snapshots());

    assertThat(icebergSnapshots).hasSize((int) (currentVersion - firstConstructableVersion + 1));

    for (int i = 0; i < icebergSnapshots.size(); i++) {
      long deltaVersion = firstConstructableVersion + i;
      Snapshot currentIcebergSnapshot = icebergSnapshots.get(i);

      String expectedVersionTag = "delta-version-" + deltaVersion;
      icebergSnapshotRefs.get(expectedVersionTag);
      assertThat(icebergSnapshotRefs.get(expectedVersionTag)).isNotNull();
      assertThat(icebergSnapshotRefs.get(expectedVersionTag).isTag()).isTrue();
      assertThat(icebergSnapshotRefs.get(expectedVersionTag).snapshotId())
          .isEqualTo(currentIcebergSnapshot.snapshotId());

      Timestamp deltaVersionTimestamp = deltaLog.getCommitInfoAt(deltaVersion).getTimestamp();
      assertThat(deltaVersionTimestamp).isNotNull();
      String expectedTimestampTag = "delta-ts-" + deltaVersionTimestamp.getTime();

      assertThat(icebergSnapshotRefs.get(expectedTimestampTag)).isNotNull();
      assertThat(icebergSnapshotRefs.get(expectedTimestampTag).isTag()).isTrue();
      assertThat(icebergSnapshotRefs.get(expectedTimestampTag).snapshotId())
          .isEqualTo(currentIcebergSnapshot.snapshotId());
    }
  }

  private void checkIcebergTableLocation(String icebergTableIdentifier, String expectedLocation) {
    Table icebergTable = getIcebergTable(icebergTableIdentifier);
    assertThat(icebergTable.location())
        .isEqualTo(LocationUtil.stripTrailingSlash(expectedLocation));
  }

  private void checkIcebergTableProperties(
      String icebergTableIdentifier,
      Map<String, String> expectedAdditionalProperties,
      String deltaTableLocation) {
    Table icebergTable = getIcebergTable(icebergTableIdentifier);
    ImmutableMap.Builder<String, String> expectedPropertiesBuilder = ImmutableMap.builder();
    // The snapshot action will put some fixed properties to the table
    expectedPropertiesBuilder.put(SNAPSHOT_SOURCE_PROP, DELTA_SOURCE_VALUE);
    expectedPropertiesBuilder.putAll(expectedAdditionalProperties);
    ImmutableMap<String, String> expectedProperties = expectedPropertiesBuilder.build();

    assertThat(icebergTable.properties().entrySet()).containsAll(expectedProperties.entrySet());
    assertThat(icebergTable.properties()).containsEntry(ORIGINAL_LOCATION_PROP, deltaTableLocation);
  }

  private void checkDataFilePathsIntegrity(
      String icebergTableIdentifier, String deltaTableLocation) {
    Table icebergTable = getIcebergTable(icebergTableIdentifier);
    DeltaLog deltaLog = DeltaLog.forTable(spark.sessionState().newHadoopConf(), deltaTableLocation);
    // checkSnapshotIntegrity already checks the number of data files in the snapshot iceberg table
    // equals that in the original delta lake table
    List<String> deltaTableDataFilePaths =
        deltaLog.update().getAllFiles().stream()
            .map(f -> getFullFilePath(f.getPath(), deltaLog.getPath().toString()))
            .collect(Collectors.toList());
    SnapshotChanges.builderFor(icebergTable)
        .snapshot(icebergTable.currentSnapshot())
        .build()
        .addedDataFiles()
        .forEach(
            dataFile -> {
              assertThat(URI.create(dataFile.location()).isAbsolute()).isTrue();
              assertThat(deltaTableDataFilePaths).contains(dataFile.location());
            });
  }

  private Table getIcebergTable(String icebergTableIdentifier) {
    CatalogPlugin defaultCatalog = spark.sessionState().catalogManager().currentCatalog();
    Spark3Util.CatalogAndIdentifier catalogAndIdent =
        Spark3Util.catalogAndIdentifier(
            "test catalog", spark, icebergTableIdentifier, defaultCatalog);
    return Spark3Util.loadIcebergCatalog(spark, catalogAndIdent.catalog().name())
        .loadTable(TableIdentifier.parse(catalogAndIdent.identifier().toString()));
  }

  private String destName(String catalogName, String dest) {
    if (catalogName.equals(DEFAULT_SPARK_CATALOG)) {
      return NAMESPACE + "." + catalogName + "_" + dest;
    }
    return catalogName + "." + NAMESPACE + "." + catalogName + "_" + dest;
  }

  /**
   * Add parquet files manually to a delta lake table to mock the situation that some data files are
   * not in the same location as the delta lake table. The case that {@link AddFile#getPath()} or
   * {@link RemoveFile#getPath()} returns absolute path.
   *
   * <p>The known <a href="https://github.com/delta-io/connectors/issues/380">issue</a> makes it
   * necessary to manually rebuild the AddFile to avoid deserialization error when committing the
   * transaction.
   */
  private void addExternalDatafiles(
      String targetDeltaTableLocation, String sourceDeltaTableLocation) {
    DeltaLog targetLog =
        DeltaLog.forTable(spark.sessionState().newHadoopConf(), targetDeltaTableLocation);
    OptimisticTransaction transaction = targetLog.startTransaction();
    DeltaLog sourceLog =
        DeltaLog.forTable(spark.sessionState().newHadoopConf(), sourceDeltaTableLocation);
    List<AddFile> newFiles =
        sourceLog.update().getAllFiles().stream()
            .map(
                f ->
                    AddFile.builder(
                            getFullFilePath(f.getPath(), sourceLog.getPath().toString()),
                            f.getPartitionValues(),
                            f.getSize(),
                            System.currentTimeMillis(),
                            true)
                        .build())
            .collect(Collectors.toList());
    try {
      transaction.commit(newFiles, new Operation(Operation.Name.UPDATE), "Delta-Lake/2.2.0");
    } catch (DeltaConcurrentModificationException e) {
      throw new RuntimeException(e);
    }
  }

  private static String getFullFilePath(String path, String tableRoot) {
    URI dataFileUri = URI.create(path);
    try {
      String decodedPath = new URLCodec().decode(path);
      if (dataFileUri.isAbsolute()) {
        return decodedPath;
      } else {
        return tableRoot + File.separator + decodedPath;
      }
    } catch (DecoderException e) {
      throw new IllegalArgumentException(String.format("Cannot decode path %s", path), e);
    }
  }

  private void writeDeltaTable(
      Dataset<Row> df, String identifier, String path, String... partitionColumns) {
    spark.sql(String.format("DROP TABLE IF EXISTS %s", identifier));
    if (partitionColumns.length > 0) {
      df.write()
          .format("delta")
          .mode(SaveMode.Append)
          .option("path", path)
          .partitionBy(partitionColumns)
          .saveAsTable(identifier);
    } else {
      df.write().format("delta").mode(SaveMode.Append).option("path", path).saveAsTable(identifier);
    }
  }

  private long countDataFilesInDeltaLakeTable(DeltaLog deltaLog, long firstConstructableVersion) {
    long dataFilesCount = 0;

    List<AddFile> initialDataFiles =
        deltaLog.getSnapshotForVersionAsOf(firstConstructableVersion).getAllFiles();
    dataFilesCount += initialDataFiles.size();

    Iterator<VersionLog> versionLogIterator =
        deltaLog.getChanges(
            firstConstructableVersion + 1, false // not throw exception when data loss detected
            );

    while (versionLogIterator.hasNext()) {
      VersionLog versionLog = versionLogIterator.next();
      List<Action> addFiles =
          versionLog.getActions().stream()
              .filter(action -> action instanceof AddFile)
              .collect(Collectors.toList());
      dataFilesCount += addFiles.size();
    }

    return dataFilesCount;
  }
}
