diff --git a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/FieldType.java b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/FieldType.java
index 2a6a68a076d09..12628e577a900 100644
--- a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/FieldType.java
+++ b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/FieldType.java
@@ -64,7 +64,20 @@ public enum FieldType {
      * placeholder; {@link #fromMappingType} keeps working unchanged because no source
      * advertises that mapping string.
      */
-    ARRAY("array");
+    ARRAY("array"),
+
+    /**
+     * Map-typed expression result. First in-tree producer is PPL `spath`'s auto-extract mode
+     * (`JSON_EXTRACT_ALL` returns {@code MAP<VARCHAR, VARCHAR>}). Mapping string is {@code
+     * "map"} as a placeholder — no OpenSearch storage format declares this mapping today, so
+     * {@link #fromMappingType} never resolves to it through the mapping path; columns reach
+     * MAP only through {@link #fromSqlTypeName}. Capability registrations for filter / project
+     * operators on MAP columns are intentionally minimal: callers (e.g. PPL `where doc.user.name`)
+     * always wrap the MAP column in an ITEM lookup whose result type is the map's value type,
+     * so the EQUALS / sort / aggregate operators see the value-level type by the time the
+     * runtime executes them.
+     */
+    MAP("map");
 
     private final String mappingType;
 
@@ -127,6 +140,7 @@ public static FieldType fromSqlTypeName(SqlTypeName sqlTypeName) {
             case BOOLEAN -> FieldType.BOOLEAN;
             case BINARY, VARBINARY -> FieldType.BINARY;
             case ARRAY -> FieldType.ARRAY;
+            case MAP -> FieldType.MAP;
             default -> null;
         };
     }
diff --git a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java
index bf08f9e3a3982..15ab21cd80618 100644
--- a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java
+++ b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java
@@ -233,6 +233,7 @@ public enum ScalarFunction {
     JSON_DELETE(Category.SCALAR, SqlKind.OTHER_FUNCTION),
     JSON_EXTEND(Category.SCALAR, SqlKind.OTHER_FUNCTION),
     JSON_EXTRACT(Category.SCALAR, SqlKind.OTHER_FUNCTION),
+    JSON_EXTRACT_ALL(Category.SCALAR, SqlKind.OTHER_FUNCTION),
     JSON_KEYS(Category.SCALAR, SqlKind.OTHER_FUNCTION),
     JSON_SET(Category.SCALAR, SqlKind.OTHER_FUNCTION),
 
diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/json_extract_all.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/json_extract_all.rs
new file mode 100644
index 0000000000000..b87835ee6edda
--- /dev/null
+++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/json_extract_all.rs
@@ -0,0 +1,543 @@
+/*
+ * 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.
+ */
+
+//! `json_extract_all(value)` — flatten a JSON object to a `Map<Utf8, Utf8>` with
+//! dot-separated paths as keys (parity with legacy
+//! `JsonExtractAllFunctionImpl`). Objects flatten to dotted keys (`user.name`);
+//! arrays append a `{}` segment (`tags{}`); duplicate logical keys merge into a
+//! Java-`List.toString()` rendering (`[a, b, c]`); JSON nulls render as the
+//! literal string `"null"`. Null input / empty / whitespace / top-level scalar
+//! / top-level number → NULL map. Malformed JSON → empty map (we silently keep
+//! whatever was parsed before the syntax error, matching the legacy
+//! `try { … } catch (IOException) { /* swallow */ }` flow at
+//! {@code JsonExtractAllFunctionImpl#parseJson}).
+
+use std::any::Any;
+use std::sync::Arc;
+
+use datafusion::arrow::array::{ArrayRef, MapBuilder, MapFieldNames, StringBuilder};
+use datafusion::arrow::datatypes::{DataType, Field, Fields};
+use datafusion::common::ScalarValue;
+use datafusion::error::Result;
+use datafusion::execution::context::SessionContext;
+use datafusion::logical_expr::{
+    ColumnarValue, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Volatility,
+};
+use serde_json::Value;
+
+use super::json_common::{check_arity, parse, StringArrayView};
+use super::{coerce_args, CoerceMode};
+
+/// Order-preserving append-or-merge map keyed by stringified path. We use a
+/// `Vec<(String, Slot)>` rather than pulling `indexmap` into the dep tree —
+/// duplicate-key insertion is rare (only when a JSON document has a literal
+/// dotted key that collides with a nested-object path), so O(n) lookup on
+/// merge is dominated by the JSON-walk cost on every realistic input.
+type OrderedEntries = Vec<(String, Slot)>;
+
+const NAME: &str = "json_extract_all";
+const ARRAY_MARKER: &str = "{}";
+
+pub fn register_all(ctx: &SessionContext) {
+    ctx.register_udf(ScalarUDF::from(JsonExtractAllUdf::new()));
+}
+
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct JsonExtractAllUdf {
+    signature: Signature,
+}
+
+impl JsonExtractAllUdf {
+    pub fn new() -> Self {
+        Self {
+            signature: Signature::user_defined(Volatility::Immutable),
+        }
+    }
+}
+
+impl Default for JsonExtractAllUdf {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Arrow `Map<Utf8, Utf8>` (Map → Struct{keys: Utf8 non-null, values: Utf8
+/// nullable}). The legacy Calcite path returns `MAP<VARCHAR, VARCHAR NULLABLE>`
+/// (see `JsonExtractAllFunctionImpl#getReturnTypeInference`); this is the
+/// substrait-friendly Arrow shape that decodes to the same logical type.
+/// `values` is `nullable=true` so JSON nulls can round-trip as Arrow nulls
+/// inside the map (they render to the literal string `"null"` here, but a
+/// future caller might want to distinguish; keeping the slot nullable costs
+/// nothing).
+fn map_data_type() -> DataType {
+    DataType::Map(
+        Arc::new(Field::new(
+            "entries",
+            DataType::Struct(Fields::from(vec![
+                // Field names match Arrow Java's `MapVector.KEY_NAME` /
+                // `VALUE_NAME` so the SQL-plugin response marshaller can use
+                // those constants when unpacking entries.
+                Field::new("key", DataType::Utf8, false),
+                Field::new("value", DataType::Utf8, true),
+            ])),
+            false,
+        )),
+        false,
+    )
+}
+
+impl ScalarUDFImpl for JsonExtractAllUdf {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+    fn name(&self) -> &str {
+        NAME
+    }
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+    fn return_type(&self, _args: &[DataType]) -> Result<DataType> {
+        Ok(map_data_type())
+    }
+    fn coerce_types(&self, args: &[DataType]) -> Result<Vec<DataType>> {
+        coerce_args(NAME, args, &[CoerceMode::Utf8])
+    }
+
+    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+        check_arity(NAME, args.args.len(), 1)?;
+        let n = args.number_rows;
+
+        // Scalar fast-path — every row in the batch evaluates to the same map.
+        if let ColumnarValue::Scalar(sv) = &args.args[0] {
+            let s = scalar_str(sv);
+            // Build a single-row Map array, then wrap as a Scalar(List(...))
+            // by returning Array. DataFusion broadcasts a 1-row Array vs.
+            // promoting to a real Scalar for Map; the cleanest fast-path here
+            // is still to drop into the columnar builder for n rows and emit
+            // identical rows. Build once, replay n times.
+            let result = match s {
+                Some(text) => extract_all(text),
+                None => None,
+            };
+            let mut b = new_map_builder(n);
+            for _ in 0..n {
+                append_row(&mut b, result.as_ref())?;
+            }
+            return Ok(ColumnarValue::Array(Arc::new(b.finish()) as ArrayRef));
+        }
+
+        let arr = args.args[0].clone().into_array(n)?;
+        let strings = StringArrayView::from_array(&arr)?;
+        let mut b = new_map_builder(n);
+        for i in 0..n {
+            let row_result = match strings.cell(i) {
+                Some(s) => extract_all(s),
+                None => None,
+            };
+            append_row(&mut b, row_result.as_ref())?;
+        }
+        Ok(ColumnarValue::Array(Arc::new(b.finish()) as ArrayRef))
+    }
+}
+
+fn scalar_str(sv: &ScalarValue) -> Option<&str> {
+    match sv {
+        ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) | ScalarValue::Utf8View(s) => s.as_deref(),
+        _ => None,
+    }
+}
+
+fn new_map_builder(capacity: usize) -> MapBuilder<StringBuilder, StringBuilder> {
+    MapBuilder::with_capacity(
+        Some(MapFieldNames {
+            entry: "entries".to_string(),
+            key: "key".to_string(),
+            value: "value".to_string(),
+        }),
+        StringBuilder::new(),
+        StringBuilder::new(),
+        capacity,
+    )
+}
+
+/// Append one Map row: `None` → null map slot; `Some(entries)` → a map with
+/// the (already-stringified) entries. Insertion order is preserved by
+/// `IndexMap` so the wire-level JSON object key order matches Jackson's
+/// streaming-parser order in the legacy impl.
+fn append_row(
+    b: &mut MapBuilder<StringBuilder, StringBuilder>,
+    entries: Option<&OrderedEntries>,
+) -> Result<()> {
+    match entries {
+        None => b.append(false)?,
+        Some(map) => {
+            for (k, v) in map.iter() {
+                b.keys().append_value(k);
+                match v.render() {
+                    Some(rendered) => b.values().append_value(&rendered),
+                    None => b.values().append_null(),
+                }
+            }
+            b.append(true)?;
+        }
+    }
+    Ok(())
+}
+
+/// One logical value at a path. Mirrors the legacy `Object` slot —
+/// scalar-or-list polymorphism — so duplicate-key insertion can promote to a
+/// list without rewriting the whole value.
+#[derive(Debug)]
+enum Slot {
+    /// A single value. `None` here means a JSON null was inserted; it renders
+    /// as the literal string `"null"` to match Java's `String.valueOf(null)`.
+    Single(Option<String>),
+    /// Two or more values at the same path. Renders as
+    /// `[v1, v2, ..., vn]` with `null` for null elements — Java's
+    /// `List.toString()` shape.
+    List(Vec<Option<String>>),
+}
+
+impl Slot {
+    fn extend(&mut self, next: Option<String>) {
+        match self {
+            Slot::Single(first) => {
+                let mut v = Vec::with_capacity(2);
+                v.push(first.take());
+                v.push(next);
+                *self = Slot::List(v);
+            }
+            Slot::List(v) => v.push(next),
+        }
+    }
+
+    fn render(&self) -> Option<String> {
+        match self {
+            Slot::Single(None) => Some("null".to_string()),
+            Slot::Single(Some(s)) => Some(s.clone()),
+            Slot::List(items) => {
+                let parts: Vec<&str> = items
+                    .iter()
+                    .map(|o| o.as_deref().unwrap_or("null"))
+                    .collect();
+                Some(format!("[{}]", parts.join(", ")))
+            }
+        }
+    }
+}
+
+/// Public entry — returns `Some(map)` when the input parses (possibly to an
+/// empty map) and is rooted at a JSON object/array; `None` for null/empty/
+/// whitespace input and for a top-level non-container value (string, number,
+/// bool, null). Matches the legacy contract exactly.
+fn extract_all(s: &str) -> Option<OrderedEntries> {
+    if s.trim().is_empty() {
+        return None;
+    }
+    let parsed = match parse(s) {
+        // Malformed → legacy swallows the error and returns the partial map.
+        // With serde_json we get all-or-nothing (no token-level partial
+        // state), but the IT-side `testSpathAutoExtractMalformedJson` only
+        // cares that the result is the empty map (struct `{}`), which is the
+        // observable behaviour we deliver here. Unit-test-level partial-parse
+        // cases (`testInvalidJsonReturnResults`) are not part of the IT
+        // surface.
+        Some(v) => v,
+        None => return Some(Vec::new()),
+    };
+    match &parsed {
+        Value::Object(_) | Value::Array(_) => {}
+        _ => return None, // top-level scalar — legacy returns NULL.
+    }
+    let mut out: OrderedEntries = Vec::new();
+    walk(&parsed, &mut Vec::new(), &mut out);
+    Some(out)
+}
+
+/// DFS walk that mirrors the Jackson stream-walk in the legacy impl.
+/// `path` is the current path stack (field names + array markers). Empty
+/// arrays produce no entries; non-empty arrays append `{}` to the path and
+/// recurse element-by-element so every element shares the same parent path
+/// (which is what produces the duplicate-key merge behaviour).
+fn walk(v: &Value, path: &mut Vec<String>, out: &mut OrderedEntries) {
+    match v {
+        Value::Object(map) => {
+            for (key, child) in map.iter() {
+                path.push(key.clone());
+                walk(child, path, out);
+                path.pop();
+            }
+        }
+        Value::Array(arr) => {
+            path.push(ARRAY_MARKER.to_string());
+            for element in arr.iter() {
+                walk(element, path, out);
+            }
+            path.pop();
+        }
+        leaf => {
+            let key = build_path(path);
+            let value = stringify_scalar(leaf);
+            insert(out, key, value);
+        }
+    }
+}
+
+fn insert(out: &mut OrderedEntries, key: String, value: Option<String>) {
+    if let Some((_, existing)) = out.iter_mut().find(|(k, _)| *k == key) {
+        existing.extend(value);
+    } else {
+        out.push((key, Slot::Single(value)));
+    }
+}
+
+/// Join `path` into a dotted/`{}`-terminated string. Array markers `{}` carry
+/// through unchanged; field-name separators are `.`. Adjacent `{}` segments
+/// land next to each other (`a{}{}` for arrays-of-arrays).
+fn build_path(path: &[String]) -> String {
+    let mut s = String::with_capacity(path.iter().map(|p| p.len() + 1).sum::<usize>());
+    for segment in path {
+        if segment == ARRAY_MARKER {
+            s.push_str(ARRAY_MARKER);
+        } else {
+            if !s.is_empty() && !s.ends_with(ARRAY_MARKER) {
+                s.push('.');
+            } else if s.ends_with(ARRAY_MARKER) {
+                s.push('.');
+            }
+            s.push_str(segment);
+        }
+    }
+    s
+}
+
+/// Stringify a JSON scalar exactly the way the legacy `String.valueOf` chain
+/// does. Returns `None` for JSON null (rendered as `"null"` at the outer
+/// boundary — see `Slot::render`).
+fn stringify_scalar(v: &Value) -> Option<String> {
+    match v {
+        Value::Null => None,
+        Value::Bool(b) => Some(b.to_string()),
+        Value::Number(n) => Some(stringify_number(n)),
+        Value::String(s) => Some(s.clone()),
+        Value::Array(_) | Value::Object(_) => {
+            // Caller shouldn't reach this — containers are walked recursively
+            // by `walk`. Defensive: serialize as JSON so a future caller that
+            // passes a container by mistake doesn't silently lose data.
+            Some(v.to_string())
+        }
+    }
+}
+
+/// JSON-number → Java-`String.valueOf` parity:
+///  - Integers in i64 range → `123`, `9223372036854775807`
+///  - Integers above i64 → promoted to f64 and rendered in Rust's shortest
+///    round-trip form (matches `Double.toString` for the common
+///    `9.223372036854776E18` pattern; minor lowercase-`e` divergence is the
+///    only known formatting difference, not exercised by spath ITs).
+///  - Non-integers → Rust's default `{}` formatter, which is the shortest
+///    round-trip representation (matches Java's `Double.toString` for finite
+///    values that don't need exponent form: `3.14159`, `0.5`).
+fn stringify_number(n: &serde_json::Number) -> String {
+    if let Some(i) = n.as_i64() {
+        return i.to_string();
+    }
+    if let Some(u) = n.as_u64() {
+        return u.to_string();
+    }
+    if let Some(f) = n.as_f64() {
+        return f.to_string();
+    }
+    // Should be unreachable for finite JSON; fall back to the raw textual
+    // form serde_json holds internally.
+    n.to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn rendered(s: &str) -> Option<Vec<(String, String)>> {
+        extract_all(s).map(|entries| {
+            entries
+                .into_iter()
+                .map(|(k, v)| (k, v.render().unwrap_or_else(|| "null".to_string())))
+                .collect()
+        })
+    }
+
+    #[test]
+    fn null_and_empty_input_returns_none() {
+        assert!(rendered("").is_none());
+        assert!(rendered("   ").is_none());
+    }
+
+    #[test]
+    fn top_level_scalar_returns_none() {
+        // Legacy: `if (pathStack.isEmpty()) return null` on top-level scalar.
+        assert!(rendered(r#""just a string""#).is_none());
+        assert!(rendered("42").is_none());
+        assert!(rendered("null").is_none());
+        assert!(rendered("true").is_none());
+    }
+
+    #[test]
+    fn empty_object_returns_empty_map() {
+        assert_eq!(rendered(r#"{}"#).unwrap(), Vec::<(String, String)>::new());
+    }
+
+    #[test]
+    fn simple_object_dotted_keys_for_scalars() {
+        assert_eq!(
+            rendered(r#"{"name":"John","age":30}"#).unwrap(),
+            vec![
+                ("name".to_string(), "John".to_string()),
+                ("age".to_string(), "30".to_string()),
+            ]
+        );
+    }
+
+    #[test]
+    fn nested_object_flattens_to_dotted_path() {
+        assert_eq!(
+            rendered(r#"{"user":{"name":"John"},"system":"linux"}"#).unwrap(),
+            vec![
+                ("user.name".to_string(), "John".to_string()),
+                ("system".to_string(), "linux".to_string()),
+            ]
+        );
+    }
+
+    #[test]
+    fn array_of_primitives_suffixes_with_curly_marker() {
+        assert_eq!(
+            rendered(r#"{"tags":["a","b","c"]}"#).unwrap(),
+            vec![("tags{}".to_string(), "[a, b, c]".to_string())]
+        );
+    }
+
+    #[test]
+    fn empty_array_yields_no_entry() {
+        assert_eq!(
+            rendered(r#"{"empty":[]}"#).unwrap(),
+            Vec::<(String, String)>::new()
+        );
+    }
+
+    #[test]
+    fn array_of_objects_propagates_marker_into_children() {
+        assert_eq!(
+            rendered(r#"{"users":[{"name":"John"},{"name":"Jane"}]}"#).unwrap(),
+            vec![("users{}.name".to_string(), "[John, Jane]".to_string())]
+        );
+    }
+
+    #[test]
+    fn top_level_array_of_objects_roots_at_curly_marker() {
+        assert_eq!(
+            rendered(r#"[{"age":1},{"age":2}]"#).unwrap(),
+            vec![("{}.age".to_string(), "[1, 2]".to_string())]
+        );
+    }
+
+    #[test]
+    fn top_level_array_of_primitives_uses_curly_marker_as_root() {
+        assert_eq!(
+            rendered(r#"[1,2,3]"#).unwrap(),
+            vec![("{}".to_string(), "[1, 2, 3]".to_string())]
+        );
+    }
+
+    #[test]
+    fn json_null_renders_as_literal_string_null() {
+        // `{"x": null}` → key `x` present, value rendered as the literal
+        // string `"null"` (legacy `String.valueOf((Object) null) == "null"`).
+        let m = rendered(r#"{"n":30,"b":true,"x":null}"#).unwrap();
+        assert_eq!(
+            m,
+            vec![
+                ("n".to_string(), "30".to_string()),
+                ("b".to_string(), "true".to_string()),
+                ("x".to_string(), "null".to_string()),
+            ]
+        );
+    }
+
+    #[test]
+    fn null_in_array_renders_inline_with_other_elements() {
+        assert_eq!(
+            rendered(r#"[{"a":null},{"a":1}]"#).unwrap(),
+            vec![("{}.a".to_string(), "[null, 1]".to_string())]
+        );
+    }
+
+    #[test]
+    fn duplicate_logical_keys_merge_into_bracketed_list() {
+        // `{"a":{"b":1}, "a.b": 2}` — Jackson's parser emits the outer key
+        // `a.b` after the nested-object path produces the same key, so the
+        // values merge into a list. With `preserve_order` enabled,
+        // serde_json's `Value::Object` preserves the input order across
+        // duplicate logical keys, so the merge produces `[1, 2]`.
+        assert_eq!(
+            rendered(r#"{"a":{"b":1},"a.b":2}"#).unwrap(),
+            vec![("a.b".to_string(), "[1, 2]".to_string())]
+        );
+    }
+
+    #[test]
+    fn malformed_json_returns_empty_map() {
+        // Legacy: catches Jackson IOException and returns whatever was parsed.
+        // serde_json is all-or-nothing, so we return the empty map — the
+        // IT-level `testSpathAutoExtractMalformedJson` only asserts on `{}`.
+        assert_eq!(
+            rendered(r#"{"user":{"name":"#).unwrap(),
+            Vec::<(String, String)>::new()
+        );
+    }
+
+    #[test]
+    fn return_type_is_map_utf8_utf8() {
+        let dt = JsonExtractAllUdf::new().return_type(&[DataType::Utf8]).unwrap();
+        match dt {
+            DataType::Map(field, sorted) => {
+                assert!(!sorted);
+                match field.data_type() {
+                    DataType::Struct(fields) => {
+                        assert_eq!(fields.len(), 2);
+                        assert_eq!(fields[0].name(), "key");
+                        assert_eq!(fields[0].data_type(), &DataType::Utf8);
+                        assert!(!fields[0].is_nullable());
+                        assert_eq!(fields[1].name(), "value");
+                        assert_eq!(fields[1].data_type(), &DataType::Utf8);
+                        assert!(fields[1].is_nullable());
+                    }
+                    other => panic!("expected Struct inside Map, got {other:?}"),
+                }
+            }
+            other => panic!("expected Map, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn coerce_types_enforces_string_arity() {
+        let udf = JsonExtractAllUdf::new();
+        for variant in [DataType::Utf8, DataType::LargeUtf8, DataType::Utf8View] {
+            assert_eq!(
+                udf.coerce_types(&[variant.clone()]).unwrap(),
+                vec![variant.clone()],
+                "CoerceMode::Utf8 should pass {variant:?} through unchanged"
+            );
+        }
+        assert!(udf
+            .coerce_types(&[DataType::Int64])
+            .unwrap_err()
+            .to_string()
+            .contains("expected string"));
+        assert!(udf.coerce_types(&[]).is_err());
+    }
+}
diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs
index 677aed08fcd80..930b50c881643 100644
--- a/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs
+++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs
@@ -132,6 +132,7 @@ pub(crate) mod json_common;
 pub mod json_delete;
 pub mod json_extend;
 pub mod json_extract;
+pub mod json_extract_all;
 pub mod json_keys;
 pub mod json_set;
 pub mod makedate;
@@ -171,6 +172,7 @@ pub fn register_all(ctx: &SessionContext) {
     json_delete::register_all(ctx);
     json_extend::register_all(ctx);
     json_extract::register_all(ctx);
+    json_extract_all::register_all(ctx);
     json_keys::register_all(ctx);
     json_set::register_all(ctx);
     makedate::register_all(ctx);
@@ -190,7 +192,7 @@ pub fn register_all(ctx: &SessionContext) {
     time_format::register_all(ctx);
     width_bucket::register_all(ctx);
     log::info!(
-        "OpenSearch UDF register_all: convert_tz, conversion(numeric_conversion: num/auto/memk/rmcomma/rmunit/dur2sec/mstime, time_conversion: ctime/mktime), crc32, date_format, extract, from_unixtime, json_append, json_array_length, json_delete, json_extend, json_extract, json_keys, json_set, makedate, maketime, minspan_bucket, mvappend, mvfind, mvzip, range_bucket, rex_extract, rex_extract_multi, rex_offset, sha1, span_bucket, str_to_date, strftime, time_format, tonumber, tostring, width_bucket registered"
+        "OpenSearch UDF register_all: convert_tz, conversion(numeric_conversion: num/auto/memk/rmcomma/rmunit/dur2sec/mstime, time_conversion: ctime/mktime), crc32, date_format, extract, from_unixtime, json_append, json_array_length, json_delete, json_extend, json_extract, json_extract_all, json_keys, json_set, makedate, maketime, minspan_bucket, mvappend, mvfind, mvzip, range_bucket, rex_extract, rex_extract_multi, rex_offset, sha1, span_bucket, str_to_date, strftime, time_format, width_bucket registered"
     );
 }
 
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/ArrayElementAdapter.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/ArrayElementAdapter.java
index 02476ad222a4e..4fb62dfe34c60 100644
--- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/ArrayElementAdapter.java
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/ArrayElementAdapter.java
@@ -51,10 +51,10 @@
 class ArrayElementAdapter implements ScalarFunctionAdapter {
 
     /**
-     * Locally-declared target operator. Name matches DataFusion's native
-     * {@code array_element}. Return-type inference is a placeholder — the
-     * adapt method explicitly carries the original ITEM call's return type
-     * (the element type).
+     * Locally-declared target operator for the array-input case. Name matches
+     * DataFusion's native {@code array_element}. Return-type inference is a
+     * placeholder — the adapt method explicitly carries the original ITEM
+     * call's return type (the element type).
      */
     static final SqlOperator LOCAL_ARRAY_ELEMENT_OP = new SqlFunction(
         "array_element",
@@ -65,6 +65,23 @@ class ArrayElementAdapter implements ScalarFunctionAdapter {
         SqlFunctionCategory.SYSTEM
     );
 
+    /**
+     * Locally-declared target operator for the map-input case. PPL `spath`'s
+     * auto-extract mode lowers `result.user.name` to
+     * {@code ITEM(JSON_EXTRACT_ALL($n), 'user.name')} where the container is a
+     * {@code MAP<VARCHAR, VARCHAR>}. DataFusion's array_element only handles
+     * List/Array; map keys must dispatch to {@code map_extract}. Same
+     * placeholder return-type approach as {@link #LOCAL_ARRAY_ELEMENT_OP}.
+     */
+    static final SqlOperator LOCAL_MAP_EXTRACT_OP = new SqlFunction(
+        "map_extract",
+        SqlKind.OTHER_FUNCTION,
+        ReturnTypes.ARG0,
+        null,
+        OperandTypes.ANY_ANY,
+        SqlFunctionCategory.SYSTEM
+    );
+
     @Override
     public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelOptCluster cluster) {
         RexBuilder rexBuilder = cluster.getRexBuilder();
@@ -73,13 +90,39 @@ public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelO
         if (operands.size() != 2) {
             return rexBuilder.makeCall(original.getType(), LOCAL_ARRAY_ELEMENT_OP, operands);
         }
-        RexNode array = operands.get(0);
-        RexNode index = operands.get(1);
-        if (index.getType().getSqlTypeName() != SqlTypeName.BIGINT) {
+        RexNode container = operands.get(0);
+        RexNode key = operands.get(1);
+        // Map dispatch — PPL spath's JSON_EXTRACT_ALL returns MAP, so
+        // `result.user.name` lowers to `ITEM(map, 'user.name')` with a string
+        // key. DataFusion's `map_extract(map, key)` returns `List<value>`
+        // because maps in some semantics permit duplicate keys; wrap the call
+        // in `array_element(..., 1)` to project the first (and, for our
+        // unique-key maps, only) element back to a scalar that matches the
+        // PPL ITEM call's declared return type (the map's value type).
+        if (container.getType().getSqlTypeName() == SqlTypeName.MAP) {
+            RelDataType keyType = container.getType().getKeyType();
+            RelDataType valueType = container.getType().getValueType();
+            // Coerce the lookup key to the map's declared key type. Calcite
+            // produces CHAR(N) literals for `result.user.name`-style accesses,
+            // which doesn't unify with the map's VARCHAR key type variable
+            // (`any1`) inside Substrait's extension catalog. An explicit cast
+            // collapses CHAR(N) → VARCHAR before isthmus emits the call.
+            if (!key.getType().equals(keyType)) {
+                RelDataType nullableKeyType = typeFactory.createTypeWithNullability(keyType, key.getType().isNullable());
+                key = rexBuilder.makeCast(nullableKeyType, key, true, false);
+            }
+            RelDataType nullableValueType = typeFactory.createTypeWithNullability(valueType, true);
+            RelDataType listOfValue = typeFactory.createArrayType(nullableValueType, -1);
+            RexNode mapResult = rexBuilder.makeCall(listOfValue, LOCAL_MAP_EXTRACT_OP, List.of(container, key));
+            RexNode one = rexBuilder.makeLiteral(java.math.BigDecimal.ONE, typeFactory.createSqlType(SqlTypeName.BIGINT), false);
+            return rexBuilder.makeCall(original.getType(), LOCAL_ARRAY_ELEMENT_OP, List.of(mapResult, one));
+        }
+        // Array dispatch — coerce the index to BIGINT as before.
+        if (key.getType().getSqlTypeName() != SqlTypeName.BIGINT) {
             RelDataType bigint = typeFactory.createSqlType(SqlTypeName.BIGINT);
-            RelDataType nullableBigint = typeFactory.createTypeWithNullability(bigint, index.getType().isNullable());
-            index = rexBuilder.makeCast(nullableBigint, index, true, false);
+            RelDataType nullableBigint = typeFactory.createTypeWithNullability(bigint, key.getType().isNullable());
+            key = rexBuilder.makeCast(nullableBigint, key, true, false);
         }
-        return rexBuilder.makeCall(original.getType(), LOCAL_ARRAY_ELEMENT_OP, List.of(array, index));
+        return rexBuilder.makeCall(original.getType(), LOCAL_ARRAY_ELEMENT_OP, List.of(container, key));
     }
 }
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java
index d677195e8c2aa..d18c010f845ee 100644
--- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java
@@ -285,6 +285,9 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP
         ScalarFunction.JSON_DELETE,
         ScalarFunction.JSON_EXTEND,
         ScalarFunction.JSON_EXTRACT,
+        // JSON_EXTRACT_ALL — return type is MAP<VARCHAR, VARCHAR>, so its capability
+        // is registered separately via {@link #MAP_RETURNING_PROJECT_OPS} (keyed on
+        // FieldType.MAP rather than SUPPORTED_FIELD_TYPES, mirroring the ARRAY-return split).
         ScalarFunction.JSON_KEYS,
         ScalarFunction.JSON_SET,
         // Array functions whose RETURN type is element-typed (not ARRAY itself), so the
@@ -352,6 +355,22 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP
         ScalarFunction.MVAPPEND
     );
 
+    /**
+     * Project-side scalar functions whose return type is {@code MAP<VARCHAR, VARCHAR>}.
+     * Registered separately because the capability lookup at
+     * {@code OpenSearchProjectRule.resolveScalarViableBackends} keys on the call's return
+     * type, and for these the lookup resolves to {@link FieldType#MAP} — intentionally not
+     * in {@link #SUPPORTED_FIELD_TYPES} (filter / aggregate / sort operators have no
+     * meaningful semantics over map-typed values; the value-level type emerges after the
+     * {@code ITEM(map, key)} lookup that always follows a map-returning call).
+     *
+     * <p>{@code JSON_EXTRACT_ALL} flattens a JSON object to dot-path keys, returning a
+     * {@code MAP<VARCHAR, VARCHAR>}. First in-tree caller is PPL {@code spath}'s
+     * auto-extract mode; routes to the {@code json_extract_all} Rust UDF via
+     * {@link JsonFunctionAdapters.JsonExtractAllAdapter}.
+     */
+    private static final Set<ScalarFunction> MAP_RETURNING_PROJECT_OPS = Set.of(ScalarFunction.JSON_EXTRACT_ALL);
+
     private static final Set<AggregateFunction> AGG_FUNCTIONS = Set.of(
         AggregateFunction.SUM,
         AggregateFunction.SUM0,
@@ -440,6 +459,16 @@ public Set<FilterCapability> filterCapabilities() {
                     for (FieldType type : SUPPORTED_FIELD_TYPES) {
                         caps.add(new FilterCapability.Standard(op, Set.of(type), formats));
                     }
+                    // MAP-typed fields enter the filter rule when the predicate is
+                    // shape `ITEM(map_field, key) <op> literal` (PPL `where doc.user.name = 'John'`
+                    // after the spath auto-extract lowering). The filter-rule's field-index
+                    // collection sees the underlying MAP column, not the ITEM-extracted scalar,
+                    // so without this branch the WHERE rejects with
+                    // "No backend can evaluate filter predicate [...] on fields [<col>:MAP]".
+                    // Registering STANDARD_FILTER_OPS on MAP is sound because every viable
+                    // predicate against a MAP column is forced through an ITEM lookup that
+                    // emits a value-typed scalar before substrait emission.
+                    caps.add(new FilterCapability.Standard(op, Set.of(FieldType.MAP), formats));
                 }
                 return Set.copyOf(caps);
             }
@@ -461,6 +490,9 @@ public Set<ProjectCapability> projectCapabilities() {
                 for (ScalarFunction op : ARRAY_RETURNING_PROJECT_OPS) {
                     caps.add(new ProjectCapability.Scalar(op, Set.of(FieldType.ARRAY), formats, true));
                 }
+                for (ScalarFunction op : MAP_RETURNING_PROJECT_OPS) {
+                    caps.add(new ProjectCapability.Scalar(op, Set.of(FieldType.MAP), formats, true));
+                }
                 return Set.copyOf(caps);
             }
 
@@ -542,6 +574,7 @@ public Map<ScalarFunction, ScalarFunctionAdapter> scalarFunctionAdapters() {
                     Map.entry(ScalarFunction.JSON_DELETE, new JsonFunctionAdapters.JsonDeleteAdapter()),
                     Map.entry(ScalarFunction.JSON_EXTEND, new JsonFunctionAdapters.JsonExtendAdapter()),
                     Map.entry(ScalarFunction.JSON_EXTRACT, new JsonFunctionAdapters.JsonExtractAdapter()),
+                    Map.entry(ScalarFunction.JSON_EXTRACT_ALL, new JsonFunctionAdapters.JsonExtractAllAdapter()),
                     Map.entry(ScalarFunction.JSON_KEYS, new JsonFunctionAdapters.JsonKeysAdapter()),
                     Map.entry(ScalarFunction.JSON_SET, new JsonFunctionAdapters.JsonSetAdapter()),
                     Map.entry(ScalarFunction.LIKE, new LikeAdapter()),
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java
index 0344dbc4ab781..fd691b57a19df 100644
--- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java
@@ -209,6 +209,7 @@ public class DataFusionFragmentConvertor implements FragmentConvertor {
         FunctionMappings.s(JsonFunctionAdapters.JsonDeleteAdapter.LOCAL_JSON_DELETE_OP, "json_delete"),
         FunctionMappings.s(JsonFunctionAdapters.JsonExtendAdapter.LOCAL_JSON_EXTEND_OP, "json_extend"),
         FunctionMappings.s(JsonFunctionAdapters.JsonExtractAdapter.LOCAL_JSON_EXTRACT_OP, "json_extract"),
+        FunctionMappings.s(JsonFunctionAdapters.JsonExtractAllAdapter.LOCAL_JSON_EXTRACT_ALL_OP, "json_extract_all"),
         FunctionMappings.s(JsonFunctionAdapters.JsonKeysAdapter.LOCAL_JSON_KEYS_OP, "json_keys"),
         FunctionMappings.s(JsonFunctionAdapters.JsonSetAdapter.LOCAL_JSON_SET_OP, "json_set"),
         FunctionMappings.s(SqlLibraryOperators.REGEXP_CONTAINS, "regex_match"),
@@ -227,6 +228,7 @@ public class DataFusionFragmentConvertor implements FragmentConvertor {
         FunctionMappings.s(MakeArrayAdapter.LOCAL_MAKE_ARRAY_OP, "make_array"),
         FunctionMappings.s(ArrayToStringAdapter.LOCAL_ARRAY_TO_STRING_OP, "array_to_string"),
         FunctionMappings.s(ArrayElementAdapter.LOCAL_ARRAY_ELEMENT_OP, "array_element"),
+        FunctionMappings.s(ArrayElementAdapter.LOCAL_MAP_EXTRACT_OP, "map_extract"),
         FunctionMappings.s(MvzipAdapter.LOCAL_MVZIP_OP, "mvzip"),
         FunctionMappings.s(MvfindAdapter.LOCAL_MVFIND_OP, "mvfind"),
         FunctionMappings.s(MvappendAdapter.LOCAL_MVAPPEND_OP, "mvappend"),
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/JsonFunctionAdapters.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/JsonFunctionAdapters.java
index 9a416de26ae8f..354f797e1beca 100644
--- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/JsonFunctionAdapters.java
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/JsonFunctionAdapters.java
@@ -89,6 +89,32 @@ static class JsonExtractAdapter extends AbstractNameMappingAdapter {
         }
     }
 
+    /**
+     * {@code JSON_EXTRACT_ALL(value)} — flatten a JSON document to a
+     * {@code MAP<VARCHAR, VARCHAR>} keyed by dot-separated path; non-object /
+     * top-level scalar / null / empty / whitespace inputs → NULL map; malformed
+     * → empty map. Matches the legacy {@code JsonExtractAllFunctionImpl} on the
+     * v2 / Calcite path. The locally-declared return type here is a placeholder
+     * — {@link AbstractNameMappingAdapter#adapt} preserves the original PPL
+     * call's MAP RelDataType, so isthmus emits a substrait map type and the
+     * DataFusion-side substrait consumer decodes it to Arrow {@code Map<Utf8, Utf8>}.
+     */
+    static class JsonExtractAllAdapter extends AbstractNameMappingAdapter {
+
+        static final SqlOperator LOCAL_JSON_EXTRACT_ALL_OP = new SqlFunction(
+            "json_extract_all",
+            SqlKind.OTHER_FUNCTION,
+            ReturnTypes.VARCHAR_NULLABLE,
+            null,
+            OperandTypes.STRING,
+            SqlFunctionCategory.STRING
+        );
+
+        JsonExtractAllAdapter() {
+            super(LOCAL_JSON_EXTRACT_ALL_OP, List.of(), List.of());
+        }
+    }
+
     /** {@code JSON_DELETE(value, path1, [path2, ...])} — remove PPL-path matches; missing paths are no-ops. */
     static class JsonDeleteAdapter extends AbstractNameMappingAdapter {
 
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml b/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml
index 21b5fe67009db..a0fcb7b624bd4 100644
--- a/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml
@@ -612,6 +612,36 @@ scalar_functions:
         variadic: { min: 1 }
         return: string
 
+  - name: "map_extract"
+    description: >-
+      Return the list of values whose key matches `key` in `value`. DataFusion's
+      native `map_extract` returns a list because the underlying map permits
+      duplicate keys; the analytics-engine ITEM adapter wraps every
+      `map_extract` call in `array_element(..., 1)` to project back to a
+      scalar. Signature uses Substrait type variables (`any1`/`any2`) so a
+      single declaration covers every Calcite MAP variant — first caller is
+      PPL `spath`'s JSON_EXTRACT_ALL map (`map<string, string?>`).
+    impls:
+      - args: [{ value: "map<any1, any2>", name: "value" }, { value: any1, name: "key" }]
+        return: "list<any2>"
+
+  - name: "json_extract_all"
+    description: >-
+      Flatten a JSON document into a Map<string, string?> keyed by
+      dot-separated path. Objects flatten to dotted keys (`user.name`); arrays
+      append `{}` (`tags{}`); duplicate logical keys merge into a Java
+      `List.toString()` rendering (`[a, b, c]`); JSON nulls render as the
+      literal string `"null"`. Top-level scalar / null / empty / whitespace
+      input → NULL map; malformed JSON → empty map. Parity with the legacy
+      v2 / Calcite `JsonExtractAllFunctionImpl`. Return type is declared as
+      `any1` here so substrait's extension parser does not need a map-type
+      grammar — the actual call-site RelDataType (MAP<VARCHAR, VARCHAR>) is
+      what isthmus emits into the substrait IR, and the DataFusion-side UDF
+      registers its own concrete return type (`Map<Utf8, Utf8>`).
+    impls:
+      - args: [{ value: string, name: "value" }]
+        return: any1
+
   - name: "json_delete"
     description: "Remove PPL-path matches from a JSON document; missing paths are no-ops."
     impls:
diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/ArrowValues.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/ArrowValues.java
index 2a944451363cd..d3f752d22b78c 100644
--- a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/ArrowValues.java
+++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/ArrowValues.java
@@ -11,11 +11,14 @@
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.VarCharVector;
 import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.complex.MapVector;
 import org.apache.arrow.vector.util.Text;
 
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Helpers for reading Arrow vector cells as plain Java values at the
@@ -38,6 +41,26 @@ public static Object toJavaValue(FieldVector vector, int index) {
         if (vector instanceof VarCharVector v) {
             return new String(v.get(index), StandardCharsets.UTF_8);
         }
+        // MapVector check must come before the ListVector branch because
+        // MapVector extends ListVector. Arrow Map is laid out as
+        // List<Struct{keys, values}>, so MapVector.getObject(i) returns a
+        // JsonStringArrayList of entry structs rather than a Map. Reassemble
+        // entries into a LinkedHashMap (insertion-order preserving) so the
+        // downstream ExprValueUtils tuple converter sees the same shape as a
+        // legacy v2 Map<String, Object> column. Routes the values through
+        // Text→String normalization so JSON serialization doesn't choke on
+        // Arrow's UTF-8 byte wrapper. First in-tree caller is the spath
+        // command's `json_extract_all` UDF on the analytics-engine route.
+        if (vector instanceof MapVector && vector.getObject(index) instanceof List<?> entries) {
+            LinkedHashMap<String, Object> map = new LinkedHashMap<>();
+            for (Object entry : entries) {
+                if (!(entry instanceof Map<?, ?> e)) continue;
+                Object k = e.get(MapVector.KEY_NAME);
+                Object v = e.get(MapVector.VALUE_NAME);
+                map.put(k instanceof Text t ? t.toString() : String.valueOf(k), v instanceof Text t ? t.toString() : v);
+            }
+            return map;
+        }
         Object value = vector.getObject(index);
         if (vector instanceof ListVector && value instanceof List<?> raw) {
             // ListVector.getObject returns a JsonStringArrayList whose elements are the
diff --git a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/SpathCommandIT.java b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/SpathCommandIT.java
new file mode 100644
index 0000000000000..71b2e458d4519
--- /dev/null
+++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/SpathCommandIT.java
@@ -0,0 +1,406 @@
+/*
+ * 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.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Self-contained integration test for PPL {@code spath} on the analytics-engine route.
+ *
+ * <p>Mirrors {@code CalcitePPLSpathCommandIT} from the {@code opensearch-project/sql}
+ * repository one-test-method-to-one so the analytics-engine path can be verified inside
+ * core without cross-plugin dependencies on the SQL plugin. 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>Exercises both spath modes:
+ * <ul>
+ *   <li>Path mode — {@code spath path=N input=doc output=result} (lowers to
+ *       {@code json_extract(doc, 'N')} on the DataFusion side).</li>
+ *   <li>Auto-extract mode — {@code spath input=doc} (lowers to
+ *       {@code json_extract_all(doc)} which returns a {@code MAP<VARCHAR, VARCHAR>}).
+ *       The MAP column drives the ITEM-on-MAP dispatch in {@code ArrayElementAdapter},
+ *       the {@code map_extract} substrait wiring, and the {@code ArrowValues.MapVector}
+ *       response marshalling — all introduced by the spath analytics-engine PR.</li>
+ * </ul>
+ *
+ * <p>Provisions four small datasets ({@code spath_simple}, {@code spath_auto},
+ * {@code spath_cmd}, {@code spath_null}) once per class via {@link DatasetProvisioner};
+ * {@link AnalyticsRestTestCase#preserveIndicesUponCompletion()} keeps them across
+ * test methods.
+ */
+public class SpathCommandIT extends AnalyticsRestTestCase {
+
+    private static final Dataset SIMPLE = new Dataset("spath_simple", "spath_simple");
+    private static final Dataset AUTO = new Dataset("spath_auto", "spath_auto");
+    private static final Dataset CMD = new Dataset("spath_cmd", "spath_cmd");
+    private static final Dataset NULL_DATASET = new Dataset("spath_null", "spath_null");
+
+    private static boolean dataProvisioned = false;
+
+    /**
+     * Lazily provision all four datasets on first invocation. Must be called inside a
+     * test method (not {@code setUp()}) — {@code OpenSearchRestTestCase}'s static
+     * {@code client()} is not initialized until after {@code @BeforeClass}, but is
+     * reliably available inside test bodies.
+     */
+    private void ensureDataProvisioned() throws IOException {
+        if (dataProvisioned == false) {
+            DatasetProvisioner.provision(client(), SIMPLE);
+            DatasetProvisioner.provision(client(), AUTO);
+            DatasetProvisioner.provision(client(), CMD);
+            DatasetProvisioner.provision(client(), NULL_DATASET);
+            dataProvisioned = true;
+        }
+    }
+
+    // ── path mode ───────────────────────────────────────────────────────────────
+
+    public void testSimpleSpath() throws IOException {
+        assertRows(
+            "source=" + SIMPLE.indexName + " | spath input=doc output=result path=n | fields result",
+            row("1"),
+            row("2"),
+            row("3")
+        );
+    }
+
+    // ── auto-extract mode ───────────────────────────────────────────────────────
+
+    public void testSpathAutoExtract() throws IOException {
+        // schema: doc (struct); values are flattened single-key maps
+        assertMapRows(
+            "source=" + SIMPLE.indexName + " | spath input=doc",
+            mapOf("n", "1"),
+            mapOf("n", "2"),
+            mapOf("n", "3")
+        );
+    }
+
+    public void testSpathAutoExtractWithOutput() throws IOException {
+        // schema: doc (string), result (struct)
+        assertRows(
+            "source=" + SIMPLE.indexName + " | spath input=doc output=result",
+            row("{\"n\": 1}", mapOf("n", "1")),
+            row("{\"n\": 2}", mapOf("n", "2")),
+            row("{\"n\": 3}", mapOf("n", "3"))
+        );
+    }
+
+    public void testSpathAutoExtractNestedFields() throws IOException {
+        // Nested objects flatten to dotted keys: user.name
+        assertMapRows(
+            "source=" + AUTO.indexName + " | spath input=nested_doc output=result | fields result",
+            mapOf("user.name", "John")
+        );
+    }
+
+    public void testSpathAutoExtractArraySuffix() throws IOException {
+        // Arrays use {} suffix; merged values use [a, b, c] format
+        assertMapRows(
+            "source=" + AUTO.indexName + " | spath input=array_doc output=result | fields result",
+            mapOf("tags{}", "[java, sql]")
+        );
+    }
+
+    public void testSpathAutoExtractDuplicateKeysMerge() throws IOException {
+        // Nested-key path and a literal dotted key both map to "a.b" → merged into a list.
+        assertMapRows(
+            "source=" + AUTO.indexName + " | spath input=merge_doc output=result | fields result",
+            mapOf("a.b", "[1, 2]")
+        );
+    }
+
+    public void testSpathAutoExtractStringifyAndNull() throws IOException {
+        // Numbers / booleans / nulls all stringify; null renders as the literal "null".
+        assertMapRows(
+            "source=" + AUTO.indexName + " | spath input=stringify_doc output=result | fields result",
+            mapOf("n", "30", "b", "true", "x", "null")
+        );
+    }
+
+    public void testSpathAutoExtractNullInput() throws IOException {
+        // First row: doc parses to {n:1}. Second row: doc is null → null map.
+        assertMapRows(
+            "source=" + NULL_DATASET.indexName + " | spath input=doc output=result | fields result",
+            mapOf("n", "1"),
+            null
+        );
+    }
+
+    public void testSpathAutoExtractEmptyJson() throws IOException {
+        // Empty JSON object returns an empty map (not null).
+        assertMapRows(
+            "source=" + AUTO.indexName + " | spath input=empty_doc output=result | fields result",
+            mapOf()
+        );
+    }
+
+    public void testSpathAutoExtractMalformedJson() throws IOException {
+        // Malformed JSON returns the partial map (empty here — Jackson errors out before
+        // any (field-name, value) pair lands).
+        assertMapRows(
+            "source=" + AUTO.indexName + " | spath input=malformed_doc output=result | fields result",
+            mapOf()
+        );
+    }
+
+    // ── eval / where / stats / sort with auto-extract MAP column ────────────────
+
+    public void testSpathAutoExtractWithEval() throws IOException {
+        // ITEM(JSON_EXTRACT_ALL(doc), 'user.name') → scalar VARCHAR through the
+        // map_extract + array_element(...,1) wrap in the analytics-engine adapter.
+        // Bulk-load order: doc 1 (John), doc 2 (Alice).
+        assertRows(
+            "source=" + CMD.indexName + " | spath input=doc | eval name = doc.user.name | fields name",
+            row("John"),
+            row("Alice")
+        );
+    }
+
+    public void testSpathAutoExtractWithWhere() throws IOException {
+        // EQUALS on a MAP-column expression goes through the FieldType.MAP filter
+        // capability registered for the analytics-engine route.
+        assertRows(
+            "source=" + CMD.indexName + " | spath input=doc | where doc.user.name = 'John' | fields doc.user.name",
+            row("John")
+        );
+    }
+
+    public void testSpathAutoExtractWithStats() throws IOException {
+        // SUM over an ITEM-extracted MAP value, GROUP BY another ITEM-extracted value.
+        // GROUP BY result is not order-stable on the analytics-engine route; assert via
+        // set-equality.
+        assertRowsAnyOrder(
+            "source=" + CMD.indexName + " | spath input=doc | stats sum(doc.user.age) by doc.user.name",
+            row(25, "Alice"),
+            row(30, "John")
+        );
+    }
+
+    public void testSpathAutoExtractWithSort() throws IOException {
+        assertRowsInOrder(
+            "source=" + CMD.indexName + " | spath input=doc | sort doc.user.name | fields doc.user.name",
+            row("Alice"),
+            row("John")
+        );
+    }
+
+    public void testSpathAutoExtractWithMultiFieldEval() throws IOException {
+        // Two dotted-path assignments in a single eval (issue #5185 regression).
+        // Bulk-load order: doc 1 (John, 30), doc 2 (Alice, 25).
+        assertRows(
+            "source=" + CMD.indexName + " | spath input=doc"
+                + " | eval doc.user.name=doc.user.name, doc.user.age=doc.user.age"
+                + " | fields doc.user.name, doc.user.age",
+            row("John", "30"),
+            row("Alice", "25")
+        );
+    }
+
+    public void testSpathAutoExtractWithSeparateEvalCommands() throws IOException {
+        // Same two assignments in separate eval commands.
+        assertRows(
+            "source=" + CMD.indexName + " | spath input=doc"
+                + " | eval doc.user.name=doc.user.name"
+                + " | eval doc.user.age=doc.user.age"
+                + " | fields doc.user.name, doc.user.age",
+            row("John", "30"),
+            row("Alice", "25")
+        );
+    }
+
+    // ── helpers ─────────────────────────────────────────────────────────────────
+
+    /** Build an expected row from positional values. */
+    private static List<Object> row(Object... values) {
+        return Arrays.asList(values);
+    }
+
+    /**
+     * Build an expected MAP cell. The auto-extract output deserializes to a Java
+     * {@code Map<String, Object>} on the wire, so we compare against a built-from-pairs
+     * map. Insertion order is not significant — {@link #assertCellEquals} compares maps
+     * by entry-set equality.
+     */
+    private static Map<String, Object> mapOf(Object... keyValuePairs) {
+        if (keyValuePairs.length % 2 != 0) {
+            throw new IllegalArgumentException("mapOf requires an even number of arguments");
+        }
+        java.util.LinkedHashMap<String, Object> m = new java.util.LinkedHashMap<>();
+        for (int i = 0; i < keyValuePairs.length; i += 2) {
+            m.put((String) keyValuePairs[i], keyValuePairs[i + 1]);
+        }
+        return m;
+    }
+
+    /**
+     * Assert that a query returning a single MAP-typed column produces the expected
+     * sequence of maps (one map per row). Use {@code null} as an expected element to
+     * assert a row with a null cell.
+     */
+    @SafeVarargs
+    @SuppressWarnings("varargs")
+    private final void assertMapRows(String ppl, Map<String, 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' field for query: " + ppl, actualRows);
+        assertEquals("Row count mismatch for query: " + ppl, expected.length, actualRows.size());
+        for (int i = 0; i < expected.length; i++) {
+            Map<String, Object> want = expected[i];
+            List<Object> got = actualRows.get(i);
+            assertEquals("Column count mismatch at row " + i + " for query: " + ppl, 1, got.size());
+            assertCellEquals(
+                "Cell mismatch at row " + i + " for query: " + ppl,
+                want,
+                got.get(0)
+            );
+        }
+    }
+
+    /**
+     * Assert rows match the expected sequence with row-order significance (for sort tests).
+     */
+    @SafeVarargs
+    @SuppressWarnings("varargs")
+    private final void assertRowsInOrder(String ppl, List<Object>... expected) throws IOException {
+        assertRows(ppl, expected);
+    }
+
+    /**
+     * Assert rows match the expected set without row-order significance. Used for
+     * GROUP BY result tables where the analytics-engine route doesn't guarantee a sort
+     * order in the absence of an explicit {@code sort} clause.
+     */
+    @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' field 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);
+        for (List<Object> want : expected) {
+            int matchIdx = -1;
+            for (int i = 0; i < remaining.size(); i++) {
+                if (rowsEqual(want, remaining.get(i))) {
+                    matchIdx = i;
+                    break;
+                }
+            }
+            assertTrue(
+                "Expected row not found in response for query: " + ppl + " — looking for " + want + ", remaining " + remaining,
+                matchIdx >= 0
+            );
+            remaining.remove(matchIdx);
+        }
+    }
+
+    /** Numeric-tolerant per-row equality, used by {@link #assertRowsAnyOrder}. */
+    private static boolean rowsEqual(List<Object> want, List<Object> got) {
+        if (want.size() != got.size()) return false;
+        for (int i = 0; i < want.size(); i++) {
+            try {
+                assertCellEquals("rowsEqual", want.get(i), got.get(i));
+            } catch (AssertionError ae) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * Assert rows match the expected sequence. JSON parsing turns numbers into
+     * Integer/Long/Double interchangeably, so cell comparison goes through a
+     * numeric-tolerant comparator.
+     */
+    @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' field 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)
+                );
+            }
+        }
+    }
+
+    /** {@code POST /_analytics/ppl} and parse the JSON body. */
+    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);
+    }
+
+    /**
+     * Numeric-tolerant cell comparison. Numbers cross-compare via {@code doubleValue()};
+     * Maps compare key-set + per-key recursion; everything else falls back to
+     * {@link java.util.Objects#equals}.
+     */
+    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;
+        }
+        if (expected instanceof Map && actual instanceof Map) {
+            @SuppressWarnings("unchecked")
+            Map<String, Object> em = (Map<String, Object>) expected;
+            @SuppressWarnings("unchecked")
+            Map<String, Object> am = (Map<String, Object>) actual;
+            assertEquals(
+                message + ": map key set mismatch",
+                new java.util.HashSet<>(em.keySet()),
+                new java.util.HashSet<>(am.keySet())
+            );
+            for (Map.Entry<String, Object> entry : em.entrySet()) {
+                assertCellEquals(
+                    message + ": map[" + entry.getKey() + "]",
+                    entry.getValue(),
+                    am.get(entry.getKey())
+                );
+            }
+            return;
+        }
+        assertEquals(message, expected, actual);
+    }
+}
diff --git a/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_auto/bulk.json b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_auto/bulk.json
new file mode 100644
index 0000000000000..6da213f38d84a
--- /dev/null
+++ b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_auto/bulk.json
@@ -0,0 +1,3 @@
+{"index": {}}
+{"nested_doc": "{\"user\":{\"name\":\"John\"}}", "array_doc": "{\"tags\":[\"java\",\"sql\"]}", "merge_doc": "{\"a\":{\"b\":1},\"a.b\":2}", "stringify_doc": "{\"n\":30,\"b\":true,\"x\":null}", "empty_doc": "{}", "malformed_doc": "{\"user\":{\"name\":"}
+
diff --git a/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_auto/mapping.json b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_auto/mapping.json
new file mode 100644
index 0000000000000..286618f70fa06
--- /dev/null
+++ b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_auto/mapping.json
@@ -0,0 +1,16 @@
+{
+    "settings": {
+        "number_of_shards": 1,
+        "number_of_replicas": 0
+    },
+    "mappings": {
+        "properties": {
+            "nested_doc": { "type": "keyword" },
+            "array_doc": { "type": "keyword" },
+            "merge_doc": { "type": "keyword" },
+            "stringify_doc": { "type": "keyword" },
+            "empty_doc": { "type": "keyword" },
+            "malformed_doc": { "type": "keyword" }
+        }
+    }
+}
diff --git a/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_cmd/bulk.json b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_cmd/bulk.json
new file mode 100644
index 0000000000000..64e7779ae33c4
--- /dev/null
+++ b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_cmd/bulk.json
@@ -0,0 +1,5 @@
+{"index": {}}
+{"doc": "{\"user\":{\"name\":\"John\",\"age\":30}}"}
+{"index": {}}
+{"doc": "{\"user\":{\"name\":\"Alice\",\"age\":25}}"}
+
diff --git a/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_cmd/mapping.json b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_cmd/mapping.json
new file mode 100644
index 0000000000000..cda9043c22a9d
--- /dev/null
+++ b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_cmd/mapping.json
@@ -0,0 +1,11 @@
+{
+    "settings": {
+        "number_of_shards": 1,
+        "number_of_replicas": 0
+    },
+    "mappings": {
+        "properties": {
+            "doc": { "type": "keyword" }
+        }
+    }
+}
diff --git a/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_null/bulk.json b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_null/bulk.json
new file mode 100644
index 0000000000000..6e32ff0d62ec3
--- /dev/null
+++ b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_null/bulk.json
@@ -0,0 +1,5 @@
+{"index": {}}
+{"doc": "{\"n\": 1}"}
+{"index": {}}
+{"doc": null}
+
diff --git a/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_null/mapping.json b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_null/mapping.json
new file mode 100644
index 0000000000000..cda9043c22a9d
--- /dev/null
+++ b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_null/mapping.json
@@ -0,0 +1,11 @@
+{
+    "settings": {
+        "number_of_shards": 1,
+        "number_of_replicas": 0
+    },
+    "mappings": {
+        "properties": {
+            "doc": { "type": "keyword" }
+        }
+    }
+}
diff --git a/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_simple/bulk.json b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_simple/bulk.json
new file mode 100644
index 0000000000000..03a0984ccc7b4
--- /dev/null
+++ b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_simple/bulk.json
@@ -0,0 +1,7 @@
+{"index": {}}
+{"doc": "{\"n\": 1}"}
+{"index": {}}
+{"doc": "{\"n\": 2}"}
+{"index": {}}
+{"doc": "{\"n\": 3}"}
+
diff --git a/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_simple/mapping.json b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_simple/mapping.json
new file mode 100644
index 0000000000000..cda9043c22a9d
--- /dev/null
+++ b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/spath_simple/mapping.json
@@ -0,0 +1,11 @@
+{
+    "settings": {
+        "number_of_shards": 1,
+        "number_of_replicas": 0
+    },
+    "mappings": {
+        "properties": {
+            "doc": { "type": "keyword" }
+        }
+    }
+}