From 35bf78fb58ba97040f09fb350d521ab46f16e446 Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Wed, 20 May 2026 13:24:35 -0700 Subject: [PATCH 1/4] fix: instantiate DatetimeUdt normalize/output-cast rules per plan() call MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DatetimeUdtNormalizeRule and DatetimeOutputCastRule extend RelHomogeneousShuttle, which inherits a stateful Deque stack from RelShuttleImpl. DatetimeExtension.postAnalysisRules() returned the static INSTANCE of each rule, sharing the same shuttle (and the same stack) across every UnifiedQueryPlanner.plan() invocation. If any traversal ever ends with an unbalanced stack, residual entries persist to the next query. The next query's visitChild() then pops a stale or empty stack and throws NoSuchElementException at RelShuttleImpl.visitChild line 67 (the stack.pop() in the finally block) — surfacing as the cluster-side stack trace reported on analytics-engine-routed parquet indices for queries that combine aggregations over datetime UDT columns (e.g. "stats count() as field_count, distinct_count(field)"). Return fresh instances per plan() instead. Drop the INSTANCE constants and the Lombok @NoArgsConstructor on both rules; document the singleton-unsafety on each class JavaDoc. Add a regression test that runs several plan() calls in sequence against the same context, covering stats+distinct_count over both schema-declared and eval-derived datetime columns. Signed-off-by: Kai Huang --- .../api/spec/datetime/DatetimeExtension.java | 3 ++- .../spec/datetime/DatetimeOutputCastRule.java | 12 +++++------ .../datetime/DatetimeUdtNormalizeRule.java | 8 +++---- .../spec/datetime/DatetimeExtensionTest.java | 21 +++++++++++++++++++ 4 files changed, 32 insertions(+), 12 deletions(-) diff --git a/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeExtension.java b/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeExtension.java index 944ac4a4bf1..1f0b0b820a5 100644 --- a/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeExtension.java +++ b/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeExtension.java @@ -22,7 +22,8 @@ public class DatetimeExtension implements LanguageExtension { @Override public List postAnalysisRules() { - return List.of(DatetimeUdtNormalizeRule.INSTANCE, DatetimeOutputCastRule.INSTANCE); + // Fresh instances per plan() because RelHomogeneousShuttle inherits a stateful stack. + return List.of(new DatetimeUdtNormalizeRule(), new DatetimeOutputCastRule()); } /** Maps datetime UDT types to their standard Calcite equivalents. */ diff --git a/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeOutputCastRule.java b/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeOutputCastRule.java index 9a7ae25e003..edc418928cb 100644 --- a/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeOutputCastRule.java +++ b/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeOutputCastRule.java @@ -9,8 +9,6 @@ import java.util.ArrayList; import java.util.List; -import lombok.AccessLevel; -import lombok.NoArgsConstructor; import org.apache.calcite.rel.RelHomogeneousShuttle; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.logical.LogicalProject; @@ -21,12 +19,14 @@ import org.apache.calcite.rex.RexNode; import org.apache.calcite.sql.type.SqlTypeName; -/** Wraps the root output with CAST(datetime → VARCHAR) for PPL wire-format compatibility. */ -@NoArgsConstructor(access = AccessLevel.PRIVATE) +/** + * Wraps the root output with CAST(datetime → VARCHAR) for PPL wire-format compatibility. + * + *

Not a singleton: {@link RelHomogeneousShuttle} inherits a stateful {@code stack} field from + * {@link org.apache.calcite.rel.RelShuttleImpl}, so a fresh instance must be used per plan(). + */ class DatetimeOutputCastRule extends RelHomogeneousShuttle { - static final DatetimeOutputCastRule INSTANCE = new DatetimeOutputCastRule(); - @Override public RelNode visit(RelNode other) { List fields = other.getRowType().getFieldList(); diff --git a/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeUdtNormalizeRule.java b/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeUdtNormalizeRule.java index b15d830d412..7fb8a488c6e 100644 --- a/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeUdtNormalizeRule.java +++ b/api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeUdtNormalizeRule.java @@ -6,8 +6,6 @@ package org.opensearch.sql.api.spec.datetime; import java.util.Optional; -import lombok.AccessLevel; -import lombok.NoArgsConstructor; import org.apache.calcite.rel.RelHomogeneousShuttle; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.type.RelDataType; @@ -22,12 +20,12 @@ /** * Temporary patch that rewrites datetime UDT return types on RexCall nodes to standard Calcite * types. + * + *

Not a singleton: {@link RelHomogeneousShuttle} inherits a stateful {@code stack} field from + * {@link org.apache.calcite.rel.RelShuttleImpl}, so a fresh instance must be used per plan(). */ -@NoArgsConstructor(access = AccessLevel.PRIVATE) class DatetimeUdtNormalizeRule extends RelHomogeneousShuttle { - static final DatetimeUdtNormalizeRule INSTANCE = new DatetimeUdtNormalizeRule(); - @Override public RelNode visit(RelNode other) { RelNode visited = super.visit(other); diff --git a/api/src/test/java/org/opensearch/sql/api/spec/datetime/DatetimeExtensionTest.java b/api/src/test/java/org/opensearch/sql/api/spec/datetime/DatetimeExtensionTest.java index fc089150109..588d25d8d0f 100644 --- a/api/src/test/java/org/opensearch/sql/api/spec/datetime/DatetimeExtensionTest.java +++ b/api/src/test/java/org/opensearch/sql/api/spec/datetime/DatetimeExtensionTest.java @@ -172,6 +172,27 @@ public void testNonDatetimeFieldsNotWrapped() { """); } + @Test + public void testSequentialPlanCallsDoNotCorruptShuttleStack() { + // Regression test: DatetimeUdtNormalizeRule extends RelHomogeneousShuttle which inherits a + // stateful Deque stack field. Earlier implementations used a static INSTANCE shared + // across all plan() calls; under workloads with aggregations (especially count + distinct_count + // over datetime columns), the shared stack would desynchronize and visitChild's pop would throw + // NoSuchElementException on subsequent plan calls. Running several distinct plans through the + // same context confirms each invocation gets a fresh shuttle. + for (int i = 0; i < 5; i++) { + planner.plan( + "source = catalog.events" + + " | stats count() as field_count, distinct_count(created_at) as distinct_count"); + planner.plan( + "source = catalog.events" + + " | eval ts = TIMESTAMP(name)" + + " | stats count() as field_count, distinct_count(ts) as distinct_count"); + planner.plan( + "source = catalog.events | where created_at > \"2024-01-01\" | fields hire_date"); + } + } + @Test public void testOutputCastCanCompileAndExecute() throws Exception { RelNode plan = From 6e7e1f024e70f379b5db2b68defcaa2446a8277a Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Wed, 20 May 2026 13:58:05 -0700 Subject: [PATCH 2/4] test: add analytics-engine regression IT for singleton stack-corruption CalciteDatetimeUdtNormalizeRegressionIT exercises the failure pattern that triggered the cluster-side NoSuchElementException: stats + distinct_count over datetime columns, repeated 20 times to amplify any plan() carry-over. The IT is harness-aware: - Without `-Dtests.analytics.force_routing=true`: queries go through the V2 / Calcite engine path. The DatetimeUdtNormalizeRule path is not exercised, so the IT passes as a baseline correctness check. - With `-Dtests.analytics.force_routing=true -Dtests.analytics.parquet_indices=true`: every query routes through the analytics-engine path and hits the DatetimeUdtNormalizeRule shuttle that this PR fixes. The 20-iteration pattern surfaces any remaining singleton-stack carry-over. CI's :integTest task (in-process testCluster without analytics-engine) runs the IT through the V2 path, which is safe and fast. The analytics-engine verification path is via :integTestRemote against an externally-managed cluster built per `docs/dev/ppl-analytics-engine-routing.md`. Signed-off-by: Kai Huang --- ...lciteDatetimeUdtNormalizeRegressionIT.java | 117 ++++++++++++++++++ 1 file changed, 117 insertions(+) create mode 100644 integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java new file mode 100644 index 00000000000..430ab4e8d8e --- /dev/null +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java @@ -0,0 +1,117 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.sql.calcite.remote; + +import static org.junit.Assert.assertNotNull; +import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_DATE_FORMATS; + +import java.io.IOException; +import org.json.JSONObject; +import org.junit.jupiter.api.Test; +import org.opensearch.sql.ppl.PPLIntegTestCase; + +/** + * Regression IT for the {@code DatetimeUdtNormalizeRule} / {@code DatetimeOutputCastRule} + * singleton-stack-corruption bug. + * + *

Both rules extend Calcite's {@code RelHomogeneousShuttle}, which inherits a stateful {@code + * Deque} stack from {@code RelShuttleImpl}. Earlier code returned the same {@code + * INSTANCE} of each rule from {@code DatetimeExtension.postAnalysisRules()} on every {@code + * UnifiedQueryPlanner.plan()} call. If any traversal ever finished with an unbalanced stack, + * residual entries persisted to the next query and the next {@code visitChild()} popped a stale or + * empty stack — surfacing as {@code NoSuchElementException} at {@code RelShuttleImpl.visitChild} + * line 67. + * + *

The failure was reported as intermittent on dashboards issuing field-statistics queries of the + * shape {@code stats count() as field_count, distinct_count(field)} against parquet-backed indices. + * This IT runs the same query shape repeatedly against a parquet-backed (composite) index with + * multiple datetime columns to exercise the analytics-engine route end-to-end. + * + *

Run via: + * + * 

{@code
+ * ./gradlew :integ-test:integTestRemote \
+ *   -Dtests.rest.cluster=localhost:9200 -Dtests.cluster=localhost:9300 \
+ *   -Dtests.clustername=runTask \
+ *   -Dtests.analytics.force_routing=true \
+ *   -Dtests.analytics.parquet_indices=true \
+ *   --tests org.opensearch.sql.calcite.remote.CalciteDatetimeUdtNormalizeRegressionIT
+ * }
+ */ +public class CalciteDatetimeUdtNormalizeRegressionIT extends PPLIntegTestCase { + + /** Number of repetitions per test. Singleton bug surfaces order-dependent across plan() calls. */ + private static final int ITERATIONS = 20; + + @Override + public void init() throws Exception { + super.init(); + enableCalcite(); + + // DATE_FORMATS index has many date columns of different formats / precisions, which is what + // dashboard field-statistics panels iterate over. Loaded through the helper so that with + // -Dtests.analytics.parquet_indices=true it gets provisioned as a parquet-backed composite + // index — required for analytics-engine routing. + loadIndex(Index.DATE_FORMATS); + } + + @Test + public void testSequentialStatsDistinctCountOverDatetime() throws IOException { + // Bug repro: each iteration runs a stats + distinct_count over a datetime field. With the + // singleton rule, residual entries on the shuttle's internal stack from the previous plan() + // would cause NoSuchElementException on a subsequent traversal. With fresh instances per + // plan(), the stack is always empty at entry and the iterations all succeed. + String query = + String.format( + "source=%s | stats count() as field_count, distinct_count(epoch_millis) as" + + " distinct_count", + TEST_INDEX_DATE_FORMATS); + for (int i = 0; i < ITERATIONS; i++) { + JSONObject result = executeQuery(query); + // Sanity: response was returned without a cluster-side exception. + assertNotNull("iteration " + i + " produced no result", result); + } + } + + @Test + public void testInterleavedStatsAndDatetimeProjection() throws IOException { + // Bug repro variant: interleave stats+distinct_count with a plain datetime projection that + // exercises the DatetimeOutputCastRule. Different plan shapes per iteration push the rule + // through different visitChild paths and surface stack desync faster. + for (int i = 0; i < ITERATIONS; i++) { + executeQuery( + String.format( + "source=%s | stats count() as field_count, distinct_count(epoch_millis) as" + + " distinct_count", + TEST_INDEX_DATE_FORMATS)); + executeQuery( + String.format( + "source=%s | fields epoch_millis, epoch_second, date_optional_time", + TEST_INDEX_DATE_FORMATS)); + executeQuery( + String.format( + "source=%s | stats count() as field_count, distinct_count(epoch_second) as" + + " distinct_count", + TEST_INDEX_DATE_FORMATS)); + } + } + + @Test + public void testDistinctCountOverMultipleDatetimeFields() throws IOException { + // Bug repro variant: iterate distinct_count over different datetime fields in sequence — + // mirrors the dashboard field-statistics tab that probes every field in the index. + String[] datetimeFields = { + "epoch_millis", "epoch_second", "date_optional_time", "strict_date_optional_time" + }; + for (int i = 0; i < ITERATIONS; i++) { + String field = datetimeFields[i % datetimeFields.length]; + executeQuery( + String.format( + "source=%s | stats count() as field_count, distinct_count(%s) as distinct_count", + TEST_INDEX_DATE_FORMATS, field)); + } + } +} From 885052ebe7352cdda3d02b61ddafd35f6129bb94 Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Wed, 20 May 2026 14:28:22 -0700 Subject: [PATCH 3/4] test: switch regression IT to concurrent query pattern MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sequential iteration variant passed even with the singleton bug in place — local cluster doesn't carry over enough state between calls in one thread. The actual production trigger is parallel queries from a dashboard "field statistics" panel: multiple cluster threads call plan() simultaneously, all using the shared singleton's non-thread-safe ArrayDeque. Their push/pop operations interleave and corrupt the stack. Verified locally against analytics-engine path with parquet indices: - Unfixed cluster: 2-3 / 80 queries fail with NoSuchElementException (HTTP 500), matching the production stack trace exactly. - Fixed cluster: 0 / 80 failures. Uses CompletableFuture + 8-thread pool to fire 80 queries per test across: - testConcurrentStatsDistinctCountOverDatetime: same shape, varied datetime fields. - testConcurrentMixedDatetimePlans: three different plan shapes interleaved — mixed visitChild call counts amplify the race. Signed-off-by: Kai Huang --- ...lciteDatetimeUdtNormalizeRegressionIT.java | 168 +++++++++++------- 1 file changed, 102 insertions(+), 66 deletions(-) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java index 430ab4e8d8e..3daf6f81a6f 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java @@ -5,11 +5,15 @@ package org.opensearch.sql.calcite.remote; -import static org.junit.Assert.assertNotNull; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_DATE_FORMATS; -import java.io.IOException; -import org.json.JSONObject; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import org.opensearch.sql.ppl.PPLIntegTestCase; @@ -17,18 +21,18 @@ * Regression IT for the {@code DatetimeUdtNormalizeRule} / {@code DatetimeOutputCastRule} * singleton-stack-corruption bug. * - *

Both rules extend Calcite's {@code RelHomogeneousShuttle}, which inherits a stateful {@code - * Deque} stack from {@code RelShuttleImpl}. Earlier code returned the same {@code - * INSTANCE} of each rule from {@code DatetimeExtension.postAnalysisRules()} on every {@code - * UnifiedQueryPlanner.plan()} call. If any traversal ever finished with an unbalanced stack, - * residual entries persisted to the next query and the next {@code visitChild()} popped a stale or - * empty stack — surfacing as {@code NoSuchElementException} at {@code RelShuttleImpl.visitChild} - * line 67. + *

Both rules extend Calcite's {@code RelHomogeneousShuttle}, which inherits a stateful + * non-thread-safe {@code ArrayDeque} stack from {@code RelShuttleImpl}. Earlier code + * returned the same {@code INSTANCE} of each rule from {@code + * DatetimeExtension.postAnalysisRules()} on every {@code UnifiedQueryPlanner.plan()} call. Under + * parallel query load — exactly what a dashboard "field statistics" panel issues when it probes + * every field in an index concurrently — multiple cluster threads call {@code plan()} + * simultaneously and their push/pop on the shared stack interleave, leaving residual entries that + * surface on a subsequent traversal as {@code NoSuchElementException} at {@code + * RelShuttleImpl.visitChild} line 67 (the {@code stack.pop()} in the {@code finally} block). * - *

The failure was reported as intermittent on dashboards issuing field-statistics queries of the - * shape {@code stats count() as field_count, distinct_count(field)} against parquet-backed indices. - * This IT runs the same query shape repeatedly against a parquet-backed (composite) index with - * multiple datetime columns to exercise the analytics-engine route end-to-end. + *

This IT reproduces the production failure by firing many {@code distinct_count} queries over + * datetime fields concurrently against a parquet-backed (composite) index. * *

Run via: * @@ -43,75 +47,107 @@ */ public class CalciteDatetimeUdtNormalizeRegressionIT extends PPLIntegTestCase { - /** Number of repetitions per test. Singleton bug surfaces order-dependent across plan() calls. */ - private static final int ITERATIONS = 20; + /** Concurrency level — matches the rough parallelism of a dashboard field-stats panel. */ + private static final int PARALLELISM = 8; + + /** Total queries fired per test. */ + private static final int QUERIES = 80; @Override public void init() throws Exception { super.init(); enableCalcite(); - // DATE_FORMATS index has many date columns of different formats / precisions, which is what - // dashboard field-statistics panels iterate over. Loaded through the helper so that with - // -Dtests.analytics.parquet_indices=true it gets provisioned as a parquet-backed composite - // index — required for analytics-engine routing. + // DATE_FORMATS has many datetime columns of different formats/precisions. With + // -Dtests.analytics.parquet_indices=true the helper provisions it as a parquet-backed + // composite index — required for analytics-engine routing. loadIndex(Index.DATE_FORMATS); } @Test - public void testSequentialStatsDistinctCountOverDatetime() throws IOException { - // Bug repro: each iteration runs a stats + distinct_count over a datetime field. With the - // singleton rule, residual entries on the shuttle's internal stack from the previous plan() - // would cause NoSuchElementException on a subsequent traversal. With fresh instances per - // plan(), the stack is always empty at entry and the iterations all succeed. - String query = - String.format( - "source=%s | stats count() as field_count, distinct_count(epoch_millis) as" - + " distinct_count", - TEST_INDEX_DATE_FORMATS); - for (int i = 0; i < ITERATIONS; i++) { - JSONObject result = executeQuery(query); - // Sanity: response was returned without a cluster-side exception. - assertNotNull("iteration " + i + " produced no result", result); + public void testConcurrentStatsDistinctCountOverDatetime() throws Exception { + String[] fields = { + "epoch_millis", "epoch_second", "date_optional_time", "strict_date_optional_time" + }; + List queries = new ArrayList<>(QUERIES); + for (int i = 0; i < QUERIES; i++) { + String field = fields[i % fields.length]; + queries.add( + String.format( + "source=%s | stats count() as field_count, distinct_count(%s) as distinct_count", + TEST_INDEX_DATE_FORMATS, field)); } + executeConcurrent(queries); } @Test - public void testInterleavedStatsAndDatetimeProjection() throws IOException { - // Bug repro variant: interleave stats+distinct_count with a plain datetime projection that - // exercises the DatetimeOutputCastRule. Different plan shapes per iteration push the rule - // through different visitChild paths and surface stack desync faster. - for (int i = 0; i < ITERATIONS; i++) { - executeQuery( - String.format( - "source=%s | stats count() as field_count, distinct_count(epoch_millis) as" - + " distinct_count", - TEST_INDEX_DATE_FORMATS)); - executeQuery( - String.format( - "source=%s | fields epoch_millis, epoch_second, date_optional_time", - TEST_INDEX_DATE_FORMATS)); - executeQuery( - String.format( - "source=%s | stats count() as field_count, distinct_count(epoch_second) as" - + " distinct_count", - TEST_INDEX_DATE_FORMATS)); + public void testConcurrentMixedDatetimePlans() throws Exception { + // Mix three plan shapes: stats+distinct_count, plain field projection (datetime cast), and + // stats by a different field. Different plan shapes push the singleton shuttle through + // different visitChild call counts — making cross-query stack pollution more likely. + List shapes = + List.of( + "source=%s | stats count() as field_count, distinct_count(epoch_millis) as" + + " distinct_count", + "source=%s | fields epoch_millis, epoch_second, date_optional_time", + "source=%s | stats count() as field_count, distinct_count(epoch_second) as" + + " distinct_count by date_optional_time"); + List queries = new ArrayList<>(QUERIES); + for (int i = 0; i < QUERIES; i++) { + queries.add(String.format(shapes.get(i % shapes.size()), TEST_INDEX_DATE_FORMATS)); } + executeConcurrent(queries); } - @Test - public void testDistinctCountOverMultipleDatetimeFields() throws IOException { - // Bug repro variant: iterate distinct_count over different datetime fields in sequence — - // mirrors the dashboard field-statistics tab that probes every field in the index. - String[] datetimeFields = { - "epoch_millis", "epoch_second", "date_optional_time", "strict_date_optional_time" - }; - for (int i = 0; i < ITERATIONS; i++) { - String field = datetimeFields[i % datetimeFields.length]; - executeQuery( - String.format( - "source=%s | stats count() as field_count, distinct_count(%s) as distinct_count", - TEST_INDEX_DATE_FORMATS, field)); + /** + * Fire all queries through a fixed-size thread pool. Asserts every query completes without + * exception. With the singleton bug present this triggers {@code NoSuchElementException} on at + * least one task once the stack interleaving corrupts state. + */ + private void executeConcurrent(List queries) throws Exception { + var executor = Executors.newFixedThreadPool(PARALLELISM); + try { + List> futures = new ArrayList<>(queries.size()); + AtomicInteger failures = new AtomicInteger(); + List errors = new ArrayList<>(); + for (String query : queries) { + futures.add( + CompletableFuture.runAsync( + () -> { + try { + executeQuery(query); + } catch (Exception e) { + failures.incrementAndGet(); + synchronized (errors) { + errors.add(e); + } + } + }, + executor)); + } + for (CompletableFuture f : futures) { + try { + f.get(60, TimeUnit.SECONDS); + } catch (ExecutionException e) { + failures.incrementAndGet(); + synchronized (errors) { + errors.add(e.getCause()); + } + } + } + if (failures.get() > 0) { + StringBuilder msg = new StringBuilder(); + msg.append(failures.get()).append("/").append(queries.size()).append(" queries failed:"); + synchronized (errors) { + for (int i = 0; i < Math.min(3, errors.size()); i++) { + msg.append("\n - ").append(errors.get(i)); + } + } + throw new AssertionError(msg.toString()); + } + } finally { + executor.shutdown(); + executor.awaitTermination(30, TimeUnit.SECONDS); } } } From 0c1b7017aec3996ea8eeda3ba46ab78bf067bdde Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Thu, 21 May 2026 09:52:54 -0700 Subject: [PATCH 4/4] test: rename to CalcitePlannerConcurrencyIT (review nit) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit @dai-chen flagged that the IT name was over-scoped to a single rule and the file would read better as a general bucket for planner-level concurrency / state-isolation regressions. The actual surface under test is UnifiedQueryPlanner's post-analysis pipeline — any RelShuttle extension that doesn't isolate per-call state is unsafe under concurrent load, not just the datetime rules. Renames the file and class, updates the JavaDoc to describe the planner- level invariant rather than the specific Datetime* rules, and notes the current cases as the regression that motivated the suite. Test method bodies and assertions are unchanged. Signed-off-by: Kai Huang --- ....java => CalcitePlannerConcurrencyIT.java} | 40 ++++++++++--------- 1 file changed, 21 insertions(+), 19 deletions(-) rename integ-test/src/test/java/org/opensearch/sql/calcite/remote/{CalciteDatetimeUdtNormalizeRegressionIT.java => CalcitePlannerConcurrencyIT.java} (72%) diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePlannerConcurrencyIT.java similarity index 72% rename from integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java rename to integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePlannerConcurrencyIT.java index 3daf6f81a6f..375a0f10716 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteDatetimeUdtNormalizeRegressionIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePlannerConcurrencyIT.java @@ -18,21 +18,23 @@ import org.opensearch.sql.ppl.PPLIntegTestCase; /** - * Regression IT for the {@code DatetimeUdtNormalizeRule} / {@code DatetimeOutputCastRule} - * singleton-stack-corruption bug. + * Integration tests for {@code UnifiedQueryPlanner} state isolation under concurrent load. * - *

Both rules extend Calcite's {@code RelHomogeneousShuttle}, which inherits a stateful - * non-thread-safe {@code ArrayDeque} stack from {@code RelShuttleImpl}. Earlier code - * returned the same {@code INSTANCE} of each rule from {@code - * DatetimeExtension.postAnalysisRules()} on every {@code UnifiedQueryPlanner.plan()} call. Under - * parallel query load — exactly what a dashboard "field statistics" panel issues when it probes - * every field in an index concurrently — multiple cluster threads call {@code plan()} - * simultaneously and their push/pop on the shared stack interleave, leaving residual entries that - * surface on a subsequent traversal as {@code NoSuchElementException} at {@code - * RelShuttleImpl.visitChild} line 67 (the {@code stack.pop()} in the {@code finally} block). + *

The planner's post-analysis pipeline (extensions registered via {@code + * LanguageSpec.postAnalysisRules}) uses Calcite {@code RelShuttle} subclasses. {@code + * RelShuttleImpl} inherits a non-thread-safe {@code ArrayDeque} stack used by {@code + * visitChild}'s push/pop. Any extension that returns the same shuttle instance across {@code + * plan()} calls is unsafe under concurrent load: cluster threads call {@code plan()} simultaneously + * and their push/pop on the shared stack interleave, leaving residual entries that surface on a + * subsequent traversal as {@code NoSuchElementException} at {@code RelShuttleImpl.visitChild} line + * 67 (the {@code stack.pop()} in the {@code finally} block). * - *

This IT reproduces the production failure by firing many {@code distinct_count} queries over - * datetime fields concurrently against a parquet-backed (composite) index. + *

The test methods here fire many queries through a thread pool to exercise concurrent {@code + * plan()} invocations. New planner-level concurrency / state-isolation regressions belong in this + * class. The current cases cover {@code DatetimeExtension}'s {@code RelHomogeneousShuttle} + * subclasses ({@code DatetimeUdtNormalizeRule}, {@code DatetimeOutputCastRule}) which were + * previously returned as static {@code INSTANCE}s and caused the production failure that motivated + * this suite. * *

Run via: * @@ -42,10 +44,10 @@ * -Dtests.clustername=runTask \ * -Dtests.analytics.force_routing=true \ * -Dtests.analytics.parquet_indices=true \ - * --tests org.opensearch.sql.calcite.remote.CalciteDatetimeUdtNormalizeRegressionIT + * --tests org.opensearch.sql.calcite.remote.CalcitePlannerConcurrencyIT * } */ -public class CalciteDatetimeUdtNormalizeRegressionIT extends PPLIntegTestCase { +public class CalcitePlannerConcurrencyIT extends PPLIntegTestCase { /** Concurrency level — matches the rough parallelism of a dashboard field-stats panel. */ private static final int PARALLELISM = 8; @@ -83,8 +85,8 @@ public void testConcurrentStatsDistinctCountOverDatetime() throws Exception { @Test public void testConcurrentMixedDatetimePlans() throws Exception { // Mix three plan shapes: stats+distinct_count, plain field projection (datetime cast), and - // stats by a different field. Different plan shapes push the singleton shuttle through - // different visitChild call counts — making cross-query stack pollution more likely. + // stats by a different field. Different plan shapes push the planner's post-analysis shuttles + // through different visitChild call counts — amplifying any cross-query stack pollution. List shapes = List.of( "source=%s | stats count() as field_count, distinct_count(epoch_millis) as" @@ -101,8 +103,8 @@ public void testConcurrentMixedDatetimePlans() throws Exception { /** * Fire all queries through a fixed-size thread pool. Asserts every query completes without - * exception. With the singleton bug present this triggers {@code NoSuchElementException} on at - * least one task once the stack interleaving corrupts state. + * exception. With a shuttle-state-leak bug present this triggers {@code NoSuchElementException} + * on at least one task once the stack interleaving corrupts state. */ private void executeConcurrent(List queries) throws Exception { var executor = Executors.newFixedThreadPool(PARALLELISM);