From 5d2ad61df689540c8186aa423049615d8f078e16 Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Mon, 27 Jul 2026 11:46:11 -0700 Subject: [PATCH 1/8] Add non-fatal warning channel to PPL query response Introduce a first-class warnings array on the query response so the engine can attach non-fatal, structured notices (type/message/detail) to an otherwise-successful result, without turning it into an error. This is the prerequisite for returning a labeled partial result instead of failing a query that would otherwise exhaust Point-In-Time contexts. - Add Warning value type and QueryResponse.warnings (core). - Collect warnings during Calcite planning via a thread-local on CalcitePlanContext, drained in OpenSearchExecutionEngine.buildResultSet and cleared with the other lifecycle signals so nothing leaks across queries. - Thread warnings through QueryResult and emit them in SimpleJsonResponseFormatter only when non-empty, so existing responses are byte-for-byte unchanged. Scoped to the Calcite PPL path. Signed-off-by: Kai Huang --- .../sql/calcite/CalcitePlanContext.java | 27 ++++++++++++ .../sql/executor/ExecutionEngine.java | 3 ++ .../org/opensearch/sql/executor/Warning.java | 25 +++++++++++ .../executor/OpenSearchExecutionEngine.java | 1 + .../transport/TransportPPLQueryAction.java | 6 ++- .../sql/protocol/response/QueryResult.java | 15 +++++++ .../format/SimpleJsonResponseFormatter.java | 8 ++++ .../SimpleJsonResponseFormatterTest.java | 44 +++++++++++++++++++ 8 files changed, 128 insertions(+), 1 deletion(-) create mode 100644 core/src/main/java/org/opensearch/sql/executor/Warning.java diff --git a/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java b/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java index b715fdac86d..344f5d0c030 100644 --- a/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java +++ b/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java @@ -31,6 +31,7 @@ import org.opensearch.sql.calcite.utils.CalciteToolsHelper.OpenSearchRelBuilder; import org.opensearch.sql.common.setting.Settings; import org.opensearch.sql.executor.QueryType; +import org.opensearch.sql.executor.Warning; import org.opensearch.sql.expression.function.FunctionProperties; public class CalcitePlanContext { @@ -58,6 +59,15 @@ public class CalcitePlanContext { /** Timewrap series mode: "relative", "short", or "exact". */ public static final ThreadLocal timewrapSeries = new ThreadLocal<>(); + /** + * Non-fatal warnings raised during planning (e.g. a partial result over a subset of indices) to + * be attached to the query response by the execution engine. Drained in {@code + * OpenSearchExecutionEngine.buildResultSet} and cleared with the other lifecycle signals so it + * never leaks onto the next query on a pooled worker thread. + */ + private static final ThreadLocal> pendingWarnings = + ThreadLocal.withInitial(ArrayList::new); + /** Thread-local switch that tells whether the current query prefers legacy behavior. */ private static final ThreadLocal legacyPreferredFlag = ThreadLocal.withInitial(() -> true); @@ -245,6 +255,23 @@ public static void clearTimewrapSignals() { stripNullColumns.set(false); timewrapUnitName.set(null); timewrapSeries.set(null); + pendingWarnings.remove(); + } + + /** Records a non-fatal warning to be attached to the response for the current query. */ + public static void addWarning(Warning warning) { + pendingWarnings.get().add(warning); + } + + /** Returns and clears the warnings collected for the current query. */ + public static List drainWarnings() { + List warnings = pendingWarnings.get(); + if (warnings.isEmpty()) { + return List.of(); + } + List drained = List.copyOf(warnings); + pendingWarnings.remove(); + return drained; } public void pushForeachBindings( diff --git a/core/src/main/java/org/opensearch/sql/executor/ExecutionEngine.java b/core/src/main/java/org/opensearch/sql/executor/ExecutionEngine.java index 9b51876c004..6f5276424cc 100644 --- a/core/src/main/java/org/opensearch/sql/executor/ExecutionEngine.java +++ b/core/src/main/java/org/opensearch/sql/executor/ExecutionEngine.java @@ -94,6 +94,9 @@ class QueryResponse { private final Cursor cursor; @lombok.Setter private QueryProfile profile; @lombok.Setter private Throwable error; + + /** Non-fatal notices attached to a successful result; empty for a plain success. */ + @lombok.Setter private List warnings = List.of(); } @Data diff --git a/core/src/main/java/org/opensearch/sql/executor/Warning.java b/core/src/main/java/org/opensearch/sql/executor/Warning.java new file mode 100644 index 00000000000..99ca7f53f58 --- /dev/null +++ b/core/src/main/java/org/opensearch/sql/executor/Warning.java @@ -0,0 +1,25 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.executor; + +import lombok.Data; + +/** + * A non-fatal notice attached to an otherwise-successful query response. Carried through the + * response path so consumers can distinguish a correct-but-noteworthy result (e.g. a partial result + * over a subset of indices) from a plain success, without turning it into an error. + */ +@Data +public class Warning { + /** Machine-readable category, e.g. {@code PARTIAL_RESULT}. */ + private final String type; + + /** Short human-readable summary. */ + private final String message; + + /** Optional longer explanation with the specifics and remedy; may be null. */ + private final String detail; +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/executor/OpenSearchExecutionEngine.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/executor/OpenSearchExecutionEngine.java index 2e37782b72b..7a9d4684a90 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/executor/OpenSearchExecutionEngine.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/executor/OpenSearchExecutionEngine.java @@ -491,6 +491,7 @@ private QueryResponse buildResultSet( Schema schema = new Schema(columns); QueryResponse response = new QueryResponse(schema, values, null); + response.setWarnings(CalcitePlanContext.drainWarnings()); return response; } diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java b/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java index b54c2c49ab9..0cbb75d9e1c 100644 --- a/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java +++ b/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java @@ -323,7 +323,11 @@ public void onResponse(ExecutionEngine.QueryResponse response) { String responseContent = formatter.format( new QueryResult( - response.getSchema(), response.getResults(), response.getCursor(), PPL_SPEC)); + response.getSchema(), + response.getResults(), + response.getCursor(), + PPL_SPEC, + response.getWarnings())); listener.onResponse(new TransportPPLQueryResponse(responseContent)); } diff --git a/protocol/src/main/java/org/opensearch/sql/protocol/response/QueryResult.java b/protocol/src/main/java/org/opensearch/sql/protocol/response/QueryResult.java index a1ea6d462e6..cb60d808964 100644 --- a/protocol/src/main/java/org/opensearch/sql/protocol/response/QueryResult.java +++ b/protocol/src/main/java/org/opensearch/sql/protocol/response/QueryResult.java @@ -8,6 +8,7 @@ import java.util.Collection; import java.util.Iterator; import java.util.LinkedHashMap; +import java.util.List; import java.util.Locale; import java.util.Map; import lombok.Getter; @@ -15,6 +16,7 @@ import org.opensearch.sql.data.model.ExprValueUtils; import org.opensearch.sql.executor.ExecutionEngine; import org.opensearch.sql.executor.ExecutionEngine.Schema.Column; +import org.opensearch.sql.executor.Warning; import org.opensearch.sql.executor.pagination.Cursor; import org.opensearch.sql.lang.LangSpec; @@ -33,6 +35,9 @@ public class QueryResult implements Iterable { private final LangSpec langSpec; + /** Non-fatal notices attached to a successful result; empty for a plain success. */ + @Getter private final List warnings; + public QueryResult(ExecutionEngine.Schema schema, Collection exprValues) { this(schema, exprValues, Cursor.None, LangSpec.SQL_SPEC); } @@ -47,10 +52,20 @@ public QueryResult( Collection exprValues, Cursor cursor, LangSpec langSpec) { + this(schema, exprValues, cursor, langSpec, List.of()); + } + + public QueryResult( + ExecutionEngine.Schema schema, + Collection exprValues, + Cursor cursor, + LangSpec langSpec, + List warnings) { this.schema = schema; this.exprValues = exprValues; this.cursor = cursor; this.langSpec = langSpec; + this.warnings = warnings == null ? List.of() : warnings; } /** diff --git a/protocol/src/main/java/org/opensearch/sql/protocol/response/format/SimpleJsonResponseFormatter.java b/protocol/src/main/java/org/opensearch/sql/protocol/response/format/SimpleJsonResponseFormatter.java index 88afd636712..3680802ffec 100644 --- a/protocol/src/main/java/org/opensearch/sql/protocol/response/format/SimpleJsonResponseFormatter.java +++ b/protocol/src/main/java/org/opensearch/sql/protocol/response/format/SimpleJsonResponseFormatter.java @@ -10,6 +10,7 @@ import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.Singular; +import org.opensearch.sql.executor.Warning; import org.opensearch.sql.monitor.profile.MetricName; import org.opensearch.sql.monitor.profile.ProfileMetric; import org.opensearch.sql.monitor.profile.QueryProfile; @@ -55,6 +56,10 @@ public Object buildJsonObject(QueryResult response) { json.datarows(fetchDataRows(response)); + if (!response.getWarnings().isEmpty()) { + json.warnings(response.getWarnings()); + } + formatMetric.set(System.nanoTime() - formatTime); json.profile(QueryProfiling.current().finish()); @@ -83,6 +88,9 @@ public static class JsonResponse { private long total; private long size; + + /** Present only when non-empty; a plain success omits this field entirely. */ + private final List warnings; } @RequiredArgsConstructor diff --git a/protocol/src/test/java/org/opensearch/sql/protocol/response/format/SimpleJsonResponseFormatterTest.java b/protocol/src/test/java/org/opensearch/sql/protocol/response/format/SimpleJsonResponseFormatterTest.java index 778b97f892c..1734dcc2734 100644 --- a/protocol/src/test/java/org/opensearch/sql/protocol/response/format/SimpleJsonResponseFormatterTest.java +++ b/protocol/src/test/java/org/opensearch/sql/protocol/response/format/SimpleJsonResponseFormatterTest.java @@ -26,6 +26,9 @@ import org.opensearch.sql.data.model.ExprValue; import org.opensearch.sql.data.model.ExprValueUtils; import org.opensearch.sql.executor.ExecutionEngine; +import org.opensearch.sql.executor.Warning; +import org.opensearch.sql.executor.pagination.Cursor; +import org.opensearch.sql.lang.LangSpec; import org.opensearch.sql.protocol.response.QueryResult; class SimpleJsonResponseFormatterTest { @@ -89,6 +92,47 @@ void formatResponsePretty() { formatter.format(response)); } + @Test + void formatResponseWithWarnings() { + QueryResult response = + new QueryResult( + schema, + Arrays.asList(tupleValue(ImmutableMap.of("firstname", "John", "age", 20))), + Cursor.None, + LangSpec.SQL_SPEC, + List.of( + new Warning( + "PARTIAL_RESULT", + "Results exclude 1 index due to a mapping conflict.", + "Field [applicationid] is mapped as text in [logs-conflict-one]."))); + SimpleJsonResponseFormatter formatter = new SimpleJsonResponseFormatter(COMPACT); + assertEquals( + "{\"schema\":[{\"name\":\"firstname\",\"type\":\"string\"}," + + "{\"name\":\"age\",\"type\":\"integer\"}],\"datarows\":[[\"John\",20]]," + + "\"total\":1,\"size\":1," + + "\"warnings\":[{\"type\":\"PARTIAL_RESULT\"," + + "\"message\":\"Results exclude 1 index due to a mapping conflict.\"," + + "\"detail\":\"Field [applicationid] is mapped as text in [logs-conflict-one].\"}]}", + formatter.format(response)); + } + + @Test + void formatResponseWithoutWarningsOmitsField() { + QueryResult response = + new QueryResult( + schema, + Arrays.asList(tupleValue(ImmutableMap.of("firstname", "John", "age", 20))), + Cursor.None, + LangSpec.SQL_SPEC, + List.of()); + SimpleJsonResponseFormatter formatter = new SimpleJsonResponseFormatter(COMPACT); + assertEquals( + "{\"schema\":[{\"name\":\"firstname\",\"type\":\"string\"}," + + "{\"name\":\"age\",\"type\":\"integer\"}],\"datarows\":[[\"John\",20]]," + + "\"total\":1,\"size\":1}", + formatter.format(response)); + } + @Test void formatResponseSchemaWithAlias() { ExecutionEngine.Schema schema = From 418f45e8501e27b760d89b58beb1e78a49690fbc Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Mon, 27 Jul 2026 12:21:00 -0700 Subject: [PATCH 2/8] Return a partial result instead of exhausting PIT on a mapping conflict When an aggregation groups by a field that a text/keyword mapping conflict collapsed to text-without-keyword across a wildcard pattern, pushdown fails and the query falls back to a per-shard document scan that opens a Point-In-Time context on every shard, tripping search.max_open_pit_context. Add an opt-in fallback: when normal aggregate pushdown fails and plugins.calcite.partial_result.on_mapping_conflict.enabled is set, narrow the scan to the largest homogeneous subset of indices whose mapping of the grouped field is aggregatable, push the aggregation down over just that subset (size=0, no PIT), and attach a PARTIAL_RESULT warning naming the excluded indices. - New default-off setting, registered in OpenSearchSettings. - tryPartialResultAggregate in CalciteLogicalIndexScan partitions the matched indices, keeps the keyword group (or text-with-keyword if no keyword group), rebinds the pushdown context to the narrowed index preserving pushed operations such as a WHERE filter, and re-runs pushdown. - Wired into AggregateIndexScanRule as the fallback when pushdown returns null. - drainWarnings de-duplicates, since the planner may raise the warning for multiple equivalent plan alternatives. - Integration test verifies: partial off fails with PIT exhaustion; partial on returns the keyword-index counts with the warning and no PIT; a conflict-free aggregation attaches no warning. Scoped to the Calcite PPL path. Signed-off-by: Kai Huang --- .../sql/common/setting/Settings.java | 2 + .../sql/calcite/CalcitePlanContext.java | 8 +- ...lcitePartialResultOnMappingConflictIT.java | 140 +++++++++++++++++ .../planner/rules/AggregateIndexScanRule.java | 6 + .../setting/OpenSearchSettings.java | 14 ++ .../storage/scan/CalciteLogicalIndexScan.java | 146 ++++++++++++++++++ .../storage/scan/context/PushDownContext.java | 15 ++ 7 files changed, 329 insertions(+), 2 deletions(-) create mode 100644 integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java diff --git a/common/src/main/java/org/opensearch/sql/common/setting/Settings.java b/common/src/main/java/org/opensearch/sql/common/setting/Settings.java index bf3e65d8741..3f86799d047 100644 --- a/common/src/main/java/org/opensearch/sql/common/setting/Settings.java +++ b/common/src/main/java/org/opensearch/sql/common/setting/Settings.java @@ -44,6 +44,8 @@ public enum Key { CALCITE_PUSHDOWN_ROWCOUNT_ESTIMATION_FACTOR( "plugins.calcite.pushdown.rowcount.estimation.factor"), CALCITE_SUPPORT_ALL_JOIN_TYPES("plugins.calcite.all_join_types.allowed"), + CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT( + "plugins.calcite.partial_result.on_mapping_conflict.enabled"), /** Query Settings. */ FIELD_TYPE_TOLERANCE("plugins.query.field_type_tolerance"), diff --git a/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java b/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java index 344f5d0c030..dc48f643ee7 100644 --- a/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java +++ b/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java @@ -263,13 +263,17 @@ public static void addWarning(Warning warning) { pendingWarnings.get().add(warning); } - /** Returns and clears the warnings collected for the current query. */ + /** + * Returns and clears the warnings collected for the current query, de-duplicated by value. The + * planner may fire a rule that raises a warning more than once for equivalent plan alternatives, + * so identical warnings are collapsed to one. + */ public static List drainWarnings() { List warnings = pendingWarnings.get(); if (warnings.isEmpty()) { return List.of(); } - List drained = List.copyOf(warnings); + List drained = warnings.stream().distinct().toList(); pendingWarnings.remove(); return drained; } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java new file mode 100644 index 00000000000..484790bda12 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java @@ -0,0 +1,140 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.calcite.remote; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.opensearch.sql.util.MatcherUtils.rows; +import static org.opensearch.sql.util.MatcherUtils.verifyDataRows; +import static org.opensearch.sql.util.TestUtils.createIndexByRestClient; +import static org.opensearch.sql.util.TestUtils.isIndexExist; +import static org.opensearch.sql.util.TestUtils.performRequest; + +import java.io.IOException; +import org.json.JSONArray; +import org.json.JSONObject; +import org.junit.After; +import org.junit.Test; +import org.opensearch.client.Request; +import org.opensearch.client.ResponseException; +import org.opensearch.sql.common.setting.Settings; +import org.opensearch.sql.ppl.PPLIntegTestCase; + +/** + * End-to-end tests for the partial-result fallback on a text/keyword mapping conflict. A field + * mapped as {@code keyword} in one index and {@code text} (without a {@code .keyword} sub-field) in + * another collapses to text-without-keyword across the wildcard pattern, which defeats aggregation + * pushdown and forces a per-shard document scan that opens a Point-In-Time context on every shard. + * + *

When {@code plugins.calcite.partial_result.on_mapping_conflict.enabled} is on, the aggregation + * is instead pushed down over just the aggregatable (keyword) index subset — no PIT — and the + * response carries a {@code PARTIAL_RESULT} warning naming the excluded index. + */ +public class CalcitePartialResultOnMappingConflictIT extends PPLIntegTestCase { + + private static final String KEYWORD_INDEX = "partial_conflict_keyword"; + private static final String TEXT_INDEX = "partial_conflict_text"; + private static final String PATTERN = "partial_conflict_*"; + + @Override + public void init() throws Exception { + super.init(); + enableCalcite(); + createTestIndices(); + } + + @After + public void cleanup() throws IOException { + setPartialResult(false); + setPitContextLimit(null); + } + + private void createTestIndices() throws IOException { + // keyword index: env is aggregatable. Two shards so a scan needs 2 PIT contexts. + if (!isIndexExist(client(), KEYWORD_INDEX)) { + String mapping = + "{\"settings\":{\"index\":{\"number_of_shards\":2,\"number_of_replicas\":0}}," + + "\"mappings\":{\"properties\":{\"env\":{\"type\":\"keyword\"}}}}"; + createIndexByRestClient(client(), KEYWORD_INDEX, mapping); + Request bulk = new Request("POST", "/" + KEYWORD_INDEX + "/_bulk?refresh=true"); + bulk.setJsonEntity( + "{\"index\":{}}\n{\"env\":\"prod\"}\n" + + "{\"index\":{}}\n{\"env\":\"prod\"}\n" + + "{\"index\":{}}\n{\"env\":\"dev\"}\n"); + performRequest(client(), bulk); + } + // text index (no .keyword sub-field): env is NOT aggregatable -> forces the conflict collapse. + if (!isIndexExist(client(), TEXT_INDEX)) { + String mapping = + "{\"settings\":{\"index\":{\"number_of_shards\":2,\"number_of_replicas\":0}}," + + "\"mappings\":{\"properties\":{\"env\":{\"type\":\"text\"}}}}"; + createIndexByRestClient(client(), TEXT_INDEX, mapping); + Request bulk = new Request("POST", "/" + TEXT_INDEX + "/_bulk?refresh=true"); + bulk.setJsonEntity( + "{\"index\":{}}\n{\"env\":\"prod\"}\n" + "{\"index\":{}}\n{\"env\":\"qa\"}\n"); + performRequest(client(), bulk); + } + } + + @Test + public void partialResultOffFailsWhenPitExhausted() throws IOException { + setPartialResult(false); + // Below the shard count of the pattern (4 shards) so the forced scan cannot open its PITs. + setPitContextLimit("1"); + ResponseException e = + assertThrows( + ResponseException.class, + () -> executeQuery(String.format("source=%s | stats count() by env", PATTERN))); + assertTrue(e.getResponse().getStatusLine().getStatusCode() >= 400); + } + + @Test + public void partialResultOnReturnsKeywordSubsetWithWarning() throws IOException { + setPartialResult(true); + // Even with the PIT budget crippled, partial mode pushes the aggregation down (size=0), so no + // PIT is opened and the query succeeds over the aggregatable keyword index only. + setPitContextLimit("1"); + JSONObject result = + executeQuery(String.format("source=%s | stats count() by env | sort env", PATTERN)); + + // Only the keyword index contributes: prod=2, dev=1. The text index (prod=1, qa=1) is excluded. + verifyDataRows(result, rows(1, "dev"), rows(2, "prod")); + + assertTrue("response should carry a warnings array", result.has("warnings")); + JSONArray warnings = result.getJSONArray("warnings"); + assertEquals(1, warnings.length()); + JSONObject warning = warnings.getJSONObject(0); + assertEquals("PARTIAL_RESULT", warning.getString("type")); + assertTrue( + "warning detail should name the excluded text index", + warning.getString("detail").contains(TEXT_INDEX)); + } + + @Test + public void partialResultOffOmitsWarningWhenNoConflict() throws IOException { + setPartialResult(true); + setPitContextLimit(null); + // A single-index aggregatable query has no conflict, so it pushes down normally and no warning + // is attached even with partial mode enabled. + JSONObject result = + executeQuery(String.format("source=%s | stats count() by env | sort env", KEYWORD_INDEX)); + verifyDataRows(result, rows(1, "dev"), rows(2, "prod")); + assertTrue("no warning expected on a clean aggregation", !result.has("warnings")); + } + + private void setPartialResult(boolean enabled) throws IOException { + updateClusterSettings( + new ClusterSetting( + "persistent", + Settings.Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT.getKeyValue(), + Boolean.toString(enabled))); + } + + private void setPitContextLimit(String value) throws IOException { + updateClusterSettings(new ClusterSetting("transient", "search.max_open_pit_context", value)); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/AggregateIndexScanRule.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/AggregateIndexScanRule.java index bd9f17abd2a..f714bb4ec44 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/AggregateIndexScanRule.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/AggregateIndexScanRule.java @@ -98,6 +98,12 @@ protected void apply( @Nullable LogicalProject project, CalciteLogicalIndexScan scan) { AbstractRelNode newRelNode = scan.pushDownAggregate(aggregate, project); + if (newRelNode == null) { + // Normal pushdown failed (e.g. a text/keyword mapping conflict across a wildcard pattern that + // would otherwise fall back to a per-shard scan and exhaust PIT contexts). If partial results + // are enabled, push the aggregation over the aggregatable index subset instead and warn. + newRelNode = scan.tryPartialResultAggregate(aggregate, project); + } if (newRelNode != null) { call.transformTo(newRelNode); PlanUtils.tryPruneRelNodes(call); diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java index b596c7bc47a..2e82cfc6428 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java @@ -187,6 +187,13 @@ public class OpenSearchSettings extends Settings { Setting.Property.NodeScope, Setting.Property.Dynamic); + public static final Setting CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT_SETTING = + Setting.boolSetting( + Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT.getKeyValue(), + false, + Setting.Property.NodeScope, + Setting.Property.Dynamic); + public static final Setting QUERY_MEMORY_LIMIT_SETTING = Setting.memorySizeSetting( Key.QUERY_MEMORY_LIMIT.getKeyValue(), @@ -476,6 +483,12 @@ public OpenSearchSettings(ClusterSettings clusterSettings) { Key.CALCITE_SUPPORT_ALL_JOIN_TYPES, CALCITE_SUPPORT_ALL_JOIN_TYPES_SETTING, new Updater(Key.CALCITE_SUPPORT_ALL_JOIN_TYPES)); + register( + settingBuilder, + clusterSettings, + Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT, + CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT_SETTING, + new Updater(Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT)); register( settingBuilder, clusterSettings, @@ -674,6 +687,7 @@ public static List> pluginSettings() { .add(CALCITE_PUSHDOWN_ENABLED_SETTING) .add(CALCITE_PUSHDOWN_ROWCOUNT_ESTIMATION_FACTOR_SETTING) .add(CALCITE_SUPPORT_ALL_JOIN_TYPES_SETTING) + .add(CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT_SETTING) .add(DEFAULT_PATTERN_METHOD_SETTING) .add(DEFAULT_PATTERN_MODE_SETTING) .add(DEFAULT_PATTERN_MAX_SAMPLE_COUNT_SETTING) diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index 2017437e7bd..c99169d63b7 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -42,6 +42,7 @@ import org.apache.logging.log4j.Logger; import org.opensearch.search.aggregations.AggregationBuilder; import org.opensearch.sql.ast.tree.HighlightConfig; +import org.opensearch.sql.calcite.CalcitePlanContext; import org.opensearch.sql.calcite.plan.HighlightPushDown; import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory; import org.opensearch.sql.calcite.utils.PPLHintUtils; @@ -51,6 +52,7 @@ import org.opensearch.sql.expression.HighlightExpression; import org.opensearch.sql.opensearch.data.type.OpenSearchDataType; import org.opensearch.sql.opensearch.data.type.OpenSearchTextType; +import org.opensearch.sql.opensearch.mapping.IndexMapping; import org.opensearch.sql.opensearch.planner.rules.OpenSearchIndexRules; import org.opensearch.sql.opensearch.request.AggregateAnalyzer; import org.opensearch.sql.opensearch.request.PredicateAnalyzer; @@ -431,6 +433,150 @@ public AbstractRelNode pushDownAggregate(Aggregate aggregate, @Nullable Project return null; } + /** + * Partial-result fallback for a text/keyword mapping conflict. When a normal aggregate pushdown + * fails because the grouped field collapsed to text-without-keyword across a wildcard pattern (so + * it would otherwise fall back to a per-shard document scan that opens a PIT on every shard), and + * the partial-result setting is enabled, narrow the scan to the largest homogeneous subset of + * indices whose mapping of the grouped field is aggregatable, push the aggregation down over just + * that subset, and record a warning naming the excluded indices. + * + *

The result is partial — the excluded indices' documents are missing from the counts — + * so this only runs behind an explicit opt-in and always emits a warning through {@link + * CalcitePlanContext#addWarning}. Returns {@code null} (leaving the caller to fall back) when the + * setting is off, there is no conflict, only one index matches, or no clean subset exists. + */ + public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable Project project) { + if (!(Boolean) + osIndex + .getSettings() + .getSettingValue(Settings.Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT)) { + return null; + } + try { + List outputFields = aggregate.getRowType().getFieldNames(); + List bucketNames = outputFields.subList(0, aggregate.getGroupSet().cardinality()); + if (bucketNames.isEmpty()) { + return null; + } + // getIndexNames() returns the raw pattern (e.g. ["logs-*"]) for a wildcard, not the resolved + // concrete indices; getIndexMappings expands it server-side to a map keyed by concrete index + // name. So the "multiple indices" check must be on the resolved map, not the pattern array. + String[] indexExpression = osIndex.getIndexName().getIndexNames(); + Map mappings = osIndex.getClient().getIndexMappings(indexExpression); + if (mappings.size() < 2) { + return null; + } + + // Keep the largest homogeneous index group so the narrowed merge resolves to a single clean + // type: prefer bare keyword (merges to keyword), else text-with-keyword-subfield (merges to + // text-with-.keyword). A mix of the two is still a conflict, so we never keep both. + List keywordIndices = new ArrayList<>(); + List textKeywordIndices = new ArrayList<>(); + List excludedIndices = new ArrayList<>(); + for (Map.Entry entry : mappings.entrySet()) { + MappingResolution resolution = resolveBucketMappings(entry.getValue(), bucketNames); + switch (resolution) { + case KEYWORD -> keywordIndices.add(entry.getKey()); + case TEXT_WITH_KEYWORD -> textKeywordIndices.add(entry.getKey()); + default -> excludedIndices.add(entry.getKey()); + } + } + + List keptIndices; + if (keywordIndices.size() >= textKeywordIndices.size() && !keywordIndices.isEmpty()) { + keptIndices = keywordIndices; + excludedIndices.addAll(textKeywordIndices); + } else if (!textKeywordIndices.isEmpty()) { + keptIndices = textKeywordIndices; + excludedIndices.addAll(keywordIndices); + } else { + return null; // no aggregatable subset -> partial mode can't help + } + if (excludedIndices.isEmpty()) { + return null; // homogeneous already; not our case (and pushdown wouldn't have failed) + } + + OpenSearchIndex narrowedIndex = + new OpenSearchIndex( + osIndex.getClient(), osIndex.getSettings(), String.join(",", keptIndices)); + CalciteLogicalIndexScan narrowedScan = + new CalciteLogicalIndexScan( + getCluster(), + traitSet, + hints, + table, + narrowedIndex, + getRowType(), + pushDownContext.cloneWithOsIndex(narrowedIndex)); + AbstractRelNode pushed = narrowedScan.pushDownAggregate(aggregate, project); + if (pushed == null) { + return null; // narrowed subset still can't push down -> fall back + } + + excludedIndices.sort(null); + CalcitePlanContext.addWarning( + new org.opensearch.sql.executor.Warning( + "PARTIAL_RESULT", + String.format( + "Results exclude %d of %d indices due to a text/keyword mapping conflict on" + + " %s.", + excludedIndices.size(), mappings.size(), bucketNames), + String.format( + "Field %s is mapped inconsistently across the queried indices, which prevents" + + " aggregation pushdown for the whole pattern. Excluded indices: %s. Align" + + " the mapping (reindex to keyword, or add a .keyword sub-field) to include" + + " them.", + bucketNames, excludedIndices))); + return pushed; + } catch (Exception e) { + if (LOG.isDebugEnabled()) { + LOG.debug("Cannot apply partial-result aggregate pushdown for {}", aggregate, e); + } + return null; + } + } + + /** How a single index maps the grouped field(s), for partial-result partitioning. */ + private enum MappingResolution { + KEYWORD, + TEXT_WITH_KEYWORD, + NOT_AGGREGATABLE + } + + /** + * Resolve how one index maps all grouped fields. Returns the weakest resolution across the + * fields: KEYWORD only if every field is bare keyword, TEXT_WITH_KEYWORD if every field is + * aggregatable but at least one relies on a .keyword sub-field, otherwise NOT_AGGREGATABLE. + */ + private static MappingResolution resolveBucketMappings( + IndexMapping mapping, List bucketNames) { + MappingResolution combined = MappingResolution.KEYWORD; + for (String field : bucketNames) { + OpenSearchDataType type = mapping.getFieldMappings().get(field); + if (type == null) { + return MappingResolution.NOT_AGGREGATABLE; // field absent here -> can't aggregate cleanly + } + OpenSearchDataType.MappingType mappingType = type.getMappingType(); + if (mappingType == OpenSearchDataType.MappingType.Keyword) { + continue; + } else if (isTextWithKeywordSubField(type)) { + combined = MappingResolution.TEXT_WITH_KEYWORD; + } else { + return MappingResolution.NOT_AGGREGATABLE; + } + } + return combined; + } + + private static boolean isTextWithKeywordSubField(OpenSearchDataType type) { + if (type instanceof OpenSearchTextType textType) { + return textType.getFields().values().stream() + .anyMatch(f -> f.getMappingType() == OpenSearchDataType.MappingType.Keyword); + } + return false; + } + public AbstractRelNode pushDownLimit(LogicalSort sort, Integer limit, Integer offset) { try { if (pushDownContext.isAggregatePushed()) { diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/context/PushDownContext.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/context/PushDownContext.java index a622f948efb..ba266a0f846 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/context/PushDownContext.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/context/PushDownContext.java @@ -56,6 +56,21 @@ public PushDownContext clone() { return newContext; } + /** + * Clone this context but rebind it to a different {@link OpenSearchIndex}. Used by partial-result + * pushdown, which re-targets an aggregation at a narrowed index subset while preserving the + * operations already pushed onto the current scan (e.g. a WHERE filter). The pushed operations + * are index-agnostic request-builder actions, so replaying them onto the new index is safe. + */ + public PushDownContext cloneWithOsIndex(OpenSearchIndex newOsIndex) { + PushDownContext newContext = new PushDownContext(newOsIndex); + for (PushDownOperation operation : this) { + newContext.add(operation); + } + newContext.aggSpec = aggSpec; + return newContext; + } + /** * Create a new {@link PushDownContext} without the collation action. * From e0407f34c2855e5824196a2ad9840c999d03436a Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Mon, 27 Jul 2026 12:36:16 -0700 Subject: [PATCH 3/8] Gate partial results on a warning-capable response format A partial result is only safe if the response can carry the warning that says so. Refuse partial mode (fall through to the normal path) when the requested format has no warnings channel -- CSV, RAW, and VIZ -- so a knowingly-partial result is never returned silently. - QueryContext carries a warnings-supported flag, set in TransportPPLQueryAction from the request format (true only for the JSON shape), read by the producer. - tryPartialResultAggregate bails when warnings are unsupported. - Integration test: partial on + format=csv still errors on PIT exhaustion rather than returning a silently-partial CSV. Signed-off-by: Kai Huang --- .../sql/common/utils/QueryContext.java | 22 +++++++++++++++++++ ...lcitePartialResultOnMappingConflictIT.java | 14 ++++++++++++ .../storage/scan/CalciteLogicalIndexScan.java | 6 +++++ .../transport/TransportPPLQueryAction.java | 14 ++++++++++++ 4 files changed, 56 insertions(+) diff --git a/common/src/main/java/org/opensearch/sql/common/utils/QueryContext.java b/common/src/main/java/org/opensearch/sql/common/utils/QueryContext.java index 4d4301df473..d2d35756df5 100644 --- a/common/src/main/java/org/opensearch/sql/common/utils/QueryContext.java +++ b/common/src/main/java/org/opensearch/sql/common/utils/QueryContext.java @@ -22,6 +22,8 @@ public class QueryContext { private static final String PROFILE_KEY = "profile"; + private static final String WARNINGS_SUPPORTED_KEY = "warnings_supported"; + /** * Generates a random UUID and adds to the {@link ThreadContext} as the request id. * @@ -84,4 +86,24 @@ public static void setProfile(boolean profileEnabled) { public static boolean isProfileEnabled() { return Boolean.parseBoolean(ThreadContext.get(PROFILE_KEY)); } + + /** + * Record whether the requested response format can surface non-fatal warnings. Features that + * return a knowingly-partial result gate on this so they never silently drop data into a format + * (CSV/RAW) that has no warning channel. + * + * @param supported whether the response format carries a warnings channel + */ + public static void setWarningsSupported(boolean supported) { + ThreadContext.put(WARNINGS_SUPPORTED_KEY, Boolean.toString(supported)); + } + + /** + * @return true if the response format for the current request can surface warnings. Defaults to + * false when unset, so a caller that never declared support cannot get a silent partial + * result. + */ + public static boolean isWarningsSupported() { + return Boolean.parseBoolean(ThreadContext.get(WARNINGS_SUPPORTED_KEY)); + } } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java index 484790bda12..e6ac2354947 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java @@ -126,6 +126,20 @@ public void partialResultOffOmitsWarningWhenNoConflict() throws IOException { assertTrue("no warning expected on a clean aggregation", !result.has("warnings")); } + @Test + public void partialResultRefusedForCsvFormat() throws IOException { + setPartialResult(true); + setPitContextLimit("1"); + // CSV has no warnings channel, so partial mode must NOT silently drop the text index. It falls + // through to the normal path, which still exhausts PIT contexts and errors. + ResponseException e = + assertThrows( + ResponseException.class, + () -> + executeCsvQuery(String.format("source=%s | stats count() by env", PATTERN), false)); + assertTrue(e.getResponse().getStatusLine().getStatusCode() >= 400); + } + private void setPartialResult(boolean enabled) throws IOException { updateClusterSettings( new ClusterSetting( diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index c99169d63b7..a33fb07a211 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -47,6 +47,7 @@ import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory; import org.opensearch.sql.calcite.utils.PPLHintUtils; import org.opensearch.sql.common.setting.Settings; +import org.opensearch.sql.common.utils.QueryContext; import org.opensearch.sql.data.type.ExprCoreType; import org.opensearch.sql.data.type.ExprType; import org.opensearch.sql.expression.HighlightExpression; @@ -453,6 +454,11 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable .getSettingValue(Settings.Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT)) { return null; } + // A partial result is only safe if the response can surface the warning that says so. Refuse + // (fall through to the normal path) for formats without a warnings channel, e.g. CSV/RAW/VIZ. + if (!QueryContext.isWarningsSupported()) { + return null; + } try { List outputFields = aggregate.getRowType().getFieldNames(); List bucketNames = outputFields.subList(0, aggregate.getGroupSet().cardinality()); diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java b/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java index 0cbb75d9e1c..109df61c9f5 100644 --- a/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java +++ b/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java @@ -174,6 +174,9 @@ protected void doExecute( // in order to use PPL service, we need to convert TransportPPLQueryRequest to PPLQueryRequest PPLQueryRequest transformedRequest = transportRequest.toPPLQueryRequest(); QueryContext.setProfile(transformedRequest.profile()); + // Only the JSON response shape carries a warnings channel. Features that return a + // knowingly-partial result gate on this so they never silently drop data into CSV/RAW/VIZ. + QueryContext.setWarningsSupported(warningsSupported(transformedRequest)); ActionListener clearingListener = wrapWithProfilingClear(listener); // Route to analytics engine for non-Lucene (e.g., Parquet-backed) indices. @@ -349,6 +352,17 @@ private Format format(PPLQueryRequest pplRequest) { } } + /** + * Whether the requested response format carries a warnings channel. Only the JSON shape (built by + * {@code SimpleJsonResponseFormatter} -- the fallback for anything that is not CSV/RAW/VIZ) emits + * warnings; the others have no slot for them. Mirrors the format branching in {@link + * #createListener}. + */ + private boolean warningsSupported(PPLQueryRequest pplRequest) { + Format format = format(pplRequest); + return !(format.equals(Format.CSV) || format.equals(Format.RAW) || format.equals(Format.VIZ)); + } + private ActionListener wrapWithProfilingClear( ActionListener delegate) { return new ActionListener<>() { From f553b7727a9c923e9140168eb4afae45c62d9db8 Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Mon, 27 Jul 2026 13:32:20 -0700 Subject: [PATCH 4/8] Flatten per-index mappings when partitioning for partial results The partial-result partitioner looked up the grouped field in each index's raw field mappings by its dotted name. A nested/object field such as resource.attributes.applicationid is stored as an object tree, not a flat dotted key, so the lookup returned null, every index classified as NOT_AGGREGATABLE, and the producer bailed -- leaving the query to exhaust PIT contexts. This is the exact shape of the real observability field that motivated the feature. Flatten each index's field mappings with OpenSearchDataType.traverseAndFlatten (the same flattening the field-type resolver uses) before the lookup, so the dotted bucket name resolves. Add an integration test over a nested-field conflict pattern. Found by live testing the customer query; the flat-field integration test missed it. Signed-off-by: Kai Huang --- ...lcitePartialResultOnMappingConflictIT.java | 58 +++++++++++++++++++ .../storage/scan/CalciteLogicalIndexScan.java | 14 ++++- 2 files changed, 69 insertions(+), 3 deletions(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java index e6ac2354947..63f4a57162a 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java @@ -40,6 +40,10 @@ public class CalcitePartialResultOnMappingConflictIT extends PPLIntegTestCase { private static final String TEXT_INDEX = "partial_conflict_text"; private static final String PATTERN = "partial_conflict_*"; + private static final String NESTED_KEYWORD_INDEX = "partial_nested_keyword"; + private static final String NESTED_TEXT_INDEX = "partial_nested_text"; + private static final String NESTED_PATTERN = "partial_nested_*"; + @Override public void init() throws Exception { super.init(); @@ -78,6 +82,35 @@ private void createTestIndices() throws IOException { "{\"index\":{}}\n{\"env\":\"prod\"}\n" + "{\"index\":{}}\n{\"env\":\"qa\"}\n"); performRequest(client(), bulk); } + + // A nested/dotted field (resource.attributes.env) is stored as an object tree in the mapping, + // so the partitioning must flatten it to match the bucket field's dotted path. Mirrors the + // real observability shape (e.g. resource.attributes.applicationid). + if (!isIndexExist(client(), NESTED_KEYWORD_INDEX)) { + String mapping = + "{\"settings\":{\"index\":{\"number_of_shards\":2,\"number_of_replicas\":0}}," + + "\"mappings\":{\"properties\":{\"resource\":{\"properties\":{\"attributes\":" + + "{\"properties\":{\"env\":{\"type\":\"keyword\"}}}}}}}}"; + createIndexByRestClient(client(), NESTED_KEYWORD_INDEX, mapping); + Request bulk = new Request("POST", "/" + NESTED_KEYWORD_INDEX + "/_bulk?refresh=true"); + bulk.setJsonEntity( + "{\"index\":{}}\n{\"resource\":{\"attributes\":{\"env\":\"prod\"}}}\n" + + "{\"index\":{}}\n{\"resource\":{\"attributes\":{\"env\":\"prod\"}}}\n" + + "{\"index\":{}}\n{\"resource\":{\"attributes\":{\"env\":\"dev\"}}}\n"); + performRequest(client(), bulk); + } + if (!isIndexExist(client(), NESTED_TEXT_INDEX)) { + String mapping = + "{\"settings\":{\"index\":{\"number_of_shards\":2,\"number_of_replicas\":0}}," + + "\"mappings\":{\"properties\":{\"resource\":{\"properties\":{\"attributes\":" + + "{\"properties\":{\"env\":{\"type\":\"text\"}}}}}}}}"; + createIndexByRestClient(client(), NESTED_TEXT_INDEX, mapping); + Request bulk = new Request("POST", "/" + NESTED_TEXT_INDEX + "/_bulk?refresh=true"); + bulk.setJsonEntity( + "{\"index\":{}}\n{\"resource\":{\"attributes\":{\"env\":\"prod\"}}}\n" + + "{\"index\":{}}\n{\"resource\":{\"attributes\":{\"env\":\"qa\"}}}\n"); + performRequest(client(), bulk); + } } @Test @@ -126,6 +159,31 @@ public void partialResultOffOmitsWarningWhenNoConflict() throws IOException { assertTrue("no warning expected on a clean aggregation", !result.has("warnings")); } + @Test + public void partialResultOnHandlesNestedDottedField() throws IOException { + setPartialResult(true); + setPitContextLimit("1"); + // The grouped field is a nested/dotted path; the partitioning must flatten the mapping to find + // it. Only the keyword index contributes: prod=2, dev=1. + JSONObject result = + executeQuery( + String.format( + "source=%s | stats count() by resource.attributes.env | sort" + + " `resource.attributes.env`", + NESTED_PATTERN)); + verifyDataRows(result, rows(1, "dev"), rows(2, "prod")); + + assertTrue("response should carry a warnings array", result.has("warnings")); + JSONObject warning = result.getJSONArray("warnings").getJSONObject(0); + assertEquals("PARTIAL_RESULT", warning.getString("type")); + assertTrue( + "warning should name the dotted field", + warning.getString("detail").contains("resource.attributes.env")); + assertTrue( + "warning should name the excluded nested-text index", + warning.getString("detail").contains(NESTED_TEXT_INDEX)); + } + @Test public void partialResultRefusedForCsvFormat() throws IOException { setPartialResult(true); diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index a33fb07a211..ad06b13669b 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -481,7 +481,12 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable List textKeywordIndices = new ArrayList<>(); List excludedIndices = new ArrayList<>(); for (Map.Entry entry : mappings.entrySet()) { - MappingResolution resolution = resolveBucketMappings(entry.getValue(), bucketNames); + // Flatten the per-index mapping so nested object fields (e.g. a raw mapping tree + // resource -> attributes -> applicationid) are keyed by their dotted path, matching the + // bucket field name Calcite resolved. + Map flatMapping = + OpenSearchDataType.traverseAndFlatten(entry.getValue().getFieldMappings()); + MappingResolution resolution = resolveBucketMappings(flatMapping, bucketNames); switch (resolution) { case KEYWORD -> keywordIndices.add(entry.getKey()); case TEXT_WITH_KEYWORD -> textKeywordIndices.add(entry.getKey()); @@ -554,12 +559,15 @@ private enum MappingResolution { * Resolve how one index maps all grouped fields. Returns the weakest resolution across the * fields: KEYWORD only if every field is bare keyword, TEXT_WITH_KEYWORD if every field is * aggregatable but at least one relies on a .keyword sub-field, otherwise NOT_AGGREGATABLE. + * + * @param flatMapping the index's field types flattened to dotted paths (see {@link + * OpenSearchDataType#traverseAndFlatten}) */ private static MappingResolution resolveBucketMappings( - IndexMapping mapping, List bucketNames) { + Map flatMapping, List bucketNames) { MappingResolution combined = MappingResolution.KEYWORD; for (String field : bucketNames) { - OpenSearchDataType type = mapping.getFieldMappings().get(field); + OpenSearchDataType type = flatMapping.get(field); if (type == null) { return MappingResolution.NOT_AGGREGATABLE; // field absent here -> can't aggregate cleanly } From b475c5bfada3a6ef45b79f86aabc97170f1c7013 Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Mon, 27 Jul 2026 14:46:00 -0700 Subject: [PATCH 5/8] Refine partial-result index selection and warning wording Two refinements to the partial-result partitioner: - Pick the kept index group by a deterministic priority instead of a count-based majority: always keep the keyword group when any keyword index exists (the canonical aggregatable representation), fall back to the text-with-.keyword group only when there is no keyword index, and always exclude bare-text. The returned data no longer depends on how many indices of each type match, so a stray index can't flip which subset the user sees. - Correct the warning wording: the old remedy ("add a .keyword sub-field") was misleading when the excluded index already had one. Reword to say the aggregation ran over the largest consistently-mapped subset and to suggest aligning the mapping (e.g. keyword everywhere). Add an integration test where keyword is outnumbered 2:1 by text-with-.keyword indices and must still be the kept group. Signed-off-by: Kai Huang --- ...lcitePartialResultOnMappingConflictIT.java | 53 +++++++++++++++++++ .../storage/scan/CalciteLogicalIndexScan.java | 22 +++++--- 2 files changed, 67 insertions(+), 8 deletions(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java index 63f4a57162a..cfb57e84d71 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java @@ -44,6 +44,14 @@ public class CalcitePartialResultOnMappingConflictIT extends PPLIntegTestCase { private static final String NESTED_TEXT_INDEX = "partial_nested_text"; private static final String NESTED_PATTERN = "partial_nested_*"; + // Priority-ladder fixture: one keyword index vs two text-with-.keyword indices. Keyword is + // outnumbered, so a count-based majority would keep the text-with-.keyword group; the + // deterministic keyword-first rule must keep the single keyword index instead. + private static final String PRIORITY_KEYWORD_INDEX = "partial_priority_keyword"; + private static final String PRIORITY_TEXTKW_INDEX_1 = "partial_priority_textkw1"; + private static final String PRIORITY_TEXTKW_INDEX_2 = "partial_priority_textkw2"; + private static final String PRIORITY_PATTERN = "partial_priority_*"; + @Override public void init() throws Exception { super.init(); @@ -111,6 +119,31 @@ private void createTestIndices() throws IOException { + "{\"index\":{}}\n{\"resource\":{\"attributes\":{\"env\":\"qa\"}}}\n"); performRequest(client(), bulk); } + + // Priority ladder: 1 keyword index vs 2 text-with-.keyword indices (keyword outnumbered 2:1). + if (!isIndexExist(client(), PRIORITY_KEYWORD_INDEX)) { + String mapping = + "{\"settings\":{\"index\":{\"number_of_shards\":2,\"number_of_replicas\":0}}," + + "\"mappings\":{\"properties\":{\"env\":{\"type\":\"keyword\"}}}}"; + createIndexByRestClient(client(), PRIORITY_KEYWORD_INDEX, mapping); + Request bulk = new Request("POST", "/" + PRIORITY_KEYWORD_INDEX + "/_bulk?refresh=true"); + bulk.setJsonEntity( + "{\"index\":{}}\n{\"env\":\"prod\"}\n" + "{\"index\":{}}\n{\"env\":\"dev\"}\n"); + performRequest(client(), bulk); + } + String textKwMapping = + "{\"settings\":{\"index\":{\"number_of_shards\":2,\"number_of_replicas\":0}}," + + "\"mappings\":{\"properties\":{\"env\":{\"type\":\"text\",\"fields\":" + + "{\"keyword\":{\"type\":\"keyword\",\"ignore_above\":256}}}}}}"; + for (String idx : new String[] {PRIORITY_TEXTKW_INDEX_1, PRIORITY_TEXTKW_INDEX_2}) { + if (!isIndexExist(client(), idx)) { + createIndexByRestClient(client(), idx, textKwMapping); + Request bulk = new Request("POST", "/" + idx + "/_bulk?refresh=true"); + bulk.setJsonEntity( + "{\"index\":{}}\n{\"env\":\"prod\"}\n" + "{\"index\":{}}\n{\"env\":\"stage\"}\n"); + performRequest(client(), bulk); + } + } } @Test @@ -184,6 +217,26 @@ public void partialResultOnHandlesNestedDottedField() throws IOException { warning.getString("detail").contains(NESTED_TEXT_INDEX)); } + @Test + public void partialResultKeepsKeywordGroupEvenWhenOutnumbered() throws IOException { + setPartialResult(true); + setPitContextLimit("1"); + // Keyword is outnumbered 2:1 by text-with-.keyword indices. The deterministic keyword-first + // rule keeps the single keyword index (prod:1, dev:1) and excludes both text-with-.keyword + // indices -- a count-based majority would have kept the text group instead. + JSONObject result = + executeQuery( + String.format("source=%s | stats count() by env | sort env", PRIORITY_PATTERN)); + verifyDataRows(result, rows(1, "dev"), rows(1, "prod")); + + JSONObject warning = result.getJSONArray("warnings").getJSONObject(0); + assertEquals("PARTIAL_RESULT", warning.getString("type")); + assertTrue( + "both text-with-keyword indices should be excluded", + warning.getString("detail").contains(PRIORITY_TEXTKW_INDEX_1) + && warning.getString("detail").contains(PRIORITY_TEXTKW_INDEX_2)); + } + @Test public void partialResultRefusedForCsvFormat() throws IOException { setPartialResult(true); diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index ad06b13669b..8e4d30ef2d7 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -474,9 +474,9 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable return null; } - // Keep the largest homogeneous index group so the narrowed merge resolves to a single clean - // type: prefer bare keyword (merges to keyword), else text-with-keyword-subfield (merges to - // text-with-.keyword). A mix of the two is still a conflict, so we never keep both. + // Classify each index by how it maps the grouped field. The kept group must be a single + // homogeneous representation: a mix of keyword and text-with-.keyword is still a text/keyword + // conflict that would re-collapse and fail pushdown, so we never keep both. List keywordIndices = new ArrayList<>(); List textKeywordIndices = new ArrayList<>(); List excludedIndices = new ArrayList<>(); @@ -494,13 +494,18 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable } } + // Deterministic priority (not a count-based majority): always keep the keyword group when one + // exists, since keyword is the canonical aggregatable representation and the result must not + // depend on how many stray indices of another type happen to match. Fall back to the + // text-with-.keyword group only when there is no keyword index at all. Bare-text indices are + // never aggregatable and are always excluded. Recovering an excluded but aggregatable group + // (text-with-.keyword alongside keyword) is the job of the split (2a), not partial mode. List keptIndices; - if (keywordIndices.size() >= textKeywordIndices.size() && !keywordIndices.isEmpty()) { + if (!keywordIndices.isEmpty()) { keptIndices = keywordIndices; excludedIndices.addAll(textKeywordIndices); } else if (!textKeywordIndices.isEmpty()) { keptIndices = textKeywordIndices; - excludedIndices.addAll(keywordIndices); } else { return null; // no aggregatable subset -> partial mode can't help } @@ -535,9 +540,10 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable excludedIndices.size(), mappings.size(), bucketNames), String.format( "Field %s is mapped inconsistently across the queried indices, which prevents" - + " aggregation pushdown for the whole pattern. Excluded indices: %s. Align" - + " the mapping (reindex to keyword, or add a .keyword sub-field) to include" - + " them.", + + " aggregation pushdown for the whole pattern. The aggregation ran over the" + + " largest consistently-mapped subset; excluded indices: %s. Align the" + + " field's mapping across all indices (e.g. map it as keyword everywhere) to" + + " include them.", bucketNames, excludedIndices))); return pushed; } catch (Exception e) { From dad3bb3abc65852547773d22cf03442cdfa7b6c3 Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Mon, 27 Jul 2026 14:53:11 -0700 Subject: [PATCH 6/8] Truncate the excluded-index list in the partial-result warning A wide observability pattern can exclude hundreds of indices; listing them all verbatim in the warning detail produces an unreadable multi-kilobyte message. The exact count is already in the warning's summary, so spell out at most a few excluded index names in the detail and summarize the rest as "... and N more". Add an integration test with a large excluded set asserting the detail is truncated while the summary still reports the full count. Signed-off-by: Kai Huang --- ...lcitePartialResultOnMappingConflictIT.java | 49 +++++++++++++++++++ .../storage/scan/CalciteLogicalIndexScan.java | 22 ++++++++- 2 files changed, 70 insertions(+), 1 deletion(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java index cfb57e84d71..e0b5257a170 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java @@ -44,6 +44,13 @@ public class CalcitePartialResultOnMappingConflictIT extends PPLIntegTestCase { private static final String NESTED_TEXT_INDEX = "partial_nested_text"; private static final String NESTED_PATTERN = "partial_nested_*"; + // Truncation fixture: 1 keyword index + 8 bare-text indices, so the excluded list exceeds the + // warning's spell-out cap and must be summarized as "... and N more". + private static final String MANY_KEYWORD_INDEX = "partial_many_keyword"; + private static final String MANY_TEXT_PREFIX = "partial_many_text"; + private static final String MANY_PATTERN = "partial_many_*"; + private static final int MANY_TEXT_COUNT = 8; + // Priority-ladder fixture: one keyword index vs two text-with-.keyword indices. Keyword is // outnumbered, so a count-based majority would keep the text-with-.keyword group; the // deterministic keyword-first rule must keep the single keyword index instead. @@ -144,6 +151,29 @@ private void createTestIndices() throws IOException { performRequest(client(), bulk); } } + + // Truncation: 1 keyword + many bare-text indices, so the excluded list exceeds the warning cap. + if (!isIndexExist(client(), MANY_KEYWORD_INDEX)) { + String mapping = + "{\"settings\":{\"index\":{\"number_of_shards\":2,\"number_of_replicas\":0}}," + + "\"mappings\":{\"properties\":{\"env\":{\"type\":\"keyword\"}}}}"; + createIndexByRestClient(client(), MANY_KEYWORD_INDEX, mapping); + Request bulk = new Request("POST", "/" + MANY_KEYWORD_INDEX + "/_bulk?refresh=true"); + bulk.setJsonEntity("{\"index\":{}}\n{\"env\":\"prod\"}\n"); + performRequest(client(), bulk); + } + String bareTextMapping = + "{\"settings\":{\"index\":{\"number_of_shards\":2,\"number_of_replicas\":0}}," + + "\"mappings\":{\"properties\":{\"env\":{\"type\":\"text\"}}}}"; + for (int i = 1; i <= MANY_TEXT_COUNT; i++) { + String idx = MANY_TEXT_PREFIX + i; + if (!isIndexExist(client(), idx)) { + createIndexByRestClient(client(), idx, bareTextMapping); + Request bulk = new Request("POST", "/" + idx + "/_bulk?refresh=true"); + bulk.setJsonEntity("{\"index\":{}}\n{\"env\":\"prod\"}\n"); + performRequest(client(), bulk); + } + } } @Test @@ -237,6 +267,25 @@ public void partialResultKeepsKeywordGroupEvenWhenOutnumbered() throws IOExcepti && warning.getString("detail").contains(PRIORITY_TEXTKW_INDEX_2)); } + @Test + public void partialResultWarningTruncatesLargeExcludedList() throws IOException { + setPartialResult(true); + setPitContextLimit("1"); + JSONObject result = + executeQuery(String.format("source=%s | stats count() by env", MANY_PATTERN)); + + JSONObject warning = result.getJSONArray("warnings").getJSONObject(0); + // 8 bare-text indices excluded; the message reports the exact count... + assertTrue( + "message should report the full excluded count", + warning.getString("message").contains("8 of 9")); + // ...but the detail spells out only a few and summarizes the rest. + String detail = warning.getString("detail"); + assertTrue("detail should summarize the remainder", detail.contains("and 3 more")); + assertTrue( + "detail should not list every excluded index", !detail.contains(MANY_TEXT_PREFIX + "8")); + } + @Test public void partialResultRefusedForCsvFormat() throws IOException { setPartialResult(true); diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index 8e4d30ef2d7..7ac3792592b 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -544,7 +544,7 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable + " largest consistently-mapped subset; excluded indices: %s. Align the" + " field's mapping across all indices (e.g. map it as keyword everywhere) to" + " include them.", - bucketNames, excludedIndices))); + bucketNames, formatIndexList(excludedIndices, MAX_EXCLUDED_INDICES_IN_WARNING)))); return pushed; } catch (Exception e) { if (LOG.isDebugEnabled()) { @@ -561,6 +561,26 @@ private enum MappingResolution { NOT_AGGREGATABLE } + /** + * Cap on how many excluded index names to spell out in the partial-result warning. A wide + * observability pattern can exclude hundreds of indices; the exact count is already in the + * warning's message, so the detail lists only a few examples and summarizes the rest. + */ + private static final int MAX_EXCLUDED_INDICES_IN_WARNING = 5; + + /** + * Format a (sorted) index-name list for a warning: spell out up to {@code limit} names, then + * summarize any remainder as "and N more" so the message stays readable when the excluded set is + * large. + */ + private static String formatIndexList(List indices, int limit) { + if (indices.size() <= limit) { + return indices.toString(); + } + List shown = indices.subList(0, limit); + return String.format("[%s, ... and %d more]", String.join(", ", shown), indices.size() - limit); + } + /** * Resolve how one index maps all grouped fields. Returns the weakest resolution across the * fields: KEYWORD only if every field is bare keyword, TEXT_WITH_KEYWORD if every field is From 078c949e905860698ae1543183d905423e3a36c5 Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Mon, 27 Jul 2026 14:58:42 -0700 Subject: [PATCH 7/8] Fix partial-result warning wording to reflect the pushdown criterion The detail said the aggregation ran over the 'largest consistently-mapped subset', leftover from when the kept group was chosen by index count. Selection is now by whether the field is aggregatable there (keyword-first), not size, so reword to 'ran only over the indices where the field is aggregatable'. Signed-off-by: Kai Huang --- .../opensearch/storage/scan/CalciteLogicalIndexScan.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index 7ac3792592b..da1a23a31e7 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -540,10 +540,10 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable excludedIndices.size(), mappings.size(), bucketNames), String.format( "Field %s is mapped inconsistently across the queried indices, which prevents" - + " aggregation pushdown for the whole pattern. The aggregation ran over the" - + " largest consistently-mapped subset; excluded indices: %s. Align the" - + " field's mapping across all indices (e.g. map it as keyword everywhere) to" - + " include them.", + + " aggregation pushdown for the whole pattern. The aggregation ran only over" + + " the indices where the field is aggregatable; excluded indices: %s. Align" + + " the field's mapping across all indices (e.g. map it as keyword" + + " everywhere) to include them.", bucketNames, formatIndexList(excludedIndices, MAX_EXCLUDED_INDICES_IN_WARNING)))); return pushed; } catch (Exception e) { From 0419c94bd4e249dcdbac10c7829c4bd6151ec4ae Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Mon, 27 Jul 2026 16:06:27 -0700 Subject: [PATCH 8/8] Extract partial-result partitioning into its own class with unit tests Harden the partial-result fallback from POC-shaped code into a testable unit: - Move the classify/partition/priority/warning logic out of the 600-line CalciteLogicalIndexScan into a dedicated PartialResultAggregatePushdown. The scan's tryPartialResultAggregate keeps only the plan-time wiring (settings gate, mapping lookup, narrowed-scan construction, warning emission) and delegates the decision to PartialResultAggregatePushdown.plan(...). - Make the PARTIAL_RESULT warning type a shared constant (Warning.TYPE_PARTIAL_RESULT) instead of a literal, since consumers such as OpenSearch Dashboards branch on it -- a cross-surface contract. - Add a field-map constructor to IndexMapping for testability. - Add unit tests covering classification (keyword / text+keyword / bare-text / absent), multi-field weakest-resolution, the keyword-first priority ladder (including when keyword is outnumbered), null/no-op cases, excluded-list sorting, and warning-list truncation. No behavior change; the integration tests are unchanged and still pass. Signed-off-by: Kai Huang --- .../org/opensearch/sql/executor/Warning.java | 11 +- .../sql/opensearch/mapping/IndexMapping.java | 5 + .../storage/scan/CalciteLogicalIndexScan.java | 153 ++---------- .../scan/PartialResultAggregatePushdown.java | 176 ++++++++++++++ .../PartialResultAggregatePushdownTest.java | 220 ++++++++++++++++++ 5 files changed, 430 insertions(+), 135 deletions(-) create mode 100644 opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/PartialResultAggregatePushdown.java create mode 100644 opensearch/src/test/java/org/opensearch/sql/opensearch/storage/scan/PartialResultAggregatePushdownTest.java diff --git a/core/src/main/java/org/opensearch/sql/executor/Warning.java b/core/src/main/java/org/opensearch/sql/executor/Warning.java index 99ca7f53f58..daa64c2f118 100644 --- a/core/src/main/java/org/opensearch/sql/executor/Warning.java +++ b/core/src/main/java/org/opensearch/sql/executor/Warning.java @@ -14,7 +14,16 @@ */ @Data public class Warning { - /** Machine-readable category, e.g. {@code PARTIAL_RESULT}. */ + + /** + * The result is complete for the indices it covers but omits one or more indices that could not + * be served (e.g. a mapping conflict that prevents aggregation pushdown). This is a cross-surface + * contract: consumers such as OpenSearch Dashboards branch on this {@code type} value, so it must + * not change without coordinating those consumers. + */ + public static final String TYPE_PARTIAL_RESULT = "PARTIAL_RESULT"; + + /** Machine-readable category, e.g. {@link #TYPE_PARTIAL_RESULT}. */ private final String type; /** Short human-readable summary. */ diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/mapping/IndexMapping.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/mapping/IndexMapping.java index 87aa9d93dda..6e49081acd7 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/mapping/IndexMapping.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/mapping/IndexMapping.java @@ -33,6 +33,11 @@ public IndexMapping(MappingMetadata metaData) { (Map) metaData.getSourceAsMap().getOrDefault("properties", null)); } + /** Construct directly from parsed field mappings. Visible for testing. */ + public IndexMapping(Map fieldMappings) { + this.fieldMappings = fieldMappings; + } + /** * How many fields in the index (after flatten). * diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java index da1a23a31e7..a93b8bd980b 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java @@ -435,17 +435,19 @@ public AbstractRelNode pushDownAggregate(Aggregate aggregate, @Nullable Project } /** - * Partial-result fallback for a text/keyword mapping conflict. When a normal aggregate pushdown - * fails because the grouped field collapsed to text-without-keyword across a wildcard pattern (so - * it would otherwise fall back to a per-shard document scan that opens a PIT on every shard), and - * the partial-result setting is enabled, narrow the scan to the largest homogeneous subset of - * indices whose mapping of the grouped field is aggregatable, push the aggregation down over just - * that subset, and record a warning naming the excluded indices. + * Partial-result fallback for a text/keyword mapping conflict on the aggregation's group key. + * When normal aggregate pushdown fails because the field collapsed to text-without-keyword across + * a wildcard pattern (so it would otherwise fall back to a per-shard document scan that opens a + * PIT on every shard), and the partial-result setting is enabled, narrow the scan to the subset + * of indices where the field is aggregatable, push the aggregation down over just that subset, + * and record a warning naming the excluded indices. * *

The result is partial — the excluded indices' documents are missing from the counts — - * so this only runs behind an explicit opt-in and always emits a warning through {@link - * CalcitePlanContext#addWarning}. Returns {@code null} (leaving the caller to fall back) when the - * setting is off, there is no conflict, only one index matches, or no clean subset exists. + * so this only runs behind an explicit opt-in and only when the response format can surface the + * warning ({@link QueryContext#isWarningsSupported}). The partitioning decision lives in {@link + * PartialResultAggregatePushdown}; this method owns the plan-time wiring (settings gate, mapping + * lookup, narrowed-scan construction, warning emission). Returns {@code null} — leaving the + * caller to fall back — whenever partial mode does not apply. */ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable Project project) { if (!(Boolean) @@ -462,60 +464,20 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable try { List outputFields = aggregate.getRowType().getFieldNames(); List bucketNames = outputFields.subList(0, aggregate.getGroupSet().cardinality()); - if (bucketNames.isEmpty()) { - return null; - } // getIndexNames() returns the raw pattern (e.g. ["logs-*"]) for a wildcard, not the resolved // concrete indices; getIndexMappings expands it server-side to a map keyed by concrete index - // name. So the "multiple indices" check must be on the resolved map, not the pattern array. - String[] indexExpression = osIndex.getIndexName().getIndexNames(); - Map mappings = osIndex.getClient().getIndexMappings(indexExpression); - if (mappings.size() < 2) { + // name, which is what the partitioning classifies. + Map mappings = + osIndex.getClient().getIndexMappings(osIndex.getIndexName().getIndexNames()); + PartialResultAggregatePushdown.Plan plan = + PartialResultAggregatePushdown.plan(bucketNames, mappings); + if (plan == null) { return null; } - // Classify each index by how it maps the grouped field. The kept group must be a single - // homogeneous representation: a mix of keyword and text-with-.keyword is still a text/keyword - // conflict that would re-collapse and fail pushdown, so we never keep both. - List keywordIndices = new ArrayList<>(); - List textKeywordIndices = new ArrayList<>(); - List excludedIndices = new ArrayList<>(); - for (Map.Entry entry : mappings.entrySet()) { - // Flatten the per-index mapping so nested object fields (e.g. a raw mapping tree - // resource -> attributes -> applicationid) are keyed by their dotted path, matching the - // bucket field name Calcite resolved. - Map flatMapping = - OpenSearchDataType.traverseAndFlatten(entry.getValue().getFieldMappings()); - MappingResolution resolution = resolveBucketMappings(flatMapping, bucketNames); - switch (resolution) { - case KEYWORD -> keywordIndices.add(entry.getKey()); - case TEXT_WITH_KEYWORD -> textKeywordIndices.add(entry.getKey()); - default -> excludedIndices.add(entry.getKey()); - } - } - - // Deterministic priority (not a count-based majority): always keep the keyword group when one - // exists, since keyword is the canonical aggregatable representation and the result must not - // depend on how many stray indices of another type happen to match. Fall back to the - // text-with-.keyword group only when there is no keyword index at all. Bare-text indices are - // never aggregatable and are always excluded. Recovering an excluded but aggregatable group - // (text-with-.keyword alongside keyword) is the job of the split (2a), not partial mode. - List keptIndices; - if (!keywordIndices.isEmpty()) { - keptIndices = keywordIndices; - excludedIndices.addAll(textKeywordIndices); - } else if (!textKeywordIndices.isEmpty()) { - keptIndices = textKeywordIndices; - } else { - return null; // no aggregatable subset -> partial mode can't help - } - if (excludedIndices.isEmpty()) { - return null; // homogeneous already; not our case (and pushdown wouldn't have failed) - } - OpenSearchIndex narrowedIndex = new OpenSearchIndex( - osIndex.getClient(), osIndex.getSettings(), String.join(",", keptIndices)); + osIndex.getClient(), osIndex.getSettings(), String.join(",", plan.keptIndices())); CalciteLogicalIndexScan narrowedScan = new CalciteLogicalIndexScan( getCluster(), @@ -530,21 +492,7 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable return null; // narrowed subset still can't push down -> fall back } - excludedIndices.sort(null); - CalcitePlanContext.addWarning( - new org.opensearch.sql.executor.Warning( - "PARTIAL_RESULT", - String.format( - "Results exclude %d of %d indices due to a text/keyword mapping conflict on" - + " %s.", - excludedIndices.size(), mappings.size(), bucketNames), - String.format( - "Field %s is mapped inconsistently across the queried indices, which prevents" - + " aggregation pushdown for the whole pattern. The aggregation ran only over" - + " the indices where the field is aggregatable; excluded indices: %s. Align" - + " the field's mapping across all indices (e.g. map it as keyword" - + " everywhere) to include them.", - bucketNames, formatIndexList(excludedIndices, MAX_EXCLUDED_INDICES_IN_WARNING)))); + CalcitePlanContext.addWarning(plan.warning()); return pushed; } catch (Exception e) { if (LOG.isDebugEnabled()) { @@ -554,69 +502,6 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable } } - /** How a single index maps the grouped field(s), for partial-result partitioning. */ - private enum MappingResolution { - KEYWORD, - TEXT_WITH_KEYWORD, - NOT_AGGREGATABLE - } - - /** - * Cap on how many excluded index names to spell out in the partial-result warning. A wide - * observability pattern can exclude hundreds of indices; the exact count is already in the - * warning's message, so the detail lists only a few examples and summarizes the rest. - */ - private static final int MAX_EXCLUDED_INDICES_IN_WARNING = 5; - - /** - * Format a (sorted) index-name list for a warning: spell out up to {@code limit} names, then - * summarize any remainder as "and N more" so the message stays readable when the excluded set is - * large. - */ - private static String formatIndexList(List indices, int limit) { - if (indices.size() <= limit) { - return indices.toString(); - } - List shown = indices.subList(0, limit); - return String.format("[%s, ... and %d more]", String.join(", ", shown), indices.size() - limit); - } - - /** - * Resolve how one index maps all grouped fields. Returns the weakest resolution across the - * fields: KEYWORD only if every field is bare keyword, TEXT_WITH_KEYWORD if every field is - * aggregatable but at least one relies on a .keyword sub-field, otherwise NOT_AGGREGATABLE. - * - * @param flatMapping the index's field types flattened to dotted paths (see {@link - * OpenSearchDataType#traverseAndFlatten}) - */ - private static MappingResolution resolveBucketMappings( - Map flatMapping, List bucketNames) { - MappingResolution combined = MappingResolution.KEYWORD; - for (String field : bucketNames) { - OpenSearchDataType type = flatMapping.get(field); - if (type == null) { - return MappingResolution.NOT_AGGREGATABLE; // field absent here -> can't aggregate cleanly - } - OpenSearchDataType.MappingType mappingType = type.getMappingType(); - if (mappingType == OpenSearchDataType.MappingType.Keyword) { - continue; - } else if (isTextWithKeywordSubField(type)) { - combined = MappingResolution.TEXT_WITH_KEYWORD; - } else { - return MappingResolution.NOT_AGGREGATABLE; - } - } - return combined; - } - - private static boolean isTextWithKeywordSubField(OpenSearchDataType type) { - if (type instanceof OpenSearchTextType textType) { - return textType.getFields().values().stream() - .anyMatch(f -> f.getMappingType() == OpenSearchDataType.MappingType.Keyword); - } - return false; - } - public AbstractRelNode pushDownLimit(LogicalSort sort, Integer limit, Integer offset) { try { if (pushDownContext.isAggregatePushed()) { diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/PartialResultAggregatePushdown.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/PartialResultAggregatePushdown.java new file mode 100644 index 00000000000..f4e2493a00b --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/PartialResultAggregatePushdown.java @@ -0,0 +1,176 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.scan; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import javax.annotation.Nullable; +import org.opensearch.sql.executor.Warning; +import org.opensearch.sql.opensearch.data.type.OpenSearchDataType; +import org.opensearch.sql.opensearch.data.type.OpenSearchDataType.MappingType; +import org.opensearch.sql.opensearch.data.type.OpenSearchTextType; +import org.opensearch.sql.opensearch.mapping.IndexMapping; + +/** + * Partial-result fallback for a text/keyword mapping conflict on an aggregation's group key. + * + *

When a field is mapped as {@code keyword} in some indices of a wildcard pattern and {@code + * text} in others, the multi-index type merge must pick a single type valid on every shard, so it + * collapses the field to {@code text}-without-{@code .keyword}. Text has no doc values, so the + * aggregation cannot push down and instead falls back to a per-shard document scan that opens a + * Point-In-Time (PIT) context on every shard -- exhausting {@code search.max_open_pit_context} on a + * wide pattern. + * + *

This class computes which indices to keep so the aggregation can still push down: it + * partitions the matched indices by how each maps the group field, then picks a single homogeneous + * group by a deterministic priority (keyword first) and the warning describing what was excluded. + * The caller ({@link CalciteLogicalIndexScan}) uses {@link Plan#keptIndices()} to build a narrowed + * scan and re-runs pushdown over it. + * + *

Only the text/keyword conflict collapses aggregation pushdown this way, so this class handles + * that case specifically rather than a general conflict framework. + */ +final class PartialResultAggregatePushdown { + + /** Max excluded index names to spell out in the warning; the rest are summarized as "N more". */ + static final int MAX_EXCLUDED_INDICES_IN_WARNING = 5; + + private PartialResultAggregatePushdown() {} + + /** How one index maps the grouped field(s), in decreasing preference for a clean pushdown. */ + enum MappingResolution { + /** Every group field is a bare {@code keyword} -> merges to a clean {@code keyword}. */ + KEYWORD, + /** Every group field is aggregatable, at least one via a {@code .keyword} sub-field. */ + TEXT_WITH_KEYWORD, + /** At least one group field is not aggregatable here (bare {@code text}, or absent). */ + NOT_AGGREGATABLE + } + + /** + * The outcome of partitioning: which indices to aggregate over and the warning to attach. Absent + * (see {@link #plan}) when partial mode cannot or need not apply. + */ + record Plan(List keptIndices, List excludedIndices, Warning warning) {} + + /** + * Decide the partial-result plan for a group key over a set of per-index mappings. + * + * @param bucketNames the aggregation's group-by field names (dotted paths) + * @param mappings per-index field mappings, keyed by concrete index name (from {@code + * getIndexMappings}); the wildcard has already been resolved to concrete indices + * @return a plan naming the kept and excluded indices plus the warning, or {@code null} when + * partial mode does not apply: fewer than two indices, no aggregatable subset, or nothing + * excluded (the query would have pushed down normally) + */ + @Nullable + static Plan plan(List bucketNames, Map mappings) { + if (bucketNames.isEmpty() || mappings.size() < 2) { + return null; + } + + List keywordIndices = new ArrayList<>(); + List textKeywordIndices = new ArrayList<>(); + List excludedIndices = new ArrayList<>(); + for (Map.Entry entry : mappings.entrySet()) { + // Flatten so a nested object field (mapping tree resource -> attributes -> applicationid) is + // keyed by its dotted path, matching the bucket field name Calcite resolved. + Map flatMapping = + OpenSearchDataType.traverseAndFlatten(entry.getValue().getFieldMappings()); + switch (resolveBucketMapping(flatMapping, bucketNames)) { + case KEYWORD -> keywordIndices.add(entry.getKey()); + case TEXT_WITH_KEYWORD -> textKeywordIndices.add(entry.getKey()); + default -> excludedIndices.add(entry.getKey()); + } + } + + // Deterministic priority, not a count-based majority: keep the keyword group whenever one + // exists (the canonical aggregatable representation), so the returned data never depends on how + // many stray indices of another type match. Fall back to text-with-.keyword only when there is + // no keyword index. A mix of the two is still a text/keyword conflict that would re-collapse, + // so + // we never keep both -- recovering the excluded-but-aggregatable group needs a split-and-union. + List keptIndices; + if (!keywordIndices.isEmpty()) { + keptIndices = keywordIndices; + excludedIndices.addAll(textKeywordIndices); + } else if (!textKeywordIndices.isEmpty()) { + keptIndices = textKeywordIndices; + } else { + return null; // no aggregatable subset -> partial mode can't help + } + if (excludedIndices.isEmpty()) { + return null; // homogeneous already -> pushdown would not have failed + } + + excludedIndices.sort(null); + return new Plan( + keptIndices, excludedIndices, buildWarning(bucketNames, excludedIndices, mappings.size())); + } + + /** + * Resolve how one index maps all grouped fields, as the weakest resolution across them: {@link + * MappingResolution#KEYWORD} only if every field is bare keyword; {@link + * MappingResolution#TEXT_WITH_KEYWORD} if every field is aggregatable but at least one relies on + * a {@code .keyword} sub-field; otherwise {@link MappingResolution#NOT_AGGREGATABLE}. + */ + static MappingResolution resolveBucketMapping( + Map flatMapping, List bucketNames) { + MappingResolution combined = MappingResolution.KEYWORD; + for (String field : bucketNames) { + OpenSearchDataType type = flatMapping.get(field); + if (type == null) { + return MappingResolution.NOT_AGGREGATABLE; // field absent here -> can't aggregate cleanly + } + if (type.getMappingType() == MappingType.Keyword) { + continue; + } else if (hasKeywordSubField(type)) { + combined = MappingResolution.TEXT_WITH_KEYWORD; + } else { + return MappingResolution.NOT_AGGREGATABLE; + } + } + return combined; + } + + private static boolean hasKeywordSubField(OpenSearchDataType type) { + return type instanceof OpenSearchTextType textType + && textType.getFields().values().stream() + .anyMatch(f -> f.getMappingType() == MappingType.Keyword); + } + + private static Warning buildWarning( + List bucketNames, List excludedIndices, int totalIndices) { + String message = + String.format( + "Results exclude %d of %d indices due to a text/keyword mapping conflict on %s.", + excludedIndices.size(), totalIndices, bucketNames); + String detail = + String.format( + "Field %s is mapped inconsistently across the queried indices, which prevents" + + " aggregation pushdown for the whole pattern. The aggregation ran only over the" + + " indices where the field is aggregatable; excluded indices: %s. Align the" + + " field's mapping across all indices (e.g. map it as keyword everywhere) to" + + " include them.", + bucketNames, formatIndexList(excludedIndices, MAX_EXCLUDED_INDICES_IN_WARNING)); + return new Warning(Warning.TYPE_PARTIAL_RESULT, message, detail); + } + + /** + * Format a (sorted) index-name list for a warning: spell out up to {@code limit} names, then + * summarize any remainder as "and N more" so the message stays readable when the excluded set is + * large. + */ + static String formatIndexList(List indices, int limit) { + if (indices.size() <= limit) { + return indices.toString(); + } + return String.format( + "[%s, ... and %d more]", + String.join(", ", indices.subList(0, limit)), indices.size() - limit); + } +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/scan/PartialResultAggregatePushdownTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/scan/PartialResultAggregatePushdownTest.java new file mode 100644 index 00000000000..f87ff7dcae3 --- /dev/null +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/scan/PartialResultAggregatePushdownTest.java @@ -0,0 +1,220 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.scan; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.opensearch.sql.opensearch.storage.scan.PartialResultAggregatePushdown.MappingResolution.KEYWORD; +import static org.opensearch.sql.opensearch.storage.scan.PartialResultAggregatePushdown.MappingResolution.NOT_AGGREGATABLE; +import static org.opensearch.sql.opensearch.storage.scan.PartialResultAggregatePushdown.MappingResolution.TEXT_WITH_KEYWORD; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import org.junit.jupiter.api.Test; +import org.opensearch.sql.executor.Warning; +import org.opensearch.sql.opensearch.data.type.OpenSearchDataType; +import org.opensearch.sql.opensearch.data.type.OpenSearchDataType.MappingType; +import org.opensearch.sql.opensearch.data.type.OpenSearchTextType; +import org.opensearch.sql.opensearch.mapping.IndexMapping; +import org.opensearch.sql.opensearch.storage.scan.PartialResultAggregatePushdown.MappingResolution; +import org.opensearch.sql.opensearch.storage.scan.PartialResultAggregatePushdown.Plan; + +class PartialResultAggregatePushdownTest { + + private static final OpenSearchDataType KEYWORD_TYPE = OpenSearchDataType.of(MappingType.Keyword); + private static final OpenSearchDataType BARE_TEXT_TYPE = OpenSearchTextType.of(); + private static final OpenSearchDataType TEXT_WITH_KEYWORD_TYPE = + OpenSearchTextType.of(Map.of("keyword", OpenSearchDataType.of(MappingType.Keyword))); + + // ---- resolveBucketMapping ---- + + @Test + void resolveKeywordField() { + assertEquals( + KEYWORD, resolveOne(Map.of("f", KEYWORD_TYPE)), "bare keyword resolves to KEYWORD"); + } + + @Test + void resolveTextWithKeywordField() { + assertEquals( + TEXT_WITH_KEYWORD, + resolveOne(Map.of("f", TEXT_WITH_KEYWORD_TYPE)), + "text with a .keyword sub-field is aggregatable via the sub-field"); + } + + @Test + void resolveBareTextField() { + assertEquals( + NOT_AGGREGATABLE, + resolveOne(Map.of("f", BARE_TEXT_TYPE)), + "bare text (no .keyword) is not aggregatable"); + } + + @Test + void resolveAbsentField() { + assertEquals( + NOT_AGGREGATABLE, + PartialResultAggregatePushdown.resolveBucketMapping(Map.of(), List.of("f")), + "a field absent from the index cannot be aggregated cleanly"); + } + + @Test + void resolveMultiFieldTakesWeakestResolution() { + // One keyword + one text-with-.keyword group key -> the weaker TEXT_WITH_KEYWORD wins. + Map mapping = + Map.of("a", KEYWORD_TYPE, "b", TEXT_WITH_KEYWORD_TYPE); + assertEquals( + TEXT_WITH_KEYWORD, + PartialResultAggregatePushdown.resolveBucketMapping(mapping, List.of("a", "b"))); + // Add a bare-text key -> the whole index becomes NOT_AGGREGATABLE. + Map withBareText = + Map.of("a", KEYWORD_TYPE, "b", TEXT_WITH_KEYWORD_TYPE, "c", BARE_TEXT_TYPE); + assertEquals( + NOT_AGGREGATABLE, + PartialResultAggregatePushdown.resolveBucketMapping(withBareText, List.of("a", "b", "c"))); + } + + // ---- plan: not-applicable cases return null ---- + + @Test + void planNullForSingleIndex() { + assertNull(PartialResultAggregatePushdown.plan(List.of("f"), Map.of("idx", keywordIndex()))); + } + + @Test + void planNullForEmptyBucketNames() { + assertNull( + PartialResultAggregatePushdown.plan( + List.of(), Map.of("kw", keywordIndex(), "txt", bareTextIndex()))); + } + + @Test + void planNullWhenNoConflict() { + // Two keyword indices -> nothing excluded -> pushdown would have worked normally. + assertNull( + PartialResultAggregatePushdown.plan( + List.of("f"), Map.of("kw1", keywordIndex(), "kw2", keywordIndex()))); + } + + @Test + void planNullWhenNoAggregatableSubset() { + // Every index is bare text -> partial mode can't help. + assertNull( + PartialResultAggregatePushdown.plan( + List.of("f"), Map.of("txt1", bareTextIndex(), "txt2", bareTextIndex()))); + } + + // ---- plan: partitioning ---- + + @Test + void planKeepsKeywordExcludesBareText() { + Plan plan = + PartialResultAggregatePushdown.plan( + List.of("f"), ordered("kw", keywordIndex(), "txt", bareTextIndex())); + assertEquals(List.of("kw"), plan.keptIndices()); + assertEquals(List.of("txt"), plan.excludedIndices()); + assertEquals(Warning.TYPE_PARTIAL_RESULT, plan.warning().getType()); + } + + @Test + void planKeepsKeywordEvenWhenTextWithKeywordOutnumbersIt() { + // 1 keyword vs 3 text-with-.keyword. Keyword-first keeps the single keyword index; a count + // majority would have kept the 3. + Map mappings = + ordered( + "kw", keywordIndex(), + "tk1", textWithKeywordIndex(), + "tk2", textWithKeywordIndex(), + "tk3", textWithKeywordIndex()); + Plan plan = PartialResultAggregatePushdown.plan(List.of("f"), mappings); + assertEquals(List.of("kw"), plan.keptIndices()); + assertEquals(List.of("tk1", "tk2", "tk3"), plan.excludedIndices()); + } + + @Test + void planFallsBackToTextWithKeywordWhenNoKeywordIndex() { + Map mappings = + ordered("tk", textWithKeywordIndex(), "txt", bareTextIndex()); + Plan plan = PartialResultAggregatePushdown.plan(List.of("f"), mappings); + assertEquals(List.of("tk"), plan.keptIndices()); + assertEquals(List.of("txt"), plan.excludedIndices()); + } + + @Test + void planExcludedIndicesAreSorted() { + Map mappings = + ordered( + "kw", keywordIndex(), + "txt-c", bareTextIndex(), + "txt-a", bareTextIndex(), + "txt-b", bareTextIndex()); + Plan plan = PartialResultAggregatePushdown.plan(List.of("f"), mappings); + assertEquals(List.of("txt-a", "txt-b", "txt-c"), plan.excludedIndices()); + } + + // ---- formatIndexList ---- + + @Test + void formatShortListInFull() { + assertEquals( + "[a, b, c]", PartialResultAggregatePushdown.formatIndexList(List.of("a", "b", "c"), 5)); + } + + @Test + void formatLongListTruncated() { + List many = + IntStream.rangeClosed(1, 8).mapToObj(i -> "idx" + i).collect(Collectors.toList()); + String formatted = PartialResultAggregatePushdown.formatIndexList(many, 5); + assertTrue(formatted.contains("idx1"), "spells out the first few"); + assertTrue(formatted.contains("idx5"), "spells out up to the cap"); + assertTrue(formatted.contains("and 3 more"), "summarizes the remainder"); + assertTrue(!formatted.contains("idx6"), "does not list beyond the cap"); + } + + // ---- warning content ---- + + @Test + void warningNamesFieldExcludedIndicesAndCount() { + Plan plan = + PartialResultAggregatePushdown.plan( + List.of("f"), ordered("kw", keywordIndex(), "txt", bareTextIndex())); + Warning w = plan.warning(); + assertTrue(w.getMessage().contains("1 of 2 indices"), "message reports the excluded count"); + assertTrue(w.getMessage().contains("f"), "message names the field"); + assertTrue(w.getDetail().contains("txt"), "detail names the excluded index"); + } + + // ---- helpers ---- + + private static MappingResolution resolveOne(Map mapping) { + return PartialResultAggregatePushdown.resolveBucketMapping(mapping, List.of("f")); + } + + private static IndexMapping keywordIndex() { + return new IndexMapping(Map.of("f", KEYWORD_TYPE)); + } + + private static IndexMapping bareTextIndex() { + return new IndexMapping(Map.of("f", BARE_TEXT_TYPE)); + } + + private static IndexMapping textWithKeywordIndex() { + return new IndexMapping(Map.of("f", TEXT_WITH_KEYWORD_TYPE)); + } + + /** Build an insertion-ordered map so kept/excluded assertions are deterministic. */ + private static Map ordered(Object... keyThenValue) { + Map map = new LinkedHashMap<>(); + for (int i = 0; i < keyThenValue.length; i += 2) { + map.put((String) keyThenValue[i], (IndexMapping) keyThenValue[i + 1]); + } + return map; + } +}