[#2824] bugFix: Fixed abnormal timestamp type conversion between iceberg and gravitino

catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/converter/FromIcebergType.java

@@ -80,9 +80,9 @@ public Type primitive(org.apache.iceberg.types.Type.PrimitiveType primitive) {
       case TIMESTAMP:
         Types.TimestampType ts = (Types.TimestampType) primitive;
         if (ts.shouldAdjustToUTC()) {
-          return com.datastrato.gravitino.rel.types.Types.TimestampType.withoutTimeZone();
-        } else {
           return com.datastrato.gravitino.rel.types.Types.TimestampType.withTimeZone();
+        } else {
+          return com.datastrato.gravitino.rel.types.Types.TimestampType.withoutTimeZone();
         }
       case STRING:
         return com.datastrato.gravitino.rel.types.Types.StringType.get();

catalogs/catalog-lakehouse-iceberg/src/test/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/converter/TestConvertUtil.java

@@ -278,12 +278,20 @@ public void testFormIcebergType() {
     Assertions.assertTrue(
         ConvertUtil.formIcebergType(Types.TimeType.get())
             instanceof com.datastrato.gravitino.rel.types.Types.TimeType);
+    com.datastrato.gravitino.rel.types.Type TimestampTypeWithoutZone =
+        ConvertUtil.formIcebergType(Types.TimestampType.withoutZone());
     Assertions.assertTrue(
-        ConvertUtil.formIcebergType(Types.TimestampType.withoutZone())
-            instanceof com.datastrato.gravitino.rel.types.Types.TimestampType);
+        TimestampTypeWithoutZone instanceof com.datastrato.gravitino.rel.types.Types.TimestampType);
+    Assertions.assertFalse(
+        ((com.datastrato.gravitino.rel.types.Types.TimestampType) TimestampTypeWithoutZone)
+            .hasTimeZone());
+    com.datastrato.gravitino.rel.types.Type TimestampTypeWithZone =
+        ConvertUtil.formIcebergType(Types.TimestampType.withZone());
     Assertions.assertTrue(
-        ConvertUtil.formIcebergType(Types.TimestampType.withZone())
-            instanceof com.datastrato.gravitino.rel.types.Types.TimestampType);
+        TimestampTypeWithZone instanceof com.datastrato.gravitino.rel.types.Types.TimestampType);
+    Assertions.assertTrue(
+        ((com.datastrato.gravitino.rel.types.Types.TimestampType) TimestampTypeWithZone)
+            .hasTimeZone());
     Assertions.assertTrue(
         ConvertUtil.formIcebergType(Types.DoubleType.get())
             instanceof com.datastrato.gravitino.rel.types.Types.DoubleType);

catalogs/catalog-lakehouse-iceberg/src/test/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/integration/test/CatalogIcebergIT.java

@@ -439,6 +439,61 @@ void testCreateAndLoadIcebergTable() {
                     sortOrders));
   }
 
+  @Test
+  void testTimestampTypeConversion() {
+
+    Column col1 =
+        Column.of("iceberg_column_1", Types.TimestampType.withTimeZone(), "col_1_comment");
+    Column col2 =
+        Column.of("iceberg_column_2", Types.TimestampType.withoutTimeZone(), "col_2_comment");
+
+    Column[] columns = new Column[] {col1, col2};
+
+    String timestampTableName = "timestamp_table";
+
+    NameIdentifier tableIdentifier =
+        NameIdentifier.of(metalakeName, catalogName, schemaName, timestampTableName);
+
+    Map<String, String> properties = createProperties();
+    TableCatalog tableCatalog = catalog.asTableCatalog();
+    Table createdTable =
+        tableCatalog.createTable(tableIdentifier, columns, table_comment, properties);
+    Assertions.assertEquals("iceberg_column_1", createdTable.columns()[0].name());
+    Assertions.assertEquals(
+        Types.TimestampType.withTimeZone(), createdTable.columns()[0].dataType());
+    Assertions.assertEquals("col_1_comment", createdTable.columns()[0].comment());
+
+    Assertions.assertEquals("iceberg_column_2", createdTable.columns()[1].name());
+    Assertions.assertEquals(
+        Types.TimestampType.withoutTimeZone(), createdTable.columns()[1].dataType());
+    Assertions.assertEquals("col_2_comment", createdTable.columns()[1].comment());
+
+    Table loadTable = tableCatalog.loadTable(tableIdentifier);
+    Assertions.assertEquals("iceberg_column_1", loadTable.columns()[0].name());
+    Assertions.assertEquals(Types.TimestampType.withTimeZone(), loadTable.columns()[0].dataType());
+    Assertions.assertEquals("col_1_comment", loadTable.columns()[0].comment());
+
+    Assertions.assertEquals("iceberg_column_2", loadTable.columns()[1].name());
+    Assertions.assertEquals(
+        Types.TimestampType.withoutTimeZone(), loadTable.columns()[1].dataType());
+    Assertions.assertEquals("col_2_comment", loadTable.columns()[1].comment());
+
+    org.apache.iceberg.Table table =
+        hiveCatalog.loadTable(IcebergTableOpsHelper.buildIcebergTableIdentifier(tableIdentifier));
+    org.apache.iceberg.Schema icebergSchema = table.schema();
+    Assertions.assertEquals("iceberg_column_1", icebergSchema.columns().get(0).name());
+    Assertions.assertEquals(
+        org.apache.iceberg.types.Types.TimestampType.withZone(),
+        icebergSchema.columns().get(0).type());
+    Assertions.assertEquals("col_1_comment", icebergSchema.columns().get(0).doc());
+
+    Assertions.assertEquals("iceberg_column_2", icebergSchema.columns().get(1).name());
+    Assertions.assertEquals(
+        org.apache.iceberg.types.Types.TimestampType.withoutZone(),
+        icebergSchema.columns().get(1).type());
+    Assertions.assertEquals("col_2_comment", icebergSchema.columns().get(1).doc());
+  }
+
   @Test
   void testListAndDropIcebergTable() {
     Column[] columns = createColumns();