Use the precision of the original timestamp column in HOP/TUMBLE

Signed-off-by: Mihai Budiu <mbudiu@feldera.com>

Author: mihaibudiu

core/src/main/java/org/apache/calcite/sql/SqlWindowTableFunction.java

@@ -40,6 +40,8 @@
 
 import static org.apache.calcite.util.Static.RESOURCE;
 
+import static java.util.Objects.requireNonNull;
+
 /**
  * Base class for a table-valued function that computes windows. Examples
  * include {@code TUMBLE}, {@code HOP} and {@code SESSION}.
@@ -107,14 +109,37 @@ public SqlWindowTableFunction(String name, SqlOperandMetadata operandMetadata) {
   private static RelDataType inferRowType(SqlOperatorBinding opBinding) {
     final RelDataType inputRowType = opBinding.getOperandType(0);
     final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+    int precision = getTimestampPrecision(opBinding);
     return typeFactory.builder()
         .kind(inputRowType.getStructKind())
         .addAll(inputRowType.getFieldList())
-        .add("window_start", SqlTypeName.TIMESTAMP)
-        .add("window_end", SqlTypeName.TIMESTAMP)
+        .add("window_start", SqlTypeName.TIMESTAMP, precision)
+        .add("window_end", SqlTypeName.TIMESTAMP, precision)
         .build();
   }
 
+  /** Extract the precision for the start_window, end_window columns from the column supplied as
+   * DESCRIPTOR for the window function. */
+  private static int getTimestampPrecision(SqlOperatorBinding opBinding) {
+    RelDataType inputRowType = opBinding.getOperandType(0);
+    SqlCallBinding callBinding = (SqlCallBinding) opBinding;
+    // Locate the "descriptor" argument
+    for (SqlNode operand : callBinding.operands()) {
+      if (operand instanceof SqlCall) {
+        SqlCall opCall = (SqlCall) operand;
+        if (opCall.getOperator().getKind() == SqlKind.DESCRIPTOR) {
+          SqlNode descriptor = opCall.operand(0);
+          SqlIdentifier id = (SqlIdentifier) descriptor;
+          RelDataTypeField field =
+              inputRowType.getField(id.getSimple(), false, false);
+          return requireNonNull(field, "field").getType().getPrecision();
+        }
+      }
+    }
+    // Should be unreachable, since validation succeeded
+    throw new RuntimeException("Could not locate DESCRIPTOR column");
+  }
+
   /** Partial implementation of operand type checker. */
   protected abstract static class AbstractOperandMetadata
       implements SqlOperandMetadata {

core/src/test/java/org/apache/calcite/test/SqlToRelConverterTest.java

@@ -2502,6 +2502,13 @@ void checkCorrelatedMapSubQuery(boolean expand) {
     sql(sql).ok();
   }
 
+  @Test void testTableFunctionTumbleConvert() {
+    final String sql = "with t as (select CAST(rowtime AS TIMESTAMP(2)) as rowtime FROM Shipments) "
+        + "select *\n"
+        + "from table(tumble(table t, descriptor(rowtime), INTERVAL '1.5' SECOND))";
+    sql(sql).ok();
+  }
+
   @Test void testTableFunctionTumbleWithParamNames() {
     final String sql = "select *\n"
         + "from table(\n"

core/src/test/resources/org/apache/calcite/test/SqlToRelConverterTest.xml

@@ -8588,6 +8588,20 @@ LogicalProject(ORDERID=[$0], ROWTIME=[$1], window_start=[$2], window_end=[$3])
   LogicalTableFunctionScan(invocation=[TUMBLE(DESCRIPTOR($1), 60000:INTERVAL MINUTE)], rowType=[RecordType(INTEGER ORDERID, TIMESTAMP(0) ROWTIME, TIMESTAMP(0) window_start, TIMESTAMP(0) window_end)])
     LogicalProject(ORDERID=[$0], ROWTIME=[$1])
       LogicalTableScan(table=[[CATALOG, SALES, SHIPMENTS]])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testTableFunctionTumbleConvert">
+    <Resource name="sql">
+      <![CDATA[with t as (select CAST(rowtime AS TIMESTAMP(2)) as rowtime FROM Shipments) select *
+from table(tumble(table t, descriptor(rowtime), INTERVAL '1.5' SECOND))]]>
+    </Resource>
+    <Resource name="plan">
+      <![CDATA[
+LogicalProject(ROWTIME=[$0], window_start=[$1], window_end=[$2])
+  LogicalTableFunctionScan(invocation=[TUMBLE(DESCRIPTOR($0), 1500:INTERVAL SECOND)], rowType=[RecordType(TIMESTAMP(2) ROWTIME, TIMESTAMP(2) window_start, TIMESTAMP(2) window_end)])
+    LogicalProject(ROWTIME=[CAST($1):TIMESTAMP(2) NOT NULL])
+      LogicalTableScan(table=[[CATALOG, SALES, SHIPMENTS]])
 ]]>
     </Resource>
   </TestCase>

core/src/test/resources/sql/stream.iq

@@ -17,6 +17,24 @@
 #
 !use orinoco
 !set outputformat mysql
+
+# Test case for [CALCITE-7410] TIMESTAMP type for TUMBLE and HOP is hardwired to TIMESTAMP(3)
+# Since we cast the input column to TIMESTAMP(3), we expect window_start to have the same type.
+WITH S AS (SELECT *, CAST(ROWTIME AS TIMESTAMP(3)) + INTERVAL '0.5' SECONDS AS TS FROM ORDERS)
+SELECT * FROM TABLE(TUMBLE((SELECT * FROM S), DESCRIPTOR(TS), INTERVAL '20:10.525' MINUTE TO SECOND));
++---------------------+----+---------+-------+-------------------------+-------------------------+-------------------------+
+| ROWTIME             | ID | PRODUCT | UNITS | TS                      | window_start            | window_end              |
++---------------------+----+---------+-------+-------------------------+-------------------------+-------------------------+
+| 2015-02-15 10:15:00 |  1 | paint   |    10 | 2015-02-15 10:15:00.500 | 2015-02-15 10:10:31.125 | 2015-02-15 10:30:41.650 |
+| 2015-02-15 10:24:15 |  2 | paper   |     5 | 2015-02-15 10:24:15.500 | 2015-02-15 10:10:31.125 | 2015-02-15 10:30:41.650 |
+| 2015-02-15 10:24:45 |  3 | brush   |    12 | 2015-02-15 10:24:45.500 | 2015-02-15 10:10:31.125 | 2015-02-15 10:30:41.650 |
+| 2015-02-15 10:58:00 |  4 | paint   |     3 | 2015-02-15 10:58:00.500 | 2015-02-15 10:50:52.175 | 2015-02-15 11:11:02.700 |
+| 2015-02-15 11:10:00 |  5 | paint   |     3 | 2015-02-15 11:10:00.500 | 2015-02-15 10:50:52.175 | 2015-02-15 11:11:02.700 |
++---------------------+----+---------+-------+-------------------------+-------------------------+-------------------------+
+(5 rows)
+
+!ok
+
 SELECT * FROM TABLE(
   TUMBLE(
     DATA => TABLE ORDERS,