You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2023/07/26 16:27:25 UTC
[cassandra] 01/22: Add Index Group Interface for Storage Attached Index
This is an automated email from the ASF dual-hosted git repository.
maedhroz pushed a commit to branch cep-7-sai
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit dd0285599d1935067f129cdf4429b51b13cdf792
Author: Zhao Yang <zh...@gmail.com>
AuthorDate: Tue May 19 17:08:36 2020 +0800
Add Index Group Interface for Storage Attached Index
patch by Zhao Yang; reviewed by Caleb Rackliffe and Andres de la Peña for CASSANDRA-16092
Co-authored-by: Caleb Rackliffe <ca...@gmail.com>
Co-authored-by: Andres de la Peña <a....@gmail.com>
---
.../cassandra/db/CassandraTableWriteHandler.java | 5 +-
.../org/apache/cassandra/db/ColumnFamilyStore.java | 21 +-
src/java/org/apache/cassandra/db/Keyspace.java | 8 +-
.../cassandra/db/PartitionRangeReadCommand.java | 38 +-
src/java/org/apache/cassandra/db/ReadCommand.java | 94 ++--
.../cassandra/db/ReadExecutionController.java | 12 +-
.../cassandra/db/SinglePartitionReadCommand.java | 36 +-
.../org/apache/cassandra/db/TableWriteHandler.java | 3 +-
.../db/compaction/AbstractCompactionStrategy.java | 13 +-
.../db/compaction/AbstractStrategyHolder.java | 2 +-
.../db/compaction/CompactionIterator.java | 5 +-
.../cassandra/db/compaction/CompactionManager.java | 8 +-
.../db/compaction/CompactionStrategyHolder.java | 4 +-
.../db/compaction/CompactionStrategyManager.java | 4 +-
.../db/compaction/PendingRepairHolder.java | 4 +-
.../apache/cassandra/db/compaction/Upgrader.java | 10 +-
.../compaction/writers/CompactionAwareWriter.java | 8 +-
.../writers/DefaultCompactionWriter.java | 8 +-
.../writers/MajorLeveledCompactionWriter.java | 2 +-
src/java/org/apache/cassandra/index/Index.java | 380 ++++++++++++--
.../org/apache/cassandra/index/IndexRegistry.java | 97 +++-
.../cassandra/index/SecondaryIndexManager.java | 306 ++++++++---
.../cassandra/index/SingletonIndexGroup.java | 114 +++++
.../cassandra/index/SingletonIndexQueryPlan.java | 87 ++++
.../cassandra/index/internal/CassandraIndex.java | 21 +-
.../index/internal/CassandraIndexSearcher.java | 6 +
.../org/apache/cassandra/index/sasi/SASIIndex.java | 30 +-
.../sasi/SASIIndexGroup.java} | 21 +-
.../index/sasi/disk/PerSSTableIndexWriter.java | 10 +-
.../index/sasi/plan/SASIIndexQueryPlan.java | 51 ++
.../{QueryPlan.java => SASIIndexSearcher.java} | 17 +-
.../org/apache/cassandra/io/sstable/SSTable.java | 4 -
.../cassandra/io/sstable/SSTableFlushObserver.java | 21 +-
.../cassandra/io/sstable/SSTableTxnWriter.java | 10 +-
.../io/sstable/SimpleSSTableMultiWriter.java | 8 +-
.../cassandra/io/sstable/format/SSTableWriter.java | 49 +-
.../io/sstable/format/SortedTableWriter.java | 6 +-
.../cassandra/io/sstable/format/big/BigFormat.java | 2 +-
.../io/sstable/format/big/BigTableWriter.java | 10 +-
.../io/sstable/format/bti/BtiTableWriter.java | 6 +-
.../metrics/ClientRangeRequestMetrics.java | 47 ++
.../org/apache/cassandra/schema/IndexMetadata.java | 22 +-
.../paxos/uncommitted/PaxosUncommittedIndex.java | 9 +-
.../cassandra/service/reads/DataResolver.java | 32 +-
.../service/reads/ShortReadRowsProtection.java | 2 +-
.../service/reads/range/RangeCommandIterator.java | 5 +-
.../service/reads/range/RangeCommands.java | 2 +-
.../upgrade/MixedModeIndexTestBase.java | 135 +++++
.../distributed/upgrade/MixedModeIndexV40Test.java | 16 +-
.../apache/cassandra/db/SecondaryIndexTest.java | 12 +-
.../cassandra/db/SerializationHeaderTest.java | 6 +-
.../db/compaction/ActiveCompactionsTest.java | 4 +-
.../db/lifecycle/RealTransactionsTest.java | 4 +-
.../apache/cassandra/index/CustomIndexTest.java | 563 ++++++++++++++++++++-
.../cassandra/index/SecondaryIndexManagerTest.java | 4 +-
.../unit/org/apache/cassandra/index/StubIndex.java | 30 +-
.../org/apache/cassandra/index/StubIndexGroup.java | 99 ++++
.../index/internal/CustomCassandraIndex.java | 19 +-
.../apache/cassandra/index/sasi/SASIIndexTest.java | 6 +-
.../index/sasi/disk/PerSSTableIndexWriterTest.java | 5 +-
.../io/sstable/SSTableFlushObserverTest.java | 5 +-
.../cassandra/io/sstable/SSTableReaderTest.java | 2 +-
.../io/sstable/SSTableWriterTestBase.java | 4 +-
.../org/apache/cassandra/io/sstable/ScrubTest.java | 2 +-
.../metrics/ClientRequestMetricsTest.java | 23 +-
65 files changed, 2142 insertions(+), 457 deletions(-)
diff --git a/src/java/org/apache/cassandra/db/CassandraTableWriteHandler.java b/src/java/org/apache/cassandra/db/CassandraTableWriteHandler.java
index 146539c26e..671c06d1e5 100644
--- a/src/java/org/apache/cassandra/db/CassandraTableWriteHandler.java
+++ b/src/java/org/apache/cassandra/db/CassandraTableWriteHandler.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.db;
import org.apache.cassandra.db.partitions.PartitionUpdate;
-import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.tracing.Tracing;
public class CassandraTableWriteHandler implements TableWriteHandler
@@ -33,10 +32,10 @@ public class CassandraTableWriteHandler implements TableWriteHandler
@Override
@SuppressWarnings("resource")
- public void write(PartitionUpdate update, WriteContext context, UpdateTransaction updateTransaction)
+ public void write(PartitionUpdate update, WriteContext context, boolean updateIndexes)
{
CassandraWriteContext ctx = CassandraWriteContext.fromContext(context);
Tracing.trace("Adding to {} memtable", update.metadata().name);
- cfs.apply(update, updateTransaction, ctx.getGroup(), ctx.getPosition());
+ cfs.apply(update, ctx, updateIndexes);
}
}
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index ef18097a0a..5631faac71 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -660,7 +660,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
public SSTableMultiWriter createSSTableMultiWriter(Descriptor descriptor, long keyCount, long repairedAt, TimeUUID pendingRepair, boolean isTransient, IntervalSet<CommitLogPosition> commitLogPositions, int sstableLevel, SerializationHeader header, LifecycleNewTracker lifecycleNewTracker)
{
- return getCompactionStrategyManager().createSSTableMultiWriter(descriptor, keyCount, repairedAt, pendingRepair, isTransient, commitLogPositions, sstableLevel, header, indexManager.listIndexes(), lifecycleNewTracker);
+ return getCompactionStrategyManager().createSSTableMultiWriter(descriptor, keyCount, repairedAt, pendingRepair, isTransient, commitLogPositions, sstableLevel, header, indexManager.listIndexGroups(), lifecycleNewTracker);
}
public boolean supportsEarlyOpen()
@@ -1423,17 +1423,21 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
/**
* Insert/Update the column family for this key.
* Caller is responsible for acquiring Keyspace.switchLock
- * param @ lock - lock that needs to be used.
- * param @ key - key for update/insert
- * param @ columnFamily - columnFamily changes
+ * @param update to be applied
+ * @param context write context for current update
+ * @param updateIndexes whether secondary indexes should be updated
*/
- public void apply(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, CommitLogPosition commitLogPosition)
+ @SuppressWarnings("resource") // opGroup
+ public void apply(PartitionUpdate update, CassandraWriteContext context, boolean updateIndexes)
{
long start = nanoTime();
+ OpOrder.Group opGroup = context.getGroup();
+ CommitLogPosition commitLogPosition = context.getPosition();
try
{
Memtable mt = data.getMemtableFor(opGroup, commitLogPosition);
+ UpdateTransaction indexer = newUpdateTransaction(update, context, updateIndexes, mt);
long timeDelta = mt.put(update, indexer, opGroup);
DecoratedKey key = update.partitionKey();
invalidateCachedPartition(key);
@@ -1457,6 +1461,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
+ getKeyspaceName() + ", table: " + name, e);
}
}
+
+ private UpdateTransaction newUpdateTransaction(PartitionUpdate update, CassandraWriteContext context, boolean updateIndexes, Memtable memtable)
+ {
+ return updateIndexes
+ ? indexManager.newUpdateTransaction(update, context, FBUtilities.nowInSeconds(), memtable)
+ : UpdateTransaction.NO_OP;
+ }
public static class VersionedLocalRanges extends ArrayList<Splitter.WeightedRange>
{
diff --git a/src/java/org/apache/cassandra/db/Keyspace.java b/src/java/org/apache/cassandra/db/Keyspace.java
index a6df38646a..a3f05e6048 100644
--- a/src/java/org/apache/cassandra/db/Keyspace.java
+++ b/src/java/org/apache/cassandra/db/Keyspace.java
@@ -50,7 +50,6 @@ import org.apache.cassandra.db.view.ViewManager;
import org.apache.cassandra.exceptions.WriteTimeoutException;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.SecondaryIndexManager;
-import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
@@ -67,7 +66,6 @@ import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.service.snapshot.TableSnapshot;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
@@ -611,7 +609,6 @@ public class Keyspace
columnFamilyStores.get(tableId).metric.viewLockAcquireTime.update(acquireTime, MILLISECONDS);
}
}
- long nowInSec = FBUtilities.nowInSeconds();
try (WriteContext ctx = getWriteHandler().beginWrite(mutation, makeDurable))
{
for (PartitionUpdate upd : mutation.getPartitionUpdates())
@@ -640,10 +637,7 @@ public class Keyspace
}
}
- UpdateTransaction indexTransaction = updateIndexes
- ? cfs.indexManager.newUpdateTransaction(upd, ctx, nowInSec)
- : UpdateTransaction.NO_OP;
- cfs.getWriteHandler().write(upd, ctx, indexTransaction);
+ cfs.getWriteHandler().write(upd, ctx, updateIndexes);
if (requiresViewUpdate)
baseComplete.set(currentTimeMillis());
diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index 36d7e93681..88b38a4d7c 100644
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@ -50,7 +50,6 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.metrics.TableMetrics;
import org.apache.cassandra.net.Verb;
-import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.StorageProxy;
@@ -75,10 +74,10 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
RowFilter rowFilter,
DataLimits limits,
DataRange dataRange,
- IndexMetadata index,
+ Index.QueryPlan indexQueryPlan,
boolean trackWarnings)
{
- super(Kind.PARTITION_RANGE, isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, index, trackWarnings);
+ super(Kind.PARTITION_RANGE, isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, indexQueryPlan, trackWarnings);
this.dataRange = dataRange;
this.requestedSlices = dataRange.clusteringIndexFilter.getSlices(metadata());
@@ -93,7 +92,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
RowFilter rowFilter,
DataLimits limits,
DataRange dataRange,
- IndexMetadata index,
+ Index.QueryPlan indexQueryPlan,
boolean trackWarnings)
{
if (metadata.isVirtual())
@@ -107,7 +106,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
rowFilter,
limits,
dataRange,
- index,
+ indexQueryPlan,
trackWarnings);
}
return new PartitionRangeReadCommand(isDigest,
@@ -119,7 +118,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
rowFilter,
limits,
dataRange,
- index,
+ indexQueryPlan,
trackWarnings);
}
@@ -139,7 +138,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
rowFilter,
limits,
dataRange,
- findIndex(metadata, rowFilter),
+ findIndexQueryPlan(metadata, rowFilter),
false);
}
@@ -212,7 +211,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
rowFilter(),
isRangeContinuation ? limits() : limits().withoutState(),
dataRange().forSubRange(range),
- indexMetadata(),
+ indexQueryPlan(),
isTrackingWarnings());
}
@@ -227,7 +226,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
rowFilter(),
limits(),
dataRange(),
- indexMetadata(),
+ indexQueryPlan(),
isTrackingWarnings());
}
@@ -243,7 +242,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
rowFilter(),
limits(),
dataRange(),
- indexMetadata(),
+ indexQueryPlan(),
isTrackingWarnings());
}
@@ -259,7 +258,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
rowFilter(),
limits(),
dataRange(),
- indexMetadata(),
+ indexQueryPlan(),
isTrackingWarnings());
}
@@ -275,7 +274,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
rowFilter(),
newLimits,
dataRange(),
- indexMetadata(),
+ indexQueryPlan(),
isTrackingWarnings());
}
@@ -291,7 +290,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
rowFilter(),
newLimits,
newDataRange,
- indexMetadata(),
+ indexQueryPlan(),
isTrackingWarnings());
}
@@ -472,9 +471,8 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
*/
public PartitionIterator postReconciliationProcessing(PartitionIterator result)
{
- ColumnFamilyStore cfs = Keyspace.open(metadata().keyspace).getColumnFamilyStore(metadata().name);
- Index index = getIndex(cfs);
- return index == null ? result : index.postProcessorFor(this).apply(result, this);
+ Index.QueryPlan queryPlan = indexQueryPlan();
+ return queryPlan == null ? result : queryPlan.postProcessor().apply(result);
}
@Override
@@ -528,11 +526,11 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
ColumnFilter columnFilter,
RowFilter rowFilter,
DataLimits limits,
- IndexMetadata index)
+ Index.QueryPlan indexQueryPlan)
throws IOException
{
DataRange range = DataRange.serializer.deserialize(in, version, metadata);
- return PartitionRangeReadCommand.create(isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, range, index, false);
+ return new PartitionRangeReadCommand(isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, range, indexQueryPlan, false);
}
}
@@ -547,10 +545,10 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
RowFilter rowFilter,
DataLimits limits,
DataRange dataRange,
- IndexMetadata index,
+ Index.QueryPlan indexQueryPlan,
boolean trackWarnings)
{
- super(isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, dataRange, index, trackWarnings);
+ super(isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, dataRange, indexQueryPlan, trackWarnings);
}
@Override
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java
index 2979b69180..d46ce1efa2 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.LongPredicate;
import java.util.function.Function;
+import java.util.stream.Collectors;
import javax.annotation.Nullable;
@@ -51,8 +52,6 @@ import org.apache.cassandra.db.transform.StoppingTransformation;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.exceptions.UnknownIndexException;
import org.apache.cassandra.index.Index;
-import org.apache.cassandra.index.IndexNotAvailableException;
-import org.apache.cassandra.index.IndexRegistry;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputPlus;
@@ -107,7 +106,7 @@ public abstract class ReadCommand extends AbstractReadQuery
private boolean trackWarnings;
@Nullable
- private final IndexMetadata index;
+ private final Index.QueryPlan indexQueryPlan;
protected static abstract class SelectionDeserializer
{
@@ -121,7 +120,7 @@ public abstract class ReadCommand extends AbstractReadQuery
ColumnFilter columnFilter,
RowFilter rowFilter,
DataLimits limits,
- IndexMetadata index) throws IOException;
+ Index.QueryPlan indexQueryPlan) throws IOException;
}
protected enum Kind
@@ -146,7 +145,7 @@ public abstract class ReadCommand extends AbstractReadQuery
ColumnFilter columnFilter,
RowFilter rowFilter,
DataLimits limits,
- IndexMetadata index,
+ Index.QueryPlan indexQueryPlan,
boolean trackWarnings)
{
super(metadata, nowInSec, columnFilter, rowFilter, limits);
@@ -157,7 +156,7 @@ public abstract class ReadCommand extends AbstractReadQuery
this.isDigestQuery = isDigestQuery;
this.digestVersion = digestVersion;
this.acceptsTransient = acceptsTransient;
- this.index = index;
+ this.indexQueryPlan = indexQueryPlan;
this.trackWarnings = trackWarnings;
}
@@ -245,14 +244,20 @@ public abstract class ReadCommand extends AbstractReadQuery
}
/**
- * Index (metadata) chosen for this query. Can be null.
+ * Index query plan chosen for this query. Can be null.
*
- * @return index (metadata) chosen for this query
+ * @return index query plan chosen for this query
*/
@Nullable
- public IndexMetadata indexMetadata()
+ public Index.QueryPlan indexQueryPlan()
{
- return index;
+ return indexQueryPlan;
+ }
+
+ @VisibleForTesting
+ public Index.Searcher indexSearcher()
+ {
+ return indexQueryPlan == null ? null : indexQueryPlan.searcherFor(this);
}
/**
@@ -353,30 +358,19 @@ public abstract class ReadCommand extends AbstractReadQuery
long indexSerializedSize(int version)
{
- return null != index
- ? IndexMetadata.serializer.serializedSize(index, version)
+ return null != indexQueryPlan
+ ? IndexMetadata.serializer.serializedSize(indexQueryPlan.getFirst().getIndexMetadata(), version)
: 0;
}
- public Index getIndex(ColumnFamilyStore cfs)
- {
- return null != index
- ? cfs.indexManager.getIndex(index)
- : null;
- }
-
- static IndexMetadata findIndex(TableMetadata table, RowFilter rowFilter)
+ static Index.QueryPlan findIndexQueryPlan(TableMetadata table, RowFilter rowFilter)
{
if (table.indexes.isEmpty() || rowFilter.isEmpty())
return null;
ColumnFamilyStore cfs = Keyspace.openAndGetStore(table);
- Index index = cfs.indexManager.getBestIndexFor(rowFilter);
-
- return null != index
- ? index.getIndexMetadata()
- : null;
+ return cfs.indexManager.getBestIndexQueryPlanFor(rowFilter);
}
/**
@@ -387,8 +381,10 @@ public abstract class ReadCommand extends AbstractReadQuery
*/
public void maybeValidateIndex()
{
- if (null != index)
- IndexRegistry.obtain(metadata()).getIndex(index).validate(this);
+ if (null != indexQueryPlan)
+ {
+ indexQueryPlan.validate(this);
+ }
}
/**
@@ -408,16 +404,23 @@ public abstract class ReadCommand extends AbstractReadQuery
try
{
ColumnFamilyStore cfs = Keyspace.openAndGetStore(metadata());
- Index index = getIndex(cfs);
+ Index.QueryPlan indexQueryPlan = indexQueryPlan();
Index.Searcher searcher = null;
- if (index != null)
+ if (indexQueryPlan != null)
{
- if (!cfs.indexManager.isIndexQueryable(index))
- throw new IndexNotAvailableException(index);
-
- searcher = index.searcherFor(this);
- Tracing.trace("Executing read on {}.{} using index {}", cfs.metadata.keyspace, cfs.metadata.name, index.getIndexMetadata().name);
+ cfs.indexManager.checkQueryability(indexQueryPlan);
+
+ Index index = indexQueryPlan.getFirst();
+ searcher = indexQueryPlan.searcherFor(this);
+ Tracing.trace("Executing read on {}.{} using index{} {}",
+ cfs.metadata.keyspace,
+ cfs.metadata.name,
+ indexQueryPlan.getIndexes().size() == 1 ? "" : "es",
+ indexQueryPlan.getIndexes()
+ .stream()
+ .map(i -> i.getIndexMetadata().name)
+ .collect(Collectors.joining(",")));
}
UnfilteredPartitionIterator iterator = (null == searcher) ? queryStorage(cfs, executionController) : searcher.search(executionController);
@@ -433,7 +436,7 @@ public abstract class ReadCommand extends AbstractReadQuery
// If we've used a 2ndary index, we know the result already satisfy the primary expression used, so
// no point in checking it again.
- RowFilter filter = (null == searcher) ? rowFilter() : index.getPostIndexQueryFilter(rowFilter());
+ RowFilter filter = (null == searcher) ? rowFilter() : indexQueryPlan.postIndexQueryFilter();
/*
* TODO: We'll currently do filtering by the rowFilter here because it's convenient. However,
@@ -1067,7 +1070,7 @@ public abstract class ReadCommand extends AbstractReadQuery
out.writeByte(command.kind.ordinal());
out.writeByte(
digestFlag(command.isDigestQuery())
- | indexFlag(null != command.indexMetadata())
+ | indexFlag(null != command.indexQueryPlan())
| acceptsTransientFlag(command.acceptsTransient())
);
if (command.isDigestQuery())
@@ -1077,8 +1080,11 @@ public abstract class ReadCommand extends AbstractReadQuery
ColumnFilter.serializer.serialize(command.columnFilter(), out, version);
RowFilter.serializer.serialize(command.rowFilter(), out, version);
DataLimits.serializer.serialize(command.limits(), out, version, command.metadata().comparator);
- if (null != command.index)
- IndexMetadata.serializer.serialize(command.index, out, version);
+ // Using the name of one of the indexes in the plan to identify the index group because we want
+ // to keep compatibility with legacy nodes. Each replica can create its own different index query plan
+ // from the index name.
+ if (null != command.indexQueryPlan)
+ IndexMetadata.serializer.serialize(command.indexQueryPlan.getFirst().getIndexMetadata(), out, version);
command.serializeSelection(out, version);
}
@@ -1104,9 +1110,17 @@ public abstract class ReadCommand extends AbstractReadQuery
ColumnFilter columnFilter = ColumnFilter.serializer.deserialize(in, version, metadata);
RowFilter rowFilter = RowFilter.serializer.deserialize(in, version, metadata);
DataLimits limits = DataLimits.serializer.deserialize(in, version, metadata);
- IndexMetadata index = hasIndex ? deserializeIndexMetadata(in, version, metadata) : null;
- return kind.selectionDeserializer.deserialize(in, version, isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, index);
+ Index.QueryPlan indexQueryPlan = null;
+ if (hasIndex)
+ {
+ IndexMetadata index = deserializeIndexMetadata(in, version, metadata);
+ Index.Group indexGroup = Keyspace.openAndGetStore(metadata).indexManager.getIndexGroup(index);
+ if (indexGroup != null)
+ indexQueryPlan = indexGroup.queryPlanFor(rowFilter);
+ }
+
+ return kind.selectionDeserializer.deserialize(in, version, isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, indexQueryPlan);
}
private IndexMetadata deserializeIndexMetadata(DataInputPlus in, int version, TableMetadata metadata) throws IOException
diff --git a/src/java/org/apache/cassandra/db/ReadExecutionController.java b/src/java/org/apache/cassandra/db/ReadExecutionController.java
index c2ecfe27d8..cd86350385 100644
--- a/src/java/org/apache/cassandra/db/ReadExecutionController.java
+++ b/src/java/org/apache/cassandra/db/ReadExecutionController.java
@@ -124,7 +124,7 @@ public class ReadExecutionController implements AutoCloseable
static ReadExecutionController forCommand(ReadCommand command, boolean trackRepairedStatus)
{
ColumnFamilyStore baseCfs = Keyspace.openAndGetStore(command.metadata());
- ColumnFamilyStore indexCfs = maybeGetIndexCfs(baseCfs, command);
+ ColumnFamilyStore indexCfs = maybeGetIndexCfs(command);
long createdAtNanos = baseCfs.metric.topLocalReadQueryTime.isEnabled() ? clock.now() : NO_SAMPLING;
@@ -165,10 +165,14 @@ public class ReadExecutionController implements AutoCloseable
}
}
- private static ColumnFamilyStore maybeGetIndexCfs(ColumnFamilyStore baseCfs, ReadCommand command)
+ private static ColumnFamilyStore maybeGetIndexCfs(ReadCommand command)
{
- Index index = command.getIndex(baseCfs);
- return index == null ? null : index.getBackingTable().orElse(null);
+ Index.QueryPlan queryPlan = command.indexQueryPlan();
+ if (queryPlan == null)
+ return null;
+
+ // only the index groups with a single member are allowed to have a backing table
+ return queryPlan.getFirst().getBackingTable().orElse(null);
}
public TableMetadata metadata()
diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
index 87ff16a1cd..2ec10a6093 100644
--- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
+++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
@@ -64,6 +64,7 @@ import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.db.virtual.VirtualKeyspaceRegistry;
import org.apache.cassandra.db.virtual.VirtualTable;
import org.apache.cassandra.exceptions.RequestExecutionException;
+import org.apache.cassandra.index.Index;
import org.apache.cassandra.io.sstable.SSTableReadsListener;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputPlus;
@@ -71,7 +72,6 @@ import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.metrics.TableMetrics;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.ColumnMetadata;
-import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.service.ClientState;
@@ -101,10 +101,10 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
DataLimits limits,
DecoratedKey partitionKey,
ClusteringIndexFilter clusteringIndexFilter,
- IndexMetadata index,
+ Index.QueryPlan indexQueryPlan,
boolean trackWarnings)
{
- super(Kind.SINGLE_PARTITION, isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, index, trackWarnings);
+ super(Kind.SINGLE_PARTITION, isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, indexQueryPlan, trackWarnings);
assert partitionKey.getPartitioner() == metadata.partitioner;
this.partitionKey = partitionKey;
this.clusteringIndexFilter = clusteringIndexFilter;
@@ -120,7 +120,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
DataLimits limits,
DecoratedKey partitionKey,
ClusteringIndexFilter clusteringIndexFilter,
- IndexMetadata index,
+ Index.QueryPlan indexQueryPlan,
boolean trackWarnings)
{
if (metadata.isVirtual())
@@ -135,7 +135,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
limits,
partitionKey,
clusteringIndexFilter,
- index,
+ indexQueryPlan,
trackWarnings);
}
return new SinglePartitionReadCommand(isDigest,
@@ -148,7 +148,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
limits,
partitionKey,
clusteringIndexFilter,
- index,
+ indexQueryPlan,
trackWarnings);
}
@@ -162,7 +162,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
* @param limits the limits to use for the query.
* @param partitionKey the partition key for the partition to query.
* @param clusteringIndexFilter the clustering index filter to use for the query.
- * @param indexMetadata explicitly specified index to use for the query
+ * @param indexQueryPlan explicitly specified index to use for the query
*
* @return a newly created read command.
*/
@@ -173,7 +173,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
DataLimits limits,
DecoratedKey partitionKey,
ClusteringIndexFilter clusteringIndexFilter,
- IndexMetadata indexMetadata)
+ Index.QueryPlan indexQueryPlan)
{
return create(false,
0,
@@ -185,7 +185,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
limits,
partitionKey,
clusteringIndexFilter,
- indexMetadata,
+ indexQueryPlan,
false);
}
@@ -217,7 +217,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
limits,
partitionKey,
clusteringIndexFilter,
- findIndex(metadata, rowFilter));
+ findIndexQueryPlan(metadata, rowFilter));
}
/**
@@ -362,7 +362,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
limits(),
partitionKey(),
clusteringIndexFilter(),
- indexMetadata(),
+ indexQueryPlan(),
isTrackingWarnings());
}
@@ -379,7 +379,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
limits(),
partitionKey(),
clusteringIndexFilter(),
- indexMetadata(),
+ indexQueryPlan(),
isTrackingWarnings());
}
@@ -396,7 +396,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
limits(),
partitionKey(),
clusteringIndexFilter(),
- indexMetadata(),
+ indexQueryPlan(),
isTrackingWarnings());
}
@@ -413,7 +413,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
newLimits,
partitionKey(),
clusteringIndexFilter(),
- indexMetadata(),
+ indexQueryPlan(),
isTrackingWarnings());
}
@@ -1318,12 +1318,12 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
ColumnFilter columnFilter,
RowFilter rowFilter,
DataLimits limits,
- IndexMetadata index)
+ Index.QueryPlan indexQueryPlan)
throws IOException
{
DecoratedKey key = metadata.partitioner.decorateKey(metadata.partitionKeyType.readBuffer(in, DatabaseDescriptor.getMaxValueSize()));
ClusteringIndexFilter filter = ClusteringIndexFilter.serializer.deserialize(in, version, metadata);
- return SinglePartitionReadCommand.create(isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, key, filter, index, false);
+ return SinglePartitionReadCommand.create(isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, key, filter, indexQueryPlan, false);
}
}
@@ -1367,10 +1367,10 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
DataLimits limits,
DecoratedKey partitionKey,
ClusteringIndexFilter clusteringIndexFilter,
- IndexMetadata index,
+ Index.QueryPlan indexQueryPlan,
boolean trackWarnings)
{
- super(isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, clusteringIndexFilter, index, trackWarnings);
+ super(isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, clusteringIndexFilter, indexQueryPlan, trackWarnings);
}
@Override
diff --git a/src/java/org/apache/cassandra/db/TableWriteHandler.java b/src/java/org/apache/cassandra/db/TableWriteHandler.java
index 4e47221925..7aa709fe52 100644
--- a/src/java/org/apache/cassandra/db/TableWriteHandler.java
+++ b/src/java/org/apache/cassandra/db/TableWriteHandler.java
@@ -19,9 +19,8 @@
package org.apache.cassandra.db;
import org.apache.cassandra.db.partitions.PartitionUpdate;
-import org.apache.cassandra.index.transactions.UpdateTransaction;
public interface TableWriteHandler
{
- void write(PartitionUpdate update, WriteContext context, UpdateTransaction updateTransaction);
+ void write(PartitionUpdate update, WriteContext context, boolean updateIndexes);
}
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
index facff1495d..34e148075d 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
@@ -254,10 +254,10 @@ public abstract class AbstractCompactionStrategy
* allow for a more memory efficient solution if we know the sstable don't overlap (see
* LeveledCompactionStrategy for instance).
*/
- @SuppressWarnings("resource")
+ @SuppressWarnings({"resource", "RedundantSuppression"})
public ScannerList getScanners(Collection<SSTableReader> sstables, Collection<Range<Token>> ranges)
{
- ArrayList<ISSTableScanner> scanners = new ArrayList<ISSTableScanner>();
+ ArrayList<ISSTableScanner> scanners = new ArrayList<>();
try
{
for (SSTableReader sstable : sstables)
@@ -327,9 +327,6 @@ public abstract class AbstractCompactionStrategy
*
* Not called when repair status changes (which is also metadata), because this results in the
* sstable getting removed from the compaction strategy instance.
- *
- * @param oldMetadata
- * @param sstable
*/
public void metadataChanged(StatsMetadata oldMetadata, SSTableReader sstable)
{
@@ -368,7 +365,7 @@ public abstract class AbstractCompactionStrategy
for (int i=0, isize=scanners.size(); i<isize; i++)
{
- @SuppressWarnings("resource")
+ @SuppressWarnings({"resource", "RedundantSuppression"})
ISSTableScanner scanner = scanners.get(i);
compressed += scanner.getCompressedLengthInBytes();
uncompressed += scanner.getLengthInBytes();
@@ -565,7 +562,7 @@ public abstract class AbstractCompactionStrategy
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
- Collection<Index> indexes,
+ Collection<Index.Group> indexGroups,
LifecycleNewTracker lifecycleNewTracker)
{
return SimpleSSTableMultiWriter.create(descriptor,
@@ -577,7 +574,7 @@ public abstract class AbstractCompactionStrategy
commitLogPositions,
sstableLevel,
header,
- indexes,
+ indexGroups,
lifecycleNewTracker, cfs);
}
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractStrategyHolder.java b/src/java/org/apache/cassandra/db/compaction/AbstractStrategyHolder.java
index 38123d32d5..97721cc9c3 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractStrategyHolder.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractStrategyHolder.java
@@ -198,7 +198,7 @@ public abstract class AbstractStrategyHolder
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
- Collection<Index> indexes,
+ Collection<Index.Group> indexGroups,
LifecycleNewTracker lifecycleNewTracker);
/**
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
index ce3e7d0363..a31e137e04 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
@@ -38,6 +38,7 @@ import org.apache.cassandra.db.EmptyIterators;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.db.transform.DuplicateRowChecker;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.partitions.PurgeFunction;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
@@ -50,10 +51,10 @@ 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.db.rows.WrappingUnfilteredRowIterator;
-import org.apache.cassandra.db.transform.DuplicateRowChecker;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.index.transactions.CompactionTransaction;
+import org.apache.cassandra.index.transactions.IndexTransaction;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.metrics.TopPartitionTracker;
@@ -221,7 +222,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
CompactionIterator.this.updateCounterFor(merged);
if ( (type != OperationType.COMPACTION && type != OperationType.MAJOR_COMPACTION)
- || !controller.cfs.indexManager.hasIndexes() )
+ || !controller.cfs.indexManager.handles(IndexTransaction.Type.COMPACTION) )
{
return null;
}
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index a32242d7c4..cf6904dda7 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -1623,8 +1623,8 @@ public class CompactionManager implements CompactionManagerMBean
.setTableMetadataRef(cfs.metadata)
.setMetadataCollector(new MetadataCollector(cfs.metadata().comparator).sstableLevel(sstable.getSSTableLevel()))
.setSerializationHeader(sstable.header)
- .addDefaultComponents()
- .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexes(), txn.opType())
+ .addDefaultComponents(cfs.indexManager.listIndexGroups())
+ .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get())
.build(txn, cfs);
}
@@ -1663,8 +1663,8 @@ public class CompactionManager implements CompactionManagerMBean
.setTableMetadataRef(cfs.metadata)
.setMetadataCollector(new MetadataCollector(sstables, cfs.metadata().comparator).sstableLevel(minLevel))
.setSerializationHeader(SerializationHeader.make(cfs.metadata(), sstables))
- .addDefaultComponents()
- .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexes(), txn.opType())
+ .addDefaultComponents(cfs.indexManager.listIndexGroups())
+ .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get())
.build(txn, cfs);
}
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionStrategyHolder.java b/src/java/org/apache/cassandra/db/compaction/CompactionStrategyHolder.java
index 9a31752d69..9b7f4a68e2 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionStrategyHolder.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionStrategyHolder.java
@@ -224,7 +224,7 @@ public class CompactionStrategyHolder extends AbstractStrategyHolder
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
- Collection<Index> indexes,
+ Collection<Index.Group> indexGroups,
LifecycleNewTracker lifecycleNewTracker)
{
if (isRepaired)
@@ -249,7 +249,7 @@ public class CompactionStrategyHolder extends AbstractStrategyHolder
commitLogPositions,
sstableLevel,
header,
- indexes,
+ indexGroups,
lifecycleNewTracker);
}
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java
index 05add93d7f..c082a1a2b2 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java
@@ -1237,7 +1237,7 @@ public class CompactionStrategyManager implements INotificationConsumer
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
- Collection<Index> indexes,
+ Collection<Index.Group> indexGroups,
LifecycleNewTracker lifecycleNewTracker)
{
SSTable.validateRepairedMetadata(repairedAt, pendingRepair, isTransient);
@@ -1253,7 +1253,7 @@ public class CompactionStrategyManager implements INotificationConsumer
commitLogPositions,
sstableLevel,
header,
- indexes,
+ indexGroups,
lifecycleNewTracker);
}
finally
diff --git a/src/java/org/apache/cassandra/db/compaction/PendingRepairHolder.java b/src/java/org/apache/cassandra/db/compaction/PendingRepairHolder.java
index 639481322a..534bf3ae82 100644
--- a/src/java/org/apache/cassandra/db/compaction/PendingRepairHolder.java
+++ b/src/java/org/apache/cassandra/db/compaction/PendingRepairHolder.java
@@ -242,7 +242,7 @@ public class PendingRepairHolder extends AbstractStrategyHolder
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
- Collection<Index> indexes,
+ Collection<Index.Group> indexGroups,
LifecycleNewTracker lifecycleNewTracker)
{
Preconditions.checkArgument(repairedAt == ActiveRepairService.UNREPAIRED_SSTABLE,
@@ -259,7 +259,7 @@ public class PendingRepairHolder extends AbstractStrategyHolder
commitLogPositions,
sstableLevel,
header,
- indexes,
+ indexGroups,
lifecycleNewTracker);
}
diff --git a/src/java/org/apache/cassandra/db/compaction/Upgrader.java b/src/java/org/apache/cassandra/db/compaction/Upgrader.java
index 58b3e89cd6..89e73646bc 100644
--- a/src/java/org/apache/cassandra/db/compaction/Upgrader.java
+++ b/src/java/org/apache/cassandra/db/compaction/Upgrader.java
@@ -17,7 +17,7 @@
*/
package org.apache.cassandra.db.compaction;
-import java.util.Arrays;
+import java.util.Collections;
import java.util.function.LongPredicate;
import com.google.common.base.Throwables;
@@ -64,8 +64,8 @@ public class Upgrader
this.controller = new UpgradeController(cfs);
this.strategyManager = cfs.getCompactionStrategyManager();
- long estimatedTotalKeys = Math.max(cfs.metadata().params.minIndexInterval, SSTableReader.getApproximateKeyCount(Arrays.asList(this.sstable)));
- long estimatedSSTables = Math.max(1, SSTableReader.getTotalBytes(Arrays.asList(this.sstable)) / strategyManager.getMaxSSTableBytes());
+ long estimatedTotalKeys = Math.max(cfs.metadata().params.minIndexInterval, SSTableReader.getApproximateKeyCount(Collections.singletonList(this.sstable)));
+ long estimatedSSTables = Math.max(1, SSTableReader.getTotalBytes(Collections.singletonList(this.sstable)) / strategyManager.getMaxSSTableBytes());
this.estimatedRows = (long) Math.ceil((double) estimatedTotalKeys / estimatedSSTables);
}
@@ -83,8 +83,8 @@ public class Upgrader
.setTableMetadataRef(cfs.metadata)
.setMetadataCollector(sstableMetadataCollector)
.setSerializationHeader(SerializationHeader.make(cfs.metadata(), Sets.newHashSet(sstable)))
- .addDefaultComponents()
- .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexes(), transaction.opType())
+ .addDefaultComponents(cfs.indexManager.listIndexGroups())
+ .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), transaction, cfs.metadata.get())
.build(transaction, cfs);
}
diff --git a/src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java b/src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java
index 82c56eeb2b..be014ff4b3 100644
--- a/src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java
+++ b/src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java
@@ -262,7 +262,6 @@ public abstract class CompactionAwareWriter extends Transactional.AbstractTransa
* Return a directory where we can expect expectedWriteSize to fit.
*
* @param sstables the sstables to compact
- * @return
*/
public Directories.DataDirectory getWriteDirectory(Iterable<SSTableReader> sstables, long estimatedWriteSize)
{
@@ -312,9 +311,6 @@ public abstract class CompactionAwareWriter extends Transactional.AbstractTransa
* - {@link SSTableWriter.Builder#setKeyCount(long)},
* - {@link SSTableWriter.Builder#setSerializationHeader(SerializationHeader)} and,
* - {@link SSTableWriter.Builder#setMetadataCollector(MetadataCollector)}
- *
- * @param descriptor
- * @return
*/
protected SSTableWriter.Builder<?, ?> newWriterBuilder(Descriptor descriptor)
{
@@ -323,7 +319,7 @@ public abstract class CompactionAwareWriter extends Transactional.AbstractTransa
.setTransientSSTable(isTransient)
.setRepairedAt(minRepairedAt)
.setPendingRepair(pendingRepair)
- .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexes(), txn.opType())
- .addDefaultComponents();
+ .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get())
+ .addDefaultComponents(cfs.indexManager.listIndexGroups());
}
}
diff --git a/src/java/org/apache/cassandra/db/compaction/writers/DefaultCompactionWriter.java b/src/java/org/apache/cassandra/db/compaction/writers/DefaultCompactionWriter.java
index 6c049b2343..4f64ff776e 100644
--- a/src/java/org/apache/cassandra/db/compaction/writers/DefaultCompactionWriter.java
+++ b/src/java/org/apache/cassandra/db/compaction/writers/DefaultCompactionWriter.java
@@ -42,13 +42,7 @@ public class DefaultCompactionWriter extends CompactionAwareWriter
this(cfs, directories, txn, nonExpiredSSTables, false, 0);
}
- @Deprecated
- public DefaultCompactionWriter(ColumnFamilyStore cfs, Directories directories, LifecycleTransaction txn, Set<SSTableReader> nonExpiredSSTables, boolean offline, boolean keepOriginals, int sstableLevel)
- {
- this(cfs, directories, txn, nonExpiredSSTables, keepOriginals, sstableLevel);
- }
-
- @SuppressWarnings("resource")
+ @SuppressWarnings({ "resource", "RedundantSuppression" })
public DefaultCompactionWriter(ColumnFamilyStore cfs, Directories directories, LifecycleTransaction txn, Set<SSTableReader> nonExpiredSSTables, boolean keepOriginals, int sstableLevel)
{
super(cfs, directories, txn, nonExpiredSSTables, keepOriginals);
diff --git a/src/java/org/apache/cassandra/db/compaction/writers/MajorLeveledCompactionWriter.java b/src/java/org/apache/cassandra/db/compaction/writers/MajorLeveledCompactionWriter.java
index a9f2627b2b..0c31cbf5cd 100644
--- a/src/java/org/apache/cassandra/db/compaction/writers/MajorLeveledCompactionWriter.java
+++ b/src/java/org/apache/cassandra/db/compaction/writers/MajorLeveledCompactionWriter.java
@@ -47,7 +47,7 @@ public class MajorLeveledCompactionWriter extends CompactionAwareWriter
this(cfs, directories, txn, nonExpiredSSTables, maxSSTableSize, false);
}
- @SuppressWarnings("resource")
+ @SuppressWarnings({ "resource", "RedundantSuppression" })
public MajorLeveledCompactionWriter(ColumnFamilyStore cfs,
Directories directories,
LifecycleTransaction txn,
diff --git a/src/java/org/apache/cassandra/index/Index.java b/src/java/org/apache/cassandra/index/Index.java
index bd7f3db9f7..de738019c2 100644
--- a/src/java/org/apache/cassandra/index/Index.java
+++ b/src/java/org/apache/cassandra/index/Index.java
@@ -21,10 +21,15 @@
package org.apache.cassandra.index;
import java.util.Collection;
+import java.util.Collections;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Callable;
-import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.db.ColumnFamilyStore;
@@ -35,8 +40,8 @@ import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.WriteContext;
-import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.memtable.Memtable;
import org.apache.cassandra.db.partitions.PartitionIterator;
@@ -46,12 +51,16 @@ import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.internal.CollatedViewIndexBuilder;
import org.apache.cassandra.index.transactions.IndexTransaction;
+import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.ReducingKeyIterator;
import org.apache.cassandra.io.sstable.SSTableFlushObserver;
+import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.IndexMetadata;
+import org.apache.cassandra.schema.TableMetadata;
+
/**
* Consisting of a top level Index interface and two sub-interfaces which handle read and write operations,
@@ -171,7 +180,7 @@ public interface Index
*/
interface IndexBuildingSupport
{
- SecondaryIndexBuilder getIndexBuildTask(ColumnFamilyStore cfs, Set<Index> indexes, Collection<SSTableReader> sstables);
+ SecondaryIndexBuilder getIndexBuildTask(ColumnFamilyStore cfs, Set<Index> indexes, Collection<SSTableReader> sstables, boolean isFullRebuild);
}
/**
@@ -180,8 +189,8 @@ public interface Index
*/
public static class CollatedViewIndexBuildingSupport implements IndexBuildingSupport
{
- @SuppressWarnings("resource")
- public SecondaryIndexBuilder getIndexBuildTask(ColumnFamilyStore cfs, Set<Index> indexes, Collection<SSTableReader> sstables)
+ @SuppressWarnings({"resource", "RedundantSuppression"})
+ public SecondaryIndexBuilder getIndexBuildTask(ColumnFamilyStore cfs, Set<Index> indexes, Collection<SSTableReader> sstables, boolean isFullRebuild)
{
return new CollatedViewIndexBuilder(cfs, indexes, new ReducingKeyIterator(sstables), sstables);
}
@@ -267,9 +276,10 @@ public interface Index
public void register(IndexRegistry registry);
/**
- * If the index implementation uses a local table to store its index data this method should return a
- * handle to it. If not, an empty Optional should be returned. Typically, this is useful for the built-in
- * Index implementations.
+ * If the index implementation uses a local table to store its index data, this method should return a
+ * handle to it. If not, an empty {@link Optional} should be returned. This exists to support legacy
+ * implementations, and should always be empty for indexes not belonging to a {@link SingletonIndexGroup}.
+ *
* @return an Optional referencing the Index's backing storage table if it has one, or Optional.empty() if not.
*/
public Optional<ColumnFamilyStore> getBackingTable();
@@ -343,11 +353,11 @@ public interface Index
* Get flush observer to observe partition/cell events generated by flushing SSTable (memtable flush or compaction).
*
* @param descriptor The descriptor of the sstable observer is requested for.
- * @param opType The type of the operation which requests observer e.g. memtable flush or compaction.
+ * @param tracker The {@link LifecycleNewTracker} associated with the SSTable being written
*
* @return SSTable flush observer.
*/
- default SSTableFlushObserver getFlushObserver(Descriptor descriptor, OperationType opType)
+ default SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker)
{
return null;
}
@@ -414,6 +424,17 @@ public interface Index
*/
public long getEstimatedResultRows();
+ /**
+ * Check if current index is queryable based on the index status.
+ *
+ * @param status current status of the index
+ * @return true if index should be queryable, false if index should be non-queryable
+ */
+ default boolean isQueryable(Status status)
+ {
+ return true;
+ }
+
/*
* Input validation
*/
@@ -425,10 +446,19 @@ public interface Index
* cell values for each row in the update may be checked by index
* implementations
* @param update PartitionUpdate containing the values to be validated by registered Index implementations
- * @throws InvalidRequestException
*/
public void validate(PartitionUpdate update) throws InvalidRequestException;
+ /**
+ * Returns the SSTable-attached {@link Component}s created by this index.
+ *
+ * @return the SSTable components created by this index
+ */
+ default Set<Component> getComponents()
+ {
+ return Collections.emptySet();
+ }
+
/*
* Update processing
*/
@@ -444,6 +474,8 @@ public interface Index
* @param ctx WriteContext spanning the update operation
* @param transactionType indicates what kind of update is being performed on the base data
* i.e. a write time insert/update/delete or the result of compaction
+ * @param memtable current memtable that the write goes into. It's to make sure memtable and index memtable
+ * are in sync.
* @return the newly created indexer or {@code null} if the index is not interested by the update
* (this could be because the index doesn't care about that particular partition, doesn't care about
* that type of transaction, ...).
@@ -452,7 +484,8 @@ public interface Index
RegularAndStaticColumns columns,
long nowInSec,
WriteContext ctx,
- IndexTransaction.Type transactionType);
+ IndexTransaction.Type transactionType,
+ Memtable memtable);
/**
* Listener for processing events emitted during a single partition update.
@@ -483,7 +516,6 @@ public interface Index
/**
* Notification of a top level partition delete.
- * @param deletionTime
*/
public void partitionDelete(DeletionTime deletionTime);
@@ -491,7 +523,6 @@ public interface Index
* Notification of a RangeTombstone.
* An update of a single partition may contain multiple RangeTombstones,
* and a notification will be passed for each of them.
- * @param tombstone
*/
public void rangeTombstone(RangeTombstone tombstone);
@@ -573,39 +604,6 @@ public interface Index
{
}
- /**
- * Tells whether this index supports replica fitering protection or not.
- *
- * Replica filtering protection might need to run the query row filter in the coordinator to detect stale results.
- * An index implementation will be compatible with this protection mechanism if it returns the same results for the
- * row filter as CQL will return with {@code ALLOW FILTERING} and without using the index. This means that index
- * implementations using custom query syntax or applying transformations to the indexed data won't support it.
- * See CASSANDRA-8272 for further details.
- *
- * @param rowFilter rowFilter of query to decide if it supports replica filtering protection or not
- * @return true if this index supports replica filtering protection, false otherwise
- */
- default boolean supportsReplicaFilteringProtection(RowFilter rowFilter)
- {
- return true;
- }
-
- /**
- * Return a function which performs post processing on the results of a partition range read command.
- * In future, this may be used as a generalized mechanism for transforming results on the coordinator prior
- * to returning them to the caller.
- *
- * This is used on the coordinator during execution of a range command to perform post
- * processing of merged results obtained from the necessary replicas. This is the only way in which results are
- * transformed in this way but this may change over time as usage is generalized.
- * See CASSANDRA-8717 for further discussion.
- *
- * The function takes a PartitionIterator of the results from the replicas which has already been collated
- * and reconciled, along with the command being executed. It returns another PartitionIterator containing the results
- * of the transformation (which may be the same as the input if the transformation is a no-op).
- */
- public BiFunction<PartitionIterator, ReadCommand, PartitionIterator> postProcessorFor(ReadCommand command);
-
/**
* Factory method for query time search helper.
*
@@ -621,10 +619,298 @@ public interface Index
*/
public interface Searcher
{
+ /**
+ * Returns the {@link ReadCommand} for which this searcher has been created.
+ *
+ * @return the base read command
+ */
+ ReadCommand command();
+
/**
* @param executionController the collection of OpOrder.Groups which the ReadCommand is being performed under.
* @return partitions from the base table matching the criteria of the search.
*/
public UnfilteredPartitionIterator search(ReadExecutionController executionController);
+
+ /**
+ * Replica filtering protection may fetch data that doesn't match query conditions.
+ *
+ * On coordinator, we need to filter the replicas' responses again.
+ *
+ * This will not be called if {@link QueryPlan#supportsReplicaFilteringProtection(RowFilter)} returns false.
+ *
+ * @return filtered response that satisfied query conditions
+ */
+ default PartitionIterator filterReplicaFilteringProtection(PartitionIterator fullResponse)
+ {
+ return command().rowFilter().filter(fullResponse, command().metadata(), command().nowInSec());
+ }
+ }
+
+ /**
+ * Class providing grouped operations for indexes that communicate with each other.
+ *
+ * Index implementations should provide a {@code Group} implementation calling to
+ * {@link SecondaryIndexManager#registerIndex(Index, Object, Supplier)} during index registering
+ * at {@link #register(IndexRegistry)} method.
+ */
+ interface Group
+ {
+ /**
+ * Returns the indexes that are members of this group.
+ *
+ * @return the indexes that are members of this group
+ */
+ Set<Index> getIndexes();
+
+ /**
+ * Adds the specified {@link Index} as a member of this group.
+ *
+ * @param index the index to be added
+ */
+ void addIndex(Index index);
+
+ /**
+ * Removes the specified {@link Index} from the members of this group.
+ *
+ * @param index the index to be removed
+ */
+ void removeIndex(Index index);
+
+ /**
+ * Returns if this group contains the specified {@link Index}.
+ *
+ * @param index the index to be removed
+ * @return {@code true} if this group contains {@code index}, {@code false} otherwise
+ */
+ boolean containsIndex(Index index);
+
+ /**
+ * Creates an new {@code Indexer} object for updates to a given partition.
+ *
+ * @param indexSelector a predicate selecting the targeted members
+ * @param key key of the partition being modified
+ * @param columns the regular and static columns the created indexer will have to deal with.
+ * This can be empty as an update might only contain partition, range and row deletions, but
+ * the indexer is guaranteed to not get any cells for a column that is not part of {@code columns}.
+ * @param nowInSec current time of the update operation
+ * @param ctx WriteContext spanning the update operation
+ * @param transactionType indicates what kind of update is being performed on the base data
+ * i.e. a write time insert/update/delete or the result of compaction
+ * @param memtable the {@link Memtable} to which the updates are being applied or {@code null}
+ * if the source of the updates is an existing {@link SSTable}
+ *
+ * @return the newly created indexer or {@code null} if the index is not interested by the update
+ * (this could be because the index doesn't care about that particular partition, doesn't care about
+ * that type of transaction, ...).
+ */
+ Indexer indexerFor(Predicate<Index> indexSelector,
+ DecoratedKey key,
+ RegularAndStaticColumns columns,
+ long nowInSec,
+ WriteContext ctx,
+ IndexTransaction.Type transactionType,
+ Memtable memtable);
+
+ /**
+ * Returns a new {@link QueryPlan} for the specified {@link RowFilter}, or {@code null} if none of the indexes in
+ * this group supports the expression in the row filter.
+ *
+ * @param rowFilter a row filter
+ * @return a new query plan for the specified {@link RowFilter} if it's supported, {@code null} otherwise
+ */
+ @Nullable
+ QueryPlan queryPlanFor(RowFilter rowFilter);
+
+ /**
+ * Get flush observer to observe partition/cell events generated by flushing SSTable (memtable flush or compaction).
+ *
+ * @param descriptor The descriptor of the sstable observer is requested for.
+ * @param tracker The {@link LifecycleNewTracker} associated with the SSTable being written
+ * @param tableMetadata The immutable metadata of the table at the moment the SSTable is flushed
+ *
+ * @return SSTable flush observer.
+ */
+ SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker, TableMetadata tableMetadata);
+
+ /**
+ * @param type index transaction type
+ * @return true if index will be able to handle given index transaction type
+ */
+ default boolean handles(IndexTransaction.Type type)
+ {
+ return true;
+ }
+
+ /**
+ * Called when the table associated with this group has been invalidated. Implementations
+ * should dispose of any resources tied to the lifecycle of the {@link Group}.
+ */
+ default void invalidate() { }
+
+ /**
+ * Returns the SSTable-attached {@link Component}s created by this index group.
+ *
+ * @return the SSTable components created by this group
+ */
+ Set<Component> getComponents();
+ }
+
+ /**
+ * Specifies a set of compatible indexes to be used with a query according to its {@link RowFilter}, ignoring data
+ * ranges, limits, etc. All the indexes in it should belong to the same {@link Group}.
+ * <p>
+ * It's created by {@link Group#queryPlanFor} from the {@link RowFilter} that is common to all the subcommands of a
+ * user query's {@link ReadCommand}, so it can be reused by those subcommands along the cluster nodes. The
+ * {@link #searcherFor(ReadCommand)} method provides the {@link Searcher} object to read the index for each
+ * particular (sub)command.
+ */
+ interface QueryPlan extends Comparable<QueryPlan>
+ {
+ /**
+ * Returns the indexes selected by this query plan, all of them belonging to the same {@link Group}.
+ *
+ * It should never be empty.
+ *
+ * @return the indexes selected by this query plan, which is never empty
+ */
+ Set<Index> getIndexes();
+
+ /**
+ * Returns the first index in this plan.
+ *
+ * @return the first index
+ */
+ @Nonnull
+ default Index getFirst()
+ {
+ return getIndexes().iterator().next();
+ }
+
+ /**
+ * Return an estimate of the number of results this plan is expected to return for any given {@link ReadCommand}
+ * that it can be used to answer. Used by {@link SecondaryIndexManager#getBestIndexQueryPlanFor(RowFilter)}
+ * to determine the {@link Group} with the most selective plan for a given {@link RowFilter}.
+ * Additionally, this is also used by StorageProxy.estimateResultsPerRange to calculate the initial concurrency
+ * factor for range requests
+ *
+ * @return the estimated average number of results a Searcher may return for any given command
+ */
+ default long getEstimatedResultRows()
+ {
+ // CQL only supports AND expressions, so the estimated number of results for multiple indexes will be the
+ // the lowest of the estimates for each index
+ return getIndexes().stream()
+ .mapToLong(Index::getEstimatedResultRows)
+ .min()
+ .orElseThrow(AssertionError::new); // registered groups are never empty
+ }
+
+ /**
+ * Used to determine whether to estimate initial concurrency during remote range reads. Default is true, each
+ * implementation must override this method if they choose a different strategy.
+ *
+ * @return true if the {@link QueryPlan} should estimate initial concurrency, false otherwise
+ */
+ default boolean shouldEstimateInitialConcurrency()
+ {
+ return true;
+ }
+
+ @Override
+ default int compareTo(QueryPlan other)
+ {
+ // initially, we prefer the plan with less estimated results
+ int results = Long.compare(getEstimatedResultRows(), other.getEstimatedResultRows());
+ if (results != 0)
+ return results;
+
+ // In case of having the same number of estimated results, we favour the plan that involves more indexes.
+ // This way, we honour the possible absence of ALLOW FILTERING in the CQL query. Also, this criteria should
+ // not break the transitivity of this method because the estimated number of results for a plan is the
+ // minimum of the estimates of its members.
+ return Integer.compare(getIndexes().size(), other.getIndexes().size());
+ }
+
+ /**
+ * Used to validate the various parameters of a supplied {@link ReadCommand} against the indexes in this plan.
+ *
+ * @param command a ReadCommand whose parameters are to be verified
+ * @throws InvalidRequestException if the details of the command fail to meet the validation rules of the
+ * indexes in the query plan
+ */
+ default void validate(ReadCommand command) throws InvalidRequestException
+ {
+ getIndexes().forEach(i -> i.validate(command));
+ }
+
+ /**
+ * Factory method for query time search helper.
+ *
+ * @param command the read command being executed
+ * @return an Searcher with which to perform the supplied command
+ */
+ Searcher searcherFor(ReadCommand command);
+
+ /**
+ * Return a function which performs post processing on the results of a partition range read command.
+ * In future, this may be used as a generalized mechanism for transforming results on the coordinator prior
+ * to returning them to the caller.
+ *
+ * This is used on the coordinator during execution of a range command to perform post
+ * processing of merged results obtained from the necessary replicas. This is the only way in which results are
+ * transformed in this way but this may change over time as usage is generalized.
+ * See CASSANDRA-8717 for further discussion.
+ *
+ * The function takes a PartitionIterator of the results from the replicas which has already been collated
+ * and reconciled, along with the command being executed. It returns another PartitionIterator containing the results
+ * of the transformation (which may be the same as the input if the transformation is a no-op).
+ */
+ default Function<PartitionIterator, PartitionIterator> postProcessor()
+ {
+ return partitions -> partitions;
+ }
+
+ /**
+ * Transform an initial {@link RowFilter} into the filter that will still need to applied to a set of Rows after
+ * the index has performed it's initial scan.
+ *
+ * Used in {@link ReadCommand#executeLocally(ReadExecutionController)} to reduce the amount of filtering performed on the
+ * results of the index query.
+ *
+ * @return the (hopefully) reduced filter that would still need to be applied after
+ * the index was used to narrow the initial result set
+ */
+ RowFilter postIndexQueryFilter();
+
+
+ /**
+ * Tells whether this index supports replica fitering protection or not.
+ *
+ * Replica filtering protection might need to run the query row filter in the coordinator to detect stale results.
+ * An index implementation will be compatible with this protection mechanism if it returns the same results for the
+ * row filter as CQL will return with {@code ALLOW FILTERING} and without using the index. This means that index
+ * implementations using custom query syntax or applying transformations to the indexed data won't support it.
+ * See CASSANDRA-8272 for further details.
+ *
+ * @param rowFilter rowFilter of query to decide if it supports replica filtering protection or not
+ * @return true if this index supports replica filtering protection, false otherwise
+ */
+ default boolean supportsReplicaFilteringProtection(RowFilter rowFilter)
+ {
+ return true;
+ }
+ }
+
+ /*
+ * Status of index used to determine queryability
+ */
+ enum Status
+ {
+ UNKNOWN,
+ FULL_REBUILD_STARTED,
+ BUILD_FAILED,
+ BUILD_SUCCEEDED,
+ DROPPED
}
}
diff --git a/src/java/org/apache/cassandra/index/IndexRegistry.java b/src/java/org/apache/cassandra/index/IndexRegistry.java
index 32acfb8fb1..4d49c693d6 100644
--- a/src/java/org/apache/cassandra/index/IndexRegistry.java
+++ b/src/java/org/apache/cassandra/index/IndexRegistry.java
@@ -23,19 +23,27 @@ package org.apache.cassandra.index;
import java.util.Collection;
import java.util.Collections;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.Callable;
-import java.util.function.BiFunction;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+
+import javax.annotation.Nullable;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
-import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.memtable.Memtable;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.transactions.IndexTransaction;
+import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.SSTableFlushObserver;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.TableMetadata;
@@ -56,19 +64,20 @@ public interface IndexRegistry
public static final IndexRegistry EMPTY = new IndexRegistry()
{
@Override
- public void unregisterIndex(Index index)
+ public void registerIndex(Index index, Object groupKey, Supplier<Index.Group> groupSupplier)
{
}
@Override
- public void registerIndex(Index index)
+ public Collection<Index> listIndexes()
{
+ return Collections.emptyList();
}
@Override
- public Collection<Index> listIndexes()
+ public Collection<Index.Group> listIndexGroups()
{
- return Collections.emptyList();
+ return Collections.emptySet();
}
@Override
@@ -97,7 +106,7 @@ public interface IndexRegistry
*/
public static final IndexRegistry NON_DAEMON = new IndexRegistry()
{
- Index index = new Index()
+ final Index index = new Index()
{
public Callable<?> getInitializationTask()
{
@@ -173,27 +182,70 @@ public interface IndexRegistry
{
}
- public Indexer indexerFor(DecoratedKey key, RegularAndStaticColumns columns, long nowInSec, WriteContext ctx, IndexTransaction.Type transactionType)
+ public Indexer indexerFor(DecoratedKey key, RegularAndStaticColumns columns, long nowInSec, WriteContext ctx, IndexTransaction.Type transactionType, Memtable memtable)
{
return null;
}
- public BiFunction<PartitionIterator, ReadCommand, PartitionIterator> postProcessorFor(ReadCommand command)
+ public Searcher searcherFor(ReadCommand command)
{
return null;
}
+ };
- public Searcher searcherFor(ReadCommand command)
+ final Index.Group group = new Index.Group()
+ {
+ @Override
+ public Set<Index> getIndexes()
+ {
+ return Collections.singleton(index);
+ }
+
+ @Override
+ public void addIndex(Index index)
+ {
+ }
+
+ @Override
+ public void removeIndex(Index index)
+ {
+ }
+
+ @Override
+ public boolean containsIndex(Index i)
+ {
+ return index == i;
+ }
+
+ @Nullable
+ @Override
+ public Index.Indexer indexerFor(Predicate<Index> indexSelector, DecoratedKey key, RegularAndStaticColumns columns, long nowInSec, WriteContext ctx, IndexTransaction.Type transactionType, Memtable memtable)
{
return null;
}
- };
- public void registerIndex(Index index)
- {
- }
+ @Nullable
+ @Override
+ public Index.QueryPlan queryPlanFor(RowFilter rowFilter)
+ {
+ return null;
+ }
- public void unregisterIndex(Index index)
+ @Nullable
+ @Override
+ public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker, TableMetadata tableMetadata)
+ {
+ return null;
+ }
+
+ @Override
+ public Set<Component> getComponents()
+ {
+ return null;
+ }
+ };
+
+ public void registerIndex(Index index, Object groupKey, Supplier<Index.Group> groupSupplier)
{
}
@@ -207,6 +259,12 @@ public interface IndexRegistry
return Collections.singletonList(index);
}
+ @Override
+ public Collection<Index.Group> listIndexGroups()
+ {
+ return Collections.singletonList(group);
+ }
+
public Optional<Index> getBestIndexFor(RowFilter.Expression expression)
{
return Optional.empty();
@@ -214,12 +272,15 @@ public interface IndexRegistry
public void validate(PartitionUpdate update)
{
-
}
};
- void registerIndex(Index index);
- void unregisterIndex(Index index);
+ default void registerIndex(Index index)
+ {
+ registerIndex(index, index, () -> new SingletonIndexGroup(index));
+ }
+ public void registerIndex(Index index, Object groupKey, Supplier<Index.Group> groupSupplier);
+ Collection<Index.Group> listIndexGroups();
Index getIndex(IndexMetadata indexMetadata);
Collection<Index> listIndexes();
diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
index 31d3e84561..826e8666a9 100644
--- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
+++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
@@ -24,14 +24,21 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import javax.annotation.Nullable;
+
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.base.Strings;
-import com.google.common.collect.*;
-import com.google.common.primitives.Longs;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
import com.google.common.util.concurrent.FutureCallback;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
@@ -149,6 +156,11 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
*/
private final Map<String, Index> writableIndexes = Maps.newConcurrentMap();
+ /**
+ * The groups of all the registered indexes
+ */
+ private final Map<Object, Index.Group> indexGroups = Maps.newConcurrentMap();
+
/**
* The count of pending index builds for each index.
*/
@@ -212,9 +224,15 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
markIndexesBuilding(ImmutableSet.of(index), true, isNewCF);
+ return buildIndex(index);
+ }
+
+ @VisibleForTesting
+ public Future<Void> buildIndex(final Index index)
+ {
FutureTask<?> initialBuildTask = null;
// if the index didn't register itself, we can probably assume that no initialization needs to happen
- if (indexes.containsKey(indexDef.name))
+ if (indexes.containsKey(index.getIndexMetadata().name))
{
try
{
@@ -281,7 +299,23 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
{
return queryableIndexes.contains(index.getIndexMetadata().name);
}
-
+
+ /**
+ * Throws an {@link IndexNotAvailableException} if any of the indexes in the specified {@link Index.QueryPlan} is
+ * not queryable, as it's defined by {@link #isIndexQueryable(Index)}.
+ *
+ * @param queryPlan a query plan
+ * @throws IndexNotAvailableException if the query plan has any index that is not queryable
+ */
+ public void checkQueryability(Index.QueryPlan queryPlan)
+ {
+ for (Index index : queryPlan.getIndexes())
+ {
+ if (!isIndexQueryable(index))
+ throw new IndexNotAvailableException(index);
+ }
+ }
+
/**
* Checks if the specified index is writable.
*
@@ -490,7 +524,7 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
{
logger.info("Submitting index {} of {} for data in {}",
isFullRebuild ? "recovery" : "build",
- indexes.stream().map(i -> i.getIndexMetadata().name).collect(Collectors.joining(",")),
+ commaSeparated(indexes),
sstables.stream().map(SSTableReader::toString).collect(Collectors.joining(",")));
// Group all building tasks
@@ -508,7 +542,7 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
List<Future<?>> futures = new ArrayList<>(byType.size());
byType.forEach((buildingSupport, groupedIndexes) ->
{
- SecondaryIndexBuilder builder = buildingSupport.getIndexBuildTask(baseCfs, groupedIndexes, sstables);
+ SecondaryIndexBuilder builder = buildingSupport.getIndexBuildTask(baseCfs, groupedIndexes, sstables, isFullRebuild);
final AsyncPromise<Object> build = new AsyncPromise<>();
CompactionManager.instance.submitIndexBuild(builder).addCallback(new FutureCallback()
{
@@ -643,7 +677,10 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
AtomicInteger counter = inProgressBuilds.computeIfAbsent(indexName, ignored -> new AtomicInteger(0));
if (isFullRebuild)
+ {
needsFullRebuild.remove(indexName);
+ makeIndexNonQueryable(index, Index.Status.FULL_REBUILD_STARTED);
+ }
if (counter.getAndIncrement() == 0 && DatabaseDescriptor.isDaemonInitialized() && !isNewCF)
SystemKeyspace.setIndexRemoved(keyspaceName, indexName);
@@ -661,14 +698,8 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
{
String indexName = index.getIndexMetadata().name;
if (isFullRebuild)
- {
- if (queryableIndexes.add(indexName))
- logger.info("Index [{}] became queryable after successful build.", indexName);
-
- if (writableIndexes.put(indexName, index) == null)
- logger.info("Index [{}] became writable after successful build.", indexName);
- }
-
+ makeIndexQueryable(index, Index.Status.BUILD_SUCCEEDED);
+
AtomicInteger counter = inProgressBuilds.get(indexName);
if (counter != null)
{
@@ -748,8 +779,10 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
if (indexDef.isCustom())
{
assert indexDef.options != null;
- String className = indexDef.options.get(IndexTarget.CUSTOM_INDEX_OPTION_NAME);
+ // Find any aliases to the fully qualified index class name:
+ String className = IndexMetadata.expandAliases(indexDef.options.get(IndexTarget.CUSTOM_INDEX_OPTION_NAME));
assert !Strings.isNullOrEmpty(className);
+
try
{
Class<? extends Index> indexClass = FBUtilities.classForName(className, "Index");
@@ -784,6 +817,9 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
markAllIndexesRemoved();
if (dropData)
invalidateAllIndexesBlocking();
+
+ // TODO: Determine whether "dropData" should guard this or be passed to Group#invalidate()
+ indexGroups.forEach((key, group) -> group.invalidate());
}
@VisibleForTesting
@@ -927,12 +963,14 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
try (UnfilteredRowIterator partition = page.next())
{
- Set<Index.Indexer> indexers = indexes.stream()
- .map(index -> index.indexerFor(key,
- partition.columns(),
- nowInSec,
- ctx,
- IndexTransaction.Type.UPDATE))
+ Set<Index.Indexer> indexers = indexGroups.values().stream()
+ .map(g -> g.indexerFor(indexes::contains,
+ key,
+ partition.columns(),
+ nowInSec,
+ ctx,
+ IndexTransaction.Type.UPDATE,
+ null))
.filter(Objects::nonNull)
.collect(Collectors.toSet());
@@ -1033,6 +1071,9 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
*/
public void deletePartition(UnfilteredRowIterator partition, long nowInSec)
{
+ if (!handles(IndexTransaction.Type.CLEANUP))
+ return;
+
// we need to acquire memtable lock because secondary index deletion may
// cause a race (see CASSANDRA-3712). This is done internally by the
// index transaction when it commits
@@ -1080,15 +1121,14 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
* ReadOrderGroup, or an estimate of the result size from an average index query.
*
* @param rowFilter RowFilter of the command to be executed
- * @return an Index instance, ready to use during execution of the command, or null if none
- * of the registered indexes can support the command.
+ * @return the best available index query plan for the row filter, or {@code null} if none of the registered indexes
+ * can support the command.
*/
- public Index getBestIndexFor(RowFilter rowFilter)
+ public Index.QueryPlan getBestIndexQueryPlanFor(RowFilter rowFilter)
{
if (indexes.isEmpty() || rowFilter.isEmpty())
return null;
- Set<Index> searchableIndexes = new HashSet<>();
for (RowFilter.Expression expression : rowFilter)
{
if (expression.isCustom())
@@ -1098,41 +1138,49 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
RowFilter.CustomExpression customExpression = (RowFilter.CustomExpression) expression;
logger.trace("Command contains a custom index expression, using target index {}", customExpression.getTargetIndex().name);
Tracing.trace("Command contains a custom index expression, using target index {}", customExpression.getTargetIndex().name);
- return indexes.get(customExpression.getTargetIndex().name);
- }
- else if (!expression.isUserDefined())
- {
- indexes.values().stream()
- .filter(index -> index.supportsExpression(expression.column(), expression.operator()))
- .forEach(searchableIndexes::add);
+ Index.Group group = getIndexGroup(customExpression.getTargetIndex());
+ return group == null ? null : group.queryPlanFor(rowFilter);
}
}
- if (searchableIndexes.isEmpty())
+ Set<Index.QueryPlan> queryPlans = indexGroups.values()
+ .stream()
+ .map(g -> g.queryPlanFor(rowFilter))
+ .filter(Objects::nonNull)
+ .collect(Collectors.toSet());
+
+ if (queryPlans.isEmpty())
{
logger.trace("No applicable indexes found");
Tracing.trace("No applicable indexes found");
return null;
}
- Index selected = searchableIndexes.size() == 1
- ? Iterables.getOnlyElement(searchableIndexes)
- : searchableIndexes.stream()
- .min((a, b) -> Longs.compare(a.getEstimatedResultRows(),
- b.getEstimatedResultRows()))
- .orElseThrow(() -> new AssertionError("Could not select most selective index"));
+ // find the best plan
+ Index.QueryPlan selected = queryPlans.size() == 1
+ ? Iterables.getOnlyElement(queryPlans)
+ : queryPlans.stream()
+ .min(Comparator.naturalOrder())
+ .orElseThrow(() -> new AssertionError("Could not select most selective index"));
// pay for an additional threadlocal get() rather than build the strings unnecessarily
if (Tracing.isTracing())
{
Tracing.trace("Index mean cardinalities are {}. Scanning with {}.",
- searchableIndexes.stream().map(i -> i.getIndexMetadata().name + ':' + i.getEstimatedResultRows())
- .collect(Collectors.joining(",")),
- selected.getIndexMetadata().name);
+ queryPlans.stream()
+ .map(p -> commaSeparated(p.getIndexes()) + ':' + p.getEstimatedResultRows())
+ .collect(Collectors.joining(",")),
+ commaSeparated(selected.getIndexes()));
}
return selected;
}
+ private static String commaSeparated(Collection<Index> indexes)
+ {
+ return indexes.stream().map(i -> i.getIndexMetadata().name).collect(Collectors.joining(","));
+ }
+
+
public Optional<Index> getBestIndexFor(RowFilter.Expression expression)
{
return indexes.values().stream().filter((i) -> i.supportsExpression(expression.column(), expression.operator())).findFirst();
@@ -1157,23 +1205,50 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
/*
* IndexRegistry methods
*/
-
- public void registerIndex(Index index)
+ public void registerIndex(Index index, Object groupKey, Supplier<Index.Group> groupSupplier)
{
String name = index.getIndexMetadata().name;
indexes.put(name, index);
logger.trace("Registered index {}", name);
- }
- public void unregisterIndex(Index index)
- {
- unregisterIndex(index.getIndexMetadata().name);
+ // instantiate and add the index group if it hasn't been already added
+ Index.Group group = indexGroups.computeIfAbsent(groupKey, k -> groupSupplier.get());
+
+ // add the created index to its group if it is not a singleton group
+ if (!(group instanceof SingletonIndexGroup))
+ {
+ if (index.getBackingTable().isPresent())
+ throw new InvalidRequestException("Indexes belonging to a group of indexes shouldn't have a backing table");
+
+ group.addIndex(index);
+ }
}
private Index unregisterIndex(String name)
{
Index removed = indexes.remove(name);
logger.trace(removed == null ? "Index {} was not registered" : "Removed index {} from registry", name);
+
+ if (removed != null)
+ {
+ // Remove the index from any non-singleton groups...
+ for (Index.Group group : listIndexGroups())
+ {
+ if (!(group instanceof SingletonIndexGroup) && group.containsIndex(removed))
+ {
+ group.removeIndex(removed);
+
+ if (group.getIndexes().isEmpty())
+ {
+ indexGroups.remove(group);
+ }
+ }
+ }
+
+ // ...and remove singleton groups entirely.
+ indexGroups.remove(removed);
+ }
+
return removed;
}
@@ -1187,6 +1262,42 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
return ImmutableSet.copyOf(indexes.values());
}
+ public Set<Index.Group> listIndexGroups()
+ {
+ return ImmutableSet.copyOf(indexGroups.values());
+ }
+
+ public Index.Group getIndexGroup(Object key)
+ {
+ return indexGroups.get(key);
+ }
+
+ /**
+ * Returns the {@link Index.Group} the specified index belongs to, as specified during registering with
+ * {@link #registerIndex(Index, Object, Supplier)}.
+ *
+ * @param metadata the index metadata
+ * @return the group the index belongs to, or {@code null} if the index is not registered or if it hasn't been
+ * associated to any group
+ */
+ @Nullable
+ public Index.Group getIndexGroup(IndexMetadata metadata)
+ {
+ Index index = getIndex(metadata);
+ return index == null ? null : getIndexGroup(index);
+ }
+
+ @VisibleForTesting
+ public boolean needsFullRebuild(String index)
+ {
+ return needsFullRebuild.contains(index);
+ }
+
+ public Index.Group getIndexGroup(Index index)
+ {
+ return indexGroups.values().stream().filter(g -> g.containsIndex(index)).findAny().orElse(null);
+ }
+
/*
* Handling of index updates.
* Implementations of the various IndexTransaction interfaces, for keeping indexes in sync with base data
@@ -1196,23 +1307,28 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
/**
* Transaction for updates on the write path.
*/
- public UpdateTransaction newUpdateTransaction(PartitionUpdate update, WriteContext ctx, long nowInSec)
+ public UpdateTransaction newUpdateTransaction(PartitionUpdate update, WriteContext ctx, long nowInSec, Memtable memtable)
{
if (!hasIndexes())
return UpdateTransaction.NO_OP;
- ArrayList<Index.Indexer> idxrs = new ArrayList<>();
- for (Index i : writableIndexes.values())
- {
- Index.Indexer idxr = i.indexerFor(update.partitionKey(), update.columns(), nowInSec, ctx, IndexTransaction.Type.UPDATE);
- if (idxr != null)
- idxrs.add(idxr);
- }
+ Index.Indexer[] indexers = listIndexGroups().stream()
+ .map(g -> g.indexerFor(writableIndexSelector(),
+ update.partitionKey(),
+ update.columns(),
+ nowInSec,
+ ctx,
+ IndexTransaction.Type.UPDATE,
+ memtable))
+ .filter(Objects::nonNull)
+ .toArray(Index.Indexer[]::new);
+
+ return indexers.length == 0 ? UpdateTransaction.NO_OP : new WriteTimeTransaction(indexers);
+ }
- if (idxrs.size() == 0)
- return UpdateTransaction.NO_OP;
- else
- return new WriteTimeTransaction(idxrs.toArray(new Index.Indexer[idxrs.size()]));
+ private Predicate<Index> writableIndexSelector()
+ {
+ return index -> writableIndexes.containsKey(index.getIndexMetadata().name);
}
/**
@@ -1224,7 +1340,7 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
long nowInSec)
{
// the check for whether there are any registered indexes is already done in CompactionIterator
- return new IndexGCTransaction(key, regularAndStaticColumns, keyspace, versions, nowInSec, writableIndexes.values());
+ return new IndexGCTransaction(key, regularAndStaticColumns, keyspace, versions, nowInSec, listIndexGroups(), writableIndexSelector());
}
/**
@@ -1237,7 +1353,21 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
if (!hasIndexes())
return CleanupTransaction.NO_OP;
- return new CleanupGCTransaction(key, regularAndStaticColumns, keyspace, nowInSec, writableIndexes.values());
+ return new CleanupGCTransaction(key, regularAndStaticColumns, keyspace, nowInSec, listIndexGroups(), writableIndexSelector());
+ }
+
+ /**
+ * @param type index transaction type
+ * @return true if at least one of the indexes will be able to handle given index transaction type
+ */
+ public boolean handles(IndexTransaction.Type type)
+ {
+ for (Index.Group group : indexGroups.values())
+ {
+ if (group.handles(type))
+ return true;
+ }
+ return false;
}
/**
@@ -1353,7 +1483,8 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
private final Keyspace keyspace;
private final int versions;
private final long nowInSec;
- private final Collection<Index> indexes;
+ private final Collection<Index.Group> indexGroups;
+ private final Predicate<Index> writableIndexSelector;
private Row[] rows;
@@ -1362,14 +1493,16 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
Keyspace keyspace,
int versions,
long nowInSec,
- Collection<Index> indexes)
+ Collection<Index.Group> indexGroups,
+ Predicate<Index> writableIndexSelector)
{
this.key = key;
this.columns = columns;
this.keyspace = keyspace;
this.versions = versions;
- this.indexes = indexes;
+ this.indexGroups = indexGroups;
this.nowInSec = nowInSec;
+ this.writableIndexSelector = writableIndexSelector;
}
public void start()
@@ -1430,9 +1563,9 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
try (WriteContext ctx = keyspace.getWriteHandler().createContextForIndexing())
{
- for (Index index : indexes)
+ for (Index.Group group : indexGroups)
{
- Index.Indexer indexer = index.indexerFor(key, columns, nowInSec, ctx, Type.COMPACTION);
+ Index.Indexer indexer = group.indexerFor(writableIndexSelector, key, columns, nowInSec, ctx, Type.COMPACTION, null);
if (indexer == null)
continue;
@@ -1458,7 +1591,8 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
private final RegularAndStaticColumns columns;
private final Keyspace keyspace;
private final long nowInSec;
- private final Collection<Index> indexes;
+ private final Collection<Index.Group> indexGroups;
+ private final Predicate<Index> writableIndexSelector;
private Row row;
private DeletionTime partitionDelete;
@@ -1467,13 +1601,15 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
RegularAndStaticColumns columns,
Keyspace keyspace,
long nowInSec,
- Collection<Index> indexes)
+ Collection<Index.Group> indexGroups,
+ Predicate<Index> writableIndexSelector)
{
this.key = key;
this.columns = columns;
this.keyspace = keyspace;
- this.indexes = indexes;
+ this.indexGroups = indexGroups;
this.nowInSec = nowInSec;
+ this.writableIndexSelector = writableIndexSelector;
}
public void start()
@@ -1497,9 +1633,9 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
try (WriteContext ctx = keyspace.getWriteHandler().createContextForIndexing())
{
- for (Index index : indexes)
+ for (Index.Group group : indexGroups)
{
- Index.Indexer indexer = index.indexerFor(key, columns, nowInSec, ctx, Type.CLEANUP);
+ Index.Indexer indexer = group.indexerFor(writableIndexSelector, key, columns, nowInSec, ctx, Type.CLEANUP, null);
if (indexer == null)
continue;
@@ -1572,4 +1708,30 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
shutdown(asyncExecutor, blockingExecutor);
awaitTermination(timeout, units, asyncExecutor, blockingExecutor);
}
+
+ public void makeIndexNonQueryable(Index index, Index.Status status)
+ {
+ String name = index.getIndexMetadata().name;
+ if (indexes.get(name) == index)
+ {
+ if (!index.isQueryable(status))
+ queryableIndexes.remove(name);
+ }
+ }
+
+ public void makeIndexQueryable(Index index, Index.Status status)
+ {
+ String name = index.getIndexMetadata().name;
+ if (indexes.get(name) == index)
+ {
+ if (index.isQueryable(status))
+ {
+ if (queryableIndexes.add(name))
+ logger.info("Index [{}] became queryable after successful build.", name);
+ }
+
+ if (writableIndexes.put(name, index) == null)
+ logger.info("Index [{}] became writable after successful build.", name);
+ }
+ }
}
diff --git a/src/java/org/apache/cassandra/index/SingletonIndexGroup.java b/src/java/org/apache/cassandra/index/SingletonIndexGroup.java
new file mode 100644
index 0000000000..304d35d6d4
--- /dev/null
+++ b/src/java/org/apache/cassandra/index/SingletonIndexGroup.java
@@ -0,0 +1,114 @@
+/*
+ *
+ * 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.index;
+
+import java.util.Collections;
+import java.util.Set;
+import java.util.function.Predicate;
+
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.RegularAndStaticColumns;
+import org.apache.cassandra.db.WriteContext;
+import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
+import org.apache.cassandra.db.memtable.Memtable;
+import org.apache.cassandra.index.transactions.IndexTransaction;
+import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.SSTableFlushObserver;
+import org.apache.cassandra.schema.TableMetadata;
+
+/**
+ * An {@link Index.Group} containing a single {@link Index}, to which it just delegates the calls.
+ */
+public class SingletonIndexGroup implements Index.Group
+{
+ private final Index delegate;
+ private final Set<Index> indexes;
+
+ protected SingletonIndexGroup(Index delegate)
+ {
+ this.delegate = delegate;
+ this.indexes = Collections.singleton(delegate);
+ }
+
+ @Override
+ public Set<Index> getIndexes()
+ {
+ return indexes;
+ }
+
+ public Index getIndex()
+ {
+ return delegate;
+ }
+
+ @Override
+ public void addIndex(Index index)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void removeIndex(Index index)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean containsIndex(Index index)
+ {
+ return index.equals(delegate);
+ }
+
+ @Override
+ public Index.Indexer indexerFor(Predicate<Index> indexSelector,
+ DecoratedKey key,
+ RegularAndStaticColumns columns,
+ long nowInSec,
+ WriteContext ctx,
+ IndexTransaction.Type transactionType,
+ Memtable memtable)
+ {
+ return indexSelector.test(delegate)
+ ? delegate.indexerFor(key, columns, nowInSec, ctx, transactionType, memtable)
+ : null;
+ }
+
+ @Override
+ public Index.QueryPlan queryPlanFor(RowFilter rowFilter)
+ {
+ return SingletonIndexQueryPlan.create(delegate, rowFilter);
+ }
+
+ @Override
+ public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker, TableMetadata tableMetadata)
+ {
+ return delegate.getFlushObserver(descriptor, tracker);
+ }
+
+ @Override
+ public Set<Component> getComponents()
+ {
+ return delegate.getComponents();
+ }
+}
diff --git a/src/java/org/apache/cassandra/index/SingletonIndexQueryPlan.java b/src/java/org/apache/cassandra/index/SingletonIndexQueryPlan.java
new file mode 100644
index 0000000000..b475cee145
--- /dev/null
+++ b/src/java/org/apache/cassandra/index/SingletonIndexQueryPlan.java
@@ -0,0 +1,87 @@
+/*
+ *
+ * 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.index;
+
+import java.util.Collections;
+import java.util.Set;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.filter.RowFilter;
+
+public class SingletonIndexQueryPlan implements Index.QueryPlan
+{
+ private final Index index;
+ private final Set<Index> indexes;
+ private final RowFilter postIndexFilter;
+
+ protected SingletonIndexQueryPlan(Index index, RowFilter postIndexFilter)
+ {
+ this.index = index;
+ this.indexes = Collections.singleton(index);
+ this.postIndexFilter = postIndexFilter;
+ }
+
+ @Nullable
+ protected static SingletonIndexQueryPlan create(Index index, RowFilter rowFilter)
+ {
+ for (RowFilter.Expression e : rowFilter.getExpressions())
+ {
+ if (index.supportsExpression(e.column(), e.operator()))
+ return new SingletonIndexQueryPlan(index, index.getPostIndexQueryFilter(rowFilter));
+ }
+
+ return null;
+ }
+
+ @Override
+ public Set<Index> getIndexes()
+ {
+ return indexes;
+ }
+
+ @Override
+ @Nonnull
+ public Index getFirst()
+ {
+ return index;
+ }
+
+ @Override
+ public long getEstimatedResultRows()
+ {
+ return index.getEstimatedResultRows();
+ }
+
+ @Override
+ public Index.Searcher searcherFor(ReadCommand command)
+ {
+ return index.searcherFor(command);
+ }
+
+ @Override
+ public RowFilter postIndexQueryFilter()
+ {
+ return postIndexFilter;
+ }
+}
diff --git a/src/java/org/apache/cassandra/index/internal/CassandraIndex.java b/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
index 4f274cf707..5d3e45089a 100644
--- a/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
+++ b/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
@@ -24,7 +24,6 @@ import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
-import java.util.function.BiFunction;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
@@ -46,7 +45,7 @@ import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.CollectionType;
-import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.memtable.Memtable;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.dht.LocalPartitioner;
@@ -55,7 +54,6 @@ import org.apache.cassandra.index.*;
import org.apache.cassandra.index.internal.composites.CompositesSearcher;
import org.apache.cassandra.index.internal.keys.KeysSearcher;
import org.apache.cassandra.index.transactions.IndexTransaction;
-import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.io.sstable.ReducingKeyIterator;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.IndexMetadata;
@@ -272,14 +270,6 @@ public abstract class CassandraIndex implements Index
return indexCfs.getMeanRowCount();
}
- /**
- * No post processing of query results, just return them unchanged
- */
- public BiFunction<PartitionIterator, ReadCommand, PartitionIterator> postProcessorFor(ReadCommand command)
- {
- return (partitionIterator, readCommand) -> partitionIterator;
- }
-
public RowFilter getPostIndexQueryFilter(RowFilter filter)
{
return getTargetExpression(filter.getExpressions()).map(filter::without)
@@ -341,9 +331,10 @@ public abstract class CassandraIndex implements Index
final RegularAndStaticColumns columns,
final long nowInSec,
final WriteContext ctx,
- final IndexTransaction.Type transactionType)
+ final IndexTransaction.Type transactionType,
+ Memtable memtable)
{
- /**
+ /*
* Indexes on regular and static columns (the non primary-key ones) only care about updates with live
* data for the column they index. In particular, they don't care about having just row or range deletions
* as they don't know how to update the index table unless they know exactly the value that is deleted.
@@ -525,7 +516,7 @@ public abstract class CassandraIndex implements Index
cell));
Row row = BTreeRow.noCellLiveRow(buildIndexClustering(rowKey, clustering, cell), info);
PartitionUpdate upd = partitionUpdate(valueKey, row);
- indexCfs.getWriteHandler().write(upd, ctx, UpdateTransaction.NO_OP);
+ indexCfs.getWriteHandler().write(upd, ctx, false);
logger.trace("Inserted entry into index for value {}", valueKey);
}
@@ -571,7 +562,7 @@ public abstract class CassandraIndex implements Index
{
Row row = BTreeRow.emptyDeletedRow(indexClustering, Row.Deletion.regular(deletion));
PartitionUpdate upd = partitionUpdate(indexKey, row);
- indexCfs.getWriteHandler().write(upd, ctx, UpdateTransaction.NO_OP);
+ indexCfs.getWriteHandler().write(upd, ctx, false);
logger.trace("Removed index entry for value {}", indexKey);
}
diff --git a/src/java/org/apache/cassandra/index/internal/CassandraIndexSearcher.java b/src/java/org/apache/cassandra/index/internal/CassandraIndexSearcher.java
index 0935005e0e..720de13c40 100644
--- a/src/java/org/apache/cassandra/index/internal/CassandraIndexSearcher.java
+++ b/src/java/org/apache/cassandra/index/internal/CassandraIndexSearcher.java
@@ -54,6 +54,12 @@ public abstract class CassandraIndexSearcher implements Index.Searcher
this.index = index;
}
+ @Override
+ public ReadCommand command()
+ {
+ return command;
+ }
+
@SuppressWarnings("resource") // Both the OpOrder and 'indexIter' are closed on exception, or through the closing of the result
// of this method.
public UnfilteredPartitionIterator search(ReadExecutionController executionController)
diff --git a/src/java/org/apache/cassandra/index/sasi/SASIIndex.java b/src/java/org/apache/cassandra/index/sasi/SASIIndex.java
index 8ad641e5f3..4668844101 100644
--- a/src/java/org/apache/cassandra/index/sasi/SASIIndex.java
+++ b/src/java/org/apache/cassandra/index/sasi/SASIIndex.java
@@ -27,7 +27,6 @@ import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.Callable;
-import java.util.function.BiFunction;
import com.googlecode.concurrenttrees.common.Iterables;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -44,9 +43,10 @@ import org.apache.cassandra.db.WriteContext;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.db.lifecycle.Tracker;
import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.memtable.Memtable;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.dht.Murmur3Partitioner;
@@ -60,7 +60,7 @@ import org.apache.cassandra.index.sasi.conf.ColumnIndex;
import org.apache.cassandra.index.sasi.conf.IndexMode;
import org.apache.cassandra.index.sasi.disk.OnDiskIndexBuilder.Mode;
import org.apache.cassandra.index.sasi.disk.PerSSTableIndexWriter;
-import org.apache.cassandra.index.sasi.plan.QueryPlan;
+import org.apache.cassandra.index.sasi.plan.SASIIndexSearcher;
import org.apache.cassandra.index.transactions.IndexTransaction;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SSTableFlushObserver;
@@ -90,7 +90,8 @@ public class SASIIndex implements Index, INotificationConsumer
{
public SecondaryIndexBuilder getIndexBuildTask(ColumnFamilyStore cfs,
Set<Index> indexes,
- Collection<SSTableReader> sstablesToRebuild)
+ Collection<SSTableReader> sstablesToRebuild,
+ boolean isFullRebuild)
{
NavigableMap<SSTableReader, Map<ColumnMetadata, ColumnIndex>> sstables = new TreeMap<>(SSTableReader.idComparator);
@@ -182,9 +183,10 @@ public class SASIIndex implements Index, INotificationConsumer
return Collections.emptyMap();
}
+ @Override
public void register(IndexRegistry registry)
{
- registry.registerIndex(this);
+ registry.registerIndex(this, this, () -> new SASIIndexGroup(this));
}
public IndexMetadata getIndexMetadata()
@@ -267,12 +269,7 @@ public class SASIIndex implements Index, INotificationConsumer
{}
@Override
- public boolean supportsReplicaFilteringProtection(RowFilter rowFilter)
- {
- return false;
- }
-
- public Indexer indexerFor(DecoratedKey key, RegularAndStaticColumns columns, long nowInSec, WriteContext context, IndexTransaction.Type transactionType)
+ public Indexer indexerFor(DecoratedKey key, RegularAndStaticColumns columns, long nowInSec, WriteContext context, IndexTransaction.Type transactionType, Memtable memtable)
{
return new Indexer()
{
@@ -320,17 +317,12 @@ public class SASIIndex implements Index, INotificationConsumer
{
TableMetadata config = command.metadata();
ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreInstance(config.id);
- return controller -> new QueryPlan(cfs, command, DatabaseDescriptor.getRangeRpcTimeout(MILLISECONDS)).execute(controller);
- }
-
- public SSTableFlushObserver getFlushObserver(Descriptor descriptor, OperationType opType)
- {
- return newWriter(baseCfs.metadata().partitionKeyType, descriptor, Collections.singletonMap(index.getDefinition(), index), opType);
+ return new SASIIndexSearcher(cfs, command, DatabaseDescriptor.getRangeRpcTimeout(MILLISECONDS));
}
- public BiFunction<PartitionIterator, ReadCommand, PartitionIterator> postProcessorFor(ReadCommand command)
+ public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker)
{
- return (partitionIterator, readCommand) -> partitionIterator;
+ return newWriter(baseCfs.metadata().partitionKeyType, descriptor, Collections.singletonMap(index.getDefinition(), index), tracker.opType());
}
public IndexBuildingSupport getBuildTaskSupport()
diff --git a/src/java/org/apache/cassandra/db/TableWriteHandler.java b/src/java/org/apache/cassandra/index/sasi/SASIIndexGroup.java
similarity index 59%
copy from src/java/org/apache/cassandra/db/TableWriteHandler.java
copy to src/java/org/apache/cassandra/index/sasi/SASIIndexGroup.java
index 4e47221925..8b4c6fa0cb 100644
--- a/src/java/org/apache/cassandra/db/TableWriteHandler.java
+++ b/src/java/org/apache/cassandra/index/sasi/SASIIndexGroup.java
@@ -16,12 +16,23 @@
* limitations under the License.
*/
-package org.apache.cassandra.db;
+package org.apache.cassandra.index.sasi;
-import org.apache.cassandra.db.partitions.PartitionUpdate;
-import org.apache.cassandra.index.transactions.UpdateTransaction;
+import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.index.Index;
+import org.apache.cassandra.index.SingletonIndexGroup;
+import org.apache.cassandra.index.sasi.plan.SASIIndexQueryPlan;
-public interface TableWriteHandler
+public class SASIIndexGroup extends SingletonIndexGroup
{
- void write(PartitionUpdate update, WriteContext context, UpdateTransaction updateTransaction);
+ public SASIIndexGroup(SASIIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Index.QueryPlan queryPlanFor(RowFilter rowFilter)
+ {
+ return SASIIndexQueryPlan.create((SASIIndex) getIndex(), rowFilter);
+ }
}
diff --git a/src/java/org/apache/cassandra/index/sasi/disk/PerSSTableIndexWriter.java b/src/java/org/apache/cassandra/index/sasi/disk/PerSSTableIndexWriter.java
index 6d3db81bcb..f6ace2e1b5 100644
--- a/src/java/org/apache/cassandra/index/sasi/disk/PerSSTableIndexWriter.java
+++ b/src/java/org/apache/cassandra/index/sasi/disk/PerSSTableIndexWriter.java
@@ -282,7 +282,7 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
{
logger.info("Scheduling index flush to {}", outputFile);
- getExecutor().submit((Runnable) () -> {
+ getExecutor().submit(() -> {
long start1 = nanoTime();
OnDiskIndex[] parts = new OnDiskIndex[segments.size() + 1];
@@ -299,7 +299,7 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
// parts are present but there is something still in memory, let's flush that inline
if (!currentBuilder.isEmpty())
{
- @SuppressWarnings("resource")
+ @SuppressWarnings({ "resource", "RedundantSuppression" })
OnDiskIndex last = scheduleSegmentFlush(false).call();
segments.add(ImmediateFuture.success(last));
}
@@ -309,7 +309,7 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
for (Future<OnDiskIndex> f : segments)
{
- @SuppressWarnings("resource")
+ @SuppressWarnings({ "resource", "RedundantSuppression" })
OnDiskIndex part = f.get();
if (part == null)
continue;
@@ -335,7 +335,7 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
for (int segment = 0; segment < segmentNumber; segment++)
{
- @SuppressWarnings("resource")
+ @SuppressWarnings({ "resource", "RedundantSuppression" })
OnDiskIndex part = parts[segment];
if (part != null)
@@ -378,6 +378,6 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
public boolean equals(Object o)
{
- return !(o == null || !(o instanceof PerSSTableIndexWriter)) && descriptor.equals(((PerSSTableIndexWriter) o).descriptor);
+ return o instanceof PerSSTableIndexWriter && descriptor.equals(((PerSSTableIndexWriter) o).descriptor);
}
}
diff --git a/src/java/org/apache/cassandra/index/sasi/plan/SASIIndexQueryPlan.java b/src/java/org/apache/cassandra/index/sasi/plan/SASIIndexQueryPlan.java
new file mode 100644
index 0000000000..06664a1834
--- /dev/null
+++ b/src/java/org/apache/cassandra/index/sasi/plan/SASIIndexQueryPlan.java
@@ -0,0 +1,51 @@
+/*
+ * 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.index.sasi.plan;
+
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.index.SingletonIndexQueryPlan;
+import org.apache.cassandra.index.sasi.SASIIndex;
+
+public class SASIIndexQueryPlan extends SingletonIndexQueryPlan
+{
+ private SASIIndexQueryPlan(SASIIndex index, RowFilter postIndexFilter)
+ {
+ super(index, postIndexFilter);
+ }
+
+ @Nullable
+ public static SASIIndexQueryPlan create(SASIIndex index, RowFilter rowFilter)
+ {
+ for (RowFilter.Expression e : rowFilter.getExpressions())
+ {
+ if (index.supportsExpression(e.column(), e.operator()))
+ return new SASIIndexQueryPlan(index, index.getPostIndexQueryFilter(rowFilter));
+ }
+
+ return null;
+ }
+
+ @Override
+ public boolean supportsReplicaFilteringProtection(RowFilter rowFilter)
+ {
+ return false;
+ }
+}
diff --git a/src/java/org/apache/cassandra/index/sasi/plan/QueryPlan.java b/src/java/org/apache/cassandra/index/sasi/plan/SASIIndexSearcher.java
similarity index 92%
rename from src/java/org/apache/cassandra/index/sasi/plan/QueryPlan.java
rename to src/java/org/apache/cassandra/index/sasi/plan/SASIIndexSearcher.java
index a54dfc8dd5..10d2aab5c4 100644
--- a/src/java/org/apache/cassandra/index/sasi/plan/QueryPlan.java
+++ b/src/java/org/apache/cassandra/index/sasi/plan/SASIIndexSearcher.java
@@ -23,22 +23,30 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.sasi.disk.Token;
import org.apache.cassandra.index.sasi.plan.Operation.OperationType;
-import org.apache.cassandra.exceptions.RequestTimeoutException;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.AbstractIterator;
-public class QueryPlan
+public class SASIIndexSearcher implements Index.Searcher
{
+ private final ReadCommand command;
private final QueryController controller;
- public QueryPlan(ColumnFamilyStore cfs, ReadCommand command, long executionQuotaMs)
+ public SASIIndexSearcher(ColumnFamilyStore cfs, ReadCommand command, long executionQuotaMs)
{
+ this.command = command;
this.controller = new QueryController(cfs, (PartitionRangeReadCommand) command, executionQuotaMs);
}
+ @Override
+ public ReadCommand command()
+ {
+ return command;
+ }
+
/**
* Converts expressions into operation tree (which is currently just a single AND).
*
@@ -63,7 +71,8 @@ public class QueryPlan
}
}
- public UnfilteredPartitionIterator execute(ReadExecutionController executionController) throws RequestTimeoutException
+ @Override
+ public UnfilteredPartitionIterator search(ReadExecutionController executionController)
{
return new ResultIterator(analyze(), controller, executionController);
}
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTable.java b/src/java/org/apache/cassandra/io/sstable/SSTable.java
index 5156adc4e1..519aa39824 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTable.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTable.java
@@ -33,8 +33,6 @@ import com.google.common.base.Predicates;
import com.google.common.collect.Collections2;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.cassandra.cache.ChunkCache;
import org.apache.cassandra.config.CassandraRelevantProperties;
@@ -65,8 +63,6 @@ import static org.apache.cassandra.service.ActiveRepairService.UNREPAIRED_SSTABL
*/
public abstract class SSTable
{
- static final Logger logger = LoggerFactory.getLogger(SSTable.class);
-
public static final int TOMBSTONE_HISTOGRAM_BIN_SIZE = 100;
public static final int TOMBSTONE_HISTOGRAM_SPOOL_SIZE = 100000;
public static final int TOMBSTONE_HISTOGRAM_TTL_ROUND_SECONDS = CassandraRelevantProperties.STREAMING_HISTOGRAM_ROUND_SECONDS.getInt();
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableFlushObserver.java b/src/java/org/apache/cassandra/io/sstable/SSTableFlushObserver.java
index 025d8cbaf4..159e0d43bd 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableFlushObserver.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableFlushObserver.java
@@ -38,12 +38,12 @@ public interface SSTableFlushObserver
*
* @param key the key being appended to SSTable.
* @param keyPosition the position of the key in the SSTable data file
- * @param KeyPositionForSASI SSTable format specific key position for storage attached indexes, it can be
+ * @param keyPositionForSASI SSTable format specific key position for storage attached indexes, it can be
* in data file or in some index file. It is the same position as returned by
* {@link KeyReader#keyPositionForSecondaryIndex()} for the same format, and the same
* position as expected by {@link SSTableReader#keyAtPositionFromSecondaryIndex(long)}.
*/
- void startPartition(DecoratedKey key, long keyPosition, long KeyPositionForSASI);
+ void startPartition(DecoratedKey key, long keyPosition, long keyPositionForSASI);
/**
* Called when a static row is being written to the sstable. If static columns are present in the table, it is called
@@ -54,16 +54,23 @@ public interface SSTableFlushObserver
void staticRow(Row staticRow);
/**
- * Called after the unfiltered cluster is written to the sstable.
- * Will be preceded by a call to {@code startPartition(DecoratedKey, long)},
- * and the cluster should be assumed to belong to that partition.
+ * Called after an unfiltered is written to the sstable.
*
- * @param unfilteredCluster The unfiltered cluster being added to SSTable.
+ * Will be preceded by a call to {@link #startPartition(DecoratedKey, long, long)},
+ * and the unfiltered should be assumed to belong to that partition.
+ *
+ * @param unfiltered the unfiltered being written to the SSTable
*/
- void nextUnfilteredCluster(Unfiltered unfilteredCluster);
+ void nextUnfilteredCluster(Unfiltered unfiltered);
/**
* Called when all data is written to the file and it's ready to be finished up.
*/
void complete();
+
+ /**
+ * Clean up resources on error. There should be no side effects if called multiple times.
+ */
+ @SuppressWarnings("unused")
+ default void abort(Throwable accumulator) {}
}
\ No newline at end of file
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java
index 8bdc1a1917..de9d4f6200 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java
@@ -96,7 +96,7 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
return writer.finished();
}
- @SuppressWarnings("resource") // log and writer closed during doPostCleanup
+ @SuppressWarnings({"resource", "RedundantSuppression"}) // log and writer closed during doPostCleanup
public static SSTableTxnWriter create(ColumnFamilyStore cfs, Descriptor descriptor, long keyCount, long repairedAt, TimeUUID pendingRepair, boolean isTransient, SerializationHeader header)
{
LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.WRITE);
@@ -105,7 +105,7 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
}
- @SuppressWarnings("resource") // log and writer closed during doPostCleanup
+ @SuppressWarnings({"resource", "RedundantSuppression"}) // log and writer closed during doPostCleanup
public static SSTableTxnWriter createRangeAware(TableMetadataRef metadata,
long keyCount,
long repairedAt,
@@ -132,7 +132,7 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
return new SSTableTxnWriter(txn, writer);
}
- @SuppressWarnings("resource") // log and writer closed during doPostCleanup
+ @SuppressWarnings({"resource", "RedundantSuppression"}) // log and writer closed during doPostCleanup
public static SSTableTxnWriter create(TableMetadataRef metadata,
Descriptor descriptor,
long keyCount,
@@ -140,12 +140,12 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
TimeUUID pendingRepair,
boolean isTransient,
SerializationHeader header,
- Collection<Index> indexes,
+ Collection<Index.Group> indexGroups,
SSTable.Owner owner)
{
// if the column family store does not exist, we create a new default SSTableMultiWriter to use:
LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.WRITE);
- SSTableMultiWriter writer = SimpleSSTableMultiWriter.create(descriptor, keyCount, repairedAt, pendingRepair, isTransient, metadata, null, 0, header, indexes, txn, owner);
+ SSTableMultiWriter writer = SimpleSSTableMultiWriter.create(descriptor, keyCount, repairedAt, pendingRepair, isTransient, metadata, null, 0, header, indexGroups, txn, owner);
return new SSTableTxnWriter(txn, writer);
}
}
diff --git a/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java b/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java
index f5f6a443e3..affc0d86b1 100644
--- a/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java
@@ -106,7 +106,7 @@ public class SimpleSSTableMultiWriter implements SSTableMultiWriter
writer.close();
}
- @SuppressWarnings("resource") // SimpleSSTableMultiWriter closes writer
+ @SuppressWarnings({"resource", "RedundantSuppression"}) // SimpleSSTableMultiWriter closes writer
public static SSTableMultiWriter create(Descriptor descriptor,
long keyCount,
long repairedAt,
@@ -116,7 +116,7 @@ public class SimpleSSTableMultiWriter implements SSTableMultiWriter
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
- Collection<Index> indexes,
+ Collection<Index.Group> indexGroups,
LifecycleNewTracker lifecycleNewTracker,
SSTable.Owner owner)
{
@@ -131,8 +131,8 @@ public class SimpleSSTableMultiWriter implements SSTableMultiWriter
.setTableMetadataRef(metadata)
.setMetadataCollector(metadataCollector)
.setSerializationHeader(header)
- .addDefaultComponents()
- .addFlushObserversForSecondaryIndexes(indexes, lifecycleNewTracker.opType())
+ .addDefaultComponents(indexGroups)
+ .addFlushObserversForSecondaryIndexes(indexGroups, lifecycleNewTracker, metadata.get())
.build(lifecycleNewTracker, owner);
return new SimpleSSTableMultiWriter(writer, lifecycleNewTracker);
}
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java
index 8eb92cbf2e..c9cc2b1c2d 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java
@@ -21,8 +21,10 @@ package org.apache.cassandra.io.sstable.format;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -31,7 +33,6 @@ import com.google.common.collect.ImmutableSet;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.SerializationHeader;
-import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.dht.AbstractBounds;
@@ -39,6 +40,7 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.sstable.AbstractRowIndexEntry;
+import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.SSTableFlushObserver;
@@ -49,6 +51,7 @@ import org.apache.cassandra.io.sstable.metadata.MetadataComponent;
import org.apache.cassandra.io.sstable.metadata.MetadataType;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
import org.apache.cassandra.io.util.MmappedRegionsCache;
+import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.Transactional;
@@ -239,7 +242,14 @@ public abstract class SSTableWriter extends SSTable implements Transactional
public final Throwable abort(Throwable accumulate)
{
- return txnProxy.abort(accumulate);
+ try
+ {
+ return txnProxy.abort(accumulate);
+ }
+ finally
+ {
+ observers.forEach(observer -> observer.abort(accumulate));
+ }
}
public final void close()
@@ -249,7 +259,14 @@ public abstract class SSTableWriter extends SSTable implements Transactional
public final void abort()
{
- txnProxy.abort();
+ try
+ {
+ txnProxy.abort();
+ }
+ finally
+ {
+ observers.forEach(observer -> observer.abort(null));
+ }
}
protected Map<MetadataType, MetadataComponent> finalizeMetadata()
@@ -406,7 +423,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional
return (B) this;
}
- public B addDefaultComponents()
+ public B addDefaultComponents(Collection<Index.Group> indexGroups)
{
checkNotNull(getTableMetadataRef());
@@ -423,21 +440,35 @@ public abstract class SSTableWriter extends SSTable implements Transactional
addComponents(ImmutableSet.of(Components.CRC));
}
+ if (!indexGroups.isEmpty())
+ addComponents(indexComponents(indexGroups));
+
return (B) this;
}
- public B addFlushObserversForSecondaryIndexes(Collection<Index> indexes, OperationType operationType)
+ private static Set<Component> indexComponents(Collection<Index.Group> indexGroups)
+ {
+ Set<Component> components = new HashSet<>();
+ for (Index.Group group : indexGroups)
+ {
+ components.addAll(group.getComponents());
+ }
+
+ return components;
+ }
+
+ public B addFlushObserversForSecondaryIndexes(Collection<Index.Group> indexGroups, LifecycleNewTracker tracker, TableMetadata metadata)
{
- if (indexes == null)
+ if (indexGroups == null)
return (B) this;
Collection<SSTableFlushObserver> current = this.flushObservers != null ? this.flushObservers : Collections.emptyList();
- List<SSTableFlushObserver> observers = new ArrayList<>(indexes.size() + current.size());
+ List<SSTableFlushObserver> observers = new ArrayList<>(indexGroups.size() + current.size());
observers.addAll(current);
- for (Index index : indexes)
+ for (Index.Group group : indexGroups)
{
- SSTableFlushObserver observer = index.getFlushObserver(descriptor, operationType);
+ SSTableFlushObserver observer = group.getFlushObserver(descriptor, tracker, metadata);
if (observer != null)
{
observer.begin();
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java
index bd71c4758b..cf3918d05b 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.io.sstable.format;
import java.io.IOException;
import java.nio.BufferOverflowException;
+import java.util.Collection;
import java.util.Set;
import java.util.function.Consumer;
@@ -42,6 +43,7 @@ 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.index.Index;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.compress.CompressedSequentialWriter;
import org.apache.cassandra.io.compress.CompressionMetadata;
@@ -436,9 +438,9 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter> ex
}
@Override
- public B addDefaultComponents()
+ public B addDefaultComponents(Collection<Index.Group> indexGroups)
{
- super.addDefaultComponents();
+ super.addDefaultComponents(indexGroups);
if (FilterComponent.shouldUseBloomFilter(getTableMetadataRef().getLocal().params.bloomFilterFpChance))
{
diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java
index e1cb14b905..f7f2e4d850 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java
@@ -40,11 +40,11 @@ import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.GaugeProvider;
import org.apache.cassandra.io.sstable.IScrubber;
import org.apache.cassandra.io.sstable.MetricsProviders;
-import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.filter.BloomFilterMetrics;
import org.apache.cassandra.io.sstable.format.AbstractSSTableFormat;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java
index f0eb2235e6..6a837946c1 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.io.sstable.format.big;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;
@@ -34,6 +35,7 @@ import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
+import org.apache.cassandra.index.Index;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.sstable.AbstractRowIndexEntry;
import org.apache.cassandra.io.sstable.Descriptor;
@@ -143,7 +145,7 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
return entry;
}
- @SuppressWarnings("resource")
+ @SuppressWarnings({"resource", "RedundantSuppression"})
private BigTableReader openInternal(IndexSummaryBuilder.ReadableBoundary boundary, SSTableReader.OpenReason openReason)
{
assert boundary == null || (boundary.indexLength > 0 && boundary.dataLength > 0);
@@ -278,7 +280,7 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
summary = new IndexSummaryBuilder(b.getKeyCount(), b.getTableMetadataRef().getLocal().params.minIndexInterval, Downsampling.BASE_SAMPLING_LEVEL);
// register listeners to be alerted when the data files are flushed
writer.setPostFlushListener(() -> summary.markIndexSynced(writer.getLastFlushOffset()));
- @SuppressWarnings("resource")
+ @SuppressWarnings({"resource", "RedundantSuppression"})
SequentialWriter dataWriter = b.getDataWriter();
dataWriter.setPostFlushListener(() -> summary.markDataSynced(dataWriter.getLastFlushOffset()));
}
@@ -384,9 +386,9 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
}
@Override
- public Builder addDefaultComponents()
+ public Builder addDefaultComponents(Collection<Index.Group> indexGroups)
{
- super.addDefaultComponents();
+ super.addDefaultComponents(indexGroups);
addComponents(ImmutableSet.of(Components.PRIMARY_INDEX, Components.SUMMARY));
diff --git a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableWriter.java
index 29706fc048..0e462125a9 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableWriter.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.io.sstable.format.bti;
import java.io.IOException;
+import java.util.Collection;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -31,6 +32,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
+import org.apache.cassandra.index.Index;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.sstable.AbstractRowIndexEntry;
@@ -379,9 +381,9 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
}
@Override
- public Builder addDefaultComponents()
+ public Builder addDefaultComponents(Collection<Index.Group> indexGroups)
{
- super.addDefaultComponents();
+ super.addDefaultComponents(indexGroups);
addComponents(ImmutableSet.of(Components.PARTITION_INDEX, Components.ROW_INDEX));
diff --git a/src/java/org/apache/cassandra/metrics/ClientRangeRequestMetrics.java b/src/java/org/apache/cassandra/metrics/ClientRangeRequestMetrics.java
new file mode 100644
index 0000000000..c974651381
--- /dev/null
+++ b/src/java/org/apache/cassandra/metrics/ClientRangeRequestMetrics.java
@@ -0,0 +1,47 @@
+/*
+ * 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.metrics;
+
+import com.codahale.metrics.Histogram;
+
+import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
+
+/**
+ * Metrics for tracking information about range requests.
+ *
+ */
+public class ClientRangeRequestMetrics extends ClientRequestMetrics
+{
+ /**
+ * Metric for tracking the number of round trips made for a range request.
+ */
+ public final Histogram roundTrips;
+
+ public ClientRangeRequestMetrics(String scope)
+ {
+ super(scope);
+ roundTrips = Metrics.histogram(factory.createMetricName("RoundTripsPerReadHistogram"), false);
+ }
+
+ public void release()
+ {
+ super.release();
+ Metrics.remove(factory.createMetricName("RoundTripsPerReadHistogram"));
+ }
+}
diff --git a/src/java/org/apache/cassandra/schema/IndexMetadata.java b/src/java/org/apache/cassandra/schema/IndexMetadata.java
index 4adbce170b..cdba139b5d 100644
--- a/src/java/org/apache/cassandra/schema/IndexMetadata.java
+++ b/src/java/org/apache/cassandra/schema/IndexMetadata.java
@@ -21,6 +21,7 @@ package org.apache.cassandra.schema;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -37,6 +38,7 @@ import org.apache.cassandra.cql3.statements.schema.IndexTarget;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.UnknownIndexException;
import org.apache.cassandra.index.Index;
+import org.apache.cassandra.index.sasi.SASIIndex;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.FBUtilities;
@@ -55,6 +57,16 @@ public final class IndexMetadata
public static final Serializer serializer = new Serializer();
+ /**
+ * A mapping of user-friendly index names to their fully qualified index class names.
+ */
+ private static final Map<String, String> indexNameAliases = new ConcurrentHashMap<>();
+
+ static
+ {
+ indexNameAliases.put(SASIIndex.class.getSimpleName(), SASIIndex.class.getCanonicalName());
+ }
+
public enum Kind
{
KEYS, CUSTOM, COMPOSITES
@@ -122,7 +134,10 @@ public final class IndexMetadata
if (options == null || !options.containsKey(IndexTarget.CUSTOM_INDEX_OPTION_NAME))
throw new ConfigurationException(String.format("Required option missing for index %s : %s",
name, IndexTarget.CUSTOM_INDEX_OPTION_NAME));
- String className = options.get(IndexTarget.CUSTOM_INDEX_OPTION_NAME);
+
+ // Find any aliases to the fully qualified index class name:
+ String className = expandAliases(options.get(IndexTarget.CUSTOM_INDEX_OPTION_NAME));
+
Class<Index> indexerClass = FBUtilities.classForName(className, "custom indexer");
if (!Index.class.isAssignableFrom(indexerClass))
throw new ConfigurationException(String.format("Specified Indexer class (%s) does not implement the Indexer interface", className));
@@ -130,6 +145,11 @@ public final class IndexMetadata
}
}
+ public static String expandAliases(String className)
+ {
+ return indexNameAliases.getOrDefault(className, className);
+ }
+
private void validateCustomIndexOptions(TableMetadata table, Class<? extends Index> indexerClass, Map<String, String> options)
{
try
diff --git a/src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosUncommittedIndex.java b/src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosUncommittedIndex.java
index c42923607b..bfb73dba05 100644
--- a/src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosUncommittedIndex.java
+++ b/src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosUncommittedIndex.java
@@ -25,7 +25,6 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.Callable;
-import java.util.function.BiFunction;
import java.util.stream.Collectors;
import com.google.common.base.Preconditions;
@@ -48,7 +47,6 @@ import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.memtable.Memtable;
-import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
@@ -245,16 +243,11 @@ public class PaxosUncommittedIndex implements Index, PaxosUncommittedTracker.Upd
}
- public Indexer indexerFor(DecoratedKey key, RegularAndStaticColumns columns, long nowInSec, WriteContext ctx, IndexTransaction.Type transactionType)
+ public Indexer indexerFor(DecoratedKey key, RegularAndStaticColumns columns, long nowInSec, WriteContext ctx, IndexTransaction.Type transactionType, Memtable memtable)
{
return indexer;
}
- public BiFunction<PartitionIterator, ReadCommand, PartitionIterator> postProcessorFor(ReadCommand command)
- {
- return null;
- }
-
public Searcher searcherFor(ReadCommand command)
{
throw new UnsupportedOperationException();
diff --git a/src/java/org/apache/cassandra/service/reads/DataResolver.java b/src/java/org/apache/cassandra/service/reads/DataResolver.java
index e7e49521c1..66719554b9 100644
--- a/src/java/org/apache/cassandra/service/reads/DataResolver.java
+++ b/src/java/org/apache/cassandra/service/reads/DataResolver.java
@@ -29,7 +29,6 @@ import javax.annotation.Nullable;
import com.google.common.base.Joiner;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.ReadCommand;
@@ -51,7 +50,6 @@ import org.apache.cassandra.index.Index;
import org.apache.cassandra.locator.Endpoints;
import org.apache.cassandra.locator.ReplicaPlan;
import org.apache.cassandra.net.Message;
-import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.reads.repair.NoopReadRepair;
import org.apache.cassandra.service.reads.repair.ReadRepair;
@@ -137,22 +135,11 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
if (command.rowFilter().isEmpty())
return false;
- IndexMetadata indexMetadata = command.indexMetadata();
-
- if (indexMetadata == null || !indexMetadata.isCustom())
- {
+ Index.QueryPlan queryPlan = command.indexQueryPlan();
+ if (queryPlan == null )
return true;
- }
-
- ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(command.metadata().id);
-
- assert cfs != null;
-
- Index index = command.getIndex(cfs);
- assert index != null;
-
- return index.supportsReplicaFilteringProtection(command.rowFilter());
+ return queryPlan.supportsReplicaFilteringProtection(command.rowFilter());
}
private class ResolveContext
@@ -258,13 +245,24 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
PartitionIterator completedPartitions = resolveWithReadRepair(secondPhaseContext,
i -> rfp.queryProtectedPartitions(firstPhasePartitions, i),
- results -> command.rowFilter().filter(results, command.metadata(), command.nowInSec()),
+ preCountFilterForReplicaFilteringProtection(),
repairedDataTracker);
// Ensure that the RFP instance has a chance to record metrics when the iterator closes.
return PartitionIterators.doOnClose(completedPartitions, firstPhasePartitions::close);
}
+ private UnaryOperator<PartitionIterator> preCountFilterForReplicaFilteringProtection()
+ {
+ return results -> {
+ Index.Searcher searcher = command.indexSearcher();
+ // in case of "ALLOW FILTERING" without index
+ if (searcher == null)
+ return command.rowFilter().filter(results, command.metadata(), command.nowInSec());
+ return searcher.filterReplicaFilteringProtection(results);
+ };
+ }
+
@SuppressWarnings("resource")
private PartitionIterator resolveInternal(ResolveContext context,
UnfilteredPartitionIterators.MergeListener mergeListener,
diff --git a/src/java/org/apache/cassandra/service/reads/ShortReadRowsProtection.java b/src/java/org/apache/cassandra/service/reads/ShortReadRowsProtection.java
index 8061f0a008..4ed9e32956 100644
--- a/src/java/org/apache/cassandra/service/reads/ShortReadRowsProtection.java
+++ b/src/java/org/apache/cassandra/service/reads/ShortReadRowsProtection.java
@@ -184,6 +184,6 @@ class ShortReadRowsProtection extends Transformation implements MoreRows<Unfilte
command.limits().forShortReadRetry(toQuery),
partitionKey,
filter,
- command.indexMetadata());
+ command.indexQueryPlan());
}
}
diff --git a/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java b/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java
index b353b4bbad..8881308534 100644
--- a/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java
+++ b/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java
@@ -42,7 +42,7 @@ import org.apache.cassandra.exceptions.UnavailableException;
import org.apache.cassandra.locator.EndpointsForRange;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.locator.ReplicaPlan;
-import org.apache.cassandra.metrics.ClientRequestMetrics;
+import org.apache.cassandra.metrics.ClientRangeRequestMetrics;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageProxy;
@@ -60,7 +60,7 @@ public class RangeCommandIterator extends AbstractIterator<RowIterator> implemen
{
private static final Logger logger = LoggerFactory.getLogger(RangeCommandIterator.class);
- private static final ClientRequestMetrics rangeMetrics = new ClientRequestMetrics("RangeSlice");
+ public static final ClientRangeRequestMetrics rangeMetrics = new ClientRangeRequestMetrics("RangeSlice");
private final CloseableIterator<ReplicaPlan.ForRangeRead> replicaPlans;
private final int totalRangeCount;
@@ -270,6 +270,7 @@ public class RangeCommandIterator extends AbstractIterator<RowIterator> implemen
{
long latency = nanoTime() - startTime;
rangeMetrics.addNano(latency);
+ rangeMetrics.roundTrips.update(batchesRequested);
Keyspace.openAndGetStore(command.metadata()).metric.coordinatorScanLatency.update(latency, TimeUnit.NANOSECONDS);
}
}
diff --git a/src/java/org/apache/cassandra/service/reads/range/RangeCommands.java b/src/java/org/apache/cassandra/service/reads/range/RangeCommands.java
index 1c5ce7f147..ca3c220d70 100644
--- a/src/java/org/apache/cassandra/service/reads/range/RangeCommands.java
+++ b/src/java/org/apache/cassandra/service/reads/range/RangeCommands.java
@@ -109,7 +109,7 @@ public class RangeCommands
static float estimateResultsPerRange(PartitionRangeReadCommand command, Keyspace keyspace)
{
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(command.metadata().id);
- Index index = command.getIndex(cfs);
+ Index.QueryPlan index = command.indexQueryPlan();
float maxExpectedResults = index == null
? command.limits().estimateTotalResults(cfs)
: index.getEstimatedResultRows();
diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeIndexTestBase.java b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeIndexTestBase.java
new file mode 100644
index 0000000000..923f29abd4
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeIndexTestBase.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.upgrade;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.vdurmont.semver4j.Semver;
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.IUpgradeableInstance;
+
+import static java.lang.String.format;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL;
+import static org.apache.cassandra.distributed.api.ConsistencyLevel.ONE;
+import static org.apache.cassandra.distributed.api.ConsistencyLevel.QUORUM;
+import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
+import static org.apache.cassandra.distributed.shared.AssertUtils.row;
+
+public class MixedModeIndexTestBase extends UpgradeTestBase
+{
+ private static final AtomicInteger TABLE_INDEX = new AtomicInteger(1);
+
+ protected static void testIndex(Semver initial) throws Throwable
+ {
+ List<Tester> testers = new ArrayList<>();
+ testers.addAll(Tester.create(1, ALL));
+ testers.addAll(Tester.create(2, ALL, QUORUM));
+ testers.addAll(Tester.create(3, ALL, QUORUM, ONE));
+
+ new TestCase()
+ .nodes(3)
+ .nodesToUpgrade(1)
+ .upgradesToCurrentFrom(initial)
+ .withConfig(config -> config.set("read_request_timeout_in_ms", SECONDS.toMillis(30))
+ .set("write_request_timeout_in_ms", SECONDS.toMillis(30)))
+ .setup(cluster -> {
+ for (Tester tester : testers)
+ {
+ tester.createTable(cluster);
+ tester.createIndex(cluster);
+ tester.writeRows(cluster);
+ }
+ }).runAfterNodeUpgrade((cluster, node) -> {
+ for (Tester tester : testers)
+ tester.readRows(cluster);
+ }).run();
+ }
+
+ private static class Tester
+ {
+ private final String table;
+ private final int numWrittenReplicas;
+ private final ConsistencyLevel readConsistencyLevel;
+
+ private Tester(int numWrittenReplicas, ConsistencyLevel readConsistencyLevel)
+ {
+ this.table = "cf_" + TABLE_INDEX.getAndIncrement();
+ this.numWrittenReplicas = numWrittenReplicas;
+ this.readConsistencyLevel = readConsistencyLevel;
+ }
+
+ private static List<Tester> create(int numWrittenReplicas, ConsistencyLevel... readConsistencyLevels)
+ {
+ return Stream.of(readConsistencyLevels)
+ .map(readConsistencyLevel -> new Tester(numWrittenReplicas, readConsistencyLevel))
+ .collect(Collectors.toList());
+ }
+
+ private void createTable(UpgradeableCluster cluster)
+ {
+ cluster.schemaChange(withKeyspace(String.format("CREATE TABLE %%s.%s (k int, c int, v int, PRIMARY KEY (k, c))", table)));
+ }
+
+ public void createIndex(UpgradeableCluster cluster)
+ {
+ cluster.schemaChange(withKeyspace(String.format("CREATE INDEX ON %%s.%s (v);", table)));
+ }
+
+ private void writeRows(UpgradeableCluster cluster)
+ {
+ String query = withKeyspace(String.format("INSERT INTO %%s.%s (k, c, v) VALUES (?, ?, ?)", table));
+ for (int i = 1; i <= numWrittenReplicas; i++)
+ {
+ IUpgradeableInstance node = cluster.get(i);
+ node.executeInternal(query, 1, 1, 10);
+ node.executeInternal(query, 1, 2, 20);
+ node.executeInternal(query, 1, 3, 30);
+ }
+ }
+
+ private void readRows(UpgradeableCluster cluster)
+ {
+ String query = withKeyspace(String.format("SELECT * FROM %%s.%s WHERE v = ? ALLOW FILTERING", table));
+ int coordinator = 1;
+ try
+ {
+ for (coordinator = 1; coordinator <= cluster.size(); coordinator++)
+ {
+ assertRows(cluster.coordinator(coordinator).execute(query, readConsistencyLevel, 10),
+ row(1, 1, 10));
+ assertRows(cluster.coordinator(coordinator).execute(query, readConsistencyLevel, 20),
+ row(1, 2, 20));
+ assertRows(cluster.coordinator(coordinator).execute(query, readConsistencyLevel, 30),
+ row(1, 3, 30));
+ }
+ }
+ catch (Throwable t)
+ {
+ String format = "Unexpected error reading rows with %d written replicas, CL=%s and coordinator=%s";
+ throw new AssertionError(format(format, numWrittenReplicas, readConsistencyLevel, coordinator), t);
+ }
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/db/TableWriteHandler.java b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeIndexV40Test.java
similarity index 73%
copy from src/java/org/apache/cassandra/db/TableWriteHandler.java
copy to test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeIndexV40Test.java
index 4e47221925..bd4488b928 100644
--- a/src/java/org/apache/cassandra/db/TableWriteHandler.java
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeIndexV40Test.java
@@ -16,12 +16,18 @@
* limitations under the License.
*/
-package org.apache.cassandra.db;
+package org.apache.cassandra.distributed.upgrade;
-import org.apache.cassandra.db.partitions.PartitionUpdate;
-import org.apache.cassandra.index.transactions.UpdateTransaction;
+import org.junit.Test;
-public interface TableWriteHandler
+/**
+ * {@link MixedModeIndexTestBase} for upgrades from 4.0.
+ */
+public class MixedModeIndexV40Test extends MixedModeIndexTestBase
{
- void write(PartitionUpdate update, WriteContext context, UpdateTransaction updateTransaction);
+ @Test
+ public void testIndex() throws Throwable
+ {
+ testIndex(v40);
+ }
}
diff --git a/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java b/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java
index 14137f13dc..c13ef60018 100644
--- a/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java
+++ b/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java
@@ -120,7 +120,7 @@ public class SecondaryIndexTest
.filterOn("birthdate", Operator.EQ, 1L)
.build();
- Index.Searcher searcher = rc.getIndex(cfs).searcherFor(rc);
+ Index.Searcher searcher = rc.indexSearcher();
try (ReadExecutionController executionController = rc.executionController();
UnfilteredPartitionIterator pi = searcher.search(executionController))
{
@@ -207,7 +207,7 @@ public class SecondaryIndexTest
// verify that it's not being indexed under any other value either
ReadCommand rc = Util.cmd(cfs).build();
- assertNull(rc.getIndex(cfs));
+ assertNull(rc.indexSearcher());
// resurrect w/ a newer timestamp
new RowUpdateBuilder(cfs.metadata(), 2, "k1").clustering("c").add("birthdate", 1L).build().apply();;
@@ -225,13 +225,13 @@ public class SecondaryIndexTest
// todo - checking the # of index searchers for the command is probably not the best thing to test here
RowUpdateBuilder.deleteRow(cfs.metadata(), 3, "k1", "c").applyUnsafe();
rc = Util.cmd(cfs).build();
- assertNull(rc.getIndex(cfs));
+ assertNull(rc.indexSearcher());
// make sure obsolete mutations don't generate an index entry
// todo - checking the # of index searchers for the command is probably not the best thing to test here
new RowUpdateBuilder(cfs.metadata(), 3, "k1").clustering("c").add("birthdate", 1L).build().apply();;
rc = Util.cmd(cfs).build();
- assertNull(rc.getIndex(cfs));
+ assertNull(rc.indexSearcher());
}
@Test
@@ -546,7 +546,7 @@ public class SecondaryIndexTest
.filterOn("notbirthdate", Operator.EQ, 0L)
.build();
- assertEquals("notbirthdate_key_index", rc.indexMetadata().name);
+ assertEquals("notbirthdate_key_index", rc.indexQueryPlan().getFirst().getIndexMetadata().name);
}
private void assertIndexedNone(ColumnFamilyStore cfs, ByteBuffer col, Object val)
@@ -562,7 +562,7 @@ public class SecondaryIndexTest
ColumnMetadata cdef = cfs.metadata().getColumn(col);
ReadCommand rc = Util.cmd(cfs).filterOn(cdef.name.toString(), Operator.EQ, ((AbstractType) cdef.cellValueType()).decompose(val)).build();
- Index.Searcher searcher = rc.getIndex(cfs).searcherFor(rc);
+ Index.Searcher searcher = rc.indexSearcher();
if (count != 0)
assertNotNull(searcher);
diff --git a/test/unit/org/apache/cassandra/db/SerializationHeaderTest.java b/test/unit/org/apache/cassandra/db/SerializationHeaderTest.java
index d77e1fefc0..2b8db525f1 100644
--- a/test/unit/org/apache/cassandra/db/SerializationHeaderTest.java
+++ b/test/unit/org/apache/cassandra/db/SerializationHeaderTest.java
@@ -56,7 +56,7 @@ import org.apache.cassandra.schema.TableMetadataRef;
public class SerializationHeaderTest
{
- private static String KEYSPACE = "SerializationHeaderTest";
+ private static final String KEYSPACE = "SerializationHeaderTest";
static
{
@@ -102,7 +102,7 @@ public class SerializationHeaderTest
.setSerializationHeader(header)
.setFlushObservers(Collections.emptyList())
.setMetadataCollector(new MetadataCollector(schema.comparator))
- .addDefaultComponents()
+ .addDefaultComponents(Collections.emptySet())
.build(txn, null))
{
ColumnMetadata cd = schema.getColumn(v);
@@ -139,7 +139,7 @@ public class SerializationHeaderTest
{
UnfilteredRowIterator partition = partitions.next();
long value = Int32Type.instance.compose(((Row)partition.next()).getCell(columnRegular).buffer());
- Assert.assertEquals(value, (long)i);
+ Assert.assertEquals(value, i);
Assert.assertTrue(partition.staticRow().isEmpty());
Assert.assertFalse(partition.hasNext());
}
diff --git a/test/unit/org/apache/cassandra/db/compaction/ActiveCompactionsTest.java b/test/unit/org/apache/cassandra/db/compaction/ActiveCompactionsTest.java
index 78497965d7..cb6d5c8241 100644
--- a/test/unit/org/apache/cassandra/db/compaction/ActiveCompactionsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/ActiveCompactionsTest.java
@@ -82,7 +82,7 @@ public class ActiveCompactionsTest extends CQLTester
for (int ii = 0; ii < loopCount; ii++)
{
CountDownLatch trigger = new CountDownLatch(1);
- SecondaryIndexBuilder builder = idx.getBuildTaskSupport().getIndexBuildTask(getCurrentColumnFamilyStore(), Collections.singleton(idx), sstables);
+ SecondaryIndexBuilder builder = idx.getBuildTaskSupport().getIndexBuildTask(getCurrentColumnFamilyStore(), Collections.singleton(idx), sstables, true);
Future<?> f1 = es.submit(() -> {
Uninterruptibles.awaitUninterruptibly(trigger);
try
@@ -119,7 +119,7 @@ public class ActiveCompactionsTest extends CQLTester
Index idx = getCurrentColumnFamilyStore().indexManager.getIndexByName(idxName);
Set<SSTableReader> sstables = getCurrentColumnFamilyStore().getLiveSSTables();
- SecondaryIndexBuilder builder = idx.getBuildTaskSupport().getIndexBuildTask(getCurrentColumnFamilyStore(), Collections.singleton(idx), sstables);
+ SecondaryIndexBuilder builder = idx.getBuildTaskSupport().getIndexBuildTask(getCurrentColumnFamilyStore(), Collections.singleton(idx), sstables, false);
MockActiveCompactions mockActiveCompactions = new MockActiveCompactions();
CompactionManager.instance.submitIndexBuild(builder, mockActiveCompactions).get();
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/RealTransactionsTest.java b/test/unit/org/apache/cassandra/db/lifecycle/RealTransactionsTest.java
index bf03f9cd95..6d447ddb25 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/RealTransactionsTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/RealTransactionsTest.java
@@ -163,9 +163,9 @@ public class RealTransactionsTest extends SchemaLoader
rewriter.switchWriter(desc.getFormat().getWriterFactory().builder(desc)
.setTableMetadataRef(metadata)
.setSerializationHeader(SerializationHeader.make(cfs.metadata(), txn.originals()))
- .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexes(), txn.opType())
+ .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, metadata.get())
.setMetadataCollector(new MetadataCollector(cfs.metadata().comparator))
- .addDefaultComponents()
+ .addDefaultComponents(cfs.indexManager.listIndexGroups())
.build(txn, cfs));
while (ci.hasNext())
{
diff --git a/test/unit/org/apache/cassandra/index/CustomIndexTest.java b/test/unit/org/apache/cassandra/index/CustomIndexTest.java
index c30481a08c..a296ea779f 100644
--- a/test/unit/org/apache/cassandra/index/CustomIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/CustomIndexTest.java
@@ -24,31 +24,42 @@ import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.junit.Test;
import com.datastax.driver.core.exceptions.QueryValidationException;
import org.apache.cassandra.Util;
-import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.cql3.restrictions.IndexRestrictions;
import org.apache.cassandra.cql3.restrictions.StatementRestrictions;
import org.apache.cassandra.cql3.statements.schema.IndexTarget;
import org.apache.cassandra.cql3.statements.ModificationStatement;
+import org.apache.cassandra.db.ColumnFamilyStore.FlushReason;
import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.memtable.Memtable;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.Row;
+import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.SSTableFlushObserver;
import org.apache.cassandra.index.transactions.IndexTransaction;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.Indexes;
+import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
@@ -227,7 +238,7 @@ public class CustomIndexTest extends CQLTester
}
@Test
- public void defaultIndexNameContainsTargetColumns() throws Throwable
+ public void defaultIndexNameContainsTargetColumns()
{
createTable("CREATE TABLE %s(k int, c int, v1 int, v2 int, PRIMARY KEY(k,c))");
@@ -253,7 +264,7 @@ public class CustomIndexTest extends CQLTester
}
@Test
- public void createMultiColumnIndexes() throws Throwable
+ public void createMultiColumnIndexes()
{
// smoke test for various permutations of multicolumn indexes
createTable("CREATE TABLE %s (" +
@@ -326,7 +337,7 @@ public class CustomIndexTest extends CQLTester
}
@Test
- public void createMultiColumnIndexIncludingUserTypeColumn() throws Throwable
+ public void createMultiColumnIndexIncludingUserTypeColumn()
{
String myType = KEYSPACE + '.' + createType("CREATE TYPE %s (a int, b int)");
createTable("CREATE TABLE %s (k int PRIMARY KEY, v1 int, v2 frozen<" + myType + ">)");
@@ -535,7 +546,7 @@ public class CustomIndexTest extends CQLTester
}
@Test
- public void reloadIndexMetadataOnBaseCfsReload() throws Throwable
+ public void reloadIndexMetadataOnBaseCfsReload()
{
// verify that whenever the base table TableMetadata is reloaded, a reload of the index
// metadata is performed
@@ -607,7 +618,7 @@ public class CustomIndexTest extends CQLTester
}
@Test
- public void validateOptions() throws Throwable
+ public void validateOptions()
{
createTable("CREATE TABLE %s(k int, c int, v1 int, v2 int, PRIMARY KEY(k,c))");
createIndex(String.format("CREATE CUSTOM INDEX ON %%s(c, v2) USING '%s' WITH OPTIONS = {'foo':'bar'}",
@@ -617,7 +628,7 @@ public class CustomIndexTest extends CQLTester
}
@Test
- public void validateOptionsWithTableMetadata() throws Throwable
+ public void validateOptionsWithTableMetadata()
{
createTable("CREATE TABLE %s(k int, c int, v1 int, v2 int, PRIMARY KEY(k,c))");
createIndex(String.format("CREATE CUSTOM INDEX ON %%s(c, v2) USING '%s' WITH OPTIONS = {'foo':'bar'}",
@@ -829,6 +840,7 @@ public class CustomIndexTest extends CQLTester
// Used for index creation above
+ @SuppressWarnings("unused")
public static class BrokenCustom2I extends StubIndex
{
public BrokenCustom2I(ColumnFamilyStore baseCfs, IndexMetadata metadata)
@@ -842,12 +854,10 @@ public class CustomIndexTest extends CQLTester
}
}
- private void testCreateIndex(String indexName, String... targetColumnNames) throws Throwable
+ private void testCreateIndex(String indexName, String... targetColumnNames)
{
createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(%s) USING '%s'",
- indexName,
- Arrays.stream(targetColumnNames).collect(Collectors.joining(",")),
- StubIndex.class.getName()));
+ indexName, String.join(",", targetColumnNames), StubIndex.class.getName()));
assertIndexCreated(indexName, targetColumnNames);
}
@@ -1023,6 +1033,7 @@ public class CustomIndexTest extends CQLTester
super(baseCfs, metadata);
}
+ @SuppressWarnings("unused")
public static Map<String, String> validateOptions(Map<String, String> options)
{
IndexWithValidateOptions.options = options;
@@ -1040,6 +1051,7 @@ public class CustomIndexTest extends CQLTester
super(baseCfs, metadata);
}
+ @SuppressWarnings("unused")
public static Map<String, String> validateOptions(Map<String, String> options, TableMetadata table)
{
IndexWithOverloadedValidateOptions.options = options;
@@ -1081,11 +1093,13 @@ public class CustomIndexTest extends CQLTester
// When we're done indexing the partition, the test checks the states of the
// various OpOrder.Groups, which it can obtain from this index.
+ @Override
public Indexer indexerFor(final DecoratedKey key,
RegularAndStaticColumns columns,
long nowInSec,
WriteContext ctx,
- IndexTransaction.Type transactionType)
+ IndexTransaction.Type transactionType,
+ Memtable memtable)
{
CassandraWriteContext cassandraWriteContext = (CassandraWriteContext) ctx;
if (readOrderingAtStart == null)
@@ -1134,4 +1148,529 @@ public class CustomIndexTest extends CQLTester
};
}
}
+
+
+ @Test
+ public void testFlushObserver() throws Throwable
+ {
+ createTable("CREATE TABLE %s (k int, c int, s int static, v int, PRIMARY KEY (k, c))");
+ String indexName = "test_index_with_flush_observer";
+ createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v) USING '%s'",
+ indexName, IndexWithFlushObserver.class.getName()));
+
+ execute("INSERT INTO %s (k, c, s, v) VALUES (?, ?, ?, ?)", 0, 0, 0, 0);
+ execute("INSERT INTO %s (k, c, s, v) VALUES (?, ?, ?, ?)", 0, 1, 1, 1);
+ execute("INSERT INTO %s (k, c, s, v) VALUES (?, ?, ?, ?)", 1, 0, 2, 2);
+ execute("INSERT INTO %s (k, c, s, v) VALUES (?, ?, ?, ?)", 1, 1, 3, 3);
+
+ ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+ SecondaryIndexManager indexManager = cfs.indexManager;
+ IndexWithFlushObserver index = (IndexWithFlushObserver) indexManager.getIndexByName(indexName);
+
+ assertEquals(0, index.beginFlushCalls.get());
+ assertEquals(0, index.flushedPartitions.get());
+ assertEquals(0, index.flushedStaticRows.get());
+ assertEquals(0, index.flushedUnfiltereds.get());
+ assertEquals(0, index.completeFlushCalls.get());
+
+ cfs.forceBlockingFlush(FlushReason.UNIT_TESTS);
+
+ assertEquals(1, index.beginFlushCalls.get());
+ assertEquals(2, index.flushedPartitions.get());
+ assertEquals(2, index.flushedStaticRows.get());
+ assertEquals(4, index.flushedUnfiltereds.get());
+ assertEquals(1, index.completeFlushCalls.get());
+
+ execute("DELETE FROM %s WHERE k=?", 0);
+ execute("DELETE FROM %s WHERE k=? AND c>=?", 1, 1);
+ index.reset();
+ cfs.forceBlockingFlush(FlushReason.UNIT_TESTS);
+
+ assertEquals(1, index.beginFlushCalls.get());
+ assertEquals(2, index.flushedPartitions.get());
+ assertEquals(0, index.flushedStaticRows.get()); // flushed data has no static values..
+ assertEquals(2, index.flushedUnfiltereds.get());
+ assertEquals(1, index.completeFlushCalls.get());
+ }
+
+ /**
+ * A {@link StubIndex} using a {@link SSTableFlushObserver} that just keeps count of operations.
+ */
+ public static final class IndexWithFlushObserver extends StubIndex
+ {
+
+ AtomicInteger beginFlushCalls = new AtomicInteger();
+ AtomicInteger flushedPartitions = new AtomicInteger();
+ AtomicInteger flushedStaticRows = new AtomicInteger();
+ AtomicInteger flushedUnfiltereds = new AtomicInteger();
+ AtomicInteger completeFlushCalls = new AtomicInteger();
+
+ public IndexWithFlushObserver(ColumnFamilyStore baseCfs, IndexMetadata metadata)
+ {
+ super(baseCfs, metadata);
+ }
+
+ @Override
+ public void reset()
+ {
+ super.reset();
+ beginFlushCalls.set(0);
+ flushedPartitions.set(0);
+ flushedStaticRows.set(0);
+ flushedUnfiltereds.set(0);
+ completeFlushCalls.set(0);
+ }
+
+ @Override
+ public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker)
+ {
+ return new SSTableFlushObserver() {
+
+ @Override
+ public void begin()
+ {
+ beginFlushCalls.incrementAndGet();
+ }
+
+ @Override
+ public void startPartition(DecoratedKey key, long keyPosition, long keyPositionForSASI)
+ {
+ flushedPartitions.incrementAndGet();
+ }
+
+ @Override
+ public void staticRow(Row staticRow)
+ {
+ flushedStaticRows.incrementAndGet();
+ }
+
+ @Override
+ public void nextUnfilteredCluster(Unfiltered unfiltered)
+ {
+ flushedUnfiltereds.incrementAndGet();
+ }
+
+ @Override
+ public void complete()
+ {
+ completeFlushCalls.incrementAndGet();
+ }
+ };
+ }
+ }
+
+ /**
+ * Verify that writes for indexes in the same {@link Index.Group} are grouped.
+ */
+ @Test
+ public void testGroupedWrites() throws Throwable
+ {
+ // create the schema with two indexes in the same group
+ String indexClassName = IndexWithSharedGroup.class.getName();
+ createTable("CREATE TABLE %s (k int, c int, s int static, v int, PRIMARY KEY (k,c))");
+ createIndex(String.format("CREATE CUSTOM INDEX grouped_index_c ON %%s(c) USING '%s'", indexClassName));
+ createIndex(String.format("CREATE CUSTOM INDEX grouped_index_v ON %%s(v) USING '%s'", indexClassName));
+
+ // retrieve the indexes and their shared group
+ ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+ SecondaryIndexManager indexManager = cfs.indexManager;
+ StubIndex index1 = (IndexWithSharedGroup) indexManager.getIndexByName("grouped_index_c");
+ StubIndex index2 = (IndexWithSharedGroup) indexManager.getIndexByName("grouped_index_v");
+ IndexWithSharedGroup.Group group = indexManager.listIndexGroups()
+ .stream()
+ .filter(g -> g instanceof IndexWithSharedGroup.Group)
+ .map(g -> (IndexWithSharedGroup.Group) g)
+ .findAny()
+ .orElseThrow(AssertionError::new);
+
+ // verify that row insertions get to the index group and they are propagated to their members
+ execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 0, 0);
+ execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 1, 1);
+ execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 1, 0, 3);
+ assertEquals(3, group.rowsInserted.get());
+ assertEquals(3, index1.rowsInserted.size());
+ assertEquals(3, index2.rowsInserted.size());
+
+ // verify that row updates get to the index group and they are propagated to their members
+ execute("UPDATE %s SET v=? WHERE k=? AND c=?", 10, 0, 0);
+ execute("UPDATE %s SET v=? WHERE k=? AND c=?", 10, 1, 0);
+ assertEquals(2, group.rowsUpdated.get());
+ assertEquals(2, index1.rowsUpdated.size());
+ assertEquals(2, index2.rowsUpdated.size());
+
+ // verify that partition deletions get to the index group and its members
+ ReadCommand cmd = Util.cmd(cfs, 0).build();
+ try (ReadExecutionController executionController = cmd.executionController();
+ UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController))
+ {
+ assertTrue(iterator.hasNext());
+ cfs.indexManager.deletePartition(iterator.next(), FBUtilities.nowInSeconds());
+ }
+ assertEquals(1, group.partitionDeletions.get());
+ assertEquals(1, index1.partitionDeletions.size());
+ assertEquals(1, index2.partitionDeletions.size());
+
+ // verify that the row deletions produced by the previous partition deletion get to the group and its members
+ assertEquals(2, group.rowsDeleted.get());
+ assertEquals(2, index1.rowsDeleted.size());
+ assertEquals(2, index2.rowsDeleted.size());
+
+ // verify that range tombstones get to the index group and its members
+ execute("DELETE FROM %s WHERE k=? AND c>?", 0, 0);
+ execute("DELETE FROM %s WHERE k=? AND c>?", 1, 1);
+ assertEquals(2, group.rangeTombstones.get());
+ assertEquals(2, index1.rangeTombstones.size());
+ assertEquals(2, index2.rangeTombstones.size());
+
+ // verify the total number of begin calls
+ assertEquals(10, group.beginCalls.get());
+ assertEquals(10, index1.beginCalls);
+ assertEquals(10, index2.beginCalls);
+
+ // verify the total number of finish calls
+ assertEquals(10, group.finishCalls.get());
+ assertEquals(10, index1.finishCalls);
+ assertEquals(10, index2.finishCalls);
+
+ // flush the previous data to get rid of it, reset the group counters and flush a new memtable
+ cfs.forceBlockingFlush(FlushReason.UNIT_TESTS);
+ group.reset();
+ execute("INSERT INTO %s (k, s) VALUES (?, ?)", 1, 0);
+ execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 1, 0, 0);
+ execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 1, 1, 0);
+ execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 2, 0, 0);
+ execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 2, 1, 0);
+ execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 2, 2, 0);
+ execute("DELETE FROM %s WHERE k=? AND c=?", 2, 3);
+ execute("DELETE FROM %s WHERE k=?", 3);
+ cfs.forceBlockingFlush(FlushReason.UNIT_TESTS);
+
+ // verify that the flush observer calls get only once to the group
+ assertEquals(1, group.beginFlushCalls.get());
+ assertEquals(3, group.flushedPartitions.get());
+ assertEquals(3, group.flushedStaticRows.get());
+ assertEquals(6, group.flushedUnfiltereds.get());
+ assertEquals(1, group.completeFlushCalls.get());
+
+ // verify that the index rebuilds can be directed only to the first index
+ group.reset();
+ indexManager.rebuildIndexesBlocking(Collections.singleton(index1.getIndexMetadata().name));
+ assertEquals(8, group.rowsInserted.get());
+ assertEquals(8, index1.rowsInserted.size());
+ assertEquals(0, index2.rowsInserted.size());
+
+ // verify that the index rebuilds can be directed only to the second index
+ group.reset();
+ indexManager.rebuildIndexesBlocking(Collections.singleton(index2.getIndexMetadata().name));
+ assertEquals(8, group.rowsInserted.get());
+ assertEquals(0, index1.rowsInserted.size());
+ assertEquals(8, index2.rowsInserted.size());
+ }
+
+ @Test
+ public void testIndexGroupsInstancesManagement() throws Throwable
+ {
+ String indexClassName = IndexWithSharedGroup.class.getName();
+ createTable("CREATE TABLE %s (k int PRIMARY KEY, v1 int, v2 int, v3 int, v4 int, v5 int)");
+ SecondaryIndexManager indexManager = getCurrentColumnFamilyStore().indexManager;
+
+ // create two indexes belonging to the same group and verify that only one group is added to the manager
+ String idx1 = createIndex(String.format("CREATE CUSTOM INDEX ON %%s(v1) USING '%s'", indexClassName));
+ String idx2 = createIndex(String.format("CREATE CUSTOM INDEX ON %%s(v2) USING '%s'", indexClassName));
+ IndexWithSharedGroup.Group group = indexManager.listIndexGroups()
+ .stream()
+ .filter(g -> g instanceof IndexWithSharedGroup.Group)
+ .map(g -> (IndexWithSharedGroup.Group) g)
+ .findAny()
+ .orElseThrow(AssertionError::new);
+
+ // verify that only one group has been added to the manager
+ assertEquals(2, indexManager.listIndexes().size());
+ assertEquals(1, indexManager.listIndexGroups().size());
+ assertEquals(2, group.indexes.size());
+
+ // create two indexes belonging to their own singleton group and verify that two groups are added to the manager
+ String idx3 = createIndex("CREATE INDEX ON %s(v3)");
+ String idx4 = createIndex("CREATE INDEX ON %s(v4)");
+ assertEquals(4, indexManager.listIndexes().size());
+ assertEquals(3, indexManager.listIndexGroups().size());
+
+ // create another index to the shared group and verify that they are added to the existing group instance
+ String idx5 = createIndex(String.format("CREATE CUSTOM INDEX ON %%s(v5) USING '%s'", indexClassName));
+ assertEquals(5, indexManager.listIndexes().size());
+ assertEquals(3, indexManager.listIndexGroups().size());
+ assertEquals(3, group.indexes.size());
+
+ // drop one of the shared group members and verify that the manager still has the same group count
+ dropIndex("DROP INDEX %s." + idx1);
+ assertEquals(4, indexManager.listIndexes().size());
+ assertEquals(3, indexManager.listIndexGroups().size());
+ assertEquals(2, group.indexes.size());
+
+ // drop the standalone indexes and verify that their singleton groups are removed from the manager
+ dropIndex("DROP INDEX %s." + idx3);
+ dropIndex("DROP INDEX %s." + idx4);
+ assertEquals(2, indexManager.listIndexes().size());
+ assertEquals(1, indexManager.listIndexGroups().size());
+
+ // drop the remaining members of the shared group and verify that it is kept empty in the manager
+ dropIndex("DROP INDEX %s." + idx2);
+ dropIndex("DROP INDEX %s." + idx5);
+ assertEquals(0, indexManager.listIndexes().size());
+ assertEquals(1, indexManager.listIndexGroups().size());
+ assertEquals(0, group.indexes.size());
+
+ // create the sharing group members again and verify that they are added to the existing group instance
+ createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v1) USING '%s'", idx1, indexClassName));
+ createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v2) USING '%s'", idx2, indexClassName));
+ createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v3) USING '%s'", idx3, indexClassName));
+ assertEquals(3, indexManager.listIndexes().size());
+ assertEquals(1, indexManager.listIndexGroups().size());
+ assertEquals(3, group.indexes.size());
+ }
+
+ /**
+ * {@link StubIndex} implementation that uses the same {@link Index.Group} for all its instances.
+ * That group keeps count of the calls and passes them to its members.
+ */
+ public static final class IndexWithSharedGroup extends StubIndex
+ {
+ public IndexWithSharedGroup(ColumnFamilyStore baseCfs, IndexMetadata metadata)
+ {
+ super(baseCfs, metadata);
+ }
+
+ @Override
+ public boolean shouldBuildBlocking()
+ {
+ return true;
+ }
+
+ @Override
+ public void register(IndexRegistry registry)
+ {
+ registry.registerIndex(this, Group.class, Group::new);
+ }
+
+ private static class Group implements Index.Group
+ {
+ Map<String, IndexWithSharedGroup> indexes = Maps.newConcurrentMap();
+
+ AtomicInteger beginCalls = new AtomicInteger();
+ AtomicInteger finishCalls = new AtomicInteger();
+ AtomicInteger partitionDeletions = new AtomicInteger();
+ AtomicInteger rangeTombstones = new AtomicInteger();
+ AtomicInteger rowsInserted = new AtomicInteger();
+ AtomicInteger rowsDeleted = new AtomicInteger();
+ AtomicInteger rowsUpdated = new AtomicInteger();
+
+ AtomicInteger beginFlushCalls = new AtomicInteger();
+ AtomicInteger flushedPartitions = new AtomicInteger();
+ AtomicInteger flushedStaticRows = new AtomicInteger();
+ AtomicInteger flushedUnfiltereds = new AtomicInteger();
+ AtomicInteger completeFlushCalls = new AtomicInteger();
+
+ public void reset()
+ {
+ beginCalls.set(0);
+ finishCalls.set(0);
+ partitionDeletions.set(0);
+ rangeTombstones.set(0);
+ rowsInserted.set(0);
+ rowsDeleted.set(0);
+ rowsUpdated.set(0);
+ beginFlushCalls.set(0);
+ flushedPartitions.set(0);
+ flushedStaticRows.set(0);
+ flushedUnfiltereds.set(0);
+ completeFlushCalls.set(0);
+ indexes.values().forEach(IndexWithSharedGroup::reset);
+ }
+
+ @Override
+ public Set<Index> getIndexes()
+ {
+ return ImmutableSet.copyOf(indexes.values());
+ }
+
+ @Override
+ public void addIndex(Index index)
+ {
+ indexes.put(index.getIndexMetadata().name, (IndexWithSharedGroup) index);
+ }
+
+ @Override
+ public void removeIndex(Index index)
+ {
+ indexes.remove(index.getIndexMetadata().name);
+ }
+
+ @Override
+ public boolean containsIndex(Index index)
+ {
+ return indexes.containsKey(index.getIndexMetadata().name);
+ }
+
+ @Override
+ public Index.Indexer indexerFor(Predicate<Index> indexSelector,
+ DecoratedKey key,
+ RegularAndStaticColumns columns,
+ long nowInSec,
+ WriteContext context,
+ IndexTransaction.Type transactionType,
+ Memtable memtable)
+ {
+ Set<Index.Indexer> indexers = indexes.values()
+ .stream()
+ .filter(indexSelector)
+ .map(i -> i.indexerFor(key, columns, nowInSec, context, transactionType, memtable))
+ .filter(Objects::nonNull)
+ .collect(Collectors.toSet());
+
+ return indexers.isEmpty() ? null : new Index.Indexer() {
+
+ @Override
+ public void begin()
+ {
+ beginCalls.incrementAndGet();
+ indexers.forEach(Indexer::begin);
+ }
+
+ @Override
+ public void partitionDelete(DeletionTime deletionTime)
+ {
+ partitionDeletions.incrementAndGet();
+ indexers.forEach(indexer -> indexer.partitionDelete(deletionTime));
+ }
+
+ @Override
+ public void rangeTombstone(RangeTombstone tombstone)
+ {
+ rangeTombstones.incrementAndGet();
+ indexers.forEach(indexer -> indexer.rangeTombstone(tombstone));
+ }
+
+ @Override
+ public void insertRow(Row row)
+ {
+ rowsInserted.incrementAndGet();
+ indexers.forEach(indexer -> indexer.insertRow(row));
+ }
+
+ @Override
+ public void removeRow(Row row)
+ {
+ rowsDeleted.incrementAndGet();
+ indexers.forEach(indexer -> indexer.removeRow(row));
+ }
+
+ @Override
+ public void updateRow(Row oldRow, Row newRow)
+ {
+ rowsUpdated.incrementAndGet();
+ indexers.forEach(indexer -> indexer.updateRow(oldRow, newRow));
+ }
+
+ @Override
+ public void finish()
+ {
+ finishCalls.incrementAndGet();
+ indexers.forEach(Indexer::finish);
+ }
+ };
+ }
+
+ @Override
+ public QueryPlan queryPlanFor(RowFilter rowFilter)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker, TableMetadata tableMetadata)
+ {
+ Set<SSTableFlushObserver> observers = indexes.values()
+ .stream()
+ .map(i -> i.getFlushObserver(descriptor, tracker))
+ .filter(Objects::nonNull)
+ .collect(Collectors.toSet());
+
+ return new SSTableFlushObserver() {
+
+ @Override
+ public void begin()
+ {
+ beginFlushCalls.incrementAndGet();
+ observers.forEach(SSTableFlushObserver::begin);
+ }
+
+ @Override
+ public void startPartition(DecoratedKey key, long position, long keyPositionForSASI)
+ {
+ flushedPartitions.incrementAndGet();
+ observers.forEach(o -> o.startPartition(key, position, keyPositionForSASI));
+ }
+
+ @Override
+ public void staticRow(Row staticRow)
+ {
+ flushedStaticRows.incrementAndGet();
+ observers.forEach(o -> o.staticRow(staticRow));
+ }
+
+ @Override
+ public void nextUnfilteredCluster(Unfiltered unfiltered)
+ {
+ flushedUnfiltereds.incrementAndGet();
+ observers.forEach(o -> o.nextUnfilteredCluster(unfiltered));
+ }
+
+ @Override
+ public void complete()
+ {
+ completeFlushCalls.incrementAndGet();
+ observers.forEach(SSTableFlushObserver::complete);
+ }
+ };
+ }
+
+ @Override
+ public Set<Component> getComponents()
+ {
+ return Collections.emptySet();
+ }
+ }
+ }
+
+ @Test
+ public void testMulticolumnIndexWithBaseTable() throws Throwable
+ {
+ createTable("CREATE TABLE %s(k int PRIMARY KEY, v int)");
+ assertInvalidMessage("Indexes belonging to a group of indexes shouldn't have a backing table",
+ String.format("CREATE CUSTOM INDEX ON %%s(v) USING '%s'",
+ MulticolumnIndexWithBaseTable.class.getName()));
+ }
+
+ public static final class MulticolumnIndexWithBaseTable extends StubIndex
+ {
+ private final ColumnFamilyStore baseCfs;
+
+ public MulticolumnIndexWithBaseTable(ColumnFamilyStore baseCfs, IndexMetadata metadata)
+ {
+ super(baseCfs, metadata);
+ this.baseCfs = baseCfs;
+ }
+
+ @Override
+ public void register(IndexRegistry registry)
+ {
+ registry.registerIndex(this, MulticolumnIndexWithBaseTable.class, StubIndexGroup::new);
+ }
+
+ @Override
+ public Optional<ColumnFamilyStore> getBackingTable()
+ {
+ return Optional.of(baseCfs);
+ }
+ }
}
diff --git a/test/unit/org/apache/cassandra/index/SecondaryIndexManagerTest.java b/test/unit/org/apache/cassandra/index/SecondaryIndexManagerTest.java
index d8fb99f40f..c9dde2ae69 100644
--- a/test/unit/org/apache/cassandra/index/SecondaryIndexManagerTest.java
+++ b/test/unit/org/apache/cassandra/index/SecondaryIndexManagerTest.java
@@ -766,7 +766,7 @@ public class SecondaryIndexManagerTest extends CQLTester
{
return new CollatedViewIndexBuildingSupport()
{
- public SecondaryIndexBuilder getIndexBuildTask(ColumnFamilyStore cfs, Set<Index> indexes, Collection<SSTableReader> sstables)
+ public SecondaryIndexBuilder getIndexBuildTask(ColumnFamilyStore cfs, Set<Index> indexes, Collection<SSTableReader> sstables, boolean isFullRebuild)
{
try
{
@@ -775,7 +775,7 @@ public class SecondaryIndexManagerTest extends CQLTester
buildWaitLatch.countDown();
buildLatch.await();
}
- final SecondaryIndexBuilder builder = super.getIndexBuildTask(cfs, indexes, sstables);
+ final SecondaryIndexBuilder builder = super.getIndexBuildTask(cfs, indexes, sstables, isFullRebuild);
return new SecondaryIndexBuilder()
{
diff --git a/test/unit/org/apache/cassandra/index/StubIndex.java b/test/unit/org/apache/cassandra/index/StubIndex.java
index f8a6ae6b65..d3366c7766 100644
--- a/test/unit/org/apache/cassandra/index/StubIndex.java
+++ b/test/unit/org/apache/cassandra/index/StubIndex.java
@@ -20,16 +20,16 @@ package org.apache.cassandra.index;
import java.util.*;
import java.util.concurrent.Callable;
-import java.util.function.BiFunction;
import org.apache.cassandra.Util;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.UTF8Type;
-import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.memtable.Memtable;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.exceptions.InvalidRequestException;
@@ -100,7 +100,8 @@ public class StubIndex implements Index
RegularAndStaticColumns columns,
long nowInSec,
WriteContext ctx,
- IndexTransaction.Type transactionType)
+ IndexTransaction.Type transactionType,
+ Memtable memtable)
{
return new Indexer()
{
@@ -205,11 +206,28 @@ public class StubIndex implements Index
public Searcher searcherFor(final ReadCommand command)
{
- return (controller) -> Util.executeLocally((PartitionRangeReadCommand)command, baseCfs, controller);
+ return new Searcher(command);
}
- public BiFunction<PartitionIterator, ReadCommand, PartitionIterator> postProcessorFor(ReadCommand readCommand)
+ protected class Searcher implements Index.Searcher
{
- return (iter, command) -> iter;
+ private final ReadCommand command;
+
+ Searcher(ReadCommand command)
+ {
+ this.command = command;
+ }
+
+ @Override
+ public ReadCommand command()
+ {
+ return command;
+ }
+
+ @Override
+ public UnfilteredPartitionIterator search(ReadExecutionController executionController)
+ {
+ return Util.executeLocally((PartitionRangeReadCommand)command, baseCfs, executionController);
+ }
}
}
diff --git a/test/unit/org/apache/cassandra/index/StubIndexGroup.java b/test/unit/org/apache/cassandra/index/StubIndexGroup.java
new file mode 100644
index 0000000000..0b6ad1c37d
--- /dev/null
+++ b/test/unit/org/apache/cassandra/index/StubIndexGroup.java
@@ -0,0 +1,99 @@
+/*
+ * 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.index;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.function.Predicate;
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.RegularAndStaticColumns;
+import org.apache.cassandra.db.WriteContext;
+import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
+import org.apache.cassandra.db.memtable.Memtable;
+import org.apache.cassandra.index.transactions.IndexTransaction;
+import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.SSTableFlushObserver;
+import org.apache.cassandra.schema.TableMetadata;
+
+/**
+ * Basic custom index group implementation for testing.
+ */
+public class StubIndexGroup implements Index.Group
+{
+ private final Set<Index> indexes = new HashSet<>();
+
+ @Override
+ public Set<Index> getIndexes()
+ {
+ return indexes;
+ }
+
+ @Override
+ public void addIndex(Index index)
+ {
+ indexes.add(index);
+ }
+
+ @Override
+ public void removeIndex(Index index)
+ {
+ indexes.remove(index);
+ }
+
+ @Override
+ public boolean containsIndex(Index index)
+ {
+ return indexes.contains(index);
+ }
+
+ @Override
+ public Index.Indexer indexerFor(Predicate<Index> indexSelector,
+ DecoratedKey key,
+ RegularAndStaticColumns columns,
+ long nowInSec,
+ WriteContext context,
+ IndexTransaction.Type transactionType,
+ Memtable memtable)
+ {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public Index.QueryPlan queryPlanFor(RowFilter rowFilter)
+ {
+ return null;
+ }
+
+ @Override
+ public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker, TableMetadata tableMetadata)
+ {
+ return null;
+ }
+
+ public Set<Component> getComponents()
+ {
+ return Collections.emptySet();
+ }
+}
\ No newline at end of file
diff --git a/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java b/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
index 9640af1357..f67f2666b9 100644
--- a/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
+++ b/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
@@ -24,7 +24,6 @@ import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
-import java.util.function.BiFunction;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
@@ -46,7 +45,7 @@ import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.memtable.Memtable;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.exceptions.InvalidRequestException;
@@ -54,7 +53,6 @@ import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.IndexRegistry;
import org.apache.cassandra.index.SecondaryIndexBuilder;
import org.apache.cassandra.index.transactions.IndexTransaction;
-import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.io.sstable.ReducingKeyIterator;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.IndexMetadata;
@@ -210,14 +208,6 @@ public class CustomCassandraIndex implements Index
return indexCfs.getMeanEstimatedCellPerPartitionCount();
}
- /**
- * No post processing of query results, just return them unchanged
- */
- public BiFunction<PartitionIterator, ReadCommand, PartitionIterator> postProcessorFor(ReadCommand command)
- {
- return (partitionIterator, readCommand) -> partitionIterator;
- }
-
public RowFilter getPostIndexQueryFilter(RowFilter filter)
{
return getTargetExpression(filter.getExpressions()).map(filter::without)
@@ -290,7 +280,8 @@ public class CustomCassandraIndex implements Index
final RegularAndStaticColumns columns,
final long nowInSec,
final WriteContext ctx,
- final IndexTransaction.Type transactionType)
+ final IndexTransaction.Type transactionType,
+ final Memtable memtable)
{
if (!isPrimaryKeyIndex() && !columns.contains(indexedColumn))
return null;
@@ -462,7 +453,7 @@ public class CustomCassandraIndex implements Index
cell));
Row row = BTreeRow.noCellLiveRow(buildIndexClustering(rowKey, clustering, cell), info);
PartitionUpdate upd = partitionUpdate(valueKey, row);
- indexCfs.getWriteHandler().write(upd, ctx, UpdateTransaction.NO_OP);
+ indexCfs.getWriteHandler().write(upd, ctx, false);
logger.debug("Inserted entry into index for value {}", valueKey);
}
@@ -508,7 +499,7 @@ public class CustomCassandraIndex implements Index
{
Row row = BTreeRow.emptyDeletedRow(indexClustering, Row.Deletion.regular(deletion));
PartitionUpdate upd = partitionUpdate(indexKey, row);
- indexCfs.getWriteHandler().write(upd, ctx, UpdateTransaction.NO_OP);
+ indexCfs.getWriteHandler().write(upd, ctx, false);
logger.debug("Removed index entry for value {}", indexKey);
}
diff --git a/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java b/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java
index 3c33327c97..fd13659535 100644
--- a/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java
@@ -110,7 +110,7 @@ import org.apache.cassandra.index.sasi.disk.Token;
import org.apache.cassandra.index.sasi.exceptions.TimeQuotaExceededException;
import org.apache.cassandra.index.sasi.memory.IndexMemtable;
import org.apache.cassandra.index.sasi.plan.QueryController;
-import org.apache.cassandra.index.sasi.plan.QueryPlan;
+import org.apache.cassandra.index.sasi.plan.SASIIndexSearcher;
import org.apache.cassandra.index.sasi.utils.RangeIterator;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
@@ -1506,7 +1506,7 @@ public class SASIIndexTest
DataRange.allData(store.metadata().partitioner));
try
{
- new QueryPlan(store, command, 0).execute(ReadExecutionController.empty());
+ new SASIIndexSearcher(store, command, 0).search(ReadExecutionController.empty());
Assert.fail();
}
catch (TimeQuotaExceededException e)
@@ -1523,7 +1523,7 @@ public class SASIIndexTest
try (ReadExecutionController controller = command.executionController())
{
- Set<String> rows = getKeys(new QueryPlan(store, command, DatabaseDescriptor.getRangeRpcTimeout(MILLISECONDS)).execute(controller));
+ Set<String> rows = getKeys(new SASIIndexSearcher(store, command, DatabaseDescriptor.getRangeRpcTimeout(MILLISECONDS)).search(controller));
assertRows(rows, "key1", "key2", "key3", "key4");
}
}
diff --git a/test/unit/org/apache/cassandra/index/sasi/disk/PerSSTableIndexWriterTest.java b/test/unit/org/apache/cassandra/index/sasi/disk/PerSSTableIndexWriterTest.java
index 6707aa592e..02e87b374c 100644
--- a/test/unit/org/apache/cassandra/index/sasi/disk/PerSSTableIndexWriterTest.java
+++ b/test/unit/org/apache/cassandra/index/sasi/disk/PerSSTableIndexWriterTest.java
@@ -41,6 +41,7 @@ import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.compaction.OperationType;
+import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.UTF8Type;
@@ -93,7 +94,7 @@ public class PerSSTableIndexWriterTest extends SchemaLoader
File directory = cfs.getDirectories().getDirectoryForNewSSTables();
Descriptor descriptor = cfs.newSSTableDescriptor(directory);
- PerSSTableIndexWriter indexWriter = (PerSSTableIndexWriter) sasi.getFlushObserver(descriptor, OperationType.FLUSH);
+ PerSSTableIndexWriter indexWriter = (PerSSTableIndexWriter) sasi.getFlushObserver(descriptor, LifecycleTransaction.offline(OperationType.FLUSH));
SortedMap<DecoratedKey, Row> expectedKeys = new TreeMap<>(DecoratedKey.comparator);
@@ -187,7 +188,7 @@ public class PerSSTableIndexWriterTest extends SchemaLoader
File directory = cfs.getDirectories().getDirectoryForNewSSTables();
Descriptor descriptor = cfs.newSSTableDescriptor(directory);
- PerSSTableIndexWriter indexWriter = (PerSSTableIndexWriter) sasi.getFlushObserver(descriptor, OperationType.FLUSH);
+ PerSSTableIndexWriter indexWriter = (PerSSTableIndexWriter) sasi.getFlushObserver(descriptor, LifecycleTransaction.offline(OperationType.FLUSH));
final long now = System.currentTimeMillis();
diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableFlushObserverTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableFlushObserverTest.java
index 58ca05ba0f..fbeabb95cb 100644
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableFlushObserverTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableFlushObserverTest.java
@@ -107,10 +107,10 @@ public class SSTableFlushObserverTest
.setMetadataCollector(new MetadataCollector(cfm.comparator).sstableLevel(0))
.setSerializationHeader(new SerializationHeader(true, cfm, cfm.regularAndStaticColumns(), EncodingStats.NO_STATS))
.setFlushObservers(Collections.singletonList(observer))
- .addDefaultComponents()
+ .addDefaultComponents(Collections.emptySet())
.build(transaction, null);
- SSTableReader reader = null;
+ SSTableReader reader;
Multimap<ByteBuffer, Cell<?>> expected = ArrayListMultimap.create();
try
@@ -151,7 +151,6 @@ public class SSTableFlushObserverTest
for (Triple<ByteBuffer, Long, Long> e : observer.rows.keySet())
{
ByteBuffer key = e.getLeft();
- long dataPosition = e.getMiddle();
long indexPosition = e.getRight();
DecoratedKey indexKey = reader.keyAtPositionFromSecondaryIndex(indexPosition);
diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java
index f5fe75eb9c..48ea2c6d20 100644
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java
@@ -1060,7 +1060,7 @@ public class SSTableReaderTest
.columns("birthdate")
.filterOn("birthdate", Operator.EQ, 1L)
.build();
- Index.Searcher searcher = rc.getIndex(indexedCFS).searcherFor(rc);
+ Index.Searcher searcher = rc.indexSearcher();
assertNotNull(searcher);
try (ReadExecutionController executionController = rc.executionController())
{
diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTestBase.java b/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTestBase.java
index 91b44ed27e..0cdd8ac18b 100644
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTestBase.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTestBase.java
@@ -163,9 +163,9 @@ public class SSTableWriterTestBase extends SchemaLoader
.setPendingRepair(pendingRepair)
.setTransientSSTable(isTransient)
.setSerializationHeader(new SerializationHeader(true, cfs.metadata(), cfs.metadata().regularAndStaticColumns(), EncodingStats.NO_STATS))
- .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexes(), txn.opType())
+ .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get())
.setMetadataCollector(new MetadataCollector(cfs.metadata().comparator))
- .addDefaultComponents()
+ .addDefaultComponents(cfs.indexManager.listIndexGroups())
.build(txn, cfs);
}
diff --git a/test/unit/org/apache/cassandra/io/sstable/ScrubTest.java b/test/unit/org/apache/cassandra/io/sstable/ScrubTest.java
index c756d9a81b..bd5b4a5c18 100644
--- a/test/unit/org/apache/cassandra/io/sstable/ScrubTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/ScrubTest.java
@@ -840,7 +840,7 @@ public class ScrubTest
.setMetadataCollector(collector)
.setSerializationHeader(header)
.setFlushObservers(Collections.emptyList())
- .addDefaultComponents()
+ .addDefaultComponents(Collections.emptySet())
.build(txn, cfs);
return new TestMultiWriter(writer, txn);
diff --git a/test/unit/org/apache/cassandra/metrics/ClientRequestMetricsTest.java b/test/unit/org/apache/cassandra/metrics/ClientRequestMetricsTest.java
index 650bd95637..b4e9b085d2 100644
--- a/test/unit/org/apache/cassandra/metrics/ClientRequestMetricsTest.java
+++ b/test/unit/org/apache/cassandra/metrics/ClientRequestMetricsTest.java
@@ -24,6 +24,7 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+import com.codahale.metrics.Histogram;
import com.datastax.driver.core.BatchStatement;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.Cluster;
@@ -33,10 +34,13 @@ import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.service.EmbeddedCassandraService;
+import org.apache.cassandra.service.reads.range.RangeCommandIterator;
-import static com.datastax.driver.core.Cluster.builder;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
+import static com.datastax.driver.core.Cluster.builder;
+
public class ClientRequestMetricsTest
{
private static Cluster cluster;
@@ -161,6 +165,23 @@ public class ClientRequestMetricsTest
assertEquals(0, writeMetricsContainer.compareRemoteRequest());
}
+ @Test
+ public void testRangeRead() throws Throwable
+ {
+ clearHistogram(RangeCommandIterator.rangeMetrics.roundTrips);
+ long latencyCount = RangeCommandIterator.rangeMetrics.latency.getCount();
+
+ session.execute("SELECT * FROM system.peers");
+
+ assertThat(RangeCommandIterator.rangeMetrics.roundTrips.getCount()).isGreaterThan(0);
+ assertThat(RangeCommandIterator.rangeMetrics.roundTrips.getSnapshot().getMax()).isEqualTo(1);
+ assertThat(RangeCommandIterator.rangeMetrics.latency.getCount()).isEqualTo(latencyCount + 1);
+ }
+
+ private void clearHistogram(Histogram histogram)
+ {
+ ((ClearableHistogram) histogram).clear();
+ }
private static class ClientRequestMetricsContainer
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org