Return-Path: 
 <commits-return-4229-apmail-tajo-commits-archive=tajo.apache.org@tajo.apache.org>
X-Original-To: apmail-tajo-commits-archive@minotaur.apache.org
Delivered-To: apmail-tajo-commits-archive@minotaur.apache.org
Received: from mail.apache.org (hermes.apache.org [140.211.11.3])
	by minotaur.apache.org (Postfix) with SMTP id 230DE18A5C
	for <apmail-tajo-commits-archive@minotaur.apache.org>;
 Thu, 30 Jul 2015 04:39:00 +0000 (UTC)
Received: (qmail 66790 invoked by uid 500); 30 Jul 2015 04:39:00 -0000
Delivered-To: apmail-tajo-commits-archive@tajo.apache.org
Received: (qmail 66696 invoked by uid 500); 30 Jul 2015 04:39:00 -0000
Mailing-List: contact commits-help@tajo.apache.org; run by ezmlm
Precedence: bulk
List-Help: <mailto:commits-help@tajo.apache.org>
List-Unsubscribe: <mailto:commits-unsubscribe@tajo.apache.org>
List-Post: <mailto:commits@tajo.apache.org>
List-Id: <commits.tajo.apache.org>
Reply-To: dev@tajo.apache.org
Delivered-To: mailing list commits@tajo.apache.org
Received: (qmail 66561 invoked by uid 99); 30 Jul 2015 04:38:59 -0000
Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org)
 (140.211.11.23)
    by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 30 Jul 2015 04:38:59 +0000
Received: by git1-us-west.apache.org (ASF Mail Server at
 git1-us-west.apache.org, from userid 33)
	id C0757E6824; Thu, 30 Jul 2015 04:38:59 +0000 (UTC)
Content-Type: text/plain; charset="us-ascii"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
From: jihoonson@apache.org
To: commits@tajo.apache.org
Date: Thu, 30 Jul 2015 04:39:02 -0000
Message-Id: <037fe2a2b03a40b3911f954d953ee715@git.apache.org>
In-Reply-To: <997c85b40af34d339f70614ea1a88f22@git.apache.org>
References: <997c85b40af34d339f70614ea1a88f22@git.apache.org>
X-Mailer: ASF-Git Admin Mailer
Subject: [4/5] tajo git commit: TAJO-1300: Merge the index branch into the
 master branch.

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
index c6b7d36..2e9c340 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
@@ -27,9 +27,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.tajo.annotation.Nullable;
-import org.apache.tajo.catalog.CatalogConstants;
-import org.apache.tajo.catalog.CatalogUtil;
-import org.apache.tajo.catalog.FunctionDesc;
+import org.apache.tajo.catalog.*;
 import org.apache.tajo.catalog.exception.*;
 import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.catalog.proto.CatalogProtos.*;
@@ -2177,36 +2175,55 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
     Connection conn = null;
     PreparedStatement pstmt = null;
 
-    String databaseName = proto.getTableIdentifier().getDatabaseName();
-    String tableName = proto.getTableIdentifier().getTableName();
-    String columnName = CatalogUtil.extractSimpleName(proto.getColumn().getName());
+    final String databaseName = proto.getTableIdentifier().getDatabaseName();
+    final String tableName = CatalogUtil.extractSimpleName(proto.getTableIdentifier().getTableName());
 
     try {
+      // indexes table
       int databaseId = getDatabaseId(databaseName);
       int tableId = getTableId(databaseId, databaseName, tableName);
 
       String sql = "INSERT INTO " + TB_INDEXES +
           " (" + COL_DATABASES_PK + ", " + COL_TABLES_PK + ", INDEX_NAME, " +
-          "" + COL_COLUMN_NAME + ", DATA_TYPE, INDEX_TYPE, IS_UNIQUE, IS_CLUSTERED, IS_ASCENDING) " +
-        "VALUES (?,?,?,?,?,?,?,?,?)";
+          "INDEX_TYPE, PATH, COLUMN_NAMES, DATA_TYPES, ORDERS, NULL_ORDERS, IS_UNIQUE, IS_CLUSTERED) " +
+          "VALUES (?,?,?,?,?,?,?,?,?,?,?)";
 
       if (LOG.isDebugEnabled()) {
         LOG.debug(sql);
       }
 
+      StringBuilder columnNamesBuilder = new StringBuilder();
+      StringBuilder dataTypesBuilder= new StringBuilder();
+      StringBuilder ordersBuilder = new StringBuilder();
+      StringBuilder nullOrdersBuilder = new StringBuilder();
+      for (SortSpecProto columnSpec : proto.getKeySortSpecsList()) {
+        // Since the key columns are always sorted in order of their occurrence position in the relation schema,
+        // the concatenated name can be uniquely identified.
+        columnNamesBuilder.append(columnSpec.getColumn().getName()).append(",");
+        dataTypesBuilder.append(columnSpec.getColumn().getDataType().getType().name()).append(",");
+        ordersBuilder.append(columnSpec.getAscending()).append(",");
+        nullOrdersBuilder.append(columnSpec.getNullFirst()).append(",");
+      }
+      columnNamesBuilder.deleteCharAt(columnNamesBuilder.length()-1);
+      dataTypesBuilder.deleteCharAt(dataTypesBuilder.length()-1);
+      ordersBuilder.deleteCharAt(ordersBuilder.length()-1);
+      nullOrdersBuilder.deleteCharAt(nullOrdersBuilder.length()-1);
+
       conn = getConnection();
       conn.setAutoCommit(false);
 
       pstmt = conn.prepareStatement(sql);
       pstmt.setInt(1, databaseId);
       pstmt.setInt(2, tableId);
-      pstmt.setString(3, proto.getIndexName());
-      pstmt.setString(4, columnName);
-      pstmt.setString(5, proto.getColumn().getDataType().getType().name());
-      pstmt.setString(6, proto.getIndexMethod().toString());
-      pstmt.setBoolean(7, proto.hasIsUnique() && proto.getIsUnique());
-      pstmt.setBoolean(8, proto.hasIsClustered() && proto.getIsClustered());
-      pstmt.setBoolean(9, proto.hasIsAscending() && proto.getIsAscending());
+      pstmt.setString(3, proto.getIndexName()); // index name
+      pstmt.setString(4, proto.getIndexMethod().toString()); // index type
+      pstmt.setString(5, proto.getIndexPath()); // index path
+      pstmt.setString(6, columnNamesBuilder.toString());
+      pstmt.setString(7, dataTypesBuilder.toString());
+      pstmt.setString(8, ordersBuilder.toString());
+      pstmt.setString(9, nullOrdersBuilder.toString());
+      pstmt.setBoolean(10, proto.hasIsUnique() && proto.getIsUnique());
+      pstmt.setBoolean(11, proto.hasIsClustered() && proto.getIsClustered());
       pstmt.executeUpdate();
       conn.commit();
     } catch (SQLException se) {
@@ -2260,9 +2277,7 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
   }
 
   final static String GET_INDEXES_SQL =
-      "SELECT " + COL_TABLES_PK + ", INDEX_NAME, COLUMN_NAME, DATA_TYPE, INDEX_TYPE, IS_UNIQUE, " +
-          "IS_CLUSTERED, IS_ASCENDING FROM " + TB_INDEXES;
-
+      "SELECT * FROM " + TB_INDEXES;
 
   @Override
   public IndexDescProto getIndexByName(String databaseName, final String indexName)
@@ -2293,6 +2308,7 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
       resultToIndexDescProtoBuilder(builder, res);
       String tableName = getTableName(conn, res.getInt(COL_TABLES_PK));
       builder.setTableIdentifier(CatalogUtil.buildTableIdentifier(databaseName, tableName));
+      builder.setTargetRelationSchema(getTable(databaseName, tableName).getSchema());
       proto = builder.build();
     } catch (SQLException se) {
       throw new TajoInternalError(se);
@@ -2304,9 +2320,8 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
   }
 
   @Override
-  public IndexDescProto getIndexByColumn(final String databaseName,
-                                         final String tableName,
-                                         final String columnName) throws CatalogException {
+  public IndexDescProto getIndexByColumns(String databaseName, String tableName, String[] columnNames)
+      throws CatalogException {
     Connection conn = null;
     ResultSet res = null;
     PreparedStatement pstmt = null;
@@ -2314,25 +2329,35 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
 
     try {
       int databaseId = getDatabaseId(databaseName);
+      int tableId = getTableId(databaseId, databaseName, tableName);
+      TableDescProto tableDescProto = getTable(databaseName, tableName);
 
-      String sql = GET_INDEXES_SQL + " WHERE " + COL_DATABASES_PK + "=? AND COLUMN_NAME=?";
+      String sql = GET_INDEXES_SQL + " WHERE " + COL_DATABASES_PK + "=? AND " +
+          COL_TABLES_PK + "=? AND COLUMN_NAMES=?";
 
       if (LOG.isDebugEnabled()) {
         LOG.debug(sql);
       }
 
+      // Since the column names in the unified name are always sorted
+      // in order of occurrence position in the relation schema,
+      // they can be uniquely identified.
+      String unifiedName = CatalogUtil.buildFQName(databaseName, tableName,
+          CatalogUtil.getUnifiedSimpleColumnName(new Schema(tableDescProto.getSchema()), columnNames));
       conn = getConnection();
       pstmt = conn.prepareStatement(sql);
       pstmt.setInt(1, databaseId);
-      ;
-      pstmt.setString(2, columnName);
+      pstmt.setInt(2, tableId);
+      pstmt.setString(3, unifiedName);
       res = pstmt.executeQuery();
       if (!res.next()) {
-        throw new TajoInternalError("ERROR: there is no index matched to " + columnName);
+        throw new TajoInternalError("ERROR: there is no index matched to " + unifiedName);
       }
+
       IndexDescProto.Builder builder = IndexDescProto.newBuilder();
       resultToIndexDescProtoBuilder(builder, res);
       builder.setTableIdentifier(CatalogUtil.buildTableIdentifier(databaseName, tableName));
+      builder.setTargetRelationSchema(tableDescProto.getSchema());
       proto = builder.build();
     } catch (SQLException se) {
       throw new TajoInternalError(se);
@@ -2377,7 +2402,7 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
   }
 
   @Override
-  public boolean existIndexByColumn(String databaseName, String tableName, String columnName)
+  public boolean existIndexByColumns(String databaseName, String tableName, String[] columnNames)
       throws CatalogException {
     Connection conn = null;
     ResultSet res = null;
@@ -2387,18 +2412,27 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
 
     try {
       int databaseId = getDatabaseId(databaseName);
+      int tableId = getTableId(databaseId, databaseName, tableName);
+      Schema relationSchema = new Schema(getTable(databaseName, tableName).getSchema());
 
       String sql =
-          "SELECT INDEX_NAME FROM " + TB_INDEXES + " WHERE " + COL_DATABASES_PK + "=? AND COLUMN_NAME=?";
+          "SELECT " + COL_INDEXES_PK + " FROM " + TB_INDEXES +
+              " WHERE " + COL_DATABASES_PK + "=? AND " + COL_TABLES_PK + "=? AND COLUMN_NAMES=?";
 
       if (LOG.isDebugEnabled()) {
         LOG.debug(sql);
       }
 
+      // Since the column names in the unified name are always sorted
+      // in order of occurrence position in the relation schema,
+      // they can be uniquely identified.
+      String unifiedName = CatalogUtil.buildFQName(databaseName, tableName,
+          CatalogUtil.getUnifiedSimpleColumnName(new Schema(relationSchema), columnNames));
       conn = getConnection();
       pstmt = conn.prepareStatement(sql);
       pstmt.setInt(1, databaseId);
-      pstmt.setString(2, columnName);
+      pstmt.setInt(2, tableId);
+      pstmt.setString(3, unifiedName);
       res = pstmt.executeQuery();
       exist = res.next();
     } catch (SQLException se) {
@@ -2410,22 +2444,18 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
   }
 
   @Override
-  public IndexDescProto[] getIndexes(String databaseName, final String tableName)
+  public List<String> getAllIndexNamesByTable(final String databaseName, final String tableName)
       throws CatalogException {
-    Connection conn = null;
     ResultSet res = null;
     PreparedStatement pstmt = null;
-    final List<IndexDescProto> protos = new ArrayList<IndexDescProto>();
+    final List<String> indexNames = new ArrayList<String>();
 
     try {
       final int databaseId = getDatabaseId(databaseName);
       final int tableId = getTableId(databaseId, databaseName, tableName);
-      final TableIdentifierProto tableIdentifier = CatalogUtil.buildTableIdentifier(databaseName, tableName);
-
 
       String sql = GET_INDEXES_SQL + " WHERE " + COL_DATABASES_PK + "=? AND " + COL_TABLES_PK + "=?";
 
-
       if (LOG.isDebugEnabled()) {
         LOG.debug(sql);
       }
@@ -2437,10 +2467,7 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
       res = pstmt.executeQuery();
 
       while (res.next()) {
-        IndexDescProto.Builder builder = IndexDescProto.newBuilder();
-        resultToIndexDescProtoBuilder(builder, res);
-        builder.setTableIdentifier(tableIdentifier);
-        protos.add(builder.build());
+        indexNames.add(res.getString("index_name"));
       }
     } catch (SQLException se) {
       throw new TajoInternalError(se);
@@ -2448,56 +2475,73 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
       CatalogUtil.closeQuietly(pstmt, res);
     }
 
-    return protos.toArray(new IndexDescProto[protos.size()]);
+    return indexNames;
   }
-  
-  @Override
-  public List<IndexProto> getAllIndexes() throws CatalogException {
-    Connection conn = null;
-    Statement stmt = null;
-    ResultSet resultSet = null;
 
-    List<IndexProto> indexes = new ArrayList<IndexProto>();
+  @Override
+  public boolean existIndexesByTable(String databaseName, String tableName) throws CatalogException {
+    ResultSet res = null;
+    PreparedStatement pstmt = null;
+    final List<String> indexNames = new ArrayList<String>();
 
     try {
-      String sql = "SELECT " + COL_DATABASES_PK + ", " + COL_TABLES_PK + ", INDEX_NAME, " +
-        "COLUMN_NAME, DATA_TYPE, INDEX_TYPE, IS_UNIQUE, IS_CLUSTERED, IS_ASCENDING FROM " + TB_INDEXES;
+      final int databaseId = getDatabaseId(databaseName);
+      final int tableId = getTableId(databaseId, databaseName, tableName);
 
-      conn = getConnection();
-      stmt = conn.createStatement();
-      resultSet = stmt.executeQuery(sql);
-      while (resultSet.next()) {
-        IndexProto.Builder builder = IndexProto.newBuilder();
-        
-        builder.setDbId(resultSet.getInt(COL_DATABASES_PK));
-        builder.setTId(resultSet.getInt(COL_TABLES_PK));
-        builder.setIndexName(resultSet.getString("INDEX_NAME"));
-        builder.setColumnName(resultSet.getString("COLUMN_NAME"));
-        builder.setDataType(resultSet.getString("DATA_TYPE"));
-        builder.setIndexType(resultSet.getString("INDEX_TYPE"));
-        builder.setIsUnique(resultSet.getBoolean("IS_UNIQUE"));
-        builder.setIsClustered(resultSet.getBoolean("IS_CLUSTERED"));
-        builder.setIsAscending(resultSet.getBoolean("IS_ASCENDING"));
-        
-        indexes.add(builder.build());
+      String sql = GET_INDEXES_SQL + " WHERE " + COL_DATABASES_PK + "=? AND " + COL_TABLES_PK + "=?";
+
+      if (LOG.isDebugEnabled()) {
+        LOG.debug(sql);
       }
+
+      conn = getConnection();
+      pstmt = conn.prepareStatement(sql);
+      pstmt.setInt(1, databaseId);
+      pstmt.setInt(2, tableId);
+      res = pstmt.executeQuery();
+
+      return res.next();
     } catch (SQLException se) {
       throw new TajoInternalError(se);
     } finally {
-      CatalogUtil.closeQuietly(stmt, resultSet);
+      CatalogUtil.closeQuietly(pstmt, res);
     }
-    
-    return indexes;
+  }
+
+  @Override
+  public List<IndexDescProto> getAllIndexes() throws CatalogException {
+    List<IndexDescProto> indexDescProtos = TUtil.newList();
+    for (String databaseName : getAllDatabaseNames()) {
+      for (String tableName : getAllTableNames(databaseName)) {
+        for (String indexName: getAllIndexNamesByTable(databaseName, tableName)) {
+          indexDescProtos.add(getIndexByName(databaseName, indexName));
+        }
+      }
+    }
+    return indexDescProtos;
   }
 
   private void resultToIndexDescProtoBuilder(IndexDescProto.Builder builder,
                                              final ResultSet res) throws SQLException {
     builder.setIndexName(res.getString("index_name"));
-    builder.setColumn(indexResultToColumnProto(res));
     builder.setIndexMethod(getIndexMethod(res.getString("index_type").trim()));
+    builder.setIndexPath(res.getString("path"));
+    String[] columnNames, dataTypes, orders, nullOrders;
+    columnNames = res.getString("column_names").trim().split(",");
+    dataTypes = res.getString("data_types").trim().split(",");
+    orders = res.getString("orders").trim().split(",");
+    nullOrders = res.getString("null_orders").trim().split(",");
+    int columnNum = columnNames.length;
+    for (int i = 0; i < columnNum; i++) {
+      SortSpecProto.Builder colSpecBuilder = SortSpecProto.newBuilder();
+      colSpecBuilder.setColumn(ColumnProto.newBuilder().setName(columnNames[i])
+          .setDataType(CatalogUtil.newSimpleDataType(getDataType(dataTypes[i]))).build());
+      colSpecBuilder.setAscending(orders[i].equals("true"));
+      colSpecBuilder.setNullFirst(nullOrders[i].equals("true"));
+      builder.addKeySortSpecs(colSpecBuilder.build());
+    }
     builder.setIsUnique(res.getBoolean("is_unique"));
     builder.setIsClustered(res.getBoolean("is_clustered"));
-    builder.setIsAscending(res.getBoolean("is_ascending"));
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
index 4ffedcf..d8d0103 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
@@ -20,14 +20,7 @@ package org.apache.tajo.catalog.store;
 
 import org.apache.tajo.catalog.FunctionDesc;
 import org.apache.tajo.catalog.proto.CatalogProtos;
-import org.apache.tajo.catalog.proto.CatalogProtos.ColumnProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.DatabaseProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.IndexDescProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.IndexProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.TableDescriptorProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.TableOptionProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.TablePartitionProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.TableStatsProto;
+import org.apache.tajo.catalog.proto.CatalogProtos.*;
 
 import java.io.Closeable;
 
@@ -38,10 +31,6 @@ import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.KeyValueProto;
 import java.util.Collection;
 import java.util.List;
 
-import static org.apache.tajo.catalog.proto.CatalogProtos.AlterTablespaceProto;
-import static org.apache.tajo.catalog.proto.CatalogProtos.PartitionMethodProto;
-import static org.apache.tajo.catalog.proto.CatalogProtos.TablespaceProto;
-
 public interface CatalogStore extends Closeable {
   /*************************** Tablespace ******************************/
   void createTablespace(String spaceName, String spaceUri) throws CatalogException;
@@ -123,18 +112,20 @@ public interface CatalogStore extends Closeable {
   void dropIndex(String databaseName, String indexName) throws CatalogException;
   
   IndexDescProto getIndexByName(String databaseName, String indexName) throws CatalogException;
-  
-  IndexDescProto getIndexByColumn(String databaseName, String tableName, String columnName)
+
+  IndexDescProto getIndexByColumns(String databaseName, String tableName, String[] columnNames)
       throws CatalogException;
   
   boolean existIndexByName(String databaseName, String indexName) throws CatalogException;
-  
-  boolean existIndexByColumn(String databaseName, String tableName, String columnName)
+
+  boolean existIndexByColumns(String databaseName, String tableName, String[] columnNames)
       throws CatalogException;
 
-  IndexDescProto [] getIndexes(String databaseName, String tableName) throws CatalogException;
-  
-  List<IndexProto> getAllIndexes() throws CatalogException;
+  List<String> getAllIndexNamesByTable(String databaseName, String tableName) throws CatalogException;
+
+  boolean existIndexesByTable(String databaseName, String tableName) throws CatalogException;
+
+  List<IndexDescProto> getAllIndexes() throws CatalogException;
 
   /************************** FUNCTION *****************************/
 

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
index 5763f31..74b6023 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
@@ -26,15 +26,17 @@ import com.google.common.collect.Maps;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.tajo.TajoConstants;
+import org.apache.tajo.catalog.CatalogConstants;
 import org.apache.tajo.catalog.CatalogUtil;
 import org.apache.tajo.catalog.FunctionDesc;
+import org.apache.tajo.catalog.Schema;
 import org.apache.tajo.catalog.TableMeta;
 import org.apache.tajo.catalog.exception.*;
 import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.catalog.proto.CatalogProtos.ColumnProto;
 import org.apache.tajo.catalog.proto.CatalogProtos.DatabaseProto;
 import org.apache.tajo.catalog.proto.CatalogProtos.IndexDescProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.IndexProto;
+import org.apache.tajo.catalog.proto.CatalogProtos.SortSpecProto;
 import org.apache.tajo.catalog.proto.CatalogProtos.TableDescProto;
 import org.apache.tajo.catalog.proto.CatalogProtos.TableDescriptorProto;
 import org.apache.tajo.catalog.proto.CatalogProtos.TableOptionProto;
@@ -64,7 +66,6 @@ public class MemStore implements CatalogStore {
 
   public MemStore(Configuration conf) {
   }
-
   
   public void close() throws IOException {
     databases.clear();
@@ -151,6 +152,8 @@ public class MemStore implements CatalogStore {
     }
 
     databases.put(databaseName, new HashMap<String, CatalogProtos.TableDescProto>());
+    indexes.put(databaseName, new HashMap<String, IndexDescProto>());
+    indexesByColumn.put(databaseName, new HashMap<String, IndexDescProto>());
   }
 
   @Override
@@ -164,6 +167,8 @@ public class MemStore implements CatalogStore {
       throw new UndefinedDatabaseException(databaseName);
     }
     databases.remove(databaseName);
+    indexes.remove(databaseName);
+    indexesByColumn.remove(databaseName);
   }
 
   @Override
@@ -609,17 +614,23 @@ public class MemStore implements CatalogStore {
   @Override
   public void createIndex(IndexDescProto proto) throws CatalogException {
     final String databaseName = proto.getTableIdentifier().getDatabaseName();
+    final String tableName = CatalogUtil.extractSimpleName(proto.getTableIdentifier().getTableName());
 
     Map<String, IndexDescProto> index = checkAndGetDatabaseNS(indexes, databaseName);
     Map<String, IndexDescProto> indexByColumn = checkAndGetDatabaseNS(indexesByColumn, databaseName);
+    TableDescProto tableDescProto = getTable(databaseName, tableName);
 
     if (index.containsKey(proto.getIndexName())) {
       throw new DuplicateIndexException(proto.getIndexName());
     }
 
     index.put(proto.getIndexName(), proto);
-    indexByColumn.put(proto.getTableIdentifier().getTableName() + "."
-        + CatalogUtil.extractSimpleName(proto.getColumn().getName()), proto);
+    String originalTableName = proto.getTableIdentifier().getTableName();
+    String simpleTableName = CatalogUtil.extractSimpleName(originalTableName);
+    indexByColumn.put(CatalogUtil.buildFQName(proto.getTableIdentifier().getDatabaseName(),
+            simpleTableName,
+            getUnifiedNameForIndexByColumn(proto)),
+        proto);
   }
 
   /* (non-Javadoc)
@@ -628,10 +639,19 @@ public class MemStore implements CatalogStore {
   @Override
   public void dropIndex(String databaseName, String indexName) throws CatalogException {
     Map<String, IndexDescProto> index = checkAndGetDatabaseNS(indexes, databaseName);
+    Map<String, IndexDescProto> indexByColumn = checkAndGetDatabaseNS(indexesByColumn, databaseName);
     if (!index.containsKey(indexName)) {
       throw new UndefinedIndexException(indexName);
     }
+    IndexDescProto proto = index.get(indexName);
+    final String tableName = CatalogUtil.extractSimpleName(proto.getTableIdentifier().getTableName());
+    TableDescProto tableDescProto = getTable(databaseName, tableName);
     index.remove(indexName);
+    String originalTableName = proto.getTableIdentifier().getTableName();
+    String simpleTableName = CatalogUtil.extractSimpleName(originalTableName);
+    indexByColumn.remove(CatalogUtil.buildFQName(proto.getTableIdentifier().getDatabaseName(),
+        simpleTableName,
+        getUnifiedNameForIndexByColumn(proto)));
   }
 
   /* (non-Javadoc)
@@ -647,19 +667,18 @@ public class MemStore implements CatalogStore {
     return index.get(indexName);
   }
 
-  /* (non-Javadoc)
-   * @see CatalogStore#getIndexByName(java.lang.String, java.lang.String)
-   */
   @Override
-  public IndexDescProto getIndexByColumn(String databaseName, String tableName, String columnName)
-      throws CatalogException {
-
+  public IndexDescProto getIndexByColumns(String databaseName, String tableName, String[] columnNames) throws CatalogException {
     Map<String, IndexDescProto> indexByColumn = checkAndGetDatabaseNS(indexesByColumn, databaseName);
-    if (!indexByColumn.containsKey(columnName)) {
-      throw new UndefinedIndexException(CatalogUtil.buildFQName(databaseName, tableName), columnName);
+    String simpleTableName = CatalogUtil.extractSimpleName(tableName);
+    TableDescProto tableDescProto = getTable(databaseName, simpleTableName);
+    String qualifiedColumnName = CatalogUtil.buildFQName(databaseName, simpleTableName,
+        CatalogUtil.getUnifiedSimpleColumnName(new Schema(tableDescProto.getSchema()), columnNames));
+    if (!indexByColumn.containsKey(qualifiedColumnName)) {
+      throw new UndefinedIndexException(qualifiedColumnName);
     }
 
-    return indexByColumn.get(columnName);
+    return indexByColumn.get(qualifiedColumnName);
   }
 
   @Override
@@ -669,50 +688,47 @@ public class MemStore implements CatalogStore {
   }
 
   @Override
-  public boolean existIndexByColumn(String databaseName, String tableName, String columnName)
-      throws CatalogException {
+  public boolean existIndexByColumns(String databaseName, String tableName, String[] columnNames) throws CatalogException {
     Map<String, IndexDescProto> indexByColumn = checkAndGetDatabaseNS(indexesByColumn, databaseName);
-    return indexByColumn.containsKey(columnName);
+    TableDescProto tableDescProto = getTable(databaseName, tableName);
+    return indexByColumn.containsKey(
+        CatalogUtil.buildFQName(databaseName, CatalogUtil.extractSimpleName(tableName),
+            CatalogUtil.getUnifiedSimpleColumnName(new Schema(tableDescProto.getSchema()), columnNames)));
   }
 
   @Override
-  public IndexDescProto[] getIndexes(String databaseName, String tableName) throws CatalogException {
-    List<IndexDescProto> protos = new ArrayList<IndexDescProto>();
+  public List<String> getAllIndexNamesByTable(String databaseName, String tableName) throws CatalogException {
+    List<String> indexNames = new ArrayList<String>();
     Map<String, IndexDescProto> indexByColumn = checkAndGetDatabaseNS(indexesByColumn, databaseName);
+    String simpleTableName = CatalogUtil.extractSimpleName(tableName);
     for (IndexDescProto proto : indexByColumn.values()) {
-      if (proto.getTableIdentifier().getTableName().equals(tableName)) {
-        protos.add(proto);
+      if (proto.getTableIdentifier().getTableName().equals(simpleTableName)) {
+        indexNames.add(proto.getIndexName());
       }
     }
 
-    return protos.toArray(new IndexDescProto[protos.size()]);
+    return indexNames;
   }
-  
+
   @Override
-  public List<IndexProto> getAllIndexes() throws CatalogException {
-    List<IndexProto> indexList = new ArrayList<CatalogProtos.IndexProto>();
-    Set<String> databases = indexes.keySet();
-    
-    for (String databaseName: databases) {
-      Map<String, IndexDescProto> indexMap = indexes.get(databaseName);
-      
-      for (Map.Entry<String, IndexDescProto> entry: indexMap.entrySet()) {
-        IndexDescProto indexDesc = entry.getValue();
-        IndexProto.Builder builder = IndexProto.newBuilder();
-        
-        builder.setColumnName(indexDesc.getColumn().getName());
-        builder.setDataType(indexDesc.getColumn().getDataType().getType().toString());
-        builder.setIndexName(entry.getKey());
-        builder.setIndexType(indexDesc.getIndexMethod().toString());
-        builder.setIsAscending(indexDesc.hasIsAscending() && indexDesc.getIsAscending());
-        builder.setIsClustered(indexDesc.hasIsClustered() && indexDesc.getIsClustered());
-        builder.setIsUnique(indexDesc.hasIsUnique() && indexDesc.getIsUnique());
-        
-        indexList.add(builder.build());
+  public boolean existIndexesByTable(String databaseName, String tableName) throws CatalogException {
+    Map<String, IndexDescProto> indexByColumn = checkAndGetDatabaseNS(indexesByColumn, databaseName);
+    String simpleTableName = CatalogUtil.extractSimpleName(tableName);
+    for (IndexDescProto proto : indexByColumn.values()) {
+      if (proto.getTableIdentifier().getTableName().equals(simpleTableName)) {
+        return true;
       }
     }
-    
-    return indexList;
+    return false;
+  }
+
+  @Override
+  public List<IndexDescProto> getAllIndexes() throws CatalogException {
+    List<IndexDescProto> indexDescProtos = TUtil.newList();
+    for (Map<String,IndexDescProto> indexMap : indexes.values()) {
+      indexDescProtos.addAll(indexMap.values());
+    }
+    return indexDescProtos;
   }
 
   @Override
@@ -736,4 +752,13 @@ public class MemStore implements CatalogStore {
     return null;
   }
 
+  public static String getUnifiedNameForIndexByColumn(IndexDescProto proto) {
+    StringBuilder sb = new StringBuilder();
+    for (SortSpecProto columnSpec : proto.getKeySortSpecsList()) {
+      String[] identifiers = columnSpec.getColumn().getName().split(CatalogConstants.IDENTIFIER_DELIMITER_REGEXP);
+      sb.append(identifiers[identifiers.length-1]).append("_");
+    }
+    sb.deleteCharAt(sb.length()-1);
+    return sb.toString();
+  }
 }

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/derby/derby.xml
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/derby/derby.xml b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/derby/derby.xml
index e0bd469..7ed9118 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/derby/derby.xml
+++ b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/derby/derby.xml
@@ -19,12 +19,14 @@
 <tns:store xmlns:tns="http://tajo.apache.org/catalogstore" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.xsd ">
   <!--
       Catalog base version history
+      * 6 - 2015-07-24: Merge the index branch into the master branch (TAJO-1300)
+      * 5 - 2015-06-15: Implement TablespaceManager to load Tablespaces (TAJO-1616)
       * 4 - 2015-03-27: Partition Schema (TAJO-1284)
       * 3 - 2015-03-12: Nested Schema (TAJO-1329)
       * 2 - 2014-06-09: First versioning
       * 1-  Before 2013-03-20
     -->
-	<tns:base version="5">
+	<tns:base version="6">
 		<tns:objects>
 			<tns:Object order="0" type="table" name="META">
 				<tns:sql><![CDATA[CREATE TABLE META (VERSION INT NOT NULL)]]></tns:sql>
@@ -105,16 +107,22 @@
 			<tns:Object order="11" type="table" name="INDEXES">
 				<tns:sql><![CDATA[
 				CREATE TABLE INDEXES (
-  				DB_ID INT NOT NULL REFERENCES DATABASES_ (DB_ID) ON DELETE CASCADE,
-  				TID INT NOT NULL REFERENCES TABLES (TID) ON DELETE CASCADE,
-  				INDEX_NAME VARCHAR(128) NOT NULL,
-  				COLUMN_NAME VARCHAR(128) NOT NULL,
-  				DATA_TYPE VARCHAR(128) NOT NULL,
-  				INDEX_TYPE CHAR(32) NOT NULL,
-  				IS_UNIQUE BOOLEAN NOT NULL,
-  				IS_CLUSTERED BOOLEAN NOT NULL,
-  				IS_ASCENDING BOOLEAN NOT NULL,
-  				CONSTRAINT C_INDEXES_PK PRIMARY KEY (DB_ID, INDEX_NAME)
+					INDEX_ID INT NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1),
+					DB_ID INT NOT NULL REFERENCES DATABASES_ (DB_ID) ON DELETE CASCADE,
+					TID INT NOT NULL REFERENCES TABLES (TID) ON DELETE CASCADE,
+					INDEX_NAME VARCHAR(128) NOT NULL,
+					INDEX_TYPE CHAR(32) NOT NULL,
+					PATH VARCHAR(4096),
+					COLUMN_NAMES VARCHAR(128) NOT NULL, -- array of column names
+					DATA_TYPES VARCHAR(128) NOT NULL, -- array of column types
+					ORDERS VARCHAR(128) NOT NULL, -- array of column orders
+					NULL_ORDERS VARCHAR(128) NOT NULL, -- array of null orderings
+					IS_UNIQUE BOOLEAN NOT NULL,
+					IS_CLUSTERED BOOLEAN NOT NULL,
+					FOREIGN KEY (DB_ID) REFERENCES DATABASES_ (DB_ID) ON DELETE CASCADE,
+					FOREIGN KEY (TID) REFERENCES TABLES (TID) ON DELETE CASCADE,
+					CONSTRAINT INDEXES_PK PRIMARY KEY (INDEX_ID),
+					CONSTRAINT C_INDEXES_UNIQ UNIQUE (DB_ID, INDEX_NAME)
 				)]]>
 				</tns:sql>
 			</tns:Object>
@@ -122,7 +130,7 @@
 				<tns:sql><![CDATA[CREATE UNIQUE INDEX idx_indexes_pk ON INDEXES (DB_ID,index_name)]]></tns:sql>
 			</tns:Object>
 			<tns:Object order="13" type="index" name="IDX_INDEXES_COLUMNS" dependsOn="INDEXES">
-				<tns:sql><![CDATA[CREATE INDEX idx_indexes_columns ON INDEXES (DB_ID,column_name)]]></tns:sql>
+				<tns:sql><![CDATA[CREATE INDEX idx_col_names ON INDEXES (DB_ID,TID,column_names)]]></tns:sql>
 			</tns:Object>
 			<tns:Object order="14" type="table" name="STATS">
 				<tns:sql><![CDATA[

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mariadb/mariadb.xml
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mariadb/mariadb.xml b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mariadb/mariadb.xml
index 7485da1..8750d2b 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mariadb/mariadb.xml
+++ b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mariadb/mariadb.xml
@@ -19,13 +19,14 @@
 <tns:store xmlns:tns="http://tajo.apache.org/catalogstore" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.xsd ">
   <!--
       Catalog base version history
+      * 6 - 2015-07-24: Merge the index branch into the master branch (TAJO-1300)
       * 5 - 2015-06-15: Implement TablespaceManager to load Tablespaces (TAJO-1616)
       * 4 - 2015-03-27: Partition Schema (TAJO-1284)
       * 3 - 2015-03-12: Nested Schema (TAJO-1329)
       * 2 - 2014-06-09: First versioning
       * 1-  Before 2013-03-20
     -->
-  <tns:base version="5">
+  <tns:base version="6">
     <tns:objects>
       <tns:Object order="0" type="table" name="META">
         <tns:sql><![CDATA[CREATE TABLE META (VERSION INT NOT NULL)]]></tns:sql>
@@ -95,20 +96,21 @@
       <tns:Object order="6" type="table" name="INDEXES">
         <tns:sql><![CDATA[
         CREATE TABLE INDEXES (
+          INDEX_ID INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
           DB_ID INT NOT NULL,
           TID INT NOT NULL,
-          INDEX_NAME VARCHAR(128) BINARY NOT NULL,
-          COLUMN_NAME VARCHAR(128) BINARY NOT NULL,
-          DATA_TYPE VARCHAR(128) NOT NULL,
-          INDEX_TYPE CHAR(32) NOT NULL,
+          INDEX_NAME VARCHAR(128) NOT NULL,
+          INDEX_TYPE CHAR(32) BINARY NOT NULL,
+          PATH VARCHAR(4096) NOT NULL,
+          COLUMN_NAMES VARCHAR(256) NOT NULL, -- array of column names
+          DATA_TYPES VARCHAR(128) NOT NULL, -- array of column types
+          ORDERS VARCHAR(128) NOT NULL, -- array of column orders
+          NULL_ORDERS VARCHAR(128) NOT NULL, -- array of null orderings
           IS_UNIQUE BOOLEAN NOT NULL,
           IS_CLUSTERED BOOLEAN NOT NULL,
-          IS_ASCENDING BOOLEAN NOT NULL,
-          PRIMARY KEY (DB_ID, INDEX_NAME),
           FOREIGN KEY (DB_ID) REFERENCES DATABASES_ (DB_ID) ON DELETE CASCADE,
           FOREIGN KEY (TID) REFERENCES TABLES (TID) ON DELETE CASCADE,
-          UNIQUE INDEX IDX_DB_ID_NAME (DB_ID, INDEX_NAME),
-          INDEX IDX_TID_COLUMN_NAME (TID, COLUMN_NAME)
+          UNIQUE INDEX IDX_DB_ID_NAME (DB_ID, INDEX_NAME)
         )]]>
         </tns:sql>
       </tns:Object>

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mysql/mysql.xml
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mysql/mysql.xml b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mysql/mysql.xml
index 2bde04f..763d0f7 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mysql/mysql.xml
+++ b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mysql/mysql.xml
@@ -19,13 +19,14 @@
 <tns:store xmlns:tns="http://tajo.apache.org/catalogstore" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.xsd ">
   <!--
     Catalog base version history
+    * 6 - 2015-07-24: Merge the index branch into the master branch (TAJO-1300)
     * 5 - 2015-06-15: Implement TablespaceManager to load Tablespaces (TAJO-1616)
     * 4 - 2015-03-27: Partition Schema (TAJO-1284)
     * 3 - 2015-03-12: Nested Schema (TAJO-1329)
     * 2 - 2014-06-09: First versioning
     * 1-  Before 2013-03-20
   -->
-  <tns:base version="5">
+  <tns:base version="6">
     <tns:objects>
       <tns:Object order="0" type="table" name="META">
         <tns:sql><![CDATA[CREATE TABLE META (VERSION INT NOT NULL)]]></tns:sql>
@@ -96,20 +97,21 @@
       <tns:Object order="6" type="table" name="INDEXES">
         <tns:sql><![CDATA[
         CREATE TABLE INDEXES (
+          INDEX_ID INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
           DB_ID INT NOT NULL,
           TID INT NOT NULL,
-          INDEX_NAME VARCHAR(128) BINARY NOT NULL,
-          COLUMN_NAME VARCHAR(128) BINARY NOT NULL,
-          DATA_TYPE VARCHAR(128) NOT NULL,
-          INDEX_TYPE CHAR(32) NOT NULL,
+          INDEX_NAME VARCHAR(128) NOT NULL,
+          INDEX_TYPE CHAR(32) BINARY NOT NULL,
+          PATH VARCHAR(4096) NOT NULL,
+          COLUMN_NAMES VARCHAR(256) NOT NULL, -- array of column names
+          DATA_TYPES VARCHAR(128) NOT NULL, -- array of column types
+          ORDERS VARCHAR(128) NOT NULL, -- array of column orders
+          NULL_ORDERS VARCHAR(128) NOT NULL, -- array of null orderings
           IS_UNIQUE BOOLEAN NOT NULL,
           IS_CLUSTERED BOOLEAN NOT NULL,
-          IS_ASCENDING BOOLEAN NOT NULL,
-          PRIMARY KEY (DB_ID, INDEX_NAME),
           FOREIGN KEY (DB_ID) REFERENCES DATABASES_ (DB_ID) ON DELETE CASCADE,
           FOREIGN KEY (TID) REFERENCES TABLES (TID) ON DELETE CASCADE,
-          UNIQUE INDEX IDX_DB_ID_NAME (DB_ID, INDEX_NAME),
-          INDEX IDX_TID_COLUMN_NAME (TID, COLUMN_NAME)
+          UNIQUE INDEX IDX_DB_ID_NAME (DB_ID, INDEX_NAME)
         )]]>
         </tns:sql>
       </tns:Object>

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/oracle/oracle.xml
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/oracle/oracle.xml b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/oracle/oracle.xml
index 2778e0c..3d8ef30 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/oracle/oracle.xml
+++ b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/oracle/oracle.xml
@@ -19,13 +19,14 @@
 <tns:store xmlns:tns="http://tajo.apache.org/catalogstore" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.xsd ">
   <!--
       Catalog base version history
+      * 6 - 2015-07-24: Merge the index branch into the master branch (TAJO-1300)
       * 5 - 2015-06-15: Implement TablespaceManager to load Tablespaces (TAJO-1616)
       * 4 - 2015-03-27: Partition Schema (TAJO-1284)
       * 3 - 2015-03-12: Nested Schema (TAJO-1329)
       * 2 - 2014-06-09: First versioning
       * 1-  Before 2013-03-20
     -->
-  <tns:base version="5">
+  <tns:base version="6">
     <tns:objects>
   		<tns:Object order="0" type="table" name="meta">
   			<tns:sql><![CDATA[
@@ -147,18 +148,21 @@
   		<tns:Object order="14" type="table" name="INDEXES">
   			<tns:sql><![CDATA[
   			CREATE TABLE INDEXES (
-					DB_ID INT NOT NULL,
-					TID INT NOT NULL,
-					INDEX_NAME VARCHAR2(128) NOT NULL,
-					COLUMN_NAME VARCHAR2(128) NOT NULL,
-					DATA_TYPE VARCHAR2(128) NOT NULL,
-					INDEX_TYPE CHAR(32) NOT NULL,
-					IS_UNIQUE CHAR NOT NULL,
-					IS_CLUSTERED CHAR NOT NULL,
-					IS_ASCENDING CHAR NOT NULL,
-					CONSTRAINT INDEXES_PKEY PRIMARY KEY (DB_ID, INDEX_NAME),
-					FOREIGN KEY (DB_ID) REFERENCES DATABASES_ (DB_ID) ON DELETE CASCADE,
-					FOREIGN KEY (TID) REFERENCES TABLES (TID) ON DELETE CASCADE
+  				INDEX_ID NUMBER(10) NOT NULL PRIMARY KEY,
+  				DB_ID INT NOT NULL,
+  				TID INT NOT NULL,
+  				INDEX_NAME VARCHAR2(128) NOT NULL,
+  				INDEX_TYPE CHAR(32) NOT NULL,
+  				PATH VARCHAR(4096) NOT NULL,
+  				COLUMN_NAMES VARCHAR(256) NOT NULL, -- array of column names
+  				DATA_TYPES VARCHAR(128) NOT NULL, -- array of column types
+  				ORDERS VARCHAR(128) NOT NULL, -- array of column orders
+  				NULL_ORDERS VARCHAR(128) NOT NULL, -- array of null orderings
+  				IS_UNIQUE CHAR NOT NULL,
+  				IS_CLUSTERED CHAR NOT NULL,
+  				IS_ASCENDING CHAR NOT NULL,
+  				FOREIGN KEY (DB_ID) REFERENCES DATABASES_ (DB_ID) ON DELETE CASCADE,
+  				FOREIGN KEY (TID) REFERENCES TABLES (TID) ON DELETE CASCADE
 				)]]>
 				</tns:sql>
   		</tns:Object>

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/postgresql/postgresql.xml
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/postgresql/postgresql.xml b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/postgresql/postgresql.xml
index 0051242..0dd72b9 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/postgresql/postgresql.xml
+++ b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/postgresql/postgresql.xml
@@ -21,6 +21,7 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.xsd ">
   <!--
       Catalog base version history
+      * 6 - 2015-07-24: Merge the index branch into the master branch (TAJO-1300)
       * 5 - 2015-06-15: Implement TablespaceManager to load Tablespaces (TAJO-1616)
       * 5 - 2015-06-15: Implement TablespaceManager to load Tablespaces (TAJO-1616)
       * 4 - 2015-03-27: Partition Schema (TAJO-1284)
@@ -28,7 +29,7 @@ xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.
       * 2 - 2014-06-09: First versioning
       * 1-  Before 2013-03-20
     -->
-	<tns:base version="5">
+	<tns:base version="6">
 		<tns:objects>
 			<tns:Object name="META" type="table" order="0">
 				<tns:sql><![CDATA[CREATE TABLE META (VERSION INT NOT NULL)]]></tns:sql>
@@ -107,18 +108,21 @@ xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.
 			<tns:Object name="INDEXES" type="table" order="10">
 				<tns:sql><![CDATA[
 				CREATE TABLE INDEXES (
-  				DB_ID INT NOT NULL,
-  				TID INT NOT NULL,
-  				INDEX_NAME VARCHAR(128) NOT NULL,
-  				COLUMN_NAME VARCHAR(128) NOT NULL,
-  				DATA_TYPE VARCHAR(128) NOT NULL,
-  				INDEX_TYPE CHAR(32) NOT NULL,
-  				IS_UNIQUE BOOLEAN NOT NULL,
-  				IS_CLUSTERED BOOLEAN NOT NULL,
-  				IS_ASCENDING BOOLEAN NOT NULL,
-  				CONSTRAINT INDEXES_PKEY PRIMARY KEY (DB_ID, INDEX_NAME),
-  				FOREIGN KEY (DB_ID) REFERENCES DATABASES_ (DB_ID) ON DELETE CASCADE,
-  				FOREIGN KEY (TID) REFERENCES TABLES (TID) ON DELETE CASCADE
+  					INDEX_ID SERIAL NOT NULL PRIMARY KEY,
+  					DB_ID INT NOT NULL,
+  					TID INT NOT NULL,
+  					INDEX_NAME VARCHAR(128) NOT NULL,
+  					INDEX_TYPE CHAR(32) NOT NULL,
+  					PATH VARCHAR(4096) NOT NULL,
+  					COLUMN_NAMES VARCHAR(256) NOT NULL, -- array of column names
+  					DATA_TYPES VARCHAR(128) NOT NULL, -- array of column types
+  					ORDERS VARCHAR(128) NOT NULL, -- array of column orders
+  					NULL_ORDERS VARCHAR(128) NOT NULL, -- array of null orderings
+  					IS_UNIQUE BOOLEAN NOT NULL,
+  					IS_CLUSTERED BOOLEAN NOT NULL,
+  					CONSTRAINT INDEXES_PKEY PRIMARY KEY (INDEX_ID),
+  					FOREIGN KEY (DB_ID) REFERENCES DATABASES_ (DB_ID) ON DELETE CASCADE,
+  					FOREIGN KEY (TID) REFERENCES TABLES (TID) ON DELETE CASCADE
 				)]]>
 				</tns:sql>
 			</tns:Object>
@@ -126,7 +130,7 @@ xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.
 				<tns:sql><![CDATA[CREATE UNIQUE INDEX INDEXES_IDX_DB_ID_NAME on INDEXES (DB_ID, INDEX_NAME)]]></tns:sql>
 			</tns:Object>
 			<tns:Object name="INDEXES_IDX_TID_COLUMN_NAME" type="index" order="12" dependsOn="INDEXES">
-				<tns:sql><![CDATA[CREATE INDEX INDEXES_IDX_TID_COLUMN_NAME on INDEXES (TID, COLUMN_NAME)]]></tns:sql>
+				<tns:sql><![CDATA[CREATE INDEX INDEXES_IDX_TID_COLUMN_NAME on INDEXES (DB_ID, TID, COLUMN_NAMES)]]></tns:sql>
 			</tns:Object>
 			<tns:Object name="STATS" type="table" order="13">
 				<tns:sql><![CDATA[

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java b/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
index 40b731e..1206bfa 100644
--- a/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
+++ b/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
@@ -19,23 +19,18 @@
 package org.apache.tajo.catalog;
 
 import com.google.common.collect.Sets;
-
 import org.apache.hadoop.fs.Path;
 import org.apache.tajo.TajoConstants;
 import org.apache.tajo.catalog.dictionary.InfoSchemaMetadataDictionary;
 import org.apache.tajo.catalog.exception.CatalogException;
 import org.apache.tajo.catalog.exception.UndefinedFunctionException;
 import org.apache.tajo.catalog.partition.PartitionDesc;
-import org.apache.tajo.catalog.store.PostgreSQLStore;
 import org.apache.tajo.catalog.partition.PartitionMethodDesc;
 import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.catalog.proto.CatalogProtos.FunctionType;
 import org.apache.tajo.catalog.proto.CatalogProtos.IndexMethod;
 import org.apache.tajo.catalog.proto.CatalogProtos.PartitionKeyProto;
-import org.apache.tajo.catalog.store.DerbyStore;
-import org.apache.tajo.catalog.store.MySQLStore;
-import org.apache.tajo.catalog.store.MariaDBStore;
-import org.apache.tajo.catalog.store.OracleStore;
+import org.apache.tajo.catalog.store.*;
 import org.apache.tajo.common.TajoDataTypes;
 import org.apache.tajo.common.TajoDataTypes.Type;
 import org.apache.tajo.conf.TajoConf;
@@ -49,6 +44,8 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 
 import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.*;
 
 import static org.apache.tajo.TajoConstants.DEFAULT_DATABASE_NAME;
@@ -537,36 +534,43 @@ public class TestCatalog {
   static IndexDesc desc1;
   static IndexDesc desc2;
   static IndexDesc desc3;
-
-  static {
-    desc1 = new IndexDesc(
-        "idx_test", DEFAULT_DATABASE_NAME, "indexed", new Column("id", Type.INT4),
-        IndexMethod.TWO_LEVEL_BIN_TREE, true, true, true);
-
-    desc2 = new IndexDesc(
-        "idx_test2", DEFAULT_DATABASE_NAME, "indexed", new Column("score", Type.FLOAT8),
-        IndexMethod.TWO_LEVEL_BIN_TREE, false, false, false);
-
-    desc3 = new IndexDesc(
-        "idx_test", DEFAULT_DATABASE_NAME, "indexed", new Column("id", Type.INT4),
-        IndexMethod.TWO_LEVEL_BIN_TREE, true, true, true);
-  }
+  static Schema relationSchema;
 
   public static TableDesc prepareTable() throws IOException {
-    Schema schema = new Schema();
-    schema.addColumn("indexed.id", Type.INT4)
-        .addColumn("indexed.name", Type.TEXT)
-        .addColumn("indexed.age", Type.INT4)
-        .addColumn("indexed.score", Type.FLOAT8);
+    relationSchema = new Schema();
+    relationSchema.addColumn(DEFAULT_DATABASE_NAME + ".indexed.id", Type.INT4)
+        .addColumn(DEFAULT_DATABASE_NAME + ".indexed.name", Type.TEXT)
+        .addColumn(DEFAULT_DATABASE_NAME + ".indexed.age", Type.INT4)
+        .addColumn(DEFAULT_DATABASE_NAME + ".indexed.score", Type.FLOAT8);
 
     String tableName = "indexed";
 
     TableMeta meta = CatalogUtil.newTableMeta("TEXT");
     return new TableDesc(
-        CatalogUtil.buildFQName(TajoConstants.DEFAULT_DATABASE_NAME, tableName), schema, meta,
+        CatalogUtil.buildFQName(TajoConstants.DEFAULT_DATABASE_NAME, tableName), relationSchema, meta,
         new Path(CommonTestingUtil.getTestDir(), "indexed").toUri());
   }
 
+  public static void prepareIndexDescs() throws IOException, URISyntaxException {
+    SortSpec[] colSpecs1 = new SortSpec[1];
+    colSpecs1[0] = new SortSpec(new Column("default.indexed.id", Type.INT4), true, true);
+    desc1 = new IndexDesc(DEFAULT_DATABASE_NAME, "indexed",
+        "idx_test", new URI("idx_test"), colSpecs1,
+        IndexMethod.TWO_LEVEL_BIN_TREE, true, true, relationSchema);
+
+    SortSpec[] colSpecs2 = new SortSpec[1];
+    colSpecs2[0] = new SortSpec(new Column("default.indexed.score", Type.FLOAT8), false, false);
+    desc2 = new IndexDesc(DEFAULT_DATABASE_NAME, "indexed",
+        "idx_test2", new URI("idx_test2"), colSpecs2,
+        IndexMethod.TWO_LEVEL_BIN_TREE, false, false, relationSchema);
+
+    SortSpec[] colSpecs3 = new SortSpec[1];
+    colSpecs3[0] = new SortSpec(new Column("default.indexed.id", Type.INT4), true, false);
+    desc3 = new IndexDesc(DEFAULT_DATABASE_NAME, "indexed",
+        "idx_test", new URI("idx_test"), colSpecs3,
+        IndexMethod.TWO_LEVEL_BIN_TREE, true, true, relationSchema);
+  }
+
   @Test
   public void testCreateSameTables() throws IOException {
     assertTrue(catalog.createDatabase("tmpdb3", TajoConstants.DEFAULT_TABLESPACE_NAME));
@@ -602,25 +606,34 @@ public class TestCatalog {
 	@Test
 	public void testAddAndDelIndex() throws Exception {
 	  TableDesc desc = prepareTable();
+    prepareIndexDescs();
 	  assertTrue(catalog.createTable(desc));
 	  
-	  assertFalse(catalog.existIndexByName("db1", desc1.getIndexName()));
-	  assertFalse(catalog.existIndexByColumn(DEFAULT_DATABASE_NAME, "indexed", "id"));
+	  assertFalse(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc1.getName()));
+	  assertFalse(catalog.existIndexByColumnNames(DEFAULT_DATABASE_NAME, "indexed", new String[]{"id"}));
 	  catalog.createIndex(desc1);
-	  assertTrue(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc1.getIndexName()));
-	  assertTrue(catalog.existIndexByColumn(DEFAULT_DATABASE_NAME, "indexed", "id"));
+	  assertTrue(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc1.getName()));
+	  assertTrue(catalog.existIndexByColumnNames(DEFAULT_DATABASE_NAME, "indexed", new String[]{"id"}));
 
 
-	  assertFalse(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc2.getIndexName()));
-	  assertFalse(catalog.existIndexByColumn(DEFAULT_DATABASE_NAME, "indexed", "score"));
+	  assertFalse(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc2.getName()));
+	  assertFalse(catalog.existIndexByColumnNames(DEFAULT_DATABASE_NAME, "indexed", new String[]{"score"}));
 	  catalog.createIndex(desc2);
-	  assertTrue(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc2.getIndexName()));
-	  assertTrue(catalog.existIndexByColumn(DEFAULT_DATABASE_NAME, "indexed", "score"));
+	  assertTrue(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc2.getName()));
+	  assertTrue(catalog.existIndexByColumnNames(DEFAULT_DATABASE_NAME, "indexed", new String[]{"score"}));
+
+    Set<IndexDesc> indexDescs = TUtil.newHashSet();
+    indexDescs.add(desc1);
+    indexDescs.add(desc2);
+    indexDescs.add(desc3);
+    for (IndexDesc index : catalog.getAllIndexesByTable(DEFAULT_DATABASE_NAME, "indexed")) {
+      assertTrue(indexDescs.contains(index));
+    }
 	  
-	  catalog.dropIndex(DEFAULT_DATABASE_NAME, desc1.getIndexName());
-	  assertFalse(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc1.getIndexName()));
-	  catalog.dropIndex(DEFAULT_DATABASE_NAME, desc2.getIndexName());
-	  assertFalse(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc2.getIndexName()));
+	  catalog.dropIndex(DEFAULT_DATABASE_NAME, desc1.getName());
+	  assertFalse(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc1.getName()));
+	  catalog.dropIndex(DEFAULT_DATABASE_NAME, desc2.getName());
+	  assertFalse(catalog.existIndexByName(DEFAULT_DATABASE_NAME, desc2.getName()));
 	  
 	  catalog.dropTable(desc.getName());
     assertFalse(catalog.existsTable(desc.getName()));

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-cli/src/main/java/org/apache/tajo/cli/tools/TajoDump.java
----------------------------------------------------------------------
diff --git a/tajo-cli/src/main/java/org/apache/tajo/cli/tools/TajoDump.java b/tajo-cli/src/main/java/org/apache/tajo/cli/tools/TajoDump.java
index b48a9ec..12515a8 100644
--- a/tajo-cli/src/main/java/org/apache/tajo/cli/tools/TajoDump.java
+++ b/tajo-cli/src/main/java/org/apache/tajo/cli/tools/TajoDump.java
@@ -21,10 +21,8 @@ package org.apache.tajo.cli.tools;
 import com.google.protobuf.ServiceException;
 import org.apache.commons.cli.*;
 import org.apache.tajo.auth.UserRoleInfo;
-import org.apache.tajo.catalog.CatalogConstants;
-import org.apache.tajo.catalog.CatalogUtil;
-import org.apache.tajo.catalog.DDLBuilder;
-import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.catalog.*;
+import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.client.TajoClient;
 import org.apache.tajo.client.TajoClientImpl;
 import org.apache.tajo.conf.TajoConf;
@@ -186,6 +184,14 @@ public class TajoDump {
         } else {
           writer.write(DDLBuilder.buildDDLForBaseTable(table));
         }
+
+        if (client.hasIndexes(tableName)) {
+          List<CatalogProtos.IndexDescProto> indexeProtos = client.getIndexes(tableName);
+          for (CatalogProtos.IndexDescProto eachIndexProto : indexeProtos) {
+            writer.write("\n\n");
+            writer.write(DDLBuilder.buildDDLForIndex(new IndexDesc(eachIndexProto)));
+          }
+        }
         writer.write("\n\n");
       } catch (Exception e) {
         // dump for each table can throw any exception. We need to skip the exception case.

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-cli/src/main/java/org/apache/tajo/cli/tsql/commands/DescTableCommand.java
----------------------------------------------------------------------
diff --git a/tajo-cli/src/main/java/org/apache/tajo/cli/tsql/commands/DescTableCommand.java b/tajo-cli/src/main/java/org/apache/tajo/cli/tsql/commands/DescTableCommand.java
index 6b2905a..4b7006d 100644
--- a/tajo-cli/src/main/java/org/apache/tajo/cli/tsql/commands/DescTableCommand.java
+++ b/tajo-cli/src/main/java/org/apache/tajo/cli/tsql/commands/DescTableCommand.java
@@ -21,9 +21,12 @@ package org.apache.tajo.cli.tsql.commands;
 import org.apache.commons.lang.CharUtils;
 import org.apache.commons.lang.StringEscapeUtils;
 import org.apache.tajo.TajoConstants;
+import org.apache.tajo.catalog.CatalogUtil;
 import org.apache.tajo.catalog.Column;
 import org.apache.tajo.catalog.TableDesc;
 import org.apache.tajo.catalog.partition.PartitionMethodDesc;
+import org.apache.tajo.catalog.proto.CatalogProtos.IndexDescProto;
+import org.apache.tajo.catalog.proto.CatalogProtos.SortSpecProto;
 import org.apache.tajo.cli.tsql.TajoCli;
 import org.apache.tajo.util.FileUtil;
 import org.apache.tajo.util.StringUtils;
@@ -51,6 +54,22 @@ public class DescTableCommand extends TajoShellCommand {
         context.getOutput().println("Did not find any relation named \"" + tableName + "\"");
       } else {
         context.getOutput().println(toFormattedString(desc));
+        // If there exists any indexes for the table, print index information
+        if (client.hasIndexes(tableName)) {
+          StringBuilder sb = new StringBuilder();
+          sb.append("Indexes:\n");
+          for (IndexDescProto index : client.getIndexes(tableName)) {
+            sb.append("\"").append(index.getIndexName()).append("\" ");
+            sb.append(index.getIndexMethod()).append(" (");
+            for (SortSpecProto key : index.getKeySortSpecsList()) {
+              sb.append(CatalogUtil.extractSimpleName(key.getColumn().getName()));
+              sb.append(key.getAscending() ? " ASC" : " DESC");
+              sb.append(key.getNullFirst() ? " NULLS FIRST, " : " NULLS LAST, ");
+            }
+            sb.delete(sb.length()-2, sb.length()-1).append(")\n");
+          }
+          context.getOutput().println(sb.toString());
+        }
       }
     } else if (cmd.length == 1) {
       List<String> tableList = client.getTableList(null);

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClient.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClient.java b/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClient.java
index c020ef5..5e923e0 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClient.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClient.java
@@ -24,6 +24,7 @@ import org.apache.tajo.catalog.TableDesc;
 import org.apache.tajo.catalog.TableMeta;
 import org.apache.tajo.catalog.partition.PartitionMethodDesc;
 import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.catalog.proto.CatalogProtos.IndexDescProto;
 
 import java.io.Closeable;
 import java.net.URI;
@@ -135,4 +136,18 @@ public interface CatalogAdminClient extends Closeable {
   TableDesc getTableDesc(final String tableName) throws SQLException;
 
   List<CatalogProtos.FunctionDescProto> getFunctions(final String functionName) throws SQLException;
+
+  IndexDescProto getIndex(final String indexName) throws SQLException;
+
+  boolean existIndex(final String indexName) throws SQLException;
+
+  List<IndexDescProto> getIndexes(final String tableName) throws SQLException;
+
+  boolean hasIndexes(final String tableName) throws SQLException;
+
+  IndexDescProto getIndex(final String tableName, final String[] columnNames) throws SQLException;
+
+  boolean existIndex(final String tableName, final String[] columnName) throws SQLException;
+
+  boolean dropIndex(final String indexName) throws SQLException;
 }

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClientImpl.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClientImpl.java b/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClientImpl.java
index e73a032..8745d19 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClientImpl.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClientImpl.java
@@ -26,8 +26,10 @@ import org.apache.tajo.catalog.TableDesc;
 import org.apache.tajo.catalog.TableMeta;
 import org.apache.tajo.catalog.partition.PartitionMethodDesc;
 import org.apache.tajo.catalog.proto.CatalogProtos;
-import org.apache.tajo.catalog.proto.CatalogProtos.FunctionDescProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.TableResponse;
+import org.apache.tajo.catalog.proto.CatalogProtos.*;
+import org.apache.tajo.error.Errors.ResultCode;
+import org.apache.tajo.ipc.ClientProtos;
+import org.apache.tajo.ipc.ClientProtos.*;
 import org.apache.tajo.exception.SQLExceptionUtil;
 import org.apache.tajo.ipc.ClientProtos;
 import org.apache.tajo.ipc.ClientProtos.DropTableRequest;
@@ -220,6 +222,110 @@ public class CatalogAdminClientImpl implements CatalogAdminClient {
   }
 
   @Override
+  public IndexDescProto getIndex(final String indexName) throws SQLException {
+    final BlockingInterface stub = conn.getTMStub();
+
+    IndexResponse res;
+    try {
+      res = stub.getIndexWithName(null, conn.getSessionedString(indexName));
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+
+    throwIfError(res.getState());
+    return res.getIndexDesc();
+  }
+
+  @Override
+  public boolean existIndex(final String indexName) throws SQLException {
+    final BlockingInterface stub = conn.getTMStub();
+
+    try {
+      return isSuccess(stub.existIndexWithName(null, conn.getSessionedString(indexName)));
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public List<IndexDescProto> getIndexes(final String tableName) throws SQLException {
+    final BlockingInterface stub = conn.getTMStub();
+
+    IndexListResponse response;
+    try {
+      response = stub.getIndexesForTable(null,
+          conn.getSessionedString(tableName));
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+
+    throwIfError(response.getState());
+    return response.getIndexDescList();
+  }
+
+  @Override
+  public boolean hasIndexes(final String tableName) throws SQLException {
+    final BlockingInterface stub = conn.getTMStub();
+
+    try {
+      return isSuccess(stub.existIndexesForTable(null, conn.getSessionedString(tableName)));
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public IndexDescProto getIndex(final String tableName, final String[] columnNames) throws SQLException {
+    final BlockingInterface stub = conn.getTMStub();
+
+    GetIndexWithColumnsRequest.Builder builder = GetIndexWithColumnsRequest.newBuilder();
+    builder.setSessionId(conn.sessionId);
+    builder.setTableName(tableName);
+    for (String eachColumnName : columnNames) {
+      builder.addColumnNames(eachColumnName);
+    }
+
+    IndexResponse response;
+    try {
+      response = stub.getIndexWithColumns(null, builder.build());
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+
+    throwIfError(response.getState());
+    return response.getIndexDesc();
+  }
+
+  @Override
+  public boolean existIndex(final String tableName, final String[] columnName) throws SQLException {
+    final BlockingInterface stub = conn.getTMStub();
+
+    GetIndexWithColumnsRequest.Builder builder = GetIndexWithColumnsRequest.newBuilder();
+    builder.setSessionId(conn.sessionId);
+    builder.setTableName(tableName);
+    for (String eachColumnName : columnName) {
+      builder.addColumnNames(eachColumnName);
+    }
+
+    try {
+      return isSuccess(stub.existIndexWithColumns(null, builder.build()));
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public boolean dropIndex(final String indexName) throws SQLException {
+    final BlockingInterface stub = conn.getTMStub();
+
+    try {
+      return isSuccess(stub.dropIndex(null, conn.getSessionedString(indexName)));
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
   public void close() throws IOException {
   }
 }

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java b/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
index aeca72b..e7317e5 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
@@ -161,6 +161,7 @@ public class QueryClientImpl implements QueryClient {
     }
 
     return response;
+
   }
 
   @Override
@@ -305,7 +306,6 @@ public class QueryClientImpl implements QueryClient {
         .setFetchRowNum(fetchRowNum)
         .build();
 
-
     GetQueryResultDataResponse response;
     try {
       response = stub.getQueryResultData(null, request);
@@ -451,7 +451,7 @@ public class QueryClientImpl implements QueryClient {
   public int getMaxRows() {
   	return this.maxRows;
   }
-  
+
   public QueryInfoProto getQueryInfo(final QueryId queryId) throws SQLException {
 
     final BlockingInterface stub = conn.getTMStub();

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java b/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
index f875335..4900188 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
@@ -374,7 +374,6 @@ public class SessionConnection implements Closeable {
         builder.setBaseDatabaseName(baseDatabase);
       }
 
-
       CreateSessionResponse response = null;
 
       try {

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java b/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
index c81fafc..b66d451 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
@@ -28,6 +28,7 @@ import org.apache.tajo.catalog.TableDesc;
 import org.apache.tajo.catalog.TableMeta;
 import org.apache.tajo.catalog.partition.PartitionMethodDesc;
 import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.catalog.proto.CatalogProtos.IndexDescProto;
 import org.apache.tajo.ipc.ClientProtos.*;
 import org.apache.tajo.jdbc.TajoMemoryResultSet;
 import org.apache.tajo.service.ServiceTracker;
@@ -227,4 +228,39 @@ public class TajoClientImpl extends SessionConnection implements TajoClient, Que
   public List<CatalogProtos.FunctionDescProto> getFunctions(final String functionName) throws SQLException {
     return catalogClient.getFunctions(functionName);
   }
+
+  @Override
+  public IndexDescProto getIndex(String indexName) throws SQLException {
+    return catalogClient.getIndex(indexName);
+  }
+
+  @Override
+  public boolean existIndex(String indexName) throws SQLException {
+    return catalogClient.existIndex(indexName);
+  }
+
+  @Override
+  public List<IndexDescProto> getIndexes(String tableName) throws SQLException {
+    return catalogClient.getIndexes(tableName);
+  }
+
+  @Override
+  public boolean hasIndexes(String tableName) throws SQLException {
+    return catalogClient.hasIndexes(tableName);
+  }
+
+  @Override
+  public IndexDescProto getIndex(String tableName, String[] columnNames) throws SQLException {
+    return catalogClient.getIndex(tableName, columnNames);
+  }
+
+  @Override
+  public boolean existIndex(String tableName, String[] columnName) throws SQLException {
+    return catalogClient.existIndex(tableName, columnName);
+  }
+
+  @Override
+  public boolean dropIndex(String indexName) throws SQLException {
+    return catalogClient.dropIndex(indexName);
+  }
 }

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-client/src/main/proto/ClientProtos.proto
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/proto/ClientProtos.proto b/tajo-client/src/main/proto/ClientProtos.proto
index ecfbbd9..49b0c40 100644
--- a/tajo-client/src/main/proto/ClientProtos.proto
+++ b/tajo-client/src/main/proto/ClientProtos.proto
@@ -261,3 +261,13 @@ message GetQueryInfoResponse {
   optional QueryInfoProto queryInfo = 2;
 }
 
+message CreateIndexResponse {
+  required ReturnState result = 1;
+  optional IndexDescProto indexDesc = 2;
+}
+
+message GetIndexWithColumnsRequest {
+  required SessionIdProto sessionId = 1;
+  required string tableName = 2;
+  repeated string columnNames = 3;
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-client/src/main/proto/TajoMasterClientProtocol.proto
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/proto/TajoMasterClientProtocol.proto b/tajo-client/src/main/proto/TajoMasterClientProtocol.proto
index 57bb2db..78f0f30 100644
--- a/tajo-client/src/main/proto/TajoMasterClientProtocol.proto
+++ b/tajo-client/src/main/proto/TajoMasterClientProtocol.proto
@@ -70,4 +70,13 @@ service TajoMasterClientProtocolService {
   rpc getTableList(SessionedStringProto) returns (StringListResponse);
   rpc getTableDesc(SessionedStringProto) returns (TableResponse);
   rpc getFunctionList(SessionedStringProto) returns (FunctionListResponse);
+
+  // Index Management APIs
+  rpc getIndexWithName(SessionedStringProto) returns (IndexResponse);
+  rpc existIndexWithName(SessionedStringProto) returns (ReturnState);
+  rpc getIndexesForTable(SessionedStringProto) returns (IndexListResponse);
+  rpc existIndexesForTable(SessionedStringProto) returns (ReturnState);
+  rpc getIndexWithColumns(GetIndexWithColumnsRequest) returns (IndexResponse);
+  rpc existIndexWithColumns(GetIndexWithColumnsRequest) returns (ReturnState);
+  rpc dropIndex(SessionedStringProto) returns (ReturnState);
 }

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-common/src/main/java/org/apache/tajo/OverridableConf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/OverridableConf.java b/tajo-common/src/main/java/org/apache/tajo/OverridableConf.java
index c22f054..0062c43 100644
--- a/tajo-common/src/main/java/org/apache/tajo/OverridableConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/OverridableConf.java
@@ -178,7 +178,7 @@ public class OverridableConf extends KeyValueSet {
 
   @Override
   public float getFloat(ConfigKey key) {
-    return getLong(key, null);
+    return getFloat(key, null);
   }
 
   public void put(ConfigKey key, String val) {

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/SessionVars.java b/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
index 832a5b4..6c7d075 100644
--- a/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
+++ b/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
@@ -126,6 +126,11 @@ public enum SessionVars implements ConfigKey {
   NULL_CHAR(ConfVars.$TEXT_NULL, "null char of text file output", DEFAULT),
   CODEGEN(ConfVars.$CODEGEN, "Runtime code generation enabled (experiment)", DEFAULT),
 
+  // for index
+  INDEX_ENABLED(ConfVars.$INDEX_ENABLED, "index scan enabled", DEFAULT),
+  INDEX_SELECTIVITY_THRESHOLD(ConfVars.$INDEX_SELECTIVITY_THRESHOLD, "the selectivity threshold for index scan", DEFAULT),
+
+  // for partition overwrite
   PARTITION_NO_RESULT_OVERWRITE_ENABLED(ConfVars.$PARTITION_NO_RESULT_OVERWRITE_ENABLED,
     "If True, a partitioned table is overwritten even if a sub query leads to no result. "
     + "Otherwise, the table data will be kept if there is no result", DEFAULT),

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
index 3336be6..4b4de62 100644
--- a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
@@ -332,6 +332,10 @@ public class TajoConf extends Configuration {
     $MAX_OUTPUT_FILE_SIZE("tajo.query.max-outfile-size-mb", 0), // zero means infinite
     $CODEGEN("tajo.executor.codegen.enabled", false), // Runtime code generation (todo this is broken)
 
+    // for index
+    $INDEX_ENABLED("tajo.query.index.enabled", false),
+    $INDEX_SELECTIVITY_THRESHOLD("tajo.query.index.selectivity.threshold", 0.05f),
+
     // Client -----------------------------------------------------------------
     $CLIENT_SESSION_EXPIRY_TIME("tajo.client.session.expiry-time-sec", 3600), // default time is one hour.
 
@@ -517,7 +521,7 @@ public class TajoConf extends Configuration {
   }
 
   public static long getLongVar(Configuration conf, ConfVars var) {
-    assert (var.valClass == Long.class || var.valClass == Integer.class);
+    assert (var.valClass == Long.class || var.valClass == Integer.class || var.valClass == Float.class);
     if (var.valClass == Integer.class) {
       return conf.getInt(var.varname, var.defaultIntVal);
     } else {

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-common/src/main/java/org/apache/tajo/exception/ErrorMessages.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/exception/ErrorMessages.java b/tajo-common/src/main/java/org/apache/tajo/exception/ErrorMessages.java
index 088fb91..3f05f44 100644
--- a/tajo-common/src/main/java/org/apache/tajo/exception/ErrorMessages.java
+++ b/tajo-common/src/main/java/org/apache/tajo/exception/ErrorMessages.java
@@ -64,6 +64,9 @@ public class ErrorMessages {
     ADD_MESSAGE(UNDEFINED_FUNCTION, "function does not exist: %s", 1);
     ADD_MESSAGE(UNDEFINED_PARTITION, "partition '%s' does not exist", 1);
     ADD_MESSAGE(UNDEFINED_OPERATOR, "operator does not exist: '%s'", 1);
+    ADD_MESSAGE(UNDEFINED_INDEX_FOR_TABLE, "index ''%s' does not exist", 1);
+    ADD_MESSAGE(UNDEFINED_INDEX_FOR_COLUMNS, "index does not exist for '%s' columns of '%s' table", 2);
+    ADD_MESSAGE(UNDEFINED_INDEX_NAME, "index name '%s' does not exist", 1);
 
     ADD_MESSAGE(DUPLICATE_TABLESPACE, "tablespace '%s' already exists", 1);
     ADD_MESSAGE(DUPLICATE_DATABASE, "database '%s' already exists", 1);

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-common/src/main/java/org/apache/tajo/exception/ReturnStateUtil.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/exception/ReturnStateUtil.java b/tajo-common/src/main/java/org/apache/tajo/exception/ReturnStateUtil.java
index 862889e..fb6b9a5 100644
--- a/tajo-common/src/main/java/org/apache/tajo/exception/ReturnStateUtil.java
+++ b/tajo-common/src/main/java/org/apache/tajo/exception/ReturnStateUtil.java
@@ -30,8 +30,10 @@ import org.apache.tajo.exception.TajoExceptionInterface;
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.ReturnState;
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.StringListResponse;
+import org.apache.tajo.util.StringUtils;
 
 import java.util.Collection;
+import java.util.List;
 
 public class ReturnStateUtil {
 
@@ -170,8 +172,13 @@ public class ReturnStateUtil {
     return returnError(ResultCode.UNDEFINED_PARTITION_METHOD, tbName);
   }
 
-  public static ReturnState errUndefinedIndex(String tbName, String columnName) {
-    return returnError(ResultCode.UNDEFINED_INDEX, tbName, columnName);
+  public static ReturnState errUndefinedIndex(String tbName) {
+    return returnError(ResultCode.UNDEFINED_INDEX_FOR_TABLE, tbName);
+  }
+
+  public static ReturnState errUndefinedIndex(String tbName, List<String> columnNameList) {
+    String columnNames = StringUtils.join(columnNameList, ",");
+    return returnError(ResultCode.UNDEFINED_INDEX_FOR_COLUMNS, columnNames, tbName);
   }
 
   public static ReturnState errUndefinedIndexName(String indexName) {

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-common/src/main/java/org/apache/tajo/util/TUtil.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/TUtil.java b/tajo-common/src/main/java/org/apache/tajo/util/TUtil.java
index 66e8acc..f24be61 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/TUtil.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/TUtil.java
@@ -21,6 +21,8 @@ package org.apache.tajo.util;
 import com.google.common.base.Objects;
 
 import java.lang.reflect.Array;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -268,6 +270,14 @@ public class TUtil {
     return element.getClassName() + ":" + element.getMethodName() + "(" + element.getLineNumber() +")";
   }
 
+  public static URI stringToURI(String str) {
+    try {
+      return new URI(str);
+    } catch (URISyntaxException e) {
+      throw new RuntimeException("Cannot convert " + str + " to the URI type", e);
+    }
+  }
+
   public static <T> T checkTypeAndGet(Object instance, Class<T> type) {
     if (!type.isInstance(instance)) {
       throw new IllegalArgumentException(instance.getClass().getSimpleName()

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-common/src/main/proto/errors.proto
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/proto/errors.proto b/tajo-common/src/main/proto/errors.proto
index 1ad9c85..5e2ecc0 100644
--- a/tajo-common/src/main/proto/errors.proto
+++ b/tajo-common/src/main/proto/errors.proto
@@ -103,11 +103,12 @@ enum ResultCode {
   UNDEFINED_TABLE                       = 514; // ?
   UNDEFINED_COLUMN                      = 515; // SQLState: 42703
   UNDEFINED_FUNCTION                    = 516; // SQLState: 42883
-  UNDEFINED_INDEX                       = 517; // ?
-  UNDEFINED_INDEX_NAME                  = 518; // ?
-  UNDEFINED_PARTITION                   = 519; // ?
-  UNDEFINED_PARTITION_METHOD            = 520; // ?
-  UNDEFINED_OPERATOR                    = 521; // SQLState: 42883 (=UNDEFINED_FUNCTION)
+  UNDEFINED_INDEX_FOR_TABLE             = 517; // ?
+  UNDEFINED_INDEX_FOR_COLUMNS           = 518; // ?
+  UNDEFINED_INDEX_NAME                  = 519; // ?
+  UNDEFINED_PARTITION                   = 520; // ?
+  UNDEFINED_PARTITION_METHOD            = 521; // ?
+  UNDEFINED_OPERATOR                    = 522; // SQLState: 42883 (=UNDEFINED_FUNCTION)
 
   DUPLICATE_TABLESPACE                  = 531;
   DUPLICATE_DATABASE                    = 532; // SQLState: 42P04

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-core/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4 b/tajo-core/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4
index 1137b30..b07fb8f 100644
--- a/tajo-core/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4
+++ b/tajo-core/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4
@@ -77,8 +77,17 @@ schema_statement
   ;
 
 index_statement
-  : CREATE (u=UNIQUE)? INDEX n=identifier ON t=table_name (m=method_specifier)?
-    LEFT_PAREN s=sort_specifier_list RIGHT_PAREN p=param_clause?
+  : create_index_statement
+  | drop_index_statement
+  ;
+
+create_index_statement
+  : CREATE (u=UNIQUE)? INDEX index_name = identifier ON table_name (method_specifier)?
+    LEFT_PAREN sort_specifier_list RIGHT_PAREN param_clause? (where_clause)? (LOCATION path=Character_String_Literal)?
+  ;
+
+drop_index_statement
+  : DROP INDEX index_name = identifier
   ;
 
 database_definition

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-core/src/main/java/org/apache/tajo/engine/codegen/ExecutorPreCompiler.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/ExecutorPreCompiler.java b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/ExecutorPreCompiler.java
index 87d33de..fbd88ad 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/ExecutorPreCompiler.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/ExecutorPreCompiler.java
@@ -214,4 +214,11 @@ public class ExecutorPreCompiler extends BasicLogicalPlanVisitor<ExecutorPreComp
 
     return node;
   }
+
+  @Override
+  public LogicalNode visitIndexScan(CompilationContext context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                    IndexScanNode node, Stack<LogicalNode> stack) throws TajoException {
+    visitScan(context, plan, block, node, stack);
+    return node;
+  }
 }

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java b/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
index d99a5cf..c50d5be 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
@@ -27,7 +27,9 @@ import org.antlr.v4.runtime.tree.TerminalNode;
 import org.apache.tajo.SessionVars;
 import org.apache.tajo.algebra.*;
 import org.apache.tajo.algebra.Aggregation.GroupType;
+import org.apache.tajo.algebra.CreateIndex.IndexMethodSpec;
 import org.apache.tajo.algebra.LiteralValue.LiteralType;
+import org.apache.tajo.algebra.Sort.SortSpec;
 import org.apache.tajo.engine.parser.SQLParser.*;
 import org.apache.tajo.storage.StorageConstants;
 import org.apache.tajo.util.StringUtils;
@@ -1221,6 +1223,51 @@ public class SQLAnalyzer extends SQLParserBaseVisitor<Expr> {
   }
 
   @Override
+  public Expr visitCreate_index_statement(SQLParser.Create_index_statementContext ctx) {
+    String indexName = ctx.index_name.getText();
+    String tableName = ctx.table_name().getText();
+    Relation relation = new Relation(tableName);
+    SortSpec[] sortSpecs = buildSortSpecs(ctx.sort_specifier_list());
+    NamedExpr[] targets = new NamedExpr[sortSpecs.length];
+    Projection projection = new Projection();
+    int i = 0;
+    for (SortSpec sortSpec : sortSpecs) {
+      targets[i++] = new NamedExpr(sortSpec.getKey());
+    }
+    projection.setNamedExprs(targets);
+    projection.setChild(relation);
+
+    CreateIndex createIndex = new CreateIndex(indexName, sortSpecs);
+    if (checkIfExist(ctx.UNIQUE())) {
+      createIndex.setUnique(true);
+    }
+    if (checkIfExist(ctx.method_specifier())) {
+      String methodName = ctx.method_specifier().identifier().getText();
+      createIndex.setMethodSpec(new IndexMethodSpec(methodName));
+    }
+    if (checkIfExist(ctx.param_clause())) {
+      Map<String, String> params = getParams(ctx.param_clause());
+      createIndex.setParams(params);
+    }
+    if (checkIfExist(ctx.where_clause())) {
+      Selection selection = visitWhere_clause(ctx.where_clause());
+      selection.setChild(relation);
+      projection.setChild(selection);
+    }
+    if (checkIfExist(ctx.LOCATION())) {
+      createIndex.setIndexPath(stripQuote(ctx.path.getText()));
+    }
+    createIndex.setChild(projection);
+    return createIndex;
+  }
+
+  @Override
+  public Expr visitDrop_index_statement(SQLParser.Drop_index_statementContext ctx) {
+    String indexName = ctx.identifier().getText();
+    return new DropIndex(indexName);
+  }
+
+  @Override
   public Expr visitDatabase_definition(@NotNull SQLParser.Database_definitionContext ctx) {
     return new CreateDatabase(ctx.identifier().getText(), null, checkIfExist(ctx.if_not_exists()));
   }

http://git-wip-us.apache.org/repos/asf/tajo/blob/9840d378/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
index 377aebe..ef3d039 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
@@ -29,7 +29,6 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.fs.Path;
 import org.apache.tajo.SessionVars;
-import org.apache.tajo.catalog.CatalogUtil;
 import org.apache.tajo.catalog.Column;
 import org.apache.tajo.catalog.SortSpec;
 import org.apache.tajo.catalog.proto.CatalogProtos;
@@ -48,12 +47,13 @@ import org.apache.tajo.plan.LogicalPlan;
 import org.apache.tajo.plan.logical.*;
 import org.apache.tajo.plan.serder.LogicalNodeDeserializer;
 import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.storage.*;
+import org.apache.tajo.storage.FileTablespace;
+import org.apache.tajo.storage.StorageConstants;
+import org.apache.tajo.storage.TablespaceManager;
 import org.apache.tajo.storage.fragment.FileFragment;
 import org.apache.tajo.storage.fragment.Fragment;
 import org.apache.tajo.storage.fragment.FragmentConvertor;
 import org.apache.tajo.util.FileUtil;
-import org.apache.tajo.util.IndexUtil;
 import org.apache.tajo.util.StringUtils;
 import org.apache.tajo.util.TUtil;
 import org.apache.tajo.worker.TaskAttemptContext;
@@ -235,11 +235,18 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
         return new LimitExec(ctx, limitNode.getInSchema(),
             limitNode.getOutSchema(), leftExec, limitNode);
 
-      case BST_INDEX_SCAN:
+      case INDEX_SCAN:
         IndexScanNode indexScanNode = (IndexScanNode) logicalNode;
         leftExec = createIndexScanExec(ctx, indexScanNode);
         return leftExec;
 
+      case CREATE_INDEX:
+        CreateIndexNode createIndexNode = (CreateIndexNode) logicalNode;
+        stack.push(createIndexNode);
+        leftExec = createPlanRecursive(ctx, createIndexNode.getChild(), stack);
+        stack.pop();
+        return new StoreIndexExec(ctx, createIndexNode, leftExec);
+
       default:
         return null;
     }
@@ -1185,21 +1192,10 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
     Preconditions.checkNotNull(ctx.getTable(annotation.getCanonicalName()),
         "Error: There is no table matched to %s", annotation.getCanonicalName());
 
-    FragmentProto [] fragmentProtos = ctx.getTables(annotation.getTableName());
-    List<FileFragment> fragments =
-        FragmentConvertor.convert(ctx.getConf(), fragmentProtos);
-
-    String indexName = IndexUtil.getIndexNameOfFrag(fragments.get(0), annotation.getSortKeys());
-    FileTablespace sm = (FileTablespace) TablespaceManager.get(fragments.get(0).getPath().toUri()).get();
-    String dbName = CatalogUtil.extractQualifier(annotation.getTableName());
-    String simpleName = CatalogUtil.extractSimpleName(annotation.getTableName());
-    Path indexPath = new Path(new Path(sm.getTableUri(dbName, simpleName)), "index");
-
-    TupleComparator comp = new BaseTupleComparator(annotation.getKeySchema(),
-        annotation.getSortKeys());
-    return new BSTIndexScanExec(ctx, annotation, fragments.get(0), new Path(indexPath, indexName),
-        annotation.getKeySchema(), comp, annotation.getDatum());
-
+    FragmentProto [] fragments = ctx.getTables(annotation.getTableName());
+    Preconditions.checkState(fragments.length == 1);
+    return new BSTIndexScanExec(ctx, annotation, fragments[0], annotation.getIndexPath(),
+        annotation.getKeySchema(), annotation.getPredicates());
   }
 
   public static EnforceProperty getAlgorithmEnforceProperty(Enforcer enforcer, LogicalNode node) {