diff --git a/CHANGES.txt b/CHANGES.txt index c04ad58029..9ce2972887 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -2,6 +2,7 @@ * Fixed nodetool cfstats printing index name twice (CASSANDRA-14903) * Add flag to disable SASI indexes, and warnings on creation (CASSANDRA-14866) Merged from 3.0: + * Fix mixed mode partition range scans with limit (CASSANDRA-15072) * cassandra-stress works with frozen collections: list and set (CASSANDRA-14907) * Fix handling FS errors on writing and reading flat files - LogTransaction and hints (CASSANDRA-15053) * Avoid double closing the iterator to avoid overcounting the number of requests (CASSANDRA-15058) diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 56fa9ccabf..bb968cf128 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -953,6 +953,8 @@ public abstract class ReadCommand extends MonitorableImpl 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 205d745ad7..c6918ebdaf 100644 --- a/src/java/org/apache/cassandra/db/filter/DataLimits.java +++ b/src/java/org/apache/cassandra/db/filter/DataLimits.java @@ -83,6 +83,25 @@ public abstract class DataLimits return cqlRowLimit == NO_LIMIT ? NONE : 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 cqlRowLimit == NO_LIMIT && perPartitionLimit == NO_LIMIT 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 0000000000..5c45d5237d --- /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 { + 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 { + + 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(); + } +}