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

Make CQLSSTableWriter to support building of SAI indexes 

patch by Stefan Miklosovic; reviewed by Caleb Rackliffe, Doug Rohrer for CASSANDRA-18714
This commit is contained in:
Stefan Miklosovic 2024-01-08 21:13:23 +01:00
parent 9f5e45e5a2
commit 016dd6ca37
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
26 changed files with 916 additions and 319 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.0-beta2
* 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;
@ -1164,7 +1165,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

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

@ -200,22 +200,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.
@ -394,7 +396,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
indexManager.reload();
memtableFactory = metadata().params.memtable.factory();
switchMemtableOrNotify(FlushReason.SCHEMA_CHANGE, Memtable::metadataUpdated);
if (DatabaseDescriptor.isDaemonInitialized())
switchMemtableOrNotify(FlushReason.SCHEMA_CHANGE, Memtable::metadataUpdated);
}
public static Runnable getBackgroundCompactionTaskSubmitter()
@ -855,26 +859,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

@ -238,7 +238,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);
}
@ -502,6 +502,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
@ -509,14 +510,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);
}

149
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,14 @@ 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.Directories.DataDirectory;
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,10 +58,12 @@ 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.SchemaTransformations;
@ -68,6 +77,7 @@ import org.apache.cassandra.service.ClientState;
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;
@ -106,7 +116,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).
@ -127,15 +137,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());
}
/**
@ -154,7 +164,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)
@ -175,7 +185,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)
@ -207,10 +217,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)
@ -231,7 +241,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)
@ -246,7 +256,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)
@ -273,7 +283,8 @@ public class CQLSSTableWriter implements Closeable
try
{
if (modificationStatement.hasSlices()) {
if (modificationStatement.hasSlices())
{
Slices slices = modificationStatement.createSlices(options);
for (ByteBuffer key : keys)
@ -298,7 +309,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();
}
}
@ -311,10 +322,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)
@ -368,9 +379,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.
*/
@ -382,6 +394,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;
@ -389,10 +402,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<>();
}
/**
@ -402,7 +417,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)
@ -417,7 +431,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)
@ -447,9 +460,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)
{
@ -457,6 +469,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>
@ -483,11 +509,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)
{
@ -526,9 +551,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)
@ -546,9 +571,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)
@ -579,6 +604,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)
@ -600,7 +637,6 @@ public class CQLSSTableWriter implements Closeable
synchronized (CQLSSTableWriter.class)
{
String keyspaceName = schemaStatement.keyspace();
Schema.instance.transform(SchemaTransformations.addKeyspace(KeyspaceMetadata.create(keyspaceName,
@ -618,6 +654,37 @@ public class CQLSSTableWriter implements Closeable
Types types = createTypes(keyspaceName);
Schema.instance.transform(SchemaTransformations.addTypes(types, true));
tableMetadata = createTable(types);
if (buildIndexes && !indexStatements.isEmpty())
{
tableMetadata = applyIndexes(ksm.withSwapped(ksm.tables.with(tableMetadata)));
Keyspace ks = Keyspace.openWithoutSSTables(keyspaceName);
Directories directories = new Directories(tableMetadata, Collections.singleton(new DataDirectory(new File(directory.toPath()))));
ColumnFamilyStore cfs = ColumnFamilyStore.createColumnFamilyStore(ks,
tableMetadata.name,
TableMetadataRef.forOfflineTools(tableMetadata),
directories,
false,
false,
true);
ks.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;
}
}
}
Schema.instance.transform(SchemaTransformations.addTable(tableMetadata, true));
}
@ -631,6 +698,13 @@ public class CQLSSTableWriter implements Closeable
if (format != null)
writer.setSSTableFormatType(format);
if (buildIndexes && !indexStatements.isEmpty())
{
StorageAttachedIndexGroup saiGroup = StorageAttachedIndexGroup.getIndexGroup(Schema.instance.getColumnFamilyStoreInstance(tableMetadata.id));
if (saiGroup != null)
writer.addIndexGroup(saiGroup);
}
return new CQLSSTableWriter(writer, preparedModificationStatement, preparedModificationStatement.getBindVariables());
}
}
@ -648,6 +722,23 @@ public class CQLSSTableWriter implements Closeable
return builder.build();
}
/**
* Applies any provided index definitions to the target table
*
* @param ksm the KeyspaceMetadata object that has the table defined
* @return an updated TableMetadata instance with the indexe create statements applied
*/
private TableMetadata applyIndexes(KeyspaceMetadata ksm)
{
ClientState state = ClientState.forInternalCalls();
Keyspaces keyspaces = Keyspaces.of(ksm);
for (CreateIndexStatement.Raw statement : indexStatements)
keyspaces = statement.prepare(state).apply(keyspaces);
return keyspaces.get(ksm.name).get().tables.get(schemaStatement.table()).get();
}
/**
* Creates the table according to schema statement
*

3
src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
View File

@ -324,7 +324,8 @@ public abstract class AbstractReplicationStrategy
try
{
Constructor<? extends AbstractReplicationStrategy> constructor = strategyClass.getConstructor(parameterTypes);
strategy = constructor.newInstance(keyspaceName, tokenMetadata, snitch, strategyOptions);
IEndpointSnitch endpointSnitch = snitch == null && DatabaseDescriptor.isClientOrToolInitialized() ? new SimpleSnitch() : snitch;
strategy = constructor.newInstance(keyspaceName, tokenMetadata, endpointSnitch, strategyOptions);
}
catch (InvocationTargetException e)
{

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

@ -1544,9 +1544,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;

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

@ -40,17 +40,29 @@ 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.locator.TokenMetadata;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
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;
@ -132,7 +144,6 @@ public class ImportTest extends CQLTester
importer.importNewSSTables(options);
assertEquals(20, execute("select * from %s").size());
}
@ -710,6 +721,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"));
createIndexAndWait(String.format("CREATE INDEX idx1 ON %s.%s (d) USING 'sai'", KEYSPACE, "sai_test"), "idx1");
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)
createIndexAndWait(String.format("CREATE INDEX idx1 ON %s.%s (d) USING 'sai'", KEYSPACE, "sai_test"), "idx1");
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"));
createIndexAndWait(String.format("CREATE INDEX idx1 ON %s.%s (d) USING 'sai'", KEYSPACE, "sai_test"), "idx1");
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"));
createIndexAndWait(String.format("CREATE INDEX idx1 ON %s.%s (d) USING 'sai'", KEYSPACE, "sai_test"), "idx1");
// 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)
@ -717,4 +937,11 @@ public class ImportTest extends CQLTester
super(cfs.keyspace, cfs.getTableName(), Util.newSeqGen(), cfs.metadata, dirs, false, false, true);
}
}
private void createIndexAndWait(String query, String indexName)
{
schemaChange(query);
waitForIndexQueryable(KEYSPACE, indexName);
}
}

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;

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

@ -17,72 +17,37 @@
*/
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.io.util.FileUtils;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.dht.IPartitioner;
import static org.junit.Assert.assertEquals;
public class CQLSSTableWriterClientTest
public class CQLSSTableWriterClientTest extends CQLSSTableWriterTest
{
private File testDirectory;
private IPartitioner oldPartitioner;
@Before
public void setUp()
{
this.testDirectory = new File(Files.createTempDir());
DatabaseDescriptor.clientInitialization();
}
@Test
public void testWriterInClientMode() throws IOException, InvalidRequestException
{
final String TABLE1 = "table1";
final String TABLE2 = "table2";
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[]{ dataDir.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

@ -46,7 +46,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();

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

@ -0,0 +1,40 @@
/*
* 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.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.commitlog.CommitLog;
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();
StorageService.instance.initServer();
}
}

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

@ -15,13 +15,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.io.sstable;
package org.apache.cassandra.io.sstable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiPredicate;
@ -30,31 +37,36 @@ import java.util.stream.StreamSupport;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.apache.cassandra.io.sstable.format.big.BigFormat;
import org.apache.cassandra.io.util.File;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import com.datastax.driver.core.utils.UUIDs;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.*;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.cql3.functions.types.*;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.functions.types.DataType;
import org.apache.cassandra.cql3.functions.types.LocalDate;
import org.apache.cassandra.cql3.functions.types.TypeCodec;
import org.apache.cassandra.cql3.functions.types.UDTValue;
import org.apache.cassandra.cql3.functions.types.UserType;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.dht.*;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
import org.apache.cassandra.index.sai.utils.IndexIdentifier;
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.schema.Schema;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.JavaDriverUtils;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.junit.Assert.assertEquals;
@ -63,30 +75,19 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class CQLSSTableWriterTest
@Ignore
public abstract class CQLSSTableWriterTest
{
private static final AtomicInteger idGen = new AtomicInteger(0);
private String keyspace;
private String table;
private String qualifiedTable;
private File dataDir;
static
{
DatabaseDescriptor.daemonInitialization();
}
private static final int NUMBER_WRITES_IN_RUNNABLE = 10;
@Rule
public TemporaryFolder tempFolder = new TemporaryFolder();
@BeforeClass
public static void setup() throws Exception
{
CommitLog.instance.start();
SchemaLoader.cleanupAndLeaveDirs();
Keyspace.setInitialized();
StorageService.instance.initServer();
}
private String keyspace;
protected String table;
private String qualifiedTable;
protected File dataDir;
@Before
public void perTestSetup() throws IOException
@ -94,7 +95,7 @@ public class CQLSSTableWriterTest
keyspace = "cql_keyspace" + idGen.incrementAndGet();
table = "table" + idGen.incrementAndGet();
qualifiedTable = keyspace + '.' + table;
dataDir = new File(tempFolder.newFolder().getAbsolutePath() + File.pathSeparator() + keyspace + File.pathSeparator() + table);
dataDir = new File(tempFolder.getRoot().getAbsolutePath() + File.pathSeparator() + keyspace + File.pathSeparator() + table);
assert dataDir.tryCreateDirectories();
}
@ -104,10 +105,10 @@ public class CQLSSTableWriterTest
try (AutoCloseable switcher = Util.switchPartitioner(ByteOrderedPartitioner.instance))
{
String schema = "CREATE TABLE " + qualifiedTable + " ("
+ " k int PRIMARY KEY,"
+ " v1 text,"
+ " v2 int"
+ ")";
+ " k int PRIMARY KEY,"
+ " v1 text,"
+ " v2 int"
+ ")";
String insert = "INSERT INTO " + qualifiedTable + " (k, v1, v2) VALUES (?, ?, ?)";
CQLSSTableWriter writer = CQLSSTableWriter.builder()
.inDirectory(dataDir)
@ -182,9 +183,9 @@ public class CQLSSTableWriterTest
// To do that simply, we use a writer with a buffer of 1MiB, and write 2 rows in the same partition with a value
// > 1MiB and validate that this created more than 1 sstable.
String schema = "CREATE TABLE " + qualifiedTable + " ("
+ " k int PRIMARY KEY,"
+ " v blob"
+ ")";
+ " k int PRIMARY KEY,"
+ " v blob"
+ ")";
String insert = "INSERT INTO " + qualifiedTable + " (k, v) VALUES (?, ?)";
CQLSSTableWriter writer = CQLSSTableWriter.builder()
.inDirectory(dataDir)
@ -203,7 +204,6 @@ public class CQLSSTableWriterTest
assert dataDir.tryListNames(filterDataFiles).length > 1 : Arrays.toString(dataDir.tryListNames(filterDataFiles));
}
@Test
public void testSyncNoEmptyRows() throws Exception
{
@ -221,11 +221,11 @@ public class CQLSSTableWriterTest
.withBufferSizeInMB(1)
.build();
for (int i = 0 ; i < 50000 ; i++) {
for (int i = 0; i < 50000; i++)
{
writer.addRow(UUID.randomUUID(), 0);
}
writer.close();
}
@Test
@ -466,11 +466,11 @@ public class CQLSSTableWriterTest
.build();
CQLSSTableWriter deleteWriter = CQLSSTableWriter.builder()
.inDirectory(dataDir)
.forTable(schema)
.using("DELETE v FROM " + qualifiedTable +
" WHERE k = ? AND c1 = ? AND c2 = ?")
.build();
.inDirectory(dataDir)
.forTable(schema)
.using("DELETE v FROM " + qualifiedTable +
" WHERE k = ? AND c1 = ? AND c2 = ?")
.build();
insertWriter.addRow("v0.2", "a", 1, 2);
insertWriter.close();
@ -509,50 +509,6 @@ public class CQLSSTableWriterTest
assertEquals(2, modifiedRow.getInt("c2"));
}
private static final int NUMBER_WRITES_IN_RUNNABLE = 10;
private class WriterThread extends Thread
{
private final File dataDir;
private final int id;
private final String qualifiedTable;
public volatile Exception exception;
public WriterThread(File dataDir, int id, String qualifiedTable)
{
this.dataDir = dataDir;
this.id = id;
this.qualifiedTable = qualifiedTable;
}
@Override
public void run()
{
String schema = "CREATE TABLE " + qualifiedTable + " ("
+ " k int,"
+ " v int,"
+ " PRIMARY KEY (k, v)"
+ ")";
String insert = "INSERT INTO " + qualifiedTable + " (k, v) VALUES (?, ?)";
CQLSSTableWriter writer = CQLSSTableWriter.builder()
.inDirectory(dataDir)
.forTable(schema)
.using(insert).build();
try
{
for (int i = 0; i < NUMBER_WRITES_IN_RUNNABLE; i++)
{
writer.addRow(id, i);
}
writer.close();
}
catch (Exception e)
{
exception = e;
}
}
}
@Test
public void testConcurrentWriters() throws Exception
{
@ -627,7 +583,8 @@ public class CQLSSTableWriterTest
assertEquals(resultSet.size(), 100);
int cnt = 0;
for (UntypedResultSet.Row row: resultSet) {
for (UntypedResultSet.Row row : resultSet)
{
assertEquals(cnt,
row.getInt("k"));
List<UDTValue> values = (List<UDTValue>) collectionCodec.deserialize(row.getBytes("v1"),
@ -689,7 +646,8 @@ public class CQLSSTableWriterTest
assertEquals(resultSet.size(), 100);
int cnt = 0;
for (UntypedResultSet.Row row: resultSet) {
for (UntypedResultSet.Row row : resultSet)
{
assertEquals(cnt,
row.getInt("k"));
UDTValue nestedTpl = (UDTValue) nestedTupleCodec.deserialize(row.getBytes("v1"),
@ -933,7 +891,8 @@ public class CQLSSTableWriterTest
.build();
final int ID_OFFSET = 1000;
for (int i = 0; i < 100 ; i++) {
for (int i = 0; i < 100; i++)
{
// Use old-style integer as date to test backwards-compatibility
writer.addRow(i, i - Integer.MIN_VALUE); // old-style raw integer needs to be offset
// Use new-style `LocalDate` for date value.
@ -945,8 +904,9 @@ public class CQLSSTableWriterTest
UntypedResultSet rs = QueryProcessor.executeInternal("SELECT * FROM " + qualifiedTable + ";");
assertEquals(200, rs.size());
Map<Integer, LocalDate> map = StreamSupport.stream(rs.spliterator(), false)
.collect(Collectors.toMap( r -> r.getInt("k"), r -> r.getDate("c")));
for (int i = 0; i < 100; i++) {
.collect(Collectors.toMap(r -> r.getInt("k"), r -> r.getDate("c")));
for (int i = 0; i < 100; i++)
{
final LocalDate expected = LocalDate.fromDaysSinceEpoch(i);
assertEquals(expected, map.get(i + ID_OFFSET));
assertEquals(expected, map.get(i));
@ -1136,13 +1096,13 @@ public class CQLSSTableWriterTest
.inDirectory(dataDir)
.forTable(schema)
.using("INSERT INTO " + qualifiedTable +
" (k, v1, v2, v3) VALUES (?,?,?,?) using timestamp ?" )
" (k, v1, v2, v3) VALUES (?,?,?,?) using timestamp ?")
.build();
// Note that, all other things being equal, Cassandra will sort these rows lexicographically, so we use "higher" values in the
// row we expect to "win" so that we're sure that it isn't just accidentally picked due to the row sorting.
writer.addRow( 1, 4, 5, "b", now); // This write should be the one found at the end because it has a higher timestamp
writer.addRow( 1, 2, 3, "a", then);
writer.addRow(1, 4, 5, "b", now); // This write should be the one found at the end because it has a higher timestamp
writer.addRow(1, 2, 3, "a", then);
writer.close();
loadSSTables(dataDir, keyspace);
@ -1157,6 +1117,7 @@ public class CQLSSTableWriterTest
assertEquals("b", r1.getString("v3"));
assertFalse(iter.hasNext());
}
@Test
public void testWriteWithTtl() throws Exception
{
@ -1169,15 +1130,15 @@ public class CQLSSTableWriterTest
+ ")";
CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder()
.inDirectory(dataDir)
.forTable(schema)
.using("INSERT INTO " + qualifiedTable +
" (k, v1, v2, v3) VALUES (?,?,?,?) using TTL ?");
.inDirectory(dataDir)
.forTable(schema)
.using("INSERT INTO " + qualifiedTable +
" (k, v1, v2, v3) VALUES (?,?,?,?) using TTL ?");
CQLSSTableWriter writer = builder.build();
// add a row that _should_ show up - 1 hour TTL
writer.addRow( 1, 2, 3, "a", 3600);
writer.addRow(1, 2, 3, "a", 3600);
// Insert a row with a TTL of 1 second - should not appear in results once we sleep
writer.addRow( 2, 4, 5, "b", 1);
writer.addRow(2, 4, 5, "b", 1);
writer.close();
Thread.sleep(1200); // Slightly over 1 second, just to make sure
loadSSTables(dataDir, keyspace);
@ -1193,6 +1154,7 @@ public class CQLSSTableWriterTest
assertEquals("a", r1.getString("v3"));
assertFalse(iter.hasNext());
}
@Test
public void testWriteWithTimestampsAndTtl() throws Exception
{
@ -1208,7 +1170,7 @@ public class CQLSSTableWriterTest
.inDirectory(dataDir)
.forTable(schema)
.using("INSERT INTO " + qualifiedTable +
" (k, v1, v2, v3) VALUES (?,?,?,?) using timestamp ? AND TTL ?" )
" (k, v1, v2, v3) VALUES (?,?,?,?) using timestamp ? AND TTL ?")
.build();
// NOTE: It would be easier to make this a timestamp in the past, but Cassandra also has a _local_ deletion time
// which is based on the server's timestamp, so simply setting the timestamp to some time in the past
@ -1216,9 +1178,9 @@ public class CQLSSTableWriterTest
long oneSecondFromNow = TimeUnit.MILLISECONDS.toMicros(currentTimeMillis() + 1000);
// Insert some rows with a timestamp of 1 second from now, and different TTLs
// add a row that _should_ show up - 1 hour TTL
writer.addRow( 1, 2, 3, "a", oneSecondFromNow, 3600);
writer.addRow(1, 2, 3, "a", oneSecondFromNow, 3600);
// Insert a row "two seconds ago" with a TTL of 1 second - should not appear in results
writer.addRow( 2, 4, 5, "b", oneSecondFromNow, 1);
writer.addRow(2, 4, 5, "b", oneSecondFromNow, 1);
writer.close();
loadSSTables(dataDir, keyspace);
UntypedResultSet resultSet = QueryProcessor.executeInternal("SELECT * FROM " + qualifiedTable);
@ -1245,7 +1207,7 @@ public class CQLSSTableWriterTest
.inDirectory(dataDir)
.forTable(schema)
.using("INSERT INTO " + qualifiedTable +
" (k, v) VALUES (?, text_as_blob(?))" )
" (k, v) VALUES (?, text_as_blob(?))")
.sorted()
.build();
int rowCount = 10_000;
@ -1276,7 +1238,7 @@ public class CQLSSTableWriterTest
.inDirectory(dataDir)
.forTable(schema)
.using("INSERT INTO " + qualifiedTable +
" (k, v) VALUES (?, text_as_blob(?))" )
" (k, v) VALUES (?, text_as_blob(?))")
.sorted()
.withMaxSSTableSizeInMiB(1)
.build();
@ -1309,25 +1271,191 @@ public class CQLSSTableWriterTest
}
}
private static void loadSSTables(File dataDir, String ks) throws ExecutionException, InterruptedException
@Test
public void testMultipleWritersWithDistinctTables() throws IOException
{
SSTableLoader loader = new SSTableLoader(dataDir, new SSTableLoader.Client()
testWriters("table1", "table2");
}
@Test
public void testMultipleWritersWithSameTable() throws IOException
{
testWriters("table1", "table1");
}
private void testWriters(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(dataDir)
.forTable(String.format(schema, table1))
.using(String.format(insert, table1))
.build();
CQLSSTableWriter writer2 = CQLSSTableWriter.builder()
.inDirectory(dataDir)
.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 = dataDir.tryList(filter);
assertEquals(2, dataFiles.length);
}
@Test
public void testWriteWithSAI() throws Exception
{
writeWithSaiInternal();
writeWithSaiInternal();
}
private void writeWithSaiInternal() throws Exception
{
String schema = "CREATE TABLE " + qualifiedTable + " ("
+ " k int PRIMARY KEY,"
+ " v1 text,"
+ " v2 int )";
String v1Index = "CREATE INDEX idx1 ON " + qualifiedTable + " (v1) USING 'sai'";
String v2Index = "CREATE INDEX idx2 ON " + qualifiedTable + " (v2) USING 'sai'";
String insert = "INSERT INTO " + qualifiedTable + " (k, v1, v2) VALUES (?, ?, ?)";
CQLSSTableWriter writer = CQLSSTableWriter.builder()
.inDirectory(dataDir)
.forTable(schema)
.using(insert)
.withIndexes(v1Index, v2Index)
.withBuildIndexes(true)
.withPartitioner(Murmur3Partitioner.instance)
.build();
int rowCount = 30_000;
for (int i = 0; i < rowCount; i++)
writer.addRow(i, UUID.randomUUID().toString(), i);
writer.close();
File[] dataFiles = dataDir.list(f -> f.name().endsWith('-' + BigFormat.Components.DATA.type.repr));
assertNotNull(dataFiles);
IndexDescriptor indexDescriptor = IndexDescriptor.create(Descriptor.fromFile(dataFiles[0]),
Murmur3Partitioner.instance,
Schema.instance.getTableMetadata(keyspace, table).comparator);
assertTrue(indexDescriptor.isPerColumnIndexBuildComplete(new IndexIdentifier(keyspace, table, "idx1")));
assertTrue(indexDescriptor.isPerColumnIndexBuildComplete(new IndexIdentifier(keyspace, table, "idx2")));
if (PathUtils.isDirectory(dataDir.toPath()))
PathUtils.forEach(dataDir.toPath(), PathUtils::deleteRecursive);
}
@Test
public void testSkipBuildingIndexesWithSAI() throws Exception
{
String schema = "CREATE TABLE " + qualifiedTable + " ("
+ " k int PRIMARY KEY,"
+ " v1 text,"
+ " v2 int )";
String v1Index = "CREATE INDEX idx1 ON " + qualifiedTable + " (v1) USING 'sai'";
String v2Index = "CREATE INDEX idx2 ON " + qualifiedTable + " (v2) USING 'sai'";
String insert = "INSERT INTO " + qualifiedTable + " (k, v1, v2) VALUES (?, ?, ?)";
CQLSSTableWriter writer = CQLSSTableWriter.builder()
.inDirectory(dataDir)
.forTable(schema)
.using(insert)
.withIndexes(v1Index, v2Index)
// not building indexes here so no SAI components will be present
.withBuildIndexes(false)
.withPartitioner(Murmur3Partitioner.instance)
.build();
int rowCount = 30_000;
for (int i = 0; i < rowCount; i++)
writer.addRow(i, UUID.randomUUID().toString(), i);
writer.close();
File[] dataFiles = dataDir.list(f -> f.name().endsWith('-' + BigFormat.Components.DATA.type.repr));
assertNotNull(dataFiles);
IndexDescriptor indexDescriptor = IndexDescriptor.create(Descriptor.fromFile(dataFiles[0]),
Murmur3Partitioner.instance,
Schema.instance.getTableMetadata(keyspace, table).comparator);
// no indexes built due to withBuildIndexes set to false
assertFalse(indexDescriptor.isPerColumnIndexBuildComplete(new IndexIdentifier(keyspace, table, "idx1")));
assertFalse(indexDescriptor.isPerColumnIndexBuildComplete(new IndexIdentifier(keyspace, table, "idx2")));
}
protected void loadSSTables(File dataDir, String ksName)
{
ColumnFamilyStore cfs = Keyspace.openWithoutSSTables(ksName).getColumnFamilyStore(table);
Set<String> dataFilePaths = Set.of(dataDir.absolutePath());
cfs.importNewSSTables(dataFilePaths, false, false, false,
false, false, false, false,
true, false);
}
private class WriterThread extends Thread
{
private final File dataDir;
private final int id;
private final String qualifiedTable;
public volatile Exception exception;
public WriterThread(File dataDir, int id, String qualifiedTable)
{
private String keyspace;
this.dataDir = dataDir;
this.id = id;
this.qualifiedTable = qualifiedTable;
}
public void init(String keyspace)
@Override
public void run()
{
String schema = "CREATE TABLE " + qualifiedTable + " ("
+ " k int,"
+ " v int,"
+ " PRIMARY KEY (k, v)"
+ ")";
String insert = "INSERT INTO " + qualifiedTable + " (k, v) VALUES (?, ?)";
CQLSSTableWriter writer = CQLSSTableWriter.builder()
.inDirectory(dataDir)
.forTable(schema)
.using(insert).build();
try
{
this.keyspace = keyspace;
for (Range<Token> range : StorageService.instance.getLocalReplicas(ks).ranges())
addRangeForEndpoint(range, FBUtilities.getBroadcastAddressAndPort());
for (int i = 0; i < NUMBER_WRITES_IN_RUNNABLE; i++)
{
writer.addRow(id, i);
}
writer.close();
}
public TableMetadataRef getTableMetadata(String cfName)
catch (Exception e)
{
return Schema.instance.getTableMetadataRef(keyspace, cfName);
exception = e;
}
}, new OutputHandler.SystemOutput(false, false));
loader.stream().get();
}
}
}