Add support for tuple type in the Cassandra connector

This adds support to read columns of type tuple from Cassandra as
VARCHAR. Remaining unsupported types will now throw a NotSupportedException
error instead of failing a null pointer check.

Author: BryanCutler

presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraType.java

@@ -17,8 +17,10 @@
 import com.datastax.driver.core.LocalDate;
 import com.datastax.driver.core.ProtocolVersion;
 import com.datastax.driver.core.Row;
+import com.datastax.driver.core.TupleValue;
 import com.datastax.driver.core.utils.Bytes;
 import com.facebook.presto.cassandra.util.CassandraCqlUtils;
+import com.facebook.presto.common.NotSupportedException;
 import com.facebook.presto.common.predicate.NullableValue;
 import com.facebook.presto.common.type.BigintType;
 import com.facebook.presto.common.type.BooleanType;
@@ -54,6 +56,7 @@
 import static io.airlift.slice.Slices.wrappedBuffer;
 import static java.lang.Float.floatToRawIntBits;
 import static java.lang.Float.intBitsToFloat;
+import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 
 public enum CassandraType
@@ -81,7 +84,8 @@ public enum CassandraType
     VARINT(createUnboundedVarcharType(), BigInteger.class),
     LIST(createUnboundedVarcharType(), null),
     MAP(createUnboundedVarcharType(), null),
-    SET(createUnboundedVarcharType(), null);
+    SET(createUnboundedVarcharType(), null),
+    TUPLE(createUnboundedVarcharType(), null);
 
     private static class Constants
     {
@@ -162,14 +166,16 @@ public static CassandraType getCassandraType(DataType.Name name)
                 return TIMEUUID;
             case TINYINT:
                 return TINYINT;
+            case TUPLE:
+                return TUPLE;
             case UUID:
                 return UUID;
             case VARCHAR:
                 return VARCHAR;
             case VARINT:
                 return VARINT;
             default:
-                return null;
+                throw new NotSupportedException(format("Unsupported Cassandra type: %s", name));
         }
     }
 
@@ -231,6 +237,9 @@ public static NullableValue getColumnValue(Row row, int position, CassandraType
                 case MAP:
                     checkTypeArguments(cassandraType, 2, typeArguments);
                     return NullableValue.of(nativeType, utf8Slice(buildMapValue(row, position, typeArguments.get(0), typeArguments.get(1))));
+                case TUPLE:
+                    TupleValue tupleValue = row.getTupleValue(position);
+                    return NullableValue.of(nativeType, utf8Slice(tupleValue.toString()));
                 default:
                     throw new IllegalStateException("Handling of type " + cassandraType
                             + " is not implemented");

presto-cassandra/src/main/java/com/facebook/presto/cassandra/NativeCassandraSession.java

@@ -28,6 +28,7 @@
 import com.datastax.driver.core.Session;
 import com.datastax.driver.core.Statement;
 import com.datastax.driver.core.TableMetadata;
+import com.datastax.driver.core.TableOptionsMetadata;
 import com.datastax.driver.core.TokenRange;
 import com.datastax.driver.core.VersionNumber;
 import com.datastax.driver.core.exceptions.NoHostAvailableException;
@@ -65,6 +66,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Stream;
@@ -216,10 +218,10 @@ public CassandraTable getTable(SchemaTableName schemaTableName)
         }
 
         // check if there is a comment to establish column ordering
-        String comment = tableMeta.getOptions().getComment();
+        Optional<String> comment = Optional.ofNullable(tableMeta.getOptions()).map(TableOptionsMetadata::getComment);
         Set<String> hiddenColumns = ImmutableSet.of();
-        if (comment != null && comment.startsWith(PRESTO_COMMENT_METADATA)) {
-            String columnOrderingString = comment.substring(PRESTO_COMMENT_METADATA.length());
+        if (comment.isPresent() && comment.get().startsWith(PRESTO_COMMENT_METADATA)) {
+            String columnOrderingString = comment.get().substring(PRESTO_COMMENT_METADATA.length());
 
             // column ordering
             List<ExtraColumnMetadata> extras = extraColumnMetadataCodec.fromJson(columnOrderingString);
@@ -361,7 +363,7 @@ private CassandraColumnHandle buildColumnHandle(AbstractTableMetadata tableMetad
     {
         CassandraType cassandraType = CassandraType.getCassandraType(columnMeta.getType().getName());
         List<CassandraType> typeArguments = null;
-        if (cassandraType != null && cassandraType.getTypeArgumentSize() > 0) {
+        if (cassandraType.getTypeArgumentSize() > 0) {
             List<DataType> typeArgs = columnMeta.getType().getTypeArguments();
             switch (cassandraType.getTypeArgumentSize()) {
                 case 1:

presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraServer.java

@@ -14,6 +14,7 @@
 package com.facebook.presto.cassandra;
 
 import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Metadata;
 import com.datastax.driver.core.ResultSet;
 import com.datastax.driver.core.Row;
 import com.facebook.airlift.json.JsonCodec;
@@ -58,6 +59,7 @@ public class CassandraServer
     private final GenericContainer<?> dockerContainer;
 
     private final CassandraSession session;
+    private final Metadata metadata;
 
     public CassandraServer()
             throws Exception
@@ -82,6 +84,7 @@ public CassandraServer()
                 JsonCodec.listJsonCodec(ExtraColumnMetadata.class),
                 cluster,
                 new Duration(1, MINUTES));
+        this.metadata = cluster.getMetadata();
 
         try {
             checkConnectivity(session);
@@ -117,6 +120,11 @@ public CassandraSession getSession()
         return requireNonNull(session, "cluster is null");
     }
 
+    public Metadata getMetadata()
+    {
+        return metadata;
+    }
+
     public String getHost()
     {
         return dockerContainer.getContainerIpAddress();

presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraTestingUtils.java

@@ -13,6 +13,9 @@
  */
 package com.facebook.presto.cassandra;
 
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.TupleType;
 import com.datastax.driver.core.querybuilder.Insert;
 import com.datastax.driver.core.querybuilder.QueryBuilder;
 import com.facebook.presto.spi.SchemaTableName;
@@ -42,12 +45,12 @@ public class CassandraTestingUtils
 
     private CassandraTestingUtils() {}
 
-    public static void createTestTables(CassandraSession cassandraSession, String keyspace, Date date)
+    public static void createTestTables(CassandraSession cassandraSession, Metadata metadata, String keyspace, Date date)
     {
         createKeyspace(cassandraSession, keyspace);
-        createTableAllTypes(cassandraSession, new SchemaTableName(keyspace, TABLE_ALL_TYPES), date, 9);
-        createTableAllTypes(cassandraSession, new SchemaTableName(keyspace, TABLE_ALL_TYPES_INSERT), date, 0);
-        createTableAllTypesPartitionKey(cassandraSession, new SchemaTableName(keyspace, TABLE_ALL_TYPES_PARTITION_KEY), date);
+        createTableAllTypes(cassandraSession, metadata, new SchemaTableName(keyspace, TABLE_ALL_TYPES), date, 9);
+        createTableAllTypes(cassandraSession, metadata, new SchemaTableName(keyspace, TABLE_ALL_TYPES_INSERT), date, 0);
+        createTableAllTypesPartitionKey(cassandraSession, metadata, new SchemaTableName(keyspace, TABLE_ALL_TYPES_PARTITION_KEY), date);
         createTableClusteringKeys(cassandraSession, new SchemaTableName(keyspace, TABLE_CLUSTERING_KEYS), 9);
         createTableClusteringKeys(cassandraSession, new SchemaTableName(keyspace, TABLE_CLUSTERING_KEYS_LARGE), 1000);
         createTableMultiPartitionClusteringKeys(cassandraSession, new SchemaTableName(keyspace, TABLE_MULTI_PARTITION_CLUSTERING_KEYS));
@@ -142,7 +145,7 @@ public static void insertIntoTableClusteringKeysInequality(CassandraSession sess
         assertEquals(session.execute("SELECT COUNT(*) FROM " + table).all().get(0).getLong(0), rowsCount);
     }
 
-    public static void createTableAllTypes(CassandraSession session, SchemaTableName table, Date date, int rowsCount)
+    public static void createTableAllTypes(CassandraSession session, Metadata metadata, SchemaTableName table, Date date, int rowsCount)
     {
         session.execute("DROP TABLE IF EXISTS " + table);
         session.execute("CREATE TABLE " + table + " (" +
@@ -164,11 +167,12 @@ public static void createTableAllTypes(CassandraSession session, SchemaTableName
                 " typelist list<text>, " +
                 " typemap map<int, bigint>, " +
                 " typeset set<boolean>, " +
+                " typetuple tuple<bigint, varchar>, " +
                 ")");
-        insertTestData(session, table, date, rowsCount);
+        insertTestData(session, metadata, table, date, rowsCount);
     }
 
-    public static void createTableAllTypesPartitionKey(CassandraSession session, SchemaTableName table, Date date)
+    public static void createTableAllTypesPartitionKey(CassandraSession session, Metadata metadata, SchemaTableName table, Date date)
     {
         session.execute("DROP TABLE IF EXISTS " + table);
 
@@ -191,6 +195,7 @@ public static void createTableAllTypesPartitionKey(CassandraSession session, Sch
                 " typelist frozen <list<text>>, " +
                 " typemap frozen <map<int, bigint>>, " +
                 " typeset frozen <set<boolean>>, " +
+                " typetuple frozen <tuple<bigint, varchar>>, " +
                 " PRIMARY KEY ((" +
                 "   key, " +
                 "   typeuuid, " +
@@ -213,15 +218,18 @@ public static void createTableAllTypesPartitionKey(CassandraSession session, Sch
                 // TODO: NOT YET SUPPORTED AS A PARTITION KEY
                 "   typelist, " +
                 "   typemap, " +
-                "   typeset" +
+                "   typeset," +
+                "   typetuple" +
                 " ))" +
                 ")");
 
-        insertTestData(session, table, date, 9);
+        insertTestData(session, metadata, table, date, 9);
     }
 
-    private static void insertTestData(CassandraSession session, SchemaTableName table, Date date, int rowsCount)
+    private static void insertTestData(CassandraSession session, Metadata metadata, SchemaTableName table, Date date, int rowsCount)
     {
+        TupleType tupleType = metadata.newTupleType(DataType.bigint(), DataType.varchar());
+
         for (int rowNumber = 1; rowNumber <= rowsCount; rowNumber++) {
             Insert insert = QueryBuilder.insertInto(table.getSchemaName(), table.getTableName())
                     .value("key", "key " + rowNumber)
@@ -241,7 +249,8 @@ private static void insertTestData(CassandraSession session, SchemaTableName tab
                     .value("typetimeuuid", UUID.fromString(String.format("d2177dd0-eaa2-11de-a572-001b779c76e%d", rowNumber)))
                     .value("typelist", ImmutableList.of("list-value-1" + rowNumber, "list-value-2" + rowNumber))
                     .value("typemap", ImmutableMap.of(rowNumber, rowNumber + 1L, rowNumber + 2, rowNumber + 3L))
-                    .value("typeset", ImmutableSet.of(false, true));
+                    .value("typeset", ImmutableSet.of(false, true))
+                    .value("typetuple", tupleType.newValue((long) rowNumber, "row=" + rowNumber));
 
             session.execute(insert);
         }

presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraConnector.java

@@ -110,7 +110,7 @@ public void setup()
         this.server = new CassandraServer();
 
         String keyspace = "test_connector";
-        createTestTables(server.getSession(), keyspace, DATE);
+        createTestTables(server.getSession(), server.getMetadata(), keyspace, DATE);
 
         String connectorId = "cassandra-test";
         CassandraConnectorFactory connectorFactory = new CassandraConnectorFactory(connectorId);

presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraIntegrationSmokeTest.java

@@ -96,7 +96,7 @@ protected QueryRunner createQueryRunner()
         CassandraServer server = new CassandraServer();
         this.server = server;
         this.session = server.getSession();
-        createTestTables(session, KEYSPACE, DATE_TIME_LOCAL);
+        createTestTables(session, server.getMetadata(), KEYSPACE, DATE_TIME_LOCAL);
         return createCassandraQueryRunner(server);
     }
 

presto-docs/src/main/sphinx/connector/cassandra.rst

@@ -220,6 +220,7 @@ SET<?>            VARCHAR
 TEXT              VARCHAR
 TIMESTAMP         TIMESTAMP
 TIMEUUID          VARCHAR
+TUPLE             VARCHAR
 VARCHAR           VARCHAR
 VARINT            VARCHAR
 SMALLINT          INTEGER
@@ -230,7 +231,7 @@ DATE              DATE
 Any collection (LIST/MAP/SET) can be designated as FROZEN, and the value is
 mapped to VARCHAR. Additionally, blobs have the limitation that they cannot be empty.
 
-Types not mentioned in the table above are not supported (e.g. tuple or UDT).
+Data types not listed in the table above, such as UDT, are not supported.
 
 Partition keys can only be of the following types: