/*
 * 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 static com.datastax.driver.core.schemabuilder.SchemaBuilder.dateTieredStrategy;
import static com.datastax.driver.core.schemabuilder.SchemaBuilder.percentile;
import static com.datastax.driver.core.schemabuilder.SchemaBuilder.rows;
import static com.datastax.driver.core.schemabuilder.SchemaBuilder.snappy;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;

import com.datastax.driver.core.CCMTestsSupport;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.schemabuilder.TableOptions.CompactionOptions.DateTieredCompactionStrategyOptions.TimeStampResolution;
import com.datastax.driver.core.utils.CassandraVersion;
import java.util.Iterator;
import org.testng.annotations.Test;

public class SchemaBuilderIT extends CCMTestsSupport {

  @Test(groups = "short")
  @CassandraVersion("2.1.2")
  public void should_modify_table_metadata() {
    // Create a table
    session()
        .execute(
            SchemaBuilder.createTable("ks", "TableMetadata")
                .addPartitionKey("a", DataType.cint())
                .addPartitionKey("b", DataType.cint())
                .addClusteringColumn("c", DataType.cint())
                .addClusteringColumn("d", DataType.cint())
                .withOptions()
                .compactStorage());

    // Modify the table metadata
    session()
        .execute(
            SchemaBuilder.alterTable("TableMetadata")
                .withOptions()
                .defaultTimeToLive(1337)
                .bloomFilterFPChance(0.42)
                .caching(SchemaBuilder.KeyCaching.ALL, rows(1))
                .gcGraceSeconds(1234567890)
                .minIndexInterval(6)
                .indexInterval(7)
                .maxIndexInterval(8)
                .comment("Useful comment")
                .readRepairChance(0.123456)
                .speculativeRetry(percentile(50))
                .dcLocalReadRepairChance(0.84)
                .memtableFlushPeriodInMillis(1234567890)
                .compactionOptions(
                    dateTieredStrategy()
                        .baseTimeSeconds(1)
                        .minThreshold(2)
                        .maxThreshold(3)
                        .maxSSTableAgeDays(4)
                        .timestampResolution(TimeStampResolution.MILLISECONDS))
                .compressionOptions(snappy()));

    // Retrieve the metadata from Cassandra
    ResultSet rows =
        session()
            .execute(
                "SELECT "
                    + "bloom_filter_fp_chance, "
                    + "caching, "
                    + "cf_id, "
                    + "column_aliases, "
                    + "comment, "
                    + "compaction_strategy_class, "
                    + "compaction_strategy_options, "
                    + "comparator, "
                    + "compression_parameters, "
                    + "default_time_to_live, "
                    + "default_validator, "
                    + "dropped_columns, "
                    + "gc_grace_seconds, "
                    + "index_interval, "
                    + "is_dense, "
                    + "key_aliases, "
                    + "key_validator, "
                    + "local_read_repair_chance, "
                    + "max_compaction_threshold, "
                    + "max_index_interval, "
                    + "memtable_flush_period_in_ms, "
                    + "min_compaction_threshold, "
                    + "min_index_interval, "
                    + "read_repair_chance, "
                    + "speculative_retry, "
                    + "subcomparator, "
                    + "type, "
                    + "value_alias "
                    + "FROM system.schema_columnfamilies "
                    + "WHERE keyspace_name='ks' AND columnfamily_name='tablemetadata'");
    for (Row row : rows) {
      // There should be only one row
      // Verify that every property we modified is correctly set
      assertThat(row.getDouble("bloom_filter_fp_chance")).isEqualTo(0.42);
      assertThat(row.getString("caching"))
          .isEqualTo("{\"keys\":\"ALL\", \"rows_per_partition\":\"1\"}");
      assertThat(row.getUUID("cf_id")).isNotNull();
      assertThat(row.getString("column_aliases")).isEqualTo("[\"c\",\"d\"]");
      assertThat(row.getString("comment")).isEqualTo("Useful comment");
      assertThat(row.getString("compaction_strategy_class"))
          .isEqualTo("org.apache.cassandra.db.compaction.DateTieredCompactionStrategy");
      assertThat(row.getString("compaction_strategy_options"))
          .isEqualTo(
              "{\"base_time_seconds\":\"1\",\"timestamp_resolution\":\"MILLISECONDS\",\"max_sstable_age_days\":\"4\",\"min_threshold\":\"2\",\"max_threshold\":\"3\"}");
      assertThat(row.getString("compression_parameters"))
          .isEqualTo(
              "{\"sstable_compression\":\"org.apache.cassandra.io.compress.SnappyCompressor\"}");
      assertThat(row.getInt("default_time_to_live")).isEqualTo(1337);
      assertThat(row.getInt("gc_grace_seconds")).isEqualTo(1234567890);
      assertThat(row.getInt("min_index_interval")).isEqualTo(6);
      assertThat(row.getInt("max_index_interval")).isEqualTo(8);
      assertThat(row.getString("key_aliases")).isEqualTo("[\"a\",\"b\"]");
      assertThat(row.getDouble("local_read_repair_chance")).isEqualTo(0.84);
      assertThat(row.getInt("max_compaction_threshold")).isEqualTo(3);
      assertThat(row.getInt("memtable_flush_period_in_ms")).isEqualTo(1234567890);
      assertThat(row.getInt("min_compaction_threshold")).isEqualTo(2);
      assertThat(row.getDouble("read_repair_chance")).isEqualTo(0.123456);
      assertThat(row.getString("speculative_retry")).isEqualTo("50.0PERCENTILE");
    }
  }

  @Test(groups = "short")
  @CassandraVersion("2.1.0")
  public void should_create_a_table_and_a_udt() {
    // Create a UDT and a table
    session()
        .execute(SchemaBuilder.createType("MyUDT").ifNotExists().addColumn("x", DataType.cint()));
    UDTType myUDT = UDTType.frozen("MyUDT");
    session()
        .execute(
            SchemaBuilder.createTable("ks", "CreateTable")
                .ifNotExists()
                .addPartitionKey("a", DataType.cint())
                .addUDTPartitionKey("b", myUDT)
                .addClusteringColumn("c", DataType.ascii())
                .addUDTClusteringColumn("d", myUDT)
                .addUDTColumn("e", myUDT)
                .addStaticColumn("f", DataType.bigint())
                .addUDTStaticColumn("g", myUDT)
                .addUDTListColumn("h", myUDT)
                .addUDTMapColumn("i", DataType.cboolean(), myUDT)
                .addUDTMapColumn("j", myUDT, DataType.cboolean())
                .addUDTSetColumn("k", myUDT));

    // Check columns a to k
    ResultSet rows =
        session()
            .execute(
                "SELECT column_name, type, validator "
                    + "FROM system.schema_columns "
                    + "WHERE keyspace_name='ks' AND columnfamily_name='createtable'");
    Iterator<Row> iterator = rows.iterator();
    verifyNextColumnDefinition(
        iterator, "a", "partition_key", "org.apache.cassandra.db.marshal.Int32Type");
    verifyNextColumnDefinition(
        iterator, "b", "partition_key", "org.apache.cassandra.db.marshal.UserType");
    verifyNextColumnDefinition(
        iterator, "c", "clustering_key", "org.apache.cassandra.db.marshal.AsciiType");
    verifyNextColumnDefinition(
        iterator, "d", "clustering_key", "org.apache.cassandra.db.marshal.UserType");
    verifyNextColumnDefinition(
        iterator, "e", "regular", "org.apache.cassandra.db.marshal.UserType");
    verifyNextColumnDefinition(iterator, "f", "static", "org.apache.cassandra.db.marshal.LongType");
    verifyNextColumnDefinition(iterator, "g", "static", "org.apache.cassandra.db.marshal.UserType");
    verifyNextColumnDefinition(
        iterator,
        "h",
        "regular",
        "org.apache.cassandra.db.marshal.ListType",
        "org.apache.cassandra.db.marshal.UserType");
    verifyNextColumnDefinition(
        iterator,
        "i",
        "regular",
        "org.apache.cassandra.db.marshal.MapType",
        "org.apache.cassandra.db.marshal.BooleanType",
        "org.apache.cassandra.db.marshal.UserType");
    verifyNextColumnDefinition(
        iterator,
        "j",
        "regular",
        "org.apache.cassandra.db.marshal.MapType",
        "org.apache.cassandra.db.marshal.UserType",
        "org.apache.cassandra.db.marshal.BooleanType");
    verifyNextColumnDefinition(
        iterator,
        "k",
        "regular",
        "org.apache.cassandra.db.marshal.SetType",
        "org.apache.cassandra.db.marshal.UserType");
  }

  @Test(groups = "short")
  public void should_add_and_drop_a_column() {
    // Create a table, add a column to it with an alter table statement and delete that column
    session()
        .execute(
            SchemaBuilder.createTable("ks", "DropColumn")
                .ifNotExists()
                .addPartitionKey("a", DataType.cint()));

    // Add and then drop a column
    session()
        .execute(SchemaBuilder.alterTable("ks", "DropColumn").addColumn("b").type(DataType.cint()));
    session().execute(SchemaBuilder.alterTable("ks", "DropColumn").dropColumn("b"));

    // Check that only column a exist
    ResultSet rows =
        session()
            .execute(
                "SELECT column_name, type, validator "
                    + "FROM system.schema_columns "
                    + "WHERE keyspace_name='ks' AND columnfamily_name='dropcolumn'");
    Iterator<Row> iterator = rows.iterator();
    verifyNextColumnDefinition(
        iterator, "a", "partition_key", "org.apache.cassandra.db.marshal.Int32Type");
    assertThat(iterator.hasNext()).isFalse();
  }

  private void verifyNextColumnDefinition(
      Iterator<Row> rowIterator, String columnName, String type, String... validatorFragments) {
    Row rowA = rowIterator.next();
    assertThat(rowA.getString("column_name")).isEqualTo(columnName);
    assertThat(rowA.getString("type")).isEqualTo(type);
    for (String validatorFragment : validatorFragments) {
      assertThat(rowA.getString("validator")).contains(validatorFragment);
    }
  }

  @Test(groups = "short")
  public void should_drop_a_table() {
    // Create a table
    session()
        .execute(
            SchemaBuilder.createTable("ks", "DropTable").addPartitionKey("a", DataType.cint()));

    // Drop the table
    session().execute(SchemaBuilder.dropTable("ks", "DropTable"));
    session().execute(SchemaBuilder.dropTable("DropTable").ifExists());

    ResultSet rows =
        session()
            .execute(
                "SELECT columnfamily_name "
                    + "FROM system.schema_columnfamilies "
                    + "WHERE keyspace_name='ks' AND columnfamily_name='droptable'");
    if (rows.iterator().hasNext()) {
      fail("This table should have been deleted");
    }
  }

  @Test(groups = "short")
  public void should_create_an_index() {
    // Create a table
    session()
        .execute(
            SchemaBuilder.createTable("ks", "CreateIndex")
                .addPartitionKey("a", DataType.cint())
                .addClusteringColumn("b", DataType.cint())
                .addColumn("c", DataType.map(DataType.cint(), DataType.cint())));

    // Create an index on a regular column of the table
    session()
        .execute(SchemaBuilder.createIndex("ks_Index").onTable("ks", "CreateIndex").andColumn("b"));
    session()
        .execute(
            SchemaBuilder.createIndex("ks_IndexOnMap")
                .onTable("ks", "CreateIndex")
                .andKeysOfColumn("c"));

    // Verify that the indexes exist on the right columns
    ResultSet rows =
        session()
            .execute(
                "SELECT column_name, index_name, index_options, index_type, component_index "
                    + "FROM system.schema_columns "
                    + "WHERE keyspace_name='ks' "
                    + "AND columnfamily_name='createindex' "
                    + "AND column_name IN ('b', 'c')");
    Iterator<Row> iterator = rows.iterator();
    verifyNextIndexDefinition(iterator, "ks_Index", "{}", "COMPOSITES", 0);
    verifyNextIndexDefinition(iterator, "ks_IndexOnMap", "{\"index_keys\":\"\"}", "COMPOSITES", 1);
    assertThat(iterator.hasNext()).isFalse();
  }

  private void verifyNextIndexDefinition(
      Iterator<Row> iterator, String name, String options, String type, int index) {
    Row nextIndex = iterator.next();
    assertThat(nextIndex.getString("index_name")).isEqualTo(name);
    assertThat(nextIndex.getString("index_options")).isEqualTo(options);
    assertThat(nextIndex.getString("index_type")).isEqualTo(type);
    assertThat(nextIndex.getInt("component_index")).isEqualTo(index);
  }

  @Test(groups = "short")
  public void should_drop_an_index() {
    // Create a table
    session()
        .execute(
            SchemaBuilder.createTable("ks", "DropIndex")
                .addPartitionKey("a", DataType.cint())
                .addClusteringColumn("b", DataType.cint()));

    // Create an index
    // Note: we have to pick a lower-case name because Cassandra uses the CamelCase index name at
    // creation
    // but a lowercase index name at deletion
    // See : https://issues.apache.org/jira/browse/CASSANDRA-8365
    session()
        .execute(SchemaBuilder.createIndex("ks_index").onTable("ks", "DropIndex").andColumn("b"));

    // Verify that the PK index and the secondary indexes both exist
    assertThat(numberOfIndexedColumns()).isEqualTo(1);

    // Delete the index
    session().execute(SchemaBuilder.dropIndex("ks", "ks_index"));

    // Verify that only the PK index exists
    assertThat(numberOfIndexedColumns()).isEqualTo(0);
  }

  private int numberOfIndexedColumns() {
    ResultSet columns =
        session()
            .execute(
                "SELECT * "
                    + "FROM system.schema_columns "
                    + "WHERE keyspace_name='ks' "
                    + "AND columnfamily_name='dropindex' ");
    int count = 0;
    for (Row column : columns) {
      if (column.getString("index_name") != null) count += 1;
    }
    return count;
  }
}
