[CASSANALYTICS-183] - Pass quoted identifiers to sidecar API calls for mixed-case keyspace/table restore

CHANGES.txt

@@ -1,5 +1,6 @@
 0.5.0
 -----
+ * Mixed-case keyspace restore fails with keyspace does not exist when quoteIdentifiers is set (CASSANALYTICS-183)
  * TokenPartitioner fails to detect range gap in reader (CASSANALYTICS-180)
  * CDC reader stats silently dropped in SidecarCdcBuilder (CASSANALYTICS-191)
  * Add CapturePublishedSchema metric to SidecarCdcStats (CASSANALYTICS-189)

cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/QualifiedTableName.java

@@ -65,6 +65,14 @@ public String keyspace()
         return keyspace;
     }
 
+    /**
+     * @return the keyspace name, quoted with double quotes when {@code quoteIdentifiers} is set
+     */
+    public String maybeQuotedKeyspace()
+    {
+        return maybeQuote(keyspace);
+    }
+
     /**
      * @return the table name in Cassandra
      */
@@ -73,6 +81,14 @@ public String table()
         return table;
     }
 
+    /**
+     * @return the table name, quoted with double quotes when {@code quoteIdentifiers} is set
+     */
+    public String maybeQuotedTable()
+    {
+        return maybeQuote(table);
+    }
+
     /**
      * @return the identifiers should be quoted
      */

cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/QualifiedTableNameTest.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cassandra.spark.data;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class QualifiedTableNameTest
+{
+    @Test
+    void testUnquotedIdentifiersReturnedAsIs()
+    {
+        QualifiedTableName name = new QualifiedTableName("mykeyspace", "mytable", false);
+        assertThat(name.keyspace()).isEqualTo("mykeyspace");
+        assertThat(name.table()).isEqualTo("mytable");
+        assertThat(name.maybeQuotedKeyspace()).isEqualTo("mykeyspace");
+        assertThat(name.maybeQuotedTable()).isEqualTo("mytable");
+    }
+
+    @Test
+    void testQuotedIdentifiersWrappedInDoubleQuotes()
+    {
+        QualifiedTableName name = new QualifiedTableName("MyKeyspace", "MyTable", true);
+        assertThat(name.maybeQuotedKeyspace()).isEqualTo("\"MyKeyspace\"");
+        assertThat(name.maybeQuotedTable()).isEqualTo("\"MyTable\"");
+    }
+
+    @Test
+    void testRawAccessorsUnaffectedByQuoteFlag()
+    {
+        QualifiedTableName name = new QualifiedTableName("MyKeyspace", "MyTable", true);
+        assertThat(name.keyspace()).isEqualTo("MyKeyspace");
+        assertThat(name.table()).isEqualTo("MyTable");
+    }
+
+    @Test
+    void testToStringQuotesBothWhenFlagSet()
+    {
+        QualifiedTableName name = new QualifiedTableName("MyKeyspace", "MyTable", true);
+        assertThat(name.toString()).isEqualTo("\"MyKeyspace\".\"MyTable\"");
+    }
+
+    @Test
+    void testToStringUnquotedWhenFlagNotSet()
+    {
+        QualifiedTableName name = new QualifiedTableName("mykeyspace", "mytable", false);
+        assertThat(name.toString()).isEqualTo("mykeyspace.mytable");
+    }
+
+    @Test
+    void testDefaultConstructorDoesNotQuote()
+    {
+        QualifiedTableName name = new QualifiedTableName("MyKeyspace", "MyTable");
+        assertThat(name.quoteIdentifiers()).isFalse();
+        assertThat(name.maybeQuotedKeyspace()).isEqualTo("MyKeyspace");
+        assertThat(name.maybeQuotedTable()).isEqualTo("MyTable");
+    }
+
+    @Test
+    void testReservedWordIdentifiersQuoted()
+    {
+        QualifiedTableName name = new QualifiedTableName("keyspace", "table", true);
+        assertThat(name.maybeQuotedKeyspace()).isEqualTo("\"keyspace\"");
+        assertThat(name.maybeQuotedTable()).isEqualTo("\"table\"");
+    }
+}

cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java

@@ -289,11 +289,12 @@ protected String getCurrentKeyspaceSchema() throws Exception
     private TokenRangeReplicasResponse getTokenRangesAndReplicaSets()
     {
         CassandraContext context = getCassandraContext();
+        String quotedKeyspace = maybeQuotedIdentifier(bridge(), conf.quoteIdentifiers, conf.keyspace);
         try
         {
             long start = System.nanoTime();
             TokenRangeReplicasResponse response = context.getSidecarClient()
-                                                         .tokenRangeReplicas(new ArrayList<>(context.getCluster()), conf.keyspace)
+                                                         .tokenRangeReplicas(new ArrayList<>(context.getCluster()), quotedKeyspace)
                                                          .get();
             long elapsedTimeNanos = System.nanoTime() - start;
             LOGGER.info("Retrieved token ranges for {} instances in {} milliseconds",
@@ -303,8 +304,8 @@ private TokenRangeReplicasResponse getTokenRangesAndReplicaSets()
         }
         catch (ExecutionException | InterruptedException exception)
         {
-            LOGGER.error("Failed to get token ranges for keyspace {}", conf.keyspace, exception);
-            throw new SidecarApiCallException("Failed to get token ranges for keyspace" + conf.keyspace, exception);
+            LOGGER.error("Failed to get token ranges for keyspace {}", quotedKeyspace, exception);
+            throw new SidecarApiCallException("Failed to get token ranges for keyspace " + quotedKeyspace, exception);
         }
     }
 

cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/CloudStorageDataTransferApiImpl.java

@@ -103,8 +103,8 @@ public void createRestoreJob(CreateRestoreJobRequestPayload createRestoreJobRequ
         try
         {
             QualifiedTableName qualifiedTableName = jobInfo.qualifiedTableName();
-            sidecarClient.createRestoreJob(qualifiedTableName.keyspace(),
-                                           qualifiedTableName.table(),
+            sidecarClient.createRestoreJob(qualifiedTableName.maybeQuotedKeyspace(),
+                                           qualifiedTableName.maybeQuotedTable(),
                                            createRestoreJobRequestPayload).get();
         }
         catch (Exception exception)
@@ -121,8 +121,8 @@ public RestoreJobSummaryResponsePayload restoreJobSummary() throws SidecarApiCal
         try
         {
             QualifiedTableName qualifiedTableName = jobInfo.qualifiedTableName();
-            return sidecarClient.restoreJobSummary(qualifiedTableName.keyspace(),
-                                                   qualifiedTableName.table(),
+            return sidecarClient.restoreJobSummary(qualifiedTableName.maybeQuotedKeyspace(),
+                                                   qualifiedTableName.maybeQuotedTable(),
                                                    jobId).get();
         }
         catch (Exception exception)
@@ -168,8 +168,8 @@ public void updateRestoreJob(UpdateRestoreJobRequestPayload updateRestoreJobRequ
         {
             LOGGER.info("Updating the restore job. clusterId={} restoreJobId={}", clusterId, jobId);
             QualifiedTableName qualifiedTableName = jobInfo.qualifiedTableName();
-            sidecarClient.updateRestoreJob(qualifiedTableName.keyspace(),
-                                           qualifiedTableName.table(),
+            sidecarClient.updateRestoreJob(qualifiedTableName.maybeQuotedKeyspace(),
+                                           qualifiedTableName.maybeQuotedTable(),
                                            jobId,
                                            updateRestoreJobRequestPayload).get();
         }
@@ -188,8 +188,8 @@ public void abortRestoreJob() throws SidecarApiCallException
         {
             LOGGER.info("Abort job. clusterId={} restoreJobId={}", clusterId, jobId);
             QualifiedTableName qualifiedTableName = jobInfo.qualifiedTableName();
-            sidecarClient.abortRestoreJob(qualifiedTableName.keyspace(),
-                                          qualifiedTableName.table(),
+            sidecarClient.abortRestoreJob(qualifiedTableName.maybeQuotedKeyspace(),
+                                          qualifiedTableName.maybeQuotedTable(),
                                           jobId).get();
         }
         catch (Exception exception)
@@ -207,8 +207,8 @@ private CompletableFuture<Void> createRestoreSliceWithCustomRetry(SidecarInstanc
                                                                       RetryPolicy retryPolicy)
     {
         QualifiedTableName qualifiedTableName = jobInfo.qualifiedTableName();
-        CreateRestoreJobSliceRequest request = new CreateRestoreJobSliceRequest(qualifiedTableName.keyspace(),
-                                                                                qualifiedTableName.table(),
+        CreateRestoreJobSliceRequest request = new CreateRestoreJobSliceRequest(qualifiedTableName.maybeQuotedKeyspace(),
+                                                                                qualifiedTableName.maybeQuotedTable(),
                                                                                 jobInfo.getRestoreJobId(clusterId),
                                                                                 createSliceRequestPayload);
         return sidecarClient.executeRequestAsync(sidecarClient.requestBuilder()

cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCloudStorageDataTransferApi.java

@@ -169,8 +169,8 @@ private void createRestoreSliceInternal(String clusterId,
         SidecarClient sidecarClient = dataTransferApi.sidecarClient();
         QualifiedTableName qualifiedTableName = jobInfo.qualifiedTableName();
         UUID restoreJobId = jobInfo.getRestoreJobId(clusterId);
-        CreateRestoreJobSliceRequest request = new CreateRestoreJobSliceRequest(qualifiedTableName.keyspace(),
-                                                                                qualifiedTableName.table(),
+        CreateRestoreJobSliceRequest request = new CreateRestoreJobSliceRequest(qualifiedTableName.maybeQuotedKeyspace(),
+                                                                                qualifiedTableName.maybeQuotedTable(),
                                                                                 restoreJobId,
                                                                                 createSliceRequestPayload);
         RetryPolicy retryPolicy = new CloudStorageDataTransferApiImpl.ExecutorCreateSliceRetryPolicy(sidecarClient);
@@ -197,8 +197,8 @@ private void restoreJobProgressInternal(String clusterId,
         JobInfo jobInfo = dataTransferApi.jobInfo();
         QualifiedTableName qualifiedTableName = jobInfo.qualifiedTableName();
         UUID restoreJobId = jobInfo.getRestoreJobId(clusterId);
-        RestoreJobProgressRequestParams requestParams = new RestoreJobProgressRequestParams(qualifiedTableName.keyspace(),
-                                                                                            qualifiedTableName.table(),
+        RestoreJobProgressRequestParams requestParams = new RestoreJobProgressRequestParams(qualifiedTableName.maybeQuotedKeyspace(),
+                                                                                            qualifiedTableName.maybeQuotedTable(),
                                                                                             restoreJobId,
                                                                                             fetchPolicy);
         RestoreJobProgressResponsePayload jobProgress = restoreJobProgress(dataTransferApi, requestParams);

cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java

@@ -1189,7 +1189,9 @@ protected Sizing getSizing(CompletableFuture<RingResponse> ringFuture,
                                ReplicationFactor replicationFactor,
                                ClientConfig options)
     {
-        return SizingFactory.create(replicationFactor, options, consistencyLevel, keyspace, table, datacenter, sidecar, sidecarPort, ringFuture);
+        return SizingFactory.create(replicationFactor, options, consistencyLevel,
+                                    maybeQuotedKeyspace, maybeQuotedTable, datacenter,
+                                    sidecar, sidecarPort, ringFuture);
     }
 
     protected void await(CountDownLatch latch)

cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/CloudStorageDataTransferApiImplTest.java

@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cassandra.spark.bulkwriter.cloudstorage;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import o.a.c.sidecar.client.shaded.common.request.data.CreateRestoreJobRequestPayload;
+import o.a.c.sidecar.client.shaded.common.request.data.UpdateRestoreJobRequestPayload;
+import o.a.c.sidecar.client.shaded.common.response.data.RestoreJobSummaryResponsePayload;
+import o.a.c.sidecar.client.shaded.client.SidecarClient;
+import org.apache.cassandra.spark.bulkwriter.JobInfo;
+import org.apache.cassandra.spark.data.QualifiedTableName;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class CloudStorageDataTransferApiImplTest
+{
+    private static final String QUOTED_KEYSPACE = "\"MyKeyspace\"";
+    private static final String QUOTED_TABLE = "\"MyTable\"";
+    private static final UUID JOB_ID = UUID.randomUUID();
+
+    private SidecarClient sidecarClient;
+    private JobInfo jobInfo;
+    private CloudStorageDataTransferApiImpl api;
+
+    @BeforeEach
+    void setup()
+    {
+        sidecarClient = mock(SidecarClient.class);
+        jobInfo = mock(JobInfo.class);
+        when(jobInfo.qualifiedTableName()).thenReturn(new QualifiedTableName("MyKeyspace", "MyTable", true));
+        when(jobInfo.getRestoreJobId()).thenReturn(JOB_ID);
+        when(jobInfo.getRestoreJobId(null)).thenReturn(JOB_ID);
+        api = new CloudStorageDataTransferApiImpl(jobInfo, sidecarClient, mock(StorageClient.class), null);
+    }
+
+    @Test
+    void testCreateRestoreJobUsesQuotedIdentifiers() throws Exception
+    {
+        when(sidecarClient.createRestoreJob(eq(QUOTED_KEYSPACE), eq(QUOTED_TABLE), any()))
+            .thenReturn(CompletableFuture.completedFuture(null));
+
+        api.createRestoreJob(mock(CreateRestoreJobRequestPayload.class));
+
+        verify(sidecarClient).createRestoreJob(eq(QUOTED_KEYSPACE), eq(QUOTED_TABLE), any());
+    }
+
+    @Test
+    void testRestoreJobSummaryUsesQuotedIdentifiers() throws Exception
+    {
+        RestoreJobSummaryResponsePayload response = mock(RestoreJobSummaryResponsePayload.class);
+        when(sidecarClient.restoreJobSummary(eq(QUOTED_KEYSPACE), eq(QUOTED_TABLE), eq(JOB_ID)))
+            .thenReturn(CompletableFuture.completedFuture(response));
+
+        RestoreJobSummaryResponsePayload result = api.restoreJobSummary();
+
+        assertThat(result).isSameAs(response);
+        verify(sidecarClient).restoreJobSummary(eq(QUOTED_KEYSPACE), eq(QUOTED_TABLE), eq(JOB_ID));
+    }
+
+    @Test
+    void testUpdateRestoreJobUsesQuotedIdentifiers() throws Exception
+    {
+        when(sidecarClient.updateRestoreJob(eq(QUOTED_KEYSPACE), eq(QUOTED_TABLE), eq(JOB_ID), any()))
+            .thenReturn(CompletableFuture.completedFuture(null));
+
+        api.updateRestoreJob(mock(UpdateRestoreJobRequestPayload.class));
+
+        verify(sidecarClient).updateRestoreJob(eq(QUOTED_KEYSPACE), eq(QUOTED_TABLE), eq(JOB_ID), any());
+    }
+
+    @Test
+    void testAbortRestoreJobUsesQuotedIdentifiers() throws Exception
+    {
+        when(sidecarClient.abortRestoreJob(eq(QUOTED_KEYSPACE), eq(QUOTED_TABLE), eq(JOB_ID)))
+            .thenReturn(CompletableFuture.completedFuture(null));
+
+        api.abortRestoreJob();
+
+        verify(sidecarClient).abortRestoreJob(eq(QUOTED_KEYSPACE), eq(QUOTED_TABLE), eq(JOB_ID));
+    }
+
+    @Test
+    void testUnquotedIdentifiersPassedAsIsWhenFlagNotSet() throws Exception
+    {
+        when(jobInfo.qualifiedTableName()).thenReturn(new QualifiedTableName("mykeyspace", "mytable", false));
+        when(sidecarClient.createRestoreJob(eq("mykeyspace"), eq("mytable"), any()))
+            .thenReturn(CompletableFuture.completedFuture(null));
+
+        api.createRestoreJob(mock(CreateRestoreJobRequestPayload.class));
+
+        verify(sidecarClient).createRestoreJob(eq("mykeyspace"), eq("mytable"), any());
+    }
+}