Migrate Cassandra product test to integration test

plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraConnectorTest.java

@@ -909,6 +909,43 @@ public void testClusteringKeyPushdownInequality()
         }
     }
 
+    @Test
+    void testMultiColumnKey()
+    {
+        String tableName = "test_multi_column_key" + randomNameSuffix();
+        onCassandra("CREATE TABLE tpch." + tableName + "(user_id text, key text, updated_at timestamp, value text, PRIMARY KEY (user_id, key, updated_at))");
+        onCassandra("INSERT INTO tpch." + tableName + " (user_id, key, updated_at, value) VALUES ('Alice', 'a1', '2015-01-01 01:01:01', 'Test value 1')");
+        onCassandra("INSERT INTO tpch." + tableName + " (user_id, key, updated_at, value) VALUES ('Bob', 'b1', '2014-02-02 03:04:05', 'Test value 2')");
+
+        assertContainsEventually(() -> computeActual("SHOW TABLES FROM cassandra.tpch"), resultBuilder(getSession(), VARCHAR)
+                .row(tableName)
+                .build(), new Duration(1, MINUTES));
+
+        // equality filter on clustering key
+        assertQuery("SELECT value FROM " + tableName + " WHERE key = 'a1'", "VALUES 'Test value 1'");
+
+        // equality filter on primary and clustering keys
+        assertQuery("SELECT value FROM " + tableName + " WHERE user_id = 'Alice' and key = 'a1' and updated_at = TIMESTAMP '2015-01-01 01:01:01Z'",
+                "VALUES 'Test value 1'");
+
+        // mixed filter on primary and clustering keys
+        assertQuery("SELECT value FROM " + tableName + " WHERE user_id = 'Alice' and key < 'b' and updated_at >= TIMESTAMP '2015-01-01 01:01:01Z'",
+                "VALUES 'Test value 1'");
+
+        // filter on primary key doesn't match
+        assertQueryReturnsEmptyResult("SELECT value FROM " + tableName + " WHERE user_id = 'George'");
+
+        // filter on prefix of clustering key
+        assertQuery("SELECT value FROM " + tableName + " WHERE user_id = 'Bob' and key = 'b1'",
+                "VALUES 'Test value 2'");
+
+        // filter on second clustering key
+        assertQuery("SELECT value FROM " + tableName + " WHERE user_id = 'Bob' and updated_at = TIMESTAMP '2014-02-02 03:04:05Z'",
+                "VALUES 'Test value 2'");
+
+        onCassandra("DROP TABLE tpch." + tableName);
+    }
+
     @Test
     public void testSelectWithSecondaryIndex()
     {