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

import java.io.File;
import java.net.URI;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.iceberg.ManifestEntry.Status;
import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.junit.jupiter.api.TestTemplate;

public class TestSequenceNumberForV2Table extends TestBase {

  @Parameters(name = "formatVersion = {0}")
  protected static List<Object> parameters() {
    return Arrays.asList(2);
  }

  @TestTemplate
  public void testRewrite() {
    table.newFastAppend().appendFile(FILE_A).commit();
    Snapshot snap1 = table.currentSnapshot();
    long commitId1 = snap1.snapshotId();
    ManifestFile manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(null, snap1, 1, FILE_A);
    validateManifest(manifestFile, dataSeqs(1L), fileSeqs(1L), ids(commitId1), files(FILE_A));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap1.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());

    table.newFastAppend().appendFile(FILE_B).commit();
    Snapshot snap2 = table.currentSnapshot();
    long commitId2 = snap2.snapshotId();
    manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(snap1, snap2, 2, FILE_B);
    validateManifest(manifestFile, dataSeqs(2L), fileSeqs(2L), ids(commitId2), files(FILE_B));
    V2Assert.assertEquals("Snapshot sequence number should be 2", 2, snap2.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());

    table.rewriteManifests().clusterBy(file -> "").commit();
    Snapshot snap3 = table.currentSnapshot();
    ManifestFile newManifest =
        snap3.allManifests(table.io()).stream()
            .filter(manifest -> manifest.snapshotId() == snap3.snapshotId())
            .collect(Collectors.toList())
            .get(0);

    V2Assert.assertEquals("Snapshot sequence number should be 3", 3, snap3.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 3", 3, readMetadata().lastSequenceNumber());

    // FILE_A and FILE_B in manifest may reorder
    for (ManifestEntry<DataFile> entry :
        ManifestFiles.read(newManifest, FILE_IO, table.specs()).entries()) {
      if (entry.file().location().equals(FILE_A.location())) {
        V2Assert.assertEquals(
            "FILE_A sequence number should be 1", 1, entry.dataSequenceNumber().longValue());
        V2Assert.assertEquals(
            "FILE_A sequence number should be 1", 1, entry.file().dataSequenceNumber().longValue());
        V2Assert.assertEquals(
            "FILE_A file sequence number should be 1", 1, entry.fileSequenceNumber().longValue());
        V2Assert.assertEquals(
            "FILE_A file sequence number should be 1",
            1,
            entry.file().fileSequenceNumber().longValue());
      }

      if (entry.file().location().equals(FILE_B.location())) {
        V2Assert.assertEquals(
            "FILE_b sequence number should be 2", 2, entry.dataSequenceNumber().longValue());
        V2Assert.assertEquals(
            "FILE_b sequence number should be 2", 2, entry.file().dataSequenceNumber().longValue());
        V2Assert.assertEquals(
            "FILE_B file sequence number should be 2", 2, entry.fileSequenceNumber().longValue());
        V2Assert.assertEquals(
            "FILE_B file sequence number should be 2",
            2,
            entry.file().fileSequenceNumber().longValue());
      }
    }
  }

  @TestTemplate
  public void testCommitConflict() {
    AppendFiles appendA = table.newFastAppend();
    appendA.appendFile(FILE_A).apply();

    table.updateProperties().set(TableProperties.COMMIT_NUM_RETRIES, "0").commit();

    table.ops().failCommits(1);

    assertThatThrownBy(() -> table.newFastAppend().appendFile(FILE_B).commit())
        .isInstanceOf(CommitFailedException.class)
        .hasMessage("Injected failure");

    table.updateProperties().set(TableProperties.COMMIT_NUM_RETRIES, "5").commit();

    appendA.commit();
    Snapshot snap1 = table.currentSnapshot();
    long commitId1 = snap1.snapshotId();
    ManifestFile manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(null, snap1, 1, FILE_A);
    validateManifest(manifestFile, dataSeqs(1L), fileSeqs(1L), ids(commitId1), files(FILE_A));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap1.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());

    AppendFiles appendFiles = table.newFastAppend().appendFile(FILE_C);
    appendFiles.apply();
    table.newFastAppend().appendFile(FILE_D).commit();
    Snapshot snap2 = table.currentSnapshot();
    long commitId2 = snap2.snapshotId();
    manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(snap1, snap2, 2, FILE_D);
    validateManifest(manifestFile, dataSeqs(2L), fileSeqs(2L), ids(commitId2), files(FILE_D));
    V2Assert.assertEquals("Snapshot sequence number should be 2", 2, snap2.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());

    appendFiles.commit();
    Snapshot snap3 = table.currentSnapshot();
    long commitId3 = snap3.snapshotId();
    manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateManifest(manifestFile, dataSeqs(3L), fileSeqs(3L), ids(commitId3), files(FILE_C));
    validateSnapshot(snap2, snap3, 3, FILE_C);
    V2Assert.assertEquals("Snapshot sequence number should be 3", 3, snap3.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 3", 3, readMetadata().lastSequenceNumber());
  }

  @TestTemplate
  public void testRollBack() {
    table.newFastAppend().appendFile(FILE_A).commit();
    Snapshot snap1 = table.currentSnapshot();
    long commitId1 = snap1.snapshotId();
    ManifestFile manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(null, snap1, 1, FILE_A);
    validateManifest(manifestFile, dataSeqs(1L), fileSeqs(1L), ids(commitId1), files(FILE_A));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap1.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());

    table.newFastAppend().appendFile(FILE_B).commit();
    Snapshot snap2 = table.currentSnapshot();
    long commitId2 = snap2.snapshotId();
    manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(snap1, snap2, 2, FILE_B);
    validateManifest(manifestFile, dataSeqs(2L), fileSeqs(2L), ids(commitId2), files(FILE_B));
    V2Assert.assertEquals("Snapshot sequence number should be 2", 2, snap2.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());

    table.manageSnapshots().rollbackTo(commitId1).commit();
    Snapshot snap3 = table.currentSnapshot();
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap3.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());

    table.newFastAppend().appendFile(FILE_C).commit();
    Snapshot snap4 = table.currentSnapshot();
    long commitId4 = snap4.snapshotId();
    manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(snap3, snap4, 3, FILE_C);
    validateManifest(manifestFile, dataSeqs(3L), fileSeqs(3L), ids(commitId4), files(FILE_C));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 3, snap4.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 3", 3, readMetadata().lastSequenceNumber());
  }

  @TestTemplate
  public void testSingleTransaction() {
    Transaction txn = table.newTransaction();
    txn.newAppend().appendFile(FILE_A).commit();
    txn.commitTransaction();
    Snapshot snap = table.currentSnapshot();
    long commitId = snap.snapshotId();
    ManifestFile manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(null, snap, 1, FILE_A);
    validateManifest(manifestFile, dataSeqs(1L), fileSeqs(1L), ids(commitId), files(FILE_A));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());
  }

  @TestTemplate
  public void testConcurrentTransaction() {
    Transaction txn1 = table.newTransaction();
    Transaction txn2 = table.newTransaction();
    Transaction txn3 = table.newTransaction();
    Transaction txn4 = table.newTransaction();

    txn1.newFastAppend().appendFile(FILE_A).commit();
    txn3.newOverwrite().addFile(FILE_C).commit();
    txn4.newDelete().deleteFile(FILE_A).commit();
    txn2.newAppend().appendFile(FILE_B).commit();

    txn1.commitTransaction();
    Snapshot snap1 = table.currentSnapshot();
    long commitId1 = snap1.snapshotId();
    ManifestFile manifestFile1 = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(null, snap1, 1, FILE_A);
    validateManifest(manifestFile1, dataSeqs(1L), fileSeqs(1L), ids(commitId1), files(FILE_A));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap1.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());

    txn2.commitTransaction();
    Snapshot snap2 = table.currentSnapshot();
    long commitId2 = snap2.snapshotId();
    ManifestFile manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(snap1, snap2, 2, FILE_B);
    validateManifest(manifestFile, dataSeqs(2L), fileSeqs(2L), ids(commitId2), files(FILE_B));
    V2Assert.assertEquals("Snapshot sequence number should be 2", 2, snap2.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());

    txn3.commitTransaction();
    Snapshot snap3 = table.currentSnapshot();
    long commitId3 = snap3.snapshotId();
    manifestFile =
        table.currentSnapshot().allManifests(table.io()).stream()
            .filter(manifest -> manifest.snapshotId() == commitId3)
            .collect(Collectors.toList())
            .get(0);
    validateManifest(manifestFile, dataSeqs(3L), fileSeqs(3L), ids(commitId3), files(FILE_C));
    validateSnapshot(snap2, snap3, 3, FILE_C);
    V2Assert.assertEquals("Snapshot sequence number should be 3", 3, snap3.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 3", 3, readMetadata().lastSequenceNumber());

    txn4.commitTransaction();
    Snapshot snap4 = table.currentSnapshot();
    long commitId4 = snap4.snapshotId();
    manifestFile =
        table.currentSnapshot().allManifests(table.io()).stream()
            .filter(manifest -> manifest.snapshotId() == commitId4)
            .collect(Collectors.toList())
            .get(0);
    validateManifest(
        manifestFile,
        dataSeqs(1L),
        fileSeqs(1L),
        ids(commitId4),
        files(FILE_A),
        statuses(Status.DELETED));
    V2Assert.assertEquals("Snapshot sequence number should be 4", 4, snap4.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 4", 4, readMetadata().lastSequenceNumber());
  }

  @TestTemplate
  public void testMultipleOperationsTransaction() {
    Transaction txn = table.newTransaction();
    txn.newFastAppend().appendFile(FILE_A).commit();
    Snapshot snap1 = txn.table().currentSnapshot();
    long commitId1 = snap1.snapshotId();
    ManifestFile manifestFile = snap1.allManifests(table.io()).get(0);
    validateSnapshot(null, snap1, 1, FILE_A);
    validateManifest(manifestFile, dataSeqs(1L), fileSeqs(1L), ids(commitId1), files(FILE_A));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap1.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 0", 0, readMetadata().lastSequenceNumber());

    Set<DataFile> toAddFiles = Sets.newHashSet();
    Set<DataFile> toDeleteFiles = Sets.newHashSet();
    toAddFiles.add(FILE_B);
    toDeleteFiles.add(FILE_A);
    txn.newRewrite().rewriteFiles(toDeleteFiles, toAddFiles).commit();
    txn.commitTransaction();

    Snapshot snap2 = table.currentSnapshot();
    long commitId2 = snap2.snapshotId();
    manifestFile =
        snap2.allManifests(table.io()).stream()
            .filter(manifest -> manifest.snapshotId() == commitId2)
            .collect(Collectors.toList())
            .get(0);

    validateManifest(manifestFile, dataSeqs(2L), fileSeqs(2L), ids(commitId2), files(FILE_B));
    V2Assert.assertEquals("Snapshot sequence number should be 2", 2, snap2.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());
  }

  @TestTemplate
  public void testExpirationInTransaction() {
    table.newFastAppend().appendFile(FILE_A).commit();
    Snapshot snap1 = table.currentSnapshot();
    long commitId1 = snap1.snapshotId();
    ManifestFile manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(null, snap1, 1, FILE_A);
    validateManifest(manifestFile, dataSeqs(1L), fileSeqs(1L), ids(commitId1), files(FILE_A));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap1.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());
    V2Assert.assertEquals(
        "Should be 1 manifest list",
        1,
        listManifestLists(new File(URI.create(table.location()))).size());

    table.newAppend().appendFile(FILE_B).commit();
    Snapshot snap2 = table.currentSnapshot();
    long commitId2 = snap2.snapshotId();
    manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(snap1, snap2, 2, FILE_B);
    validateManifest(manifestFile, dataSeqs(2L), fileSeqs(2L), ids(commitId2), files(FILE_B));
    V2Assert.assertEquals("Snapshot sequence number should be 2", 2, snap2.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());
    V2Assert.assertEquals(
        "Should be 2 manifest lists",
        2,
        listManifestLists(new File(URI.create(table.location()))).size());

    Transaction txn = table.newTransaction();
    txn.expireSnapshots().expireSnapshotId(commitId1).commit();
    txn.commitTransaction();
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());
    V2Assert.assertEquals(
        "Should be 1 manifest list as 1 was deleted",
        1,
        listManifestLists(new File(URI.create(table.location()))).size());
  }

  @TestTemplate
  public void testTransactionFailure() {
    table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
    Snapshot snap1 = table.currentSnapshot();
    long commitId1 = snap1.snapshotId();
    ManifestFile manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateSnapshot(null, snap1, 1, FILE_A, FILE_B);
    validateManifest(
        manifestFile,
        dataSeqs(1L, 1L),
        fileSeqs(1L, 1L),
        ids(commitId1, commitId1),
        files(FILE_A, FILE_B));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap1.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());

    table.updateProperties().set(TableProperties.COMMIT_NUM_RETRIES, "0").commit();

    table.ops().failCommits(1);

    Transaction txn = table.newTransaction();
    txn.newAppend().appendFile(FILE_C).commit();

    assertThatThrownBy(txn::commitTransaction)
        .isInstanceOf(CommitFailedException.class)
        .hasMessage("Injected failure");

    V2Assert.assertEquals(
        "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());
  }

  @TestTemplate
  public void testCherryPicking() {
    table.newAppend().appendFile(FILE_A).commit();
    Snapshot snap1 = table.currentSnapshot();
    long commitId1 = snap1.snapshotId();
    ManifestFile manifestFile = snap1.allManifests(table.io()).get(0);
    validateSnapshot(null, snap1, 1, FILE_A);
    validateManifest(manifestFile, dataSeqs(1L), fileSeqs(1L), ids(commitId1), files(FILE_A));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap1.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());

    table.newAppend().appendFile(FILE_B).stageOnly().commit();

    Snapshot snap2 = table.currentSnapshot();
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap2.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());

    // pick the snapshot that's staged but not committed
    Snapshot stagedSnapshot = readMetadata().snapshots().get(1);
    V2Assert.assertEquals(
        "Snapshot sequence number should be 2", 2, stagedSnapshot.sequenceNumber());

    // table has new commit
    table.newAppend().appendFile(FILE_C).commit();

    Snapshot snap3 = table.currentSnapshot();
    long commitId3 = snap3.snapshotId();
    manifestFile = snap3.allManifests(table.io()).get(0);
    validateManifest(manifestFile, dataSeqs(3L), fileSeqs(3L), ids(commitId3), files(FILE_C));
    validateSnapshot(snap2, snap3, 3, FILE_C);
    V2Assert.assertEquals("Snapshot sequence number should be 3", 3, snap3.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 3", 3, readMetadata().lastSequenceNumber());

    // cherry-pick snapshot
    table.manageSnapshots().cherrypick(stagedSnapshot.snapshotId()).commit();
    Snapshot snap4 = table.currentSnapshot();
    long commitId4 = snap4.snapshotId();
    manifestFile = table.currentSnapshot().allManifests(table.io()).get(0);
    validateManifest(manifestFile, dataSeqs(4L), fileSeqs(4L), ids(commitId4), files(FILE_B));
    validateSnapshot(snap3, snap4, 4, FILE_B);
    V2Assert.assertEquals("Snapshot sequence number should be 4", 4, snap4.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 4", 4, readMetadata().lastSequenceNumber());
  }

  @TestTemplate
  public void testCherryPickFastForward() {
    table.newAppend().appendFile(FILE_A).commit();
    Snapshot snap1 = table.currentSnapshot();
    long commitId1 = snap1.snapshotId();
    ManifestFile manifestFile = snap1.allManifests(table.io()).get(0);
    validateSnapshot(null, snap1, 1, FILE_A);
    validateManifest(manifestFile, dataSeqs(1L), fileSeqs(1L), ids(commitId1), files(FILE_A));
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap1.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 1", 1, readMetadata().lastSequenceNumber());

    table.newAppend().appendFile(FILE_B).stageOnly().commit();
    Snapshot snap2 = table.currentSnapshot();
    V2Assert.assertEquals("Snapshot sequence number should be 1", 1, snap2.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());

    // pick the snapshot that's staged but not committed
    Snapshot stagedSnapshot = readMetadata().snapshots().get(1);
    V2Assert.assertEquals(
        "Snapshot sequence number should be 2", 2, stagedSnapshot.sequenceNumber());

    // cherry-pick snapshot, this will fast forward
    table.manageSnapshots().cherrypick(stagedSnapshot.snapshotId()).commit();
    Snapshot snap3 = table.currentSnapshot();
    long commitId3 = snap3.snapshotId();
    manifestFile = snap3.allManifests(table.io()).get(0);
    validateManifest(manifestFile, dataSeqs(2L), fileSeqs(2L), ids(commitId3), files(FILE_B));
    validateSnapshot(snap2, snap3, 2, FILE_B);
    V2Assert.assertEquals("Snapshot sequence number should be 2", 2, snap3.sequenceNumber());
    V2Assert.assertEquals(
        "Last sequence number should be 2", 2, readMetadata().lastSequenceNumber());
  }
}
