fix: [iceberg] Add LogicalTypeAnnotation in ParquetColumnSpec

common/src/main/java/org/apache/comet/parquet/ParquetColumnSpec.java

@@ -19,6 +19,8 @@
 
 package org.apache.comet.parquet;
 
+import java.util.Map;
+
 public class ParquetColumnSpec {
 
   private final String[] path;
@@ -27,20 +29,26 @@ public class ParquetColumnSpec {
   private final boolean isRepeated;
   private final int maxDefinitionLevel;
   private final int maxRepetitionLevel;
+  private String logicalTypeName;
+  private Map<String, String> logicalTypeParams;
 
   public ParquetColumnSpec(
       String[] path,
       String physicalType,
       int typeLength,
       boolean isRepeated,
       int maxDefinitionLevel,
-      int maxRepetitionLevel) {
+      int maxRepetitionLevel,
+      String logicalTypeName,
+      Map<String, String> logicalTypeParams) {
     this.path = path;
     this.physicalType = physicalType;
     this.typeLength = typeLength;
     this.isRepeated = isRepeated;
     this.maxDefinitionLevel = maxDefinitionLevel;
     this.maxRepetitionLevel = maxRepetitionLevel;
+    this.logicalTypeName = logicalTypeName;
+    this.logicalTypeParams = logicalTypeParams;
   }
 
   public String[] getPath() {
@@ -66,4 +74,12 @@ public int getMaxRepetitionLevel() {
   public int getMaxDefinitionLevel() {
     return maxDefinitionLevel;
   }
+
+  public String getLogicalTypeName() {
+    return logicalTypeName;
+  }
+
+  public Map<String, String> getLogicalTypeParams() {
+    return logicalTypeParams;
+  }
 }

common/src/main/java/org/apache/comet/parquet/Utils.java

@@ -24,6 +24,7 @@
 import org.apache.parquet.schema.MessageType;
 import org.apache.parquet.schema.PrimitiveType;
 import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
 import org.apache.spark.sql.types.*;
 
 import org.apache.comet.CometSchemaImporter;
@@ -290,15 +291,129 @@ public static ColumnDescriptor buildColumnDescriptor(ParquetColumnSpec columnSpe
     }
 
     String name = columnSpec.getPath()[columnSpec.getPath().length - 1];
+    // Reconstruct the logical type from parameters
+    LogicalTypeAnnotation logicalType = null;
+    if (columnSpec.getLogicalTypeName() != null) {
+      logicalType =
+          reconstructLogicalType(
+              columnSpec.getLogicalTypeName(), columnSpec.getLogicalTypeParams());
+    }
 
     PrimitiveType primitiveType;
     if (primType == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) {
-      primitiveType = new PrimitiveType(repetition, primType, columnSpec.getTypeLength(), name);
+      primitiveType =
+          Types.primitive(primType, repetition)
+              .length(columnSpec.getTypeLength())
+              .as(logicalType)
+              .named(name);
     } else {
-      primitiveType = new PrimitiveType(repetition, primType, name);
+      primitiveType = Types.primitive(primType, repetition).as(logicalType).named(name);
     }
 
     MessageType schema = new MessageType("root", primitiveType);
     return schema.getColumnDescription(columnSpec.getPath());
   }
+
+  private static LogicalTypeAnnotation reconstructLogicalType(
+      String logicalTypeName, java.util.Map<String, String> params) {
+
+    switch (logicalTypeName) {
+        // MAP
+      case "MapLogicalTypeAnnotation":
+        return LogicalTypeAnnotation.mapType();
+
+        // LIST
+      case "ListLogicalTypeAnnotation":
+        return LogicalTypeAnnotation.listType();
+
+        // STRING
+      case "StringLogicalTypeAnnotation":
+        return LogicalTypeAnnotation.stringType();
+
+        // MAP_KEY_VALUE
+      case "MapKeyValueLogicalTypeAnnotation":
+        return LogicalTypeAnnotation.MapKeyValueTypeAnnotation.getInstance();
+
+        // ENUM
+      case "EnumLogicalTypeAnnotation":
+        return LogicalTypeAnnotation.enumType();
+
+        // DECIMAL
+      case "DecimalLogicalTypeAnnotation":
+        int scale = Integer.parseInt(params.get("scale"));
+        int precision = Integer.parseInt(params.get("precision"));
+        return LogicalTypeAnnotation.decimalType(scale, precision);
+
+        // DATE
+      case "DateLogicalTypeAnnotation":
+        return LogicalTypeAnnotation.dateType();
+
+        // TIME
+      case "TimeLogicalTypeAnnotation":
+        boolean isUTC = Boolean.parseBoolean(params.getOrDefault("isAdjustedToUTC", "true"));
+        String timeUnitStr = params.getOrDefault("unit", "MICROS");
+
+        LogicalTypeAnnotation.TimeUnit timeUnit;
+        switch (timeUnitStr) {
+          case "MILLIS":
+            timeUnit = LogicalTypeAnnotation.TimeUnit.MILLIS;
+            break;
+          case "MICROS":
+            timeUnit = LogicalTypeAnnotation.TimeUnit.MICROS;
+            break;
+          case "NANOS":
+            timeUnit = LogicalTypeAnnotation.TimeUnit.NANOS;
+            break;
+          default:
+            timeUnit = LogicalTypeAnnotation.TimeUnit.MICROS;
+        }
+        return LogicalTypeAnnotation.timeType(isUTC, timeUnit);
+
+        // TIMESTAMP
+      case "TimestampLogicalTypeAnnotation":
+        boolean isAdjustedToUTC = Boolean.parseBoolean(params.get("isAdjustedToUTC"));
+        String unitStr = params.getOrDefault("unit", "MICROS");
+
+        LogicalTypeAnnotation.TimeUnit unit;
+        switch (unitStr) {
+          case "MILLIS":
+            unit = LogicalTypeAnnotation.TimeUnit.MILLIS;
+            break;
+          case "MICROS":
+            unit = LogicalTypeAnnotation.TimeUnit.MICROS;
+            break;
+          case "NANOS":
+            unit = LogicalTypeAnnotation.TimeUnit.NANOS;
+            break;
+          default:
+            unit = LogicalTypeAnnotation.TimeUnit.MICROS;
+        }
+        return LogicalTypeAnnotation.timestampType(isAdjustedToUTC, unit);
+
+        // INTEGER
+      case "IntLogicalTypeAnnotation":
+        boolean isSigned = Boolean.parseBoolean(params.get("isSigned"));
+        int bitWidth = Integer.parseInt(params.get("bitWidth"));
+        return LogicalTypeAnnotation.intType(bitWidth, isSigned);
+
+        // JSON
+      case "JsonLogicalTypeAnnotation":
+        return LogicalTypeAnnotation.jsonType();
+
+        // BSON
+      case "BsonLogicalTypeAnnotation":
+        return LogicalTypeAnnotation.bsonType();
+
+        // UUID
+      case "UUIDLogicalTypeAnnotation":
+        return LogicalTypeAnnotation.uuidType();
+
+        // INTERVAL
+      case "IntervalLogicalTypeAnnotation":
+        return LogicalTypeAnnotation.IntervalLogicalTypeAnnotation.getInstance();
+
+      default:
+        throw new IllegalArgumentException("Unknown logical type: " + logicalTypeName);
+    }
+  }
 }