cgivre commented on code in PR #3067: URL: https://github.com/apache/drill/pull/3067#discussion_r3814420299
########## contrib/storage-accumulo/src/main/java/org/apache/drill/exec/store/accumulo/AccumuloFilterBuilder.java: ########## @@ -0,0 +1,326 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.drill.exec.store.accumulo; + +import java.util.Arrays; +import java.util.List; + +import org.apache.drill.common.FunctionNames; +import org.apache.drill.common.expression.BooleanOperator; +import org.apache.drill.common.expression.FunctionCall; +import org.apache.drill.common.expression.LogicalExpression; +import org.apache.drill.common.expression.SchemaPath; +import org.apache.drill.common.expression.visitors.AbstractExprVisitor; + +/** + * Builds Accumulo scan specifications from Drill filter expressions. + * + * <p>This class converts Drill's LogicalExpression filter representation into + * Accumulo scan parameters (start row, stop row). It focuses on row key + * predicates since those can be efficiently pushed down to Accumulo's scan range.</p> + * + * <p>Supported predicates on row_key:</p> + * <ul> + * <li>row_key = 'value' → exact range</li> + * <li>row_key > 'value' → start row (exclusive)</li> + * <li>row_key >= 'value' → start row (inclusive)</li> + * <li>row_key < 'value' → stop row (exclusive)</li> + * <li>row_key <= 'value' → stop row (inclusive)</li> + * <li>AND combinations → intersect ranges</li> + * <li>OR combinations → union ranges (if contiguous)</li> + * </ul> + */ +public class AccumuloFilterBuilder + extends AbstractExprVisitor<AccumuloScanSpec, Void, RuntimeException> + implements DrillAccumuloConstants { + + private final AccumuloGroupScan groupScan; + private final LogicalExpression filterExpression; + private boolean allExpressionsConverted = true; + + public AccumuloFilterBuilder(AccumuloGroupScan groupScan, LogicalExpression filterExpression) { + this.groupScan = groupScan; + this.filterExpression = filterExpression; + } + + /** + * Parses the filter expression and returns an updated scan specification. + * + * @return the scan spec with row key ranges, or null if no filters can be pushed + */ + public AccumuloScanSpec parseTree() { + AccumuloScanSpec parsedSpec = filterExpression.accept(this, null); + if (parsedSpec != null) { + // Merge with existing scan spec + parsedSpec = mergeScanSpecs(FunctionNames.AND, groupScan.getScanSpec(), parsedSpec); + } + return parsedSpec; + } + + /** + * Returns true if all filter expressions were converted to Accumulo scan parameters. + * If false, the filter operator should remain in the plan for client-side filtering. + */ + public boolean isAllExpressionsConverted() { + return allExpressionsConverted; + } + + @Override + public AccumuloScanSpec visitUnknown(LogicalExpression e, Void value) throws RuntimeException { + allExpressionsConverted = false; + return null; + } + + @Override + public AccumuloScanSpec visitBooleanOperator(BooleanOperator op, Void value) + throws RuntimeException { + return visitFunctionCall(op, value); + } + + @Override + public AccumuloScanSpec visitFunctionCall(FunctionCall call, Void value) + throws RuntimeException { + AccumuloScanSpec nodeScanSpec = null; + String functionName = call.getName(); + List<LogicalExpression> args = call.args(); + + if (AccumuloCompareFunctionsProcessor.isCompareFunction(functionName)) { + AccumuloCompareFunctionsProcessor processor = + AccumuloCompareFunctionsProcessor.createFunctionsProcessorInstance(call); + if (processor.isSuccess()) { + nodeScanSpec = createScanSpecFromComparison(processor); + } + } else { + switch (functionName) { + case FunctionNames.AND: + case FunctionNames.OR: + AccumuloScanSpec firstScanSpec = args.get(0).accept(this, null); + for (int i = 1; i < args.size(); ++i) { + AccumuloScanSpec nextScanSpec = args.get(i).accept(this, null); + if (firstScanSpec != null && nextScanSpec != null) { + nodeScanSpec = mergeScanSpecs(functionName, firstScanSpec, nextScanSpec); + } else { + allExpressionsConverted = false; + if (FunctionNames.AND.equals(functionName)) { + // For AND, keep whichever spec we have + nodeScanSpec = firstScanSpec == null ? nextScanSpec : firstScanSpec; + } + // For OR, if either is null we can't push down the whole OR + } + firstScanSpec = nodeScanSpec; + } + break; + default: + // Unknown function + break; + } + } + + if (nodeScanSpec == null) { + allExpressionsConverted = false; + } + + return nodeScanSpec; + } + + /** + * Creates a scan spec from a comparison processor result. + */ + private AccumuloScanSpec createScanSpecFromComparison( + AccumuloCompareFunctionsProcessor processor) { + + String functionName = processor.getFunctionName(); + SchemaPath field = processor.getPath(); + byte[] fieldValue = processor.getValue(); + + // Only handle row_key predicates for now + boolean isRowKey = field.getRootSegmentPath().equalsIgnoreCase(ROW_KEY); + if (!isRowKey) { + // Column predicates require iterators - not supported in Option A + return null; + } + + byte[] startRow = null; + byte[] stopRow = null; + boolean startRowInclusive = true; + boolean stopRowInclusive = false; + + switch (functionName) { + case FunctionNames.EQ: + // row_key = 'value' → scan exactly that row + startRow = fieldValue; + // Stop row should be just after the value + stopRow = Arrays.copyOf(fieldValue, fieldValue.length + 1); + startRowInclusive = true; + stopRowInclusive = false; + break; + + case FunctionNames.NE: + // row_key != 'value' → can't efficiently push down (would need full scan minus one row) + return null; + + case FunctionNames.GE: + // row_key >= 'value' → start at value (inclusive) + startRow = fieldValue; + startRowInclusive = true; + break; + + case FunctionNames.GT: + // row_key > 'value' → start just after value + startRow = Arrays.copyOf(fieldValue, fieldValue.length + 1); + startRowInclusive = true; + break; + + case FunctionNames.LE: + // row_key <= 'value' → stop just after value + stopRow = Arrays.copyOf(fieldValue, fieldValue.length + 1); + stopRowInclusive = false; + break; + + case FunctionNames.LT: + // row_key < 'value' → stop at value (exclusive) + stopRow = fieldValue; + stopRowInclusive = false; + break; + + default: + return null; + } + + return new AccumuloScanSpec( + groupScan.getTableName(), + startRow, + stopRow, + startRowInclusive, + stopRowInclusive, + groupScan.getScanSpec().getColumns(), + null, // No filter expression needed when using row ranges + groupScan.getScanSpec().getLimit(), + groupScan.getScanSpec().isUseSortedScanner(), + groupScan.getScanSpec().isSortDescending()); + } + + /** + * Merges two scan specs using AND or OR logic. + */ + private AccumuloScanSpec mergeScanSpecs( + String functionName, + AccumuloScanSpec leftSpec, + AccumuloScanSpec rightSpec) { + + byte[] startRow = null; + byte[] stopRow = null; + boolean startRowInclusive = true; + boolean stopRowInclusive = false; + + switch (functionName) { + case FunctionNames.AND: + // AND: Take the intersection (max of starts, min of stops) + startRow = maxOfStartRows(leftSpec.getStartRow(), rightSpec.getStartRow()); + stopRow = minOfStopRows(leftSpec.getStopRow(), rightSpec.getStopRow()); + break; + + case FunctionNames.OR: + // OR: Take the union (min of starts, max of stops) Review Comment: Good catch, this was a real correctness bug. The union of the two ranges is a superset of the disjunction, but the builder still reported `allExpressionsConverted == true`, so `AccumuloPushFilterIntoScan` dropped the filter and the scan returned everything between 'a' and 'z'. Fixed in a7b5c66: an OR merge now sets `allExpressionsConverted = false`, so the range still narrows the scan but the Drill filter stays in the plan to discard the rows in between. While in there I also fixed a second bug in the same path: the OR union used the AND helpers, which treat a null bound as "take the other side" rather than "unbounded". That made `row_key < 'x' OR row_key > 'y'` collapse into an inverted range. Union now propagates null (unbounded). Added two integration tests for this — `testFilterOnRowKeyDisjunction` and `testFilterOnRowKeyDisjunctionOfRanges`. Both fail on the old code (the first returned 9 rows instead of 2) and pass now. ########## contrib/storage-accumulo/src/main/resources/drill-module.conf: ########## @@ -0,0 +1,36 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# This file tells Drill to consider this module when class path scanning. +# This file can also include any supplementary configuration information. +# This file is in HOCON format, see https://github.com/typesafehub/config/blob/master/HOCON.md for more information. + +drill: { + classpath.scanning: { + packages += "org.apache.drill.exec.store.accumulo" + } + + exec: { + accumulo.scan: { + # Number of rows to sample for schema inference + samplerows.count: 100, Review Comment: Right, those were leftovers — the values are hardcoded in Java (`DrillAccumuloConstants.DEFAULT_BATCH_SIZE` and `DrillAccumuloTable.COLUMN_FAMILY_SAMPLE_SIZE`) and nothing ever read the config keys. Removed the `exec.accumulo.scan` block in a7b5c66, and the record reader now uses `DEFAULT_BATCH_SIZE` instead of its own duplicate 4000. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
