This is an automated email from the ASF dual-hosted git repository.
sarankk pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-sidecar.git
The following commit(s) were added to refs/heads/trunk by this push:
new 99476afc4 CASSSIDECAR-505: Use describe statement to fetch schema
instead of Java driver (#390)
99476afc4 is described below
commit 99476afc4dd3bec0f20d264febbed00e5b155f27
Author: Saranya Krishnakumar <[email protected]>
AuthorDate: Mon Oct 26 15:16:03 2026 -0700
CASSSIDECAR-505: Use describe statement to fetch schema instead of Java
driver (#390)
Patch by Saranya Krishnakumar; Reviewed by Mansi Khara, Jyothsna Konisa,
Yifan Cai for CASSSIDECAR-505
---
CHANGES.txt | 1 +
docs/src/user.adoc | 2 +
.../CassandraSchemaRouteIntegrationTest.java | 55 +++
.../sidecar/config/DriverConfiguration.java | 10 +-
.../config/yaml/DriverConfigurationImpl.java | 40 ++-
.../sidecar/db/DriverUnsupportedSchemaCache.java | 293 ----------------
.../apache/cassandra/sidecar/db/SchemaCache.java | 379 +++++++++++++++++++++
.../sidecar/handlers/KeyspaceSchemaHandler.java | 120 +++----
.../cassandra/sidecar/handlers/SchemaHandler.java | 8 +-
.../sstableuploads/SSTableUploadHandler.java | 12 +-
.../validations/ValidateTableExistenceHandler.java | 14 +-
.../sidecar/modules/CassandraOperationsModule.java | 12 +-
.../cassandra/sidecar/modules/CdcModule.java | 6 +-
.../sidecar/modules/ConfigurationModule.java | 14 +-
.../modules/multibindings/PeriodicTaskMapKeys.java | 2 +-
.../tasks/CassandraClusterSchemaMonitor.java | 11 +-
.../sidecar/testing/IntegrationTestModule.java | 2 +-
.../org/apache/cassandra/sidecar/TestModule.java | 9 +-
.../sidecar/db/CQLSchemaAccessorTest.java | 58 +++-
.../db/DriverUnsupportedSchemaCacheTest.java | 128 -------
.../cassandra/sidecar/db/SchemaCacheTest.java | 361 ++++++++++++++++++++
.../sidecar/handlers/SchemaHandlerTest.java | 57 +++-
.../tasks/CassandraClusterSchemaMonitorTest.java | 23 +-
23 files changed, 1051 insertions(+), 566 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index bebe79cd7..9987d8b5f 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Use describe statement to fetch schema instead of Java driver
(CASSSIDECAR-505)
* Make the CDC CassandraBridge injectable (CASSSIDECAR-511)
* Introduce server-side configuration to enforce allowed credential modes for
restore jobs (CASSSIDECAR-501)
* Sanitize/Nullify Cloud Credentials on RestoreJob Completion
(CASSSIDECAR-500)
diff --git a/docs/src/user.adoc b/docs/src/user.adoc
index f90f6d0e7..50b095631 100644
--- a/docs/src/user.adoc
+++ b/docs/src/user.adoc
@@ -353,6 +353,8 @@ The `driver_parameters` section of the `sidecar.yaml` file
is used to configure
* `ssl`: This defines how the Cassandra driver can connect to Cassandra over
SSL. See the above <<ssl, SSL/TLS Configuration>> section for more information.
* `num_connections`: This defines the number of connections that the Cassandra
driver will open to each Cassandra instance. By default, 2.
* `local_dc`: This configures the local data center of the Cassandra driver.
Ensure that this is the same value as the data center of the Cassandra nodes
which this Cassandra Sidecar manages.
+* `schema_refresh_time`: This defines how frequently Cassandra Sidecar
refreshes its cache of the schema read from Cassandra. Defaults to 5 minutes
(`5m`).
+* `unsupported_table_schema_refresh_time`: Deprecated in favor of
`schema_refresh_time`, which falls back to this property when it is the only
one configured.
[[healthcheck]]
### healthcheck
diff --git
a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/CassandraSchemaRouteIntegrationTest.java
b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/CassandraSchemaRouteIntegrationTest.java
index 56cb507a3..af2082417 100644
---
a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/CassandraSchemaRouteIntegrationTest.java
+++
b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/CassandraSchemaRouteIntegrationTest.java
@@ -18,10 +18,15 @@
package org.apache.cassandra.sidecar.routes;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.Test;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
import io.vertx.core.http.HttpResponseExpectation;
import org.apache.cassandra.sidecar.common.response.SchemaResponse;
import
org.apache.cassandra.sidecar.testing.SharedClusterSidecarIntegrationTestBase;
@@ -104,4 +109,54 @@ class CassandraSchemaRouteIntegrationTest extends
SharedClusterSidecarIntegratio
assertThat(response.keyspace()).isEqualTo("keyspace");
assertThat(response.schema()).isNotNull();
}
+
+ @Test
+ void testSchemaResponseForEachKeyspace()
+ {
+ for (String keyspace : List.of("testkeyspace", "\"Cycling\"",
"\"keyspace\""))
+ {
+ String testRoute = "/api/v1/schema/keyspaces/" + keyspace;
+ SchemaResponse response =
getBlocking(trustedClient().get(serverWrapper.serverPort, "localhost",
testRoute)
+ .send()
+
.expecting(HttpResponseExpectation.SC_OK))
+ .bodyAsJson(SchemaResponse.class);
+ assertThat(response.schema())
+ .describedAs("Schema of %s must be the schema rendered by
Cassandra", keyspace)
+ .isEqualTo(convertToUpperCase(describeKeyspace(keyspace)));
+ }
+ }
+
+ @Test
+ void testAllKeyspacesPresentInFullSchema()
+ {
+ String testRoute = "/api/v1/schema/keyspaces";
+ SchemaResponse response =
getBlocking(trustedClient().get(serverWrapper.serverPort, "localhost",
testRoute)
+ .send()
+
.expecting(HttpResponseExpectation.SC_OK))
+ .bodyAsJson(SchemaResponse.class);
+
assertThat(response.schema()).contains(describeKeyspace("testkeyspace"));
+
assertThat(response.schema()).contains(describeKeyspace("\"Cycling\""));
+
assertThat(response.schema()).contains(describeKeyspace("\"keyspace\""));
+ assertThat(response.schema()).contains("CREATE KEYSPACE
system_traces");
+ }
+
+ String describeKeyspace(String maybeQuotedKeyspace)
+ {
+ try (Cluster driverCluster = createDriverCluster(cluster.delegate());
+ Session session = driverCluster.connect())
+ {
+ List<String> statements = new ArrayList<>();
+ for (Row row : session.execute("DESCRIBE KEYSPACE " +
maybeQuotedKeyspace).all())
+ {
+ statements.add(row.getString("create_statement"));
+ }
+ assertThat(statements).isNotEmpty();
+ return String.join("\n\n", statements);
+ }
+ }
+
+ String convertToUpperCase(String keyspaceSchema)
+ {
+ return keyspaceSchema.replaceFirst("WITH replication", "WITH
REPLICATION");
+ }
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/config/DriverConfiguration.java
b/server/src/main/java/org/apache/cassandra/sidecar/config/DriverConfiguration.java
index 0ffea4cde..3097f5411 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/config/DriverConfiguration.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/config/DriverConfiguration.java
@@ -76,8 +76,16 @@ public interface DriverConfiguration
SslConfiguration sslConfiguration();
/**
+ * @deprecated use {@link #schemaRefreshTime()} instead
* @return Refresh interval of table schemas not supported by Java
driver's metadata
- * (not parseable, e.g. including vector type).
+ * (not parseable, e.g. including vector type), {@code null} when it is
not configured.
*/
+ @Deprecated
SecondBoundConfiguration unsupportedTableSchemaRefreshTime();
+
+ /**
+ * @return Refresh interval of schema cache. When it is not configured,
the deprecated
+ * {@link #unsupportedTableSchemaRefreshTime()} is used instead, if it is
configured.
+ */
+ SecondBoundConfiguration schemaRefreshTime();
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/DriverConfigurationImpl.java
b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/DriverConfigurationImpl.java
index fdd1f0dd2..77fe1bbe6 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/DriverConfigurationImpl.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/DriverConfigurationImpl.java
@@ -35,7 +35,7 @@ import org.apache.cassandra.sidecar.config.SslConfiguration;
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);
+ private static final SecondBoundConfiguration DEFAULT_SCHEMA_REFRESH_TIME
= new SecondBoundConfiguration(5, TimeUnit.MINUTES);
@JsonProperty("contact_points")
private final List<InetSocketAddress> contactPoints;
@@ -58,9 +58,13 @@ public class DriverConfigurationImpl implements
DriverConfiguration
@JsonProperty("ssl")
private final SslConfiguration sslConfiguration;
+ @Deprecated
@JsonProperty("unsupported_table_schema_refresh_time")
private final SecondBoundConfiguration unsupportedTableSchemaRefreshTime;
+ @JsonProperty("schema_refresh_time")
+ private final SecondBoundConfiguration schemaRefreshTime;
+
public DriverConfigurationImpl()
{
this(builder());
@@ -76,6 +80,7 @@ public class DriverConfigurationImpl implements
DriverConfiguration
authProvider = builder.authProvider;
sslConfiguration = builder.sslConfiguration;
unsupportedTableSchemaRefreshTime =
builder.unsupportedTableSchemaRefreshTime;
+ schemaRefreshTime = builder.schemaRefreshTime;
}
/**
@@ -153,6 +158,7 @@ public class DriverConfigurationImpl implements
DriverConfiguration
/**
* {@inheritDoc}
*/
+ @Deprecated
@Override
@JsonProperty("unsupported_table_schema_refresh_time")
public SecondBoundConfiguration unsupportedTableSchemaRefreshTime()
@@ -160,6 +166,21 @@ public class DriverConfigurationImpl implements
DriverConfiguration
return unsupportedTableSchemaRefreshTime;
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @JsonProperty("schema_refresh_time")
+ public SecondBoundConfiguration schemaRefreshTime()
+ {
+ if (schemaRefreshTime != null)
+ {
+ return schemaRefreshTime;
+ }
+ return unsupportedTableSchemaRefreshTime != null ?
unsupportedTableSchemaRefreshTime
+ :
DEFAULT_SCHEMA_REFRESH_TIME;
+ }
+
public static Builder builder()
{
return new Builder();
@@ -177,7 +198,8 @@ public class DriverConfigurationImpl implements
DriverConfiguration
private String password;
private ParameterizedClassConfiguration authProvider;
private SslConfiguration sslConfiguration;
- private SecondBoundConfiguration unsupportedTableSchemaRefreshTime =
DEFAULT_UNSUPPORTED_TABLE_SCHEMA_REFRESH_TIME;
+ private SecondBoundConfiguration unsupportedTableSchemaRefreshTime;
+ private SecondBoundConfiguration schemaRefreshTime;
private Builder()
{
@@ -291,12 +313,26 @@ public class DriverConfigurationImpl implements
DriverConfiguration
*
* @param unsupportedTableSchemaRefreshTime the {@code
unsupportedTableSchemaRefreshTime} to set
* @return a reference to this Builder
+ * @deprecated use {@link
#schemaRefreshTime(SecondBoundConfiguration)} instead
*/
+ @Deprecated
public Builder
unsupportedTableSchemaRefreshTime(SecondBoundConfiguration
unsupportedTableSchemaRefreshTime)
{
return update(b -> b.unsupportedTableSchemaRefreshTime =
unsupportedTableSchemaRefreshTime);
}
+ /**
+ * Sets the {@code schemaRefreshTime} and returns a reference to this
Builder enabling
+ * method chaining.
+ *
+ * @param schemaRefreshTime the {@code schemaRefreshTime} to set
+ * @return a reference to this Builder
+ */
+ public Builder schemaRefreshTime(SecondBoundConfiguration
schemaRefreshTime)
+ {
+ return update(b -> b.schemaRefreshTime = schemaRefreshTime);
+ }
+
/**
* Returns a {@code DriverConfigurationImpl} built from the parameters
previously set.
*
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/db/DriverUnsupportedSchemaCache.java
b/server/src/main/java/org/apache/cassandra/sidecar/db/DriverUnsupportedSchemaCache.java
deleted file mode 100644
index 2250147ee..000000000
---
a/server/src/main/java/org/apache/cassandra/sidecar/db/DriverUnsupportedSchemaCache.java
+++ /dev/null
@@ -1,293 +0,0 @@
-/*
- * 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.Collections;
-import java.util.Comparator;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.SortedMap;
-import java.util.TreeMap;
-import java.util.function.Predicate;
-import java.util.stream.Collectors;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.KeyspaceMetadata;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.TableMetadata;
-import com.google.inject.Singleton;
-import io.vertx.core.Promise;
-import org.apache.cassandra.sidecar.common.server.CQLSessionProvider;
-import org.apache.cassandra.sidecar.common.server.data.Name;
-import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
-import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
-import org.apache.cassandra.sidecar.common.utils.StringUtils;
-import org.apache.cassandra.sidecar.config.SidecarConfiguration;
-import org.apache.cassandra.sidecar.exceptions.CassandraUnavailableException;
-import org.apache.cassandra.sidecar.tasks.PeriodicTask;
-import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
-import org.jetbrains.annotations.VisibleForTesting;
-
-/**
- * The {@link DriverUnsupportedSchemaCache} class maintains cache of CQL
schema for tables whose definition is not
- * supported by driver natively. Java driver 3.x does not return {@link
TableMetadata} for tables which schema
- * could not be parsed. Since it does not currently support vector type, any
table containing vector would not
- * be visible. Cache is refreshed for the first time as soon as CQL connection
is established or upon first lookup.
- * Later, it is periodically refreshed according to configured schedule.
- * TODO: Remove after upgrade to Java driver 4.x (CASSSIDECAR-421).
- */
-@Singleton
-public class DriverUnsupportedSchemaCache implements PeriodicTask
-{
- private static final Logger LOGGER =
LoggerFactory.getLogger(DriverUnsupportedSchemaCache.class);
- private static final String STATEMENT_DELIMITER = "\n\n";
-
- private final SidecarConfiguration sidecarConfiguration;
- private final CQLSessionProvider sessionProvider;
- private final CQLSchemaAccessor schemaAccessor;
-
- // cache contains only schemas unparseable by Java driver
- // during every cache refresh, reference is being reassigned to achieve
atomic swap
- private volatile SortedMap<QualifiedTableName, String> schemaCache;
- private volatile boolean initialized; // flag indicating whether cache has
been populated at least once
-
- private PreparedStatement tableListStatement;
-
- public DriverUnsupportedSchemaCache(SidecarConfiguration
sidecarConfiguration,
- CQLSessionProvider sessionProvider)
- {
- this.sidecarConfiguration = sidecarConfiguration;
- this.sessionProvider = sessionProvider;
- this.schemaAccessor = new CQLSchemaAccessor(sessionProvider);
- this.schemaCache = createCache();
- this.initialized = false;
- }
-
- /**
- * @return Schema for all tables across all keyspaces (only those not
supported by Java driver).
- */
- @NotNull
- public String getFullSchema()
- {
- // we cannot know whether schema was updated after last
- // periodic refresh, so results may be stale
- refreshIfUninitialized();
- return getUnsupportedSchema(table -> true);
- }
-
- /**
- * @return Schema for all tables within given keyspaces (only those not
supported by Java driver).
- */
- @NotNull
- public String getKeyspaceSchema(@NotNull Name keyspace)
- {
- // we cannot know whether schema was updated after last
- // periodic refresh, so results may be stale
- refreshIfUninitialized();
- return getUnsupportedSchema(table ->
keyspace.equals(table.getKeyspace()));
- }
-
- /**
- * @return Schema for table not supported by Java driver's metadata,
{@code null} otherwise.
- */
- @Nullable
- public String getTableSchema(@NotNull Name keyspace, @NotNull Name table)
- {
- return getTableSchema(keyspace, table, true);
- }
-
- /**
- * @param allowRefresh Flag indicating whether lookup of table schema via
CQL query
- * is allowed, when data not found in cache.
- * @return Schema for table not supported by Java driver's metadata,
{@code null} otherwise.
- */
- @Nullable
- public String getTableSchema(@NotNull Name keyspace, @NotNull Name table,
boolean allowRefresh)
- {
- refreshIfUninitialized();
- QualifiedTableName name = new QualifiedTableName(keyspace, table);
- String schema = schemaCache.get(name);
- if (schema == null && allowRefresh)
- {
- // proactively fetch table schema, not to provide
- // false-negative when table is not cached yet
- // attention, method can block
- schema = populateSchemaCache(schemaCache, name);
- }
- return schema;
- }
-
- @Override
- public DurationSpec delay()
- {
- return
sidecarConfiguration.driverConfiguration().unsupportedTableSchemaRefreshTime();
- }
-
- @Override
- public void execute(Promise<Void> promise)
- {
- try
- {
- refresh(false);
- promise.tryComplete();
- }
- catch (Throwable t)
- {
- promise.fail(t);
- }
- }
-
- private String getUnsupportedSchema(Predicate<QualifiedTableName>
condition)
- {
- StringBuilder result = new StringBuilder();
- for (Map.Entry<QualifiedTableName, String> entry :
schemaCache.entrySet())
- {
- if (condition.test(entry.getKey()))
- {
- result.append(entry.getValue()).append(STATEMENT_DELIMITER);
- }
- }
- return result.toString().trim();
- }
-
- private void refreshIfUninitialized()
- {
- if (!initialized)
- {
- refresh(true);
- }
- }
-
- public synchronized void refresh(boolean initializeOnly)
- {
- if (initialized && initializeOnly)
- {
- // cache has been already initialized, early exit
- return;
- }
- try
- {
- Session session = sessionProvider.get();
- prepareStatements(session);
-
- Set<QualifiedTableName> tables = queryAllTables(session);
- Set<QualifiedTableName> driverKnownTables =
driverKnownTables(session);
-
- tables.removeAll(driverKnownTables);
-
- SortedMap<QualifiedTableName, String> newCache = createCache();
- if (!tables.isEmpty())
- {
- LOGGER.debug("Tables not supported by Java driver metadata:
{}", tables);
- tables.forEach(table -> populateSchemaCache(newCache, table));
- }
- // replacing cache, because some tables might have been removed in
the meanwhile
- schemaCache = newCache;
-
- initialized = true;
- }
- catch (CassandraUnavailableException ignored)
- {
- LOGGER.debug("Not yet connected to Cassandra cluster");
- }
- }
-
- private String populateSchemaCache(Map<QualifiedTableName, String> cache,
QualifiedTableName table)
- {
- Name keyspaceName = table.getKeyspace();
- Name tableName = table.table();
- if (keyspaceName == null || tableName == null)
- {
- throw new IllegalArgumentException("Invalid table name: " + table);
- }
- List<String> cqlSchema = schemaAccessor.getTableSchema(keyspaceName,
tableName);
- if (cqlSchema != null)
- {
- String schema = String.join(STATEMENT_DELIMITER, cqlSchema);
- cache.put(table, schema);
- return schema;
- }
- return null;
- }
-
- private Set<QualifiedTableName> queryAllTables(Session session)
- {
- List<Row> rows = session.execute(tableListStatement.bind()).all();
- return rows.stream()
- .map(r -> new
QualifiedTableName(r.getString("keyspace_name"),
- r.getString("table_name")))
- .collect(Collectors.toSet());
- }
-
- private Set<QualifiedTableName> driverKnownTables(Session session)
- {
- Set<QualifiedTableName> result = new HashSet<>();
- Cluster cluster = session.getCluster();
- for (KeyspaceMetadata keyspace : cluster.getMetadata().getKeyspaces())
- {
- for (TableMetadata table : keyspace.getTables())
- {
- result.add(new QualifiedTableName(keyspace.getName(),
table.getName()));
- }
- }
- return result;
- }
-
- private void prepareStatements(Session session)
- {
- if (tableListStatement == null)
- {
- tableListStatement = session.prepare("SELECT keyspace_name,
table_name FROM system_schema.tables");
- }
- }
-
- private SortedMap<QualifiedTableName, String> createCache()
- {
- // use sorted map for repeatable results when retrieving full schema
- // synchronized, because different threads may indirectly add items to
the map using getTableSchema() method
- return Collections.synchronizedSortedMap(new
TreeMap<>(Comparator.comparing(QualifiedTableName::toString)));
- }
-
- @VisibleForTesting
- void setInitialized(boolean initialized)
- {
- this.initialized = initialized;
- }
-
- public static String concatSchemas(String ... schemas)
- {
- StringBuilder result = new StringBuilder();
- for (String schema : schemas)
- {
- if (StringUtils.isNotEmpty(schema))
- {
- result.append(schema.trim()).append(STATEMENT_DELIMITER);
- }
- }
- return result.toString().trim();
- }
-}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/db/SchemaCache.java
b/server/src/main/java/org/apache/cassandra/sidecar/db/SchemaCache.java
new file mode 100644
index 000000000..d9395af6c
--- /dev/null
+++ b/server/src/main/java/org/apache/cassandra/sidecar/db/SchemaCache.java
@@ -0,0 +1,379 @@
+/*
+ * 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.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TableMetadata;
+import com.google.inject.Singleton;
+import io.vertx.core.Promise;
+import org.apache.cassandra.sidecar.common.server.CQLSessionProvider;
+import org.apache.cassandra.sidecar.common.server.data.Name;
+import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
+import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
+import org.apache.cassandra.sidecar.common.utils.StringUtils;
+import org.apache.cassandra.sidecar.config.SidecarConfiguration;
+import org.apache.cassandra.sidecar.exceptions.CassandraUnavailableException;
+import org.apache.cassandra.sidecar.tasks.PeriodicTask;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.VisibleForTesting;
+
+/**
+ * The {@link SchemaCache} class maintains cache of keyspace, table schemas
read through {@code DESCRIBE} call. Schema
+ * provided by Java driver could be partial as it drops objects it cannot
parse. It also stores driver unparseable
+ * table level schemas, used for table existence checks.
+ * <p>
+ * Cache is regularly refreshed, but schema changes applied between refreshes
may not be immediately visible
+ */
+@Singleton
+public class SchemaCache implements PeriodicTask
+{
+ private static final Logger LOGGER =
LoggerFactory.getLogger(SchemaCache.class);
+ private static final String STATEMENT_DELIMITER = "\n\n";
+
+ private final SidecarConfiguration sidecarConfiguration;
+ private final CQLSessionProvider sessionProvider;
+ private final CQLSchemaAccessor schemaAccessor;
+
+ // Caches table schemas those that could not be parsed by Java driver
+ private volatile ConcurrentNavigableMap<QualifiedTableName, String>
unsupportedTableSchemaCache;
+ private volatile ConcurrentNavigableMap<String, String>
keyspaceSchemaCache;
+ // CQL schema of all keyspaces. It is invalidated whenever the schema of a
keyspace is added to keyspaceSchemaCache.
+ private final AtomicReference<FullSchema> fullSchemaCache = new
AtomicReference<>(new FullSchema(null));
+ private volatile boolean initialized;
+
+ private PreparedStatement tableListStatement;
+
+ public SchemaCache(SidecarConfiguration sidecarConfiguration,
CQLSessionProvider sessionProvider)
+ {
+ this.sidecarConfiguration = sidecarConfiguration;
+ this.sessionProvider = sessionProvider;
+ this.schemaAccessor = new CQLSchemaAccessor(sessionProvider);
+ this.unsupportedTableSchemaCache = createTableSchemaCache();
+ this.keyspaceSchemaCache = createKeyspaceSchemaCache();
+ this.initialized = false;
+ }
+
+ /**
+ * @return CQL schema of all keyspaces. Keyspaces created after the last
cache refresh are not included, unless
+ * their schema has been looked up individually in the meanwhile, so
results may be stale.
+ */
+ @NotNull
+ public String getFullSchema()
+ {
+ refreshIfUninitialized();
+ FullSchema cached = fullSchemaCache.get();
+ if (cached.schema != null)
+ {
+ return cached.schema;
+ }
+ String fullSchema = assembleFullSchema(keyspaceSchemaCache);
+ // Cache only if nothing was added while assembling, the result would
miss it otherwise.
+ // The assembled schema is returned either way
+ fullSchemaCache.compareAndSet(cached, new FullSchema(fullSchema));
+ return fullSchema;
+ }
+
+ /**
+ * @return CQL schema of the keyspace, {@code null} when the keyspace does
not exist. If the keyspace was updated
+ * after last refresh, results maybe stale until next refresh.
+ */
+ @Nullable
+ public String getKeyspaceSchema(@NotNull String keyspace)
+ {
+ refreshIfUninitialized();
+ ConcurrentNavigableMap<String, String> cache = keyspaceSchemaCache;
+ String schema = cache.get(keyspace);
+ if (schema == null)
+ {
+ schema = populateKeyspaceSchemaCache(cache, keyspace);
+ if (schema != null)
+ {
+ // the keyspace was added to the live cache, so the cached
full schema is outdated
+ setFullSchema(null);
+ }
+ }
+ return schema;
+ }
+
+ /**
+ * @return Schema for all tables across all keyspaces (only those not
supported by Java driver). we cannot know
+ * whether schema was updated after last periodic refresh, so results may
be stale.
+ */
+ @NotNull
+ public String getUnsupportedTableSchema()
+ {
+ refreshIfUninitialized();
+ StringBuilder result = new StringBuilder();
+ for (Map.Entry<QualifiedTableName, String> entry :
unsupportedTableSchemaCache.entrySet())
+ {
+ result.append(entry.getValue()).append(STATEMENT_DELIMITER);
+ }
+ return result.toString().trim();
+ }
+
+ /**
+ * @return Schema for table not supported by Java driver's metadata,
{@code null} otherwise.
+ */
+ @Nullable
+ public String getUnsupportedTableSchema(@NotNull Name keyspace, @NotNull
Name table)
+ {
+ return getUnsupportedTableSchema(keyspace, table, true);
+ }
+
+ /**
+ * @param allowRefresh Flag indicating whether lookup of table schema via
CQL query
+ * is allowed, when data not found in cache.
+ * @return Schema for table not supported by Java driver's metadata,
{@code null} otherwise.
+ */
+ @Nullable
+ public String getUnsupportedTableSchema(@NotNull Name keyspace, @NotNull
Name table, boolean allowRefresh)
+ {
+ refreshIfUninitialized();
+ // the cache is keyed by the names as stored by Cassandra
+ QualifiedTableName name = new
QualifiedTableName(foldIfUnquoted(keyspace), foldIfUnquoted(table));
+ String schema = unsupportedTableSchemaCache.get(name);
+ if (schema == null && allowRefresh)
+ {
+ // proactively fetch table schema, not to provide
+ // false-negative when table is not cached yet
+ // attention, method can block
+ schema = populateTableSchemaCache(unsupportedTableSchemaCache,
name);
+ }
+ return schema;
+ }
+
+ @Override
+ public DurationSpec delay()
+ {
+ return sidecarConfiguration.driverConfiguration().schemaRefreshTime();
+ }
+
+ @Override
+ public void execute(Promise<Void> promise)
+ {
+ try
+ {
+ refresh(false);
+ promise.tryComplete();
+ }
+ catch (Throwable t)
+ {
+ promise.fail(t);
+ }
+ }
+
+ private void refreshIfUninitialized()
+ {
+ if (!initialized)
+ {
+ refresh(true);
+ }
+ }
+
+ public synchronized void refresh(boolean initializeOnly)
+ {
+ if (initialized && initializeOnly)
+ {
+ return;
+ }
+ try
+ {
+ Session session = sessionProvider.get();
+ prepareStatements(session);
+ Set<QualifiedTableName> tables = queryAllTables(session);
+ Set<QualifiedTableName> driverKnownTables =
driverKnownTables(session);
+ tables.removeAll(driverKnownTables);
+
+ ConcurrentNavigableMap<QualifiedTableName, String>
newTableSchemaCache = createTableSchemaCache();
+ if (!tables.isEmpty())
+ {
+ LOGGER.debug("Tables not supported by Java driver metadata:
{}", tables);
+ tables.forEach(table ->
populateTableSchemaCache(newTableSchemaCache, table));
+ }
+ ConcurrentNavigableMap<String, String> newKeyspaceSchemaCache =
createKeyspaceSchemaCache();
+ for (Name keyspace : schemaAccessor.getKeyspaces())
+ {
+ populateKeyspaceSchemaCache(newKeyspaceSchemaCache,
keyspace.name());
+ }
+ unsupportedTableSchemaCache = newTableSchemaCache;
+ keyspaceSchemaCache = newKeyspaceSchemaCache;
+ setFullSchema(assembleFullSchema(newKeyspaceSchemaCache));
+ initialized = true;
+ }
+ catch (CassandraUnavailableException ignored)
+ {
+ LOGGER.debug("Error refreshing SchemaCache, not yet connected to
Cassandra");
+ }
+ }
+
+ @Nullable
+ private String populateKeyspaceSchemaCache(Map<String, String> cache,
String keyspace)
+ {
+ List<String> cqlSchema = schemaAccessor.getKeyspaceSchema(new
Name(Metadata.quoteIfNecessary(keyspace)));
+ if (cqlSchema == null || cqlSchema.isEmpty())
+ {
+ return null;
+ }
+ String schema = String.join(STATEMENT_DELIMITER, cqlSchema);
+ cache.put(keyspace, schema);
+ return schema;
+ }
+
+ private void setFullSchema(@Nullable String fullSchema)
+ {
+ fullSchemaCache.set(new FullSchema(fullSchema));
+ }
+
+ private static String assembleFullSchema(Map<String, String>
keyspaceSchemas)
+ {
+ StringBuilder result = new StringBuilder();
+ for (String keyspaceSchema : keyspaceSchemas.values())
+ {
+ result.append(keyspaceSchema).append(STATEMENT_DELIMITER);
+ }
+ return result.toString().trim();
+ }
+
+ private String populateTableSchemaCache(Map<QualifiedTableName, String>
cache, QualifiedTableName table)
+ {
+ Name keyspaceName = table.getKeyspace();
+ Name tableName = table.table();
+ if (keyspaceName == null || tableName == null)
+ {
+ throw new IllegalArgumentException("Invalid table name: " + table);
+ }
+ // Cassandra stores the names unquoted, while DESCRIBE requires them
to be quoted when the name is case
+ // sensitive or a reserved keyword
+ List<String> cqlSchema = schemaAccessor.getTableSchema(new
Name(Metadata.quoteIfNecessary(keyspaceName.name())),
+ new
Name(Metadata.quoteIfNecessary(tableName.name())));
+ if (cqlSchema != null)
+ {
+ String schema = String.join(STATEMENT_DELIMITER, cqlSchema);
+ cache.put(table, schema);
+ return schema;
+ }
+ return null;
+ }
+
+ private Set<QualifiedTableName> queryAllTables(Session session)
+ {
+ List<Row> rows = session.execute(tableListStatement.bind()).all();
+ return rows.stream()
+ .map(r -> new
QualifiedTableName(r.getString("keyspace_name"),
+ r.getString("table_name")))
+ .collect(Collectors.toSet());
+ }
+
+ private Set<QualifiedTableName> driverKnownTables(Session session)
+ {
+ Set<QualifiedTableName> result = new HashSet<>();
+ Cluster cluster = session.getCluster();
+ for (KeyspaceMetadata keyspace : cluster.getMetadata().getKeyspaces())
+ {
+ for (TableMetadata table : keyspace.getTables())
+ {
+ result.add(new QualifiedTableName(keyspace.getName(),
table.getName()));
+ }
+ }
+ return result;
+ }
+
+ private void prepareStatements(Session session)
+ {
+ if (tableListStatement == null)
+ {
+ tableListStatement = session.prepare("SELECT keyspace_name,
table_name FROM system_schema.tables");
+ }
+ }
+
+ /**
+ * Normalizes a name obtained from a request into the form Cassandra
stores it in: CQL folds unquoted
+ * names to lowercase, while quoted names keep their case.
+ */
+ private static String foldIfUnquoted(Name name)
+ {
+ return name.isSourceQuoted() ? name.name() :
name.name().toLowerCase(Locale.ROOT);
+ }
+
+ private ConcurrentNavigableMap<QualifiedTableName, String>
createTableSchemaCache()
+ {
+ // uses ConcurrentSkipListMap to have assembled full schema or table
schemas consistent across calls
+ return new
ConcurrentSkipListMap<>(Comparator.comparing(QualifiedTableName::toString));
+ }
+
+ private ConcurrentNavigableMap<String, String> createKeyspaceSchemaCache()
+ {
+ // uses ConcurrentSkipListMap to have assembled full schema or table
schemas consistent across calls
+ return new ConcurrentSkipListMap<>();
+ }
+
+ @VisibleForTesting
+ void setInitialized(boolean initialized)
+ {
+ this.initialized = initialized;
+ }
+
+ public static String concatSchemas(String ... schemas)
+ {
+ StringBuilder result = new StringBuilder();
+ for (String schema : schemas)
+ {
+ if (StringUtils.isNotEmpty(schema))
+ {
+ result.append(schema.trim()).append(STATEMENT_DELIMITER);
+ }
+ }
+ return result.toString().trim();
+ }
+
+ /**
+ * Holder of the cached full schema. Compared by identity, so every update
is distinguishable from the previous
+ * one even when the cached value is the same
+ */
+ private static final class FullSchema
+ {
+ @Nullable
+ final String schema;
+
+ FullSchema(@Nullable String schema)
+ {
+ this.schema = schema;
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/KeyspaceSchemaHandler.java
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/KeyspaceSchemaHandler.java
index 3a43976bf..2fd3c91ec 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/KeyspaceSchemaHandler.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/KeyspaceSchemaHandler.java
@@ -25,17 +25,17 @@ import com.datastax.driver.core.Metadata;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import io.netty.handler.codec.http.HttpResponseStatus;
-import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.http.HttpServerRequest;
import io.vertx.core.net.SocketAddress;
import io.vertx.ext.auth.authorization.Authorization;
import io.vertx.ext.web.RoutingContext;
+import io.vertx.ext.web.handler.HttpException;
import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
import org.apache.cassandra.sidecar.common.response.SchemaResponse;
import org.apache.cassandra.sidecar.common.server.data.Name;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
-import org.apache.cassandra.sidecar.db.DriverUnsupportedSchemaCache;
+import org.apache.cassandra.sidecar.db.SchemaCache;
import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
import org.apache.cassandra.sidecar.utils.MetadataUtils;
@@ -49,7 +49,7 @@ import static
org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpExceptio
@Singleton
public class KeyspaceSchemaHandler extends AbstractHandler<Name> implements
AccessProtected
{
- private final DriverUnsupportedSchemaCache driverUnsupportedSchemaCache;
+ private final SchemaCache schemaCache;
/**
* Constructs a handler with the provided {@code metadataFetcher}
@@ -57,16 +57,16 @@ public class KeyspaceSchemaHandler extends
AbstractHandler<Name> implements Acce
* @param metadataFetcher the interface to retrieve metadata
* @param executorPools executor pools for blocking executions
* @param validator a validator instance to validate
Cassandra-specific input
- * @param driverUnsupportedSchemaCache cache of unparseable table schemas
by Java driver
+ * @param schemaCache cache of CQL schema read from Cassandra
*/
@Inject
protected KeyspaceSchemaHandler(InstanceMetadataFetcher metadataFetcher,
ExecutorPools executorPools,
CassandraInputValidator validator,
- DriverUnsupportedSchemaCache
driverUnsupportedSchemaCache)
+ SchemaCache schemaCache)
{
super(metadataFetcher, executorPools, validator);
- this.driverUnsupportedSchemaCache = driverUnsupportedSchemaCache;
+ this.schemaCache = schemaCache;
}
@Override
@@ -85,84 +85,68 @@ public class KeyspaceSchemaHandler extends
AbstractHandler<Name> implements Acce
SocketAddress remoteAddress,
Name keyspace)
{
- Future.all(metadata(host), unsupportedSchema(keyspace))
- .onFailure(cause -> processFailure(cause, context, host,
remoteAddress, keyspace))
- .onSuccess(future -> handleWithMetadata(context, keyspace,
future));
+ describeSchema(host, keyspace)
+ .onSuccess(context::json)
+ .onFailure(cause -> processFailure(cause, context, host,
remoteAddress, keyspace));
+ }
+
+ private Future<SchemaResponse> describeSchema(String host, Name keyspace)
+ {
+ return executorPools.service()
+ .executeBlocking(() -> {
+ if (keyspace == null)
+ {
+ String schema =
schemaCache.getFullSchema();
+ if (schema.isEmpty())
+ {
+ throw schemaNotAvailable();
+ }
+ return new SchemaResponse(schema);
+ }
+
+ String keyspaceSchema =
schemaCache.getKeyspaceSchema(resolveKeyspace(host, keyspace));
+ if (keyspaceSchema == null)
+ {
+ throw keyspaceDoesNotExist(keyspace);
+ }
+ return new SchemaResponse(keyspace.name(),
upperCaseReplication(keyspaceSchema));
+ });
}
/**
- * Handles the request with the Cassandra {@link Metadata metadata}.
- *
- * @param context the event to handle
- * @param keyspace the keyspace parsed from the request
- * @param future composite future containing result of:
- * - the metadata on the connected cluster, including
known nodes and schema definitions
- * - additional schema which could not be parsed by Java
driver
+ * {@code DESCRIBE KEYSPACE} emits the replication in lower case, whereas
the schema previously returned
+ * by this endpoint, generated by Java driver, used upper case. Keywords
are upper cased to keep the
+ * response compatible with existing clients.
*/
- private void handleWithMetadata(RoutingContext context, Name keyspace,
CompositeFuture future)
+ private String upperCaseReplication(String schema)
{
- Metadata metadata = future.resultAt(0);
- String unparseableSchema = future.resultAt(1);
- if (keyspace == null)
- {
- String fullSchema =
DriverUnsupportedSchemaCache.concatSchemas(metadata.exportSchemaAsString(),
-
unparseableSchema);
- SchemaResponse schemaResponse = new SchemaResponse(fullSchema);
- context.json(schemaResponse);
- return;
- }
+ return schema.replaceFirst("WITH replication", "WITH REPLICATION");
+ }
- // retrieve keyspace metadata
+ /**
+ * Checks keyspace exists and resolves to String stored in
+ */
+ private String resolveKeyspace(String host, Name keyspace)
+ {
+ Metadata metadata = metadataFetcher.delegate(host).metadata();
KeyspaceMetadata ksMetadata = MetadataUtils.keyspace(metadata,
keyspace);
-
if (ksMetadata == null)
{
- // set request as failed and return
- // keyspace does not exist
- String errorMessage = String.format("Keyspace '%s' does not
exist.", keyspace);
- context.fail(wrapHttpException(HttpResponseStatus.NOT_FOUND,
errorMessage));
- return;
+ throw keyspaceDoesNotExist(keyspace);
}
-
- String keyspaceSchema =
DriverUnsupportedSchemaCache.concatSchemas(ksMetadata.exportAsString(),
-
unparseableSchema);
- SchemaResponse schemaResponse = new SchemaResponse(keyspace.name(),
keyspaceSchema);
- context.json(schemaResponse);
+ return ksMetadata.getName();
}
- /**
- * Gets cluster metadata asynchronously.
- *
- * @param host the Cassandra instance host
- * @return {@link Future} containing {@link Metadata}
- */
- private Future<Metadata> metadata(String host)
+ private HttpException keyspaceDoesNotExist(Name keyspace)
{
- return executorPools.service().executeBlocking(() -> {
- // metadata can block so we need to run in a blocking thread
- return metadataFetcher.delegate(host).metadata();
- });
+ String errorMessage = String.format("Keyspace '%s' does not exist.",
keyspace.name());
+ return wrapHttpException(HttpResponseStatus.NOT_FOUND, errorMessage);
}
- /**
- * Get CQL schema not parseable by Java driver (and therefore not present
in {@link Metadata}).
- *
- * @param keyspace optional keyspace name, {@code null} includes schema
for all keyspaces
- * @return {@link Future} containing CQL schema
- */
- private Future<String> unsupportedSchema(Name keyspace)
+ private HttpException schemaNotAvailable()
{
- return executorPools.service().executeBlocking(() -> {
- // schema cache can block if it was not successfully initialized
at least once
- if (keyspace == null)
- {
- return driverUnsupportedSchemaCache.getFullSchema();
- }
- else
- {
- return
driverUnsupportedSchemaCache.getKeyspaceSchema(keyspace);
- }
- });
+ return wrapHttpException(HttpResponseStatus.SERVICE_UNAVAILABLE,
+ "Schema is not available yet, Cassandra could
not be reached.");
}
/**
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/SchemaHandler.java
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/SchemaHandler.java
index 351115b80..4ba144ea6 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/SchemaHandler.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/SchemaHandler.java
@@ -27,7 +27,7 @@ import io.vertx.ext.web.RoutingContext;
import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
import org.apache.cassandra.sidecar.common.server.data.Name;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
-import org.apache.cassandra.sidecar.db.DriverUnsupportedSchemaCache;
+import org.apache.cassandra.sidecar.db.SchemaCache;
import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
@@ -43,15 +43,15 @@ public class SchemaHandler extends KeyspaceSchemaHandler
* @param metadataFetcher the interface to retrieve metadata
* @param executorPools executor pools for blocking executions
* @param validator a validator instance to validate
Cassandra-specific input
- * @param driverUnsupportedSchemaCache cache of unparseable table schemas
by Java driver
+ * @param schemaCache cache of schemas retrieved by describe stmt call
*/
@Inject
protected SchemaHandler(InstanceMetadataFetcher metadataFetcher,
ExecutorPools executorPools,
CassandraInputValidator validator,
- DriverUnsupportedSchemaCache
driverUnsupportedSchemaCache)
+ SchemaCache schemaCache)
{
- super(metadataFetcher, executorPools, validator,
driverUnsupportedSchemaCache);
+ super(metadataFetcher, executorPools, validator, schemaCache);
}
@Override
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandler.java
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandler.java
index 3a9a55551..670b697f1 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandler.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandler.java
@@ -42,7 +42,7 @@ import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
import org.apache.cassandra.sidecar.config.SSTableUploadConfiguration;
import org.apache.cassandra.sidecar.config.ServiceConfiguration;
-import org.apache.cassandra.sidecar.db.DriverUnsupportedSchemaCache;
+import org.apache.cassandra.sidecar.db.SchemaCache;
import org.apache.cassandra.sidecar.handlers.AbstractHandler;
import org.apache.cassandra.sidecar.handlers.AccessProtected;
import org.apache.cassandra.sidecar.handlers.data.SSTableUploadRequestParam;
@@ -73,7 +73,7 @@ public class SSTableUploadHandler extends
AbstractHandler<SSTableUploadRequestPa
private final SSTableUploadsPathBuilder uploadPathBuilder;
private final ConcurrencyLimiter limiter;
private final DigestVerifierFactory digestVerifierFactory;
- private final DriverUnsupportedSchemaCache driverUnsupportedSchemaCache;
+ private final SchemaCache schemaCache;
/**
* Constructs a handler with the provided params.
@@ -86,7 +86,7 @@ public class SSTableUploadHandler extends
AbstractHandler<SSTableUploadRequestPa
* @param executorPools executor pools for blocking
executions
* @param validator a validator instance to validate
Cassandra-specific input
* @param digestVerifierFactory a factory of checksum verifiers
- * @param driverUnsupportedSchemaCache cache of unparseable table schemas
by Java driver
+ * @param schemaCache cache containing unparseable table
schemas by Java driver
*/
@Inject
protected SSTableUploadHandler(Vertx vertx,
@@ -97,7 +97,7 @@ public class SSTableUploadHandler extends
AbstractHandler<SSTableUploadRequestPa
ExecutorPools executorPools,
CassandraInputValidator validator,
DigestVerifierFactory digestVerifierFactory,
- DriverUnsupportedSchemaCache
driverUnsupportedSchemaCache)
+ SchemaCache schemaCache)
{
super(metadataFetcher, executorPools, validator);
this.fs = vertx.fileSystem();
@@ -106,7 +106,7 @@ public class SSTableUploadHandler extends
AbstractHandler<SSTableUploadRequestPa
this.uploadPathBuilder = uploadPathBuilder;
this.limiter = new
ConcurrencyLimiter(configuration::concurrentUploadsLimit);
this.digestVerifierFactory = digestVerifierFactory;
- this.driverUnsupportedSchemaCache = driverUnsupportedSchemaCache;
+ this.schemaCache = schemaCache;
}
@Override
@@ -209,7 +209,7 @@ public class SSTableUploadHandler extends
AbstractHandler<SSTableUploadRequestPa
{
TaskExecutorPool pool = executorPools.service();
return Future.all(pool.executeBlocking(() ->
metadataFetcher.delegate(host).metadata()),
- pool.executeBlocking(() ->
driverUnsupportedSchemaCache.getTableSchema(request.keyspace(),
request.table(), false)))
+ pool.executeBlocking(() ->
schemaCache.getUnsupportedTableSchema(request.keyspace(), request.table(),
false)))
.compose(compositeFuture -> {
Metadata metadata = compositeFuture.resultAt(0);
String unparseableSchema =
compositeFuture.resultAt(1);
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/validations/ValidateTableExistenceHandler.java
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/validations/ValidateTableExistenceHandler.java
index f18964edf..8b729744d 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/validations/ValidateTableExistenceHandler.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/validations/ValidateTableExistenceHandler.java
@@ -31,7 +31,7 @@ import io.vertx.ext.web.RoutingContext;
import org.apache.cassandra.sidecar.common.server.data.Name;
import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
-import org.apache.cassandra.sidecar.db.DriverUnsupportedSchemaCache;
+import org.apache.cassandra.sidecar.db.SchemaCache;
import org.apache.cassandra.sidecar.handlers.AbstractHandler;
import org.apache.cassandra.sidecar.routes.RoutingContextUtils;
import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
@@ -50,16 +50,16 @@ import static
org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpExceptio
@Singleton
public class ValidateTableExistenceHandler extends
AbstractHandler<QualifiedTableName>
{
- private final DriverUnsupportedSchemaCache driverUnsupportedSchemaCache;
+ private final SchemaCache schemaCache;
@Inject
public ValidateTableExistenceHandler(InstanceMetadataFetcher
metadataFetcher,
ExecutorPools executorPools,
CassandraInputValidator validator,
- DriverUnsupportedSchemaCache
driverUnsupportedSchemaCache)
+ SchemaCache schemaCache)
{
super(metadataFetcher, executorPools, validator);
- this.driverUnsupportedSchemaCache = driverUnsupportedSchemaCache;
+ this.schemaCache = schemaCache;
}
// It is a validator, and it does not assume values (keyspace and table)
are present.
@@ -144,9 +144,9 @@ public class ValidateTableExistenceHandler extends
AbstractHandler<QualifiedTabl
return null;
}
// schema cache can block if it was not successfully initialized
at least once
- return driverUnsupportedSchemaCache.getTableSchema(new
Name(keyspace),
- new Name(table),
- false);
+ return schemaCache.getUnsupportedTableSchema(new Name(keyspace),
+ new Name(table),
+ false);
});
}
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java
b/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java
index 49c1af468..789ab0a31 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java
@@ -322,7 +322,8 @@ public class CassandraOperationsModule extends
AbstractModule
@GET
@Path(ApiEndpointsV1.ALL_KEYSPACES_SCHEMA_ROUTE)
@Operation(summary = "Get all keyspaces schema",
- description = "Returns the schema information for all keyspaces
in the Cassandra cluster")
+ description = "Returns the schema information for all keyspaces
in the Cassandra cluster, fetches " +
+ "schema using describe schema call")
@APIResponse(description = "Schema information retrieved successfully",
responseCode = "200",
content = @Content(mediaType = "application/json",
@@ -339,7 +340,8 @@ public class CassandraOperationsModule extends
AbstractModule
@GET
@Path(ApiEndpointsV1.DEPRECATED_ALL_KEYSPACES_SCHEMA_ROUTE)
@Operation(summary = "Get all keyspaces schema (deprecated)",
- description = "Returns the schema information for all keyspaces
in the Cassandra cluster. This endpoint is deprecated.")
+ description = "Returns the schema information for all keyspaces
in the Cassandra cluster, fetches " +
+ "schema using describe schema call. This endpoint
is deprecated.")
@APIResponse(description = "Schema information retrieved successfully",
responseCode = "200",
content = @Content(mediaType = "application/json",
@@ -355,7 +357,8 @@ public class CassandraOperationsModule extends
AbstractModule
@GET
@Path(ApiEndpointsV1.KEYSPACE_SCHEMA_ROUTE)
@Operation(summary = "Get keyspace schema",
- description = "Returns the schema information for a specific
keyspace")
+ description = "Returns the schema information for a specific
keyspace, fetches schema using describe " +
+ "schema call")
@APIResponse(description = "Keyspace schema retrieved successfully",
responseCode = "200",
content = @Content(mediaType = "application/json",
@@ -372,7 +375,8 @@ public class CassandraOperationsModule extends
AbstractModule
@GET
@Path(ApiEndpointsV1.DEPRECATED_KEYSPACE_SCHEMA_ROUTE)
@Operation(summary = "Get keyspace schema (deprecated)",
- description = "Returns the schema information for a specific
keyspace. This endpoint is deprecated.")
+ description = "Returns the schema information for a specific
keyspace, fetches schema using describe " +
+ "schema call. This endpoint is deprecated.")
@APIResponse(description = "Keyspace schema retrieved successfully",
responseCode = "200",
content = @Content(mediaType = "application/json",
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/modules/CdcModule.java
b/server/src/main/java/org/apache/cassandra/sidecar/modules/CdcModule.java
index 09c33b691..043e14eeb 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/modules/CdcModule.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/modules/CdcModule.java
@@ -83,8 +83,8 @@ import
org.apache.cassandra.sidecar.coordination.TokenRingProvider;
import org.apache.cassandra.sidecar.db.CdcConfigAccessor;
import org.apache.cassandra.sidecar.db.CdcDatabaseAccessor;
import org.apache.cassandra.sidecar.db.CdcSystemViewsDatabaseAccessor;
-import org.apache.cassandra.sidecar.db.DriverUnsupportedSchemaCache;
import org.apache.cassandra.sidecar.db.KafkaConfigAccessor;
+import org.apache.cassandra.sidecar.db.SchemaCache;
import org.apache.cassandra.sidecar.db.TokenSplitConfigAccessor;
import org.apache.cassandra.sidecar.db.schema.CdcStatesSchema;
import org.apache.cassandra.sidecar.db.schema.ConfigsSchema;
@@ -158,12 +158,12 @@ public class CdcModule extends AbstractModule
@Singleton
CassandraClusterSchemaMonitor
cassandraClusterSchemaMonitorInstance(InstanceMetadataFetcher
instanceMetadataFetcher,
CdcDatabaseAccessor databaseAccessor,
-
DriverUnsupportedSchemaCache driverUnsupportedSchemaCache,
+
SchemaCache schemaCache,
SidecarConfiguration configuration,
Provider<CassandraBridge> cassandraBridgeProvider)
{
return new CassandraClusterSchemaMonitor(instanceMetadataFetcher,
databaseAccessor,
- driverUnsupportedSchemaCache,
configuration,
+ schemaCache, configuration,
cassandraBridgeProvider);
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/modules/ConfigurationModule.java
b/server/src/main/java/org/apache/cassandra/sidecar/modules/ConfigurationModule.java
index 2da63b21d..811d6d6c7 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/modules/ConfigurationModule.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/modules/ConfigurationModule.java
@@ -59,7 +59,7 @@ import org.apache.cassandra.sidecar.config.JmxConfiguration;
import org.apache.cassandra.sidecar.config.ParameterizedClassConfiguration;
import org.apache.cassandra.sidecar.config.ServiceConfiguration;
import org.apache.cassandra.sidecar.config.SidecarConfiguration;
-import org.apache.cassandra.sidecar.db.DriverUnsupportedSchemaCache;
+import org.apache.cassandra.sidecar.db.SchemaCache;
import org.apache.cassandra.sidecar.db.schema.TableSchemaFetcher;
import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
import org.apache.cassandra.sidecar.metrics.MetricRegistryFactory;
@@ -172,11 +172,11 @@ public class ConfigurationModule extends AbstractModule
@Provides
@Singleton
- DriverUnsupportedSchemaCache driverUnsupportedSchemaCache(Vertx vertx,
-
SidecarConfiguration sidecarConfiguration,
-
CQLSessionProvider cqlSessionProvider)
+ SchemaCache schemaCache(Vertx vertx,
+ SidecarConfiguration sidecarConfiguration,
+ CQLSessionProvider cqlSessionProvider)
{
- DriverUnsupportedSchemaCache schemaCache = new
DriverUnsupportedSchemaCache(sidecarConfiguration, cqlSessionProvider);
+ SchemaCache schemaCache = new SchemaCache(sidecarConfiguration,
cqlSessionProvider);
// trigger immediate cache population once Sidecar connects to
Cassandra node
EventBusUtils.onceLocalConsumer(vertx.eventBus(),
ON_CASSANDRA_CQL_READY.address(),
@@ -188,8 +188,8 @@ public class ConfigurationModule extends AbstractModule
}
@ProvidesIntoMap
- @KeyClassMapKey(PeriodicTaskMapKeys.UnsupportedSchemaCacheTaskKey.class)
- PeriodicTask
driverUnsupportedSchemaCachePeriodicTask(DriverUnsupportedSchemaCache
schemaCache)
+ @KeyClassMapKey(PeriodicTaskMapKeys.SchemaCacheTaskKey.class)
+ PeriodicTask schemaCachePeriodicTask(SchemaCache schemaCache)
{
return schemaCache;
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/modules/multibindings/PeriodicTaskMapKeys.java
b/server/src/main/java/org/apache/cassandra/sidecar/modules/multibindings/PeriodicTaskMapKeys.java
index fcf7055fa..688d740a8 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/modules/multibindings/PeriodicTaskMapKeys.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/modules/multibindings/PeriodicTaskMapKeys.java
@@ -37,5 +37,5 @@ public interface PeriodicTaskMapKeys
interface CdcPublisherTaskKey extends ClassKey {}
interface CdcConfigRefresherNotifierKey extends ClassKey {}
interface CassandraClusterSchemaTaskKey extends ClassKey {}
- interface UnsupportedSchemaCacheTaskKey extends ClassKey {}
+ interface SchemaCacheTaskKey extends ClassKey {}
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitor.java
b/server/src/main/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitor.java
index e3bd0a649..b56b1aa86 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitor.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitor.java
@@ -42,7 +42,7 @@ import
org.apache.cassandra.sidecar.common.response.NodeSettings;
import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
import org.apache.cassandra.sidecar.config.SidecarConfiguration;
import org.apache.cassandra.sidecar.db.CdcDatabaseAccessor;
-import org.apache.cassandra.sidecar.db.DriverUnsupportedSchemaCache;
+import org.apache.cassandra.sidecar.db.SchemaCache;
import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
import org.apache.cassandra.spark.data.CqlTable;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
@@ -79,7 +79,7 @@ public class CassandraClusterSchemaMonitor implements
PeriodicTask
private final AtomicReference<Set<TableIdentifier>> lastRegisteredTables =
new AtomicReference<>(Collections.emptySet());
private final ConcurrentHashMap<TableIdentifier, UUID> tableIdCache = new
ConcurrentHashMap<>();
private final CdcDatabaseAccessor databaseAccessor;
- private final DriverUnsupportedSchemaCache driverUnsupportedSchemaCache;
+ private final SchemaCache schemaCache;
private final CopyOnWriteArrayList<Runnable> schemaChangeListeners = new
CopyOnWriteArrayList<>();
private final SidecarConfiguration sidecarConfiguration;
private final InstanceMetadataFetcher instanceFetcher;
@@ -87,14 +87,14 @@ public class CassandraClusterSchemaMonitor implements
PeriodicTask
public CassandraClusterSchemaMonitor(InstanceMetadataFetcher
instanceFetcher,
CdcDatabaseAccessor databaseAccessor,
- DriverUnsupportedSchemaCache
driverUnsupportedSchemaCache,
+ SchemaCache schemaCache,
SidecarConfiguration
sidecarConfiguration,
Provider<CassandraBridge>
cassandraBridgeProvider)
{
this.instanceFetcher = instanceFetcher;
this.databaseAccessor = databaseAccessor;
- this.driverUnsupportedSchemaCache = driverUnsupportedSchemaCache;
+ this.schemaCache = schemaCache;
this.sidecarConfiguration = sidecarConfiguration;
this.cassandraBridgeProvider = cassandraBridgeProvider;
}
@@ -113,8 +113,7 @@ public class CassandraClusterSchemaMonitor implements
PeriodicTask
try
{
LOGGER.debug("Checking for schema changes...");
- String fullSchemaText =
DriverUnsupportedSchemaCache.concatSchemas(databaseAccessor.fullSchema(),
-
driverUnsupportedSchemaCache.getFullSchema());
+ String fullSchemaText =
SchemaCache.concatSchemas(databaseAccessor.fullSchema(),
schemaCache.getUnsupportedTableSchema());
if (!fullSchemaText.equals(currSchemaText.get()))
{
LOGGER.info("Schema change detected, refreshing CDC tables");
diff --git
a/server/src/test/integration/org/apache/cassandra/sidecar/testing/IntegrationTestModule.java
b/server/src/test/integration/org/apache/cassandra/sidecar/testing/IntegrationTestModule.java
index ce6c4590b..816018364 100644
---
a/server/src/test/integration/org/apache/cassandra/sidecar/testing/IntegrationTestModule.java
+++
b/server/src/test/integration/org/apache/cassandra/sidecar/testing/IntegrationTestModule.java
@@ -135,7 +135,7 @@ public class IntegrationTestModule extends AbstractModule
DriverConfiguration driverConfiguration =
DriverConfigurationImpl.builder()
.localDc("dc1")
.numConnections(1)
-
.unsupportedTableSchemaRefreshTime(new SecondBoundConfiguration(3,
TimeUnit.SECONDS))
+
.schemaRefreshTime(new SecondBoundConfiguration(3, TimeUnit.SECONDS))
.authProvider(authProvider)
.build();
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/TestModule.java
b/server/src/test/java/org/apache/cassandra/sidecar/TestModule.java
index f2629483e..d67d58e3f 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/TestModule.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/TestModule.java
@@ -65,7 +65,7 @@ import
org.apache.cassandra.sidecar.config.yaml.SchemaKeyspaceConfigurationImpl;
import org.apache.cassandra.sidecar.config.yaml.SidecarConfigurationImpl;
import org.apache.cassandra.sidecar.config.yaml.TestServiceConfiguration;
import org.apache.cassandra.sidecar.config.yaml.ThrottleConfigurationImpl;
-import org.apache.cassandra.sidecar.db.DriverUnsupportedSchemaCache;
+import org.apache.cassandra.sidecar.db.SchemaCache;
import org.apache.cassandra.sidecar.utils.CassandraVersionProvider;
import static org.apache.cassandra.sidecar.utils.TestMetricUtils.registry;
@@ -268,11 +268,12 @@ public class TestModule extends AbstractModule
@Provides
@Singleton
- public DriverUnsupportedSchemaCache driverUnsupportedSchemaCache()
+ public SchemaCache schemaCache()
{
- DriverUnsupportedSchemaCache mock =
mock(DriverUnsupportedSchemaCache.class);
- when(mock.getFullSchema()).thenReturn("");
+ SchemaCache mock = mock(SchemaCache.class);
+ when(mock.getUnsupportedTableSchema()).thenReturn("");
when(mock.getKeyspaceSchema(any())).thenReturn("");
+ when(mock.getFullSchema()).thenReturn("");
return mock;
}
}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/db/CQLSchemaAccessorTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/db/CQLSchemaAccessorTest.java
index e9c6ad626..b8635af10 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/db/CQLSchemaAccessorTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/db/CQLSchemaAccessorTest.java
@@ -35,6 +35,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.startsWith;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -42,14 +43,15 @@ class CQLSchemaAccessorTest
{
static final Name KEYSPACE = new Name("keyspace1");
static final Name TABLE = new Name("table1");
- static final String SCHEMA = "CREATE TABLE keyspace1.table1 (a int, b int,
PRIMARY KEY (a));";
+ static final String VIRTUAL_KEYSPACE = "system_views";
+ static final String TABLE_SCHEMA = "CREATE TABLE keyspace1.table1 (a int,
b int, PRIMARY KEY (a));";
CQLSchemaAccessor schemaAccessor;
@BeforeEach
void setUp()
{
- CQLSessionProvider sessionProvider =
mockCQLSessionProvider(KEYSPACE.name(), TABLE.name(), SCHEMA);
+ CQLSessionProvider sessionProvider =
mockCQLSessionProvider(KEYSPACE.name(), TABLE.name(), TABLE_SCHEMA);
schemaAccessor = new CQLSchemaAccessor(sessionProvider);
}
@@ -62,7 +64,7 @@ class CQLSchemaAccessorTest
@Test
void testGetExistingTable()
{
- assertThat(schemaAccessor.getTableSchema(KEYSPACE,
TABLE)).containsExactly(SCHEMA);
+ assertThat(schemaAccessor.getTableSchema(KEYSPACE,
TABLE)).containsExactly(TABLE_SCHEMA);
}
@Test
@@ -71,7 +73,26 @@ class CQLSchemaAccessorTest
assertThat(schemaAccessor.getTableSchema(KEYSPACE, new
Name("unknown"))).isNull();
}
- static CQLSessionProvider mockCQLSessionProvider(String keyspace, String
table, String schema)
+ @Test
+ void testGetExistingKeyspace()
+ {
+ assertThat(schemaAccessor.getKeyspaceSchema(KEYSPACE))
+ .containsExactly(createKeyspaceStatement(KEYSPACE.name()),
TABLE_SCHEMA);
+ }
+
+ @Test
+ void testGetNotExistingKeyspace()
+ {
+ assertThat(schemaAccessor.getKeyspaceSchema(new
Name("unknown"))).isNull();
+ }
+
+ @Test
+ void testGetKeyspaceWithoutSchema()
+ {
+ assertThat(schemaAccessor.getKeyspaceSchema(new
Name(VIRTUAL_KEYSPACE))).isEmpty();
+ }
+
+ static CQLSessionProvider mockCQLSessionProvider(String keyspace, String
table, String tableSchema)
{
Session session = mock(Session.class, RETURNS_DEEP_STUBS);
@@ -87,17 +108,44 @@ class CQLSchemaAccessorTest
String describeTable = String.format("DESCRIBE TABLE %s.%s", keyspace,
table);
when(session.execute(eq(describeTable))).then(invocation -> {
ResultSet resultSet = mock(ResultSet.class);
- Row row = mockRow(Map.of("create_statement", schema));
+ Row row = mockRow(Map.of("create_statement", tableSchema));
when(resultSet.all()).thenReturn(List.of(row));
return resultSet;
});
+ when(session.execute(startsWith("DESCRIBE KEYSPACE "))).thenThrow(new
InvalidQueryException("Unknown keyspace"));
+
+ String describeKeyspace = String.format("DESCRIBE KEYSPACE %s",
keyspace);
+ when(session.execute(eq(describeKeyspace))).then(invocation -> {
+ ResultSet resultSet = mock(ResultSet.class);
+ doReturn(List.of(mockRow(Map.of("create_statement",
createKeyspaceStatement(keyspace))),
+ mockRow(Map.of("create_statement",
tableSchema)))).when(resultSet).all();
+ return resultSet;
+ });
+
+ when(session.execute(eq(String.format("DESCRIBE KEYSPACE %s",
VIRTUAL_KEYSPACE)))).then(invocation -> {
+ ResultSet resultSet = mock(ResultSet.class);
+ when(resultSet.all()).thenReturn(List.of());
+ return resultSet;
+ });
+
CQLSessionProvider cqlSession = mock(CQLSessionProvider.class);
when(cqlSession.get()).thenReturn(session);
return cqlSession;
}
+ static String createKeyspaceStatement(String keyspace)
+ {
+ return String.format("CREATE KEYSPACE %s WITH replication = {'class':
'SimpleStrategy',"
+ + " 'replication_factor': '1'} AND
durable_writes = true;", keyspace);
+ }
+
+ static String keyspaceSchema(String keyspace, String schema)
+ {
+ return createKeyspaceStatement(keyspace) + "\n\n" + schema;
+ }
+
static Row mockRow(Map<String, String> values)
{
Row row = mock(Row.class);
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/db/DriverUnsupportedSchemaCacheTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/db/DriverUnsupportedSchemaCacheTest.java
deleted file mode 100644
index 0c9de1ebe..000000000
---
a/server/src/test/java/org/apache/cassandra/sidecar/db/DriverUnsupportedSchemaCacheTest.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * 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.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import com.datastax.driver.core.BoundStatement;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import io.vertx.core.Future;
-import io.vertx.core.Promise;
-import org.apache.cassandra.sidecar.common.server.CQLSessionProvider;
-import org.apache.cassandra.sidecar.common.server.data.Name;
-import
org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
-import org.apache.cassandra.sidecar.config.DriverConfiguration;
-import org.apache.cassandra.sidecar.config.SidecarConfiguration;
-import org.apache.cassandra.sidecar.exceptions.CassandraUnavailableException;
-
-import static org.apache.cassandra.sidecar.db.CQLSchemaAccessorTest.mockRow;
-import static
org.apache.cassandra.sidecar.exceptions.CassandraUnavailableException.Service.CQL;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-class DriverUnsupportedSchemaCacheTest
-{
- static final Name KEYSPACE = new Name("keyspace1");
- static final Name TABLE = new Name("table1");
- static final String SCHEMA = "CREATE TABLE keyspace1.table1 (a int, b
vector<float, 3>, PRIMARY KEY (a));";
-
- CQLSessionProvider sessionProvider;
- Session session;
- DriverUnsupportedSchemaCache schemaCache;
-
- @BeforeEach
- void setUp()
- {
- sessionProvider =
CQLSchemaAccessorTest.mockCQLSessionProvider(KEYSPACE.name(), TABLE.name(),
SCHEMA);
- session = sessionProvider.get();
- mockPreparedQuery(session,
- "SELECT keyspace_name, table_name FROM
system_schema.tables",
- List.of(Map.of("keyspace_name", KEYSPACE.name(),
"table_name", TABLE.name())));
- SidecarConfiguration sidecarConfiguration = mockSidecarConfiguration();
- schemaCache = new DriverUnsupportedSchemaCache(sidecarConfiguration,
sessionProvider);
- }
-
- @Test
- void testImmediateSchemaLookup()
- {
- assertThat(schemaCache.getFullSchema()).isEqualTo(SCHEMA);
- assertThat(schemaCache.getKeyspaceSchema(KEYSPACE)).isEqualTo(SCHEMA);
- assertThat(schemaCache.getTableSchema(KEYSPACE,
TABLE)).isEqualTo(SCHEMA);
-
- assertThat(schemaCache.getTableSchema(new Name("unknown"),
TABLE)).isNull();
- assertThat(schemaCache.getTableSchema(KEYSPACE, new
Name("unknown"))).isNull();
- }
-
- @Test
- void testSchemaLookupAfterRefresh()
- {
- // mark cache as initialized, but effectively empty
- schemaCache.setInitialized(true);
-
- assertThat(schemaCache.getFullSchema()).isEqualTo("");
- assertThat(schemaCache.getKeyspaceSchema(KEYSPACE)).isEqualTo("");
-
- Promise<Void> p = Promise.promise();
- schemaCache.execute(p);
- Future<Void> future = p.future();
- assertThat(future.succeeded()).isTrue();
-
- // simulate Cassandra unavailability to verify data is taken from cache
- when(sessionProvider.get()).thenThrow(new
CassandraUnavailableException(CQL, "CQL unavailable"));
-
- assertThat(schemaCache.getFullSchema()).isEqualTo(SCHEMA);
- assertThat(schemaCache.getKeyspaceSchema(KEYSPACE)).isEqualTo(SCHEMA);
- assertThat(schemaCache.getTableSchema(KEYSPACE,
TABLE)).isEqualTo(SCHEMA);
- }
-
- static SidecarConfiguration mockSidecarConfiguration()
- {
- SidecarConfiguration sidecarConfiguration =
mock(SidecarConfiguration.class);
- DriverConfiguration driverConfiguration =
mock(DriverConfiguration.class);
-
when(driverConfiguration.unsupportedTableSchemaRefreshTime()).thenReturn(new
SecondBoundConfiguration(5, TimeUnit.SECONDS));
-
when(sidecarConfiguration.driverConfiguration()).thenReturn(driverConfiguration);
- return sidecarConfiguration;
- }
-
- static void mockPreparedQuery(Session session, String statement,
List<Map<String, String>> rows)
- {
- PreparedStatement preparedStatement = mock(PreparedStatement.class);
- BoundStatement boundStatement = mock(BoundStatement.class);
- when(session.prepare(eq(statement))).thenReturn(preparedStatement);
- when(preparedStatement.bind()).thenReturn(boundStatement);
- when(session.execute(eq(boundStatement))).then(invocation -> {
- ResultSet resultSet = mock(ResultSet.class);
- List<Row> mockRows = new ArrayList<>();
- rows.forEach(row -> mockRows.add(mockRow(row)));
- when(resultSet.all()).thenReturn(mockRows);
- return resultSet;
- });
- }
-}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/db/SchemaCacheTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/db/SchemaCacheTest.java
new file mode 100644
index 000000000..fbed45d5d
--- /dev/null
+++ b/server/src/test/java/org/apache/cassandra/sidecar/db/SchemaCacheTest.java
@@ -0,0 +1,361 @@
+/*
+ * 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.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TableMetadata;
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
+import org.apache.cassandra.sidecar.common.server.CQLSessionProvider;
+import org.apache.cassandra.sidecar.common.server.data.Name;
+import
org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
+import org.apache.cassandra.sidecar.config.DriverConfiguration;
+import org.apache.cassandra.sidecar.config.SidecarConfiguration;
+import org.apache.cassandra.sidecar.exceptions.CassandraUnavailableException;
+
+import static
org.apache.cassandra.sidecar.db.CQLSchemaAccessorTest.VIRTUAL_KEYSPACE;
+import static
org.apache.cassandra.sidecar.db.CQLSchemaAccessorTest.createKeyspaceStatement;
+import static
org.apache.cassandra.sidecar.db.CQLSchemaAccessorTest.keyspaceSchema;
+import static org.apache.cassandra.sidecar.db.CQLSchemaAccessorTest.mockRow;
+import static
org.apache.cassandra.sidecar.exceptions.CassandraUnavailableException.Service.CQL;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class SchemaCacheTest
+{
+ static final Name KEYSPACE = new Name("keyspace1");
+ static final Name TABLE = new Name("table1");
+ static final String TABLE_LIST_QUERY = "SELECT keyspace_name, table_name
FROM system_schema.tables";
+ // a table the Java driver cannot parse, because of the vector data type
+ static final String SCHEMA = "CREATE TABLE keyspace1.table1 (a int, b
vector<float, 3>, PRIMARY KEY (a));";
+ static final String KEYSPACE_SCHEMA = keyspaceSchema(KEYSPACE.name(),
SCHEMA);
+
+ CQLSessionProvider sessionProvider;
+ Session session;
+ SchemaCache schemaCache;
+ List<Map<String, String>> tableRows;
+ List<Map<String, String>> keyspaceRows;
+
+ @BeforeEach
+ void setUp()
+ {
+ sessionProvider =
CQLSchemaAccessorTest.mockCQLSessionProvider(KEYSPACE.name(), TABLE.name(),
SCHEMA);
+ session = sessionProvider.get();
+ tableRows = new ArrayList<>();
+ tableRows.add(Map.of("keyspace_name", KEYSPACE.name(), "table_name",
TABLE.name()));
+ mockPreparedQuery(session, TABLE_LIST_QUERY, tableRows);
+ keyspaceRows = new ArrayList<>();
+ keyspaceRows.add(Map.of("keyspace_name", KEYSPACE.name()));
+ mockQuery(session, "DESCRIBE KEYSPACES", keyspaceRows);
+ SidecarConfiguration sidecarConfiguration = mockSidecarConfiguration();
+ schemaCache = new SchemaCache(sidecarConfiguration, sessionProvider);
+ }
+
+ @Test
+ void testImmediateUnsupportedTableSchemaLookup()
+ {
+ assertThat(schemaCache.getUnsupportedTableSchema()).isEqualTo(SCHEMA);
+ assertThat(schemaCache.getUnsupportedTableSchema(KEYSPACE,
TABLE)).isEqualTo(SCHEMA);
+
+ assertThat(schemaCache.getUnsupportedTableSchema(new Name("unknown"),
TABLE)).isNull();
+ assertThat(schemaCache.getUnsupportedTableSchema(KEYSPACE, new
Name("unknown"))).isNull();
+ }
+
+ @Test
+ void testUnsupportedTableSchemaLookupWithoutRefresh()
+ {
+ assertThat(schemaCache.getUnsupportedTableSchema(KEYSPACE, TABLE,
false)).isEqualTo(SCHEMA);
+
+ // for a table created after the last refresh
+ Name table = new Name("table2");
+ assertThat(schemaCache.getUnsupportedTableSchema(KEYSPACE, table,
false)).isNull();
+ verify(session, never()).execute("DESCRIBE TABLE keyspace1.table2");
+
+ mockDescribeTable(session, KEYSPACE.name(), table.name(), "CREATE
TABLE keyspace1.table2 (a int PRIMARY KEY);");
+ assertThat(schemaCache.getUnsupportedTableSchema(KEYSPACE, table,
true))
+ .isEqualTo("CREATE TABLE keyspace1.table2 (a int PRIMARY KEY);");
+ verify(session, times(1)).execute("DESCRIBE TABLE keyspace1.table2");
+ }
+
+ @Test
+ void testUnsupportedTableSchemaLookupForUnquotedNames()
+ {
+ assertThat(schemaCache.getUnsupportedTableSchema(new Name("Keyspace1"),
+ new Name("TABLE1"),
+ false))
+ .isEqualTo(SCHEMA);
+
+ assertThat(schemaCache.getUnsupportedTableSchema(new
Name("\"Keyspace1\""),
+ new
Name("\"Table1\""),
+ false))
+ .isNull();
+
+ assertThat(schemaCache.getUnsupportedTableSchema(new
Name("\"keyspace1\""),
+ new
Name("\"table1\""),
+ false))
+ .isEqualTo(SCHEMA);
+ }
+
+ @Test
+ void testUnsupportedTableSchemaLookupAfterRefresh()
+ {
+ schemaCache.setInitialized(true);
+
+ assertThat(schemaCache.getUnsupportedTableSchema()).isEqualTo("");
+
+ Promise<Void> p = Promise.promise();
+ schemaCache.execute(p);
+ Future<Void> future = p.future();
+ assertThat(future.succeeded()).isTrue();
+
+ when(sessionProvider.get()).thenThrow(new
CassandraUnavailableException(CQL, "CQL unavailable"));
+
+ assertThat(schemaCache.getUnsupportedTableSchema()).isEqualTo(SCHEMA);
+ assertThat(schemaCache.getUnsupportedTableSchema(KEYSPACE,
TABLE)).isEqualTo(SCHEMA);
+ }
+
+ @Test
+ void testUnsupportedTableSchemaIsAssembledInStableOrder()
+ {
+ String otherSchema = "CREATE TABLE keyspace1.table0 (a int, b
vector<float, 3>, PRIMARY KEY (a));";
+ mockDescribeTable(session, KEYSPACE.name(), "table0", otherSchema);
+ tableRows.add(Map.of("keyspace_name", KEYSPACE.name(), "table_name",
"table0"));
+
+
assertThat(schemaCache.getUnsupportedTableSchema()).isEqualTo(otherSchema +
"\n\n" + SCHEMA);
+ schemaCache.refresh(false);
+
assertThat(schemaCache.getUnsupportedTableSchema()).isEqualTo(otherSchema +
"\n\n" + SCHEMA);
+ }
+
+ @Test
+ void testDroppedTableIsRemovedOnRefresh()
+ {
+ assertThat(schemaCache.getUnsupportedTableSchema()).isEqualTo(SCHEMA);
+ tableRows.clear();
+ schemaCache.refresh(false);
+ assertThat(schemaCache.getUnsupportedTableSchema()).isEqualTo("");
+ assertThat(schemaCache.getUnsupportedTableSchema(KEYSPACE, TABLE,
false)).isNull();
+ }
+
+ @Test
+ void testTableKnownToDriverIsNotCached()
+ {
+ Cluster mockCluster = mock(Cluster.class);
+ when(session.getCluster()).thenReturn(mockCluster);
+ Metadata mockMetadata = mock(Metadata.class);
+ when(mockCluster.getMetadata()).thenReturn(mockMetadata);
+ doReturn(List.of(mockKeyspaceMetadata(KEYSPACE.name(),
TABLE.name()))).when(mockMetadata).getKeyspaces();
+
+ assertThat(schemaCache.getUnsupportedTableSchema()).isEqualTo("");
+ assertThat(schemaCache.getUnsupportedTableSchema(KEYSPACE, TABLE,
false)).isNull();
+ }
+
+ @Test
+ void testImmediateDescribeSchemaLookup()
+ {
+ assertThat(schemaCache.getFullSchema()).isEqualTo(KEYSPACE_SCHEMA);
+
assertThat(schemaCache.getKeyspaceSchema(KEYSPACE.name())).isEqualTo(KEYSPACE_SCHEMA);
+ assertThat(schemaCache.getKeyspaceSchema("unknown")).isNull();
+
+ verify(session, times(1)).execute("DESCRIBE KEYSPACE " +
KEYSPACE.name());
+ }
+
+ @Test
+ void testDescribeSchemaLookupOnDemand()
+ {
+ schemaCache.setInitialized(true);
+ assertThat(schemaCache.getFullSchema()).isEqualTo("");
+
+
assertThat(schemaCache.getKeyspaceSchema(KEYSPACE.name())).isEqualTo(KEYSPACE_SCHEMA);
+
assertThat(schemaCache.getKeyspaceSchema(KEYSPACE.name())).isEqualTo(KEYSPACE_SCHEMA);
+ verify(session, times(1)).execute("DESCRIBE KEYSPACE " +
KEYSPACE.name());
+
+ when(sessionProvider.get()).thenThrow(new
CassandraUnavailableException(CQL, "CQL unavailable"));
+
+ assertThat(schemaCache.getFullSchema()).isEqualTo(KEYSPACE_SCHEMA);
+
assertThat(schemaCache.getKeyspaceSchema(KEYSPACE.name())).isEqualTo(KEYSPACE_SCHEMA);
+ }
+
+ @Test
+ void testDescribeSchemaOfAllKeyspacesIsInvalidatedOnDemandLookup()
+ {
+ assertThat(schemaCache.getFullSchema()).isEqualTo(KEYSPACE_SCHEMA);
+
+ String otherSchema = "CREATE TABLE keyspace2.table1 (a int PRIMARY
KEY);";
+ mockDescribeKeyspace(session, "keyspace2", otherSchema);
+
+
assertThat(schemaCache.getKeyspaceSchema("keyspace2")).isEqualTo(keyspaceSchema("keyspace2",
otherSchema));
+ assertThat(schemaCache.getFullSchema())
+ .isEqualTo(KEYSPACE_SCHEMA + "\n\n" + keyspaceSchema("keyspace2",
otherSchema));
+ }
+
+ @Test
+ void testDescribeSchemaIsAssembledInStableOrder()
+ {
+ String otherSchema = "CREATE TABLE aaa_keyspace.table1 (a int PRIMARY
KEY);";
+ mockDescribeKeyspace(session, "aaa_keyspace", otherSchema);
+ keyspaceRows.add(Map.of("keyspace_name", "aaa_keyspace"));
+
+ assertThat(schemaCache.getFullSchema())
+ .isEqualTo(keyspaceSchema("aaa_keyspace", otherSchema) + "\n\n" +
KEYSPACE_SCHEMA);
+ schemaCache.refresh(false);
+ assertThat(schemaCache.getFullSchema())
+ .isEqualTo(keyspaceSchema("aaa_keyspace", otherSchema) + "\n\n" +
KEYSPACE_SCHEMA);
+ }
+
+ @Test
+ void testDroppedKeyspaceIsRemovedOnRefresh()
+ {
+ assertThat(schemaCache.getFullSchema()).isEqualTo(KEYSPACE_SCHEMA);
+
+ keyspaceRows.clear();
+ schemaCache.refresh(false);
+
+ assertThat(schemaCache.getFullSchema()).isEqualTo("");
+ }
+
+ @Test
+ void testKeyspaceWithoutSchemaIsNotCached()
+ {
+ keyspaceRows.add(Map.of("keyspace_name", VIRTUAL_KEYSPACE));
+
+ assertThat(schemaCache.getFullSchema()).isEqualTo(KEYSPACE_SCHEMA);
+ assertThat(schemaCache.getKeyspaceSchema(VIRTUAL_KEYSPACE)).isNull();
+ }
+
+ @Test
+ void testLookupWhileCassandraIsUnavailable()
+ {
+ CQLSessionProvider unavailable = mock(CQLSessionProvider.class);
+ when(unavailable.get()).thenThrow(new
CassandraUnavailableException(CQL, "CQL unavailable"));
+ SchemaCache cache = new SchemaCache(mockSidecarConfiguration(),
unavailable);
+
+ assertThat(cache.getFullSchema()).isEqualTo("");
+ assertThat(cache.getUnsupportedTableSchema()).isEqualTo("");
+
+ assertThatThrownBy(() -> cache.getKeyspaceSchema(KEYSPACE.name()))
+ .isInstanceOf(CassandraUnavailableException.class);
+ assertThatThrownBy(() -> cache.getUnsupportedTableSchema(KEYSPACE,
TABLE))
+ .isInstanceOf(CassandraUnavailableException.class);
+
+ assertThat(cache.getUnsupportedTableSchema(KEYSPACE, TABLE,
false)).isNull();
+ }
+
+ @Test
+ void testExecuteFailsWhenRefreshFails()
+ {
+ CQLSessionProvider failing = mock(CQLSessionProvider.class);
+ when(failing.get()).thenThrow(new RuntimeException("cannot get
session"));
+ SchemaCache cache = new SchemaCache(mockSidecarConfiguration(),
failing);
+
+ Promise<Void> p = Promise.promise();
+ cache.execute(p);
+ Future<Void> future = p.future();
+ assertThat(future.failed()).isTrue();
+
assertThat(future.cause()).isInstanceOf(RuntimeException.class).hasMessage("cannot
get session");
+ }
+
+ @Test
+ void testConcatSchemas()
+ {
+ assertThat(SchemaCache.concatSchemas()).isEqualTo("");
+ assertThat(SchemaCache.concatSchemas("", null)).isEqualTo("");
+ assertThat(SchemaCache.concatSchemas(SCHEMA)).isEqualTo(SCHEMA);
+ assertThat(SchemaCache.concatSchemas(KEYSPACE_SCHEMA, null, "", "\n" +
SCHEMA + "\n"))
+ .isEqualTo(KEYSPACE_SCHEMA + "\n\n" + SCHEMA);
+ }
+
+ private SidecarConfiguration mockSidecarConfiguration()
+ {
+ SidecarConfiguration sidecarConfiguration =
mock(SidecarConfiguration.class);
+ DriverConfiguration driverConfiguration =
mock(DriverConfiguration.class);
+ when(driverConfiguration.schemaRefreshTime()).thenReturn(new
SecondBoundConfiguration(5, TimeUnit.SECONDS));
+
when(sidecarConfiguration.driverConfiguration()).thenReturn(driverConfiguration);
+ return sidecarConfiguration;
+ }
+
+ private void mockPreparedQuery(Session session, String statement,
List<Map<String, String>> rows)
+ {
+ PreparedStatement preparedStatement = mock(PreparedStatement.class);
+ BoundStatement boundStatement = mock(BoundStatement.class);
+ when(session.prepare(eq(statement))).thenReturn(preparedStatement);
+ when(preparedStatement.bind()).thenReturn(boundStatement);
+ when(session.execute(eq(boundStatement))).then(invocation ->
mockResultSet(rows));
+ }
+
+ private void mockDescribeKeyspace(Session session, String keyspace, String
schema)
+ {
+ mockQuery(session,
+ "DESCRIBE KEYSPACE " + keyspace,
+ List.of(Map.of("create_statement",
createKeyspaceStatement(keyspace)),
+ Map.of("create_statement", schema)));
+ }
+
+ private void mockDescribeTable(Session session, String keyspace, String
table, String schema)
+ {
+ mockQuery(session,
+ String.format("DESCRIBE TABLE %s.%s", keyspace, table),
+ List.of(Map.of("create_statement", schema)));
+ }
+
+ private void mockQuery(Session session, String statement, List<Map<String,
String>> rows)
+ {
+ when(session.execute(eq(statement))).then(invocation ->
mockResultSet(rows));
+ }
+
+ private KeyspaceMetadata mockKeyspaceMetadata(String keyspace, String
table)
+ {
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.getName()).thenReturn(table);
+ KeyspaceMetadata keyspaceMetadata = mock(KeyspaceMetadata.class);
+ when(keyspaceMetadata.getName()).thenReturn(keyspace);
+ when(keyspaceMetadata.getTables()).thenReturn(List.of(tableMetadata));
+ return keyspaceMetadata;
+ }
+
+ private ResultSet mockResultSet(List<Map<String, String>> rows)
+ {
+ ResultSet resultSet = mock(ResultSet.class);
+ List<Row> mockRows = new ArrayList<>();
+ rows.forEach(row -> mockRows.add(mockRow(row)));
+ when(resultSet.all()).thenReturn(mockRows);
+ return resultSet;
+ }
+}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/handlers/SchemaHandlerTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/handlers/SchemaHandlerTest.java
index 37ebd3e50..b4a3115f6 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/handlers/SchemaHandlerTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/handlers/SchemaHandlerTest.java
@@ -53,14 +53,14 @@ import
org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
-import org.apache.cassandra.sidecar.db.DriverUnsupportedSchemaCache;
+import org.apache.cassandra.sidecar.db.SchemaCache;
import org.apache.cassandra.sidecar.modules.SidecarModules;
import org.apache.cassandra.sidecar.server.Server;
import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static
io.netty.handler.codec.http.HttpResponseStatus.SERVICE_UNAVAILABLE;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -71,17 +71,22 @@ import static org.mockito.Mockito.when;
class SchemaHandlerTest
{
static final Logger LOGGER =
LoggerFactory.getLogger(SchemaHandlerTest.class);
+ static final String TEST_KEYSPACE = "testKeyspace";
Vertx vertx;
Server server;
@TempDir
File instanceDir;
String testKeyspaceSchema;
+ SchemaCache mockSchemaCache;
@BeforeEach
void before() throws InterruptedException, IOException
{
ClassLoader cl = getClass().getClassLoader();
testKeyspaceSchema =
IOUtils.readFully(cl.getResourceAsStream("schema/test_keyspace_schema.cql"));
+ mockSchemaCache = mock(SchemaCache.class);
+ when(mockSchemaCache.getFullSchema()).thenReturn("MOCK SCHEMA");
+
when(mockSchemaCache.getKeyspaceSchema(TEST_KEYSPACE)).thenReturn(testKeyspaceSchema);
Injector injector;
injector = Guice.createInjector(Modules.override(SidecarModules.all())
@@ -119,7 +124,22 @@ class SchemaHandlerTest
JsonObject jsonObject = response.bodyAsJsonObject();
assertThat(jsonObject.getString("keyspace")).isNull();
assertThat(jsonObject.getString("schema"))
- .isEqualTo("FULL SCHEMA");
+ .isEqualTo("MOCK SCHEMA");
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testSchemaNotAvailable(VertxTestContext context)
+ {
+ when(mockSchemaCache.getFullSchema()).thenReturn("");
+
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/api/v1/schema/keyspaces";
+ client.get(server.actualPort(), "127.0.0.1", testRoute)
+ .expect(ResponsePredicate.SC_SERVICE_UNAVAILABLE)
+ .send(context.succeeding(response -> context.verify(() -> {
+
assertThat(response.statusCode()).isEqualTo(SERVICE_UNAVAILABLE.code());
context.completeNow();
})));
}
@@ -134,9 +154,24 @@ class SchemaHandlerTest
.send(context.succeeding(response -> context.verify(() -> {
assertThat(response.statusCode()).isEqualTo(OK.code());
JsonObject jsonObject = response.bodyAsJsonObject();
-
assertThat(jsonObject.getString("keyspace")).isEqualTo("testKeyspace");
+
assertThat(jsonObject.getString("keyspace")).isEqualTo(TEST_KEYSPACE);
assertThat(jsonObject.getString("schema"))
- .isEqualTo(testKeyspaceSchema.trim());
+ .isEqualTo(testKeyspaceSchema);
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testKeyspaceSchemaNotAvailable(VertxTestContext context)
+ {
+
when(mockSchemaCache.getKeyspaceSchema(TEST_KEYSPACE)).thenReturn(null);
+
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/api/v1/schema/keyspaces/testKeyspace";
+ client.get(server.actualPort(), "127.0.0.1", testRoute)
+ .expect(ResponsePredicate.SC_NOT_FOUND)
+ .send(context.succeeding(response -> context.verify(() -> {
+
assertThat(response.statusCode()).isEqualTo(NOT_FOUND.code());
context.completeNow();
})));
}
@@ -175,9 +210,8 @@ class SchemaHandlerTest
when(instanceMetadata.delegate()).thenReturn(mockCassandraAdapterDelegate);
Metadata mockMetadata = mock(Metadata.class);
KeyspaceMetadata mockKeyspaceMetadata =
mock(KeyspaceMetadata.class);
- when(mockMetadata.exportSchemaAsString()).thenReturn("FULL
SCHEMA");
-
when(mockMetadata.getKeyspace("testKeyspace")).thenReturn(mockKeyspaceMetadata);
-
when(mockKeyspaceMetadata.exportAsString()).thenReturn(testKeyspaceSchema);
+
when(mockMetadata.getKeyspace(TEST_KEYSPACE)).thenReturn(mockKeyspaceMetadata);
+ when(mockKeyspaceMetadata.getName()).thenReturn(TEST_KEYSPACE);
when(mockCassandraAdapterDelegate.metadata()).thenReturn(mockMetadata);
@@ -191,12 +225,9 @@ class SchemaHandlerTest
@Provides
@Singleton
- public DriverUnsupportedSchemaCache driverUnsupportedSchemaCache()
+ public SchemaCache schemaCache()
{
- DriverUnsupportedSchemaCache schemaAccessor =
mock(DriverUnsupportedSchemaCache.class);
- when(schemaAccessor.getFullSchema()).thenReturn("");
- when(schemaAccessor.getKeyspaceSchema(any())).thenReturn("");
- return schemaAccessor;
+ return mockSchemaCache;
}
}
}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitorTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitorTest.java
index e225efc6d..242641e1f 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitorTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitorTest.java
@@ -41,7 +41,7 @@ import
org.apache.cassandra.sidecar.config.SchemaKeyspaceConfiguration;
import org.apache.cassandra.sidecar.config.ServiceConfiguration;
import org.apache.cassandra.sidecar.config.SidecarConfiguration;
import org.apache.cassandra.sidecar.db.CdcDatabaseAccessor;
-import org.apache.cassandra.sidecar.db.DriverUnsupportedSchemaCache;
+import org.apache.cassandra.sidecar.db.SchemaCache;
import org.apache.cassandra.sidecar.utils.CdcUtil;
import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
import org.apache.cassandra.spark.data.CqlTable;
@@ -72,7 +72,7 @@ class CassandraClusterSchemaMonitorTest
private CassandraClusterSchemaMonitor clusterSchema;
private InstanceMetadataFetcher mockInstanceFetcher;
private CdcDatabaseAccessor mockDatabaseAccessor;
- private DriverUnsupportedSchemaCache mockDriverUnsupportedSchemaCache;
+ private SchemaCache mockSchemaCache;
private SidecarConfiguration mockSidecarConfiguration;
private ServiceConfiguration mockServiceConfiguration;
private CdcConfiguration mockCdcConfiguration;
@@ -102,7 +102,7 @@ class CassandraClusterSchemaMonitorTest
{
mockInstanceFetcher = mock(InstanceMetadataFetcher.class);
mockDatabaseAccessor = mock(CdcDatabaseAccessor.class);
- mockDriverUnsupportedSchemaCache =
mock(DriverUnsupportedSchemaCache.class);
+ mockSchemaCache = mock(SchemaCache.class);
mockSidecarConfiguration = mock(SidecarConfiguration.class);
mockServiceConfiguration = mock(ServiceConfiguration.class);
mockCdcConfiguration = mock(CdcConfiguration.class);
@@ -131,7 +131,7 @@ class CassandraClusterSchemaMonitorTest
// Setup database accessor
when(mockDatabaseAccessor.fullSchema()).thenReturn(INITIAL_SCHEMA);
- when(mockDriverUnsupportedSchemaCache.getFullSchema()).thenReturn("");
+ when(mockSchemaCache.getUnsupportedTableSchema()).thenReturn("");
when(mockDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(mockDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenReturn(UUID.randomUUID());
mockCdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
@@ -139,7 +139,7 @@ class CassandraClusterSchemaMonitorTest
clusterSchema = new CassandraClusterSchemaMonitor(
mockInstanceFetcher,
mockDatabaseAccessor,
- mockDriverUnsupportedSchemaCache,
+ mockSchemaCache,
mockSidecarConfiguration,
() -> mockCassandraBridge
);
@@ -238,7 +238,6 @@ class CassandraClusterSchemaMonitorTest
// Verify initial schema processing
verify(mockDatabaseAccessor, times(1)).fullSchema();
- verify(mockDriverUnsupportedSchemaCache, times(1)).getFullSchema();
verify(mockCdcBridge,
times(1)).updateCdcSchema(tablesCaptor.capture(),
eq(Partitioner.Murmur3Partitioner), any());
assertThat(tablesCaptor.getValue()).hasSize(1);
@@ -247,7 +246,6 @@ class CassandraClusterSchemaMonitorTest
// Verify schema change detection and update
verify(mockDatabaseAccessor, times(2)).fullSchema();
- verify(mockDriverUnsupportedSchemaCache, times(2)).getFullSchema();
verify(mockCdcBridge,
times(2)).updateCdcSchema(tablesCaptor.capture(),
eq(Partitioner.Murmur3Partitioner), any());
assertThat(tablesCaptor.getValue()).hasSize(2);
}
@@ -322,7 +320,7 @@ class CassandraClusterSchemaMonitorTest
// Same table set every refresh — nothing ever becomes stale.
when(mockDatabaseAccessor.fullSchema())
.thenReturn(INITIAL_SCHEMA)
- .thenReturn(INITIAL_SCHEMA + " "); // trailing space forces a
"schema changed" refresh without changing the table set
+ .thenReturn(INITIAL_SCHEMA + " -- changed"); // forces a "schema
changed" refresh without changing the table set
clusterSchema.refresh();
clusterSchema.refresh();
@@ -346,10 +344,9 @@ class CassandraClusterSchemaMonitorTest
String updatedCdcTableStmt = "CREATE TABLE test.cdc_table (id text
PRIMARY KEY, data text) WITH cdc = true;";
String updatedMismatchedSchema = updatedCdcTableStmt +
nonCdcTableStmt;
// Third refresh: schema "changes" again but the table set is
identical — appending
- // a non-whitespace comment-like suffix (whitespace alone would be
stripped by
- // DriverUnsupportedSchemaCache.concatSchemas()'s trim(), never
triggering a
- // "schema changed" refresh at all) is enough to force refresh()
to re-run its
- // schema-changed branch and retry unregistration.
+ // a non-whitespace comment-like suffix (whitespace alone is
trimmed when SchemaCache
+ // assembles the full schema, never triggering a "schema changed"
refresh at all) is
+ // enough to force refresh() to re-run its schema-changed branch
and retry unregistration.
String thirdRefreshSchema = updatedMismatchedSchema + " --
refresh3";
Map<TableIdentifier, CdcUtil.TableSchema> initialTables = Map.of(
@@ -474,7 +471,7 @@ class CassandraClusterSchemaMonitorTest
void testRefreshHandlesGenericExceptionGracefully()
{
- when(mockDatabaseAccessor.fullSchema()).thenThrow(new
RuntimeException("Unexpected error"));
+ when(mockDatabaseAccessor.fullSchema()).thenThrow(new
IllegalStateException("Unexpected error"));
// Should now throw the exception instead of handling it gracefully
try
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
