[Java] Fix CassandraIO ReadFn to quote column names for reserved keywords (#36459)

* Fix CassandraIO ReadFn to quote column names for reserved keywords

* Minor fixes

Author: Suvrat1629

sdks/java/io/cassandra/src/main/java/org/apache/beam/sdk/io/cassandra/ReadFn.java

@@ -50,6 +50,7 @@ public void processElement(@Element Read<T> read, OutputReceiver<T> receiver) th
           session.getCluster().getMetadata().getKeyspace(read.keyspace().get())
               .getTable(read.table().get()).getPartitionKey().stream()
               .map(ColumnMetadata::getName)
+              .map(ReadFn::quoteIdentifier)
               .collect(Collectors.joining(","));
 
       String query = generateRangeQuery(read, partitionKey, read.ringRanges() != null);
@@ -148,4 +149,13 @@ private static String buildInitialQuery(Read<?> spec, Boolean hasRingRange) {
   private static String getJoinerClause(String queryString) {
     return queryString.toUpperCase().contains("WHERE") ? " AND " : " WHERE ";
   }
+
+  static String quoteIdentifier(String identifier) {
+    if (identifier == null) {
+      return null;
+    }
+    // Escape any existing double quotes by doubling them
+    String escaped = identifier.replace("\"", "\"\"");
+    return "\"" + escaped + "\"";
+  }
 }