You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bd...@apache.org on 2019/04/24 22:37:27 UTC
[cassandra] branch cassandra-3.0 updated: Fix mixed mode partition
range scans with limit
This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch cassandra-3.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-3.0 by this push:
new d27c3ad Fix mixed mode partition range scans with limit
d27c3ad is described below
commit d27c3ad0d2d006a5f156f0a2f2a24286d31c5069
Author: Blake Eggleston <bd...@gmail.com>
AuthorDate: Thu Apr 4 14:50:12 2019 -0700
Fix mixed mode partition range scans with limit
Patch by Blake Eggleston; Reviewed by Sam Tunnicliffe for CASSANDRA-15072
---
CHANGES.txt | 1 +
src/java/org/apache/cassandra/db/ReadCommand.java | 2 +
.../org/apache/cassandra/db/filter/DataLimits.java | 19 ++++
.../upgrade/CompactStorage2to3UpgradeTest.java | 102 +++++++++++++++++++++
4 files changed, 124 insertions(+)
diff --git a/CHANGES.txt b/CHANGES.txt
index b38ab06..4f76c70 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.19
+ * Fix mixed mode partition range scans with limit (CASSANDRA-15072)
* cassandra-stress works with frozen collections: list and set (CASSANDRA-14907)
* For nodetool listsnapshots output, put spaces between columns, and increase snapshot padding (CASSANDRA-14876)
* Fix handling FS errors on writing and reading flat files - LogTransaction and hints (CASSANDRA-15053)
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java
index fd453ef..b499daf 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -856,6 +856,8 @@ public abstract class ReadCommand implements ReadQuery
limits = DataLimits.distinctLimits(maxResults);
else if (compositesToGroup == -1)
limits = DataLimits.thriftLimits(maxResults, perPartitionLimit);
+ else if (metadata.isStaticCompactTable())
+ limits = DataLimits.legacyCompactStaticCqlLimits(maxResults);
else
limits = DataLimits.cqlLimits(maxResults);
diff --git a/src/java/org/apache/cassandra/db/filter/DataLimits.java b/src/java/org/apache/cassandra/db/filter/DataLimits.java
index 4c57a76..46a1c6d 100644
--- a/src/java/org/apache/cassandra/db/filter/DataLimits.java
+++ b/src/java/org/apache/cassandra/db/filter/DataLimits.java
@@ -80,6 +80,25 @@ public abstract class DataLimits
return new CQLLimits(cqlRowLimit);
}
+ // mixed mode partition range scans on compact storage tables without clustering columns coordinated by 2.x are
+ // returned as one (cql) row per cell, but we need to count each partition as a single row. So we just return a
+ // CQLLimits instance that doesn't count rows towards it's limit. See CASSANDRA-15072
+ public static DataLimits legacyCompactStaticCqlLimits(int cqlRowLimits)
+ {
+ return new CQLLimits(cqlRowLimits) {
+ public Counter newCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData, boolean enforceStrictLiveness)
+ {
+ return new CQLCounter(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData, enforceStrictLiveness) {
+ public Row applyToRow(Row row)
+ {
+ // noop: only count full partitions
+ return row;
+ }
+ };
+ }
+ };
+ }
+
public static DataLimits cqlLimits(int cqlRowLimit, int perPartitionLimit)
{
return new CQLLimits(cqlRowLimit, perPartitionLimit);
diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage2to3UpgradeTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage2to3UpgradeTest.java
new file mode 100644
index 0000000..5c45d52
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage2to3UpgradeTest.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.upgrade;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.ICoordinator;
+import org.apache.cassandra.distributed.impl.Versions;
+import org.apache.cassandra.distributed.test.DistributedTestBase;
+
+public class CompactStorage2to3UpgradeTest extends UpgradeTestBase
+{
+ @Test
+ public void multiColumn() throws Throwable
+ {
+ new TestCase()
+ .upgrade(Versions.Major.v22, Versions.Major.v30)
+ .setup(cluster -> {
+ assert cluster.size() == 3;
+ int rf = cluster.size() - 1;
+ assert rf == 2;
+ cluster.schemaChange("CREATE KEYSPACE ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + (cluster.size() - 1) + "};");
+ cluster.schemaChange("CREATE TABLE ks.tbl (pk int, v1 int, v2 text, PRIMARY KEY (pk)) WITH COMPACT STORAGE");
+ ICoordinator coordinator = cluster.coordinator(1);
+ // these shouldn't be replicated by the 3rd node
+ coordinator.execute("INSERT INTO ks.tbl (pk, v1, v2) VALUES (3, 3, '3')", ConsistencyLevel.ALL);
+ coordinator.execute("INSERT INTO ks.tbl (pk, v1, v2) VALUES (9, 9, '9')", ConsistencyLevel.ALL);
+ for (int i=0; i<cluster.size(); i++)
+ {
+ int nodeNum = i+1;
+ System.out.println(String.format("****** node %s: %s", nodeNum, cluster.get(nodeNum).config()));
+ }
+
+ })
+ .runAfterNodeUpgrade(((cluster, node) -> {
+ if (node != 2)
+ return;
+
+ Object[][] rows = cluster.coordinator(3).execute("SELECT * FROM ks.tbl LIMIT 2", ConsistencyLevel.ALL);
+ Object[][] expected = {
+ DistributedTestBase.row(9, 9, "9"),
+ DistributedTestBase.row(3, 3, "3")
+ };
+ DistributedTestBase.assertRows(rows, expected);
+
+ })).run();
+ }
+
+ @Test
+ public void singleColumn() throws Throwable
+ {
+ new TestCase()
+ .upgrade(Versions.Major.v22, Versions.Major.v30)
+ .setup(cluster -> {
+ assert cluster.size() == 3;
+ int rf = cluster.size() - 1;
+ assert rf == 2;
+ cluster.schemaChange("CREATE KEYSPACE ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + (cluster.size() - 1) + "};");
+ cluster.schemaChange("CREATE TABLE ks.tbl (pk int, v int, PRIMARY KEY (pk)) WITH COMPACT STORAGE");
+ ICoordinator coordinator = cluster.coordinator(1);
+ // these shouldn't be replicated by the 3rd node
+ coordinator.execute("INSERT INTO ks.tbl (pk, v) VALUES (3, 3)", ConsistencyLevel.ALL);
+ coordinator.execute("INSERT INTO ks.tbl (pk, v) VALUES (9, 9)", ConsistencyLevel.ALL);
+ for (int i=0; i<cluster.size(); i++)
+ {
+ int nodeNum = i+1;
+ System.out.println(String.format("****** node %s: %s", nodeNum, cluster.get(nodeNum).config()));
+ }
+
+ })
+ .runAfterNodeUpgrade(((cluster, node) -> {
+
+ if (node < 2)
+ return;
+
+ Object[][] rows = cluster.coordinator(3).execute("SELECT * FROM ks.tbl LIMIT 2", ConsistencyLevel.ALL);
+ Object[][] expected = {
+ DistributedTestBase.row(9, 9),
+ DistributedTestBase.row(3, 3)
+ };
+ DistributedTestBase.assertRows(rows, expected);
+
+ })).run();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org