hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From zg...@apache.org
Subject hbase git commit: HBASE-20625 refactor some WALCellCodec related code
Date Thu, 14 Jun 2018 11:41:00 GMT
Repository: hbase
Updated Branches:
  refs/heads/master 9e9db3245 -> 0b28155d2


HBASE-20625 refactor some WALCellCodec related code

Signed-off-by: Guanghao Zhang <zghao@apache.org>


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

Branch: refs/heads/master
Commit: 0b28155d274910b4e667b949d51f78809a1eff0b
Parents: 9e9db32
Author: jingyuntian <tianjy1990@gmail.com>
Authored: Thu Jun 14 10:25:24 2018 +0800
Committer: Guanghao Zhang <zghao@apache.org>
Committed: Thu Jun 14 19:37:01 2018 +0800

----------------------------------------------------------------------
 .../hbase/protobuf/ReplicationProtbufUtil.java  | 61 ++++----------
 .../wal/AbstractProtobufLogWriter.java          |  3 +
 .../wal/AsyncProtobufLogWriter.java             |  1 -
 .../regionserver/wal/CompressionContext.java    | 54 ++++++------
 .../regionserver/wal/ProtobufLogReader.java     |  2 +
 .../regionserver/wal/ProtobufLogWriter.java     |  1 -
 .../hbase/regionserver/wal/ReaderBase.java      |  3 -
 .../wal/SecureProtobufLogReader.java            |  1 +
 .../hbase/regionserver/wal/WALCellCodec.java    | 87 ++++++++++++++------
 .../replication/ClusterMarkingEntryFilter.java  |  2 -
 .../java/org/apache/hadoop/hbase/wal/WAL.java   | 11 ---
 .../org/apache/hadoop/hbase/wal/WALKeyImpl.java | 60 +++++---------
 .../wal/FaultyProtobufLogReader.java            |  3 -
 13 files changed, 134 insertions(+), 155 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ReplicationProtbufUtil.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ReplicationProtbufUtil.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ReplicationProtbufUtil.java
index 81dd59e..157ad1b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ReplicationProtbufUtil.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ReplicationProtbufUtil.java
@@ -24,29 +24,25 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
-import java.util.NavigableMap;
-import java.util.UUID;
 
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellScanner;
-import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.PrivateCellUtil;
-import org.apache.hadoop.hbase.wal.WALKeyImpl;
+import org.apache.hadoop.hbase.regionserver.wal.WALCellCodec;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.hadoop.hbase.io.SizedCellScanner;
 import org.apache.hadoop.hbase.ipc.HBaseRpcController;
 import org.apache.hadoop.hbase.ipc.HBaseRpcControllerImpl;
 import org.apache.hadoop.hbase.wal.WALEdit;
-import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
-import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.hadoop.hbase.wal.WAL.Entry;
 
+import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
+
 @InterfaceAudience.Private
 public class ReplicationProtbufUtil {
   /**
@@ -81,7 +77,7 @@ public class ReplicationProtbufUtil {
    * found.
    */
   public static Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner>
-      buildReplicateWALEntryRequest(final Entry[] entries) {
+      buildReplicateWALEntryRequest(final Entry[] entries) throws IOException {
     return buildReplicateWALEntryRequest(entries, null, null, null, null);
   }
 
@@ -97,53 +93,30 @@ public class ReplicationProtbufUtil {
    */
   public static Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner>
       buildReplicateWALEntryRequest(final Entry[] entries, byte[] encodedRegionName,
-          String replicationClusterId, Path sourceBaseNamespaceDir, Path sourceHFileArchiveDir)
{
+          String replicationClusterId, Path sourceBaseNamespaceDir, Path sourceHFileArchiveDir)
+          throws IOException {
     // Accumulate all the Cells seen in here.
     List<List<? extends Cell>> allCells = new ArrayList<>(entries.length);
     int size = 0;
-    WALProtos.FamilyScope.Builder scopeBuilder = WALProtos.FamilyScope.newBuilder();
     AdminProtos.WALEntry.Builder entryBuilder = AdminProtos.WALEntry.newBuilder();
     AdminProtos.ReplicateWALEntryRequest.Builder builder =
       AdminProtos.ReplicateWALEntryRequest.newBuilder();
-    HBaseProtos.UUID.Builder uuidBuilder = HBaseProtos.UUID.newBuilder();
+
     for (Entry entry: entries) {
       entryBuilder.clear();
-      // TODO: this duplicates a lot in WALKeyImpl#getBuilder
-      WALProtos.WALKey.Builder keyBuilder = entryBuilder.getKeyBuilder();
-      WALKeyImpl key = entry.getKey();
-      keyBuilder.setEncodedRegionName(
-          UnsafeByteOperations.unsafeWrap(encodedRegionName == null
-            ? key.getEncodedRegionName()
-            : encodedRegionName));
-      keyBuilder.setTableName(UnsafeByteOperations.unsafeWrap(key.getTableName().getName()));
-      long sequenceId = key.getSequenceId();
-      keyBuilder.setLogSequenceNumber(sequenceId);
-      keyBuilder.setWriteTime(key.getWriteTime());
-      if (key.getNonce() != HConstants.NO_NONCE) {
-        keyBuilder.setNonce(key.getNonce());
+      WALProtos.WALKey.Builder keyBuilder;
+      try {
+        keyBuilder = entry.getKey().getBuilder(WALCellCodec.getNoneCompressor());
+      } catch (IOException e) {
+        throw new IOException(
+            "There should not throw exception since NoneCompressor do not throw any exceptions",
e);
       }
-      if (key.getNonceGroup() != HConstants.NO_NONCE) {
-        keyBuilder.setNonceGroup(key.getNonceGroup());
-      }
-      for(UUID clusterId : key.getClusterIds()) {
-        uuidBuilder.setLeastSigBits(clusterId.getLeastSignificantBits());
-        uuidBuilder.setMostSigBits(clusterId.getMostSignificantBits());
-        keyBuilder.addClusterIds(uuidBuilder.build());
-      }
-      if (key.getOrigLogSeqNum() > 0) {
-        keyBuilder.setOrigSequenceNumber(key.getOrigLogSeqNum());
+      if(encodedRegionName != null){
+        keyBuilder.setEncodedRegionName(
+            UnsafeByteOperations.unsafeWrap(encodedRegionName));
       }
+      entryBuilder.setKey(keyBuilder.build());
       WALEdit edit = entry.getEdit();
-      NavigableMap<byte[], Integer> scopes = key.getReplicationScopes();
-      if (scopes != null && !scopes.isEmpty()) {
-        for (Map.Entry<byte[], Integer> scope: scopes.entrySet()) {
-          scopeBuilder.setFamily(UnsafeByteOperations.unsafeWrap(scope.getKey()));
-          WALProtos.ScopeType scopeType =
-              WALProtos.ScopeType.valueOf(scope.getValue().intValue());
-          scopeBuilder.setScopeType(scopeType);
-          keyBuilder.addScopes(scopeBuilder.build());
-        }
-      }
       List<Cell> cells = edit.getCells();
       // Add up the size.  It is used later serializing out the kvs.
       for (Cell cell: cells) {

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractProtobufLogWriter.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractProtobufLogWriter.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractProtobufLogWriter.java
index 50ac101..ae084a4 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractProtobufLogWriter.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractProtobufLogWriter.java
@@ -183,6 +183,8 @@ public abstract class AbstractProtobufLogWriter {
     this.cellEncoder = codec.getEncoder(getOutputStreamForCellEncoder());
     if (doCompress) {
       this.compressor = codec.getByteStringCompressor();
+    } else {
+      this.compressor = WALCellCodec.getNoneCompressor();
     }
   }
 
@@ -198,6 +200,7 @@ public abstract class AbstractProtobufLogWriter {
       this.cellEncoder = codec.getEncoder(getOutputStreamForCellEncoder());
       // We do not support compression
       this.compressionContext = null;
+      this.compressor = WALCellCodec.getNoneCompressor();
     } else {
       initAfterHeader0(doCompress);
     }

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncProtobufLogWriter.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncProtobufLogWriter.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncProtobufLogWriter.java
index abdc24e..6368fb7 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncProtobufLogWriter.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncProtobufLogWriter.java
@@ -120,7 +120,6 @@ public class AsyncProtobufLogWriter extends AbstractProtobufLogWriter
   @Override
   public void append(Entry entry) {
     int buffered = output.buffered();
-    entry.setCompressionContext(compressionContext);
     try {
       entry.getKey().
         getBuilder(compressor).setFollowingKvCount(entry.getEdit().size()).build()

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java
index 22ac499..16866e1 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java
@@ -20,6 +20,8 @@ package org.apache.hadoop.hbase.regionserver.wal;
 
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
+import java.util.EnumMap;
+import java.util.Map;
 
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -35,12 +37,12 @@ public class CompressionContext {
   static final String ENABLE_WAL_TAGS_COMPRESSION =
       "hbase.regionserver.wal.tags.enablecompression";
 
-  // visible only for WALKey, until we move everything into o.a.h.h.wal
-  public final Dictionary regionDict;
-  public final Dictionary tableDict;
-  public final Dictionary familyDict;
-  final Dictionary qualifierDict;
-  final Dictionary rowDict;
+  public enum DictionaryIndex {
+    REGION, TABLE, FAMILY, QUALIFIER, ROW
+  }
+
+  private final Map<DictionaryIndex, Dictionary> dictionaries =
+      new EnumMap<>(DictionaryIndex.class);
   // Context used for compressing tags
   TagCompressionContext tagCompressionContext = null;
 
@@ -49,33 +51,35 @@ public class CompressionContext {
       InstantiationException, IllegalAccessException, InvocationTargetException {
     Constructor<? extends Dictionary> dictConstructor =
         dictType.getConstructor();
-    regionDict = dictConstructor.newInstance();
-    tableDict = dictConstructor.newInstance();
-    familyDict = dictConstructor.newInstance();
-    qualifierDict = dictConstructor.newInstance();
-    rowDict = dictConstructor.newInstance();
-    if (recoveredEdits) {
-      // This will never change
-      regionDict.init(1);
-      tableDict.init(1);
+    for (DictionaryIndex dictionaryIndex : DictionaryIndex.values()) {
+      Dictionary newDictionary = dictConstructor.newInstance();
+      dictionaries.put(dictionaryIndex, newDictionary);
+    }
+    if(recoveredEdits) {
+      getDictionary(DictionaryIndex.REGION).init(1);
+      getDictionary(DictionaryIndex.TABLE).init(1);
     } else {
-      regionDict.init(Short.MAX_VALUE);
-      tableDict.init(Short.MAX_VALUE);
+      getDictionary(DictionaryIndex.REGION).init(Short.MAX_VALUE);
+      getDictionary(DictionaryIndex.TABLE).init(Short.MAX_VALUE);
     }
-    rowDict.init(Short.MAX_VALUE);
-    familyDict.init(Byte.MAX_VALUE);
-    qualifierDict.init(Byte.MAX_VALUE);
+
+    getDictionary(DictionaryIndex.ROW).init(Short.MAX_VALUE);
+    getDictionary(DictionaryIndex.FAMILY).init(Byte.MAX_VALUE);
+    getDictionary(DictionaryIndex.QUALIFIER).init(Byte.MAX_VALUE);
+
     if (hasTagCompression) {
       tagCompressionContext = new TagCompressionContext(dictType, Short.MAX_VALUE);
     }
   }
 
+  public Dictionary getDictionary(Enum dictIndex) {
+    return dictionaries.get(dictIndex);
+  }
+
   void clear() {
-    regionDict.clear();
-    tableDict.clear();
-    familyDict.clear();
-    qualifierDict.clear();
-    rowDict.clear();
+    for(Dictionary dictionary : dictionaries.values()){
+      dictionary.clear();
+    }
     if (tagCompressionContext != null) {
       tagCompressionContext.clear();
     }

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogReader.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogReader.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogReader.java
index 5d8d8c0..83398bd 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogReader.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogReader.java
@@ -312,6 +312,8 @@ public class ProtobufLogReader extends ReaderBase {
     this.cellDecoder = codec.getDecoder(this.inputStream);
     if (this.hasCompression) {
       this.byteStringUncompressor = codec.getByteStringUncompressor();
+    } else {
+      this.byteStringUncompressor = WALCellCodec.getNoneUncompressor();
     }
   }
 

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogWriter.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogWriter.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogWriter.java
index 2852047..b4e2cbf 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogWriter.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogWriter.java
@@ -47,7 +47,6 @@ public class ProtobufLogWriter extends AbstractProtobufLogWriter
 
   @Override
   public void append(Entry entry) throws IOException {
-    entry.setCompressionContext(compressionContext);
     entry.getKey().getBuilder(compressor).
         setFollowingKvCount(entry.getEdit().size()).build().writeDelimitedTo(output);
     for (Cell cell : entry.getEdit().getCells()) {

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ReaderBase.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ReaderBase.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ReaderBase.java
index 4338f6d..27d40b2 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ReaderBase.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ReaderBase.java
@@ -92,9 +92,6 @@ public abstract class ReaderBase implements AbstractFSWALProvider.Reader
{
     if (e == null) {
       e = new Entry();
     }
-    if (compressionContext != null) {
-      e.setCompressionContext(compressionContext);
-    }
 
     boolean hasEntry = false;
     try {

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SecureProtobufLogReader.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SecureProtobufLogReader.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SecureProtobufLogReader.java
index b1f17ad..e43d140 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SecureProtobufLogReader.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SecureProtobufLogReader.java
@@ -141,6 +141,7 @@ public class SecureProtobufLogReader extends ProtobufLogReader {
       this.cellDecoder = codec.getDecoder(this.inputStream);
       // We do not support compression with WAL encryption
       this.compressionContext = null;
+      this.byteStringUncompressor = WALCellCodec.getNoneUncompressor();
       this.hasCompression = false;
     } else {
       super.initAfterCompression(cellCodecClsName);

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALCellCodec.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALCellCodec.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALCellCodec.java
index 2922a10..34d83f7 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALCellCodec.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALCellCodec.java
@@ -46,6 +46,7 @@ import org.apache.hadoop.hbase.util.ReflectionUtils;
 import org.apache.hadoop.io.IOUtils;
 
 import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
+import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
 
 
 /**
@@ -62,12 +63,6 @@ public class WALCellCodec implements Codec {
   public static final String WAL_CELL_CODEC_CLASS_KEY = "hbase.regionserver.wal.codec";
 
   protected final CompressionContext compression;
-  protected final ByteStringUncompressor statelessUncompressor = new ByteStringUncompressor()
{
-    @Override
-    public byte[] uncompress(ByteString data, Dictionary dict) throws IOException {
-      return WALCellCodec.uncompressByteString(data, dict);
-    }
-  };
 
   /**
    * <b>All subclasses must implement a no argument constructor</b>
@@ -132,17 +127,32 @@ public class WALCellCodec implements Codec {
   }
 
   public interface ByteStringCompressor {
-    ByteString compress(byte[] data, Dictionary dict) throws IOException;
+    ByteString compress(byte[] data, Enum dictIndex) throws IOException;
   }
 
   public interface ByteStringUncompressor {
-    byte[] uncompress(ByteString data, Dictionary dict) throws IOException;
+    byte[] uncompress(ByteString data, Enum dictIndex) throws IOException;
+  }
+
+  static class StatelessUncompressor implements ByteStringUncompressor {
+    CompressionContext compressionContext;
+
+    public StatelessUncompressor(CompressionContext compressionContext) {
+      this.compressionContext = compressionContext;
+    }
+
+    @Override
+    public byte[] uncompress(ByteString data, Enum dictIndex) throws IOException {
+      return WALCellCodec.uncompressByteString(data, compressionContext.getDictionary(dictIndex));
+    }
   }
 
-  // TODO: it sucks that compression context is in WAL.Entry. It'd be nice if it was here.
-  //       Dictionary could be gotten by enum; initially, based on enum, context would create
-  //       an array of dictionaries.
   static class BaosAndCompressor extends ByteArrayOutputStream implements ByteStringCompressor
{
+    private CompressionContext compressionContext;
+
+    public BaosAndCompressor(CompressionContext compressionContext) {
+      this.compressionContext = compressionContext;
+    }
     public ByteString toByteString() {
       // We need this copy to create the ByteString as the byte[] 'buf' is not immutable.
We reuse
       // them.
@@ -150,8 +160,8 @@ public class WALCellCodec implements Codec {
     }
 
     @Override
-    public ByteString compress(byte[] data, Dictionary dict) throws IOException {
-      writeCompressed(data, dict);
+    public ByteString compress(byte[] data, Enum dictIndex) throws IOException {
+      writeCompressed(data, dictIndex);
       // We need this copy to create the ByteString as the byte[] 'buf' is not immutable.
We reuse
       // them.
       ByteString result = ByteString.copyFrom(this.buf, 0, this.count);
@@ -159,7 +169,8 @@ public class WALCellCodec implements Codec {
       return result;
     }
 
-    private void writeCompressed(byte[] data, Dictionary dict) throws IOException {
+    private void writeCompressed(byte[] data, Enum dictIndex) throws IOException {
+      Dictionary dict = compressionContext.getDictionary(dictIndex);
       assert dict != null;
       short dictIdx = dict.findEntry(data, 0, data.length);
       if (dictIdx == Dictionary.NOT_IN_DICTIONARY) {
@@ -172,6 +183,22 @@ public class WALCellCodec implements Codec {
     }
   }
 
+  static class NoneCompressor implements ByteStringCompressor {
+
+    @Override
+    public ByteString compress(byte[] data, Enum dictIndex) {
+      return UnsafeByteOperations.unsafeWrap(data);
+    }
+  }
+
+  static class NoneUncompressor implements ByteStringUncompressor {
+
+    @Override
+    public byte[] uncompress(ByteString data, Enum dictIndex) {
+      return data.toByteArray();
+    }
+  }
+
   private static byte[] uncompressByteString(ByteString bs, Dictionary dict) throws IOException
{
     InputStream in = bs.newInput();
     byte status = (byte)in.read();
@@ -209,9 +236,12 @@ public class WALCellCodec implements Codec {
       // To support tags
       int tagsLength = cell.getTagsLength();
       StreamUtils.writeRawVInt32(out, tagsLength);
-      PrivateCellUtil.compressRow(out, cell, compression.rowDict);
-      PrivateCellUtil.compressFamily(out, cell, compression.familyDict);
-      PrivateCellUtil.compressQualifier(out, cell, compression.qualifierDict);
+      PrivateCellUtil.compressRow(out, cell,
+        compression.getDictionary(CompressionContext.DictionaryIndex.ROW));
+      PrivateCellUtil.compressFamily(out, cell,
+        compression.getDictionary(CompressionContext.DictionaryIndex.FAMILY));
+      PrivateCellUtil.compressQualifier(out, cell,
+        compression.getDictionary(CompressionContext.DictionaryIndex.QUALIFIER));
       // Write timestamp, type and value as uncompressed.
       StreamUtils.writeLong(out, cell.getTimestamp());
       out.write(cell.getTypeByte());
@@ -255,19 +285,22 @@ public class WALCellCodec implements Codec {
       pos = Bytes.putInt(backingArray, pos, vlength);
 
       // the row
-      int elemLen = readIntoArray(backingArray, pos + Bytes.SIZEOF_SHORT, compression.rowDict);
+      int elemLen = readIntoArray(backingArray, pos + Bytes.SIZEOF_SHORT,
+        compression.getDictionary(CompressionContext.DictionaryIndex.ROW));
       checkLength(elemLen, Short.MAX_VALUE);
       pos = Bytes.putShort(backingArray, pos, (short)elemLen);
       pos += elemLen;
 
       // family
-      elemLen = readIntoArray(backingArray, pos + Bytes.SIZEOF_BYTE, compression.familyDict);
+      elemLen = readIntoArray(backingArray, pos + Bytes.SIZEOF_BYTE,
+        compression.getDictionary(CompressionContext.DictionaryIndex.FAMILY));
       checkLength(elemLen, Byte.MAX_VALUE);
       pos = Bytes.putByte(backingArray, pos, (byte)elemLen);
       pos += elemLen;
 
       // qualifier
-      elemLen = readIntoArray(backingArray, pos, compression.qualifierDict);
+      elemLen = readIntoArray(backingArray, pos,
+        compression.getDictionary(CompressionContext.DictionaryIndex.QUALIFIER));
       pos += elemLen;
 
       // timestamp, type and value
@@ -354,12 +387,18 @@ public class WALCellCodec implements Codec {
   }
 
   public ByteStringCompressor getByteStringCompressor() {
-    // TODO: ideally this should also encapsulate compressionContext
-    return new BaosAndCompressor();
+    return new BaosAndCompressor(compression);
   }
 
   public ByteStringUncompressor getByteStringUncompressor() {
-    // TODO: ideally this should also encapsulate compressionContext
-    return this.statelessUncompressor;
+    return new StatelessUncompressor(compression);
+  }
+
+  public static ByteStringCompressor getNoneCompressor() {
+    return new NoneCompressor();
+  }
+
+  public static ByteStringUncompressor getNoneUncompressor() {
+    return new NoneUncompressor();
   }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ClusterMarkingEntryFilter.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ClusterMarkingEntryFilter.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ClusterMarkingEntryFilter.java
index 6dc5001..5f92bbf 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ClusterMarkingEntryFilter.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ClusterMarkingEntryFilter.java
@@ -60,8 +60,6 @@ public class ClusterMarkingEntryFilter implements WALEntryFilter {
       if (edit != null && !edit.isEmpty()) {
         // Mark that the current cluster has the change
         logKey.addClusterId(clusterId);
-        // We need to set the CC to null else it will be compressed when sent to the sink
-        entry.setCompressionContext(null);
         return entry;
       }
     }

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WAL.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WAL.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WAL.java
index 3c85737..cf367cd 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WAL.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WAL.java
@@ -23,7 +23,6 @@ import java.util.Map;
 import java.util.Set;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.client.RegionInfo;
-import org.apache.hadoop.hbase.regionserver.wal.CompressionContext;
 import org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException;
 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
 import org.apache.hadoop.hbase.regionserver.wal.WALCoprocessorHost;
@@ -278,16 +277,6 @@ public interface WAL extends Closeable, WALFileLengthProvider {
       return key;
     }
 
-    /**
-     * Set compression context for this entry.
-     *
-     * @param compressionContext
-     *          Compression context
-     */
-    public void setCompressionContext(CompressionContext compressionContext) {
-      key.setCompressionContext(compressionContext);
-    }
-
     @Override
     public String toString() {
       return this.key + "=" + this.edit;

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALKeyImpl.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALKeyImpl.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALKeyImpl.java
index 8828239..71cbaf4 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALKeyImpl.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALKeyImpl.java
@@ -37,7 +37,6 @@ import org.apache.yetus.audience.InterfaceAudience;
 
 import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
-import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
 
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;
@@ -117,8 +116,6 @@ public class WALKeyImpl implements WALKey {
    */
   private MultiVersionConcurrencyControl.WriteEntry writeEntry;
 
-  private CompressionContext compressionContext;
-
   public WALKeyImpl() {
     init(null, null, 0L, HConstants.LATEST_TIMESTAMP,
         new ArrayList<>(), HConstants.NO_NONCE, HConstants.NO_NONCE, null, null);
@@ -332,13 +329,6 @@ public class WALKeyImpl implements WALKey {
     this.sequenceId = sequenceId;
   }
 
-  /**
-   * @param compressionContext Compression context to use
-   */
-  public void setCompressionContext(CompressionContext compressionContext) {
-    this.compressionContext = compressionContext;
-  }
-
   /** @return encoded region name */
   @Override
   public byte [] getEncodedRegionName() {
@@ -517,18 +507,13 @@ public class WALKeyImpl implements WALKey {
     this.encodedRegionName = encodedRegionName;
   }
 
-  public WALProtos.WALKey.Builder getBuilder(
-      WALCellCodec.ByteStringCompressor compressor) throws IOException {
+  public WALProtos.WALKey.Builder getBuilder(WALCellCodec.ByteStringCompressor compressor)
+      throws IOException {
     WALProtos.WALKey.Builder builder = WALProtos.WALKey.newBuilder();
-    if (compressionContext == null) {
-      builder.setEncodedRegionName(UnsafeByteOperations.unsafeWrap(this.encodedRegionName));
-      builder.setTableName(UnsafeByteOperations.unsafeWrap(this.tablename.getName()));
-    } else {
-      builder.setEncodedRegionName(compressor.compress(this.encodedRegionName,
-          compressionContext.regionDict));
-      builder.setTableName(compressor.compress(this.tablename.getName(),
-          compressionContext.tableDict));
-    }
+    builder.setEncodedRegionName(
+      compressor.compress(this.encodedRegionName, CompressionContext.DictionaryIndex.REGION));
+    builder.setTableName(
+      compressor.compress(this.tablename.getName(), CompressionContext.DictionaryIndex.TABLE));
     builder.setLogSequenceNumber(getSequenceId());
     builder.setWriteTime(writeTime);
     if (this.origLogSeqNum > 0) {
@@ -548,29 +533,22 @@ public class WALKeyImpl implements WALKey {
     }
     if (replicationScope != null) {
       for (Map.Entry<byte[], Integer> e : replicationScope.entrySet()) {
-        ByteString family = (compressionContext == null)
-            ? UnsafeByteOperations.unsafeWrap(e.getKey())
-            : compressor.compress(e.getKey(), compressionContext.familyDict);
-        builder.addScopes(FamilyScope.newBuilder()
-            .setFamily(family).setScopeType(ScopeType.forNumber(e.getValue())));
+        ByteString family =
+            compressor.compress(e.getKey(), CompressionContext.DictionaryIndex.FAMILY);
+        builder.addScopes(FamilyScope.newBuilder().setFamily(family)
+            .setScopeType(ScopeType.forNumber(e.getValue())));
       }
     }
     return builder;
   }
 
   public void readFieldsFromPb(WALProtos.WALKey walKey,
-                               WALCellCodec.ByteStringUncompressor uncompressor)
-      throws IOException {
-    if (this.compressionContext != null) {
-      this.encodedRegionName = uncompressor.uncompress(
-          walKey.getEncodedRegionName(), compressionContext.regionDict);
-      byte[] tablenameBytes = uncompressor.uncompress(
-          walKey.getTableName(), compressionContext.tableDict);
-      this.tablename = TableName.valueOf(tablenameBytes);
-    } else {
-      this.encodedRegionName = walKey.getEncodedRegionName().toByteArray();
-      this.tablename = TableName.valueOf(walKey.getTableName().toByteArray());
-    }
+      WALCellCodec.ByteStringUncompressor uncompressor) throws IOException {
+    this.encodedRegionName = uncompressor.uncompress(walKey.getEncodedRegionName(),
+      CompressionContext.DictionaryIndex.REGION);
+    byte[] tablenameBytes =
+        uncompressor.uncompress(walKey.getTableName(), CompressionContext.DictionaryIndex.TABLE);
+    this.tablename = TableName.valueOf(tablenameBytes);
     clusterIds.clear();
     for (HBaseProtos.UUID clusterId : walKey.getClusterIdsList()) {
       clusterIds.add(new UUID(clusterId.getMostSigBits(), clusterId.getLeastSigBits()));
@@ -585,14 +563,14 @@ public class WALKeyImpl implements WALKey {
     if (walKey.getScopesCount() > 0) {
       this.replicationScope = new TreeMap<>(Bytes.BYTES_COMPARATOR);
       for (FamilyScope scope : walKey.getScopesList()) {
-        byte[] family = (compressionContext == null) ? scope.getFamily().toByteArray() :
-          uncompressor.uncompress(scope.getFamily(), compressionContext.familyDict);
+        byte[] family =
+            uncompressor.uncompress(scope.getFamily(), CompressionContext.DictionaryIndex.FAMILY);
         this.replicationScope.put(family, scope.getScopeType().getNumber());
       }
     }
     setSequenceId(walKey.getLogSequenceNumber());
     this.writeTime = walKey.getWriteTime();
-    if(walKey.hasOrigSequenceNumber()) {
+    if (walKey.hasOrigSequenceNumber()) {
       this.origLogSeqNum = walKey.getOrigSequenceNumber();
     }
   }

http://git-wip-us.apache.org/repos/asf/hbase/blob/0b28155d/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/FaultyProtobufLogReader.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/FaultyProtobufLogReader.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/FaultyProtobufLogReader.java
index f1508e5..2c74f80 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/FaultyProtobufLogReader.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/FaultyProtobufLogReader.java
@@ -45,9 +45,6 @@ public class FaultyProtobufLogReader extends ProtobufLogReader {
       boolean b;
       do {
         Entry e = new Entry();
-        if (compressionContext != null) {
-          e.setCompressionContext(compressionContext);
-        }
         b = readNext(e);
         nextQueue.offer(e);
         numberOfFileEntries++;


Mime
View raw message