1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17 package org.apache.hadoop.hbase.io.encoding;
18
19 import java.io.DataInputStream;
20 import java.io.DataOutputStream;
21 import java.io.IOException;
22 import java.nio.ByteBuffer;
23
24 import org.apache.hadoop.hbase.Cell;
25 import org.apache.hadoop.hbase.CellComparator;
26 import org.apache.hadoop.hbase.CellUtil;
27 import org.apache.hadoop.hbase.HConstants;
28 import org.apache.hadoop.hbase.KeyValue;
29 import org.apache.hadoop.hbase.KeyValue.KVComparator;
30 import org.apache.hadoop.hbase.KeyValue.SamePrefixComparator;
31 import org.apache.hadoop.hbase.KeyValue.Type;
32 import org.apache.hadoop.hbase.KeyValueUtil;
33 import org.apache.hadoop.hbase.SettableSequenceId;
34 import org.apache.hadoop.hbase.classification.InterfaceAudience;
35 import org.apache.hadoop.hbase.io.HeapSize;
36 import org.apache.hadoop.hbase.io.TagCompressionContext;
37 import org.apache.hadoop.hbase.io.hfile.BlockType;
38 import org.apache.hadoop.hbase.io.hfile.HFileContext;
39 import org.apache.hadoop.hbase.io.util.LRUDictionary;
40 import org.apache.hadoop.hbase.util.ByteBufferUtils;
41 import org.apache.hadoop.hbase.util.Bytes;
42 import org.apache.hadoop.hbase.util.ClassSize;
43 import org.apache.hadoop.io.WritableUtils;
44
45
46
47
48 @InterfaceAudience.Private
49 abstract class BufferedDataBlockEncoder implements DataBlockEncoder {
50
51
52
53 private static int INITIAL_KEY_BUFFER_SIZE = 512;
54
55 @Override
56 public ByteBuffer decodeKeyValues(DataInputStream source,
57 HFileBlockDecodingContext blkDecodingCtx) throws IOException {
58 if (blkDecodingCtx.getClass() != HFileBlockDefaultDecodingContext.class) {
59 throw new IOException(this.getClass().getName() + " only accepts "
60 + HFileBlockDefaultDecodingContext.class.getName() + " as the decoding context.");
61 }
62
63 HFileBlockDefaultDecodingContext decodingCtx =
64 (HFileBlockDefaultDecodingContext) blkDecodingCtx;
65 if (decodingCtx.getHFileContext().isIncludesTags()
66 && decodingCtx.getHFileContext().isCompressTags()) {
67 if (decodingCtx.getTagCompressionContext() != null) {
68
69
70 decodingCtx.getTagCompressionContext().clear();
71 } else {
72 try {
73 TagCompressionContext tagCompressionContext = new TagCompressionContext(
74 LRUDictionary.class, Byte.MAX_VALUE);
75 decodingCtx.setTagCompressionContext(tagCompressionContext);
76 } catch (Exception e) {
77 throw new IOException("Failed to initialize TagCompressionContext", e);
78 }
79 }
80 }
81 return internalDecodeKeyValues(source, 0, 0, decodingCtx);
82 }
83
84 protected static class SeekerState implements Cell {
85 protected ByteBuffer currentBuffer;
86 protected TagCompressionContext tagCompressionContext;
87 protected int valueOffset = -1;
88 protected int keyLength;
89 protected int valueLength;
90 protected int lastCommonPrefix;
91 protected int tagsLength = 0;
92 protected int tagsOffset = -1;
93 protected int tagsCompressedLength = 0;
94 protected boolean uncompressTags = true;
95
96
97 protected byte[] keyBuffer = new byte[INITIAL_KEY_BUFFER_SIZE];
98 protected byte[] tagsBuffer = new byte[INITIAL_KEY_BUFFER_SIZE];
99
100 protected long memstoreTS;
101 protected int nextKvOffset;
102 protected KeyValue.KeyOnlyKeyValue currentKey = new KeyValue.KeyOnlyKeyValue();
103
104 protected boolean isValid() {
105 return valueOffset != -1;
106 }
107
108 protected void invalidate() {
109 valueOffset = -1;
110 tagsCompressedLength = 0;
111 currentKey = new KeyValue.KeyOnlyKeyValue();
112 uncompressTags = true;
113 currentBuffer = null;
114 }
115
116 protected void ensureSpaceForKey() {
117 if (keyLength > keyBuffer.length) {
118
119 int newKeyBufferLength = Math.max(keyBuffer.length, 1) * 2;
120 while (keyLength > newKeyBufferLength) {
121 newKeyBufferLength *= 2;
122 }
123 byte[] newKeyBuffer = new byte[newKeyBufferLength];
124 System.arraycopy(keyBuffer, 0, newKeyBuffer, 0, keyBuffer.length);
125 keyBuffer = newKeyBuffer;
126 }
127 }
128
129 protected void ensureSpaceForTags() {
130 if (tagsLength > tagsBuffer.length) {
131
132 int newTagsBufferLength = Math.max(tagsBuffer.length, 1) * 2;
133 while (tagsLength > newTagsBufferLength) {
134 newTagsBufferLength *= 2;
135 }
136 byte[] newTagsBuffer = new byte[newTagsBufferLength];
137 System.arraycopy(tagsBuffer, 0, newTagsBuffer, 0, tagsBuffer.length);
138 tagsBuffer = newTagsBuffer;
139 }
140 }
141
142 protected void setKey(byte[] keyBuffer, long memTS) {
143 currentKey.setKey(keyBuffer, 0, keyLength);
144 memstoreTS = memTS;
145 }
146
147
148
149
150
151
152 protected void copyFromNext(SeekerState nextState) {
153 if (keyBuffer.length != nextState.keyBuffer.length) {
154 keyBuffer = nextState.keyBuffer.clone();
155 } else if (!isValid()) {
156
157
158 System.arraycopy(nextState.keyBuffer, 0, keyBuffer, 0,
159 nextState.keyLength);
160 } else {
161
162 System.arraycopy(nextState.keyBuffer, nextState.lastCommonPrefix,
163 keyBuffer, nextState.lastCommonPrefix, nextState.keyLength
164 - nextState.lastCommonPrefix);
165 }
166 currentKey = nextState.currentKey;
167
168 valueOffset = nextState.valueOffset;
169 keyLength = nextState.keyLength;
170 valueLength = nextState.valueLength;
171 lastCommonPrefix = nextState.lastCommonPrefix;
172 nextKvOffset = nextState.nextKvOffset;
173 memstoreTS = nextState.memstoreTS;
174 currentBuffer = nextState.currentBuffer;
175 tagsOffset = nextState.tagsOffset;
176 tagsLength = nextState.tagsLength;
177 if (nextState.tagCompressionContext != null) {
178 tagCompressionContext = nextState.tagCompressionContext;
179 }
180 }
181
182 @Override
183 public byte[] getRowArray() {
184 return currentKey.getRowArray();
185 }
186
187 @Override
188 public int getRowOffset() {
189 return Bytes.SIZEOF_SHORT;
190 }
191
192 @Override
193 public short getRowLength() {
194 return currentKey.getRowLength();
195 }
196
197 @Override
198 public byte[] getFamilyArray() {
199 return currentKey.getFamilyArray();
200 }
201
202 @Override
203 public int getFamilyOffset() {
204 return currentKey.getFamilyOffset();
205 }
206
207 @Override
208 public byte getFamilyLength() {
209 return currentKey.getFamilyLength();
210 }
211
212 @Override
213 public byte[] getQualifierArray() {
214 return currentKey.getQualifierArray();
215 }
216
217 @Override
218 public int getQualifierOffset() {
219 return currentKey.getQualifierOffset();
220 }
221
222 @Override
223 public int getQualifierLength() {
224 return currentKey.getQualifierLength();
225 }
226
227 @Override
228 public long getTimestamp() {
229 return currentKey.getTimestamp();
230 }
231
232 @Override
233 public byte getTypeByte() {
234 return currentKey.getTypeByte();
235 }
236
237 @Override
238 public long getMvccVersion() {
239 return memstoreTS;
240 }
241
242 @Override
243 public long getSequenceId() {
244 return memstoreTS;
245 }
246
247 @Override
248 public byte[] getValueArray() {
249 return currentBuffer.array();
250 }
251
252 @Override
253 public int getValueOffset() {
254 return currentBuffer.arrayOffset() + valueOffset;
255 }
256
257 @Override
258 public int getValueLength() {
259 return valueLength;
260 }
261
262 @Override
263 public byte[] getTagsArray() {
264 if (tagCompressionContext != null) {
265 return tagsBuffer;
266 }
267 return currentBuffer.array();
268 }
269
270 @Override
271 public int getTagsOffset() {
272 if (tagCompressionContext != null) {
273 return 0;
274 }
275 return currentBuffer.arrayOffset() + tagsOffset;
276 }
277
278 @Override
279 public int getTagsLength() {
280 return tagsLength;
281 }
282
283 @Override
284 @Deprecated
285 public byte[] getValue() {
286 throw new UnsupportedOperationException("getValue() not supported");
287 }
288
289 @Override
290 @Deprecated
291 public byte[] getFamily() {
292 throw new UnsupportedOperationException("getFamily() not supported");
293 }
294
295 @Override
296 @Deprecated
297 public byte[] getQualifier() {
298 throw new UnsupportedOperationException("getQualifier() not supported");
299 }
300
301 @Override
302 @Deprecated
303 public byte[] getRow() {
304 throw new UnsupportedOperationException("getRow() not supported");
305 }
306
307 @Override
308 public String toString() {
309 return KeyValue.keyToString(this.keyBuffer, 0, KeyValueUtil.keyLength(this)) + "/vlen="
310 + getValueLength() + "/seqid=" + memstoreTS;
311 }
312
313 public Cell shallowCopy() {
314 return new ClonedSeekerState(currentBuffer, keyBuffer, currentKey.getRowLength(),
315 currentKey.getFamilyOffset(), currentKey.getFamilyLength(), keyLength,
316 currentKey.getQualifierOffset(), currentKey.getQualifierLength(),
317 currentKey.getTimestamp(), currentKey.getTypeByte(), valueLength, valueOffset,
318 memstoreTS, tagsOffset, tagsLength, tagCompressionContext, tagsBuffer);
319 }
320 }
321
322
323
324
325
326
327
328
329
330
331
332 protected static class ClonedSeekerState implements Cell, HeapSize, SettableSequenceId {
333 private static final long FIXED_OVERHEAD = ClassSize.align(ClassSize.OBJECT
334 + (4 * ClassSize.REFERENCE) + (2 * Bytes.SIZEOF_LONG) + (7 * Bytes.SIZEOF_INT)
335 + (Bytes.SIZEOF_SHORT) + (2 * Bytes.SIZEOF_BYTE) + (2 * ClassSize.ARRAY));
336 private byte[] keyOnlyBuffer;
337 private ByteBuffer currentBuffer;
338 private short rowLength;
339 private int familyOffset;
340 private byte familyLength;
341 private int qualifierOffset;
342 private int qualifierLength;
343 private long timestamp;
344 private byte typeByte;
345 private int valueOffset;
346 private int valueLength;
347 private int tagsLength;
348 private int tagsOffset;
349 private byte[] cloneTagsBuffer;
350 private long seqId;
351 private TagCompressionContext tagCompressionContext;
352
353 protected ClonedSeekerState(ByteBuffer currentBuffer, byte[] keyBuffer, short rowLength,
354 int familyOffset, byte familyLength, int keyLength, int qualOffset, int qualLength,
355 long timeStamp, byte typeByte, int valueLen, int valueOffset, long seqId,
356 int tagsOffset, int tagsLength, TagCompressionContext tagCompressionContext,
357 byte[] tagsBuffer) {
358 this.currentBuffer = currentBuffer;
359 keyOnlyBuffer = new byte[keyLength];
360 this.tagCompressionContext = tagCompressionContext;
361 this.rowLength = rowLength;
362 this.familyOffset = familyOffset;
363 this.familyLength = familyLength;
364 this.qualifierOffset = qualOffset;
365 this.qualifierLength = qualLength;
366 this.timestamp = timeStamp;
367 this.typeByte = typeByte;
368 this.valueLength = valueLen;
369 this.valueOffset = valueOffset;
370 this.tagsOffset = tagsOffset;
371 this.tagsLength = tagsLength;
372 System.arraycopy(keyBuffer, 0, keyOnlyBuffer, 0, keyLength);
373 if (tagCompressionContext != null) {
374 this.cloneTagsBuffer = new byte[tagsLength];
375 System.arraycopy(tagsBuffer, 0, this.cloneTagsBuffer, 0, tagsLength);
376 }
377 setSequenceId(seqId);
378 }
379
380 @Override
381 public byte[] getRowArray() {
382 return keyOnlyBuffer;
383 }
384
385 @Override
386 public byte[] getFamilyArray() {
387 return keyOnlyBuffer;
388 }
389
390 @Override
391 public byte[] getQualifierArray() {
392 return keyOnlyBuffer;
393 }
394
395 @Override
396 public int getRowOffset() {
397 return Bytes.SIZEOF_SHORT;
398 }
399
400 @Override
401 public short getRowLength() {
402 return rowLength;
403 }
404
405 @Override
406 public int getFamilyOffset() {
407 return familyOffset;
408 }
409
410 @Override
411 public byte getFamilyLength() {
412 return familyLength;
413 }
414
415 @Override
416 public int getQualifierOffset() {
417 return qualifierOffset;
418 }
419
420 @Override
421 public int getQualifierLength() {
422 return qualifierLength;
423 }
424
425 @Override
426 public long getTimestamp() {
427 return timestamp;
428 }
429
430 @Override
431 public byte getTypeByte() {
432 return typeByte;
433 }
434
435 @Override
436 @Deprecated
437 public long getMvccVersion() {
438 return getSequenceId();
439 }
440
441 @Override
442 public long getSequenceId() {
443 return seqId;
444 }
445
446 @Override
447 public byte[] getValueArray() {
448 return currentBuffer.array();
449 }
450
451 @Override
452 public int getValueOffset() {
453 return currentBuffer.arrayOffset() + valueOffset;
454 }
455
456 @Override
457 public int getValueLength() {
458 return valueLength;
459 }
460
461 @Override
462 public byte[] getTagsArray() {
463 if (tagCompressionContext != null) {
464 return cloneTagsBuffer;
465 }
466 return currentBuffer.array();
467 }
468
469 @Override
470 public int getTagsOffset() {
471 if (tagCompressionContext != null) {
472 return 0;
473 }
474 return currentBuffer.arrayOffset() + tagsOffset;
475 }
476
477 @Override
478 public int getTagsLength() {
479 return tagsLength;
480 }
481
482 @Override
483 @Deprecated
484 public byte[] getValue() {
485 return CellUtil.cloneValue(this);
486 }
487
488 @Override
489 @Deprecated
490 public byte[] getFamily() {
491 return CellUtil.cloneFamily(this);
492 }
493
494 @Override
495 @Deprecated
496 public byte[] getQualifier() {
497 return CellUtil.cloneQualifier(this);
498 }
499
500 @Override
501 @Deprecated
502 public byte[] getRow() {
503 return CellUtil.cloneRow(this);
504 }
505
506 @Override
507 public String toString() {
508 return KeyValue.keyToString(this.keyOnlyBuffer, 0, KeyValueUtil.keyLength(this)) + "/vlen="
509 + getValueLength() + "/seqid=" + seqId;
510 }
511
512 @Override
513 public void setSequenceId(long seqId) {
514 this.seqId = seqId;
515 }
516
517 @Override
518 public long heapSize() {
519 return FIXED_OVERHEAD + rowLength + familyLength + qualifierLength + valueLength + tagsLength;
520 }
521 }
522
523 protected abstract static class
524 BufferedEncodedSeeker<STATE extends SeekerState>
525 implements EncodedSeeker {
526 protected HFileBlockDecodingContext decodingCtx;
527 protected final KVComparator comparator;
528 protected final SamePrefixComparator<byte[]> samePrefixComparator;
529 protected ByteBuffer currentBuffer;
530 protected STATE current = createSeekerState();
531 protected STATE previous = createSeekerState();
532 protected TagCompressionContext tagCompressionContext = null;
533
534 public BufferedEncodedSeeker(KVComparator comparator,
535 HFileBlockDecodingContext decodingCtx) {
536 this.comparator = comparator;
537 this.samePrefixComparator = comparator;
538 this.decodingCtx = decodingCtx;
539 if (decodingCtx.getHFileContext().isCompressTags()) {
540 try {
541 tagCompressionContext = new TagCompressionContext(LRUDictionary.class, Byte.MAX_VALUE);
542 } catch (Exception e) {
543 throw new RuntimeException("Failed to initialize TagCompressionContext", e);
544 }
545 }
546 }
547
548 protected boolean includesMvcc() {
549 return this.decodingCtx.getHFileContext().isIncludesMvcc();
550 }
551
552 protected boolean includesTags() {
553 return this.decodingCtx.getHFileContext().isIncludesTags();
554 }
555
556 @Override
557 public int compareKey(KVComparator comparator, byte[] key, int offset, int length) {
558 return comparator.compareFlatKey(key, offset, length,
559 current.keyBuffer, 0, current.keyLength);
560 }
561
562 @Override
563 public int compareKey(KVComparator comparator, Cell key) {
564 return comparator.compareOnlyKeyPortion(key,
565 new KeyValue.KeyOnlyKeyValue(current.keyBuffer, 0, current.keyLength));
566 }
567
568 @Override
569 public void setCurrentBuffer(ByteBuffer buffer) {
570 if (this.tagCompressionContext != null) {
571 this.tagCompressionContext.clear();
572 }
573 currentBuffer = buffer;
574 current.currentBuffer = currentBuffer;
575 if(tagCompressionContext != null) {
576 current.tagCompressionContext = tagCompressionContext;
577 }
578 decodeFirst();
579 current.setKey(current.keyBuffer, current.memstoreTS);
580 previous.invalidate();
581 }
582
583 @Override
584 public ByteBuffer getKeyDeepCopy() {
585 ByteBuffer keyBuffer = ByteBuffer.allocate(current.keyLength);
586 keyBuffer.put(current.keyBuffer, 0, current.keyLength);
587 keyBuffer.rewind();
588 return keyBuffer;
589 }
590
591 @Override
592 public ByteBuffer getValueShallowCopy() {
593 ByteBuffer dup = currentBuffer.duplicate();
594 dup.position(current.valueOffset);
595 dup.limit(current.valueOffset + current.valueLength);
596 return dup.slice();
597 }
598
599 @Override
600 public Cell getKeyValue() {
601 return current.shallowCopy();
602 }
603
604 @Override
605 public void rewind() {
606 currentBuffer.rewind();
607 if (tagCompressionContext != null) {
608 tagCompressionContext.clear();
609 }
610 decodeFirst();
611 current.setKey(current.keyBuffer, current.memstoreTS);
612 previous.invalidate();
613 }
614
615 @Override
616 public boolean next() {
617 if (!currentBuffer.hasRemaining()) {
618 return false;
619 }
620 decodeNext();
621 current.setKey(current.keyBuffer, current.memstoreTS);
622 previous.invalidate();
623 return true;
624 }
625
626 protected void decodeTags() {
627 current.tagsLength = ByteBufferUtils.readCompressedInt(currentBuffer);
628 if (tagCompressionContext != null) {
629 if (current.uncompressTags) {
630
631 current.ensureSpaceForTags();
632 try {
633 current.tagsCompressedLength = tagCompressionContext.uncompressTags(currentBuffer,
634 current.tagsBuffer, 0, current.tagsLength);
635 } catch (IOException e) {
636 throw new RuntimeException("Exception while uncompressing tags", e);
637 }
638 } else {
639 ByteBufferUtils.skip(currentBuffer, current.tagsCompressedLength);
640 current.uncompressTags = true;
641 }
642 current.tagsOffset = -1;
643 } else {
644
645
646 current.tagsOffset = currentBuffer.position();
647 ByteBufferUtils.skip(currentBuffer, current.tagsLength);
648 }
649 }
650
651 @Override
652 public int seekToKeyInBlock(byte[] key, int offset, int length, boolean seekBefore) {
653 return seekToKeyInBlock(new KeyValue.KeyOnlyKeyValue(key, offset, length), seekBefore);
654 }
655
656 @Override
657 public int seekToKeyInBlock(Cell seekCell, boolean seekBefore) {
658 int rowCommonPrefix = 0;
659 int familyCommonPrefix = 0;
660 int qualCommonPrefix = 0;
661 previous.invalidate();
662 KeyValue.KeyOnlyKeyValue currentCell = new KeyValue.KeyOnlyKeyValue();
663 do {
664 int comp;
665 if (samePrefixComparator != null) {
666 currentCell.setKey(current.keyBuffer, 0, current.keyLength);
667 if (current.lastCommonPrefix != 0) {
668
669
670
671
672
673 rowCommonPrefix = Math.min(rowCommonPrefix, current.lastCommonPrefix - 2);
674 }
675 if (current.lastCommonPrefix <= 2) {
676 rowCommonPrefix = 0;
677 }
678 rowCommonPrefix += CellComparator.findCommonPrefixInRowPart(seekCell, currentCell,
679 rowCommonPrefix);
680 comp = CellComparator.compareCommonRowPrefix(seekCell, currentCell, rowCommonPrefix);
681 if (comp == 0) {
682 comp = compareTypeBytes(seekCell, currentCell);
683 if (comp == 0) {
684
685 familyCommonPrefix = Math.max(
686 0,
687 Math.min(familyCommonPrefix,
688 current.lastCommonPrefix - (3 + currentCell.getRowLength())));
689 familyCommonPrefix += CellComparator.findCommonPrefixInFamilyPart(seekCell,
690 currentCell, familyCommonPrefix);
691 comp = CellComparator.compareCommonFamilyPrefix(seekCell, currentCell,
692 familyCommonPrefix);
693 if (comp == 0) {
694
695
696 qualCommonPrefix = Math.max(
697 0,
698 Math.min(
699 qualCommonPrefix,
700 current.lastCommonPrefix
701 - (3 + currentCell.getRowLength() + currentCell.getFamilyLength())));
702 qualCommonPrefix += CellComparator.findCommonPrefixInQualifierPart(seekCell,
703 currentCell, qualCommonPrefix);
704 comp = CellComparator.compareCommonQualifierPrefix(seekCell, currentCell,
705 qualCommonPrefix);
706 if (comp == 0) {
707 comp = CellComparator.compareTimestamps(seekCell, currentCell);
708 if (comp == 0) {
709
710
711
712
713
714
715
716
717 comp = (0xff & currentCell.getTypeByte()) - (0xff & seekCell.getTypeByte());
718 }
719 }
720 }
721 }
722 }
723 } else {
724 Cell r = new KeyValue.KeyOnlyKeyValue(current.keyBuffer, 0, current.keyLength);
725 comp = comparator.compareOnlyKeyPortion(seekCell, r);
726 }
727 if (comp == 0) {
728 if (seekBefore) {
729 if (!previous.isValid()) {
730
731
732 throw new IllegalStateException("Cannot seekBefore if "
733 + "positioned at the first key in the block: key="
734 + Bytes.toStringBinary(seekCell.getRowArray()));
735 }
736 moveToPrevious();
737 return 1;
738 }
739 return 0;
740 }
741
742 if (comp < 0) {
743 if (previous.isValid()) {
744 moveToPrevious();
745 } else {
746 return HConstants.INDEX_KEY_MAGIC;
747 }
748 return 1;
749 }
750
751
752 if (currentBuffer.hasRemaining()) {
753 previous.copyFromNext(current);
754 decodeNext();
755 current.setKey(current.keyBuffer, current.memstoreTS);
756 } else {
757 break;
758 }
759 } while (true);
760
761
762 return 1;
763 }
764
765 private int compareTypeBytes(Cell key, Cell right) {
766 if (key.getFamilyLength() + key.getQualifierLength() == 0
767 && key.getTypeByte() == Type.Minimum.getCode()) {
768
769 return 1;
770 }
771 if (right.getFamilyLength() + right.getQualifierLength() == 0
772 && right.getTypeByte() == Type.Minimum.getCode()) {
773 return -1;
774 }
775 return 0;
776 }
777
778
779 private void moveToPrevious() {
780 if (!previous.isValid()) {
781 throw new IllegalStateException(
782 "Can move back only once and not in first key in the block.");
783 }
784
785 STATE tmp = previous;
786 previous = current;
787 current = tmp;
788
789
790 currentBuffer.position(current.nextKvOffset);
791
792
793
794
795
796
797 current.tagsBuffer = previous.tagsBuffer;
798 current.tagsCompressedLength = previous.tagsCompressedLength;
799 current.uncompressTags = false;
800 current.setKey(current.keyBuffer, current.memstoreTS);
801 previous.invalidate();
802 }
803
804 @SuppressWarnings("unchecked")
805 protected STATE createSeekerState() {
806
807
808 return (STATE) new SeekerState();
809 }
810
811 abstract protected void decodeFirst();
812 abstract protected void decodeNext();
813 }
814
815
816
817
818
819
820
821
822 protected final int afterEncodingKeyValue(Cell cell, DataOutputStream out,
823 HFileBlockDefaultEncodingContext encodingCtx) throws IOException {
824 int size = 0;
825 if (encodingCtx.getHFileContext().isIncludesTags()) {
826 int tagsLength = cell.getTagsLength();
827 ByteBufferUtils.putCompressedInt(out, tagsLength);
828
829 if (tagsLength > 0) {
830 TagCompressionContext tagCompressionContext = encodingCtx.getTagCompressionContext();
831
832
833 if (tagCompressionContext != null) {
834 tagCompressionContext
835 .compressTags(out, cell.getTagsArray(), cell.getTagsOffset(), tagsLength);
836 } else {
837 out.write(cell.getTagsArray(), cell.getTagsOffset(), tagsLength);
838 }
839 }
840 size += tagsLength + KeyValue.TAGS_LENGTH_SIZE;
841 }
842 if (encodingCtx.getHFileContext().isIncludesMvcc()) {
843
844 long memstoreTS = cell.getSequenceId();
845 WritableUtils.writeVLong(out, memstoreTS);
846
847
848 size += WritableUtils.getVIntSize(memstoreTS);
849 }
850 return size;
851 }
852
853 protected final void afterDecodingKeyValue(DataInputStream source,
854 ByteBuffer dest, HFileBlockDefaultDecodingContext decodingCtx) throws IOException {
855 if (decodingCtx.getHFileContext().isIncludesTags()) {
856 int tagsLength = ByteBufferUtils.readCompressedInt(source);
857
858 dest.put((byte) ((tagsLength >> 8) & 0xff));
859 dest.put((byte) (tagsLength & 0xff));
860 if (tagsLength > 0) {
861 TagCompressionContext tagCompressionContext = decodingCtx.getTagCompressionContext();
862
863
864 if (tagCompressionContext != null) {
865 tagCompressionContext.uncompressTags(source, dest, tagsLength);
866 } else {
867 ByteBufferUtils.copyFromStreamToBuffer(dest, source, tagsLength);
868 }
869 }
870 }
871 if (decodingCtx.getHFileContext().isIncludesMvcc()) {
872 long memstoreTS = -1;
873 try {
874
875
876 memstoreTS = WritableUtils.readVLong(source);
877 ByteBufferUtils.writeVLong(dest, memstoreTS);
878 } catch (IOException ex) {
879 throw new RuntimeException("Unable to copy memstore timestamp " +
880 memstoreTS + " after decoding a key/value");
881 }
882 }
883 }
884
885 @Override
886 public HFileBlockEncodingContext newDataBlockEncodingContext(DataBlockEncoding encoding,
887 byte[] header, HFileContext meta) {
888 return new HFileBlockDefaultEncodingContext(encoding, header, meta);
889 }
890
891 @Override
892 public HFileBlockDecodingContext newDataBlockDecodingContext(HFileContext meta) {
893 return new HFileBlockDefaultDecodingContext(meta);
894 }
895
896 protected abstract ByteBuffer internalDecodeKeyValues(DataInputStream source,
897 int allocateHeaderLength, int skipLastBytes, HFileBlockDefaultDecodingContext decodingCtx)
898 throws IOException;
899
900
901
902
903
904
905
906
907 protected static void ensureSpace(ByteBuffer out, int length)
908 throws EncoderBufferTooSmallException {
909 if (out.position() + length > out.limit()) {
910 throw new EncoderBufferTooSmallException(
911 "Buffer position=" + out.position() +
912 ", buffer limit=" + out.limit() +
913 ", length to be written=" + length);
914 }
915 }
916
917 @Override
918 public void startBlockEncoding(HFileBlockEncodingContext blkEncodingCtx, DataOutputStream out)
919 throws IOException {
920 if (blkEncodingCtx.getClass() != HFileBlockDefaultEncodingContext.class) {
921 throw new IOException (this.getClass().getName() + " only accepts "
922 + HFileBlockDefaultEncodingContext.class.getName() + " as the " +
923 "encoding context.");
924 }
925
926 HFileBlockDefaultEncodingContext encodingCtx =
927 (HFileBlockDefaultEncodingContext) blkEncodingCtx;
928 encodingCtx.prepareEncoding(out);
929 if (encodingCtx.getHFileContext().isIncludesTags()
930 && encodingCtx.getHFileContext().isCompressTags()) {
931 if (encodingCtx.getTagCompressionContext() != null) {
932
933
934 encodingCtx.getTagCompressionContext().clear();
935 } else {
936 try {
937 TagCompressionContext tagCompressionContext = new TagCompressionContext(
938 LRUDictionary.class, Byte.MAX_VALUE);
939 encodingCtx.setTagCompressionContext(tagCompressionContext);
940 } catch (Exception e) {
941 throw new IOException("Failed to initialize TagCompressionContext", e);
942 }
943 }
944 }
945 ByteBufferUtils.putInt(out, 0);
946 blkEncodingCtx.setEncodingState(new BufferedDataBlockEncodingState());
947 }
948
949 private static class BufferedDataBlockEncodingState extends EncodingState {
950 int unencodedDataSizeWritten = 0;
951 }
952
953 @Override
954 public int encode(Cell cell, HFileBlockEncodingContext encodingCtx, DataOutputStream out)
955 throws IOException {
956 BufferedDataBlockEncodingState state = (BufferedDataBlockEncodingState) encodingCtx
957 .getEncodingState();
958 int encodedKvSize = internalEncode(cell, (HFileBlockDefaultEncodingContext) encodingCtx, out);
959 state.unencodedDataSizeWritten += encodedKvSize;
960 return encodedKvSize;
961 }
962
963 public abstract int internalEncode(Cell cell, HFileBlockDefaultEncodingContext encodingCtx,
964 DataOutputStream out) throws IOException;
965
966 @Override
967 public void endBlockEncoding(HFileBlockEncodingContext encodingCtx, DataOutputStream out,
968 byte[] uncompressedBytesWithHeader) throws IOException {
969 BufferedDataBlockEncodingState state = (BufferedDataBlockEncodingState) encodingCtx
970 .getEncodingState();
971
972 Bytes.putInt(uncompressedBytesWithHeader,
973 HConstants.HFILEBLOCK_HEADER_SIZE + DataBlockEncoding.ID_SIZE, state.unencodedDataSizeWritten
974 );
975 if (encodingCtx.getDataBlockEncoding() != DataBlockEncoding.NONE) {
976 encodingCtx.postEncoding(BlockType.ENCODED_DATA);
977 } else {
978 encodingCtx.postEncoding(BlockType.DATA);
979 }
980 }
981 }