You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by sl...@apache.org on 2017/01/27 08:32:44 UTC
[1/6] cassandra git commit: Revert CASSANDRA-12768 (and update fix of
CASSANDRA-12694) due to upgrade regression
Repository: cassandra
Updated Branches:
refs/heads/cassandra-3.0 4bbf99372 -> e1da99a1d
refs/heads/cassandra-3.11 237e14dd9 -> b234ca3ee
refs/heads/trunk 2987a7093 -> 26eab4546
Revert CASSANDRA-12768 (and update fix of CASSANDRA-12694) due to upgrade regression
patch by Sylvain Lebresne; reviewed by Benjamin Lerer for CASSANDRA-13025
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/e1da99a1
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/e1da99a1
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/e1da99a1
Branch: refs/heads/cassandra-3.0
Commit: e1da99a1d82b5313f022307c2e110afc07e112b6
Parents: 4bbf993
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Mon Jan 23 10:08:40 2017 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 27 09:24:07 2017 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 -
.../cql3/statements/CQL3CasRequest.java | 27 ++++----
.../cql3/statements/SelectStatement.java | 26 ++------
.../db/filter/ClusteringIndexNamesFilter.java | 12 ++--
.../cassandra/db/filter/ColumnFilter.java | 67 +++++---------------
.../org/apache/cassandra/db/rows/BTreeRow.java | 2 +-
6 files changed, 38 insertions(+), 97 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 396fa3f..3796a8d 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -17,7 +17,6 @@
* Nodetool compactionstats fails with NullPointerException (CASSANDRA-13021)
* Thread local pools never cleaned up (CASSANDRA-13033)
* Set RPC_READY to false when draining or if a node is marked as shutdown (CASSANDRA-12781)
- * CQL often queries static columns unnecessarily (CASSANDRA-12768)
* Make sure sstables only get committed when it's safe to discard commit log records (CASSANDRA-12956)
* Reject default_time_to_live option when creating or altering MVs (CASSANDRA-12868)
* Nodetool should use a more sane max heap size (CASSANDRA-12739)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
index db8653d..e226a2a 100644
--- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
+++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
@@ -157,21 +157,16 @@ public class CQL3CasRequest implements CASRequest
private PartitionColumns columnsToRead()
{
- // If all our conditions are columns conditions (IF x = ?), then it's enough to query
- // the columns from the conditions. If we have a IF EXISTS or IF NOT EXISTS however,
- // we need to query all columns for the row since if the condition fails, we want to
- // return everything to the user. Static columns make this a bit more complex, in that
- // if an insert only static columns, then the existence condition applies only to the
- // static columns themselves, and so we don't want to include regular columns in that
- // case.
- if (hasExists)
- {
- PartitionColumns allColumns = cfm.partitionColumns();
- Columns statics = updatesStaticRow ? allColumns.statics : Columns.NONE;
- Columns regulars = updatesRegularRows ? allColumns.regulars : Columns.NONE;
- return new PartitionColumns(statics, regulars);
- }
- return conditionColumns;
+ PartitionColumns allColumns = cfm.partitionColumns();
+
+ // If we update static row, we won't have any conditions on regular rows.
+ // If we update regular row, we have to fetch all regular rows (which would satisfy column condition) and
+ // static rows that take part in column condition.
+ // In both cases, we're fetching enough rows to distinguish between "all conditions are nulls" and "row does not exist".
+ // We have to do this as we can't rely on row marker for that (see #6623)
+ Columns statics = updatesStaticRow ? allColumns.statics : conditionColumns.statics;
+ Columns regulars = updatesRegularRows ? allColumns.regulars : conditionColumns.regulars;
+ return new PartitionColumns(statics, regulars);
}
public SinglePartitionReadCommand readCommand(int nowInSec)
@@ -179,7 +174,7 @@ public class CQL3CasRequest implements CASRequest
assert staticConditions != null || !conditions.isEmpty();
// Fetch all columns, but query only the selected ones
- ColumnFilter columnFilter = ColumnFilter.selection(cfm, columnsToRead());
+ ColumnFilter columnFilter = ColumnFilter.selection(columnsToRead());
// With only a static condition, we still want to make the distinction between a non-existing partition and one
// that exists (has some live data) but has not static content. So we query the first live row of the partition.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/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 f2aa030..aca6146 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -141,19 +141,13 @@ public class SelectStatement implements CQLStatement
if (selection.isWildcard())
return ColumnFilter.all(cfm);
- ColumnFilter.Builder builder = ColumnFilter.allRegularColumnsBuilder(cfm);
+ ColumnFilter.Builder builder = ColumnFilter.allColumnsBuilder(cfm);
// Adds all selected columns
for (ColumnDefinition 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 (cfm.hasStaticColumns() && returnStaticContentOnPartitionWithNoRows())
- builder.addAll(cfm.partitionColumns().statics);
-
return builder.build();
}
@@ -740,20 +734,6 @@ public class SelectStatement implements CQLStatement
}
}
- // 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());
- }
-
// Used by ModificationStatement for CAS operations
void processPartition(RowIterator partition, QueryOptions options, Selection.ResultSetBuilder result, int nowInSec)
throws InvalidRequestException
@@ -764,10 +744,12 @@ public class SelectStatement implements CQLStatement
Row staticRow = partition.staticRow();
// If there is no rows, then provided the select was a full partition selection
+ // (i.e. not a 2ndary index search and there was no condition on clustering columns),
// we want to include static columns and we're done.
if (!partition.hasNext())
{
- if (!staticRow.isEmpty() && returnStaticContentOnPartitionWithNoRows())
+ if (!staticRow.isEmpty() && (!restrictions.usesSecondaryIndexing() || cfm.isStaticCompactTable())
+ && !restrictions.hasClusteringColumnsRestriction())
{
result.newRow(protocolVersion);
for (ColumnDefinition def : selection.getColumns())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java b/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
index ea5cf55..a81a7a6 100644
--- a/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
@@ -178,12 +178,12 @@ public class ClusteringIndexNamesFilter extends AbstractClusteringIndexFilter
{
final SearchIterator<Clustering, Row> searcher = partition.searchIterator(columnFilter, reversed);
return new AbstractUnfilteredRowIterator(partition.metadata(),
- partition.partitionKey(),
- partition.partitionLevelDeletion(),
- columnFilter.fetchedColumns(),
- searcher.next(Clustering.STATIC_CLUSTERING),
- reversed,
- partition.stats())
+ partition.partitionKey(),
+ partition.partitionLevelDeletion(),
+ columnFilter.fetchedColumns(),
+ searcher.next(Clustering.STATIC_CLUSTERING),
+ reversed,
+ partition.stats())
{
private final Iterator<Clustering> clusteringIter = clusteringsInQueryOrder.iterator();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/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 2377ad0..df91781 100644
--- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
@@ -37,8 +37,8 @@ import org.apache.cassandra.io.util.DataOutputPlus;
* 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
+ * 1) most user queries have to internally query all 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 #6588for 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
@@ -51,11 +51,8 @@ public class ColumnFilter
{
public static final Serializer serializer = new Serializer();
- // 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.
+ // Distinguish between the 2 cases described above: if 'isFetchAll' is true, then all columns will be retrieved
+ // by the query, but 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;
@@ -114,21 +111,12 @@ public class ColumnFilter
*/
public PartitionColumns fetchedColumns()
{
- if (!isFetchAll)
- return selection;
-
- // We always fetch all regulars, but only fetch the statics in selection. Unless selection is null, in which
- // case it's a wildcard and we fetch everything.
- PartitionColumns all = metadata.partitionColumns();
- return selection == null || all.statics.isEmpty()
- ? all
- : new PartitionColumns(selection.statics, all.regulars);
+ return isFetchAll ? metadata.partitionColumns() : selection;
}
- public boolean includesAllColumns(boolean isStatic)
+ public boolean includesAllColumns()
{
- // Static columns are never all included, unless selection == null
- return isStatic ? selection == null : isFetchAll;
+ return isFetchAll;
}
/**
@@ -136,11 +124,6 @@ public class ColumnFilter
*/
public boolean includes(ColumnDefinition column)
{
- // 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);
}
@@ -192,13 +175,8 @@ public class ColumnFilter
}
/**
- * Creates a new {@code Tester} to efficiently test the inclusion of cells
- * of an included complex 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.
+ * Creates a new {@code Tester} to efficiently test the inclusion of cells of complex column
+ * {@code column}.
*/
public Tester newTester(ColumnDefinition column)
{
@@ -209,15 +187,14 @@ public class ColumnFilter
if (s.isEmpty())
return null;
- // isFetchAll only imply everything if fetches for regular
- return new Tester(isFetchAll && !column.isStatic(), s.iterator());
+ return new Tester(isFetchAll, s.iterator());
}
/**
* 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).
*/
- public static Builder allRegularColumnsBuilder(CFMetaData metadata)
+ public static Builder allColumnsBuilder(CFMetaData metadata)
{
return new Builder(metadata);
}
@@ -233,36 +210,24 @@ public class ColumnFilter
public static class Tester
{
- private final boolean isFetched; // if true, all cells are included
+ private final boolean isFetchAll;
private ColumnSubselection current;
private final Iterator<ColumnSubselection> iterator;
- private Tester(boolean isFetched, Iterator<ColumnSubselection> iterator)
+ private Tester(boolean isFetchAll, Iterator<ColumnSubselection> iterator)
{
- this.isFetched = isFetched;
+ this.isFetchAll = isFetchAll;
this.iterator = iterator;
}
public boolean includes(CellPath 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 isFetchAll || includedBySubselection(path);
}
- /**
- * Must only be called if {@code includes(path) == true}.
- */
public boolean canSkipValue(CellPath 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 isFetchAll && !includedBySubselection(path);
}
private boolean includedBySubselection(CellPath path)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/src/java/org/apache/cassandra/db/rows/BTreeRow.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/rows/BTreeRow.java b/src/java/org/apache/cassandra/db/rows/BTreeRow.java
index 18f3dec..ea1d9e0 100644
--- a/src/java/org/apache/cassandra/db/rows/BTreeRow.java
+++ b/src/java/org/apache/cassandra/db/rows/BTreeRow.java
@@ -237,7 +237,7 @@ public class BTreeRow extends AbstractRow
{
Map<ByteBuffer, CFMetaData.DroppedColumn> droppedColumns = metadata.getDroppedColumns();
- if (filter.includesAllColumns(isStatic()) && (activeDeletion.isLive() || deletion.supersedes(activeDeletion)) && droppedColumns.isEmpty())
+ if (filter.includesAllColumns() && (activeDeletion.isLive() || deletion.supersedes(activeDeletion)) && droppedColumns.isEmpty())
return this;
boolean mayHaveShadowed = activeDeletion.supersedes(deletion.time());
[2/6] cassandra git commit: Revert CASSANDRA-12768 (and update fix of
CASSANDRA-12694) due to upgrade regression
Posted by sl...@apache.org.
Revert CASSANDRA-12768 (and update fix of CASSANDRA-12694) due to upgrade regression
patch by Sylvain Lebresne; reviewed by Benjamin Lerer for CASSANDRA-13025
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/e1da99a1
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/e1da99a1
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/e1da99a1
Branch: refs/heads/cassandra-3.11
Commit: e1da99a1d82b5313f022307c2e110afc07e112b6
Parents: 4bbf993
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Mon Jan 23 10:08:40 2017 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 27 09:24:07 2017 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 -
.../cql3/statements/CQL3CasRequest.java | 27 ++++----
.../cql3/statements/SelectStatement.java | 26 ++------
.../db/filter/ClusteringIndexNamesFilter.java | 12 ++--
.../cassandra/db/filter/ColumnFilter.java | 67 +++++---------------
.../org/apache/cassandra/db/rows/BTreeRow.java | 2 +-
6 files changed, 38 insertions(+), 97 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 396fa3f..3796a8d 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -17,7 +17,6 @@
* Nodetool compactionstats fails with NullPointerException (CASSANDRA-13021)
* Thread local pools never cleaned up (CASSANDRA-13033)
* Set RPC_READY to false when draining or if a node is marked as shutdown (CASSANDRA-12781)
- * CQL often queries static columns unnecessarily (CASSANDRA-12768)
* Make sure sstables only get committed when it's safe to discard commit log records (CASSANDRA-12956)
* Reject default_time_to_live option when creating or altering MVs (CASSANDRA-12868)
* Nodetool should use a more sane max heap size (CASSANDRA-12739)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
index db8653d..e226a2a 100644
--- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
+++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
@@ -157,21 +157,16 @@ public class CQL3CasRequest implements CASRequest
private PartitionColumns columnsToRead()
{
- // If all our conditions are columns conditions (IF x = ?), then it's enough to query
- // the columns from the conditions. If we have a IF EXISTS or IF NOT EXISTS however,
- // we need to query all columns for the row since if the condition fails, we want to
- // return everything to the user. Static columns make this a bit more complex, in that
- // if an insert only static columns, then the existence condition applies only to the
- // static columns themselves, and so we don't want to include regular columns in that
- // case.
- if (hasExists)
- {
- PartitionColumns allColumns = cfm.partitionColumns();
- Columns statics = updatesStaticRow ? allColumns.statics : Columns.NONE;
- Columns regulars = updatesRegularRows ? allColumns.regulars : Columns.NONE;
- return new PartitionColumns(statics, regulars);
- }
- return conditionColumns;
+ PartitionColumns allColumns = cfm.partitionColumns();
+
+ // If we update static row, we won't have any conditions on regular rows.
+ // If we update regular row, we have to fetch all regular rows (which would satisfy column condition) and
+ // static rows that take part in column condition.
+ // In both cases, we're fetching enough rows to distinguish between "all conditions are nulls" and "row does not exist".
+ // We have to do this as we can't rely on row marker for that (see #6623)
+ Columns statics = updatesStaticRow ? allColumns.statics : conditionColumns.statics;
+ Columns regulars = updatesRegularRows ? allColumns.regulars : conditionColumns.regulars;
+ return new PartitionColumns(statics, regulars);
}
public SinglePartitionReadCommand readCommand(int nowInSec)
@@ -179,7 +174,7 @@ public class CQL3CasRequest implements CASRequest
assert staticConditions != null || !conditions.isEmpty();
// Fetch all columns, but query only the selected ones
- ColumnFilter columnFilter = ColumnFilter.selection(cfm, columnsToRead());
+ ColumnFilter columnFilter = ColumnFilter.selection(columnsToRead());
// With only a static condition, we still want to make the distinction between a non-existing partition and one
// that exists (has some live data) but has not static content. So we query the first live row of the partition.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/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 f2aa030..aca6146 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -141,19 +141,13 @@ public class SelectStatement implements CQLStatement
if (selection.isWildcard())
return ColumnFilter.all(cfm);
- ColumnFilter.Builder builder = ColumnFilter.allRegularColumnsBuilder(cfm);
+ ColumnFilter.Builder builder = ColumnFilter.allColumnsBuilder(cfm);
// Adds all selected columns
for (ColumnDefinition 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 (cfm.hasStaticColumns() && returnStaticContentOnPartitionWithNoRows())
- builder.addAll(cfm.partitionColumns().statics);
-
return builder.build();
}
@@ -740,20 +734,6 @@ public class SelectStatement implements CQLStatement
}
}
- // 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());
- }
-
// Used by ModificationStatement for CAS operations
void processPartition(RowIterator partition, QueryOptions options, Selection.ResultSetBuilder result, int nowInSec)
throws InvalidRequestException
@@ -764,10 +744,12 @@ public class SelectStatement implements CQLStatement
Row staticRow = partition.staticRow();
// If there is no rows, then provided the select was a full partition selection
+ // (i.e. not a 2ndary index search and there was no condition on clustering columns),
// we want to include static columns and we're done.
if (!partition.hasNext())
{
- if (!staticRow.isEmpty() && returnStaticContentOnPartitionWithNoRows())
+ if (!staticRow.isEmpty() && (!restrictions.usesSecondaryIndexing() || cfm.isStaticCompactTable())
+ && !restrictions.hasClusteringColumnsRestriction())
{
result.newRow(protocolVersion);
for (ColumnDefinition def : selection.getColumns())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java b/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
index ea5cf55..a81a7a6 100644
--- a/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
@@ -178,12 +178,12 @@ public class ClusteringIndexNamesFilter extends AbstractClusteringIndexFilter
{
final SearchIterator<Clustering, Row> searcher = partition.searchIterator(columnFilter, reversed);
return new AbstractUnfilteredRowIterator(partition.metadata(),
- partition.partitionKey(),
- partition.partitionLevelDeletion(),
- columnFilter.fetchedColumns(),
- searcher.next(Clustering.STATIC_CLUSTERING),
- reversed,
- partition.stats())
+ partition.partitionKey(),
+ partition.partitionLevelDeletion(),
+ columnFilter.fetchedColumns(),
+ searcher.next(Clustering.STATIC_CLUSTERING),
+ reversed,
+ partition.stats())
{
private final Iterator<Clustering> clusteringIter = clusteringsInQueryOrder.iterator();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/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 2377ad0..df91781 100644
--- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
@@ -37,8 +37,8 @@ import org.apache.cassandra.io.util.DataOutputPlus;
* 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
+ * 1) most user queries have to internally query all 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 #6588for 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
@@ -51,11 +51,8 @@ public class ColumnFilter
{
public static final Serializer serializer = new Serializer();
- // 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.
+ // Distinguish between the 2 cases described above: if 'isFetchAll' is true, then all columns will be retrieved
+ // by the query, but 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;
@@ -114,21 +111,12 @@ public class ColumnFilter
*/
public PartitionColumns fetchedColumns()
{
- if (!isFetchAll)
- return selection;
-
- // We always fetch all regulars, but only fetch the statics in selection. Unless selection is null, in which
- // case it's a wildcard and we fetch everything.
- PartitionColumns all = metadata.partitionColumns();
- return selection == null || all.statics.isEmpty()
- ? all
- : new PartitionColumns(selection.statics, all.regulars);
+ return isFetchAll ? metadata.partitionColumns() : selection;
}
- public boolean includesAllColumns(boolean isStatic)
+ public boolean includesAllColumns()
{
- // Static columns are never all included, unless selection == null
- return isStatic ? selection == null : isFetchAll;
+ return isFetchAll;
}
/**
@@ -136,11 +124,6 @@ public class ColumnFilter
*/
public boolean includes(ColumnDefinition column)
{
- // 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);
}
@@ -192,13 +175,8 @@ public class ColumnFilter
}
/**
- * Creates a new {@code Tester} to efficiently test the inclusion of cells
- * of an included complex 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.
+ * Creates a new {@code Tester} to efficiently test the inclusion of cells of complex column
+ * {@code column}.
*/
public Tester newTester(ColumnDefinition column)
{
@@ -209,15 +187,14 @@ public class ColumnFilter
if (s.isEmpty())
return null;
- // isFetchAll only imply everything if fetches for regular
- return new Tester(isFetchAll && !column.isStatic(), s.iterator());
+ return new Tester(isFetchAll, s.iterator());
}
/**
* 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).
*/
- public static Builder allRegularColumnsBuilder(CFMetaData metadata)
+ public static Builder allColumnsBuilder(CFMetaData metadata)
{
return new Builder(metadata);
}
@@ -233,36 +210,24 @@ public class ColumnFilter
public static class Tester
{
- private final boolean isFetched; // if true, all cells are included
+ private final boolean isFetchAll;
private ColumnSubselection current;
private final Iterator<ColumnSubselection> iterator;
- private Tester(boolean isFetched, Iterator<ColumnSubselection> iterator)
+ private Tester(boolean isFetchAll, Iterator<ColumnSubselection> iterator)
{
- this.isFetched = isFetched;
+ this.isFetchAll = isFetchAll;
this.iterator = iterator;
}
public boolean includes(CellPath 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 isFetchAll || includedBySubselection(path);
}
- /**
- * Must only be called if {@code includes(path) == true}.
- */
public boolean canSkipValue(CellPath 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 isFetchAll && !includedBySubselection(path);
}
private boolean includedBySubselection(CellPath path)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/src/java/org/apache/cassandra/db/rows/BTreeRow.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/rows/BTreeRow.java b/src/java/org/apache/cassandra/db/rows/BTreeRow.java
index 18f3dec..ea1d9e0 100644
--- a/src/java/org/apache/cassandra/db/rows/BTreeRow.java
+++ b/src/java/org/apache/cassandra/db/rows/BTreeRow.java
@@ -237,7 +237,7 @@ public class BTreeRow extends AbstractRow
{
Map<ByteBuffer, CFMetaData.DroppedColumn> droppedColumns = metadata.getDroppedColumns();
- if (filter.includesAllColumns(isStatic()) && (activeDeletion.isLive() || deletion.supersedes(activeDeletion)) && droppedColumns.isEmpty())
+ if (filter.includesAllColumns() && (activeDeletion.isLive() || deletion.supersedes(activeDeletion)) && droppedColumns.isEmpty())
return this;
boolean mayHaveShadowed = activeDeletion.supersedes(deletion.time());
[6/6] cassandra git commit: Merge commit
'b234ca3ee4101a34b761948d633ef4a12aa70ca2' into trunk
Posted by sl...@apache.org.
Merge commit 'b234ca3ee4101a34b761948d633ef4a12aa70ca2' into trunk
* commit 'b234ca3ee4101a34b761948d633ef4a12aa70ca2':
Revert CASSANDRA-12768 (and update fix of CASSANDRA-12694) due to upgrade regression
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/26eab454
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/26eab454
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/26eab454
Branch: refs/heads/trunk
Commit: 26eab45465a140be5ef5fae45dc8d7cbb33192b9
Parents: 2987a70 b234ca3
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Fri Jan 27 09:27:44 2017 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 27 09:28:41 2017 +0100
----------------------------------------------------------------------
.../org/apache/cassandra/db/ReadResponse.java | 26 ++++++------
.../cassandra/db/filter/ColumnFilter.java | 44 ++++++++++++++++++--
.../apache/cassandra/net/MessagingService.java | 3 +-
3 files changed, 55 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/26eab454/src/java/org/apache/cassandra/db/ReadResponse.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/ReadResponse.java
index 107bcc4,cca21f8..7cf04a4
--- a/src/java/org/apache/cassandra/db/ReadResponse.java
+++ b/src/java/org/apache/cassandra/db/ReadResponse.java
@@@ -52,7 -79,7 +52,7 @@@ public abstract class ReadRespons
@VisibleForTesting
public static ReadResponse createRemoteDataResponse(UnfilteredPartitionIterator data, ReadCommand command)
{
-- return new RemoteDataResponse(LocalDataResponse.build(data, command.columnFilter()));
++ return new RemoteDataResponse(LocalDataResponse.build(data, command.columnFilter()), MessagingService.current_version);
}
public static ReadResponse createDigestResponse(UnfilteredPartitionIterator data, ReadCommand command)
@@@ -108,7 -135,7 +108,7 @@@
{
private LocalDataResponse(UnfilteredPartitionIterator iter, ReadCommand command)
{
- super(build(iter, command.columnFilter()), SerializationHelper.Flag.LOCAL);
- super(command, build(iter, command.columnFilter()), SerializationHelper.Flag.LOCAL);
++ super(build(iter, command.columnFilter()), MessagingService.current_version, SerializationHelper.Flag.LOCAL);
}
private static ByteBuffer build(UnfilteredPartitionIterator iter, ColumnFilter selection)
@@@ -129,9 -156,9 +129,9 @@@
// built on the coordinator node receiving a response
private static class RemoteDataResponse extends DataResponse
{
-- protected RemoteDataResponse(ByteBuffer data)
++ protected RemoteDataResponse(ByteBuffer data, int version)
{
- super(data, SerializationHelper.Flag.FROM_REMOTE);
- super(null, data, SerializationHelper.Flag.FROM_REMOTE);
++ super(data, version, SerializationHelper.Flag.FROM_REMOTE);
}
}
@@@ -140,12 -167,12 +140,14 @@@
// TODO: can the digest be calculated over the raw bytes now?
// The response, serialized in the current messaging version
private final ByteBuffer data;
++ private final int dataSerializationVersion;
private final SerializationHelper.Flag flag;
- protected DataResponse(ByteBuffer data, SerializationHelper.Flag flag)
- protected DataResponse(ReadCommand command, ByteBuffer data, SerializationHelper.Flag flag)
++ protected DataResponse(ByteBuffer data, int dataSerializationVersion, SerializationHelper.Flag flag)
{
- super(command);
+ super();
this.data = data;
++ this.dataSerializationVersion = dataSerializationVersion;
this.flag = flag;
}
@@@ -157,7 -184,7 +159,7 @@@
// the later can be null (for RemoteDataResponse as those are created in the serializers and
// those don't have easy access to the command). This is also why we need the command as parameter here.
return UnfilteredPartitionIterators.serializerForIntraNode().deserialize(in,
-- MessagingService.current_version,
++ dataSerializationVersion,
command.metadata(),
command.columnFilter(),
flag);
@@@ -204,11 -377,9 +206,8 @@@
if (digest.hasRemaining())
return new DigestResponse(digest);
- // Note that we can only get there if version == 3.0, which is the current_version. When we'll change the
- // version, we'll have to deserialize/re-serialize the data to be in the proper version.
-- assert version == MessagingService.VERSION_30;
ByteBuffer data = ByteBufferUtil.readWithVIntLength(in);
-- return new RemoteDataResponse(data);
++ return new RemoteDataResponse(data, version);
}
public long serializedSize(ReadResponse response, int version)
@@@ -219,9 -390,31 +218,10 @@@
long size = ByteBufferUtil.serializedSizeWithVIntLength(digest);
if (!isDigest)
{
-- // Note that we can only get there if version == 3.0, which is the current_version. When we'll change the
-- // version, we'll have to deserialize/re-serialize the data to be in the proper version.
-- assert version == MessagingService.VERSION_30;
++ // In theory, we should deserialize/re-serialize if the version asked is different from the current
++ // version as the content could have a different serialization format. So far though, we haven't made
++ // change to partition iterators serialization since 3.0 so we skip this.
++ assert version >= MessagingService.VERSION_30;
ByteBuffer data = ((DataResponse)response).data;
size += ByteBufferUtil.serializedSizeWithVIntLength(data);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/26eab454/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/filter/ColumnFilter.java
index d320fc3,93a848e..b3ae505
--- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
@@@ -20,6 -20,6 +20,7 @@@ package org.apache.cassandra.db.filter
import java.io.IOException;
import java.util.*;
++import com.google.common.collect.Iterables;
import com.google.common.collect.SortedSetMultimap;
import com.google.common.collect.TreeMultimap;
@@@ -30,6 -30,6 +31,7 @@@ import org.apache.cassandra.db.rows.Cel
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
++import org.apache.cassandra.net.MessagingService;
/**
* Represents which (non-PK) columns (and optionally which sub-part of a column for complex columns) are selected
@@@ -62,17 -62,16 +64,17 @@@ public class ColumnFilte
{
public static final Serializer serializer = new Serializer();
- // True if _fetched_ includes all regular columns (an any static in _queried_), in which case metadata must not be
- // 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_.
- private final boolean isFetchAll;
++ // True if _fetched_ includes all regular columns (and 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 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 queried,
SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections)
@@@ -429,19 -401,19 +431,42 @@@
public static class Serializer
{
-- private static final int IS_FETCH_ALL_MASK = 0x01;
++ private static final int FETCH_ALL_MASK = 0x01;
private static final int HAS_QUERIED_MASK = 0x02;
private static final int HAS_SUB_SELECTIONS_MASK = 0x04;
private static int makeHeaderByte(ColumnFilter selection)
{
- return (selection.fetchAllRegulars ? IS_FETCH_ALL_MASK : 0)
- return (selection.isFetchAll ? IS_FETCH_ALL_MASK : 0)
++ return (selection.fetchAllRegulars ? FETCH_ALL_MASK : 0)
| (selection.queried != null ? HAS_QUERIED_MASK : 0)
| (selection.subSelections != null ? HAS_SUB_SELECTIONS_MASK : 0);
}
++ private static ColumnFilter maybeUpdateForBackwardCompatility(ColumnFilter selection, int version)
++ {
++ if (version > MessagingService.VERSION_30 || !selection.fetchAllRegulars || selection.queried == null)
++ return selection;
++
++ // The meaning of fetchAllRegulars changed (at least when queried != null) due to CASSANDRA-12768: in
++ // pre-4.0 it means that *all* columns are fetched, not just the regular ones, and so 3.0/3.X nodes
++ // would send us more than we'd like. So instead recreating a filter that correspond to what we
++ // actually want (it's a tiny bit less efficient as we include all columns manually and will mark as
++ // queried some columns that are actually only fetched, but it's fine during upgrade).
++ // More concretely, we replace our filter by a non-fetch-all one that queries every columns that our
++ // current filter fetches.
++ Columns allRegulars = selection.metadata.partitionColumns().regulars;
++ Set<ColumnDefinition> queriedStatic = new HashSet<>();
++ Iterables.addAll(queriedStatic, Iterables.filter(selection.queried, ColumnDefinition::isStatic));
++ return new ColumnFilter(false,
++ null,
++ new PartitionColumns(Columns.from(queriedStatic), allRegulars),
++ selection.subSelections);
++ }
++
public void serialize(ColumnFilter selection, DataOutputPlus out, int version) throws IOException
{
++ selection = maybeUpdateForBackwardCompatility(selection, version);
++
out.writeByte(makeHeaderByte(selection));
if (selection.queried != null)
@@@ -461,7 -433,7 +486,7 @@@
public ColumnFilter deserialize(DataInputPlus in, int version, CFMetaData metadata) throws IOException
{
int header = in.readUnsignedByte();
-- boolean isFetchAll = (header & IS_FETCH_ALL_MASK) != 0;
++ boolean isFetchAll = (header & FETCH_ALL_MASK) != 0;
boolean hasQueried = (header & HAS_QUERIED_MASK) != 0;
boolean hasSubSelections = (header & HAS_SUB_SELECTIONS_MASK) != 0;
@@@ -485,11 -457,11 +510,22 @@@
}
}
++ // Same concern than in serialize/serializedSize: we should be wary of the change in meaning for isFetchAll.
++ // If we get a filter with isFetchAll from 3.0/3.x, it actually expects all static columns to be fetched,
++ // make sure we do that (note that if queried == null, that's already what we do).
++ // Note that here again this will make us do a bit more work that necessary, namely we'll _query_ all
++ // statics even though we only care about _fetching_ them all, but that's a minor inefficiency, so fine
++ // during upgrade.
++ if (version <= MessagingService.VERSION_30 && isFetchAll && queried != null)
++ queried = new PartitionColumns(metadata.partitionColumns().statics, queried.regulars);
++
return new ColumnFilter(isFetchAll, isFetchAll ? metadata : null, queried, subSelections);
}
public long serializedSize(ColumnFilter selection, int version)
{
++ selection = maybeUpdateForBackwardCompatility(selection, version);
++
long size = 1; // header byte
if (selection.queried != null)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/26eab454/src/java/org/apache/cassandra/net/MessagingService.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/net/MessagingService.java
index 38c1cd2,f82e80b..7215397
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@@ -88,8 -88,12 +88,9 @@@ public final class MessagingService imp
public static final String MBEAN_NAME = "org.apache.cassandra.net:type=MessagingService";
// 8 bits version, so don't waste versions
- public static final int VERSION_12 = 6;
- public static final int VERSION_20 = 7;
- public static final int VERSION_21 = 8;
- public static final int VERSION_22 = 9;
public static final int VERSION_30 = 10;
-- public static final int current_version = VERSION_30;
++ public static final int VERSION_40 = 11;
++ public static final int current_version = VERSION_40;
public static final String FAILURE_CALLBACK_PARAM = "CAL_BAC";
public static final byte[] ONE_BYTE = new byte[1];
[3/6] cassandra git commit: Revert CASSANDRA-12768 (and update fix of
CASSANDRA-12694) due to upgrade regression
Posted by sl...@apache.org.
Revert CASSANDRA-12768 (and update fix of CASSANDRA-12694) due to upgrade regression
patch by Sylvain Lebresne; reviewed by Benjamin Lerer for CASSANDRA-13025
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/e1da99a1
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/e1da99a1
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/e1da99a1
Branch: refs/heads/trunk
Commit: e1da99a1d82b5313f022307c2e110afc07e112b6
Parents: 4bbf993
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Mon Jan 23 10:08:40 2017 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 27 09:24:07 2017 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 -
.../cql3/statements/CQL3CasRequest.java | 27 ++++----
.../cql3/statements/SelectStatement.java | 26 ++------
.../db/filter/ClusteringIndexNamesFilter.java | 12 ++--
.../cassandra/db/filter/ColumnFilter.java | 67 +++++---------------
.../org/apache/cassandra/db/rows/BTreeRow.java | 2 +-
6 files changed, 38 insertions(+), 97 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 396fa3f..3796a8d 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -17,7 +17,6 @@
* Nodetool compactionstats fails with NullPointerException (CASSANDRA-13021)
* Thread local pools never cleaned up (CASSANDRA-13033)
* Set RPC_READY to false when draining or if a node is marked as shutdown (CASSANDRA-12781)
- * CQL often queries static columns unnecessarily (CASSANDRA-12768)
* Make sure sstables only get committed when it's safe to discard commit log records (CASSANDRA-12956)
* Reject default_time_to_live option when creating or altering MVs (CASSANDRA-12868)
* Nodetool should use a more sane max heap size (CASSANDRA-12739)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
index db8653d..e226a2a 100644
--- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
+++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
@@ -157,21 +157,16 @@ public class CQL3CasRequest implements CASRequest
private PartitionColumns columnsToRead()
{
- // If all our conditions are columns conditions (IF x = ?), then it's enough to query
- // the columns from the conditions. If we have a IF EXISTS or IF NOT EXISTS however,
- // we need to query all columns for the row since if the condition fails, we want to
- // return everything to the user. Static columns make this a bit more complex, in that
- // if an insert only static columns, then the existence condition applies only to the
- // static columns themselves, and so we don't want to include regular columns in that
- // case.
- if (hasExists)
- {
- PartitionColumns allColumns = cfm.partitionColumns();
- Columns statics = updatesStaticRow ? allColumns.statics : Columns.NONE;
- Columns regulars = updatesRegularRows ? allColumns.regulars : Columns.NONE;
- return new PartitionColumns(statics, regulars);
- }
- return conditionColumns;
+ PartitionColumns allColumns = cfm.partitionColumns();
+
+ // If we update static row, we won't have any conditions on regular rows.
+ // If we update regular row, we have to fetch all regular rows (which would satisfy column condition) and
+ // static rows that take part in column condition.
+ // In both cases, we're fetching enough rows to distinguish between "all conditions are nulls" and "row does not exist".
+ // We have to do this as we can't rely on row marker for that (see #6623)
+ Columns statics = updatesStaticRow ? allColumns.statics : conditionColumns.statics;
+ Columns regulars = updatesRegularRows ? allColumns.regulars : conditionColumns.regulars;
+ return new PartitionColumns(statics, regulars);
}
public SinglePartitionReadCommand readCommand(int nowInSec)
@@ -179,7 +174,7 @@ public class CQL3CasRequest implements CASRequest
assert staticConditions != null || !conditions.isEmpty();
// Fetch all columns, but query only the selected ones
- ColumnFilter columnFilter = ColumnFilter.selection(cfm, columnsToRead());
+ ColumnFilter columnFilter = ColumnFilter.selection(columnsToRead());
// With only a static condition, we still want to make the distinction between a non-existing partition and one
// that exists (has some live data) but has not static content. So we query the first live row of the partition.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/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 f2aa030..aca6146 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -141,19 +141,13 @@ public class SelectStatement implements CQLStatement
if (selection.isWildcard())
return ColumnFilter.all(cfm);
- ColumnFilter.Builder builder = ColumnFilter.allRegularColumnsBuilder(cfm);
+ ColumnFilter.Builder builder = ColumnFilter.allColumnsBuilder(cfm);
// Adds all selected columns
for (ColumnDefinition 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 (cfm.hasStaticColumns() && returnStaticContentOnPartitionWithNoRows())
- builder.addAll(cfm.partitionColumns().statics);
-
return builder.build();
}
@@ -740,20 +734,6 @@ public class SelectStatement implements CQLStatement
}
}
- // 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());
- }
-
// Used by ModificationStatement for CAS operations
void processPartition(RowIterator partition, QueryOptions options, Selection.ResultSetBuilder result, int nowInSec)
throws InvalidRequestException
@@ -764,10 +744,12 @@ public class SelectStatement implements CQLStatement
Row staticRow = partition.staticRow();
// If there is no rows, then provided the select was a full partition selection
+ // (i.e. not a 2ndary index search and there was no condition on clustering columns),
// we want to include static columns and we're done.
if (!partition.hasNext())
{
- if (!staticRow.isEmpty() && returnStaticContentOnPartitionWithNoRows())
+ if (!staticRow.isEmpty() && (!restrictions.usesSecondaryIndexing() || cfm.isStaticCompactTable())
+ && !restrictions.hasClusteringColumnsRestriction())
{
result.newRow(protocolVersion);
for (ColumnDefinition def : selection.getColumns())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java b/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
index ea5cf55..a81a7a6 100644
--- a/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
@@ -178,12 +178,12 @@ public class ClusteringIndexNamesFilter extends AbstractClusteringIndexFilter
{
final SearchIterator<Clustering, Row> searcher = partition.searchIterator(columnFilter, reversed);
return new AbstractUnfilteredRowIterator(partition.metadata(),
- partition.partitionKey(),
- partition.partitionLevelDeletion(),
- columnFilter.fetchedColumns(),
- searcher.next(Clustering.STATIC_CLUSTERING),
- reversed,
- partition.stats())
+ partition.partitionKey(),
+ partition.partitionLevelDeletion(),
+ columnFilter.fetchedColumns(),
+ searcher.next(Clustering.STATIC_CLUSTERING),
+ reversed,
+ partition.stats())
{
private final Iterator<Clustering> clusteringIter = clusteringsInQueryOrder.iterator();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/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 2377ad0..df91781 100644
--- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
@@ -37,8 +37,8 @@ import org.apache.cassandra.io.util.DataOutputPlus;
* 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
+ * 1) most user queries have to internally query all 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 #6588for 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
@@ -51,11 +51,8 @@ public class ColumnFilter
{
public static final Serializer serializer = new Serializer();
- // 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.
+ // Distinguish between the 2 cases described above: if 'isFetchAll' is true, then all columns will be retrieved
+ // by the query, but 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;
@@ -114,21 +111,12 @@ public class ColumnFilter
*/
public PartitionColumns fetchedColumns()
{
- if (!isFetchAll)
- return selection;
-
- // We always fetch all regulars, but only fetch the statics in selection. Unless selection is null, in which
- // case it's a wildcard and we fetch everything.
- PartitionColumns all = metadata.partitionColumns();
- return selection == null || all.statics.isEmpty()
- ? all
- : new PartitionColumns(selection.statics, all.regulars);
+ return isFetchAll ? metadata.partitionColumns() : selection;
}
- public boolean includesAllColumns(boolean isStatic)
+ public boolean includesAllColumns()
{
- // Static columns are never all included, unless selection == null
- return isStatic ? selection == null : isFetchAll;
+ return isFetchAll;
}
/**
@@ -136,11 +124,6 @@ public class ColumnFilter
*/
public boolean includes(ColumnDefinition column)
{
- // 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);
}
@@ -192,13 +175,8 @@ public class ColumnFilter
}
/**
- * Creates a new {@code Tester} to efficiently test the inclusion of cells
- * of an included complex 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.
+ * Creates a new {@code Tester} to efficiently test the inclusion of cells of complex column
+ * {@code column}.
*/
public Tester newTester(ColumnDefinition column)
{
@@ -209,15 +187,14 @@ public class ColumnFilter
if (s.isEmpty())
return null;
- // isFetchAll only imply everything if fetches for regular
- return new Tester(isFetchAll && !column.isStatic(), s.iterator());
+ return new Tester(isFetchAll, s.iterator());
}
/**
* 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).
*/
- public static Builder allRegularColumnsBuilder(CFMetaData metadata)
+ public static Builder allColumnsBuilder(CFMetaData metadata)
{
return new Builder(metadata);
}
@@ -233,36 +210,24 @@ public class ColumnFilter
public static class Tester
{
- private final boolean isFetched; // if true, all cells are included
+ private final boolean isFetchAll;
private ColumnSubselection current;
private final Iterator<ColumnSubselection> iterator;
- private Tester(boolean isFetched, Iterator<ColumnSubselection> iterator)
+ private Tester(boolean isFetchAll, Iterator<ColumnSubselection> iterator)
{
- this.isFetched = isFetched;
+ this.isFetchAll = isFetchAll;
this.iterator = iterator;
}
public boolean includes(CellPath 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 isFetchAll || includedBySubselection(path);
}
- /**
- * Must only be called if {@code includes(path) == true}.
- */
public boolean canSkipValue(CellPath 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 isFetchAll && !includedBySubselection(path);
}
private boolean includedBySubselection(CellPath path)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e1da99a1/src/java/org/apache/cassandra/db/rows/BTreeRow.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/rows/BTreeRow.java b/src/java/org/apache/cassandra/db/rows/BTreeRow.java
index 18f3dec..ea1d9e0 100644
--- a/src/java/org/apache/cassandra/db/rows/BTreeRow.java
+++ b/src/java/org/apache/cassandra/db/rows/BTreeRow.java
@@ -237,7 +237,7 @@ public class BTreeRow extends AbstractRow
{
Map<ByteBuffer, CFMetaData.DroppedColumn> droppedColumns = metadata.getDroppedColumns();
- if (filter.includesAllColumns(isStatic()) && (activeDeletion.isLive() || deletion.supersedes(activeDeletion)) && droppedColumns.isEmpty())
+ if (filter.includesAllColumns() && (activeDeletion.isLive() || deletion.supersedes(activeDeletion)) && droppedColumns.isEmpty())
return this;
boolean mayHaveShadowed = activeDeletion.supersedes(deletion.time());
[4/6] cassandra git commit: Merge commit
'e1da99a1d82b5313f022307c2e110afc07e112b6' into cassandra-3.11
Posted by sl...@apache.org.
Merge commit 'e1da99a1d82b5313f022307c2e110afc07e112b6' into cassandra-3.11
* commit 'e1da99a1d82b5313f022307c2e110afc07e112b6':
Revert CASSANDRA-12768 (and update fix of CASSANDRA-12694) due to upgrade regression
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b234ca3e
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b234ca3e
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b234ca3e
Branch: refs/heads/trunk
Commit: b234ca3ee4101a34b761948d633ef4a12aa70ca2
Parents: 237e14d e1da99a
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Fri Jan 27 09:26:14 2017 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 27 09:26:53 2017 +0100
----------------------------------------------------------------------
.../cql3/statements/CQL3CasRequest.java | 27 +++---
.../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 +-
6 files changed, 51 insertions(+), 100 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b234ca3e/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b234ca3e/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index 1744e70,aca6146..038d4bd
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@@ -809,12 -743,15 +791,14 @@@ public class SelectStatement implement
ByteBuffer[] keyComponents = getComponents(cfm, partition.partitionKey());
Row staticRow = partition.staticRow();
- // If there is no rows, we include the static content if we should and we're done.
- // If there is no rows, then provided the select was a full partition selection
- // (i.e. not a 2ndary index search and there was no condition on clustering columns),
++ // 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),
+ // we want to include static columns and we're done.
if (!partition.hasNext())
{
- if (!staticRow.isEmpty() && returnStaticContentOnPartitionWithNoRows())
- if (!staticRow.isEmpty() && (!restrictions.usesSecondaryIndexing() || cfm.isStaticCompactTable())
- && !restrictions.hasClusteringColumnsRestriction())
++ if (!staticRow.isEmpty() && (!restrictions.hasClusteringColumnsRestriction() || cfm.isStaticCompactTable()))
{
- 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/b234ca3e/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/filter/ColumnFilter.java
index d320fc3,df91781..93a848e
--- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
@@@ -35,53 -36,39 +35,52 @@@ 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 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 #6588for 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 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).
++ * 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).
+ *
+ * Note that in practice:
+ * - the _queried_ columns set is always included in the _fetched_ 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.
++ * - 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.
+ * - 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_ 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;
- // Distinguish between the 2 cases described above: if 'isFetchAll' is true, then all columns will be retrieved
- // by the query, but 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.
++ // 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_.
+ private final boolean isFetchAll;
- private final CFMetaData metadata; // can be null if !fetchAllRegulars
+ private final CFMetaData metadata; // can be null if !isFetchAll
- 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 PartitionColumns selection; // can be null if isFetchAll and we don't want to skip any value
++ private final PartitionColumns queried; // can be null if isFetchAll and _fetched_ == _queried_
private final SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections; // can be null
- private ColumnFilter(boolean fetchAllRegulars,
+ private ColumnFilter(boolean isFetchAll,
CFMetaData metadata,
- PartitionColumns columns,
+ PartitionColumns queried,
SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections)
{
- assert !fetchAllRegulars || metadata != null;
- assert fetchAllRegulars || queried != null;
- this.fetchAllRegulars = fetchAllRegulars;
++ assert !isFetchAll || metadata != null;
++ assert isFetchAll || queried != null;
+ this.isFetchAll = isFetchAll;
this.metadata = metadata;
- this.selection = columns;
+ this.queried = queried;
this.subSelections = subSelections;
}
@@@ -121,99 -111,61 +120,73 @@@
*/
public PartitionColumns fetchedColumns()
{
- if (!fetchAllRegulars)
- return queried;
-
- // 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 queried == null || all.statics.isEmpty()
- ? all
- : new PartitionColumns(queried.statics, all.regulars);
- return isFetchAll ? metadata.partitionColumns() : selection;
++ return isFetchAll ? metadata.partitionColumns() : queried;
+ }
+
+ /**
+ * 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 || fetchAllRegulars;
++ assert queried != null || isFetchAll;
+ return queried == null ? metadata.partitionColumns() : queried;
}
- /**
- * 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 fetchesAllColumns(boolean isStatic)
- public boolean includesAllColumns()
++ public boolean fetchesAllColumns()
{
- return isStatic ? queried == null : fetchAllRegulars;
+ return isFetchAll;
}
/**
- * Whether the provided column is selected by this selection.
+ * Whether _fetched_ == _queried_ for this filter, and so if the {@code isQueried()} methods
+ * can return {@code false} for some column/cell.
*/
- public boolean includes(ColumnDefinition column)
+ public boolean allFetchedColumnsAreQueried()
{
- return !fetchAllRegulars || queried == null;
- return isFetchAll || selection.contains(column);
++ return !isFetchAll || (queried == null && subSelections == null);
}
/**
- * Whether we can skip the value for the provided selected column.
+ * Whether the provided column is fetched by this filter.
*/
- public boolean canSkipValue(ColumnDefinition column)
+ public boolean fetches(ColumnDefinition column)
{
- // 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 regulars, if 'fetchAllRegulars', then it's included automatically. Otherwise, it depends on _queried_.
- return fetchAllRegulars || queried.contains(column);
- // We don't use that currently, see #10655 for more details.
- return false;
++ return isFetchAll || queried.contains(column);
}
/**
- * Whether the provided cell of a complex column is selected by this selection.
+ * 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 includes(Cell cell)
+ public boolean fetchedColumnIsQueried(ColumnDefinition column)
{
- return !fetchAllRegulars || queried == null || 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 (ColumnSubselection subSel : s)
- if (subSel.compareInclusionOf(cell.path()) == 0)
- return true;
-
- return false;
++ return !isFetchAll || 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 (!fetchAllRegulars || subSelections == null)
++ if (!isFetchAll || 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)
@@@ -225,10 -177,6 +198,9 @@@
/**
* Creates a new {@code Tester} to efficiently test the inclusion of cells of complex column
* {@code column}.
+ *
- * @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)
{
@@@ -243,10 -191,10 +215,10 @@@
}
/**
- * Returns a {@code ColumnFilter}} builder that fetches all regular 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 columns (and queries the columns
+ * added to the builder, or everything if no column is added).
*/
- public static Builder allRegularColumnsBuilder(CFMetaData metadata)
+ public static Builder allColumnsBuilder(CFMetaData metadata)
{
return new Builder(metadata);
}
@@@ -271,20 -220,17 +243,20 @@@
this.iterator = iterator;
}
- public boolean includes(CellPath path)
+ public boolean fetches(CellPath path)
{
- return isFetched || hasSubselection(path);
- return isFetchAll || includedBySubselection(path);
++ return isFetchAll || hasSubselection(path);
}
- public boolean canSkipValue(CellPath path)
+ /**
+ * Must only be called if {@code fetches(path) == true}.
+ */
+ public boolean fetchedCellIsQueried(CellPath path)
{
- return !isFetched || hasSubselection(path);
- return isFetchAll && !includedBySubselection(path);
++ return !isFetchAll || hasSubselection(path);
}
- private boolean includedBySubselection(CellPath path)
+ private boolean hasSubselection(CellPath path)
{
while (current != null || iterator.hasNext())
{
@@@ -387,13 -321,13 +359,13 @@@
@Override
public String toString()
{
- if (fetchAllRegulars && queried == null)
+ if (isFetchAll)
return "*";
- if (selection.isEmpty())
+ if (queried.isEmpty())
return "";
- Iterator<ColumnDefinition> defs = selection.selectOrderIterator();
+ Iterator<ColumnDefinition> defs = queried.selectOrderIterator();
if (!defs.hasNext())
return "<none>";
@@@ -435,8 -366,8 +407,8 @@@
private static int makeHeaderByte(ColumnFilter selection)
{
- return (selection.fetchAllRegulars ? IS_FETCH_ALL_MASK : 0)
+ return (selection.isFetchAll ? IS_FETCH_ALL_MASK : 0)
- | (selection.selection != null ? HAS_SELECTION_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/b234ca3e/src/java/org/apache/cassandra/db/rows/BTreeRow.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/rows/BTreeRow.java
index 54da16b,ea1d9e0..0eccb6e
--- 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(isStatic());
- if (filter.includesAllColumns() && (activeDeletion.isLive() || deletion.supersedes(activeDeletion)) && droppedColumns.isEmpty())
++ boolean mayFilterColumns = !filter.fetchesAllColumns() || !filter.allFetchedColumnsAreQueried();
+ 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/b234ca3e/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/schema/SchemaKeyspace.java
index 1617aa7,84a5e13..2dffe58
--- 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.allRegularColumnsBuilder(partition.metadata());
++ ColumnFilter.Builder builder = ColumnFilter.allColumnsBuilder(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/b234ca3e/test/unit/org/apache/cassandra/db/ReadCommandTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/db/ReadCommandTest.java
index 9b7775da,0000000..2aef2a7
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.allRegularColumnsBuilder(cfs.metadata).build();
++ ColumnFilter columnFilter = ColumnFilter.allColumnsBuilder(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);
+ }
+ }
+}
[5/6] cassandra git commit: Merge commit
'e1da99a1d82b5313f022307c2e110afc07e112b6' into cassandra-3.11
Posted by sl...@apache.org.
Merge commit 'e1da99a1d82b5313f022307c2e110afc07e112b6' into cassandra-3.11
* commit 'e1da99a1d82b5313f022307c2e110afc07e112b6':
Revert CASSANDRA-12768 (and update fix of CASSANDRA-12694) due to upgrade regression
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b234ca3e
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b234ca3e
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b234ca3e
Branch: refs/heads/cassandra-3.11
Commit: b234ca3ee4101a34b761948d633ef4a12aa70ca2
Parents: 237e14d e1da99a
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Fri Jan 27 09:26:14 2017 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 27 09:26:53 2017 +0100
----------------------------------------------------------------------
.../cql3/statements/CQL3CasRequest.java | 27 +++---
.../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 +-
6 files changed, 51 insertions(+), 100 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b234ca3e/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b234ca3e/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index 1744e70,aca6146..038d4bd
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@@ -809,12 -743,15 +791,14 @@@ public class SelectStatement implement
ByteBuffer[] keyComponents = getComponents(cfm, partition.partitionKey());
Row staticRow = partition.staticRow();
- // If there is no rows, we include the static content if we should and we're done.
- // If there is no rows, then provided the select was a full partition selection
- // (i.e. not a 2ndary index search and there was no condition on clustering columns),
++ // 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),
+ // we want to include static columns and we're done.
if (!partition.hasNext())
{
- if (!staticRow.isEmpty() && returnStaticContentOnPartitionWithNoRows())
- if (!staticRow.isEmpty() && (!restrictions.usesSecondaryIndexing() || cfm.isStaticCompactTable())
- && !restrictions.hasClusteringColumnsRestriction())
++ if (!staticRow.isEmpty() && (!restrictions.hasClusteringColumnsRestriction() || cfm.isStaticCompactTable()))
{
- 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/b234ca3e/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/filter/ColumnFilter.java
index d320fc3,df91781..93a848e
--- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
@@@ -35,53 -36,39 +35,52 @@@ 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 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 #6588for 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 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).
++ * 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).
+ *
+ * Note that in practice:
+ * - the _queried_ columns set is always included in the _fetched_ 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.
++ * - 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.
+ * - 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_ 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;
- // Distinguish between the 2 cases described above: if 'isFetchAll' is true, then all columns will be retrieved
- // by the query, but 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.
++ // 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_.
+ private final boolean isFetchAll;
- private final CFMetaData metadata; // can be null if !fetchAllRegulars
+ private final CFMetaData metadata; // can be null if !isFetchAll
- 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 PartitionColumns selection; // can be null if isFetchAll and we don't want to skip any value
++ private final PartitionColumns queried; // can be null if isFetchAll and _fetched_ == _queried_
private final SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections; // can be null
- private ColumnFilter(boolean fetchAllRegulars,
+ private ColumnFilter(boolean isFetchAll,
CFMetaData metadata,
- PartitionColumns columns,
+ PartitionColumns queried,
SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections)
{
- assert !fetchAllRegulars || metadata != null;
- assert fetchAllRegulars || queried != null;
- this.fetchAllRegulars = fetchAllRegulars;
++ assert !isFetchAll || metadata != null;
++ assert isFetchAll || queried != null;
+ this.isFetchAll = isFetchAll;
this.metadata = metadata;
- this.selection = columns;
+ this.queried = queried;
this.subSelections = subSelections;
}
@@@ -121,99 -111,61 +120,73 @@@
*/
public PartitionColumns fetchedColumns()
{
- if (!fetchAllRegulars)
- return queried;
-
- // 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 queried == null || all.statics.isEmpty()
- ? all
- : new PartitionColumns(queried.statics, all.regulars);
- return isFetchAll ? metadata.partitionColumns() : selection;
++ return isFetchAll ? metadata.partitionColumns() : queried;
+ }
+
+ /**
+ * 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 || fetchAllRegulars;
++ assert queried != null || isFetchAll;
+ return queried == null ? metadata.partitionColumns() : queried;
}
- /**
- * 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 fetchesAllColumns(boolean isStatic)
- public boolean includesAllColumns()
++ public boolean fetchesAllColumns()
{
- return isStatic ? queried == null : fetchAllRegulars;
+ return isFetchAll;
}
/**
- * Whether the provided column is selected by this selection.
+ * Whether _fetched_ == _queried_ for this filter, and so if the {@code isQueried()} methods
+ * can return {@code false} for some column/cell.
*/
- public boolean includes(ColumnDefinition column)
+ public boolean allFetchedColumnsAreQueried()
{
- return !fetchAllRegulars || queried == null;
- return isFetchAll || selection.contains(column);
++ return !isFetchAll || (queried == null && subSelections == null);
}
/**
- * Whether we can skip the value for the provided selected column.
+ * Whether the provided column is fetched by this filter.
*/
- public boolean canSkipValue(ColumnDefinition column)
+ public boolean fetches(ColumnDefinition column)
{
- // 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 regulars, if 'fetchAllRegulars', then it's included automatically. Otherwise, it depends on _queried_.
- return fetchAllRegulars || queried.contains(column);
- // We don't use that currently, see #10655 for more details.
- return false;
++ return isFetchAll || queried.contains(column);
}
/**
- * Whether the provided cell of a complex column is selected by this selection.
+ * 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 includes(Cell cell)
+ public boolean fetchedColumnIsQueried(ColumnDefinition column)
{
- return !fetchAllRegulars || queried == null || 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 (ColumnSubselection subSel : s)
- if (subSel.compareInclusionOf(cell.path()) == 0)
- return true;
-
- return false;
++ return !isFetchAll || 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 (!fetchAllRegulars || subSelections == null)
++ if (!isFetchAll || 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)
@@@ -225,10 -177,6 +198,9 @@@
/**
* Creates a new {@code Tester} to efficiently test the inclusion of cells of complex column
* {@code column}.
+ *
- * @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)
{
@@@ -243,10 -191,10 +215,10 @@@
}
/**
- * Returns a {@code ColumnFilter}} builder that fetches all regular 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 columns (and queries the columns
+ * added to the builder, or everything if no column is added).
*/
- public static Builder allRegularColumnsBuilder(CFMetaData metadata)
+ public static Builder allColumnsBuilder(CFMetaData metadata)
{
return new Builder(metadata);
}
@@@ -271,20 -220,17 +243,20 @@@
this.iterator = iterator;
}
- public boolean includes(CellPath path)
+ public boolean fetches(CellPath path)
{
- return isFetched || hasSubselection(path);
- return isFetchAll || includedBySubselection(path);
++ return isFetchAll || hasSubselection(path);
}
- public boolean canSkipValue(CellPath path)
+ /**
+ * Must only be called if {@code fetches(path) == true}.
+ */
+ public boolean fetchedCellIsQueried(CellPath path)
{
- return !isFetched || hasSubselection(path);
- return isFetchAll && !includedBySubselection(path);
++ return !isFetchAll || hasSubselection(path);
}
- private boolean includedBySubselection(CellPath path)
+ private boolean hasSubselection(CellPath path)
{
while (current != null || iterator.hasNext())
{
@@@ -387,13 -321,13 +359,13 @@@
@Override
public String toString()
{
- if (fetchAllRegulars && queried == null)
+ if (isFetchAll)
return "*";
- if (selection.isEmpty())
+ if (queried.isEmpty())
return "";
- Iterator<ColumnDefinition> defs = selection.selectOrderIterator();
+ Iterator<ColumnDefinition> defs = queried.selectOrderIterator();
if (!defs.hasNext())
return "<none>";
@@@ -435,8 -366,8 +407,8 @@@
private static int makeHeaderByte(ColumnFilter selection)
{
- return (selection.fetchAllRegulars ? IS_FETCH_ALL_MASK : 0)
+ return (selection.isFetchAll ? IS_FETCH_ALL_MASK : 0)
- | (selection.selection != null ? HAS_SELECTION_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/b234ca3e/src/java/org/apache/cassandra/db/rows/BTreeRow.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/rows/BTreeRow.java
index 54da16b,ea1d9e0..0eccb6e
--- 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(isStatic());
- if (filter.includesAllColumns() && (activeDeletion.isLive() || deletion.supersedes(activeDeletion)) && droppedColumns.isEmpty())
++ boolean mayFilterColumns = !filter.fetchesAllColumns() || !filter.allFetchedColumnsAreQueried();
+ 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/b234ca3e/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/schema/SchemaKeyspace.java
index 1617aa7,84a5e13..2dffe58
--- 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.allRegularColumnsBuilder(partition.metadata());
++ ColumnFilter.Builder builder = ColumnFilter.allColumnsBuilder(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/b234ca3e/test/unit/org/apache/cassandra/db/ReadCommandTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/db/ReadCommandTest.java
index 9b7775da,0000000..2aef2a7
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.allRegularColumnsBuilder(cfs.metadata).build();
++ ColumnFilter columnFilter = ColumnFilter.allColumnsBuilder(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);
+ }
+ }
+}