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/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/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java b/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java index b715fdac86d..dc48f643ee7 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,27 @@ 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, 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 = warnings.stream().distinct().toList(); + 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..daa64c2f118 --- /dev/null +++ b/core/src/main/java/org/opensearch/sql/executor/Warning.java @@ -0,0 +1,34 @@ +/* + * 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 { + + /** + * 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. */ + private final String message; + + /** Optional longer explanation with the specifics and remedy; may be null. */ + private final String detail; +} 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..e0b5257a170 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java @@ -0,0 +1,314 @@ +/* + * 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_*"; + + 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_*"; + + // 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. + 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(); + 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); + } + + // 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); + } + + // 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); + } + } + + // 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 + 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")); + } + + @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 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 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); + 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( + "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/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/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/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..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 @@ -42,15 +42,18 @@ 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; 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; 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 +434,74 @@ public AbstractRelNode pushDownAggregate(Aggregate aggregate, @Nullable Project return null; } + /** + * 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 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) + osIndex + .getSettings() + .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()); + // 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, 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; + } + + OpenSearchIndex narrowedIndex = + new OpenSearchIndex( + osIndex.getClient(), osIndex.getSettings(), String.join(",", plan.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 + } + + CalcitePlanContext.addWarning(plan.warning()); + return pushed; + } catch (Exception e) { + if (LOG.isDebugEnabled()) { + LOG.debug("Cannot apply partial-result aggregate pushdown for {}", aggregate, e); + } + return null; + } + } + 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/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. * 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; + } +} 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..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. @@ -323,7 +326,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)); } @@ -345,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<>() { 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 =