From d64afe45eaf297e326a7afbad20e4885abdfda0f Mon Sep 17 00:00:00 2001 From: Marc Handalian Date: Tue, 1 Sep 2026 21:20:11 +0000 Subject: [PATCH] Support OpenSearch object fields in the analytics engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An `object` couldn't be addressed as a whole value — `fields city`, `stats ... by city` failed with "Field [city] not found", because the schema flattened objects into dotted leaf columns and never added the parent. Six of the eight APM service-map PPL queries hit this. Expose the object as a ROW column and have ObjectStructMaterializer rewrite each scan to read only the leaves, rebuilding the object above it with a new `make_struct`. The project reproduces the scan's original row type exactly, so input refs above stay valid; sub-objects nest, so any depth resolves in one pass. The struct can't survive into the scan — an object has no physical storage. Runs before trimFields so unreferenced make_structs are dropped and leaf predicates still push down. MakeStructCallConverter serializes it to substrait, bypassing isthmus' signature matching — that matcher needs one type for all variadic operands, which interleaved (name, value) pairs don't have, and enumerating arities would cap struct width. Struct cells render like _source: an empty object is null, an empty sub-object is omitted, nested leaves are formatted. A shapeless `{"type": "object"}` stays addressable and null, matching vanilla. Verified against a lucene-only cluster — identical types and nested values across `select *`, whole objects, intermediate objects and leaves. Known: the response schema says `type: unknown` rather than `struct`, needing a one-line opensearch-sql fix. Signed-off-by: Marc Handalian Signed-off-by: Marc Handalian --- .../schema/OpenSearchSchemaBuilder.java | 82 +++++- .../opensearch/analytics/spi/FieldType.java | 5 + .../analytics/spi/MakeStructFunction.java | 81 ++++++ .../analytics/spi/ScalarFunction.java | 11 +- .../DataFusionAnalyticsBackendPlugin.java | 22 ++ .../DataFusionFragmentConvertor.java | 71 ++++- .../be/datafusion/DataFusionPlugin.java | 2 + .../datafusion/MakeStructCallConverter.java | 93 +++++++ .../opensearch_struct_functions.yaml | 38 +++ .../MakeStructCallConverterTests.java | 144 ++++++++++ .../analytics/exec/ArrowValues.java | 43 +++ .../planner/ObjectStructMaterializer.java | 257 ++++++++++++++++++ .../analytics/planner/PlannerImpl.java | 2 + .../engine/OpenSearchSchemaBuilderTests.java | 73 ++++- .../analytics/exec/ArrowValuesTests.java | 97 +++++++ .../planner/BasePlannerRulesTests.java | 2 +- .../planner/MockDataFusionBackend.java | 7 +- .../planner/ObjectStructPlanShapeTests.java | 217 +++++++++++++++ .../opensearch/parquet/ParquetSettings.java | 14 +- .../src/main/rust/src/merge/schema.rs | 10 + .../analytics/qa/ApmServiceMapObjectIT.java | 224 +++++++++++++++ .../analytics/qa/ObjectFieldIT.java | 149 ++++++++-- .../analytics/qa/ObjectFieldMultiShardIT.java | 94 +++++++ .../datasets/apm_service_map/bulk.json | 6 + .../datasets/apm_service_map/mapping.json | 63 +++++ 25 files changed, 1778 insertions(+), 29 deletions(-) create mode 100644 sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/MakeStructFunction.java create mode 100644 sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/MakeStructCallConverter.java create mode 100644 sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_struct_functions.yaml create mode 100644 sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/MakeStructCallConverterTests.java create mode 100644 sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/ObjectStructMaterializer.java create mode 100644 sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/ObjectStructPlanShapeTests.java create mode 100644 sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ApmServiceMapObjectIT.java create mode 100644 sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ObjectFieldMultiShardIT.java create mode 100644 sandbox/qa/analytics-engine-rest/src/test/resources/datasets/apm_service_map/bulk.json create mode 100644 sandbox/qa/analytics-engine-rest/src/test/resources/datasets/apm_service_map/mapping.json diff --git a/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java b/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java index 28aa6cd2f06ea..3ef3a4dc05702 100644 --- a/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java +++ b/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java @@ -27,6 +27,7 @@ import org.opensearch.core.common.Strings; import org.opensearch.index.IndexNotFoundException; +import java.util.ArrayList; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; @@ -350,8 +351,34 @@ private static void addLeafFields( // Recurse into sub-properties so dotted leaf paths ("city.location.latitude") appear as flat columns. if (fieldType == null || "object".equals(fieldType)) { Map nested = (Map) fieldProps.get("properties"); - if (nested != null) { + if (nested == null) { + if ("object".equals(fieldType) == false) { + // No "type" AND no "properties" — a malformed mapping entry, not an object. + // Skip it, as before; only an explicit `"type": "object"` is shapeless. + continue; + } + // Shapeless object — `{"type": "object"}` with no properties, which is what + // dynamic mapping leaves before any document populates it. Nothing is known + // about its shape, but the field must stay addressable and resolve to null, as + // vanilla does. A field-less ROW gives that: ObjectStructMaterializer finds no + // leaves to assemble and emits a typed NULL. Note the parent of a shapeless + // child does NOT carry it in its own struct type (buildObjectType skips it), so + // `fields outer` omits it while `fields outer.shapeless` returns null — again + // matching vanilla. + builder.add(fieldName, emptyStruct(typeFactory)); + continue; + } + { addLeafFields(builder, typeFactory, nested, fieldName); + // Also expose the object itself as a struct (ROW) column, so a query can + // address the whole object (`fields nested_metadata`, `stats … by obj`) and + // not just its leaves. The object has no physical storage — the scan reads + // the leaves — so ObjectStructMaterializer strips this column from the scan + // and re-assembles it with make_struct in a project directly above it. + RelDataType structType = buildObjectType(typeFactory, nested, fieldName); + if (structType != null) { + builder.add(fieldName, structType); + } } continue; } @@ -372,6 +399,59 @@ private static void addLeafFields( } } + /** Nullable ROW with no fields — an object whose shape is unknown; always resolves to null. */ + private static RelDataType emptyStruct(RelDataTypeFactory typeFactory) { + return typeFactory.createTypeWithNullability(typeFactory.createStructType(List.of(), List.of()), true); + } + + /** + * Builds the struct (ROW) type for an {@code object} mapping: one struct field per supported + * sub-field, recursing for sub-objects. Field names are the local names (the struct + * nesting already carries the path), while the flat leaf columns added by + * {@link #addLeafFields} keep their dotted paths — that dotted convention is how + * {@code ObjectStructMaterializer} pairs a struct field back to its backing column. + * + *

Returns {@code null} when the object contributes no supported field, so callers omit the + * column entirely rather than declaring an empty struct. + */ + @SuppressWarnings("unchecked") + private static RelDataType buildObjectType(RelDataTypeFactory typeFactory, Map properties, String pathPrefix) { + List types = new ArrayList<>(); + List names = new ArrayList<>(); + for (Map.Entry fieldEntry : properties.entrySet()) { + String localName = fieldEntry.getKey(); + Map fieldProps = (Map) fieldEntry.getValue(); + String fieldType = (String) fieldProps.get("type"); + RelDataType childType; + if (fieldType == null || "object".equals(fieldType)) { + Map nested = (Map) fieldProps.get("properties"); + childType = nested == null ? null : buildObjectType(typeFactory, nested, pathPrefix + "." + localName); + } else if ("nested".equals(fieldType)) { + // Array-of-sub-docs needs LIST + UNNEST; out of scope here. + childType = null; + } else { + // An ordinary scalar leaf (keyword, long, date_nanos, …). Typed exactly as + // addLeafFields types the equivalent flat column — including `scaling_factor`, + // the mapping parameter of `scaled_float` (a float persisted as a scaled long), + // which buildLeafType needs to type that field. parseScalingFactor yields NaN + // when absent, which buildLeafType treats as "not a scaled_float". + double scalingFactor = parseScalingFactor(fieldProps.get(SCALING_FACTOR_FIELD)); + childType = buildLeafType(fieldType, (String) fieldProps.get("format"), scalingFactor, typeFactory); + } + if (childType == null) { + // Unsupported sub-field (geo_point, nested, …) — dropped from the struct for the + // same reason it is dropped from the flat columns. + continue; + } + names.add(localName); + types.add(childType); + } + if (names.isEmpty()) { + return null; + } + return typeFactory.createTypeWithNullability(typeFactory.createStructType(types, names), true); + } + /** * Normalizes a scaling_factor value (Number or String from mapping JSON) to a double. * Returns {@link Double#NaN} when the value is null or unparseable, signaling the caller 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 1148b4ddb6a83..5cc88b5a0b70f 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 @@ -144,6 +144,11 @@ public static FieldType fromSqlTypeName(SqlTypeName sqlTypeName) { case BINARY, VARBINARY -> FieldType.BINARY; case ARRAY -> FieldType.ARRAY; case MAP -> FieldType.MAP; + // A Calcite ROW is the struct shape an `object` mapping materializes into (see + // ObjectStructMaterializer). Without this case, any struct-typed expression fails + // capability resolution in OpenSearchProjectRule and surfaces as + // UnsupportedFunctionException rather than dispatching to a backend. + case ROW -> OBJECT; default -> null; }; } diff --git a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/MakeStructFunction.java b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/MakeStructFunction.java new file mode 100644 index 0000000000000..cd715b9a21104 --- /dev/null +++ b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/MakeStructFunction.java @@ -0,0 +1,81 @@ +/* + * 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.spi; + +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rex.RexBuilder; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.SqlFunction; +import org.apache.calcite.sql.SqlFunctionCategory; +import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.type.OperandTypes; +import org.apache.calcite.sql.type.SqlTypeName; + +import java.util.ArrayList; +import java.util.List; + +/** + * {@code make_struct('name0', value0, 'name1', value1, ...)} → ROW. + * + *

Reassembles an OpenSearch {@code object} from the flat dotted leaf columns a scan produces, so + * a projection or aggregate can address the object as one value. Sub-objects nest the call: + * + * 

+ * make_struct('top', $1, 'properties', make_struct('name', $2, 'value', $3))
+ *
+ * + *

The interleaved {@code (name, value, …)} form is what backends consume; how one lowers and + * serializes the call is its own concern. The return type is always supplied by the caller via + * {@link #makeCall} — operand-driven inference is a placeholder, since the authoritative ROW type + * comes from the index mapping. + * + * @opensearch.internal + */ +public final class MakeStructFunction { + + /** The function name used in Calcite plans and Substrait serialization. */ + public static final String NAME = "make_struct"; + + /** Singleton Calcite SqlFunction: {@code make_struct(VARCHAR, ANY, ...) → ROW}. */ + public static final SqlFunction FUNCTION = new SqlFunction( + NAME, + SqlKind.OTHER_FUNCTION, + opBinding -> opBinding.getTypeFactory().createSqlType(SqlTypeName.ANY), + null, + OperandTypes.VARIADIC, + SqlFunctionCategory.USER_DEFINED_FUNCTION + ); + + private MakeStructFunction() {} + + /** + * Builds {@code make_struct('f0', v0, 'f1', v1, ...)} with an explicit ROW return type. + * + * @param rexBuilder builder for the enclosing plan + * @param structType the ROW type this call produces (from the index mapping) + * @param fieldNames struct field names, in order + * @param fieldValues struct field value expressions, positionally paired with {@code fieldNames} + */ + public static RexNode makeCall(RexBuilder rexBuilder, RelDataType structType, List fieldNames, List fieldValues) { + if (fieldNames.size() != fieldValues.size()) { + throw new IllegalArgumentException( + "make_struct requires one value per field name; got " + fieldNames.size() + " names and " + fieldValues.size() + " values" + ); + } + // VARCHAR, not CHAR: makeLiteral(String) yields CHAR(n), whose padding semantics are wrong + // for a field name, and backends depend on the distinction (see MakeStructCallConverter). + RelDataType nameType = rexBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR); + List operands = new ArrayList<>(fieldNames.size() * 2); + for (int i = 0; i < fieldNames.size(); i++) { + operands.add(rexBuilder.makeLiteral(fieldNames.get(i), nameType, true)); + operands.add(fieldValues.get(i)); + } + return rexBuilder.makeCall(structType, FUNCTION, operands); + } +} 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 eb74cc1719497..37a3ef309da51 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 @@ -455,7 +455,16 @@ public enum ScalarFunction { MINSPAN_BUCKET(Category.SCALAR, SqlKind.OTHER_FUNCTION), /** PPL range_bucket(value, data_min, data_max, start_param, end_param). VARCHAR label. */ - RANGE_BUCKET(Category.SCALAR, SqlKind.OTHER_FUNCTION); + RANGE_BUCKET(Category.SCALAR, SqlKind.OTHER_FUNCTION), + + // ── Composite (object / struct) construction ───────────────────── + /** + * {@code make_struct('name0', v0, 'name1', v1, ...)} → ROW. Materializes an OpenSearch + * {@code object} field from the flat dotted leaf columns a scan produces + * (see {@code ObjectStructMaterializer}). Nests for sub-objects. + * See {@link MakeStructFunction}. + */ + MAKE_STRUCT(Category.SCALAR, SqlKind.OTHER_FUNCTION); /** * Category of scalar function. 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 1a96960f0e09a..e64b9f8e77284 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 @@ -473,6 +473,13 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP */ private static final Set POLYMORPHIC_RETURN_PROJECT_OPS = Set.of(ScalarFunction.CAST, ScalarFunction.SAFE_CAST); + /** + * {@code MAKE_STRUCT} returns OBJECT rather than a scalar, so capability lookup — which keys on + * return type — needs it declared here rather than in SUPPORTED_FIELD_TYPES. Widening that set + * would wrongly claim filter/sort/aggregate support over structs. + */ + private static final Set OBJECT_RETURNING_PROJECT_OPS = Set.of(ScalarFunction.MAKE_STRUCT); + private static final Set AGG_FUNCTIONS = Set.of( AggregateFunction.SUM, AggregateFunction.SUM0, @@ -635,6 +642,9 @@ public Set projectCapabilities() { for (ScalarFunction op : MAP_RETURNING_PROJECT_OPS) { caps.add(new ProjectCapability.Scalar(op, Set.of(FieldType.MAP), formats, true)); } + for (ScalarFunction op : OBJECT_RETURNING_PROJECT_OPS) { + caps.add(new ProjectCapability.Scalar(op, Set.of(FieldType.OBJECT), formats, true)); + } for (ScalarFunction op : POLYMORPHIC_RETURN_PROJECT_OPS) { for (FieldType ft : FieldType.values()) { caps.add(new ProjectCapability.Scalar(op, Set.of(ft), formats, true)); @@ -647,6 +657,18 @@ public Set projectCapabilities() { public Set aggregateCapabilities() { Set formats = Set.copyOf(plugin.getSupportedFormats()); Set caps = new HashSet<>(); + // Aggregates over a struct-typed (object) column. Registered separately because + // FieldType.OBJECT is deliberately NOT in SUPPORTED_FIELD_TYPES — most aggregates + // are meaningless on a struct (no ordering ⇒ MIN/MAX, no arithmetic ⇒ SUM/AVG), + // and adding OBJECT there would also wrongly claim filter/sort support. + // + // COUNT is the meaningful one (non-null count) and is what the SQL plugin's + // Lucene path supports today, i.e. `stats count()`. Without this the + // capability lookup for (COUNT, OBJECT) misses and the query fails at plan time + // with "Function [COUNT] is not currently supported as an aggregate function". + for (FieldType objectType : Set.of(FieldType.OBJECT)) { + caps.add(new AggregateCapability(AggregateFunction.COUNT, Set.of(objectType), formats)); + } for (AggregateFunction func : AGG_FUNCTIONS) { for (FieldType type : SUPPORTED_FIELD_TYPES) { // 3-arg constructor leaves decomposition=null so the 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 56a911e309abe..ff88d986c3fe7 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 @@ -57,6 +57,7 @@ import java.math.MathContext; import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.function.Function; @@ -66,6 +67,7 @@ import io.substrait.expression.ImmutableAggregateFunctionInvocation; import io.substrait.extension.ExtensionCollector; import io.substrait.extension.SimpleExtension; +import io.substrait.isthmus.CallConverter; import io.substrait.isthmus.ConverterProvider; import io.substrait.isthmus.SubstraitRelVisitor; import io.substrait.isthmus.TypeConverter; @@ -577,7 +579,8 @@ private byte[] convertToSubstrait(RelNode fragment) { throw new IllegalStateException("Substrait conversion rejected the plan: " + e.getMessage(), e); } - List fieldNames = root.fields.stream().map(field -> field.getValue()).toList(); + // Root output names; flattened so struct columns contribute their nested names too. + List fieldNames = flattenNamesForSubstrait(root.fields, preprocessed.getRowType()); Plan.Root substraitRoot = Plan.Root.builder().input(substraitRel).names(fieldNames).build(); Plan plan = Plan.builder().addRoots(substraitRoot).build(); @@ -610,7 +613,59 @@ static Plan rewire(Plan inner, Rel wrapper, List wrapperNames) { /** Wrapper's output column names from its Calcite row type. */ private static List fieldNames(RelNode fragment) { - return fragment.getRowType().getFieldList().stream().map(RelDataTypeField::getName).toList(); + return flattenNamesForSubstrait(fragment.getRowType()); + } + + /** + * Flattens column names for a Substrait {@code Plan.Root} / {@code NamedStruct}: one flat + * depth-first list naming every field at every nesting level. + * + * 
+     * row type:  id INTEGER, meta ROW(top VARCHAR, props ROW(name VARCHAR))
+     * Substrait: ["id", "meta", "top", "props", "name"]
+     *
+ * + * It must be complete because a Substrait struct expression is positional — values only, no + * names — so emitting just the top level fails with "Named schema must contain names for all + * fields". + * + *

This overload takes {@link RelRoot} fields, whose names may be aliases and whose index into + * {@code rowType} need not match list position, so types are looked up by that index. The + * {@link #flattenNamesForSubstrait(RelDataType)} overload pairs by position instead. + */ + private static List flattenNamesForSubstrait(List> rootFields, RelDataType rowType) { + List fields = rowType.getFieldList(); + List flattened = new ArrayList<>(rootFields.size()); + for (Map.Entry rootField : rootFields) { + flattened.add(rootField.getValue()); + int index = rootField.getKey(); + if (index >= 0 && index < fields.size() && fields.get(index).getType().isStruct()) { + appendNestedNames(flattened, fields.get(index).getType()); + } + } + return flattened; + } + + /** Row-type-driven overload: names come from the row type itself, so positions align by construction. */ + private static List flattenNamesForSubstrait(RelDataType rowType) { + List flattened = new ArrayList<>(rowType.getFieldCount()); + for (RelDataTypeField field : rowType.getFieldList()) { + flattened.add(field.getName()); + if (field.getType().isStruct()) { + appendNestedNames(flattened, field.getType()); + } + } + return flattened; + } + + /** Appends a struct's field names depth-first (children before the next sibling). */ + private static void appendNestedNames(List out, RelDataType structType) { + for (RelDataTypeField child : structType.getFieldList()) { + out.add(child.getName()); + if (child.getType().isStruct()) { + appendNestedNames(out, child.getType()); + } + } } private static Rel replaceInput(Rel wrapper, Rel newInput) { @@ -798,7 +853,17 @@ protected ImmutableList getSigs() { aggConverter, windowConverter, typeConverter - ); + ) { + @Override + public List getCallConverters() { + // Struct construction is offered before signature matching — see + // MakeStructCallConverter for why the matcher can't handle it. + List converters = new ArrayList<>(); + converters.add(new MakeStructCallConverter(extensions, typeConverter)); + converters.addAll(super.getCallConverters()); + return converters; + } + }; return new SubstraitRelVisitor(converterProvider) { @Override public Rel visit(org.apache.calcite.rel.core.Aggregate aggregate) { diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionPlugin.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionPlugin.java index e05d6127c9c0d..662729360b705 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionPlugin.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionPlugin.java @@ -738,6 +738,7 @@ private static SimpleExtension.ExtensionCollection loadSubstraitExtensions() { SimpleExtension.ExtensionCollection delegationExtensions = SimpleExtension.load(List.of("/delegation_functions.yaml")); SimpleExtension.ExtensionCollection scalarExtensions = SimpleExtension.load(List.of("/opensearch_scalar_functions.yaml")); SimpleExtension.ExtensionCollection arrayExtensions = SimpleExtension.load(List.of("/opensearch_array_functions.yaml")); + SimpleExtension.ExtensionCollection structExtensions = SimpleExtension.load(List.of("/opensearch_struct_functions.yaml")); SimpleExtension.ExtensionCollection aggregateExtensions = SimpleExtension.load(List.of("/opensearch_aggregate_functions.yaml")); SimpleExtension.ExtensionCollection windowExtensions = SimpleExtension.load(List.of("/opensearch_window_functions.yaml")); SimpleExtension.ExtensionCollection arithmeticOverloads = SimpleExtension.load( @@ -746,6 +747,7 @@ private static SimpleExtension.ExtensionCollection loadSubstraitExtensions() { return DefaultExtensionCatalog.DEFAULT_COLLECTION.merge(delegationExtensions) .merge(scalarExtensions) .merge(arrayExtensions) + .merge(structExtensions) .merge(aggregateExtensions) .merge(windowExtensions) .merge(arithmeticOverloads); diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/MakeStructCallConverter.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/MakeStructCallConverter.java new file mode 100644 index 0000000000000..67b30e6117b71 --- /dev/null +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/MakeStructCallConverter.java @@ -0,0 +1,93 @@ +/* + * 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.be.datafusion; + +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexNode; +import org.opensearch.analytics.spi.MakeStructFunction; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.function.Function; + +import io.substrait.expression.Expression; +import io.substrait.expression.FunctionArg; +import io.substrait.extension.SimpleExtension; +import io.substrait.isthmus.CallConverter; +import io.substrait.isthmus.TypeConverter; + +/** + * Serializes {@code make_struct} to a Substrait {@link Expression.ScalarFunctionInvocation} built + * directly, rather than letting isthmus match it against a declared signature. Operands are + * forwarded unchanged, including the field-name literals, since DataFusion's {@code named_struct} + * takes the interleaved {@code (name, value, …)} form. + * + *

Bypassing the matcher is what leaves the number of struct fields unbounded: isthmus binds a + * variadic function through a {@code SingularArgumentMatcher}, which derives one type every operand + * must satisfy, and {@code named_struct}'s interleaved operands have no such common type. Declaring + * one impl per field count matches but caps struct width. The extension declaration is therefore + * used only as an anchor (name + URN) for the consumer to resolve by name. + * + *

Substrait's native {@code Expression.NestedStruct} would be cleaner, but DataFusion rejects it + * at execution ("Nested struct expressions are not yet supported"). Revisit if that closes. + * + * @opensearch.internal + */ +class MakeStructCallConverter implements CallConverter { + + /** DataFusion's native struct constructor — the name the consumer resolves. */ + static final String NAMED_STRUCT = "named_struct"; + + private final SimpleExtension.ExtensionCollection extensions; + private final TypeConverter typeConverter; + + MakeStructCallConverter(SimpleExtension.ExtensionCollection extensions, TypeConverter typeConverter) { + this.extensions = extensions; + this.typeConverter = typeConverter; + } + + @Override + public Optional convert(RexCall call, Function topLevelConverter) { + String operator = call.getOperator().getName(); + // The engine emits `make_struct`; `named_struct` is accepted too so the converter stays + // correct if a rename ever reaches it first. + if (!MakeStructFunction.NAME.equalsIgnoreCase(operator) && !NAMED_STRUCT.equalsIgnoreCase(operator)) { + return Optional.empty(); + } + + Optional declaration = findNamedStructDeclaration(); + if (declaration.isEmpty()) { + // No anchor to reference — decline so the failure surfaces as isthmus' normal + // "Unable to convert call" rather than an NPE deep in proto serialization. + return Optional.empty(); + } + + List arguments = new ArrayList<>(call.getOperands().size()); + for (RexNode operand : call.getOperands()) { + arguments.add(topLevelConverter.apply(operand)); + } + + return Optional.of( + Expression.ScalarFunctionInvocation.builder() + .declaration(declaration.get()) + .addAllArguments(arguments) + .outputType(typeConverter.toSubstrait(call.getType())) + .build() + ); + } + + /** + * Any declared {@code named_struct} variant works as the anchor — the consumer resolves the + * function by name, and the arity we attach is independent of the variant's declared arity. + */ + private Optional findNamedStructDeclaration() { + return extensions.scalarFunctions().stream().filter(variant -> NAMED_STRUCT.equals(variant.name())).findFirst(); + } +} diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_struct_functions.yaml b/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_struct_functions.yaml new file mode 100644 index 0000000000000..1d2e397fac9f8 --- /dev/null +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_struct_functions.yaml @@ -0,0 +1,38 @@ +%YAML 1.2 +--- +# Substrait extension declaring DataFusion's native struct constructor. +# +# This file exists ONLY to give the function an extension anchor (URN + name) that the consumer +# resolves by name. The argument list is attached by MakeStructCallConverter, which builds the +# Expression.ScalarFunctionInvocation directly and therefore never goes through isthmus' +# signature matcher — so the single impl below does NOT bound the number of struct fields. +# +# WHY THE MATCHER IS BYPASSED (and why this is not an enumeration of arities): +# isthmus binds a variadic function via a SingularArgumentMatcher — it derives ONE type that every +# variadic operand must satisfy (the same reason MakeArrayAdapter widens make_array's operands to a +# common element type first). named_struct interleaves `string` field names with values of +# unrelated types, so no such type exists. Verified against isthmus 0.89.1: a variadic impl matches +# neither as `value: any1` nor as unconstrained `value: any`, with +# `parameterConsistency: INCONSISTENT` in both cases. Enumerating one impl per field count DOES +# match but is bounded, and an object's width is data-dependent (an OTel span's `attributes` +# carries ~55 sub-fields and grows), so enumeration is not a shippable answer. +# +# Substrait models struct construction natively as Expression.NestedStruct, which would need no +# declaration at all, but DataFusion's substrait consumer rejects it at execution: +# "This feature is not implemented: Nested struct expressions are not yet supported". +# DataFusion does implement `named_struct` as a scalar function at any arity, so the +# function-invocation form is the one that executes. +urn: extension:org.opensearch:struct_functions +scalar_functions: + - name: named_struct + description: >- + Construct a struct from interleaved (field-name, value) pairs: + named_struct('f0', v0, 'f1', v1, ...). Anchor declaration only — see the header. The real + arguments are attached by MakeStructCallConverter and are not limited to this arity. + impls: + - args: + - value: string + name: "name" + - value: any1 + name: "value" + return: any diff --git a/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/MakeStructCallConverterTests.java b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/MakeStructCallConverterTests.java new file mode 100644 index 0000000000000..1aa0e7b165ddd --- /dev/null +++ b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/MakeStructCallConverterTests.java @@ -0,0 +1,144 @@ +/* + * 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.be.datafusion; + +import org.apache.calcite.jdbc.JavaTypeFactoryImpl; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rex.RexBuilder; +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.fun.SqlStdOperatorTable; +import org.apache.calcite.sql.type.SqlTypeName; +import org.opensearch.analytics.spi.MakeStructFunction; +import org.opensearch.test.OpenSearchTestCase; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.function.Function; + +import io.substrait.expression.Expression; +import io.substrait.extension.SimpleExtension; +import io.substrait.isthmus.TypeConverter; + +/** + * Substrait-serialization tests for {@link MakeStructCallConverter} — the layer both struct bugs + * lived in, and one a plan-shape test can't reach (it stops before Substrait) nor a REST IT cheaply + * (needs a live cluster plus the Rust native library). + */ +public class MakeStructCallConverterTests extends OpenSearchTestCase { + + private RelDataTypeFactory typeFactory; + private RexBuilder rexBuilder; + private MakeStructCallConverter converter; + + /** Converts operands the way isthmus would; enough for asserting arity and shape. */ + private Function operandConverter; + + @Override + public void setUp() throws Exception { + super.setUp(); + typeFactory = new JavaTypeFactoryImpl(); + rexBuilder = new RexBuilder(typeFactory); + SimpleExtension.ExtensionCollection extensions = SimpleExtension.load(List.of("/opensearch_struct_functions.yaml")); + converter = new MakeStructCallConverter(extensions, TypeConverter.DEFAULT); + // Stand-in for isthmus' recursive conversion; enough to count arguments. + operandConverter = rex -> Expression.StrLiteral.builder().value(rex.toString()).build(); + } + + /** 100 fields → one invocation with 200 operands. Fails if struct construction is ever routed + * back through signature matching, which cannot bind a variadic call of mixed operand types. */ + public void testWideStructHasNoArityCeiling() { + int fieldCount = 100; + Optional converted = converter.convert(makeStructCall(fieldCount), operandConverter); + + assertTrue("converter must handle a 100-field struct", converted.isPresent()); + Expression.ScalarFunctionInvocation invocation = asInvocation(converted.get()); + assertEquals("one argument per name and per value", fieldCount * 2, invocation.arguments().size()); + assertEquals("named_struct", invocation.declaration().name()); + } + + /** The common narrow case still works, and keeps the interleaved (name, value) ordering. */ + public void testNarrowStructKeepsInterleavedOperands() { + Optional converted = converter.convert(makeStructCall(2), operandConverter); + + assertTrue(converted.isPresent()); + assertEquals(4, asInvocation(converted.get()).arguments().size()); + } + + /** Unrelated operators must fall through so other converters (and function matching) still run. */ + public void testDeclinesUnrelatedOperator() { + RexNode left = rexBuilder.makeLiteral("a"); + RexNode right = rexBuilder.makeLiteral("b"); + RexCall unrelated = (RexCall) rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, left, right); + + assertFalse("EQUALS must not be claimed by the struct converter", converter.convert(unrelated, operandConverter).isPresent()); + } + + /** The converter's odd-operand guard is unreachable from our emitter: makeCall rejects + * mismatched name/value lists first. Asserts that contract, since a hand-rolled odd-arity + * RexCall isn't constructible under -ea. */ + public void testEmitterRejectsMismatchedNameAndValueCounts() { + List twoNames = List.of("f0", "f1"); + List oneValue = List.of(rexBuilder.makeLiteral("v0")); + + IllegalArgumentException thrown = expectThrows( + IllegalArgumentException.class, + () -> MakeStructFunction.makeCall(rexBuilder, structTypeOf(2), twoNames, oneValue) + ); + assertTrue( + "message should name the mismatch, got: " + thrown.getMessage(), + thrown.getMessage().contains("one value per field name") + ); + } + + /** No anchor declaration → decline, so it surfaces as "Unable to convert call" rather than an + * NPE. Guards the coupling to opensearch_struct_functions.yaml being on the classpath. */ + public void testDeclinesWhenAnchorDeclarationMissing() throws Exception { + MakeStructCallConverter withoutAnchor = new MakeStructCallConverter( + SimpleExtension.ExtensionCollection.builder().build(), + TypeConverter.DEFAULT + ); + + assertFalse(withoutAnchor.convert(makeStructCall(2), operandConverter).isPresent()); + } + + // ── helpers ────────────────────────────────────────────────────────────────────── + + /** {@code make_struct('f0', 'v0', 'f1', 'v1', …)} with {@code fieldCount} pairs. */ + private RexCall makeStructCall(int fieldCount) { + List names = new ArrayList<>(fieldCount); + List values = new ArrayList<>(fieldCount); + for (int i = 0; i < fieldCount; i++) { + names.add("f" + i); + values.add(rexBuilder.makeLiteral("v" + i)); + } + return (RexCall) MakeStructFunction.makeCall(rexBuilder, structTypeOf(fieldCount), names, values); + } + + /** ROW type with {@code fieldCount} VARCHAR fields. */ + private RelDataType structTypeOf(int fieldCount) { + List types = new ArrayList<>(fieldCount); + List names = new ArrayList<>(fieldCount); + for (int i = 0; i < fieldCount; i++) { + types.add(typeFactory.createSqlType(SqlTypeName.VARCHAR)); + names.add("f" + i); + } + return typeFactory.createStructType(types, names); + } + + private static Expression.ScalarFunctionInvocation asInvocation(Expression expression) { + assertTrue( + "expected a ScalarFunctionInvocation, got " + expression.getClass().getSimpleName(), + expression instanceof Expression.ScalarFunctionInvocation + ); + return (Expression.ScalarFunctionInvocation) expression; + } +} 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 e79af1e694ff8..2f6e236cf3d9b 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 @@ -13,6 +13,7 @@ import org.apache.arrow.vector.VectorSchemaRoot; import org.apache.arrow.vector.complex.ListVector; import org.apache.arrow.vector.complex.MapVector; +import org.apache.arrow.vector.complex.StructVector; import org.apache.arrow.vector.types.TimeUnit; import org.apache.arrow.vector.types.pojo.ArrowType; import org.apache.arrow.vector.types.pojo.Field; @@ -127,6 +128,9 @@ public static Object toJavaValue(FieldVector vector, int index) { if (vector instanceof VarCharVector v) { return spaceSeparator(new String(v.get(index), StandardCharsets.UTF_8)); } + if (vector instanceof StructVector sv) { + return structToMap(sv, index); + } // MapVector extends ListVector — must come first. if (vector instanceof MapVector && vector.getObject(index) instanceof List entries) { LinkedHashMap map = new LinkedHashMap<>(); @@ -150,6 +154,45 @@ public static Object toJavaValue(FieldVector vector, int index) { return normalize(value); } + /** + * Converts a struct cell to a sparse map, matching how {@code _source} renders an OpenSearch + * {@code object}. + * + *

Sparseness itself comes for free — {@code NonNullableStructVector.getObject} already skips + * children whose value is null, so an absent leaf is omitted while a leaf genuinely holding + * {@code ""} is kept. This walks the children explicitly for two things {@code getObject} does + * not do: + * + *

    + *
  • An object with nothing populated becomes {@code null}, not {@code {}}. + * That matches vanilla, where e.g. an unpopulated {@code traceGroupFields} comes back + * null. An empty sub-object is then dropped by its parent's loop, since the + * recursive call returns null and null children are skipped.
    • + *
  • Nested leaves are formatted like top-level ones. Recursing through + * {@link #toJavaValue} keeps {@link Text} → {@code String} and formats a nested + * date/time/timestamp instead of leaving a raw epoch number. {@code getObject} cannot: + * it hands back child values already stripped of the {@link Field} that says how to + * format them, and {@link #normalize}'s map branch has no type information to recover it. + * Compare the list path, which threads the child field through + * {@code normalizeList(raw, lv.getDataVector().getField())}.
    • + *
+ */ + private static Map structToMap(StructVector vector, int index) { + Map out = new LinkedHashMap<>(); + for (Field child : vector.getField().getChildren()) { + FieldVector childVector = vector.getChild(child.getName()); + if (childVector == null || childVector.isNull(index)) { + continue; + } + Object value = toJavaValue(childVector, index); + if (value == null) { + continue; + } + out.put(child.getName(), value); + } + return out.isEmpty() ? null : out; + } + /** ISO-T temporal → space separator; other strings unchanged. */ private static String spaceSeparator(String s) { if (s == null) return null; diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/ObjectStructMaterializer.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/ObjectStructMaterializer.java new file mode 100644 index 0000000000000..e0c9e02334970 --- /dev/null +++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/ObjectStructMaterializer.java @@ -0,0 +1,257 @@ +/* + * 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.planner; + +import org.apache.calcite.plan.RelOptAbstractTable; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.RelShuttleImpl; +import org.apache.calcite.rel.core.TableScan; +import org.apache.calcite.rel.logical.LogicalProject; +import org.apache.calcite.rel.logical.LogicalTableScan; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rel.type.RelDataTypeField; +import org.apache.calcite.rex.RexBuilder; +import org.apache.calcite.rex.RexNode; +import org.opensearch.analytics.spi.MakeStructFunction; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** + * Rebuilds OpenSearch {@code object} fields as structs in a project directly above the table scan. + * + *

An object is stored as flat dotted leaf columns and has no physical column of its own — + * {@code FieldStorageResolver} recurses past object parents because "object fields themselves have + * no storage" — while the schema still exposes it as a ROW column so queries can name it. So this + * strips struct columns from the scan and reassembles them above it: + * + * 

+ * LogicalProject(id=[$0], meta.top=[$1], meta.props.name=[$2],
+ *                meta=[make_struct('top', $1, 'props', make_struct('name', $2))])
+ *   LogicalTableScan(table=[[t]])      // leaves only
+ *
+ * + *

The project reproduces the scan's original row type exactly, so every {@code RexInputRef} + * above stays valid. Sub-objects nest another {@code make_struct}, so any depth resolves in one + * pass. + * + *

One-shot pass rather than a HEP rule: a rule matching {@code TableScan} and producing + * {@code Project(TableScan)} would re-match its own output. + * + *

Must run before {@code trimFields}. It emits a {@code make_struct} per object the scan + * declares, and the trimmer drops the unreferenced ones — otherwise a query filtering on one leaf + * pays for, and can fail on, an unrelated object. Leaf pushdown still works, since the leaves stay + * in this project's output and {@code FILTER_PROJECT_TRANSPOSE} moves filters through it. + * + *

TODO: store objects as native Parquet structs instead of flat dotted leaves. A Parquet + * {@code STRUCT} yields the same leaf column chunks we write today, so pruning and encoding are + * unchanged — the only difference is a group node in the schema. The reader would then hand the + * object back assembled, making this pass, {@code MakeStructFunction}, and + * {@code MakeStructCallConverter} unnecessary for {@code object} support. Keep {@code make_struct} + * regardless: reconstructing {@code nested} from parallel {@code LIST} leaves needs it. + * + *

Blocked on the Rust merge path, which handles only top-level columns — + * {@code build_parquet_root_schema} unions segment schemas by top-level name with first-writer-wins, + * and {@code ColumnMapping} never null-fills a missing struct child, so a struct gaining a sub-field + * between segments fails the merge. The writer also has no {@code Struct} in + * {@code ParquetSettings.ARROW_TYPE_NAME_TO_INSTANCE}. + * + *

What changes on the search side, since it is more than deleting this class. Native storage + * gives the scan the ROW column only — there are no {@code city.name} columns — so every + * consumer of the dotted-leaf convention moves: + * + *

    + *
  • {@code OpenSearchSchemaBuilder} adds flat leaf columns today; they would become derived + * names resolving to struct children.
    • + *
  • {@code FieldStorageResolver} keys {@code fieldStorage} on the same dotted paths, built + * independently of the schema builder — the two agree by convention, not by contract. Keeping + * that naming stable is what decides whether this migration is cheap: if dotted names still + * resolve (to struct children instead of columns), nothing above those two files changes.
    • + *
  • {@link Materializer#resolveLeaf} becomes {@code GET_FIELD}; see its javadoc.
    • + *
  • Leaf predicate pushdown pushes a field access rather than a column reference.
    • + *
  • The plan-shape goldens in {@code ObjectStructPlanShapeTests} pin flat-leaf plans and will + * all change. Expected, not a regression.
    • + *
  • The companion rewrite noted for pushing struct-authored predicates down + * ({@code GET_FIELD(struct, 'x') → leaf ref}) becomes unnecessary — native storage wants the + * opposite direction.
    • + *
+ * + *

Acceptance tests already exist and are storage-agnostic: {@code ObjectFieldIT}, + * {@code ApmServiceMapObjectIT}, and {@code ObjectFieldMultiShardIT} assert PPL results — nested + * JSON shape, types, group counts — not plan shapes, so they should pass unchanged. + * + * @opensearch.internal + */ +public final class ObjectStructMaterializer { + + private ObjectStructMaterializer() {} + + /** + * Rewrites scans that expose struct-typed columns into a leaf-only scan plus a + * struct-materializing project. + * + * @return the rewritten plan, or {@link Optional#empty()} when the plan has no object + * columns (callers keep the original plan unchanged) + */ + public static Optional rewrite(RelNode root) { + Materializer materializer = new Materializer(); + RelNode rewritten = root.accept(materializer); + return materializer.changed ? Optional.of(rewritten) : Optional.empty(); + } + + private static final class Materializer extends RelShuttleImpl { + + private boolean changed = false; + + @Override + public RelNode visit(TableScan scan) { + RelDataType originalRowType = scan.getRowType(); + List originalFields = originalRowType.getFieldList(); + if (originalFields.stream().noneMatch(f -> f.getType().isStruct())) { + return scan; + } + + // Scan keeps only physically-stored (non-struct) columns. + RelDataTypeFactory typeFactory = scan.getCluster().getTypeFactory(); + RelDataTypeFactory.Builder leafTypeBuilder = typeFactory.builder(); + Map leafIndexByName = new HashMap<>(); + for (RelDataTypeField field : originalFields) { + if (field.getType().isStruct()) { + continue; + } + leafIndexByName.put(field.getName(), leafTypeBuilder.getFieldCount()); + leafTypeBuilder.add(field.getName(), field.getType()); + } + RelDataType leafRowType = leafTypeBuilder.build(); + if (leafRowType.getFieldCount() == 0) { + // Nothing physical to read — don't emit a scan with an empty row type. + return scan; + } + + RelOptTable leafTable = new LeafOnlyTable(scan.getTable(), leafRowType); + RelNode leafScan = LogicalTableScan.create(scan.getCluster(), leafTable, scan.getHints()); + + // Rebuild the original row type: leaves pass through, structs assemble in place. + RexBuilder rexBuilder = scan.getCluster().getRexBuilder(); + List projects = new ArrayList<>(originalFields.size()); + List names = new ArrayList<>(originalFields.size()); + for (RelDataTypeField field : originalFields) { + names.add(field.getName()); + if (field.getType().isStruct()) { + RexNode struct = buildStruct(rexBuilder, leafScan, field.getName(), field.getType(), leafIndexByName); + if (struct == null) { + // A backing leaf is absent — typed NULL, never a partial struct. + projects.add(rexBuilder.makeNullLiteral(field.getType())); + } else { + projects.add(struct); + } + } else { + projects.add(rexBuilder.makeInputRef(leafScan, leafIndexByName.get(field.getName()))); + } + } + + changed = true; + return LogicalProject.create(leafScan, List.of(), projects, names); + } + + /** + * Recursively builds {@code make_struct} for {@code structType}, resolving each leaf to the + * trimmed scan's column named {@code path + "." + fieldName}. Returns {@code null} when any + * leaf is missing, signaling the caller to skip materialization for this column. + */ + private static RexNode buildStruct( + RexBuilder rexBuilder, + RelNode leafScan, + String path, + RelDataType structType, + Map leafIndexByName + ) { + List fieldNames = new ArrayList<>(); + List fieldValues = new ArrayList<>(); + for (RelDataTypeField child : structType.getFieldList()) { + String childPath = path + "." + child.getName(); + RexNode value; + if (child.getType().isStruct()) { + value = buildStruct(rexBuilder, leafScan, childPath, child.getType(), leafIndexByName); + } else { + value = resolveLeaf(rexBuilder, leafScan, childPath, leafIndexByName); + } + if (value == null) { + return null; + } + fieldNames.add(child.getName()); + fieldValues.add(value); + } + if (fieldNames.isEmpty()) { + return null; + } + return MakeStructFunction.makeCall(rexBuilder, structType, fieldNames, fieldValues); + } + + /** + * Produces the value of one object leaf, given its dotted path. The single point where this + * pass depends on leaves being physical columns named by their dotted path. + * + *

Deliberately isolated: if objects are ever stored as native Parquet structs (see the + * TODO on this class), the leaf is no longer a column of its own and this becomes + * {@code GET_FIELD(structRef, name)}. Swapping one implementation is the whole change on this + * side; the recursion above it doesn't move. + * + * @return {@code null} when the leaf has no backing column, which tells the caller to skip + * materializing this object rather than emit a partial struct + */ + private static RexNode resolveLeaf( + RexBuilder rexBuilder, + RelNode leafScan, + String dottedPath, + Map leafIndexByName + ) { + Integer leafIndex = leafIndexByName.get(dottedPath); + return leafIndex == null ? null : rexBuilder.makeInputRef(leafScan, leafIndex); + } + } + + /** + * Wraps the scanned table with a row type stripped of struct columns, so downstream physical + * resolution ({@code FieldStorageResolver}) only ever sees fields that actually have storage. + * Mirrors the {@code IndexNameTable} wrapper in {@code OpenSearchTableScanRule}. + */ + private static final class LeafOnlyTable extends RelOptAbstractTable { + + private final RelOptTable delegate; + + LeafOnlyTable(RelOptTable delegate, RelDataType leafRowType) { + super(delegate.getRelOptSchema(), delegate.getQualifiedName().getLast(), leafRowType); + this.delegate = delegate; + } + + @Override + public List getQualifiedName() { + // Preserve the original qualified name: OpenSearchTableScanRule resolves the index + // from it, and RelOptAbstractTable would otherwise report a single-segment name. + return delegate.getQualifiedName(); + } + + @Override + public double getRowCount() { + return delegate.getRowCount(); + } + + @Override + public T unwrap(Class clazz) { + T unwrapped = super.unwrap(clazz); + return unwrapped != null ? unwrapped : delegate.unwrap(clazz); + } + } +} diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/PlannerImpl.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/PlannerImpl.java index fa6b73dc7ee2c..a6618927ecee1 100644 --- a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/PlannerImpl.java +++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/PlannerImpl.java @@ -128,6 +128,8 @@ public static RelNode runAllOptimizations(RelNode rawRelNode, PlannerContext con RelNode modifiedRelNode = rawRelNode; modifiedRelNode = removeSubQueries(modifiedRelNode, listener); + // Must run before trimFields — see ObjectStructMaterializer. + modifiedRelNode = ObjectStructMaterializer.rewrite(modifiedRelNode).orElse(modifiedRelNode); modifiedRelNode = trimFields(modifiedRelNode); modifiedRelNode = extractLiteralAgg(modifiedRelNode, listener); modifiedRelNode = reduceExpressions(modifiedRelNode, listener); diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java index 0cd11566a34e3..6cc443f210db5 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java @@ -120,7 +120,8 @@ public void testMultipleIndicesProduceMultipleTables() throws Exception { /** * Test that nested/object fields are skipped. */ - public void testNestedAndObjectFieldsSkipped() throws Exception { + /** {@code nested} is skipped; a shapeless {@code object} stays addressable (always null). */ + public void testNestedSkippedAndShapelessObjectAddressable() throws Exception { ClusterState clusterState = buildClusterState( Map.of("nested_index", Map.of("name", "keyword", "address", "object", "tags", "nested")) ); @@ -131,8 +132,10 @@ public void testNestedAndObjectFieldsSkipped() throws Exception { assertNotNull(table); RelDataType rowType = table.getRowType(new org.apache.calcite.jdbc.JavaTypeFactoryImpl()); - assertEquals("Should only have 'name' field, skipping object/nested", 1, rowType.getFieldCount()); + assertEquals("name plus the shapeless object; nested is skipped", 2, rowType.getFieldCount()); assertFieldType(rowType, "name", SqlTypeName.VARCHAR); + assertNotNull("shapeless object is addressable", rowType.getField("address", true, false)); + assertNull("nested stays unsupported", rowType.getField("tags", true, false)); } /** @@ -259,10 +262,22 @@ public void testNestedObjectWithUnsupportedLeafDropsOnlyThatLeaf() throws Except assertNotNull(table); RelDataType rowType = table.getRowType(new org.apache.calcite.jdbc.JavaTypeFactoryImpl()); - assertEquals("Only 2 supported nested leaves should remain", 2, rowType.getFieldCount()); + // 2 supported leaves + the struct-typed `customer` parent column (the object itself is + // now addressable; ObjectStructMaterializer assembles it from these leaves). + assertEquals("2 supported nested leaves plus the object parent", 3, rowType.getFieldCount()); assertFieldType(rowType, "customer.id", SqlTypeName.VARCHAR); assertFieldType(rowType, "customer.age", SqlTypeName.INTEGER); assertNull("nested geo_point leaf must be dropped", rowType.getField("customer.home", true, false)); + + // The unsupported sub-field is dropped from the struct too, for the same reason it is + // dropped from the flat columns. + RelDataTypeField parent = rowType.getField("customer", true, false); + assertNotNull("object parent must be exposed as a column", parent); + assertEquals(SqlTypeName.ROW, parent.getType().getSqlTypeName()); + assertEquals("struct carries only the supported sub-fields", 2, parent.getType().getFieldCount()); + assertNotNull(parent.getType().getField("id", true, false)); + assertNotNull(parent.getType().getField("age", true, false)); + assertNull("geo_point sub-field must be dropped from the struct", parent.getType().getField("home", true, false)); } /** @@ -857,4 +872,56 @@ private static String collectMessages(Throwable t) { } return sb.toString(); } + + /** + * A bare {@code {"type": "object"}} with no {@code properties} — what dynamic mapping leaves + * behind before any document populates the object. Nothing is known about its shape, so it gets + * a field-less ROW: addressable, and always resolving to null. Matches vanilla, measured on a + * lucene-only 3.8.0 cluster — {@code fields attrs} there gives schema {@code [(attrs, struct)]} + * with row {@code [null]}, and the field appears under {@code *} too. + */ + public void testBareObjectWithoutPropertiesIsAddressableAndEmpty() throws Exception { + String mapping = "{\"properties\":{" + "\"id\":{\"type\":\"keyword\"}," + "\"attrs\":{\"type\":\"object\"}" + "}}"; + ClusterState clusterState = buildClusterStateRaw("bare_object", mapping); + + SchemaPlus schema = OpenSearchSchemaBuilder.buildSchema(clusterState); + RelDataType rowType = schema.getTable("bare_object").getRowType(new org.apache.calcite.jdbc.JavaTypeFactoryImpl()); + + RelDataTypeField attrs = rowType.getField("attrs", true, false); + assertNotNull("a shapeless object must stay addressable", attrs); + assertTrue("must be a struct type", attrs.getType().isStruct()); + assertEquals("with no fields, since its shape is unknown", 0, attrs.getType().getFieldCount()); + assertTrue("nullable, since it always resolves to null", attrs.getType().isNullable()); + assertFieldType(rowType, "id", SqlTypeName.VARCHAR); + } + + /** + * Same one level down, and the asymmetry that matches vanilla: {@code outer.shapeless} is + * addressable and null, while {@code outer}'s own struct type does not carry it — so + * {@code fields outer} returns {@code {name: x}} and {@code fields outer.shapeless} returns + * null, exactly as measured against vanilla. + */ + public void testNestedBareObjectIsAddressableButAbsentFromParentStruct() throws Exception { + String mapping = "{\"properties\":{" + + "\"outer\":{\"properties\":{" + + "\"name\":{\"type\":\"keyword\"}," + + "\"shapeless\":{\"type\":\"object\"}" + + "}}" + + "}}"; + ClusterState clusterState = buildClusterStateRaw("bare_nested", mapping); + + SchemaPlus schema = OpenSearchSchemaBuilder.buildSchema(clusterState); + RelDataType rowType = schema.getTable("bare_nested").getRowType(new org.apache.calcite.jdbc.JavaTypeFactoryImpl()); + + RelDataTypeField shapeless = rowType.getField("outer.shapeless", true, false); + assertNotNull("nested shapeless object must stay addressable", shapeless); + assertEquals(0, shapeless.getType().getFieldCount()); + assertFieldType(rowType, "outer.name", SqlTypeName.VARCHAR); + + RelDataTypeField outer = rowType.getField("outer", true, false); + assertNotNull(outer); + assertEquals("parent's struct carries only the resolvable leaf", 1, outer.getType().getFieldCount()); + assertNotNull(outer.getType().getField("name", true, false)); + assertNull("shapeless child is not part of the parent struct", outer.getType().getField("shapeless", true, false)); + } } diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/ArrowValuesTests.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/ArrowValuesTests.java index 6914a1fddca93..4e127b5a15e0e 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/ArrowValuesTests.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/ArrowValuesTests.java @@ -29,6 +29,7 @@ import org.apache.arrow.vector.VarCharVector; import org.apache.arrow.vector.VectorSchemaRoot; import org.apache.arrow.vector.complex.ListVector; +import org.apache.arrow.vector.complex.StructVector; import org.apache.arrow.vector.complex.impl.UnionListWriter; import org.apache.arrow.vector.types.TimeUnit; import org.apache.arrow.vector.types.pojo.ArrowType; @@ -328,4 +329,100 @@ public void testToSourceMapDropsNullsAndPreservesFieldOrder() { assertFalse(out.containsKey("missing")); } } + + // ── struct cells: sparse like _source, and formatted at any depth ────────────────── + // + // getObject already skips null children, so the first test guards behaviour we inherit rather + // than behaviour structToMap adds. The other three fail without it: getObject returns {} for an + // all-absent struct instead of null, and hands back child values stripped of the Field needed + // to format a nested temporal. + + /** An absent leaf is omitted; a leaf genuinely holding "" is kept. Inherited from getObject. */ + public void testStructOmitsAbsentLeafButKeepsEmptyString() { + try (StructVector sv = StructVector.empty("obj", allocator)) { + VarCharVector present = sv.addOrGet("present", FieldType.nullable(new ArrowType.Utf8()), VarCharVector.class); + VarCharVector empty = sv.addOrGet("empty", FieldType.nullable(new ArrowType.Utf8()), VarCharVector.class); + VarCharVector absent = sv.addOrGet("absent", FieldType.nullable(new ArrowType.Utf8()), VarCharVector.class); + sv.allocateNew(); + present.setSafe(0, "v".getBytes(java.nio.charset.StandardCharsets.UTF_8)); + empty.setSafe(0, new byte[0]); + absent.setNull(0); + sv.setIndexDefined(0); + sv.setValueCount(1); + present.setValueCount(1); + empty.setValueCount(1); + absent.setValueCount(1); + + Object value = ArrowValues.toJavaValue(sv, 0); + + assertEquals(Map.of("present", "v", "empty", ""), value); + } + } + + /** A struct with nothing populated is null, not an empty map — matches vanilla. */ + public void testStructWithNothingPopulatedIsNull() { + try (StructVector sv = StructVector.empty("obj", allocator)) { + VarCharVector a = sv.addOrGet("a", FieldType.nullable(new ArrowType.Utf8()), VarCharVector.class); + VarCharVector b = sv.addOrGet("b", FieldType.nullable(new ArrowType.Utf8()), VarCharVector.class); + sv.allocateNew(); + a.setNull(0); + b.setNull(0); + sv.setIndexDefined(0); + sv.setValueCount(1); + a.setValueCount(1); + b.setValueCount(1); + + assertNull(ArrowValues.toJavaValue(sv, 0)); + } + } + + /** An empty sub-object is dropped by its parent rather than appearing as {} or null. */ + public void testEmptySubObjectIsOmittedFromParent() { + try (StructVector parent = StructVector.empty("parent", allocator)) { + VarCharVector kept = parent.addOrGet("kept", FieldType.nullable(new ArrowType.Utf8()), VarCharVector.class); + StructVector child = parent.addOrGetStruct("child"); + VarCharVector grandchild = child.addOrGet("g", FieldType.nullable(new ArrowType.Utf8()), VarCharVector.class); + parent.allocateNew(); + kept.setSafe(0, "x".getBytes(java.nio.charset.StandardCharsets.UTF_8)); + grandchild.setNull(0); + child.setIndexDefined(0); + parent.setIndexDefined(0); + parent.setValueCount(1); + kept.setValueCount(1); + child.setValueCount(1); + grandchild.setValueCount(1); + + assertEquals(Map.of("kept", "x"), ArrowValues.toJavaValue(parent, 0)); + } + } + + /** + * A nested timestamp is formatted the same as a top-level one. Reading child values out of + * {@code getObject} would return the raw epoch instead, because the values arrive stripped of + * the Field that says how to format them. + */ + public void testNestedTimestampIsFormattedLikeTopLevel() { + try (StructVector sv = StructVector.empty("obj", allocator)) { + TimeStampMilliVector ts = sv.addOrGet( + "at", + FieldType.nullable(new ArrowType.Timestamp(TimeUnit.MILLISECOND, null)), + TimeStampMilliVector.class + ); + sv.allocateNew(); + ts.setSafe(0, 1_700_000_000_000L); + sv.setIndexDefined(0); + sv.setValueCount(1); + ts.setValueCount(1); + + @SuppressWarnings("unchecked") + Map out = (Map) ArrowValues.toJavaValue(sv, 0); + + try (TimeStampMilliVector standalone = new TimeStampMilliVector("at", allocator)) { + standalone.allocateNew(1); + standalone.setSafe(0, 1_700_000_000_000L); + standalone.setValueCount(1); + assertEquals(ArrowValues.toJavaValue(standalone, 0), out.get("at")); + } + } + } } diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/BasePlannerRulesTests.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/BasePlannerRulesTests.java index 1eb99a6e97403..c0b17a631e6cb 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/BasePlannerRulesTests.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/BasePlannerRulesTests.java @@ -253,7 +253,7 @@ protected RelOptTable mockNullableTable(String tableName, String... fieldNames) return mockTable(tableName, rowTypeBuilder.build()); } - private RelOptTable mockTable(String tableName, RelDataType rowType) { + protected RelOptTable mockTable(String tableName, RelDataType rowType) { RelOptTable table = mock(RelOptTable.class); when(table.getQualifiedName()).thenReturn(List.of(tableName)); when(table.getRowType()).thenReturn(rowType); diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockDataFusionBackend.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockDataFusionBackend.java index 52ca31f44d8c8..ff5e1faa3e33e 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockDataFusionBackend.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockDataFusionBackend.java @@ -53,6 +53,9 @@ public class MockDataFusionBackend extends MockBackend implements SearchBackEndP SUPPORTED_TYPES.addAll(FieldType.keyword()); SUPPORTED_TYPES.addAll(FieldType.date()); SUPPORTED_TYPES.add(FieldType.BOOLEAN); + // Calcite ROW maps to FieldType.OBJECT — needed for make_struct's return type to + // resolve a capable backend in OpenSearchProjectRule. + SUPPORTED_TYPES.add(FieldType.OBJECT); } private static final Set STANDARD_OPS = Set.of( @@ -206,7 +209,9 @@ protected Set scanCapabilities() { ScalarFunction.CONCAT, ScalarFunction.UPPER, ScalarFunction.SIN, - ScalarFunction.ABS + ScalarFunction.ABS, + // Object materialization: make_struct over the flat dotted leaves (ObjectStructMaterializer). + ScalarFunction.MAKE_STRUCT ); private static final Set PROJECT_CAPS; diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/ObjectStructPlanShapeTests.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/ObjectStructPlanShapeTests.java new file mode 100644 index 0000000000000..d71b492618f84 --- /dev/null +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/ObjectStructPlanShapeTests.java @@ -0,0 +1,217 @@ +/* + * 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.planner; + +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.plan.RelOptUtil; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.AggregateCall; +import org.apache.calcite.rel.logical.LogicalProject; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.sql.fun.SqlStdOperatorTable; +import org.apache.calcite.sql.type.SqlTypeName; +import org.apache.calcite.util.ImmutableBitSet; + +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** + * End-to-end plan-shape coverage for {@link ObjectStructMaterializer}: an OpenSearch + * {@code object} field is re-assembled from its flat dotted leaf columns into a + * {@code make_struct} call in a project directly above the scan. + * + *

The scan carries only leaf columns — an object parent has no physical storage + * ({@code FieldStorageResolver.populateFromProperties} recurses past object parents), so the + * struct is appended by the project, never read from the scan: + * + * 

+ * id                               INTEGER
+ * nested_metadata.top              VARCHAR
+ * nested_metadata.properties.name  VARCHAR
+ * nested_metadata.properties.value VARCHAR
+ *
+ */ +public class ObjectStructPlanShapeTests extends PlanShapeTestBase { + + /** + * Table as the schema builder now produces it for an {@code object} mapping: flat dotted leaf + * columns PLUS the struct-typed parent column. + */ + private RelOptTable objectTable() { + RelDataType varchar = typeFactory.createSqlType(SqlTypeName.VARCHAR); + RelDataType properties = typeFactory.createStructType(List.of(varchar, varchar), List.of("name", "value")); + RelDataType meta = typeFactory.createStructType(List.of(varchar, properties), List.of("top", "properties")); + + RelDataTypeFactory.Builder builder = typeFactory.builder(); + builder.add("id", typeFactory.createSqlType(SqlTypeName.INTEGER)); + builder.add("nested_metadata.top", varchar); + builder.add("nested_metadata.properties.name", varchar); + builder.add("nested_metadata.properties.value", varchar); + builder.add("nested_metadata", meta); + return mockTable("test_index", builder.build()); + } + + /** Field mappings for the leaf columns — the object parent is intentionally absent. */ + private Map> leafFieldMappings() { + return Map.of( + "id", + Map.of("type", "integer"), + "nested_metadata.top", + Map.of("type", "keyword"), + "nested_metadata.properties.name", + Map.of("type", "keyword"), + "nested_metadata.properties.value", + Map.of("type", "keyword") + ); + } + + /** + * The core rewrite: a project above the scan passes the leaves through and appends the + * object, nesting a second {@code make_struct} for the {@code properties} sub-object. + */ + public void testMaterializerAppendsNestedStructProjectAboveScan() { + RelNode scan = stubScan(objectTable()); + + Optional rewritten = ObjectStructMaterializer.rewrite(scan); + + assertTrue("materializer should fire when an object's leaves are present", rewritten.isPresent()); + assertPlanShape( + """ + LogicalProject(id=[$0], nested_metadata.top=[$1], nested_metadata.properties.name=[$2], nested_metadata.properties.value=[$3], nested_metadata=[make_struct('top':VARCHAR, $1, 'properties':VARCHAR, make_struct('name':VARCHAR, $2, 'value':VARCHAR, $3))]) + LogicalTableScan(table=[[test_index]]) + """, + rewritten.get() + ); + } + + /** No object spec ⇒ no rewrite, so plans without objects are untouched. */ + public void testMaterializerNoOpWithoutObjectSpec() { + RelNode scan = stubScan(mockTable("test_index", "status", "size")); + assertFalse(ObjectStructMaterializer.rewrite(scan).isPresent()); + } + + /** + * A struct column whose backing leaves are absent from the scan (e.g. an unsupported sub-field + * type was dropped from the schema) yields a typed NULL, never a partially-filled struct. + */ + public void testMaterializerEmitsNullForObjectWithMissingLeaf() { + RelDataType varchar = typeFactory.createSqlType(SqlTypeName.VARCHAR); + RelDataType meta = typeFactory.createStructType(List.of(varchar), List.of("top")); + RelDataTypeFactory.Builder builder = typeFactory.builder(); + builder.add("id", typeFactory.createSqlType(SqlTypeName.INTEGER)); + // NOTE: no "nested_metadata.top" leaf column — the struct cannot be assembled. + builder.add("nested_metadata", meta); + RelNode scan = stubScan(mockTable("test_index", builder.build())); + + RelNode rewritten = ObjectStructMaterializer.rewrite(scan).orElseThrow(); + + String shape = RelOptUtil.toString(rewritten); + assertFalse("no partial struct should be emitted, got:\n" + shape, shape.contains("make_struct")); + assertTrue("expected a typed NULL for the unassemblable object, got:\n" + shape, shape.contains("null:RecordType")); + } + + /** + * Projecting the object returns the whole object: the materialized struct survives the full + * planner (marking, CBO) and sits above the scan. + */ + public void testProjectOnObjectReturnsWholeObjectThroughPlanner() { + RelNode scan = stubScan(objectTable()); + RelNode materialized = ObjectStructMaterializer.rewrite(scan).orElseThrow(); + int objectIndex = materialized.getRowType().getFieldCount() - 1; + RelNode plan = LogicalProject.create( + materialized, + List.of(), + List.of(rexBuilder.makeInputRef(materialized, 0), rexBuilder.makeInputRef(materialized, objectIndex)), + List.of("id", "nested_metadata") + ); + + RelNode result = runPlanner(plan, buildContext("parquet", 1, leafFieldMappings())); + + String shape = RelOptUtil.toString(result); + assertTrue("expected a materialized struct, got:\n" + shape, shape.contains("make_struct")); + assertTrue( + "sub-object must nest a second make_struct, got:\n" + shape, + shape.indexOf("make_struct") != shape.lastIndexOf("make_struct") + ); + assertTrue("struct assembly must sit above the scan, got:\n" + shape, shape.indexOf("make_struct") < shape.indexOf("TableScan")); + } + + /** + * {@code stats count() by nested_metadata} — the struct is materialized in a project below the + * aggregate, so the aggregate receives an already-assembled object, not raw leaves. + * + *

Grouping (rather than {@code count(nested_metadata)}) is the meaningful probe: counting a + * non-nullable column is equivalent to {@code count(*)}, so Calcite drops the column reference + * and trims the project away — correctly, but it proves nothing about materialization. A group + * key genuinely needs the struct's value. + */ + public void testAggregateOnObjectMaterializesStructBeforeAgg() { + RelNode scan = stubScan(objectTable()); + RelNode materialized = ObjectStructMaterializer.rewrite(scan).orElseThrow(); + int objectIndex = materialized.getRowType().getFieldCount() - 1; + AggregateCall count = AggregateCall.create( + SqlStdOperatorTable.COUNT, + false, + List.of(), + -1, + materialized, + typeFactory.createSqlType(SqlTypeName.BIGINT), + "cnt" + ); + RelNode plan = makeAggregate(materialized, ImmutableBitSet.of(objectIndex), count); + + RelNode result = runPlanner(plan, buildContext("parquet", 1, leafFieldMappings())); + + String shape = RelOptUtil.toString(result); + assertTrue("expected a materialized struct, got:\n" + shape, shape.contains("make_struct")); + int aggAt = shape.indexOf("Aggregate"); + int structAt = shape.indexOf("make_struct"); + assertTrue("aggregate must sit above the struct-materializing project, got:\n" + shape, aggAt >= 0 && aggAt < structAt); + assertTrue("struct assembly must sit above the scan, got:\n" + shape, structAt < shape.indexOf("TableScan")); + } + + /** + * Multi-shard: grouping on the materialized object still splits into PARTIAL/FINAL. The group set + * is {@code {0}} over a single-column input, so it satisfies + * {@code OpenSearchAggregateSplitRule.shouldSkipPartialFinalSplit}'s prefix check and the + * aggregate distributes normally — the struct as a group key costs throughput (DataFusion has no + * {@code Struct} specialization in its columnar group-values path, so it row-encodes) but does + * not cost distribution. Pinned because the reduce path is unreachable at one shard. + */ + public void testAggregateOnObject_2shard() { + RelNode scan = stubScan(objectTable()); + RelNode materialized = ObjectStructMaterializer.rewrite(scan).orElseThrow(); + int objectIndex = materialized.getRowType().getFieldCount() - 1; + AggregateCall count = AggregateCall.create( + SqlStdOperatorTable.COUNT, + false, + List.of(), + -1, + materialized, + typeFactory.createSqlType(SqlTypeName.BIGINT), + "cnt" + ); + RelNode plan = makeAggregate(materialized, ImmutableBitSet.of(objectIndex), count); + + RelNode result = runPlanner(plan, buildContext("parquet", 2, leafFieldMappings())); + + assertPlanShape( + """ + OpenSearchAggregate(group=[{0}], cnt=[SUM($1)], mode=[FINAL], viableBackends=[[mock-parquet]]) + OpenSearchExchangeReducer(viableBackends=[[mock-parquet]], exchange=[ExchangeInfo[distributionType=SINGLETON, partitionKeyIndices=[], partitionCount=0]]) + OpenSearchAggregate(group=[{0}], cnt=[COUNT()], mode=[PARTIAL], viableBackends=[[mock-parquet]]) + OpenSearchProject(nested_metadata=[ANNOTATED_PROJECT_EXPR(id=1, backends=[mock-parquet], make_struct('top':VARCHAR, $1, 'properties':VARCHAR, ANNOTATED_PROJECT_EXPR(id=0, backends=[mock-parquet], make_struct('name':VARCHAR, $2, 'value':VARCHAR, $3))))], viableBackends=[[mock-parquet]]) + OpenSearchTableScan(table=[[test_index]], viableBackends=[[mock-parquet]]) + """, + result + ); + } +} diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java index dd8bf0800b53a..8dd4999531896 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java @@ -416,7 +416,19 @@ private static long safeTotalRamBytes() { /** Set of ArrowType classes that do NOT support dictionary encoding. */ private static final Set> DICTIONARY_INCOMPATIBLE_TYPES = Set.of(ArrowType.Bool.class); - /** Maps arrow type name strings to representative ArrowType instances for compatibility checks. */ + /** + * Maps arrow type name strings to representative ArrowType instances for compatibility checks. + * + *

TODO: add {@code Struct} so OpenSearch {@code object} fields can be written as native + * Parquet structs instead of flat dotted leaf columns. A struct yields the same leaf column + * chunks we already write, so pruning and encoding are unchanged — but the reader could then + * hand the object back assembled, removing the need to reassemble it at query time (see + * {@code ObjectStructMaterializer}, which documents the full search-side impact). Also needs the + * merge path to recurse into struct children: {@code build_parquet_root_schema} unions segment + * schemas by top-level name only, and {@code ColumnMapping} null-fills a whole missing column but + * never a missing struct child, so a struct gaining a sub-field between segments fails the merge. + * {@code List} is the same gap for {@code nested}. + */ public static final Map ARROW_TYPE_NAME_TO_INSTANCE = Map.ofEntries( Map.entry("int8", new ArrowType.Int(8, true)), Map.entry("int16", new ArrowType.Int(16, true)), diff --git a/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/schema.rs b/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/schema.rs index 73e572f9b951f..cfb2fe7e1f729 100644 --- a/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/schema.rs +++ b/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/schema.rs @@ -21,6 +21,13 @@ pub const ROW_ID_COLUMN_NAME: &str = "__row_id__"; /// Builds the output Parquet schema as the union of pre-read schema descriptors. /// +/// TODO: recurse into struct children. The union is by TOP-LEVEL name with first-writer-wins, so a +/// nested group's subtree is taken wholesale from whichever segment is seen first. If a struct gains +/// a field in a later segment, the merged schema keeps the older, narrower type and the merge then +/// fails in `ColumnMapping::pad_batch` at `RecordBatch::try_new` on a type mismatch. This is the +/// blocker for storing OpenSearch `object` fields as native structs (and `nested` as +/// `LIST>`); dynamic mapping makes a struct gaining a field routine. +/// /// The output schema contains every column seen across all inputs, except: /// - Any existing `__row_id__` column is removed. /// - A fresh `__row_id__` INT64 REQUIRED column is appended at the end. @@ -121,6 +128,9 @@ impl ColumnMapping { } /// Remap a batch using the precomputed mapping. Zero-copy when schemas match. + /// + /// TODO: null-fill a missing struct *child*, not just a whole missing column. Paired with the + /// recursive union in `build_parquet_root_schema`, this is what native struct storage needs. #[inline] pub fn pad_batch(&self, batch: &RecordBatch) -> MergeResult { if self.is_identity { diff --git a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ApmServiceMapObjectIT.java b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ApmServiceMapObjectIT.java new file mode 100644 index 0000000000000..104652767be96 --- /dev/null +++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ApmServiceMapObjectIT.java @@ -0,0 +1,224 @@ +/* + * 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 java.io.IOException; +import java.util.List; +import java.util.Map; + +/** + * Regression contract for the dashboards-observability APM service-map experience: the eight PPL + * queries in {@code apm/query_services/query_requests/ppl_queries.ts}. + * + *

Six project a parent object ({@code sourceNode.keyAttributes}, …) and used to fail with + * {@code Field [sourceNode.keyAttributes] not found}, blanking topology, service detail, and + * dependency lists — only the two {@code distinct_count} widgets on leaf scalars worked. + * {@link #testOperatorMatrixOnParentObject} additionally covers {@code fields} / {@code eval} / + * {@code dedup} / {@code sort} / {@code isnotnull}, since the bug was broader than projection. + * + *

Fixture topology: frontend→checkout, frontend→payment, checkout→inventory. + */ +public class ApmServiceMapObjectIT extends AnalyticsRestTestCase { + + private static final Dataset DATASET = new Dataset("apm_service_map", "otel-apm-service-map"); + + /** Back-quoted because the index name contains hyphens, exactly as the dashboards plugin sends it. */ + private static final String SOURCE = "source = `otel-apm-service-map`"; + + /** Time clause matching {@code buildTimeFilterClause}'s 'YYYY-MM-DD HH:mm:ss.SSS' rendering. */ + private static final String TIME = " | where timestamp >= '2026-08-26 00:00:00.000'" + + " and timestamp <= '2026-08-27 00:00:00.000'"; + + private static boolean dataProvisioned = false; + + @Override + protected void onBeforeQuery() throws IOException { + if (dataProvisioned == false) { + DatasetProvisioner.provision(client(), DATASET); + dataProvisioned = true; + } + } + + // ── The six queries that projected parent objects (previously all HTTP 500) ──────── + + /** {@code getQueryListServices} — projects all four parent objects at once. */ + public void testListServices() throws IOException { + List> rows = rows( + SOURCE + + TIME + + " | dedup nodeConnectionHash" + + " | fields sourceNode.keyAttributes, sourceNode.groupByAttributes," + + " targetNode.keyAttributes, targetNode.groupByAttributes" + ); + assertEquals("one row per connection", 3, rows.size()); + // Every projected cell must be a materialized object, not a scalar or null. + for (List row : rows) { + assertEquals(4, row.size()); + for (Object cell : row) { + assertTrue("expected an object, got: " + cell, cell instanceof Map); + } + } + } + + /** {@code getQueryGetService} — parent objects plus the two leaf equality filters. */ + public void testGetService() throws IOException { + List> rows = rows( + SOURCE + + TIME + + " | where sourceNode.keyAttributes.environment = 'prod'" + + " | where sourceNode.keyAttributes.name = 'frontend'" + + " | dedup nodeConnectionHash" + + " | fields sourceNode.keyAttributes, sourceNode.groupByAttributes" + ); + // Two frontend-sourced connections (h1, h2) survive dedup on the hash. + assertEquals(2, rows.size()); + assertEquals("frontend", keyAttribute(rows.get(0).get(0), "name")); + assertEquals("js", keyAttribute(rows.get(0).get(1), "telemetry_sdk_language")); + } + + /** {@code getQueryServiceAttributes} — parent objects with {@code sort - timestamp | head 1}. */ + public void testServiceAttributes() throws IOException { + List> rows = rows( + SOURCE + + TIME + + " | where sourceNode.keyAttributes.environment = 'prod'" + + " | where sourceNode.keyAttributes.name = 'frontend'" + + " | fields sourceNode.keyAttributes, sourceNode.groupByAttributes, timestamp" + + " | sort - timestamp" + + " | head 1" + ); + assertEquals(1, rows.size()); + assertEquals("frontend", keyAttribute(rows.get(0).get(0), "name")); + // Descending sort picks the later of the two frontend rows (10:05, not 10:00). + assertTrue("expected the newest row, got: " + rows.get(0).get(2), rows.get(0).get(2).toString().contains("10:05")); + } + + /** + * {@code getQueryListServiceOperations} — mixes parent objects with leaf scalars in one + * projection. {@code getQueryListServiceDependencies} builds a byte-for-byte identical + * pipeline, so this covers both. + */ + public void testListServiceOperationsAndDependencies() throws IOException { + List> rows = rows( + SOURCE + + TIME + + " | dedup operationConnectionHash" + + " | fields sourceNode.keyAttributes, sourceOperation.name," + + " targetNode.keyAttributes, targetOperation.name" + ); + assertEquals(3, rows.size()); + for (List row : rows) { + assertTrue("col0 must be an object", row.get(0) instanceof Map); + assertTrue("col1 must be a scalar operation name", row.get(1) instanceof String); + assertTrue("col2 must be an object", row.get(2) instanceof Map); + assertTrue("col3 must be a scalar operation name", row.get(3) instanceof String); + } + } + + /** {@code getQueryGetServiceMap} — the topology query; four parent objects, interleaved order. */ + public void testGetServiceMap() throws IOException { + List> rows = rows( + SOURCE + + TIME + + " | dedup nodeConnectionHash" + + " | fields sourceNode.keyAttributes, targetNode.keyAttributes," + + " sourceNode.groupByAttributes, targetNode.groupByAttributes" + ); + assertEquals(3, rows.size()); + for (List row : rows) { + assertEquals(4, row.size()); + for (Object cell : row) { + assertTrue("expected an object, got: " + cell, cell instanceof Map); + } + } + } + + // ── The two count widgets that already worked — guard against regression ────────── + + /** {@code getQueryOperationDependenciesCount} — frontend's 'GET /cart' fans out to 2 services. */ + public void testOperationDependenciesCount() throws IOException { + List> rows = rows( + SOURCE + + TIME + + " | where sourceNode.keyAttributes.environment = 'prod'" + + " | where sourceNode.keyAttributes.name = 'frontend'" + + " | where sourceOperation.name = 'GET /cart'" + + " | stats distinct_count(targetNode.keyAttributes.name) as dependency_count" + ); + assertEquals(1, rows.size()); + assertEquals(2, ((Number) rows.get(0).get(0)).intValue()); + } + + /** {@code getQueryDependencyDownstreamCount} — checkout has 1 downstream dependency. */ + public void testDependencyDownstreamCount() throws IOException { + List> rows = rows( + SOURCE + + TIME + + " | where sourceNode.keyAttributes.environment = 'prod'" + + " | where sourceNode.keyAttributes.name = 'checkout'" + + " | stats distinct_count(targetNode.keyAttributes.name) as dependency_count" + ); + assertEquals(1, rows.size()); + assertEquals(1, ((Number) rows.get(0).get(0)).intValue()); + } + + // ── Operator matrix from the bug report ────────────────────────────────────────── + + /** + * Every operator that takes a parent-object identifier as an argument. The original report + * called out that the failure was not projection-specific: {@code eval}, {@code dedup}, + * {@code sort} and {@code isnotnull} threw the same "Field [...] not found" because the + * object was absent from the row type. One schema change fixes all of them, so all are + * pinned here. + */ + public void testOperatorMatrixOnParentObject() throws IOException { + // fields on a top-level object + assertFalse(rows(SOURCE + " | fields sourceNode | head 1").isEmpty()); + // fields on a one-level-deep object + assertFalse(rows(SOURCE + " | fields sourceNode.keyAttributes | head 1").isEmpty()); + // eval with an object on the RHS + assertFalse(rows(SOURCE + " | eval ka = sourceNode.keyAttributes | fields ka | head 1").isEmpty()); + // dedup keyed on an object + assertFalse(rows(SOURCE + " | dedup sourceNode.keyAttributes | head 1").isEmpty()); + // sort keyed on an object + assertFalse(rows(SOURCE + " | sort sourceNode.keyAttributes | head 1").isEmpty()); + // null-test on an object + assertFalse(rows(SOURCE + " | where isnotnull(sourceNode.keyAttributes) | fields nodeConnectionHash | head 1").isEmpty()); + } + + /** Control rows from the matrix: scalar-leaf filter and leaf-only aggregation still work. */ + public void testOperatorMatrixLeafControls() throws IOException { + List> filtered = rows( + SOURCE + " | where sourceNode.keyAttributes.name = 'frontend' | fields nodeConnectionHash" + ); + assertEquals(2, filtered.size()); + + List> aggregated = rows( + SOURCE + " | stats count() by sourceNode.keyAttributes.name" + ); + assertEquals("two distinct source services", 2, aggregated.size()); + } + + // ── helpers ────────────────────────────────────────────────────────────────────── + + @SuppressWarnings("unchecked") + private List> rows(String ppl) throws IOException { + Map response = executePpl(ppl); + List> rows = (List>) response.get("datarows"); + assertNotNull("Response missing 'datarows' for query: " + ppl, rows); + return rows; + } + + /** Reads a sub-field out of a materialized object cell. */ + @SuppressWarnings("unchecked") + private static Object keyAttribute(Object objectCell, String field) { + assertTrue("expected an object cell, got: " + objectCell, objectCell instanceof Map); + return ((Map) objectCell).get(field); + } +} diff --git a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ObjectFieldIT.java b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ObjectFieldIT.java index 0aa3bad9318e4..9c2e1ed7fd921 100644 --- a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ObjectFieldIT.java +++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ObjectFieldIT.java @@ -17,12 +17,9 @@ import java.util.Map; /** - * Diagnostic integration tests for PPL access to OpenSearch {@code object} fields - * via dotted-path notation ({@code city.name}, {@code city.location.latitude}) on the - * analytics-engine route. Mirrors the shape of the sql repo's - * {@code ObjectFieldOperateIT}. Every test here is expected to fail initially — - * the purpose is to surface exact failure modes for follow-up debugging, not to - * exercise a working implementation. + * PPL access to OpenSearch {@code object} fields — leaves via dotted paths + * ({@code city.location.latitude}), whole objects, and objects as group keys. Mirrors the sql repo's + * {@code ObjectFieldOperateIT}. */ public class ObjectFieldIT extends AnalyticsRestTestCase { @@ -102,18 +99,14 @@ public void testFilterOnDeeplyNestedObjectField() throws IOException { ); } - // ── Object-parent projection (gated on query-then-fetch) ────────────────── + // ── Object-parent projection ─────────────────────────────────────────────── // // Projecting an object parent (top-level "city" or intermediate "city.location") - // returns a nested JSON value reconstructed from _source. Analytics-engine emits - // only flat leaves into the Calcite row type today, so parent references fall - // through QualifiedNameResolver and throw "Field [city.location] not found". - // - // Support requires query-then-fetch (QTF): coordinator returns docIds post-filter, - // a fetch stage pulls the doc from the shard, and the parent sub-object is - // reconstructed from _source or from parquet rows. QTF is tracked separately. + // returns the nested object. No query-then-fetch / _source read is needed: the + // schema exposes the object as a struct (ROW) column and ObjectStructMaterializer + // re-assembles it with make_struct over the flat leaf columns the scan already + // produces, in a project directly above the scan. - @AwaitsFix(bugUrl = "Object parent projection requires query-then-fetch (QTF) for source-based materialization") public void testSelectIntermediateObjectField() throws IOException { assertRowsEqual( "source=" + DATASET.indexName + " | fields city.location | head 1", @@ -121,7 +114,6 @@ public void testSelectIntermediateObjectField() throws IOException { ); } - @AwaitsFix(bugUrl = "Object parent projection requires query-then-fetch (QTF) for source-based materialization") public void testSelectTopLevelObjectField() throws IOException { assertRowsEqual( "source=" + DATASET.indexName + " | fields city | head 1", @@ -129,7 +121,6 @@ public void testSelectTopLevelObjectField() throws IOException { ); } - @AwaitsFix(bugUrl = "Object parent projection requires query-then-fetch (QTF) for source-based materialization") public void testSelectTopLevelObjectFieldWithSiblings() throws IOException { assertRowsEqual( "source=" + DATASET.indexName + " | fields city, account | head 1", @@ -140,7 +131,6 @@ public void testSelectTopLevelObjectFieldWithSiblings() throws IOException { ); } - @AwaitsFix(bugUrl = "Object parent projection requires query-then-fetch (QTF) for source-based materialization") public void testSelectParentAndLeafMixed() throws IOException { assertRowsEqual( "source=" + DATASET.indexName + " | fields city.name, city.location | head 1", @@ -148,8 +138,38 @@ public void testSelectParentAndLeafMixed() throws IOException { ); } + // ── Aggregation involving object fields ─────────────────────────────────── + // + // Leaf aggregations (min/max/sum on city.population, city.location.latitude, …) are covered + // above. These cover aggregating on the OBJECT VALUE itself — the group key is a struct + // materialized by ObjectStructMaterializer, so the aggregate receives an assembled object. + + /** Group by an intermediate object ({@code city.location}) — 3 distinct locations. */ + public void testGroupByIntermediateObjectField() throws IOException { + assertRowCount("source=" + DATASET.indexName + " | stats count() by city.location", 3); + } + + /** Group by a top-level object ({@code city}) — 3 distinct cities. */ + public void testGroupByTopLevelObjectField() throws IOException { + assertRowCount("source=" + DATASET.indexName + " | stats count() by city", 3); + } + + /** Aggregate a leaf while grouping by an object value. */ + public void testAggregateLeafGroupedByObjectField() throws IOException { + assertRowCount("source=" + DATASET.indexName + " | stats max(city.population) by city.location", 3); + } + // ── helpers (mirrored from FieldsCommandIT) ──────────────────────────────── + /** Asserts only the row count — group order is not deterministic for a struct key. */ + private void assertRowCount(String ppl, int expected) throws IOException { + Map response = executePpl(ppl); + @SuppressWarnings("unchecked") + List> actualRows = (List>) response.get("datarows"); + assertNotNull("Response missing 'datarows' for query: " + ppl, actualRows); + assertEquals("Row count mismatch for query: " + ppl, expected, actualRows.size()); + } + private static List row(Object... values) { return Arrays.asList(values); } @@ -173,4 +193,97 @@ private final void assertRowsEqual(String ppl, List... expected) throws } + + // ── select * ────────────────────────────────────────────────────────────────────── + // + // Nothing here names an object, so coverage depends entirely on how `*` expands. Verified + // against the legacy engine on the same mapping: three top-level fields, objects as nested + // JSON. The flat dotted leaves must NOT appear — an object's data is returned once, not twice. + + /** {@code source=idx} with no field list: objects come back as whole nested values. */ + public void testSelectStarReturnsObjectsAsNestedStructs() throws IOException { + Map response = executePpl("source=" + DATASET.indexName + " | head 1"); + assertStarShape(response, "source=... | head 1"); + } + + /** Explicit {@code fields *} must behave identically to the implicit form above. */ + public void testFieldsStarReturnsObjectsAsNestedStructs() throws IOException { + Map response = executePpl("source=" + DATASET.indexName + " | fields * | head 1"); + assertStarShape(response, "source=... | fields * | head 1"); + } + + /** + * Asserts the star-expansion contract: exactly the top-level fields (no dotted leaves), with + * each object materialized as a nested map. Column order is not asserted — it is not part of + * the contract and differs from legacy — so the row is checked by column name. + */ + private void assertStarShape(Map response, String context) { + List columns = extractColumnNames(response); + assertEquals( + "star expansion must yield only top-level fields (no dotted leaves) for " + context, + List.of("account", "city", "id"), + columns.stream().sorted().toList() + ); + + @SuppressWarnings("unchecked") + List> rows = (List>) response.get("datarows"); + assertNotNull("missing datarows for " + context, rows); + assertEquals("expected a single row for " + context, 1, rows.size()); + Map row = new java.util.HashMap<>(); + for (int i = 0; i < columns.size(); i++) { + row.put(columns.get(i), rows.get(0).get(i)); + } + + assertEquals("id for " + context, "1", row.get("id")); + assertEquals( + "account must be a whole nested object for " + context, + Map.of("owner", "alice", "balance", 1000.5), + row.get("account") + ); + // Nested sub-object arrives nested, not flattened to a dotted key. + assertEquals( + "city must nest location for " + context, + Map.of( + "name", + "Seattle", + "population", + 750000, + "location", + Map.of("latitude", 47.6062, "longitude", -122.3321) + ), + row.get("city") + ); + } + + /** + * A shapeless {@code {"type": "object"}} — no {@code properties}, which is what dynamic mapping + * leaves before any document populates it — is addressable and resolves to null, as vanilla does. + * The schema gives it a field-less ROW, so this is also the end-to-end check that such a type + * survives Substrait serialization and DataFusion rather than only the schema builder. + */ + public void testShapelessObjectResolvesToNull() throws IOException { + String index = "shapeless_object_it"; + try { + client().performRequest(new Request("DELETE", "/" + index)); + } catch (Exception ignored) {} + Request create = new Request("PUT", "/" + index); + create.setJsonEntity( + "{\"settings\":{\"index.pluggable.dataformat.enabled\":true," + + "\"index.pluggable.dataformat\":\"composite\"," + + "\"index.composite.primary_data_format\":\"parquet\"," + + "\"index.composite.secondary_data_formats\":[\"lucene\"]," + + "\"number_of_shards\":1,\"number_of_replicas\":0}," + + "\"mappings\":{\"properties\":{\"id\":{\"type\":\"keyword\"}," + + "\"attrs\":{\"type\":\"object\"}}}}" + ); + client().performRequest(create); + // No custom _id: parquet indices are append-only and reject one. + Request doc = new Request("POST", "/" + index + "/_bulk?refresh=true"); + doc.setJsonEntity("{\"index\":{}}\n{\"id\":\"1\"}\n"); + doc.setOptions(doc.getOptions().toBuilder().addHeader("Content-Type", "application/x-ndjson")); + client().performRequest(doc); + + assertRowsEqual("source=" + index + " | fields attrs", row((Object) null)); + assertRowsEqual("source=" + index + " | fields id, attrs", row("1", null)); + } } diff --git a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ObjectFieldMultiShardIT.java b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ObjectFieldMultiShardIT.java new file mode 100644 index 0000000000000..aab22330be2ce --- /dev/null +++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/ObjectFieldMultiShardIT.java @@ -0,0 +1,94 @@ +/* + * 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 java.io.IOException; +import java.util.List; +import java.util.Map; + +/** + * Multi-shard coverage for grouping on an {@code object}. {@link ObjectFieldIT} exercises the same + * queries at 1 shard, which cannot reach the PARTIAL/FINAL reduce path. + * + *

Grouping on a struct value goes through DataFusion's generic row-encoded group column + * ({@code RowsGroupColumn}) rather than a per-type columnar builder, since {@code Struct} has no + * specialization yet. That is a throughput question, not a correctness one — but the PARTIAL/FINAL + * reduce it runs through is untested at 1 shard, so these pin the results it produces. + * + *

Reuses {@code object_fields}' mapping and bulk data under a distinct index name so the 1-shard + * index in {@link ObjectFieldIT} is untouched. + */ +public class ObjectFieldMultiShardIT extends AnalyticsRestTestCase { + + private static final Dataset DATASET = new Dataset("object_fields", "object_fields_multishard"); + + /** Seattle, Portland, Austin — one group per city in the dataset. */ + private static final int DISTINCT_CITIES = 3; + + private static boolean provisioned = false; + + @Override + protected void onBeforeQuery() throws IOException { + if (provisioned == false) { + DatasetProvisioner.provision(client(), DATASET, 2); + provisioned = true; + } + } + + /** + * Grouping on the whole object across shards: one group per city, every group key returned as a + * nested object (not flattened), and the per-shard counts reduced to the full document count. + */ + @SuppressWarnings("unchecked") + public void testGroupByTopLevelObjectFieldAtTwoShards() throws IOException { + List> rows = rowsOf("source=" + DATASET.indexName + " | stats count() by city"); + assertEquals("one group per city", DISTINCT_CITIES, rows.size()); + + long total = 0; + for (List row : rows) { + // stats output is [count, groupKey]; locate the map rather than assuming a position. + Map city = null; + for (Object cell : row) { + if (cell instanceof Map map) { + city = (Map) map; + } else if (cell instanceof Number n) { + total += n.longValue(); + } + } + assertNotNull("group key must be a nested object, got row: " + row, city); + assertTrue("object must carry its leaves, got: " + city, city.containsKey("name") && city.containsKey("population")); + assertTrue("sub-object must stay nested, got: " + city, city.get("location") instanceof Map); + } + assertEquals("reduced counts must cover every document", DISTINCT_CITIES, total); + } + + /** An intermediate object as the group key — the nested make_struct path, across shards. */ + public void testGroupByIntermediateObjectFieldAtTwoShards() throws IOException { + assertEquals( + DISTINCT_CITIES, + rowsOf("source=" + DATASET.indexName + " | stats count() by city.location").size() + ); + } + + /** A leaf aggregate grouped by an object: agg call and struct group key reduced together. */ + public void testAggregateLeafGroupedByObjectFieldAtTwoShards() throws IOException { + assertEquals( + DISTINCT_CITIES, + rowsOf("source=" + DATASET.indexName + " | stats max(city.population) by city.location").size() + ); + } + + @SuppressWarnings("unchecked") + private List> rowsOf(String ppl) throws IOException { + Map response = executePpl(ppl); + List> rows = (List>) response.get("datarows"); + assertNotNull("Response missing 'datarows' for query: " + ppl, rows); + return rows; + } +} diff --git a/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/apm_service_map/bulk.json b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/apm_service_map/bulk.json new file mode 100644 index 0000000000000..1a4b90b7f6620 --- /dev/null +++ b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/apm_service_map/bulk.json @@ -0,0 +1,6 @@ +{"index": {}} +{"timestamp": "2026-08-26T10:00:00.000Z", "nodeConnectionHash": "h1", "operationConnectionHash": "o1", "sourceNode": {"keyAttributes": {"name": "frontend", "environment": "prod", "type": "Service"}, "groupByAttributes": {"telemetry_sdk_language": "js", "deployment_env": "prod"}}, "targetNode": {"keyAttributes": {"name": "checkout", "environment": "prod", "type": "Service"}, "groupByAttributes": {"telemetry_sdk_language": "java", "deployment_env": "prod"}}, "sourceOperation": {"name": "GET /cart"}, "targetOperation": {"name": "POST /checkout"}} +{"index": {}} +{"timestamp": "2026-08-26T10:05:00.000Z", "nodeConnectionHash": "h2", "operationConnectionHash": "o2", "sourceNode": {"keyAttributes": {"name": "frontend", "environment": "prod", "type": "Service"}, "groupByAttributes": {"telemetry_sdk_language": "js", "deployment_env": "prod"}}, "targetNode": {"keyAttributes": {"name": "payment", "environment": "prod", "type": "Service"}, "groupByAttributes": {"telemetry_sdk_language": "go", "deployment_env": "prod"}}, "sourceOperation": {"name": "GET /cart"}, "targetOperation": {"name": "POST /pay"}} +{"index": {}} +{"timestamp": "2026-08-26T10:10:00.000Z", "nodeConnectionHash": "h3", "operationConnectionHash": "o3", "sourceNode": {"keyAttributes": {"name": "checkout", "environment": "prod", "type": "Service"}, "groupByAttributes": {"telemetry_sdk_language": "java", "deployment_env": "prod"}}, "targetNode": {"keyAttributes": {"name": "inventory", "environment": "prod", "type": "Service"}, "groupByAttributes": {"telemetry_sdk_language": "java", "deployment_env": "prod"}}, "sourceOperation": {"name": "POST /checkout"}, "targetOperation": {"name": "GET /stock"}} diff --git a/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/apm_service_map/mapping.json b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/apm_service_map/mapping.json new file mode 100644 index 0000000000000..9c224af5deb08 --- /dev/null +++ b/sandbox/qa/analytics-engine-rest/src/test/resources/datasets/apm_service_map/mapping.json @@ -0,0 +1,63 @@ +{ + "settings": { + "number_of_shards": 1, + "number_of_replicas": 0 + }, + "mappings": { + "properties": { + "timestamp": { + "type": "date" + }, + "nodeConnectionHash": { + "type": "keyword" + }, + "operationConnectionHash": { + "type": "keyword" + }, + "sourceNode": { + "properties": { + "keyAttributes": { + "properties": { + "name": { "type": "keyword" }, + "environment": { "type": "keyword" }, + "type": { "type": "keyword" } + } + }, + "groupByAttributes": { + "properties": { + "telemetry_sdk_language": { "type": "keyword" }, + "deployment_env": { "type": "keyword" } + } + } + } + }, + "targetNode": { + "properties": { + "keyAttributes": { + "properties": { + "name": { "type": "keyword" }, + "environment": { "type": "keyword" }, + "type": { "type": "keyword" } + } + }, + "groupByAttributes": { + "properties": { + "telemetry_sdk_language": { "type": "keyword" }, + "deployment_env": { "type": "keyword" } + } + } + } + }, + "sourceOperation": { + "properties": { + "name": { "type": "keyword" } + } + }, + "targetOperation": { + "properties": { + "name": { "type": "keyword" } + } + } + } + } +}