src/main/java/com/thinkaurelius/titan/diskstorage/cassandra/utils/CassandraHelper.java

   1 package com.thinkaurelius.titan.diskstorage.cassandra.utils;
   2
   3 import java.nio.ByteBuffer;
   4 import java.util.*;
   5
   6 import com.google.common.base.Function;
   7 import com.google.common.base.Preconditions;
   8 import com.google.common.base.Predicate;
   9 import com.google.common.collect.Iterators;
  10 import com.thinkaurelius.titan.diskstorage.EntryList;
  11 import com.thinkaurelius.titan.diskstorage.StaticBuffer;
  12 import com.thinkaurelius.titan.diskstorage.Entry;
  13 import com.thinkaurelius.titan.diskstorage.keycolumnvalue.KeyRange;
  14 import com.thinkaurelius.titan.diskstorage.util.BufferUtil;
  15 import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer;
  16 import com.thinkaurelius.titan.diskstorage.util.StaticArrayEntry;
  17 import com.thinkaurelius.titan.diskstorage.util.StaticArrayEntryList;
  18 import org.apache.cassandra.dht.BytesToken;
  19 import org.apache.cassandra.dht.Range;
  20 import org.apache.cassandra.dht.Token;
  21
  22 import javax.annotation.Nullable;
  23
  24 public class CassandraHelper {
  25
  26     public static List<ByteBuffer> convert(List<StaticBuffer> keys) {
  27         List<ByteBuffer> requestKeys = new ArrayList<ByteBuffer>(keys.size());
  28         for (int i = 0; i < keys.size(); i++) {
  29             requestKeys.add(keys.get(i).asByteBuffer());
  30         }
  31         return requestKeys;
  32     }
  33
  34     /**
  35      * Constructs an {@link EntryList} from the Iterable of entries while excluding the end slice
  36      * (since the method contract states that the end slice is exclusive, yet Cassandra treats it as
  37      * inclusive) and respecting the limit.
  38      *
  39      * @param entries
  40      * @param getter
  41      * @param lastColumn TODO: make this StaticBuffer so we can avoid the conversion and provide equals method
  42      * @param limit
  43      * @param <E>
  44      * @return
  45      */
  46     public static<E> EntryList makeEntryList(final Iterable<E> entries,
  47                                              final StaticArrayEntry.GetColVal<E,ByteBuffer> getter,
  48                                              final StaticBuffer lastColumn, final int limit) {
  49         return StaticArrayEntryList.ofByteBuffer(new Iterable<E>() {
  50             @Override
  51             public Iterator<E> iterator() {
  52                 return Iterators.filter(entries.iterator(),new FilterResultColumns<E>(lastColumn,limit,getter));
  53             }
  54         },getter);
  55     }
  56
  57     private static class FilterResultColumns<E> implements Predicate<E> {
  58
  59         private int count = 0;
  60
  61         private final int limit;
  62         private final StaticBuffer lastColumn;
  63         private final StaticArrayEntry.GetColVal<E,ByteBuffer> getter;
  64
  65         private FilterResultColumns(StaticBuffer lastColumn, int limit, StaticArrayEntry.GetColVal<E, ByteBuffer> getter) {
  66             this.limit = limit;
  67             this.lastColumn = lastColumn;
  68             this.getter = getter;
  69         }
  70
  71         @Override
  72         public boolean apply(@Nullable E e) {
  73             assert e!=null;
  74             if (count>=limit || BufferUtil.equals(lastColumn, getter.getColumn(e))) return false;
  75             count++;
  76             return true;
  77         }
  78
  79     }
  80
  81     public static<E> Iterator<Entry> makeEntryIterator(final Iterable<E> entries,
  82                                              final StaticArrayEntry.GetColVal<E,ByteBuffer> getter,
  83                                              final StaticBuffer lastColumn, final int limit) {
  84         return Iterators.transform(Iterators.filter(entries.iterator(),
  85                 new FilterResultColumns<E>(lastColumn, limit, getter)), new Function<E, Entry>() {
  86             @Nullable
  87             @Override
  88             public Entry apply(@Nullable E e) {
  89                 return StaticArrayEntry.ofByteBuffer(e,getter);
  90             }
  91         });
  92     }
  93
  94
  95     public static KeyRange transformRange(Range<Token> range) {
  96         return transformRange(range.left, range.right);
  97     }
  98
  99     public static KeyRange transformRange(Token leftKeyExclusive, Token rightKeyInclusive) {
 100         if (!(leftKeyExclusive instanceof BytesToken))
 101             throw new UnsupportedOperationException();
 102
 103         // if left part is BytesToken, right part should be too, otherwise there is no sense in the ring
 104         assert rightKeyInclusive instanceof BytesToken;
 105
 106         // l is exclusive, r is inclusive
 107         BytesToken l = (BytesToken) leftKeyExclusive;
 108         BytesToken r = (BytesToken) rightKeyInclusive;
 109
 110         byte[] leftTokenValue = l.getTokenValue();
 111         byte[] rightTokenValue = r.getTokenValue();
 112
 113         Preconditions.checkArgument(leftTokenValue.length == rightTokenValue.length, "Tokens have unequal length");
 114         int tokenLength = leftTokenValue.length;
 115
 116         byte[][] tokens = new byte[][]{leftTokenValue, rightTokenValue};
 117         byte[][] plusOne = new byte[2][tokenLength];
 118
 119         for (int j = 0; j < 2; j++) {
 120             boolean carry = true;
 121             for (int i = tokenLength - 1; i >= 0; i--) {
 122                 byte b = tokens[j][i];
 123                 if (carry) {
 124                     b++;
 125                     carry = false;
 126                 }
 127                 if (b == 0) carry = true;
 128                 plusOne[j][i] = b;
 129             }
 130         }
 131
 132         StaticBuffer lb = StaticArrayBuffer.of(plusOne[0]);
 133         StaticBuffer rb = StaticArrayBuffer.of(plusOne[1]);
 134         Preconditions.checkArgument(lb.length() == tokenLength, lb.length());
 135         Preconditions.checkArgument(rb.length() == tokenLength, rb.length());
 136
 137         return new KeyRange(lb, rb);
 138     }
 139 }