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

Record latencies for SAI post-filtering reads against local storage 

patch by Caleb Rackliffe; reviewed by Mike Adamson for CASSANDRA-18940
This commit is contained in:
Caleb Rackliffe 2024-02-22 15:08:23 -06:00
parent 939fbfa63f
commit 1d7bae3697
6 changed files with 33 additions and 34 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.0-beta2
* Record latencies for SAI post-filtering reads against local storage (CASSANDRA-18940)
* Fix VectorMemoryIndex#update logic to compare vectors. Fix Index view (CASSANDRA-19168)
* Deprecate native_transport_port_ssl (CASSANDRA-19392)
* Update packaging shell includes (CASSANDRA-19283)

4
src/java/org/apache/cassandra/index/sai/metrics/TableQueryMetrics.java
View File

@ -32,6 +32,8 @@ public class TableQueryMetrics extends AbstractMetrics
{
public static final String TABLE_QUERY_METRIC_TYPE = "TableQueryMetrics";
public final Timer postFilteringReadLatency;
private final PerQueryMetrics perQueryMetrics;
private final Counter totalQueryTimeouts;
@ -45,6 +47,8 @@ public class TableQueryMetrics extends AbstractMetrics
perQueryMetrics = new PerQueryMetrics(table);
postFilteringReadLatency = Metrics.timer(createMetricName("PostFilteringReadLatency"));
totalPartitionReads = Metrics.counter(createMetricName("TotalPartitionReads"));
totalRowsFiltered = Metrics.counter(createMetricName("TotalRowsFiltered"));
totalQueriesCompleted = Metrics.counter(createMetricName("TotalQueriesCompleted"));

14
src/java/org/apache/cassandra/index/sai/plan/QueryController.java
View File

@ -57,7 +57,6 @@ import org.apache.cassandra.index.sai.iterators.KeyRangeIntersectionIterator;
import org.apache.cassandra.index.sai.iterators.KeyRangeIterator;
import org.apache.cassandra.index.sai.iterators.KeyRangeOrderingIterator;
import org.apache.cassandra.index.sai.iterators.KeyRangeUnionIterator;
import org.apache.cassandra.index.sai.metrics.TableQueryMetrics;
import org.apache.cassandra.index.sai.utils.PrimaryKey;
import org.apache.cassandra.net.ParamType;
import org.apache.cassandra.schema.TableMetadata;
@ -73,7 +72,6 @@ public class QueryController
private final ColumnFamilyStore cfs;
private final ReadCommand command;
private final QueryContext queryContext;
private final TableQueryMetrics tableQueryMetrics;
private final RowFilter filterOperation;
private final List<DataRange> ranges;
private final AbstractBounds<PartitionPosition> mergeRange;
@ -85,13 +83,11 @@ public class QueryController
public QueryController(ColumnFamilyStore cfs,
ReadCommand command,
RowFilter filterOperation,
QueryContext queryContext,
TableQueryMetrics tableQueryMetrics)
QueryContext queryContext)
{
this.cfs = cfs;
this.command = command;
this.queryContext = queryContext;
this.tableQueryMetrics = tableQueryMetrics;
this.filterOperation = filterOperation;
this.ranges = dataRanges(command);
DataRange first = ranges.get(0);
@ -249,14 +245,6 @@ public class QueryController
return key.kind() == PrimaryKey.Kind.WIDE && !command.clusteringIndexFilter(key.partitionKey()).selects(key.clustering());
}
/**
* Used to release all resources and record metrics when query finishes.
*/
public void finish()
{
if (tableQueryMetrics != null) tableQueryMetrics.record(queryContext);
}
// This is an ANN only query
public KeyRangeIterator getTopKRows(RowFilter.Expression expression)
{

33
src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java
View File

@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@ -54,12 +55,14 @@ import org.apache.cassandra.index.sai.utils.PrimaryKey;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.Clock;
public class StorageAttachedIndexSearcher implements Index.Searcher
{
private final ReadCommand command;
private final QueryController queryController;
private final QueryContext queryContext;
private final TableQueryMetrics tableQueryMetrics;
public StorageAttachedIndexSearcher(ColumnFamilyStore cfs,
TableQueryMetrics tableQueryMetrics,
@ -69,7 +72,8 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
{
this.command = command;
this.queryContext = new QueryContext(command, executionQuotaMs);
this.queryController = new QueryController(cfs, command, filterOperation, queryContext, tableQueryMetrics);
this.queryController = new QueryController(cfs, command, filterOperation, queryContext);
this.tableQueryMetrics = tableQueryMetrics;
}
@Override
@ -95,10 +99,10 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
public UnfilteredPartitionIterator search(ReadExecutionController executionController) throws RequestTimeoutException
{
if (!command.isTopK())
return new ResultRetriever(queryController, executionController, queryContext, false);
return new ResultRetriever(executionController, false);
else
{
Supplier<ResultRetriever> resultSupplier = () -> new ResultRetriever(queryController, executionController, queryContext, true);
Supplier<ResultRetriever> resultSupplier = () -> new ResultRetriever(executionController, true);
// VSTODO performance: if there is shadowed primary keys, we have to at least query twice.
// First time to find out there are shadow keys, second time to find out there are no more shadow keys.
@ -115,7 +119,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
}
}
private static class ResultRetriever extends AbstractIterator<UnfilteredRowIterator> implements UnfilteredPartitionIterator
private class ResultRetriever extends AbstractIterator<UnfilteredRowIterator> implements UnfilteredPartitionIterator
{
private final PrimaryKey firstPrimaryKey;
private final PrimaryKey lastPrimaryKey;
@ -124,26 +128,20 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
private final KeyRangeIterator resultKeyIterator;
private final FilterTree filterTree;
private final QueryController queryController;
private final ReadExecutionController executionController;
private final QueryContext queryContext;
private final PrimaryKey.Factory keyFactory;
private final boolean topK;
private PrimaryKey lastKey;
private ResultRetriever(QueryController queryController,
ReadExecutionController executionController,
QueryContext queryContext,
private ResultRetriever(ReadExecutionController executionController,
boolean topK)
{
this.keyRanges = queryController.dataRanges().iterator();
this.currentKeyRange = keyRanges.next().keyRange();
this.resultKeyIterator = Operation.buildIterator(queryController);
this.filterTree = Operation.buildFilter(queryController);
this.queryController = queryController;
this.executionController = executionController;
this.queryContext = queryContext;
this.keyFactory = queryController.primaryKeyFactory();
this.firstPrimaryKey = queryController.firstPrimaryKeyInRange();
this.lastPrimaryKey = queryController.lastPrimaryKeyInRange();
@ -370,13 +368,20 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
return null;
lastKey = key;
long startTimeNanos = Clock.Global.nanoTime();
try (UnfilteredRowIterator partition = queryController.queryStorage(key, executionController))
{
queryContext.partitionsRead++;
queryContext.checkpoint();
return applyIndexFilter(key, partition, filterTree, queryContext);
UnfilteredRowIterator filtered = applyIndexFilter(key, partition, filterTree, queryContext);
// Note that we record the duration of the read after post-filtering, which actually
// materializes the rows from disk.
tableQueryMetrics.postFilteringReadLatency.update(Clock.Global.nanoTime() - startTimeNanos, TimeUnit.NANOSECONDS);
return filtered;
}
}
@ -428,7 +433,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
return new PartitionIterator(partition, staticRow, Iterators.filter(clusters.iterator(), u -> !((Row)u).isStatic()));
}
private static class PartitionIterator extends AbstractUnfilteredRowIterator
private class PartitionIterator extends AbstractUnfilteredRowIterator
{
private final Iterator<Unfiltered> rows;
@ -462,7 +467,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
public void close()
{
FileUtils.closeQuietly(resultKeyIterator);
queryController.finish();
if (tableQueryMetrics != null) tableQueryMetrics.record(queryContext);
}
}

6
test/unit/org/apache/cassandra/index/sai/metrics/QueryMetricsTest.java
View File

@ -38,7 +38,7 @@ public class QueryMetricsTest extends AbstractMetricsTest
public ExpectedException exception = ExpectedException.none();
@Test
public void testSameIndexNameAcrossKeyspaces() throws Throwable
public void testSameIndexNameAcrossKeyspaces()
{
String table = "test_same_index_name_across_keyspaces";
String index = "test_same_index_name_across_keyspaces_index";
@ -58,7 +58,9 @@ public class QueryMetricsTest extends AbstractMetricsTest
assertEquals(1, rows.all().size());
assertEquals(1L, getTableQueryMetrics(keyspace1, table, "TotalQueriesCompleted"));
assertEquals(1L, getTableQueryMetrics(keyspace1, table, "PostFilteringReadLatency"));
assertEquals(0L, getTableQueryMetrics(keyspace2, table, "TotalQueriesCompleted"));
assertEquals(0L, getTableQueryMetrics(keyspace2, table, "PostFilteringReadLatency"));
execute("INSERT INTO " + keyspace2 + '.' + table + " (id1, v1, v2) VALUES ('0', 0, '0')");
execute("INSERT INTO " + keyspace2 + '.' + table + " (id1, v1, v2) VALUES ('1', 1, '1')");
@ -71,6 +73,8 @@ public class QueryMetricsTest extends AbstractMetricsTest
assertEquals(2L, getTableQueryMetrics(keyspace1, table, "TotalQueriesCompleted"));
assertEquals(1L, getTableQueryMetrics(keyspace2, table, "TotalQueriesCompleted"));
assertEquals(2L, getTableQueryMetrics(keyspace1, table, "PostFilteringReadLatency"));
assertEquals(1L, getTableQueryMetrics(keyspace2, table, "PostFilteringReadLatency"));
}
@Test

9
test/unit/org/apache/cassandra/index/sai/plan/OperationTest.java
View File

@ -115,22 +115,19 @@ public class OperationTest
controller = new QueryController(BACKEND,
command,
null,
new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS)),
null);
new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS)));
command = PartitionRangeReadCommand.allDataRead(CLUSTERING_BACKEND.metadata(), FBUtilities.nowInSeconds());
controllerClustering = new QueryController(CLUSTERING_BACKEND,
command,
null,
new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS)),
null);
new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS)));
command = PartitionRangeReadCommand.allDataRead(STATIC_BACKEND.metadata(), FBUtilities.nowInSeconds());
controllerStatic = new QueryController(STATIC_BACKEND,
command,
null,
new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS)),
null);
new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS)));
}
@Test