From 95b86ff01aeae7e21f09e0405ffaf86d806b8b6a Mon Sep 17 00:00:00 2001
From: Walaa Eldin Moustafa <wmoustafa@linkedin.com>
Date: Sat, 1 Nov 2025 02:23:31 -0700
Subject: [PATCH 1/6] Add Coral type system

---
 .../coral/common/types/CoralArrayType.java    | 65 +++++++++++++++
 .../coral/common/types/CoralCharType.java     | 68 +++++++++++++++
 .../coral/common/types/CoralDecimalType.java  | 83 +++++++++++++++++++
 .../coral/common/types/CoralMapType.java      | 77 +++++++++++++++++
 .../common/types/CoralPrimitiveType.java      | 58 +++++++++++++
 .../coral/common/types/CoralStructField.java  | 63 ++++++++++++++
 .../coral/common/types/CoralStructType.java   | 76 +++++++++++++++++
 .../common/types/CoralTimestampType.java      | 73 ++++++++++++++++
 .../coral/common/types/CoralVarcharType.java  | 68 +++++++++++++++
 .../common/types/CoralTypeSystemTest.java     |  1 -
 10 files changed, 631 insertions(+), 1 deletion(-)
 create mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralArrayType.java
 create mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralCharType.java
 create mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralDecimalType.java
 create mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralMapType.java
 create mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralPrimitiveType.java
 create mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructField.java
 create mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructType.java
 create mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralTimestampType.java
 create mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralVarcharType.java

diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralArrayType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralArrayType.java
new file mode 100644
index 000000000..653348892
--- /dev/null
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralArrayType.java
@@ -0,0 +1,65 @@
+/**
+ * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
+ * Licensed under the BSD-2 Clause license.
+ * See LICENSE in the project root for license information.
+ */
+package com.linkedin.coral.common.types;
+
+import java.util.Objects;
+
+
+/**
+ * Represents an array data type in the Coral type system.
+ */
+public final class CoralArrayType implements CoralDataType {
+  private final CoralDataType elementType;
+  private final boolean nullable;
+
+  /**
+   * Creates a new array type.
+   * @param elementType the type of elements in the array
+   * @param nullable whether this type allows null values
+   */
+  public CoralArrayType(CoralDataType elementType, boolean nullable) {
+    this.elementType = Objects.requireNonNull(elementType, "Element type cannot be null");
+    this.nullable = nullable;
+  }
+
+  /**
+   * Returns the type of elements in this array.
+   * @return the element type
+   */
+  public CoralDataType getElementType() {
+    return elementType;
+  }
+
+  @Override
+  public CoralTypeKind getKind() {
+    return CoralTypeKind.ARRAY;
+  }
+
+  @Override
+  public boolean isNullable() {
+    return nullable;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    CoralArrayType that = (CoralArrayType) o;
+    return nullable == that.nullable && Objects.equals(elementType, that.elementType);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(elementType, nullable);
+  }
+
+  @Override
+  public String toString() {
+    return "ARRAY<" + elementType + ">" + (nullable ? " NULL" : " NOT NULL");
+  }
+}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralCharType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralCharType.java
new file mode 100644
index 000000000..70c186161
--- /dev/null
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralCharType.java
@@ -0,0 +1,68 @@
+/**
+ * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
+ * Licensed under the BSD-2 Clause license.
+ * See LICENSE in the project root for license information.
+ */
+package com.linkedin.coral.common.types;
+
+import java.util.Objects;
+
+
+/**
+ * Represents a fixed-length character data type in the Coral type system.
+ */
+public final class CoralCharType implements CoralDataType {
+  private final int length;
+  private final boolean nullable;
+
+  /**
+   * Creates a new CHAR type.
+   * @param length the fixed length of the character string
+   * @param nullable whether this type allows null values
+   */
+  public CoralCharType(int length, boolean nullable) {
+    if (length <= 0) {
+      throw new IllegalArgumentException("Length must be positive, got: " + length);
+    }
+    this.length = length;
+    this.nullable = nullable;
+  }
+
+  /**
+   * Returns the fixed length of this CHAR type.
+   * @return the length
+   */
+  public int getLength() {
+    return length;
+  }
+
+  @Override
+  public CoralTypeKind getKind() {
+    return CoralTypeKind.CHAR;
+  }
+
+  @Override
+  public boolean isNullable() {
+    return nullable;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    CoralCharType that = (CoralCharType) o;
+    return length == that.length && nullable == that.nullable;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(length, nullable);
+  }
+
+  @Override
+  public String toString() {
+    return "CHAR(" + length + ")" + (nullable ? " NULL" : " NOT NULL");
+  }
+}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralDecimalType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralDecimalType.java
new file mode 100644
index 000000000..7e9ade3ea
--- /dev/null
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralDecimalType.java
@@ -0,0 +1,83 @@
+/**
+ * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
+ * Licensed under the BSD-2 Clause license.
+ * See LICENSE in the project root for license information.
+ */
+package com.linkedin.coral.common.types;
+
+import java.util.Objects;
+
+
+/**
+ * Represents a decimal data type with precision and scale in the Coral type system.
+ */
+public final class CoralDecimalType implements CoralDataType {
+  private final int precision;
+  private final int scale;
+  private final boolean nullable;
+
+  /**
+   * Creates a new decimal type.
+   * @param precision the total number of digits
+   * @param scale the number of digits after the decimal point
+   * @param nullable whether this type allows null values
+   */
+  public CoralDecimalType(int precision, int scale, boolean nullable) {
+    if (precision <= 0) {
+      throw new IllegalArgumentException("Precision must be positive, got: " + precision);
+    }
+    if (scale < 0 || scale > precision) {
+      throw new IllegalArgumentException(
+          "Scale must be non-negative and <= precision, got scale=" + scale + ", precision=" + precision);
+    }
+    this.precision = precision;
+    this.scale = scale;
+    this.nullable = nullable;
+  }
+
+  /**
+   * Returns the precision (total number of digits).
+   * @return the precision
+   */
+  public int getPrecision() {
+    return precision;
+  }
+
+  /**
+   * Returns the scale (number of digits after decimal point).
+   * @return the scale
+   */
+  public int getScale() {
+    return scale;
+  }
+
+  @Override
+  public CoralTypeKind getKind() {
+    return CoralTypeKind.DECIMAL;
+  }
+
+  @Override
+  public boolean isNullable() {
+    return nullable;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    CoralDecimalType that = (CoralDecimalType) o;
+    return precision == that.precision && scale == that.scale && nullable == that.nullable;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(precision, scale, nullable);
+  }
+
+  @Override
+  public String toString() {
+    return "DECIMAL(" + precision + "," + scale + ")" + (nullable ? " NULL" : " NOT NULL");
+  }
+}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralMapType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralMapType.java
new file mode 100644
index 000000000..d2607eda1
--- /dev/null
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralMapType.java
@@ -0,0 +1,77 @@
+/**
+ * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
+ * Licensed under the BSD-2 Clause license.
+ * See LICENSE in the project root for license information.
+ */
+package com.linkedin.coral.common.types;
+
+import java.util.Objects;
+
+
+/**
+ * Represents a map data type in the Coral type system.
+ */
+public final class CoralMapType implements CoralDataType {
+  private final CoralDataType keyType;
+  private final CoralDataType valueType;
+  private final boolean nullable;
+
+  /**
+   * Creates a new map type.
+   * @param keyType the type of keys in the map
+   * @param valueType the type of values in the map
+   * @param nullable whether this type allows null values
+   */
+  public CoralMapType(CoralDataType keyType, CoralDataType valueType, boolean nullable) {
+    this.keyType = Objects.requireNonNull(keyType, "Key type cannot be null");
+    this.valueType = Objects.requireNonNull(valueType, "Value type cannot be null");
+    this.nullable = nullable;
+  }
+
+  /**
+   * Returns the type of keys in this map.
+   * @return the key type
+   */
+  public CoralDataType getKeyType() {
+    return keyType;
+  }
+
+  /**
+   * Returns the type of values in this map.
+   * @return the value type
+   */
+  public CoralDataType getValueType() {
+    return valueType;
+  }
+
+  @Override
+  public CoralTypeKind getKind() {
+    return CoralTypeKind.MAP;
+  }
+
+  @Override
+  public boolean isNullable() {
+    return nullable;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    CoralMapType that = (CoralMapType) o;
+    return nullable == that.nullable && Objects.equals(keyType, that.keyType)
+        && Objects.equals(valueType, that.valueType);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(keyType, valueType, nullable);
+  }
+
+  @Override
+  public String toString() {
+    return "MAP<" + keyType + "," + valueType + ">" + (nullable ? " NULL" : " NOT NULL");
+  }
+}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralPrimitiveType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralPrimitiveType.java
new file mode 100644
index 000000000..0b75ff47e
--- /dev/null
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralPrimitiveType.java
@@ -0,0 +1,58 @@
+/**
+ * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
+ * Licensed under the BSD-2 Clause license.
+ * See LICENSE in the project root for license information.
+ */
+package com.linkedin.coral.common.types;
+
+import java.util.Objects;
+
+
+/**
+ * Represents a primitive data type in the Coral type system.
+ * This includes basic types like BOOLEAN, INT, DOUBLE, STRING, etc.
+ */
+public final class CoralPrimitiveType implements CoralDataType {
+  private final CoralTypeKind kind;
+  private final boolean nullable;
+
+  /**
+   * Creates a new primitive type.
+   * @param kind the type kind (must be a primitive type)
+   * @param nullable whether this type allows null values
+   */
+  public CoralPrimitiveType(CoralTypeKind kind, boolean nullable) {
+    this.kind = Objects.requireNonNull(kind, "Type kind cannot be null");
+    this.nullable = nullable;
+  }
+
+  @Override
+  public CoralTypeKind getKind() {
+    return kind;
+  }
+
+  @Override
+  public boolean isNullable() {
+    return nullable;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    CoralPrimitiveType that = (CoralPrimitiveType) o;
+    return nullable == that.nullable && kind == that.kind;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(kind, nullable);
+  }
+
+  @Override
+  public String toString() {
+    return kind.name() + (nullable ? " NULL" : " NOT NULL");
+  }
+}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructField.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructField.java
new file mode 100644
index 000000000..bcead3cdb
--- /dev/null
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructField.java
@@ -0,0 +1,63 @@
+/**
+ * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
+ * Licensed under the BSD-2 Clause license.
+ * See LICENSE in the project root for license information.
+ */
+package com.linkedin.coral.common.types;
+
+import java.util.Objects;
+
+
+/**
+ * Represents a field in a struct data type in the Coral type system.
+ */
+public final class CoralStructField {
+  private final String name;
+  private final CoralDataType type;
+
+  /**
+   * Creates a new struct field.
+   * @param name the name of the field
+   * @param type the type of the field
+   */
+  public CoralStructField(String name, CoralDataType type) {
+    this.name = Objects.requireNonNull(name, "Field name cannot be null");
+    this.type = Objects.requireNonNull(type, "Field type cannot be null");
+  }
+
+  /**
+   * Returns the name of this field.
+   * @return the field name
+   */
+  public String getName() {
+    return name;
+  }
+
+  /**
+   * Returns the type of this field.
+   * @return the field type
+   */
+  public CoralDataType getType() {
+    return type;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    CoralStructField that = (CoralStructField) o;
+    return Objects.equals(name, that.name) && Objects.equals(type, that.type);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(name, type);
+  }
+
+  @Override
+  public String toString() {
+    return name + ": " + type;
+  }
+}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructType.java
new file mode 100644
index 000000000..df20985c3
--- /dev/null
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructType.java
@@ -0,0 +1,76 @@
+/**
+ * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
+ * Licensed under the BSD-2 Clause license.
+ * See LICENSE in the project root for license information.
+ */
+package com.linkedin.coral.common.types;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+
+/**
+ * Represents a struct data type in the Coral type system.
+ */
+public final class CoralStructType implements CoralDataType {
+  private final List<CoralStructField> fields;
+  private final boolean nullable;
+
+  /**
+   * Creates a new struct type.
+   * @param fields the fields in the struct
+   * @param nullable whether this type allows null values
+   */
+  public CoralStructType(List<CoralStructField> fields, boolean nullable) {
+    this.fields = Collections.unmodifiableList(Objects.requireNonNull(fields, "Fields list cannot be null"));
+    this.nullable = nullable;
+  }
+
+  /**
+   * Returns the fields in this struct.
+   * @return an unmodifiable list of fields
+   */
+  public List<CoralStructField> getFields() {
+    return fields;
+  }
+
+  @Override
+  public CoralTypeKind getKind() {
+    return CoralTypeKind.STRUCT;
+  }
+
+  @Override
+  public boolean isNullable() {
+    return nullable;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    CoralStructType that = (CoralStructType) o;
+    return nullable == that.nullable && Objects.equals(fields, that.fields);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(fields, nullable);
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    sb.append("STRUCT<");
+    for (int i = 0; i < fields.size(); i++) {
+      if (i > 0)
+        sb.append(",");
+      sb.append(fields.get(i));
+    }
+    sb.append(">");
+    sb.append(nullable ? " NULL" : " NOT NULL");
+    return sb.toString();
+  }
+}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTimestampType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTimestampType.java
new file mode 100644
index 000000000..e371cfc20
--- /dev/null
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTimestampType.java
@@ -0,0 +1,73 @@
+/**
+ * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
+ * Licensed under the BSD-2 Clause license.
+ * See LICENSE in the project root for license information.
+ */
+package com.linkedin.coral.common.types;
+
+import java.util.Objects;
+
+
+/**
+ * Represents a TIMESTAMP type with fractional second precision in the Coral type system.
+ *
+ * Precision indicates the number of fractional digits of seconds, e.g.:
+ *  - 0: seconds
+ *  - 3: milliseconds
+ *  - 6: microseconds
+ *  - 9: nanoseconds
+ */
+public final class CoralTimestampType implements CoralDataType {
+  private final int precision;
+  private final boolean nullable;
+
+  /**
+   * Create a TIMESTAMP type with the given precision and nullability.
+   * @param precision fractional second precision (0-9)
+   * @param nullable whether this type allows null values
+   */
+  public CoralTimestampType(int precision, boolean nullable) {
+    if (precision < 0 || precision > 9) {
+      throw new IllegalArgumentException("Timestamp precision must be in range [0, 9], got: " + precision);
+    }
+    this.precision = precision;
+    this.nullable = nullable;
+  }
+
+  /**
+   * @return the fractional second precision (0-9)
+   */
+  public int getPrecision() {
+    return precision;
+  }
+
+  @Override
+  public CoralTypeKind getKind() {
+    return CoralTypeKind.TIMESTAMP;
+  }
+
+  @Override
+  public boolean isNullable() {
+    return nullable;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    CoralTimestampType that = (CoralTimestampType) o;
+    return precision == that.precision && nullable == that.nullable;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(precision, nullable);
+  }
+
+  @Override
+  public String toString() {
+    return "TIMESTAMP(" + precision + ")" + (nullable ? " NULL" : " NOT NULL");
+  }
+}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralVarcharType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralVarcharType.java
new file mode 100644
index 000000000..c9fd44a9a
--- /dev/null
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralVarcharType.java
@@ -0,0 +1,68 @@
+/**
+ * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
+ * Licensed under the BSD-2 Clause license.
+ * See LICENSE in the project root for license information.
+ */
+package com.linkedin.coral.common.types;
+
+import java.util.Objects;
+
+
+/**
+ * Represents a variable-length character data type in the Coral type system.
+ */
+public final class CoralVarcharType implements CoralDataType {
+  private final int length;
+  private final boolean nullable;
+
+  /**
+   * Creates a new VARCHAR type.
+   * @param length the maximum length of the character string
+   * @param nullable whether this type allows null values
+   */
+  public CoralVarcharType(int length, boolean nullable) {
+    if (length <= 0) {
+      throw new IllegalArgumentException("Length must be positive, got: " + length);
+    }
+    this.length = length;
+    this.nullable = nullable;
+  }
+
+  /**
+   * Returns the maximum length of this VARCHAR type.
+   * @return the length
+   */
+  public int getLength() {
+    return length;
+  }
+
+  @Override
+  public CoralTypeKind getKind() {
+    return CoralTypeKind.VARCHAR;
+  }
+
+  @Override
+  public boolean isNullable() {
+    return nullable;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    CoralVarcharType that = (CoralVarcharType) o;
+    return length == that.length && nullable == that.nullable;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(length, nullable);
+  }
+
+  @Override
+  public String toString() {
+    return "VARCHAR(" + length + ")" + (nullable ? " NULL" : " NOT NULL");
+  }
+}
diff --git a/coral-common/src/test/java/com/linkedin/coral/common/types/CoralTypeSystemTest.java b/coral-common/src/test/java/com/linkedin/coral/common/types/CoralTypeSystemTest.java
index 294fa1f57..ce1f47b02 100644
--- a/coral-common/src/test/java/com/linkedin/coral/common/types/CoralTypeSystemTest.java
+++ b/coral-common/src/test/java/com/linkedin/coral/common/types/CoralTypeSystemTest.java
@@ -203,7 +203,6 @@ public void testTypeEquality() {
     PrimitiveType intType1 = PrimitiveType.of(CoralTypeKind.INT, false);
     PrimitiveType intType2 = PrimitiveType.of(CoralTypeKind.INT, false);
     PrimitiveType nullableIntType = PrimitiveType.of(CoralTypeKind.INT, true);
-
     assertEquals(intType1, intType2);
     assertNotEquals(intType1, nullableIntType);
     assertEquals(intType1.hashCode(), intType2.hashCode());

From 5ea6754535bf789c57f912199e54c90faf4c6439 Mon Sep 17 00:00:00 2001
From: Walaa Eldin Moustafa <wmoustafa@linkedin.com>
Date: Sat, 8 Nov 2025 10:24:36 -0800
Subject: [PATCH 2/6] Remove Coral prefix from type names

---
 .../coral/common/types/CoralArrayType.java    | 65 ---------------
 .../coral/common/types/CoralCharType.java     | 68 ---------------
 .../coral/common/types/CoralDecimalType.java  | 83 -------------------
 .../coral/common/types/CoralMapType.java      | 77 -----------------
 .../common/types/CoralPrimitiveType.java      | 58 -------------
 .../coral/common/types/CoralStructField.java  | 63 --------------
 .../coral/common/types/CoralStructType.java   | 76 -----------------
 .../common/types/CoralTimestampType.java      | 73 ----------------
 .../coral/common/types/CoralVarcharType.java  | 68 ---------------
 9 files changed, 631 deletions(-)
 delete mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralArrayType.java
 delete mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralCharType.java
 delete mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralDecimalType.java
 delete mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralMapType.java
 delete mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralPrimitiveType.java
 delete mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructField.java
 delete mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructType.java
 delete mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralTimestampType.java
 delete mode 100644 coral-common/src/main/java/com/linkedin/coral/common/types/CoralVarcharType.java

diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralArrayType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralArrayType.java
deleted file mode 100644
index 653348892..000000000
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralArrayType.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/**
- * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
- * Licensed under the BSD-2 Clause license.
- * See LICENSE in the project root for license information.
- */
-package com.linkedin.coral.common.types;
-
-import java.util.Objects;
-
-
-/**
- * Represents an array data type in the Coral type system.
- */
-public final class CoralArrayType implements CoralDataType {
-  private final CoralDataType elementType;
-  private final boolean nullable;
-
-  /**
-   * Creates a new array type.
-   * @param elementType the type of elements in the array
-   * @param nullable whether this type allows null values
-   */
-  public CoralArrayType(CoralDataType elementType, boolean nullable) {
-    this.elementType = Objects.requireNonNull(elementType, "Element type cannot be null");
-    this.nullable = nullable;
-  }
-
-  /**
-   * Returns the type of elements in this array.
-   * @return the element type
-   */
-  public CoralDataType getElementType() {
-    return elementType;
-  }
-
-  @Override
-  public CoralTypeKind getKind() {
-    return CoralTypeKind.ARRAY;
-  }
-
-  @Override
-  public boolean isNullable() {
-    return nullable;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o)
-      return true;
-    if (o == null || getClass() != o.getClass())
-      return false;
-    CoralArrayType that = (CoralArrayType) o;
-    return nullable == that.nullable && Objects.equals(elementType, that.elementType);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(elementType, nullable);
-  }
-
-  @Override
-  public String toString() {
-    return "ARRAY<" + elementType + ">" + (nullable ? " NULL" : " NOT NULL");
-  }
-}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralCharType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralCharType.java
deleted file mode 100644
index 70c186161..000000000
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralCharType.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/**
- * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
- * Licensed under the BSD-2 Clause license.
- * See LICENSE in the project root for license information.
- */
-package com.linkedin.coral.common.types;
-
-import java.util.Objects;
-
-
-/**
- * Represents a fixed-length character data type in the Coral type system.
- */
-public final class CoralCharType implements CoralDataType {
-  private final int length;
-  private final boolean nullable;
-
-  /**
-   * Creates a new CHAR type.
-   * @param length the fixed length of the character string
-   * @param nullable whether this type allows null values
-   */
-  public CoralCharType(int length, boolean nullable) {
-    if (length <= 0) {
-      throw new IllegalArgumentException("Length must be positive, got: " + length);
-    }
-    this.length = length;
-    this.nullable = nullable;
-  }
-
-  /**
-   * Returns the fixed length of this CHAR type.
-   * @return the length
-   */
-  public int getLength() {
-    return length;
-  }
-
-  @Override
-  public CoralTypeKind getKind() {
-    return CoralTypeKind.CHAR;
-  }
-
-  @Override
-  public boolean isNullable() {
-    return nullable;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o)
-      return true;
-    if (o == null || getClass() != o.getClass())
-      return false;
-    CoralCharType that = (CoralCharType) o;
-    return length == that.length && nullable == that.nullable;
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(length, nullable);
-  }
-
-  @Override
-  public String toString() {
-    return "CHAR(" + length + ")" + (nullable ? " NULL" : " NOT NULL");
-  }
-}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralDecimalType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralDecimalType.java
deleted file mode 100644
index 7e9ade3ea..000000000
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralDecimalType.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/**
- * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
- * Licensed under the BSD-2 Clause license.
- * See LICENSE in the project root for license information.
- */
-package com.linkedin.coral.common.types;
-
-import java.util.Objects;
-
-
-/**
- * Represents a decimal data type with precision and scale in the Coral type system.
- */
-public final class CoralDecimalType implements CoralDataType {
-  private final int precision;
-  private final int scale;
-  private final boolean nullable;
-
-  /**
-   * Creates a new decimal type.
-   * @param precision the total number of digits
-   * @param scale the number of digits after the decimal point
-   * @param nullable whether this type allows null values
-   */
-  public CoralDecimalType(int precision, int scale, boolean nullable) {
-    if (precision <= 0) {
-      throw new IllegalArgumentException("Precision must be positive, got: " + precision);
-    }
-    if (scale < 0 || scale > precision) {
-      throw new IllegalArgumentException(
-          "Scale must be non-negative and <= precision, got scale=" + scale + ", precision=" + precision);
-    }
-    this.precision = precision;
-    this.scale = scale;
-    this.nullable = nullable;
-  }
-
-  /**
-   * Returns the precision (total number of digits).
-   * @return the precision
-   */
-  public int getPrecision() {
-    return precision;
-  }
-
-  /**
-   * Returns the scale (number of digits after decimal point).
-   * @return the scale
-   */
-  public int getScale() {
-    return scale;
-  }
-
-  @Override
-  public CoralTypeKind getKind() {
-    return CoralTypeKind.DECIMAL;
-  }
-
-  @Override
-  public boolean isNullable() {
-    return nullable;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o)
-      return true;
-    if (o == null || getClass() != o.getClass())
-      return false;
-    CoralDecimalType that = (CoralDecimalType) o;
-    return precision == that.precision && scale == that.scale && nullable == that.nullable;
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(precision, scale, nullable);
-  }
-
-  @Override
-  public String toString() {
-    return "DECIMAL(" + precision + "," + scale + ")" + (nullable ? " NULL" : " NOT NULL");
-  }
-}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralMapType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralMapType.java
deleted file mode 100644
index d2607eda1..000000000
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralMapType.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/**
- * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
- * Licensed under the BSD-2 Clause license.
- * See LICENSE in the project root for license information.
- */
-package com.linkedin.coral.common.types;
-
-import java.util.Objects;
-
-
-/**
- * Represents a map data type in the Coral type system.
- */
-public final class CoralMapType implements CoralDataType {
-  private final CoralDataType keyType;
-  private final CoralDataType valueType;
-  private final boolean nullable;
-
-  /**
-   * Creates a new map type.
-   * @param keyType the type of keys in the map
-   * @param valueType the type of values in the map
-   * @param nullable whether this type allows null values
-   */
-  public CoralMapType(CoralDataType keyType, CoralDataType valueType, boolean nullable) {
-    this.keyType = Objects.requireNonNull(keyType, "Key type cannot be null");
-    this.valueType = Objects.requireNonNull(valueType, "Value type cannot be null");
-    this.nullable = nullable;
-  }
-
-  /**
-   * Returns the type of keys in this map.
-   * @return the key type
-   */
-  public CoralDataType getKeyType() {
-    return keyType;
-  }
-
-  /**
-   * Returns the type of values in this map.
-   * @return the value type
-   */
-  public CoralDataType getValueType() {
-    return valueType;
-  }
-
-  @Override
-  public CoralTypeKind getKind() {
-    return CoralTypeKind.MAP;
-  }
-
-  @Override
-  public boolean isNullable() {
-    return nullable;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o)
-      return true;
-    if (o == null || getClass() != o.getClass())
-      return false;
-    CoralMapType that = (CoralMapType) o;
-    return nullable == that.nullable && Objects.equals(keyType, that.keyType)
-        && Objects.equals(valueType, that.valueType);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(keyType, valueType, nullable);
-  }
-
-  @Override
-  public String toString() {
-    return "MAP<" + keyType + "," + valueType + ">" + (nullable ? " NULL" : " NOT NULL");
-  }
-}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralPrimitiveType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralPrimitiveType.java
deleted file mode 100644
index 0b75ff47e..000000000
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralPrimitiveType.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/**
- * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
- * Licensed under the BSD-2 Clause license.
- * See LICENSE in the project root for license information.
- */
-package com.linkedin.coral.common.types;
-
-import java.util.Objects;
-
-
-/**
- * Represents a primitive data type in the Coral type system.
- * This includes basic types like BOOLEAN, INT, DOUBLE, STRING, etc.
- */
-public final class CoralPrimitiveType implements CoralDataType {
-  private final CoralTypeKind kind;
-  private final boolean nullable;
-
-  /**
-   * Creates a new primitive type.
-   * @param kind the type kind (must be a primitive type)
-   * @param nullable whether this type allows null values
-   */
-  public CoralPrimitiveType(CoralTypeKind kind, boolean nullable) {
-    this.kind = Objects.requireNonNull(kind, "Type kind cannot be null");
-    this.nullable = nullable;
-  }
-
-  @Override
-  public CoralTypeKind getKind() {
-    return kind;
-  }
-
-  @Override
-  public boolean isNullable() {
-    return nullable;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o)
-      return true;
-    if (o == null || getClass() != o.getClass())
-      return false;
-    CoralPrimitiveType that = (CoralPrimitiveType) o;
-    return nullable == that.nullable && kind == that.kind;
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(kind, nullable);
-  }
-
-  @Override
-  public String toString() {
-    return kind.name() + (nullable ? " NULL" : " NOT NULL");
-  }
-}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructField.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructField.java
deleted file mode 100644
index bcead3cdb..000000000
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructField.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/**
- * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
- * Licensed under the BSD-2 Clause license.
- * See LICENSE in the project root for license information.
- */
-package com.linkedin.coral.common.types;
-
-import java.util.Objects;
-
-
-/**
- * Represents a field in a struct data type in the Coral type system.
- */
-public final class CoralStructField {
-  private final String name;
-  private final CoralDataType type;
-
-  /**
-   * Creates a new struct field.
-   * @param name the name of the field
-   * @param type the type of the field
-   */
-  public CoralStructField(String name, CoralDataType type) {
-    this.name = Objects.requireNonNull(name, "Field name cannot be null");
-    this.type = Objects.requireNonNull(type, "Field type cannot be null");
-  }
-
-  /**
-   * Returns the name of this field.
-   * @return the field name
-   */
-  public String getName() {
-    return name;
-  }
-
-  /**
-   * Returns the type of this field.
-   * @return the field type
-   */
-  public CoralDataType getType() {
-    return type;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o)
-      return true;
-    if (o == null || getClass() != o.getClass())
-      return false;
-    CoralStructField that = (CoralStructField) o;
-    return Objects.equals(name, that.name) && Objects.equals(type, that.type);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(name, type);
-  }
-
-  @Override
-  public String toString() {
-    return name + ": " + type;
-  }
-}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructType.java
deleted file mode 100644
index df20985c3..000000000
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralStructType.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/**
- * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
- * Licensed under the BSD-2 Clause license.
- * See LICENSE in the project root for license information.
- */
-package com.linkedin.coral.common.types;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Objects;
-
-
-/**
- * Represents a struct data type in the Coral type system.
- */
-public final class CoralStructType implements CoralDataType {
-  private final List<CoralStructField> fields;
-  private final boolean nullable;
-
-  /**
-   * Creates a new struct type.
-   * @param fields the fields in the struct
-   * @param nullable whether this type allows null values
-   */
-  public CoralStructType(List<CoralStructField> fields, boolean nullable) {
-    this.fields = Collections.unmodifiableList(Objects.requireNonNull(fields, "Fields list cannot be null"));
-    this.nullable = nullable;
-  }
-
-  /**
-   * Returns the fields in this struct.
-   * @return an unmodifiable list of fields
-   */
-  public List<CoralStructField> getFields() {
-    return fields;
-  }
-
-  @Override
-  public CoralTypeKind getKind() {
-    return CoralTypeKind.STRUCT;
-  }
-
-  @Override
-  public boolean isNullable() {
-    return nullable;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o)
-      return true;
-    if (o == null || getClass() != o.getClass())
-      return false;
-    CoralStructType that = (CoralStructType) o;
-    return nullable == that.nullable && Objects.equals(fields, that.fields);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(fields, nullable);
-  }
-
-  @Override
-  public String toString() {
-    StringBuilder sb = new StringBuilder();
-    sb.append("STRUCT<");
-    for (int i = 0; i < fields.size(); i++) {
-      if (i > 0)
-        sb.append(",");
-      sb.append(fields.get(i));
-    }
-    sb.append(">");
-    sb.append(nullable ? " NULL" : " NOT NULL");
-    return sb.toString();
-  }
-}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTimestampType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTimestampType.java
deleted file mode 100644
index e371cfc20..000000000
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTimestampType.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/**
- * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
- * Licensed under the BSD-2 Clause license.
- * See LICENSE in the project root for license information.
- */
-package com.linkedin.coral.common.types;
-
-import java.util.Objects;
-
-
-/**
- * Represents a TIMESTAMP type with fractional second precision in the Coral type system.
- *
- * Precision indicates the number of fractional digits of seconds, e.g.:
- *  - 0: seconds
- *  - 3: milliseconds
- *  - 6: microseconds
- *  - 9: nanoseconds
- */
-public final class CoralTimestampType implements CoralDataType {
-  private final int precision;
-  private final boolean nullable;
-
-  /**
-   * Create a TIMESTAMP type with the given precision and nullability.
-   * @param precision fractional second precision (0-9)
-   * @param nullable whether this type allows null values
-   */
-  public CoralTimestampType(int precision, boolean nullable) {
-    if (precision < 0 || precision > 9) {
-      throw new IllegalArgumentException("Timestamp precision must be in range [0, 9], got: " + precision);
-    }
-    this.precision = precision;
-    this.nullable = nullable;
-  }
-
-  /**
-   * @return the fractional second precision (0-9)
-   */
-  public int getPrecision() {
-    return precision;
-  }
-
-  @Override
-  public CoralTypeKind getKind() {
-    return CoralTypeKind.TIMESTAMP;
-  }
-
-  @Override
-  public boolean isNullable() {
-    return nullable;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o)
-      return true;
-    if (o == null || getClass() != o.getClass())
-      return false;
-    CoralTimestampType that = (CoralTimestampType) o;
-    return precision == that.precision && nullable == that.nullable;
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(precision, nullable);
-  }
-
-  @Override
-  public String toString() {
-    return "TIMESTAMP(" + precision + ")" + (nullable ? " NULL" : " NOT NULL");
-  }
-}
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralVarcharType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralVarcharType.java
deleted file mode 100644
index c9fd44a9a..000000000
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralVarcharType.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/**
- * Copyright 2024-2025 LinkedIn Corporation. All rights reserved.
- * Licensed under the BSD-2 Clause license.
- * See LICENSE in the project root for license information.
- */
-package com.linkedin.coral.common.types;
-
-import java.util.Objects;
-
-
-/**
- * Represents a variable-length character data type in the Coral type system.
- */
-public final class CoralVarcharType implements CoralDataType {
-  private final int length;
-  private final boolean nullable;
-
-  /**
-   * Creates a new VARCHAR type.
-   * @param length the maximum length of the character string
-   * @param nullable whether this type allows null values
-   */
-  public CoralVarcharType(int length, boolean nullable) {
-    if (length <= 0) {
-      throw new IllegalArgumentException("Length must be positive, got: " + length);
-    }
-    this.length = length;
-    this.nullable = nullable;
-  }
-
-  /**
-   * Returns the maximum length of this VARCHAR type.
-   * @return the length
-   */
-  public int getLength() {
-    return length;
-  }
-
-  @Override
-  public CoralTypeKind getKind() {
-    return CoralTypeKind.VARCHAR;
-  }
-
-  @Override
-  public boolean isNullable() {
-    return nullable;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o)
-      return true;
-    if (o == null || getClass() != o.getClass())
-      return false;
-    CoralVarcharType that = (CoralVarcharType) o;
-    return length == that.length && nullable == that.nullable;
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(length, nullable);
-  }
-
-  @Override
-  public String toString() {
-    return "VARCHAR(" + length + ")" + (nullable ? " NULL" : " NOT NULL");
-  }
-}

From 0c4b97c8565ee010b2d5a203e6b0a04347424a65 Mon Sep 17 00:00:00 2001
From: Aastha Agrrawal <aagrrawal@linkedin.com>
Date: Sun, 16 Nov 2025 16:59:16 -0800
Subject: [PATCH 3/6] enhance coraltype system

---
 .../com/linkedin/coral/common/HiveTable.java  | 98 ++++++++++++++++++-
 .../common/HiveToCoralTypeConverter.java      | 18 +++-
 .../coral/common/types/CoralTypeKind.java     |  3 +
 .../CoralTypeToRelDataTypeConverter.java      | 10 +-
 .../coral/common/types/TimestampType.java     | 23 ++++-
 .../common/HiveToCoralTypeConverterTest.java  | 19 ++--
 6 files changed, 153 insertions(+), 18 deletions(-)

diff --git a/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java b/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
index 1df9fa940..07adbf32b 100644
--- a/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
+++ b/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
@@ -16,7 +16,10 @@
 import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
-
+import com.linkedin.coral.common.types.CoralDataType;
+import com.linkedin.coral.common.types.CoralTypeToRelDataTypeConverter;
+import com.linkedin.coral.common.types.StructField;
+import com.linkedin.coral.common.types.StructType;
 import org.apache.calcite.DataContext;
 import org.apache.calcite.config.CalciteConnectionConfig;
 import org.apache.calcite.linq4j.Enumerable;
@@ -134,12 +137,70 @@ private void checkDaliTable() {
     // Preconditions.checkState(isDaliTable());
   }
 
+  /**
+   * Returns the row type (schema) for this table.
+   *
+   * Two conversion paths are supported:
+   * 1. Two-stage (preferred): Hive → Coral → Calcite
+   * 2. Direct (legacy): Hive → Calcite (for backward compatibility)
+   *
+   * The two-stage conversion enables using Coral type system as an intermediary,
+   * allowing better type system unification and testing.
+   *
+   * @param typeFactory Calcite type factory
+   * @return RelDataType representing the table schema
+   */
   @Override
   public RelDataType getRowType(RelDataTypeFactory typeFactory) {
+    // Use two-stage conversion if HiveCoralTable is available
+    try {
+      return getRowTypeViaCoralTypeSystem(typeFactory);
+    } catch (Exception e) {
+      // Fall back to direct conversion if two-stage conversion fails
+      LOG.warn("Two-stage type conversion failed for table {}, falling back to direct conversion. Error: {}",
+          hiveTable.getTableName(), e.getMessage(), e);
+      return getRowTypeDirectConversion(typeFactory);
+    }
+  }
+
+  /**
+   * Two-stage conversion: Hive → Coral → Calcite.
+   * This is the preferred path when using CoralCatalog.
+   */
+  private RelDataType getRowTypeViaCoralTypeSystem(RelDataTypeFactory typeFactory) {
+    // Stage 1: Hive → Coral
+    CoralDataType coralSchema = getCoralSchema();
+
+    // Stage 2: Coral → Calcite
+    if (!(coralSchema instanceof StructType)) {
+      throw new IllegalStateException("Expected StructType from getCoralSchema(), got: " + coralSchema.getClass());
+    }
+
+    StructType structType = (StructType) coralSchema;
+    List<StructField> fields = structType.getFields();
+
+    List<RelDataType> fieldTypes = new ArrayList<>(fields.size());
+    List<String> fieldNames = new ArrayList<>(fields.size());
+
+    for (StructField field : fields) {
+      fieldNames.add(field.getName());
+      RelDataType fieldType = CoralTypeToRelDataTypeConverter.convert(field.getType(), typeFactory);
+      fieldTypes.add(fieldType);
+    }
+
+    return typeFactory.createStructType(fieldTypes, fieldNames);
+  }
+
+  /**
+   * Direct conversion: Hive → Calcite.
+   * This is the legacy path for backward compatibility.
+   */
+  private RelDataType getRowTypeDirectConversion(RelDataTypeFactory typeFactory) {
     final List<FieldSchema> cols = getColumns();
     final List<RelDataType> fieldTypes = new ArrayList<>(cols.size());
     final List<String> fieldNames = new ArrayList<>(cols.size());
     final Iterable<FieldSchema> allCols = Iterables.concat(cols, hiveTable.getPartitionKeys());
+
     allCols.forEach(col -> {
       final TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(col.getType());
       final RelDataType relType = TypeConverter.convert(typeInfo, typeFactory);
@@ -153,6 +214,41 @@ public RelDataType getRowType(RelDataTypeFactory typeFactory) {
     return typeFactory.createStructType(fieldTypes, fieldNames);
   }
 
+  /**
+   * Returns the table schema in Coral type system.
+   * This includes both regular columns (from StorageDescriptor) and partition columns.
+   * Converts Hive TypeInfo to Coral types using HiveToCoralTypeConverter.
+   *
+   * @return StructType representing the full table schema (columns + partitions)
+   */
+  @Override
+  public CoralDataType getCoralSchema() {
+    final List<FieldSchema> cols = getColumns();
+    final List<StructField> fields = new ArrayList<>();
+    final List<String> fieldNames = new ArrayList<>();
+
+    // Combine regular columns and partition keys (same as HiveTable.getRowType)
+    final Iterable<FieldSchema> allCols = Iterables.concat(cols, hiveTable.getPartitionKeys());
+
+    for (FieldSchema col : allCols) {
+      final String colName = col.getName();
+
+      // Skip duplicate columns (partition keys might overlap with regular columns)
+      if (!fieldNames.contains(colName)) {
+        // Convert Hive type string to TypeInfo, then to CoralDataType
+        final TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(col.getType());
+        final CoralDataType coralType = HiveToCoralTypeConverter.convert(typeInfo);
+
+        fields.add(StructField.of(colName, coralType));
+        fieldNames.add(colName);
+      }
+    }
+
+    // Return struct type representing the table schema
+    // Table-level struct is nullable (Hive convention)
+    return StructType.of(fields, true);
+  }
+
   private List<FieldSchema> getColumns() {
     StorageDescriptor sd = hiveTable.getSd();
     String serDeLib = getSerializationLib();
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/HiveToCoralTypeConverter.java b/coral-common/src/main/java/com/linkedin/coral/common/HiveToCoralTypeConverter.java
index b832134f3..a5cd99e71 100644
--- a/coral-common/src/main/java/com/linkedin/coral/common/HiveToCoralTypeConverter.java
+++ b/coral-common/src/main/java/com/linkedin/coral/common/HiveToCoralTypeConverter.java
@@ -72,8 +72,9 @@ private static CoralDataType convertPrimitive(PrimitiveTypeInfo type) {
       case DATE:
         return PrimitiveType.of(CoralTypeKind.DATE, nullable);
       case TIMESTAMP:
-        // Default to microsecond precision (6)
-        return TimestampType.of(3, nullable);
+        // Hive TIMESTAMP has no explicit precision (matches TypeConverter behavior)
+        // Use PRECISION_NOT_SPECIFIED (-1) to match Calcite's behavior
+        return TimestampType.of(TimestampType.PRECISION_NOT_SPECIFIED, nullable);
       case BINARY:
         return PrimitiveType.of(CoralTypeKind.BINARY, nullable);
       case DECIMAL:
@@ -86,6 +87,7 @@ private static CoralDataType convertPrimitive(PrimitiveTypeInfo type) {
         CharTypeInfo charType = (CharTypeInfo) type;
         return CharType.of(charType.getLength(), nullable);
       case VOID:
+        return PrimitiveType.of(CoralTypeKind.NULL, true);
       case UNKNOWN:
         return PrimitiveType.of(CoralTypeKind.STRING, true); // Map to nullable string as a fallback
       default:
@@ -118,12 +120,18 @@ private static CoralDataType convertStruct(StructTypeInfo structType) {
   }
 
   private static CoralDataType convertUnion(UnionTypeInfo unionType) {
-    // For UNION types, we'll create a struct with all possible fields
-    // This is similar to how some systems handle union types
+    // For UNION types, create a struct conforming to Trino's union representation
+    // Schema: {tag, field0, field1, ..., fieldN}
+    // See: https://github.com/trinodb/trino/pull/3483
     List<TypeInfo> memberTypes = unionType.getAllUnionObjectTypeInfos();
 
-    // Create fields for each possible type in the union
+    // Create fields: "tag" field first (INTEGER), then "field0", "field1", etc.
     List<StructField> fields = new ArrayList<>();
+
+    // Add "tag" field (INTEGER) to indicate which union member is active
+    fields.add(StructField.of("tag", PrimitiveType.of(CoralTypeKind.INT, true)));
+
+    // Add fields for each possible type in the union
     for (int i = 0; i < memberTypes.size(); i++) {
       CoralDataType fieldType = convert(memberTypes.get(i));
       fields.add(StructField.of("field" + i, fieldType));
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTypeKind.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTypeKind.java
index e7aff9806..8d0160cc2 100644
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTypeKind.java
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTypeKind.java
@@ -34,6 +34,9 @@ public enum CoralTypeKind {
   // Binary types
   BINARY,
 
+  // Special types
+  NULL,
+
   // Complex types
   ARRAY,
   MAP,
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTypeToRelDataTypeConverter.java b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTypeToRelDataTypeConverter.java
index dd783cd4c..6b571c8f5 100644
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTypeToRelDataTypeConverter.java
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/CoralTypeToRelDataTypeConverter.java
@@ -36,7 +36,13 @@ public static RelDataType convert(CoralDataType type, RelDataTypeFactory factory
       relType = convertPrimitive((PrimitiveType) type, factory);
     } else if (type instanceof TimestampType) {
       TimestampType ts = (TimestampType) type;
-      relType = factory.createSqlType(SqlTypeName.TIMESTAMP, ts.getPrecision());
+      // Handle unspecified precision (Hive compatibility)
+      if (ts.hasPrecision()) {
+        relType = factory.createSqlType(SqlTypeName.TIMESTAMP, ts.getPrecision());
+      } else {
+        // No precision specified - matches TypeConverter behavior
+        relType = factory.createSqlType(SqlTypeName.TIMESTAMP);
+      }
     } else if (type instanceof DecimalType) {
       DecimalType dec = (DecimalType) type;
       relType = factory.createSqlType(SqlTypeName.DECIMAL, dec.getPrecision(), dec.getScale());
@@ -107,6 +113,8 @@ private static RelDataType convertPrimitive(PrimitiveType prim, RelDataTypeFacto
         return factory.createSqlType(SqlTypeName.TIME);
       case BINARY:
         return factory.createSqlType(SqlTypeName.BINARY);
+      case NULL:
+        return factory.createSqlType(SqlTypeName.NULL);
       default:
         // Fallback for unsupported primitive types
         return factory.createSqlType(SqlTypeName.ANY);
diff --git a/coral-common/src/main/java/com/linkedin/coral/common/types/TimestampType.java b/coral-common/src/main/java/com/linkedin/coral/common/types/TimestampType.java
index cb9f3f5b6..e875e7fed 100644
--- a/coral-common/src/main/java/com/linkedin/coral/common/types/TimestampType.java
+++ b/coral-common/src/main/java/com/linkedin/coral/common/types/TimestampType.java
@@ -12,23 +12,28 @@
  * Represents a TIMESTAMP type with fractional second precision in the Coral type system.
  *
  * Precision indicates the number of fractional digits of seconds, e.g.:
+ *  - -1: unspecified (PRECISION_NOT_SPECIFIED, for Hive compatibility)
  *  - 0: seconds
  *  - 3: milliseconds
  *  - 6: microseconds
  *  - 9: nanoseconds
  */
 public final class TimestampType implements CoralDataType {
+  /** Constant for unspecified precision (matches Calcite's RelDataType.PRECISION_NOT_SPECIFIED) */
+  public static final int PRECISION_NOT_SPECIFIED = -1;
+
   private final int precision;
   private final boolean nullable;
 
   /**
    * Create a TIMESTAMP type with the given precision and nullability.
-   * @param precision fractional second precision (0-9)
+   * @param precision fractional second precision (-1 for unspecified, or 0-9)
    * @param nullable whether this type allows null values
    */
   public static TimestampType of(int precision, boolean nullable) {
-    if (precision < 0 || precision > 9) {
-      throw new IllegalArgumentException("Timestamp precision must be in range [0, 9], got: " + precision);
+    if (precision != PRECISION_NOT_SPECIFIED && (precision < 0 || precision > 9)) {
+      throw new IllegalArgumentException(
+          "Timestamp precision must be -1 (unspecified) or in range [0, 9], got: " + precision);
     }
     return new TimestampType(precision, nullable);
   }
@@ -39,12 +44,19 @@ private TimestampType(int precision, boolean nullable) {
   }
 
   /**
-   * @return the fractional second precision (0-9)
+   * @return the fractional second precision (-1 for unspecified, or 0-9)
    */
   public int getPrecision() {
     return precision;
   }
 
+  /**
+   * @return true if precision is explicitly specified, false if unspecified
+   */
+  public boolean hasPrecision() {
+    return precision != PRECISION_NOT_SPECIFIED;
+  }
+
   @Override
   public CoralTypeKind getKind() {
     return CoralTypeKind.TIMESTAMP;
@@ -72,6 +84,7 @@ public int hashCode() {
 
   @Override
   public String toString() {
-    return "TIMESTAMP(" + precision + ")" + (nullable ? " NULL" : " NOT NULL");
+    String precisionStr = precision == PRECISION_NOT_SPECIFIED ? "" : "(" + precision + ")";
+    return "TIMESTAMP" + precisionStr + (nullable ? " NULL" : " NOT NULL");
   }
 }
diff --git a/coral-common/src/test/java/com/linkedin/coral/common/HiveToCoralTypeConverterTest.java b/coral-common/src/test/java/com/linkedin/coral/common/HiveToCoralTypeConverterTest.java
index be9348542..39ebe91e1 100644
--- a/coral-common/src/test/java/com/linkedin/coral/common/HiveToCoralTypeConverterTest.java
+++ b/coral-common/src/test/java/com/linkedin/coral/common/HiveToCoralTypeConverterTest.java
@@ -23,6 +23,9 @@ public class HiveToCoralTypeConverterTest {
 
   @Test
   public void testPrimitiveTypes() {
+    // Test void/null type
+    testPrimitiveType(TypeInfoFactory.voidTypeInfo, CoralTypeKind.NULL, true, null, null);
+
     // Test boolean
     testPrimitiveType(TypeInfoFactory.booleanTypeInfo, CoralTypeKind.BOOLEAN, true, null, null);
 
@@ -39,7 +42,8 @@ public void testPrimitiveTypes() {
 
     // Test date/time types
     testPrimitiveType(TypeInfoFactory.dateTypeInfo, CoralTypeKind.DATE, true, null, null);
-    testPrimitiveType(TypeInfoFactory.timestampTypeInfo, CoralTypeKind.TIMESTAMP, true, 3, null);
+    // TIMESTAMP has PRECISION_NOT_SPECIFIED (-1) to match legacy TypeConverter behavior
+    testPrimitiveType(TypeInfoFactory.timestampTypeInfo, CoralTypeKind.TIMESTAMP, true, -1, null);
 
     // Test binary
     testPrimitiveType(TypeInfoFactory.binaryTypeInfo, CoralTypeKind.BINARY, true, null, null);
@@ -193,13 +197,16 @@ public void testUnionType() {
     assertTrue(result instanceof StructType);
     StructType structType = (StructType) result;
 
-    // Union is converted to a struct with fields for each possible type
+    // Union is converted to a struct with "tag" field first, then fields for each possible type
+    // This matches the Trino union representation: {tag, field0, field1, ...}
     List<StructField> fields = structType.getFields();
-    assertEquals(fields.size(), 2);
-    assertEquals(fields.get(0).getName(), "field0");
+    assertEquals(fields.size(), 3); // tag + 2 union member fields
+    assertEquals(fields.get(0).getName(), "tag");
     assertEquals(fields.get(0).getType().getKind(), CoralTypeKind.INT);
-    assertEquals(fields.get(1).getName(), "field1");
-    assertEquals(fields.get(1).getType().getKind(), CoralTypeKind.STRING);
+    assertEquals(fields.get(1).getName(), "field0");
+    assertEquals(fields.get(1).getType().getKind(), CoralTypeKind.INT);
+    assertEquals(fields.get(2).getName(), "field1");
+    assertEquals(fields.get(2).getType().getKind(), CoralTypeKind.STRING);
   }
 
   @Test(expectedExceptions = IllegalArgumentException.class)

From 5ac359f0830932764c4d60ee3717088555dac9af Mon Sep 17 00:00:00 2001
From: Aastha Agrrawal <aagrrawal@linkedin.com>
Date: Sun, 16 Nov 2025 18:42:00 -0800
Subject: [PATCH 4/6] fix build

---
 .../java/com/linkedin/coral/common/HiveTable.java   | 13 +++++++------
 1 file changed, 7 insertions(+), 6 deletions(-)

diff --git a/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java b/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
index 07adbf32b..987b95df6 100644
--- a/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
+++ b/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
@@ -1,5 +1,5 @@
 /**
- * Copyright 2017-2023 LinkedIn Corporation. All rights reserved.
+ * Copyright 2017-2025 LinkedIn Corporation. All rights reserved.
  * Licensed under the BSD-2 Clause license.
  * See LICENSE in the project root for license information.
  */
@@ -16,10 +16,7 @@
 import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
-import com.linkedin.coral.common.types.CoralDataType;
-import com.linkedin.coral.common.types.CoralTypeToRelDataTypeConverter;
-import com.linkedin.coral.common.types.StructField;
-import com.linkedin.coral.common.types.StructType;
+
 import org.apache.calcite.DataContext;
 import org.apache.calcite.config.CalciteConnectionConfig;
 import org.apache.calcite.linq4j.Enumerable;
@@ -42,6 +39,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.linkedin.coral.common.types.CoralDataType;
+import com.linkedin.coral.common.types.CoralTypeToRelDataTypeConverter;
+import com.linkedin.coral.common.types.StructField;
+import com.linkedin.coral.common.types.StructType;
+
 
 /**
  * Adaptor class from Hive {@link org.apache.hadoop.hive.metastore.api.Table} representation to
@@ -221,7 +223,6 @@ private RelDataType getRowTypeDirectConversion(RelDataTypeFactory typeFactory) {
    *
    * @return StructType representing the full table schema (columns + partitions)
    */
-  @Override
   public CoralDataType getCoralSchema() {
     final List<FieldSchema> cols = getColumns();
     final List<StructField> fields = new ArrayList<>();

From 080722c09d7518457568315e1b2ecd568a95415f Mon Sep 17 00:00:00 2001
From: Aastha Agrrawal <aagrrawal@linkedin.com>
Date: Wed, 19 Nov 2025 16:58:12 -0800
Subject: [PATCH 5/6] rename pvt methods

---
 .../main/java/com/linkedin/coral/common/HiveTable.java    | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java b/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
index 987b95df6..d84babbb4 100644
--- a/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
+++ b/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
@@ -156,12 +156,12 @@ private void checkDaliTable() {
   public RelDataType getRowType(RelDataTypeFactory typeFactory) {
     // Use two-stage conversion if HiveCoralTable is available
     try {
-      return getRowTypeViaCoralTypeSystem(typeFactory);
+      return getRowTypeFromCoralType(typeFactory);
     } catch (Exception e) {
       // Fall back to direct conversion if two-stage conversion fails
       LOG.warn("Two-stage type conversion failed for table {}, falling back to direct conversion. Error: {}",
           hiveTable.getTableName(), e.getMessage(), e);
-      return getRowTypeDirectConversion(typeFactory);
+      return getRowTypeFromHiveType(typeFactory);
     }
   }
 
@@ -169,7 +169,7 @@ public RelDataType getRowType(RelDataTypeFactory typeFactory) {
    * Two-stage conversion: Hive → Coral → Calcite.
    * This is the preferred path when using CoralCatalog.
    */
-  private RelDataType getRowTypeViaCoralTypeSystem(RelDataTypeFactory typeFactory) {
+  private RelDataType getRowTypeFromCoralType(RelDataTypeFactory typeFactory) {
     // Stage 1: Hive → Coral
     CoralDataType coralSchema = getCoralSchema();
 
@@ -197,7 +197,7 @@ private RelDataType getRowTypeViaCoralTypeSystem(RelDataTypeFactory typeFactory)
    * Direct conversion: Hive → Calcite.
    * This is the legacy path for backward compatibility.
    */
-  private RelDataType getRowTypeDirectConversion(RelDataTypeFactory typeFactory) {
+  private RelDataType getRowTypeFromHiveType(RelDataTypeFactory typeFactory) {
     final List<FieldSchema> cols = getColumns();
     final List<RelDataType> fieldTypes = new ArrayList<>(cols.size());
     final List<String> fieldNames = new ArrayList<>(cols.size());

From eb470f25aa28c71c4e93def5481fc535aa2e621f Mon Sep 17 00:00:00 2001
From: Aastha Agrrawal <aagrrawal@linkedin.com>
Date: Thu, 20 Nov 2025 10:59:44 -0800
Subject: [PATCH 6/6] prioritize original conversion over coral type system

---
 .../com/linkedin/coral/common/HiveTable.java  | 34 +++++++++++++------
 1 file changed, 23 insertions(+), 11 deletions(-)

diff --git a/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java b/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
index d84babbb4..e5807ffaf 100644
--- a/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
+++ b/coral-common/src/main/java/com/linkedin/coral/common/HiveTable.java
@@ -142,27 +142,39 @@ private void checkDaliTable() {
   /**
    * Returns the row type (schema) for this table.
    *
-   * Two conversion paths are supported:
-   * 1. Two-stage (preferred): Hive → Coral → Calcite
-   * 2. Direct (legacy): Hive → Calcite (for backward compatibility)
+   * Current behavior (validation/shadow mode):
+   * - Always returns the legacy Hive → Calcite direct conversion
+   * - Validates against the new Hive → Coral → Calcite two-stage conversion
+   * - Logs warnings if conversions don't match or if validation fails
    *
-   * The two-stage conversion enables using Coral type system as an intermediary,
-   * allowing better type system unification and testing.
+   * This allows safe validation of the new conversion path in production
+   * before switching to use it as the primary path.
    *
    * @param typeFactory Calcite type factory
    * @return RelDataType representing the table schema
    */
   @Override
   public RelDataType getRowType(RelDataTypeFactory typeFactory) {
-    // Use two-stage conversion if HiveCoralTable is available
+    // Always compute and return the legacy Hive direct conversion (production path)
+    RelDataType hiveType = getRowTypeFromHiveType(typeFactory);
+
+    // Validate against new two-stage Coral conversion (shadow/validation mode)
     try {
-      return getRowTypeFromCoralType(typeFactory);
+      RelDataType coralType = getRowTypeFromCoralType(typeFactory);
+
+      // Compare the two type representations
+      if (!hiveType.equals(coralType)) {
+        LOG.warn("Hive and Coral type conversion mismatch for table {}.{}. Hive: {}, Coral: {}", hiveTable.getDbName(),
+            hiveTable.getTableName(), hiveType, coralType);
+      }
     } catch (Exception e) {
-      // Fall back to direct conversion if two-stage conversion fails
-      LOG.warn("Two-stage type conversion failed for table {}, falling back to direct conversion. Error: {}",
-          hiveTable.getTableName(), e.getMessage(), e);
-      return getRowTypeFromHiveType(typeFactory);
+      // Log validation failure but continue with Hive type (zero production impact)
+      LOG.warn("Coral type validation failed for table {}.{}. Proceeding with Hive type. Error: {}",
+          hiveTable.getDbName(), hiveTable.getTableName(), e.getMessage(), e);
     }
+
+    // Always return the battle-tested Hive conversion result
+    return hiveType;
   }
 
   /**