http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/selection/Selection.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/selection/Selection.java b/src/java/org/apache/cassandra/cql3/selection/Selection.java index 078438b..4a636a1 100644 --- a/src/java/org/apache/cassandra/cql3/selection/Selection.java +++ b/src/java/org/apache/cassandra/cql3/selection/Selection.java @@ -27,18 +27,12 @@ import com.google.common.collect.Iterators; import org.apache.cassandra.cql3.*; import org.apache.cassandra.cql3.functions.Function; -import org.apache.cassandra.db.Clustering; -import org.apache.cassandra.db.DecoratedKey; -import org.apache.cassandra.db.aggregation.AggregationSpecification; -import org.apache.cassandra.db.aggregation.GroupMaker; -import org.apache.cassandra.db.context.CounterContext; +import org.apache.cassandra.db.filter.ColumnFilter; import org.apache.cassandra.db.marshal.UTF8Type; -import org.apache.cassandra.db.rows.Cell; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.transport.ProtocolVersion; -import org.apache.cassandra.utils.ByteBufferUtil; public abstract class Selection { @@ -50,22 +44,28 @@ public abstract class Selection private final TableMetadata table; private final List<ColumnMetadata> columns; private final SelectionColumnMapping columnMapping; - private final ResultSet.ResultMetadata metadata; - private final boolean collectTimestamps; - private final boolean collectTTLs; + protected final ResultSet.ResultMetadata metadata; + protected final ColumnFilterFactory columnFilterFactory; + protected final boolean isJson; protected Selection(TableMetadata table, - List<ColumnMetadata> columns, + List<ColumnMetadata> selectedColumns, + Set<ColumnMetadata> orderingColumns, SelectionColumnMapping columnMapping, - boolean collectTimestamps, - boolean collectTTLs) + ColumnFilterFactory columnFilterFactory, + boolean isJson) { this.table = table; - this.columns = columns; + this.columns = selectedColumns; this.columnMapping = columnMapping; this.metadata = new ResultSet.ResultMetadata(columnMapping.getColumnSpecifications()); - this.collectTimestamps = collectTimestamps; - this.collectTTLs = collectTTLs; + this.columnFilterFactory = columnFilterFactory; + this.isJson = isJson; + + // If we order post-query, the sorted column needs to be in the ResultSet for sorting, + // even if we don't ultimately ship them to the client (CASSANDRA-4911). + this.columns.addAll(orderingColumns); + this.metadata.addNonSerializedColumns(orderingColumns); } // Overriden by SimpleSelection when appropriate. @@ -89,28 +89,7 @@ public abstract class Selection return !Iterables.isEmpty(Iterables.filter(columns, STATIC_COLUMN_FILTER)); } - /** - * Checks if this selection contains only static columns. - * @return <code>true</code> if this selection contains only static columns, <code>false</code> otherwise; - */ - public boolean containsOnlyStaticColumns() - { - if (!containsStaticColumns()) - return false; - - if (isWildcard()) - return false; - - for (ColumnMetadata def : getColumns()) - { - if (!def.isPartitionKey() && !def.isStatic()) - return false; - } - - return true; - } - - public ResultSet.ResultMetadata getResultMetadata(boolean isJson) + public ResultSet.ResultMetadata getResultMetadata() { if (!isJson) return metadata; @@ -120,50 +99,88 @@ public abstract class Selection return new ResultSet.ResultMetadata(Arrays.asList(jsonSpec)); } - public static Selection wildcard(TableMetadata table) + public static Selection wildcard(TableMetadata table, boolean isJson) { List<ColumnMetadata> all = new ArrayList<>(table.columns().size()); Iterators.addAll(all, table.allColumnsInSelectOrder()); - return new SimpleSelection(table, all, true); + return new SimpleSelection(table, all, Collections.emptySet(), true, isJson); } public static Selection forColumns(TableMetadata table, List<ColumnMetadata> columns) { - return new SimpleSelection(table, columns, false); - } - - public int addColumnForOrdering(ColumnMetadata c) - { - columns.add(c); - metadata.addNonSerializedColumn(c); - return columns.size() - 1; + return new SimpleSelection(table, columns, Collections.emptySet(), false, false); } public void addFunctionsTo(List<Function> functions) { } - private static boolean processesSelection(List<RawSelector> rawSelectors) + private static boolean processesSelection(List<Selectable> selectables) { - for (RawSelector rawSelector : rawSelectors) + for (Selectable selectable : selectables) { - if (rawSelector.processesSelection()) + if (selectable.processesSelection()) return true; } return false; } - public static Selection fromSelectors(TableMetadata table, List<RawSelector> rawSelectors, VariableSpecifications boundNames, boolean hasGroupBy) + public static Selection fromSelectors(TableMetadata table, + List<Selectable> selectables, + VariableSpecifications boundNames, + Set<ColumnMetadata> orderingColumns, + Set<ColumnMetadata> nonPKRestrictedColumns, + boolean hasGroupBy, + boolean isJson) { - List<ColumnMetadata> defs = new ArrayList<>(); + List<ColumnMetadata> selectedColumns = new ArrayList<>(); SelectorFactories factories = - SelectorFactories.createFactoriesAndCollectColumnDefinitions(RawSelector.toSelectables(rawSelectors, table), null, table, defs, boundNames); - SelectionColumnMapping mapping = collectColumnMappings(table, rawSelectors, factories); + SelectorFactories.createFactoriesAndCollectColumnDefinitions(selectables, null, table, selectedColumns, boundNames); + SelectionColumnMapping mapping = collectColumnMappings(table, factories); + + Set<ColumnMetadata> filteredOrderingColumns = filterOrderingColumns(orderingColumns, + selectedColumns, + factories); + + return (processesSelection(selectables) || selectables.size() != selectedColumns.size() || hasGroupBy) + ? new SelectionWithProcessing(table, + selectedColumns, + filteredOrderingColumns, + nonPKRestrictedColumns, + mapping, + factories, + isJson) + : new SimpleSelection(table, + selectedColumns, + filteredOrderingColumns, + nonPKRestrictedColumns, + mapping, + isJson); + } + + /** + * Removes the ordering columns that are already selected. + * + * @param orderingColumns the columns used to order the results + * @param selectedColumns the selected columns + * @param factories the factory used to create the selectors + * @return the ordering columns that are not part of the selection + */ + private static Set<ColumnMetadata> filterOrderingColumns(Set<ColumnMetadata> orderingColumns, + List<ColumnMetadata> selectedColumns, + SelectorFactories factories) + { + Set<ColumnMetadata> filteredOrderingColumns = new LinkedHashSet<>(orderingColumns.size()); + for (ColumnMetadata orderingColumn : orderingColumns) + { + int index = selectedColumns.indexOf(orderingColumn); + if (index >= 0 && factories.indexOfSimpleSelectorFactory(index) >= 0) + continue; - return (processesSelection(rawSelectors) || rawSelectors.size() != defs.size() || hasGroupBy) - ? new SelectionWithProcessing(table, defs, mapping, factories) - : new SimpleSelection(table, defs, mapping, false); + filteredOrderingColumns.add(orderingColumn); + } + return filteredOrderingColumns; } /** @@ -183,29 +200,22 @@ public abstract class Selection */ protected final int getColumnIndex(ColumnMetadata c) { - for (int i = 0, m = columns.size(); i < m; i++) - if (columns.get(i).name.equals(c.name)) - return i; - return -1; + return columns.indexOf(c); } private static SelectionColumnMapping collectColumnMappings(TableMetadata table, - List<RawSelector> rawSelectors, SelectorFactories factories) { SelectionColumnMapping selectionColumns = SelectionColumnMapping.newMapping(); - Iterator<RawSelector> iter = rawSelectors.iterator(); for (Selector.Factory factory : factories) { ColumnSpecification colSpec = factory.getColumnSpecification(table); - ColumnIdentifier alias = iter.next().alias; - factory.addColumnMapping(selectionColumns, - alias == null ? colSpec : colSpec.withAlias(alias)); + factory.addColumnMapping(selectionColumns, colSpec); } return selectionColumns; } - protected abstract Selectors newSelectors(QueryOptions options) throws InvalidRequestException; + public abstract Selectors newSelectors(QueryOptions options); /** * @return the list of CQL3 columns value this SelectionClause needs. @@ -223,17 +233,6 @@ public abstract class Selection return columnMapping; } - public ResultSetBuilder resultSetBuilder(QueryOptions options, boolean isJson) - { - return new ResultSetBuilder(options, isJson); - } - - public ResultSetBuilder resultSetBuilder(QueryOptions options, boolean isJson, AggregationSpecification aggregationSpec) - { - return aggregationSpec == null ? new ResultSetBuilder(options, isJson) - : new ResultSetBuilder(options, isJson, aggregationSpec.newGroupMaker()); - } - public abstract boolean isAggregate(); @Override @@ -243,12 +242,10 @@ public abstract class Selection .add("columns", columns) .add("columnMapping", columnMapping) .add("metadata", metadata) - .add("collectTimestamps", collectTimestamps) - .add("collectTTLs", collectTTLs) .toString(); } - public static List<ByteBuffer> rowToJson(List<ByteBuffer> row, ProtocolVersion protocolVersion, ResultSet.ResultMetadata metadata) + private static List<ByteBuffer> rowToJson(List<ByteBuffer> row, ProtocolVersion protocolVersion, ResultSet.ResultMetadata metadata) { StringBuilder sb = new StringBuilder("{"); for (int i = 0; i < metadata.names.size(); i++) @@ -274,159 +271,47 @@ public abstract class Selection return Collections.singletonList(UTF8Type.instance.getSerializer().serialize(sb.toString())); } - public class ResultSetBuilder + public static interface Selectors { - private final ResultSet resultSet; - private final ProtocolVersion protocolVersion; - /** - * As multiple thread can access a <code>Selection</code> instance each <code>ResultSetBuilder</code> will use - * its own <code>Selectors</code> instance. + * Returns the {@code ColumnFilter} corresponding to those selectors + * @return the {@code ColumnFilter} corresponding to those selectors */ - private final Selectors selectors; + public ColumnFilter getColumnFilter(); /** - * The <code>GroupMaker</code> used to build the aggregates. - */ - private final GroupMaker groupMaker; - - /* - * We'll build CQL3 row one by one. - * The currentRow is the values for the (CQL3) columns we've fetched. - * We also collect timestamps and ttls for the case where the writetime and - * ttl functions are used. Note that we might collect timestamp and/or ttls - * we don't care about, but since the array below are allocated just once, - * it doesn't matter performance wise. + * Checks if one of the selectors perform some aggregations. + * @return {@code true} if one of the selectors perform some aggregations, {@code false} otherwise. */ - List<ByteBuffer> current; - final long[] timestamps; - final int[] ttls; - - private final boolean isJson; - - private ResultSetBuilder(QueryOptions options, boolean isJson) - { - this(options, isJson, null); - } - - private ResultSetBuilder(QueryOptions options, boolean isJson, GroupMaker groupMaker) - { - this.resultSet = new ResultSet(getResultMetadata(isJson).copy(), new ArrayList<List<ByteBuffer>>()); - this.protocolVersion = options.getProtocolVersion(); - this.selectors = newSelectors(options); - this.groupMaker = groupMaker; - this.timestamps = collectTimestamps ? new long[columns.size()] : null; - this.ttls = collectTTLs ? new int[columns.size()] : null; - this.isJson = isJson; - - // We use MIN_VALUE to indicate no timestamp and -1 for no ttl - if (timestamps != null) - Arrays.fill(timestamps, Long.MIN_VALUE); - if (ttls != null) - Arrays.fill(ttls, -1); - } - - public void add(ByteBuffer v) - { - current.add(v); - } - - public void add(Cell c, int nowInSec) - { - if (c == null) - { - current.add(null); - return; - } - - current.add(value(c)); - - if (timestamps != null) - timestamps[current.size() - 1] = c.timestamp(); - - if (ttls != null) - ttls[current.size() - 1] = remainingTTL(c, nowInSec); - } - - private int remainingTTL(Cell c, int nowInSec) - { - if (!c.isExpiring()) - return -1; - - int remaining = c.localDeletionTime() - nowInSec; - return remaining >= 0 ? remaining : -1; - } - - private ByteBuffer value(Cell c) - { - return c.isCounterCell() - ? ByteBufferUtil.bytes(CounterContext.instance().total(c.value())) - : c.value(); - } + public boolean isAggregate(); /** - * Notifies this <code>Builder</code> that a new row is being processed. - * - * @param partitionKey the partition key of the new row - * @param clustering the clustering of the new row + * Returns the number of fetched columns + * @return the number of fetched columns */ - public void newRow(DecoratedKey partitionKey, Clustering clustering) - { - // The groupMaker needs to be called for each row - boolean isNewAggregate = groupMaker == null || groupMaker.isNewGroup(partitionKey, clustering); - if (current != null) - { - selectors.addInputRow(protocolVersion, this); - if (isNewAggregate) - { - resultSet.addRow(getOutputRow()); - selectors.reset(); - } - } - current = new ArrayList<>(columns.size()); - } + public int numberOfFetchedColumns(); /** - * Builds the <code>ResultSet</code> + * Checks if one of the selectors collect TTLs. + * @return {@code true} if one of the selectors collect TTLs, {@code false} otherwise. */ - public ResultSet build() - { - if (current != null) - { - selectors.addInputRow(protocolVersion, this); - resultSet.addRow(getOutputRow()); - selectors.reset(); - current = null; - } - - // For aggregates we need to return a row even it no records have been found - if (resultSet.isEmpty() && groupMaker != null && groupMaker.returnAtLeastOneRow()) - resultSet.addRow(getOutputRow()); - return resultSet; - } - - private List<ByteBuffer> getOutputRow() - { - List<ByteBuffer> outputRow = selectors.getOutputRow(protocolVersion); - return isJson ? rowToJson(outputRow, protocolVersion, metadata) - : outputRow; - } - } + public boolean collectTTLs(); - private static interface Selectors - { - public boolean isAggregate(); + /** + * Checks if one of the selectors collect timestamps. + * @return {@code true} if one of the selectors collect timestamps, {@code false} otherwise. + */ + public boolean collectTimestamps(); /** * Adds the current row of the specified <code>ResultSetBuilder</code>. * - * @param protocolVersion * @param rs the <code>ResultSetBuilder</code> * @throws InvalidRequestException */ - public void addInputRow(ProtocolVersion protocolVersion, ResultSetBuilder rs) throws InvalidRequestException; + public void addInputRow(ResultSetBuilder rs); - public List<ByteBuffer> getOutputRow(ProtocolVersion protocolVersion) throws InvalidRequestException; + public List<ByteBuffer> getOutputRow(); public void reset(); } @@ -436,22 +321,52 @@ public abstract class Selection { private final boolean isWildcard; - public SimpleSelection(TableMetadata table, List<ColumnMetadata> columns, boolean isWildcard) + public SimpleSelection(TableMetadata table, + List<ColumnMetadata> selectedColumns, + Set<ColumnMetadata> orderingColumns, + boolean isWildcard, + boolean isJson) { - this(table, columns, SelectionColumnMapping.simpleMapping(columns), isWildcard); + this(table, + selectedColumns, + orderingColumns, + SelectionColumnMapping.simpleMapping(selectedColumns), + isWildcard ? ColumnFilterFactory.wildcard(table) + : ColumnFilterFactory.fromColumns(table, selectedColumns, orderingColumns, Collections.emptySet()), + isWildcard, + isJson); } public SimpleSelection(TableMetadata table, - List<ColumnMetadata> columns, - SelectionColumnMapping metadata, - boolean isWildcard) + List<ColumnMetadata> selectedColumns, + Set<ColumnMetadata> orderingColumns, + Set<ColumnMetadata> nonPKRestrictedColumns, + SelectionColumnMapping mapping, + boolean isJson) + { + this(table, + selectedColumns, + orderingColumns, + mapping, + ColumnFilterFactory.fromColumns(table, selectedColumns, orderingColumns, nonPKRestrictedColumns), + false, + isJson); + } + + private SimpleSelection(TableMetadata table, + List<ColumnMetadata> selectedColumns, + Set<ColumnMetadata> orderingColumns, + SelectionColumnMapping mapping, + ColumnFilterFactory columnFilterFactory, + boolean isWildcard, + boolean isJson) { /* * In theory, even a simple selection could have multiple time the same column, so we * could filter those duplicate out of columns. But since we're very unlikely to * get much duplicate in practice, it's more efficient not to bother. */ - super(table, columns, metadata, false, false); + super(table, selectedColumns, orderingColumns, mapping, columnFilterFactory, isJson); this.isWildcard = isWildcard; } @@ -466,7 +381,7 @@ public abstract class Selection return false; } - protected Selectors newSelectors(QueryOptions options) + public Selectors newSelectors(QueryOptions options) { return new Selectors() { @@ -477,12 +392,12 @@ public abstract class Selection current = null; } - public List<ByteBuffer> getOutputRow(ProtocolVersion protocolVersion) + public List<ByteBuffer> getOutputRow() { - return current; + return isJson ? rowToJson(current, options.getProtocolVersion(), metadata) : current; } - public void addInputRow(ProtocolVersion protocolVersion, ResultSetBuilder rs) throws InvalidRequestException + public void addInputRow(ResultSetBuilder rs) throws InvalidRequestException { current = rs.current; } @@ -491,6 +406,32 @@ public abstract class Selection { return false; } + + @Override + public int numberOfFetchedColumns() + { + return getColumns().size(); + } + + @Override + public boolean collectTTLs() + { + return false; + } + + @Override + public boolean collectTimestamps() + { + return false; + } + + @Override + public ColumnFilter getColumnFilter() + { + // In the case of simple selection we know that the ColumnFilter has already been computed and + // that by consequence the selectors argument has not impact on the output. + return columnFilterFactory.newInstance(null); + } }; } } @@ -498,19 +439,32 @@ public abstract class Selection private static class SelectionWithProcessing extends Selection { private final SelectorFactories factories; + private final boolean collectTimestamps; + private final boolean collectTTLs; public SelectionWithProcessing(TableMetadata table, List<ColumnMetadata> columns, + Set<ColumnMetadata> orderingColumns, + Set<ColumnMetadata> nonPKRestrictedColumns, SelectionColumnMapping metadata, - SelectorFactories factories) throws InvalidRequestException + SelectorFactories factories, + boolean isJson) { super(table, columns, + orderingColumns, metadata, - factories.containsWritetimeSelectorFactory(), - factories.containsTTLSelectorFactory()); + ColumnFilterFactory.fromSelectorFactories(table, factories, orderingColumns, nonPKRestrictedColumns), + isJson); this.factories = factories; + this.collectTimestamps = factories.containsWritetimeSelectorFactory(); + this.collectTTLs = factories.containsTTLSelectorFactory();; + + for (ColumnMetadata orderingColumn : orderingColumns) + { + factories.addSelectorForOrdering(orderingColumn, getColumnIndex(orderingColumn)); + } } @Override @@ -522,24 +476,7 @@ public abstract class Selection @Override public int getResultSetIndex(ColumnMetadata c) { - int index = getColumnIndex(c); - - if (index < 0) - return -1; - - for (int i = 0, m = factories.size(); i < m; i++) - if (factories.get(i).isSimpleSelectorFactory(index)) - return i; - - return -1; - } - - @Override - public int addColumnForOrdering(ColumnMetadata c) - { - int index = super.addColumnForOrdering(c); - factories.addSelectorForOrdering(c, index); - return factories.size() - 1; + return factories.indexOfSimpleSelectorFactory(super.getResultSetIndex(c)); } public boolean isAggregate() @@ -547,7 +484,7 @@ public abstract class Selection return factories.doesAggregation(); } - protected Selectors newSelectors(final QueryOptions options) throws InvalidRequestException + public Selectors newSelectors(final QueryOptions options) throws InvalidRequestException { return new Selectors() { @@ -564,20 +501,44 @@ public abstract class Selection return factories.doesAggregation(); } - public List<ByteBuffer> getOutputRow(ProtocolVersion protocolVersion) throws InvalidRequestException + public List<ByteBuffer> getOutputRow() { List<ByteBuffer> outputRow = new ArrayList<>(selectors.size()); for (Selector selector: selectors) - outputRow.add(selector.getOutput(protocolVersion)); + outputRow.add(selector.getOutput(options.getProtocolVersion())); - return outputRow; + return isJson ? rowToJson(outputRow, options.getProtocolVersion(), metadata) : outputRow; } - public void addInputRow(ProtocolVersion protocolVersion, ResultSetBuilder rs) throws InvalidRequestException + public void addInputRow(ResultSetBuilder rs) throws InvalidRequestException { for (Selector selector : selectors) - selector.addInput(protocolVersion, rs); + selector.addInput(options.getProtocolVersion(), rs); + } + + @Override + public int numberOfFetchedColumns() + { + return getColumns().size(); + } + + @Override + public boolean collectTTLs() + { + return collectTTLs; + } + + @Override + public boolean collectTimestamps() + { + return collectTimestamps; + } + + @Override + public ColumnFilter getColumnFilter() + { + return columnFilterFactory.newInstance(selectors); } }; }
http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/selection/Selector.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/selection/Selector.java b/src/java/org/apache/cassandra/cql3/selection/Selector.java index 420af9c..3262b9c 100644 --- a/src/java/org/apache/cassandra/cql3/selection/Selector.java +++ b/src/java/org/apache/cassandra/cql3/selection/Selector.java @@ -25,7 +25,7 @@ import org.apache.cassandra.cql3.ColumnIdentifier; import org.apache.cassandra.cql3.ColumnSpecification; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.functions.Function; -import org.apache.cassandra.cql3.selection.Selection.ResultSetBuilder; +import org.apache.cassandra.db.filter.ColumnFilter; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.transport.ProtocolVersion; @@ -54,7 +54,7 @@ public abstract class Selector * @param table the table meta data * @return a column specification */ - public final ColumnSpecification getColumnSpecification(TableMetadata table) + public ColumnSpecification getColumnSpecification(TableMetadata table) { return new ColumnSpecification(table.keyspace, table.name, @@ -106,13 +106,25 @@ public abstract class Selector } /** + * Checks if this factory creates <code>Selector</code>s that simply return a column value. + * + * @param index the column index + * @return <code>true</code> if this factory creates <code>Selector</code>s that simply return a column value, + * <code>false</code> otherwise. + */ + public boolean isSimpleSelectorFactory() + { + return false; + } + + /** * Checks if this factory creates <code>Selector</code>s that simply return the specified column. * * @param index the column index * @return <code>true</code> if this factory creates <code>Selector</code>s that simply return * the specified column, <code>false</code> otherwise. */ - public boolean isSimpleSelectorFactory(int index) + public boolean isSimpleSelectorFactoryFor(int index) { return false; } @@ -144,9 +156,33 @@ public abstract class Selector * by the Selector are to be mapped */ protected abstract void addColumnMapping(SelectionColumnMapping mapping, ColumnSpecification resultsColumn); + + /** + * Checks if all the columns fetched by the selector created by this factory are known + * @return {@code true} if all the columns fetched by the selector created by this factory are known, + * {@code false} otherwise. + */ + abstract boolean areAllFetchedColumnsKnown(); + + /** + * Adds the columns fetched by the selector created by this factory to the provided builder, assuming the + * factory is terminal (i.e. that {@code isTerminal() == true}). + * + * @param builder the column builder to add fetched columns (and potential subselection) to. + * @throws AssertionError if the method is called on a factory where {@code isTerminal()} returns {@code false}. + */ + abstract void addFetchedColumns(ColumnFilter.Builder builder); } /** + * Add to the provided builder the column (and potential subselections) to fetch for this + * selection. + * + * @param builder the builder to add columns and subselections to. + */ + public abstract void addFetchedColumns(ColumnFilter.Builder builder); + + /** * Add the current value from the specified <code>ResultSetBuilder</code>. * * @param protocolVersion protocol version used for serialization http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java b/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java index 25a1059..7f4bcb3 100644 --- a/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java +++ b/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java @@ -27,6 +27,7 @@ import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.VariableSpecifications; import org.apache.cassandra.cql3.functions.Function; import org.apache.cassandra.cql3.selection.Selector.Factory; +import org.apache.cassandra.db.filter.ColumnFilter.Builder; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.exceptions.InvalidRequestException; @@ -118,6 +119,22 @@ final class SelectorFactories implements Iterable<Selector.Factory> } /** + * Returns the index of the {@code SimpleSelector.Factory} for the specified column. + * + * @param columnIndex the index of the column + * @return the index of the {@code SimpleSelector.Factory} for the specified column or -1 if it does not exist. + */ + public int indexOfSimpleSelectorFactory(int columnIndex) + { + for (int i = 0, m = factories.size(); i < m; i++) + { + if (factories.get(i).isSimpleSelectorFactoryFor(columnIndex)) + return i; + } + return -1; + } + + /** * Adds a new <code>Selector.Factory</code> for a column that is needed only for ORDER BY purposes. * @param def the column that is needed for ordering * @param index the index of the column definition in the Selection's list of columns @@ -211,6 +228,22 @@ final class SelectorFactories implements Iterable<Selector.Factory> }); } + boolean areAllFetchedColumnsKnown() + { + for (Factory factory : factories) + { + if (!factory.areAllFetchedColumnsKnown()) + return false; + } + return true; + } + + void addFetchedColumns(Builder builder) + { + for (Factory factory : factories) + factory.addFetchedColumns(builder); + } + /** * Returns the number of factories. * @return the number of factories http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/selection/SetSelector.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/selection/SetSelector.java b/src/java/org/apache/cassandra/cql3/selection/SetSelector.java index 2ee086e..6693121 100644 --- a/src/java/org/apache/cassandra/cql3/selection/SetSelector.java +++ b/src/java/org/apache/cassandra/cql3/selection/SetSelector.java @@ -24,7 +24,7 @@ import java.util.TreeSet; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.Sets; -import org.apache.cassandra.cql3.selection.Selection.ResultSetBuilder; +import org.apache.cassandra.db.filter.ColumnFilter.Builder; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.SetType; import org.apache.cassandra.exceptions.InvalidRequestException; @@ -63,6 +63,13 @@ final class SetSelector extends Selector }; } + @Override + public void addFetchedColumns(Builder builder) + { + for (int i = 0, m = elements.size(); i < m; i++) + elements.get(i).addFetchedColumns(builder); + } + public void addInput(ProtocolVersion protocolVersion, ResultSetBuilder rs) throws InvalidRequestException { for (int i = 0, m = elements.size(); i < m; i++) http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java b/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java index cbd65a9..31b1911 100644 --- a/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java +++ b/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java @@ -22,52 +22,94 @@ import java.nio.ByteBuffer; import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.cql3.ColumnSpecification; import org.apache.cassandra.cql3.QueryOptions; -import org.apache.cassandra.cql3.selection.Selection.ResultSetBuilder; +import org.apache.cassandra.db.filter.ColumnFilter; +import org.apache.cassandra.db.filter.ColumnFilter.Builder; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.transport.ProtocolVersion; public final class SimpleSelector extends Selector { - private final String columnName; + /** + * The Factory for {@code SimpleSelector}. + */ + public static final class SimpleSelectorFactory extends Factory + { + private final int idx; + + private final ColumnMetadata column; + + private SimpleSelectorFactory(int idx, ColumnMetadata def) + { + this.idx = idx; + this.column = def; + } + + @Override + protected String getColumnName() + { + return column.name.toString(); + } + + @Override + protected AbstractType<?> getReturnType() + { + return column.type; + } + + protected void addColumnMapping(SelectionColumnMapping mapping, ColumnSpecification resultColumn) + { + mapping.addMapping(resultColumn, column); + } + + @Override + public Selector newInstance(QueryOptions options) + { + return new SimpleSelector(column, idx); + } + + @Override + public boolean isSimpleSelectorFactory() + { + return true; + } + + @Override + public boolean isSimpleSelectorFactoryFor(int index) + { + return index == idx; + } + + public boolean areAllFetchedColumnsKnown() + { + return true; + } + + public void addFetchedColumns(ColumnFilter.Builder builder) + { + builder.add(column); + } + + public ColumnMetadata getColumn() + { + return column; + } + } + + public final ColumnMetadata column; private final int idx; - private final AbstractType<?> type; private ByteBuffer current; private boolean isSet; public static Factory newFactory(final ColumnMetadata def, final int idx) { - return new Factory() - { - @Override - protected String getColumnName() - { - return def.name.toString(); - } - - @Override - protected AbstractType<?> getReturnType() - { - return def.type; - } - - protected void addColumnMapping(SelectionColumnMapping mapping, ColumnSpecification resultColumn) - { - mapping.addMapping(resultColumn, def); - } - - @Override - public Selector newInstance(QueryOptions options) - { - return new SimpleSelector(def.name.toString(), idx, def.type); - } - - @Override - public boolean isSimpleSelectorFactory(int index) - { - return index == idx; - } - }; + return new SimpleSelectorFactory(idx, def); + } + + @Override + public void addFetchedColumns(Builder builder) + { + builder.add(column); } @Override @@ -96,19 +138,18 @@ public final class SimpleSelector extends Selector @Override public AbstractType<?> getType() { - return type; + return column.type; } @Override public String toString() { - return columnName; + return column.name.toString(); } - private SimpleSelector(String columnName, int idx, AbstractType<?> type) + private SimpleSelector(ColumnMetadata column, int idx) { - this.columnName = columnName; + this.column = column; this.idx = idx; - this.type = type; } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/selection/TermSelector.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/selection/TermSelector.java b/src/java/org/apache/cassandra/cql3/selection/TermSelector.java index bdb4953..321cd27 100644 --- a/src/java/org/apache/cassandra/cql3/selection/TermSelector.java +++ b/src/java/org/apache/cassandra/cql3/selection/TermSelector.java @@ -23,6 +23,7 @@ import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.cql3.ColumnSpecification; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.Term; +import org.apache.cassandra.db.filter.ColumnFilter; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.transport.ProtocolVersion; @@ -61,6 +62,15 @@ public class TermSelector extends Selector { return new TermSelector(term.bindAndGet(options), type); } + + public void addFetchedColumns(ColumnFilter.Builder builder) + { + } + + public boolean areAllFetchedColumnsKnown() + { + return true; + } }; } @@ -70,7 +80,11 @@ public class TermSelector extends Selector this.type = type; } - public void addInput(ProtocolVersion protocolVersion, Selection.ResultSetBuilder rs) throws InvalidRequestException + public void addFetchedColumns(ColumnFilter.Builder builder) + { + } + + public void addInput(ProtocolVersion protocolVersion, ResultSetBuilder rs) throws InvalidRequestException { } http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/selection/TupleSelector.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/selection/TupleSelector.java b/src/java/org/apache/cassandra/cql3/selection/TupleSelector.java index 9f4c381..898085b 100644 --- a/src/java/org/apache/cassandra/cql3/selection/TupleSelector.java +++ b/src/java/org/apache/cassandra/cql3/selection/TupleSelector.java @@ -22,7 +22,7 @@ import java.util.List; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.Tuples; -import org.apache.cassandra.cql3.selection.Selection.ResultSetBuilder; +import org.apache.cassandra.db.filter.ColumnFilter.Builder; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.TupleType; import org.apache.cassandra.exceptions.InvalidRequestException; @@ -60,6 +60,13 @@ final class TupleSelector extends Selector }; } + @Override + public void addFetchedColumns(Builder builder) + { + for (int i = 0, m = elements.size(); i < m; i++) + elements.get(i).addFetchedColumns(builder); + } + public void addInput(ProtocolVersion protocolVersion, ResultSetBuilder rs) throws InvalidRequestException { for (int i = 0, m = elements.size(); i < m; i++) http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/selection/UserTypeSelector.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/selection/UserTypeSelector.java b/src/java/org/apache/cassandra/cql3/selection/UserTypeSelector.java index 7600e1d..61faf8d 100644 --- a/src/java/org/apache/cassandra/cql3/selection/UserTypeSelector.java +++ b/src/java/org/apache/cassandra/cql3/selection/UserTypeSelector.java @@ -29,7 +29,8 @@ import org.apache.cassandra.cql3.FieldIdentifier; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.UserTypes; import org.apache.cassandra.cql3.functions.Function; -import org.apache.cassandra.cql3.selection.Selection.ResultSetBuilder; +import org.apache.cassandra.db.filter.ColumnFilter; +import org.apache.cassandra.db.filter.ColumnFilter.Builder; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.TupleType; import org.apache.cassandra.db.marshal.UserType; @@ -130,9 +131,33 @@ final class UserTypeSelector extends Selector } return false; } + + @Override + boolean areAllFetchedColumnsKnown() + { + for (Factory factory : factories.values()) + { + if (!factory.areAllFetchedColumnsKnown()) + return false; + } + return true; + } + + @Override + void addFetchedColumns(Builder builder) + { + for (Factory factory : factories.values()) + factory.addFetchedColumns(builder); + } }; } + public void addFetchedColumns(ColumnFilter.Builder builder) + { + for (Selector field : fields.values()) + field.addFetchedColumns(builder); + } + public void addInput(ProtocolVersion protocolVersion, ResultSetBuilder rs) throws InvalidRequestException { for (Selector field : fields.values()) http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/selection/WritetimeOrTTLSelector.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/selection/WritetimeOrTTLSelector.java b/src/java/org/apache/cassandra/cql3/selection/WritetimeOrTTLSelector.java index 1e38337..95586f2 100644 --- a/src/java/org/apache/cassandra/cql3/selection/WritetimeOrTTLSelector.java +++ b/src/java/org/apache/cassandra/cql3/selection/WritetimeOrTTLSelector.java @@ -22,7 +22,7 @@ import java.nio.ByteBuffer; import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.ColumnSpecification; -import org.apache.cassandra.cql3.selection.Selection.ResultSetBuilder; +import org.apache.cassandra.db.filter.ColumnFilter; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.Int32Type; import org.apache.cassandra.db.marshal.LongType; @@ -31,7 +31,7 @@ import org.apache.cassandra.utils.ByteBufferUtil; final class WritetimeOrTTLSelector extends Selector { - private final String columnName; + private final ColumnMetadata column; private final int idx; private final boolean isWritetime; private ByteBuffer current; @@ -58,7 +58,7 @@ final class WritetimeOrTTLSelector extends Selector public Selector newInstance(QueryOptions options) { - return new WritetimeOrTTLSelector(def.name.toString(), idx, isWritetime); + return new WritetimeOrTTLSelector(def, idx, isWritetime); } public boolean isWritetimeSelectorFactory() @@ -70,9 +70,24 @@ final class WritetimeOrTTLSelector extends Selector { return !isWritetime; } + + public boolean areAllFetchedColumnsKnown() + { + return true; + } + + public void addFetchedColumns(ColumnFilter.Builder builder) + { + builder.add(def); + } }; } + public void addFetchedColumns(ColumnFilter.Builder builder) + { + builder.add(column); + } + public void addInput(ProtocolVersion protocolVersion, ResultSetBuilder rs) { if (isSet) @@ -111,14 +126,13 @@ final class WritetimeOrTTLSelector extends Selector @Override public String toString() { - return columnName; + return column.name.toString(); } - private WritetimeOrTTLSelector(String columnName, int idx, boolean isWritetime) + private WritetimeOrTTLSelector(ColumnMetadata column, int idx, boolean isWritetime) { - this.columnName = columnName; + this.column = column; this.idx = idx; this.isWritetime = isWritetime; } - } http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/statements/CreateViewStatement.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/statements/CreateViewStatement.java b/src/java/org/apache/cassandra/cql3/statements/CreateViewStatement.java index 8e92534..270533d 100644 --- a/src/java/org/apache/cassandra/cql3/statements/CreateViewStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/CreateViewStatement.java @@ -160,6 +160,10 @@ public class CreateViewStatement extends SchemaAlteringStatement throw new InvalidRequestException("Cannot use function when defining a materialized view"); if (selectable instanceof Selectable.WritetimeOrTTL.Raw) throw new InvalidRequestException("Cannot use function when defining a materialized view"); + if (selectable instanceof Selectable.WithElementSelection.Raw) + throw new InvalidRequestException("Cannot use collection element selection when defining a materialized view"); + if (selectable instanceof Selectable.WithSliceSelection.Raw) + throw new InvalidRequestException("Cannot use collection slice selection when defining a materialized view"); if (selector.alias != null) throw new InvalidRequestException("Cannot use alias when defining a materialized view"); http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index e82d840..8a22262 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -36,7 +36,9 @@ import org.apache.cassandra.cql3.conditions.ColumnConditions; import org.apache.cassandra.cql3.conditions.Conditions; import org.apache.cassandra.cql3.functions.Function; import org.apache.cassandra.cql3.restrictions.StatementRestrictions; +import org.apache.cassandra.cql3.selection.ResultSetBuilder; import org.apache.cassandra.cql3.selection.Selection; +import org.apache.cassandra.cql3.selection.Selection.Selectors; import org.apache.cassandra.db.*; import org.apache.cassandra.db.filter.*; import org.apache.cassandra.db.marshal.BooleanType; @@ -536,7 +538,7 @@ public abstract class ModificationStatement implements CQLStatement Selection selection; if (columnsWithConditions == null) { - selection = Selection.wildcard(metadata); + selection = Selection.wildcard(metadata, false); } else { @@ -552,7 +554,8 @@ public abstract class ModificationStatement implements CQLStatement } - Selection.ResultSetBuilder builder = selection.resultSetBuilder(options, false); + Selectors selectors = selection.newSelectors(options); + ResultSetBuilder builder = new ResultSetBuilder(selection.getResultMetadata(), selectors); SelectStatement.forSelection(metadata, selection).processPartition(partition, options, builder, http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 652b549..a39416b 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -34,9 +34,13 @@ import org.apache.cassandra.cql3.*; import org.apache.cassandra.cql3.functions.Function; import org.apache.cassandra.cql3.restrictions.StatementRestrictions; import org.apache.cassandra.cql3.selection.RawSelector; +import org.apache.cassandra.cql3.selection.ResultSetBuilder; +import org.apache.cassandra.cql3.selection.Selectable; import org.apache.cassandra.cql3.selection.Selection; +import org.apache.cassandra.cql3.selection.Selection.Selectors; import org.apache.cassandra.db.*; import org.apache.cassandra.db.aggregation.AggregationSpecification; +import org.apache.cassandra.db.aggregation.GroupMaker; import org.apache.cassandra.db.filter.*; import org.apache.cassandra.db.marshal.CollectionType; import org.apache.cassandra.db.marshal.CompositeType; @@ -104,8 +108,6 @@ public class SelectStatement implements CQLStatement */ private final Comparator<List<ByteBuffer>> orderingComparator; - private final ColumnFilter queriedColumns; - // Used by forSelection below private static final Parameters defaultParameters = new Parameters(Collections.emptyMap(), Collections.emptyList(), @@ -134,7 +136,6 @@ public class SelectStatement implements CQLStatement this.parameters = parameters; this.limit = limit; this.perPartitionLimit = perPartitionLimit; - this.queriedColumns = gatherQueriedColumns(); } public Iterable<Function> getFunctions() @@ -156,36 +157,13 @@ public class SelectStatement implements CQLStatement perPartitionLimit.addFunctionsTo(functions); } - // Note that the queried columns internally is different from the one selected by the - // user as it also include any column for which we have a restriction on. - private ColumnFilter gatherQueriedColumns() - { - if (selection.isWildcard()) - return ColumnFilter.all(table); - - ColumnFilter.Builder builder = ColumnFilter.allRegularColumnsBuilder(table); - // Adds all selected columns - for (ColumnMetadata def : selection.getColumns()) - if (!def.isPrimaryKeyColumn()) - builder.add(def); - // as well as any restricted column (so we can actually apply the restriction) - builder.addAll(restrictions.nonPKRestrictedColumns(true)); - - // In a number of cases, we want to distinguish between a partition truly empty and one with only static content - // (but no rows). In those cases, we should force querying all static columns (to make the distinction). - if (table.hasStaticColumns() && returnStaticContentOnPartitionWithNoRows()) - builder.addAll(table.staticColumns()); - - return builder.build(); - } - /** * The columns to fetch internally for this SELECT statement (which can be more than the one selected by the * user as it also include any restricted column in particular). */ public ColumnFilter queriedColumns() { - return queriedColumns; + return selection.newSelectors(QueryOptions.DEFAULT).getColumnFilter(); } // Creates a simple select based on the given selection. @@ -207,7 +185,7 @@ public class SelectStatement implements CQLStatement public ResultSet.ResultMetadata getResultMetadata() { - return selection.getResultMetadata(parameters.isJson); + return selection.getResultMetadata(); } public int getBoundTerms() @@ -248,42 +226,62 @@ public class SelectStatement implements CQLStatement int userLimit = getLimit(options); int userPerPartitionLimit = getPerPartitionLimit(options); int pageSize = options.getPageSize(); - ReadQuery query = getQuery(options, nowInSec, userLimit, userPerPartitionLimit, pageSize); + + Selectors selectors = selection.newSelectors(options); + ReadQuery query = getQuery(options, selectors.getColumnFilter(), nowInSec, userLimit, userPerPartitionLimit, pageSize); if (aggregationSpec == null && (pageSize <= 0 || (query.limits().count() <= pageSize))) - return execute(query, options, state, nowInSec, userLimit, queryStartNanoTime); + return execute(query, options, state, selectors, nowInSec, userLimit, queryStartNanoTime); QueryPager pager = getPager(query, options); - return execute(Pager.forDistributedQuery(pager, cl, state.getClientState()), options, pageSize, nowInSec, userLimit, queryStartNanoTime); + return execute(Pager.forDistributedQuery(pager, cl, state.getClientState()), + options, + selectors, + pageSize, + nowInSec, + userLimit, + queryStartNanoTime); } public ReadQuery getQuery(QueryOptions options, int nowInSec) throws RequestValidationException { - return getQuery(options, nowInSec, getLimit(options), getPerPartitionLimit(options), options.getPageSize()); + Selectors selectors = selection.newSelectors(options); + return getQuery(options, + selectors.getColumnFilter(), + nowInSec, + getLimit(options), + getPerPartitionLimit(options), + options.getPageSize()); } - public ReadQuery getQuery(QueryOptions options, int nowInSec, int userLimit, int perPartitionLimit, int pageSize) + public ReadQuery getQuery(QueryOptions options, + ColumnFilter columnFilter, + int nowInSec, + int userLimit, + int perPartitionLimit, + int pageSize) { boolean isPartitionRangeQuery = restrictions.isKeyRange() || restrictions.usesSecondaryIndexing(); DataLimits limit = getDataLimits(userLimit, perPartitionLimit, pageSize); if (isPartitionRangeQuery) - return getRangeCommand(options, limit, nowInSec); + return getRangeCommand(options, columnFilter, limit, nowInSec); - return getSliceCommands(options, limit, nowInSec); + return getSliceCommands(options, columnFilter, limit, nowInSec); } private ResultMessage.Rows execute(ReadQuery query, QueryOptions options, QueryState state, + Selectors selectors, int nowInSec, int userLimit, long queryStartNanoTime) throws RequestValidationException, RequestExecutionException { try (PartitionIterator data = query.execute(options.getConsistency(), state.getClientState(), queryStartNanoTime)) { - return processResults(data, options, nowInSec, userLimit); + return processResults(data, options, selectors, nowInSec, userLimit); } } @@ -356,6 +354,7 @@ public class SelectStatement implements CQLStatement private ResultMessage.Rows execute(Pager pager, QueryOptions options, + Selectors selectors, int pageSize, int nowInSec, int userLimit, @@ -382,7 +381,7 @@ public class SelectStatement implements CQLStatement ResultMessage.Rows msg; try (PartitionIterator page = pager.fetchPage(pageSize, queryStartNanoTime)) { - msg = processResults(page, options, nowInSec, userLimit); + msg = processResults(page, options, selectors, nowInSec, userLimit); } // Please note that the isExhausted state of the pager only gets updated when we've closed the page, so this @@ -401,10 +400,11 @@ public class SelectStatement implements CQLStatement private ResultMessage.Rows processResults(PartitionIterator partitions, QueryOptions options, + Selectors selectors, int nowInSec, int userLimit) throws RequestValidationException { - ResultSet rset = process(partitions, options, nowInSec, userLimit); + ResultSet rset = process(partitions, options, selectors, nowInSec, userLimit); return new ResultMessage.Rows(rset); } @@ -418,7 +418,9 @@ public class SelectStatement implements CQLStatement int userLimit = getLimit(options); int userPerPartitionLimit = getPerPartitionLimit(options); int pageSize = options.getPageSize(); - ReadQuery query = getQuery(options, nowInSec, userLimit, userPerPartitionLimit, pageSize); + + Selectors selectors = selection.newSelectors(options); + ReadQuery query = getQuery(options, selectors.getColumnFilter(), nowInSec, userLimit, userPerPartitionLimit, pageSize); try (ReadExecutionController executionController = query.executionController()) { @@ -426,15 +428,19 @@ public class SelectStatement implements CQLStatement { try (PartitionIterator data = query.executeInternal(executionController)) { - return processResults(data, options, nowInSec, userLimit); + return processResults(data, options, selectors, nowInSec, userLimit); } } - else - { - QueryPager pager = getPager(query, options); - return execute(Pager.forInternalQuery(pager, executionController), options, pageSize, nowInSec, userLimit, queryStartNanoTime); - } + QueryPager pager = getPager(query, options); + + return execute(Pager.forInternalQuery(pager, executionController), + options, + selectors, + pageSize, + nowInSec, + userLimit, + queryStartNanoTime); } } @@ -450,7 +456,9 @@ public class SelectStatement implements CQLStatement public ResultSet process(PartitionIterator partitions, int nowInSec) throws InvalidRequestException { - return process(partitions, QueryOptions.DEFAULT, nowInSec, getLimit(QueryOptions.DEFAULT)); + QueryOptions options = QueryOptions.DEFAULT; + Selectors selectors = selection.newSelectors(options); + return process(partitions, options, selectors, nowInSec, getLimit(options)); } public String keyspace() @@ -479,13 +487,13 @@ public class SelectStatement implements CQLStatement return restrictions; } - private ReadQuery getSliceCommands(QueryOptions options, DataLimits limit, int nowInSec) throws RequestValidationException + private ReadQuery getSliceCommands(QueryOptions options, ColumnFilter columnFilter, DataLimits limit, int nowInSec) { Collection<ByteBuffer> keys = restrictions.getPartitionKeys(options); if (keys.isEmpty()) return ReadQuery.EMPTY; - ClusteringIndexFilter filter = makeClusteringIndexFilter(options); + ClusteringIndexFilter filter = makeClusteringIndexFilter(options, columnFilter); if (filter == null) return ReadQuery.EMPTY; @@ -498,7 +506,7 @@ public class SelectStatement implements CQLStatement { QueryProcessor.validateKey(key); DecoratedKey dk = table.partitioner.decorateKey(ByteBufferUtil.clone(key)); - commands.add(SinglePartitionReadCommand.create(table, nowInSec, queriedColumns, rowFilter, limit, dk, filter)); + commands.add(SinglePartitionReadCommand.create(table, nowInSec, columnFilter, rowFilter, limit, dk, filter)); } return new SinglePartitionReadCommand.Group(commands, limit); @@ -515,7 +523,8 @@ public class SelectStatement implements CQLStatement public Slices clusteringIndexFilterAsSlices() { QueryOptions options = QueryOptions.forInternalCalls(Collections.emptyList()); - ClusteringIndexFilter filter = makeClusteringIndexFilter(options); + ColumnFilter columnFilter = selection.newSelectors(options).getColumnFilter(); + ClusteringIndexFilter filter = makeClusteringIndexFilter(options, columnFilter); if (filter instanceof ClusteringIndexSliceFilter) return ((ClusteringIndexSliceFilter)filter).requestedSlices(); @@ -532,9 +541,10 @@ public class SelectStatement implements CQLStatement public SinglePartitionReadCommand internalReadForView(DecoratedKey key, int nowInSec) { QueryOptions options = QueryOptions.forInternalCalls(Collections.emptyList()); - ClusteringIndexFilter filter = makeClusteringIndexFilter(options); + ColumnFilter columnFilter = selection.newSelectors(options).getColumnFilter(); + ClusteringIndexFilter filter = makeClusteringIndexFilter(options, columnFilter); RowFilter rowFilter = getRowFilter(options); - return SinglePartitionReadCommand.create(table, nowInSec, queriedColumns, rowFilter, DataLimits.NONE, key, filter); + return SinglePartitionReadCommand.create(table, nowInSec, columnFilter, rowFilter, DataLimits.NONE, key, filter); } /** @@ -545,9 +555,9 @@ public class SelectStatement implements CQLStatement return getRowFilter(QueryOptions.forInternalCalls(Collections.emptyList())); } - private ReadQuery getRangeCommand(QueryOptions options, DataLimits limit, int nowInSec) throws RequestValidationException + private ReadQuery getRangeCommand(QueryOptions options, ColumnFilter columnFilter, DataLimits limit, int nowInSec) { - ClusteringIndexFilter clusteringIndexFilter = makeClusteringIndexFilter(options); + ClusteringIndexFilter clusteringIndexFilter = makeClusteringIndexFilter(options, columnFilter); if (clusteringIndexFilter == null) return ReadQuery.EMPTY; @@ -561,7 +571,7 @@ public class SelectStatement implements CQLStatement PartitionRangeReadCommand command = new PartitionRangeReadCommand(table, nowInSec, - queriedColumns, + columnFilter, rowFilter, limit, new DataRange(keyBounds, clusteringIndexFilter), @@ -576,8 +586,7 @@ public class SelectStatement implements CQLStatement return command; } - private ClusteringIndexFilter makeClusteringIndexFilter(QueryOptions options) - throws InvalidRequestException + private ClusteringIndexFilter makeClusteringIndexFilter(QueryOptions options, ColumnFilter columnFilter) { if (parameters.isDistinct) { @@ -599,17 +608,15 @@ public class SelectStatement implements CQLStatement return new ClusteringIndexSliceFilter(slices, isReversed); } - else - { - NavigableSet<Clustering> clusterings = getRequestedRows(options); - // We can have no clusterings if either we're only selecting the static columns, or if we have - // a 'IN ()' for clusterings. In that case, we still want to query if some static columns are - // queried. But we're fine otherwise. - if (clusterings.isEmpty() && queriedColumns.fetchedColumns().statics.isEmpty()) - return null; - return new ClusteringIndexNamesFilter(clusterings, isReversed); - } + NavigableSet<Clustering> clusterings = getRequestedRows(options); + // We can have no clusterings if either we're only selecting the static columns, or if we have + // a 'IN ()' for clusterings. In that case, we still want to query if some static columns are + // queried. But we're fine otherwise. + if (clusterings.isEmpty() && columnFilter.fetchedColumns().statics.isEmpty()) + return null; + + return new ClusteringIndexNamesFilter(clusterings, isReversed); } private Slices makeSlices(QueryOptions options) @@ -754,10 +761,12 @@ public class SelectStatement implements CQLStatement private ResultSet process(PartitionIterator partitions, QueryOptions options, + Selectors selectors, int nowInSec, int userLimit) throws InvalidRequestException { - Selection.ResultSetBuilder result = selection.resultSetBuilder(options, parameters.isJson, aggregationSpec); + GroupMaker groupMaker = aggregationSpec == null ? null : aggregationSpec.newGroupMaker(); + ResultSetBuilder result = new ResultSetBuilder(getResultMetadata(), selectors, groupMaker); while (partitions.hasNext()) { @@ -802,7 +811,7 @@ public class SelectStatement implements CQLStatement } // Used by ModificationStatement for CAS operations - void processPartition(RowIterator partition, QueryOptions options, Selection.ResultSetBuilder result, int nowInSec) + void processPartition(RowIterator partition, QueryOptions options, ResultSetBuilder result, int nowInSec) throws InvalidRequestException { ProtocolVersion protocolVersion = options.getProtocolVersion(); @@ -869,7 +878,7 @@ public class SelectStatement implements CQLStatement return !restrictions.hasClusteringColumnsRestrictions() && !restrictions.hasRegularColumnsRestrictions(); } - private static void addValue(Selection.ResultSetBuilder result, ColumnMetadata def, Row row, int nowInSec, ProtocolVersion protocolVersion) + private static void addValue(ResultSetBuilder result, ColumnMetadata def, Row row, int nowInSec, ProtocolVersion protocolVersion) { if (def.isComplex()) { @@ -937,11 +946,26 @@ public class SelectStatement implements CQLStatement TableMetadata table = Schema.instance.validateTable(keyspace(), columnFamily()); VariableSpecifications boundNames = getBoundVariables(); - Selection selection = selectClause.isEmpty() - ? Selection.wildcard(table) - : Selection.fromSelectors(table, selectClause, boundNames, !parameters.groups.isEmpty()); + List<Selectable> selectables = RawSelector.toSelectables(selectClause, table); + boolean containsOnlyStaticColumns = selectOnlyStaticColumns(table, selectables); + + StatementRestrictions restrictions = prepareRestrictions(table, boundNames, containsOnlyStaticColumns, forView); - StatementRestrictions restrictions = prepareRestrictions(table, boundNames, selection, forView); + // If we order post-query, the sorted column needs to be in the ResultSet for sorting, + // even if we don't ultimately ship them to the client (CASSANDRA-4911). + Map<ColumnMetadata, Boolean> orderingColumns = getOrderingColumns(table); + Set<ColumnMetadata> resultSetOrderingColumns = restrictions.keyIsInRelation() ? orderingColumns.keySet() + : Collections.emptySet(); + + Selection selection = selectables.isEmpty() + ? Selection.wildcard(table, parameters.isJson) + : Selection.fromSelectors(table, + selectables, + boundNames, + resultSetOrderingColumns, + restrictions.nonPKRestrictedColumns(false), + !parameters.groups.isEmpty(), + parameters.isJson); if (parameters.isDistinct) { @@ -960,12 +984,12 @@ public class SelectStatement implements CQLStatement Comparator<List<ByteBuffer>> orderingComparator = null; boolean isReversed = false; - if (!parameters.orderings.isEmpty()) + if (!orderingColumns.isEmpty()) { assert !forView; verifyOrderingIsAllowed(restrictions); - orderingComparator = getOrderingComparator(table, selection, restrictions); - isReversed = isReversed(table); + orderingComparator = getOrderingComparator(table, selection, restrictions, orderingColumns); + isReversed = isReversed(table, orderingColumns); if (isReversed) orderingComparator = Collections.reverseOrder(orderingComparator); } @@ -987,24 +1011,58 @@ public class SelectStatement implements CQLStatement } /** + * Checks if the specified selectables select only partition key columns or static columns + * + * @param table the table metadata + * @param selectables the selectables to check + * @return {@code true} if the specified selectables select only partition key columns or static columns, + * {@code false} otherwise. + */ + private boolean selectOnlyStaticColumns(TableMetadata table, List<Selectable> selectables) + { + if (!table.hasStaticColumns() || selectables.isEmpty()) + return false; + + return Selectable.selectColumns(selectables, (column) -> column.isStatic()) + && !Selectable.selectColumns(selectables, (column) -> !column.isPartitionKey() && !column.isStatic()); + } + + /** + * Returns the columns used to order the data. + * @return the columns used to order the data. + */ + private Map<ColumnMetadata, Boolean> getOrderingColumns(TableMetadata table) + { + if (parameters.orderings.isEmpty()) + return Collections.emptyMap(); + + Map<ColumnMetadata, Boolean> orderingColumns = new LinkedHashMap<>(); + for (Map.Entry<ColumnMetadata.Raw, Boolean> entry : parameters.orderings.entrySet()) + { + orderingColumns.put(entry.getKey().prepare(table), entry.getValue()); + } + return orderingColumns; + } + + /** * Prepares the restrictions. * * @param metadata the column family meta data * @param boundNames the variable specifications - * @param selection the selection + * @param selectsOnlyStaticColumns {@code true} if the query select only static columns, {@code false} otherwise. * @return the restrictions * @throws InvalidRequestException if a problem occurs while building the restrictions */ private StatementRestrictions prepareRestrictions(TableMetadata metadata, VariableSpecifications boundNames, - Selection selection, + boolean selectsOnlyStaticColumns, boolean forView) throws InvalidRequestException { return new StatementRestrictions(StatementType.SELECT, metadata, whereClause, boundNames, - selection.containsOnlyStaticColumns(), + selectsOnlyStaticColumns, parameters.allowFiltering, forView); } @@ -1111,20 +1169,20 @@ public class SelectStatement implements CQLStatement private Comparator<List<ByteBuffer>> getOrderingComparator(TableMetadata metadata, Selection selection, - StatementRestrictions restrictions) + StatementRestrictions restrictions, + Map<ColumnMetadata, Boolean> orderingColumns) throws InvalidRequestException { if (!restrictions.keyIsInRelation()) return null; - Map<ColumnIdentifier, Integer> orderingIndexes = getOrderingIndex(metadata, selection); + Map<ColumnIdentifier, Integer> orderingIndexes = getOrderingIndex(metadata, selection, orderingColumns); List<Integer> idToSort = new ArrayList<Integer>(); List<Comparator<ByteBuffer>> sorters = new ArrayList<Comparator<ByteBuffer>>(); - for (ColumnMetadata.Raw raw : parameters.orderings.keySet()) + for (ColumnMetadata orderingColumn : orderingColumns.keySet()) { - ColumnMetadata orderingColumn = raw.prepare(metadata); idToSort.add(orderingIndexes.get(orderingColumn.name)); sorters.add(orderingColumn.type); } @@ -1132,31 +1190,26 @@ public class SelectStatement implements CQLStatement : new CompositeComparator(sorters, idToSort); } - private Map<ColumnIdentifier, Integer> getOrderingIndex(TableMetadata table, Selection selection) - throws InvalidRequestException + private Map<ColumnIdentifier, Integer> getOrderingIndex(TableMetadata table, + Selection selection, + Map<ColumnMetadata, Boolean> orderingColumns) { - // If we order post-query (see orderResults), the sorted column needs to be in the ResultSet for sorting, - // even if we don't - // ultimately ship them to the client (CASSANDRA-4911). Map<ColumnIdentifier, Integer> orderingIndexes = new HashMap<>(); - for (ColumnMetadata.Raw raw : parameters.orderings.keySet()) + for (ColumnMetadata def : orderingColumns.keySet()) { - final ColumnMetadata def = raw.prepare(table); int index = selection.getResultSetIndex(def); - if (index < 0) - index = selection.addColumnForOrdering(def); orderingIndexes.put(def.name, index); } return orderingIndexes; } - private boolean isReversed(TableMetadata table) throws InvalidRequestException + private boolean isReversed(TableMetadata table, Map<ColumnMetadata, Boolean> orderingColumns) throws InvalidRequestException { Boolean[] reversedMap = new Boolean[table.clusteringColumns().size()]; int i = 0; - for (Map.Entry<ColumnMetadata.Raw, Boolean> entry : parameters.orderings.entrySet()) + for (Map.Entry<ColumnMetadata, Boolean> entry : orderingColumns.entrySet()) { - ColumnMetadata def = entry.getKey().prepare(table); + ColumnMetadata def = entry.getKey(); boolean reversed = entry.getValue(); checkTrue(def.isClusteringColumn(), http://git-wip-us.apache.org/repos/asf/cassandra/blob/4ebab661/src/java/org/apache/cassandra/db/filter/ColumnFilter.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java index 6f6fc08..b568704 100644 --- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java +++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java @@ -21,11 +21,13 @@ import java.io.IOException; import java.util.*; import com.google.common.collect.Iterables; +import com.google.common.collect.Iterators; import com.google.common.collect.SortedSetMultimap; import com.google.common.collect.TreeMultimap; import org.apache.cassandra.cql3.ColumnIdentifier; import org.apache.cassandra.db.*; +import org.apache.cassandra.db.rows.Cell; import org.apache.cassandra.db.rows.CellPath; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; @@ -245,6 +247,23 @@ public class ColumnFilter } /** + * Given an iterator on the cell of a complex column, returns an iterator that only include the cells selected by + * this filter. + * + * @param column the (complex) column for which the cells are. + * @param cells the cells to filter. + * @return a filtered iterator that only include the cells from {@code cells} that are included by this filter. + */ + public Iterator<Cell> filterComplexCells(ColumnMetadata column, Iterator<Cell> cells) + { + Tester tester = newTester(column); + if (tester == null) + return cells; + + return Iterators.filter(cells, cell -> tester.fetchedCellIsQueried(cell.path())); + } + + /** * Returns a {@code ColumnFilter}} builder that fetches all regular columns (and queries the columns * added to the builder, or everything if no column is added). */ @@ -317,6 +336,12 @@ public class ColumnFilter * all columns, not querying none (but if you know you want to query all columns, prefer * {@link ColumnFilter#all(TableMetadata)}. For selectionBuilder, adding no queried columns means no column will be * fetched (so the builder will return {@code PartitionColumns.NONE}). + * + * Also, if only a subselection of a complex column should be queried, then only the corresponding + * subselection method of the builder ({@link #slice} or {@link #select}) should be called for the + * column, but {@link #add} shouldn't. if {@link #add} is also called, the whole column will be + * queried and the subselection(s) will be ignored. This is done for correctness of CQL where + * if you do "SELECT m, m[2..5]", you are really querying the whole collection. */ public static class Builder { @@ -324,6 +349,8 @@ public class ColumnFilter private RegularAndStaticColumns.Builder queriedBuilder; private List<ColumnSubselection> subSelections; + private Set<ColumnMetadata> fullySelectedComplexColumns; + private Builder(TableMetadata metadata) { this.metadata = metadata; @@ -331,23 +358,38 @@ public class ColumnFilter public Builder add(ColumnMetadata c) { - if (queriedBuilder == null) - queriedBuilder = RegularAndStaticColumns.builder(); - queriedBuilder.add(c); - return this; + if (c.isComplex() && c.type.isMultiCell()) + { + if (fullySelectedComplexColumns == null) + fullySelectedComplexColumns = new HashSet<>(); + fullySelectedComplexColumns.add(c); + } + return addInternal(c); } public Builder addAll(Iterable<ColumnMetadata> columns) { + for (ColumnMetadata column : columns) + add(column); + return this; + } + + private Builder addInternal(ColumnMetadata c) + { + if (c.isPrimaryKeyColumn()) + return this; + if (queriedBuilder == null) queriedBuilder = RegularAndStaticColumns.builder(); - queriedBuilder.addAll(columns); + queriedBuilder.add(c); return this; } private Builder addSubSelection(ColumnSubselection subSelection) { - add(subSelection.column()); + ColumnMetadata column = subSelection.column(); + assert column.isComplex() && column.type.isMultiCell(); + addInternal(column); if (subSelections == null) subSelections = new ArrayList<>(); subSelections.add(subSelection); @@ -379,7 +421,10 @@ public class ColumnFilter { s = TreeMultimap.create(Comparator.<ColumnIdentifier>naturalOrder(), Comparator.<ColumnSubselection>naturalOrder()); for (ColumnSubselection subSelection : subSelections) - s.put(subSelection.column().name, subSelection); + { + if (fullySelectedComplexColumns == null || !fullySelectedComplexColumns.contains(subSelection.column())) + s.put(subSelection.column().name, subSelection); + } } return new ColumnFilter(isFetchAll, metadata, queried, s); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
