[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 5dc4c9a)

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 rows.size() > maxRows after adding each row, but this means it can accumulate maxRows + 1 rows before throwing. If maxRows is exactly 1_000_000 (the default), the check triggers at 1_000_001 rows, potentially exceeding the intended ceiling. The check should be rows.size() >= maxRows or placed before the add. 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 silently returns null on any IOException during filter parsing. If the filter JSON is malformed due to corruption or a manual edit, the method treats it as "no discriminant" rather than signaling an error. This could allow an append to proceed on a corrupted alias when it should fail, or cause a false "no discriminant" rejection. Consider logging the parse failure or propagating it as an error. 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) { // {"term":{"__lookup":{"value":"<uuid>"}}} Object nested = valueObject.get("value"); return nested == null ? null : nested.toString(); } return value == null ? null : value.toString(); // {"term":{"__lookup":"<uuid>"}} } catch (IOException e) { return null; } Possible Issue The retry loop in flushBatch uses System.nanoTime() to enforce a timeout, but if the system clock jumps backward (e.g., NTP correction), System.nanoTime() >= deadlineNanos may never become true, causing the loop to retry indefinitely. Use elapsed time (System.nanoTime() - startNanos >= RETRY_TIMEOUT.nanos()) instead of an absolute deadline to avoid this. 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()); Possible Issue The canonical method for BigDecimal calls stripTrailingZeros().toPlainString(), which can produce exponential notation for very large or very small decimals (e.g., 1E+50). This breaks the deterministic encoding contract because the same logical value can encode differently depending on scale. Use toPlainString() without stripTrailingZeros(), or explicitly format to avoid scientific notation. if (value instanceof BigDecimal bigDecimal) { // longValue() would truncate the fraction and collide distinct decimals onto one id return bigDecimal.stripTrailingZeros().toPlainString();

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 5dc4c9a

Explore these optional code suggestions:

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

---

Previous suggestions

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

Suggestions up to commit dade0d5

Category	Suggestion	Impact
Possible issue	Validate alias has backing indices When appending to an existing alias, writeSlice is called with target.primaryIndex(), but if target.aliasIndices() is empty, primaryIndex() returns null, which will cause a NullPointerException. Verify that target.aliasIndices() is not empty before calling primaryIndex() to prevent a potential crash. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [88-122] case ALIAS: { if (!target.filtered()) { throw new IllegalArgumentException(...); } if (append) { if (target.lookupUuid() == null) { throw new IllegalArgumentException(...); } + if (target.aliasIndices().isEmpty()) { + throw new IllegalStateException( + "outputlookup alias [" + name + "] has no backing indices"); + } writeSlice( client, target.primaryIndex(), fields, mode, keyFields, rows, target.lookupUuid()); } else { - String uuid = newUuid(); - writeSlice(client, backingIndex, fields, mode, keyFields, rows, uuid); - repointFilteredAlias(client, name, target.aliasIndices(), backingIndex, uuid); ... } break; } Suggestion importance[1-10]: 8 __ Why: The suggestion identifies a potential NullPointerException when target.aliasIndices() is empty and primaryIndex() returns null (line 104). Adding a check before calling primaryIndex() would prevent a crash and provide a clearer error message.	Medium
General	Validate max 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 stream has already been consumed and cannot be retried. Move this validation to occur before calling drain() to prevent resource waste and ensure the input remains unconsumed on validation failure. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [58-69] 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); } +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 code already performs validation before drain() is called (lines 58-65), so the suggestion's premise is incorrect. The improved_code doesn't change the order.	Medium
	Check maxRows before adding row The maxRows check occurs after adding the row to the list, so when rows.size() equals maxRows + 1, the exception is thrown. This means the method allows maxRows + 1 rows to be accumulated in memory before failing. Check the size before adding the row to fail exactly at the maxRows boundary and prevent unnecessary memory allocation. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [139-158] 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]: 6 __ Why: The suggestion correctly identifies that the maxRows check occurs after adding the row (line 147), allowing one extra row in memory before failing. Moving the check before adding the row would enforce the exact boundary and prevent unnecessary allocation, though the impact is minimal.	Low
	Check deadline before sleeping in retry The retry loop does not check the deadline before sleeping, which can cause the thread to sleep past the deadline and waste time. Check System.nanoTime() >= deadlineNanos before calling Thread.sleep() to avoid unnecessary delays when the timeout has already been exceeded. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [92-140] Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); long deadlineNanos = System.nanoTime() + RETRY_TIMEOUT.nanos(); while (true) { BulkResponse response = client.bulk(pending).actionGet(); ... if (!backoff.hasNext() || System.nanoTime() >= deadlineNanos) { ... throw new BulkWriteException(...); } + long sleepMillis = backoff.next().millis(); + if (System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(sleepMillis) >= deadlineNanos) { + throw new BulkWriteException(...); + } try { - Thread.sleep(backoff.next().millis()); + Thread.sleep(sleepMillis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new BulkWriteException("interrupted during bulk retry backoff", List.of()); } Suggestion importance[1-10]: 5 __ Why: The suggestion proposes checking the deadline before sleeping to avoid unnecessary delays. While this could marginally improve timeout precision, the existing check at line 119 already prevents excessive retries. The improvement is minor and adds complexity.	Low

Suggestions up to commit 4ebd6a0

Category	Suggestion	Impact
Possible issue	Check row limit before adding The maxRows check occurs after adding the row, so when rows.size() equals maxRows + 1, the exception is thrown. This means the method allows one more row than intended before failing. The check should occur before adding to enforce the exact limit. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [138-157] 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]: 7 __ Why: The suggestion correctly identifies that the maxRows check occurs after adding the row, allowing one more row than intended before failing. Moving the check before adding the row ensures the exact limit is enforced, which is a meaningful correctness improvement for the boundary condition.	Medium
General	Optimize field lookup performance Using indexOf for each key field in a loop has O(n*m) complexity where n is the number of key fields and m is the number of fields. For large field lists or multiple key fields, this becomes inefficient. Consider building 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; + 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 a performance issue with repeated indexOf calls in a loop. Building a field-to-index map improves complexity from O(n*m) to O(n+m). However, the impact depends on the typical size of keyFields and fields lists, and the optimization adds memory overhead for the map.	Low
	Handle flush exceptions properly The add method should handle potential exceptions from flushBatch() to prevent leaving the writer in an inconsistent state. If flushBatch() throws, buffered has already been incremented but the batch may not have been cleared, leading to incorrect state tracking. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [53-59] public void add(Object[] row) { ensureOpen(); batch.add(toIndexRequest(row)); - if (++buffered >= cfg.batchSize()) { + buffered++; + if (buffered >= cfg.batchSize()) { flushBatch(); } } Suggestion importance[1-10]: 3 __ Why: The suggestion to move the increment after checking the condition is a minor improvement that could prevent state inconsistency if flushBatch() throws. However, the impact is limited since flushBatch() resets buffered to 0 and creates a new batch, so the state would be corrected on the next call.	Low

Suggestions up to commit 3e25b5f

Category	Suggestion	Impact
General	Handle missing lookup on empty guard When append is false, rows is empty, and overrideIfEmpty is false, the method returns early without creating the backing index. However, if the lookup does not exist yet, a subsequent read will fail because the alias was never created. Consider creating the backing index and alias even in this case, or document that the lookup must exist before using override_if_empty=false. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [68-74] List<Object[]> rows = drain(input, max, maxRows); if (!append && rows.isEmpty() && !overrideIfEmpty) { + // Ensure the backing index and alias exist even when skipping the write, + // so subsequent reads do not fail on a missing lookup. + LookupsIndex.ensureExists(client, backingIndex); + Target target = resolveTarget(client, name); + if (target.kind() == Kind.ABSENT) { + String uuid = stableUuid(name); + addFilteredAlias(client, name, backingIndex, uuid); + } return 0; } LookupsIndex.ensureExists(client, backingIndex); Suggestion importance[1-10]: 7 __ Why: When override_if_empty=false and the result is empty, the method returns early without ensuring the lookup exists. This could cause subsequent reads to fail if the lookup was never created. The suggestion addresses a real usability issue.	Medium
	Avoid oversleeping past retry deadline The retry loop checks System.nanoTime() >= deadlineNanos after a failed bulk attempt, but it does not check the deadline before sleeping. If the deadline is already exceeded, the thread will still sleep for the full backoff duration before throwing. Check the deadline before sleeping to avoid unnecessary delay when the timeout has already been reached. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [92-141] Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); long deadlineNanos = System.nanoTime() + RETRY_TIMEOUT.nanos(); while (true) { BulkResponse response = client.bulk(pending).actionGet(); ... - if (!backoff.hasNext() || System.nanoTime() >= deadlineNanos) { + long now = System.nanoTime(); + if (!backoff.hasNext() || now >= deadlineNanos) { ... throw new BulkWriteException(...); } + TimeValue sleep = backoff.next(); + if (now + sleep.nanos() > deadlineNanos) { + sleep = TimeValue.timeValueNanos(deadlineNanos - now); + } try { - Thread.sleep(backoff.next().millis()); + Thread.sleep(sleep.millis()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new BulkWriteException("interrupted during bulk retry backoff", List.of()); } pending = retry; } Suggestion importance[1-10]: 6 __ Why: The retry loop could sleep past the deadline, causing unnecessary delay. The suggestion improves the timeout handling by capping the sleep duration, which is a minor but useful optimization for responsiveness.	Low
	Clarify max boundary inclusivity The validation logic allows max to equal maxRows, but the error message says "between 1 and the operator ceiling", which typically implies exclusivity. Either change the condition to max >= maxRows if the ceiling should be exclusive, or update the error message to clarify that max can equal maxRows (e.g., "between 1 and plugins.ppl.outputlookup.max_rows (inclusive)"). opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [57-64] 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 " + + " (inclusive, the plugins.ppl.outputlookup.max_rows ceiling), but was " + max); } Suggestion importance[1-10]: 5 __ Why: The error message could be clearer about whether maxRows is inclusive or exclusive. The condition allows max == maxRows, so the message should reflect this. However, this is a minor clarity improvement rather than a correctness issue.	Low
	Optimize key field index lookup The method uses fields.indexOf(keyField) inside a loop over keyFields, which is O(n*m) where n is the number of key fields and m is the number of fields. For large field lists, this can be inefficient. Consider building a Map<String, Integer> from fields to their indices once before the loop, reducing the complexity to O(n+m). 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(); + Map<String, Integer> fieldIndices = new java.util.HashMap<>(); + for (int i = 0; i < fields.size(); i++) { + fieldIndices.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 = fieldIndices.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"); } 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()); } Suggestion importance[1-10]: 5 __ Why: Using indexOf in a loop is O(n*m), which could be inefficient for large field lists. Building a map upfront reduces complexity to O(n+m). This is a valid optimization but only matters for large field counts, making it a moderate improvement.	Low

Suggestions up to commit 91bf200

Category	Suggestion	Impact
Possible issue	Fix off-by-one error 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 limit is enforced at maxRows + 1 instead of maxRows. Move the check before adding the row to enforce the exact limit. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [138-157] 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( - "outputlookup wrote nothing because the result has more than " - + maxRows - + " rows..."); - } } return rows; } Suggestion importance[1-10]: 9 __ Why: The check rows.size() > maxRows occurs after adding the row, allowing maxRows + 1 rows before throwing. This is a critical off-by-one bug that violates the stated limit. Moving the check before adding the row ensures the exact limit is enforced.	High
General	Validate maxRows is positive before use The validation checks max against maxRows, but maxRows could theoretically be less than 1 if misconfigured. Validate that maxRows itself is positive before using it as an upper bound to prevent confusing error messages. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [57-64] +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]: 5 __ Why: The suggestion correctly identifies that maxRows could theoretically be misconfigured. However, the setting definition at line 155-161 in OpenSearchSettings.java already enforces a minimum of 1, making this validation redundant in practice. Still, defensive validation improves robustness.	Low
	Prevent incorrect row count on flush failure If flushBatch() throws an exception, buffered has already been incremented but the batch may not have been fully processed. This could lead to incorrect row counts. Consider incrementing buffered only after successful batch addition or resetting it on flush failure. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [53-59] public void add(Object[] row) { ensureOpen(); batch.add(toIndexRequest(row)); - if (++buffered >= cfg.batchSize()) { + buffered++; + if (buffered >= cfg.batchSize()) { flushBatch(); } } Suggestion importance[1-10]: 3 __ Why: The suggestion raises a valid concern about exception handling, but the existing code already increments buffered before calling flushBatch(), and the 'improved_code' is identical to the 'existing_code'. The issue is minor since flushBatch() resets buffered to 0 at line 90, mitigating the concern.	Low

[github-actions]

PR Code Analyzer ❗

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

Path	Line	Severity	Description
opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java	285	medium	applyLoadSettings() sets replicas=0 and translog.durability=async on the target index before writing, but restoreServeSettings() is called outside the try-with-resources block. If an exception is thrown during bulk write or the subsequent refresh call, restoreServeSettings() is never called, permanently leaving the index with 0 replicas and async durability until manually corrected. This is not malicious but the pattern is unusual enough to warrant review — a deliberate sabotage of index durability settings would look exactly like this.
opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java	249	low	lookupFilter() builds an alias filter JSON string by direct string concatenation using the uuid value. While UUIDs produced by UUID.randomUUID() and UUID.nameUUIDFromBytes() are safe, the method accepts an arbitrary String parameter with no validation, so a caller passing a crafted uuid could inject content into the alias filter JSON. In the current call graph the inputs are always generated UUIDs, so exploitability is low.

The table above displays the top 10 most important findings.

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

---

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

[github-actions]

Persistent review updated to latest commit 995ed79

[github-actions]

Persistent review updated to latest commit 4ccbe20

[github-actions]

Persistent review updated to latest commit 7963644

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

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

nit: fix this import?

[songkant-aws]

nit: fix imports

[songkant-aws]

nit: fix imports

[songkant-aws]

nit: fix imports

[songkant-aws]

nit: fix imports

[songkant-aws]

It will never restore settings if writer or client throws exception

[songkant-aws]

nit: fix imports

[songkant-aws]

nit: fix imports

[songkant-aws]

It will create an empty index when there is an existing lookup index, right? Then the empty index is never used.

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

[github-actions]

Persistent review updated to latest commit 5dc4c9a