Fix timestamp precision (#255)

* Support timestamp-millis and timestamp-micros independently

Timestamp columns surfaced as TIMESTAMP(0) because Avro timestamp logical
types were mapped to TIMESTAMP with unspecified precision, and Calcite's
default RelDataTypeSystem caps datetime precision at 3 (milliseconds), so
micro/nanosecond precision could never survive.

- Add HoptimatorTypeSystem, which raises the datetime precision ceiling from
  3 to 9 (matching Flink SQL's TIMESTAMP(0..9)).
- Map Avro timestamp-millis to TIMESTAMP(3) and timestamp-micros to
  TIMESTAMP(6) in AvroConverter, and choose the logical type by precision on
  the reverse path.
- Route the JDBC planner/validator through HoptimatorTypeSystem via the
  Calcite typeSystem connection property, and repoint the standalone type
  factories in the Avro conversion path.
- Carry datetime precision through generated DDL in ScriptImplementor (e.g.
  TIMESTAMP(6)); a bare TIMESTAMP stays precision-less.

Adds unit coverage (type system, Avro mapping both directions, DDL
rendering), an end-to-end DDL Quidem test, and an integration script case.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Normalize tests onto HoptimatorTypeSystem

Swap RelDataTypeSystem.DEFAULT for HoptimatorTypeSystem.INSTANCE in test type
factories so tests mirror the type system used in production. Raising the
datetime precision ceiling does not affect existing assertions.

HoptimatorTypeSystemTest intentionally keeps RelDataTypeSystem.DEFAULT, since
it asserts the default's clamping behavior as the premise for the custom type
system. The delete-marked planner and catalog modules are left untouched.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Normalize remaining type factories onto HoptimatorTypeSystem

Route the remaining standalone type factories through HoptimatorTypeSystem so
nothing silently falls back to Calcite's default (which caps datetime
precision at 3).

- Main code: the compatibility validators (AvroTableValidator, Backward/
  ForwardCompatibilityValidator) and RemoteTable/ArrayTable built row types
  via new JavaTypeFactoryImpl() (default type system). With the default,
  micro/nanosecond timestamps collapse to TIMESTAMP(3), so the compat
  validators could not distinguish millis from micros. They now use
  HoptimatorTypeSystem.INSTANCE.
- Tests: normalize the remaining JavaTypeFactoryImpl() and
  RelDataTypeSystem.DEFAULT usages (demodb, planner, catalog, jdbc),
  adding a test-scoped hoptimator-avro dependency where needed.

HoptimatorTypeSystemTest intentionally still references
RelDataTypeSystem.DEFAULT, since it exists to contrast the default's
clamping behavior against the custom type system.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Preserve JDBC temporal fractional-seconds precision (MySQL)

A JDBC source's temporal columns were mapped using only the SQL type name, so
MySqlTable produced TIMESTAMP(0) for a TIMESTAMP(3)/(6) column — the
fractional-seconds precision JDBC exposes via DECIMAL_DIGITS was dropped at the
JDBC layer. Downstream, the planner's row type showed TIMESTAMP(0) even though
the source carried milli/microsecond precision.

Read DECIMAL_DIGITS for temporal types (TIME/TIMESTAMP and their local-time-zone
variants) and build the column with that precision, so TIMESTAMP(3)/(6) survive.

Adds MySqlTableTest cases for millis/micros/time/no-fraction, and an integration
script (mysql-timestamp-precision.id) with an `events` source table whose
TIMESTAMP(3)/(6) columns are asserted end-to-end via !describe.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Preserve temporal precision across the JDBC federation

HoptimatorJdbcTable derived its row type from the wrapped Calcite JdbcTable,
whose type comes from DatabaseMetaData. That round-trip drops temporal
fractional-seconds precision: CalciteMetaImpl reports a TIMESTAMP's precision in
COLUMN_SIZE while JdbcSchema#getRelDataType reads it from DECIMAL_DIGITS (left
null) and builds the proto type with RelDataTypeSystem.DEFAULT (capped at
millis). The net effect is TIMESTAMP(3)/(6) collapsing to TIMESTAMP(0) in the
planner even when the source carries milli/microsecond precision.

Calcite exposes no usable hook here (getRelDataType is package-private and its
proto is memoized; the THREAD_METADATA/Foo hook is experimental "do not use" and
only lists tables). So prefer the upstream table's own row type — the same
source of truth already used for the Avro key/value schemas — which preserves
precision, falling back to the JDBC-metadata path for non-Calcite upstreams.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Report temporal precision in DECIMAL_DIGITS across the JDBC federation

Replaces the upstream-table bypass with a native Calcite metadata fix. The
precision was being dropped inside Calcite's own JDBC federation: its metadata
producer (CalciteMetaImpl.columns) reports a TIMESTAMP/TIME column's precision
in COLUMN_SIZE and leaves DECIMAL_DIGITS null, but its reader
(JdbcSchema.getRelDataType) reads temporal precision from DECIMAL_DIGITS. So the
two halves disagree and TIMESTAMP(3) surfaces as TIMESTAMP(0).

Fix it the way an external JDBC driver would — populate DECIMAL_DIGITS — via
Calcite's native metaColumnFactory connection property (honored by Driver.
createMeta). HoptimatorMetaColumnFactory mirrors COLUMN_SIZE into DECIMAL_DIGITS
for temporal types, and CalciteDriver sets the property so every adapter
connection reports it. No federation bypass, no metadata round-trip change.

This restores millisecond precision across the federation; precision beyond
millis is still clamped by JdbcSchema.getRelDataType's hardcoded
RelDataTypeSystem.DEFAULT, which no connection knob can override.

Reverts the HoptimatorJdbcTable.getRowType bypass from the previous commit.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Remove flawed timestamp-precision integration scenarios

The mysql-timestamp-precision and kafka ts-precision integration scenarios
asserted micro/nanosecond precision (TIMESTAMP(6)/(9)) surviving a source read,
but that is unattainable through Calcite's JDBC federation: JdbcSchema.
getRelDataType builds its proto type with RelDataTypeSystem.DEFAULT, which caps
datetime precision at milliseconds. So those assertions could never hold
regardless of the metadata fix.

The MySQL scenario additionally depended on MySQL Connector/J reporting
DECIMAL_DIGITS for a TIMESTAMP(3) column, which is a separate, driver-specific
variable from the Calcite-over-Calcite federation fix.

Removes both scenarios and the associated events table. The millisecond fix
remains covered by unit tests (HoptimatorMetaColumnFactoryTest, MySqlTableTest).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix redundant check

* Fix MySQL precision/scale handling + support in quidem

* fix up tests

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: jogrogan

deploy/docker/mysql/init.sql

@@ -20,6 +20,7 @@ CREATE TABLE IF NOT EXISTS orders (
     quantity INT NOT NULL,
     price DECIMAL(10, 2) NOT NULL,
     order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    order_date_ms TIMESTAMP(3) DEFAULT CURRENT_TIMESTAMP(3),
     status VARCHAR(20) DEFAULT 'pending',
     FOREIGN KEY (user_id) REFERENCES users(user_id)
 );

hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/AvroConverter.java

@@ -1,12 +1,12 @@
 package com.linkedin.hoptimator.avro;
 
+import org.apache.avro.LogicalType;
 import org.apache.avro.LogicalTypes;
 import org.apache.avro.Schema;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rel.type.RelDataTypeImpl;
-import org.apache.calcite.rel.type.RelDataTypeSystem;
 import org.apache.calcite.rel.type.RelProtoDataType;
 import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
 import org.apache.calcite.sql.type.SqlTypeName;
@@ -27,6 +27,10 @@ public final class AvroConverter {
   private static final String KEY_OPTION = "key.fields";
   private static final String KEY_PREFIX_OPTION = "key.fields-prefix";
 
+  // Fractional-second precisions for Avro's timestamp logical types.
+  private static final int MILLIS_PRECISION = 3;
+  private static final int MICROS_PRECISION = 6;
+
   private AvroConverter() {
   }
 
@@ -75,8 +79,12 @@ public static Schema avro(String namespace, String name, RelDataType dataType) {
           return createAvroSchemaWithNullability(
               LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT)), dataType.isNullable());
         case TIMESTAMP:
+          // Avro distinguishes millisecond- and microsecond-precision timestamps. Map any
+          // precision beyond millis (3) to timestamp-micros, otherwise timestamp-millis.
+          LogicalType timestampLogicalType = dataType.getPrecision() > MILLIS_PRECISION
+              ? LogicalTypes.timestampMicros() : LogicalTypes.timestampMillis();
           return createAvroSchemaWithNullability(
-              LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG)), dataType.isNullable());
+              timestampLogicalType.addToSchema(Schema.create(Schema.Type.LONG)), dataType.isNullable());
         case BOOLEAN:
           return createAvroTypeWithNullability(Schema.Type.BOOLEAN, dataType.isNullable());
         case ARRAY:
@@ -101,7 +109,7 @@ public static Schema avro(String namespace, String name, RelDataType dataType) {
   }
 
   public static Schema avro(String namespace, String name, RelProtoDataType relProtoDataType) {
-    RelDataTypeFactory factory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+    RelDataTypeFactory factory = new SqlTypeFactoryImpl(HoptimatorTypeSystem.INSTANCE);
     return avro(namespace, name, relProtoDataType.apply(factory));
   }
 
@@ -118,7 +126,7 @@ public static Pair<Schema, Schema> avroKeyPayloadSchema(String namespace, String
       return new Pair<>(null, avro(namespace, payloadSchemaName, dataType));
     }
 
-    RelDataTypeFactory typeFactory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+    RelDataTypeFactory typeFactory = new SqlTypeFactoryImpl(HoptimatorTypeSystem.INSTANCE);
     RelDataTypeFactory.Builder keyBuilder = new RelDataTypeFactory.Builder(typeFactory);
     RelDataTypeFactory.Builder payloadBuilder = new RelDataTypeFactory.Builder(typeFactory);
 
@@ -289,8 +297,11 @@ public static RelDataType rel(Schema schema, RelDataTypeFactory typeFactory, boo
       case LONG:
         if (schema.getLogicalType() != null) {
           String logicalName = schema.getLogicalType().getName();
-          if ("timestamp-millis".equals(logicalName) || "timestamp-micros".equals(logicalName)) {
-            return createRelType(typeFactory, SqlTypeName.TIMESTAMP, nullable);
+          if ("timestamp-millis".equals(logicalName)) {
+            return createRelType(typeFactory, SqlTypeName.TIMESTAMP, MILLIS_PRECISION, nullable);
+          }
+          if ("timestamp-micros".equals(logicalName)) {
+            return createRelType(typeFactory, SqlTypeName.TIMESTAMP, MICROS_PRECISION, nullable);
           }
         }
         return createRelType(typeFactory, SqlTypeName.BIGINT, nullable);
@@ -335,7 +346,7 @@ public static RelDataType rel(Schema schema, RelDataTypeFactory typeFactory, boo
   }
 
   public static RelDataType rel(Schema schema) {
-    return rel(schema, new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT));
+    return rel(schema, new SqlTypeFactoryImpl(HoptimatorTypeSystem.INSTANCE));
   }
 
   private static RelDataType createRelType(RelDataTypeFactory typeFactory, SqlTypeName typeName, boolean nullable) {
@@ -349,7 +360,7 @@ private static RelDataType createRelType(RelDataTypeFactory typeFactory, SqlType
   }
 
   public static RelProtoDataType proto(Schema schema) {
-    return RelDataTypeImpl.proto(rel(schema, new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT)));
+    return RelDataTypeImpl.proto(rel(schema, new SqlTypeFactoryImpl(HoptimatorTypeSystem.INSTANCE)));
   }
 
   private static String describe(RelDataTypeField dataType) {

hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/AvroTableValidator.java

@@ -53,7 +53,7 @@ public void validate(Issues issues, DeploymentContext context) {
   }
 
   private void validate(SchemaPlus schema, Table table, Table originalTable, Issues issues) {
-    RelDataTypeFactory typeFactory = new JavaTypeFactoryImpl();
+    RelDataTypeFactory typeFactory = new JavaTypeFactoryImpl(HoptimatorTypeSystem.INSTANCE);
     RelDataType rowType = table.getRowType(typeFactory);
     RelDataType originalRowType = originalTable.getRowType(typeFactory);
     Schema avroSchema = AvroConverter.avro("ns", "n", rowType);

hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/HoptimatorTypeSystem.java

@@ -0,0 +1,39 @@
+package com.linkedin.hoptimator.avro;
+
+import org.apache.calcite.rel.type.RelDataTypeSystem;
+import org.apache.calcite.rel.type.RelDataTypeSystemImpl;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+
+/**
+ * Hoptimator's {@link RelDataTypeSystem}. It behaves exactly like Calcite's default system except
+ * that it raises the maximum precision of the datetime types from Calcite's hard-coded
+ * {@link SqlTypeName#MAX_DATETIME_PRECISION} (3, i.e. milliseconds) up to 9 (nanoseconds), matching
+ * what Flink SQL supports. Without this, {@code createSqlType(TIMESTAMP, 6)} — as produced for an
+ * Avro {@code timestamp-micros} column — is silently clamped down to {@code TIMESTAMP(3)}, so
+ * micro/nanosecond precision could never survive.
+ *
+ * <p>Exposed via a public {@link #INSTANCE} so it can be named through the {@code typeSystem} Calcite
+ * connection property (see {@code HoptimatorDriver}), which routes the whole planner/validator
+ * through it. Nothing is forced to a higher precision; the ceiling is merely raised.
+ */
+public final class HoptimatorTypeSystem extends RelDataTypeSystemImpl {
+
+  public static final RelDataTypeSystem INSTANCE = new HoptimatorTypeSystem();
+
+  // Flink SQL supports TIMESTAMP(0) through TIMESTAMP(9).
+  private static final int MAX_DATETIME_PRECISION = 9;
+
+  @Override
+  public int getMaxPrecision(SqlTypeName typeName) {
+    switch (typeName) {
+      case TIME:
+      case TIME_WITH_LOCAL_TIME_ZONE:
+      case TIMESTAMP:
+      case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+        return MAX_DATETIME_PRECISION;
+      default:
+        return super.getMaxPrecision(typeName);
+    }
+  }
+}