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..dbff64884f8 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 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/CalciteOtelLogsDataStreamIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java new file mode 100644 index 00000000000..a569a93f7e2 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelLogsDataStreamIT.java @@ -0,0 +1,24 @@ +/* + * 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); + } + + @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..fa6829375b4 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteOtelTracesDataStreamIT.java @@ -0,0 +1,24 @@ +/* + * 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); + } + + @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..a386088db5a --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/MappingConflictFixtures.java @@ -0,0 +1,142 @@ +/* + * 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 known 500s with 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 and {@code {F}} for the field under + * test. + */ + record Command(String name, String template) { + String query(String source, String field) { + return template.replace("{I}", source).replace("{F}", field); + } + } + + /** + * 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})"), + 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}"), + new Command("timechart_by", "source={I} | timechart span=1d count() 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"), + 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}"), + 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")); + + /** The source a known 500 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("*"); + } + } + + /** + * 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 Known500( + String issue, List commands, List pairs, String winner, Source source) { + Known500 { + if (commands.isEmpty() || pairs.isEmpty()) { + throw new IllegalArgumentException("The " + issue + " known 500 needs commands and pairs"); + } + } + + /** 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 observedWinner, String command, String observedSource) { + return pairs.contains(pair) + && commands.contains(command) + && (winner == null || winner.equals(observedWinner)) + && source.matches(observedSource); + } + } + + /** + * 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 = + List.of( + // #5811 Composite aggregation pushdown does not support a field whose type differs across + // indices + 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 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 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 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 Known500( + "#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..5fdc24e42e8 --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/OtelDataStreamConflictTestCase.java @@ -0,0 +1,394 @@ +/* + * 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.Known500; +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, 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 { + + private static final int BASE_DOCS = 5; + + /** Creates the stream with its base documents in 000001. */ + protected abstract OtelDataStream create(String name) throws IOException; + + @Override + public void init() throws Exception { + super.init(); + enableCalcite(); + } + + /** + * 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 { + 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.unlisted500s()); + failures.addAll(observed.staleKnown500s()); + if (!failures.isEmpty()) { + Set distinct = new TreeSet<>(failures); + fail(distinct.size() + " failures\n" + String.join("\n", distinct)); + } + } + + /** + * 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"))); + if (nodes.length() != 1) { + throw new IllegalStateException( + "The scenarios need one node to hold the merge order fixed, the cluster has " + + nodes.length()); + } + } + + /** + * 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 driftFactory) { + + /** + * Every pair of {@code types} twice, once with each type in 000001, then each type removed and + * each type added. + */ + 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))); + } + } + for (String type : types) { + String field = "attributes.types." + type; + 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, + streamName -> + Drift.add(added, type, baseValues(streamName, "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 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 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(observeScenario(name, scenario)); + } + return new Observations(all); + } + + /** + * 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 observeScenario(String name, Scenario scenario) throws IOException { + OtelDataStream stream = create(name); + List observations = new ArrayList<>(); + try { + String field = "`" + scenario.field() + "`"; + 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); + for (Command command : REPRESENTATIVE) { + String query = command.query(source, field); + Response response = run(query); + observations.add( + new Observation( + scenario.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 observations; + } + + 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()); + } + } + logger.info( + "{} queries, {} return 500 across {} pair, winner and command keys", + observed.all().size(), + serverErrors, + keys.size()); + } + + /** Every observation of a run, with the checks that turn them into failure lines. */ + private record Observations(List all) { + + /** + * 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() + .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()) + .distinct() + .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(); + } + + /** 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( + 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, 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) { + List covered = all.stream().filter(o -> o.isCoveredBy(known)).toList(); + 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 {@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() + ? response.type() + : "unread-" + response.status(); + } + + /** Each base document's value for a flattened {@code attributes} key. */ + private static List baseValues(String streamName, String attribute) throws IOException { + JSONObject body = + new JSONObject( + getResponseBody( + 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(attribute)); + } + if (values.size() != BASE_DOCS) { + throw new IllegalStateException( + streamName + " 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) {} + + /** + * 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, + String command, + String source, + String query, + int status, + String error) { + + /** The pair, winner and command, the unit the run counts 500s by. */ + String key() { + return pair + " " + winner + " " + command; + } + + /** The observation as a failure line reads it. */ + String label() { + return "pair " + pair + ", winner " + winner + ", command " + command; + } + + boolean isCoveredBy(Known500 known) { + return known.covers(pair, winner, command, source); + } + } +}