github-actions[bot] commented on code in PR #69339: URL: https://github.com/apache/doris/pull/69339#discussion_r4323613497
########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownSearchTopNIntoLanceScan.java: ########## @@ -0,0 +1,318 @@ +// 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.doris.nereids.rules.rewrite; + +import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.TableSnapshot; +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.datasource.lance.LanceExternalTable; +import org.apache.doris.datasource.lance.metadata.LanceRefSelector; +import org.apache.doris.datasource.lance.metadata.LanceTableMetadata; +import org.apache.doris.datasource.lance.source.LancePredicateConverter; +import org.apache.doris.nereids.CascadesContext; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.glue.translator.ExpressionTranslator; +import org.apache.doris.nereids.glue.translator.PlanTranslatorContext; +import org.apache.doris.nereids.rules.Rule; +import org.apache.doris.nereids.rules.RuleType; +import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.GreaterThan; +import org.apache.doris.nereids.trees.expressions.GreaterThanEqual; +import org.apache.doris.nereids.trees.expressions.LessThan; +import org.apache.doris.nereids.trees.expressions.LessThanEqual; +import org.apache.doris.nereids.trees.expressions.Match; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Properties; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.Subtract; +import org.apache.doris.nereids.trees.expressions.functions.scalar.InnerProductApproximate; +import org.apache.doris.nereids.trees.expressions.functions.scalar.L2DistanceApproximate; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Score; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Sqrt; +import org.apache.doris.nereids.trees.expressions.functions.table.LanceSearch; +import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral; +import org.apache.doris.nereids.trees.expressions.literal.FloatLiteral; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalTVFRelation; +import org.apache.doris.nereids.trees.plans.logical.LogicalTopN; +import org.apache.doris.nereids.types.DoubleType; +import org.apache.doris.nereids.types.FloatType; +import org.apache.doris.nereids.util.ExpressionUtils; +import org.apache.doris.tablefunction.FullTextSearchTableValuedFunction; +import org.apache.doris.tablefunction.TableValuedFunctionIf; +import org.apache.doris.tablefunction.VectorSearchTableValuedFunction; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import org.apache.arrow.vector.types.FloatingPointPrecision; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** Reuses the fixed-snapshot Lance search relations for ordinary SQL search expressions. */ +public class PushDownSearchTopNIntoLanceScan extends OneRewriteRuleFactory { + @Override + public Rule build() { + return logicalTopN(logicalProject(any())) + .thenApply(ctx -> rewrite(ctx.root, ctx.cascadesContext)) + .toRule(RuleType.PUSH_DOWN_SEARCH_TOPN_INTO_LANCE_SCAN); + } + + private Plan rewrite(LogicalTopN<LogicalProject<Plan>> topN, CascadesContext context) { + LogicalProject<Plan> project = topN.child(); + Plan input = project.child(); + Set<Expression> predicates = ImmutableSet.of(); + if (input instanceof LogicalFilter) { + predicates = ((LogicalFilter<?>) input).getConjuncts(); + input = input.child(0); + } + if (!(input instanceof LogicalFileScan) || !(((LogicalFileScan) input).getTable() + instanceof LanceExternalTable) || project.isDistinct()) { Review Comment: [P2] Preserve non-movable projections before pruning candidates This rule also matches a project containing `assert_true`, which is explicitly a `NoneMovableFunction`: ```text TopN(d ASC, LIMIT 1) Project(assert_true(id > 0, 'bad') AS checked, l2_distance_approximate(embedding, [0,0]) AS d) LanceScan(id, embedding) ``` Put `(id=-1, embedding=[100,100])` and `(id=1, embedding=[0,0])` in one fragment without an ANN index. The scalar project raises `bad` before TopN. The rewrite requests native `top_k=1`, drops the failing row, and evaluates the retained project only on `id=1`, so the query succeeds. `LogicalProject.pruneOutputs` deliberately preserves these expressions, but this guard only checks DISTINCT. Skip the rewrite for non-movable projections, or preserve their evaluation before native pruning, and cover a failing assertion outside the candidate window. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownSearchTopNIntoLanceScan.java: ########## @@ -0,0 +1,318 @@ +// 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.doris.nereids.rules.rewrite; + +import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.TableSnapshot; +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.datasource.lance.LanceExternalTable; +import org.apache.doris.datasource.lance.metadata.LanceRefSelector; +import org.apache.doris.datasource.lance.metadata.LanceTableMetadata; +import org.apache.doris.datasource.lance.source.LancePredicateConverter; +import org.apache.doris.nereids.CascadesContext; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.glue.translator.ExpressionTranslator; +import org.apache.doris.nereids.glue.translator.PlanTranslatorContext; +import org.apache.doris.nereids.rules.Rule; +import org.apache.doris.nereids.rules.RuleType; +import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.GreaterThan; +import org.apache.doris.nereids.trees.expressions.GreaterThanEqual; +import org.apache.doris.nereids.trees.expressions.LessThan; +import org.apache.doris.nereids.trees.expressions.LessThanEqual; +import org.apache.doris.nereids.trees.expressions.Match; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Properties; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.Subtract; +import org.apache.doris.nereids.trees.expressions.functions.scalar.InnerProductApproximate; +import org.apache.doris.nereids.trees.expressions.functions.scalar.L2DistanceApproximate; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Score; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Sqrt; +import org.apache.doris.nereids.trees.expressions.functions.table.LanceSearch; +import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral; +import org.apache.doris.nereids.trees.expressions.literal.FloatLiteral; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalTVFRelation; +import org.apache.doris.nereids.trees.plans.logical.LogicalTopN; +import org.apache.doris.nereids.types.DoubleType; +import org.apache.doris.nereids.types.FloatType; +import org.apache.doris.nereids.util.ExpressionUtils; +import org.apache.doris.tablefunction.FullTextSearchTableValuedFunction; +import org.apache.doris.tablefunction.TableValuedFunctionIf; +import org.apache.doris.tablefunction.VectorSearchTableValuedFunction; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import org.apache.arrow.vector.types.FloatingPointPrecision; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** Reuses the fixed-snapshot Lance search relations for ordinary SQL search expressions. */ +public class PushDownSearchTopNIntoLanceScan extends OneRewriteRuleFactory { + @Override + public Rule build() { + return logicalTopN(logicalProject(any())) + .thenApply(ctx -> rewrite(ctx.root, ctx.cascadesContext)) + .toRule(RuleType.PUSH_DOWN_SEARCH_TOPN_INTO_LANCE_SCAN); + } + + private Plan rewrite(LogicalTopN<LogicalProject<Plan>> topN, CascadesContext context) { + LogicalProject<Plan> project = topN.child(); + Plan input = project.child(); + Set<Expression> predicates = ImmutableSet.of(); + if (input instanceof LogicalFilter) { + predicates = ((LogicalFilter<?>) input).getConjuncts(); + input = input.child(0); + } + if (!(input instanceof LogicalFileScan) || !(((LogicalFileScan) input).getTable() + instanceof LanceExternalTable) || project.isDistinct()) { + return null; + } + LogicalFileScan scan = (LogicalFileScan) input; + if (scan.getTableSample().isPresent() || scan.getScanParams().isPresent() + || topN.getOrderKeys().size() != 1 || topN.getLimit() <= 0) { + return null; + } + Expression order = topN.getOrderKeys().get(0).getExpr(); + Expression expression = project.getProjects().stream() + .filter(p -> p.toSlot().equals(order) && p instanceof Alias) + .map(p -> ((Alias) p).child()).findFirst().orElse(null); + boolean l2 = expression instanceof L2DistanceApproximate; + boolean dot = expression instanceof InnerProductApproximate; + boolean score = expression instanceof Score; + if (!l2 && !dot && !score) { + return null; + } + // Each split can prune only the nearest (or highest-scoring) candidates. Reversing + // that order or adding tie-breakers would require a different local candidate contract. + if (topN.getOrderKeys().get(0).isAsc() != l2) { + if (score) { + throw new AnalysisException("Lance BM25 Top-N requires ORDER BY score() DESC"); + } + return null; + } + if (topN.getOffset() < 0 || topN.getLimit() > 0xffff_ffffL - topN.getOffset()) { + throw new AnalysisException("Lance search requires limit + offset <= 4294967295"); + } + LanceExternalTable table = (LanceExternalTable) scan.getTable(); + LanceTableMetadata metadata = table.getMetadata(scan.getRelationSnapshot()); + if (!score && metadata.getSchema().getFields().stream().anyMatch( + field -> field.getName().equalsIgnoreCase(VectorSearchTableValuedFunction.DISTANCE_COLUMN))) { + // The TVF reserves _distance, but an ordinary table may legally store that column. + // Keep its existing scalar execution rather than turning a valid SQL query into an error. + return null; + } + Map<String, String> properties = new LinkedHashMap<>(); + properties.put("table", scan.qualified().stream() + .map(part -> "`" + part.replace("`", "``") + "`").collect(Collectors.joining("."))); + // OFFSET belongs to the global Doris TopN, never independently to each split. + properties.put("top_k", Long.toString(topN.getLimit() + topN.getOffset())); + properties.put("version", Long.toString(metadata.getVersion())); + List<Expression> filters = new ArrayList<>(predicates); + List<Expression> scoreFilters = new ArrayList<>(); + if (score) { + List<Match> matches = predicates.stream().filter(Match.class::isInstance) + .map(Match.class::cast).collect(Collectors.toList()); + if (matches.size() != 1) { + throw new AnalysisException("Lance BM25 Top-N requires one top-level MATCH_ANY, MATCH_ALL " + + "or MATCH_PHRASE predicate on a single column"); + } + Match match = matches.get(0); + if (!(match.left() instanceof SlotReference) || !(match.right() instanceof StringLikeLiteral) + || match.getAnalyzer().isPresent()) { + throw new AnalysisException("Lance MATCH requires a string literal and the index's stored analyzer"); + } + properties.put("column", physicalName((SlotReference) match.left())); + properties.put("query", ((StringLikeLiteral) match.right()).getStringValue()); + switch (match.op()) { + case MATCH_ANY: + properties.put("operator", "or"); + break; + case MATCH_ALL: + properties.put("operator", "and"); + break; + case MATCH_PHRASE: + properties.put("query_type", "phrase"); + break; + default: + throw new AnalysisException("Unsupported Lance BM25 MATCH operator: " + match.op()); + } + filters.remove(match); + for (Expression predicate : filters) { + if (predicate.anyMatch(Score.class::isInstance)) { + if (!isScoreLowerBound(predicate)) { + throw new AnalysisException("Lance BM25 supports only score() > literal or score() >= literal"); + } + scoreFilters.add(predicate); + } + } + filters.removeAll(scoreFilters); + } else { + Expression column = expression.child(0); + if (column instanceof Cast && column.getDataType().equals(column.child(0).getDataType())) { + column = column.child(0); + } + Expression query = expression.child(1); + if (!(column instanceof SlotReference) || !(query instanceof ArrayLiteral)) { + return null; + } + Field field = metadata.getSchema().findField(physicalName((SlotReference) column)); + // These SQL functions consume ARRAY<FLOAT>. Searching a wider physical vector + // directly would bypass Doris's element casts and can change distances/ranking. + if (field == null || !(field.getType() instanceof ArrowType.FixedSizeList) Review Comment: [P2] Preserve the scalar error for NULL vector inputs The physical-type guard admits nullable fixed-size-list vectors and nullable elements. For example: ```text TopN(d ASC, LIMIT 1) Project(id, l2_distance_approximate(v, [0,0]) AS d) LanceScan(id, v) // rows: (1,NULL), (2,[1,0]) ``` Both approximate scalar functions call `FunctionArrayDistance::_validate_no_nulls`, so the original plan throws on row 1. The pinned native flat-KNN path propagates vector validity into distances and filters out NULL distances (`vector/flat.rs` and `io/exec/knn.rs`), so the rewritten plan silently returns row 2. NULL elements also lose Doris's explicit validation. Preserve scalar execution unless the required non-null invariants are proven, or validate surviving inputs before native ranking; add outer-NULL and element-NULL differential cases for both functions. ########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/LanceExternalSearchTableValuedFunction.java: ########## @@ -108,6 +108,16 @@ protected LanceExternalSearchTableValuedFunction(PreparedSearch prepared) { this.offset = common.offset; } + protected LanceExternalSearchTableValuedFunction(PreparedSearch prepared, byte[] substraitFilter) { + this(prepared); + // Automatic rewrites must apply the original WHERE before choosing search candidates. + searchRequest.setPrefilter(true); + if (substraitFilter.length > 0) { + searchRequest.setSearchFilter(new TSearchFilter().setFormat(TSearchFilterFormat.SUBSTRAIT) Review Comment: [P2] Gate Substrait search filters on backend support During a smooth upgrade, old source BEs remain query/load available and `FederationBackendPolicy` can assign Lance splits to them. An ordinary query such as `SELECT id, l2_distance_approximate(v,[0,0]) AS d FROM catalog.db.items WHERE id > 0 ORDER BY d LIMIT 1` now unconditionally emits this SUBSTRAIT search filter. The pre-change BE accepts only SQL search filters in `_validate_external_search_request` and returns `NotSupported` for this request, although it could execute the original scalar scan. The existing `LanceScanNode.checkAdditionalTypeBackendCompatibility` does not cover ordinary BIGINT/FLOAT/vector schemas, so it does not prevent the failure. Gate the rewrite/request on backend support, retaining scalar execution for vector queries or selecting only capable BEs, and cover the mixed-version path. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownSearchTopNIntoLanceScan.java: ########## @@ -0,0 +1,318 @@ +// 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.doris.nereids.rules.rewrite; + +import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.TableSnapshot; +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.datasource.lance.LanceExternalTable; +import org.apache.doris.datasource.lance.metadata.LanceRefSelector; +import org.apache.doris.datasource.lance.metadata.LanceTableMetadata; +import org.apache.doris.datasource.lance.source.LancePredicateConverter; +import org.apache.doris.nereids.CascadesContext; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.glue.translator.ExpressionTranslator; +import org.apache.doris.nereids.glue.translator.PlanTranslatorContext; +import org.apache.doris.nereids.rules.Rule; +import org.apache.doris.nereids.rules.RuleType; +import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.GreaterThan; +import org.apache.doris.nereids.trees.expressions.GreaterThanEqual; +import org.apache.doris.nereids.trees.expressions.LessThan; +import org.apache.doris.nereids.trees.expressions.LessThanEqual; +import org.apache.doris.nereids.trees.expressions.Match; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Properties; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.Subtract; +import org.apache.doris.nereids.trees.expressions.functions.scalar.InnerProductApproximate; +import org.apache.doris.nereids.trees.expressions.functions.scalar.L2DistanceApproximate; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Score; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Sqrt; +import org.apache.doris.nereids.trees.expressions.functions.table.LanceSearch; +import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral; +import org.apache.doris.nereids.trees.expressions.literal.FloatLiteral; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalTVFRelation; +import org.apache.doris.nereids.trees.plans.logical.LogicalTopN; +import org.apache.doris.nereids.types.DoubleType; +import org.apache.doris.nereids.types.FloatType; +import org.apache.doris.nereids.util.ExpressionUtils; +import org.apache.doris.tablefunction.FullTextSearchTableValuedFunction; +import org.apache.doris.tablefunction.TableValuedFunctionIf; +import org.apache.doris.tablefunction.VectorSearchTableValuedFunction; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import org.apache.arrow.vector.types.FloatingPointPrecision; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** Reuses the fixed-snapshot Lance search relations for ordinary SQL search expressions. */ +public class PushDownSearchTopNIntoLanceScan extends OneRewriteRuleFactory { + @Override + public Rule build() { + return logicalTopN(logicalProject(any())) + .thenApply(ctx -> rewrite(ctx.root, ctx.cascadesContext)) + .toRule(RuleType.PUSH_DOWN_SEARCH_TOPN_INTO_LANCE_SCAN); + } + + private Plan rewrite(LogicalTopN<LogicalProject<Plan>> topN, CascadesContext context) { + LogicalProject<Plan> project = topN.child(); + Plan input = project.child(); + Set<Expression> predicates = ImmutableSet.of(); + if (input instanceof LogicalFilter) { + predicates = ((LogicalFilter<?>) input).getConjuncts(); + input = input.child(0); + } + if (!(input instanceof LogicalFileScan) || !(((LogicalFileScan) input).getTable() + instanceof LanceExternalTable) || project.isDistinct()) { + return null; + } + LogicalFileScan scan = (LogicalFileScan) input; + if (scan.getTableSample().isPresent() || scan.getScanParams().isPresent() + || topN.getOrderKeys().size() != 1 || topN.getLimit() <= 0) { + return null; + } + Expression order = topN.getOrderKeys().get(0).getExpr(); + Expression expression = project.getProjects().stream() + .filter(p -> p.toSlot().equals(order) && p instanceof Alias) + .map(p -> ((Alias) p).child()).findFirst().orElse(null); + boolean l2 = expression instanceof L2DistanceApproximate; + boolean dot = expression instanceof InnerProductApproximate; + boolean score = expression instanceof Score; + if (!l2 && !dot && !score) { + return null; + } + // Each split can prune only the nearest (or highest-scoring) candidates. Reversing + // that order or adding tie-breakers would require a different local candidate contract. + if (topN.getOrderKeys().get(0).isAsc() != l2) { + if (score) { + throw new AnalysisException("Lance BM25 Top-N requires ORDER BY score() DESC"); + } + return null; + } + if (topN.getOffset() < 0 || topN.getLimit() > 0xffff_ffffL - topN.getOffset()) { + throw new AnalysisException("Lance search requires limit + offset <= 4294967295"); + } + LanceExternalTable table = (LanceExternalTable) scan.getTable(); + LanceTableMetadata metadata = table.getMetadata(scan.getRelationSnapshot()); + if (!score && metadata.getSchema().getFields().stream().anyMatch( + field -> field.getName().equalsIgnoreCase(VectorSearchTableValuedFunction.DISTANCE_COLUMN))) { + // The TVF reserves _distance, but an ordinary table may legally store that column. + // Keep its existing scalar execution rather than turning a valid SQL query into an error. + return null; + } + Map<String, String> properties = new LinkedHashMap<>(); + properties.put("table", scan.qualified().stream() + .map(part -> "`" + part.replace("`", "``") + "`").collect(Collectors.joining("."))); + // OFFSET belongs to the global Doris TopN, never independently to each split. + properties.put("top_k", Long.toString(topN.getLimit() + topN.getOffset())); + properties.put("version", Long.toString(metadata.getVersion())); + List<Expression> filters = new ArrayList<>(predicates); + List<Expression> scoreFilters = new ArrayList<>(); + if (score) { + List<Match> matches = predicates.stream().filter(Match.class::isInstance) + .map(Match.class::cast).collect(Collectors.toList()); + if (matches.size() != 1) { + throw new AnalysisException("Lance BM25 Top-N requires one top-level MATCH_ANY, MATCH_ALL " + + "or MATCH_PHRASE predicate on a single column"); + } + Match match = matches.get(0); + if (!(match.left() instanceof SlotReference) || !(match.right() instanceof StringLikeLiteral) + || match.getAnalyzer().isPresent()) { + throw new AnalysisException("Lance MATCH requires a string literal and the index's stored analyzer"); + } + properties.put("column", physicalName((SlotReference) match.left())); + properties.put("query", ((StringLikeLiteral) match.right()).getStringValue()); + switch (match.op()) { + case MATCH_ANY: + properties.put("operator", "or"); + break; + case MATCH_ALL: + properties.put("operator", "and"); + break; + case MATCH_PHRASE: + properties.put("query_type", "phrase"); + break; + default: + throw new AnalysisException("Unsupported Lance BM25 MATCH operator: " + match.op()); + } + filters.remove(match); + for (Expression predicate : filters) { + if (predicate.anyMatch(Score.class::isInstance)) { + if (!isScoreLowerBound(predicate)) { + throw new AnalysisException("Lance BM25 supports only score() > literal or score() >= literal"); + } + scoreFilters.add(predicate); + } + } + filters.removeAll(scoreFilters); + } else { + Expression column = expression.child(0); + if (column instanceof Cast && column.getDataType().equals(column.child(0).getDataType())) { + column = column.child(0); + } + Expression query = expression.child(1); + if (!(column instanceof SlotReference) || !(query instanceof ArrayLiteral)) { + return null; + } + Field field = metadata.getSchema().findField(physicalName((SlotReference) column)); + // These SQL functions consume ARRAY<FLOAT>. Searching a wider physical vector + // directly would bypass Doris's element casts and can change distances/ranking. + if (field == null || !(field.getType() instanceof ArrowType.FixedSizeList) + || field.getChildren().size() != 1 + || !(field.getChildren().get(0).getType() instanceof ArrowType.FloatingPoint) + || ((ArrowType.FloatingPoint) field.getChildren().get(0).getType()).getPrecision() + != FloatingPointPrecision.SINGLE) { + return null; + } + properties.put("column", field.getName()); + properties.put("query_vector", query.toSql()); + properties.put("metric", l2 ? "l2" : "dot"); + } + Optional<byte[]> filter = convertFilters(filters, scan, metadata, context); + if (!filter.isPresent()) { + if (score) { + throw new AnalysisException("Lance BM25 WHERE conditions must be fully pushable before Top-N"); + } + return null; + } + metadata = withIndexes(table, metadata); + TableValuedFunctionIf function; + String name; + String resultName; + try { + if (score) { + function = FullTextSearchTableValuedFunction.forScan(properties, table, metadata, filter.get()); + name = FullTextSearchTableValuedFunction.NAME; + resultName = FullTextSearchTableValuedFunction.SCORE_COLUMN; + } else { + function = VectorSearchTableValuedFunction.forScan(properties, table, metadata, filter.get()); + name = VectorSearchTableValuedFunction.NAME; + resultName = VectorSearchTableValuedFunction.DISTANCE_COLUMN; + } + } catch (org.apache.doris.common.AnalysisException e) { + throw new AnalysisException(e.getMessage(), e); + } + LogicalTVFRelation relation = new LogicalTVFRelation(scan.getRelationId(), + new LanceSearch(name, new Properties(properties), function), ImmutableList.of()); + Map<String, SlotReference> sourceSlots = new HashMap<>(); + scan.getOutput().forEach(slot -> sourceSlots.put(physicalName((SlotReference) slot), (SlotReference) slot)); + List<Slot> outputs = new ArrayList<>(); + Slot result = null; + for (Slot slot : relation.getOutput()) { + if (slot.getName().equals(resultName)) { + result = slot; + } + SlotReference source = sourceSlots.get(slot.getName()); + outputs.add(source == null ? slot : ((SlotReference) slot) + .withExprId(source.getExprId()).withQualifier(source.getQualifier())); + } + if (result == null) { + throw new AnalysisException("Lance search result column is missing: " + resultName); + } + Expression value = score ? result : l2 + ? new Cast(new Sqrt(new Cast(result, DoubleType.INSTANCE)), FloatType.INSTANCE) + : new Subtract(new FloatLiteral(1.0f), result); + Map<Expression, Expression> replacements = new HashMap<>(); + replacements.put(expression, value); + List<NamedExpression> projects = ExpressionUtils.replaceNamedExpressions(project.getProjects(), replacements); + List<Expression> residuals = scoreFilters.stream().map(e -> ExpressionUtils.replace(e, replacements)) + .collect(Collectors.toList()); + Set<Slot> required = new HashSet<>(); + projects.forEach(p -> required.addAll(p.getInputSlots())); + residuals.forEach(p -> required.addAll(p.getInputSlots())); + // This rewrite runs after column pruning. Do not reintroduce the embedding or wide + // payload columns just because they are part of the underlying search TVF's schema. + relation = relation.withCachedOutputs(outputs.stream().filter(required::contains).collect(Collectors.toList())); + Plan child = residuals.isEmpty() ? relation : new LogicalFilter<>(ImmutableSet.copyOf(residuals), relation); + context.getStatementContext().getSqlCacheContext().ifPresent(cache -> cache.setCannotProcessExpression(true)); + return topN.withChildren(project.withProjectsAndChild(projects, child)); Review Comment: [P2] Preserve rows whose scalar distance is NaN A supported Lance dataset can contain a non-NULL float32 vector `[NaN,0]`. With that as its only row and no ANN index, the input plan is: ```text TopN(d ASC, LIMIT 1) Project(id, l2_distance_approximate(v, [0,0]) AS d) LanceScan(id, v) ``` Ordinary Arrow ingestion copies the float values unchanged; the scalar distance returns NaN and Doris TopN retains its only row. The pinned native flat-KNN implementation explicitly removes NaN distances (`io/exec/knn.rs:1047-1054`), so replacing the scan here returns zero rows. This is a cardinality change even without an approximate index or candidate competition. It also survives the NULL eligibility fix because the vector and its elements are non-NULL. Preserve scalar row eligibility for this case, or restrict the rewrite to a proven compatible input domain, and add a differential regression with stored NaN vectors. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
