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

Add mutation tracking summary to SSTables 

Patch by Abe Ratnofsky; reviewed by Ariel Weisberg, Blake Eggleston for CASSANDRA-20336
This commit is contained in:
Abe Ratnofsky 2025-05-14 12:35:25 -04:00 committed by Blake Eggleston
parent fa017c6334
commit 51e63f0ff3
172 changed files with 1215 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
View File

@ -711,6 +711,9 @@ public abstract class AlterTableStatement extends AlterSchemaStatement
params = validateAndUpdateTransactionalMigration(table.isCounter(), table.params, params);
if (table.replicationType().isTracked() && params.memtable.factory().streamFromMemtable())
throw ire("Cannot use mutation tracking with persistent memtables");
return keyspace.withSwapped(keyspace.tables.withSwapped(table.withSwapped(params)));
}
}

5
src/java/org/apache/cassandra/db/CassandraTableWriteHandler.java
View File

@ -19,6 +19,7 @@
package org.apache.cassandra.db;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.tracing.Tracing;
public class CassandraTableWriteHandler implements TableWriteHandler
@ -31,10 +32,10 @@ public class CassandraTableWriteHandler implements TableWriteHandler
}
@Override
public void write(PartitionUpdate update, WriteContext context, boolean updateIndexes)
public void write(MutationId mutationId, PartitionUpdate update, WriteContext context, boolean updateIndexes)
{
CassandraWriteContext ctx = CassandraWriteContext.fromContext(context);
Tracing.trace("Adding to {} memtable", update.metadata().name);
cfs.apply(update, ctx, updateIndexes);
cfs.apply(mutationId, update, ctx, updateIndexes);
}
}

20
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -136,6 +136,7 @@ import org.apache.cassandra.metrics.TopPartitionTracker;
import org.apache.cassandra.repair.TableRepairManager;
import org.apache.cassandra.repair.consistent.admin.CleanupSummary;
import org.apache.cassandra.repair.consistent.admin.PendingStat;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.CompactionParams;
import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
@ -674,19 +675,19 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
return memtableFactory.streamFromMemtable();
}
public SSTableMultiWriter createSSTableMultiWriter(Descriptor descriptor, long keyCount, long repairedAt, TimeUUID pendingRepair, boolean isTransient, SerializationHeader header, ILifecycleTransaction txn)
public SSTableMultiWriter createSSTableMultiWriter(Descriptor descriptor, long keyCount, long repairedAt, TimeUUID pendingRepair, boolean isTransient, CoordinatorLogBoundaries coordinatorLogBoundaries, SerializationHeader header, ILifecycleTransaction txn)
{
return createSSTableMultiWriter(descriptor, keyCount, repairedAt, pendingRepair, isTransient, null, 0, header, txn);
return createSSTableMultiWriter(descriptor, keyCount, repairedAt, pendingRepair, isTransient, coordinatorLogBoundaries, null, 0, header, txn);
}
public SSTableMultiWriter createSSTableMultiWriter(Descriptor descriptor, long keyCount, long repairedAt, TimeUUID pendingRepair, boolean isTransient, IntervalSet<CommitLogPosition> commitLogPositions, SerializationHeader header, ILifecycleTransaction txn)
public SSTableMultiWriter createSSTableMultiWriter(Descriptor descriptor, long keyCount, long repairedAt, TimeUUID pendingRepair, boolean isTransient, CoordinatorLogBoundaries coordinatorLogBoundaries, IntervalSet<CommitLogPosition> commitLogPositions, SerializationHeader header, ILifecycleTransaction txn)
{
return createSSTableMultiWriter(descriptor, keyCount, repairedAt, pendingRepair, isTransient, commitLogPositions, 0, header, txn);
return createSSTableMultiWriter(descriptor, keyCount, repairedAt, pendingRepair, isTransient, coordinatorLogBoundaries, commitLogPositions, 0, header, txn);
}
public SSTableMultiWriter createSSTableMultiWriter(Descriptor descriptor, long keyCount, long repairedAt, TimeUUID pendingRepair, boolean isTransient, IntervalSet<CommitLogPosition> commitLogPositions, int sstableLevel, SerializationHeader header, ILifecycleTransaction txn)
public SSTableMultiWriter createSSTableMultiWriter(Descriptor descriptor, long keyCount, long repairedAt, TimeUUID pendingRepair, boolean isTransient, CoordinatorLogBoundaries coordinatorLogBoundaries, IntervalSet<CommitLogPosition> commitLogPositions, int sstableLevel, SerializationHeader header, ILifecycleTransaction txn)
{
return getCompactionStrategyManager().createSSTableMultiWriter(descriptor, keyCount, repairedAt, pendingRepair, isTransient, commitLogPositions, sstableLevel, header, indexManager.listIndexGroups(), txn);
return getCompactionStrategyManager().createSSTableMultiWriter(descriptor, keyCount, repairedAt, pendingRepair, isTransient, coordinatorLogBoundaries, commitLogPositions, sstableLevel, header, indexManager.listIndexGroups(), txn);
}
public boolean supportsEarlyOpen()
@ -1498,7 +1499,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
* @param context write context for current update
* @param updateIndexes whether secondary indexes should be updated
*/
public void apply(PartitionUpdate update, CassandraWriteContext context, boolean updateIndexes)
public void apply(MutationId mutationId, PartitionUpdate update, CassandraWriteContext context, boolean updateIndexes)
{
long start = nanoTime();
@ -1508,7 +1509,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
{
Memtable mt = data.getMemtableFor(opGroup, commitLogPosition);
UpdateTransaction indexer = newUpdateTransaction(update, context, updateIndexes, mt);
long timeDelta = mt.put(update, indexer, opGroup);
long timeDelta = mt.put(mutationId, update, indexer, opGroup);
DecoratedKey key = update.partitionKey();
invalidateCachedPartition(key);
metric.topWritePartitionFrequency.addSample(key.getKey(), 1);
@ -2419,12 +2420,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
return null;
List<Memtable.FlushablePartitionSet<?>> dataSets = new ArrayList<>(ranges.size());
CoordinatorLogBoundariesBuilder boundaries = new CoordinatorLogBoundariesBuilder();
IntervalSet.Builder<CommitLogPosition> commitLogIntervals = new IntervalSet.Builder();
long keys = 0;
for (Range<PartitionPosition> range : ranges)
{
Memtable.FlushablePartitionSet<?> dataSet = current.getFlushSet(range.left, range.right);
dataSets.add(dataSet);
boundaries.addAll(dataSet.coordinatorLogBoundaries());
commitLogIntervals.add(dataSet.commitLogLowerBound(), dataSet.commitLogUpperBound());
keys += dataSet.partitionCount();
}
@ -2438,6 +2441,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
0,
repairSessionID,
false,
boundaries.build(),
commitLogIntervals.build(),
new SerializationHeader(true,
firstDataSet.metadata(),

88
src/java/org/apache/cassandra/db/CoordinatorLogBoundaries.java Normal file
View File

@ -0,0 +1,88 @@
/*
* 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.db;
import java.io.IOException;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.replication.CoordinatorLogId;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.utils.vint.VIntCoding;
/**
* Max mutation ID present in this SSTable for each coordinator log, to determine whether an SSTable is reconciled or
* not. Once max mutation IDs are reconciled, next compaction can safely mark this SSTabled as repaired. Note that peers
* may have reconciled all mutations included in an SSTable, but {@link StatsMetadata#repairedAt} is dependent on
* compaction timing, so "nodetool repair --validate" may report temporary disagreements on the repaired set.
* <p>
* A reference to this class should be treated as immutable. Do not cast to {@link CoordinatorLogBoundariesMap}.
* Iterable over {@link CoordinatorLogId}.
*/
public interface CoordinatorLogBoundaries extends Iterable<Long>
{
int maxOffset(long logId);
MutationId max(long logId);
int size();
IVersionedSerializer<CoordinatorLogBoundaries> serializer = new IVersionedSerializer<>()
{
@Override
public void serialize(CoordinatorLogBoundaries boundaries, DataOutputPlus out, int version) throws IOException
{
if (version < MessagingService.VERSION_52)
return;
out.writeUnsignedVInt32(boundaries.size());
for (long logId : boundaries)
MutationId.serializer.serialize(boundaries.max(logId), out, version);
}
@Override
public CoordinatorLogBoundaries deserialize(DataInputPlus in, int version) throws IOException
{
if (version < MessagingService.VERSION_52)
return CoordinatorLogBoundaries.NONE;
int size = in.readUnsignedVInt32();
CoordinatorLogBoundariesMap boundaries = new CoordinatorLogBoundariesMap(size);
for (int i = 0; i < size; i++)
{
MutationId mutationId = MutationId.serializer.deserialize(in, version);
boundaries.add(mutationId);
}
return boundaries;
}
@Override
public long serializedSize(CoordinatorLogBoundaries boundaries, int version)
{
if (version < MessagingService.VERSION_52)
return 0;
long size = 0;
size += VIntCoding.computeUnsignedVIntSize(boundaries.size());
for (long logId : boundaries)
size += MutationId.serializer.serializedSize(boundaries.max(logId), version);
return size;
}
};
CoordinatorLogBoundaries NONE = new CoordinatorLogBoundariesMap();
}

92
src/java/org/apache/cassandra/db/CoordinatorLogBoundariesBuilder.java Normal file
View File

@ -0,0 +1,92 @@
/*
* 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.db;
import java.util.Iterator;
import javax.annotation.concurrent.NotThreadSafe;
import com.google.common.collect.Iterators;
import org.agrona.collections.Long2ObjectHashMap;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.replication.ShortMutationId;
@NotThreadSafe
public class CoordinatorLogBoundariesBuilder
{
private static final MutationId NONE = MutationId.none();
private static final int NONE_OFFSET = NONE.offset();
private final Long2ObjectHashMap<MutationId> ids;
public CoordinatorLogBoundariesBuilder()
{
this.ids = new Long2ObjectHashMap<>();
}
public CoordinatorLogBoundariesBuilder add(MutationId mutationId)
{
if (mutationId.isNone())
return this;
long logId = mutationId.logId();
MutationId existing = ids.get(logId);
if (existing == null || ShortMutationId.comparator.compare(existing, mutationId) < 0)
ids.put(mutationId.logId(), mutationId);
return this;
}
public CoordinatorLogBoundariesBuilder addAll(CoordinatorLogBoundaries boundaries)
{
for (long logId : boundaries)
add(boundaries.max(logId));
return this;
}
public CoordinatorLogBoundaries build()
{
return new CoordinatorLogBoundaries()
{
@Override
public int maxOffset(long logId)
{
MutationId id = ids.get(logId);
return id == null ? NONE_OFFSET : id.offset();
}
@Override
public MutationId max(long logId)
{
return ids.getOrDefault(logId, CoordinatorLogBoundariesBuilder.NONE);
}
@Override
public int size()
{
return ids.size();
}
@Override
public Iterator<Long> iterator()
{
return Iterators.unmodifiableIterator(ids.keySet().iterator());
}
};
}
}

80
src/java/org/apache/cassandra/db/CoordinatorLogBoundariesMap.java Normal file
View File

@ -0,0 +1,80 @@
/*
* 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.db;
import java.util.Iterator;
import javax.annotation.concurrent.ThreadSafe;
import com.google.common.collect.Iterators;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.replication.ShortMutationId;
import org.jctools.maps.NonBlockingHashMapLong;
/**
* A replica can only receive writes from another replica it shares ranges with, and tracked writes are executed by
* coordinators, so this should contain up to (2*RF - 1) keys.
* Consider wrapping value in AtomicReference to avoid false sharing, see: https://trishagee.com/2011/07/22/dissecting_the_disruptor_why_its_so_fast_part_two__magic_cache_line_padding/
*/
@ThreadSafe
class CoordinatorLogBoundariesMap extends NonBlockingHashMapLong<MutationId> implements MutableCoordinatorLogBoundaries
{
private static final MutationId NONE = MutationId.none();
private static final int NONE_OFFSET = NONE.offset();
protected CoordinatorLogBoundariesMap(int size)
{
super(size);
}
protected CoordinatorLogBoundariesMap()
{
super();
}
public void add(MutationId mutationId)
{
long logId = mutationId.logId();
merge(logId, mutationId, (existing, updating) -> {
if (ShortMutationId.comparator.compare(existing, updating) < 0)
return updating;
return existing;
});
}
@Override
public int maxOffset(long logId)
{
MutationId id = get(logId);
return id == null ? NONE_OFFSET : id.offset();
}
@Override
public MutationId max(long logId)
{
return getOrDefault(logId, NONE);
}
@Override
public Iterator<Long> iterator()
{
return Iterators.unmodifiableIterator(keySet().iterator());
}
}

4
src/java/org/apache/cassandra/db/Keyspace.java
View File

@ -583,7 +583,7 @@ public class Keyspace
}
}
cfs.getWriteHandler().write(upd, ctx, updateIndexes);
cfs.getWriteHandler().write(mutation.id(), upd, ctx, updateIndexes);
if (requiresViewUpdate)
baseComplete.set(currentTimeMillis());
@ -628,7 +628,7 @@ public class Keyspace
continue;
}
cfs.getWriteHandler().write(upd, ctx, true);
cfs.getWriteHandler().write(mutation.id(), upd, ctx, true);
}
}

46
src/java/org/apache/cassandra/db/MutableCoordinatorLogBoundaries.java Normal file
View File

@ -0,0 +1,46 @@
/*
* 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.db;
import org.apache.cassandra.replication.MutationId;
public interface MutableCoordinatorLogBoundaries extends CoordinatorLogBoundaries
{
void add(MutationId mutationId);
default void addAll(CoordinatorLogBoundaries from)
{
for (long logId : from)
{
MutationId max = from.max(logId);
if (!max.isNone())
add(max);
}
}
static MutableCoordinatorLogBoundaries create()
{
return new CoordinatorLogBoundariesMap();
}
static MutableCoordinatorLogBoundaries create(int size)
{
return new CoordinatorLogBoundariesMap(size);
}
}

8
src/java/org/apache/cassandra/db/SSTableImporter.java
View File

@ -45,6 +45,7 @@ import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.OutputHandler;
@ -80,6 +81,11 @@ public class SSTableImporter
UUID importID = UUID.randomUUID();
logger.info("[{}] Loading new SSTables for {}/{}: {}", importID, cfs.getKeyspaceName(), cfs.getTableName(), options);
// This will be supported in the future
TableMetadata metadata = cfs.metadata();
if (metadata.replicationType() != null && metadata.replicationType().isTracked())
throw new IllegalStateException("Can't import into tables with mutation tracking enabled");
List<Pair<Directories.SSTableLister, String>> listers = getSSTableListers(options.srcPaths);
Set<Descriptor> currentDescriptors = new HashSet<>();
@ -110,7 +116,7 @@ public class SSTableImporter
{
IndexDescriptor indexDescriptor = IndexDescriptor.create(descriptor,
cfs.getPartitioner(),
cfs.metadata().comparator);
metadata.comparator);
String keyspace = cfs.getKeyspaceName();
String table = cfs.getTableName();

3
src/java/org/apache/cassandra/db/TableWriteHandler.java
View File

@ -19,8 +19,9 @@
package org.apache.cassandra.db;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.replication.MutationId;
public interface TableWriteHandler
{
void write(PartitionUpdate update, WriteContext context, boolean updateIndexes);
void write(MutationId mutationId, PartitionUpdate update, WriteContext context, boolean updateIndexes);
}

2
src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
View File

@ -560,6 +560,7 @@ public abstract class AbstractCompactionStrategy
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
@ -571,6 +572,7 @@ public abstract class AbstractCompactionStrategy
repairedAt,
pendingRepair,
isTransient,
coordinatorLogBoundaries,
cfs.metadata,
commitLogPositions,
sstableLevel,

2
src/java/org/apache/cassandra/db/compaction/AbstractStrategyHolder.java
View File

@ -28,6 +28,7 @@ import java.util.function.Supplier;
import com.google.common.base.Preconditions;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.db.commitlog.IntervalSet;
@ -196,6 +197,7 @@ public abstract class AbstractStrategyHolder
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,

18
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -70,6 +70,8 @@ import org.apache.cassandra.concurrent.ExecutorFactory;
import org.apache.cassandra.concurrent.WrappedExecutorPlus;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.CoordinatorLogBoundariesBuilder;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.DiskBoundaries;
@ -1650,7 +1652,8 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan
CompactionIterator ci = new CompactionIterator(OperationType.CLEANUP, Collections.singletonList(scanner), controller, nowInSec, nextTimeUUID(), active, null))
{
StatsMetadata metadata = sstable.getSSTableMetadata();
writer.switchWriter(createWriter(cfs, compactionFileLocation, expectedBloomFilterSize, metadata.repairedAt, metadata.pendingRepair, metadata.isTransient, sstable, txn));
// TODO(aratnofsky): filter coordinatorLogBoundaries to exclude any CoordinatorLogIds we're no longer responsible for, after ownership change
writer.switchWriter(createWriter(cfs, compactionFileLocation, expectedBloomFilterSize, metadata.repairedAt, metadata.pendingRepair, metadata.isTransient, metadata.coordinatorLogBoundaries, sstable, txn));
long lastBytesScanned = 0;
while (ci.hasNext())
@ -1815,6 +1818,7 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
SSTableReader sstable,
LifecycleTransaction txn)
{
@ -1826,6 +1830,7 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan
.setRepairedAt(repairedAt)
.setPendingRepair(pendingRepair)
.setTransientSSTable(isTransient)
.setCoordinatorLogBoundaries(coordinatorLogBoundaries)
.setTableMetadataRef(cfs.metadata)
.setMetadataCollector(new MetadataCollector(cfs.metadata().comparator).sstableLevel(sstable.getSSTableLevel()))
.setSerializationHeader(sstable.header)
@ -1846,14 +1851,16 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan
{
FileUtils.createDirectory(compactionFileLocation);
int minLevel = Integer.MAX_VALUE;
// if all sstables have the same level, we can compact them together without creating overlap during anticompaction
// note that we only anticompact from unrepaired sstables, which is not leveled, but we still keep original level
// after first migration to be able to drop the sstables back in their original place in the repaired sstable manifest
CoordinatorLogBoundariesBuilder boundaries = new CoordinatorLogBoundariesBuilder();
for (SSTableReader sstable : sstables)
{
boundaries.addAll(sstable.getCoordinatorLogBoundaries());
// if all sstables have the same level, we can compact them together without creating overlap during anticompaction
// note that we only anticompact from unrepaired sstables, which is not leveled, but we still keep original level
// after first migration to be able to drop the sstables back in their original place in the repaired sstable manifest
if (minLevel == Integer.MAX_VALUE)
minLevel = sstable.getSSTableLevel();
if (minLevel != sstable.getSSTableLevel())
{
minLevel = 0;
@ -1866,6 +1873,7 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan
.setKeyCount(expectedBloomFilterSize)
.setRepairedAt(repairedAt)
.setPendingRepair(pendingRepair)
.setCoordinatorLogBoundaries(boundaries.build())
.setTransientSSTable(isTransient)
.setTableMetadataRef(cfs.metadata)
.setMetadataCollector(new MetadataCollector(sstables, cfs.metadata().comparator).sstableLevel(minLevel))

3
src/java/org/apache/cassandra/db/compaction/CompactionStrategyHolder.java
View File

@ -26,6 +26,7 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.db.commitlog.IntervalSet;
@ -226,6 +227,7 @@ public class CompactionStrategyHolder extends AbstractStrategyHolder
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
@ -251,6 +253,7 @@ public class CompactionStrategyHolder extends AbstractStrategyHolder
repairedAt,
pendingRepair,
isTransient,
coordinatorLogBoundaries,
commitLogPositions,
sstableLevel,
header,

3
src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java
View File

@ -50,6 +50,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.DiskBoundaries;
import org.apache.cassandra.db.SerializationHeader;
@ -1411,6 +1412,7 @@ public class CompactionStrategyManager implements INotificationConsumer
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
@ -1427,6 +1429,7 @@ public class CompactionStrategyManager implements INotificationConsumer
repairedAt,
pendingRepair,
isTransient,
coordinatorLogBoundaries,
commitLogPositions,
sstableLevel,
header,

9
src/java/org/apache/cassandra/db/compaction/CompactionTask.java
View File

@ -37,6 +37,8 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.CoordinatorLogBoundariesBuilder;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.WriteContext;
@ -458,6 +460,13 @@ public class CompactionTask extends AbstractCompactionTask
return isTransient;
}
public static CoordinatorLogBoundaries getCoordinatorLogBoundaries(Set<SSTableReader> sstables)
{
CoordinatorLogBoundariesBuilder builder = new CoordinatorLogBoundariesBuilder();
for (SSTableReader sstable : sstables)
builder.addAll(sstable.getCoordinatorLogBoundaries());
return builder.build();
}
/*
* Checks if we have enough disk space to execute the compaction. Drops the largest sstable out of the Task until

3
src/java/org/apache/cassandra/db/compaction/PendingRepairHolder.java
View File

@ -27,6 +27,7 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.db.commitlog.IntervalSet;
@ -246,6 +247,7 @@ public class PendingRepairHolder extends AbstractStrategyHolder
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
@ -263,6 +265,7 @@ public class PendingRepairHolder extends AbstractStrategyHolder
repairedAt,
pendingRepair,
isTransient,
coordinatorLogBoundaries,
commitLogPositions,
sstableLevel,
header,

2
src/java/org/apache/cassandra/db/compaction/UnifiedCompactionStrategy.java
View File

@ -300,6 +300,7 @@ public class UnifiedCompactionStrategy extends AbstractCompactionStrategy
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
SerializationHeader header,
@ -317,6 +318,7 @@ public class UnifiedCompactionStrategy extends AbstractCompactionStrategy
repairedAt,
pendingRepair,
isTransient,
coordinatorLogBoundaries,
commitLogPositions,
header,
indexGroups,

1
src/java/org/apache/cassandra/db/compaction/Upgrader.java
View File

@ -80,6 +80,7 @@ public class Upgrader
.setRepairedAt(metadata.repairedAt)
.setPendingRepair(metadata.pendingRepair)
.setTransientSSTable(metadata.isTransient)
.setCoordinatorLogBoundaries(metadata.coordinatorLogBoundaries)
.setTableMetadataRef(cfs.metadata)
.setMetadataCollector(sstableMetadataCollector)
.setSerializationHeader(SerializationHeader.make(cfs.metadata(), Sets.newHashSet(sstable)))

5
src/java/org/apache/cassandra/db/compaction/unified/ShardedMultiWriter.java
View File

@ -26,6 +26,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
@ -61,6 +62,7 @@ public class ShardedMultiWriter implements SSTableMultiWriter
private final long repairedAt;
private final TimeUUID pendingRepair;
private final boolean isTransient;
private final CoordinatorLogBoundaries coordinatorLogBoundaries;
private final IntervalSet<CommitLogPosition> commitLogPositions;
private final SerializationHeader header;
private final Collection<Index.Group> indexGroups;
@ -75,6 +77,7 @@ public class ShardedMultiWriter implements SSTableMultiWriter
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
IntervalSet<CommitLogPosition> commitLogPositions,
SerializationHeader header,
Collection<Index.Group> indexGroups,
@ -87,6 +90,7 @@ public class ShardedMultiWriter implements SSTableMultiWriter
this.repairedAt = repairedAt;
this.pendingRepair = pendingRepair;
this.isTransient = isTransient;
this.coordinatorLogBoundaries = coordinatorLogBoundaries;
this.commitLogPositions = commitLogPositions;
this.header = header;
this.indexGroups = indexGroups;
@ -112,6 +116,7 @@ public class ShardedMultiWriter implements SSTableMultiWriter
.setKeyCount(forSplittingKeysBy(boundaries.count()))
.setRepairedAt(repairedAt)
.setPendingRepair(pendingRepair)
.setCoordinatorLogBoundaries(coordinatorLogBoundaries)
.setTransientSSTable(isTransient)
.setTableMetadataRef(cfs.metadata)
.setMetadataCollector(metadataCollector)

4
src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java
View File

@ -27,6 +27,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.DiskBoundaries;
@ -61,6 +62,7 @@ public abstract class CompactionAwareWriter extends Transactional.AbstractTransa
protected final long minRepairedAt;
protected final TimeUUID pendingRepair;
protected final boolean isTransient;
protected final CoordinatorLogBoundaries coordinatorLogBoundaries;
protected final SSTableRewriter sstableWriter;
protected final ILifecycleTransaction txn;
@ -96,6 +98,7 @@ public abstract class CompactionAwareWriter extends Transactional.AbstractTransa
minRepairedAt = CompactionTask.getMinRepairedAt(nonExpiredSSTables);
pendingRepair = CompactionTask.getPendingRepair(nonExpiredSSTables);
isTransient = CompactionTask.getIsTransient(nonExpiredSSTables);
coordinatorLogBoundaries = CompactionTask.getCoordinatorLogBoundaries(nonExpiredSSTables);
DiskBoundaries db = cfs.getDiskBoundaries();
diskBoundaries = db.positions;
locations = db.directories;
@ -328,6 +331,7 @@ public abstract class CompactionAwareWriter extends Transactional.AbstractTransa
.setTransientSSTable(isTransient)
.setRepairedAt(minRepairedAt)
.setPendingRepair(pendingRepair)
.setCoordinatorLogBoundaries(coordinatorLogBoundaries)
.setSecondaryIndexGroups(cfs.indexManager.listIndexGroups())
.addDefaultComponents(cfs.indexManager.listIndexGroups())
.setCompressionDictionaryManager(cfs.compressionDictionaryManager());

1
src/java/org/apache/cassandra/db/memtable/Flushing.java
View File

@ -244,6 +244,7 @@ public class Flushing
ActiveRepairService.UNREPAIRED_SSTABLE,
ActiveRepairService.NO_PENDING_REPAIR,
false,
flushSet.coordinatorLogBoundaries(),
new IntervalSet<>(flushSet.commitLogLowerBound(),
flushSet.commitLogUpperBound()),
new SerializationHeader(true,

11
src/java/org/apache/cassandra/db/memtable/Memtable.java
View File

@ -26,6 +26,7 @@ import javax.annotation.concurrent.NotThreadSafe;
import org.apache.cassandra.db.CellSourceIdentifier;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
@ -37,6 +38,7 @@ import org.apache.cassandra.db.rows.UnfilteredSource;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.io.sstable.format.SSTableWriter;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.FBUtilities;
@ -183,9 +185,9 @@ public interface Memtable extends Comparable<Memtable>, UnfilteredSource, CellSo
// Main write and read operations
default long put(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup)
default long put(MutationId mutationId, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup)
{
return put(update, indexer, opGroup, false);
return put(mutationId, update, indexer, opGroup, false);
}
/**
@ -202,7 +204,7 @@ public interface Memtable extends Comparable<Memtable>, UnfilteredSource, CellSo
* timestamp delta being computed as the difference between the cells and DeletionTimes from any existing partition
* and those in {@code update}. See CASSANDRA-7979.
*/
long put(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing);
long put(MutationId mutationId, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing);
// Read operations are provided by the UnfilteredSource interface.
@ -334,6 +336,9 @@ public interface Memtable extends Comparable<Memtable>, UnfilteredSource, CellSo
/** Statistics required for writing an sstable efficiently */
EncodingStats encodingStats();
/** The boundaries in coordinator logs for all included tracked mutations */
CoordinatorLogBoundaries coordinatorLogBoundaries();
default TableMetadata metadata()
{
return memtable().metadata();

34
src/java/org/apache/cassandra/db/memtable/ShardedSkipListMemtable.java
View File

@ -32,8 +32,11 @@ import org.github.jamm.Unmetered;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.CoordinatorLogBoundariesBuilder;
import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.MutableCoordinatorLogBoundaries;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.Slices;
@ -55,6 +58,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.io.sstable.SSTableReadsListener;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.concurrent.OpOrder;
@ -139,11 +143,11 @@ public class ShardedSkipListMemtable extends AbstractShardedMemtable
*
* commitLogSegmentPosition should only be null if this is a secondary index, in which case it is *expected* to be null
*/
public long put(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing)
public long put(MutationId mutationId, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing)
{
DecoratedKey key = update.partitionKey();
MemtableShard shard = shards[boundaries.getShardForKey(key)];
return shard.put(key, update, indexer, opGroup, assumeMissing);
return shard.put(mutationId, key, update, indexer, opGroup, assumeMissing);
}
/**
@ -287,7 +291,7 @@ public class ShardedSkipListMemtable extends AbstractShardedMemtable
int keyCount = 0;
TableMetadata currentTableMetadata = metadata();
for (Iterator<AtomicBTreePartition> it = getPartitionIterator(from, true, to,false); it.hasNext();)
for (Iterator<AtomicBTreePartition> it = getPartitionIterator(from, true, to, false); it.hasNext(); )
{
AtomicBTreePartition en = it.next();
keySize += en.partitionKey().getKey().remaining();
@ -295,7 +299,15 @@ public class ShardedSkipListMemtable extends AbstractShardedMemtable
}
long partitionKeySize = keySize;
int partitionCount = keyCount;
Iterator<AtomicBTreePartition> toFlush = getPartitionIterator(from, true, to,false);
Iterator<AtomicBTreePartition> toFlush = getPartitionIterator(from, true, to, false);
CoordinatorLogBoundaries flushableBoundaries;
{
CoordinatorLogBoundariesBuilder builder = new CoordinatorLogBoundariesBuilder();
for (MemtableShard shard : shards)
builder.addAll(shard.coordinatorLogBoundaries);
flushableBoundaries = builder.build();
}
return new AbstractFlushablePartitionSet<AtomicBTreePartition>()
{
@ -336,6 +348,12 @@ public class ShardedSkipListMemtable extends AbstractShardedMemtable
{
return tableMetadata;
}
@Override
public CoordinatorLogBoundaries coordinatorLogBoundaries()
{
return flushableBoundaries;
}
};
}
@ -361,6 +379,7 @@ public class ShardedSkipListMemtable extends AbstractShardedMemtable
private final ColumnsCollector columnsCollector;
private final StatsCollector statsCollector;
private final MutableCoordinatorLogBoundaries coordinatorLogBoundaries = MutableCoordinatorLogBoundaries.create();
@Unmetered // total pool size should not be included in memtable's deep size
private final MemtableAllocator allocator;
@ -376,7 +395,7 @@ public class ShardedSkipListMemtable extends AbstractShardedMemtable
this.metadata = metadata;
}
public long put(DecoratedKey key, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing)
public long put(MutationId mutationId, DecoratedKey key, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing)
{
Cloner cloner = allocator.cloner(opGroup);
AtomicBTreePartition previous = assumeMissing ? null : partitions.get(key);
@ -405,6 +424,7 @@ public class ShardedSkipListMemtable extends AbstractShardedMemtable
liveDataSize.addAndGet(initialSize + updater.dataSize);
columnsCollector.update(update.columns());
statsCollector.update(update.stats());
coordinatorLogBoundaries.add(mutationId);
currentOperations.addAndGet(update.operationCount());
return updater.colUpdateTimeDelta;
}
@ -514,13 +534,13 @@ public class ShardedSkipListMemtable extends AbstractShardedMemtable
*
* commitLogSegmentPosition should only be null if this is a secondary index, in which case it is *expected* to be null
*/
public long put(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing)
public long put(MutationId mutationId, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing)
{
DecoratedKey key = update.partitionKey();
MemtableShard shard = shards[boundaries.getShardForKey(key)];
synchronized (shard)
{
return shard.put(key, update, indexer, opGroup, assumeMissing);
return shard.put(mutationId, key, update, indexer, opGroup, assumeMissing);
}
}

18
src/java/org/apache/cassandra/db/memtable/SkipListMemtable.java
View File

@ -32,8 +32,11 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.BufferDecoratedKey;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.CoordinatorLogBoundariesBuilder;
import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.MutableCoordinatorLogBoundaries;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.Slices;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
@ -54,6 +57,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.io.sstable.SSTableReadsListener;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.ObjectSizes;
@ -86,6 +90,8 @@ public class SkipListMemtable extends AbstractAllocatorMemtable
// actually only store DecoratedKey.
private final ConcurrentNavigableMap<PartitionPosition, AtomicBTreePartition> partitions = new ConcurrentSkipListMap<>();
private final MutableCoordinatorLogBoundaries coordinatorLogBoundaries = MutableCoordinatorLogBoundaries.create();
private final AtomicLong liveDataSize = new AtomicLong(0);
protected SkipListMemtable(AtomicReference<CommitLogPosition> commitLogLowerBound, TableMetadataRef metadataRef, Owner owner)
@ -115,7 +121,7 @@ public class SkipListMemtable extends AbstractAllocatorMemtable
* commitLogSegmentPosition should only be null if this is a secondary index, in which case it is *expected* to be null
*/
@Override
public long put(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing)
public long put(MutationId mutationId, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing)
{
long initialSize = 0;
Cloner cloner = allocator.cloner(opGroup);
@ -143,6 +149,7 @@ public class SkipListMemtable extends AbstractAllocatorMemtable
liveDataSize.addAndGet(initialSize + updater.dataSize);
columnsCollector.update(update.columns());
statsCollector.update(update.stats());
coordinatorLogBoundaries.add(mutationId);
currentOperations.addAndGet(update.operationCount());
return updater.colUpdateTimeDelta;
}
@ -288,6 +295,9 @@ public class SkipListMemtable extends AbstractAllocatorMemtable
}
final long partitionKeysSize = keysSize;
final long partitionCount = keyCount;
CoordinatorLogBoundaries flushableBoundaries = new CoordinatorLogBoundariesBuilder()
.addAll(coordinatorLogBoundaries)
.build();
return new AbstractFlushablePartitionSet<AtomicBTreePartition>()
{
@ -334,6 +344,12 @@ public class SkipListMemtable extends AbstractAllocatorMemtable
{
return tableMetadata;
}
@Override
public CoordinatorLogBoundaries coordinatorLogBoundaries()
{
return flushableBoundaries;
}
};
}

26
src/java/org/apache/cassandra/db/memtable/TrieMemtable.java
View File

@ -38,9 +38,12 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.BufferDecoratedKey;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.CoordinatorLogBoundariesBuilder;
import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.DeletionInfo;
import org.apache.cassandra.db.MutableCoordinatorLogBoundaries;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.Slices;
@ -69,6 +72,7 @@ import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.sstable.SSTableReadsListener;
import org.apache.cassandra.metrics.TrieMemtableMetricsView;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.Clock;
@ -183,13 +187,13 @@ public class TrieMemtable extends AbstractShardedMemtable
* commitLogSegmentPosition should only be null if this is a secondary index, in which case it is *expected* to be null
*/
@Override
public long put(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing)
public long put(MutationId mutationId, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, boolean assumeMissing)
{
try
{
DecoratedKey key = update.partitionKey();
MemtableShard shard = shards[boundaries.getShardForKey(key)];
long colUpdateTimeDelta = shard.put(key, update, indexer, opGroup);
long colUpdateTimeDelta = shard.put(mutationId, key, update, indexer, opGroup);
if (shard.data.reachedAllocatedSizeThreshold() && !switchRequested.getAndSet(true))
{
@ -447,6 +451,14 @@ public class TrieMemtable extends AbstractShardedMemtable
partitionKeySize = keySize;
partitionCount = keyCount;
CoordinatorLogBoundaries flushableBoundaries;
{
CoordinatorLogBoundariesBuilder builder = new CoordinatorLogBoundariesBuilder();
for (MemtableShard shard : shards)
builder.addAll(shard.coordinatorLogBoundaries);
flushableBoundaries = builder.build();
}
return new AbstractFlushablePartitionSet<MemtablePartition>()
{
private final TableMetadata tableMetadata = TrieMemtable.this.metadata();
@ -489,6 +501,12 @@ public class TrieMemtable extends AbstractShardedMemtable
{
return tableMetadata;
}
@Override
public CoordinatorLogBoundaries coordinatorLogBoundaries()
{
return flushableBoundaries;
}
};
}
@ -530,6 +548,7 @@ public class TrieMemtable extends AbstractShardedMemtable
private final ColumnsCollector columnsCollector;
private final StatsCollector statsCollector;
private final MutableCoordinatorLogBoundaries coordinatorLogBoundaries = MutableCoordinatorLogBoundaries.create();
@Unmetered // total pool size should not be included in memtable's deep size
private final MemtableAllocator allocator;
@ -547,7 +566,7 @@ public class TrieMemtable extends AbstractShardedMemtable
this.metrics = metrics;
}
public long put(DecoratedKey key, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup) throws InMemoryTrie.SpaceExhaustedException
public long put(MutationId mutationId, DecoratedKey key, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup) throws InMemoryTrie.SpaceExhaustedException
{
BTreePartitionUpdater updater = new BTreePartitionUpdater(allocator, allocator.cloner(opGroup), opGroup, indexer);
boolean locked = writeLock.tryLock();
@ -586,6 +605,7 @@ public class TrieMemtable extends AbstractShardedMemtable
columnsCollector.update(update.columns());
statsCollector.update(update.stats());
coordinatorLogBoundaries.add(mutationId);
}
}
finally

2
src/java/org/apache/cassandra/db/streaming/CassandraCompressedStreamReader.java
View File

@ -74,7 +74,7 @@ public class CassandraCompressedStreamReader extends CassandraStreamReader
try (CompressedInputStream cis = new CompressedInputStream(inputPlus, compressionInfo, ChecksumType.CRC32, cfs::getCrcCheckChance))
{
TrackedDataInputPlus in = new TrackedDataInputPlus(cis);
writer = createWriter(cfs, totalSize, repairedAt, pendingRepair, inputVersion.format);
writer = createWriter(cfs, totalSize, repairedAt, pendingRepair, coordinatorLogBoundaries, inputVersion.format);
deserializer = new StreamDeserializer(cfs.metadata(), in, inputVersion, getHeader(cfs.metadata()), session, writer);
String filename = writer.getFilename();
String sectionName = filename + '-' + fileSeqNum;

7
src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java
View File

@ -26,6 +26,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
@ -151,6 +152,12 @@ public class CassandraOutgoingFile implements OutgoingStream
return ref.get().getPendingRepair();
}
@Override
public CoordinatorLogBoundaries getCoordinatorLogBoundaries()
{
return ref.get().getCoordinatorLogBoundaries();
}
@Override
public void write(StreamSession session, StreamingDataOutputPlus out, int version) throws IOException
{

9
src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java
View File

@ -31,6 +31,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.Directories;
@ -87,6 +88,7 @@ public class CassandraStreamReader implements IStreamReader
protected final Version inputVersion;
protected final long repairedAt;
protected final TimeUUID pendingRepair;
protected final CoordinatorLogBoundaries coordinatorLogBoundaries;
protected final int sstableLevel;
protected final SerializationHeader.Component header;
protected final int fileSeqNum;
@ -106,6 +108,7 @@ public class CassandraStreamReader implements IStreamReader
this.inputVersion = streamHeader.version;
this.repairedAt = header.repairedAt;
this.pendingRepair = header.pendingRepair;
this.coordinatorLogBoundaries = header.coordinatorLogBoundaries;
this.sstableLevel = streamHeader.sstableLevel;
this.header = streamHeader.serializationHeader;
this.fileSeqNum = header.sequenceNumber;
@ -135,7 +138,7 @@ public class CassandraStreamReader implements IStreamReader
try (StreamCompressionInputStream streamCompressionInputStream = new StreamCompressionInputStream(inputPlus, current_version))
{
TrackedDataInputPlus in = new TrackedDataInputPlus(streamCompressionInputStream);
writer = createWriter(cfs, totalSize, repairedAt, pendingRepair, inputVersion.format);
writer = createWriter(cfs, totalSize, repairedAt, pendingRepair, coordinatorLogBoundaries, inputVersion.format);
deserializer = getDeserializer(cfs.metadata(), in, inputVersion, session, writer);
String sequenceName = writer.getFilename() + '-' + fileSeqNum;
long lastBytesRead = 0;
@ -176,7 +179,7 @@ public class CassandraStreamReader implements IStreamReader
{
return header != null? header.toHeader(metadata) : null; //pre-3.0 sstable have no SerializationHeader
}
protected SSTableTxnSingleStreamWriter createWriter(ColumnFamilyStore cfs, long totalSize, long repairedAt, TimeUUID pendingRepair, SSTableFormat<?, ?> format) throws IOException
protected SSTableTxnSingleStreamWriter createWriter(ColumnFamilyStore cfs, long totalSize, long repairedAt, TimeUUID pendingRepair, CoordinatorLogBoundaries coordinatorLogBoundaries, SSTableFormat<?, ?> format) throws IOException
{
Directories.DataDirectory localDir = cfs.getDirectories().getWriteableLocation(totalSize);
if (localDir == null)
@ -185,7 +188,7 @@ public class CassandraStreamReader implements IStreamReader
StreamReceiver streamReceiver = session.getAggregator(tableId);
Preconditions.checkState(streamReceiver instanceof CassandraStreamReceiver);
ILifecycleTransaction txn = createTxn();
RangeAwareSSTableWriter writer = new RangeAwareSSTableWriter(cfs, estimatedKeys, repairedAt, pendingRepair, false, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()));
RangeAwareSSTableWriter writer = new RangeAwareSSTableWriter(cfs, estimatedKeys, repairedAt, pendingRepair, false, coordinatorLogBoundaries, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()));
return new SSTableTxnSingleStreamWriter(txn, writer);
}

5
src/java/org/apache/cassandra/index/internal/CassandraIndex.java
View File

@ -76,6 +76,7 @@ import org.apache.cassandra.index.internal.keys.KeysSearcher;
import org.apache.cassandra.index.transactions.IndexTransaction;
import org.apache.cassandra.io.sstable.ReducingKeyIterator;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.TableMetadata;
@ -544,7 +545,7 @@ public abstract class CassandraIndex implements Index
cell));
Row row = BTreeRow.noCellLiveRow(buildIndexClustering(rowKey, clustering, cell), info);
PartitionUpdate upd = partitionUpdate(valueKey, row);
indexCfs.getWriteHandler().write(upd, ctx, false);
indexCfs.getWriteHandler().write(MutationId.fixme(), upd, ctx, false);
logger.trace("Inserted entry into index for value {}", valueKey);
}
@ -590,7 +591,7 @@ public abstract class CassandraIndex implements Index
{
Row row = BTreeRow.emptyDeletedRow(indexClustering, Row.Deletion.regular(deletion));
PartitionUpdate upd = partitionUpdate(indexKey, row);
indexCfs.getWriteHandler().write(upd, ctx, false);
indexCfs.getWriteHandler().write(MutationId.fixme(), upd, ctx, false);
logger.trace("Removed index entry for value {}", indexKey);
}

13
src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
View File

@ -30,7 +30,9 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.stream.Stream;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.SerializationHeader;
@ -69,6 +71,9 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
indexGroups = new ArrayList<>();
}
@VisibleForTesting
public abstract long bytesWritten();
protected void setSSTableFormatType(SSTableFormat<?, ?> type)
{
this.format = type;
@ -117,10 +122,15 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
protected SSTableTxnWriter createWriter(SSTable.Owner owner) throws IOException
{
// This will prevent cassandra-analytics from producing SSTables for tables with mutation tracking enabled.
// We'll eventually support this with coordinated nodetool import
if (metadata.get().keyspaceReplicationType.isTracked())
throw new IllegalStateException("Can't create writer for table with mutation tracking enabled");
SerializationHeader header = new SerializationHeader(true, metadata.get(), columns, EncodingStats.NO_STATS);
if (makeRangeAware)
return SSTableTxnWriter.createRangeAware(metadata, 0, ActiveRepairService.UNREPAIRED_SSTABLE, ActiveRepairService.NO_PENDING_REPAIR, false, format, header);
return SSTableTxnWriter.createRangeAware(metadata, 0, ActiveRepairService.UNREPAIRED_SSTABLE, ActiveRepairService.NO_PENDING_REPAIR, false, CoordinatorLogBoundaries.NONE, format, header);
SSTable.Owner effectiveOwner;
@ -142,6 +152,7 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
ActiveRepairService.UNREPAIRED_SSTABLE,
ActiveRepairService.NO_PENDING_REPAIR,
false,
CoordinatorLogBoundaries.NONE,
header,
indexGroups,
effectiveOwner);

7
src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
View File

@ -35,6 +35,7 @@ import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Sets;
@ -161,6 +162,12 @@ public class CQLSSTableWriter implements Closeable
.collect(Collectors.toList());
}
@VisibleForTesting
public long bytesWritten()
{
return writer.bytesWritten();
}
/**
* Returns a new builder for a CQLSSTableWriter.
*

9
src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java
View File

@ -23,6 +23,7 @@ import java.util.Collection;
import java.util.List;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.DiskBoundaries;
@ -45,6 +46,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
private final long repairedAt;
private final TimeUUID pendingRepair;
private final boolean isTransient;
private final CoordinatorLogBoundaries coordinatorLogBoundaries;
private final SSTableFormat<?, ?> format;
private final SerializationHeader header;
private final ILifecycleTransaction txn;
@ -53,7 +55,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
private final List<SSTableMultiWriter> finishedWriters = new ArrayList<>();
private SSTableMultiWriter currentWriter = null;
public RangeAwareSSTableWriter(ColumnFamilyStore cfs, long estimatedKeys, long repairedAt, TimeUUID pendingRepair, boolean isTransient, SSTableFormat<?, ?> format, int sstableLevel, long totalSize, ILifecycleTransaction txn, SerializationHeader header) throws IOException
public RangeAwareSSTableWriter(ColumnFamilyStore cfs, long estimatedKeys, long repairedAt, TimeUUID pendingRepair, boolean isTransient, CoordinatorLogBoundaries coordinatorLogBoundaries, SSTableFormat<?, ?> format, int sstableLevel, long totalSize, ILifecycleTransaction txn, SerializationHeader header) throws IOException
{
DiskBoundaries db = cfs.getDiskBoundaries();
directories = db.directories;
@ -63,6 +65,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
this.repairedAt = repairedAt;
this.pendingRepair = pendingRepair;
this.isTransient = isTransient;
this.coordinatorLogBoundaries = coordinatorLogBoundaries;
this.format = format;
this.txn = txn;
this.header = header;
@ -74,7 +77,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
throw new IOException(String.format("Insufficient disk space to store %s",
FBUtilities.prettyPrintMemory(totalSize)));
Descriptor desc = cfs.newSSTableDescriptor(cfs.getDirectories().getLocationForDisk(localDir), format);
currentWriter = cfs.createSSTableMultiWriter(desc, estimatedKeys, repairedAt, pendingRepair, isTransient, null, sstableLevel, header, txn);
currentWriter = cfs.createSSTableMultiWriter(desc, estimatedKeys, repairedAt, pendingRepair, isTransient, coordinatorLogBoundaries, null, sstableLevel, header, txn);
}
}
@ -96,7 +99,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
finishedWriters.add(currentWriter);
Descriptor desc = cfs.newSSTableDescriptor(cfs.getDirectories().getLocationForDisk(directories.get(currentIndex)), format);
currentWriter = cfs.createSSTableMultiWriter(desc, estimatedKeys, repairedAt, pendingRepair, isTransient, null, sstableLevel, header, txn);
currentWriter = cfs.createSSTableMultiWriter(desc, estimatedKeys, repairedAt, pendingRepair, isTransient, coordinatorLogBoundaries, null, sstableLevel, header, txn);
}
}

8
src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
View File

@ -61,6 +61,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
private Buffer buffer = new Buffer();
private final long maxSStableSizeInBytes;
private long currentSize;
private long syncedSize;
// Used to compute the row serialized size
private final SerializationHeader header;
@ -100,6 +101,12 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
return previous;
}
@Override
public long bytesWritten()
{
return currentSize + syncedSize;
}
private void countRow(Row row)
{
// Note that the accounting of a row is a bit inaccurate (it doesn't take some of the file format optimization into account)
@ -164,6 +171,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
put(buffer);
buffer = new Buffer();
syncedSize += currentSize;
currentSize = 0;
}

8
src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java
View File

@ -52,6 +52,7 @@ class SSTableSimpleWriter extends AbstractSSTableSimpleWriter
protected PartitionUpdate.Builder update;
private SSTableTxnWriter writer;
private long flushedBytes;
/**
* Create a SSTable writer for sorted input data.
@ -160,6 +161,7 @@ class SSTableSimpleWriter extends AbstractSSTableSimpleWriter
if (writer == null)
return;
flushedBytes += writer.getOnDiskBytesWritten();
Collection<SSTableReader> finished = writer.finish(shouldOpenSSTables());
notifySSTableProduced(finished);
}
@ -175,4 +177,10 @@ class SSTableSimpleWriter extends AbstractSSTableSimpleWriter
{
getOrCreateWriter().append(update.unfilteredIterator());
}
@Override
public long bytesWritten()
{
return flushedBytes + (writer != null ? writer.getOnDiskBytesWritten() : 0);
}
}

16
src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java
View File

@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.Collection;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.compaction.OperationType;
@ -76,6 +77,11 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
return writer.getOnDiskBytesWritten();
}
public long getBytesWritten()
{
return writer.getBytesWritten();
}
protected Throwable doCommit(Throwable accumulate)
{
return writer.commit(txn.commit(accumulate));
@ -108,10 +114,10 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
}
@SuppressWarnings({"resource", "RedundantSuppression"}) // log and writer closed during doPostCleanup
public static SSTableTxnWriter create(ColumnFamilyStore cfs, Descriptor descriptor, long keyCount, long repairedAt, TimeUUID pendingRepair, boolean isTransient, SerializationHeader header)
public static SSTableTxnWriter create(ColumnFamilyStore cfs, Descriptor descriptor, long keyCount, long repairedAt, TimeUUID pendingRepair, boolean isTransient, CoordinatorLogBoundaries coordinatorLogBoundaries, SerializationHeader header)
{
LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.WRITE);
SSTableMultiWriter writer = cfs.createSSTableMultiWriter(descriptor, keyCount, repairedAt, pendingRepair, isTransient, header, txn);
SSTableMultiWriter writer = cfs.createSSTableMultiWriter(descriptor, keyCount, repairedAt, pendingRepair, isTransient, coordinatorLogBoundaries, header, txn);
return new SSTableTxnWriter(txn, writer);
}
@ -120,6 +126,7 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
SSTableFormat<?, ?> type,
SerializationHeader header)
{
@ -129,7 +136,7 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
SSTableMultiWriter writer;
try
{
writer = new RangeAwareSSTableWriter(cfs, keyCount, repairedAt, pendingRepair, isTransient, type, 0, 0, txn, header);
writer = new RangeAwareSSTableWriter(cfs, keyCount, repairedAt, pendingRepair, isTransient, coordinatorLogBoundaries, type, 0, 0, txn, header);
}
catch (IOException e)
{
@ -148,13 +155,14 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
SerializationHeader header,
Collection<Index.Group> indexGroups,
SSTable.Owner owner)
{
// if the column family store does not exist, we create a new default SSTableMultiWriter to use:
LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.WRITE);
SSTableMultiWriter writer = SimpleSSTableMultiWriter.create(descriptor, keyCount, repairedAt, pendingRepair, isTransient, metadata, null, 0, header, indexGroups, txn, owner);
SSTableMultiWriter writer = SimpleSSTableMultiWriter.create(descriptor, keyCount, repairedAt, pendingRepair, isTransient, coordinatorLogBoundaries, metadata, null, 0, header, indexGroups, txn, owner);
return new SSTableTxnWriter(txn, writer);
}
}

3
src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.io.sstable;
import java.util.Collection;
import java.util.Collections;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.db.commitlog.IntervalSet;
@ -117,6 +118,7 @@ public class SimpleSSTableMultiWriter implements SSTableMultiWriter
long repairedAt,
TimeUUID pendingRepair,
boolean isTransient,
CoordinatorLogBoundaries coordinatorLogBoundaries,
TableMetadataRef metadata,
IntervalSet<CommitLogPosition> commitLogPositions,
int sstableLevel,
@ -136,6 +138,7 @@ public class SimpleSSTableMultiWriter implements SSTableMultiWriter
.setRepairedAt(repairedAt)
.setPendingRepair(pendingRepair)
.setTransientSSTable(isTransient)
.setCoordinatorLogBoundaries(coordinatorLogBoundaries)
.setTableMetadataRef(metadata)
.setMetadataCollector(metadataCollector)
.setSerializationHeader(header)

6
src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
View File

@ -58,6 +58,7 @@ import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.SerializationHeader;
@ -1229,6 +1230,11 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource,
return sstableMetadata.pendingRepair;
}
public CoordinatorLogBoundaries getCoordinatorLogBoundaries()
{
return sstableMetadata.coordinatorLogBoundaries;
}
public long getRepairedAt()
{
return sstableMetadata.repairedAt;

17
src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java
View File

@ -38,6 +38,7 @@ import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.compression.CompressionDictionaryManager;
@ -77,6 +78,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional
protected long repairedAt;
protected TimeUUID pendingRepair;
protected boolean isTransient;
protected CoordinatorLogBoundaries coordinatorLogBoundaries;
protected long maxDataAge = -1;
protected final long keyCount;
public final MetadataCollector metadataCollector;
@ -100,11 +102,13 @@ public abstract class SSTableWriter extends SSTable implements Transactional
checkNotNull(builder.getIndexGroups());
checkNotNull(builder.getMetadataCollector());
checkNotNull(builder.getSerializationHeader());
checkNotNull(builder.getCoordinatorLogBoundaries());
this.keyCount = builder.getKeyCount();
this.repairedAt = builder.getRepairedAt();
this.pendingRepair = builder.getPendingRepair();
this.isTransient = builder.isTransientSSTable();
this.coordinatorLogBoundaries = builder.getCoordinatorLogBoundaries();
this.metadataCollector = builder.getMetadataCollector();
this.header = builder.getSerializationHeader();
this.mmappedRegionsCache = builder.getMmappedRegionsCache();
@ -348,6 +352,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional
repairedAt,
pendingRepair,
isTransient,
coordinatorLogBoundaries,
header,
first.retainable().getKey(),
last.retainable().getKey());
@ -454,6 +459,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional
private List<Index.Group> indexGroups;
@Nullable
private CompressionDictionaryManager compressionDictionaryManager;
private CoordinatorLogBoundaries coordinatorLogBoundaries;
public B setMetadataCollector(MetadataCollector metadataCollector)
{
@ -479,6 +485,12 @@ public abstract class SSTableWriter extends SSTable implements Transactional
return (B) this;
}
public B setCoordinatorLogBoundaries(CoordinatorLogBoundaries coordinatorLogBoundaries)
{
this.coordinatorLogBoundaries = coordinatorLogBoundaries;
return (B) this;
}
public B setTransientSSTable(boolean transientSSTable)
{
this.transientSSTable = transientSSTable;
@ -569,6 +581,11 @@ public abstract class SSTableWriter extends SSTable implements Transactional
return transientSSTable;
}
public CoordinatorLogBoundaries getCoordinatorLogBoundaries()
{
return coordinatorLogBoundaries;
}
public SerializationHeader getSerializationHeader()
{
return serializationHeader;

4
src/java/org/apache/cassandra/io/sstable/format/SortedTableScrubber.java
View File

@ -190,7 +190,7 @@ public abstract class SortedTableScrubber<R extends SSTableReaderWithFilter> imp
Refs<SSTableReader> refs = Refs.ref(Collections.singleton(sstable)))
{
StatsMetadata metadata = sstable.getSSTableMetadata();
writer.switchWriter(CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, metadata.repairedAt, metadata.pendingRepair, metadata.isTransient, sstable, transaction));
writer.switchWriter(CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, metadata.repairedAt, metadata.pendingRepair, metadata.isTransient, metadata.coordinatorLogBoundaries, sstable, transaction));
scrubInternal(writer);
@ -240,7 +240,7 @@ public abstract class SortedTableScrubber<R extends SSTableReaderWithFilter> imp
// out of order partitions/rows, but no bad partition found - we can keep our repairedAt time
long repairedAt = badPartitions > 0 ? ActiveRepairService.UNREPAIRED_SSTABLE : sstable.getSSTableMetadata().repairedAt;
SSTableReader newInOrderSstable;
try (SSTableWriter inOrderWriter = CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, repairedAt, metadata.pendingRepair, metadata.isTransient, sstable, transaction))
try (SSTableWriter inOrderWriter = CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, repairedAt, metadata.pendingRepair, metadata.isTransient, metadata.coordinatorLogBoundaries, sstable, transaction))
{
for (Partition partition : outOfOrder)
inOrderWriter.append(partition.unfilteredIterator());

2
src/java/org/apache/cassandra/io/sstable/format/Version.java
View File

@ -96,6 +96,8 @@ public abstract class Version
*/
public abstract boolean hasTokenSpaceCoverage();
public abstract boolean hasMutationTrackingMetadata();
/**
* Records in th stats if the sstable has any partition deletions.
*/

13
src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java
View File

@ -434,7 +434,7 @@ public class BigFormat extends AbstractSSTableFormat<BigTableReader, BigTableWri
static class BigVersion extends Version
{
public static final String current_version = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? "nb" :
DatabaseDescriptor.getStorageCompatibilityMode().isBefore(6) ? "oa" : "pa";
DatabaseDescriptor.getStorageCompatibilityMode().isBefore(6) ? "oa" : "pb";
public static final String earliest_supported_version = "ma";
// ma (3.0.0): swap bf hash order
@ -450,6 +450,7 @@ public class BigFormat extends AbstractSSTableFormat<BigTableReader, BigTableWri
// Long deletionTime to prevent TTL overflow
// token space coverage
// pa (6.0): compression dictionary metadata in CompressionInfo component
// pb (6.x): mutation tracking metadata
//
// NOTE: When adding a new version:
// - Please add it to LegacySSTableTest
@ -471,6 +472,7 @@ public class BigFormat extends AbstractSSTableFormat<BigTableReader, BigTableWri
private final boolean hasKeyRange;
private final boolean hasUintDeletionTime;
private final boolean hasTokenSpaceCoverage;
private final boolean hasMutationTrackingMetadata;
/**
* CASSANDRA-9067: 4.0 bloom filter representation changed (two longs just swapped)
@ -485,7 +487,7 @@ public class BigFormat extends AbstractSSTableFormat<BigTableReader, BigTableWri
isLatestVersion = version.compareTo(current_version) == 0;
// Note that, we probably forgot to change that to 40 for N version, and therefore we cannot do it now.
correspondingMessagingVersion = version.compareTo("oa") >= 0 ? MessagingService.VERSION_50 : MessagingService.VERSION_30;
correspondingMessagingVersion = version.compareTo("ob") >= 0 ? (version.compareTo("oa") > 0 ? MessagingService.VERSION_61 : MessagingService.VERSION_60) : MessagingService.VERSION_30;
hasCommitLogLowerBound = version.compareTo("mb") >= 0;
hasCommitLogIntervals = version.compareTo("mc") >= 0;
@ -503,6 +505,7 @@ public class BigFormat extends AbstractSSTableFormat<BigTableReader, BigTableWri
hasKeyRange = version.compareTo("oa") >= 0;
hasUintDeletionTime = version.compareTo("oa") >= 0;
hasTokenSpaceCoverage = version.compareTo("oa") >= 0;
hasMutationTrackingMetadata = version.compareTo("ob") >= 0;
}
@Override
@ -589,6 +592,12 @@ public class BigFormat extends AbstractSSTableFormat<BigTableReader, BigTableWri
return hasTokenSpaceCoverage;
}
@Override
public boolean hasMutationTrackingMetadata()
{
return hasMutationTrackingMetadata;
}
@Override
public boolean hasPartitionLevelDeletionsPresenceMarker()
{

13
src/java/org/apache/cassandra/io/sstable/format/bti/BtiFormat.java
View File

@ -287,24 +287,27 @@ public class BtiFormat extends AbstractSSTableFormat<BtiTableReader, BtiTableWri
static class BtiVersion extends Version
{
public static final String current_version = "ea";
public static final String current_version = "eb";
public static final String earliest_supported_version = "da";
// versions aa-cz are not supported in OSS
// da (5.0): initial version of the BTI format
// ea (6.0): compression dictionary metadata in CompressionInfo component
// eb (6.x): mutation tracking metadata
// NOTE: when adding a new version, please add that to LegacySSTableTest, too.
private final boolean isLatestVersion;
private final int correspondingMessagingVersion;
private final boolean hasMutationTrackingMetadata;
BtiVersion(BtiFormat format, String version)
{
super(format, version);
isLatestVersion = version.compareTo(current_version) == 0;
correspondingMessagingVersion = MessagingService.VERSION_50;
correspondingMessagingVersion = version.compareTo("ea") > 0 ? MessagingService.VERSION_61 : MessagingService.VERSION_50;
hasMutationTrackingMetadata = version.compareTo("db") >= 0;
}
@Override
@ -389,6 +392,12 @@ public class BtiFormat extends AbstractSSTableFormat<BtiTableReader, BtiTableWri
return true;
}
@Override
public boolean hasMutationTrackingMetadata()
{
return hasMutationTrackingMetadata;
}
@Override
public boolean hasPartitionLevelDeletionsPresenceMarker()
{

36
src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.io.sstable.metadata;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.EnumMap;
import java.util.Map;
import java.util.UUID;
@ -33,6 +32,7 @@ import org.apache.cassandra.db.ClusteringBound;
import org.apache.cassandra.db.ClusteringBoundOrBoundary;
import org.apache.cassandra.db.ClusteringComparator;
import org.apache.cassandra.db.ClusteringPrefix;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.LivenessInfo;
import org.apache.cassandra.db.SerializationHeader;
@ -48,9 +48,7 @@ import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.io.sstable.ClusteringDescriptor;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.EstimatedHistogram;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MurmurHash;
@ -82,35 +80,6 @@ public class MetadataCollector implements PartitionStatisticsCollector
return TombstoneHistogram.createDefault();
}
public static StatsMetadata defaultStatsMetadata()
{
return new StatsMetadata(defaultPartitionSizeHistogram(),
defaultCellPerPartitionCountHistogram(),
IntervalSet.empty(),
Long.MIN_VALUE,
Long.MAX_VALUE,
Integer.MAX_VALUE,
Integer.MAX_VALUE,
0,
Integer.MAX_VALUE,
NO_COMPRESSION_RATIO,
defaultTombstoneDropTimeHistogram(),
0,
Collections.emptyList(),
Slice.ALL,
true,
ActiveRepairService.UNREPAIRED_SSTABLE,
-1,
-1,
Double.NaN,
null,
null,
false,
true,
ByteBufferUtil.EMPTY_BYTE_BUFFER,
ByteBufferUtil.EMPTY_BYTE_BUFFER);
}
protected EstimatedHistogram estimatedPartitionSize = defaultPartitionSizeHistogram();
// TODO: count the number of row per partition (either with the number of cells, or instead)
protected EstimatedHistogram estimatedCellPerPartitionCount = defaultCellPerPartitionCountHistogram();
@ -444,7 +413,7 @@ public class MetadataCollector implements PartitionStatisticsCollector
return totalRows;
}
public Map<MetadataType, MetadataComponent> finalizeMetadata(String partitioner, double bloomFilterFPChance, long repairedAt, TimeUUID pendingRepair, boolean isTransient, SerializationHeader header, ByteBuffer firstKey, ByteBuffer lastKey)
public Map<MetadataType, MetadataComponent> finalizeMetadata(String partitioner, double bloomFilterFPChance, long repairedAt, TimeUUID pendingRepair, boolean isTransient, CoordinatorLogBoundaries coordinatorLogBoundaries, SerializationHeader header, ByteBuffer firstKey, ByteBuffer lastKey)
{
assert minClustering.kind() == ClusteringPrefix.Kind.CLUSTERING || minClustering.kind().isStart();
assert maxClustering.kind() == ClusteringPrefix.Kind.CLUSTERING || maxClustering.kind().isEnd();
@ -481,6 +450,7 @@ public class MetadataCollector implements PartitionStatisticsCollector
pendingRepair,
isTransient,
hasPartitionLevelDeletions,
coordinatorLogBoundaries,
firstKey,
lastKey));
components.put(MetadataType.COMPACTION, new CompactionMetadata(cardinality));

14
src/java/org/apache/cassandra/io/sstable/metadata/MetadataSerializer.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.io.sstable.metadata;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.file.NoSuchFileException;
import java.util.Collections;
import java.util.EnumMap;
import java.util.EnumSet;
@ -124,17 +125,10 @@ public class MetadataSerializer implements IMetadataSerializer
logger.trace("Load metadata for {}", descriptor);
File statsFile = descriptor.fileFor(Components.STATS);
if (!statsFile.exists())
throw new NoSuchFileException("Stats component of sstable " + descriptor + " is missing");
try (RandomAccessReader r = RandomAccessReader.open(statsFile))
{
logger.trace("No sstable stats for {}", descriptor);
components = new EnumMap<>(MetadataType.class);
components.put(MetadataType.STATS, MetadataCollector.defaultStatsMetadata());
}
else
{
try (RandomAccessReader r = RandomAccessReader.open(statsFile))
{
components = deserialize(descriptor, r, types);
}
components = deserialize(descriptor, r, types);
}
return components;
}

19
src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java
View File

@ -30,6 +30,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.BufferClusteringBound;
import org.apache.cassandra.db.ClusteringBound;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.Slice;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
@ -79,6 +80,7 @@ public class StatsMetadata extends MetadataComponent
public final UUID originatingHostId;
public final TimeUUID pendingRepair;
public final boolean isTransient;
public final CoordinatorLogBoundaries coordinatorLogBoundaries;
// just holds the current encoding stats to avoid allocating - it is not serialized
public final EncodingStats encodingStats;
@ -122,6 +124,7 @@ public class StatsMetadata extends MetadataComponent
TimeUUID pendingRepair,
boolean isTransient,
boolean hasPartitionLevelDeletions,
CoordinatorLogBoundaries coordinatorLogBoundaries,
ByteBuffer firstKey,
ByteBuffer lastKey)
{
@ -147,6 +150,7 @@ public class StatsMetadata extends MetadataComponent
this.originatingHostId = originatingHostId;
this.pendingRepair = pendingRepair;
this.isTransient = isTransient;
this.coordinatorLogBoundaries = coordinatorLogBoundaries;
this.encodingStats = new EncodingStats(minTimestamp, minLocalDeletionTime, minTTL);
this.hasPartitionLevelDeletions = hasPartitionLevelDeletions;
this.firstKey = firstKey;
@ -207,6 +211,7 @@ public class StatsMetadata extends MetadataComponent
pendingRepair,
isTransient,
hasPartitionLevelDeletions,
coordinatorLogBoundaries,
firstKey,
lastKey);
}
@ -236,6 +241,7 @@ public class StatsMetadata extends MetadataComponent
newPendingRepair,
newIsTransient,
hasPartitionLevelDeletions,
coordinatorLogBoundaries,
firstKey,
lastKey);
}
@ -269,6 +275,7 @@ public class StatsMetadata extends MetadataComponent
.append(originatingHostId, that.originatingHostId)
.append(pendingRepair, that.pendingRepair)
.append(hasPartitionLevelDeletions, that.hasPartitionLevelDeletions)
.append(coordinatorLogBoundaries, that.coordinatorLogBoundaries)
.append(firstKey, that.firstKey)
.append(lastKey, that.lastKey)
.build();
@ -299,6 +306,7 @@ public class StatsMetadata extends MetadataComponent
.append(originatingHostId)
.append(pendingRepair)
.append(hasPartitionLevelDeletions)
.append(coordinatorLogBoundaries)
.append(firstKey)
.append(lastKey)
.build();
@ -386,6 +394,9 @@ public class StatsMetadata extends MetadataComponent
size += Double.BYTES;
}
if (version.hasMutationTrackingMetadata())
size += CoordinatorLogBoundaries.serializer.serializedSize(component.coordinatorLogBoundaries, version.correspondingMessagingVersion());
return size;
}
@ -509,6 +520,9 @@ public class StatsMetadata extends MetadataComponent
{
out.writeDouble(component.tokenSpaceCoverage);
}
if (version.hasMutationTrackingMetadata())
CoordinatorLogBoundaries.serializer.serialize(component.coordinatorLogBoundaries, out, version.correspondingMessagingVersion());
}
private void serializeImprovedMinMax(Version version, StatsMetadata component, DataOutputPlus out) throws IOException
@ -654,6 +668,10 @@ public class StatsMetadata extends MetadataComponent
tokenSpaceCoverage = in.readDouble();
}
CoordinatorLogBoundaries coordinatorLogBoundaries = CoordinatorLogBoundaries.NONE;
if (version.hasMutationTrackingMetadata())
coordinatorLogBoundaries = CoordinatorLogBoundaries.serializer.deserialize(in, version.correspondingMessagingVersion());
return new StatsMetadata(partitionSizes,
columnCounts,
commitLogIntervals,
@ -677,6 +695,7 @@ public class StatsMetadata extends MetadataComponent
pendingRepair,
isTransient,
hasPartitionLevelDeletions,
coordinatorLogBoundaries,
firstKey,
lastKey);
}

5
src/java/org/apache/cassandra/replication/CoordinatorLogId.java
View File

@ -26,6 +26,8 @@ import java.io.IOException;
import java.io.Serializable;
import java.util.Comparator;
import com.google.common.annotations.VisibleForTesting;
public class CoordinatorLogId implements Serializable
{
private static final CoordinatorLogId NONE = new CoordinatorLogId(Integer.MIN_VALUE, Integer.MIN_VALUE);
@ -66,7 +68,8 @@ public class CoordinatorLogId implements Serializable
return asLong(hostId, hostLogId);
}
static long asLong(int hostId, int hostLogId)
@VisibleForTesting
public static long asLong(int hostId, int hostLogId)
{
return ((long) hostId << 32) | hostLogId;
}

3
src/java/org/apache/cassandra/schema/TableMetadata.java
View File

@ -654,6 +654,9 @@ public class TableMetadata implements SchemaElement
indexes.validate(this);
if (replicationType() != null && replicationType().isTracked() && params.memtable.factory().streamFromMemtable())
except("Cannot use mutation tracking with persistent memtables");
for (ColumnMetadata columnMetadata : columns())
{
ColumnConstraints constraints = columnMetadata.getColumnConstraints();

3
src/java/org/apache/cassandra/streaming/OutgoingStream.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.streaming;
import java.io.IOException;
import java.util.List;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.schema.TableId;
@ -48,6 +49,8 @@ public interface OutgoingStream
long getRepairedAt();
TimeUUID getPendingRepair();
CoordinatorLogBoundaries getCoordinatorLogBoundaries();
String getName();
/**

3
src/java/org/apache/cassandra/streaming/messages/OutgoingStreamMessage.java
View File

@ -74,7 +74,8 @@ public class OutgoingStreamMessage extends StreamMessage
session.sessionIndex(),
sequenceNumber,
stream.getRepairedAt(),
stream.getPendingRepair());
stream.getPendingRepair(),
stream.getCoordinatorLogBoundaries());
}
public synchronized void serialize(StreamingDataOutputPlus out, int version, StreamSession session) throws IOException

11
src/java/org/apache/cassandra/streaming/messages/StreamMessageHeader.java
View File

@ -21,6 +21,7 @@ import java.io.IOException;
import com.google.common.base.Objects;
import org.apache.cassandra.db.CoordinatorLogBoundaries;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -46,6 +47,7 @@ public class StreamMessageHeader
public final int sequenceNumber;
public final long repairedAt;
public final TimeUUID pendingRepair;
public final CoordinatorLogBoundaries coordinatorLogBoundaries;
public final InetAddressAndPort sender;
public StreamMessageHeader(TableId tableId,
@ -55,7 +57,8 @@ public class StreamMessageHeader
int sessionIndex,
int sequenceNumber,
long repairedAt,
TimeUUID pendingRepair)
TimeUUID pendingRepair,
CoordinatorLogBoundaries coordinatorLogBoundaries)
{
this.tableId = tableId;
this.sender = sender;
@ -65,6 +68,7 @@ public class StreamMessageHeader
this.sequenceNumber = sequenceNumber;
this.repairedAt = repairedAt;
this.pendingRepair = pendingRepair;
this.coordinatorLogBoundaries = coordinatorLogBoundaries;
}
@Override
@ -119,6 +123,7 @@ public class StreamMessageHeader
{
header.pendingRepair.serialize(out);
}
CoordinatorLogBoundaries.serializer.serialize(header.coordinatorLogBoundaries, out, version);
}
public StreamMessageHeader deserialize(DataInputPlus in, int version) throws IOException
@ -131,8 +136,9 @@ public class StreamMessageHeader
int sequenceNumber = in.readInt();
long repairedAt = in.readLong();
TimeUUID pendingRepair = in.readBoolean() ? TimeUUID.deserialize(in) : null;
CoordinatorLogBoundaries coordinatorLogBoundaries = CoordinatorLogBoundaries.serializer.deserialize(in, version);
return new StreamMessageHeader(tableId, sender, planId, sendByFollower, sessionIndex, sequenceNumber, repairedAt, pendingRepair);
return new StreamMessageHeader(tableId, sender, planId, sendByFollower, sessionIndex, sequenceNumber, repairedAt, pendingRepair, coordinatorLogBoundaries);
}
public long serializedSize(StreamMessageHeader header, int version)
@ -146,6 +152,7 @@ public class StreamMessageHeader
size += TypeSizes.sizeof(header.repairedAt);
size += TypeSizes.sizeof(header.pendingRepair != null);
size += header.pendingRepair != null ? TimeUUID.sizeInBytes() : 0;
size += CoordinatorLogBoundaries.serializer.serializedSize(header.coordinatorLogBoundaries, version);
return size;
}

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-CompressionInfo.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Data.db Normal file
View File

Binary file not shown.

1
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Digest.crc32 Normal file
View File

@ -0,0 +1 @@
2530067741

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Filter.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Partitions.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Rows.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Statistics.db Normal file
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-TOC.txt Normal file
View File

@ -0,0 +1,8 @@
Data.db
Statistics.db
Digest.crc32
TOC.txt
CompressionInfo.db
Filter.db
Partitions.db
Rows.db

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-CompressionInfo.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Data.db Normal file
View File

Binary file not shown.

1
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Digest.crc32 Normal file
View File

@ -0,0 +1 @@
1370392555

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Filter.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Partitions.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Rows.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Statistics.db Normal file
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-TOC.txt Normal file
View File

@ -0,0 +1,8 @@
CompressionInfo.db
Data.db
Digest.crc32
Filter.db
Partitions.db
Rows.db
Statistics.db
TOC.txt

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-CompressionInfo.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Data.db Normal file
View File

Binary file not shown.

1
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Digest.crc32 Normal file
View File

@ -0,0 +1 @@
53972413

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Filter.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Partitions.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Rows.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Statistics.db Normal file
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-TOC.txt Normal file
View File

@ -0,0 +1,8 @@
Data.db
Statistics.db
Digest.crc32
TOC.txt
CompressionInfo.db
Filter.db
Partitions.db
Rows.db

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-CompressionInfo.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Data.db Normal file
View File

Binary file not shown.

1
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Digest.crc32 Normal file
View File

@ -0,0 +1 @@
3597571582

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Filter.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Partitions.db Normal file
View File

Binary file not shown.

0
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Rows.db Normal file
View File

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Statistics.db Normal file
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-TOC.txt Normal file
View File

@ -0,0 +1,8 @@
Data.db
Statistics.db
Digest.crc32
TOC.txt
CompressionInfo.db
Filter.db
Partitions.db
Rows.db

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-CompressionInfo.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Data.db Normal file
View File

Binary file not shown.

1
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Digest.crc32 Normal file
View File

@ -0,0 +1 @@
3514184066

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Filter.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Partitions.db Normal file
View File

Binary file not shown.

0
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Rows.db Normal file
View File

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Statistics.db Normal file
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-TOC.txt Normal file
View File

@ -0,0 +1,8 @@
Data.db
Statistics.db
Digest.crc32
TOC.txt
CompressionInfo.db
Filter.db
Partitions.db
Rows.db

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-CompressionInfo.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Data.db Normal file
View File

Binary file not shown.

1
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Digest.crc32 Normal file
View File

@ -0,0 +1 @@
1960712659

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Filter.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Partitions.db Normal file
View File

Binary file not shown.

0
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Rows.db Normal file
View File

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Statistics.db Normal file
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-TOC.txt Normal file
View File

@ -0,0 +1,8 @@
Data.db
Statistics.db
Digest.crc32
TOC.txt
CompressionInfo.db
Filter.db
Partitions.db
Rows.db

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-CompressionInfo.db Normal file
View File

Binary file not shown.

Some files were not shown because too many files have changed in this diff Show More