/*
 * 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.ambari.server.orm.db;

import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Charsets;
import com.google.common.base.Joiner;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.io.Resources;

/**
 * <p>
 * Utility to parse DDL scripts for verification and comparision. Parsing is not complete, only things relevant for
 * the unit tests are considered, e.g. the column data types are not captured as they may differ across different SQL
 * dialects.
 * </p>
 * <p>
 * Parsing is mostly done by regexp matches, so the parser has limited capabilities. The following known restrictions apply:
 * </p>
 * <ul>
 *   <li>Multiple statements in a single line are not supported, e.g: CREATE TABLE X(...); CREATE TABLE Y(...)</li>
 *   <li>Mutiple definitions in the same line within a create statement is supported though, e.g: name varchar(255), id bigint, ...</li>
 *   <li>Statements must be terminated by ;</li>
 * </ul>
 *
 */
public class DDLTestUtils {
  private static final Logger LOG = LoggerFactory.getLogger(DDLTestUtils.class);

  // These patterns are used during the initial line by line parsing of a DDL.
  // The patterns help
  // 1. filter out irrelevant lines (comment, empty line, go/commit),
  // 2. recognize statement starters (create table, alter table, create index)
  // 3. recognize statement terminators (;)
  private static final Pattern CommentLine =  Pattern.compile("^\\s*--.*");
  private static final Pattern EmptyLine =    Pattern.compile("^\\s*$");
  private static final Pattern CommitLine =   Pattern.compile("^\\s*(go|commit).*$");
  private static final Pattern CreateTable =  Pattern.compile("^\\s*create\\s+table.*$");
  private static final Pattern AlterTable =   Pattern.compile("^\\s*alter\\s+table.*$");
  private static final Pattern CreateIndex =  Pattern.compile("^\\s*create\\s+index.*$");
  private static final Pattern EndStatement = Pattern.compile("(.*\\;)\\s*$");

  // These patterns are used to match column/constraint definitons in a create table statement
  // to capture the table name
  private static final Pattern TableName =   Pattern.compile("^\\s*create table\\s+([\\`\\w\\.\\_\\`]+).*$");
  // to capture the name and columns in a primary key
  private static final Pattern PK =          Pattern.compile("^.*constraint\\s+([\\w\\.\\_]+)\\s+primary\\s+key\\s*\\(([^\\)]+)\\).*$");
  // to capture the name and columns in a clusterd primary key
  private static final Pattern PKClustered = Pattern.compile("^.*constraint\\s+([\\w\\.\\_]+)\\s+primary\\s+key\\s+clustered\\s*\\(([^\\)]+)\\).*$");
  // to capture the name and columns in a unique constraint
  private static final Pattern UQ =          Pattern.compile("^.*constraint\\s+([\\w\\.\\_]+)\\s+unique\\s*\\(([^\\)]+)\\).*$");
  // to capture the name and columns and the referred columnd in a foreign key
  private static final Pattern FK =          Pattern.compile("^.*constraint\\s+([\\w\\.\\_]+)\\s+foreign\\s+key\\s*\\(([^\\)]*)\\)\\s*references\\s+([\\`\\w\\_\\.\\`]+)\\s*\\(([^\\)]+)\\).*$");
  // to capture the name of a columns
  private static final Pattern Col =         Pattern.compile("^\\s*([\\`\\\"\\[\\]\\w\\.\\_]+)\\s+.*$");
  // to capture column lists within the create table statement, such as the column list in "primary key (name, id)"
  // in such lists commas are replaced with a | so that we can reliably use comma as a definition separator within the create table statement
  private static final Pattern InnerList =   Pattern.compile("(\\([^\\(^\\)]+\\))");

  // These patterns represent Unnamed constraints
  private static final Pattern UnnamedPK =          Pattern.compile("^\\s*primary\\s+key[\\sclustered]*\\(([^\\)]+)\\).*$"); // e.g: primary key [clustered] (name)
  private static final Pattern UnnamedUQ =          Pattern.compile("^\\s*unique\\s*\\(([^\\)]+)\\).*$"); // e.g: unique (name)
  private static final Pattern UnnamedFK =          Pattern.compile("^\\s*foreign\\s+key\\s*\\(([^\\)]+)\\)\\s*references\\s+([\\w\\_\\.]+)\\s*\\(([^\\)]+)\\).*$"); // e.g: foreign key (name) references other_table(name)
  private static final Pattern PKColumn =          Pattern.compile("^.*[\\w\\.\\_]+\\s.*primary\\s+key[\\sclustered\\,\\;\\)]*$"); // e.g: name varchar(255) not null primary key [clustered]
  private static final Pattern UQColumn =          Pattern.compile("^\\s*[\\w\\.\\_]+\\s.*unique[\\s\\;\\,\\)]*$"); // e.g: name varchar(255) not null unique
  private static final List<Pattern> CheckedUnnamedConstraints = ImmutableList.of(UnnamedPK);
  private static final List<Pattern> UncheckedUnnamedConstraints = ImmutableList.of(UnnamedUQ, UnnamedFK, PKColumn, UQColumn);

  private static final LoadingCache<String, DDL> ddlCache = CacheBuilder.newBuilder().build(
    new CacheLoader<String, DDL>() {
      @Override public DDL load(String key) throws Exception { return loadDdl(key); }
    }
  );

  /**
   * List of supported databases.
   */
  public static final List<String> DATABASES = ImmutableList.of(
    "Derby",
    "MySQL",
    "Oracle",
    "Postgres",
    "Postgres-EMBEDDED",
    "SQLAnywhere",
    "SQLServer");

  /**
   * Loads and parses the DDL for a specific database type (e.g. Postgres). Already loaded DDL's are cached.
   * @param dbType One of the supported databases @see #DATABASES
   * @return the parsed DDL
   * @throws Exception if exception occurs during loading/parsing
   */
  public static DDL getDdl(String dbType) throws Exception{
    return ddlCache.get(dbType);
  }

  private static URL getDdlUrl(String dbType) {
    return Resources.getResource("Ambari-DDL-" + dbType + "-CREATE.sql");
  }

  private static List<String> loadFile(String dbType) throws Exception {
    List<String> lines = Resources.readLines(getDdlUrl(dbType), Charsets.UTF_8);
    List<String> replaced = new ArrayList<>(lines.size());
    for (String line: lines) { replaced.add(line.toLowerCase()); }
    return replaced;
  }

  /**
   * Groups the load DDL file into statements. Currently CREATE TABLE's, ALTER TABLE's and CREATE INDEX-es
   * are considered statements. Multiple statements in a single line are not supported. Comments, empty lines and
   * GO / COMMIT commands are discarded.
   *
   * @param ddlFile the loaded DDL file (as list of strings)
   * @return a list of strings containing the statements
   */
  private static List<String> groupStatements(List<String> ddlFile) {
    List<String> statements = new ArrayList<>();
    Optional<ArrayList<String>> currentStmt = Optional.absent();
    for (String line: ddlFile) {
      // These lines should be skipped
      if (CommentLine.matcher(line).matches() ||
          EmptyLine.matcher(line).matches() ||
          CommitLine.matcher(line).matches());
      // These lines indicate the start of a CREATE TABLE / ALTER TABLE / CREATE INDEX statement
      else if (CreateTable.matcher(line).matches() ||
          AlterTable.matcher(line).matches() ||
          CreateIndex.matcher(line).matches()) {
        // Prepare to collect subsequent lines as part of the new statement
        if(currentStmt.isPresent()) throw new IllegalStateException(
            "Unfinished statement: " + currentStmt.get() + "\nnew statement: " +line);
        currentStmt = Optional.of(new ArrayList<String>());
        currentStmt.get().add(stripComment(line));
        // If the statement is a one liner, close it right away
        if (line.contains(";")) {
          statements.add(Joiner.on(' ').join(currentStmt.get()));
          currentStmt = Optional.absent();
        }
      }
      // Process terminating line (containing ;): add to the current statement and close current statement
      else if (currentStmt.isPresent() && EndStatement.matcher(line).matches()) {
        currentStmt.get().add(stripComment(line));
        statements.add(Joiner.on(' ').join(currentStmt.get()));
        currentStmt = Optional.absent();
      }
      // Collect all other lines as part of the current statement
      else if (currentStmt.isPresent()) {
        currentStmt.get().add(stripComment(line));
      }
    }
    return statements;
  }

  private static String stripComment(String line) {
    return line.contains("--") ? line.substring(0, line.indexOf("--")) : line;
  }

  private static Collection<String> toColumns(String cols) {
    List<String> columns = new ArrayList<>();
    for (String col: Splitter.on('|').split(cols)) {
      columns.add( stripPrefixQuotationAndBrackets(col.trim()));
    }
    return columns;
  }

  /**
   * Strips out quotation characters ('"[]) and schema prefixes from identifiers
   * (table / columns / constraint names)
   * @param input an identifier
   * @return the important part of the identifier
   */
  private static String stripPrefixQuotationAndBrackets(String input) {
    String output = input.replaceAll("[\\`\\\"\\[\\]]", "").replaceAll("[^\\.]*\\.", "");
    return output;
  }

  private static Optional<String> firstMatchingGroup(Pattern p, String s) {
    Matcher m = p.matcher(s);
    if (m.matches()) {
      return Optional.of(m.group(1));
    }
    else {
      return Optional.absent();
    }
  }

  private static Map<String, Table> parseTableDefs(List<String> statements) {
    // Find all CREATE TABLE statements
    List<String> createTables = new ArrayList<>();
    for (String stmt: statements) {
      if (stmt.matches(".*create\\s+table.*")) {
        String content = stmt.substring(stmt.indexOf('(') + 1, stmt.lastIndexOf(')'));
        // Replace , with | within PK/FK/UQ definitions so that we will be able to partition column/constraint definitions by ,
        Matcher m = InnerList.matcher(content);
        while (m.find()) {
          String innerList = m.group();
          stmt = stmt.replace(innerList, innerList.replaceAll("\\,", "|"));
        }
        createTables.add(stmt);
      }
    }
    List<Table> tables = new ArrayList<>();
    // Parse CREATE TABLE statements
    for(String ct: createTables) {
      String tableName = stripPrefixQuotationAndBrackets(firstMatchingGroup(TableName, ct).get());
      List<String> columns = new ArrayList<>();
      Optional<SimpleConstraint> pk = Optional.absent();
      List<FKConstraint> fks = new ArrayList<>();
      List<SimpleConstraint> uqs = new ArrayList<>();
      final String innerPart = ct.substring(ct.indexOf('(') + 1, ct.lastIndexOf(')'));
      for (String definition: Splitter.on(',').split(innerPart)) {
        definition = definition.trim();
        assertNounnamedConstraint(tableName, definition);
        Matcher pkMatcher = PK.matcher(definition);
        Matcher pkClustMatcher = PKClustered.matcher(definition);
        Matcher unnamedPkMatcher = UnnamedPK.matcher(definition);
        Matcher pkColumnMatcher = PKColumn.matcher(definition);
        Matcher fkMatcher = FK.matcher(definition);
        Matcher uqMatcher = UQ.matcher(definition);
        Matcher unnamedFkMatcher = UnnamedFK.matcher(definition);
        Matcher unnamedUqMatcher = UnnamedUQ.matcher(definition);
        Matcher uqColumnMatcher = UQColumn.matcher(definition);
        Matcher colMatcher = Col.matcher(definition);
        if (pkMatcher.matches()) {
          pk = Optional.of(Constraint.pk(pkMatcher.group(1),toColumns(pkMatcher.group(2))));
        } else if (pkMatcher.matches()) {
          pk = Optional.of(Constraint.pk(stripPrefixQuotationAndBrackets(pkMatcher.group(1)),toColumns(pkMatcher.group(2))));
        } else if (pkClustMatcher.matches()) {
          pk = Optional.of(Constraint.pk(stripPrefixQuotationAndBrackets(pkClustMatcher.group(1)),toColumns(pkClustMatcher.group(2))));
        } else if (unnamedPkMatcher.matches()) {
          pk = Optional.of(Constraint.pk("<default>",toColumns(unnamedPkMatcher.group(1))));
        } else if (fkMatcher.matches()) {
          fks.add(Constraint.fk(fkMatcher.group(1), toColumns(fkMatcher.group(2)), stripPrefixQuotationAndBrackets(fkMatcher.group(3)), toColumns(fkMatcher.group(4))));
        } else if (unnamedFkMatcher.matches()) {
          fks.add(Constraint.fk("<default>", toColumns(unnamedFkMatcher.group(1)), stripPrefixQuotationAndBrackets(unnamedFkMatcher.group(2)), toColumns(unnamedFkMatcher.group(3))));
        } else if (uqMatcher.matches()) {
          uqs.add(Constraint.uq(stripPrefixQuotationAndBrackets(uqMatcher.group(1)),toColumns(uqMatcher.group(2))));
        } else if (unnamedUqMatcher.matches()) {
          uqs.add(Constraint.uq("<default>", toColumns(unnamedUqMatcher.group(1))));
        } else if (colMatcher.matches()) {
          String colName = stripPrefixQuotationAndBrackets(colMatcher.group(1));
          columns.add(colName);
          // column definitions can include PK/UQ declaration, e.g: x integer not null primary key
          if (pkColumnMatcher.matches()) {
            pk = Optional.of(Constraint.pk("<default>", Collections.singleton(colName)));
          } else if (uqColumnMatcher.matches()) {
            uqs.add(Constraint.uq("<default>", Collections.singleton(colName)));
          }
        } else {
          LOG.warn("Unexpected definition: {}, context: {}", definition, ct);
        }
      }
      if (columns.isEmpty()) {
        throw new IllegalStateException("No columns found in table " + tableName);
      }
      checkDupes("columns of table " + tableName, columns);
      tables.add(new Table(tableName,
          ImmutableSet.copyOf(columns),
          pk,
          ImmutableSet.copyOf(fks),
          ImmutableSet.copyOf(uqs)));
    }
    Map<String, Table> tableMap = Maps.newHashMap();
    for(Table t: tables) {
      if (tableMap.containsKey(t.name)) throw new IllegalStateException("Duplicate table definition: " + t.name);
      tableMap.put(t.name, t);
    }
    return tableMap;
  }

  private static void checkDupes(String objectName, List<? extends Object> items) {
    Set<Object> set = Sets.newHashSet(items);
    if (set.size() < items.size()) {
      throw new IllegalStateException(String.format("Duplicates found in %s: %s", objectName, Iterables.toString(items)));
    }
  }

  /**
   * Currently we only fail on unnamed primary keys.
   * @param tableName
   * @param definition
   */
  private static void assertNounnamedConstraint(String tableName, String definition) {
    if (tableName.contains("qrtz")) {
      LOG.debug("Skipp checking quartz table: {}", tableName);
    }
    else {
      for (Pattern unnamedConstraint: CheckedUnnamedConstraints) {
        if (unnamedConstraint.matcher(definition).matches()) {
          throw new IllegalStateException(
              String.format("Found invalid (unnamed) constraint in table %s: %s", tableName, definition));
        }
      }
      for (Pattern unnamedConstraint: UncheckedUnnamedConstraints) {
        if (unnamedConstraint.matcher(definition).matches()) {
          LOG.info("Found unnamed constraint in table {}: {}", tableName, definition);
        }
      }
    }
  }

  private static DDL loadDdl(String dbType) throws Exception {
    List<String> lines = loadFile(dbType);
    List<String> statements = groupStatements(lines);
    Map<String, Table> tables = parseTableDefs(statements);
    List<String> alterTables = new ArrayList<>();
    for (String stmt: statements) {
      if (stmt.matches(".*alter\\s+table.*")) alterTables.add(stmt);
    }
    return new DDL(dbType, tables, alterTables);
  }

}

/**
 * Represents a DDL
 */
class DDL {
  final String dbType;
  final Map<String, Table> tables;
  final List<String> alterTables;

  Set<String> tableNames() { return tables.keySet(); }

  DDL(String dbType, Map<String, Table> tables, List<String> alterTables) {
    this.dbType = dbType;
    this.tables = tables;
    this.alterTables = alterTables;
  }
}

/**
 * Represents a datbase table
 */
class Table {
  final String name;
  final ImmutableSet<String> columns;
  final Optional<SimpleConstraint> primaryKey;
  final ImmutableSet<FKConstraint> foreignKeys;
  final ImmutableSet<SimpleConstraint> uniqueConstraints;

  Table(String name, Set<String> columns, Optional<SimpleConstraint> primaryKey, Set<FKConstraint> foreignKeys, Set<SimpleConstraint> uniqueConstraints) {
    this.name = name;
    this.columns =
        (columns instanceof ImmutableSet) ? (ImmutableSet<String>)columns : ImmutableSet.copyOf(columns);
    this.primaryKey = primaryKey;
    this.foreignKeys =
        (foreignKeys instanceof ImmutableSet) ? (ImmutableSet<FKConstraint>)foreignKeys : ImmutableSet.copyOf(foreignKeys);
    this.uniqueConstraints =
        (uniqueConstraints instanceof ImmutableSet) ? (ImmutableSet<SimpleConstraint>) uniqueConstraints : ImmutableSet.copyOf(uniqueConstraints);
  }

  @Override
  public String toString() {
    return String.format("TABLE name: %s, columns: %s, pk: %s, fks: %s, uqs: %s",
        name, Iterables.toString(columns), primaryKey, Iterables.toString(foreignKeys), Iterables.toString(uniqueConstraints));
  }
}

/**
 * Represents a constraint.
 */
abstract class Constraint<ContentType> {
  abstract String name();
  abstract ContentType content();

  static SimpleConstraint pk(String name, Collection<String> columns) {
    Preconditions.checkArgument(!columns.isEmpty(), "Columns must not be empty.");
    return new SimpleConstraint(name, "PK", columns);
  }

  static SimpleConstraint uq(String name, Collection<String> columns) {
    Preconditions.checkArgument(!columns.isEmpty(), "Columns must not be empty.");
    return new SimpleConstraint(name, "PK", columns);
  }

  static FKConstraint fk(String name, Collection<String> columns, String referredTableName, Collection<String> referredColumns) {
    Preconditions.checkArgument(!columns.isEmpty(), "Columns must not be empty.");
    Preconditions.checkArgument(!referredColumns.isEmpty(), "Referred columns must not be empty.");
    return new FKConstraint(name, columns, referredTableName, referredColumns);
  }

}

/**
 * Represents a simple constraint (PK/UQ)
 */
class SimpleConstraint extends Constraint<Set<String>> {
  final String name;
  final String type;
  final ImmutableSet<String> columns; // These have favorable equals/hashcode semantics

  SimpleConstraint(String name, String type, Collection<String> columns) {
    this.name = name;
    this.type = type;
    this.columns = (columns instanceof ImmutableSet) ? (ImmutableSet<String>) columns : ImmutableSet.copyOf(columns);
  }

  public String name() {
    return name;
  }

  public ImmutableSet<String> content() { return columns; }

  @Override public String toString() {
    return String.format("%s %s [%s]", type, name, Joiner.on(',').join(columns));
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) return true;
    if (o == null || getClass() != o.getClass()) return false;
    SimpleConstraint that = (SimpleConstraint) o;
    if (!name.equals(that.name)) return false;
    if (!type.equals(that.type)) return false;
    return columns.equals(that.columns);
  }

  @Override
  public int hashCode() {
    int result = name.hashCode();
    result = 31 * result + type.hashCode();
    result = 31 * result + columns.hashCode();
    return result;
  }

}

class FKConstraintContent {
  final ImmutableSet<String> columns; // These have favorable equals/hashcode semantics
  final String referredTable;
  final ImmutableSet<String> referredColumns; // These have favorable equals/hashcode semantics

  public FKConstraintContent(Collection<String> columns, String referredTable, Collection<String> referredColumns) {
    this.columns = columns instanceof ImmutableSet ? (ImmutableSet<String>)columns : ImmutableSet.copyOf(columns);
    this.referredTable = referredTable;
    this.referredColumns = referredColumns instanceof ImmutableSet ? (ImmutableSet<String>)referredColumns :
        ImmutableSet.copyOf(referredColumns);
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) return true;
    if (o == null || getClass() != o.getClass()) return false;
    FKConstraintContent that = (FKConstraintContent) o;
    if (!columns.equals(that.columns)) return false;
    if (!referredTable.equals(that.referredTable)) return false;
    return referredColumns.equals(that.referredColumns);

  }

  @Override
  public int hashCode() {
    int result = columns.hashCode();
    result = 31 * result + referredTable.hashCode();
    result = 31 * result + referredColumns.hashCode();
    return result;
  }

  @Override public String toString() {
    return String.format("[%s] --> %s [%s]", Joiner.on(',').join(columns), referredTable, Joiner.on(',').join(referredColumns));
  }

}

class FKConstraint extends Constraint<FKConstraintContent> {
  final String name;
  final FKConstraintContent content;

  FKConstraint(String name, Collection<String> columns, String referredTable, Collection<String> referredColumns) {
    this.name = name;
    this.content = new FKConstraintContent(columns, referredTable, referredColumns);
  }

  public String name() {
    return name;
  }

  public FKConstraintContent content() {
    return content;
  }

  @Override public String toString() {
    return String.format("FK name:%s content: %s", name, content);
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) return true;
    if (o == null || getClass() != o.getClass()) return false;
    FKConstraint that = (FKConstraint) o;
    if (!name.equals(that.name)) return false;
    return content.equals(that.content);
  }

  @Override
  public int hashCode() {
    int result = name.hashCode();
    result = 31 * result + content.hashCode();
    return result;
  }
}
