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

Merge branch 'cassandra-5.0' into trunk

This commit is contained in:
Stefan Miklosovic 2024-01-22 12:44:59 +01:00
parent aa644c9dfa 016dd6ca37
commit e81b2f54b4
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
25 changed files with 1371 additions and 647 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -8,6 +8,7 @@
* Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
* Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
Merged from 5.0:
* Make CQLSSTableWriter to support building of SAI indexes (CASSANDRA-18714)
* Append additional JVM options when using JDK17+ (CASSANDRA-19001)
* Upgrade Python driver to 3.29.0 (CASSANDRA-19245)
* Creating a SASI index after creating an SAI index does not break secondary index queries (CASSANDRA-18939)

18
src/java/org/apache/cassandra/config/Config.java
View File

@ -39,6 +39,7 @@ import org.apache.cassandra.audit.AuditLogOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.fql.FullQueryLoggerOptions;
import org.apache.cassandra.index.internal.CassandraIndex;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.sstable.format.big.BigFormat;
import org.apache.cassandra.service.StartupChecks.StartupCheckType;
import org.apache.cassandra.utils.StorageCompatibilityMode;
@ -1174,7 +1175,22 @@ public class Config
unslabbed_heap_buffers_logged,
heap_buffers,
offheap_buffers,
offheap_objects
offheap_objects;
public BufferType toBufferType()
{
switch (this)
{
case unslabbed_heap_buffers:
case heap_buffers:
return BufferType.ON_HEAP;
case offheap_buffers:
case offheap_objects:
return BufferType.OFF_HEAP;
default:
throw new AssertionError();
}
}
}
public enum DiskFailurePolicy

88
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -202,22 +202,24 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
are finished. By having flushExecutor size the same size as each of the perDiskflushExecutors we make sure we can
have that many flushes going at the same time.
*/
private static final ExecutorPlus flushExecutor = executorFactory()
.withJmxInternal()
.pooled("MemtableFlushWriter", getFlushWriters());
private static final ExecutorPlus flushExecutor = DatabaseDescriptor.isDaemonInitialized()
? executorFactory().withJmxInternal().pooled("MemtableFlushWriter", getFlushWriters())
: null;
// post-flush executor is single threaded to provide guarantee that any flush Future on a CF will never return until prior flushes have completed
private static final ExecutorPlus postFlushExecutor = executorFactory()
.withJmxInternal()
.sequential("MemtablePostFlush");
private static final ExecutorPlus postFlushExecutor = DatabaseDescriptor.isDaemonInitialized()
? executorFactory().withJmxInternal().sequential("MemtablePostFlush")
: null;
private static final ExecutorPlus reclaimExecutor = executorFactory()
.withJmxInternal()
.sequential("MemtableReclaimMemory");
private static final ExecutorPlus reclaimExecutor = DatabaseDescriptor.isDaemonInitialized()
? executorFactory().withJmxInternal().sequential("MemtableReclaimMemory")
: null;
private static final PerDiskFlushExecutors perDiskflushExecutors = new PerDiskFlushExecutors(DatabaseDescriptor.getFlushWriters(),
DatabaseDescriptor.getNonLocalSystemKeyspacesDataFileLocations(),
DatabaseDescriptor.useSpecificLocationForLocalSystemData());
private static final PerDiskFlushExecutors perDiskflushExecutors = DatabaseDescriptor.isDaemonInitialized()
? new PerDiskFlushExecutors(DatabaseDescriptor.getFlushWriters(),
DatabaseDescriptor.getNonLocalSystemKeyspacesDataFileLocations(),
DatabaseDescriptor.useSpecificLocationForLocalSystemData())
: null;
/**
* Reason for initiating a memtable flush.
@ -402,7 +404,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
indexManager.reload(tableMetadata);
memtableFactory = tableMetadata.params.memtable.factory();
switchMemtableOrNotify(FlushReason.SCHEMA_CHANGE, tableMetadata, Memtable::metadataUpdated);
if (DatabaseDescriptor.isDaemonInitialized())
switchMemtableOrNotify(FlushReason.SCHEMA_CHANGE, tableMetadata, Memtable::metadataUpdated);
}
public static Runnable getBackgroundCompactionTaskSubmitter()
@ -863,26 +866,53 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
sstableImporter.importNewSSTables(options);
}
/**
* #{@inheritDoc}
*/
public synchronized List<String> importNewSSTables(Set<String> srcPaths, boolean resetLevel, boolean clearRepaired, boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches, boolean extendedVerify, boolean copyData)
@Override
public List<String> importNewSSTables(Set<String> srcPaths, boolean resetLevel, boolean clearRepaired,
boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches,
boolean extendedVerify, boolean copyData)
{
SSTableImporter.Options options = SSTableImporter.Options.options(srcPaths)
.resetLevel(resetLevel)
.clearRepaired(clearRepaired)
.verifySSTables(verifySSTables)
.verifyTokens(verifyTokens)
.invalidateCaches(invalidateCaches)
.extendedVerify(extendedVerify)
.copyData(copyData).build();
return sstableImporter.importNewSSTables(options);
return sstableImporter.importNewSSTables(SSTableImporter.Options.options(srcPaths)
.resetLevel(resetLevel)
.clearRepaired(clearRepaired)
.verifySSTables(verifySSTables)
.verifyTokens(verifyTokens)
.invalidateCaches(invalidateCaches)
.extendedVerify(extendedVerify)
.copyData(copyData).build());
}
public List<String> importNewSSTables(Set<String> srcPaths, boolean resetLevel, boolean clearRepaired, boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches, boolean extendedVerify)
@Override
public List<String> importNewSSTables(Set<String> srcPaths, boolean resetLevel, boolean clearRepaired,
boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches,
boolean extendedVerify)
{
return importNewSSTables(srcPaths, resetLevel, clearRepaired, verifySSTables, verifyTokens, invalidateCaches, extendedVerify, false);
return sstableImporter.importNewSSTables(SSTableImporter.Options.options(srcPaths)
.resetLevel(resetLevel)
.clearRepaired(clearRepaired)
.verifySSTables(verifySSTables)
.verifyTokens(verifyTokens)
.invalidateCaches(invalidateCaches)
.extendedVerify(extendedVerify)
.build());
}
@Override
public List<String> importNewSSTables(Set<String> srcPaths, boolean resetLevel, boolean clearRepaired,
boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches,
boolean extendedVerify, boolean copyData, boolean failOnMissingIndex,
boolean validateIndexChecksum)
{
return sstableImporter.importNewSSTables(SSTableImporter.Options.options(srcPaths)
.resetLevel(resetLevel)
.clearRepaired(clearRepaired)
.verifySSTables(verifySSTables)
.verifyTokens(verifyTokens)
.invalidateCaches(invalidateCaches)
.extendedVerify(extendedVerify)
.failOnMissingIndex(failOnMissingIndex)
.validateIndexChecksum(validateIndexChecksum)
.copyData(copyData)
.build());
}
Descriptor getUniqueDescriptorFor(Descriptor descriptor, File targetDirectory)

41
src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java
View File

@ -189,12 +189,12 @@ public interface ColumnFamilyStoreMBean
/** @deprecated See CASSANDRA-16407 */
@Deprecated(since = "4.0")
public List<String> importNewSSTables(Set<String> srcPaths,
boolean resetLevel,
boolean clearRepaired,
boolean verifySSTables,
boolean verifyTokens,
boolean invalidateCaches,
boolean extendedVerify);
boolean resetLevel,
boolean clearRepaired,
boolean verifySSTables,
boolean verifyTokens,
boolean invalidateCaches,
boolean extendedVerify);
/**
* Load new sstables from the given directory
@ -210,6 +210,8 @@ public interface ColumnFamilyStoreMBean
*
* @return list of failed import directories
*/
/** @deprecated See CASSANDRA-18714 */
@Deprecated(since = "5.0")
public List<String> importNewSSTables(Set<String> srcPaths,
boolean resetLevel,
boolean clearRepaired,
@ -219,6 +221,33 @@ public interface ColumnFamilyStoreMBean
boolean extendedVerify,
boolean copyData);
/**
* Load new sstables from the given directory
*
* @param srcPaths the path to the new sstables - if it is an empty set, the data directories will be scanned
* @param resetLevel if the level should be reset to 0 on the new sstables
* @param clearRepaired if repaired info should be wiped from the new sstables
* @param verifySSTables if the new sstables should be verified that they are not corrupt
* @param verifyTokens if the tokens in the new sstables should be verified that they are owned by the current node
* @param invalidateCaches if row cache should be invalidated for the keys in the new sstables
* @param extendedVerify if we should run an extended verify checking all values in the new sstables
* @param copyData if we should copy data from source paths instead of moving them
* @param validateIndexChecksum if we should also validate checksum for SAI indexes
* @param failOnMissingIndex if loading should fail when SSTables do not contain built SAI indexes too
*
* @return list of failed import directories
*/
public List<String> importNewSSTables(Set<String> srcPaths,
boolean resetLevel,
boolean clearRepaired,
boolean verifySSTables,
boolean verifyTokens,
boolean invalidateCaches,
boolean extendedVerify,
boolean copyData,
boolean failOnMissingIndex,
boolean validateIndexChecksum);
/** @deprecated See CASSANDRA-6719 */
@Deprecated(since = "4.0")
public void loadNewSSTables();

72
src/java/org/apache/cassandra/db/SSTableImporter.java
View File

@ -32,6 +32,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.sai.StorageAttachedIndexGroup;
import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
import org.apache.cassandra.index.sai.utils.IndexIdentifier;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.IVerifier;
@ -83,7 +87,7 @@ public class SSTableImporter
List<String> failedDirectories = new ArrayList<>();
// verify first to avoid starting to copy sstables to the data directories and then have to abort.
if (options.verifySSTables || options.verifyTokens)
if (options.verifySSTables || options.verifyTokens || options.failOnMissingIndex)
{
for (Pair<Directories.SSTableLister, String> listerPair : listers)
{
@ -97,7 +101,39 @@ public class SSTableImporter
try
{
abortIfDraining();
verifySSTableForImport(descriptor, entry.getValue(), options.verifyTokens, options.verifySSTables, options.extendedVerify);
if (options.failOnMissingIndex)
{
Index.Group saiIndexGroup = cfs.indexManager.getIndexGroup(StorageAttachedIndexGroup.GROUP_KEY);
if (saiIndexGroup != null)
{
IndexDescriptor indexDescriptor = IndexDescriptor.create(descriptor,
cfs.getPartitioner(),
cfs.metadata().comparator);
String keyspace = cfs.getKeyspaceName();
String table = cfs.getTableName();
if (!indexDescriptor.isPerSSTableIndexBuildComplete())
throw new IllegalStateException(String.format("Missing SAI index to import for SSTable %s on %s.%s",
indexDescriptor.sstableDescriptor.toString(),
keyspace,
table));
for (Index index : saiIndexGroup.getIndexes())
{
IndexIdentifier indexIdentifier = new IndexIdentifier(keyspace, table, index.getIndexMetadata().name);
if (!indexDescriptor.isPerColumnIndexBuildComplete(indexIdentifier))
throw new IllegalStateException(String.format("Missing SAI index to import for index %s on %s.%s",
index.getIndexMetadata().name,
keyspace,
table));
}
}
}
if (options.verifySSTables || options.verifyTokens)
verifySSTableForImport(descriptor, entry.getValue(), options.verifyTokens, options.verifySSTables, options.extendedVerify);
}
catch (Throwable t)
{
@ -188,7 +224,7 @@ public class SSTableImporter
abortIfDraining();
// Validate existing SSTable-attached indexes, and then build any that are missing:
if (!cfs.indexManager.validateSSTableAttachedIndexes(newSSTables, false))
if (!cfs.indexManager.validateSSTableAttachedIndexes(newSSTables, false, options.validateIndexChecksum))
cfs.indexManager.buildSSTableAttachedIndexesBlocking(newSSTables);
cfs.getTracker().addSSTables(newSSTables);
@ -441,8 +477,13 @@ public class SSTableImporter
private final boolean invalidateCaches;
private final boolean extendedVerify;
private final boolean copyData;
private final boolean failOnMissingIndex;
public final boolean validateIndexChecksum;
public Options(Set<String> srcPaths, boolean resetLevel, boolean clearRepaired, boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches, boolean extendedVerify, boolean copyData)
public Options(Set<String> srcPaths, boolean resetLevel, boolean clearRepaired,
boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches,
boolean extendedVerify, boolean copyData, boolean failOnMissingIndex,
boolean validateIndexChecksum)
{
this.srcPaths = srcPaths;
this.resetLevel = resetLevel;
@ -452,6 +493,8 @@ public class SSTableImporter
this.invalidateCaches = invalidateCaches;
this.extendedVerify = extendedVerify;
this.copyData = copyData;
this.failOnMissingIndex = failOnMissingIndex;
this.validateIndexChecksum = validateIndexChecksum;
}
public static Builder options(String srcDir)
@ -481,6 +524,8 @@ public class SSTableImporter
", invalidateCaches=" + invalidateCaches +
", extendedVerify=" + extendedVerify +
", copyData= " + copyData +
", failOnMissingIndex= " + failOnMissingIndex +
", validateIndexChecksum= " + validateIndexChecksum +
'}';
}
@ -494,6 +539,8 @@ public class SSTableImporter
private boolean invalidateCaches = false;
private boolean extendedVerify = false;
private boolean copyData = false;
private boolean failOnMissingIndex = false;
private boolean validateIndexChecksum = true;
private Builder(Set<String> srcPath)
{
@ -543,9 +590,24 @@ public class SSTableImporter
return this;
}
public Builder failOnMissingIndex(boolean value)
{
failOnMissingIndex = value;
return this;
}
public Builder validateIndexChecksum(boolean value)
{
validateIndexChecksum = value;
return this;
}
public Options build()
{
return new Options(srcPaths, resetLevel, clearRepaired, verifySSTables, verifyTokens, invalidateCaches, extendedVerify, copyData);
return new Options(srcPaths, resetLevel, clearRepaired,
verifySSTables, verifyTokens, invalidateCaches,
extendedVerify, copyData, failOnMissingIndex,
validateIndexChecksum);
}
}
}

19
src/java/org/apache/cassandra/db/memtable/TrieMemtable.java
View File

@ -91,24 +91,7 @@ public class TrieMemtable extends AbstractShardedMemtable
private static final Logger logger = LoggerFactory.getLogger(TrieMemtable.class);
/** Buffer type to use for memtable tries (on- vs off-heap) */
public static final BufferType BUFFER_TYPE;
static
{
switch (DatabaseDescriptor.getMemtableAllocationType())
{
case unslabbed_heap_buffers:
case heap_buffers:
BUFFER_TYPE = BufferType.ON_HEAP;
break;
case offheap_buffers:
case offheap_objects:
BUFFER_TYPE = BufferType.OFF_HEAP;
break;
default:
throw new AssertionError();
}
}
public static final BufferType BUFFER_TYPE = DatabaseDescriptor.getMemtableAllocationType().toBufferType();
/** If keys is below this length, we will use a recursive procedure for inserting data in the memtable trie. */
@VisibleForTesting

2
src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
View File

@ -249,7 +249,7 @@ public class CassandraStreamReceiver implements StreamReceiver
// via the SSTable flush observer, and an error there would have aborted the streaming transaction.
if (receivedEntireSSTable)
// If we do validate, any exception thrown doing so will also abort the streaming transaction:
cfs.indexManager.validateSSTableAttachedIndexes(readers, true);
cfs.indexManager.validateSSTableAttachedIndexes(readers, true, true);
finishTransaction();

13
src/java/org/apache/cassandra/index/Index.java
View File

@ -849,18 +849,19 @@ public interface Index
Set<Component> getComponents();
/**
* Validates all indexes in the group against the specified SSTables.
*
* Validates all indexes in the group against the specified SSTables.
*
* @param sstables SSTables for which indexes in the group should be built
* @param throwOnIncomplete whether to throw an error if any index in the group is incomplete
*
* @param validateChecksum whether checksum will be tested as part of the validation
*
* @return true if all indexes in the group are complete and valid
* false if any index is incomplete and {@code throwOnIncomplete} is false
*
* false if any index is incomplete and {@code throwOnIncomplete} is false
*
* @throws IllegalStateException if {@code throwOnIncomplete} is true and any index in the group is incomplete
* @throws UncheckedIOException if there is a problem validating any on-disk component of an index in the group
*/
default boolean validateSSTableAttachedIndexes(Collection<SSTableReader> sstables, boolean throwOnIncomplete)
default boolean validateSSTableAttachedIndexes(Collection<SSTableReader> sstables, boolean throwOnIncomplete, boolean validateChecksum)
{
return true;
}

7
src/java/org/apache/cassandra/index/SecondaryIndexManager.java
View File

@ -239,7 +239,7 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
{
try
{
Callable<?> call = index.getInitializationTask();
Callable<?> call = DatabaseDescriptor.isDaemonInitialized() ? index.getInitializationTask() : null;
if (call != null)
initialBuildTask = new FutureTask<>(call);
}
@ -503,6 +503,7 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
*
* @param sstables SSTables for which indexes in the group should be built
* @param throwOnIncomplete whether to throw an error if any index in the group is incomplete
* @param validateChecksum whether to validate checksum or not
*
* @return true if all indexes in all groups are complete and valid
* false if an index in any group is incomplete and {@code throwOnIncomplete} is false
@ -510,14 +511,14 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
* @throws IllegalStateException if {@code throwOnIncomplete} is true and an index in any group is incomplete
* @throws UncheckedIOException if there is a problem validating any on-disk component in any group
*/
public boolean validateSSTableAttachedIndexes(Collection<SSTableReader> sstables, boolean throwOnIncomplete)
public boolean validateSSTableAttachedIndexes(Collection<SSTableReader> sstables, boolean throwOnIncomplete, boolean validateChecksum)
{
boolean complete = true;
for (Index.Group group : indexGroups.values())
{
if (group.getIndexes().stream().anyMatch(Index::isSSTableAttached))
complete &= group.validateSSTableAttachedIndexes(sstables, throwOnIncomplete);
complete &= group.validateSSTableAttachedIndexes(sstables, throwOnIncomplete, validateChecksum);
}
return complete;

2
src/java/org/apache/cassandra/index/sai/SSTableContextManager.java
View File

@ -77,7 +77,7 @@ public class SSTableContextManager
try
{
// Only validate on restart or newly refreshed SSTable. Newly built files are unlikely to be corrupted.
if (!sstableContexts.containsKey(sstable) && !indexDescriptor.validatePerSSTableComponents(validation))
if (!sstableContexts.containsKey(sstable) && !indexDescriptor.validatePerSSTableComponents(validation, true, false))
{
invalid.add(sstable);
removeInvalidSSTableContext(sstable);

2
src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java
View File

@ -264,7 +264,7 @@ public class StorageAttachedIndexBuilder extends SecondaryIndexBuilder
// if per-table files are incomplete, full rebuild is requested, or checksum fails
if (!indexDescriptor.isPerSSTableIndexBuildComplete()
|| isFullRebuild
|| !indexDescriptor.validatePerSSTableComponents(IndexValidation.CHECKSUM))
|| !indexDescriptor.validatePerSSTableComponents(IndexValidation.CHECKSUM, true, false))
{
CountDownLatch latch = CountDownLatch.newCountDownLatch(1);
if (inProgress.putIfAbsent(sstable, latch) == null)

12
src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java
View File

@ -343,7 +343,7 @@ public class StorageAttachedIndexGroup implements Index.Group, INotificationCons
}
@Override
public boolean validateSSTableAttachedIndexes(Collection<SSTableReader> sstables, boolean throwOnIncomplete)
public boolean validateSSTableAttachedIndexes(Collection<SSTableReader> sstables, boolean throwOnIncomplete, boolean validateChecksum)
{
boolean complete = true;
@ -353,21 +353,21 @@ public class StorageAttachedIndexGroup implements Index.Group, INotificationCons
if (indexDescriptor.isPerSSTableIndexBuildComplete())
{
indexDescriptor.checksumPerSSTableComponents();
indexDescriptor.validatePerSSTableComponents(IndexValidation.CHECKSUM, validateChecksum, true);
for (StorageAttachedIndex index : indexes)
{
if (indexDescriptor.isPerColumnIndexBuildComplete(index.identifier()))
indexDescriptor.checksumPerIndexComponents(index.termType(), index.identifier());
indexDescriptor.validatePerIndexComponents(index.termType(), index.identifier(), IndexValidation.CHECKSUM, validateChecksum, true);
else if (throwOnIncomplete)
throw new IllegalStateException(indexDescriptor.logMessage("Incomplete per-column index build"));
throw new IllegalStateException(indexDescriptor.logMessage("Incomplete per-column index build for SSTable " + sstable.descriptor.toString()));
else
complete = false;
}
}
else if (throwOnIncomplete)
{
throw new IllegalStateException(indexDescriptor.logMessage("Incomplete per-SSTable index build"));
throw new IllegalStateException(indexDescriptor.logMessage("Incomplete per-SSTable index build" + sstable.descriptor.toString()));
}
else
{
@ -375,7 +375,7 @@ public class StorageAttachedIndexGroup implements Index.Group, INotificationCons
}
}
return complete;
return complete;
}
/**

35
src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java
View File

@ -364,56 +364,49 @@ public class IndexDescriptor
}
@SuppressWarnings("BooleanMethodIsAlwaysInverted")
public boolean validatePerIndexComponents(IndexTermType indexTermType, IndexIdentifier indexIdentifier, IndexValidation validation)
public boolean validatePerIndexComponents(IndexTermType indexTermType, IndexIdentifier indexIdentifier, IndexValidation validation, boolean validateChecksum, boolean rethrow)
{
if (validation == IndexValidation.NONE)
return true;
logger.info(logMessage("Validating per-column index components for {} using mode {}"), indexIdentifier, validation);
boolean checksum = validation == IndexValidation.CHECKSUM;
logger.info(logMessage("Validating per-column index components for {} for SSTable {} using mode {}"), indexIdentifier, sstableDescriptor.toString(), validation);
try
{
version.onDiskFormat().validatePerColumnIndexComponents(this, indexTermType, indexIdentifier, checksum);
version.onDiskFormat().validatePerColumnIndexComponents(this, indexTermType, indexIdentifier, validation == IndexValidation.CHECKSUM && validateChecksum);
return true;
}
catch (UncheckedIOException e)
{
return false;
if (rethrow)
throw e;
else
return false;
}
}
@SuppressWarnings("BooleanMethodIsAlwaysInverted")
public boolean validatePerSSTableComponents(IndexValidation validation)
public boolean validatePerSSTableComponents(IndexValidation validation, boolean validateChecksum, boolean rethrow)
{
if (validation == IndexValidation.NONE)
return true;
logger.info(logMessage("Validating per-sstable index components using mode " + validation));
boolean checksum = validation == IndexValidation.CHECKSUM;
logger.info(logMessage("Validating per-sstable index components for SSTable {} using mode {}"), sstableDescriptor.toString(), validation);
try
{
version.onDiskFormat().validatePerSSTableIndexComponents(this, checksum);
version.onDiskFormat().validatePerSSTableIndexComponents(this, validation == IndexValidation.CHECKSUM && validateChecksum);
return true;
}
catch (UncheckedIOException e)
{
return false;
if (rethrow)
throw e;
else
return false;
}
}
public void checksumPerIndexComponents(IndexTermType indexTermType, IndexIdentifier indexIdentifier)
{
version.onDiskFormat().validatePerColumnIndexComponents(this, indexTermType, indexIdentifier, true);
}
@SuppressWarnings("BooleanMethodIsAlwaysInverted")
public void checksumPerSSTableComponents()
{
version.onDiskFormat().validatePerSSTableIndexComponents(this, true);
}
public void deletePerSSTableIndexComponents()
{
version.onDiskFormat()

4
src/java/org/apache/cassandra/index/sai/disk/v1/segment/SegmentTrieBuffer.java
View File

@ -22,7 +22,7 @@ import java.util.Map;
import java.util.concurrent.atomic.LongAdder;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.cassandra.db.memtable.TrieMemtable;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.tries.InMemoryTrie;
import org.apache.cassandra.index.sai.postings.PostingList;
import org.apache.cassandra.index.sai.utils.IndexEntry;
@ -45,7 +45,7 @@ public class SegmentTrieBuffer
public SegmentTrieBuffer()
{
trie = new InMemoryTrie<>(TrieMemtable.BUFFER_TYPE);
trie = new InMemoryTrie<>(DatabaseDescriptor.getMemtableAllocationType().toBufferType());
postingsAccumulator = new PostingsAccumulator();
}

2
src/java/org/apache/cassandra/index/sai/view/IndexViewManager.java
View File

@ -178,7 +178,7 @@ public class IndexViewManager
{
if (validation != IndexValidation.NONE)
{
if (!sstableContext.indexDescriptor.validatePerIndexComponents(index.termType(), index.identifier(), validation))
if (!sstableContext.indexDescriptor.validatePerIndexComponents(index.termType(), index.identifier(), validation, true, false))
{
invalid.add(sstableContext);
continue;

13
src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
View File

@ -23,7 +23,8 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Collections;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;
@ -33,6 +34,7 @@ import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.EncodingStats;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.schema.TableMetadataRef;
@ -49,12 +51,14 @@ abstract class AbstractSSTableSimpleWriter implements Closeable
protected SSTableFormat<?, ?> format = DatabaseDescriptor.getSelectedSSTableFormat();
protected static final AtomicReference<SSTableId> id = new AtomicReference<>(SSTableIdFactory.instance.defaultBuilder().generator(Stream.empty()).get());
protected boolean makeRangeAware = false;
protected final Collection<Index.Group> indexGroups;
protected AbstractSSTableSimpleWriter(File directory, TableMetadataRef metadata, RegularAndStaticColumns columns)
{
this.metadata = metadata;
this.directory = directory;
this.columns = columns;
indexGroups = new ArrayList<>();
}
protected void setSSTableFormatType(SSTableFormat<?, ?> type)
@ -67,6 +71,11 @@ abstract class AbstractSSTableSimpleWriter implements Closeable
this.makeRangeAware = makeRangeAware;
}
protected void addIndexGroup(Index.Group indexGroup)
{
this.indexGroups.add(indexGroup);
}
protected SSTableTxnWriter createWriter(SSTable.Owner owner) throws IOException
{
SerializationHeader header = new SerializationHeader(true, metadata.get(), columns, EncodingStats.NO_STATS);
@ -81,7 +90,7 @@ abstract class AbstractSSTableSimpleWriter implements Closeable
ActiveRepairService.NO_PENDING_REPAIR,
false,
header,
Collections.emptySet(),
indexGroups,
owner);
}

171
src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
View File

@ -19,7 +19,9 @@ package org.apache.cassandra.io.sstable;
import java.io.Closeable;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.nio.file.NoSuchFileException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@ -41,9 +43,13 @@ import org.apache.cassandra.cql3.UpdateParameters;
import org.apache.cassandra.cql3.functions.types.TypeCodec;
import org.apache.cassandra.cql3.functions.types.UserType;
import org.apache.cassandra.cql3.statements.ModificationStatement;
import org.apache.cassandra.cql3.statements.schema.CreateIndexStatement;
import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
import org.apache.cassandra.cql3.statements.schema.CreateTypeStatement;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Slice;
import org.apache.cassandra.db.Slices;
import org.apache.cassandra.db.marshal.AbstractType;
@ -51,12 +57,15 @@ import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.SyntaxException;
import org.apache.cassandra.index.sai.StorageAttachedIndexGroup;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.SchemaTransformation;
import org.apache.cassandra.schema.SchemaTransformations;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
@ -65,10 +74,13 @@ import org.apache.cassandra.schema.Types;
import org.apache.cassandra.schema.UserFunctions;
import org.apache.cassandra.schema.Views;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.transformations.AlterSchema;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.JavaDriverUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -107,7 +119,7 @@ import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
* // Close the writer, finalizing the sstable
* writer.close();
* </pre>
*
* <p>
* Please note that {@code CQLSSTableWriter} is <b>not</b> thread-safe (multiple threads cannot access the
* same instance). It is however safe to use multiple instances in parallel (even if those instance write
* sstables for the same table).
@ -129,15 +141,15 @@ public class CQLSSTableWriter implements Closeable
private final AbstractSSTableSimpleWriter writer;
private final ModificationStatement modificationStatement;
private final List<ColumnSpecification> boundNames;
private final List<TypeCodec> typeCodecs;
private final List<TypeCodec<?>> typeCodecs;
private CQLSSTableWriter(AbstractSSTableSimpleWriter writer, ModificationStatement modificationStatement, List<ColumnSpecification> boundNames)
{
this.writer = writer;
this.modificationStatement = modificationStatement;
this.boundNames = boundNames;
this.typeCodecs = boundNames.stream().map(bn -> JavaDriverUtils.codecFor(JavaDriverUtils.driverType(bn.type)))
.collect(Collectors.toList());
this.typeCodecs = boundNames.stream().map(bn -> JavaDriverUtils.codecFor(JavaDriverUtils.driverType(bn.type)))
.collect(Collectors.toList());
}
/**
@ -156,7 +168,7 @@ public class CQLSSTableWriter implements Closeable
* This is a shortcut for {@code addRow(Arrays.asList(values))}.
*
* @param values the row values (corresponding to the bind variables of the
* modification statement used when creating by this writer).
* modification statement used when creating by this writer).
* @return this writer.
*/
public CQLSSTableWriter addRow(Object... values)
@ -177,7 +189,7 @@ public class CQLSSTableWriter implements Closeable
* {@link #rawAddRow} instead.
*
* @param values the row values (corresponding to the bind variables of the
* modification statement used when creating by this writer).
* modification statement used when creating by this writer).
* @return this writer.
*/
public CQLSSTableWriter addRow(List<Object> values)
@ -209,10 +221,10 @@ public class CQLSSTableWriter implements Closeable
* (in which case the map key must use the exact case of the column).
*
* @param values a map of colum name to column values representing the new
* row to add. Note that if a column is not part of the map, it's value will
* be {@code null}. If the map contains keys that does not correspond to one
* of the column of the modification statement used when creating this writer, the
* the corresponding value is ignored.
* row to add. Note that if a column is not part of the map, it's value will
* be {@code null}. If the map contains keys that does not correspond to one
* of the column of the modification statement used when creating this writer, the
* the corresponding value is ignored.
* @return this writer.
*/
public CQLSSTableWriter addRow(Map<String, Object> values)
@ -233,7 +245,7 @@ public class CQLSSTableWriter implements Closeable
* Adds a new row to the writer given already serialized values.
*
* @param values the row values (corresponding to the bind variables of the
* modification statement used when creating by this writer) as binary.
* modification statement used when creating by this writer) as binary.
* @return this writer.
*/
public CQLSSTableWriter rawAddRow(ByteBuffer... values)
@ -248,7 +260,7 @@ public class CQLSSTableWriter implements Closeable
* This is a shortcut for {@code rawAddRow(Arrays.asList(values))}.
*
* @param values the row values (corresponding to the bind variables of the
* modification statement used when creating by this writer) as binary.
* modification statement used when creating by this writer) as binary.
* @return this writer.
*/
public CQLSSTableWriter rawAddRow(List<ByteBuffer> values)
@ -275,7 +287,8 @@ public class CQLSSTableWriter implements Closeable
try
{
if (modificationStatement.hasSlices()) {
if (modificationStatement.hasSlices())
{
Slices slices = modificationStatement.createSlices(options);
for (ByteBuffer key : keys)
@ -300,7 +313,7 @@ public class CQLSSTableWriter implements Closeable
{
// If we use a BufferedWriter and had a problem writing to disk, the IOException has been
// wrapped in a SyncException (see BufferedWriter below). We want to extract that IOE.
throw (IOException)e.getCause();
throw (IOException) e.getCause();
}
}
@ -313,10 +326,10 @@ public class CQLSSTableWriter implements Closeable
* this write.
*
* @param values a map of colum name to column values representing the new
* row to add. Note that if a column is not part of the map, it's value will
* be {@code null}. If the map contains keys that does not correspond to one
* of the column of the modification statement used when creating this writer, the
* the corresponding value is ignored.
* row to add. Note that if a column is not part of the map, it's value will
* be {@code null}. If the map contains keys that does not correspond to one
* of the column of the modification statement used when creating this writer, the
* the corresponding value is ignored.
* @return this writer.
*/
public CQLSSTableWriter rawAddRow(Map<String, ByteBuffer> values)
@ -370,9 +383,10 @@ public class CQLSSTableWriter implements Closeable
{
// For backwards-compatibility with consumers that may be passing
// an Integer for a Date field, for example.
return ((AbstractType)columnSpecification.type).decompose(value);
return ((AbstractType) columnSpecification.type).decompose(value);
}
}
/**
* A Builder for a CQLSSTableWriter object.
*/
@ -384,6 +398,7 @@ public class CQLSSTableWriter implements Closeable
protected SSTableFormat<?, ?> format = null;
private final List<CreateTypeStatement.Raw> typeStatements;
private final List<CreateIndexStatement.Raw> indexStatements;
private File directory;
private CreateTableStatement.Raw schemaStatement;
@ -391,10 +406,12 @@ public class CQLSSTableWriter implements Closeable
private IPartitioner partitioner;
private boolean sorted = false;
private long maxSSTableSizeInMiB = -1L;
private boolean buildIndexes = true;
protected Builder()
{
this.typeStatements = new ArrayList<>();
this.indexStatements = new ArrayList<>();
}
/**
@ -404,7 +421,6 @@ public class CQLSSTableWriter implements Closeable
*
* @param directory the directory to use, which should exists and be writable.
* @return this builder.
*
* @throws IllegalArgumentException if {@code directory} doesn't exist or is not writable.
*/
public Builder inDirectory(String directory)
@ -419,7 +435,6 @@ public class CQLSSTableWriter implements Closeable
*
* @param directory the directory to use, which should exists and be writable.
* @return this builder.
*
* @throws IllegalArgumentException if {@code directory} doesn't exist or is not writable.
*/
public Builder inDirectory(File directory)
@ -449,9 +464,8 @@ public class CQLSSTableWriter implements Closeable
*
* @param schema the schema of the table for which sstables are to be created.
* @return this builder.
*
* @throws IllegalArgumentException if {@code schema} is not a valid CREATE TABLE statement
* or does not have a fully-qualified table name.
* or does not have a fully-qualified table name.
*/
public Builder forTable(String schema)
{
@ -459,6 +473,20 @@ public class CQLSSTableWriter implements Closeable
return this;
}
/**
* The schema (CREATE INDEX statement) for index to be created for the table. Only SAI indexes are supported.
*
* @param indexes CQL statements representing SAI indexes to be created.
* @return this builder
*/
public Builder withIndexes(String... indexes)
{
for (String index : indexes)
indexStatements.add(QueryProcessor.parseStatement(index, CreateIndexStatement.Raw.class, "CREATE INDEX"));
return this;
}
/**
* The partitioner to use.
* <p>
@ -485,11 +513,10 @@ public class CQLSSTableWriter implements Closeable
* This is a mandatory option.
*
* @param modificationStatement an insert, update, or delete statement that defines the order
* of column values to use.
* of column values to use.
* @return this builder.
*
* @throws IllegalArgumentException if {@code modificationStatement} is not a valid insert, update, or delete
* statement, does not have a fully-qualified table name or have no bind variables.
* statement, does not have a fully-qualified table name or have no bind variables.
*/
public Builder using(String modificationStatement)
{
@ -528,9 +555,9 @@ public class CQLSSTableWriter implements Closeable
* The default is 128MiB, which should be reasonable for a 1GiB heap. If you experience
* OOM while using the writer, you should lower this value.
*
* @deprecated This method is deprecated in favor of the new withMaxSSTableSizeInMiB(int size)
* @param size the size to use in MiB.
* @return this builder.
* @deprecated This method is deprecated in favor of the new withMaxSSTableSizeInMiB(int size)
*/
@Deprecated(since = "5.0")
public Builder withBufferSizeInMiB(int size)
@ -548,9 +575,9 @@ public class CQLSSTableWriter implements Closeable
* The default is 128MiB, which should be reasonable for a 1GiB heap. If you experience
* OOM while using the writer, you should lower this value.
*
* @deprecated This method is deprecated in favor of the new withBufferSizeInMiB(int size). See CASSANDRA-17675
* @param size the size to use in MiB.
* @return this builder.
* @deprecated This method is deprecated in favor of the new withBufferSizeInMiB(int size). See CASSANDRA-17675
*/
@Deprecated(since = "4.1")
public Builder withBufferSizeInMB(int size)
@ -581,6 +608,18 @@ public class CQLSSTableWriter implements Closeable
return this;
}
/**
* Whether indexes should be built and serialized to disk along data. Defaults to true.
*
* @param buildIndexes true if indexes should be built, false otherwise
* @return this builder
*/
public Builder withBuildIndexes(boolean buildIndexes)
{
this.buildIndexes = buildIndexes;
return this;
}
public CQLSSTableWriter build()
{
if (directory == null)
@ -603,6 +642,7 @@ public class CQLSSTableWriter implements Closeable
synchronized (CQLSSTableWriter.class)
{
String keyspaceName = schemaStatement.keyspace();
String tableName = schemaStatement.table();
Schema.instance.submit(SchemaTransformations.addKeyspace(KeyspaceMetadata.create(keyspaceName,
KeyspaceParams.simple(1),
@ -618,13 +658,56 @@ public class CQLSSTableWriter implements Closeable
Types.none(),
UserFunctions.none());
TableMetadata tableMetadata = ksm.tables.getNullable(schemaStatement.table());
TableMetadata tableMetadata = Schema.instance.getTableMetadata(keyspaceName, tableName);
if (tableMetadata == null)
{
Types types = createTypes(keyspaceName);
Schema.instance.submit(SchemaTransformations.addTypes(types, true));
tableMetadata = createTable(types);
Schema.instance.submit(SchemaTransformations.addTable(tableMetadata, true));
if (buildIndexes && !indexStatements.isEmpty())
{
// we need to commit keyspace metadata first so applyIndexes sees that keyspace from TCM
commitKeyspaceMetadata(ksm.withSwapped(ksm.tables.with(tableMetadata)));
applyIndexes(keyspaceName);
}
KeyspaceMetadata keyspaceMetadata = ClusterMetadata.current().schema.getKeyspaceMetadata(keyspaceName);
tableMetadata = keyspaceMetadata.tables.getNullable(tableName);
Schema.instance.submit(SchemaTransformations.addTable(tableMetadata, true));
}
ColumnFamilyStore cfs = null;
if (buildIndexes && !indexStatements.isEmpty())
{
KeyspaceMetadata keyspaceMetadata = ClusterMetadata.current().schema.getKeyspaceMetadata(keyspaceName);
Keyspace keyspace = Keyspace.mockKS(keyspaceMetadata);
Directories directories = new Directories(tableMetadata, Collections.singleton(new Directories.DataDirectory(new File(directory.toPath()))));
cfs = ColumnFamilyStore.createColumnFamilyStore(keyspace,
tableName,
tableMetadata,
directories,
false,
false);
keyspace.initCfCustom(cfs);
// this is the empty directory / leftover from times we initialized ColumnFamilyStore
// it will automatically create directories for keyspace and table on disk after initialization
// we set that directory to the destination of generated SSTables so we just remove empty directories here
try
{
new File(directory, keyspaceName).deleteRecursive();
}
catch (UncheckedIOException ex)
{
if (!(ex.getCause() instanceof NoSuchFileException))
{
throw ex;
}
}
}
ModificationStatement preparedModificationStatement = prepareModificationStatement();
@ -637,6 +720,13 @@ public class CQLSSTableWriter implements Closeable
if (format != null)
writer.setSSTableFormatType(format);
if (buildIndexes && !indexStatements.isEmpty() && cfs != null)
{
StorageAttachedIndexGroup saiGroup = StorageAttachedIndexGroup.getIndexGroup(cfs);
if (saiGroup != null)
writer.addIndexGroup(saiGroup);
}
return new CQLSSTableWriter(writer, preparedModificationStatement, preparedModificationStatement.getBindVariables());
}
}
@ -654,6 +744,29 @@ public class CQLSSTableWriter implements Closeable
return builder.build();
}
/**
* Applies any provided index definitions to the target table
*
* @param keyspaceName name of the keyspace to apply indexes for
* @return table metadata reflecting applied indexes
*/
private void applyIndexes(String keyspaceName)
{
ClientState state = ClientState.forInternalCalls();
for (CreateIndexStatement.Raw statement : indexStatements)
{
Keyspaces keyspaces = statement.prepare(state).apply(ClusterMetadata.current());
commitKeyspaceMetadata(keyspaces.getNullable(keyspaceName));
}
}
private void commitKeyspaceMetadata(KeyspaceMetadata keyspaceMetadata)
{
SchemaTransformation schemaTransformation = metadata -> metadata.schema.getKeyspaces().withAddedOrUpdated(keyspaceMetadata);
ClusterMetadataService.instance().commit(new AlterSchema(schemaTransformation, Schema.instance));
}
/**
* Creates the table according to schema statement
*

10
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -1561,9 +1561,15 @@ public class NodeProbe implements AutoCloseable
ssProxy.loadNewSSTables(ksName, cfName);
}
public List<String> importNewSSTables(String ksName, String cfName, Set<String> srcPaths, boolean resetLevel, boolean clearRepaired, boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches, boolean extendedVerify, boolean copyData)
public List<String> importNewSSTables(String ksName, String cfName, Set<String> srcPaths, boolean resetLevel,
boolean clearRepaired, boolean verifySSTables, boolean verifyTokens,
boolean invalidateCaches, boolean extendedVerify, boolean copyData,
boolean failOnMissingIndex, boolean validateIndexChecksum)
{
return getCfsProxy(ksName, cfName).importNewSSTables(srcPaths, resetLevel, clearRepaired, verifySSTables, verifyTokens, invalidateCaches, extendedVerify, copyData);
return getCfsProxy(ksName, cfName).importNewSSTables(srcPaths, resetLevel, clearRepaired,
verifySSTables, verifyTokens, invalidateCaches,
extendedVerify, copyData, failOnMissingIndex,
validateIndexChecksum);
}
public void rebuildIndex(String ksName, String cfName, String... idxNames)

15
src/java/org/apache/cassandra/tools/nodetool/Import.java
View File

@ -80,6 +80,16 @@ public class Import extends NodeToolCmd
description = "Copy data from source directories instead of moving them")
private boolean copyData = false;
@Option(title = "require_index_components",
name = {"-ri", "--require-index-components"},
description = "Require existing index components for SSTables with attached indexes")
private boolean failOnMissingIndex = false;
@Option(title = "no_index_validation",
name = {"-niv", "--no-index-validation"},
description = "Skip SSTable-attached index checksum validation")
private boolean noIndexValidation = false;
@Override
public void execute(NodeProbe probe)
{
@ -92,9 +102,12 @@ public class Import extends NodeToolCmd
noInvalidateCaches = true;
noVerify = true;
extendedVerify = false;
noIndexValidation = true;
}
List<String> srcPaths = Lists.newArrayList(args.subList(2, args.size()));
List<String> failedDirs = probe.importNewSSTables(args.get(0), args.get(1), new HashSet<>(srcPaths), !keepLevel, !keepRepaired, !noVerify, !noVerifyTokens, !noInvalidateCaches, extendedVerify, copyData);
List<String> failedDirs = probe.importNewSSTables(args.get(0), args.get(1), new HashSet<>(srcPaths), !keepLevel,
!keepRepaired, !noVerify, !noVerifyTokens, !noInvalidateCaches,
extendedVerify, copyData, failOnMissingIndex, !noIndexValidation);
if (!failedDirs.isEmpty())
{
PrintStream err = probe.output().err;

222
test/unit/org/apache/cassandra/db/ImportTest.java
View File

@ -40,11 +40,20 @@ import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.dht.BootStrapper;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.index.sai.disk.format.IndexComponent;
import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
import org.apache.cassandra.index.sai.disk.io.IndexOutputWriter;
import org.apache.cassandra.index.sai.disk.v1.SAICodecUtils;
import org.apache.cassandra.index.sai.utils.IndexIdentifier;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.big.BigFormat;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.PathUtils;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.membership.NodeAddresses;
@ -53,6 +62,9 @@ import org.apache.cassandra.tcm.transformations.Register;
import org.apache.cassandra.tcm.transformations.UnsafeJoin;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.lucene.codecs.CodecUtil.FOOTER_MAGIC;
import static org.apache.lucene.codecs.CodecUtil.writeBEInt;
import static org.apache.lucene.codecs.CodecUtil.writeBELong;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@ -134,7 +146,6 @@ public class ImportTest extends CQLTester
importer.importNewSSTables(options);
assertEquals(20, execute("select * from %s").size());
}
@ -702,6 +713,215 @@ public class ImportTest extends CQLTester
}
}
@Test
public void mustNotFailOnBuiltSAIIndexesWhenRequiredTest() throws Throwable
{
try
{
schemaChange(String.format("CREATE TABLE %s.%s (id int primary key, d int)", KEYSPACE, "sai_test"));
schemaChange(String.format("CREATE INDEX idx1 ON %s.%s (d) USING 'sai'", KEYSPACE, "sai_test"));
for (int i = 0; i < 10; i++)
execute(String.format("INSERT INTO %s.%s (id, d) values (?, ?)", KEYSPACE, "sai_test"), i, i);
ColumnFamilyStore cfs = getColumnFamilyStore(KEYSPACE, "sai_test");
Util.flush(cfs);
Set<SSTableReader> sstables = cfs.getLiveSSTables();
cfs.clearUnsafe();
File backupDir = moveToBackupDir(sstables);
assertEquals(0, execute(String.format("SELECT * FROM %s.%s", KEYSPACE, "sai_test")).size());
SSTableImporter importer = new SSTableImporter(cfs);
SSTableImporter.Options options = SSTableImporter.Options.options(backupDir.toString())
.copyData(true)
.failOnMissingIndex(true)
.build();
assertTrue(importer.importNewSSTables(options).isEmpty());
assertEquals(10, execute(String.format("SELECT * FROM %s.%s WHERE d >= 0", KEYSPACE, "sai_test")).size());
}
finally
{
execute(String.format("DROP TABLE IF EXISTS %s.%s", KEYSPACE, "sai_test"));
}
}
@Test
public void mustNotFailOnMissingSAIIndexWhenSAIDoesNotExistTest() throws Throwable
{
try
{
schemaChange(String.format("CREATE TABLE %s.%s (id int primary key, d int)", KEYSPACE, "sai_less_test"));
for (int i = 0; i < 10; i++)
execute(String.format("INSERT INTO %s.%s (id, d) values (?, ?)", KEYSPACE, "sai_less_test"), i, i);
ColumnFamilyStore cfs = getColumnFamilyStore(KEYSPACE, "sai_less_test");
Util.flush(cfs);
Set<SSTableReader> sstables = cfs.getLiveSSTables();
cfs.clearUnsafe();
File backupDir = moveToBackupDir(sstables);
assertEquals(0, execute(String.format("SELECT * FROM %s.%s", KEYSPACE, "sai_less_test")).size());
SSTableImporter importer = new SSTableImporter(cfs);
SSTableImporter.Options options = SSTableImporter.Options.options(backupDir.toString())
.copyData(true)
// this does not mean anything
// because our table does not have any SAI index
.failOnMissingIndex(true)
.build();
assertTrue(importer.importNewSSTables(options).isEmpty());
assertEquals(10, execute(String.format("SELECT * FROM %s.%s", KEYSPACE, "sai_less_test")).size());
}
finally
{
execute(String.format("DROP TABLE IF EXISTS %s.%s", KEYSPACE, "sai_less_test"));
}
}
@Test
public void mustFailOnMissingSAIWhenRequiredTest() throws Throwable
{
File backupDir = null;
try
{
schemaChange(String.format("CREATE TABLE %s.%s (id int primary key, d int)", KEYSPACE, "sai_test"));
for (int i = 0; i < 10; i++)
execute(String.format("INSERT INTO %s.%s (id, d) values (?, ?)", KEYSPACE, "sai_test"), i, i);
ColumnFamilyStore cfs = getColumnFamilyStore(KEYSPACE, "sai_test");
Util.flush(cfs);
Set<SSTableReader> sstables = cfs.getLiveSSTables();
cfs.clearUnsafe();
backupDir = moveToBackupDir(sstables);
assertEquals(0, execute(String.format("SELECT * FROM %s.%s", KEYSPACE, "sai_test")).size());
// create index and load sstables, they will be without indexes (because we created
// data when index was not created yet)
schemaChange(String.format("CREATE INDEX idx1 ON %s.%s (d) USING 'sai'", KEYSPACE, "sai_test"));
SSTableImporter importer = new SSTableImporter(cfs);
SSTableImporter.Options options = SSTableImporter.Options.options(backupDir.toString())
.copyData(true)
.failOnMissingIndex(true)
.build();
assertFalse(importer.importNewSSTables(options).isEmpty());
assertEquals(0, execute(String.format("SELECT * FROM %s.%s WHERE d >= 0", KEYSPACE, "sai_test")).size());
}
finally
{
if (backupDir != null)
{
backupDir.deleteRecursive();
backupDir.parent().deleteRecursive();
}
execute(String.format("DROP TABLE IF EXISTS %s.%s", KEYSPACE, "sai_test"));
}
}
@Test
public void skipIndexChecksumOnSAITest() throws Throwable
{
try
{
schemaChange(String.format("CREATE TABLE %s.%s (id int primary key, d int)", KEYSPACE, "sai_test"));
schemaChange(String.format("CREATE INDEX idx1 ON %s.%s (d) USING 'sai'", KEYSPACE, "sai_test"));
for (int i = 0; i < 10; i++)
execute(String.format("INSERT INTO %s.%s (id, d) values (?, ?)", KEYSPACE, "sai_test"), i, i);
ColumnFamilyStore cfs = getColumnFamilyStore(KEYSPACE, "sai_test");
Util.flush(cfs);
Set<SSTableReader> sstables = cfs.getLiveSSTables();
cfs.clearUnsafe();
File backupDir = moveToBackupDir(sstables);
File[] dataFiles = backupDir.list(f -> f.name().endsWith('-' + BigFormat.Components.DATA.type.repr));
IndexDescriptor indexDescriptor = IndexDescriptor.create(Descriptor.fromFile(dataFiles[0]),
Murmur3Partitioner.instance,
Schema.instance.getTableMetadata(KEYSPACE, "sai_test").comparator);
IndexIdentifier indexIdentifier = new IndexIdentifier(KEYSPACE, "sai_test", "idx1");
// corrupt one of index files
try (IndexOutputWriter output = indexDescriptor.openPerIndexOutput(IndexComponent.COLUMN_COMPLETION_MARKER, indexIdentifier))
{
SAICodecUtils.writeHeader(output);
output.writeByte((byte) 0);
// taken from SAICodecUtils#writeFooter
writeBEInt(output, FOOTER_MAGIC);
writeBEInt(output, 0);
writeBELong(output, 123); // some garbage checksum value to prove the point
}
assertEquals(0, execute(String.format("SELECT * FROM %s.%s", KEYSPACE, "sai_test")).size());
SSTableImporter importer = new SSTableImporter(cfs);
SSTableImporter.Options options = SSTableImporter.Options.options(backupDir.toString())
.copyData(true)
.failOnMissingIndex(true)
.validateIndexChecksum(false)
.build();
// even with corrupted column completion marker (wrong checksum), it will import
assertTrue(importer.importNewSSTables(options).isEmpty());
assertEquals(10, execute(String.format("SELECT * FROM %s.%s", KEYSPACE, "sai_test")).size());
}
finally
{
execute(String.format("DROP TABLE IF EXISTS %s.%s", KEYSPACE, "sai_test"));
}
}
@Test
public void skipEmptyIndexChecksumOnSAITest() throws Throwable
{
try
{
schemaChange(String.format("CREATE TABLE %s.%s (id int primary key, d int)", KEYSPACE, "sai_test"));
schemaChange(String.format("CREATE INDEX idx1 ON %s.%s (d) USING 'sai'", KEYSPACE, "sai_test"));
// no data in indexed column = empty index
for (int i = 0; i < 10; i++)
execute(String.format("INSERT INTO %s.%s (id) values (?)", KEYSPACE, "sai_test"), i);
ColumnFamilyStore cfs = getColumnFamilyStore(KEYSPACE, "sai_test");
Util.flush(cfs);
Set<SSTableReader> sstables = cfs.getLiveSSTables();
cfs.clearUnsafe();
File backupDir = moveToBackupDir(sstables);
assertEquals(0, execute(String.format("SELECT * FROM %s.%s", KEYSPACE, "sai_test")).size());
SSTableImporter importer = new SSTableImporter(cfs);
SSTableImporter.Options options = SSTableImporter.Options.options(backupDir.toString())
.copyData(true)
.failOnMissingIndex(true)
.validateIndexChecksum(true)
.build();
assertTrue(importer.importNewSSTables(options).isEmpty());
assertEquals(10, execute(String.format("SELECT * FROM %s.%s", KEYSPACE, "sai_test")).size());
}
finally
{
execute(String.format("DROP TABLE IF EXISTS %s.%s", KEYSPACE, "sai_test"));
}
}
private static class MockCFS extends ColumnFamilyStore
{
public MockCFS(ColumnFamilyStore cfs, Directories dirs)

4
test/unit/org/apache/cassandra/index/sai/SAITester.java
View File

@ -379,8 +379,8 @@ public abstract class SAITester extends CQLTester
for (SSTableReader sstable : cfs.getLiveSSTables())
{
IndexDescriptor indexDescriptor = IndexDescriptor.create(sstable);
if (!indexDescriptor.validatePerSSTableComponents(IndexValidation.CHECKSUM)
|| !indexDescriptor.validatePerIndexComponents(indexContext, indexIdentifier, IndexValidation.CHECKSUM))
if (!indexDescriptor.validatePerSSTableComponents(IndexValidation.CHECKSUM, true, false)
|| !indexDescriptor.validatePerIndexComponents(indexContext, indexIdentifier, IndexValidation.CHECKSUM, true, false))
return false;
}
return true;

78
test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
View File

@ -18,79 +18,47 @@
package org.apache.cassandra.io.sstable;
import java.io.IOException;
import java.util.function.BiPredicate;
import com.google.common.io.Files;
import org.apache.cassandra.io.util.File;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import static org.junit.Assert.assertEquals;
public class CQLSSTableWriterClientTest
public class CQLSSTableWriterClientTest extends CQLSSTableWriterTest
{
private File testDirectory;
private IPartitioner oldPartitioner;
@Before
public void setUp()
public void setup()
{
// setting this to true will execute a CQL query to table
// and this path is not enabled in client mode
verifyDataAfterLoading = false;
this.testDirectory = new File(Files.createTempDir());
DatabaseDescriptor.clientInitialization();
}
@Test
public void testMultipleWritersWithDistinctTables() throws IOException
{
testWriterInClientMode("table1", "table2");
}
@Test
public void testMultipleWritersWithSameTable() throws IOException
{
testWriterInClientMode("table1", "table1");
}
public void testWriterInClientMode(String table1, String table2) throws IOException, InvalidRequestException
{
String schema = "CREATE TABLE client_test.%s ("
+ " k int PRIMARY KEY,"
+ " v1 text,"
+ " v2 int"
+ ")";
String insert = "INSERT INTO client_test.%s (k, v1, v2) VALUES (?, ?, ?)";
CQLSSTableWriter writer = CQLSSTableWriter.builder()
.inDirectory(this.testDirectory)
.forTable(String.format(schema, table1))
.using(String.format(insert, table1)).build();
CQLSSTableWriter writer2 = CQLSSTableWriter.builder()
.inDirectory(this.testDirectory)
.forTable(String.format(schema, table2))
.using(String.format(insert, table2)).build();
writer.addRow(0, "A", 0);
writer2.addRow(0, "A", 0);
writer.addRow(1, "B", 1);
writer2.addRow(1, "B", 1);
writer.close();
writer2.close();
BiPredicate<File, String> filter = (dir, name) -> name.endsWith("-Data.db");
File[] dataFiles = this.testDirectory.tryList(filter);
assertEquals(2, dataFiles.length);
DatabaseDescriptor.clientInitialization(true,
() -> {
Config config = new Config();
config.data_file_directories = new String[]{ testDirectory.absolutePath() };
return config;
});
CassandraRelevantProperties.FORCE_LOAD_LOCAL_KEYSPACES.setBoolean(true);
oldPartitioner = DatabaseDescriptor.setPartitionerUnsafe(ByteOrderedPartitioner.instance);
Keyspace.setInitialized();
}
@After
public void tearDown()
{
FileUtils.deleteRecursive(this.testDirectory);
DatabaseDescriptor.setPartitionerUnsafe(oldPartitioner);
}
}

2
test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterConcurrencyTest.java
View File

@ -45,7 +45,7 @@ import static org.assertj.core.api.Assertions.assertThat;
*/
public class CQLSSTableWriterConcurrencyTest extends CQLTester
{
private static final Logger LOGGER = LoggerFactory.getLogger(CQLSSTableWriterTest.class);
private static final Logger LOGGER = LoggerFactory.getLogger(CQLSSTableWriterConcurrencyTest.class);
@Rule
public TemporaryFolder tempFolder = new TemporaryFolder();

44
test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterDaemonTest.java Normal file
View File

@ -0,0 +1,44 @@
/*
* 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.io.sstable;
import org.junit.BeforeClass;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
public class CQLSSTableWriterDaemonTest extends CQLSSTableWriterTest
{
@BeforeClass
public static void setup() throws Exception
{
DatabaseDescriptor.daemonInitialization();
CommitLog.instance.start();
SchemaLoader.cleanupAndLeaveDirs();
Keyspace.setInitialized();
ServerTestUtils.prepareServerNoRegister();
MessagingService.instance().waitUntilListeningUnchecked();
StorageService.instance.initServer();
}
}

1141
test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
View File

File diff suppressed because it is too large Load Diff