Java 类org.apache.hadoop.hbase.index.IndexSpecification 实例源码

private void prepareIndexMutations(List<IndexSpecification> indices, HRegion userRegion,
    Mutation mutation, String tableName, HRegion indexRegion) throws IOException {
  IndexEdits indexEdits = threadLocal.get();
  if (mutation instanceof Put) {
    for (IndexSpecification index : indices) {
      // Handle each of the index
      Mutation indexPut = IndexUtils.prepareIndexPut((Put) mutation, index, indexRegion);
      if (null != indexPut) {
        // This mutation can be null when the user table mutation is not
        // containing all of the indexed col value.
        indexEdits.add(indexPut);
      }
    }
  } else if (mutation instanceof Delete) {
    Collection<? extends Mutation> indexDeletes =
        prepareIndexDeletes((Delete) mutation, userRegion, indices, indexRegion);
    indexEdits.addAll(indexDeletes);
  } else {
    // TODO : Log or throw exception
  }
}

FilterNode evalFilterForIndexSelection(Filter filter, List<IndexSpecification> indices) {
  if (filter instanceof FilterList) {
    FilterList fList = (FilterList) filter;
    GroupingCondition condition =
        (fList.getOperator() == Operator.MUST_PASS_ALL) ? GroupingCondition.AND
            : GroupingCondition.OR;
    NonLeafFilterNode nonLeafFilterNode = new NonLeafFilterNode(condition);
    List<Filter> filters = fList.getFilters();
    for (Filter fltr : filters) {
      FilterNode node = evalFilterForIndexSelection(fltr, indices);
      nonLeafFilterNode.addFilterNode(node);
    }
    return handleNonLeafFilterNode(nonLeafFilterNode);
  } else if (filter instanceof SingleColumnValueFilter) {
    // Check for the availability of index
    return selectBestFitAndPossibleIndicesForSCVF(indices, (SingleColumnValueFilter) filter);
  } else if (filter instanceof SingleColumnRangeFilter) {
    return selectBestFitAndPossibleIndicesForSCRF(indices, (SingleColumnRangeFilter) filter);
  }
  return new NoIndexFilterNode();
}

private FilterNode handleORCondition(NonLeafFilterNode nonLeafFilterNode) {
  Iterator<FilterNode> nonLeafFilterNodeItr = nonLeafFilterNode.getFilterNodes().iterator();
  while (nonLeafFilterNodeItr.hasNext()) {
    FilterNode filterNode = nonLeafFilterNodeItr.next();
    if (filterNode instanceof IndexFilterNode) {
      FilterColumnValueDetail filterColumnValueDetail =
          ((IndexFilterNode) filterNode).getFilterColumnValueDetail();
      IndexSpecification indexToUse = ((IndexFilterNode) filterNode).getBestIndex();
      nonLeafFilterNode.addIndicesToUse(filterColumnValueDetail, indexToUse);
      nonLeafFilterNodeItr.remove();
    } else if (filterNode instanceof PossibleIndexFilterNode
        || filterNode instanceof NoIndexFilterNode) {
      // The moment an OR condition contains a column on which there is no index which can be
      // used, the entire OR node becomes as non indexed.
      return new NoIndexFilterNode();
    }
    // A NonLeafFilterNode under the OR node need to be kept as it is.
  }
  return nonLeafFilterNode;
}

private FilterNode selectBestFitAndPossibleIndicesForSCVF(List<IndexSpecification> indices,
    SingleColumnValueFilter filter) {
  if (CompareOp.NOT_EQUAL == filter.getOperator() || CompareOp.NO_OP == filter.getOperator()) {
    return new NoIndexFilterNode();
  }
  FilterColumnValueDetail detail = null;
  if (filter instanceof SingleColumnValuePartitionFilter) {
    SingleColumnValuePartitionFilter escvf = (SingleColumnValuePartitionFilter) filter;
    detail =
        new FilterColumnValueDetail(escvf.getFamily(), escvf.getQualifier(), escvf
            .getComparator().getValue(), escvf.getValuePartition(), escvf.getOperator());
  } else {
    detail =
        new FilterColumnValueDetail(filter.getFamily(), filter.getQualifier(), filter
            .getComparator().getValue(), filter.getOperator());
  }
  return selectBestFitIndexForColumn(indices, detail);
}

public void testAddIndexForTable() throws Exception {

    IndexManager im = IndexManager.getInstance();
    assertNotNull("Index Manager should not be null.", im);

    List<IndexSpecification> indexList = new ArrayList<IndexSpecification>(1);
    IndexSpecification iSpec = new IndexSpecification("index_name");

    iSpec.addIndexColumn(new HColumnDescriptor("cf"), "cq", null, 10);
    indexList.add(iSpec);
    im.addIndexForTable("index_name", indexList);
    indexList = im.getIndicesForTable("index_name");
    assertEquals("Index name should be equal with actual value.", "index_name", indexList.get(0)
        .getName());
    assertTrue("Column qualifier state mismatch.",
      indexList.get(0).getIndexColumns().contains(new ColumnQualifier("cf", "cq", null, 10)));

  }

@Test(timeout = 180000)
public void testPreCreateShouldNotBeSuccessfulIfIndicesAreNotSameAtBothTypeAndLength()
    throws IOException, KeeperException, InterruptedException {
  String userTableName = "testNotConsisIndex4";
  HTableDescriptor ihtd = new HTableDescriptor(TableName.valueOf(userTableName));
  HColumnDescriptor hcd = new HColumnDescriptor("col");
  IndexSpecification iSpec1 = new IndexSpecification("Index1");
  iSpec1.addIndexColumn(hcd, "q1", ValueType.String, 10);
  iSpec1.addIndexColumn(hcd, "q2", ValueType.String, 10);
  ihtd.addFamily(hcd);
  IndexSpecification iSpec2 = new IndexSpecification("Index2");
  iSpec2.addIndexColumn(hcd, "q1", ValueType.Int, 10);
  iSpec2.addIndexColumn(hcd, "q2", ValueType.String, 7);
  TableIndices indices = new TableIndices();
  indices.addIndex(iSpec1);
  indices.addIndex(iSpec2);
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  boolean returnVal = false;
  try {
    admin.createTable(ihtd);
    fail("IOException should be thrown");
  } catch (IOException e) {
    returnVal = true;
  }
  Assert.assertTrue(returnVal);
}

@Test(timeout = 180000)
public void testPreCreateShouldBeSuccessfulIfIndicesAreSame() throws IOException,
    KeeperException, InterruptedException {
  String userTableName = "testConsistIndex";
  HTableDescriptor ihtd = new HTableDescriptor(TableName.valueOf(userTableName));
  HColumnDescriptor hcd = new HColumnDescriptor("col");
  IndexSpecification iSpec1 = new IndexSpecification("Index1");
  iSpec1.addIndexColumn(hcd, "q1", ValueType.String, 10);
  ihtd.addFamily(hcd);
  IndexSpecification iSpec2 = new IndexSpecification("Index2");
  iSpec2.addIndexColumn(hcd, "q1", ValueType.String, 10);
  TableIndices indices = new TableIndices();
  indices.addIndex(iSpec1);
  indices.addIndex(iSpec2);
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  try {
    admin.createTable(ihtd);
  } catch (IOException e) {
    fail("Exception should not be thrown");
  }
}

@Test(timeout = 180000)
public void testIndexPutWithOffsetAndLength() throws IOException {
  Path basedir = new Path(DIR + "TestIndexPut");
  Configuration conf = TEST_UTIL.getConfiguration();
  HTableDescriptor htd =
      new HTableDescriptor(TableName.valueOf("testIndexPutWithOffsetAndLength"));
  HRegionInfo info = new HRegionInfo(htd.getTableName(), "ABC".getBytes(), "BBB".getBytes(), false);
  HRegion region = HRegion.createHRegion(info, basedir, conf, htd);
  IndexSpecification spec = new IndexSpecification("index");
  spec.addIndexColumn(new HColumnDescriptor("col"), "ql1", new SpatialPartition(20, 2),
    ValueType.String, 18);

  byte[] value1 = "AB---CD---EF---GH---IJ---KL---MN---OP---".getBytes();
  Put p = new Put("row".getBytes());
  p.add("col".getBytes(), "ql1".getBytes(), value1);
  Put indexPut = IndexUtils.prepareIndexPut(p, spec, region);
  byte[] indexRowKey = indexPut.getRow();
  byte[] actualResult = new byte[2];
  System.arraycopy(indexRowKey, 22, actualResult, 0, actualResult.length);
  byte[] expectedResult = new byte[2];
  System.arraycopy("IJ".getBytes(), 0, expectedResult, 0, "IJ".getBytes().length);
  Assert.assertTrue(Bytes.equals(actualResult, expectedResult));
}

@Test(timeout = 180000)
public void testIndexPutwithPositiveIntDataTypes() throws IOException {
  Path basedir = new Path(DIR + "TestIndexPut");
  Configuration conf = TEST_UTIL.getConfiguration();
  HTableDescriptor htd =
      new HTableDescriptor(TableName.valueOf("testIndexPutwithPositiveIntDataTypes"));
  HRegionInfo info = new HRegionInfo(htd.getTableName(), "ABC".getBytes(), "BBB".getBytes(), false);
  // HLog hlog = UTIL.getMiniHBaseCluster().getRegionServer(0).getWAL();
  HRegion region = HRegion.createHRegion(info, basedir, conf, htd);
  IndexSpecification spec = new IndexSpecification("index");
  spec.addIndexColumn(new HColumnDescriptor("col"), "ql1", ValueType.Int, 4);
  spec.addIndexColumn(new HColumnDescriptor("col"), "ql2", ValueType.Float, 4);

  byte[] value1 = Bytes.toBytes(1000);
  Put p = new Put("row".getBytes());
  p.add("col".getBytes(), "ql1".getBytes(), value1);
  Put indexPut1 = IndexUtils.prepareIndexPut(p, spec, region);
  int a = 1000;
  byte[] expectedResult = Bytes.toBytes(a ^ (1 << 31));
  byte[] actualResult = new byte[4];
  byte[] indexRowKey = indexPut1.getRow();
  System.arraycopy(indexRowKey, 22, actualResult, 0, actualResult.length);
  Assert.assertTrue(Bytes.equals(expectedResult, actualResult));
}

@Test(timeout = 180000)
public void testIndexPutWithNegativeIntDataTypes() throws IOException {
  Path basedir = new Path(DIR + "TestIndexPut");
  Configuration conf = TEST_UTIL.getConfiguration();
  HTableDescriptor htd =
      new HTableDescriptor(TableName.valueOf("testIndexPutWithNegativeIntDataTypes"));
  HRegionInfo info = new HRegionInfo(htd.getTableName(), "ABC".getBytes(), "BBB".getBytes(), false);
  // HLog hlog = UTIL.getMiniHBaseCluster().getRegionServer(0).getWAL();
  HRegion region = HRegion.createHRegion(info, basedir, conf, htd);
  IndexSpecification spec = new IndexSpecification("index");
  spec.addIndexColumn(new HColumnDescriptor("col"), "ql1", ValueType.Int, 4);
  spec.addIndexColumn(new HColumnDescriptor("col"), "ql2", ValueType.Float, 4);

  byte[] value1 = Bytes.toBytes(-2562351);
  Put p = new Put("row".getBytes());
  p.add("col".getBytes(), "ql1".getBytes(), value1);
  Put indexPut = IndexUtils.prepareIndexPut(p, spec, region);
  int a = -2562351;
  byte[] expectedResult = Bytes.toBytes(a ^ (1 << 31));
  byte[] actualResult = new byte[4];
  byte[] indexRowKey = indexPut.getRow();
  System.arraycopy(indexRowKey, 22, actualResult, 0, actualResult.length);
  Assert.assertTrue(Bytes.equals(expectedResult, actualResult));
}

@Test(timeout = 180000)
public void testIndexPutWithLongDataTypes() throws IOException {
  Path basedir = new Path(DIR + "TestIndexPut");
  Configuration conf = TEST_UTIL.getConfiguration();
  HTableDescriptor htd = new HTableDescriptor("testIndexPutWithNegativeIntDataTypes");
  HRegionInfo info = new HRegionInfo(htd.getTableName(), "ABC".getBytes(), "BBB".getBytes(), false);
  // HLog hlog = UTIL.getMiniHBaseCluster().getRegionServer(0).getWAL();
  HRegion region = HRegion.createHRegion(info, basedir, conf, htd);
  IndexSpecification spec = new IndexSpecification("index");
  spec.addIndexColumn(new HColumnDescriptor("col"), "ql1", ValueType.Long, 4);

  byte[] value1 = Bytes.toBytes(-2562351L);
  Put p = new Put("row".getBytes());
  p.add("col".getBytes(), "ql1".getBytes(), value1);
  Put indexPut = IndexUtils.prepareIndexPut(p, spec, region);
  long a = -2562351L;
  byte[] expectedResult = Bytes.toBytes(a ^ (1L << 63));
  byte[] actualResult = new byte[8];
  byte[] indexRowKey = indexPut.getRow();
  System.arraycopy(indexRowKey, 22, actualResult, 0, actualResult.length);
  Assert.assertTrue(Bytes.equals(expectedResult, actualResult));
}

@Test
public void testSingleIndexExpressionWithOneEqualsExpression() throws Exception {
  String indexName = "idx1";
  SingleIndexExpression singleIndexExpression = new SingleIndexExpression(indexName);
  byte[] value = "1".getBytes();
  Column column = new Column(FAMILY1, QUALIFIER1);
  EqualsExpression equalsExpression = new EqualsExpression(column, value);
  singleIndexExpression.addEqualsExpression(equalsExpression);

  Scan scan = new Scan();
  scan.setAttribute(Constants.INDEX_EXPRESSION, IndexUtils.toBytes(singleIndexExpression));
  Filter filter = new SingleColumnValueFilter(FAMILY1, QUALIFIER1, CompareOp.EQUAL, value);
  scan.setFilter(filter);
  ScanFilterEvaluator evaluator = new ScanFilterEvaluator();
  List<IndexSpecification> indices = new ArrayList<IndexSpecification>();
  IndexSpecification index = new IndexSpecification(indexName);
  HColumnDescriptor colDesc = new HColumnDescriptor(FAMILY1);
  index.addIndexColumn(colDesc, COL1, ValueType.String, 10);
  indices.add(index);
  HRegion region =
      initHRegion(tableName.getBytes(), null, null,
        "testSingleIndexExpressionWithOneEqualsExpression", TEST_UTIL.getConfiguration(), FAMILY1);
  IndexRegionScanner scanner = evaluator.evaluate(scan, indices, new byte[0], region, tableName);
  // TODO add assertions
}

@Test(timeout = 180000)
public void testPostOpenCoprocessor() throws IOException, KeeperException, InterruptedException {
  String userTableName = "testPostOpenCoprocessor";

  HTableDescriptor ihtd =
    TestUtils.createIndexedHTableDescriptor(userTableName, "col", "Index1", "col", "ql");
  admin.createTable(ihtd);

  // Check the number of indices
  List<IndexSpecification> list = IndexManager.getInstance().getIndicesForTable(userTableName);
  Assert.assertEquals(1, list.size());

  // Check the index name
  boolean bool = false;
  for (IndexSpecification e : list) {
    if (e.getName().equals("Index1")) bool = true;
  }
  Assert.assertTrue(bool);
}

/**
 * @param tableName
 * @param indexName
 * @return index specification
 */
public IndexSpecification getIndex(String tableName, byte[] indexName) {
  Map<byte[], IndexSpecification> indices = this.tableIndexMap.get(tableName);
  if (indices != null) {
    return indices.get(indexName);
  }
  return null;
}

@Override
public boolean preWALWrite(ObserverContext<WALCoprocessorEnvironment> ctx, HRegionInfo info,
    HLogKey logKey, WALEdit logEdit) throws IOException {
  TableName tableName = info.getTable();
  if (IndexUtils.isCatalogOrSystemTable(tableName) || IndexUtils.isIndexTable(tableName)) {
    return true;
  }
  List<IndexSpecification> indices = indexManager.getIndicesForTable(tableName.getNameAsString());
  if (indices != null && !indices.isEmpty()) {
    LOG.trace("Entering preWALWrite for the table " + tableName);
    String indexTableName = IndexUtils.getIndexTableName(tableName);
    IndexEdits iEdits = IndexRegionObserver.threadLocal.get();
    WALEdit indexWALEdit = iEdits.getWALEdit();
    // This size will be 0 when none of the Mutations to the user table to be indexed.
    // or write to WAL is disabled for the Mutations
    if (indexWALEdit.getKeyValues().size() == 0) {
      return true;
    }
    LOG.trace("Adding indexWALEdits into WAL for table " + tableName);
    HRegion indexRegion = iEdits.getRegion();
    // TS in all KVs within WALEdit will be the same. So considering the 1st one.
    Long time = indexWALEdit.getKeyValues().get(0).getTimestamp();
    indexRegion.getLog().appendNoSync(indexRegion.getRegionInfo(),
      TableName.valueOf(indexTableName), indexWALEdit, logKey.getClusterIds(), time,
      indexRegion.getTableDesc(), indexRegion.getSequenceId(), true, HConstants.NO_NONCE,
      HConstants.NO_NONCE);
    LOG.trace("Exiting preWALWrite for the table " + tableName);
  }
  return true;
}

@Override
public void postOpen(ObserverContext<RegionCoprocessorEnvironment> contx) {
  HTableDescriptor tableDesc = contx.getEnvironment().getRegion().getTableDesc();
  RegionServerServices rss = contx.getEnvironment().getRegionServerServices();
  TableName tableName = tableDesc.getTableName();
  if (isNotIndexedTableDescriptor(tableDesc)) {
    return;
  }
  LOG.trace("Entering postOpen for the table " + tableName);
  this.indexManager.incrementRegionCount(tableName.getNameAsString());
  List<IndexSpecification> list = indexManager.getIndicesForTable(tableName.getNameAsString());
  if (null != list) {
    LOG.trace("Index Manager already contains an entry for the table "
        + ". Hence returning from postOpen");
    return;
  }
  byte[] indexBytes = tableDesc.getValue(Constants.INDEX_SPEC_KEY);
  TableIndices tableIndices = new TableIndices();
  try {
    tableIndices.readFields(indexBytes);
  } catch (IOException e) {
    rss.abort("Some unidentified scenario while reading from the "
        + "table descriptor . Aborting RegionServer", e);
  }
  list = tableIndices.getIndices();
  if (list != null && list.size() > 0) {
    indexManager.addIndexForTable(tableName.getNameAsString(), list);
    LOG.trace("Added index Specification in the Manager for the " + tableName);
  }
  LOG.trace("Exiting postOpen for the table " + tableName);
}

@Test(timeout = 180000)
public void testAndOperationWithProperStartAndStopRow() throws Exception {
  Configuration conf = UTIL.getConfiguration();
  String userTableName = "testAndOperationWithProperStartAndStopRow";
  HTableDescriptor ihtd = new HTableDescriptor(userTableName);
  HColumnDescriptor hcd = new HColumnDescriptor("cf1");
  ihtd.addFamily(hcd);
  TableIndices indices = new TableIndices();
  IndexSpecification indexSpecification =
      createIndexSpecification(hcd, ValueType.String, 10, new String[] { "c1" }, "idx1");
  indices.addIndex(indexSpecification);
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  admin.createTable(ihtd);
  HTable table = new HTable(conf, userTableName);
  rangePutForIdx2(table);
  FilterList masterFilter = new FilterList(Operator.MUST_PASS_ALL);
  SingleColumnValueFilter scvf =
      new SingleColumnValueFilter("cf1".getBytes(), "c1".getBytes(), CompareOp.EQUAL,
          Bytes.toBytes("5"));
  masterFilter.addFilter(scvf);
  Scan scan = new Scan();
  scan.setFilter(masterFilter);
  ResultScanner scanner = table.getScanner(scan);
  List<Result> testRes = new ArrayList<Result>();
  Result[] result = scanner.next(1);
  while (result != null && result.length > 0) {
    testRes.add(result[0]);
    result = scanner.next(1);
  }
  assertTrue(testRes.size() == 1);

}

Collection<? extends Mutation> prepareIndexDeletes(Delete delete, HRegion userRegion,
    List<IndexSpecification> indexSpecs, HRegion indexRegion) throws IOException {
  Collection<Delete> indexDeletes = new LinkedHashSet<Delete>();
  for (Entry<byte[], List<Cell>> entry : delete.getFamilyCellMap().entrySet()) {
    for (Cell cell : entry.getValue()) {
      indexDeletes.addAll(getIndexDeletes(indexSpecs, userRegion, indexRegion,
        KeyValueUtil.ensureKeyValue(cell)));
    }
  }
  return indexDeletes;
}

private IndexSpecification createIndexSpecification(HColumnDescriptor hcd, ValueType type,
    int maxValueLength, String[] qualifiers, String name) {
  IndexSpecification index = new IndexSpecification(name.getBytes());
  for (String qualifier : qualifiers) {
    index.addIndexColumn(hcd, qualifier, type, maxValueLength);
  }
  return index;
}

@Override
public void postBatchMutate(final ObserverContext<RegionCoprocessorEnvironment> ctx,
    final MiniBatchOperationInProgress<Mutation> miniBatchOp) {
  HTableDescriptor userTableDesc = ctx.getEnvironment().getRegion().getTableDesc();
  String tableName = userTableDesc.getNameAsString();
  if (isNotIndexedTableDescriptor(userTableDesc)) {
    return;
  }
  List<IndexSpecification> indices = indexManager.getIndicesForTable(tableName);
  if (indices == null || indices.isEmpty()) {
    LOG.trace("skipping postBatchMutate for the table " + tableName + " as there are no indices");
    return;
  }
  LOG.trace("Entering postBatchMutate for the table " + tableName);
  IndexEdits indexEdits = threadLocal.get();
  List<Mutation> indexMutations = indexEdits.getIndexMutations();

  if (indexMutations.size() == 0) {
    return;
  }
  HRegion hr = indexEdits.getRegion();
  try {
    hr.batchMutateForIndex(indexMutations.toArray(new Mutation[indexMutations.size()]));
  } catch (IOException e) {
    // TODO This can come? If so we need to revert the actual put
    // and make the op failed.
    LOG.error("Error putting data into the index region", e);
  }
  LOG.trace("Exiting postBatchMutate for the table " + tableName);
}

private boolean isValidIndexMutation(HTableDescriptor userTableDesc) {
  String tableName = userTableDesc.getTableName().getNameAsString();
  if (IndexUtils.isCatalogOrSystemTable(userTableDesc.getTableName())
      || IndexUtils.isIndexTable(tableName)) {
    return false;
  }
  List<IndexSpecification> indices = indexManager.getIndicesForTable(tableName);
  if (indices == null || indices.isEmpty()) {
    LOG.trace("skipping preBatchMutate for the table " + tableName + " as there are no indices");
    return false;
  }
  return true;
}

@SuppressWarnings("unchecked")
private Map<List<Column>, IndexSpecification> finalizeIndexForLeafNodes(
    Map<Column, LeafFilterNode> leafNodes) {
  // go with the breakups and check
  // suppose there are 5 cols under the AND condition and are c1,c2,c3,c4,c5
  // There can be different break ups for the cols possible.
  // [5],[4,1],[3,2],[3,1,1],[2,2,1],[2,1,1,1],[1,1,1,1,1]
  // In each of these breakup also we can get many columns combinations.
  // Except in first and last where either all cols in one group or 1 column only.
  // For [4,1] there can be 5c1 combinations possible.
  Set<List<List<List<Column>>>> colBreakUps = getColsBreakUps(leafNodes.keySet());
  ColBreakUpIndexDetails bestColBreakUpIndexDetails = null;
  for (List<List<List<Column>>> colBreakUp : colBreakUps) {
    ColBreakUpIndexDetails colBreakUpIndexDetails =
        findBestIndicesForColSplitsInBreakUp(colBreakUp, leafNodes);
    if (colBreakUpIndexDetails == null) {
      continue;
    }
    if (colBreakUpIndexDetails.isBestIndex) {
      // This means this is THE best index. It solves all the columns and exactly those cols only
      // there as part of the indices too.. What else we need...
      bestColBreakUpIndexDetails = colBreakUpIndexDetails;
      break;
    } else {
      if (bestColBreakUpIndexDetails == null
          || isIndicesGroupBetterThanCurBest(colBreakUpIndexDetails, bestColBreakUpIndexDetails)) {
        bestColBreakUpIndexDetails = colBreakUpIndexDetails;
      }
    }
  }
  // TODO some more logging of the output..
  return bestColBreakUpIndexDetails == null ? null
      : bestColBreakUpIndexDetails.bestIndicesForBreakUp;
}

private void addIndicesToNonLeafAndNode(Map<List<Column>, IndexSpecification> colsVsIndex,
    NonLeafFilterNode nonLeafFilterNode, Map<Column, LeafFilterNode> leafNodes) {
  for (Entry<List<Column>, IndexSpecification> entry : colsVsIndex.entrySet()) {
    List<Column> cols = entry.getKey();
    int colsSize = cols.size();
    IndexSpecification index = entry.getValue();
    // The FilterColumnValueDetail for cols need to be in the same order as that of cols
    // in the index. This order will be important for creating the start/stop keys for
    // index scan.
    List<FilterColumnValueDetail> fcvds = new ArrayList<FilterColumnValueDetail>(colsSize);
    int i = 0;
    for (ColumnQualifier cq : index.getIndexColumns()) {
      FilterColumnValueDetail fcvd =
          leafNodes.get(
            new Column(cq.getColumnFamily(), cq.getQualifier(), cq.getValuePartition()))
              .getFilterColumnValueDetail();
      assert fcvd != null;
      fcvds.add(fcvd);
      i++;
      if (i == colsSize) {
        // The selected index might be on more cols than those we are interested in now.
        // All those will be towards the end.
        break;
      }
    }
    LOG.info("Index using for the columns " + cols + " : " + index);
    nonLeafFilterNode.addIndicesToUse(fcvds, index);
  }
}

private IndexSpecification
    findBestIndex(List<Column> cols, Map<Column, LeafFilterNode> leafNodes) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Trying to find a best index for the cols : " + cols);
  }
  Set<IndexSpecification> indicesToUse = getPossibleIndicesForCols(cols, leafNodes);
  // indicesToUse will never come as null....
  if (LOG.isDebugEnabled()) {
    LOG.debug("Possible indices for cols " + cols + " : " + indicesToUse);
  }
  IndexSpecification bestIndex = null;
  int bestIndexCardinality = -1;
  for (IndexSpecification index : indicesToUse) {
    if (isIndexSuitable(index, cols, leafNodes)) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Index " + index + " seems to be suitable for the columns " + cols);
      }
      if (index.getIndexColumns().size() == cols.size()) {
        // Yea we got the best index. Juts return this. No need to loop through and check
        // with other indices
        return index;
      }
      // Compare this index with the current best. This will be better if its cardinality
      // is better(lesser) than the current best's
      // TODO pluggable interface to decide which index to be used when both this and current
      // best index having same cardinality.
      if (bestIndex == null || index.getIndexColumns().size() < bestIndexCardinality) {
        bestIndex = index;
        bestIndexCardinality = index.getIndexColumns().size();
      }
    }
  }
  return bestIndex;
}

private boolean isIndexSuitable(IndexSpecification index, List<Column> cols,
    Map<Column, LeafFilterNode> leafNodes) {
  int matchedCols = 0;
  for (ColumnQualifier cq : index.getIndexColumns()) {
    Column column = new Column(cq.getColumnFamily(), cq.getQualifier(), cq.getValuePartition());
    if (cols.contains(column)) {
      matchedCols++;
      // leafNodes.get(column) will never be null.. Don't worry
      if (leafNodes.get(column).getFilterColumnValueDetail() instanceof FilterColumnValueRange) {
        // When the condition on the column is a range condition, we need to ensure in this index
        // 1. The column is the last column
        // or
        // 2. There are no columns in this index which is part of the cols list
        if (matchedCols != cols.size()) {
          return false;
        }
      }
    } else {
      if (matchedCols != cols.size()) {
        return false;
      }
    }
    if (matchedCols == cols.size()) {
      return true;
    }
  }
  return false;
}

@Test(timeout = 180000)
public void testVerifyTheStopRowIsCorrectInCaseOfGreaterOperatorsInSCVF() throws Exception {
  Configuration conf = UTIL.getConfiguration();
  String tableName = "testDeleteIncosistent";
  IndexSpecification spec1 = new IndexSpecification("idx1");
  HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
  // HTableDescriptor htd = new HTableDescriptor(tableName);
  HColumnDescriptor hcd = new HColumnDescriptor("cf");
  spec1.addIndexColumn(hcd, "detail", ValueType.String, 10);
  htd.addFamily(hcd);
  TableIndices indices = new TableIndices();
  indices.addIndex(spec1);
  htd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  admin.createTable(htd);
  HTable table = new HTable(conf, "testDeleteIncosistent");
  HTable table2 = new HTable(conf, "testDeleteIncosistent_idx");
  Put p = new Put(Bytes.toBytes("row5"));
  p.add("cf".getBytes(), "detail".getBytes(), "5".getBytes());
  table2.put(IndexUtils.prepareIndexPut(p, spec1, HConstants.EMPTY_START_ROW));
  p = new Put(Bytes.toBytes("row6"));
  p.add("cf".getBytes(), "detail".getBytes(), "6".getBytes());
  table.put(p);
  Scan s = new Scan();
  SingleColumnValueFilter filter1 =
      new SingleColumnValueFilter("cf".getBytes(), "detail".getBytes(), CompareOp.GREATER,
          "5".getBytes());
  s.setFilter(filter1);
  ResultScanner scanner = table.getScanner(s);
  int i = 0;
  for (Result result : scanner) {
    i++;
  }
  assertEquals(1, i);
}

@Override
public Map<Column, List<Pair<IndexSpecification, Integer>>> getPossibleFutureUseIndices() {
  // TODO avoid create of Map instance all the time...
  Map<Column, List<Pair<IndexSpecification, Integer>>> reply =
      new HashMap<Column, List<Pair<IndexSpecification, Integer>>>();
  reply.put(filterColumnValueDetail.getColumn(), possibleFutureUseIndices);
  return reply;
}

public IndexFilterNode(IndexSpecification indexToUse,
    List<Pair<IndexSpecification, Integer>> possibleUseIndices,
    List<Pair<IndexSpecification, Integer>> possibleFutureUseIndices,
    FilterColumnValueDetail filterColumnValueDetail) {
  this.indexToUse = indexToUse;
  this.possibleUseIndices = possibleUseIndices;
  this.possibleFutureUseIndices = possibleFutureUseIndices;
  this.filterColumnValueDetail = filterColumnValueDetail;
}

@Override
public Map<List<FilterColumnValueDetail>, IndexSpecification> getIndexToUse() {
  // TODO avoid create of Map instance all the time...
  Map<List<FilterColumnValueDetail>, IndexSpecification> reply =
      new HashMap<List<FilterColumnValueDetail>, IndexSpecification>();
  List<FilterColumnValueDetail> key = new ArrayList<FilterColumnValueDetail>(1);
  key.add(filterColumnValueDetail);
  reply.put(key, indexToUse);
  return reply;
}

@Test(timeout = 180000)
public void testShouldRetrieveNegativeFloatValueWithGreaterCondition() throws Exception {
  Configuration conf = UTIL.getConfiguration();
  String userTableName = "testShouldRetrieveNegativeFloatValueWithGreaterCondition";
  HTableDescriptor ihtd = new HTableDescriptor(TableName.valueOf(userTableName));
  HColumnDescriptor hcd = new HColumnDescriptor("cf1");
  ihtd.addFamily(hcd);
  TableIndices indices = new TableIndices();
  IndexSpecification indexSpecification =
      createIndexSpecification(hcd, ValueType.Float, 10, new String[] { "c1" }, "idx1");
  indices.addIndex(indexSpecification);
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  admin.createTable(ihtd);
  HTable table = new HTable(conf, userTableName);
  rangePutForIdx2WithFloat(table);
  FilterList masterFilter = new FilterList(Operator.MUST_PASS_ALL);
  SingleColumnValueFilter scvf =
      new SingleColumnValueFilter("cf1".getBytes(), "c1".getBytes(), CompareOp.GREATER,
          new FloatComparator(Bytes.toBytes(-5f)));
  masterFilter.addFilter(scvf);
  Scan scan = new Scan();
  scan.setFilter(masterFilter);
  ResultScanner scanner = table.getScanner(scan);
  List<Result> testRes = new ArrayList<Result>();
  Result[] result = scanner.next(1);
  while (result != null && result.length > 0) {
    testRes.add(result[0]);
    result = scanner.next(1);
  }
  assertTrue(testRes.size() == 4);
}

@Test(timeout = 180000)
public void testShouldRetrieveNegativeFloatValueWithLessCondition() throws Exception {
  Configuration conf = UTIL.getConfiguration();
  String userTableName = "testShouldRetrieveNegativeFloatValueWithLessCondition";
  HTableDescriptor ihtd = new HTableDescriptor(TableName.valueOf(userTableName));
  HColumnDescriptor hcd = new HColumnDescriptor("cf1");
  ihtd.addFamily(hcd);
  TableIndices indices = new TableIndices();
  IndexSpecification indexSpecification =
      createIndexSpecification(hcd, ValueType.Float, 10, new String[] { "c1" }, "idx1");
  indices.addIndex(indexSpecification);
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  admin.createTable(ihtd);
  HTable table = new HTable(conf, userTableName);
  rangePutForIdx2WithFloat(table);
  FilterList masterFilter = new FilterList(Operator.MUST_PASS_ALL);
  SingleColumnValueFilter scvf =
      new SingleColumnValueFilter("cf1".getBytes(), "c1".getBytes(), CompareOp.LESS,
          new FloatComparator(Bytes.toBytes(-5f)));

  masterFilter.addFilter(scvf);
  Scan scan = new Scan();
  scan.setFilter(masterFilter);
  ResultScanner scanner = table.getScanner(scan);
  List<Result> testRes = new ArrayList<Result>();
  Result[] result = scanner.next(1);
  while (result != null && result.length > 0) {
    testRes.add(result[0]);
    result = scanner.next(1);
  }
  assertTrue(testRes.size() == 2);
}

public void testAddIndexForTableWhenStringAndValLengthIsZero() throws Exception {
  IndexManager im = IndexManager.getInstance();
  assertNotNull("Index Manager should not be null.", im);

  List<IndexSpecification> indexList = new ArrayList<IndexSpecification>(1);
  IndexSpecification iSpec = new IndexSpecification("index_name");

  iSpec.addIndexColumn(new HColumnDescriptor("cf"), "cq", null, 0);
  indexList.add(iSpec);
  im.addIndexForTable("index_name", indexList);
  indexList = im.getIndicesForTable("index_name");
  assertEquals("the total value length should be 2", 2, indexList.get(0).getTotalValueLength());
}

@Test(timeout = 180000)
public void testPreCreateShouldNotBeSuccessfulIfIndicesAreNotSame() throws IOException,
    KeeperException, InterruptedException {
  ZooKeeperWatcher zkw = HBaseTestingUtility.getZooKeeperWatcher(UTIL);
  String userTableName = "testNotConsisIndex1";
  HTableDescriptor ihtd = new HTableDescriptor(TableName.valueOf(userTableName));
  HColumnDescriptor hcd = new HColumnDescriptor("col");
  IndexSpecification iSpec1 = new IndexSpecification("Index1");
  iSpec1.addIndexColumn(hcd, "q1", ValueType.String, 10);
  ihtd.addFamily(hcd);
  TableIndices indices = new TableIndices();
  indices.addIndex(iSpec1);
  IndexSpecification iSpec2 = new IndexSpecification("Index2");
  iSpec2.addIndexColumn(hcd, "q1", ValueType.Int, 10);
  indices.addIndex(iSpec2);
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  boolean returnVal = false;
  try {
    admin.createTable(ihtd);
    fail("Exception should be thrown");
  } catch (IOException e) {

    returnVal = true;
  }
  Assert.assertTrue(returnVal);
  ZKAssign.blockUntilNoRIT(zkw);
}

@Test(timeout = 180000)
public void testPreCreateShouldNotBeSuccessfulIfIndicesAreNotSameAtLength() throws IOException,
    KeeperException, InterruptedException {
  ZooKeeperWatcher zkw = HBaseTestingUtility.getZooKeeperWatcher(UTIL);
  String userTableName = "testNotConsisIndex2";
  HTableDescriptor ihtd = new HTableDescriptor(TableName.valueOf(userTableName));
  HColumnDescriptor hcd = new HColumnDescriptor("col");
  IndexSpecification iSpec1 = new IndexSpecification("Index1");
  iSpec1.addIndexColumn(hcd, "q1", ValueType.String, 10);
  iSpec1.addIndexColumn(hcd, "q2", ValueType.String, 4);
  ihtd.addFamily(hcd);
  TableIndices indices = new TableIndices();
  indices.addIndex(iSpec1);
  IndexSpecification iSpec2 = new IndexSpecification("Index2");
  iSpec2.addIndexColumn(hcd, "q3", ValueType.String, 10);
  iSpec2.addIndexColumn(hcd, "q2", ValueType.String, 10);
  indices.addIndex(iSpec2);
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  boolean returnVal = false;
  try {
    admin.createTable(ihtd);
    fail("Exception should be thrown");
  } catch (IOException e) {
    returnVal = true;
  }
  Assert.assertTrue(returnVal);
  ZKAssign.blockUntilNoRIT(zkw);
}

@Test(timeout = 180000)
public void testPreCreateShouldNotBeSuccessfulIfIndicesAreNotSameAtType() throws IOException,
    KeeperException, InterruptedException {
  ZooKeeperWatcher zkw = HBaseTestingUtility.getZooKeeperWatcher(UTIL);
  String userTableName = "testNotConsisIndex3";
  HTableDescriptor ihtd = new HTableDescriptor(TableName.valueOf(userTableName));
  HColumnDescriptor hcd = new HColumnDescriptor("col");
  IndexSpecification iSpec1 = new IndexSpecification("Index1");
  iSpec1.addIndexColumn(hcd, "q1", ValueType.String, 10);
  iSpec1.addIndexColumn(hcd, "q2", ValueType.String, 10);
  ihtd.addFamily(hcd);
  IndexSpecification iSpec2 = new IndexSpecification("Index2");
  iSpec2.addIndexColumn(hcd, "q1", ValueType.Int, 10);
  iSpec2.addIndexColumn(hcd, "q3", ValueType.String, 10);
  TableIndices indices = new TableIndices();
  indices.addIndex(iSpec1);
  indices.addIndex(iSpec2);
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  boolean returnVal = false;
  try {
    admin.createTable(ihtd);
    fail("Exception should be thrown");
  } catch (IOException e) {
    returnVal = true;
  }
  Assert.assertTrue(returnVal);
  ZKAssign.blockUntilNoRIT(zkw);
}

public void testWhenORWithSameColumnAppearsinDiffChild() throws Exception {
  Configuration conf = HBaseConfiguration.create();
  region = initHRegion(tableName, "tesWhenORWithSameColumnAppearsinDiffChild", conf, family);
  ScanFilterEvaluator mapper = new ScanFilterEvaluator();
  List<IndexSpecification> indices = new ArrayList<IndexSpecification>();
  // create the indices.
  indices.add(createIndexSpecification("cf1", ValueType.String, 10, new String[] { "c2", "c3" },
    "idx1"));

  indices
      .add(createIndexSpecification("cf1", ValueType.String, 10, new String[] { "c1" }, "idx2"));

  FilterList masterFilter = new FilterList(Operator.MUST_PASS_ONE);
  // create the filter
  FilterList filter = new FilterList(Operator.MUST_PASS_ALL);
  SingleColumnValueFilter iscvf1 =
      new SingleColumnValueFilter("cf1".getBytes(), "c1".getBytes(), CompareOp.GREATER,
          "a".getBytes());
  SingleColumnValueFilter iscvf2 =
      new SingleColumnValueFilter("cf1".getBytes(), "c2".getBytes(), CompareOp.EQUAL,
          "K".getBytes());
  filter.addFilter(iscvf1);
  filter.addFilter(iscvf2);

  SingleColumnValueFilter iscvf3 =
      new SingleColumnValueFilter("cf1".getBytes(), "c1".getBytes(), CompareOp.GREATER,
          "d".getBytes());
  // filter2.addFilter(iscvf3);

  masterFilter.addFilter(filter);
  masterFilter.addFilter(iscvf3);
  Scan scan = new Scan();
  scan.setFilter(masterFilter);

  IndexManager.getInstance().addIndexForTable(this.region.getTableDesc().getNameAsString(),
    indices);
  mapper.evaluate(scan, indices, new byte[0], this.region, this.region.getTableDesc()
      .getNameAsString());
}

@Test(timeout = 180000)
public void testIndexPutWithOffsetAndLengthWhenPutIsSmallerThanOffset() throws IOException {
  Path basedir = new Path(DIR + "TestIndexPut");
  Configuration conf = TEST_UTIL.getConfiguration();
  HTableDescriptor htd =
      new HTableDescriptor(
          TableName.valueOf("testIndexPutWithOffsetAndLengthWhenPutIsSmallerThanOffset"));
  HRegionInfo info = new HRegionInfo(htd.getTableName(), "ABC".getBytes(), "BBB".getBytes(), false);
  HRegion region = HRegion.createHRegion(info, basedir, conf, htd);
  IndexSpecification spec = new IndexSpecification("index");
  spec.addIndexColumn(new HColumnDescriptor("col"), "ql1", new SpatialPartition(20, 2),
    ValueType.String, 18);

  byte[] value1 = "AB---CD---EF---GH".getBytes();
  Put p = new Put("row".getBytes());
  p.add("col".getBytes(), "ql1".getBytes(), value1);
  Put indexPut = IndexUtils.prepareIndexPut(p, spec, region);
  byte[] indexRowKey = indexPut.getRow();
  byte[] actualResult = new byte[2];
  System.arraycopy(indexRowKey, 22, actualResult, 0, actualResult.length);
  byte[] expectedResult = new byte[2];
  Assert.assertTrue(Bytes.equals(actualResult, expectedResult));

  value1 = "AB---CD---EF---GH---I".getBytes();
  p = new Put("row".getBytes());
  p.add("col".getBytes(), "ql1".getBytes(), value1);
  indexPut = IndexUtils.prepareIndexPut(p, spec, region);
  indexRowKey = indexPut.getRow();
  actualResult = new byte[2];
  System.arraycopy(indexRowKey, 22, actualResult, 0, actualResult.length);
  expectedResult = new byte[2];
  expectedResult[0] = 'I';
  Assert.assertTrue(Bytes.equals(actualResult, expectedResult));
}

@Test(timeout = 180000)
public void testIndexManagerCleanUp() throws Exception {
  Configuration conf = UTIL.getConfiguration();
  conf.setBoolean("hbase.use.secondary.index", true);
  String userTableName = "testIndexManagerCleanUp";
  HTableDescriptor ihtd = new HTableDescriptor(TableName.valueOf(userTableName));
  HColumnDescriptor hcd = new HColumnDescriptor("col1");
  ihtd.addFamily(hcd);
  IndexSpecification iSpec = new IndexSpecification("Index1");
  iSpec.addIndexColumn(hcd, "ql", ValueType.String, 10);
  TableIndices indices = new TableIndices();
  indices.addIndex(iSpec);
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());

  byte[][] splits = new byte[10][];
  char c = 'A';
  for (int i = 0; i < 10; i++) {
    byte[] b = { (byte) c };
    splits[i] = b;
    c++;
  }
  admin.createTable(ihtd, splits);
  IndexManager instance = IndexManager.getInstance();
  int regionCount = instance.getTableRegionCount(userTableName);
  Assert.assertEquals(11, regionCount);

  admin.disableTable(Bytes.toBytes(userTableName));
  regionCount = instance.getTableRegionCount(userTableName);
  Assert.assertEquals(0, regionCount);

  admin.enableTable(userTableName);
  regionCount = instance.getTableRegionCount(userTableName);
  Assert.assertEquals(11, regionCount);
}

@Test(timeout = 180000)
public void testCachingWithValuesDistributedAmongMulitpleRegions() throws Exception {
  Configuration conf = UTIL.getConfiguration();
  String userTableName = "testCachingWithValuesDistributedAmongMulitpleRegions";
  HTableDescriptor ihtd = new HTableDescriptor(TableName.valueOf(userTableName));
  HColumnDescriptor hcd = new HColumnDescriptor("cf1");
  ihtd.addFamily(hcd);
  TableIndices indices = new TableIndices();
  IndexSpecification indexSpecification =
      createIndexSpecification(hcd, ValueType.String, 10, new String[] { "c1" }, "idx1");
  indices.addIndex(indexSpecification);
  byte[][] split =
      new byte[][] { Bytes.toBytes("row1"), Bytes.toBytes("row2"), Bytes.toBytes("row3"),
          Bytes.toBytes("row4") };
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  admin.createTable(ihtd, split);
  HTable table = new HTable(conf, userTableName);
  insert100Rows(table);
  FilterList masterFilter = new FilterList(Operator.MUST_PASS_ALL);
  SingleColumnValueFilter scvf =
      new SingleColumnValueFilter("cf1".getBytes(), "c1".getBytes(), CompareOp.EQUAL,
          Bytes.toBytes("5"));
  masterFilter.addFilter(scvf);
  Scan scan = new Scan();
  scan.setCaching(5);
  scan.setFilter(masterFilter);
  ResultScanner scanner = table.getScanner(scan);
  List<Result> testRes = new ArrayList<Result>();
  Result[] result = scanner.next(1);
  int i = 0;
  while (result != null && result.length > 0) {
    System.out.println(Bytes.toString(result[0].getRow()));
    testRes.add(result[0]);
    result = scanner.next(1);
    i++;
  }
  assertEquals(8, i);
}

@Test(timeout = 180000)
public void testShouldRetrieveNegativeDoubleValueWithLesserThanEqualsCondition() throws Exception {
  Configuration conf = UTIL.getConfiguration();
  String userTableName = "testShouldRetrieveNegativeDoubleValueWithLesserThanEqualsCondition";
  HTableDescriptor ihtd = new HTableDescriptor(TableName.valueOf(userTableName));
  HColumnDescriptor hcd = new HColumnDescriptor("cf1");
  ihtd.addFamily(hcd);
  TableIndices indices = new TableIndices();
  IndexSpecification indexSpecification =
      createIndexSpecification(hcd, ValueType.Double, 10, new String[] { "c1" }, "idx1");
  indices.addIndex(indexSpecification);
  ihtd.setValue(Constants.INDEX_SPEC_KEY, indices.toByteArray());
  admin.createTable(ihtd);
  HTable table = new HTable(conf, userTableName);
  rangePutForIdx2WithDouble(table);
  FilterList masterFilter = new FilterList(Operator.MUST_PASS_ALL);
  SingleColumnValueFilter scvf =
      new SingleColumnValueFilter("cf1".getBytes(), "c1".getBytes(), CompareOp.LESS_OR_EQUAL,
          new DoubleComparator(Bytes.toBytes(-5.3d)));
  masterFilter.addFilter(scvf);
  Scan scan = new Scan();
  scan.setFilter(masterFilter);
  ResultScanner scanner = table.getScanner(scan);
  List<Result> testRes = new ArrayList<Result>();
  Result[] result = scanner.next(1);
  while (result != null && result.length > 0) {
    testRes.add(result[0]);
    result = scanner.next(1);
  }
  assertTrue(testRes.size() == 2);
}