pzhdfy commented on code in PR #68988: URL: https://github.com/apache/doris/pull/68988#discussion_r4321192181
########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonRustEligibility.java: ########## @@ -0,0 +1,298 @@ +// 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.datasource.paimon.source; + +import org.apache.doris.datasource.paimon.PaimonScanParams; +import org.apache.doris.system.Backend; + +import org.apache.commons.lang3.StringUtils; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.rest.RESTTokenFileIO; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; + +import java.util.Collection; +import java.util.Locale; +import java.util.Map; + +/** + * The table-, storage- and backend-level gates of the paimon-rust reader, + * computed once from the table handle and shared by every call site that must + * agree on them: {@code PaimonScanNode}'s per-split gate (both the batch + * reader and the PK-vector search) and the {@code PushDownVectorTopNIntoPaimonScan} + * rewrite's pre-flight via {@code PaimonVectorSearch.vectorScanEligible}. The + * rewrite is one-way (a rewritten PK-vector split has no JNI fallback), so + * both sites must evaluate the same conditions; this class is that single + * definition. + * + * <p>Covers only what is a property of the table handle, the storage shape or + * the cluster's backends. Everything that depends on the individual split + * (native DataSplit shape, unmaterialized deletion vectors, external-path + * files, the PaimonRustReaderCapabilities whitelist) or on the final scan + * tuple (ORC TIMESTAMP_LTZ, projected VARIANT) stays with the scan node. + */ +final class PaimonRustEligibility { + + // BE opens the table via paimon_table_from_schema_json, which needs the resolved + // TableSchema that only FileStoreTable exposes via schema(). A table that is not a + // FileStoreTable (e.g. a sys table backed by DataSplit) cannot ship a schema JSON, + // so it falls back to JNI rather than sending an incomplete PAIMON_RUST request + // that BE would reject. + final boolean isFileStoreTable; + + // Both sides of a FallbackReadFileStoreTable wrap their splits, and the pinned + // rust decoder rejects a FallbackDataSplit outright ("trailing bytes after + // DataSplit" — it requires full-buffer consumption); the wrapper is gated as a + // whole (any split from it routes to JNI) until the rust ABI represents both + // sides. The per-split FallbackSplit shape is judged separately in PaimonScanNode. + final boolean fallbackReadTable; + + // A renewable REST token reached the shipped properties as a plain value; only + // the table's FileIO type reveals it expires: doInitialize snapshots + // RESTTokenFileIO.validToken().token() into the backend storage properties, + // discarding expireAtMillis and the REST refresh context, so the shipped + // credentials look static — but the pinned rust table reuses one option map with + // no refresh callback, while the JNI RESTTokenFileIO checks expiry before each + // file operation and obtains a replacement token. A queued or long scan that + // crosses the token TTL would start on rust and later fail authentication. + // Null-safe: a table handle whose FileIO is not resolved stays rust-eligible, + // mirroring the CoreOptions null-safety below. + final boolean restTokenTable; + + // query-auth.enabled tables stay on JNI: when catalog authorization succeeds with + // no row filter or column mask, Paimon still leaves an ordinary DataSplit + // (restricted results use QueryAuthSplit and are already handled by the + // nativeSplit gate), so this table shape passes the compound gate — but the + // shipped schema keeps query-auth.enabled=true and the pinned rust ReadBuilder + // rejects every such table (its CoreOptions::ensure_read_authorized fails closed + // because the client cannot enforce the row filter / column masking), turning a + // valid authorized scan into a BE-open failure. Until the authorization result + // can be transported and enforced by the rust ABI, these tables route to JNI. + final boolean queryAuthTable; + + // Partial-update / aggregation tables with deletion vectors: the pinned rust + // read_pk rejects merge-engine=partial-update/aggregation with + // deletion-vectors.merge-on-read=true outright, so that table option routes the + // whole table to JNI. merge-on-read has no Java accessor in the pinned paimon, so + // it is read raw from the schema options map — the one the BE rust reader + // deserializes from the shipped schema JSON, so it cannot diverge from what BE + // sees — with the rust parsing semantics (any case-insensitive "true" is on, + // default false). For the remaining PU/AGG DV tables, per-split materialization + // is judged in PaimonScanNode (splitDvNotMaterialized); Deduplicate stays + // rust-eligible (its read_pk routes uncompacted splits to the KV reader, which + // applies the attached per-file DVs). + final boolean puAggDeletionVectors; + final boolean dvMergeOnRead; + + // deduplicate.ignore-delete=true tables stay on JNI: Java's + // DeduplicateMergeFunction skips retract records when the option is set — + // including old, uncompacted files that still contain them — but the pinned rust + // read_pk does not pass table options into its deduplicate merge: it picks the + // latest row and omits the key when that row is DELETE/UPDATE_BEFORE. An + // uncompacted insert followed by a delete therefore returns the insert through + // JNI but silently disappears through rust. Gate the option until the rust + // merge implements it. + final boolean deduplicateIgnoreDelete; + + // Non-DV merge options the pinned rust read rejects: Java supports + // partial-update.remove-record-on-delete / aggregation.remove-record-on-delete + // and the wider per-field retract matrix, but the rust PartialUpdateConfig / + // AggregationConfig validations return Unsupported for them — and the DV-derived + // gates above only cover deletion-vector tables, so an ordinary non-DV DataSplit + // with one of these options would pass the compound gate and fail during the + // rust merge construction. Mirrors the exact rust key matrix (presence, not + // values) against the same schema options map BE deserializes. + final boolean rustUnsupportedMergeOption; + + // Branch tables can retain persisted modes without a selector: the shipped + // schema options carry scan.mode, and the pinned Rust builders only accept the + // default. + final boolean scanModeNotDefault; + + // The credential provider shape is evaluated by providerModeTranslatable below. + final boolean providerModeTranslatable; + + // Scheme capability gate: the pinned paimon-rust storage dispatcher (io/storage.rs) + // selects the FileIO parser from the table location's URI scheme, and + // libpaimon_c.a compiles in separate COS, OBS, GCS and Azdls parsers besides the + // OSS and S3 ones. Doris normalizes every object store's credentials into the + // AWS_* / use_path_style aliases, which the BE rust bridge translates only into + // the fs.oss.* and s3.* key families — a cosn:// / obs:// / gs:// / abfs:// + // warehouse would reach its scheme's parser without the key family it reads and + // fail the open instead of using JNI. Only the schemes whose property translation + // is implemented and open-tested (s3 / s3a / oss) plus the credential-free hdfs + // and local-filesystem parsers stay rust-eligible. A null location also routes to + // JNI: the rust path needs the paimon_table that only a real location can provide. + final boolean schemeCapabilityVerified; + + // An hdfs:// location is scheme-verified only together with the credential-free + // backend shape: the backend storage properties that ship to BE also carry an + // HDFS catalog's authentication (kerberos principal / keytab, proxy user, HA + // nameservice config), none of which the pinned rust HDFS parser reads — the + // scan would open as the BE process's ambient identity instead of the catalog's + // configured one and fail the access JNI honors. See isRustVerifiedHdfsBackend. + final boolean hdfsBackendVerified; + + // The BE capability negotiation requires every candidate backend to report + // supports_paimon_rust_reader. + final boolean allBackendsRustCapable; + + private PaimonRustEligibility(boolean isFileStoreTable, boolean fallbackReadTable, + boolean restTokenTable, boolean queryAuthTable, boolean puAggDeletionVectors, + boolean dvMergeOnRead, boolean deduplicateIgnoreDelete, boolean rustUnsupportedMergeOption, + boolean scanModeNotDefault, boolean providerModeTranslatable, boolean schemeCapabilityVerified, + boolean hdfsBackendVerified, boolean allBackendsRustCapable) { + this.isFileStoreTable = isFileStoreTable; + this.fallbackReadTable = fallbackReadTable; + this.restTokenTable = restTokenTable; + this.queryAuthTable = queryAuthTable; + this.puAggDeletionVectors = puAggDeletionVectors; + this.dvMergeOnRead = dvMergeOnRead; + this.deduplicateIgnoreDelete = deduplicateIgnoreDelete; + this.rustUnsupportedMergeOption = rustUnsupportedMergeOption; + this.scanModeNotDefault = scanModeNotDefault; + this.providerModeTranslatable = providerModeTranslatable; + this.schemeCapabilityVerified = schemeCapabilityVerified; + this.hdfsBackendVerified = hdfsBackendVerified; + this.allBackendsRustCapable = allBackendsRustCapable; + } + + /** + * Whether the credential provider shape in the backend storage properties can be + * translated to the pinned paimon-rust reader without changing the identity the + * catalog configured. The rust S3 bridge maps static credentials, anonymous access + * (AWS_CREDENTIALS_PROVIDER_TYPE=ANONYMOUS -> s3.anonymous) and assume-role, but + * the remaining provider modes are ambient JVM provider chains (ENV, + * SYSTEM_PROPERTIES, WEB_IDENTITY, CONTAINER, INSTANCE_PROFILE) with no paimon-rust + * equivalent — rust would silently sign with whatever its own ambient chain + * resolves to. Lives here as one of the shared eligibility gates; also called + * directly for the JNI fallback-reason logging. + */ + static boolean providerModeTranslatable(String location, + Map<String, String> backendStorageProperties) { + String providerType = backendStorageProperties == null + ? null : backendStorageProperties.get("AWS_CREDENTIALS_PROVIDER_TYPE"); + String mode = providerType == null ? "DEFAULT" : providerType.trim().toUpperCase(Locale.ROOT); + if (mode.isEmpty()) { + mode = "DEFAULT"; + } + if (location != null && (location.startsWith("s3://") || location.startsWith("s3a://"))) { + // Java DEFAULT may resolve JVM properties or anonymous credentials; + // Rust's ambient chain is different. Only explicit keys or explicit + // anonymous access can cross this boundary without changing identity. + String accessKey = backendStorageProperties == null + ? null : backendStorageProperties.get("AWS_ACCESS_KEY"); + String secretKey = backendStorageProperties == null + ? null : backendStorageProperties.get("AWS_SECRET_KEY"); + boolean staticKeys = accessKey != null && !accessKey.trim().isEmpty() + && secretKey != null && !secretKey.trim().isEmpty(); + // Hadoop's Simple provider gives static keys precedence over anonymous, + // role and token settings. Rust interprets those settings differently. + boolean conflictingStaticSettings = staticKeys && (mode.equals("ANONYMOUS") + || StringUtils.isNotEmpty(backendStorageProperties.get("AWS_ROLE_ARN")) + || StringUtils.isNotEmpty(backendStorageProperties.get("AWS_TOKEN"))); + return !conflictingStaticSettings + && (mode.equals("ANONYMOUS") || (mode.equals("DEFAULT") && staticKeys)); + } + if (providerType != null) { + // OSS has no anonymous FileIO mode in the pinned Rust dependency. + return mode.equals("DEFAULT") + || (mode.equals("ANONYMOUS") && (location == null + || !location.startsWith("oss://"))); + } + return true; + } + + static PaimonRustEligibility of(Table table, String tableLocation, + Map<String, String> backendStorageProperties, Collection<Backend> backends) { + boolean isFileStoreTable = table instanceof FileStoreTable; + boolean fallbackReadTable = table instanceof FallbackReadFileStoreTable; + boolean restTokenTable = false; + boolean queryAuthTable = false; + boolean puAggDeletionVectors = false; + boolean dvMergeOnRead = false; + boolean deduplicateIgnoreDelete = false; + boolean rustUnsupportedMergeOption = false; + if (isFileStoreTable) { + FileStoreTable fileStoreTable = (FileStoreTable) table; + restTokenTable = fileStoreTable.fileIO() instanceof RESTTokenFileIO; + CoreOptions resolvedCoreOptions = fileStoreTable.coreOptions(); + // Null-safe: a table handle whose CoreOptions is not resolved (e.g. some + // wrapper shapes) stays rust-eligible rather than failing here — the rust + // open itself rejects such a table if the option is really set. + if (resolvedCoreOptions != null) { + queryAuthTable = resolvedCoreOptions.queryAuthEnabled(); + CoreOptions.MergeEngine mergeEngine = resolvedCoreOptions.mergeEngine(); + if (resolvedCoreOptions.deletionVectorsEnabled() + && (mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE + || mergeEngine == CoreOptions.MergeEngine.AGGREGATE)) { + puAggDeletionVectors = true; + TableSchema dvSchema = fileStoreTable.schema(); + Map<String, String> dvOptions = dvSchema == null ? null : dvSchema.options(); + String mergeOnRead = dvOptions == null + ? null : dvOptions.get(PaimonScanNode.DELETION_VECTORS_MERGE_ON_READ); + dvMergeOnRead = "true".equalsIgnoreCase(mergeOnRead); + } + if (mergeEngine == CoreOptions.MergeEngine.DEDUPLICATE + && resolvedCoreOptions.ignoreDelete()) { + deduplicateIgnoreDelete = true; + } + if (mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE + || mergeEngine == CoreOptions.MergeEngine.AGGREGATE) { + TableSchema mergeSchema = fileStoreTable.schema(); + Map<String, String> mergeOptions = + mergeSchema == null ? null : mergeSchema.options(); + rustUnsupportedMergeOption = mergeOptions != null + && PaimonScanNode.hasRustUnsupportedMergeOption(mergeOptions, mergeEngine); + } + } + } + // Null-safe on a null schema, mirroring the CoreOptions null-safety above: + // a fallback-shaped table can carry a null schema. + boolean scanModeNotDefault = false; + if (isFileStoreTable && ((FileStoreTable) table).schema() != null) { + TableSchema schema = ((FileStoreTable) table).schema(); + String scanMode = PaimonScanParams.withoutTimeTravelSelectors(schema) + .options().get(CoreOptions.SCAN_MODE.key()); + scanModeNotDefault = scanMode != null && !"default".equalsIgnoreCase(scanMode); + } + boolean schemeCapabilityVerified = PaimonScanNode.isRustVerifiedLocationScheme(tableLocation); + boolean hdfsBackendVerified = !PaimonScanNode.isHdfsLocationScheme(tableLocation) + || PaimonScanNode.isRustVerifiedHdfsBackend(backendStorageProperties); + boolean providerModeTranslatable = providerModeTranslatable( + tableLocation, backendStorageProperties); + boolean allBackendsRustCapable = backends != null && !backends.isEmpty() + && backends.stream().allMatch(Backend::isPaimonRustReaderSupported); Review Comment: Confirmed the mixed-version break: base BEs report supports_paimon_rust_reader=true with no vector protocol, so the rewrite would ship paimon_split="" plus the payload and the old _validate_rust_split rejects "missing paimon_split". Fixed as a separately negotiated bit: TBackendInfo field 12 supports_paimon_rust_vector_reader (12 is free on master — field 11 there is annotated as reserved for this PR's forward-port), set by BEs built from this branch, mapped through HeartbeatMgr/BackendHbResponse/Backend (with downgrade clearing), and required by a new vectorEligible() on the shared eligibility computation used by both the rewrite pre-flight and the scan-node gate. An old BE simply declines the vector rewrite; its batch rust reads are unaffected. Covered by capability UT (batch-only bit, both bits, downgrade, replay) and a rewrite UT; a mixed-cluster e2e cannot be built locally. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownVectorTopNIntoPaimonScan.java: ########## @@ -0,0 +1,513 @@ +// 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.TableScanParams; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.Type; +import org.apache.doris.datasource.credentials.CredentialUtils; +import org.apache.doris.datasource.credentials.VendedCredentialsFactory; +import org.apache.doris.datasource.paimon.PaimonExternalCatalog; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.datasource.paimon.PaimonUtil; +import org.apache.doris.datasource.paimon.source.PaimonVectorSearch; +import org.apache.doris.datasource.property.storage.StorageProperties; +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.ComparisonPredicate; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.InPredicate; +import org.apache.doris.nereids.trees.expressions.IsNull; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Not; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; +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.literal.ArrayLiteral; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.plans.AnnTopNInfo; +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.LogicalTopN; +import org.apache.doris.nereids.util.ExpressionUtils; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.thrift.TVectorMetric; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Maps; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeRoot; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.MultisetType; +import org.apache.paimon.types.RowType; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** + * Push a primary-key vector (ANN) top-N down into the Paimon scan (index-only). + * + * <p>Recognizes {@code TopN[order by dis] -> Project[dis := dist_fn(vec, q)] -> [Filter] -> PaimonScan} + * where {@code dist_fn} is {@code l2_distance_approximate} (ASC) or + * {@code inner_product_approximate} (DESC) and the Paimon table has a PK-vector + * index on {@code vec} with a matching distance metric. + * + * <p>The rewrite mirrors the internal-table ANN path (see + * {@code PushDownVectorTopNIntoOlapScan}): the scan is told to produce a + * reader-generated {@code __paimon_search_score_to_dis} column via the PK-vector index, and + * every occurrence of {@code dist_fn(vec, q)} is replaced by that column's slot. The + * order key slot keeps its identity, so the surrounding TopN merges/limits unchanged. + * The distance always comes from the index, never from the raw vector. + * + * <p>No arithmetic is emitted here even though paimon-rust reports a score rather than + * a distance: the BE reader undoes the metric-dependent score transform in place while + * filling the block, exactly as the internal table sqrt's faiss's squared distance inside + * the index layer. So the column arrives already holding what {@code dist_fn} would + * return, and the plan stays a bare slot reference -- which also keeps {@code dis} + * non-nullable (an expression over the score would not be, since {@code Divide} and + * {@code Sqrt} are always nullable) and makes predicates like + * {@code dist_fn(vec, q) < 10} correct without any direction flip. + * + * <p>Being index-only is just an optimization here, not a precondition: if nothing + * else references the raw vector column it is left unreferenced and dropped by column + * pruning (pure index-only read); if the query still needs it ({@code SELECT vec} or a + * predicate on {@code vec}), it stays a normal projected column and the BE vector + * reader materializes it alongside the score. Either way ANN is applied -- the rewrite + * never falls back merely because the vector is read. + */ +public class PushDownVectorTopNIntoPaimonScan implements RewriteRuleFactory { + + /** Doris column types the BE paimon-rust predicate converter can push; see + * PaimonRustPredicateConverter._is_supported_slot_type on the BE side. */ + private static final Set<Type> VECTOR_FILTER_SUPPORTED_COLUMN_TYPES = ImmutableSet.of( + Type.BOOLEAN, Type.TINYINT, Type.SMALLINT, Type.INT, Type.BIGINT, + Type.STRING, Type.DATE, Type.DATEV2); + + @Override + public List<Rule> buildRules() { + return ImmutableList.of( + logicalTopN(logicalProject(logicalFileScan())).when(t -> t.getOrderKeys().size() == 1).then(topN -> { + LogicalProject<LogicalFileScan> project = topN.child(); + LogicalFileScan scan = project.child(); + return pushDown(topN, project, scan, Optional.empty()); + }).toRule(RuleType.PUSH_DOWN_VECTOR_TOPN_INTO_PAIMON_SCAN), + logicalTopN(logicalProject(logicalFilter(logicalFileScan()))) + .when(t -> t.getOrderKeys().size() == 1).then(topN -> { + LogicalProject<LogicalFilter<LogicalFileScan>> project = topN.child(); + LogicalFilter<LogicalFileScan> filter = project.child(); + LogicalFileScan scan = filter.child(); + return pushDown(topN, project, scan, Optional.of(filter)); + }).toRule(RuleType.PUSH_DOWN_VECTOR_TOPN_INTO_PAIMON_SCAN) + ); + } + + private Plan pushDown( + LogicalTopN<?> topN, + LogicalProject<?> project, + LogicalFileScan scan, + Optional<LogicalFilter<?>> optionalFilter) { + // Only Paimon external tables carry PK-vector indexes. + if (!(scan.getTable() instanceof PaimonExternalTable)) { + return null; + } + + // The index-only search path is executed exclusively by the BE paimon-rust reader, + // which only FileScannerV2 can select (the V1 FileScanner explicitly rejects + // PAIMON_RUST). If the rust reader or the V2 scanner is disabled, the split would + // fall through to the JNI reader, which ignores the vector payload and would + // return a non-ANN scan with an empty distance column. Fall back to a normal + // scan then. + ConnectContext ctx = ConnectContext.get(); + if (ctx == null || ctx.getSessionVariable() == null + || !ctx.getSessionVariable().isEnablePaimonRustReader() + || !ctx.getSessionVariable().enableFileScannerV2) { + return null; + } + + // An incremental scan (t@incr / BETWEEN) must not be rewritten into a + // PK-vector search: the VectorScan builder has no incremental mode, so a + // vector payload would silently return snapshot rows for a query that asked + // for changes. Keeping the query un-rewritten runs it as the ordinary + // incremental scan the user requested, with the distance computed in Doris. + // (The scan node's vector gate keeps !isIncremental as the authoritative + // backstop for any path that produces a vector split regardless.) + // + // The same holds for a pinned file-creation-time read state: the ordinary + // path enforces `scan.file-creation-time-millis` by dropping manifest + // entries older than the bound, but the vector search plans its own splits + // and never consults that bound, so old files' nearest neighbors could leak + // into the result. Snapshot/tag/version selectors are fine -- they are + // resolved into the table handle before the vector search -- so only the + // file-level bound forces the decline. + if (scan.getScanParams().isPresent()) { + TableScanParams scanParams = scan.getScanParams().get(); + if (scanParams.incrementalRead()) { + return null; + } + Map<String, String> rawParams = scanParams.getMapParams(); + if (rawParams.containsKey(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())) { + return null; + } + } + + // A bounded vector top-N is only correct when every filter conjunct is + // applied inside the search, before the limit is exhausted. The BE + // paimon-rust predicate converter skips conjuncts it cannot represent (and + // the synthetic distance column is not in its field registry at all), and + // the engine re-applies those residuals on the returned block -- after the + // top-K rows were already selected. A skipped residual can therefore drop + // rows out of a bounded result: with candidate distances 0, 2 and 3, a + // `WHERE dis > 1 ORDER BY dis LIMIT 1` would fetch only the distance-0 row, + // filter it out and return nothing instead of the distance-2 row. So the + // rewrite only fires when each conjunct is provably of a form the converter + // pushes: a simple comparison / IN / null-check on a table column whose + // Doris type is in the converter's supported set (see + // PaimonRustPredicateConverter._is_supported_slot_type on BE). Anything + // else -- an expression predicate, a predicate on the distance itself, an + // unsupported column type -- declines the rewrite and the query runs as an + // ordinary scan with the distance computed in Doris, which is always + // correct. + if (optionalFilter.isPresent() && !conjunctsRideIntoVectorSearch(optionalFilter.get())) { + return null; + } + + // The order key must be a SlotReference produced by a Project alias. + Expression orderKey = topN.getOrderKeys().get(0).getExpr(); + if (!(orderKey instanceof SlotReference)) { + return null; + } + SlotReference keySlot = (SlotReference) orderKey; + Alias orderKeyAlias = null; + Expression orderKeyExpr = null; + for (NamedExpression projection : project.getProjects()) { + if (projection.toSlot().equals(keySlot) && projection instanceof Alias) { + orderKeyAlias = (Alias) projection; + orderKeyExpr = orderKeyAlias.child(); + break; + } + } + if (orderKeyExpr == null) { + return null; + } + + // l2 must sort ascending (smaller distance = closer); inner_product descending. + boolean isAsc = topN.getOrderKeys().get(0).isAsc(); + boolean l2Dist = orderKeyExpr instanceof L2DistanceApproximate; + boolean innerProduct = orderKeyExpr instanceof InnerProductApproximate; + if (!(l2Dist && isAsc) && !(innerProduct && !isAsc)) { + return null; + } + + // The indexed column must be a direct slot reference. A Cast chain over the + // column is NOT stripped: a legal cast can change vector values (for example + // an ARRAY<FLOAT> -> ARRAY<INT> -> ARRAY<FLOAT> round trip loses precision), + // and searching the original vectors while the query means the casted ones + // changes both the distance and the row order. Doris's implicit coercions do + // not produce a cast here for a matching ARRAY<FLOAT> column, so anything + // wrapped in a Cast simply declines the rewrite. + Expression left; + Expression right; + if (l2Dist) { + left = ((L2DistanceApproximate) orderKeyExpr).left(); + right = ((L2DistanceApproximate) orderKeyExpr).right(); + } else { + left = ((InnerProductApproximate) orderKeyExpr).left(); + right = ((InnerProductApproximate) orderKeyExpr).right(); + } + if (!(left instanceof SlotReference && right.isConstant() && right instanceof ArrayLiteral)) { + return null; + } + SlotReference vectorSlot = (SlotReference) left; + if (!vectorSlot.getOriginalColumn().isPresent() || !vectorSlot.getOriginalTable().isPresent()) { + return null; + } + String annColumnName = vectorSlot.getOriginalColumn().get().getName(); + + // Validate the PK-vector index and metric on the Paimon table. + PaimonExternalTable paimonTable = (PaimonExternalTable) scan.getTable(); + CoreOptions coreOptions; + try { + Table basePaimonTable = paimonTable.getPaimonTable(Optional.empty()); + coreOptions = CoreOptions.fromMap(basePaimonTable.options()); + } catch (Throwable t) { + return null; + } + if (!coreOptions.primaryKeyVectorIndexEnabled()) { + return null; + } + boolean columnIndexed = coreOptions.primaryKeyVectorIndexColumns().stream() + .anyMatch(c -> c.equalsIgnoreCase(annColumnName)); + if (!columnIndexed) { + return null; + } + // v1 supports only l2 / inner_product; the function must match the index metric. + String metric = coreOptions.primaryKeyVectorDistanceMetric(annColumnName); + String expectedMetric = l2Dist ? "l2" : "inner_product"; + if (!expectedMetric.equals(metric)) { + return null; + } + + // Pre-flight the projection-dependent bounds the vector split would hit at + // the scan node: a PK-vector split has no JNI fallback (it would return an + // empty distance column), so a query whose projected columns would set + // projectedVariant -- or an ORC table with TIMESTAMP_LTZ in its schema, the + // orc-ltz-shape bound -- must not be rewritten at all. Without the rewrite + // those queries run through the ordinary paths, where they are supported. + try { + if (projectedColumnsContainVariant(project, optionalFilter) + || orcSchemaContainsTimestampLtz(paimonTable)) { + return null; + } + } catch (Throwable t) { + return null; + } + + // The thrift-level enum the AnnTopNInfo will carry. expectedMetric is kept as a + // string for the comparison above (coreOptions returns a string), but the + // AnnTopNInfo / wire contract uses TVectorMetric, so map once here. DOT_PRODUCT + // is the thrift enum that paimon-rust's "inner_product" score transform maps to + // (see the BE reader's parse_score_transform). + TVectorMetric expectedMetricEnum = l2Dist ? TVectorMetric.L2 : TVectorMetric.DOT_PRODUCT; + + // Pre-flight the rust eligibility the scan node will gate on. The rewrite is + // one-way: a PK-vector split has no JNI fallback (the JNI reader ignores the + // vector payload and would return a non-ANN scan with an empty distance + // column), so the scan node can only fail loudly — a table whose gate would + // fail (credential provider shape, REST-token IO, query-auth, merge-option + // matrix, deletion-vector merge-on-read, scan-mode, storage scheme or HDFS + // credentials, no rust-capable backend) must not be rewritten at all. Without + // the rewrite the query still runs, computing the distance in Doris. The + // projection-dependent bounds (ORC LTZ, projected variant) need the final scan + // tuple and stay scan-node-side, and the scan node re-evaluates on the + // relation-option-processed table, so it stays authoritative for t@options + // overrides this raw-table pre-flight cannot see. + try { + Table baseTable = paimonTable.getPaimonTable(Optional.empty()); + PaimonExternalCatalog catalog = (PaimonExternalCatalog) paimonTable.getCatalog(); + Map<StorageProperties.Type, StorageProperties> storagePropertiesMap = + VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials( + catalog.getCatalogProperty().getMetastoreProperties(), + catalog.getCatalogProperty().getStoragePropertiesMap(), + baseTable); + Map<String, String> backendStorageProperties = + CredentialUtils.getBackendPropertiesFromStorageMap(storagePropertiesMap); + String tableLocation = baseTable instanceof FileStoreTable + ? ((FileStoreTable) baseTable).location().toString() + : baseTable.options().get("path"); + if (!PaimonVectorSearch.vectorScanEligible(baseTable, tableLocation, + backendStorageProperties, + Env.getCurrentSystemInfo().getAllClusterBackends(true))) { + return null; + } + } catch (Throwable t) { + return null; + } + // NOTE: the FE-side data-predicate interception is intentionally not done. The Review Comment: Confirmed verbatim against the pinned rust source: PkVectorSearchParams::resolve rejects every data predicate unless deletion vectors are enabled and merge-on-read is disabled, and the failure lands after the one-way rewrite replaced the distance expression. The rewrite now pre-flights the table mode via PaimonScanNode.vectorSearchSupportsDataPredicate (raw option keys, the same map the rust reader deserializes) — a non-partition data conjunct on a table without physical-row-read capability declines. Partition-only conjuncts are exempt: they are enforced by split planning and produce no per-row residual. The regression adds a first-row no-DV PK-vector fixture: unfiltered still rewrites, filtered declines and both legs agree. -- 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]
