CASSSIDECAR-443: Support column types not parseable by Java 3.x driver

integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/CassandraNodeOperationsIntegrationTest.java

@@ -76,7 +76,7 @@ void testNodeDrainOperationSuccess()
         trustedClient().put(serverWrapper.serverPort, "localhost", ApiEndpointsV1.NODE_DRAIN_ROUTE)
                        .send());
 
-        assertThat(drainResponse.statusCode()).isEqualTo(OK.code());
+        assertThat(drainResponse.statusCode()).isIn(OK.code(), ACCEPTED.code());
 
         JsonObject responseBody = drainResponse.bodyAsJsonObject();
         assertThat(responseBody).isNotNull();

integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/CassandraVectorSchemaRouteIntegrationTest.java

@@ -0,0 +1,70 @@
+/*
+ * 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.sidecar.routes;
+
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import io.vertx.core.http.HttpResponseExpectation;
+import org.apache.cassandra.sidecar.common.response.SchemaResponse;
+import org.apache.cassandra.sidecar.testing.QualifiedName;
+import org.apache.cassandra.sidecar.testing.SharedClusterSidecarIntegrationTestBase;
+import org.apache.cassandra.sidecar.utils.SimpleCassandraVersion;
+
+import static org.apache.cassandra.testing.utils.AssertionUtils.getBlocking;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assumptions.assumeThat;
+
+class CassandraVectorSchemaRouteIntegrationTest extends SharedClusterSidecarIntegrationTestBase
+{
+    protected static final int MIN_VERSION_WITH_VECTOR = 5;
+
+    @Override
+    protected void initializeSchemaForTest()
+    {
+        createTestKeyspace("test_keyspace", Map.of("replication_factor", 1));
+        createTestTable(new QualifiedName("test_keyspace", "int_table"),
+                        "CREATE TABLE IF NOT EXISTS %s (a int, b int, PRIMARY KEY (a))");
+        createTestTable(new QualifiedName("test_keyspace", "vector_table"),
+                        "CREATE TABLE IF NOT EXISTS %s (a int, b vector<float, 3>, PRIMARY KEY (a))");
+    }
+
+    @Override
+    protected void beforeClusterProvisioning()
+    {
+        assumeThat(SimpleCassandraVersion.create(testVersion.version()).major)
+        .as("Vector type is supported since Cassandra 5.0")
+        .isGreaterThanOrEqualTo(MIN_VERSION_WITH_VECTOR);
+    }
+
+    @Test
+    void testSchemaHandlerWithVectorTable()
+    {
+        String testRoute = "/api/v1/schema/keyspaces/test_keyspace";
+        SchemaResponse response = getBlocking(trustedClient()
+                                              .get(serverWrapper.serverPort, "localhost", testRoute)
+                                              .send()
+                                              .expecting(HttpResponseExpectation.SC_OK))
+                                  .bodyAsJson(SchemaResponse.class);
+        assertThat(response).isNotNull();
+        assertThat(response.keyspace()).isEqualTo("test_keyspace");
+        assertThat(response.schema()).contains("vector_table");
+    }
+}

server-common/src/main/java/org/apache/cassandra/sidecar/common/server/data/QualifiedTableName.java

@@ -117,6 +117,27 @@ public String maybeQuotedTableName()
         return table;
     }
 
+    @Override
+    public boolean equals(Object o)
+    {
+        if (this == o)
+        {
+            return true;
+        }
+        if (!(o instanceof QualifiedTableName))
+        {
+            return false;
+        }
+        QualifiedTableName that = (QualifiedTableName) o;
+        return Objects.equals(keyspace, that.keyspace) && Objects.equals(table, that.table);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Objects.hash(keyspace, table);
+    }
+
     /**
      * {@inheritDoc}
      */

server/src/main/java/org/apache/cassandra/bridge/CassandraBridgeFactory.java

@@ -30,7 +30,6 @@
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
-
 import jakarta.inject.Singleton;
 import org.jetbrains.annotations.NotNull;
 

server/src/main/java/org/apache/cassandra/sidecar/config/DriverConfiguration.java

@@ -21,6 +21,8 @@
 import java.net.InetSocketAddress;
 import java.util.List;
 
+import org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
+
 /**
  * The driver configuration to use when connecting to Cassandra
  */
@@ -61,4 +63,10 @@ public interface DriverConfiguration
      * Cassandra instance.
      */
     SslConfiguration sslConfiguration();
+
+    /**
+     * @return Refresh interval of table schemas not supported by Java driver's metadata
+     * (not parseable, e.g. including vector type).
+     */
+    SecondBoundConfiguration unsupportedTableSchemaRefreshTime();
 }

server/src/main/java/org/apache/cassandra/sidecar/config/yaml/DriverConfigurationImpl.java

@@ -21,8 +21,10 @@
 import java.net.InetSocketAddress;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
 import org.apache.cassandra.sidecar.config.DriverConfiguration;
 import org.apache.cassandra.sidecar.config.SslConfiguration;
 
@@ -32,6 +34,8 @@
 public class DriverConfigurationImpl implements DriverConfiguration
 {
     private static final int DEFAULT_NUM_CONNECTIONS = 1000;
+    private static final SecondBoundConfiguration DEFAULT_UNSUPPORTED_TABLE_SCHEMA_REFRESH_TIME = new SecondBoundConfiguration(5, TimeUnit.MINUTES);
+
     @JsonProperty("contact_points")
     private final List<InetSocketAddress> contactPoints;
 
@@ -50,24 +54,29 @@ public class DriverConfigurationImpl implements DriverConfiguration
     @JsonProperty("ssl")
     private final SslConfiguration sslConfiguration;
 
+    @JsonProperty("unsupported_table_schema_refresh_time")
+    private final SecondBoundConfiguration unsupportedTableSchemaRefreshTime;
+
     public DriverConfigurationImpl()
     {
-        this(Collections.emptyList(), null, DEFAULT_NUM_CONNECTIONS, null, null, null);
+        this(Collections.emptyList(), null, DEFAULT_NUM_CONNECTIONS, null, null, null, DEFAULT_UNSUPPORTED_TABLE_SCHEMA_REFRESH_TIME);
     }
 
     public DriverConfigurationImpl(List<InetSocketAddress> contactPoints,
                                    String localDc,
                                    int numConnections,
                                    String username,
                                    String password,
-                                   SslConfiguration sslConfiguration)
+                                   SslConfiguration sslConfiguration,
+                                   SecondBoundConfiguration unsupportedTableSchemaRefreshTime)
     {
         this.contactPoints = contactPoints;
         this.localDc = localDc;
         this.numConnections = numConnections;
         this.username = username;
         this.password = password;
         this.sslConfiguration = sslConfiguration;
+        this.unsupportedTableSchemaRefreshTime = unsupportedTableSchemaRefreshTime;
     }
 
     /**
@@ -129,4 +138,14 @@ public SslConfiguration sslConfiguration()
     {
         return sslConfiguration;
     }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @JsonProperty("unsupported_table_schema_refresh_time")
+    public SecondBoundConfiguration unsupportedTableSchemaRefreshTime()
+    {
+        return unsupportedTableSchemaRefreshTime;
+    }
 }

server/src/main/java/org/apache/cassandra/sidecar/db/CQLSchemaAccessor.java

@@ -0,0 +1,98 @@
+/*
+ * 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.sidecar.db;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.exceptions.InvalidQueryException;
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.server.CQLSessionProvider;
+import org.apache.cassandra.sidecar.common.server.data.Name;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Component allowing to read CQL schema by executing {@code DESCRIBE} statement.
+ * TODO: Remove after upgrade to Java driver 4.x (CASSSIDECAR-421).
+ */
+@Singleton
+public class CQLSchemaAccessor
+{
+    private final CQLSessionProvider sessionProvider;
+
+    public CQLSchemaAccessor(CQLSessionProvider sessionProvider)
+    {
+        this.sessionProvider = sessionProvider;
+    }
+
+    @NotNull
+    public Set<Name> getKeyspaces()
+    {
+        Session session = sessionProvider.get();
+        Set<Name> keyspaces = new HashSet<>();
+        List<Row> rows = session.execute("DESCRIBE KEYSPACES").all();
+        for (Row row : rows)
+        {
+            Name keyspaceName = new Name(row.getString("keyspace_name"));
+            keyspaces.add(keyspaceName);
+        }
+        return keyspaces;
+    }
+
+    @Nullable
+    public List<String> getKeyspaceSchema(@NotNull Name keyspace)
+    {
+        Session session = sessionProvider.get();
+        String statement = String.format("DESCRIBE KEYSPACE %s", keyspace.maybeQuotedName());
+        return describe(session, statement);
+    }
+
+    @Nullable
+    public List<String> getTableSchema(@NotNull Name keyspace, @NotNull Name table)
+    {
+        Session session = sessionProvider.get();
+        String statement = String.format("DESCRIBE TABLE %s.%s", keyspace.maybeQuotedName(), table.maybeQuotedName());
+        return describe(session, statement);
+    }
+
+    private List<String> describe(Session session, String describeStatement)
+    {
+        try
+        {
+            List<Row> rows = session.execute(describeStatement).all();
+            List<String> result = new ArrayList<>(rows.size());
+            for (Row row : rows)
+            {
+                String createStatement = row.getString("create_statement");
+                result.add(createStatement);
+            }
+            return result;
+        }
+        catch (InvalidQueryException e)
+        {
+            // keyspace or table not found
+            return null;
+        }
+    }
+}