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

Add LIMIT and PER PARTITION LIMIT to the AST SingleNodeTableWalkTest 

patch by David Capwell; reviewed by Abe Ratnofsky for CASSANDRA-20399
This commit is contained in:
David Capwell 2025-03-17 14:07:00 -07:00
parent 19007f374c
commit 779bae02c8
10 changed files with 615 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
View File

@ -72,10 +72,10 @@ public abstract class AbstractCompactionStrategy
protected static final boolean DEFAULT_UNCHECKED_TOMBSTONE_COMPACTION_OPTION = false;
protected static final boolean DEFAULT_LOG_ALL_OPTION = false;
protected static final String TOMBSTONE_THRESHOLD_OPTION = "tombstone_threshold";
public static final String TOMBSTONE_THRESHOLD_OPTION = "tombstone_threshold";
protected static final String TOMBSTONE_COMPACTION_INTERVAL_OPTION = "tombstone_compaction_interval";
// disable range overlap check when deciding if an SSTable is candidate for tombstone compaction (CASSANDRA-6563)
protected static final String UNCHECKED_TOMBSTONE_COMPACTION_OPTION = "unchecked_tombstone_compaction";
public static final String UNCHECKED_TOMBSTONE_COMPACTION_OPTION = "unchecked_tombstone_compaction";
protected static final String LOG_ALL_OPTION = "log_all";
protected static final String COMPACTION_ENABLED = "enabled";
public static final String ONLY_PURGE_REPAIRED_TOMBSTONES = "only_purge_repaired_tombstones";

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

@ -47,10 +47,10 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.TOLERATE_S
public class LeveledCompactionStrategy extends AbstractCompactionStrategy
{
private static final Logger logger = LoggerFactory.getLogger(LeveledCompactionStrategy.class);
private static final String SSTABLE_SIZE_OPTION = "sstable_size_in_mb";
public static final String SSTABLE_SIZE_OPTION = "sstable_size_in_mb";
private static final boolean tolerateSstableSize = TOLERATE_SSTABLE_SIZE.getBoolean();
private static final String LEVEL_FANOUT_SIZE_OPTION = "fanout_size";
private static final String SINGLE_SSTABLE_UPLEVEL_OPTION = "single_sstable_uplevel";
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;
@VisibleForTesting

6
src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategyOptions.java
View File

@ -26,9 +26,9 @@ public final class SizeTieredCompactionStrategyOptions
protected static final long DEFAULT_MIN_SSTABLE_SIZE = 50L * 1024L * 1024L;
protected static final double DEFAULT_BUCKET_LOW = 0.5;
protected static final double DEFAULT_BUCKET_HIGH = 1.5;
protected static final String MIN_SSTABLE_SIZE_KEY = "min_sstable_size";
protected static final String BUCKET_LOW_KEY = "bucket_low";
protected static final String BUCKET_HIGH_KEY = "bucket_high";
public static final String MIN_SSTABLE_SIZE_KEY = "min_sstable_size";
public static final String BUCKET_LOW_KEY = "bucket_low";
public static final String BUCKET_HIGH_KEY = "bucket_high";
protected long minSSTableSize;
protected double bucketLow;

18
src/java/org/apache/cassandra/db/compaction/unified/Controller.java
View File

@ -50,7 +50,7 @@ public class Controller
* The scaling parameters W, one per bucket index and separated by a comma.
* Higher indexes will use the value of the last index with a W specified.
*/
final static String SCALING_PARAMETERS_OPTION = "scaling_parameters";
public final static String SCALING_PARAMETERS_OPTION = "scaling_parameters";
private final static String DEFAULT_SCALING_PARAMETERS =
CassandraRelevantProperties.UCS_SCALING_PARAMETER.getString();
@ -58,7 +58,7 @@ public class Controller
* The minimum sstable size. Sharded writers split sstables over shard only if they are at least as large as the
* minimum size.
*/
static final String MIN_SSTABLE_SIZE_OPTION = "min_sstable_size";
public static final String MIN_SSTABLE_SIZE_OPTION = "min_sstable_size";
private static final String DEFAULT_MIN_SSTABLE_SIZE = CassandraRelevantProperties.UCS_MIN_SSTABLE_SIZE.getString();
@ -66,9 +66,9 @@ public class Controller
* Override for the flush size in MB. The database should be able to calculate this from executing flushes, this
* should only be necessary in rare cases.
*/
static final String FLUSH_SIZE_OVERRIDE_OPTION = "flush_size_override";
public static final String FLUSH_SIZE_OVERRIDE_OPTION = "flush_size_override";
static final String BASE_SHARD_COUNT_OPTION = "base_shard_count";
public static final String BASE_SHARD_COUNT_OPTION = "base_shard_count";
/**
* Default base shard count, used when a base count is not explicitly supplied. This value applies as long as the
* table is not a system one, and directories are not defined.
@ -79,10 +79,10 @@ public class Controller
public static final int DEFAULT_BASE_SHARD_COUNT =
CassandraRelevantProperties.UCS_BASE_SHARD_COUNT.getInt();
static final String TARGET_SSTABLE_SIZE_OPTION = "target_sstable_size";
public static final String TARGET_SSTABLE_SIZE_OPTION = "target_sstable_size";
public static final long DEFAULT_TARGET_SSTABLE_SIZE =
CassandraRelevantProperties.UCS_TARGET_SSTABLE_SIZE.getSizeInBytes();
static final long MIN_TARGET_SSTABLE_SIZE = 1L << 20;
public static final long MIN_TARGET_SSTABLE_SIZE = 1L << 20;
/**
* Provision for growth of the constructed SSTables as the size of the data grows. By default, the target SSTable
@ -109,7 +109,7 @@ public class Controller
* base count of 4, the number of SSTables will be 4 (~256GiB each) for a growth value of 1, 128 (~8GiB each) for
* a growth value of 0.333, and 64 (~16GiB each) for a growth value of 0.5.
*/
static final String SSTABLE_GROWTH_OPTION = "sstable_growth";
public static final String SSTABLE_GROWTH_OPTION = "sstable_growth";
private static final double DEFAULT_SSTABLE_GROWTH = CassandraRelevantProperties.UCS_SSTABLE_GROWTH.getDouble();
/**
@ -127,7 +127,7 @@ public class Controller
*
* If the fanout factor is larger than the maximum number of sstables, the strategy will ignore the latter.
*/
static final String MAX_SSTABLES_TO_COMPACT_OPTION = "max_sstables_to_compact";
public static final String MAX_SSTABLES_TO_COMPACT_OPTION = "max_sstables_to_compact";
static final String ALLOW_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_OPTION = "unsafe_aggressive_sstable_expiration";
static final boolean ALLOW_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION =
@ -147,7 +147,7 @@ public class Controller
* that overlap with participating (LCS-like, higher concurrency during upgrades but some double compaction),
* TRANSITIVE to include overlaps of overlaps (likely to trigger whole level compactions, safest).
*/
static final String OVERLAP_INCLUSION_METHOD_OPTION = "overlap_inclusion_method";
public static final String OVERLAP_INCLUSION_METHOD_OPTION = "overlap_inclusion_method";
static final Overlaps.InclusionMethod DEFAULT_OVERLAP_INCLUSION_METHOD =
CassandraRelevantProperties.UCS_OVERLAP_INCLUSION_METHOD.getEnum(Overlaps.InclusionMethod.TRANSITIVE);

7
test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java
View File

@ -130,7 +130,7 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
}
}
Select select = builder.build();
return state.command(rs, select, (wholePartition ? "Whole Partition" : "Single Row"));
return state.command(rs, select, (wholePartition ? "By Partition Key" : "By Primary Key"));
}
public Property.Command<State, Void, ?> selectToken(RandomSource rs, State state)
@ -357,7 +357,10 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
//TODO (coverage): partition is defined at the cluster level, so have to hard code in this model as the table is changed rather than cluster being recreated... this limits coverage
return toGen(new TableMetadataBuilder()
.withTableKinds(TableMetadata.Kind.REGULAR)
.withKnownMemtables()
.withParams(b -> b.withKnownMemtables()
.withCaching()
.withCompaction()
.withCompression())
.withKeyspaceName(ks).withTableName("tbl")
.withSimpleColumnNames()
.withDefaultTypeGen(supportedTypes())

53
test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java
View File

@ -99,8 +99,7 @@ public class StatefulASTBase extends TestBaseImpl
*/
protected static boolean CQL_DEBUG_APPLY_OPERATOR = false;
protected static final Gen<Gen<Boolean>> BIND_OR_LITERAL_DISTRO = Gens.bools().mixedDistribution();
protected static final Gen<Gen<Boolean>> BETWEEN_EQ_DISTRO = Gens.bools().mixedDistribution();
protected static final Gen<Gen<Boolean>> BOOL_DISTRIBUTION = Gens.bools().mixedDistribution();
protected static final Gen<Gen<Conditional.Where.Inequality>> LESS_THAN_DISTRO = Gens.mixedDistribution(Stream.of(Conditional.Where.Inequality.values())
.filter(i -> i == Conditional.Where.Inequality.LESS_THAN || i == Conditional.Where.Inequality.LESS_THAN_EQ)
.collect(Collectors.toList()));
@ -111,6 +110,7 @@ public class StatefulASTBase extends TestBaseImpl
.filter(i -> i != Conditional.Where.Inequality.EQUAL && i != Conditional.Where.Inequality.NOT_EQUAL)
.collect(Collectors.toList()));
protected static final Gen<Gen.IntGen> FETCH_SIZE_DISTRO = Gens.mixedDistribution(new int[] {1, 10, 100, 1000, 5000});
protected static final Gen<Gen.IntGen> LIMIT_DISTRO = Gens.mixedDistribution(1, 1001);
static
{
@ -196,6 +196,8 @@ public class StatefulASTBase extends TestBaseImpl
protected final Session session;
protected final Gen<Boolean> bindOrLiteralGen;
protected final Gen<Boolean> betweenEqGen;
protected final Gen<Boolean> useFetchSizeGen, usePerPartitionLimitGen, useLimitGen;
protected final Gen.IntGen perPartitionLimitGen, limitGen;
protected final Gen<Conditional.Where.Inequality> lessThanGen;
protected final Gen<Conditional.Where.Inequality> greaterThanGen;
protected final Gen<Conditional.Where.Inequality> rangeInequalityGen;
@ -221,12 +223,17 @@ public class StatefulASTBase extends TestBaseImpl
this.debug = CQL_DEBUG_APPLY_OPERATOR ? CompositeVisitor.of(StandardVisitors.APPLY_OPERATOR, StandardVisitors.DEBUG)
: StandardVisitors.DEBUG;
this.bindOrLiteralGen = BIND_OR_LITERAL_DISTRO.next(rs);
this.betweenEqGen = BETWEEN_EQ_DISTRO.next(rs);
this.bindOrLiteralGen = BOOL_DISTRIBUTION.next(rs);
this.betweenEqGen = BOOL_DISTRIBUTION.next(rs);
this.lessThanGen = LESS_THAN_DISTRO.next(rs);
this.greaterThanGen = GREATER_THAN_DISTRO.next(rs);
this.rangeInequalityGen = RANGE_INEQUALITY_DISTRO.next(rs);
this.fetchSizeGen = FETCH_SIZE_DISTRO.next(rs);
this.useFetchSizeGen = BOOL_DISTRIBUTION.next(rs);
this.usePerPartitionLimitGen = BOOL_DISTRIBUTION.next(rs);
this.useLimitGen = BOOL_DISTRIBUTION.next(rs);
this.perPartitionLimitGen = LIMIT_DISTRO.next(rs);
this.limitGen = LIMIT_DISTRO.next(rs);
this.enoughMemtables = rs.pickInt(3, 10, 50);
this.enoughSSTables = rs.pickInt(3, 10, 50);
@ -260,18 +267,50 @@ public class StatefulASTBase extends TestBaseImpl
return command(rs, select, null);
}
protected boolean allowLimit(Select select)
{
//TODO (coverage): allow this in the model!
// LIMIT with IN clause on partition columns is non-deterministic which is not currently supported by the model
if (select.where.isEmpty()) return true;
return !select.where.get()
.streamRecursive(true)
.filter(e -> e instanceof Conditional.In)
.anyMatch(e -> {
var in = (Conditional.In) e;
// when expression is size 1, then this is deterministic
if (in.expressions.size() == 1) return false;
return model.factory.partitionColumns.contains(in.ref);
});
}
protected boolean allowPerPartitionLimit(Select select)
{
return true;
}
protected boolean allowPaging(Select select)
{
return true;
}
protected <S extends BaseState> Property.Command<S, Void, ?> command(RandomSource rs, Select select, @Nullable String annotate)
{
var inst = selectInstance(rs);
//TODO (coverage): don't limit this to all selects, only those doing range queries!
int fetchSize = fetchSizeGen.nextInt(rs);
if (allowPerPartitionLimit(select) && usePerPartitionLimitGen.next(rs))
select = select.withPerPartitionLimit(perPartitionLimitGen.nextInt(rs));
if (allowLimit(select) && useLimitGen.next(rs))
select = select.withLimit(limitGen.nextInt(rs));
int fetchSize = allowPaging(select) && useFetchSizeGen.next(rs)
? fetchSizeGen.nextInt(rs)
: Integer.MAX_VALUE;
String postfix = "on " + inst;
if (fetchSize != Integer.MAX_VALUE)
postfix += ", fetch size " + fetchSize;
if (annotate == null) annotate = postfix;
else annotate += ", " + postfix;
Select finalSelect = select;
return new Property.SimpleCommand<>(humanReadable(select, annotate), s -> {
s.model.validate(s.executeQuery(inst, fetchSize, s.selectCl(), select), select);
s.model.validate(s.executeQuery(inst, fetchSize, s.selectCl(), finalSelect), finalSelect);
});
}

39
test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModel.java
View File

@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.function.IntFunction;
@ -52,9 +53,11 @@ import org.apache.cassandra.cql3.ast.Mutation;
import org.apache.cassandra.cql3.ast.Select;
import org.apache.cassandra.cql3.ast.StandardVisitors;
import org.apache.cassandra.cql3.ast.Symbol;
import org.apache.cassandra.cql3.ast.Value;
import org.apache.cassandra.db.BufferClustering;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.harry.model.BytesPartitionState.PrimaryKey;
import org.apache.cassandra.harry.util.StringUtils;
@ -572,7 +575,7 @@ public class ASTSingleTableModel
private SelectResult getRowsAsByteBuffer(Select select)
{
if (select.where.isEmpty())
return all();
return new SelectResult(getRowsAsByteBuffer(applyLimits(all(), select.perPartitionLimit, select.limit)), false);
LookupContext ctx = context(select);
List<PrimaryKey> primaryKeys;
if (ctx.unmatchable)
@ -598,11 +601,41 @@ public class ASTSingleTableModel
// partial tested (handles many columns, tests are single column)
primaryKeys = search(ctx);
}
primaryKeys = applyLimits(primaryKeys, select.perPartitionLimit, select.limit);
//TODO (correctness): now that we have the rows we need to handle the selections/aggregation/limit/group-by/etc.
return new SelectResult(getRowsAsByteBuffer(primaryKeys), ctx.unordered);
}
private SelectResult all()
private List<PrimaryKey> applyLimits(List<PrimaryKey> primaryKeys, Optional<Value> perPartitionLimitOpt, Optional<Value> limitOpt)
{
if (perPartitionLimitOpt.isPresent())
{
int limit = Int32Type.instance.compose(eval(perPartitionLimitOpt.get()));
var it = primaryKeys.iterator();
BytesPartitionState.Ref current = null;
int count = 0;
while (it.hasNext())
{
PrimaryKey next = it.next();
if (current == null || !current.equals(next.partition))
{
current = next.partition;
count = 0;
}
if (++count > limit)
it.remove();
}
}
if (limitOpt.isPresent())
{
int limit = Int32Type.instance.compose(eval(limitOpt.get()));
if (primaryKeys.size() > limit)
primaryKeys = primaryKeys.subList(0, limit);
}
return primaryKeys;
}
private List<PrimaryKey> all()
{
List<PrimaryKey> primaryKeys = new ArrayList<>();
for (var partition : partitions.values())
@ -610,7 +643,7 @@ public class ASTSingleTableModel
if (partition.staticOnly()) primaryKeys.add(partition.partitionRowRef());
else partition.rows().stream().map(BytesPartitionState.Row::ref).forEach(primaryKeys::add);
}
return new SelectResult(getRowsAsByteBuffer(primaryKeys), false);
return primaryKeys;
}
public ByteBuffer[][] getRowsAsByteBuffer(List<PrimaryKey> primaryKeys)

102
test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModelTest.java
View File

@ -26,6 +26,7 @@ import java.util.List;
import java.util.TreeMap;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@ -83,8 +84,36 @@ public class ASTSingleTableModelTest
Select.Builder builder = Select.builder().table(metadata);
for (var pk : metadata.partitionKeyColumns())
builder.value(new Symbol(pk), ZERO);
Select select = builder.build();
model.validate(expected, select);
model.validate(expected, builder.build());
}
}
@Test
public void singlePartitionLimit()
{
for (TableMetadata metadata : defaultTables())
{
if (metadata.clusteringColumns().isEmpty()) continue;
ASTSingleTableModel model = new ASTSingleTableModel(metadata);
Function<ByteBuffer, ColumnValue> update = bb -> partitionLevelUpdate(ZERO, bb);
ByteBuffer[][] expected = new ByteBuffer[][]{
insert(model, ZERO),
insert(model, update.apply(ONE)),
insert(model, update.apply(TWO)),
insert(model, update.apply(THREE)),
};
Select.Builder builder = Select.builder().table(metadata);
for (var pk : metadata.partitionKeyColumns())
builder.value(new Symbol(pk), ZERO);
// without limit
model.validate(expected, builder.build());
for (int limit = 1; limit <= expected.length; limit++)
{
builder.limit(limit);
model.validate(Arrays.copyOf(expected, limit), builder.build());
}
}
}
@ -328,7 +357,38 @@ public class ASTSingleTableModelTest
modelModel.add(insert(model, value));
}
model.validate(modelModel.all(), Select.builder(metadata).build());
var builder = Select.builder(metadata);
ByteBuffer[][] all = modelModel.all();
model.validate(all, builder.build());
for (int i = 1; i < all.length; i++)
model.validate(Arrays.copyOf(all, i), builder.limit(i).build());
model.validate(all, builder.limit(all.length).build());
model.validate(all, builder.limit(all.length + 1).build());
}
}
@Test
public void selectStarPerPartitionLimit()
{
List<ByteBuffer> values = Arrays.asList(ZERO, ONE, TWO, THREE);
for (TableMetadata metadata : defaultTables())
{
if (metadata.clusteringColumns().isEmpty()) continue;
ASTSingleTableModel model = new ASTSingleTableModel(metadata);
ModelModel modelModel = new ModelModel(model);
for (ByteBuffer pk : values)
{
for (ByteBuffer row : values)
modelModel.add(insert(model, partitionLevelUpdate(pk, row)));
}
var builder = Select.builder(metadata);
model.validate(modelModel.all(), builder.build());
for (int i = 1; i < values.size(); i++)
model.validate(modelModel.allPerPartitionLimit(i), builder.perPartitionLimit(i).build());
model.validate(modelModel.all(), builder.perPartitionLimit(values.size()).build());
}
}
@ -522,6 +582,20 @@ public class ASTSingleTableModelTest
return insert(model, (i1, i2) -> value);
}
private static ColumnValue partitionLevelUpdate(ByteBuffer partitionLevel, ByteBuffer rowLevel)
{
return (kind, offset) -> {
switch (kind)
{
case PARTITION_KEY:
case STATIC:
return partitionLevel;
default:
return rowLevel;
}
};
}
private static ByteBuffer[] insert(ASTSingleTableModel model, ColumnValue fn)
{
TableMetadata metadata = model.factory.metadata;
@ -617,6 +691,28 @@ public class ASTSingleTableModelTest
return allWhere(i -> true);
}
public ByteBuffer[][] allPerPartitionLimit(int limit)
{
class State
{
BytesPartitionState.Ref current = null;
int count = 0;
boolean nextPartition(BytesPartitionState.Ref ref)
{
current = ref;
count = 0;
return true;
}
boolean nextRow(ByteBuffer[] row)
{
return ++count <= limit;
}
}
State state = new State();
return allWhere(state::nextPartition, state::nextRow);
}
public ByteBuffer[][] allEq(Symbol column, ByteBuffer value)
{
return allWhere(column, Inequality.EQUAL, value);

58
test/unit/org/apache/cassandra/cql3/ast/Select.java
View File

@ -39,6 +39,7 @@ FROM [keyspace_name.] table_name
[AND clustering_filters
[AND static_filters]]]
[ORDER BY PK_column_name ASC|DESC]
[PER PARTITION LIMIT N]
[LIMIT N]
[ALLOW FILTERING]
*/
@ -49,7 +50,7 @@ FROM [keyspace_name.] table_name
// where
public final Optional<Conditional> where;
public final Optional<OrderBy> orderBy;
public final Optional<Value> limit;
public final Optional<Value> perPartitionLimit, limit;
public final boolean allowFiltering;
public Select(List<Expression> selections)
@ -59,15 +60,16 @@ FROM [keyspace_name.] table_name
public Select(List<Expression> selections, Optional<TableReference> source, Optional<Conditional> where, Optional<OrderBy> orderBy, Optional<Value> limit)
{
this(selections, source, where, orderBy, limit, false);
this(selections, source, where, orderBy, Optional.empty(), limit, false);
}
public Select(List<Expression> selections, Optional<TableReference> source, Optional<Conditional> where, Optional<OrderBy> orderBy, Optional<Value> limit, boolean allowFiltering)
public Select(List<Expression> selections, Optional<TableReference> source, Optional<Conditional> where, Optional<OrderBy> orderBy, Optional<Value> perPartitionLimit, Optional<Value> limit, boolean allowFiltering)
{
this.selections = selections;
this.source = source;
this.where = where;
this.orderBy = orderBy;
this.perPartitionLimit = perPartitionLimit;
this.limit = limit;
this.allowFiltering = allowFiltering;
@ -96,7 +98,17 @@ FROM [keyspace_name.] table_name
public Select withAllowFiltering()
{
return new Select(selections, source, where, orderBy, limit, true);
return new Select(selections, source, where, orderBy, perPartitionLimit, limit, true);
}
public Select withLimit(int limit)
{
return new Select(selections, source, where, orderBy, perPartitionLimit, Optional.of(Literal.of(limit)), allowFiltering);
}
public Select withPerPartitionLimit(int perPartitionLimit)
{
return new Select(selections, source, where, orderBy, Optional.of(Literal.of(perPartitionLimit)), limit, allowFiltering);
}
@Override
@ -132,6 +144,12 @@ FROM [keyspace_name.] table_name
sb.append("ORDER BY ");
orderBy.get().toCQL(sb, formatter);
}
if (perPartitionLimit.isPresent())
{
formatter.section(sb);
sb.append("PER PARTITION LIMIT ");
perPartitionLimit.get().toCQL(sb, formatter);
}
if (limit.isPresent())
{
formatter.section(sb);
@ -154,6 +172,7 @@ FROM [keyspace_name.] table_name
+ (source.isPresent() ? 1 : 0)
+ (where.isPresent() ? 1 : 0)
+ (orderBy.isPresent() ? 1 : 0)
+ (perPartitionLimit.isPresent() ? 1 : 0)
+ (limit.isPresent() ? 1 : 0));
es.addAll(selections);
if (source.isPresent())
@ -162,6 +181,8 @@ FROM [keyspace_name.] table_name
es.add(where.get());
if (orderBy.isPresent())
es.add(orderBy.get());
if (perPartitionLimit.isPresent())
es.add(perPartitionLimit.get());
if (limit.isPresent())
es.add(limit.get());
return es.stream();
@ -204,6 +225,18 @@ FROM [keyspace_name.] table_name
{
where = this.where;
}
Optional<Value> perPartitionLimit;
if (this.perPartitionLimit.isPresent())
{
var l = this.perPartitionLimit.get();
var update = l.visit(v);
updated |= l != update;
perPartitionLimit = Optional.ofNullable(update);
}
else
{
perPartitionLimit = this.perPartitionLimit;
}
Optional<Value> limit;
if (this.limit.isPresent())
{
@ -217,9 +250,10 @@ FROM [keyspace_name.] table_name
limit = this.limit;
}
if (!updated) return this;
return new Select(selections, source, where, orderBy, limit, allowFiltering);
return new Select(selections, source, where, orderBy, perPartitionLimit, limit, allowFiltering);
}
public static class OrderBy implements Element
{
public enum Ordering
@ -313,6 +347,7 @@ FROM [keyspace_name.] table_name
protected Optional<TableReference> source = Optional.empty();
private Conditional.Builder where = new Conditional.Builder();
private OrderBy.Builder orderBy = new OrderBy.Builder();
private Optional<Value> perPartitionLimit = Optional.empty();
private Optional<Value> limit = Optional.empty();
private boolean allowFiltering = false;
@ -377,6 +412,17 @@ FROM [keyspace_name.] table_name
return (T) this;
}
public T perPartitionLimit(Value limit)
{
this.perPartitionLimit = Optional.of(limit);
return (T) this;
}
public T perPartitionLimit(int limit)
{
return perPartitionLimit(Bind.of(limit));
}
public T limit(Value limit)
{
this.limit = Optional.of(limit);
@ -394,7 +440,7 @@ FROM [keyspace_name.] table_name
source,
where.isEmpty() ? Optional.empty() : Optional.of(where.build()),
orderBy.isEmpty() ? Optional.empty() : Optional.of(orderBy.build()),
limit,
perPartitionLimit, limit,
allowFiltering);
}
}

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

@ -28,6 +28,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@ -51,6 +52,7 @@ import com.google.common.collect.Sets;
import org.apache.commons.lang3.builder.MultilineRecursiveToStringStyle;
import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
import org.apache.cassandra.config.DataStorageSpec;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.cql3.Duration;
@ -60,6 +62,12 @@ import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.SchemaCQLHelper;
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.Slices;
import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
import org.apache.cassandra.db.compaction.LeveledCompactionStrategy;
import org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy;
import org.apache.cassandra.db.compaction.SizeTieredCompactionStrategyOptions;
import org.apache.cassandra.db.compaction.UnifiedCompactionStrategy;
import org.apache.cassandra.db.compaction.unified.Controller;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.CompositeType;
@ -82,6 +90,8 @@ import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.HeartBeatState;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.io.compress.LZ4Compressor;
import org.apache.cassandra.io.compress.ZstdCompressor;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.InetAddressAndPort;
@ -94,7 +104,10 @@ import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.NoPayload;
import org.apache.cassandra.net.PingRequest;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.CachingParams;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.CompactionParams;
import org.apache.cassandra.schema.CompressionParams;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.MemtableParams;
@ -196,6 +209,71 @@ public final class CassandraGenerators
}
private static String humanReadableSignPrefix(RandomnessSource rnd)
{
switch (SourceDSL.integers().between(0, 2).generate(rnd))
{
case 0: return "";
case 1: return "-";
case 2: return "+";
default:
throw new AssertionError();
}
}
public static Gen<String> humanReadableStorageValue()
{
Gen<Long> valueGen = SourceDSL.longs().between(0, 1000);
return rnd -> {
// [+-]?\d+(\.\d+)?([eE]([+-]?)\d+)?
StringBuilder sb = new StringBuilder();
sb.append(humanReadableSignPrefix(rnd));
sb.append(valueGen.generate(rnd));
if (nextBoolean(rnd))
{
sb.append('.');
sb.append(valueGen.generate(rnd));
}
if (nextBoolean(rnd))
{
sb.append('E');
sb.append(humanReadableSignPrefix(rnd));
sb.append(valueGen.generate(rnd));
}
return sb.toString();
};
}
public static Gen<String> humanReadableStorage()
{
Gen<DataStorageSpec.DataStorageUnit> unitGen = SourceDSL.arbitrary().enumValues(DataStorageSpec.DataStorageUnit.class);
return rnd -> {
DataStorageSpec.DataStorageUnit unit = unitGen.generate(rnd);
String value;
switch (SourceDSL.integers().between(0, 2).generate(rnd))
{
case 0:
value = "NaN";
break;
case 1:
value = humanReadableSignPrefix(rnd) + "Infinity";
break;
case 2:
value = humanReadableStorageValue().generate(rnd);
break;
default:
throw new AssertionError();
}
return value + ' ' + unit.getSymbol();
};
}
public static Gen<String> humanReadableStorageSimple()
{
Gen<DataStorageSpec.DataStorageUnit> unitGen = SourceDSL.arbitrary().enumValues(DataStorageSpec.DataStorageUnit.class);
return rnd -> humanReadableStorageValue().generate(rnd) + ' ' + unitGen.generate(rnd).getSymbol();
}
public static Set<UserType> extractUDTs(TableMetadata metadata)
{
Set<UserType> matches = new HashSet<>();
@ -413,10 +491,261 @@ public final class CassandraGenerators
}
}
public static Gen<CachingParams> cachingParamsGen()
{
return rnd -> {
boolean cacheKeys = nextBoolean(rnd);
int rowsPerPartitionToCache;
switch (SourceDSL.integers().between(1, 3).generate(rnd))
{
case 1: // ALL
rowsPerPartitionToCache = Integer.MAX_VALUE;
break;
case 2: // NONE
rowsPerPartitionToCache = 0;
break;
case 3: // num values
rowsPerPartitionToCache = Math.toIntExact(rnd.next(Constraint.between(1, Integer.MAX_VALUE - 1)));
break;
default:
throw new AssertionError();
}
return new CachingParams(cacheKeys, rowsPerPartitionToCache);
};
}
public enum KnownCompactionAlgo
{
SizeTiered(SizeTieredCompactionStrategy.class),
Leveled(LeveledCompactionStrategy.class),
Unified(UnifiedCompactionStrategy.class);
private final Class<? extends AbstractCompactionStrategy> klass;
KnownCompactionAlgo(Class<? extends AbstractCompactionStrategy> klass)
{
this.klass = klass;
}
}
public static class CompactionParamsBuilder
{
private Gen<KnownCompactionAlgo> algoGen = SourceDSL.arbitrary().enumValues(KnownCompactionAlgo.class);
private Gen<CompactionParams.TombstoneOption> tombstoneOptionGen = SourceDSL.arbitrary().enumValues(CompactionParams.TombstoneOption.class);
private Gen<Map<String, String>> sizeTieredOptions = rnd -> {
if (nextBoolean(rnd)) return Map.of();
Map<String, String> options = new HashMap<>();
if (nextBoolean(rnd))
// computes mb then converts to bytes
options.put(SizeTieredCompactionStrategyOptions.MIN_SSTABLE_SIZE_KEY, Long.toString(SourceDSL.longs().between(1, 100).generate(rnd) * 1024L * 1024L));
if (nextBoolean(rnd))
options.put(SizeTieredCompactionStrategyOptions.BUCKET_LOW_KEY, Double.toString(SourceDSL.doubles().between(0.1, 0.9).generate(rnd)));
if (nextBoolean(rnd))
options.put(SizeTieredCompactionStrategyOptions.BUCKET_HIGH_KEY, Double.toString(SourceDSL.doubles().between(1.1, 1.9).generate(rnd)));
return options;
};
private Gen<Map<String, String>> leveledOptions = rnd -> {
if (nextBoolean(rnd)) return Map.of();
Map<String, String> options = new HashMap<>();
if (nextBoolean(rnd))
options.putAll(sizeTieredOptions.generate(rnd));
if (nextBoolean(rnd))
// size in mb
options.put(LeveledCompactionStrategy.SSTABLE_SIZE_OPTION, SourceDSL.integers().between(1, 2_000).generate(rnd).toString());
if (nextBoolean(rnd))
options.put(LeveledCompactionStrategy.LEVEL_FANOUT_SIZE_OPTION, SourceDSL.integers().between(1, 100).generate(rnd).toString());
if (nextBoolean(rnd))
options.put(LeveledCompactionStrategy.SINGLE_SSTABLE_UPLEVEL_OPTION, nextBoolean(rnd).toString());
return options;
};
private Gen<Map<String, String>> unifiedOptions = rnd -> {
if (nextBoolean(rnd)) return Map.of();
Gen<String> storageSizeGen = Generators.filter(humanReadableStorageSimple(), s -> Controller.MIN_TARGET_SSTABLE_SIZE <= FBUtilities.parseHumanReadableBytes(s));
Map<String, String> options = new HashMap<>();
if (nextBoolean(rnd))
options.put(Controller.BASE_SHARD_COUNT_OPTION, SourceDSL.integers().between(1, 10).generate(rnd).toString());
if (nextBoolean(rnd))
options.put(Controller.FLUSH_SIZE_OVERRIDE_OPTION, storageSizeGen.generate(rnd));
if (nextBoolean(rnd))
options.put(Controller.MAX_SSTABLES_TO_COMPACT_OPTION, SourceDSL.integers().between(0, 32).generate(rnd).toString());
if (nextBoolean(rnd))
options.put(Controller.SSTABLE_GROWTH_OPTION, SourceDSL.integers().between(0, 100).generate(rnd) + "%");
if (nextBoolean(rnd))
options.put(Controller.OVERLAP_INCLUSION_METHOD_OPTION, SourceDSL.arbitrary().enumValues(Overlaps.InclusionMethod.class).generate(rnd).name());
if (nextBoolean(rnd))
{
int numLevels = SourceDSL.integers().between(1, 10).generate(rnd);
String[] scalingParams = new String[numLevels];
Gen<Integer> levelSize = SourceDSL.integers().between(2, 10);
for (int i = 0; i < numLevels; i++)
{
String value;
switch (SourceDSL.integers().between(0, 3).generate(rnd))
{
case 0:
value = "N";
break;
case 1:
value = "L" + levelSize.generate(rnd);
break;
case 2:
value = "T" + levelSize.generate(rnd);
break;
case 3:
value = SourceDSL.integers().all().generate(rnd).toString();
break;
default:
throw new AssertionError();
}
scalingParams[i] = value;
}
options.put(Controller.SCALING_PARAMETERS_OPTION, String.join(",", scalingParams));
}
if (nextBoolean(rnd))
{
// Calculate TARGET then compute the MIN from that. The issue is that there is a hidden relationship
// between these 2 fields more complex than simple comparability, MIN must be < 70% * TARGET!
// See CASSANDRA-20398
// 1MiB to 128MiB target
long targetBytes = SourceDSL.longs().between(1L << 20, 1L << 27).generate(rnd);
long limit = (long) Math.ceil(targetBytes * Math.sqrt(0.5));
long minBytes = SourceDSL.longs().between(1, limit - 1).generate(rnd);
options.put(Controller.MIN_SSTABLE_SIZE_OPTION, minBytes + "B");
options.put(Controller.TARGET_SSTABLE_SIZE_OPTION, targetBytes + "B");
}
return options;
};
//TODO (coverage): doesn't look to validate > 1, what does that even mean?
private Gen<Float> tombstoneThreshold = SourceDSL.floats().between(0, 1);
private Gen<Boolean> uncheckedTombstoneCompaction = SourceDSL.booleans().all();
private Gen<Boolean> onlyPurgeRepairedTombstones = SourceDSL.booleans().all();
public Gen<CompactionParams> build()
{
return rnd -> {
KnownCompactionAlgo algo = algoGen.generate(rnd);
Map<String, String> options = new HashMap<>();
if (nextBoolean(rnd))
options.put(CompactionParams.Option.PROVIDE_OVERLAPPING_TOMBSTONES.toString(), tombstoneOptionGen.generate(rnd).name());
if (CompactionParams.supportsThresholdParams(algo.klass) && nextBoolean(rnd))
{
options.put(CompactionParams.Option.MIN_THRESHOLD.toString(), Long.toString(rnd.next(Constraint.between(2, 4))));
options.put(CompactionParams.Option.MAX_THRESHOLD.toString(), Long.toString(rnd.next(Constraint.between(5, 32))));
}
if (nextBoolean(rnd))
options.put(AbstractCompactionStrategy.TOMBSTONE_THRESHOLD_OPTION, tombstoneThreshold.generate(rnd).toString());
if (nextBoolean(rnd))
options.put(AbstractCompactionStrategy.UNCHECKED_TOMBSTONE_COMPACTION_OPTION, uncheckedTombstoneCompaction.generate(rnd).toString());
if (nextBoolean(rnd))
options.put(AbstractCompactionStrategy.ONLY_PURGE_REPAIRED_TOMBSTONES, onlyPurgeRepairedTombstones.generate(rnd).toString());
switch (algo)
{
case SizeTiered:
options.putAll(sizeTieredOptions.generate(rnd));
break;
case Leveled:
options.putAll(leveledOptions.generate(rnd));
break;
case Unified:
options.putAll(unifiedOptions.generate(rnd));
break;
default:
throw new UnsupportedOperationException(algo.name());
}
return CompactionParams.create(algo.klass, options);
};
}
}
private static Boolean nextBoolean(RandomnessSource rnd)
{
return SourceDSL.booleans().all().generate(rnd);
}
public static Gen<CompactionParams> compactionParamsGen()
{
return new CompactionParamsBuilder().build();
}
public enum KnownCompressionAlgo
{
snappy("SnappyCompressor"),
deflate("DeflateCompressor"),
lz4("LZ4Compressor"),
zstd("ZstdCompressor"),
noop("NoopCompressor");
private final String compressor;
KnownCompressionAlgo(String compressor)
{
this.compressor = compressor;
}
}
public static class CompressionParamsBuilder
{
private Gen<Boolean> enabledGen = SourceDSL.booleans().all();
private Gen<KnownCompressionAlgo> algoGen = SourceDSL.arbitrary().enumValues(KnownCompressionAlgo.class);
private Gen<Map<String, String>> lz4OptionsGen = rnd -> {
if (nextBoolean(rnd))
return Map.of();
Map<String, String> options = new HashMap<>();
if (nextBoolean(rnd))
options.put(LZ4Compressor.LZ4_COMPRESSOR_TYPE, nextBoolean(rnd) ? LZ4Compressor.LZ4_FAST_COMPRESSOR : LZ4Compressor.LZ4_HIGH_COMPRESSOR);
if (nextBoolean(rnd))
options.put(LZ4Compressor.LZ4_HIGH_COMPRESSION_LEVEL, Integer.toString(Math.toIntExact(rnd.next(Constraint.between(1, 17)))));
return options;
};
private Gen<Map<String, String>> zstdOptionsGen = rnd -> {
if (nextBoolean(rnd))
return Map.of();
int level = Math.toIntExact(rnd.next(Constraint.between(ZstdCompressor.FAST_COMPRESSION_LEVEL, ZstdCompressor.BEST_COMPRESSION_LEVEL)));
return Map.of(ZstdCompressor.COMPRESSION_LEVEL_OPTION_NAME, Integer.toString(level));
};
public Gen<CompressionParams> build()
{
return rnd -> {
if (!enabledGen.generate(rnd))
return CompressionParams.noCompression();
KnownCompressionAlgo algo = algoGen.generate(rnd);
if (algo == KnownCompressionAlgo.noop)
return CompressionParams.noop();
// when null disabled
int chunkLength = CompressionParams.DEFAULT_CHUNK_LENGTH;
double minCompressRatio = CompressionParams.DEFAULT_MIN_COMPRESS_RATIO;
Map<String, String> options;
switch (algo)
{
case lz4:
options = lz4OptionsGen.generate(rnd);
break;
case zstd:
options = zstdOptionsGen.generate(rnd);
break;
default:
options = Map.of();
}
return new CompressionParams(algo.compressor, options, chunkLength, minCompressRatio);
};
}
}
public static Gen<CompressionParams> compressionParamsGen()
{
return new CompressionParamsBuilder().build();
}
public static class TableParamsBuilder
{
@Nullable
private Gen<String> memtableKeyGen = null;
@Nullable
private Gen<CachingParams> cachingParamsGen = null;
@Nullable
private Gen<CompactionParams> compactionParamsGen = null;
@Nullable
private Gen<CompressionParams> compressionParamsGen = null;
public TableParamsBuilder withKnownMemtables()
{
@ -427,12 +756,36 @@ public final class CassandraGenerators
return this;
}
public TableParamsBuilder withCaching()
{
cachingParamsGen = cachingParamsGen();
return this;
}
public TableParamsBuilder withCompaction()
{
compactionParamsGen = compactionParamsGen();
return this;
}
public TableParamsBuilder withCompression()
{
compressionParamsGen = compressionParamsGen();
return this;
}
public Gen<TableParams> build()
{
return rnd -> {
TableParams.Builder params = TableParams.builder();
if (memtableKeyGen != null)
params.memtable(MemtableParams.get(memtableKeyGen.generate(rnd)));
if (cachingParamsGen != null)
params.caching(cachingParamsGen.generate(rnd));
if (compactionParamsGen != null)
params.compaction(compactionParamsGen.generate(rnd));
if (compressionParamsGen != null)
params.compression(compressionParamsGen.generate(rnd));
return params.build();
};
}
@ -515,6 +868,12 @@ public final class CassandraGenerators
return this;
}
public TableMetadataBuilder withParams(Consumer<TableParamsBuilder> fn)
{
fn.accept(paramsBuilder);
return this;
}
public TableMetadataBuilder withKeyspaceName(Gen<String> ksNameGen)
{
this.ksNameGen = ksNameGen;
@ -1358,7 +1717,7 @@ public final class CassandraGenerators
public static Gen<Epoch> epochs()
{
return rnd -> {
if (SourceDSL.booleans().all().generate(rnd))
if (nextBoolean(rnd))
{
switch (SourceDSL.arbitrary().enumValues(EpochConstants.class).generate(rnd))
{