From c349f83e977971caf7923a8968d2e8a2b61e02b8 Mon Sep 17 00:00:00 2001 From: Chayanin Noramuttha Date: Tue, 6 Oct 2026 09:28:02 -0700 Subject: [PATCH 01/10] Add OTel data stream ITs that pin schema evolution 500s Adds CalciteOtelLogsDataStreamIT and CalciteOtelTracesDataStreamIT for #5806. Each rolls an OTel data stream over so that the second backing index changes, removes or adds one attributes.types field, then runs PPL through the stream name and a wildcard at pushdown on. The 21 pairs among text, keyword, byte, integer, long, date and object each run twice with the types swapped between the backing indices, so both merge winners run on one node. A 500 fails the run unless a pin ties it to an issue, and a pinned cell that stops returning 500 fails until its pin is updated. A 200 or a 4xx passes. Today 93 keys, 186 queries, in each IT are pinned to #5811, #5825, #5835 and #5836. MappingConflictFixtures holds the field types, the 19 PPL templates and the pins, so a fix that clears a 500 edits only that file. OtelDataStreamConflictTestCase holds the flow both ITs share. The ITs run in a new mappingConflictIntegTest task. Its single node disables the InternalComposite asserts, since with assertions on the node exits on a mixed-type composite key. integTest and integTestRemote exclude both ITs, and both integration steps of sql-test-and-build-workflow.yml run the task. Signed-off-by: Chayanin Noramuttha --- .../workflows/sql-test-and-build-workflow.yml | 4 +- integ-test/build.gradle | 53 ++++ .../remote/CalciteOtelLogsDataStreamIT.java | 29 ++ .../remote/CalciteOtelTracesDataStreamIT.java | 29 ++ .../remote/MappingConflictFixtures.java | 124 ++++++++ .../OtelDataStreamConflictTestCase.java | 297 ++++++++++++++++++ 6 files changed, 534 insertions(+), 2 deletions(-) create mode 100644 integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java create mode 100644 integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java create mode 100644 integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java create mode 100644 integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java diff --git a/.github/workflows/sql-test-and-build-workflow.yml b/.github/workflows/sql-test-and-build-workflow.yml index ccc7f11d605..8ae19c6a961 100644 --- a/.github/workflows/sql-test-and-build-workflow.yml +++ b/.github/workflows/sql-test-and-build-workflow.yml @@ -108,7 +108,7 @@ jobs: if [ "${{ matrix.test-type }}" = "unit" ]; then ./gradlew --continue build -x spotlessCheck -x integTest -x tracingIntegTest -x yamlRestTest -x doctest elif [ "${{ matrix.test-type }}" = "integration" ]; then - ./gradlew --continue integTest tracingIntegTest yamlRestTest + ./gradlew --continue integTest tracingIntegTest yamlRestTest mappingConflictIntegTest else ./gradlew --continue doctest fi @@ -213,7 +213,7 @@ jobs: if [ "${{ matrix.test-type }}" = "unit" ]; then ./gradlew --continue build -x spotlessCheck -x integTest -x tracingIntegTest -x yamlRestTest -x doctest ${{ matrix.entry.os_build_args }} elif [ "${{ matrix.test-type }}" = "integration" ]; then - ./gradlew --continue integTest tracingIntegTest yamlRestTest ${{ matrix.entry.os_build_args }} + ./gradlew --continue integTest tracingIntegTest yamlRestTest mappingConflictIntegTest ${{ matrix.entry.os_build_args }} else ./gradlew --continue doctest ${{ matrix.entry.os_build_args }} fi diff --git a/integ-test/build.gradle b/integ-test/build.gradle index 6026e19f018..44aeee23883 100644 --- a/integ-test/build.gradle +++ b/integ-test/build.gradle @@ -507,6 +507,17 @@ testClusters { 'io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter' setting 'telemetry.otel.tracer.exporter.delay', '50ms' } + mappingConflictIntegTest { + testDistribution = 'archive' + plugin(getJobSchedulerPlugin()) + plugin ":opensearch-sql-plugin" + setting "plugins.query.datasources.encryption.masterkey", "1234567812345678" + // Mixed-type bucket keys trip the key-class assert in InternalBucket.compareKey, and the + // AssertionError exits the node. Off here only, so the node returns the 500 a production + // node returns. Named by the top-level class, because a nested class takes its assertion + // status from there. The shared integTest cluster keeps every assert. + jvmArgs '-da:org.opensearch.search.aggregations.bucket.composite.InternalComposite' + } yamlRestTest { testDistribution = 'archive' plugin(getJobSchedulerPlugin()) @@ -917,6 +928,9 @@ integTest { exclude 'org/opensearch/sql/calcite/tracing/**' + // Runs in :mappingConflictIntegTest, whose cluster disables the InternalComposite asserts. + exclude 'org/opensearch/sql/calcite/remote/CalciteOtel*DataStreamIT.class' + // Explain IT is dependent on internal implementation of old engine so it's not necessary // to run these with new engine and not necessary to make this consistent with old engine. exclude 'org/opensearch/sql/legacy/ExplainIT.class' @@ -1026,6 +1040,40 @@ task tracingIntegTest(type: RestIntegTestTask) { } } +task mappingConflictIntegTest(type: RestIntegTestTask) { + useCluster testClusters.mappingConflictIntegTest + dependsOn ':opensearch-sql-plugin:bundlePlugin' + + systemProperty 'tests.security.manager', 'false' + systemProperty 'project.root', project.projectDir.absolutePath + systemProperty 'defaultQuerySizeLimit', '10000' + + doFirst { + getClusters().forEach { cluster -> + String allHttpSocketURI = cluster.nodes.stream().flatMap { node -> + node.getAllHttpSocketURI().stream() + }.collect(Collectors.joining(",")) + String allTransportSocketURI = cluster.nodes.stream().flatMap { node -> + node.getAllTransportPortURI().stream() + }.collect(Collectors.joining(",")) + systemProperty "tests.rest.${cluster.name}.http_hosts", "${-> allHttpSocketURI}" + systemProperty "tests.rest.${cluster.name}.transport_hosts", "${-> allTransportSocketURI}" + } + systemProperty "https", "false" + systemProperty "user", "admin" + systemProperty "password", "admin" + systemProperty 'cluster.debug', getDebug() + } + + testLogging { + events "passed", "skipped", "failed" + } + + filter { + includeTestsMatching 'org.opensearch.sql.calcite.remote.CalciteOtel*DataStreamIT' + } +} + task comparisonTest(type: RestIntegTestTask) { testLogging { events "passed", "skipped", "failed" @@ -1596,4 +1644,9 @@ task integTestRemote(type: RestIntegTestTask) { // externally provided cluster, so these suites cannot pass here. integTest still covers them. exclude 'org/opensearch/sql/datasource/DataSourceAPIsIT.class' exclude 'org/opensearch/sql/datasource/DataSourceEnabledIT.class' + + // Suites that pin the 500s of one build on a single node with the composite assert off. An + // external cluster can add nodes, which redraws the merge winner per request, and cannot drop + // the assert, so they run only in :mappingConflictIntegTest. + exclude 'org/opensearch/sql/calcite/remote/CalciteOtel*DataStreamIT.class' } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java new file mode 100644 index 00000000000..026a7620df9 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java @@ -0,0 +1,29 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.calcite.remote; + +import java.io.IOException; +import org.junit.Test; +import org.opensearch.sql.util.OtelDataStream; + +/** Schema evolution across the backing indices of an OTel logs data stream. */ +public class CalciteOtelLogsDataStreamIT extends OtelDataStreamConflictTestCase { + + @Override + protected OtelDataStream create(String name) throws IOException { + return OtelDataStream.logs(client(), name); + } + + @Override + protected String control() { + return "`resource.attributes.service.name`"; + } + + @Test + public void typeDrifts() throws IOException { + runDrifts("otel-logs-drift"); + } +} diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java new file mode 100644 index 00000000000..84d243d89f3 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java @@ -0,0 +1,29 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.calcite.remote; + +import java.io.IOException; +import org.junit.Test; +import org.opensearch.sql.util.OtelDataStream; + +/** Schema evolution across the backing indices of an OTel traces data stream. */ +public class CalciteOtelTracesDataStreamIT extends OtelDataStreamConflictTestCase { + + @Override + protected OtelDataStream create(String name) throws IOException { + return OtelDataStream.traces(client(), name); + } + + @Override + protected String control() { + return "serviceName"; + } + + @Test + public void typeDrifts() throws IOException { + runDrifts("otel-traces-drift"); + } +} diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java new file mode 100644 index 00000000000..d2dcb85cc11 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java @@ -0,0 +1,124 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.calcite.remote; + +import java.util.List; + +/** + * The data the OTel data stream ITs run over, meaning the field types the drifts cross, the PPL + * commands, and the 500s pinned to their issues. {@link OtelDataStreamConflictTestCase} holds the + * flow, so a fix that clears a 500 edits only this file. + */ +final class MappingConflictFixtures { + + /** The field types the drifts cross, each against the others and against the field absent. */ + static final List TYPES = + List.of("text", "keyword", "byte", "integer", "long", "date", "object"); + + /** + * A PPL command template, holding {@code {I}} for the source, {@code {F}} for the field under + * test and {@code {K}} for a control field. + */ + record Command(String name, String template) { + String query(String source, String field, String control) { + return template.replace("{I}", source).replace("{F}", field).replace("{K}", control); + } + } + + /** + * Representative commands, one per way PPL reads, filters, aggregates, orders or windows the + * field, covering each command family the mapping conflict issues in #5829 report as a 500. + */ + static final List REPRESENTATIVE = + List.of( + new Command("where_isnotnull", "source={I} | where isnotnull({F}) | fields {K}"), + new Command("fields_keep", "source={I} | fields {F}"), + new Command("eval_cast_string", "source={I} | eval x = cast({F} as string) | fields x"), + new Command("stats_count_by", "source={I} | stats count() by {F}"), + new Command("stats_max", "source={I} | stats max({F})"), + new Command("stats_dc", "source={I} | stats dc({F})"), + new Command("stats", "source={I} | stats count()"), + new Command("eventstats_by", "source={I} | eventstats count() by {F}"), + new Command( + "streamstats_count_by", "source={I} | streamstats count() by {F} | fields {K}"), + new Command("timechart_by", "source={I} | timechart span=1d count() by {F}"), + new Command("chart_by", "source={I} | chart count() over {K} by {F}"), + new Command("sort", "source={I} | sort {F}"), + new Command("reverse_after_sort", "source={I} | sort {F} | reverse | fields {K}"), + new Command("head_after_sort", "source={I} | sort {F} | head 1 | fields {K}"), + new Command("dedup", "source={I} | dedup {F}"), + new Command("top", "source={I} | top {F}"), + new Command("rare", "source={I} | rare {F}"), + new Command("rex_named_group", "source={I} | rex field={F} '(?.+)' | fields rr"), + new Command( + "join_self", + "source={I} | head 100 | inner join left = l right = r on l.{F} = r.{F} [ source={I}" + + " | head 100 ] | head 10")); + + /** + * Every listed command over every listed pair returns a 500 under {@code winner}, the PPL type + * the merged schema reports, or under every winner the pair shows when {@code winner} is null. A + * fix narrows its pin to the cells that still fail, splitting the pin if they no longer form one + * set of commands over one set of pairs, and the run fails until it does. + */ + record Pin(String issue, List commands, List pairs, String winner) { + Pin { + if (commands.isEmpty() || pairs.isEmpty()) { + throw new IllegalArgumentException("The " + issue + " pin needs commands and pairs"); + } + } + + boolean covers(String pair, String cellWinner, String command) { + return pairs.contains(pair) + && commands.contains(command) + && (winner == null || winner.equals(cellWinner)); + } + } + + /** + * The cells that return a 500 today, grouped by the issue that tracks them. Logs and traces + * return the same ones, since the drift lands on the same fields. Each pin lists its own commands + * and pairs, so a fix edits only its own entry. + */ + static final List KNOWN_500S = + List.of( + // #5811 Composite aggregation pushdown does not support a field whose type differs across + // indices + new Pin( + "#5811", + List.of("stats_count_by", "dedup", "top", "rare", "chart_by", "timechart_by"), + List.of("keyword/byte", "keyword/integer", "keyword/long"), + null), + // #5811 Composite aggregation pushdown does not support a field whose type differs across + // indices + new Pin( + "#5811", + List.of("stats_count_by", "dedup", "top", "rare", "chart_by", "timechart_by"), + List.of("keyword/date"), + "string"), + // #5825 sort returns 500 when a field is keyword or ip in one index and a number, date or + // boolean in another + new Pin( + "#5825", + List.of("sort", "reverse_after_sort", "head_after_sort", "join_self"), + List.of("keyword/byte", "keyword/integer", "keyword/long", "keyword/date"), + null), + // #5835 stats max and stats min return 500 when a field is keyword or ip and boolean or a + // number in another + new Pin( + "#5835", + List.of("stats_max"), + List.of("keyword/byte", "keyword/integer", "keyword/long", "keyword/date"), + "string"), + // #5836 sort, stats max, and stats min on an object or nested field return 500 + new Pin( + "#5836", + List.of("sort", "reverse_after_sort", "head_after_sort", "join_self", "stats_max"), + List.of("text/object", "object/absent", "absent/object"), + "struct")); + + private MappingConflictFixtures() {} +} diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java new file mode 100644 index 00000000000..f6e404042cc --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java @@ -0,0 +1,297 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.calcite.remote; + +import static org.junit.Assert.fail; +import static org.opensearch.sql.calcite.remote.MappingConflictFixtures.KNOWN_500S; +import static org.opensearch.sql.calcite.remote.MappingConflictFixtures.REPRESENTATIVE; +import static org.opensearch.sql.calcite.remote.MappingConflictFixtures.TYPES; +import static org.opensearch.sql.legacy.TestUtils.getResponseBody; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.TreeSet; +import java.util.stream.Collectors; +import org.json.JSONArray; +import org.json.JSONObject; +import org.opensearch.client.Request; +import org.opensearch.client.ResponseException; +import org.opensearch.sql.calcite.remote.MappingConflictFixtures.Command; +import org.opensearch.sql.calcite.remote.MappingConflictFixtures.Pin; +import org.opensearch.sql.ppl.PPLIntegTestCase; +import org.opensearch.sql.util.OtelDataStream; +import org.opensearch.sql.util.OtelDataStream.Drift; + +/** + * Runs PPL commands over an OTel data stream whose second backing index changes the type of one + * {@code attributes.types} field, removes it, or adds one, through the stream name and a wildcard. + * Every pair of {@link MappingConflictFixtures#TYPES} runs twice, once with each type in 000001. + * One node reads the backing indices in a fixed order, so the two passes feed the merge both orders + * a two-index stream can produce and cover every winner production can return. A 500 fails the run + * unless {@link MappingConflictFixtures#KNOWN_500S} pins its cell to an issue, and a pinned cell + * that stops returning 500 fails too. An unpinned 200 or 4xx passes, since this checks only that + * schema evolution never returns a 500. + */ +public abstract class OtelDataStreamConflictTestCase extends PPLIntegTestCase { + + private static final int BASE_DOCS = 5; + + /** Creates the stream with its base documents in 000001. */ + protected abstract OtelDataStream create(String name) throws IOException; + + /** A keyword field no drift touches, for {@code {K}}. */ + protected abstract String control(); + + @Override + public void init() throws Exception { + super.init(); + enableCalcite(); + } + + /** Runs every drift under {@code name} and fails on any unpinned 500 or stale pin. */ + protected void runDrifts(String name) throws IOException { + JSONArray nodes = new JSONArray(executeRequest(new Request("GET", "/_cat/nodes?format=json"))); + if (nodes.length() != 1) { + throw new IllegalStateException( + "The swap needs one node to fix the merge order, the cluster has " + nodes.length()); + } + List commands = REPRESENTATIVE; + Set names = commands.stream().map(Command::name).collect(Collectors.toSet()); + List cells = new ArrayList<>(); + List failures = new ArrayList<>(); + for (int i = 0; i < TYPES.size(); i++) { + for (int j = i + 1; j < TYPES.size(); j++) { + String a = TYPES.get(i); + String b = TYPES.get(j); + String pair = a + "/" + b; + cells.addAll(pass(name, pair, a, retype(a, b), commands)); + cells.addAll(pass(name, pair, b, retype(b, a), commands)); + } + } + for (String type : TYPES) { + String field = "attributes.types." + type; + cells.addAll( + pass(name, type + "/absent", type, n -> Drift.remove(field, BASE_DOCS), commands)); + String added = "added_" + type; + cells.addAll( + pass( + name, + "absent/" + type, + added, + n -> + Drift.add( + "attributes.types." + added, type, baseValues(n, "types." + type).toArray()), + commands)); + } + cells.stream() + .filter(c -> c.winner().startsWith("unread-")) + .forEach( + c -> + failures.add( + c.pair() + " via " + c.source() + " could not read its winner, " + c.winner())); + String tag = tag(); + Set serverErrors = new TreeSet<>(); + for (Cell cell : cells) { + if (cell.status() >= 500) { + serverErrors.add(cell.key()); + System.out.println( + tag + "500 " + cell.key() + " | " + cell.source() + " | " + cell.error()); + if (KNOWN_500S.stream() + .noneMatch(p -> p.covers(cell.pair(), cell.winner(), cell.command()))) { + failures.add( + "unpinned 500, " + + cell.label() + + "\n query: " + + cell.query() + + "\n error: " + + cell.error().substring(0, Math.min(200, cell.error().length()))); + } + } + } + System.out.println( + tag + + cells.size() + + " queries, " + + cells.stream().filter(c -> c.status() >= 500).count() + + " return 500 across " + + serverErrors.size() + + " pair, winner and command keys"); + Set pairs = cells.stream().map(Cell::pair).collect(Collectors.toSet()); + for (Pin pin : KNOWN_500S) { + pin.commands().stream() + .filter(c -> !names.contains(c)) + .forEach( + c -> failures.add("the " + pin.issue() + " pin lists " + c + ", which never runs")); + List covered = + cells.stream().filter(c -> pin.covers(c.pair(), c.winner(), c.command())).toList(); + pin.pairs().stream() + .filter(p -> covered.stream().noneMatch(c -> c.pair().equals(p))) + .forEach( + p -> + failures.add( + "the " + + pin.issue() + + " pin lists " + + p + + (pairs.contains(p) + ? ", which yields no cell under its winner, so fix or remove it" + : ", which never runs, so check its spelling and TYPES order"))); + covered.stream() + .filter(c -> c.status() < 500) + .forEach( + c -> + failures.add( + "pinned to " + + pin.issue() + + " but no longer returns 500, " + + c.label() + + ", narrow that pin to the cells that still fail\n query: " + + c.query())); + } + if (!failures.isEmpty()) { + Set distinct = new TreeSet<>(failures); + fail(distinct.size() + " failures\n" + String.join("\n", distinct)); + } + } + + /** Builds a drift once the stream exists, since its values come from the base documents. */ + @FunctionalInterface + private interface DriftFactory { + Drift drift(String name) throws IOException; + } + + /** Changes {@code from}'s field to {@code to}, filled with {@code to}'s base values. */ + private DriftFactory retype(String from, String to) { + return n -> + Drift.changeType("attributes.types." + from, to, baseValues(n, "types." + to).toArray()); + } + + /** + * Rolls the stream over to the drift and runs every command through the stream name and the + * wildcard, each under the winner its merged schema reports. + */ + private List pass( + String name, String pair, String type, DriftFactory factory, List commands) + throws IOException { + OtelDataStream stream = create(name); + List cells = new ArrayList<>(); + try { + String field = "`attributes.types." + type + "`"; + stream.rollover(factory.drift(name)); + for (String source : List.of(name, name + "*")) { + String resolved = winner(source, field); + System.out.println( + tag() + "pass " + pair + ", field " + field + " via " + source + " reads " + resolved); + for (Command command : commands) { + String query = command.query(source, field, control()); + Response response = run(query); + cells.add( + new Cell( + pair, + resolved, + command.name(), + source, + query, + response.status(), + response.error())); + } + } + } catch (IOException | RuntimeException e) { + try { + stream.delete(); + } catch (IOException | RuntimeException cleanup) { + e.addSuppressed(cleanup); + } + throw e; + } + stream.delete(); + return cells; + } + + private String tag() { + return "[" + getClass().getSimpleName() + "] "; + } + + /** The PPL type the merged schema gives the field, or the status when it cannot be read. */ + private String winner(String source, String field) throws IOException { + Response response = run("source=" + source + " | fields " + field + " | head 1"); + return response.status() == 200 && !response.type().isEmpty() + ? response.type() + : "unread-" + response.status(); + } + + /** Each base document's value for a flattened {@code attributes} key. */ + private List baseValues(String name, String key) throws IOException { + JSONObject body = + new JSONObject( + getResponseBody( + client().performRequest(new Request("GET", "/" + name + "/_search?size=100")))); + JSONArray hits = body.getJSONObject("hits").getJSONArray("hits"); + List values = new ArrayList<>(); + for (int i = 0; i < hits.length(); i++) { + values.add( + hits.getJSONObject(i).getJSONObject("_source").getJSONObject("attributes").get(key)); + } + if (values.size() != BASE_DOCS) { + throw new IllegalStateException( + name + " holds " + values.size() + " base documents, expected " + BASE_DOCS); + } + return values; + } + + private Response run(String query) throws IOException { + Request request = new Request("POST", "/_plugins/_ppl"); + request.setJsonEntity(new JSONObject().put("query", query).toString()); + try { + JSONObject body = new JSONObject(getResponseBody(client().performRequest(request))); + JSONArray schema = body.optJSONArray("schema"); + return new Response( + 200, + schema == null || schema.isEmpty() ? "" : schema.getJSONObject(0).optString("type"), + ""); + } catch (ResponseException e) { + String error = ""; + try { + JSONObject json = new JSONObject(getResponseBody(e.getResponse())).optJSONObject("error"); + if (json != null) { + String details = json.optString("details"); + error = + json.optString("type") + + ": " + + (details.isEmpty() ? json.optString("reason") : details); + } + } catch (IOException | RuntimeException ignored) { + // keep the status alone + } + error = error.replace('\n', ' '); + return new Response(e.getResponse().getStatusLine().getStatusCode(), "", error); + } + } + + private record Response(int status, String type, String error) {} + + /** One command over one source under one winner. */ + private record Cell( + String pair, + String winner, + String command, + String source, + String query, + int status, + String error) { + + String key() { + return pair + " " + winner + " " + command; + } + + /** The cell as a failure line reads it. */ + String label() { + return "pair " + pair + ", winner " + winner + ", command " + command; + } + } +} From a2b14dd35b6388d2e9dc3b8bcb48008bf0ad4d39 Mon Sep 17 00:00:00 2001 From: Chayanin Noramuttha Date: Tue, 6 Oct 2026 14:39:27 -0700 Subject: [PATCH 02/10] Accept 4xx winner probes, require a 200, and scope pins by source A winner probe that returns a 4xx on a type pair no longer fails the run, so a fix that rejects conflicting types with a 400 can pass. A 4xx probe on a pass where one side has no field still fails, since nothing conflicts there. The run now fails when a command never returns 200, so a template that every query rejects cannot pass while testing nothing. Pin takes an optional Source, DATA_STREAM, WILDCARD or BOTH, so a 500 that only one source returns can be pinned. The existing constructor defaults to BOTH, and the five pins are unchanged. Signed-off-by: Chayanin Noramuttha --- .../remote/MappingConflictFixtures.java | 31 +++++++++++++++---- .../OtelDataStreamConflictTestCase.java | 25 +++++++++++---- 2 files changed, 44 insertions(+), 12 deletions(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java index d2dcb85cc11..439e756d3d8 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java @@ -58,23 +58,42 @@ String query(String source, String field, String control) { "source={I} | head 100 | inner join left = l right = r on l.{F} = r.{F} [ source={I}" + " | head 100 ] | head 10")); + /** The source a pin covers, the data stream name, the wildcard over it, or both. */ + enum Source { + DATA_STREAM, + WILDCARD, + BOTH; + + boolean matches(String source) { + return this == BOTH || (this == WILDCARD) == source.endsWith("*"); + } + } + /** * Every listed command over every listed pair returns a 500 under {@code winner}, the PPL type - * the merged schema reports, or under every winner the pair shows when {@code winner} is null. A - * fix narrows its pin to the cells that still fail, splitting the pin if they no longer form one - * set of commands over one set of pairs, and the run fails until it does. + * the merged schema reports, or under every winner the pair shows when {@code winner} is null, + * through {@code source}. A fix narrows its pin to the cells that still fail, splitting the pin + * if they no longer form one set of commands over one set of pairs, and the run fails until it + * does. */ - record Pin(String issue, List commands, List pairs, String winner) { + record Pin( + String issue, List commands, List pairs, String winner, Source source) { Pin { if (commands.isEmpty() || pairs.isEmpty()) { throw new IllegalArgumentException("The " + issue + " pin needs commands and pairs"); } } - boolean covers(String pair, String cellWinner, String command) { + /** A pin that covers both sources. */ + Pin(String issue, List commands, List pairs, String winner) { + this(issue, commands, pairs, winner, Source.BOTH); + } + + boolean covers(String pair, String cellWinner, String command, String cellSource) { return pairs.contains(pair) && commands.contains(command) - && (winner == null || winner.equals(cellWinner)); + && (winner == null || winner.equals(cellWinner)) + && source.matches(cellSource); } } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java index f6e404042cc..2bf591170ae 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java @@ -35,7 +35,8 @@ * a two-index stream can produce and cover every winner production can return. A 500 fails the run * unless {@link MappingConflictFixtures#KNOWN_500S} pins its cell to an issue, and a pinned cell * that stops returning 500 fails too. An unpinned 200 or 4xx passes, since this checks only that - * schema evolution never returns a 500. + * schema evolution never returns a 500, but a command that never returns 200 fails, since it tests + * nothing. */ public abstract class OtelDataStreamConflictTestCase extends PPLIntegTestCase { @@ -53,7 +54,10 @@ public void init() throws Exception { enableCalcite(); } - /** Runs every drift under {@code name} and fails on any unpinned 500 or stale pin. */ + /** + * Runs every drift under {@code name} and fails on any unpinned 500, stale pin, or command that + * never returns 200. + */ protected void runDrifts(String name) throws IOException { JSONArray nodes = new JSONArray(executeRequest(new Request("GET", "/_cat/nodes?format=json"))); if (nodes.length() != 1) { @@ -89,11 +93,17 @@ protected void runDrifts(String name) throws IOException { commands)); } cells.stream() - .filter(c -> c.winner().startsWith("unread-")) + .filter( + c -> + c.winner().startsWith("unread-") + && (c.pair().contains("absent") || !c.winner().startsWith("unread-4"))) .forEach( c -> failures.add( c.pair() + " via " + c.source() + " could not read its winner, " + c.winner())); + names.stream() + .filter(n -> cells.stream().noneMatch(c -> c.command().equals(n) && c.status() == 200)) + .forEach(n -> failures.add(n + " never returned 200, so it tests nothing")); String tag = tag(); Set serverErrors = new TreeSet<>(); for (Cell cell : cells) { @@ -102,7 +112,7 @@ protected void runDrifts(String name) throws IOException { System.out.println( tag + "500 " + cell.key() + " | " + cell.source() + " | " + cell.error()); if (KNOWN_500S.stream() - .noneMatch(p -> p.covers(cell.pair(), cell.winner(), cell.command()))) { + .noneMatch(p -> p.covers(cell.pair(), cell.winner(), cell.command(), cell.source()))) { failures.add( "unpinned 500, " + cell.label() @@ -128,7 +138,9 @@ protected void runDrifts(String name) throws IOException { .forEach( c -> failures.add("the " + pin.issue() + " pin lists " + c + ", which never runs")); List covered = - cells.stream().filter(c -> pin.covers(c.pair(), c.winner(), c.command())).toList(); + cells.stream() + .filter(c -> pin.covers(c.pair(), c.winner(), c.command(), c.source())) + .toList(); pin.pairs().stream() .filter(p -> covered.stream().noneMatch(c -> c.pair().equals(p))) .forEach( @@ -139,7 +151,8 @@ protected void runDrifts(String name) throws IOException { + " pin lists " + p + (pairs.contains(p) - ? ", which yields no cell under its winner, so fix or remove it" + ? ", which yields no cell under its winner and source, so fix or" + + " remove it" : ", which never runs, so check its spelling and TYPES order"))); covered.stream() .filter(c -> c.status() < 500) From 94c5e5e08260318211cfa0d663c840154ea05731 Mon Sep 17 00:00:00 2001 From: Chayanin Noramuttha Date: Tue, 6 Oct 2026 15:21:06 -0700 Subject: [PATCH 03/10] Log mapping conflict diagnostics through the test logger Signed-off-by: Chayanin Noramuttha --- .../OtelDataStreamConflictTestCase.java | 24 ++++++------------- 1 file changed, 7 insertions(+), 17 deletions(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java index 2bf591170ae..934291262ad 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java @@ -104,13 +104,11 @@ protected void runDrifts(String name) throws IOException { names.stream() .filter(n -> cells.stream().noneMatch(c -> c.command().equals(n) && c.status() == 200)) .forEach(n -> failures.add(n + " never returned 200, so it tests nothing")); - String tag = tag(); Set serverErrors = new TreeSet<>(); for (Cell cell : cells) { if (cell.status() >= 500) { serverErrors.add(cell.key()); - System.out.println( - tag + "500 " + cell.key() + " | " + cell.source() + " | " + cell.error()); + logger.info("500 {} | {} | {}", cell.key(), cell.source(), cell.error()); if (KNOWN_500S.stream() .noneMatch(p -> p.covers(cell.pair(), cell.winner(), cell.command(), cell.source()))) { failures.add( @@ -123,14 +121,11 @@ protected void runDrifts(String name) throws IOException { } } } - System.out.println( - tag - + cells.size() - + " queries, " - + cells.stream().filter(c -> c.status() >= 500).count() - + " return 500 across " - + serverErrors.size() - + " pair, winner and command keys"); + logger.info( + "{} queries, {} return 500 across {} pair, winner and command keys", + cells.size(), + cells.stream().filter(c -> c.status() >= 500).count(), + serverErrors.size()); Set pairs = cells.stream().map(Cell::pair).collect(Collectors.toSet()); for (Pin pin : KNOWN_500S) { pin.commands().stream() @@ -198,8 +193,7 @@ private List pass( stream.rollover(factory.drift(name)); for (String source : List.of(name, name + "*")) { String resolved = winner(source, field); - System.out.println( - tag() + "pass " + pair + ", field " + field + " via " + source + " reads " + resolved); + logger.info("pass {}, field {} via {} reads {}", pair, field, source, resolved); for (Command command : commands) { String query = command.query(source, field, control()); Response response = run(query); @@ -226,10 +220,6 @@ private List pass( return cells; } - private String tag() { - return "[" + getClass().getSimpleName() + "] "; - } - /** The PPL type the merged schema gives the field, or the status when it cannot be read. */ private String winner(String source, String field) throws IOException { Response response = run("source=" + source + " | fields " + field + " | head 1"); From 8301b185d6e9ba6c68ac7c3c2d8e203335a0ecdc Mon Sep 17 00:00:00 2001 From: Chayanin Noramuttha Date: Tue, 6 Oct 2026 15:31:25 -0700 Subject: [PATCH 04/10] Rename Cell, Pin, pass() and control() for readability Cell becomes Observation, the outcome of one query. Pin becomes Known500, a group of queries that return a 500 today, tied to the issue that tracks them. pass() becomes observe() and control() becomes stableField(). The Javadocs and failure messages drop the pin and cell wording to match. Signed-off-by: Chayanin Noramuttha --- .../remote/CalciteOtelLogsDataStreamIT.java | 2 +- .../remote/CalciteOtelTracesDataStreamIT.java | 2 +- .../remote/MappingConflictFixtures.java | 54 +++---- .../OtelDataStreamConflictTestCase.java | 136 ++++++++++-------- 4 files changed, 102 insertions(+), 92 deletions(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java index 026a7620df9..1717c04a8c6 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java @@ -18,7 +18,7 @@ protected OtelDataStream create(String name) throws IOException { } @Override - protected String control() { + protected String stableField() { return "`resource.attributes.service.name`"; } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java index 84d243d89f3..81c6309ddfe 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java @@ -18,7 +18,7 @@ protected OtelDataStream create(String name) throws IOException { } @Override - protected String control() { + protected String stableField() { return "serviceName"; } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java index 439e756d3d8..bcb6f34ef21 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java @@ -9,7 +9,7 @@ /** * The data the OTel data stream ITs run over, meaning the field types the drifts cross, the PPL - * commands, and the 500s pinned to their issues. {@link OtelDataStreamConflictTestCase} holds the + * commands, and the known 500s with their issues. {@link OtelDataStreamConflictTestCase} holds the * flow, so a fix that clears a 500 edits only this file. */ final class MappingConflictFixtures { @@ -20,11 +20,11 @@ final class MappingConflictFixtures { /** * A PPL command template, holding {@code {I}} for the source, {@code {F}} for the field under - * test and {@code {K}} for a control field. + * test and {@code {K}} for a stable field that no drift touches. */ record Command(String name, String template) { - String query(String source, String field, String control) { - return template.replace("{I}", source).replace("{F}", field).replace("{K}", control); + String query(String source, String field, String stableField) { + return template.replace("{I}", source).replace("{F}", field).replace("{K}", stableField); } } @@ -58,7 +58,7 @@ String query(String source, String field, String control) { "source={I} | head 100 | inner join left = l right = r on l.{F} = r.{F} [ source={I}" + " | head 100 ] | head 10")); - /** The source a pin covers, the data stream name, the wildcard over it, or both. */ + /** The source a known 500 covers, the data stream name, the wildcard over it, or both. */ enum Source { DATA_STREAM, WILDCARD, @@ -70,70 +70,70 @@ boolean matches(String source) { } /** - * Every listed command over every listed pair returns a 500 under {@code winner}, the PPL type - * the merged schema reports, or under every winner the pair shows when {@code winner} is null, - * through {@code source}. A fix narrows its pin to the cells that still fail, splitting the pin - * if they no longer form one set of commands over one set of pairs, and the run fails until it - * does. + * A group of queries that return a 500 today, tied to the issue that tracks them. Every listed + * command over every listed pair returns a 500 under {@code winner}, the PPL type the merged + * schema reports, or under every winner the pair shows when {@code winner} is null, through + * {@code source}. A fix narrows its entry to the queries that still fail, splitting the entry if + * they no longer form one set of commands over one set of pairs, and the run fails until it does. */ - record Pin( + record Known500( String issue, List commands, List pairs, String winner, Source source) { - Pin { + Known500 { if (commands.isEmpty() || pairs.isEmpty()) { - throw new IllegalArgumentException("The " + issue + " pin needs commands and pairs"); + throw new IllegalArgumentException("The " + issue + " known 500 needs commands and pairs"); } } - /** A pin that covers both sources. */ - Pin(String issue, List commands, List pairs, String winner) { + /** A known 500 on both sources. */ + Known500(String issue, List commands, List pairs, String winner) { this(issue, commands, pairs, winner, Source.BOTH); } - boolean covers(String pair, String cellWinner, String command, String cellSource) { + boolean covers(String pair, String observedWinner, String command, String observedSource) { return pairs.contains(pair) && commands.contains(command) - && (winner == null || winner.equals(cellWinner)) - && source.matches(cellSource); + && (winner == null || winner.equals(observedWinner)) + && source.matches(observedSource); } } /** - * The cells that return a 500 today, grouped by the issue that tracks them. Logs and traces - * return the same ones, since the drift lands on the same fields. Each pin lists its own commands - * and pairs, so a fix edits only its own entry. + * The queries that return a 500 today, grouped by the issue that tracks them. Logs and traces + * return the same ones, since the drift lands on the same fields. Each entry lists its own + * commands and pairs, so a fix edits only its own entry. */ - static final List KNOWN_500S = + static final List KNOWN_500S = List.of( // #5811 Composite aggregation pushdown does not support a field whose type differs across // indices - new Pin( + new Known500( "#5811", List.of("stats_count_by", "dedup", "top", "rare", "chart_by", "timechart_by"), List.of("keyword/byte", "keyword/integer", "keyword/long"), null), // #5811 Composite aggregation pushdown does not support a field whose type differs across // indices - new Pin( + new Known500( "#5811", List.of("stats_count_by", "dedup", "top", "rare", "chart_by", "timechart_by"), List.of("keyword/date"), "string"), // #5825 sort returns 500 when a field is keyword or ip in one index and a number, date or // boolean in another - new Pin( + new Known500( "#5825", List.of("sort", "reverse_after_sort", "head_after_sort", "join_self"), List.of("keyword/byte", "keyword/integer", "keyword/long", "keyword/date"), null), // #5835 stats max and stats min return 500 when a field is keyword or ip and boolean or a // number in another - new Pin( + new Known500( "#5835", List.of("stats_max"), List.of("keyword/byte", "keyword/integer", "keyword/long", "keyword/date"), "string"), // #5836 sort, stats max, and stats min on an object or nested field return 500 - new Pin( + new Known500( "#5836", List.of("sort", "reverse_after_sort", "head_after_sort", "join_self", "stats_max"), List.of("text/object", "object/absent", "absent/object"), diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java index 934291262ad..9072c068606 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java @@ -22,7 +22,7 @@ import org.opensearch.client.Request; import org.opensearch.client.ResponseException; import org.opensearch.sql.calcite.remote.MappingConflictFixtures.Command; -import org.opensearch.sql.calcite.remote.MappingConflictFixtures.Pin; +import org.opensearch.sql.calcite.remote.MappingConflictFixtures.Known500; import org.opensearch.sql.ppl.PPLIntegTestCase; import org.opensearch.sql.util.OtelDataStream; import org.opensearch.sql.util.OtelDataStream.Drift; @@ -33,10 +33,10 @@ * Every pair of {@link MappingConflictFixtures#TYPES} runs twice, once with each type in 000001. * One node reads the backing indices in a fixed order, so the two passes feed the merge both orders * a two-index stream can produce and cover every winner production can return. A 500 fails the run - * unless {@link MappingConflictFixtures#KNOWN_500S} pins its cell to an issue, and a pinned cell - * that stops returning 500 fails too. An unpinned 200 or 4xx passes, since this checks only that - * schema evolution never returns a 500, but a command that never returns 200 fails, since it tests - * nothing. + * unless {@link MappingConflictFixtures#KNOWN_500S} lists it as a known 500 tied to an issue, and a + * known 500 that stops returning 500 fails too. A 200 or 4xx outside {@code KNOWN_500S} passes, + * since this checks only that schema evolution never returns a 500, but a command that never + * returns 200 fails, since it tests nothing. */ public abstract class OtelDataStreamConflictTestCase extends PPLIntegTestCase { @@ -46,7 +46,7 @@ public abstract class OtelDataStreamConflictTestCase extends PPLIntegTestCase { protected abstract OtelDataStream create(String name) throws IOException; /** A keyword field no drift touches, for {@code {K}}. */ - protected abstract String control(); + protected abstract String stableField(); @Override public void init() throws Exception { @@ -55,8 +55,8 @@ public void init() throws Exception { } /** - * Runs every drift under {@code name} and fails on any unpinned 500, stale pin, or command that - * never returns 200. + * Runs every drift under {@code name} and fails on a 500 missing from {@code KNOWN_500S}, a stale + * {@code KNOWN_500S} entry, or a command that never returns 200. */ protected void runDrifts(String name) throws IOException { JSONArray nodes = new JSONArray(executeRequest(new Request("GET", "/_cat/nodes?format=json"))); @@ -66,24 +66,24 @@ protected void runDrifts(String name) throws IOException { } List commands = REPRESENTATIVE; Set names = commands.stream().map(Command::name).collect(Collectors.toSet()); - List cells = new ArrayList<>(); + List observations = new ArrayList<>(); List failures = new ArrayList<>(); for (int i = 0; i < TYPES.size(); i++) { for (int j = i + 1; j < TYPES.size(); j++) { String a = TYPES.get(i); String b = TYPES.get(j); String pair = a + "/" + b; - cells.addAll(pass(name, pair, a, retype(a, b), commands)); - cells.addAll(pass(name, pair, b, retype(b, a), commands)); + observations.addAll(observe(name, pair, a, retype(a, b), commands)); + observations.addAll(observe(name, pair, b, retype(b, a), commands)); } } for (String type : TYPES) { String field = "attributes.types." + type; - cells.addAll( - pass(name, type + "/absent", type, n -> Drift.remove(field, BASE_DOCS), commands)); + observations.addAll( + observe(name, type + "/absent", type, n -> Drift.remove(field, BASE_DOCS), commands)); String added = "added_" + type; - cells.addAll( - pass( + observations.addAll( + observe( name, "absent/" + type, added, @@ -92,74 +92,84 @@ protected void runDrifts(String name) throws IOException { "attributes.types." + added, type, baseValues(n, "types." + type).toArray()), commands)); } - cells.stream() + observations.stream() .filter( - c -> - c.winner().startsWith("unread-") - && (c.pair().contains("absent") || !c.winner().startsWith("unread-4"))) + o -> + o.winner().startsWith("unread-") + && (o.pair().contains("absent") || !o.winner().startsWith("unread-4"))) .forEach( - c -> + o -> failures.add( - c.pair() + " via " + c.source() + " could not read its winner, " + c.winner())); + o.pair() + " via " + o.source() + " could not read its winner, " + o.winner())); names.stream() - .filter(n -> cells.stream().noneMatch(c -> c.command().equals(n) && c.status() == 200)) + .filter( + n -> observations.stream().noneMatch(o -> o.command().equals(n) && o.status() == 200)) .forEach(n -> failures.add(n + " never returned 200, so it tests nothing")); Set serverErrors = new TreeSet<>(); - for (Cell cell : cells) { - if (cell.status() >= 500) { - serverErrors.add(cell.key()); - logger.info("500 {} | {} | {}", cell.key(), cell.source(), cell.error()); + for (Observation observation : observations) { + if (observation.status() >= 500) { + serverErrors.add(observation.key()); + logger.info( + "500 {} | {} | {}", observation.key(), observation.source(), observation.error()); if (KNOWN_500S.stream() - .noneMatch(p -> p.covers(cell.pair(), cell.winner(), cell.command(), cell.source()))) { + .noneMatch( + k -> + k.covers( + observation.pair(), + observation.winner(), + observation.command(), + observation.source()))) { failures.add( - "unpinned 500, " - + cell.label() + "500 missing from KNOWN_500S, " + + observation.label() + "\n query: " - + cell.query() + + observation.query() + "\n error: " - + cell.error().substring(0, Math.min(200, cell.error().length()))); + + observation.error().substring(0, Math.min(200, observation.error().length()))); } } } logger.info( "{} queries, {} return 500 across {} pair, winner and command keys", - cells.size(), - cells.stream().filter(c -> c.status() >= 500).count(), + observations.size(), + observations.stream().filter(o -> o.status() >= 500).count(), serverErrors.size()); - Set pairs = cells.stream().map(Cell::pair).collect(Collectors.toSet()); - for (Pin pin : KNOWN_500S) { - pin.commands().stream() + Set pairs = observations.stream().map(Observation::pair).collect(Collectors.toSet()); + for (Known500 known : KNOWN_500S) { + known.commands().stream() .filter(c -> !names.contains(c)) .forEach( - c -> failures.add("the " + pin.issue() + " pin lists " + c + ", which never runs")); - List covered = - cells.stream() - .filter(c -> pin.covers(c.pair(), c.winner(), c.command(), c.source())) + c -> + failures.add( + "the " + known.issue() + " known 500 lists " + c + ", which never runs")); + List covered = + observations.stream() + .filter(o -> known.covers(o.pair(), o.winner(), o.command(), o.source())) .toList(); - pin.pairs().stream() - .filter(p -> covered.stream().noneMatch(c -> c.pair().equals(p))) + known.pairs().stream() + .filter(p -> covered.stream().noneMatch(o -> o.pair().equals(p))) .forEach( p -> failures.add( "the " - + pin.issue() - + " pin lists " + + known.issue() + + " known 500 lists " + p + (pairs.contains(p) - ? ", which yields no cell under its winner and source, so fix or" + ? ", which yields no query under its winner and source, so fix or" + " remove it" : ", which never runs, so check its spelling and TYPES order"))); covered.stream() - .filter(c -> c.status() < 500) + .filter(o -> o.status() < 500) .forEach( - c -> + o -> failures.add( - "pinned to " - + pin.issue() - + " but no longer returns 500, " - + c.label() - + ", narrow that pin to the cells that still fail\n query: " - + c.query())); + "listed under " + + known.issue() + + " in KNOWN_500S but no longer returns 500, " + + o.label() + + ", narrow that entry to the queries that still fail\n query: " + + o.query())); } if (!failures.isEmpty()) { Set distinct = new TreeSet<>(failures); @@ -183,22 +193,22 @@ private DriftFactory retype(String from, String to) { * Rolls the stream over to the drift and runs every command through the stream name and the * wildcard, each under the winner its merged schema reports. */ - private List pass( + private List observe( String name, String pair, String type, DriftFactory factory, List commands) throws IOException { OtelDataStream stream = create(name); - List cells = new ArrayList<>(); + List observations = new ArrayList<>(); try { String field = "`attributes.types." + type + "`"; stream.rollover(factory.drift(name)); for (String source : List.of(name, name + "*")) { String resolved = winner(source, field); - logger.info("pass {}, field {} via {} reads {}", pair, field, source, resolved); + logger.info("pair {}, field {} via {} reads {}", pair, field, source, resolved); for (Command command : commands) { - String query = command.query(source, field, control()); + String query = command.query(source, field, stableField()); Response response = run(query); - cells.add( - new Cell( + observations.add( + new Observation( pair, resolved, command.name(), @@ -217,7 +227,7 @@ private List pass( throw e; } stream.delete(); - return cells; + return observations; } /** The PPL type the merged schema gives the field, or the status when it cannot be read. */ @@ -278,8 +288,8 @@ private Response run(String query) throws IOException { private record Response(int status, String type, String error) {} - /** One command over one source under one winner. */ - private record Cell( + /** The outcome of one query, meaning one command over one source under one winner. */ + private record Observation( String pair, String winner, String command, @@ -292,7 +302,7 @@ String key() { return pair + " " + winner + " " + command; } - /** The cell as a failure line reads it. */ + /** The observation as a failure line reads it. */ String label() { return "pair " + pair + ", winner " + winner + ", command " + command; } From cef9230ce92314b8e533017c497bdae497a136e8 Mon Sep 17 00:00:00 2001 From: Chayanin Noramuttha Date: Tue, 6 Oct 2026 15:54:21 -0700 Subject: [PATCH 05/10] Split runDrifts into scenarios, observations and named checks runDrifts now observes Scenario.allDrifts(TYPES) into an Observations record, logs the summary, and collects failures from four checks on that record. They cover unreadable winners, commands that never return 200, 500s missing from KNOWN_500S, and stale KNOWN_500S entries. A plant that fires every failure path gave a byte-identical failure message before and after the split, under the same merge read order. Signed-off-by: Chayanin Noramuttha --- .../OtelDataStreamConflictTestCase.java | 301 ++++++++++-------- 1 file changed, 174 insertions(+), 127 deletions(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java index 9072c068606..bdcfc54af2f 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java @@ -31,12 +31,12 @@ * Runs PPL commands over an OTel data stream whose second backing index changes the type of one * {@code attributes.types} field, removes it, or adds one, through the stream name and a wildcard. * Every pair of {@link MappingConflictFixtures#TYPES} runs twice, once with each type in 000001. - * One node reads the backing indices in a fixed order, so the two passes feed the merge both orders - * a two-index stream can produce and cover every winner production can return. A 500 fails the run - * unless {@link MappingConflictFixtures#KNOWN_500S} lists it as a known 500 tied to an issue, and a - * known 500 that stops returning 500 fails too. A 200 or 4xx outside {@code KNOWN_500S} passes, - * since this checks only that schema evolution never returns a 500, but a command that never - * returns 200 fails, since it tests nothing. + * One node reads the backing indices in a fixed order, so the two scenarios of a pair feed the + * merge both orders a two-index stream can produce and cover every winner production can return. A + * 500 fails the run unless {@link MappingConflictFixtures#KNOWN_500S} lists it as a known 500 tied + * to an issue, and a known 500 that stops returning 500 fails too. A 200 or 4xx outside {@code + * KNOWN_500S} passes, since this checks only that schema evolution never returns a 500, but a + * command that never returns 200 fails, since it tests nothing. */ public abstract class OtelDataStreamConflictTestCase extends PPLIntegTestCase { @@ -59,157 +59,102 @@ public void init() throws Exception { * {@code KNOWN_500S} entry, or a command that never returns 200. */ protected void runDrifts(String name) throws IOException { + requireOneNode(); + Observations observed = observe(name, Scenario.allDrifts(TYPES)); + logSummary(observed); + List failures = new ArrayList<>(); + failures.addAll(observed.unreadableWinners()); + failures.addAll(observed.commandsThatNeverReturn200()); + failures.addAll(observed.missingKnown500s()); + failures.addAll(observed.staleKnown500s()); + if (!failures.isEmpty()) { + Set distinct = new TreeSet<>(failures); + fail(distinct.size() + " failures\n" + String.join("\n", distinct)); + } + } + + private void requireOneNode() throws IOException { JSONArray nodes = new JSONArray(executeRequest(new Request("GET", "/_cat/nodes?format=json"))); if (nodes.length() != 1) { throw new IllegalStateException( "The swap needs one node to fix the merge order, the cluster has " + nodes.length()); } - List commands = REPRESENTATIVE; - Set names = commands.stream().map(Command::name).collect(Collectors.toSet()); - List observations = new ArrayList<>(); - List failures = new ArrayList<>(); - for (int i = 0; i < TYPES.size(); i++) { - for (int j = i + 1; j < TYPES.size(); j++) { - String a = TYPES.get(i); - String b = TYPES.get(j); - String pair = a + "/" + b; - observations.addAll(observe(name, pair, a, retype(a, b), commands)); - observations.addAll(observe(name, pair, b, retype(b, a), commands)); - } - } - for (String type : TYPES) { - String field = "attributes.types." + type; - observations.addAll( - observe(name, type + "/absent", type, n -> Drift.remove(field, BASE_DOCS), commands)); - String added = "added_" + type; - observations.addAll( - observe( - name, - "absent/" + type, - added, - n -> - Drift.add( - "attributes.types." + added, type, baseValues(n, "types." + type).toArray()), - commands)); - } - observations.stream() - .filter( - o -> - o.winner().startsWith("unread-") - && (o.pair().contains("absent") || !o.winner().startsWith("unread-4"))) - .forEach( - o -> - failures.add( - o.pair() + " via " + o.source() + " could not read its winner, " + o.winner())); - names.stream() - .filter( - n -> observations.stream().noneMatch(o -> o.command().equals(n) && o.status() == 200)) - .forEach(n -> failures.add(n + " never returned 200, so it tests nothing")); - Set serverErrors = new TreeSet<>(); - for (Observation observation : observations) { - if (observation.status() >= 500) { - serverErrors.add(observation.key()); - logger.info( - "500 {} | {} | {}", observation.key(), observation.source(), observation.error()); - if (KNOWN_500S.stream() - .noneMatch( - k -> - k.covers( - observation.pair(), - observation.winner(), - observation.command(), - observation.source()))) { - failures.add( - "500 missing from KNOWN_500S, " - + observation.label() - + "\n query: " - + observation.query() - + "\n error: " - + observation.error().substring(0, Math.min(200, observation.error().length()))); + } + + /** + * One rollover, meaning the pair its queries report under, the field they read, and the drift + * that 000002 applies. + */ + private record Scenario(String pair, String field, DriftFactory drift) { + + /** + * Every pair of {@code types} twice, once with each type in 000001, then each type removed and + * each type added, so 56 scenarios for 7 types. + */ + static List allDrifts(List types) { + List scenarios = new ArrayList<>(); + for (int i = 0; i < types.size(); i++) { + for (int j = i + 1; j < types.size(); j++) { + String a = types.get(i); + String b = types.get(j); + String pair = a + "/" + b; + scenarios.add(new Scenario(pair, "attributes.types." + a, retype(a, b))); + scenarios.add(new Scenario(pair, "attributes.types." + b, retype(b, a))); } } - } - logger.info( - "{} queries, {} return 500 across {} pair, winner and command keys", - observations.size(), - observations.stream().filter(o -> o.status() >= 500).count(), - serverErrors.size()); - Set pairs = observations.stream().map(Observation::pair).collect(Collectors.toSet()); - for (Known500 known : KNOWN_500S) { - known.commands().stream() - .filter(c -> !names.contains(c)) - .forEach( - c -> - failures.add( - "the " + known.issue() + " known 500 lists " + c + ", which never runs")); - List covered = - observations.stream() - .filter(o -> known.covers(o.pair(), o.winner(), o.command(), o.source())) - .toList(); - known.pairs().stream() - .filter(p -> covered.stream().noneMatch(o -> o.pair().equals(p))) - .forEach( - p -> - failures.add( - "the " - + known.issue() - + " known 500 lists " - + p - + (pairs.contains(p) - ? ", which yields no query under its winner and source, so fix or" - + " remove it" - : ", which never runs, so check its spelling and TYPES order"))); - covered.stream() - .filter(o -> o.status() < 500) - .forEach( - o -> - failures.add( - "listed under " - + known.issue() - + " in KNOWN_500S but no longer returns 500, " - + o.label() - + ", narrow that entry to the queries that still fail\n query: " - + o.query())); - } - if (!failures.isEmpty()) { - Set distinct = new TreeSet<>(failures); - fail(distinct.size() + " failures\n" + String.join("\n", distinct)); + for (String type : types) { + String field = "attributes.types." + type; + scenarios.add(new Scenario(type + "/absent", field, n -> Drift.remove(field, BASE_DOCS))); + String added = "attributes.types.added_" + type; + scenarios.add( + new Scenario( + "absent/" + type, + added, + n -> Drift.add(added, type, baseValues(n, "types." + type).toArray()))); + } + return scenarios; } } /** Builds a drift once the stream exists, since its values come from the base documents. */ @FunctionalInterface private interface DriftFactory { - Drift drift(String name) throws IOException; + Drift build(String name) throws IOException; } /** Changes {@code from}'s field to {@code to}, filled with {@code to}'s base values. */ - private DriftFactory retype(String from, String to) { + private static DriftFactory retype(String from, String to) { return n -> Drift.changeType("attributes.types." + from, to, baseValues(n, "types." + to).toArray()); } + private Observations observe(String name, List scenarios) throws IOException { + List all = new ArrayList<>(); + for (Scenario scenario : scenarios) { + all.addAll(observe(name, scenario)); + } + return new Observations(all); + } + /** - * Rolls the stream over to the drift and runs every command through the stream name and the - * wildcard, each under the winner its merged schema reports. + * Creates the stream, rolls it over to the scenario's drift, and runs every command through the + * stream name and the wildcard, each under the winner its merged schema reports. */ - private List observe( - String name, String pair, String type, DriftFactory factory, List commands) - throws IOException { + private List observe(String name, Scenario scenario) throws IOException { OtelDataStream stream = create(name); List observations = new ArrayList<>(); try { - String field = "`attributes.types." + type + "`"; - stream.rollover(factory.drift(name)); + String field = "`" + scenario.field() + "`"; + stream.rollover(scenario.drift().build(name)); for (String source : List.of(name, name + "*")) { String resolved = winner(source, field); - logger.info("pair {}, field {} via {} reads {}", pair, field, source, resolved); - for (Command command : commands) { + logger.info("pair {}, field {} via {} reads {}", scenario.pair(), field, source, resolved); + for (Command command : REPRESENTATIVE) { String query = command.query(source, field, stableField()); Response response = run(query); observations.add( new Observation( - pair, + scenario.pair(), resolved, command.name(), source, @@ -230,6 +175,104 @@ private List observe( return observations; } + private void logSummary(Observations observed) { + Set keys = new TreeSet<>(); + for (Observation o : observed.all()) { + if (o.status() >= 500) { + keys.add(o.key()); + logger.info("500 {} | {} | {}", o.key(), o.source(), o.error()); + } + } + logger.info( + "{} queries, {} return 500 across {} pair, winner and command keys", + observed.all().size(), + observed.all().stream().filter(o -> o.status() >= 500).count(), + keys.size()); + } + + /** Every observation of a run, with the checks that turn them into failure lines. */ + private record Observations(List all) { + + /** + * A winner probe that failed. A 4xx on a type pair passes, since a fix may reject the conflict, + * but not on a pair with an absent side, where nothing conflicts. + */ + List unreadableWinners() { + return all.stream() + .filter( + o -> + o.winner().startsWith("unread-") + && (o.pair().contains("absent") || !o.winner().startsWith("unread-4"))) + .map(o -> o.pair() + " via " + o.source() + " could not read its winner, " + o.winner()) + .toList(); + } + + List commandsThatNeverReturn200() { + return REPRESENTATIVE.stream() + .map(Command::name) + .filter(n -> all.stream().noneMatch(o -> o.command().equals(n) && o.status() == 200)) + .map(n -> n + " never returned 200, so it tests nothing") + .toList(); + } + + List missingKnown500s() { + return all.stream() + .filter(o -> o.status() >= 500 && KNOWN_500S.stream().noneMatch(o::isCoveredBy)) + .map( + o -> + "500 missing from KNOWN_500S, " + + o.label() + + "\n query: " + + o.query() + + "\n error: " + + o.error().substring(0, Math.min(200, o.error().length()))) + .toList(); + } + + /** + * A {@code KNOWN_500S} entry that lists a command or pair the run never reaches, or covers a + * query that no longer returns 500. + */ + List staleKnown500s() { + Set pairs = all.stream().map(Observation::pair).collect(Collectors.toSet()); + List failures = new ArrayList<>(); + for (Known500 known : KNOWN_500S) { + known.commands().stream() + .filter(c -> REPRESENTATIVE.stream().noneMatch(command -> command.name().equals(c))) + .forEach( + c -> + failures.add( + "the " + known.issue() + " known 500 lists " + c + ", which never runs")); + List covered = all.stream().filter(o -> o.isCoveredBy(known)).toList(); + known.pairs().stream() + .filter(p -> covered.stream().noneMatch(o -> o.pair().equals(p))) + .forEach( + p -> + failures.add( + "the " + + known.issue() + + " known 500 lists " + + p + + (pairs.contains(p) + ? ", which yields no query under its winner and source, so fix or" + + " remove it" + : ", which never runs, so check its spelling and TYPES order"))); + covered.stream() + .filter(o -> o.status() < 500) + .forEach( + o -> + failures.add( + "listed under " + + known.issue() + + " in KNOWN_500S but no longer returns 500, " + + o.label() + + ", narrow that entry to the queries that still fail\n query: " + + o.query())); + } + return failures; + } + } + /** The PPL type the merged schema gives the field, or the status when it cannot be read. */ private String winner(String source, String field) throws IOException { Response response = run("source=" + source + " | fields " + field + " | head 1"); @@ -239,7 +282,7 @@ private String winner(String source, String field) throws IOException { } /** Each base document's value for a flattened {@code attributes} key. */ - private List baseValues(String name, String key) throws IOException { + private static List baseValues(String name, String key) throws IOException { JSONObject body = new JSONObject( getResponseBody( @@ -306,5 +349,9 @@ String key() { String label() { return "pair " + pair + ", winner " + winner + ", command " + command; } + + boolean isCoveredBy(Known500 known) { + return known.covers(pair, winner, command, source); + } } } From 9e04e3be26abf7cb66bf151ba4550f29d9e4517a Mon Sep 17 00:00:00 2001 From: Chayanin Noramuttha Date: Tue, 6 Oct 2026 16:05:45 -0700 Subject: [PATCH 06/10] Explain why the mapping conflict ITs need one node The merge order comes from the index names and a per-JVM salt, so one node keeps every scenario on one order and a/b then b/a lets each type win once. The node check message drops the undefined word swap. Signed-off-by: Chayanin Noramuttha --- .../calcite/remote/OtelDataStreamConflictTestCase.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java index bdcfc54af2f..23154be6bef 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java @@ -73,11 +73,18 @@ protected void runDrifts(String name) throws IOException { } } + /** + * The merged schema keeps the type of the backing index it iterates last, in an order set by the + * index names and a salt each JVM draws at start. Every scenario recreates the same two names, so + * one node gives them all one order, and running a pair as a/b and then b/a lets each type win + * once. Each node draws its own salt and requests spread across nodes, so several nodes break it. + */ private void requireOneNode() throws IOException { JSONArray nodes = new JSONArray(executeRequest(new Request("GET", "/_cat/nodes?format=json"))); if (nodes.length() != 1) { throw new IllegalStateException( - "The swap needs one node to fix the merge order, the cluster has " + nodes.length()); + "The scenarios need one node to hold the merge order fixed, the cluster has " + + nodes.length()); } } From 8d29e3621d5dcd16a5315014cfc0e6f1913c0801 Mon Sep 17 00:00:00 2001 From: Chayanin Noramuttha Date: Tue, 6 Oct 2026 16:35:09 -0700 Subject: [PATCH 07/10] Define drift, pair, winner and key in the mapping conflict ITs Signed-off-by: Chayanin Noramuttha --- .../remote/OtelDataStreamConflictTestCase.java | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java index 23154be6bef..3c01078e770 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java @@ -30,11 +30,13 @@ /** * Runs PPL commands over an OTel data stream whose second backing index changes the type of one * {@code attributes.types} field, removes it, or adds one, through the stream name and a wildcard. - * Every pair of {@link MappingConflictFixtures#TYPES} runs twice, once with each type in 000001. - * One node reads the backing indices in a fixed order, so the two scenarios of a pair feed the - * merge both orders a two-index stream can produce and cover every winner production can return. A - * 500 fails the run unless {@link MappingConflictFixtures#KNOWN_500S} lists it as a known 500 tied - * to an issue, and a known 500 that stops returning 500 fails too. A 200 or 4xx outside {@code + * That change is the drift. A pair names the two types the field takes, with {@code absent} for the + * index that lacks it, and the winner is the type the merged schema reports for the field. Every + * pair of {@link MappingConflictFixtures#TYPES} runs twice, once with each type in 000001. One node + * reads the backing indices in a fixed order, so the two scenarios of a pair feed the merge both + * orders a two-index stream can produce and cover every winner production can return. A 500 fails + * the run unless {@link MappingConflictFixtures#KNOWN_500S} lists it as a known 500 tied to an + * issue, and a known 500 that stops returning 500 fails too. A 200 or 4xx outside {@code * KNOWN_500S} passes, since this checks only that schema evolution never returns a 500, but a * command that never returns 200 fails, since it tests nothing. */ @@ -348,6 +350,7 @@ private record Observation( int status, String error) { + /** The pair, winner and command, the unit the run counts 500s by. */ String key() { return pair + " " + winner + " " + command; } From 4f988ecca763548d184011f8efe2693a8d578524 Mon Sep 17 00:00:00 2001 From: Chayanin Noramuttha Date: Tue, 6 Oct 2026 20:46:42 -0700 Subject: [PATCH 08/10] Clarify names and Javadocs in the mapping conflict ITs Rename missingKnown500s to unlisted500s, the per-scenario observe to observeScenario, and Scenario.drift to driftFactory. Split staleKnown500s into commandsThatNeverRun, unreachedPairs and noLonger500s. Move the drift, pair and winner definitions onto Scenario and Observation, state the pair order rule, and name the pairs an earlier merge rule decides. A plant that fires every failure path gave a byte-identical failure message before and after, under the same merge read order. Signed-off-by: Chayanin Noramuttha --- integ-test/build.gradle | 6 +- .../remote/MappingConflictFixtures.java | 2 +- .../OtelDataStreamConflictTestCase.java | 166 +++++++++++------- 3 files changed, 102 insertions(+), 72 deletions(-) diff --git a/integ-test/build.gradle b/integ-test/build.gradle index 44aeee23883..dbff64884f8 100644 --- a/integ-test/build.gradle +++ b/integ-test/build.gradle @@ -1645,8 +1645,8 @@ task integTestRemote(type: RestIntegTestTask) { exclude 'org/opensearch/sql/datasource/DataSourceAPIsIT.class' exclude 'org/opensearch/sql/datasource/DataSourceEnabledIT.class' - // Suites that pin the 500s of one build on a single node with the composite assert off. An - // external cluster can add nodes, which redraws the merge winner per request, and cannot drop - // the assert, so they run only in :mappingConflictIntegTest. + // Suites that list the known 500s of one build on a single node with the InternalComposite + // asserts off. An external cluster can add nodes, which redraws the merge winner per request, + // and cannot drop the assert, so they run only in :mappingConflictIntegTest. exclude 'org/opensearch/sql/calcite/remote/CalciteOtel*DataStreamIT.class' } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java index bcb6f34ef21..3e649fbf0a6 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java @@ -20,7 +20,7 @@ final class MappingConflictFixtures { /** * A PPL command template, holding {@code {I}} for the source, {@code {F}} for the field under - * test and {@code {K}} for a stable field that no drift touches. + * test and {@code {K}} for a keyword field that no drift touches. */ record Command(String name, String template) { String query(String source, String field, String stableField) { diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java index 3c01078e770..093f6649122 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java @@ -30,15 +30,11 @@ /** * Runs PPL commands over an OTel data stream whose second backing index changes the type of one * {@code attributes.types} field, removes it, or adds one, through the stream name and a wildcard. - * That change is the drift. A pair names the two types the field takes, with {@code absent} for the - * index that lacks it, and the winner is the type the merged schema reports for the field. Every - * pair of {@link MappingConflictFixtures#TYPES} runs twice, once with each type in 000001. One node - * reads the backing indices in a fixed order, so the two scenarios of a pair feed the merge both - * orders a two-index stream can produce and cover every winner production can return. A 500 fails - * the run unless {@link MappingConflictFixtures#KNOWN_500S} lists it as a known 500 tied to an - * issue, and a known 500 that stops returning 500 fails too. A 200 or 4xx outside {@code - * KNOWN_500S} passes, since this checks only that schema evolution never returns a 500, but a - * command that never returns 200 fails, since it tests nothing. + * Every pair of {@link MappingConflictFixtures#TYPES} runs twice, once with each type in 000001, to + * cover every type the merged schema can keep. A 500 fails the run unless {@link + * MappingConflictFixtures#KNOWN_500S} lists it as a known 500 tied to an issue, and a known 500 + * that stops returning 500 fails too. A command's 200 or 4xx outside {@code KNOWN_500S} passes, + * since this checks only for 500s. A command that never returns 200 fails, since it tests nothing. */ public abstract class OtelDataStreamConflictTestCase extends PPLIntegTestCase { @@ -67,7 +63,7 @@ protected void runDrifts(String name) throws IOException { List failures = new ArrayList<>(); failures.addAll(observed.unreadableWinners()); failures.addAll(observed.commandsThatNeverReturn200()); - failures.addAll(observed.missingKnown500s()); + failures.addAll(observed.unlisted500s()); failures.addAll(observed.staleKnown500s()); if (!failures.isEmpty()) { Set distinct = new TreeSet<>(failures); @@ -76,10 +72,11 @@ protected void runDrifts(String name) throws IOException { } /** - * The merged schema keeps the type of the backing index it iterates last, in an order set by the - * index names and a salt each JVM draws at start. Every scenario recreates the same two names, so - * one node gives them all one order, and running a pair as a/b and then b/a lets each type win - * once. Each node draws its own salt and requests spread across nodes, so several nodes break it. + * Unless an earlier merge rule decides the pair, as for text/keyword, the merged schema keeps the + * type of the backing index it iterates last, in an order set by the index names and a salt each + * JVM draws at start. Every scenario recreates the same two names, so one node gives them all one + * order, and running each pair once with each type in 000001 lets each type win once. Each node + * draws its own salt and requests spread across nodes, so several nodes break it. */ private void requireOneNode() throws IOException { JSONArray nodes = new JSONArray(executeRequest(new Request("GET", "/_cat/nodes?format=json"))); @@ -91,14 +88,16 @@ private void requireOneNode() throws IOException { } /** - * One rollover, meaning the pair its queries report under, the field they read, and the drift - * that 000002 applies. + * One rollover, where the drift is what 000002 does to the field. A type pair names the field's + * two types in {@code TYPES} order, whichever one sits in 000001. A removed or added field puts + * {@code absent} on the side of the index that lacks it, so {@code object/absent} removes it and + * {@code absent/object} adds it. */ - private record Scenario(String pair, String field, DriftFactory drift) { + private record Scenario(String pair, String field, DriftFactory driftFactory) { /** * Every pair of {@code types} twice, once with each type in 000001, then each type removed and - * each type added, so 56 scenarios for 7 types. + * each type added. */ static List allDrifts(List types) { List scenarios = new ArrayList<>(); @@ -113,13 +112,15 @@ static List allDrifts(List types) { } for (String type : types) { String field = "attributes.types." + type; - scenarios.add(new Scenario(type + "/absent", field, n -> Drift.remove(field, BASE_DOCS))); + scenarios.add( + new Scenario(type + "/absent", field, streamName -> Drift.remove(field, BASE_DOCS))); String added = "attributes.types.added_" + type; scenarios.add( new Scenario( "absent/" + type, added, - n -> Drift.add(added, type, baseValues(n, "types." + type).toArray()))); + streamName -> + Drift.add(added, type, baseValues(streamName, "types." + type).toArray()))); } return scenarios; } @@ -128,19 +129,20 @@ static List allDrifts(List types) { /** Builds a drift once the stream exists, since its values come from the base documents. */ @FunctionalInterface private interface DriftFactory { - Drift build(String name) throws IOException; + Drift build(String streamName) throws IOException; } /** Changes {@code from}'s field to {@code to}, filled with {@code to}'s base values. */ private static DriftFactory retype(String from, String to) { - return n -> - Drift.changeType("attributes.types." + from, to, baseValues(n, "types." + to).toArray()); + return streamName -> + Drift.changeType( + "attributes.types." + from, to, baseValues(streamName, "types." + to).toArray()); } private Observations observe(String name, List scenarios) throws IOException { List all = new ArrayList<>(); for (Scenario scenario : scenarios) { - all.addAll(observe(name, scenario)); + all.addAll(observeScenario(name, scenario)); } return new Observations(all); } @@ -149,12 +151,12 @@ private Observations observe(String name, List scenarios) throws IOExc * Creates the stream, rolls it over to the scenario's drift, and runs every command through the * stream name and the wildcard, each under the winner its merged schema reports. */ - private List observe(String name, Scenario scenario) throws IOException { + private List observeScenario(String name, Scenario scenario) throws IOException { OtelDataStream stream = create(name); List observations = new ArrayList<>(); try { String field = "`" + scenario.field() + "`"; - stream.rollover(scenario.drift().build(name)); + stream.rollover(scenario.driftFactory().build(name)); for (String source : List.of(name, name + "*")) { String resolved = winner(source, field); logger.info("pair {}, field {} via {} reads {}", scenario.pair(), field, source, resolved); @@ -186,8 +188,10 @@ private List observe(String name, Scenario scenario) throws IOExcep private void logSummary(Observations observed) { Set keys = new TreeSet<>(); + int serverErrors = 0; for (Observation o : observed.all()) { if (o.status() >= 500) { + serverErrors++; keys.add(o.key()); logger.info("500 {} | {} | {}", o.key(), o.source(), o.error()); } @@ -195,7 +199,7 @@ private void logSummary(Observations observed) { logger.info( "{} queries, {} return 500 across {} pair, winner and command keys", observed.all().size(), - observed.all().stream().filter(o -> o.status() >= 500).count(), + serverErrors, keys.size()); } @@ -203,8 +207,8 @@ private void logSummary(Observations observed) { private record Observations(List all) { /** - * A winner probe that failed. A 4xx on a type pair passes, since a fix may reject the conflict, - * but not on a pair with an absent side, where nothing conflicts. + * A {@code winner} query that did not return a type. A 4xx on a type pair passes, since a fix + * may reject the conflict, but not on a pair with an absent side, where nothing conflicts. */ List unreadableWinners() { return all.stream() @@ -213,6 +217,7 @@ List unreadableWinners() { o.winner().startsWith("unread-") && (o.pair().contains("absent") || !o.winner().startsWith("unread-4"))) .map(o -> o.pair() + " via " + o.source() + " could not read its winner, " + o.winner()) + .distinct() .toList(); } @@ -224,7 +229,8 @@ List commandsThatNeverReturn200() { .toList(); } - List missingKnown500s() { + /** A 500 that no {@code KNOWN_500S} entry lists. */ + List unlisted500s() { return all.stream() .filter(o -> o.status() >= 500 && KNOWN_500S.stream().noneMatch(o::isCoveredBy)) .map( @@ -239,50 +245,65 @@ List missingKnown500s() { } /** - * A {@code KNOWN_500S} entry that lists a command or pair the run never reaches, or covers a - * query that no longer returns 500. + * A {@code KNOWN_500S} entry that lists a command or pair the run never reaches, lists a pair + * with no query under its winner and source, or covers a query that no longer returns 500. */ List staleKnown500s() { Set pairs = all.stream().map(Observation::pair).collect(Collectors.toSet()); List failures = new ArrayList<>(); for (Known500 known : KNOWN_500S) { - known.commands().stream() - .filter(c -> REPRESENTATIVE.stream().noneMatch(command -> command.name().equals(c))) - .forEach( - c -> - failures.add( - "the " + known.issue() + " known 500 lists " + c + ", which never runs")); List covered = all.stream().filter(o -> o.isCoveredBy(known)).toList(); - known.pairs().stream() - .filter(p -> covered.stream().noneMatch(o -> o.pair().equals(p))) - .forEach( - p -> - failures.add( - "the " - + known.issue() - + " known 500 lists " - + p - + (pairs.contains(p) - ? ", which yields no query under its winner and source, so fix or" - + " remove it" - : ", which never runs, so check its spelling and TYPES order"))); - covered.stream() - .filter(o -> o.status() < 500) - .forEach( - o -> - failures.add( - "listed under " - + known.issue() - + " in KNOWN_500S but no longer returns 500, " - + o.label() - + ", narrow that entry to the queries that still fail\n query: " - + o.query())); + failures.addAll(commandsThatNeverRun(known)); + failures.addAll(unreachedPairs(known, covered, pairs)); + failures.addAll(noLonger500s(known, covered)); } return failures; } + + private static List commandsThatNeverRun(Known500 known) { + return known.commands().stream() + .filter(c -> REPRESENTATIVE.stream().noneMatch(command -> command.name().equals(c))) + .map(c -> "the " + known.issue() + " known 500 lists " + c + ", which never runs") + .toList(); + } + + /** A listed pair that never runs, or yields no query under the entry's winner and source. */ + private static List unreachedPairs( + Known500 known, List covered, Set pairs) { + return known.pairs().stream() + .filter(p -> covered.stream().noneMatch(o -> o.pair().equals(p))) + .map( + p -> + "the " + + known.issue() + + " known 500 lists " + + p + + (pairs.contains(p) + ? ", which yields no query under its winner and source, so fix or" + + " remove it" + : ", which never runs, so check its spelling and TYPES order")) + .toList(); + } + + private static List noLonger500s(Known500 known, List covered) { + return covered.stream() + .filter(o -> o.status() < 500) + .map( + o -> + "listed under " + + known.issue() + + " in KNOWN_500S but no longer returns 500, " + + o.label() + + ", narrow that entry to the queries that still fail\n query: " + + o.query()) + .toList(); + } } - /** The PPL type the merged schema gives the field, or the status when it cannot be read. */ + /** + * The PPL type the merged schema gives the field, or {@code unread-} when the query + * returns no type, so every 4xx starts with {@code unread-4}. + */ private String winner(String source, String field) throws IOException { Response response = run("source=" + source + " | fields " + field + " | head 1"); return response.status() == 200 && !response.type().isEmpty() @@ -291,20 +312,24 @@ private String winner(String source, String field) throws IOException { } /** Each base document's value for a flattened {@code attributes} key. */ - private static List baseValues(String name, String key) throws IOException { + private static List baseValues(String streamName, String attribute) throws IOException { JSONObject body = new JSONObject( getResponseBody( - client().performRequest(new Request("GET", "/" + name + "/_search?size=100")))); + client() + .performRequest(new Request("GET", "/" + streamName + "/_search?size=100")))); JSONArray hits = body.getJSONObject("hits").getJSONArray("hits"); List values = new ArrayList<>(); for (int i = 0; i < hits.length(); i++) { values.add( - hits.getJSONObject(i).getJSONObject("_source").getJSONObject("attributes").get(key)); + hits.getJSONObject(i) + .getJSONObject("_source") + .getJSONObject("attributes") + .get(attribute)); } if (values.size() != BASE_DOCS) { throw new IllegalStateException( - name + " holds " + values.size() + " base documents, expected " + BASE_DOCS); + streamName + " holds " + values.size() + " base documents, expected " + BASE_DOCS); } return values; } @@ -340,7 +365,12 @@ private Response run(String query) throws IOException { private record Response(int status, String type, String error) {} - /** The outcome of one query, meaning one command over one source under one winner. */ + /** + * The outcome of one query, meaning one command over one source under one winner. The winner is + * the PPL type, such as {@code string} or {@code struct}, that the merged schema reports for the + * field, or the status that {@link OtelDataStreamConflictTestCase#winner(String, String)} encodes + * when it cannot be read. + */ private record Observation( String pair, String winner, From 61726e39a6ff70e908f78824bd77e9cab7bd7848 Mon Sep 17 00:00:00 2001 From: Chayanin Noramuttha Date: Tue, 6 Oct 2026 23:05:54 -0700 Subject: [PATCH 09/10] Drop the stable field from the mapping conflict ITs Five command templates read a second keyword field, {K}, next to the field under test. The run gives the same 500s without it, so the templates now read only {F}, and stableField() is gone from the base class and both ITs. Both ITs still run 2128 queries with 186 returning 500 across 93 keys, every one listed in KNOWN_500S. Signed-off-by: Chayanin Noramuttha --- .../remote/CalciteOtelLogsDataStreamIT.java | 5 ----- .../remote/CalciteOtelTracesDataStreamIT.java | 5 ----- .../remote/MappingConflictFixtures.java | 19 +++++++++---------- .../OtelDataStreamConflictTestCase.java | 5 +---- 4 files changed, 10 insertions(+), 24 deletions(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java index 1717c04a8c6..a569a93f7e2 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java @@ -17,11 +17,6 @@ protected OtelDataStream create(String name) throws IOException { return OtelDataStream.logs(client(), name); } - @Override - protected String stableField() { - return "`resource.attributes.service.name`"; - } - @Test public void typeDrifts() throws IOException { runDrifts("otel-logs-drift"); diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java index 81c6309ddfe..fa6829375b4 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java @@ -17,11 +17,6 @@ protected OtelDataStream create(String name) throws IOException { return OtelDataStream.traces(client(), name); } - @Override - protected String stableField() { - return "serviceName"; - } - @Test public void typeDrifts() throws IOException { runDrifts("otel-traces-drift"); diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java index 3e649fbf0a6..a386088db5a 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java @@ -19,12 +19,12 @@ final class MappingConflictFixtures { List.of("text", "keyword", "byte", "integer", "long", "date", "object"); /** - * A PPL command template, holding {@code {I}} for the source, {@code {F}} for the field under - * test and {@code {K}} for a keyword field that no drift touches. + * A PPL command template, holding {@code {I}} for the source and {@code {F}} for the field under + * test. */ record Command(String name, String template) { - String query(String source, String field, String stableField) { - return template.replace("{I}", source).replace("{F}", field).replace("{K}", stableField); + String query(String source, String field) { + return template.replace("{I}", source).replace("{F}", field); } } @@ -34,7 +34,7 @@ String query(String source, String field, String stableField) { */ static final List REPRESENTATIVE = List.of( - new Command("where_isnotnull", "source={I} | where isnotnull({F}) | fields {K}"), + new Command("where_isnotnull", "source={I} | where isnotnull({F})"), new Command("fields_keep", "source={I} | fields {F}"), new Command("eval_cast_string", "source={I} | eval x = cast({F} as string) | fields x"), new Command("stats_count_by", "source={I} | stats count() by {F}"), @@ -42,13 +42,12 @@ String query(String source, String field, String stableField) { new Command("stats_dc", "source={I} | stats dc({F})"), new Command("stats", "source={I} | stats count()"), new Command("eventstats_by", "source={I} | eventstats count() by {F}"), - new Command( - "streamstats_count_by", "source={I} | streamstats count() by {F} | fields {K}"), + new Command("streamstats_count_by", "source={I} | streamstats count() by {F}"), new Command("timechart_by", "source={I} | timechart span=1d count() by {F}"), - new Command("chart_by", "source={I} | chart count() over {K} by {F}"), + new Command("chart_by", "source={I} | chart count() by {F}"), new Command("sort", "source={I} | sort {F}"), - new Command("reverse_after_sort", "source={I} | sort {F} | reverse | fields {K}"), - new Command("head_after_sort", "source={I} | sort {F} | head 1 | fields {K}"), + new Command("reverse_after_sort", "source={I} | sort {F} | reverse"), + new Command("head_after_sort", "source={I} | sort {F} | head 1"), new Command("dedup", "source={I} | dedup {F}"), new Command("top", "source={I} | top {F}"), new Command("rare", "source={I} | rare {F}"), diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java index 093f6649122..5fdc24e42e8 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java @@ -43,9 +43,6 @@ public abstract class OtelDataStreamConflictTestCase extends PPLIntegTestCase { /** Creates the stream with its base documents in 000001. */ protected abstract OtelDataStream create(String name) throws IOException; - /** A keyword field no drift touches, for {@code {K}}. */ - protected abstract String stableField(); - @Override public void init() throws Exception { super.init(); @@ -161,7 +158,7 @@ private List observeScenario(String name, Scenario scenario) throws String resolved = winner(source, field); logger.info("pair {}, field {} via {} reads {}", scenario.pair(), field, source, resolved); for (Command command : REPRESENTATIVE) { - String query = command.query(source, field, stableField()); + String query = command.query(source, field); Response response = run(query); observations.add( new Observation( From 3a325f2950551898dc5d29eeb8d84022dd392167 Mon Sep 17 00:00:00 2001 From: Chayanin Noramuttha Date: Thu, 8 Oct 2026 22:52:00 -0700 Subject: [PATCH 10/10] Re-run CI after the security fixture fix in #5847 Signed-off-by: Chayanin Noramuttha