[Feature] Add PPL collect command

[noCharger]

Terminal pass-through write: appends pipeline rows to a pre-existing index and
returns them as the result.

• Grammar (ppl / language-grammar / async-query-core) + Collect AST + AstBuilder + anonymizer
• visitCollect builds LogicalTableSpool; EnumerableOpenSearchTableSpool + converter rule do the eager-drain batched-bulk write (429 retry/backoff) with pass-through output
• Options: source/host/sourcetype/marker stamps + testmode dry-run
• Plan-time safety: destination pre-existence + dot/hidden-index refusal
• Tests: CalcitePPLCollectTest, CollectWriteStrategyTest, NewAddedCommandsIT, CalcitePPLCollectIT
• User doc: docs/user/ppl/cmd/collect.md

Description

#5596

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

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 Code Analyzer ❗

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

Path	Line	Severity	Description
docs/user/ppl/admin/security.md	15	medium	The collect command's async fire-and-forget design intentionally allows the foreground query to succeed and return preview rows even when the background materialization task lacks write or materialize permissions. Authorization failures are recorded only on the background task observable via _tasks API, not surfaced synchronously. While documented, this means a caller cannot reliably distinguish an authorized write from a silently-denied one from the query response alone, weakening the security posture for callers that depend on write confirmation.

The table above displays the top 10 most important findings.

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

---

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]

PR Reviewer Guide 🔍

(Review updated until commit f50378f)

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 PassThroughWriter drains the entire input eagerly on first moveNext() call, but if the client stops consuming results early (e.g., due to an error or early termination), the drain continues to completion and writes all rows anyway. This means a client that fetches only the first page of results still triggers a full write of potentially millions of rows, which may not be the intended behavior for a "pass-through" operation where the write should arguably stop when consumption stops. @Override public boolean moveNext() { if (!drained) { drainAndWrite(); drained = true; } if (passThrough.hasNext()) { current = passThrough.next(); return true; } return false; } Possible Issue The bulkWithRetry method retries only 429 (TOO_MANY_REQUESTS) failures but silently ignores all other per-item failures after the retry loop exhausts or when no 429s remain. This means non-retriable write failures (e.g., mapper_parsing_exception, version_conflict) are swallowed without surfacing to the caller or logging, so rows may be silently dropped without any indication that the write was incomplete. private static void bulkWithRetry(OpenSearchClient client, BulkRequest request) { java.util.Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); BulkRequest pending = request; while (true) { BulkResponse response = client.bulk(pending); if (!response.hasFailures()) { return; } BulkRequest retry = new BulkRequest(); for (BulkItemResponse item : response.getItems()) { if (item.isFailed() && item.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS) { retry.add(pending.requests().get(item.getItemId())); } } if (retry.numberOfActions() == 0 || !backoff.hasNext()) { return; } try { Thread.sleep(backoff.next().millis()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } pending = retry; } } Possible Issue The isIndexNotFound method uses string matching on exception messages ("no such index", "IndexNotFound") to detect index-not-found conditions. This is fragile because exception messages can change across OpenSearch versions or be localized, causing the detection to fail and the wrong exception to propagate (e.g., a raw RuntimeException instead of the intended destinationDoesNotExist error). private static boolean isIndexNotFound(Throwable e) { for (Throwable t = e; t != null; t = t.getCause()) { if ("IndexNotFoundException".equals(t.getClass().getSimpleName())) { return true; } String msg = t.getMessage(); if (msg != null && (msg.contains("no such index") || msg.contains("IndexNotFound"))) { return true; } } return false; } Possible Issue The submitAsyncCollect method checks if the destination index exists using clusterServiceRef.state().routingTable().hasIndex(destIndex), but this check is not atomic with the subsequent write operation. If the index is deleted between the check and the background task execution, the background task will fail with an index-not-found error that was supposed to be caught synchronously. This race condition means the plan-time safety guarantee can be violated under concurrent index deletion. if (!clusterServiceRef.state().routingTable().hasIndex(destIndex)) { listener.onFailure( org.opensearch.sql.calcite.CollectValidation.destinationDoesNotExist(destIndex)); return;

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to f50378f

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Surface non-retriable bulk write failures The retry logic silently discards non-429 bulk failures by returning without throwing an exception. This means mapper parsing errors or other permanent failures are lost, contradicting the test testCollectSurfacesNonRetriableBulkFailure which expects failures to surface. The method should throw an exception when non-retriable failures remain after exhausting retries. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [187-212] private static void bulkWithRetry(OpenSearchClient client, BulkRequest request) { java.util.Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); BulkRequest pending = request; while (true) { BulkResponse response = client.bulk(pending); if (!response.hasFailures()) { return; } BulkRequest retry = new BulkRequest(); + boolean hasNonRetriableFailure = false; for (BulkItemResponse item : response.getItems()) { - if (item.isFailed() && item.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS) { - retry.add(pending.requests().get(item.getItemId())); + if (item.isFailed()) { + if (item.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS) { + retry.add(pending.requests().get(item.getItemId())); + } else { + hasNonRetriableFailure = true; + } } } - if (retry.numberOfActions() == 0 || !backoff.hasNext()) { + if (retry.numberOfActions() == 0) { + if (hasNonRetriableFailure) { + throw new IllegalStateException("Bulk write failed with non-retriable errors: " + response.buildFailureMessage()); + } return; + } + if (!backoff.hasNext()) { + throw new IllegalStateException("Bulk write failed after exhausting retries: " + response.buildFailureMessage()); } ... } } Suggestion importance[1-10]: 9 __ Why: This is a critical correctness issue. The current code silently discards non-429 failures, contradicting the test testCollectSurfacesNonRetriableBulkFailure which expects failures to surface. The improved code properly throws exceptions for non-retriable errors, ensuring failures are observable.	High
	Handle bulk write exceptions properly The flush() method does not handle exceptions from bulkWithRetry, which could leave the writer in an inconsistent state if a non-retriable error occurs. Consider wrapping the bulk operation in a try-catch block to ensure proper error propagation and resource cleanup, preventing silent failures during the write operation. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [177-181] private void flush() { - bulkWithRetry(client, bulk); - bulk = new BulkRequest(); - buffered = 0; + try { + bulkWithRetry(client, bulk); + } finally { + bulk = new BulkRequest(); + buffered = 0; + } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that exceptions from bulkWithRetry could leave state inconsistent. Using a finally block ensures bulk and buffered are reset even on failure, improving robustness. However, the impact is moderate since the writer is single-use per drain.	Medium
General	Validate destination index existence robustly The destination index existence check using hasIndex() may produce false negatives for aliases or data streams, which are valid write targets but may not appear in the routing table the same way as concrete indices. Consider using a more robust resolution mechanism that handles aliases and data streams correctly to avoid incorrectly rejecting valid destinations. plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java [259-266] if (destIndex.startsWith(".")) { listener.onFailure(org.opensearch.sql.calcite.CollectValidation.dotIndexRefused(destIndex)); return; } -if (!clusterServiceRef.state().routingTable().hasIndex(destIndex)) { +try { + clusterServiceRef.state().metadata().index(destIndex); +} catch (Exception e) { listener.onFailure( org.opensearch.sql.calcite.CollectValidation.destinationDoesNotExist(destIndex)); return; } Suggestion importance[1-10]: 6 __ Why: The suggestion raises a valid concern about hasIndex() potentially missing aliases or data streams. However, the improved code using metadata().index() may not be the complete solution either, and the current implementation may already handle these cases correctly through the later getTableForMember call in CalciteRelNodeVisitor.	Low
	Ensure resource cleanup on drain failure The drainAndWrite() method does not handle exceptions during the drain loop, which could leave the bulk request partially filled and the source enumerator in an inconsistent state. If flush() or toIndexRequest() throws an exception mid-drain, resources may leak. Wrap the drain logic in a try-finally block to ensure proper cleanup of the source enumerator. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [159-175] private void drainAndWrite() { List<@Nullable Object> passThroughBuffer = new ArrayList<>(); - while (source.moveNext()) { - Object row = source.current(); - bulk.add(toIndexRequest(row)); - if (++buffered >= BATCH_SIZE) { + try { + while (source.moveNext()) { + Object row = source.current(); + bulk.add(toIndexRequest(row)); + if (++buffered >= BATCH_SIZE) { + flush(); + } + if (passThroughBuffer.size() < passThroughCap) { + passThroughBuffer.add(row); + } + } + if (buffered > 0) { flush(); } - if (passThroughBuffer.size() < passThroughCap) { - passThroughBuffer.add(row); - } + } finally { + passThrough = passThroughBuffer.iterator(); } - if (buffered > 0) { - flush(); - } - passThrough = passThroughBuffer.iterator(); } Suggestion importance[1-10]: 5 __ Why: While the suggestion to use finally for setting passThrough is reasonable for defensive programming, the actual impact is limited. If an exception occurs, the enumerator is likely unusable anyway, and the passThrough iterator would simply be empty. The improvement is marginal.	Low

---

Previous suggestions

Suggestions up to commit c339218

Category	Suggestion	Impact
Possible issue	Surface non-retriable bulk write failures The retry logic silently discards non-429 bulk failures after exhausting retries. When retry.numberOfActions() == 0 (all failures are non-retriable) or !backoff.hasNext() (retries exhausted), the method returns without surfacing the failure. This causes data loss: rows pass through to the client as if written, but the write silently failed. Throw an exception when failures remain after retry exhaustion so the caller can handle or propagate the error. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [187-212] private static void bulkWithRetry(OpenSearchClient client, BulkRequest request) { java.util.Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); BulkRequest pending = request; while (true) { BulkResponse response = client.bulk(pending); if (!response.hasFailures()) { return; } BulkRequest retry = new BulkRequest(); for (BulkItemResponse item : response.getItems()) { if (item.isFailed() && item.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS) { retry.add(pending.requests().get(item.getItemId())); } } if (retry.numberOfActions() == 0 || !backoff.hasNext()) { + if (response.hasFailures()) { + throw new IllegalStateException("Bulk write failed with non-retriable errors: " + response.buildFailureMessage()); + } return; } ... } } Suggestion importance[1-10]: 9 __ Why: Critical correctness issue: the retry logic silently discards non-retriable bulk failures, causing data loss where rows appear written to the client but actually failed. The suggestion correctly identifies that an exception should be thrown when failures remain after retry exhaustion.	High
General	Document pass-through preview cap behavior The pass-through buffer is capped at passThroughCap but the write drains the entire source. If the source exceeds passThroughCap, rows beyond the cap are written but never returned to the client, creating a mismatch between what the client sees and what was written. This violates the pass-through contract. Consider documenting this behavior explicitly or ensuring the client is aware that only a preview is returned when the source is large. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [159-175] private void drainAndWrite() { List<@Nullable Object> passThroughBuffer = new ArrayList<>(); while (source.moveNext()) { Object row = source.current(); bulk.add(toIndexRequest(row)); if (++buffered >= BATCH_SIZE) { flush(); } + // Cap pass-through at QUERY_SIZE_LIMIT; rows beyond are written but not returned (preview). if (passThroughBuffer.size() < passThroughCap) { passThroughBuffer.add(row); } } if (buffered > 0) { flush(); } passThrough = passThroughBuffer.iterator(); } Suggestion importance[1-10]: 5 __ Why: Valid observation about the pass-through cap behavior, but this is already documented in the RFC and integration tests. The suggestion to add an inline comment is reasonable for code clarity, though the behavior is intentional and documented elsewhere.	Low
	Verify dot-prefix check handles edge cases The dot-prefix check uses startsWith(".") which only catches indices starting with a single dot. Hidden indices can have multiple leading dots (e.g., ..hidden). Use a more robust check that handles all dot-prefixed patterns to ensure system/hidden indices are consistently refused. plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java [259-266] +if (destIndex.startsWith(".")) { + listener.onFailure(org.opensearch.sql.calcite.CollectValidation.dotIndexRefused(destIndex)); + return; +} +if (!clusterServiceRef.state().routingTable().hasIndex(destIndex)) { + listener.onFailure( + org.opensearch.sql.calcite.CollectValidation.destinationDoesNotExist(destIndex)); + return; +} - Suggestion importance[1-10]: 3 __ Why: The concern about multiple leading dots (e.g., ..hidden) is theoretically valid, but startsWith(".") already catches all dot-prefixed indices including those with multiple dots. The improved_code is identical to existing_code, indicating no actual change is needed. The suggestion is more of a verification request than a concrete improvement.	Low

Suggestions up to commit 41bb297

Category	Suggestion	Impact
Possible issue	Initialize passThrough in finally block The drainAndWrite method does not handle exceptions during bulk write operations. If flush() throws an exception mid-drain, the passThroughBuffer may be incomplete and passThrough will never be initialized, causing a NullPointerException in moveNext(). Wrap the drain loop in a try-finally block to ensure passThrough is always initialized, even if the write fails. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [159-175] private void drainAndWrite() { List<@Nullable Object> passThroughBuffer = new ArrayList<>(); - while (source.moveNext()) { - Object row = source.current(); - bulk.add(toIndexRequest(row)); - if (++buffered >= BATCH_SIZE) { + try { + while (source.moveNext()) { + Object row = source.current(); + bulk.add(toIndexRequest(row)); + if (++buffered >= BATCH_SIZE) { + flush(); + } + if (passThroughBuffer.size() < passThroughCap) { + passThroughBuffer.add(row); + } + } + if (buffered > 0) { flush(); } - if (passThroughBuffer.size() < passThroughCap) { - passThroughBuffer.add(row); - } + } finally { + passThrough = passThroughBuffer.iterator(); } - if (buffered > 0) { - flush(); - } - passThrough = passThroughBuffer.iterator(); } Suggestion importance[1-10]: 8 __ Why: If flush() throws an exception mid-drain, passThrough remains null, causing a NullPointerException in moveNext(). The finally block ensures passThrough is always initialized, preventing a critical runtime failure.	Medium
	Surface non-retriable bulk write failures The retry logic silently returns when non-429 failures occur or backoff is exhausted, discarding the failure information. This makes it impossible to detect partial write failures. Instead, throw an exception when non-retriable failures remain or retry budget is exhausted, so the caller can handle or surface the error appropriately. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [187-212] private static void bulkWithRetry(OpenSearchClient client, BulkRequest request) { java.util.Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); BulkRequest pending = request; while (true) { BulkResponse response = client.bulk(pending); if (!response.hasFailures()) { return; } BulkRequest retry = new BulkRequest(); for (BulkItemResponse item : response.getItems()) { if (item.isFailed() && item.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS) { retry.add(pending.requests().get(item.getItemId())); } } - if (retry.numberOfActions() == 0 || !backoff.hasNext()) { - return; + if (retry.numberOfActions() == 0) { + throw new IllegalStateException("Bulk write failed with non-retriable errors: " + response.buildFailureMessage()); + } + if (!backoff.hasNext()) { + throw new IllegalStateException("Bulk write retry exhausted after backoff"); } ... } } Suggestion importance[1-10]: 7 __ Why: The current implementation silently discards non-429 failures and exhausted retries, making it impossible to detect partial write failures. Throwing an exception improves observability and allows proper error handling, though the impact depends on whether silent failures are acceptable for this use case.	Medium
General	Verify destination index is open The pre-existence check uses hasIndex() which only verifies the index exists in the routing table, but does not confirm the index is open and writable. A closed index will pass this check but fail at write time. Add a check to ensure the index state is open before submitting the background task. plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java [282-291] if (!clusterServiceRef.state().routingTable().hasIndex(destIndex)) { listener.onFailure( new SemanticCheckException( String.format( Locale.ROOT, "collect destination index [%s] does not exist; collect requires a pre-existing" + " destination index", destIndex))); return; } +if (clusterServiceRef.state().metadata().index(destIndex).getState() != IndexMetadata.State.OPEN) { + listener.onFailure( + new SemanticCheckException( + String.format( + Locale.ROOT, + "collect destination index [%s] is not open", + destIndex))); + return; +} Suggestion importance[1-10]: 6 __ Why: The current check only verifies the index exists in the routing table but not whether it's open and writable. A closed index would pass this check but fail at write time. Adding an open-state check improves early error detection, though the write would eventually fail anyway.	Low
	Validate row-schema field count match The method silently skips fields when values.length < fields.size(), which can occur if the row shape mismatches the schema. This produces incomplete documents without any indication of data loss. Add validation to ensure the row has the expected number of fields, or log a warning when fields are skipped due to length mismatch. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [214-224] private IndexRequest toIndexRequest(@Nullable Object row) { Object[] values = (row instanceof Object[] arr) ? arr : new Object[] {row}; + if (values.length < fields.size()) { + throw new IllegalStateException( + String.format("Row has %d values but schema expects %d fields", values.length, fields.size())); + } Map<String, Object> doc = new LinkedHashMap<>(); for (int i = 0; i < fields.size() && i < values.length; i++) { String name = fields.get(i); if (values[i] != null && !OpenSearchIndex.METADATAFIELD_TYPE_MAP.containsKey(name)) { doc.put(name, values[i]); } } return new IndexRequest(indexName).source(doc); } Suggestion importance[1-10]: 5 __ Why: The method silently skips fields when values.length < fields.size(), potentially producing incomplete documents. Adding validation or logging improves data integrity awareness, though the current behavior may be intentional for handling variable-length rows.	Low

Suggestions up to commit 6098080

Category	Suggestion	Impact
Possible issue	Surface non-retryable bulk write failures The retry logic silently discards non-429 failures after exhausting retries. Critical write failures (permission errors, mapping conflicts) should be surfaced to the caller rather than silently ignored, as they indicate data loss. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [187-212] private static void bulkWithRetry(OpenSearchClient client, BulkRequest request) { java.util.Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); BulkRequest pending = request; while (true) { BulkResponse response = client.bulk(pending); if (!response.hasFailures()) { return; } BulkRequest retry = new BulkRequest(); + boolean hasNonRetryableFailures = false; for (BulkItemResponse item : response.getItems()) { - if (item.isFailed() && item.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS) { - retry.add(pending.requests().get(item.getItemId())); + if (item.isFailed()) { + if (item.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS) { + retry.add(pending.requests().get(item.getItemId())); + } else { + hasNonRetryableFailures = true; + } } + } + if (hasNonRetryableFailures) { + throw new IllegalStateException("Bulk write failed with non-retryable errors: " + response.buildFailureMessage()); } if (retry.numberOfActions() == 0 || !backoff.hasNext()) { return; } ... } } Suggestion importance[1-10]: 9 __ Why: The current implementation silently discards non-429 failures (permission errors, mapping conflicts), which can lead to silent data loss. This is a critical issue that should be surfaced to the caller.	High
	Flush pending writes on exception If source.moveNext() throws an exception mid-stream, the partially buffered bulk request is never flushed, causing silent data loss. Wrap the drain loop in try-finally to ensure pending writes are flushed even on failure. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [159-175] private void drainAndWrite() { List<@Nullable Object> passThroughBuffer = new ArrayList<>(); - while (source.moveNext()) { - Object row = source.current(); - bulk.add(toIndexRequest(row)); - if (++buffered >= BATCH_SIZE) { + try { + while (source.moveNext()) { + Object row = source.current(); + bulk.add(toIndexRequest(row)); + if (++buffered >= BATCH_SIZE) { + flush(); + } + if (passThroughBuffer.size() < passThroughCap) { + passThroughBuffer.add(row); + } + } + } finally { + if (buffered > 0) { flush(); } - if (passThroughBuffer.size() < passThroughCap) { - passThroughBuffer.add(row); - } - } - if (buffered > 0) { - flush(); } passThrough = passThroughBuffer.iterator(); } Suggestion importance[1-10]: 8 __ Why: If source.moveNext() throws an exception mid-stream, the partially buffered bulk request is never flushed, causing potential data loss. The try-finally pattern ensures pending writes are flushed even on failure.	Medium
Security	Strengthen system index validation The dot-prefix check only validates the first character. Malicious or malformed index names like foo.bar or ..hidden could bypass this check. Use a more robust validation that checks if the entire index name represents a system/hidden index pattern. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [810-816] -if (node.getIndexName().startsWith(".")) { +String indexName = node.getIndexName(); +if (indexName.startsWith(".") || indexName.contains("/.")) { throw new SemanticCheckException( String.format( "collect cannot write to system or hidden index [%s]; dot-prefixed indices are" + " refused", - node.getIndexName())); + indexName)); } Suggestion importance[1-10]: 3 __ Why: The suggested check for contains("/.") is incorrect for OpenSearch index names, which don't contain slashes. The existing startsWith(".") check is sufficient for detecting dot-prefixed system/hidden indices.	Low

Suggestions up to commit 9293d6b

Category	Suggestion	Impact
Possible issue	Add maximum retry limit The infinite while (true) loop lacks a maximum retry limit, which could cause the method to retry indefinitely if backoff exhaustion doesn't terminate it. Add an explicit retry counter to prevent unbounded retries and ensure the method eventually exits even under persistent failures. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [187-212] private static void bulkWithRetry(OpenSearchClient client, BulkRequest request) { java.util.Iterator<TimeValue> backoff = BackoffPolicy.exponentialBackoff().iterator(); BulkRequest pending = request; - while (true) { + int maxRetries = 10; + int retryCount = 0; + while (retryCount < maxRetries) { BulkResponse response = client.bulk(pending); if (!response.hasFailures()) { return; } BulkRequest retry = new BulkRequest(); for (BulkItemResponse item : response.getItems()) { if (item.isFailed() && item.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS) { retry.add(pending.requests().get(item.getItemId())); } } if (retry.numberOfActions() == 0 || !backoff.hasNext()) { return; } try { Thread.sleep(backoff.next().millis()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } pending = retry; + retryCount++; } } Suggestion importance[1-10]: 7 __ Why: The infinite while (true) loop relies solely on backoff.hasNext() to terminate, which may not provide a hard upper bound on retries. Adding an explicit maxRetries counter improves robustness by ensuring the method eventually exits even if the backoff policy doesn't terminate as expected, preventing unbounded retries under persistent failures.	Medium
General	Validate row-schema field count match When values.length is less than fields.size(), the loop silently skips trailing fields without logging or validation. This could mask data loss if the row structure doesn't match the expected schema. Add validation to detect and handle schema mismatches explicitly. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/EnumerableOpenSearchTableSpool.java [214-224] private IndexRequest toIndexRequest(@Nullable Object row) { Object[] values = (row instanceof Object[] arr) ? arr : new Object[] {row}; + if (values.length < fields.size()) { + throw new IllegalStateException( + String.format("Row has %d values but schema expects %d fields", values.length, fields.size())); + } Map<String, Object> doc = new LinkedHashMap<>(); for (int i = 0; i < fields.size() && i < values.length; i++) { String name = fields.get(i); if (values[i] != null && !OpenSearchIndex.METADATAFIELD_TYPE_MAP.containsKey(name)) { doc.put(name, values[i]); } } return new IndexRequest(indexName).source(doc); } Suggestion importance[1-10]: 5 __ Why: The suggestion adds validation to detect when values.length is less than fields.size(), which could indicate a schema mismatch. However, the current code already handles this gracefully by iterating only up to the minimum of the two lengths (i < fields.size() && i < values.length), so the validation may be overly strict and could reject valid cases where trailing fields are intentionally absent.	Low
	Validate index name before prefix check The dot-prefix check only validates the first character but doesn't handle edge cases like empty strings or null values. Add explicit null/empty validation before the prefix check to prevent potential NullPointerException or unexpected behavior. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [810-816] -if (node.getIndexName().startsWith(".")) { +String indexName = node.getIndexName(); +if (indexName == null || indexName.isEmpty()) { + throw new SemanticCheckException("collect destination index name cannot be null or empty"); +} +if (indexName.startsWith(".")) { throw new SemanticCheckException( String.format( "collect cannot write to system or hidden index [%s]; dot-prefixed indices are" + " refused", - node.getIndexName())); + indexName)); } Suggestion importance[1-10]: 4 __ Why: The suggestion adds null/empty validation for node.getIndexName() before the dot-prefix check. While this improves defensive programming, the indexName is already validated earlier in the code path (it's a required field in the Collect AST node and is resolved through the catalog), so a null or empty value at this point would indicate a deeper bug rather than a user input issue.	Low

[github-actions]

Persistent review updated to latest commit 6098080

[github-actions]

Persistent review updated to latest commit 41bb297

[github-actions]

Persistent review updated to latest commit c339218

[github-actions]

Persistent review updated to latest commit f50378f