mirror of https://github.com/apache/cassandra
Merge branch 'cassandra-3.11' into trunk
This commit is contained in:
Andrés de la Peña
commit
2c11dfcb5b
|
|
@ -2,6 +2,8 @@
|
|||
* Test org.apache.cassandra.net.AsyncPromiseTest FAILED (CASSANDRA-16596)
|
||||
Merged from 3.11:
|
||||
* Ignore stale acks received in the shadow round (CASSANDRA-16588)
|
||||
Merged from 3.0:
|
||||
* Fix materialized view builders inserting truncated data (CASSANDRA-16567)
|
||||
|
||||
4.0-rc1
|
||||
* Allow for setting buffer max capacity to increase it dynamically as needed (CASSANDRA-16524)
|
||||
|
|
|
|||
|
|
@ -2241,6 +2241,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
|
|||
// position in the System keyspace.
|
||||
logger.info("Truncating {}.{}", keyspace.getName(), name);
|
||||
|
||||
viewManager.stopBuild();
|
||||
|
||||
final long truncatedAt;
|
||||
final CommitLogPosition replayAfter;
|
||||
|
||||
|
|
@ -2298,6 +2300,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
|
|||
};
|
||||
|
||||
runWithCompactionsDisabled(Executors.callable(truncateRunnable), true, true);
|
||||
|
||||
viewManager.build();
|
||||
|
||||
logger.info("Truncate of {}.{} is complete", keyspace.getName(), name);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -23,7 +23,6 @@ import java.util.concurrent.TimeUnit;
|
|||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import com.google.common.collect.ImmutableMap;
|
||||
import com.google.common.collect.Iterables;
|
||||
import com.google.common.collect.Iterators;
|
||||
import com.google.common.collect.PeekingIterator;
|
||||
|
|
@ -93,6 +92,16 @@ public class TableViews extends AbstractCollection<View>
|
|||
return Iterables.transform(views, view -> keyspace.getColumnFamilyStore(view.getDefinition().name()));
|
||||
}
|
||||
|
||||
public void build()
|
||||
{
|
||||
views.forEach(View::build);
|
||||
}
|
||||
|
||||
public void stopBuild()
|
||||
{
|
||||
views.forEach(View::stopBuild);
|
||||
}
|
||||
|
||||
public void forceBlockingFlush()
|
||||
{
|
||||
for (ColumnFamilyStore viewCfs : allViewsCfs())
|
||||
|
|
|
|||
|
|
@ -220,7 +220,7 @@ public class View
|
|||
{
|
||||
if (builder != null)
|
||||
{
|
||||
logger.debug("Stopping current view builder due to schema change");
|
||||
logger.debug("Stopping current view builder due to schema change or truncate");
|
||||
builder.stop();
|
||||
builder = null;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,11 +18,10 @@
|
|||
|
||||
package org.apache.cassandra.cql3;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import com.google.common.util.concurrent.Uninterruptibles;
|
||||
|
|
@ -32,12 +31,12 @@ import org.junit.After;
|
|||
import org.junit.Before;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import com.datastax.driver.core.ResultSet;
|
||||
import com.datastax.driver.core.Row;
|
||||
import com.datastax.driver.core.exceptions.InvalidQueryException;
|
||||
import com.datastax.driver.core.exceptions.OperationTimedOutException;
|
||||
import org.apache.cassandra.concurrent.SEPExecutor;
|
||||
import org.apache.cassandra.concurrent.Stage;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.db.view.View;
|
||||
|
|
@ -48,13 +47,25 @@ import org.apache.cassandra.db.Keyspace;
|
|||
import org.apache.cassandra.db.SystemKeyspace;
|
||||
import org.apache.cassandra.db.compaction.CompactionManager;
|
||||
import org.apache.cassandra.service.ClientWarn;
|
||||
import org.apache.cassandra.transport.ProtocolVersion;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
import org.awaitility.Awaitility;
|
||||
import org.jboss.byteman.contrib.bmunit.BMRule;
|
||||
import org.jboss.byteman.contrib.bmunit.BMRules;
|
||||
import org.jboss.byteman.contrib.bmunit.BMUnitRunner;
|
||||
|
||||
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
@RunWith(BMUnitRunner.class)
|
||||
public class ViewTest extends CQLTester
|
||||
{
|
||||
ProtocolVersion protocolVersion = ProtocolVersion.V4;
|
||||
/** Latch used by {@link #testTruncateWhileBuilding()} Byteman injections. */
|
||||
@SuppressWarnings("unused")
|
||||
private static final CountDownLatch blockViewBuild = new CountDownLatch(1);
|
||||
|
||||
private final List<String> views = new ArrayList<>();
|
||||
|
||||
@BeforeClass
|
||||
|
|
@ -72,14 +83,14 @@ public class ViewTest extends CQLTester
|
|||
public void end() throws Throwable
|
||||
{
|
||||
for (String viewName : views)
|
||||
executeNet(protocolVersion, "DROP MATERIALIZED VIEW " + viewName);
|
||||
executeNet("DROP MATERIALIZED VIEW " + viewName);
|
||||
}
|
||||
|
||||
private void createView(String name, String query) throws Throwable
|
||||
{
|
||||
try
|
||||
{
|
||||
executeNet(protocolVersion, String.format(query, name));
|
||||
executeNet(String.format(query, name));
|
||||
// If exception is thrown, the view will not be added to the list; since it shouldn't have been created, this is
|
||||
// the desired behavior
|
||||
views.add(name);
|
||||
|
|
@ -94,12 +105,16 @@ public class ViewTest extends CQLTester
|
|||
|
||||
private void updateView(String query, Object... params) throws Throwable
|
||||
{
|
||||
executeNet(protocolVersion, query, params);
|
||||
while (!(((SEPExecutor) Stage.VIEW_MUTATION.executor()).getPendingTaskCount() == 0
|
||||
&& ((SEPExecutor) Stage.VIEW_MUTATION.executor()).getActiveTaskCount() == 0))
|
||||
{
|
||||
Thread.sleep(1);
|
||||
}
|
||||
executeNet(query, params);
|
||||
waitForViewMutations();
|
||||
}
|
||||
|
||||
private void waitForViewMutations()
|
||||
{
|
||||
Awaitility.await()
|
||||
.atMost(5, TimeUnit.MINUTES)
|
||||
.until(() -> Stage.VIEW_MUTATION.executor().getPendingTaskCount() == 0
|
||||
&& Stage.VIEW_MUTATION.executor().getActiveTaskCount() == 0);
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -129,7 +144,7 @@ public class ViewTest extends CQLTester
|
|||
createTable("CREATE TABLE %s (k1 int, c1 int, c2 int, v1 int, v2 int, PRIMARY KEY (k1, c1, c2))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("view1",
|
||||
"CREATE MATERIALIZED VIEW view1 AS SELECT * FROM %%s WHERE k1 IS NOT NULL AND c1 IS NOT NULL AND c2 IS NOT NULL PRIMARY KEY (k1, c2, c1)");
|
||||
|
|
@ -166,7 +181,7 @@ public class ViewTest extends CQLTester
|
|||
createTable("CREATE TABLE %s (k1 int, c1 int , val int, PRIMARY KEY (k1, c1))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("view1", "CREATE MATERIALIZED VIEW view1 AS SELECT k1, c1, val FROM %%s WHERE k1 IS NOT NULL AND c1 IS NOT NULL AND val IS NOT NULL PRIMARY KEY (val, k1, c1)");
|
||||
|
||||
|
|
@ -188,7 +203,7 @@ public class ViewTest extends CQLTester
|
|||
createTable("CREATE TABLE %s (k1 int, c1 int , val int, PRIMARY KEY (k1, c1))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("view1", "CREATE MATERIALIZED VIEW view1 AS SELECT val, k1, c1 FROM %%s WHERE k1 IS NOT NULL AND c1 IS NOT NULL AND val IS NOT NULL PRIMARY KEY (val, k1, c1)");
|
||||
|
||||
|
|
@ -200,7 +215,7 @@ public class ViewTest extends CQLTester
|
|||
createTable("CREATE TABLE %s (k1 int, c1 int , val int, PRIMARY KEY (k1, c1))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("view1", "CREATE MATERIALIZED VIEW view1 AS SELECT k1, c1, val FROM %%s WHERE k1 IS NOT NULL AND c1 IS NOT NULL AND val IS NOT NULL PRIMARY KEY (val, k1, c1)");
|
||||
|
||||
|
|
@ -226,7 +241,7 @@ public class ViewTest extends CQLTester
|
|||
"PRIMARY KEY((k, asciival)))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
// Must include "IS NOT NULL" for primary keys
|
||||
try
|
||||
|
|
@ -286,7 +301,7 @@ public class ViewTest extends CQLTester
|
|||
"PRIMARY KEY(k,c))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
try
|
||||
{
|
||||
|
|
@ -340,7 +355,7 @@ public class ViewTest extends CQLTester
|
|||
"PRIMARY KEY(k,c))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv_tstest", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE val IS NOT NULL AND k IS NOT NULL AND c IS NOT NULL PRIMARY KEY (val,k,c)");
|
||||
|
||||
|
|
@ -379,7 +394,7 @@ public class ViewTest extends CQLTester
|
|||
"val int)");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv_rctstest", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE k IS NOT NULL AND c IS NOT NULL PRIMARY KEY (k,c)");
|
||||
|
||||
|
|
@ -408,7 +423,7 @@ public class ViewTest extends CQLTester
|
|||
"count counter)");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
try
|
||||
{
|
||||
|
|
@ -428,7 +443,7 @@ public class ViewTest extends CQLTester
|
|||
"result duration)");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
try
|
||||
{
|
||||
|
|
@ -458,25 +473,25 @@ public class ViewTest extends CQLTester
|
|||
createTable("CREATE TABLE %s (a int, b int, c int, d int, e int, PRIMARY KEY (a, b))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
Keyspace ks = Keyspace.open(keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL AND c IS NOT NULL PRIMARY KEY (c, a, b)");
|
||||
ks.getColumnFamilyStore("mv").disableAutoCompaction();
|
||||
|
||||
//Set initial values TS=0, leaving e null and verify view
|
||||
executeNet(protocolVersion, "INSERT INTO %s (a, b, c, d) VALUES (0, 0, 1, 0) USING TIMESTAMP 0");
|
||||
executeNet("INSERT INTO %s (a, b, c, d) VALUES (0, 0, 1, 0) USING TIMESTAMP 0");
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(0));
|
||||
|
||||
//update c's timestamp TS=2
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 2 SET c = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
executeNet("UPDATE %s USING TIMESTAMP 2 SET c = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(0));
|
||||
|
||||
if (flush)
|
||||
FBUtilities.waitOnFutures(ks.flush());
|
||||
|
||||
// change c's value and TS=3, tombstones c=1 and adds c=0 record
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 3 SET c = ? WHERE a = ? and b = ? ", 0, 0, 0);
|
||||
executeNet("UPDATE %s USING TIMESTAMP 3 SET c = ? WHERE a = ? and b = ? ", 0, 0, 0);
|
||||
if (flush)
|
||||
FBUtilities.waitOnFutures(ks.flush());
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0));
|
||||
|
|
@ -489,7 +504,7 @@ public class ViewTest extends CQLTester
|
|||
|
||||
|
||||
//change c's value back to 1 with TS=4, check we can see d
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 4 SET c = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
executeNet("UPDATE %s USING TIMESTAMP 4 SET c = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
if (flush)
|
||||
{
|
||||
ks.getColumnFamilyStore("mv").forceMajorCompaction();
|
||||
|
|
@ -500,7 +515,7 @@ public class ViewTest extends CQLTester
|
|||
|
||||
|
||||
//Add e value @ TS=1
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 1 SET e = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
executeNet("UPDATE %s USING TIMESTAMP 1 SET e = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
assertRows(execute("SELECT d,e from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(0, 1));
|
||||
|
||||
if (flush)
|
||||
|
|
@ -508,7 +523,7 @@ public class ViewTest extends CQLTester
|
|||
|
||||
|
||||
//Change d value @ TS=2
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 2 SET d = ? WHERE a = ? and b = ? ", 2, 0, 0);
|
||||
executeNet("UPDATE %s USING TIMESTAMP 2 SET d = ? WHERE a = ? and b = ? ", 2, 0, 0);
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(2));
|
||||
|
||||
if (flush)
|
||||
|
|
@ -516,16 +531,16 @@ public class ViewTest extends CQLTester
|
|||
|
||||
|
||||
//Change d value @ TS=3
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 3 SET d = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
executeNet("UPDATE %s USING TIMESTAMP 3 SET d = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(1));
|
||||
|
||||
|
||||
//Tombstone c
|
||||
executeNet(protocolVersion, "DELETE FROM %s WHERE a = ? and b = ?", 0, 0);
|
||||
executeNet("DELETE FROM %s WHERE a = ? and b = ?", 0, 0);
|
||||
assertRows(execute("SELECT d from mv"));
|
||||
|
||||
//Add back without D
|
||||
executeNet(protocolVersion, "INSERT INTO %s (a, b, c) VALUES (0, 0, 1)");
|
||||
executeNet("INSERT INTO %s (a, b, c) VALUES (0, 0, 1)");
|
||||
|
||||
//Make sure D doesn't pop back in.
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row((Object) null));
|
||||
|
|
@ -533,24 +548,24 @@ public class ViewTest extends CQLTester
|
|||
|
||||
//New partition
|
||||
// insert a row with timestamp 0
|
||||
executeNet(protocolVersion, "INSERT INTO %s (a, b, c, d, e) VALUES (?, ?, ?, ?, ?) USING TIMESTAMP 0", 1, 0, 0, 0, 0);
|
||||
executeNet("INSERT INTO %s (a, b, c, d, e) VALUES (?, ?, ?, ?, ?) USING TIMESTAMP 0", 1, 0, 0, 0, 0);
|
||||
|
||||
// overwrite pk and e with timestamp 1, but don't overwrite d
|
||||
executeNet(protocolVersion, "INSERT INTO %s (a, b, c, e) VALUES (?, ?, ?, ?) USING TIMESTAMP 1", 1, 0, 0, 0);
|
||||
executeNet("INSERT INTO %s (a, b, c, e) VALUES (?, ?, ?, ?) USING TIMESTAMP 1", 1, 0, 0, 0);
|
||||
|
||||
// delete with timestamp 0 (which should only delete d)
|
||||
executeNet(protocolVersion, "DELETE FROM %s USING TIMESTAMP 0 WHERE a = ? AND b = ?", 1, 0);
|
||||
executeNet("DELETE FROM %s USING TIMESTAMP 0 WHERE a = ? AND b = ?", 1, 0);
|
||||
assertRows(execute("SELECT a, b, c, d, e from mv WHERE c = ? and a = ? and b = ?", 0, 1, 0),
|
||||
row(1, 0, 0, null, 0)
|
||||
);
|
||||
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 2 SET c = ? WHERE a = ? AND b = ?", 1, 1, 0);
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 3 SET c = ? WHERE a = ? AND b = ?", 0, 1, 0);
|
||||
executeNet("UPDATE %s USING TIMESTAMP 2 SET c = ? WHERE a = ? AND b = ?", 1, 1, 0);
|
||||
executeNet("UPDATE %s USING TIMESTAMP 3 SET c = ? WHERE a = ? AND b = ?", 0, 1, 0);
|
||||
assertRows(execute("SELECT a, b, c, d, e from mv WHERE c = ? and a = ? and b = ?", 0, 1, 0),
|
||||
row(1, 0, 0, null, 0)
|
||||
);
|
||||
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 3 SET d = ? WHERE a = ? AND b = ?", 0, 1, 0);
|
||||
executeNet("UPDATE %s USING TIMESTAMP 3 SET d = ? WHERE a = ? AND b = ?", 0, 1, 0);
|
||||
assertRows(execute("SELECT a, b, c, d, e from mv WHERE c = ? and a = ? and b = ?", 0, 1, 0),
|
||||
row(1, 0, 0, 0, 0)
|
||||
);
|
||||
|
|
@ -568,7 +583,7 @@ public class ViewTest extends CQLTester
|
|||
"PRIMARY KEY (k, c))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
|
||||
for(int i = 0; i < 1024; i++)
|
||||
|
|
@ -597,7 +612,7 @@ public class ViewTest extends CQLTester
|
|||
")");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv_test1", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE textval2 IS NOT NULL AND k IS NOT NULL AND asciival IS NOT NULL AND bigintval IS NOT NULL AND textval1 IS NOT NULL PRIMARY KEY ((textval2, k), asciival, bigintval, textval1)");
|
||||
|
||||
|
|
@ -644,7 +659,7 @@ public class ViewTest extends CQLTester
|
|||
")");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE textval1 IS NOT NULL AND k IS NOT NULL AND asciival IS NOT NULL AND bigintval IS NOT NULL PRIMARY KEY ((textval1, k), asciival, bigintval)");
|
||||
|
||||
|
|
@ -677,7 +692,7 @@ public class ViewTest extends CQLTester
|
|||
")");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE textval1 IS NOT NULL AND k IS NOT NULL AND asciival IS NOT NULL AND bigintval IS NOT NULL PRIMARY KEY ((textval1, k), asciival, bigintval)");
|
||||
|
||||
|
|
@ -710,7 +725,7 @@ public class ViewTest extends CQLTester
|
|||
TableMetadata metadata = currentTableMetadata();
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
for (ColumnMetadata def : new HashSet<>(metadata.columns()))
|
||||
{
|
||||
|
|
@ -829,7 +844,7 @@ public class ViewTest extends CQLTester
|
|||
"PRIMARY KEY (k))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE k IS NOT NULL AND intval IS NOT NULL PRIMARY KEY (intval, k)");
|
||||
|
||||
|
|
@ -849,7 +864,7 @@ public class ViewTest extends CQLTester
|
|||
createTable("CREATE TABLE %s (k int, intval int, listval frozen<list<tuple<text,text>>>, PRIMARY KEY (k))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv",
|
||||
"CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE k IS NOT NULL AND listval IS NOT NULL PRIMARY KEY (k, listval)");
|
||||
|
|
@ -885,7 +900,7 @@ public class ViewTest extends CQLTester
|
|||
"PRIMARY KEY (k))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE k IS NOT NULL AND intval IS NOT NULL PRIMARY KEY (intval, k)");
|
||||
|
||||
|
|
@ -911,7 +926,7 @@ public class ViewTest extends CQLTester
|
|||
"PRIMARY KEY (a, b))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT a, b, c FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (b, a)");
|
||||
|
||||
|
|
@ -949,7 +964,7 @@ public class ViewTest extends CQLTester
|
|||
"d int," +
|
||||
"PRIMARY KEY (a, b))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE c IS NOT NULL AND a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (c, a, b)");
|
||||
|
||||
|
|
@ -959,7 +974,7 @@ public class ViewTest extends CQLTester
|
|||
updateView("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 1, 1, 2);
|
||||
|
||||
Thread.sleep(TimeUnit.SECONDS.toMillis(5));
|
||||
List<Row> results = executeNet(protocolVersion, "SELECT d FROM mv WHERE c = 2 AND a = 1 AND b = 1").all();
|
||||
List<Row> results = executeNet("SELECT d FROM mv WHERE c = 2 AND a = 1 AND b = 1").all();
|
||||
Assert.assertEquals(1, results.size());
|
||||
Assert.assertTrue("There should be a null result given back due to ttl expiry", results.get(0).isNull(0));
|
||||
}
|
||||
|
|
@ -974,14 +989,14 @@ public class ViewTest extends CQLTester
|
|||
"d int," +
|
||||
"PRIMARY KEY (a, b))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE c IS NOT NULL AND a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (c, a, b)");
|
||||
|
||||
updateView("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?) USING TTL 3", 1, 1, 1, 1);
|
||||
|
||||
Thread.sleep(TimeUnit.SECONDS.toMillis(4));
|
||||
Assert.assertEquals(0, executeNet(protocolVersion, "SELECT * FROM mv WHERE c = 1 AND a = 1 AND b = 1").all().size());
|
||||
Assert.assertEquals(0, executeNet("SELECT * FROM mv WHERE c = 1 AND a = 1 AND b = 1").all().size());
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -994,14 +1009,14 @@ public class ViewTest extends CQLTester
|
|||
"d int," +
|
||||
"PRIMARY KEY (a, b))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE c IS NOT NULL AND a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (c, a, b)");
|
||||
|
||||
String table = keyspace() + "." + currentTable();
|
||||
updateView("DELETE FROM " + table + " USING TIMESTAMP 6 WHERE a = 1 AND b = 1;");
|
||||
updateView("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?) USING TIMESTAMP 3", 1, 1, 1, 1);
|
||||
Assert.assertEquals(0, executeNet(protocolVersion, "SELECT * FROM mv WHERE c = 1 AND a = 1 AND b = 1").all().size());
|
||||
Assert.assertEquals(0, executeNet("SELECT * FROM mv WHERE c = 1 AND a = 1 AND b = 1").all().size());
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -1013,7 +1028,7 @@ public class ViewTest extends CQLTester
|
|||
"c int," +
|
||||
"PRIMARY KEY (a, b))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE c IS NOT NULL AND a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (c, a, b)");
|
||||
|
||||
|
|
@ -1022,11 +1037,11 @@ public class ViewTest extends CQLTester
|
|||
updateView("INSERT INTO %s (a, b, c) VALUES (?, ?, ?) USING TIMESTAMP 1", 1, 1, i);
|
||||
}
|
||||
|
||||
ResultSet mvRows = executeNet(protocolVersion, "SELECT c FROM mv");
|
||||
List<Row> rows = executeNet(protocolVersion, "SELECT c FROM %s").all();
|
||||
ResultSet mvRows = executeNet("SELECT c FROM mv");
|
||||
List<Row> rows = executeNet("SELECT c FROM %s").all();
|
||||
Assert.assertEquals("There should be exactly one row in base", 1, rows.size());
|
||||
int expected = rows.get(0).getInt("c");
|
||||
assertRowsNet(protocolVersion, mvRows, row(expected));
|
||||
assertRowsNet(mvRows, row(expected));
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -1040,7 +1055,7 @@ public class ViewTest extends CQLTester
|
|||
"PRIMARY KEY (a, b, c))" +
|
||||
"WITH CLUSTERING ORDER BY (b ASC, c DESC)");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv1", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL AND c IS NOT NULL PRIMARY KEY (a, b, c) WITH CLUSTERING ORDER BY (b DESC, c ASC)");
|
||||
createView("mv2", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL AND c IS NOT NULL PRIMARY KEY (a, c, b) WITH CLUSTERING ORDER BY (c ASC, b ASC)");
|
||||
|
|
@ -1050,25 +1065,17 @@ public class ViewTest extends CQLTester
|
|||
updateView("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 1, 1, 1, 1);
|
||||
updateView("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 1, 2, 2, 2);
|
||||
|
||||
ResultSet mvRows = executeNet(protocolVersion, "SELECT b FROM mv1");
|
||||
assertRowsNet(protocolVersion, mvRows,
|
||||
row(2),
|
||||
row(1));
|
||||
ResultSet mvRows = executeNet("SELECT b FROM mv1");
|
||||
assertRowsNet(mvRows, row(2), row(1));
|
||||
|
||||
mvRows = executeNet(protocolVersion, "SELECT c FROM mv2");
|
||||
assertRowsNet(protocolVersion, mvRows,
|
||||
row(1),
|
||||
row(2));
|
||||
mvRows = executeNet("SELECT c FROM mv2");
|
||||
assertRowsNet(mvRows, row(1), row(2));
|
||||
|
||||
mvRows = executeNet(protocolVersion, "SELECT b FROM mv3");
|
||||
assertRowsNet(protocolVersion, mvRows,
|
||||
row(1),
|
||||
row(2));
|
||||
mvRows = executeNet("SELECT b FROM mv3");
|
||||
assertRowsNet(mvRows, row(1), row(2));
|
||||
|
||||
mvRows = executeNet(protocolVersion, "SELECT c FROM mv4");
|
||||
assertRowsNet(protocolVersion, mvRows,
|
||||
row(2),
|
||||
row(1));
|
||||
mvRows = executeNet("SELECT c FROM mv4");
|
||||
assertRowsNet(mvRows, row(2), row(1));
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -1079,7 +1086,7 @@ public class ViewTest extends CQLTester
|
|||
"b int," +
|
||||
"PRIMARY KEY (a, b))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv1", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (b, a)");
|
||||
|
||||
|
|
@ -1087,19 +1094,16 @@ public class ViewTest extends CQLTester
|
|||
updateView("INSERT INTO %s (a, b) VALUES (?, ?)", 1, 2);
|
||||
updateView("INSERT INTO %s (a, b) VALUES (?, ?)", 1, 3);
|
||||
|
||||
ResultSet mvRows = executeNet(protocolVersion, "SELECT a, b FROM mv1");
|
||||
assertRowsNet(protocolVersion, mvRows,
|
||||
row(1, 1),
|
||||
row(1, 2),
|
||||
row(1, 3));
|
||||
ResultSet mvRows = executeNet("SELECT a, b FROM mv1");
|
||||
assertRowsNet(mvRows, row(1, 1), row(1, 2), row(1, 3));
|
||||
|
||||
updateView(String.format("BEGIN UNLOGGED BATCH " +
|
||||
"DELETE FROM %s WHERE a = 1 AND b > 1 AND b < 3;" +
|
||||
"DELETE FROM %s WHERE a = 1;" +
|
||||
"APPLY BATCH", currentTable(), currentTable()));
|
||||
|
||||
mvRows = executeNet(protocolVersion, "SELECT a, b FROM mv1");
|
||||
assertRowsNet(protocolVersion, mvRows);
|
||||
mvRows = executeNet("SELECT a, b FROM mv1");
|
||||
assertRowsNet(mvRows);
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -1110,15 +1114,15 @@ public class ViewTest extends CQLTester
|
|||
"b int," +
|
||||
"PRIMARY KEY (a, b))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
// Cannot use SELECT *, as those are always handled by the includeAll shortcut in View.updateAffectsView
|
||||
createView("mv1", "CREATE MATERIALIZED VIEW %s AS SELECT a, b FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (b, a)");
|
||||
|
||||
updateView("INSERT INTO %s (a, b) VALUES (?, ?)", 1, 1);
|
||||
|
||||
ResultSet mvRows = executeNet(protocolVersion, "SELECT a, b FROM mv1");
|
||||
assertRowsNet(protocolVersion, mvRows, row(1, 1));
|
||||
ResultSet mvRows = executeNet("SELECT a, b FROM mv1");
|
||||
assertRowsNet(mvRows, row(1, 1));
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -1129,15 +1133,15 @@ public class ViewTest extends CQLTester
|
|||
"b int," +
|
||||
"PRIMARY KEY ((a, b)))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
// Cannot use SELECT *, as those are always handled by the includeAll shortcut in View.updateAffectsView
|
||||
createView("mv1", "CREATE MATERIALIZED VIEW %s AS SELECT a, b FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (b, a)");
|
||||
|
||||
updateView("INSERT INTO %s (a, b) VALUES (?, ?)", 1, 1);
|
||||
|
||||
ResultSet mvRows = executeNet(protocolVersion, "SELECT a, b FROM mv1");
|
||||
assertRowsNet(protocolVersion, mvRows, row(1, 1));
|
||||
ResultSet mvRows = executeNet("SELECT a, b FROM mv1");
|
||||
assertRowsNet(mvRows, row(1, 1));
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -1150,19 +1154,19 @@ public class ViewTest extends CQLTester
|
|||
"d int," +
|
||||
"PRIMARY KEY (a, b))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
createView("mv1", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL AND d IS NOT NULL PRIMARY KEY (a, d, b)");
|
||||
|
||||
updateView("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 0, 0, 0, 0);
|
||||
ResultSet mvRows = executeNet(protocolVersion, "SELECT a, d, b, c FROM mv1");
|
||||
assertRowsNet(protocolVersion, mvRows, row(0, 0, 0, 0));
|
||||
ResultSet mvRows = executeNet("SELECT a, d, b, c FROM mv1");
|
||||
assertRowsNet(mvRows, row(0, 0, 0, 0));
|
||||
|
||||
updateView("DELETE c FROM %s WHERE a = ? AND b = ?", 0, 0);
|
||||
mvRows = executeNet(protocolVersion, "SELECT a, d, b, c FROM mv1");
|
||||
assertRowsNet(protocolVersion, mvRows, row(0, 0, 0, null));
|
||||
mvRows = executeNet("SELECT a, d, b, c FROM mv1");
|
||||
assertRowsNet(mvRows, row(0, 0, 0, null));
|
||||
|
||||
updateView("DELETE d FROM %s WHERE a = ? AND b = ?", 0, 0);
|
||||
mvRows = executeNet(protocolVersion, "SELECT a, d, b FROM mv1");
|
||||
mvRows = executeNet("SELECT a, d, b FROM mv1");
|
||||
assertTrue(mvRows.isExhausted());
|
||||
}
|
||||
|
||||
|
|
@ -1176,19 +1180,19 @@ public class ViewTest extends CQLTester
|
|||
"d int," +
|
||||
"PRIMARY KEY (a, b))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
createView("mv1", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL AND d IS NOT NULL PRIMARY KEY (d, a, b)");
|
||||
|
||||
updateView("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 0, 0, 0, 0);
|
||||
ResultSet mvRows = executeNet(protocolVersion, "SELECT a, d, b, c FROM mv1");
|
||||
assertRowsNet(protocolVersion, mvRows, row(0, 0, 0, 0));
|
||||
ResultSet mvRows = executeNet("SELECT a, d, b, c FROM mv1");
|
||||
assertRowsNet(mvRows, row(0, 0, 0, 0));
|
||||
|
||||
updateView("DELETE c FROM %s WHERE a = ? AND b = ?", 0, 0);
|
||||
mvRows = executeNet(protocolVersion, "SELECT a, d, b, c FROM mv1");
|
||||
assertRowsNet(protocolVersion, mvRows, row(0, 0, 0, null));
|
||||
mvRows = executeNet("SELECT a, d, b, c FROM mv1");
|
||||
assertRowsNet(mvRows, row(0, 0, 0, null));
|
||||
|
||||
updateView("DELETE d FROM %s WHERE a = ? AND b = ?", 0, 0);
|
||||
mvRows = executeNet(protocolVersion, "SELECT a, d, b FROM mv1");
|
||||
mvRows = executeNet("SELECT a, d, b FROM mv1");
|
||||
assertTrue(mvRows.isExhausted());
|
||||
}
|
||||
|
||||
|
|
@ -1201,20 +1205,20 @@ public class ViewTest extends CQLTester
|
|||
"c map<int, text>," +
|
||||
"PRIMARY KEY (a))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
createView("mvmap", "CREATE MATERIALIZED VIEW %s AS SELECT a, b FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (b, a)");
|
||||
|
||||
updateView("INSERT INTO %s (a, b) VALUES (?, ?)", 0, 0);
|
||||
ResultSet mvRows = executeNet(protocolVersion, "SELECT a, b FROM mvmap WHERE b = ?", 0);
|
||||
assertRowsNet(protocolVersion, mvRows, row(0, 0));
|
||||
ResultSet mvRows = executeNet("SELECT a, b FROM mvmap WHERE b = ?", 0);
|
||||
assertRowsNet(mvRows, row(0, 0));
|
||||
|
||||
updateView("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 1, 1, map(1, "1"));
|
||||
mvRows = executeNet(protocolVersion, "SELECT a, b FROM mvmap WHERE b = ?", 1);
|
||||
assertRowsNet(protocolVersion, mvRows, row(1, 1));
|
||||
mvRows = executeNet("SELECT a, b FROM mvmap WHERE b = ?", 1);
|
||||
assertRowsNet(mvRows, row(1, 1));
|
||||
|
||||
updateView("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 0, 0, map(0, "0"));
|
||||
mvRows = executeNet(protocolVersion, "SELECT a, b FROM mvmap WHERE b = ?", 0);
|
||||
assertRowsNet(protocolVersion, mvRows, row(0, 0));
|
||||
mvRows = executeNet("SELECT a, b FROM mvmap WHERE b = ?", 0);
|
||||
assertRowsNet(mvRows, row(0, 0));
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -1253,7 +1257,7 @@ public class ViewTest extends CQLTester
|
|||
{
|
||||
createTable("CREATE TABLE %s (id1 int, id2 int, v1 text, v2 text, PRIMARY KEY (id1, id2))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv",
|
||||
"CREATE MATERIALIZED VIEW %s AS" +
|
||||
|
|
@ -1265,16 +1269,16 @@ public class ViewTest extends CQLTester
|
|||
|
||||
execute("INSERT INTO %s (id1, id2, v1, v2) VALUES (?, ?, ?, ?)", 0, 1, "foo", "bar");
|
||||
|
||||
assertRowsNet(protocolVersion, executeNet(protocolVersion, "SELECT * FROM %s"), row(0, 1, "foo", "bar"));
|
||||
assertRowsNet(protocolVersion, executeNet(protocolVersion, "SELECT * FROM mv"), row(0, "foo", 1, "bar"));
|
||||
assertRowsNet(executeNet("SELECT * FROM %s"), row(0, 1, "foo", "bar"));
|
||||
assertRowsNet(executeNet("SELECT * FROM mv"), row(0, "foo", 1, "bar"));
|
||||
|
||||
executeNet(protocolVersion, "UPDATE %s SET v1=? WHERE id1=? AND id2=?", null, 0, 1);
|
||||
assertRowsNet(protocolVersion, executeNet(protocolVersion, "SELECT * FROM %s"), row(0, 1, null, "bar"));
|
||||
assertRowsNet(protocolVersion, executeNet(protocolVersion, "SELECT * FROM mv"));
|
||||
executeNet("UPDATE %s SET v1=? WHERE id1=? AND id2=?", null, 0, 1);
|
||||
assertRowsNet(executeNet("SELECT * FROM %s"), row(0, 1, null, "bar"));
|
||||
assertRowsNet(executeNet("SELECT * FROM mv"));
|
||||
|
||||
executeNet(protocolVersion, "UPDATE %s SET v2=? WHERE id1=? AND id2=?", "rab", 0, 1);
|
||||
assertRowsNet(protocolVersion, executeNet(protocolVersion, "SELECT * FROM %s"), row(0, 1, null, "rab"));
|
||||
assertRowsNet(protocolVersion, executeNet(protocolVersion, "SELECT * FROM mv"));
|
||||
executeNet("UPDATE %s SET v2=? WHERE id1=? AND id2=?", "rab", 0, 1);
|
||||
assertRowsNet(executeNet("SELECT * FROM %s"), row(0, 1, null, "rab"));
|
||||
assertRowsNet(executeNet("SELECT * FROM mv"));
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -1282,7 +1286,7 @@ public class ViewTest extends CQLTester
|
|||
{
|
||||
createTable("CREATE TABLE %s (\"token\" int PRIMARY KEY, \"keyspace\" int)");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("mv",
|
||||
"CREATE MATERIALIZED VIEW %s AS" +
|
||||
|
|
@ -1293,8 +1297,8 @@ public class ViewTest extends CQLTester
|
|||
|
||||
execute("INSERT INTO %s (\"token\", \"keyspace\") VALUES (?, ?)", 0, 1);
|
||||
|
||||
assertRowsNet(protocolVersion, executeNet(protocolVersion, "SELECT * FROM %s"), row(0, 1));
|
||||
assertRowsNet(protocolVersion, executeNet(protocolVersion, "SELECT * FROM mv"), row(1, 0));
|
||||
assertRowsNet(executeNet("SELECT * FROM %s"), row(0, 1));
|
||||
assertRowsNet(executeNet("SELECT * FROM mv"), row(1, 0));
|
||||
}
|
||||
|
||||
public void testCreateMvWithTTL() throws Throwable
|
||||
|
|
@ -1305,7 +1309,7 @@ public class ViewTest extends CQLTester
|
|||
"val int) WITH default_time_to_live = 60");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
// Must NOT include "default_time_to_live" for Materialized View creation
|
||||
try
|
||||
|
|
@ -1331,7 +1335,7 @@ public class ViewTest extends CQLTester
|
|||
// Must NOT include "default_time_to_live" on alter Materialized View
|
||||
try
|
||||
{
|
||||
executeNet(protocolVersion, "ALTER MATERIALIZED VIEW %s WITH default_time_to_live = 30");
|
||||
executeNet("ALTER MATERIALIZED VIEW %s WITH default_time_to_live = 30");
|
||||
Assert.fail("Should fail if TTL is provided while altering materialized view");
|
||||
}
|
||||
catch (Exception e)
|
||||
|
|
@ -1348,7 +1352,7 @@ public class ViewTest extends CQLTester
|
|||
"PRIMARY KEY(k,c))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
CompactionManager.instance.setConcurrentViewBuilders(concurrentViewBuilders);
|
||||
CompactionManager.instance.setCoreCompactorThreads(1);
|
||||
|
|
@ -1433,7 +1437,7 @@ public class ViewTest extends CQLTester
|
|||
{
|
||||
createTable("CREATE TABLE %s (k int PRIMARY KEY, v int)");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
boolean enableMaterializedViews = DatabaseDescriptor.getEnableMaterializedViews();
|
||||
try
|
||||
|
|
@ -1458,7 +1462,7 @@ public class ViewTest extends CQLTester
|
|||
{
|
||||
createTable("CREATE TABLE %s (\"theKey\" int, \"theClustering_1\" int, \"theClustering_2\" int, \"theValue\" int, PRIMARY KEY (\"theKey\", \"theClustering_1\", \"theClustering_2\"))");
|
||||
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
|
||||
createView("view1", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE \"theKey\" IS NOT NULL AND \"theClustering_1\" IS NOT NULL AND \"theClustering_2\" IS NOT NULL AND \"theValue\" IS NOT NULL PRIMARY KEY (\"theKey\", \"theClustering_1\", \"theClustering_2\");");
|
||||
createView("view2", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE \"theKey\" IS NOT NULL AND (\"theClustering_1\", \"theClustering_2\") = (1, 2) AND \"theValue\" IS NOT NULL PRIMARY KEY (\"theKey\", \"theClustering_1\", \"theClustering_2\");");
|
||||
|
|
@ -1469,4 +1473,61 @@ public class ViewTest extends CQLTester
|
|||
row("\"theKey\" IS NOT NULL AND (\"theClustering_1\", \"theClustering_2\") = (1, 2) AND \"theValue\" IS NOT NULL"),
|
||||
row("token(\"theKey\") > token(1) AND \"theClustering_1\" = 1 AND \"theClustering_2\" > 2 AND \"theValue\" IS NOT NULL"));
|
||||
}
|
||||
|
||||
/**
|
||||
* Tests that truncating a table stops the ongoing builds of its materialized views,
|
||||
* so they don't write into the MV data that has been truncated in the base table.
|
||||
*
|
||||
* See CASSANDRA-16567 for further details.
|
||||
*/
|
||||
@Test
|
||||
@BMRules(rules = {
|
||||
@BMRule(name = "Block view builder tasks",
|
||||
targetClass = "ViewBuilderTask",
|
||||
targetMethod = "buildKey",
|
||||
action = "com.google.common.util.concurrent.Uninterruptibles.awaitUninterruptibly" +
|
||||
"(org.apache.cassandra.cql3.ViewTest.blockViewBuild);"),
|
||||
@BMRule(name = "Unblock view builder tasks",
|
||||
targetClass = "ColumnFamilyStore",
|
||||
targetMethod = "truncateBlocking",
|
||||
action = "org.apache.cassandra.cql3.ViewTest.blockViewBuild.countDown();")
|
||||
})
|
||||
public void testTruncateWhileBuilding() throws Throwable
|
||||
{
|
||||
createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY(k, c))");
|
||||
execute("USE " + keyspace());
|
||||
executeNet("USE " + keyspace());
|
||||
execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 0, 0);
|
||||
createView("mv",
|
||||
"CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s " +
|
||||
"WHERE k IS NOT NULL AND c IS NOT NULL AND v IS NOT NULL " +
|
||||
"PRIMARY KEY (v, c, k)");
|
||||
|
||||
// check that the delayed view builder tasks are either running or pending,
|
||||
// and that they haven't written anything yet
|
||||
assertThat(runningViewBuilds()).isPositive();
|
||||
assertFalse(SystemKeyspace.isViewBuilt(KEYSPACE, "mv"));
|
||||
waitForViewMutations();
|
||||
assertRows(execute("SELECT * FROM mv"));
|
||||
|
||||
// truncate the view, this should unblock the view builders, wait for their cancellation,
|
||||
// drop the sstables and, finally, start a new view build
|
||||
updateView("TRUNCATE %s");
|
||||
|
||||
// check that there aren't any rows after truncating
|
||||
assertRows(execute("SELECT * FROM mv"));
|
||||
|
||||
// check that the view builder tasks finish and that the view is still empty after that
|
||||
Awaitility.await().untilAsserted(() -> assertEquals(0, runningViewBuilds()));
|
||||
assertTrue(SystemKeyspace.isViewBuilt(KEYSPACE, "mv"));
|
||||
waitForViewMutations();
|
||||
assertRows(execute("SELECT * FROM mv"));
|
||||
}
|
||||
|
||||
private static int runningViewBuilds()
|
||||
{
|
||||
return Metrics.getThreadPoolMetrics("ViewBuildExecutor")
|
||||
.map(p -> p.activeTasks.getValue() + p.pendingTasks.getValue())
|
||||
.orElse(0);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue