/*
 * 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 java.io.IOException;
import java.util.List;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.types.TypeUtil;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

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

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

    Table entriesTable = new ManifestEntriesTable(table);

    Schema readSchema = ManifestEntry.getSchema(table.spec().partitionType());
    Schema expectedSchema =
        TypeUtil.join(readSchema, MetricsUtil.readableMetricsSchema(table.schema(), readSchema));

    assertThat(entriesTable.schema().asStruct())
        .as("A tableScan.select() should prune the schema")
        .isEqualTo(expectedSchema.asStruct());
  }

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

    Table entriesTable = new ManifestEntriesTable(table);
    TableScan scan = entriesTable.newScan();

    Schema readSchema = ManifestEntry.getSchema(table.spec().partitionType());
    Schema expectedSchema =
        TypeUtil.join(readSchema, MetricsUtil.readableMetricsSchema(table.schema(), readSchema));

    assertThat(scan.schema().asStruct())
        .as("A tableScan.select() should prune the schema")
        .isEqualTo(expectedSchema.asStruct());

    FileScanTask file = Iterables.getOnlyElement(scan.planFiles());
    assertThat(file.file().location())
        .as("Data file should be the table's manifest")
        .isEqualTo(table.currentSnapshot().allManifests(table.io()).get(0).path());

    assertThat(file.file().recordCount()).as("Should contain 2 data file records").isEqualTo(2);
  }

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

    table.newAppend().appendFile(FILE_C).appendFile(FILE_D).commit();

    // set the split size to a large value so that both manifests are in 1 split
    table
        .updateProperties()
        .set(TableProperties.METADATA_SPLIT_SIZE, String.valueOf(128 * 1024 * 1024))
        .commit();

    Table entriesTable = new ManifestEntriesTable(table);

    assertThat(entriesTable.newScan().planTasks()).hasSize(1);

    // set the split size to a small value so that manifests end up in different splits
    table.updateProperties().set(TableProperties.METADATA_SPLIT_SIZE, String.valueOf(1)).commit();

    assertThat(entriesTable.newScan().planTasks()).hasSize(2);

    // override the table property with a large value so that both manifests are in 1 split
    TableScan scan =
        entriesTable
            .newScan()
            .option(TableProperties.SPLIT_SIZE, String.valueOf(128 * 1024 * 1024));

    assertThat(scan.planTasks()).hasSize(1);
  }

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

    int splitSize =
        (int) TableProperties.METADATA_SPLIT_SIZE_DEFAULT; // default split size is 32 MB

    Table entriesTable = new ManifestEntriesTable(table);
    assertThat(entriesTable.currentSnapshot().allManifests(table.io())).hasSize(1);

    int expectedSplits =
        ((int) entriesTable.currentSnapshot().allManifests(table.io()).get(0).length()
                + splitSize
                - 1)
            / splitSize;

    TableScan scan = entriesTable.newScan();

    assertThat(scan.planTasks()).hasSize(expectedSplits);
  }

  @TestTemplate
  public void entriesTableWithDeleteManifests() {
    assumeThat(formatVersion).as("Only V2 Tables Support Deletes").isGreaterThanOrEqualTo(2);
    table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();

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

    Table entriesTable = new ManifestEntriesTable(table);
    TableScan scan = entriesTable.newScan();

    Schema readSchema = ManifestEntry.getSchema(table.spec().partitionType());
    Schema expectedSchema =
        TypeUtil.join(readSchema, MetricsUtil.readableMetricsSchema(table.schema(), readSchema));

    assertThat(scan.schema().asStruct())
        .as("A tableScan.select() should prune the schema")
        .isEqualTo(expectedSchema.asStruct());

    List<FileScanTask> files = ImmutableList.copyOf(scan.planFiles());
    assertThat(files.get(0).file().location())
        .as("Data file should be the table's manifest")
        .isEqualTo(table.currentSnapshot().dataManifests(table.io()).get(0).path());
    assertThat(files.get(0).file().recordCount())
        .as("Should contain 2 data file records")
        .isEqualTo(2);
    assertThat(files.get(1).file().location())
        .as("Delete file should be in the table manifest")
        .isEqualTo(table.currentSnapshot().deleteManifests(table.io()).get(0).path());
    assertThat(files.get(1).file().recordCount())
        .as("Should contain 1 delete file record")
        .isEqualTo(1);
  }

  @TestTemplate
  public void ignoreResidualsPreservesManifestContentPruning() {
    assumeThat(formatVersion).as("Only V2 Tables Support Deletes").isGreaterThanOrEqualTo(2);

    table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
    table.newRowDelta().addDeletes(fileADeletes()).commit();

    String dataManifestPath = table.currentSnapshot().dataManifests(table.io()).get(0).path();
    String deleteManifestPath = table.currentSnapshot().deleteManifests(table.io()).get(0).path();

    List<Table> entriesTables =
        ImmutableList.of(new ManifestEntriesTable(table), new AllEntriesTable(table));

    for (Table entriesTable : entriesTables) {
      TableScan scan =
          entriesTable
              .newScan()
              .filter(Expressions.equal("data_file.content", FileContent.POSITION_DELETES.id()))
              .ignoreResiduals();

      List<FileScanTask> tasks = ImmutableList.copyOf(scan.planFiles());

      assertThat(tasks)
          .as("Should retain manifest content pruning for %s", entriesTable.name())
          .extracting(task -> task.file().location())
          .contains(deleteManifestPath)
          .doesNotContain(dataManifestPath);

      assertThat(tasks)
          .as("Should ignore residuals for %s", entriesTable.name())
          .allSatisfy(task -> assertThat(task.residual()).isEqualTo(Expressions.alwaysTrue()));
    }
  }

  @TestTemplate
  public void invalidContentPredicatesPruneAllManifests() throws IOException {
    assumeThat(formatVersion).as("Only V2+ tables support delete files").isGreaterThanOrEqualTo(2);

    table.newAppend().appendFile(FILE_A).commit();
    table.newRowDelta().addDeletes(fileADeletes()).commit();

    List<Table> metadataTables =
        ImmutableList.of(new ManifestEntriesTable(table), new AllEntriesTable(table));

    List<Expression> filters =
        ImmutableList.of(
            Expressions.equal("data_file.content", 3), Expressions.in("data_file.content", 3, 4));

    for (Table metadataTable : metadataTables) {
      for (Expression filter : filters) {
        try (CloseableIterable<FileScanTask> tasks =
            metadataTable.newScan().filter(filter).planFiles()) {
          assertThat(tasks)
              .as("Should not plan manifests for %s with filter %s", metadataTable.name(), filter)
              .isEmpty();
        }
      }
    }
  }
}
