Merge branch 'cassandra-3.0' into trunk

Conflicts:
	src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java
	src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java
	src/java/org/apache/cassandra/service/pager/QueryPager.java
This commit is contained in:
Benedict Elliott Smith 2015-10-26 21:08:07 +00:00
commit 928e4c28c6
67 changed files with 2237 additions and 1712 deletions

View File

@ -5,6 +5,7 @@
3.0
* Flatten Iterator Transformation Hierarchy (CASSANDRA-9975)
* Remove token generator (CASSANDRA-5261)
* RolesCache should not be created for any authenticator that does not requireAuthentication (CASSANDRA-10562)
* Fix LogTransaction checking only a single directory for files (CASSANDRA-10421)

View File

@ -0,0 +1,214 @@
/*
* 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.NoSuchElementException;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.partitions.BasePartitionIterator;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.*;
public class EmptyIterators
{
private static class EmptyBasePartitionIterator<R extends BaseRowIterator<?>> implements BasePartitionIterator<R>
{
EmptyBasePartitionIterator()
{
}
public void close()
{
}
public boolean hasNext()
{
return false;
}
public R next()
{
throw new NoSuchElementException();
}
}
private static class EmptyUnfilteredPartitionIterator extends EmptyBasePartitionIterator<UnfilteredRowIterator> implements UnfilteredPartitionIterator
{
final CFMetaData metadata;
final boolean isForThrift;
public EmptyUnfilteredPartitionIterator(CFMetaData metadata, boolean isForThrift)
{
this.metadata = metadata;
this.isForThrift = isForThrift;
}
public boolean isForThrift()
{
return isForThrift;
}
public CFMetaData metadata()
{
return metadata;
}
}
private static class EmptyPartitionIterator extends EmptyBasePartitionIterator<RowIterator> implements PartitionIterator
{
public static final EmptyPartitionIterator instance = new EmptyPartitionIterator();
private EmptyPartitionIterator()
{
super();
}
}
private static class EmptyBaseRowIterator<U extends Unfiltered> implements BaseRowIterator<U>
{
final PartitionColumns columns;
final CFMetaData metadata;
final DecoratedKey partitionKey;
final boolean isReverseOrder;
final Row staticRow;
EmptyBaseRowIterator(PartitionColumns columns, CFMetaData metadata, DecoratedKey partitionKey, boolean isReverseOrder, Row staticRow)
{
this.columns = columns;
this.metadata = metadata;
this.partitionKey = partitionKey;
this.isReverseOrder = isReverseOrder;
this.staticRow = staticRow;
}
public CFMetaData metadata()
{
return metadata;
}
public boolean isReverseOrder()
{
return isReverseOrder;
}
public PartitionColumns columns()
{
return columns;
}
public DecoratedKey partitionKey()
{
return partitionKey;
}
public Row staticRow()
{
return staticRow;
}
public void close()
{
}
public boolean isEmpty()
{
return staticRow == Rows.EMPTY_STATIC_ROW;
}
public boolean hasNext()
{
return false;
}
public U next()
{
throw new NoSuchElementException();
}
}
private static class EmptyUnfilteredRowIterator extends EmptyBaseRowIterator<Unfiltered> implements UnfilteredRowIterator
{
final DeletionTime partitionLevelDeletion;
public EmptyUnfilteredRowIterator(PartitionColumns columns, CFMetaData metadata, DecoratedKey partitionKey,
boolean isReverseOrder, Row staticRow, DeletionTime partitionLevelDeletion)
{
super(columns, metadata, partitionKey, isReverseOrder, staticRow);
this.partitionLevelDeletion = partitionLevelDeletion;
}
public boolean isEmpty()
{
return partitionLevelDeletion == DeletionTime.LIVE && super.isEmpty();
}
public DeletionTime partitionLevelDeletion()
{
return partitionLevelDeletion;
}
public EncodingStats stats()
{
return EncodingStats.NO_STATS;
}
}
private static class EmptyRowIterator extends EmptyBaseRowIterator<Row> implements RowIterator
{
public EmptyRowIterator(CFMetaData metadata, DecoratedKey partitionKey, boolean isReverseOrder, Row staticRow)
{
super(PartitionColumns.NONE, metadata, partitionKey, isReverseOrder, staticRow);
}
}
public static UnfilteredPartitionIterator unfilteredPartition(CFMetaData metadata, boolean isForThrift)
{
return new EmptyUnfilteredPartitionIterator(metadata, isForThrift);
}
public static PartitionIterator partition()
{
return EmptyPartitionIterator.instance;
}
// this method is the only one that can return a non-empty iterator, but it still has no rows, so it seems cleanest to keep it here
public static UnfilteredRowIterator unfilteredRow(CFMetaData metadata, DecoratedKey partitionKey, boolean isReverseOrder, Row staticRow, DeletionTime partitionDeletion)
{
PartitionColumns columns = PartitionColumns.NONE;
if (!staticRow.isEmpty())
columns = new PartitionColumns(Columns.from(staticRow.columns()), Columns.NONE);
else
staticRow = Rows.EMPTY_STATIC_ROW;
if (partitionDeletion.isLive())
partitionDeletion = DeletionTime.LIVE;
return new EmptyUnfilteredRowIterator(columns, metadata, partitionKey, isReverseOrder, staticRow, partitionDeletion);
}
public static UnfilteredRowIterator unfilteredRow(CFMetaData metadata, DecoratedKey partitionKey, boolean isReverseOrder)
{
return new EmptyUnfilteredRowIterator(PartitionColumns.NONE, metadata, partitionKey, isReverseOrder, Rows.EMPTY_STATIC_ROW, DeletionTime.LIVE);
}
public static RowIterator row(CFMetaData metadata, DecoratedKey partitionKey, boolean isReverseOrder)
{
return new EmptyRowIterator(metadata, partitionKey, isReverseOrder, Rows.EMPTY_STATIC_ROW);
}
}

View File

@ -30,7 +30,8 @@ import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.BaseRowIterator;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.index.Index;
@ -226,10 +227,10 @@ public class PartitionRangeReadCommand extends ReadCommand
private UnfilteredPartitionIterator checkCacheFilter(UnfilteredPartitionIterator iter, final ColumnFamilyStore cfs)
{
return new WrappingUnfilteredPartitionIterator(iter)
class CacheFilter extends Transformation
{
@Override
public UnfilteredRowIterator computeNext(UnfilteredRowIterator iter)
public BaseRowIterator applyToPartition(BaseRowIterator iter)
{
// Note that we rely on the fact that until we actually advance 'iter', no really costly operation is actually done
// (except for reading the partition key from the index file) due to the call to mergeLazily in queryStorage.
@ -249,7 +250,8 @@ public class PartitionRangeReadCommand extends ReadCommand
return iter;
}
};
}
return Transformation.apply(iter, new CacheFilter());
}
public MessageOut<ReadCommand> createMessage(int version)

View File

@ -31,6 +31,8 @@ import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.monitoring.MonitorableImpl;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.transform.StoppingTransformation;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.io.IVersionedSerializer;
@ -386,7 +388,7 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery
*/
private UnfilteredPartitionIterator withMetricsRecording(UnfilteredPartitionIterator iter, final TableMetrics metric, final long startTimeNanos)
{
return new WrappingUnfilteredPartitionIterator(iter)
class MetricRecording extends Transformation<UnfilteredRowIterator>
{
private final int failureThreshold = DatabaseDescriptor.getTombstoneFailureThreshold();
private final int warningThreshold = DatabaseDescriptor.getTombstoneWarnThreshold();
@ -399,114 +401,105 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery
private DecoratedKey currentKey;
@Override
public UnfilteredRowIterator computeNext(UnfilteredRowIterator iter)
public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator iter)
{
currentKey = iter.partitionKey();
return new AlteringUnfilteredRowIterator(iter)
{
@Override
protected Row computeNextStatic(Row row)
{
return computeNext(row);
}
@Override
protected Row computeNext(Row row)
{
if (row.hasLiveData(ReadCommand.this.nowInSec()))
++liveRows;
for (Cell cell : row.cells())
{
if (!cell.isLive(ReadCommand.this.nowInSec()))
countTombstone(row.clustering());
}
return row;
}
@Override
protected RangeTombstoneMarker computeNext(RangeTombstoneMarker marker)
{
countTombstone(marker.clustering());
return marker;
}
private void countTombstone(ClusteringPrefix clustering)
{
++tombstones;
if (tombstones > failureThreshold && respectTombstoneThresholds)
{
String query = ReadCommand.this.toCQLString();
Tracing.trace("Scanned over {} tombstones for query {}; query aborted (see tombstone_failure_threshold)", failureThreshold, query);
throw new TombstoneOverwhelmingException(tombstones, query, ReadCommand.this.metadata(), currentKey, clustering);
}
}
};
return Transformation.apply(iter, this);
}
@Override
public void close()
public Row applyToStatic(Row row)
{
try
return applyToRow(row);
}
@Override
public Row applyToRow(Row row)
{
if (row.hasLiveData(ReadCommand.this.nowInSec()))
++liveRows;
for (Cell cell : row.cells())
{
super.close();
if (!cell.isLive(ReadCommand.this.nowInSec()))
countTombstone(row.clustering());
}
finally
return row;
}
@Override
public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
{
countTombstone(marker.clustering());
return marker;
}
private void countTombstone(ClusteringPrefix clustering)
{
++tombstones;
if (tombstones > failureThreshold && respectTombstoneThresholds)
{
recordLatency(metric, System.nanoTime() - startTimeNanos);
metric.tombstoneScannedHistogram.update(tombstones);
metric.liveScannedHistogram.update(liveRows);
boolean warnTombstones = tombstones > warningThreshold && respectTombstoneThresholds;
if (warnTombstones)
{
String msg = String.format("Read %d live rows and %d tombstone cells for query %1.512s (see tombstone_warn_threshold)", liveRows, tombstones, ReadCommand.this.toCQLString());
ClientWarn.warn(msg);
logger.warn(msg);
}
Tracing.trace("Read {} live and {} tombstone cells{}", liveRows, tombstones, (warnTombstones ? " (see tombstone_warn_threshold)" : ""));
String query = ReadCommand.this.toCQLString();
Tracing.trace("Scanned over {} tombstones for query {}; query aborted (see tombstone_failure_threshold)", failureThreshold, query);
throw new TombstoneOverwhelmingException(tombstones, query, ReadCommand.this.metadata(), currentKey, clustering);
}
}
@Override
public void onClose()
{
recordLatency(metric, System.nanoTime() - startTimeNanos);
metric.tombstoneScannedHistogram.update(tombstones);
metric.liveScannedHistogram.update(liveRows);
boolean warnTombstones = tombstones > warningThreshold && respectTombstoneThresholds;
if (warnTombstones)
{
String msg = String.format("Read %d live rows and %d tombstone cells for query %1.512s (see tombstone_warn_threshold)", liveRows, tombstones, ReadCommand.this.toCQLString());
ClientWarn.warn(msg);
logger.warn(msg);
}
Tracing.trace("Read {} live and {} tombstone cells{}", liveRows, tombstones, (warnTombstones ? " (see tombstone_warn_threshold)" : ""));
}
};
return Transformation.apply(iter, new MetricRecording());
}
protected class CheckForAbort extends StoppingTransformation<BaseRowIterator<?>>
{
protected BaseRowIterator<?> applyToPartition(BaseRowIterator partition)
{
maybeAbort();
return partition;
}
protected Row applyToRow(Row row)
{
maybeAbort();
return row;
}
private void maybeAbort()
{
if (isAborted())
stop();
if (TEST_ITERATION_DELAY_MILLIS > 0)
maybeDelayForTesting();
}
}
protected UnfilteredPartitionIterator withStateTracking(UnfilteredPartitionIterator iter)
{
return new WrappingUnfilteredPartitionIterator(iter)
{
@Override
public UnfilteredRowIterator computeNext(UnfilteredRowIterator iter)
{
if (isAborted())
return null;
if (TEST_ITERATION_DELAY_MILLIS > 0)
maybeDelayForTesting();
return iter;
}
};
return Transformation.apply(iter, new CheckForAbort());
}
protected UnfilteredRowIterator withStateTracking(UnfilteredRowIterator iter)
{
return new WrappingUnfilteredRowIterator(iter)
{
@Override
public boolean hasNext()
{
if (isAborted())
return false;
if (TEST_ITERATION_DELAY_MILLIS > 0)
maybeDelayForTesting();
return super.hasNext();
}
};
return Transformation.apply(iter, new CheckForAbort());
}
private void maybeDelayForTesting()
@ -527,13 +520,20 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery
// are to some extend an artefact of compaction lagging behind and hence counting them is somewhat unintuitive).
protected UnfilteredPartitionIterator withoutPurgeableTombstones(UnfilteredPartitionIterator iterator, ColumnFamilyStore cfs)
{
return new PurgingPartitionIterator(iterator, cfs.gcBefore(nowInSec()), oldestUnrepairedTombstone(), cfs.getCompactionStrategyManager().onlyPurgeRepairedTombstones())
final boolean isForThrift = iterator.isForThrift();
class WithoutPurgeableTombstones extends PurgeFunction
{
public WithoutPurgeableTombstones()
{
super(isForThrift, cfs.gcBefore(nowInSec()), oldestUnrepairedTombstone(), cfs.getCompactionStrategyManager().onlyPurgeRepairedTombstones());
}
protected long getMaxPurgeableTimestamp()
{
return Long.MAX_VALUE;
}
};
}
return Transformation.apply(iterator, new WithoutPurgeableTombstones());
}
/**

View File

@ -42,12 +42,12 @@ public interface ReadQuery
public PartitionIterator execute(ConsistencyLevel consistency, ClientState clientState) throws RequestExecutionException
{
return PartitionIterators.EMPTY;
return EmptyIterators.partition();
}
public PartitionIterator executeInternal(ReadExecutionController controller)
{
return PartitionIterators.EMPTY;
return EmptyIterators.partition();
}
public DataLimits limits()

View File

@ -153,7 +153,7 @@ public class SinglePartitionNamesCommand extends SinglePartitionReadCommand<Clus
cfs.metric.updateSSTableIterated(sstablesIterated);
if (result == null || result.isEmpty())
return UnfilteredRowIterators.emptyIterator(metadata(), partitionKey(), false);
return EmptyIterators.unfilteredRow(metadata(), partitionKey(), false);
DecoratedKey key = result.partitionKey();
cfs.metric.samplers.get(TableMetrics.Sampler.READS).addSample(key.getKey(), key.hashCode(), 1);

View File

@ -233,7 +233,7 @@ public class SinglePartitionSliceCommand extends SinglePartitionReadCommand<Clus
cfs.metric.updateSSTableIterated(sstablesIterated);
if (iterators.isEmpty())
return UnfilteredRowIterators.emptyIterator(cfs.metadata, partitionKey(), filter.isReversed());
return EmptyIterators.unfilteredRow(cfs.metadata, partitionKey(), filter.isReversed());
Tracing.trace("Merging data from memtables and {} sstables", sstablesIterated);

View File

@ -25,10 +25,11 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.partitions.PurgingPartitionIterator;
import org.apache.cassandra.db.partitions.PurgeFunction;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.index.transactions.CompactionTransaction;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.metrics.CompactionMetrics;
@ -98,9 +99,11 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
if (metrics != null)
metrics.beginCompaction(this);
this.compacted = scanners.isEmpty()
? UnfilteredPartitionIterators.empty(controller.cfs.metadata)
: new PurgeIterator(UnfilteredPartitionIterators.merge(scanners, nowInSec, listener()), controller);
UnfilteredPartitionIterator merged = scanners.isEmpty()
? EmptyIterators.unfilteredPartition(controller.cfs.metadata, false)
: UnfilteredPartitionIterators.merge(scanners, nowInSec, listener());
boolean isForThrift = merged.isForThrift(); // to stop capture of iterator in Purger, which is confusing for debug
this.compacted = Transformation.apply(merged, new Purger(isForThrift, controller));
}
public boolean isForThrift()
@ -251,7 +254,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
return this.getCompactionInfo().toString();
}
private class PurgeIterator extends PurgingPartitionIterator
private class Purger extends PurgeFunction
{
private final CompactionController controller;
@ -261,9 +264,9 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
private long compactedUnfiltered;
private PurgeIterator(UnfilteredPartitionIterator toPurge, CompactionController controller)
private Purger(boolean isForThrift, CompactionController controller)
{
super(toPurge, controller.gcBefore, controller.compactingRepaired() ? Integer.MIN_VALUE : Integer.MAX_VALUE, controller.cfs.getCompactionStrategyManager().onlyPurgeRepairedTombstones());
super(isForThrift, controller.gcBefore, controller.compactingRepaired() ? Integer.MIN_VALUE : Integer.MAX_VALUE, controller.cfs.getCompactionStrategyManager().onlyPurgeRepairedTombstones());
this.controller = controller;
}

View File

@ -25,6 +25,7 @@ import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -107,20 +108,21 @@ public class ClusteringIndexNamesFilter extends AbstractClusteringIndexFilter
{
// Note that we don't filter markers because that's a bit trickier (we don't know in advance until when
// the range extend) and it's harmless to left them.
return new AlteringUnfilteredRowIterator(iterator)
class FilterNotIndexed extends Transformation
{
@Override
public Row computeNextStatic(Row row)
public Row applyToStatic(Row row)
{
return columnFilter.fetchedColumns().statics.isEmpty() ? null : row.filter(columnFilter, iterator.metadata());
}
@Override
public Row computeNext(Row row)
public Row applyToRow(Row row)
{
return clusterings.contains(row.clustering()) ? row.filter(columnFilter, iterator.metadata()) : null;
}
};
}
return Transformation.apply(iterator, new FilterNotIndexed());
}
public UnfilteredRowIterator filter(final SliceableUnfilteredRowIterator iter)

View File

@ -26,6 +26,7 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.partitions.CachedPartition;
import org.apache.cassandra.db.partitions.Partition;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -91,26 +92,26 @@ public class ClusteringIndexSliceFilter extends AbstractClusteringIndexFilter
// Note that we don't filter markers because that's a bit trickier (we don't know in advance until when
// the range extend) and it's harmless to leave them.
return new AlteringUnfilteredRowIterator(iterator)
class FilterNotIndexed extends Transformation
{
@Override
public boolean hasNext()
public boolean isDoneForPartition()
{
return !tester.isDone() && super.hasNext();
return tester.isDone();
}
@Override
public Row computeNextStatic(Row row)
{
return columnFilter.fetchedColumns().statics.isEmpty() ? null : row.filter(columnFilter, iterator.metadata());
}
@Override
public Row computeNext(Row row)
public Row applyToRow(Row row)
{
return tester.includes(row.clustering()) ? row.filter(columnFilter, iterator.metadata()) : null;
}
};
@Override
public Row applyToStatic(Row row)
{
return columnFilter.fetchedColumns().statics.isEmpty() ? Rows.EMPTY_STATIC_ROW : row.filter(columnFilter, iterator.metadata());
}
}
return Transformation.apply(iterator, new FilterNotIndexed());
}
public UnfilteredRowIterator filter(SliceableUnfilteredRowIterator iterator)

View File

@ -23,6 +23,10 @@ import java.nio.ByteBuffer;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.transform.BasePartitions;
import org.apache.cassandra.db.transform.BaseRows;
import org.apache.cassandra.db.transform.StoppingTransformation;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -125,17 +129,17 @@ public abstract class DataLimits
public UnfilteredPartitionIterator filter(UnfilteredPartitionIterator iter, int nowInSec)
{
return new CountingUnfilteredPartitionIterator(iter, newCounter(nowInSec, false));
return this.newCounter(nowInSec, false).applyTo(iter);
}
public UnfilteredRowIterator filter(UnfilteredRowIterator iter, int nowInSec)
{
return new CountingUnfilteredRowIterator(iter, newCounter(nowInSec, false));
return this.newCounter(nowInSec, false).applyTo(iter);
}
public PartitionIterator filter(PartitionIterator iter, int nowInSec)
{
return new CountingPartitionIterator(iter, this, nowInSec);
return this.newCounter(nowInSec, true).applyTo(iter);
}
/**
@ -144,11 +148,36 @@ public abstract class DataLimits
*/
public abstract float estimateTotalResults(ColumnFamilyStore cfs);
public interface Counter
public static abstract class Counter extends StoppingTransformation<BaseRowIterator<?>>
{
public void newPartition(DecoratedKey partitionKey, Row staticRow);
public void newRow(Row row);
public void endOfPartition();
// false means we do not propagate our stop signals onto the iterator, we only count
private boolean enforceLimits = true;
public Counter onlyCount()
{
this.enforceLimits = false;
return this;
}
public PartitionIterator applyTo(PartitionIterator partitions)
{
return Transformation.apply(partitions, this);
}
public UnfilteredPartitionIterator applyTo(UnfilteredPartitionIterator partitions)
{
return Transformation.apply(partitions, this);
}
public UnfilteredRowIterator applyTo(UnfilteredRowIterator partition)
{
return (UnfilteredRowIterator) applyToPartition(partition);
}
public RowIterator applyTo(RowIterator partition)
{
return (RowIterator) applyToPartition(partition);
}
/**
* The number of results counted.
@ -157,12 +186,40 @@ public abstract class DataLimits
*
* @return the number of results counted.
*/
public int counted();
public abstract int counted();
public abstract int countedInCurrentPartition();
public int countedInCurrentPartition();
public abstract boolean isDone();
public abstract boolean isDoneForPartition();
public boolean isDone();
public boolean isDoneForPartition();
@Override
protected BaseRowIterator<?> applyToPartition(BaseRowIterator<?> partition)
{
return partition instanceof UnfilteredRowIterator ? Transformation.apply((UnfilteredRowIterator) partition, this)
: Transformation.apply((RowIterator) partition, this);
}
// called before we process a given partition
protected abstract void applyToPartition(DecoratedKey partitionKey, Row staticRow);
@Override
protected void attachTo(BasePartitions partitions)
{
if (enforceLimits)
super.attachTo(partitions);
if (isDone())
stop();
}
@Override
protected void attachTo(BaseRows rows)
{
if (enforceLimits)
super.attachTo(rows);
applyToPartition(rows.partitionKey(), rows.staticRow());
if (isDoneForPartition())
stopInPartition();
}
}
/**
@ -241,13 +298,15 @@ public abstract class DataLimits
return false;
// Otherwise, we need to re-count
DataLimits.Counter counter = newCounter(nowInSec, false);
try (UnfilteredRowIterator cacheIter = cached.unfilteredIterator(ColumnFilter.selection(cached.columns()), Slices.ALL, false);
CountingUnfilteredRowIterator iter = new CountingUnfilteredRowIterator(cacheIter, newCounter(nowInSec, false)))
UnfilteredRowIterator iter = counter.applyTo(cacheIter))
{
// Consume the iterator until we've counted enough
while (iter.hasNext() && !iter.counter().isDone())
while (iter.hasNext())
iter.next();
return iter.counter().isDone();
return counter.isDone();
}
}
@ -274,7 +333,7 @@ public abstract class DataLimits
return rowsPerPartition * (cfs.estimateKeys());
}
protected class CQLCounter implements Counter
protected class CQLCounter extends Counter
{
protected final int nowInSec;
protected final boolean assumeLiveData;
@ -290,23 +349,39 @@ public abstract class DataLimits
this.assumeLiveData = assumeLiveData;
}
public void newPartition(DecoratedKey partitionKey, Row staticRow)
@Override
public void applyToPartition(DecoratedKey partitionKey, Row staticRow)
{
rowInCurrentPartition = 0;
if (!staticRow.isEmpty() && (assumeLiveData || staticRow.hasLiveData(nowInSec)))
hasLiveStaticRow = true;
}
public void endOfPartition()
@Override
public Row applyToRow(Row row)
{
if (assumeLiveData || row.hasLiveData(nowInSec))
incrementRowCount();
return row;
}
@Override
public void onPartitionClose()
{
// Normally, we don't count static rows as from a CQL point of view, it will be merge with other
// rows in the partition. However, if we only have the static row, it will be returned as one row
// so count it.
if (hasLiveStaticRow && rowInCurrentPartition == 0)
{
++rowCounted;
++rowInCurrentPartition;
}
incrementRowCount();
super.onPartitionClose();
}
private void incrementRowCount()
{
if (++rowCounted >= rowLimit)
stop();
if (++rowInCurrentPartition >= perPartitionLimit)
stopInPartition();
}
public int counted()
@ -328,15 +403,6 @@ public abstract class DataLimits
{
return isDone() || rowInCurrentPartition >= perPartitionLimit;
}
public void newRow(Row row)
{
if (assumeLiveData || row.hasLiveData(nowInSec))
{
++rowCounted;
++rowInCurrentPartition;
}
}
}
@Override
@ -402,7 +468,7 @@ public abstract class DataLimits
}
@Override
public void newPartition(DecoratedKey partitionKey, Row staticRow)
public void applyToPartition(DecoratedKey partitionKey, Row staticRow)
{
if (partitionKey.getKey().equals(lastReturnedKey))
{
@ -415,7 +481,7 @@ public abstract class DataLimits
}
else
{
super.newPartition(partitionKey, staticRow);
super.applyToPartition(partitionKey, staticRow);
}
}
}
@ -481,13 +547,14 @@ public abstract class DataLimits
return false;
// Otherwise, we need to re-count
DataLimits.Counter counter = newCounter(nowInSec, false);
try (UnfilteredRowIterator cacheIter = cached.unfilteredIterator(ColumnFilter.selection(cached.columns()), Slices.ALL, false);
CountingUnfilteredRowIterator iter = new CountingUnfilteredRowIterator(cacheIter, newCounter(nowInSec, false)))
UnfilteredRowIterator iter = counter.applyTo(cacheIter))
{
// Consume the iterator until we've counted enough
while (iter.hasNext() && !iter.counter().isDone())
while (iter.hasNext())
iter.next();
return iter.counter().isDone();
return counter.isDone();
}
}
@ -513,7 +580,7 @@ public abstract class DataLimits
return cellsPerPartition * cfs.estimateKeys();
}
protected class ThriftCounter implements Counter
protected class ThriftCounter extends Counter
{
protected final int nowInSec;
protected final boolean assumeLiveData;
@ -528,16 +595,35 @@ public abstract class DataLimits
this.assumeLiveData = assumeLiveData;
}
public void newPartition(DecoratedKey partitionKey, Row staticRow)
@Override
public void applyToPartition(DecoratedKey partitionKey, Row staticRow)
{
cellsInCurrentPartition = 0;
if (!staticRow.isEmpty())
newRow(staticRow);
applyToRow(staticRow);
}
public void endOfPartition()
@Override
public Row applyToRow(Row row)
{
++partitionsCounted;
for (Cell cell : row.cells())
{
if (assumeLiveData || cell.isLive(nowInSec))
{
++cellsCounted;
if (++cellsInCurrentPartition >= cellPerPartitionLimit)
stopInPartition();
}
}
return row;
}
@Override
public void onPartitionClose()
{
if (++partitionsCounted >= partitionLimit)
stop();
super.onPartitionClose();
}
public int counted()
@ -559,18 +645,6 @@ public abstract class DataLimits
{
return isDone() || cellsInCurrentPartition >= cellPerPartitionLimit;
}
public void newRow(Row row)
{
for (Cell cell : row.cells())
{
if (assumeLiveData || cell.isLive(nowInSec))
{
++cellsCounted;
++cellsInCurrentPartition;
}
}
}
}
@Override
@ -625,14 +699,17 @@ public abstract class DataLimits
super(nowInSec, assumeLiveData);
}
public void newRow(Row row)
@Override
public Row applyToRow(Row row)
{
// In the internal format, a row == a super column, so that's what we want to count.
if (assumeLiveData || row.hasLiveData(nowInSec))
{
++cellsCounted;
++cellsInCurrentPartition;
if (++cellsInCurrentPartition >= cellPerPartitionLimit)
stopInPartition();
}
return row;
}
}
}

View File

@ -22,7 +22,6 @@ import java.nio.ByteBuffer;
import java.util.*;
import com.google.common.base.Objects;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
@ -31,6 +30,7 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -222,28 +222,29 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
if (expressions.isEmpty())
return iter;
return new AlteringUnfilteredPartitionIterator(iter)
class IsSatisfiedFilter extends Transformation<UnfilteredRowIterator>
{
protected Row computeNext(DecoratedKey partitionKey, Row row)
DecoratedKey pk;
public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
// We filter tombstones when passing the row to isSatisfiedBy so that the method doesn't have to bother with them.
Row purged = row.purge(DeletionPurger.PURGE_ALL, nowInSec);
return purged != null && CQLFilter.this.isSatisfiedBy(partitionKey, purged) ? row : null;
pk = partition.partitionKey();
return Transformation.apply(partition, this);
}
};
}
/**
* Returns whether the provided row (with it's partition key) satisfies
* this row filter or not (that is, if it satisfies all of its expressions).
*/
private boolean isSatisfiedBy(DecoratedKey partitionKey, Row row)
{
for (Expression e : expressions)
if (!e.isSatisfiedBy(partitionKey, row))
return false;
public Row applyToRow(Row row)
{
Row purged = row.purge(DeletionPurger.PURGE_ALL, nowInSec);
if (purged == null)
return null;
return true;
for (Expression e : expressions)
if (!e.isSatisfiedBy(pk, purged))
return null;
return row;
}
}
return Transformation.apply(iter, new IsSatisfiedFilter());
}
protected RowFilter withNewExpressions(List<Expression> expressions)
@ -264,16 +265,17 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
if (expressions.isEmpty())
return iter;
return new WrappingUnfilteredPartitionIterator(iter)
class IsSatisfiedThriftFilter extends Transformation<UnfilteredRowIterator>
{
@Override
public UnfilteredRowIterator computeNext(final UnfilteredRowIterator iter)
public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator iter)
{
// Thrift does not filter rows, it filters entire partition if any of the expression is not
// satisfied, which forces us to materialize the result (in theory we could materialize only
// what we need which might or might not be everything, but we keep it simple since in practice
// it's not worth that it has ever been).
ImmutableBTreePartition result = ImmutableBTreePartition.create(iter);
iter.close();
// The partition needs to have a row for every expression, and the expression needs to be valid.
for (Expression expr : expressions)
@ -286,7 +288,8 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
// If we get there, it means all expressions where satisfied, so return the original result
return result.unfilteredIterator();
}
};
}
return Transformation.apply(iter, new IsSatisfiedThriftFilter());
}
protected RowFilter withNewExpressions(List<Expression> expressions)

View File

@ -1,72 +0,0 @@
/*
* 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.partitions;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.rows.*;
/**
* A partition iterator that allows to filter/modify the unfiltered from the
* underlying iterators.
*/
public abstract class AlteringUnfilteredPartitionIterator extends WrappingUnfilteredPartitionIterator
{
protected AlteringUnfilteredPartitionIterator(UnfilteredPartitionIterator wrapped)
{
super(wrapped);
}
protected Row computeNextStatic(DecoratedKey partitionKey, Row row)
{
return row;
}
protected Row computeNext(DecoratedKey partitionKey, Row row)
{
return row;
}
protected RangeTombstoneMarker computeNext(DecoratedKey partitionKey, RangeTombstoneMarker marker)
{
return marker;
}
@Override
protected UnfilteredRowIterator computeNext(UnfilteredRowIterator iter)
{
final DecoratedKey partitionKey = iter.partitionKey();
return new AlteringUnfilteredRowIterator(iter)
{
protected Row computeNextStatic(Row row)
{
return AlteringUnfilteredPartitionIterator.this.computeNextStatic(partitionKey, row);
}
protected Row computeNext(Row row)
{
return AlteringUnfilteredPartitionIterator.this.computeNext(partitionKey, row);
}
protected RangeTombstoneMarker computeNext(RangeTombstoneMarker marker)
{
return AlteringUnfilteredPartitionIterator.this.computeNext(partitionKey, marker);
}
};
}
}

View File

@ -0,0 +1,27 @@
/*
* 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.partitions;
import org.apache.cassandra.db.rows.BaseRowIterator;
import org.apache.cassandra.utils.CloseableIterator;
public interface BasePartitionIterator<I extends BaseRowIterator<?>> extends CloseableIterator<I>
{
public void close();
}

View File

@ -1,58 +0,0 @@
/*
* 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.partitions;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.filter.DataLimits;
public class CountingPartitionIterator extends WrappingPartitionIterator
{
protected final DataLimits.Counter counter;
public CountingPartitionIterator(PartitionIterator result, DataLimits.Counter counter)
{
super(result);
this.counter = counter;
}
public CountingPartitionIterator(PartitionIterator result, DataLimits limits, int nowInSec)
{
this(result, limits.newCounter(nowInSec, true));
}
public DataLimits.Counter counter()
{
return counter;
}
@Override
public boolean hasNext()
{
if (counter.isDone())
return false;
return super.hasNext();
}
@Override
@SuppressWarnings("resource") // Close through the closing of the returned 'CountingRowIterator' (and CountingRowIterator shouldn't throw)
public RowIterator next()
{
return new CountingRowIterator(super.next(), counter);
}
}

View File

@ -1,58 +0,0 @@
/*
* 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.partitions;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.filter.DataLimits;
public class CountingRowIterator extends WrappingRowIterator
{
protected final DataLimits.Counter counter;
public CountingRowIterator(RowIterator iter, DataLimits.Counter counter)
{
super(iter);
this.counter = counter;
counter.newPartition(iter.partitionKey(), iter.staticRow());
}
@Override
public boolean hasNext()
{
if (counter.isDoneForPartition())
return false;
return super.hasNext();
}
@Override
public Row next()
{
Row row = super.next();
counter.newRow(row);
return row;
}
@Override
public void close()
{
super.close();
counter.endOfPartition();
}
}

View File

@ -1,52 +0,0 @@
/*
* 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.partitions;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.filter.DataLimits;
public class CountingUnfilteredPartitionIterator extends WrappingUnfilteredPartitionIterator
{
protected final DataLimits.Counter counter;
public CountingUnfilteredPartitionIterator(UnfilteredPartitionIterator result, DataLimits.Counter counter)
{
super(result);
this.counter = counter;
}
public DataLimits.Counter counter()
{
return counter;
}
@Override
public boolean hasNext()
{
if (counter.isDone())
return false;
return super.hasNext();
}
@Override
public UnfilteredRowIterator computeNext(UnfilteredRowIterator iter)
{
return new CountingUnfilteredRowIterator(iter, counter);
}
}

View File

@ -1,64 +0,0 @@
/*
* 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.partitions;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.filter.DataLimits;
public class CountingUnfilteredRowIterator extends WrappingUnfilteredRowIterator
{
private final DataLimits.Counter counter;
public CountingUnfilteredRowIterator(UnfilteredRowIterator iter, DataLimits.Counter counter)
{
super(iter);
this.counter = counter;
counter.newPartition(iter.partitionKey(), iter.staticRow());
}
public DataLimits.Counter counter()
{
return counter;
}
@Override
public boolean hasNext()
{
if (counter.isDoneForPartition())
return false;
return super.hasNext();
}
@Override
public Unfiltered next()
{
Unfiltered next = super.next();
if (next.isRow())
counter.newRow((Row)next);
return next;
}
@Override
public void close()
{
super.close();
counter.endOfPartition();
}
}

View File

@ -33,7 +33,6 @@ import org.apache.cassandra.db.rows.*;
* reference on the returned objects for longer than the iteration, it must
* make a copy of it explicitely.
*/
public interface PartitionIterator extends Iterator<RowIterator>, AutoCloseable
public interface PartitionIterator extends BasePartitionIterator<RowIterator>
{
public void close();
}

View File

@ -20,37 +20,18 @@ package org.apache.cassandra.db.partitions;
import java.util.*;
import java.security.MessageDigest;
import org.apache.cassandra.db.EmptyIterators;
import org.apache.cassandra.db.transform.MorePartitions;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.io.util.FileUtils;
public abstract class PartitionIterators
{
private PartitionIterators() {}
public static final PartitionIterator EMPTY = new PartitionIterator()
{
public boolean hasNext()
{
return false;
}
public RowIterator next()
{
throw new NoSuchElementException();
}
public void remove()
{
}
public void close()
{
}
};
@SuppressWarnings("resource") // The created resources are returned right away
public static RowIterator getOnlyElement(final PartitionIterator iter, SinglePartitionReadCommand<?> command)
{
@ -58,30 +39,24 @@ public abstract class PartitionIterators
// want a RowIterator out of this method, so we return an empty one.
RowIterator toReturn = iter.hasNext()
? iter.next()
: RowIterators.emptyIterator(command.metadata(),
command.partitionKey(),
command.clusteringIndexFilter().isReversed());
: EmptyIterators.row(command.metadata(),
command.partitionKey(),
command.clusteringIndexFilter().isReversed());
// Note that in general, we should wrap the result so that it's close method actually
// close the whole PartitionIterator.
return new WrappingRowIterator(toReturn)
class Close extends Transformation
{
public void close()
public void onPartitionClose()
{
try
{
super.close();
}
finally
{
// asserting this only now because it bothers UnfilteredPartitionIterators.Serializer (which might be used
// under the provided DataIter) if hasNext() is called before the previously returned iterator hasn't been fully consumed.
assert !iter.hasNext();
iter.close();
}
// asserting this only now because it bothers UnfilteredPartitionIterators.Serializer (which might be used
// under the provided DataIter) if hasNext() is called before the previously returned iterator hasn't been fully consumed.
boolean hadNext = iter.hasNext();
iter.close();
assert !hadNext;
}
};
}
return Transformation.apply(toReturn, new Close());
}
@SuppressWarnings("resource") // The created resources are returned right away
@ -90,39 +65,17 @@ public abstract class PartitionIterators
if (iterators.size() == 1)
return iterators.get(0);
return new PartitionIterator()
class Extend implements MorePartitions<PartitionIterator>
{
private int idx = 0;
public boolean hasNext()
int i = 1;
public PartitionIterator moreContents()
{
while (idx < iterators.size())
{
if (iterators.get(idx).hasNext())
return true;
++idx;
}
return false;
if (i >= iterators.size())
return null;
return iterators.get(i++);
}
public RowIterator next()
{
if (!hasNext())
throw new NoSuchElementException();
return iterators.get(idx).next();
}
public void remove()
{
throw new UnsupportedOperationException();
}
public void close()
{
FileUtils.closeQuietly(iterators);
}
};
}
return MorePartitions.extend(iterators.get(0), new Extend());
}
public static void digest(PartitionIterator iterator, MessageDigest digest)
@ -162,13 +115,14 @@ public abstract class PartitionIterators
@SuppressWarnings("resource") // The created resources are returned right away
public static PartitionIterator loggingIterator(PartitionIterator iterator, final String id)
{
return new WrappingPartitionIterator(iterator)
class Logger extends Transformation<RowIterator>
{
public RowIterator next()
public RowIterator applyToPartition(RowIterator partition)
{
return RowIterators.loggingIterator(super.next(), id);
return RowIterators.loggingIterator(partition, id);
}
};
}
return Transformation.apply(iterator, new Logger());
}
private static class SingletonPartitionIterator extends AbstractIterator<RowIterator> implements PartitionIterator

View File

@ -0,0 +1,120 @@
/*
* 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.partitions;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.transform.Transformation;
public abstract class PurgeFunction extends Transformation<UnfilteredRowIterator>
{
private final boolean isForThrift;
private final DeletionPurger purger;
private final int gcBefore;
private boolean isReverseOrder;
public PurgeFunction(boolean isForThrift, int gcBefore, int oldestUnrepairedTombstone, boolean onlyPurgeRepairedTombstones)
{
this.isForThrift = isForThrift;
this.gcBefore = gcBefore;
this.purger = (timestamp, localDeletionTime) ->
!(onlyPurgeRepairedTombstones && localDeletionTime >= oldestUnrepairedTombstone)
&& localDeletionTime < gcBefore
&& timestamp < getMaxPurgeableTimestamp();
}
protected abstract long getMaxPurgeableTimestamp();
// Called at the beginning of each new partition
protected void onNewPartition(DecoratedKey partitionKey)
{
}
// Called for each partition that had only purged infos and are empty post-purge.
protected void onEmptyPartitionPostPurge(DecoratedKey partitionKey)
{
}
// Called for every unfiltered. Meant for CompactionIterator to update progress
protected void updateProgress()
{
}
public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
onNewPartition(partition.partitionKey());
isReverseOrder = partition.isReverseOrder();
UnfilteredRowIterator purged = Transformation.apply(partition, this);
if (!isForThrift && purged.isEmpty())
{
onEmptyPartitionPostPurge(purged.partitionKey());
purged.close();
return null;
}
return purged;
}
public DeletionTime applyToDeletion(DeletionTime deletionTime)
{
return purger.shouldPurge(deletionTime) ? DeletionTime.LIVE : deletionTime;
}
public Row applyToStatic(Row row)
{
updateProgress();
return row.purge(purger, gcBefore);
}
public Row applyToRow(Row row)
{
updateProgress();
return row.purge(purger, gcBefore);
}
public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
{
updateProgress();
boolean reversed = isReverseOrder;
if (marker.isBoundary())
{
// We can only skip the whole marker if both deletion time are purgeable.
// If only one of them is, filterTombstoneMarker will deal with it.
RangeTombstoneBoundaryMarker boundary = (RangeTombstoneBoundaryMarker)marker;
boolean shouldPurgeClose = purger.shouldPurge(boundary.closeDeletionTime(reversed));
boolean shouldPurgeOpen = purger.shouldPurge(boundary.openDeletionTime(reversed));
if (shouldPurgeClose)
{
if (shouldPurgeOpen)
return null;
return boundary.createCorrespondingOpenMarker(reversed);
}
return shouldPurgeOpen
? boundary.createCorrespondingCloseMarker(reversed)
: marker;
}
else
{
return purger.shouldPurge(((RangeTombstoneBoundMarker)marker).deletionTime()) ? null : marker;
}
}
}

View File

@ -1,156 +0,0 @@
/*
* 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.partitions;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
public abstract class PurgingPartitionIterator extends WrappingUnfilteredPartitionIterator
{
private final DeletionPurger purger;
private final int gcBefore;
private UnfilteredRowIterator next;
public PurgingPartitionIterator(UnfilteredPartitionIterator iterator, int gcBefore, int oldestUnrepairedTombstone, boolean onlyPurgeRepairedTombstones)
{
super(iterator);
this.gcBefore = gcBefore;
this.purger = new DeletionPurger()
{
public boolean shouldPurge(long timestamp, int localDeletionTime)
{
if (onlyPurgeRepairedTombstones && localDeletionTime >= oldestUnrepairedTombstone)
return false;
return timestamp < getMaxPurgeableTimestamp() && localDeletionTime < gcBefore;
}
};
}
protected abstract long getMaxPurgeableTimestamp();
// Called at the beginning of each new partition
protected void onNewPartition(DecoratedKey partitionKey)
{
}
// Called for each partition that had only purged infos and are empty post-purge.
protected void onEmptyPartitionPostPurge(DecoratedKey partitionKey)
{
}
// Called for every unfiltered. Meant for CompactionIterator to update progress
protected void updateProgress()
{
}
@Override
@SuppressWarnings("resource") // 'purged' closes wrapped 'iterator'
public boolean hasNext()
{
while (next == null && super.hasNext())
{
UnfilteredRowIterator iterator = super.next();
onNewPartition(iterator.partitionKey());
UnfilteredRowIterator purged = purge(iterator);
if (isForThrift() || !purged.isEmpty())
{
next = purged;
return true;
}
onEmptyPartitionPostPurge(purged.partitionKey());
purged.close();
}
return next != null;
}
@Override
public UnfilteredRowIterator next()
{
UnfilteredRowIterator toReturn = next;
next = null;
updateProgress();
return toReturn;
}
private UnfilteredRowIterator purge(final UnfilteredRowIterator iter)
{
return new AlteringUnfilteredRowIterator(iter)
{
@Override
public DeletionTime partitionLevelDeletion()
{
DeletionTime dt = iter.partitionLevelDeletion();
return purger.shouldPurge(dt) ? DeletionTime.LIVE : dt;
}
@Override
public Row computeNextStatic(Row row)
{
return row.purge(purger, gcBefore);
}
@Override
public Row computeNext(Row row)
{
return row.purge(purger, gcBefore);
}
@Override
public RangeTombstoneMarker computeNext(RangeTombstoneMarker marker)
{
boolean reversed = isReverseOrder();
if (marker.isBoundary())
{
// We can only skip the whole marker if both deletion time are purgeable.
// If only one of them is, filterTombstoneMarker will deal with it.
RangeTombstoneBoundaryMarker boundary = (RangeTombstoneBoundaryMarker)marker;
boolean shouldPurgeClose = purger.shouldPurge(boundary.closeDeletionTime(reversed));
boolean shouldPurgeOpen = purger.shouldPurge(boundary.openDeletionTime(reversed));
if (shouldPurgeClose)
{
if (shouldPurgeOpen)
return null;
return boundary.createCorrespondingOpenMarker(reversed);
}
return shouldPurgeOpen
? boundary.createCorrespondingCloseMarker(reversed)
: marker;
}
else
{
return purger.shouldPurge(((RangeTombstoneBoundMarker)marker).deletionTime()) ? null : marker;
}
}
@Override
public Unfiltered next()
{
Unfiltered next = super.next();
updateProgress();
return next;
}
};
}
};

View File

@ -17,8 +17,6 @@
*/
package org.apache.cassandra.db.partitions;
import java.util.Iterator;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
@ -30,7 +28,7 @@ import org.apache.cassandra.db.rows.UnfilteredRowIterator;
* reference on the returned objects for longer than the iteration, it must
* make a copy of it explicitely.
*/
public interface UnfilteredPartitionIterator extends Iterator<UnfilteredRowIterator>, AutoCloseable
public interface UnfilteredPartitionIterator extends BasePartitionIterator<UnfilteredRowIterator>
{
/**
* Whether that partition iterator is for a thrift queries.
@ -44,6 +42,4 @@ public interface UnfilteredPartitionIterator extends Iterator<UnfilteredRowItera
public boolean isForThrift();
public CFMetaData metadata();
public void close();
}

View File

@ -22,19 +22,16 @@ import java.io.IOException;
import java.security.MessageDigest;
import java.util.*;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.Lists;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.transform.FilteredPartitions;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.MergeIterator;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.*;
/**
* Static methods to work with partition iterators.
@ -53,33 +50,6 @@ public abstract class UnfilteredPartitionIterators
public void close();
}
public static UnfilteredPartitionIterator empty(final CFMetaData metadata)
{
return new AbstractUnfilteredPartitionIterator()
{
public boolean isForThrift()
{
return false;
}
public CFMetaData metadata()
{
return metadata;
}
public boolean hasNext()
{
return false;
}
public UnfilteredRowIterator next()
{
throw new NoSuchElementException();
}
};
}
@SuppressWarnings("resource") // The created resources are returned right away
public static UnfilteredRowIterator getOnlyElement(final UnfilteredPartitionIterator iter, SinglePartitionReadCommand<?> command)
{
@ -87,30 +57,24 @@ public abstract class UnfilteredPartitionIterators
// want a RowIterator out of this method, so we return an empty one.
UnfilteredRowIterator toReturn = iter.hasNext()
? iter.next()
: UnfilteredRowIterators.emptyIterator(command.metadata(),
command.partitionKey(),
command.clusteringIndexFilter().isReversed());
: EmptyIterators.unfilteredRow(command.metadata(),
command.partitionKey(),
command.clusteringIndexFilter().isReversed());
// Note that in general, we should wrap the result so that it's close method actually
// close the whole UnfilteredPartitionIterator.
return new WrappingUnfilteredRowIterator(toReturn)
class Close extends Transformation
{
public void close()
public void onPartitionClose()
{
try
{
super.close();
}
finally
{
// asserting this only now because it bothers Serializer if hasNext() is called before
// the previously returned iterator hasn't been fully consumed.
assert !iter.hasNext();
iter.close();
}
// asserting this only now because it bothers Serializer if hasNext() is called before
// the previously returned iterator hasn't been fully consumed.
boolean hadNext = iter.hasNext();
iter.close();
assert !hadNext;
}
};
}
return Transformation.apply(toReturn, new Close());
}
public static PartitionIterator mergeAndFilter(List<UnfilteredPartitionIterator> iterators, int nowInSec, MergeListener listener)
@ -121,55 +85,7 @@ public abstract class UnfilteredPartitionIterators
public static PartitionIterator filter(final UnfilteredPartitionIterator iterator, final int nowInSec)
{
return new PartitionIterator()
{
private RowIterator next;
public boolean hasNext()
{
while (next == null && iterator.hasNext())
{
@SuppressWarnings("resource") // closed either directly if empty, or, if assigned to next, by either
// the caller of next() or close()
UnfilteredRowIterator rowIterator = iterator.next();
next = UnfilteredRowIterators.filter(rowIterator, nowInSec);
if (!iterator.isForThrift() && next.isEmpty())
{
rowIterator.close();
next = null;
}
}
return next != null;
}
public RowIterator next()
{
if (next == null && !hasNext())
throw new NoSuchElementException();
RowIterator toReturn = next;
next = null;
return toReturn;
}
public void remove()
{
throw new UnsupportedOperationException();
}
public void close()
{
try
{
iterator.close();
}
finally
{
if (next != null)
next.close();
}
}
};
return FilteredPartitions.filter(iterator, nowInSec);
}
public static UnfilteredPartitionIterator merge(final List<? extends UnfilteredPartitionIterator> iterators, final int nowInSec, final MergeListener listener)
@ -204,7 +120,7 @@ public abstract class UnfilteredPartitionIterators
// Replace nulls by empty iterators
for (int i = 0; i < toMerge.size(); i++)
if (toMerge.get(i) == null)
toMerge.set(i, UnfilteredRowIterators.emptyIterator(metadata, partitionKey, isReverseOrder));
toMerge.set(i, EmptyIterators.unfilteredRow(metadata, partitionKey, isReverseOrder));
return UnfilteredRowIterators.merge(toMerge, nowInSec, rowListener);
}
@ -353,13 +269,14 @@ public abstract class UnfilteredPartitionIterators
*/
public static UnfilteredPartitionIterator loggingIterator(UnfilteredPartitionIterator iterator, final String id, final boolean fullDetails)
{
return new WrappingUnfilteredPartitionIterator(iterator)
class Logging extends Transformation<UnfilteredRowIterator>
{
public UnfilteredRowIterator next()
public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
return UnfilteredRowIterators.loggingIterator(super.next(), id, fullDetails);
return UnfilteredRowIterators.loggingIterator(partition, id, fullDetails);
}
};
}
return Transformation.apply(iterator, new Logging());
}
/**

View File

@ -1,50 +0,0 @@
/*
* 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.partitions;
import org.apache.cassandra.db.rows.RowIterator;
public abstract class WrappingPartitionIterator implements PartitionIterator
{
protected final PartitionIterator wrapped;
protected WrappingPartitionIterator(PartitionIterator wrapped)
{
this.wrapped = wrapped;
}
public boolean hasNext()
{
return wrapped.hasNext();
}
public RowIterator next()
{
return wrapped.next();
}
public void remove()
{
wrapped.remove();
}
public void close()
{
wrapped.close();
}
}

View File

@ -1,126 +0,0 @@
/*
* 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.partitions;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
/**
* A utility class for writing partition iterators that filter/modify other
* partition iterators.
*
* This work a little bit like Guava's AbstractIterator in that you only need
* to implement the computeNext() method, though that method takes as argument
* the UnfilteredRowIterator to filter from the wrapped partition iterator.
*/
public abstract class WrappingUnfilteredPartitionIterator extends AbstractUnfilteredPartitionIterator
{
protected final UnfilteredPartitionIterator wrapped;
private UnfilteredRowIterator next;
protected WrappingUnfilteredPartitionIterator(UnfilteredPartitionIterator wrapped)
{
this.wrapped = wrapped;
}
public boolean isForThrift()
{
return wrapped.isForThrift();
}
public CFMetaData metadata()
{
return wrapped.metadata();
}
public boolean hasNext()
{
prepareNext();
return next != null;
}
public UnfilteredRowIterator next()
{
prepareNext();
assert next != null;
UnfilteredRowIterator toReturn = next;
next = null;
return toReturn;
}
private void prepareNext()
{
while (next == null && wrapped.hasNext())
{
@SuppressWarnings("resource") // Closed on exception, right away if empty or ignored by computeNext, or if assigned to 'next',
// either by the caller to next(), or in close().
UnfilteredRowIterator wrappedNext = wrapped.next();
try
{
UnfilteredRowIterator maybeNext = computeNext(wrappedNext);
// As the wrappd iterator shouldn't return an empty iterator, if computeNext
// gave us back it's input we save the isEmpty check.
if (maybeNext != null && (isForThrift() || maybeNext == wrappedNext || !maybeNext.isEmpty()))
{
next = maybeNext;
return;
}
else
{
wrappedNext.close();
}
}
catch (RuntimeException | Error e)
{
wrappedNext.close();
throw e;
}
}
}
/**
* Given the next UnfilteredRowIterator from the wrapped partition iterator, return
* the (potentially modified) UnfilteredRowIterator to return. Please note that the
* result will be skipped if it's either {@code null} of if it's empty.
*
* The default implementation return it's input unchanged to make it easier
* to write wrapping partition iterators that only change the close method.
*/
protected UnfilteredRowIterator computeNext(UnfilteredRowIterator iter)
{
return iter;
}
@Override
public void close()
{
try
{
wrapped.close();
}
finally
{
if (next != null)
next.close();
}
}
}

View File

@ -1,98 +0,0 @@
/*
* 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.rows;
import java.util.NoSuchElementException;
import com.google.common.collect.UnmodifiableIterator;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.*;
/**
* Class that makes it easier to write unfiltered iterators that filter or modify
* the returned unfiltered.
*
* The methods you want to override are {@code computeNextStatic} and the {@code computeNext} methods.
* All of these methods are allowed to return a {@code null} value with the meaning of ignoring
* the entry.
*/
public abstract class AlteringUnfilteredRowIterator extends WrappingUnfilteredRowIterator
{
private Row staticRow;
private Unfiltered next;
protected AlteringUnfilteredRowIterator(UnfilteredRowIterator wrapped)
{
super(wrapped);
}
protected Row computeNextStatic(Row row)
{
return row;
}
protected Row computeNext(Row row)
{
return row;
}
protected RangeTombstoneMarker computeNext(RangeTombstoneMarker marker)
{
return marker;
}
public Row staticRow()
{
if (staticRow == null)
{
Row row = computeNextStatic(super.staticRow());
staticRow = row == null ? Rows.EMPTY_STATIC_ROW : row;
}
return staticRow;
}
public boolean hasNext()
{
while (next == null && super.hasNext())
{
Unfiltered unfiltered = super.next();
if (unfiltered.isRow())
{
Row row = computeNext((Row)unfiltered);
if (row != null && !row.isEmpty())
next = row;
}
else
{
next = computeNext((RangeTombstoneMarker)unfiltered);
}
}
return next != null;
}
public Unfiltered next()
{
if (!hasNext())
throw new NoSuchElementException();
Unfiltered toReturn = next;
next = null;
return toReturn;
}
}

View File

@ -0,0 +1,64 @@
/*
* 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.rows;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.PartitionColumns;
import org.apache.cassandra.utils.CloseableIterator;
/**
* A common interface for Row and Unfiltered, that permits sharing of the (majority) common
* methods and functionality
*/
public interface BaseRowIterator<U extends Unfiltered> extends CloseableIterator<U>
{
/**
* The metadata for the table this iterator on.
*/
public CFMetaData metadata();
/**
* Whether or not the rows returned by this iterator are in reversed
* clustering order.
*/
public boolean isReverseOrder();
/**
* A subset of the columns for the (static and regular) rows returned by this iterator.
* Every row returned by this iterator must guarantee that it has only those columns.
*/
public PartitionColumns columns();
/**
* The partition key of the partition this in an iterator over.
*/
public DecoratedKey partitionKey();
/**
* The static part corresponding to this partition (this can be an empty
* row).
*/
public Row staticRow();
/**
* Returns whether the provided iterator has no data.
*/
public boolean isEmpty();
}

View File

@ -34,38 +34,8 @@ import org.apache.cassandra.db.*;
* reverse clustering order if isReverseOrder is true), and the Row objects returned
* by next() are only valid until the next call to hasNext() or next().
*/
public interface RowIterator extends Iterator<Row>, AutoCloseable
public interface RowIterator extends BaseRowIterator<Row>
{
/**
* The metadata for the table this iterator on.
*/
public CFMetaData metadata();
/**
* Whether or not the rows returned by this iterator are in reversed
* clustering order.
*/
public boolean isReverseOrder();
/**
* A subset of the columns for the (static and regular) rows returned by this iterator.
* Every row returned by this iterator must guarantee that it has only those columns.
*/
public PartitionColumns columns();
/**
* The partition key of the partition this in an iterator over.
*/
public DecoratedKey partitionKey();
/**
* The static part corresponding to this partition (this can be an empty
* row).
*/
public Row staticRow();
public void close();
/**
* Returns whether the provided iterator has no data.
*/

View File

@ -17,14 +17,13 @@
*/
package org.apache.cassandra.db.rows;
import java.util.*;
import java.security.MessageDigest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.utils.FBUtilities;
/**
@ -49,54 +48,6 @@ public abstract class RowIterators
iterator.next().digest(digest);
}
public static RowIterator emptyIterator(CFMetaData cfm, DecoratedKey partitionKey, boolean isReverseOrder)
{
return iterator(cfm, partitionKey, isReverseOrder, Collections.emptyIterator());
}
public static RowIterator iterator(CFMetaData cfm, DecoratedKey partitionKey, boolean isReverseOrder, Iterator<Row> iterator)
{
return new RowIterator()
{
public CFMetaData metadata()
{
return cfm;
}
public boolean isReverseOrder()
{
return isReverseOrder;
}
public PartitionColumns columns()
{
return PartitionColumns.NONE;
}
public DecoratedKey partitionKey()
{
return partitionKey;
}
public Row staticRow()
{
return Rows.EMPTY_STATIC_ROW;
}
public void close() { }
public boolean hasNext()
{
return iterator.hasNext();
}
public Row next()
{
return iterator.next();
}
};
}
/**
* Wraps the provided iterator so it logs the returned rows for debugging purposes.
* <p>
@ -113,24 +64,23 @@ public abstract class RowIterators
metadata.getKeyValidator().getString(iterator.partitionKey().getKey()),
iterator.isReverseOrder());
return new WrappingRowIterator(iterator)
class Log extends Transformation
{
@Override
public Row staticRow()
public Row applyToStatic(Row row)
{
Row row = super.staticRow();
if (!row.isEmpty())
logger.info("[{}] {}", id, row.toString(metadata()));
logger.info("[{}] {}", id, row.toString(metadata));
return row;
}
@Override
public Row next()
public Row applyToRow(Row row)
{
Row next = super.next();
logger.info("[{}] {}", id, next.toString(metadata()));
return next;
logger.info("[{}] {}", id, row.toString(metadata));
return row;
}
};
}
return Transformation.apply(iterator, new Log());
}
}

View File

@ -46,41 +46,13 @@ import org.apache.cassandra.db.*;
* the returned objects for longer than the iteration, it must make a copy of
* it explicitly.
*/
public interface UnfilteredRowIterator extends Iterator<Unfiltered>, AutoCloseable
public interface UnfilteredRowIterator extends BaseRowIterator<Unfiltered>
{
/**
* The metadata for the table this iterator on.
*/
public CFMetaData metadata();
/**
* A subset of the columns for the (static and regular) rows returned by this iterator.
* Every row returned by this iterator must guarantee that it has only those columns.
*/
public PartitionColumns columns();
/**
* Whether or not the atom returned by this iterator are in reversed
* clustering order.
*/
public boolean isReverseOrder();
/**
* The partition key of the partition this in an iterator over.
*/
public DecoratedKey partitionKey();
/**
* The partition level deletion for the partition this iterate over.
*/
public DeletionTime partitionLevelDeletion();
/**
* The static part corresponding to this partition (this can be an empty
* row).
*/
public Row staticRow();
/**
* Return "statistics" about what is returned by this iterator. Those are used for
* performance reasons (for delta-encoding for instance) and code should not
@ -88,8 +60,6 @@ public interface UnfilteredRowIterator extends Iterator<Unfiltered>, AutoCloseab
*/
public EncodingStats stats();
public void close();
/**
* Returns whether this iterator has no data (including no deletion data).
*/

View File

@ -202,7 +202,7 @@ public class UnfilteredRowIteratorSerializer
public UnfilteredRowIterator deserialize(DataInputPlus in, int version, CFMetaData metadata, SerializationHelper.Flag flag, Header header) throws IOException
{
if (header.isEmpty)
return UnfilteredRowIterators.emptyIterator(metadata, header.key, header.isReversed);
return EmptyIterators.unfilteredRow(metadata, header.key, header.isReversed);
final SerializationHelper helper = new SerializationHelper(metadata, version, flag);
final SerializationHeader sHeader = header.sHeader;

View File

@ -22,10 +22,12 @@ import java.security.MessageDigest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.AbstractIterator;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.transform.FilteredRows;
import org.apache.cassandra.db.transform.MoreRows;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
import org.apache.cassandra.serializers.MarshalException;
@ -63,8 +65,7 @@ public abstract class UnfilteredRowIterators
*/
public static RowIterator filter(UnfilteredRowIterator iter, int nowInSec)
{
return new FilteringIterator(iter, nowInSec);
return FilteredRows.filter(iter, nowInSec);
}
/**
@ -90,72 +91,12 @@ public abstract class UnfilteredRowIterators
return UnfilteredRowMergeIterator.create(iterators, nowInSec, mergeListener);
}
public static UnfilteredRowIterator emptyIterator(final CFMetaData cfm, final DecoratedKey partitionKey, final boolean isReverseOrder)
{
return noRowsIterator(cfm, partitionKey, Rows.EMPTY_STATIC_ROW, DeletionTime.LIVE, isReverseOrder);
}
/**
* Returns an empty atom iterator for a given partition.
*/
public static UnfilteredRowIterator noRowsIterator(final CFMetaData cfm, final DecoratedKey partitionKey, final Row staticRow, final DeletionTime partitionDeletion, final boolean isReverseOrder)
{
PartitionColumns columns = staticRow == Rows.EMPTY_STATIC_ROW ? PartitionColumns.NONE
: new PartitionColumns(Columns.from(staticRow.columns()), Columns.NONE);
return new UnfilteredRowIterator()
{
public CFMetaData metadata()
{
return cfm;
}
public boolean isReverseOrder()
{
return isReverseOrder;
}
public PartitionColumns columns()
{
return columns;
}
public DecoratedKey partitionKey()
{
return partitionKey;
}
public DeletionTime partitionLevelDeletion()
{
return partitionDeletion;
}
public Row staticRow()
{
return staticRow;
}
public EncodingStats stats()
{
return EncodingStats.NO_STATS;
}
public boolean hasNext()
{
return false;
}
public Unfiltered next()
{
throw new NoSuchElementException();
}
public void remove()
{
}
public void close()
{
}
};
return EmptyIterators.unfilteredRow(cfm, partitionKey, isReverseOrder, staticRow, partitionDeletion);
}
/**
@ -201,65 +142,45 @@ public abstract class UnfilteredRowIterators
&& iter1.columns().equals(iter2.columns())
&& iter1.staticRow().equals(iter2.staticRow());
return new AbstractUnfilteredRowIterator(iter1.metadata(),
iter1.partitionKey(),
iter1.partitionLevelDeletion(),
iter1.columns(),
iter1.staticRow(),
iter1.isReverseOrder(),
iter1.stats())
class Extend implements MoreRows<UnfilteredRowIterator>
{
protected Unfiltered computeNext()
boolean returned = false;
public UnfilteredRowIterator moreContents()
{
if (iter1.hasNext())
return iter1.next();
return iter2.hasNext() ? iter2.next() : endOfData();
if (returned)
return null;
returned = true;
return iter2;
}
}
@Override
public void close()
{
try
{
iter1.close();
}
finally
{
iter2.close();
}
}
};
return MoreRows.extend(iter1, new Extend());
}
public static UnfilteredRowIterator cloningIterator(UnfilteredRowIterator iterator, final AbstractAllocator allocator)
{
return new AlteringUnfilteredRowIterator(iterator)
class Cloner extends Transformation
{
private Row.Builder regularBuilder;
private final Row.Builder builder = allocator.cloningBTreeRowBuilder();
@Override
protected Row computeNextStatic(Row row)
public Row applyToStatic(Row row)
{
Row.Builder staticBuilder = allocator.cloningBTreeRowBuilder();
return Rows.copy(row, staticBuilder).build();
return Rows.copy(row, builder).build();
}
@Override
protected Row computeNext(Row row)
public Row applyToRow(Row row)
{
if (regularBuilder == null)
regularBuilder = allocator.cloningBTreeRowBuilder();
return Rows.copy(row, regularBuilder).build();
return Rows.copy(row, builder).build();
}
@Override
protected RangeTombstoneMarker computeNext(RangeTombstoneMarker marker)
public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
{
return marker.copy(allocator);
}
};
}
return Transformation.apply(iterator, new Cloner());
}
/**
@ -277,24 +198,24 @@ public abstract class UnfilteredRowIterators
*/
public static UnfilteredRowIterator withValidation(UnfilteredRowIterator iterator, final String filename)
{
return new AlteringUnfilteredRowIterator(iterator)
class Validator extends Transformation
{
@Override
protected Row computeNextStatic(Row row)
public Row applyToStatic(Row row)
{
validate(row);
return row;
}
@Override
protected Row computeNext(Row row)
public Row applyToRow(Row row)
{
validate(row);
return row;
}
@Override
protected RangeTombstoneMarker computeNext(RangeTombstoneMarker marker)
public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
{
validate(marker);
return marker;
@ -311,7 +232,8 @@ public abstract class UnfilteredRowIterators
throw new CorruptSSTableException(me, filename);
}
}
};
}
return Transformation.apply(iterator, new Validator());
}
/**
@ -331,30 +253,31 @@ public abstract class UnfilteredRowIterators
iterator.isReverseOrder(),
iterator.partitionLevelDeletion().markedForDeleteAt());
return new AlteringUnfilteredRowIterator(iterator)
class Logger extends Transformation
{
@Override
protected Row computeNextStatic(Row row)
public Row applyToStatic(Row row)
{
if (!row.isEmpty())
logger.info("[{}] {}", id, row.toString(metadata(), fullDetails));
logger.info("[{}] {}", id, row.toString(metadata, fullDetails));
return row;
}
@Override
protected Row computeNext(Row row)
public Row applyToRow(Row row)
{
logger.info("[{}] {}", id, row.toString(metadata(), fullDetails));
logger.info("[{}] {}", id, row.toString(metadata, fullDetails));
return row;
}
@Override
protected RangeTombstoneMarker computeNext(RangeTombstoneMarker marker)
public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
{
logger.info("[{}] {}", id, marker.toString(metadata()));
logger.info("[{}] {}", id, marker.toString(metadata));
return marker;
}
};
}
return Transformation.apply(iterator, new Logger());
}
/**
@ -577,66 +500,4 @@ public abstract class UnfilteredRowIterators
}
}
}
private static class FilteringIterator extends AbstractIterator<Row> implements RowIterator
{
private final UnfilteredRowIterator iter;
private final int nowInSec;
public FilteringIterator(UnfilteredRowIterator iter, int nowInSec)
{
this.iter = iter;
this.nowInSec = nowInSec;
}
public CFMetaData metadata()
{
return iter.metadata();
}
public boolean isReverseOrder()
{
return iter.isReverseOrder();
}
public PartitionColumns columns()
{
return iter.columns();
}
public DecoratedKey partitionKey()
{
return iter.partitionKey();
}
public Row staticRow()
{
Row row = iter.staticRow();
if (row.isEmpty())
return Rows.EMPTY_STATIC_ROW;
row = row.purge(DeletionPurger.PURGE_ALL, nowInSec);
return row == null ? Rows.EMPTY_STATIC_ROW : row;
}
protected Row computeNext()
{
while (iter.hasNext())
{
Unfiltered next = iter.next();
if (next.isRangeTombstoneMarker())
continue;
Row row = ((Row)next).purge(DeletionPurger.PURGE_ALL, nowInSec);
if (row != null)
return row;
}
return endOfData();
}
public void close()
{
iter.close();
}
}
}

View File

@ -1,79 +0,0 @@
/*
* 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.rows;
import com.google.common.collect.UnmodifiableIterator;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.*;
/**
* Abstract class to make writing atom iterators that wrap another iterator
* easier. By default, the wrapping iterator simply delegate every call to
* the wrapped iterator so concrete implementations will override some of the
* methods.
*/
public abstract class WrappingRowIterator extends UnmodifiableIterator<Row> implements RowIterator
{
protected final RowIterator wrapped;
protected WrappingRowIterator(RowIterator wrapped)
{
this.wrapped = wrapped;
}
public CFMetaData metadata()
{
return wrapped.metadata();
}
public boolean isReverseOrder()
{
return wrapped.isReverseOrder();
}
public PartitionColumns columns()
{
return wrapped.columns();
}
public DecoratedKey partitionKey()
{
return wrapped.partitionKey();
}
public Row staticRow()
{
return wrapped.staticRow();
}
public boolean hasNext()
{
return wrapped.hasNext();
}
public Row next()
{
return wrapped.next();
}
public void close()
{
wrapped.close();
}
}

View File

@ -29,7 +29,7 @@ import org.apache.cassandra.db.*;
* some of the methods.
* <p>
* Note that if most of what you want to do is modifying/filtering the returned
* {@code Unfiltered}, {@link AlteringUnfilteredRowIterator} can be a simpler option.
* {@code Unfiltered}, {@link org.apache.cassandra.db.transform.Transformation.apply} can be a simpler option.
*/
public abstract class WrappingUnfilteredRowIterator extends UnmodifiableIterator<Unfiltered> implements UnfilteredRowIterator
{

View File

@ -0,0 +1,129 @@
package org.apache.cassandra.db.transform;
import java.util.Iterator;
import java.util.NoSuchElementException;
import net.nicoulaj.compilecommand.annotations.DontInline;
import org.apache.cassandra.utils.CloseableIterator;
import static org.apache.cassandra.utils.Throwables.maybeFail;
import static org.apache.cassandra.utils.Throwables.merge;
abstract class BaseIterator<V, I extends CloseableIterator<? extends V>, O extends V> extends Stack implements AutoCloseable, Iterator<O>
{
I input;
V next;
Stop stop; // applies at the end of the current next()
static class Stop
{
// TODO: consider moving "next" into here, so that a stop() when signalled outside of a function call (e.g. in attach)
// can take effect immediately; this doesn't seem to be necessary at the moment, but it might cause least surprise in future
boolean isSignalled;
}
// responsibility for initialising next lies with the subclass
BaseIterator(BaseIterator<? extends V, ? extends I, ?> copyFrom)
{
super(copyFrom);
this.input = copyFrom.input;
this.next = copyFrom.next;
this.stop = copyFrom.stop;
}
BaseIterator(I input)
{
this.input = input;
this.stop = new Stop();
}
/**
* run the corresponding runOnClose method for the first length transformations.
*
* used in hasMoreContents to close the methods preceding the MoreContents
*/
protected abstract Throwable runOnClose(int length);
/**
* apply the relevant method from the transformation to the value.
*
* used in hasMoreContents to apply the functions that follow the MoreContents
*/
protected abstract V applyOne(V value, Transformation transformation);
public final void close()
{
Throwable fail = runOnClose(length);
if (next instanceof AutoCloseable)
{
try { ((AutoCloseable) next).close(); }
catch (Throwable t) { fail = merge(fail, t); }
}
try { input.close(); }
catch (Throwable t) { fail = merge(fail, t); }
maybeFail(fail);
}
public final O next()
{
if (next == null && !hasNext())
throw new NoSuchElementException();
O next = (O) this.next;
this.next = null;
return next;
}
// may set next != null if the next contents are a transforming iterator that already has data to return,
// in which case we immediately have more contents to yield
protected final boolean hasMoreContents()
{
return moreContents.length > 0 && tryGetMoreContents();
}
@DontInline
private boolean tryGetMoreContents()
{
for (int i = 0 ; i < moreContents.length ; i++)
{
MoreContentsHolder holder = moreContents[i];
MoreContents provider = holder.moreContents;
I newContents = (I) provider.moreContents();
if (newContents == null)
continue;
input.close();
input = newContents;
Stack prefix = EMPTY;
if (newContents instanceof BaseIterator)
{
// we're refilling with transformed contents, so swap in its internals directly
// TODO: ensure that top-level data is consistent. i.e. staticRow, partitionlevelDeletion etc are same?
BaseIterator abstr = (BaseIterator) newContents;
prefix = abstr;
input = (I) abstr.input;
next = apply((V) abstr.next, holder.length); // must apply all remaining functions to the next, if any
}
// since we're truncating our transformation stack to only those occurring after the extend transformation
// we have to run any prior runOnClose methods
maybeFail(runOnClose(holder.length));
refill(prefix, holder, i);
if (next != null || input.hasNext())
return true;
i = -1;
}
return false;
}
// apply the functions [from..length)
private V apply(V next, int from)
{
while (next != null & from < length)
next = applyOne(next, stack[from++]);
return next;
}
}

View File

@ -0,0 +1,100 @@
package org.apache.cassandra.db.transform;
import java.util.Collections;
import org.apache.cassandra.db.partitions.BasePartitionIterator;
import org.apache.cassandra.db.rows.BaseRowIterator;
import org.apache.cassandra.utils.Throwables;
import static org.apache.cassandra.utils.Throwables.merge;
public abstract class BasePartitions<R extends BaseRowIterator<?>, I extends BasePartitionIterator<? extends BaseRowIterator<?>>>
extends BaseIterator<BaseRowIterator<?>, I, R>
implements BasePartitionIterator<R>
{
public BasePartitions(I input)
{
super(input);
}
BasePartitions(BasePartitions<?, ? extends I> copyFrom)
{
super(copyFrom);
}
// *********************************
protected BaseRowIterator<?> applyOne(BaseRowIterator<?> value, Transformation transformation)
{
return value == null ? null : transformation.applyToPartition(value);
}
void add(Transformation transformation)
{
transformation.attachTo(this);
super.add(transformation);
next = applyOne(next, transformation);
}
protected Throwable runOnClose(int length)
{
Throwable fail = null;
Transformation[] fs = stack;
for (int i = 0 ; i < length ; i++)
{
try
{
fs[i].onClose();
}
catch (Throwable t)
{
fail = merge(fail, t);
}
}
return fail;
}
public final boolean hasNext()
{
BaseRowIterator<?> next = null;
try
{
Stop stop = this.stop;
while (this.next == null)
{
Transformation[] fs = stack;
int len = length;
while (!stop.isSignalled && input.hasNext())
{
next = input.next();
for (int i = 0 ; next != null & i < len ; i++)
next = fs[i].applyToPartition(next);
if (next != null)
{
this.next = next;
return true;
}
}
if (stop.isSignalled || !hasMoreContents())
return false;
}
return true;
}
catch (Throwable t)
{
if (next != null)
Throwables.close(t, Collections.singleton(next));
throw t;
}
}
}

View File

@ -0,0 +1,139 @@
package org.apache.cassandra.db.transform;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.PartitionColumns;
import org.apache.cassandra.db.rows.*;
import static org.apache.cassandra.utils.Throwables.merge;
public abstract class BaseRows<R extends Unfiltered, I extends BaseRowIterator<? extends Unfiltered>>
extends BaseIterator<Unfiltered, I, R>
implements BaseRowIterator<R>
{
private Row staticRow;
public BaseRows(I input)
{
super(input);
staticRow = input.staticRow();
}
// swap parameter order to avoid casting errors
BaseRows(BaseRows<?, ? extends I> copyFrom)
{
super(copyFrom);
staticRow = copyFrom.staticRow;
}
public CFMetaData metadata()
{
return input.metadata();
}
public boolean isReverseOrder()
{
return input.isReverseOrder();
}
public PartitionColumns columns()
{
return input.columns();
}
public DecoratedKey partitionKey()
{
return input.partitionKey();
}
public Row staticRow()
{
return staticRow;
}
// **************************
@Override
protected Throwable runOnClose(int length)
{
Throwable fail = null;
Transformation[] fs = stack;
for (int i = 0 ; i < length ; i++)
{
try
{
fs[i].onPartitionClose();
}
catch (Throwable t)
{
fail = merge(fail, t);
}
}
return fail;
}
@Override
void add(Transformation transformation)
{
transformation.attachTo(this);
super.add(transformation);
// transform any existing data
staticRow = transformation.applyToStatic(staticRow);
next = applyOne(next, transformation);
}
@Override
protected Unfiltered applyOne(Unfiltered value, Transformation transformation)
{
return value == null
? null
: value instanceof Row
? transformation.applyToRow((Row) value)
: transformation.applyToMarker((RangeTombstoneMarker) value);
}
@Override
public final boolean hasNext()
{
Stop stop = this.stop;
while (this.next == null)
{
Transformation[] fs = stack;
int len = length;
while (!stop.isSignalled && input.hasNext())
{
Unfiltered next = input.next();
if (next.isRow())
{
Row row = (Row) next;
for (int i = 0 ; row != null && i < len ; i++)
row = fs[i].applyToRow(row);
next = row;
}
else
{
RangeTombstoneMarker rtm = (RangeTombstoneMarker) next;
for (int i = 0 ; rtm != null && i < len ; i++)
rtm = fs[i].applyToMarker(rtm);
next = rtm;
}
if (next != null)
{
this.next = next;
return true;
}
}
if (stop.isSignalled || !hasMoreContents())
return false;
}
return true;
}
}

View File

@ -0,0 +1,56 @@
package org.apache.cassandra.db.transform;
import org.apache.cassandra.db.DeletionPurger;
import org.apache.cassandra.db.rows.*;
final class Filter extends Transformation
{
private final boolean filterEmpty; // generally maps to !isForThrift, but also false for direct row filtration
private final int nowInSec;
public Filter(boolean filterEmpty, int nowInSec)
{
this.filterEmpty = filterEmpty;
this.nowInSec = nowInSec;
}
public RowIterator applyToPartition(BaseRowIterator iterator)
{
RowIterator filtered = iterator instanceof UnfilteredRows
? new FilteredRows(this, (UnfilteredRows) iterator)
: new FilteredRows((UnfilteredRowIterator) iterator, this);
if (filterEmpty && closeIfEmpty(filtered))
return null;
return filtered;
}
public Row applyToStatic(Row row)
{
if (row.isEmpty())
return Rows.EMPTY_STATIC_ROW;
row = row.purge(DeletionPurger.PURGE_ALL, nowInSec);
return row == null ? Rows.EMPTY_STATIC_ROW : row;
}
public Row applyToRow(Row row)
{
return row.purge(DeletionPurger.PURGE_ALL, nowInSec);
}
public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
{
return null;
}
private static boolean closeIfEmpty(BaseRowIterator<?> iter)
{
if (iter.isEmpty())
{
iter.close();
return true;
}
return false;
}
}

View File

@ -0,0 +1,40 @@
package org.apache.cassandra.db.transform;
import org.apache.cassandra.db.partitions.BasePartitionIterator;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.RowIterator;
public final class FilteredPartitions extends BasePartitions<RowIterator, BasePartitionIterator<?>> implements PartitionIterator
{
// wrap basic iterator for transformation
FilteredPartitions(PartitionIterator input)
{
super(input);
}
// wrap basic unfiltered iterator for transformation, applying filter as first transformation
FilteredPartitions(UnfilteredPartitionIterator input, Filter filter)
{
super(input);
add(filter);
}
// copy from an UnfilteredPartitions, applying a filter to convert it
FilteredPartitions(Filter filter, UnfilteredPartitions copyFrom)
{
super(copyFrom);
add(filter);
}
/**
* Filter any RangeTombstoneMarker from the iterator's iterators, transforming it into a PartitionIterator.
*/
public static PartitionIterator filter(UnfilteredPartitionIterator iterator, int nowInSecs)
{
Filter filter = new Filter(!iterator.isForThrift(), nowInSecs);
if (iterator instanceof UnfilteredPartitions)
return new FilteredPartitions(filter, (UnfilteredPartitions) iterator);
return new FilteredPartitions(iterator, filter);
}
}

View File

@ -0,0 +1,40 @@
package org.apache.cassandra.db.transform;
import org.apache.cassandra.db.rows.BaseRowIterator;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
public final class FilteredRows extends BaseRows<Row, BaseRowIterator<?>> implements RowIterator
{
FilteredRows(RowIterator input)
{
super(input);
}
FilteredRows(UnfilteredRowIterator input, Filter filter)
{
super(input);
add(filter);
}
FilteredRows(Filter filter, UnfilteredRows input)
{
super(input);
add(filter);
}
@Override
public boolean isEmpty()
{
return staticRow().isEmpty() && !hasNext();
}
/**
* Filter any RangeTombstoneMarker from the iterator, transforming it into a RowIterator.
*/
public static RowIterator filter(UnfilteredRowIterator iterator, int nowInSecs)
{
return new Filter(false, nowInSecs).applyToPartition(iterator);
}
}

View File

@ -0,0 +1,8 @@
package org.apache.cassandra.db.transform;
// a shared internal interface, that is hidden to provide type-safety to the user
interface MoreContents<I>
{
public abstract I moreContents();
}

View File

@ -0,0 +1,35 @@
package org.apache.cassandra.db.transform;
import org.apache.cassandra.db.partitions.BasePartitionIterator;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import static org.apache.cassandra.db.transform.Transformation.add;
import static org.apache.cassandra.db.transform.Transformation.mutable;
/**
* An interface for providing new partitions for a partitions iterator.
*
* The new contents are produced as a normal arbitrary PartitionIterator or UnfilteredPartitionIterator (as appropriate)
*
* The transforming iterator invokes this method when any current source is exhausted, then then inserts the
* new contents as the new source.
*
* If the new source is itself a product of any transformations, the two transforming iterators are merged
* so that control flow always occurs at the outermost point
*/
public interface MorePartitions<I extends BasePartitionIterator<?>> extends MoreContents<I>
{
public static UnfilteredPartitionIterator extend(UnfilteredPartitionIterator iterator, MorePartitions<? super UnfilteredPartitionIterator> more)
{
return add(mutable(iterator), more);
}
public static PartitionIterator extend(PartitionIterator iterator, MorePartitions<? super PartitionIterator> more)
{
return add(mutable(iterator), more);
}
}

View File

@ -0,0 +1,36 @@
package org.apache.cassandra.db.transform;
import org.apache.cassandra.db.rows.BaseRowIterator;
import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import static org.apache.cassandra.db.transform.Transformation.add;
import static org.apache.cassandra.db.transform.Transformation.mutable;
/**
* An interface for providing new row contents for a partition.
*
* The new contents are produced as a normal arbitrary RowIterator or UnfilteredRowIterator (as appropriate),
* with matching staticRow, partitionKey and partitionLevelDeletion.
*
* The transforming iterator invokes this method when any current source is exhausted, then then inserts the
* new contents as the new source.
*
* If the new source is itself a product of any transformations, the two transforming iterators are merged
* so that control flow always occurs at the outermost point
*/
public interface MoreRows<I extends BaseRowIterator<?>> extends MoreContents<I>
{
public static UnfilteredRowIterator extend(UnfilteredRowIterator iterator, MoreRows<? super UnfilteredRowIterator> more)
{
return add(mutable(iterator), more);
}
public static RowIterator extend(RowIterator iterator, MoreRows<? super RowIterator> more)
{
return add(mutable(iterator), more);
}
}

View File

@ -0,0 +1,81 @@
package org.apache.cassandra.db.transform;
import java.util.Arrays;
class Stack
{
static final Stack EMPTY = new Stack();
Transformation[] stack;
int length; // number of used stack entries
MoreContentsHolder[] moreContents; // stack of more contents providers (if any; usually zero or one)
// an internal placeholder for a MoreContents, storing the associated stack length at time it was applied
static class MoreContentsHolder
{
final MoreContents moreContents;
int length;
private MoreContentsHolder(MoreContents moreContents, int length)
{
this.moreContents = moreContents;
this.length = length;
}
}
Stack()
{
stack = new Transformation[0];
moreContents = new MoreContentsHolder[0];
}
Stack(Stack copy)
{
stack = copy.stack;
length = copy.length;
moreContents = copy.moreContents;
}
void add(Transformation add)
{
if (length == stack.length)
stack = resize(stack);
stack[length++] = add;
}
void add(MoreContents more)
{
this.moreContents = Arrays.copyOf(moreContents, moreContents.length + 1);
this.moreContents[moreContents.length - 1] = new MoreContentsHolder(more, length);
}
private static <E> E[] resize(E[] array)
{
int newLen = array.length == 0 ? 5 : array.length * 2;
return Arrays.copyOf(array, newLen);
}
// reinitialise the transformations after a moreContents applies
void refill(Stack prefix, MoreContentsHolder holder, int index)
{
// drop the transformations that were present when the MoreContents was attached,
// and prefix any transformations in the new contents (if it's a transformer)
moreContents = splice(prefix.moreContents, prefix.moreContents.length, moreContents, index, moreContents.length);
stack = splice(prefix.stack, prefix.length, stack, holder.length, length);
length += prefix.length - holder.length;
holder.length = prefix.length;
}
private static <E> E[] splice(E[] prefix, int prefixCount, E[] keep, int keepFrom, int keepTo)
{
int keepCount = keepTo - keepFrom;
int newCount = prefixCount + keepCount;
if (newCount > keep.length)
keep = Arrays.copyOf(keep, newCount);
if (keepFrom != prefixCount)
System.arraycopy(keep, keepFrom, keep, prefixCount, keepCount);
if (prefixCount != 0)
System.arraycopy(prefix, 0, keep, 0, prefixCount);
return keep;
}
}

View File

@ -0,0 +1,60 @@
package org.apache.cassandra.db.transform;
import net.nicoulaj.compilecommand.annotations.DontInline;
import org.apache.cassandra.db.rows.BaseRowIterator;
// A Transformation that can stop an iterator earlier than its natural exhaustion
public abstract class StoppingTransformation<I extends BaseRowIterator<?>> extends Transformation<I>
{
private BaseIterator.Stop stop;
private BaseIterator.Stop stopInPartition;
/**
* If invoked by a subclass, any partitions iterator this transformation has been applied to will terminate
* after any currently-processing item is returned, as will any row/unfiltered iterator
*/
@DontInline
protected void stop()
{
if (stop != null)
stop.isSignalled = true;
stopInPartition();
}
/**
* If invoked by a subclass, any rows/unfiltered iterator this transformation has been applied to will terminate
* after any currently-processing item is returned
*/
@DontInline
protected void stopInPartition()
{
if (stopInPartition != null)
stopInPartition.isSignalled = true;
}
@Override
protected void attachTo(BasePartitions partitions)
{
assert this.stop == null;
this.stop = partitions.stop;
}
@Override
protected void attachTo(BaseRows rows)
{
assert this.stopInPartition == null;
this.stopInPartition = rows.stop;
}
@Override
protected void onClose()
{
stop = null;
}
@Override
protected void onPartitionClose()
{
stopInPartition = null;
}
}

View File

@ -0,0 +1,145 @@
package org.apache.cassandra.db.transform;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.*;
/**
* We have a single common superclass for all Transformations to make implementation efficient.
* we have a shared stack for all transformations, and can share the same transformation across partition and row
* iterators, reducing garbage. Internal code is also simplified by always having a basic no-op implementation to invoke.
*
* Only the necessary methods need be overridden. Early termination is provided by invoking the method's stop or stopInPartition
* methods, rather than having their own abstract method to invoke, as this is both more efficient and simpler to reason about.
*/
public abstract class Transformation<I extends BaseRowIterator<?>>
{
// internal methods for StoppableTransformation only
void attachTo(BasePartitions partitions) { }
void attachTo(BaseRows rows) { }
/**
* Run on the close of any (logical) partitions iterator this function was applied to
*
* We stipulate logical, because if applied to a transformed iterator the lifetime of the iterator
* object may be longer than the lifetime of the "logical" iterator it was applied to; if the iterator
* is refilled with MoreContents, for instance, the iterator may outlive this function
*/
protected void onClose() { }
/**
* Run on the close of any (logical) rows iterator this function was applied to
*
* We stipulate logical, because if applied to a transformed iterator the lifetime of the iterator
* object may be longer than the lifetime of the "logical" iterator it was applied to; if the iterator
* is refilled with MoreContents, for instance, the iterator may outlive this function
*/
protected void onPartitionClose() { }
/**
* Applied to any rows iterator (partition) we encounter in a partitions iterator
*/
protected I applyToPartition(I partition)
{
return partition;
}
/**
* Applied to any row we encounter in a rows iterator
*/
protected Row applyToRow(Row row)
{
return row;
}
/**
* Applied to any RTM we encounter in a rows/unfiltered iterator
*/
protected RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
{
return marker;
}
/**
* Applied to the static row of any rows iterator.
*
* NOTE that this is only applied to the first iterator in any sequence of iterators filled by a MoreContents;
* the static data for such iterators is all expected to be equal
*/
protected Row applyToStatic(Row row)
{
return row;
}
/**
* Applied to the partition-level deletion of any rows iterator.
*
* NOTE that this is only applied to the first iterator in any sequence of iterators filled by a MoreContents;
* the static data for such iterators is all expected to be equal
*/
protected DeletionTime applyToDeletion(DeletionTime deletionTime)
{
return deletionTime;
}
//******************************************************
// Static Application Methods
//******************************************************
public static UnfilteredPartitionIterator apply(UnfilteredPartitionIterator iterator, Transformation<? super UnfilteredRowIterator> transformation)
{
return add(mutable(iterator), transformation);
}
public static PartitionIterator apply(PartitionIterator iterator, Transformation<? super RowIterator> transformation)
{
return add(mutable(iterator), transformation);
}
public static UnfilteredRowIterator apply(UnfilteredRowIterator iterator, Transformation<?> transformation)
{
return add(mutable(iterator), transformation);
}
public static RowIterator apply(RowIterator iterator, Transformation<?> transformation)
{
return add(mutable(iterator), transformation);
}
static UnfilteredPartitions mutable(UnfilteredPartitionIterator iterator)
{
return iterator instanceof UnfilteredPartitions
? (UnfilteredPartitions) iterator
: new UnfilteredPartitions(iterator);
}
static FilteredPartitions mutable(PartitionIterator iterator)
{
return iterator instanceof FilteredPartitions
? (FilteredPartitions) iterator
: new FilteredPartitions(iterator);
}
static UnfilteredRows mutable(UnfilteredRowIterator iterator)
{
return iterator instanceof UnfilteredRows
? (UnfilteredRows) iterator
: new UnfilteredRows(iterator);
}
static FilteredRows mutable(RowIterator iterator)
{
return iterator instanceof FilteredRows
? (FilteredRows) iterator
: new FilteredRows(iterator);
}
static <E extends BaseIterator> E add(E to, Transformation add)
{
to.add(add);
return to;
}
static <E extends BaseIterator> E add(E to, MoreContents add)
{
to.add(add);
return to;
}
}

View File

@ -0,0 +1,27 @@
package org.apache.cassandra.db.transform;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
final class UnfilteredPartitions extends BasePartitions<UnfilteredRowIterator, UnfilteredPartitionIterator> implements UnfilteredPartitionIterator
{
final boolean isForThrift;
// wrap an iterator for transformation
public UnfilteredPartitions(UnfilteredPartitionIterator input)
{
super(input);
this.isForThrift = input.isForThrift();
}
public boolean isForThrift()
{
return isForThrift;
}
public CFMetaData metadata()
{
return input.metadata();
}
}

View File

@ -0,0 +1,40 @@
package org.apache.cassandra.db.transform;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.rows.EncodingStats;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
final class UnfilteredRows extends BaseRows<Unfiltered, UnfilteredRowIterator> implements UnfilteredRowIterator
{
private DeletionTime partitionLevelDeletion;
public UnfilteredRows(UnfilteredRowIterator input)
{
super(input);
partitionLevelDeletion = input.partitionLevelDeletion();
}
@Override
void add(Transformation add)
{
super.add(add);
partitionLevelDeletion = add.applyToDeletion(partitionLevelDeletion);
}
public DeletionTime partitionLevelDeletion()
{
return partitionLevelDeletion;
}
public EncodingStats stats()
{
return input.stats();
}
@Override
public boolean isEmpty()
{
return staticRow().isEmpty() && partitionLevelDeletion().isLive() && !hasNext();
}
}

View File

@ -71,14 +71,7 @@ public class SecondaryIndexBuilder extends CompactionInfo.Holder
}
finally
{
try
{
iter.close();
}
catch (IOException e)
{
throw new RuntimeException(e);
}
iter.close();
}
}
}

View File

@ -31,6 +31,7 @@ import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.index.internal.CassandraIndex;
import org.apache.cassandra.index.internal.CassandraIndexSearcher;
import org.apache.cassandra.index.internal.IndexEntry;
@ -203,18 +204,13 @@ public class CompositesSearcher extends CassandraIndexSearcher
});
}
return new AlteringUnfilteredRowIterator(dataIter)
ClusteringComparator comparator = dataIter.metadata().comparator;
class Transform extends Transformation
{
private int entriesIdx;
public void close()
{
deleteAllEntries(staleEntries, writeOp, nowInSec);
super.close();
}
@Override
protected Row computeNext(Row row)
public Row applyToRow(Row row)
{
IndexEntry entry = findEntry(row.clustering());
if (!index.isStale(row, indexValue, nowInSec))
@ -234,7 +230,7 @@ public class CompositesSearcher extends CassandraIndexSearcher
// next entry, the one at 'entriesIdx'. However, we can have stale entries, entries
// that have no corresponding row in the base table typically because of a range
// tombstone or partition level deletion. Delete such stale entries.
int cmp = metadata().comparator.compare(entry.indexedEntryClustering, clustering);
int cmp = comparator.compare(entry.indexedEntryClustering, clustering);
assert cmp <= 0; // this would means entries are not in clustering order, which shouldn't happen
if (cmp == 0)
return entry;
@ -244,6 +240,14 @@ public class CompositesSearcher extends CassandraIndexSearcher
// entries correspond to the rows we've queried, so we shouldn't have a row that has no corresponding entry.
throw new AssertionError();
}
};
@Override
public void onClose()
{
deleteAllEntries(staleEntries, writeOp, nowInSec);
}
}
return Transformation.apply(dataIter, new Transform());
}
}

View File

@ -17,7 +17,6 @@
*/
package org.apache.cassandra.io.sstable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
@ -70,12 +69,10 @@ public class ReducingKeyIterator implements CloseableIterator<DecoratedKey>
}
}
public void close() throws IOException
public void close()
{
if (mi != null)
{
mi.close();
}
}
public long getTotalBytes()

View File

@ -22,6 +22,7 @@ import java.util.Map;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.io.sstable.*;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.SSTableWriter;
@ -143,11 +144,11 @@ public class BigTableWriter extends SSTableWriter
long startPosition = beforeAppend(key);
try (StatsCollector withStats = new StatsCollector(iterator, metadataCollector))
try (UnfilteredRowIterator collecting = Transformation.apply(iterator, new StatsCollector(metadataCollector)))
{
ColumnIndex index = ColumnIndex.writeAndBuildIndex(withStats, dataFile, header, descriptor.version);
ColumnIndex index = ColumnIndex.writeAndBuildIndex(collecting, dataFile, header, descriptor.version);
RowIndexEntry entry = RowIndexEntry.create(startPosition, iterator.partitionLevelDeletion(), index);
RowIndexEntry entry = RowIndexEntry.create(startPosition, collecting.partitionLevelDeletion(), index);
long endPosition = dataFile.position();
long rowSize = endPosition - startPosition;
@ -171,20 +172,18 @@ public class BigTableWriter extends SSTableWriter
}
}
private static class StatsCollector extends AlteringUnfilteredRowIterator
private static class StatsCollector extends Transformation
{
private final MetadataCollector collector;
private int cellCount;
StatsCollector(UnfilteredRowIterator iter, MetadataCollector collector)
StatsCollector(MetadataCollector collector)
{
super(iter);
this.collector = collector;
collector.update(iter.partitionLevelDeletion());
}
@Override
protected Row computeNextStatic(Row row)
public Row applyToStatic(Row row)
{
if (!row.isEmpty())
cellCount += Rows.collectStats(row, collector);
@ -192,7 +191,7 @@ public class BigTableWriter extends SSTableWriter
}
@Override
protected Row computeNext(Row row)
public Row applyToRow(Row row)
{
collector.updateClusteringValues(row.clustering());
cellCount += Rows.collectStats(row, collector);
@ -200,7 +199,7 @@ public class BigTableWriter extends SSTableWriter
}
@Override
protected RangeTombstoneMarker computeNext(RangeTombstoneMarker marker)
public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
{
collector.updateClusteringValues(marker.clustering());
if (marker.isBoundary())
@ -217,10 +216,16 @@ public class BigTableWriter extends SSTableWriter
}
@Override
public void close()
public void onPartitionClose()
{
collector.addCellPerPartitionCount(cellCount);
super.close();
}
@Override
public DeletionTime applyToDeletion(DeletionTime deletionTime)
{
collector.update(deletionTime);
return deletionTime;
}
}

View File

@ -23,6 +23,7 @@ import java.util.concurrent.TimeoutException;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.*;
@ -30,6 +31,8 @@ import org.apache.cassandra.db.filter.ClusteringIndexFilter;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.transform.MoreRows;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.net.*;
import org.apache.cassandra.tracing.Tracing;
@ -67,7 +70,7 @@ public class DataResolver extends ResponseResolver
// Even though every responses should honor the limit, we might have more than requested post reconciliation,
// so ensure we're respecting the limit.
DataLimits.Counter counter = command.limits().newCounter(command.nowInSec(), true);
return new CountingPartitionIterator(mergeWithShortReadProtection(iters, sources, counter), counter);
return counter.applyTo(mergeWithShortReadProtection(iters, sources, counter));
}
private PartitionIterator mergeWithShortReadProtection(List<UnfilteredPartitionIterator> results, InetAddress[] sources, DataLimits.Counter resultCounter)
@ -80,11 +83,11 @@ public class DataResolver extends ResponseResolver
// So-called "short reads" stems from nodes returning only a subset of the results they have for a partition due to the limit,
// but that subset not being enough post-reconciliation. So if we don't have limit, don't bother.
if (command.limits().isUnlimited())
return UnfilteredPartitionIterators.mergeAndFilter(results, command.nowInSec(), listener);
for (int i = 0; i < results.size(); i++)
results.set(i, new ShortReadProtectedIterator(sources[i], results.get(i), resultCounter));
if (!command.limits().isUnlimited())
{
for (int i = 0; i < results.size(); i++)
results.set(i, Transformation.apply(results.get(i), new ShortReadProtection(sources[i], resultCounter)));
}
return UnfilteredPartitionIterators.mergeAndFilter(results, command.nowInSec(), listener);
}
@ -281,78 +284,53 @@ public class DataResolver extends ResponseResolver
}
}
private class ShortReadProtectedIterator extends CountingUnfilteredPartitionIterator
private class ShortReadProtection extends Transformation<UnfilteredRowIterator>
{
private final InetAddress source;
private final DataLimits.Counter counter;
private final DataLimits.Counter postReconciliationCounter;
private ShortReadProtectedIterator(InetAddress source, UnfilteredPartitionIterator iterator, DataLimits.Counter postReconciliationCounter)
private ShortReadProtection(InetAddress source, DataLimits.Counter postReconciliationCounter)
{
super(iterator, command.limits().newCounter(command.nowInSec(), false));
this.source = source;
this.counter = command.limits().newCounter(command.nowInSec(), false).onlyCount();
this.postReconciliationCounter = postReconciliationCounter;
}
@Override
public UnfilteredRowIterator next()
public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
return new ShortReadProtectedRowIterator(super.next());
partition = Transformation.apply(partition, counter);
// must apply and extend with same protection instance
ShortReadRowProtection protection = new ShortReadRowProtection(partition.metadata(), partition.partitionKey());
partition = MoreRows.extend(partition, protection);
partition = Transformation.apply(partition, protection); // apply after, so it is retained when we extend (in case we need to reextend)
return partition;
}
private class ShortReadProtectedRowIterator extends WrappingUnfilteredRowIterator
private class ShortReadRowProtection extends Transformation implements MoreRows<UnfilteredRowIterator>
{
private boolean initialReadIsDone;
private UnfilteredRowIterator shortReadContinuation;
private Clustering lastClustering;
final CFMetaData metadata;
final DecoratedKey partitionKey;
Clustering lastClustering;
int lastCount = 0;
ShortReadProtectedRowIterator(UnfilteredRowIterator iter)
private ShortReadRowProtection(CFMetaData metadata, DecoratedKey partitionKey)
{
super(iter);
this.metadata = metadata;
this.partitionKey = partitionKey;
}
@Override
public boolean hasNext()
public Row applyToRow(Row row)
{
if (super.hasNext())
return true;
initialReadIsDone = true;
if (shortReadContinuation != null && shortReadContinuation.hasNext())
return true;
return checkForShortRead();
lastClustering = row.clustering();
return row;
}
@Override
public Unfiltered next()
public UnfilteredRowIterator moreContents()
{
Unfiltered next = initialReadIsDone ? shortReadContinuation.next() : super.next();
if (next.kind() == Unfiltered.Kind.ROW)
lastClustering = ((Row)next).clustering();
return next;
}
@Override
public void close()
{
try
{
super.close();
}
finally
{
if (shortReadContinuation != null)
shortReadContinuation.close();
}
}
private boolean checkForShortRead()
{
assert shortReadContinuation == null || !shortReadContinuation.hasNext();
// We have a short read if the node this is the result of has returned the requested number of
// rows for that partition (i.e. it has stopped returning results due to the limit), but some of
// those results haven't made it in the final result post-reconciliation due to other nodes
@ -363,8 +341,9 @@ public class DataResolver extends ResponseResolver
// Also note that we only get here once all the results for this node have been returned, and so
// if the node had returned the requested number but we still get there, it imply some results were
// skipped during reconciliation.
if (!counter.isDoneForPartition())
return false;
if (lastCount == counter.counted() || !counter.isDoneForPartition())
return null;
lastCount = counter.counted();
assert !postReconciliationCounter.isDoneForPartition();
@ -378,23 +357,20 @@ public class DataResolver extends ResponseResolver
// counting iterator.
int n = postReconciliationCounter.countedInCurrentPartition();
int x = counter.countedInCurrentPartition();
int toQuery = x == 0
? n * 2 // We didn't got any answer, so (somewhat randomly) ask for twice as much
: Math.max(((n * n) / x) - n, 1);
int toQuery = Math.max(((n * n) / x) - n, 1);
DataLimits retryLimits = command.limits().forShortReadRetry(toQuery);
ClusteringIndexFilter filter = command.clusteringIndexFilter(partitionKey());
ClusteringIndexFilter retryFilter = lastClustering == null ? filter : filter.forPaging(metadata().comparator, lastClustering, false);
ClusteringIndexFilter filter = command.clusteringIndexFilter(partitionKey);
ClusteringIndexFilter retryFilter = lastClustering == null ? filter : filter.forPaging(metadata.comparator, lastClustering, false);
SinglePartitionReadCommand<?> cmd = SinglePartitionReadCommand.create(command.metadata(),
command.nowInSec(),
command.columnFilter(),
command.rowFilter(),
retryLimits,
partitionKey(),
partitionKey,
retryFilter);
shortReadContinuation = doShortReadRetry(cmd);
return shortReadContinuation.hasNext();
return doShortReadRetry(cmd);
}
private UnfilteredRowIterator doShortReadRetry(SinglePartitionReadCommand<?> retryCommand)
@ -402,7 +378,7 @@ public class DataResolver extends ResponseResolver
DataResolver resolver = new DataResolver(keyspace, retryCommand, ConsistencyLevel.ONE, 1);
ReadCallback handler = new ReadCallback(resolver, ConsistencyLevel.ONE, retryCommand, Collections.singletonList(source));
if (StorageProxy.canDoLocalRequest(source))
StageManager.getStage(Stage.READ).maybeExecuteImmediately(new StorageProxy.LocalReadRunnable(retryCommand, handler));
StageManager.getStage(Stage.READ).maybeExecuteImmediately(new StorageProxy.LocalReadRunnable(retryCommand, handler));
else
MessagingService.instance().sendRRWithFailure(retryCommand.createMessage(MessagingService.current_version), source, handler);

View File

@ -1914,7 +1914,8 @@ public class StorageProxy implements StorageProxyMBean
private final ConsistencyLevel consistency;
private final long startTime;
private CountingPartitionIterator sentQueryIterator;
private DataLimits.Counter counter;
private PartitionIterator sentQueryIterator;
private int concurrencyFactor;
// The two following "metric" are maintained to improve the concurrencyFactor
@ -1944,7 +1945,7 @@ public class StorageProxy implements StorageProxyMBean
// else, sends the next batch of concurrent queries (after having close the previous iterator)
if (sentQueryIterator != null)
{
liveReturned += sentQueryIterator.counter().counted();
liveReturned += counter.counted();
sentQueryIterator.close();
// It's not the first batch of queries and we're not done, so we we can use what has been
@ -2005,7 +2006,7 @@ public class StorageProxy implements StorageProxyMBean
return new SingleRangeResponse(handler);
}
private CountingPartitionIterator sendNextRequests()
private PartitionIterator sendNextRequests()
{
List<PartitionIterator> concurrentQueries = new ArrayList<>(concurrencyFactor);
for (int i = 0; i < concurrencyFactor && ranges.hasNext(); i++)
@ -2017,7 +2018,8 @@ public class StorageProxy implements StorageProxyMBean
Tracing.trace("Submitted {} concurrent range requests", concurrentQueries.size());
// We want to count the results for the sake of updating the concurrency factor (see updateConcurrencyFactor) but we don't want to
// enforce any particular limit at this point (this could break code than rely on postReconciliationProcessing), hence the DataLimits.NONE.
return new CountingPartitionIterator(PartitionIterators.concat(concurrentQueries), DataLimits.NONE, command.nowInSec());
counter = DataLimits.NONE.newCounter(command.nowInSec(), true);
return counter.applyTo(PartitionIterators.concat(concurrentQueries));
}
public void close()

View File

@ -17,12 +17,11 @@
*/
package org.apache.cassandra.service.pager;
import java.util.NoSuchElementException;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.exceptions.RequestValidationException;
import org.apache.cassandra.service.ClientState;
@ -61,87 +60,65 @@ abstract class AbstractQueryPager implements QueryPager
public PartitionIterator fetchPage(int pageSize, ConsistencyLevel consistency, ClientState clientState) throws RequestValidationException, RequestExecutionException
{
if (isExhausted())
return PartitionIterators.EMPTY;
return EmptyIterators.partition();
pageSize = Math.min(pageSize, remaining);
return new PagerIterator(nextPageReadCommand(pageSize).execute(consistency, clientState), limits.forPaging(pageSize), command.nowInSec());
Pager pager = new Pager(limits.forPaging(pageSize), command.nowInSec());
return Transformation.apply(nextPageReadCommand(pageSize).execute(consistency, clientState), pager);
}
public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException
{
if (isExhausted())
return PartitionIterators.EMPTY;
return EmptyIterators.partition();
pageSize = Math.min(pageSize, remaining);
return new PagerIterator(nextPageReadCommand(pageSize).executeInternal(executionController), limits.forPaging(pageSize), command.nowInSec());
Pager pager = new Pager(limits.forPaging(pageSize), command.nowInSec());
return Transformation.apply(nextPageReadCommand(pageSize).executeInternal(executionController), pager);
}
private class PagerIterator extends CountingPartitionIterator
private class Pager extends Transformation<RowIterator>
{
private final DataLimits pageLimits;
private final DataLimits.Counter counter;
private Row lastRow;
private boolean isFirstPartition = true;
private RowIterator nextPartition;
private PagerIterator(PartitionIterator iter, DataLimits pageLimits, int nowInSec)
private Pager(DataLimits pageLimits, int nowInSec)
{
super(iter, pageLimits, nowInSec);
this.counter = pageLimits.newCounter(nowInSec, true);
this.pageLimits = pageLimits;
}
@Override
@SuppressWarnings("resource") // iter is closed by closing the result or in close()
public boolean hasNext()
public RowIterator applyToPartition(RowIterator partition)
{
while (nextPartition == null && super.hasNext())
DecoratedKey key = partition.partitionKey();
if (lastKey == null || !lastKey.equals(key))
remainingInPartition = limits.perPartitionCount();
lastKey = key;
// If this is the first partition of this page, this could be the continuation of a partition we've started
// on the previous page. In which case, we could have the problem that the partition has no more "regular"
// rows (but the page size is such we didn't knew before) but it does has a static row. We should then skip
// the partition as returning it would means to the upper layer that the partition has "only" static columns,
// which is not the case (and we know the static results have been sent on the previous page).
if (isFirstPartition)
{
if (nextPartition == null)
nextPartition = super.next();
DecoratedKey key = nextPartition.partitionKey();
if (lastKey == null || !lastKey.equals(key))
remainingInPartition = limits.perPartitionCount();
lastKey = key;
// If this is the first partition of this page, this could be the continuation of a partition we've started
// on the previous page. In which case, we could have the problem that the partition has no more "regular"
// rows (but the page size is such we didn't knew before) but it does has a static row. We should then skip
// the partition as returning it would means to the upper layer that the partition has "only" static columns,
// which is not the case (and we know the static results have been sent on the previous page).
if (isFirstPartition && isPreviouslyReturnedPartition(key) && !nextPartition.hasNext())
{
nextPartition.close();
nextPartition = null;
}
isFirstPartition = false;
if (isPreviouslyReturnedPartition(key) && !partition.hasNext())
{
partition.close();
return null;
}
}
return nextPartition != null;
return Transformation.apply(counter.applyTo(partition), this);
}
@Override
@SuppressWarnings("resource") // iter is closed by closing the result
public RowIterator next()
public void onClose()
{
if (!hasNext())
throw new NoSuchElementException();
RowIterator toReturn = nextPartition;
nextPartition = null;
return new RowPagerIterator(toReturn);
}
@Override
public void close()
{
super.close();
if (nextPartition != null)
nextPartition.close();
recordLast(lastKey, lastRow);
int counted = counter.counted();
@ -159,28 +136,18 @@ abstract class AbstractQueryPager implements QueryPager
exhausted = counted < pageLimits.count();
}
private class RowPagerIterator extends WrappingRowIterator
public Row applyToStatic(Row row)
{
RowPagerIterator(RowIterator iter)
{
super(iter);
}
if (!row.isEmpty())
lastRow = row;
return row;
}
@Override
public Row staticRow()
{
Row staticRow = super.staticRow();
if (!staticRow.isEmpty())
lastRow = staticRow;
return staticRow;
}
@Override
public Row next()
{
lastRow = super.next();
return lastRow;
}
@Override
public Row applyToRow(Row row)
{
lastRow = row;
return row;
}
}

View File

@ -17,8 +17,6 @@
*/
package org.apache.cassandra.service.pager;
import java.util.List;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.db.*;
@ -125,9 +123,9 @@ public class MultiPartitionPager implements QueryPager
{
int toQuery = Math.min(remaining, pageSize);
PagersIterator iter = new PagersIterator(toQuery, consistency, clientState, null);
CountingPartitionIterator countingIter = new CountingPartitionIterator(iter, limit.forPaging(toQuery), nowInSec);
iter.setCounter(countingIter.counter());
return countingIter;
DataLimits.Counter counter = limit.forPaging(toQuery).newCounter(nowInSec, true);
iter.setCounter(counter);
return counter.applyTo(iter);
}
@SuppressWarnings("resource") // iter closed via countingIter
@ -135,9 +133,9 @@ public class MultiPartitionPager implements QueryPager
{
int toQuery = Math.min(remaining, pageSize);
PagersIterator iter = new PagersIterator(toQuery, null, null, executionController);
CountingPartitionIterator countingIter = new CountingPartitionIterator(iter, limit.forPaging(toQuery), nowInSec);
iter.setCounter(countingIter.counter());
return countingIter;
DataLimits.Counter counter = limit.forPaging(toQuery).newCounter(nowInSec, true);
iter.setCounter(counter);
return counter.applyTo(iter);
}
private class PagersIterator extends AbstractIterator<RowIterator> implements PartitionIterator

View File

@ -19,8 +19,8 @@ package org.apache.cassandra.service.pager;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.EmptyIterators;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionIterators;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.exceptions.RequestValidationException;
import org.apache.cassandra.service.ClientState;
@ -55,12 +55,12 @@ public interface QueryPager
public PartitionIterator fetchPage(int pageSize, ConsistencyLevel consistency, ClientState clientState) throws RequestValidationException, RequestExecutionException
{
return PartitionIterators.EMPTY;
return EmptyIterators.partition();
}
public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException
{
return PartitionIterators.EMPTY;
return EmptyIterators.partition();
}
public boolean isExhausted()

View File

@ -53,10 +53,11 @@ public class QueryPagers
int count = 0;
while (!pager.isExhausted())
{
try (CountingPartitionIterator iter = new CountingPartitionIterator(pager.fetchPage(pageSize, consistencyLevel, state), limits, nowInSec))
try (PartitionIterator iter = pager.fetchPage(pageSize, consistencyLevel, state))
{
PartitionIterators.consume(iter);
count += iter.counter().counted();
DataLimits.Counter counter = limits.newCounter(nowInSec, true);
PartitionIterators.consume(counter.applyTo(iter));
count += counter.counted();
}
}
return count;

View File

@ -21,6 +21,7 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.NoSuchElementException;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.utils.AbstractIterator;
import com.google.common.collect.Iterators;
import com.google.common.collect.PeekingIterator;
@ -59,13 +60,12 @@ import org.apache.cassandra.db.partitions.*;
* "c5": { value : 4 }
* "c7": { value : 1 }
*/
public class ThriftResultsMerger extends WrappingUnfilteredPartitionIterator
public class ThriftResultsMerger extends Transformation<UnfilteredRowIterator>
{
private final int nowInSec;
private ThriftResultsMerger(UnfilteredPartitionIterator wrapped, int nowInSec)
private ThriftResultsMerger(int nowInSec)
{
super(wrapped);
this.nowInSec = nowInSec;
}
@ -74,7 +74,7 @@ public class ThriftResultsMerger extends WrappingUnfilteredPartitionIterator
if (!metadata.isStaticCompactTable() && !metadata.isSuper())
return iterator;
return new ThriftResultsMerger(iterator, nowInSec);
return Transformation.apply(iterator, new ThriftResultsMerger(nowInSec));
}
public static UnfilteredRowIterator maybeWrap(UnfilteredRowIterator iterator, int nowInSec)
@ -83,14 +83,15 @@ public class ThriftResultsMerger extends WrappingUnfilteredPartitionIterator
return iterator;
return iterator.metadata().isSuper()
? new SuperColumnsPartitionMerger(iterator, nowInSec)
? Transformation.apply(iterator, new SuperColumnsPartitionMerger(iterator, nowInSec))
: new PartitionMerger(iterator, nowInSec);
}
protected UnfilteredRowIterator computeNext(UnfilteredRowIterator iter)
@Override
public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator iter)
{
return iter.metadata().isSuper()
? new SuperColumnsPartitionMerger(iter, nowInSec)
? Transformation.apply(iter, new SuperColumnsPartitionMerger(iter, nowInSec))
: new PartitionMerger(iter, nowInSec);
}
@ -204,20 +205,19 @@ public class ThriftResultsMerger extends WrappingUnfilteredPartitionIterator
}
}
private static class SuperColumnsPartitionMerger extends AlteringUnfilteredRowIterator
private static class SuperColumnsPartitionMerger extends Transformation
{
private final int nowInSec;
private final Row.Builder builder;
private final ColumnDefinition superColumnMapColumn;
private final AbstractType<?> columnComparator;
private SuperColumnsPartitionMerger(UnfilteredRowIterator results, int nowInSec)
private SuperColumnsPartitionMerger(UnfilteredRowIterator applyTo, int nowInSec)
{
super(results);
assert results.metadata().isSuper();
assert applyTo.metadata().isSuper();
this.nowInSec = nowInSec;
this.superColumnMapColumn = results.metadata().compactValueColumn();
this.superColumnMapColumn = applyTo.metadata().compactValueColumn();
assert superColumnMapColumn != null && superColumnMapColumn.type instanceof MapType;
this.builder = BTreeRow.sortedBuilder();
@ -225,7 +225,7 @@ public class ThriftResultsMerger extends WrappingUnfilteredPartitionIterator
}
@Override
protected Row computeNext(Row row)
public Row applyToRow(Row row)
{
PeekingIterator<Cell> staticCells = Iterators.peekingIterator(simpleCellsIterator(row));
if (!staticCells.hasNext())

View File

@ -21,6 +21,7 @@ import java.io.Closeable;
import java.util.Iterator;
// so we can instantiate anonymous classes implementing both interfaces
public interface CloseableIterator<T> extends Iterator<T>, AutoCloseable, Closeable
public interface CloseableIterator<T> extends Iterator<T>, AutoCloseable
{
public void close();
}

View File

@ -76,13 +76,18 @@ public final class Throwables
@SafeVarargs
public static <E extends Exception> void perform(DiscreteAction<? extends E> ... actions) throws E
{
perform(Arrays.stream(actions));
perform(Stream.of(actions));
}
public static <E extends Exception> void perform(Stream<? extends DiscreteAction<? extends E>> stream, DiscreteAction<? extends E> ... extra) throws E
{
perform(Stream.concat(stream, Stream.of(extra)));
}
@SuppressWarnings("unchecked")
public static <E extends Exception> void perform(Stream<DiscreteAction<? extends E>> actions) throws E
{
Throwable fail = perform(null, actions);
Throwable fail = perform((Throwable) null, actions);
if (failIfCanCast(fail, null))
throw (E) fail;
}

View File

@ -103,9 +103,9 @@ public class Util
return row.getCell(column);
}
public static ClusteringPrefix clustering(ClusteringComparator comparator, Object... o)
public static Clustering clustering(ClusteringComparator comparator, Object... o)
{
return comparator.make(o).clustering();
return comparator.make(o);
}
public static Token token(String key)

View File

@ -0,0 +1,325 @@
/*
* 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.ArrayList;
import java.util.List;
import org.junit.Test;
import junit.framework.Assert;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.transform.FilteredRows;
import org.apache.cassandra.db.transform.MoreRows;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.ByteBufferUtil;
public class TransformerTest
{
static final CFMetaData metadata = metadata();
static final DecoratedKey partitionKey = new BufferDecoratedKey(new Murmur3Partitioner.LongToken(0L), ByteBufferUtil.EMPTY_BYTE_BUFFER);
static final Row staticRow = BTreeRow.singleCellRow(Clustering.STATIC_CLUSTERING, new BufferCell(metadata.partitionColumns().columns(true).getSimple(0), 0L, 0, 0, ByteBufferUtil.bytes(-1), null));
static CFMetaData metadata()
{
CFMetaData.Builder builder = CFMetaData.Builder.create("", "");
builder.addPartitionKey("pk", BytesType.instance);
builder.addClusteringColumn("c", Int32Type.instance);
builder.addStaticColumn("s", Int32Type.instance);
builder.addRegularColumn("v", Int32Type.instance);
return builder.build();
}
// Mock Data
static abstract class AbstractBaseRowIterator<U extends Unfiltered> extends AbstractIterator<U> implements BaseRowIterator<U>
{
private final int i;
private boolean returned;
protected AbstractBaseRowIterator(int i)
{
this.i = i;
}
protected U computeNext()
{
if (returned)
return endOfData();
returned = true;
return (U) row(i);
}
public CFMetaData metadata()
{
return metadata;
}
public boolean isReverseOrder()
{
return false;
}
public PartitionColumns columns()
{
return metadata.partitionColumns();
}
public DecoratedKey partitionKey()
{
return partitionKey;
}
public Row staticRow()
{
return staticRow;
}
public boolean isEmpty()
{
return false;
}
public void close()
{
}
}
private static UnfilteredRowIterator unfiltered(int i)
{
class Iter extends AbstractBaseRowIterator<Unfiltered> implements UnfilteredRowIterator
{
protected Iter(int i)
{
super(i);
}
public DeletionTime partitionLevelDeletion()
{
return DeletionTime.LIVE;
}
public EncodingStats stats()
{
return EncodingStats.NO_STATS;
}
}
return new Iter(i);
}
private static RowIterator filtered(int i)
{
class Iter extends AbstractBaseRowIterator<Row> implements RowIterator
{
protected Iter(int i)
{
super(i);
}
}
return new Iter(i);
}
private static Row row(int i)
{
return BTreeRow.singleCellRow(Util.clustering(metadata.comparator, i),
new BufferCell(metadata.partitionColumns().columns(false).getSimple(0), 1L, BufferCell.NO_TTL, BufferCell.NO_DELETION_TIME, ByteBufferUtil.bytes(i), null));
}
// Transformations that check mock data ranges
private static Transformation expect(int from, int to, List<Check> checks)
{
Expect expect = new Expect(from, to);
checks.add(expect);
return expect;
}
abstract static class Check extends Transformation
{
public abstract void check();
}
static class Expect extends Check
{
final int from, to;
int cur;
boolean closed;
Expect(int from, int to)
{
this.from = from;
this.to = to;
this.cur = from;
}
public Row applyToRow(Row row)
{
Assert.assertEquals(cur++, ByteBufferUtil.toInt(row.clustering().get(0)));
return row;
}
public void onPartitionClose()
{
Assert.assertEquals(to, cur);
closed = true;
}
public void check()
{
Assert.assertTrue(closed);
}
}
// Combinations of mock data and checks for an empty, singleton, and extending (sequential) range
private static enum Filter
{
INIT, APPLY_INNER, APPLY_OUTER, NONE
}
private static BaseRowIterator<?> empty(Filter filter, List<Check> checks)
{
switch (filter)
{
case INIT:
return Transformation.apply(EmptyIterators.row(metadata, partitionKey, false), expect(0, 0, checks));
case APPLY_INNER:
return Transformation.apply(FilteredRows.filter(Transformation.apply(EmptyIterators.unfilteredRow(metadata, partitionKey, false), expect(0, 0, checks)), Integer.MAX_VALUE), expect(0, 0, checks));
case APPLY_OUTER:
case NONE:
return Transformation.apply(EmptyIterators.unfilteredRow(metadata, partitionKey, false), expect(0, 0, checks));
default:
throw new IllegalStateException();
}
}
private static BaseRowIterator<?> singleton(Filter filter, int i, List<Check> checks)
{
switch (filter)
{
case INIT:
return Transformation.apply(filtered(i), expect(i, i + 1, checks));
case APPLY_INNER:
return FilteredRows.filter(Transformation.apply(unfiltered(i), expect(i, i + 1, checks)), Integer.MAX_VALUE);
case APPLY_OUTER:
case NONE:
return Transformation.apply(unfiltered(i), expect(i, i + 1, checks));
default:
throw new IllegalStateException();
}
}
private static BaseRowIterator<?> extendingIterator(int count, Filter filter, List<Check> checks)
{
class RefillNested extends Expect implements MoreRows<BaseRowIterator<?>>
{
boolean returnedEmpty, returnedSingleton, returnedNested;
RefillNested(int from)
{
super(from, count);
}
public BaseRowIterator<?> moreContents()
{
// first call return an empty iterator,
// second call return a singleton iterator (with a function that expects to be around to receive just that item)
// third call return a nested version of ourselves, with a function that expects to receive all future values
// fourth call, return null, indicating no more iterators to return
if (!returnedEmpty)
{
returnedEmpty = true;
return empty(filter, checks);
}
if (!returnedSingleton)
{
returnedSingleton = true;
return singleton(filter, from, checks);
}
if (from + 1 >= to)
return null;
if (!returnedNested)
{
returnedNested = true;
RefillNested refill = new RefillNested(from + 1);
checks.add(refill);
return refill.applyTo(empty(filter, checks));
}
return null;
}
BaseRowIterator<?> applyTo(BaseRowIterator<?> iter)
{
if (iter instanceof UnfilteredRowIterator)
return Transformation.apply(MoreRows.extend((UnfilteredRowIterator) iter, this), this);
else
return Transformation.apply(MoreRows.extend((RowIterator) iter, this), this);
}
}
RefillNested refill = new RefillNested(0);
checks.add(refill);
BaseRowIterator<?> iter = empty(filter, checks);
switch (filter)
{
case APPLY_OUTER:
return FilteredRows.filter((UnfilteredRowIterator) refill.applyTo(iter), Integer.MAX_VALUE);
case APPLY_INNER:
case INIT:
case NONE:
return refill.applyTo(iter);
default:
throw new IllegalStateException();
}
}
@Test
public void testRowExtension()
{
for (Filter filter : Filter.values())
{
List<Check> checks = new ArrayList<>();
BaseRowIterator<?> iter = extendingIterator(5, filter, checks);
for (int i = 0 ; i < 5 ; i++)
{
Unfiltered u = iter.next();
assert u instanceof Row;
Assert.assertEquals(i, ByteBufferUtil.toInt(u.clustering().get(0)));
}
iter.close();
for (Check check : checks)
check.check();
}
}
}

View File

@ -20,8 +20,6 @@ package org.apache.cassandra.repair;
import java.net.InetAddress;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.junit.After;
@ -32,8 +30,8 @@ import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.BufferDecoratedKey;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.EmptyIterators;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
@ -45,7 +43,6 @@ import org.apache.cassandra.repair.messages.RepairMessage;
import org.apache.cassandra.repair.messages.ValidationComplete;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
@ -126,7 +123,7 @@ public class ValidatorTest
// add a row
Token mid = partitioner.midpoint(range.left, range.right);
validator.add(UnfilteredRowIterators.emptyIterator(cfs.metadata, new BufferDecoratedKey(mid, ByteBufferUtil.bytes("inconceivable!")), false));
validator.add(EmptyIterators.unfilteredRow(cfs.metadata, new BufferDecoratedKey(mid, ByteBufferUtil.bytes("inconceivable!")), false));
validator.complete();
// confirm that the tree was validated

View File

@ -327,7 +327,7 @@ public class DataResolverTest
.add("c2", "v2")
.buildUpdate())));
InetAddress peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, UnfilteredPartitionIterators.empty(cfm)));
resolver.preprocess(readResponseMessage(peer2, EmptyIterators.unfilteredPartition(cfm, false)));
try(PartitionIterator data = resolver.resolve();
RowIterator rows = Iterators.getOnlyElement(data))
@ -349,8 +349,8 @@ public class DataResolverTest
public void testResolveWithBothEmpty()
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2);
resolver.preprocess(readResponseMessage(peer(), UnfilteredPartitionIterators.empty(cfm)));
resolver.preprocess(readResponseMessage(peer(), UnfilteredPartitionIterators.empty(cfm)));
resolver.preprocess(readResponseMessage(peer(), EmptyIterators.unfilteredPartition(cfm, false)));
resolver.preprocess(readResponseMessage(peer(), EmptyIterators.unfilteredPartition(cfm, false)));
try(PartitionIterator data = resolver.resolve())
{