feat(dqe): timestamp sort, DATE literal, and type support

- SortOperator: add TimestampType and BooleanType comparison
- ExpressionCompiler: handle GenericLiteral (DATE, TIMESTAMP types)
  to convert date strings to epoch microseconds for comparison
- ConstantExpression: add TimestampType and IntegerType block building

Fixes Q25, Q27 (timestamp sort), Q37, Q38, Q42 (DATE comparisons).

Author: penghuo

dqe/src/main/java/org/opensearch/sql/dqe/function/expression/ConstantExpression.java

@@ -13,6 +13,8 @@
 import io.trino.spi.type.BigintType;
 import io.trino.spi.type.BooleanType;
 import io.trino.spi.type.DoubleType;
+import io.trino.spi.type.IntegerType;
+import io.trino.spi.type.TimestampType;
 import io.trino.spi.type.Type;
 import io.trino.spi.type.VarcharType;
 import lombok.Getter;
@@ -51,8 +53,18 @@ private static Block buildSingleValueBlock(Object value, Type type) {
       VarcharType.VARCHAR.writeSlice(builder, Slices.utf8Slice(value.toString()));
     } else if (type instanceof BooleanType) {
       BooleanType.BOOLEAN.writeBoolean(builder, (Boolean) value);
+    } else if (type instanceof TimestampType) {
+      // Timestamp values are stored as long (microseconds since epoch)
+      type.writeLong(builder, ((Number) value).longValue());
+    } else if (type instanceof IntegerType) {
+      IntegerType.INTEGER.writeLong(builder, ((Number) value).intValue());
     } else {
-      throw new UnsupportedOperationException("Unsupported constant type: " + type);
+      // Fallback: try writeLong for numeric types
+      try {
+        type.writeLong(builder, ((Number) value).longValue());
+      } catch (Exception e) {
+        throw new UnsupportedOperationException("Unsupported constant type: " + type);
+      }
     }
     return builder.build();
   }

dqe/src/main/java/org/opensearch/sql/dqe/function/expression/ExpressionCompiler.java

@@ -9,6 +9,7 @@
 import io.trino.spi.type.BooleanType;
 import io.trino.spi.type.DoubleType;
 import io.trino.spi.type.IntegerType;
+import io.trino.spi.type.TimestampType;
 import io.trino.spi.type.Type;
 import io.trino.spi.type.VarcharType;
 import io.trino.sql.tree.ArithmeticBinaryExpression;
@@ -22,6 +23,7 @@
 import io.trino.sql.tree.DoubleLiteral;
 import io.trino.sql.tree.Expression;
 import io.trino.sql.tree.FunctionCall;
+import io.trino.sql.tree.GenericLiteral;
 import io.trino.sql.tree.Identifier;
 import io.trino.sql.tree.InListExpression;
 import io.trino.sql.tree.InPredicate;
@@ -81,6 +83,8 @@ public BlockExpression compile(Expression expr) {
       return new ConstantExpression(((BooleanLiteral) expr).getValue(), BooleanType.BOOLEAN);
     } else if (expr instanceof NullLiteral) {
       return new ConstantExpression(null, BigintType.BIGINT);
+    } else if (expr instanceof GenericLiteral generic) {
+      return compileGenericLiteral(generic);
     } else if (expr instanceof ComparisonExpression cmp) {
       return new ComparisonBlockExpression(
           cmp.getOperator(), compile(cmp.getLeft()), compile(cmp.getRight()));
@@ -140,6 +144,45 @@ public BlockExpression compile(Expression expr) {
         "Unsupported expression type: " + expr.getClass().getSimpleName());
   }
 
+  /**
+   * Compile a GenericLiteral (e.g., DATE '2013-07-01', TIMESTAMP '2013-07-01 12:00:00'). Converts
+   * the literal value to the appropriate Trino internal representation.
+   */
+  private BlockExpression compileGenericLiteral(GenericLiteral literal) {
+    String typeName = literal.getType().toUpperCase(java.util.Locale.ROOT);
+    String value = literal.getValue();
+
+    switch (typeName) {
+      case "DATE":
+        {
+          // Parse date string and convert to Trino timestamp micros
+          java.time.LocalDate date = java.time.LocalDate.parse(value);
+          long epochMicros = date.toEpochDay() * 86_400_000_000L;
+          return new ConstantExpression(epochMicros, TimestampType.TIMESTAMP_MILLIS);
+        }
+      case "TIMESTAMP":
+        {
+          // Parse timestamp string and convert to Trino timestamp micros
+          java.time.LocalDateTime dt =
+              java.time.LocalDateTime.parse(
+                  value, java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
+          long epochMicros =
+              dt.toEpochSecond(java.time.ZoneOffset.UTC) * 1_000_000L + dt.getNano() / 1_000L;
+          return new ConstantExpression(epochMicros, TimestampType.TIMESTAMP_MILLIS);
+        }
+      case "INTEGER":
+      case "INT":
+        return new ConstantExpression(Long.parseLong(value), BigintType.BIGINT);
+      case "BIGINT":
+        return new ConstantExpression(Long.parseLong(value), BigintType.BIGINT);
+      case "DOUBLE":
+        return new ConstantExpression(Double.parseDouble(value), DoubleType.DOUBLE);
+      default:
+        // Treat as VARCHAR for unknown types
+        return new ConstantExpression(value, VarcharType.VARCHAR);
+    }
+  }
+
   private BlockExpression compileIdentifier(Identifier identifier) {
     String name = identifier.getValue();
     Integer colIdx = columnIndexMap.get(name);

dqe/src/main/java/org/opensearch/sql/dqe/operator/SortOperator.java

@@ -9,8 +9,10 @@
 import io.trino.spi.block.Block;
 import io.trino.spi.block.BlockBuilder;
 import io.trino.spi.type.BigintType;
+import io.trino.spi.type.BooleanType;
 import io.trino.spi.type.DoubleType;
 import io.trino.spi.type.IntegerType;
+import io.trino.spi.type.TimestampType;
 import io.trino.spi.type.Type;
 import io.trino.spi.type.VarcharType;
 import java.util.ArrayList;
@@ -173,8 +175,19 @@ private static int compareValues(Block block, int posA, int posB, Type type) {
       return VarcharType.VARCHAR
           .getSlice(block, posA)
           .compareTo(VarcharType.VARCHAR.getSlice(block, posB));
+    } else if (type instanceof TimestampType) {
+      // TimestampType stores values as long (microseconds since epoch)
+      return Long.compare(type.getLong(block, posA), type.getLong(block, posB));
+    } else if (type instanceof BooleanType) {
+      return Boolean.compare(
+          BooleanType.BOOLEAN.getBoolean(block, posA), BooleanType.BOOLEAN.getBoolean(block, posB));
+    }
+    // Fallback: try comparing as longs for any other numeric type
+    try {
+      return Long.compare(type.getLong(block, posA), type.getLong(block, posB));
+    } catch (Exception e) {
+      throw new UnsupportedOperationException("Unsupported type for sorting: " + type);
     }
-    throw new UnsupportedOperationException("Unsupported type for sorting: " + type);
   }
 
   @Override