cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yu...@apache.org
Subject [11/15] cassandra git commit: Merge branch 'cassandra-2.2' into cassandra-3.0
Date Wed, 18 Nov 2015 01:32:17 GMT
Merge branch 'cassandra-2.2' into cassandra-3.0


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5cc02dd9
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5cc02dd9
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5cc02dd9

Branch: refs/heads/cassandra-3.1
Commit: 5cc02dd9ac4e9f081540586769e82e1544532e1e
Parents: 63a9f9b ae64cc0
Author: Yuki Morishita <yukim@apache.org>
Authored: Tue Nov 17 19:31:38 2015 -0600
Committer: Yuki Morishita <yukim@apache.org>
Committed: Tue Nov 17 19:31:38 2015 -0600

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../io/compress/CompressionMetadata.java        | 30 +++++++++++
 .../streaming/messages/FileMessageHeader.java   | 55 ++++++++++++++++++--
 .../streaming/messages/IncomingFileMessage.java |  2 +-
 .../streaming/messages/OutgoingFileMessage.java | 14 ++---
 .../compression/CompressedInputStreamTest.java  |  7 +++
 6 files changed, 93 insertions(+), 16 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/5cc02dd9/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index b68cf0d,572afc2..4510462
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -9,7 -3,17 +9,8 @@@ Merged from 2.2
   * Fix SimpleDateType type compatibility (CASSANDRA-10027)
   * (Hadoop) fix splits calculation (CASSANDRA-10640)
   * (Hadoop) ensure that Cluster instances are always closed (CASSANDRA-10058)
 - * (cqlsh) show partial trace if incomplete after max_trace_wait (CASSANDRA-7645)
 - * Use most up-to-date version of schema for system tables (CASSANDRA-10652)
 - * Deprecate memory_allocator in cassandra.yaml (CASSANDRA-10581,10628)
 - * Expose phi values from failure detector via JMX and tweak debug
 -   and trace logging (CASSANDRA-9526)
 - * Fix RangeNamesQueryPager (CASSANDRA-10509)
 - * Deprecate Pig support (CASSANDRA-10542)
 - * Reduce contention getting instances of CompositeType (CASSANDRA-10433)
  Merged from 2.1:
+  * Create compression chunk for sending file only (CASSANDRA-10680)
 - * Make buffered read size configurable (CASSANDRA-10249)
   * Forbid compact clustering column type changes in ALTER TABLE (CASSANDRA-8879)
   * Reject incremental repair with subrange repair (CASSANDRA-10422)
   * Add a nodetool command to refresh size_estimates (CASSANDRA-9579)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5cc02dd9/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5cc02dd9/src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java
index 123b983,e9a727f..e1e13b7
--- a/src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java
+++ b/src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java
@@@ -52,10 -49,15 +52,16 @@@ public class FileMessageHeade
      public final SSTableFormat.Type format;
      public final long estimatedKeys;
      public final List<Pair<Long, Long>> sections;
+     /**
+      * Compression info for SSTable to send. Can be null if SSTable is not compressed.
+      * On sender, this field is always null to avoid holding large number of Chunks.
+      * Use compressionMetadata instead.
+      */
      public final CompressionInfo compressionInfo;
+     private final CompressionMetadata compressionMetadata;
      public final long repairedAt;
      public final int sstableLevel;
 +    public final SerializationHeader.Component header;
  
      public FileMessageHeader(UUID cfId,
                               int sequenceNumber,
@@@ -75,11 -76,38 +81,41 @@@
          this.estimatedKeys = estimatedKeys;
          this.sections = sections;
          this.compressionInfo = compressionInfo;
+         this.compressionMetadata = null;
+         this.repairedAt = repairedAt;
+         this.sstableLevel = sstableLevel;
++        this.header = header;
+     }
+ 
+     public FileMessageHeader(UUID cfId,
+                              int sequenceNumber,
 -                             String version,
++                             Version version,
+                              SSTableFormat.Type format,
+                              long estimatedKeys,
+                              List<Pair<Long, Long>> sections,
+                              CompressionMetadata compressionMetadata,
+                              long repairedAt,
 -                             int sstableLevel)
++                             int sstableLevel,
++                             SerializationHeader.Component header)
+     {
+         this.cfId = cfId;
+         this.sequenceNumber = sequenceNumber;
+         this.version = version;
+         this.format = format;
+         this.estimatedKeys = estimatedKeys;
+         this.sections = sections;
+         this.compressionInfo = null;
+         this.compressionMetadata = compressionMetadata;
          this.repairedAt = repairedAt;
          this.sstableLevel = sstableLevel;
 +        this.header = header;
      }
  
+     public boolean isCompressed()
+     {
+         return compressionInfo != null || compressionMetadata != null;
+     }
+ 
      /**
       * @return total file size to transfer in bytes
       */
@@@ -156,15 -188,17 +196,20 @@@
                  out.writeLong(section.left);
                  out.writeLong(section.right);
              }
-             CompressionInfo.serializer.serialize(header.compressionInfo, out, version);
+             // construct CompressionInfo here to avoid holding large number of Chunks on
heap.
+             CompressionInfo compressionInfo = null;
+             if (header.compressionMetadata != null)
+                 compressionInfo = new CompressionInfo(header.compressionMetadata.getChunksForSections(header.sections),
header.compressionMetadata.parameters);
+             CompressionInfo.serializer.serialize(compressionInfo, out, version);
              out.writeLong(header.repairedAt);
              out.writeInt(header.sstableLevel);
 +
 +            if (version >= StreamMessage.VERSION_30)
 +                SerializationHeader.serializer.serialize(header.version, header.header,
out);
+             return compressionInfo;
          }
  
 -        public FileMessageHeader deserialize(DataInput in, int version) throws IOException
 +        public FileMessageHeader deserialize(DataInputPlus in, int version) throws IOException
          {
              UUID cfId = UUIDSerializer.serializer.deserialize(in, MessagingService.current_version);
              int sequenceNumber = in.readInt();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5cc02dd9/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java
index 19f9e12,31ab2a8..d881d43
--- a/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java
+++ b/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java
@@@ -40,9 -39,9 +40,9 @@@ public class IncomingFileMessage extend
          @SuppressWarnings("resource")
          public IncomingFileMessage deserialize(ReadableByteChannel in, int version, StreamSession
session) throws IOException
          {
 -            DataInputStream input = new DataInputStream(Channels.newInputStream(in));
 +            DataInputPlus input = new DataInputStreamPlus(Channels.newInputStream(in));
              FileMessageHeader header = FileMessageHeader.serializer.deserialize(input, version);
-             StreamReader reader = header.compressionInfo == null ? new StreamReader(header,
session)
+             StreamReader reader = !header.isCompressed() ? new StreamReader(header, session)
                      : new CompressedStreamReader(header, session);
  
              try

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5cc02dd9/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
index 82e6620,c8175ea..f10b42e
--- a/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
+++ b/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
@@@ -62,22 -62,15 +62,16 @@@ public class OutgoingFileMessage extend
  
          SSTableReader sstable = ref.get();
          filename = sstable.getFilename();
-         CompressionInfo compressionInfo = null;
-         if (sstable.compression)
-         {
-             CompressionMetadata meta = sstable.getCompressionMetadata();
-             compressionInfo = new CompressionInfo(meta.getChunksForSections(sections), meta.parameters);
-         }
          this.header = new FileMessageHeader(sstable.metadata.cfId,
                                              sequenceNumber,
 -                                            sstable.descriptor.version.toString(),
 +                                            sstable.descriptor.version,
                                              sstable.descriptor.formatType,
                                              estimatedKeys,
                                              sections,
-                                             compressionInfo,
+                                             sstable.compression ? sstable.getCompressionMetadata()
: null,
                                              repairedAt,
 -                                            keepSSTableLevel ? sstable.getSSTableLevel()
: 0);
 +                                            keepSSTableLevel ? sstable.getSSTableLevel()
: 0,
 +                                            sstable.header == null ? null : sstable.header.toComponent());
      }
  
      public synchronized void serialize(DataOutputStreamPlus out, int version, StreamSession
session) throws IOException

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5cc02dd9/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
index db05a3e,0000000..5646592
mode 100644,000000..100644
--- a/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
+++ b/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
@@@ -1,122 -1,0 +1,129 @@@
 +/*
 + * 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.streaming.compression;
 +
 +import java.io.*;
 +import java.util.*;
 +
 +import org.junit.Test;
 +import org.apache.cassandra.db.ClusteringComparator;
 +import org.apache.cassandra.db.marshal.BytesType;
 +import org.apache.cassandra.io.compress.CompressedSequentialWriter;
 +import org.apache.cassandra.io.compress.CompressionMetadata;
 +import org.apache.cassandra.schema.CompressionParams;
 +import org.apache.cassandra.io.sstable.Component;
 +import org.apache.cassandra.io.sstable.Descriptor;
 +import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
 +import org.apache.cassandra.streaming.compress.CompressedInputStream;
 +import org.apache.cassandra.streaming.compress.CompressionInfo;
 +import org.apache.cassandra.utils.ChecksumType;
 +import org.apache.cassandra.utils.Pair;
 +
++import static org.junit.Assert.assertEquals;
++
 +/**
 + */
 +public class CompressedInputStreamTest
 +{
 +    @Test
 +    public void testCompressedRead() throws Exception
 +    {
 +        testCompressedReadWith(new long[]{0L}, false);
 +        testCompressedReadWith(new long[]{1L}, false);
 +        testCompressedReadWith(new long[]{100L}, false);
 +
 +        testCompressedReadWith(new long[]{1L, 122L, 123L, 124L, 456L}, false);
 +    }
 +
 +    @Test(expected = EOFException.class)
 +    public void testTruncatedRead() throws Exception
 +    {
 +        testCompressedReadWith(new long[]{1L, 122L, 123L, 124L, 456L}, true);
 +    }
 +    /**
 +     * @param valuesToCheck array of longs of range(0-999)
 +     * @throws Exception
 +     */
 +    private void testCompressedReadWith(long[] valuesToCheck, boolean testTruncate) throws
Exception
 +    {
 +        assert valuesToCheck != null && valuesToCheck.length > 0;
 +
 +        // write compressed data file of longs
 +        File tmp = new File(File.createTempFile("cassandra", "unittest").getParent(), "ks-cf-ib-1-Data.db");
 +        Descriptor desc = Descriptor.fromFilename(tmp.getAbsolutePath());
 +        MetadataCollector collector = new MetadataCollector(new ClusteringComparator(BytesType.instance));
 +        CompressionParams param = CompressionParams.snappy(32);
 +        CompressedSequentialWriter writer = new CompressedSequentialWriter(tmp, desc.filenameFor(Component.COMPRESSION_INFO),
param, collector);
 +        Map<Long, Long> index = new HashMap<Long, Long>();
 +        for (long l = 0L; l < 1000; l++)
 +        {
 +            index.put(l, writer.position());
 +            writer.writeLong(l);
 +        }
 +        writer.finish();
 +
 +        CompressionMetadata comp = CompressionMetadata.create(tmp.getAbsolutePath());
 +        List<Pair<Long, Long>> sections = new ArrayList<Pair<Long, Long>>();
 +        for (long l : valuesToCheck)
 +        {
 +            long position = index.get(l);
 +            sections.add(Pair.create(position, position + 8));
 +        }
 +        CompressionMetadata.Chunk[] chunks = comp.getChunksForSections(sections);
++        long totalSize = comp.getTotalSizeForSections(sections);
++        long expectedSize = 0;
++        for (CompressionMetadata.Chunk c : chunks)
++            expectedSize += c.length + 4;
++        assertEquals(expectedSize, totalSize);
 +
 +        // buffer up only relevant parts of file
 +        int size = 0;
 +        for (CompressionMetadata.Chunk c : chunks)
 +            size += (c.length + 4); // 4bytes CRC
 +        byte[] toRead = new byte[size];
 +
 +        RandomAccessFile f = new RandomAccessFile(tmp, "r");
 +        int pos = 0;
 +        for (CompressionMetadata.Chunk c : chunks)
 +        {
 +            f.seek(c.offset);
 +            pos += f.read(toRead, pos, c.length + 4);
 +        }
 +        f.close();
 +
 +        if (testTruncate)
 +        {
 +            byte [] actuallyRead = new byte[50];
 +            System.arraycopy(toRead, 0, actuallyRead, 0, 50);
 +            toRead = actuallyRead;
 +        }
 +
 +        // read buffer using CompressedInputStream
 +        CompressionInfo info = new CompressionInfo(chunks, param);
 +        CompressedInputStream input = new CompressedInputStream(new ByteArrayInputStream(toRead),
info,
 +                                                                ChecksumType.CRC32, () ->
1.0);
 +        DataInputStream in = new DataInputStream(input);
 +
 +        for (int i = 0; i < sections.size(); i++)
 +        {
 +            input.position(sections.get(i).left);
 +            long readValue = in.readLong();
 +            assert readValue == valuesToCheck[i] : "expected " + valuesToCheck[i] + " but
was " + readValue;
 +        }
 +    }
 +}


Mime
View raw message