JingsongLi commented on code in PR #10777: URL: https://github.com/apache/paimon/pull/10777#discussion_r4374024061
########## paimon-elasticsearch/src/main/java/org/apache/paimon/eslib/javaindex/JavaEsQueryParser.java: ########## @@ -0,0 +1,368 @@ +/* + * 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.paimon.eslib.javaindex; + +import org.apache.paimon.globalindex.EsSearchQuery; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode; + +import org.apache.lucene.analysis.Analyzer; +import org.apache.lucene.document.DoublePoint; +import org.apache.lucene.document.FloatPoint; +import org.apache.lucene.document.IntPoint; +import org.apache.lucene.document.LongPoint; +import org.apache.lucene.index.Term; +import org.apache.lucene.search.BooleanClause; +import org.apache.lucene.search.BooleanQuery; +import org.apache.lucene.search.ConstantScoreQuery; +import org.apache.lucene.search.FieldExistsQuery; +import org.apache.lucene.search.KnnFloatVectorQuery; +import org.apache.lucene.search.MatchAllDocsQuery; +import org.apache.lucene.search.MatchNoDocsQuery; +import org.apache.lucene.search.Query; +import org.apache.lucene.search.TermInSetQuery; +import org.apache.lucene.search.TermQuery; +import org.apache.lucene.search.TermRangeQuery; +import org.apache.lucene.util.BytesRef; +import org.apache.lucene.util.QueryBuilder; +import org.apache.lucene.util.VectorUtil; +import org.elasticsearch.eslib.analyzer.BuiltinAnalyzers; +import org.elasticsearch.eslib.api.model.BuiltinAnalyzer; +import org.elasticsearch.eslib.api.model.FieldIndexConfig; +import org.elasticsearch.eslib.api.model.ScalarFieldType; +import org.elasticsearch.eslib.api.model.VectorAlgorithm; + +import java.math.BigDecimal; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import static org.apache.paimon.globalindex.EsSearchQuery.require; + +/** Translates the validated, bounded search subset to the Java archive's Lucene queries. */ +final class JavaEsQueryParser { + private final Map<String, FieldIndexConfig> fields; + + JavaEsQueryParser(Map<String, FieldIndexConfig> fields) { + this.fields = fields; + } + + Query parse(EsSearchQuery query) { + return clause(EsSearchQuery.parse(query.queryJson())); + } + + float[] vector(JsonNode body) { + String name = body.get("field").textValue(); + FieldIndexConfig field = field(name); + require( + field.indexType() == FieldIndexConfig.IndexType.VECTOR, + "knn requires a vector field"); + JsonNode values = body.get("query_vector"); + require( + values.size() == field.dimension(), + "Query vector dimension does not match the index"); + float[] vector = new float[values.size()]; + for (int i = 0; i < vector.length; i++) { + vector[i] = values.get(i).floatValue(); + } + if (field.algorithm() == VectorAlgorithm.DISKBBQ + && "cosine".equalsIgnoreCase(field.metric())) { + VectorUtil.l2normalize(vector); + } + return vector; + } + + private Query clause(JsonNode node) { + String kind = node.fieldNames().next(); + JsonNode body = node.get(kind); + switch (kind) { + case "match_all": + return new MatchAllDocsQuery(); + case "match_none": + return new MatchNoDocsQuery(); + case "bool": + return bool(body); + case "exists": + String name = body.get("field").textValue(); + field(name); + return new FieldExistsQuery(name); + case "knn": + return new KnnFloatVectorQuery( + body.get("field").textValue(), + vector(body), + body.get("num_candidates").intValue(), + body.has("filter") ? clause(body.get("filter")) : null); + default: + String fieldName = body.fieldNames().next(); + FieldIndexConfig field = field(fieldName); + JsonNode value = body.get(fieldName); + switch (kind) { + case "term": + return field.indexType() == FieldIndexConfig.IndexType.FULLTEXT + ? new TermQuery(new Term(fieldName, value.asText())) + : terms(field, Collections.singletonList(value)); + case "terms": + List<JsonNode> values = new ArrayList<>(); + value.forEach(values::add); + return terms(field, values); + case "range": + return range(field, value); + case "match": + case "match_phrase": + return match(field, value, "match_phrase".equals(kind)); + default: + throw new IllegalArgumentException("Unsupported Java ES query: " + kind); + } + } + } + + private FieldIndexConfig field(String name) { + FieldIndexConfig field = fields.get(name); + require(field != null, "Field is absent from the selected ES index: " + name); + return field; + } + + private Query bool(JsonNode body) { + BooleanQuery.Builder result = new BooleanQuery.Builder(); + int positive = 0; + int required = 0; + int should = 0; + int total = 0; + for (String key : Arrays.asList("must", "filter", "should", "must_not")) { + JsonNode value = body.get(key); + if (value == null) { + continue; + } + BooleanClause.Occur occur; + switch (key) { + case "must": + occur = BooleanClause.Occur.MUST; + break; + case "filter": + occur = BooleanClause.Occur.FILTER; + break; + case "should": + occur = BooleanClause.Occur.SHOULD; + break; + default: + occur = BooleanClause.Occur.MUST_NOT; + } + Iterable<JsonNode> children = + value.isArray() ? value : Collections.singletonList(value); + for (JsonNode child : children) { + result.add(clause(child), occur); + total++; + if (occur != BooleanClause.Occur.MUST_NOT) { + positive++; + } + if (occur == BooleanClause.Occur.MUST || occur == BooleanClause.Occur.FILTER) { + required++; + } + if (occur == BooleanClause.Occur.SHOULD) { + should++; + } + } + } + int minimum = + body.has("minimum_should_match") + ? body.get("minimum_should_match").intValue() + : (required == 0 && should > 0 ? 1 : 0); + if (minimum > should) { + return new MatchNoDocsQuery(); + } + if (total == 0) { + return new MatchAllDocsQuery(); + } + if (positive == 0) { + result.add(new MatchAllDocsQuery(), BooleanClause.Occur.MUST); Review Comment: [P2] Keep pure-negative bool clauses out of the relevance score The synthetic match-all is added as MUST, so each pure-negative bool contributes 1.0 instead of acting only as an exclusion. This changes result membership after Top-K truncation. With row 0 `(tag=x,n=1)` and row 1 `(tag=y,n=2)`, the query `{"bool":{"should":[{"bool":{"must_not":{"term":{"tag":"x"}}}},{"bool":{"must_not":{"term":{"n":1}}}},{"term":{"n":1}}]}}` ranks row 1 at 2.0 ahead of row 0 at 1.0. With `size=1`, the positive numeric match is lost. I reproduced this against the exact PR parser. Elasticsearch's [fixNegativeQueryIfNeeded](https://github.com/elastic/elasticsearch/blob/v8.17.0/server/src/main/java/org/elasticsearch/common/lucene/search/Queries.java#L95-L103) adds the match-all as FILTER; that produces row 0 at 1.0, then row 1 at 0.0. Please use FILTER here and add a nested-negative score/ranking regression. ########## paimon-core/src/main/java/org/apache/paimon/globalindex/EsSearchResult.java: ########## @@ -0,0 +1,130 @@ +/* + * 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.paimon.globalindex; + +import org.apache.paimon.globalindex.EsIndexSearch.Hit; +import org.apache.paimon.utils.JsonSerdeUtil; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +/** Bounded HIT/PAGE_END records shared by Spark, Flink and the Java SDK. */ +public final class EsSearchResult { + private static final String[] COLUMNS = { + "record_kind", + "query_id", + "query_ordinal", + "execution_query_id", + "target", + "table_name", + "branch", + "snapshot_id", + "row_id", + "score", + "rank", + "sort_values_json", + "source_schema_json", + "source_binary", + "source_encoding", + "metadata_json", + "index_fields_json" + }; + + private final List<Object[]> records; + private final List<Hit> hits; + + EsSearchResult(EsSearchRequest request, EsIndexSearch.Plan plan, List<Hit> hits) + throws Exception { + this.hits = Collections.unmodifiableList(new ArrayList<>(hits)); + Map<Long, byte[]> source = request.fetchSource ? plan.fetch(hits) : Collections.emptyMap(); + String schema = request.fetchSource ? JsonSerdeUtil.toJson(plan.sourceType()) : null; + String queryId = UUID.randomUUID().toString(); + ObjectNode metadata = JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.createObjectNode(); + metadata.put("snapshotId", Long.toString(plan.snapshotId)); + metadata.put("indexType", "es-index"); + metadata.put("coverageComplete", true); + metadata.put("scoring", "QUERY_THEN_FETCH"); + metadata.put("totalHitsTracked", false); + metadata.put("hitCount", hits.size()); + records = new ArrayList<>(); + for (int i = 0; i <= hits.size(); i++) { + boolean end = i == hits.size(); + Object[] row = new Object[COLUMNS.length]; + row[0] = end ? "PAGE_END" : "HIT"; + row[1] = queryId; + row[2] = 0; + row[3] = queryId; + row[4] = request.target; + row[5] = request.database + "." + request.table; + row[6] = "main"; Review Comment: [P2] Reject non-main tables before emitting main-branch identities The request validator checks only the separate `branch` member. `"table":"T$branch_dev"` is accepted and `Identifier.create` resolves it to the dev branch; the public `prepare(FileStoreTable, request)` overload also accepts a dev table. I reproduced a complete SDK search with a real indexed dev branch and source fetch: `resolvedBranch=dev, emittedBranch=main, hits=4`. Every returned hit and PAGE_END therefore identifies a dev snapshot/row as main, despite the documented main-only contract. Please reject non-main identifiers/resolved tables before planning (including the FileStoreTable overload), or explicitly support branches and emit the actual identity. -- 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]
