cassandra-commits mailing list archives

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


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

Branch: refs/heads/cassandra-3.1
Commit: ae64cc0da4b83289a88478c48ae73f608990a0ec
Parents: afe3fe3 8385bb6
Author: Yuki Morishita <yukim@apache.org>
Authored: Tue Nov 17 18:07:42 2015 -0600
Committer: Yuki Morishita <yukim@apache.org>
Committed: Tue Nov 17 18:07:42 2015 -0600

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


http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae64cc0d/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 4cb9275,008d4d4..572afc2
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,17 -1,5 +1,18 @@@
 -2.1.12
 +2.2.4
 + * Don't do anticompaction after subrange repair (CASSANDRA-10422)
 + * 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)

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

http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae64cc0d/src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java
index e9c99fe,34d9a01..e9a727f
--- a/src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java
+++ b/src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java
@@@ -44,14 -43,16 +44,20 @@@ public class FileMessageHeade
      public final int sequenceNumber;
      /** SSTable version */
      public final String version;
 +
 +    /** SSTable format **/
 +    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 FileMessageHeader(UUID cfId,
                               int sequenceNumber,
@@@ -70,10 -68,33 +76,38 @@@
          this.estimatedKeys = estimatedKeys;
          this.sections = sections;
          this.compressionInfo = compressionInfo;
+         this.compressionMetadata = null;
+         this.repairedAt = repairedAt;
++        this.sstableLevel = sstableLevel;
+     }
+ 
+     public FileMessageHeader(UUID cfId,
+                              int sequenceNumber,
+                              String version,
++                             SSTableFormat.Type format,
+                              long estimatedKeys,
+                              List<Pair<Long, Long>> sections,
+                              CompressionMetadata compressionMetadata,
 -                             long repairedAt)
++                             long repairedAt,
++                             int sstableLevel)
+     {
+         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;
      }
  
+     public boolean isCompressed()
+     {
+         return compressionInfo != null || compressionMetadata != null;
+     }
+ 
      /**
       * @return total file size to transfer in bytes
       */
@@@ -101,12 -126,10 +139,12 @@@
          sb.append("cfId: ").append(cfId);
          sb.append(", #").append(sequenceNumber);
          sb.append(", version: ").append(version);
 +        sb.append(", format: ").append(format);
          sb.append(", estimated keys: ").append(estimatedKeys);
          sb.append(", transfer size: ").append(size());
-         sb.append(", compressed?: ").append(compressionInfo != null);
+         sb.append(", compressed?: ").append(isCompressed());
          sb.append(", repairedAt: ").append(repairedAt);
 +        sb.append(", level: ").append(sstableLevel);
          sb.append(')');
          return sb.toString();
      }
@@@ -150,9 -166,13 +188,14 @@@
                  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);
+             return compressionInfo;
          }
  
          public FileMessageHeader deserialize(DataInput in, int version) throws IOException

http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae64cc0d/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae64cc0d/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
index 5b34bd8,71902e1..c8175ea
--- a/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
+++ b/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
@@@ -65,35 -66,26 +65,29 @@@ public class OutgoingFileMessage extend
          this.header = new FileMessageHeader(sstable.metadata.cfId,
                                              sequenceNumber,
                                              sstable.descriptor.version.toString(),
 +                                            sstable.descriptor.formatType,
                                              estimatedKeys,
                                              sections,
-                                             compressionInfo,
+                                             sstable.compression ? sstable.getCompressionMetadata()
: null,
 -                                            repairedAt);
 +                                            repairedAt,
 +                                            keepSSTableLevel ? sstable.getSSTableLevel()
: 0);
      }
  
 -    public synchronized void serialize(DataOutputStreamAndChannel out, int version, StreamSession
session) throws IOException
 +    public synchronized void serialize(DataOutputStreamPlus out, int version, StreamSession
session) throws IOException
      {
          if (completed)
          {
              return;
          }
  
-         FileMessageHeader.serializer.serialize(header, out, version);
+         CompressionInfo compressionInfo = FileMessageHeader.serializer.serialize(header,
out, version);
  
          final SSTableReader reader = ref.get();
-         StreamWriter writer = header.compressionInfo == null ?
+         StreamWriter writer = compressionInfo == null ?
                                        new StreamWriter(reader, header.sections, session)
:
 -                                      new CompressedStreamWriter(reader, header.sections,
compressionInfo, session);
 -        writer.write(out.getChannel());
 +                                      new CompressedStreamWriter(reader, header.sections,
-                                                                  header.compressionInfo,
session);
++                                                                 compressionInfo, session);
 +        writer.write(out);
      }
  
      public synchronized void complete()

http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae64cc0d/test/unit/org/apache/cassandra/streaming/compress/CompressedInputStreamTest.java
----------------------------------------------------------------------


Mime
View raw message