[Feature] Add PPL outputlookup command (synchronous terminal write sink)

[noCharger]

Description

Implements the PPL outputlookup command from RFC #5625 — a synchronous, terminal write sink that materializes the current pipeline result into a lookup and returns a single rows_written count.

... | outputlookup [append=<bool>] [override_if_empty=<bool>] [key_field=<f1>(,<f2>)*] [max=<int>] <name>

Semantics, substrate, consistency contract, <name> resolution/migration, permissions, and alternatives are all in the RFC and are not restated here.

This PR also adds the operator ceiling plugins.ppl.outputlookup.max_rows (NodeScope, Dynamic, default 1_000_000): a single call exceeding it fails with 400 and writes nothing (fail-loud, no truncated slice), orthogonal to the per-query max=<int> truncation.

Tests: CalcitePPLOutputLookupIT (18, incl. testMaxRowsSettingRejectsExceeding), OutputLookupPermissionsIT (2, incl. a read-privileged user reading the lookup through its alias), and unit tests.

Benchmark (3-node m5.xlarge): write-bound; per-batch refresh is ~2.5x of plain bulk (the main follow-up lever); 1M rows written un-truncated with no OOM; same-name concurrent overwrite is last-writer-wins with 13,532 reads and 0 torn/partial across 30 atomic repoints; crash-window self-heals on re-run. Full perf + chaos results and charts are in the comment below.

Related Issues

Addresses #5625

Check List

• New functionality includes testing.
• New functionality has been documented.
  • New functionality has javadoc added.
  • New functionality has a user manual doc added.
• New PPL command checklist all confirmed.
• 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.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit f9d9260)

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 drain method checks if rows.size() > maxRows after adding each row, but this means it can accumulate maxRows+1 rows before throwing. If maxRows is 1_000_000, the list could hold 1_000_001 rows before the exception fires, exceeding the intended ceiling. The check should be rows.size() >= maxRows or should occur before adding the row. Enumerator<@Nullable Object> input, @Nullable Integer max, int maxRows) { List<Object[]> rows = new ArrayList<>(); while (input.moveNext()) { Object cur = input.current(); rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); if (max != null && rows.size() >= max) { break; } if (rows.size() > maxRows) { throw new IllegalArgumentException( "outputlookup wrote nothing because the result has more than " + maxRows + " rows, the maximum allowed for a single write. To write fewer rows, add" + " max=<n> to your query. To allow more, raise the" + " plugins.ppl.outputlookup.max_rows setting."); Possible Issue The extractLookupUuid method parses JSON from a filter string but swallows IOException silently and returns null. If the filter is malformed or the parsing fails for any reason, the method returns null without logging or signaling the issue. This could mask configuration errors or corrupted alias metadata. A corrupted filter would be treated as "no discriminant" and the append would be rejected, but the root cause (parse failure) would be invisible to the operator. private static @Nullable String extractLookupUuid(@Nullable String filterJson) { if (filterJson == null) { return null; } try (XContentParser parser = XContentType.JSON .xContent() .createParser( NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, filterJson)) { Map<String, Object> root = parser.map(); if (!(root.get("term") instanceof Map<?, ?> term)) { return null; } Object value = term.get(LookupsIndex.LOOKUP_FIELD); if (value instanceof Map<?, ?> valueObject) { Object nested = valueObject.get("value"); return nested == null ? null : nested.toString(); } return value == null ? null : value.toString(); } catch (IOException e) { return null; } } Possible Issue The flushBatch retry loop uses System.nanoTime() to enforce a timeout, but if the system clock jumps backward (e.g., NTP correction), deadlineNanos could be in the future indefinitely, causing the retry loop to run far longer than RETRY_TIMEOUT. Use a monotonic elapsed-time approach or accept that rare clock adjustments may extend the timeout window. long deadlineNanos = System.nanoTime() + RETRY_TIMEOUT.nanos(); while (true) { BulkResponse response = client.bulk(pending).actionGet(); if (!response.hasFailures()) { written += pending.numberOfActions(); return; } BulkRequest retry = new BulkRequest(); retry.setRefreshPolicy(cfg.refresh()); List<ItemFailure> fatal = new ArrayList<>(); int succeeded = 0; for (BulkItemResponse item : response.getItems()) { if (!item.isFailed()) { succeeded++; } else if (item.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS) { retry.add(pending.requests().get(item.getItemId())); } else { fatal.add(new ItemFailure(item.getItemId(), item.getId(), item.getFailureMessage())); } } written += succeeded; if (!fatal.isEmpty()) { throw new BulkWriteException("outputlookup bulk write hit non-retryable failures", fatal); } if (retry.numberOfActions() == 0) { return; } if (!backoff.hasNext() || System.nanoTime() >= deadlineNanos) { List<ItemFailure> exhausted = new ArrayList<>(); for (int i = 0; i < retry.numberOfActions(); i++) { exhausted.add(new ItemFailure(i, null, "429 retry budget exhausted")); } throw new BulkWriteException( "outputlookup bulk write exhausted 429 retries or hit the " + RETRY_TIMEOUT + " retry timeout", exhausted); } try { Thread.sleep(backoff.next().millis()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new BulkWriteException("interrupted during bulk retry backoff", List.of()); } pending = retry;

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to f9d9260

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Validate maxRows is positive The validation logic checks max against maxRows but does not validate that maxRows itself is positive. If the setting plugins.ppl.outputlookup.max_rows is misconfigured to zero or negative, the validation could pass with max=null and later cause unexpected behavior when draining rows. Add a guard to ensure maxRows is positive before using it. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [59-66] +if (maxRows < 1) { + throw new IllegalStateException( + "plugins.ppl.outputlookup.max_rows must be at least 1, but was " + maxRows); +} if (max != null && (max < 1 || max > maxRows)) { throw new IllegalArgumentException( "outputlookup max must be between 1 and the operator ceiling" + " plugins.ppl.outputlookup.max_rows (" + maxRows + "), but was " + max); } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that maxRows should be validated before use. However, the setting already has a minimum of 1 defined in OUTPUTLOOKUP_MAX_ROWS_SETTING, so this is a defensive check rather than a critical bug fix.	Medium
General	Prevent sleep from exceeding deadline The retry loop checks the deadline after determining there are items to retry, but the sleep that follows could push execution past the deadline. If the sleep duration exceeds the remaining time, the thread will sleep longer than intended. Check the deadline before sleeping to avoid overshooting the timeout. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [119-140] if (retry.numberOfActions() == 0) { return; } -if (!backoff.hasNext() || System.nanoTime() >= deadlineNanos) { +long remainingNanos = deadlineNanos - System.nanoTime(); +if (!backoff.hasNext() || remainingNanos <= 0) { List<ItemFailure> exhausted = new ArrayList<>(); for (int i = 0; i < retry.numberOfActions(); i++) { exhausted.add(new ItemFailure(i, null, "429 retry budget exhausted")); } throw new BulkWriteException( "outputlookup bulk write exhausted 429 retries or hit the " + RETRY_TIMEOUT + " retry timeout", exhausted); } +TimeValue nextBackoff = backoff.next(); +long sleepMillis = Math.min(nextBackoff.millis(), remainingNanos / 1_000_000); +try { + Thread.sleep(sleepMillis); +} catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new BulkWriteException("interrupted during bulk retry backoff", List.of()); +} Suggestion importance[1-10]: 6 __ Why: The suggestion addresses a minor timing precision issue where the sleep could exceed the deadline. While this improves accuracy, the impact is limited since the timeout is already generous (60 seconds) and the overshoot would be bounded by one backoff interval.	Low
	Optimize field lookup performance The code uses fields.indexOf(keyField) inside a loop over keyFields, resulting in O(n*m) complexity where n is the number of key fields and m is the number of fields. For large field lists, this could become a performance bottleneck. Pre-compute a field-to-index map once before the loop to achieve O(n+m) complexity. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java [35-53] +Map<String, Integer> fieldIndex = new HashMap<>(); +for (int i = 0; i < fields.size(); i++) { + fieldIndex.put(fields.get(i), i); +} for (String keyField : keyFields) { - int idx = fields.indexOf(keyField); - Object value = (idx >= 0 && idx < row.length) ? row[idx] : MISSING; + Integer idx = fieldIndex.get(keyField); + Object value = (idx != null && idx < row.length) ? row[idx] : MISSING; if (value == null || value == MISSING) { digest.update((byte) 'N'); continue; } if (value instanceof Object[] || value instanceof Collection) { throw new IllegalArgumentException( "outputlookup key_field [" + keyField + "] must not be multivalue"); } ... } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies an O(n*m) complexity issue and proposes a valid optimization using a HashMap. However, the impact is limited since keyFields is typically small (1-3 fields), and the method is called once per row rather than in a tight inner loop.	Low
	Avoid draining rows unnecessarily The code drains all rows into memory before checking overrideIfEmpty. If the result is large and overrideIfEmpty=false, the memory is wasted because the function returns early without writing. Move the overrideIfEmpty check before draining to avoid unnecessary memory allocation when the result will be discarded. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [70-76] +Target target = resolveTarget(client, name); List<Object[]> rows = drain(input, max, maxRows); if (!append && rows.isEmpty() && !overrideIfEmpty) { return 0; } -Target target = resolveTarget(client, name); - Suggestion importance[1-10]: 3 __ Why: The suggestion is incorrect. The overrideIfEmpty check requires knowing whether rows.isEmpty(), which necessitates draining first. Moving resolveTarget before draining doesn't avoid the drain operation and could introduce unnecessary work if the target resolution fails.	Low

---

Previous suggestions

Suggestions up to commit 9d9a533

Category	Suggestion	Impact
General	Check maxRows limit before adding row The maxRows check occurs after adding the row to the list, meaning the list can grow to maxRows + 1 before the exception is thrown. This wastes memory and is inconsistent with the error message stating the result "has more than maxRows rows". Check the size before adding the row to fail at exactly maxRows + 1 rows. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [141-159] private static List<Object[]> drain( Enumerator<@Nullable Object> input, @Nullable Integer max, int maxRows) { List<Object[]> rows = new ArrayList<>(); while (input.moveNext()) { + if (rows.size() >= maxRows) { + throw new IllegalArgumentException( + "outputlookup wrote nothing because the result has more than " + + maxRows + + " rows..."); + } Object cur = input.current(); rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); if (max != null && rows.size() >= max) { break; } - if (rows.size() > maxRows) { - throw new IllegalArgumentException(...); - } } return rows; } Suggestion importance[1-10]: 7 __ Why: Valid observation that the check occurs after adding the row, allowing the list to grow to maxRows + 1 before failing. Moving the check before the add operation is more accurate and prevents unnecessary memory allocation for the rejected row.	Medium
	Track original item IDs in retry failures The loop creates ItemFailure objects with sequential indices i, but these indices don't correspond to the original request item IDs. The retry request contains only the failed items, so i is the position within the retry batch, not the original batch. This could mislead debugging efforts when tracking which original items failed. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [122-132] if (!backoff.hasNext() || System.nanoTime() >= deadlineNanos) { List<ItemFailure> exhausted = new ArrayList<>(); for (int i = 0; i < retry.numberOfActions(); i++) { - exhausted.add(new ItemFailure(i, null, "429 retry budget exhausted")); + int originalItemId = pending.requests().indexOf(retry.requests().get(i)); + exhausted.add(new ItemFailure(originalItemId, null, "429 retry budget exhausted")); } - throw new BulkWriteException( - "outputlookup bulk write exhausted 429 retries or hit the " - + RETRY_TIMEOUT - + " retry timeout", - exhausted); + throw new BulkWriteException(...); } Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that the loop index i doesn't correspond to original item IDs, which could hinder debugging. However, the proposed fix using indexOf may not work correctly if the same request object appears multiple times, and adds O(n²) complexity.	Low
	Optimize field lookup with map Using indexOf inside a loop over keyFields results in O(n*m) complexity where n is the number of key fields and m is the number of fields. For large field lists, this could become a performance bottleneck. Build a field-to-index map once before the loop to achieve O(n+m) complexity. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java [35-55] public static String encode(List<String> keyFields, List<String> fields, Object[] row) { MessageDigest digest = sha256(); + Map<String, Integer> fieldIndex = new HashMap<>(); + for (int i = 0; i < fields.size(); i++) { + fieldIndex.put(fields.get(i), i); + } for (String keyField : keyFields) { - int idx = fields.indexOf(keyField); - Object value = (idx >= 0 && idx < row.length) ? row[idx] : MISSING; + Integer idx = fieldIndex.get(keyField); + Object value = (idx != null && idx < row.length) ? row[idx] : MISSING; ... } return Base64.getUrlEncoder().withoutPadding().encodeToString(digest.digest()); } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies O(n*m) complexity from repeated indexOf calls. Using a map reduces this to O(n+m). However, for typical small key field counts, the performance gain may be negligible, and the added map allocation has its own cost.	Low
	Clarify max boundary validation message The validation logic checks max > maxRows but the error message says "between 1 and the operator ceiling". This is inconsistent because max == maxRows should be allowed based on the error message, but the condition rejects it. Change the condition to max < 1 || max > maxRows should be max < 1 || max > maxRows is correct, but reconsider if the boundary should be inclusive. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [59-66] if (max != null && (max < 1 || max > maxRows)) { throw new IllegalArgumentException( - "outputlookup max must be between 1 and the operator ceiling" - + " plugins.ppl.outputlookup.max_rows (" + "outputlookup max must be between 1 and " + maxRows - + "), but was " + + " (plugins.ppl.outputlookup.max_rows), but was " + max); } Suggestion importance[1-10]: 4 __ Why: The validation logic is correct (max > maxRows properly rejects values exceeding the ceiling), but the error message could be clearer by stating the actual ceiling value first. This is a minor readability improvement.	Low

Suggestions up to commit 9c2adda

Category	Suggestion	Impact
General	Preserve document IDs in retry failures The retry loop does not preserve the original document IDs from failed items when constructing the exhausted list. The ItemFailure records use the retry batch position i instead of the original item ID from the response, making it difficult to trace which specific documents failed after retry exhaustion. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [122-132] if (!backoff.hasNext() || System.nanoTime() >= deadlineNanos) { List<ItemFailure> exhausted = new ArrayList<>(); for (int i = 0; i < retry.numberOfActions(); i++) { - exhausted.add(new ItemFailure(i, null, "429 retry budget exhausted")); + IndexRequest req = (IndexRequest) retry.requests().get(i); + exhausted.add(new ItemFailure(i, req.id(), "429 retry budget exhausted")); } - throw new BulkWriteException( - "outputlookup bulk write exhausted 429 retries or hit the " - + RETRY_TIMEOUT - + " retry timeout", - exhausted); + throw new BulkWriteException(...); } Suggestion importance[1-10]: 7 __ Why: Valid improvement for debugging failed retries. Preserving the document _id in ItemFailure records makes it easier to trace which specific documents failed after retry exhaustion, improving observability and troubleshooting.	Medium
	Optimize field lookup with map The encode() method uses fields.indexOf(keyField) inside a loop over keyFields, resulting in O(n*m) complexity where n is the number of key fields and m is the total number of fields. For queries with many fields or composite keys, this can become a performance bottleneck. Consider building a field-to-index map once before the loop. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java [35-54] public static String encode(List<String> keyFields, List<String> fields, Object[] row) { MessageDigest digest = sha256(); + Map<String, Integer> fieldIndex = new HashMap<>(); + for (int i = 0; i < fields.size(); i++) { + fieldIndex.put(fields.get(i), i); + } for (String keyField : keyFields) { - int idx = fields.indexOf(keyField); - Object value = (idx >= 0 && idx < row.length) ? row[idx] : MISSING; - if (value == null || value == MISSING) { - digest.update((byte) 'N'); - continue; - } - if (value instanceof Object[] || value instanceof Collection) { - throw new IllegalArgumentException( - "outputlookup key_field [" + keyField + "] must not be multivalue"); - } + Integer idx = fieldIndex.get(keyField); + Object value = (idx != null && idx < row.length) ? row[idx] : MISSING; ... } ... } Suggestion importance[1-10]: 7 __ Why: Valid performance optimization. Using indexOf() inside a loop creates O(n*m) complexity. Building a Map<String, Integer> once before the loop reduces this to O(n+m), which is beneficial for queries with many fields or composite keys.	Medium
	Check row limit before adding The drain() method checks rows.size() > maxRows after adding each row, meaning it will collect maxRows + 1 rows in memory before throwing. For the default limit of 1,000,000 rows, this wastes memory on the extra row. Check the size before adding to fail exactly at the limit. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [142-159] private static List<Object[]> drain( Enumerator<@Nullable Object> input, @Nullable Integer max, int maxRows) { List<Object[]> rows = new ArrayList<>(); while (input.moveNext()) { + if (rows.size() >= maxRows) { + throw new IllegalArgumentException( + "outputlookup wrote nothing because the result has more than " + + maxRows + + " rows..."); + } Object cur = input.current(); rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); if (max != null && rows.size() >= max) { break; } - if (rows.size() > maxRows) { - throw new IllegalArgumentException(...); - } } return rows; } Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that checking rows.size() > maxRows after adding allows one extra row in memory. Moving the check before adding prevents this minor memory waste and fails exactly at the limit, though the impact is minimal (one row overhead).	Low
	Validate effective row limit early The validation logic should check the result size against maxRows before attempting to write, not just validate the max parameter. If max is null and the result exceeds maxRows, the validation in drain() will throw after collecting all rows in memory, potentially causing OOM for very large results. Consider adding an early check or streaming validation. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [59-66] if (max != null && (max < 1 || max > maxRows)) { throw new IllegalArgumentException( "outputlookup max must be between 1 and the operator ceiling" + " plugins.ppl.outputlookup.max_rows (" + maxRows + "), but was " + max); } +// Effective limit is the smaller of max and maxRows +int effectiveLimit = (max != null) ? Math.min(max, maxRows) : maxRows; Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that validation happens after drain() collects rows, but the proposed fix doesn't address the core issue. The drain() method already enforces maxRows during collection. Adding effectiveLimit here is redundant since max is already validated and drain() handles the actual enforcement.	Low

Suggestions up to commit 306bf6e

Category	Suggestion	Impact
General	Enforce maxRows limit before buffering row The maxRows check occurs after adding the row, so when rows.size() equals maxRows + 1, the exception is thrown. This means the method allows maxRows + 1 rows to be buffered in memory before failing, which could exceed the intended memory limit. Check the size before adding the row to enforce the exact ceiling. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [156-171] -private static List<Object[]> drain( - Enumerator<@Nullable Object> input, @Nullable Integer max, int maxRows) { - List<Object[]> rows = new ArrayList<>(); - while (input.moveNext()) { - Object cur = input.current(); - rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); - if (max != null && rows.size() >= max) { - break; - } - if (rows.size() > maxRows) { - throw new IllegalArgumentException(...); - } +while (input.moveNext()) { + if (rows.size() >= maxRows) { + throw new IllegalArgumentException( + "outputlookup wrote nothing because the result has more than " + + maxRows + + " rows, the maximum allowed for a single write. To write fewer rows, add" + + " max=<n> to your query. To allow more, raise the" + + " plugins.ppl.outputlookup.max_rows setting."); } - return rows; + Object cur = input.current(); + rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); + if (max != null && rows.size() >= max) { + break; + } } Suggestion importance[1-10]: 8 __ Why: Correctly identifies an off-by-one issue where the method allows maxRows + 1 rows to be buffered before throwing an exception. This could cause memory issues when maxRows is set to the maximum allowed value. The suggested fix properly enforces the limit before adding the row.	Medium
	Prevent double-close of input enumerator The input.close() in the finally block may be called twice if an exception occurs after drain() completes, since drain() already closes the input when it finishes. This could lead to resource cleanup issues or exceptions. Ensure the enumerator is closed exactly once by tracking its state or removing the redundant close. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [145-147] -} finally { +} catch (Exception e) { input.close(); + throw e; } Suggestion importance[1-10]: 3 __ Why: The concern about double-close is valid, but the suggested solution (removing the finally block) is incorrect. The drain() method doesn't close the input; it only iterates through it. The finally block at lines 145-147 is necessary to ensure cleanup. The improved code would actually break proper resource management.	Low
Possible issue	Validate max parameter before draining input The validation logic should check max against maxRows before draining rows from the input enumerator. If validation fails after draining, the input data is already consumed but the write is rejected, potentially losing data. Move this validation to occur before drain() is called. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [61-68] if (max != null && (max < 1 || max > maxRows)) { + input.close(); throw new IllegalArgumentException( "outputlookup max must be between 1 and the operator ceiling" + " plugins.ppl.outputlookup.max_rows (" + maxRows + "), but was " + max); } +List<Object[]> rows = drain(input, max, maxRows); + Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that validation should occur before consuming the input stream. However, the improved code doesn't fully address the issue since it only closes the input but doesn't move the validation before drain() is called. The validation still occurs at the same location (lines 61-68), while drain() is called at line 72.	Medium
	Report pending items on interrupt When an InterruptedException occurs during retry backoff, the current batch in pending is lost without being written or reported. This silently drops data. The exception should include the pending items as failures so the caller knows which rows were not written. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [133-138] } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new BulkWriteException("interrupted during bulk retry backoff", List.of()); + List<ItemFailure> interrupted = new ArrayList<>(); + for (int i = 0; i < pending.numberOfActions(); i++) { + interrupted.add(new ItemFailure(i, null, "interrupted during retry")); + } + throw new BulkWriteException("interrupted during bulk retry backoff", interrupted); } Suggestion importance[1-10]: 6 __ Why: Valid suggestion to improve error reporting by tracking which items were lost during interruption. This helps with debugging and data loss tracking, though it's not a critical bug since the interrupt scenario is exceptional and the thread interruption is properly restored.	Low

Suggestions up to commit 5dc4c9a

Category	Suggestion	Impact
Possible issue	Fix off-by-one in row limit The maxRows check occurs after adding the row, so when rows.size() equals maxRows + 1, the exception is thrown. This means the actual limit enforced is maxRows + 1, not maxRows. Move the check before adding the row to enforce the correct limit. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [150-169] private static List<Object[]> drain( Enumerator<@Nullable Object> input, @Nullable Integer max, int maxRows) { List<Object[]> rows = new ArrayList<>(); while (input.moveNext()) { + if (rows.size() >= maxRows) { + throw new IllegalArgumentException( + "outputlookup wrote nothing because the result has more than " + + maxRows + + " rows, the maximum allowed for a single write. To write fewer rows, add" + + " max=<n> to your query. To allow more, raise the" + + " plugins.ppl.outputlookup.max_rows setting."); + } Object cur = input.current(); rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); if (max != null && rows.size() >= max) { break; } - if (rows.size() > maxRows) { - throw new IllegalArgumentException(...); - } } return rows; } Suggestion importance[1-10]: 9 __ Why: This is a critical bug. The check occurs after adding the row, allowing maxRows + 1 rows instead of maxRows. This violates the documented limit and could cause unexpected behavior when the limit is reached. Moving the check before adding the row correctly enforces the intended limit.	High
General	Preserve refresh exceptions during cleanup If the refresh operation throws an exception, restoreServeSettings is still called in the finally block, but the exception from refresh is lost. This could leave the index in an inconsistent state without proper error reporting. Ensure exceptions from critical operations like refresh are properly propagated. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [175-212] -private static void writeSlice( - NodeClient client, - String index, - List<String> fields, - WriteMode mode, - List<String> keyFields, - List<Object[]> rows, - String uuid) { +private static void writeSlice(...) { ... applyLoadSettings(client, index); try { WriteConfig cfg = new WriteConfig(...); try (OpenSearchBulkWriter writer = new OpenSearchBulkWriter(client, cfg)) { for (Object[] row : rows) { ... } } - client.admin().indices().refresh(new RefreshRequest(index)).actionGet(); + try { + client.admin().indices().refresh(new RefreshRequest(index)).actionGet(); + } catch (Exception e) { + LOGGER.error("Failed to refresh index {} after write", index, e); + throw e; + } } finally { - restoreServeSettings(client, index); + try { + restoreServeSettings(client, index); + } catch (Exception e) { + LOGGER.error("Failed to restore settings for index {}", index, e); + } } } Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that exceptions from the refresh operation could be masked by the finally block. However, the current code structure already propagates exceptions from the try block before the finally executes. The improvement adds explicit logging and exception handling in the finally block, which is a good practice but not critical.	Low
	Add maximum retry count guard The retry loop lacks a maximum iteration count, relying solely on the timeout. If backoff.hasNext() always returns true or the timeout check fails, the loop could run indefinitely. Add an explicit maximum retry count as a safety guard to prevent unbounded retries. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [94-141] +private static final int MAX_RETRIES = 10; + private void flushBatch() { BulkRequest pending = batch; pending.setRefreshPolicy(cfg.refresh()); batch = new BulkRequest(); buffered = 0; Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); long deadlineNanos = System.nanoTime() + RETRY_TIMEOUT.nanos(); + int retryCount = 0; while (true) { BulkResponse response = client.bulk(pending).actionGet(); ... + if (!backoff.hasNext() || System.nanoTime() >= deadlineNanos || ++retryCount >= MAX_RETRIES) { + ... Suggestion importance[1-10]: 5 __ Why: Adding a maximum retry count is a reasonable safety guard, but the existing timeout mechanism already prevents unbounded retries. The BackoffPolicy.exponentialBackoff() iterator is finite and the timeout check provides a hard deadline, so this is a minor defensive improvement rather than fixing a critical issue.	Low
	Optimize repeated field index lookups Using fields.indexOf(keyField) for every key field in every row is O(n*m) where n is the number of rows and m is the number of fields. For large datasets with many key fields, this becomes a performance bottleneck. Pre-compute field indices once before the loop to achieve O(1) lookups. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java [35-56] public static String encode(List<String> keyFields, List<String> fields, Object[] row) { MessageDigest digest = sha256(); for (String keyField : keyFields) { int idx = fields.indexOf(keyField); - Object value = (idx >= 0 && idx < row.length) ? row[idx] : MISSING; - if (value == null || value == MISSING) { - digest.update((byte) 'N'); - continue; - } - if (value instanceof Object[] || value instanceof Collection) { - throw new IllegalArgumentException(...); - } - byte[] bytes = canonical(value).getBytes(StandardCharsets.UTF_8); - digest.update((byte) 'V'); - digest.update(typeTag(value)); - digest.update(intToBytes(bytes.length)); - digest.update(bytes); + ... } - return Base64.getUrlEncoder().withoutPadding().encodeToString(digest.digest()); + ... } +// Add a method that pre-computes indices: +public static java.util.function.Function<Object[], String> createEncoder( + List<String> keyFields, List<String> fields) { + int[] indices = new int[keyFields.size()]; + for (int i = 0; i < keyFields.size(); i++) { + indices[i] = fields.indexOf(keyFields.get(i)); + } + return row -> encodeWithIndices(keyFields, indices, row); +} + Suggestion importance[1-10]: 4 __ Why: The performance concern is valid for high-volume scenarios, but the encode method is called once per row during bulk writes, not in a tight inner loop. The O(n*m) complexity applies across all rows, not per row. Pre-computing indices would require API changes and the current implementation is acceptable for typical lookup sizes. This is a minor optimization rather than a significant issue.	Low

Suggestions up to commit b84bad6

Category	Suggestion	Impact
General	Check row limit before allocation The method accumulates all rows in memory before checking if maxRows is exceeded. For very large result sets, this can cause an out-of-memory error before the check triggers. Check the limit immediately after adding each row to fail fast and avoid unnecessary memory allocation. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [141-160] private static List<Object[]> drain( Enumerator<@Nullable Object> input, @Nullable Integer max, int maxRows) { List<Object[]> rows = new ArrayList<>(); while (input.moveNext()) { + if (rows.size() >= maxRows) { + throw new IllegalArgumentException( + "outputlookup wrote nothing because the result has more than " + + maxRows + + " rows, the maximum allowed for a single write. To write fewer rows, add" + + " max=<n> to your query. To allow more, raise the" + + " plugins.ppl.outputlookup.max_rows setting."); + } Object cur = input.current(); rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); if (max != null && rows.size() >= max) { break; } - if (rows.size() > maxRows) { - throw new IllegalArgumentException(...); - } } return rows; } Suggestion importance[1-10]: 8 __ Why: This is a valuable optimization that prevents unnecessary memory allocation and provides faster failure feedback. Moving the maxRows check before adding the row avoids accumulating excess rows in memory, which is important for large result sets.	Medium
	Add maximum retry count guard The retry loop lacks a maximum iteration count, relying solely on the timeout. If System.nanoTime() wraps or the backoff iterator is unbounded, the loop could run indefinitely. Add an explicit maximum retry count as a safety guard to prevent infinite loops even if the timeout check fails. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [94-141] +private static final int MAX_RETRIES = 10; + private void flushBatch() { BulkRequest pending = batch; pending.setRefreshPolicy(cfg.refresh()); batch = new BulkRequest(); buffered = 0; Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); long deadlineNanos = System.nanoTime() + RETRY_TIMEOUT.nanos(); + int retryCount = 0; while (true) { BulkResponse response = client.bulk(pending).actionGet(); + ... + if (retry.numberOfActions() == 0) { + return; + } + if (!backoff.hasNext() || System.nanoTime() >= deadlineNanos || ++retryCount >= MAX_RETRIES) { + ... + } ... } } Suggestion importance[1-10]: 7 __ Why: Adding a maximum retry count is a good defensive practice to prevent infinite loops, though the existing timeout check already provides protection. The suggestion improves robustness but is not critical since System.nanoTime() wrapping is extremely rare and BackoffPolicy.exponentialBackoff() is bounded.	Medium
	Optimize field lookup with map The method calls fields.indexOf(keyField) inside a loop for every key field, resulting in O(n*m) complexity where n is the number of key fields and m is the number of fields. For large field lists, this becomes a performance bottleneck. Pre-compute a field-to-index map once before the loop to achieve O(n+m) complexity. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java [33-53] public static String encode(List<String> keyFields, List<String> fields, Object[] row) { MessageDigest digest = sha256(); + java.util.Map<String, Integer> fieldIndex = new java.util.HashMap<>(); + for (int i = 0; i < fields.size(); i++) { + fieldIndex.put(fields.get(i), i); + } for (String keyField : keyFields) { - int idx = fields.indexOf(keyField); - Object value = (idx >= 0 && idx < row.length) ? row[idx] : MISSING; - if (value == null || value == MISSING) { - digest.update((byte) 'N'); - continue; - } - if (value instanceof Object[] || value instanceof Collection) { - throw new IllegalArgumentException( - "outputlookup key_field [" + keyField + "] must not be multivalue"); - } - byte[] bytes = canonical(value).getBytes(StandardCharsets.UTF_8); - digest.update((byte) 'V'); - digest.update(typeTag(value)); - digest.update(intToBytes(bytes.length)); - digest.update(bytes); + Integer idx = fieldIndex.get(keyField); + Object value = (idx != null && idx < row.length) ? row[idx] : MISSING; + ... } return Base64.getUrlEncoder().withoutPadding().encodeToString(digest.digest()); } Suggestion importance[1-10]: 7 __ Why: Pre-computing a field-to-index map improves performance from O(n*m) to O(n+m), which is beneficial when dealing with many fields or key fields. However, for typical use cases with small field counts, the performance gain may be marginal.	Medium
	Document data loss risk during load Setting number_of_replicas to 0 during load eliminates redundancy, but if the node crashes before restoreServeSettings runs, data loss occurs. Consider checking if the index already has replicas configured and preserving that value, or document this risk explicitly for operators. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [205-214] private static void applyLoadSettings(NodeClient client, String index) { + // Note: Temporarily disabling replicas during load improves write throughput but risks + // data loss if the node crashes before restoreServeSettings() completes. Ensure the + // cluster is stable or accept this trade-off for performance. updateSettings( client, index, Settings.builder() .put("index.number_of_replicas", 0) .put("index.translog.durability", "async") .put("index.refresh_interval", "-1") .build()); } Suggestion importance[1-10]: 6 __ Why: Adding documentation about the data loss risk is helpful for operators, but the trade-off appears intentional for performance during bulk loading. The suggestion improves clarity but doesn't change behavior or fix a bug.	Low

[github-actions]

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit 306bf6e.

Path	Line	Severity	Description
opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java	288	medium	lookupFilter() constructs alias filter JSON via string concatenation. The uuid argument is always a standard UUID string (safe in practice), but this pattern is fragile: if the input source changes (e.g., a refactor passes user-supplied text), it becomes a JSON injection vector affecting alias filter logic and potentially routing reads to unintended index slices.
opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java	205	low	drain() materializes the entire pipeline result into a heap ArrayList before writing. Although bounded by the maxRows setting (default 1,000,000), a large result set combined with wide rows can exhaust coordinator heap, causing OOM or degraded cluster performance for other queries running concurrently.
opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java	247	low	applyLoadSettings() temporarily sets the backing index to 0 replicas and async translog durability. If the node crashes between applyLoadSettings and the restoreServeSettings finally block, the index is left permanently under-replicated with reduced durability, silently affecting data safety for subsequent reads even after the write appears to succeed.

The table above displays the top 10 most important findings.

Total: 3 | Critical: 0 | High: 0 | Medium: 1 | Low: 2

---

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

---

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

[github-actions]

Persistent review updated to latest commit 9476e41

[github-actions]

Persistent review updated to latest commit 92cfd5f

[github-actions]

Persistent review updated to latest commit 4c9fb21

[github-actions]

Persistent review updated to latest commit d39455c

[noCharger]

Benchmark: performance and resilience

3-node m5.xlarge, node-side, single run per cell (directional).

Performance (current build)

Measured on the current build: per-batch RefreshPolicy.NONE + one refresh before the alias repoint, plus build-time slice load settings (0 replicas, async translog, no auto refresh during the load, all restored to serving settings before publish) and bulk batch size 5000.

• Write-bound, not read-bound: the source scan is 0.01 to 0.2s at every size; the cost is the write.
• outputlookup write throughput is now on par with or faster than a plain _bulk load of the same rows. Overwrite, rows/s:

write	outputlookup	plain _bulk	ratio
100K / 2 cols	25,802	27,533	0.94x
100K / 5 cols	30,980	19,984	1.55x
100K / 20 cols	11,145	11,329	0.98x
1M / 2 cols	44,358	31,275	1.42x
1M / 5 cols	32,982	24,043	1.37x
1M / 20 cols	13,774	10,976	1.25x

• An earlier build using per-batch RefreshPolicy.IMMEDIATE was about 2.5x slower than plain bulk. Removing the per-batch refresh and loading the pre-publish slice with build-time settings eliminates that tax.
• Why it can match or beat plain bulk: the slice is invisible until the atomic alias repoint, so it is loaded like a build-time index (0 replicas, async translog, no auto refresh) and restored to serving settings before publish. The plain _bulk baseline used default settings (1 replica, request durability), so the ratios reflect each path at its realistic write config rather than identical durability.
• overwrite is approximately equal to append and to keyed upsert at scale: the alias repoint and the uuid-salted _id are minor next to the write.
• 1M rows written un-truncated via PIT, no OOM even at 1M x 20 cols (20M cells) on an 8GB heap.
• Read tax negligible: a 100-row lookup reads in about 11 to 15ms with 0 / 9 / 99 orphan slices in the backing index; the __lookup term filter is efficient, so the reaper is storage-justified, not read-latency-justified, at this scale.

Resilience (chaos)

• Same-name concurrent overwrite, last-writer-wins, reads never torn: 6 writers x 5 rounds (30 overwrites), a dense poller over the alias ran 13,532 reads with 0 violations (no empty, partial, or mixed-writer read) across 30 observed atomic repoints; the alias settled on exactly one uuid with complete data.
• Crash-window self-heal: reconstructing the exact post-crash state (slice written, alias not yet repointed or added) and re-running the real command heals in both windows (overwrite-before-repoint, create-before-alias); during the window a read returns only the complete old slice, or does not resolve, the orphan is invisible through the filter and enumerable for the reaper.

[songkant-aws]

The class has no shared mutable state, but the lookup lifecycle is not concurrency-safe. In particular, two concurrent appends to an absent lookup can both resolve ABSENT, write separate UUID slices, and race to install the alias. Both requests may return success, while only the last alias target remains visible. Concurrent append and overwrite has a similar lost-visibility problem. What concurrency contract do we want for same-name writes?

For example, similar concurrent queries like ... | outputlookup append=true my_lookup generate the following sequence, resulting orphan slice in the backing index:

Timestamp	Request A	Request B	alias status
T1	resolve → ABSENT		non-existent
T2		resolve → ABSENT	non-existent
T3	write uuid-A		non-existent
T4		write uuid-B	non-existent
T5	add alias → uuid-A		my_lookup → uuid-A
T6	return success	add alias → uuid-B	my_lookup → uuid-B
T7		return success	my_lookup → uuid-B

[noCharger]

The class has no shared mutable state, but the lookup lifecycle is not concurrency-safe. In particular, two concurrent appends to an absent lookup can both resolve ABSENT, write separate UUID slices, and race to install the alias. Both requests may return success, while only the last alias target remains visible. Concurrent append and overwrite has a similar lost-visibility problem. What concurrency contract do we want for same-name writes?

For example, similar concurrent queries like ... | outputlookup append=true my_lookup generate the following sequence, resulting orphan slice in the backing index:

Timestamp Request A Request B alias status
T1 resolve → ABSENT   non-existent
T2   resolve → ABSENT non-existent
T3 write uuid-A   non-existent
T4   write uuid-B non-existent
T5 add alias → uuid-A   my_lookup → uuid-A
T6 return success add alias → uuid-B my_lookup → uuid-B
T7   return success my_lookup → uuid-B

Contract defined as append-to-absent useing a deterministic per-lookup discriminant so concurrent first-appends converge into one slice (no lost write); overwrite is last-writer-wins on the atomic repoint. Added a concurrent IT.

[noCharger]

Contract for concurrent same-name writes:

• Concurrent first-time appends to an absent lookup converge on one slice via a deterministic per-lookup discriminant, so every appended row persists and the alias resolves to that slice. Covered by testConcurrentAppendToAbsentDoesNotLoseWrites.
• A concurrent overwrite is last-writer-wins on the atomic repoint. An append that races an overwrite may write into the slice the overwrite orphans, so those appended rows are acknowledged while staying outside the published lookup. Concrete interleaving on an existing lookup hosts (alias -> U0):

T	A: outputlookup append=true hosts	B: outputlookup hosts (overwrite)	alias
T1	resolve -> ALIAS (U0, primary hosts__lookup)		hosts -> U0
T2		resolve -> ALIAS (U0)	hosts -> U0
T3		writeSlice(hosts__lookup, U1)	hosts -> U0
T4		repoint -> U1	hosts -> U1 (U0 orphaned)
T5	writeSlice(hosts__lookup, U0)		hosts -> U1
T6	return success		hosts -> U1

A wrote into U0, which the overwrite orphaned, so A's rows stay outside the published lookup while A reports success.

• When both writers observe the lookup as absent (one append, one overwrite), the alias resolves to whichever installs it last, so the surviving data comes from that writer.

This matches the merged data importer (OpenSearch-Dashboards#11303), which uses add-only alias updates and leaves cross-request coordination to the caller. I documented this contract in docs/user/ppl/cmd/outputlookup.md (Concurrency section) and recommend serializing writes to a given lookup, for example a single scheduled refresh per lookup, for deterministic results under contention. A per-lookup lock (control-index doc with seqNo/primaryTerm CAS) would eliminate the append-vs-overwrite window; I have it as a follow-up rather than in this PR to stay aligned with the importer precedent.

[github-actions]

Persistent review updated to latest commit 91bf200

[github-actions]

Persistent review updated to latest commit 4ebd6a0

[github-actions]

Persistent review updated to latest commit dade0d5

[github-actions]

Persistent review updated to latest commit b84bad6

[songkant-aws]

I’m still concerned that maintaining multiple UUID slices in one backing index introduces unnecessary lifecycle complexity. Before the alias is repointed, a newly written slice is indistinguishable from an orphan to the proposed reaper, so safe cleanup would require additional coordination or a grace period. The generations also share mappings, index-level settings, and the reserved __lookup field, and cleanup requires delete_by_query.
Would one physical index per generation plus an atomic unfiltered-alias switch be simpler? Orphan generations would then be unreferenced managed indices and could be removed by deleting the entire index.

[noCharger]

Aligned this with the merged precedent.

OpenSearch-Dashboards#11303 (the data importer) uses the same model: one index holding multiple slices keyed by a __lookup=<uuid> field, with a filtered alias {term:{__lookup:<uuid>}} published on a clean load. Field name, type, and alias-filter convention match, so outputlookup and the importer stay wire-compatible on the read side.

On orphan handling: #11303's alias update is add-only (a single add action), so a re-import repoints the alias and leaves the previous slice in place; it keeps the slice model and defers reclamation. I matched that here by removing the reaper/delete_by_query design. Overwrite repoints the filtered alias to the new slice in one atomic aliases request; reclaiming the superseded slice is deferred to a follow-up, consistent with the importer.

One deliberate difference, in the safer direction: the backing index is per-lookup (<name>__lookup), giving each lookup its own mapping and write boundary while keeping the identical __lookup + filtered-alias convention.

[songkant-aws]

Ok, I'm not quite familiar with data importer background. If our goal is to align with previous behavior, then we‘d better clarify all of limitations. Please also get the signoff from data importer author or other developer whoever knows the previous design decisions.

[github-actions]

Persistent review updated to latest commit 5dc4c9a

[github-actions]

Persistent review updated to latest commit 306bf6e

[github-actions]

Persistent review updated to latest commit 9c2adda

[github-actions]

Persistent review updated to latest commit 9d9a533

[github-actions]

Persistent review updated to latest commit f9d9260

[songkant-aws]

nit: need to be changed to 3.9

[songkant-aws]

nit: A better namespace control could be metadata enrichment to avoid index naming conflicts with other business indices. Curious if we could add reserved metadata info to _meta to strengthen it?

[songkant-aws]

Could we validate the target overwrite index to be valid lookup index?

[songkant-aws]

What would be partial write success/failure behavior? In case of append mode, could it generate partial duplicate rows? If we don't expect to handle it, add it to limitations.

[songkant-aws]

Ok, I'm not quite familiar with data importer background. If our goal is to align with previous behavior, then we‘d better clarify all of limitations. Please also get the signoff from data importer author or other developer whoever knows the previous design decisions.