[Feature] Add PPL rest command

common/src/main/java/org/opensearch/sql/common/setting/Settings.java

@@ -36,6 +36,8 @@ public enum Key {
     PPL_SYNTAX_LEGACY_PREFERRED("plugins.ppl.syntax.legacy.preferred"),
     PPL_SUBSEARCH_MAXOUT("plugins.ppl.subsearch.maxout"),
     PPL_JOIN_SUBSEARCH_MAXOUT("plugins.ppl.join.subsearch_maxout"),
+    PPL_REST_REDACTION_ENABLED("plugins.ppl.rest.redaction.enabled"),
+    PPL_REST_ALLOWED_ENDPOINTS("plugins.ppl.rest.allowed_endpoints"),
 
     /** Enable Calcite as execution engine */
     CALCITE_ENGINE_ENABLED("plugins.calcite.enabled"),

core/src/main/java/org/opensearch/sql/ast/tree/RestRelation.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ast.tree;
+
+import java.util.Collections;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+import org.opensearch.sql.ast.expression.UnresolvedExpression;
+
+/**
+ * Extend Relation to mark a {@code rest} leading command. The single table name is a reserved,
+ * encoded token (produced by {@link org.opensearch.sql.utils.SystemIndexUtils#restTable}) that
+ * carries the validated REST endpoint spec; it resolves through the storage engine to a REST source
+ * table on the Calcite path, exactly as {@link DescribeRelation} resolves to a system index.
+ */
+@ToString
+@EqualsAndHashCode(callSuper = false)
+public class RestRelation extends Relation {
+  public RestRelation(UnresolvedExpression tableName) {
+    super(Collections.singletonList(tableName));
+  }
+}

core/src/main/java/org/opensearch/sql/utils/SystemIndexUtils.java

@@ -5,6 +5,9 @@
 
 package org.opensearch.sql.utils;
 
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.Map;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.experimental.UtilityClass;
@@ -34,6 +37,121 @@ public static Boolean isSystemIndex(String indexName) {
     return indexName.endsWith(SYS_TABLES_SUFFIX);
   }
 
+  /**
+   * Reserved suffix marking a {@code rest} source table. Distinct from {@link #SYS_TABLES_SUFFIX}
+   * so {@link #isSystemIndex} and {@link #isRestSource} never overlap. The whole reserved name is a
+   * single Calcite identifier (REST + lowercase hex + this suffix), so it survives name resolution
+   * the same way the uppercase system-mapping names do.
+   */
+  private static final String REST_SOURCE_SUFFIX = "__REST_SOURCE";
+
+  private static final String REST_SOURCE_PREFIX = "REST";
+
+  /** True if the resolved table name is a {@code rest} source token. */
+  public static boolean isRestSource(String indexName) {
+    return indexName.endsWith(REST_SOURCE_SUFFIX);
+  }
+
+  /**
+   * Encode a validated {@link RestSpec} into a single reserved table name. Mirrors {@link
+   * #mappingTable}: structured metadata travels inside a reserved name rather than a side channel.
+   * The endpoint/args have already been allow-list-validated before this is called.
+   */
+  public static String restTable(RestSpec spec) {
+    StringBuilder sb = new StringBuilder();
+    sb.append("endpoint=").append(spec.getEndpoint());
+    if (spec.getCount() != null) {
+      sb.append('\n').append("count=").append(spec.getCount());
+    }
+    if (spec.getTimeout() != null) {
+      sb.append('\n').append("timeout=").append(spec.getTimeout());
+    }
+    if (spec.getArgs() != null) {
+      for (Map.Entry<String, String> e : spec.getArgs().entrySet()) {
+        sb.append('\n').append("arg.").append(e.getKey()).append('=').append(e.getValue());
+      }
+    }
+    return REST_SOURCE_PREFIX + toHex(sb.toString()) + REST_SOURCE_SUFFIX;
+  }
+
+  /** Decode a reserved {@code rest} table name back into its {@link RestSpec}. */
+  public static RestSpec decodeRestSpec(String indexName) {
+    // Validate the token shape before slicing it: callers gate on isRestSource today, but a
+    // public decoder must not assume its precondition, otherwise a malformed token would throw an
+    // opaque StringIndexOutOfBoundsException from substring rather than a clear input error.
+    if (!isRestSource(indexName)) {
+      throw new IllegalArgumentException("not a valid rest source token: " + indexName);
+    }
+    String body =
+        indexName.substring(
+            REST_SOURCE_PREFIX.length(), indexName.length() - REST_SOURCE_SUFFIX.length());
+    String decoded = fromHex(body);
+    String endpoint = null;
+    Integer count = null;
+    String timeout = null;
+    LinkedHashMap<String, String> args = new LinkedHashMap<>();
+    for (String line : decoded.split("\n")) {
+      if (line.isEmpty()) {
+        continue;
+      }
+      int eq = line.indexOf('=');
+      if (eq < 0) {
+        continue;
+      }
+      String k = line.substring(0, eq);
+      String v = line.substring(eq + 1);
+      if (k.equals("endpoint")) {
+        endpoint = v;
+      } else if (k.equals("count")) {
+        count = Integer.parseInt(v);
+      } else if (k.equals("timeout")) {
+        timeout = v;
+      } else if (k.startsWith("arg.")) {
+        args.put(k.substring("arg.".length()), v);
+      }
+    }
+    if (endpoint == null) {
+      throw new IllegalArgumentException("rest source token is missing the endpoint: " + indexName);
+    }
+    return new RestSpec(endpoint, args, count, timeout);
+  }
+
+  private static String toHex(String s) {
+    StringBuilder h = new StringBuilder();
+    for (byte b : s.getBytes(StandardCharsets.UTF_8)) {
+      h.append(Character.forDigit((b >> 4) & 0xF, 16)).append(Character.forDigit(b & 0xF, 16));
+    }
+    return h.toString();
+  }
+
+  private static String fromHex(String h) {
+    if (h.length() % 2 != 0) {
+      throw new IllegalArgumentException("not a valid rest source token: odd-length hex body");
+    }
+    byte[] bytes = new byte[h.length() / 2];
+    for (int i = 0; i < bytes.length; i++) {
+      bytes[i] =
+          (byte)
+              ((Character.digit(h.charAt(2 * i), 16) << 4)
+                  + Character.digit(h.charAt(2 * i + 1), 16));
+    }
+    return new String(bytes, StandardCharsets.UTF_8);
+  }
+
+  /**
+   * The validated spec for a {@code rest} command: an allow-listed read-only endpoint plus optional
+   * count/timeout/query args. Lives in core so the parser (encode) and the storage engine (decode)
+   * share it without a cross-module dependency.
+   */
+  @Getter
+  @RequiredArgsConstructor
+  public static class RestSpec {
+    private final String endpoint;
+    private final Map<String, String> args;
+    private final Integer count;
+    private final String timeout;
+  }
+
   /**
    * Compose system mapping table.
    *

docs/category.json

@@ -34,6 +34,7 @@
     "user/ppl/cmd/rename.md",
     "user/ppl/cmd/multisearch.md",
     "user/ppl/cmd/replace.md",
+    "user/ppl/cmd/rest.md",
     "user/ppl/cmd/rex.md",
     "user/ppl/cmd/search.md",
     "user/ppl/cmd/showdatasources.md",

docs/user/ppl/cmd/rest.md

@@ -0,0 +1,94 @@
+# rest
+
+The `rest` command is a leading command that reads an allow-listed, read-only in-cluster management endpoint (cluster/cat/nodes) and emits the response as PPL rows. Its rows come from the endpoint dispatch, not from an index, so `rest` appears at the start of a query.
+
+> **Note**: The `rest` command is supported only on the Calcite query engine (`plugins.calcite.enabled=true`). Each endpoint has a fixed output schema, and the dispatch runs under the caller's security context, so a user who cannot call an endpoint directly cannot call it through `rest`. The command is read-only; mutating and non-allow-listed endpoints are rejected. Each endpoint requires the same cluster-monitor privilege as calling it natively, so `rest` grants no extra access. Some allow-listed endpoints surface operational metadata (for example `/_cat/nodes` exposes node addresses and resource utilization, `/_cat/plugins` the installed plugin inventory, and `/_cluster/state` cluster-state identifiers); this is a deliberate, read-only, monitor-privileged trade-off. `/_cluster/settings` is redacted with the node's setting filter so `Property.Filtered` keys are not surfaced.
+
+## Enabling the command
+
+The `rest` command is **disabled by default**: `plugins.ppl.rest.allowed_endpoints` defaults to an empty list, so every endpoint is rejected until a deployment explicitly opts in. Enable specific endpoints by setting the allow-list (a node-level setting, so it is applied at node startup and cannot be changed at runtime):
+
+```yaml
+plugins.ppl.rest.allowed_endpoints: ["/_cluster/health", "/_cat/nodes"]
+```
+
+Use `["*"]` to allow every endpoint in the curated list below. An empty list (the default) disables the command entirely.
+
+The `rest` command also supports optional response redaction of network identifiers (IPv4/IPv6 addresses, `inet[...]` forms, EC2-style host names, and availability-zone names) in `/_cat/*` and `/_cluster/settings` cell values, controlled by `plugins.ppl.rest.redaction.enabled` (a node-level setting, default `false`). Managed deployments that must not expose host topology should set it to `true`.
+
+## Syntax
+
+The `rest` command has the following syntax:
+
+```syntax
+rest <endpoint-path> [count=<int>] [timeout=<duration>] [<get-arg>=<value> ...]
+```
+
+## Parameters
+
+The `rest` command supports the following parameters.
+
+| Parameter | Required/Optional | Description |
+| --- | --- | --- |
+| `<endpoint-path>` | Required | An allow-listed, read-only endpoint path (see the allow-list below), for example `/_cluster/health`. |
+| `count=<int>` | Optional | Caps the number of emitted rows. |
+| `timeout=<duration>` | Optional | Reserved for forward compatibility. It is currently rejected with a clear error, because a single uniform timeout does not map cleanly across the different endpoints. |
+| `<get-arg>=<value>` | Optional | Endpoint query arguments, validated per endpoint by both key and value (for example `local=true` for `/_cluster/health`, `health=green` for `/_cat/indices`, `expand_wildcards=open` for `/_resolve/index`). |
+
+## Allow-list
+
+`rest` resolves only an explicit, curated set of read-only endpoints. Anything outside the list, including any mutating endpoint, is rejected with a clear error.
+
+| Endpoint | Output columns | Accepted args |
+| --- | --- | --- |
+| `/_cluster/health` | `cluster_name` (string), `status` (string), `number_of_nodes` (integer), `number_of_data_nodes` (integer), `active_primary_shards` (integer), `active_shards` (integer), `relocating_shards` (integer), `initializing_shards` (integer), `unassigned_shards` (integer), `timed_out` (boolean) | `local` |
+| `/_cluster/state` | `cluster_name` (string), `state_uuid` (string), `version` (long), `cluster_manager_node` (string) | (none) |
+| `/_cluster/settings` | `setting` (string), `value` (string), `tier` (string) | (none) |
+| `/_cat/indices` | `index` (string), `health` (string), `pri` (integer), `rep` (integer), `active_shards` (integer) | `health` |
+| `/_cat/nodes` | `name` (string), `ip` (string), `node_role` (string), `heap_percent` (integer), `ram_percent` (integer), `cpu` (integer) | (none) |
+| `/_cat/cluster_manager` | `id` (string), `host` (string), `ip` (string), `node` (string) | (none) |
+| `/_cat/plugins` | `name` (string), `component` (string), `version` (string) | (none) |
+| `/_cat/shards` | `index` (string), `shard` (integer), `prirep` (string), `state` (string), `node` (string) | (none) |
+| `/_resolve/index` | `name` (string), `type` (string) | `expand_wildcards` |
+
+## Example 1: Counting the nodes in the cluster
+
+The following query reads cluster health and projects a column that is deterministic on a single-node cluster:
+
+```ppl
+| rest '/_cluster/health' | fields number_of_nodes
+```
+
+The query returns the following results:
+
+```text
+fetched rows / total rows = 1/1
++-----------------+
+| number_of_nodes |
+|-----------------|
+| 1               |
++-----------------+
+```
+
+`/_cluster/health` also exposes `status`, `active_shards`, and the other columns listed in the allow-list, which you can project and filter the same way.
+
+## Example 2: Composing downstream commands over a cat endpoint
+
+The `rest` row source composes with downstream `where`, `sort`, `stats`, and `fields` exactly like an index scan. The following query reads `/_cat/cluster_manager` and counts the rows:
+
+```ppl
+| rest '/_cat/cluster_manager' | stats count() as managers
+```
+
+The query returns the following results:
+
+```text
+fetched rows / total rows = 1/1
++----------+
+| managers |
+|----------|
+| 1        |
++----------+
+```
+
+For example, `| rest '/_cat/indices' | where health = 'green' | sort index | fields index, health, pri` lists green indexes; the projected columns come from the endpoint's fixed schema.

docs/user/ppl/index.md

@@ -74,6 +74,7 @@ source=accounts
 | [lookup command](cmd/lookup.md) | 3.0 | experimental (since 3.0) | Add or replace data from a lookup index. |
 | [multisearch command](cmd/multisearch.md) | 3.4 | experimental (since 3.4) | Execute multiple search queries and combine their results. |
 | [union command](cmd/union.md) | 3.7 | experimental (since 3.7) | Combine results from multiple datasets using UNION ALL semantics. |
+| [rest command](cmd/rest.md) | 3.8 | experimental (since 3.8) | Read an allow-listed, read-only in-cluster management endpoint (cluster/cat/nodes) as rows. Calcite engine only. |
 | [ml command](cmd/ml.md) | 2.5 | stable (since 2.5) | Apply machine learning algorithms to analyze data. |
 | [kmeans command](cmd/kmeans.md) | 1.3 | stable (since 1.3) | Apply the kmeans algorithm on the search result returned by a PPL command. |
 | [ad command](cmd/ad.md) | 1.3 | deprecated (since 2.5) | Apply Random Cut Forest algorithm on the search result returned by a PPL command. |

doctest/build.gradle

@@ -205,6 +205,11 @@ testClusters {
         plugin(getJobSchedulerPlugin())
         plugin ':opensearch-sql-plugin'
         testDistribution = 'archive'
+        // The rest command is disabled by default (empty allow-list). Opt the doctest cluster into
+        // the registered endpoints so the rest.md examples run against an enabled command. A literal
+        // "*" cannot be used because a bare * is a YAML alias indicator in opensearch.yml.
+        setting 'plugins.ppl.rest.allowed_endpoints',
+                '/_cluster/health,/_cluster/state,/_cluster/settings,/_cat/indices,/_cat/nodes,/_cat/cluster_manager,/_cat/plugins,/_cat/shards,/_resolve/index'
     }
 }
 tasks.register("runRestTestCluster", RunTask) {

integ-test/build.gradle

@@ -387,6 +387,11 @@ testClusters {
         plugin(getGeoSpatialPlugin())
         plugin ":opensearch-sql-plugin"
         setting "plugins.query.datasources.encryption.masterkey", "1234567812345678"
+        // The rest command is disabled by default (empty allow-list). Opt this cluster into the
+        // registered endpoints so the rest integration tests exercise the enabled path. A literal
+        // "*" cannot be used because a bare * is a YAML alias indicator in opensearch.yml.
+        setting 'plugins.ppl.rest.allowed_endpoints',
+                '/_cluster/health,/_cluster/state,/_cluster/settings,/_cat/indices,/_cat/nodes,/_cat/cluster_manager,/_cat/plugins,/_cat/shards,/_resolve/index'
     }
     yamlRestTest {
         testDistribution = 'archive'
@@ -405,6 +410,9 @@ testClusters {
         testDistribution = 'archive'
         plugin(getJobSchedulerPlugin())
         plugin ":opensearch-sql-plugin"
+        // Opt into the rest endpoints (disabled by default) so RestCommandSecurityIT runs.
+        setting 'plugins.ppl.rest.allowed_endpoints',
+                '/_cluster/health,/_cluster/state,/_cluster/settings,/_cat/indices,/_cat/nodes,/_cat/cluster_manager,/_cat/plugins,/_cat/shards,/_resolve/index'
     }
     remoteIntegTestWithSecurity {
         testDistribution = 'archive'
@@ -419,6 +427,8 @@ testClusters {
         plugin(getArrowFlightRpcPlugin())
         plugin(getAnalyticsEnginePlugin())
         plugin ":opensearch-sql-plugin"
+        // Composite-default cluster: PPL queries route to the analytics engine unless excluded.
+        setting 'cluster.pluggable.dataformat', 'composite'
     }
 }
 

integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java

@@ -57,6 +57,7 @@
   CalciteObjectFieldOperateIT.class,
   CalciteOperatorIT.class,
   CalciteParseCommandIT.class,
+  CalcitePPLRestIT.class,
   CalcitePPLAggregationIT.class,
   CalcitePPLAppendcolIT.class,
   CalcitePPLAppendCommandIT.class,

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java

@@ -64,6 +64,15 @@ public void init() throws Exception {
     loadIndex(Index.GRAPH_EMPLOYEES);
   }
 
+  // Only for Calcite: the rest row source explains as a CalciteScannableCatalogScan.
+  @Test
+  public void explainRestCommand() throws IOException {
+    String result = explainQueryToString("| rest '/_cluster/health' | fields status");
+    Assert.assertTrue(
+        "Expected a rest scan node in the explain output, got: " + result,
+        result.contains("CatalogScan"));
+  }
+
   @Override
   @Ignore("test only in v2")
   public void testExplainModeUnsupportedInV2() throws IOException {}