@Test public void testGetIndexPredications() throws Exception { Column rowkeyColumn = new Column("rk", Type.TEXT); ScanNode scanNode = new ScanNode(1); EvalNode evalNode1 = new BinaryEval(EvalType.GEQ, new FieldEval(rowkeyColumn), new ConstEval(new TextDatum("020"))); EvalNode evalNode2 = new BinaryEval(EvalType.LEQ, new FieldEval(rowkeyColumn), new ConstEval(new TextDatum("055"))); EvalNode evalNodeA = new BinaryEval(EvalType.AND, evalNode1, evalNode2); scanNode.setQual(evalNodeA); HBaseTablespace storageManager = (HBaseTablespace) TablespaceManager.getByName("cluster1"); List<Set<EvalNode>> indexEvals = storageManager.findIndexablePredicateSet(scanNode.getQual(), new Column[]{rowkeyColumn}); assertNotNull(indexEvals); assertEquals(1, indexEvals.size()); Pair<Datum, Datum> indexPredicateValue = storageManager.getIndexablePredicateValue(null, indexEvals.get(0)); assertEquals("020", indexPredicateValue.getFirst().asChars()); assertEquals("055", indexPredicateValue.getSecond().asChars()); EvalNode evalNode3 = new BinaryEval(EvalType.EQUAL, new FieldEval(rowkeyColumn),new ConstEval(new TextDatum("075"))); EvalNode evalNodeB = new BinaryEval(EvalType.OR, evalNodeA, evalNode3); scanNode.setQual(evalNodeB); indexEvals = storageManager.findIndexablePredicateSet(scanNode.getQual(), new Column[]{rowkeyColumn}); assertEquals(2, indexEvals.size()); indexPredicateValue = storageManager.getIndexablePredicateValue(null, indexEvals.get(0)); assertEquals("020", indexPredicateValue.getFirst().asChars()); assertEquals("055", indexPredicateValue.getSecond().asChars()); indexPredicateValue = storageManager.getIndexablePredicateValue(null, indexEvals.get(1)); assertEquals("075", indexPredicateValue.getFirst().asChars()); assertEquals("075", indexPredicateValue.getSecond().asChars()); EvalNode evalNode4 = new BinaryEval(EvalType.GEQ, new FieldEval(rowkeyColumn), new ConstEval(new TextDatum("072"))); EvalNode evalNode5 = new BinaryEval(EvalType.LEQ, new FieldEval(rowkeyColumn), new ConstEval(new TextDatum("078"))); EvalNode evalNodeC = new BinaryEval(EvalType.AND, evalNode4, evalNode5); EvalNode evalNodeD = new BinaryEval(EvalType.OR, evalNodeA, evalNodeC); scanNode.setQual(evalNodeD); indexEvals = storageManager.findIndexablePredicateSet(scanNode.getQual(), new Column[]{rowkeyColumn}); assertEquals(2, indexEvals.size()); indexPredicateValue = storageManager.getIndexablePredicateValue(null, indexEvals.get(0)); assertEquals("020", indexPredicateValue.getFirst().asChars()); assertEquals("055", indexPredicateValue.getSecond().asChars()); indexPredicateValue = storageManager.getIndexablePredicateValue(null, indexEvals.get(1)); assertEquals("072", indexPredicateValue.getFirst().asChars()); assertEquals("078", indexPredicateValue.getSecond().asChars()); evalNode4 = new BinaryEval(EvalType.GEQ, new FieldEval(rowkeyColumn), new ConstEval(new TextDatum("072"))); evalNode5 = new BinaryEval(EvalType.LEQ, new FieldEval(rowkeyColumn), new ConstEval(new TextDatum("078"))); evalNodeC = new BinaryEval(EvalType.AND, evalNode4, evalNode5); EvalNode evalNode6 = new BinaryEval(EvalType.GEQ, new FieldEval(rowkeyColumn), new ConstEval(new TextDatum("073"))); evalNodeD = new BinaryEval(EvalType.AND, evalNodeC, evalNode6); EvalNode evalNodeE = new BinaryEval(EvalType.OR, evalNodeA, evalNodeD); scanNode.setQual(evalNodeE); indexEvals = storageManager.findIndexablePredicateSet(scanNode.getQual(), new Column[]{rowkeyColumn}); assertEquals(2, indexEvals.size()); indexPredicateValue = storageManager.getIndexablePredicateValue(null, indexEvals.get(0)); assertEquals("020", indexPredicateValue.getFirst().asChars()); assertEquals("055", indexPredicateValue.getSecond().asChars()); indexPredicateValue = storageManager.getIndexablePredicateValue(null, indexEvals.get(1)); assertEquals("073", indexPredicateValue.getFirst().asChars()); assertEquals("078", indexPredicateValue.getSecond().asChars()); }
public List<IndexPredication> getIndexPredications(ColumnMapping columnMapping, TableDesc tableDesc, @Nullable EvalNode filterCondition) throws IOException, MissingTablePropertyException, InvalidTablePropertyException { final Column[] indexableColumns = getIndexableColumns(tableDesc); if (indexableColumns == null || indexableColumns.length == 0) { return Collections.EMPTY_LIST; } return findIndexablePredicateSet(filterCondition, indexableColumns).stream() .map(set -> getIndexablePredicateValue(columnMapping, set)) .filter(value -> value != null) .map(value -> new IndexPredication( indexableColumns[0], tableDesc.getLogicalSchema().getColumnId(indexableColumns[0].getQualifiedName()), value.getFirst(), value.getSecond())) .collect(Collectors.toList()); }
HBaseTablespace extends Tablespace { public List<IndexPredication> getIndexPredications(ColumnMapping columnMapping, TableDesc tableDesc, @Nullable EvalNode filterCondition) throws IOException, MissingTablePropertyException, InvalidTablePropertyException { final Column[] indexableColumns = getIndexableColumns(tableDesc); if (indexableColumns == null || indexableColumns.length == 0) { return Collections.EMPTY_LIST; } return findIndexablePredicateSet(filterCondition, indexableColumns).stream() .map(set -> getIndexablePredicateValue(columnMapping, set)) .filter(value -> value != null) .map(value -> new IndexPredication( indexableColumns[0], tableDesc.getLogicalSchema().getColumnId(indexableColumns[0].getQualifiedName()), value.getFirst(), value.getSecond())) .collect(Collectors.toList()); } }
HBaseTablespace extends Tablespace { public List<IndexPredication> getIndexPredications(ColumnMapping columnMapping, TableDesc tableDesc, @Nullable EvalNode filterCondition) throws IOException, MissingTablePropertyException, InvalidTablePropertyException { final Column[] indexableColumns = getIndexableColumns(tableDesc); if (indexableColumns == null || indexableColumns.length == 0) { return Collections.EMPTY_LIST; } return findIndexablePredicateSet(filterCondition, indexableColumns).stream() .map(set -> getIndexablePredicateValue(columnMapping, set)) .filter(value -> value != null) .map(value -> new IndexPredication( indexableColumns[0], tableDesc.getLogicalSchema().getColumnId(indexableColumns[0].getQualifiedName()), value.getFirst(), value.getSecond())) .collect(Collectors.toList()); } HBaseTablespace(String spaceName, URI uri, JSONObject config); }
HBaseTablespace extends Tablespace { public List<IndexPredication> getIndexPredications(ColumnMapping columnMapping, TableDesc tableDesc, @Nullable EvalNode filterCondition) throws IOException, MissingTablePropertyException, InvalidTablePropertyException { final Column[] indexableColumns = getIndexableColumns(tableDesc); if (indexableColumns == null || indexableColumns.length == 0) { return Collections.EMPTY_LIST; } return findIndexablePredicateSet(filterCondition, indexableColumns).stream() .map(set -> getIndexablePredicateValue(columnMapping, set)) .filter(value -> value != null) .map(value -> new IndexPredication( indexableColumns[0], tableDesc.getLogicalSchema().getColumnId(indexableColumns[0].getQualifiedName()), value.getFirst(), value.getSecond())) .collect(Collectors.toList()); } HBaseTablespace(String spaceName, URI uri, JSONObject config); @Override void storageInit(); Configuration getHbaseConf(); @Override long getTableVolume(TableDesc table, Optional<EvalNode> filter); @Override void close(); @Override void createTable(TableDesc tableDesc, boolean ifNotExists); static HTableDescriptor parseHTableDescriptor(TableMeta tableMeta, Schema schema); @Override void purgeTable(TableDesc tableDesc); @Override URI getTableUri(TableMeta meta, String databaseName, String tableName); @Override List<Fragment> getSplits(String inputSourceId,
                                  TableDesc table,
                                  boolean requireSorted,
                                  @Nullable EvalNode filterCondition); @Override Appender getAppenderForInsertRow(OverridableConf queryContext,
                                          TaskAttemptId taskAttemptId,
                                          TableMeta meta,
                                          Schema schema,
                                          Path workDir); @Override Appender getAppender(OverridableConf queryContext,
                              TaskAttemptId taskAttemptId, TableMeta meta, Schema schema, Path workDir); HConnection getConnection(); List<IndexPredication> getIndexPredications(ColumnMapping columnMapping,
                                                     TableDesc tableDesc,
                                                     @Nullable EvalNode filterCondition); List<Set<EvalNode>> findIndexablePredicateSet(@Nullable EvalNode qual,
                                                       Column[] indexableColumns); static boolean isIndexableOperator(EvalNode expr); Pair<Datum, Datum> getIndexablePredicateValue(ColumnMapping columnMapping,
                                                       Set<EvalNode> evalNodes); @Override Path commitTable(OverridableConf queryContext, ExecutionBlockId finalEbId,
                          LogicalPlan plan, Schema schema,
                          TableDesc tableDesc); @Override TupleRange[] getInsertSortRanges(OverridableConf queryContext, TableDesc tableDesc,
                                          Schema inputSchema, SortSpec[] sortSpecs, TupleRange dataRange); @Override void rewritePlan(OverridableConf context, LogicalPlan plan); @Override StorageProperty getProperty(); @Override FormatProperty getFormatProperty(TableMeta meta); void prepareTable(LogicalNode node); @Override void rollbackTable(LogicalNode node); @Override URI getStagingUri(OverridableConf context, String queryId, TableMeta meta); URI prepareStagingSpace(TajoConf conf, String queryId, OverridableConf context,
                                 TableMeta meta); @Override void verifySchemaToWrite(TableDesc tableDesc, Schema outSchema); }
HBaseTablespace extends Tablespace { public List<IndexPredication> getIndexPredications(ColumnMapping columnMapping, TableDesc tableDesc, @Nullable EvalNode filterCondition) throws IOException, MissingTablePropertyException, InvalidTablePropertyException { final Column[] indexableColumns = getIndexableColumns(tableDesc); if (indexableColumns == null || indexableColumns.length == 0) { return Collections.EMPTY_LIST; } return findIndexablePredicateSet(filterCondition, indexableColumns).stream() .map(set -> getIndexablePredicateValue(columnMapping, set)) .filter(value -> value != null) .map(value -> new IndexPredication( indexableColumns[0], tableDesc.getLogicalSchema().getColumnId(indexableColumns[0].getQualifiedName()), value.getFirst(), value.getSecond())) .collect(Collectors.toList()); } HBaseTablespace(String spaceName, URI uri, JSONObject config); @Override void storageInit(); Configuration getHbaseConf(); @Override long getTableVolume(TableDesc table, Optional<EvalNode> filter); @Override void close(); @Override void createTable(TableDesc tableDesc, boolean ifNotExists); static HTableDescriptor parseHTableDescriptor(TableMeta tableMeta, Schema schema); @Override void purgeTable(TableDesc tableDesc); @Override URI getTableUri(TableMeta meta, String databaseName, String tableName); @Override List<Fragment> getSplits(String inputSourceId,
                                  TableDesc table,
                                  boolean requireSorted,
                                  @Nullable EvalNode filterCondition); @Override Appender getAppenderForInsertRow(OverridableConf queryContext,
                                          TaskAttemptId taskAttemptId,
                                          TableMeta meta,
                                          Schema schema,
                                          Path workDir); @Override Appender getAppender(OverridableConf queryContext,
                              TaskAttemptId taskAttemptId, TableMeta meta, Schema schema, Path workDir); HConnection getConnection(); List<IndexPredication> getIndexPredications(ColumnMapping columnMapping,
                                                     TableDesc tableDesc,
                                                     @Nullable EvalNode filterCondition); List<Set<EvalNode>> findIndexablePredicateSet(@Nullable EvalNode qual,
                                                       Column[] indexableColumns); static boolean isIndexableOperator(EvalNode expr); Pair<Datum, Datum> getIndexablePredicateValue(ColumnMapping columnMapping,
                                                       Set<EvalNode> evalNodes); @Override Path commitTable(OverridableConf queryContext, ExecutionBlockId finalEbId,
                          LogicalPlan plan, Schema schema,
                          TableDesc tableDesc); @Override TupleRange[] getInsertSortRanges(OverridableConf queryContext, TableDesc tableDesc,
                                          Schema inputSchema, SortSpec[] sortSpecs, TupleRange dataRange); @Override void rewritePlan(OverridableConf context, LogicalPlan plan); @Override StorageProperty getProperty(); @Override FormatProperty getFormatProperty(TableMeta meta); void prepareTable(LogicalNode node); @Override void rollbackTable(LogicalNode node); @Override URI getStagingUri(OverridableConf context, String queryId, TableMeta meta); URI prepareStagingSpace(TajoConf conf, String queryId, OverridableConf context,
                                 TableMeta meta); @Override void verifySchemaToWrite(TableDesc tableDesc, Schema outSchema); static final StorageProperty HBASE_STORAGE_PROPERTIES; static final FormatProperty HFILE_FORMAT_PROPERTIES; static final FormatProperty PUT_MODE_PROPERTIES; static final List<byte []> EMPTY_START_ROW_KEY; static final List<byte []> EMPTY_END_ROW_KEY; }