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

import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION;
import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION_LEVEL;
import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
import static org.apache.iceberg.TableProperties.DELETE_AVRO_COMPRESSION;
import static org.apache.iceberg.TableProperties.DELETE_AVRO_COMPRESSION_LEVEL;
import static org.apache.iceberg.TableProperties.DELETE_DEFAULT_FILE_FORMAT;
import static org.apache.iceberg.TableProperties.DELETE_DISTRIBUTION_MODE;
import static org.apache.iceberg.TableProperties.DELETE_ORC_COMPRESSION;
import static org.apache.iceberg.TableProperties.DELETE_ORC_COMPRESSION_STRATEGY;
import static org.apache.iceberg.TableProperties.DELETE_PARQUET_COMPRESSION;
import static org.apache.iceberg.TableProperties.DELETE_PARQUET_COMPRESSION_LEVEL;
import static org.apache.iceberg.TableProperties.MERGE_DISTRIBUTION_MODE;
import static org.apache.iceberg.TableProperties.ORC_COMPRESSION;
import static org.apache.iceberg.TableProperties.ORC_COMPRESSION_STRATEGY;
import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION;
import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION_LEVEL;
import static org.apache.iceberg.TableProperties.PARQUET_SHRED_VARIANTS;
import static org.apache.iceberg.TableProperties.UPDATE_DISTRIBUTION_MODE;
import static org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE;
import static org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE_HASH;
import static org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE_NONE;
import static org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE_RANGE;
import static org.apache.iceberg.spark.SparkSQLProperties.COMPRESSION_CODEC;
import static org.apache.iceberg.spark.SparkSQLProperties.COMPRESSION_LEVEL;
import static org.apache.spark.sql.connector.write.RowLevelOperation.Command.DELETE;
import static org.apache.spark.sql.connector.write.RowLevelOperation.Command.MERGE;
import static org.apache.spark.sql.connector.write.RowLevelOperation.Command.UPDATE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import org.apache.iceberg.DistributionMode;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.deletes.DeleteGranularity;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.spark.sql.internal.SQLConf;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

@ExtendWith(ParameterizedTestExtension.class)
public class TestSparkWriteConf extends TestBaseWithCatalog {

  @BeforeEach
  public void before() {
    super.before();
    sql(
        "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) "
            + "USING iceberg "
            + "PARTITIONED BY (date, days(ts))",
        tableName);
  }

  @AfterEach
  public void after() {
    sql("DROP TABLE IF EXISTS %s", tableName);
  }

  @TestTemplate
  public void testOptionCaseInsensitive() {
    Table table = validationCatalog.loadTable(tableIdent);
    Map<String, String> options = ImmutableMap.of("option", "value");
    SparkConfParser parser = new SparkConfParser(spark, table, options);
    String parsedValue = parser.stringConf().option("oPtIoN").parseOptional();
    assertThat(parsedValue).isEqualTo("value");
  }

  @TestTemplate
  public void testCamelCaseSparkSessionConf() {
    Table table = validationCatalog.loadTable(tableIdent);
    String confName = "spark.sql.iceberg.some-int-conf";
    String sparkConfName = "spark.sql.iceberg.someIntConf";

    withSQLConf(
        ImmutableMap.of(sparkConfName, "1"),
        () -> {
          SparkConfParser parser = new SparkConfParser(spark, table, ImmutableMap.of());
          Integer value = parser.intConf().sessionConf(confName).parseOptional();
          assertThat(value).isEqualTo(1);
        });
  }

  @TestTemplate
  public void testCamelCaseSparkOption() {
    Table table = validationCatalog.loadTable(tableIdent);
    String option = "some-int-option";
    String sparkOption = "someIntOption";
    Map<String, String> options = ImmutableMap.of(sparkOption, "1");
    SparkConfParser parser = new SparkConfParser(spark, table, options);
    Integer value = parser.intConf().option(option).parseOptional();
    assertThat(value).isEqualTo(1);
  }

  @TestTemplate
  public void testDurationConf() {
    Table table = validationCatalog.loadTable(tableIdent);
    String confName = "spark.sql.iceberg.some-duration-conf";

    withSQLConf(
        ImmutableMap.of(confName, "10s"),
        () -> {
          SparkConfParser parser = new SparkConfParser(spark, table, ImmutableMap.of());
          Duration duration = parser.durationConf().sessionConf(confName).parseOptional();
          assertThat(duration).hasSeconds(10);
        });

    withSQLConf(
        ImmutableMap.of(confName, "2m"),
        () -> {
          SparkConfParser parser = new SparkConfParser(spark, table, ImmutableMap.of());
          Duration duration = parser.durationConf().sessionConf(confName).parseOptional();
          assertThat(duration).hasMinutes(2);
        });
  }

  @TestTemplate
  public void testDeleteGranularityDefault() {
    Table table = validationCatalog.loadTable(tableIdent);
    SparkWriteConf writeConf = new SparkWriteConf(spark, table);

    DeleteGranularity value = writeConf.deleteGranularity();
    assertThat(value).isEqualTo(DeleteGranularity.FILE);
  }

  @TestTemplate
  public void testDeleteGranularityTableProperty() {
    Table table = validationCatalog.loadTable(tableIdent);

    table
        .updateProperties()
        .set(TableProperties.DELETE_GRANULARITY, DeleteGranularity.PARTITION.toString())
        .commit();

    SparkWriteConf writeConf = new SparkWriteConf(spark, table);

    DeleteGranularity value = writeConf.deleteGranularity();
    assertThat(value).isEqualTo(DeleteGranularity.PARTITION);
  }

  @TestTemplate
  public void testDeleteGranularityWriteOption() {
    Table table = validationCatalog.loadTable(tableIdent);

    table
        .updateProperties()
        .set(TableProperties.DELETE_GRANULARITY, DeleteGranularity.PARTITION.toString())
        .commit();

    Map<String, String> options =
        ImmutableMap.of(SparkWriteOptions.DELETE_GRANULARITY, DeleteGranularity.FILE.toString());

    SparkWriteConf writeConf =
        new SparkWriteConf(spark, table, new CaseInsensitiveStringMap(options));

    DeleteGranularity value = writeConf.deleteGranularity();
    assertThat(value).isEqualTo(DeleteGranularity.FILE);
  }

  @TestTemplate
  public void testDeleteGranularityInvalidValue() {
    Table table = validationCatalog.loadTable(tableIdent);

    table.updateProperties().set(TableProperties.DELETE_GRANULARITY, "invalid").commit();

    SparkWriteConf writeConf = new SparkWriteConf(spark, table);

    assertThatThrownBy(writeConf::deleteGranularity)
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessageContaining("Unknown delete granularity");
  }

  @TestTemplate
  public void testAdvisoryPartitionSize() {
    Table table = validationCatalog.loadTable(tableIdent);

    SparkWriteConf writeConf = new SparkWriteConf(spark, table);

    long value1 = writeConf.writeRequirements().advisoryPartitionSize();
    assertThat(value1).isGreaterThan(64L * 1024 * 1024).isLessThan(2L * 1024 * 1024 * 1024);

    spark.conf().set(SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES().key(), "2GB");
    long value2 = writeConf.writeRequirements().advisoryPartitionSize();
    assertThat(value2).isEqualTo(2L * 1024 * 1024 * 1024);

    spark.conf().set(SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES().key(), "10MB");
    long value3 = writeConf.writeRequirements().advisoryPartitionSize();
    assertThat(value3).isGreaterThan(10L * 1024 * 1024);
  }

  @TestTemplate
  public void testSparkWriteConfDistributionDefault() {
    Table table = validationCatalog.loadTable(tableIdent);

    SparkWriteConf writeConf = new SparkWriteConf(spark, table);

    checkMode(DistributionMode.HASH, writeConf);
  }

  @TestTemplate
  public void testSparkWriteConfDistributionModeWithWriteOption() {
    Table table = validationCatalog.loadTable(tableIdent);

    CaseInsensitiveStringMap writeOptions =
        new CaseInsensitiveStringMap(
            ImmutableMap.of(SparkWriteOptions.DISTRIBUTION_MODE, DistributionMode.NONE.modeName()));

    SparkWriteConf writeConf = new SparkWriteConf(spark, table, writeOptions);
    checkMode(DistributionMode.NONE, writeConf);
  }

  @TestTemplate
  public void testSparkWriteConfDistributionModeWithSessionConfig() {
    withSQLConf(
        ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.NONE.modeName()),
        () -> {
          Table table = validationCatalog.loadTable(tableIdent);
          SparkWriteConf writeConf = new SparkWriteConf(spark, table);
          checkMode(DistributionMode.NONE, writeConf);
        });
  }

  @TestTemplate
  public void testSparkWriteConfDistributionModeWithTableProperties() {
    Table table = validationCatalog.loadTable(tableIdent);

    table
        .updateProperties()
        .set(WRITE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_NONE)
        .set(DELETE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_NONE)
        .set(UPDATE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_NONE)
        .set(MERGE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_NONE)
        .commit();

    SparkWriteConf writeConf = new SparkWriteConf(spark, table);
    checkMode(DistributionMode.NONE, writeConf);
  }

  @TestTemplate
  public void testSparkWriteConfDistributionModeWithTblPropAndSessionConfig() {
    withSQLConf(
        ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.NONE.modeName()),
        () -> {
          Table table = validationCatalog.loadTable(tableIdent);

          table
              .updateProperties()
              .set(WRITE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_RANGE)
              .set(DELETE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_RANGE)
              .set(UPDATE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_RANGE)
              .set(MERGE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_RANGE)
              .commit();

          SparkWriteConf writeConf = new SparkWriteConf(spark, table);
          // session config overwrite the table properties
          checkMode(DistributionMode.NONE, writeConf);
        });
  }

  @TestTemplate
  public void testSparkWriteConfDistributionModeWithWriteOptionAndSessionConfig() {
    withSQLConf(
        ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.RANGE.modeName()),
        () -> {
          Table table = validationCatalog.loadTable(tableIdent);

          CaseInsensitiveStringMap writeOptions =
              new CaseInsensitiveStringMap(
                  ImmutableMap.of(
                      SparkWriteOptions.DISTRIBUTION_MODE, DistributionMode.NONE.modeName()));

          SparkWriteConf writeConf = new SparkWriteConf(spark, table, writeOptions);
          // write options overwrite the session config
          checkMode(DistributionMode.NONE, writeConf);
        });
  }

  @TestTemplate
  public void testSparkWriteConfDistributionModeWithEverything() {
    withSQLConf(
        ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.RANGE.modeName()),
        () -> {
          Table table = validationCatalog.loadTable(tableIdent);

          CaseInsensitiveStringMap writeOptions =
              new CaseInsensitiveStringMap(
                  ImmutableMap.of(
                      SparkWriteOptions.DISTRIBUTION_MODE, DistributionMode.NONE.modeName()));

          table
              .updateProperties()
              .set(WRITE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_HASH)
              .set(DELETE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_HASH)
              .set(UPDATE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_HASH)
              .set(MERGE_DISTRIBUTION_MODE, WRITE_DISTRIBUTION_MODE_HASH)
              .commit();

          SparkWriteConf writeConf = new SparkWriteConf(spark, table, writeOptions);
          // write options take the highest priority
          checkMode(DistributionMode.NONE, writeConf);
        });
  }

  @TestTemplate
  public void testSparkConfOverride() {
    List<List<Map<String, String>>> propertiesSuites =
        Lists.newArrayList(
            Lists.newArrayList(
                ImmutableMap.of(COMPRESSION_CODEC, "zstd", COMPRESSION_LEVEL, "3"),
                ImmutableMap.of(
                    DEFAULT_FILE_FORMAT,
                    "parquet",
                    DELETE_DEFAULT_FILE_FORMAT,
                    "parquet",
                    TableProperties.PARQUET_COMPRESSION,
                    "gzip",
                    TableProperties.DELETE_PARQUET_COMPRESSION,
                    "snappy"),
                ImmutableMap.of(
                    PARQUET_SHRED_VARIANTS,
                    "false",
                    DELETE_PARQUET_COMPRESSION,
                    "zstd",
                    PARQUET_COMPRESSION,
                    "zstd",
                    PARQUET_COMPRESSION_LEVEL,
                    "3",
                    DELETE_PARQUET_COMPRESSION_LEVEL,
                    "3")),
            Lists.newArrayList(
                ImmutableMap.of(
                    COMPRESSION_CODEC,
                    "zstd",
                    SparkSQLProperties.COMPRESSION_STRATEGY,
                    "compression"),
                ImmutableMap.of(
                    DEFAULT_FILE_FORMAT,
                    "orc",
                    DELETE_DEFAULT_FILE_FORMAT,
                    "orc",
                    ORC_COMPRESSION,
                    "zlib",
                    DELETE_ORC_COMPRESSION,
                    "snappy"),
                ImmutableMap.of(
                    DELETE_ORC_COMPRESSION,
                    "zstd",
                    ORC_COMPRESSION,
                    "zstd",
                    DELETE_ORC_COMPRESSION_STRATEGY,
                    "compression",
                    ORC_COMPRESSION_STRATEGY,
                    "compression")),
            Lists.newArrayList(
                ImmutableMap.of(COMPRESSION_CODEC, "zstd", COMPRESSION_LEVEL, "9"),
                ImmutableMap.of(
                    DEFAULT_FILE_FORMAT,
                    "avro",
                    DELETE_DEFAULT_FILE_FORMAT,
                    "avro",
                    AVRO_COMPRESSION,
                    "gzip",
                    DELETE_AVRO_COMPRESSION,
                    "snappy"),
                ImmutableMap.of(
                    DELETE_AVRO_COMPRESSION,
                    "zstd",
                    AVRO_COMPRESSION,
                    "zstd",
                    AVRO_COMPRESSION_LEVEL,
                    "9",
                    DELETE_AVRO_COMPRESSION_LEVEL,
                    "9")));
    for (List<Map<String, String>> propertiesSuite : propertiesSuites) {
      testWriteProperties(propertiesSuite);
    }
  }

  @TestTemplate
  public void testExtraSnapshotMetadataReflectsSessionConfig() {
    withSQLConf(
        ImmutableMap.of("spark.sql.iceberg.snapshot-property.test-key", "session-value"),
        () -> {
          Table table = validationCatalog.loadTable(tableIdent);
          SparkWriteConf writeConf = new SparkWriteConf(spark, table);

          Map<String, String> metadata = writeConf.extraSnapshotMetadata();

          assertThat(metadata).containsEntry("test-key", "session-value");
        });
  }

  @TestTemplate
  public void testExtraSnapshotMetadataWriteOptionsOverrideSessionConfig() {
    withSQLConf(
        ImmutableMap.of("spark.sql.iceberg.snapshot-property.test-key", "session-value"),
        () -> {
          Table table = validationCatalog.loadTable(tableIdent);
          CaseInsensitiveStringMap writeOptions =
              new CaseInsensitiveStringMap(
                  ImmutableMap.of("snapshot-property.test-key", "write-option-value"));
          SparkWriteConf writeConf = new SparkWriteConf(spark, table, writeOptions);

          Map<String, String> metadata = writeConf.extraSnapshotMetadata();

          // Assert that writeOptions take precedence over session config
          assertThat(metadata).containsEntry("test-key", "write-option-value");
        });
  }

  @TestTemplate
  public void testExtraSnapshotMetadataPersistedOnWrite() {
    String propertyKey = "test-key";
    String propertyValue = "session-value";

    withSQLConf(
        ImmutableMap.of("spark.sql.iceberg.snapshot-property." + propertyKey, propertyValue),
        () -> {
          spark.sql(
              String.format(
                  "INSERT INTO %s VALUES (1, 'a', DATE '2021-01-01', TIMESTAMP '2021-01-01 00:00:00')",
                  tableName));
          Table table = validationCatalog.loadTable(tableIdent);
          assertThat(table.currentSnapshot().summary()).containsEntry(propertyKey, propertyValue);
        });
  }

  @TestTemplate
  public void testDataPropsDefaultsAsDeleteProps() {
    List<List<Map<String, String>>> propertiesSuites =
        Lists.newArrayList(
            Lists.newArrayList(
                ImmutableMap.of(),
                ImmutableMap.of(
                    DEFAULT_FILE_FORMAT,
                    "parquet",
                    DELETE_DEFAULT_FILE_FORMAT,
                    "parquet",
                    PARQUET_COMPRESSION,
                    "zstd",
                    PARQUET_COMPRESSION_LEVEL,
                    "5"),
                ImmutableMap.of(
                    PARQUET_SHRED_VARIANTS,
                    "false",
                    DELETE_PARQUET_COMPRESSION,
                    "zstd",
                    PARQUET_COMPRESSION,
                    "zstd",
                    PARQUET_COMPRESSION_LEVEL,
                    "5",
                    DELETE_PARQUET_COMPRESSION_LEVEL,
                    "5")),
            Lists.newArrayList(
                ImmutableMap.of(),
                ImmutableMap.of(
                    DEFAULT_FILE_FORMAT,
                    "orc",
                    DELETE_DEFAULT_FILE_FORMAT,
                    "orc",
                    ORC_COMPRESSION,
                    "snappy",
                    ORC_COMPRESSION_STRATEGY,
                    "speed"),
                ImmutableMap.of(
                    DELETE_ORC_COMPRESSION,
                    "snappy",
                    ORC_COMPRESSION,
                    "snappy",
                    ORC_COMPRESSION_STRATEGY,
                    "speed",
                    DELETE_ORC_COMPRESSION_STRATEGY,
                    "speed")),
            Lists.newArrayList(
                ImmutableMap.of(),
                ImmutableMap.of(
                    DEFAULT_FILE_FORMAT,
                    "avro",
                    DELETE_DEFAULT_FILE_FORMAT,
                    "avro",
                    AVRO_COMPRESSION,
                    "snappy",
                    AVRO_COMPRESSION_LEVEL,
                    "9"),
                ImmutableMap.of(
                    DELETE_AVRO_COMPRESSION,
                    "snappy",
                    AVRO_COMPRESSION,
                    "snappy",
                    AVRO_COMPRESSION_LEVEL,
                    "9",
                    DELETE_AVRO_COMPRESSION_LEVEL,
                    "9")));
    for (List<Map<String, String>> propertiesSuite : propertiesSuites) {
      testWriteProperties(propertiesSuite);
    }
  }

  @TestTemplate
  public void testDeleteFileWriteConf() {
    List<List<Map<String, String>>> propertiesSuites =
        Lists.newArrayList(
            Lists.newArrayList(
                ImmutableMap.of(),
                ImmutableMap.of(
                    DEFAULT_FILE_FORMAT,
                    "parquet",
                    DELETE_DEFAULT_FILE_FORMAT,
                    "parquet",
                    TableProperties.PARQUET_COMPRESSION,
                    "zstd",
                    PARQUET_COMPRESSION_LEVEL,
                    "5",
                    DELETE_PARQUET_COMPRESSION_LEVEL,
                    "6"),
                ImmutableMap.of(
                    PARQUET_SHRED_VARIANTS,
                    "false",
                    DELETE_PARQUET_COMPRESSION,
                    "zstd",
                    PARQUET_COMPRESSION,
                    "zstd",
                    PARQUET_COMPRESSION_LEVEL,
                    "5",
                    DELETE_PARQUET_COMPRESSION_LEVEL,
                    "6")),
            Lists.newArrayList(
                ImmutableMap.of(),
                ImmutableMap.of(
                    DEFAULT_FILE_FORMAT,
                    "orc",
                    DELETE_DEFAULT_FILE_FORMAT,
                    "orc",
                    ORC_COMPRESSION,
                    "snappy",
                    ORC_COMPRESSION_STRATEGY,
                    "speed",
                    DELETE_ORC_COMPRESSION,
                    "zstd",
                    DELETE_ORC_COMPRESSION_STRATEGY,
                    "compression"),
                ImmutableMap.of(
                    DELETE_ORC_COMPRESSION,
                    "zstd",
                    ORC_COMPRESSION,
                    "snappy",
                    ORC_COMPRESSION_STRATEGY,
                    "speed",
                    DELETE_ORC_COMPRESSION_STRATEGY,
                    "compression")),
            Lists.newArrayList(
                ImmutableMap.of(),
                ImmutableMap.of(
                    DEFAULT_FILE_FORMAT,
                    "avro",
                    DELETE_DEFAULT_FILE_FORMAT,
                    "avro",
                    AVRO_COMPRESSION,
                    "snappy",
                    AVRO_COMPRESSION_LEVEL,
                    "9",
                    DELETE_AVRO_COMPRESSION,
                    "zstd",
                    DELETE_AVRO_COMPRESSION_LEVEL,
                    "16"),
                ImmutableMap.of(
                    DELETE_AVRO_COMPRESSION,
                    "zstd",
                    AVRO_COMPRESSION,
                    "snappy",
                    AVRO_COMPRESSION_LEVEL,
                    "9",
                    DELETE_AVRO_COMPRESSION_LEVEL,
                    "16")));
    for (List<Map<String, String>> propertiesSuite : propertiesSuites) {
      testWriteProperties(propertiesSuite);
    }
  }

  @TestTemplate
  public void testDVWriteConf() {
    Table table = validationCatalog.loadTable(tableIdent);
    table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
    SparkWriteConf writeConf = new SparkWriteConf(spark, table);
    assertThat(writeConf.deleteFileFormat()).isEqualTo(FileFormat.PUFFIN);
  }

  @TestTemplate
  public void testSortOrderWriteConf() {
    Table table = validationCatalog.loadTable(tableIdent);

    table.replaceSortOrder().asc("id").commit();

    SparkWriteConf writeConf =
        new SparkWriteConf(
            spark,
            table,
            new CaseInsensitiveStringMap(
                ImmutableMap.of(SparkWriteOptions.OUTPUT_SORT_ORDER_ID, "1")));

    assertThat(writeConf.outputSortOrderId(SparkWriteRequirements.EMPTY))
        .isEqualTo(table.sortOrder().orderId());
  }

  @TestTemplate
  public void testSortOrderWriteConfWithInvalidId() {
    Table table = validationCatalog.loadTable(tableIdent);

    table.replaceSortOrder().asc("id").commit();

    SparkWriteConf writeConfForUnknownSortOrder =
        new SparkWriteConf(
            spark,
            table,
            new CaseInsensitiveStringMap(
                ImmutableMap.of(SparkWriteOptions.OUTPUT_SORT_ORDER_ID, "999")));

    assertThatIllegalArgumentException()
        .isThrownBy(
            () -> writeConfForUnknownSortOrder.outputSortOrderId(SparkWriteRequirements.EMPTY))
        .withMessage(
            "Cannot use output sort order id 999 because the table does not contain a sort order with that id");
  }

  @TestTemplate
  public void testSortOrderWriteConfWithNoOption() {
    Table table = validationCatalog.loadTable(tableIdent);

    table.replaceSortOrder().asc("id").commit();

    SparkWriteConf writeConfNoOption = new SparkWriteConf(spark, table);

    assertThat(writeConfNoOption.outputSortOrderId(writeConfNoOption.writeRequirements()))
        .isEqualTo(table.sortOrder().orderId());

    assertThat(writeConfNoOption.outputSortOrderId(SparkWriteRequirements.EMPTY)).isEqualTo(0);
  }

  private void testWriteProperties(List<Map<String, String>> propertiesSuite) {
    withSQLConf(
        propertiesSuite.get(0),
        () -> {
          Table table = validationCatalog.loadTable(tableIdent);
          Map<String, String> tableProperties = propertiesSuite.get(1);
          UpdateProperties updateProperties = table.updateProperties();
          for (Map.Entry<String, String> entry : tableProperties.entrySet()) {
            updateProperties.set(entry.getKey(), entry.getValue());
          }

          updateProperties.commit();

          SparkWriteConf writeConf = new SparkWriteConf(spark, table);
          Map<String, String> writeProperties = writeConf.writeProperties();
          Map<String, String> expectedProperties = propertiesSuite.get(2);
          assertThat(writeConf.writeProperties()).hasSameSizeAs(expectedProperties);
          for (Map.Entry<String, String> entry : writeProperties.entrySet()) {
            assertThat(expectedProperties).containsEntry(entry.getKey(), entry.getValue());
          }

          table.refresh();
          updateProperties = table.updateProperties();
          for (Map.Entry<String, String> entry : tableProperties.entrySet()) {
            updateProperties.remove(entry.getKey());
          }

          updateProperties.commit();
        });
  }

  private void checkMode(DistributionMode expectedMode, SparkWriteConf writeConf) {
    assertThat(writeConf.distributionMode()).isEqualTo(expectedMode);
    assertThat(writeConf.copyOnWriteDistributionMode(DELETE)).isEqualTo(expectedMode);
    assertThat(writeConf.positionDeltaDistributionMode(DELETE)).isEqualTo(expectedMode);
    assertThat(writeConf.copyOnWriteDistributionMode(UPDATE)).isEqualTo(expectedMode);
    assertThat(writeConf.positionDeltaDistributionMode(UPDATE)).isEqualTo(expectedMode);
    assertThat(writeConf.copyOnWriteDistributionMode(MERGE)).isEqualTo(expectedMode);
    assertThat(writeConf.positionDeltaDistributionMode(MERGE)).isEqualTo(expectedMode);
  }

  @TestTemplate
  public void testShredVariantsDefault() {
    Table table = validationCatalog.loadTable(tableIdent);
    SparkWriteConf writeConf = new SparkWriteConf(spark, table);
    assertThat(writeConf.shredVariants()).isFalse();
  }

  @TestTemplate
  public void testVariantInferenceBufferSizeDefault() {
    Table table = validationCatalog.loadTable(tableIdent);
    SparkWriteConf writeConf = new SparkWriteConf(spark, table);
    assertThat(writeConf.variantInferenceBufferSize())
        .isEqualTo(TableProperties.PARQUET_VARIANT_BUFFER_SIZE_DEFAULT);
  }

  @TestTemplate
  public void testVariantInferenceBufferSizeTableProperty() {
    Table table = validationCatalog.loadTable(tableIdent);

    table.updateProperties().set(TableProperties.PARQUET_VARIANT_BUFFER_SIZE, "500").commit();

    SparkWriteConf writeConf = new SparkWriteConf(spark, table);
    assertThat(writeConf.variantInferenceBufferSize()).isEqualTo(500);
  }

  @TestTemplate
  public void testShredVariantsSessionOverridesTableProperty() {
    Table table = validationCatalog.loadTable(tableIdent);
    table.updateProperties().set(TableProperties.PARQUET_SHRED_VARIANTS, "false").commit();

    withSQLConf(
        ImmutableMap.of(SparkSQLProperties.SHRED_VARIANTS, "true"),
        () -> {
          SparkWriteConf writeConf = new SparkWriteConf(spark, table);
          assertThat(writeConf.shredVariants()).isTrue();
        });
  }

  @TestTemplate
  public void testShredVariantsWriteOptionOverridesSessionConf() {
    withSQLConf(
        ImmutableMap.of(SparkSQLProperties.SHRED_VARIANTS, "false"),
        () -> {
          Table table = validationCatalog.loadTable(tableIdent);
          SparkWriteConf writeConf =
              new SparkWriteConf(
                  spark,
                  table,
                  new CaseInsensitiveStringMap(
                      ImmutableMap.of(SparkWriteOptions.SHRED_VARIANTS, "true")));
          assertThat(writeConf.shredVariants()).isTrue();
        });
  }

  @TestTemplate
  public void testVariantInferenceBufferSizeSessionConf() {
    withSQLConf(
        ImmutableMap.of(SparkSQLProperties.VARIANT_INFERENCE_BUFFER_SIZE, "250"),
        () -> {
          Table table = validationCatalog.loadTable(tableIdent);
          SparkWriteConf writeConf = new SparkWriteConf(spark, table);
          assertThat(writeConf.variantInferenceBufferSize()).isEqualTo(250);
        });
  }

  @TestTemplate
  public void testWritePropertiesIncludeVariantShredding() {
    Table table = validationCatalog.loadTable(tableIdent);
    table.updateProperties().set(TableProperties.PARQUET_SHRED_VARIANTS, "true").commit();
    table.updateProperties().set(TableProperties.PARQUET_VARIANT_BUFFER_SIZE, "200").commit();

    SparkWriteConf writeConf = new SparkWriteConf(spark, table);
    Map<String, String> writeProperties = writeConf.writeProperties();
    assertThat(writeProperties).containsEntry(PARQUET_SHRED_VARIANTS, "true");
    assertThat(writeProperties).containsEntry(TableProperties.PARQUET_VARIANT_BUFFER_SIZE, "200");
  }
}
