Add a generic, extensible rest-endpoint provider SPI

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

@@ -36,6 +36,7 @@ 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_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

@@ -35,6 +35,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,60 @@
+# rest
+
+The `rest` command is a leading command that reads an allow-listed, read-only in-cluster management endpoint 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.
+
+The `rest` command is a generic, extensible framework: a plugin contributes additional read-only endpoints through the `RestEndpointProvider` extension point without changing the grammar. This first version ships a single built-in endpoint, `/_cluster/health`. Additional endpoints (for example `/_cat/nodes`, `/_cat/shards`, `/_cluster/state`, `/_cluster/settings`) can be added in follow-ups, together with optional response redaction applied centrally at the endpoint choke point.
+
+## Enabling the command
+
+`/_cluster/health` is **enabled by default**: `plugins.ppl.rest.allowed_endpoints` defaults to `["/_cluster/health"]`. Any other endpoint is rejected until a deployment adds it to the allow-list (a node-level setting, applied at node startup and not changeable at runtime):
+
+```yaml
+plugins.ppl.rest.allowed_endpoints: ["/_cluster/health"]
+```
+
+Use `["*"]` to allow every endpoint in the curated list below. Set an empty list to disable the command entirely.
+
+## Syntax
+
+```syntax
+rest <endpoint-path> [count=<int>] [<get-arg>=<value> ...]
+```
+
+## 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. |
+| `<get-arg>=<value>` | Optional | Endpoint query arguments, validated per endpoint by both key and value (for example `local=true` for `/_cluster/health`). |
+
+## 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` |
+
+## Example: 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. The `rest` row source composes with downstream `where`, `sort`, `stats`, and `fields` exactly like an index scan, for example `| rest '/_cluster/health' | where status = 'green' | fields status, active_shards`.

docs/user/ppl/index.md

@@ -75,6 +75,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.9 | experimental (since 3.9) | 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,8 @@ testClusters {
         plugin(getJobSchedulerPlugin())
         plugin ':opensearch-sql-plugin'
         testDistribution = 'archive'
+        // Only /_cluster/health is registered; pin the allow-list to it so the rest.md examples run.
+        setting 'plugins.ppl.rest.allowed_endpoints', '/_cluster/health'
     }
 }
 tasks.register("runRestTestCluster", RunTask) {

integ-test/build.gradle

@@ -387,6 +387,8 @@ testClusters {
         plugin(getGeoSpatialPlugin())
         plugin ":opensearch-sql-plugin"
         setting "plugins.query.datasources.encryption.masterkey", "1234567812345678"
+        // Only /_cluster/health is registered; pin the allow-list to it for the rest ITs.
+        setting 'plugins.ppl.rest.allowed_endpoints', '/_cluster/health'
     }
     yamlRestTest {
         testDistribution = 'archive'
@@ -405,6 +407,8 @@ testClusters {
         testDistribution = 'archive'
         plugin(getJobSchedulerPlugin())
         plugin ":opensearch-sql-plugin"
+        // Only /_cluster/health is registered; pin the allow-list to it for RestCommandSecurityIT.
+        setting 'plugins.ppl.rest.allowed_endpoints', '/_cluster/health'
     }
     remoteIntegTestWithSecurity {
         testDistribution = 'archive'
@@ -419,6 +423,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

@@ -59,6 +59,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

@@ -66,6 +66,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 {}

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

@@ -0,0 +1,94 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.calcite.remote;
+
+import static org.opensearch.sql.util.MatcherUtils.rows;
+import static org.opensearch.sql.util.MatcherUtils.schema;
+import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
+import static org.opensearch.sql.util.MatcherUtils.verifySchema;
+
+import java.io.IOException;
+import org.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.opensearch.client.ResponseException;
+import org.opensearch.sql.ppl.PPLIntegTestCase;
+
+/**
+ * Integration tests for the {@code rest} leading command on the Calcite path. This first version
+ * ships a single built-in endpoint, {@code /_cluster/health} (a deterministic single-row endpoint
+ * that carries no network identifiers and needs no redaction). These tests exercise it end to end
+ * and verify the allow-list and per-arg gates.
+ */
+public class CalcitePPLRestIT extends PPLIntegTestCase {
+
+  @Override
+  public void init() throws Exception {
+    super.init();
+    enableCalcite();
+  }
+
+  @Test
+  public void testRestClusterHealthSchema() throws IOException {
+    JSONObject result = executeQuery("| rest '/_cluster/health' | fields status, number_of_nodes");
+    verifySchema(result, schema("status", "string"), schema("number_of_nodes", "int"));
+  }
+
+  @Test
+  public void testRestClusterHealthDataRows() throws IOException {
+    // Single-node test cluster: exactly one node.
+    JSONObject result = executeQuery("| rest '/_cluster/health' | fields number_of_nodes");
+    verifyDataRows(result, rows(1));
+  }
+
+  @Test
+  public void testRestClusterHealthComposesDownstream() throws IOException {
+    // The rest row source composes with downstream stats exactly like an index scan.
+    JSONObject result = executeQuery("| rest '/_cluster/health' | stats count() as cnt");
+    verifyDataRows(result, rows(1));
+  }
+
+  @Test
+  public void testRestClusterHealthLocalArg() throws IOException {
+    // local=true reads health from the local node; on a single-node cluster the row is unchanged.
+    JSONObject result =
+        executeQuery("| rest '/_cluster/health' local='true' | fields number_of_nodes");
+    verifyDataRows(result, rows(1));
+  }
+
+  @Test
+  public void testRestRejectsNonAllowListedEndpoint() {
+    // /_cat/nodes is not registered in this version; it is refused before any transport call.
+    assertRestBadRequest("| rest '/_cat/nodes'", "allow-list");
+  }
+
+  @Test
+  public void testRestRejectsEmptyEndpoint() {
+    assertRestBadRequest("| rest ''", "non-empty path");
+  }
+
+  @Test
+  public void testRestRejectsDisallowedArg() {
+    assertRestBadRequest("| rest '/_cluster/health' h='name'", "does not accept arg");
+  }
+
+  @Test
+  public void testRestRejectsNegativeCount() {
+    assertRestBadRequest("| rest '/_cluster/health' count=-1", "non-negative");
+  }
+
+  /**
+   * Assert a {@code rest} query is refused as a client error: HTTP 400 (not a 500 system error)
+   * with the given substring in the response body. Covers allow-list and bad-argument rejection.
+   */
+  private void assertRestBadRequest(String query, String expectedSubstring) {
+    ResponseException e =
+        org.junit.Assert.assertThrows(ResponseException.class, () -> executeQuery(query));
+    org.junit.Assert.assertEquals(400, e.getResponse().getStatusLine().getStatusCode());
+    org.junit.Assert.assertTrue(
+        "expected [" + expectedSubstring + "] in response body: " + e.getMessage(),
+        e.getMessage().contains(expectedSubstring));
+  }
+}