package org.verdictdb.core;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.verdictdb.JdbcResultSet;
import org.verdictdb.connection.DataTypeConverter;
import org.verdictdb.connection.DbmsConnection;
import org.verdictdb.connection.DbmsQueryResult;
import org.verdictdb.sql.syntax.SyntaxAbstract;

public class DbmsMetadataCache {

  private SyntaxAbstract syntax;

  private DbmsConnection connection;

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

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

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

  private List<Pair<String,Integer>> columnsCache = new ArrayList<>();

  public DbmsMetadataCache(DbmsConnection connection) {
    this.syntax = connection.getSyntax();
    this.connection = connection;
  }
  
  public List<String> getSchemas() throws SQLException {
    if (!schemaCache.isEmpty()){
      return schemaCache;
    }
    schemaCache.addAll(connection.getSchemas());
    return schemaCache;
  }
  
  public List<String> getTables(String schema) throws SQLException {
    if (!tablesCache.isEmpty()){
      return tablesCache;
    }
    tablesCache.addAll(connection.getTables(schema));
    return tablesCache;
  }

  public List<Pair<String, Integer>> getColumns(String schema, String table) throws SQLException {
    if (!columnsCache.isEmpty()){
      return columnsCache;
    }
    columnsCache.addAll(connection.getColumns(schema, table));
    return columnsCache;
  }
  
  /**
   * Only needed for the DBMS that supports partitioning.
   * 
   * @param schema
   * @param table
   * @return
   */
  public List<String> getPartitionColumns(String schema, String table) throws SQLException {
    if (!syntax.doesSupportTablePartitioning()) {
      throw new SQLException("Database does not support table partitioning");
    }
    if (!partitionCache.isEmpty()){
      return partitionCache;
    }
    partitionCache.addAll(getPartitionColumns(schema, table));
    return partitionCache;
  }

}

@@ -624,32 +624,48 @@
 %3C%3E();%0A%0A  private
+ HashMap%3CString,
  List%3CString%3E ta
@@ -661,16 +661,17 @@
 %3CString%3E
+%3E
  tablesC
@@ -677,33 +677,31 @@
 Cache = new 
-ArrayList
+HashMap
 %3C%3E();%0A%0A  pri
@@ -760,24 +760,54 @@
 ;%0A%0A  private
+ HashMap%3CPair%3CString, String%3E,
  List%3CPair%3CS
@@ -821,16 +821,17 @@
 nteger%3E%3E
+%3E
  columns
@@ -838,33 +838,31 @@
 Cache = new 
-ArrayList
+HashMap
 %3C%3E();%0A%0A  pub
@@ -1075,32 +1075,33 @@
 Cache.isEmpty())
+ 
 %7B%0A      return s
@@ -1275,16 +1275,49 @@
     if (
+tablesCache.containsKey(schema)&&
 !tablesC
@@ -1317,24 +1317,36 @@
 tablesCache.
+get(schema).
 isEmpty())%7B%0A
@@ -1335,32 +1335,33 @@
 hema).isEmpty())
+ 
 %7B%0A      return t
@@ -1366,24 +1366,36 @@
  tablesCache
+.get(schema)
 ;%0A    %7D%0A    
@@ -1402,31 +1402,36 @@
 tablesCache.
-addAll(
+put(schema, 
 connection.g
@@ -1467,24 +1467,36 @@
  tablesCache
+.get(schema)
 ;%0A  %7D%0A%0A  pub
@@ -1593,20 +1593,119 @@
 n %7B%0A    
-if (
+Pair%3CString, String%3E key = new ImmutablePair%3C%3E(schema,table);%0A    if (columnsCache.containsKey(key) && 
 !columns
@@ -1706,24 +1706,33 @@
 olumnsCache.
+get(key).
 isEmpty())%7B%0A
@@ -1721,32 +1721,33 @@
 (key).isEmpty())
+ 
 %7B%0A      return c
@@ -1753,24 +1753,33 @@
 columnsCache
+.get(key)
 ;%0A    %7D%0A    
@@ -1787,31 +1787,33 @@
 olumnsCache.
-addAll(
+put(key, 
 connection.g
@@ -1858,24 +1858,33 @@
 columnsCache
+.get(key)
 ;%0A  %7D%0A  %0A  /
@@ -2267,16 +2267,17 @@
 Empty())
+ 
 %7B%0A      

Make DbmsMetaDataCache independent of Jdbc