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

Merge branch 'cassandra-5.0' into trunk 

* cassandra-5.0:
  Avoid availability gap between UP and queryability marking for already built SAI indexes on bounce
This commit is contained in:
Caleb Rackliffe 2025-06-27 11:30:46 -05:00
parent d516ea2f0f f553acea3e
commit 1c2307bd62
8 changed files with 95 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -200,6 +200,7 @@
* Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
* Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
Merged from 5.0:
* 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

@ -332,9 +332,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();
}
@ -847,15 +868,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);
@ -892,8 +910,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

@ -682,6 +682,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
initialized = true;
}
@VisibleForTesting
public void unsafeSetUninitialized()
{
initialized = false;
}
public boolean isInitialized()
{
return initialized;

8
test/unit/org/apache/cassandra/index/sai/cql/AllowFilteringTest.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.restrictions.StatementRestrictions;
@ -26,6 +27,7 @@ import org.apache.cassandra.index.sai.SAITester;
import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.inject.Injections;
import org.apache.cassandra.inject.InvokePointBuilder;
import org.apache.cassandra.service.StorageService;
import static java.lang.String.format;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@ -36,6 +38,12 @@ import static org.junit.Assert.assertNotNull;
*/
public class AllowFilteringTest extends SAITester
{
@BeforeClass
public static void setup()
{
StorageService.instance.unsafeSetInitialized();
}
@Test
public void testAllowFilteringOnFirstClusteringKeyColumn() throws Throwable
{

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

@ -18,13 +18,21 @@
package org.apache.cassandra.index.sai.cql;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.index.sai.SAITester;
import org.apache.cassandra.service.StorageService;
public class EmptyStringLifecycleTest extends SAITester
{
@BeforeClass
public static void setup()
{
StorageService.instance.unsafeSetInitialized();
}
@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 not yet be initialized for pre-emptive index validation to occur:
StorageService.instance.unsafeSetUninitialized();
// Simply getting the initialization task (and not running it) will validate and mark the index queryable again:
cfs.indexManager.buildIndex(index);
StorageService.instance.unsafeSetInitialized();
rows = executeNet("SELECT id1 FROM %s WHERE v1 >= 5");
assertEquals(5, rows.all().size());
}
@Test
public void shouldRejectQueriesWithCustomExpressions()
{

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

@ -20,6 +20,8 @@
*/
package org.apache.cassandra.index.sai.functional;
import org.assertj.core.api.Assertions;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.db.marshal.Int32Type;
@ -30,12 +32,18 @@ import org.apache.cassandra.index.sai.utils.IndexIdentifier;
import org.apache.cassandra.index.sai.utils.IndexTermType;
import org.apache.cassandra.inject.Injection;
import org.apache.cassandra.inject.Injections;
import org.assertj.core.api.Assertions;
import org.apache.cassandra.service.StorageService;
import static org.junit.Assert.assertEquals;
public class FailureTest extends SAITester
{
@BeforeClass
public static void setup()
{
StorageService.instance.unsafeSetInitialized();
}
@Test
public void shouldMakeIndexNonQueryableOnSSTableContextFailureDuringFlush() throws Throwable
{

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

@ -29,6 +29,7 @@ import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.inject.Injections;
import org.apache.cassandra.inject.InvokePointBuilder;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.StorageService;
/**
* Tests the virtual table exposing storage-attached column index metadata.
@ -58,6 +59,8 @@ public class IndexesSystemViewTest extends SAITester
public static void setup()
{
VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(SchemaConstants.VIRTUAL_VIEWS, ImmutableList.of(new ColumnIndexesSystemView(SchemaConstants.VIRTUAL_VIEWS))));
StorageService.instance.unsafeSetInitialized();
}
@Test