From 37f39c40c35cdf2b43949ae66c475363bad855d7 Mon Sep 17 00:00:00 2001
From: Michael Stack <stack@apache.org>
Date: Mon, 27 Aug 2012 18:47:49 +0000
Subject: [PATCH] HBASE-6477 Use PB filter definitions in RPC

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1377788 13f79535-47bb-0310-9956-ffa450edef68
---
 .../hadoop/hbase/io/HbaseObjectWritable.java      | 15 +++++++++++++++
 .../hadoop/hbase/protobuf/ProtobufUtil.java       | 14 ++++++++------
 2 files changed, 23 insertions(+), 6 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/HbaseObjectWritable.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/HbaseObjectWritable.java
index d1aeb16914f..d2fb1bd29e6 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/HbaseObjectWritable.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/HbaseObjectWritable.java
@@ -85,6 +85,8 @@ import org.apache.hadoop.hbase.filter.SkipFilter;
 import org.apache.hadoop.hbase.filter.ValueFilter;
 import org.apache.hadoop.hbase.filter.WhileMatchFilter;
 import org.apache.hadoop.hbase.filter.WritableByteArrayComparable;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
 import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.regionserver.RegionOpeningState;
 import org.apache.hadoop.hbase.regionserver.wal.HLog;
@@ -375,6 +377,8 @@ public class HbaseObjectWritable implements Writable, WritableWithSize, Configur
         code = CLASS_TO_CODE.get(Message.class);
       } else if (Serializable.class.isAssignableFrom(c)){
         code = CLASS_TO_CODE.get(Serializable.class);
+      } else if (Scan.class.isAssignableFrom(c)) {
+        code = CLASS_TO_CODE.get(Scan.class);
       }
     }
     return code;
@@ -543,6 +547,11 @@ public class HbaseObjectWritable implements Writable, WritableWithSize, Configur
         if(bos!=null) bos.close();
         if(oos!=null) oos.close();
       }
+    } else if (Scan.class.isAssignableFrom(declClass)) {
+      Scan scan = (Scan)instanceObj;
+      byte [] scanBytes = ProtobufUtil.toScan(scan).toByteArray();
+      out.writeInt(scanBytes.length);
+      out.write(scanBytes);
     } else {
       throw new IOException("Can't write: "+instanceObj+" as "+declClass);
     }
@@ -668,6 +677,12 @@ public class HbaseObjectWritable implements Writable, WritableWithSize, Configur
         LOG.error("Can't find class " + className, e);
         throw new IOException("Can't find class " + className, e);
       }
+    } else if (Scan.class.isAssignableFrom(declaredClass)) {
+      int length = in.readInt();
+      byte [] scanBytes = new byte[length];
+      in.readFully(scanBytes);
+      ClientProtos.Scan.Builder scanProto = ClientProtos.Scan.newBuilder();
+      instance = ProtobufUtil.toScan(scanProto.mergeFrom(scanBytes).build());
     } else {                                      // Writable or Serializable
       Class instanceClass = null;
       int b = (byte)WritableUtils.readVInt(in);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java
index 8e82bce1c1d..17dbddaad5d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java
@@ -32,6 +32,7 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeSet;
 import java.util.Map.Entry;
 import java.util.NavigableMap;
 import java.util.NavigableSet;
@@ -693,16 +694,17 @@ public final class ProtobufUtil {
       scan.setAttribute(attribute.getName(), attribute.getValue().toByteArray());
     }
     if (proto.getColumnCount() > 0) {
+      TreeMap<byte [], NavigableSet<byte[]>> familyMap =
+        new TreeMap<byte [], NavigableSet<byte []>>(Bytes.BYTES_COMPARATOR);
       for (Column column: proto.getColumnList()) {
         byte[] family = column.getFamily().toByteArray();
-        if (column.getQualifierCount() > 0) {
-          for (ByteString qualifier: column.getQualifierList()) {
-            scan.addColumn(family, qualifier.toByteArray());
-          }
-        } else {
-          scan.addFamily(family);
+        TreeSet<byte []> set = new TreeSet<byte []>(Bytes.BYTES_COMPARATOR);
+        for (ByteString qualifier: column.getQualifierList()) {
+          set.add(qualifier.toByteArray());
         }
+        familyMap.put(family, set);
       }
+      scan.setFamilyMap(familyMap);
     }
     return scan;
   }