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

Merge branch 'cassandra-5.0' into trunk

This commit is contained in:
David Capwell 2025-04-23 17:58:50 -07:00
parent 7160c49ed7 07831c9cc7
commit df24dacc0c
8 changed files with 182 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

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

3
src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java
View File

@ -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<String, String> 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);

15
src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java
View File

@ -146,12 +146,15 @@ public class MemtableIndexWriter implements PerColumnIndexWriter
{
final Iterator<Pair<ByteComparable, LongArrayList>> 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();

40
test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
View File

@ -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<Range> 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());
});
}
}

96
test/distributed/org/apache/cassandra/distributed/test/sai/PartialWritesWithRepairTest.java Normal file
View File

@ -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<bigint, 2>, 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<Range> 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<Range> completeRanges(List<Range> ranges)
{
ranges.sort(Comparator.comparingLong(Range::left));
List<Range> 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;
}
}

2
test/distributed/org/apache/cassandra/fuzz/sai/AccordFullMultiNodeSAITest.java
View File

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

2
test/distributed/org/apache/cassandra/fuzz/sai/AccordInteropMultiNodeSAITest.java
View File

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

34
test/unit/org/apache/cassandra/utils/CassandraGenerators.java
View File

@ -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<String, String> 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<Integer> 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;