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.db;
  
  import static org.junit.Assert.*;
  
  import java.io.*;
  
  import com.google.common.base.Predicate;
  
  import org.junit.BeforeClass;
  import org.junit.Test;
  import org.apache.cassandra.SchemaLoader;
  import org.apache.cassandra.Util;
  import org.apache.cassandra.schema.ColumnMetadata;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.db.commitlog.CommitLogTestReplayer;
  import org.apache.cassandra.db.partitions.PartitionUpdate;
  import org.apache.cassandra.db.rows.Row;
  import org.apache.cassandra.db.marshal.AsciiType;
  import org.apache.cassandra.db.marshal.BytesType;
  import org.apache.cassandra.db.partitions.FilteredPartition;
  import org.apache.cassandra.exceptions.ConfigurationException;
  import org.apache.cassandra.io.IVersionedSerializer;
  import org.apache.cassandra.io.util.DataInputBuffer;
  import org.apache.cassandra.io.util.DataInputPlus;
  import org.apache.cassandra.io.util.DataOutputBuffer;
  import org.apache.cassandra.net.MessagingService;
  import org.apache.cassandra.schema.KeyspaceParams;
  import org.apache.cassandra.utils.ByteBufferUtil;
  
  public class ReadMessageTest
  {
      private static final String KEYSPACE1 = "ReadMessageTest1";
      private static final String KEYSPACENOCOMMIT = "ReadMessageTest_NoCommit";
      private static final String CF = "Standard1";
      private static final String CF_FOR_READ_TEST = "Standard2";
      private static final String CF_FOR_COMMIT_TEST = "Standard3";
  
      @BeforeClass
      public static void defineSchema() throws ConfigurationException
      {
          DatabaseDescriptor.daemonInitialization();
  
          TableMetadata.Builder cfForReadMetadata =
              TableMetadata.builder(KEYSPACE1, CF_FOR_READ_TEST)
                           .addPartitionKeyColumn("key", BytesType.instance)
                           .addClusteringColumn("col1", AsciiType.instance)
                           .addClusteringColumn("col2", AsciiType.instance)
                           .addRegularColumn("a", AsciiType.instance)
                           .addRegularColumn("b", AsciiType.instance);
  
          TableMetadata.Builder cfForCommitMetadata1 =
              TableMetadata.builder(KEYSPACE1, CF_FOR_COMMIT_TEST)
                           .addPartitionKeyColumn("key", BytesType.instance)
                           .addClusteringColumn("name", AsciiType.instance)
                           .addRegularColumn("commit1", AsciiType.instance);
  
          TableMetadata.Builder cfForCommitMetadata2 =
              TableMetadata.builder(KEYSPACENOCOMMIT, CF_FOR_COMMIT_TEST)
                           .addPartitionKeyColumn("key", BytesType.instance)
                           .addClusteringColumn("name", AsciiType.instance)
                           .addRegularColumn("commit2", AsciiType.instance);
  
          SchemaLoader.prepareServer();
  
          SchemaLoader.createKeyspace(KEYSPACE1,
                                      KeyspaceParams.simple(1),
                                      SchemaLoader.standardCFMD(KEYSPACE1, CF),
                                      cfForReadMetadata,
                                      cfForCommitMetadata1);
  
          SchemaLoader.createKeyspace(KEYSPACENOCOMMIT,
                                      KeyspaceParams.simpleTransient(1),
                                      SchemaLoader.standardCFMD(KEYSPACENOCOMMIT, CF),
                                      cfForCommitMetadata2);
      }
  
      @Test
      public void testMakeReadMessage() throws IOException
      {
          ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_FOR_READ_TEST);
          ReadCommand rm, rm2;
  
          rm = Util.cmd(cfs, Util.dk("key1"))
                   .includeRow("col1", "col2")
                   .build();
          rm2 = serializeAndDeserializeReadMessage(rm);
          assertEquals(rm.toString(), rm2.toString());
  
          rm = Util.cmd(cfs, Util.dk("key1"))
                   .includeRow("col1", "col2")
                   .reverse()
                   .build();
          rm2 = serializeAndDeserializeReadMessage(rm);
          assertEquals(rm.toString(), rm2.toString());
  
          rm = Util.cmd(cfs)
                   .build();
          rm2 = serializeAndDeserializeReadMessage(rm);
          assertEquals(rm.toString(), rm2.toString());
  
          rm = Util.cmd(cfs)
                   .fromKeyIncl(ByteBufferUtil.bytes("key1"))
                   .toKeyIncl(ByteBufferUtil.bytes("key2"))
                   .build();
          rm2 = serializeAndDeserializeReadMessage(rm);
          assertEquals(rm.toString(), rm2.toString());
  
          rm = Util.cmd(cfs)
                   .columns("a")
                   .build();
          rm2 = serializeAndDeserializeReadMessage(rm);
          assertEquals(rm.toString(), rm2.toString());
  
          rm = Util.cmd(cfs)
                   .includeRow("col1", "col2")
                   .columns("a")
                   .build();
          rm2 = serializeAndDeserializeReadMessage(rm);
          assertEquals(rm.toString(), rm2.toString());
  
          rm = Util.cmd(cfs)
                   .fromKeyIncl(ByteBufferUtil.bytes("key1"))
                   .includeRow("col1", "col2")
                   .columns("a")
                   .build();
          rm2 = serializeAndDeserializeReadMessage(rm);
          assertEquals(rm.toString(), rm2.toString());
      }
  
      private ReadCommand serializeAndDeserializeReadMessage(ReadCommand rm) throws IOException
      {
          IVersionedSerializer<ReadCommand> rms = ReadCommand.serializer;
          DataOutputBuffer out = new DataOutputBuffer();
  
          rms.serialize(rm, out, MessagingService.current_version);
  
          DataInputPlus dis = new DataInputBuffer(out.getData());
          return rms.deserialize(dis, MessagingService.current_version);
      }
  
  
      @Test
      public void testGetColumn()
      {
          ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF);
  
          new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key1"))
                  .clustering("Column1")
                  .add("val", ByteBufferUtil.bytes("abcd"))
                  .build()
                  .apply();
  
          ColumnMetadata col = cfs.metadata().getColumn(ByteBufferUtil.bytes("val"));
          int found = 0;
          for (FilteredPartition partition : Util.getAll(Util.cmd(cfs).build()))
          {
              for (Row r : partition)
              {
                  if (r.getCell(col).value().equals(ByteBufferUtil.bytes("abcd")))
                      ++found;
              }
          }
          assertEquals(1, found);
      }
  
      @Test
      public void testNoCommitLog() throws Exception
      {
          ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_FOR_COMMIT_TEST);
  
          ColumnFamilyStore cfsnocommit = Keyspace.open(KEYSPACENOCOMMIT).getColumnFamilyStore(CF_FOR_COMMIT_TEST);
  
          new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("row"))
                  .clustering("c")
                  .add("commit1", ByteBufferUtil.bytes("abcd"))
                  .build()
                  .apply();
  
          new RowUpdateBuilder(cfsnocommit.metadata(), 0, ByteBufferUtil.bytes("row"))
                  .clustering("c")
                  .add("commit2", ByteBufferUtil.bytes("abcd"))
                  .build()
                  .apply();
  
          Checker checker = new Checker(cfs.metadata().getColumn(ByteBufferUtil.bytes("commit1")),
                                        cfsnocommit.metadata().getColumn(ByteBufferUtil.bytes("commit2")));
  
          CommitLogTestReplayer replayer = new CommitLogTestReplayer(checker);
          replayer.examineCommitLog();
  
          assertTrue(checker.commitLogMessageFound);
          assertFalse(checker.noCommitLogMessageFound);
      }
  
      static class Checker implements Predicate<Mutation>
      {
          private final ColumnMetadata withCommit;
          private final ColumnMetadata withoutCommit;
  
          boolean commitLogMessageFound = false;
          boolean noCommitLogMessageFound = false;
  
          public Checker(ColumnMetadata withCommit, ColumnMetadata withoutCommit)
          {
              this.withCommit = withCommit;
              this.withoutCommit = withoutCommit;
          }
  
          public boolean apply(Mutation mutation)
          {
              for (PartitionUpdate upd : mutation.getPartitionUpdates())
              {
                  Row r = upd.getRow(Clustering.make(ByteBufferUtil.bytes("c")));
                  if (r != null)
                  {
                      if (r.getCell(withCommit) != null)
                          commitLogMessageFound = true;
                      if (r.getCell(withoutCommit) != null)
                          noCommitLogMessageFound = true;
                  }
              }
              return true;
          }
      }
  }
  