From e26387400e81acf1269f00f30ac3a340bf44b29b Mon Sep 17 00:00:00 2001 From: Kunal Khatua Date: Fri, 17 Jul 2026 21:30:01 -0700 Subject: [PATCH 1/6] feat(cat): Add summarization for _cat APIs via h= aggregation Add grouped + aggregated views to _cat APIs, expressed entirely through the h= parameter: a bare column token (e.g. index) is a GROUP BY key and a func(field) token (sum, count, avg, min, max) is an aggregate over the grouped rows, mirroring SELECT index, sum(docs) ... GROUP BY index. When h= contains no aggregation function the table is returned unchanged, so ordinary listings behave exactly as before. Because the group key is always a bare h= token, it is always rendered. Aggregation preserves numeric source types (Long/Integer/Short/Byte, ByteSizeValue, SizeValue, TimeValue) so downstream bytes=/time=/size= rendering keeps working, and avg divides by the count of non-null values so null cells do not drag the average toward zero. Signed-off-by: Kunal Khatua --- .../opensearch/rest/action/cat/RestTable.java | 7 + .../rest/action/cat/TableSummarizer.java | 330 +++++++++++++ .../rest/action/cat/RestTableTests.java | 53 ++ .../rest/action/cat/TableSummarizerTests.java | 466 ++++++++++++++++++ 4 files changed, 856 insertions(+) create mode 100644 server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java create mode 100644 server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java b/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java index e27c8212e16a0..6555bf7f4763b 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java @@ -72,6 +72,13 @@ public class RestTable { public static RestResponse buildResponse(Table table, RestChannel channel) throws Exception { RestRequest request = channel.request(); + String hParam = request.param("h"); + // Grouping/aggregation is inferred from h=: if any h= token is an aggregation function + // (e.g. sum(docs)), summarize the table; bare tokens are treated as GROUP BY keys. Plain + // h= listings (no functions) are returned unchanged by TableSummarizer.summarize. + if (TableSummarizer.hasAggregation(hParam)) { + table = TableSummarizer.summarize(table, hParam); + } MediaType mediaType = getXContentType(request); if (mediaType != null) { return buildXContentBuilder(table, channel); diff --git a/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java b/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java new file mode 100644 index 0000000000000..725c2ec85496b --- /dev/null +++ b/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java @@ -0,0 +1,330 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.rest.action.cat; + +import org.opensearch.common.Table; +import org.opensearch.common.unit.SizeValue; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.common.Strings; +import org.opensearch.core.common.unit.ByteSizeValue; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Builds a grouped + aggregated view of a {@link Table} for {@code _cat} APIs. Grouping and + * aggregation are both expressed entirely through the {@code h=} parameter: any bare column token + * (e.g. {@code index}) is treated as a GROUP BY key, and any function token (e.g. {@code sum(docs)}) + * is treated as an aggregate over the grouped rows. This mirrors the SQL mental model + * {@code SELECT index, sum(docs) ... GROUP BY index} without a separate {@code summarize=} parameter. + * + *

When {@code h=} contains no aggregation function the table is returned unchanged, so ordinary + * {@code h=} listings behave exactly as they did before summarization existed. + * + *

Supported functions: {@code sum}, {@code count}, {@code avg}, {@code min}, {@code max}. + * Numeric types {@link Long}, {@link Integer}, {@link Short}, {@link Byte}, {@link ByteSizeValue}, + * {@link SizeValue}, {@link TimeValue} are preserved through aggregation so downstream rendering + * with {@code bytes=}/{@code time=}/{@code size=} continues to work on aggregated values. + * + *

The {@code avg} function computes {@code sum / count_of_non_null_values}; rows whose value is + * null for a given aggregation column do not contribute to either the numerator or denominator + * for that column's average. + * + * @opensearch.internal + */ +public final class TableSummarizer { + + /** Matches function syntax like {@code sum(docs.count)}, {@code count(shard)}. */ + private static final Pattern AGG_FUNC_PATTERN = Pattern.compile("^(sum|count|avg|min|max)\\((.+)\\)$"); + + private TableSummarizer() {} + + /** + * Returns true if at least one token in the {@code h=} parameter is an aggregation function + * invocation like {@code sum(field)}. Used by callers that want to short-circuit upstream row + * construction when neither sort nor aggregation is requested. + */ + public static boolean hasAggregation(String headersParam) { + if (headersParam == null || headersParam.isEmpty()) { + return false; + } + for (String header : Strings.splitStringByCommaToArray(headersParam)) { + if (AGG_FUNC_PATTERN.matcher(header.trim()).matches()) { + return true; + } + } + return false; + } + + /** + * Build a summarized table by inferring grouping and aggregation from the {@code h=} parameter. + * Bare column tokens become GROUP BY keys; {@code func(field)} tokens become aggregates. Returns + * the original {@code table} unchanged when {@code hParam} is null/empty or contains no + * aggregation function (i.e. an ordinary listing was requested). + * + * @param hParam value of the {@code h=} query parameter; may be null + */ + public static Table summarize(Table table, String hParam) { + if (hParam == null || hParam.isEmpty()) { + return table; + } + Map aliasMap = table.getAliasMap(); + + // Split h= into group-by columns (bare tokens) and aggregation columns (func(field) tokens). + List groupByTokens = new ArrayList<>(); + List aggColumns = new ArrayList<>(); + for (String token : Strings.splitStringByCommaToArray(hParam)) { + String h = token.trim(); + if (h.isEmpty()) { + continue; + } + Matcher m = AGG_FUNC_PATTERN.matcher(h); + if (m.matches()) { + String func = m.group(1).toLowerCase(Locale.ROOT); + String field = aliasMap.getOrDefault(m.group(2).trim(), m.group(2).trim()); + aggColumns.add(new AggColumn(func, field, h)); + } else { + groupByTokens.add(h); + } + } + + // No aggregation requested => ordinary listing; return the table unchanged so that plain + // h= requests behave exactly as they did before summarization existed. + if (aggColumns.isEmpty()) { + return table; + } + + // Resolve group-by field aliases to canonical names. The raw token is kept for the result + // header display name so it still matches the h= token during downstream column filtering. + String[] groupByFields = groupByTokens.toArray(new String[0]); + String[] resolvedGroupBy = new String[groupByFields.length]; + for (int i = 0; i < groupByFields.length; i++) { + resolvedGroupBy[i] = aliasMap.getOrDefault(groupByFields[i], groupByFields[i]); + } + + Map> colMap = table.getAsMap(); + int rowCount = table.getRows().size(); + + // Single-pass online aggregation. For each row we resolve its group key and update the + // running aggregator for every (group, aggColumn) pair. Memory is O(G * C) rather than + // O(N) for the row-index lists used by the previous two-pass design. + Map groups = new LinkedHashMap<>(); + for (int row = 0; row < rowCount; row++) { + // Build key from group-by values (uses List.equals/hashCode — collision-free). + List keyValues = new ArrayList<>(resolvedGroupBy.length); + for (String gb : resolvedGroupBy) { + List col = colMap.get(gb); + keyValues.add(col != null ? col.get(row).value : null); + } + GroupKey key = new GroupKey(keyValues); + Aggregator[] aggregators = groups.computeIfAbsent(key, k -> { + Aggregator[] arr = new Aggregator[aggColumns.size()]; + for (int i = 0; i < arr.length; i++) { + arr[i] = new Aggregator(); + } + return arr; + }); + for (int i = 0; i < aggColumns.size(); i++) { + AggColumn agg = aggColumns.get(i); + List col = colMap.get(agg.field); + Object val = col != null ? col.get(row).value : null; + aggregators[i].add(val); + } + } + + // Build the result table. + Table result = new Table(); + result.startHeaders(); + for (int g = 0; g < resolvedGroupBy.length; g++) { + Table.Cell origHeader = table.getHeaderMap().get(resolvedGroupBy[g]); + result.addCell(groupByFields[g], copyHeaderAttrString(origHeader, groupByFields[g])); + } + for (AggColumn agg : aggColumns) { + result.addCell(agg.displayName, "text-align:right;desc:" + agg.func + "(" + agg.field + ")"); + } + result.endHeaders(); + + for (Map.Entry entry : groups.entrySet()) { + result.startRow(); + for (Object val : entry.getKey().values) { + result.addCell(val); + } + Aggregator[] aggregators = entry.getValue(); + for (int i = 0; i < aggColumns.size(); i++) { + result.addCell(aggregators[i].getValue(aggColumns.get(i).func)); + } + result.endRow(); + } + return result; + } + + /** + * Reconstructs a safe attribute string for the result header that mirrors the original column's + * metadata. Values containing the attribute delimiters ({@code ;} or {@code :}) are skipped to + * avoid producing a string that cannot be parsed back correctly. + */ + private static String copyHeaderAttrString(Table.Cell origHeader, String fallbackDesc) { + if (origHeader == null || origHeader.attr == null || origHeader.attr.isEmpty()) { + return "desc:" + fallbackDesc; + } + StringBuilder sb = new StringBuilder(); + boolean hasDesc = false; + for (Map.Entry e : origHeader.attr.entrySet()) { + String v = e.getValue(); + if (v == null) continue; + if (v.indexOf(';') >= 0 || v.indexOf(':') >= 0) continue; // skip un-encodable values + if (sb.length() > 0) sb.append(';'); + sb.append(e.getKey()).append(':').append(v); + if ("desc".equals(e.getKey())) hasDesc = true; + } + if (!hasDesc) { + if (sb.length() > 0) sb.append(';'); + sb.append("desc:").append(fallbackDesc); + } + return sb.toString(); + } + + static final class AggColumn { + final String func; + final String field; + final String displayName; + + AggColumn(String func, String field, String displayName) { + this.func = func; + this.field = field; + this.displayName = displayName; + } + } + + /** + * Hashable group key over the resolved group-by values. Uses {@link List#equals(Object)} / + * {@link List#hashCode()} so the key is collision-free regardless of cell value contents + * (the previous design used {@code String.join("\0", ...)} which could in theory collide). + */ + static final class GroupKey { + final List values; + private final int hash; + + GroupKey(List values) { + this.values = values; + this.hash = values.hashCode(); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof GroupKey)) return false; + return values.equals(((GroupKey) o).values); + } + + @Override + public int hashCode() { + return hash; + } + } + + /** + * Per-(group, column) running aggregator. Maintains the sum, min, max, count of contributing + * (non-null and numerically-parseable) values, and a {@code countAll} including null/unparseable + * rows for the {@code count} function. The first non-null value seen is captured as a + * {@code sampleValue} so the aggregated output can be wrapped back into its source type + * (ByteSizeValue, TimeValue, etc.). + */ + static final class Aggregator { + long countAll = 0; + long countContributing = 0; + Double sum = null; + Double min = null; + Double max = null; + Object sampleValue = null; + + void add(Object value) { + countAll++; + if (value != null) { + if (sampleValue == null) sampleValue = value; + Double num = parseAsDouble(value); + if (num != null) { + if (sum == null) sum = 0.0; + sum += num; + if (min == null || num < min) min = num; + if (max == null || num > max) max = num; + countContributing++; + } + } + } + + Object getValue(String function) { + switch (function) { + case "count": + return countAll; + case "sum": + return sum == null ? null : formatValue(sum, sampleValue); + case "min": + return min == null ? null : formatValue(min, sampleValue); + case "max": + return max == null ? null : formatValue(max, sampleValue); + case "avg": + // sum of non-nulls divided by count of non-nulls — null values do not pull the + // average toward zero (catSummary divided by countAll instead, which is wrong + // when any row contributes null for the aggregation column). + if (countContributing == 0 || sum == null) return null; + double avg = sum / countContributing; + // For non-type-preserving outputs we keep two decimal places (matches the + // legacy inline implementation in RestTable). + if (sampleValue instanceof ByteSizeValue || sampleValue instanceof SizeValue || sampleValue instanceof TimeValue) { + return formatValue(avg, sampleValue); + } + return formatValue(Math.round(avg * 100.0) / 100.0, sampleValue); + default: + return null; + } + } + } + + /** Parse a cell value into a double for aggregation. Returns null if the value isn't numeric. */ + static Double parseAsDouble(Object value) { + if (value == null) return null; + if (value instanceof Number) return ((Number) value).doubleValue(); + if (value instanceof ByteSizeValue) return (double) ((ByteSizeValue) value).getBytes(); + if (value instanceof TimeValue) return (double) ((TimeValue) value).millis(); + if (value instanceof SizeValue) return (double) ((SizeValue) value).singles(); + if (value instanceof String) { + try { + return Double.parseDouble((String) value); + } catch (NumberFormatException e) { + return null; + } + } + return null; + } + + /** Wrap an aggregated double back into the original source type so renderers preserve units. */ + static Object formatValue(double num, Object sampleValue) { + if (sampleValue instanceof ByteSizeValue) return new ByteSizeValue((long) num); + if (sampleValue instanceof SizeValue) return new SizeValue((long) num); + if (sampleValue instanceof TimeValue) return new TimeValue((long) num); + if (sampleValue instanceof Long || sampleValue instanceof Integer || sampleValue instanceof Short || sampleValue instanceof Byte) { + return (long) num; + } + if (sampleValue instanceof String) { + if (num == Math.floor(num) && !Double.isInfinite(num)) { + return String.valueOf((long) num); + } + return String.format(Locale.ROOT, "%.2f", num); + } + // Default: return Long when integral (consistent with the original inline impl), else Double. + if (num == Math.floor(num) && !Double.isInfinite(num)) return (long) num; + return num; + } +} diff --git a/server/src/test/java/org/opensearch/rest/action/cat/RestTableTests.java b/server/src/test/java/org/opensearch/rest/action/cat/RestTableTests.java index f475d0ff1efda..74e279e9843a3 100644 --- a/server/src/test/java/org/opensearch/rest/action/cat/RestTableTests.java +++ b/server/src/test/java/org/opensearch/rest/action/cat/RestTableTests.java @@ -341,6 +341,59 @@ private Table getPaginatedTable() { return paginatedTable; } + public void testHeaderAggregationTriggersSummarizedJsonOutput() throws Exception { + Table t = new Table(); + t.startHeaders(); + t.addCell("index", "desc:index name"); + t.addCell("docs", "text-align:right;desc:doc count"); + t.endHeaders(); + for (Object[] row : Arrays.asList(new Object[] { "idx-a", 100L }, new Object[] { "idx-a", 200L }, new Object[] { "idx-b", 50L })) { + t.startRow(); + t.addCell(row[0]); + t.addCell(row[1]); + t.endRow(); + } + + FakeRestRequest req = new FakeRestRequest.Builder(xContentRegistry()).withHeaders( + Collections.singletonMap(ACCEPT, Collections.singletonList(APPLICATION_JSON)) + ).build(); + // A function token in h= triggers summarization; `index` is inferred as the GROUP BY key. + req.params().put("h", "index,sum(docs)"); + RestResponse response = buildResponse(t, new AbstractRestChannel(req, true) { + @Override + public void sendResponse(RestResponse response) {} + }); + assertThat( + response.content().utf8ToString(), + equalTo("[{\"index\":\"idx-a\",\"sum(docs)\":\"300\"},{\"index\":\"idx-b\",\"sum(docs)\":\"50\"}]") + ); + } + + public void testHeaderWithoutAggregationIsNotSummarized() throws Exception { + Table t = new Table(); + t.startHeaders(); + t.addCell("index", "desc:index name"); + t.addCell("docs", "text-align:right;desc:doc count"); + t.endHeaders(); + for (Object[] row : Arrays.asList(new Object[] { "idx-a", 100L }, new Object[] { "idx-a", 200L }, new Object[] { "idx-b", 50L })) { + t.startRow(); + t.addCell(row[0]); + t.addCell(row[1]); + t.endRow(); + } + + FakeRestRequest req = new FakeRestRequest.Builder(xContentRegistry()).withHeaders( + Collections.singletonMap(ACCEPT, Collections.singletonList(APPLICATION_JSON)) + ).build(); + // Bare columns only => ordinary listing, all three rows preserved (no grouping). + req.params().put("h", "index"); + RestResponse response = buildResponse(t, new AbstractRestChannel(req, true) { + @Override + public void sendResponse(RestResponse response) {} + }); + assertThat(response.content().utf8ToString(), equalTo("[{\"index\":\"idx-a\"},{\"index\":\"idx-a\"},{\"index\":\"idx-b\"}]")); + } + private void addHeaders(Table table) { table.startHeaders(); table.addCell("bulk.foo", "alias:f;desc:foo"); diff --git a/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java b/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java new file mode 100644 index 0000000000000..bcdaa0b1fcc4b --- /dev/null +++ b/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java @@ -0,0 +1,466 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.rest.action.cat; + +import org.opensearch.common.Table; +import org.opensearch.common.unit.SizeValue; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.common.unit.ByteSizeValue; +import org.opensearch.rest.RestRequest; +import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.test.rest.FakeRestRequest; + +import java.util.concurrent.TimeUnit; + +import static org.hamcrest.Matchers.equalTo; + +public class TableSummarizerTests extends OpenSearchTestCase { + + // ---------- helpers ---------- + + private Table buildNumericTable() { + Table t = new Table(); + t.startHeaders(); + t.addCell("index", "desc:index name"); + t.addCell("shard", "desc:shard id"); + t.addCell("docs", "text-align:right;desc:doc count"); + t.addCell("size", "text-align:right;desc:size in bytes"); + t.endHeaders(); + + t.startRow(); + t.addCell("idx-a"); + t.addCell("0"); + t.addCell(100L); + t.addCell(new ByteSizeValue(1024)); + t.endRow(); + + t.startRow(); + t.addCell("idx-a"); + t.addCell("1"); + t.addCell(200L); + t.addCell(new ByteSizeValue(2048)); + t.endRow(); + + t.startRow(); + t.addCell("idx-b"); + t.addCell("0"); + t.addCell(50L); + t.addCell(new ByteSizeValue(512)); + t.endRow(); + + return t; + } + + private Table buildTimeTable() { + Table t = new Table(); + t.startHeaders(); + t.addCell("group", "desc:group name"); + t.addCell("duration", "text-align:right;desc:duration"); + t.endHeaders(); + + t.startRow(); + t.addCell("tasks"); + t.addCell(new TimeValue(10, TimeUnit.NANOSECONDS)); + t.endRow(); + + t.startRow(); + t.addCell("tasks"); + t.addCell(new TimeValue(20, TimeUnit.MILLISECONDS)); + t.endRow(); + + t.startRow(); + t.addCell("tasks"); + t.addCell(new TimeValue(75, TimeUnit.SECONDS)); + t.endRow(); + + return t; + } + + private String renderValueViaResponse(FakeRestRequest request, Object value) { + try { + java.lang.reflect.Method m = RestTable.class.getDeclaredMethod("renderValue", RestRequest.class, Object.class); + m.setAccessible(true); + return (String) m.invoke(null, request, value); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + // ---------- hasAggregation ---------- + + public void testHasAggregationDetectsFunctionTokens() { + assertTrue(TableSummarizer.hasAggregation("index,sum(docs)")); + assertTrue(TableSummarizer.hasAggregation("count(shard)")); + assertTrue(TableSummarizer.hasAggregation("index , avg(size) , node")); + } + + public void testHasAggregationFalseWhenNoFunctionTokens() { + assertFalse(TableSummarizer.hasAggregation("index,shard,node")); + assertFalse(TableSummarizer.hasAggregation("")); + assertFalse(TableSummarizer.hasAggregation(null)); + } + + // ---------- summarize returns original table when h= is null/empty or has no aggregation ---------- + + public void testNullOrEmptyHeadersReturnsOriginal() { + Table t = buildNumericTable(); + assertSame(t, TableSummarizer.summarize(t, null)); + assertSame(t, TableSummarizer.summarize(t, "")); + } + + public void testNoAggregationFunctionReturnsOriginal() { + Table t = buildNumericTable(); + // h= with only bare columns (no func()) is an ordinary listing => table returned unchanged. + assertSame(t, TableSummarizer.summarize(t, "index,shard")); + } + + // ---------- aggregation correctness ---------- + + public void testSummarizeSum() { + Table t = buildNumericTable(); + Table result = TableSummarizer.summarize(t, "index,sum(docs),sum(size)"); + assertThat(result.getRows().size(), equalTo(2)); + // idx-a: 100+200=300 + assertThat(result.getAsMap().get("sum(docs)").get(0).value, equalTo(300L)); + // idx-b: 50 + assertThat(result.getAsMap().get("sum(docs)").get(1).value, equalTo(50L)); + } + + public void testSummarizeCount() { + Table t = buildNumericTable(); + Table result = TableSummarizer.summarize(t, "index,count(shard)"); + assertThat(result.getAsMap().get("count(shard)").get(0).value, equalTo(2L)); // idx-a has 2 rows + assertThat(result.getAsMap().get("count(shard)").get(1).value, equalTo(1L)); // idx-b has 1 row + } + + public void testSummarizeAvg() { + Table t = buildNumericTable(); + Table result = TableSummarizer.summarize(t, "index,avg(docs)"); + // idx-a: (100+200)/2 = 150 (integral result returned as long) + assertThat(((Number) result.getAsMap().get("avg(docs)").get(0).value).longValue(), equalTo(150L)); + } + + public void testSummarizeMinMax() { + Table t = buildNumericTable(); + Table result = TableSummarizer.summarize(t, "index,min(docs),max(docs)"); + assertThat(result.getAsMap().get("min(docs)").get(0).value, equalTo(100L)); + assertThat(result.getAsMap().get("max(docs)").get(0).value, equalTo(200L)); + assertThat(result.getAsMap().get("min(docs)").get(1).value, equalTo(50L)); + assertThat(result.getAsMap().get("max(docs)").get(1).value, equalTo(50L)); + } + + public void testSummarizeMultipleGroupByFields() { + Table t = buildNumericTable(); + Table result = TableSummarizer.summarize(t, "index,shard,sum(docs)"); + assertThat(result.getRows().size(), equalTo(3)); + } + + public void testBareColumnsWithoutAggregationDoNotGroup() { + Table t = buildNumericTable(); + // With group-by inferred from h=, a bare column and no aggregation function is a plain + // listing, not a distinct-values grouping: the original table is returned unchanged. + Table result = TableSummarizer.summarize(t, "index"); + assertSame(t, result); + assertThat(result.getRows().size(), equalTo(3)); + } + + public void testBareColumnBecomesGroupByWhenAggregationPresent() { + Table t = buildNumericTable(); + // The bare `index` token is the GROUP BY key; count(shard) triggers summarization. + Table result = TableSummarizer.summarize(t, "index,count(shard)"); + assertThat(result.getRows().size(), equalTo(2)); // 2 distinct indices + // The group-by column is present in the output under its raw token name. + assertThat(result.getAsMap().get("index").get(0).value, equalTo("idx-a")); + assertThat(result.getAsMap().get("index").get(1).value, equalTo("idx-b")); + } + + public void testAggregationWithNoGroupByProducesSingleRow() { + Table t = buildNumericTable(); + // No bare column => a single global aggregate row (grand total). + Table result = TableSummarizer.summarize(t, "sum(docs)"); + assertThat(result.getRows().size(), equalTo(1)); + assertThat(result.getAsMap().get("sum(docs)").get(0).value, equalTo(350L)); // 100+200+50 + } + + // ---------- TableSummarizer no longer caps groups itself ---------- + // The `limit` parameter is now exclusively handled by RestTable.getRowOrder; the summarizer + // always returns all groups so that downstream sort + top-K can operate on the full set. + + public void testSummarizerDoesNotTruncateGroups() { + Table t = buildNumericTable(); + // Even if the user passes limit=1 in the request, the summarizer signature no longer takes + // it — it always returns all groups. Truncation is applied by RestTable.getRowOrder. + Table result = TableSummarizer.summarize(t, "index,sum(docs)"); + assertThat(result.getRows().size(), equalTo(2)); + } + + // ---------- type preservation ---------- + + public void testSummarizePreservesByteSizeValue() { + Table t = buildNumericTable(); + Table result = TableSummarizer.summarize(t, "index,sum(size),avg(size)"); + + Object sumVal = result.getAsMap().get("sum(size)").get(0).value; + assertTrue("sum(size) should be ByteSizeValue but was " + sumVal.getClass(), sumVal instanceof ByteSizeValue); + assertThat(((ByteSizeValue) sumVal).getBytes(), equalTo(3072L)); + + Object avgVal = result.getAsMap().get("avg(size)").get(0).value; + assertTrue("avg(size) should be ByteSizeValue but was " + avgVal.getClass(), avgVal instanceof ByteSizeValue); + assertThat(((ByteSizeValue) avgVal).getBytes(), equalTo(1536L)); + } + + public void testSummarizeTimeValuePreservation() { + Table t = buildTimeTable(); + Table result = TableSummarizer.summarize(t, "group,sum(duration),avg(duration),min(duration),max(duration)"); + + Object sumVal = result.getAsMap().get("sum(duration)").get(0).value; + Object avgVal = result.getAsMap().get("avg(duration)").get(0).value; + Object minVal = result.getAsMap().get("min(duration)").get(0).value; + Object maxVal = result.getAsMap().get("max(duration)").get(0).value; + + assertTrue("sum should be TimeValue", sumVal instanceof TimeValue); + assertTrue("avg should be TimeValue", avgVal instanceof TimeValue); + assertTrue("min should be TimeValue", minVal instanceof TimeValue); + assertTrue("max should be TimeValue", maxVal instanceof TimeValue); + + // Aggregation uses millis precision: 10ns→0ms, 20ms→20ms, 75s→75000ms + assertThat(((TimeValue) sumVal).millis(), equalTo(75020L)); + assertThat(((TimeValue) minVal).millis(), equalTo(0L)); + assertThat(((TimeValue) maxVal).millis(), equalTo(75000L)); + // avg: (0 + 20 + 75000) / 3 = 25006.67 → cast back to TimeValue gives 25006ms + assertThat(((TimeValue) avgVal).millis(), equalTo(25006L)); + } + + public void testSummarizeTimeValueHumanReadable() { + Table t = buildTimeTable(); + Table result = TableSummarizer.summarize(t, "group,sum(duration),min(duration),max(duration)"); + + Object sumVal = result.getAsMap().get("sum(duration)").get(0).value; + Object minVal = result.getAsMap().get("min(duration)").get(0).value; + Object maxVal = result.getAsMap().get("max(duration)").get(0).value; + + assertThat(sumVal.toString(), equalTo("1.2m")); + assertThat(minVal.toString(), equalTo("0s")); + assertThat(maxVal.toString(), equalTo("1.2m")); + } + + public void testSummarizeTimeValueWithExplicitUnit() { + Table t = buildTimeTable(); + Table result = TableSummarizer.summarize(t, "group,sum(duration),min(duration),max(duration)"); + + FakeRestRequest msReq = new FakeRestRequest(); + msReq.params().put("time", "ms"); + assertThat(renderValueViaResponse(msReq, result.getAsMap().get("sum(duration)").get(0).value), equalTo("75020")); + assertThat(renderValueViaResponse(msReq, result.getAsMap().get("max(duration)").get(0).value), equalTo("75000")); + + FakeRestRequest sReq = new FakeRestRequest(); + sReq.params().put("time", "s"); + assertThat(renderValueViaResponse(sReq, result.getAsMap().get("sum(duration)").get(0).value), equalTo("75")); + + FakeRestRequest microsReq = new FakeRestRequest(); + microsReq.params().put("time", "micros"); + assertThat(renderValueViaResponse(microsReq, result.getAsMap().get("min(duration)").get(0).value), equalTo("0")); + assertThat(renderValueViaResponse(microsReq, result.getAsMap().get("sum(duration)").get(0).value), equalTo("75020000")); + + FakeRestRequest nanosReq = new FakeRestRequest(); + nanosReq.params().put("time", "nanos"); + assertThat(renderValueViaResponse(nanosReq, result.getAsMap().get("min(duration)").get(0).value), equalTo("0")); + assertThat(renderValueViaResponse(nanosReq, result.getAsMap().get("sum(duration)").get(0).value), equalTo("75020000000")); + } + + public void testSummarizeByteSizeValueHumanReadable() { + Table t = buildNumericTable(); + Table result = TableSummarizer.summarize(t, "index,sum(size),min(size),max(size)"); + + Object sumVal = result.getAsMap().get("sum(size)").get(0).value; + Object minVal = result.getAsMap().get("min(size)").get(0).value; + Object maxVal = result.getAsMap().get("max(size)").get(0).value; + + assertThat(sumVal.toString(), equalTo("3kb")); + assertThat(minVal.toString(), equalTo("1kb")); + assertThat(maxVal.toString(), equalTo("2kb")); + } + + public void testSummarizeByteSizeValueWithExplicitUnit() { + Table t = buildNumericTable(); + Table result = TableSummarizer.summarize(t, "index,sum(size)"); + + FakeRestRequest bReq = new FakeRestRequest(); + bReq.params().put("bytes", "b"); + assertThat(renderValueViaResponse(bReq, result.getAsMap().get("sum(size)").get(0).value), equalTo("3072")); + + FakeRestRequest kbReq = new FakeRestRequest(); + kbReq.params().put("bytes", "kb"); + assertThat(renderValueViaResponse(kbReq, result.getAsMap().get("sum(size)").get(0).value), equalTo("3")); + } + + // ---------- broadened numeric type preservation ---------- + + public void testIntegerShortByteRoundTrip() { + Table t = new Table(); + t.startHeaders(); + t.addCell("g", "desc:group"); + t.addCell("vi", "desc:int"); + t.addCell("vs", "desc:short"); + t.addCell("vb", "desc:byte"); + t.endHeaders(); + + t.startRow(); + t.addCell("a"); + t.addCell((Integer) 10); + t.addCell((Short) (short) 2); + t.addCell((Byte) (byte) 3); + t.endRow(); + + t.startRow(); + t.addCell("a"); + t.addCell((Integer) 20); + t.addCell((Short) (short) 4); + t.addCell((Byte) (byte) 5); + t.endRow(); + + Table result = TableSummarizer.summarize(t, "g,sum(vi),sum(vs),sum(vb)"); + // All Number-derived inputs collapse to Long on output (Number/Integer/Short/Byte case). + assertThat(result.getAsMap().get("sum(vi)").get(0).value, equalTo(30L)); + assertThat(result.getAsMap().get("sum(vs)").get(0).value, equalTo(6L)); + assertThat(result.getAsMap().get("sum(vb)").get(0).value, equalTo(8L)); + } + + public void testSizeValueRoundTrip() { + Table t = new Table(); + t.startHeaders(); + t.addCell("g"); + t.addCell("count_v"); + t.endHeaders(); + + t.startRow(); + t.addCell("a"); + t.addCell(new SizeValue(100)); + t.endRow(); + + t.startRow(); + t.addCell("a"); + t.addCell(new SizeValue(250)); + t.endRow(); + + Table result = TableSummarizer.summarize(t, "g,sum(count_v)"); + Object sumVal = result.getAsMap().get("sum(count_v)").get(0).value; + assertTrue("sum(count_v) should be SizeValue but was " + sumVal.getClass(), sumVal instanceof SizeValue); + assertThat(((SizeValue) sumVal).singles(), equalTo(350L)); + } + + public void testStringNumericRoundTrip() { + Table t = new Table(); + t.startHeaders(); + t.addCell("g"); + t.addCell("vstr"); + t.endHeaders(); + + t.startRow(); + t.addCell("a"); + t.addCell("10"); + t.endRow(); + + t.startRow(); + t.addCell("a"); + t.addCell("25"); + t.endRow(); + + Table result = TableSummarizer.summarize(t, "g,sum(vstr)"); + // String numerics: integral results round-trip to String to preserve type. + assertEquals("35", result.getAsMap().get("sum(vstr)").get(0).value); + } + + // ---------- avg semantics ---------- + + public void testAvgIgnoresNullValuesInDenominator() { + // catSummary's bug: avg = sum / count_all_rows including null rows. We compute + // sum / count_non_null_rows so null cells don't drag the average toward zero. + Table t = new Table(); + t.startHeaders(); + t.addCell("g"); + t.addCell("v"); + t.endHeaders(); + + t.startRow(); + t.addCell("a"); + t.addCell(10L); + t.endRow(); + + t.startRow(); + t.addCell("a"); + t.addCell(20L); + t.endRow(); + + t.startRow(); + t.addCell("a"); + t.addCell((Object) null); // contributes to countAll but not avg denominator + t.endRow(); + + Table result = TableSummarizer.summarize(t, "g,avg(v),count(v)"); + // avg = (10 + 20) / 2 = 15 (NOT 30/3 = 10) + assertThat(((Number) result.getAsMap().get("avg(v)").get(0).value).longValue(), equalTo(15L)); + // count counts ALL rows in the group, including the null row. + assertThat(result.getAsMap().get("count(v)").get(0).value, equalTo(3L)); + } + + public void testAvgIsNullWhenAllValuesAreNull() { + Table t = new Table(); + t.startHeaders(); + t.addCell("g"); + t.addCell("v"); + t.endHeaders(); + + t.startRow(); + t.addCell("a"); + t.addCell((Object) null); + t.endRow(); + t.startRow(); + t.addCell("a"); + t.addCell((Object) null); + t.endRow(); + + Table result = TableSummarizer.summarize(t, "g,avg(v),count(v),sum(v)"); + assertNull(result.getAsMap().get("avg(v)").get(0).value); + // sum is null when no values contributed (we don't synthesize 0). + assertNull(result.getAsMap().get("sum(v)").get(0).value); + // count counts all rows regardless. + assertThat(result.getAsMap().get("count(v)").get(0).value, equalTo(2L)); + } + + // ---------- GroupKey correctness ---------- + + public void testGroupKeyDoesNotCollideOnNulCharInValue() { + // Previous string-based key joined values with '\0'. A value containing '\0' could collide + // with a multi-column key. With the new GroupKey (List.equals/hashCode), it cannot. + Table t = new Table(); + t.startHeaders(); + t.addCell("a"); + t.addCell("b"); + t.addCell("v"); + t.endHeaders(); + + // Old hash: "x\0y\0" vs "x\0y" — could collide depending on how values stringify. + // Use a value containing '\0' to demonstrate keys remain distinct. + t.startRow(); + t.addCell("x"); + t.addCell("y\0z"); + t.addCell(1L); + t.endRow(); + t.startRow(); + t.addCell("x\0y"); + t.addCell("z"); + t.addCell(2L); + t.endRow(); + + Table result = TableSummarizer.summarize(t, "a,b,sum(v)"); + // Two distinct groups (rather than one collapsed group). + assertThat(result.getRows().size(), equalTo(2)); + } +} From a07e28e630ecc9167b5a4a58404a60519cc40789 Mon Sep 17 00:00:00 2001 From: Kunal Khatua Date: Fri, 17 Jul 2026 21:31:17 -0700 Subject: [PATCH 2/6] feat(cat): Add limit parameter for _cat APIs Add a 'limit' query parameter that caps the number of rows in a _cat response. Applied at the RestTable level after sort, so ?s=field:desc&limit=N yields the top-N rows by the sort key. The limit applies uniformly to summarized and unsummarized output: - Without aggregation: caps the response to N rows - With aggregation: caps to N groups, with sort applied to the aggregated result so aggregation + sort + limit yields the top-N groups by sort key Example: GET _cat/segments?h=index,sum(docs.count)&s=sum(docs.count):desc&limit=5 Signed-off-by: Kunal Khatua --- .../rest/action/cat/AbstractCatAction.java | 2 +- .../opensearch/rest/action/cat/RestTable.java | 7 + .../rest/action/cat/RestTableTests.java | 180 ++++++++++++++---- 3 files changed, 153 insertions(+), 36 deletions(-) diff --git a/server/src/main/java/org/opensearch/rest/action/cat/AbstractCatAction.java b/server/src/main/java/org/opensearch/rest/action/cat/AbstractCatAction.java index 6b9fa248c0174..7848e0339ceae 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/AbstractCatAction.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/AbstractCatAction.java @@ -91,7 +91,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC } static Set RESPONSE_PARAMS = Collections.unmodifiableSet( - new HashSet<>(Arrays.asList("format", "h", "v", "ts", "pri", "bytes", "size", "time", "s", "timeout")) + new HashSet<>(Arrays.asList("format", "h", "v", "ts", "pri", "bytes", "size", "time", "s", "timeout", "limit")) ); @Override diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java b/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java index 6555bf7f4763b..98af7bf159dd8 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java @@ -212,6 +212,13 @@ static List getRowOrder(Table table, RestRequest request) { } Collections.sort(rowOrder, new TableIndexComparator(table, ordering)); } + + // Apply the limit parameter (if positive). Applied after sort so `?s=field:desc&limit=N` + // yields the top-N rows by the sort key. This also caps unsummarized responses. + int limit = request.paramAsInt("limit", -1); + if (limit > 0 && rowOrder.size() > limit) { + rowOrder = new ArrayList<>(rowOrder.subList(0, limit)); + } return rowOrder; } diff --git a/server/src/test/java/org/opensearch/rest/action/cat/RestTableTests.java b/server/src/test/java/org/opensearch/rest/action/cat/RestTableTests.java index 74e279e9843a3..ae04cd2945222 100644 --- a/server/src/test/java/org/opensearch/rest/action/cat/RestTableTests.java +++ b/server/src/test/java/org/opensearch/rest/action/cat/RestTableTests.java @@ -291,54 +291,99 @@ public void testMultiSort() { assertEquals(Arrays.asList(1, 0, 2), rowOrder); } - private RestResponse assertResponseContentType(Map> headers, String mediaType) throws Exception { - return assertResponseContentType(headers, mediaType, table); + public void testLimitWithoutSummarize() { + Table table = buildSimpleSortableTable(); + restRequest.params().put("limit", "2"); + List rowOrder = RestTable.getRowOrder(table, restRequest); + assertThat(rowOrder.size(), equalTo(2)); + // No sort: rows returned in insertion order (0, 1). + assertEquals(Arrays.asList(0, 1), rowOrder); } - private RestResponse assertResponseContentType(Map> headers, String mediaType, Table table) throws Exception { - FakeRestRequest requestWithAcceptHeader = new FakeRestRequest.Builder(xContentRegistry()).withHeaders(headers).build(); - table.startRow(); - table.addCell("foo"); - table.addCell("foo"); - table.addCell("foo"); - table.addCell("foo"); - table.addCell("foo"); - table.addCell("foo"); - table.addCell("foo"); - table.addCell("foo"); - table.endRow(); - RestResponse response = buildResponse(table, new AbstractRestChannel(requestWithAcceptHeader, true) { - @Override - public void sendResponse(RestResponse response) {} - }); + public void testLimitWithSortReturnsTopN() { + Table table = buildSimpleSortableTable(); + restRequest.params().put("s", "compare:desc"); + restRequest.params().put("limit", "2"); + List rowOrder = RestTable.getRowOrder(table, restRequest); + // Values are [3, 1, 2] at indices [0, 1, 2]. Descending sort: [3, 2, 1] => indices [0, 2, 1]. + // Top-2 by sort key => indices [0, 2]. + assertEquals(Arrays.asList(0, 2), rowOrder); + } - assertThat(response.contentType(), equalTo(mediaType)); - return response; + public void testLimitGreaterThanRowsReturnsAll() { + Table table = buildSimpleSortableTable(); + restRequest.params().put("limit", "100"); + List rowOrder = RestTable.getRowOrder(table, restRequest); + assertThat(rowOrder.size(), equalTo(3)); } - private void assertResponse(Map> headers, String mediaType, String body) throws Exception { - assertResponse(headers, mediaType, body, table); + public void testLimitEqualToRowsReturnsAll() { + Table table = buildSimpleSortableTable(); + restRequest.params().put("limit", "3"); + List rowOrder = RestTable.getRowOrder(table, restRequest); + assertThat(rowOrder.size(), equalTo(3)); } - private void assertResponse(Map> headers, String mediaType, String body, Table table) throws Exception { - RestResponse response = assertResponseContentType(headers, mediaType, table); - assertThat(response.content().utf8ToString(), equalTo(body)); + public void testLimitZeroOrNegativeIsIgnored() { + Table table = buildSimpleSortableTable(); + // limit=0 and any negative value are treated as "no limit" (only positive values apply). + restRequest.params().put("limit", "0"); + assertThat(RestTable.getRowOrder(table, restRequest).size(), equalTo(3)); + + restRequest.params().put("limit", "-5"); + assertThat(RestTable.getRowOrder(table, restRequest).size(), equalTo(3)); } - private List getHeaderNames(List headers) { - List headerNames = new ArrayList<>(); - for (RestTable.DisplayHeader header : headers) { - headerNames.add(header.name); + public void testNoLimitParamReturnsAll() { + Table table = buildSimpleSortableTable(); + // Regression: no limit param at all should return all rows (default behavior unchanged). + List rowOrder = RestTable.getRowOrder(table, restRequest); + assertThat(rowOrder.size(), equalTo(3)); + } + + public void testLimitAppliesToTextOutput() throws Exception { + Table t = new Table(); + t.startHeaders(); + t.addCell("name"); + t.endHeaders(); + for (String n : Arrays.asList("a", "b", "c", "d", "e")) { + t.startRow(); + t.addCell(n); + t.endRow(); } - return headerNames; + FakeRestRequest req = new FakeRestRequest.Builder(xContentRegistry()).withHeaders( + Collections.singletonMap(ACCEPT, Collections.singletonList(TEXT_PLAIN)) + ).build(); + req.params().put("limit", "2"); + RestResponse response = buildResponse(t, new AbstractRestChannel(req, true) { + @Override + public void sendResponse(RestResponse response) {} + }); + // Two lines, each ending with newline. Body should be "a\nb\n". + assertThat(response.content().utf8ToString(), equalTo("a\nb\n")); } - private Table getPaginatedTable() { - PageToken pageToken = new PageToken("foo", "entities"); - Table paginatedTable = new Table(pageToken); - addHeaders(paginatedTable); - return paginatedTable; + public void testLimitAppliesToJsonOutput() throws Exception { + Table t = new Table(); + t.startHeaders(); + t.addCell("name"); + t.endHeaders(); + for (String n : Arrays.asList("a", "b", "c", "d", "e")) { + t.startRow(); + t.addCell(n); + t.endRow(); + } + + FakeRestRequest req = new FakeRestRequest.Builder(xContentRegistry()).withHeaders( + Collections.singletonMap(ACCEPT, Collections.singletonList(APPLICATION_JSON)) + ).build(); + req.params().put("limit", "3"); + RestResponse response = buildResponse(t, new AbstractRestChannel(req, true) { + @Override + public void sendResponse(RestResponse response) {} + }); + assertThat(response.content().utf8ToString(), equalTo("[{\"name\":\"a\"},{\"name\":\"b\"},{\"name\":\"c\"}]")); } public void testHeaderAggregationTriggersSummarizedJsonOutput() throws Exception { @@ -394,6 +439,69 @@ public void sendResponse(RestResponse response) {} assertThat(response.content().utf8ToString(), equalTo("[{\"index\":\"idx-a\"},{\"index\":\"idx-a\"},{\"index\":\"idx-b\"}]")); } + private Table buildSimpleSortableTable() { + Table table = new Table(); + table.startHeaders(); + table.addCell("compare"); + table.endHeaders(); + for (Integer v : Arrays.asList(3, 1, 2)) { + table.startRow(); + table.addCell(v); + table.endRow(); + } + return table; + } + + private RestResponse assertResponseContentType(Map> headers, String mediaType) throws Exception { + return assertResponseContentType(headers, mediaType, table); + } + + private RestResponse assertResponseContentType(Map> headers, String mediaType, Table table) throws Exception { + FakeRestRequest requestWithAcceptHeader = new FakeRestRequest.Builder(xContentRegistry()).withHeaders(headers).build(); + table.startRow(); + table.addCell("foo"); + table.addCell("foo"); + table.addCell("foo"); + table.addCell("foo"); + table.addCell("foo"); + table.addCell("foo"); + table.addCell("foo"); + table.addCell("foo"); + table.endRow(); + RestResponse response = buildResponse(table, new AbstractRestChannel(requestWithAcceptHeader, true) { + @Override + public void sendResponse(RestResponse response) {} + }); + + assertThat(response.contentType(), equalTo(mediaType)); + return response; + } + + private void assertResponse(Map> headers, String mediaType, String body) throws Exception { + assertResponse(headers, mediaType, body, table); + } + + private void assertResponse(Map> headers, String mediaType, String body, Table table) throws Exception { + RestResponse response = assertResponseContentType(headers, mediaType, table); + assertThat(response.content().utf8ToString(), equalTo(body)); + } + + private List getHeaderNames(List headers) { + List headerNames = new ArrayList<>(); + for (RestTable.DisplayHeader header : headers) { + headerNames.add(header.name); + } + + return headerNames; + } + + private Table getPaginatedTable() { + PageToken pageToken = new PageToken("foo", "entities"); + Table paginatedTable = new Table(pageToken); + addHeaders(paginatedTable); + return paginatedTable; + } + private void addHeaders(Table table) { table.startHeaders(); table.addCell("bulk.foo", "alias:f;desc:foo"); @@ -410,4 +518,6 @@ private void addHeaders(Table table) { table.addCell("epoch", "alias:t"); table.endHeaders(); } + + // --- Summarize tests moved to TableSummarizerTests when buildSummarizedTable was extracted --- } From 5e7944d6d5600fe7cc488778729f79ab557d1246 Mon Sep 17 00:00:00 2001 From: Kunal Khatua Date: Fri, 17 Jul 2026 21:32:10 -0700 Subject: [PATCH 3/6] perf(cat): Skip IndicesStats fan-out when no stats columns requested The _cat/shards transport always issued an IndicesStats broadcast to every data node, even when the response did not need a single stats-derived column. For high-shard clusters this is a substantial fan-out cost that contributed nothing to the response. Detect at the REST layer whether any column in h= or s= requires per-shard stats. When all referenced columns are derivable from cluster state (index, shard, prirep, state, node/ip/id, unassigned.*, recoverysource.type, and their aliases), set a new indicesStatsRequired=false flag on the CatShardsRequest. The transport action honors the flag by returning an empty IndicesStatsResponse and short-circuiting the broadcast. The fast path is intentionally conservative: - h= must be explicitly set (default headers include docs/store) - every h= and s= token must match a known routing-only column - wildcards force the slow path (we don't expand them here) - unknown tokens force the slow path Backwards compatibility for the new request field is gated on Version.V_3_7_0 with the default (true) preserving today's behavior on mixed-version clusters. Signed-off-by: Kunal Khatua --- .../cluster/shards/CatShardsRequest.java | 18 ++++ .../shards/TransportCatShardsAction.java | 9 ++ .../rest/action/cat/RestShardsAction.java | 89 +++++++++++++++++++ .../action/cat/RestShardsActionTests.java | 54 +++++++++++ 4 files changed, 170 insertions(+) diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java index 9511a45423e63..152ce09c4f301 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java @@ -32,6 +32,10 @@ public class CatShardsRequest extends ClusterManagerNodeReadRequest headers) { return new ClusterAdminTask(id, type, action, parentTaskId, headers, this.cancelAfterTimeInterval); diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/shards/TransportCatShardsAction.java b/server/src/main/java/org/opensearch/action/admin/cluster/shards/TransportCatShardsAction.java index e044646897481..29ecd1053e7ff 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/shards/TransportCatShardsAction.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/shards/TransportCatShardsAction.java @@ -118,6 +118,15 @@ public void onResponse(ClusterStateResponse clusterStateResponse) { cancellableListener.onResponse(catShardsResponse); return; } + // Fast path: if the REST layer determined that no requested column (h=/s=) + // requires per-shard stats, skip the IndicesStats broadcast entirely. The + // resulting Table has null cells for stats columns, which are not displayed + // because they were not in h=. + if (shardsRequest.isIndicesStatsRequired() == false) { + catShardsResponse.setIndicesStatsResponse(IndicesStatsResponse.getEmptyResponse()); + cancellableListener.onResponse(catShardsResponse); + return; + } IndicesStatsRequest indicesStatsRequest = new IndicesStatsRequest(); indicesStatsRequest.setShouldCancelOnTimeout(true); indicesStatsRequest.all(); diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java b/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java index 7a58c51d10cc9..021d3b3f6de95 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java @@ -71,6 +71,7 @@ import java.time.Instant; import java.util.List; import java.util.Locale; +import java.util.Set; import java.util.function.Function; import static java.util.Arrays.asList; @@ -123,6 +124,7 @@ public RestChannelConsumer doCatRequest(final RestRequest request, final NodeCli shardsRequest.setIndices(indices); shardsRequest.setRequestLimitCheckSupported(isRequestLimitCheckSupported()); shardsRequest.setPageParams(pageParams); + shardsRequest.setIndicesStatsRequired(requestNeedsIndicesStats(request)); parseDeprecatedMasterTimeoutParameter(shardsRequest, request, deprecationLogger, getName()); return channel -> client.execute(CatShardsAction.INSTANCE, shardsRequest, new RestResponseListener(channel) { @Override @@ -141,6 +143,93 @@ public RestResponse buildResponse(CatShardsResponse catShardsResponse) throws Ex }); } + /** + * Canonical names and aliases for shard columns that are derivable purely from cluster state + * (ShardRouting + DiscoveryNodes + UnassignedInfo). These do NOT require a per-shard + * IndicesStats broadcast. + */ + private static final Set ROUTING_ONLY_COLUMNS = Set.of( + // index + "index", + "i", + "idx", + // shard + "shard", + "s", + "sh", + // prirep + "prirep", + "p", + "pr", + "primaryOrReplica", + // state + "state", + "st", + // node identity + "ip", + "id", + "node", + "n", + // unassigned info + "unassigned.reason", + "ur", + "unassigned.at", + "ua", + "unassigned.for", + "uf", + "unassigned.details", + "ud", + // recovery source type (derived from ShardRouting.recoverySource) + "recoverysource.type", + "rs" + ); + + /** + * Returns true if the IndicesStats broadcast is required to build the response. The fast path + * (false) applies only when: + *
    + *
  • h= is explicitly set (the default header set includes 'docs' and 'store' which need stats)
    • + *
  • every token in h= matches a routing-only column (no wildcards, no stats columns)
    • + *
  • s= (if set) only references routing-only columns
    • + *
+ * In all other cases we err on the side of correctness and fetch stats as today. + * + * Package-private for testing. + */ + static boolean requestNeedsIndicesStats(RestRequest request) { + String hParam = request.param("h"); + if (hParam == null || hParam.isEmpty()) { + return true; // default headers include stats columns + } + for (String token : Strings.splitStringByCommaToArray(hParam)) { + if (isRoutingOnlyToken(token) == false) { + return true; + } + } + String sParam = request.param("s"); + if (sParam != null && sParam.isEmpty() == false) { + for (String token : Strings.splitStringByCommaToArray(sParam)) { + // strip :asc / :desc suffix before checking + String col = token; + if (col.endsWith(":desc")) col = col.substring(0, col.length() - 5); + else if (col.endsWith(":asc")) col = col.substring(0, col.length() - 4); + if (isRoutingOnlyToken(col) == false) { + return true; + } + } + } + return false; + } + + private static boolean isRoutingOnlyToken(String token) { + if (token == null) return false; + String trimmed = token.trim(); + if (trimmed.isEmpty()) return false; + // Wildcards conservatively force the slow path — we don't expand them here. + if (trimmed.indexOf('*') >= 0 || trimmed.indexOf('?') >= 0) return false; + return ROUTING_ONLY_COLUMNS.contains(trimmed); + } + @Override protected Table getTableWithHeader(final RestRequest request) { return getTableWithHeader(request, null); diff --git a/server/src/test/java/org/opensearch/rest/action/cat/RestShardsActionTests.java b/server/src/test/java/org/opensearch/rest/action/cat/RestShardsActionTests.java index 8f763f882f250..73b388ac0cff4 100644 --- a/server/src/test/java/org/opensearch/rest/action/cat/RestShardsActionTests.java +++ b/server/src/test/java/org/opensearch/rest/action/cat/RestShardsActionTests.java @@ -145,6 +145,60 @@ public void testBuildTableWithPageToken() { assertEquals("test", table.getPageToken().getPaginatedEntity()); } + // --- Phase 2: routing-only fast path tests --- + + public void testIndicesStatsRequiredWhenNoHParam() { + // No h= → default headers include 'docs' and 'store' which need stats. + assertTrue(RestShardsAction.requestNeedsIndicesStats(new FakeRestRequest())); + } + + public void testIndicesStatsRequiredWhenHIsRoutingOnly() { + FakeRestRequest req = new FakeRestRequest(); + req.params().put("h", "index,shard,prirep,state,node,ip"); + assertFalse(RestShardsAction.requestNeedsIndicesStats(req)); + } + + public void testIndicesStatsRequiredWhenHUsesAliases() { + FakeRestRequest req = new FakeRestRequest(); + // i=index, sh=shard, p=prirep, st=state, n=node, ip=ip, ur=unassigned.reason + req.params().put("h", "i,sh,p,st,n,ip,ur"); + assertFalse(RestShardsAction.requestNeedsIndicesStats(req)); + } + + public void testIndicesStatsRequiredWhenHContainsStatsColumn() { + FakeRestRequest req = new FakeRestRequest(); + req.params().put("h", "index,shard,docs"); // docs requires stats + assertTrue(RestShardsAction.requestNeedsIndicesStats(req)); + } + + public void testIndicesStatsRequiredWhenHContainsWildcard() { + FakeRestRequest req = new FakeRestRequest(); + // Wildcards conservatively force the slow path. + req.params().put("h", "index,shard*"); + assertTrue(RestShardsAction.requestNeedsIndicesStats(req)); + } + + public void testIndicesStatsRequiredWhenSortReferencesStatsColumn() { + FakeRestRequest req = new FakeRestRequest(); + req.params().put("h", "index,shard,node"); + req.params().put("s", "docs:desc"); // sort on stats column → stats still needed + assertTrue(RestShardsAction.requestNeedsIndicesStats(req)); + } + + public void testIndicesStatsNotRequiredWhenSortIsRoutingOnly() { + FakeRestRequest req = new FakeRestRequest(); + req.params().put("h", "index,shard,node"); + req.params().put("s", "index:asc,shard:desc"); // routing columns only + assertFalse(RestShardsAction.requestNeedsIndicesStats(req)); + } + + public void testIndicesStatsRequiredOnUnknownColumn() { + FakeRestRequest req = new FakeRestRequest(); + // Unknown column conservatively forces slow path (don't optimize unrecognized tokens). + req.params().put("h", "index,shard,not_a_column"); + assertTrue(RestShardsAction.requestNeedsIndicesStats(req)); + } + private void assertTable(Table table) { // now, verify the table is correct List headers = table.getHeaders(); From ba9bed9ae0e1a5413f5f533cf0ad6feb52bdb322 Mon Sep 17 00:00:00 2001 From: Kunal Khatua Date: Sun, 19 Jul 2026 08:55:31 -0700 Subject: [PATCH 4/6] test(cat): Avoid reflection in TableSummarizerTests forbiddenApisTest failed because TableSummarizerTests reached the private RestTable#renderValue via Class#getDeclaredMethod + AccessibleObject#setAccessible, which the check bans. Make RestTable#renderValue package-private (the test is in the same package) and call it directly, removing the reflection and the now unused RestRequest import. Signed-off-by: Kunal Khatua --- .../java/org/opensearch/rest/action/cat/RestTable.java | 3 ++- .../opensearch/rest/action/cat/TableSummarizerTests.java | 9 +-------- 2 files changed, 3 insertions(+), 9 deletions(-) diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java b/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java index 98af7bf159dd8..b279eca112da9 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java @@ -405,7 +405,8 @@ public static void pad(Table.Cell cell, int width, RestRequest request, UTF8Stre } } - private static String renderValue(RestRequest request, Object value) { + // package-private for testing + static String renderValue(RestRequest request, Object value) { if (value == null) { return null; } diff --git a/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java b/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java index bcdaa0b1fcc4b..92f28c4f38ee9 100644 --- a/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java +++ b/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java @@ -12,7 +12,6 @@ import org.opensearch.common.unit.SizeValue; import org.opensearch.common.unit.TimeValue; import org.opensearch.core.common.unit.ByteSizeValue; -import org.opensearch.rest.RestRequest; import org.opensearch.test.OpenSearchTestCase; import org.opensearch.test.rest.FakeRestRequest; @@ -83,13 +82,7 @@ private Table buildTimeTable() { } private String renderValueViaResponse(FakeRestRequest request, Object value) { - try { - java.lang.reflect.Method m = RestTable.class.getDeclaredMethod("renderValue", RestRequest.class, Object.class); - m.setAccessible(true); - return (String) m.invoke(null, request, value); - } catch (Exception e) { - throw new RuntimeException(e); - } + return RestTable.renderValue(request, value); } // ---------- hasAggregation ---------- From 30b97ead592b80971787189c84e135f790580d2d Mon Sep 17 00:00:00 2001 From: Kunal Khatua Date: Sun, 19 Jul 2026 19:53:06 -0700 Subject: [PATCH 5/6] fix(cat): Guard aggregate overflow and accept mixed-case tokens Address review feedback on the _cat summarization/limit change: - TableSummarizer.formatValue now converts aggregated doubles to long via a safeLong helper that rounds half-up and clamps to Long.MIN/MAX_VALUE instead of truncating and silently overflowing. This prevents a sum of large ByteSizeValues from wrapping to a negative value (which would throw from the ByteSizeValue constructor). - AGG_FUNC_PATTERN is compiled CASE_INSENSITIVE so SUM(docs)/Count(shard) are recognized consistently by hasAggregation and summarize. - RestShardsAction stats-skip sort parsing trims whitespace and strips the :asc/:desc suffix case-insensitively, so index:DESC or " shard:Asc" still resolve to routing-only columns and take the fast path. Adds tests for case-insensitive aggregation, safeLong rounding/clamping, and mixed-case/whitespace sort suffixes, and updates the TimeValue avg assertion to the corrected round-half-up result. Signed-off-by: Kunal Khatua --- .../rest/action/cat/RestShardsAction.java | 9 +++--- .../rest/action/cat/TableSummarizer.java | 30 ++++++++++++++----- .../action/cat/RestShardsActionTests.java | 8 +++++ .../rest/action/cat/TableSummarizerTests.java | 30 +++++++++++++++++-- 4 files changed, 63 insertions(+), 14 deletions(-) diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java b/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java index 021d3b3f6de95..97cf742f3df96 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java @@ -209,10 +209,11 @@ static boolean requestNeedsIndicesStats(RestRequest request) { String sParam = request.param("s"); if (sParam != null && sParam.isEmpty() == false) { for (String token : Strings.splitStringByCommaToArray(sParam)) { - // strip :asc / :desc suffix before checking - String col = token; - if (col.endsWith(":desc")) col = col.substring(0, col.length() - 5); - else if (col.endsWith(":asc")) col = col.substring(0, col.length() - 4); + // strip :asc / :desc suffix before checking (case-insensitive, whitespace-tolerant) + String col = token.trim(); + String lower = col.toLowerCase(Locale.ROOT); + if (lower.endsWith(":desc")) col = col.substring(0, col.length() - 5); + else if (lower.endsWith(":asc")) col = col.substring(0, col.length() - 4); if (isRoutingOnlyToken(col) == false) { return true; } diff --git a/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java b/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java index 725c2ec85496b..1bb6f616f3b3d 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java @@ -45,8 +45,8 @@ */ public final class TableSummarizer { - /** Matches function syntax like {@code sum(docs.count)}, {@code count(shard)}. */ - private static final Pattern AGG_FUNC_PATTERN = Pattern.compile("^(sum|count|avg|min|max)\\((.+)\\)$"); + /** Matches function syntax like {@code sum(docs.count)}, {@code count(shard)}. Case-insensitive so {@code SUM(docs)} also matches. */ + private static final Pattern AGG_FUNC_PATTERN = Pattern.compile("^(sum|count|avg|min|max)\\((.+)\\)$", Pattern.CASE_INSENSITIVE); private TableSummarizer() {} @@ -311,20 +311,34 @@ static Double parseAsDouble(Object value) { /** Wrap an aggregated double back into the original source type so renderers preserve units. */ static Object formatValue(double num, Object sampleValue) { - if (sampleValue instanceof ByteSizeValue) return new ByteSizeValue((long) num); - if (sampleValue instanceof SizeValue) return new SizeValue((long) num); - if (sampleValue instanceof TimeValue) return new TimeValue((long) num); + if (sampleValue instanceof ByteSizeValue) return new ByteSizeValue(safeLong(num)); + if (sampleValue instanceof SizeValue) return new SizeValue(safeLong(num)); + if (sampleValue instanceof TimeValue) return new TimeValue(safeLong(num)); if (sampleValue instanceof Long || sampleValue instanceof Integer || sampleValue instanceof Short || sampleValue instanceof Byte) { - return (long) num; + return safeLong(num); } if (sampleValue instanceof String) { if (num == Math.floor(num) && !Double.isInfinite(num)) { - return String.valueOf((long) num); + return String.valueOf(safeLong(num)); } return String.format(Locale.ROOT, "%.2f", num); } // Default: return Long when integral (consistent with the original inline impl), else Double. - if (num == Math.floor(num) && !Double.isInfinite(num)) return (long) num; + if (num == Math.floor(num) && !Double.isInfinite(num)) return safeLong(num); return num; } + + /** + * Convert an aggregated double to a long using round-half-up semantics, clamping to + * {@link Long#MIN_VALUE}/{@link Long#MAX_VALUE} instead of silently overflowing. This matters + * for sums of many large {@link ByteSizeValue}s (which can exceed {@code Long.MAX_VALUE} as a + * double) and prevents {@code ByteSizeValue}/{@code SizeValue} from being constructed with a + * wrapped negative value. + */ + static long safeLong(double num) { + if (Double.isNaN(num)) return 0L; + if (num >= (double) Long.MAX_VALUE) return Long.MAX_VALUE; + if (num <= (double) Long.MIN_VALUE) return Long.MIN_VALUE; + return Math.round(num); + } } diff --git a/server/src/test/java/org/opensearch/rest/action/cat/RestShardsActionTests.java b/server/src/test/java/org/opensearch/rest/action/cat/RestShardsActionTests.java index 73b388ac0cff4..ace02d8d442b4 100644 --- a/server/src/test/java/org/opensearch/rest/action/cat/RestShardsActionTests.java +++ b/server/src/test/java/org/opensearch/rest/action/cat/RestShardsActionTests.java @@ -192,6 +192,14 @@ public void testIndicesStatsNotRequiredWhenSortIsRoutingOnly() { assertFalse(RestShardsAction.requestNeedsIndicesStats(req)); } + public void testIndicesStatsNotRequiredWhenSortSuffixIsUpperCaseOrSpaced() { + FakeRestRequest req = new FakeRestRequest(); + req.params().put("h", "index,shard,node"); + // Mixed-case direction suffix and surrounding whitespace must still resolve to routing-only. + req.params().put("s", "index:DESC, shard:Asc"); + assertFalse(RestShardsAction.requestNeedsIndicesStats(req)); + } + public void testIndicesStatsRequiredOnUnknownColumn() { FakeRestRequest req = new FakeRestRequest(); // Unknown column conservatively forces slow path (don't optimize unrecognized tokens). diff --git a/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java b/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java index 92f28c4f38ee9..0a89fef1dd6eb 100644 --- a/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java +++ b/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java @@ -99,6 +99,32 @@ public void testHasAggregationFalseWhenNoFunctionTokens() { assertFalse(TableSummarizer.hasAggregation(null)); } + public void testAggregationFunctionIsCaseInsensitive() { + // Users may pass mixed-case function names; detection and summarization must both accept them. + assertTrue(TableSummarizer.hasAggregation("index,SUM(docs)")); + assertTrue(TableSummarizer.hasAggregation("Count(shard)")); + + Table t = buildNumericTable(); + Table result = TableSummarizer.summarize(t, "index,SUM(docs),Avg(docs)"); + // idx-a: sum=300, avg=150 + assertThat(result.getAsMap().get("SUM(docs)").get(0).value, equalTo(300L)); + assertThat(((Number) result.getAsMap().get("Avg(docs)").get(0).value).longValue(), equalTo(150L)); + } + + public void testSafeLongClampsOverflowAndRounds() { + // Round-half-up rather than truncate. + assertThat(TableSummarizer.safeLong(2.6), equalTo(3L)); + assertThat(TableSummarizer.safeLong(-2.6), equalTo(-3L)); + // Clamp instead of silently wrapping to a negative long. + assertThat(TableSummarizer.safeLong(Double.MAX_VALUE), equalTo(Long.MAX_VALUE)); + assertThat(TableSummarizer.safeLong(-Double.MAX_VALUE), equalTo(Long.MIN_VALUE)); + assertThat(TableSummarizer.safeLong(Double.NaN), equalTo(0L)); + // A ByteSizeValue sum that exceeds Long.MAX_VALUE as a double clamps rather than throwing. + Object clamped = TableSummarizer.formatValue(Double.MAX_VALUE, new ByteSizeValue(1)); + assertTrue(clamped instanceof ByteSizeValue); + assertThat(((ByteSizeValue) clamped).getBytes(), equalTo(Long.MAX_VALUE)); + } + // ---------- summarize returns original table when h= is null/empty or has no aggregation ---------- public void testNullOrEmptyHeadersReturnsOriginal() { @@ -226,8 +252,8 @@ public void testSummarizeTimeValuePreservation() { assertThat(((TimeValue) sumVal).millis(), equalTo(75020L)); assertThat(((TimeValue) minVal).millis(), equalTo(0L)); assertThat(((TimeValue) maxVal).millis(), equalTo(75000L)); - // avg: (0 + 20 + 75000) / 3 = 25006.67 → cast back to TimeValue gives 25006ms - assertThat(((TimeValue) avgVal).millis(), equalTo(25006L)); + // avg: (0 + 20 + 75000) / 3 = 25006.67 → round-half-up to TimeValue gives 25007ms + assertThat(((TimeValue) avgVal).millis(), equalTo(25007L)); } public void testSummarizeTimeValueHumanReadable() { From e38cfd0ea879ddc27114c7b776ec423fffa3f3e6 Mon Sep 17 00:00:00 2001 From: Kunal Khatua Date: Wed, 29 Jul 2026 20:19:40 -0700 Subject: [PATCH 6/6] fix(cat): Address review feedback on summarize + limit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the remaining review comments on PR #22502: - Version gate for the new indicesStatsRequired field in CatShardsRequest moves from V_3_7_0 to V_3_8_0. The 3.8.0 branch has been cut from main, so this change lands in 3.9.0-development; gating on V_3_8_0 keeps the intent (feature only serialized between peers that have it) without fabricating a V_3_9_0 constant that doesn't yet exist in Version.java. Adds mixed-version wire-compatibility tests covering the round-trip on Version.CURRENT and the safe default (indicesStatsRequired=true) when the negotiated peer version is before the gate. - RestTable.buildResponse now reads h= via request.params().get("h") instead of request.param("h"). The buildDisplayHeaders call later in the pipeline still calls request.param("h") and marks it consumed for unrecognized-parameter bookkeeping; using the non-consuming raw map here keeps that ordering unchanged. - TableSummarizer.copyHeaderAttrString now sanitizes ';' and ':' inside attribute values (replacing them with spaces) instead of silently dropping the whole attribute. This prevents a future header with a colon in its 'desc' from losing metadata during the summarize round trip. Includes a unit test that exercises the sanitizer directly with a manually-constructed attr map (Table.addCell's parser can't itself represent a value containing ':', which is precisely the scenario the sanitizer defends against). - Aggregator now tracks countNonNull separately from countAll and countContributing. count(field) returns countNonNull, matching SQL COUNT(field) semantics — a group with three rows one of which is null now reports count(field)=2. Updated the two existing avg/null tests to assert the new semantics. - TableSummarizer.summarize validates that every column referenced in h= (both bare group-by tokens and func(field) arguments) resolves to an existing table column, either directly or via aliasMap. Unknown tokens raise IllegalArgumentException with a message naming the offending token so users no longer get silent empty output from a typo. Tests cover the error paths and confirm alias references continue to work. Test count after this change (cat suites): CatShardsRequestTests=6, TableSummarizerTests=33, RestTableTests=28, all passing. Signed-off-by: Kunal Khatua --- .../cluster/shards/CatShardsRequest.java | 4 +- .../opensearch/rest/action/cat/RestTable.java | 4 +- .../rest/action/cat/TableSummarizer.java | 41 +++++-- .../cluster/shards/CatShardsRequestTests.java | 47 ++++++++ .../rest/action/cat/TableSummarizerTests.java | 112 +++++++++++++++++- 5 files changed, 194 insertions(+), 14 deletions(-) diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java index 152ce09c4f301..d8cd12081fe4b 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java @@ -49,7 +49,7 @@ public CatShardsRequest(StreamInput in) throws IOException { } requestLimitCheckSupported = in.readBoolean(); } - if (in.getVersion().onOrAfter(Version.V_3_7_0)) { + if (in.getVersion().onOrAfter(Version.V_3_8_0)) { indicesStatsRequired = in.readBoolean(); } } @@ -70,7 +70,7 @@ public void writeTo(StreamOutput out) throws IOException { } out.writeBoolean(requestLimitCheckSupported); } - if (out.getVersion().onOrAfter(Version.V_3_7_0)) { + if (out.getVersion().onOrAfter(Version.V_3_8_0)) { out.writeBoolean(indicesStatsRequired); } } diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java b/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java index b279eca112da9..99a1e28e1d263 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/RestTable.java @@ -72,7 +72,9 @@ public class RestTable { public static RestResponse buildResponse(Table table, RestChannel channel) throws Exception { RestRequest request = channel.request(); - String hParam = request.param("h"); + // Non-consuming read: buildDisplayHeaders later calls request.param("h") to mark it + // consumed. Accessing the raw map here avoids reordering param-consumption semantics. + String hParam = request.params().get("h"); // Grouping/aggregation is inferred from h=: if any h= token is an aggregation function // (e.g. sum(docs)), summarize the table; bare tokens are treated as GROUP BY keys. Plain // h= listings (no functions) are returned unchanged by TableSummarizer.summarize. diff --git a/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java b/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java index 1bb6f616f3b3d..e3d3520aac735 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/TableSummarizer.java @@ -114,6 +114,21 @@ public static Table summarize(Table table, String hParam) { } Map> colMap = table.getAsMap(); + + // Validate that every referenced column exists in the table. Silently producing empty or + // misaligned output on a typo would be surprising — reject the request up-front with a + // clear error naming the offending token. + for (int i = 0; i < groupByFields.length; i++) { + if (colMap.containsKey(resolvedGroupBy[i]) == false) { + throw new IllegalArgumentException("Unknown column in h= (group-by): '" + groupByFields[i] + "'"); + } + } + for (AggColumn agg : aggColumns) { + if (colMap.containsKey(agg.field) == false) { + throw new IllegalArgumentException("Unknown column in h= aggregation " + agg.func + "(...): '" + agg.field + "'"); + } + } + int rowCount = table.getRows().size(); // Single-pass online aggregation. For each row we resolve its group key and update the @@ -174,7 +189,8 @@ public static Table summarize(Table table, String hParam) { * metadata. Values containing the attribute delimiters ({@code ;} or {@code :}) are skipped to * avoid producing a string that cannot be parsed back correctly. */ - private static String copyHeaderAttrString(Table.Cell origHeader, String fallbackDesc) { + // Package-private for direct unit testing (delimiter-in-value handling). + static String copyHeaderAttrString(Table.Cell origHeader, String fallbackDesc) { if (origHeader == null || origHeader.attr == null || origHeader.attr.isEmpty()) { return "desc:" + fallbackDesc; } @@ -183,7 +199,13 @@ private static String copyHeaderAttrString(Table.Cell origHeader, String fallbac for (Map.Entry e : origHeader.attr.entrySet()) { String v = e.getValue(); if (v == null) continue; - if (v.indexOf(';') >= 0 || v.indexOf(':') >= 0) continue; // skip un-encodable values + // Attribute strings are parsed on ';' and ':', so delimiter chars inside a value would + // otherwise mangle the parse. Replace them with spaces so the attribute survives the + // round-trip (the desc text is slightly reformatted; alias/text-align are unaffected + // because those values never contain these characters in practice). + if (v.indexOf(';') >= 0 || v.indexOf(':') >= 0) { + v = v.replace(';', ' ').replace(':', ' '); + } if (sb.length() > 0) sb.append(';'); sb.append(e.getKey()).append(':').append(v); if ("desc".equals(e.getKey())) hasDesc = true; @@ -236,13 +258,15 @@ public int hashCode() { /** * Per-(group, column) running aggregator. Maintains the sum, min, max, count of contributing - * (non-null and numerically-parseable) values, and a {@code countAll} including null/unparseable - * rows for the {@code count} function. The first non-null value seen is captured as a - * {@code sampleValue} so the aggregated output can be wrapped back into its source type - * (ByteSizeValue, TimeValue, etc.). + * (non-null and numerically-parseable) values, and separate row-counters for {@code count()} + * semantics: {@code countNonNull} for {@code count(field)} (matches SQL {@code COUNT(field)}), + * and {@code countAll} which counts every row seen regardless of value nullness. The first + * non-null value seen is captured as a {@code sampleValue} so the aggregated output can be + * wrapped back into its source type (ByteSizeValue, TimeValue, etc.). */ static final class Aggregator { long countAll = 0; + long countNonNull = 0; long countContributing = 0; Double sum = null; Double min = null; @@ -252,6 +276,7 @@ static final class Aggregator { void add(Object value) { countAll++; if (value != null) { + countNonNull++; if (sampleValue == null) sampleValue = value; Double num = parseAsDouble(value); if (num != null) { @@ -267,7 +292,9 @@ void add(Object value) { Object getValue(String function) { switch (function) { case "count": - return countAll; + // SQL semantics: COUNT(field) excludes null values. Use countAll if the caller + // wants row count regardless of null (not currently exposed through h=). + return countNonNull; case "sum": return sum == null ? null : formatValue(sum, sampleValue); case "min": diff --git a/server/src/test/java/org/opensearch/action/admin/cluster/shards/CatShardsRequestTests.java b/server/src/test/java/org/opensearch/action/admin/cluster/shards/CatShardsRequestTests.java index e161342c3c609..6fd1214259921 100644 --- a/server/src/test/java/org/opensearch/action/admin/cluster/shards/CatShardsRequestTests.java +++ b/server/src/test/java/org/opensearch/action/admin/cluster/shards/CatShardsRequestTests.java @@ -109,4 +109,51 @@ public void testSerializationWithOlderVersionsParametersNotSerialized() throws E } } } + + // --- Mixed-version wire-compatibility coverage for the indicesStatsRequired field (V_3_8_0 gate) --- + + /** + * At a version on/after the gate (both nodes on a version that knows the field), the flag must + * round-trip exactly in both states. Models same-version feature-to-feature transport. + */ + public void testIndicesStatsRequiredRoundTripsOnCurrentVersion() throws Exception { + for (boolean flag : new boolean[] { true, false }) { + CatShardsRequest request = new CatShardsRequest(); + request.setIndicesStatsRequired(flag); + assertTrue(Version.CURRENT.onOrAfter(Version.V_3_8_0)); + try (BytesStreamOutput out = new BytesStreamOutput()) { + out.setVersion(Version.CURRENT); + request.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + in.setVersion(Version.CURRENT); + CatShardsRequest deserialized = new CatShardsRequest(in); + assertEquals(flag, deserialized.isIndicesStatsRequired()); + } + } + } + } + + /** + * When the negotiated peer version is BEFORE the gate (a true old node that predates the + * feature), the field is NOT written and NOT read, so the stream stays aligned and the reader + * falls back to the safe default (true = fetch stats, i.e. legacy behavior). This is the mixed + * cluster safety guarantee: an old peer never desynchronizes the stream, and the optimization + * simply does not engage. + */ + public void testIndicesStatsRequiredDefaultsTrueWhenPeerBeforeGate() throws Exception { + Version oldVersion = VersionUtils.getPreviousVersion(Version.V_3_8_0); + assertTrue(oldVersion.before(Version.V_3_8_0)); + CatShardsRequest request = new CatShardsRequest(); + request.setIndicesStatsRequired(false); // even if the coordinator wanted to skip stats... + try (BytesStreamOutput out = new BytesStreamOutput()) { + out.setVersion(oldVersion); + request.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + in.setVersion(oldVersion); + CatShardsRequest deserialized = new CatShardsRequest(in); + // ...an old peer reads the safe default and fetches stats as before (no skip). + assertTrue(deserialized.isIndicesStatsRequired()); + } + } + } } diff --git a/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java b/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java index 0a89fef1dd6eb..87c658b0fcf86 100644 --- a/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java +++ b/server/src/test/java/org/opensearch/rest/action/cat/TableSummarizerTests.java @@ -17,6 +17,7 @@ import java.util.concurrent.TimeUnit; +import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.equalTo; public class TableSummarizerTests extends OpenSearchTestCase { @@ -425,8 +426,8 @@ public void testAvgIgnoresNullValuesInDenominator() { Table result = TableSummarizer.summarize(t, "g,avg(v),count(v)"); // avg = (10 + 20) / 2 = 15 (NOT 30/3 = 10) assertThat(((Number) result.getAsMap().get("avg(v)").get(0).value).longValue(), equalTo(15L)); - // count counts ALL rows in the group, including the null row. - assertThat(result.getAsMap().get("count(v)").get(0).value, equalTo(3L)); + // count(field) matches SQL COUNT(field): non-null rows only. Row with null value is excluded. + assertThat(result.getAsMap().get("count(v)").get(0).value, equalTo(2L)); } public void testAvgIsNullWhenAllValuesAreNull() { @@ -449,8 +450,8 @@ public void testAvgIsNullWhenAllValuesAreNull() { assertNull(result.getAsMap().get("avg(v)").get(0).value); // sum is null when no values contributed (we don't synthesize 0). assertNull(result.getAsMap().get("sum(v)").get(0).value); - // count counts all rows regardless. - assertThat(result.getAsMap().get("count(v)").get(0).value, equalTo(2L)); + // count(field) excludes nulls; when every row is null, count is zero. + assertThat(result.getAsMap().get("count(v)").get(0).value, equalTo(0L)); } // ---------- GroupKey correctness ---------- @@ -482,4 +483,107 @@ public void testGroupKeyDoesNotCollideOnNulCharInValue() { // Two distinct groups (rather than one collapsed group). assertThat(result.getRows().size(), equalTo(2)); } + + // ---------- unknown-column validation ---------- + + public void testUnknownGroupByColumnThrows() { + Table t = buildNumericTable(); + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> TableSummarizer.summarize(t, "not_a_column,sum(docs)") + ); + assertEquals("Unknown column in h= (group-by): 'not_a_column'", e.getMessage()); + } + + public void testUnknownAggregationColumnThrows() { + Table t = buildNumericTable(); + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> TableSummarizer.summarize(t, "index,sum(not_a_column)") + ); + assertEquals("Unknown column in h= aggregation sum(...): 'not_a_column'", e.getMessage()); + } + + public void testAliasedGroupByStillResolves() { + // buildNumericTable's "index" column has no alias, so use the shard-cat pattern via the + // aliasMap indirectly: register a synthetic alias by building a table that declares one. + Table t = new Table(); + t.startHeaders(); + t.addCell("index", "alias:i,idx;desc:index name"); + t.addCell("docs", "text-align:right;desc:doc count"); + t.endHeaders(); + t.startRow(); + t.addCell("a"); + t.addCell(10L); + t.endRow(); + t.startRow(); + t.addCell("a"); + t.addCell(20L); + t.endRow(); + + // Alias "i" resolves to "index" — validation must accept aliased references. + Table result = TableSummarizer.summarize(t, "i,sum(docs)"); + assertThat(result.getRows().size(), equalTo(1)); + assertThat(result.getAsMap().get("sum(docs)").get(0).value, equalTo(30L)); + } + + public void testAliasedAggregationFieldStillResolves() { + Table t = new Table(); + t.startHeaders(); + t.addCell("index", "desc:index name"); + t.addCell("docs", "alias:d,dc;text-align:right;desc:doc count"); + t.endHeaders(); + t.startRow(); + t.addCell("a"); + t.addCell(10L); + t.endRow(); + t.startRow(); + t.addCell("a"); + t.addCell(20L); + t.endRow(); + + // Alias "dc" inside sum(...) resolves to "docs". + Table result = TableSummarizer.summarize(t, "index,sum(dc)"); + assertThat(result.getRows().size(), equalTo(1)); + assertThat(result.getAsMap().get("sum(dc)").get(0).value, equalTo(30L)); + } + + // ---------- header attribute sanitization ---------- + + public void testHeaderAttrDelimitersAreSanitizedNotDropped() { + // If a header's desc contains a colon or semicolon, previously copyHeaderAttrString would + // silently drop the whole attribute. It now sanitizes the value (replacing ; and : with + // spaces) so the attribute survives the round-trip. + // + // We call copyHeaderAttrString directly with a manually-constructed attr map because the + // Table.addCell attribute parser cannot itself parse a raw string with ':' inside a value + // — that's exactly the scenario the sanitizer defends against for programmatically-built + // headers. + Table.Cell origHeader = new Table.Cell("index"); + origHeader.attr.put("alias", "i"); + origHeader.attr.put("desc", "contains: colon and; semicolon"); + + String out = TableSummarizer.copyHeaderAttrString(origHeader, "fallback"); + + // The 'desc' attribute must survive (not be silently dropped) and be delimiter-free. + assertThat(out, containsString("desc:contains colon and semicolon")); + // The alias attribute (which never had delimiters) is preserved as-is. + assertThat(out, containsString("alias:i")); + // No stray raw delimiters remain inside values (the outer ';'/':' separators are fine). + // Split on ';' and inspect each entry for balanced key:value shape. + for (String entry : out.split(";")) { + int firstColon = entry.indexOf(':'); + assertTrue("each attr entry should have a key:value shape: " + entry, firstColon > 0); + String value = entry.substring(firstColon + 1); + assertFalse("value must not contain raw ';': " + value, value.contains(";")); + assertFalse("value must not contain raw ':': " + value, value.contains(":")); + } + } + + public void testHeaderAttrEmptyFallsBackToDefaultDesc() { + // Sanity check: an empty attr map yields the fallback desc, and no other attributes. + Table.Cell origHeader = new Table.Cell("index"); + String out = TableSummarizer.copyHeaderAttrString(origHeader, "fallback"); + assertEquals("desc:fallback", out); + } }