From 60cd893e9d3e01ff9c641f8c952f21d82cb92025 Mon Sep 17 00:00:00 2001 From: Heng Qian Date: Fri, 4 Jul 2025 18:26:06 +0800 Subject: [PATCH 1/9] Support partial filter push down Signed-off-by: Heng Qian --- .../sql/calcite/remote/CalciteExplainIT.java | 12 +++++++ .../calcite/explain_partial_filter_push.json | 6 ++++ .../OpenSearchFilterIndexScanRule.java | 13 ++++--- .../opensearch/request/PredicateAnalyzer.java | 26 ++++++++------ .../scan/AbstractCalciteIndexScan.java | 17 ++++++--- .../storage/scan/CalciteLogicalIndexScan.java | 36 +++++++++++++------ 6 files changed, 81 insertions(+), 29 deletions(-) create mode 100644 integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java index 666b0bda365..ee0476749d2 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java @@ -59,4 +59,16 @@ public void supportSearchSargPushDown_timeRange() throws IOException { + "| where birthdate >= '2016-12-08 00:00:00.000000000' " + "and birthdate < '2018-11-09 00:00:00.000000000' ")); } + + // Only for Calcite + @Test + public void supportPartialPushDown() throws IOException { + // field `address` is text type without keyword subfield, so we cannot push it down. + String query = + "source=opensearch-sql_test_index_account | where age >= 1 and address = '880 Holmes Lane'" + + " | fields age, address"; + var result = explainQueryToString(query); + String expected = loadFromFile("expectedOutput/calcite/explain_partial_filter_push.json"); + assertJsonEqualsIgnoreId(expected, result); + } } diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json new file mode 100644 index 00000000000..c6e737532b2 --- /dev/null +++ b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json @@ -0,0 +1,6 @@ +{ + "calcite": { + "logical": "LogicalProject(age=[$8], address=[$2])\n LogicalFilter(condition=[AND(>=($8, 1), =($2, '880 Holmes Lane'))])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n", + "physical": "EnumerableCalc(expr#0..1=[{inputs}], expr#2=[1], expr#3=[>=($t1, $t2)], expr#4=['880 Holmes Lane':VARCHAR], expr#5=[=($t0, $t4)], expr#6=[AND($t3, $t5)], age=[$t1], address=[$t0], $condition=[$t6])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[FILTER->AND(>=($8, 1), =($2, '880 Holmes Lane')), PROJECT->[address, age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"range\":{\"age\":{\"from\":1,\"to\":null,\"include_lower\":true,\"include_upper\":true,\"boost\":1.0}}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"_source\":{\"includes\":[\"address\",\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n" + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/physical/OpenSearchFilterIndexScanRule.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/physical/OpenSearchFilterIndexScanRule.java index 234dbaf4e43..1c14a07a510 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/physical/OpenSearchFilterIndexScanRule.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/physical/OpenSearchFilterIndexScanRule.java @@ -7,9 +7,11 @@ import java.util.function.Predicate; import org.apache.calcite.plan.RelOptRuleCall; import org.apache.calcite.plan.RelRule; +import org.apache.calcite.rel.AbstractRelNode; import org.apache.calcite.rel.core.Filter; import org.apache.calcite.rel.logical.LogicalFilter; import org.immutables.value.Value; +import org.opensearch.sql.opensearch.storage.scan.AbstractCalciteIndexScan.PushDownType; import org.opensearch.sql.opensearch.storage.scan.CalciteLogicalIndexScan; /** Planner rule that push a {@link LogicalFilter} down to {@link CalciteLogicalIndexScan} */ @@ -27,7 +29,10 @@ public void onMatch(RelOptRuleCall call) { // the ordinary variant final LogicalFilter filter = call.rel(0); final CalciteLogicalIndexScan scan = call.rel(1); - apply(call, filter, scan); + // If the same filter has already been pushed down, skip it. + if (!scan.getPushDownContext().containsDigest(PushDownType.FILTER, filter.getCondition())) { + apply(call, filter, scan); + } } else { throw new AssertionError( String.format( @@ -37,9 +42,9 @@ public void onMatch(RelOptRuleCall call) { } protected void apply(RelOptRuleCall call, Filter filter, CalciteLogicalIndexScan scan) { - CalciteLogicalIndexScan newScan = scan.pushDownFilter(filter); - if (newScan != null) { - call.transformTo(newScan); + AbstractRelNode newRel = scan.pushDownFilter(filter); + if (newRel != null) { + call.transformTo(newRel); } } diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java index a29e148783d..94b71af3b02 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java @@ -144,17 +144,15 @@ private PredicateAnalyzer() {} public static QueryBuilder analyze( RexNode expression, List schema, Map filedTypes) throws ExpressionNotAnalyzableException { + return analyze_(expression, schema, filedTypes).builder(); + } + + public static QueryExpression analyze_( + RexNode expression, List schema, Map filedTypes) + throws ExpressionNotAnalyzableException { requireNonNull(expression, "expression"); try { - // visits expression tree - QueryExpression queryExpression = - (QueryExpression) expression.accept(new Visitor(schema, filedTypes)); - - if (queryExpression != null && queryExpression.isPartial()) { - throw new UnsupportedOperationException( - "Can't handle partial QueryExpression: " + queryExpression); - } - return queryExpression != null ? queryExpression.builder() : null; + return (QueryExpression) expression.accept(new Visitor(schema, filedTypes)); } catch (Throwable e) { Throwables.throwIfInstanceOf(e, UnsupportedOperationException.class); throw new ExpressionNotAnalyzableException("Can't convert " + expression, e); @@ -689,7 +687,7 @@ private static boolean isColumn( interface Expression {} /** Main expression operators (like {@code equals}, {@code gt}, {@code exists} etc.) */ - abstract static class QueryExpression implements Expression { + public abstract static class QueryExpression implements Expression { public abstract QueryBuilder builder(); @@ -964,7 +962,13 @@ private String getFieldReference() { } private String getFieldReferenceForTermQuery() { - return rel.getReferenceForTermQuery(); + String reference = rel.getReferenceForTermQuery(); + // Throw exception in advance of method builder() to trigger partial push down. + if (reference == null) { + throw new PredicateAnalyzerException( + "Field reference for term query cannot be null for " + rel.getRootName()); + } + return reference; } private SimpleQueryExpression(NamedFieldExpression rel) { diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java index 45d16ceb6bd..075566bffc6 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java @@ -121,9 +121,17 @@ public boolean isAggregatePushed() { isAggregatePushed = !isEmpty() && super.peekLast().type == PushDownType.AGGREGATION; return isAggregatePushed; } + + public boolean containsDigest(PushDownType type, Object digest) { + return stream().anyMatch(action -> action.type == type && action.digest.equals(digest)); + } + + public boolean containsArg(PushDownType type, Object arg) { + return stream().anyMatch(action -> action.type == type && action.arg.equals(arg)); + } } - protected enum PushDownType { + public enum PushDownType { FILTER, PROJECT, AGGREGATION, @@ -133,9 +141,10 @@ protected enum PushDownType { // NESTED } - public record PushDownAction(PushDownType type, Object digest, AbstractAction action) { - static PushDownAction of(PushDownType type, Object digest, AbstractAction action) { - return new PushDownAction(type, digest, action); + public record PushDownAction( + PushDownType type, Object digest, AbstractAction action, Object arg) { + static PushDownAction of(PushDownType type, Object digest, AbstractAction action, Object arg) { + return new PushDownAction(type, digest, action, arg); } public String toString() { diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index a8edc18735d..62a3f3e8cf8 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -19,6 +19,7 @@ import org.apache.calcite.plan.RelOptRule; import org.apache.calcite.plan.RelOptTable; import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.AbstractRelNode; import org.apache.calcite.rel.RelCollation; import org.apache.calcite.rel.RelCollations; import org.apache.calcite.rel.RelFieldCollation; @@ -48,6 +49,7 @@ import org.opensearch.sql.opensearch.planner.physical.OpenSearchIndexRules; import org.opensearch.sql.opensearch.request.AggregateAnalyzer; import org.opensearch.sql.opensearch.request.PredicateAnalyzer; +import org.opensearch.sql.opensearch.request.PredicateAnalyzer.QueryExpression; import org.opensearch.sql.opensearch.response.agg.OpenSearchAggregationResponseParser; import org.opensearch.sql.opensearch.storage.OpenSearchIndex; @@ -97,20 +99,30 @@ public void register(RelOptPlanner planner) { } } - public CalciteLogicalIndexScan pushDownFilter(Filter filter) { + public AbstractRelNode pushDownFilter(Filter filter) { try { - CalciteLogicalIndexScan newScan = this.copyWithNewSchema(filter.getRowType()); List schema = this.getRowType().getFieldNames(); Map filedTypes = this.osIndex.getFieldTypes(); - QueryBuilder filterBuilder = - PredicateAnalyzer.analyze(filter.getCondition(), schema, filedTypes); + QueryExpression queryExpression = + PredicateAnalyzer.analyze_(filter.getCondition(), schema, filedTypes); + QueryBuilder queryBuilder = queryExpression.builder(); + // If the same QueryBuilder has already been pushed down, skip it. + if (this.pushDownContext.containsArg(PushDownType.FILTER, queryBuilder)) return null; + CalciteLogicalIndexScan newScan = this.copyWithNewSchema(filter.getRowType()); + // TODO: handle the case where condition contains a score function newScan.pushDownContext.add( PushDownAction.of( PushDownType.FILTER, filter.getCondition(), - requestBuilder -> requestBuilder.pushDownFilter(filterBuilder))); + requestBuilder -> requestBuilder.pushDownFilter(queryBuilder), + queryBuilder)); - // TODO: handle the case where condition contains a score function + // If the query expression is partial, we need to replace the input of the filter with the + // partial pushed scan and then return the original filter itself. + if (queryExpression.isPartial()) { + filter.replaceInput(0, newScan); + return filter; + } return newScan; } catch (Exception e) { if (LOG.isDebugEnabled()) { @@ -161,7 +173,8 @@ public CalciteLogicalIndexScan pushDownProject(List selectedColumns) { PushDownAction.of( PushDownType.PROJECT, newSchema.getFieldNames(), - requestBuilder -> requestBuilder.pushDownProjectStream(projectedFields.stream()))); + requestBuilder -> requestBuilder.pushDownProjectStream(projectedFields.stream()), + projectedFields)); return newScan; } @@ -216,7 +229,8 @@ public CalciteLogicalIndexScan pushDownAggregate(Aggregate aggregate, Project pr requestBuilder -> { requestBuilder.pushDownAggregation(aggregationBuilder); requestBuilder.pushTypeMapping(extendedTypeMapping); - })); + }, + aggregationBuilder)); return newScan; } catch (Exception e) { if (LOG.isDebugEnabled()) { @@ -235,7 +249,8 @@ public CalciteLogicalIndexScan pushDownLimit(Integer limit, Integer offset) { PushDownAction.of( PushDownType.LIMIT, limit, - requestBuilder -> requestBuilder.pushDownLimit(limit, offset))); + requestBuilder -> requestBuilder.pushDownLimit(limit, offset), + Pair.of(limit, offset))); return newScan; } catch (Exception e) { if (LOG.isDebugEnabled()) { @@ -303,7 +318,8 @@ public CalciteLogicalIndexScan pushDownSort(List collations) PushDownAction.of( PushDownType.SORT, builders.toString(), - requestBuilder -> requestBuilder.pushDownSort(builders))); + requestBuilder -> requestBuilder.pushDownSort(builders), + builders)); return newScan; } catch (Exception e) { if (LOG.isDebugEnabled()) { From 996a82bde78439c34cd031a31a6805a6425c4565 Mon Sep 17 00:00:00 2001 From: Heng Qian Date: Fri, 4 Jul 2025 18:30:23 +0800 Subject: [PATCH 2/9] Add doc for PushDownAction Signed-off-by: Heng Qian --- .../storage/scan/AbstractCalciteIndexScan.java | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java index 075566bffc6..8a195e7e525 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java @@ -141,6 +141,15 @@ public enum PushDownType { // NESTED } + /** + * Represents a push down action that can be applied to an OpenSearchRequestBuilder. + * + * @param type PushDownType enum + * @param digest the digest of the pushed down operator + * @param action the lambda action to apply on the OpenSearchRequestBuilder + * @param arg the argument for the action, which can be null if not needed, currently only used + * for identifying whether the same QueryBuilder is ever pushed down + */ public record PushDownAction( PushDownType type, Object digest, AbstractAction action, Object arg) { static PushDownAction of(PushDownType type, Object digest, AbstractAction action, Object arg) { From b23d7ce68b01c54a5d6198ea3566d4c8359eed2d Mon Sep 17 00:00:00 2001 From: Heng Qian Date: Fri, 4 Jul 2025 21:02:55 +0800 Subject: [PATCH 3/9] Fix IT Signed-off-by: Heng Qian --- .../expectedOutput/calcite/explain_partial_filter_push.json | 2 +- .../sql/opensearch/storage/scan/CalciteLogicalIndexScan.java | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json index c6e737532b2..0f268b0ddaa 100644 --- a/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json +++ b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json @@ -1,6 +1,6 @@ { "calcite": { "logical": "LogicalProject(age=[$8], address=[$2])\n LogicalFilter(condition=[AND(>=($8, 1), =($2, '880 Holmes Lane'))])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n", - "physical": "EnumerableCalc(expr#0..1=[{inputs}], expr#2=[1], expr#3=[>=($t1, $t2)], expr#4=['880 Holmes Lane':VARCHAR], expr#5=[=($t0, $t4)], expr#6=[AND($t3, $t5)], age=[$t1], address=[$t0], $condition=[$t6])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[FILTER->AND(>=($8, 1), =($2, '880 Holmes Lane')), PROJECT->[address, age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"range\":{\"age\":{\"from\":1,\"to\":null,\"include_lower\":true,\"include_upper\":true,\"boost\":1.0}}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"_source\":{\"includes\":[\"address\",\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n" + "physical": "EnumerableCalc(expr#0..1=[{inputs}], expr#2=[1], expr#3=[>=($t1, $t2)], expr#4=['880 Holmes Lane':VARCHAR], expr#5=[=($t0, $t4)], expr#6=[AND($t3, $t5)], age=[$t1], address=[$t0], $condition=[$t6])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[address, age], FILTER->AND(>=($1, 1), =($0, '880 Holmes Lane'))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"range\":{\"age\":{\"from\":1,\"to\":null,\"include_lower\":true,\"include_upper\":true,\"boost\":1.0}}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"_source\":{\"includes\":[\"address\",\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n" } } diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index 62a3f3e8cf8..1e29bfa14bf 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -120,8 +120,7 @@ public AbstractRelNode pushDownFilter(Filter filter) { // If the query expression is partial, we need to replace the input of the filter with the // partial pushed scan and then return the original filter itself. if (queryExpression.isPartial()) { - filter.replaceInput(0, newScan); - return filter; + return filter.copy(filter.getTraitSet(), newScan, filter.getCondition()); } return newScan; } catch (Exception e) { From f55ca70689f43d67187ab4f7a6367c84e765c828 Mon Sep 17 00:00:00 2001 From: Heng Qian Date: Mon, 7 Jul 2025 15:28:13 +0800 Subject: [PATCH 4/9] Refine code to only keep non-push-down condition in the new filter Signed-off-by: Heng Qian --- .../calcite/explain_partial_filter_push.json | 2 +- .../OpenSearchFilterIndexScanRule.java | 6 +- .../opensearch/request/PredicateAnalyzer.java | 285 ++++++++---------- .../storage/scan/CalciteLogicalIndexScan.java | 16 +- 4 files changed, 137 insertions(+), 172 deletions(-) diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json index 0f268b0ddaa..613ab805751 100644 --- a/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json +++ b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json @@ -1,6 +1,6 @@ { "calcite": { "logical": "LogicalProject(age=[$8], address=[$2])\n LogicalFilter(condition=[AND(>=($8, 1), =($2, '880 Holmes Lane'))])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n", - "physical": "EnumerableCalc(expr#0..1=[{inputs}], expr#2=[1], expr#3=[>=($t1, $t2)], expr#4=['880 Holmes Lane':VARCHAR], expr#5=[=($t0, $t4)], expr#6=[AND($t3, $t5)], age=[$t1], address=[$t0], $condition=[$t6])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[address, age], FILTER->AND(>=($1, 1), =($0, '880 Holmes Lane'))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"range\":{\"age\":{\"from\":1,\"to\":null,\"include_lower\":true,\"include_upper\":true,\"boost\":1.0}}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"_source\":{\"includes\":[\"address\",\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n" + "physical": "EnumerableCalc(expr#0..1=[{inputs}], expr#2=['880 Holmes Lane':VARCHAR], expr#3=[=($t0, $t2)], age=[$t1], address=[$t0], $condition=[$t3])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[address, age], FILTER->AND(>=($1, 1), =($0, '880 Holmes Lane'))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"range\":{\"age\":{\"from\":1,\"to\":null,\"include_lower\":true,\"include_upper\":true,\"boost\":1.0}}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"_source\":{\"includes\":[\"address\",\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n" } } diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/physical/OpenSearchFilterIndexScanRule.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/physical/OpenSearchFilterIndexScanRule.java index 1c14a07a510..4ee5148b826 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/physical/OpenSearchFilterIndexScanRule.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/physical/OpenSearchFilterIndexScanRule.java @@ -11,7 +11,6 @@ import org.apache.calcite.rel.core.Filter; import org.apache.calcite.rel.logical.LogicalFilter; import org.immutables.value.Value; -import org.opensearch.sql.opensearch.storage.scan.AbstractCalciteIndexScan.PushDownType; import org.opensearch.sql.opensearch.storage.scan.CalciteLogicalIndexScan; /** Planner rule that push a {@link LogicalFilter} down to {@link CalciteLogicalIndexScan} */ @@ -29,10 +28,7 @@ public void onMatch(RelOptRuleCall call) { // the ordinary variant final LogicalFilter filter = call.rel(0); final CalciteLogicalIndexScan scan = call.rel(1); - // If the same filter has already been pushed down, skip it. - if (!scan.getPushDownContext().containsDigest(PushDownType.FILTER, filter.getCondition())) { - apply(call, filter, scan); - } + apply(call, filter, scan); } else { throw new AssertionError( String.format( diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java index 94b71af3b02..a041f8131eb 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java @@ -52,6 +52,7 @@ import java.util.Locale; import java.util.Map; import java.util.Set; +import lombok.Getter; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rex.RexCall; @@ -542,6 +543,7 @@ private QueryExpression andOr(RexCall call) { QueryExpression[] expressions = new QueryExpression[call.getOperands().size()]; PredicateAnalyzerException firstError = null; boolean partial = false; + int failedCount = 0; for (int i = 0; i < call.getOperands().size(); i++) { try { Expression expr = call.getOperands().get(i).accept(this); @@ -556,6 +558,10 @@ private QueryExpression andOr(RexCall call) { firstError = e; } partial = true; + ++failedCount; + // If we cannot analyze the operand, wrap the RexNode with UnAnalyzableQueryExpression and + // record them in the array. We will reuse them later. + expressions[i] = new UnAnalyzableQueryExpression(call.getOperands().get(i)); } } @@ -571,6 +577,11 @@ private QueryExpression andOr(RexCall call) { } return CompoundQueryExpression.or(expressions); case AND: + if (failedCount == call.getOperands().size()) { + // If all operands failed, we cannot analyze the AND expression. + throw new PredicateAnalyzerException( + "All expressions in AND failed to analyze: " + call); + } return CompoundQueryExpression.and(partial, expressions); default: String message = format(Locale.ROOT, "Unable to handle call: [%s]", call); @@ -695,68 +706,126 @@ public boolean isPartial() { return false; } - public abstract QueryExpression contains(LiteralExpression literal); - /** Negate {@code this} QueryExpression (not the next one). */ - public abstract QueryExpression not(); - - public abstract QueryExpression exists(); + QueryExpression not() { + throw new PredicateAnalyzerException("not cannot be applied to " + this.getClass()); + } - public abstract QueryExpression notExists(); + QueryExpression exists() { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['exists'] " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression like(LiteralExpression literal); + QueryExpression notExists() { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['notExists'] " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression notLike(LiteralExpression literal); + QueryExpression contains(LiteralExpression literal) { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['contains'] " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression equals(LiteralExpression literal); + QueryExpression between(Range literal, boolean isTimeStamp) { + throw new PredicateAnalyzer.PredicateAnalyzerException( + "between cannot be applied to " + this.getClass()); + } - public abstract QueryExpression in(LiteralExpression literal); + QueryExpression like(LiteralExpression literal) { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['like'] " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression notIn(LiteralExpression literal); + QueryExpression notLike(LiteralExpression literal) { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['notLike'] " + "cannot be applied to " + this.getClass()); + } - public QueryExpression between(Range literal, boolean isTimeStamp) { - throw new PredicateAnalyzer.PredicateAnalyzerException( - "between cannot be applied to " + this.getClass()); + QueryExpression equals(LiteralExpression literal) { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['='] " + "cannot be applied to " + this.getClass()); } - public QueryExpression equals(Object point, boolean isTimeStamp) { + QueryExpression equals(Object point, boolean isTimeStamp) { throw new PredicateAnalyzer.PredicateAnalyzerException( "equals cannot be applied to " + this.getClass()); } - public abstract QueryExpression notEquals(LiteralExpression literal); + QueryExpression notEquals(LiteralExpression literal) { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['not'] " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression gt(LiteralExpression literal); + QueryExpression gt(LiteralExpression literal) { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['>'] " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression gte(LiteralExpression literal); + QueryExpression gte(LiteralExpression literal) { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['>='] " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression lt(LiteralExpression literal); + QueryExpression lt(LiteralExpression literal) { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['<'] " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression lte(LiteralExpression literal); + QueryExpression lte(LiteralExpression literal) { + throw new PredicateAnalyzerException( + "SqlOperatorImpl ['<='] " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression match(String query, Map optionalArguments); + QueryExpression match(String query, Map optionalArguments) { + throw new PredicateAnalyzerException("Match " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression matchPhrase( - String query, Map optionalArguments); + QueryExpression matchPhrase(String query, Map optionalArguments) { + throw new PredicateAnalyzerException( + "MatchPhrase " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression matchBoolPrefix( - String query, Map optionalArguments); + QueryExpression matchBoolPrefix(String query, Map optionalArguments) { + throw new PredicateAnalyzerException( + "MatchBoolPrefix " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression matchPhrasePrefix( - String query, Map optionalArguments); + QueryExpression matchPhrasePrefix(String query, Map optionalArguments) { + throw new PredicateAnalyzerException( + "MatchPhrasePrefix " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression simpleQueryString( - RexCall fieldsRexCall, String query, Map optionalArguments); + QueryExpression simpleQueryString( + RexCall fieldsRexCall, String query, Map optionalArguments) { + throw new PredicateAnalyzerException( + "SimpleQueryString " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression queryString( - RexCall fieldsRexCall, String query, Map optionalArguments); + QueryExpression queryString( + RexCall fieldsRexCall, String query, Map optionalArguments) { + throw new PredicateAnalyzerException( + "QueryString " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression multiMatch( - RexCall fieldsRexCall, String query, Map optionalArguments); + QueryExpression multiMatch( + RexCall fieldsRexCall, String query, Map optionalArguments) { + throw new PredicateAnalyzerException( + "MultiMatch " + "cannot be applied to " + this.getClass()); + } - public abstract QueryExpression isTrue(); + QueryExpression isTrue() { + throw new PredicateAnalyzerException("isTrue cannot be applied to " + this.getClass()); + } + + QueryExpression in(LiteralExpression literal) { + throw new PredicateAnalyzerException("in cannot be applied to " + this.getClass()); + } + + QueryExpression notIn(LiteralExpression literal) { + throw new PredicateAnalyzerException("notIn cannot be applied to " + this.getClass()); + } - public static QueryExpression create(TerminalExpression expression) { + static QueryExpression create(TerminalExpression expression) { if (expression instanceof CastExpression) { expression = CastExpression.unpack(expression); } @@ -770,11 +839,26 @@ public static QueryExpression create(TerminalExpression expression) { } } + @Getter + static class UnAnalyzableQueryExpression extends QueryExpression { + RexNode rexNode; + + public UnAnalyzableQueryExpression(RexNode rexNode) { + this.rexNode = requireNonNull(rexNode, "rexNode"); + } + + @Override + public QueryBuilder builder() { + return null; + } + } + /** Builds conjunctions / disjunctions based on existing expressions. */ - static class CompoundQueryExpression extends QueryExpression { + public static class CompoundQueryExpression extends QueryExpression { private final boolean partial; private final BoolQueryBuilder builder; + @Getter private final List unAnalyzableNodes = new ArrayList<>(); public static CompoundQueryExpression or(QueryExpression... expressions) { CompoundQueryExpression bqe = new CompoundQueryExpression(false); @@ -794,7 +878,9 @@ public static CompoundQueryExpression or(QueryExpression... expressions) { public static CompoundQueryExpression and(boolean partial, QueryExpression... expressions) { CompoundQueryExpression bqe = new CompoundQueryExpression(partial); for (QueryExpression expression : expressions) { - if (expression != null) { // partial expressions have nulls for missing nodes + if (expression instanceof UnAnalyzableQueryExpression unAnalyzableQueryExpression) { + bqe.unAnalyzableNodes.add(unAnalyzableQueryExpression.rexNode); + } else { bqe.builder.must(expression.builder()); } } @@ -824,131 +910,6 @@ public QueryBuilder builder() { public QueryExpression not() { return new CompoundQueryExpression(partial, boolQuery().mustNot(builder())); } - - @Override - public QueryExpression exists() { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['exists'] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression contains(LiteralExpression literal) { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['contains'] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression notExists() { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['notExists'] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression like(LiteralExpression literal) { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['like'] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression notLike(LiteralExpression literal) { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['notLike'] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression equals(LiteralExpression literal) { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['='] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression notEquals(LiteralExpression literal) { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['not'] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression gt(LiteralExpression literal) { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['>'] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression gte(LiteralExpression literal) { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['>='] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression lt(LiteralExpression literal) { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['<'] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression lte(LiteralExpression literal) { - throw new PredicateAnalyzerException( - "SqlOperatorImpl ['<='] " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression match(String query, Map optionalArguments) { - throw new PredicateAnalyzerException("Match " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression matchPhrase(String query, Map optionalArguments) { - throw new PredicateAnalyzerException( - "MatchPhrase " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression matchBoolPrefix(String query, Map optionalArguments) { - throw new PredicateAnalyzerException( - "MatchBoolPrefix " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression matchPhrasePrefix(String query, Map optionalArguments) { - throw new PredicateAnalyzerException( - "MatchPhrasePrefix " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression simpleQueryString( - RexCall fieldsRexCall, String query, Map optionalArguments) { - throw new PredicateAnalyzerException( - "SimpleQueryString " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression queryString( - RexCall fieldsRexCall, String query, Map optionalArguments) { - throw new PredicateAnalyzerException( - "QueryString " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression multiMatch( - RexCall fieldsRexCall, String query, Map optionalArguments) { - throw new PredicateAnalyzerException( - "MultiMatch " + "cannot be applied to a compound expression"); - } - - @Override - public QueryExpression isTrue() { - throw new PredicateAnalyzerException("isTrue cannot be applied to a compound expression"); - } - - @Override - public QueryExpression in(LiteralExpression literal) { - throw new PredicateAnalyzerException("in cannot be applied to a compound expression"); - } - - @Override - public QueryExpression notIn(LiteralExpression literal) { - throw new PredicateAnalyzerException("notIn cannot be applied to a compound expression"); - } } /** Usually basic expression of type {@code a = 'val'} or {@code b > 42}. */ diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index 1e29bfa14bf..96724f2405a 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -31,6 +31,8 @@ import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rel.type.RelDataTypeFactory; import org.apache.calcite.rel.type.RelDataTypeField; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.fun.SqlStdOperatorTable; import org.apache.commons.lang3.tuple.Pair; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -49,6 +51,7 @@ import org.opensearch.sql.opensearch.planner.physical.OpenSearchIndexRules; import org.opensearch.sql.opensearch.request.AggregateAnalyzer; import org.opensearch.sql.opensearch.request.PredicateAnalyzer; +import org.opensearch.sql.opensearch.request.PredicateAnalyzer.CompoundQueryExpression; import org.opensearch.sql.opensearch.request.PredicateAnalyzer.QueryExpression; import org.opensearch.sql.opensearch.response.agg.OpenSearchAggregationResponseParser; import org.opensearch.sql.opensearch.storage.OpenSearchIndex; @@ -106,8 +109,6 @@ public AbstractRelNode pushDownFilter(Filter filter) { QueryExpression queryExpression = PredicateAnalyzer.analyze_(filter.getCondition(), schema, filedTypes); QueryBuilder queryBuilder = queryExpression.builder(); - // If the same QueryBuilder has already been pushed down, skip it. - if (this.pushDownContext.containsArg(PushDownType.FILTER, queryBuilder)) return null; CalciteLogicalIndexScan newScan = this.copyWithNewSchema(filter.getRowType()); // TODO: handle the case where condition contains a score function newScan.pushDownContext.add( @@ -118,9 +119,16 @@ public AbstractRelNode pushDownFilter(Filter filter) { queryBuilder)); // If the query expression is partial, we need to replace the input of the filter with the - // partial pushed scan and then return the original filter itself. + // partial pushed scan and the filter condition with non-pushed-down conditions. if (queryExpression.isPartial()) { - return filter.copy(filter.getTraitSet(), newScan, filter.getCondition()); + // Only CompoundQueryExpression could be partial. + List conditions = + ((CompoundQueryExpression) queryExpression).getUnAnalyzableNodes(); + RexNode newCondition = + conditions.size() > 1 + ? getCluster().getRexBuilder().makeCall(SqlStdOperatorTable.AND, conditions) + : conditions.get(0); + return filter.copy(filter.getTraitSet(), newScan, newCondition); } return newScan; } catch (Exception e) { From a1159e181ec7bd88926ec09695f62b1adaf87ef1 Mon Sep 17 00:00:00 2001 From: Heng Qian Date: Wed, 9 Jul 2025 11:15:26 +0800 Subject: [PATCH 5/9] Refine code Signed-off-by: Heng Qian --- .../storage/scan/AbstractCalciteIndexScan.java | 17 +++-------------- .../storage/scan/CalciteLogicalIndexScan.java | 15 +++++---------- 2 files changed, 8 insertions(+), 24 deletions(-) diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java index 8a195e7e525..1bada937191 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/AbstractCalciteIndexScan.java @@ -121,14 +121,6 @@ public boolean isAggregatePushed() { isAggregatePushed = !isEmpty() && super.peekLast().type == PushDownType.AGGREGATION; return isAggregatePushed; } - - public boolean containsDigest(PushDownType type, Object digest) { - return stream().anyMatch(action -> action.type == type && action.digest.equals(digest)); - } - - public boolean containsArg(PushDownType type, Object arg) { - return stream().anyMatch(action -> action.type == type && action.arg.equals(arg)); - } } public enum PushDownType { @@ -147,13 +139,10 @@ public enum PushDownType { * @param type PushDownType enum * @param digest the digest of the pushed down operator * @param action the lambda action to apply on the OpenSearchRequestBuilder - * @param arg the argument for the action, which can be null if not needed, currently only used - * for identifying whether the same QueryBuilder is ever pushed down */ - public record PushDownAction( - PushDownType type, Object digest, AbstractAction action, Object arg) { - static PushDownAction of(PushDownType type, Object digest, AbstractAction action, Object arg) { - return new PushDownAction(type, digest, action, arg); + public record PushDownAction(PushDownType type, Object digest, AbstractAction action) { + static PushDownAction of(PushDownType type, Object digest, AbstractAction action) { + return new PushDownAction(type, digest, action); } public String toString() { diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index 96724f2405a..613c17254f1 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -115,8 +115,7 @@ public AbstractRelNode pushDownFilter(Filter filter) { PushDownAction.of( PushDownType.FILTER, filter.getCondition(), - requestBuilder -> requestBuilder.pushDownFilter(queryBuilder), - queryBuilder)); + requestBuilder -> requestBuilder.pushDownFilter(queryBuilder))); // If the query expression is partial, we need to replace the input of the filter with the // partial pushed scan and the filter condition with non-pushed-down conditions. @@ -180,8 +179,7 @@ public CalciteLogicalIndexScan pushDownProject(List selectedColumns) { PushDownAction.of( PushDownType.PROJECT, newSchema.getFieldNames(), - requestBuilder -> requestBuilder.pushDownProjectStream(projectedFields.stream()), - projectedFields)); + requestBuilder -> requestBuilder.pushDownProjectStream(projectedFields.stream()))); return newScan; } @@ -236,8 +234,7 @@ public CalciteLogicalIndexScan pushDownAggregate(Aggregate aggregate, Project pr requestBuilder -> { requestBuilder.pushDownAggregation(aggregationBuilder); requestBuilder.pushTypeMapping(extendedTypeMapping); - }, - aggregationBuilder)); + })); return newScan; } catch (Exception e) { if (LOG.isDebugEnabled()) { @@ -256,8 +253,7 @@ public CalciteLogicalIndexScan pushDownLimit(Integer limit, Integer offset) { PushDownAction.of( PushDownType.LIMIT, limit, - requestBuilder -> requestBuilder.pushDownLimit(limit, offset), - Pair.of(limit, offset))); + requestBuilder -> requestBuilder.pushDownLimit(limit, offset))); return newScan; } catch (Exception e) { if (LOG.isDebugEnabled()) { @@ -325,8 +321,7 @@ public CalciteLogicalIndexScan pushDownSort(List collations) PushDownAction.of( PushDownType.SORT, builders.toString(), - requestBuilder -> requestBuilder.pushDownSort(builders), - builders)); + requestBuilder -> requestBuilder.pushDownSort(builders))); return newScan; } catch (Exception e) { if (LOG.isDebugEnabled()) { From 1c7e1f50ea32000747aecf99dfa8c88530c609bb Mon Sep 17 00:00:00 2001 From: Heng Qian Date: Mon, 14 Jul 2025 16:58:35 +0800 Subject: [PATCH 6/9] Only show the pushed conditions in the PushDownContext Signed-off-by: Heng Qian --- .../sql/calcite/remote/CalciteExplainIT.java | 17 +++++- .../calcite/explain_partial_filter_push.json | 4 +- .../calcite/explain_partial_filter_push2.json | 6 ++ .../opensearch/request/PredicateAnalyzer.java | 57 +++++++++++++++++-- .../storage/scan/CalciteLogicalIndexScan.java | 21 ++++--- 5 files changed, 88 insertions(+), 17 deletions(-) create mode 100644 integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push2.json diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java index ee0476749d2..962b16230ff 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java @@ -65,10 +65,23 @@ public void supportSearchSargPushDown_timeRange() throws IOException { public void supportPartialPushDown() throws IOException { // field `address` is text type without keyword subfield, so we cannot push it down. String query = - "source=opensearch-sql_test_index_account | where age >= 1 and address = '880 Holmes Lane'" - + " | fields age, address"; + "source=opensearch-sql_test_index_account | where (state = 'Seattle' or age < 10) and (age" + + " >= 1 and address = '880 Holmes Lane') | fields age, address"; var result = explainQueryToString(query); String expected = loadFromFile("expectedOutput/calcite/explain_partial_filter_push.json"); assertJsonEqualsIgnoreId(expected, result); } + + // Only for Calcite + @Test + public void supportPartialPushDown2() throws IOException { + // field `gender` and `address` are both text type without keyword subfield, so we cannot push + // them down. + String query = + "source=opensearch-sql_test_index_account | where (gender = 'M' or age < 10) and (age >= 1" + + " and address = '880 Holmes Lane') | fields age, address"; + var result = explainQueryToString(query); + String expected = loadFromFile("expectedOutput/calcite/explain_partial_filter_push2.json"); + assertJsonEqualsIgnoreId(expected, result); + } } diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json index 613ab805751..642fe5cd517 100644 --- a/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json +++ b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push.json @@ -1,6 +1,6 @@ { "calcite": { - "logical": "LogicalProject(age=[$8], address=[$2])\n LogicalFilter(condition=[AND(>=($8, 1), =($2, '880 Holmes Lane'))])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n", - "physical": "EnumerableCalc(expr#0..1=[{inputs}], expr#2=['880 Holmes Lane':VARCHAR], expr#3=[=($t0, $t2)], age=[$t1], address=[$t0], $condition=[$t3])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[address, age], FILTER->AND(>=($1, 1), =($0, '880 Holmes Lane'))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"range\":{\"age\":{\"from\":1,\"to\":null,\"include_lower\":true,\"include_upper\":true,\"boost\":1.0}}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"_source\":{\"includes\":[\"address\",\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n" + "logical": "LogicalProject(age=[$8], address=[$2])\n LogicalFilter(condition=[AND(OR(=($7, 'Seattle'), <($8, 10)), >=($8, 1), =($2, '880 Holmes Lane'))])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n", + "physical": "EnumerableCalc(expr#0..1=[{inputs}], expr#2=['880 Holmes Lane':VARCHAR], expr#3=[=($t0, $t2)], age=[$t1], address=[$t0], $condition=[$t3])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[address, state, age], FILTER->AND(OR(=($1, 'Seattle'), <($2, 10)), >=($2, 1)), PROJECT->[address, age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"bool\":{\"should\":[{\"term\":{\"state.keyword\":{\"value\":\"Seattle\",\"boost\":1.0}}},{\"range\":{\"age\":{\"from\":null,\"to\":10,\"include_lower\":true,\"include_upper\":false,\"boost\":1.0}}}],\"adjust_pure_negative\":true,\"boost\":1.0}},{\"range\":{\"age\":{\"from\":1,\"to\":null,\"include_lower\":true,\"include_upper\":true,\"boost\":1.0}}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"_source\":{\"includes\":[\"address\",\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n" } } diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push2.json b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push2.json new file mode 100644 index 00000000000..5e19f97f0c6 --- /dev/null +++ b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push2.json @@ -0,0 +1,6 @@ +{ + "calcite": { + "logical": "LogicalProject(age=[$8], address=[$2])\n LogicalFilter(condition=[AND(OR(=($4, 'M'), <($8, 10)), >=($8, 1), =($2, '880 Holmes Lane'))])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n", + "physical": "EnumerableCalc(expr#0..2=[{inputs}], expr#3=['M'], expr#4=[=($t1, $t3)], expr#5=[10], expr#6=[<($t2, $t5)], expr#7=[OR($t4, $t6)], expr#8=['880 Holmes Lane':VARCHAR], expr#9=[=($t0, $t8)], expr#10=[AND($t7, $t9)], age=[$t2], address=[$t0], $condition=[$t10])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[address, gender, age], FILTER->>=($2, 1)], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"range\":{\"age\":{\"from\":1,\"to\":null,\"include_lower\":true,\"include_upper\":true,\"boost\":1.0}}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"_source\":{\"includes\":[\"address\",\"gender\",\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n" + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java index a041f8131eb..bbb17554c91 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java @@ -551,6 +551,9 @@ private QueryExpression andOr(RexCall call) { // nop currently } else { expressions[i] = (QueryExpression) call.getOperands().get(i).accept(this); + // Update or simplify the analyzed node list if it is not partial. + if (!expressions[i].isPartial()) + expressions[i].updateAnalyzedNodes(call.getOperands().get(i)); } partial |= expressions[i].isPartial(); } catch (PredicateAnalyzerException e) { @@ -702,6 +705,12 @@ public abstract static class QueryExpression implements Expression { public abstract QueryBuilder builder(); + public abstract List getAnalyzedNodes(); + + public abstract void updateAnalyzedNodes(RexNode rexNode); + + public abstract List getUnAnalyzableNodes(); + public boolean isPartial() { return false; } @@ -841,16 +850,32 @@ static QueryExpression create(TerminalExpression expression) { @Getter static class UnAnalyzableQueryExpression extends QueryExpression { - RexNode rexNode; + final RexNode unAnalyzableRexNode; public UnAnalyzableQueryExpression(RexNode rexNode) { - this.rexNode = requireNonNull(rexNode, "rexNode"); + this.unAnalyzableRexNode = requireNonNull(rexNode, "rexNode"); } @Override public QueryBuilder builder() { return null; } + + @Override + public List getUnAnalyzableNodes() { + return List.of(unAnalyzableRexNode); + } + + @Override + public List getAnalyzedNodes() { + return List.of(); + } + + @Override + public void updateAnalyzedNodes(RexNode rexNode) { + throw new IllegalStateException( + "UnAnalyzableQueryExpression does not support unAnalyzableNodes"); + } } /** Builds conjunctions / disjunctions based on existing expressions. */ @@ -858,6 +883,7 @@ public static class CompoundQueryExpression extends QueryExpression { private final boolean partial; private final BoolQueryBuilder builder; + @Getter private List analyzedNodes = new ArrayList<>(); @Getter private final List unAnalyzableNodes = new ArrayList<>(); public static CompoundQueryExpression or(QueryExpression... expressions) { @@ -878,9 +904,9 @@ public static CompoundQueryExpression or(QueryExpression... expressions) { public static CompoundQueryExpression and(boolean partial, QueryExpression... expressions) { CompoundQueryExpression bqe = new CompoundQueryExpression(partial); for (QueryExpression expression : expressions) { - if (expression instanceof UnAnalyzableQueryExpression unAnalyzableQueryExpression) { - bqe.unAnalyzableNodes.add(unAnalyzableQueryExpression.rexNode); - } else { + bqe.analyzedNodes.addAll(expression.getAnalyzedNodes()); + bqe.unAnalyzableNodes.addAll(expression.getUnAnalyzableNodes()); + if (!(expression instanceof UnAnalyzableQueryExpression)) { bqe.builder.must(expression.builder()); } } @@ -906,6 +932,11 @@ public QueryBuilder builder() { return builder; } + @Override + public void updateAnalyzedNodes(RexNode rexNode) { + this.analyzedNodes = List.of(rexNode); + } + @Override public QueryExpression not() { return new CompoundQueryExpression(partial, boolQuery().mustNot(builder())); @@ -915,6 +946,7 @@ public QueryExpression not() { /** Usually basic expression of type {@code a = 'val'} or {@code b > 42}. */ static class SimpleQueryExpression extends QueryExpression { + private RexNode analyzedRexNode; private final NamedFieldExpression rel; private QueryBuilder builder; @@ -949,6 +981,21 @@ public QueryBuilder builder() { return builder; } + @Override + public List getUnAnalyzableNodes() { + return List.of(); + } + + @Override + public List getAnalyzedNodes() { + return List.of(analyzedRexNode); + } + + @Override + public void updateAnalyzedNodes(RexNode rexNode) { + this.analyzedRexNode = rexNode; + } + @Override public QueryExpression not() { builder = boolQuery().mustNot(builder()); diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index 613c17254f1..af2ac9eba7e 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -31,6 +31,7 @@ import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rel.type.RelDataTypeFactory; import org.apache.calcite.rel.type.RelDataTypeField; +import org.apache.calcite.rex.RexBuilder; import org.apache.calcite.rex.RexNode; import org.apache.calcite.sql.fun.SqlStdOperatorTable; import org.apache.commons.lang3.tuple.Pair; @@ -51,7 +52,6 @@ import org.opensearch.sql.opensearch.planner.physical.OpenSearchIndexRules; import org.opensearch.sql.opensearch.request.AggregateAnalyzer; import org.opensearch.sql.opensearch.request.PredicateAnalyzer; -import org.opensearch.sql.opensearch.request.PredicateAnalyzer.CompoundQueryExpression; import org.opensearch.sql.opensearch.request.PredicateAnalyzer.QueryExpression; import org.opensearch.sql.opensearch.response.agg.OpenSearchAggregationResponseParser; import org.opensearch.sql.opensearch.storage.OpenSearchIndex; @@ -114,19 +114,18 @@ public AbstractRelNode pushDownFilter(Filter filter) { newScan.pushDownContext.add( PushDownAction.of( PushDownType.FILTER, - filter.getCondition(), + queryExpression.isPartial() + ? constructCondition( + queryExpression.getAnalyzedNodes(), getCluster().getRexBuilder()) + : filter.getCondition(), requestBuilder -> requestBuilder.pushDownFilter(queryBuilder))); // If the query expression is partial, we need to replace the input of the filter with the // partial pushed scan and the filter condition with non-pushed-down conditions. if (queryExpression.isPartial()) { // Only CompoundQueryExpression could be partial. - List conditions = - ((CompoundQueryExpression) queryExpression).getUnAnalyzableNodes(); - RexNode newCondition = - conditions.size() > 1 - ? getCluster().getRexBuilder().makeCall(SqlStdOperatorTable.AND, conditions) - : conditions.get(0); + List conditions = queryExpression.getUnAnalyzableNodes(); + RexNode newCondition = constructCondition(conditions, getCluster().getRexBuilder()); return filter.copy(filter.getTraitSet(), newScan, newCondition); } return newScan; @@ -140,6 +139,12 @@ public AbstractRelNode pushDownFilter(Filter filter) { return null; } + private static RexNode constructCondition(List conditions, RexBuilder rexBuilder) { + return conditions.size() > 1 + ? rexBuilder.makeCall(SqlStdOperatorTable.AND, conditions) + : conditions.get(0); + } + /** * When pushing down a project, we need to create a new CalciteLogicalIndexScan with the updated * schema since we cannot override getRowType() which is defined to be final. From 86ed34a3e7c18dee90ef0c2ef00c64cc173576d0 Mon Sep 17 00:00:00 2001 From: Heng Qian Date: Mon, 14 Jul 2025 17:53:44 +0800 Subject: [PATCH 7/9] Ignore test when push down disabled Signed-off-by: Heng Qian --- .../org/opensearch/sql/calcite/remote/CalciteExplainIT.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java index 52c502496cc..335deb2f573 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java @@ -8,6 +8,7 @@ import static org.opensearch.sql.util.MatcherUtils.assertJsonEqualsIgnoreId; import java.io.IOException; +import org.junit.Assume; import org.junit.Ignore; import org.junit.Test; import org.opensearch.sql.ppl.ExplainIT; @@ -60,6 +61,7 @@ public void supportSearchSargPushDown_timeRange() throws IOException { // Only for Calcite @Test public void supportPartialPushDown() throws IOException { + Assume.assumeTrue("This test is only for push down enabled", isPushdownEnabled()); // field `address` is text type without keyword subfield, so we cannot push it down. String query = "source=opensearch-sql_test_index_account | where (state = 'Seattle' or age < 10) and (age" @@ -72,6 +74,7 @@ public void supportPartialPushDown() throws IOException { // Only for Calcite @Test public void supportPartialPushDown2() throws IOException { + Assume.assumeTrue("This test is only for push down enabled", isPushdownEnabled()); // field `gender` and `address` are both text type without keyword subfield, so we cannot push // them down. String query = From ac0c48d09aabb2fe8b6c2d7688a33c17412d0432 Mon Sep 17 00:00:00 2001 From: Heng Qian Date: Tue, 15 Jul 2025 11:01:46 +0800 Subject: [PATCH 8/9] Fix IT after merging main Signed-off-by: Heng Qian --- .../remote/CalciteRelevanceFunctionIT.java | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteRelevanceFunctionIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteRelevanceFunctionIT.java index 85ac2554e09..d95fa514056 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteRelevanceFunctionIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteRelevanceFunctionIT.java @@ -5,6 +5,10 @@ package org.opensearch.sql.calcite.remote; +import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BEER; + +import java.io.IOException; +import org.junit.Assume; import org.opensearch.sql.ppl.RelevanceFunctionIT; public class CalciteRelevanceFunctionIT extends RelevanceFunctionIT { @@ -14,4 +18,17 @@ public void init() throws Exception { enableCalcite(); disallowCalciteFallback(); } + + // For Calcite, this PPL won't throw exception since it supports partial pushdown and has + // optimization rule `FilterProjectTransposeRule` to push down the filter through the project. + @Override + public void not_pushdown_throws_exception() throws IOException { + Assume.assumeTrue("This test is only for push down enabled", isPushdownEnabled()); + String query1 = + "SOURCE=" + + TEST_INDEX_BEER + + " | EVAL answerId = AcceptedAnswerId + 1" + + " | WHERE simple_query_string(['Tags'], 'taste') and answerId > 200"; + assertEquals(5, executeQuery(query1).getInt("total")); + } } From 4649db801f9b7e0c689211e232ef73573e4cc92a Mon Sep 17 00:00:00 2001 From: Heng Qian Date: Fri, 18 Jul 2025 10:44:02 +0800 Subject: [PATCH 9/9] Fix IT because mapping changed after merging main Signed-off-by: Heng Qian --- .../opensearch/sql/calcite/remote/CalciteExplainIT.java | 9 ++++----- .../calcite/explain_partial_filter_push2.json | 4 ++-- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java index a96868bf84b..a0c4eb6eab5 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java @@ -84,13 +84,12 @@ public void supportPartialPushDown() throws IOException { // Only for Calcite @Test - public void supportPartialPushDown2() throws IOException { + public void supportPartialPushDown_NoPushIfAllFailed() throws IOException { Assume.assumeTrue("This test is only for push down enabled", isPushdownEnabled()); - // field `gender` and `address` are both text type without keyword subfield, so we cannot push - // them down. + // field `address` is text type without keyword subfield, so we cannot push it down. String query = - "source=opensearch-sql_test_index_account | where (gender = 'M' or age < 10) and (age >= 1" - + " and address = '880 Holmes Lane') | fields age, address"; + "source=opensearch-sql_test_index_account | where (address = '671 Bristol Street' or age <" + + " 10) and (age >= 10 or address = '880 Holmes Lane') | fields age, address"; var result = explainQueryToString(query); String expected = loadFromFile("expectedOutput/calcite/explain_partial_filter_push2.json"); assertJsonEqualsIgnoreId(expected, result); diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push2.json b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push2.json index 5e19f97f0c6..087d2dfbea1 100644 --- a/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push2.json +++ b/integ-test/src/test/resources/expectedOutput/calcite/explain_partial_filter_push2.json @@ -1,6 +1,6 @@ { "calcite": { - "logical": "LogicalProject(age=[$8], address=[$2])\n LogicalFilter(condition=[AND(OR(=($4, 'M'), <($8, 10)), >=($8, 1), =($2, '880 Holmes Lane'))])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n", - "physical": "EnumerableCalc(expr#0..2=[{inputs}], expr#3=['M'], expr#4=[=($t1, $t3)], expr#5=[10], expr#6=[<($t2, $t5)], expr#7=[OR($t4, $t6)], expr#8=['880 Holmes Lane':VARCHAR], expr#9=[=($t0, $t8)], expr#10=[AND($t7, $t9)], age=[$t2], address=[$t0], $condition=[$t10])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[address, gender, age], FILTER->>=($2, 1)], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"range\":{\"age\":{\"from\":1,\"to\":null,\"include_lower\":true,\"include_upper\":true,\"boost\":1.0}}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"_source\":{\"includes\":[\"address\",\"gender\",\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n" + "logical": "LogicalProject(age=[$8], address=[$2])\n LogicalFilter(condition=[AND(OR(=($2, '671 Bristol Street'), <($8, 10)), OR(>=($8, 10), =($2, '880 Holmes Lane')))])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n", + "physical": "EnumerableCalc(expr#0..1=[{inputs}], expr#2=['671 Bristol Street':VARCHAR], expr#3=[=($t0, $t2)], expr#4=[10], expr#5=[<($t1, $t4)], expr#6=[OR($t3, $t5)], expr#7=[>=($t1, $t4)], expr#8=['880 Holmes Lane':VARCHAR], expr#9=[=($t0, $t8)], expr#10=[OR($t7, $t9)], expr#11=[AND($t6, $t10)], age=[$t1], address=[$t0], $condition=[$t11])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[address, age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"address\",\"age\"],\"excludes\":[]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n" } }