diff --git a/CHANGES.txt b/CHANGES.txt index 268bfa42b3..a7859ee9ec 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index df1cf62d15..d5b91ef530 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -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 diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 1ef67878d5..bcf4dc7073 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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 importNewSSTables(Set srcPaths, boolean resetLevel, boolean clearRepaired, boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches, boolean extendedVerify, boolean copyData) + @Override + public List importNewSSTables(Set 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 importNewSSTables(Set srcPaths, boolean resetLevel, boolean clearRepaired, boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches, boolean extendedVerify) + @Override + public List importNewSSTables(Set 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 importNewSSTables(Set 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) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java index 46e1b4d50a..7d7b9e58ee 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java @@ -189,12 +189,12 @@ public interface ColumnFamilyStoreMBean /** @deprecated See CASSANDRA-16407 */ @Deprecated(since = "4.0") public List importNewSSTables(Set 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 importNewSSTables(Set 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 importNewSSTables(Set 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(); diff --git a/src/java/org/apache/cassandra/db/SSTableImporter.java b/src/java/org/apache/cassandra/db/SSTableImporter.java index c2544f7d06..990265c61c 100644 --- a/src/java/org/apache/cassandra/db/SSTableImporter.java +++ b/src/java/org/apache/cassandra/db/SSTableImporter.java @@ -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 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 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 srcPaths, boolean resetLevel, boolean clearRepaired, boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches, boolean extendedVerify, boolean copyData) + public Options(Set 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 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); } } } diff --git a/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java b/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java index 294f1472f0..83b02db06a 100644 --- a/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java +++ b/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java @@ -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 diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java index cfb1909329..50f87c799e 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java @@ -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(); diff --git a/src/java/org/apache/cassandra/index/Index.java b/src/java/org/apache/cassandra/index/Index.java index bc7057a0fb..18fed751af 100644 --- a/src/java/org/apache/cassandra/index/Index.java +++ b/src/java/org/apache/cassandra/index/Index.java @@ -849,18 +849,19 @@ public interface Index Set 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 sstables, boolean throwOnIncomplete) + default boolean validateSSTableAttachedIndexes(Collection sstables, boolean throwOnIncomplete, boolean validateChecksum) { return true; } diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java index cc2a8182af..da15282b39 100644 --- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java +++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java @@ -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 sstables, boolean throwOnIncomplete) + public boolean validateSSTableAttachedIndexes(Collection 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; diff --git a/src/java/org/apache/cassandra/index/sai/SSTableContextManager.java b/src/java/org/apache/cassandra/index/sai/SSTableContextManager.java index 1c6406e0c1..b2df0f29c2 100644 --- a/src/java/org/apache/cassandra/index/sai/SSTableContextManager.java +++ b/src/java/org/apache/cassandra/index/sai/SSTableContextManager.java @@ -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); diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java index 5bf27b27b0..55f6381859 100644 --- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java +++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java @@ -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) diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java index fd04e44dac..30d23f5de6 100644 --- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java +++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java @@ -343,7 +343,7 @@ public class StorageAttachedIndexGroup implements Index.Group, INotificationCons } @Override - public boolean validateSSTableAttachedIndexes(Collection sstables, boolean throwOnIncomplete) + public boolean validateSSTableAttachedIndexes(Collection 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; } /** diff --git a/src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java b/src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java index 8280cf3664..4501aec7f9 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java +++ b/src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java @@ -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() diff --git a/src/java/org/apache/cassandra/index/sai/disk/v1/segment/SegmentTrieBuffer.java b/src/java/org/apache/cassandra/index/sai/disk/v1/segment/SegmentTrieBuffer.java index d1e79f91f5..72c9add313 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/v1/segment/SegmentTrieBuffer.java +++ b/src/java/org/apache/cassandra/index/sai/disk/v1/segment/SegmentTrieBuffer.java @@ -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(); } diff --git a/src/java/org/apache/cassandra/index/sai/view/IndexViewManager.java b/src/java/org/apache/cassandra/index/sai/view/IndexViewManager.java index a6237b45dc..5bbaccd691 100644 --- a/src/java/org/apache/cassandra/index/sai/view/IndexViewManager.java +++ b/src/java/org/apache/cassandra/index/sai/view/IndexViewManager.java @@ -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; diff --git a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java index b12806c157..8d8543d25a 100644 --- a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java @@ -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 id = new AtomicReference<>(SSTableIdFactory.instance.defaultBuilder().generator(Stream.empty()).get()); protected boolean makeRangeAware = false; + protected final Collection 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); } diff --git a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java index c164066ba3..7e503829aa 100644 --- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java @@ -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(); * - * + *

* Please note that {@code CQLSSTableWriter} is not 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 boundNames; - private final List typeCodecs; + private final List> typeCodecs; private CQLSSTableWriter(AbstractSSTableSimpleWriter writer, ModificationStatement modificationStatement, List 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 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 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 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 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 typeStatements; + private final List 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. *

@@ -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 * diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index 2debc1adf9..c32b7be2b0 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -324,7 +324,8 @@ public abstract class AbstractReplicationStrategy try { Constructor 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) { diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index ab38720c94..f541eabe54 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -1544,9 +1544,15 @@ public class NodeProbe implements AutoCloseable ssProxy.loadNewSSTables(ksName, cfName); } - public List importNewSSTables(String ksName, String cfName, Set srcPaths, boolean resetLevel, boolean clearRepaired, boolean verifySSTables, boolean verifyTokens, boolean invalidateCaches, boolean extendedVerify, boolean copyData) + public List importNewSSTables(String ksName, String cfName, Set 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) diff --git a/src/java/org/apache/cassandra/tools/nodetool/Import.java b/src/java/org/apache/cassandra/tools/nodetool/Import.java index 73fa3144bb..f64c7893dd 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/Import.java +++ b/src/java/org/apache/cassandra/tools/nodetool/Import.java @@ -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 srcPaths = Lists.newArrayList(args.subList(2, args.size())); - List failedDirs = probe.importNewSSTables(args.get(0), args.get(1), new HashSet<>(srcPaths), !keepLevel, !keepRepaired, !noVerify, !noVerifyTokens, !noInvalidateCaches, extendedVerify, copyData); + List 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; diff --git a/test/unit/org/apache/cassandra/db/ImportTest.java b/test/unit/org/apache/cassandra/db/ImportTest.java index 4a1bd79ce4..8f9c5de6fd 100644 --- a/test/unit/org/apache/cassandra/db/ImportTest.java +++ b/test/unit/org/apache/cassandra/db/ImportTest.java @@ -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 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 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 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 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 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); + } + } diff --git a/test/unit/org/apache/cassandra/index/sai/SAITester.java b/test/unit/org/apache/cassandra/index/sai/SAITester.java index 8c451c05ed..ffa3d6a75e 100644 --- a/test/unit/org/apache/cassandra/index/sai/SAITester.java +++ b/test/unit/org/apache/cassandra/index/sai/SAITester.java @@ -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; diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java index 5ee2768c7b..a402059ed4 100644 --- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java @@ -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 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); } } diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterConcurrencyTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterConcurrencyTest.java index c140049703..1fa406072e 100644 --- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterConcurrencyTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterConcurrencyTest.java @@ -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(); diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterDaemonTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterDaemonTest.java new file mode 100644 index 0000000000..8e96073705 --- /dev/null +++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterDaemonTest.java @@ -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(); + } +} diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java index c5e69fb65d..de592283cc 100644 --- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java @@ -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 values = (List) 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 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 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 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 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(); + } } }