Add cluster command for PPL

[ritvibhatt]

Description

Description

Adds the cluster command to PPL, which groups documents into clusters based on text similarity. More info on decisions made in #5255

Syntax

cluster <field> [t=<threshold>] [match=<algorithm>] [labelfield=<field>] [countfield=<field>] [showcount=<bool>] [labelonly=<bool>] [delims=<string>]

Supported algorithms

• termlist (default) — positional term frequency, word order matters
• termset — bag-of-words, word order ignored
• ngramset — character trigram frequency for fuzzy matching

All algorithms use cosine similarity against the configured threshold to determine cluster membership.

Behavior

• By default (labelonly=false), deduplicates results to one
  representative row per cluster
• labelonly=true keeps all rows and adds the cluster label
• showcount=true adds a count of how many documents belong to each cluster
• Rows where the source field is null are filtered out before clustering

Changes

• CalciteRelNodeVisitor: implements visitCluster with window-based clustering, null filtering, and optional dedup/showcount
• ClusterLabelAggFunction: window aggregate that buffers rows, runs greedy clustering, and returns an array of cluster labels
• TextSimilarityClustering: cosine similarity computation with LRU-cached vectorization for termlist, termset, and ngramset modes
• Cluster AST node, parser rules, and PPL grammar additions
• Documentation with 7 doctest-verified examples
• Unit tests for logical plan verification
• Integration tests covering all parameters, dedup behavior, labelonly mode, and multi-row clustering

Related Issues

Resolves #5255

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

Path	Line	Severity	Description
common/src/main/java/org/opensearch/sql/common/cluster/TextSimilarityClustering.java	80	medium	Cache eviction is performed inside ConcurrentHashMap.computeIfAbsent() via parallelStream().forEach(vectorCache::remove). Modifying a ConcurrentHashMap while a computeIfAbsent call is in progress on the same map can cause a deadlock or infinite loop (known JDK issue). If triggered under adversarial input designed to hit MAX_CACHE_SIZE boundary conditions, this could be used as a denial-of-service vector against the query engine. The pattern is anomalous and not necessary — eviction could be done outside the lambda.
ppl/src/main/antlr/OpenSearchPPLParser.g4	1592	low	The ident grammar rule was restructured: previously keywords-as-identifiers fell through a separate alternative; now they are inlined as (ID | keywordsCanBeId) in the primary alternative. Combined with adding the single-character token T as both a parameter keyword and a searchable keyword, this increases the ambiguity surface of the PPL grammar. Misidentification of tokens could allow carefully crafted queries to bypass identifier-based access control checks, though no specific exploit path is evident from this diff alone.

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]

Failed to generate code suggestions for PR

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit a88cc9a)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 Multiple PR themes Sub-PR theme: Core clustering engine: TextSimilarityClustering and ClusterLabelAggFunction UDF Relevant files: common/src/main/java/org/opensearch/sql/common/cluster/TextSimilarityClustering.java core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java core/src/main/java/org/opensearch/sql/expression/function/BuiltinFunctionName.java core/src/main/java/org/opensearch/sql/expression/function/PPLBuiltinOperators.java core/src/main/java/org/opensearch/sql/expression/function/PPLFuncImpTable.java Sub-PR theme: PPL grammar, AST node, and parser support for cluster command Relevant files: core/src/main/java/org/opensearch/sql/ast/tree/Cluster.java core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java ppl/src/main/antlr/OpenSearchPPLLexer.g4 ppl/src/main/antlr/OpenSearchPPLParser.g4 ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java Sub-PR theme: Calcite plan visitor, integration tests, and documentation for cluster command Relevant files: core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteClusterCommandIT.java integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLClusterTest.java docs/user/ppl/cmd/cluster.md docs/user/ppl/index.md

⚡ Recommended focus areas for review Thread Safety The ClusterLabelAggFunction stores mutable state (threshold, matchMode, delims, bufferLimit, maxClusters) as instance fields that are mutated in the add method. Since Calcite may reuse aggregate function instances across rows or threads, this is not thread-safe and could lead to incorrect clustering results or race conditions in concurrent query execution. private int bufferLimit = 50000; // Configurable buffer size private int maxClusters = 10000; // Limit cluster count to prevent memory explosion private double threshold = 0.8; private String matchMode = "termlist"; private String delims = " "; @Override public Acc init() { return new Acc(); } @Override public Object result(Acc acc) { return acc.value(threshold, matchMode, delims, maxClusters); } @Override public Acc add(Acc acc, Object... values) { // Handle case where Calcite calls generic method with null field value if (values.length == 1) { String field = (values[0] != null) ? values[0].toString() : null; return add(acc, field); } throw new SyntaxCheckException( "Unsupported function signature for cluster aggregate. Valid parameters include (field:" + " required string), (t: optional double threshold 0.0-1.0, default 0.8), (match:" + " optional string algorithm 'termlist'|'termset'|'ngramset', default 'termlist')," + " (delims: optional string delimiters, default ' '), (labelfield: optional string" + " output field name, default 'cluster_label'), (countfield: optional string count" + " field name, default 'cluster_count')"); } public Acc add( Acc acc, String field, double threshold, String matchMode, String delims, int bufferLimit, int maxClusters) { // Process all rows, even when field is null - convert null to empty string // This ensures the result array matches input row count String processedField = (field != null) ? field : ""; this.threshold = threshold; this.matchMode = matchMode; this.delims = delims; this.bufferLimit = bufferLimit; this.maxClusters = maxClusters; acc.evaluate(processedField); if (bufferLimit > 0 && acc.bufferSize() == bufferLimit) { acc.partialMerge(threshold, matchMode, delims, maxClusters); acc.clearBuffer(); } return acc; } Partial Merge Bug When bufferLimit is reached, partialMerge is called and then clearBuffer is called. However, the remaining rows after the last partial merge are processed in value() via another partialMerge. The issue is that after a partial merge, the allLabels list grows incrementally, but the array indexing in visitCluster uses ROW_NUMBER() (1-based) to index into the returned list. If partial merges happen mid-stream, the returned list from value() may not have labels in the correct positions corresponding to input row order, since clearBuffer discards the buffer but the global cluster state is preserved. This could produce misaligned label-to-row mappings. if (bufferLimit > 0 && acc.bufferSize() == bufferLimit) { acc.partialMerge(threshold, matchMode, delims, maxClusters); acc.clearBuffer(); } return acc; Lexer Conflict Adding T as a dedicated lexer token (for the threshold parameter) is very broad and may conflict with existing identifiers or keywords named T in user queries. Any field or variable named t in PPL queries could be tokenized as this keyword instead of an identifier, potentially breaking existing queries. T: 'T'; Array Index Mismatch The cluster label array is indexed using ROW_NUMBER() OVER () (unbounded, no partition), which is computed over the already-filtered relation (after null filtering). However, the window aggregate cluster_label(...) also runs over the same filtered relation. If the row ordering is not deterministic or if the ROW_NUMBER window frame differs from the cluster_label window frame, the index used to extract labels from the array may not correctly correspond to the row's position in the array returned by the aggregate function. // Add ROW_NUMBER to index into the array (1-based). String rowNumAlias = "_cluster_row_idx"; RexNode rowNum = context .relBuilder .aggregateCall(SqlStdOperatorTable.ROW_NUMBER) .over() .rowsBetween(RexWindowBounds.UNBOUNDED_PRECEDING, RexWindowBounds.CURRENT_ROW) .as(rowNumAlias); context.relBuilder.projectPlus(rowNum); // Extract the label for this row: array[row_number] (ITEM access is 1-based). RexNode rowIdxAsInt = context.rexBuilder.makeCast( context.rexBuilder.getTypeFactory().createSqlType(SqlTypeName.INTEGER), context.relBuilder.field(rowNumAlias)); RexNode labelExpr = context.rexBuilder.makeCall( SqlStdOperatorTable.ITEM, context.relBuilder.field(arrayAlias), rowIdxAsInt); context.relBuilder.projectPlus(context.relBuilder.alias(labelExpr, node.getLabelField())); Cache Thread Safety The vectorCache is a LinkedHashMap with access-order enabled (true in the constructor), used as an LRU cache. LinkedHashMap is not thread-safe, and if TextSimilarityClustering instances are shared across threads (e.g., via the aggregate function), concurrent access to vectorCache could cause ConcurrentModificationException or data corruption. private final Map<String, Map<CharSequence, Integer>> vectorCache = new LinkedHashMap<>(MAX_CACHE_SIZE, 0.75f, true) { @Override protected boolean removeEldestEntry(Map.Entry<String, Map<CharSequence, Integer>> eldest) { return size() > MAX_CACHE_SIZE; } };

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to a88cc9a

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Avoid mutable shared state in aggregate function The add method mutates instance-level fields (this.threshold, this.matchMode, etc.) which are shared state on the ClusterLabelAggFunction instance. In a concurrent or multi-query environment, this can cause race conditions where one query's parameters overwrite another's. These configuration values should be passed through to result() and partialMerge() via the Acc accumulator rather than stored as mutable instance fields. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [58-84] public Acc add( Acc acc, String field, double threshold, String matchMode, String delims, int bufferLimit, int maxClusters) { - // Process all rows, even when field is null - convert null to empty string - // This ensures the result array matches input row count String processedField = (field != null) ? field : ""; - this.threshold = threshold; - this.matchMode = matchMode; - this.delims = delims; - this.bufferLimit = bufferLimit; - this.maxClusters = maxClusters; - ... + // Store config in accumulator instead of mutable instance fields + acc.threshold = threshold; + acc.matchMode = matchMode; + acc.delims = delims; + acc.bufferLimit = bufferLimit; + acc.maxClusters = maxClusters; + acc.evaluate(processedField); + + if (bufferLimit > 0 && acc.bufferSize() == bufferLimit) { + acc.partialMerge(threshold, matchMode, delims, maxClusters); + acc.clearBuffer(); + } + + return acc; +} + Suggestion importance[1-10]: 6 __ Why: Mutating instance-level fields like this.threshold and this.matchMode in the add method is a valid concurrency concern. However, the improved_code references fields (acc.threshold, acc.bufferLimit, etc.) that don't exist in the Acc class as shown in the PR, making the suggested fix incomplete as written.	Low
	Handle all parameter counts in varargs method The generic add(Acc acc, Object... values) method only handles the case where values.length == 1, but the actual invocation from visitCluster passes 4 parameters (field, threshold, matchMode, delims). This means any call with the correct 4-argument signature via the varargs path will throw a SyntaxCheckException. The varargs overload should also handle the 4-argument case to match the parameters passed in visitCluster. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [42-56] @Override public Acc add(Acc acc, Object... values) { - // Handle case where Calcite calls generic method with null field value if (values.length == 1) { String field = (values[0] != null) ? values[0].toString() : null; return add(acc, field); } + if (values.length == 4) { + String field = (values[0] != null) ? values[0].toString() : null; + double threshold = ((Number) values[1]).doubleValue(); + String matchMode = (values[2] != null) ? values[2].toString() : this.matchMode; + String delims = (values[3] != null) ? values[3].toString() : this.delims; + return add(acc, field, threshold, matchMode, delims); + } throw new SyntaxCheckException( - "Unsupported function signature for cluster aggregate. ..."); + "Unsupported function signature for cluster aggregate. Valid parameters include (field:" + + " required string), (t: optional double threshold 0.0-1.0, default 0.8), (match:" + + " optional string algorithm 'termlist'|'termset'|'ngramset', default 'termlist')," + + " (delims: optional string delimiters, default ' '), (labelfield: optional string" + + " output field name, default 'cluster_label'), (countfield: optional string count" + + " field name, default 'cluster_count')"); } Suggestion importance[1-10]: 5 __ Why: The suggestion identifies a potential gap in the varargs add method, but Calcite's reflection-based dispatch typically routes to the most specific typed overload (e.g., add(Acc, String, double, String, String)) rather than the varargs method. The actual risk depends on how Calcite invokes the method, making this a moderate concern worth investigating.	Low
General	Fix static field declaration order The vectorCache field references MAX_CACHE_SIZE before it is declared as a static field. In Java, instance field initializers run in declaration order, and while static fields are initialized before instance fields, referencing a static field that appears later in the source can cause confusion and potential issues with some compilers or static analysis tools. Move the MAX_CACHE_SIZE declaration before the vectorCache field to ensure clear and correct ordering. common/src/main/java/org/opensearch/sql/common/cluster/TextSimilarityClustering.java [26-33] +private static final int MAX_CACHE_SIZE = 10000; + private final Map<String, Map<CharSequence, Integer>> vectorCache = new LinkedHashMap<>(MAX_CACHE_SIZE, 0.75f, true) { @Override protected boolean removeEldestEntry(Map.Entry<String, Map<CharSequence, Integer>> eldest) { return size() > MAX_CACHE_SIZE; } }; -private static final int MAX_CACHE_SIZE = 10000; Suggestion importance[1-10]: 4 __ Why: While MAX_CACHE_SIZE is a static field and will be initialized before instance fields in Java, moving it before vectorCache improves readability and follows conventional ordering. This is a style/clarity improvement rather than a bug fix.	Low
	Include all parameters in anonymized output The anonymizer always appends labelfield and countfield regardless of whether showcount or labelonly were set by the user. This means the anonymized output may not accurately reflect the original query structure, and parameters like showcount, labelonly, and delims are silently dropped. The anonymizer should include all relevant parameters to produce a faithful (but masked) representation of the query. ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java [971-987] @Override public String visitCluster(org.opensearch.sql.ast.tree.Cluster node, String context) { String child = node.getChild().get(0).accept(this, context); - ... + String sourceField = visitExpression(node.getSourceField()); + StringBuilder command = new StringBuilder(); + command.append(child).append(" | cluster ").append(sourceField); + + command.append(" t=").append(node.getThreshold()); + + if (!"termlist".equals(node.getMatchMode())) { + command.append(" match=").append(node.getMatchMode()); + } + command.append(" labelfield=").append(MASK_COLUMN); command.append(" countfield=").append(MASK_COLUMN); + + if (node.isShowCount()) { + command.append(" showcount=true"); + } + if (node.isLabelOnly()) { + command.append(" labelonly=true"); + } return command.toString(); } Suggestion importance[1-10]: 4 __ Why: The suggestion to include showcount and labelonly in the anonymized output is reasonable for completeness, but the improved_code is nearly identical to the existing_code with only minor additions. The existing anonymizer tests in the PR don't test these parameters, so this is a minor improvement to query representation fidelity.	Low

---

Previous suggestions

Suggestions up to commit ca93d37

Category	Suggestion	Impact
Possible issue	Fix static field declaration order to avoid zero value MAX_CACHE_SIZE is a static final field but is referenced in the instance initializer of vectorCache before it is declared in the class body. In Java, static fields are initialized in declaration order, so referencing MAX_CACHE_SIZE in the field initializer of vectorCache (which appears before the MAX_CACHE_SIZE declaration) will result in MAX_CACHE_SIZE being 0 at the time the LinkedHashMap is constructed. Move the MAX_CACHE_SIZE declaration before vectorCache. common/src/main/java/org/opensearch/sql/common/cluster/TextSimilarityClustering.java [26-33] +private static final int MAX_CACHE_SIZE = 10000; + private final Map<String, Map<CharSequence, Integer>> vectorCache = new LinkedHashMap<>(MAX_CACHE_SIZE, 0.75f, true) { @Override protected boolean removeEldestEntry(Map.Entry<String, Map<CharSequence, Integer>> eldest) { return size() > MAX_CACHE_SIZE; } }; -private static final int MAX_CACHE_SIZE = 10000; Suggestion importance[1-10]: 9 __ Why: This is a real bug: MAX_CACHE_SIZE is declared after vectorCache which references it in its initializer. In Java, static fields are initialized in declaration order, so MAX_CACHE_SIZE will be 0 when vectorCache is constructed, causing the LinkedHashMap to be initialized with capacity 0 and the removeEldestEntry check to always return true, effectively making the cache non-functional.	High
	Avoid shared mutable state in aggregate function The add method mutates instance fields (this.threshold, this.matchMode, etc.) which are shared state on the ClusterLabelAggFunction instance. In a concurrent or multi-threaded execution environment (e.g., parallel query execution), this creates a race condition where one thread's parameters can overwrite another's. These configuration values should be stored in the Acc accumulator instead of on the function instance. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [58-84] -public Acc add( - Acc acc, - String field, - double threshold, - String matchMode, - String delims, - int bufferLimit, - int maxClusters) { - ... - this.threshold = threshold; - this.matchMode = matchMode; - this.delims = delims; - this.bufferLimit = bufferLimit; - this.maxClusters = maxClusters; +public static class Acc implements Accumulator { + private final List<String> buffer = new ArrayList<>(); + private final List<ClusterRepresentative> globalClusters = new ArrayList<>(); + private final List<Integer> allLabels = new ArrayList<>(); + private int nextClusterId = 1; + // Store config in accumulator to avoid shared mutable state + double threshold = 0.8; + String matchMode = "termlist"; + String delims = " "; + int bufferLimit = 50000; + int maxClusters = 10000; ... } Suggestion importance[1-10]: 7 __ Why: The add method mutates instance-level fields (this.threshold, this.matchMode, etc.) which is a real thread-safety concern in concurrent query execution. Moving these configuration values into the Acc accumulator would be a more correct design, though the impact depends on whether Calcite instantiates a new function object per query.	Medium
	Handle all argument counts in generic add method The generic add(Acc, Object...) method only handles the case where values.length == 1, but Calcite may call this method with 4 parameters (field, threshold, matchMode, delims) when the function is invoked with all arguments. This would cause a SyntaxCheckException at runtime instead of routing to the correct overload. Add handling for the 4-argument case (and potentially others) before throwing the exception. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [42-56] @Override public Acc add(Acc acc, Object... values) { - // Handle case where Calcite calls generic method with null field value if (values.length == 1) { String field = (values[0] != null) ? values[0].toString() : null; return add(acc, field); } + if (values.length == 4) { + String field = (values[0] != null) ? values[0].toString() : null; + double threshold = ((Number) values[1]).doubleValue(); + String matchMode = (String) values[2]; + String delims = (String) values[3]; + return add(acc, field, threshold, matchMode, delims); + } throw new SyntaxCheckException( - "Unsupported function signature for cluster aggregate. ..."); + "Unsupported function signature for cluster aggregate. Valid parameters include (field:" + + " required string), (t: optional double threshold 0.0-1.0, default 0.8), (match:" + + " optional string algorithm 'termlist'|'termset'|'ngramset', default 'termlist')," + + " (delims: optional string delimiters, default ' '), (labelfield: optional string" + + " output field name, default 'cluster_label'), (countfield: optional string count" + + " field name, default 'cluster_count')"); } Suggestion importance[1-10]: 6 __ Why: The generic add method only handles values.length == 1, but Calcite may invoke it with more arguments (e.g., 4 for field, threshold, matchMode, delims). However, looking at the code, the typed overloads (add(Acc, String, double, String, String), etc.) are what Calcite should call via reflection, so this may not be a real runtime issue in practice. Still, it's a defensive improvement worth considering.	Low
General	Include all boolean parameters in anonymized output The anonymizer always appends labelfield and countfield regardless of whether showcount or labelonly were set, potentially producing output that doesn't match the original query structure. Additionally, labelonly and showcount boolean parameters are never included in the anonymized output, which means the anonymized query may not be semantically equivalent to the original for audit/logging purposes. ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java [971-987] @Override public String visitCluster(org.opensearch.sql.ast.tree.Cluster node, String context) { String child = node.getChild().get(0).accept(this, context); - ... + String sourceField = visitExpression(node.getSourceField()); + StringBuilder command = new StringBuilder(); + command.append(child).append(" | cluster ").append(sourceField); + + command.append(" t=").append(node.getThreshold()); + + if (!"termlist".equals(node.getMatchMode())) { + command.append(" match=").append(node.getMatchMode()); + } + command.append(" labelfield=").append(MASK_COLUMN); command.append(" countfield=").append(MASK_COLUMN); + + if (node.isLabelOnly()) { + command.append(" labelonly=true"); + } + if (node.isShowCount()) { + command.append(" showcount=true"); + } return command.toString(); } Suggestion importance[1-10]: 4 __ Why: The anonymizer omits labelonly and showcount boolean parameters from the output, making the anonymized query not fully representative of the original. This is a minor completeness issue for audit/logging purposes, but doesn't affect correctness of query execution.	Low

Suggestions up to commit 9563361

Category	Suggestion	Impact
Possible issue	Fix forward reference to static constant MAX_CACHE_SIZE is referenced in the field initializer before it is declared as a static final field. In Java, forward references to static fields in instance field initializers are allowed only if the static field is declared before the instance field. Since MAX_CACHE_SIZE is declared after vectorCache, this will cause a compile error or unexpected behavior. Move the MAX_CACHE_SIZE declaration before vectorCache. common/src/main/java/org/opensearch/sql/common/cluster/TextSimilarityClustering.java [26-33] +private static final int MAX_CACHE_SIZE = 10000; + private final Map<String, Map<CharSequence, Integer>> vectorCache = new LinkedHashMap<>(MAX_CACHE_SIZE, 0.75f, true) { @Override protected boolean removeEldestEntry(Map.Entry<String, Map<CharSequence, Integer>> eldest) { return size() > MAX_CACHE_SIZE; } }; -private static final int MAX_CACHE_SIZE = 10000; Suggestion importance[1-10]: 8 __ Why: MAX_CACHE_SIZE is used in the vectorCache field initializer but declared after it on line 33. In Java, static fields referenced in instance field initializers must be declared before the instance field to avoid a compile error or illegal forward reference. Moving MAX_CACHE_SIZE before vectorCache is a necessary fix.	Medium
	Remove mutable shared state from aggregate function The add method mutates instance-level fields (this.threshold, this.matchMode, etc.) which are shared state on the ClusterLabelAggFunction instance. When Calcite reuses the same function instance across multiple aggregations or concurrent queries, this creates a race condition and incorrect behavior. These parameters should be passed directly to acc.partialMerge and acc.value rather than stored as mutable instance state. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [58-84] public Acc add( Acc acc, String field, double threshold, String matchMode, String delims, int bufferLimit, int maxClusters) { - ... - this.threshold = threshold; - this.matchMode = matchMode; - this.delims = delims; - this.bufferLimit = bufferLimit; - this.maxClusters = maxClusters; - ... + String processedField = (field != null) ? field : ""; + + acc.evaluate(processedField); + + if (bufferLimit > 0 && acc.bufferSize() == bufferLimit) { + acc.partialMerge(threshold, matchMode, delims, maxClusters); + acc.clearBuffer(); + } + + return acc; } Suggestion importance[1-10]: 7 __ Why: Mutating instance-level fields (this.threshold, this.matchMode, etc.) in the add method creates a potential race condition if the function instance is reused across concurrent queries. The improved code correctly avoids storing these as mutable state and passes them directly, which is a meaningful correctness improvement.	Medium
	Handle all argument counts in generic add method The generic add(Acc, Object...) method only handles the case where values.length == 1, but Calcite may call this method with 4 parameters (field, threshold, matchMode, delims) when the function is invoked with all arguments. This would cause a SyntaxCheckException at runtime instead of routing to the correct overload. Add handling for the 4-argument case to properly dispatch to the full add overload. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [42-56] @Override public Acc add(Acc acc, Object... values) { - // Handle case where Calcite calls generic method with null field value if (values.length == 1) { String field = (values[0] != null) ? values[0].toString() : null; return add(acc, field); + } else if (values.length == 4) { + String field = (values[0] != null) ? values[0].toString() : null; + double threshold = ((Number) values[1]).doubleValue(); + String matchMode = (values[2] != null) ? values[2].toString() : this.matchMode; + String delims = (values[3] != null) ? values[3].toString() : this.delims; + return add(acc, field, threshold, matchMode, delims); } throw new SyntaxCheckException( - "Unsupported function signature for cluster aggregate. ..."); + "Unsupported function signature for cluster aggregate. Valid parameters include (field:" + + " required string), (t: optional double threshold 0.0-1.0, default 0.8), (match:" + + " optional string algorithm 'termlist'|'termset'|'ngramset', default 'termlist')," + + " (delims: optional string delimiters, default ' '), (labelfield: optional string" + + " output field name, default 'cluster_label'), (countfield: optional string count" + + " field name, default 'cluster_count')"); } Suggestion importance[1-10]: 6 __ Why: The generic add method only handles values.length == 1, but Calcite may call it with 4 arguments when all parameters are provided. This could cause a SyntaxCheckException at runtime. However, the actual dispatch mechanism depends on how Calcite resolves overloaded methods, and in practice Calcite typically calls the most specific typed overload directly rather than the varargs version.	Low
General	Guard against empty child list access node.getChild().get(0) will throw an IndexOutOfBoundsException if the cluster node has no child (e.g., when the node is visited before being attached to a pipeline). The getChild() method returns an empty list when child is null, so this should be guarded. Check if the child list is non-empty before accessing index 0. ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java [971-987] @Override public String visitCluster(org.opensearch.sql.ast.tree.Cluster node, String context) { - String child = node.getChild().get(0).accept(this, context); + List<UnresolvedPlan> children = node.getChild(); + String child = children.isEmpty() ? "" : children.get(0).accept(this, context); ... } Suggestion importance[1-10]: 4 __ Why: Accessing node.getChild().get(0) without checking if the list is empty could throw an IndexOutOfBoundsException in edge cases. However, in normal usage the anonymizer is called on fully-constructed AST nodes that always have a child, so this is a low-probability defensive improvement.	Low

Suggestions up to commit 5bfaf69

Category	Suggestion	Impact
Possible issue	Handle all expected argument counts in generic add method The generic add(Acc, Object...) method only handles the case where values.length == 1, but Calcite may call this method with 4 parameters (field, threshold, matchMode, delims) as configured in visitCluster. This would cause an unexpected SyntaxCheckException at runtime instead of routing to the correct overload. Add handling for the 4-argument case to match the parameters passed from visitCluster. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [42-56] @Override public Acc add(Acc acc, Object... values) { - // Handle case where Calcite calls generic method with null field value if (values.length == 1) { String field = (values[0] != null) ? values[0].toString() : null; return add(acc, field); } + if (values.length == 4) { + String field = (values[0] != null) ? values[0].toString() : null; + double threshold = ((Number) values[1]).doubleValue(); + String matchMode = (values[2] != null) ? values[2].toString() : this.matchMode; + String delims = (values[3] != null) ? values[3].toString() : this.delims; + return add(acc, field, threshold, matchMode, delims); + } throw new SyntaxCheckException( - "Unsupported function signature for cluster aggregate. ..."); + "Unsupported function signature for cluster aggregate. Valid parameters include (field:" + + " required string), (t: optional double threshold 0.0-1.0, default 0.8), (match:" + + " optional string algorithm 'termlist'|'termset'|'ngramset', default 'termlist')," + + " (delims: optional string delimiters, default ' '), (labelfield: optional string" + + " output field name, default 'cluster_label'), (countfield: optional string count" + + " field name, default 'cluster_count')"); } Suggestion importance[1-10]: 7 __ Why: The generic add method only handles values.length == 1, but Calcite may invoke it with 4 arguments (field, threshold, matchMode, delims) as configured in visitCluster. Without handling this case, a SyntaxCheckException would be thrown at runtime instead of routing to the correct overload.	Medium
	Avoid race conditions from mutable shared instance state The add method mutates instance fields (this.threshold, this.matchMode, etc.) which are shared state on the ClusterLabelAggFunction instance. In a concurrent or multi-threaded execution environment (common in query engines), this creates a race condition where one thread's parameters can overwrite another's. These configuration values should be stored in the Acc accumulator instead of on the function instance. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [58-84] -public Acc add( - Acc acc, - String field, - double threshold, - String matchMode, - String delims, - int bufferLimit, - int maxClusters) { - ... - this.threshold = threshold; - this.matchMode = matchMode; - this.delims = delims; - this.bufferLimit = bufferLimit; - this.maxClusters = maxClusters; +public static class Acc implements Accumulator { + private final List<String> buffer = new ArrayList<>(); + private final List<ClusterRepresentative> globalClusters = new ArrayList<>(); + private final List<Integer> allLabels = new ArrayList<>(); + private int nextClusterId = 1; + // Store config in accumulator to avoid shared mutable state + double threshold = 0.8; + String matchMode = "termlist"; + String delims = " "; + int bufferLimit = 50000; + int maxClusters = 10000; ... } +// In add(...) method, update acc fields instead of this fields: +acc.threshold = threshold; +acc.matchMode = matchMode; +acc.delims = delims; +acc.bufferLimit = bufferLimit; +acc.maxClusters = maxClusters; Suggestion importance[1-10]: 6 __ Why: Mutating instance fields like this.threshold, this.matchMode, etc. in the add method creates potential race conditions in concurrent environments. Moving these configuration values into the Acc accumulator would be safer, though the actual risk depends on how Calcite instantiates and uses the aggregate function.	Low
General	Conditionally include optional parameters in anonymized output The anonymizer always appends labelfield and countfield regardless of whether showcount or labelonly were set, which produces output that doesn't match the original command structure and may confuse log analysis. Additionally, labelonly and showcount boolean parameters are never included in the anonymized output, losing information about the command's behavior. Include these parameters conditionally to produce a more accurate anonymized representation. ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java [971-987] @Override public String visitCluster(org.opensearch.sql.ast.tree.Cluster node, String context) { String child = node.getChild().get(0).accept(this, context); - ... + String sourceField = visitExpression(node.getSourceField()); + StringBuilder command = new StringBuilder(); + command.append(child).append(" | cluster ").append(sourceField); + + command.append(" t=").append(node.getThreshold()); + + if (!"termlist".equals(node.getMatchMode())) { + command.append(" match=").append(node.getMatchMode()); + } + command.append(" labelfield=").append(MASK_COLUMN); - command.append(" countfield=").append(MASK_COLUMN); + + if (node.isShowCount()) { + command.append(" countfield=").append(MASK_COLUMN); + command.append(" showcount=true"); + } + + if (node.isLabelOnly()) { + command.append(" labelonly=true"); + } return command.toString(); } Suggestion importance[1-10]: 4 __ Why: The anonymizer always appends labelfield and countfield and omits showcount/labelonly flags, producing output that doesn't accurately reflect the original command structure. The improved code conditionally includes these parameters for a more faithful anonymized representation.	Low
	Declare constant before use in field initializer MAX_CACHE_SIZE is referenced in the field initializer before it is declared as a static final field below it. In Java, instance field initializers can reference static fields declared later in the class, but this ordering is fragile and may cause confusion or issues in some edge cases. Move the MAX_CACHE_SIZE declaration before the vectorCache field declaration. common/src/main/java/org/opensearch/sql/common/cluster/TextSimilarityClustering.java [26-33] +private static final int MAX_CACHE_SIZE = 10000; + private final Map<String, Map<CharSequence, Integer>> vectorCache = new LinkedHashMap<>(MAX_CACHE_SIZE, 0.75f, true) { @Override protected boolean removeEldestEntry(Map.Entry<String, Map<CharSequence, Integer>> eldest) { return size() > MAX_CACHE_SIZE; } }; -private static final int MAX_CACHE_SIZE = 10000; Suggestion importance[1-10]: 3 __ Why: In Java, static fields are initialized before instance fields regardless of declaration order, so MAX_CACHE_SIZE being declared after vectorCache is technically valid. This is a minor code style/readability improvement with no functional impact.	Low

Suggestions up to commit 3fbd709

Category	Suggestion	Impact
Possible issue	Handle all expected argument counts in generic add method The generic add(Acc, Object...) method only handles the case where values.length == 1, but Calcite may call this method with 4 parameters (field, threshold, matchMode, delims) as defined in the window function invocation. This would cause an unexpected SyntaxCheckException at runtime instead of routing to the correct overloaded add method. Add handling for the 4-argument case (and potentially others) before throwing the exception. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [42-56] @Override public Acc add(Acc acc, Object... values) { - // Handle case where Calcite calls generic method with null field value if (values.length == 1) { String field = (values[0] != null) ? values[0].toString() : null; return add(acc, field); } + if (values.length == 4) { + String field = (values[0] != null) ? values[0].toString() : null; + double threshold = values[1] != null ? ((Number) values[1]).doubleValue() : this.threshold; + String matchMode = values[2] != null ? values[2].toString() : this.matchMode; + String delims = values[3] != null ? values[3].toString() : this.delims; + return add(acc, field, threshold, matchMode, delims); + } throw new SyntaxCheckException( - "Unsupported function signature for cluster aggregate. ..."); + "Unsupported function signature for cluster aggregate. Valid parameters include (field:" + + " required string), (t: optional double threshold 0.0-1.0, default 0.8), (match:" + + " optional string algorithm 'termlist'|'termset'|'ngramset', default 'termlist')," + + " (delims: optional string delimiters, default ' '), (labelfield: optional string" + + " output field name, default 'cluster_label'), (countfield: optional string count" + + " field name, default 'cluster_count')"); } Suggestion importance[1-10]: 7 __ Why: The generic add method only handles values.length == 1, but Calcite may invoke it with 4 arguments (field, threshold, matchMode, delims) as passed in the window function call. This could cause a runtime SyntaxCheckException instead of routing to the correct overload, making this a real potential issue.	Medium
	Avoid mutable shared state in aggregate function instance Mutating instance fields (this.threshold, this.matchMode, etc.) inside the add method is not thread-safe and can cause incorrect behavior when the aggregate function instance is shared across rows or threads. These configuration values should be stored in the Acc accumulator instead of as mutable instance fields, or the instance should be treated as stateless with parameters passed through the accumulator. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [58-84] -public Acc add( - Acc acc, - String field, - double threshold, - String matchMode, - String delims, - int bufferLimit, - int maxClusters) { - ... - this.threshold = threshold; - this.matchMode = matchMode; - this.delims = delims; - this.bufferLimit = bufferLimit; - this.maxClusters = maxClusters; - ... +// Move threshold, matchMode, delims, bufferLimit, maxClusters into Acc +public static class Acc implements Accumulator { + private final List<String> buffer = new ArrayList<>(); + private final List<ClusterRepresentative> globalClusters = new ArrayList<>(); + private final List<Integer> allLabels = new ArrayList<>(); + private int nextClusterId = 1; + // Configuration stored per-accumulator + double threshold = 0.8; + String matchMode = "termlist"; + String delims = " "; + int bufferLimit = 50000; + int maxClusters = 10000; + // ... rest of Acc } Suggestion importance[1-10]: 6 __ Why: Mutating instance fields like this.threshold, this.matchMode, etc. inside add is not thread-safe and can cause incorrect behavior if the function instance is shared. Moving these into the Acc accumulator would be a cleaner design, though the practical impact depends on how Calcite manages function instances.	Low
General	Declare constant before its use in field initializer MAX_CACHE_SIZE is referenced in the field initializer before it is declared as a static final field below it. In Java, instance field initializers can reference static fields declared later in the class, but this ordering is confusing and error-prone. Move the MAX_CACHE_SIZE declaration before the vectorCache field to make the dependency order explicit and avoid potential issues. common/src/main/java/org/opensearch/sql/common/cluster/TextSimilarityClustering.java [26-33] +private static final int MAX_CACHE_SIZE = 10000; + private final Map<String, Map<CharSequence, Integer>> vectorCache = new LinkedHashMap<>(MAX_CACHE_SIZE, 0.75f, true) { @Override protected boolean removeEldestEntry(Map.Entry<String, Map<CharSequence, Integer>> eldest) { return size() > MAX_CACHE_SIZE; } }; -private static final int MAX_CACHE_SIZE = 10000; Suggestion importance[1-10]: 4 __ Why: While Java technically allows referencing a static final field declared later in the class from an instance field initializer, the current ordering of vectorCache before MAX_CACHE_SIZE is confusing and error-prone. Moving MAX_CACHE_SIZE before vectorCache improves readability and makes the dependency explicit.	Low
	Include all parameters in anonymized cluster command output The anonymizer always appends labelfield and countfield regardless of whether showcount or labelonly were set, which produces output that doesn't match the original command structure and may confuse log analysis. Additionally, labelonly and showcount parameters are silently dropped from the anonymized output, losing information about the command's behavior. These parameters should be included in the anonymized output. ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java [971-987] @Override public String visitCluster(org.opensearch.sql.ast.tree.Cluster node, String context) { String child = node.getChild().get(0).accept(this, context); - ... + String sourceField = visitExpression(node.getSourceField()); + StringBuilder command = new StringBuilder(); + command.append(child).append(" | cluster ").append(sourceField); + + command.append(" t=").append(node.getThreshold()); + + if (!"termlist".equals(node.getMatchMode())) { + command.append(" match=").append(node.getMatchMode()); + } + command.append(" labelfield=").append(MASK_COLUMN); command.append(" countfield=").append(MASK_COLUMN); + + if (node.isLabelOnly()) { + command.append(" labelonly=true"); + } + if (node.isShowCount()) { + command.append(" showcount=true"); + } return command.toString(); } Suggestion importance[1-10]: 4 __ Why: The visitCluster anonymizer silently drops labelonly and showcount parameters, which reduces the fidelity of anonymized output for log analysis. Including these boolean flags (which don't contain sensitive data) would make the anonymized representation more accurate.	Low

Suggestions up to commit 1763fb8

Category	Suggestion	Impact
Possible issue	Handle all expected argument counts in generic add method The generic add(Acc, Object...) method only handles the case where values.length == 1, but Calcite may call this method with 4 parameters (field, threshold, matchMode, delims) as configured in visitCluster. This would cause a SyntaxCheckException at runtime instead of routing to the correct overload. Add handling for the 4-argument case to match the parameters passed from visitCluster. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [42-56] @Override public Acc add(Acc acc, Object... values) { - // Handle case where Calcite calls generic method with null field value if (values.length == 1) { String field = (values[0] != null) ? values[0].toString() : null; return add(acc, field); } + if (values.length == 4) { + String field = (values[0] != null) ? values[0].toString() : null; + double threshold = ((Number) values[1]).doubleValue(); + String matchMode = (values[2] != null) ? values[2].toString() : this.matchMode; + String delims = (values[3] != null) ? values[3].toString() : this.delims; + return add(acc, field, threshold, matchMode, delims); + } throw new SyntaxCheckException( - "Unsupported function signature for cluster aggregate. ..."); + "Unsupported function signature for cluster aggregate. Valid parameters include (field:" + + " required string), (t: optional double threshold 0.0-1.0, default 0.8), (match:" + + " optional string algorithm 'termlist'|'termset'|'ngramset', default 'termlist')," + + " (delims: optional string delimiters, default ' '), (labelfield: optional string" + + " output field name, default 'cluster_label'), (countfield: optional string count" + + " field name, default 'cluster_count')"); } Suggestion importance[1-10]: 6 __ Why: The generic add method only handles values.length == 1, but Calcite may call it with 4 arguments (field, threshold, matchMode, delims) as configured in visitCluster. This could cause a SyntaxCheckException at runtime. However, Calcite's reflection-based dispatch typically routes to the most specific typed overload, so this may not be triggered in practice.	Low
	Avoid mutating shared instance state in accumulator method Mutating instance fields (this.threshold, this.matchMode, etc.) inside the add method is not thread-safe and can cause incorrect behavior when the aggregate function is used concurrently across multiple partitions or threads. These configuration values should be set once at construction time or passed explicitly through the accumulator rather than stored as mutable instance state. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [58-84] public Acc add( Acc acc, String field, double threshold, String matchMode, String delims, int bufferLimit, int maxClusters) { - ... - this.threshold = threshold; - this.matchMode = matchMode; - this.delims = delims; - this.bufferLimit = bufferLimit; - this.maxClusters = maxClusters; - ... + String processedField = (field != null) ? field : ""; + + acc.evaluate(processedField); + + if (bufferLimit > 0 && acc.bufferSize() == bufferLimit) { + acc.partialMerge(threshold, matchMode, delims, maxClusters); + acc.clearBuffer(); + } + + return acc; } Suggestion importance[1-10]: 6 __ Why: Mutating instance fields like this.threshold, this.matchMode, etc. inside add is not thread-safe and can cause incorrect behavior in concurrent scenarios. The improved code removes these mutations and passes parameters directly, which is a valid correctness concern.	Low
General	Declare constant before its use in field initializer MAX_CACHE_SIZE is referenced in the field initializer before it is declared as a static final field below it. In Java, instance field initializers can reference static fields declared later in the class, but this ordering is confusing and error-prone. Move the MAX_CACHE_SIZE declaration before the vectorCache field to make the dependency explicit and avoid potential issues. common/src/main/java/org/opensearch/sql/common/cluster/TextSimilarityClustering.java [26-33] +private static final int MAX_CACHE_SIZE = 10000; + private final Map<String, Map<CharSequence, Integer>> vectorCache = new LinkedHashMap<>(MAX_CACHE_SIZE, 0.75f, true) { @Override protected boolean removeEldestEntry(Map.Entry<String, Map<CharSequence, Integer>> eldest) { return size() > MAX_CACHE_SIZE; } }; -private static final int MAX_CACHE_SIZE = 10000; Suggestion importance[1-10]: 4 __ Why: While Java allows static fields to be referenced in instance initializers regardless of declaration order, moving MAX_CACHE_SIZE before vectorCache improves readability and avoids confusion. This is a minor style/maintainability improvement.	Low
	Guard against empty child list before access Calling node.getChild().get(0) without checking if the child list is empty will throw an IndexOutOfBoundsException if the cluster node has no child (e.g., when used as the root node or in error scenarios). Add a null/empty check before accessing the first child, consistent with how getChild() can return an empty list per the Cluster.getChild() implementation. ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java [971-987] @Override public String visitCluster(org.opensearch.sql.ast.tree.Cluster node, String context) { - String child = node.getChild().get(0).accept(this, context); - ... + List<UnresolvedPlan> children = node.getChild(); + String child = children.isEmpty() ? "" : children.get(0).accept(this, context); + String sourceField = visitExpression(node.getSourceField()); + StringBuilder command = new StringBuilder(); + command.append(child).append(" | cluster ").append(sourceField); + + command.append(" t=").append(node.getThreshold()); + + if (!"termlist".equals(node.getMatchMode())) { + command.append(" match=").append(node.getMatchMode()); + } + + command.append(" labelfield=").append(MASK_COLUMN); + command.append(" countfield=").append(MASK_COLUMN); + + return command.toString(); } Suggestion importance[1-10]: 4 __ Why: Calling node.getChild().get(0) without checking for an empty list could throw an IndexOutOfBoundsException in edge cases. However, in normal PPL query processing, the cluster node always has a child, making this a low-probability defensive improvement.	Low

[github-actions]

Persistent review updated to latest commit 152d2d4

[github-actions]

Persistent review updated to latest commit 74001d0

[github-actions]

Failed to generate code suggestions for PR

[github-actions]

Failed to generate code suggestions for PR

[github-actions]

Persistent review updated to latest commit 5792e98

[github-actions]

Persistent review updated to latest commit e009b93

[github-actions]

Persistent review updated to latest commit 2d99d92

[github-actions]

Persistent review updated to latest commit 1763fb8

[github-actions]

Persistent review updated to latest commit 3fbd709

[github-actions]

Persistent review updated to latest commit 5bfaf69

[opensearch-trigger-bot]

This PR is stalled because it has been open for 2 weeks with no activity.

[github-actions]

Persistent review updated to latest commit 9563361

[github-actions]

Persistent review updated to latest commit ca93d37

[github-actions]

Persistent review updated to latest commit a88cc9a

[opensearch-trigger-bot]

This PR is stalled because it has been open for 2 weeks with no activity.

[Swiddis]

merge conflicts

[Swiddis]

ideally this should be enum instead of str, which makes it hard to misuse the interface

[Swiddis]

if minor: this probably also could stand to be tagged somehow instead of just working directly on strings

[penghuo]

Did u benchmark cluster command on http_logs dataset?

[penghuo]

what is allowed value of delims?

[penghuo]

initialCapacity equal to MAX_CACHE_SIZE?

[penghuo]

Does this setting configurable by user?