github-actions[bot] commented on code in PR #69493: URL: https://github.com/apache/doris/pull/69493#discussion_r4440105683
########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/PaimonVectorSearchTableValuedFunction.java: ########## @@ -0,0 +1,314 @@ +// 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.tablefunction; + +import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.SlotDescriptor; +import org.apache.doris.analysis.SlotRef; +import org.apache.doris.analysis.TableName; +import org.apache.doris.analysis.TableScanParams; +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.catalog.Type; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.datasource.lance.metadata.LanceRefSelector; +import org.apache.doris.datasource.mvcc.MvccSnapshot; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.datasource.paimon.PaimonMvccSnapshot; +import org.apache.doris.datasource.paimon.PaimonScanParams; +import org.apache.doris.datasource.paimon.PaimonSnapshot; +import org.apache.doris.datasource.paimon.PaimonUtil; +import org.apache.doris.datasource.paimon.source.PaimonScanNode; +import org.apache.doris.datasource.paimon.source.PaimonVectorSearch; +import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.glue.translator.ExpressionTranslator; +import org.apache.doris.nereids.glue.translator.PlanTranslatorContext; +import org.apache.doris.nereids.rules.rewrite.PushDownVectorTopNIntoPaimonScan; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.planner.PlanNodeId; +import org.apache.doris.planner.ScanNode; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.SessionVariable; +import org.apache.doris.thrift.TExternalSearchQuery; +import org.apache.doris.thrift.TExternalSearchRequest; +import org.apache.doris.thrift.TPaimonVectorSearchOptions; +import org.apache.doris.thrift.TSearchVector; +import org.apache.doris.thrift.TVectorMetric; +import org.apache.doris.thrift.TVectorSearchParams; + +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 org.apache.arrow.vector.types.pojo.FieldType; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeRoot; +import org.apache.paimon.types.VectorType; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** Paimon PK-vector implementation of the shared vector_search SQL interface. */ +public class PaimonVectorSearchTableValuedFunction extends ExternalSearchTableValuedFunction { + private static final String NAME = VectorSearchTableValuedFunction.NAME; + private static final String DISTANCE_COLUMN = VectorSearchTableValuedFunction.DISTANCE_COLUMN; + // The Rust search eagerly materializes each split before Doris can account for its batches. + private static final int MAX_CANDIDATES_PER_SPLIT = 100_000; + private static final Set<String> PROPERTIES = ImmutableSet.of( + "table", "column", "query_vector", "top_k", "offset", "metric", "filter", "prefilter", + "nprobes", "refine_factor", "use_index", "version", "timestamp", "tag", "branch"); + + private final TableName tableName; + private final PaimonExternalTable sourceTable; + private final FileStoreTable table; + private final List<Column> columns; + private final TExternalSearchRequest request; + private final long topK; + private final long offset; + private final boolean emptySnapshot; + private final Optional<String> filter; + private final boolean prefilter; + private List<Expression> boundPrefilter = Collections.emptyList(); + + public PaimonVectorSearchTableValuedFunction(Map<String, String> properties, boolean deferQueryVector) + throws AnalysisException { + Map<String, String> params = normalizeProperties(properties, PROPERTIES, NAME); + tableName = LanceExternalSearchTableValuedFunction.parseTableName(required(params, "table", NAME)); + sourceTable = findPaimonTable(tableName); + LanceRefSelector selector = LanceExternalSearchTableValuedFunction.parseSelector(params); + if (selector.getBranch().isPresent() && selector.getSnapshot().isPresent()) { + throw new AnalysisException("Paimon does not support combining branch with version or timestamp"); + } + TableScanParams scanParams = selector.getBranch().map(name -> new TableScanParams( + "branch", Collections.emptyMap(), Collections.singletonList(name))).orElse(null); + if (selector.getTag().isPresent()) { + scanParams = new TableScanParams("tag", Collections.emptyMap(), + Collections.singletonList(selector.getTag().get())); + } + // A TVF does not pass through ordinary relation binding. Load its source snapshot here + // so selectors work and latest reads share the statement's existing snapshot fence. + ConnectContext context = ConnectContext.get(); + Optional<MvccSnapshot> snapshot = context != null && context.getStatementContext() != null + ? context.getStatementContext().loadSnapshots(sourceTable, + selector.getSnapshot(), Optional.ofNullable(scanParams)) + : Optional.of(sourceTable.loadSnapshot(selector.getSnapshot(), Optional.ofNullable(scanParams))); + Table selected = sourceTable.getPaimonTable(snapshot); + if (!(selected instanceof FileStoreTable)) { + throw new AnalysisException("Paimon vector_search requires a file-store table"); + } + FileStoreTable selectedTable = (FileStoreTable) selected; + long snapshotId = ((PaimonMvccSnapshot) snapshot.get()).getSnapshotCacheValue() + .getSnapshot().getSnapshotId(); + emptySnapshot = snapshotId == PaimonSnapshot.INVALID_SNAPSHOT_ID; + // Preserve the bound schema, including columns added after the last data commit. + // Retained tags can also outlive the snapshot file and must keep their read handle. + table = emptySnapshot || selector.getTag().isPresent() ? selectedTable + : selectedTable.copyWithoutTimeTravel(PaimonScanParams.isolateSnapshotRead(snapshotId)); + topK = parseLong(params.getOrDefault("top_k", "10"), "top_k", 1, MAX_CANDIDATES_PER_SPLIT); + offset = parseLong(params.getOrDefault("offset", "0"), "offset", 0, MAX_CANDIDATES_PER_SPLIT - topK); + filter = Optional.ofNullable(params.get("filter")); + if (filter.isPresent()) { + validateAndEncodeSqlFilter(filter.get()); + } + prefilter = params.containsKey("prefilter") && parseBoolean(params.get("prefilter"), "prefilter"); + + String column = required(params, "column", NAME); + DataType vectorType = null; + for (org.apache.paimon.types.DataField field : table.rowType().getFields()) { + if (field.name().equalsIgnoreCase(column)) { + column = field.name(); + vectorType = field.type(); + break; + } + } + if (!(vectorType instanceof VectorType) + || ((VectorType) vectorType).getElementType().getTypeRoot() != DataTypeRoot.FLOAT) { + throw new AnalysisException("Paimon vector_search requires a VECTOR<FLOAT> column"); + } + CoreOptions options = CoreOptions.fromMap(table.options()); + if (!options.primaryKeyVectorIndexEnabled() + || !options.primaryKeyVectorIndexColumns().contains(column)) { + throw new AnalysisException("Paimon vector_search requires a PK-vector index on '" + column + "'"); + } + String metric = params.getOrDefault("metric", "l2").toLowerCase(Locale.ROOT); + TVectorMetric vectorMetric; + if (metric.equals("l2")) { + vectorMetric = TVectorMetric.L2; + } else if (metric.equals("dot") || metric.equals("dot_product") || metric.equals("inner_product")) { + vectorMetric = TVectorMetric.DOT_PRODUCT; + } else { + throw new AnalysisException("Paimon vector_search supports only l2 and dot metrics"); + } + String indexMetric = options.primaryKeyVectorDistanceMetric(column); + if (!(vectorMetric == TVectorMetric.L2 ? "l2" : "inner_product").equals(indexMetric)) { + throw new AnalysisException("Paimon vector_search metric does not match the PK-vector index"); + } + TSearchVector query = null; + if (!deferQueryVector) { + Field field = new Field(column, FieldType.nullable( + new ArrowType.FixedSizeList(((VectorType) vectorType).getLength())), + Collections.singletonList(Field.nullable("item", + new ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE)))); + query = VectorSearchTableValuedFunction.parseAndEncodeQueryVector( + field, required(params, "query_vector", NAME)); + ByteBuffer values = ByteBuffer.wrap(query.getValues()).order(ByteOrder.LITTLE_ENDIAN); + while (values.hasRemaining()) { + if (!Float.isFinite(values.getFloat())) { + throw new AnalysisException("Paimon query_vector coordinates must be finite FLOAT values"); + } + } + } + Map<String, String> searchOptions = new HashMap<>(); + if (params.containsKey("nprobes")) { + searchOptions.put("ivf.nprobe", String.valueOf(parseLong( + params.get("nprobes"), "nprobes", 1, Integer.MAX_VALUE))); + } + if (params.containsKey("refine_factor")) { + // Refinement oversamples the bounded candidate set before returning the final Top-K. + searchOptions.put("refine_factor", String.valueOf(parseLong( + params.get("refine_factor"), "refine_factor", 1, MAX_CANDIDATES_PER_SPLIT / (topK + offset)))); + } + if (params.containsKey("use_index") && !parseBoolean(params.get("use_index"), "use_index")) { + searchOptions.put("vector-index.search-mode", "full"); + } + request = new TExternalSearchRequest().setSchemaVersion(1) + .setSearchQuery(TExternalSearchQuery.vector_search(new TVectorSearchParams() + .setColumn(column).setQueryVector(query).setMetric(vectorMetric) + .setTopK(topK).setOffset(offset))) + .setPaimonOptions(new TPaimonVectorSearchOptions().setOptions(searchOptions).setTvfDistance(true)); + List<Column> schema = new ArrayList<>(PaimonUtil.parseSchema(table, + sourceTable.getCatalog().getEnableMappingVarbinary(), + sourceTable.getCatalog().getEnableMappingTimestampTz())); + for (Column field : schema) { + if (field.getName().equalsIgnoreCase(DISTANCE_COLUMN) + || field.getName().equalsIgnoreCase(PaimonVectorSearch.SEARCH_DISTANCE_COLUMN) + || field.getName().equalsIgnoreCase("__paimon_search_score") + || field.getName().equalsIgnoreCase("_PKEY_VECTOR_POSITION")) { + throw new AnalysisException("Paimon vector_search column name is reserved: " + field.getName()); + } + } + schema.add(new Column(DISTANCE_COLUMN, Type.FLOAT, false)); + columns = Collections.unmodifiableList(schema); + } + + static PaimonExternalTable findPaimonTable(TableName name) throws AnalysisException { + if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ConnectContext.get(), name, PrivPredicate.SELECT)) { + throw new AnalysisException("SELECT privilege is required on " + name); + } + CatalogIf<?> catalog = Env.getCurrentEnv().getCatalogMgr().getCatalog(name.getCtl()); + TableIf table = catalog.getDbOrAnalysisException(name.getDb()).getTableOrAnalysisException(name.getTbl()); + if (!(table instanceof PaimonExternalTable)) { + throw new AnalysisException("Paimon vector_search requires a Paimon data table"); + } + return (PaimonExternalTable) table; + } + + public Optional<String> getFilter() { + return filter; + } + + public void bindFilter(LogicalFilter<?> filterPlan) { + if (!prefilter) { + return; + } + if (filterPlan.getInputSlots().isEmpty()) { + return; + } + if (!PushDownVectorTopNIntoPaimonScan.conjunctsRideIntoVectorSearch(filterPlan)) { + throw new org.apache.doris.nereids.exceptions.AnalysisException( + "Paimon prefilter requires supported scalar column comparisons, IN, or null checks"); + } + boolean dataFilter = filterPlan.getInputSlots().stream() + .anyMatch(slot -> !table.partitionKeys().contains(slot.getName())); + if (dataFilter && !PaimonScanNode.vectorSearchSupportsDataPredicate(table)) { + throw new org.apache.doris.nereids.exceptions.AnalysisException( + "Paimon data prefilter requires deletion vectors without merge-on-read"); + } + boundPrefilter = Collections.unmodifiableList(new ArrayList<>(filterPlan.getConjuncts())); + } + + @Override + public long getTopK() { + return topK; + } + + @Override + public long getOffset() { + return offset; + } + + public TExternalSearchRequest getSearchRequest() { + return request.deepCopy(); + } + + @Override + public String getTableName() { + return "PaimonVectorSearch<" + tableName + ">"; + } + + @Override + public List<Column> getTableColumns() { + return columns; + } + + @Override + public ScanNode getScanNode(PlanNodeId id, TupleDescriptor desc, SessionVariable sv) { + TExternalSearchRequest scanRequest = request.deepCopy(); + PlanTranslatorContext translation = new PlanTranslatorContext(); + List<Expr> predicates = new ArrayList<>(); + List<Expr> partitionAndDataPredicates = new ArrayList<>(); + for (Expression expression : boundPrefilter) { + for (Slot slot : expression.getInputSlots()) { + SlotDescriptor descriptor = desc.getSlots().stream() + .filter(candidate -> candidate.getColumn().getName().equalsIgnoreCase(slot.getName())) + .findFirst().orElseThrow(() -> new IllegalStateException( + "Paimon prefilter column was pruned: " + slot.getName())); + translation.addExprIdSlotRefPair(slot.getExprId(), new SlotRef(descriptor)); + } + Expr translated = ExpressionTranslator.translate(expression, translation); Review Comment: [P2] Preserve prefilter input slots through column pruning On a DV-backed indexed table with non-partition `id BIGINT NOT NULL`, this supported query fails when it projects only `_distance`: ```text Project(_distance) TopN(_distance ASC, 1) Filter(id IS NOT NULL) vector_search(..., filter='id IS NOT NULL', prefilter='true') ``` `EliminateNotNull` removes the tautological filter, then `ColumnPruning` inserts a projection containing only `_distance`. The private `boundPrefilter` still carries `id`: this loop serializes its slot before the lower project's translator calls `updateScanSlotsMaterialization` and removes `id` from the scan tuple. `DescriptorTable.toThrift` emits only the remaining tuple slots, so the saved request references an absent descriptor; the Rust reader fails while preparing that prefilter's `VSlotRef`. Keeping `id` in SELECT avoids the failure. Preserve prefilter inputs as scan dependencies, or reconcile the request with the final optimized predicates. Correct comparison normalization alone does not preserve these later-pruned dependencies. ########## gensrc/thrift/PlanNodes.thrift: ########## @@ -385,6 +406,13 @@ struct TPaimonFileDesc { // paimon-rust: non-default branch name. Unset means main branch (matches // upstream paimon commit 742da63: null-if-DEFAULT_MAIN_BRANCH). 19: optional string paimon_branch + // Primary-key vector (ANN) search payload for this split. When set, BE + // routes to the paimon-rust vector-search read path instead of the normal + // split read; paimon_split is set but empty in that case. + // Field 21: id 20 is reserved as `original_file_path` on master and must not + // be reused here — a mixed-version cluster skips mismatched wire types + // instead of receiving either field. + 21: optional TPaimonVectorPayload vector_payload; Review Comment: [P2] Reserve the Paimon field IDs already used on branch-4.1 `TPaimonFileDesc.21` is already `optional list<TPaimonBlobFileDesc> blob_files` on active branch-4.1 (verified at `dd72933ec78a8a4daf3e1d204a6ea96d6443ada7`); fields 22 and 23 hold its anchor row coordinates. This assigns the same ID an unrelated STRUCT, while the comment reserves only master's field 20. Thrift skips a LIST/STRUCT mismatch, so these definitions cannot transport both meanings compatibly or coexist when the changes are picked across releases. A vector range deliberately has an empty `paimon_split`, making a lost vector payload unusable. Reserve 21-23 and choose an ID unused on every active branch, keeping its meaning identical in picks. Also make the new payload's members optional with explicit reader validation, as required by the shared-definition guide; the optional outer envelope does not make the conflicting ID safe. ########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/PaimonVectorSearchTableValuedFunction.java: ########## @@ -0,0 +1,314 @@ +// 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.tablefunction; + +import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.SlotDescriptor; +import org.apache.doris.analysis.SlotRef; +import org.apache.doris.analysis.TableName; +import org.apache.doris.analysis.TableScanParams; +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.catalog.Type; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.datasource.lance.metadata.LanceRefSelector; +import org.apache.doris.datasource.mvcc.MvccSnapshot; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.datasource.paimon.PaimonMvccSnapshot; +import org.apache.doris.datasource.paimon.PaimonScanParams; +import org.apache.doris.datasource.paimon.PaimonSnapshot; +import org.apache.doris.datasource.paimon.PaimonUtil; +import org.apache.doris.datasource.paimon.source.PaimonScanNode; +import org.apache.doris.datasource.paimon.source.PaimonVectorSearch; +import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.glue.translator.ExpressionTranslator; +import org.apache.doris.nereids.glue.translator.PlanTranslatorContext; +import org.apache.doris.nereids.rules.rewrite.PushDownVectorTopNIntoPaimonScan; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.planner.PlanNodeId; +import org.apache.doris.planner.ScanNode; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.SessionVariable; +import org.apache.doris.thrift.TExternalSearchQuery; +import org.apache.doris.thrift.TExternalSearchRequest; +import org.apache.doris.thrift.TPaimonVectorSearchOptions; +import org.apache.doris.thrift.TSearchVector; +import org.apache.doris.thrift.TVectorMetric; +import org.apache.doris.thrift.TVectorSearchParams; + +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 org.apache.arrow.vector.types.pojo.FieldType; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeRoot; +import org.apache.paimon.types.VectorType; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** Paimon PK-vector implementation of the shared vector_search SQL interface. */ +public class PaimonVectorSearchTableValuedFunction extends ExternalSearchTableValuedFunction { + private static final String NAME = VectorSearchTableValuedFunction.NAME; + private static final String DISTANCE_COLUMN = VectorSearchTableValuedFunction.DISTANCE_COLUMN; + // The Rust search eagerly materializes each split before Doris can account for its batches. + private static final int MAX_CANDIDATES_PER_SPLIT = 100_000; + private static final Set<String> PROPERTIES = ImmutableSet.of( + "table", "column", "query_vector", "top_k", "offset", "metric", "filter", "prefilter", + "nprobes", "refine_factor", "use_index", "version", "timestamp", "tag", "branch"); + + private final TableName tableName; + private final PaimonExternalTable sourceTable; + private final FileStoreTable table; + private final List<Column> columns; + private final TExternalSearchRequest request; + private final long topK; + private final long offset; + private final boolean emptySnapshot; + private final Optional<String> filter; + private final boolean prefilter; + private List<Expression> boundPrefilter = Collections.emptyList(); + + public PaimonVectorSearchTableValuedFunction(Map<String, String> properties, boolean deferQueryVector) + throws AnalysisException { + Map<String, String> params = normalizeProperties(properties, PROPERTIES, NAME); + tableName = LanceExternalSearchTableValuedFunction.parseTableName(required(params, "table", NAME)); + sourceTable = findPaimonTable(tableName); + LanceRefSelector selector = LanceExternalSearchTableValuedFunction.parseSelector(params); + if (selector.getBranch().isPresent() && selector.getSnapshot().isPresent()) { + throw new AnalysisException("Paimon does not support combining branch with version or timestamp"); + } + TableScanParams scanParams = selector.getBranch().map(name -> new TableScanParams( + "branch", Collections.emptyMap(), Collections.singletonList(name))).orElse(null); + if (selector.getTag().isPresent()) { + scanParams = new TableScanParams("tag", Collections.emptyMap(), + Collections.singletonList(selector.getTag().get())); + } + // A TVF does not pass through ordinary relation binding. Load its source snapshot here + // so selectors work and latest reads share the statement's existing snapshot fence. + ConnectContext context = ConnectContext.get(); + Optional<MvccSnapshot> snapshot = context != null && context.getStatementContext() != null + ? context.getStatementContext().loadSnapshots(sourceTable, + selector.getSnapshot(), Optional.ofNullable(scanParams)) + : Optional.of(sourceTable.loadSnapshot(selector.getSnapshot(), Optional.ofNullable(scanParams))); + Table selected = sourceTable.getPaimonTable(snapshot); + if (!(selected instanceof FileStoreTable)) { + throw new AnalysisException("Paimon vector_search requires a file-store table"); + } + FileStoreTable selectedTable = (FileStoreTable) selected; + long snapshotId = ((PaimonMvccSnapshot) snapshot.get()).getSnapshotCacheValue() + .getSnapshot().getSnapshotId(); + emptySnapshot = snapshotId == PaimonSnapshot.INVALID_SNAPSHOT_ID; + // Preserve the bound schema, including columns added after the last data commit. + // Retained tags can also outlive the snapshot file and must keep their read handle. + table = emptySnapshot || selector.getTag().isPresent() ? selectedTable + : selectedTable.copyWithoutTimeTravel(PaimonScanParams.isolateSnapshotRead(snapshotId)); + topK = parseLong(params.getOrDefault("top_k", "10"), "top_k", 1, MAX_CANDIDATES_PER_SPLIT); + offset = parseLong(params.getOrDefault("offset", "0"), "offset", 0, MAX_CANDIDATES_PER_SPLIT - topK); + filter = Optional.ofNullable(params.get("filter")); + if (filter.isPresent()) { + validateAndEncodeSqlFilter(filter.get()); + } + prefilter = params.containsKey("prefilter") && parseBoolean(params.get("prefilter"), "prefilter"); + + String column = required(params, "column", NAME); + DataType vectorType = null; + for (org.apache.paimon.types.DataField field : table.rowType().getFields()) { + if (field.name().equalsIgnoreCase(column)) { + column = field.name(); + vectorType = field.type(); + break; + } + } + if (!(vectorType instanceof VectorType) + || ((VectorType) vectorType).getElementType().getTypeRoot() != DataTypeRoot.FLOAT) { + throw new AnalysisException("Paimon vector_search requires a VECTOR<FLOAT> column"); + } + CoreOptions options = CoreOptions.fromMap(table.options()); + if (!options.primaryKeyVectorIndexEnabled() + || !options.primaryKeyVectorIndexColumns().contains(column)) { + throw new AnalysisException("Paimon vector_search requires a PK-vector index on '" + column + "'"); + } + String metric = params.getOrDefault("metric", "l2").toLowerCase(Locale.ROOT); + TVectorMetric vectorMetric; + if (metric.equals("l2")) { + vectorMetric = TVectorMetric.L2; + } else if (metric.equals("dot") || metric.equals("dot_product") || metric.equals("inner_product")) { + vectorMetric = TVectorMetric.DOT_PRODUCT; + } else { + throw new AnalysisException("Paimon vector_search supports only l2 and dot metrics"); + } + String indexMetric = options.primaryKeyVectorDistanceMetric(column); + if (!(vectorMetric == TVectorMetric.L2 ? "l2" : "inner_product").equals(indexMetric)) { + throw new AnalysisException("Paimon vector_search metric does not match the PK-vector index"); + } + TSearchVector query = null; + if (!deferQueryVector) { + Field field = new Field(column, FieldType.nullable( + new ArrowType.FixedSizeList(((VectorType) vectorType).getLength())), + Collections.singletonList(Field.nullable("item", + new ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE)))); + query = VectorSearchTableValuedFunction.parseAndEncodeQueryVector( + field, required(params, "query_vector", NAME)); + ByteBuffer values = ByteBuffer.wrap(query.getValues()).order(ByteOrder.LITTLE_ENDIAN); + while (values.hasRemaining()) { + if (!Float.isFinite(values.getFloat())) { + throw new AnalysisException("Paimon query_vector coordinates must be finite FLOAT values"); + } + } + } + Map<String, String> searchOptions = new HashMap<>(); + if (params.containsKey("nprobes")) { + searchOptions.put("ivf.nprobe", String.valueOf(parseLong( + params.get("nprobes"), "nprobes", 1, Integer.MAX_VALUE))); + } + if (params.containsKey("refine_factor")) { + // Refinement oversamples the bounded candidate set before returning the final Top-K. + searchOptions.put("refine_factor", String.valueOf(parseLong( + params.get("refine_factor"), "refine_factor", 1, MAX_CANDIDATES_PER_SPLIT / (topK + offset)))); + } + if (params.containsKey("use_index") && !parseBoolean(params.get("use_index"), "use_index")) { + searchOptions.put("vector-index.search-mode", "full"); + } + request = new TExternalSearchRequest().setSchemaVersion(1) + .setSearchQuery(TExternalSearchQuery.vector_search(new TVectorSearchParams() + .setColumn(column).setQueryVector(query).setMetric(vectorMetric) + .setTopK(topK).setOffset(offset))) + .setPaimonOptions(new TPaimonVectorSearchOptions().setOptions(searchOptions).setTvfDistance(true)); + List<Column> schema = new ArrayList<>(PaimonUtil.parseSchema(table, + sourceTable.getCatalog().getEnableMappingVarbinary(), + sourceTable.getCatalog().getEnableMappingTimestampTz())); + for (Column field : schema) { + if (field.getName().equalsIgnoreCase(DISTANCE_COLUMN) + || field.getName().equalsIgnoreCase(PaimonVectorSearch.SEARCH_DISTANCE_COLUMN) + || field.getName().equalsIgnoreCase("__paimon_search_score") + || field.getName().equalsIgnoreCase("_PKEY_VECTOR_POSITION")) { + throw new AnalysisException("Paimon vector_search column name is reserved: " + field.getName()); + } + } + schema.add(new Column(DISTANCE_COLUMN, Type.FLOAT, false)); + columns = Collections.unmodifiableList(schema); + } + + static PaimonExternalTable findPaimonTable(TableName name) throws AnalysisException { + if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ConnectContext.get(), name, PrivPredicate.SELECT)) { + throw new AnalysisException("SELECT privilege is required on " + name); + } + CatalogIf<?> catalog = Env.getCurrentEnv().getCatalogMgr().getCatalog(name.getCtl()); + TableIf table = catalog.getDbOrAnalysisException(name.getDb()).getTableOrAnalysisException(name.getTbl()); + if (!(table instanceof PaimonExternalTable)) { + throw new AnalysisException("Paimon vector_search requires a Paimon data table"); + } + return (PaimonExternalTable) table; + } + + public Optional<String> getFilter() { + return filter; + } + + public void bindFilter(LogicalFilter<?> filterPlan) { + if (!prefilter) { + return; + } + if (filterPlan.getInputSlots().isEmpty()) { + return; + } + if (!PushDownVectorTopNIntoPaimonScan.conjunctsRideIntoVectorSearch(filterPlan)) { + throw new org.apache.doris.nereids.exceptions.AnalysisException( + "Paimon prefilter requires supported scalar column comparisons, IN, or null checks"); + } + boolean dataFilter = filterPlan.getInputSlots().stream() + .anyMatch(slot -> !table.partitionKeys().contains(slot.getName())); + if (dataFilter && !PaimonScanNode.vectorSearchSupportsDataPredicate(table)) { + throw new org.apache.doris.nereids.exceptions.AnalysisException( + "Paimon data prefilter requires deletion vectors without merge-on-read"); + } + boundPrefilter = Collections.unmodifiableList(new ArrayList<>(filterPlan.getConjuncts())); Review Comment: [P1] Normalize explicit prefilters before retaining them For an indexed table with a non-partition `id BIGINT` column, deletion vectors enabled and merge-on-read disabled, `filter='id <=> 7', prefilter='true'` is accepted here. The reduced plan is: ```text TopN(_distance ASC, 1) Filter(id <=> 7) vector_search(...) ``` `BindExpression` captures this analyzed expression before `NullSafeEqualToEqual` runs. Later optimization changes the plan filter, but `boundPrefilter` keeps the original `NullSafeEqual`; `getScanNode` translates it directly into `EQ_FOR_NULL`. `PaimonRustPredicateConverter::_convert_binary` converts a non-NULL literal `EQ_FOR_NULL` into `paimon_predicate_is_null`, so the search excludes an existing `id=7` row and returns no match. The converted-conjunct count succeeds and cannot detect the wrong semantics. The accepted literal-left form `filter='7 < id'` similarly bypasses normalization and fails conversion. Normalize the retained predicate using the required planner rules, or make the admitted raw forms correctly supported by the converter before capture. -- 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]
