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

import java.time.Duration;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.flink.maintenance.operator.DeleteFilesProcessor;
import org.apache.iceberg.flink.maintenance.operator.ExpireSnapshotsProcessor;
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;

/** Deletes expired snapshots and the corresponding files. */
public class ExpireSnapshots {
  private static final int DELETE_BATCH_SIZE_DEFAULT = 1000;
  private static final String EXECUTOR_OPERATOR_NAME = "Expire Snapshot";
  @VisibleForTesting static final String DELETE_FILES_OPERATOR_NAME = "Delete file";

  private ExpireSnapshots() {}

  /** Creates the builder for creating a stream which expires snapshots for the table. */
  public static Builder builder() {
    return new Builder();
  }

  public static class Builder extends MaintenanceTaskBuilder<ExpireSnapshots.Builder> {
    private Duration maxSnapshotAge = null;
    private Integer numSnapshots = null;
    private Integer planningWorkerPoolSize;
    private int deleteBatchSize = DELETE_BATCH_SIZE_DEFAULT;
    private Boolean cleanExpiredMetadata = null;

    @Override
    String maintenanceTaskName() {
      return "ExpireSnapshots";
    }

    /**
     * The snapshots older than this age will be removed.
     *
     * @param newMaxSnapshotAge of the snapshots to be removed
     */
    public Builder maxSnapshotAge(Duration newMaxSnapshotAge) {
      this.maxSnapshotAge = newMaxSnapshotAge;
      return this;
    }

    /**
     * The minimum number of {@link Snapshot}s to retain. For more details description see {@link
     * org.apache.iceberg.ExpireSnapshots#retainLast(int)}.
     *
     * @param newNumSnapshots number of snapshots to retain
     */
    public Builder retainLast(int newNumSnapshots) {
      this.numSnapshots = newNumSnapshots;
      return this;
    }

    /**
     * The worker pool size used to calculate the files to delete. If not set, the shared worker
     * pool is used.
     *
     * @param newPlanningWorkerPoolSize for planning files to delete
     */
    public Builder planningWorkerPoolSize(int newPlanningWorkerPoolSize) {
      this.planningWorkerPoolSize = newPlanningWorkerPoolSize;
      return this;
    }

    /**
     * Size of the batch used to deleting the files.
     *
     * @param newDeleteBatchSize used for deleting
     */
    public Builder deleteBatchSize(int newDeleteBatchSize) {
      this.deleteBatchSize = newDeleteBatchSize;
      return this;
    }

    /**
     * Expires unused table metadata such as partition specs and schemas.
     *
     * @param newCleanExpiredMetadata remove unused partition specs, schemas, or other metadata when
     *     true
     * @return this for method chaining
     */
    public Builder cleanExpiredMetadata(boolean newCleanExpiredMetadata) {
      this.cleanExpiredMetadata = newCleanExpiredMetadata;
      return this;
    }

    public Builder config(ExpireSnapshotsConfig expireSnapshotsConfig) {
      scheduleOnCommitCount(expireSnapshotsConfig.scheduleOnCommitCount());
      scheduleOnInterval(Duration.ofSeconds(expireSnapshotsConfig.scheduleOnIntervalSecond()));
      deleteBatchSize(expireSnapshotsConfig.deleteBatchSize());
      cleanExpiredMetadata(expireSnapshotsConfig.cleanExpiredMetadata());
      planningWorkerPoolSize(expireSnapshotsConfig.planningWorkerPoolSize());

      Integer retainLast = expireSnapshotsConfig.retainLast();
      if (retainLast != null) {
        retainLast(retainLast);
      }

      Long maxSnapshotAgeSeconds = expireSnapshotsConfig.maxSnapshotAgeSeconds();
      if (maxSnapshotAgeSeconds != null) {
        maxSnapshotAge(Duration.ofSeconds(maxSnapshotAgeSeconds));
      }

      return this;
    }

    @Override
    DataStream<TaskResult> append(DataStream<Trigger> trigger) {
      Preconditions.checkNotNull(tableLoader(), "TableLoader should not be null");

      SingleOutputStreamOperator<TaskResult> result =
          setSlotSharingGroup(
              trigger
                  .process(
                      new ExpireSnapshotsProcessor(
                          tableLoader(),
                          maxSnapshotAge == null ? null : maxSnapshotAge.toMillis(),
                          numSnapshots,
                          planningWorkerPoolSize,
                          cleanExpiredMetadata))
                  .name(operatorName(EXECUTOR_OPERATOR_NAME))
                  .uid(EXECUTOR_OPERATOR_NAME + uidSuffix())
                  .forceNonParallel());

      setSlotSharingGroup(
          result
              .getSideOutput(ExpireSnapshotsProcessor.DELETE_STREAM)
              .rebalance()
              .transform(
                  operatorName(DELETE_FILES_OPERATOR_NAME),
                  TypeInformation.of(Void.class),
                  new DeleteFilesProcessor(
                      tableLoader().loadTable(), taskName(), index(), deleteBatchSize))
              .uid(DELETE_FILES_OPERATOR_NAME + uidSuffix())
              .setParallelism(parallelism()));

      // Ignore the file deletion result and return the DataStream<TaskResult> directly
      return result;
    }
  }
}
