Add PPL multikv command (fixed-schema)

[noCharger]

Description

Adds the PPL multikv command: a streaming, row-multiplying command that extracts fields from an input field and emits one row per source record. It requires the Calcite (v3) engine.

The input field is selected with field=<name> (defaults to _raw); output columns are declared with fields <col>.... multikv dispatches on the input field's plan-time type:

• Text (VARCHAR): parse table-formatted text (for example ps / top / netstat / df output) into columns. Extracted values are typed string.
• Array of objects (ARRAY<ANY>): explode into one row per element, reading each declared column from the element and preserving its type.
• Single object (MAP<VARCHAR,ANY>): read each declared column, preserving its type; emits one row.

Nested container values are returned serialized (matching the merged makeresults convention); extract deeper fields downstream with spath or another multikv field=<subfield>.

Design, semantics, and deferred scope (runtime auto-header, aligned-offset parsing, filter/rmorig, parse-to-MAP optimization) are in the RFC: #5640.

Changes

• Grammar + Multikv AST + Calcite lowering, including the field=<name> input selector, forceheader, and noheader.
• Type-dispatch in visitMultikv: text → MULTIKV_SPLIT / mvexpand / MULTIKV_EXTRACT; array-of-objects → mvexpand + INTERNAL_ITEM; single object → INTERNAL_ITEM only.
• Bare multikv (no fields, no noheader) is rejected at the semantic layer with actionable guidance (v1 is fixed-schema).
• V2 (non-Calcite) engine returns an unsupported-command error, matching the merged makeresults convention.
• User docs: docs/user/ppl/cmd/multikv.md and an index.md row.

Test coverage

• Unit: CalcitePPLMultikvTest, AstBuilderTest#testMultikvCommand (incl. field=), PPLQueryDataAnonymizerTest.
• IT: CalcitePPLMultikvCommandIT (covering field=, text, array-of-objects, and single-object modes), NewAddedCommandsIT#testMultikv.

Check List

• New functionality includes testing.
• New functionality has been documented.
• Commits are signed per the DCO using --signoff.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit e1b3333)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue In the parse method, when forceHeader > 0, the code computes hIdx = Math.min(forceHeader - 1, lines.size() - 1). If forceHeader exceeds the number of lines, hIdx points to the last line, which becomes the header, and dataStart = hIdx + 1 equals lines.size(), leaving no data rows. This silently returns an empty result instead of signaling that the forced header line does not exist. A user specifying forceheader=10 on a 3-line input would expect an error or at least a warning, not silent success with zero rows. } else if (forceHeader > 0) { int hIdx = Math.min(forceHeader - 1, lines.size() - 1); header = splitCols(lines.get(hIdx)); dataStart = hIdx + 1; Possible Issue In visitMultikv, after probing the child to determine the input field type, the code discards the probe build with context.relBuilder.build() and resets projectVisited for the text-input path. However, if the probe build modified other context state (for example, added operators to the builder stack or altered internal flags), those changes are not fully rolled back. This could leave the builder in an inconsistent state when the text pipeline is built from scratch, potentially causing incorrect plans or runtime errors if the builder stack is not empty or if other context flags are stale. context.relBuilder.build(); context.setProjectVisited(savedProjectVisited);

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to e1b3333
Explore these optional code suggestions:

Category	Suggestion	Impact
General	Validate forceHeader bounds strictly When forceHeader exceeds the number of available lines, the code silently uses the last line as the header. This could lead to incorrect parsing where a data row is misinterpreted as a header. Consider throwing an exception or logging a warning when forceHeader is out of bounds. core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java [65-72] if (noHeader) { header = null; dataStart = 0; } else if (forceHeader > 0) { - int hIdx = Math.min(forceHeader - 1, lines.size() - 1); + if (forceHeader > lines.size()) { + throw new IllegalArgumentException( + "forceheader line " + forceHeader + " exceeds available lines (" + lines.size() + ")"); + } + int hIdx = forceHeader - 1; header = splitCols(lines.get(hIdx)); dataStart = hIdx + 1; } Suggestion importance[1-10]: 8 __ Why: The current code silently uses the last line when forceHeader exceeds available lines, which could lead to incorrect parsing. Strict validation prevents data misinterpretation and provides clear error messages.	Medium
	Use appropriate error code The error code FIELD_NOT_FOUND is misleading for this validation failure. The issue is not that a field is missing, but that the schema cannot be determined at plan time. Use a more appropriate error code like SEMANTIC_CHECK_EXCEPTION or create a specific code for schema resolution failures. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [4503-4512] if (noFields && !node.isNoHeader()) { throw ErrorReport.wrap( new SemanticCheckException( "multikv has no declared output columns. Add an explicit fields clause, for" + " example: multikv fields <col1> <col2>")) - .code(ErrorCode.FIELD_NOT_FOUND) + .code(ErrorCode.SEMANTIC_CHECK_EXCEPTION) .location("while resolving the output schema for multikv") .context("command", "multikv") .build(); } Suggestion importance[1-10]: 5 __ Why: While using SEMANTIC_CHECK_EXCEPTION instead of FIELD_NOT_FOUND is more semantically accurate, this is a minor improvement in error reporting clarity and doesn't affect functionality.	Low
Security	Add input size validation The LINE.split(raw) operation can produce a large array for very large input strings, potentially causing memory issues. Consider using a streaming approach or adding a size limit check before splitting to prevent excessive memory consumption from malicious or malformed input. core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java [48-61] public static List<String> parse( String raw, int forceHeader, boolean noHeader, List<String> filterTerms) { if (raw == null) { return Collections.emptyList(); + } + if (raw.length() > 10_000_000) { // 10MB limit + throw new IllegalArgumentException("Input text exceeds maximum size limit"); } List<String> lines = new ArrayList<>(); for (String l : LINE.split(raw)) { if (!l.trim().isEmpty()) { lines.add(l); } } if (lines.isEmpty()) { return Collections.emptyList(); } Suggestion importance[1-10]: 7 __ Why: Adding input size validation prevents potential memory exhaustion from very large inputs. This is a reasonable security enhancement, though the specific limit value may need tuning based on actual use cases.	Medium

---

Previous suggestions

Suggestions up to commit eedb47b

Category	Suggestion	Impact
General	Validate schema before building child The error is thrown after building the child node (probe), which may have already performed expensive operations. Move this validation earlier in the method, before building the child, to fail fast and avoid unnecessary computation when the query is invalid. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [4495-4519] +List<Field> fields = node.getFields(); +boolean noFields = (fields == null || fields.isEmpty()); + if (noFields && !node.isNoHeader()) { throw ErrorReport.wrap( new SemanticCheckException( "multikv has no declared output columns. Add an explicit fields clause, for" + " example: multikv fields <col1> <col2>")) .code(ErrorCode.FIELD_NOT_FOUND) .location("while resolving the output schema for multikv") .context("command", "multikv") .build(); } +boolean savedProjectVisited = context.isProjectVisited(); +RelNode probe = node.getChild().get(0).accept(this, context); + Suggestion importance[1-10]: 8 __ Why: Moving validation before building the child node is a good optimization that fails fast and avoids unnecessary computation. This improves performance when queries are invalid.	Medium
Security	Add input size validation The LINE.split(raw) operation can produce a large array when processing very large input strings, potentially causing memory issues. Consider using a streaming approach or adding a size limit check before splitting to prevent excessive memory consumption from malicious or malformed input. core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java [48-61] public static List<String> parse( String raw, int forceHeader, boolean noHeader, List<String> filterTerms) { if (raw == null) { return Collections.emptyList(); + } + if (raw.length() > 10_000_000) { // 10MB limit + throw new IllegalArgumentException("Input text exceeds maximum allowed size"); } List<String> lines = new ArrayList<>(); for (String l : LINE.split(raw)) { if (!l.trim().isEmpty()) { lines.add(l); } } if (lines.isEmpty()) { return Collections.emptyList(); } Suggestion importance[1-10]: 7 __ Why: Adding input size validation prevents potential memory exhaustion from very large inputs. However, the hardcoded 10MB limit may be too restrictive for legitimate use cases and should ideally be configurable.	Medium
	Limit maximum field count The FS_SPLIT.split(record, -1) can produce a large array for maliciously crafted records with many field separators, leading to potential memory exhaustion. Add a limit on the number of fields to prevent denial-of-service attacks through excessive field counts. core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java [90-104] +private static final int MAX_FIELDS = 1000; + public static String extract(String record, String col) { if (record == null || col == null) { return null; } - for (String pair : FS_SPLIT.split(record, -1)) { + String[] pairs = FS_SPLIT.split(record, MAX_FIELDS + 1); + if (pairs.length > MAX_FIELDS) { + throw new IllegalArgumentException("Record exceeds maximum field count"); + } + for (String pair : pairs) { int idx = pair.indexOf(KV); if (idx < 0) { continue; } if (pair.substring(0, idx).equals(col)) { return pair.substring(idx + 1); } } return null; } Suggestion importance[1-10]: 7 __ Why: Adding a field count limit prevents potential DoS attacks through excessive field separators. However, the hardcoded limit of 1000 fields may need adjustment based on actual use cases.	Medium

[github-actions]

Persistent review updated to latest commit e1b3333