/*
 * 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.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.*;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.PeekingIterator;
import org.slf4j.Logger;

import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;


Wrapper class for handling of multiple MerkleTrees at once. The MerkleTree's are divided in Ranges of non-overlapping tokens.
/** * Wrapper class for handling of multiple MerkleTrees at once. * * The MerkleTree's are divided in Ranges of non-overlapping tokens. */
public class MerkleTrees implements Iterable<Map.Entry<Range<Token>, MerkleTree>> { public static final MerkleTreesSerializer serializer = new MerkleTreesSerializer(); private Map<Range<Token>, MerkleTree> merkleTrees = new TreeMap<>(new TokenRangeComparator()); private IPartitioner partitioner;
Creates empty MerkleTrees object.
Params:
  • partitioner – The partitioner to use
/** * Creates empty MerkleTrees object. * * @param partitioner The partitioner to use */
public MerkleTrees(IPartitioner partitioner) { this(partitioner, new ArrayList<>()); } private MerkleTrees(IPartitioner partitioner, Collection<MerkleTree> merkleTrees) { this.partitioner = partitioner; addTrees(merkleTrees); }
Get the ranges that these merkle trees covers.
Returns:
/** * Get the ranges that these merkle trees covers. * * @return */
public Collection<Range<Token>> ranges() { return merkleTrees.keySet(); }
Get the partitioner in use.
Returns:
/** * Get the partitioner in use. * * @return */
public IPartitioner partitioner() { return partitioner; }
Add merkle tree's with the defined maxsize and ranges.
Params:
  • maxsize –
  • ranges –
/** * Add merkle tree's with the defined maxsize and ranges. * * @param maxsize * @param ranges */
public void addMerkleTrees(int maxsize, Collection<Range<Token>> ranges) { for (Range<Token> range : ranges) { addMerkleTree(maxsize, range); } }
Add a MerkleTree with the defined size and range.
Params:
  • maxsize –
  • range –
Returns:The created merkle tree.
/** * Add a MerkleTree with the defined size and range. * * @param maxsize * @param range * @return The created merkle tree. */
public MerkleTree addMerkleTree(int maxsize, Range<Token> range) { return addMerkleTree(maxsize, MerkleTree.RECOMMENDED_DEPTH, range); } @VisibleForTesting public MerkleTree addMerkleTree(int maxsize, byte hashdepth, Range<Token> range) { MerkleTree tree = new MerkleTree(partitioner, range, hashdepth, maxsize); addTree(tree); return tree; }
Get the MerkleTree.Range responsible for the given token.
Params:
  • t –
Returns:
/** * Get the MerkleTree.Range responsible for the given token. * * @param t * @return */
@VisibleForTesting public MerkleTree.TreeRange get(Token t) { return getMerkleTree(t).get(t); }
Init all MerkleTree's with an even tree distribution.
/** * Init all MerkleTree's with an even tree distribution. */
public void init() { for (Range<Token> range : merkleTrees.keySet()) { init(range); } }
Init a selected MerkleTree with an even tree distribution.
Params:
  • range –
/** * Init a selected MerkleTree with an even tree distribution. * * @param range */
public void init(Range<Token> range) { merkleTrees.get(range).init(); }
Split the MerkleTree responsible for the given token.
Params:
  • t –
Returns:
/** * Split the MerkleTree responsible for the given token. * * @param t * @return */
public boolean split(Token t) { return getMerkleTree(t).split(t); }
Invalidate the MerkleTree responsible for the given token.
Params:
  • t –
/** * Invalidate the MerkleTree responsible for the given token. * * @param t */
@VisibleForTesting public void invalidate(Token t) { getMerkleTree(t).invalidate(t); }
Get the MerkleTree responsible for the given token range.
Params:
  • range –
Returns:
/** * Get the MerkleTree responsible for the given token range. * * @param range * @return */
public MerkleTree getMerkleTree(Range<Token> range) { return merkleTrees.get(range); } public long size() { long size = 0; for (MerkleTree tree : merkleTrees.values()) { size += tree.size(); } return size; } @VisibleForTesting public void maxsize(Range<Token> range, int maxsize) { getMerkleTree(range).maxsize(maxsize); }
Get the MerkleTree responsible for the given token.
Params:
  • t –
Returns:The given MerkleTree or null if none exist.
/** * Get the MerkleTree responsible for the given token. * * @param t * @return The given MerkleTree or null if none exist. */
private MerkleTree getMerkleTree(Token t) { for (Range<Token> range : merkleTrees.keySet()) { if (range.contains(t)) return merkleTrees.get(range); } throw new AssertionError("Expected tree for token " + t); } private void addTrees(Collection<MerkleTree> trees) { for (MerkleTree tree : trees) { addTree(tree); } } private void addTree(MerkleTree tree) { assert validateNonOverlapping(tree) : "Range [" + tree.fullRange + "] is intersecting an existing range"; merkleTrees.put(tree.fullRange, tree); } private boolean validateNonOverlapping(MerkleTree tree) { for (Range<Token> range : merkleTrees.keySet()) { if (tree.fullRange.intersects(range)) return false; } return true; }
Get an iterator for all the invalids generated by the MerkleTrees.
Returns:
/** * Get an iterator for all the invalids generated by the MerkleTrees. * * @return */
public TreeRangeIterator invalids() { return new TreeRangeIterator(); }
Log the row count per leaf for all MerkleTrees.
Params:
  • logger –
/** * Log the row count per leaf for all MerkleTrees. * * @param logger */
public void logRowCountPerLeaf(Logger logger) { for (MerkleTree tree : merkleTrees.values()) { tree.histogramOfRowCountPerLeaf().log(logger); } }
Log the row size per leaf for all MerkleTrees.
Params:
  • logger –
/** * Log the row size per leaf for all MerkleTrees. * * @param logger */
public void logRowSizePerLeaf(Logger logger) { for (MerkleTree tree : merkleTrees.values()) { tree.histogramOfRowSizePerLeaf().log(logger); } } @VisibleForTesting public byte[] hash(Range<Token> range) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); boolean hashed = false; try { for (Range<Token> rt : merkleTrees.keySet()) { if (rt.intersects(range)) { byte[] bytes = merkleTrees.get(rt).hash(range); if (bytes != null) { baos.write(bytes); hashed = true; } } } } catch (IOException e) { throw new RuntimeException("Unable to append merkle tree hash to result"); } return hashed ? baos.toByteArray() : null; }
Get an iterator of all ranges and their MerkleTrees.
/** * Get an iterator of all ranges and their MerkleTrees. */
public Iterator<Map.Entry<Range<Token>, MerkleTree>> iterator() { return merkleTrees.entrySet().iterator(); } public long rowCount() { long totalCount = 0; for (MerkleTree tree : merkleTrees.values()) { totalCount += tree.rowCount(); } return totalCount; } public class TreeRangeIterator extends AbstractIterator<MerkleTree.TreeRange> implements Iterable<MerkleTree.TreeRange>, PeekingIterator<MerkleTree.TreeRange> { private final Iterator<MerkleTree> it; private MerkleTree.TreeRangeIterator current = null; private TreeRangeIterator() { it = merkleTrees.values().iterator(); } public MerkleTree.TreeRange computeNext() { if (current == null || !current.hasNext()) return nextIterator(); return current.next(); } private MerkleTree.TreeRange nextIterator() { if (it.hasNext()) { current = it.next().invalids(); return current.next(); } return endOfData(); } public Iterator<MerkleTree.TreeRange> iterator() { return this; } }
Get the differences between the two sets of MerkleTrees.
Params:
  • ltree –
  • rtree –
Returns:
/** * Get the differences between the two sets of MerkleTrees. * * @param ltree * @param rtree * @return */
public static List<Range<Token>> difference(MerkleTrees ltree, MerkleTrees rtree) { List<Range<Token>> differences = new ArrayList<>(); for (MerkleTree tree : ltree.merkleTrees.values()) { differences.addAll(MerkleTree.difference(tree, rtree.getMerkleTree(tree.fullRange))); } return differences; } public static class MerkleTreesSerializer implements IVersionedSerializer<MerkleTrees> { public void serialize(MerkleTrees trees, DataOutputPlus out, int version) throws IOException { out.writeInt(trees.merkleTrees.size()); for (MerkleTree tree : trees.merkleTrees.values()) { MerkleTree.serializer.serialize(tree, out, version); } } public MerkleTrees deserialize(DataInputPlus in, int version) throws IOException { IPartitioner partitioner = null; int nTrees = in.readInt(); Collection<MerkleTree> trees = new ArrayList<>(nTrees); if (nTrees > 0) { for (int i = 0; i < nTrees; i++) { MerkleTree tree = MerkleTree.serializer.deserialize(in, version); trees.add(tree); if (partitioner == null) partitioner = tree.partitioner(); else assert tree.partitioner() == partitioner; } } return new MerkleTrees(partitioner, trees); } public long serializedSize(MerkleTrees trees, int version) { assert trees != null; long size = TypeSizes.sizeof(trees.merkleTrees.size()); for (MerkleTree tree : trees.merkleTrees.values()) { size += MerkleTree.serializer.serializedSize(tree, version); } return size; } } private static class TokenRangeComparator implements Comparator<Range<Token>> { @Override public int compare(Range<Token> rt1, Range<Token> rt2) { if (rt1.left.compareTo(rt2.left) == 0) return 0; return rt1.compareTo(rt2); } } }