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

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

import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

@ExtendWith(ParameterizedTestExtension.class)
public class TestSnapshot extends TestBase {

  @TestTemplate
  public void snapshotValueMethodsIncludeMetadataFields() {
    Snapshot snapshot =
        new BaseSnapshot(
            0, 1, null, 0, DataOperations.APPEND, null, 1, "ml.avro", 10L, 5L, "key-1");
    Snapshot same =
        new BaseSnapshot(
            0, 1, null, 0, DataOperations.APPEND, null, 1, "ml.avro", 10L, 5L, "key-1");
    Snapshot differentFirstRowId =
        new BaseSnapshot(
            0, 1, null, 0, DataOperations.APPEND, null, 1, "ml.avro", 11L, 5L, "key-1");
    Snapshot differentAddedRows =
        new BaseSnapshot(
            0, 1, null, 0, DataOperations.APPEND, null, 1, "ml.avro", 10L, 6L, "key-1");
    Snapshot differentKeyId =
        new BaseSnapshot(
            0, 1, null, 0, DataOperations.APPEND, null, 1, "ml.avro", 10L, 5L, "key-2");
    Snapshot withoutMetadataFields =
        new BaseSnapshot(
            0, 1, null, 0, DataOperations.APPEND, null, 1, "ml.avro", null, null, null);

    assertThat(snapshot).isEqualTo(same);
    assertThat(snapshot.hashCode()).isEqualTo(same.hashCode());
    assertThat(snapshot).isNotEqualTo(differentFirstRowId);
    assertThat(snapshot).isNotEqualTo(differentAddedRows);
    assertThat(snapshot).isNotEqualTo(differentKeyId);
    assertThat(snapshot).isNotEqualTo(withoutMetadataFields);
    assertThat(snapshot.toString()).contains("first-row-id=10", "added-rows=5", "key-id=key-1");
  }

  @TestTemplate
  public void testAppendFilesFromTable() {
    table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit();

    // collect data files from deserialization
    Iterable<DataFile> filesToAdd = SnapshotChanges.builderFor(table).build().addedDataFiles();

    table.newDelete().deleteFile(FILE_A).deleteFile(FILE_B).commit();

    Snapshot oldSnapshot = table.currentSnapshot();

    AppendFiles fastAppend = table.newFastAppend();
    for (DataFile file : filesToAdd) {
      fastAppend.appendFile(file);
    }

    Snapshot newSnapshot = fastAppend.apply();
    validateSnapshot(oldSnapshot, newSnapshot, FILE_A, FILE_B);
  }

  @TestTemplate
  public void testAppendFoundFiles() {
    table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit();

    Iterable<DataFile> filesToAdd =
        FindFiles.in(table)
            .inPartition(table.spec(), StaticDataTask.Row.of(0))
            .inPartition(table.spec(), StaticDataTask.Row.of(1))
            .collect();

    table.newDelete().deleteFile(FILE_A).deleteFile(FILE_B).commit();

    Snapshot oldSnapshot = table.currentSnapshot();

    AppendFiles fastAppend = table.newFastAppend();
    for (DataFile file : filesToAdd) {
      fastAppend.appendFile(file);
    }

    Snapshot newSnapshot = fastAppend.apply();
    validateSnapshot(oldSnapshot, newSnapshot, FILE_A, FILE_B);
  }

  @TestTemplate
  public void testCachedDataFiles() {
    table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit();

    table.updateSpec().addField(Expressions.truncate("data", 2)).commit();

    DataFile secondSnapshotDataFile = newDataFile("data_bucket=8/data_trunc_2=aa");

    table.newFastAppend().appendFile(secondSnapshotDataFile).commit();

    DataFile thirdSnapshotDataFile = newDataFile("data_bucket=8/data_trunc_2=bb");

    table.newOverwrite().deleteFile(FILE_A).addFile(thirdSnapshotDataFile).commit();

    Snapshot thirdSnapshot = table.currentSnapshot();

    SnapshotChanges changes = SnapshotChanges.builderFor(table).snapshot(thirdSnapshot).build();
    Iterable<DataFile> removedDataFiles = changes.removedDataFiles();
    assertThat(removedDataFiles).as("Must have 1 removed data file").hasSize(1);

    DataFile removedDataFile = Iterables.getOnlyElement(removedDataFiles);
    assertThat(removedDataFile.location()).isEqualTo(FILE_A.location());
    assertThat(removedDataFile.specId()).isEqualTo(FILE_A.specId());
    assertThat(removedDataFile.partition()).isEqualTo(FILE_A.partition());

    Iterable<DataFile> addedDataFiles = changes.addedDataFiles();
    assertThat(addedDataFiles).as("Must have 1 added data file").hasSize(1);

    DataFile addedDataFile = Iterables.getOnlyElement(addedDataFiles);
    assertThat(addedDataFile.location()).isEqualTo(thirdSnapshotDataFile.location());
    assertThat(addedDataFile.specId()).isEqualTo(thirdSnapshotDataFile.specId());
    assertThat(addedDataFile.partition()).isEqualTo(thirdSnapshotDataFile.partition());
  }

  @TestTemplate
  public void testCachedDeleteFiles() {
    assumeThat(formatVersion).as("Delete files only supported in V2").isGreaterThanOrEqualTo(2);

    table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit();

    table.updateSpec().addField(Expressions.truncate("data", 2)).commit();

    DataFile secondSnapshotDataFile = newDataFile("data_bucket=8/data_trunc_2=aa");
    DeleteFile secondSnapshotDeleteFile = newDeletes(secondSnapshotDataFile);

    table
        .newRowDelta()
        .addRows(secondSnapshotDataFile)
        .addDeletes(secondSnapshotDeleteFile)
        .commit();

    DeleteFile thirdSnapshotDeleteFile = newDeletes(secondSnapshotDataFile);

    ImmutableSet<DeleteFile> replacedDeleteFiles = ImmutableSet.of(secondSnapshotDeleteFile);
    ImmutableSet<DeleteFile> newDeleteFiles = ImmutableSet.of(thirdSnapshotDeleteFile);

    table
        .newRewrite()
        .rewriteFiles(ImmutableSet.of(), replacedDeleteFiles, ImmutableSet.of(), newDeleteFiles)
        .commit();

    Snapshot thirdSnapshot = table.currentSnapshot();

    SnapshotChanges changes = SnapshotChanges.builderFor(table).snapshot(thirdSnapshot).build();
    Iterable<DeleteFile> removedDeleteFiles = changes.removedDeleteFiles();
    assertThat(removedDeleteFiles).as("Must have 1 removed delete file").hasSize(1);

    DeleteFile removedDeleteFile = Iterables.getOnlyElement(removedDeleteFiles);
    assertThat(removedDeleteFile.location()).isEqualTo(secondSnapshotDeleteFile.location());
    assertThat(removedDeleteFile.specId()).isEqualTo(secondSnapshotDeleteFile.specId());
    assertThat(removedDeleteFile.partition()).isEqualTo(secondSnapshotDeleteFile.partition());

    Iterable<DeleteFile> addedDeleteFiles = changes.addedDeleteFiles();
    assertThat(addedDeleteFiles).as("Must have 1 added delete file").hasSize(1);

    DeleteFile addedDeleteFile = Iterables.getOnlyElement(addedDeleteFiles);
    assertThat(addedDeleteFile.location()).isEqualTo(thirdSnapshotDeleteFile.location());
    assertThat(addedDeleteFile.specId()).isEqualTo(thirdSnapshotDeleteFile.specId());
    assertThat(addedDeleteFile.partition()).isEqualTo(thirdSnapshotDeleteFile.partition());
  }

  @TestTemplate
  public void testSequenceNumbersInAddedDataFiles() {
    long expectedSequenceNumber = 0L;
    if (formatVersion >= 2) {
      expectedSequenceNumber = 1L;
    }

    runAddedDataFileSequenceNumberTest(expectedSequenceNumber);

    if (formatVersion >= 2) {
      ++expectedSequenceNumber;
    }

    runAddedDataFileSequenceNumberTest(expectedSequenceNumber);
  }

  private void runAddedDataFileSequenceNumberTest(long expectedSequenceNumber) {
    table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit();

    Snapshot snapshot = table.currentSnapshot();
    Iterable<DataFile> addedDataFiles =
        SnapshotChanges.builderFor(table).snapshot(snapshot).build().addedDataFiles();

    assertThat(snapshot.sequenceNumber())
        .as("Sequence number mismatch in Snapshot")
        .isEqualTo(expectedSequenceNumber);

    for (DataFile df : addedDataFiles) {
      assertThat(df.dataSequenceNumber().longValue())
          .as("Data sequence number mismatch")
          .isEqualTo(expectedSequenceNumber);
      assertThat(df.fileSequenceNumber().longValue())
          .as("File sequence number mismatch")
          .isEqualTo(expectedSequenceNumber);
    }
  }

  @TestTemplate
  public void testSequenceNumbersInRemovedDataFiles() {
    table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit();

    long expectedSnapshotSequenceNumber = 0L;
    if (formatVersion >= 2) {
      expectedSnapshotSequenceNumber = 2L;
    }

    long expectedFileSequenceNumber = 0L;
    if (formatVersion >= 2) {
      expectedFileSequenceNumber = 1L;
    }

    runRemovedDataFileSequenceNumberTest(
        FILE_A, expectedSnapshotSequenceNumber, expectedFileSequenceNumber);

    if (formatVersion >= 2) {
      ++expectedSnapshotSequenceNumber;
    }

    runRemovedDataFileSequenceNumberTest(
        FILE_B, expectedSnapshotSequenceNumber, expectedFileSequenceNumber);
  }

  private void runRemovedDataFileSequenceNumberTest(
      DataFile fileToRemove, long expectedSnapshotSequenceNumber, long expectedFileSequenceNumber) {
    table.newDelete().deleteFile(fileToRemove).commit();

    Snapshot snapshot = table.currentSnapshot();
    Iterable<DataFile> removedDataFiles =
        SnapshotChanges.builderFor(table).snapshot(snapshot).build().removedDataFiles();
    assertThat(removedDataFiles).as("Must have 1 removed data file").hasSize(1);

    DataFile removedDataFile = Iterables.getOnlyElement(removedDataFiles);

    assertThat(snapshot.sequenceNumber())
        .as("Sequence number mismatch in Snapshot")
        .isEqualTo(expectedSnapshotSequenceNumber);
    assertThat(removedDataFile.dataSequenceNumber().longValue())
        .as("Data sequence number mismatch")
        .isEqualTo(expectedFileSequenceNumber);
    assertThat(removedDataFile.fileSequenceNumber().longValue())
        .as("File sequence number mismatch")
        .isEqualTo(expectedFileSequenceNumber);
  }

  @TestTemplate
  public void testSequenceNumbersInAddedDeleteFiles() {
    assumeThat(formatVersion).as("Delete files only supported in V2").isGreaterThanOrEqualTo(2);

    table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit();

    runAddedDeleteFileSequenceNumberTest(newDeletes(FILE_A), 2);

    runAddedDeleteFileSequenceNumberTest(newDeletes(FILE_B), 3);
  }

  private void runAddedDeleteFileSequenceNumberTest(
      DeleteFile deleteFileToAdd, long expectedSequenceNumber) {
    table.newRowDelta().addDeletes(deleteFileToAdd).commit();

    Snapshot snapshot = table.currentSnapshot();
    Iterable<DeleteFile> addedDeleteFiles =
        SnapshotChanges.builderFor(table).snapshot(snapshot).build().addedDeleteFiles();
    assertThat(addedDeleteFiles).as("Must have 1 added delete file").hasSize(1);

    DeleteFile addedDeleteFile = Iterables.getOnlyElement(addedDeleteFiles);

    assertThat(snapshot.sequenceNumber())
        .as("Sequence number mismatch in Snapshot")
        .isEqualTo(expectedSequenceNumber);
    assertThat(addedDeleteFile.dataSequenceNumber().longValue())
        .as("Data sequence number mismatch")
        .isEqualTo(expectedSequenceNumber);
    assertThat(addedDeleteFile.fileSequenceNumber().longValue())
        .as("File sequence number mismatch")
        .isEqualTo(expectedSequenceNumber);
  }
}
