Fix Cassandra case insensitive issue (#2897)

Author: KodaiD

core/src/main/java/com/scalar/db/storage/cassandra/CassandraAdmin.java

@@ -139,7 +139,7 @@ private void upsertIntoNamespacesTable(String keyspace) {
     String insertQuery =
         QueryBuilder.insertInto(
                 quoteIfNecessary(metadataKeyspace), quoteIfNecessary(NAMESPACES_TABLE))
-            .value(NAMESPACES_NAME_COL, quoteIfNecessary(keyspace))
+            .value(NAMESPACES_NAME_COL, keyspace)
             .toString();
     clusterManager.getSession().execute(insertQuery);
   }
@@ -180,7 +180,7 @@ private void deleteFromNamespacesTable(String keyspace) {
     String deleteQuery =
         QueryBuilder.delete()
             .from(quoteIfNecessary(metadataKeyspace), quoteIfNecessary(NAMESPACES_TABLE))
-            .where(QueryBuilder.eq(NAMESPACES_NAME_COL, quoteIfNecessary(keyspace)))
+            .where(QueryBuilder.eq(NAMESPACES_NAME_COL, keyspace))
             .toString();
     clusterManager.getSession().execute(deleteQuery);
   }
@@ -279,7 +279,9 @@ private TableMetadata createTableMetadata(com.datastax.driver.core.TableMetadata
       ClusteringOrder clusteringOrder = metadata.getClusteringOrder().get(i);
       builder.addClusteringKey(clusteringColumnName, convertOrder(clusteringOrder));
     }
-    metadata.getIndexes().forEach(i -> builder.addSecondaryIndex(i.getTarget()));
+    metadata
+        .getIndexes()
+        .forEach(i -> builder.addSecondaryIndex(unquoteIfNecessary(i.getTarget())));
     return builder.build();
   }
 
@@ -349,7 +351,7 @@ public boolean namespaceExists(String namespace) throws ExecutionException {
       String query =
           QueryBuilder.select(NAMESPACES_NAME_COL)
               .from(quoteIfNecessary(metadataKeyspace), quoteIfNecessary(NAMESPACES_TABLE))
-              .where(QueryBuilder.eq(NAMESPACES_NAME_COL, quoteIfNecessary(namespace)))
+              .where(QueryBuilder.eq(NAMESPACES_NAME_COL, namespace))
               .toString();
       ResultSet resultSet = clusterManager.getSession().execute(query);
 
@@ -396,8 +398,8 @@ public void addNewColumnToTable(
     }
     try {
       String alterTableQuery =
-          SchemaBuilder.alterTable(namespace, table)
-              .addColumn(columnName)
+          SchemaBuilder.alterTable(quoteIfNecessary(namespace), quoteIfNecessary(table))
+              .addColumn(quoteIfNecessary(columnName))
               .type(toCassandraDataType(columnType))
               .getQueryString();
 
@@ -686,4 +688,14 @@ public String toString() {
       return strategyName;
     }
   }
+
+  private String unquoteIfNecessary(String identifier) {
+    if (identifier == null) {
+      return null;
+    }
+    if (identifier.length() >= 2 && identifier.startsWith("\"") && identifier.endsWith("\"")) {
+      return identifier.substring(1, identifier.length() - 1).replace("\"\"", "\"");
+    }
+    return identifier;
+  }
 }

core/src/test/java/com/scalar/db/storage/cassandra/CassandraAdminTest.java

@@ -229,7 +229,7 @@ private void verifyInsertIntoKeyspacesTableQuery(String keyspace) {
         QueryBuilder.insertInto(
                 quoteIfNecessary(METADATA_KEYSPACE),
                 quoteIfNecessary(CassandraAdmin.NAMESPACES_TABLE))
-            .value(CassandraAdmin.NAMESPACES_NAME_COL, quoteIfNecessary(keyspace))
+            .value(CassandraAdmin.NAMESPACES_NAME_COL, keyspace)
             .toString();
     verify(cassandraSession).execute(query);
   }
@@ -569,7 +569,7 @@ private void verifyDeleteFromKeyspacesTableQuery(String keyspace) {
             .from(
                 quoteIfNecessary(METADATA_KEYSPACE),
                 quoteIfNecessary(CassandraAdmin.NAMESPACES_TABLE))
-            .where(QueryBuilder.eq(CassandraAdmin.NAMESPACES_NAME_COL, quoteIfNecessary(keyspace)))
+            .where(QueryBuilder.eq(CassandraAdmin.NAMESPACES_NAME_COL, keyspace))
             .toString();
     verify(cassandraSession).execute(query);
   }