/**
 * Serialize column family to data block encoding map to configuration.
 * Invoked while configuring the MR job for incremental load.
 *
 * @param table to read the properties from
 * @param conf to persist serialized values into
 * @throws IOException
 *           on failure to read column family descriptors
 */
@VisibleForTesting
static void configureDataBlockEncoding(HTableDescriptor tableDescriptor, Configuration conf)
        throws UnsupportedEncodingException {
    if (tableDescriptor == null) {
        // could happen with mock table instance
        return;
    }
    StringBuilder dataBlockEncodingConfigValue = new StringBuilder();
    Collection<HColumnDescriptor> families = tableDescriptor.getFamilies();
    int i = 0;
    for (HColumnDescriptor familyDescriptor : families) {
        if (i++ > 0) {
            dataBlockEncodingConfigValue.append('&');
        }
        dataBlockEncodingConfigValue.append(URLEncoder.encode(familyDescriptor.getNameAsString(), "UTF-8"));
        dataBlockEncodingConfigValue.append('=');
        DataBlockEncoding encoding = familyDescriptor.getDataBlockEncoding();
        if (encoding == null) {
            encoding = DataBlockEncoding.NONE;
        }
        dataBlockEncodingConfigValue.append(URLEncoder.encode(encoding.toString(), "UTF-8"));
    }
    conf.set(DATABLOCK_ENCODING_FAMILIES_CONF_KEY, dataBlockEncodingConfigValue.toString());
}
 

public HFileWriterImpl(final Configuration conf, CacheConfig cacheConf, Path path,
    FSDataOutputStream outputStream, HFileContext fileContext) {
  this.outputStream = outputStream;
  this.path = path;
  this.name = path != null ? path.getName() : outputStream.toString();
  this.hFileContext = fileContext;
  DataBlockEncoding encoding = hFileContext.getDataBlockEncoding();
  if (encoding != DataBlockEncoding.NONE) {
    this.blockEncoder = new HFileDataBlockEncoderImpl(encoding);
  } else {
    this.blockEncoder = NoOpDataBlockEncoder.INSTANCE;
  }
  closeOutputStream = path != null;
  this.cacheConf = cacheConf;
  float encodeBlockSizeRatio = conf.getFloat(UNIFIED_ENCODED_BLOCKSIZE_RATIO, 1f);
  this.encodedBlockSizeLimit = (int)(hFileContext.getBlocksize() * encodeBlockSizeRatio);
  finishInit(conf);
  if (LOG.isTraceEnabled()) {
    LOG.trace("Writer" + (path != null ? " for " + path : "") +
      " initialized with cacheConf: " + cacheConf +
      " fileContext: " + fileContext);
  }
}
 

public static HFileDataBlockEncoder createFromFileInfo(
    HFileInfo fileInfo) throws IOException {
  DataBlockEncoding encoding = DataBlockEncoding.NONE;
  byte[] dataBlockEncodingType = fileInfo.get(DATA_BLOCK_ENCODING);
  if (dataBlockEncodingType != null) {
    String dataBlockEncodingStr = Bytes.toString(dataBlockEncodingType);
    try {
      encoding = DataBlockEncoding.valueOf(dataBlockEncodingStr);
    } catch (IllegalArgumentException ex) {
      throw new IOException("Invalid data block encoding type in file info: "
        + dataBlockEncodingStr, ex);
    }
  }

  if (encoding == DataBlockEncoding.NONE) {
    return NoOpDataBlockEncoder.INSTANCE;
  }
  return new HFileDataBlockEncoderImpl(encoding);
}
 

private void writeHFile(Configuration conf, FileSystem fs, Path hfilePath, Algorithm compression,
    DataBlockEncoding encoding, int cellCount) throws IOException {
  HFileContext context =
      new HFileContextBuilder().withBlockSize(1).withDataBlockEncoding(DataBlockEncoding.NONE)
          .withCompression(compression).withDataBlockEncoding(encoding).build();
  try (HFile.Writer writer =
      new HFile.WriterFactory(conf, new CacheConfig(conf)).withPath(fs, hfilePath)
          .withFileContext(context).create()) {
    Random rand = new Random(9713312); // Just a fixed seed.
    for (int i = 0; i < cellCount; ++i) {
      byte[] keyBytes = Bytes.add(Bytes.toBytes(i), SUFFIX);

      // A random-length random value.
      byte[] valueBytes = RandomKeyValueUtil.randomValue(rand);
      KeyValue keyValue =
          new KeyValue(keyBytes, FAMILY, QUALIFIER, HConstants.LATEST_TIMESTAMP, valueBytes);
      if (firstCell == null) {
        firstCell = keyValue;
      } else if (secondCell == null) {
        secondCell = keyValue;
      }
      writer.append(keyValue);
    }
  }
}
 

HFileContext(boolean useHBaseChecksum, boolean includesMvcc, boolean includesTags,
             Compression.Algorithm compressAlgo, boolean compressTags, ChecksumType checksumType,
             int bytesPerChecksum, int blockSize, DataBlockEncoding encoding,
             Encryption.Context cryptoContext, long fileCreateTime, String hfileName,
             byte[] columnFamily, byte[] tableName, CellComparator cellComparator) {
  this.usesHBaseChecksum = useHBaseChecksum;
  this.includesMvcc =  includesMvcc;
  this.includesTags = includesTags;
  this.compressAlgo = compressAlgo;
  this.compressTags = compressTags;
  this.checksumType = checksumType;
  this.bytesPerChecksum = bytesPerChecksum;
  this.blocksize = blockSize;
  if (encoding != null) {
    this.encoding = encoding;
  }
  this.cryptoContext = cryptoContext;
  this.fileCreateTime = fileCreateTime;
  this.hfileName = hfileName;
  this.columnFamily = columnFamily;
  this.tableName = tableName;
  // If no cellComparator specified, make a guess based off tablename. If hbase:meta, then should
  // be the meta table comparator. Comparators are per table.
  this.cellComparator = cellComparator != null ? cellComparator : this.tableName != null ?
    CellComparatorImpl.getCellComparator(this.tableName) : CellComparator.getInstance();
}
 

/**
 * Updates the current block to be the given {@link HFileBlock}. Seeks to the the first
 * key/value pair.
 * @param newBlock the block to make current, and read by {@link HFileReaderImpl#readBlock},
 *          it's a totally new block with new allocated {@link ByteBuff}, so if no further
 *          reference to this block, we should release it carefully.
 */
@Override
protected void updateCurrentBlock(HFileBlock newBlock) throws CorruptHFileException {
  try {
    // sanity checks
    if (newBlock.getBlockType() != BlockType.ENCODED_DATA) {
      throw new IllegalStateException("EncodedScanner works only on encoded data blocks");
    }
    short dataBlockEncoderId = newBlock.getDataBlockEncodingId();
    if (!DataBlockEncoding.isCorrectEncoder(dataBlockEncoder, dataBlockEncoderId)) {
      String encoderCls = dataBlockEncoder.getClass().getName();
      throw new CorruptHFileException("Encoder " + encoderCls +
        " doesn't support data block encoding " +
        DataBlockEncoding.getNameFromId(dataBlockEncoderId) + ",path=" + reader.getPath());
    }
    updateCurrBlockRef(newBlock);
    ByteBuff encodedBuffer = getEncodedBuffer(newBlock);
    seeker.setCurrentBuffer(encodedBuffer);
  } finally {
    releaseIfNotCurBlock(newBlock);
  }
  // Reset the next indexed key
  this.nextIndexedKey = null;
}
 

public void setUp(long threshold, String TN, DataBlockEncoding encoding)
    throws Exception {
  tableDescriptor =
    new TableDescriptorBuilder.ModifyableTableDescriptor(TableName.valueOf(TN));
  columnFamilyDescriptor =
    new ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor(family);
  columnFamilyDescriptor.setMobEnabled(true);
  columnFamilyDescriptor.setMobThreshold(threshold);
  columnFamilyDescriptor.setMaxVersions(4);
  columnFamilyDescriptor.setDataBlockEncoding(encoding);
  tableDescriptor.setColumnFamily(columnFamilyDescriptor);
  admin = TEST_UTIL.getAdmin();
  admin.createTable(tableDescriptor);
  table = ConnectionFactory.createConnection(TEST_UTIL.getConfiguration())
    .getTable(TableName.valueOf(TN));
}
 

public void testDataBlockEncoding(DataBlockEncoding encoding) throws Exception {
  String TN = "testDataBlockEncoding" + encoding;
  setUp(defaultThreshold, TN, encoding);
  long ts1 = System.currentTimeMillis();
  long ts2 = ts1 + 1;
  long ts3 = ts1 + 2;
  byte[] value = generateMobValue((int) defaultThreshold + 1);

  Put put1 = new Put(row1);
  put1.addColumn(family, qf1, ts3, value);
  put1.addColumn(family, qf2, ts2, value);
  put1.addColumn(family, qf3, ts1, value);
  table.put(put1);
  admin.flush(TableName.valueOf(TN));

  Scan scan = new Scan();
  scan.readVersions(4);
  MobTestUtil.assertCellsValue(table, scan, value, 3);
}
 

/**
 * @return a map from column family names to compression algorithms for
 *         testing column family compression. Column family names have special characters
 */
private Map<String, DataBlockEncoding>
    getMockColumnFamiliesForDataBlockEncoding (int numCfs) {
  Map<String, DataBlockEncoding> familyToDataBlockEncoding = new HashMap<>();
  // use column family names having special characters
  if (numCfs-- > 0) {
    familyToDataBlockEncoding.put("[email protected]#[email protected]#&", DataBlockEncoding.DIFF);
  }
  if (numCfs-- > 0) {
    familyToDataBlockEncoding.put("Family2=asdads&!AASD",
        DataBlockEncoding.FAST_DIFF);
  }
  if (numCfs-- > 0) {
    familyToDataBlockEncoding.put("Family2=asdads&!AASD",
        DataBlockEncoding.PREFIX);
  }
  if (numCfs-- > 0) {
    familyToDataBlockEncoding.put("Family3", DataBlockEncoding.NONE);
  }
  return familyToDataBlockEncoding;
}
 

private void testSplitStoreFileWithDifferentEncoding(DataBlockEncoding bulkloadEncoding,
    DataBlockEncoding cfEncoding) throws IOException {
  Path dir = util.getDataTestDirOnTestFS("testSplitHFileWithDifferentEncoding");
  FileSystem fs = util.getTestFileSystem();
  Path testIn = new Path(dir, "testhfile");
  ColumnFamilyDescriptor familyDesc =
    ColumnFamilyDescriptorBuilder.newBuilder(FAMILY).setDataBlockEncoding(cfEncoding).build();
  HFileTestUtil.createHFileWithDataBlockEncoding(util.getConfiguration(), fs, testIn,
    bulkloadEncoding, FAMILY, QUALIFIER, Bytes.toBytes("aaa"), Bytes.toBytes("zzz"), 1000);

  Path bottomOut = new Path(dir, "bottom.out");
  Path topOut = new Path(dir, "top.out");

  BulkLoadHFilesTool.splitStoreFile(util.getConfiguration(), testIn, familyDesc,
    Bytes.toBytes("ggg"), bottomOut, topOut);

  int rowCount = verifyHFile(bottomOut);
  rowCount += verifyHFile(topOut);
  assertEquals(1000, rowCount);
}
 

/**
 * Creates a pre-split table for load testing. If the table already exists,
 * logs a warning and continues.
 * @return the number of regions the table was split into
 */
public static int createPreSplitLoadTestTable(Configuration conf,
    TableName tableName, byte[] columnFamily, Algorithm compression,
    DataBlockEncoding dataBlockEncoding, int numRegionsPerServer, int regionReplication,
    Durability durability)
        throws IOException {
  TableDescriptorBuilder.ModifyableTableDescriptor tableDescriptor =
    new TableDescriptorBuilder.ModifyableTableDescriptor(tableName);
  tableDescriptor.setDurability(durability);
  tableDescriptor.setRegionReplication(regionReplication);
  ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor familyDescriptor =
    new ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor(columnFamily);
  familyDescriptor.setDataBlockEncoding(dataBlockEncoding);
  familyDescriptor.setCompressionType(compression);
  return createPreSplitLoadTestTable(conf, tableDescriptor, familyDescriptor,
    numRegionsPerServer);
}
 

/**
 * Create a set of column descriptors with the combination of compression,
 * encoding, bloom codecs available.
 * @param prefix family names prefix
 * @return the list of column descriptors
 */
public static List<ColumnFamilyDescriptor> generateColumnDescriptors(final String prefix) {
  List<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>();
  long familyId = 0;
  for (Compression.Algorithm compressionType: getSupportedCompressionAlgorithms()) {
    for (DataBlockEncoding encodingType: DataBlockEncoding.values()) {
      for (BloomType bloomType: BloomType.values()) {
        String name = String.format("%[email protected]#&-%[email protected]#", prefix, familyId);
        ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder =
          ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(name));
        columnFamilyDescriptorBuilder.setCompressionType(compressionType);
        columnFamilyDescriptorBuilder.setDataBlockEncoding(encodingType);
        columnFamilyDescriptorBuilder.setBloomFilterType(bloomType);
        columnFamilyDescriptors.add(columnFamilyDescriptorBuilder.build());
        familyId++;
      }
    }
  }
  return columnFamilyDescriptors;
}
 

private void testEncodingWithCacheInternals(boolean useTag) throws IOException {
  List<KeyValue> kvs = generator.generateTestKeyValues(60, useTag);
  HFileBlock block = getSampleHFileBlock(kvs, useTag);
  HFileBlock cacheBlock = createBlockOnDisk(kvs, block, useTag);

  LruBlockCache blockCache =
      new LruBlockCache(8 * 1024 * 1024, 32 * 1024);
  BlockCacheKey cacheKey = new BlockCacheKey("test", 0);
  blockCache.cacheBlock(cacheKey, cacheBlock);

  HeapSize heapSize = blockCache.getBlock(cacheKey, false, false, true);
  assertTrue(heapSize instanceof HFileBlock);

  HFileBlock returnedBlock = (HFileBlock) heapSize;

  if (blockEncoder.getDataBlockEncoding() ==
      DataBlockEncoding.NONE) {
    assertEquals(block.getBufferReadOnly(), returnedBlock.getBufferReadOnly());
  } else {
    if (BlockType.ENCODED_DATA != returnedBlock.getBlockType()) {
      System.out.println(blockEncoder);
    }
    assertEquals(BlockType.ENCODED_DATA, returnedBlock.getBlockType());
  }
}
 

@Test
public void testDisabledBlockCache() throws Exception {
  writeHFile(conf, fs, hfilePath, Algorithm.NONE, DataBlockEncoding.NONE, CELL_COUNT);
  // Set LruBlockCache
  conf.setFloat(HFILE_BLOCK_CACHE_SIZE_KEY, 0.0f);
  BlockCache defaultBC = BlockCacheFactory.createBlockCache(conf);
  Assert.assertNull(defaultBC);
  CacheConfig cacheConfig = new CacheConfig(conf, null, defaultBC, allocator);
  Assert.assertFalse(cacheConfig.isCombinedBlockCache()); // Must be LruBlockCache.
  HFile.Reader reader = HFile.createReader(fs, hfilePath, cacheConfig, true, conf);
  Assert.assertTrue(reader instanceof HFileReaderImpl);
  // We've build a HFile tree with index = 16.
  Assert.assertEquals(16, reader.getTrailer().getNumDataIndexLevels());

  HFileBlock block1 = reader.getDataBlockIndexReader()
      .loadDataBlockWithScanInfo(firstCell, null, true, true, false,
          DataBlockEncoding.NONE, reader).getHFileBlock();

  Assert.assertTrue(block1.isSharedMem());
  Assert.assertTrue(block1 instanceof SharedMemHFileBlock);
  Assert.assertEquals(1, block1.refCnt());
  Assert.assertTrue(block1.release());
}
 

public void majorCompactionWithDataBlockEncoding(boolean inCacheOnly) throws Exception {
  Map<HStore, HFileDataBlockEncoder> replaceBlockCache = new HashMap<>();
  for (HStore store : r.getStores()) {
    HFileDataBlockEncoder blockEncoder = store.getDataBlockEncoder();
    replaceBlockCache.put(store, blockEncoder);
    final DataBlockEncoding inCache = DataBlockEncoding.PREFIX;
    final DataBlockEncoding onDisk = inCacheOnly ? DataBlockEncoding.NONE : inCache;
    ((HStore) store).setDataBlockEncoderInTest(new HFileDataBlockEncoderImpl(onDisk));
  }

  majorCompaction();

  // restore settings
  for (Entry<HStore, HFileDataBlockEncoder> entry : replaceBlockCache.entrySet()) {
    ((HStore) entry.getKey()).setDataBlockEncoderInTest(entry.getValue());
  }
}
 

/**
 * Check if data block encoding information is saved correctly in HFile's file info.
 */
@Test
public void testDataBlockEncodingMetaData() throws IOException {
  // Make up a directory hierarchy that has a regiondir ("7e0102") and familyname.
  Path dir = new Path(new Path(testDir, "7e0102"), "familyname");
  Path path = new Path(dir, "1234567890");

  DataBlockEncoding dataBlockEncoderAlgo = DataBlockEncoding.FAST_DIFF;
  cacheConf = new CacheConfig(conf);
  HFileContext meta =
    new HFileContextBuilder().withBlockSize(BLOCKSIZE_SMALL).withChecksumType(CKTYPE)
      .withBytesPerCheckSum(CKBYTES).withDataBlockEncoding(dataBlockEncoderAlgo).build();
  // Make a store file and write data to it.
  StoreFileWriter writer = new StoreFileWriter.Builder(conf, cacheConf, this.fs)
    .withFilePath(path).withMaxKeyCount(2000).withFileContext(meta).build();
  writer.close();

  HStoreFile storeFile =
    new HStoreFile(fs, writer.getPath(), conf, cacheConf, BloomType.NONE, true);
  storeFile.initReader();
  StoreFileReader reader = storeFile.getReader();

  Map<byte[], byte[]> fileInfo = reader.loadFileInfo();
  byte[] value = fileInfo.get(HFileDataBlockEncoder.DATA_BLOCK_ENCODING);
  assertArrayEquals(dataBlockEncoderAlgo.getNameInBytes(), value);
}
 

/**
 * Runs inside the task to deserialize column family to data block encoding
 * type map from the configuration.
 *
 * @param conf to read the serialized values from
 * @return a map from column family to HFileDataBlockEncoder for the
 *         configured data block type for the family
 */
@VisibleForTesting
static Map<byte[], DataBlockEncoding> createFamilyDataBlockEncodingMap(Configuration conf) {
    Map<byte[], String> stringMap = createFamilyConfValueMap(conf, DATABLOCK_ENCODING_FAMILIES_CONF_KEY);
    Map<byte[], DataBlockEncoding> encoderMap = new TreeMap<byte[], DataBlockEncoding>(Bytes.BYTES_COMPARATOR);
    for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
        encoderMap.put(e.getKey(), DataBlockEncoding.valueOf((e.getValue())));
    }
    return encoderMap;
}
 

private void setupMockColumnFamiliesForDataBlockEncoding(Table table,
    Map<String, DataBlockEncoding> familyToDataBlockEncoding) throws IOException {
  TableDescriptorBuilder mockTableDescriptor =
    TableDescriptorBuilder.newBuilder(TABLE_NAMES[0]);
  for (Entry<String, DataBlockEncoding> entry : familyToDataBlockEncoding.entrySet()) {
    ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder
      .newBuilder(Bytes.toBytes(entry.getKey()))
      .setMaxVersions(1)
      .setDataBlockEncoding(entry.getValue())
      .setBlockCacheEnabled(false)
      .setTimeToLive(0).build();
    mockTableDescriptor.setColumnFamily(columnFamilyDescriptor);
  }
  Mockito.doReturn(mockTableDescriptor).when(table).getDescriptor();
}
 

private static void createTable(HBaseGraphConfiguration config, Admin admin, String name, int ttl) throws IOException {
    TableName tableName = getTableName(config, name);
    if (admin.tableExists(tableName)) return;
    try {
        HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
        tableDescriptor.setDurability(config.getTableAsyncWAL() ? Durability.ASYNC_WAL : Durability.USE_DEFAULT);
        HColumnDescriptor columnDescriptor = new HColumnDescriptor(DEFAULT_FAMILY)
                .setCompressionType(Compression.Algorithm.valueOf(config.getCompressionAlgorithm().toUpperCase()))
                .setBloomFilterType(BloomType.ROW)
                .setDataBlockEncoding(DataBlockEncoding.FAST_DIFF)
                .setMaxVersions(1)
                .setMinVersions(0)
                .setBlocksize(32768)
                .setBlockCacheEnabled(true)
                .setTimeToLive(ttl);
        tableDescriptor.addFamily(columnDescriptor);
        int regionCount = config.getRegionCount();
        if (regionCount <= 1) {
            admin.createTable(tableDescriptor);
        } else {
            admin.createTable(tableDescriptor, getStartKey(regionCount), getEndKey(regionCount), regionCount);
        }
    } catch (IOException e) {
        LOGGER.error("Could not create table " + tableName, e);
        throw e;
    }
}
 

private void setCFOptions(HColumnDescriptor cdesc, int ttlInSeconds) {
    if (null != compression && !compression.equals(COMPRESSION_DEFAULT))
        compat.setCompression(cdesc, compression);

    if (ttlInSeconds > 0)
        cdesc.setTimeToLive(ttlInSeconds);

    cdesc.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF);
}
 

private void parseColumnFamilyOptions(CommandLine cmd) {
  String dataBlockEncodingStr = cmd.getOptionValue(HFileTestUtil.OPT_DATA_BLOCK_ENCODING);
  dataBlockEncodingAlgo = dataBlockEncodingStr == null ? null :
      DataBlockEncoding.valueOf(dataBlockEncodingStr);

  String compressStr = cmd.getOptionValue(OPT_COMPRESSION);
  compressAlgo = compressStr == null ? Compression.Algorithm.NONE :
      Compression.Algorithm.valueOf(compressStr);

  String bloomStr = cmd.getOptionValue(OPT_BLOOM);
  bloomType = bloomStr == null ? BloomType.ROW :
      BloomType.valueOf(bloomStr);

  if (bloomType == BloomType.ROWPREFIX_FIXED_LENGTH) {
    if (!cmd.hasOption(OPT_BLOOM_PARAM)) {
      LOG.error("the parameter of bloom filter {} is not specified", bloomType.name());
    } else {
      conf.set(BloomFilterUtil.PREFIX_LENGTH_KEY, cmd.getOptionValue(OPT_BLOOM_PARAM));
    }
  }

  inMemoryCF = cmd.hasOption(OPT_INMEMORY);
  if (cmd.hasOption(OPT_ENCRYPTION)) {
    cipher = Encryption.getCipher(conf, cmd.getOptionValue(OPT_ENCRYPTION));
  }

}
 

/**
 * Runs inside the task to deserialize column family to data block encoding
 * type map from the configuration.
 *
 * @param conf to read the serialized values from
 * @return a map from column family to HFileDataBlockEncoder for the
 *         configured data block type for the family
 */
@VisibleForTesting
static Map<byte[], DataBlockEncoding> createFamilyDataBlockEncodingMap(
    Configuration conf) {
  Map<byte[], String> stringMap = createFamilyConfValueMap(conf,
      DATABLOCK_ENCODING_FAMILIES_CONF_KEY);
  Map<byte[], DataBlockEncoding> encoderMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
  for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
    encoderMap.put(e.getKey(), DataBlockEncoding.valueOf((e.getValue())));
  }
  return encoderMap;
}
 

/**
 * Runs inside the task to deserialize column family to data block encoding
 * type map from the configuration.
 *
 * @param conf to read the serialized values from
 * @return a map from column family to HFileDataBlockEncoder for the
 *         configured data block type for the family
 */
@VisibleForTesting
static Map<byte[], DataBlockEncoding> createFamilyDataBlockEncodingMap(Configuration conf) {
    Map<byte[], String> stringMap = createFamilyConfValueMap(conf, DATABLOCK_ENCODING_FAMILIES_CONF_KEY);
    Map<byte[], DataBlockEncoding> encoderMap = new TreeMap<byte[], DataBlockEncoding>(Bytes.BYTES_COMPARATOR);
    for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
        encoderMap.put(e.getKey(), DataBlockEncoding.valueOf((e.getValue())));
    }
    return encoderMap;
}
 

public static DataBlockEncoding dataBlockEncodingFromThrift(TDataBlockEncoding in) {
  switch (in.getValue()) {
    case 0: return DataBlockEncoding.NONE;
    case 2: return DataBlockEncoding.PREFIX;
    case 3: return DataBlockEncoding.DIFF;
    case 4: return DataBlockEncoding.FAST_DIFF;
    case 7: return DataBlockEncoding.ROW_INDEX_V1;
    default: return DataBlockEncoding.NONE;
  }
}
 

@Parameters
public static Collection<Object[]> parameters() {
  List<Object[]> paramList = new ArrayList<>();
  for (DataBlockEncoding encoding : DataBlockEncoding.values()) {
    paramList.add(new Object[] { encoding });
  }
  return paramList;
}
 

/**
 * Check DataBlockEncodings of column families are compatible.
 *
 * @return number of column families with incompatible DataBlockEncoding
 * @throws IOException if a remote or network exception occurs
 */
private int validateDBE() throws IOException {
  int incompatibilities = 0;

  LOG.info("Validating Data Block Encodings");

  try (Connection connection = ConnectionFactory.createConnection(getConf());
      Admin admin = connection.getAdmin()) {
    List<TableDescriptor> tableDescriptors = admin.listTableDescriptors();
    String encoding = "";

    for (TableDescriptor td : tableDescriptors) {
      ColumnFamilyDescriptor[] columnFamilies = td.getColumnFamilies();
      for (ColumnFamilyDescriptor cfd : columnFamilies) {
        try {
          encoding = Bytes.toString(cfd.getValue(DATA_BLOCK_ENCODING));
          // IllegalArgumentException will be thrown if encoding is incompatible with 2.0
          DataBlockEncoding.valueOf(encoding);
        } catch (IllegalArgumentException e) {
          incompatibilities++;
          LOG.warn("Incompatible DataBlockEncoding for table: {}, cf: {}, encoding: {}",
              td.getTableName().getNameAsString(), cfd.getNameAsString(), encoding);
        }
      }
    }
  }

  if (incompatibilities > 0) {
    LOG.warn("There are {} column families with incompatible Data Block Encodings. Do not "
        + "upgrade until these encodings are converted to a supported one. "
        + "Check https://s.apache.org/prefixtree for instructions.", incompatibilities);
  } else {
    LOG.info("The used Data Block Encodings are compatible with HBase 2.0.");
  }

  return incompatibilities;
}
 

protected void finishFileInfo() throws IOException {
  if (lastCell != null) {
    // Make a copy. The copy is stuffed into our fileinfo map. Needs a clean
    // byte buffer. Won't take a tuple.
    byte [] lastKey = PrivateCellUtil.getCellKeySerializedAsKeyValueKey(this.lastCell);
    fileInfo.append(HFileInfo.LASTKEY, lastKey, false);
  }

  // Average key length.
  int avgKeyLen =
      entryCount == 0 ? 0 : (int) (totalKeyLength / entryCount);
  fileInfo.append(HFileInfo.AVG_KEY_LEN, Bytes.toBytes(avgKeyLen), false);
  fileInfo.append(HFileInfo.CREATE_TIME_TS, Bytes.toBytes(hFileContext.getFileCreateTime()),
    false);

  // Average value length.
  int avgValueLen =
      entryCount == 0 ? 0 : (int) (totalValueLength / entryCount);
  fileInfo.append(HFileInfo.AVG_VALUE_LEN, Bytes.toBytes(avgValueLen), false);
  if (hFileContext.isIncludesTags()) {
    // When tags are not being written in this file, MAX_TAGS_LEN is excluded
    // from the FileInfo
    fileInfo.append(HFileInfo.MAX_TAGS_LEN, Bytes.toBytes(this.maxTagsLength), false);
    boolean tagsCompressed = (hFileContext.getDataBlockEncoding() != DataBlockEncoding.NONE)
      && hFileContext.isCompressTags();
    fileInfo.append(HFileInfo.TAGS_COMPRESSED, Bytes.toBytes(tagsCompressed), false);
  }
}
 

@Override
public DataBlockEncoding getEffectiveEncodingInCache(boolean isCompaction) {
  if (!useEncodedScanner(isCompaction)) {
    return DataBlockEncoding.NONE;
  }
  return encoding;
}
 

@Override
public void startBlockEncoding(HFileBlockEncodingContext encodingCtx, DataOutputStream out)
    throws IOException {
  if (this.encoding != null && this.encoding != DataBlockEncoding.NONE) {
    this.encoding.getEncoder().startBlockEncoding(encodingCtx, out);
  }
}
 

@Test
public void testDropViewKeepsHTable() throws Exception {
    Connection conn = getConnection();
    Admin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin();
    String hbaseNativeViewName = generateUniqueName();

    byte[] hbaseNativeBytes = SchemaUtil.getTableNameAsBytes(HBASE_NATIVE_SCHEMA_NAME, hbaseNativeViewName);
    try {
         TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(TableName.valueOf(hbaseNativeBytes));
        ColumnFamilyDescriptor columnDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(FAMILY_NAME)
                .setKeepDeletedCells(KeepDeletedCells.TRUE).build();
        builder.addColumnFamily(columnDescriptor);
        admin.createTable(builder.build());
    } finally {
        admin.close();
    }
    
    conn.createStatement().execute("create view " + hbaseNativeViewName+
            "   (uint_key unsigned_int not null," +
            "    ulong_key unsigned_long not null," +
            "    string_key varchar not null,\n" +
            "    \"1\".uint_col unsigned_int," +
            "    \"1\".ulong_col unsigned_long" +
            "    CONSTRAINT pk PRIMARY KEY (uint_key, ulong_key, string_key))\n" +
            ColumnFamilyDescriptorBuilder.DATA_BLOCK_ENCODING + "='" + DataBlockEncoding.NONE + "'");
    conn.createStatement().execute("drop view " + hbaseNativeViewName);
    conn.close();
}