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

Extend nodetool verify to (optionally) validate SAI files 

patch by Sunil Ramchandra Pawar; reviewed by Caleb Rackliffe and David Capwell for CASSANDRA-20949
This commit is contained in:
Sunil Ramchandra Pawar 2025-10-14 23:01:56 +05:30 committed by Caleb Rackliffe
parent ba9790a6f0
commit a84abe7ba1
9 changed files with 292 additions and 18 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.1
* Extend nodetool verify to (optionally) validate SAI files (CASSANDRA-20949)
* Fix CompressionDictionary being closed while still in use (CASSANDRA-21047)
* When updating a multi cell collection element, if the update is rejected then the shared Row.Builder is not freed causing all future mutations to be rejected (CASSANDRA-21055)
* Schema annotations escape validation on CREATE and ALTER DDL statements (CASSANDRA-21046)

46
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -89,6 +89,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.index.SecondaryIndexBuilder;
import org.apache.cassandra.index.sai.StorageAttachedIndexGroup;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.IScrubber;
@ -669,6 +670,15 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan
public AllSSTableOpStatus performVerify(ColumnFamilyStore cfs, IVerifier.Options options) throws InterruptedException, ExecutionException
{
assert !cfs.isIndex();
StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(cfs);
boolean skipSaiCheck = indexGroup == null;
if (options.onlySai && skipSaiCheck)
{
logger.info("Skipping table {} during SAI-only verify because it has no SAI indexes.", cfs.getTableName());
return AllSSTableOpStatus.SUCCESSFUL;
}
return parallelAllSSTableOperation(cfs, new OneSSTableOperation()
{
@Override
@ -1493,17 +1503,37 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan
@VisibleForTesting
void verifyOne(ColumnFamilyStore cfs, SSTableReader sstable, IVerifier.Options options, ActiveCompactionsTracker activeCompactions)
{
CompactionInfo.Holder verifyInfo = null;
try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, options))
StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(cfs);
boolean skipSaiCheck = indexGroup == null;
// Skip early if no SAI indexes on table
if (options.onlySai && skipSaiCheck)
{
verifyInfo = verifier.getVerifyInfo();
activeCompactions.beginCompaction(verifyInfo);
verifier.verify();
logger.info("Skipping SAI validation for table {} because it has no SAI indexes.", cfs.getTableName());
return;
}
finally
CompactionInfo.Holder verifyInfo = null;
if (!options.onlySai)
{
if (verifyInfo != null)
activeCompactions.finishCompaction(verifyInfo);
try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, options))
{
verifyInfo = verifier.getVerifyInfo();
activeCompactions.beginCompaction(verifyInfo);
verifier.verify();
}
finally
{
if (verifyInfo != null)
activeCompactions.finishCompaction(verifyInfo);
}
}
if ((options.onlySai || options.includeSai) && !skipSaiCheck)
{
cfs.indexManager.validateSSTableAttachedIndexes(Collections.singleton(sstable), true, true);
}
}

35
src/java/org/apache/cassandra/io/sstable/IVerifier.java
View File

@ -60,6 +60,16 @@ public interface IVerifier extends Closeable
*/
public final boolean quick;
/**
* To verify only SAI checksum
*/
public final boolean onlySai;
/**
* To include SAI verification along with data files
*/
public final boolean includeSai;
public final Function<String, ? extends Collection<Range<Token>>> tokenLookup;
private Options(boolean invokeDiskFailurePolicy,
@ -68,14 +78,21 @@ public interface IVerifier extends Closeable
boolean mutateRepairStatus,
boolean checkOwnsTokens,
boolean quick,
boolean onlySai,
boolean includeSai,
Function<String, ? extends Collection<Range<Token>>> tokenLookup)
{
if (onlySai && includeSai)
throw new IllegalArgumentException("onlySai and includeSai both cannot be true at a time.");
this.invokeDiskFailurePolicy = invokeDiskFailurePolicy;
this.extendedVerification = extendedVerification;
this.checkVersion = checkVersion;
this.mutateRepairStatus = mutateRepairStatus;
this.checkOwnsTokens = checkOwnsTokens;
this.quick = quick;
this.onlySai = onlySai;
this.includeSai = includeSai;
this.tokenLookup = tokenLookup;
}
@ -89,6 +106,8 @@ public interface IVerifier extends Closeable
", mutateRepairStatus=" + mutateRepairStatus +
", checkOwnsTokens=" + checkOwnsTokens +
", quick=" + quick +
", onlySai=" + onlySai +
", includeSai=" + includeSai +
'}';
}
@ -100,6 +119,8 @@ public interface IVerifier extends Closeable
private boolean mutateRepairStatus = false; // mutating repair status can be dangerous
private boolean checkOwnsTokens = false;
private boolean quick = false;
private boolean onlySai = false;
private boolean includeSai = false;
private Function<String, ? extends Collection<Range<Token>>> tokenLookup = StorageService.instance::getLocalAndPendingRanges;
public Builder invokeDiskFailurePolicy(boolean param)
@ -138,6 +159,18 @@ public interface IVerifier extends Closeable
return this;
}
public Builder onlySai(boolean param)
{
this.onlySai = param;
return this;
}
public Builder includeSai(boolean param)
{
this.includeSai = param;
return this;
}
public Builder tokenLookup(Function<String, ? extends Collection<Range<Token>>> tokenLookup)
{
this.tokenLookup = tokenLookup;
@ -146,7 +179,7 @@ public interface IVerifier extends Closeable
public Options build()
{
return new Options(invokeDiskFailurePolicy, extendedVerification, checkVersion, mutateRepairStatus, checkOwnsTokens, quick, tokenLookup);
return new Options(invokeDiskFailurePolicy, extendedVerification, checkVersion, mutateRepairStatus, checkOwnsTokens, quick, onlySai, includeSai, tokenLookup);
}
}
}

14
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -2690,10 +2690,18 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
@Deprecated(since = "4.0")
public int verify(boolean extendedVerify, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException
{
return verify(extendedVerify, false, false, false, false, false, keyspaceName, tableNames);
return verify(extendedVerify, false, false, false, false, false, false, false, keyspaceName, tableNames);
}
/**
* Kept for backward compatibility with existing clients.
*/
public int verify(boolean extendedVerify, boolean checkVersion, boolean diskFailurePolicy, boolean mutateRepairStatus, boolean checkOwnsTokens, boolean quick, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException
{
return verify(extendedVerify, checkVersion, diskFailurePolicy, mutateRepairStatus, checkOwnsTokens, quick, false, false, keyspaceName, tableNames);
}
public int verify(boolean extendedVerify, boolean checkVersion, boolean diskFailurePolicy, boolean mutateRepairStatus, boolean checkOwnsTokens, boolean quick, boolean onlySai, boolean includeSai, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException
{
CompactionManager.AllSSTableOpStatus status = CompactionManager.AllSSTableOpStatus.SUCCESSFUL;
IVerifier.Options options = IVerifier.options().invokeDiskFailurePolicy(diskFailurePolicy)
@ -2701,7 +2709,9 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
.checkVersion(checkVersion)
.mutateRepairStatus(mutateRepairStatus)
.checkOwnsTokens(checkOwnsTokens)
.quick(quick).build();
.quick(quick)
.onlySai(onlySai)
.includeSai(includeSai).build();
logger.info("Staring {} on {}.{} with options = {}", OperationType.VERIFY, keyspaceName, Arrays.toString(tableNames), options);
for (ColumnFamilyStore cfStore : getValidColumnFamilies(false, false, keyspaceName, tableNames))
{

9
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -452,8 +452,17 @@ public interface StorageServiceMBean extends NotificationEmitter
* The entire sstable will be read to ensure each cell validates if extendedVerify is true
*/
public int verify(boolean extendedVerify, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException;
/**
* Kept for backward compatibility with existing clients.
*/
public int verify(boolean extendedVerify, boolean checkVersion, boolean diskFailurePolicy, boolean mutateRepairStatus, boolean checkOwnsTokens, boolean quick, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException;
/**
* Verify checksums of the given keyspace with extended options including SAI index validation.
*/
public int verify(boolean extendedVerify, boolean checkVersion, boolean diskFailurePolicy, boolean mutateRepairStatus, boolean checkOwnsTokens, boolean quick, boolean onlySai, boolean includeSai, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException;
/**
* Rewrite all sstables to the latest version.
* Unlike scrub, it doesn't skip bad rows and do not snapshot sstables first.

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

@ -391,9 +391,9 @@ public class NodeProbe implements AutoCloseable
return ssProxy.scrub(disableSnapshot, skipCorrupted, checkData, reinsertOverflowedTTL, jobs, keyspaceName, tables);
}
public int verify(boolean extendedVerify, boolean checkVersion, boolean diskFailurePolicy, boolean mutateRepairStatus, boolean checkOwnsTokens, boolean quick, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException
public int verify(boolean extendedVerify, boolean checkVersion, boolean diskFailurePolicy, boolean mutateRepairStatus, boolean checkOwnsTokens, boolean quick, boolean onlySai, boolean includeSai, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException
{
return ssProxy.verify(extendedVerify, checkVersion, diskFailurePolicy, mutateRepairStatus, checkOwnsTokens, quick, keyspaceName, tableNames);
return ssProxy.verify(extendedVerify, checkVersion, diskFailurePolicy, mutateRepairStatus, checkOwnsTokens, quick, onlySai, includeSai, keyspaceName, tableNames);
}
public int upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion, long maxSSTableTimestamp, int jobs, String... tableNames) throws IOException, ExecutionException, InterruptedException
@ -434,10 +434,10 @@ public class NodeProbe implements AutoCloseable
"scrubbing");
}
public void verify(PrintStream out, boolean extendedVerify, boolean checkVersion, boolean diskFailurePolicy, boolean mutateRepairStatus, boolean checkOwnsTokens, boolean quick, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException
public void verify(PrintStream out, boolean extendedVerify, boolean checkVersion, boolean diskFailurePolicy, boolean mutateRepairStatus, boolean checkOwnsTokens, boolean quick, boolean onlySai, boolean includeSai, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException
{
perform(out, keyspaceName,
() -> verify(extendedVerify, checkVersion, diskFailurePolicy, mutateRepairStatus, checkOwnsTokens, quick, keyspaceName, tableNames),
() -> verify(extendedVerify, checkVersion, diskFailurePolicy, mutateRepairStatus, checkOwnsTokens, quick, onlySai, includeSai, keyspaceName, tableNames),
"verifying");
}

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

@ -78,6 +78,16 @@ public class Verify extends AbstractCommand
description = "Do a quick check - avoid reading all data to verify checksums")
private boolean quick = false;
@Option(paramLabel = "sai_only",
names = { "-s", "--sai-only"},
description = "Verify only sai index")
private boolean onlySai = false;
@Option(paramLabel = "include_sai",
names = { "-i", "--include-sai"},
description = "Include SAI index verification along with data files")
private boolean includeSai = false;
@Override
public void execute(NodeProbe probe)
{
@ -89,6 +99,9 @@ public class Verify extends AbstractCommand
System.exit(1);
}
if (onlySai && includeSai)
throw new IllegalArgumentException("Cannot specify both --sai-only and --include-sai");
List<String> keyspaces = parseOptionalKeyspace(args, probe);
String[] tableNames = parseOptionalTables(args);
@ -103,7 +116,7 @@ public class Verify extends AbstractCommand
{
try
{
probe.verify(out, extendedVerify, checkVersion, diskFailurePolicy, mutateRepairStatus, checkOwnsTokens, quick, keyspace, tableNames);
probe.verify(out, extendedVerify, checkVersion, diskFailurePolicy, mutateRepairStatus, checkOwnsTokens, quick, onlySai, includeSai, keyspace, tableNames);
} catch (Exception e)
{
throw new RuntimeException("Error occurred during verifying", e);

11
test/resources/nodetool/help/verify
View File

@ -7,8 +7,9 @@ SYNOPSIS
[(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]
[(-u <username> | --username <username>)] verify
[(-c | --check-version)] [(-d | --dfp)] [(-e | --extended-verify)]
[(-f | --force)] [(-q | --quick)] [(-r | --rsc)]
[(-t | --check-tokens)] [--] [<keyspace> <tables>...]
[(-f | --force)] [(-i | --include-sai)] [(-q | --quick)] [(-r | --rsc)]
[(-s | --sai-only)] [(-t | --check-tokens)] [--] [<keyspace>
<tables>...]
OPTIONS
-c, --check-version
@ -26,6 +27,9 @@ OPTIONS
-h <host>, --host <host>
Node hostname or ip address
-i, --include-sai
Include SAI index verification along with data files
-p <port>, --port <port>
Remote jmx agent port number
@ -44,6 +48,9 @@ OPTIONS
-r, --rsc
Mutate the repair status on corrupt sstables
-s, --sai-only
Verify only sai index
-t, --check-tokens
Verify that all tokens in sstables are owned by this node

171
test/unit/org/apache/cassandra/tools/nodetool/VerifyTest.java Normal file
View File

@ -0,0 +1,171 @@
/*
* 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.tools.nodetool;
import java.io.IOException;
import java.nio.channels.FileChannel;
import java.util.Set;
import java.util.stream.Collectors;
import com.google.common.collect.Iterables;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.index.sai.StorageAttachedIndexGroup;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.tools.ToolRunner;
import static org.apache.cassandra.tools.ToolRunner.invokeNodetool;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
public class VerifyTest extends CQLTester
{
@BeforeClass
public static void setup() throws Throwable
{
requireNetwork();
startJMXServer();
}
@Test
public void testVerifyDefault()
{
createTable("CREATE TABLE %s (pk int, ck int, a int, b int, PRIMARY KEY(pk,ck))");
createIndex("CREATE INDEX idx1 ON %s(a) USING 'sai'");
for (int i = 0; i < 10; i++)
{
execute("INSERT INTO %s (pk, ck, a, b) VALUES (?, ?, ?, ?)", i, i * 2, i * 3, i * 4);
}
flush(keyspace());
invokeNodetool("verify", "--force", keyspace(), currentTable()).assertOnCleanExit();
}
@Test
public void testVerifySaiOnly()
{
createTable("CREATE TABLE %s (pk int, ck int, a int, b int, PRIMARY KEY (pk, ck))");
createIndex("CREATE INDEX idx1 ON %s(a) USING 'sai'");
for (int i = 0; i < 10; i++)
{
execute("INSERT INTO %s (pk, ck, a, b) VALUES (?, ?, ?, ?)", i, i * 2, i * 3, i * 4);
}
flush(keyspace());
// SAI-only mode: verify only SAI components
invokeNodetool("verify", "--force", "--sai-only", keyspace(), currentTable()).assertOnCleanExit();
}
@Test
public void testVerifyIncludeSai()
{
createTable("CREATE TABLE %s (pk int, ck int, a int, b int, PRIMARY KEY (pk, ck))");
createIndex("CREATE INDEX idx1 ON %s(a) USING 'sai'");
for (int i = 0; i < 10; i++)
{
execute("INSERT INTO %s (pk, ck, a, b) VALUES (?, ?, ?, ?)", i, i * 2, i * 3, i * 4);
}
flush(keyspace());
// Include-sai mode: verify both data files and SAI components
invokeNodetool("verify", "--force", "--include-sai", keyspace(), currentTable()).assertOnCleanExit();
}
@Test
public void testVerifySaiOnlyWithoutSaiIndex()
{
createTable("CREATE TABLE %s (pk int, ck int, a int, b int, PRIMARY KEY (pk, ck))");
// No SAI index created
for (int i = 0; i < 10; i++)
{
execute("INSERT INTO %s (pk, ck, a, b) VALUES (?, ?, ?, ?)", i, i * 2, i * 3, i * 4);
}
flush(keyspace());
// SAI-only mode on table without SAI should succeed (skipped)
invokeNodetool("verify", "--force", "--sai-only", keyspace(), currentTable()).assertOnCleanExit();
}
@Test
public void testVerifyConflictingFlags()
{
createTable("CREATE TABLE %s (pk int, ck int, a int, b int, PRIMARY KEY (pk, ck))");
// Both --sai-only and --include-sai should fail
ToolRunner.ToolResult result = invokeNodetool("verify", "--force", "--sai-only", "--include-sai", keyspace(), currentTable());
result.asserts().failure();
result.getStdout().contains("Cannot specify both --sai-only and --include-sai");
}
@Test
public void testVerifySaiWithCorruption ()
{
createTable("CREATE TABLE %s (pk int, ck int, a int, b int, PRIMARY KEY (pk, ck))");
createIndex("CREATE INDEX idx1 ON %s(a) USING 'sai'");
for (int i = 0; i < 10; i++)
{
execute("INSERT INTO %s (pk, ck, a, b) VALUES (?, ?, ?, ?)", i, i * 2, i * 3, i * 4);
}
flush(keyspace());
ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
StorageAttachedIndexGroup group = StorageAttachedIndexGroup.getIndexGroup(cfs);
assertNotNull("SAI group should exist", group);
SSTableReader sstable = Iterables.getOnlyElement(cfs.getLiveSSTables());
Set<Component> saiComponents = StorageAttachedIndexGroup.getLiveComponents(sstable,
group.getIndexes().stream().map(i -> (StorageAttachedIndex)i).collect(Collectors.toSet()));
assertFalse("SAI components should exist", saiComponents.isEmpty());
Component componentToCorrupt = saiComponents.iterator().next();
File saiFile = sstable.descriptor.fileFor(componentToCorrupt);
assertTrue("SAI file should exist", saiFile.exists());
try (FileChannel channel = saiFile.newReadWriteChannel())
{
channel.truncate(10);
}
catch (IOException e)
{
throw new RuntimeException(e);
}
invokeNodetool("verify", "--force", keyspace(), currentTable()).asserts().success();
invokeNodetool("verify", "--force", "--sai-only", keyspace(), currentTable()).asserts().failure().errorContains("file truncated");
invokeNodetool("verify", "--force", "--include-sai", keyspace(), currentTable()).asserts().failure().errorContains("file truncated");
}
}