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

Avoid availability gap between UP and queryability marking for already built SAI indexes on bounce 

patch by Caleb Rackliffe; reviewed by David Capwell and Dmitry Konstantinov for CASSANDRA-20732
This commit is contained in:
Caleb Rackliffe 2025-06-26 11:58:33 -05:00
parent 8de4c9250e
commit f553acea3e
7 changed files with 85 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.0.5
* Avoid availability gap between UP and queryability marking for already built SAI indexes on bounce (CASSANDRA-20732)
* Make Commitlog flush data safely in Direct IO mode (CASSANDRA-20692)
* Get SAI MemtableIndex refs before SSTableIndex refs at query time (CASSANDRA-20709)
* Fix MAX_SEGMENT_SIZE < chunkSize in MmappedRegions::updateState (CASSANDRA-20636)

33
src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
View File

@ -331,9 +331,30 @@ public class StorageAttachedIndex implements Index
public Callable<?> getInitializationTask()
{
// New storage-attached indexes will be available for queries after on disk index data are built.
// Memtable data will be indexed via flushing triggered by schema change
// We only want to validate the index files if we are starting up
IndexValidation validation = StorageService.instance.isStarting() ? IndexValidation.HEADER_FOOTER : IndexValidation.NONE;
// Memtable data will be indexed via flushing triggered by schema change.
// We only want to validate the index files if we are starting up.
boolean isStarting = StorageService.instance.isStarting();
IndexValidation validation = isStarting ? IndexValidation.HEADER_FOOTER : IndexValidation.NONE;
// Only attempt to make the index queryable if we are starting up. Otherwise, if we create a new index on top
// of nothing but existing Memtable data (i.e. no SSTables), that data will temporarily be lost until flush.
if (isStarting)
{
StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(baseCfs);
assert indexGroup != null : "Index group does not exist for table " + baseCfs.keyspace + '.' + baseCfs.name;
Collection<SSTableReader> nonIndexed = findNonIndexedSSTables(baseCfs, indexGroup, validation);
if (nonIndexed.isEmpty())
{
// If the index is complete, mark it queryable and avoid an initial build:
baseCfs.indexManager.makeIndexQueryable(this, Status.BUILD_SUCCEEDED);
logger.debug(indexIdentifier.logMessage("Skipping initial build, as index is already queryable..."));
initBuildStarted = true;
return () -> ImmediateFuture.success(null);
}
}
return () -> startInitialBuild(baseCfs, validation).get();
}
@ -843,15 +864,12 @@ public class StorageAttachedIndex implements Index
// Force another flush to make sure on disk index is generated for memtable data before marking it queryable.
// In the case of offline scrub, there are no live memtables.
if (!baseCfs.getTracker().getView().liveMemtables.isEmpty())
{
baseCfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.INDEX_BUILD_STARTED);
}
// It is now safe to flush indexes directly from flushing Memtables.
initBuildStarted = true;
StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(baseCfs);
assert indexGroup != null : "Index group does not exist for table " + baseCfs.keyspace + '.' + baseCfs.name;
List<SSTableReader> nonIndexed = findNonIndexedSSTables(baseCfs, indexGroup, validation);
@ -888,8 +906,7 @@ public class StorageAttachedIndex implements Index
}
StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(baseCfs);
assert indexGroup != null : "Index group does not exist for table";
assert indexGroup != null : "Index group does not exist for table " + baseCfs.keyspace + '.' + baseCfs.name;
Collection<SSTableReader> nonIndexed = findNonIndexedSSTables(baseCfs, indexGroup, IndexValidation.HEADER_FOOTER);

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

@ -2029,6 +2029,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
setMode(Mode.NORMAL, true);
}
@VisibleForTesting
public void setStartingModeUnsafe()
{
setMode(Mode.STARTING, true);
}
private void setMode(Mode m, boolean log)
{
setMode(m, null, log);

8
test/unit/org/apache/cassandra/index/sai/cql/EmptyStringLifecycleTest.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.index.sai.cql;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.cql3.UntypedResultSet;
@ -25,6 +26,13 @@ import org.apache.cassandra.index.sai.SAITester;
public class EmptyStringLifecycleTest extends SAITester
{
@BeforeClass
public static void setup()
{
setUpClass();
requireNetwork(); // Ensure the node has advanced out of STARTING mode
}
@Test
public void testBeforeAndAfterFlush()
{

35
test/unit/org/apache/cassandra/index/sai/cql/StorageAttachedIndexDDLTest.java
View File

@ -76,6 +76,7 @@ import org.apache.cassandra.inject.InvokePointBuilder;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.Throwables;
import org.assertj.core.api.Assertions;
import org.mockito.Mockito;
@ -1311,6 +1312,40 @@ public class StorageAttachedIndexDDLTest extends SAITester
assertTrue(verifyChecksum(numericIndexTermType, numericIndexIdentifier));
}
@Test
public void shouldMarkQueryableInInitializationTask() throws Throwable
{
createTable(CREATE_TABLE_TEMPLATE);
disableCompaction(KEYSPACE);
IndexIdentifier idxIdentifier = createIndexIdentifier(createIndexAsync(String.format(CREATE_INDEX_TEMPLATE, "v1")));
// create 10 SSTables
for (int i = 0; i < 10; i++)
{
execute("INSERT INTO %s (id1, v1, v2) VALUES (?, ?, ?)", String.valueOf(i), i, String.valueOf(i));
flush();
}
ResultSet rows = executeNet("SELECT id1 FROM %s WHERE v1 >= 5");
assertEquals(5, rows.all().size());
// Make the index artificially non-queryable:
ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
Index index = cfs.indexManager.getIndexByName(idxIdentifier.indexName);
cfs.indexManager.makeIndexNonQueryable(index, Index.Status.BUILD_FAILED);
// Query should fail with the index in an artificially non-queryable state:
assertThatThrownBy(() -> executeNet("SELECT id1 FROM %s WHERE v1 >= 5")).isInstanceOf(ReadFailureException.class);
// Node must be in STARTING mode for it to be necessary for the initialization task to pre-emptively validate:
StorageService.instance.setStartingModeUnsafe();
// Simply getting the initialization task (and not running it) will validate and mark the index queryable again:
cfs.indexManager.buildIndex(index);
StorageService.instance.setNormalModeUnsafe();
rows = executeNet("SELECT id1 FROM %s WHERE v1 >= 5");
assertEquals(5, rows.all().size());
}
@Test
public void shouldRejectQueriesWithCustomExpressions()
{

8
test/unit/org/apache/cassandra/index/sai/functional/FailureTest.java
View File

@ -20,6 +20,7 @@
*/
package org.apache.cassandra.index.sai.functional;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.db.marshal.Int32Type;
@ -36,6 +37,13 @@ import static org.junit.Assert.assertEquals;
public class FailureTest extends SAITester
{
@BeforeClass
public static void setup()
{
setUpClass();
requireNetwork(); // Ensure the node has advanced out of STARTING mode
}
@Test
public void shouldMakeIndexNonQueryableOnSSTableContextFailureDuringFlush() throws Throwable
{

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

@ -21,7 +21,6 @@ import com.google.common.collect.ImmutableList;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.virtual.VirtualKeyspace;
import org.apache.cassandra.db.virtual.VirtualKeyspaceRegistry;
@ -60,7 +59,8 @@ public class IndexesSystemViewTest extends SAITester
{
VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(SchemaConstants.VIRTUAL_VIEWS, ImmutableList.of(new ColumnIndexesSystemView(SchemaConstants.VIRTUAL_VIEWS))));
CQLTester.setUpClass();
setUpClass();
requireNetwork(); // Ensure the node has advanced out of STARTING mode
}
@Test