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 9296e74)

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 is positive but exceeds the number of lines, the code clamps it to lines.size() - 1 and uses that as the header index. If lines.size() is 1, hIdx becomes 0, and dataStart becomes 1, which is out of bounds. The loop for (int i = dataStart; i < lines.size(); i++) then never executes, returning an empty list. This is silent data loss when the user specifies a valid forceheader value that happens to exceed the available lines. The user receives no indication that their header line was not found. int hIdx = Math.min(forceHeader - 1, lines.size() - 1); header = splitCols(lines.get(hIdx)); dataStart = hIdx + 1; Possible Issue The visitMultikv method discards the probe build by calling context.relBuilder.build() when the input is text, but does not restore the builder's state. If the builder had accumulated other operations before the probe, those are lost. This can corrupt the plan when multikv is not the first operation in the pipeline. context.relBuilder.build(); context.setProjectVisited(savedProjectVisited); Possible Issue The extract method uses indexOf(KV) to find the key-value separator, but does not handle the case where a column name itself contains the KV character (U+0002). If a column name is "foo\u0002bar", the method will incorrectly split at the first occurrence of KV within the name, returning a partial column name as the key. This breaks extraction for any column whose name contains the separator character. for (String pair : FS_SPLIT.split(record, -1)) { int idx = pair.indexOf(KV); if (idx < 0) { continue; } if (pair.substring(0, idx).equals(col)) { return pair.substring(idx + 1);

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 9296e74

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Escape separator characters in values The serialize method does not escape or validate cell values that might contain the special separators FS or KV. If a cell value contains these characters, the serialized record will be corrupted and extract will return incorrect results. Add escaping or validation to prevent separator injection. core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java [123-135] private static String serialize(List<String> header, List<String> cells) { int n = (header != null) ? Math.max(header.size(), cells.size()) : cells.size(); StringBuilder sb = new StringBuilder(); for (int i = 0; i < n; i++) { String name = (header != null && i < header.size()) ? header.get(i) : ("Column_" + (i + 1)); String val = (i < cells.size()) ? cells.get(i) : ""; + // Escape special separators to prevent injection + val = val.replace(FS, "").replace(KV, ""); if (sb.length() > 0) { sb.append(FS); } sb.append(name).append(KV).append(val); } return sb.toString(); } Suggestion importance[1-10]: 9 __ Why: This identifies a critical data corruption issue where cell values containing FS or KV separators would break the serialization format, causing extract to return incorrect results. This is a correctness bug that could lead to silent data corruption.	High
General	Validate forceHeader bounds When forceHeader is set to a very large value and lines.size() is small, hIdx will be clamped to lines.size() - 1, causing the last line to be treated as the header. This silently changes behavior instead of validating the input. Validate that forceHeader is within bounds and throw an exception if it exceeds the available lines. core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java [65-75] 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; } else { header = splitCols(lines.get(0)); dataStart = 1; } Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies that forceHeader values exceeding available lines are silently clamped rather than validated. This could lead to unexpected behavior. Explicit validation with a clear error message improves correctness and user experience.	Medium
Security	Add line count limit The LINE.split(raw) operation can produce a very large array when raw contains many newlines, potentially causing memory exhaustion. Consider adding a limit on the number of lines processed or implementing streaming line processing to prevent resource exhaustion attacks. core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java [48-61] +private static final int MAX_LINES = 10000; + public static List<String> parse( String raw, int forceHeader, boolean noHeader, List<String> filterTerms) { if (raw == null) { return Collections.emptyList(); } List<String> lines = new ArrayList<>(); for (String l : LINE.split(raw)) { if (!l.trim().isEmpty()) { lines.add(l); + if (lines.size() >= MAX_LINES) { + break; + } } } if (lines.isEmpty()) { return Collections.emptyList(); } Suggestion importance[1-10]: 7 __ Why: The suggestion addresses a potential resource exhaustion vulnerability where processing very large inputs could cause memory issues. Adding a line limit is a reasonable safeguard, though the specific limit value should be configurable.	Medium

---

Previous suggestions

Suggestions up to commit e146e91

Category	Suggestion	Impact
Possible issue	Validate forceHeader bounds explicitly When forceHeader is set to a value greater than the number of lines, the code silently uses the last line as the header. This could lead to incorrect parsing without any indication to the user. Consider validating that forceHeader is within bounds and providing clear feedback. core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java [65-75] 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; } else { header = splitCols(lines.get(0)); dataStart = 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. Explicit validation with a clear error message improves correctness and user experience.	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 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() > 1_000_000) { // Add reasonable 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. However, the hardcoded limit of 1MB may be too restrictive for legitimate use cases and should ideally be configurable.	Medium
General	Optimize column extraction performance The extract method performs a linear search through all field-value pairs for each column lookup. When extracting multiple columns from the same record (common use case), this results in O(n*m) complexity. Consider parsing the record into a map once and reusing it for multiple extractions. core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java [90-104] public static String extract(String record, String col) { if (record == null || col == null) { return null; } + String searchKey = col + KV; for (String pair : FS_SPLIT.split(record, -1)) { - int idx = pair.indexOf(KV); - if (idx < 0) { - continue; - } - if (pair.substring(0, idx).equals(col)) { - return pair.substring(idx + 1); + if (pair.startsWith(searchKey)) { + return pair.substring(searchKey.length()); } } return null; } Suggestion importance[1-10]: 6 __ Why: The optimization using startsWith is a minor improvement that reduces substring operations. However, the performance gain is marginal for typical use cases, and the original implementation is already reasonably efficient for single-column extraction.	Low

Suggestions up to commit e1b3333

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

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

[github-actions]

Persistent review updated to latest commit e146e91

[github-actions]

Persistent review updated to latest commit 9296e74

[songkant-aws]

Avoid Splunk wording in comments.

[songkant-aws]

We don't have such metadata field yet. Is it expected to be explicitly generated by user in makeresults command?

[songkant-aws]

We need to print real output for each example ppl queries.

[songkant-aws]

nit: version 3.9?

[songkant-aws]

Avoid Splunk wording in comments.

[songkant-aws]

Per the RFC discussion, we should call out that users must explicitly cast expanded key-value fields from VARCHAR or ANY to the expected types when type-sensitive processing is required.

Please confirm that this reflects an outcome discussed and agreed upon with users.

[songkant-aws]

Add more complex ITs to let downstream commands correctly consume multikv results after casting

[songkant-aws]

Please also add tests to CalciteExplainIT to assert printed logical plans are correct.

[songkant-aws]

For input types other than ARRAY and MAP, the current implementation silently falls back to the text path and treats them as strings. Should we validate the input type here and reject unsupported types during planning?

[songkant-aws]

The type-based dispatch is reasonable, but the current implementation visits the child twice. The first visit mutates the shared RelBuilder and CalcitePlanContext; text mode then calls build() and only restores isProjectVisited before rebuilding the child. This is not a full rollback and may affect features such as correlation binding or plan-node tracking. Could we build the child once and lower text mode from the resulting RelNode? Is there a specific reason to discard the build in text mode? The schema is known at stack peek and it seems feasible to build text mode specific MvExpand Relnode on top of it.