diff --git a/CHANGES.txt b/CHANGES.txt index 02f658422b..c28d3bd250 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/db/EmptyIterators.java b/src/java/org/apache/cassandra/db/EmptyIterators.java new file mode 100644 index 0000000000..6bf8fffdf8 --- /dev/null +++ b/src/java/org/apache/cassandra/db/EmptyIterators.java @@ -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> implements BasePartitionIterator + { + EmptyBasePartitionIterator() + { + } + + public void close() + { + } + + public boolean hasNext() + { + return false; + } + + public R next() + { + throw new NoSuchElementException(); + } + } + + private static class EmptyUnfilteredPartitionIterator extends EmptyBasePartitionIterator 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 implements PartitionIterator + { + public static final EmptyPartitionIterator instance = new EmptyPartitionIterator(); + private EmptyPartitionIterator() + { + super(); + } + } + + private static class EmptyBaseRowIterator implements BaseRowIterator + { + 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 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 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); + } +} diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java index 8fd53a7236..06ef64311e 100644 --- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java +++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java @@ -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 createMessage(int version) diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 1ed8bb4f97..f50a8cfc23 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -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 { 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> + { + 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()); } /** diff --git a/src/java/org/apache/cassandra/db/ReadQuery.java b/src/java/org/apache/cassandra/db/ReadQuery.java index 2b5c09c0f9..ba7b893fcd 100644 --- a/src/java/org/apache/cassandra/db/ReadQuery.java +++ b/src/java/org/apache/cassandra/db/ReadQuery.java @@ -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() diff --git a/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java index 1181485c6b..de572d6743 100644 --- a/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java +++ b/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java @@ -153,7 +153,7 @@ public class SinglePartitionNamesCommand extends SinglePartitionReadCommand> { - 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; } } } diff --git a/src/java/org/apache/cassandra/db/filter/RowFilter.java b/src/java/org/apache/cassandra/db/filter/RowFilter.java index 0ff30af348..09dc3427e8 100644 --- a/src/java/org/apache/cassandra/db/filter/RowFilter.java +++ b/src/java/org/apache/cassandra/db/filter/RowFilter.java @@ -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 if (expressions.isEmpty()) return iter; - return new AlteringUnfilteredPartitionIterator(iter) + class IsSatisfiedFilter extends Transformation { - 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 expressions) @@ -264,16 +265,17 @@ public abstract class RowFilter implements Iterable if (expressions.isEmpty()) return iter; - return new WrappingUnfilteredPartitionIterator(iter) + class IsSatisfiedThriftFilter extends Transformation { @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 // 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 expressions) diff --git a/src/java/org/apache/cassandra/db/partitions/AlteringUnfilteredPartitionIterator.java b/src/java/org/apache/cassandra/db/partitions/AlteringUnfilteredPartitionIterator.java deleted file mode 100644 index f7d722289a..0000000000 --- a/src/java/org/apache/cassandra/db/partitions/AlteringUnfilteredPartitionIterator.java +++ /dev/null @@ -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); - } - }; - } -} - diff --git a/src/java/org/apache/cassandra/db/partitions/BasePartitionIterator.java b/src/java/org/apache/cassandra/db/partitions/BasePartitionIterator.java new file mode 100644 index 0000000000..214f416d18 --- /dev/null +++ b/src/java/org/apache/cassandra/db/partitions/BasePartitionIterator.java @@ -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> extends CloseableIterator +{ + public void close(); +} diff --git a/src/java/org/apache/cassandra/db/partitions/CountingPartitionIterator.java b/src/java/org/apache/cassandra/db/partitions/CountingPartitionIterator.java deleted file mode 100644 index 16445e7839..0000000000 --- a/src/java/org/apache/cassandra/db/partitions/CountingPartitionIterator.java +++ /dev/null @@ -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); - } -} diff --git a/src/java/org/apache/cassandra/db/partitions/CountingRowIterator.java b/src/java/org/apache/cassandra/db/partitions/CountingRowIterator.java deleted file mode 100644 index 4ad321e939..0000000000 --- a/src/java/org/apache/cassandra/db/partitions/CountingRowIterator.java +++ /dev/null @@ -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(); - } -} diff --git a/src/java/org/apache/cassandra/db/partitions/CountingUnfilteredPartitionIterator.java b/src/java/org/apache/cassandra/db/partitions/CountingUnfilteredPartitionIterator.java deleted file mode 100644 index 52eedd4db4..0000000000 --- a/src/java/org/apache/cassandra/db/partitions/CountingUnfilteredPartitionIterator.java +++ /dev/null @@ -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); - } -} diff --git a/src/java/org/apache/cassandra/db/partitions/CountingUnfilteredRowIterator.java b/src/java/org/apache/cassandra/db/partitions/CountingUnfilteredRowIterator.java deleted file mode 100644 index e5d1e759fc..0000000000 --- a/src/java/org/apache/cassandra/db/partitions/CountingUnfilteredRowIterator.java +++ /dev/null @@ -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(); - } -} diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionIterator.java b/src/java/org/apache/cassandra/db/partitions/PartitionIterator.java index 36358fc69d..529a9e2c4a 100644 --- a/src/java/org/apache/cassandra/db/partitions/PartitionIterator.java +++ b/src/java/org/apache/cassandra/db/partitions/PartitionIterator.java @@ -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, AutoCloseable +public interface PartitionIterator extends BasePartitionIterator { - public void close(); } diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java b/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java index eeb6a4b846..0b43c19ee5 100644 --- a/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java +++ b/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java @@ -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 { - 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 { - 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 implements PartitionIterator diff --git a/src/java/org/apache/cassandra/db/partitions/PurgeFunction.java b/src/java/org/apache/cassandra/db/partitions/PurgeFunction.java new file mode 100644 index 0000000000..b7b01d6490 --- /dev/null +++ b/src/java/org/apache/cassandra/db/partitions/PurgeFunction.java @@ -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 +{ + 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; + } + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/db/partitions/PurgingPartitionIterator.java b/src/java/org/apache/cassandra/db/partitions/PurgingPartitionIterator.java deleted file mode 100644 index 2093f53139..0000000000 --- a/src/java/org/apache/cassandra/db/partitions/PurgingPartitionIterator.java +++ /dev/null @@ -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; - } - }; - } -}; diff --git a/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterator.java b/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterator.java index 10989dfdc1..201c9343b1 100644 --- a/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterator.java +++ b/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterator.java @@ -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, AutoCloseable +public interface UnfilteredPartitionIterator extends BasePartitionIterator { /** * Whether that partition iterator is for a thrift queries. @@ -44,6 +42,4 @@ public interface UnfilteredPartitionIterator extends Iterator 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 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 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 { - 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()); } /** diff --git a/src/java/org/apache/cassandra/db/partitions/WrappingPartitionIterator.java b/src/java/org/apache/cassandra/db/partitions/WrappingPartitionIterator.java deleted file mode 100644 index 4d4be70727..0000000000 --- a/src/java/org/apache/cassandra/db/partitions/WrappingPartitionIterator.java +++ /dev/null @@ -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(); - } -} diff --git a/src/java/org/apache/cassandra/db/partitions/WrappingUnfilteredPartitionIterator.java b/src/java/org/apache/cassandra/db/partitions/WrappingUnfilteredPartitionIterator.java deleted file mode 100644 index ebf3c28d72..0000000000 --- a/src/java/org/apache/cassandra/db/partitions/WrappingUnfilteredPartitionIterator.java +++ /dev/null @@ -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(); - } - } -} diff --git a/src/java/org/apache/cassandra/db/rows/AlteringUnfilteredRowIterator.java b/src/java/org/apache/cassandra/db/rows/AlteringUnfilteredRowIterator.java deleted file mode 100644 index a390badf21..0000000000 --- a/src/java/org/apache/cassandra/db/rows/AlteringUnfilteredRowIterator.java +++ /dev/null @@ -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; - } -} diff --git a/src/java/org/apache/cassandra/db/rows/BaseRowIterator.java b/src/java/org/apache/cassandra/db/rows/BaseRowIterator.java new file mode 100644 index 0000000000..fb9e9083ee --- /dev/null +++ b/src/java/org/apache/cassandra/db/rows/BaseRowIterator.java @@ -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 extends CloseableIterator +{ + /** + * 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(); +} diff --git a/src/java/org/apache/cassandra/db/rows/RowIterator.java b/src/java/org/apache/cassandra/db/rows/RowIterator.java index 69994ddafb..f0b4499a6d 100644 --- a/src/java/org/apache/cassandra/db/rows/RowIterator.java +++ b/src/java/org/apache/cassandra/db/rows/RowIterator.java @@ -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, AutoCloseable +public interface RowIterator extends BaseRowIterator { - /** - * 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. */ diff --git a/src/java/org/apache/cassandra/db/rows/RowIterators.java b/src/java/org/apache/cassandra/db/rows/RowIterators.java index 30f5c50cdc..551edb806c 100644 --- a/src/java/org/apache/cassandra/db/rows/RowIterators.java +++ b/src/java/org/apache/cassandra/db/rows/RowIterators.java @@ -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 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. *

@@ -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()); } } diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterator.java b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterator.java index 649fd8bf6c..a969858db2 100644 --- a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterator.java +++ b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterator.java @@ -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, AutoCloseable +public interface UnfilteredRowIterator extends BaseRowIterator { - /** - * 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, AutoCloseab */ public EncodingStats stats(); - public void close(); - /** * Returns whether this iterator has no data (including no deletion data). */ diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java index 3a0558e052..932ca4c4bc 100644 --- a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java +++ b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java @@ -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; diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java index 22628e2cd7..ea929d7cb3 100644 --- a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java +++ b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java @@ -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 { - 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 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(); - } - } } diff --git a/src/java/org/apache/cassandra/db/rows/WrappingRowIterator.java b/src/java/org/apache/cassandra/db/rows/WrappingRowIterator.java deleted file mode 100644 index 8847a4740f..0000000000 --- a/src/java/org/apache/cassandra/db/rows/WrappingRowIterator.java +++ /dev/null @@ -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 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(); - } -} diff --git a/src/java/org/apache/cassandra/db/rows/WrappingUnfilteredRowIterator.java b/src/java/org/apache/cassandra/db/rows/WrappingUnfilteredRowIterator.java index 84713ebeb6..8b1855424f 100644 --- a/src/java/org/apache/cassandra/db/rows/WrappingUnfilteredRowIterator.java +++ b/src/java/org/apache/cassandra/db/rows/WrappingUnfilteredRowIterator.java @@ -29,7 +29,7 @@ import org.apache.cassandra.db.*; * some of the methods. *

* 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 implements UnfilteredRowIterator { diff --git a/src/java/org/apache/cassandra/db/transform/BaseIterator.java b/src/java/org/apache/cassandra/db/transform/BaseIterator.java new file mode 100644 index 0000000000..9b95dfa9cc --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/BaseIterator.java @@ -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, O extends V> extends Stack implements AutoCloseable, Iterator +{ + 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 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; + } +} + diff --git a/src/java/org/apache/cassandra/db/transform/BasePartitions.java b/src/java/org/apache/cassandra/db/transform/BasePartitions.java new file mode 100644 index 0000000000..e795760a40 --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/BasePartitions.java @@ -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, I extends BasePartitionIterator>> +extends BaseIterator, I, R> +implements BasePartitionIterator +{ + + public BasePartitions(I input) + { + super(input); + } + + BasePartitions(BasePartitions 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; + } + } + +} + diff --git a/src/java/org/apache/cassandra/db/transform/BaseRows.java b/src/java/org/apache/cassandra/db/transform/BaseRows.java new file mode 100644 index 0000000000..78526e86ae --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/BaseRows.java @@ -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> +extends BaseIterator +implements BaseRowIterator +{ + + private Row staticRow; + + public BaseRows(I input) + { + super(input); + staticRow = input.staticRow(); + } + + // swap parameter order to avoid casting errors + BaseRows(BaseRows 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; + } +} diff --git a/src/java/org/apache/cassandra/db/transform/Filter.java b/src/java/org/apache/cassandra/db/transform/Filter.java new file mode 100644 index 0000000000..3bf831f03c --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/Filter.java @@ -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; + } +} diff --git a/src/java/org/apache/cassandra/db/transform/FilteredPartitions.java b/src/java/org/apache/cassandra/db/transform/FilteredPartitions.java new file mode 100644 index 0000000000..5a802dcd1e --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/FilteredPartitions.java @@ -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> 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); + } +} diff --git a/src/java/org/apache/cassandra/db/transform/FilteredRows.java b/src/java/org/apache/cassandra/db/transform/FilteredRows.java new file mode 100644 index 0000000000..b21b451834 --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/FilteredRows.java @@ -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> 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); + } +} diff --git a/src/java/org/apache/cassandra/db/transform/MoreContents.java b/src/java/org/apache/cassandra/db/transform/MoreContents.java new file mode 100644 index 0000000000..7e392ca827 --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/MoreContents.java @@ -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 +{ + public abstract I moreContents(); +} + diff --git a/src/java/org/apache/cassandra/db/transform/MorePartitions.java b/src/java/org/apache/cassandra/db/transform/MorePartitions.java new file mode 100644 index 0000000000..5cfcc4c304 --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/MorePartitions.java @@ -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> extends MoreContents +{ + + public static UnfilteredPartitionIterator extend(UnfilteredPartitionIterator iterator, MorePartitions more) + { + return add(mutable(iterator), more); + } + + public static PartitionIterator extend(PartitionIterator iterator, MorePartitions more) + { + return add(mutable(iterator), more); + } + +} + diff --git a/src/java/org/apache/cassandra/db/transform/MoreRows.java b/src/java/org/apache/cassandra/db/transform/MoreRows.java new file mode 100644 index 0000000000..f406a490d0 --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/MoreRows.java @@ -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> extends MoreContents +{ + + public static UnfilteredRowIterator extend(UnfilteredRowIterator iterator, MoreRows more) + { + return add(mutable(iterator), more); + } + + public static RowIterator extend(RowIterator iterator, MoreRows more) + { + return add(mutable(iterator), more); + } + +} + diff --git a/src/java/org/apache/cassandra/db/transform/Stack.java b/src/java/org/apache/cassandra/db/transform/Stack.java new file mode 100644 index 0000000000..aac1679b4e --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/Stack.java @@ -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[] 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[] 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; + } +} + diff --git a/src/java/org/apache/cassandra/db/transform/StoppingTransformation.java b/src/java/org/apache/cassandra/db/transform/StoppingTransformation.java new file mode 100644 index 0000000000..f3afdc0e14 --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/StoppingTransformation.java @@ -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> extends Transformation +{ + 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; + } +} diff --git a/src/java/org/apache/cassandra/db/transform/Transformation.java b/src/java/org/apache/cassandra/db/transform/Transformation.java new file mode 100644 index 0000000000..29e2e15d14 --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/Transformation.java @@ -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> +{ + // 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 transformation) + { + return add(mutable(iterator), transformation); + } + public static PartitionIterator apply(PartitionIterator iterator, Transformation 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 add(E to, Transformation add) + { + to.add(add); + return to; + } + static E add(E to, MoreContents add) + { + to.add(add); + return to; + } + +} diff --git a/src/java/org/apache/cassandra/db/transform/UnfilteredPartitions.java b/src/java/org/apache/cassandra/db/transform/UnfilteredPartitions.java new file mode 100644 index 0000000000..4e405455a3 --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/UnfilteredPartitions.java @@ -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 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(); + } +} diff --git a/src/java/org/apache/cassandra/db/transform/UnfilteredRows.java b/src/java/org/apache/cassandra/db/transform/UnfilteredRows.java new file mode 100644 index 0000000000..98640ae0af --- /dev/null +++ b/src/java/org/apache/cassandra/db/transform/UnfilteredRows.java @@ -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 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(); + } +} diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexBuilder.java b/src/java/org/apache/cassandra/index/SecondaryIndexBuilder.java index d6ae8e2b4a..e66f0a30a9 100644 --- a/src/java/org/apache/cassandra/index/SecondaryIndexBuilder.java +++ b/src/java/org/apache/cassandra/index/SecondaryIndexBuilder.java @@ -71,14 +71,7 @@ public class SecondaryIndexBuilder extends CompactionInfo.Holder } finally { - try - { - iter.close(); - } - catch (IOException e) - { - throw new RuntimeException(e); - } + iter.close(); } } } diff --git a/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java b/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java index d77b889215..7303cbe2f9 100644 --- a/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java +++ b/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java @@ -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()); } } diff --git a/src/java/org/apache/cassandra/io/sstable/ReducingKeyIterator.java b/src/java/org/apache/cassandra/io/sstable/ReducingKeyIterator.java index bbc56cc541..6f395f8cb4 100644 --- a/src/java/org/apache/cassandra/io/sstable/ReducingKeyIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/ReducingKeyIterator.java @@ -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 } } - public void close() throws IOException + public void close() { if (mi != null) - { mi.close(); - } } public long getTotalBytes() diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java index e02c91977f..b6077e098a 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java @@ -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; } } diff --git a/src/java/org/apache/cassandra/service/DataResolver.java b/src/java/org/apache/cassandra/service/DataResolver.java index f24c29f5d9..2de02f6401 100644 --- a/src/java/org/apache/cassandra/service/DataResolver.java +++ b/src/java/org/apache/cassandra/service/DataResolver.java @@ -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 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 { 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 { - 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); diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 82e2b6e182..424909e629 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -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 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() diff --git a/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java b/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java index de4e54bc59..386d7aeab9 100644 --- a/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java +++ b/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java @@ -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 { 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; } } diff --git a/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java b/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java index fca0165655..922df2ede5 100644 --- a/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java +++ b/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java @@ -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 implements PartitionIterator diff --git a/src/java/org/apache/cassandra/service/pager/QueryPager.java b/src/java/org/apache/cassandra/service/pager/QueryPager.java index 1d5a739e07..e2d7f5eb89 100644 --- a/src/java/org/apache/cassandra/service/pager/QueryPager.java +++ b/src/java/org/apache/cassandra/service/pager/QueryPager.java @@ -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() diff --git a/src/java/org/apache/cassandra/service/pager/QueryPagers.java b/src/java/org/apache/cassandra/service/pager/QueryPagers.java index eee94e604c..02b5de2679 100644 --- a/src/java/org/apache/cassandra/service/pager/QueryPagers.java +++ b/src/java/org/apache/cassandra/service/pager/QueryPagers.java @@ -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; diff --git a/src/java/org/apache/cassandra/thrift/ThriftResultsMerger.java b/src/java/org/apache/cassandra/thrift/ThriftResultsMerger.java index 72e4399a81..14c0dca4d7 100644 --- a/src/java/org/apache/cassandra/thrift/ThriftResultsMerger.java +++ b/src/java/org/apache/cassandra/thrift/ThriftResultsMerger.java @@ -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 { 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 staticCells = Iterators.peekingIterator(simpleCellsIterator(row)); if (!staticCells.hasNext()) diff --git a/src/java/org/apache/cassandra/utils/CloseableIterator.java b/src/java/org/apache/cassandra/utils/CloseableIterator.java index 7474f3dc11..a7c4300dbd 100644 --- a/src/java/org/apache/cassandra/utils/CloseableIterator.java +++ b/src/java/org/apache/cassandra/utils/CloseableIterator.java @@ -21,6 +21,7 @@ import java.io.Closeable; import java.util.Iterator; // so we can instantiate anonymous classes implementing both interfaces -public interface CloseableIterator extends Iterator, AutoCloseable, Closeable +public interface CloseableIterator extends Iterator, AutoCloseable { + public void close(); } diff --git a/src/java/org/apache/cassandra/utils/Throwables.java b/src/java/org/apache/cassandra/utils/Throwables.java index 923b7238a5..8ef6a633c8 100644 --- a/src/java/org/apache/cassandra/utils/Throwables.java +++ b/src/java/org/apache/cassandra/utils/Throwables.java @@ -76,13 +76,18 @@ public final class Throwables @SafeVarargs public static void perform(DiscreteAction ... actions) throws E { - perform(Arrays.stream(actions)); + perform(Stream.of(actions)); + } + + public static void perform(Stream> stream, DiscreteAction ... extra) throws E + { + perform(Stream.concat(stream, Stream.of(extra))); } @SuppressWarnings("unchecked") public static void perform(Stream> actions) throws E { - Throwable fail = perform(null, actions); + Throwable fail = perform((Throwable) null, actions); if (failIfCanCast(fail, null)) throw (E) fail; } diff --git a/test/data/legacy-sstables/jb/Keyspace1/Keyspace1-Standard1-jb-0-Summary.db b/test/data/legacy-sstables/jb/Keyspace1/Keyspace1-Standard1-jb-0-Summary.db index 83c68cecf1..1fbe0405e0 100644 Binary files a/test/data/legacy-sstables/jb/Keyspace1/Keyspace1-Standard1-jb-0-Summary.db and b/test/data/legacy-sstables/jb/Keyspace1/Keyspace1-Standard1-jb-0-Summary.db differ diff --git a/test/unit/org/apache/cassandra/Util.java b/test/unit/org/apache/cassandra/Util.java index 4e122ed36e..9162ed951c 100644 --- a/test/unit/org/apache/cassandra/Util.java +++ b/test/unit/org/apache/cassandra/Util.java @@ -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) diff --git a/test/unit/org/apache/cassandra/db/TransformerTest.java b/test/unit/org/apache/cassandra/db/TransformerTest.java new file mode 100644 index 0000000000..d56d8cd345 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/TransformerTest.java @@ -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 extends AbstractIterator implements BaseRowIterator + { + 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 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 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 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 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 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 checks) + { + class RefillNested extends Expect implements MoreRows> + { + 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 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(); + } + } +} diff --git a/test/unit/org/apache/cassandra/repair/ValidatorTest.java b/test/unit/org/apache/cassandra/repair/ValidatorTest.java index 8fe76c3322..14f5707710 100644 --- a/test/unit/org/apache/cassandra/repair/ValidatorTest.java +++ b/test/unit/org/apache/cassandra/repair/ValidatorTest.java @@ -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 diff --git a/test/unit/org/apache/cassandra/service/DataResolverTest.java b/test/unit/org/apache/cassandra/service/DataResolverTest.java index 6048b9cc5b..b94db671ff 100644 --- a/test/unit/org/apache/cassandra/service/DataResolverTest.java +++ b/test/unit/org/apache/cassandra/service/DataResolverTest.java @@ -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()) {