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

import java.time.Duration;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.flink.maintenance.api.ConvertEqualityDeletesConfig;
import org.apache.iceberg.flink.maintenance.api.DeleteOrphanFilesConfig;
import org.apache.iceberg.flink.maintenance.api.ExpireSnapshotsConfig;
import org.apache.iceberg.flink.maintenance.api.RewriteDataFilesConfig;
import org.apache.iceberg.flink.sink.shuffle.StatisticsType;

/** Flink sink write options */
public class FlinkWriteOptions {

  private FlinkWriteOptions() {}

  // File format for write operations(default: Table write.format.default )
  public static final ConfigOption<String> WRITE_FORMAT =
      ConfigOptions.key("write-format").stringType().noDefaultValue();

  // Overrides this table's write.target-file-size-bytes
  public static final ConfigOption<Long> TARGET_FILE_SIZE_BYTES =
      ConfigOptions.key("target-file-size-bytes").longType().noDefaultValue();

  // Overrides this table's write.<FILE_FORMAT>.compression-codec
  public static final ConfigOption<String> COMPRESSION_CODEC =
      ConfigOptions.key("compression-codec").stringType().noDefaultValue();

  // Overrides this table's write.<FILE_FORMAT>.compression-level
  public static final ConfigOption<String> COMPRESSION_LEVEL =
      ConfigOptions.key("compression-level").stringType().noDefaultValue();

  // Overrides this table's write.<FILE_FORMAT>.compression-strategy
  public static final ConfigOption<String> COMPRESSION_STRATEGY =
      ConfigOptions.key("compression-strategy").stringType().noDefaultValue();

  // Overrides this table's write.upsert.enabled
  public static final ConfigOption<Boolean> WRITE_UPSERT_ENABLED =
      ConfigOptions.key("upsert-enabled").booleanType().noDefaultValue();

  public static final ConfigOption<Boolean> OVERWRITE_MODE =
      ConfigOptions.key("overwrite-enabled").booleanType().defaultValue(false);

  // Overrides the table's write.distribution-mode
  public static final ConfigOption<String> DISTRIBUTION_MODE =
      ConfigOptions.key("distribution-mode").stringType().noDefaultValue();

  public static final ConfigOption<String> RANGE_DISTRIBUTION_STATISTICS_TYPE =
      ConfigOptions.key("range-distribution-statistics-type")
          .stringType()
          .defaultValue(StatisticsType.Auto.name())
          .withDescription("Type of statistics collection: Auto, Map, Sketch");

  public static final ConfigOption<Double> RANGE_DISTRIBUTION_SORT_KEY_BASE_WEIGHT =
      ConfigOptions.key("range-distribution-sort-key-base-weight")
          .doubleType()
          .defaultValue(0.0d)
          .withDescription(
              "Base weight for every sort key relative to target weight per writer task");

  // Branch to write to
  public static final ConfigOption<String> BRANCH =
      ConfigOptions.key("branch").stringType().defaultValue(SnapshotRef.MAIN_BRANCH);

  public static final ConfigOption<Integer> WRITE_PARALLELISM =
      ConfigOptions.key("write-parallelism").intType().noDefaultValue();

  public static final ConfigOption<Boolean> COMPACTION_ENABLE =
      ConfigOptions.key(RewriteDataFilesConfig.PREFIX + "enabled")
          .booleanType()
          .defaultValue(false)
          .withDeprecatedKeys("compaction-enabled");

  public static final ConfigOption<Boolean> EXPIRE_SNAPSHOTS_ENABLE =
      ConfigOptions.key(ExpireSnapshotsConfig.PREFIX + "enabled").booleanType().defaultValue(false);

  public static final ConfigOption<Boolean> DELETE_ORPHAN_FILES_ENABLE =
      ConfigOptions.key(DeleteOrphanFilesConfig.PREFIX + "enabled")
          .booleanType()
          .defaultValue(false);

  public static final ConfigOption<Boolean> CONVERT_EQUALITY_DELETES_ENABLE =
      ConfigOptions.key(ConvertEqualityDeletesConfig.PREFIX + "enabled")
          .booleanType()
          .defaultValue(false);

  @Experimental
  public static final ConfigOption<Duration> TABLE_REFRESH_INTERVAL =
      ConfigOptions.key("table-refresh-interval").durationType().noDefaultValue();

  //  specify the uidSuffix to be used for the underlying IcebergSink
  public static final ConfigOption<String> UID_SUFFIX =
      ConfigOptions.key("uid-suffix").stringType().defaultValue("");

  public static final ConfigOption<Boolean> SHRED_VARIANTS =
      ConfigOptions.key("shred-variants").booleanType().defaultValue(false);

  public static final ConfigOption<Integer> VARIANT_INFERENCE_BUFFER_SIZE =
      ConfigOptions.key("variant-inference-buffer-size").intType().noDefaultValue();
}
