Revert "HBASE-22664 Move protobuf stuff in hbase-rsgroup to hbase-protocol-shaded...
[hbase.git] / hbase-client / src / main / java / org / apache / hadoop / hbase / protobuf / ProtobufUtil.java
blobac0695b35ddd64e2bab99e22748e3ce7d07f0ab1
1 /**
2 * Licensed to the Apache Software Foundation (ASF) under one
3 * or more contributor license agreements. See the NOTICE file
4 * distributed with this work for additional information
5 * regarding copyright ownership. The ASF licenses this file
6 * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance
8 * with the License. You may obtain a copy of the License at
10 * http://www.apache.org/licenses/LICENSE-2.0
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
18 package org.apache.hadoop.hbase.protobuf;
20 import static org.apache.hadoop.hbase.protobuf.ProtobufMagic.PB_MAGIC;
22 import com.google.protobuf.ByteString;
23 import com.google.protobuf.CodedInputStream;
24 import com.google.protobuf.InvalidProtocolBufferException;
25 import com.google.protobuf.Message;
26 import com.google.protobuf.Parser;
27 import com.google.protobuf.RpcChannel;
28 import com.google.protobuf.RpcController;
29 import com.google.protobuf.Service;
30 import com.google.protobuf.ServiceException;
31 import com.google.protobuf.TextFormat;
32 import java.io.IOException;
33 import java.lang.reflect.Constructor;
34 import java.lang.reflect.Method;
35 import java.security.AccessController;
36 import java.security.PrivilegedAction;
37 import java.util.ArrayList;
38 import java.util.List;
39 import java.util.Map;
40 import java.util.NavigableSet;
41 import java.util.function.Function;
42 import org.apache.hadoop.conf.Configuration;
43 import org.apache.hadoop.hbase.Cell;
44 import org.apache.hadoop.hbase.Cell.Type;
45 import org.apache.hadoop.hbase.CellBuilderType;
46 import org.apache.hadoop.hbase.CellScanner;
47 import org.apache.hadoop.hbase.CellUtil;
48 import org.apache.hadoop.hbase.DoNotRetryIOException;
49 import org.apache.hadoop.hbase.ExtendedCellBuilder;
50 import org.apache.hadoop.hbase.ExtendedCellBuilderFactory;
51 import org.apache.hadoop.hbase.HBaseConfiguration;
52 import org.apache.hadoop.hbase.HBaseIOException;
53 import org.apache.hadoop.hbase.HConstants;
54 import org.apache.hadoop.hbase.KeyValue;
55 import org.apache.hadoop.hbase.ServerName;
56 import org.apache.hadoop.hbase.TableName;
57 import org.apache.hadoop.hbase.client.Append;
58 import org.apache.hadoop.hbase.client.Consistency;
59 import org.apache.hadoop.hbase.client.Delete;
60 import org.apache.hadoop.hbase.client.Durability;
61 import org.apache.hadoop.hbase.client.Get;
62 import org.apache.hadoop.hbase.client.Increment;
63 import org.apache.hadoop.hbase.client.Mutation;
64 import org.apache.hadoop.hbase.client.PackagePrivateFieldAccessor;
65 import org.apache.hadoop.hbase.client.Put;
66 import org.apache.hadoop.hbase.client.Result;
67 import org.apache.hadoop.hbase.client.Scan;
68 import org.apache.hadoop.hbase.client.SnapshotType;
69 import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
70 import org.apache.hadoop.hbase.exceptions.DeserializationException;
71 import org.apache.hadoop.hbase.filter.ByteArrayComparable;
72 import org.apache.hadoop.hbase.filter.Filter;
73 import org.apache.hadoop.hbase.io.TimeRange;
74 import org.apache.hadoop.hbase.net.Address;
75 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
76 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetServerInfoRequest;
77 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetServerInfoResponse;
78 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ServerInfo;
79 import org.apache.hadoop.hbase.protobuf.generated.CellProtos;
80 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
81 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.Column;
82 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto;
83 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.ColumnValue;
84 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.ColumnValue.QualifierValue;
85 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.DeleteType;
86 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType;
87 import org.apache.hadoop.hbase.protobuf.generated.ComparatorProtos;
88 import org.apache.hadoop.hbase.protobuf.generated.FilterProtos;
89 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
90 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.NameBytesPair;
91 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier;
92 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier.RegionSpecifierType;
93 import org.apache.hadoop.hbase.protobuf.generated.MapReduceProtos;
94 import org.apache.hadoop.hbase.protobuf.generated.RSGroupProtos;
95 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
96 import org.apache.hadoop.hbase.rsgroup.RSGroupInfo;
97 import org.apache.hadoop.hbase.util.Addressing;
98 import org.apache.hadoop.hbase.util.ByteStringer;
99 import org.apache.hadoop.hbase.util.Bytes;
100 import org.apache.hadoop.hbase.util.DynamicClassLoader;
101 import org.apache.hadoop.hbase.util.ExceptionUtil;
102 import org.apache.hadoop.hbase.util.Methods;
103 import org.apache.hadoop.ipc.RemoteException;
104 import org.apache.yetus.audience.InterfaceAudience;
107 * Protobufs utility.
108 * NOTE: This class OVERLAPS ProtobufUtil in the subpackage 'shaded'. The latter is used
109 * internally and has more methods. This Class is for Coprocessor Endpoints only though they
110 * should not be using this private class. It should not be depended upon. Most methods here
111 * are COPIED from the shaded ProtobufUtils with only difference being they refer to non-shaded
112 * protobufs.
113 * @see ProtobufUtil
115 // TODO: Generate this class from the shaded version.
116 @InterfaceAudience.Private // TODO: some clients (Hive, etc) use this class.
117 public final class ProtobufUtil {
118 private ProtobufUtil() {
122 * Many results are simple: no cell, exists true or false. To save on object creations,
123 * we reuse them across calls.
125 // TODO: PICK THESE UP FROM THE SHADED PROTOBUF.
126 private final static Cell[] EMPTY_CELL_ARRAY = new Cell[]{};
127 private final static Result EMPTY_RESULT = Result.create(EMPTY_CELL_ARRAY);
128 final static Result EMPTY_RESULT_EXISTS_TRUE = Result.create(null, true);
129 final static Result EMPTY_RESULT_EXISTS_FALSE = Result.create(null, false);
130 private final static Result EMPTY_RESULT_STALE = Result.create(EMPTY_CELL_ARRAY, null, true);
131 private final static Result EMPTY_RESULT_EXISTS_TRUE_STALE
132 = Result.create((Cell[])null, true, true);
133 private final static Result EMPTY_RESULT_EXISTS_FALSE_STALE
134 = Result.create((Cell[])null, false, true);
136 private final static ClientProtos.Result EMPTY_RESULT_PB;
137 private final static ClientProtos.Result EMPTY_RESULT_PB_EXISTS_TRUE;
138 private final static ClientProtos.Result EMPTY_RESULT_PB_EXISTS_FALSE;
139 private final static ClientProtos.Result EMPTY_RESULT_PB_STALE;
140 private final static ClientProtos.Result EMPTY_RESULT_PB_EXISTS_TRUE_STALE;
141 private final static ClientProtos.Result EMPTY_RESULT_PB_EXISTS_FALSE_STALE;
144 static {
145 ClientProtos.Result.Builder builder = ClientProtos.Result.newBuilder();
147 builder.setExists(true);
148 builder.setAssociatedCellCount(0);
149 EMPTY_RESULT_PB_EXISTS_TRUE = builder.build();
151 builder.setStale(true);
152 EMPTY_RESULT_PB_EXISTS_TRUE_STALE = builder.build();
153 builder.clear();
155 builder.setExists(false);
156 builder.setAssociatedCellCount(0);
157 EMPTY_RESULT_PB_EXISTS_FALSE = builder.build();
158 builder.setStale(true);
159 EMPTY_RESULT_PB_EXISTS_FALSE_STALE = builder.build();
161 builder.clear();
162 builder.setAssociatedCellCount(0);
163 EMPTY_RESULT_PB = builder.build();
164 builder.setStale(true);
165 EMPTY_RESULT_PB_STALE = builder.build();
169 * Dynamic class loader to load filter/comparators
171 private final static class ClassLoaderHolder {
172 private final static ClassLoader CLASS_LOADER;
174 static {
175 ClassLoader parent = ProtobufUtil.class.getClassLoader();
176 Configuration conf = HBaseConfiguration.create();
177 CLASS_LOADER = AccessController.doPrivileged((PrivilegedAction<ClassLoader>)
178 () -> new DynamicClassLoader(conf, parent)
184 * Prepend the passed bytes with four bytes of magic, {@link ProtobufMagic#PB_MAGIC},
185 * to flag what follows as a protobuf in hbase. Prepend these bytes to all content written to
186 * znodes, etc.
187 * @param bytes Bytes to decorate
188 * @return The passed <code>bytes</code> with magic prepended (Creates a new
189 * byte array that is <code>bytes.length</code> plus {@link ProtobufMagic#PB_MAGIC}.length.
191 public static byte [] prependPBMagic(final byte [] bytes) {
192 return Bytes.add(PB_MAGIC, bytes);
196 * @param bytes Bytes to check.
197 * @return True if passed <code>bytes</code> has {@link ProtobufMagic#PB_MAGIC} for a prefix.
199 public static boolean isPBMagicPrefix(final byte [] bytes) {
200 return ProtobufMagic.isPBMagicPrefix(bytes);
204 * @param bytes Bytes to check.
205 * @param offset offset to start at
206 * @param len length to use
207 * @return True if passed <code>bytes</code> has {@link ProtobufMagic#PB_MAGIC} for a prefix.
209 public static boolean isPBMagicPrefix(final byte [] bytes, int offset, int len) {
210 return ProtobufMagic.isPBMagicPrefix(bytes, offset, len);
214 * @param bytes bytes to check
215 * @throws DeserializationException if we are missing the pb magic prefix
217 public static void expectPBMagicPrefix(final byte[] bytes) throws DeserializationException {
218 if (!isPBMagicPrefix(bytes)) {
219 String bytesPrefix = bytes == null ? "null" : Bytes.toStringBinary(bytes, 0, PB_MAGIC.length);
220 throw new DeserializationException(
221 "Missing pb magic " + Bytes.toString(PB_MAGIC) + " prefix, bytes: " + bytesPrefix);
226 * @return Length of {@link ProtobufMagic#lengthOfPBMagic()}
228 public static int lengthOfPBMagic() {
229 return ProtobufMagic.lengthOfPBMagic();
233 * Return the IOException thrown by the remote server wrapped in
234 * ServiceException as cause.
236 * @param se ServiceException that wraps IO exception thrown by the server
237 * @return Exception wrapped in ServiceException or
238 * a new IOException that wraps the unexpected ServiceException.
240 public static IOException getRemoteException(ServiceException se) {
241 return makeIOExceptionOfException(se);
245 * Return the Exception thrown by the remote server wrapped in
246 * ServiceException as cause. RemoteException are left untouched.
248 * @param e ServiceException that wraps IO exception thrown by the server
249 * @return Exception wrapped in ServiceException.
251 public static IOException getServiceException(org.apache.hbase.thirdparty.com.google.protobuf.ServiceException e) {
252 Throwable t = e.getCause();
253 if (ExceptionUtil.isInterrupt(t)) {
254 return ExceptionUtil.asInterrupt(t);
256 return t instanceof IOException ? (IOException) t : new HBaseIOException(t);
260 * Like {@link #getRemoteException(ServiceException)} but more generic, able to handle more than
261 * just {@link ServiceException}. Prefer this method to
262 * {@link #getRemoteException(ServiceException)} because trying to
263 * contain direct protobuf references.
265 public static IOException handleRemoteException(Throwable e) {
266 return makeIOExceptionOfException(e);
269 private static IOException makeIOExceptionOfException(Throwable e) {
270 Throwable t = e;
271 if (e instanceof ServiceException ||
272 e instanceof org.apache.hbase.thirdparty.com.google.protobuf.ServiceException) {
273 t = e.getCause();
275 if (ExceptionUtil.isInterrupt(t)) {
276 return ExceptionUtil.asInterrupt(t);
278 if (t instanceof RemoteException) {
279 t = ((RemoteException)t).unwrapRemoteException();
281 return t instanceof IOException? (IOException)t: new HBaseIOException(t);
285 * Convert a ServerName to a protocol buffer ServerName
287 * @param serverName the ServerName to convert
288 * @return the converted protocol buffer ServerName
289 * @see #toServerName(org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.ServerName)
291 public static HBaseProtos.ServerName
292 toServerName(final ServerName serverName) {
293 if (serverName == null) return null;
294 HBaseProtos.ServerName.Builder builder =
295 HBaseProtos.ServerName.newBuilder();
296 builder.setHostName(serverName.getHostname());
297 if (serverName.getPort() >= 0) {
298 builder.setPort(serverName.getPort());
300 if (serverName.getStartcode() >= 0) {
301 builder.setStartCode(serverName.getStartcode());
303 return builder.build();
307 * Convert a protocol buffer ServerName to a ServerName
309 * @param proto the protocol buffer ServerName to convert
310 * @return the converted ServerName
312 public static ServerName toServerName(final HBaseProtos.ServerName proto) {
313 if (proto == null) return null;
314 String hostName = proto.getHostName();
315 long startCode = -1;
316 int port = -1;
317 if (proto.hasPort()) {
318 port = proto.getPort();
320 if (proto.hasStartCode()) {
321 startCode = proto.getStartCode();
323 return ServerName.valueOf(hostName, port, startCode);
327 * Convert a protobuf Durability into a client Durability
329 public static Durability toDurability(
330 final ClientProtos.MutationProto.Durability proto) {
331 switch(proto) {
332 case USE_DEFAULT:
333 return Durability.USE_DEFAULT;
334 case SKIP_WAL:
335 return Durability.SKIP_WAL;
336 case ASYNC_WAL:
337 return Durability.ASYNC_WAL;
338 case SYNC_WAL:
339 return Durability.SYNC_WAL;
340 case FSYNC_WAL:
341 return Durability.FSYNC_WAL;
342 default:
343 return Durability.USE_DEFAULT;
348 * Convert a client Durability into a protbuf Durability
350 public static ClientProtos.MutationProto.Durability toDurability(
351 final Durability d) {
352 switch(d) {
353 case USE_DEFAULT:
354 return ClientProtos.MutationProto.Durability.USE_DEFAULT;
355 case SKIP_WAL:
356 return ClientProtos.MutationProto.Durability.SKIP_WAL;
357 case ASYNC_WAL:
358 return ClientProtos.MutationProto.Durability.ASYNC_WAL;
359 case SYNC_WAL:
360 return ClientProtos.MutationProto.Durability.SYNC_WAL;
361 case FSYNC_WAL:
362 return ClientProtos.MutationProto.Durability.FSYNC_WAL;
363 default:
364 return ClientProtos.MutationProto.Durability.USE_DEFAULT;
369 * Convert a protocol buffer Get to a client Get
371 * @param proto the protocol buffer Get to convert
372 * @return the converted client Get
373 * @throws IOException
375 public static Get toGet(final ClientProtos.Get proto) throws IOException {
376 if (proto == null) return null;
377 byte[] row = proto.getRow().toByteArray();
378 Get get = new Get(row);
379 if (proto.hasCacheBlocks()) {
380 get.setCacheBlocks(proto.getCacheBlocks());
382 if (proto.hasMaxVersions()) {
383 get.readVersions(proto.getMaxVersions());
385 if (proto.hasStoreLimit()) {
386 get.setMaxResultsPerColumnFamily(proto.getStoreLimit());
388 if (proto.hasStoreOffset()) {
389 get.setRowOffsetPerColumnFamily(proto.getStoreOffset());
391 if (proto.getCfTimeRangeCount() > 0) {
392 for (HBaseProtos.ColumnFamilyTimeRange cftr : proto.getCfTimeRangeList()) {
393 TimeRange timeRange = protoToTimeRange(cftr.getTimeRange());
394 get.setColumnFamilyTimeRange(cftr.getColumnFamily().toByteArray(),
395 timeRange.getMin(), timeRange.getMax());
398 if (proto.hasTimeRange()) {
399 TimeRange timeRange = protoToTimeRange(proto.getTimeRange());
400 get.setTimeRange(timeRange.getMin(), timeRange.getMax());
402 if (proto.hasFilter()) {
403 FilterProtos.Filter filter = proto.getFilter();
404 get.setFilter(ProtobufUtil.toFilter(filter));
406 for (NameBytesPair attribute: proto.getAttributeList()) {
407 get.setAttribute(attribute.getName(), attribute.getValue().toByteArray());
409 if (proto.getColumnCount() > 0) {
410 for (Column column: proto.getColumnList()) {
411 byte[] family = column.getFamily().toByteArray();
412 if (column.getQualifierCount() > 0) {
413 for (ByteString qualifier: column.getQualifierList()) {
414 get.addColumn(family, qualifier.toByteArray());
416 } else {
417 get.addFamily(family);
421 if (proto.hasExistenceOnly() && proto.getExistenceOnly()){
422 get.setCheckExistenceOnly(true);
424 if (proto.hasConsistency()) {
425 get.setConsistency(toConsistency(proto.getConsistency()));
427 if (proto.hasLoadColumnFamiliesOnDemand()) {
428 get.setLoadColumnFamiliesOnDemand(proto.getLoadColumnFamiliesOnDemand());
430 return get;
433 public static Consistency toConsistency(ClientProtos.Consistency consistency) {
434 switch (consistency) {
435 case STRONG : return Consistency.STRONG;
436 case TIMELINE : return Consistency.TIMELINE;
437 default : return Consistency.STRONG;
441 public static ClientProtos.Consistency toConsistency(Consistency consistency) {
442 switch (consistency) {
443 case STRONG : return ClientProtos.Consistency.STRONG;
444 case TIMELINE : return ClientProtos.Consistency.TIMELINE;
445 default : return ClientProtos.Consistency.STRONG;
450 * Convert a protocol buffer Mutate to a Put.
452 * @param proto The protocol buffer MutationProto to convert
453 * @return A client Put.
454 * @throws IOException
456 public static Put toPut(final MutationProto proto)
457 throws IOException {
458 return toPut(proto, null);
462 * Convert a protocol buffer Mutate to a Put.
464 * @param proto The protocol buffer MutationProto to convert
465 * @param cellScanner If non-null, the Cell data that goes with this proto.
466 * @return A client Put.
467 * @throws IOException
469 public static Put toPut(final MutationProto proto, final CellScanner cellScanner)
470 throws IOException {
471 // TODO: Server-side at least why do we convert back to the Client types? Why not just pb it?
472 MutationType type = proto.getMutateType();
473 assert type == MutationType.PUT: type.name();
474 long timestamp = proto.hasTimestamp()? proto.getTimestamp(): HConstants.LATEST_TIMESTAMP;
475 Put put = proto.hasRow() ? new Put(proto.getRow().toByteArray(), timestamp) : null;
476 int cellCount = proto.hasAssociatedCellCount()? proto.getAssociatedCellCount(): 0;
477 if (cellCount > 0) {
478 // The proto has metadata only and the data is separate to be found in the cellScanner.
479 if (cellScanner == null) {
480 throw new DoNotRetryIOException("Cell count of " + cellCount + " but no cellScanner: " +
481 toShortString(proto));
483 for (int i = 0; i < cellCount; i++) {
484 if (!cellScanner.advance()) {
485 throw new DoNotRetryIOException("Cell count of " + cellCount + " but at index " + i +
486 " no cell returned: " + toShortString(proto));
488 Cell cell = cellScanner.current();
489 if (put == null) {
490 put = new Put(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(), timestamp);
492 put.add(cell);
494 } else {
495 if (put == null) {
496 throw new IllegalArgumentException("row cannot be null");
498 // The proto has the metadata and the data itself
499 ExtendedCellBuilder cellBuilder = ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY);
500 for (ColumnValue column: proto.getColumnValueList()) {
501 byte[] family = column.getFamily().toByteArray();
502 for (QualifierValue qv: column.getQualifierValueList()) {
503 if (!qv.hasValue()) {
504 throw new DoNotRetryIOException(
505 "Missing required field: qualifier value");
507 long ts = timestamp;
508 if (qv.hasTimestamp()) {
509 ts = qv.getTimestamp();
511 byte[] allTagsBytes;
512 if (qv.hasTags()) {
513 allTagsBytes = qv.getTags().toByteArray();
514 if(qv.hasDeleteType()) {
515 put.add(cellBuilder.clear()
516 .setRow(put.getRow())
517 .setFamily(family)
518 .setQualifier(qv.hasQualifier() ? qv.getQualifier().toByteArray() : null)
519 .setTimestamp(ts)
520 .setType(fromDeleteType(qv.getDeleteType()).getCode())
521 .setTags(allTagsBytes)
522 .build());
523 } else {
524 put.add(cellBuilder.clear()
525 .setRow(put.getRow())
526 .setFamily(family)
527 .setQualifier(qv.hasQualifier() ? qv.getQualifier().toByteArray() : null)
528 .setTimestamp(ts)
529 .setType(Cell.Type.Put)
530 .setValue(qv.hasValue() ? qv.getValue().toByteArray() : null)
531 .setTags(allTagsBytes)
532 .build());
534 } else {
535 if(qv.hasDeleteType()) {
536 put.add(cellBuilder.clear()
537 .setRow(put.getRow())
538 .setFamily(family)
539 .setQualifier(qv.hasQualifier() ? qv.getQualifier().toByteArray() : null)
540 .setTimestamp(ts)
541 .setType(fromDeleteType(qv.getDeleteType()).getCode())
542 .build());
543 } else{
544 put.add(cellBuilder.clear()
545 .setRow(put.getRow())
546 .setFamily(family)
547 .setQualifier(qv.hasQualifier() ? qv.getQualifier().toByteArray() : null)
548 .setTimestamp(ts)
549 .setType(Type.Put)
550 .setValue(qv.hasValue() ? qv.getValue().toByteArray() : null)
551 .build());
557 put.setDurability(toDurability(proto.getDurability()));
558 for (NameBytesPair attribute: proto.getAttributeList()) {
559 put.setAttribute(attribute.getName(), attribute.getValue().toByteArray());
561 return put;
565 * Convert a protocol buffer Mutate to a Delete
567 * @param proto the protocol buffer Mutate to convert
568 * @return the converted client Delete
569 * @throws IOException
571 public static Delete toDelete(final MutationProto proto)
572 throws IOException {
573 return toDelete(proto, null);
577 * Convert a protocol buffer Mutate to a Delete
579 * @param proto the protocol buffer Mutate to convert
580 * @param cellScanner if non-null, the data that goes with this delete.
581 * @return the converted client Delete
582 * @throws IOException
584 public static Delete toDelete(final MutationProto proto, final CellScanner cellScanner)
585 throws IOException {
586 MutationType type = proto.getMutateType();
587 assert type == MutationType.DELETE : type.name();
588 long timestamp = proto.hasTimestamp() ? proto.getTimestamp() : HConstants.LATEST_TIMESTAMP;
589 Delete delete = proto.hasRow() ? new Delete(proto.getRow().toByteArray(), timestamp) : null;
590 int cellCount = proto.hasAssociatedCellCount()? proto.getAssociatedCellCount(): 0;
591 if (cellCount > 0) {
592 // The proto has metadata only and the data is separate to be found in the cellScanner.
593 if (cellScanner == null) {
594 // TextFormat should be fine for a Delete since it carries no data, just coordinates.
595 throw new DoNotRetryIOException("Cell count of " + cellCount + " but no cellScanner: " +
596 TextFormat.shortDebugString(proto));
598 for (int i = 0; i < cellCount; i++) {
599 if (!cellScanner.advance()) {
600 // TextFormat should be fine for a Delete since it carries no data, just coordinates.
601 throw new DoNotRetryIOException("Cell count of " + cellCount + " but at index " + i +
602 " no cell returned: " + TextFormat.shortDebugString(proto));
604 Cell cell = cellScanner.current();
605 if (delete == null) {
606 delete =
607 new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(), timestamp);
609 delete.add(cell);
611 } else {
612 if (delete == null) {
613 throw new IllegalArgumentException("row cannot be null");
615 for (ColumnValue column: proto.getColumnValueList()) {
616 byte[] family = column.getFamily().toByteArray();
617 for (QualifierValue qv: column.getQualifierValueList()) {
618 DeleteType deleteType = qv.getDeleteType();
619 byte[] qualifier = null;
620 if (qv.hasQualifier()) {
621 qualifier = qv.getQualifier().toByteArray();
623 long ts = HConstants.LATEST_TIMESTAMP;
624 if (qv.hasTimestamp()) {
625 ts = qv.getTimestamp();
627 if (deleteType == DeleteType.DELETE_ONE_VERSION) {
628 delete.addColumn(family, qualifier, ts);
629 } else if (deleteType == DeleteType.DELETE_MULTIPLE_VERSIONS) {
630 delete.addColumns(family, qualifier, ts);
631 } else if (deleteType == DeleteType.DELETE_FAMILY_VERSION) {
632 delete.addFamilyVersion(family, ts);
633 } else {
634 delete.addFamily(family, ts);
639 delete.setDurability(toDurability(proto.getDurability()));
640 for (NameBytesPair attribute: proto.getAttributeList()) {
641 delete.setAttribute(attribute.getName(), attribute.getValue().toByteArray());
643 return delete;
646 @FunctionalInterface
647 private interface ConsumerWithException <T, U> {
648 void accept(T t, U u) throws IOException;
651 private static <T extends Mutation> T toDelta(Function<Bytes, T> supplier, ConsumerWithException<T, Cell> consumer,
652 final MutationProto proto, final CellScanner cellScanner) throws IOException {
653 byte[] row = proto.hasRow() ? proto.getRow().toByteArray() : null;
654 T mutation = row == null ? null : supplier.apply(new Bytes(row));
655 int cellCount = proto.hasAssociatedCellCount() ? proto.getAssociatedCellCount() : 0;
656 if (cellCount > 0) {
657 // The proto has metadata only and the data is separate to be found in the cellScanner.
658 if (cellScanner == null) {
659 throw new DoNotRetryIOException("Cell count of " + cellCount + " but no cellScanner: " +
660 toShortString(proto));
662 for (int i = 0; i < cellCount; i++) {
663 if (!cellScanner.advance()) {
664 throw new DoNotRetryIOException("Cell count of " + cellCount + " but at index " + i +
665 " no cell returned: " + toShortString(proto));
667 Cell cell = cellScanner.current();
668 if (mutation == null) {
669 mutation = supplier.apply(new Bytes(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
671 consumer.accept(mutation, cell);
673 } else {
674 if (mutation == null) {
675 throw new IllegalArgumentException("row cannot be null");
677 for (ColumnValue column : proto.getColumnValueList()) {
678 byte[] family = column.getFamily().toByteArray();
679 for (QualifierValue qv : column.getQualifierValueList()) {
680 byte[] qualifier = qv.getQualifier().toByteArray();
681 if (!qv.hasValue()) {
682 throw new DoNotRetryIOException(
683 "Missing required field: qualifier value");
685 byte[] value = qv.getValue().toByteArray();
686 byte[] tags = null;
687 if (qv.hasTags()) {
688 tags = qv.getTags().toByteArray();
690 consumer.accept(mutation, CellUtil.createCell(mutation.getRow(), family, qualifier, qv.getTimestamp(),
691 KeyValue.Type.Put, value, tags));
695 mutation.setDurability(toDurability(proto.getDurability()));
696 for (NameBytesPair attribute : proto.getAttributeList()) {
697 mutation.setAttribute(attribute.getName(), attribute.getValue().toByteArray());
699 return mutation;
703 * Convert a protocol buffer Mutate to an Append
704 * @param cellScanner
705 * @param proto the protocol buffer Mutate to convert
706 * @return the converted client Append
707 * @throws IOException
709 public static Append toAppend(final MutationProto proto, final CellScanner cellScanner)
710 throws IOException {
711 MutationType type = proto.getMutateType();
712 assert type == MutationType.APPEND : type.name();
713 Append append = toDelta((Bytes row) -> new Append(row.get(), row.getOffset(), row.getLength()),
714 Append::add, proto, cellScanner);
715 if (proto.hasTimeRange()) {
716 TimeRange timeRange = protoToTimeRange(proto.getTimeRange());
717 append.setTimeRange(timeRange.getMin(), timeRange.getMax());
719 return append;
723 * Convert a protocol buffer Mutate to an Increment
725 * @param proto the protocol buffer Mutate to convert
726 * @return the converted client Increment
727 * @throws IOException
729 public static Increment toIncrement(final MutationProto proto, final CellScanner cellScanner)
730 throws IOException {
731 MutationType type = proto.getMutateType();
732 assert type == MutationType.INCREMENT : type.name();
733 Increment increment = toDelta((Bytes row) -> new Increment(row.get(), row.getOffset(), row.getLength()),
734 Increment::add, proto, cellScanner);
735 if (proto.hasTimeRange()) {
736 TimeRange timeRange = protoToTimeRange(proto.getTimeRange());
737 increment.setTimeRange(timeRange.getMin(), timeRange.getMax());
739 return increment;
743 * Convert a MutateRequest to Mutation
745 * @param proto the protocol buffer Mutate to convert
746 * @return the converted Mutation
747 * @throws IOException
749 public static Mutation toMutation(final MutationProto proto) throws IOException {
750 MutationType type = proto.getMutateType();
751 if (type == MutationType.APPEND) {
752 return toAppend(proto, null);
754 if (type == MutationType.DELETE) {
755 return toDelete(proto, null);
757 if (type == MutationType.PUT) {
758 return toPut(proto, null);
760 throw new IOException("Unknown mutation type " + type);
764 * Convert a protocol buffer Mutate to a Get.
765 * @param proto the protocol buffer Mutate to convert.
766 * @param cellScanner
767 * @return the converted client get.
768 * @throws IOException
770 public static Get toGet(final MutationProto proto, final CellScanner cellScanner)
771 throws IOException {
772 MutationType type = proto.getMutateType();
773 assert type == MutationType.INCREMENT || type == MutationType.APPEND : type.name();
774 byte[] row = proto.hasRow() ? proto.getRow().toByteArray() : null;
775 Get get = null;
776 int cellCount = proto.hasAssociatedCellCount() ? proto.getAssociatedCellCount() : 0;
777 if (cellCount > 0) {
778 // The proto has metadata only and the data is separate to be found in the cellScanner.
779 if (cellScanner == null) {
780 throw new DoNotRetryIOException("Cell count of " + cellCount + " but no cellScanner: "
781 + TextFormat.shortDebugString(proto));
783 for (int i = 0; i < cellCount; i++) {
784 if (!cellScanner.advance()) {
785 throw new DoNotRetryIOException("Cell count of " + cellCount + " but at index " + i
786 + " no cell returned: " + TextFormat.shortDebugString(proto));
788 Cell cell = cellScanner.current();
789 if (get == null) {
790 get = new Get(Bytes.copy(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
792 get.addColumn(
793 Bytes.copy(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()),
794 Bytes.copy(cell.getQualifierArray(), cell.getQualifierOffset(),
795 cell.getQualifierLength()));
797 } else {
798 get = new Get(row);
799 for (ColumnValue column : proto.getColumnValueList()) {
800 byte[] family = column.getFamily().toByteArray();
801 for (QualifierValue qv : column.getQualifierValueList()) {
802 byte[] qualifier = qv.getQualifier().toByteArray();
803 if (!qv.hasValue()) {
804 throw new DoNotRetryIOException("Missing required field: qualifier value");
806 get.addColumn(family, qualifier);
810 if (proto.hasTimeRange()) {
811 TimeRange timeRange = protoToTimeRange(proto.getTimeRange());
812 get.setTimeRange(timeRange.getMin(), timeRange.getMax());
814 for (NameBytesPair attribute : proto.getAttributeList()) {
815 get.setAttribute(attribute.getName(), attribute.getValue().toByteArray());
817 return get;
820 public static ClientProtos.Scan.ReadType toReadType(Scan.ReadType readType) {
821 switch (readType) {
822 case DEFAULT:
823 return ClientProtos.Scan.ReadType.DEFAULT;
824 case STREAM:
825 return ClientProtos.Scan.ReadType.STREAM;
826 case PREAD:
827 return ClientProtos.Scan.ReadType.PREAD;
828 default:
829 throw new IllegalArgumentException("Unknown ReadType: " + readType);
833 public static Scan.ReadType toReadType(ClientProtos.Scan.ReadType readType) {
834 switch (readType) {
835 case DEFAULT:
836 return Scan.ReadType.DEFAULT;
837 case STREAM:
838 return Scan.ReadType.STREAM;
839 case PREAD:
840 return Scan.ReadType.PREAD;
841 default:
842 throw new IllegalArgumentException("Unknown ReadType: " + readType);
847 * Convert a client Scan to a protocol buffer Scan
849 * @param scan the client Scan to convert
850 * @return the converted protocol buffer Scan
851 * @throws IOException
853 public static ClientProtos.Scan toScan(
854 final Scan scan) throws IOException {
855 ClientProtos.Scan.Builder scanBuilder =
856 ClientProtos.Scan.newBuilder();
857 scanBuilder.setCacheBlocks(scan.getCacheBlocks());
858 if (scan.getBatch() > 0) {
859 scanBuilder.setBatchSize(scan.getBatch());
861 if (scan.getMaxResultSize() > 0) {
862 scanBuilder.setMaxResultSize(scan.getMaxResultSize());
864 if (scan.isSmall()) {
865 scanBuilder.setSmall(scan.isSmall());
867 if (scan.getAllowPartialResults()) {
868 scanBuilder.setAllowPartialResults(scan.getAllowPartialResults());
870 Boolean loadColumnFamiliesOnDemand = scan.getLoadColumnFamiliesOnDemandValue();
871 if (loadColumnFamiliesOnDemand != null) {
872 scanBuilder.setLoadColumnFamiliesOnDemand(loadColumnFamiliesOnDemand);
874 scanBuilder.setMaxVersions(scan.getMaxVersions());
875 scan.getColumnFamilyTimeRange().forEach((cf, timeRange) -> {
876 scanBuilder.addCfTimeRange(HBaseProtos.ColumnFamilyTimeRange.newBuilder()
877 .setColumnFamily(ByteStringer.wrap(cf))
878 .setTimeRange(toTimeRange(timeRange))
879 .build());
881 scanBuilder.setTimeRange(toTimeRange(scan.getTimeRange()));
882 Map<String, byte[]> attributes = scan.getAttributesMap();
883 if (!attributes.isEmpty()) {
884 NameBytesPair.Builder attributeBuilder = NameBytesPair.newBuilder();
885 for (Map.Entry<String, byte[]> attribute: attributes.entrySet()) {
886 attributeBuilder.setName(attribute.getKey());
887 attributeBuilder.setValue(ByteStringer.wrap(attribute.getValue()));
888 scanBuilder.addAttribute(attributeBuilder.build());
891 byte[] startRow = scan.getStartRow();
892 if (startRow != null && startRow.length > 0) {
893 scanBuilder.setStartRow(ByteStringer.wrap(startRow));
895 byte[] stopRow = scan.getStopRow();
896 if (stopRow != null && stopRow.length > 0) {
897 scanBuilder.setStopRow(ByteStringer.wrap(stopRow));
899 if (scan.hasFilter()) {
900 scanBuilder.setFilter(ProtobufUtil.toFilter(scan.getFilter()));
902 if (scan.hasFamilies()) {
903 Column.Builder columnBuilder = Column.newBuilder();
904 for (Map.Entry<byte[],NavigableSet<byte []>>
905 family: scan.getFamilyMap().entrySet()) {
906 columnBuilder.setFamily(ByteStringer.wrap(family.getKey()));
907 NavigableSet<byte []> qualifiers = family.getValue();
908 columnBuilder.clearQualifier();
909 if (qualifiers != null && qualifiers.size() > 0) {
910 for (byte [] qualifier: qualifiers) {
911 columnBuilder.addQualifier(ByteStringer.wrap(qualifier));
914 scanBuilder.addColumn(columnBuilder.build());
917 if (scan.getMaxResultsPerColumnFamily() >= 0) {
918 scanBuilder.setStoreLimit(scan.getMaxResultsPerColumnFamily());
920 if (scan.getRowOffsetPerColumnFamily() > 0) {
921 scanBuilder.setStoreOffset(scan.getRowOffsetPerColumnFamily());
923 if (scan.isReversed()) {
924 scanBuilder.setReversed(scan.isReversed());
926 if (scan.getConsistency() == Consistency.TIMELINE) {
927 scanBuilder.setConsistency(toConsistency(scan.getConsistency()));
929 if (scan.getCaching() > 0) {
930 scanBuilder.setCaching(scan.getCaching());
932 long mvccReadPoint = PackagePrivateFieldAccessor.getMvccReadPoint(scan);
933 if (mvccReadPoint > 0) {
934 scanBuilder.setMvccReadPoint(mvccReadPoint);
936 if (!scan.includeStartRow()) {
937 scanBuilder.setIncludeStartRow(false);
939 scanBuilder.setIncludeStopRow(scan.includeStopRow());
940 if (scan.getReadType() != Scan.ReadType.DEFAULT) {
941 scanBuilder.setReadType(toReadType(scan.getReadType()));
943 return scanBuilder.build();
947 * Convert a protocol buffer Scan to a client Scan
949 * @param proto the protocol buffer Scan to convert
950 * @return the converted client Scan
951 * @throws IOException
953 public static Scan toScan(
954 final ClientProtos.Scan proto) throws IOException {
955 byte[] startRow = HConstants.EMPTY_START_ROW;
956 byte[] stopRow = HConstants.EMPTY_END_ROW;
957 boolean includeStartRow = true;
958 boolean includeStopRow = false;
959 if (proto.hasStartRow()) {
960 startRow = proto.getStartRow().toByteArray();
962 if (proto.hasStopRow()) {
963 stopRow = proto.getStopRow().toByteArray();
965 if (proto.hasIncludeStartRow()) {
966 includeStartRow = proto.getIncludeStartRow();
968 if (proto.hasIncludeStopRow()) {
969 includeStopRow = proto.getIncludeStopRow();
971 Scan scan =
972 new Scan().withStartRow(startRow, includeStartRow).withStopRow(stopRow, includeStopRow);
973 if (proto.hasCacheBlocks()) {
974 scan.setCacheBlocks(proto.getCacheBlocks());
976 if (proto.hasMaxVersions()) {
977 scan.setMaxVersions(proto.getMaxVersions());
979 if (proto.hasStoreLimit()) {
980 scan.setMaxResultsPerColumnFamily(proto.getStoreLimit());
982 if (proto.hasStoreOffset()) {
983 scan.setRowOffsetPerColumnFamily(proto.getStoreOffset());
985 if (proto.hasLoadColumnFamiliesOnDemand()) {
986 scan.setLoadColumnFamiliesOnDemand(proto.getLoadColumnFamiliesOnDemand());
988 if (proto.getCfTimeRangeCount() > 0) {
989 for (HBaseProtos.ColumnFamilyTimeRange cftr : proto.getCfTimeRangeList()) {
990 TimeRange timeRange = protoToTimeRange(cftr.getTimeRange());
991 scan.setColumnFamilyTimeRange(cftr.getColumnFamily().toByteArray(),
992 timeRange.getMin(), timeRange.getMax());
995 if (proto.hasTimeRange()) {
996 TimeRange timeRange = protoToTimeRange(proto.getTimeRange());
997 scan.setTimeRange(timeRange.getMin(), timeRange.getMax());
999 if (proto.hasFilter()) {
1000 FilterProtos.Filter filter = proto.getFilter();
1001 scan.setFilter(ProtobufUtil.toFilter(filter));
1003 if (proto.hasBatchSize()) {
1004 scan.setBatch(proto.getBatchSize());
1006 if (proto.hasMaxResultSize()) {
1007 scan.setMaxResultSize(proto.getMaxResultSize());
1009 if (proto.hasSmall()) {
1010 scan.setSmall(proto.getSmall());
1012 if (proto.hasAllowPartialResults()) {
1013 scan.setAllowPartialResults(proto.getAllowPartialResults());
1015 for (NameBytesPair attribute: proto.getAttributeList()) {
1016 scan.setAttribute(attribute.getName(), attribute.getValue().toByteArray());
1018 if (proto.getColumnCount() > 0) {
1019 for (Column column: proto.getColumnList()) {
1020 byte[] family = column.getFamily().toByteArray();
1021 if (column.getQualifierCount() > 0) {
1022 for (ByteString qualifier: column.getQualifierList()) {
1023 scan.addColumn(family, qualifier.toByteArray());
1025 } else {
1026 scan.addFamily(family);
1030 if (proto.hasReversed()) {
1031 scan.setReversed(proto.getReversed());
1033 if (proto.hasConsistency()) {
1034 scan.setConsistency(toConsistency(proto.getConsistency()));
1036 if (proto.hasCaching()) {
1037 scan.setCaching(proto.getCaching());
1039 if (proto.hasMvccReadPoint()) {
1040 PackagePrivateFieldAccessor.setMvccReadPoint(scan, proto.getMvccReadPoint());
1042 if (scan.isSmall()) {
1043 scan.setReadType(Scan.ReadType.PREAD);
1044 } else if (proto.hasReadType()) {
1045 scan.setReadType(toReadType(proto.getReadType()));
1047 return scan;
1051 * Create a protocol buffer Get based on a client Get.
1053 * @param get the client Get
1054 * @return a protocol buffer Get
1055 * @throws IOException
1057 public static ClientProtos.Get toGet(
1058 final Get get) throws IOException {
1059 ClientProtos.Get.Builder builder =
1060 ClientProtos.Get.newBuilder();
1061 builder.setRow(ByteStringer.wrap(get.getRow()));
1062 builder.setCacheBlocks(get.getCacheBlocks());
1063 builder.setMaxVersions(get.getMaxVersions());
1064 if (get.getFilter() != null) {
1065 builder.setFilter(ProtobufUtil.toFilter(get.getFilter()));
1067 get.getColumnFamilyTimeRange().forEach((cf, timeRange) ->
1068 builder.addCfTimeRange(HBaseProtos.ColumnFamilyTimeRange.newBuilder()
1069 .setColumnFamily(ByteStringer.wrap(cf))
1070 .setTimeRange(toTimeRange(timeRange)).build())
1072 builder.setTimeRange(toTimeRange(get.getTimeRange()));
1073 Map<String, byte[]> attributes = get.getAttributesMap();
1074 if (!attributes.isEmpty()) {
1075 NameBytesPair.Builder attributeBuilder = NameBytesPair.newBuilder();
1076 for (Map.Entry<String, byte[]> attribute: attributes.entrySet()) {
1077 attributeBuilder.setName(attribute.getKey());
1078 attributeBuilder.setValue(ByteStringer.wrap(attribute.getValue()));
1079 builder.addAttribute(attributeBuilder.build());
1082 if (get.hasFamilies()) {
1083 Column.Builder columnBuilder = Column.newBuilder();
1084 Map<byte[], NavigableSet<byte[]>> families = get.getFamilyMap();
1085 for (Map.Entry<byte[], NavigableSet<byte[]>> family: families.entrySet()) {
1086 NavigableSet<byte[]> qualifiers = family.getValue();
1087 columnBuilder.setFamily(ByteStringer.wrap(family.getKey()));
1088 columnBuilder.clearQualifier();
1089 if (qualifiers != null && qualifiers.size() > 0) {
1090 for (byte[] qualifier: qualifiers) {
1091 columnBuilder.addQualifier(ByteStringer.wrap(qualifier));
1094 builder.addColumn(columnBuilder.build());
1097 if (get.getMaxResultsPerColumnFamily() >= 0) {
1098 builder.setStoreLimit(get.getMaxResultsPerColumnFamily());
1100 if (get.getRowOffsetPerColumnFamily() > 0) {
1101 builder.setStoreOffset(get.getRowOffsetPerColumnFamily());
1103 if (get.isCheckExistenceOnly()){
1104 builder.setExistenceOnly(true);
1106 if (get.getConsistency() != null && get.getConsistency() != Consistency.STRONG) {
1107 builder.setConsistency(toConsistency(get.getConsistency()));
1110 Boolean loadColumnFamiliesOnDemand = get.getLoadColumnFamiliesOnDemandValue();
1111 if (loadColumnFamiliesOnDemand != null) {
1112 builder.setLoadColumnFamiliesOnDemand(loadColumnFamiliesOnDemand);
1115 return builder.build();
1118 public static MutationProto toMutation(final MutationType type, final Mutation mutation)
1119 throws IOException {
1120 return toMutation(type, mutation, HConstants.NO_NONCE);
1124 * Create a protocol buffer Mutate based on a client Mutation
1126 * @param type
1127 * @param mutation
1128 * @return a protobuf'd Mutation
1129 * @throws IOException
1131 public static MutationProto toMutation(final MutationType type, final Mutation mutation,
1132 final long nonce) throws IOException {
1133 return toMutation(type, mutation, MutationProto.newBuilder(), nonce);
1136 public static MutationProto toMutation(final MutationType type, final Mutation mutation,
1137 MutationProto.Builder builder) throws IOException {
1138 return toMutation(type, mutation, builder, HConstants.NO_NONCE);
1141 public static MutationProto toMutation(final MutationType type, final Mutation mutation,
1142 MutationProto.Builder builder, long nonce)
1143 throws IOException {
1144 builder = getMutationBuilderAndSetCommonFields(type, mutation, builder);
1145 if (nonce != HConstants.NO_NONCE) {
1146 builder.setNonce(nonce);
1148 if (type == MutationType.INCREMENT) {
1149 builder.setTimeRange(toTimeRange(((Increment) mutation).getTimeRange()));
1151 if (type == MutationType.APPEND) {
1152 builder.setTimeRange(toTimeRange(((Append) mutation).getTimeRange()));
1154 ColumnValue.Builder columnBuilder = ColumnValue.newBuilder();
1155 QualifierValue.Builder valueBuilder = QualifierValue.newBuilder();
1156 for (Map.Entry<byte[],List<Cell>> family: mutation.getFamilyCellMap().entrySet()) {
1157 columnBuilder.clear();
1158 columnBuilder.setFamily(ByteStringer.wrap(family.getKey()));
1159 for (Cell cell: family.getValue()) {
1160 valueBuilder.clear();
1161 valueBuilder.setQualifier(ByteStringer.wrap(
1162 cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
1163 valueBuilder.setValue(ByteStringer.wrap(
1164 cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
1165 valueBuilder.setTimestamp(cell.getTimestamp());
1166 if (type == MutationType.DELETE || (type == MutationType.PUT && CellUtil.isDelete(cell))) {
1167 KeyValue.Type keyValueType = KeyValue.Type.codeToType(cell.getTypeByte());
1168 valueBuilder.setDeleteType(toDeleteType(keyValueType));
1170 columnBuilder.addQualifierValue(valueBuilder.build());
1172 builder.addColumnValue(columnBuilder.build());
1174 return builder.build();
1178 * Create a protocol buffer MutationProto based on a client Mutation. Does NOT include data.
1179 * Understanding is that the Cell will be transported other than via protobuf.
1180 * @param type
1181 * @param mutation
1182 * @param builder
1183 * @return a protobuf'd Mutation
1184 * @throws IOException
1186 public static MutationProto toMutationNoData(final MutationType type, final Mutation mutation,
1187 final MutationProto.Builder builder) throws IOException {
1188 return toMutationNoData(type, mutation, builder, HConstants.NO_NONCE);
1192 * Create a protocol buffer MutationProto based on a client Mutation. Does NOT include data.
1193 * Understanding is that the Cell will be transported other than via protobuf.
1194 * @param type
1195 * @param mutation
1196 * @return a protobuf'd Mutation
1197 * @throws IOException
1199 public static MutationProto toMutationNoData(final MutationType type, final Mutation mutation)
1200 throws IOException {
1201 MutationProto.Builder builder = MutationProto.newBuilder();
1202 return toMutationNoData(type, mutation, builder);
1205 public static MutationProto toMutationNoData(final MutationType type, final Mutation mutation,
1206 final MutationProto.Builder builder, long nonce) throws IOException {
1207 getMutationBuilderAndSetCommonFields(type, mutation, builder);
1208 builder.setAssociatedCellCount(mutation.size());
1209 if (mutation instanceof Increment) {
1210 builder.setTimeRange(toTimeRange(((Increment)mutation).getTimeRange()));
1212 if (mutation instanceof Append) {
1213 builder.setTimeRange(toTimeRange(((Append)mutation).getTimeRange()));
1215 if (nonce != HConstants.NO_NONCE) {
1216 builder.setNonce(nonce);
1218 return builder.build();
1222 * Code shared by {@link #toMutation(MutationType, Mutation)} and
1223 * {@link #toMutationNoData(MutationType, Mutation)}
1224 * @param type
1225 * @param mutation
1226 * @return A partly-filled out protobuf'd Mutation.
1228 private static MutationProto.Builder getMutationBuilderAndSetCommonFields(final MutationType type,
1229 final Mutation mutation, MutationProto.Builder builder) {
1230 builder.setRow(ByteStringer.wrap(mutation.getRow()));
1231 builder.setMutateType(type);
1232 builder.setDurability(toDurability(mutation.getDurability()));
1233 builder.setTimestamp(mutation.getTimestamp());
1234 Map<String, byte[]> attributes = mutation.getAttributesMap();
1235 if (!attributes.isEmpty()) {
1236 NameBytesPair.Builder attributeBuilder = NameBytesPair.newBuilder();
1237 for (Map.Entry<String, byte[]> attribute: attributes.entrySet()) {
1238 attributeBuilder.setName(attribute.getKey());
1239 attributeBuilder.setValue(ByteStringer.wrap(attribute.getValue()));
1240 builder.addAttribute(attributeBuilder.build());
1243 return builder;
1247 * Convert a client Result to a protocol buffer Result
1249 * @param result the client Result to convert
1250 * @return the converted protocol buffer Result
1252 public static ClientProtos.Result toResult(final Result result) {
1253 if (result.getExists() != null) {
1254 return toResult(result.getExists(), result.isStale());
1257 Cell[] cells = result.rawCells();
1258 if (cells == null || cells.length == 0) {
1259 return result.isStale() ? EMPTY_RESULT_PB_STALE : EMPTY_RESULT_PB;
1262 ClientProtos.Result.Builder builder = ClientProtos.Result.newBuilder();
1263 for (Cell c : cells) {
1264 builder.addCell(toCell(c));
1267 builder.setStale(result.isStale());
1268 builder.setPartial(result.mayHaveMoreCellsInRow());
1270 return builder.build();
1274 * Convert a client Result to a protocol buffer Result
1276 * @param existence the client existence to send
1277 * @return the converted protocol buffer Result
1279 public static ClientProtos.Result toResult(final boolean existence, boolean stale) {
1280 if (stale){
1281 return existence ? EMPTY_RESULT_PB_EXISTS_TRUE_STALE : EMPTY_RESULT_PB_EXISTS_FALSE_STALE;
1282 } else {
1283 return existence ? EMPTY_RESULT_PB_EXISTS_TRUE : EMPTY_RESULT_PB_EXISTS_FALSE;
1288 * Convert a client Result to a protocol buffer Result.
1289 * The pb Result does not include the Cell data. That is for transport otherwise.
1291 * @param result the client Result to convert
1292 * @return the converted protocol buffer Result
1294 public static ClientProtos.Result toResultNoData(final Result result) {
1295 if (result.getExists() != null) return toResult(result.getExists(), result.isStale());
1296 int size = result.size();
1297 if (size == 0) return result.isStale() ? EMPTY_RESULT_PB_STALE : EMPTY_RESULT_PB;
1298 ClientProtos.Result.Builder builder = ClientProtos.Result.newBuilder();
1299 builder.setAssociatedCellCount(size);
1300 builder.setStale(result.isStale());
1301 return builder.build();
1305 * Convert a protocol buffer Result to a client Result
1307 * @param proto the protocol buffer Result to convert
1308 * @return the converted client Result
1310 public static Result toResult(final ClientProtos.Result proto) {
1311 if (proto.hasExists()) {
1312 if (proto.getStale()) {
1313 return proto.getExists() ? EMPTY_RESULT_EXISTS_TRUE_STALE :EMPTY_RESULT_EXISTS_FALSE_STALE;
1315 return proto.getExists() ? EMPTY_RESULT_EXISTS_TRUE : EMPTY_RESULT_EXISTS_FALSE;
1318 List<CellProtos.Cell> values = proto.getCellList();
1319 if (values.isEmpty()){
1320 return proto.getStale() ? EMPTY_RESULT_STALE : EMPTY_RESULT;
1323 List<Cell> cells = new ArrayList<>(values.size());
1324 ExtendedCellBuilder builder = ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY);
1325 for (CellProtos.Cell c : values) {
1326 cells.add(toCell(builder, c));
1328 return Result.create(cells, null, proto.getStale(), proto.getPartial());
1332 * Convert a protocol buffer Result to a client Result
1334 * @param proto the protocol buffer Result to convert
1335 * @param scanner Optional cell scanner.
1336 * @return the converted client Result
1337 * @throws IOException
1339 public static Result toResult(final ClientProtos.Result proto, final CellScanner scanner)
1340 throws IOException {
1341 List<CellProtos.Cell> values = proto.getCellList();
1343 if (proto.hasExists()) {
1344 if ((values != null && !values.isEmpty()) ||
1345 (proto.hasAssociatedCellCount() && proto.getAssociatedCellCount() > 0)) {
1346 throw new IllegalArgumentException("bad proto: exists with cells is no allowed " + proto);
1348 if (proto.getStale()) {
1349 return proto.getExists() ? EMPTY_RESULT_EXISTS_TRUE_STALE :EMPTY_RESULT_EXISTS_FALSE_STALE;
1351 return proto.getExists() ? EMPTY_RESULT_EXISTS_TRUE : EMPTY_RESULT_EXISTS_FALSE;
1354 // TODO: Unit test that has some Cells in scanner and some in the proto.
1355 List<Cell> cells = null;
1356 if (proto.hasAssociatedCellCount()) {
1357 int count = proto.getAssociatedCellCount();
1358 cells = new ArrayList<>(count + values.size());
1359 for (int i = 0; i < count; i++) {
1360 if (!scanner.advance()) throw new IOException("Failed get " + i + " of " + count);
1361 cells.add(scanner.current());
1365 if (!values.isEmpty()){
1366 if (cells == null) cells = new ArrayList<>(values.size());
1367 ExtendedCellBuilder builder = ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY);
1368 for (CellProtos.Cell c: values) {
1369 cells.add(toCell(builder, c));
1373 return (cells == null || cells.isEmpty())
1374 ? (proto.getStale() ? EMPTY_RESULT_STALE : EMPTY_RESULT)
1375 : Result.create(cells, null, proto.getStale());
1380 * Convert a ByteArrayComparable to a protocol buffer Comparator
1382 * @param comparator the ByteArrayComparable to convert
1383 * @return the converted protocol buffer Comparator
1385 public static ComparatorProtos.Comparator toComparator(ByteArrayComparable comparator) {
1386 ComparatorProtos.Comparator.Builder builder = ComparatorProtos.Comparator.newBuilder();
1387 builder.setName(comparator.getClass().getName());
1388 builder.setSerializedComparator(ByteStringer.wrap(comparator.toByteArray()));
1389 return builder.build();
1393 * Convert a protocol buffer Comparator to a ByteArrayComparable
1395 * @param proto the protocol buffer Comparator to convert
1396 * @return the converted ByteArrayComparable
1398 @SuppressWarnings("unchecked")
1399 public static ByteArrayComparable toComparator(ComparatorProtos.Comparator proto)
1400 throws IOException {
1401 String type = proto.getName();
1402 String funcName = "parseFrom";
1403 byte [] value = proto.getSerializedComparator().toByteArray();
1404 try {
1405 Class<?> c = Class.forName(type, true, ClassLoaderHolder.CLASS_LOADER);
1406 Method parseFrom = c.getMethod(funcName, byte[].class);
1407 if (parseFrom == null) {
1408 throw new IOException("Unable to locate function: " + funcName + " in type: " + type);
1410 return (ByteArrayComparable)parseFrom.invoke(null, value);
1411 } catch (Exception e) {
1412 throw new IOException(e);
1417 * Convert a protocol buffer Filter to a client Filter
1419 * @param proto the protocol buffer Filter to convert
1420 * @return the converted Filter
1422 @SuppressWarnings("unchecked")
1423 public static Filter toFilter(FilterProtos.Filter proto) throws IOException {
1424 String type = proto.getName();
1425 final byte [] value = proto.getSerializedFilter().toByteArray();
1426 String funcName = "parseFrom";
1427 try {
1428 Class<?> c = Class.forName(type, true, ClassLoaderHolder.CLASS_LOADER);
1429 Method parseFrom = c.getMethod(funcName, byte[].class);
1430 if (parseFrom == null) {
1431 throw new IOException("Unable to locate function: " + funcName + " in type: " + type);
1433 return (Filter)parseFrom.invoke(c, value);
1434 } catch (Exception e) {
1435 // Either we couldn't instantiate the method object, or "parseFrom" failed.
1436 // In either case, let's not retry.
1437 throw new DoNotRetryIOException(e);
1442 * Convert a client Filter to a protocol buffer Filter
1444 * @param filter the Filter to convert
1445 * @return the converted protocol buffer Filter
1447 public static FilterProtos.Filter toFilter(Filter filter) throws IOException {
1448 FilterProtos.Filter.Builder builder = FilterProtos.Filter.newBuilder();
1449 builder.setName(filter.getClass().getName());
1450 builder.setSerializedFilter(ByteStringer.wrap(filter.toByteArray()));
1451 return builder.build();
1455 * Convert a delete KeyValue type to protocol buffer DeleteType.
1457 * @param type
1458 * @return protocol buffer DeleteType
1459 * @throws IOException
1461 public static DeleteType toDeleteType(
1462 KeyValue.Type type) throws IOException {
1463 switch (type) {
1464 case Delete:
1465 return DeleteType.DELETE_ONE_VERSION;
1466 case DeleteColumn:
1467 return DeleteType.DELETE_MULTIPLE_VERSIONS;
1468 case DeleteFamily:
1469 return DeleteType.DELETE_FAMILY;
1470 case DeleteFamilyVersion:
1471 return DeleteType.DELETE_FAMILY_VERSION;
1472 default:
1473 throw new IOException("Unknown delete type: " + type);
1478 * Convert a protocol buffer DeleteType to delete KeyValue type.
1480 * @param type The DeleteType
1481 * @return The type.
1482 * @throws IOException
1484 public static KeyValue.Type fromDeleteType(
1485 DeleteType type) throws IOException {
1486 switch (type) {
1487 case DELETE_ONE_VERSION:
1488 return KeyValue.Type.Delete;
1489 case DELETE_MULTIPLE_VERSIONS:
1490 return KeyValue.Type.DeleteColumn;
1491 case DELETE_FAMILY:
1492 return KeyValue.Type.DeleteFamily;
1493 case DELETE_FAMILY_VERSION:
1494 return KeyValue.Type.DeleteFamilyVersion;
1495 default:
1496 throw new IOException("Unknown delete type: " + type);
1501 * Convert a stringified protocol buffer exception Parameter to a Java Exception
1503 * @param parameter the protocol buffer Parameter to convert
1504 * @return the converted Exception
1505 * @throws IOException if failed to deserialize the parameter
1507 @SuppressWarnings("unchecked")
1508 public static Throwable toException(final NameBytesPair parameter) throws IOException {
1509 if (parameter == null || !parameter.hasValue()) return null;
1510 String desc = parameter.getValue().toStringUtf8();
1511 String type = parameter.getName();
1512 try {
1513 Class<? extends Throwable> c =
1514 (Class<? extends Throwable>)Class.forName(type, true, ClassLoaderHolder.CLASS_LOADER);
1515 Constructor<? extends Throwable> cn = null;
1516 try {
1517 cn = c.getDeclaredConstructor(String.class);
1518 return cn.newInstance(desc);
1519 } catch (NoSuchMethodException e) {
1520 // Could be a raw RemoteException. See HBASE-8987.
1521 cn = c.getDeclaredConstructor(String.class, String.class);
1522 return cn.newInstance(type, desc);
1524 } catch (Exception e) {
1525 throw new IOException(e);
1529 // Start helpers for Client
1531 @SuppressWarnings("unchecked")
1532 public static <T extends Service> T newServiceStub(Class<T> service, RpcChannel channel)
1533 throws Exception {
1534 return (T)Methods.call(service, null, "newStub",
1535 new Class[]{ RpcChannel.class }, new Object[]{ channel });
1538 // End helpers for Client
1539 // Start helpers for Admin
1542 * A helper to get the info of a region server using admin protocol.
1543 * @return the server name
1545 public static ServerInfo getServerInfo(final RpcController controller,
1546 final AdminService.BlockingInterface admin)
1547 throws IOException {
1548 GetServerInfoRequest request = buildGetServerInfoRequest();
1549 try {
1550 GetServerInfoResponse response = admin.getServerInfo(controller, request);
1551 return response.getServerInfo();
1552 } catch (ServiceException se) {
1553 throw getRemoteException(se);
1559 * @see #buildGetServerInfoRequest()
1561 private static GetServerInfoRequest GET_SERVER_INFO_REQUEST =
1562 GetServerInfoRequest.newBuilder().build();
1565 * Create a new GetServerInfoRequest
1567 * @return a GetServerInfoRequest
1569 public static GetServerInfoRequest buildGetServerInfoRequest() {
1570 return GET_SERVER_INFO_REQUEST;
1573 public static ScanMetrics toScanMetrics(final byte[] bytes) {
1574 Parser<MapReduceProtos.ScanMetrics> parser = MapReduceProtos.ScanMetrics.PARSER;
1575 MapReduceProtos.ScanMetrics pScanMetrics = null;
1576 try {
1577 pScanMetrics = parser.parseFrom(bytes);
1578 } catch (InvalidProtocolBufferException e) {
1579 //Ignored there are just no key values to add.
1581 ScanMetrics scanMetrics = new ScanMetrics();
1582 if (pScanMetrics != null) {
1583 for (HBaseProtos.NameInt64Pair pair : pScanMetrics.getMetricsList()) {
1584 if (pair.hasName() && pair.hasValue()) {
1585 scanMetrics.setCounter(pair.getName(), pair.getValue());
1589 return scanMetrics;
1593 * Unwraps an exception from a protobuf service into the underlying (expected) IOException.
1594 * This method will <strong>always</strong> throw an exception.
1595 * @param se the {@code ServiceException} instance to convert into an {@code IOException}
1597 public static void toIOException(ServiceException se) throws IOException {
1598 if (se == null) {
1599 throw new NullPointerException("Null service exception passed!");
1602 Throwable cause = se.getCause();
1603 if (cause != null && cause instanceof IOException) {
1604 throw (IOException)cause;
1606 throw new IOException(se);
1609 public static CellProtos.Cell toCell(final Cell kv) {
1610 // Doing this is going to kill us if we do it for all data passed.
1611 // St.Ack 20121205
1612 CellProtos.Cell.Builder kvbuilder = CellProtos.Cell.newBuilder();
1613 kvbuilder.setRow(ByteStringer.wrap(kv.getRowArray(), kv.getRowOffset(),
1614 kv.getRowLength()));
1615 kvbuilder.setFamily(ByteStringer.wrap(kv.getFamilyArray(),
1616 kv.getFamilyOffset(), kv.getFamilyLength()));
1617 kvbuilder.setQualifier(ByteStringer.wrap(kv.getQualifierArray(),
1618 kv.getQualifierOffset(), kv.getQualifierLength()));
1619 kvbuilder.setCellType(CellProtos.CellType.valueOf(kv.getTypeByte()));
1620 kvbuilder.setTimestamp(kv.getTimestamp());
1621 kvbuilder.setValue(ByteStringer.wrap(kv.getValueArray(), kv.getValueOffset(),
1622 kv.getValueLength()));
1623 return kvbuilder.build();
1626 public static Cell toCell(ExtendedCellBuilder cellBuilder, final CellProtos.Cell cell) {
1627 return cellBuilder.clear()
1628 .setRow(cell.getRow().toByteArray())
1629 .setFamily(cell.getFamily().toByteArray())
1630 .setQualifier(cell.getQualifier().toByteArray())
1631 .setTimestamp(cell.getTimestamp())
1632 .setType((byte) cell.getCellType().getNumber())
1633 .setValue(cell.getValue().toByteArray())
1634 .build();
1638 * Print out some subset of a MutationProto rather than all of it and its data
1639 * @param proto Protobuf to print out
1640 * @return Short String of mutation proto
1642 static String toShortString(final MutationProto proto) {
1643 return "row=" + Bytes.toString(proto.getRow().toByteArray()) +
1644 ", type=" + proto.getMutateType().toString();
1647 public static TableName toTableName(HBaseProtos.TableName tableNamePB) {
1648 return TableName.valueOf(tableNamePB.getNamespace().asReadOnlyByteBuffer(),
1649 tableNamePB.getQualifier().asReadOnlyByteBuffer());
1652 public static HBaseProtos.TableName toProtoTableName(TableName tableName) {
1653 return HBaseProtos.TableName.newBuilder()
1654 .setNamespace(ByteStringer.wrap(tableName.getNamespace()))
1655 .setQualifier(ByteStringer.wrap(tableName.getQualifier())).build();
1659 * This version of protobuf's mergeFrom avoids the hard-coded 64MB limit for decoding
1660 * buffers when working with byte arrays
1661 * @param builder current message builder
1662 * @param b byte array
1663 * @throws IOException
1665 public static void mergeFrom(Message.Builder builder, byte[] b) throws IOException {
1666 final CodedInputStream codedInput = CodedInputStream.newInstance(b);
1667 codedInput.setSizeLimit(b.length);
1668 builder.mergeFrom(codedInput);
1669 codedInput.checkLastTagWas(0);
1673 * This version of protobuf's mergeFrom avoids the hard-coded 64MB limit for decoding
1674 * buffers when working with byte arrays
1675 * @param builder current message builder
1676 * @param b byte array
1677 * @param offset
1678 * @param length
1679 * @throws IOException
1681 public static void mergeFrom(Message.Builder builder, byte[] b, int offset, int length)
1682 throws IOException {
1683 final CodedInputStream codedInput = CodedInputStream.newInstance(b, offset, length);
1684 codedInput.setSizeLimit(length);
1685 builder.mergeFrom(codedInput);
1686 codedInput.checkLastTagWas(0);
1689 private static TimeRange protoToTimeRange(HBaseProtos.TimeRange timeRange) throws IOException {
1690 long minStamp = 0;
1691 long maxStamp = Long.MAX_VALUE;
1692 if (timeRange.hasFrom()) {
1693 minStamp = timeRange.getFrom();
1695 if (timeRange.hasTo()) {
1696 maxStamp = timeRange.getTo();
1698 return new TimeRange(minStamp, maxStamp);
1702 * Creates {@link org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.SnapshotDescription.Type}
1703 * from {@link SnapshotType}
1704 * @param type the SnapshotDescription type
1705 * @return the protobuf SnapshotDescription type
1707 public static HBaseProtos.SnapshotDescription.Type
1708 createProtosSnapShotDescType(SnapshotType type) {
1709 return HBaseProtos.SnapshotDescription.Type.valueOf(type.name());
1713 * Convert a byte array to a protocol buffer RegionSpecifier
1715 * @param type the region specifier type
1716 * @param value the region specifier byte array value
1717 * @return a protocol buffer RegionSpecifier
1719 public static RegionSpecifier buildRegionSpecifier(
1720 final RegionSpecifierType type, final byte[] value) {
1721 RegionSpecifier.Builder regionBuilder = RegionSpecifier.newBuilder();
1722 regionBuilder.setValue(ByteStringer.wrap(value));
1723 regionBuilder.setType(type);
1724 return regionBuilder.build();
1728 * Get a ServerName from the passed in data bytes.
1729 * @param data Data with a serialize server name in it; can handle the old style
1730 * servername where servername was host and port. Works too with data that
1731 * begins w/ the pb 'PBUF' magic and that is then followed by a protobuf that
1732 * has a serialized {@link ServerName} in it.
1733 * @return Returns null if <code>data</code> is null else converts passed data
1734 * to a ServerName instance.
1735 * @throws DeserializationException
1737 public static ServerName toServerName(final byte [] data) throws DeserializationException {
1738 if (data == null || data.length <= 0) return null;
1739 if (ProtobufMagic.isPBMagicPrefix(data)) {
1740 int prefixLen = ProtobufMagic.lengthOfPBMagic();
1741 try {
1742 ZooKeeperProtos.Master rss =
1743 ZooKeeperProtos.Master.PARSER.parseFrom(data, prefixLen, data.length - prefixLen);
1744 org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.ServerName sn =
1745 rss.getMaster();
1746 return ServerName.valueOf(sn.getHostName(), sn.getPort(), sn.getStartCode());
1747 } catch (/*InvalidProtocolBufferException*/IOException e) {
1748 // A failed parse of the znode is pretty catastrophic. Rather than loop
1749 // retrying hoping the bad bytes will changes, and rather than change
1750 // the signature on this method to add an IOE which will send ripples all
1751 // over the code base, throw a RuntimeException. This should "never" happen.
1752 // Fail fast if it does.
1753 throw new DeserializationException(e);
1756 // The str returned could be old style -- pre hbase-1502 -- which was
1757 // hostname and port seperated by a colon rather than hostname, port and
1758 // startcode delimited by a ','.
1759 String str = Bytes.toString(data);
1760 int index = str.indexOf(ServerName.SERVERNAME_SEPARATOR);
1761 if (index != -1) {
1762 // Presume its ServerName serialized with versioned bytes.
1763 return ServerName.parseVersionedServerName(data);
1765 // Presume it a hostname:port format.
1766 String hostname = Addressing.parseHostname(str);
1767 int port = Addressing.parsePort(str);
1768 return ServerName.valueOf(hostname, port, -1L);
1771 public static RSGroupInfo toGroupInfo(RSGroupProtos.RSGroupInfo proto) {
1772 RSGroupInfo RSGroupInfo = new RSGroupInfo(proto.getName());
1773 for(HBaseProtos.ServerName el: proto.getServersList()) {
1774 RSGroupInfo.addServer(Address.fromParts(el.getHostName(), el.getPort()));
1776 for(HBaseProtos.TableName pTableName: proto.getTablesList()) {
1777 RSGroupInfo.addTable(ProtobufUtil.toTableName(pTableName));
1779 return RSGroupInfo;
1782 public static HBaseProtos.TimeRange toTimeRange(TimeRange timeRange) {
1783 if (timeRange == null) {
1784 timeRange = TimeRange.allTime();
1786 return HBaseProtos.TimeRange.newBuilder().setFrom(timeRange.getMin())
1787 .setTo(timeRange.getMax())
1788 .build();