/*
 * 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 com.datastax.driver.core.schemabuilder;

import com.google.common.base.Joiner;
import com.google.common.base.Optional;
import com.google.common.base.Strings;
import java.util.ArrayList;
import java.util.List;

/**
 * The table options used in a CREATE TABLE or ALTER TABLE statement.
 *
 * <p>Implementation notes: this class is abstract and not meant to use directly. The type parameter
 * {@code T} allows usage of <strong>covariant return type</strong> and makes the builder pattern
 * work for different sub-classes.
 *
 * @param <T> the concrete sub-class of {@link com.datastax.driver.core.schemabuilder.TableOptions}
 * @see <a href="http://docs.datastax.com/en/cql/3.1/cql/cql_reference/tabProp.html"
 *     target="_blank">details on table options</a>
 */
public abstract class TableOptions<T extends TableOptions> extends SchemaStatement {

  private StatementStart statementStart;

  private Optional<SchemaBuilder.Caching> cassandra20Caching = Optional.absent();
  private Optional<SchemaBuilder.KeyCaching> cassandra21KeyCaching = Optional.absent();
  private Optional<CachingRowsPerPartition> cassandra21RowCaching = Optional.absent();

  private Optional<Double> bloomFilterFPChance = Optional.absent();

  private Optional<String> comment = Optional.absent();

  private Optional<CompressionOptions> compressionOptions = Optional.absent();

  private Optional<CompactionOptions> compactionOptions = Optional.absent();

  private Optional<Double> dcLocalReadRepairChance = Optional.absent();

  private Optional<Integer> defaultTTL = Optional.absent();

  private Optional<Integer> gcGraceSeconds = Optional.absent();

  private Optional<Integer> indexInterval = Optional.absent();
  private Optional<Integer> minIndexInterval = Optional.absent();
  private Optional<Integer> maxIndexInterval = Optional.absent();

  private Optional<Integer> memtableFlushPeriodInMillis = Optional.absent();

  private Optional<Boolean> populateIOOnCacheFlush = Optional.absent();

  private Optional<Double> readRepairChance = Optional.absent();

  private Optional<Boolean> replicateOnWrite = Optional.absent();

  private Optional<SpeculativeRetryValue> speculativeRetry = Optional.absent();

  private Optional<Boolean> cdc = Optional.absent();

  private List<String> customOptions = new ArrayList<String>();

  private Optional<ReadRepairValue> readReapair = Optional.absent();

  private Optional<AdditionalWritePolicyValue> additionalWritePolicy = Optional.absent();

  @SuppressWarnings("unchecked")
  private final T self = (T) this;

  TableOptions(StatementStart statementStart) {
    this.statementStart = statementStart;
  }

  /**
   * Define the caching type for Cassandra 2.0.x.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is {@link
   * SchemaBuilder.Caching#KEYS_ONLY}.
   *
   * @param caching the caching type (all enum values are allowed).
   * @return this {@code TableOptions} object.
   */
  public T caching(SchemaBuilder.Caching caching) {
    this.cassandra20Caching = Optional.fromNullable(caching);
    return self;
  }

  /**
   * Define the caching options for Cassandra 2.1.x.
   *
   * <p>If no call is made to this method, the default values set by Cassandra are keys = {@link
   * SchemaBuilder.Caching#ALL} and rows_per_partition = {@link
   * com.datastax.driver.core.schemabuilder.SchemaBuilder#noRows()}.
   *
   * @param keys the key cache type.
   * @param rowsPerPartition defines the number of rows to be cached per partition when Row Caching
   *     is enabled. To create instances, use {@link SchemaBuilder#noRows()}, {@link
   *     SchemaBuilder#allRows()} or {@link SchemaBuilder#rows(int)}.
   * @return this {@code TableOptions} object.
   */
  public T caching(SchemaBuilder.KeyCaching keys, CachingRowsPerPartition rowsPerPartition) {
    this.cassandra21KeyCaching = Optional.fromNullable(keys);
    this.cassandra21RowCaching = Optional.fromNullable(rowsPerPartition);
    return self;
  }

  /**
   * Define the desired false-positive probability for SSTable Bloom filters.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is:
   *
   * <ul>
   *   <li><strong>0.01</strong> for the size-tiered compaction strategy;
   *   <li><strong>0.1</strong> for the leveled compaction strategy.
   * </ul>
   *
   * @param fpChance the false positive change. This value should be between 0 and 1.0.
   * @return this {@code TableOptions} object.
   */
  public T bloomFilterFPChance(Double fpChance) {
    validateRateValue(fpChance, "Bloom filter false positive change");
    this.bloomFilterFPChance = Optional.fromNullable(fpChance);
    return self;
  }

  /**
   * Define a human readable comment describing the table.
   *
   * @param comment the comment.
   * @return this {@code TableOptions} object.
   */
  public T comment(String comment) {
    this.comment = Optional.fromNullable(comment);
    return self;
  }

  /**
   * Define the compression options.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is {@link
   * SchemaBuilder#lz4()}.
   *
   * @param compressionOptions the compression options. To create instances, use {@link
   *     SchemaBuilder#noCompression()}, {@link SchemaBuilder#lz4()}, {@link SchemaBuilder#snappy()}
   *     or {@link SchemaBuilder#deflate()}.
   * @return this {@code TableOptions} object.
   */
  public T compressionOptions(CompressionOptions compressionOptions) {
    this.compressionOptions = Optional.fromNullable(compressionOptions);
    return self;
  }

  /**
   * Define the compaction options.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is {@link
   * SchemaBuilder#sizedTieredStategy()}.
   *
   * @param compactionOptions the compaction options. To create instances, use {@link
   *     SchemaBuilder#sizedTieredStategy()}, {@link SchemaBuilder#leveledStrategy()} or {@link
   *     SchemaBuilder#dateTieredStrategy()}
   * @return this {@code TableOptions} object.
   */
  public T compactionOptions(CompactionOptions compactionOptions) {
    this.compactionOptions = Optional.fromNullable(compactionOptions);
    return self;
  }

  /**
   * Define the probability of read repairs being invoked over all replicas in the current data
   * center.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is 0.0.
   *
   * @param dcLocalReadRepairChance the probability.
   * @return this {@code TableOptions} object.
   */
  public T dcLocalReadRepairChance(Double dcLocalReadRepairChance) {
    validateRateValue(dcLocalReadRepairChance, "DC local read repair chance");
    this.dcLocalReadRepairChance = Optional.fromNullable(dcLocalReadRepairChance);
    return self;
  }

  /**
   * Define the default expiration time in seconds for a table.
   *
   * <p>
   *
   * <p>Used in MapReduce/Hive scenarios when you have no control of TTL.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is 0.
   *
   * @param defaultTimeToLive the default time to live in seconds for a table.
   * @return this {@code TableOptions} object.
   */
  public T defaultTimeToLive(Integer defaultTimeToLive) {
    this.defaultTTL = Optional.fromNullable(defaultTimeToLive);
    return self;
  }

  /**
   * Define the time to wait before garbage collecting tombstones (deletion markers).
   *
   * <p>The default value allows a great deal of time for consistency to be achieved prior to
   * deletion. In many deployments this interval can be reduced, and in a single-node cluster it can
   * be safely set to zero.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is 864000 secs (10
   * days).
   *
   * @param gcGraceSeconds the grace period.
   * @return this {@code TableOptions} object.
   */
  public T gcGraceSeconds(Integer gcGraceSeconds) {
    this.gcGraceSeconds = Optional.fromNullable(gcGraceSeconds);
    return self;
  }

  /**
   * Define the index interval for Cassandra 2.0.
   *
   * <p>To control the sampling of entries from the primary row index, configure sample frequency of
   * the partition summary by changing the index interval. After changing the index interval,
   * SSTables are written to disk with new information. The interval corresponds to the number of
   * index entries that are skipped between taking each sample. By default, Cassandra samples one
   * row key out of every 128. The larger the interval, the smaller and less effective the sampling.
   * The larger the sampling, the more effective the index, but with increased memory usage. In
   * Cassandra 2.0.x, generally, the best trade off between memory usage and performance is a value
   * between 128 and 512 in combination with a large table key cache. However, if you have small
   * rows (many to an OS page), you may want to increase the sample size, which often lowers memory
   * usage without an impact on performance. For large rows, decreasing the sample size may improve
   * read performance.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is 128.
   *
   * @param indexInterval the index interval.
   * @return this {@code TableOptions} object.
   */
  public T indexInterval(Integer indexInterval) {
    this.indexInterval = Optional.fromNullable(indexInterval);
    return self;
  }

  /**
   * Define the minimum index interval for Cassandra 2.1.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is 128.
   *
   * @param minIndexInterval the minimum index interval.
   * @return this {@code TableOptions} object.
   * @see #indexInterval(Integer)
   */
  public T minIndexInterval(Integer minIndexInterval) {
    this.minIndexInterval = Optional.fromNullable(minIndexInterval);
    return self;
  }

  /**
   * Define the maximum index interval for Cassandra 2.1.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is 2048.
   *
   * @param maxIndexInterval the maximum index interval.
   * @return this {@code TableOptions} object.
   * @see #indexInterval(Integer)
   */
  public T maxIndexInterval(Integer maxIndexInterval) {
    this.maxIndexInterval = Optional.fromNullable(maxIndexInterval);
    return self;
  }

  /**
   * Define the memtable flush period.
   *
   * <p>If set, this forces flushing of the memtable after the specified time elapses.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is 0.
   *
   * @param memtableFlushPeriodInMillis the memtable flush period in milliseconds.
   * @return this {@code TableOptions} object.
   */
  public T memtableFlushPeriodInMillis(Integer memtableFlushPeriodInMillis) {
    this.memtableFlushPeriodInMillis = Optional.fromNullable(memtableFlushPeriodInMillis);
    return self;
  }

  /**
   * Define whether to populate IO cache on flush of sstables.
   *
   * <p>If set, Cassandra adds newly flushed or compacted sstables to the operating system page
   * cache, potentially evicting other cached data to make room. Enable when all data in the table
   * is expected to fit in memory.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is {@code false}.
   *
   * @param populateIOOnCacheFlush whether to populate IO cache on flush of sstables.
   * @return this {@code TableOptions} object.
   * @see <a
   *     href="http://docs.datastax.com/en/cassandra/2.0/cassandra/configuration/configCassandra_yaml_r.html?scroll=reference_ds_qfg_n1r_1k__compaction_preheat_key_cache">the
   *     global option compaction_preheat_key_cache</a>
   */
  public T populateIOCacheOnFlush(Boolean populateIOOnCacheFlush) {
    this.populateIOOnCacheFlush = Optional.fromNullable(populateIOOnCacheFlush);
    return self;
  }

  /**
   * Define the probability with which read repairs should be invoked on non-quorum reads. The value
   * must be between 0 and 1.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is 0.1.
   *
   * @param readRepairChance the read repair chance.
   * @return this {@code TableOptions} object.
   */
  public T readRepairChance(Double readRepairChance) {
    validateRateValue(readRepairChance, "Read repair chance");
    this.readRepairChance = Optional.fromNullable(readRepairChance);
    return self;
  }

  /**
   * Define whether to replicate data on write (Cassandra 2.0.x only).
   *
   * <p>When set to {@code true}, replicates writes to all affected replicas regardless of the
   * consistency level specified by the client for a write request. For counter tables, this should
   * always be set to {@code true}.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is {@code true}.
   *
   * @param replicateOnWrite whether to replicate data on write.
   * @return this {@code TableOptions} object.
   */
  public T replicateOnWrite(Boolean replicateOnWrite) {
    this.replicateOnWrite = Optional.fromNullable(replicateOnWrite);
    return self;
  }

  /**
   * To override normal read timeout when read_repair_chance is not 1.0, sending another request to
   * read, choose one of these values and use the property to create or alter the table:
   *
   * <ul>
   *   <li>ALWAYS: Retry reads of all replicas.
   *   <li>Xpercentile: Retry reads based on the effect on throughput and latency.
   *   <li>Yms: Retry reads after specified milliseconds.
   *   <li>NONE: Do not retry reads.
   * </ul>
   *
   * <p>Using the speculative retry property, you can configure rapid read protection in Cassandra
   * 2.0.2 and later. Use this property to retry a request after some milliseconds have passed or
   * after a percentile of the typical read latency has been reached, which is tracked per table.
   *
   * <p>
   *
   * <p>If no call is made to this method, the default value set by Cassandra is {code
   * 99percentile}.
   *
   * @param speculativeRetry the speculative retry. To create instances, use {@link
   *     SchemaBuilder#noSpeculativeRetry()}, {@link SchemaBuilder#always()}, {@link
   *     SchemaBuilder#percentile(int)} or {@link SchemaBuilder#millisecs(int)}.
   * @return this {@code TableOptions} object.
   */
  public T speculativeRetry(SpeculativeRetryValue speculativeRetry) {
    this.speculativeRetry = Optional.fromNullable(speculativeRetry);
    return self;
  }

  /**
   * Define whether or not change data capture is enabled on this table.
   *
   * <p>Note that using this option with a version of Apache Cassandra less than 3.8 will raise a
   * syntax error.
   *
   * <p>If no call is made to this method, the default value set by Cassandra is {@code false}.
   *
   * @param cdc Whether or not change data capture should be enabled for this table.
   * @return this {@code TableOptions} object.
   */
  public T cdc(Boolean cdc) {
    this.cdc = Optional.fromNullable(cdc);
    return self;
  }

  /**
   * Define a free-form option as a key/value pair.
   *
   * <p>This method is provided as a fallback if the SchemaBuilder is used with a more recent
   * version of Cassandra that has new, unsupported options.
   *
   * @param key the name of the option.
   * @param value the value of the option. If it's a {@code String}, it will be included in single
   *     quotes, otherwise the result of invoking its {@code toString} method will be used unquoted.
   * @return this {@code TableOptions} object.
   */
  public T freeformOption(String key, Object value) {
    if (Strings.isNullOrEmpty(key)) {
      throw new IllegalArgumentException("Key for custom option should not be null or blank");
    }
    customOptions.add(buildCustomOption(key, value));
    return self;
  }

  /**
   * Define the read_repair mode for this table. Possible values are:
   *
   * <ul>
   *   <li>BLOCKING
   *   <li>NONE
   * </ul>
   *
   * <p>This option is for Cassandra 4.0+. The default value is {@code BLOCKING}, which means that
   * when a read repair is started, a read will block on writes sent to other replicas until the CL
   * is reached by the writes. This provides monotonic quorum reads, but not partition level write
   * atomicity.
   *
   * <p>If {@code NONE} is specified, the coordinator will reconcile any differences between
   * replicas, but will not attempt to repair them. This provides partition level write atomicity,
   * but not monotonic quorum reads.
   *
   * @param readRepair the read repair mode.
   * @return this {@code TableOptions} object.
   */
  public T readRepair(ReadRepairValue readRepair) {
    this.readReapair = Optional.fromNullable(readRepair);
    return self;
  }

  /**
   * Define the additional_write_policy for this table. This specifies the threshold at which a
   * cheap quorum write will be upgraded to include transient replicas.
   *
   * <p>This option is for Cassandra 4.0+. The default value is "99p".
   *
   * @param additionalWritePolicy the additional write policy threshold.
   * @return this {@code TableOptions} object.
   */
  public T additionalWritePolicy(AdditionalWritePolicyValue additionalWritePolicy) {
    this.additionalWritePolicy = Optional.fromNullable(additionalWritePolicy);
    return self;
  }

  private static String buildCustomOption(String key, Object value) {
    return String.format(
        "%s = %s", key, (value instanceof String) ? "'" + value + "'" : value.toString());
  }

  private List<String> buildCommonOptions() {
    List<String> options = new ArrayList<String>();

    buildCachingOptions(options);

    if (bloomFilterFPChance.isPresent()) {
      options.add("bloom_filter_fp_chance = " + bloomFilterFPChance.get());
    }

    if (comment.isPresent()) {
      options.add("comment = '" + comment.get() + "'");
    }

    if (compressionOptions.isPresent()) {
      options.add("compression = " + compressionOptions.get().build());
    }

    if (compactionOptions.isPresent()) {
      options.add("compaction = " + compactionOptions.get().build());
    }

    if (dcLocalReadRepairChance.isPresent()) {
      options.add("dclocal_read_repair_chance = " + dcLocalReadRepairChance.get());
    }

    if (defaultTTL.isPresent()) {
      options.add("default_time_to_live = " + defaultTTL.get());
    }

    if (gcGraceSeconds.isPresent()) {
      options.add("gc_grace_seconds = " + gcGraceSeconds.get());
    }

    if (indexInterval.isPresent()) {
      options.add("index_interval = " + indexInterval.get());
    }

    if (minIndexInterval.isPresent()) {
      options.add("min_index_interval = " + minIndexInterval.get());
    }

    if (maxIndexInterval.isPresent()) {
      options.add("max_index_interval = " + maxIndexInterval.get());
    }

    if (memtableFlushPeriodInMillis.isPresent()) {
      options.add("memtable_flush_period_in_ms = " + memtableFlushPeriodInMillis.get());
    }

    if (populateIOOnCacheFlush.isPresent()) {
      options.add("populate_io_cache_on_flush = " + populateIOOnCacheFlush.get());
    }

    if (readRepairChance.isPresent()) {
      options.add("read_repair_chance = " + readRepairChance.get());
    }

    if (replicateOnWrite.isPresent()) {
      options.add("replicate_on_write = " + replicateOnWrite.get());
    }

    if (speculativeRetry.isPresent()) {
      options.add("speculative_retry = " + speculativeRetry.get().value());
    }

    if (cdc.isPresent()) {
      options.add("cdc = " + cdc.get());
    }

    if (readReapair.isPresent()) {
      options.add("read_repair = " + readReapair.get().value());
    }

    if (additionalWritePolicy.isPresent()) {
      options.add("additional_write_policy = " + additionalWritePolicy.get().value());
    }

    options.addAll(customOptions);

    return options;
  }

  private void buildCachingOptions(List<String> options) {
    if (cassandra20Caching.isPresent() && cassandra21KeyCaching.isPresent()) {
      throw new IllegalStateException(
          "Can't use Cassandra 2.0 and 2.1 caching at the same time, you must call only one version of caching()");
    } else if (cassandra20Caching.isPresent()) {
      options.add("caching = " + cassandra20Caching.get().value());
    } else if (cassandra21KeyCaching.isPresent() && cassandra21RowCaching.isPresent()) {
      options.add(
          String.format(
              "caching = {'keys' : %s, 'rows_per_partition' : %s}",
              cassandra21KeyCaching.get().value(), cassandra21RowCaching.get().value()));
    }
  }

  protected abstract void addSpecificOptions(List<String> options);

  @Override
  public final String buildInternal() {
    List<String> options = buildCommonOptions();
    addSpecificOptions(options);
    return statementStart.buildInternal()
        + STATEMENT_START
        + "WITH "
        + Joiner.on(" AND ").join(options);
  }

  static void validateRateValue(Double rateValue, String property) {
    if (rateValue != null && (rateValue < 0 || rateValue > 1.0)) {
      throw new IllegalArgumentException(property + " should be between 0 and 1");
    }
  }

  /**
   * Compaction options for a CREATE or ALTER TABLE statement.
   *
   * <p>To create instances, use {@link SchemaBuilder#sizedTieredStategy()}, {@link
   * SchemaBuilder#leveledStrategy()}, {@link SchemaBuilder#dateTieredStrategy()} or {@link
   * SchemaBuilder#timeWindowCompactionStrategy()}
   */
  public abstract static class CompactionOptions<T extends CompactionOptions> {

    private Strategy strategy;

    private Optional<Boolean> enabled = Optional.absent();

    private Optional<Integer> tombstoneCompactionIntervalInDay = Optional.absent();

    private Optional<Double> tombstoneThreshold = Optional.absent();

    private Optional<Boolean> uncheckedTombstoneCompaction = Optional.absent();

    private List<String> customOptions = new ArrayList<String>();

    @SuppressWarnings("unchecked")
    private final T self = (T) this;

    CompactionOptions(Strategy compactionStrategy) {
      this.strategy = compactionStrategy;
    }

    /**
     * Enable or disable background compaction.
     *
     * <p>If no call is made to this method, the default value set by Cassandra is {code true}.
     *
     * @param enabled whether to enable background compaction for the table.
     * @return this object (for call chaining).
     */
    public T enabled(Boolean enabled) {
      this.enabled = Optional.fromNullable(enabled);
      return self;
    }

    /**
     * Define the minimum number of days to wait after an SSTable creation time before considering
     * the SSTable for tombstone compaction. Tombstone compaction is the compaction triggered if the
     * SSTable has more garbage-collectable tombstones than tombstone_threshold.
     *
     * <p>If no call is made to this method, the default value set by Cassandra is 1.
     *
     * @param tombstoneCompactionInterval the tombstone compaction interval in day.
     * @return this object (for call chaining).
     */
    public T tombstoneCompactionIntervalInDay(Integer tombstoneCompactionInterval) {
      this.tombstoneCompactionIntervalInDay = Optional.fromNullable(tombstoneCompactionInterval);
      return self;
    }

    /**
     * Define the ratio of garbage-collectable tombstones to all contained columns, which if
     * exceeded by the SSTable triggers compaction (with no other SSTables) for the purpose of
     * purging the tombstones.
     *
     * <p>If no call is made to this method, the default value set by Cassandra is 0.2.
     *
     * @param tombstoneThreshold the threshold.
     * @return this object (for call chaining).
     */
    public T tombstoneThreshold(Double tombstoneThreshold) {
      validateRateValue(tombstoneThreshold, "Tombstone threshold");
      this.tombstoneThreshold = Optional.fromNullable(tombstoneThreshold);
      return self;
    }

    /**
     * Enables more aggressive than normal tombstone compactions. A single SSTable tombstone
     * compaction runs without checking the likelihood of success (Cassandra 2.0.9 and later).
     *
     * <p>If no call is made to this method, the default value set by Cassandra is {code false}.
     *
     * @param uncheckedTombstoneCompaction whether to enable the feature.
     * @return this object (for call chaining).
     */
    public T uncheckedTombstoneCompaction(Boolean uncheckedTombstoneCompaction) {
      this.uncheckedTombstoneCompaction = Optional.fromNullable(uncheckedTombstoneCompaction);
      return self;
    }

    /**
     * Define a free-form option as a key/value pair.
     *
     * <p>This method is provided as a fallback if the SchemaBuilder is used with a more recent
     * version of Cassandra that has new, unsupported options.
     *
     * @param key the name of the option.
     * @param value the value of the option. If it's a {@code CharSequence}, it will be included in
     *     single quotes, otherwise the result of invoking its {@code toString} method will be used
     *     unquoted.
     * @return this object (for call chaining).
     */
    public T freeformOption(String key, Object value) {
      if (Strings.isNullOrEmpty(key)) {
        throw new IllegalArgumentException("Key for custom option should not be null or blank");
      }
      customOptions.add(buildCustomOption(key, value));
      return self;
    }

    private static String buildCustomOption(String key, Object value) {
      return String.format(
          "'%s' : %s", key, (value instanceof CharSequence) ? "'" + value + "'" : value.toString());
    }

    List<String> buildCommonOptions() {

      List<String> options = new ArrayList<String>();
      options.add("'class' : " + strategy.strategyClass());

      if (enabled.isPresent()) {
        options.add("'enabled' : " + enabled.get());
      }

      if (tombstoneCompactionIntervalInDay.isPresent()) {
        options.add("'tombstone_compaction_interval' : " + tombstoneCompactionIntervalInDay.get());
      }

      if (tombstoneThreshold.isPresent()) {
        options.add("'tombstone_threshold' : " + tombstoneThreshold.get());
      }

      if (uncheckedTombstoneCompaction.isPresent()) {
        options.add("'unchecked_tombstone_compaction' : " + uncheckedTombstoneCompaction.get());
      }

      options.addAll(customOptions);

      return options;
    }

    public abstract String build();

    /** Compaction options specific to SizeTiered strategy */
    public static class SizeTieredCompactionStrategyOptions
        extends CompactionOptions<SizeTieredCompactionStrategyOptions> {

      private Optional<Double> bucketHigh = Optional.absent();

      private Optional<Double> bucketLow = Optional.absent();

      private Optional<Double> coldReadsRatioToOmit = Optional.absent();

      private Optional<Integer> minThreshold = Optional.absent();

      private Optional<Integer> maxThreshold = Optional.absent();

      private Optional<Long> minSSTableSizeInBytes = Optional.absent();

      SizeTieredCompactionStrategyOptions() {
        super(Strategy.SIZED_TIERED);
      }

      /**
       * Size-tiered compaction strategy (STCS) considers SSTables to be within the same bucket if
       * the SSTable size diverges by 50% or less from the default bucket_low and default
       * bucket_high values: [average-size × bucket_low, average-size × bucket_high].
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 1.5.
       *
       * @param bucketHigh the new value.
       * @return this object (for call chaining).
       */
      public SizeTieredCompactionStrategyOptions bucketHigh(Double bucketHigh) {
        this.bucketHigh = Optional.fromNullable(bucketHigh);
        return this;
      }

      /**
       * Size-tiered compaction strategy (STCS) considers SSTables to be within the same bucket if
       * the SSTable size diverges by 50% or less from the default bucket_low and default
       * bucket_high values: [average-size × bucket_low, average-size × bucket_high].
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 0.5.
       *
       * @param bucketLow the new value.
       * @return this object (for call chaining).
       */
      public SizeTieredCompactionStrategyOptions bucketLow(Double bucketLow) {
        this.bucketLow = Optional.fromNullable(bucketLow);
        return this;
      }

      /**
       * The maximum percentage of reads/sec that ignored SSTables may account for. The recommended
       * range of values is 0.0 and 1.0. In Cassandra 2.0.3 and later, you can enable the
       * cold_reads_to_omit property to tune performace per table. The <a
       * href="http://www.datastax.com/dev/blog/optimizations-around-cold-sstables">Optimizations
       * around Cold SSTables</a> blog includes detailed information tuning performance using this
       * property, which avoids compacting cold SSTables. Use the ALTER TABLE command to configure
       * cold_reads_to_omit.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 0.0 (disabled).
       *
       * @param coldReadsRatio the new value.
       * @return this object (for call chaining).
       */
      public SizeTieredCompactionStrategyOptions coldReadsRatioToOmit(Double coldReadsRatio) {
        validateRateValue(coldReadsRatio, "Cold read ratio to omit ");
        this.coldReadsRatioToOmit = Optional.fromNullable(coldReadsRatio);
        return this;
      }

      /**
       * Sets the minimum number of SSTables to trigger a minor compaction
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 4.
       *
       * @param minThreshold the new value.
       * @return this object (for call chaining).
       */
      public SizeTieredCompactionStrategyOptions minThreshold(Integer minThreshold) {
        this.minThreshold = Optional.fromNullable(minThreshold);
        return this;
      }

      /**
       * Sets the maximum number of SSTables to allow in a minor compaction. In
       * LeveledCompactionStrategy (LCS), it applies to L0 when L0 gets behind, that is, when L0
       * accumulates more than MAX_COMPACTING_L0 SSTables.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 32.
       *
       * @param maxThreshold the new value.
       * @return this object (for call chaining).
       */
      public SizeTieredCompactionStrategyOptions maxThreshold(Integer maxThreshold) {
        this.maxThreshold = Optional.fromNullable(maxThreshold);
        return this;
      }

      /**
       * The SizeTieredCompactionStrategy groups SSTables for compaction into buckets. The bucketing
       * process groups SSTables that differ in size by less than 50%. This results in a bucketing
       * process that is too fine grained for small SSTables. If your SSTables are small, use
       * min_sstable_size to define a size threshold (in bytes) below which all SSTables belong to
       * one unique bucket
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 52428800 (50
       * MB).
       *
       * @param minSSTableSize the new value.
       * @return this object (for call chaining).
       */
      public SizeTieredCompactionStrategyOptions minSSTableSizeInBytes(Long minSSTableSize) {
        this.minSSTableSizeInBytes = Optional.fromNullable(minSSTableSize);
        return this;
      }

      @Override
      public String build() {
        final List<String> generalOptions = super.buildCommonOptions();

        List<String> options = new ArrayList<String>(generalOptions);

        if (bucketHigh.isPresent()) {
          options.add("'bucket_high' : " + bucketHigh.get());
        }

        if (bucketLow.isPresent()) {
          options.add("'bucket_low' : " + bucketLow.get());
        }

        if (coldReadsRatioToOmit.isPresent()) {
          options.add("'cold_reads_to_omit' : " + coldReadsRatioToOmit.get());
        }

        if (minThreshold.isPresent()) {
          options.add("'min_threshold' : " + minThreshold.get());
        }

        if (maxThreshold.isPresent()) {
          options.add("'max_threshold' : " + maxThreshold.get());
        }

        if (minSSTableSizeInBytes.isPresent()) {
          options.add("'min_sstable_size' : " + minSSTableSizeInBytes.get());
        }
        return "{" + Joiner.on(", ").join(options) + "}";
      }
    }

    /** Compaction options specific to Leveled strategy */
    public static class LeveledCompactionStrategyOptions
        extends CompactionOptions<LeveledCompactionStrategyOptions> {

      private Optional<Integer> ssTableSizeInMB = Optional.absent();

      LeveledCompactionStrategyOptions() {
        super(Strategy.LEVELED);
      }

      /**
       * The target size for SSTables that use the leveled compaction strategy. Although SSTable
       * sizes should be less or equal to sstable_size_in_mb, it is possible to have a larger
       * SSTable during compaction. This occurs when data for a given partition key is exceptionally
       * large. The data is not split into two SSTables
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 160.
       *
       * @param ssTableSizeInMB the new value.
       * @return this object (for call chaining).
       */
      public LeveledCompactionStrategyOptions ssTableSizeInMB(Integer ssTableSizeInMB) {
        this.ssTableSizeInMB = Optional.fromNullable(ssTableSizeInMB);
        return this;
      }

      @Override
      public String build() {
        final List<String> generalOptions = super.buildCommonOptions();

        List<String> options = new ArrayList<String>(generalOptions);

        if (ssTableSizeInMB.isPresent()) {
          options.add("'sstable_size_in_mb' : " + ssTableSizeInMB.get());
        }
        return "{" + Joiner.on(", ").join(options) + "}";
      }
    }

    /** Compaction options specific to the date-tiered strategy. */
    public static class DateTieredCompactionStrategyOptions
        extends CompactionOptions<DateTieredCompactionStrategyOptions> {

      public enum TimeStampResolution {
        MICROSECONDS,
        MILLISECONDS
      }

      private Optional<Integer> baseTimeSeconds = Optional.absent();

      private Optional<Integer> maxSSTableAgeDays = Optional.absent();

      private Optional<Integer> minThreshold = Optional.absent();

      private Optional<Integer> maxThreshold = Optional.absent();

      private Optional<TimeStampResolution> timestampResolution = Optional.absent();

      DateTieredCompactionStrategyOptions() {
        super(Strategy.DATE_TIERED);
      }

      /**
       * Sets the size of the first window.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 3600 (1 hour).
       *
       * @param baseTimeSeconds the size of the first window.
       * @return this object (for call chaining).
       */
      public DateTieredCompactionStrategyOptions baseTimeSeconds(Integer baseTimeSeconds) {
        this.baseTimeSeconds = Optional.fromNullable(baseTimeSeconds);
        return this;
      }

      /**
       * Stop compacting SSTables only having data older than these specified days.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 365.
       *
       * @param maxSSTableAgeDays the maximum age of the SSTables to compact.
       * @return this object (for call chaining).
       */
      public DateTieredCompactionStrategyOptions maxSSTableAgeDays(Integer maxSSTableAgeDays) {
        this.maxSSTableAgeDays = Optional.fromNullable(maxSSTableAgeDays);
        return this;
      }

      /**
       * Sets the minimum number of SSTables to trigger a minor compaction
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 4.
       *
       * @param minThreshold the new value.
       * @return this object (for call chaining).
       */
      public DateTieredCompactionStrategyOptions minThreshold(Integer minThreshold) {
        this.minThreshold = Optional.fromNullable(minThreshold);
        return this;
      }

      /**
       * Sets the maximum number of SSTables to allow in a minor compaction. In
       * LeveledCompactionStrategy (LCS), it applies to L0 when L0 gets behind, that is, when L0
       * accumulates more than MAX_COMPACTING_L0 SSTables.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 32.
       *
       * @param maxThreshold the new value.
       * @return this object (for call chaining).
       */
      public DateTieredCompactionStrategyOptions maxThreshold(Integer maxThreshold) {
        this.maxThreshold = Optional.fromNullable(maxThreshold);
        return this;
      }

      /**
       * Sets the timestamp resolution, depending on the timestamp unit of the data you insert.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is {@code
       * MICROSECONDS}.
       *
       * @param timestampResolution {@link TimeStampResolution#MICROSECONDS} or {@link
       *     TimeStampResolution#MILLISECONDS}.
       * @return this object (for call chaining).
       */
      public DateTieredCompactionStrategyOptions timestampResolution(
          TimeStampResolution timestampResolution) {
        this.timestampResolution = Optional.fromNullable(timestampResolution);
        return this;
      }

      @Override
      public String build() {
        final List<String> generalOptions = super.buildCommonOptions();

        List<String> options = new ArrayList<String>(generalOptions);

        if (baseTimeSeconds.isPresent()) {
          options.add("'base_time_seconds' : " + baseTimeSeconds.get());
        }

        if (maxSSTableAgeDays.isPresent()) {
          options.add("'max_sstable_age_days' : " + maxSSTableAgeDays.get());
        }

        if (minThreshold.isPresent()) {
          options.add("'min_threshold' : " + minThreshold.get());
        }

        if (maxThreshold.isPresent()) {
          options.add("'max_threshold' : " + maxThreshold.get());
        }

        if (timestampResolution.isPresent()) {
          options.add("'timestamp_resolution' : '" + timestampResolution.get() + "'");
        }

        return "{" + Joiner.on(", ").join(options) + "}";
      }
    }

    /** Compaction options specific to the time window strategy. */
    public static class TimeWindowCompactionStrategyOptions
        extends CompactionOptions<TimeWindowCompactionStrategyOptions> {

      public enum CompactionWindowUnit {
        MINUTES,
        HOURS,
        DAYS
      }

      public enum TimeStampResolution {
        MICROSECONDS,
        MILLISECONDS
      }

      private Optional<Double> bucketHigh = Optional.absent();

      private Optional<Double> bucketLow = Optional.absent();

      private Optional<CompactionWindowUnit> compactionWindowUnit = Optional.absent();

      private Optional<Integer> compactionWindowSize = Optional.absent();

      private Optional<Integer> minThreshold = Optional.absent();

      private Optional<Integer> maxThreshold = Optional.absent();

      private Optional<Long> minSSTableSizeInBytes = Optional.absent();

      private Optional<TimeStampResolution> timestampResolution = Optional.absent();

      private Optional<Boolean> unsafeAggressiveSSTableExpiration = Optional.absent();

      TimeWindowCompactionStrategyOptions() {
        super(Strategy.TIME_WINDOW);
      }

      /**
       * Size-tiered compaction strategy (STCS) is used in the newest window, this method sets the
       * bucketHigh value used in STCS.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 1.5.
       *
       * @param bucketHigh the new value.
       * @return this object (for call chaining).
       */
      public TimeWindowCompactionStrategyOptions bucketHigh(Double bucketHigh) {
        this.bucketHigh = Optional.fromNullable(bucketHigh);
        return this;
      }

      /**
       * Size-tiered compaction strategy (STCS) is used in the newest window, this method sets the
       * bucketLow value used in STCS.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 0.5.
       *
       * @param bucketLow the new value.
       * @return this object (for call chaining).
       */
      public TimeWindowCompactionStrategyOptions bucketLow(Double bucketLow) {
        this.bucketLow = Optional.fromNullable(bucketLow);
        return this;
      }

      /**
       * Sets the time unit used to define the window size
       *
       * <p>If no call is made to this method, the default value set by Cassandra is {@code DAYS}.
       *
       * @param compactionWindowUnit {@link CompactionWindowUnit#MINUTES}, {@link
       *     CompactionWindowUnit#HOURS} or {@link CompactionWindowUnit#DAYS}.
       * @return this object (for call chaining).
       */
      public TimeWindowCompactionStrategyOptions compactionWindowUnit(
          CompactionWindowUnit compactionWindowUnit) {
        this.compactionWindowUnit = Optional.fromNullable(compactionWindowUnit);
        return this;
      }

      /**
       * Sets the number of units that make up a window.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 1.
       *
       * @param compactionWindowSize the size of the first window.
       * @return this object (for call chaining).
       */
      public TimeWindowCompactionStrategyOptions compactionWindowSize(
          Integer compactionWindowSize) {
        this.compactionWindowSize = Optional.fromNullable(compactionWindowSize);
        return this;
      }

      /**
       * Sets the minimum number of SSTables to trigger a minor compaction
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 4.
       *
       * @param minThreshold the new value.
       * @return this object (for call chaining).
       */
      public TimeWindowCompactionStrategyOptions minThreshold(Integer minThreshold) {
        this.minThreshold = Optional.fromNullable(minThreshold);
        return this;
      }

      /**
       * Sets the maximum number of SSTables to allow in a minor compaction.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 32.
       *
       * @param maxThreshold the new value.
       * @return this object (for call chaining).
       */
      public TimeWindowCompactionStrategyOptions maxThreshold(Integer maxThreshold) {
        this.maxThreshold = Optional.fromNullable(maxThreshold);
        return this;
      }

      /**
       * Size-tiered compaction strategy (STCS) is used in the newest window, this method sets the
       * minSSTableSize value used in STCS.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is 52428800 (50
       * MB).
       *
       * @param minSSTableSize the new value.
       * @return this object (for call chaining).
       */
      public TimeWindowCompactionStrategyOptions minSSTableSizeInBytes(Long minSSTableSize) {
        this.minSSTableSizeInBytes = Optional.fromNullable(minSSTableSize);
        return this;
      }

      /**
       * Sets the timestamp resolution, depending on the timestamp unit of the data you insert.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is {@code
       * MICROSECONDS}.
       *
       * @param timestampResolution {@link TimeStampResolution#MICROSECONDS} or {@link
       *     TimeStampResolution#MILLISECONDS}.
       * @return this object (for call chaining).
       */
      public TimeWindowCompactionStrategyOptions timestampResolution(
          TimeStampResolution timestampResolution) {
        this.timestampResolution = Optional.fromNullable(timestampResolution);
        return this;
      }

      /**
       * Allow expired sstables to be dropped without checking if its data is shadowing other
       * sstables. This is a potentially risky option that can lead to data loss or deleted data
       * re-appearing.
       *
       * <p>If no call is made to this method, the default value set by Cassandra is false.
       *
       * @param unsafeAggressiveSSTableExpiration whether to enable unsafe aggressive sstable
       *     expiration option.
       * @return this object (for call chaining).
       */
      public TimeWindowCompactionStrategyOptions unsafeAggressiveSSTableExpiration(
          Boolean unsafeAggressiveSSTableExpiration) {
        this.unsafeAggressiveSSTableExpiration =
            Optional.fromNullable(unsafeAggressiveSSTableExpiration);
        return this;
      }

      @Override
      public String build() {
        final List<String> generalOptions = super.buildCommonOptions();

        List<String> options = new ArrayList<String>(generalOptions);

        if (bucketHigh.isPresent()) {
          options.add("'bucket_high' : " + bucketHigh.get());
        }

        if (bucketLow.isPresent()) {
          options.add("'bucket_low' : " + bucketLow.get());
        }

        if (compactionWindowUnit.isPresent()) {
          options.add("'compaction_window_unit' : '" + compactionWindowUnit.get() + "'");
        }

        if (compactionWindowSize.isPresent()) {
          options.add("'compaction_window_size' : " + compactionWindowSize.get());
        }

        if (minThreshold.isPresent()) {
          options.add("'min_threshold' : " + minThreshold.get());
        }

        if (maxThreshold.isPresent()) {
          options.add("'max_threshold' : " + maxThreshold.get());
        }

        if (minSSTableSizeInBytes.isPresent()) {
          options.add("'min_sstable_size' : " + minSSTableSizeInBytes.get());
        }

        if (timestampResolution.isPresent()) {
          options.add("'timestamp_resolution' : '" + timestampResolution.get() + "'");
        }

        if (unsafeAggressiveSSTableExpiration.isPresent()) {
          options.add(
              "'unsafe_aggressive_sstable_expiration' : '"
                  + unsafeAggressiveSSTableExpiration.get()
                  + "'");
        }

        return "{" + Joiner.on(", ").join(options) + "}";
      }
    }

    /**
     * Compaction strategies. Possible values: SIZED_TIERED, LEVELED, DATE_TIERED AND TIME_WINDOW
     */
    public static enum Strategy {
      SIZED_TIERED("'SizeTieredCompactionStrategy'"),
      LEVELED("'LeveledCompactionStrategy'"),
      DATE_TIERED("'DateTieredCompactionStrategy'"),
      TIME_WINDOW("'TimeWindowCompactionStrategy'");

      private String strategyClass;

      Strategy(String strategyClass) {
        this.strategyClass = strategyClass;
      }

      public String strategyClass() {
        return strategyClass;
      }

      @Override
      public String toString() {
        return strategyClass;
      }
    }
  }

  /**
   * The compression options for a CREATE or ALTER TABLE statement.
   *
   * <p>To create instances, use {@link SchemaBuilder#noCompression()}, {@link SchemaBuilder#lz4()},
   * {@link SchemaBuilder#snappy()} or {@link SchemaBuilder#deflate()}.
   */
  public static class CompressionOptions {

    private Algorithm algorithm;

    private Optional<Integer> chunkLengthInKb = Optional.absent();

    private Optional<Double> crcCheckChance = Optional.absent();

    CompressionOptions(Algorithm algorithm) {
      this.algorithm = algorithm;
    }

    /**
     * On disk, SSTables are compressed by block to allow random reads. This subproperty of
     * compression defines the size (in KB) of the block. Values larger than the default value might
     * improve the compression rate, but increases the minimum size of data to be read from disk
     * when a read occurs. The default value is a good middle-ground for compressing tables. Adjust
     * compression size to account for read/write access patterns (how much data is typically
     * requested at once) and the average size of rows in the table.
     *
     * <p>If no call is made to this method, the default value set by Cassandra is 54.
     *
     * @param chunkLengthInKb the new value.
     * @return this object (for call chaining).
     */
    public CompressionOptions withChunkLengthInKb(Integer chunkLengthInKb) {
      this.chunkLengthInKb = Optional.fromNullable(chunkLengthInKb);
      return this;
    }

    /**
     * When compression is enabled, each compressed block includes a checksum of that block for the
     * purpose of detecting disk bitrate and avoiding the propagation of corruption to other
     * replica. This option defines the probability with which those checksums are checked during
     * read. By default they are always checked. Set to 0 to disable checksum checking and to 0.5,
     * for instance, to check them on every other read.
     *
     * <p>If no call is made to this method, the default value set by Cassandra is 1.0 (always
     * check).
     *
     * @param crcCheckChance the new value.
     * @return this object (for call chaining).
     */
    public CompressionOptions withCRCCheckChance(Double crcCheckChance) {
      validateRateValue(crcCheckChance, "CRC check chance");
      this.crcCheckChance = Optional.fromNullable(crcCheckChance);
      return this;
    }

    public String build() {
      List<String> options = new ArrayList<String>();
      options.add("'sstable_compression' : " + algorithm.value());

      if (chunkLengthInKb.isPresent()) {
        options.add("'chunk_length_kb' : " + chunkLengthInKb.get());
      }

      if (crcCheckChance.isPresent()) {
        options.add("'crc_check_chance' : " + crcCheckChance.get());
      }
      return "{" + Joiner.on(", ").join(options) + "}";
    }

    /** Compression algorithms. Possible values: NONE, LZ4, SNAPPY, DEFLATE */
    public static enum Algorithm {
      NONE("''"),
      LZ4("'LZ4Compressor'"),
      SNAPPY("'SnappyCompressor'"),
      DEFLATE("'DeflateCompressor'");

      private String value;

      Algorithm(String value) {
        this.value = value;
      }

      public String value() {
        return value;
      }

      @Override
      public String toString() {
        return value;
      }
    }

    public static class NoCompression extends CompressionOptions {

      public NoCompression() {
        super(Algorithm.NONE);
      }

      @Override
      public CompressionOptions withChunkLengthInKb(Integer chunkLengthInKb) {
        return this;
      }

      @Override
      public CompressionOptions withCRCCheckChance(Double crcCheckChance) {
        return this;
      }
    }
  }

  /**
   * The speculative retry options.
   *
   * <p>To create instances, use {@link SchemaBuilder#noSpeculativeRetry()}, {@link
   * SchemaBuilder#always()}, {@link SchemaBuilder#percentile(int)} or {@link
   * SchemaBuilder#millisecs(int)}.
   */
  public static class SpeculativeRetryValue {

    private String value;

    SpeculativeRetryValue(String value) {
      this.value = value;
    }

    String value() {
      return value;
    }
  }

  /**
   * Define the number of rows to be cached per partition when row caching is enabled (this feature
   * is only applicable to Cassandra 2.1.x).
   *
   * <p>To create instances, use {@link SchemaBuilder#noRows()}, {@link SchemaBuilder#allRows()} or
   * {@link SchemaBuilder#rows(int)}.
   */
  public static class CachingRowsPerPartition {
    private String value;

    CachingRowsPerPartition(String value) {
      this.value = value;
    }

    public String value() {
      return value;
    }
  }

  /** Read Repair modes. Possible values: BLOCKING, NONE. */
  public static enum ReadRepairValue {
    BLOCKING("'BLOCKING'"),
    NONE("'NONE'");

    private String value;

    private ReadRepairValue(String value) {
      this.value = value;
    }

    public String value() {
      return value;
    }

    @Override
    public String toString() {
      return value;
    }
  }

  /**
   * Additional Write Policy. Default value is 99p.
   *
   * <p>To create instances, use {@link SchemaBuilder#additionalWritePolicyNone() ()}, {@link
   * SchemaBuilder#additionalWritePolicyAlways()}, {@link
   * SchemaBuilder#additionalWritePolicyMillisecs(int)} or {@link
   * SchemaBuilder#additionalWritePolicyMillisecs(int)}.
   */
  public static class AdditionalWritePolicyValue {
    private String value;

    public AdditionalWritePolicyValue(String value) {
      this.value = value;
    }

    public String value() {
      return value;
    }
  }
}
