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

Flush snapshot table on every write and introduce reversed long local partitioner. 

Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-19391
This commit is contained in:
Marcus Eriksson 2024-02-12 14:23:25 +01:00
parent 0e5fb78c01
commit f31d72e875
2 changed files with 241 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -62,7 +62,6 @@ import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.db.compaction.CompactionHistoryTabularData;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.ReversedType;
import org.apache.cassandra.db.marshal.TimeUUIDType;
import org.apache.cassandra.db.marshal.TupleType;
import org.apache.cassandra.db.marshal.UTF8Type;
@ -73,6 +72,7 @@ import org.apache.cassandra.db.rows.Rows;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.ReversedLongLocalPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.gms.EndpointState;
@ -505,7 +505,7 @@ public final class SystemKeyspace
"epoch bigint PRIMARY KEY," +
"period bigint," +
"snapshot blob)")
.partitioner(new LocalPartitioner(ReversedType.getInstance(LongType.instance)))
.partitioner(ReversedLongLocalPartitioner.instance)
.build();
@Deprecated(since = "4.0")
@ -1990,6 +1990,7 @@ public final class SystemKeyspace
logger.info("Storing snapshot of cluster metadata at epoch {} (period {})", epoch, period);
String query = String.format("INSERT INTO %s.%s (epoch, period, snapshot) VALUES (?, ?, ?)", SchemaConstants.SYSTEM_KEYSPACE_NAME, SNAPSHOT_TABLE_NAME);
executeInternal(query, epoch.getEpoch(), period, snapshot);
forceBlockingFlush(SNAPSHOT_TABLE_NAME);
}
public static ByteBuffer getSnapshot(Epoch epoch)
@ -2003,12 +2004,23 @@ public final class SystemKeyspace
}
/**
* WARNING: we use token(epoch) >= search in the query below - this is due the fact that we use LocalPartitioner with a reversed LongToken
* and this is not quite supported (yet), so the query is actually `token(epoch) <= search` which is what we want here
* We use ReversedLongLocalPartitioner here, which calculates token as Long.MAX_VALUE - key
*
* table is something like (assuming Long.MAX_VALUE is 1000 for easier calculations...):
* epoch | token(epoch)
* --------------------
* 100 | 900
* 90 | 910
* 80 | 920
* 70 | 970
* ...
*
* so to find the first snapshot before epoch 85, we query the table with token(epoch) >= token(85)=915. Which gives us
* epoch 80, 70... and the first row is the first snapshot before `search`
*/
public static ByteBuffer findSnapshotBefore(Epoch search)
{
String query = String.format("SELECT snapshot FROM %s.%s WHERE token(epoch) >= ? LIMIT 1", SchemaConstants.SYSTEM_KEYSPACE_NAME, SNAPSHOT_TABLE_NAME);
String query = String.format("SELECT snapshot FROM %s.%s WHERE token(epoch) >= token(?) LIMIT 1", SchemaConstants.SYSTEM_KEYSPACE_NAME, SNAPSHOT_TABLE_NAME);
UntypedResultSet res = executeInternal(query, search.getEpoch());
if (res != null && !res.isEmpty())

224
src/java/org/apache/cassandra/dht/ReversedLongLocalPartitioner.java Normal file
View File

@ -0,0 +1,224 @@
/*
* 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.dht;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Random;
import com.google.common.primitives.Longs;
import org.apache.cassandra.db.CachedHashDecoratedKey;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.bytecomparable.ByteComparable;
import org.apache.cassandra.utils.bytecomparable.ByteSource;
import org.apache.cassandra.utils.bytecomparable.ByteSourceInverse;
import org.apache.cassandra.utils.memory.HeapCloner;
/**
* Special ordered partitioner for the TCM snapshots table, sorting in reverse. Keys are longs (epoch), and tokens
* become Long.MAX_VALUE - key. Keys are required to be >= 0.
*/
public class ReversedLongLocalPartitioner implements IPartitioner
{
public static ReversedLongLocalPartitioner instance = new ReversedLongLocalPartitioner();
private static final ReversedLongLocalToken MIN_TOKEN = new ReversedLongLocalToken(Long.MIN_VALUE);
private static final long HEAP_SIZE = ObjectSizes.measure(MIN_TOKEN);
private ReversedLongLocalPartitioner() {}
public DecoratedKey decorateKey(ByteBuffer key)
{
return new CachedHashDecoratedKey(getToken(key), key); // CachedHashDecoratedKey is used for bloom filter hash calculation
}
public Token midpoint(Token left, Token right)
{
throw new UnsupportedOperationException();
}
public Token split(Token left, Token right, double ratioToLeft)
{
throw new UnsupportedOperationException();
}
public Token getMinimumToken()
{
return MIN_TOKEN;
}
public Token getToken(ByteBuffer key)
{
if (!key.hasRemaining())
return MIN_TOKEN;
long longKey = ByteBufferUtil.toLong(HeapCloner.instance.clone(key));
assert longKey >= 0 : "ReversedLocalLongToken only supports non-negative keys, not " + longKey;
return new ReversedLongLocalToken(Long.MAX_VALUE - longKey);
}
public Token getRandomToken()
{
throw new UnsupportedOperationException();
}
public Token getRandomToken(Random random)
{
throw new UnsupportedOperationException();
}
public Token.TokenFactory getTokenFactory()
{
return tokenFactory;
}
private final Token.TokenFactory tokenFactory = new Token.TokenFactory()
{
public Token fromComparableBytes(ByteSource.Peekable comparableBytes, ByteComparable.Version version)
{
long tokenData = ByteSourceInverse.getSignedLong(comparableBytes);
return new ReversedLongLocalToken(tokenData);
}
public ByteBuffer toByteArray(Token token)
{
ReversedLongLocalToken longToken = (ReversedLongLocalToken) token;
return ByteBufferUtil.bytes(longToken.token);
}
public Token fromByteArray(ByteBuffer bytes)
{
return new ReversedLongLocalToken(ByteBufferUtil.toLong(bytes));
}
public String toString(Token token)
{
return token.toString();
}
public void validate(String token)
{
LongType.instance.validate(LongType.instance.fromString(token));
}
public Token fromString(String string)
{
return new ReversedLongLocalToken(Long.parseLong(string));
}
};
public boolean preservesOrder()
{
return true;
}
public Map<Token, Float> describeOwnership(List<Token> sortedTokens)
{
return Collections.singletonMap(getMinimumToken(), 1.0F);
}
public AbstractType<?> getTokenValidator()
{
return LongType.instance;
}
public AbstractType<?> partitionOrdering()
{
return LongType.instance;
}
private static class ReversedLongLocalToken extends Token
{
private final long token;
public ReversedLongLocalToken(long token)
{
this.token = token;
}
@Override
public IPartitioner getPartitioner()
{
return ReversedLongLocalPartitioner.instance;
}
@Override
public long getHeapSize()
{
return HEAP_SIZE;
}
@Override
public Object getTokenValue()
{
return token;
}
@Override
public ByteSource asComparableBytes(ByteComparable.Version version)
{
return ByteSource.of(token);
}
@Override
public double size(Token next)
{
throw new UnsupportedOperationException(String.format("Token type %s does not support token allocation.",
getClass().getSimpleName()));
}
@Override
public Token nextValidToken()
{
throw new UnsupportedOperationException(String.format("Token type %s does not support token allocation.",
getClass().getSimpleName()));
}
@Override
public int compareTo(Token o)
{
return Long.compare(token, ((ReversedLongLocalToken)o).token);
}
@Override
public String toString()
{
return String.valueOf(token);
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (!(o instanceof ReversedLongLocalToken)) return false;
ReversedLongLocalToken that = (ReversedLongLocalToken) o;
return token == that.token;
}
@Override
public int hashCode()
{
return Longs.hashCode(token);
}
}
}