ClusteringComparator.java
/*
* 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.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableList;
import org.apache.cassandra.db.marshal.ValueAccessor;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.io.sstable.IndexInfo;
import org.apache.cassandra.utils.bytecomparable.ByteComparable;
import org.apache.cassandra.utils.bytecomparable.ByteSource;
import static org.apache.cassandra.utils.bytecomparable.ByteSource.EXCLUDED;
import static org.apache.cassandra.utils.bytecomparable.ByteSource.NEXT_COMPONENT;
import static org.apache.cassandra.utils.bytecomparable.ByteSource.NEXT_COMPONENT_EMPTY;
import static org.apache.cassandra.utils.bytecomparable.ByteSource.NEXT_COMPONENT_EMPTY_REVERSED;
import static org.apache.cassandra.utils.bytecomparable.ByteSource.NEXT_COMPONENT_NULL;
import static org.apache.cassandra.utils.bytecomparable.ByteSource.TERMINATOR;
/**
* A comparator of clustering prefixes (or more generally of {@link Clusterable}}.
* <p>
* This is essentially just a composite comparator that the clustering values of the provided
* clustering prefixes in lexicographical order, with each component being compared based on
* the type of the clustering column this is a value of.
*/
public class ClusteringComparator implements Comparator<Clusterable>
{
private final List<AbstractType<?>> clusteringTypes;
private final Comparator<IndexInfo> indexComparator;
private final Comparator<IndexInfo> indexReverseComparator;
private final Comparator<Clusterable> reverseComparator;
private final Comparator<Row> rowComparator = (r1, r2) -> compare((ClusteringPrefix<?>) r1.clustering(),
(ClusteringPrefix<?>) r2.clustering());
public ClusteringComparator(AbstractType<?>... clusteringTypes)
{
this(ImmutableList.copyOf(clusteringTypes));
}
public ClusteringComparator(Iterable<AbstractType<?>> clusteringTypes)
{
// copy the list to ensure despatch is monomorphic
this.clusteringTypes = ImmutableList.copyOf(clusteringTypes);
this.indexComparator = (o1, o2) -> ClusteringComparator.this.compare((ClusteringPrefix<?>) o1.lastName,
(ClusteringPrefix<?>) o2.lastName);
this.indexReverseComparator = (o1, o2) -> ClusteringComparator.this.compare((ClusteringPrefix<?>) o1.firstName,
(ClusteringPrefix<?>) o2.firstName);
this.reverseComparator = (c1, c2) -> ClusteringComparator.this.compare(c2, c1);
for (AbstractType<?> type : clusteringTypes)
type.checkComparable(); // this should already be enforced by TableMetadata.Builder.addColumn, but we check again for other constructors
}
/**
* The number of clustering columns for the table this is the comparator of.
*/
public int size()
{
return clusteringTypes.size();
}
/**
* The "subtypes" of this clustering comparator, that is the types of the clustering
* columns for the table this is a comparator of.
*/
public List<AbstractType<?>> subtypes()
{
return clusteringTypes;
}
/**
* Returns the type of the ith clustering column of the table.
*/
public AbstractType<?> subtype(int i)
{
return clusteringTypes.get(i);
}
/**
* Creates a row clustering based on the clustering values.
* <p>
* Every argument can either be a {@code ByteBuffer}, in which case it is used as-is, or a object
* corresponding to the type of the corresponding clustering column, in which case it will be
* converted to a byte buffer using the column type.
*
* @param values the values to use for the created clustering. There should be exactly {@code size()}
* values which must be either byte buffers or of the type the column expect.
*
* @return the newly created clustering.
*/
public Clustering<?> make(Object... values)
{
if (values.length != size())
throw new IllegalArgumentException(String.format("Invalid number of components, expecting %d but got %d", size(), values.length));
CBuilder builder = CBuilder.create(this);
for (Object val : values)
{
if (val instanceof ByteBuffer)
builder.add((ByteBuffer) val);
else
builder.add(val);
}
return builder.build();
}
public int compare(Clusterable c1, Clusterable c2)
{
return compare((ClusteringPrefix<?>) c1.clustering(), (ClusteringPrefix<?>) c2.clustering());
}
public <V1, V2> int compare(ClusteringPrefix<V1> c1, ClusteringPrefix<V2> c2)
{
int s1 = c1.size();
int s2 = c2.size();
int minSize = Math.min(s1, s2);
for (int i = 0; i < minSize; i++)
{
int cmp = compareComponent(i, c1.get(i), c1.accessor(), c2.get(i), c2.accessor());
if (cmp != 0)
return cmp;
}
if (s1 == s2)
return ClusteringPrefix.Kind.compare(c1.kind(), c2.kind());
return s1 < s2 ? c1.kind().comparedToClustering : -c2.kind().comparedToClustering;
}
public <V1, V2> int compare(Clustering<V1> c1, Clustering<V2> c2)
{
return compare(c1, c2, size());
}
/**
* Compares the specified part of the specified clusterings.
*
* @param c1 the first clustering
* @param c2 the second clustering
* @param size the number of components to compare
* @return a negative integer, zero, or a positive integer as the first argument is less than,
* equal to, or greater than the second.
*/
public <V1, V2> int compare(Clustering<V1> c1, Clustering<V2> c2, int size)
{
for (int i = 0; i < size; i++)
{
int cmp = compareComponent(i, c1.get(i), c1.accessor(), c2.get(i), c2.accessor());
if (cmp != 0)
return cmp;
}
return 0;
}
public <V1, V2> int compareComponent(int i, V1 v1, ValueAccessor<V1> accessor1, V2 v2, ValueAccessor<V2> accessor2)
{
if (v1 == null)
return v2 == null ? 0 : -1;
if (v2 == null)
return 1;
return clusteringTypes.get(i).compare(v1, accessor1, v2, accessor2);
}
public <V1, V2> int compareComponent(int i, ClusteringPrefix<V1> v1, ClusteringPrefix<V2> v2)
{
return compareComponent(i, v1.get(i), v1.accessor(), v2.get(i), v2.accessor());
}
/**
* Returns whether this clustering comparator is compatible with the provided one,
* that is if the provided one can be safely replaced by this new one.
*
* @param previous the previous comparator that we want to replace and test
* compatibility with.
*
* @return whether {@code previous} can be safely replaced by this comparator.
*/
public boolean isCompatibleWith(ClusteringComparator previous)
{
if (this == previous)
return true;
// Extending with new components is fine, shrinking is not
if (size() < previous.size())
return false;
for (int i = 0; i < previous.size(); i++)
{
AbstractType<?> tprev = previous.subtype(i);
AbstractType<?> tnew = subtype(i);
if (!tnew.isCompatibleWith(tprev))
return false;
}
return true;
}
/**
* Validates the provided prefix for corrupted data.
*
* @param clustering the clustering prefix to validate.
*
* @throws MarshalException if {@code clustering} contains some invalid data.
*/
public <T> void validate(ClusteringPrefix<T> clustering)
{
ValueAccessor<T> accessor = clustering.accessor();
for (int i = 0; i < clustering.size(); i++)
{
T value = clustering.get(i);
if (value != null)
subtype(i).validate(value, accessor);
}
}
/**
* Produce a prefix-free byte-comparable representation of the given value, i.e. such a sequence of bytes that any
* pair x, y of valid values of this type
* compare(x, y) == compareLexicographicallyUnsigned(asByteComparable(x), asByteComparable(y))
* and
* asByteComparable(x) is not a prefix of asByteComparable(y)
*/
public <V> ByteComparable asByteComparable(ClusteringPrefix<V> clustering)
{
return new ByteComparableClustering<>(clustering);
}
/**
* A prefix-free byte-comparable representation for a clustering or prefix.
*
* Adds a NEXT_COMPONENT byte before each component (allowing inclusive/exclusive bounds over incomplete prefixes
* of that length) and finishes with a suitable byte for the clustering kind. Also deals with null entries.
*
* Since all types' encodings are weakly prefix-free, this is guaranteed to be prefix-free as long as the
* bound/ClusteringPrefix terminators are different from the separator byte. It is okay for the terminator for
* Clustering to be the same as the separator, as all Clusterings must be completely specified.
*
* See also {@link AbstractType#asComparableBytes}.
*
* Some examples:
* "A", 0005, Clustering -> 40 4100 40 0005 40
* "B", 0006, InclusiveEnd -> 40 4200 40 0006 60
* "A", ExclusiveStart -> 40 4100 60
* "", null, Clustering -> 40 00 3F 40
* "", 0000, Clustering -> 40 00 40 0000 40
* BOTTOM -> 20
*/
private class ByteComparableClustering<V> implements ByteComparable
{
private final ClusteringPrefix<V> src;
ByteComparableClustering(ClusteringPrefix<V> src)
{
this.src = src;
}
@Override
public ByteSource asComparableBytes(Version version)
{
return new ByteSource()
{
private ByteSource current = null;
private int srcnum = -1;
@Override
public int next()
{
if (current != null)
{
int b = current.next();
if (b > END_OF_STREAM)
return b;
current = null;
}
int sz = src.size();
if (srcnum == sz)
return END_OF_STREAM;
++srcnum;
if (srcnum == sz)
return src.kind().asByteComparableValue(version);
final V nextComponent = src.get(srcnum);
// We can have a null as the clustering component (this is a relic of COMPACT STORAGE, but also
// can appear in indexed partitions with no rows but static content),
if (nextComponent == null)
{
if (version != Version.LEGACY)
return NEXT_COMPONENT_NULL; // always sorts before non-nulls, including for reversed types
else
{
// legacy version did not permit nulls in clustering keys and treated these as null values
return subtype(srcnum).isReversed() ? NEXT_COMPONENT_EMPTY_REVERSED : NEXT_COMPONENT_EMPTY;
}
}
current = subtype(srcnum).asComparableBytes(src.accessor(), nextComponent, version);
// and also null values for some types (e.g. int, varint but not text) that are encoded as empty
// buffers.
if (current == null)
return subtype(srcnum).isReversed() ? NEXT_COMPONENT_EMPTY_REVERSED : NEXT_COMPONENT_EMPTY;
return NEXT_COMPONENT;
}
};
}
public String toString()
{
return src.clusteringString(subtypes());
}
}
/**
* Produces a clustering from the given byte-comparable value. The method will throw an exception if the value
* does not correctly encode a clustering of this type, including if it encodes a position before or after a
* clustering (i.e. a bound/boundary).
*
* @param accessor Accessor to use to construct components.
* @param comparable The clustering encoded as a byte-comparable sequence.
*/
public <V> Clustering<V> clusteringFromByteComparable(ValueAccessor<V> accessor, ByteComparable comparable)
{
ByteComparable.Version version = ByteComparable.Version.OSS50;
ByteSource.Peekable orderedBytes = ByteSource.peekable(comparable.asComparableBytes(version));
if (orderedBytes == null)
return null;
// First check for special cases (partition key only, static clustering) that can do without buffers.
int sep = orderedBytes.next();
switch (sep)
{
case TERMINATOR:
assert size() == 0 : "Terminator should be after " + size() + " components, got 0";
return accessor.factory().clustering();
case EXCLUDED:
return accessor.factory().staticClustering();
default:
// continue with processing
}
int cc = 0;
V[] components = accessor.createArray(size());
while (true)
{
switch (sep)
{
case NEXT_COMPONENT_NULL:
components[cc] = null;
break;
case NEXT_COMPONENT_EMPTY:
case NEXT_COMPONENT_EMPTY_REVERSED:
components[cc] = subtype(cc).fromComparableBytes(accessor, null, version);
break;
case NEXT_COMPONENT:
// Decode the next component, consuming bytes from orderedBytes.
components[cc] = subtype(cc).fromComparableBytes(accessor, orderedBytes, version);
break;
case TERMINATOR:
assert cc == size() : "Terminator should be after " + size() + " components, got " + cc;
return accessor.factory().clustering(components);
case EXCLUDED:
throw new AssertionError("Unexpected static terminator after the first component");
default:
throw new AssertionError("Unexpected separator " + Integer.toHexString(sep) + " in Clustering encoding");
}
++cc;
sep = orderedBytes.next();
}
}
/**
* Produces a clustering bound from the given byte-comparable value. The method will throw an exception if the value
* does not correctly encode a bound position of this type, including if it encodes an exact clustering.
*
* Note that the encoded clustering position cannot specify the type of bound (i.e. start/end/boundary) because to
* correctly compare clustering positions the encoding must be the same for the different types (e.g. the position
* for a exclusive end and an inclusive start is the same, before the exact clustering). The type must be supplied
* separately (in the bound... vs boundary... call and isEnd argument).
*
* @param accessor Accessor to use to construct components.
* @param comparable The clustering position encoded as a byte-comparable sequence.
* @param isEnd true if the bound marks the end of a range, false is it marks the start.
*/
public <V> ClusteringBound<V> boundFromByteComparable(ValueAccessor<V> accessor,
ByteComparable comparable,
boolean isEnd)
{
ByteComparable.Version version = ByteComparable.Version.OSS50;
ByteSource.Peekable orderedBytes = ByteSource.peekable(comparable.asComparableBytes(version));
int sep = orderedBytes.next();
int cc = 0;
V[] components = accessor.createArray(size());
while (true)
{
switch (sep)
{
case NEXT_COMPONENT_NULL:
components[cc] = null;
break;
case NEXT_COMPONENT_EMPTY:
case NEXT_COMPONENT_EMPTY_REVERSED:
components[cc] = subtype(cc).fromComparableBytes(accessor, null, version);
break;
case NEXT_COMPONENT:
// Decode the next component, consuming bytes from orderedBytes.
components[cc] = subtype(cc).fromComparableBytes(accessor, orderedBytes, version);
break;
case ByteSource.LT_NEXT_COMPONENT:
return accessor.factory().bound(isEnd ? ClusteringPrefix.Kind.EXCL_END_BOUND
: ClusteringPrefix.Kind.INCL_START_BOUND,
Arrays.copyOf(components, cc));
case ByteSource.GT_NEXT_COMPONENT:
return accessor.factory().bound(isEnd ? ClusteringPrefix.Kind.INCL_END_BOUND
: ClusteringPrefix.Kind.EXCL_START_BOUND,
Arrays.copyOf(components, cc));
case ByteSource.LTLT_NEXT_COMPONENT:
case ByteSource.GTGT_NEXT_COMPONENT:
throw new AssertionError("Unexpected sstable lower/upper bound - byte comparable representation of artificial sstable bounds is not supported");
default:
throw new AssertionError("Unexpected separator " + Integer.toHexString(sep) + " in ClusteringBound encoding");
}
++cc;
sep = orderedBytes.next();
}
}
/**
* Produces a clustering boundary from the given byte-comparable value. The method will throw an exception if the
* value does not correctly encode a bound position of this type, including if it encodes an exact clustering.
*
* Note that the encoded clustering position cannot specify the type of bound (i.e. start/end/boundary) because to
* correctly compare clustering positions the encoding must be the same for the different types (e.g. the position
* for a exclusive end and an inclusive start is the same, before the exact clustering). The type must be supplied
* separately (in the bound... vs boundary... call and isEnd argument).
*
* @param accessor Accessor to use to construct components.
* @param comparable The clustering position encoded as a byte-comparable sequence.
*/
public <V> ClusteringBoundary<V> boundaryFromByteComparable(ValueAccessor<V> accessor, ByteComparable comparable)
{
ByteComparable.Version version = ByteComparable.Version.OSS50;
ByteSource.Peekable orderedBytes = ByteSource.peekable(comparable.asComparableBytes(version));
int sep = orderedBytes.next();
int cc = 0;
V[] components = accessor.createArray(size());
while (true)
{
switch (sep)
{
case NEXT_COMPONENT_NULL:
components[cc] = null;
break;
case NEXT_COMPONENT_EMPTY:
case NEXT_COMPONENT_EMPTY_REVERSED:
components[cc] = subtype(cc).fromComparableBytes(accessor, null, version);
break;
case NEXT_COMPONENT:
// Decode the next component, consuming bytes from orderedBytes.
components[cc] = subtype(cc).fromComparableBytes(accessor, orderedBytes, version);
break;
case ByteSource.LT_NEXT_COMPONENT:
return accessor.factory().boundary(ClusteringPrefix.Kind.EXCL_END_INCL_START_BOUNDARY,
Arrays.copyOf(components, cc));
case ByteSource.GT_NEXT_COMPONENT:
return accessor.factory().boundary(ClusteringPrefix.Kind.INCL_END_EXCL_START_BOUNDARY,
Arrays.copyOf(components, cc));
default:
throw new AssertionError("Unexpected separator " + Integer.toHexString(sep) + " in ClusteringBoundary encoding");
}
++cc;
sep = orderedBytes.next();
}
}
/**
* A comparator for rows.
*
* A {@code Row} is a {@code Clusterable} so {@code ClusteringComparator} can be used
* to compare rows directly, but when we know we deal with rows (and not {@code Clusterable} in
* general), this is a little faster because by knowing we compare {@code Clustering} objects,
* we know that 1) they all have the same size and 2) they all have the same kind.
*/
public Comparator<Row> rowComparator()
{
return rowComparator;
}
public Comparator<IndexInfo> indexComparator(boolean reversed)
{
return reversed ? indexReverseComparator : indexComparator;
}
public Comparator<Clusterable> reversed()
{
return reverseComparator;
}
@Override
public String toString()
{
return String.format("comparator(%s)", Joiner.on(", ").join(clusteringTypes));
}
@Override
public boolean equals(Object o)
{
if (this == o)
return true;
if (!(o instanceof ClusteringComparator))
return false;
ClusteringComparator that = (ClusteringComparator)o;
return this.clusteringTypes.equals(that.clusteringTypes);
}
@Override
public int hashCode()
{
return Objects.hashCode(clusteringTypes);
}
}