diff --git a/core/src/main/java/org/opensearch/sql/analysis/Analyzer.java b/core/src/main/java/org/opensearch/sql/analysis/Analyzer.java index 2fdd8d6b843..36d2637c63e 100644 --- a/core/src/main/java/org/opensearch/sql/analysis/Analyzer.java +++ b/core/src/main/java/org/opensearch/sql/analysis/Analyzer.java @@ -81,6 +81,7 @@ import org.opensearch.sql.ast.tree.ML; import org.opensearch.sql.ast.tree.Multisearch; import org.opensearch.sql.ast.tree.MvCombine; +import org.opensearch.sql.ast.tree.MvExpand; import org.opensearch.sql.ast.tree.NoMv; import org.opensearch.sql.ast.tree.Paginate; import org.opensearch.sql.ast.tree.Parse; @@ -552,6 +553,11 @@ public LogicalPlan visitNoMv(NoMv node, AnalysisContext context) { throw getOnlyForCalciteException("nomv"); } + @Override + public LogicalPlan visitMvExpand(MvExpand node, AnalysisContext context) { + throw getOnlyForCalciteException("mvexpand"); + } + /** Build {@link ParseExpression} to context and skip to child nodes. */ @Override public LogicalPlan visitParse(Parse node, AnalysisContext context) { diff --git a/core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java b/core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java index 9a8fac2588d..f882314ea01 100644 --- a/core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java +++ b/core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java @@ -69,6 +69,7 @@ import org.opensearch.sql.ast.tree.ML; import org.opensearch.sql.ast.tree.Multisearch; import org.opensearch.sql.ast.tree.MvCombine; +import org.opensearch.sql.ast.tree.MvExpand; import org.opensearch.sql.ast.tree.NoMv; import org.opensearch.sql.ast.tree.Paginate; import org.opensearch.sql.ast.tree.Parse; @@ -480,4 +481,8 @@ public T visitMvCombine(MvCombine node, C context) { public T visitNoMv(NoMv node, C context) { return visitChildren(node, context); } + + public T visitMvExpand(MvExpand node, C context) { + return visitChildren(node, context); + } } diff --git a/core/src/main/java/org/opensearch/sql/ast/dsl/AstDSL.java b/core/src/main/java/org/opensearch/sql/ast/dsl/AstDSL.java index 8b129c6267a..b2731ebbd40 100644 --- a/core/src/main/java/org/opensearch/sql/ast/dsl/AstDSL.java +++ b/core/src/main/java/org/opensearch/sql/ast/dsl/AstDSL.java @@ -63,6 +63,7 @@ import org.opensearch.sql.ast.tree.Limit; import org.opensearch.sql.ast.tree.MinSpanBin; import org.opensearch.sql.ast.tree.MvCombine; +import org.opensearch.sql.ast.tree.MvExpand; import org.opensearch.sql.ast.tree.Parse; import org.opensearch.sql.ast.tree.Patterns; import org.opensearch.sql.ast.tree.Project; @@ -477,6 +478,16 @@ public static MvCombine mvcombine(Field field, String delim) { return new MvCombine(field, delim); } + /** + * Build an MVEXPAND plan node and attach it to the input plan. + * + *

`@param` input input plan `@param` field field to expand `@param` limit optional + * per-document limit `@return` MvExpand plan attached to the input + */ + public static UnresolvedPlan mvexpand(UnresolvedPlan input, Field field, Integer limit) { + return new MvExpand(field, limit).attach(input); + } + public static List sortOptions() { return exprList(argument("desc", booleanLiteral(false))); } diff --git a/core/src/main/java/org/opensearch/sql/ast/tree/MvExpand.java b/core/src/main/java/org/opensearch/sql/ast/tree/MvExpand.java new file mode 100644 index 00000000000..29dc89c541b --- /dev/null +++ b/core/src/main/java/org/opensearch/sql/ast/tree/MvExpand.java @@ -0,0 +1,46 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.ast.tree; + +import com.google.common.collect.ImmutableList; +import java.util.List; +import javax.annotation.Nullable; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.ToString; +import org.opensearch.sql.ast.AbstractNodeVisitor; +import org.opensearch.sql.ast.expression.Field; + +/** AST node representing the {@code mvexpand} PPL command: {@code mvexpand [limit=N]}. */ +@ToString +@EqualsAndHashCode(callSuper = false) +public class MvExpand extends UnresolvedPlan { + + private UnresolvedPlan child; + @Getter private final Field field; + @Getter @Nullable private final Integer limit; + + public MvExpand(Field field, @Nullable Integer limit) { + this.field = field; + this.limit = limit; + } + + @Override + public MvExpand attach(UnresolvedPlan child) { + this.child = child; + return this; + } + + @Override + public List getChild() { + return this.child == null ? ImmutableList.of() : ImmutableList.of(this.child); + } + + @Override + public T accept(AbstractNodeVisitor nodeVisitor, C context) { + return nodeVisitor.visitMvExpand(this, context); + } +} diff --git a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java index 84386dd0084..8e94aa976a4 100644 --- a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java +++ b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java @@ -126,6 +126,7 @@ import org.opensearch.sql.ast.tree.ML; import org.opensearch.sql.ast.tree.Multisearch; import org.opensearch.sql.ast.tree.MvCombine; +import org.opensearch.sql.ast.tree.MvExpand; import org.opensearch.sql.ast.tree.NoMv; import org.opensearch.sql.ast.tree.Paginate; import org.opensearch.sql.ast.tree.Parse; @@ -930,7 +931,11 @@ public RelNode visitPatterns(Patterns node, CalcitePlanContext context) { .toList(); context.relBuilder.aggregate(context.relBuilder.groupKey(groupByList), aggCall); buildExpandRelNode( - context.relBuilder.field(node.getAlias()), node.getAlias(), node.getAlias(), context); + context.relBuilder.field(node.getAlias()), + node.getAlias(), + node.getAlias(), + null, + context); flattenParsedPattern( node.getAlias(), context.relBuilder.field(node.getAlias()), @@ -3127,7 +3132,7 @@ public RelNode visitExpand(Expand expand, CalcitePlanContext context) { RexInputRef arrayFieldRex = (RexInputRef) rexVisitor.analyze(arrayField, context); String alias = expand.getAlias(); - buildExpandRelNode(arrayFieldRex, arrayField.getField().toString(), alias, context); + buildExpandRelNode(arrayFieldRex, arrayField.getField().toString(), alias, null, context); return context.relBuilder.peek(); } @@ -3320,6 +3325,61 @@ public RelNode visitNoMv(NoMv node, CalcitePlanContext context) { return visitEval((Eval) node.rewriteAsEval(), context); } + /** + * MVExpand command visitor. + * + *

Expands a multi-value (array) field into separate rows using Calcite's CORRELATE join with + * UNCOLLECT. Each element of the array becomes a separate row while preserving all other fields + * from the original row. + * + *

Implementation uses {@link #buildExpandRelNode} to create a correlate join between the + * original relation and an uncollected (unnested) version of the target array field. + * + *

Behavior: + * + *

+ * + * @param mvExpand MVExpand command containing the field to expand and optional limit + * @param context CalcitePlanContext containing the RelBuilder and planning context + * @return RelNode representing the relation with the expanded multi-value field + * @throws SemanticCheckException if the target field does not exist in the schema + */ + @Override + public RelNode visitMvExpand(MvExpand mvExpand, CalcitePlanContext context) { + visitChildren(mvExpand, context); + + final RelBuilder relBuilder = context.relBuilder; + final Field field = mvExpand.getField(); + final String fieldName = field.getField().toString(); + + final RelDataType inputType = relBuilder.peek().getRowType(); + final RelDataTypeField inputField = + inputType.getField(fieldName, /*caseSensitive*/ true, /*elideRecord*/ false); + + if (inputField == null) { + throw new SemanticCheckException( + String.format("Field '%s' not found in the schema", fieldName)); + } + + final RexInputRef arrayFieldRex = (RexInputRef) rexVisitor.analyze(field, context); + + final RelDataType fieldType = arrayFieldRex.getType(); + if (!(SqlTypeUtil.isArray(fieldType) || SqlTypeUtil.isMultiset(fieldType))) { + // For non-array/multiset fields (scalars), mvexpand just returns the field unchanged. + // This treats single-value fields as if they were arrays with one element. + return relBuilder.peek(); + } + + buildExpandRelNode(arrayFieldRex, fieldName, fieldName, mvExpand.getLimit(), context); + + return relBuilder.peek(); + } + @Override public RelNode visitValues(Values values, CalcitePlanContext context) { if (values.getValues() == null || values.getValues().isEmpty()) { @@ -3564,7 +3624,11 @@ private void flattenParsedPattern( } private void buildExpandRelNode( - RexInputRef arrayFieldRex, String arrayFieldName, String alias, CalcitePlanContext context) { + RexInputRef arrayFieldRex, + String arrayFieldName, + String alias, + @Nullable Integer perDocLimit, + CalcitePlanContext context) { // 3. Capture the outer row in a CorrelationId Holder correlVariable = Holder.empty(); context.relBuilder.variable(correlVariable::set); @@ -3579,14 +3643,17 @@ private void buildExpandRelNode( RelNode leftNode = context.relBuilder.build(); // 5. Build join right node and expand the array field using uncollect - RelNode rightNode = - context - .relBuilder - // fake input, see convertUnnest and convertExpression in Calcite SqlToRelConverter - .push(LogicalValues.createOneRow(context.relBuilder.getCluster())) - .project(List.of(correlArrayFieldAccess), List.of(arrayFieldName)) - .uncollect(List.of(), false) - .build(); + context + .relBuilder + // fake input, see convertUnnest and convertExpression in Calcite SqlToRelConverter + .push(LogicalValues.createOneRow(context.relBuilder.getCluster())) + .project(List.of(correlArrayFieldAccess), List.of(arrayFieldName)) + .uncollect(List.of(), false); + + if (perDocLimit != null) { + context.relBuilder.limit(0, perDocLimit); + } + RelNode rightNode = context.relBuilder.build(); // 6. Perform a nested-loop join (correlate) between the original table and the expanded // array field. diff --git a/core/src/main/java/org/opensearch/sql/expression/function/PPLFuncImpTable.java b/core/src/main/java/org/opensearch/sql/expression/function/PPLFuncImpTable.java index 29463c95f45..89a8f59397b 100644 --- a/core/src/main/java/org/opensearch/sql/expression/function/PPLFuncImpTable.java +++ b/core/src/main/java/org/opensearch/sql/expression/function/PPLFuncImpTable.java @@ -1105,6 +1105,10 @@ void populate() { OperandTypes.family(SqlTypeFamily.ARRAY, SqlTypeFamily.INTEGER) .or(OperandTypes.family(SqlTypeFamily.MAP, SqlTypeFamily.ANY)), false)); + registerOperator( + INTERNAL_ITEM, + SqlStdOperatorTable.ITEM, + PPLTypeChecker.family(SqlTypeFamily.IGNORE, SqlTypeFamily.CHARACTER)); registerOperator( XOR, SqlStdOperatorTable.NOT_EQUALS, diff --git a/docs/category.json b/docs/category.json index f90acc03692..5e9b6f954a5 100644 --- a/docs/category.json +++ b/docs/category.json @@ -26,6 +26,7 @@ "user/ppl/cmd/lookup.md", "user/ppl/cmd/mvcombine.md", "user/ppl/cmd/nomv.md", + "user/ppl/cmd/mvexpand.md", "user/ppl/cmd/parse.md", "user/ppl/cmd/patterns.md", "user/ppl/cmd/rare.md", diff --git a/docs/user/ppl/cmd/mvexpand.md b/docs/user/ppl/cmd/mvexpand.md new file mode 100644 index 00000000000..6fdd9bca365 --- /dev/null +++ b/docs/user/ppl/cmd/mvexpand.md @@ -0,0 +1,137 @@ +# mvexpand + +## Description +The `mvexpand` command expands each value in a multivalue (array) field into a separate row. For each document, every element in the specified array field is returned as a new row. + + +## Syntax +``` +mvexpand [limit=] +``` + +- ``: The multivalue (array) field to expand. (Required) +- `limit`: Maximum number of values per document to expand. If not specified, all array elements are expanded. (Optional) + + +### Output field naming +After `mvexpand`, the expanded value remains under the same field name (for example, `tags` or `ids`). +If the array contains objects, you can reference subfields (for example, `skills.name`). + + +## Examples + +### Example 1: Basic Expansion (single document) +Input document (case "basic") contains three tag values. + +PPL query: +```ppl +source=people +| eval tags = array('error', 'warning', 'info') +| fields tags +| head 1 +| mvexpand tags +| fields tags +``` + +Expected output: +```text +fetched rows / total rows = 3/3 ++---------+ +| tags | +|---------| +| error | +| warning | +| info | ++---------+ +``` + +### Example 2: Expansion with Limit +Input document (case "ids") contains an array of integers; expand and apply limit. + +PPL query: +```ppl +source=people +| eval ids = array(1, 2, 3, 4, 5) +| fields ids +| head 1 +| mvexpand ids limit=3 +| fields ids +``` + +Expected output: +```text +fetched rows / total rows = 3/3 ++-----+ +| ids | +|-----| +| 1 | +| 2 | +| 3 | ++-----+ +``` + +### Example 3: Expand projects +This example demonstrates expanding a multivalue `projects` field into one row per project. + +PPL query: +```ppl +source=people +| head 1 +| fields projects +| mvexpand projects +| fields projects.name +``` + +Expected output: +```text +fetched rows / total rows = 3/3 ++--------------------------------+ +| projects.name | +|--------------------------------| +| AWS Redshift Spectrum querying | +| AWS Redshift security | +| AWS Aurora security | ++--------------------------------+ +``` + +### Example 4: Single-value array (case "single") +Single-element array should expand to one row. + +PPL query: +```ppl +source=people +| eval tags = array('error') +| fields tags +| head 1 +| mvexpand tags +| fields tags +``` + +Expected output: +```text +fetched rows / total rows = 1/1 ++-------+ +| tags | +|-------| +| error | ++-------+ +``` + +### Example 5: Missing Field +If the field does not exist in the input schema (for example, it is not mapped or was projected out earlier), mvexpand throws a semantic check exception. + +PPL query: +```ppl +source=people +| eval some_field = 'x' +| fields some_field +| head 1 +| mvexpand tags +| fields tags +``` + +Expected output: +```text +{'reason': 'Invalid Query', 'details': "Field 'tags' not found in the schema", 'type': 'SemanticCheckException'} +Error: Query returned no data +``` \ No newline at end of file diff --git a/docs/user/ppl/index.md b/docs/user/ppl/index.md index 262cdf4f024..0608975fa4e 100644 --- a/docs/user/ppl/index.md +++ b/docs/user/ppl/index.md @@ -83,6 +83,7 @@ source=accounts | [transpose command](cmd/transpose.md) | 3.5 | stable (since 3.5) | Transpose rows to columns. | | [mvcombine command](cmd/mvcombine.md) | 3.5 | stable (since 3.4) | Combines values of a specified field across rows identical on all other fields. | | [nomv command](cmd/nomv.md) | 3.6 | stable (since 3.6) | Converts a multivalue field to a single-value string by joining elements with newlines. | +| [mvexpand command](cmd/mvexpand.md) | 3.6 | stable (since 3.6) | Expand a multi-valued field into separate documents (one per value). | - [Syntax](cmd/syntax.md) - PPL query structure and command syntax formatting * **Functions** diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java index c564c0b86cc..47acf06e6a3 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java @@ -109,7 +109,8 @@ CalciteWhereCommandIT.class, CalcitePPLTpchIT.class, CalciteMvCombineCommandIT.class, - CalciteNoMvCommandIT.class + CalciteNoMvCommandIT.class, + CalciteMvExpandCommandIT.class, }) public class CalciteNoPushdownIT { private static boolean wasPushdownEnabled; diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java index 3f901bddec7..8e980d8973b 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java @@ -59,6 +59,7 @@ public void init() throws Exception { loadIndex(Index.DATA_TYPE_ALIAS); loadIndex(Index.DEEP_NESTED); loadIndex(Index.CASCADED_NESTED); + loadIndex(Index.MVEXPAND_EDGE_CASES); } @Override @@ -2553,6 +2554,15 @@ public void testExplainNoMv() throws IOException { assertYamlEqualsIgnoreId(expected, actual); } + @Test + public void testMvexpandExplain() throws IOException { + String expected = loadExpectedPlan("explain_mvexpand.yaml"); + String actual = + explainQueryYaml( + "source=mvexpand_edge_cases | eval skills_arr = array(1, 2, 3) | mvexpand skills_arr"); + assertYamlEqualsIgnoreId(expected, actual); + } + // ==================== fetch_size explain tests ==================== @Test diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteMvExpandCommandIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteMvExpandCommandIT.java new file mode 100644 index 00000000000..99334dcb6c2 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteMvExpandCommandIT.java @@ -0,0 +1,282 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.calcite.remote; + +import static org.opensearch.sql.util.MatcherUtils.rows; +import static org.opensearch.sql.util.MatcherUtils.schema; +import static org.opensearch.sql.util.MatcherUtils.verifyDataRows; +import static org.opensearch.sql.util.MatcherUtils.verifyNumOfRows; +import static org.opensearch.sql.util.MatcherUtils.verifySchema; + +import org.json.JSONObject; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.opensearch.sql.ppl.PPLIntegTestCase; + +public class CalciteMvExpandCommandIT extends PPLIntegTestCase { + + private static final String INDEX = Index.MVEXPAND_EDGE_CASES.getName(); + + @Override + public void init() throws Exception { + super.init(); + enableCalcite(); + loadIndex(Index.MVEXPAND_EDGE_CASES); + } + + @Test + public void testMvexpandSingleElement() throws Exception { + String q1 = + String.format( + "source=%s | mvexpand skills | where username='single' | fields username, skills", + INDEX); + JSONObject r1 = executeQuery(q1); + + assertSingleRowNestedFieldEquals(r1, "skills", "name", "go"); + + String q2 = + String.format( + "source=%s | mvexpand skills | where username='single' | fields username, skills.name", + INDEX); + JSONObject r2 = executeQuery(q2); + verifyDataRows(r2, rows("single", "go")); + } + + /** + * Asserts the result has exactly one row and that the given column is a MAP/object containing + * nestedKey=nestedValue. + */ + private static void assertSingleRowNestedFieldEquals( + JSONObject result, String mapColumn, String nestedKey, String expectedValue) { + var dataRows = result.getJSONArray("datarows"); + Assertions.assertEquals(1, dataRows.length(), "Expected exactly one row"); + + var schema = result.getJSONArray("schema"); + + int mapIdx = -1; + for (int i = 0; i < schema.length(); i++) { + if (mapColumn.equals(schema.getJSONObject(i).getString("name"))) { + mapIdx = i; + break; + } + } + Assertions.assertTrue(mapIdx >= 0, "Column not found in schema: " + mapColumn); + + var row0 = dataRows.getJSONArray(0); + var skillsObj = row0.getJSONObject(mapIdx); // this is the MAP/object + Assertions.assertEquals(expectedValue, skillsObj.optString(nestedKey, null)); + } + + @Test + public void testMvexpandEmptyArray() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills | where username='empty' | fields username, skills.name", + INDEX); + JSONObject result = executeQuery(query); + verifyDataRows(result); + } + + @Test + public void testMvexpandNullArray() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills | where username='nullskills' | fields username," + + " skills.name", + INDEX); + JSONObject result = executeQuery(query); + verifyDataRows(result); + } + + @Test + public void testMvexpandNoArrayField() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills | where username='noskills' | fields username," + + " skills.name", + INDEX); + JSONObject result = executeQuery(query); + verifyDataRows(result); + } + + @Test + public void testMvexpandDuplicate() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills | where username='duplicate' | fields username," + + " skills.name | sort skills.name", + INDEX); + JSONObject result = executeQuery(query); + verifyDataRows(result, rows("duplicate", "dup"), rows("duplicate", "dup")); + } + + @Test + public void testMvexpandHappyMultipleElements() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills | where username='happy' | fields username, skills.name |" + + " sort skills.name", + INDEX); + JSONObject result = executeQuery(query); + verifyDataRows(result, rows("happy", "java"), rows("happy", "python"), rows("happy", "sql")); + } + + @Test + public void testMvexpandPartialElementMissingName() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills | where username='partial' | fields username, skills.name" + + " | sort skills.name", + INDEX); + JSONObject result = executeQuery(query); + verifyDataRows( + result, + rows("partial", "kotlin"), + rows("partial", (String) null), + rows("partial", (String) null)); + } + + @Test + public void testMvexpandMixedShapesKeepsAllElements() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills | where username='mixed_shapes' | fields username," + + " skills.name | sort skills.name", + INDEX); + JSONObject result = executeQuery(query); + verifyDataRows(result, rows("mixed_shapes", "elixir"), rows("mixed_shapes", "haskell")); + } + + @Test + public void testMvexpandFlattenedSchemaPresence() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills | where username='complex' | fields username," + + " skills.level, skills.name", + INDEX); + JSONObject result = executeQuery(query); + + verifySchema( + result, + schema("username", "string"), + schema("skills.level", "string"), + schema("skills.name", "string")); + + verifyDataRows( + result, + rows("complex", "expert", "ml"), + rows("complex", (String) null, "ai"), + rows("complex", "novice", (String) null)); + } + + @Test + public void testMvexpandOnNonArrayFieldMapping() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills_not_array | where username='u1' | fields username," + + " skills_not_array", + INDEX); + + JSONObject result = executeQuery(query); + + verifyNumOfRows(result, 1); + verifyDataRows(result, rows("u1", "scala")); + } + + @Test + public void testMvexpandMissingFieldReturnsEmpty() throws Exception { + // single-index version: username='noskills' doc has no "skills" field at all + String query = + String.format( + "source=%s | mvexpand skills | where username='noskills' | fields username, skills", + INDEX); + + JSONObject result = executeQuery(query); + verifyDataRows(result); + } + + @Test + public void testMvexpandLimitParameter() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills limit=3 | where username='limituser' | fields username," + + " skills.name", + INDEX); + JSONObject result = executeQuery(query); + verifyNumOfRows(result, 3); + verifyDataRows(result, rows("limituser", "a"), rows("limituser", "b"), rows("limituser", "c")); + } + + @Test + public void testMvexpandMultiDocumentLimitParameter() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills limit=2 | where username='happy' OR username='limituser'" + + " | fields username, skills.name | sort username, skills.name", + INDEX); + JSONObject result = executeQuery(query); + + verifyNumOfRows(result, 4); + + verifyDataRows( + result, + rows("happy", "java"), + rows("happy", "python"), + rows("limituser", "a"), + rows("limituser", "b")); + } + + @Test + public void testMvexpandTypeInferenceForHeterogeneousSubfields() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills | where username='hetero_types' | fields username," + + " skills.level", + INDEX); + JSONObject result = executeQuery(query); + + verifyDataRows(result, rows("hetero_types", "senior"), rows("hetero_types", "3")); + } + + @Test + public void testMvexpandLargeArrayElements() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills | where username='large' | fields username, skills.name |" + + " sort skills.name", + INDEX); + JSONObject result = executeQuery(query); + + verifyNumOfRows(result, 10); + + verifyDataRows( + result, + rows("large", "s1"), + rows("large", "s10"), + rows("large", "s2"), + rows("large", "s3"), + rows("large", "s4"), + rows("large", "s5"), + rows("large", "s6"), + rows("large", "s7"), + rows("large", "s8"), + rows("large", "s9")); + } + + @Test + public void testMvexpandOnIntegerFieldMapping() throws Exception { + String query = + String.format( + "source=%s | mvexpand skills_int | where username='u_int' | fields username," + + " skills_int", + INDEX); + + JSONObject result = executeQuery(query); + + verifyNumOfRows(result, 1); + verifyDataRows(result, rows("u_int", 5)); + } +} diff --git a/integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java b/integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java index c9de7a584c6..a5228c226e4 100644 --- a/integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java +++ b/integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java @@ -686,6 +686,11 @@ public enum Index { "_doc", getNestedSimpleIndexMapping(), "src/test/resources/nested_simple.json"), + MVEXPAND_EDGE_CASES( + "mvexpand_edge_cases", + "mvexpand_edge_cases", + getMappingFile("mvexpand_edge_cases_mapping.json"), + "src/test/resources/mvexpand_edge_cases.json"), DEEP_NESTED( TestsConstants.TEST_INDEX_DEEP_NESTED, "_doc", diff --git a/integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java b/integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java index ad8a232bab3..cd6c72bffa8 100644 --- a/integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java +++ b/integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java @@ -92,6 +92,7 @@ public class TestsConstants { public static final String TEST_INDEX_LOGS = TEST_INDEX + "_logs"; public static final String TEST_INDEX_OTEL_LOGS = TEST_INDEX + "_otel_logs"; public static final String TEST_INDEX_TIME_DATE_NULL = TEST_INDEX + "_time_date_null"; + public static final String TEST_INDEX_MVEXPAND_EDGE_CASES = "mvexpand_edge_cases"; public static final String DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"; public static final String TS_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss.SSS"; diff --git a/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java b/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java index 5a7f7be9220..a5164ec7fef 100644 --- a/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java @@ -10,9 +10,11 @@ import static org.opensearch.sql.common.setting.Settings.Key.CALCITE_ENGINE_ENABLED; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_DOG; +import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_MVEXPAND_EDGE_CASES; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_STRINGS; import java.io.IOException; +import org.json.JSONArray; import org.json.JSONObject; import org.junit.jupiter.api.Test; import org.opensearch.client.ResponseException; @@ -25,6 +27,7 @@ public void init() throws Exception { loadIndex(Index.BANK); loadIndex(Index.DOG); loadIndex(Index.STRINGS); + loadIndex(Index.MVEXPAND_EDGE_CASES); } @Test @@ -268,4 +271,217 @@ public void testNoMvUnsupportedInV2() throws IOException { } verifyQuery(result); } + + @Test + public void testMvExpandCommandBasicExpansion() throws IOException { + JSONObject result; + try { + result = + executeQuery( + String.format( + "search source=%s | mvexpand skills | where username='happy' | fields username," + + " skills.name | sort skills.name", + TEST_INDEX_MVEXPAND_EDGE_CASES)); + } catch (ResponseException e) { + result = new JSONObject(TestUtils.getResponseBody(e.getResponse())); + } + + if (isCalciteEnabled()) { + assertThat(result.getJSONArray("datarows").length(), equalTo(3)); + + JSONArray datarows = result.getJSONArray("datarows"); + assertThat(datarows.getJSONArray(0).getString(0), equalTo("happy")); + assertThat(datarows.getJSONArray(0).getString(1), equalTo("java")); + assertThat(datarows.getJSONArray(1).getString(1), equalTo("python")); + assertThat(datarows.getJSONArray(2).getString(1), equalTo("sql")); + } else { + JSONObject error = result.getJSONObject("error"); + assertThat( + error.getString("details"), + containsString( + "is supported only when " + CALCITE_ENGINE_ENABLED.getKeyValue() + "=true")); + assertThat(error.getString("type"), equalTo("UnsupportedOperationException")); + } + } + + @Test + public void testMvExpandCommandNullInput() throws IOException { + JSONObject result; + try { + result = + executeQuery( + String.format( + "search source=%s | mvexpand skills | where username='nullskills' | fields" + + " username, skills.name", + TEST_INDEX_MVEXPAND_EDGE_CASES)); + } catch (ResponseException e) { + result = new JSONObject(TestUtils.getResponseBody(e.getResponse())); + } + + if (isCalciteEnabled()) { + assertThat(result.getJSONArray("datarows").length(), equalTo(0)); + } else { + JSONObject error = result.getJSONObject("error"); + assertThat( + error.getString("details"), + containsString( + "is supported only when " + CALCITE_ENGINE_ENABLED.getKeyValue() + "=true")); + assertThat(error.getString("type"), equalTo("UnsupportedOperationException")); + } + } + + @Test + public void testMvExpandCommandEmptyArray() throws IOException { + JSONObject result; + try { + result = + executeQuery( + String.format( + "search source=%s | mvexpand skills | where username='empty' | fields username," + + " skills.name", + TEST_INDEX_MVEXPAND_EDGE_CASES)); + } catch (ResponseException e) { + result = new JSONObject(TestUtils.getResponseBody(e.getResponse())); + } + + if (isCalciteEnabled()) { + assertThat(result.getJSONArray("datarows").length(), equalTo(0)); + } else { + JSONObject error = result.getJSONObject("error"); + assertThat( + error.getString("details"), + containsString( + "is supported only when " + CALCITE_ENGINE_ENABLED.getKeyValue() + "=true")); + assertThat(error.getString("type"), equalTo("UnsupportedOperationException")); + } + } + + @Test + public void testMvExpandCommandNonArrayField() throws IOException { + JSONObject result; + try { + result = + executeQuery( + String.format( + "search source=%s | mvexpand skills_not_array | where username='u1' | fields" + + " username, skills_not_array", + TEST_INDEX_MVEXPAND_EDGE_CASES)); + } catch (ResponseException e) { + result = new JSONObject(TestUtils.getResponseBody(e.getResponse())); + } + + if (isCalciteEnabled()) { + assertThat(result.getJSONArray("datarows").length(), equalTo(1)); + assertThat(result.getJSONArray("datarows").getJSONArray(0).getString(1), equalTo("scala")); + } else { + JSONObject error = result.getJSONObject("error"); + assertThat( + error.getString("details"), + containsString( + "is supported only when " + CALCITE_ENGINE_ENABLED.getKeyValue() + "=true")); + assertThat(error.getString("type"), equalTo("UnsupportedOperationException")); + } + } + + @Test + public void testMvExpandCommandLimitBoundary() throws IOException { + JSONObject result; + try { + result = + executeQuery( + String.format( + "search source=%s | mvexpand skills limit=3 | where username='limituser' | fields" + + " username, skills.name", + TEST_INDEX_MVEXPAND_EDGE_CASES)); + } catch (ResponseException e) { + result = new JSONObject(TestUtils.getResponseBody(e.getResponse())); + } + + if (isCalciteEnabled()) { + assertThat(result.getJSONArray("datarows").length(), equalTo(3)); + } else { + JSONObject error = result.getJSONObject("error"); + assertThat( + error.getString("details"), + containsString( + "is supported only when " + CALCITE_ENGINE_ENABLED.getKeyValue() + "=true")); + assertThat(error.getString("type"), equalTo("UnsupportedOperationException")); + } + } + + @Test + public void testMvExpandCommandMultiDocument() throws IOException { + JSONObject result; + try { + result = + executeQuery( + String.format( + "search source=%s | mvexpand skills | where username='happy' OR username='single'" + + " | fields username, skills.name | sort username, skills.name", + TEST_INDEX_MVEXPAND_EDGE_CASES)); + } catch (ResponseException e) { + result = new JSONObject(TestUtils.getResponseBody(e.getResponse())); + } + + if (isCalciteEnabled()) { + assertThat(result.getJSONArray("datarows").length(), equalTo(4)); + + JSONArray datarows = result.getJSONArray("datarows"); + assertThat(datarows.getJSONArray(0).getString(0), equalTo("happy")); + assertThat(datarows.getJSONArray(3).getString(0), equalTo("single")); + } else { + JSONObject error = result.getJSONObject("error"); + assertThat( + error.getString("details"), + containsString( + "is supported only when " + CALCITE_ENGINE_ENABLED.getKeyValue() + "=true")); + assertThat(error.getString("type"), equalTo("UnsupportedOperationException")); + } + } + + @Test + public void testMvExpandInvalidLimitZero() throws IOException { + if (!isCalciteEnabled()) { + return; // Skip test when Calcite is disabled + } + try { + executeQuery( + String.format( + "search source=%s | mvexpand skills limit=0 | fields username, skills.name", + TEST_INDEX_MVEXPAND_EDGE_CASES)); + fail("Expected IllegalArgumentException for limit=0"); + } catch (ResponseException e) { + JSONObject result = new JSONObject(TestUtils.getResponseBody(e.getResponse())); + JSONObject error = result.getJSONObject("error"); + String details = error.getString("details"); + assertThat( + "Error message should mention limit or positive", + details.toLowerCase(), + containsString("limit")); + assertThat(error.getString("type"), equalTo("IllegalArgumentException")); + } + } + + @Test + public void testMvExpandInvalidLimitNegative() throws IOException { + if (!isCalciteEnabled()) { + return; // Skip test when Calcite is disabled + } + try { + executeQuery( + String.format( + "search source=%s | mvexpand skills limit=-1 | fields username, skills.name", + TEST_INDEX_MVEXPAND_EDGE_CASES)); + fail("Expected SyntaxCheckException for negative limit"); + } catch (ResponseException e) { + JSONObject result = new JSONObject(TestUtils.getResponseBody(e.getResponse())); + JSONObject error = result.getJSONObject("error"); + String details = error.getString("details"); + assertThat( + "Error message should mention parsing error", + details.toLowerCase(), + containsString("extraneous")); + assertThat(error.getString("type"), equalTo("SyntaxCheckException")); + } + } } diff --git a/integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterSearchIT.java b/integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterSearchIT.java index 0c101ce4723..13dbdce4bce 100644 --- a/integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterSearchIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterSearchIT.java @@ -31,6 +31,8 @@ protected void init() throws Exception { loadIndex(Index.ACCOUNT, remoteClient()); loadIndex(Index.TIME_TEST_DATA); loadIndex(Index.TIME_TEST_DATA, remoteClient()); + loadIndex(Index.MVEXPAND_EDGE_CASES); + loadIndex(Index.MVEXPAND_EDGE_CASES, remoteClient()); enableCalcite(); } @@ -435,4 +437,28 @@ public void testCrossClusterNoMv() throws IOException { verifyDataRows(result, rows("Hattie", "Hattie\nBond")); } + + @Test + public void testCrossClusterMvExpandBasic() throws IOException { + JSONObject result = + executeQuery( + String.format( + "search source=%s | mvexpand skills | where username='happy' | fields username," + + " skills.name | sort skills.name", + TEST_INDEX_MVEXPAND_REMOTE)); + verifySchema(result, schema("username", "string"), schema("skills.name", "string")); + verifyDataRows(result, rows("happy", "java"), rows("happy", "python"), rows("happy", "sql")); + } + + @Test + public void testCrossClusterMvExpandWithLimit() throws IOException { + JSONObject result = + executeQuery( + String.format( + "search source=%s | mvexpand skills limit=2 | where username='limituser' | fields" + + " username, skills.name | sort skills.name", + TEST_INDEX_MVEXPAND_REMOTE)); + verifySchema(result, schema("username", "string"), schema("skills.name", "string")); + verifyDataRows(result, rows("limituser", "a"), rows("limituser", "b")); + } } diff --git a/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterTestBase.java b/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterTestBase.java index d9de95c663b..dc4d7d0dafd 100644 --- a/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterTestBase.java +++ b/integ-test/src/test/java/org/opensearch/sql/security/CrossClusterTestBase.java @@ -8,6 +8,7 @@ import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_ACCOUNT; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_DOG; +import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_MVEXPAND_EDGE_CASES; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_TIME_DATA; import org.opensearch.sql.ppl.PPLIntegTestCase; @@ -36,6 +37,8 @@ public class CrossClusterTestBase extends PPLIntegTestCase { REMOTE_CLUSTER + ":" + TEST_INDEX_ACCOUNT; protected static final String TEST_INDEX_TIME_DATA_REMOTE = REMOTE_CLUSTER + ":" + TEST_INDEX_TIME_DATA; + protected static final String TEST_INDEX_MVEXPAND_REMOTE = + REMOTE_CLUSTER + ":" + TEST_INDEX_MVEXPAND_EDGE_CASES; @Override protected void init() throws Exception { diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_mvexpand.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_mvexpand.yaml new file mode 100644 index 00000000000..c07f8d5f063 --- /dev/null +++ b/integ-test/src/test/resources/expectedOutput/calcite/explain_mvexpand.yaml @@ -0,0 +1,19 @@ +calcite: + logical: | + LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT]) + LogicalProject(skills=[$0], skills_int=[$3], skills_not_array=[$4], username=[$5], skills_arr=[$13]) + LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{12}]) + LogicalProject(skills=[$0], skills.level=[$1], skills.name=[$2], skills_int=[$3], skills_not_array=[$4], username=[$5], _id=[$6], _index=[$7], _score=[$8], _maxscore=[$9], _sort=[$10], _routing=[$11], skills_arr=[array(1, 2, 3)]) + CalciteLogicalIndexScan(table=[[OpenSearch, mvexpand_edge_cases]]) + Uncollect + LogicalProject(skills_arr=[$cor0.skills_arr]) + LogicalValues(tuples=[[{ 0 }]]) + physical: | + EnumerableLimit(fetch=[10000]) + EnumerableCalc(expr#0..5=[{inputs}], proj#0..3=[{exprs}], skills_arr=[$t5]) + EnumerableCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{4}]) + EnumerableCalc(expr#0..3=[{inputs}], expr#4=[1], expr#5=[2], expr#6=[3], expr#7=[array($t4, $t5, $t6)], proj#0..3=[{exprs}], skills_arr=[$t7]) + CalciteEnumerableIndexScan(table=[[OpenSearch, mvexpand_edge_cases]], PushDownContext=[[PROJECT->[skills, skills_int, skills_not_array, username]], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","_source":{"includes":["skills","skills_int","skills_not_array","username"],"excludes":[]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)]) + EnumerableUncollect + EnumerableCalc(expr#0=[{inputs}], expr#1=[$cor0], expr#2=[$t1.skills_arr], skills_arr=[$t2]) + EnumerableValues(tuples=[[{ 0 }]]) diff --git a/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_mvexpand.yaml b/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_mvexpand.yaml new file mode 100644 index 00000000000..5efc2d80da9 --- /dev/null +++ b/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_mvexpand.yaml @@ -0,0 +1,19 @@ +calcite: + logical: | + LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT]) + LogicalProject(skills=[$0], skills_int=[$3], skills_not_array=[$4], username=[$5], skills_arr=[$13]) + LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{12}]) + LogicalProject(skills=[$0], skills.level=[$1], skills.name=[$2], skills_int=[$3], skills_not_array=[$4], username=[$5], _id=[$6], _index=[$7], _score=[$8], _maxscore=[$9], _sort=[$10], _routing=[$11], skills_arr=[array(1, 2, 3)]) + CalciteLogicalIndexScan(table=[[OpenSearch, mvexpand_edge_cases]]) + Uncollect + LogicalProject(skills_arr=[$cor0.skills_arr]) + LogicalValues(tuples=[[{ 0 }]]) + physical: | + EnumerableLimit(fetch=[10000]) + EnumerableCalc(expr#0..5=[{inputs}], proj#0..3=[{exprs}], skills_arr=[$t5]) + EnumerableCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{4}]) + EnumerableCalc(expr#0..11=[{inputs}], expr#12=[1], expr#13=[2], expr#14=[3], expr#15=[array($t12, $t13, $t14)], skills=[$t0], skills_int=[$t3], skills_not_array=[$t4], username=[$t5], skills_arr=[$t15]) + CalciteEnumerableIndexScan(table=[[OpenSearch, mvexpand_edge_cases]]) + EnumerableUncollect + EnumerableCalc(expr#0=[{inputs}], expr#1=[$cor0], expr#2=[$t1.skills_arr], skills_arr=[$t2]) + EnumerableValues(tuples=[[{ 0 }]]) diff --git a/integ-test/src/test/resources/indexDefinitions/mvexpand_edge_cases_mapping.json b/integ-test/src/test/resources/indexDefinitions/mvexpand_edge_cases_mapping.json new file mode 100644 index 00000000000..a0b5519d176 --- /dev/null +++ b/integ-test/src/test/resources/indexDefinitions/mvexpand_edge_cases_mapping.json @@ -0,0 +1,16 @@ +{ + "mappings": { + "properties": { + "username": { "type": "keyword" }, + "skills": { + "type": "nested", + "properties": { + "name": { "type": "keyword" }, + "level": { "type": "keyword" } + } + }, + "skills_not_array": { "type": "keyword" }, + "skills_int": { "type": "integer" } + } + } +} diff --git a/integ-test/src/test/resources/mvexpand_edge_cases.json b/integ-test/src/test/resources/mvexpand_edge_cases.json new file mode 100644 index 00000000000..c7632bb1e24 --- /dev/null +++ b/integ-test/src/test/resources/mvexpand_edge_cases.json @@ -0,0 +1,30 @@ +{"index":{}} +{"username":"happy","skills":[{"name":"python"},{"name":"java"},{"name":"sql"}]} +{"index":{}} +{"username":"single","skills":[{"name":"go"}]} +{"index":{}} +{"username":"empty","skills":[]} +{"index":{}} +{"username":"nullskills","skills":null} +{"index":{}} +{"username":"noskills"} +{"index":{}} +{"username":"missingattr","skills":[{"name":"c"},{"level":"advanced"}]} +{"index":{}} +{"username":"complex","skills":[{"name":"ml","level":"expert"},{"name":"ai"},{"level":"novice"}]} +{"index":{}} +{"username":"duplicate","skills":[{"name":"dup"},{"name":"dup"}]} +{"index":{}} +{"username":"large","skills":[{"name":"s1"},{"name":"s2"},{"name":"s3"},{"name":"s4"},{"name":"s5"},{"name":"s6"},{"name":"s7"},{"name":"s8"},{"name":"s9"},{"name":"s10"}]} +{"index":{}} +{"username":"partial","skills":[{"name":"kotlin"},{"level":"intermediate"},{"level":"advanced"}]} +{"index":{}} +{"username":"mixed_shapes","skills":[{"name":"elixir"},{"name":"haskell"}]} +{"index":{}} +{"username":"hetero_types","skills":[{"level":"senior"},{"level":"3"}]} +{"index":{}} +{"username":"limituser","skills":[{"name":"a"},{"name":"b"},{"name":"c"},{"name":"d"},{"name":"e"}]} +{"index":{}} +{"username":"u1","skills_not_array":"scala"} +{"index":{}} +{"username":"u_int","skills_int":5} diff --git a/ppl/src/main/antlr/OpenSearchPPLLexer.g4 b/ppl/src/main/antlr/OpenSearchPPLLexer.g4 index 54d69da39b9..8ccea8fcde5 100644 --- a/ppl/src/main/antlr/OpenSearchPPLLexer.g4 +++ b/ppl/src/main/antlr/OpenSearchPPLLexer.g4 @@ -56,6 +56,7 @@ ADDCOLTOTALS: 'ADDCOLTOTALS'; ROW: 'ROW'; COL: 'COL'; EXPAND: 'EXPAND'; +MVEXPAND: 'MVEXPAND'; SIMPLE_PATTERN: 'SIMPLE_PATTERN'; BRAIN: 'BRAIN'; VARIABLE_COUNT_THRESHOLD: 'VARIABLE_COUNT_THRESHOLD'; diff --git a/ppl/src/main/antlr/OpenSearchPPLParser.g4 b/ppl/src/main/antlr/OpenSearchPPLParser.g4 index 455fd92c3d2..3b7df6c5d93 100644 --- a/ppl/src/main/antlr/OpenSearchPPLParser.g4 +++ b/ppl/src/main/antlr/OpenSearchPPLParser.g4 @@ -80,6 +80,7 @@ commands | addcoltotalsCommand | appendCommand | expandCommand + | mvexpandCommand | flattenCommand | reverseCommand | regexCommand @@ -123,6 +124,7 @@ commandName | ML | FILLNULL | EXPAND + | MVEXPAND | FLATTEN | TRENDLINE | TIMECHART @@ -561,6 +563,10 @@ nomvCommand : NOMV fieldExpression ; +mvexpandCommand + : MVEXPAND fieldExpression (LIMIT EQUAL INTEGER_LITERAL)? + ; + flattenCommand : FLATTEN fieldExpression (AS aliases = identifierSeq)? ; diff --git a/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java b/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java index 118ca94dbd6..71fc860ba30 100644 --- a/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java +++ b/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java @@ -93,6 +93,7 @@ import org.opensearch.sql.ast.tree.MinSpanBin; import org.opensearch.sql.ast.tree.Multisearch; import org.opensearch.sql.ast.tree.MvCombine; +import org.opensearch.sql.ast.tree.MvExpand; import org.opensearch.sql.ast.tree.NoMv; import org.opensearch.sql.ast.tree.Parse; import org.opensearch.sql.ast.tree.Patterns; @@ -915,6 +916,19 @@ public UnresolvedPlan visitNomvCommand(OpenSearchPPLParser.NomvCommandContext ct return new NoMv(field); } + @Override + public UnresolvedPlan visitMvexpandCommand(OpenSearchPPLParser.MvexpandCommandContext ctx) { + Field field = (Field) expressionBuilder.visit(ctx.fieldExpression()); + Integer limit = + ctx.INTEGER_LITERAL() != null ? Integer.parseInt(ctx.INTEGER_LITERAL().getText()) : null; + + if (limit != null && limit <= 0) { + throw new IllegalArgumentException("Limit must be a positive number, got: " + limit); + } + + return new MvExpand(field, limit); + } + @Override public UnresolvedPlan visitGrokCommand(OpenSearchPPLParser.GrokCommandContext ctx) { UnresolvedExpression sourceField = internalVisitExpression(ctx.source_field); diff --git a/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java b/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java index a47db85c3cb..147f2422677 100644 --- a/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java +++ b/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java @@ -83,6 +83,7 @@ import org.opensearch.sql.ast.tree.MinSpanBin; import org.opensearch.sql.ast.tree.Multisearch; import org.opensearch.sql.ast.tree.MvCombine; +import org.opensearch.sql.ast.tree.MvExpand; import org.opensearch.sql.ast.tree.NoMv; import org.opensearch.sql.ast.tree.Parse; import org.opensearch.sql.ast.tree.Patterns; @@ -482,6 +483,16 @@ public String visitNoMv(NoMv node, String context) { return StringUtils.format("%s | nomv %s", child, field); } + @Override + public String visitMvExpand(MvExpand node, String context) { + String child = node.getChild().get(0).accept(this, context); + String field = MASK_COLUMN; // Always anonymize field names + if (node.getLimit() != null) { + return StringUtils.format("%s | mvexpand %s limit=%s", child, field, MASK_LITERAL); + } + return StringUtils.format("%s | mvexpand %s", child, field); + } + /** Build {@link LogicalSort}. */ @Override public String visitSort(Sort node, String context) { diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMvExpandTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMvExpandTest.java new file mode 100644 index 00000000000..ac37dab12a9 --- /dev/null +++ b/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMvExpandTest.java @@ -0,0 +1,279 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.ppl.calcite; + +import com.google.common.collect.ImmutableList; +import java.util.Arrays; +import java.util.List; +import org.apache.calcite.config.CalciteConnectionConfig; +import org.apache.calcite.plan.RelTraitDef; +import org.apache.calcite.rel.RelCollations; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rel.type.RelProtoDataType; +import org.apache.calcite.schema.Schema; +import org.apache.calcite.schema.SchemaPlus; +import org.apache.calcite.schema.Statistic; +import org.apache.calcite.schema.Statistics; +import org.apache.calcite.schema.Table; +import org.apache.calcite.sql.SqlCall; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.parser.SqlParser; +import org.apache.calcite.sql.type.SqlTypeName; +import org.apache.calcite.test.CalciteAssert; +import org.apache.calcite.tools.Frameworks; +import org.apache.calcite.tools.Programs; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.junit.Assert; +import org.junit.Test; + +public class CalcitePPLMvExpandTest extends CalcitePPLAbstractTest { + + public CalcitePPLMvExpandTest() { + super(CalciteAssert.SchemaSpec.SCOTT_WITH_TEMPORAL); + } + + /** + * There is no existing table with arrays. We create one for test purpose. + * + *

This mirrors CalcitePPLExpandTest.TableWithArray. + */ + public static class TableWithArray implements Table { + protected final RelProtoDataType protoRowType = + factory -> + factory + .builder() + .add("DEPTNO", SqlTypeName.INTEGER) + .add( + "EMPNOS", + factory.createArrayType(factory.createSqlType(SqlTypeName.INTEGER), -1)) + .add( + "TAGS", + factory.createMultisetType(factory.createSqlType(SqlTypeName.VARCHAR), -1)) + .build(); + + @Override + public RelDataType getRowType(RelDataTypeFactory typeFactory) { + return protoRowType.apply(typeFactory); + } + + @Override + public Statistic getStatistic() { + return Statistics.of(0d, ImmutableList.of(), RelCollations.createSingleton(0)); + } + + @Override + public Schema.TableType getJdbcTableType() { + return Schema.TableType.TABLE; + } + + @Override + public boolean isRolledUp(String column) { + return false; + } + + @Override + public boolean rolledUpColumnValidInsideAgg( + String column, + SqlCall call, + @Nullable SqlNode parent, + @Nullable CalciteConnectionConfig config) { + return false; + } + } + + @Override + protected Frameworks.ConfigBuilder config(CalciteAssert.SchemaSpec... schemaSpecs) { + final SchemaPlus rootSchema = Frameworks.createRootSchema(true); + final SchemaPlus schema = CalciteAssert.addSchema(rootSchema, schemaSpecs); + schema.add("DEPT", new TableWithArray()); + return Frameworks.newConfigBuilder() + .parserConfig(SqlParser.Config.DEFAULT) + .defaultSchema(schema) + .traitDefs((List) null) + .programs(Programs.heuristicJoinOrder(Programs.RULE_SET, true, 2)); + } + + @Test + public void testMvExpandBasic() { + String ppl = "source=DEPT | mvexpand EMPNOS"; + RelNode root = getRelNode(ppl); + String expectedLogical = + "LogicalProject(DEPTNO=[$0], TAGS=[$2], EMPNOS=[$3])\n" + + " LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])\n" + + " LogicalTableScan(table=[[scott, DEPT]])\n" + + " Uncollect\n" + + " LogicalProject(EMPNOS=[$cor0.EMPNOS])\n" + + " LogicalValues(tuples=[[{ 0 }]])\n"; + verifyLogical(root, expectedLogical); + + String expectedSparkSql = + "SELECT `$cor0`.`DEPTNO`, `$cor0`.`TAGS`, `t00`.`EMPNOS`\n" + + "FROM `scott`.`DEPT` `$cor0`,\n" + + "LATERAL UNNEST((SELECT `$cor0`.`EMPNOS`\n" + + "FROM (VALUES (0)) `t` (`ZERO`))) `t00` (`EMPNOS`)"; + verifyPPLToSparkSQL(root, expectedSparkSql); + } + + @Test + public void testMvExpandWithLimitParameter() { + String ppl = "source=DEPT | mvexpand EMPNOS limit=2"; + RelNode root = getRelNode(ppl); + + assertContains(root, "LogicalCorrelate"); + assertContains(root, "Uncollect"); + assertAnyContains(root, "fetch=", "LIMIT", "RowNumber", "Window"); + + String expectedSparkSql = + "SELECT `$cor0`.`DEPTNO`, `$cor0`.`TAGS`, `t1`.`EMPNOS`\n" + + "FROM `scott`.`DEPT` `$cor0`,\n" + + "LATERAL (SELECT `EMPNOS`\n" + + "FROM UNNEST((SELECT `$cor0`.`EMPNOS`\n" + + "FROM (VALUES (0)) `t` (`ZERO`))) `t0` (`EMPNOS`)\n" + + "LIMIT 2) `t1`"; + verifyPPLToSparkSQL(root, expectedSparkSql); + } + + @Test + public void testMvExpandProjectNested() { + String ppl = "source=DEPT | mvexpand EMPNOS | fields DEPTNO, EMPNOS"; + RelNode root = getRelNode(ppl); + + assertContains(root, "LogicalCorrelate"); + assertContains(root, "Uncollect"); + assertContains(root, "LogicalProject"); + + String expectedSparkSql = + "SELECT `$cor0`.`DEPTNO`, `t00`.`EMPNOS`\n" + + "FROM `scott`.`DEPT` `$cor0`,\n" + + "LATERAL UNNEST((SELECT `$cor0`.`EMPNOS`\n" + + "FROM (VALUES (0)) `t` (`ZERO`))) `t00` (`EMPNOS`)"; + verifyPPLToSparkSQL(root, expectedSparkSql); + } + + @Test + public void testMvExpandEmptyOrNullArray() { + RelNode root = getRelNode("source=DEPT | where isnull(EMPNOS) | mvexpand EMPNOS"); + assertContains(root, "LogicalCorrelate"); + assertContains(root, "Uncollect"); + } + + @Test + public void testMvExpandWithDuplicates() { + RelNode root = getRelNode("source=DEPT | where DEPTNO in (10, 10, 20) | mvexpand EMPNOS"); + assertContains(root, "LogicalCorrelate"); + assertContains(root, "Uncollect"); + } + + @Test + public void testMvExpandLargeArray() { + RelNode root = getRelNode("source=DEPT | where DEPTNO = 999 | mvexpand EMPNOS"); + assertContains(root, "LogicalCorrelate"); + assertContains(root, "Uncollect"); + } + + @Test + public void testMvExpandPrimitiveArray() { + RelNode root = getRelNode("source=DEPT | mvexpand EMPNOS"); + assertContains(root, "LogicalCorrelate"); + assertContains(root, "Uncollect"); + } + + @Test + public void testMvExpandInvalidLimitZero() { + String ppl = "source=DEPT | mvexpand EMPNOS limit=0"; + Exception ex = Assert.assertThrows(Exception.class, () -> getRelNode(ppl)); + String msg = String.valueOf(ex.getMessage()); + Assert.assertTrue( + "Expected error message for limit=0. Actual: " + msg, + msg.toLowerCase().contains("limit") || msg.toLowerCase().contains("positive")); + } + + @Test + public void testMvExpandInvalidLimitNegative() { + String ppl = "source=DEPT | mvexpand EMPNOS limit=-1"; + Exception ex = Assert.assertThrows(Exception.class, () -> getRelNode(ppl)); + String msg = String.valueOf(ex.getMessage()); + Assert.assertTrue( + "Expected parsing error for negative limit. Actual: " + msg, + msg.toLowerCase().contains("extraneous") + || msg.toLowerCase().contains("unexpected") + || msg.toLowerCase().contains("expecting")); + } + + @Test + public void testMvExpandNonArrayField() { + String ppl = "source=DEPT | mvexpand DEPTNO"; + RelNode root = getRelNode(ppl); + + Assert.assertNotNull("Query should produce a valid plan", root); + + String plan = root.explain(); + Assert.assertTrue( + "Plan should contain LogicalTableScan", + plan.contains("LogicalTableScan") || plan.contains("LogicalProject")); + + Assert.assertFalse( + "Non-array field should not generate Uncollect operation", plan.contains("Uncollect")); + } + + @Test + public void testMvExpandMultisetField() { + // Test that MULTISET types are handled the same as ARRAY types + // This verifies the fix for the MULTISET handling issue identified in code review + String ppl = "source=DEPT | mvexpand TAGS"; + RelNode root = getRelNode(ppl); + + // MULTISET fields should generate the same plan structure as ARRAY fields + assertContains(root, "LogicalCorrelate"); + assertContains(root, "Uncollect"); + + String expectedSparkSql = + "SELECT `$cor0`.`DEPTNO`, `$cor0`.`EMPNOS`, `t00`.`TAGS`\n" + + "FROM `scott`.`DEPT` `$cor0`,\n" + + "LATERAL UNNEST((SELECT `$cor0`.`TAGS`\n" + + "FROM (VALUES (0)) `t` (`ZERO`))) `t00` (`TAGS`)"; + verifyPPLToSparkSQL(root, expectedSparkSql); + } + + @Test + public void testMvExpandMultisetWithLimit() { + // Test that MULTISET types work correctly with limit parameter + String ppl = "source=DEPT | mvexpand TAGS limit=3"; + RelNode root = getRelNode(ppl); + + assertContains(root, "LogicalCorrelate"); + assertContains(root, "Uncollect"); + assertAnyContains(root, "fetch=", "LIMIT", "RowNumber", "Window"); + + String expectedSparkSql = + "SELECT `$cor0`.`DEPTNO`, `$cor0`.`EMPNOS`, `t1`.`TAGS`\n" + + "FROM `scott`.`DEPT` `$cor0`,\n" + + "LATERAL (SELECT `TAGS`\n" + + "FROM UNNEST((SELECT `$cor0`.`TAGS`\n" + + "FROM (VALUES (0)) `t` (`ZERO`))) `t0` (`TAGS`)\n" + + "LIMIT 3) `t1`"; + verifyPPLToSparkSQL(root, expectedSparkSql); + } + + private static void assertContains(RelNode root, String token) { + String plan = root.explain(); + Assert.assertTrue( + "Expected plan to contain [" + token + "] but got:\n" + plan, plan.contains(token)); + } + + private static void assertAnyContains(RelNode root, String... tokens) { + String plan = root.explain(); + for (String token : tokens) { + if (plan.contains(token)) { + return; + } + } + Assert.fail( + "Expected plan to contain one of " + Arrays.toString(tokens) + " but got:\n" + plan); + } +} diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java index b043c8c5aa3..eefc678f706 100644 --- a/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java +++ b/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java @@ -1042,4 +1042,16 @@ public void testMvcombineCommandWithDelim() { public void testNoMvCommand() { assertEquals("source=table | nomv identifier", anonymize("source=t | nomv firstname")); } + + @Test + public void testMvexpandCommand() { + assertEquals("source=table | mvexpand identifier", anonymize("source=t | mvexpand skills")); + } + + @Test + public void testMvexpandCommandWithLimit() { + assertEquals( + "source=table | mvexpand identifier limit=***", + anonymize("source=t | mvexpand skills limit=5")); + } }