content: /*
   * 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.index.sai.plan;
  
  import java.nio.ByteBuffer;
  import java.util.Arrays;
  import java.util.EnumMap;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.TimeUnit;
  
  import com.google.common.base.Preconditions;
  import com.google.common.collect.Sets;
  import org.junit.Before;
  import org.junit.BeforeClass;
  import org.junit.Test;
  
  import org.apache.cassandra.SchemaLoader;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.cql3.Operator;
  import org.apache.cassandra.cql3.statements.schema.IndexTarget;
  import org.apache.cassandra.db.Clustering;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.db.DecoratedKey;
  import org.apache.cassandra.db.Keyspace;
  import org.apache.cassandra.db.PartitionRangeReadCommand;
  import org.apache.cassandra.db.ReadCommand;
  import org.apache.cassandra.db.filter.RowFilter;
  import org.apache.cassandra.db.marshal.AbstractType;
  import org.apache.cassandra.db.marshal.CollectionType;
  import org.apache.cassandra.db.marshal.CompositeType;
  import org.apache.cassandra.db.marshal.DoubleType;
  import org.apache.cassandra.db.marshal.Int32Type;
  import org.apache.cassandra.db.marshal.ListType;
  import org.apache.cassandra.db.marshal.LongType;
  import org.apache.cassandra.db.marshal.MapType;
  import org.apache.cassandra.db.marshal.UTF8Type;
  import org.apache.cassandra.db.rows.BTreeRow;
  import org.apache.cassandra.db.rows.BufferCell;
  import org.apache.cassandra.db.rows.Cell;
  import org.apache.cassandra.db.rows.Row;
  import org.apache.cassandra.db.rows.Unfiltered;
  import org.apache.cassandra.dht.Murmur3Partitioner;
  import org.apache.cassandra.exceptions.ConfigurationException;
  import org.apache.cassandra.index.sai.QueryContext;
  import org.apache.cassandra.index.sai.StorageAttachedIndex;
  import org.apache.cassandra.schema.ColumnMetadata;
  import org.apache.cassandra.schema.IndexMetadata;
  import org.apache.cassandra.schema.Indexes;
  import org.apache.cassandra.schema.KeyspaceParams;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.utils.FBUtilities;
  
  import static org.apache.cassandra.config.CassandraRelevantProperties.CASSANDRA_CONFIG;
  import static org.apache.cassandra.db.marshal.Int32Type.instance;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertTrue;
  
  public class OperationTest
  {
      private static final String KS_NAME = "sai";
      private static final String CF_NAME = "test_cf";
      private static final String CLUSTERING_CF_NAME = "clustering_test_cf";
      private static final String STATIC_CF_NAME = "static_sai_test_cf";
  
      private static ColumnFamilyStore BACKEND;
      private static ColumnFamilyStore CLUSTERING_BACKEND;
      private static ColumnFamilyStore STATIC_BACKEND;
  
      private QueryController controller;
      private QueryController controllerClustering;
      private QueryController controllerStatic;
  
      @BeforeClass
      public static void loadSchema() throws ConfigurationException
      {
          CASSANDRA_CONFIG.setString("cassandra-murmur.yaml");
  
          SchemaLoader.loadSchema();
  
          SchemaLoader.createKeyspace(KS_NAME,
                                      KeyspaceParams.simpleTransient(1),
                                      skinnySAITableMetadata(KS_NAME, CF_NAME),
                                      clusteringSAITableMetadata(KS_NAME, CLUSTERING_CF_NAME),
                                      staticSAITableMetadata(KS_NAME, STATIC_CF_NAME));
  
          BACKEND = Keyspace.open(KS_NAME).getColumnFamilyStore(CF_NAME);
          CLUSTERING_BACKEND = Keyspace.open(KS_NAME).getColumnFamilyStore(CLUSTERING_CF_NAME);
          STATIC_BACKEND = Keyspace.open(KS_NAME).getColumnFamilyStore(STATIC_CF_NAME);
      }
  
      @Before
      public void beforeTest()
      {
          ReadCommand command = PartitionRangeReadCommand.allDataRead(BACKEND.metadata(), FBUtilities.nowInSeconds());
          controller = new QueryController(BACKEND, command, RowFilter.none(), contextWithUnrepairedMatches(command));
  
          command = PartitionRangeReadCommand.allDataRead(CLUSTERING_BACKEND.metadata(), FBUtilities.nowInSeconds());
          controllerClustering = new QueryController(CLUSTERING_BACKEND, command, RowFilter.none(), contextWithUnrepairedMatches(command));
  
          command = PartitionRangeReadCommand.allDataRead(STATIC_BACKEND.metadata(), FBUtilities.nowInSeconds());
          controllerStatic = new QueryController(STATIC_BACKEND, command, RowFilter.none(), contextWithUnrepairedMatches(command));
      }
  
      private static QueryContext contextWithUnrepairedMatches(ReadCommand command)
      {
          QueryContext context = new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS));
          context.hasUnrepairedMatches = true;
          return context;
      }
  
      @Test
      public void testAnalyze()
      {
          final ColumnMetadata age = getColumn(UTF8Type.instance.decompose("age"));
  
          // age > 1 AND age < 7
          Map<Expression.IndexOperator, Expression> expressions = convert(Operation.buildIndexExpressions(controller,
                                                                                                          Arrays.asList(new SimpleExpression(age, Operator.GT, Int32Type.instance.decompose(1)),
                                                                                                                        new SimpleExpression(age, Operator.LT, Int32Type.instance.decompose(7)))));
  
          assertEquals(1, expressions.size());
  
          Expression rangeExpression = expressions.get(Expression.IndexOperator.RANGE);
  
          assertExpression(rangeExpression, Expression.IndexOperator.RANGE, Int32Type.instance.decompose(1), false, Int32Type.instance.decompose(7), false);
      }
  
      @Test
      public void testSatisfiedBy()
      {
          final ColumnMetadata timestamp = getColumn(UTF8Type.instance.decompose("timestamp"));
          final ColumnMetadata age = getColumn(UTF8Type.instance.decompose("age"));
  
          Operation.Node node = new Operation.ExpressionNode(new SimpleExpression(age, Operator.EQ, Int32Type.instance.decompose(5)));
          FilterTree filterTree = node.buildFilter(controller, true);
  
          DecoratedKey key = buildKey("0");
          Unfiltered row = buildRow(buildCell(age, instance.decompose(6), System.currentTimeMillis()));
          Row staticRow = buildRow(Clustering.STATIC_CLUSTERING);
  
          assertFalse(filterTree.isSatisfiedBy(key, (Row) row, staticRow));
  
          row = buildRow(buildCell(age, instance.decompose(5), System.currentTimeMillis()));
  
          assertTrue(filterTree.isSatisfiedBy(key, (Row) row, staticRow));
  
          row = buildRow(buildCell(age, instance.decompose(6), System.currentTimeMillis()));
  
          assertFalse(filterTree.isSatisfiedBy(key, (Row) row, staticRow));
  
          // range with exclusions - age > 1 AND age <= 10
          node = new Operation.AndNode();
          node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.GT, Int32Type.instance.decompose(1))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.LTE, Int32Type.instance.decompose(10))));
          filterTree = node.buildFilter(controller, true);
  
          Set<Integer> exclusions = Sets.newHashSet(0, 1, 11);
          for (int i = 0; i <= 11; i++)
          {
              row = buildRow(buildCell(age, instance.decompose(i), System.currentTimeMillis()));
  
              boolean result = filterTree.isSatisfiedBy(key, (Row) row, staticRow);
              assertTrue(exclusions.contains(i) != result);
          }
  
          // now let's test aggregated AND commands
          node = new Operation.AndNode();
  
          node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.GTE, Int32Type.instance.decompose(0))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.LT, Int32Type.instance.decompose(10))));
  
          filterTree = node.buildFilter(controller, true);
  
          for (int i = 0; i < 10; i++)
          {
              row = buildRow(buildCell(age, instance.decompose(i), System.currentTimeMillis()));
  
              boolean result = filterTree.isSatisfiedBy(key, (Row) row, staticRow);
              assertTrue(result);
          }
  
          // multiple analyzed expressions in the Operation timestamp >= 10 AND age = 5
          node = new Operation.AndNode();
          node.add(new Operation.ExpressionNode(new SimpleExpression(timestamp, Operator.GTE, LongType.instance.decompose(10L))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.EQ, Int32Type.instance.decompose(5))));
  
          FilterTree filterTreeStrict = node.buildFilter(controller, true);
          FilterTree filterTreeNonStrict = node.buildFilter(controller, false);
  
          long startTime = System.currentTimeMillis();
          row = buildRow(buildCell(age, instance.decompose(6), startTime),
                         buildCell(timestamp, LongType.instance.decompose(11L), startTime + 1));
  
          assertFalse(filterTreeStrict.isSatisfiedBy(key, (Row) row, staticRow));
          assertTrue(filterTreeNonStrict.isSatisfiedBy(key, (Row) row, staticRow)); // matches on timestamp >= 10
  
          row = buildRow(buildCell(age, instance.decompose(5), startTime + 2),
                         buildCell(timestamp, LongType.instance.decompose(22L), startTime + 3));
  
          assertTrue(filterTreeStrict.isSatisfiedBy(key, (Row) row, staticRow));
          assertTrue(filterTreeNonStrict.isSatisfiedBy(key, (Row) row, staticRow));
  
          row = buildRow(buildCell(age, instance.decompose(5), startTime + 4),
                         buildCell(timestamp, LongType.instance.decompose(9L), startTime + 5));
  
          assertFalse(filterTreeStrict.isSatisfiedBy(key, (Row) row, staticRow));
          assertTrue(filterTreeNonStrict.isSatisfiedBy(key, (Row) row, staticRow)); // matches on age = 5
      }
  
      @Test
      public void testAnalyzeNotIndexedButDefinedColumn()
      {
          final ColumnMetadata firstName = getColumn(UTF8Type.instance.decompose("first_name"));
          final ColumnMetadata height = getColumn(UTF8Type.instance.decompose("height"));
  
          // first_name = 'a' AND height > 5
          Map<Expression.IndexOperator, Expression> expressions;
          expressions = convert(Operation.buildIndexExpressions(controller,
                                                                Arrays.asList(new SimpleExpression(firstName, Operator.EQ, UTF8Type.instance.decompose("a")),
                                                                     new SimpleExpression(height, Operator.GT, Int32Type.instance.decompose(5)))));
  
          assertEquals(2, expressions.size());
  
          expressions = convert(Operation.buildIndexExpressions(controller,
                                                                Arrays.asList(new SimpleExpression(firstName, Operator.EQ, UTF8Type.instance.decompose("a")),
                                                                     new SimpleExpression(height, Operator.GT, Int32Type.instance.decompose(0)),
                                                                     new SimpleExpression(height, Operator.EQ, Int32Type.instance.decompose(5)))));
  
          assertEquals(2, expressions.size());
  
          Expression rangeExpression = expressions.get(Expression.IndexOperator.RANGE);
  
          assertExpression(rangeExpression, Expression.IndexOperator.RANGE, Int32Type.instance.decompose(0), false, Int32Type.instance.decompose(5), true);
  
          expressions = convert(Operation.buildIndexExpressions(controller,
                                                                Arrays.asList(new SimpleExpression(firstName, Operator.EQ, UTF8Type.instance.decompose("a")),
                                                                              new SimpleExpression(height, Operator.GTE, Int32Type.instance.decompose(0)),
                                                                              new SimpleExpression(height, Operator.LT, Int32Type.instance.decompose(10)))));
  
          assertEquals(2, expressions.size());
  
          rangeExpression = expressions.get(Expression.IndexOperator.RANGE);
  
          assertExpression(rangeExpression, Expression.IndexOperator.RANGE, Int32Type.instance.decompose(0), true, Int32Type.instance.decompose(10), false);
      }
  
      @Test
      public void testSatisfiedByWithClustering()
      {
          ColumnMetadata location = getColumn(CLUSTERING_BACKEND, UTF8Type.instance.decompose("location"));
          ColumnMetadata age = getColumn(CLUSTERING_BACKEND, UTF8Type.instance.decompose("age"));
          ColumnMetadata height = getColumn(CLUSTERING_BACKEND, UTF8Type.instance.decompose("height"));
          ColumnMetadata score = getColumn(CLUSTERING_BACKEND, UTF8Type.instance.decompose("score"));
  
          DecoratedKey key = buildKey(CLUSTERING_BACKEND, "0");
          Row row = buildRow(Clustering.make(UTF8Type.instance.fromString("US"), Int32Type.instance.decompose(27)),
                                    buildCell(height, instance.decompose(182), System.currentTimeMillis()),
                                    buildCell(score, DoubleType.instance.decompose(1.0d), System.currentTimeMillis()));
          Row staticRow = buildRow(Clustering.STATIC_CLUSTERING);
  
          Operation.Node node = new Operation.AndNode();
          node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.EQ, Int32Type.instance.decompose(27))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(height, Operator.EQ, Int32Type.instance.decompose(182))));
  
          assertTrue(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
  
          node = new Operation.AndNode();
  
          node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.EQ, Int32Type.instance.decompose(28))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(height, Operator.EQ, Int32Type.instance.decompose(182))));
  
          assertFalse(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
  
          node = new Operation.AndNode();
          node.add(new Operation.ExpressionNode(new SimpleExpression(location, Operator.EQ, UTF8Type.instance.decompose("US"))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.GTE, Int32Type.instance.decompose(27))));
  
          assertTrue(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
  
          node = new Operation.AndNode();
          node.add(new Operation.ExpressionNode(new SimpleExpression(location, Operator.EQ, UTF8Type.instance.decompose("BY"))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.GTE, Int32Type.instance.decompose(28))));
  
          assertFalse(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
  
          node = new Operation.AndNode();
          node.add(new Operation.ExpressionNode(new SimpleExpression(location, Operator.EQ, UTF8Type.instance.decompose("US"))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.LTE, Int32Type.instance.decompose(27))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(height, Operator.GTE, Int32Type.instance.decompose(182))));
  
          assertTrue(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
  
          node = new Operation.AndNode();
          node.add(new Operation.ExpressionNode(new SimpleExpression(location, Operator.EQ, UTF8Type.instance.decompose("US"))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(height, Operator.GTE, Int32Type.instance.decompose(182))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(score, Operator.EQ, DoubleType.instance.decompose(1.0d))));
  
          assertTrue(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
  
          node = new Operation.AndNode();
          node.add(new Operation.ExpressionNode(new SimpleExpression(height, Operator.GTE, Int32Type.instance.decompose(182))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(score, Operator.EQ, DoubleType.instance.decompose(1.0d))));
  
          assertTrue(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
      }
  
      private Map<Expression.IndexOperator, Expression> convert(Operation.Expressions expressions)
      {
          Map<Expression.IndexOperator, Expression> converted = new EnumMap<>(Expression.IndexOperator.class);
          for (Expression expression : expressions.all())
          {
              Expression column = converted.get(expression.getIndexOperator());
              assert column == null; // sanity check
              converted.put(expression.getIndexOperator(), expression);
          }
  
          return converted;
      }
  
      @Test
      public void testSatisfiedByWithStatic()
      {
          final ColumnMetadata sensorType = getColumn(STATIC_BACKEND, UTF8Type.instance.decompose("sensor_type"));
          final ColumnMetadata value = getColumn(STATIC_BACKEND, UTF8Type.instance.decompose("value"));
  
          DecoratedKey key = buildKey(STATIC_BACKEND, 0);
          Row row = buildRow(Clustering.make(UTF8Type.instance.fromString("date"), LongType.instance.decompose(20160401L)),
                                    buildCell(value, DoubleType.instance.decompose(24.56), System.currentTimeMillis()));
          Row staticRow = buildRow(Clustering.STATIC_CLUSTERING,
                                   buildCell(sensorType, UTF8Type.instance.decompose("TEMPERATURE"), System.currentTimeMillis()));
  
          // sensor_type ='TEMPERATURE' AND value = 24.56
          Operation.Node node = new Operation.AndNode();
          node.add(new Operation.ExpressionNode(new SimpleExpression(sensorType, Operator.EQ, UTF8Type.instance.decompose("TEMPERATURE"))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(value, Operator.EQ, DoubleType.instance.decompose(24.56))));
  
          assertTrue(node.buildFilter(controllerStatic, true).isSatisfiedBy(key, row, staticRow));
  
          // sensor_type ='TEMPERATURE' AND value = 30
          node = new Operation.AndNode();
          node.add(new Operation.ExpressionNode(new SimpleExpression(sensorType, Operator.EQ, UTF8Type.instance.decompose("TEMPERATURE"))));
          node.add(new Operation.ExpressionNode(new SimpleExpression(value, Operator.EQ, DoubleType.instance.decompose(30.00))));
  
          assertFalse(node.buildFilter(controllerStatic, true).isSatisfiedBy(key, row, staticRow));
      }
  
      public static TableMetadata.Builder skinnySAITableMetadata(String keyspace, String table)
      {
          TableMetadata.Builder builder =
          TableMetadata.builder(keyspace, table)
                       .addPartitionKeyColumn("id", UTF8Type.instance)
                       .addRegularColumn("first_name", UTF8Type.instance)
                       .addRegularColumn("last_name", UTF8Type.instance)
                       .addRegularColumn("age", Int32Type.instance)
                       .addRegularColumn("height", Int32Type.instance)
                       .addRegularColumn("timestamp", LongType.instance)
                       .addRegularColumn("address", UTF8Type.instance)
                       .addRegularColumn("score", DoubleType.instance);
  
          Indexes.Builder indexes = Indexes.builder();
          addIndex(indexes, table, "first_name");
          addIndex(indexes, table, "last_name");
          addIndex(indexes, table, "age");
          addIndex(indexes, table, "timestamp");
          addIndex(indexes, table, "address");
          addIndex(indexes, table, "score");
  
          return builder.indexes(indexes.build());
      }
  
      public static TableMetadata.Builder clusteringSAITableMetadata(String keyspace, String table)
      {
          return clusteringSAITableMetadata(keyspace, table, "location", "age", "height", "score");
      }
  
      public static TableMetadata.Builder clusteringSAITableMetadata(String keyspace, String table, String...indexedColumns)
      {
          Indexes.Builder indexes = Indexes.builder();
          for (String indexedColumn : indexedColumns)
          {
              addIndex(indexes, table, indexedColumn);
          }
  
          return TableMetadata.builder(keyspace, table)
                              .addPartitionKeyColumn("name", UTF8Type.instance)
                              .addClusteringColumn("location", UTF8Type.instance)
                              .addClusteringColumn("age", Int32Type.instance)
                              .addRegularColumn("height", Int32Type.instance)
                              .addRegularColumn("score", DoubleType.instance)
                              .indexes(indexes.build());
      }
  
      public static TableMetadata.Builder staticSAITableMetadata(String keyspace, String table)
      {
          TableMetadata.Builder builder =
          TableMetadata.builder(keyspace, table)
                       .addPartitionKeyColumn("sensor_id", Int32Type.instance)
                       .addStaticColumn("sensor_type", UTF8Type.instance)
                       .addClusteringColumn("date", LongType.instance)
                       .addRegularColumn("value", DoubleType.instance)
                       .addRegularColumn("variance", Int32Type.instance);
  
          Indexes.Builder indexes = Indexes.builder();
  
          addIndex(indexes, table, "sensor_type");
          addIndex(indexes, table, "value");
          addIndex(indexes, table, "variance");
  
          return builder.indexes(indexes.build());
      }
  
      private void assertExpression(Expression expression, Expression.IndexOperator indexOperator, ByteBuffer lower,
                                    boolean lowerInclusive, ByteBuffer upper, boolean upperInclusive)
      {
          assertEquals(indexOperator, expression.getIndexOperator());
          assertEquals(lower, expression.lower().value.raw);
          assertEquals(lowerInclusive, expression.lower().inclusive);
          assertEquals(upper, expression.upper().value.raw);
          assertEquals(upperInclusive, expression.upper().inclusive);
      }
  
      private static void addIndex(Indexes.Builder indexes, String table, String column)
      {
          String indexName = table + '_' + column;
          indexes.add(IndexMetadata.fromSchemaMetadata(indexName, IndexMetadata.Kind.CUSTOM, new HashMap<String, String>()
          {{
              put(IndexTarget.CUSTOM_INDEX_OPTION_NAME, StorageAttachedIndex.class.getName());
              put(IndexTarget.TARGET_OPTION_NAME, column);
          }}));
      }
  
      private static DecoratedKey buildKey(Object... key)
      {
          return buildKey(BACKEND, key);
      }
  
      private static DecoratedKey buildKey(ColumnFamilyStore cfs, Object... key)
      {
          AbstractType<?> type = cfs.metadata().partitionKeyType;
          ByteBuffer decomposed;
          if(type instanceof CompositeType)
          {
              Preconditions.checkArgument(key.length == type.subTypes().size());
              decomposed = ((CompositeType) type).decompose(key);
          }
          else
          {
              Preconditions.checkArgument(key.length == 1);
              decomposed = ((AbstractType) type).decompose(key[0]);
          }
          return Murmur3Partitioner.instance.decorateKey(decomposed);
      }
  
      private static Unfiltered buildRow(Cell<?>... cells)
      {
          return buildRow(Clustering.EMPTY, cells);
      }
  
      private static Row buildRow(Clustering<?> clustering, Cell<?>... cells)
      {
          Row.Builder rowBuilder = BTreeRow.sortedBuilder();
          rowBuilder.newRow(clustering);
          for (Cell<?> c : cells)
              rowBuilder.addCell(c);
  
          return rowBuilder.build();
      }
  
      private static Cell<?> buildCell(ColumnMetadata column, ByteBuffer value, long timestamp)
      {
          return BufferCell.live(column, timestamp, value);
      }
  
      private static ColumnMetadata getColumn(ByteBuffer name)
      {
          return getColumn(BACKEND, name);
      }
  
      private static ColumnMetadata getColumn(ColumnFamilyStore cfs, ByteBuffer name)
      {
          return cfs.metadata().getColumn(name);
      }
  
      private static class SimpleExpression extends RowFilter.Expression
      {
          SimpleExpression(ColumnMetadata column, Operator operator, ByteBuffer value)
          {
              super(column, operator, value);
          }
  
          @Override
          public Kind kind()
          {
              return Kind.SIMPLE;
          }
  
          @Override
          public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)
          {
              throw new UnsupportedOperationException();
          }
  
          @Override
          protected String toString(boolean cql)
          {
              AbstractType<?> type = column.type;
              switch (operator)
              {
                  case CONTAINS:
                      assert type instanceof CollectionType;
                      CollectionType<?> ct = (CollectionType<?>)type;
                      type = ct.kind == CollectionType.Kind.SET ? ct.nameComparator() : ct.valueComparator();
                      break;
                  case CONTAINS_KEY:
                      assert type instanceof MapType;
                      type = ((MapType<?, ?>)type).nameComparator();
                      break;
                  case IN:
                      type = ListType.getInstance(type, false);
                      break;
                  default:
                      break;
              }
              return cql
                     ? String.format("%s %s %s", column.name.toCQLString(), operator, type.toCQLString(value) )
                     : String.format("%s %s %s", column.name.toString(), operator, type.getString(value));
          }
      }
  }
  