diff --git a/CHANGES.txt b/CHANGES.txt index d37705c1e8..45ea128630 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -180,6 +180,7 @@ * Add the ability to disable bulk loading of SSTables (CASSANDRA-18781) * Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787) Merged from 5.0: + * SAI marks an index as non-empty when a partial partition/row modifications is flushed due to repair (CASSANDRA-20567) * SAI fails queries when multiple columns exist and a non-indexed column is a composite with a map (CASSANDRA-19891) * Avoid purging deletions in RowFilter when reconciliation is required (CASSANDRA-20541) * Fixed multiple single-node SAI query bugs relating to static columns (CASSANDRA-20338) diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java index a68efa120c..fbf2894187 100644 --- a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java @@ -52,6 +52,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy public static final String LEVEL_FANOUT_SIZE_OPTION = "fanout_size"; public static final String SINGLE_SSTABLE_UPLEVEL_OPTION = "single_sstable_uplevel"; public static final int DEFAULT_LEVEL_FANOUT_SIZE = 10; + public static final int DEFAULT_MAX_SSTABLE_SIZE_MIB = 160; @VisibleForTesting final LeveledManifest manifest; @@ -62,7 +63,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy public LeveledCompactionStrategy(ColumnFamilyStore cfs, Map options) { super(cfs, options); - int configuredMaxSSTableSize = 160; + int configuredMaxSSTableSize = DEFAULT_MAX_SSTABLE_SIZE_MIB; int configuredLevelFanoutSize = DEFAULT_LEVEL_FANOUT_SIZE; boolean configuredSingleSSTableUplevel = false; SizeTieredCompactionStrategyOptions localOptions = new SizeTieredCompactionStrategyOptions(options); diff --git a/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java b/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java index 04d3185bfc..c5f833f4ac 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java +++ b/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java @@ -146,12 +146,15 @@ public class MemtableIndexWriter implements PerColumnIndexWriter { final Iterator> iterator = rowMapping.merge(memtable); - try (MemtableTermsIterator terms = new MemtableTermsIterator(memtable.getMinTerm(), memtable.getMaxTerm(), iterator)) + long cellCount = 0; + if (iterator.hasNext()) { - long cellCount = flush(terms); - - completeIndexFlush(cellCount, start, stopwatch); + try (MemtableTermsIterator terms = new MemtableTermsIterator(memtable.getMinTerm(), memtable.getMaxTerm(), iterator)) + { + cellCount = flush(terms); + } } + completeIndexFlush(cellCount, start, stopwatch); } } catch (Throwable t) @@ -217,8 +220,8 @@ public class MemtableIndexWriter implements PerColumnIndexWriter private void completeIndexFlush(long cellCount, long startTime, Stopwatch stopwatch) throws IOException { - // create a completion marker indicating that the index is complete and not-empty - ColumnCompletionMarkerUtil.create(indexDescriptor, indexIdentifier, false); + // create a completion marker indicating that the index is complete + ColumnCompletionMarkerUtil.create(indexDescriptor, indexIdentifier, cellCount == 0); indexMetrics.memtableIndexFlushCount.inc(); diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java index 694729f8bd..bdff5b8900 100644 --- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java +++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java @@ -49,6 +49,9 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Strings; import com.google.common.collect.ImmutableList; import com.google.common.util.concurrent.Futures; +import org.apache.cassandra.tcm.compatibility.TokenRingUtils; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Shared; import org.assertj.core.api.Assertions; import org.junit.Assert; import org.slf4j.Logger; @@ -1646,4 +1649,41 @@ public class ClusterUtils }); }); } + + @Shared + public static class Range implements Serializable + { + public final String left, right; + + public Range(String left, String right) + { + this.left = left; + this.right = right; + } + + public Range(long left, long right) + { + this(Long.toString(left), Long.toString(right)); + } + + public long left() + { + return Long.parseLong(left); + } + + public long right() + { + return Long.parseLong(right); + } + } + + public static List getPrimaryRanges(IInvokableInstance instance, String keyspace) + { + return instance.callOnInstance(() -> { + var ranges = TokenRingUtils.getPrimaryRangesForEndpoint(keyspace, FBUtilities.getBroadcastAddressAndPort()); + return ranges.stream() + .flatMap(r -> r.unwrap().stream().map(r2 -> new Range(r2.left.toString(), r2.right.toString()))) + .collect(Collectors.toList()); + }); + } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/PartialWritesWithRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/PartialWritesWithRepairTest.java new file mode 100644 index 0000000000..703137c5c1 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/PartialWritesWithRepairTest.java @@ -0,0 +1,96 @@ +/* + * 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.test.sai; + +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.shared.ClusterUtils; +import org.apache.cassandra.distributed.shared.ClusterUtils.Range; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.junit.Test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +public class PartialWritesWithRepairTest extends TestBaseImpl +{ + @Test + public void test() throws IOException + { + try (Cluster cluster = Cluster.build(2) + .withConfig(c -> c.with(Feature.values())) + .start()) + { + init(cluster); + cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk vector, ck int, s1 int static, v1 int, v2 int, PRIMARY KEY (pk, ck))")); + cluster.schemaChange(withKeyspace("CREATE INDEX ON %s.tbl(s1) USING 'sai'")); + cluster.schemaChange(withKeyspace("CREATE INDEX ON %s.tbl(v1) USING 'sai'")); + cluster.schemaChange(withKeyspace("CREATE INDEX ON %s.tbl(v2) USING 'sai'")); + IInvokableInstance node1 = cluster.get(1); + IInvokableInstance node2 = cluster.get(2); + // see org.apache.cassandra.service.StorageService.repair + List partialRanges = ClusterUtils.getPrimaryRanges(node1, KEYSPACE); + var completeRanges = completeRanges(partialRanges); + + // write to each column for the complete set + // avoid writing to one of the columns for the partial set + for (var range : completeRanges) + { + ByteBuffer pk = key(range); + node2.executeInternal(withKeyspace("INSERT INTO %s.tbl(pk, ck, s1, v1, v2) VALUES (?, ?, ?, ?, ?)"), pk, 0, 0, 0, 0); + node2.executeInternal(withKeyspace("INSERT INTO %s.tbl(pk, ck, s1, v1, v2) VALUES (?, ?, ?, ?, ?)"), pk, 1, 0, 1, 1); + } + for (var range : partialRanges) + { + ByteBuffer pk = key(range); + node2.executeInternal(withKeyspace("INSERT INTO %s.tbl(pk, ck, v1) VALUES (?, ?, ?)"), pk, 0, 0); + node2.executeInternal(withKeyspace("INSERT INTO %s.tbl(pk, ck, v1) VALUES (?, ?, ?)"), pk, 1, 1); + } + + node1.nodetoolResult("repair", KEYSPACE, "-pr").asserts().success(); + } + } + + private static ByteBuffer key(Range range) + { + return Murmur3Partitioner.LongToken.keyForToken(range.right()); + } + + private static List completeRanges(List ranges) + { + ranges.sort(Comparator.comparingLong(Range::left)); + List list = new ArrayList<>(); + Range previous = ranges.get(0); + if (previous.left() != Long.MIN_VALUE) + list.add(new Range(Long.MIN_VALUE, ranges.get(0).left())); + for (int i = 1; i < ranges.size(); i++) + { + Range next = ranges.get(i); + if (!previous.right.equals(next.left)) + list.add(new Range(previous.right, next.left)); + previous = next; + } + return list; + } +} diff --git a/test/distributed/org/apache/cassandra/fuzz/sai/AccordFullMultiNodeSAITest.java b/test/distributed/org/apache/cassandra/fuzz/sai/AccordFullMultiNodeSAITest.java index 23b52833af..291ef7dcaa 100644 --- a/test/distributed/org/apache/cassandra/fuzz/sai/AccordFullMultiNodeSAITest.java +++ b/test/distributed/org/apache/cassandra/fuzz/sai/AccordFullMultiNodeSAITest.java @@ -25,7 +25,7 @@ import org.apache.cassandra.harry.gen.Generator; import org.apache.cassandra.harry.gen.SchemaGenerators; import org.apache.cassandra.service.consensus.TransactionalMode; -@Ignore("CASSANDRA-20567: Repair is failing due to missing SAI index files when using zero copy streaming") +@Ignore("It was believed that these tests were failing due to CASSANDRA-20567, but in fixing that issue it was found that the tests are still failing! Harry is detecting an incorrect response...") public class AccordFullMultiNodeSAITest extends MultiNodeSAITestBase { public AccordFullMultiNodeSAITest() diff --git a/test/distributed/org/apache/cassandra/fuzz/sai/AccordInteropMultiNodeSAITest.java b/test/distributed/org/apache/cassandra/fuzz/sai/AccordInteropMultiNodeSAITest.java index ba937c0f91..2d665eb47d 100644 --- a/test/distributed/org/apache/cassandra/fuzz/sai/AccordInteropMultiNodeSAITest.java +++ b/test/distributed/org/apache/cassandra/fuzz/sai/AccordInteropMultiNodeSAITest.java @@ -25,7 +25,7 @@ import org.apache.cassandra.harry.gen.Generator; import org.apache.cassandra.harry.gen.SchemaGenerators; import org.apache.cassandra.service.consensus.TransactionalMode; -@Ignore("CASSANDRA-20567: Repair is failing due to missing SAI index files when using zero copy streaming") +@Ignore("It was believed that these tests were failing due to CASSANDRA-20567, but in fixing that issue it was found that the tests are still failing! Harry is detecting an incorrect response...") public class AccordInteropMultiNodeSAITest extends MultiNodeSAITestBase { public AccordInteropMultiNodeSAITest() diff --git a/test/unit/org/apache/cassandra/utils/CassandraGenerators.java b/test/unit/org/apache/cassandra/utils/CassandraGenerators.java index 33ebdb6c3a..08b06c0730 100644 --- a/test/unit/org/apache/cassandra/utils/CassandraGenerators.java +++ b/test/unit/org/apache/cassandra/utils/CassandraGenerators.java @@ -50,6 +50,8 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Sets; + +import org.apache.cassandra.db.compaction.LeveledManifest; import org.apache.cassandra.schema.*; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; import org.apache.cassandra.tcm.extensions.ExtensionKey; @@ -563,11 +565,39 @@ public final class CassandraGenerators Map options = new HashMap<>(); if (nextBoolean(rnd)) options.putAll(sizeTieredOptions.generate(rnd)); + int maxSSTableSizeInMB = LeveledCompactionStrategy.DEFAULT_MAX_SSTABLE_SIZE_MIB; if (nextBoolean(rnd)) + { // size in mb - options.put(LeveledCompactionStrategy.SSTABLE_SIZE_OPTION, SourceDSL.integers().between(1, 2_000).generate(rnd).toString()); + maxSSTableSizeInMB = SourceDSL.integers().between(1, 2_000).generate(rnd); + options.put(LeveledCompactionStrategy.SSTABLE_SIZE_OPTION, Integer.toString(maxSSTableSizeInMB)); + } if (nextBoolean(rnd)) - options.put(LeveledCompactionStrategy.LEVEL_FANOUT_SIZE_OPTION, SourceDSL.integers().between(1, 100).generate(rnd).toString()); + { + // there is a relationship between sstable size and fanout, so respect it + // see CASSANDRA-20570: Leveled Compaction doesn't validate maxBytesForLevel when the table is altered/created + long maxSSTableSizeInBytes = maxSSTableSizeInMB * 1024L * 1024L; + Gen gen = SourceDSL.integers().between(1, 100); + Integer value = gen.generate(rnd); + while (true) + { + try + { + // see org.apache.cassandra.db.compaction.LeveledGenerations.MAX_LEVEL_COUNT for why 8 is hard coded here + LeveledManifest.maxBytesForLevel(8, value, maxSSTableSizeInBytes); + break; // value is good, keep it + } + catch (RuntimeException e) + { + // this value is too large... lets shrink it + if (value.intValue() == 1) + throw new AssertionError("There is no possible fanout size that works with maxSSTableSizeInMB=" + maxSSTableSizeInMB); + gen = SourceDSL.integers().between(1, value - 1); + value = gen.generate(rnd); + } + } + options.put(LeveledCompactionStrategy.LEVEL_FANOUT_SIZE_OPTION, value.toString()); + } if (nextBoolean(rnd)) options.put(LeveledCompactionStrategy.SINGLE_SSTABLE_UPLEVEL_OPTION, nextBoolean(rnd).toString()); return options;