fix: instantiate DatetimeUdt normalize/output-cast rules per plan() call

DatetimeUdtNormalizeRule and DatetimeOutputCastRule extend
RelHomogeneousShuttle, which inherits a stateful Deque<RelNode> 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 <ahkcs@amazon.com>

Author: ahkcs

api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeExtension.java

@@ -22,7 +22,8 @@ public class DatetimeExtension implements LanguageExtension {
 
   @Override
   public List<RelShuttle> 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. */

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.
+ *
+ * <p>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<RelDataTypeField> fields = other.getRowType().getFieldList();

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.
+ *
+ * <p>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);

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<RelNode> 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 =