Replace Calcite PIVOT with post-processing pivot, add series parameter

- Remove Calcite PIVOT from timewrap: no more MAX_PERIODS limit or crash risk
- Add post-processing pivot in execution engine: dynamically builds columns
  from actual data using HashMap grouping
- Add series parameter: series=relative (default), series=short (s0, s1)
- Add series=exact grammar support (falls back to short at runtime)
- Add time_format grammar support for series=exact
- Benchmark: 1000 period columns in ~50ms (Calcite PIVOT crashed at 1000)
- 32 IT tests, all using verifySchema + verifyDataRows

Signed-off-by: Jialiang Li <jialiang.li@hey.com>
Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

core/src/main/java/org/opensearch/sql/ast/tree/Timewrap.java

@@ -24,6 +24,8 @@ public class Timewrap extends UnresolvedPlan {
   private final SpanUnit unit;
   private final int value;
   private final String align; // "end" or "now"
+  private final String series; // "relative", "short", or "exact"
+  private final String timeFormat; // format string for series=exact, nullable
   private final Literal spanLiteral; // original span literal for display
 
   private UnresolvedPlan child;

core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java

@@ -53,6 +53,12 @@ public class CalcitePlanContext {
    */
   public static final ThreadLocal<String> timewrapUnitName = new ThreadLocal<>();
 
+  /** Timewrap series mode: "relative", "short", or "exact". */
+  public static final ThreadLocal<String> timewrapSeries = new ThreadLocal<>();
+
+  /** Timewrap time_format string for series=exact (e.g., "%Y-%m-%d"). */
+  public static final ThreadLocal<String> timewrapTimeFormat = new ThreadLocal<>();
+
   /** Thread-local switch that tells whether the current query prefers legacy behavior. */
   private static final ThreadLocal<Boolean> legacyPreferredFlag =
       ThreadLocal.withInitial(() -> true);

core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java

@@ -3835,6 +3835,8 @@ public RelNode visitTimewrap(Timewrap node, CalcitePlanContext context) {
     CalcitePlanContext.stripNullColumns.set(true);
     CalcitePlanContext.timewrapUnitName.set(
         TimewrapUtils.unitBaseName(node.getUnit(), node.getValue()) + "|_before");
+    CalcitePlanContext.timewrapSeries.set(node.getSeries());
+    CalcitePlanContext.timewrapTimeFormat.set(node.getTimeFormat());
 
     RelBuilder b = context.relBuilder;
     RexBuilder rx = context.rexBuilder;
@@ -3950,44 +3952,10 @@ public RelNode visitTimewrap(Timewrap node, CalcitePlanContext context) {
     projections.add(b.alias(periodNum, "__period__"));
     b.project(projections);
 
-    // Step 4: PIVOT on period, grouped by [offset, __base_offset__]
-    // __base_offset__ is constant across all rows so it doesn't affect grouping,
-    // but survives the PIVOT so the execution engine can use it for absolute column naming
-    b.pivot(
-        b.groupKey(b.field(tsFieldName), b.field("__base_offset__")),
-        valueFieldNames.stream().map(f -> (RelBuilder.AggCall) b.max(b.field(f)).as("")).toList(),
-        ImmutableList.of(b.field("__period__")),
-        IntStream.rangeClosed(1, TimewrapUtils.MAX_PERIODS)
-            .map(i -> TimewrapUtils.MAX_PERIODS + 1 - i) // reverse: oldest period first
-            .mapToObj(
-                i ->
-                    Map.entry(
-                        // Use placeholder relative names; execution engine renames to absolute
-                        String.valueOf(i), ImmutableList.of((RexNode) b.literal((long) i))))
-            .collect(Collectors.toList()));
-
-    // Step 5: Rename columns — add agg name prefix, clean up pivot artifacts
-    // Use relative period numbers as temporary names; the execution engine will compute
-    // absolute offsets using __base_offset__ and rename accordingly
-    List<String> pivotColNames = b.peek().getRowType().getFieldNames();
-    List<String> cleanNames = new ArrayList<>();
-    cleanNames.add(tsFieldName);
-    cleanNames.add("__base_offset__");
-    for (int i = 2; i < pivotColNames.size(); i++) {
-      String name = pivotColNames.get(i);
-      if (name.endsWith("_")) {
-        name = name.substring(0, name.length() - 1);
-      }
-      // Prefix with agg name and _before suffix
-      if (valueFieldNames.size() == 1) {
-        name = valueFieldNames.get(0) + "_" + name + "_before";
-      }
-      cleanNames.add(name);
-    }
-    b.rename(cleanNames);
-
-    // Step 6: Sort by offset
-    b.sort(b.field(0));
+    // Step 4: Sort by offset, then period (execution engine will pivot)
+    // No Calcite PIVOT -- the execution engine pivots dynamically after reading all rows.
+    // Output schema: [display_timestamp, value_columns..., __base_offset__, __period__]
+    b.sort(b.field(tsFieldName), b.field("__period__"));
 
     return b.peek();
   }

core/src/main/java/org/opensearch/sql/calcite/utils/TimewrapUtils.java

@@ -276,18 +276,31 @@ public static long calendarUnitNumberFromEpoch(long epochSec, SpanUnit unit, int
   }
 
   /**
-   * Walk the AST from a Timewrap node to find a WHERE clause with an upper bound on the timestamp
-   * field. Returns the upper bound as epoch seconds, or null if not found.
+   * Walk the AST from a Timewrap node to the deepest Filter node and extract the timestamp upper
+   * bound. The frontend time picker always appends the timestamp filter as the first pipe (closest
+   * to source), making it the deepest Filter in the AST chain:
+   *
+   * <pre>
+   *   Timewrap -> Chart -> [user filters] -> Filter(@timestamp >= X AND @timestamp <= Y) -> Source
+   * </pre>
+   *
+   * We walk all Filter nodes and return the last (deepest) timestamp upper bound found. This
+   * ensures user filters like `where age > 30` between timechart and the time picker filter don't
+   * interfere.
    */
   public static Long extractTimestampUpperBound(Timewrap node) {
     Node current = node;
+    Long lastBound = null;
     while (current != null && !current.getChild().isEmpty()) {
       current = current.getChild().get(0);
       if (current instanceof Filter filter) {
-        return findUpperBound(filter.getCondition());
+        Long bound = findUpperBound(filter.getCondition());
+        if (bound != null) {
+          lastBound = bound;
+        }
       }
     }
-    return null;
+    return lastBound;
   }
 
   private static Long findUpperBound(UnresolvedExpression expr) {

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteTimewrapCommandIT.java

@@ -143,23 +143,32 @@ public void testTimewrapIncompletePeriodNullFill() throws IOException {
         schema("sum(requests)_2days_before", "bigint"),
         schema("sum(requests)_1day_before", "bigint"),
         schema("sum(requests)_latest_day", "bigint"));
-    verifyDataRowsSome(result, rows("2024-07-04 12:00:00", null, 310, 330, 285));
-    verifyDataRowsSome(result, rows("2024-07-04 18:00:00", null, 190, 215, 165));
+    verifyDataRowsInOrder(
+        result,
+        rows("2024-07-04 00:00:00", 180, 205, 165, 80),
+        rows("2024-07-04 06:00:00", 240, 260, 225, 100),
+        rows("2024-07-04 12:00:00", null, 310, 330, 285),
+        rows("2024-07-04 18:00:00", null, 190, 215, 165));
   }
 
   // --- Different timescales ---
 
   @Test
   public void testTimewrapWeekSpanSinglePeriod() throws IOException {
+    // 3 days of daily data in 1 week -> single period, 3 rows
     JSONObject result =
         executeQuery(
             "source=timewrap_test"
-                + WHERE_ALL
-                + " | timechart span=6h sum(requests) | timewrap 1week");
+                + WHERE_JUL1_TO_JUL3
+                + " | timechart span=1day sum(requests) | timewrap 1week");
 
     verifySchema(
         result, schema("@timestamp", "timestamp"), schema("sum(requests)_latest_week", "bigint"));
-    verifyDataRowsSome(result, rows("2024-07-04 00:00:00", 80), rows("2024-07-04 06:00:00", 100));
+    verifyDataRowsInOrder(
+        result,
+        rows("2024-07-01 00:00:00", 920),
+        rows("2024-07-02 00:00:00", 1010),
+        rows("2024-07-03 00:00:00", 840));
   }
 
   @Test
@@ -180,7 +189,10 @@ public void testTimewrapTwelveHourSpan() throws IOException {
         schema("sum(requests)_24hours_before", "bigint"),
         schema("sum(requests)_12hours_before", "bigint"),
         schema("sum(requests)_latest_hour", "bigint"));
-    verifyDataRowsSome(result, rows("2024-07-04 00:00:00", 180, 310, 205, 330, 165, 285, 80));
+    verifyDataRowsInOrder(
+        result,
+        rows("2024-07-04 00:00:00", 180, 310, 205, 330, 165, 285, 80),
+        rows("2024-07-04 06:00:00", 240, 190, 260, 215, 225, 165, 100));
   }
 
   @Test
@@ -531,15 +543,17 @@ public void testTimewrapDaySpan() throws IOException {
 
   @Test
   public void testTimewrapWeekSpan() throws IOException {
+    // 2 days of daily data in 1 week -> single period, 2 rows
     JSONObject result =
         executeQuery(
             "source=timewrap_test"
-                + WHERE_ALL
-                + " | timechart span=6h sum(requests) | timewrap 1week");
+                + WHERE_JUL1_TO_JUL2
+                + " | timechart span=1day sum(requests) | timewrap 1week");
 
     verifySchema(
         result, schema("@timestamp", "timestamp"), schema("sum(requests)_latest_week", "bigint"));
-    verifyDataRowsSome(result, rows("2024-07-04 00:00:00", 80), rows("2024-07-04 06:00:00", 100));
+    verifyDataRowsInOrder(
+        result, rows("2024-07-01 00:00:00", 920), rows("2024-07-02 00:00:00", 1010));
   }
 
   @Test
@@ -552,7 +566,12 @@ public void testTimewrapMonthSpan() throws IOException {
 
     verifySchema(
         result, schema("@timestamp", "timestamp"), schema("sum(requests)_latest_month", "bigint"));
-    verifyDataRowsSome(result, rows("2024-07-01 00:00:00", 920), rows("2024-07-04 00:00:00", 180));
+    verifyDataRowsInOrder(
+        result,
+        rows("2024-07-01 00:00:00", 920),
+        rows("2024-07-02 00:00:00", 1010),
+        rows("2024-07-03 00:00:00", 840),
+        rows("2024-07-04 00:00:00", 180));
   }
 
   @Test
@@ -570,24 +589,174 @@ public void testTimewrapQuarterSpan() throws IOException {
         schema("@timestamp", "timestamp"),
         schema("sum(requests)_1quarter_before", "bigint"),
         schema("sum(requests)_latest_quarter", "bigint"));
-    // Day 15 of each quarter aligns — both values on the same row
-    verifyDataRowsSome(result, rows("2024-04-15 00:00:00", 300, 350));
+    // Day 15 of each quarter aligns -- both values on the same row
+    verifyDataRows(result, rows("2024-04-15 00:00:00", 300, 350));
   }
 
   @Test
   public void testTimewrapYearSpan() throws IOException {
-    // Jan 2024 to Jan 2025 → 2 year periods (2024 and 2025)
-    // Jan 15 offset exists in both years: 2024=300, 2025=400
+    // Jan 15 2024 (300) and Jan 15 2025 (400) -- 2 data points in 2 different years
+    // timechart span=1year: 2 yearly buckets
+    // timewrap 1year: 2 periods, 1 offset row
     JSONObject result =
         executeQuery(
-            "source=timewrap_test | where @timestamp >= '2024-01-15 00:00:00' and @timestamp <="
-                + " '2025-01-15 12:00:00' | timechart span=1day sum(requests) | timewrap 1year");
+            "source=timewrap_test | where @timestamp >= '2024-01-15 12:00:00' and @timestamp <="
+                + " '2025-01-15 12:00:00' | timechart span=1year sum(requests) | timewrap 1year");
 
     verifySchema(
         result,
         schema("@timestamp", "timestamp"),
         schema("sum(requests)_1year_before", "bigint"),
         schema("sum(requests)_latest_year", "bigint"));
-    verifyDataRowsSome(result, rows("2025-01-15 00:00:00", 300, 400));
+    // 2024 yearly sum = all 2024 data in WHERE range; 2025 = Jan 15 only (400)
+    verifyDataRows(result, rows("2025-01-01 00:00:00", 4050, 400));
+  }
+
+  // --- series parameter ---
+
+  @Test
+  public void testTimewrapSeriesRelativeIsDefault() throws IOException {
+    // series=relative is the default — same as no series parameter
+    JSONObject resultDefault =
+        executeQuery(
+            "source=timewrap_test"
+                + WHERE_JUL1_TO_JUL3
+                + " | timechart span=6h sum(requests) | timewrap 1day");
+    JSONObject resultRelative =
+        executeQuery(
+            "source=timewrap_test"
+                + WHERE_JUL1_TO_JUL3
+                + " | timechart span=6h sum(requests) | timewrap 1day series=relative");
+
+    verifySchema(
+        resultRelative,
+        schema("@timestamp", "timestamp"),
+        schema("sum(requests)_2days_before", "bigint"),
+        schema("sum(requests)_1day_before", "bigint"),
+        schema("sum(requests)_latest_day", "bigint"));
+    verifyDataRowsInOrder(
+        resultRelative,
+        rows("2024-07-03 00:00:00", 180, 205, 165),
+        rows("2024-07-03 06:00:00", 240, 260, 225),
+        rows("2024-07-03 12:00:00", 310, 330, 285),
+        rows("2024-07-03 18:00:00", 190, 215, 165));
+    verifySchema(
+        resultDefault,
+        schema("@timestamp", "timestamp"),
+        schema("sum(requests)_2days_before", "bigint"),
+        schema("sum(requests)_1day_before", "bigint"),
+        schema("sum(requests)_latest_day", "bigint"));
+    verifyDataRowsInOrder(
+        resultDefault,
+        rows("2024-07-03 00:00:00", 180, 205, 165),
+        rows("2024-07-03 06:00:00", 240, 260, 225),
+        rows("2024-07-03 12:00:00", 310, 330, 285),
+        rows("2024-07-03 18:00:00", 190, 215, 165));
   }
+
+  @Test
+  public void testTimewrapSeriesShort() throws IOException {
+    // series=short: columns named <agg>_s<absolutePeriod>
+    // With align=end and WHERE upper bound = Jul 3 18:00, baseOffset=0
+    // Periods: oldest=2, middle=1, newest=0
+    JSONObject result =
+        executeQuery(
+            "source=timewrap_test"
+                + WHERE_JUL1_TO_JUL3
+                + " | timechart span=6h sum(requests) | timewrap 1day series=short");
+
+    verifySchema(
+        result,
+        schema("@timestamp", "timestamp"),
+        schema("sum(requests)_s2", "bigint"),
+        schema("sum(requests)_s1", "bigint"),
+        schema("sum(requests)_s0", "bigint"));
+    verifyDataRowsInOrder(
+        result,
+        rows("2024-07-03 00:00:00", 180, 205, 165),
+        rows("2024-07-03 06:00:00", 240, 260, 225),
+        rows("2024-07-03 12:00:00", 310, 330, 285),
+        rows("2024-07-03 18:00:00", 190, 215, 165));
+  }
+
+  @Test
+  public void testTimewrapSeriesShortWithCount() throws IOException {
+    // series=short with count aggregation
+    JSONObject result =
+        executeQuery(
+            "source=timewrap_test"
+                + WHERE_JUL2_TO_JUL3
+                + " | timechart span=6h count() | timewrap 1day series=short");
+
+    verifySchema(
+        result,
+        schema("@timestamp", "timestamp"),
+        schema("count()_s1", "bigint"),
+        schema("count()_s0", "bigint"));
+    verifyDataRowsInOrder(
+        result,
+        rows("2024-07-03 00:00:00", 2, 2),
+        rows("2024-07-03 06:00:00", 2, 2),
+        rows("2024-07-03 12:00:00", 2, 2),
+        rows("2024-07-03 18:00:00", 2, 2));
+  }
+
+  @Test
+  public void testTimewrapSeriesShortWeekSpan() throws IOException {
+    // series=short with week span, single period = s0, daily buckets
+    JSONObject result =
+        executeQuery(
+            "source=timewrap_test"
+                + WHERE_JUL1_TO_JUL2
+                + " | timechart span=1day sum(requests) | timewrap 1week series=short");
+
+    verifySchema(result, schema("@timestamp", "timestamp"), schema("sum(requests)_s0", "bigint"));
+    verifyDataRowsInOrder(
+        result, rows("2024-07-01 00:00:00", 920), rows("2024-07-02 00:00:00", 1010));
+  }
+
+  @Test
+  public void testTimewrapSeriesShortWithAvg() throws IOException {
+    JSONObject result =
+        executeQuery(
+            "source=timewrap_test"
+                + WHERE_JUL1_TO_JUL2
+                + " | timechart span=6h avg(requests) | timewrap 1day series=short");
+
+    verifySchema(
+        result,
+        schema("@timestamp", "timestamp"),
+        schema("avg(requests)_s1", "double"),
+        schema("avg(requests)_s0", "double"));
+    verifyDataRowsInOrder(
+        result,
+        rows("2024-07-02 00:00:00", 90.0, 102.5),
+        rows("2024-07-02 06:00:00", 120.0, 130.0),
+        rows("2024-07-02 12:00:00", 155.0, 165.0),
+        rows("2024-07-02 18:00:00", 95.0, 107.5));
+  }
+
+  @Test
+  public void testTimewrapSeriesShortWithErrors() throws IOException {
+    JSONObject result =
+        executeQuery(
+            "source=timewrap_test"
+                + WHERE_JUL2_TO_JUL3
+                + " | timechart span=6h sum(errors) | timewrap 1day series=short");
+
+    verifySchema(
+        result,
+        schema("@timestamp", "timestamp"),
+        schema("sum(errors)_s1", "bigint"),
+        schema("sum(errors)_s0", "bigint"));
+    verifyDataRowsInOrder(
+        result,
+        rows("2024-07-03 00:00:00", 4, 1),
+        rows("2024-07-03 06:00:00", 6, 3),
+        rows("2024-07-03 12:00:00", 9, 6),
+        rows("2024-07-03 18:00:00", 3, 1));
+  }
+
+  // BY clause tests are pending -- blocked by timechart BY output format gap.
+  // See docs/dev/ppl-timewrap-command.md for design options.
 }