Java 类org.apache.cassandra.thrift.KeyRange 实例源码

private static String toString(KeyRange keyRange) {
    ByteBuffer startKey = keyRange.start_key;
    String startKeyStr = "<null>";
    if (startKey != null) {
        startKeyStr = Utils.toString(startKey.array(), startKey.arrayOffset(), startKey.limit());
    }
    if (startKeyStr.length() == 0) {
        startKeyStr = "<first>";
    }
    ByteBuffer endKey = keyRange.end_key;
    String endKeyStr = "<null>";
    if (endKey != null) {
        endKeyStr = Utils.toString(endKey.array(), endKey.arrayOffset(), endKey.limit());
    }
    if (endKeyStr.length() == 0) {
        endKeyStr = "<last>";
    }
    StringBuilder buffer = new StringBuilder();
    if (startKeyStr.equals("<first>") && endKeyStr.equals("<last>")) {
        buffer.append("Keys(<all>)");
    } else if (startKeyStr.equals(endKeyStr)) {
        buffer.append("Key('");
        buffer.append(startKeyStr);
        buffer.append("')");
    } else {
        buffer.append("Keys('");
        buffer.append(startKeyStr);
        buffer.append("' to '");
        buffer.append(endKeyStr);
        buffer.append("')");
    }
    return buffer.toString();
}

/**
 * Create a KeyRange that begins at the given row key.
 * 
 * @param startRowKey   Starting row key as a byte[].
 * @return              KeyRange that starts at the given row, open-ended.
 */
static KeyRange keyRangeStartRow(byte[] startRowKey) {
    KeyRange keyRange = new KeyRange();
    keyRange.setStart_key(startRowKey);
    keyRange.setEnd_key(EMPTY_BYTE_BUFFER);
    keyRange.setCount(MAX_ROWS_BATCH_SIZE);
    return keyRange;
}

static KeyRange keyRangeStartRow(byte[] startRowKey, int count) {
    KeyRange keyRange = new KeyRange();
    keyRange.setStart_key(startRowKey == null ? EMPTY_BYTES : startRowKey);
    keyRange.setEnd_key(EMPTY_BYTES);
    keyRange.setCount(count);
    return keyRange;
}

/**
 * Create a KeyRange that selects a single row with the given key.
 * 
 * @param rowKey    Row key as a byte[].
 * @return          KeyRange that starts and ends with the given key.
 */
static KeyRange keyRangeSingleRow(byte[] rowKey) {
    KeyRange keyRange = new KeyRange();
    keyRange.setStart_key(rowKey);
    keyRange.setEnd_key(rowKey);
    keyRange.setCount(1);
    return keyRange;
}

private static KeyRange keyRangeFromString(String st) {
    assert st != null;
    TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
    KeyRange keyRange = new KeyRange();
    try {
        deserializer.deserialize(keyRange, Hex.hexToBytes(st));
    } catch (TException e) {
        throw new RuntimeException(e);
    }
    return keyRange;
}

private static String keyRangeToString(KeyRange keyRange)
{
    assert keyRange != null;
    TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
    try
    {
        return FBUtilities.bytesToHex(serializer.serialize(keyRange));
    }
    catch (TException e)
    {
        throw new RuntimeException(e);
    }
}

private static KeyRange keyRangeFromString(String st)
{
    assert st != null;
    TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
    KeyRange keyRange = new KeyRange();
    try
    {
        deserializer.deserialize(keyRange, FBUtilities.hexToBytes(st));
    }
    catch (TException e)
    {
        throw new RuntimeException(e);
    }
    return keyRange;
}

/**
 * get
 *
 * @throws Exception
 */
@Test
public void getByKey() throws Exception {
    String KEYSPACE = "mock";
    client.set_keyspace(KEYSPACE);

    String COLUMN_FAMILY = "student";
    // 讀取整筆
    ColumnParent columnParent = new ColumnParent(COLUMN_FAMILY);
    // 術語
    SlicePredicate predicate = new SlicePredicate();

    // InvalidRequestException(why:predicate column_names and slice_range
    // may not both be present)

    // 範圍
    // SliceRange sliceRange = new SliceRange();
    // sliceRange.setStart(new byte[0]);// 開始
    // sliceRange.setFinish(new byte[0]);// 結束
    // predicate.setSlice_range(sliceRange);

    // 讀取1個column
    predicate.addToColumn_names(ByteBufferHelper.toByteBuffer("grad"));

    // key範圍
    KeyRange keyRange = new KeyRange();
    keyRange.setStart_key(new byte[0]);
    keyRange.setEnd_key(new byte[0]);
    keyRange.setCount(100);

    // 結果
    // column_parent, predicate, range, consistency_level
    List<KeySlice> results = client.get_range_slices(columnParent,
            predicate, keyRange, ConsistencyLevel.ONE);

    for (KeySlice keySlice : results) {

        for (ColumnOrSuperColumn cos : keySlice.getColumns()) {
            Column column = cos.column;
            System.out.println(ByteHelper.toString(keySlice.getKey())
                    + ", " + ByteHelper.toString(column.getName()) + ": "
                    + ByteHelper.toString(column.getValue()) + ", "
                    + column.getTimestamp());
            // Rose, grad, 4, 1380931646061000
            // Jack, art, 87, 1380933848350
            // Jack, grad, 5, 1380932164492000
            // Jack, math, 97, 1380933848305
        }
    }
}

public void sliceModeInit(CassandraColumnMetaData meta,
    List<String> colNames, int maxRows, int maxCols, int rowBatchSize,
    int colBatchSize) throws KettleException {

  m_newSliceQuery = true;
  m_requestedCols = colNames;
  m_sliceRowsMax = maxRows;
  m_sliceColsMax = maxCols;
  m_sliceRowsBatchSize = rowBatchSize;
  m_sliceColsBatchSize = colBatchSize;
  m_rowIndex = 0;
  m_colIndex = 0;

  if (m_sliceColsBatchSize <= 0) {
    m_sliceColsBatchSize = Integer.MAX_VALUE;
  }

  if (m_sliceRowsBatchSize <= 0) {
    m_sliceRowsBatchSize = Integer.MAX_VALUE;
  }

  List<ByteBuffer> specificCols = null;
  if (m_requestedCols != null && m_requestedCols.size() > 0) {
    specificCols = new ArrayList<ByteBuffer>();

    // encode the textual column names
    for (String colName : m_requestedCols) {
      ByteBuffer encoded = meta.columnNameToByteBuffer(colName);
      specificCols.add(encoded);
    }
  }

  m_slicePredicate = new SlicePredicate();

  if (specificCols == null) {
    m_sliceRange = new SliceRange(ByteBuffer.wrap(new byte[0]),
        ByteBuffer.wrap(new byte[0]), false, m_sliceColsBatchSize);
    m_slicePredicate.setSlice_range(m_sliceRange);
  } else {
    m_slicePredicate.setColumn_names(specificCols);
  }

  m_keyRange = new KeyRange(m_sliceRowsBatchSize);
  m_keyRange.setStart_key(new byte[0]);
  m_keyRange.setEnd_key(new byte[0]);

  m_colParent = new ColumnParent(meta.getColumnFamilyName());
  m_converted = new ArrayList<Object[]>();
}

public List<POI> findPOIByHotel(String hotel) throws Exception {

        // /query
        SlicePredicate predicate = new SlicePredicate();
        SliceRange sliceRange = new SliceRange();
        sliceRange.setStart(hotel.getBytes());
        sliceRange.setFinish(hotel.getBytes());
        predicate.setSlice_range(sliceRange);

        // read all columns in the row
        String scFamily = "PointOfInterest";
        ColumnParent parent = new ColumnParent(scFamily);

        KeyRange keyRange = new KeyRange();
        keyRange.start_key = bytes("");
        keyRange.end_key = bytes("");

        List<POI> pois = new ArrayList<POI>();

        // instead of a simple list, we get a map whose keys are row keys
        // and the values the list of columns returned for each
        // only row key + first column are indexed
        Connector cl = new Connector();
        Cassandra.Client client = cl.connect();
        List<KeySlice> slices = client.get_range_slices(parent, predicate, keyRange, CL);

        for (KeySlice slice : slices) {
            List<ColumnOrSuperColumn> cols = slice.columns;

            POI poi = new POI();
            poi.name = new String(ByteBufferUtil.string(slice.key));

            for (ColumnOrSuperColumn cosc : cols) {
                SuperColumn sc = cosc.super_column;

                List<Column> colsInSc = sc.columns;

                for (Column c : colsInSc) {
                    String colName = new String(c.name.array(), UTF8);
                    if (colName.equals("desc")) {
                        poi.desc = new String(c.value.array(), UTF8);
                    }
                    if (colName.equals("phone")) {
                        poi.phone = new String(c.value.array(), UTF8);
                    }
                }

                LOG.debug("Found something neat nearby: " + poi.name + ". \nDesc: " + poi.desc + ". \nPhone: "
                        + poi.phone);
                pois.add(poi);
            }
        }

        cl.close();
        return pois;
    }

public List<Hotel> findHotelByCity(String city, String state) throws Exception {

        LOG.debug("Seaching for hotels in " + city + ", " + state);

        String key = city + ":" + state.toUpperCase();

        // /query
        SlicePredicate predicate = new SlicePredicate();
        SliceRange sliceRange = new SliceRange();
        sliceRange.setStart(new byte[0]);
        sliceRange.setFinish(new byte[0]);
        predicate.setSlice_range(sliceRange);

        // read all columns in the row
        String columnFamily = "HotelByCity";
        ColumnParent parent = new ColumnParent(columnFamily);

        KeyRange keyRange = new KeyRange();
        keyRange.setStart_key(key.getBytes());
        keyRange.setEnd_key("".getBytes()); // just outside lexical range
        keyRange.count = 5;

        Connector cl = new Connector();
        Cassandra.Client client = cl.connect();
        List<KeySlice> keySlices = client.get_range_slices(parent, predicate, keyRange, CL);

        List<Hotel> results = new ArrayList<Hotel>();

        for (KeySlice ks : keySlices) {
            List<ColumnOrSuperColumn> coscs = ks.columns;
            LOG.debug(new String("Using key " + ks.key));

            for (ColumnOrSuperColumn cs : coscs) {

                Hotel hotel = new Hotel();
                hotel.name = ByteBufferUtil.string(cs.column.name);
                hotel.city = city;
                hotel.state = state;

                results.add(hotel);
                LOG.debug("Found hotel result for " + hotel.name);
            }
        }
        // /end query
        cl.close();

        return results;
    }

@Override
public void run(final ThriftClient client) throws IOException
{
    final SlicePredicate predicate = new SlicePredicate()
            .setSlice_range(
                    new SliceRange(
                            ByteBufferUtil.EMPTY_BYTE_BUFFER,
                            ByteBufferUtil.EMPTY_BYTE_BUFFER,
                            false,
                            state.settings.columns.maxColumnsPerKey
                    )
            );

    final ByteBuffer start = getKey();
    final KeyRange range =
            new KeyRange(state.settings.columns.maxColumnsPerKey)
                    .setStart_key(start)
                    .setEnd_key(ByteBufferUtil.EMPTY_BYTE_BUFFER)
                    .setCount(((SettingsCommandMulti)state.settings.command).keysAtOnce);

    for (final ColumnParent parent : state.columnParents)
    {
        timeWithRetry(new RunOp()
        {
            private int count = 0;
            @Override
            public boolean run() throws Exception
            {
                return (count = client.get_range_slices(parent, predicate, range, state.settings.command.consistencyLevel).size()) != 0;
            }

            @Override
            public String key()
            {
                return new String(range.bufferForStart_key().array());
            }

            @Override
            public int keyCount()
            {
                return count;
            }
        });
    }
}

@Override
public void run(final ThriftClient client) throws IOException
{
    final SlicePredicate predicate = new SlicePredicate()
            .setSlice_range(
                    new SliceRange(
                            ByteBufferUtil.EMPTY_BYTE_BUFFER,
                            ByteBufferUtil.EMPTY_BYTE_BUFFER,
                            false,
                            state.settings.columns.maxColumnsPerKey
                    )
            );

    final ByteBuffer start = getKey();
    final KeyRange range =
            new KeyRange(state.settings.columns.maxColumnsPerKey)
                    .setStart_key(start)
                    .setEnd_key(ByteBufferUtil.EMPTY_BYTE_BUFFER)
                    .setCount(state.settings.command.keysAtOnce);

    for (final ColumnParent parent : state.columnParents)
    {
        timeWithRetry(new RunOp()
        {
            private int count = 0;
            @Override
            public boolean run() throws Exception
            {
                return (count = client.get_range_slices(parent, predicate, range, state.settings.command.consistencyLevel).size()) != 0;
            }

            @Override
            public String key()
            {
                return new String(range.bufferForStart_key().array());
            }

            @Override
            public int keyCount()
            {
                return count;
            }
        });
    }
}

/**
 * may be null if unset
 */
public static KeyRange getInputKeyRange(Configuration conf) {
    String str = conf.get(INPUT_KEYRANGE_CONFIG);
    return str == null ? null : keyRangeFromString(str);
}

private void maybeInit() {
  // check if we need another row
  if (rows != null && columnsRead < rowPageSize) {
    columnsRead = 0;
    startToken = partitioner.getTokenFactory().toString(partitioner.getToken(rows.get(0).key));
    predicate.getSlice_range().setStart(startSlicePredicate);
    rows = null;
    prevStartSlice = null;
    totalRead++;
  }

  if (startToken == null) {
    startToken = split.getStartToken();
  } else if (startToken.equals(split.getEndToken()) && rows == null) {
    // reached end of the split
    return;
  }

  KeyRange keyRange = new KeyRange(batchRowCount)
                            .setStart_token(startToken)
                            .setEnd_token(split.getEndToken());
  try {
    rows = client.get_range_slices(new ColumnParent(cfName),
                                           predicate,
                                           keyRange,
                                           consistencyLevel);

    // nothing new? reached the end
    if (rows.isEmpty()) {
      rows = null;
      return;
    }

    //detect infinite loop
    if (prevStartSlice != null && ByteBufferUtil.compareUnsigned(prevStartSlice, predicate.slice_range.start) == 0) {
        rows = null;
        return;
    }

    // prepare for the next slice to be read
    KeySlice row = rows.get(0);

    if (row.getColumnsSize() > 0) {

      ColumnOrSuperColumn cosc = row.getColumns().get(row.getColumnsSize() - 1);

      prevStartSlice = predicate.slice_range.start;

      //prepare next slice
      if (cosc.column != null) {
        predicate.slice_range.start = cosc.column.name;
      }

      if (cosc.super_column != null) {
        predicate.slice_range.start = cosc.super_column.name;
      }

      if (cosc.counter_column != null) {
        predicate.slice_range.start = cosc.counter_column.name;
      }

      if (cosc.counter_super_column != null) {
        predicate.slice_range.start = cosc.counter_super_column.name;
      }

      columnsRead = row.getColumnsSize();

      //If we've hit the max columns then rm the last column
      //to make sure we don't know where to start next without overlap
      if (columnsRead == rowPageSize) {
        row.getColumns().remove(columnsRead - 1);
      }
    } 
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
}

/**
 * Set the KeyRange to limit the rows.
 * @param conf Job configuration you are about to run
 */
public static void setInputRange(Configuration conf, String startToken, String endToken)
{
    KeyRange range = new KeyRange().setStart_token(startToken).setEnd_token(endToken);
    conf.set(INPUT_KEYRANGE_CONFIG, keyRangeToString(range));
}

/** may be null if unset */
public static KeyRange getInputKeyRange(Configuration conf)
{
    String str = conf.get(INPUT_KEYRANGE_CONFIG);
    return null != str ? keyRangeFromString(str) : null;
}

/**
 * Set the KeyRange to limit the rows.
 *
 * @param conf Job configuration you are about to run
 */
public static void setInputRange(Configuration conf, String startToken, String endToken) {
    KeyRange range = new KeyRange().setStart_token(startToken).setEnd_token(endToken);
    conf.set(INPUT_KEYRANGE_CONFIG, thriftToString(range));
}

/**
 * Set the KeyRange to limit the rows.
 *
 * @param conf Job configuration you are about to run
 */
public static void setInputRange(Configuration conf, String startToken, String endToken, List<IndexExpression> filter) {
    KeyRange range = new KeyRange().setStart_token(startToken).setEnd_token(endToken).setRow_filter(filter);
    conf.set(INPUT_KEYRANGE_CONFIG, thriftToString(range));
}

/**
 * Set the KeyRange to limit the rows.
 *
 * @param conf Job configuration you are about to run
 */
public static void setInputRange(Configuration conf, List<IndexExpression> filter) {
    KeyRange range = new KeyRange().setRow_filter(filter);
    conf.set(INPUT_KEYRANGE_CONFIG, thriftToString(range));
}