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

[noCharger]

Description

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

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

The command's semantics, substrate (plain index per lookup), consistency contract (weak/eventual), the <name> resolution/migration matrix, permission model, and alternatives considered are all covered in RFC #5625 and are not restated here — this PR is the implementation.

Implementation highlights:

• Terminal Calcite TableModify (INSERT) node — the optimizer treats it as a mandatory side effect (never dropped/reordered) and provides the rowcount row type.
• Full-result write with no truncation: the source scan pages unbounded via PIT.
• key_field upsert via a deterministic, collision-free _id; a multivalue key is rejected.
• Created lookup indices are tagged _meta.lookup; a non-lookup index is refused rather than clobbered.
• Backward-compatible with the Dashboards data importer (#11303): overwriting a filtered-alias lookup migrates it onto a dedicated index without deleting the shared index.
• Writes run under the caller's security context; no cluster-level permission.
• Tested by CalcitePPLOutputLookupIT, OutputLookupPermissionsIT, and unit tests.

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 a986700)

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 Interrupt Handling After catching InterruptedException and restoring the interrupt flag, the code throws BulkWriteException. If the caller does not check Thread.interrupted(), the interrupt may be lost. This can prevent timely thread shutdown in a multi-threaded environment where interruption is used for cancellation. } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new BulkWriteException("interrupted during bulk retry backoff", List.of()); } Unbounded Memory The execute method loads all input rows into a List in memory before writing. If the input is large (e.g., millions of rows) and max is not set, this can cause an OutOfMemoryError. The code should either enforce a reasonable default max or stream rows to the writer without buffering them all. List<Object[]> rows = new ArrayList<>(); while (input.moveNext()) { if (max != null && rows.size() >= max) { break; } Object cur = input.current(); rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); } Orphaned Slice When migrating a filtered alias, the code removes the alias but never deletes the orphaned __lookup slice from the shared index. Over time, repeated migrations can accumulate stale data in the shared index, wasting storage. The comment acknowledges this is out of scope, but it is a real resource leak that should be tracked or mitigated. removeAlias(client, name, target.aliasIndices()); createIndex(client, name, mapping);

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to a986700

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Close enumerator to prevent resource leak The enumerator input is never closed after draining, which may leak resources (e.g., open PIT contexts). Wrap the enumeration in a try-with-resources or explicitly close the enumerator after the loop completes. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [139-147] List<Object[]> rows = new ArrayList<>(); -while (input.moveNext()) { - if (max != null && rows.size() >= max) { - break; +try { + while (input.moveNext()) { + if (max != null && rows.size() >= max) { + break; + } + Object cur = input.current(); + rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); } - Object cur = input.current(); - rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); +} finally { + input.close(); } Suggestion importance[1-10]: 8 __ Why: The enumerator input is not explicitly closed after use, which could lead to resource leaks (e.g., open PIT contexts). Wrapping the enumeration in a try-finally block to ensure closure is a critical fix for preventing resource exhaustion.	Medium
	Add null check for row parameter The add method lacks null-safety checks for the row parameter. A null row would cause a NullPointerException in toIndexRequest. Add a null check at the method entry to fail fast with a clear error message. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [51-57] public void add(Object[] row) { ensureOpen(); + if (row == null) { + throw new IllegalArgumentException("row must not be null"); + } batch.add(toIndexRequest(row)); if (++buffered >= cfg.batchSize()) { flushBatch(); } } Suggestion importance[1-10]: 7 __ Why: Adding a null check for the row parameter prevents a potential NullPointerException in toIndexRequest. This is a valid defensive programming practice that improves robustness, though the likelihood of a null row being passed may be low in practice.	Medium
General	Add maximum retry iteration guard The infinite retry loop lacks a maximum iteration guard. If backoff.hasNext() unexpectedly always returns true due to a bug, the loop runs forever. Add a hard iteration limit as a safety net to prevent infinite loops. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [91-134] +int maxRetries = 10; +int retryCount = 0; while (true) { + if (retryCount++ > maxRetries) { + throw new BulkWriteException("outputlookup bulk write exceeded maximum retry attempts", List.of()); + } BulkResponse response = client.bulk(pending).actionGet(); - ... - if (!backoff.hasNext()) { - 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", exhausted); - } ... } Suggestion importance[1-10]: 6 __ Why: Adding a hard iteration limit to the retry loop is a good safety measure to prevent infinite loops in case of unexpected behavior. However, the BackoffPolicy.exponentialBackoff() iterator is expected to be finite, so this is more of a defensive safeguard than a critical fix.	Low
	Use case-insensitive field name validation The validation uses List.contains() for field name lookup, which is case-sensitive and may not match OpenSearch's field name normalization. Use case-insensitive comparison or normalize field names before validation to prevent false rejections. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [934-945] if (!node.getKeyFields().isEmpty()) { java.util.List<String> resultFields = child.getRowType().getFieldNames(); + java.util.Set<String> normalizedFields = resultFields.stream() + .map(String::toLowerCase) + .collect(java.util.stream.Collectors.toSet()); for (String keyField : node.getKeyFields()) { - if (!resultFields.contains(keyField)) { + if (!normalizedFields.contains(keyField.toLowerCase())) { throw new IllegalArgumentException( "outputlookup key_field [" + keyField + "] is not a field of the result; available fields: " + resultFields); } } } Suggestion importance[1-10]: 5 __ Why: The suggestion to use case-insensitive field name comparison is reasonable if OpenSearch normalizes field names. However, without confirmation that field names are case-insensitive in this context, this change could introduce false positives. The impact is moderate and depends on the actual field name handling semantics.	Low

---

Previous suggestions

Suggestions up to commit 7faebf1

Category	Suggestion	Impact
General	Stream rows instead of buffering all The code loads all rows into memory before writing, which can cause OutOfMemoryError for large result sets. Consider streaming rows directly to the writer instead of buffering them all in a list, especially since the writer already handles batching internally. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [137-145] -List<Object[]> rows = new ArrayList<>(); -while (input.moveNext()) { - if (max != null && rows.size() >= max) { - break; +long count = 0; +WriteConfig cfg = new WriteConfig(name, fields, mode, keyFields, BATCH_SIZE, RefreshPolicy.IMMEDIATE); +try (OpenSearchBulkWriter writer = new OpenSearchBulkWriter(client, cfg)) { + while (input.moveNext()) { + if (max != null && count >= max) { + break; + } + Object cur = input.current(); + Object[] row = cur instanceof Object[] arr ? arr : new Object[] {cur}; + writer.add(row); + count++; } - Object cur = input.current(); - rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); } Suggestion importance[1-10]: 9 __ Why: This is a critical improvement that addresses a potential OutOfMemoryError for large result sets. The current implementation buffers all rows in memory before writing, which is inefficient and risky. Streaming rows directly to the writer leverages its existing batching mechanism and significantly improves memory efficiency.	High
	Use ByteBuffer for integer serialization The intToBytes method uses unsigned right shift for the most significant byte, which is correct, but the length prefix could theoretically be negative if a malformed value is passed. Consider using ByteBuffer for safer and more maintainable integer serialization. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java [81-83] private static byte[] intToBytes(int v) { - return new byte[] {(byte) (v >>> 24), (byte) (v >>> 16), (byte) (v >>> 8), (byte) v}; + return java.nio.ByteBuffer.allocate(4).putInt(v).array(); } Suggestion importance[1-10]: 4 __ Why: While using ByteBuffer is more idiomatic and potentially clearer, the existing bit-shifting implementation is correct and efficient. The suggestion's concern about "negative length prefix" is unfounded since the method correctly handles all integer values. This is a minor style improvement rather than a bug fix.	Low
Possible issue	Validate key field is non-null and non-empty The validation checks if keyField exists in resultFields but does not verify that keyField itself is non-null or non-empty. A null or empty string in keyFields would pass validation but cause issues downstream. Add validation to reject null or empty key field names. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [936-947] if (!node.getKeyFields().isEmpty()) { java.util.List<String> resultFields = child.getRowType().getFieldNames(); for (String keyField : node.getKeyFields()) { + if (keyField == null || keyField.isEmpty()) { + throw new IllegalArgumentException("outputlookup key_field cannot be null or empty"); + } if (!resultFields.contains(keyField)) { throw new IllegalArgumentException( "outputlookup key_field [" + keyField + "] is not a field of the result; available fields: " + resultFields); } } } Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies that null or empty keyField values could pass validation but cause issues downstream. Adding this validation prevents potential bugs and provides clearer error messages to users.	Medium
	Add null check for row parameter The add method does not validate that row is non-null before processing it. A null row would cause a NullPointerException in toIndexRequest. Add a null check at the start of the method to fail fast with a clear error message. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [51-57] public void add(Object[] row) { ensureOpen(); + if (row == null) { + throw new IllegalArgumentException("row cannot be null"); + } batch.add(toIndexRequest(row)); if (++buffered >= cfg.batchSize()) { flushBatch(); } } Suggestion importance[1-10]: 7 __ Why: Adding a null check for the row parameter prevents a potential NullPointerException in toIndexRequest. This is a valid defensive programming practice that improves robustness, though the likelihood of a null row being passed may be low in practice.	Medium

Suggestions up to commit d39455c

Category	Suggestion	Impact
General	Stream rows instead of buffering Loading all rows into memory before writing could cause out-of-memory errors for large result sets, especially when max is null or very large. Consider streaming rows directly to the writer instead of buffering them all in a list. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [110-117] -List<Object[]> rows = new ArrayList<>(); +long rowCount = 0; while (input.moveNext()) { - if (max != null && rows.size() >= max) { + if (max != null && rowCount >= max) { break; } Object cur = input.current(); - rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); + Object[] row = cur instanceof Object[] arr ? arr : new Object[] {cur}; + // Write row immediately instead of buffering + writer.add(row); + rowCount++; } Suggestion importance[1-10]: 7 __ Why: Important memory optimization. Buffering all rows at lines 110-117 can cause OOM for large result sets. However, the current design requires the full row list for the overwrite logic (lines 131-152) to handle override_if_empty correctly. Streaming would require architectural changes to the overwrite path, making this a more complex refactoring than suggested.	Medium
	Optimize field lookup performance The indexOf call inside the loop has O(n) complexity, making the overall encoding O(k*n) where k is the number of key fields. For large field lists, this could become a performance bottleneck. Pre-compute a field-to-index map to achieve O(k) encoding time. 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> fieldIndexMap = new HashMap<>(); + for (int i = 0; i < fields.size(); i++) { + fieldIndexMap.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 = fieldIndexMap.get(keyField); + Object value = (idx != null && idx < row.length) ? row[idx] : MISSING; ... } return Base64.getUrlEncoder().withoutPadding().encodeToString(digest.digest()); } Suggestion importance[1-10]: 6 __ Why: Valid performance optimization. The indexOf call at line 36 is O(n) per key field, making the overall complexity O(k*n). Pre-computing a map reduces this to O(n + k). For typical use cases with moderate field counts, the impact is modest, but for large schemas this is a meaningful improvement.	Low
	Add retry limit safeguard The retry loop lacks a maximum iteration limit, which could cause indefinite blocking if the backoff policy unexpectedly continues. Add an explicit retry counter or timeout to prevent potential infinite loops even when the backoff iterator should be exhausted. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [84-136] private void flushBatch() { BulkRequest pending = batch; pending.setRefreshPolicy(cfg.refresh()); batch = new BulkRequest(); buffered = 0; Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); - while (true) { + int retryCount = 0; + int maxRetries = 10; // safety limit + while (retryCount < maxRetries) { BulkResponse response = client.bulk(pending).actionGet(); + retryCount++; ... } } Suggestion importance[1-10]: 3 __ Why: The suggestion adds a safety counter, but the existing code already has a built-in safeguard via backoff.hasNext() at line 122, which prevents infinite loops. The added counter provides marginal defensive value but is not critical.	Low
Possible issue	Add alias swap error handling The alias swap operation lacks error handling for the case where newBacking doesn't exist or the operation fails mid-execution. If the add action succeeds but a remove action fails, the alias could point to multiple indices. Wrap the operation in a try-catch and verify the final alias state. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [225-233] private static void swapAlias( NodeClient client, String alias, String newBacking, List<String> oldBackings) { IndicesAliasesRequest req = new IndicesAliasesRequest(); req.addAliasAction(AliasActions.add().index(newBacking).alias(alias)); for (String old : oldBackings) { req.addAliasAction(AliasActions.remove().index(old).alias(alias)); } - client.admin().indices().aliases(req).actionGet(); + try { + client.admin().indices().aliases(req).actionGet(); + } catch (Exception e) { + // Verify alias state and potentially rollback + throw new IllegalStateException("Alias swap failed for " + alias, e); + } } Suggestion importance[1-10]: 4 __ Why: The suggestion correctly identifies that error handling could be improved for the alias swap operation. However, the IndicesAliasesRequest is atomic (all actions succeed or all fail), so the concern about partial application is less severe. The try-catch adds defensive error messaging but doesn't fundamentally change correctness.	Low

Suggestions up to commit 4c9fb21

Category	Suggestion	Impact
Possible issue	Stream rows instead of buffering Loading all rows into memory before writing can cause out-of-memory errors for large result sets. Since OpenSearchBulkWriter already supports batched writes, stream rows directly to the writer instead of buffering them all in a list. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [110-117] -List<Object[]> rows = new ArrayList<>(); +long rowCount = 0; while (input.moveNext()) { - if (max != null && rows.size() >= max) { + if (max != null && rowCount >= max) { break; } Object cur = input.current(); - rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); + Object[] row = cur instanceof Object[] arr ? arr : new Object[] {cur}; + // Write directly or collect for empty-check if needed + rowCount++; } Suggestion importance[1-10]: 8 __ Why: Valid concern about memory usage for large result sets. The current code buffers all rows before writing, which can cause OOM. Streaming rows directly to the writer would be more memory-efficient, though the implementation needs careful handling of the overrideIfEmpty logic which requires knowing if the result is empty before starting writes.	Medium
	Handle exceptions before incrementing counter The add method does not handle exceptions from toIndexRequest, which could throw during document ID encoding (e.g., multivalue key field). If an exception occurs after incrementing buffered but before adding to the batch, the counter becomes inconsistent with the actual batch size, potentially causing incorrect flush behavior or lost rows. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [51-56] public void add(Object[] row) { ensureOpen(); - batch.add(toIndexRequest(row)); + IndexRequest request = toIndexRequest(row); + batch.add(request); if (++buffered >= cfg.batchSize()) { flushBatch(); } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that toIndexRequest could throw (e.g., multivalue key field rejection), and extracting it before incrementing buffered prevents counter inconsistency. However, the impact is moderate since the exception would still propagate and fail the write, just with a cleaner state.	Medium
General	Prevent exception masking during cleanup If the deleteIndex call in the finally block throws an exception, it will mask the original exception that caused the write or swap to fail. Wrap the cleanup in a try-catch to ensure the original failure is not suppressed. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [143-147] } finally { if (!swapped) { - deleteIndex(client, backing); + try { + deleteIndex(client, backing); + } catch (Exception cleanupEx) { + // Log but don't mask the original exception + } } } Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that an exception during cleanup could mask the original failure. Wrapping the cleanup in try-catch is a good practice, though the impact is moderate since cleanup failures are less common and the original exception would still be visible in most cases.	Low
	Optimize key field validation performance The validation uses contains() on a list, which has O(n) complexity per lookup. For queries with many key fields and large result schemas, this becomes O(n*m). Convert resultFields to a HashSet for O(1) lookup performance. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [936-947] if (!node.getKeyFields().isEmpty()) { - java.util.List<String> resultFields = child.getRowType().getFieldNames(); + java.util.Set<String> resultFields = new java.util.HashSet<>(child.getRowType().getFieldNames()); for (String keyField : node.getKeyFields()) { if (!resultFields.contains(keyField)) { throw new IllegalArgumentException( "outputlookup key_field [" + keyField + "] is not a field of the result; available fields: " + resultFields); } } } Suggestion importance[1-10]: 4 __ Why: The optimization from O(n*m) to O(m) is valid, but the practical impact is minimal since key field counts and result schema sizes are typically small in real queries. The improvement is correct but offers marginal benefit for the typical use case.	Low

Suggestions up to commit 92cfd5f

Category	Suggestion	Impact
General	Optimize field lookup performance Using indexOf for each key field in every row is inefficient for large datasets. Pre-compute a map from field names to indices once during initialization to avoid repeated linear searches through the fields list on every row. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java [34-37] +// Pre-compute field indices map (do this once, not per-row) +Map<String, Integer> fieldIndices = new HashMap<>(); +for (int i = 0; i < fields.size(); i++) { + fieldIndices.put(fields.get(i), i); +} + +// Then in the loop: 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; Suggestion importance[1-10]: 7 __ Why: Valid performance optimization. Using indexOf in a loop for each row is O(n*m) where n is rows and m is fields. Pre-computing a map reduces this to O(n). This is a meaningful improvement for large datasets.	Medium
	Handle empty overwrite case explicitly When rows.isEmpty() and overrideIfEmpty is true, the code creates a fresh backing index, writes zero rows, swaps the alias, and deletes old backings. This leaves an empty backing index aliased. Consider explicitly handling this case to avoid unnecessary index creation or document the intended behavior. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [131-152] -if (rows.isEmpty() && !overrideIfEmpty) { +if (rows.isEmpty()) { + if (!overrideIfEmpty) { + return 0; + } + // Empty result with overrideIfEmpty=true: clear the lookup by creating empty backing + String backing = freshBacking(alias); + createBacking(client, backing, mapping); + swapAlias(client, alias, backing, oldBackings); + for (String old : oldBackings) { + deleteIndex(client, old); + } return 0; } String backing = freshBacking(alias); createBacking(client, backing, mapping); -boolean swapped = false; -try { - writeRows(client, backing, fields, mode, keyFields, rows); - swapAlias(client, alias, backing, oldBackings); - swapped = true; -} finally { - if (!swapped) { - deleteIndex(client, backing); - } -} +... Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that empty results with overrideIfEmpty=true create an empty backing index. The proposed refactoring improves code clarity by explicitly handling this case, though the current implementation is functionally correct.	Low
	Use mutable list for keyFields The keyFields field is mutable (has a setter) but initialized with an immutable empty list from List.of(). If setKeyFields is called with a new list, this works, but direct modification attempts would fail. Consider using new ArrayList<>() for consistency with the mutable design. core/src/main/java/org/opensearch/sql/ast/tree/OutputLookup.java [42-43] /** Fields whose values form the document {@code _id} for upsert; empty means auto-generated id. */ -private List<String> keyFields = java.util.List.of(); +private List<String> keyFields = new ArrayList<>(); Suggestion importance[1-10]: 4 __ Why: The suggestion identifies a potential inconsistency: keyFields is initialized with an immutable list but has a setter. However, the setter replaces the entire list (line 43 shows private List<String> keyFields = java.util.List.of();), so direct modification isn't expected. Using ArrayList would be more conventional for mutable fields.	Low
	Remove unused tolerated failures list The tolerated list is initialized but never populated in the code. Since the class raises BulkWriteException on any non-retryable failure rather than tolerating it, this field appears unused. Either remove it or document its intended future use to avoid confusion. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [39-43] private BulkRequest batch = new BulkRequest(); private int buffered = 0; private long written = 0; -private final List<ItemFailure> tolerated = new ArrayList<>(); private boolean closed = false; Suggestion importance[1-10]: 3 __ Why: The tolerated list is initialized but never populated, as the code raises exceptions on failures. However, it's returned in flush() at line 65, so it's not entirely unused. The suggestion is partially incorrect but highlights potential dead code.	Low

Suggestions up to commit 9476e41

Category	Suggestion	Impact
General	Clean up orphaned backing index When overrideIfEmpty is false and rows are empty, the method returns early without cleaning up resources. However, if rows are not empty but an exception occurs during write or swap, the newly created backing index is orphaned and never deleted, causing index pollution. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [113-121] if (rows.isEmpty() && !overrideIfEmpty) { return 0; } String backing = freshBacking(alias); createBacking(client, backing, mapping); -writeRows(client, backing, fields, mode, keyFields, rows); -swapAlias(client, alias, backing, oldBackings); +try { + writeRows(client, backing, fields, mode, keyFields, rows); + swapAlias(client, alias, backing, oldBackings); +} catch (Exception e) { + deleteIndex(client, backing); + throw e; +} Suggestion importance[1-10]: 9 __ Why: This is a critical resource leak issue. If an exception occurs after creating the backing index but before the alias swap completes, the orphaned index will remain in the cluster indefinitely, causing index pollution and wasting resources.	High
	Handle flush exceptions properly The add method does not handle exceptions during flushBatch, which could leave the writer in an inconsistent state with buffered incremented but the batch not properly flushed. Wrap the flush in a try-catch to ensure proper error handling and state consistency. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java [51-57] public void add(Object[] row) { ensureOpen(); batch.add(toIndexRequest(row)); - if (++buffered >= cfg.batchSize()) { - flushBatch(); + buffered++; + if (buffered >= cfg.batchSize()) { + try { + flushBatch(); + } catch (Exception e) { + buffered = 0; + batch = new BulkRequest(); + throw e; + } } } Suggestion importance[1-10]: 6 __ Why: The suggestion addresses a potential state inconsistency issue where buffered is incremented before flushBatch() is called. However, the impact is moderate since exceptions during flush would likely propagate and prevent further operations anyway.	Low
Possible issue	Stream rows instead of buffering Buffering all rows in memory before writing can cause out-of-memory errors for large result sets. Consider streaming rows directly to the bulk writer instead of collecting them into a list first, especially since the bulk writer already handles batching internally. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java [93-100] -List<Object[]> rows = new ArrayList<>(); -while (input.moveNext()) { - if (max != null && rows.size() >= max) { - break; - } - Object cur = input.current(); - rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); +long rowCount = 0; +String backing = append ? resolveBacking(client, alias) : freshBacking(alias); +if (backing == null) { + backing = freshBacking(alias); + createBacking(client, backing, mapping); + swapAlias(client, alias, backing, List.of()); } +WriteConfig cfg = new WriteConfig(backing, fields, mode, keyFields, BATCH_SIZE, RefreshPolicy.IMMEDIATE); +try (OpenSearchBulkWriter writer = new OpenSearchBulkWriter(client, cfg)) { + while (input.moveNext()) { + if (max != null && rowCount >= max) { + break; + } + Object cur = input.current(); + Object[] row = cur instanceof Object[] arr ? arr : new Object[] {cur}; + writer.add(row); + rowCount++; + } +} + Suggestion importance[1-10]: 8 __ Why: Buffering all rows in memory before writing can cause out-of-memory errors for large datasets. The suggestion to stream rows directly to the bulk writer is valid and addresses a significant scalability issue, especially since the bulk writer already handles batching.	Medium
Security	Validate against internal index patterns The validation only checks for dot-prefixed system indices but does not prevent writing to other protected or internal indices. Consider adding validation against a comprehensive list of reserved index patterns to prevent accidental writes to critical system resources. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [922-927] String indexName = node.getIndexName(); -if (indexName.startsWith(".")) { +if (indexName.startsWith(".") || indexName.startsWith("_")) { throw new IllegalArgumentException( "outputlookup destination [" + indexName - + "] must not be a system (dot-prefixed) index"); + + "] must not be a system or internal index"); } Suggestion importance[1-10]: 7 __ Why: The suggestion improves security by preventing writes to underscore-prefixed internal indices in addition to dot-prefixed system indices. This is a reasonable enhancement to protect critical system resources, though the current validation already covers the most common system indices.	Medium

[github-actions]

PR Code Analyzer ❗

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

Path	Line	Severity	Description
opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java	110	medium	The outputlookup command grants authenticated PPL users unrestricted write, create, and delete access to OpenSearch indices with no PPL-layer authorization check beyond the dot-prefix guard. If the prior security model assumed PPL was read-only, this represents a privilege escalation path: any PPL-capable user can now overwrite or destroy arbitrary non-system indices.
core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java	924	medium	The system-index guard only blocks dot-prefixed index names. Non-dot-prefixed but operationally sensitive indices (e.g., audit logs, ILM-managed indices, security configuration indices) receive no protection. A malicious or misconfigured query could silently overwrite or truncate them via the overwrite-by-default semantics.
opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java	47	low	By design, a null key-field value and a missing key field both encode to the same canonical byte 'N', causing them to share the same document _id. In upsert mode this silently merges documents whose key fields differ only in presence vs. null, which can corrupt lookup data in ways that are hard to diagnose.

The table above displays the top 10 most important findings.

Total: 3 | Critical: 0 | High: 0 | Medium: 2 | 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 7faebf1

[github-actions]

Persistent review updated to latest commit a986700