[QA] Add AppendPipeCommandIT and TableCommandIT for the analytics-engine REST path

sandbox/plugins/test-ppl-frontend/src/main/java/org/opensearch/ppl/action/UnifiedQueryService.java

@@ -80,6 +80,11 @@ protected Map<String, Table> getTableMap() {
                 .language(QueryType.PPL)
                 .catalog(DEFAULT_CATALOG, flatSchema)
                 .defaultNamespace(DEFAULT_CATALOG)
+                // The unified PPL parser reuses the v2 AstBuilder, which gates Calcite-only
+                // commands (table, regex, rex, convert) on plugins.calcite.enabled. The unified
+                // path is by definition Calcite-based — flag it on so those commands lower
+                // through the same Project/Filter RelNodes as their non-aliased counterparts.
+                .setting("plugins.calcite.enabled", true)
                 .build()
         ) {
 

sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/AppendPipeCommandIT.java

@@ -0,0 +1,232 @@
+/*
+ * 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.analytics.qa;
+
+import org.opensearch.client.Request;
+import org.opensearch.client.Response;
+import org.opensearch.client.ResponseException;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Self-contained integration test for PPL {@code appendpipe} on the analytics-engine route.
+ *
+ * <p>Mirrors {@code CalcitePPLAppendPipeCommandIT} from the {@code opensearch-project/sql}
+ * repository so the analytics-engine path can be verified inside core without cross-plugin
+ * dependencies. Each test sends a PPL query through {@code POST /_analytics/ppl} (exposed
+ * by the {@code test-ppl-frontend} plugin), which runs the same {@code UnifiedQueryPlanner}
+ * → {@code CalciteRelNodeVisitor} → Substrait → DataFusion pipeline as the SQL plugin's
+ * force-routed analytics path.
+ *
+ * <p>{@code appendpipe} differs from {@code append} (covered by {@link AppendCommandIT}):
+ * {@code appendpipe [pipeline]} duplicates the current intermediate result, applies the
+ * inline {@code [pipeline]} to the duplicate, and appends the duplicate's output to the
+ * original. {@code append [search]} runs an entirely separate sub-query and unions its
+ * output. Both lower to a Calcite {@code LogicalUnion} but the upper-stage shape differs
+ * because {@code appendpipe} reuses the original's row stream as its input rather than
+ * starting a fresh {@code source=...}.
+ *
+ * <p>Provisions the {@code calcs} dataset once. {@link AnalyticsRestTestCase#preserveIndicesUponCompletion()}
+ * keeps it across test methods.
+ */
+public class AppendPipeCommandIT extends AnalyticsRestTestCase {
+
+    private static final Dataset DATASET = new Dataset("calcs", "calcs");
+
+    private static boolean dataProvisioned = false;
+
+    private void ensureDataProvisioned() throws IOException {
+        if (dataProvisioned == false) {
+            DatasetProvisioner.provision(client(), DATASET);
+            dataProvisioned = true;
+        }
+    }
+
+    // ── duplicate + inline sort, then head ──────────────────────────────────────
+
+    public void testAppendPipeSort() throws IOException {
+        // Branch: stats sum(int0) by str0 → 3 rows (FURNITURE=1, OFFICE SUPPLIES=18, TECHNOLOGY=49).
+        // Outer `sort str0` pins the original to alphabetical order. `appendpipe [sort -sum_int0_by_str0]`
+        // duplicates the 3 rows and re-sorts them descending, then appends. `head 5` keeps the first
+        // 5 of the 6 total rows: original 3 + first 2 of the descending duplicate.
+        assertRows(
+            "source="
+                + DATASET.indexName
+                + " | stats sum(int0) as sum_int0_by_str0 by str0 | sort str0"
+                + " | appendpipe [ sort -sum_int0_by_str0 ]"
+                + " | head 5",
+            row(1, "FURNITURE"),
+            row(18, "OFFICE SUPPLIES"),
+            row(49, "TECHNOLOGY"),
+            row(49, "TECHNOLOGY"),
+            row(18, "OFFICE SUPPLIES")
+        );
+    }
+
+    // ── duplicate + inline stats producing a smaller schema (merged column) ─────
+
+    public void testAppendPipeWithMergedColumn() throws IOException {
+        // Outer stats: sum(int0) by str0 → 3 rows. `appendpipe [stats sum(sum) as sum]` runs an inner
+        // stats over the duplicate, collapsing it to a single row carrying only the `sum` column.
+        // Schema unification keeps both the original branch's `str0` and the inner branch's
+        // `sum` column; the inner row is null-padded for the missing `str0`. The two branches
+        // arrive at the coordinator's union in non-deterministic order (each is its own data-node
+        // stage), so compare as a multiset rather than positionally.
+        assertRowsAnyOrder(
+            "source="
+                + DATASET.indexName
+                + " | stats sum(int0) as sum by str0 | sort str0"
+                + " | appendpipe [ stats sum(sum) as sum ]",
+            row(1, "FURNITURE"),
+            row(18, "OFFICE SUPPLIES"),
+            row(49, "TECHNOLOGY"),
+            row(68, null)
+        );
+    }
+
+    // ── duplicate + inline cast that clashes with the original's column type ───
+
+    public void testAppendPipeWithConflictTypeColumn() {
+        // Branch 1 produces `sum` as BIGINT (sum over int0). The inner pipeline of
+        // `appendpipe [eval sum = cast(sum as double)]` rewrites the same-named column to
+        // DOUBLE. SchemaUnifier refuses to merge the diverging types and surfaces a
+        // planner-side validation error before execution.
+        assertErrorContains(
+            "source="
+                + DATASET.indexName
+                + " | stats sum(int0) as sum by str0 | sort str0"
+                + " | appendpipe [ eval sum = cast(sum as double) ]"
+                + " | head 5",
+            "due to incompatible types"
+        );
+    }
+
+    // ── helpers ─────────────────────────────────────────────────────────────────
+
+    private static List<Object> row(Object... values) {
+        return Arrays.asList(values);
+    }
+
+    /**
+     * Multiset comparison — branch ordering at the coordinator's Union is non-deterministic.
+     * Used by {@link #testAppendPipeWithMergedColumn} where the original-branch stats output
+     * (3 rows) and the inner-branch collapsed-sum (1 row) can arrive in either order.
+     */
+    @SafeVarargs
+    @SuppressWarnings("varargs")
+    private final void assertRowsAnyOrder(String ppl, List<Object>... expected) throws IOException {
+        Map<String, Object> response = executePpl(ppl);
+        @SuppressWarnings("unchecked")
+        List<List<Object>> actualRows = (List<List<Object>>) response.get("rows");
+        assertNotNull("Response missing 'rows' for query: " + ppl, actualRows);
+        assertEquals("Row count mismatch for query: " + ppl, expected.length, actualRows.size());
+        java.util.List<List<Object>> remaining = new java.util.ArrayList<>(actualRows);
+        outer:
+        for (List<Object> want : expected) {
+            for (int i = 0; i < remaining.size(); i++) {
+                if (rowsEqual(want, remaining.get(i))) {
+                    remaining.remove(i);
+                    continue outer;
+                }
+            }
+            fail("Expected row not found for query: " + ppl + " — missing: " + want + " in actual: " + actualRows);
+        }
+    }
+
+    private static boolean rowsEqual(List<Object> a, List<Object> b) {
+        if (a.size() != b.size()) return false;
+        for (int i = 0; i < a.size(); i++) {
+            Object ax = a.get(i);
+            Object bx = b.get(i);
+            if (ax == null || bx == null) {
+                if (ax != bx) return false;
+                continue;
+            }
+            if (ax instanceof Number && bx instanceof Number) {
+                if (Double.compare(((Number) ax).doubleValue(), ((Number) bx).doubleValue()) != 0) return false;
+                continue;
+            }
+            if (!ax.equals(bx)) return false;
+        }
+        return true;
+    }
+
+    @SafeVarargs
+    @SuppressWarnings("varargs")
+    private final void assertRows(String ppl, List<Object>... expected) throws IOException {
+        Map<String, Object> response = executePpl(ppl);
+        @SuppressWarnings("unchecked")
+        List<List<Object>> actualRows = (List<List<Object>>) response.get("rows");
+        assertNotNull("Response missing 'rows' for query: " + ppl, actualRows);
+        assertEquals("Row count mismatch for query: " + ppl, expected.length, actualRows.size());
+        for (int i = 0; i < expected.length; i++) {
+            List<Object> want = expected[i];
+            List<Object> got = actualRows.get(i);
+            assertEquals(
+                "Column count mismatch at row " + i + " for query: " + ppl,
+                want.size(),
+                got.size()
+            );
+            for (int j = 0; j < want.size(); j++) {
+                assertCellEquals(
+                    "Cell mismatch at row " + i + ", col " + j + " for query: " + ppl,
+                    want.get(j),
+                    got.get(j)
+                );
+            }
+        }
+    }
+
+    private void assertErrorContains(String ppl, String expectedSubstring) {
+        try {
+            Map<String, Object> response = executePpl(ppl);
+            fail("Expected query to fail with [" + expectedSubstring + "] but got response: " + response);
+        } catch (ResponseException e) {
+            String body;
+            try {
+                body = org.opensearch.test.rest.OpenSearchRestTestCase.entityAsMap(e.getResponse()).toString();
+            } catch (IOException ioe) {
+                body = e.getMessage();
+            }
+            assertTrue(
+                "Expected response body to contain [" + expectedSubstring + "] but was: " + body,
+                body.contains(expectedSubstring)
+            );
+        } catch (IOException e) {
+            fail("Unexpected IOException: " + e);
+        }
+    }
+
+    private Map<String, Object> executePpl(String ppl) throws IOException {
+        ensureDataProvisioned();
+        Request request = new Request("POST", "/_analytics/ppl");
+        request.setJsonEntity("{\"query\": \"" + escapeJson(ppl) + "\"}");
+        Response response = client().performRequest(request);
+        return assertOkAndParse(response, "PPL: " + ppl);
+    }
+
+    private static void assertCellEquals(String message, Object expected, Object actual) {
+        if (expected == null || actual == null) {
+            assertEquals(message, expected, actual);
+            return;
+        }
+        if (expected instanceof Number && actual instanceof Number) {
+            double e = ((Number) expected).doubleValue();
+            double a = ((Number) actual).doubleValue();
+            if (Double.compare(e, a) != 0) {
+                fail(message + ": expected <" + expected + "> but was <" + actual + ">");
+            }
+            return;
+        }
+        assertEquals(message, expected, actual);
+    }
+}

sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/TableCommandIT.java

@@ -0,0 +1,143 @@
+/*
+ * 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.analytics.qa;
+
+import org.opensearch.client.Request;
+import org.opensearch.client.Response;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Self-contained integration test for PPL {@code table} on the analytics-engine route.
+ *
+ * <p>{@code table} is a syntactic alias of {@code fields} — the SQL plugin's
+ * {@code AstBuilder.visitTableCommand} reuses {@code buildProjectCommand} (the same
+ * code path {@code fields} dispatches to) once {@code plugins.calcite.enabled=true} is
+ * propagated through the {@code UnifiedQueryContext} (see
+ * <a href="https://github.com/opensearch-project/sql/pull/5413">opensearch-project/sql#5413</a>).
+ * The added value of {@code table} is a more permissive token shape: it accepts
+ * space-delimited field lists, leading-{@code -} exclusion forms, and mixes those with
+ * commas — surfaces {@code fields} doesn't expose.
+ *
+ * <p>This IT covers the surfaces specific to the {@code table} keyword to lock in that
+ * the analytics path lowers them to the same Calcite {@code Project} RelNode as the v2 /
+ * Calcite path does. Plain projection semantics (already covered by {@code FieldsCommandIT})
+ * are not duplicated here.
+ *
+ * <p>Reuses the {@code calcs} parquet-backed dataset.
+ */
+public class TableCommandIT extends AnalyticsRestTestCase {
+
+    private static final Dataset DATASET = new Dataset("calcs", "calcs");
+
+    private static boolean dataProvisioned = false;
+
+    private void ensureDataProvisioned() throws IOException {
+        if (dataProvisioned == false) {
+            DatasetProvisioner.provision(client(), DATASET);
+            dataProvisioned = true;
+        }
+    }
+
+    public void testTableCommaDelimited() throws IOException {
+        // Comma-delimited form — same shape as `fields a, b`. Sanity check that the table
+        // keyword reaches buildProjectCommand without falling back to the v2-only error.
+        assertColumns(
+            "source=" + DATASET.indexName + " | table str0, num0 | head 3",
+            "str0",
+            "num0"
+        );
+    }
+
+    public void testTableSpaceDelimited() throws IOException {
+        // Space-delimited form — unique to `table`. Validates the lexer accepts whitespace as
+        // a separator and the AstBuilder folds the multi-token list into a single Project.
+        assertColumns(
+            "source=" + DATASET.indexName + " | table str0 num0 int0 | head 3",
+            "str0",
+            "num0",
+            "int0"
+        );
+    }
+
+    public void testTableSuffixWildcard() throws IOException {
+        // *0 expands at parse time to all columns ending in '0'. Identical to
+        // FieldsCommandIT.testFieldsSuffixWildcard on the analytics path; pinned here
+        // for the `table` lowering specifically. Order is analyzer-dependent, so set-equality.
+        Map<String, Object> response = executePpl(
+            "source=" + DATASET.indexName + " | table *0 | head 1"
+        );
+        @SuppressWarnings("unchecked")
+        List<String> columns = (List<String>) response.get("columns");
+        assertNotNull("Response missing 'columns'", columns);
+        java.util.Set<String> actual = new java.util.HashSet<>(columns);
+        java.util.Set<String> expected = new java.util.HashSet<>(
+            java.util.Arrays.asList("num0", "str0", "int0", "bool0", "date0", "time0", "datetime0")
+        );
+        assertEquals("Wildcard *0 column set", expected, actual);
+    }
+
+    public void testTableMinusExclusion() throws IOException {
+        // `table - num0, num1, num2, num3, num4` removes those five columns. The leading
+        // minus form is unique to `table`; `fields` uses `fields - a, b, ...` with a
+        // comma-separated list (no space-delimiting). Validates analytics path retains
+        // exclusion semantics.
+        Map<String, Object> response = executePpl(
+            "source=" + DATASET.indexName + " | table - num0, num1, num2, num3, num4 | head 1"
+        );
+        @SuppressWarnings("unchecked")
+        List<String> columns = (List<String>) response.get("columns");
+        assertNotNull("Response missing 'columns'", columns);
+        for (String name : columns) {
+            assertFalse("Excluded column should not appear: " + name, name.startsWith("num"));
+        }
+    }
+
+    public void testFieldsAndTableEquivalence() throws IOException {
+        // Cross-check that `fields a, b, c` and `table a, b, c` produce identical
+        // schema + rows. Makes the alias claim explicit at the response level so a
+        // future divergence (e.g. `table` accidentally adds a Sort or rewires the
+        // Project) is caught here.
+        Map<String, Object> fieldsResp = executePpl(
+            "source=" + DATASET.indexName + " | fields str0, num0, int0 | head 3"
+        );
+        Map<String, Object> tableResp = executePpl(
+            "source=" + DATASET.indexName + " | table str0, num0, int0 | head 3"
+        );
+        assertEquals("columns from fields vs table", fieldsResp.get("columns"), tableResp.get("columns"));
+        assertEquals("rows from fields vs table", fieldsResp.get("rows"), tableResp.get("rows"));
+    }
+
+    // ── helpers ─────────────────────────────────────────────────────────────────
+
+    private void assertColumns(String ppl, String... expectedColumns) throws IOException {
+        Map<String, Object> response = executePpl(ppl);
+        @SuppressWarnings("unchecked")
+        List<String> columns = (List<String>) response.get("columns");
+        assertNotNull("Response missing 'columns' for query: " + ppl, columns);
+        assertEquals("Column count for query: " + ppl, expectedColumns.length, columns.size());
+        for (int i = 0; i < expectedColumns.length; i++) {
+            assertEquals(
+                "Column at position " + i + " for query: " + ppl,
+                expectedColumns[i],
+                columns.get(i)
+            );
+        }
+    }
+
+    private Map<String, Object> executePpl(String ppl) throws IOException {
+        ensureDataProvisioned();
+        Request request = new Request("POST", "/_analytics/ppl");
+        request.setJsonEntity("{\"query\": \"" + escapeJson(ppl) + "\"}");
+        Response response = client().performRequest(request);
+        return assertOkAndParse(response, "PPL: " + ppl);
+    }
+}