natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-4.1' into cassandra-5.0 

* cassandra-4.1:
  Retrieve keyspaces metadata and schema version concistently in DescribeStatement
This commit is contained in:
Jacek Lewandowski 2023-10-31 11:15:45 +01:00
parent 69f5df0b00 9ad5ec9664
commit f2c46cbf3b
6 changed files with 301 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -30,6 +30,7 @@ Merged from 4.1:
* Internode legacy SSL storage port certificate is not hot reloaded on update (CASSANDRA-18681)
* Nodetool paxos-only repair is no longer incremental (CASSANDRA-18466)
Merged from 4.0:
* Retrieve keyspaces metadata and schema version concistently in DescribeStatement (CASSANDRA-18921)
* Gossip NPE due to shutdown event corrupting empty statuses (CASSANDRA-18913)
* Update hdrhistogram to 2.1.12 (CASSANDRA-18893)
* Improve performance of compactions when table does not have an index (CASSANDRA-18773)

19
src/java/org/apache/cassandra/cql3/statements/DescribeStatement.java
View File

@ -132,14 +132,13 @@ public abstract class DescribeStatement<T> 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<T> 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<? extends T> stream = describe(state.getClientState(), keyspaces);
@ -174,9 +173,7 @@ public abstract class DescribeStatement<T> 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);
}

18
src/java/org/apache/cassandra/schema/Schema.java
View File

@ -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.
* <p>
* 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()
{

2
src/java/org/apache/cassandra/schema/SchemaEvent.java
View File

@ -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())

325
test/unit/org/apache/cassandra/cql3/statements/DescribeStatementTest.java
View File

@ -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<Row> iter = rs.iterator();
assertTrue(iter.hasNext());
SimpleStatement stmt = new SimpleStatement("DESCRIBE KEYSPACES");
stmt.setFetchSize(1);
ResultSet rs = executeNet(ProtocolVersion.CURRENT, stmt);
Iterator<Row> 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<int>, list<frozen<tuple<int, text>>>, tuple<frozen<tuple<int, text>>, text>)",
"CREATE FUNCTION " + function + "(t tuple<int>, l list<frozen<tuple<int, text>>>, nt tuple<frozen<tuple<int, text>>, text>)\n" +
" CALLED ON NULL INPUT\n" +
" RETURNS tuple<int, text>\n" +
" LANGUAGE java\n" +
" AS $$throw new RuntimeException();$$;"));
assertRowsNet(executeDescribeNet("DESCRIBE FUNCTION " + function),
row(KEYSPACE_PER_TEST,
"function",
shortFunctionName(function) + "(tuple<int>, list<frozen<tuple<int, text>>>, tuple<frozen<tuple<int, text>>, text>)",
"CREATE FUNCTION " + function + "(t tuple<int>, l list<frozen<tuple<int, text>>>, nt tuple<frozen<tuple<int, text>>, text>)\n" +
" CALLED ON NULL INPUT\n" +
" RETURNS tuple<int, text>\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:
*
* <p>
* - 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.
*
* <p>
* 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()

110
test/unit/org/apache/cassandra/schema/SchemaTest.java
View File

@ -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<SchemaTransformation.SchemaTransformationResult> 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<UUID> 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<Mutation> mutations = Arrays.asList(SchemaKeyspace.makeCreateKeyspaceMutation(KeyspaceMetadata.create("ks0", KeyspaceParams.simple(3)), FBUtilities.timestampMicros()).build(),