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

Add system_views.partition_key_statistics for querying SSTable metadata 

patch by Chris Lohfink; reviewed by David Capwell, Stefan Miklosovic for CASSANDRA-20161
This commit is contained in:
Chris Lohfink 2024-12-21 10:07:27 -06:00 committed by Stefan Miklosovic
parent 90f51b9c0b
commit 69dc5d05ef
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
20 changed files with 856 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1
* Add system_views.partition_key_statistics for querying SSTable metadata (CASSANDRA-20161)
* CEP-42 - Add Constraints Framework (CASSANDRA-19947)
* Add table metric PurgeableTombstoneScannedHistogram and a tracing event for scanned purgeable tombstones (CASSANDRA-20132)
* Make sure we can parse the expanded CQL before writing it to the log or sending it to replicas (CASSANDRA-20218)

47
doc/modules/cassandra/pages/managing/operating/virtualtables.adoc
View File

@ -516,6 +516,53 @@ SELECT total - progress AS remaining
FROM system_views.sstable_tasks;
----
=== Virtual table for primary id's
Since https://issues.apache.org/jira/browse/CASSANDRA-20161[CASSANDRA-20161], there is
a virtual table `system_views.partition_key_statistics` to allow users to query partition keys and related metadata for a specific table within a keyspace. This feature provides insights into SSTable-level details, such as token values, size estimates, and SSTable counts, without requiring expensive disk-based operations.
[source,console]
----
cassandra@cqlsh> select * from ks.tbl;
id | cl1 | i
----+-----+-----
1 | 202 | 999
1 | 200 | 999
1 | 101 | 600
2 | 101 | 600
----
[source,console]
----
cassandra@cqlsh> use system_views;
cassandra@cqlsh> select * from partition_key_statistics where keyspace_name = 'ks' and table_name = 'tbl' and key = '1';
@ Row 1
---------------+----------------------
keyspace_name | ks
table_name | tbl
token_value | -4069959284402364209
key | 1
size_estimate | 83
sstables | 3
cassandra@cqlsh> select * from partition_key_statistics where keyspace_name = 'ks' and table_name = 'tbl' and key = '2';
@ Row 1
---------------+----------------------
keyspace_name | ks
table_name | tbl
token_value | -3248873570005575792
key | 2
size_estimate | 25
sstables | 1
----
The value in `sstables` column means how many SSTables a particular key is located in.
If there is a composite partition key, you can separate the values by a colon for `key`.
=== Other Virtual Tables
Some examples of using other virtual tables are as follows.

10
src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
View File

@ -345,7 +345,7 @@ public final class StatementRestrictions
}
else
{
if (!allowFiltering && requiresAllowFilteringIfNotSpecified())
if (!allowFiltering && requiresAllowFilteringIfNotSpecified(table))
throw invalidRequest(allowFilteringMessage(state));
}
@ -356,12 +356,12 @@ public final class StatementRestrictions
validateSecondaryIndexSelections();
}
public boolean requiresAllowFilteringIfNotSpecified()
public static boolean requiresAllowFilteringIfNotSpecified(TableMetadata metadata)
{
if (!table.isVirtual())
if (!metadata.isVirtual())
return true;
VirtualTable tableNullable = VirtualKeyspaceRegistry.instance.getTableNullable(table.id);
VirtualTable tableNullable = VirtualKeyspaceRegistry.instance.getTableNullable(metadata.id);
assert tableNullable != null;
return !tableNullable.allowFilteringImplicitly();
}
@ -568,7 +568,7 @@ public final class StatementRestrictions
// components must have a EQ. Only the last partition key component can be in IN relation.
if (partitionKeyRestrictions.needFiltering())
{
if (!allowFiltering && !forView && !hasQueriableIndex && requiresAllowFilteringIfNotSpecified())
if (!allowFiltering && !forView && !hasQueriableIndex && requiresAllowFilteringIfNotSpecified(table))
throw new InvalidRequestException(allowFilteringMessage(state));
isKeyRange = true;

5
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View File

@ -88,6 +88,7 @@ import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import static java.lang.String.format;
import static org.apache.cassandra.cql3.restrictions.StatementRestrictions.requiresAllowFilteringIfNotSpecified;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkNotNull;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkNull;
@ -1357,7 +1358,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
boundNames,
orderings,
selectsOnlyStaticColumns,
parameters.allowFiltering,
parameters.allowFiltering || !requiresAllowFilteringIfNotSpecified(metadata),
forView);
}
@ -1583,7 +1584,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
{
// We will potentially filter data if the row filter is not the identity and there isn't any index group
// supporting all the expressions in the filter.
if (restrictions.requiresAllowFilteringIfNotSpecified())
if (requiresAllowFilteringIfNotSpecified(table))
checkFalse(restrictions.needFiltering(table), StatementRestrictions.REQUIRES_ALLOW_FILTERING_MESSAGE);
}
}

2
src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
View File

@ -565,7 +565,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
{
VirtualTable view = VirtualKeyspaceRegistry.instance.getTableNullable(metadata().id);
UnfilteredPartitionIterator resultIterator = view.select(dataRange, columnFilter());
UnfilteredPartitionIterator resultIterator = view.select(dataRange, columnFilter(), rowFilter());
return limits().filter(rowFilter().filter(resultIterator, nowInSec()), nowInSec(), selectsFullPartition());
}

2
src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
View File

@ -1395,7 +1395,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
{
VirtualTable view = VirtualKeyspaceRegistry.instance.getTableNullable(metadata().id);
UnfilteredPartitionIterator resultIterator = view.select(partitionKey, clusteringIndexFilter, columnFilter());
UnfilteredPartitionIterator resultIterator = view.select(partitionKey, clusteringIndexFilter, columnFilter(), rowFilter());
return limits().filter(rowFilter().filter(resultIterator, nowInSec()), nowInSec(), selectsFullPartition());
}

5
src/java/org/apache/cassandra/db/virtual/AbstractVirtualTable.java
View File

@ -29,6 +29,7 @@ import org.apache.cassandra.db.EmptyIterators;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.partitions.AbstractUnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.SingletonUnfilteredPartitionIterator;
@ -75,7 +76,7 @@ public abstract class AbstractVirtualTable implements VirtualTable
}
@Override
public final UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter)
public final UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter)
{
Partition partition = data(partitionKey).getPartition(partitionKey);
@ -88,7 +89,7 @@ public abstract class AbstractVirtualTable implements VirtualTable
}
@Override
public final UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter)
public final UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter)
{
DataSet data = data();

6
src/java/org/apache/cassandra/db/virtual/CollectionVirtualTableAdapter.java
View File

@ -51,6 +51,7 @@ import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.EmptyIterators;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BooleanType;
import org.apache.cassandra.db.marshal.ByteType;
@ -307,7 +308,8 @@ public class CollectionVirtualTableAdapter<R> implements VirtualTable
@Override
public UnfilteredPartitionIterator select(DecoratedKey partitionKey,
ClusteringIndexFilter clusteringFilter,
ColumnFilter columnFilter)
ColumnFilter columnFilter,
RowFilter rowFilter)
{
if (!data.iterator().hasNext())
return EmptyIterators.unfilteredPartition(metadata);
@ -348,7 +350,7 @@ public class CollectionVirtualTableAdapter<R> implements VirtualTable
}
@Override
public UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter)
public UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter)
{
return createPartitionIterator(metadata, new AbstractIterator<>()
{

364
src/java/org/apache/cassandra/db/virtual/PartitionKeyStatsTable.java Normal file
View File

@ -0,0 +1,364 @@
/*
* 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.db.virtual;
import java.io.IOException;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.Slice;
import org.apache.cassandra.db.Slices;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.db.marshal.CounterColumnType;
import org.apache.cassandra.db.marshal.IntegerType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.SingletonUnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.AbstractUnfilteredRowIterator;
import org.apache.cassandra.db.rows.BTreeRow;
import org.apache.cassandra.db.rows.BufferCell;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.EncodingStats;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.Rows;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.io.sstable.KeyReader;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.serializers.MarshalException;
import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest;
/**
* A virtual table for querying partition keys of SSTables in a specific keyspace.
*
* <p>This table is implemented as a virtual table in Cassandra, meaning it does not
* store data persistently on disk but instead derives its data from live metadata.
*
* <p>The CQL equivalent of this virtual table is:
* <pre>
* CREATE TABLE system_views.partition_key_statistics (
* keyspace_name TEXT,
* table_name TEXT,
* token_value INT,
* key TEXT,
* size_estimate COUNTER,
* sstables COUNTER,
* PRIMARY KEY ((keyspace_name, table_name), token_value, key)
* );
* </pre>
*
* <p>Note:
* <ul>
* <li>The `size_estimate` and `sstables` columns represent aggregate information about SSTable sizes and counts.</li>
* <li>Range queries across multiple tables and updates are not supported as this is a read-only table.</li>
* </ul>
*/
public class PartitionKeyStatsTable implements VirtualTable
{
private static final Logger logger = LoggerFactory.getLogger(PartitionKeyStatsTable.class);
public static final String NAME = "partition_key_statistics";
private static final String TABLE_READ_ONLY_ERROR = "The specified table is read-only.";
private static final String UNSUPPORTED_RANGE_QUERY_ERROR = "Range queries are not supported. Please provide both a keyspace and a table name.";
private static final String REVERSED_QUERY_ERROR = "Reversed queries are not supported.";
private static final String KEYSPACE_NOT_EXIST_ERROR = "The keyspace '%s' does not exist.";
private static final String TABLE_NOT_EXIST_ERROR = "The table '%s' does not exist in the keyspace '%s'.";
private static final String KEY_ONLY_EQUALS_ERROR = "The 'key' column can only be used in an equality query for this virtual table.";
private static final String KEY_NOT_WITHIN_BOUNDS_ERROR = "The specified 'key' is not within the provided token value bounds.";
private static final String PARTITIONER_NOT_SUPPORTED = "Partitioner '%s' for table '%s' in keyspace '%s' is not supported.";
private static final String COLUMN_KEYSPACE_NAME = "keyspace_name";
private static final String COLUMN_TABLE_NAME = "table_name";
private static final String COLUMN_TOKEN_VALUE = "token_value";
private static final String COLUMN_KEY = "key";
private static final String COLUMN_SIZE_ESTIMATE = "size_estimate";
private static final String COLUMN_SSTABLES = "sstables";
private final TableMetadata metadata;
private final ColumnMetadata sizeEstimateColumn;
private final ColumnMetadata sstablesColumn;
@VisibleForTesting
final CopyOnWriteArrayList<Consumer<DecoratedKey>> readListener = new CopyOnWriteArrayList<>();
public PartitionKeyStatsTable(String keyspace)
{
this.metadata = TableMetadata.builder(keyspace, NAME)
.kind(TableMetadata.Kind.VIRTUAL)
.partitioner(new LocalPartitioner(CompositeType.getInstance(UTF8Type.instance, UTF8Type.instance)))
.addPartitionKeyColumn(COLUMN_KEYSPACE_NAME, UTF8Type.instance)
.addPartitionKeyColumn(COLUMN_TABLE_NAME, UTF8Type.instance)
.addClusteringColumn(COLUMN_TOKEN_VALUE, IntegerType.instance)
.addClusteringColumn(COLUMN_KEY, UTF8Type.instance)
.addRegularColumn(COLUMN_SIZE_ESTIMATE, CounterColumnType.instance)
.addRegularColumn(COLUMN_SSTABLES, CounterColumnType.instance)
.build();
sizeEstimateColumn = metadata.regularColumns().getSimple(0);
sstablesColumn = metadata.regularColumns().getSimple(1);
}
@Override
public UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter)
{
if (clusteringIndexFilter.isReversed())
throw new InvalidRequestException(REVERSED_QUERY_ERROR);
ByteBuffer[] key = ((CompositeType) this.metadata.partitionKeyType).split(partitionKey.getKey());
String keyspace = UTF8Type.instance.getString(key[0]);
String table = UTF8Type.instance.getString(key[1]);
KeyspaceMetadata ksm = Schema.instance.getKeyspaceMetadata(keyspace);
if (ksm == null)
throw invalidRequest(KEYSPACE_NOT_EXIST_ERROR, keyspace);
TableMetadata metadata = ksm.getTableOrViewNullable(table);
if (metadata == null)
throw invalidRequest(TABLE_NOT_EXIST_ERROR, table, keyspace);
if (!metadata.partitioner.supportsSplitting())
throw invalidRequest(PARTITIONER_NOT_SUPPORTED, metadata.partitioner.getClass().getName(), table, keyspace);
AbstractBounds<PartitionPosition> range = getBounds(metadata, clusteringIndexFilter, rowFilter);
return new SingletonUnfilteredPartitionIterator(select(partitionKey, metadata, clusteringIndexFilter, range));
}
private List<SSTableReader> getSStables(TableMetadata metadata, AbstractBounds<PartitionPosition> range)
{
return Lists.newArrayList(ColumnFamilyStore.getIfExists(metadata).getTracker().getView().liveSSTablesInBounds(range.left, range.right));
}
private UnfilteredRowIterator select(DecoratedKey partitionKey, TableMetadata metadata, ClusteringIndexFilter clusteringIndexFilter, AbstractBounds<PartitionPosition> range)
{
List<SSTableReader> sstables = getSStables(metadata, range);
if (sstables.isEmpty())
return UnfilteredRowIterators.noRowsIterator(metadata, partitionKey, Rows.EMPTY_STATIC_ROW, DeletionTime.LIVE, false);
List<UnfilteredRowIterator> sstableIterators = Lists.newArrayList();
for (SSTableReader sstable : sstables)
sstableIterators.add(getSStableRowIterator(metadata, partitionKey, sstable, clusteringIndexFilter, range));
return UnfilteredRowIterators.merge(sstableIterators);
}
private UnfilteredRowIterator getSStableRowIterator(TableMetadata target, DecoratedKey partitionKey, SSTableReader sstable, ClusteringIndexFilter filter, AbstractBounds<PartitionPosition> range)
{
final KeyReader reader;
try
{
// ignore warning on try-with-resources, the reader will be closed on endOfData or close
reader = sstable.keyReader(range.left);
}
catch (IOException e)
{
logger.error("Error generating keyReader for SSTable: {}", sstable, e);
throw new RuntimeException(e);
}
return new AbstractUnfilteredRowIterator(metadata, partitionKey, DeletionTime.LIVE,
metadata.regularAndStaticColumns(), Rows.EMPTY_STATIC_ROW,
false, EncodingStats.NO_STATS)
{
public Unfiltered endOfData()
{
reader.close();
return super.endOfData();
}
public void close()
{
reader.close();
}
private Row buildRow(Clustering<?> clustering, long size)
{
Row.Builder row = BTreeRow.sortedBuilder();
row.newRow(clustering);
row.addCell(cell(sizeEstimateColumn, CounterContext.instance().createUpdate(size)));
row.addCell(cell(sstablesColumn, CounterContext.instance().createUpdate(1)));
return row.build();
}
@Override
protected Unfiltered computeNext()
{
while (!reader.isExhausted())
{
DecoratedKey key = target.partitioner.decorateKey(reader.key());
for (Consumer<DecoratedKey> listener : readListener)
listener.accept(key);
// Store the reader's current data position to calculate size later
long lastPosition = reader.dataPosition();
try
{
// Advance the reader to the next key for the next iteration. Also by moving to next key
// we move the dataPosition to the start of the next key for calculating size
reader.advance();
}
catch (IOException e)
{
logger.error("Error advancing reader for SSTable: {}", sstable, e);
return endOfData();
}
// Calculate the size of the current key. If EOF use the length of the file
long current = reader.dataPosition() == -1 ? sstable.uncompressedLength() : reader.dataPosition();
long size = current - lastPosition;
String keyString = target.partitionKeyType.getString(key.getKey());
// Check if the current key is outside the queried range; if so, stop
if (range.right.compareTo(key) < 0)
return endOfData();
// Convert the token to a string and create a clustering object
String tokenString = key.getToken().toString();
Clustering<?> clustering = Clustering.make(
IntegerType.instance.decompose(new BigInteger(tokenString)),
UTF8Type.instance.decompose(keyString)
);
// Check if the current clustering matches the filter; if so, return the row
if (filter.selects(clustering))
return buildRow(clustering, size);
}
return endOfData();
}
};
}
/**
* This converts the clustering token/key into the partition level token/key for the target table. Also provides an
* optimization from RowFilter when a `key` is specified with or without the clustering `token` being set.
*/
private AbstractBounds<PartitionPosition> getBounds(TableMetadata target, ClusteringIndexFilter clusteringIndexFilter, RowFilter rowFilter)
{
Slices s = clusteringIndexFilter.getSlices(target);
Token startToken = target.partitioner.getMinimumToken();
Token endToken = target.partitioner.getMaximumToken();
BigInteger startTokenValue = new BigInteger(endToken.getTokenValue().toString(), 10);
BigInteger endTokenValue = new BigInteger(startToken.getTokenValue().toString(), 10);
// find min/max token values from the clustering key
for (int i = 0; i < s.size(); i++)
{
Slice slice = s.get(i);
if (!slice.start().isEmpty())
{
startTokenValue = startTokenValue.min(IntegerType.instance.compose(slice.start().bufferAt(0)));
startToken = target.partitioner.getTokenFactory().fromString(startTokenValue.toString());
}
if (!slice.end().isEmpty())
{
endTokenValue = endTokenValue.max(IntegerType.instance.compose(slice.end().bufferAt(0)));
endToken = target.partitioner.getTokenFactory().fromString(endTokenValue.toString());
}
}
// override min/max of token if the `key` is specified
for (RowFilter.Expression expression : rowFilter.getExpressions())
{
if (expression.column().name.toString().equals(COLUMN_KEY))
{
if (expression.operator() != Operator.EQ)
throw new InvalidRequestException(KEY_ONLY_EQUALS_ERROR);
String keyString = UTF8Type.instance.compose(expression.getIndexValue());
ByteBuffer keyAsBB;
try
{
keyAsBB = target.partitionKeyType.fromString(keyString);
}
catch (MarshalException ex)
{
throw new InvalidRequestException(ex.getMessage());
}
DecoratedKey decoratedKey = target.partitioner.decorateKey(keyAsBB);
if (!DataRange.forKeyRange(new Range<>(startToken.minKeyBound(), endToken.maxKeyBound())).contains(decoratedKey.getToken().minKeyBound()))
throw new InvalidRequestException(KEY_NOT_WITHIN_BOUNDS_ERROR);
return Bounds.bounds(decoratedKey, true, decoratedKey, true);
}
}
return Bounds.bounds(startToken.minKeyBound(), true, endToken.maxKeyBound(), true);
}
private static Cell<?> cell(ColumnMetadata column, ByteBuffer value)
{
return BufferCell.live(column, 1L, value);
}
@Override
public TableMetadata metadata()
{
return this.metadata;
}
@Override
public UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter)
{
throw new InvalidRequestException(UNSUPPORTED_RANGE_QUERY_ERROR);
}
@Override
public void truncate()
{
throw new InvalidRequestException(TABLE_READ_ONLY_ERROR);
}
@Override
public void apply(PartitionUpdate update)
{
throw new InvalidRequestException(TABLE_READ_ONLY_ERROR);
}
}

1
src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java
View File

@ -56,6 +56,7 @@ public final class SystemViewsKeyspace extends VirtualKeyspace
.add(new RolesCacheKeysTable(VIRTUAL_VIEWS))
.add(new CQLMetricsTable(VIRTUAL_VIEWS))
.add(new BatchMetricsTable(VIRTUAL_VIEWS))
.add(new PartitionKeyStatsTable(VIRTUAL_VIEWS))
.add(new StreamingVirtualTable(VIRTUAL_VIEWS))
.add(new GossipInfoTable(VIRTUAL_VIEWS))
.add(new QueriesTable(VIRTUAL_VIEWS))

7
src/java/org/apache/cassandra/db/virtual/VirtualTable.java
View File

@ -21,6 +21,7 @@ import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.schema.TableMetadata;
@ -59,18 +60,20 @@ public interface VirtualTable
* @param partitionKey the partition key
* @param clusteringIndexFilter the clustering columns to selected
* @param columnFilter the selected columns
* @param rowFilter filter on which rows a given query should include or exclude
* @return the rows corresponding to the requested data.
*/
UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter);
UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter);
/**
* Selects the rows from a range of partitions.
*
* @param dataRange the range of data to retrieve
* @param columnFilter the selected columns
* @param rowFilter filter on which rows a given query should include or exclude
* @return the rows corresponding to the requested data.
*/
UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter);
UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter);
/**
* Truncates data from the underlying source, if supported.

9
src/java/org/apache/cassandra/dht/IPartitioner.java
View File

@ -74,6 +74,15 @@ public interface IPartitioner
throw new UnsupportedOperationException("If you are using a splitting partitioner, getMaximumToken has to be implemented");
}
/**
*
* @return true if supports splitting as per {@link IPartitioner#split(Token, Token, double)}, false otherwise. Defaults to false.
*/
default boolean supportsSplitting()
{
return false;
}
/**
* @return a Token that can be used to route a given key
* (This is NOT a method to create a Token from its string representation;

5
src/java/org/apache/cassandra/dht/Murmur3Partitioner.java
View File

@ -138,6 +138,11 @@ public class Murmur3Partitioner implements IPartitioner
return new LongToken(newToken);
}
public boolean supportsSplitting()
{
return true;
}
public LongToken getMinimumToken()
{
return MINIMUM;

5
src/java/org/apache/cassandra/dht/RandomPartitioner.java
View File

@ -135,6 +135,11 @@ public class RandomPartitioner implements IPartitioner
return new BigIntegerToken(newToken);
}
public boolean supportsSplitting()
{
return true;
}
public BigIntegerToken getMinimumToken()
{
return MINIMUM;

8
src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
View File

@ -903,6 +903,14 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource,
*/
public abstract KeyReader keyReader() throws IOException;
/**
* Returns a {@link KeyReader} over all keys in the sstable after a given key.
* @param key
* @return
* @throws IOException
*/
public abstract KeyReader keyReader(PartitionPosition key) throws IOException;
/**
* Returns a {@link KeyIterator} over all keys in the sstable.
*/

7
src/java/org/apache/cassandra/io/sstable/format/big/BigTableKeyReader.java
View File

@ -53,7 +53,12 @@ public class BigTableKeyReader implements KeyReader
public static BigTableKeyReader create(RandomAccessReader indexFileReader, IndexSerializer serializer) throws IOException
{
BigTableKeyReader iterator = new BigTableKeyReader(null, indexFileReader, serializer);
return create(null, indexFileReader, serializer);
}
public static BigTableKeyReader create(FileHandle indexFile, RandomAccessReader indexFileReader, IndexSerializer serializer) throws IOException
{
BigTableKeyReader iterator = new BigTableKeyReader(indexFile, indexFileReader, serializer);
try
{
iterator.advance();

22
src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java
View File

@ -155,6 +155,28 @@ public class BigTableReader extends SSTableReaderWithFilter implements IndexSumm
return BigTableKeyReader.create(ifile, rowIndexEntrySerializer);
}
@Override
public KeyReader keyReader(PartitionPosition key) throws IOException
{
FileHandle iFile = ifile.sharedCopy();
RandomAccessReader reader = iFile.createReader();
reader.seek(getIndexScanPosition(key));
KeyReader keys = BigTableKeyReader.create(iFile, reader, rowIndexEntrySerializer);
boolean hasMoreKeys = true;
while (hasMoreKeys)
{
ByteBuffer indexKey = keys.key();
DecoratedKey indexDecoratedKey = decorateKey(indexKey);
if (indexDecoratedKey.compareTo(key) >= 0)
break;
// Advance the iterator and check if more keys are available
hasMoreKeys = keys.advance();
}
return keys;
}
/**
* Finds and returns the first key beyond a given token in this SSTable or null if no such key exists.
*/

10
src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReader.java
View File

@ -45,6 +45,7 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.IVerifier;
import org.apache.cassandra.io.sstable.KeyReader;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.SSTableReadsListener;
import org.apache.cassandra.io.sstable.SSTableReadsListener.SelectionReason;
@ -124,6 +125,15 @@ public class BtiTableReader extends SSTableReaderWithFilter
return partitionIndex == null ? 0 : partitionIndex.size();
}
@Override
public KeyReader keyReader(PartitionPosition key) throws IOException
{
return PartitionIterator.create(partitionIndex, metadata().partitioner, rowIndexFile, dfile,
key, -1,
metadata().partitioner.getMaximumToken().maxKeyBound(), 0,
descriptor.version);
}
@Override
protected TrieIndexEntry getRowIndexEntry(PartitionPosition key,
Operator operator,

5
test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java
View File

@ -243,6 +243,11 @@ public abstract class ForwardingSSTableReader extends SSTableReader
return delegate.keyReader();
}
public KeyReader keyReader(PartitionPosition key) throws IOException
{
return delegate.keyReader(key);
}
@Override
public KeyIterator keyIterator() throws IOException
{

351
test/unit/org/apache/cassandra/db/virtual/PartitionKeyStatsTableTest.java Normal file
View File

@ -0,0 +1,351 @@
/*
* 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.db.virtual;
import java.math.BigInteger;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.collect.ImmutableList;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.exceptions.InvalidQueryException;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.sstable.format.bti.BtiFormat;
import org.bouncycastle.util.encoders.Hex;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@RunWith(Parameterized.class)
public class PartitionKeyStatsTableTest extends CQLTester
{
private static final String KS_NAME = "vts";
private String table;
private AtomicInteger scanned;
private final boolean useBtiFormat;
@Parameters(name = "Use BtiFormat = {0}")
public static Collection<Object[]> parameters()
{
return Arrays.asList(new Object[][]{ { false }, { true } });
}
public PartitionKeyStatsTableTest(boolean useBtiFormat)
{
this.useBtiFormat = useBtiFormat;
}
@Before
public void before()
{
if (useBtiFormat)
DatabaseDescriptor.setSelectedSSTableFormat(new BtiFormat.BtiFormatFactory().getInstance(Collections.emptyMap()));
PartitionKeyStatsTable primaryIdTable = new PartitionKeyStatsTable(KS_NAME);
scanned = new AtomicInteger();
VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(KS_NAME, ImmutableList.of(primaryIdTable)));
table = createTable("CREATE TABLE %s (key blob PRIMARY KEY, value blob)");
ByteBuffer value = ByteBuffer.wrap(new byte[1]);
for (int i = -10; i < 1000; i++)
{
ByteBuffer key = Murmur3Partitioner.LongToken.keyForToken(i);
execute("INSERT INTO %s (key, value) VALUES (?, ?)", key, value);
}
Util.flushTable(KEYSPACE, table);
primaryIdTable.readListener.add(unused -> scanned.incrementAndGet());
}
@Test
public void testPrimaryIdTable()
{
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ?",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(1010, all.size());
assertResults(all, -10, 1000);
// 1010 + 100 for the 1 per 10 page, +1 for the last
assertEquals(1111, scanned.get());
}
@Test
public void testTokenValueGreaterThanZero()
{
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value > 0",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(999, all.size());
assertResults(all, 1, 1000);
assertEquals(1099, scanned.get());
}
@Test
public void testTokenValueGreaterThanNegativeFive()
{
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value > -5",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(1004, all.size());
assertResults(all, -4, 1000);
// 1004 + 100 for the 1 per 10 page, +1 for the last
assertEquals(1105, scanned.get());
}
@Test
public void testTokenValueLessThanOrEqualToFive()
{
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value <= 5",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(16, all.size());
assertResults(all, -10, 5);
assertEquals(18, scanned.get());
}
@Test
public void testTokenValueEqualToZero()
{
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value = 0",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(1, all.size());
Row row = all.get(0);
assertEquals(BigInteger.valueOf(0), row.get("token_value", BigInteger.class));
assertEquals(2, scanned.get());
}
@Test
public void testTokenValueBounds()
{
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value > 0 AND token_value < 15",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(14, all.size());
assertResults(all, 1, 14);
// 0->10 = 11, 10->16 = 7
assertEquals(18, scanned.get());
}
@Test
public void testTokenValueBoundsWithBetween()
{
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value BETWEEN 0 AND 15",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(16, all.size());
assertResults(all, 0, 15);
assertEquals(18, scanned.get());
}
@Test
public void testTokenValueBoundsWithIn()
{
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value IN (1,3,6)",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(3, all.size());
assertEquals(BigInteger.valueOf(1), all.get(0).get("token_value", BigInteger.class));
assertEquals(BigInteger.valueOf(3), all.get(1).get("token_value", BigInteger.class));
assertEquals(BigInteger.valueOf(6), all.get(2).get("token_value", BigInteger.class));
assertEquals(7, scanned.get());
}
@Test
public void testTokenValueBoundsWithKey()
{
ByteBuffer ten = Murmur3Partitioner.LongToken.keyForToken(10);
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value > 0 AND token_value < 15 AND key = ?",
10, KEYSPACE, table, Hex.toHexString(ten.array()));
List<Row> all = rs.all();
assertEquals(1, all.size());
Row row = all.get(0);
assertEquals(BigInteger.valueOf(10), row.get("token_value", BigInteger.class));
assertEquals(2, scanned.get());
}
@Test
public void testByKey()
{
ByteBuffer ten = Murmur3Partitioner.LongToken.keyForToken(10);
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND key = ?",
10, KEYSPACE, table, Hex.toHexString(ten.array()));
List<Row> all = rs.all();
assertEquals(1, all.size());
Row row = all.get(0);
assertEquals(BigInteger.valueOf(10), row.get("token_value", BigInteger.class));
assertEquals(2, scanned.get());
}
@Test
public void testIgnoreSStableOutOfRange()
{
ByteBuffer twok = Murmur3Partitioner.LongToken.keyForToken(2000);
execute("INSERT INTO %s (key, value) VALUES (?, ?)", twok, ByteBuffer.wrap(new byte[1]));
Util.flushTable(KEYSPACE, table);
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value > 1500",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(1, all.size());
Row row = all.get(0);
assertEquals(BigInteger.valueOf(2000), row.get("token_value", BigInteger.class));
assertEquals(1L, row.get("sstables", Long.class).longValue());
assertEquals(1, scanned.get());
}
@Test
public void testNoResults()
{
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value < -1000",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(0, all.size());
assertEquals(0, scanned.get()); // sstables shouldn't even of been touched
}
@Test(expected = InvalidQueryException.class)
public void testNonExistantKeyspace()
{
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = 'non_existent' AND table_name = ?",
10, table);
List<Row> all = rs.all();
assertEquals(0, all.size());
assertEquals(0, scanned.get());
}
@Test
public void testNoResultsWithSSTables()
{
ByteBuffer o1 = Murmur3Partitioner.LongToken.keyForToken(10000);
ByteBuffer o2 = Murmur3Partitioner.LongToken.keyForToken(10002);
ByteBuffer value = ByteBuffer.wrap(new byte[10]);
execute("INSERT INTO %s (key, value) VALUES (?, ?)", o1, value);
execute("INSERT INTO %s (key, value) VALUES (?, ?)", o2, value);
Util.flushTable(KEYSPACE, table);
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value = 10001",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(0, all.size());
assertEquals(1, scanned.get());
}
@Test
public void testPrimaryIdTableDuplicates()
{
// 0xc25f118f072d6ba5cab7fb1468ace617 hashes to 1563004846366
ByteBuffer dup = Murmur3Partitioner.LongToken.keyForToken(1563004846366L);
// -19, 68, -61 (0xed44c3) hashes to 1563004846366
ByteBuffer dup2 = ByteBuffer.wrap(new byte[]{ -19, 68, -61 });
ByteBuffer value = ByteBuffer.wrap(new byte[10]);
execute("INSERT INTO %s (key, value) VALUES (?, ?)", dup, value);
execute("INSERT INTO %s (key, value) VALUES (?, ?)", dup2, value);
Util.flushTable(KEYSPACE, table);
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND token_value = 1563004846366",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(2, all.size());
assertEquals(BigInteger.valueOf(1563004846366L), all.get(0).get("token_value", BigInteger.class));
assertEquals(BigInteger.valueOf(1563004846366L), all.get(1).get("token_value", BigInteger.class));
assertEquals("c25f118f072d6ba5cab7fb1468ace617", all.get(0).getString("key"));
assertEquals("ed44c3", all.get(1).getString("key"));
assertEquals(2, scanned.get());
}
@Test
public void testCompositeType() throws UnknownHostException
{
String table = createTable("CREATE TABLE %s (key text, keytwo inet, value text, primary key ((key, keytwo)))");
execute("INSERT INTO %s (key, keytwo, value) VALUES (?, ?, ?)", "testkey", InetAddress.getByName("127.0.0.1"), "value");
Util.flushTable(KEYSPACE, table);
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND key = 'testkey:127.0.0.1'",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(1, all.size());
}
@Test
public void testTextType()
{
String table = createTable("CREATE TABLE %s (key text PRIMARY KEY, value text)");
execute("INSERT INTO %s (key, value) VALUES (?, ?)", "testkey", "value");
Util.flushTable(KEYSPACE, table);
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ? AND key = 'testkey'",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(1, all.size());
}
@Test
public void testSameKeyInMultipleSSTables()
{
String table = createTable("CREATE TABLE %s (key blob PRIMARY KEY, value blob)");
ByteBuffer key = Murmur3Partitioner.LongToken.keyForToken(1);
ByteBuffer value = ByteBuffer.wrap(new byte[10]);
execute("INSERT INTO %s (key, value) VALUES (?, ?)", key, value);
Util.flushTable(KEYSPACE, table);
value = ByteBuffer.wrap(new byte[100]);
execute("INSERT INTO %s (key, value) VALUES (?, ?)", key, value);
Util.flushTable(KEYSPACE, table);
ResultSet rs = executeNetWithPaging("SELECT * FROM vts.partition_key_statistics WHERE keyspace_name = ? AND table_name = ?",
10, KEYSPACE, table);
List<Row> all = rs.all();
assertEquals(1, all.size());
Row row = all.get(0);
assertEquals(BigInteger.valueOf(1), row.get("token_value", BigInteger.class));
long size = row.get("size_estimate", Long.class);
// providing a range since with timestamp delta vint encoding worried this may drift with time or in wierd
// VMs so just want to make sure it's in the right ballpark
assertTrue(size >= 110 && size < 200);
assertEquals(2L, row.get("sstables", Long.class).longValue());
assertEquals(2, scanned.get());
}
private static void assertResults(List<Row> all, int start, int end)
{
for (int i = start, offset = 0; i < end; i++, offset++)
{
Row row = all.get(offset);
}
}
}