You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2024/11/08 20:42:31 UTC
(cassandra) branch cassandra-5.0 updated: Batch clusterings into single SAI partition post-filtering reads
This is an automated email from the ASF dual-hosted git repository.
maedhroz pushed a commit to branch cassandra-5.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-5.0 by this push:
new cccb9821b4 Batch clusterings into single SAI partition post-filtering reads
cccb9821b4 is described below
commit cccb9821b4e4558e16b2c4577c398f8864612687
Author: Caleb Rackliffe <ca...@gmail.com>
AuthorDate: Fri Nov 1 16:40:49 2024 -0500
Batch clusterings into single SAI partition post-filtering reads
patch by Caleb Rackliffe; reviewed by Zhao Yang for CASSANDRA-19497
---
CHANGES.txt | 1 +
.../cassandra/index/sai/VectorQueryContext.java | 4 +-
.../cassandra/index/sai/plan/QueryController.java | 41 ++-
.../sai/plan/StorageAttachedIndexSearcher.java | 198 +++++++++-----
.../cassandra/index/sai/utils/PrimaryKey.java | 5 +
.../utils/InsertionOrderedNavigableSet.java | 297 +++++++++++++++++++++
6 files changed, 456 insertions(+), 90 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 56128a4663..36e021cf56 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.0.3
+ * Batch clusterings into single SAI partition post-filtering reads (CASSANDRA-19497)
* Ban the usage of "var" instead of full types in the production code (CASSANDRA-20038)
* Suppress CVE-2024-45772 from lucene-core-9.7.0.jar (CASSANDRA-20024)
Merged from 4.1:
diff --git a/src/java/org/apache/cassandra/index/sai/VectorQueryContext.java b/src/java/org/apache/cassandra/index/sai/VectorQueryContext.java
index d65723b79b..a0533a0fe8 100644
--- a/src/java/org/apache/cassandra/index/sai/VectorQueryContext.java
+++ b/src/java/org/apache/cassandra/index/sai/VectorQueryContext.java
@@ -64,11 +64,11 @@ public class VectorQueryContext
return limit;
}
- public void recordShadowedPrimaryKey(PrimaryKey primaryKey)
+ public void recordShadowedPrimaryKeys(Set<PrimaryKey> keys)
{
if (shadowedPrimaryKeys == null)
shadowedPrimaryKeys = new TreeSet<>();
- shadowedPrimaryKeys.add(primaryKey);
+ shadowedPrimaryKeys.addAll(keys);
}
// Returns true if the row ID will be included or false if the row ID will be shadowed
diff --git a/src/java/org/apache/cassandra/index/sai/plan/QueryController.java b/src/java/org/apache/cassandra/index/sai/plan/QueryController.java
index 91adc5d17e..b462fa3ad5 100644
--- a/src/java/org/apache/cassandra/index/sai/plan/QueryController.java
+++ b/src/java/org/apache/cassandra/index/sai/plan/QueryController.java
@@ -24,6 +24,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.NavigableSet;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
@@ -31,6 +32,7 @@ import javax.annotation.Nullable;
import com.google.common.collect.Lists;
import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
@@ -62,7 +64,7 @@ import org.apache.cassandra.index.sai.utils.PrimaryKey;
import org.apache.cassandra.net.ParamType;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.tracing.Tracing;
-import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.InsertionOrderedNavigableSet;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.Throwables;
@@ -82,6 +84,8 @@ public class QueryController
private final PrimaryKey lastPrimaryKey;
private final int orderChunkSize;
+ private final NavigableSet<Clustering<?>> nextClusterings;
+
public QueryController(ColumnFamilyStore cfs,
ReadCommand command,
RowFilter indexFilter,
@@ -99,6 +103,7 @@ public class QueryController
this.firstPrimaryKey = keyFactory.create(mergeRange.left.getToken());
this.lastPrimaryKey = keyFactory.create(mergeRange.right.getToken());
this.orderChunkSize = SAI_VECTOR_SEARCH_ORDER_CHUNK_SIZE.getInt();
+ this.nextClusterings = new InsertionOrderedNavigableSet<>(cfs.metadata().comparator);
}
public PrimaryKey.Factory primaryKeyFactory()
@@ -151,18 +156,18 @@ public class QueryController
return index != null && index.hasAnalyzer();
}
- public UnfilteredRowIterator queryStorage(PrimaryKey key, ReadExecutionController executionController)
+ public UnfilteredRowIterator queryStorage(List<PrimaryKey> keys, ReadExecutionController executionController)
{
- if (key == null)
- throw new IllegalArgumentException("non-null key required");
+ if (keys.isEmpty())
+ throw new IllegalArgumentException("At least one primary key is required!");
SinglePartitionReadCommand partition = SinglePartitionReadCommand.create(cfs.metadata(),
command.nowInSec(),
command.columnFilter(),
RowFilter.none(),
DataLimits.NONE,
- key.partitionKey(),
- makeFilter(key));
+ keys.get(0).partitionKey(),
+ makeFilter(keys));
return partition.queryMemtableAndDisk(cfs, executionController);
}
@@ -409,21 +414,29 @@ public class QueryController
// Note: This method assumes that the selects method has already been called for the
// key to avoid having to (potentially) call selects twice
- private ClusteringIndexFilter makeFilter(PrimaryKey key)
+ private ClusteringIndexFilter makeFilter(List<PrimaryKey> keys)
{
- ClusteringIndexFilter clusteringIndexFilter = command.clusteringIndexFilter(key.partitionKey());
+ PrimaryKey firstKey = keys.get(0);
- assert cfs.metadata().comparator.size() == 0 && !key.kind().hasClustering ||
- cfs.metadata().comparator.size() > 0 && key.kind().hasClustering :
- "PrimaryKey " + key + " clustering does not match table. There should be a clustering of size " + cfs.metadata().comparator.size();
+ assert cfs.metadata().comparator.size() == 0 && !firstKey.kind().hasClustering ||
+ cfs.metadata().comparator.size() > 0 && firstKey.kind().hasClustering :
+ "PrimaryKey " + firstKey + " clustering does not match table. There should be a clustering of size " + cfs.metadata().comparator.size();
+ ClusteringIndexFilter clusteringIndexFilter = command.clusteringIndexFilter(firstKey.partitionKey());
+
// If we have skinny partitions or the key is for a static row then we need to get the partition as
// requested by the original query.
- if (cfs.metadata().comparator.size() == 0 || key.kind() == PrimaryKey.Kind.STATIC)
+ if (cfs.metadata().comparator.size() == 0 || firstKey.kind() == PrimaryKey.Kind.STATIC)
+ {
return clusteringIndexFilter;
+ }
else
- return new ClusteringIndexNamesFilter(FBUtilities.singleton(key.clustering(), cfs.metadata().comparator),
- clusteringIndexFilter.isReversed());
+ {
+ nextClusterings.clear();
+ for (PrimaryKey key : keys)
+ nextClusterings.add(key.clustering());
+ return new ClusteringIndexNamesFilter(nextClusterings, clusteringIndexFilter.isReversed());
+ }
}
/**
diff --git a/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java b/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java
index 279f0d82e2..b235c251ef 100644
--- a/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java
+++ b/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java
@@ -19,14 +19,19 @@
package org.apache.cassandra.index.sai.plan;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import io.netty.util.concurrent.FastThreadLocal;
+import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
@@ -57,11 +62,22 @@ import org.apache.cassandra.utils.Clock;
public class StorageAttachedIndexSearcher implements Index.Searcher
{
+ private static final int PARTITION_ROW_BATCH_SIZE = 100;
+
private final ReadCommand command;
private final QueryController queryController;
private final QueryContext queryContext;
private final TableQueryMetrics tableQueryMetrics;
+ private static final FastThreadLocal<List<PrimaryKey>> nextKeys = new FastThreadLocal<>()
+ {
+ @Override
+ protected List<PrimaryKey> initialValue()
+ {
+ return new ArrayList<>(PARTITION_ROW_BATCH_SIZE);
+ }
+ };
+
public StorageAttachedIndexSearcher(ColumnFamilyStore cfs,
TableQueryMetrics tableQueryMetrics,
ReadCommand command,
@@ -129,11 +145,11 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
private final ReadExecutionController executionController;
private final PrimaryKey.Factory keyFactory;
private final boolean topK;
+ private final int partitionRowBatchSize;
private PrimaryKey lastKey;
- private ResultRetriever(ReadExecutionController executionController,
- boolean topK)
+ private ResultRetriever(ReadExecutionController executionController, boolean topK)
{
this.keyRanges = queryController.dataRanges().iterator();
this.currentKeyRange = keyRanges.next().keyRange();
@@ -144,6 +160,9 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
this.firstPrimaryKey = queryController.firstPrimaryKeyInRange();
this.lastPrimaryKey = queryController.lastPrimaryKeyInRange();
this.topK = topK;
+
+ // Ensure we don't fetch larger batches than the provided LIMIT to avoid fetching keys we won't use:
+ this.partitionRowBatchSize = Math.min(PARTITION_ROW_BATCH_SIZE, command.limits().count());
}
@Override
@@ -164,33 +183,75 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
// saying this iterator must not return the same partition twice.
skipToNextPartition();
- UnfilteredRowIterator iterator = nextRowIterator(this::nextSelectedKeyInRange);
- return iterator != null
- ? iteratePartition(iterator)
- : endOfData();
+ UnfilteredRowIterator iterator = nextRowIterator(this::nextSelectedKeysInRange);
+ return iterator != null ? iteratePartition(iterator) : endOfData();
}
/**
- * Tries to obtain a row iterator for one of the supplied keys by repeatedly calling
+ * Tries to obtain a row iterator for the supplied keys by repeatedly calling
* {@link ResultRetriever#queryStorageAndFilter} until it gives a non-null result.
- * The keySupplier should return the next key with every call to get() and
- * null when there are no more keys to try.
+ * The keysSupplier should return the next batch of keys with every call to get()
+ * and null when there are no more keys to try.
*
* @return an iterator or null if all keys were tried with no success
*/
- private @Nullable UnfilteredRowIterator nextRowIterator(@Nonnull Supplier<PrimaryKey> keySupplier)
+ private @Nullable UnfilteredRowIterator nextRowIterator(@Nonnull Supplier<List<PrimaryKey>> keysSupplier)
{
UnfilteredRowIterator iterator = null;
while (iterator == null)
{
- PrimaryKey key = keySupplier.get();
- if (key == null)
+ List<PrimaryKey> keys = keysSupplier.get();
+ if (keys.isEmpty())
return null;
- iterator = queryStorageAndFilter(key);
+ iterator = queryStorageAndFilter(keys);
}
return iterator;
}
+ /**
+ * Retrieves the next batch of primary keys (i.e. up to {@link #partitionRowBatchSize} of them) that are
+ * contained by one of the query key ranges and selected by the {@link QueryController}. If the next key falls
+ * out of the current key range, it skips to the next key range, and so on. If no more keys accepted by
+ * the controller are available, and empty list is returned.
+ *
+ * @return a list of up to {@link #partitionRowBatchSize} primary keys
+ */
+ private List<PrimaryKey> nextSelectedKeysInRange()
+ {
+ List<PrimaryKey> threadLocalNextKeys = nextKeys.get();
+ threadLocalNextKeys.clear();
+ PrimaryKey firstKey;
+
+ do
+ {
+ firstKey = nextKeyInRange();
+
+ if (firstKey == null)
+ return Collections.emptyList();
+ }
+ while (queryController.doesNotSelect(firstKey) || firstKey.equals(lastKey));
+
+ lastKey = firstKey;
+ threadLocalNextKeys.add(firstKey);
+ fillNextSelectedKeysInPartition(firstKey.partitionKey(), threadLocalNextKeys);
+ return threadLocalNextKeys;
+ }
+
+ /**
+ * Retrieves the next batch of primary keys (i.e. up to {@link #partitionRowBatchSize} of them) that belong to
+ * the given partition and are selected by the query controller, advancing the underlying iterator only while
+ * the next key belongs to that partition.
+ *
+ * @return a list of up to {@link #partitionRowBatchSize} primary keys within the given partition
+ */
+ private List<PrimaryKey> nextSelectedKeysInPartition(DecoratedKey partitionKey)
+ {
+ List<PrimaryKey> threadLocalNextKeys = nextKeys.get();
+ threadLocalNextKeys.clear();
+ fillNextSelectedKeysInPartition(partitionKey, threadLocalNextKeys);
+ return threadLocalNextKeys;
+ }
+
/**
* Returns the next available key contained by one of the keyRanges.
* If the next key falls out of the current key range, it skips to the next key range, and so on.
@@ -219,48 +280,23 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
return key;
}
- /**
- * Returns the next available key contained by one of the keyRanges and selected by the queryController.
- * If the next key falls out of the current key range, it skips to the next key range, and so on.
- * If no more keys acceptd by the controller are available, returns null.
- */
- private @Nullable PrimaryKey nextSelectedKeyInRange()
+ private void fillNextSelectedKeysInPartition(DecoratedKey partitionKey, List<PrimaryKey> nextPrimaryKeys)
{
- PrimaryKey key;
- do
+ while (resultKeyIterator.hasNext()
+ && resultKeyIterator.peek().partitionKey().equals(partitionKey)
+ && nextPrimaryKeys.size() < partitionRowBatchSize)
{
- key = nextKeyInRange();
- }
- while (key != null && queryController.doesNotSelect(key));
- return key;
- }
+ PrimaryKey key = nextKey();
- /**
- * Retrieves the next primary key that belongs to the given partition and is selected by the query controller.
- * The underlying key iterator is advanced only if the key belongs to the same partition.
- * <p>
- * Returns null if:
- * <ul>
- * <li>there are no more keys</li>
- * <li>the next key is beyond the upper bound</li>
- * <li>the next key belongs to a different partition</li>
- * </ul>
- * </p>
- */
- private @Nullable PrimaryKey nextSelectedKeyInPartition(DecoratedKey partitionKey)
- {
- PrimaryKey key;
- do
- {
- if (!resultKeyIterator.hasNext())
- return null;
- if (!resultKeyIterator.peek().partitionKey().equals(partitionKey))
- return null;
+ if (key == null)
+ break;
+
+ if (queryController.doesNotSelect(key) || key.equals(lastKey))
+ continue;
- key = nextKey();
+ nextPrimaryKeys.add(key);
+ lastKey = key;
}
- while (key != null && queryController.doesNotSelect(key));
- return key;
}
/**
@@ -315,8 +351,8 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
/**
* Returns an iterator over the rows in the partition associated with the given iterator.
* Initially, it retrieves the rows from the given iterator until it runs out of data.
- * Then it iterates the primary keys obtained from the index until the end of the partition
- * and lazily constructs new row itertors for each of the keys. At a given time, only one row iterator is open.
+ * Then it iterates the remaining primary keys obtained from the index in batches until the end of the
+ * partition, lazily constructing an itertor for each batch. Only one row iterator is open at a time.
* <p>
* The rows are retrieved in the order of primary keys provided by the underlying index.
* The iterator is complete when the next key to be fetched belongs to different partition
@@ -343,7 +379,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
while (!currentIter.hasNext())
{
currentIter.close();
- currentIter = nextRowIterator(() -> nextSelectedKeyInPartition(partitionKey));
+ currentIter = nextRowIterator(() -> nextSelectedKeysInPartition(partitionKey));
if (currentIter == null)
return endOfData();
}
@@ -359,21 +395,16 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
};
}
- private UnfilteredRowIterator queryStorageAndFilter(PrimaryKey key)
+ private UnfilteredRowIterator queryStorageAndFilter(List<PrimaryKey> keys)
{
- // Key reads are lazy, delayed all the way to this point. Skip if we've already seen this one:
- if (key.equals(lastKey))
- return null;
-
- lastKey = key;
long startTimeNanos = Clock.Global.nanoTime();
- try (UnfilteredRowIterator partition = queryController.queryStorage(key, executionController))
+ try (UnfilteredRowIterator partition = queryController.queryStorage(keys, executionController))
{
queryContext.partitionsRead++;
queryContext.checkpoint();
- UnfilteredRowIterator filtered = applyIndexFilter(key, partition, filterTree);
+ UnfilteredRowIterator filtered = filterPartition(keys, partition, filterTree);
// Note that we record the duration of the read after post-filtering, which actually
// materializes the rows from disk.
@@ -383,15 +414,14 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
}
}
- private UnfilteredRowIterator applyIndexFilter(PrimaryKey key, UnfilteredRowIterator partition, FilterTree tree)
+ private UnfilteredRowIterator filterPartition(List<PrimaryKey> keys, UnfilteredRowIterator partition, FilterTree tree)
{
Row staticRow = partition.staticRow();
- List<Unfiltered> matchingRows = new ArrayList<>();
+ DecoratedKey partitionKey = partition.partitionKey();
+ List<Unfiltered> matches = new ArrayList<>();
boolean hasMatch = false;
+ Set<PrimaryKey> keysToShadow = topK ? new HashSet<>(keys) : Collections.emptySet();
- // We need to filter the partition rows before filtering on the static row. If this is done in the other
- // order then we get incorrect results if we are filtering on a partition key index on a table with a
- // composite partition key.
while (partition.hasNext())
{
Unfiltered unfiltered = partition.next();
@@ -400,28 +430,48 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
{
queryContext.rowsFiltered++;
- if (tree.isSatisfiedBy(partition.partitionKey(), (Row) unfiltered, staticRow))
+ if (tree.isSatisfiedBy(partitionKey, (Row) unfiltered, staticRow))
{
- matchingRows.add(unfiltered);
+ matches.add(unfiltered);
hasMatch = true;
+
+ if (topK)
+ {
+ PrimaryKey shadowed = keyFactory.hasClusteringColumns()
+ ? keyFactory.create(partitionKey, ((Row) unfiltered).clustering())
+ : keyFactory.create(partitionKey);
+ keysToShadow.remove(shadowed);
+ }
}
}
}
+ // If any non-static rows match the filter, there should be no need to shadow the static primary key:
+ if (topK && hasMatch && keyFactory.hasClusteringColumns())
+ keysToShadow.remove(keyFactory.create(partitionKey, Clustering.STATIC_CLUSTERING));
+
+ // We may not have any non-static row data to filter...
if (!hasMatch)
{
queryContext.rowsFiltered++;
- if (tree.isSatisfiedBy(key.partitionKey(), staticRow, staticRow))
+ if (tree.isSatisfiedBy(partitionKey, staticRow, staticRow))
+ {
hasMatch = true;
+
+ if (topK)
+ keysToShadow.clear();
+ }
}
- if (!hasMatch)
+ if (topK && !keysToShadow.isEmpty())
{
- // shadowed by expired TTL or row tombstone or range tombstone
- if (topK)
- queryContext.vectorContext().recordShadowedPrimaryKey(key);
+ // Record primary keys shadowed by expired TTLs, row tombstones, or range tombstones:
+ queryContext.vectorContext().recordShadowedPrimaryKeys(keysToShadow);
+ }
+ if (!hasMatch)
+ {
// If there are no matches, return an empty partition. If reconciliation is required at the
// coordinator, replica filtering protection may make a second round trip to complete its view
// of the partition.
@@ -429,7 +479,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
}
// Return all matches found, along with the static row...
- return new PartitionIterator(partition, staticRow, matchingRows.iterator());
+ return new PartitionIterator(partition, staticRow, matches.iterator());
}
private class PartitionIterator extends AbstractUnfilteredRowIterator
diff --git a/src/java/org/apache/cassandra/index/sai/utils/PrimaryKey.java b/src/java/org/apache/cassandra/index/sai/utils/PrimaryKey.java
index c079a66c4c..4e5081ad0c 100644
--- a/src/java/org/apache/cassandra/index/sai/utils/PrimaryKey.java
+++ b/src/java/org/apache/cassandra/index/sai/utils/PrimaryKey.java
@@ -120,6 +120,11 @@ public interface PrimaryKey extends Comparable<PrimaryKey>, ByteComparable
return clustering == Clustering.STATIC_CLUSTERING ? new StaticPrimaryKey(partitionKey) : new WidePrimaryKey(partitionKey, clustering);
}
+ public boolean hasClusteringColumns()
+ {
+ return clusteringComparator != null && clusteringComparator.size() > 0;
+ }
+
/**
* Create a {@link PrimaryKey} from a {@link ByteSource}. This should only be used with {@link ByteSource} instances
* created by calls to {@link PrimaryKey#asComparableBytes(Version)}.
diff --git a/src/java/org/apache/cassandra/utils/InsertionOrderedNavigableSet.java b/src/java/org/apache/cassandra/utils/InsertionOrderedNavigableSet.java
new file mode 100644
index 0000000000..7d9d841046
--- /dev/null
+++ b/src/java/org/apache/cassandra/utils/InsertionOrderedNavigableSet.java
@@ -0,0 +1,297 @@
+/*
+ * 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.utils;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.NavigableSet;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import java.util.SortedSet;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * A {@link NavigableSet} that enforces in-order insertion of elements. This is helpful when we
+ * have an already-ordered collection with no duplicates and want constant time insertion.
+ * <p>
+ * Note: Not all methods of {@link NavigableSet} are implemented.
+ *
+ * @param <E> the type of elements maintained by this set
+ */
+public class InsertionOrderedNavigableSet<E> implements NavigableSet<E>
+{
+ private final ArrayList<E> elements;
+ private final Comparator<? super E> comparator;
+
+ public InsertionOrderedNavigableSet(Comparator<? super E> comparator)
+ {
+ this.elements = new ArrayList<>();
+ this.comparator = comparator;
+ }
+
+ @Override
+ public E lower(E e)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public E floor(E e)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public E ceiling(E e)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public E higher(E e)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public E pollFirst()
+ {
+ if (isEmpty())
+ return null;
+
+ return elements.remove(0);
+ }
+
+ @Override
+ public E pollLast()
+ {
+ if (isEmpty())
+ return null;
+
+ return elements.remove(size() - 1);
+ }
+
+ @Override
+ public int size()
+ {
+ return elements.size();
+ }
+
+ @Override
+ public boolean isEmpty()
+ {
+ return elements.isEmpty();
+ }
+
+ @Override
+ public boolean contains(Object o)
+ {
+ return elements.contains(o);
+ }
+
+ @Override
+ public Iterator<E> iterator()
+ {
+ return elements.iterator();
+ }
+
+ @Override
+ public Object[] toArray()
+ {
+ return elements.toArray();
+ }
+
+ @Override
+ public <T> T[] toArray(T[] a)
+ {
+ return elements.toArray(a);
+ }
+
+ @Override
+ public boolean add(E e)
+ {
+ if (!isEmpty() && comparator.compare(e, last()) <= 0)
+ throw new IllegalStateException("Cannot add element " + e + " as it is not greater than the current last element " + last());
+
+ return elements.add(e);
+ }
+
+ @Override
+ public boolean remove(Object o)
+ {
+ return elements.remove(o);
+ }
+
+ @Override
+ public boolean containsAll(Collection<?> c)
+ {
+ return elements.containsAll(c);
+ }
+
+ @Override
+ public boolean addAll(Collection<? extends E> c)
+ {
+ boolean modified = false;
+ for (E element : c)
+ if (add(element))
+ modified = true;
+ return modified;
+ }
+
+ @Override
+ public boolean retainAll(Collection<?> c)
+ {
+ return elements.retainAll(c);
+ }
+
+ @Override
+ public boolean removeAll(Collection<?> c)
+ {
+ return elements.removeAll(c);
+ }
+
+ @Override
+ public void clear()
+ {
+ elements.clear();
+ }
+
+ @Override
+ public NavigableSet<E> descendingSet()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterator<E> descendingIterator()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NavigableSet<E> subSet(E fromElement, boolean fromInclusive, E toElement, boolean toInclusive)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NavigableSet<E> headSet(E toElement, boolean inclusive)
+ {
+ Preconditions.checkNotNull(toElement);
+
+ if (isEmpty())
+ return Collections.emptyNavigableSet();
+
+ NavigableSet<E> head = new InsertionOrderedNavigableSet<>(comparator);
+
+ for (E element : elements)
+ {
+ int comparison = comparator.compare(element, toElement);
+ if (comparison > 0 || comparison == 0 && !inclusive)
+ break;
+
+ head.add(element);
+ }
+
+ return head;
+ }
+
+ @Override
+ public NavigableSet<E> tailSet(E fromElement, boolean inclusive)
+ {
+ Preconditions.checkNotNull(fromElement);
+
+ if (isEmpty())
+ return Collections.emptyNavigableSet();
+
+ NavigableSet<E> tail = new InsertionOrderedNavigableSet<>(comparator);
+
+ for (E element : elements)
+ {
+ int comparison = comparator.compare(element, fromElement);
+ if (comparison < 0 || comparison == 0 && !inclusive)
+ continue;
+
+ tail.add(element);
+ }
+
+ return tail;
+ }
+
+ @Override
+ public Comparator<? super E> comparator()
+ {
+ return comparator;
+ }
+
+ @Override
+ public SortedSet<E> subSet(E fromElement, E toElement)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SortedSet<E> headSet(E toElement)
+ {
+ return headSet(toElement, false);
+ }
+
+ @Override
+ public SortedSet<E> tailSet(E fromElement)
+ {
+ return tailSet(fromElement, true);
+ }
+
+ @Override
+ public E first()
+ {
+ if (isEmpty())
+ throw new NoSuchElementException();
+
+ return elements.get(0);
+ }
+
+ @Override
+ public E last()
+ {
+ if (isEmpty())
+ throw new NoSuchElementException();
+
+ return elements.get(size() - 1);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ InsertionOrderedNavigableSet<?> that = (InsertionOrderedNavigableSet<?>) o;
+ return Objects.equals(elements, that.elements) && Objects.equals(comparator, that.comparator);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(elements, comparator);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org