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 d7b8ceb1bff..ce3d39e53f6 100644 --- a/core/src/main/java/org/opensearch/sql/analysis/Analyzer.java +++ b/core/src/main/java/org/opensearch/sql/analysis/Analyzer.java @@ -561,6 +561,12 @@ public LogicalPlan visitNoMv(NoMv node, AnalysisContext context) { throw getOnlyForCalciteException("nomv"); } + @Override + public LogicalPlan visitOutputLookup( + org.opensearch.sql.ast.tree.OutputLookup node, AnalysisContext context) { + throw getOnlyForCalciteException("outputlookup"); + } + @Override public LogicalPlan visitMvExpand(MvExpand node, AnalysisContext context) { throw getOnlyForCalciteException("mvexpand"); 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 6d8415fd7ea..aeb5734b7c5 100644 --- a/core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java +++ b/core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java @@ -294,6 +294,10 @@ public T visitReverse(Reverse node, C context) { return visitChildren(node, context); } + public T visitOutputLookup(org.opensearch.sql.ast.tree.OutputLookup node, C context) { + return visitChildren(node, context); + } + public T visitTranspose(Transpose node, C context) { return visitChildren(node, context); } diff --git a/core/src/main/java/org/opensearch/sql/ast/tree/OutputLookup.java b/core/src/main/java/org/opensearch/sql/ast/tree/OutputLookup.java new file mode 100644 index 00000000000..04a75ff09d0 --- /dev/null +++ b/core/src/main/java/org/opensearch/sql/ast/tree/OutputLookup.java @@ -0,0 +1,65 @@ +/* + * 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 lombok.AllArgsConstructor; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import lombok.Setter; +import lombok.ToString; +import org.opensearch.sql.ast.AbstractNodeVisitor; + +/** + * AST node for the {@code outputlookup} command: a terminal write sink that materializes pipeline + * rows into a lookup index. Overwrite-by-default (set {@code append=true} to append instead). See + * the outputlookup PPL design. + */ +@Getter +@Setter +@ToString +@EqualsAndHashCode(callSuper = false) +@RequiredArgsConstructor +@AllArgsConstructor +public class OutputLookup extends UnresolvedPlan { + + private UnresolvedPlan child; + + /** Destination lookup name (a plain index). */ + private final String indexName; + + /** false (default) overwrites the destination; true appends to it. */ + private boolean append = false; + + /** true (default) clears the destination on empty results; false keeps it. */ + private boolean overrideIfEmpty = true; + + /** + * Fields whose values form the document {@code _id} for upsert; empty means auto-generated id. + */ + private List keyFields = java.util.List.of(); + + /** Cap on the number of rows written; null means unbounded. */ + private Integer max; + + @Override + public OutputLookup 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.visitOutputLookup(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 13ccdf15197..d73fee64c55 100644 --- a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java +++ b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java @@ -919,6 +919,67 @@ public RelNode visitReverse( return context.relBuilder.peek(); } + @Override + public RelNode visitOutputLookup( + org.opensearch.sql.ast.tree.OutputLookup node, CalcitePlanContext context) { + visitChildren(node, context); + String indexName = node.getIndexName(); + if (indexName.startsWith(".")) { + throw new IllegalArgumentException( + "outputlookup destination [" + indexName + "] must not be a system (dot-prefixed) index"); + } + RelNode child = context.relBuilder.build(); + // Validate key_field names against the result schema: a missing/misspelled key field would + // otherwise encode identically for every row and collapse them into a single document. + if (!node.getKeyFields().isEmpty()) { + java.util.List resultFields = child.getRowType().getFieldNames(); + for (String keyField : node.getKeyFields()) { + if (!resultFields.contains(keyField)) { + throw new IllegalArgumentException( + "outputlookup key_field [" + + keyField + + "] is not a field of the result; available fields: " + + resultFields); + } + } + } + org.apache.calcite.plan.RelOptTable sourceTable = findSourceTable(child); + if (sourceTable == null) { + throw new IllegalArgumentException( + "outputlookup requires an OpenSearch source scan in the pipeline"); + } + org.apache.calcite.plan.RelOptSchema relOptSchema = context.relBuilder.getRelOptSchema(); + if (!(relOptSchema instanceof org.apache.calcite.prepare.Prepare.CatalogReader)) { + throw new IllegalStateException("outputlookup could not obtain a Calcite catalog reader"); + } + RelNode sink = + OutputLookupTableModify.create( + child, + sourceTable, + (org.apache.calcite.prepare.Prepare.CatalogReader) relOptSchema, + indexName, + node.isAppend(), + node.isOverrideIfEmpty(), + node.getKeyFields(), + node.getMax()); + context.relBuilder.push(sink); + return context.relBuilder.peek(); + } + + /** Walk down to the first scan carrying a table, so the write rule can reach the client. */ + private static org.apache.calcite.plan.RelOptTable findSourceTable(RelNode rel) { + if (rel.getTable() != null) { + return rel.getTable(); + } + for (RelNode input : rel.getInputs()) { + org.apache.calcite.plan.RelOptTable found = findSourceTable(input); + if (found != null) { + return found; + } + } + return null; + } + @Override public RelNode visitTranspose( org.opensearch.sql.ast.tree.Transpose node, CalcitePlanContext context) { diff --git a/core/src/main/java/org/opensearch/sql/calcite/OutputLookupTableModify.java b/core/src/main/java/org/opensearch/sql/calcite/OutputLookupTableModify.java new file mode 100644 index 00000000000..ada91d185d8 --- /dev/null +++ b/core/src/main/java/org/opensearch/sql/calcite/OutputLookupTableModify.java @@ -0,0 +1,94 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.calcite; + +import java.util.List; +import lombok.Getter; +import org.apache.calcite.plan.Convention; +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.prepare.Prepare.CatalogReader; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.TableModify; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Terminal write node for outputlookup, modeled as a Calcite {@link TableModify} INSERT so the + * optimizer treats it as a mandatory table-modifying side effect (never dropped or reordered) and + * it exposes the standard rowcount row type. The referenced table is the source scan, used only so + * the lowering rule can reach the in-cluster client; the real destination lookup index is created + * and written at execution time by the physical operator, so a preexisting destination table is not + * required. Subclasses {@link TableModify} rather than LogicalTableModify so the built-in + * EnumerableTableModifyRule (which needs a ModifiableTable) does not fire on it. + */ +@Getter +public class OutputLookupTableModify extends TableModify { + + private final String indexName; + private final boolean append; + private final boolean overrideIfEmpty; + private final java.util.List keyFields; + private final @Nullable Integer max; + + public OutputLookupTableModify( + RelOptCluster cluster, + RelTraitSet traits, + RelOptTable table, + CatalogReader catalogReader, + RelNode input, + String indexName, + boolean append, + boolean overrideIfEmpty, + java.util.List keyFields, + @Nullable Integer max) { + super(cluster, traits, table, catalogReader, input, Operation.INSERT, null, null, false); + this.indexName = indexName; + this.append = append; + this.overrideIfEmpty = overrideIfEmpty; + this.keyFields = keyFields; + this.max = max; + } + + public static OutputLookupTableModify create( + RelNode input, + RelOptTable table, + CatalogReader catalogReader, + String indexName, + boolean append, + boolean overrideIfEmpty, + java.util.List keyFields, + @Nullable Integer max) { + RelOptCluster cluster = input.getCluster(); + RelTraitSet traits = cluster.traitSetOf(Convention.NONE); + return new OutputLookupTableModify( + cluster, + traits, + table, + catalogReader, + input, + indexName, + append, + overrideIfEmpty, + keyFields, + max); + } + + @Override + public RelNode copy(RelTraitSet traitSet, List inputs) { + return new OutputLookupTableModify( + getCluster(), + traitSet, + getTable(), + getCatalogReader(), + inputs.get(0), + indexName, + append, + overrideIfEmpty, + keyFields, + max); + } +} diff --git a/docs/category.json b/docs/category.json index 04f2bbae22e..7c06bde6903 100644 --- a/docs/category.json +++ b/docs/category.json @@ -24,6 +24,7 @@ "user/ppl/cmd/head.md", "user/ppl/cmd/join.md", "user/ppl/cmd/lookup.md", + "user/ppl/cmd/outputlookup.md", "user/ppl/cmd/mvcombine.md", "user/ppl/cmd/nomv.md", "user/ppl/cmd/mvexpand.md", diff --git a/docs/user/ppl/cmd/outputlookup.md b/docs/user/ppl/cmd/outputlookup.md new file mode 100644 index 00000000000..e7f78921b79 --- /dev/null +++ b/docs/user/ppl/cmd/outputlookup.md @@ -0,0 +1,95 @@ + +# outputlookup + +The `outputlookup` command is a terminal write sink: it materializes the current pipeline result into a lookup index and returns a single `rows_written` count. Use it to build or refresh a lookup (dimension) dataset from a search, which can then be read back with `source=` or enriched into other searches with the `lookup` command. + +A lookup name refers to a plain index. Overwrite replaces that index with the current result (its schema is replaced each run); append adds the result to it. Writes are eventually consistent: a reader during an overwrite may briefly see the lookup being rebuilt. + +## Syntax + +The `outputlookup` command has the following syntax: + +```syntax +outputlookup [append=] [override_if_empty=] [key_field=(, )*] [max=] +``` + +The following are examples of the `outputlookup` command syntax: + +```syntax +source = table | outputlookup my_lookup +source = table | where status = 'active' | outputlookup active_hosts +source = table | stats count by region | outputlookup region_counts +source = table | fields id, name | outputlookup append=true my_lookup +source = table | outputlookup override_if_empty=false my_lookup +source = table | fields id, name | outputlookup key_field=id my_lookup +source = table | fields region, host, val | outputlookup key_field=region, host my_lookup +source = table | outputlookup max=1000 sample_lookup +``` + +## Parameters + +The `outputlookup` command supports the following parameters. + +| Parameter | Required/Optional | Description | +| --- | --- | --- | +| `` | Required | The lookup name to write to. It is a plain index, created on demand if it does not exist and replaced on overwrite. Lookup indices are tagged with a `_meta.lookup` marker; if the name is an existing **non-lookup** index (no marker), the command refuses rather than overwriting it — delete it first to reuse the name. If the name is a filtered alias created by the OpenSearch Dashboards data importer, overwrite migrates it onto a dedicated plain index (the shared index is never deleted). | +| `append` | Optional | When `false` (default), overwrites the lookup with the result. When `true`, appends the result to the existing lookup. Because OpenSearch is schemaless, appended rows may introduce new fields. Default is `false`. | +| `override_if_empty` | Optional | When `true` (default), an empty result clears the existing lookup. When `false`, an empty result leaves the existing lookup intact. Default is `true`. | +| `key_field` | Optional | One or more fields (comma-separated) used as the upsert key. Rows are written by a deterministic `_id` derived from the key values, so re-running the same command updates matching rows in place instead of creating duplicates. Setting `key_field` implies `append=true`. Every field listed must be a field of the result; a multivalue key value is rejected. | +| `max` | Optional | Caps the number of rows written. | + +## Example 1: Building a lookup from an aggregation + +The following query writes per-region counts into a lookup and returns the number of rows written: + +```ppl ignore +source = events + | stats count as cnt by region + | outputlookup region_counts +``` + +The query returns the following result: + +```text ++--------------+ +| rows_written | +|--------------| +| 3 | ++--------------+ +``` + +The lookup can then be read back: + +```ppl ignore +source = region_counts +``` + +## Example 2: Idempotent upsert with `key_field` + +The following query upserts by `id`, so re-running it updates existing rows instead of duplicating them: + +```ppl ignore +source = users + | fields id, name, department + | outputlookup key_field=id users_lookup +``` + +Running the command a second time with overlapping `id` values leaves the row count unchanged for the overlapping keys and only inserts genuinely new keys. + +## Example 3: Preserving a lookup on an empty result + +The following query refreshes `error_hosts` only when the search returns rows; an empty result keeps the previous lookup intact: + +```ppl ignore +source = logs + | where level = 'ERROR' + | fields host + | outputlookup override_if_empty=false error_hosts +``` + +## Limitations + +- `outputlookup` is a terminal command: it returns a `rows_written` count rather than forwarding the input rows. +- The destination is always a lookup index; there is no file output target. +- Overwrite is eventually consistent: it deletes and recreates the index, so a concurrent read during the rebuild may briefly see the lookup absent, and an interrupted overwrite may leave it partially written. +- The write executes under the caller's security context. The caller needs write, create-index, delete, and get privileges on the destination (and alias privileges only when migrating a data-importer lookup); no cluster-level privilege is required. diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLOutputLookupIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLOutputLookupIT.java new file mode 100644 index 00000000000..fd6949b69e1 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLOutputLookupIT.java @@ -0,0 +1,505 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.calcite.remote; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.util.Locale; +import org.json.JSONArray; +import org.json.JSONObject; +import org.junit.jupiter.api.Test; +import org.opensearch.client.Request; +import org.opensearch.client.Response; +import org.opensearch.client.ResponseException; +import org.opensearch.sql.ppl.PPLIntegTestCase; + +/** + * End-to-end tests for the PPL {@code outputlookup} command (substrate C2: one plain index per + * lookup, weak/eventual overwrite) on the internal integ-test cluster. A lookup {@code } is a + * plain index; overwrite deletes and recreates it from the current result, append bulk-writes into + * it, and the command returns a single {@code rows_written} count (terminal, not pass-through). A + * name that is a #11303 filtered alias is migrated onto a dedicated plain index on overwrite. + */ +public class CalcitePPLOutputLookupIT extends PPLIntegTestCase { + + @Override + public void init() throws Exception { + super.init(); + enableCalcite(); + } + + private void seedSrc(String src) throws IOException { + Request create = new Request("PUT", "/" + src); + create.setJsonEntity( + "{\"mappings\":{\"properties\":{\"id\":{\"type\":\"integer\"}," + + "\"name\":{\"type\":\"keyword\"},\"status\":{\"type\":\"integer\"}}}}"); + client().performRequest(create); + indexDoc(src, 1, "alice", 200); + indexDoc(src, 2, "bob", 500); + indexDoc(src, 3, "carol", 404); + } + + private void indexDoc(String index, int id, String name, int status) throws IOException { + Request req = new Request("PUT", "/" + index + "/_doc/" + id + "?refresh=true"); + req.setJsonEntity( + String.format(Locale.ROOT, "{\"id\":%d,\"name\":\"%s\",\"status\":%d}", id, name, status)); + client().performRequest(req); + } + + private void refresh(String index) throws IOException { + client().performRequest(new Request("POST", "/" + index + "/_refresh")); + } + + /** rows_written count returned by the outputlookup query. */ + private long rowsWritten(JSONObject result) { + return result.getJSONArray("datarows").getJSONArray(0).getLong(0); + } + + private long docCount(String indexOrAlias) throws IOException { + Response resp = client().performRequest(new Request("GET", "/" + indexOrAlias + "/_count")); + return new JSONObject(new String(resp.getEntity().getContent().readAllBytes())) + .getLong("count"); + } + + private String mappingJson(String indexOrAlias) throws IOException { + Response resp = client().performRequest(new Request("GET", "/" + indexOrAlias + "/_mapping")); + return new String(resp.getEntity().getContent().readAllBytes()); + } + + // Overwrite returns the count, replaces rows, and drops fields absent from the new result. + @Test + public void testOverwriteReturnsCountReplacesAndDropsAbsentFields() throws IOException { + String src = "olkc_src_ow"; + String dest = "olkc_dest_ow"; + seedSrc(src); + + JSONObject r1 = + executeQuery(String.format("source=%s | fields id, name | outputlookup %s", src, dest)); + assertEquals("returns rows_written count", 3L, rowsWritten(r1)); + refresh(dest); + assertEquals(3L, docCount(dest)); + assertTrue("name present after first write", mappingJson(dest).contains("\"name\"")); + assertTrue("lookup marker set on create", mappingJson(dest).contains("_meta")); + + JSONObject r2 = + executeQuery( + String.format("source=%s | where id=1 | fields id | outputlookup %s", src, dest)); + assertEquals(1L, rowsWritten(r2)); + refresh(dest); + assertEquals("overwrite replaced all rows", 1L, docCount(dest)); + assertFalse("absent field dropped by fresh backing", mappingJson(dest).contains("\"name\"")); + } + + // A6#1: append keeps existing rows AND keeps a new field (OpenSearch is schemaless). + @Test + public void testAppendKeepsNewFieldUnlikeSplunk() throws IOException { + String src = "olkc_src_ap"; + String dest = "olkc_dest_ap"; + seedSrc(src); + + executeQuery(String.format("source=%s | where id<3 | fields id | outputlookup %s", src, dest)); + refresh(dest); + assertEquals(2L, docCount(dest)); + assertFalse(mappingJson(dest).contains("\"name\"")); + + JSONObject r = + executeQuery( + String.format( + "source=%s | where id=3 | fields id, name | outputlookup append=true %s", + src, dest)); + assertEquals(1L, rowsWritten(r)); + refresh(dest); + assertEquals("append adds without clearing", 3L, docCount(dest)); + assertTrue("new field kept (schemaless)", mappingJson(dest).contains("\"name\"")); + } + + // override_if_empty=false leaves the existing lookup intact on an empty result. + @Test + public void testOverrideIfEmptyFalseKeeps() throws IOException { + String src = "olkc_src_oif"; + String dest = "olkc_dest_oif"; + seedSrc(src); + + executeQuery(String.format("source=%s | fields id | outputlookup %s", src, dest)); + refresh(dest); + assertEquals(3L, docCount(dest)); + + JSONObject r = + executeQuery( + String.format( + "source=%s | where id=999 | outputlookup override_if_empty=false %s", src, dest)); + assertEquals("empty write reports 0", 0L, rowsWritten(r)); + refresh(dest); + assertEquals("empty guard kept existing lookup", 3L, docCount(dest)); + } + + // Default override_if_empty=true clears the lookup on an empty result. + @Test + public void testOverrideIfEmptyTrueClears() throws IOException { + String src = "olkc_src_oit"; + String dest = "olkc_dest_oit"; + seedSrc(src); + + executeQuery(String.format("source=%s | fields id | outputlookup %s", src, dest)); + refresh(dest); + assertEquals(3L, docCount(dest)); + + executeQuery(String.format("source=%s | where id=999 | outputlookup %s", src, dest)); + refresh(dest); + assertEquals("empty result cleared the lookup", 0L, docCount(dest)); + } + + // key_field upserts by id (append-default) — re-run updates in place, no duplicates. + @Test + public void testKeyFieldUpsertNoDuplicate() throws IOException { + String src = "olkc_src_kf"; + String dest = "olkc_dest_kf"; + seedSrc(src); + + executeQuery( + String.format( + "source=%s | where id<3 | fields id, name | outputlookup key_field=id %s", src, dest)); + refresh(dest); + assertEquals(2L, docCount(dest)); + + executeQuery( + String.format( + "source=%s | where id=1 or id=3 | fields id, name | outputlookup key_field=id %s", + src, dest)); + refresh(dest); + assertEquals("upsert updates in place, no duplicate", 3L, docCount(dest)); + } + + // max caps rows written (and the returned count). + @Test + public void testMaxCapsRows() throws IOException { + String src = "olkc_src_max"; + String dest = "olkc_dest_max"; + seedSrc(src); + + JSONObject r = + executeQuery(String.format("source=%s | fields id | outputlookup max=2 %s", src, dest)); + assertEquals("max caps the count", 2L, rowsWritten(r)); + refresh(dest); + assertEquals("max caps written rows", 2L, docCount(dest)); + } + + // A6#2: a multivalue field is preserved as a native array. + @Test + public void testMultivaluePreservedAsArray() throws IOException { + String src = "olkc_src_mv"; + String dest = "olkc_dest_mv"; + Request createSrc = new Request("PUT", "/" + src); + createSrc.setJsonEntity( + "{\"mappings\":{\"properties\":{\"id\":{\"type\":\"integer\"}," + + "\"tags\":{\"type\":\"keyword\"}}}}"); + client().performRequest(createSrc); + Request doc = new Request("PUT", "/" + src + "/_doc/1?refresh=true"); + doc.setJsonEntity("{\"id\":1,\"tags\":[\"x\",\"y\",\"z\"]}"); + client().performRequest(doc); + + executeQuery(String.format("source=%s | fields id, tags | outputlookup %s", src, dest)); + refresh(dest); + + Response resp = client().performRequest(new Request("GET", "/" + dest + "/_search")); + JSONObject json = new JSONObject(new String(resp.getEntity().getContent().readAllBytes())); + JSONObject source = + json.getJSONObject("hits").getJSONArray("hits").getJSONObject(0).getJSONObject("_source"); + assertTrue("tags present", source.has("tags")); + JSONArray tags = source.getJSONArray("tags"); + assertEquals("multivalue preserved as a native array", 3, tags.length()); + } + + // B: source larger than the 10k result window is written in full (no silent truncation). + @Test + public void testLargeSourceWritesPastQuerySizeLimit() throws IOException { + String src = "olkc_src_big"; + String dest = "olkc_dest_big"; + Request create = new Request("PUT", "/" + src); + create.setJsonEntity( + "{\"settings\":{\"index.max_result_window\":10000}," + + "\"mappings\":{\"properties\":{\"id\":{\"type\":\"integer\"}}}}"); + client().performRequest(create); + + int total = 12000; // > default max_result_window (10000): only PIT paging reaches all rows + StringBuilder bulk = new StringBuilder(); + for (int i = 1; i <= total; i++) { + bulk.append( + String.format(Locale.ROOT, "{\"index\":{\"_index\":\"%s\",\"_id\":%d}}%n", src, i)); + bulk.append(String.format(Locale.ROOT, "{\"id\":%d}%n", i)); + if (i % 3000 == 0 || i == total) { + Request r = new Request("POST", "/_bulk?refresh=true"); + r.setJsonEntity(bulk.toString()); + client().performRequest(r); + bulk.setLength(0); + } + } + + JSONObject res = + executeQuery(String.format("source=%s | fields id | outputlookup %s", src, dest)); + assertEquals("writes all rows past the 10k window", (long) total, rowsWritten(res)); + refresh(dest); + assertEquals("destination holds all rows", (long) total, docCount(dest)); + } + + // A7 idempotency end-to-end: a composite (multi-field) key_field upserts by the pair, so + // re-running updates matching pairs in place with no duplicates. + @Test + public void testMultiFieldKeyFieldUpsertNoDuplicate() throws IOException { + String src = "olkc_src_mkf"; + String dest = "olkc_dest_mkf"; + Request create = new Request("PUT", "/" + src); + create.setJsonEntity( + "{\"mappings\":{\"properties\":{\"region\":{\"type\":\"keyword\"}," + + "\"host\":{\"type\":\"keyword\"},\"val\":{\"type\":\"integer\"}}}}"); + client().performRequest(create); + indexRegionHost(src, 1, "us", "h1", 10); + indexRegionHost(src, 2, "us", "h2", 20); + indexRegionHost(src, 3, "eu", "h1", 30); + + // Run 1: upsert (us,h1) and (us,h2) -> 2 pairs. + executeQuery( + String.format( + "source=%s | where val<25 | fields region, host, val | outputlookup" + + " key_field=region,host %s", + src, dest)); + refresh(dest); + assertEquals(2L, docCount(dest)); + + // Run 2: host=h1 matches (us,h1) [update in place] and (eu,h1) [insert] -> 3 distinct pairs. + executeQuery( + String.format( + "source=%s | where host='h1' | fields region, host, val | outputlookup" + + " key_field=region,host %s", + src, dest)); + refresh(dest); + assertEquals( + "composite-key upsert: (us,h1) updated, (eu,h1) inserted, no duplicate", + 3L, + docCount(dest)); + } + + // P0 #1: a key_field that is not a result field is rejected (would otherwise collapse all rows + // into one document). + @Test + public void testMissingKeyFieldRejected() throws IOException { + String src = "olkc_src_kfbad"; + seedSrc(src); + ResponseException ex = + assertThrows( + ResponseException.class, + () -> + executeQuery( + String.format( + "source=%s | fields id | outputlookup key_field=does_not_exist %s", + src, "olkc_dest_kfbad"))); + assertTrue( + "error should name the offending key_field", + ex.getMessage().contains("key_field") && ex.getMessage().contains("does_not_exist")); + } + + // #1 marker guard: overwrite refuses a name that is an existing NON-lookup index (no + // `_meta.lookup` marker), so it never clobbers unrelated data. Also covers dest == source in + // shape. + @Test + public void testOverwriteRefusesForeignPlainIndex() throws IOException { + String src = "olkc_src_coll"; + String dest = "olkc_dest_coll"; + seedSrc(src); + // Pre-create dest as a plain, unmarked index with unrelated content. + Request create = new Request("PUT", "/" + dest); + create.setJsonEntity("{\"mappings\":{\"properties\":{\"other\":{\"type\":\"keyword\"}}}}"); + client().performRequest(create); + Request doc = new Request("PUT", "/" + dest + "/_doc/x?refresh=true"); + doc.setJsonEntity("{\"other\":\"keep\"}"); + client().performRequest(doc); + + ResponseException ex = + assertThrows( + ResponseException.class, + () -> + executeQuery(String.format("source=%s | fields id | outputlookup %s", src, dest))); + assertTrue( + "error should flag the non-lookup index", ex.getMessage().contains("non-lookup index")); + refresh(dest); + assertEquals("foreign index left untouched", 1L, docCount(dest)); + assertTrue("foreign content preserved", mappingJson(dest).contains("\"other\"")); + } + + // #1 marker guard: append likewise refuses a non-lookup index. + @Test + public void testAppendRefusesForeignPlainIndex() throws IOException { + String src = "olkc_src_apf"; + String dest = "olkc_dest_apf"; + seedSrc(src); + Request create = new Request("PUT", "/" + dest); + create.setJsonEntity("{\"mappings\":{\"properties\":{\"other\":{\"type\":\"keyword\"}}}}"); + client().performRequest(create); + Request doc = new Request("PUT", "/" + dest + "/_doc/x?refresh=true"); + doc.setJsonEntity("{\"other\":\"keep\"}"); + client().performRequest(doc); + + ResponseException ex = + assertThrows( + ResponseException.class, + () -> + executeQuery( + String.format( + "source=%s | fields id | outputlookup append=true %s", src, dest))); + assertTrue("append refused on non-lookup index", ex.getMessage().contains("non-lookup index")); + refresh(dest); + assertEquals("foreign index left untouched", 1L, docCount(dest)); + } + + // A multivalue field cannot be a key_field: it has no deterministic single-value _id encoding, so + // it is rejected (F3). (Under C2 there is no backing index / orphan concept to assert.) + @Test + public void testMultivalueKeyFieldRejected() throws IOException { + String src = "olkc_src_orph"; + Request createSrc = new Request("PUT", "/" + src); + createSrc.setJsonEntity( + "{\"mappings\":{\"properties\":{\"id\":{\"type\":\"integer\"}," + + "\"tags\":{\"type\":\"keyword\"}}}}"); + client().performRequest(createSrc); + Request doc = new Request("PUT", "/" + src + "/_doc/1?refresh=true"); + doc.setJsonEntity("{\"id\":1,\"tags\":[\"x\",\"y\",\"z\"]}"); + client().performRequest(doc); + + assertThrows( + ResponseException.class, + () -> + executeQuery( + String.format( + "source=%s | fields id, tags | outputlookup key_field=tags %s", + src, "olkc_dest_orph"))); + } + + // Backward-compat with the Dashboards data importer (#11303): a lookup created there is a + // FILTERED alias ({term:{__lookup:}}) over a SHARED index holding several lookups. + // Overwrite + // must migrate the touched lookup onto a dedicated backing by repointing the alias, and must NOT + // delete the shared index (that would destroy the other lookups sharing it). + @Test + public void testOverwriteMigratesDataImporterLookupWithoutDeletingSharedIndex() + throws IOException { + String src = "olkc_src_mig"; + String shared = "olkc_shared_imp"; + String aliasA = "olkc_impa"; + String aliasB = "olkc_impb"; + seedSrc(src); // 3 rows: alice/bob/carol + + // Shared importer-style index: 2 docs for lookup A, 2 for lookup B, each tagged with __lookup. + Request createShared = new Request("PUT", "/" + shared); + createShared.setJsonEntity( + "{\"mappings\":{\"properties\":{\"__lookup\":{\"type\":\"keyword\"}," + + "\"host_name\":{\"type\":\"keyword\"}}}}"); + client().performRequest(createShared); + Request bulk = new Request("POST", "/" + shared + "/_bulk?refresh=true"); + bulk.setJsonEntity( + "{\"index\":{}}\n{\"__lookup\":\"A\",\"host_name\":\"a1\"}\n" + + "{\"index\":{}}\n{\"__lookup\":\"A\",\"host_name\":\"a2\"}\n" + + "{\"index\":{}}\n{\"__lookup\":\"B\",\"host_name\":\"b1\"}\n" + + "{\"index\":{}}\n{\"__lookup\":\"B\",\"host_name\":\"b2\"}\n"); + client().performRequest(bulk); + + // Two filtered aliases over the shared index (the #11303 lookup shape). + Request aliases = new Request("POST", "/_aliases"); + aliases.setJsonEntity( + "{\"actions\":[" + + "{\"add\":{\"index\":\"" + + shared + + "\",\"alias\":\"" + + aliasA + + "\",\"filter\":{\"term\":{\"__lookup\":\"A\"}}}}," + + "{\"add\":{\"index\":\"" + + shared + + "\",\"alias\":\"" + + aliasB + + "\",\"filter\":{\"term\":{\"__lookup\":\"B\"}}}}]}"); + client().performRequest(aliases); + assertEquals("alias A sees only its slice before overwrite", 2L, docCount(aliasA)); + + // Overwrite lookup A with a fresh 3-row result. + JSONObject res = + executeQuery(String.format("source=%s | fields id, name | outputlookup %s", src, aliasA)); + assertEquals(3L, rowsWritten(res)); + refresh(aliasA); + + // The shared index must still exist and lookup B must be untouched. + Response cat = + client().performRequest(new Request("GET", "/_cat/indices/" + shared + "?format=json")); + JSONArray sharedIdx = new JSONArray(new String(cat.getEntity().getContent().readAllBytes())); + assertEquals("shared importer index must NOT be deleted", 1, sharedIdx.length()); + assertEquals("other lookup B on the shared index is intact", 2L, docCount(aliasB)); + + // Lookup A is migrated: the name is now a dedicated PLAIN INDEX (no alias, no __ol_ backing) + // holding the 3 new rows. _cat/indices/ returns exactly that index (an alias would + // instead resolve to a differently-named backing). + Response migrated = + client().performRequest(new Request("GET", "/_cat/indices/" + aliasA + "?format=json")); + JSONArray migratedIdx = + new JSONArray(new String(migrated.getEntity().getContent().readAllBytes())); + assertEquals("lookup A migrated onto a single dedicated index", 1, migratedIdx.length()); + assertEquals( + "migrated name is now a concrete index (not an alias over a backing)", + aliasA, + migratedIdx.getJSONObject(0).getString("index")); + assertEquals("name A now resolves to the migrated 3-row result", 3L, docCount(aliasA)); + } + + // Append onto a shared/filtered (data-importer) lookup is refused: writing our rows there would + // be invisible through the filtered alias or mutate a shared index. Overwrite migrates first. + @Test + public void testAppendToDataImporterLookupRefused() throws IOException { + String src = "olkc_src_appmig"; + String shared = "olkc_shared_appmig"; + String alias = "olkc_impc"; + seedSrc(src); + + Request createShared = new Request("PUT", "/" + shared); + createShared.setJsonEntity( + "{\"mappings\":{\"properties\":{\"__lookup\":{\"type\":\"keyword\"}," + + "\"host_name\":{\"type\":\"keyword\"}}}}"); + client().performRequest(createShared); + Request doc = new Request("PUT", "/" + shared + "/_doc/1?refresh=true"); + doc.setJsonEntity("{\"__lookup\":\"C\",\"host_name\":\"c1\"}"); + client().performRequest(doc); + Request aliases = new Request("POST", "/_aliases"); + aliases.setJsonEntity( + "{\"actions\":[{\"add\":{\"index\":\"" + + shared + + "\",\"alias\":\"" + + alias + + "\",\"filter\":{\"term\":{\"__lookup\":\"C\"}}}}]}"); + client().performRequest(aliases); + + ResponseException ex = + assertThrows( + ResponseException.class, + () -> + executeQuery( + String.format( + "source=%s | fields id, name | outputlookup append=true %s", src, alias))); + assertTrue( + "append onto a shared/filtered lookup should be refused with migration guidance", + ex.getMessage().contains("shared or") || ex.getMessage().contains("migrate")); + // The shared index is untouched by the refused append. + assertEquals(1L, docCount(alias)); + } + + private void indexRegionHost(String index, int id, String region, String host, int val) + throws IOException { + Request req = new Request("PUT", "/" + index + "/_doc/" + id + "?refresh=true"); + req.setJsonEntity( + String.format( + Locale.ROOT, "{\"region\":\"%s\",\"host\":\"%s\",\"val\":%d}", region, host, val)); + client().performRequest(req); + } +} diff --git a/integ-test/src/test/java/org/opensearch/sql/security/OutputLookupPermissionsIT.java b/integ-test/src/test/java/org/opensearch/sql/security/OutputLookupPermissionsIT.java new file mode 100644 index 00000000000..55df32bac6d --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/security/OutputLookupPermissionsIT.java @@ -0,0 +1,63 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.security; + +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.util.Locale; +import org.junit.Test; +import org.opensearch.client.Request; +import org.opensearch.client.ResponseException; + +/** + * Verifies that outputlookup writes run under the calling user's permissions (not a privileged + * plugin context). A read-only user can read the source but has no create, alias, or write + * permission on the destination, so the command must be denied. If the write silently ran as a + * system identity, this query would succeed, which would be a privilege escalation. + */ +public class OutputLookupPermissionsIT extends SecurityTestBase { + + private static final String SRC = "olk_perm_src"; + private static final String DEST = "olk_perm_dest"; + private static final String RO_ROLE = "olk_ro_role"; + private static final String RO_USER = "olk_ro_user"; + + private boolean initialized = false; + + @Override + protected void init() throws Exception { + super.init(); + if (!initialized) { + Request create = new Request("PUT", "/" + SRC); + create.setJsonEntity("{\"mappings\":{\"properties\":{\"id\":{\"type\":\"integer\"}}}}"); + client().performRequest(create); + Request doc = new Request("PUT", "/" + SRC + "/_doc/1?refresh=true"); + doc.setJsonEntity("{\"id\":1}"); + client().performRequest(doc); + + // Read-only role over the source: search + PIT + ppl, but no write/create/alias. + createRoleWithIndexAccess(RO_ROLE, SRC); + createUser(RO_USER, RO_ROLE); + initialized = true; + } + enableCalcite(); + allowCalciteFallback(); + } + + @Test + public void writeDeniedForReadOnlyUser() { + String query = String.format(Locale.ROOT, "source=%s | fields id | outputlookup %s", SRC, DEST); + ResponseException ex = + assertThrows(ResponseException.class, () -> executeQueryAsUser(query, RO_USER)); + String body = ex.getMessage(); + assertTrue( + "denial should be a security/permissions error, got: " + body, + body.contains("security_exception") + || body.contains("no permissions") + || body.contains("unauthorized")); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/EnumerableOutputLookupRule.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/EnumerableOutputLookupRule.java new file mode 100644 index 00000000000..09d7728389f --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/EnumerableOutputLookupRule.java @@ -0,0 +1,75 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.planner.rules; + +import org.apache.calcite.adapter.enumerable.EnumerableConvention; +import org.apache.calcite.plan.Convention; +import org.apache.calcite.plan.RelOptRuleCall; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.convert.ConverterRule; +import org.opensearch.sql.calcite.OutputLookupTableModify; +import org.opensearch.sql.opensearch.storage.OpenSearchIndex; +import org.opensearch.sql.opensearch.storage.write.EnumerableOutputLookup; +import org.opensearch.transport.client.node.NodeClient; + +/** + * Lowers an {@link OutputLookupTableModify} whose referenced table is an {@link OpenSearchIndex} + * into the physical {@link EnumerableOutputLookup}, wiring in the in-cluster node client. + */ +public class EnumerableOutputLookupRule extends ConverterRule { + + public static final Config DEFAULT_CONFIG = + Config.INSTANCE + .as(Config.class) + .withConversion( + OutputLookupTableModify.class, + Convention.NONE, + EnumerableConvention.INSTANCE, + "EnumerableOutputLookupRule") + .withRuleFactory(EnumerableOutputLookupRule::new); + + protected EnumerableOutputLookupRule(Config config) { + super(config); + } + + @Override + public boolean matches(RelOptRuleCall call) { + return nodeClientOf((OutputLookupTableModify) call.rel(0)) != null; + } + + @Override + public RelNode convert(RelNode rel) { + OutputLookupTableModify node = (OutputLookupTableModify) rel; + NodeClient client = nodeClientOf(node); + RelTraitSet traitSet = node.getTraitSet().replace(EnumerableConvention.INSTANCE); + RelNode convertedInput = + convert( + node.getInput(), node.getInput().getTraitSet().replace(EnumerableConvention.INSTANCE)); + return new EnumerableOutputLookup( + node.getCluster(), + traitSet, + convertedInput, + node.getRowType(), + client, + node.getIndexName(), + node.isAppend(), + node.isOverrideIfEmpty(), + node.getKeyFields(), + node.getMax()); + } + + private static NodeClient nodeClientOf(OutputLookupTableModify node) { + if (node.getTable() == null) { + return null; + } + OpenSearchIndex index = node.getTable().unwrap(OpenSearchIndex.class); + if (index == null) { + return null; + } + return index.getClient().getNodeClient().orElse(null); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/OpenSearchIndexRules.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/OpenSearchIndexRules.java index c200ffa8909..b9945846368 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/OpenSearchIndexRules.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/OpenSearchIndexRules.java @@ -18,6 +18,9 @@ public class OpenSearchIndexRules { EnumerableNestedAggregateRule.DEFAULT_CONFIG.toRule(); private static final RelOptRule GRAPH_LOOKUP_RULE = EnumerableGraphLookupRule.DEFAULT_CONFIG.toRule(); + // outputlookup: lower the terminal write node (LogicalOutputLookup) to the OpenSearch write op. + private static final RelOptRule OUTPUT_LOOKUP_RULE = + EnumerableOutputLookupRule.DEFAULT_CONFIG.toRule(); // Rule that always pushes down relevance functions regardless of pushdown settings private static final RelevanceFunctionPushdownRule RELEVANCE_FUNCTION_RULE = RelevanceFunctionPushdownRule.Config.DEFAULT.toRule(); @@ -29,6 +32,7 @@ public class OpenSearchIndexRules { CATALOG_SCAN_RULE, NESTED_AGGREGATE_RULE, GRAPH_LOOKUP_RULE, + OUTPUT_LOOKUP_RULE, RELEVANCE_FUNCTION_RULE); private static final ProjectIndexScanRule PROJECT_INDEX_SCAN = diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/EnumerableOutputLookup.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/EnumerableOutputLookup.java new file mode 100644 index 00000000000..fc4dc98eba9 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/EnumerableOutputLookup.java @@ -0,0 +1,142 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.write; + +import java.lang.reflect.Method; +import java.util.List; +import java.util.Map; +import lombok.Getter; +import org.apache.calcite.adapter.enumerable.EnumerableRel; +import org.apache.calcite.adapter.enumerable.EnumerableRelImplementor; +import org.apache.calcite.adapter.enumerable.PhysType; +import org.apache.calcite.adapter.enumerable.PhysTypeImpl; +import org.apache.calcite.linq4j.Enumerable; +import org.apache.calcite.linq4j.Linq4j; +import org.apache.calcite.linq4j.tree.BlockBuilder; +import org.apache.calcite.linq4j.tree.Expression; +import org.apache.calcite.linq4j.tree.Expressions; +import org.apache.calcite.linq4j.tree.Types; +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.RelWriter; +import org.apache.calcite.rel.SingleRel; +import org.apache.calcite.rel.type.RelDataType; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.opensearch.sql.opensearch.storage.write.WriteConfig.WriteMode; +import org.opensearch.transport.client.node.NodeClient; + +/** + * Physical terminal write operator for {@code outputlookup}. Drains the child rows, materializes + * them into the lookup via {@link OutputLookupWriteExec}, and emits a single {@code BIGINT} row + * carrying the written count. + */ +@Getter +public class EnumerableOutputLookup extends SingleRel implements EnumerableRel { + + private static final Method WRITE_AND_COUNT = + Types.lookupMethod(EnumerableOutputLookup.class, "writeAndCount", Enumerable.class); + + private final transient NodeClient client; + private final RelDataType modifyRowType; + private final String indexName; + private final boolean append; + private final boolean overrideIfEmpty; + private final List keyFields; + private final @Nullable Integer max; + private final List fields; + private final Map mapping; + + public EnumerableOutputLookup( + RelOptCluster cluster, + RelTraitSet traits, + RelNode input, + RelDataType modifyRowType, + NodeClient client, + String indexName, + boolean append, + boolean overrideIfEmpty, + List keyFields, + @Nullable Integer max) { + super(cluster, traits, input); + this.modifyRowType = modifyRowType; + this.client = client; + this.indexName = indexName; + this.append = append; + this.overrideIfEmpty = overrideIfEmpty; + this.keyFields = keyFields; + this.max = max; + this.fields = input.getRowType().getFieldNames(); + this.mapping = OutputLookupWriteExec.inferMapping(input.getRowType()); + } + + @Override + protected RelDataType deriveRowType() { + return modifyRowType; + } + + /** + * Surface the write parameters in explain output so the physical plan shows the write target and + * mode (the write itself is performed at runtime by {@link OutputLookupWriteExec}, which is not a + * relational operator and therefore never appears as a plan node). Including these in the digest + * also keeps two outputlookups with different targets/modes from being deduplicated by the + * planner. + */ + @Override + public RelWriter explainTerms(RelWriter pw) { + return super.explainTerms(pw) + .item("dest", indexName) + .itemIf("key_field", keyFields, !keyFields.isEmpty()) + .item("append", append) + .item("override_if_empty", overrideIfEmpty) + .itemIf("max", max, max != null); + } + + @Override + public RelNode copy(RelTraitSet traitSet, List inputs) { + return new EnumerableOutputLookup( + getCluster(), + traitSet, + inputs.get(0), + modifyRowType, + client, + indexName, + append, + overrideIfEmpty, + keyFields, + max); + } + + @Override + public Result implement(EnumerableRelImplementor implementor, Prefer pref) { + BlockBuilder builder = new BlockBuilder(); + Result inputResult = implementor.visitChild(this, 0, (EnumerableRel) getInput(), pref); + PhysType physType = + PhysTypeImpl.of(implementor.getTypeFactory(), getRowType(), pref.preferArray()); + Expression inputEnumerable = builder.append("inputEnumerable", inputResult.block); + Expression op = implementor.stash(this, EnumerableOutputLookup.class); + builder.add(Expressions.return_(null, Expressions.call(op, WRITE_AND_COUNT, inputEnumerable))); + return implementor.result(physType, builder.toBlock()); + } + + /** Runtime entry: drain input, materialize into the lookup, emit one row with the count. */ + public Enumerable<@Nullable Object> writeAndCount(Enumerable<@Nullable Object> input) { + WriteMode mode = keyFields.isEmpty() ? WriteMode.APPEND : WriteMode.UPSERT; + long count = + OutputLookupWriteExec.execute( + client, + indexName, + fields, + mapping, + mode, + keyFields, + max, + overrideIfEmpty, + append, + input.enumerator()); + return Linq4j.asEnumerable(List.of((Object) count)); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java new file mode 100644 index 00000000000..5f363fc1ee0 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoder.java @@ -0,0 +1,92 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.write; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.Base64; +import java.util.Collection; +import java.util.List; + +/** + * Computes a deterministic document {@code _id} from the {@code key_field} values of a row, so + * re-running an upsert updates the same document instead of duplicating it. + * + *

The id is {@code base64url(SHA-256(canonical(keyValues)))}: a fixed 43-character string, well + * under the 512-byte id limit regardless of key size. The canonical encoding is a sequence of + * length-prefixed, type-tagged tokens (one per key field), so multi-field keys cannot collide + * across different field boundaries, and empty string, null, and a missing field are all distinct. + * Multivalue key fields are rejected. The raw key values still live in {@code _source}, so the + * lookup stays queryable by value. + */ +public final class LookupIdEncoder { + + private LookupIdEncoder() {} + + /** Marker for a key field that is not present in the row schema (distinct from a null value). */ + private static final Object MISSING = new Object(); + + public static String encode(List keyFields, List fields, Object[] row) { + MessageDigest digest = sha256(); + for (String keyField : keyFields) { + int idx = fields.indexOf(keyField); + Object value = (idx >= 0 && idx < row.length) ? row[idx] : MISSING; + if (value == null || value == MISSING) { + digest.update((byte) 'N'); // null and missing both encode as N, distinct from any V token + continue; + } + if (value instanceof Object[] || value instanceof Collection) { + throw new IllegalArgumentException( + "outputlookup key_field [" + keyField + "] must not be multivalue"); + } + byte[] bytes = canonical(value).getBytes(StandardCharsets.UTF_8); + digest.update((byte) 'V'); + digest.update(typeTag(value)); + digest.update(intToBytes(bytes.length)); // length prefix makes multi-field keys unambiguous + digest.update(bytes); + } + return Base64.getUrlEncoder().withoutPadding().encodeToString(digest.digest()); + } + + private static String canonical(Object value) { + if (value instanceof Boolean) { + return value.toString(); + } + if (value instanceof Float || value instanceof Double) { + return Double.toString(((Number) value).doubleValue()); + } + if (value instanceof Number) { + return Long.toString(((Number) value).longValue()); + } + return String.valueOf(value); + } + + private static byte typeTag(Object value) { + if (value instanceof Boolean) { + return 'b'; + } + if (value instanceof Float || value instanceof Double) { + return 'd'; + } + if (value instanceof Number) { + return 'i'; + } + return 's'; + } + + private static byte[] intToBytes(int v) { + return new byte[] {(byte) (v >>> 24), (byte) (v >>> 16), (byte) (v >>> 8), (byte) v}; + } + + private static MessageDigest sha256() { + try { + return MessageDigest.getInstance("SHA-256"); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 not available", e); + } + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java new file mode 100644 index 00000000000..67fb4afb36e --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriter.java @@ -0,0 +1,166 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.write; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.opensearch.action.bulk.BackoffPolicy; +import org.opensearch.action.bulk.BulkItemResponse; +import org.opensearch.action.bulk.BulkRequest; +import org.opensearch.action.bulk.BulkResponse; +import org.opensearch.action.index.IndexRequest; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.rest.RestStatus; +import org.opensearch.sql.opensearch.storage.OpenSearchIndex; +import org.opensearch.sql.opensearch.storage.write.WriteConfig.WriteMode; +import org.opensearch.sql.opensearch.storage.write.WriteResult.ItemFailure; +import org.opensearch.transport.client.node.NodeClient; + +/** + * Batched, retrying bulk writer. Pure write-side: it consumes rows, maps them to documents, and + * indexes them via the native bulk path. It owns no read, PIT, or index-lifecycle concern (create, + * replace, delete all live in the calling command). + * + *

It retries only backpressure (429) items under exponential backoff. Any non-429 item failure, + * or a 429 that survives the retry budget, is raised as a {@link BulkWriteException} rather than + * silently dropped, so callers never lose rows without knowing. + */ +public final class OpenSearchBulkWriter implements AutoCloseable { + + private final NodeClient client; + private final WriteConfig cfg; + + private BulkRequest batch = new BulkRequest(); + private int buffered = 0; + private long written = 0; + private final List tolerated = new ArrayList<>(); + private boolean closed = false; + + public OpenSearchBulkWriter(NodeClient client, WriteConfig cfg) { + this.client = client; + this.cfg = cfg; + } + + /** Buffer a row; when the batch fills, a synchronous bulk flush fires. May throw on failure. */ + public void add(Object[] row) { + ensureOpen(); + batch.add(toIndexRequest(row)); + if (++buffered >= cfg.batchSize()) { + flushBatch(); + } + } + + /** Flush the tail batch and return the cumulative result. */ + public WriteResult flush() { + ensureOpen(); + if (buffered > 0) { + flushBatch(); + } + return new WriteResult(written, List.copyOf(tolerated)); + } + + /** Idempotent: flushes any buffered rows, then blocks further writes. */ + @Override + public void close() { + if (closed) { + return; + } + flush(); + closed = true; + } + + private void ensureOpen() { + if (closed) { + throw new IllegalStateException("writer is closed"); + } + } + + private void flushBatch() { + BulkRequest pending = batch; + pending.setRefreshPolicy(cfg.refresh()); + batch = new BulkRequest(); + buffered = 0; + + Iterator backoff = BackoffPolicy.exponentialBackoff().iterator(); + while (true) { + BulkResponse response = client.bulk(pending).actionGet(); + if (!response.hasFailures()) { + written += pending.numberOfActions(); + return; + } + + BulkRequest retry = new BulkRequest(); + retry.setRefreshPolicy(cfg.refresh()); + List fatal = new ArrayList<>(); + int succeeded = 0; + for (BulkItemResponse item : response.getItems()) { + if (!item.isFailed()) { + succeeded++; + } else if (item.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS) { + retry.add(pending.requests().get(item.getItemId())); + } else { + fatal.add(new ItemFailure(item.getItemId(), item.getId(), item.getFailureMessage())); + } + } + written += succeeded; + + // G2: never swallow a non-retryable failure. + if (!fatal.isEmpty()) { + throw new BulkWriteException("outputlookup bulk write hit non-retryable failures", fatal); + } + if (retry.numberOfActions() == 0) { + return; + } + if (!backoff.hasNext()) { + List exhausted = new ArrayList<>(); + for (int i = 0; i < retry.numberOfActions(); i++) { + exhausted.add(new ItemFailure(i, null, "429 retry budget exhausted")); + } + throw new BulkWriteException("outputlookup bulk write exhausted 429 retries", exhausted); + } + try { + Thread.sleep(backoff.next().millis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new BulkWriteException("interrupted during bulk retry backoff", List.of()); + } + pending = retry; + } + } + + private IndexRequest toIndexRequest(Object[] row) { + Map doc = new LinkedHashMap<>(); + List fields = cfg.fields(); + for (int i = 0; i < fields.size() && i < row.length; i++) { + String name = fields.get(i); + if (row[i] != null && !OpenSearchIndex.METADATAFIELD_TYPE_MAP.containsKey(name)) { + doc.put(name, row[i]); + } + } + IndexRequest request = new IndexRequest(cfg.destIndex()).source(doc); + if (cfg.mode() == WriteMode.UPSERT) { + request.id(LookupIdEncoder.encode(cfg.keyFields(), fields, row)); + } + return request; + } + + /** Raised when a bulk write has failures that were not resolved by retry. Carries the details. */ + public static final class BulkWriteException extends RuntimeException { + private final transient List failures; + + public BulkWriteException(String message, List failures) { + super(message + ": " + failures); + this.failures = failures; + } + + public List getFailures() { + return failures; + } + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java new file mode 100644 index 00000000000..5113b36b701 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExec.java @@ -0,0 +1,348 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.write; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeField; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.opensearch.action.admin.indices.alias.IndicesAliasesRequest; +import org.opensearch.action.admin.indices.alias.IndicesAliasesRequest.AliasActions; +import org.opensearch.action.admin.indices.alias.get.GetAliasesRequest; +import org.opensearch.action.admin.indices.create.CreateIndexRequest; +import org.opensearch.action.admin.indices.delete.DeleteIndexRequest; +import org.opensearch.action.admin.indices.get.GetIndexRequest; +import org.opensearch.action.admin.indices.get.GetIndexResponse; +import org.opensearch.action.support.WriteRequest.RefreshPolicy; +import org.opensearch.cluster.metadata.AliasMetadata; +import org.opensearch.cluster.metadata.MappingMetadata; +import org.opensearch.index.IndexNotFoundException; +import org.opensearch.sql.opensearch.storage.OpenSearchIndex; +import org.opensearch.sql.opensearch.storage.write.WriteConfig.WriteMode; +import org.opensearch.transport.client.node.NodeClient; + +/** + * Lifecycle for {@code outputlookup} (substrate C2: one plain index per lookup, + * weak/eventual overwrite). A lookup named {@code } is a plain index; readers + * ({@code source=}, {@code lookup }) hit it directly. There is no alias, backing index, + * or uuid. + * + *

Every lookup index this command creates is tagged with the index metadata marker {@code + * _meta.lookup = true}. Overwrite and append refuse to touch a plain index that lacks the marker + * (i.e. a non-lookup index the user created for another purpose), so the command never clobbers + * unrelated data; delete such an index first to reuse the name. + * + *

Consistency contract — weak/eventual (aligned with the merged Dashboards importer #11303, + * which also writes lookups incrementally). This command does not gold-plate read-continuity: + * + *

    + *
  • overwrite ({@code append=false}) = delete the index and recreate it from the current + * result, then bulk write. The schema is fully replaced each run. There is a brief window + * where {@code } is absent (a concurrent read sees index-not-found) and a crash + * mid-overwrite can lose the lookup — both within the weak/eventual contract; we deliberately + * do not reintroduce a blue/green alias swap to close that window (see the {@code + * feature/ppl-outputlookup-alias-swap} branch for the strong-consistency alternative). + *
  • append ({@code append=true}) = bulk into the existing index. Incrementally visible, + * at-least-once; a keyed append is idempotent on re-run, a plain append may duplicate. + *
+ * + *

Backward compatibility with the Dashboards data importer (#11303). That feature + * realizes a lookup as a filtered alias ({@code {term:{__lookup:}}}) over a + * shared index. The two formats coexist permanently; a lookup is migrated only when a name + * collision forces it. Because a plain index cannot coexist with an alias of the same name, + * overwrite of a filtered alias migrates on touch: it detaches the alias from the shared + * index and creates a dedicated plain index {@code } in its place. It never deletes the + * shared index (other lookups' filtered aliases still use it); the orphaned {@code __lookup} + * slice is left in the shared index. Reclaiming that slice (a periodic reaper keyed on the {@code + * __lookup} discriminant that deletes a slice only once no alias references it) is intentionally + * out of scope here and tracked as a follow-up PR. Append onto a filtered alias is refused + * (overwrite once to migrate first). Reads of an un-migrated filtered alias remain compatible. + * + *

Permissions. All operations run under the calling user's security context (verified: a + * read-only user is denied). On the destination the caller needs {@code indices:data/write/bulk}, + * {@code indices:admin/create}, {@code indices:admin/delete}, and {@code indices:admin/get} (to + * probe the target's state / lookup marker) — plus {@code indices:admin/aliases} only on the #11303 + * migration path. No cluster-level permission is required. + */ +public final class OutputLookupWriteExec { + + private static final int BATCH_SIZE = 1000; + + /** Index-metadata marker distinguishing a lookup this command owns from an arbitrary index. */ + private static final String META_KEY = "_meta"; + + private static final String LOOKUP_MARKER = "lookup"; + + private OutputLookupWriteExec() {} + + /** Map the child row schema to an OpenSearch mapping. Pure and unit-testable. */ + public static Map inferMapping(RelDataType rowType) { + Map properties = new LinkedHashMap<>(); + for (RelDataTypeField field : rowType.getFieldList()) { + // Skip reserved metadata fields (_id, _routing, ...); OpenSearch rejects them in a mapping, + // and the writer already excludes them from the document source. + if (OpenSearchIndex.METADATAFIELD_TYPE_MAP.containsKey(field.getName())) { + continue; + } + properties.put( + field.getName(), Map.of("type", esType(field.getType().getSqlTypeName().name()))); + } + return Map.of("properties", properties); + } + + private static String esType(String sqlTypeName) { + switch (sqlTypeName) { + case "TINYINT": + case "SMALLINT": + case "INTEGER": + case "BIGINT": + return "long"; + case "FLOAT": + case "REAL": + case "DOUBLE": + case "DECIMAL": + return "double"; + case "BOOLEAN": + return "boolean"; + case "DATE": + case "TIME": + case "TIMESTAMP": + return "date"; + default: + return "keyword"; + } + } + + /** + * Drain the input, materialize it into the lookup per the outputlookup options, and return the + * number of rows written. + */ + public static long execute( + NodeClient client, + String name, + List fields, + Map mapping, + WriteMode mode, + List keyFields, + @Nullable Integer max, + boolean overrideIfEmpty, + boolean append, + Enumerator<@Nullable Object> input) { + + List rows = new ArrayList<>(); + while (input.moveNext()) { + if (max != null && rows.size() >= max) { + break; + } + Object cur = input.current(); + rows.add(cur instanceof Object[] arr ? arr : new Object[] {cur}); + } + + Target target = resolveTarget(client, name); + + if (append) { + switch (target.kind()) { + case ABSENT: + createIndex(client, name, mapping); + break; + case INDEX: + if (!target.ownLookup()) { + throw foreignIndexError(name); + } + // Append into our existing lookup index. + break; + case ALIAS: + throw appendToAliasError(name, target.filtered()); + default: + throw new IllegalStateException("unreachable target kind: " + target.kind()); + } + writeRows(client, name, fields, mode, keyFields, rows); + return rows.size(); + } + + // Overwrite: on an empty result, honor override_if_empty before touching anything. + if (rows.isEmpty() && !overrideIfEmpty) { + return 0; + } + switch (target.kind()) { + case ABSENT: + createIndex(client, name, mapping); + break; + case INDEX: + if (!target.ownLookup()) { + throw foreignIndexError(name); + } + // Weak/eventual replace: drop our old lookup index and recreate it with the current schema. + // There is a brief absent window (contract-permitted); we do not blue/green it. + deleteIndex(client, name); + createIndex(client, name, mapping); + break; + case ALIAS: + if (!target.filtered()) { + // A plain (non-filtered) alias is not a recognized lookup; do not silently take it over. + throw new IllegalArgumentException( + "outputlookup destination [" + + name + + "] is an alias, not a lookup; refusing to" + + " overwrite it"); + } + // Migrate-on-touch (#11303): detach the filtered alias from its (possibly shared) index — + // NEVER delete that index — then create a dedicated plain index in its place. The orphaned + // __lookup slice is left for a follow-up reaper. + removeAlias(client, name, target.aliasIndices()); + createIndex(client, name, mapping); + break; + default: + throw new IllegalStateException("unreachable target kind: " + target.kind()); + } + writeRows(client, name, fields, mode, keyFields, rows); + return rows.size(); + } + + private static IllegalArgumentException foreignIndexError(String name) { + return new IllegalArgumentException( + "outputlookup destination [" + + name + + "] is an existing non-lookup index (no lookup marker); refusing to overwrite it." + + " Delete it first to reuse the name for a lookup"); + } + + private static IllegalArgumentException appendToAliasError(String name, boolean filtered) { + if (filtered) { + return new IllegalArgumentException( + "outputlookup cannot append to lookup [" + + name + + "]: it is an alias over a shared index (e.g. a data-importer lookup); overwrite it" + + " once to migrate it to a dedicated index, then append"); + } + return new IllegalArgumentException( + "outputlookup cannot append to [" + + name + + "]: it is an alias, not a lookup index; overwrite it first"); + } + + private static void writeRows( + NodeClient client, + String index, + List fields, + WriteMode mode, + List keyFields, + List rows) { + WriteConfig cfg = + new WriteConfig(index, fields, mode, keyFields, BATCH_SIZE, RefreshPolicy.IMMEDIATE); + try (OpenSearchBulkWriter writer = new OpenSearchBulkWriter(client, cfg)) { + for (Object[] row : rows) { + writer.add(row); + } + } + } + + /** Create the lookup index, tagging it with the {@code _meta.lookup} ownership marker. */ + private static void createIndex(NodeClient client, String index, Map mapping) { + Map withMarker = new LinkedHashMap<>(mapping); + withMarker.put(META_KEY, Map.of(LOOKUP_MARKER, true)); + client.admin().indices().create(new CreateIndexRequest(index).mapping(withMarker)).actionGet(); + } + + private static void deleteIndex(NodeClient client, String index) { + try { + client.admin().indices().delete(new DeleteIndexRequest(index)).actionGet(); + } catch (IndexNotFoundException ignored) { + // already gone + } + } + + /** Detach the alias {@code name} from the given indices (used on the #11303 migration path). */ + private static void removeAlias(NodeClient client, String alias, List indices) { + IndicesAliasesRequest req = new IndicesAliasesRequest(); + for (String index : indices) { + req.addAliasAction(AliasActions.remove().index(index).alias(alias)); + } + client.admin().indices().aliases(req).actionGet(); + } + + /** What the lookup name currently resolves to. */ + private enum Kind { + /** The name does not exist. */ + ABSENT, + /** The name is a plain concrete index. */ + INDEX, + /** The name is an alias (e.g. a #11303 filtered alias over a shared index). */ + ALIAS + } + + /** + * Resolved state of the lookup name. {@code ownLookup} is meaningful for {@link Kind#INDEX} (does + * it carry our marker); {@code filtered} and {@code aliasIndices} for {@link Kind#ALIAS}. + */ + private record Target( + Kind kind, boolean ownLookup, boolean filtered, List aliasIndices) {} + + /** + * Probe the target at the indices level (no {@code cluster:monitor/state}): an alias is detected + * via get-aliases, a concrete index and its lookup marker via get-index. + */ + private static Target resolveTarget(NodeClient client, String name) { + // Alias? get-aliases filters by alias name, so a non-empty result means is an alias. + Map> aliases = getAliases(client, name); + List aliasIndices = new ArrayList<>(); + boolean filtered = false; + for (Map.Entry> e : aliases.entrySet()) { + for (AliasMetadata m : e.getValue()) { + if (name.equals(m.alias())) { + aliasIndices.add(e.getKey()); + if (m.filter() != null) { + filtered = true; + } + } + } + } + if (!aliasIndices.isEmpty()) { + return new Target(Kind.ALIAS, false, filtered, aliasIndices); + } + // Concrete index? get-index returns its mapping (with our marker) or throws if absent. + GetIndexResponse index = getIndex(client, name); + if (index == null) { + return new Target(Kind.ABSENT, false, false, List.of()); + } + return new Target(Kind.INDEX, hasLookupMarker(index, name), false, List.of()); + } + + private static Map> getAliases(NodeClient client, String name) { + try { + return client + .admin() + .indices() + .getAliases(new GetAliasesRequest(name)) + .actionGet() + .getAliases(); + } catch (IndexNotFoundException e) { + return Map.of(); + } + } + + private static @Nullable GetIndexResponse getIndex(NodeClient client, String name) { + try { + return client.admin().indices().getIndex(new GetIndexRequest().indices(name)).actionGet(); + } catch (IndexNotFoundException e) { + return null; + } + } + + /** True iff the concrete index {@code name} carries our {@code _meta.lookup = true} marker. */ + private static boolean hasLookupMarker(GetIndexResponse index, String name) { + MappingMetadata mapping = index.getMappings().get(name); + if (mapping == null) { + return false; + } + Object meta = mapping.getSourceAsMap().get(META_KEY); + return meta instanceof Map && Boolean.TRUE.equals(((Map) meta).get(LOOKUP_MARKER)); + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/WriteConfig.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/WriteConfig.java new file mode 100644 index 00000000000..7a83dd47acc --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/WriteConfig.java @@ -0,0 +1,39 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.write; + +import java.util.List; +import org.opensearch.action.support.WriteRequest.RefreshPolicy; + +/** + * Configuration for {@link OpenSearchBulkWriter}. Describes the destination, the row schema, and + * how rows map to documents. Pure write-side concern: it carries no read, PIT, or index-lifecycle + * state. + */ +public record WriteConfig( + String destIndex, + List fields, + WriteMode mode, + List keyFields, + int batchSize, + RefreshPolicy refresh) { + + public enum WriteMode { + /** Auto-generated document id: every row is a new document. */ + APPEND, + /** Explicit document id derived from {@code keyFields}: same key updates in place (upsert). */ + UPSERT + } + + public WriteConfig { + if (mode == WriteMode.UPSERT && (keyFields == null || keyFields.isEmpty())) { + throw new IllegalArgumentException("UPSERT mode requires at least one keyField"); + } + if (batchSize <= 0) { + throw new IllegalArgumentException("batchSize must be positive"); + } + } +} diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/WriteResult.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/WriteResult.java new file mode 100644 index 00000000000..4d124971361 --- /dev/null +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/write/WriteResult.java @@ -0,0 +1,19 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.write; + +import java.util.List; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Cumulative outcome of a {@link OpenSearchBulkWriter} run: rows written and any tolerated per-item + * failures. + */ +public record WriteResult(long written, List failures) { + + /** A single failed bulk item. */ + public record ItemFailure(int batchPos, @Nullable String id, String reason) {} +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoderTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoderTest.java new file mode 100644 index 00000000000..38fbbc1ded1 --- /dev/null +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/write/LookupIdEncoderTest.java @@ -0,0 +1,68 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.write; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.List; +import org.junit.jupiter.api.Test; + +class LookupIdEncoderTest { + + private static final List FIELDS = List.of("a", "b"); + + private String id(Object... row) { + return LookupIdEncoder.encode(List.of("a", "b"), FIELDS, row); + } + + @Test + void deterministicAndFixedLength() { + String first = id("alice", 42); + assertEquals(first, id("alice", 42), "same key -> same id (idempotent upsert)"); + assertEquals(43, first.length(), "base64url SHA-256 is a bounded 43-char id"); + } + + @Test + void multiFieldNoCollisionAcrossBoundaries() { + // ["a","bc"] must not collide with ["ab","c"] -- length prefixes keep field boundaries + // distinct. + assertNotEquals( + LookupIdEncoder.encode(List.of("a", "b"), FIELDS, new Object[] {"a", "bc"}), + LookupIdEncoder.encode(List.of("a", "b"), FIELDS, new Object[] {"ab", "c"})); + } + + @Test + void emptyDistinctFromNullAndMissingCollapsed() { + String nullVal = LookupIdEncoder.encode(List.of("a"), List.of("a"), new Object[] {null}); + String emptyVal = LookupIdEncoder.encode(List.of("a"), List.of("a"), new Object[] {""}); + String missing = LookupIdEncoder.encode(List.of("a"), List.of("x"), new Object[] {"v"}); + assertNotEquals(nullVal, emptyVal, "null must differ from empty string"); + assertNotEquals(emptyVal, missing, "empty string must differ from a missing field"); + // By design (A7), null and missing both encode as N, so they collapse to the same id. + assertEquals(nullVal, missing, "null and missing both encode as N"); + } + + @Test + void typedValuesDoNotCollideWithStrings() { + // integer 1 vs string "1" -- type tag keeps them apart. + assertNotEquals( + LookupIdEncoder.encode(List.of("a"), List.of("a"), new Object[] {1}), + LookupIdEncoder.encode(List.of("a"), List.of("a"), new Object[] {"1"})); + } + + @Test + void multivalueKeyRejected() { + assertThrows( + IllegalArgumentException.class, + () -> + LookupIdEncoder.encode(List.of("a"), List.of("a"), new Object[] {new Object[] {1, 2}})); + assertThrows( + IllegalArgumentException.class, + () -> LookupIdEncoder.encode(List.of("a"), List.of("a"), new Object[] {List.of(1, 2)})); + } +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriterTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriterTest.java new file mode 100644 index 00000000000..56a59d9af25 --- /dev/null +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/write/OpenSearchBulkWriterTest.java @@ -0,0 +1,128 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.write; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.List; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.opensearch.action.DocWriteRequest; +import org.opensearch.action.bulk.BulkItemResponse; +import org.opensearch.action.bulk.BulkRequest; +import org.opensearch.action.bulk.BulkResponse; +import org.opensearch.action.index.IndexRequest; +import org.opensearch.action.support.WriteRequest.RefreshPolicy; +import org.opensearch.common.action.ActionFuture; +import org.opensearch.core.rest.RestStatus; +import org.opensearch.sql.opensearch.storage.write.WriteConfig.WriteMode; +import org.opensearch.transport.client.node.NodeClient; + +@ExtendWith(MockitoExtension.class) +class OpenSearchBulkWriterTest { + + @Mock private NodeClient client; + @Mock private ActionFuture future; + + private WriteConfig cfg(WriteMode mode, List keyFields, int batchSize) { + return new WriteConfig( + "dest", List.of("id", "name"), mode, keyFields, batchSize, RefreshPolicy.NONE); + } + + private BulkResponse success() { + return new BulkResponse(new BulkItemResponse[0], 1L); + } + + @Test + void batchesAtBatchSizeAndCountsAllRows() { + when(client.bulk(any(BulkRequest.class))).thenReturn(future); + when(future.actionGet()).thenReturn(success()); + + ArgumentCaptor captor = ArgumentCaptor.forClass(BulkRequest.class); + long written; + try (OpenSearchBulkWriter writer = + new OpenSearchBulkWriter(client, cfg(WriteMode.APPEND, List.of(), 1000))) { + for (int i = 0; i < 2500; i++) { + writer.add(new Object[] {i, "n" + i}); + } + written = writer.flush().written(); + } + + // 1000 + 1000 + 500 -> three bulk calls, all rows counted. + verify(client, times(3)).bulk(captor.capture()); + assertEquals(1000, captor.getAllValues().get(0).numberOfActions()); + assertEquals(500, captor.getAllValues().get(2).numberOfActions()); + assertEquals(2500L, written); + } + + @Test + void upsertUsesKeyFieldAsExplicitId() { + when(client.bulk(any(BulkRequest.class))).thenReturn(future); + when(future.actionGet()).thenReturn(success()); + + ArgumentCaptor captor = ArgumentCaptor.forClass(BulkRequest.class); + try (OpenSearchBulkWriter writer = + new OpenSearchBulkWriter(client, cfg(WriteMode.UPSERT, List.of("id"), 1000))) { + writer.add(new Object[] {42, "alice"}); + writer.flush(); + } + + verify(client).bulk(captor.capture()); + IndexRequest req = (IndexRequest) captor.getValue().requests().get(0); + assertEquals( + LookupIdEncoder.encode(List.of("id"), List.of("id", "name"), new Object[] {42, "alice"}), + req.id()); + assertEquals(43, req.id().length()); + } + + @Test + void appendLeavesIdUnset() { + when(client.bulk(any(BulkRequest.class))).thenReturn(future); + when(future.actionGet()).thenReturn(success()); + + ArgumentCaptor captor = ArgumentCaptor.forClass(BulkRequest.class); + try (OpenSearchBulkWriter writer = + new OpenSearchBulkWriter(client, cfg(WriteMode.APPEND, List.of(), 1000))) { + writer.add(new Object[] {42, "alice"}); + writer.flush(); + } + + verify(client).bulk(captor.capture()); + IndexRequest req = (IndexRequest) captor.getValue().requests().get(0); + org.junit.jupiter.api.Assertions.assertNull(req.id()); + } + + @Test + void nonRetryableFailureThrowsInsteadOfSwallowing() { + BulkItemResponse.Failure failure = + new BulkItemResponse.Failure( + "dest", "1", new IllegalArgumentException("mapping error"), RestStatus.BAD_REQUEST); + BulkItemResponse item = new BulkItemResponse(0, DocWriteRequest.OpType.INDEX, failure); + BulkResponse withFailure = new BulkResponse(new BulkItemResponse[] {item}, 1L); + when(client.bulk(any(BulkRequest.class))).thenReturn(future); + when(future.actionGet()).thenReturn(withFailure); + + OpenSearchBulkWriter writer = + new OpenSearchBulkWriter(client, cfg(WriteMode.APPEND, List.of(), 1000)); + writer.add(new Object[] {1, "a"}); + OpenSearchBulkWriter.BulkWriteException ex = + assertThrows(OpenSearchBulkWriter.BulkWriteException.class, writer::flush); + assertEquals(1, ex.getFailures().size()); + } + + @Test + void upsertConfigRequiresKeyField() { + assertThrows(IllegalArgumentException.class, () -> cfg(WriteMode.UPSERT, List.of(), 1000)); + } +} diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExecTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExecTest.java new file mode 100644 index 00000000000..5fb1cc5c605 --- /dev/null +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/write/OutputLookupWriteExecTest.java @@ -0,0 +1,52 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.opensearch.storage.write; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.Map; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rel.type.RelDataTypeSystem; +import org.apache.calcite.sql.type.SqlTypeFactoryImpl; +import org.apache.calcite.sql.type.SqlTypeName; +import org.junit.jupiter.api.Test; + +class OutputLookupWriteExecTest { + + @SuppressWarnings("unchecked") + @Test + void inferMappingMapsSqlTypesToOpenSearchTypes() { + RelDataTypeFactory tf = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT); + RelDataType rowType = + tf.builder() + .add("id", SqlTypeName.INTEGER) + .add("count", SqlTypeName.BIGINT) + .add("name", SqlTypeName.VARCHAR) + .add("ts", SqlTypeName.TIMESTAMP) + .add("ratio", SqlTypeName.DOUBLE) + .add("ok", SqlTypeName.BOOLEAN) + .add("_routing", SqlTypeName.VARCHAR) + .build(); + + Map mapping = OutputLookupWriteExec.inferMapping(rowType); + Map props = (Map) mapping.get("properties"); + + assertEquals("long", type(props, "id")); + assertEquals("long", type(props, "count")); + assertEquals("keyword", type(props, "name")); + assertEquals("date", type(props, "ts")); + assertEquals("double", type(props, "ratio")); + assertEquals("boolean", type(props, "ok")); + org.junit.jupiter.api.Assertions.assertFalse( + props.containsKey("_routing"), "reserved metadata field must be excluded from the mapping"); + } + + @SuppressWarnings("unchecked") + private static String type(Map props, String field) { + return (String) ((Map) props.get(field)).get("type"); + } +} diff --git a/ppl/src/main/antlr/OpenSearchPPLLexer.g4 b/ppl/src/main/antlr/OpenSearchPPLLexer.g4 index f9d67bd46ff..94f917ca15e 100644 --- a/ppl/src/main/antlr/OpenSearchPPLLexer.g4 +++ b/ppl/src/main/antlr/OpenSearchPPLLexer.g4 @@ -178,6 +178,9 @@ TIME_ZONE: 'TIME_ZONE'; TRAINING_DATA_SIZE: 'TRAINING_DATA_SIZE'; ANOMALY_SCORE_THRESHOLD: 'ANOMALY_SCORE_THRESHOLD'; APPEND: 'APPEND'; +OUTPUTLOOKUP: 'OUTPUTLOOKUP'; +OVERRIDE_IF_EMPTY: 'OVERRIDE_IF_EMPTY'; +KEY_FIELD: 'KEY_FIELD'; MULTISEARCH: 'MULTISEARCH'; UNION: 'UNION'; MAXOUT: 'MAXOUT'; diff --git a/ppl/src/main/antlr/OpenSearchPPLParser.g4 b/ppl/src/main/antlr/OpenSearchPPLParser.g4 index 354ffcd425d..3abda32fbf2 100644 --- a/ppl/src/main/antlr/OpenSearchPPLParser.g4 +++ b/ppl/src/main/antlr/OpenSearchPPLParser.g4 @@ -99,6 +99,7 @@ commands | nomvCommand | graphLookupCommand | unionCommand + | outputlookupCommand | timewrapCommand ; @@ -334,6 +335,18 @@ reverseCommand : REVERSE ; +// outputlookup: terminal write sink — materialize pipeline rows into a lookup index (overwrite by default). +outputlookupCommand + : OUTPUTLOOKUP outputlookupOption* tableSourceClause + ; + +outputlookupOption + : APPEND EQUAL append = booleanLiteral + | OVERRIDE_IF_EMPTY EQUAL overrideIfEmpty = booleanLiteral + | KEY_FIELD EQUAL keyFields += qualifiedName (COMMA keyFields += qualifiedName)* + | MAX EQUAL max = integerLiteral + ; + chartCommand : CHART chartOptions* statsAggTerm (OVER rowSplit)? (BY columnSplit)? chartOptions* | CHART chartOptions* statsAggTerm BY rowSplit (COMMA)? columnSplit chartOptions* @@ -1680,6 +1693,9 @@ searchableKeyWord | EXISTS | SOURCE | INDEX + | OUTPUTLOOKUP + | OVERRIDE_IF_EMPTY + | KEY_FIELD | A | ASC | DESC 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 2f9168383b8..4b227078f42 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 @@ -99,6 +99,7 @@ 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.OutputLookup; import org.opensearch.sql.ast.tree.Parse; import org.opensearch.sql.ast.tree.Patterns; import org.opensearch.sql.ast.tree.Project; @@ -789,6 +790,36 @@ public UnresolvedPlan visitReverseCommand(OpenSearchPPLParser.ReverseCommandCont return new Reverse(); } + /** Outputlookup command visitor (write sink). Materializes rows into a lookup index. */ + @Override + public UnresolvedPlan visitOutputlookupCommand( + OpenSearchPPLParser.OutputlookupCommandContext ctx) { + String indexName = StringUtils.unquoteIdentifier(ctx.tableSourceClause().getText()); + OutputLookup node = new OutputLookup(indexName); + boolean appendExplicit = false; + for (OpenSearchPPLParser.OutputlookupOptionContext opt : ctx.outputlookupOption()) { + if (opt.append != null) { + node.setAppend(Boolean.parseBoolean(opt.append.getText())); + appendExplicit = true; + } else if (opt.overrideIfEmpty != null) { + node.setOverrideIfEmpty(Boolean.parseBoolean(opt.overrideIfEmpty.getText())); + } else if (opt.keyFields != null && !opt.keyFields.isEmpty()) { + java.util.List keyFields = new java.util.ArrayList<>(); + for (OpenSearchPPLParser.QualifiedNameContext q : opt.keyFields) { + keyFields.add(StringUtils.unquoteIdentifier(q.getText())); + } + node.setKeyFields(keyFields); + } else if (opt.max != null) { + node.setMax(Integer.parseInt(opt.max.getText())); + } + } + // Splunk parity: key_field sets append=true by default, unless append was given explicitly. + if (!node.getKeyFields().isEmpty() && !appendExplicit) { + node.setAppend(true); + } + return node; + } + /** Transpose command. */ @Override public UnresolvedPlan visitTransposeCommand(OpenSearchPPLParser.TransposeCommandContext ctx) { 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 7d849517951..32d44a45299 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 @@ -646,6 +646,12 @@ public String visitReverse(Reverse node, String context) { return StringUtils.format("%s | reverse", child); } + @Override + public String visitOutputLookup(org.opensearch.sql.ast.tree.OutputLookup node, String context) { + String child = node.getChild().get(0).accept(this, context); + return StringUtils.format("%s | outputlookup ***", child); + } + @Override public String visitTimewrap(Timewrap node, String context) { String child = node.getChild().get(0).accept(this, context); diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java index d57ca8a69bb..aedcf125d78 100644 --- a/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java +++ b/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java @@ -81,6 +81,7 @@ import org.opensearch.sql.ast.tree.Join; import org.opensearch.sql.ast.tree.Kmeans; import org.opensearch.sql.ast.tree.ML; +import org.opensearch.sql.ast.tree.OutputLookup; import org.opensearch.sql.ast.tree.RareTopN.CommandType; import org.opensearch.sql.common.antlr.SyntaxCheckException; import org.opensearch.sql.common.setting.Settings.Key; @@ -1940,6 +1941,57 @@ public void testJoinPrefixWithoutCriteriaKeywordIsSyntaxError() { assertThrows(SyntaxCheckException.class, () -> plan("source=t1 | inner join a t2")); } + @Test + public void testOutputLookupCommandDefault() { + assertEqual("source=t | outputlookup hosts", new OutputLookup("hosts").attach(relation("t"))); + } + + @Test + public void testOutputLookupCommandWithOptions() { + OutputLookup expected = new OutputLookup("hosts"); + expected.setAppend(true); + expected.setOverrideIfEmpty(false); + expected.setKeyFields(java.util.List.of("id")); + expected.setMax(1000); + assertEqual( + "source=t | outputlookup append=true override_if_empty=false key_field=id max=1000 hosts", + expected.attach(relation("t"))); + } + + @Test + public void testOutputLookupUsableAsIdentifier() { + // outputlookup must remain usable as a bare field name (keywordsCanBeId) + plan("source=t | fields outputlookup"); + } + + @Test + public void testOutputLookupKeyFieldDefaultsAppendTrue() { + // Splunk parity: key_field implies append=true by default. + OutputLookup expected = new OutputLookup("hosts"); + expected.setKeyFields(java.util.List.of("id")); + expected.setAppend(true); + assertEqual("source=t | outputlookup key_field=id hosts", expected.attach(relation("t"))); + } + + @Test + public void testOutputLookupExplicitAppendOverridesKeyFieldDefault() { + // Explicit append=false wins over the key_field append-default. + OutputLookup expected = new OutputLookup("hosts"); + expected.setKeyFields(java.util.List.of("id")); + expected.setAppend(false); + assertEqual( + "source=t | outputlookup append=false key_field=id hosts", expected.attach(relation("t"))); + } + + @Test + public void testOutputLookupMultipleKeyFields() { + OutputLookup expected = new OutputLookup("hosts"); + expected.setKeyFields(java.util.List.of("region", "host")); + expected.setAppend(true); // key_field implies append=true + assertEqual( + "source=t | outputlookup key_field=region,host hosts", expected.attach(relation("t"))); + } + // rest command tests @Test