/*
 * 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.apache.iceberg.expressions.Expressions.bucket;
import static org.apache.iceberg.expressions.Expressions.equal;
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.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.function.Function;
import org.apache.iceberg.DeleteFileIndex.EqualityDeletes;
import org.apache.iceberg.DeleteFileIndex.PositionDeletes;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.CharSequenceSet;
import org.apache.iceberg.util.ContentFileUtil;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

@ExtendWith(ParameterizedTestExtension.class)
public abstract class DeleteFileIndexTestBase<
        ScanT extends Scan<ScanT, T, G>, T extends ScanTask, G extends ScanTaskGroup<T>>
    extends TestBase {

  @Parameters(name = "formatVersion = {0}")
  protected static List<Integer> formatVersions() {
    return TestHelpers.V2_AND_ABOVE;
  }

  static final DeleteFile FILE_A_EQ_1 =
      FileMetadata.deleteFileBuilder(SPEC)
          .ofEqualityDeletes()
          .withPath("/path/to/data-a-eq-deletes.parquet")
          .withFileSizeInBytes(10)
          .withPartition(FILE_A.partition())
          .withRecordCount(1)
          .build();

  private static DataFile unpartitionedFile(PartitionSpec spec) {
    return DataFiles.builder(spec)
        .withPath("/path/to/data-unpartitioned.parquet")
        .withFileSizeInBytes(10)
        .withRecordCount(1)
        .build();
  }

  private static DeleteFile unpartitionedPosDeletes(PartitionSpec spec) {
    return FileMetadata.deleteFileBuilder(spec)
        .ofPositionDeletes()
        .withPath(UUID.randomUUID() + "/path/to/data-unpartitioned-pos-deletes.parquet")
        .withFileSizeInBytes(10)
        .withRecordCount(1)
        .build();
  }

  private static DeleteFile partitionedPosDeletes(PartitionSpec spec, StructLike partition) {
    return FileMetadata.deleteFileBuilder(spec)
        .ofPositionDeletes()
        .withPartition(partition)
        .withPath(UUID.randomUUID() + "/path/to/data-partitioned-pos-deletes.parquet")
        .withFileSizeInBytes(10)
        .withRecordCount(1)
        .build();
  }

  private static DeleteFile posDeletesWithMetrics(PartitionSpec spec, String path) {
    return FileMetadata.deleteFileBuilder(spec)
        .ofPositionDeletes()
        .withPath(UUID.randomUUID() + "/path/to/data-partitioned-pos-deletes.parquet")
        .withFileSizeInBytes(10)
        .withRecordCount(1)
        .withMetrics(
            new Metrics(
                1L,
                ImmutableMap.of(1, 1L, MetadataColumns.DELETE_FILE_PATH.fieldId(), 2L),
                ImmutableMap.of(1, 1L, MetadataColumns.DELETE_FILE_PATH.fieldId(), 2L),
                ImmutableMap.of(1, 1L, MetadataColumns.DELETE_FILE_PATH.fieldId(), 2L),
                ImmutableMap.of(1, 1L, MetadataColumns.DELETE_FILE_PATH.fieldId(), 2L),
                ImmutableMap.of(
                    1,
                    ByteBuffer.wrap(new byte[10]),
                    MetadataColumns.DELETE_FILE_PATH.fieldId(),
                    ByteBuffer.wrap(path.getBytes(StandardCharsets.UTF_8))),
                ImmutableMap.of(
                    1,
                    ByteBuffer.wrap(new byte[10]),
                    MetadataColumns.DELETE_FILE_PATH.fieldId(),
                    ByteBuffer.wrap(path.getBytes(StandardCharsets.UTF_8)))))
        .build();
  }

  private static DeleteFile eqDeletesWithMetrics(
      PartitionSpec spec, StructLike partition, int fieldId) {
    return FileMetadata.deleteFileBuilder(spec)
        .ofEqualityDeletes(fieldId)
        .withPartition(partition)
        .withPath(UUID.randomUUID() + "/path/to/data-partitioned-eq-deletes.parquet")
        .withFileSizeInBytes(10)
        .withRecordCount(1)
        .withMetrics(
            new Metrics(
                1L,
                ImmutableMap.of(1, 1L, 2, 2L),
                ImmutableMap.of(1, 1L, 2, 2L),
                ImmutableMap.of(1, 1L, 2, 2L),
                ImmutableMap.of(1, 1L, 2, 2L),
                ImmutableMap.of(1, ByteBuffer.wrap(new byte[10]), 2, ByteBuffer.wrap(new byte[20])),
                ImmutableMap.of(
                    1, ByteBuffer.wrap(new byte[10]), 2, ByteBuffer.wrap(new byte[20]))))
        .build();
  }

  private static DeleteFile unpartitionedEqDeletes(PartitionSpec spec) {
    return FileMetadata.deleteFileBuilder(spec)
        .ofEqualityDeletes()
        .withPath(UUID.randomUUID() + "/path/to/data-unpartitioned-eq-deletes.parquet")
        .withFileSizeInBytes(10)
        .withRecordCount(1)
        .build();
  }

  private static DeleteFile partitionedEqDeletes(PartitionSpec spec, StructLike partition) {
    return FileMetadata.deleteFileBuilder(spec)
        .ofEqualityDeletes()
        .withPartition(partition)
        .withPath(UUID.randomUUID() + "/path/to/data-partitioned-eq-deletes.parquet")
        .withFileSizeInBytes(10)
        .withRecordCount(1)
        .build();
  }

  @SuppressWarnings("unchecked")
  private static <F extends ContentFile<F>> F withDataSequenceNumber(long seq, F file) {
    BaseFile<F> baseFile = (BaseFile<F>) file;
    baseFile.setDataSequenceNumber(seq);
    return file;
  }

  protected abstract ScanT newScan(Table table);

  @TestTemplate
  public void testMinSequenceNumberFilteringForFiles() {
    PartitionSpec partSpec = PartitionSpec.unpartitioned();

    DeleteFile[] deleteFiles = {
      withDataSequenceNumber(4, unpartitionedEqDeletes(partSpec)),
      withDataSequenceNumber(6, unpartitionedEqDeletes(partSpec))
    };

    DeleteFileIndex index =
        DeleteFileIndex.builderFor(Arrays.asList(deleteFiles))
            .specsById(ImmutableMap.of(partSpec.specId(), partSpec, 1, SPEC))
            .afterSequenceNumber(4)
            .build();

    DataFile file = unpartitionedFile(partSpec);

    assertThat(index.hasEqualityDeletes()).isTrue();
    assertThat(index.hasPositionDeletes()).isFalse();
    assertThat(index.forDataFile(0, file)).as("Only one delete file should apply").hasSize(1);
  }

  @TestTemplate
  public void testUnpartitionedDeletes() {
    PartitionSpec partSpec = PartitionSpec.unpartitioned();

    DeleteFile[] deleteFiles = {
      withDataSequenceNumber(4, unpartitionedEqDeletes(partSpec)),
      withDataSequenceNumber(6, unpartitionedEqDeletes(partSpec)),
      withDataSequenceNumber(5, unpartitionedPosDeletes(partSpec)),
      withDataSequenceNumber(6, unpartitionedPosDeletes(partSpec))
    };

    DeleteFileIndex index =
        DeleteFileIndex.builderFor(Arrays.asList(deleteFiles))
            .specsById(ImmutableMap.of(partSpec.specId(), partSpec, 1, SPEC))
            .build();

    assertThat(index.hasEqualityDeletes()).isTrue();
    assertThat(index.hasPositionDeletes()).isTrue();

    DataFile unpartitionedFile = unpartitionedFile(partSpec);
    assertThat(index.forDataFile(0, unpartitionedFile))
        .as("All deletes should apply to seq 0")
        .isEqualTo(deleteFiles);

    assertThat(index.forDataFile(3, unpartitionedFile))
        .as("All deletes should apply to seq 3")
        .isEqualTo(deleteFiles);

    assertThat(index.forDataFile(4, unpartitionedFile))
        .as("Last 3 deletes should apply to seq 4")
        .isEqualTo(Arrays.copyOfRange(deleteFiles, 1, 4));

    assertThat(index.forDataFile(5, unpartitionedFile))
        .as("Last 3 deletes should apply to seq 5")
        .isEqualTo(Arrays.copyOfRange(deleteFiles, 1, 4));

    assertThat(index.forDataFile(6, unpartitionedFile))
        .as("Last delete should apply to seq 6")
        .isEqualTo(Arrays.copyOfRange(deleteFiles, 3, 4));

    assertThat(index.forDataFile(7, unpartitionedFile))
        .as("No deletes should apply to seq 7")
        .isEqualTo(new DataFile[0]);

    assertThat(index.forDataFile(10, unpartitionedFile))
        .as("No deletes should apply to seq 10")
        .isEqualTo(new DataFile[0]);

    // copy file A with a different spec ID
    DataFile partitionedFileA = FILE_A.copy();
    ((BaseFile<?>) partitionedFileA).setSpecId(1);
    assertThat(index.forDataFile(0, partitionedFileA))
        .as("All global equality deletes should apply to a partitioned file")
        .isEqualTo(Arrays.copyOfRange(deleteFiles, 0, 2));
  }

  @TestTemplate
  public void testPartitionedDeleteIndex() {
    DeleteFile[] deleteFiles = {
      withDataSequenceNumber(4, partitionedEqDeletes(SPEC, FILE_A.partition())),
      withDataSequenceNumber(6, partitionedEqDeletes(SPEC, FILE_A.partition())),
      withDataSequenceNumber(5, partitionedPosDeletes(SPEC, FILE_A.partition())),
      withDataSequenceNumber(6, partitionedPosDeletes(SPEC, FILE_A.partition()))
    };

    DeleteFileIndex index =
        DeleteFileIndex.builderFor(Arrays.asList(deleteFiles))
            .specsById(ImmutableMap.of(SPEC.specId(), SPEC, 1, PartitionSpec.unpartitioned()))
            .build();

    assertThat(index.hasEqualityDeletes()).isTrue();
    assertThat(index.hasPositionDeletes()).isTrue();

    assertThat(index.forDataFile(0, FILE_A))
        .as("All deletes should apply to seq 0")
        .isEqualTo(deleteFiles);

    assertThat(index.forDataFile(3, FILE_A))
        .as("All deletes should apply to seq 3")
        .isEqualTo(deleteFiles);

    assertThat(index.forDataFile(4, FILE_A))
        .as("Last 3 deletes should apply to seq 4")
        .isEqualTo(Arrays.copyOfRange(deleteFiles, 1, 4));

    assertThat(index.forDataFile(5, FILE_A))
        .as("Last 3 deletes should apply to seq 5")
        .isEqualTo(Arrays.copyOfRange(deleteFiles, 1, 4));

    assertThat(index.forDataFile(6, FILE_A))
        .as("Last delete should apply to seq 6")
        .isEqualTo(Arrays.copyOfRange(deleteFiles, 3, 4));

    assertThat(index.forDataFile(7, FILE_A))
        .as("No deletes should apply to seq 7")
        .isEqualTo(new DataFile[0]);

    assertThat(index.forDataFile(10, FILE_A))
        .as("No deletes should apply to seq 10")
        .isEqualTo(new DataFile[0]);

    assertThat(index.forDataFile(0, FILE_B))
        .as("No deletes should apply to FILE_B, partition not in index")
        .hasSize(0);

    assertThat(index.forDataFile(0, FILE_C))
        .as("No deletes should apply to FILE_C, no indexed delete files")
        .hasSize(0);

    DataFile unpartitionedFileA = FILE_A.copy();
    ((BaseFile<?>) unpartitionedFileA).setSpecId(1);
    assertThat(index.forDataFile(0, unpartitionedFileA))
        .as("No deletes should apply to FILE_A with a different specId")
        .hasSize(0);
  }

  @TestTemplate
  public void testUnpartitionedTableScan() throws IOException {
    assumeThat(formatVersion).as("Requires V2 position deletes").isEqualTo(2);
    Table unpartitioned =
        TestTables.create(tableDir, "unpartitioned", SCHEMA, PartitionSpec.unpartitioned(), 2);

    DataFile unpartitionedFile = unpartitionedFile(unpartitioned.spec());
    unpartitioned.newAppend().appendFile(unpartitionedFile).commit();

    // add a delete file
    DeleteFile unpartitionedPosDeletes = unpartitionedPosDeletes(unpartitioned.spec());
    unpartitioned.newRowDelta().addDeletes(unpartitionedPosDeletes).commit();

    List<T> tasks = Lists.newArrayList(newScan(unpartitioned).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(unpartitionedFile.location());
    assertThat(task.deletes()).as("Should have one associated delete file").hasSize(1);
    assertThat(task.deletes().get(0).location())
        .as("Should have expected delete file")
        .isEqualTo(unpartitionedPosDeletes.location());

    // add a second delete file
    DeleteFile unpartitionedEqDeletes = unpartitionedEqDeletes(unpartitioned.spec());
    unpartitioned.newRowDelta().addDeletes(unpartitionedEqDeletes).commit();

    tasks = Lists.newArrayList(newScan(unpartitioned).planFiles().iterator());
    task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(unpartitionedFile.location());
    assertThat(task.deletes()).as("Should have two associated delete files").hasSize(2);
    assertThat(Sets.newHashSet(Iterables.transform(task.deletes(), ContentFile::location)))
        .as("Should have expected delete files")
        .isEqualTo(
            Sets.newHashSet(unpartitionedPosDeletes.location(), unpartitionedEqDeletes.location()));
  }

  @TestTemplate
  public void testPartitionedTableWithPartitionPosDeletes() {
    table.newAppend().appendFile(FILE_A).commit();

    table.newRowDelta().addDeletes(fileADeletes()).commit();

    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(FILE_A.location());
    assertThat(task.deletes()).as("Should have one associated delete file").hasSize(1);
    assertThat(task.deletes().get(0).location())
        .as("Should have only pos delete file")
        .isEqualTo(fileADeletes().location());
  }

  @TestTemplate
  public void testPartitionedTableWithPartitionEqDeletes() {
    table.newAppend().appendFile(FILE_A).commit();

    table.newRowDelta().addDeletes(FILE_A_EQ_1).commit();

    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(FILE_A.location());
    assertThat(task.deletes()).as("Should have one associated delete file").hasSize(1);
    assertThat(task.deletes().get(0).location())
        .as("Should have only pos delete file")
        .isEqualTo(FILE_A_EQ_1.location());
  }

  @TestTemplate
  public void testPartitionedTableWithUnrelatedPartitionDeletes() {
    table.newAppend().appendFile(FILE_B).commit();

    table.newRowDelta().addDeletes(fileADeletes()).addDeletes(FILE_A_EQ_1).commit();

    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(FILE_B.location());
    assertThat(task.deletes()).as("Should have no delete files to apply").hasSize(0);
  }

  @TestTemplate
  public void testPartitionedTableWithOlderPartitionDeletes() {
    assumeThat(formatVersion).as("DVs are not filtered using sequence numbers").isEqualTo(2);

    table.newRowDelta().addDeletes(FILE_A_DELETES).addDeletes(FILE_A_EQ_1).commit();

    table.newAppend().appendFile(FILE_A).commit();

    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(FILE_A.location());
    assertThat(task.deletes()).as("Should have no delete files to apply").hasSize(0);
  }

  @TestTemplate
  public void testPartitionedTableScanWithGlobalDeletes() {
    assumeThat(formatVersion).as("Requires V2 position deletes").isEqualTo(2);

    table.newAppend().appendFile(FILE_A).commit();

    TableMetadata base = table.ops().current();
    table.ops().commit(base, base.updatePartitionSpec(PartitionSpec.unpartitioned()));

    // add unpartitioned equality and position deletes, but only equality deletes are global
    DeleteFile unpartitionedEqDeletes = unpartitionedEqDeletes(table.spec());
    table
        .newRowDelta()
        .addDeletes(unpartitionedPosDeletes(table.spec()))
        .addDeletes(unpartitionedEqDeletes)
        .commit();

    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(FILE_A.location());
    assertThat(task.deletes()).as("Should have one associated delete file").hasSize(1);
    assertThat(task.deletes().get(0).location())
        .as("Should have expected delete file")
        .isEqualTo(unpartitionedEqDeletes.location());
  }

  @TestTemplate
  public void testPartitionedTableScanWithGlobalAndPartitionDeletes() {
    assumeThat(formatVersion).as("Requires V2 position deletes").isEqualTo(2);

    table.newAppend().appendFile(FILE_A).commit();

    table.newRowDelta().addDeletes(FILE_A_EQ_1).commit();

    TableMetadata base = table.ops().current();
    table.ops().commit(base, base.updatePartitionSpec(PartitionSpec.unpartitioned()));

    // add unpartitioned equality and position deletes, but only equality deletes are global
    DeleteFile unpartitionedEqDeletes = unpartitionedEqDeletes(table.spec());
    table
        .newRowDelta()
        .addDeletes(unpartitionedPosDeletes(table.spec()))
        .addDeletes(unpartitionedEqDeletes)
        .commit();

    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(FILE_A.location());
    assertThat(task.deletes()).as("Should have two associated delete files").hasSize(2);
    assertThat(Sets.newHashSet(Iterables.transform(task.deletes(), ContentFile::location)))
        .as("Should have expected delete files")
        .isEqualTo(Sets.newHashSet(unpartitionedEqDeletes.location(), FILE_A_EQ_1.location()));
  }

  @TestTemplate
  public void testPartitionedTableSequenceNumbers() {
    table.newRowDelta().addRows(FILE_A).addDeletes(FILE_A_EQ_1).addDeletes(fileADeletes()).commit();

    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(FILE_A.location());
    assertThat(task.deletes()).as("Should have one associated delete file").hasSize(1);
    assertThat(task.deletes().get(0).location())
        .as("Should have only pos delete file")
        .isEqualTo(fileADeletes().location());
  }

  @TestTemplate
  public void testUnpartitionedTableSequenceNumbers() throws IOException {
    assumeThat(formatVersion).as("Requires V2 position deletes").isEqualTo(2);
    Table unpartitioned =
        TestTables.create(tableDir, "unpartitioned", SCHEMA, PartitionSpec.unpartitioned(), 2);

    // add data, pos deletes, and eq deletes in the same sequence number
    // the position deletes will be applied to the data file, but the equality deletes will not
    DataFile unpartitionedFile = unpartitionedFile(unpartitioned.spec());
    DeleteFile unpartitionedPosDeleteFile = unpartitionedPosDeletes(unpartitioned.spec());
    unpartitioned
        .newRowDelta()
        .addRows(unpartitionedFile)
        .addDeletes(unpartitionedPosDeleteFile)
        .addDeletes(unpartitionedEqDeletes(unpartitioned.spec()))
        .commit();

    assertThat(
            unpartitioned
                .currentSnapshot()
                .deleteManifests(unpartitioned.io())
                .get(0)
                .addedFilesCount())
        .as("Table should contain 2 delete files")
        .isEqualTo(2);

    List<FileScanTask> tasks = Lists.newArrayList(unpartitioned.newScan().planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(unpartitionedFile.location());
    assertThat(task.deletes()).as("Should have one associated delete file").hasSize(1);
    assertThat(task.deletes().get(0).location())
        .as("Should have only pos delete file")
        .isEqualTo(unpartitionedPosDeleteFile.location());
  }

  @TestTemplate
  public void testPartitionedTableWithExistingDeleteFile() {
    table.updateProperties().set(TableProperties.MANIFEST_MERGE_ENABLED, "false").commit();

    table.newAppend().appendFile(FILE_A).commit();

    table.newRowDelta().addDeletes(FILE_A_EQ_1).commit();

    table.newRowDelta().addDeletes(fileADeletes()).commit();

    table
        .updateProperties()
        .set(TableProperties.MANIFEST_MIN_MERGE_COUNT, "1")
        .set(TableProperties.MANIFEST_MERGE_ENABLED, "true")
        .commit();

    assertThat(table.currentSnapshot().deleteManifests(table.io()))
        .as("Should have two delete manifests")
        .hasSize(2);

    // merge delete manifests
    table.newAppend().appendFile(FILE_B).commit();

    assertThat(table.currentSnapshot().deleteManifests(table.io()))
        .as("Should have one delete manifest")
        .hasSize(1);

    assertThat(
            table.currentSnapshot().deleteManifests(table.io()).get(0).addedFilesCount().intValue())
        .as("Should have zero added delete file")
        .isEqualTo(0);

    assertThat(
            table
                .currentSnapshot()
                .deleteManifests(table.io())
                .get(0)
                .deletedFilesCount()
                .intValue())
        .as("Should have zero deleted delete file")
        .isEqualTo(0);

    assertThat(
            table
                .currentSnapshot()
                .deleteManifests(table.io())
                .get(0)
                .existingFilesCount()
                .intValue())
        .as("Should have two existing delete files")
        .isEqualTo(2);

    List<T> tasks =
        Lists.newArrayList(
            newScan(table).filter(equal(bucket("data", BUCKETS_NUMBER), 0)).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(FILE_A.location());
    assertThat(task.deletes()).as("Should have two associated delete files").hasSize(2);
    assertThat(Sets.newHashSet(Iterables.transform(task.deletes(), ContentFile::location)))
        .as("Should have expected delete files")
        .isEqualTo(Sets.newHashSet(FILE_A_EQ_1.location(), fileADeletes().location()));
  }

  @TestTemplate
  public void testPositionDeleteDiscardMetrics() {
    assumeThat(formatVersion).isEqualTo(2);

    Table table =
        TestTables.create(tableDir, "partitioned", SCHEMA, PartitionSpec.unpartitioned(), 2);

    table.newAppend().appendFile(FILE_A).commit();

    // add a delete file
    DeleteFile posDeletesWithMetrics = posDeletesWithMetrics(table.spec(), FILE_A.location());
    table.newRowDelta().addDeletes(posDeletesWithMetrics).commit();

    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(FILE_A.location());
    assertThat(task.deletes()).as("Should have one associated delete file").hasSize(1);

    // verify scanned delete file only contains metrics for 'file_path' field
    int fieldId = MetadataColumns.DELETE_FILE_PATH.fieldId();
    DeleteFile deleteFile = task.deletes().get(0);
    assertThat(deleteFile.columnSizes()).containsExactly(Map.entry(fieldId, 2L));
    assertThat(deleteFile.valueCounts()).containsExactly(Map.entry(fieldId, 2L));
    assertThat(deleteFile.nullValueCounts()).containsExactly(Map.entry(fieldId, 2L));
    assertThat(deleteFile.nanValueCounts()).containsExactly(Map.entry(fieldId, 2L));
    assertThat(deleteFile.lowerBounds())
        .containsExactly(
            Map.entry(
                fieldId, ByteBuffer.wrap(FILE_A.location().getBytes(StandardCharsets.UTF_8))));
    assertThat(deleteFile.upperBounds())
        .containsExactly(
            Map.entry(
                fieldId, ByteBuffer.wrap(FILE_A.location().getBytes(StandardCharsets.UTF_8))));
  }

  @TestTemplate
  public void testEqualityDeleteDiscardMetrics() {
    Table table = TestTables.create(tableDir, "partitioned", SCHEMA, SPEC, formatVersion);

    table.newAppend().appendFile(FILE_A).commit();

    // add a delete file
    int fieldId = 2;
    DeleteFile eqDeletesWithMetrics = eqDeletesWithMetrics(SPEC, FILE_A.partition(), fieldId);
    table.newRowDelta().addDeletes(eqDeletesWithMetrics).commit();

    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
    assertThat(tasks).as("Should have one task").hasSize(1);

    FileScanTask task = (FileScanTask) tasks.get(0);
    assertThat(task.file().location())
        .as("Should have the correct data file path")
        .isEqualTo(FILE_A.location());
    assertThat(task.deletes()).as("Should have one associated delete file").hasSize(1);

    // verify scanned delete file only contains metrics for equality fields
    DeleteFile deleteFile = task.deletes().get(0);
    assertThat(deleteFile.columnSizes()).containsExactly(Map.entry(fieldId, 2L));
    assertThat(deleteFile.valueCounts()).containsExactly(Map.entry(fieldId, 2L));
    assertThat(deleteFile.nullValueCounts()).containsExactly(Map.entry(fieldId, 2L));
    assertThat(deleteFile.nanValueCounts()).containsExactly(Map.entry(fieldId, 2L));
    assertThat(deleteFile.lowerBounds())
        .containsExactly(Map.entry(fieldId, ByteBuffer.wrap(new byte[20])));
    assertThat(deleteFile.upperBounds())
        .containsExactly(Map.entry(fieldId, ByteBuffer.wrap(new byte[20])));
  }

  @TestTemplate
  public void testPositionDeletesGroup() {
    DeleteFile file1 = withDataSequenceNumber(1, partitionedPosDeletes(SPEC, FILE_A.partition()));
    DeleteFile file2 = withDataSequenceNumber(2, partitionedPosDeletes(SPEC, FILE_A.partition()));
    DeleteFile file3 = withDataSequenceNumber(3, partitionedPosDeletes(SPEC, FILE_A.partition()));
    DeleteFile file4 = withDataSequenceNumber(4, partitionedPosDeletes(SPEC, FILE_A.partition()));

    PositionDeletes group = new PositionDeletes();
    group.add(file4);
    group.add(file2);
    group.add(file1);
    group.add(file3);

    // the group must not be empty
    assertThat(group.isEmpty()).isFalse();

    // all files must be reported as referenced
    CharSequenceSet paths =
        CharSequenceSet.of(
            Iterables.transform(group.referencedDeleteFiles(), ContentFile::location));
    assertThat(paths)
        .contains(file1.location(), file2.location(), file3.location(), file4.location());

    // position deletes are indexed by their data sequence numbers
    // so that position deletes can apply to data files added in the same snapshot
    assertThat(group.filter(0)).isEqualTo(new DeleteFile[] {file1, file2, file3, file4});
    assertThat(group.filter(1)).isEqualTo(new DeleteFile[] {file1, file2, file3, file4});
    assertThat(group.filter(2)).isEqualTo(new DeleteFile[] {file2, file3, file4});
    assertThat(group.filter(3)).isEqualTo(new DeleteFile[] {file3, file4});
    assertThat(group.filter(4)).isEqualTo(new DeleteFile[] {file4});
    assertThat(group.filter(5)).isEqualTo(new DeleteFile[] {});

    // it should not be possible to add more elements upon indexing
    assertThatThrownBy(() -> group.add(file1))
        .isInstanceOf(IllegalStateException.class)
        .hasMessage("Can't add files upon indexing");
  }

  @TestTemplate
  public void testEqualityDeletesGroup() {
    DeleteFile file1 = withDataSequenceNumber(1, partitionedEqDeletes(SPEC, FILE_A.partition()));
    DeleteFile file2 = withDataSequenceNumber(2, partitionedEqDeletes(SPEC, FILE_A.partition()));
    DeleteFile file3 = withDataSequenceNumber(3, partitionedEqDeletes(SPEC, FILE_A.partition()));
    DeleteFile file4 = withDataSequenceNumber(4, partitionedEqDeletes(SPEC, FILE_A.partition()));

    Function<Integer, Types.NestedField> fieldLookup = SCHEMA::findField;
    EqualityDeletes group = new EqualityDeletes(fieldLookup);
    group.add(file4);
    group.add(file2);
    group.add(file1);
    group.add(file3);

    // the group must not be empty
    assertThat(group.isEmpty()).isFalse();

    // all files must be reported as referenced
    CharSequenceSet paths =
        CharSequenceSet.of(
            Iterables.transform(group.referencedDeleteFiles(), ContentFile::location));
    assertThat(paths)
        .contains(file1.location(), file2.location(), file3.location(), file4.location());

    // equality deletes are indexed by data sequence number - 1 to apply to next snapshots
    assertThat(group.filter(0, FILE_A)).isEqualTo(new DeleteFile[] {file1, file2, file3, file4});
    assertThat(group.filter(1, FILE_A)).isEqualTo(new DeleteFile[] {file2, file3, file4});
    assertThat(group.filter(2, FILE_A)).isEqualTo(new DeleteFile[] {file3, file4});
    assertThat(group.filter(3, FILE_A)).isEqualTo(new DeleteFile[] {file4});
    assertThat(group.filter(4, FILE_A)).isEqualTo(new DeleteFile[] {});

    // it should not be possible to add more elements upon indexing
    assertThatThrownBy(() -> group.add(file1))
        .isInstanceOf(IllegalStateException.class)
        .hasMessage("Can't add files upon indexing");
  }

  @TestTemplate
  public void testMixDeleteFilesAndDVs() {
    assumeThat(formatVersion).isGreaterThanOrEqualTo(3);

    List<DeleteFile> deletes =
        Arrays.asList(
            withDataSequenceNumber(1, partitionedPosDeletes(SPEC, FILE_A.partition())),
            withDataSequenceNumber(2, newDV(FILE_A)),
            withDataSequenceNumber(1, partitionedPosDeletes(SPEC, FILE_B.partition())),
            withDataSequenceNumber(2, partitionedPosDeletes(SPEC, FILE_B.partition())));

    DeleteFileIndex index = DeleteFileIndex.builderFor(deletes).specsById(table.specs()).build();

    DeleteFile[] fileADeletes = index.forDataFile(0, FILE_A);
    assertThat(fileADeletes).as("Only DV should apply to FILE_A").hasSize(1);
    assertThat(ContentFileUtil.isDV(fileADeletes[0])).isTrue();
    assertThat(fileADeletes[0].referencedDataFile()).isEqualTo(FILE_A.location());

    DeleteFile[] fileBDeletes = index.forDataFile(0, FILE_B);
    assertThat(fileBDeletes).as("Two delete files should apply to FILE_B").hasSize(2);
    assertThat(ContentFileUtil.isDV(fileBDeletes[0])).isFalse();
    assertThat(ContentFileUtil.isDV(fileBDeletes[1])).isFalse();
  }

  @TestTemplate
  public void testMultipleDVs() {
    assumeThat(formatVersion).isGreaterThanOrEqualTo(3);

    DeleteFile dv1 = withDataSequenceNumber(1, newDV(FILE_A));
    DeleteFile dv2 = withDataSequenceNumber(2, newDV(FILE_A));
    List<DeleteFile> dvs = Arrays.asList(dv1, dv2);

    assertThatThrownBy(() -> DeleteFileIndex.builderFor(dvs).specsById(table.specs()).build())
        .isInstanceOf(ValidationException.class)
        .hasMessageContaining("Can't index multiple DVs for %s", FILE_A.location());
  }

  @TestTemplate
  public void testInvalidDVSequenceNumber() {
    assumeThat(formatVersion).isGreaterThanOrEqualTo(3);
    DeleteFile dv = withDataSequenceNumber(1, newDV(FILE_A));
    List<DeleteFile> dvs = Collections.singletonList(dv);
    DeleteFileIndex index = DeleteFileIndex.builderFor(dvs).specsById(table.specs()).build();
    assertThatThrownBy(() -> index.forDataFile(2, FILE_A))
        .isInstanceOf(ValidationException.class)
        .hasMessageContaining("must be greater than or equal to data file sequence number");
  }
}
