Gabriel39 commented on code in PR #69672:
URL: https://github.com/apache/doris/pull/69672#discussion_r4386561430
##########
fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsMappingUtils.java:
##########
@@ -153,9 +162,101 @@ public static EsFieldContext
resolveFieldContext(List<String> columnNames,
resolveKeywordFields(fieldContext, fieldObject, colName,
fieldType);
resolveDocValuesFields(fieldContext, fieldObject, colName,
fieldType);
}
+ resolveSourceValueSafety(fieldContext, indexMapping, mappingType);
return fieldContext;
}
+ private static void resolveSourceValueSafety(EsFieldContext context,
String indexMapping, String mappingType) {
+ try {
+ // An alias can span mappings. Every backing index must preserve
source semantics before
+ // a predicate may remove rows or doc_values may replace the value
returned to Doris.
+ List<ObjectNode> roots = new ArrayList<>();
+ Set<String> columns = new
HashSet<>(context.getColumn2typeMap().keySet());
+ for (JsonNode index : MAPPER.readTree(indexMapping)) {
+ ObjectNode root = getRootSchema((ObjectNode)
index.get("mappings"), mappingType,
+ new ArrayList<>(), true);
+ roots.add(root);
+ }
+ // Only the first mapping's schema is exposed by Doris, including
unprojected filter columns.
+
roots.get(0).path("properties").fieldNames().forEachRemaining(columns::add);
+ Set<String> inexactKeywords = new HashSet<>();
+ for (ObjectNode root : roots) {
+ JsonNode properties = root.path("properties");
+ JsonNode source = root.path("_source");
+ boolean filteredSource =
!source.path("enabled").asBoolean(true)
+ || source.has("includes") || source.has("excludes");
+ Set<String> copyTargets = new HashSet<>();
+ collectCopyTargets(properties, copyTargets);
+ for (String column : columns) {
+ JsonNode field = properties.path(column);
+ String indexedPath =
context.getFetchFieldsContext().get(column);
+ JsonNode indexedField = field;
+ if (indexedPath != null) {
+ indexedField =
field.path("fields").path(indexedPath.substring(column.length() + 1));
+ }
+ boolean inexact = filteredSource ||
root.path("runtime").has(column)
+ || copyTargets.contains(column) ||
changesSourceValueExceptLength(field)
+ || changesSourceValueExceptLength(indexedField);
+ for (JsonNode subfield : field.path("fields")) {
+ if ("keyword".equals(subfield.path("type").asText())
+ && changesSourceValueExceptLength(subfield)) {
+ inexact = true;
Review Comment:
Fixed in cff8803bf3. Query restrictions inspect the selected keyword path;
docvalue eligibility independently inspects the selected read path on every
backing. Unused normalized siblings no longer taint a raw path. Also corrected
explicit doc_values=true multi-fields to request their full subfield path, and
reject docvalue reads when a backing does not provide that selected path. Tests
cover safe/safe, safe-query/unsafe-read, unsafe-query/safe-read, keyword sniff
disabled, and later-backing read mismatches. Live ES8 SQL verifies native OR
and projected raw values across both scanners and docvalue modes.
##########
fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsMappingUtils.java:
##########
@@ -153,9 +162,101 @@ public static EsFieldContext
resolveFieldContext(List<String> columnNames,
resolveKeywordFields(fieldContext, fieldObject, colName,
fieldType);
resolveDocValuesFields(fieldContext, fieldObject, colName,
fieldType);
}
+ resolveSourceValueSafety(fieldContext, indexMapping, mappingType);
return fieldContext;
}
+ private static void resolveSourceValueSafety(EsFieldContext context,
String indexMapping, String mappingType) {
+ try {
+ // An alias can span mappings. Every backing index must preserve
source semantics before
+ // a predicate may remove rows or doc_values may replace the value
returned to Doris.
+ List<ObjectNode> roots = new ArrayList<>();
+ Set<String> columns = new
HashSet<>(context.getColumn2typeMap().keySet());
+ for (JsonNode index : MAPPER.readTree(indexMapping)) {
+ ObjectNode root = getRootSchema((ObjectNode)
index.get("mappings"), mappingType,
+ new ArrayList<>(), true);
+ roots.add(root);
+ }
+ // Only the first mapping's schema is exposed by Doris, including
unprojected filter columns.
+
roots.get(0).path("properties").fieldNames().forEachRemaining(columns::add);
+ Set<String> inexactKeywords = new HashSet<>();
+ for (ObjectNode root : roots) {
+ JsonNode properties = root.path("properties");
+ JsonNode source = root.path("_source");
+ boolean filteredSource =
!source.path("enabled").asBoolean(true)
+ || source.has("includes") || source.has("excludes");
+ Set<String> copyTargets = new HashSet<>();
+ collectCopyTargets(properties, copyTargets);
+ for (String column : columns) {
+ JsonNode field = properties.path(column);
+ String indexedPath =
context.getFetchFieldsContext().get(column);
+ JsonNode indexedField = field;
+ if (indexedPath != null) {
+ indexedField =
field.path("fields").path(indexedPath.substring(column.length() + 1));
+ }
+ boolean inexact = filteredSource ||
root.path("runtime").has(column)
+ || copyTargets.contains(column) ||
changesSourceValueExceptLength(field)
+ || changesSourceValueExceptLength(indexedField);
+ for (JsonNode subfield : field.path("fields")) {
+ if ("keyword".equals(subfield.path("type").asText())
+ && changesSourceValueExceptLength(subfield)) {
+ inexact = true;
+ }
+ }
+ if (inexact || field.has("ignore_above") ||
indexedField.has("ignore_above")
+ || hasUnsafeKeywordSubfield(field)) {
+ context.getSourceValueUnsafeFields().add(column);
+ }
+ if (!inexact &&
"keyword".equals(indexedField.path("type").asText())) {
+ // Equality/IN with bounded literals stay exact:
ignored longer values cannot
+ // equal those literals. Intersect the bound over
every backing index.
+ context.getKeywordIgnoreAboveLimits().merge(column,
+
indexedField.path("ignore_above").asInt(Integer.MAX_VALUE), Math::min);
+ } else {
+ inexactKeywords.add(column);
+ }
+ }
+ }
+
inexactKeywords.forEach(context.getKeywordIgnoreAboveLimits()::remove);
+ } catch (IOException e) {
+ throw new DorisConnectorException("Failed to parse ES mapping: " +
e.getMessage());
+ }
+ // Local residuals must read the original value, including with
explicit doc_values=true.
+
context.getSourceValueUnsafeFields().forEach(context.getDocValueFieldsContext()::remove);
+ }
+
+ private static boolean hasUnsafeKeywordSubfield(JsonNode field) {
+ for (JsonNode subfield : field.path("fields")) {
+ if ("keyword".equals(subfield.path("type").asText()) &&
changesSourceValue(subfield)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean changesSourceValue(JsonNode field) {
+ return field.has("ignore_above") ||
changesSourceValueExceptLength(field);
+ }
+
+ private static boolean changesSourceValueExceptLength(JsonNode field) {
+ return field.isMissingNode() || field.has("normalizer")
+ || field.has("null_value") ||
field.path("ignore_malformed").asBoolean(false)
+ || !field.path("index").asBoolean(true);
Review Comment:
Fixed in cff8803bf3. Disabling the inverted index is separated from
source-value transformations and docvalue read safety. A keyword with docvalues
enabled, explicitly or by default, retains its existing remote query path; a
keyword with both index and docvalues disabled keeps a local filter. Added FE
controls and live ES8 native-OR plus projected-value checks for a
doc-value-only keyword.
--
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]
