You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "krummas (via GitHub)" <gi...@apache.org> on 2024/02/06 07:31:29 UTC
[PR] CASSANDRA-19189 [cassandra]
krummas opened a new pull request, #3088:
URL: https://github.com/apache/cassandra/pull/3088
- remove sealed_periods and last_sealed_period tables
- make metadata_snapshots use an ordered partitioner to be able to quickly find the most recent snapshot
- simplify the ways we query snapshots and log table
--
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] CASSANDRA-19189 [cassandra]
Posted by "krummas (via GitHub)" <gi...@apache.org>.
krummas commented on code in PR #3088:
URL: https://github.com/apache/cassandra/pull/3088#discussion_r1479754070
##########
src/java/org/apache/cassandra/tcm/log/LogState.java:
##########
@@ -237,31 +171,122 @@ public Serializer(Version serializationVersion)
@Override
public void serialize(LogState t, DataOutputPlus out, int version) throws IOException
{
- out.writeBoolean(t.baseState != null);
- if (t.baseState != null)
- VerboseMetadataSerializer.serialize(ClusterMetadata.serializer, t.baseState, out, serializationVersion);
- VerboseMetadataSerializer.serialize(Replication.serializer, t.transformations, out, serializationVersion);
+ VerboseMetadataSerializer.serialize(metadataSerializer, t, out, serializationVersion);
}
@Override
public LogState deserialize(DataInputPlus in, int version) throws IOException
{
- boolean hasSnapshot = in.readBoolean();
- ClusterMetadata snapshot = null;
- if (hasSnapshot)
- snapshot = VerboseMetadataSerializer.deserialize(ClusterMetadata.serializer, in);
- Replication replication = VerboseMetadataSerializer.deserialize(Replication.serializer, in);
- return new LogState(snapshot, replication);
+ return VerboseMetadataSerializer.deserialize(metadataSerializer, in);
}
@Override
public long serializedSize(LogState t, int version)
{
- long size = TypeSizes.BOOL_SIZE;
+ return VerboseMetadataSerializer.serializedSize(metadataSerializer, t, serializationVersion);
+ }
+ }
+
+ static final class MetadataSerializer implements org.apache.cassandra.tcm.serialization.MetadataSerializer<LogState>
+ {
+ @Override
+ public void serialize(LogState t, DataOutputPlus out, Version version) throws IOException
+ {
+ out.writeBoolean(t.baseState != null);
if (t.baseState != null)
- size += VerboseMetadataSerializer.serializedSize(ClusterMetadata.serializer, t.baseState, serializationVersion);
- size += VerboseMetadataSerializer.serializedSize(Replication.serializer, t.transformations, serializationVersion);
+ ClusterMetadata.serializer.serialize(t.baseState, out, version);
+ out.writeInt(t.entries.size());
+ for (Entry entry : t.entries)
+ Entry.serializer.serialize(entry, out, version);
+ }
+
+ @Override
+ public LogState deserialize(DataInputPlus in, Version version) throws IOException
+ {
+ ClusterMetadata baseState = null;
+ if (in.readBoolean())
+ baseState = ClusterMetadata.serializer.deserialize(in, version);
+ int size = in.readInt();
+ ImmutableList.Builder<Entry> builder = ImmutableList.builder();
+ for(int i=0;i<size;i++)
+ builder.add(Entry.serializer.deserialize(in, version));
+ return new LogState(baseState, builder.build());
+ }
+
+ @Override
+ public long serializedSize(LogState t, Version version)
+ {
+ long size = TypeSizes.sizeof(t.baseState != null);
+ if (t.baseState != null)
+ size += ClusterMetadata.serializer.serializedSize(t.baseState, version);
+ size += TypeSizes.INT_SIZE;
+ for (Entry entry : t.entries)
+ size += Entry.serializer.serializedSize(entry, version);
return size;
}
}
+
+ public static final class ReplicationHandler implements IVerbHandler<LogState>
+ {
+ private static final Logger logger = LoggerFactory.getLogger(ReplicationHandler.class);
+ private final LocalLog log;
+
+ public ReplicationHandler(LocalLog log)
+ {
+ this.log = log;
+ }
+
+ public void doVerb(Message<LogState> message) throws IOException
+ {
+ logger.info("Received logstate {} from {}", message.payload, message.from());
Review Comment:
I fixed the toString on LogState
--
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] CASSANDRA-19189 [cassandra]
Posted by "beobal (via GitHub)" <gi...@apache.org>.
beobal commented on code in PR #3088:
URL: https://github.com/apache/cassandra/pull/3088#discussion_r1479697203
##########
src/java/org/apache/cassandra/tcm/log/LogReader.java:
##########
@@ -18,16 +18,132 @@
package org.apache.cassandra.tcm.log;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.google.common.collect.ImmutableList;
+
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
-import org.apache.cassandra.tcm.Sealed;
+import org.apache.cassandra.tcm.MetadataSnapshots;
+import org.apache.cassandra.tcm.Period;
public interface LogReader
{
- default Replication getReplication(Epoch since)
+ /**
+ * Gets all entries where epoch >= since in the given period - could be empty if since is in a later epoch
+ */
+ EntryHolder getEntries(long period, Epoch since) throws IOException;
+ MetadataSnapshots snapshots();
+
+ /**
+ * Idea is to fill LogState "backwards" - we start querying the partition at currentPeriod, and if that doesn't
+ * include `since` we read currentPeriod - 1. Assuming we have something like
+
+ * epoch | period | transformation
+ * 10 | 2 | SEAL_PERIOD
+ * 11 | 3 | SOMETHING
+ * 12 | 3 | SOMETHING
+ * 13 | 3 | SEAL_PERIOD
+ * 14 | 4 | SOMETHING
+ * 15 | 4 | SOMETHING
+ * 16 | 4 | SEAL_PERIOD
+ * 17 | 5 | SOMETHING
+ *
+ * and `since` is 14, we want to return epoch 15, 16, 17 - not the snapshot at 16 + entry at 17 since it is assumed that the full
+ * snapshot is much larger than the transformations.
+ * But, if `since` is 11, we want to return the most recent snapshot (at epoch 16) + entry at 17.
+ *
+ * If a snapshot is missing we keep reading backwards until we find one, or we end up at period 0 and in that
+ * case we return all transformations in the log.
+ */
+ default LogState getLogState(long currentPeriod, Epoch since)
{
- Sealed sealed = Sealed.lookupForReplication(since);
- return getReplication(sealed.period, since);
+ try
+ {
+ EntryHolder current = getEntries(currentPeriod, since);
+ if (current.done)
+ return new LogState(null, ImmutableList.sortedCopyOf(current.entries));
+ List<Entry> allEntries = new ArrayList<>(current.entries);
+ int i = 0;
+ while (true)
+ {
+ i++;
+ EntryHolder previous = getEntries(currentPeriod - i, since);
+ allEntries.addAll(previous.entries);
+ if (isContinuous(since, allEntries) && (previous.done || currentPeriod - i == Period.FIRST))
+ {
+ return new LogState(null, ImmutableList.sortedCopyOf(allEntries));
+ }
+ else
+ {
+ if (i == 1)
+ {
+ // we end up here if `since` is not in currentPeriod or currentPeriod - 1
+ // so we should return a snapshot - we prefer returning the most recent snapshot + entries in `current`
+ // but if that doesn't exist we check previous.
+ if (current.min == null && previous.min == null) // we found no entries >= since in the tables -> since > current
+ return LogState.EMPTY;
+ ClusterMetadata snapshot = snapshots().getSnapshot(Epoch.create(current.min.getEpoch() - 1));
+ if (snapshot != null)
+ return new LogState(snapshot, ImmutableList.sortedCopyOf(current.entries));
+ }
+ ClusterMetadata snapshot = snapshots().getSnapshot(Epoch.create(previous.min.getEpoch() - 1));
Review Comment:
There's a chance we can NPE here if we are missing both a snapshot and entries from the log, which should come after `since`. This could happen in the local table if we received a snapshot at `since + x`, causing us not to have the individual entries directly following `since`, but then we lost the snapshot. I've pushed a test and fix for this to https://github.com/beobal/cassandra/commit/7d66c2997186a6dda7bc3ba95102207b988cfd4c
One question there is whether its better to return a `LogState` with a non-contiguous list of entries, or an empty one.
##########
src/java/org/apache/cassandra/tcm/log/LogState.java:
##########
@@ -237,31 +171,122 @@ public Serializer(Version serializationVersion)
@Override
public void serialize(LogState t, DataOutputPlus out, int version) throws IOException
{
- out.writeBoolean(t.baseState != null);
- if (t.baseState != null)
- VerboseMetadataSerializer.serialize(ClusterMetadata.serializer, t.baseState, out, serializationVersion);
- VerboseMetadataSerializer.serialize(Replication.serializer, t.transformations, out, serializationVersion);
+ VerboseMetadataSerializer.serialize(metadataSerializer, t, out, serializationVersion);
}
@Override
public LogState deserialize(DataInputPlus in, int version) throws IOException
{
- boolean hasSnapshot = in.readBoolean();
- ClusterMetadata snapshot = null;
- if (hasSnapshot)
- snapshot = VerboseMetadataSerializer.deserialize(ClusterMetadata.serializer, in);
- Replication replication = VerboseMetadataSerializer.deserialize(Replication.serializer, in);
- return new LogState(snapshot, replication);
+ return VerboseMetadataSerializer.deserialize(metadataSerializer, in);
}
@Override
public long serializedSize(LogState t, int version)
{
- long size = TypeSizes.BOOL_SIZE;
+ return VerboseMetadataSerializer.serializedSize(metadataSerializer, t, serializationVersion);
+ }
+ }
+
+ static final class MetadataSerializer implements org.apache.cassandra.tcm.serialization.MetadataSerializer<LogState>
+ {
+ @Override
+ public void serialize(LogState t, DataOutputPlus out, Version version) throws IOException
+ {
+ out.writeBoolean(t.baseState != null);
if (t.baseState != null)
- size += VerboseMetadataSerializer.serializedSize(ClusterMetadata.serializer, t.baseState, serializationVersion);
- size += VerboseMetadataSerializer.serializedSize(Replication.serializer, t.transformations, serializationVersion);
+ ClusterMetadata.serializer.serialize(t.baseState, out, version);
+ out.writeInt(t.entries.size());
+ for (Entry entry : t.entries)
+ Entry.serializer.serialize(entry, out, version);
+ }
+
+ @Override
+ public LogState deserialize(DataInputPlus in, Version version) throws IOException
+ {
+ ClusterMetadata baseState = null;
+ if (in.readBoolean())
+ baseState = ClusterMetadata.serializer.deserialize(in, version);
+ int size = in.readInt();
+ ImmutableList.Builder<Entry> builder = ImmutableList.builder();
+ for(int i=0;i<size;i++)
+ builder.add(Entry.serializer.deserialize(in, version));
+ return new LogState(baseState, builder.build());
+ }
+
+ @Override
+ public long serializedSize(LogState t, Version version)
+ {
+ long size = TypeSizes.sizeof(t.baseState != null);
+ if (t.baseState != null)
+ size += ClusterMetadata.serializer.serializedSize(t.baseState, version);
+ size += TypeSizes.INT_SIZE;
+ for (Entry entry : t.entries)
+ size += Entry.serializer.serializedSize(entry, version);
return size;
}
}
+
+ public static final class ReplicationHandler implements IVerbHandler<LogState>
+ {
+ private static final Logger logger = LoggerFactory.getLogger(ReplicationHandler.class);
+ private final LocalLog log;
+
+ public ReplicationHandler(LocalLog log)
+ {
+ this.log = log;
+ }
+
+ public void doVerb(Message<LogState> message) throws IOException
+ {
+ logger.info("Received logstate {} from {}", message.payload, message.from());
Review Comment:
This might be a bit verbose now. Before this, the equivalent log message included only the size & min/max epochs in the Replication. Maybe we could go back to something like that + the epoch of any base state?
##########
src/java/org/apache/cassandra/schema/DistributedMetadataLogKeyspace.java:
##########
@@ -173,105 +171,54 @@ public static boolean tryCommit(Entry.Id entryId,
}
}
- @VisibleForTesting
- public static void truncateLogState()
- {
- QueryProcessor.execute(String.format("TRUNCATE %s.%s", SchemaConstants.METADATA_KEYSPACE_NAME, TABLE_NAME), ConsistencyLevel.QUORUM);
- }
-
-
private static final LogReader localLogReader = new DistributedTableLogReader(ConsistencyLevel.NODE_LOCAL);
private static final LogReader serialLogReader = new DistributedTableLogReader(ConsistencyLevel.SERIAL);
public static LogState getLogState(Epoch since, boolean consistentFetch)
{
- return LogState.getLogState(since, ClusterMetadataService.instance().snapshotManager(), consistentFetch ? serialLogReader : localLogReader);
- }
-
- @VisibleForTesting
- public static LogState getLogState(Epoch since, LogReader logReader, MetadataSnapshots snapshots)
- {
- Retry retry = new Retry.Jitter(TCMMetrics.instance.fetchLogRetries);
- while (!retry.reachedMax())
- {
- try
- {
- return LogState.getLogState(since, snapshots, logReader);
- }
- catch (Throwable t)
- {
- retry.maybeSleep();
- }
- }
-
- throw new IllegalStateException(String.format("Could not retrieve log state after %s tries.", retry.currentTries()));
+ return (consistentFetch ? serialLogReader : localLogReader).getLogState(ClusterMetadata.current().period, since);
}
public static class DistributedTableLogReader implements LogReader
{
private final ConsistencyLevel consistencyLevel;
+ private final Supplier<MetadataSnapshots> snapshots;
- public DistributedTableLogReader(ConsistencyLevel consistencyLevel)
+ public DistributedTableLogReader(ConsistencyLevel consistencyLevel, Supplier<MetadataSnapshots> snapshots)
{
this.consistencyLevel = consistencyLevel;
+ this.snapshots = snapshots;
}
- @Override
- public Replication getReplication(long startPeriod, Epoch since)
+ public DistributedTableLogReader(ConsistencyLevel consistencyLevel)
{
- try
- {
- if (startPeriod == Period.EMPTY)
- {
- startPeriod = Period.scanLogForPeriod(Log, since);
- // There shouldn't be any entries in period 0, the pre-init transform would bump it to period 1.
- if (startPeriod == Period.EMPTY)
- return Replication.EMPTY;
- }
-
- long currentEpoch = since.getEpoch();
- long lastEpoch = since.getEpoch();
-
- long period = startPeriod;
- ImmutableList.Builder<Entry> entries = new ImmutableList.Builder<>();
-
- while (true)
- {
- boolean empty = true;
- UntypedResultSet resultSet = execute(String.format("SELECT current_epoch, period, epoch, kind, transformation, entry_id, sealed FROM %s.%s WHERE period = ? AND epoch > ?",
- SchemaConstants.METADATA_KEYSPACE_NAME, TABLE_NAME),
- consistencyLevel, period, since.getEpoch());
-
- for (UntypedResultSet.Row row : resultSet)
- {
- currentEpoch = row.getLong("current_epoch");
- long epochl = row.getLong("epoch");
- Epoch epoch = Epoch.create(epochl);
- Transformation.Kind kind = Transformation.Kind.valueOf(row.getString("kind"));
- long entryId = row.getLong("entry_id");
- Transformation transform = kind.fromVersionedBytes(row.getBlob("transformation"));
- entries.add(new Entry(new Entry.Id(entryId), epoch, transform));
-
- lastEpoch = currentEpoch;
- empty = false;
- }
-
- if (period != startPeriod && empty)
- break;
-
- period++;
- }
+ this(consistencyLevel, () -> ClusterMetadataService.instance().snapshotManager());
+ }
- assert currentEpoch == lastEpoch;
- return new Replication(entries.build());
- }
- catch (IOException t)
+ public EntryHolder getEntries(long period, Epoch since) throws IOException
+ {
+ UntypedResultSet resultSet = execute(String.format("SELECT current_epoch, period, epoch, kind, transformation, entry_id, sealed FROM %s.%s WHERE period = ? AND epoch >= ?",
Review Comment:
`current_epoch` & `period` are selected but not used
--
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] CASSANDRA-19189 [cassandra]
Posted by "krummas (via GitHub)" <gi...@apache.org>.
krummas closed pull request #3088: CASSANDRA-19189
URL: https://github.com/apache/cassandra/pull/3088
--
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