private Clause createClauseWith(ConditionalExpression e) {
  switch (e.getOperator()) {
    case EQ:
      return eq(e.getName(), bindMarker());
    case NE:
      return ne(e.getName(), bindMarker());
    case GT:
      return gt(e.getName(), bindMarker());
    case GTE:
      return gte(e.getName(), bindMarker());
    case LT:
      return lt(e.getName(), bindMarker());
    case LTE:
      return lte(e.getName(), bindMarker());
    default:
      // never comes here because ConditionalExpression accepts only above operators
      throw new IllegalArgumentException(e.getOperator() + " is not supported");
  }
}
 

protected Clause condition2Cql(Condition condition) {
    switch (condition.type()) {
        case AND:
            Condition.And and = (Condition.And) condition;
            Clause left = condition2Cql(and.left());
            Clause right = condition2Cql(and.right());
            return Clauses.and(left, right);
        case OR:
            throw new BackendException("Not support OR currently");
        case RELATION:
            Condition.Relation r = (Condition.Relation) condition;
            return relation2Cql(r);
        default:
            final String msg = "Unsupported condition: " + condition;
            throw new AssertionError(msg);
    }
}
 

public void updateShardPointer(final Shard shard){

        Assignment assignment = QueryBuilder.set(COLUMN_POINTER, shard.getPointer());

        Clause queueNameClause = QueryBuilder.eq(COLUMN_QUEUE_NAME, shard.getQueueName());
        Clause regionClause = QueryBuilder.eq(COLUMN_REGION, shard.getRegion());
        Clause activeClause = QueryBuilder.eq(COLUMN_ACTIVE, 1);
        Clause shardIdClause = QueryBuilder.eq(COLUMN_SHARD_ID, shard.getShardId());

        Statement update = QueryBuilder.update(getTableName(shard.getType()))
                .with(assignment)
                .where(queueNameClause)
                .and(regionClause)
                .and(activeClause)
                .and(shardIdClause);

        cassandraClient.getQueueMessageSession().execute(update);

    }
 

@Override
public DatabaseQueueMessageBody loadMessageData(final UUID messageId ){

    logger.trace("loadMessageData {}", messageId);

    Clause messageIdClause = QueryBuilder.eq( COLUMN_MESSAGE_ID, messageId );

    Statement select = QueryBuilder.select().from( TABLE_MESSAGE_DATA).where(messageIdClause);

    Row row = cassandraClient.getApplicationSession().execute(select).one();
    if ( row == null ) {
        return null;
    }

    return new DatabaseQueueMessageBody(
            row.getBytes( COLUMN_MESSAGE_DATA),
            row.getString( COLUMN_CONTENT_TYPE));
}
 

private Statement createDeleteAllMessagesStatement( Shard shard ) {

        Clause queueNameClause = QueryBuilder.eq(      COLUMN_QUEUE_NAME, shard.getQueueName() );
        Clause regionClause = QueryBuilder.eq(         COLUMN_REGION, shard.getRegion() );
        Clause shardIdClause = QueryBuilder.eq(        COLUMN_SHARD_ID, shard.getShardId() );

        DatabaseQueueMessage.Type dbqmType = Shard.Type.DEFAULT.equals( shard.getType() )
            ? DatabaseQueueMessage.Type.DEFAULT : DatabaseQueueMessage.Type.INFLIGHT;

        Statement deleteAll = QueryBuilder.delete().from( getTableName( dbqmType ))
            .where(queueNameClause)
            .and(regionClause)
            .and(shardIdClause);

        return deleteAll;
    }
 

private static void addWhereInClauses(Select.Where where, List<CassandraColumnHandle> partitionKeyColumns, List<Set<Object>> filterPrefixes)
{
    for (int i = 0; i < filterPrefixes.size(); i++) {
        CassandraColumnHandle column = partitionKeyColumns.get(i);
        List<Object> values = filterPrefixes.get(i)
                .stream()
                .map(value -> column.getCassandraType().getJavaValue(value))
                .collect(toList());
        Clause clause = QueryBuilder.in(CassandraCqlUtils.validColumnName(column.getName()), values);
        where.and(clause);
    }
}
 

private static void addWhereClause(Select.Where where, List<CassandraColumnHandle> partitionKeyColumns, List<Object> filterPrefix)
{
    for (int i = 0; i < filterPrefix.size(); i++) {
        CassandraColumnHandle column = partitionKeyColumns.get(i);
        Object value = column.getCassandraType().getJavaValue(filterPrefix.get(i));
        Clause clause = QueryBuilder.eq(CassandraCqlUtils.validColumnName(column.getName()), value);
        where.and(clause);
    }
}
 

protected Collection<Select> queryCondition2Select(Query query,
                                                   Select select) {
    // Query by conditions
    Set<Condition> conditions = query.conditions();
    for (Condition condition : conditions) {
        Clause clause = condition2Cql(condition);
        select.where(clause);
        if (Clauses.needAllowFiltering(clause)) {
            select.allowFiltering();
        }
    }
    return ImmutableList.of(select);
}
 

public long getCounter(CassandraSessionPool.Session session,
                       HugeType type) {
    Clause where = formatEQ(HugeKeys.SCHEMA_TYPE, type.name());
    Select select = QueryBuilder.select(formatKey(HugeKeys.ID))
                                .from(TABLE);
    select.where(where);
    Row row = session.execute(select).one();
    if (row == null) {
        return 0L;
    } else {
        return row.getLong(formatKey(HugeKeys.ID));
    }
}
 

@Override
public void delete( final MapScope scope, final String key ) {

    Statement deleteMapEntry;
    Clause equalsEntryKey = QueryBuilder.eq("key", getMapEntryPartitionKey(scope, key));
    deleteMapEntry = QueryBuilder.delete().from(MAP_ENTRIES_TABLE)
        .where(equalsEntryKey);
    session.execute(deleteMapEntry);



    // not sure which bucket the value is in, execute a delete against them all
    final int[] buckets = BUCKET_LOCATOR.getAllBuckets( scope.getName() );
    List<ByteBuffer> mapKeys = new ArrayList<>();
    for( int bucket :  buckets){
        mapKeys.add( getMapKeyPartitionKey(scope, bucket));
    }

    Statement deleteMapKey;
    Clause inKey = QueryBuilder.in("key", mapKeys);
    Clause column1Equals = QueryBuilder.eq("column1", DataType.text().serialize(key, ProtocolVersion.NEWEST_SUPPORTED));
    deleteMapKey = QueryBuilder.delete().from(MAP_KEYS_TABLE)
        .where(inKey).and(column1Equals);
    session.execute(deleteMapKey);


}
 

private ByteBuffer getValueCQL( MapScope scope, String key, final ConsistencyLevel consistencyLevel ) {

        Clause in = QueryBuilder.in("key", getMapEntryPartitionKey(scope, key) );
        Statement statement = QueryBuilder.select().all().from(MAP_ENTRIES_TABLE)
            .where(in)
            .setConsistencyLevel(consistencyLevel);

        ResultSet resultSet = session.execute(statement);
        com.datastax.driver.core.Row row = resultSet.one();

        return row != null ? row.getBytes("value") : null;
    }
 

public void deleteShard(final Shard shard){

        Clause queueNameClause = QueryBuilder.eq(COLUMN_QUEUE_NAME, shard.getQueueName());
        Clause regionClause = QueryBuilder.eq(COLUMN_REGION, shard.getRegion());
        Clause activeClause = QueryBuilder.eq(COLUMN_ACTIVE, 1);
        Clause shardIdClause = QueryBuilder.eq(COLUMN_SHARD_ID, shard.getShardId());

        Statement delete = QueryBuilder.delete().from(getTableName(shard.getType()))
                .where(queueNameClause)
                .and(regionClause)
                .and(activeClause)
                .and(shardIdClause);

        cassandraClient.getQueueMessageSession().execute(delete);
    }
 

@Override
public DatabaseQueue getQueue(String name) {

    logger.trace( "getQueue " + name );

    Clause queueNameClause = QueryBuilder.eq(COLUMN_QUEUE_NAME, name);

    Statement query = QueryBuilder.select().all().from(TABLE_QUEUES)
            .where(queueNameClause);

    Row row = cassandraClient.getApplicationSession().execute(query).one();

    if(row == null){
        return null;
    }

    final String queueName = row.getString(COLUMN_QUEUE_NAME);
    final String regions = row.getString(COLUMN_REGIONS);
    final String defaultDestinations = row.getString(COLUMN_DEFAULT_DESTINATIONS);
    final long defaultDelayMs = row.getLong(COLUMN_DEFAULT_DELAY_MS);
    final int retryCount = row.getInt(COLUMN_RETRY_COUNT);
    final int handlingTimeoutSec = row.getInt(COLUMN_HANDLING_TIMEOUT_SEC);
    final String deadLetterQueue = row.getString(COLUMN_DEAD_LETTER_QUEUE);

    return new DatabaseQueue( queueName, regions, defaultDestinations, defaultDelayMs, retryCount,
            handlingTimeoutSec, deadLetterQueue);

}
 

private Statement createDeleteMessageStatement( final String queueName,
                                                final String region,
                                                final Long shardIdOrNull,
                                                final DatabaseQueueMessage.Type type,
                                                final UUID queueMessageId ) {
    final long shardId;
    if ( shardIdOrNull == null ) {
        Shard.Type shardType = DatabaseQueueMessage.Type.DEFAULT.equals( type ) ?
            Shard.Type.DEFAULT : Shard.Type.INFLIGHT;
        Shard shard = shardStrategy.selectShard(
            queueName, region, shardType, queueMessageId );
        shardId = shard.getShardId();
    } else {
        shardId = shardIdOrNull;
    }

    Clause queueNameClause = QueryBuilder.eq(      COLUMN_QUEUE_NAME, queueName );
    Clause regionClause = QueryBuilder.eq(         COLUMN_REGION, region );
    Clause shardIdClause = QueryBuilder.eq(        COLUMN_SHARD_ID, shardId );
    Clause queueMessageIdClause = QueryBuilder.eq( COLUMN_QUEUE_MESSAGE_ID, queueMessageId);

    Statement delete = QueryBuilder.delete().from(getTableName( type ))
        .where(queueNameClause)
        .and(regionClause)
        .and(shardIdClause)
        .and(queueMessageIdClause);

    return delete;
}
 

protected Clause relation2Cql(Relation relation) {
    String key = relation.serialKey().toString();
    Object value = relation.serialValue();

    switch (relation.relation()) {
        case EQ:
            return QueryBuilder.eq(key, value);
        case GT:
            return QueryBuilder.gt(key, value);
        case GTE:
            return QueryBuilder.gte(key, value);
        case LT:
            return QueryBuilder.lt(key, value);
        case LTE:
            return QueryBuilder.lte(key, value);
        case IN:
            return QueryBuilder.in(key, value);
        case CONTAINS_VALUE:
            return QueryBuilder.contains(key, value);
        case CONTAINS_KEY:
            return QueryBuilder.containsKey(key, value);
        case SCAN:
            String[] col = pkColumnName().stream()
                                         .map(pk -> formatKey(pk))
                                         .toArray(String[]::new);
            Shard shard = (Shard) value;
            Object start = QueryBuilder.raw(shard.start());
            Object end = QueryBuilder.raw(shard.end());
            return Clauses.and(
                    QueryBuilder.gte(QueryBuilder.token(col), start),
                    QueryBuilder.lt(QueryBuilder.token(col), end));
        /*
         * Currently we can't support LIKE due to error:
         * "cassandra no viable alternative at input 'like'..."
         */
        // case LIKE:
        //    return QueryBuilder.like(key, value);
        case NEQ:
        default:
            throw new NotSupportException("relation '%s'", relation);
    }
}
 

public static final Clause formatEQ(HugeKeys key, Object value) {
    return QueryBuilder.eq(formatKey(key), value);
}
 

protected List<E> findPageWithTimeSearch(String searchView, List<Clause> clauses, TimePageLink pageLink) {
    return findPageWithTimeSearch(searchView, clauses, Collections.emptyList(), pageLink);
}
 

protected List<E> findPageWithTimeSearch(String searchView, List<Clause> clauses, Ordering ordering, TimePageLink pageLink) {
    return findPageWithTimeSearch(searchView, clauses, Collections.singletonList(ordering), pageLink);
}
 

protected List<E> findPageWithTimeSearch(String searchView, List<Clause> clauses, List<Ordering> topLevelOrderings, TimePageLink pageLink) {
    return findPageWithTimeSearch(searchView, clauses, topLevelOrderings, pageLink, ModelConstants.ID_PROPERTY);
}
 

protected List<E> findPageWithTimeSearch(String searchView, List<Clause> clauses, TimePageLink pageLink, String idColumn) {
    return findPageWithTimeSearch(searchView, clauses, Collections.emptyList(), pageLink, idColumn);
}
 

protected List<E> findPageWithTimeSearch(String searchView, List<Clause> clauses, List<Ordering> topLevelOrderings, TimePageLink pageLink, String idColumn) {
    return findListByStatement(buildQuery(searchView, clauses, topLevelOrderings, pageLink, idColumn));
}
 

public static Where buildQuery(String searchView, List<Clause> clauses, TimePageLink pageLink, String idColumn) {
    return buildQuery(searchView, clauses, Collections.emptyList(), pageLink, idColumn);
}
 

public static Where buildQuery(String searchView, List<Clause> clauses, Ordering order, TimePageLink pageLink, String idColumn) {
    return buildQuery(searchView, clauses, Collections.singletonList(order), pageLink, idColumn);
}
 

private Consumer<TimeRangeBound> constrainBound(
        Select select,
        BiFunction<String, Long, Clause> inclusive,
        BiFunction<String, Long, Clause> exclusive) {
    return bound -> select.where((bound.isInclusive() ? inclusive : exclusive).apply("eventTimestamp", bound.getInstant().toEpochMilli()));
}
 

@Override
public MapKeyResults getAllKeys(final MapScope scope, final String cursor, final int limit ){

    final int[] buckets = BUCKET_LOCATOR.getAllBuckets( scope.getName() );
    final List<ByteBuffer> partitionKeys = new ArrayList<>(NUM_BUCKETS.length);

    for (int bucket : buckets) {

        partitionKeys.add(getMapKeyPartitionKey(scope, bucket));
    }

    Clause in = QueryBuilder.in("key", partitionKeys);

    Statement statement;
    if( isBlank(cursor) ){
        statement = QueryBuilder.select().all().from(MAP_KEYS_TABLE)
            .where(in)
            .setFetchSize(limit);
    }else{
        statement = QueryBuilder.select().all().from(MAP_KEYS_TABLE)
            .where(in)
            .setFetchSize(limit)
            .setPagingState(PagingState.fromString(cursor));
    }


    ResultSet resultSet = session.execute(statement);
    PagingState pagingState = resultSet.getExecutionInfo().getPagingState();

    final List<String> keys = new ArrayList<>();
    Iterator<Row> resultIterator = resultSet.iterator();
    int size = 0;
    while( resultIterator.hasNext() && size < limit){

        size++;
        keys.add((String)DataType.text().deserialize(resultIterator.next().getBytes("column1"), ProtocolVersion.NEWEST_SUPPORTED));

    }

    return new MapKeyResults(pagingState != null ? pagingState.toString() : null, keys);

}
 

private <T> T getValuesCQL(
    final MapScope scope, final Collection<String> keys, final ResultsBuilderCQL<T> builder ) {

    final List<ByteBuffer> serializedKeys = new ArrayList<>();

    keys.forEach(key -> serializedKeys.add(getMapEntryPartitionKey(scope,key)));

    Clause in = QueryBuilder.in("key", serializedKeys );
    Statement statement = QueryBuilder.select().all().from(MAP_ENTRIES_TABLE)
        .where(in);


    ResultSet resultSet = session.execute(statement);

    return builder.buildResultsCQL( resultSet );
}
 

private void removeValueCQL(CacheScope scope, K key) {

        Preconditions.checkNotNull( scope, "scope is required");
        Preconditions.checkNotNull( key, "key is required" );

        // determine bucketed row-key based application UUID

        final String rowKeyString = scope.getApplication().getUuid().toString();
        final int bucket = BUCKET_LOCATOR.getCurrentBucket(rowKeyString);

        // determine column name based on K key to string
        final String columnName = key.toString();


        final Clause inKey = QueryBuilder.eq("key", getPartitionKey(scope, rowKeyString, bucket) );
        final Clause inColumn = QueryBuilder.eq("column1", DataType.text().serialize(columnName, ProtocolVersion.NEWEST_SUPPORTED) );

        final Statement statement = QueryBuilder.delete().from(SCOPED_CACHE_TABLE)
            .where(inKey)
            .and(inColumn);

        session.execute(statement);

    }
 

private void invalidateCQL(CacheScope scope){

        Preconditions.checkNotNull(scope, "scope is required");

        // determine bucketed row-key based application UUID
        final String rowKeyString = scope.getApplication().getUuid().toString();
        final int bucket = BUCKET_LOCATOR.getCurrentBucket(rowKeyString);

        final Clause inKey = QueryBuilder.eq("key", getPartitionKey(scope, rowKeyString, bucket) );

        final Statement statement = QueryBuilder.delete().from(SCOPED_CACHE_TABLE)
            .where(inKey);

        session.execute(statement);

    }
 

private void advance(){

        if (currentShard == null){
            currentShard = shardIterator.next();
        }

        Clause queueNameClause = QueryBuilder.eq( COLUMN_QUEUE_NAME, queueName);
        Clause regionClause    = QueryBuilder.eq( COLUMN_REGION, region);
        Clause shardIdClause   = QueryBuilder.eq( COLUMN_SHARD_ID, currentShard.getShardId());

        // if we have a pointer from the shard and this is the first seek, init from the pointer's position
        if ( currentShard.getPointer() != null && nextStart == null ){
            nextStart = currentShard.getPointer();
        }

        Statement query;

        if ( nextStart == null) {

            query = QueryBuilder.select().all().from(QueueMessageSerializationImpl.getTableName(messageType))
                    .where(queueNameClause)
                    .and(regionClause)
                    .and(shardIdClause)
                    .limit(PAGE_SIZE);

        } else {

            Clause messageIdClause = QueryBuilder.gt( COLUMN_QUEUE_MESSAGE_ID, nextStart);
            query = QueryBuilder.select().all().from(QueueMessageSerializationImpl.getTableName(messageType))
                    .where(queueNameClause)
                    .and(regionClause)
                    .and(shardIdClause)
                    .and(messageIdClause)
                    .limit(PAGE_SIZE);
        }


        List<Row> rows = cassandraClient.getQueueMessageSession().execute(query).all();

        logger.trace("results {} from query {}", rows.size(), query.toString());

        if ( (rows == null || rows.size() == 0) && shardIterator.hasNext()) {

            currentShard = shardIterator.next();
            advance();

        } else {

            currentIterator = getIteratorFromRows(rows);

        }
    }
 

@Override
public DatabaseQueueMessage loadMessage(
        final String queueName,
        final String region,
        final Long shardIdOrNull,
        final DatabaseQueueMessage.Type type,
        final UUID queueMessageId ) {

    if ( queueMessageId == null ) {
        return null;
    }

    logger.trace("loadMessage {}", queueMessageId);

    final long shardId;
    if ( shardIdOrNull == null ) {
        Shard.Type shardType = DatabaseQueueMessage.Type.DEFAULT.equals( type ) ?
                Shard.Type.DEFAULT : Shard.Type.INFLIGHT;
        Shard shard = shardStrategy.selectShard(
                queueName, region, shardType, queueMessageId );
        shardId = shard.getShardId();
    } else {
        shardId = shardIdOrNull;
    }

    Clause queueNameClause = QueryBuilder.eq(      COLUMN_QUEUE_NAME, queueName );
    Clause regionClause = QueryBuilder.eq(         COLUMN_REGION, region );
    Clause shardIdClause = QueryBuilder.eq(        COLUMN_SHARD_ID, shardId );
    Clause queueMessageIdClause = QueryBuilder.eq( COLUMN_QUEUE_MESSAGE_ID, queueMessageId);

    Statement select = QueryBuilder.select().from(getTableName( type ))
            .where(queueNameClause)
            .and(regionClause)
            .and(shardIdClause)
            .and(queueMessageIdClause);

    Row row = cassandraClient.getQueueMessageSession().execute(select).one();

    if (row == null) {
        return null;
    }

    return new DatabaseQueueMessage(
        row.getUUID(   COLUMN_MESSAGE_ID),
        type,
        row.getString( COLUMN_QUEUE_NAME),
        row.getString( COLUMN_REGION),
        row.getLong(   COLUMN_SHARD_ID),
        row.getLong(   COLUMN_QUEUED_AT),
        row.getLong(   COLUMN_INFLIGHT_AT),
        row.getUUID(   COLUMN_QUEUE_MESSAGE_ID)
    );
}