You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "sunil9977 (via GitHub)" <gi...@apache.org> on 2026/04/24 04:34:59 UTC
[PR] Migrate index status propagation from gossip to table polling [cassandra]
sunil9977 opened a new pull request, #4769:
URL: https://github.com/apache/cassandra/pull/4769
Cassandra-21264
1. Migrates index status propagation from gossip to table polling to avoid the 65KB writeUTF() serialization limit.
2. Adds two new system_distributed tables (index_build_status, index_events) with periodic polling
3. version-gated so pre-6.0 clusters continue using gossip.
patch by Sunil Ramchadnra Pawar; reviewed by Caleb Rackliffe for CASSANDRA-21264
Co-authored-by: Caleb Rackliffe
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Migrate index status propagation from gossip to table polling [cassandra]
Posted by "krummas (via GitHub)" <gi...@apache.org>.
krummas commented on code in PR #4769:
URL: https://github.com/apache/cassandra/pull/4769#discussion_r3880001781
##########
src/java/org/apache/cassandra/index/IndexStatusManager.java:
##########
@@ -344,8 +407,133 @@ private String identifier(String keyspace, String index)
return keyspace + '.' + index;
}
+ /**
+ * Load index statuses from the system_distributed.index_build_status table on startup
+ * so that index statuses are known before gossip starts.
+ */
+ public synchronized void loadIndexStatusesFromTable()
Review Comment:
could we merge this + `refreshFromFullTable` below?
##########
src/java/org/apache/cassandra/index/IndexStatusManager.java:
##########
@@ -227,28 +252,55 @@ public synchronized void propagateLocalIndexStatus(String keyspace, String index
{
try
{
- Map<String, Index.Status> statusMap = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(),
- k -> new HashMap<>());
+ NodeId localNodeId = ClusterMetadata.current().myNodeId();
+ if (localNodeId == null)
Review Comment:
this is not nullable anymore, check against `NodeId.UNREGISTERED` if the local node is not yet registered (edit, I guess it still is nullable, but after your next rebase it wont)
##########
src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java:
##########
@@ -414,50 +444,175 @@ public static void setViewRemoved(String keyspaceName, String viewName)
forceBlockingFlush(VIEW_BUILD_STATUS, ColumnFamilyStore.FlushReason.INTERNALLY_FORCED);
}
+ public static void updateIndexStatus(NodeId nodeId, String keyspace, String index, Index.Status status)
+ {
+ String query = "INSERT INTO %s.%s (node_id, keyspace_name, index_name, status) VALUES (?, ?, ?, ?)";
+ try
+ {
+ QueryProcessor.execute(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS),
+ ConsistencyLevel.QUORUM, nodeId.id(), keyspace, index, status.toString());
+ }
+ catch (Exception e)
+ {
+ logger.warn("Failed to update index status with QUORUM for {}.{}, retrying with ONE", keyspace, index);
+ QueryProcessor.execute(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS),
+ ConsistencyLevel.ONE, nodeId.id(), keyspace, index, status.toString());
+ }
+ }
+
+ public static void setIndexRemoved(NodeId nodeId, String keyspaceName, String indexName)
+ {
+ String buildReq = format("DELETE FROM %s.%s WHERE node_id = ? AND keyspace_name = ? AND index_name = ?",
+ SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS);
+ try
+ {
+ QueryProcessor.execute(buildReq, ConsistencyLevel.QUORUM, nodeId.id(), keyspaceName, indexName);
+ }
+ catch (Exception e)
+ {
+ logger.warn("Failed to remove index status with QUORUM for {}.{}, retrying with ONE", keyspaceName, indexName);
+ QueryProcessor.execute(buildReq, ConsistencyLevel.ONE, nodeId.id(), keyspaceName, indexName);
+ }
+ forceBlockingFlush(INDEX_BUILD_STATUS, ColumnFamilyStore.FlushReason.INTERNALLY_FORCED);
+ }
+
+ public static Map<NodeId, Map<String, Index.Status>> allIndexStatuses()
+ {
+ String query = format("SELECT node_id, keyspace_name, index_name, status FROM %s.%s",
+ SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS);
+ UntypedResultSet results;
+
+ try
+ {
+ try
+ {
+ results = QueryProcessor.execute(query, ConsistencyLevel.QUORUM);
+ }
+ catch (Exception e)
+ {
+ logger.warn("Failed to load index statuses with QUORUM, retrying with ONE");
+ results = QueryProcessor.execute(query, ConsistencyLevel.ONE);
+ }
+ }
+ catch (Exception e)
+ {
+ logger.warn("Unable to load index statuses from system table: {}", e.getMessage());
+ return Collections.emptyMap();
+ }
+
+ Map<NodeId, Map<String, Index.Status>> allStatuses = new HashMap<>();
+ for (UntypedResultSet.Row row : results)
+ {
+ NodeId nodeId = new NodeId(row.getInt("node_id"));
+ String identifier = row.getString("keyspace_name") + '.' + row.getString("index_name");
+ Index.Status status = Index.Status.valueOf(row.getString("status"));
+ allStatuses.computeIfAbsent(nodeId, k -> new HashMap<>()).put(identifier, status);
+ }
+
+ return allStatuses;
+ }
+
+ public static Map<String, Index.Status> allIndexStatusesForHost(NodeId nodeId)
Review Comment:
unused
##########
src/java/org/apache/cassandra/index/IndexStatusManager.java:
##########
@@ -344,8 +407,133 @@ private String identifier(String keyspace, String index)
return keyspace + '.' + index;
}
+ /**
+ * Load index statuses from the system_distributed.index_build_status table on startup
+ * so that index statuses are known before gossip starts.
+ */
+ public synchronized void loadIndexStatusesFromTable()
+ {
+ try
+ {
+ Map<NodeId, Map<String, Index.Status>> allStatuses = SystemDistributedKeyspace.allIndexStatuses();
+ for (Map.Entry<NodeId, Map<String, Index.Status>> entry : allStatuses.entrySet())
+ {
+ peerIndexStatus.putIfAbsent(entry.getKey(), entry.getValue());
+ }
+ logger.info("Loaded index statuses from system table for {} peers", allStatuses.size());
+ }
+ catch (Exception e)
+ {
+ logger.warn("Unable to load index statuses from system table: {}", e.getMessage());
+ }
+ }
+
+ /**
+ * Refresh index statuses from the full table, overwriting any existing data.
+ */
+ public synchronized void refreshFromFullTable()
+ {
+ try
+ {
+ Map<NodeId, Map<String, Index.Status>> allStatuses = SystemDistributedKeyspace.allIndexStatuses();
Review Comment:
I think this also needs to clear the old map? Otherwise dropped indexes might still be in there
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Migrate index status propagation from gossip to table polling [cassandra]
Posted by "krummas (via GitHub)" <gi...@apache.org>.
krummas commented on code in PR #4769:
URL: https://github.com/apache/cassandra/pull/4769#discussion_r3335808574
##########
src/java/org/apache/cassandra/index/IndexStatusManager.java:
##########
@@ -230,29 +257,65 @@ public synchronized void propagateLocalIndexStatus(String keyspace, String index
Map<String, Index.Status> statusMap = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(),
k -> new HashMap<>());
String keyspaceIndex = identifier(keyspace, index);
+ UUID localHostId = StorageService.instance.getLocalHostUUID();
Review Comment:
we should not use host id anymore, it should be using NodeId (`ClusterMetadata.current().myNodeId()`)
##########
src/java/org/apache/cassandra/index/IndexStatusManager.java:
##########
@@ -230,29 +257,65 @@ public synchronized void propagateLocalIndexStatus(String keyspace, String index
Map<String, Index.Status> statusMap = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(),
k -> new HashMap<>());
String keyspaceIndex = identifier(keyspace, index);
+ UUID localHostId = StorageService.instance.getLocalHostUUID();
+ CassandraVersion minVersion = ClusterMetadata.current().directory.clusterMinVersion.cassandraVersion;
if (status == Index.Status.DROPPED)
+ {
statusMap.remove(keyspaceIndex);
+ if (localHostId != null)
+ statusPropagationExecutor.submit(() -> {
Review Comment:
we don't need to update the system tables async
##########
src/java/org/apache/cassandra/index/IndexStatusManager.java:
##########
@@ -344,8 +427,136 @@ private String identifier(String keyspace, String index)
return keyspace + '.' + index;
}
+ /**
+ * Load index statuses from the system_distributed.index_build_status table on startup
+ * so that index statuses are known before gossip starts.
+ */
+ public synchronized void loadIndexStatusesFromTable()
+ {
+ try
+ {
+ Map<UUID, Map<String, Index.Status>> allStatuses = SystemDistributedKeyspace.allIndexStatuses();
+ for (Map.Entry<UUID, Map<String, Index.Status>> entry : allStatuses.entrySet())
+ {
+ InetAddressAndPort endpoint = StorageService.instance.getEndpointForHostId(entry.getKey());
+ if (endpoint == null)
+ continue;
+
+ peerIndexStatus.putIfAbsent(endpoint, entry.getValue());
+ }
+ logger.info("Loaded index statuses from system table for {} peers", allStatuses.size());
+ }
+ catch (Exception e)
+ {
+ logger.warn("Unable to load index statuses from system table: {}", e.getMessage());
+ }
+ }
+
+ /**
+ * Refresh index statuses from the full table, overwriting any existing data.
+ */
+ public synchronized void refreshFromFullTable()
+ {
+ try
+ {
+ Map<UUID, Map<String, Index.Status>> allStatuses = SystemDistributedKeyspace.allIndexStatuses();
+ for (Map.Entry<UUID, Map<String, Index.Status>> entry : allStatuses.entrySet())
+ {
+ InetAddressAndPort endpoint = StorageService.instance.getEndpointForHostId(entry.getKey());
+ if (endpoint == null)
+ continue;
+
+ peerIndexStatus.put(endpoint, entry.getValue());
+ }
+ logger.info("Refreshed index statuses from system table for {} peers", allStatuses.size());
+ }
+ catch (Exception e)
+ {
+ logger.warn("Unable to refresh index statuses from system table: {}", e.getMessage());
+ }
+ }
+
public void shutdownAndWait(long interval, TimeUnit unit) throws InterruptedException, TimeoutException
{
+ if (pollFuture != null)
+ pollFuture.cancel(false);
+
ExecutorUtils.shutdownAndWait(interval, unit, statusPropagationExecutor);
}
+
+ public void startPolling()
+ {
+ int intervalSeconds = DatabaseDescriptor.getIndexStatusPollInterval();
+ if (intervalSeconds <= 0)
+ return;
+
+ pollFuture = ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(
+ this::pollIndexEvents,
+ intervalSeconds,
+ intervalSeconds,
+ TimeUnit.SECONDS);
+ }
+
+ private synchronized void pollIndexEvents()
+ {
+ if (lastPollTimestampMillis == 0)
+ {
+ refreshFromFullTable();
+ lastPollTimestampMillis = Clock.Global.currentTimeMillis();
+ }
+ else
+ {
+ try
+ {
+ String today = LocalDate.now(ZoneOffset.UTC).toString();
+ String lastPollDate = Instant.ofEpochMilli(lastPollTimestampMillis).atZone(ZoneOffset.UTC).toLocalDate().toString();
+
+ if (!today.equals(lastPollDate))
+ {
+ // midnight crossed so get remaining events from last day.
+ UntypedResultSet yesterdayResults = SystemDistributedKeyspace.queryIndexEvents(lastPollDate, lastPollTimestampMillis);
+ if (yesterdayResults != null)
+ processEvents(yesterdayResults);
+ }
+
+ UntypedResultSet todayResults = SystemDistributedKeyspace.queryIndexEvents(today, lastPollTimestampMillis);
+ if (todayResults != null)
+ processEvents(todayResults);
+
+ lastPollTimestampMillis = Clock.Global.currentTimeMillis();
Review Comment:
I think we need to update this before querying the distributed table, otherwise we might miss any events that came after the last query but before this update
Actually we should probably use the timestamp of the newest event seen in the event log?
##########
src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java:
##########
@@ -414,50 +443,148 @@ public static void setViewRemoved(String keyspaceName, String viewName)
forceBlockingFlush(VIEW_BUILD_STATUS, ColumnFamilyStore.FlushReason.INTERNALLY_FORCED);
}
+ public static void updateIndexStatus(UUID hostId, String keyspace, String index, Index.Status status)
+ {
+ String query = "INSERT INTO %s.%s (host_id, keyspace_name, index_name, status) VALUES (?, ?, ?, ?)";
+ QueryProcessor.process(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS),
+ ConsistencyLevel.ONE,
Review Comment:
we should probably read/write with QUORUM to make sure we don't miss any events
for the `index_build_status` table we can probably downgrade to ONE if it fails?
##########
test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java:
##########
@@ -401,6 +406,102 @@ public int hashCode()
}
}
+ @Test
+ public void verifyIndexStatusPropagationViaTablePolling() throws Exception
+ {
+ try (Cluster cluster = init(Cluster.build(2)
+ .withConfig(config -> config.with(GOSSIP).with(NETWORK)
+ .set("index_status_poll_interval_in_seconds", 5)).start()))
+ {
+ String ks = "poll_test_ks";
+ String cf = "cf1";
+ String index = "cf1_poll_idx";
+
+ cluster.schemaChange(String.format(CREATE_KEYSPACE, ks, 2));
+ cluster.schemaChange(String.format(CREATE_TABLE, ks, cf));
+ cluster.schemaChange(String.format(CREATE_INDEX, index, ks, cf, "v1"));
+ waitForIndexQueryable(cluster, ks);
+
+ await().atMost(15, TimeUnit.SECONDS)
+ .until(() -> cluster.get(2).callOnInstance(() -> {
+ InetAddressAndPort node1Address = InetAddressAndPort.getByNameUnchecked("127.0.0.1");
+ int port = FBUtilities.getBroadcastAddressAndPort().getPort();
+ InetAddressAndPort node1 = InetAddressAndPort.getByAddressOverrideDefaults(node1Address.getAddress(), port);
+ return IndexStatusManager.instance.getIndexStatus(node1, ks, index) == Index.Status.BUILD_SUCCEEDED;
+ }));
+ }
+ }
+
+
+ @Test
+ public void verifyMaxSizeIndexTest() throws Exception
+ {
+ try (Cluster cluster = init(Cluster.build(1)
+ .withConfig(config -> config.with(GOSSIP).with(NETWORK))
+ .withInstanceInitializer(MixedPatchVersionHelper::setVersions)
+ .start()))
+ {
+ LogAction logs = cluster.get(1).logs();
+ long mark = logs.mark();
+
+ cluster.get(1).runOnInstance(() -> {
+ Map<String, Index.Status> localStatusMap =
+ IndexStatusManager.instance.peerIndexStatus
+ .computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(), k -> new HashMap<>());
+
+ for (int ks = 0; ks < 100; ks++)
+ for (int idx = 0; idx < 200; idx++)
+ localStatusMap.put("keyspace_" + ks + ".my_table_index_name" + idx, Index.Status.BUILD_SUCCEEDED);
+
+ IndexStatusManager.instance.propagateLocalIndexStatus("keyspace_trigger", "trigger_idx", Index.Status.BUILD_SUCCEEDED);
+ });
+
+ assertFalse(logs.watchFor(mark, "exceeds limit").getResult().isEmpty());
+ }
+ }
+
+
+ @Test
+ public void verifyMixedVersionSkipsTableWritesAndUsesGossip() throws Exception
+ {
+ try (Cluster cluster = init(Cluster.build(2)
+ .withConfig(config -> config.with(GOSSIP).with(NETWORK))
+ .withInstanceInitializer(MixedMajorVersionHelper::setVersions)
+ .start()))
+ {
+ String ks = "mixed_ks";
+ String cf = "cf1";
+ String index1 = "cf1_idx1";
+
+ cluster.schemaChange(String.format(CREATE_KEYSPACE, ks, 2));
+ cluster.schemaChange(String.format(CREATE_TABLE, ks, cf));
+ cluster.schemaChange(String.format(CREATE_INDEX, index1, ks, cf, "v1"));
+ waitForIndexQueryable(cluster, ks);
+
+ waitForIndexingStatus(cluster.get(2), ks, index1, cluster.get(1), Index.Status.BUILD_SUCCEEDED);
+
+ cluster.get(1).runOnInstance(() -> {
+ java.util.Map<java.util.UUID, java.util.Map<String, Index.Status>> allStatuses =
Review Comment:
nit, no need for fully qualified class names
##########
test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java:
##########
Review Comment:
needs `.set("index_status_poll_interval_in_seconds", "1")`
##########
src/java/org/apache/cassandra/index/IndexStatusManager.java:
##########
Review Comment:
I think we have to key this by NodeId now instead, because of ip changes - if a node changes ip address we'd lose its index status I think
##########
src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java:
##########
@@ -201,19 +208,41 @@ private SystemDistributedKeyspace()
public static final String COMPRESSION_DICTIONARIES_CQL = "CREATE TABLE IF NOT EXISTS %s (" +
"keyspace_name text," +
"table_name text," +
- "table_id text," +
"kind text," +
"dict_id bigint," +
"dict blob," +
"dict_length int," +
"dict_checksum int," +
- "created_at timestamp," +
- "PRIMARY KEY ((keyspace_name, table_name), table_id, dict_id)) " +
- "WITH CLUSTERING ORDER BY (table_id DESC, dict_id DESC)"; // in order to retrieve the latest dictionary; the contract is the newer the dictionary the larger the dict_id
+ "PRIMARY KEY ((keyspace_name, table_name), dict_id)) " +
+ "WITH CLUSTERING ORDER BY (dict_id DESC)"; // in order to retrieve the latest dictionary; the contract is the newer the dictionary the larger the dict_id
private static final TableMetadata CompressionDictionariesTable =
parse(COMPRESSION_DICTIONARIES, "Compression dictionaries for applicable tables", COMPRESSION_DICTIONARIES_CQL).build();
+ public static final String INDEX_BUILD_STATUS_CQL = "CREATE TABLE IF NOT EXISTS %s (" +
+ "host_id uuid," +
Review Comment:
`node_id int` instead of host id for both new tables
##########
src/java/org/apache/cassandra/index/IndexStatusManager.java:
##########
@@ -265,12 +328,32 @@ public synchronized void propagateLocalIndexStatus(String keyspace, String index
private static boolean shouldWriteLegacyStatusFormat(CassandraVersion minVersion)
{
- if (DatabaseDescriptor.getForceOptimizedIndexStatusFormat())
- return false;
-
return minVersion == null || (minVersion.major == 5 && minVersion.minor == 0 && minVersion.patch < 3);
}
+ private static boolean shouldWriteToIndexTables(CassandraVersion minVersion)
+ {
+ return minVersion != null && (minVersion.major >= 6);
+ }
+
+ @VisibleForTesting
+ static boolean shouldWriteToIndexTablesForTesting(CassandraVersion minVersion)
+ {
+ return shouldWriteToIndexTables(minVersion);
+ }
+
+ @VisibleForTesting
+ void processEventsForTesting(UntypedResultSet results)
Review Comment:
unused
##########
src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java:
##########
@@ -201,19 +208,41 @@ private SystemDistributedKeyspace()
public static final String COMPRESSION_DICTIONARIES_CQL = "CREATE TABLE IF NOT EXISTS %s (" +
"keyspace_name text," +
"table_name text," +
- "table_id text," +
"kind text," +
"dict_id bigint," +
"dict blob," +
"dict_length int," +
"dict_checksum int," +
- "created_at timestamp," +
- "PRIMARY KEY ((keyspace_name, table_name), table_id, dict_id)) " +
- "WITH CLUSTERING ORDER BY (table_id DESC, dict_id DESC)"; // in order to retrieve the latest dictionary; the contract is the newer the dictionary the larger the dict_id
+ "PRIMARY KEY ((keyspace_name, table_name), dict_id)) " +
+ "WITH CLUSTERING ORDER BY (dict_id DESC)"; // in order to retrieve the latest dictionary; the contract is the newer the dictionary the larger the dict_id
private static final TableMetadata CompressionDictionariesTable =
parse(COMPRESSION_DICTIONARIES, "Compression dictionaries for applicable tables", COMPRESSION_DICTIONARIES_CQL).build();
+ public static final String INDEX_BUILD_STATUS_CQL = "CREATE TABLE IF NOT EXISTS %s (" +
+ "host_id uuid," +
+ "keyspace_name text," +
+ "index_name text," +
+ "status text," +
+ "PRIMARY KEY (host_id, keyspace_name, index_name))";
+ private static final TableMetadata IndexBuildStatus =
+ parse(INDEX_BUILD_STATUS, "Index build status", INDEX_BUILD_STATUS_CQL).build();
+
+ private static final String INDEX_EVENTS_CQL = "CREATE TABLE IF NOT EXISTS %s (" +
+ "date text," +
+ "event_time timestamp," +
+ "index_name text," +
+ "host_id UUID," +
+ "event text," +
+ "PRIMARY KEY (date, event_time, index_name, host_id)) " +
+ "WITH CLUSTERING ORDER BY (event_time DESC)";
Review Comment:
this has to be `ASC` to get the events in the correct order - otherwise we might update the in-memory map in IndexStatusManager with an older event
##########
src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java:
##########
@@ -414,50 +443,148 @@ public static void setViewRemoved(String keyspaceName, String viewName)
forceBlockingFlush(VIEW_BUILD_STATUS, ColumnFamilyStore.FlushReason.INTERNALLY_FORCED);
}
+ public static void updateIndexStatus(UUID hostId, String keyspace, String index, Index.Status status)
+ {
+ String query = "INSERT INTO %s.%s (host_id, keyspace_name, index_name, status) VALUES (?, ?, ?, ?)";
+ QueryProcessor.process(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS),
+ ConsistencyLevel.ONE,
+ Lists.newArrayList(bytes(hostId),
+ bytes(keyspace),
+ bytes(index),
+ bytes(status.toString())));
+ }
+
+ public static void setIndexRemoved(UUID hostId, String keyspaceName, String indexName)
+ {
+ String buildReq = "DELETE FROM %s.%s WHERE host_id = ? AND keyspace_name = ? AND index_name = ?";
+ QueryProcessor.executeInternal(format(buildReq, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS),
Review Comment:
this should probably not use `executeInternal`?
##########
src/java/org/apache/cassandra/index/IndexStatusManager.java:
##########
@@ -230,29 +257,65 @@ public synchronized void propagateLocalIndexStatus(String keyspace, String index
Map<String, Index.Status> statusMap = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(),
k -> new HashMap<>());
String keyspaceIndex = identifier(keyspace, index);
+ UUID localHostId = StorageService.instance.getLocalHostUUID();
+ CassandraVersion minVersion = ClusterMetadata.current().directory.clusterMinVersion.cassandraVersion;
Review Comment:
nit; can do `boolean shouldWriteToIndexTables = shouldWriteToIndexTables(ClusterMetadata.current().directory.clusterMinVersion.cassandraVersion)` instead of calling this 3 times
##########
src/java/org/apache/cassandra/index/IndexStatusManager.java:
##########
@@ -230,29 +257,65 @@ public synchronized void propagateLocalIndexStatus(String keyspace, String index
Map<String, Index.Status> statusMap = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(),
k -> new HashMap<>());
String keyspaceIndex = identifier(keyspace, index);
+ UUID localHostId = StorageService.instance.getLocalHostUUID();
+ CassandraVersion minVersion = ClusterMetadata.current().directory.clusterMinVersion.cassandraVersion;
if (status == Index.Status.DROPPED)
+ {
statusMap.remove(keyspaceIndex);
+ if (localHostId != null)
+ statusPropagationExecutor.submit(() -> {
+ if (shouldWriteToIndexTables(minVersion))
+ {
+ SystemDistributedKeyspace.setIndexRemoved(localHostId, keyspace, index);
+ SystemDistributedKeyspace.recordIndexEvent(localHostId, keyspace, index, status);
+ }
+ });
+ }
else
+ {
statusMap.put(keyspaceIndex, status);
+ if (localHostId != null)
+ statusPropagationExecutor.submit(() -> {
+ if (shouldWriteToIndexTables(minVersion))
+ {
+ SystemDistributedKeyspace.updateIndexStatus(localHostId, keyspace, index, status);
+ SystemDistributedKeyspace.recordIndexEvent(localHostId, keyspace, index, status);
+ }
+ });
+ }
- // Don't try and propagate if the gossiper isn't enabled. This is primarily for tests where the
- // Gossiper has not been started. If we attempt to propagate when not started an exception is
- // logged and this causes a number of dtests to fail.
- if (Gossiper.instance.isEnabled())
+ // Only propagate via gossip when the gossiper is enabled and the cluster is not fully on 6.0+.
+ // Once all nodes are on 6.0+, index status is propagated via table polling instead.
+ if (Gossiper.instance.isEnabled() && !shouldWriteToIndexTables(minVersion))
{
// Versions 5.0.0 through 5.0.2 use a much more bloated format that duplicates keyspace names
// and writes full status names instead of their numeric codes. If the minimum cluster version is
// unknown or one of those 3 versions, continue to propagate the old format.
- CassandraVersion minVersion = ClusterMetadata.current().directory.clusterMinVersion.cassandraVersion;
-
- String newSerializedStatusMap = shouldWriteLegacyStatusFormat(minVersion) ? JsonUtils.writeAsJsonString(statusMap)
+ String newSerializedStatusMap = shouldWriteLegacyStatusFormat(minVersion) ? JsonUtils.writeAsJsonString(statusMap)
: toSerializedFormat(statusMap);
+ byte[] utf8Bytes = newSerializedStatusMap.getBytes(StandardCharsets.UTF_8);
+ String gossipPayload;
+
+ if (utf8Bytes.length > MAX_GOSSIP_VALUE_SIZE)
+ {
+ logger.error("Index status gossip payload size ({} bytes) exceeds limit ({} bytes), please consider removing unwanted indexes.",
+ utf8Bytes.length, MAX_GOSSIP_VALUE_SIZE);
+ gossipPayload = TABLE_FALLBACK_MARKER;
Review Comment:
do we actually need the fallback marker?
In mixed mode the old nodes won't understand the TABLE_FALLBACK_MARKER and will probably fail quite hard seeing it.
##########
test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java:
##########
Review Comment:
this needs `.set("index_status_poll_interval_in_seconds", "1")` to pass
##########
src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java:
##########
@@ -414,50 +443,148 @@ public static void setViewRemoved(String keyspaceName, String viewName)
forceBlockingFlush(VIEW_BUILD_STATUS, ColumnFamilyStore.FlushReason.INTERNALLY_FORCED);
}
+ public static void updateIndexStatus(UUID hostId, String keyspace, String index, Index.Status status)
+ {
+ String query = "INSERT INTO %s.%s (host_id, keyspace_name, index_name, status) VALUES (?, ?, ?, ?)";
+ QueryProcessor.process(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS),
+ ConsistencyLevel.ONE,
+ Lists.newArrayList(bytes(hostId),
+ bytes(keyspace),
+ bytes(index),
+ bytes(status.toString())));
+ }
+
+ public static void setIndexRemoved(UUID hostId, String keyspaceName, String indexName)
+ {
+ String buildReq = "DELETE FROM %s.%s WHERE host_id = ? AND keyspace_name = ? AND index_name = ?";
+ QueryProcessor.executeInternal(format(buildReq, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS),
+ hostId, keyspaceName, indexName);
+ forceBlockingFlush(INDEX_BUILD_STATUS, ColumnFamilyStore.FlushReason.INTERNALLY_FORCED);
+ }
+
+ public static Map<UUID, Map<String, Index.Status>> allIndexStatuses()
+ {
+ String query = "SELECT host_id, keyspace_name, index_name, status FROM %s.%s";
+ UntypedResultSet results;
+
+ try
+ {
+ results = QueryProcessor.execute(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS),
+ ConsistencyLevel.ONE);
+ }
+ catch (Exception e)
+ {
+ logger.warn("Unable to load index statuses from system table: {}", e.getMessage());
+ return Collections.emptyMap();
+ }
+
+ Map<UUID, Map<String, Index.Status>> allStatuses = new HashMap<>();
+ for (UntypedResultSet.Row row : results)
+ {
+ UUID hostId = row.getUUID("host_id");
+ String identifier = row.getString("keyspace_name") + '.' + row.getString("index_name");
+ Index.Status status = Index.Status.valueOf(row.getString("status"));
+ allStatuses.computeIfAbsent(hostId, k -> new HashMap<>()).put(identifier, status);
+ }
+
+ return allStatuses;
+ }
+
+ public static Map<String, Index.Status> allIndexStatusesForHost(UUID hostId)
+ {
+ String query = "SELECT keyspace_name, index_name, status FROM %s.%s WHERE host_id = ?";
+ UntypedResultSet results;
+
+ try
+ {
+ results = QueryProcessor.execute(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS),
+ ConsistencyLevel.ONE,
+ hostId);
+ }
+ catch (Exception e)
+ {
+ logger.warn("Unable to load index statuses from system table for host {}: {}", hostId, e.getMessage());
+ return Collections.emptyMap();
+ }
+
+ Map<String, Index.Status> statuses = new HashMap<>();
+ for (UntypedResultSet.Row row : results)
+ {
+ statuses.put(row.getString("keyspace_name") + '.' + row.getString("index_name"),
+ Index.Status.valueOf(row.getString("status")));
+ }
+
+ return statuses;
+ }
+
+ public static void recordIndexEvent(UUID hostId, String keyspace, String index, Index.Status status)
+ {
+ String query = "INSERT INTO %s.%s (date, event_time, index_name, host_id, event) VALUES (?, to_timestamp(now()), ?, ?, ?)";
+ QueryProcessor.process(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_EVENTS),
Review Comment:
should use `.execute(...)` and not wrap the parameters in `bytes(...)` below
##########
src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java:
##########
@@ -414,50 +443,148 @@ public static void setViewRemoved(String keyspaceName, String viewName)
forceBlockingFlush(VIEW_BUILD_STATUS, ColumnFamilyStore.FlushReason.INTERNALLY_FORCED);
}
+ public static void updateIndexStatus(UUID hostId, String keyspace, String index, Index.Status status)
+ {
+ String query = "INSERT INTO %s.%s (host_id, keyspace_name, index_name, status) VALUES (?, ?, ?, ?)";
+ QueryProcessor.process(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS),
Review Comment:
use QueryProcessor.execute instead, and avoid wrapping the parameters with `bytes(..)` below
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org