Return a partial result with a warning instead of exhausting PIT contexts on a text/keyword mapping conflict

[ahkcs]

Description

On the Calcite PPL path, an aggregation (stats/top/rare) that groups, sorts, or dedups on a field whose mapping conflicts across a wildcard index pattern — keyword in some indices, text in others — cannot push down. The type merge must pick one type valid on every shard, so it collapses the field to text-without-.keyword; text has no doc values, so the aggregation falls back to an in-process document scan that opens a Point-In-Time (PIT) context on every shard of every matched index. On a wide pattern this trips search.max_open_pit_context and the customer sees an opaque failure.

This PR adds an opt-in fallback that returns a partial result plus a structured warning instead of failing. When normal aggregate pushdown fails on such a conflict and the setting is enabled, the engine narrows the scan to the subset of indices where the field is aggregatable, pushes the aggregation down over just that subset (so size = 0 and no PIT is opened), and attaches a PARTIAL_RESULT warning naming the excluded indices.

Two parts:

1. A general warning channel. Successful PPL responses gain an optional warnings array ({type, message, detail}), emitted only when non-empty so existing responses are byte-for-byte unchanged:

   {
     "schema": [ ... ], "datarows": [ ... ], "total": 3, "size": 3,
     "warnings": [
       {
         "type": "PARTIAL_RESULT",
         "message": "Results exclude 1 of 2 indices due to a text/keyword mapping conflict on [applicationid].",
         "detail": "Field [applicationid] is mapped inconsistently ... The aggregation ran over the largest consistently-mapped subset; excluded indices: [logs-text]. Align the field's mapping across all indices (e.g. map it as keyword everywhere) to include them."
       }
     ]
   }

2. The partial-result producer, wired into AggregateIndexScanRule as the fallback when pushdown returns null.

Behavior (setting off vs on), on a keyword+text wildcard pattern with search.max_open_pit_context below the shard count:

Query	Setting off (today)	Setting on
stats count() by <conflict field>	500 — PIT contexts exhausted	200 — counts over the aggregatable subset + PARTIAL_RESULT warning, 0 PITs
top / rare <conflict field>	500 — PIT contexts exhausted	200 — partial buckets + warning, 0 PITs
stats count() by span(...) (no conflict)	200 — complete	200 — complete, no warning (unchanged)
single-index / no-conflict aggregation	200 — complete	200 — complete, no warning (unchanged)
any of the above with format=csv	500 / 200 as above	falls through to the normal path (CSV has no warning channel)

Key design points

• Opt-in, default off (plugins.calcite.partial_result.on_mapping_conflict.enabled). A partial result is a knowingly-incomplete answer, so it never happens silently.
• Refuse if the format can't warn. Only the JSON response shape carries the warnings array, so partial mode is skipped for CSV/RAW/VIZ — those fall through to the normal (failing) path rather than returning a silently-partial answer.
• Which indices are kept is deterministic, not count-based. 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; always exclude bare-text. The returned data does not depend on how many indices of each type happen to match.
• Scoped to the Calcite PPL path; the V2/legacy path is untouched.

Not in scope: recovering an excluded but aggregatable group (e.g. text-with-.keyword when a keyword group is also present) — that needs a per-group split-and-union, which is a larger, separate change. Partial mode deliberately trades that completeness for a single narrowed scan and a clear warning.

Related Issues

Check List

• New functionality includes testing.
• New functionality has been documented.
• New functionality has javadoc added.
• API changes companion pull request created.
• Commits are signed per the DCO using --signoff or -s.
• Public documentation issue/PR created.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 078c949)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue The method tryPartialResultAggregate checks if bucketNames.isEmpty() and returns null, but this check occurs after calling aggregate.getGroupSet().cardinality() to determine the size of bucketNames. If cardinality() returns 0, bucketNames will be an empty sublist, and the method returns null. However, if the aggregate has no grouping fields (a global aggregation like stats count()), this is a valid case that should either be handled or explicitly documented as unsupported. The current implementation silently falls back without logging or warning, which may mask legitimate use cases or configuration issues. List<String> outputFields = aggregate.getRowType().getFieldNames(); List<String> bucketNames = outputFields.subList(0, aggregate.getGroupSet().cardinality()); if (bucketNames.isEmpty()) { return null; } Resource Leak The method tryPartialResultAggregate calls osIndex.getClient().getIndexMappings(indexExpression) which may allocate resources or perform I/O. If an exception is thrown after this call but before the method completes (e.g., during narrowedScan.pushDownAggregate), the catch block logs and returns null, but there is no guarantee that any resources held by the mappings retrieval are properly released. While the mappings object itself is a Map and likely doesn't hold open connections, the pattern of catching all exceptions and returning null without cleanup is risky if future changes introduce resource-holding operations. Map<String, IndexMapping> mappings = osIndex.getClient().getIndexMappings(indexExpression); if (mappings.size() < 2) { 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<String> keywordIndices = new ArrayList<>(); List<String> textKeywordIndices = new ArrayList<>(); List<String> excludedIndices = new ArrayList<>(); for (Map.Entry<String, IndexMapping> 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<String, OpenSearchDataType> 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<String> 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)); 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. 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) { if (LOG.isDebugEnabled()) { LOG.debug("Cannot apply partial-result aggregate pushdown for {}", aggregate, e); } return null; } Thread Safety The methods setWarningsSupported and isWarningsSupported use ThreadContext.put and ThreadContext.get with a boolean string value. If ThreadContext.get(WARNINGS_SUPPORTED_KEY) returns null (e.g., if the key was never set), Boolean.parseBoolean(null) returns false, which is the intended default. However, if another thread or code path sets the key to an invalid string (not "true" or "false"), parseBoolean will silently return false without error. This could mask configuration bugs where the warnings-supported flag is set incorrectly. Consider validating the stored value or using a more robust storage mechanism. 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)); }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 078c949

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Always remove thread-local to prevent leaks The method checks isEmpty() on the thread-local list before removing it, but if the list is empty, pendingWarnings.remove() is never called. This can leak empty lists across pooled threads. Always call remove() to clean up the thread-local, regardless of whether warnings were collected. core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java [271-279] public static List<Warning> drainWarnings() { List<Warning> warnings = pendingWarnings.get(); + pendingWarnings.remove(); if (warnings.isEmpty()) { return List.of(); } - List<Warning> drained = warnings.stream().distinct().toList(); - pendingWarnings.remove(); - return drained; + return warnings.stream().distinct().toList(); } Suggestion importance[1-10]: 9 __ Why: The current code fails to call pendingWarnings.remove() when the list is empty, which can leak empty ArrayList instances across pooled worker threads. This is a critical resource leak that violates the documented lifecycle contract.	High
	Prevent ClassCastException on setting retrieval The cast to Boolean can throw ClassCastException if the setting returns an unexpected type. Wrap the cast in a try-catch or use a safer type-checking approach to prevent runtime failures when the setting value is misconfigured. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java [451-456] -if (!(Boolean) - osIndex - .getSettings() - .getSettingValue(Settings.Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT)) { +Object settingValue = osIndex + .getSettings() + .getSettingValue(Settings.Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT); +if (!(settingValue instanceof Boolean && (Boolean) settingValue)) { return null; } Suggestion importance[1-10]: 7 __ Why: The unchecked cast to Boolean could throw ClassCastException if the setting returns an unexpected type. While settings are typically well-typed, adding defensive type checking improves robustness against misconfiguration.	Medium
General	Clarify exclusion logic for text-keyword fallback When keywordIndices is empty but textKeywordIndices is not, the code sets keptIndices but never adds anything to excludedIndices. This means bare-text indices that were already in excludedIndices remain there, but no additional indices are excluded. Verify this is intentional or add the appropriate exclusion logic. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java [503-511] List<String> keptIndices; if (!keywordIndices.isEmpty()) { keptIndices = keywordIndices; excludedIndices.addAll(textKeywordIndices); } else if (!textKeywordIndices.isEmpty()) { keptIndices = textKeywordIndices; + // excludedIndices already contains bare-text indices from the classification loop } else { return null; } Suggestion importance[1-10]: 3 __ Why: The suggestion adds a clarifying comment but doesn't change the logic. The code is already correct—bare-text indices are added to excludedIndices during classification, so no additional exclusion is needed in the textKeywordIndices branch. The comment is helpful but has minimal impact.	Low

---

Previous suggestions

Suggestions up to commit dad3bb3

Category	Suggestion	Impact
Possible issue	Prevent ThreadLocal leak on empty warnings The method calls pendingWarnings.remove() only when warnings are non-empty, leaving the ThreadLocal populated when empty. This can cause warnings to leak across queries on pooled threads. Always clear the ThreadLocal regardless of whether warnings exist. core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java [271-279] public static List<Warning> drainWarnings() { List<Warning> warnings = pendingWarnings.get(); + pendingWarnings.remove(); if (warnings.isEmpty()) { return List.of(); } - List<Warning> drained = warnings.stream().distinct().toList(); - pendingWarnings.remove(); - return drained; + return warnings.stream().distinct().toList(); } Suggestion importance[1-10]: 9 __ Why: Critical bug fix. The current code only calls pendingWarnings.remove() when warnings are non-empty (line 277), leaving the ThreadLocal populated when empty. This can cause warnings to leak across queries on pooled threads. Moving the remove() call before the empty check ensures proper cleanup.	High
General	Validate index names before joining Creating a narrowed index with a comma-separated list of index names may fail if any index name contains special characters or commas. Verify that keptIndices elements are properly escaped or use an alternative construction method that handles index names safely. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java [516-520] +// Ensure index names are properly validated/escaped before joining +String narrowedIndexPattern = String.join(",", keptIndices); OpenSearchIndex narrowedIndex = new OpenSearchIndex( - osIndex.getClient(), osIndex.getSettings(), String.join(",", keptIndices)); + osIndex.getClient(), osIndex.getSettings(), narrowedIndexPattern); Suggestion importance[1-10]: 3 __ Why: The suggestion raises a valid concern about special characters in index names, but the improved_code doesn't actually implement validation or escaping—it just adds a comment. The score reflects the importance of the concern but the lack of a concrete solution.	Low

Suggestions up to commit b475c5b

Category	Suggestion	Impact
General	Verify Warning deduplication works correctly The distinct() operation on Warning objects requires proper equals() and hashCode() implementations in the Warning class. Verify that the @Data annotation on Warning generates these methods correctly, or the deduplication may not work as intended. core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java [271-279] public static List<Warning> drainWarnings() { List<Warning> warnings = pendingWarnings.get(); if (warnings.isEmpty()) { return List.of(); } + // Deduplicate by value; Warning must have proper equals/hashCode List<Warning> drained = warnings.stream().distinct().toList(); pendingWarnings.remove(); return drained; } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that distinct() relies on equals() and hashCode() implementations. While @Data from Lombok does generate these methods, explicitly verifying this is important for correctness. However, this is more of a verification request than a code fix, so it doesn't warrant a higher score.	Medium
	Ensure deterministic index ordering The keptIndices list is constructed from map keys which may not be in a deterministic order. This could lead to non-deterministic query behavior across executions. Sort keptIndices before joining to ensure consistent index ordering. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java [516-518] -OpenSearchIndex narrowedIndex = +keptIndices.sort(null); + OpenSearchIndex narrowedIndex = new OpenSearchIndex( osIndex.getClient(), osIndex.getSettings(), String.join(",", keptIndices)); Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that keptIndices (derived from map keys) may have non-deterministic ordering, which could affect query behavior. Sorting before joining ensures consistency. However, the impact is moderate since the query results should be functionally equivalent regardless of index order in the pattern string.	Low
Possible issue	Fix resolution downgrade logic The logic doesn't handle the case where combined is already TEXT_WITH_KEYWORD and a subsequent field is KEYWORD. This could incorrectly downgrade the resolution. Ensure that once TEXT_WITH_KEYWORD is set, it remains unless a non-aggregatable field is found. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java [580-587] if (mappingType == OpenSearchDataType.MappingType.Keyword) { + // Keep combined as is (KEYWORD or TEXT_WITH_KEYWORD) continue; } else if (isTextWithKeywordSubField(type)) { - combined = MappingResolution.TEXT_WITH_KEYWORD; + if (combined == MappingResolution.KEYWORD) { + combined = MappingResolution.TEXT_WITH_KEYWORD; + } } else { return MappingResolution.NOT_AGGREGATABLE; } Suggestion importance[1-10]: 3 __ Why: The suggestion identifies a potential logic issue, but the current implementation is actually correct. The combined variable starts as KEYWORD and can only be downgraded to TEXT_WITH_KEYWORD or exit early with NOT_AGGREGATABLE. Once set to TEXT_WITH_KEYWORD, encountering another KEYWORD field correctly keeps it as TEXT_WITH_KEYWORD (the weaker resolution). The suggested change adds unnecessary complexity without fixing an actual bug.	Low

[github-actions]

Persistent review updated to latest commit dad3bb3

[github-actions]

Persistent review updated to latest commit 078c949