* 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.

*/

* Encapsulates a completely parsed SELECT query, including the target

* column family, expression, result count, and ordering clause.

*

*/

    private static final Logger logger = LoggerFactory.getLogger(SelectStatement.class);

    private static final int DEFAULT_COUNT_PAGE_SIZE = 10000;

    private final int boundTerms;

    public final CFDefinition cfDef;

    public final Parameters parameters;

    private final Selection selection;

    private final Term limit;

    private final Restriction[] keyRestrictions;

    private final Restriction[] columnRestrictions;

    private final Map<CFDefinition.Name, Restriction> metadataRestrictions = new HashMap<CFDefinition.Name, Restriction>();

    // The name of all restricted names not covered by the key or index filter

    private final Set<CFDefinition.Name> restrictedNames = new HashSet<CFDefinition.Name>();

    private Restriction.Slice sliceRestriction;

    private boolean isReversed;

    private boolean onToken;

    private boolean isKeyRange;

    private boolean keyIsInRelation;

    private boolean usesSecondaryIndexing;

    private Map<CFDefinition.Name, Integer> orderingIndexes;

    private boolean selectsStaticColumns;

    private boolean selectsOnlyStaticColumns;

    // Used by forSelection below

    private static final Parameters defaultParameters = new Parameters(Collections.<ColumnIdentifier, Boolean>emptyMap(), false, false, null, false);

    private static final Predicate<CFDefinition.Name> isStaticFilter = new Predicate<CFDefinition.Name>()

    {

        public boolean apply(CFDefinition.Name name)

        {

            return name.kind == CFDefinition.Name.Kind.STATIC;

        }

    };

    public SelectStatement(CFDefinition cfDef, int boundTerms, Parameters parameters, Selection selection, Term limit)

    {

        this.cfDef = cfDef;

        this.boundTerms = boundTerms;

        this.selection = selection;

        this.keyRestrictions = new Restriction[cfDef.partitionKeyCount()];

        this.columnRestrictions = new Restriction[cfDef.clusteringColumnsCount()];

        this.parameters = parameters;

        this.limit = limit;

        // Now gather a few info on whether we should bother with static columns or not for this statement

        initStaticColumnsInfo();

    }

    private void initStaticColumnsInfo()

    {

        if (!cfDef.cfm.hasStaticColumns())

            return;

        // If it's a wildcard, we do select static but not only them

        if (selection.isWildcard())

        {

            selectsStaticColumns = true;

            return;

        }

        // Otherwise, check the selected columns

        selectsStaticColumns = !Iterables.isEmpty(Iterables.filter(selection.getColumns(), isStaticFilter));

        selectsOnlyStaticColumns = true;

        for (CFDefinition.Name name : selection.getColumns())

        {

            if (name.kind != CFDefinition.Name.Kind.KEY_ALIAS && name.kind != CFDefinition.Name.Kind.STATIC)

            {

                selectsOnlyStaticColumns = false;

                break;

            }

        }

    }

    // Creates a simple select based on the given selection.

    // Note that the results select statement should not be used for actual queries, but only for processing already

    // queried data through processColumnFamily.

    static SelectStatement forSelection(CFDefinition cfDef, Selection selection)

    {

        return new SelectStatement(cfDef, 0, defaultParameters, selection, null);

    }

    public ResultSet.Metadata getResultMetadata()

    {

        return parameters.isCount

             ? ResultSet.makeCountMetadata(keyspace(), columnFamily(), parameters.countAlias)

             : selection.getResultMetadata();

    }

    public long measureForPreparedCache(MemoryMeter meter)

    {

        return meter.measure(this)

             + meter.measureDeep(parameters)

             + meter.measureDeep(selection)

             + (limit == null ? 0 : meter.measureDeep(limit))

             + meter.measureDeep(keyRestrictions)

             + meter.measureDeep(columnRestrictions)

             + meter.measureDeep(metadataRestrictions)

             + meter.measureDeep(restrictedNames)

             + (sliceRestriction == null ? 0 : meter.measureDeep(sliceRestriction))

             + (orderingIndexes == null ? 0 : meter.measureDeep(orderingIndexes));

    }

    public int getBoundTerms()

    {

        return boundTerms;

    }

    public void checkAccess(ClientState state) throws InvalidRequestException, UnauthorizedException

    {

        state.hasColumnFamilyAccess(keyspace(), columnFamily(), Permission.SELECT);

    }

    public void validate(ClientState state) throws InvalidRequestException

    {

        // Nothing to do, all validation has been done by RawStatement.prepare()

    }

    public ResultMessage.Rows execute(QueryState state, QueryOptions options) throws RequestExecutionException, RequestValidationException

    {

        ConsistencyLevel cl = options.getConsistency();

        List<ByteBuffer> variables = options.getValues();

        if (cl == null)

            throw new InvalidRequestException("Invalid empty consistency level");

        cl.validateForRead(keyspace());

        int limit = getLimit(variables);

        int limitForQuery = updateLimitForQuery(limit);

        long now = System.currentTimeMillis();

        Pageable command;

        if (isKeyRange || usesSecondaryIndexing)

        {

            command = getRangeCommand(variables, limitForQuery, now);

        }

        else

        {

            List<ReadCommand> commands = getSliceCommands(variables, limitForQuery, now);

            command = commands == null ? null : new Pageable.ReadCommands(commands);

        }

        int pageSize = options.getPageSize();

        // A count query will never be paged for the user, but we always page it internally to avoid OOM.

        // If we user provided a pageSize we'll use that to page internally (because why not), otherwise we use our default

        // Note that if there are some nodes in the cluster with a version less than 2.0, we can't use paging (CASSANDRA-6707).

        if (parameters.isCount && pageSize <= 0 && MessagingService.instance().allNodesAtLeast20)

            pageSize = DEFAULT_COUNT_PAGE_SIZE;

        if (pageSize <= 0 || command == null || !QueryPagers.mayNeedPaging(command, pageSize))

        {

            return execute(command, cl, variables, limit, now);

        }

        else

        {

            QueryPager pager = QueryPagers.pager(command, cl, options.getPagingState());

            if (parameters.isCount)

                return pageCountQuery(pager, variables, pageSize, now, limit);

            // We can't properly do post-query ordering if we page (see #6722)

            if (needsPostQueryOrdering())

                throw new InvalidRequestException("Cannot page queries with both ORDER BY and a IN restriction on the partition key; you must either remove the "

                                                + "ORDER BY or the IN and sort client side, or disable paging for this query");

            List<Row> page = pager.fetchPage(pageSize);

            ResultMessage.Rows msg = processResults(page, variables, limit, now);

            if (!pager.isExhausted())

                msg.result.metadata.setHasMorePages(pager.state());

            return msg;

        }

    }

    private ResultMessage.Rows execute(Pageable command, ConsistencyLevel cl, List<ByteBuffer> variables, int limit, long now) throws RequestValidationException, RequestExecutionException

    {

        List<Row> rows;

        if (command == null)

        {

            rows = Collections.<Row>emptyList();

        }

        else

        {

            rows = command instanceof Pageable.ReadCommands

                 ? StorageProxy.read(((Pageable.ReadCommands)command).commands, cl)

                 : StorageProxy.getRangeSlice((RangeSliceCommand)command, cl);

        }

        return processResults(rows, variables, limit, now);

    }

    private ResultMessage.Rows pageCountQuery(QueryPager pager, List<ByteBuffer> variables, int pageSize, long now, int limit) throws RequestValidationException, RequestExecutionException

    {

        int count = 0;

        while (!pager.isExhausted())

        {

            int maxLimit = pager.maxRemaining();

            logger.debug("New maxLimit for paged count query is {}", maxLimit);

            ResultSet rset = process(pager.fetchPage(pageSize), variables, maxLimit, now);

            count += rset.rows.size();

        }

        // We sometimes query one more result than the user limit asks to handle exclusive bounds with compact tables (see updateLimitForQuery).

        // So do make sure the count is not greater than what the user asked for.

        ResultSet result = ResultSet.makeCountResult(keyspace(), columnFamily(), Math.min(count, limit), parameters.countAlias);

        return new ResultMessage.Rows(result);

    }

    public ResultMessage.Rows processResults(List<Row> rows, List<ByteBuffer> variables, int limit, long now) throws RequestValidationException

    {

        // Even for count, we need to process the result as it'll group some column together in sparse column families

        ResultSet rset = process(rows, variables, limit, now);

        rset = parameters.isCount ? rset.makeCountResult(parameters.countAlias) : rset;

        return new ResultMessage.Rows(rset);

    }

    static List<Row> readLocally(String keyspaceName, List<ReadCommand> cmds)

    {

        Keyspace keyspace = Keyspace.open(keyspaceName);

        List<Row> rows = new ArrayList<Row>(cmds.size());

        for (ReadCommand cmd : cmds)

            rows.add(cmd.getRow(keyspace));

        return rows;

    }

    public ResultMessage.Rows executeInternal(QueryState state) throws RequestExecutionException, RequestValidationException

    {

        List<ByteBuffer> variables = Collections.emptyList();

        int limit = getLimit(variables);

        int limitForQuery = updateLimitForQuery(limit);

        long now = System.currentTimeMillis();

        List<Row> rows;

        if (isKeyRange || usesSecondaryIndexing)

        {

            RangeSliceCommand command = getRangeCommand(variables, limitForQuery, now);

            rows = command == null ? Collections.<Row>emptyList() : command.executeLocally();

        }

        else

        {

            List<ReadCommand> commands = getSliceCommands(variables, limitForQuery, now);

            rows = commands == null ? Collections.<Row>emptyList() : readLocally(keyspace(), commands);

        }

        return processResults(rows, variables, limit, now);

    }

    public ResultSet process(List<Row> rows) throws InvalidRequestException

    {

        assert !parameters.isCount; // not yet needed

        return process(rows, Collections.<ByteBuffer>emptyList(), getLimit(Collections.<ByteBuffer>emptyList()), System.currentTimeMillis());

    }

    public String keyspace()

    {

        return cfDef.cfm.ksName;

    }

    public String columnFamily()

    {

        return cfDef.cfm.cfName;

    }

    private List<ReadCommand> getSliceCommands(List<ByteBuffer> variables, int limit, long now) throws RequestValidationException

    {

        Collection<ByteBuffer> keys = getKeys(variables);

        if (keys.isEmpty()) // in case of IN () for (the last column of) the partition key.

            return null;

        List<ReadCommand> commands = new ArrayList<ReadCommand>(keys.size());

        IDiskAtomFilter filter = makeFilter(variables, limit);

        if (filter == null)

            return null;

        // Note that we use the total limit for every key, which is potentially inefficient.

        // However, IN + LIMIT is not a very sensible choice.

        for (ByteBuffer key : keys)

        {

            QueryProcessor.validateKey(key);

            // We should not share the slice filter amongst the commands (hence the cloneShallow), due to

            // SliceQueryFilter not being immutable due to its columnCounter used by the lastCounted() method

            // (this is fairly ugly and we should change that but that's probably not a tiny refactor to do that cleanly)

            commands.add(ReadCommand.create(keyspace(), key, columnFamily(), now, filter.cloneShallow()));

        }

        return commands;

    }

    private RangeSliceCommand getRangeCommand(List<ByteBuffer> variables, int limit, long now) throws RequestValidationException

    {

        IDiskAtomFilter filter = makeFilter(variables, limit);

        if (filter == null)

            return null;

        List<IndexExpression> expressions = getIndexExpressions(variables);

        // The LIMIT provided by the user is the number of CQL row he wants returned.

        // We want to have getRangeSlice to count the number of columns, not the number of keys.

        AbstractBounds<RowPosition> keyBounds = getKeyBounds(variables);

        return keyBounds == null

             ? null

             : new RangeSliceCommand(keyspace(), columnFamily(), now,  filter, keyBounds, expressions, limit, !parameters.isDistinct, false);

    }

    private AbstractBounds<RowPosition> getKeyBounds(List<ByteBuffer> variables) throws InvalidRequestException

    {

        IPartitioner<?> p = StorageService.getPartitioner();

        if (onToken)

        {

            Token startToken = getTokenBound(Bound.START, variables, p);

            Token endToken = getTokenBound(Bound.END, variables, p);

            boolean includeStart = includeKeyBound(Bound.START);

            boolean includeEnd = includeKeyBound(Bound.END);

            /*

             * If we ask SP.getRangeSlice() for (token(200), token(200)], it will happily return the whole ring.

             * However, wrapping range doesn't really make sense for CQL, and we want to return an empty result

             * in that case (CASSANDRA-5573). So special case to create a range that is guaranteed to be empty.

             *

             * In practice, we want to return an empty result set if either startToken > endToken, or both are

             * equal but one of the bound is excluded (since [a, a] can contains something, but not (a, a], [a, a)

             * or (a, a)). Note though that in the case where startToken or endToken is the minimum token, then

             * this special case rule should not apply.

             */

            int cmp = startToken.compareTo(endToken);

            if (!startToken.isMinimum() && !endToken.isMinimum() && (cmp > 0 || (cmp == 0 && (!includeStart || !includeEnd))))

                return null;

            RowPosition start = includeStart ? startToken.minKeyBound() : startToken.maxKeyBound();

            RowPosition end = includeEnd ? endToken.maxKeyBound() : endToken.minKeyBound();

            return new Range<RowPosition>(start, end);

        }

        else

        {

            ByteBuffer startKeyBytes = getKeyBound(Bound.START, variables);

            ByteBuffer finishKeyBytes = getKeyBound(Bound.END, variables);

            RowPosition startKey = RowPosition.forKey(startKeyBytes, p);

            RowPosition finishKey = RowPosition.forKey(finishKeyBytes, p);

            if (startKey.compareTo(finishKey) > 0 && !finishKey.isMinimum(p))

                return null;

            if (includeKeyBound(Bound.START))

            {

                return includeKeyBound(Bound.END)

                     ? new Bounds<RowPosition>(startKey, finishKey)

                     : new IncludingExcludingBounds<RowPosition>(startKey, finishKey);

            }

            else

            {

                return includeKeyBound(Bound.END)

                     ? new Range<RowPosition>(startKey, finishKey)

                     : new ExcludingBounds<RowPosition>(startKey, finishKey);

            }

        }

    }

    private IDiskAtomFilter makeFilter(List<ByteBuffer> variables, int limit)

    throws InvalidRequestException

    {

        if (parameters.isDistinct)

        {

            return new SliceQueryFilter(ColumnSlice.ALL_COLUMNS_ARRAY, false, 1, -1);

        }

        else if (isColumnRange())

        {

            // For sparse, we used to ask for 'defined columns' * 'asked limit' (where defined columns includes the row marker)

            // to account for the grouping of columns.

            // Since that doesn't work for maps/sets/lists, we now use the compositesToGroup option of SliceQueryFilter.

            // But we must preserve backward compatibility too (for mixed version cluster that is).

            int toGroup = cfDef.isCompact ? -1 : cfDef.clusteringColumnsCount();

            List<ByteBuffer> startBounds = getRequestedBound(Bound.START, variables);

            List<ByteBuffer> endBounds = getRequestedBound(Bound.END, variables);

            assert startBounds.size() == endBounds.size();

            // Handles fetching static columns. Note that for 2i, the filter is just used to restrict

            // the part of the index to query so adding the static slice would be useless and confusing.

            // For 2i, static columns are retrieve in CompositesSearcher with each index hit.

            ColumnSlice staticSlice = null;

            if (selectsStaticColumns && !usesSecondaryIndexing)

            {

                ColumnNameBuilder staticPrefix = cfDef.cfm.getStaticColumnNameBuilder();

                // Note: we could use staticPrefix.build() for the start bound, but EMPTY_BYTE_BUFFER gives us the

                // same effect while saving a few CPU cycles.

                staticSlice = isReversed

                            ? new ColumnSlice(staticPrefix.buildAsEndOfRange(), ByteBufferUtil.EMPTY_BYTE_BUFFER)

                            : new ColumnSlice(ByteBufferUtil.EMPTY_BYTE_BUFFER, staticPrefix.buildAsEndOfRange());

                // In the case where we only select static columns, we want to really only check the static columns.

                // So we return early as the rest of that method would actually make us query everything

                if (selectsOnlyStaticColumns)

                    return sliceFilter(staticSlice, limit, toGroup);

            }

            // The case where startBounds == 1 is common enough that it's worth optimizing

            if (startBounds.size() == 1)

            {

                ColumnSlice slice = new ColumnSlice(startBounds.get(0), endBounds.get(0));

                if (slice.isAlwaysEmpty(cfDef.cfm.comparator, isReversed))

                    return staticSlice == null ? null : sliceFilter(staticSlice, limit, toGroup);

                return staticSlice == null

                     ? sliceFilter(slice, limit, toGroup)

                     : (slice.includes(cfDef.cfm.comparator, staticSlice.finish) ? sliceFilter(new ColumnSlice(staticSlice.start, slice.finish), limit, toGroup)

                                                                                 : sliceFilter(new ColumnSlice[]{ staticSlice, slice }, limit, toGroup));

            }

            List<ColumnSlice> l = new ArrayList<ColumnSlice>(startBounds.size());

            for (int i = 0; i < startBounds.size(); i++)

            {

                ColumnSlice slice = new ColumnSlice(startBounds.get(i), endBounds.get(i));

                if (!slice.isAlwaysEmpty(cfDef.cfm.comparator, isReversed))

                    l.add(slice);

            }

            if (l.isEmpty())

                return staticSlice == null ? null : sliceFilter(staticSlice, limit, toGroup);

            if (staticSlice == null)

                return sliceFilter(l.toArray(new ColumnSlice[l.size()]), limit, toGroup);

            // The slices should not overlap. We know the slices built from startBounds/endBounds don't, but

            // if there is a static slice, it could overlap with the 2nd slice. Check for it and correct if

            // that's the case

            ColumnSlice[] slices;

            if (isReversed)

            {

                if (l.get(l.size() - 1).includes(cfDef.cfm.comparator, staticSlice.start))

                {

                    slices = l.toArray(new ColumnSlice[l.size()]);

                    slices[slices.length-1] = new ColumnSlice(slices[slices.length-1].start, ByteBufferUtil.EMPTY_BYTE_BUFFER);

                }

                else

                {

                    slices = l.toArray(new ColumnSlice[l.size()+1]);

                    slices[slices.length-1] = staticSlice;

                }

            }

            else

            {

                if (l.get(0).includes(cfDef.cfm.comparator, staticSlice.finish))

                {

                    slices = new ColumnSlice[l.size()];

                    slices[0] = new ColumnSlice(ByteBufferUtil.EMPTY_BYTE_BUFFER, l.get(0).finish);

                    for (int i = 1; i < l.size(); i++)

                        slices[i] = l.get(i);

                }

                else

                {

                    slices = new ColumnSlice[l.size()+1];

                    slices[0] = staticSlice;

                    for (int i = 0; i < l.size(); i++)

                        slices[i+1] = l.get(i);

                }

            }

            return sliceFilter(slices, limit, toGroup);

        }

        else

        {

            SortedSet<ByteBuffer> cellNames = getRequestedColumns(variables);

            if (cellNames == null) // in case of IN () for the last column of the key

                return null;

            QueryProcessor.validateCellNames(cellNames);

            return new NamesQueryFilter(cellNames, true);

        }

    }

    private SliceQueryFilter sliceFilter(ColumnSlice slice, int limit, int toGroup)

    {

        return sliceFilter(new ColumnSlice[]{ slice }, limit, toGroup);

    }

    private SliceQueryFilter sliceFilter(ColumnSlice[] slices, int limit, int toGroup)

    {

        return new SliceQueryFilter(slices, isReversed, limit, toGroup);

    }

    private int getLimit(List<ByteBuffer> variables) throws InvalidRequestException

    {

        int l = Integer.MAX_VALUE;

        if (limit != null)

        {

            ByteBuffer b = limit.bindAndGet(variables);

            if (b == null)

                throw new InvalidRequestException("Invalid null value of limit");

            try

            {

                Int32Type.instance.validate(b);

                l = Int32Type.instance.compose(b);

            }

            catch (MarshalException e)

            {

                throw new InvalidRequestException("Invalid limit value");

            }

        }

        if (l <= 0)

            throw new InvalidRequestException("LIMIT must be strictly positive");

        return l;

    }

    private int updateLimitForQuery(int limit)

    {

        // Internally, we don't support exclusive bounds for slices. Instead, we query one more element if necessary

        // and exclude it later (in processColumnFamily)

        return sliceRestriction != null && (!sliceRestriction.isInclusive(Bound.START) || !sliceRestriction.isInclusive(Bound.END)) && limit != Integer.MAX_VALUE

             ? limit + 1

             : limit;

    }

    private Collection<ByteBuffer> getKeys(final List<ByteBuffer> variables) throws InvalidRequestException

    {

        List<ByteBuffer> keys = new ArrayList<ByteBuffer>();

        ColumnNameBuilder builder = cfDef.getKeyNameBuilder();

        for (CFDefinition.Name name : cfDef.partitionKeys())

        {

            Restriction r = keyRestrictions[name.position];

            assert r != null && !r.isSlice();

            List<ByteBuffer> values = r.values(variables);

            if (builder.remainingCount() == 1)

            {

                for (ByteBuffer val : values)

                {

                    if (val == null)

                        throw new InvalidRequestException(String.format("Invalid null value for partition key part %s", name));

                    keys.add(builder.copy().add(val).build());

                }

            }

            else

            {

                // Note: for backward compatibility reasons, we let INs with 1 value slide

                if (values.size() != 1)

                    throw new InvalidRequestException("IN is only supported on the last column of the partition key");

                ByteBuffer val = values.get(0);

                if (val == null)

                    throw new InvalidRequestException(String.format("Invalid null value for partition key part %s", name));

                builder.add(val);

            }

        }

        return keys;

    }

    private ByteBuffer getKeyBound(Bound b, List<ByteBuffer> variables) throws InvalidRequestException

    {

        // Deal with unrestricted partition key components (special-casing is required to deal with 2i queries on the first

        // component of a composite partition key).

        for (int i = 0; i < keyRestrictions.length; i++)

            if (keyRestrictions[i] == null)

                return ByteBufferUtil.EMPTY_BYTE_BUFFER;

        // We deal with IN queries for keys in other places, so we know buildBound will return only one result

        return buildBound(b, cfDef.partitionKeys(), keyRestrictions, false, cfDef.getKeyNameBuilder(), variables).get(0);

    }

    private Token getTokenBound(Bound b, List<ByteBuffer> variables, IPartitioner<?> p) throws InvalidRequestException

    {

        assert onToken;

        Restriction keyRestriction = keyRestrictions[0];

        ByteBuffer value;

        if (keyRestriction.isEQ())

        {

            value = keyRestriction.values(variables).get(0);

        }

        else

        {

            Restriction.Slice slice = (Restriction.Slice)keyRestriction;

            if (!slice.hasBound(b))

                return p.getMinimumToken();

            value = slice.bound(b, variables);

        }

        if (value == null)

            throw new InvalidRequestException("Invalid null token value");

        return p.getTokenFactory().fromByteArray(value);

    }

    private boolean includeKeyBound(Bound b)

    {

        for (Restriction r : keyRestrictions)

        {

            if (r == null)

                return true;

            else if (r.isSlice())

                return ((Restriction.Slice)r).isInclusive(b);

        }

        // All equality

        return true;

    }

    private boolean isColumnRange()

    {

        // Due to CASSANDRA-5762, we always do a slice for CQL3 tables (not compact, composite).

        // Static CF (non compact but non composite) never entails a column slice however

        if (!cfDef.isCompact)

            return cfDef.isComposite;

        // Otherwise (i.e. for compact table where we don't have a row marker anyway and thus don't care about CASSANDRA-5762),

        // it is a range query if it has at least one the column alias for which no relation is defined or is not EQ.

        for (Restriction r : columnRestrictions)

        {

            if (r == null || r.isSlice())

                return true;

        }

        return false;

    }

    private SortedSet<ByteBuffer> getRequestedColumns(List<ByteBuffer> variables) throws InvalidRequestException

    {

        // Note: getRequestedColumns don't handle static columns, but due to CASSANDRA-5762

        // we always do a slice for CQL3 tables, so it's ok to ignore them here

        assert !isColumnRange();

        ColumnNameBuilder builder = cfDef.getColumnNameBuilder();

        Iterator<CFDefinition.Name> idIter = cfDef.clusteringColumns().iterator();

        for (Restriction r : columnRestrictions)

        {

            ColumnIdentifier id = idIter.next().name;

            assert r != null && !r.isSlice();

            List<ByteBuffer> values = r.values(variables);

            if (values.size() == 1)

            {

                ByteBuffer val = values.get(0);

                if (val == null)

                    throw new InvalidRequestException(String.format("Invalid null value for clustering key part %s", id));

                builder.add(val);

            }

            else

            {

                // We have a IN, which we only support for the last column.

                // If compact, just add all values and we're done. Otherwise,

                // for each value of the IN, creates all the columns corresponding to the selection.

                if (values.isEmpty())

                    return null;

                SortedSet<ByteBuffer> columns = new TreeSet<ByteBuffer>(cfDef.cfm.comparator);

                Iterator<ByteBuffer> iter = values.iterator();

                while (iter.hasNext())

                {

                    ByteBuffer val = iter.next();

                    ColumnNameBuilder b = iter.hasNext() ? builder.copy() : builder;

                    if (val == null)

                        throw new InvalidRequestException(String.format("Invalid null value for clustering key part %s", id));

                    b.add(val);

                    if (cfDef.isCompact)

                        columns.add(b.build());

                    else

                        columns.addAll(addSelectedColumns(b));

                }

                return columns;

            }

        }

        return addSelectedColumns(builder);

    }

    private SortedSet<ByteBuffer> addSelectedColumns(ColumnNameBuilder builder)

    {

        if (cfDef.isCompact)

        {

            return FBUtilities.singleton(builder.build(), cfDef.cfm.comparator);

        }

        else

        {

            // Collections require doing a slice query because a given collection is a

            // non-know set of columns, so we shouldn't get there

            assert !selectACollection();

            SortedSet<ByteBuffer> columns = new TreeSet<ByteBuffer>(cfDef.cfm.comparator);

            // We need to query the selected column as well as the marker

            // column (for the case where the row exists but has no columns outside the PK)

            // Two exceptions are "static CF" (non-composite non-compact CF) and "super CF"

            // that don't have marker and for which we must query all columns instead

            if (cfDef.isComposite && !cfDef.cfm.isSuper())

            {

                // marker

                columns.add(builder.copy().add(ByteBufferUtil.EMPTY_BYTE_BUFFER).build());

                // selected columns

                for (ColumnIdentifier id : selection.regularAndStaticColumnsToFetch())

                    columns.add(builder.copy().add(id.key).build());

            }

            else

            {

                // We now that we're not composite so we can ignore static columns

                Iterator<CFDefinition.Name> iter = cfDef.regularColumns().iterator();

                while (iter.hasNext())

                {

                    ColumnIdentifier name = iter.next().name;

                    ColumnNameBuilder b = iter.hasNext() ? builder.copy() : builder;

                    ByteBuffer cname = b.add(name.key).build();

                    columns.add(cname);

                }

            }

            return columns;

        }

    }

    private boolean selectACollection()

    {

        if (!cfDef.hasCollections)

            return false;

        for (CFDefinition.Name name : selection.getColumns())

        {

            if (name.type instanceof CollectionType)

                return true;

        }

        return false;

    }

    private List<ByteBuffer> buildBound(Bound bound,

                                        Collection<CFDefinition.Name> names,

                                        Restriction[] restrictions,

                                        boolean isReversed,

                                        ColumnNameBuilder builder,

                                        List<ByteBuffer> variables) throws InvalidRequestException

    {

        // The end-of-component of composite doesn't depend on whether the

        // component type is reversed or not (i.e. the ReversedType is applied

        // to the component comparator but not to the end-of-component itself),

        // it only depends on whether the slice is reversed

        Bound eocBound = isReversed ? Bound.reverse(bound) : bound;

        for (Iterator<CFDefinition.Name> iter = names.iterator(); iter.hasNext();)

        {

            CFDefinition.Name name = iter.next();

            // In a restriction, we always have Bound.START < Bound.END for the "base" comparator.

            // So if we're doing a reverse slice, we must inverse the bounds when giving them as start and end of the slice filter.

            // But if the actual comparator itself is reversed, we must inversed the bounds too.

            Bound b = isReversed == isReversedType(name) ? bound : Bound.reverse(bound);

            Restriction r = restrictions[name.position];

            if (isNullRestriction(r, b))

            {

                // There wasn't any non EQ relation on that key, we select all records having the preceding component as prefix.

                // For composites, if there was preceding component and we're computing the end, we must change the last component

                // End-Of-Component, otherwise we would be selecting only one record.

                return Collections.singletonList(builder.componentCount() > 0 && eocBound == Bound.END

                                                 ? builder.buildAsEndOfRange()

                                                 : builder.build());

            }

            if (r.isSlice())

            {

                builder.add(getSliceValue(name, r, b, variables));

                Relation.Type relType = ((Restriction.Slice)r).getRelation(eocBound, b);

                // We can have more non null restriction if the "scalar" notation was used for the bound (#4851).

                // In that case, we need to add them all, and end the cell name with the correct end-of-component.

                while (iter.hasNext())

                {

                    name = iter.next();

                    r = restrictions[name.position];

                    if (isNullRestriction(r, b))

                        break;

                    builder.add(getSliceValue(name, r, b, variables));

                }

                return Collections.singletonList(builder.buildForRelation(relType));

            }

            else

            {

                List<ByteBuffer> values = r.values(variables);

                if (values.size() != 1)

                {

                    // IN query, we only support it on the clustering column

                    assert name.position == names.size() - 1;

                    // The IN query might not have listed the values in comparator order, so we need to re-sort

                    // the bounds lists to make sure the slices works correctly (also, to avoid duplicates).

                    TreeSet<ByteBuffer> s = new TreeSet<ByteBuffer>(isReversed ? cfDef.cfm.comparator.reverseComparator : cfDef.cfm.comparator);

                    for (ByteBuffer val : values)

                    {

                        if (val == null)

                            throw new InvalidRequestException(String.format("Invalid null clustering key part %s", name));

                        ColumnNameBuilder copy = builder.copy().add(val);

                        // See below for why this

                        s.add((b == Bound.END && copy.remainingCount() > 0) ? copy.buildAsEndOfRange() : copy.build());

                    }

                    return new ArrayList<ByteBuffer>(s);

                }

                ByteBuffer val = values.get(0);

                if (val == null)

                    throw new InvalidRequestException(String.format("Invalid null clustering key part %s", name));

                builder.add(val);

            }

        }

        // Means no relation at all or everything was an equal

        // Note: if the builder is "full", there is no need to use the end-of-component bit. For columns selection,

        // it would be harmless to do it. However, we use this method got the partition key too. And when a query

        // with 2ndary index is done, and with the the partition provided with an EQ, we'll end up here, and in that

        // case using the eoc would be bad, since for the random partitioner we have no guarantee that

        // builder.buildAsEndOfRange() will sort after builder.build() (see #5240).

        return Collections.singletonList((bound == Bound.END && builder.remainingCount() > 0) ? builder.buildAsEndOfRange() : builder.build());

    }

    private static boolean isNullRestriction(Restriction r, Bound b)

    {

        return r == null || (r.isSlice() && !((Restriction.Slice)r).hasBound(b));

    }

    private static ByteBuffer getSliceValue(CFDefinition.Name name, Restriction r, Bound b, List<ByteBuffer> variables) throws InvalidRequestException

    {

        Restriction.Slice slice = (Restriction.Slice)r;

        assert slice.hasBound(b);

        ByteBuffer val = slice.bound(b, variables);

        if (val == null)

            throw new InvalidRequestException(String.format("Invalid null clustering key part %s", name));

        return val;

    }

    private List<ByteBuffer> getRequestedBound(Bound b, List<ByteBuffer> variables) throws InvalidRequestException

    {

        assert isColumnRange();

        return buildBound(b, cfDef.clusteringColumns(), columnRestrictions, isReversed, cfDef.getColumnNameBuilder(), variables);

    }

    public List<IndexExpression> getIndexExpressions(List<ByteBuffer> variables) throws InvalidRequestException

    {

        if (!usesSecondaryIndexing || restrictedNames.isEmpty())

            return Collections.emptyList();

        List<IndexExpression> expressions = new ArrayList<IndexExpression>();

        for (CFDefinition.Name name : restrictedNames)

        {

            Restriction restriction;

            switch (name.kind)

            {

                case KEY_ALIAS:

                    restriction = keyRestrictions[name.position];

                    break;

                case COLUMN_ALIAS:

                    restriction = columnRestrictions[name.position];

                    break;

                case COLUMN_METADATA:

                case STATIC:

                    restriction = metadataRestrictions.get(name);

                    break;

                default:

                    // We don't allow restricting a VALUE_ALIAS for now in prepare.

                    throw new AssertionError();

            }

            if (restriction.isSlice())

            {

                Restriction.Slice slice = (Restriction.Slice)restriction;

                for (Bound b : Bound.values())

                {

                    if (slice.hasBound(b))

                    {

                        ByteBuffer value = slice.bound(b, variables);

                        if (value == null)

                            throw new InvalidRequestException(String.format("Unsupported null value for indexed column %s", name));

                        if (value.remaining() > 0xFFFF)

                            throw new InvalidRequestException("Index expression values may not be larger than 64K");

                        IndexOperator op = slice.getIndexOperator(b);

                        // If the underlying comparator for name is reversed, we need to reverse the IndexOperator: user operation

                        // always refer to the "forward" sorting even if the clustering order is reversed, but the 2ndary code does

                        // use the underlying comparator as is.

                        if (name.type instanceof ReversedType)

                            op = reverse(op);

                        expressions.add(new IndexExpression(name.name.key, op, value));

                    }

                }

            }

            else

            {

                List<ByteBuffer> values = restriction.values(variables);

                if (values.size() != 1)

                    throw new InvalidRequestException("IN restrictions are not supported on indexed columns");

                ByteBuffer value = values.get(0);

                if (value == null)

                    throw new InvalidRequestException(String.format("Unsupported null value for indexed column %s", name));

                if (value.remaining() > 0xFFFF)

                    throw new InvalidRequestException("Index expression values may not be larger than 64K");

                expressions.add(new IndexExpression(name.name.key, IndexOperator.EQ, value));

            }

        }

        return expressions;

    }

    private static IndexOperator reverse(IndexOperator op)

    {

        switch (op)

        {

            case LT:  return IndexOperator.GT;

            case LTE: return IndexOperator.GTE;

            case GT:  return IndexOperator.LT;

            case GTE: return IndexOperator.LTE;

            default: return op;

        }

    }

    private ResultSet process(List<Row> rows, List<ByteBuffer> variables, int limit, long now) throws InvalidRequestException

    {

        Selection.ResultSetBuilder result = selection.resultSetBuilder(now);

        for (org.apache.cassandra.db.Row row : rows)

        {

            // Not columns match the query, skip

            if (row.cf == null)

                continue;

            processColumnFamily(row.key.key, row.cf, variables, now, result);

        }

        ResultSet cqlRows = result.build();

        orderResults(cqlRows);

        // Internal calls always return columns in the comparator order, even when reverse was set

        if (isReversed)

            cqlRows.reverse();

        // Trim result if needed to respect the user limit

        cqlRows.trim(limit);

        return cqlRows;

    }

    // Used by ModificationStatement for CAS operations

    void processColumnFamily(ByteBuffer key, ColumnFamily cf, List<ByteBuffer> variables, long now, Selection.ResultSetBuilder result)

    throws InvalidRequestException

    {

        ByteBuffer[] keyComponents = cfDef.hasCompositeKey

                                   ? ((CompositeType)cfDef.cfm.getKeyValidator()).split(key)

                                   : new ByteBuffer[]{ key };

        if (parameters.isDistinct)

        {

            if (!cf.hasOnlyTombstones(now))

            {

                result.newRow();

                // selection.getColumns() will contain only the partition key components - all of them.

                for (CFDefinition.Name name : selection.getColumns())

                    result.add(keyComponents[name.position]);

            }

        }

        else if (cfDef.isCompact)

        {

            // One cqlRow per column

            for (Column c : cf)

            {

                if (c.isMarkedForDelete(now))

                    continue;

                ByteBuffer[] components = null;

                if (cfDef.isComposite)

                {

                    components = ((CompositeType)cfDef.cfm.comparator).split(c.name());

                }

                else if (sliceRestriction != null)

                {

                    Comparator<ByteBuffer> comp = cfDef.cfm.comparator;

                    // For dynamic CF, the column could be out of the requested bounds, filter here

                    if (!sliceRestriction.isInclusive(Bound.START) && comp.compare(c.name(), sliceRestriction.bound(Bound.START, variables)) == 0)

                        continue;

                    if (!sliceRestriction.isInclusive(Bound.END) && comp.compare(c.name(), sliceRestriction.bound(Bound.END, variables)) == 0)

                        continue;

                }

                result.newRow();

                // Respect selection order

                for (CFDefinition.Name name : selection.getColumns())

                {

                    switch (name.kind)

                    {

                        case KEY_ALIAS:

                            result.add(keyComponents[name.position]);

                            break;

                        case COLUMN_ALIAS:

                            ByteBuffer val = cfDef.isComposite

                                           ? (name.position < components.length ? components[name.position] : null)

                                           : c.name();

                            result.add(val);

                            break;

                        case VALUE_ALIAS:

                            result.add(c);

                            break;

                        case COLUMN_METADATA:

                        case STATIC:

                            // This should not happen for compact CF

                            throw new AssertionError();

                        default:

                            throw new AssertionError();

                    }

                }

            }

        }

        else if (cfDef.isComposite)

        {

            // Sparse case: group column in cqlRow when composite prefix is equal

            CompositeType composite = (CompositeType)cfDef.cfm.comparator;

            ColumnGroupMap.Builder builder = new ColumnGroupMap.Builder(composite, cfDef.hasCollections, now);

            for (Column c : cf)

            {

                if (c.isMarkedForDelete(now))

                    continue;

                builder.add(c);

            }

            ColumnGroupMap staticGroup = null;

            // Gather up static values first

            if (!builder.isEmpty() && builder.firstGroup().isStatic)

            {

                staticGroup = builder.firstGroup();

                builder.discardFirst();

                // If there was static columns but there is no actual row, then provided the select was a full

                // partition selection (i.e. not a 2ndary index search and there was no condition on clustering columns)

                // then we want to include the static columns in the result set.

                if (builder.isEmpty() && !usesSecondaryIndexing && hasNoClusteringColumnsRestriction() && hasValueForQuery(staticGroup))

                {

                    handleGroup(result, keyComponents, ColumnGroupMap.EMPTY, staticGroup);

                    return;

                }

            }

            for (ColumnGroupMap group : builder.groups())

                handleGroup(result, keyComponents, group, staticGroup);

        }

        else

        {

            if (cf.hasOnlyTombstones(now))