001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.io.encoding;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertTrue;
022import static org.junit.Assert.fail;
023
024import java.io.ByteArrayInputStream;
025import java.io.DataInputStream;
026import java.io.DataOutputStream;
027import java.io.IOException;
028import java.nio.ByteBuffer;
029import java.util.ArrayList;
030import java.util.Collection;
031import java.util.List;
032import java.util.Random;
033import org.apache.hadoop.hbase.ArrayBackedTag;
034import org.apache.hadoop.hbase.Cell;
035import org.apache.hadoop.hbase.CellComparatorImpl;
036import org.apache.hadoop.hbase.CellUtil;
037import org.apache.hadoop.hbase.HBaseClassTestRule;
038import org.apache.hadoop.hbase.HBaseTestingUtility;
039import org.apache.hadoop.hbase.HConstants;
040import org.apache.hadoop.hbase.KeyValue;
041import org.apache.hadoop.hbase.KeyValue.Type;
042import org.apache.hadoop.hbase.PrivateCellUtil;
043import org.apache.hadoop.hbase.Tag;
044import org.apache.hadoop.hbase.io.ByteArrayOutputStream;
045import org.apache.hadoop.hbase.io.compress.Compression;
046import org.apache.hadoop.hbase.io.hfile.HFileContext;
047import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
048import org.apache.hadoop.hbase.nio.SingleByteBuff;
049import org.apache.hadoop.hbase.testclassification.IOTests;
050import org.apache.hadoop.hbase.testclassification.LargeTests;
051import org.apache.hadoop.hbase.util.Bytes;
052import org.apache.hadoop.hbase.util.RedundantKVGenerator;
053import org.junit.ClassRule;
054import org.junit.Test;
055import org.junit.experimental.categories.Category;
056import org.junit.runner.RunWith;
057import org.junit.runners.Parameterized;
058import org.junit.runners.Parameterized.Parameters;
059import org.slf4j.Logger;
060import org.slf4j.LoggerFactory;
061
062/**
063 * Test all of the data block encoding algorithms for correctness. Most of the
064 * class generate data which will test different branches in code.
065 */
066@Category({IOTests.class, LargeTests.class})
067@RunWith(Parameterized.class)
068public class TestDataBlockEncoders {
069
070  @ClassRule
071  public static final HBaseClassTestRule CLASS_RULE =
072      HBaseClassTestRule.forClass(TestDataBlockEncoders.class);
073
074  private static final Logger LOG = LoggerFactory.getLogger(TestDataBlockEncoders.class);
075
076  private static int NUMBER_OF_KV = 10000;
077  private static int NUM_RANDOM_SEEKS = 1000;
078
079  private static int ENCODED_DATA_OFFSET = HConstants.HFILEBLOCK_HEADER_SIZE
080      + DataBlockEncoding.ID_SIZE;
081  static final byte[] HFILEBLOCK_DUMMY_HEADER = new byte[HConstants.HFILEBLOCK_HEADER_SIZE];
082
083  private RedundantKVGenerator generator = new RedundantKVGenerator();
084  private Random randomizer = new Random(42L);
085
086  private final boolean includesMemstoreTS;
087  private final boolean includesTags;
088  private final boolean useOffheapData;
089
090  @Parameters
091  public static Collection<Object[]> parameters() {
092    return HBaseTestingUtility.memStoreTSTagsAndOffheapCombination();
093  }
094
095  public TestDataBlockEncoders(boolean includesMemstoreTS, boolean includesTag,
096      boolean useOffheapData) {
097    this.includesMemstoreTS = includesMemstoreTS;
098    this.includesTags = includesTag;
099    this.useOffheapData = useOffheapData;
100  }
101
102  private HFileBlockEncodingContext getEncodingContext(Compression.Algorithm algo,
103      DataBlockEncoding encoding) {
104    DataBlockEncoder encoder = encoding.getEncoder();
105    HFileContext meta = new HFileContextBuilder()
106                        .withHBaseCheckSum(false)
107                        .withIncludesMvcc(includesMemstoreTS)
108                        .withIncludesTags(includesTags)
109                        .withCompression(algo).build();
110    if (encoder != null) {
111      return encoder.newDataBlockEncodingContext(encoding, HFILEBLOCK_DUMMY_HEADER, meta);
112    } else {
113      return new HFileBlockDefaultEncodingContext(encoding, HFILEBLOCK_DUMMY_HEADER, meta);
114    }
115  }
116
117  /**
118   * Test data block encoding of empty KeyValue.
119   *
120   * @throws IOException
121   *           On test failure.
122   */
123  @Test
124  public void testEmptyKeyValues() throws IOException {
125    List<KeyValue> kvList = new ArrayList<>();
126    byte[] row = new byte[0];
127    byte[] family = new byte[0];
128    byte[] qualifier = new byte[0];
129    byte[] value = new byte[0];
130    if (!includesTags) {
131      kvList.add(new KeyValue(row, family, qualifier, 0L, value));
132      kvList.add(new KeyValue(row, family, qualifier, 0L, value));
133    } else {
134      byte[] metaValue1 = Bytes.toBytes("metaValue1");
135      byte[] metaValue2 = Bytes.toBytes("metaValue2");
136      kvList.add(new KeyValue(row, family, qualifier, 0L, value,
137          new Tag[] { new ArrayBackedTag((byte) 1, metaValue1) }));
138      kvList.add(new KeyValue(row, family, qualifier, 0L, value,
139          new Tag[] { new ArrayBackedTag((byte) 1, metaValue2) }));
140    }
141    testEncodersOnDataset(kvList, includesMemstoreTS, includesTags);
142  }
143
144  /**
145   * Test KeyValues with negative timestamp.
146   *
147   * @throws IOException
148   *           On test failure.
149   */
150  @Test
151  public void testNegativeTimestamps() throws IOException {
152    List<KeyValue> kvList = new ArrayList<>();
153    byte[] row = new byte[0];
154    byte[] family = new byte[0];
155    byte[] qualifier = new byte[0];
156    byte[] value = new byte[0];
157    if (includesTags) {
158      byte[] metaValue1 = Bytes.toBytes("metaValue1");
159      byte[] metaValue2 = Bytes.toBytes("metaValue2");
160      kvList.add(new KeyValue(row, family, qualifier, 0L, value,
161          new Tag[] { new ArrayBackedTag((byte) 1, metaValue1) }));
162      kvList.add(new KeyValue(row, family, qualifier, 0L, value,
163          new Tag[] { new ArrayBackedTag((byte) 1, metaValue2) }));
164    } else {
165      kvList.add(new KeyValue(row, family, qualifier, -1L, Type.Put, value));
166      kvList.add(new KeyValue(row, family, qualifier, -2L, Type.Put, value));
167    }
168    testEncodersOnDataset(kvList, includesMemstoreTS, includesTags);
169  }
170
171
172  /**
173   * Test whether compression -> decompression gives the consistent results on
174   * pseudorandom sample.
175   * @throws IOException On test failure.
176   */
177  @Test
178  public void testExecutionOnSample() throws IOException {
179    List<KeyValue> kvList = generator.generateTestKeyValues(NUMBER_OF_KV, includesTags);
180    testEncodersOnDataset(kvList, includesMemstoreTS, includesTags);
181  }
182
183  /**
184   * Test seeking while file is encoded.
185   */
186  @Test
187  public void testSeekingOnSample() throws IOException {
188    List<KeyValue> sampleKv = generator.generateTestKeyValues(NUMBER_OF_KV, includesTags);
189
190    // create all seekers
191    List<DataBlockEncoder.EncodedSeeker> encodedSeekers = new ArrayList<>();
192    for (DataBlockEncoding encoding : DataBlockEncoding.values()) {
193      LOG.info("Encoding: " + encoding);
194      DataBlockEncoder encoder = encoding.getEncoder();
195      if (encoder == null) {
196        continue;
197      }
198      LOG.info("Encoder: " + encoder);
199      ByteBuffer encodedBuffer = encodeKeyValues(encoding, sampleKv,
200          getEncodingContext(Compression.Algorithm.NONE, encoding), this.useOffheapData);
201      HFileContext meta = new HFileContextBuilder()
202                          .withHBaseCheckSum(false)
203                          .withIncludesMvcc(includesMemstoreTS)
204                          .withIncludesTags(includesTags)
205                          .withCompression(Compression.Algorithm.NONE)
206                          .build();
207      DataBlockEncoder.EncodedSeeker seeker = encoder.createSeeker(CellComparatorImpl.COMPARATOR,
208          encoder.newDataBlockDecodingContext(meta));
209      seeker.setCurrentBuffer(new SingleByteBuff(encodedBuffer));
210      encodedSeekers.add(seeker);
211    }
212    LOG.info("Testing it!");
213    // test it!
214    // try a few random seeks
215    for (boolean seekBefore : new boolean[] { false, true }) {
216      for (int i = 0; i < NUM_RANDOM_SEEKS; ++i) {
217        int keyValueId;
218        if (!seekBefore) {
219          keyValueId = randomizer.nextInt(sampleKv.size());
220        } else {
221          keyValueId = randomizer.nextInt(sampleKv.size() - 1) + 1;
222        }
223
224        KeyValue keyValue = sampleKv.get(keyValueId);
225        checkSeekingConsistency(encodedSeekers, seekBefore, keyValue);
226      }
227    }
228
229    // check edge cases
230    LOG.info("Checking edge cases");
231    checkSeekingConsistency(encodedSeekers, false, sampleKv.get(0));
232    for (boolean seekBefore : new boolean[] { false, true }) {
233      checkSeekingConsistency(encodedSeekers, seekBefore, sampleKv.get(sampleKv.size() - 1));
234      KeyValue midKv = sampleKv.get(sampleKv.size() / 2);
235      Cell lastMidKv =PrivateCellUtil.createLastOnRowCol(midKv);
236      checkSeekingConsistency(encodedSeekers, seekBefore, lastMidKv);
237    }
238    LOG.info("Done");
239  }
240
241  static ByteBuffer encodeKeyValues(DataBlockEncoding encoding, List<KeyValue> kvs,
242      HFileBlockEncodingContext encodingContext, boolean useOffheapData) throws IOException {
243    DataBlockEncoder encoder = encoding.getEncoder();
244    ByteArrayOutputStream baos = new ByteArrayOutputStream();
245    baos.write(HFILEBLOCK_DUMMY_HEADER);
246    DataOutputStream dos = new DataOutputStream(baos);
247    encoder.startBlockEncoding(encodingContext, dos);
248    for (KeyValue kv : kvs) {
249      encoder.encode(kv, encodingContext, dos);
250    }
251    encoder.endBlockEncoding(encodingContext, dos, baos.getBuffer());
252    byte[] encodedData = new byte[baos.size() - ENCODED_DATA_OFFSET];
253    System.arraycopy(baos.toByteArray(), ENCODED_DATA_OFFSET, encodedData, 0, encodedData.length);
254    if (useOffheapData) {
255      ByteBuffer bb = ByteBuffer.allocateDirect(encodedData.length);
256      bb.put(encodedData);
257      bb.rewind();
258      return bb;
259    }
260    return ByteBuffer.wrap(encodedData);
261  }
262
263  @Test
264  public void testNextOnSample() throws IOException {
265    List<KeyValue> sampleKv = generator.generateTestKeyValues(NUMBER_OF_KV, includesTags);
266
267    for (DataBlockEncoding encoding : DataBlockEncoding.values()) {
268      if (encoding.getEncoder() == null) {
269        continue;
270      }
271      DataBlockEncoder encoder = encoding.getEncoder();
272      ByteBuffer encodedBuffer = encodeKeyValues(encoding, sampleKv,
273          getEncodingContext(Compression.Algorithm.NONE, encoding), this.useOffheapData);
274      HFileContext meta = new HFileContextBuilder()
275                          .withHBaseCheckSum(false)
276                          .withIncludesMvcc(includesMemstoreTS)
277                          .withIncludesTags(includesTags)
278                          .withCompression(Compression.Algorithm.NONE)
279                          .build();
280      DataBlockEncoder.EncodedSeeker seeker = encoder.createSeeker(CellComparatorImpl.COMPARATOR,
281          encoder.newDataBlockDecodingContext(meta));
282      seeker.setCurrentBuffer(new SingleByteBuff(encodedBuffer));
283      int i = 0;
284      do {
285        KeyValue expectedKeyValue = sampleKv.get(i);
286        Cell cell = seeker.getCell();
287        if (PrivateCellUtil.compareKeyIgnoresMvcc(CellComparatorImpl.COMPARATOR, expectedKeyValue,
288          cell) != 0) {
289          int commonPrefix = PrivateCellUtil
290              .findCommonPrefixInFlatKey(expectedKeyValue, cell, false, true);
291          fail(String.format("next() produces wrong results "
292              + "encoder: %s i: %d commonPrefix: %d" + "\n expected %s\n actual      %s", encoder
293              .toString(), i, commonPrefix, Bytes.toStringBinary(expectedKeyValue.getBuffer(),
294              expectedKeyValue.getKeyOffset(), expectedKeyValue.getKeyLength()), CellUtil.toString(
295              cell, false)));
296        }
297        i++;
298      } while (seeker.next());
299    }
300  }
301
302  /**
303   * Test whether the decompression of first key is implemented correctly.
304   * @throws IOException
305   */
306  @Test
307  public void testFirstKeyInBlockOnSample() throws IOException {
308    List<KeyValue> sampleKv = generator.generateTestKeyValues(NUMBER_OF_KV, includesTags);
309
310    for (DataBlockEncoding encoding : DataBlockEncoding.values()) {
311      if (encoding.getEncoder() == null) {
312        continue;
313      }
314      DataBlockEncoder encoder = encoding.getEncoder();
315      ByteBuffer encodedBuffer = encodeKeyValues(encoding, sampleKv,
316          getEncodingContext(Compression.Algorithm.NONE, encoding), this.useOffheapData);
317      Cell key = encoder.getFirstKeyCellInBlock(new SingleByteBuff(encodedBuffer));
318      KeyValue firstKv = sampleKv.get(0);
319      if (0 != PrivateCellUtil.compareKeyIgnoresMvcc(CellComparatorImpl.COMPARATOR, key, firstKv)) {
320        int commonPrefix = PrivateCellUtil.findCommonPrefixInFlatKey(key, firstKv, false, true);
321        fail(String.format("Bug in '%s' commonPrefix %d", encoder.toString(), commonPrefix));
322      }
323    }
324  }
325
326  private void checkSeekingConsistency(List<DataBlockEncoder.EncodedSeeker> encodedSeekers,
327      boolean seekBefore, Cell keyValue) {
328    Cell expectedKeyValue = null;
329    ByteBuffer expectedKey = null;
330    ByteBuffer expectedValue = null;
331    for (DataBlockEncoder.EncodedSeeker seeker : encodedSeekers) {
332      seeker.seekToKeyInBlock(keyValue, seekBefore);
333      seeker.rewind();
334
335      Cell actualKeyValue = seeker.getCell();
336      ByteBuffer actualKey = null;
337      actualKey = ByteBuffer.wrap(((KeyValue) seeker.getKey()).getKey());
338      ByteBuffer actualValue = seeker.getValueShallowCopy();
339
340      if (expectedKeyValue != null) {
341        assertTrue(CellUtil.equals(expectedKeyValue, actualKeyValue));
342      } else {
343        expectedKeyValue = actualKeyValue;
344      }
345
346      if (expectedKey != null) {
347        assertEquals(expectedKey, actualKey);
348      } else {
349        expectedKey = actualKey;
350      }
351
352      if (expectedValue != null) {
353        assertEquals(expectedValue, actualValue);
354      } else {
355        expectedValue = actualValue;
356      }
357    }
358  }
359
360  private void testEncodersOnDataset(List<KeyValue> kvList, boolean includesMemstoreTS,
361      boolean includesTags) throws IOException {
362    ByteBuffer unencodedDataBuf = RedundantKVGenerator.convertKvToByteBuffer(kvList,
363        includesMemstoreTS);
364    HFileContext fileContext = new HFileContextBuilder().withIncludesMvcc(includesMemstoreTS)
365        .withIncludesTags(includesTags).build();
366    for (DataBlockEncoding encoding : DataBlockEncoding.values()) {
367      DataBlockEncoder encoder = encoding.getEncoder();
368      if (encoder == null) {
369        continue;
370      }
371      HFileBlockEncodingContext encodingContext = new HFileBlockDefaultEncodingContext(encoding,
372          HFILEBLOCK_DUMMY_HEADER, fileContext);
373
374      ByteArrayOutputStream baos = new ByteArrayOutputStream();
375      baos.write(HFILEBLOCK_DUMMY_HEADER);
376      DataOutputStream dos = new DataOutputStream(baos);
377      encoder.startBlockEncoding(encodingContext, dos);
378      for (KeyValue kv : kvList) {
379        encoder.encode(kv, encodingContext, dos);
380      }
381      encoder.endBlockEncoding(encodingContext, dos, baos.getBuffer());
382      byte[] encodedData = baos.toByteArray();
383
384      testAlgorithm(encodedData, unencodedDataBuf, encoder);
385    }
386  }
387
388  @Test
389  public void testZeroByte() throws IOException {
390    List<KeyValue> kvList = new ArrayList<>();
391    byte[] row = Bytes.toBytes("abcd");
392    byte[] family = new byte[] { 'f' };
393    byte[] qualifier0 = new byte[] { 'b' };
394    byte[] qualifier1 = new byte[] { 'c' };
395    byte[] value0 = new byte[] { 'd' };
396    byte[] value1 = new byte[] { 0x00 };
397    if (includesTags) {
398      kvList.add(new KeyValue(row, family, qualifier0, 0, value0,
399          new Tag[] { new ArrayBackedTag((byte) 1, "value1") }));
400      kvList.add(new KeyValue(row, family, qualifier1, 0, value1,
401          new Tag[] { new ArrayBackedTag((byte) 1, "value1") }));
402    } else {
403      kvList.add(new KeyValue(row, family, qualifier0, 0, Type.Put, value0));
404      kvList.add(new KeyValue(row, family, qualifier1, 0, Type.Put, value1));
405    }
406    testEncodersOnDataset(kvList, includesMemstoreTS, includesTags);
407  }
408
409  private void testAlgorithm(byte[] encodedData, ByteBuffer unencodedDataBuf,
410      DataBlockEncoder encoder) throws IOException {
411    // decode
412    ByteArrayInputStream bais = new ByteArrayInputStream(encodedData, ENCODED_DATA_OFFSET,
413        encodedData.length - ENCODED_DATA_OFFSET);
414    DataInputStream dis = new DataInputStream(bais);
415    ByteBuffer actualDataset;
416    HFileContext meta = new HFileContextBuilder().withHBaseCheckSum(false)
417        .withIncludesMvcc(includesMemstoreTS).withIncludesTags(includesTags)
418        .withCompression(Compression.Algorithm.NONE).build();
419    actualDataset = encoder.decodeKeyValues(dis, encoder.newDataBlockDecodingContext(meta));
420    actualDataset.rewind();
421
422    // this is because in case of prefix tree the decoded stream will not have
423    // the
424    // mvcc in it.
425    assertEquals("Encoding -> decoding gives different results for " + encoder,
426        Bytes.toStringBinary(unencodedDataBuf), Bytes.toStringBinary(actualDataset));
427  }
428}