[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 81a06c2)

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 Thread Interruption Handling The catch block for InterruptedException at line 134 re-interrupts the thread but then throws a BulkWriteException. If the calling code catches this exception and continues, the interrupt flag is set but the thread may not actually terminate. In a long-running service, this could prevent proper shutdown when an interrupt signal is sent during a retry backoff. case INDEX: throw new IllegalArgumentException( "outputlookup destination [" + name + "] is a concrete index, not a lookup; delete it first to reuse the name"); default: Unbounded Retry Loop The retry loop at line 94 can spin indefinitely if backoff.hasNext() always returns true and the deadline check fails to trigger (e.g., due to clock skew or System.nanoTime() wrapping). While unlikely, this could cause the write operation to hang indefinitely without writing data or failing cleanly. 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; } } Potential Data Loss At line 128, when an overwrite repoints the alias away from the old slice, the log message states the previous slices are "now unreferenced" but the code does not delete them. Over time, repeated overwrites accumulate orphaned slices that consume storage but are invisible to users. While this may be intentional for recovery, it is not documented as a cleanup responsibility for operators. repointFilteredAlias(client, name, target.aliasIndices(), backingIndex, uuid); LOGGER.info( "outputlookup overwrite of [{}] wrote a new slice into [{}] and repointed the" + " alias; the previous slice(s) on {} are now unreferenced", name, backingIndex, target.aliasIndices()); }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 81a06c2

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Check row limit before allocation The method loads all rows into memory before checking the maxRows limit, which could cause an OutOfMemoryError for large result sets. Check the limit before adding each row to prevent unbounded memory consumption and fail fast when the limit is exceeded. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [149-165] 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]: 9 __ Why: This is a critical memory safety issue. The current code checks rows.size() > maxRows after adding the row, allowing one extra row beyond the limit and potentially causing OOM for large datasets. Checking before allocation prevents unbounded memory consumption and provides proper fail-fast behavior.	High
	Add maximum retry count safeguard The retry loop lacks a maximum iteration count, relying solely on the timeout. If System.nanoTime() wraps or the backoff policy produces an unbounded sequence, the loop could run indefinitely. Add an explicit maximum retry count as a secondary safeguard to prevent infinite loops. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [93-94] +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 (retryCount++ >= MAX_RETRIES) { + throw new BulkWriteException("exceeded maximum retry attempts", exhausted); + } + ... Suggestion importance[1-10]: 7 __ Why: Adding a maximum retry count is a good defensive practice to prevent infinite loops, though the existing timeout mechanism already provides protection. The suggestion improves robustness but is not critical since System.nanoTime() wrapping is extremely rare and the backoff policy is bounded.	Medium
General	Prevent alias removal race condition If oldIndices contains newIndex, the method removes the alias from newIndex and then re-adds it, which could cause a transient state where the alias is absent. Filter out newIndex from oldIndices before removing to ensure the alias remains continuously available. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [224-232] private static void repointFilteredAlias( NodeClient client, String alias, List<String> oldIndices, String newIndex, String uuid) { IndicesAliasesRequest req = new IndicesAliasesRequest(); for (String old : oldIndices) { - req.addAliasAction(AliasActions.remove().index(old).alias(alias)); + if (!old.equals(newIndex)) { + req.addAliasAction(AliasActions.remove().index(old).alias(alias)); + } } req.addAliasAction(AliasActions.add().index(newIndex).alias(alias).filter(lookupFilter(uuid))); client.admin().indices().aliases(req).actionGet(); } Suggestion importance[1-10]: 8 __ Why: This addresses a potential race condition where newIndex could be in oldIndices, causing the alias to be temporarily removed and re-added. While the atomic alias operation mitigates some risk, filtering out newIndex ensures continuous alias availability and prevents unnecessary operations.	Medium
	Optimize field lookup performance Using indexOf for each key field 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 becomes inefficient. 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-39] 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; ... Suggestion importance[1-10]: 6 __ Why: The suggestion improves performance from O(n*m) to O(n+m) by using a map for field lookups. However, the impact is limited since keyFields and fields are typically small lists in practice, making the optimization beneficial but not critical for most use cases.	Low

---

Previous suggestions

Suggestions up to commit f9d9260

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

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

[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

[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.

[github-actions]

Persistent review updated to latest commit 81a06c2