Merge commit '95d0b671d1af154eaf1c1e81992c7f3f51469eee' into cassandra-3.11
* commit '95d0b671d1af154eaf1c1e81992c7f3f51469eee': CQL often queries static columns unnecessarily Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/1e067746 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/1e067746 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/1e067746 Branch: refs/heads/trunk Commit: 1e067746e432dc0a450ad111a8ec545011bb5bc7 Parents: d19b6d8 95d0b67 Author: Sylvain Lebresne <[email protected]> Authored: Fri Dec 9 11:25:24 2016 +0100 Committer: Sylvain Lebresne <[email protected]> Committed: Fri Dec 9 11:27:30 2016 +0100 ---------------------------------------------------------------------- .../cql3/statements/SelectStatement.java | 26 ++++-- .../cassandra/db/filter/ColumnFilter.java | 92 +++++++++++++------- .../org/apache/cassandra/db/rows/BTreeRow.java | 2 +- .../apache/cassandra/schema/SchemaKeyspace.java | 2 +- .../apache/cassandra/db/ReadCommandTest.java | 2 +- 5 files changed, 84 insertions(+), 40 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/1e067746/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 038d4bd,f2aa030..1744e70 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@@ -782,6 -740,20 +788,18 @@@ public class SelectStatement implement } } + // Determines whether, when we have a partition result with not rows, we still return the static content (as a + // result set row with null for all other regular columns.) + private boolean returnStaticContentOnPartitionWithNoRows() + { - // The general rational is that if some rows are specifically selected by the query, we ignore partitions that - // are empty outside of static content, but if it's a full partition query, then we include that content. - // In practice, we consider rows are specifically selected if either there is some restrictions on the - // clustering columns or it's a 2ndary index query (the later is debatable but historical). An exception however - // is 'static compact' table, for which 2ndary index indexes full partition (and so for which we consider 2ndary - // indexquery to be full partition query). - return !restrictions.hasClusteringColumnsRestriction() - && (!restrictions.usesSecondaryIndexing() || cfm.isStaticCompactTable()); ++ // The general rational is that if some rows are specifically selected by the query (have a clustering columns ++ // restrictions), we ignore partitions that are empty outside of static content, but if it's a full partition ++ // query, then we include that content. ++ // We make an exception for "static compact" table are from a CQL standpoint we always want to show their static ++ // content for backward compatiblity. ++ return !restrictions.hasClusteringColumnsRestriction() || cfm.isStaticCompactTable(); + } + // Used by ModificationStatement for CAS operations void processPartition(RowIterator partition, QueryOptions options, Selection.ResultSetBuilder result, int nowInSec) throws InvalidRequestException @@@ -791,14 -763,13 +809,12 @@@ ByteBuffer[] keyComponents = getComponents(cfm, partition.partitionKey()); Row staticRow = partition.staticRow(); - // If there is no rows, and there's no restriction on clustering/regular columns, - // then provided the select was a full partition selection (either by partition key and/or by static column), - // If there is no rows, then provided the select was a full partition selection -- // we want to include static columns and we're done. ++ // If there is no rows, we include the static content if we should and we're done. if (!partition.hasNext()) { - if (!staticRow.isEmpty() && (!restrictions.hasClusteringColumnsRestriction() || cfm.isStaticCompactTable())) + if (!staticRow.isEmpty() && returnStaticContentOnPartitionWithNoRows()) { - result.newRow(protocolVersion); + result.newRow(partition.partitionKey(), staticRow.clustering()); for (ColumnDefinition def : selection.getColumns()) { switch (def.kind) http://git-wip-us.apache.org/repos/asf/cassandra/blob/1e067746/src/java/org/apache/cassandra/db/filter/ColumnFilter.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/db/filter/ColumnFilter.java index 20f35df,8d4f8b8..0dd0aac --- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java +++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java @@@ -35,52 -36,42 +35,53 @@@ import org.apache.cassandra.io.util.Dat * Represents which (non-PK) columns (and optionally which sub-part of a column for complex columns) are selected * by a query. * - * In practice, this class cover 2 main cases: - * 1) most user queries have to internally query all (regular) columns, because the CQL semantic requires us to know - * if a row is live or not even if it has no values for the columns requested by the user (see #6588 for more - * details). However, while we need to know for columns if it has live values, we can actually save from - * sending the values for those columns that will not be returned to the user. - * 2) for some internal queries (and for queries using #6588 if we introduce it), we're actually fine only - * actually querying some of the columns. + * We distinguish 2 sets of columns in practice: the _fetched_ columns, which are the columns that we (may, see + * below) need to fetch internally, and the _queried_ columns, which are the columns that the user has selected + * in its request. * - * For complex columns, this class allows to be more fine grained than the column by only selection some of the - * cells of the complex column (either individual cell by path name, or some slice). + * The reason for distinguishing those 2 sets is that due to the CQL semantic (see #6588 for more details), we - * often need to internally fetch all columns for the queried table, but can still do some optimizations for those - * columns that are not directly queried by the user (see #10657 for more details). ++ * often need to internally fetch all regular columns for the queried table, but can still do some optimizations for ++ * those columns that are not directly queried by the user (see #10657 for more details). + * + * Note that in practice: + * - the _queried_ columns set is always included in the _fetched_ one. - * - whenever those sets are different, we know the _fetched_ set contains all columns for the table, so we - * don't have to record this set, we just keep a pointer to the table metadata. The only set we concretely - * store is thus the _queried_ one. ++ * - whenever those sets are different, we know 1) the _fetched_ set contains all regular columns for the table and 2) ++ * _fetched_ == _queried_ for static columns, so we don't have to record this set, we just keep a pointer to the ++ * table metadata. The only set we concretely store is thus the _queried_ one. + * - in the special case of a {@code SELECT *} query, we want to query all columns, and _fetched_ == _queried. + * As this is a common case, we special case it by keeping the _queried_ set {@code null} (and we retrieve + * the columns through the metadata pointer). + * + * For complex columns, this class optionally allows to specify a subset of the cells to query for each column. + * We can either select individual cells by path name, or a slice of them. Note that this is a sub-selection of + * _queried_ cells, so if _fetched_ != _queried_, then the cell selected by this sub-selection are considered + * queried and the other ones are considered fetched (and if a column has some sub-selection, it must be a queried + * column, which is actually enforced by the Builder below). */ public class ColumnFilter { public static final Serializer serializer = new Serializer(); - // True if _fetched_ is all the columns, in which case metadata must not be null. If false, - // then _fetched_ == _queried_ and we only store _queried_. - // Distinguish between the 2 cases described above: if 'isFetchAll' is true, then all regular columns will be - // retrieved by the query. If selection is also null, then all static columns will be fetched too. If 'isFetchAll' - // is true and selection is not null, then 1) for static columns, only the ones in selection are read and 2) for - // regular columns, while all are fetches, the values for column/cells not selected by 'selection' and - // 'subSelections' will be skipped. - // Otherwise, only the column/cells returned by 'selection' and 'subSelections' will be returned at all. -- private final boolean isFetchAll; ++ // True if _fetched_ includes all regular columns (an any static in _queried_), in which case metadata must not be ++ // null. If false, then _fetched_ == _queried_ and we only store _queried_. ++ private final boolean fetchAllRegulars; -- private final CFMetaData metadata; // can be null if !isFetchAll ++ private final CFMetaData metadata; // can be null if !fetchAllRegulars - private final PartitionColumns queried; // can be null if isFetchAll and _fetched_ == _queried_ - private final PartitionColumns selection; // can be null if isFetchAll and we don't want to skip any value ++ private final PartitionColumns queried; // can be null if fetchAllRegulars, to represent a wildcard query (all ++ // static and regular columns are both _fetched_ and _queried_). private final SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections; // can be null -- private ColumnFilter(boolean isFetchAll, ++ private ColumnFilter(boolean fetchAllRegulars, CFMetaData metadata, - PartitionColumns columns, + PartitionColumns queried, SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections) { - assert !isFetchAll || metadata != null; - assert isFetchAll || queried != null; -- this.isFetchAll = isFetchAll; ++ assert !fetchAllRegulars || metadata != null; ++ assert fetchAllRegulars || queried != null; ++ this.fetchAllRegulars = fetchAllRegulars; this.metadata = metadata; - this.selection = columns; + this.queried = queried; this.subSelections = subSelections; } @@@ -111,73 -105,75 +112,99 @@@ */ public PartitionColumns fetchedColumns() { - return isFetchAll ? metadata.partitionColumns() : queried; - if (!isFetchAll) - return selection; ++ if (!fetchAllRegulars) ++ return queried; + - // We always fetch all regulars, but only fetch the statics in selection. Unless selection is null, in which ++ // We always fetch all regulars, but only fetch the statics in queried. Unless queried == null, in which + // case it's a wildcard and we fetch everything. + PartitionColumns all = metadata.partitionColumns(); - return selection == null || all.statics.isEmpty() ++ return queried == null || all.statics.isEmpty() + ? all - : new PartitionColumns(selection.statics, all.regulars); ++ : new PartitionColumns(queried.statics, all.regulars); } - public boolean includesAllColumns(boolean isStatic) + /** + * The columns actually queried by the user. + * <p> + * Note that this is in general not all the columns that are fetched internally (see {@link #fetchedColumns}). + */ + public PartitionColumns queriedColumns() { - assert queried != null || isFetchAll; - // Static columns are never all included, unless selection == null - return isStatic ? selection == null : isFetchAll; ++ assert queried != null || fetchAllRegulars; + return queried == null ? metadata.partitionColumns() : queried; } - public boolean fetchesAllColumns() + /** - * Whether the provided column is selected by this selection. ++ * Wether all the (regular or static) columns are fetched by this filter. ++ * <p> ++ * Note that this method is meant as an optimization but a negative return ++ * shouldn't be relied upon strongly: this can return {@code false} but ++ * still have all the columns fetches if those were manually selected by the ++ * user. The goal here is to cheaply avoid filtering things on wildcard ++ * queries, as those are common. ++ * ++ * @param isStatic whether to check for static columns or not. If {@code true}, ++ * the method returns if all static columns are fetched, otherwise it checks ++ * regular columns. + */ - public boolean includes(ColumnDefinition column) ++ public boolean fetchesAllColumns(boolean isStatic) { - return isFetchAll; - // For statics, it is included only if it's part of selection, or if selection is null (wildcard query). - if (column.isStatic()) - return selection == null || selection.contains(column); - - // For regulars, if 'isFetchAll', then it's included automatically. Otherwise, it depends on 'selection'. - return isFetchAll || selection.contains(column); ++ return isStatic ? queried == null : fetchAllRegulars; } /** - * Whether we can skip the value for the provided selected column. + * Whether _fetched_ == _queried_ for this filter, and so if the {@code isQueried()} methods + * can return {@code false} for some column/cell. */ - public boolean canSkipValue(ColumnDefinition column) + public boolean allFetchedColumnsAreQueried() { - return !isFetchAll || (queried == null && subSelections == null); - // We don't use that currently, see #10655 for more details. - return false; ++ return !fetchAllRegulars || queried == null; } /** - * Whether the provided cell of a complex column is selected by this selection. + * Whether the provided column is fetched by this filter. */ - public boolean includes(Cell cell) + public boolean fetches(ColumnDefinition column) { - return isFetchAll || queried.contains(column); - if (isFetchAll || subSelections == null || !cell.column().isComplex()) - return true; - - SortedSet<ColumnSubselection> s = subSelections.get(cell.column().name); - if (s.isEmpty()) - return true; ++ // For statics, it is included only if it's part of _queried_, or if _queried_ is null (wildcard query). ++ if (column.isStatic()) ++ return queried == null || queried.contains(column); + - for (ColumnSubselection subSel : s) - if (subSel.compareInclusionOf(cell.path()) == 0) - return true; ++ // For regulars, if 'fetchAllRegulars', then it's included automatically. Otherwise, it depends on _queried_. ++ return fetchAllRegulars || queried.contains(column); + } - return false; + /** + * Whether the provided column, which is assumed to be _fetched_ by this filter (so the caller must guarantee + * that {@code fetches(column) == true}, is also _queried_ by the user. + * + * !WARNING! please be sure to understand the difference between _fetched_ and _queried_ + * columns that this class made before using this method. If unsure, you probably want + * to use the {@link #fetches} method. + */ + public boolean fetchedColumnIsQueried(ColumnDefinition column) + { - return !isFetchAll || queried == null || queried.contains(column); ++ return !fetchAllRegulars || queried == null || queried.contains(column); } /** - * Whether we can skip the value of the cell of a complex column. + * Whether the provided complex cell (identified by its column and path), which is assumed to be _fetched_ by + * this filter, is also _queried_ by the user. + * + * !WARNING! please be sure to understand the difference between _fetched_ and _queried_ + * columns that this class made before using this method. If unsure, you probably want + * to use the {@link #fetches} method. */ - public boolean canSkipValue(ColumnDefinition column, CellPath path) + public boolean fetchedCellIsQueried(ColumnDefinition column, CellPath path) { - if (!isFetchAll || subSelections == null || !column.isComplex()) - return false; + assert path != null; - if (!isFetchAll || subSelections == null) ++ if (!fetchAllRegulars || subSelections == null) + return true; SortedSet<ColumnSubselection> s = subSelections.get(column.name); + // No subsection for this column means everything is queried if (s.isEmpty()) - return false; + return true; for (ColumnSubselection subSel : s) if (subSel.compareInclusionOf(path) == 0) @@@ -187,11 -183,13 +214,12 @@@ } /** - * Creates a new {@code Tester} to efficiently test the inclusion of cells - * of an included complex column. + * Creates a new {@code Tester} to efficiently test the inclusion of cells of complex column + * {@code column}. * - * @param column the complex column, which *must* be included by this - * filter (that is, we must have {@code this.includes(column)}). - * @retun the created tester or {@code null} if all the cells from {@code - * column} are included. ++ * @param column for complex column for which to create a tester. + * @return the created tester or {@code null} if all the cells from the provided column + * are queried. */ public Tester newTester(ColumnDefinition column) { @@@ -202,14 -200,15 +230,14 @@@ if (s.isEmpty()) return null; - return new Tester(isFetchAll, s.iterator()); - // isFetchAll only imply everything if fetches for regular - return new Tester(isFetchAll && !column.isStatic(), s.iterator()); ++ return new Tester(!column.isStatic() && fetchAllRegulars, s.iterator()); } /** - * Returns a {@code ColumnFilter}} builder that fetches all columns (and queries the columns - * Returns a {@code ColumnFilter}} builder that includes all columns (so the selections - * added to the builder are the columns/cells for which we shouldn't skip the values). ++ * 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). */ - public static Builder allColumnsBuilder(CFMetaData metadata) + public static Builder allRegularColumnsBuilder(CFMetaData metadata) { return new Builder(metadata); } @@@ -224,7 -224,7 +252,7 @@@ public static class Tester { - private final boolean isFetchAll; - private final boolean isFetched; // if true, all cells are included ++ private final boolean isFetched; private ColumnSubselection current; private final Iterator<ColumnSubselection> iterator; @@@ -234,20 -234,29 +262,20 @@@ this.iterator = iterator; } - public boolean includes(CellPath path) + public boolean fetches(CellPath path) { - return isFetchAll || hasSubselection(path); - // It's included if either all cells are fetched (because it's a - // regular column and the filter has 'isFetchAll == true'), or if - // it's explicitely selected. - return isFetched || includedBySubselection(path); ++ return isFetched || hasSubselection(path); } /** - * Must only be called if {@code includes(path) == true}. + * Must only be called if {@code fetches(path) == true}. */ - public boolean canSkipValue(CellPath path) + public boolean fetchedCellIsQueried(CellPath path) { - return !isFetchAll || hasSubselection(path); - // We can skip the value of an included column only if it's a - // regular column included due to the 'isFetchAll' flag, but which - // isn't explicitely selected. In practice, it's enough to not have - // the path explicitly selected as it implies the column was - // included due to 'isFetchAll' (since we require includes(path) to - // be called first). - return !includedBySubselection(path); ++ return !isFetched || hasSubselection(path); } - private boolean includedBySubselection(CellPath path) + private boolean hasSubselection(CellPath path) { while (current != null || iterator.hasNext()) { @@@ -350,13 -347,13 +378,13 @@@ @Override public String toString() { -- if (isFetchAll) ++ if (fetchAllRegulars && queried == null) return "*"; - if (selection.isEmpty()) + if (queried.isEmpty()) return ""; - Iterator<ColumnDefinition> defs = selection.selectOrderIterator(); + Iterator<ColumnDefinition> defs = queried.selectOrderIterator(); if (!defs.hasNext()) return "<none>"; @@@ -398,8 -392,8 +426,8 @@@ private static int makeHeaderByte(ColumnFilter selection) { -- return (selection.isFetchAll ? IS_FETCH_ALL_MASK : 0) - | (selection.selection != null ? HAS_SELECTION_MASK : 0) ++ return (selection.fetchAllRegulars ? IS_FETCH_ALL_MASK : 0) + | (selection.queried != null ? HAS_QUERIED_MASK : 0) | (selection.subSelections != null ? HAS_SUB_SELECTIONS_MASK : 0); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/1e067746/src/java/org/apache/cassandra/db/rows/BTreeRow.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/db/rows/BTreeRow.java index 0eccb6e,18f3dec..54da16b --- a/src/java/org/apache/cassandra/db/rows/BTreeRow.java +++ b/src/java/org/apache/cassandra/db/rows/BTreeRow.java @@@ -263,12 -237,10 +263,12 @@@ public class BTreeRow extends AbstractR { Map<ByteBuffer, CFMetaData.DroppedColumn> droppedColumns = metadata.getDroppedColumns(); - boolean mayFilterColumns = !filter.fetchesAllColumns() || !filter.allFetchedColumnsAreQueried(); - if (filter.includesAllColumns(isStatic()) && (activeDeletion.isLive() || deletion.supersedes(activeDeletion)) && droppedColumns.isEmpty()) ++ boolean mayFilterColumns = !filter.fetchesAllColumns(isStatic()); + boolean mayHaveShadowed = activeDeletion.supersedes(deletion.time()); + + if (!mayFilterColumns && !mayHaveShadowed && droppedColumns.isEmpty()) return this; - boolean mayHaveShadowed = activeDeletion.supersedes(deletion.time()); LivenessInfo newInfo = primaryKeyLivenessInfo; Deletion newDeletion = deletion; http://git-wip-us.apache.org/repos/asf/cassandra/blob/1e067746/src/java/org/apache/cassandra/schema/SchemaKeyspace.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/schema/SchemaKeyspace.java index 2dffe58,84a5e13..1617aa7 --- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java +++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java @@@ -376,29 -374,9 +376,29 @@@ public final class SchemaKeyspac } } - private static ByteBuffer getSchemaKSKey(String ksName) - { - return AsciiType.instance.fromString(ksName); + /** + * Creates a PartitionUpdate from a partition containing some schema table content. + * This is mainly calling {@code PartitionUpdate.fromIterator} except for the fact that it deals with + * the problem described in #12236. + */ + private static PartitionUpdate makeUpdateForSchema(UnfilteredRowIterator partition, ColumnFilter filter) + { + // This method is used during schema migration tasks, and if cdc is disabled, we want to force excluding the + // 'cdc' column from the TABLES schema table because it is problematic if received by older nodes (see #12236 + // and #12697). Otherwise though, we just simply "buffer" the content of the partition into a PartitionUpdate. + if (DatabaseDescriptor.isCDCEnabled() || !partition.metadata().cfName.equals(TABLES)) + return PartitionUpdate.fromIterator(partition, filter); + + // We want to skip the 'cdc' column. A simple solution for that is based on the fact that + // 'PartitionUpdate.fromIterator()' will ignore any columns that are marked as 'fetched' but not 'queried'. - ColumnFilter.Builder builder = ColumnFilter.allColumnsBuilder(partition.metadata()); ++ ColumnFilter.Builder builder = ColumnFilter.allRegularColumnsBuilder(partition.metadata()); + for (ColumnDefinition column : filter.fetchedColumns()) + { + if (!column.name.toString().equals("cdc")) + builder.add(column); + } + + return PartitionUpdate.fromIterator(partition, builder.build()); } private static boolean isSystemKeyspaceSchemaPartition(DecoratedKey partitionKey) http://git-wip-us.apache.org/repos/asf/cassandra/blob/1e067746/test/unit/org/apache/cassandra/db/ReadCommandTest.java ---------------------------------------------------------------------- diff --cc test/unit/org/apache/cassandra/db/ReadCommandTest.java index 2aef2a7,0000000..9b7775da mode 100644,000000..100644 --- a/test/unit/org/apache/cassandra/db/ReadCommandTest.java +++ b/test/unit/org/apache/cassandra/db/ReadCommandTest.java @@@ -1,311 -1,0 +1,311 @@@ +/* + * 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. + */ + +package org.apache.cassandra.db; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; + +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.Util; +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter; +import org.apache.cassandra.db.filter.ColumnFilter; +import org.apache.cassandra.db.filter.DataLimits; +import org.apache.cassandra.db.filter.RowFilter; +import org.apache.cassandra.db.marshal.AsciiType; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.db.partitions.FilteredPartition; +import org.apache.cassandra.db.partitions.PartitionIterator; +import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; +import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators; +import org.apache.cassandra.db.rows.Row; +import org.apache.cassandra.db.rows.RowIterator; +import org.apache.cassandra.db.rows.SerializationHelper; +import org.apache.cassandra.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.db.rows.UnfilteredRowIterators; +import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.io.util.DataInputBuffer; +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; + +import static org.junit.Assert.assertEquals; + +public class ReadCommandTest +{ + private static final String KEYSPACE = "ReadCommandTest"; + private static final String CF1 = "Standard1"; + private static final String CF2 = "Standard2"; + private static final String CF3 = "Standard3"; + + @BeforeClass + public static void defineSchema() throws ConfigurationException + { + DatabaseDescriptor.daemonInitialization(); + + CFMetaData metadata1 = SchemaLoader.standardCFMD(KEYSPACE, CF1); + + CFMetaData metadata2 = CFMetaData.Builder.create(KEYSPACE, CF2) + .addPartitionKey("key", BytesType.instance) + .addClusteringColumn("col", AsciiType.instance) + .addRegularColumn("a", AsciiType.instance) + .addRegularColumn("b", AsciiType.instance).build(); + + CFMetaData metadata3 = CFMetaData.Builder.create(KEYSPACE, CF3) + .addPartitionKey("key", BytesType.instance) + .addClusteringColumn("col", AsciiType.instance) + .addRegularColumn("a", AsciiType.instance) + .addRegularColumn("b", AsciiType.instance) + .addRegularColumn("c", AsciiType.instance) + .addRegularColumn("d", AsciiType.instance) + .addRegularColumn("e", AsciiType.instance) + .addRegularColumn("f", AsciiType.instance).build(); + + SchemaLoader.prepareServer(); + SchemaLoader.createKeyspace(KEYSPACE, + KeyspaceParams.simple(1), + metadata1, + metadata2, + metadata3); + } + + @Test + public void testPartitionRangeAbort() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF1); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key1")) + .clustering("Column1") + .add("val", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + cfs.forceBlockingFlush(); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key2")) + .clustering("Column1") + .add("val", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + ReadCommand readCommand = Util.cmd(cfs).build(); + assertEquals(2, Util.getAll(readCommand).size()); + + readCommand.abort(); + assertEquals(0, Util.getAll(readCommand).size()); + } + + @Test + public void testSinglePartitionSliceAbort() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2); + + cfs.truncateBlocking(); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key")) + .clustering("cc") + .add("a", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + cfs.forceBlockingFlush(); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key")) + .clustering("dd") + .add("a", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).build(); + + List<FilteredPartition> partitions = Util.getAll(readCommand); + assertEquals(1, partitions.size()); + assertEquals(2, partitions.get(0).rowCount()); + + readCommand.abort(); + assertEquals(0, Util.getAll(readCommand).size()); + } + + @Test + public void testSinglePartitionNamesAbort() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2); + + cfs.truncateBlocking(); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key")) + .clustering("cc") + .add("a", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + cfs.forceBlockingFlush(); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key")) + .clustering("dd") + .add("a", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).includeRow("cc").includeRow("dd").build(); + + List<FilteredPartition> partitions = Util.getAll(readCommand); + assertEquals(1, partitions.size()); + assertEquals(2, partitions.get(0).rowCount()); + + readCommand.abort(); + assertEquals(0, Util.getAll(readCommand).size()); + } + + @Test + public void testSinglePartitionGroupMerge() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF3); + + String[][][] groups = new String[][][] { + new String[][] { + new String[] { "1", "key1", "aa", "a" }, // "1" indicates to create the data, "-1" to delete the row + new String[] { "1", "key2", "bb", "b" }, + new String[] { "1", "key3", "cc", "c" } + }, + new String[][] { + new String[] { "1", "key3", "dd", "d" }, + new String[] { "1", "key2", "ee", "e" }, + new String[] { "1", "key1", "ff", "f" } + }, + new String[][] { + new String[] { "1", "key6", "aa", "a" }, + new String[] { "1", "key5", "bb", "b" }, + new String[] { "1", "key4", "cc", "c" } + }, + new String[][] { + new String[] { "-1", "key6", "aa", "a" }, + new String[] { "-1", "key2", "bb", "b" } + } + }; + + // Given the data above, when the keys are sorted and the deletions removed, we should + // get these clustering rows in this order + String[] expectedRows = new String[] { "aa", "ff", "ee", "cc", "dd", "cc", "bb"}; + + List<ByteBuffer> buffers = new ArrayList<>(groups.length); + int nowInSeconds = FBUtilities.nowInSeconds(); - ColumnFilter columnFilter = ColumnFilter.allColumnsBuilder(cfs.metadata).build(); ++ ColumnFilter columnFilter = ColumnFilter.allRegularColumnsBuilder(cfs.metadata).build(); + RowFilter rowFilter = RowFilter.create(); + Slice slice = Slice.make(ClusteringBound.BOTTOM, ClusteringBound.TOP); + ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(Slices.with(cfs.metadata.comparator, slice), false); + + for (String[][] group : groups) + { + cfs.truncateBlocking(); + + List<SinglePartitionReadCommand> commands = new ArrayList<>(group.length); + + for (String[] data : group) + { + if (data[0].equals("1")) + { + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes(data[1])) + .clustering(data[2]) + .add(data[3], ByteBufferUtil.bytes("blah")) + .build() + .apply(); + } + else + { + RowUpdateBuilder.deleteRow(cfs.metadata, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(data[1]), data[2]).apply(); + } + commands.add(SinglePartitionReadCommand.create(cfs.metadata, nowInSeconds, columnFilter, rowFilter, DataLimits.NONE, Util.dk(data[1]), sliceFilter)); + } + + cfs.forceBlockingFlush(); + + ReadQuery query = new SinglePartitionReadCommand.Group(commands, DataLimits.NONE); + + try (ReadExecutionController executionController = query.executionController(); + UnfilteredPartitionIterator iter = query.executeLocally(executionController); + DataOutputBuffer buffer = new DataOutputBuffer()) + { + UnfilteredPartitionIterators.serializerForIntraNode().serialize(iter, + columnFilter, + buffer, + MessagingService.current_version); + buffers.add(buffer.buffer()); + } + } + + // deserialize, merge and check the results are all there + List<UnfilteredPartitionIterator> iterators = new ArrayList<>(); + + for (ByteBuffer buffer : buffers) + { + try (DataInputBuffer in = new DataInputBuffer(buffer, true)) + { + iterators.add(UnfilteredPartitionIterators.serializerForIntraNode().deserialize(in, + MessagingService.current_version, + cfs.metadata, + columnFilter, + SerializationHelper.Flag.LOCAL)); + } + } + + try(PartitionIterator partitionIterator = UnfilteredPartitionIterators.mergeAndFilter(iterators, + nowInSeconds, + new UnfilteredPartitionIterators.MergeListener() + { + public UnfilteredRowIterators.MergeListener getRowMergeListener(DecoratedKey partitionKey, List<UnfilteredRowIterator> versions) + { + return null; + } + + public void close() + { + + } + })) + { + + int i = 0; + int numPartitions = 0; + while (partitionIterator.hasNext()) + { + numPartitions++; + try(RowIterator rowIterator = partitionIterator.next()) + { + while (rowIterator.hasNext()) + { + Row row = rowIterator.next(); + assertEquals("col=" + expectedRows[i++], row.clustering().toString(cfs.metadata)); + //System.out.print(row.toString(cfs.metadata, true)); + } + } + } + + assertEquals(5, numPartitions); + assertEquals(expectedRows.length, i); + } + } +}
