diff --git a/CHANGES.txt b/CHANGES.txt index 8c1cccac32..3ca783fb97 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -69,6 +69,7 @@ Merged from 4.1: * Waiting indefinitely on ReceivedMessage response in StreamSession#receive() can cause deadlock (CASSANDRA-18733) * Allow empty keystore_password in encryption_options (CASSANDRA-18778) Merged from 4.0: + * Retrieve keyspaces metadata and schema version concistently in DescribeStatement (CASSANDRA-18921) * JMH improvements - faster build and async profiler (CASSANDRA-18871) Merged from 3.11: * Fix NPE when using udfContext in UDF after a restart of a node (CASSANDRA-18739) diff --git a/src/java/org/apache/cassandra/cql3/statements/DescribeStatement.java b/src/java/org/apache/cassandra/cql3/statements/DescribeStatement.java index bc009f19ab..90382dfe57 100644 --- a/src/java/org/apache/cassandra/cql3/statements/DescribeStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/DescribeStatement.java @@ -132,14 +132,13 @@ public abstract class DescribeStatement extends CQLStatement.Raw implements C @Override public ResultMessage executeLocally(QueryState state, QueryOptions options) { - Keyspaces keyspaces = Schema.instance.distributedKeyspaces(); - UUID schemaVersion = Schema.instance.getVersion(); + DistributedSchema schema = Schema.instance.getDistributedSchemaBlocking(); - keyspaces = Keyspaces.builder() - .add(keyspaces) - .add(Schema.instance.getLocalKeyspaces()) - .add(VirtualKeyspaceRegistry.instance.virtualKeyspacesMetadata()) - .build(); + Keyspaces keyspaces = Keyspaces.builder() + .add(schema.getKeyspaces()) + .add(Schema.instance.getLocalKeyspaces()) + .add(VirtualKeyspaceRegistry.instance.virtualKeyspacesMetadata()) + .build(); PagingState pagingState = options.getPagingState(); @@ -157,7 +156,7 @@ public abstract class DescribeStatement extends CQLStatement.Raw implements C // (vint bytes) serialized schema hash (currently the result of Keyspaces.hashCode()) // - long offset = getOffset(pagingState, schemaVersion); + long offset = getOffset(pagingState, schema.getVersion()); int pageSize = options.getPageSize(); Stream stream = describe(state.getClientState(), keyspaces); @@ -174,9 +173,7 @@ public abstract class DescribeStatement extends CQLStatement.Raw implements C ResultSet result = new ResultSet(resultMetadata, rows); if (pageSize > 0 && rows.size() == pageSize) - { - result.metadata.setHasMorePages(getPagingState(offset + pageSize, schemaVersion)); - } + result.metadata.setHasMorePages(getPagingState(offset + pageSize, schema.getVersion())); return new ResultMessage.Rows(result); } diff --git a/src/java/org/apache/cassandra/schema/Schema.java b/src/java/org/apache/cassandra/schema/Schema.java index ac40228652..b704fd2997 100644 --- a/src/java/org/apache/cassandra/schema/Schema.java +++ b/src/java/org/apache/cassandra/schema/Schema.java @@ -487,15 +487,30 @@ public class Schema implements SchemaProvider /* Version control */ /** - * @return current schema version + * Returns the current schema version. Although, if the schema is being updated while the method was called, it + * can return a stale version which does not correspond to the current keyspaces metadata. It is because the schema + * version is unknown for the partially applied changes and is updated after the entire schema change is completed. + *

+ * This method should be used only internally by {@link Schema} or {@link SchemaUpdateHandler} implementations. + * Please use {@link #getDistributedSchemaBlocking()} to get schema version consistently in other cases. */ public UUID getVersion() { return version; } + /** + * Returns the current keyspaces metadata and version synchronouly. If the schema is in the middle of a multistep + * transformation, the method blocks until the update is completed. + */ + public synchronized DistributedSchema getDistributedSchemaBlocking() + { + return new DistributedSchema(distributedKeyspaces, version); + } + /** * Checks whether the given schema version is the same as the current local schema. + * Note that this method is non-blocking and may use a stale schema version for comparison - see {@link #getVersion()}. */ public boolean isSameVersion(UUID schemaVersion) { @@ -504,6 +519,7 @@ public class Schema implements SchemaProvider /** * Checks whether the current schema is empty. + * Note that this method is non-blocking and may use a stale schema version for comparison - see {@link #getVersion()}. */ public boolean isEmpty() { diff --git a/src/java/org/apache/cassandra/schema/SchemaEvent.java b/src/java/org/apache/cassandra/schema/SchemaEvent.java index 934516d707..d374323bbd 100644 --- a/src/java/org/apache/cassandra/schema/SchemaEvent.java +++ b/src/java/org/apache/cassandra/schema/SchemaEvent.java @@ -105,7 +105,7 @@ public final class SchemaEvent extends DiagnosticEvent this.nonSystemKeyspaces = schema.distributedKeyspaces().names(); this.userKeyspaces = schema.getUserKeyspaces().immutableCopy(); this.numberOfTables = schema.getNumberOfTables(); - this.version = schema.getVersion(); + this.version = schema.getVersion(); // TODO: rename this field to reflect that the schema version we know here is stale (before the entire transformation started) this.indexTables = schema.distributedKeyspaces().stream() .flatMap(ks -> ks.tables.indexTables().entrySet().stream()) diff --git a/test/unit/org/apache/cassandra/cql3/statements/DescribeStatementTest.java b/test/unit/org/apache/cassandra/cql3/statements/DescribeStatementTest.java index e51d8ed3ad..fc9e8fb135 100644 --- a/test/unit/org/apache/cassandra/cql3/statements/DescribeStatementTest.java +++ b/test/unit/org/apache/cassandra/cql3/statements/DescribeStatementTest.java @@ -23,16 +23,13 @@ import java.util.Optional; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; - import org.apache.commons.lang3.ArrayUtils; - import org.junit.Test; import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row; import com.datastax.driver.core.SimpleStatement; import com.datastax.driver.core.exceptions.InvalidQueryException; - import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.CQLTester; import org.apache.cassandra.cql3.CqlBuilder; @@ -47,7 +44,12 @@ import org.apache.cassandra.service.StorageService; import org.apache.cassandra.transport.ProtocolVersion; import static java.lang.String.format; -import static org.apache.cassandra.schema.SchemaConstants.*; +import static org.apache.cassandra.schema.SchemaConstants.AUTH_KEYSPACE_NAME; +import static org.apache.cassandra.schema.SchemaConstants.DISTRIBUTED_KEYSPACE_NAME; +import static org.apache.cassandra.schema.SchemaConstants.SCHEMA_KEYSPACE_NAME; +import static org.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME; +import static org.apache.cassandra.schema.SchemaConstants.TRACE_KEYSPACE_NAME; +import static org.apache.cassandra.schema.SchemaConstants.VIRTUAL_SCHEMA; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; @@ -58,24 +60,24 @@ public class DescribeStatementTest extends CQLTester @Test public void testSchemaChangeDuringPaging() { - SimpleStatement stmt = new SimpleStatement("DESCRIBE KEYSPACES"); - stmt.setFetchSize(1); - ResultSet rs = executeNet(ProtocolVersion.CURRENT, stmt); - Iterator iter = rs.iterator(); - assertTrue(iter.hasNext()); + SimpleStatement stmt = new SimpleStatement("DESCRIBE KEYSPACES"); + stmt.setFetchSize(1); + ResultSet rs = executeNet(ProtocolVersion.CURRENT, stmt); + Iterator iter = rs.iterator(); + assertTrue(iter.hasNext()); + iter.next(); + + createKeyspace("CREATE KEYSPACE %s WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1};"); + + try + { iter.next(); - - createKeyspace("CREATE KEYSPACE %s WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1};"); - - try - { - iter.next(); - fail("Expected InvalidQueryException"); - } - catch (InvalidQueryException e) - { - assertEquals(DescribeStatement.SCHEMA_CHANGED_WHILE_PAGING_MESSAGE, e.getMessage()); - } + fail("Expected InvalidQueryException"); + } + catch (InvalidQueryException e) + { + assertEquals(DescribeStatement.SCHEMA_CHANGED_WHILE_PAGING_MESSAGE, e.getMessage()); + } } @Test @@ -104,35 +106,35 @@ public class DescribeStatementTest extends CQLTester "LANGUAGE java " + "AS 'return \"Hello World\";'"); - for (String describeKeyword : new String[]{"DESCRIBE", "DESC"}) + for (String describeKeyword : new String[]{ "DESCRIBE", "DESC" }) { assertRowsNet(executeDescribeNet(describeKeyword + " FUNCTION " + fNonOverloaded), row(KEYSPACE_PER_TEST, "function", shortFunctionName(fNonOverloaded) + "()", "CREATE FUNCTION " + fNonOverloaded + "()\n" + - " CALLED ON NULL INPUT\n" + - " RETURNS int\n" + - " LANGUAGE java\n" + - " AS $$throw new RuntimeException();$$;")); + " CALLED ON NULL INPUT\n" + + " RETURNS int\n" + + " LANGUAGE java\n" + + " AS $$throw new RuntimeException();$$;")); assertRowsNet(executeDescribeNet(describeKeyword + " FUNCTION " + fOverloaded), row(KEYSPACE_PER_TEST, "function", shortFunctionName(fOverloaded) + "(int, ascii)", "CREATE FUNCTION " + fOverloaded + "(input int, other_in ascii)\n" + - " RETURNS NULL ON NULL INPUT\n" + - " RETURNS text\n" + - " LANGUAGE java\n" + - " AS $$return \"Hello World\";$$;"), + " RETURNS NULL ON NULL INPUT\n" + + " RETURNS text\n" + + " LANGUAGE java\n" + + " AS $$return \"Hello World\";$$;"), row(KEYSPACE_PER_TEST, "function", shortFunctionName(fOverloaded) + "(text, ascii)", "CREATE FUNCTION " + fOverloaded + "(input text, other_in ascii)\n" + - " RETURNS NULL ON NULL INPUT\n" + - " RETURNS text\n" + - " LANGUAGE java\n" + - " AS $$return \"Hello World\";$$;")); + " RETURNS NULL ON NULL INPUT\n" + + " RETURNS text\n" + + " LANGUAGE java\n" + + " AS $$return \"Hello World\";$$;")); assertRowsNet(executeDescribeNet(describeKeyword + " FUNCTIONS"), row(KEYSPACE_PER_TEST, @@ -164,37 +166,37 @@ public class DescribeStatementTest extends CQLTester String aNonDeterministic = createAggregate(KEYSPACE_PER_TEST, "int", format("CREATE AGGREGATE %%s(int) " + - "SFUNC %s " + - "STYPE int " + - "INITCOND 42", - shortFunctionName(fIntState))); + "SFUNC %s " + + "STYPE int " + + "INITCOND 42", + shortFunctionName(fIntState))); String aDeterministic = createAggregate(KEYSPACE_PER_TEST, "int", format("CREATE AGGREGATE %%s(int) " + - "SFUNC %s " + - "STYPE int " + - "FINALFUNC %s ", - shortFunctionName(fIntState), - shortFunctionName(fFinal))); + "SFUNC %s " + + "STYPE int " + + "FINALFUNC %s ", + shortFunctionName(fIntState), + shortFunctionName(fFinal))); - for (String describeKeyword : new String[]{"DESCRIBE", "DESC"}) + for (String describeKeyword : new String[]{ "DESCRIBE", "DESC" }) { assertRowsNet(executeDescribeNet(describeKeyword + " AGGREGATE " + aNonDeterministic), row(KEYSPACE_PER_TEST, "aggregate", shortFunctionName(aNonDeterministic) + "(int)", "CREATE AGGREGATE " + aNonDeterministic + "(int)\n" + - " SFUNC " + shortFunctionName(fIntState) + "\n" + - " STYPE int\n" + - " INITCOND 42;")); + " SFUNC " + shortFunctionName(fIntState) + "\n" + + " STYPE int\n" + + " INITCOND 42;")); assertRowsNet(executeDescribeNet(describeKeyword + " AGGREGATE " + aDeterministic), row(KEYSPACE_PER_TEST, "aggregate", shortFunctionName(aDeterministic) + "(int)", "CREATE AGGREGATE " + aDeterministic + "(int)\n" + - " SFUNC " + shortFunctionName(fIntState) + "\n" + - " STYPE int\n" + - " FINALFUNC " + shortFunctionName(fFinal) + ";")); + " SFUNC " + shortFunctionName(fIntState) + "\n" + + " STYPE int\n" + + " FINALFUNC " + shortFunctionName(fFinal) + ";")); assertRowsNet(executeDescribeNet(describeKeyword + " AGGREGATES"), row(KEYSPACE_PER_TEST, "aggregate", @@ -216,35 +218,35 @@ public class DescribeStatementTest extends CQLTester "LANGUAGE java " + "AS 'throw new RuntimeException();';"); - assertRowsNet(executeDescribeNet("DESCRIBE FUNCTION " + function), - row(KEYSPACE_PER_TEST, - "function", - shortFunctionName(function) + "(tuple, list>>, tuple>, text>)", - "CREATE FUNCTION " + function + "(t tuple, l list>>, nt tuple>, text>)\n" + - " CALLED ON NULL INPUT\n" + - " RETURNS tuple\n" + - " LANGUAGE java\n" + - " AS $$throw new RuntimeException();$$;")); + assertRowsNet(executeDescribeNet("DESCRIBE FUNCTION " + function), + row(KEYSPACE_PER_TEST, + "function", + shortFunctionName(function) + "(tuple, list>>, tuple>, text>)", + "CREATE FUNCTION " + function + "(t tuple, l list>>, nt tuple>, text>)\n" + + " CALLED ON NULL INPUT\n" + + " RETURNS tuple\n" + + " LANGUAGE java\n" + + " AS $$throw new RuntimeException();$$;")); } @Test public void testDescribeVirtualTables() throws Throwable { - assertRowsNet(executeDescribeNet("DESCRIBE ONLY KEYSPACE system_virtual_schema;"), + assertRowsNet(executeDescribeNet("DESCRIBE ONLY KEYSPACE system_virtual_schema;"), row("system_virtual_schema", "keyspace", "system_virtual_schema", - "/*\n" + + "/*\n" + "Warning: Keyspace system_virtual_schema is a virtual keyspace and cannot be recreated with CQL.\n" + "Structure, for reference:\n" + "VIRTUAL KEYSPACE system_virtual_schema;\n" + "*/")); - assertRowsNet(executeDescribeNet("DESCRIBE TABLE system_virtual_schema.columns;"), + assertRowsNet(executeDescribeNet("DESCRIBE TABLE system_virtual_schema.columns;"), row("system_virtual_schema", "table", "columns", - "/*\n" + + "/*\n" + "Warning: Table system_virtual_schema.columns is a virtual table and cannot be recreated with CQL.\n" + "Structure, for reference:\n" + "VIRTUAL TABLE system_virtual_schema.columns (\n" + @@ -282,23 +284,23 @@ public class DescribeStatementTest extends CQLTester // Test describe schema Object[][] testSchemaOutput = rows( - row(KEYSPACE, "keyspace", KEYSPACE, - "CREATE KEYSPACE " + KEYSPACE + - " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}" + - " AND durable_writes = true;"), - row(KEYSPACE_PER_TEST, "keyspace", KEYSPACE_PER_TEST, - "CREATE KEYSPACE " + KEYSPACE_PER_TEST + - " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}" + - " AND durable_writes = true;"), - row("test", "keyspace", "test", keyspaceOutput()), - row("test", "table", "has_all_types", allTypesTable()), - row("test", "table", "\"Test\"", testTableOutput()), - row("test", "index", "\"Test_col_idx\"", indexOutput("\"Test_col_idx\"", "\"Test\"", "col")), - row("test", "index", "\"Test_val_idx\"", indexOutput("\"Test_val_idx\"", "\"Test\"", "val")), - row("test", "table", "users", userTableOutput()), - row("test", "index", "myindex", indexOutput("myindex", "users", "age")), - row("test", "table", "users_mv", usersMvTableOutput()), - row("test", "materialized_view", "users_by_state", usersByStateMvOutput())); + row(KEYSPACE, "keyspace", KEYSPACE, + "CREATE KEYSPACE " + KEYSPACE + + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}" + + " AND durable_writes = true;"), + row(KEYSPACE_PER_TEST, "keyspace", KEYSPACE_PER_TEST, + "CREATE KEYSPACE " + KEYSPACE_PER_TEST + + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}" + + " AND durable_writes = true;"), + row("test", "keyspace", "test", keyspaceOutput()), + row("test", "table", "has_all_types", allTypesTable()), + row("test", "table", "\"Test\"", testTableOutput()), + row("test", "index", "\"Test_col_idx\"", indexOutput("\"Test_col_idx\"", "\"Test\"", "col")), + row("test", "index", "\"Test_val_idx\"", indexOutput("\"Test_val_idx\"", "\"Test\"", "val")), + row("test", "table", "users", userTableOutput()), + row("test", "index", "myindex", indexOutput("myindex", "users", "age")), + row("test", "table", "users_mv", usersMvTableOutput()), + row("test", "materialized_view", "users_by_state", usersByStateMvOutput())); assertRowsNet(executeDescribeNet(KEYSPACE_PER_TEST, "DESCRIBE SCHEMA"), testSchemaOutput); assertRowsNet(executeDescribeNet(KEYSPACE_PER_TEST, "DESC SCHEMA"), testSchemaOutput); @@ -315,7 +317,7 @@ public class DescribeStatementTest extends CQLTester row(VIRTUAL_SCHEMA, "keyspace", VIRTUAL_SCHEMA), row("test", "keyspace", "test")); - for (String describeKeyword : new String[]{"DESCRIBE", "DESC"}) + for (String describeKeyword : new String[]{ "DESCRIBE", "DESC" }) { assertRowsNet(executeDescribeNet(describeKeyword + " KEYSPACES"), testKeyspacesOutput); assertRowsNet(executeDescribeNet("test", describeKeyword + " KEYSPACES"), testKeyspacesOutput); @@ -334,7 +336,7 @@ public class DescribeStatementTest extends CQLTester row("test", "table", "users_mv", usersMvTableOutput()), row("test", "materialized_view", "users_by_state", usersByStateMvOutput())); - for (String describeKeyword : new String[]{"DESCRIBE", "DESC"}) + for (String describeKeyword : new String[]{ "DESCRIBE", "DESC" }) { assertRowsNet(executeDescribeNet(describeKeyword + " KEYSPACE test"), testKeyspaceOutput); assertRowsNet(executeDescribeNet(describeKeyword + " test"), testKeyspaceOutput); @@ -343,7 +345,7 @@ public class DescribeStatementTest extends CQLTester } // Test describe tables/table - for (String cmd : new String[]{"describe TABLES", "DESC tables"}) + for (String cmd : new String[]{ "describe TABLES", "DESC tables" }) assertRowsNet(executeDescribeNet("test", cmd), row("test", "table", "has_all_types"), row("test", "table", "\"Test\""), @@ -358,7 +360,7 @@ public class DescribeStatementTest extends CQLTester row("test", "index", "\"Test_val_idx\"", indexOutput("\"Test_val_idx\"", "\"Test\"", "val"))); testDescribeTable("test", "users", row("test", "table", "users", userTableOutput()), - row("test", "index", "myindex", indexOutput("myindex", "users", "age"))); + row("test", "index", "myindex", indexOutput("myindex", "users", "age"))); describeError("test", "DESCRIBE users2", "'users2' not found in keyspace 'test'"); describeError("DESCRIBE test.users2", "'users2' not found in keyspace 'test'"); @@ -384,18 +386,18 @@ public class DescribeStatementTest extends CQLTester private void testDescribeTable(String keyspace, String table, Object[]... rows) throws Throwable { - for (String describeKeyword : new String[]{"describe", "desc"}) + for (String describeKeyword : new String[]{ "describe", "desc" }) { - for (String cmd : new String[]{describeKeyword + " table " + keyspace + "." + table, - describeKeyword + " columnfamily " + keyspace + "." + table, - describeKeyword + " " + keyspace + "." + table}) + for (String cmd : new String[]{ describeKeyword + " table " + keyspace + "." + table, + describeKeyword + " columnfamily " + keyspace + "." + table, + describeKeyword + " " + keyspace + "." + table }) { assertRowsNet(executeDescribeNet(cmd), rows); } - for (String cmd : new String[]{describeKeyword + " table " + table, - describeKeyword + " columnfamily " + table, - describeKeyword + " " + table}) + for (String cmd : new String[]{ describeKeyword + " table " + table, + describeKeyword + " columnfamily " + table, + describeKeyword + " " + table }) { assertRowsNet(executeDescribeNet(keyspace, cmd), rows); } @@ -404,16 +406,16 @@ public class DescribeStatementTest extends CQLTester private void testDescribeIndex(String keyspace, String index, Object[]... rows) throws Throwable { - for (String describeKeyword : new String[]{"describe", "desc"}) + for (String describeKeyword : new String[]{ "describe", "desc" }) { - for (String cmd : new String[]{describeKeyword + " index " + keyspace + "." + index, - describeKeyword + " " + keyspace + "." + index}) + for (String cmd : new String[]{ describeKeyword + " index " + keyspace + "." + index, + describeKeyword + " " + keyspace + "." + index }) { assertRowsNet(executeDescribeNet(cmd), rows); } - for (String cmd : new String[]{describeKeyword + " index " + index, - describeKeyword + " " + index}) + for (String cmd : new String[]{ describeKeyword + " index " + index, + describeKeyword + " " + index }) { assertRowsNet(executeDescribeNet(keyspace, cmd), rows); } @@ -422,16 +424,16 @@ public class DescribeStatementTest extends CQLTester private void testDescribeMaterializedView(String keyspace, String view, Object[]... rows) throws Throwable { - for (String describeKeyword : new String[]{"describe", "desc"}) + for (String describeKeyword : new String[]{ "describe", "desc" }) { - for (String cmd : new String[]{describeKeyword + " materialized view " + keyspace + "." + view, - describeKeyword + " " + keyspace + "." + view}) + for (String cmd : new String[]{ describeKeyword + " materialized view " + keyspace + "." + view, + describeKeyword + " " + keyspace + "." + view }) { assertRowsNet(executeDescribeNet(cmd), rows); } - for (String cmd : new String[]{describeKeyword + " materialized view " + view, - describeKeyword + " " + view}) + for (String cmd : new String[]{ describeKeyword + " materialized view " + view, + describeKeyword + " " + view }) { assertRowsNet(executeDescribeNet(keyspace, cmd), rows); } @@ -441,12 +443,12 @@ public class DescribeStatementTest extends CQLTester @Test public void testDescribeCluster() throws Throwable { - for (String describeKeyword : new String[]{"DESCRIBE", "DESC"}) + for (String describeKeyword : new String[]{ "DESCRIBE", "DESC" }) { assertRowsNet(executeDescribeNet(describeKeyword + " CLUSTER"), - row("Test Cluster", - "ByteOrderedPartitioner", - DatabaseDescriptor.getEndpointSnitch().getClass().getName())); + row("Test Cluster", + "ByteOrderedPartitioner", + DatabaseDescriptor.getEndpointSnitch().getClass().getName())); assertRowsNet(executeDescribeNet("system_virtual_schema", describeKeyword + " CLUSTER"), row("Test Cluster", @@ -535,13 +537,13 @@ public class DescribeStatementTest extends CQLTester ") WITH CLUSTERING ORDER BY (ck1 ASC, ck2 DESC)\n" + " AND " + tableParametersCql(); - String mvCreateStatement ="CREATE MATERIALIZED VIEW " + KEYSPACE_PER_TEST + ".mv AS\n" + - " SELECT *\n" + - " FROM " + KEYSPACE_PER_TEST + "." + table + "\n" + - " WHERE pk2 IS NOT NULL AND pk1 IS NOT NULL AND ck2 IS NOT NULL AND ck1 IS NOT NULL\n" + - " PRIMARY KEY ((pk2, pk1), ck2, ck1)\n" + - " WITH CLUSTERING ORDER BY (ck2 DESC, ck1 ASC)\n" + - " AND " + mvParametersCql(); + String mvCreateStatement = "CREATE MATERIALIZED VIEW " + KEYSPACE_PER_TEST + ".mv AS\n" + + " SELECT *\n" + + " FROM " + KEYSPACE_PER_TEST + "." + table + "\n" + + " WHERE pk2 IS NOT NULL AND pk1 IS NOT NULL AND ck2 IS NOT NULL AND ck1 IS NOT NULL\n" + + " PRIMARY KEY ((pk2, pk1), ck2, ck1)\n" + + " WITH CLUSTERING ORDER BY (ck2 DESC, ck1 ASC)\n" + + " AND " + mvParametersCql(); try { @@ -620,7 +622,7 @@ public class DescribeStatementTest extends CQLTester execute("DROP MATERIALIZED VIEW " + KEYSPACE_PER_TEST + ".mv"); } } - + @Test public void testDescribeMissingKeyspace() throws Throwable { @@ -671,31 +673,31 @@ public class DescribeStatementTest extends CQLTester assertRowsNet(executeDescribeNet(KEYSPACE_PER_TEST, "DESCRIBE TYPE " + type2), row(KEYSPACE_PER_TEST, "type", type2, "CREATE TYPE " + KEYSPACE_PER_TEST + "." + type2 + " (\n" + - " x text,\n" + - " y text\n" + - ");")); + " x text,\n" + + " y text\n" + + ");")); assertRowsNet(executeDescribeNet(KEYSPACE_PER_TEST, "DESCRIBE TYPE " + type1), row(KEYSPACE_PER_TEST, "type", type1, "CREATE TYPE " + KEYSPACE_PER_TEST + "." + type1 + " (\n" + - " a int,\n" + - " b frozen<" + type3 + ">\n" + - ");")); + " a int,\n" + + " b frozen<" + type3 + ">\n" + + ");")); assertRowsNet(executeDescribeNet(KEYSPACE_PER_TEST, "DESCRIBE KEYSPACE " + KEYSPACE_PER_TEST), row(KEYSPACE_PER_TEST, "keyspace", KEYSPACE_PER_TEST, "CREATE KEYSPACE " + KEYSPACE_PER_TEST + - " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}" + - " AND durable_writes = true;"), + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}" + + " AND durable_writes = true;"), row(KEYSPACE_PER_TEST, "type", type2, "CREATE TYPE " + KEYSPACE_PER_TEST + "." + type2 + " (\n" + - " x text,\n" + - " y text\n" + - ");"), + " x text,\n" + + " y text\n" + + ");"), row(KEYSPACE_PER_TEST, "type", type3, "CREATE TYPE " + KEYSPACE_PER_TEST + "." + type3 + " (\n" + - " a text,\n" + - " b frozen<" + type2 + ">\n" + - ");"), + " a text,\n" + + " b frozen<" + type2 + ">\n" + + ");"), row(KEYSPACE_PER_TEST, "type", type1, "CREATE TYPE " + KEYSPACE_PER_TEST + "." + type1 + " (\n" + - " a int,\n" + - " b frozen<" + type3 + ">\n" + - ");")); + " a int,\n" + + " b frozen<" + type3 + ">\n" + + ");")); } finally { @@ -707,26 +709,26 @@ public class DescribeStatementTest extends CQLTester /** * Tests for the error reported in CASSANDRA-9064 by: - * + *

* - creating the table described in the bug report, using LCS, * - DESCRIBE-ing that table via cqlsh, then DROPping it, * - running the output of the DESCRIBE statement as a CREATE TABLE statement, and * - inserting a value into the table. - * + *

* The final two steps of the test should not fall down. If one does, that * indicates the output of DESCRIBE is not a correct CREATE TABLE statement. */ @Test public void testDescribeRoundtrip() throws Throwable { - for (String withInternals : new String[]{"", " WITH INTERNALS"}) + for (String withInternals : new String[]{ "", " WITH INTERNALS" }) { String table = createTable(KEYSPACE_PER_TEST, "CREATE TABLE %s (key int PRIMARY KEY) WITH compaction = {'class': 'LeveledCompactionStrategy'}"); String output = executeDescribeNet(KEYSPACE_PER_TEST, "DESCRIBE TABLE " + table + withInternals).all().get(0).getString("create_statement"); executeDescribeNet(KEYSPACE_PER_TEST, "CREATE MATERIALIZED VIEW " + table + "_view AS SELECT key FROM " + table - + " WHERE key IS NOT NULL PRIMARY KEY(key)"); + + " WHERE key IS NOT NULL PRIMARY KEY(key)"); String mvCreateView = executeDescribeNet(KEYSPACE_PER_TEST, "DESCRIBE MATERIALIZED VIEW " + table + "_view").all().get(0).getString("create_statement"); @@ -875,7 +877,8 @@ public class DescribeStatementTest extends CQLTester @Test public void testDescMaterializedViewShouldNotOmitQuotations() throws Throwable { - try{ + try + { execute("CREATE KEYSPACE testWithKeywords WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1};"); execute("CREATE TABLE testWithKeywords.users_mv (username varchar, password varchar, gender varchar, session_token varchar, " + "state varchar, birth_year bigint, \"token\" text, PRIMARY KEY (\"token\"));"); @@ -905,14 +908,14 @@ public class DescribeStatementTest extends CQLTester final String functionName = KEYSPACE_PER_TEST + ".\"token\""; createFunctionOverload(functionName, - "int, ascii", - "CREATE FUNCTION " + functionName + " (\"token\" int, other_in ascii) " + - "RETURNS NULL ON NULL INPUT " + - "RETURNS text " + - "LANGUAGE java " + - "AS 'return \"Hello World\";'"); + "int, ascii", + "CREATE FUNCTION " + functionName + " (\"token\" int, other_in ascii) " + + "RETURNS NULL ON NULL INPUT " + + "RETURNS text " + + "LANGUAGE java " + + "AS 'return \"Hello World\";'"); - for (String describeKeyword : new String[]{"DESCRIBE", "DESC"}) + for (String describeKeyword : new String[]{ "DESCRIBE", "DESC" }) { assertRowsNet(executeDescribeNet(describeKeyword + " FUNCTION " + functionName), @@ -929,24 +932,24 @@ public class DescribeStatementTest extends CQLTester final String aggregationFunctionName = KEYSPACE_PER_TEST + ".\"token\""; final String aggregationName = KEYSPACE_PER_TEST + ".\"token\""; createFunctionOverload(aggregationName, - "int, int", - "CREATE FUNCTION " + aggregationFunctionName + " (\"token\" int, add_to int) " + - "CALLED ON NULL INPUT " + - "RETURNS int " + - "LANGUAGE java " + - "AS 'return token + add_to;'"); + "int, int", + "CREATE FUNCTION " + aggregationFunctionName + " (\"token\" int, add_to int) " + + "CALLED ON NULL INPUT " + + "RETURNS int " + + "LANGUAGE java " + + "AS 'return token + add_to;'"); String aggregate = createAggregate(KEYSPACE_PER_TEST, - "int", - format("CREATE AGGREGATE %%s(int) " + - "SFUNC %s " + - "STYPE int " + - "INITCOND 42", - shortFunctionName(aggregationFunctionName))); + "int", + format("CREATE AGGREGATE %%s(int) " + + "SFUNC %s " + + "STYPE int " + + "INITCOND 42", + shortFunctionName(aggregationFunctionName))); - for (String describeKeyword : new String[]{"DESCRIBE", "DESC"}) + for (String describeKeyword : new String[]{ "DESCRIBE", "DESC" }) { assertRowsNet(executeDescribeNet(describeKeyword + " AGGREGATE " + aggregate), row(KEYSPACE_PER_TEST, @@ -1030,12 +1033,12 @@ public class DescribeStatementTest extends CQLTester private static String testTableOutput() { return "CREATE TABLE test.\"Test\" (\n" + - " id int,\n" + - " col int,\n" + - " val text,\n" + - " PRIMARY KEY (id, col)\n" + - ") WITH CLUSTERING ORDER BY (col ASC)\n" + - " AND " + tableParametersCql(); + " id int,\n" + + " col int,\n" + + " val text,\n" + + " PRIMARY KEY (id, col)\n" + + ") WITH CLUSTERING ORDER BY (col ASC)\n" + + " AND " + tableParametersCql(); } private static String tableParametersCql() diff --git a/test/unit/org/apache/cassandra/schema/SchemaTest.java b/test/unit/org/apache/cassandra/schema/SchemaTest.java index 1c8dc47c28..81f409f477 100644 --- a/test/unit/org/apache/cassandra/schema/SchemaTest.java +++ b/test/unit/org/apache/cassandra/schema/SchemaTest.java @@ -19,8 +19,17 @@ package org.apache.cassandra.schema; import java.io.IOException; +import java.time.Duration; import java.util.Arrays; import java.util.Collection; +import java.util.UUID; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.ForkJoinTask; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.junit.BeforeClass; import org.junit.Test; @@ -31,7 +40,9 @@ import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.db.Mutation; import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.utils.FBUtilities; +import org.awaitility.Awaitility; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; @@ -105,6 +116,105 @@ public class SchemaTest assertNull(Schema.instance.getKeyspaceInstance("test")); } + /** + * This test checks that the schema version is updated only after the entire schema change is processed. + * In particular, we expect that {@link Schema#getVersion()} returns stale schema version as it was before + * the change, but it does not block. On the other hand, {@link Schema#getDistributedSchemaBlocking()} should block + * until the schema change is processed and return the new schema version. + */ + @Test + public void testGettingSchemaVersion() + { + Duration timeout = Duration.ofMillis(DatabaseDescriptor.getReadRpcTimeout(TimeUnit.MILLISECONDS)); + + // the listener with a barrier will let us control the schema change process + CyclicBarrier barrier = new CyclicBarrier(2); + SchemaChangeListener listener = new SchemaChangeListener() + { + @Override + public void onCreateKeyspace(KeyspaceMetadata keyspace) + { + try + { + barrier.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + catch (BrokenBarrierException e) + { + throw new RuntimeException(e); + } + } + }; + + String suffix = String.valueOf(System.currentTimeMillis()); + try + { + Schema.instance.registerListener(listener); + + // initial schema version - we will expect it not to change until the entire schema change is processed + UUID v0 = Schema.instance.getDistributedSchemaBlocking().getVersion(); + + // a multi-step schema transformation + SchemaTransformation transformation = schema -> { + schema = schema.withAddedOrReplaced(KeyspaceMetadata.create("test1" + suffix, KeyspaceParams.simple(1))); + schema = schema.withAddedOrReplaced(KeyspaceMetadata.create("test2" + suffix, KeyspaceParams.simple(1))); + schema = schema.withAddedOrReplaced(KeyspaceMetadata.create("test3" + suffix, KeyspaceParams.simple(1))); + return schema; + }; + + // schema change is executed async because it is expected to wait for the barrier + ForkJoinTask resultFuture = ForkJoinPool.commonPool().submit(() -> Schema.instance.transform(transformation, true)); + + // let the first statement execute + barrier.await(timeout.toMillis(), TimeUnit.MILLISECONDS); + Awaitility.await().atMost(timeout).until(() -> Schema.instance.distributedKeyspaces().containsKeyspace("test1" + suffix)); + assertThat(Schema.instance.getVersion()).isEqualTo(v0); // we should be able to get version unsafely, and it should return stale schema version + assertThat(resultFuture.isDone()).isFalse(); // the schema change should not be done yet + + // let's query the schema version safely - in particular, this task should not finish until the schema change is done + ForkJoinTask futureSchemaVersion = ForkJoinPool.commonPool().submit(() -> Schema.instance.getDistributedSchemaBlocking().getVersion()); + + // let the second statement execute + barrier.await(timeout.toMillis(), TimeUnit.MILLISECONDS); + Awaitility.await().atMost(timeout).until(() -> Schema.instance.distributedKeyspaces().containsKeyspace("test2" + suffix)); + assertThat(Schema.instance.getVersion()).isEqualTo(v0); // we should be able to get version unsafely, and it should return stale schema version + assertThat(resultFuture.isDone()).isFalse(); // the schema change should not be done yet + assertThat(futureSchemaVersion.isDone()).isFalse(); // the schema version should not be available yet + + // let the third statement execute + barrier.await(timeout.toMillis(), TimeUnit.MILLISECONDS); + Awaitility.await().atMost(timeout).until(() -> Schema.instance.distributedKeyspaces().containsKeyspace("test3" + suffix)); + + SchemaTransformation.SchemaTransformationResult result = resultFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS); // the schema change should be done shortly + assertThat(futureSchemaVersion.isDone()).isTrue(); // the schema version should be available now + UUID v1 = futureSchemaVersion.get(); + assertThat(v1).isNotEqualTo(v0); // the schema version should be updated + assertThat(result).isNotNull(); + assertThat(Schema.instance.getVersion()).isEqualTo(v1); + assertThat(result.after.getVersion()).isEqualTo(v1); + assertThat(result.before.getVersion()).isEqualTo(v0); + } + catch (BrokenBarrierException | TimeoutException | ExecutionException e) + { + throw new RuntimeException(e); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + finally + { + Schema.instance.unregisterListener(listener); + barrier.reset(); + Schema.instance.transform(schema -> schema.without(Arrays.asList("test1" + suffix, "test2" + suffix, "test3" + suffix))); + } + } + private void saveKeyspaces() { Collection mutations = Arrays.asList(SchemaKeyspace.makeCreateKeyspaceMutation(KeyspaceMetadata.create("ks0", KeyspaceParams.simple(3)), FBUtilities.timestampMicros()).build(),