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 2014/02/20 17:51:32 UTC
[5/6] Merge branch 'cassandra-2.0' into cassandra-2.1
http://git-wip-us.apache.org/repos/asf/cassandra/blob/63b1ef4e/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
index 7112b79,cd5f2a2..ece6916
--- a/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
@@@ -33,9 -31,9 +33,9 @@@ import org.apache.cassandra.utils.Pair
*/
public class DeleteStatement extends ModificationStatement
{
- private DeleteStatement(int boundTerms, CFMetaData cfm, Attributes attrs)
- private DeleteStatement(StatementType type, CFMetaData cfm, Attributes attrs)
++ private DeleteStatement(StatementType type, int boundTerms, CFMetaData cfm, Attributes attrs)
{
- super(boundTerms, cfm, attrs);
- super(type, cfm, attrs);
++ super(type, boundTerms, cfm, attrs);
}
public boolean requireFullClusteringKey()
@@@ -43,42 -41,63 +43,45 @@@
return false;
}
- public ColumnFamily updateForKey(ByteBuffer key, Composite prefix, UpdateParameters params)
- public ColumnFamily updateForKey(ByteBuffer key, ColumnNameBuilder builder, UpdateParameters params)
++ public void addUpdateForKey(ColumnFamily cf, ByteBuffer key, Composite prefix, UpdateParameters params)
throws InvalidRequestException
{
- ColumnFamily cf = ArrayBackedSortedColumns.factory.create(cfm);
- ColumnFamily cf = TreeMapBackedSortedColumns.factory.create(cfm);
- addUpdateForKey(cf, key, builder, params);
- return cf;
- }
-
- public void addUpdateForKey(ColumnFamily cf, ByteBuffer key, ColumnNameBuilder builder, UpdateParameters params)
- throws InvalidRequestException
- {
- CFDefinition cfDef = cfm.getCfDef();
List<Operation> deletions = getOperations();
- boolean fullKey = builder.componentCount() == cfDef.clusteringColumnsCount();
- boolean isRange = cfDef.isCompact ? !fullKey : (!fullKey || deletions.isEmpty());
-
- if (!deletions.isEmpty() && isRange)
+ if (prefix.size() < cfm.clusteringColumns().size() && !deletions.isEmpty())
- throw new InvalidRequestException(String.format("Missing mandatory PRIMARY KEY part %s since %s specified", getFirstEmptyKey(), deletions.iterator().next().column.name));
+ {
- // We only get there if we have at least one non-static columns selected, as otherwise the builder will be
- // the "static" builder and isRange will be false. But we may still have static columns, so pick the first
- // non static one for the error message so it's not confusing
++ // In general, we can't delete specific columns if not all clustering columns have been specified.
++ // However, if we delete only static colums, it's fine since we won't really use the prefix anyway.
+ for (Operation deletion : deletions)
- if (cfm.getCfDef().get(deletion.columnName).kind != CFDefinition.Name.Kind.STATIC)
- throw new InvalidRequestException(String.format("Missing mandatory PRIMARY KEY part %s since %s specified", getFirstEmptyKey(), deletion.columnName));
- throw new AssertionError();
++ if (!deletion.column.isStatic())
++ throw new InvalidRequestException(String.format("Missing mandatory PRIMARY KEY part %s since %s specified", getFirstEmptyKey(), deletion.column.name));
+ }
- if (deletions.isEmpty() && builder.componentCount() == 0)
+ if (deletions.isEmpty())
{
- // No columns specified, delete the row
- cf.delete(new DeletionInfo(params.timestamp, params.localDeletionTime));
- }
- else
- {
- if (isRange)
+ // We delete the slice selected by the prefix.
+ // However, for performance reasons, we distinguish 2 cases:
+ // - It's a full internal row delete
+ // - It's a full cell name (i.e it's a dense layout and the prefix is full)
+ if (prefix.isEmpty())
+ {
+ // No columns specified, delete the row
+ cf.delete(new DeletionInfo(params.timestamp, params.localDeletionTime));
+ }
+ else if (cfm.comparator.isDense() && prefix.size() == cfm.clusteringColumns().size())
{
- assert deletions.isEmpty();
- ByteBuffer start = builder.build();
- ByteBuffer end = builder.buildAsEndOfRange();
- cf.addAtom(params.makeRangeTombstone(start, end));
+ cf.addAtom(params.makeTombstone(cfm.comparator.create(prefix, null)));
}
else
{
- // Delete specific columns
- if (cfDef.isCompact)
- {
- ByteBuffer columnName = builder.build();
- cf.addColumn(params.makeTombstone(columnName));
- }
- else
- {
- for (Operation deletion : deletions)
- deletion.execute(key, cf, builder.copy(), params);
- }
+ cf.addAtom(params.makeRangeTombstone(prefix.slice()));
}
}
+ else
+ {
+ for (Operation op : deletions)
+ op.execute(key, cf, prefix, params);
+ }
-
- return cf;
}
public static class Parsed extends ModificationStatement.Parsed
@@@ -97,9 -116,9 +100,9 @@@
this.whereClause = whereClause;
}
- protected ModificationStatement prepareInternal(CFDefinition cfDef, VariableSpecifications boundNames, Attributes attrs) throws InvalidRequestException
+ protected ModificationStatement prepareInternal(CFMetaData cfm, VariableSpecifications boundNames, Attributes attrs) throws InvalidRequestException
{
- DeleteStatement stmt = new DeleteStatement(boundNames.size(), cfm, attrs);
- DeleteStatement stmt = new DeleteStatement(ModificationStatement.StatementType.DELETE, cfDef.cfm, attrs);
++ DeleteStatement stmt = new DeleteStatement(ModificationStatement.StatementType.DELETE, boundNames.size(), cfm, attrs);
for (Operation.RawDeletion deletion : deletions)
{
@@@ -109,10 -128,10 +112,10 @@@
// For compact, we only have one value except the key, so the only form of DELETE that make sense is without a column
// list. However, we support having the value name for coherence with the static/sparse case
- if (def.kind != ColumnDefinition.Kind.REGULAR && def.kind != ColumnDefinition.Kind.COMPACT_VALUE)
- if (name.isPrimaryKeyColumn())
- throw new InvalidRequestException(String.format("Invalid identifier %s for deletion (should not be a PRIMARY KEY part)", name));
++ if (def.isPrimaryKeyColumn())
+ throw new InvalidRequestException(String.format("Invalid identifier %s for deletion (should not be a PRIMARY KEY part)", def.name));
- Operation op = deletion.prepare(name);
+ Operation op = deletion.prepare(cfm.ksName, def);
op.collectMarkerSpecification(boundNames);
stmt.addOperation(op);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/63b1ef4e/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
index 6d18f1b,ac8d2e1..db0b7a9
--- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
@@@ -20,7 -20,11 +20,9 @@@ package org.apache.cassandra.cql3.state
import java.nio.ByteBuffer;
import java.util.*;
+ import com.google.common.base.Function;
+ import com.google.common.collect.Iterables;
import org.github.jamm.MemoryMeter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.cassandra.auth.Permission;
import org.apache.cassandra.config.CFMetaData;
@@@ -50,19 -54,40 +52,36 @@@ public abstract class ModificationState
{
private static final ColumnIdentifier CAS_RESULT_COLUMN = new ColumnIdentifier("[applied]", false);
- private static final Logger logger = LoggerFactory.getLogger(ModificationStatement.class);
-
- private static boolean loggedCounterTTL = false;
- private static boolean loggedCounterTimestamp = false;
-
+ public static enum StatementType { INSERT, UPDATE, DELETE }
+ public final StatementType type;
+
+ private final int boundTerms;
public final CFMetaData cfm;
public final Attributes attrs;
private final Map<ColumnIdentifier, Restriction> processedKeys = new HashMap<ColumnIdentifier, Restriction>();
private final List<Operation> columnOperations = new ArrayList<Operation>();
- private List<Operation> columnConditions;
- private int boundTerms;
+ // Separating normal and static conditions makes things somewhat easier
+ private List<ColumnCondition> columnConditions;
+ private List<ColumnCondition> staticConditions;
private boolean ifNotExists;
- public ModificationStatement(int boundTerms, CFMetaData cfm, Attributes attrs)
+ private boolean hasNoClusteringColumns = true;
+ private boolean setsOnlyStaticColumns;
+
- private final Function<ColumnCondition, ColumnIdentifier> getColumnForCondition = new Function<ColumnCondition, ColumnIdentifier>()
++ private final Function<ColumnCondition, ColumnDefinition> getColumnForCondition = new Function<ColumnCondition, ColumnDefinition>()
+ {
- public ColumnIdentifier apply(ColumnCondition cond)
++ public ColumnDefinition apply(ColumnCondition cond)
+ {
- return cond.column.name;
++ return cond.column;
+ }
+ };
+
- public ModificationStatement(StatementType type, CFMetaData cfm, Attributes attrs)
++ public ModificationStatement(StatementType type, int boundTerms, CFMetaData cfm, Attributes attrs)
{
+ this.type = type;
+ this.boundTerms = boundTerms;
this.cfm = cfm;
this.attrs = attrs;
}
@@@ -77,7 -103,7 +97,7 @@@
}
public abstract boolean requireFullClusteringKey();
- public abstract ColumnFamily updateForKey(ByteBuffer key, Composite prefix, UpdateParameters params) throws InvalidRequestException;
- public abstract void addUpdateForKey(ColumnFamily updates, ByteBuffer key, ColumnNameBuilder builder, UpdateParameters params) throws InvalidRequestException;
++ public abstract void addUpdateForKey(ColumnFamily updates, ByteBuffer key, Composite prefix, UpdateParameters params) throws InvalidRequestException;
public int getBoundTerms()
{
@@@ -137,6 -176,15 +157,15 @@@
public void addOperation(Operation op)
{
- if (op.isStatic(cfm))
++ if (op.column.isStatic())
+ {
+ if (columnOperations.isEmpty())
+ setsOnlyStaticColumns = true;
+ }
+ else
+ {
+ setsOnlyStaticColumns = false;
+ }
columnOperations.add(op);
}
@@@ -145,12 -193,31 +174,31 @@@
return columnOperations;
}
- public void addCondition(Operation op)
- public Iterable<ColumnIdentifier> getColumnsWithConditions()
++ public Iterable<ColumnDefinition> getColumnsWithConditions()
{
- if (columnConditions == null)
- columnConditions = new ArrayList<Operation>();
+ if (ifNotExists)
+ return null;
- columnConditions.add(op);
- return Iterables.concat(columnConditions == null ? Collections.<ColumnIdentifier>emptyList() : Iterables.transform(columnConditions, getColumnForCondition),
- staticConditions == null ? Collections.<ColumnIdentifier>emptyList() : Iterables.transform(staticConditions, getColumnForCondition));
++ return Iterables.concat(columnConditions == null ? Collections.<ColumnDefinition>emptyList() : Iterables.transform(columnConditions, getColumnForCondition),
++ staticConditions == null ? Collections.<ColumnDefinition>emptyList() : Iterables.transform(staticConditions, getColumnForCondition));
+ }
+
+ public void addCondition(ColumnCondition cond) throws InvalidRequestException
+ {
+ List<ColumnCondition> conds = null;
- if (cond.column.kind == CFDefinition.Name.Kind.STATIC)
++ if (cond.column.isStatic())
+ {
+ if (staticConditions == null)
+ staticConditions = new ArrayList<ColumnCondition>();
+ conds = staticConditions;
+ }
+ else
+ {
+ if (columnConditions == null)
+ columnConditions = new ArrayList<ColumnCondition>();
+ conds = columnConditions;
+ }
+ conds.add(cond);
}
public void setIfNotExistCondition()
@@@ -158,15 -225,22 +206,22 @@@
ifNotExists = true;
}
- private void addKeyValues(ColumnIdentifier name, Restriction values) throws InvalidRequestException
+ public boolean hasIfNotExistCondition()
+ {
+ return ifNotExists;
+ }
+
- private void addKeyValues(CFDefinition.Name name, Restriction values) throws InvalidRequestException
++ private void addKeyValues(ColumnDefinition def, Restriction values) throws InvalidRequestException
{
- if (processedKeys.put(name, values) != null)
- throw new InvalidRequestException(String.format("Multiple definitions found for PRIMARY KEY part %s", name));
- if (name.kind == CFDefinition.Name.Kind.COLUMN_ALIAS)
++ if (def.kind == ColumnDefinition.Kind.CLUSTERING_COLUMN)
+ hasNoClusteringColumns = false;
- if (processedKeys.put(name.name, values) != null)
- throw new InvalidRequestException(String.format("Multiple definitions found for PRIMARY KEY part %s", name.name));
++ if (processedKeys.put(def.name, values) != null)
++ throw new InvalidRequestException(String.format("Multiple definitions found for PRIMARY KEY part %s", def.name));
}
- public void addKeyValue(ColumnIdentifier name, Term value) throws InvalidRequestException
- public void addKeyValue(CFDefinition.Name name, Term value) throws InvalidRequestException
++ public void addKeyValue(ColumnDefinition def, Term value) throws InvalidRequestException
{
-- addKeyValues(name, new Restriction.EQ(value, false));
++ addKeyValues(def, new Restriction.EQ(value, false));
}
public void processWhereClause(List<Relation> whereClause, VariableSpecifications names) throws InvalidRequestException
@@@ -211,14 -286,15 +266,13 @@@
}
else
{
- throw new InvalidRequestException(String.format("Invalid operator %s for PRIMARY KEY part %s", rel.operator(), name));
+ throw new InvalidRequestException(String.format("Invalid operator %s for PRIMARY KEY part %s", rel.operator(), def.name));
}
- addKeyValues(def.name, restriction);
- addKeyValues(name, restriction);
++ addKeyValues(def, restriction);
break;
- case COMPACT_VALUE:
- case REGULAR:
- case VALUE_ALIAS:
- case COLUMN_METADATA:
- case STATIC:
- throw new InvalidRequestException(String.format("Non PRIMARY KEY %s found in where clause", name));
++ default:
+ throw new InvalidRequestException(String.format("Non PRIMARY KEY %s found in where clause", def.name));
}
}
}
@@@ -241,8 -318,10 +295,10 @@@
for (ByteBuffer val : values)
{
if (val == null)
- throw new InvalidRequestException(String.format("Invalid null value for partition key part %s", name));
- ByteBuffer key = keyBuilder.copy().add(val).build();
+ throw new InvalidRequestException(String.format("Invalid null value for partition key part %s", def.name));
- keys.add(keyBuilder.buildWith(val).toByteBuffer());
++ ByteBuffer key = keyBuilder.buildWith(val).toByteBuffer();
+ ThriftValidation.validateKey(cfm, key);
+ keys.add(key);
}
}
else
@@@ -258,19 -337,56 +314,44 @@@
return keys;
}
- public ColumnNameBuilder createClusteringPrefixBuilder(List<ByteBuffer> variables)
+ public Composite createClusteringPrefix(List<ByteBuffer> variables)
throws InvalidRequestException
{
+ // If the only updated/deleted columns are static, then we don't need clustering columns.
+ // And in fact, unless it is an INSERT, we reject if clustering colums are provided as that
+ // suggest something unintended. For instance, given:
+ // CREATE TABLE t (k int, v int, s int static, PRIMARY KEY (k, v))
+ // it can make sense to do:
+ // INSERT INTO t(k, v, s) VALUES (0, 1, 2)
+ // but both
+ // UPDATE t SET s = 3 WHERE k = 0 AND v = 1
+ // DELETE v FROM t WHERE k = 0 AND v = 1
+ // sounds like you don't really understand what your are doing.
+ if (setsOnlyStaticColumns && columnConditions == null && (type != StatementType.INSERT || hasNoClusteringColumns))
+ {
+ // Reject if any clustering columns is set
- for (CFDefinition.Name name : cfm.getCfDef().clusteringColumns())
- if (processedKeys.get(name.name) != null)
- throw new InvalidRequestException(String.format("Invalid restriction on clustering column %s since the %s statement modifies only static columns", name.name, type));
- return cfm.getStaticColumnNameBuilder();
++ for (ColumnDefinition def : cfm.clusteringColumns())
++ if (processedKeys.get(def.name) != null)
++ throw new InvalidRequestException(String.format("Invalid restriction on clustering column %s since the %s statement modifies only static columns", def.name, type));
++ return cfm.comparator.staticPrefix();
+ }
+
+ return createClusteringPrefixBuilderInternal(variables);
+ }
+
- private ColumnNameBuilder updatePrefixFor(ByteBuffer name, ColumnNameBuilder prefix)
- {
- return isStatic(name) ? cfm.getStaticColumnNameBuilder() : prefix;
- }
-
- public boolean isStatic(ByteBuffer name)
- {
- ColumnDefinition def = cfm.getColumnDefinition(name);
- return def != null && def.type == ColumnDefinition.Type.STATIC;
- }
-
- private ColumnNameBuilder createClusteringPrefixBuilderInternal(List<ByteBuffer> variables)
++ private Composite createClusteringPrefixBuilderInternal(List<ByteBuffer> variables)
+ throws InvalidRequestException
+ {
- CFDefinition cfDef = cfm.getCfDef();
- ColumnNameBuilder builder = cfDef.getColumnNameBuilder();
- CFDefinition.Name firstEmptyKey = null;
- for (CFDefinition.Name name : cfDef.clusteringColumns())
+ CBuilder builder = cfm.comparator.prefixBuilder();
+ ColumnDefinition firstEmptyKey = null;
+ for (ColumnDefinition def : cfm.clusteringColumns())
{
- Restriction r = processedKeys.get(name.name);
+ Restriction r = processedKeys.get(def.name);
if (r == null)
{
- firstEmptyKey = name;
- if (requireFullClusteringKey() && cfDef.isComposite && !cfDef.isCompact)
- throw new InvalidRequestException(String.format("Missing mandatory PRIMARY KEY part %s", name));
+ firstEmptyKey = def;
+ if (requireFullClusteringKey() && !cfm.comparator.isDense() && cfm.comparator.isCompound())
+ throw new InvalidRequestException(String.format("Missing mandatory PRIMARY KEY part %s", def.name));
}
else if (firstEmptyKey != null)
{
@@@ -303,21 -419,21 +384,20 @@@
throws RequestExecutionException, RequestValidationException
{
// Lists SET operation incurs a read.
- Set<ColumnIdentifier> toRead = null;
- Set<ByteBuffer> toRead = null;
++ boolean requiresRead = false;
for (Operation op : columnOperations)
{
if (op.requiresRead())
{
-- if (toRead == null)
- toRead = new TreeSet<ColumnIdentifier>();
- toRead.add(op.column.name);
- toRead = new TreeSet<ByteBuffer>(UTF8Type.instance);
- toRead.add(op.columnName.key);
++ requiresRead = true;
++ break;
}
}
- return toRead == null ? null : readRows(partitionKeys, clusteringPrefix, toRead, cfm, local, cl);
- return toRead == null ? null : readRows(partitionKeys, clusteringPrefix, toRead, (CompositeType)cfm.comparator, local, cl);
++ return requiresRead ? readRows(partitionKeys, clusteringPrefix, cfm, local, cl) : null;
}
- protected Map<ByteBuffer, CQL3Row> readRows(List<ByteBuffer> partitionKeys, Composite rowPrefix, Set<ColumnIdentifier> toRead, CFMetaData cfm, boolean local, ConsistencyLevel cl)
- private Map<ByteBuffer, ColumnGroupMap> readRows(List<ByteBuffer> partitionKeys, ColumnNameBuilder clusteringPrefix, Set<ByteBuffer> toRead, CompositeType composite, boolean local, ConsistencyLevel cl)
++ protected Map<ByteBuffer, CQL3Row> readRows(List<ByteBuffer> partitionKeys, Composite rowPrefix, CFMetaData cfm, boolean local, ConsistencyLevel cl)
throws RequestExecutionException, RequestValidationException
{
try
@@@ -329,11 -445,16 +409,7 @@@
throw new InvalidRequestException(String.format("Write operation require a read but consistency %s is not supported on reads", cl));
}
-- ColumnSlice[] slices = new ColumnSlice[toRead.size()];
-- int i = 0;
- for (ColumnIdentifier name : toRead)
- slices[i++] = cfm.comparator.create(rowPrefix, name).slice();
-
- for (ByteBuffer name : toRead)
- {
- ColumnNameBuilder prefix = updatePrefixFor(name, clusteringPrefix);
- ByteBuffer start = prefix.copy().add(name).build();
- ByteBuffer finish = prefix.copy().add(name).buildAsEndOfRange();
- slices[i++] = new ColumnSlice(start, finish);
- }
-
++ ColumnSlice[] slices = new ColumnSlice[]{ rowPrefix.slice() };
List<ReadCommand> commands = new ArrayList<ReadCommand>(partitionKeys.size());
long now = System.currentTimeMillis();
for (ByteBuffer key : partitionKeys)
@@@ -414,14 -536,10 +490,10 @@@
// It's cleaner to use the query timestamp below, but it's in seconds while the conditions expects microseconds, so just
// put it back in millis (we don't really lose precision because the ultimate consumer, Column.isLive, re-divide it).
- long now = queryState.getTimestamp() * 1000;
- CASConditions conditions = ifNotExists
- ? new NotExistCondition(clusteringPrefix, now)
- : new ColumnsConditions(clusteringPrefix, cfm, key, columnConditions, variables, now);
+ CQL3CasConditions conditions = new CQL3CasConditions(cfm, queryState.getTimestamp() * 1000);
- ColumnNameBuilder prefix = createClusteringPrefixBuilder(variables);
- ColumnFamily updates = UnsortedColumns.factory.create(cfm);
++ Composite prefix = createClusteringPrefix(variables);
++ ColumnFamily updates = ArrayBackedSortedColumns.factory.create(cfm);
+ addUpdatesAndConditions(key, prefix, updates, conditions, variables, getTimestamp(queryState.getTimestamp(), variables));
ColumnFamily result = StorageProxy.cas(keyspace(),
columnFamily(),
@@@ -433,11 -551,39 +505,39 @@@
return new ResultMessage.Rows(buildCasResultSet(key, result));
}
- public void addUpdatesAndConditions(ByteBuffer key, ColumnNameBuilder clusteringPrefix, ColumnFamily updates, CQL3CasConditions conditions, List<ByteBuffer> variables, long now)
++ public void addUpdatesAndConditions(ByteBuffer key, Composite clusteringPrefix, ColumnFamily updates, CQL3CasConditions conditions, List<ByteBuffer> variables, long now)
+ throws InvalidRequestException
+ {
+ UpdateParameters updParams = new UpdateParameters(cfm, variables, now, getTimeToLive(variables), null);
+ addUpdateForKey(updates, key, clusteringPrefix, updParams);
+
+ if (ifNotExists)
+ {
+ // If we use ifNotExists, if the statement applies to any non static columns, then the condition is on the row of the non-static
- // columns and the prefix should be the rowPrefix. But if only static columns are set, then the ifNotExists apply to the existence
++ // columns and the prefix should be the clusteringPrefix. But if only static columns are set, then the ifNotExists apply to the existence
+ // of any static columns and we should use the prefix for the "static part" of the partition.
- conditions.addNotExist(setsOnlyStaticColumns ? cfm.getStaticColumnNameBuilder() : clusteringPrefix);
++ conditions.addNotExist(setsOnlyStaticColumns ? cfm.comparator.staticPrefix() : clusteringPrefix);
+ }
+ else
+ {
+ if (columnConditions != null)
+ conditions.addConditions(clusteringPrefix, columnConditions, variables);
+ if (staticConditions != null)
- conditions.addConditions(cfm.getStaticColumnNameBuilder(), staticConditions, variables);
++ conditions.addConditions(cfm.comparator.staticPrefix(), staticConditions, variables);
+ }
+ }
+
private ResultSet buildCasResultSet(ByteBuffer key, ColumnFamily cf) throws InvalidRequestException
{
+ return buildCasResultSet(keyspace(), key, columnFamily(), cf, getColumnsWithConditions(), false);
+ }
+
- public static ResultSet buildCasResultSet(String ksName, ByteBuffer key, String cfName, ColumnFamily cf, Iterable<ColumnIdentifier> columnsWithConditions, boolean isBatch)
++ public static ResultSet buildCasResultSet(String ksName, ByteBuffer key, String cfName, ColumnFamily cf, Iterable<ColumnDefinition> columnsWithConditions, boolean isBatch)
+ throws InvalidRequestException
+ {
boolean success = cf == null;
- ColumnSpecification spec = new ColumnSpecification(keyspace(), columnFamily(), CAS_RESULT_COLUMN, BooleanType.instance);
+ ColumnSpecification spec = new ColumnSpecification(ksName, cfName, CAS_RESULT_COLUMN, BooleanType.instance);
ResultSet.Metadata metadata = new ResultSet.Metadata(Collections.singletonList(spec));
List<List<ByteBuffer>> rows = Collections.singletonList(Collections.singletonList(BooleanType.instance.decompose(success)));
@@@ -457,25 -603,40 +557,39 @@@
List<ColumnSpecification> specs = new ArrayList<ColumnSpecification>(size);
specs.addAll(left.metadata.names);
specs.addAll(right.metadata.names);
- List<ByteBuffer> row = new ArrayList<ByteBuffer>(size);
- row.addAll(left.rows.get(0));
- row.addAll(right.rows.get(0));
- return new ResultSet(new ResultSet.Metadata(specs), Collections.singletonList(row));
+ List<List<ByteBuffer>> rows = new ArrayList<>(right.size());
+ for (int i = 0; i < right.size(); i++)
+ {
+ List<ByteBuffer> row = new ArrayList<ByteBuffer>(size);
+ row.addAll(left.rows.get(0));
+ row.addAll(right.rows.get(i));
+ rows.add(row);
+ }
+ return new ResultSet(new ResultSet.Metadata(specs), rows);
}
- private ResultSet buildCasFailureResultSet(ByteBuffer key, ColumnFamily cf) throws InvalidRequestException
- private static ResultSet buildCasFailureResultSet(ByteBuffer key, ColumnFamily cf, Iterable<ColumnIdentifier> columnsWithConditions, boolean isBatch)
++ private static ResultSet buildCasFailureResultSet(ByteBuffer key, ColumnFamily cf, Iterable<ColumnDefinition> columnsWithConditions, boolean isBatch)
+ throws InvalidRequestException
{
- CFDefinition cfDef = cf.metadata().getCfDef();
-
++ CFMetaData cfm = cf.metadata();
Selection selection;
- if (ifNotExists)
+ if (columnsWithConditions == null)
{
- selection = Selection.wildcard(cfDef);
+ selection = Selection.wildcard(cfm);
}
else
{
- List<ColumnDefinition> defs = new ArrayList<>(columnConditions.size());
- for (Operation condition : columnConditions)
- defs.add(condition.column);
- List<CFDefinition.Name> names = new ArrayList<CFDefinition.Name>();
++ List<ColumnDefinition> defs = new ArrayList<>();
+ // Adding the partition key for batches to disambiguate if the conditions span multipe rows (we don't add them outside
+ // of batches for compatibility sakes).
+ if (isBatch)
+ {
- names.addAll(cfDef.partitionKeys());
- names.addAll(cfDef.clusteringColumns());
++ defs.addAll(cfm.partitionKeyColumns());
++ defs.addAll(cfm.clusteringColumns());
+ }
- for (ColumnIdentifier id : columnsWithConditions)
- names.add(cfDef.get(id));
- selection = Selection.forColumns(names);
++ for (ColumnDefinition def : columnsWithConditions)
++ defs.add(def);
+ selection = Selection.forColumns(defs);
}
long now = System.currentTimeMillis();
@@@ -524,7 -681,8 +638,8 @@@
for (ByteBuffer key: keys)
{
ThriftValidation.validateKey(cfm, key);
- ColumnFamily cf = updateForKey(key, clusteringPrefix, params);
- ColumnFamily cf = UnsortedColumns.factory.create(cfm);
++ ColumnFamily cf = ArrayBackedSortedColumns.factory.create(cfm);
+ addUpdateForKey(cf, key, clusteringPrefix, params);
mutations.add(makeMutation(key, cf, cl, isBatch));
}
return mutations;
@@@ -541,98 -699,11 +656,11 @@@
}
else
{
- rm = new RowMutation(cfm.ksName, key, cf);
+ mutation = new Mutation(cfm.ksName, key, cf);
}
- return isCounter() ? new CounterMutation(rm, cl) : rm;
+ return isCounter() ? new CounterMutation(mutation, cl) : mutation;
}
- private static abstract class CQL3CasConditions implements CASConditions
- {
- protected final Composite rowPrefix;
- protected final long now;
-
- protected CQL3CasConditions(Composite rowPrefix, long now)
- {
- this.rowPrefix = rowPrefix;
- this.now = now;
- }
-
- public IDiskAtomFilter readFilter()
- {
- // We always read the row entirely as on CAS failure we want to be able to distinguish between "row exists
- // but all values on why there were conditions are null" and "row doesn't exists", and we can't rely on the
- // row marker for that (see #6623)
- return new SliceQueryFilter(rowPrefix.slice(), false, 1, rowPrefix.size());
- }
- }
-
- private static class NotExistCondition extends CQL3CasConditions
- {
- private NotExistCondition(Composite rowPrefix, long now)
- {
- super(rowPrefix, now);
- }
-
- public boolean appliesTo(ColumnFamily current)
- {
- return current == null || current.hasOnlyTombstones(now);
- }
- }
-
- private static class ColumnsConditions extends CQL3CasConditions
- {
- private final ColumnFamily expected;
-
- private ColumnsConditions(Composite rowPrefix,
- CFMetaData cfm,
- ByteBuffer key,
- Collection<Operation> conditions,
- List<ByteBuffer> variables,
- long now) throws InvalidRequestException
- {
- super(rowPrefix, now);
- this.expected = ArrayBackedSortedColumns.factory.create(cfm);
-
- // When building the conditions, we should not use a TTL. It's not useful, and if a very low ttl (1 seconds) is used, it's possible
- // for it to expire before the actual build of the conditions which would break since we would then testing for the presence of tombstones.
- UpdateParameters params = new UpdateParameters(cfm, variables, now, 0, null);
-
- // Conditions
- for (Operation condition : conditions)
- condition.execute(key, expected, rowPrefix, params);
- }
-
- public boolean appliesTo(ColumnFamily current)
- {
- if (current == null)
- return false;
-
- for (Cell e : expected)
- {
- Cell c = current.getColumn(e.name());
- if (e.isLive(now))
- {
- if (c == null || !c.isLive(now) || !c.value().equals(e.value()))
- return false;
- }
- else
- {
- // If we have a tombstone in expected, it means the condition tests that the column is
- // null, so check that we have no value
- if (c != null && c.isLive(now))
- return false;
- }
- }
- return true;
- }
-
- @Override
- public String toString()
- {
- return expected.toString();
- }
- }
-
public static abstract class Parsed extends CFStatement
{
protected final Attributes.Raw attrs;
@@@ -680,33 -756,23 +708,21 @@@
}
else
{
- for (Pair<ColumnIdentifier, Operation.RawUpdate> entry : conditions)
+ for (Pair<ColumnIdentifier, ColumnCondition.Raw> entry : conditions)
{
- CFDefinition.Name name = cfDef.get(entry.left);
- if (name == null)
+ ColumnDefinition def = metadata.getColumnDefinition(entry.left);
+ if (def == null)
throw new InvalidRequestException(String.format("Unknown identifier %s", entry.left));
- /*
- * Lists column names are based on a server-side generated timeuuid. So we can't allow lists
- * operation or that would yield unexpected results (update that should apply wouldn't). So for
- * now, we just refuse lists, which also save use from having to bother about the read that some
- * list operation involve.
- */
- if (def.type instanceof ListType)
- throw new InvalidRequestException(String.format("List operation (%s) are not allowed in conditional updates", def.name));
-
- Operation condition = entry.right.prepare(keyspace(), def);
- assert !condition.requiresRead();
-
- ColumnCondition condition = entry.right.prepare(name);
++ ColumnCondition condition = entry.right.prepare(keyspace(), def);
condition.collectMarkerSpecification(boundNames);
- switch (name.kind)
+ switch (def.kind)
{
- case KEY_ALIAS:
- case COLUMN_ALIAS:
+ case PARTITION_KEY:
+ case CLUSTERING_COLUMN:
throw new InvalidRequestException(String.format("PRIMARY KEY part %s found in SET part", entry.left));
- case COMPACT_VALUE:
- case REGULAR:
- case VALUE_ALIAS:
- case COLUMN_METADATA:
- case STATIC:
++ default:
stmt.addCondition(condition);
break;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/63b1ef4e/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index e08b960,2636c83..868d51c
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@@ -78,22 -80,60 +78,61 @@@ public class SelectStatement implement
private boolean isKeyRange;
private boolean keyIsInRelation;
private boolean usesSecondaryIndexing;
+ private boolean needOrderOnLastClustering;
- private Map<CFDefinition.Name, Integer> orderingIndexes;
+ private Map<ColumnIdentifier, Integer> orderingIndexes;
+ private boolean selectsStaticColumns;
+ private boolean selectsOnlyStaticColumns;
+
// Used by forSelection below
private static final Parameters defaultParameters = new Parameters(Collections.<ColumnIdentifier, Boolean>emptyMap(), false, false, null, false);
- private static final Predicate<CFDefinition.Name> isStaticFilter = new Predicate<CFDefinition.Name>()
++ private static final Predicate<ColumnDefinition> isStaticFilter = new Predicate<ColumnDefinition>()
+ {
- public boolean apply(CFDefinition.Name name)
++ public boolean apply(ColumnDefinition def)
+ {
- return name.kind == CFDefinition.Name.Kind.STATIC;
++ return def.isStatic();
+ }
+ };
+
- public SelectStatement(CFDefinition cfDef, int boundTerms, Parameters parameters, Selection selection, Term limit)
+ public SelectStatement(CFMetaData cfm, int boundTerms, Parameters parameters, Selection selection, Term limit)
{
- this.cfDef = cfDef;
+ this.cfm = cfm;
this.boundTerms = boundTerms;
this.selection = selection;
- this.keyRestrictions = new Restriction[cfDef.partitionKeyCount()];
- this.columnRestrictions = new Restriction[cfDef.clusteringColumnsCount()];
+ this.keyRestrictions = new Restriction[cfm.partitionKeyColumns().size()];
+ this.columnRestrictions = new Restriction[cfm.clusteringColumns().size()];
this.parameters = parameters;
this.limit = limit;
+
+ // Now gather a few info on whether we should bother with static columns or not for this statement
+ initStaticColumnsInfo();
+ }
+
+ private void initStaticColumnsInfo()
+ {
- if (!cfDef.cfm.hasStaticColumns())
++ if (!cfm.hasStaticColumns())
+ return;
+
+ // If it's a wildcard, we do select static but not only them
+ if (selection.isWildcard())
+ {
+ selectsStaticColumns = true;
+ return;
+ }
+
+ // Otherwise, check the selected columns
+ selectsStaticColumns = !Iterables.isEmpty(Iterables.filter(selection.getColumnsList(), isStaticFilter));
+ selectsOnlyStaticColumns = true;
- for (CFDefinition.Name name : selection.getColumnsList())
++ for (ColumnDefinition def : selection.getColumnsList())
+ {
- if (name.kind != CFDefinition.Name.Kind.KEY_ALIAS && name.kind != CFDefinition.Name.Kind.STATIC)
++ if (def.kind != ColumnDefinition.Kind.PARTITION_KEY && def.kind != ColumnDefinition.Kind.STATIC)
+ {
+ selectsOnlyStaticColumns = false;
+ break;
+ }
+ }
}
// Creates a simple select based on the given selection.
@@@ -378,35 -418,95 +417,89 @@@
}
else if (isColumnRange())
{
- // For sparse, we used to ask for 'defined columns' * 'asked limit' (where defined columns includes the row marker)
- // to account for the grouping of columns.
- // Since that doesn't work for maps/sets/lists, we now use the compositesToGroup option of SliceQueryFilter.
- // But we must preserve backward compatibility too (for mixed version cluster that is).
- int toGroup = cfDef.isCompact ? -1 : cfDef.clusteringColumnsCount();
- List<ByteBuffer> startBounds = getRequestedBound(Bound.START, variables);
- List<ByteBuffer> endBounds = getRequestedBound(Bound.END, variables);
+ int toGroup = cfm.comparator.isDense() ? -1 : cfm.clusteringColumns().size();
+ List<Composite> startBounds = getRequestedBound(Bound.START, variables);
+ List<Composite> endBounds = getRequestedBound(Bound.END, variables);
assert startBounds.size() == endBounds.size();
+ // Handles fetching static columns. Note that for 2i, the filter is just used to restrict
+ // the part of the index to query so adding the static slice would be useless and confusing.
+ // For 2i, static columns are retrieve in CompositesSearcher with each index hit.
+ ColumnSlice staticSlice = null;
+ if (selectsStaticColumns && !usesSecondaryIndexing)
+ {
- ColumnNameBuilder staticPrefix = cfDef.cfm.getStaticColumnNameBuilder();
- // Note: we could use staticPrefix.build() for the start bound, but EMPTY_BYTE_BUFFER gives us the
++ // Note: we could use staticPrefix.start() for the start bound, but EMPTY gives us the
+ // same effect while saving a few CPU cycles.
+ staticSlice = isReversed
- ? new ColumnSlice(staticPrefix.buildAsEndOfRange(), ByteBufferUtil.EMPTY_BYTE_BUFFER)
- : new ColumnSlice(ByteBufferUtil.EMPTY_BYTE_BUFFER, staticPrefix.buildAsEndOfRange());
++ ? new ColumnSlice(cfm.comparator.staticPrefix().end(), Composites.EMPTY)
++ : new ColumnSlice(Composites.EMPTY, cfm.comparator.staticPrefix().end());
+
+ // In the case where we only select static columns, we want to really only check the static columns.
+ // So we return early as the rest of that method would actually make us query everything
+ if (selectsOnlyStaticColumns)
+ return sliceFilter(staticSlice, limit, toGroup);
+ }
+
// The case where startBounds == 1 is common enough that it's worth optimizing
if (startBounds.size() == 1)
{
ColumnSlice slice = new ColumnSlice(startBounds.get(0), endBounds.get(0));
- if (slice.isAlwaysEmpty(cfDef.cfm.comparator, isReversed))
+ if (slice.isAlwaysEmpty(cfm.comparator, isReversed))
- return null;
- slices = new ColumnSlice[]{slice};
+ return staticSlice == null ? null : sliceFilter(staticSlice, limit, toGroup);
+
+ return staticSlice == null
+ ? sliceFilter(slice, limit, toGroup)
- : (slice.includes(cfDef.cfm.comparator, staticSlice.finish) ? sliceFilter(new ColumnSlice(staticSlice.start, slice.finish), limit, toGroup)
- : sliceFilter(new ColumnSlice[]{ staticSlice, slice }, limit, toGroup));
++ : (slice.includes(cfm.comparator, staticSlice.finish) ? sliceFilter(new ColumnSlice(staticSlice.start, slice.finish), limit, toGroup)
++ : sliceFilter(new ColumnSlice[]{ staticSlice, slice }, limit, toGroup));
+ }
+
+ List<ColumnSlice> l = new ArrayList<ColumnSlice>(startBounds.size());
+ for (int i = 0; i < startBounds.size(); i++)
+ {
+ ColumnSlice slice = new ColumnSlice(startBounds.get(i), endBounds.get(i));
- if (!slice.isAlwaysEmpty(cfDef.cfm.comparator, isReversed))
++ if (!slice.isAlwaysEmpty(cfm.comparator, isReversed))
+ l.add(slice);
+ }
+
+ if (l.isEmpty())
+ return staticSlice == null ? null : sliceFilter(staticSlice, limit, toGroup);
+ if (staticSlice == null)
+ return sliceFilter(l.toArray(new ColumnSlice[l.size()]), limit, toGroup);
+
- // The slices should not overlap. We know the slices built from startBounds/endBounds don't, but
- // if there is a static slice, it could overlap with the 2nd slice. Check for it and correct if
- // that's the case
++ // The slices should not overlap. We know the slices built from startBounds/endBounds don't, but if there is
++ // a static slice, it could overlap with the 2nd slice. Check for it and correct if that's the case
+ ColumnSlice[] slices;
+ if (isReversed)
+ {
- if (l.get(l.size() - 1).includes(cfDef.cfm.comparator, staticSlice.start))
++ if (l.get(l.size() - 1).includes(cfm.comparator, staticSlice.start))
+ {
+ slices = l.toArray(new ColumnSlice[l.size()]);
- slices[slices.length-1] = new ColumnSlice(slices[slices.length-1].start, ByteBufferUtil.EMPTY_BYTE_BUFFER);
++ slices[slices.length-1] = new ColumnSlice(slices[slices.length-1].start, Composites.EMPTY);
+ }
+ else
+ {
+ slices = l.toArray(new ColumnSlice[l.size()+1]);
+ slices[slices.length-1] = staticSlice;
+ }
}
else
{
- List<ColumnSlice> l = new ArrayList<ColumnSlice>(startBounds.size());
- for (int i = 0; i < startBounds.size(); i++)
- if (l.get(0).includes(cfDef.cfm.comparator, staticSlice.finish))
++ if (l.get(0).includes(cfm.comparator, staticSlice.finish))
{
- ColumnSlice slice = new ColumnSlice(startBounds.get(i), endBounds.get(i));
- if (!slice.isAlwaysEmpty(cfm.comparator, isReversed))
- l.add(slice);
+ slices = new ColumnSlice[l.size()];
- slices[0] = new ColumnSlice(ByteBufferUtil.EMPTY_BYTE_BUFFER, l.get(0).finish);
++ slices[0] = new ColumnSlice(Composites.EMPTY, l.get(0).finish);
+ for (int i = 1; i < l.size(); i++)
+ slices[i] = l.get(i);
+ }
+ else
+ {
+ slices = new ColumnSlice[l.size()+1];
+ slices[0] = staticSlice;
+ for (int i = 0; i < l.size(); i++)
+ slices[i] = l.get(i);
}
- if (l.isEmpty())
- return null;
- slices = l.toArray(new ColumnSlice[l.size()]);
}
-
- return new SliceQueryFilter(slices, isReversed, limit, toGroup);
+ return sliceFilter(slices, limit, toGroup);
}
else
{
@@@ -549,15 -659,17 +652,17 @@@
return false;
}
- private SortedSet<ByteBuffer> getRequestedColumns(List<ByteBuffer> variables) throws InvalidRequestException
+ private SortedSet<CellName> getRequestedColumns(List<ByteBuffer> variables) throws InvalidRequestException
{
+ // Note: getRequestedColumns don't handle static columns, but due to CASSANDRA-5762
+ // we always do a slice for CQL3 tables, so it's ok to ignore them here
assert !isColumnRange();
- ColumnNameBuilder builder = cfDef.getColumnNameBuilder();
- Iterator<CFDefinition.Name> idIter = cfDef.clusteringColumns().iterator();
+ CBuilder builder = cfm.comparator.prefixBuilder();
+ Iterator<ColumnDefinition> idIter = cfm.clusteringColumns().iterator();
for (Restriction r : columnRestrictions)
{
- ColumnIdentifier id = idIter.next().name;
+ ColumnDefinition def = idIter.next();
assert r != null && !r.isSlice();
List<ByteBuffer> values = r.values(variables);
@@@ -611,20 -726,26 +716,21 @@@
// column (for the case where the row exists but has no columns outside the PK)
// Two exceptions are "static CF" (non-composite non-compact CF) and "super CF"
// that don't have marker and for which we must query all columns instead
- if (cfDef.isComposite && !cfDef.cfm.isSuper())
+ if (cfm.comparator.isCompound() && !cfm.isSuper())
{
// marker
- columns.add(builder.copy().add(ByteBufferUtil.EMPTY_BYTE_BUFFER).build());
+ columns.add(cfm.comparator.rowMarker(prefix));
// selected columns
- for (ColumnIdentifier id : selection.regularAndStaticColumnsToFetch())
- columns.add(builder.copy().add(id.key).build());
+ for (ColumnDefinition def : selection.getColumnsList())
- if (def.kind == ColumnDefinition.Kind.REGULAR)
- columns.add(cfm.comparator.create(prefix, def.name));
++ if (def.kind == ColumnDefinition.Kind.REGULAR || def.kind == ColumnDefinition.Kind.STATIC)
++ columns.add(cfm.comparator.create(prefix, def));
}
else
{
+ // We now that we're not composite so we can ignore static columns
- Iterator<CFDefinition.Name> iter = cfDef.regularColumns().iterator();
- while (iter.hasNext())
- {
- ColumnIdentifier name = iter.next().name;
- ColumnNameBuilder b = iter.hasNext() ? builder.copy() : builder;
- ByteBuffer cname = b.add(name.key).build();
- columns.add(cname);
- }
+ for (ColumnDefinition def : cfm.regularColumns())
- columns.add(cfm.comparator.create(prefix, def.name));
++ columns.add(cfm.comparator.create(prefix, def));
}
return columns;
}
@@@ -777,22 -877,23 +883,23 @@@
return Collections.emptyList();
List<IndexExpression> expressions = new ArrayList<IndexExpression>();
- for (CFDefinition.Name name : restrictedNames)
+ for (ColumnDefinition def : restrictedColumns)
{
Restriction restriction;
- switch (name.kind)
+ switch (def.kind)
{
- case KEY_ALIAS:
- restriction = keyRestrictions[name.position];
+ case PARTITION_KEY:
+ restriction = keyRestrictions[def.position()];
break;
- case COLUMN_ALIAS:
- restriction = columnRestrictions[name.position];
+ case CLUSTERING_COLUMN:
+ restriction = columnRestrictions[def.position()];
break;
- case COLUMN_METADATA:
+ case REGULAR:
+ case STATIC:
- restriction = metadataRestrictions.get(name);
+ restriction = metadataRestrictions.get(def.name);
break;
default:
- // We don't allow restricting a VALUE_ALIAS for now in prepare.
+ // We don't allow restricting a COMPACT_VALUE for now in prepare.
throw new AssertionError();
}
@@@ -903,60 -1010,140 +1010,107 @@@
void processColumnFamily(ByteBuffer key, ColumnFamily cf, List<ByteBuffer> variables, long now, Selection.ResultSetBuilder result)
throws InvalidRequestException
{
- ByteBuffer[] keyComponents = cfDef.hasCompositeKey
- ? ((CompositeType)cfDef.cfm.getKeyValidator()).split(key)
- : new ByteBuffer[]{ key };
-
- if (parameters.isDistinct)
+ CFMetaData cfm = cf.metadata();
+ ByteBuffer[] keyComponents = null;
+ if (cfm.getKeyValidator() instanceof CompositeType)
{
- if (!cf.hasOnlyTombstones(now))
- {
- result.newRow();
- // selection.getColumnsList() will contain only the partition key components - all of them.
- for (CFDefinition.Name name : selection.getColumnsList())
- result.add(keyComponents[name.position]);
- }
+ keyComponents = ((CompositeType)cfm.getKeyValidator()).split(key);
}
- else if (cfDef.isCompact)
+ else
{
- // One cqlRow per column
- for (Column c : columnsInOrder(cf, variables))
- {
- if (c.isMarkedForDelete(now))
- continue;
+ keyComponents = new ByteBuffer[]{ key };
+ }
- ByteBuffer[] components = null;
- if (cfDef.isComposite)
- {
- components = ((CompositeType)cfDef.cfm.comparator).split(c.name());
- }
- else if (sliceRestriction != null)
- {
- // For dynamic CF, the column could be out of the requested bounds, filter here
- if (!sliceRestriction.isInclusive(Bound.START) && c.name().equals(sliceRestriction.bound(Bound.START, variables)))
- continue;
- if (!sliceRestriction.isInclusive(Bound.END) && c.name().equals(sliceRestriction.bound(Bound.END, variables)))
- continue;
- }
+ Iterator<Cell> cells = cf.getSortedColumns().iterator();
+ if (sliceRestriction != null)
+ cells = applySliceRestriction(cells, variables);
- for (Iterator<CQL3Row> iter = cfm.comparator.CQL3RowBuilder(now).group(cells); iter.hasNext();)
- result.newRow();
- // Respect selection order
- for (CFDefinition.Name name : selection.getColumnsList())
++ CQL3Row.RowIterator iter = cfm.comparator.CQL3RowBuilder(now).group(cells);
++
++ // If there is static columns but there is no non-static row, then provided the select was a full
++ // partition selection (i.e. not a 2ndary index search and there was no condition on clustering columns)
++ // then we want to include the static columns in the result set (and we're done).
++ CQL3Row staticRow = iter.getStaticRow();
++ if (staticRow != null && !iter.hasNext() && !usesSecondaryIndexing && hasNoClusteringColumnsRestriction())
++ {
++ result.newRow();
++ for (ColumnDefinition def : selection.getColumnsList())
++ {
++ switch (def.kind)
+ {
- switch (name.kind)
- {
- case KEY_ALIAS:
- result.add(keyComponents[name.position]);
- break;
- case COLUMN_ALIAS:
- ByteBuffer val = cfDef.isComposite
- ? (name.position < components.length ? components[name.position] : null)
- : c.name();
- result.add(val);
- break;
- case VALUE_ALIAS:
- result.add(c);
- break;
- case COLUMN_METADATA:
- case STATIC:
- // This should not happen for compact CF
- throw new AssertionError();
- default:
- throw new AssertionError();
- }
++ case PARTITION_KEY:
++ result.add(keyComponents[def.position()]);
++ break;
++ case STATIC:
++ addValue(result, def, staticRow);
++ break;
++ default:
++ result.add((ByteBuffer)null);
+ }
+ }
++ return;
+ }
- else if (cfDef.isComposite)
- {
- // Sparse case: group column in cqlRow when composite prefix is equal
- CompositeType composite = (CompositeType)cfDef.cfm.comparator;
+
- ColumnGroupMap.Builder builder = new ColumnGroupMap.Builder(composite, cfDef.hasCollections, now);
-
- for (Column c : cf)
- {
- if (c.isMarkedForDelete(now))
- continue;
-
- builder.add(c);
- }
++ while (iter.hasNext())
+ {
+ CQL3Row cql3Row = iter.next();
- Map<CFDefinition.Name, ByteBuffer> staticValues = Collections.emptyMap();
- // Gather up static values first
- if (!builder.isEmpty() && builder.firstGroup().isStatic)
+ // Respect requested order
+ result.newRow();
+ // Respect selection order
+ for (ColumnDefinition def : selection.getColumnsList())
{
- staticValues = new HashMap<>();
- ColumnGroupMap group = builder.firstGroup();
- for (CFDefinition.Name name : Iterables.filter(selection.getColumnsList(), isStaticFilter))
- staticValues.put(name, name.type.isCollection() ? getCollectionValue(name, group) : getSimpleValue(name, group));
- builder.discardFirst();
-
- // If there was static columns but there is no actual row, then provided the select was a full
- // partition selection (i.e. not a 2ndary index search and there was no condition on clustering columns)
- // then we want to include the static columns in the result set.
- if (!staticValues.isEmpty() && builder.isEmpty() && !usesSecondaryIndexing && hasNoClusteringColumnsRestriction())
+ switch (def.kind)
{
- result.newRow();
- for (CFDefinition.Name name : selection.getColumnsList())
- {
- if (name.kind == CFDefinition.Name.Kind.KEY_ALIAS)
- result.add(keyComponents[name.position]);
- else
- result.add(name.kind == CFDefinition.Name.Kind.STATIC ? staticValues.get(name) : null);
- }
- return;
+ case PARTITION_KEY:
+ result.add(keyComponents[def.position()]);
+ break;
+ case CLUSTERING_COLUMN:
+ result.add(cql3Row.getClusteringColumn(def.position()));
+ break;
+ case COMPACT_VALUE:
+ result.add(cql3Row.getColumn(null));
+ break;
+ case REGULAR:
- if (def.type.isCollection())
- {
- List<Cell> collection = cql3Row.getCollection(def.name);
- ByteBuffer value = collection == null
- ? null
- : ((CollectionType)def.type).serialize(collection);
- result.add(value);
- }
- else
- {
- result.add(cql3Row.getColumn(def.name));
- }
++ addValue(result, def, cql3Row);
++ break;
++ case STATIC:
++ addValue(result, def, staticRow);
+ break;
- }
}
+ }
-
- for (ColumnGroupMap group : builder.groups())
- handleGroup(selection, result, keyComponents, group, staticValues);
}
- else
+ }
+
++ private static void addValue(Selection.ResultSetBuilder result, ColumnDefinition def, CQL3Row row)
++ {
++ if (row == null)
+ {
- if (cf.hasOnlyTombstones(now))
- return;
++ result.add((ByteBuffer)null);
++ return;
++ }
+
- // Static case: One cqlRow for all columns
- result.newRow();
- for (CFDefinition.Name name : selection.getColumnsList())
- {
- if (name.kind == CFDefinition.Name.Kind.KEY_ALIAS)
- result.add(keyComponents[name.position]);
- else
- result.add(cf.getColumn(name.name.key));
- }
++ if (def.type.isCollection())
++ {
++ List<Cell> collection = row.getCollection(def.name);
++ ByteBuffer value = collection == null
++ ? null
++ : ((CollectionType)def.type).serialize(collection);
++ result.add(value);
++ return;
+ }
++
++ result.add(row.getColumn(def.name));
+ }
+
+ private boolean hasNoClusteringColumnsRestriction()
+ {
+ for (int i = 0; i < columnRestrictions.length; i++)
+ if (columnRestrictions[i] != null)
+ return false;
+ return true;
+ }
+
/**
* Orders results when multiple keys are selected (using IN)
*/
@@@ -1119,23 -1331,24 +1281,24 @@@
hasQueriableClusteringColumnIndex = true;
}
- switch (name.kind)
+ switch (def.kind)
{
- case KEY_ALIAS:
- stmt.keyRestrictions[name.position] = updateRestriction(cfm, name, stmt.keyRestrictions[name.position], rel, names);
+ case PARTITION_KEY:
+ stmt.keyRestrictions[def.position()] = updateRestriction(cfm, def, stmt.keyRestrictions[def.position()], rel, names);
break;
- case COLUMN_ALIAS:
- stmt.columnRestrictions[name.position] = updateRestriction(cfm, name, stmt.columnRestrictions[name.position], rel, names);
+ case CLUSTERING_COLUMN:
+ stmt.columnRestrictions[def.position()] = updateRestriction(cfm, def, stmt.columnRestrictions[def.position()], rel, names);
break;
- case VALUE_ALIAS:
- throw new InvalidRequestException(String.format("Predicates on the non-primary-key column (%s) of a COMPACT table are not yet supported", name.name));
- case COLUMN_METADATA:
+ case COMPACT_VALUE:
+ throw new InvalidRequestException(String.format("Predicates on the non-primary-key column (%s) of a COMPACT table are not yet supported", def.name));
+ case REGULAR:
+ case STATIC:
// We only all IN on the row key and last clustering key so far, never on non-PK columns, and this even if there's an index
- Restriction r = updateRestriction(cfm, name, stmt.metadataRestrictions.get(name), rel, names);
+ Restriction r = updateRestriction(cfm, def, stmt.metadataRestrictions.get(def.name), rel, names);
if (r.isIN() && !((Restriction.IN)r).canHaveOnlyOneValue())
// Note: for backward compatibility reason, we conside a IN of 1 value the same as a EQ, so we let that slide.
- throw new InvalidRequestException(String.format("IN predicates on non-primary-key columns (%s) is not yet supported", name));
- stmt.metadataRestrictions.put(name, r);
+ throw new InvalidRequestException(String.format("IN predicates on non-primary-key columns (%s) is not yet supported", def.name));
+ stmt.metadataRestrictions.put(def.name, r);
break;
}
}
@@@ -1222,8 -1435,11 +1385,11 @@@
// All (or none) of the partition key columns have been specified;
// hence there is no need to turn these restrictions into index expressions.
if (!stmt.usesSecondaryIndexing)
- stmt.restrictedNames.removeAll(cfDef.partitionKeys());
+ stmt.restrictedColumns.removeAll(cfm.partitionKeyColumns());
+ if (stmt.selectsOnlyStaticColumns && stmt.hasClusteringColumnsRestriction())
+ throw new InvalidRequestException("Cannot restrict clustering columns when selecting only static columns");
+
// If a clustering key column is restricted by a non-EQ relation, all preceding
// columns must have a EQ, and all following must have no restriction. Unless
// the column is indexed that is.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/63b1ef4e/src/java/org/apache/cassandra/cql3/statements/Selection.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/cql3/statements/Selection.java
index 9c2eb0c,9760311..c506cc6
--- a/src/java/org/apache/cassandra/cql3/statements/Selection.java
+++ b/src/java/org/apache/cassandra/cql3/statements/Selection.java
@@@ -55,30 -49,30 +55,36 @@@ public abstract class Selectio
this.collectTTLs = collectTTLs;
}
+ // Overriden by SimpleSelection when appropriate.
+ public boolean isWildcard()
+ {
+ return false;
+ }
+
public ResultSet.Metadata getResultMetadata()
{
- return new ResultSet.Metadata(metadata);
+ return metadata;
}
- public static Selection wildcard(CFDefinition cfDef)
+ public static Selection wildcard(CFMetaData cfm)
{
- List<CFDefinition.Name> all = new ArrayList<CFDefinition.Name>();
- for (CFDefinition.Name name : cfDef)
- all.add(name);
+ List<ColumnDefinition> all = new ArrayList<ColumnDefinition>(cfm.allColumns().size());
+ Iterators.addAll(all, cfm.allColumnsInSelectOrder());
- return new SimpleSelection(all);
+ return new SimpleSelection(all, true);
}
- public static Selection forColumns(List<CFDefinition.Name> columnsList)
+ public static Selection forColumns(List<ColumnDefinition> columnsList)
{
- return new SimpleSelection(columnsList);
+ return new SimpleSelection(columnsList, false);
}
+ public int addColumnForOrdering(ColumnDefinition c)
+ {
+ columnsList.add(c);
+ metadata.addNonSerializedColumn(c);
+ return columnsList.size() - 1;
+ }
+
private static boolean isUsingFunction(List<RawSelector> rawSelectors)
{
for (RawSelector rawSelector : rawSelectors)
@@@ -114,12 -108,12 +120,12 @@@
else if (raw.selectable instanceof Selectable.WritetimeOrTTL)
{
Selectable.WritetimeOrTTL tot = (Selectable.WritetimeOrTTL)raw.selectable;
- CFDefinition.Name name = cfDef.get(tot.id);
- if (name == null)
+ ColumnDefinition def = cfm.getColumnDefinition(tot.id);
+ if (def == null)
throw new InvalidRequestException(String.format("Undefined name %s in selection clause", tot.id));
- if (def.kind != ColumnDefinition.Kind.REGULAR && def.kind != ColumnDefinition.Kind.COMPACT_VALUE)
- if (name.isPrimaryKeyColumn())
- throw new InvalidRequestException(String.format("Cannot use selection function %s on PRIMARY KEY part %s", tot.isWritetime ? "writeTime" : "ttl", name));
- if (name.type.isCollection())
++ if (def.isPrimaryKeyColumn())
+ throw new InvalidRequestException(String.format("Cannot use selection function %s on PRIMARY KEY part %s", tot.isWritetime ? "writeTime" : "ttl", def.name));
+ if (def.type.isCollection())
throw new InvalidRequestException(String.format("Cannot use selection function %s on collections", tot.isWritetime ? "writeTime" : "ttl"));
if (metadata != null)
@@@ -229,13 -195,13 +235,13 @@@
for (RawSelector rawSelector : rawSelectors)
{
assert rawSelector.selectable instanceof ColumnIdentifier;
- CFDefinition.Name name = cfDef.get((ColumnIdentifier)rawSelector.selectable);
- if (name == null)
+ ColumnDefinition def = cfm.getColumnDefinition((ColumnIdentifier)rawSelector.selectable);
+ if (def == null)
throw new InvalidRequestException(String.format("Undefined name %s in selection clause", rawSelector.selectable));
- names.add(name);
- metadata.add(rawSelector.alias == null ? name : makeAliasSpec(cfDef, name.type, rawSelector.alias));
+ defs.add(def);
+ metadata.add(rawSelector.alias == null ? def : makeAliasSpec(cfm, def.type, rawSelector.alias));
}
- return new SimpleSelection(defs, metadata);
- return new SimpleSelection(names, metadata, false);
++ return new SimpleSelection(defs, metadata, false);
}
}
@@@ -333,12 -313,14 +339,14 @@@
// Special cased selection for when no function is used (this save some allocations).
private static class SimpleSelection extends Selection
{
- public SimpleSelection(List<ColumnDefinition> columnsList)
+ private final boolean isWildcard;
+
- public SimpleSelection(List<CFDefinition.Name> columnsList, boolean isWildcard)
++ public SimpleSelection(List<ColumnDefinition> columnsList, boolean isWildcard)
{
- this(columnsList, new ArrayList<ColumnSpecification>(columnsList));
+ this(columnsList, new ArrayList<ColumnSpecification>(columnsList), isWildcard);
}
- public SimpleSelection(List<ColumnDefinition> columnsList, List<ColumnSpecification> metadata)
- public SimpleSelection(List<CFDefinition.Name> columnsList, List<ColumnSpecification> metadata, boolean isWildcard)
++ public SimpleSelection(List<ColumnDefinition> columnsList, List<ColumnSpecification> metadata, boolean isWildcard)
{
/*
* In theory, even a simple selection could have multiple time the same column, so we
@@@ -352,20 -335,20 +361,26 @@@
{
return rs.current;
}
+
+ @Override
+ public boolean isWildcard()
+ {
+ return isWildcard;
+ }
}
- private interface Selector extends AssignementTestable
+ private static abstract class Selector implements AssignementTestable
{
- public ByteBuffer compute(ResultSetBuilder rs) throws InvalidRequestException;
+ public abstract ByteBuffer compute(ResultSetBuilder rs) throws InvalidRequestException;
+ public abstract AbstractType<?> getType();
+
+ public boolean isAssignableTo(String keyspace, ColumnSpecification receiver)
+ {
+ return getType().asCQL3Type().equals(receiver.type.asCQL3Type());
+ }
}
- private static class SimpleSelector implements Selector
+ private static class SimpleSelector extends Selector
{
private final String columnName;
private final int idx;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/63b1ef4e/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
index 6ed0e33,0e6481b..069f31f
--- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
@@@ -35,11 -33,11 +35,11 @@@ import org.apache.cassandra.utils.Pair
*/
public class UpdateStatement extends ModificationStatement
{
- private static final Operation setToEmptyOperation = new Constants.Setter(null, new Constants.Value(ByteBufferUtil.EMPTY_BYTE_BUFFER));
+ private static final Constants.Value EMPTY = new Constants.Value(ByteBufferUtil.EMPTY_BYTE_BUFFER);
- private UpdateStatement(int boundTerms, CFMetaData cfm, Attributes attrs)
- private UpdateStatement(StatementType type, CFMetaData cfm, Attributes attrs)
++ private UpdateStatement(StatementType type, int boundTerms, CFMetaData cfm, Attributes attrs)
{
- super(boundTerms, cfm, attrs);
- super(type, cfm, attrs);
++ super(type, boundTerms, cfm, attrs);
}
public boolean requireFullClusteringKey()
@@@ -61,8 -61,11 +61,8 @@@
// 'DELETE FROM t WHERE k = 1' does remove the row entirely)
//
// We never insert markers for Super CF as this would confuse the thrift side.
- if (cfm.isCQL3Table())
- if (cfDef.isComposite && !cfDef.isCompact && !cfm.isSuper())
- {
- ByteBuffer name = builder.copy().add(ByteBufferUtil.EMPTY_BYTE_BUFFER).build();
- cf.addColumn(params.makeColumn(name, ByteBufferUtil.EMPTY_BYTE_BUFFER));
- }
++ if (cfm.isCQL3Table() && !prefix.isStatic())
+ cf.addColumn(params.makeColumn(cfm.comparator.rowMarker(prefix), ByteBufferUtil.EMPTY_BYTE_BUFFER));
List<Operation> updates = getOperations();
@@@ -127,9 -129,9 +127,9 @@@
this.columnValues = columnValues;
}
- protected ModificationStatement prepareInternal(CFDefinition cfDef, VariableSpecifications boundNames, Attributes attrs) throws InvalidRequestException
+ protected ModificationStatement prepareInternal(CFMetaData cfm, VariableSpecifications boundNames, Attributes attrs) throws InvalidRequestException
{
- UpdateStatement stmt = new UpdateStatement(boundNames.size(), cfm, attrs);
- UpdateStatement stmt = new UpdateStatement(ModificationStatement.StatementType.INSERT, cfDef.cfm, attrs);
++ UpdateStatement stmt = new UpdateStatement(ModificationStatement.StatementType.INSERT,boundNames.size(), cfm, attrs);
// Created from an INSERT
if (stmt.isCounter())
@@@ -151,17 -153,18 +151,16 @@@
Term.Raw value = columnValues.get(i);
- switch (name.kind)
+ switch (def.kind)
{
- case KEY_ALIAS:
- case COLUMN_ALIAS:
- Term t = value.prepare(name);
+ case PARTITION_KEY:
+ case CLUSTERING_COLUMN:
+ Term t = value.prepare(keyspace(), def);
t.collectMarkerSpecification(boundNames);
- stmt.addKeyValue(def.name, t);
- stmt.addKeyValue(name, t);
++ stmt.addKeyValue(def, t);
break;
- case COMPACT_VALUE:
- case REGULAR:
- case VALUE_ALIAS:
- case COLUMN_METADATA:
- case STATIC:
- Operation operation = new Operation.SetValue(value).prepare(name);
++ default:
+ Operation operation = new Operation.SetValue(value).prepare(keyspace(), def);
operation.collectMarkerSpecification(boundNames);
stmt.addOperation(operation);
break;
@@@ -197,26 -200,27 +196,25 @@@
this.whereClause = whereClause;
}
- protected ModificationStatement prepareInternal(CFDefinition cfDef, VariableSpecifications boundNames, Attributes attrs) throws InvalidRequestException
+ protected ModificationStatement prepareInternal(CFMetaData cfm, VariableSpecifications boundNames, Attributes attrs) throws InvalidRequestException
{
- UpdateStatement stmt = new UpdateStatement(boundNames.size(), cfm, attrs);
- UpdateStatement stmt = new UpdateStatement(ModificationStatement.StatementType.UPDATE, cfDef.cfm, attrs);
++ UpdateStatement stmt = new UpdateStatement(ModificationStatement.StatementType.UPDATE, boundNames.size(), cfm, attrs);
for (Pair<ColumnIdentifier, Operation.RawUpdate> entry : updates)
{
- CFDefinition.Name name = cfDef.get(entry.left);
- if (name == null)
+ ColumnDefinition def = cfm.getColumnDefinition(entry.left);
+ if (def == null)
throw new InvalidRequestException(String.format("Unknown identifier %s", entry.left));
- Operation operation = entry.right.prepare(name);
+ Operation operation = entry.right.prepare(keyspace(), def);
operation.collectMarkerSpecification(boundNames);
- switch (name.kind)
+ switch (def.kind)
{
- case KEY_ALIAS:
- case COLUMN_ALIAS:
+ case PARTITION_KEY:
+ case CLUSTERING_COLUMN:
throw new InvalidRequestException(String.format("PRIMARY KEY part %s found in SET part", entry.left));
- case COMPACT_VALUE:
- case REGULAR:
- case VALUE_ALIAS:
- case COLUMN_METADATA:
- case STATIC:
++ default:
stmt.addOperation(operation);
break;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/63b1ef4e/src/java/org/apache/cassandra/db/CFRowAdder.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/CFRowAdder.java
index 9fd5e9a,0000000..f853d17
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/db/CFRowAdder.java
+++ b/src/java/org/apache/cassandra/db/CFRowAdder.java
@@@ -1,110 -1,0 +1,110 @@@
+/*
+ * 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 org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.db.composites.CellName;
+import org.apache.cassandra.db.composites.Composite;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.CollectionType;
+import org.apache.cassandra.db.marshal.ListType;
+import org.apache.cassandra.db.marshal.MapType;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.UUIDGen;
+
+/**
+ * Convenience object to populate a given CQL3 row in a ColumnFamily object.
+ *
+ * This is meant for when performance is not of the utmost importance. When
+ * performance matters, it might be worth allocating such builder.
+ */
+public class CFRowAdder
+{
+ public final ColumnFamily cf;
+ public final Composite prefix;
+ public final long timestamp;
+ private final int ldt;
+
+ public CFRowAdder(ColumnFamily cf, Composite prefix, long timestamp)
+ {
+ this.cf = cf;
+ this.prefix = prefix;
+ this.timestamp = timestamp;
+ this.ldt = (int) (System.currentTimeMillis() / 1000);
+
+ // If a CQL3 table, add the row marker
- if (cf.metadata().isCQL3Table())
++ if (cf.metadata().isCQL3Table() && !prefix.isStatic())
+ cf.addColumn(new Cell(cf.getComparator().rowMarker(prefix), ByteBufferUtil.EMPTY_BYTE_BUFFER, timestamp));
+ }
+
+ public CFRowAdder add(String cql3ColumnName, Object value)
+ {
+ ColumnDefinition def = getDefinition(cql3ColumnName);
- return add(cf.getComparator().create(prefix, def.name), def, value);
++ return add(cf.getComparator().create(prefix, def), def, value);
+ }
+
+ public CFRowAdder resetCollection(String cql3ColumnName)
+ {
+ ColumnDefinition def = getDefinition(cql3ColumnName);
+ assert def.type.isCollection();
- Composite name = cf.getComparator().create(prefix, def.name);
++ Composite name = cf.getComparator().create(prefix, def);
+ cf.addAtom(new RangeTombstone(name.start(), name.end(), timestamp - 1, ldt));
+ return this;
+ }
+
+ public CFRowAdder addMapEntry(String cql3ColumnName, Object key, Object value)
+ {
+ ColumnDefinition def = getDefinition(cql3ColumnName);
+ assert def.type instanceof MapType;
+ MapType mt = (MapType)def.type;
- CellName name = cf.getComparator().create(prefix, def.name, mt.keys.decompose(key));
++ CellName name = cf.getComparator().create(prefix, def, mt.keys.decompose(key));
+ return add(name, def, value);
+ }
+
+ public CFRowAdder addListEntry(String cql3ColumnName, Object value)
+ {
+ ColumnDefinition def = getDefinition(cql3ColumnName);
+ assert def.type instanceof ListType;
- CellName name = cf.getComparator().create(prefix, def.name, ByteBuffer.wrap(UUIDGen.getTimeUUIDBytes()));
++ CellName name = cf.getComparator().create(prefix, def, ByteBuffer.wrap(UUIDGen.getTimeUUIDBytes()));
+ return add(name, def, value);
+ }
+
+ private ColumnDefinition getDefinition(String name)
+ {
+ return cf.metadata().getColumnDefinition(new ColumnIdentifier(name, false));
+ }
+
+ private CFRowAdder add(CellName name, ColumnDefinition def, Object value)
+ {
+ if (value == null)
+ {
+ cf.addColumn(new DeletedCell(name, ldt, timestamp));
+ }
+ else
+ {
+ AbstractType valueType = def.type.isCollection()
+ ? ((CollectionType) def.type).valueComparator()
+ : def.type;
+ cf.addColumn(new Cell(name, value instanceof ByteBuffer ? (ByteBuffer)value : valueType.decompose(value), timestamp));
+ }
+ return this;
+ }
+}