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

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 Threshold Validation The threshold validation rejects exactly 0.0 and 1.0, but cosine similarity can legitimately produce these boundary values (identical texts yield 1.0, completely dissimilar texts yield 0.0). Rejecting t=1.0 prevents users from requiring exact matches, and rejecting t=0.0 prevents clustering everything together. If these boundaries are intentionally forbidden, the error message should explain why; otherwise, the validation should use <= 0.0 || >= 1.0 or < 0.0 || > 1.0 depending on intent. private static double validateThreshold(double threshold) { if (threshold <= 0.0 || threshold >= 1.0) { throw new IllegalArgumentException( "The threshold must be > 0.0 and < 1.0, got: " + threshold); } return threshold; Null Handling Mismatch The add(Acc acc, String field) method converts null to empty string (line 68), but CalciteRelNodeVisitor.visitCluster filters out null source fields before clustering (line 3165). This means the null-to-empty conversion in the aggregate function is unreachable. If nulls are always filtered upstream, the conversion is dead code. If the filter is removed or bypassed in some execution path, empty strings will cluster together, which may not be the intended behavior for missing data. 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; } Unbounded Memory Growth When maxClusters is reached, the fallback logic (lines 168-183) forces assignment to an existing cluster or creates a new one anyway. If the dataset contains more than maxClusters distinct patterns and the buffer is processed in batches, globalClusters can grow beyond maxClusters across multiple partialMerge calls, because each batch may add clusters up to the limit. Over many batches, this can cause memory exhaustion. The limit is only enforced per-batch, not globally. } else if (globalClusters.size() < maxClusters) { // Create new cluster ClusterRepresentative newCluster = new ClusterRepresentative(nextClusterId++, value, 1); globalClusters.add(newCluster); return new ClusterAssignment(newCluster.id, 1); } else { // Force into closest existing cluster when at max limit if (bestCluster != null) { bestCluster.size++; return new ClusterAssignment(bestCluster.id, bestCluster.size); } else if (!globalClusters.isEmpty()) { // Fallback: assign to cluster 1 globalClusters.get(0).size++; return new ClusterAssignment(globalClusters.get(0).id, globalClusters.get(0).size); } else { // Emergency fallback: create first cluster ClusterRepresentative newCluster = new ClusterRepresentative(1, value, 1); globalClusters.add(newCluster); nextClusterId = 2; return new ClusterAssignment(1, 1); } } Unbounded Cache Growth The vectorCache uses an LRU eviction policy with MAX_CACHE_SIZE=10000, but if the dataset contains more than 10,000 unique text values, the cache will thrash and evict entries that may be needed again (e.g., cluster representatives). This degrades performance but does not cause incorrect results. However, if a single query processes millions of unique strings, the cache provides no benefit and adds overhead. Consider whether the cache size should be configurable or if a different caching strategy is needed for large datasets. 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; Array Index Assumption The code assumes ITEM (array access) is 1-based (line 3201 comment), but the actual indexing depends on the SQL dialect. If the underlying system uses 0-based indexing, array[row_number] will be off-by-one, causing each row to receive the wrong cluster label. Verify that Calcite's ITEM operator uses 1-based indexing for arrays, or adjust the index calculation accordingly. // 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);

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to a51e8f9

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Prevent exceeding max cluster limit The emergency fallback branch creates a new cluster even when maxClusters limit is reached. This violates the max cluster constraint and can lead to exceeding the configured limit. Remove this unreachable branch or ensure it respects the maxClusters limit by forcing assignment to an existing cluster instead. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [162-183] } else if (globalClusters.size() < maxClusters) { // Create new cluster ClusterRepresentative newCluster = new ClusterRepresentative(nextClusterId++, value, 1); globalClusters.add(newCluster); return new ClusterAssignment(newCluster.id, 1); } else { // Force into closest existing cluster when at max limit if (bestCluster != null) { bestCluster.size++; return new ClusterAssignment(bestCluster.id, bestCluster.size); } else if (!globalClusters.isEmpty()) { // Fallback: assign to cluster 1 globalClusters.get(0).size++; return new ClusterAssignment(globalClusters.get(0).id, globalClusters.get(0).size); - } else { - // Emergency fallback: create first cluster - ClusterRepresentative newCluster = new ClusterRepresentative(1, value, 1); - globalClusters.add(newCluster); - nextClusterId = 2; - return new ClusterAssignment(1, 1); } + // This should never happen if maxClusters > 0 + throw new IllegalStateException("Cannot assign cluster: no clusters exist and max limit reached"); } Suggestion importance[1-10]: 7 __ Why: The emergency fallback branch at lines 177-182 creates a new cluster even when maxClusters limit is reached, violating the constraint. This branch appears unreachable given the logic flow, but if reached, it would exceed the configured limit. Replacing it with an exception or ensuring proper constraint enforcement improves correctness.	Medium
General	Reorder constant declaration before usage The MAX_CACHE_SIZE constant is declared after the vectorCache field that uses it. While Java allows forward references to static fields, this ordering is confusing and could lead to initialization issues. Move the MAX_CACHE_SIZE declaration before vectorCache for better code clarity and to avoid potential initialization order problems. 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]: 6 __ Why: While Java allows forward references to static fields, declaring MAX_CACHE_SIZE after its usage in vectorCache is confusing and reduces code readability. Moving the constant declaration before its usage improves code clarity and follows common Java conventions.	Low
	Support multi-parameter function calls The generic add method only handles single-value case but throws an exception for all other cases. This prevents the function from accepting multiple parameters (threshold, match mode, delims) as documented. Add proper handling for multi-parameter cases or ensure the overloaded methods are called correctly by Calcite. core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ClusterLabelAggFunction.java [42-56] 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 >= 2) { + String field = (values[0] != null) ? values[0].toString() : null; + double threshold = (values[1] instanceof Number) ? ((Number) values[1]).doubleValue() : this.threshold; + String matchMode = (values.length > 2 && values[2] != null) ? values[2].toString() : this.matchMode; + String delims = (values.length > 3 && values[3] != null) ? values[3].toString() : this.delims; + return add(acc, field, threshold, matchMode, delims); } 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')"); } Suggestion importance[1-10]: 5 __ Why: The generic add method only handles single-value cases and throws an exception for all others. While the overloaded methods may handle multi-parameter cases, adding explicit handling for multi-parameter scenarios in the generic method could improve robustness and prevent unexpected exceptions if Calcite calls this method with multiple parameters.	Low

---

Previous suggestions

Suggestions up to commit a88cc9a

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

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

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

[github-actions]

Persistent review updated to latest commit a51e8f9