import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
@Override
public List<Future<OperationResult<ColumnList<byte[]>>>> start() {
	ColumnFamily cf = info1.getColumnFamilyObj();
	Keyspace keyspace = columnFamilies.getKeyspace();
	
	List<Future<OperationResult<ColumnList<byte[]>>>> futures = new ArrayList<Future<OperationResult<ColumnList<byte[]>>>>();

	for(byte[] val : values) {
		Key from = new Key(val, true);
		Key to = new Key(val, true);
		byte[] rowKey = scanInfo.getRowKey();
		
		CompositeRangeBuilder range = CassandraSession.setupRangeBuilder(from, to, info1, reverse);
		ColumnFamilyQuery query = keyspace.prepareQuery(cf);
		RowQuery<byte[], byte[]> rowQuery = query.getKey(rowKey).withColumnRange(range);
		Future future = executeAsync(rowQuery);
		futures.add(future);
	}
	
	return futures;
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
@Override
public Collection<Locator> getLocators(SlotKey slotKey) throws IOException {

    Timer.Context ctx = Instrumentation.getReadTimerContext(CassandraModel.CF_METRICS_DELAYED_LOCATOR_NAME);

    try {
        RowQuery<SlotKey, Locator> query = AstyanaxIO.getKeyspace()
                .prepareQuery(CassandraModel.CF_METRICS_DELAYED_LOCATOR)
                .getKey(slotKey);
        return query.execute().getResult().getColumnNames();
    } catch (NotFoundException e) {
        Instrumentation.markNotFound(CassandraModel.CF_METRICS_DELAYED_LOCATOR_NAME);
        return Collections.emptySet();
    } catch (ConnectionException ex) {
        Instrumentation.markPoolExhausted();
        Instrumentation.markReadError();
        LOG.error("Connection exception during ADelayedLocatorIO.getLocators(" + slotKey.toString() + ")", ex);
        throw new IOException("Error reading delayed locators", ex);
    } finally {
        ctx.stop();
    }
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
/**
 * Returns the locators for a shard, i.e. those that should be rolled up, for a given shard.
 * 'Should' means:
 *  1) A locator is capable of rollup.
 *  2) A locator has had new data in the past LOCATOR_TTL seconds.
 *
 * @param shard Number of the shard you want the locators for. 0-127 inclusive.
 * @return Collection of locators
 * @throws IOException
 */
@Override
public Collection<Locator> getLocators(long shard) throws IOException {
    Timer.Context ctx = Instrumentation.getReadTimerContext(CassandraModel.CF_METRICS_LOCATOR_NAME);
    try {
        RowQuery<Long, Locator> query = AstyanaxIO.getKeyspace()
                .prepareQuery(CassandraModel.CF_METRICS_LOCATOR)
                .getKey(shard);
        if (LOG.isTraceEnabled())
            LOG.trace("ALocatorIO.getLocators() executing: select * from \"" + CassandraModel.KEYSPACE + "\"." + CassandraModel.CF_METRICS_LOCATOR_NAME + " where key=" + Long.toString(shard));
        return query.execute().getResult().getColumnNames();
    } catch (NotFoundException e) {
        Instrumentation.markNotFound(CassandraModel.CF_METRICS_LOCATOR_NAME);
        return Collections.emptySet();
    } catch (ConnectionException ex) {
        Instrumentation.markReadError(ex);
        LOG.error("Connection exception during getLocators(" + Long.toString(shard) + ")", ex);
        throw new IOException("Error reading locators", ex);
    } finally {
        ctx.stop();
    }
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
private Content resolve(long longId, Set<ContentColumn> colNames) {
    try {
        RowQuery<Long, String> query = keyspace.prepareQuery(mainCf)
                .getKey(longId);
        if (colNames != null && colNames.size() > 0) {
            query = query.withColumnSlice(Collections2.transform(
                    colNames,
                    Functions.toStringFunction()
            ));
        }
        ColumnList<String> cols = query.execute().getResult();
        if (cols.isEmpty()) {
            return null;
        }
        verifyRequiredColumns(longId, cols);
        return marshaller.unmarshallCols(cols);
    } catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
/** Executes a {@code RowQuery} with {@code autoPaginate(true)} repeatedly as necessary to fetch all pages. */
private <K, C> Iterator<Column<C>> executePaginated(final RowQuery<K, C> query) {
    return Iterators.concat(new AbstractIterator<Iterator<Column<C>>>() {
        @Override
        protected Iterator<Column<C>> computeNext() {
            ColumnList<C> page = execute(query);
            return !page.isEmpty() ? page.iterator() : endOfData();
        }
    });
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
/** Executes a {@code RowQuery} with {@code autoPaginate(true)} repeatedly as necessary to fetch all pages. */
private <K, C> Iterable<Column<C>> executePaginated(final RowQuery<K, C> query) {
    return OneTimeIterable.wrap(Iterators.concat(new AbstractIterator<Iterator<Column<C>>>() {
        @Override
        protected Iterator<Column<C>> computeNext() {
            ColumnList<C> page = execute(query);
            return !page.isEmpty() ? page.iterator() : endOfData();
        }
    }));
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
private Future executeAsync(RowQuery rowQuery) {
	try {
		return rowQuery.executeAsync();
	} catch (ConnectionException e) {
		throw new RuntimeException(e);
	}
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
private RowQuery createBasicRowQuery(byte[] rowKey, Info info1, ByteBufferRange range) {
	ColumnFamily cf = info1.getColumnFamilyObj();
	
	Keyspace keyspace = columnFamilies.getKeyspace();
	ColumnFamilyQuery query = keyspace.prepareQuery(cf);
	//ColumnFamilyQuery query = query1.setConsistencyLevel(ConsistencyLevel.CL_QUORUM);
	RowQuery rowQuery = query.getKey(rowKey)
						.autoPaginate(true)
						.withColumnRange(range);
	return rowQuery;
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
/**
 * For some dang reason with astyanax, we have to recreate the row query from scratch before we re-use it for 
 * a NEsted join.
 * @return
 */
public RowQuery<byte[], byte[]> createRowQuery() {
	CompositeRangeBuilder range = setupRangeBuilder(from, to, info1, false);
	if(batchSize != null)
		range = range.limit(batchSize);			
	return createBasicRowQuery(rowKey, info1, range);
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
/**
 * For some dang reason with astyanax, we have to recreate the row query from scratch before we re-use it for 
 * a NEsted join.
 * @return
 */
public RowQuery<byte[], byte[]> createRowQueryReverse() {
	CompositeRangeBuilder range = setupRangeBuilder(from, to, info1, true);
	if(batchSize != null)
		range = range.limit(batchSize);			
	return createBasicRowQuery(rowKey, info1, range);
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
@Override
public List<T> getGroup(String group) {
    try {
        List<T> list = new ArrayList<>();

        // Get all the row keys
        RowQuery<String, String> rowQuery = keyspace.prepareQuery(columnFamily).getKey(group)
            .autoPaginate(true)
            .withColumnRange(new RangeBuilder().setLimit(1000).build());

        List<String> rowKeys = new ArrayList<>();
        ColumnList<String> columns;
        while (!(columns = rowQuery.execute().getResult()).isEmpty()) {
            for (Column<String> c : columns) {
                rowKeys.add(c.getName());
            }
        }

        // Get values for all the fetched row keys
        RowSliceQuery<String, String> rowSliceQuery = keyspace.prepareQuery(columnFamily).getKeySlice(rowKeys);
        Rows<String, String> rows = rowSliceQuery.execute().getResult();
        for (Row<String, String> row : rows) {
            if (row.getColumns() != null && row.getColumns().size() > 0) {
                byte[] bytes = row.getColumns().getColumnByIndex(0).getByteArrayValue();
                list.add(objectMapper.readValue(codec.decompress(bytes), parameterClass));
            }
        }

        return list;
    } catch (ConnectionException | IOException e) {
        throw new RuntimeException(String.format("Exception occurred while fetching values for group '%s'", group), e);
    }
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
@Override
public Long getCounterValue(K key, String counterColumnName) {
	try {
		RowQuery<K, String> row = keyspace.prepareQuery(columnFamily).getKey(key);
		if (row != null) {
			ColumnQuery<String> column = row.getColumn(counterColumnName);
			if (column != null) {
				OperationResult<Column<String>> executeResult = column.execute();
				if (executeResult != null) {
					Column<String> result = executeResult.getResult();
					return result != null ? result.getLongValue() : 0L;
				} else {
					return 0L;
				}
			} else {
				return 0L;
			}
		} else {
			return 0L;
		}

	} catch (ConnectionException e) {
		if (log.isDebugEnabled()) {
			log.debug("HecubaClientManager error while reading key " + key.toString() + ". " +
					ExceptionUtils.getStackTrace(e));
			log.debug("Caught Exception", e);
		}
	}

	return 0L;
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
@Override
protected ColumnList<String> run() throws Exception {
     RowQuery<RowKeyType, String> rowQuery = keyspace.prepareQuery(columnFamily).getKey(rowKey);
     /* apply column slice if we have one */
     if (columns != null) {
         rowQuery = rowQuery.withColumnSlice(columns);
     }
     ColumnList<String> result = rowQuery.execute().getResult();
     return result;
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
@Override
public QueryResult readRow(String key, Integer columnCount, String startColumn, String endColumn, Boolean reversed) throws PaasException {
    invariant();
    
    try {
        // Construct the query
        RowQuery<ByteBuffer, ByteBuffer> query = keyspace
                .prepareQuery(this.columnFamily)
                .getRow(serializers.keyAsByteBuffer(key));
                
        RangeBuilder range = new RangeBuilder();
        if (columnCount != null && columnCount > 0) {
            range.setLimit(columnCount);
        }
        if (startColumn != null && !startColumn.isEmpty()) {
            range.setStart(serializers.columnAsByteBuffer(startColumn));
        }
        if (endColumn != null && !endColumn.isEmpty()) {
            range.setEnd(serializers.columnAsByteBuffer(endColumn));
        }
        range.setReversed(reversed);
        query.withColumnRange(range.build());

        // Execute the query
        ColumnList<ByteBuffer> result = query.execute().getResult();
        
        // Convert raw data into a simple sparse tree
        SchemalessRows.Builder builder = SchemalessRows.builder();
        Map<String, String> columns = Maps.newHashMap();
        if (!result.isEmpty()) {
            for (Column<ByteBuffer> column : result) { 
                columns.put(serializers.columnAsString(column.getRawName()), serializers.valueAsString(column.getRawName(), column.getByteBufferValue()));
            }
            builder.addRow(key, columns);
        }

        QueryResult dr = new QueryResult();
        dr.setSrows(builder.build());
        return dr;
    } catch (ConnectionException e) {
        throw new PaasException(
                String.format("Failed to read row '%s' in column family '%s.%s'" , 
                              key, this.keyspace.getKeyspaceName(), this.columnFamily.getName()),
                e);
    }
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
public ColumnNameIterator( RowQuery<?, C> rowQuery, final ColumnParser<C, T> parser, final boolean skipFirst ) {
    this.rowQuery = rowQuery.autoPaginate( true );
    this.parser = parser;
    this.skipFirst = skipFirst;
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
private static ColumnNameIterator<Long, Long> createIterator( final String rowKey, final boolean reversed ) {


        final ColumnParser<Long, Long> longParser = new ColumnParser<Long, Long>() {
            @Override
            public Long parseColumn( final Column<Long> column ) {
                return column.getName();
            }
        };

        final RangeBuilder forwardRange = new RangeBuilder().setLimit( 720 ).setReversed( reversed );


        final RowQuery<String, Long> forwardQuery =
                keyspace.prepareQuery( COLUMN_FAMILY ).getKey( rowKey ).withColumnRange( forwardRange.build() );


        ColumnNameIterator<Long, Long> itr = new ColumnNameIterator<>( forwardQuery, longParser, false );

        return itr;
    }
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
/**
 * Get the edge types from the search criteria.
 *
 * @param scope The org scope
 * @param search The edge type search info
 * @param cf The column family to execute on
 */
private Iterator<String> getEdgeTypes( final ApplicationScope scope, final SearchEdgeType search,
                                       final MultiTenantColumnFamily<ScopedRowKey<Id>, String> cf ) {
    ValidationUtils.validateApplicationScope( scope );
    GraphValidation.validateSearchEdgeType( search );


    final ScopedRowKey< Id> sourceKey = new ScopedRowKey<>( scope.getApplication(), search.getNode() );


    //resume from the last if specified.  Also set the range


    final RangeBuilder rangeBuilder = createRange( search );

    RowQuery<ScopedRowKey<Id>, String> query =
            keyspace.prepareQuery( cf ).getKey( sourceKey ).autoPaginate( true )
                    .withColumnRange( rangeBuilder.build() );

    return new ColumnNameIterator<>( query, PARSER, search.getLast().isPresent() );
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
/**
 * Get the id types from the specified column family
 *
 * @param scope The organization scope to use
 * @param search The search criteria
 * @param cf The column family to search
 */
public Iterator<String> getIdTypes( final ApplicationScope scope, final SearchIdType search,
                                    final MultiTenantColumnFamily<ScopedRowKey<EdgeIdTypeKey>, String> cf ) {
    ValidationUtils.validateApplicationScope( scope );
    GraphValidation.validateSearchEdgeIdType( search );


    final ScopedRowKey<EdgeIdTypeKey> sourceTypeKey =
            new ScopedRowKey<>( scope.getApplication(), new EdgeIdTypeKey( search.getNode(), search.getEdgeType() ) );


    final RangeBuilder rangeBuilder = createRange( search );


    RowQuery<ScopedRowKey<EdgeIdTypeKey>, String> query =
            keyspace.prepareQuery( cf ).getKey( sourceTypeKey ).autoPaginate( true )
                    .withColumnRange( rangeBuilder.build() );


    return new ColumnNameIterator<>( query, PARSER, search.getLast().isPresent() );
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
@Override
public Iterator<MvccEntity> loadDescendingHistory( final ApplicationScope applicationScope, final Id entityId,
                                                   final UUID version, final int fetchSize ) {

    Preconditions.checkNotNull( applicationScope, "applicationScope is required" );
    Preconditions.checkNotNull( entityId, "entity id is required" );
    Preconditions.checkNotNull( version, "version is required" );
    Preconditions.checkArgument( fetchSize > 0, "max Size must be greater than 0" );


    final Id applicationId = applicationScope.getApplication();
    final Id ownerId = applicationId;
    final String collectionName = LegacyScopeUtils.getCollectionScopeNameFromEntityType( entityId.getType() );

    final CollectionPrefixedKey<Id> collectionPrefixedKey =
            new CollectionPrefixedKey<>( collectionName, ownerId, entityId );


    final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey =
            ScopedRowKey.fromKey( applicationId, collectionPrefixedKey );


    RowQuery<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> query =
            keyspace.prepareQuery( columnFamily ).getKey( rowKey )
                    .withColumnRange( version, null, false, fetchSize );

    return new ColumnNameIterator( query, new MvccColumnParser( entityId, getEntitySerializer() ), false );
}
 

import com.netflix.astyanax.query.RowQuery; //导入依赖的package包/类
@Override
public Iterator<MvccEntity> loadAscendingHistory( final ApplicationScope applicationScope, final Id entityId,
                                                  final UUID version, final int fetchSize ) {

    Preconditions.checkNotNull( applicationScope, "applicationScope is required" );
    Preconditions.checkNotNull( entityId, "entity id is required" );
    Preconditions.checkNotNull( version, "version is required" );
    Preconditions.checkArgument( fetchSize > 0, "max Size must be greater than 0" );


    final Id applicationId = applicationScope.getApplication();
    final Id ownerId = applicationId;
    final String collectionName = LegacyScopeUtils.getCollectionScopeNameFromEntityType( entityId.getType() );

    final CollectionPrefixedKey<Id> collectionPrefixedKey =
            new CollectionPrefixedKey<>( collectionName, ownerId, entityId );


    final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey =
            ScopedRowKey.fromKey( applicationId, collectionPrefixedKey );


    RowQuery<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> query =
            keyspace.prepareQuery( columnFamily ).getKey( rowKey )
                    .withColumnRange( null, version, true, fetchSize );

    return new ColumnNameIterator( query, new MvccColumnParser( entityId, getEntitySerializer() ), false );
}