From e19b312823a69ae0c0f3c47dcf86df060b9ed92d Mon Sep 17 00:00:00 2001
From: buom <39004+buom@users.noreply.github.com>
Date: Fri, 10 Mar 2023 13:28:51 +0700
Subject: [PATCH] Support logical types  (parquet-hive)

Added support to create DATE, TIMESTAMP and INT values from KafkaConnect Date, Timestamp and Time respectively.
---
 .../io/confluent/connect/hdfs/parquet/ParquetHiveUtil.java    | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/main/java/io/confluent/connect/hdfs/parquet/ParquetHiveUtil.java b/src/main/java/io/confluent/connect/hdfs/parquet/ParquetHiveUtil.java
index 519adf433..ffab71d10 100644
--- a/src/main/java/io/confluent/connect/hdfs/parquet/ParquetHiveUtil.java
+++ b/src/main/java/io/confluent/connect/hdfs/parquet/ParquetHiveUtil.java
@@ -56,7 +56,7 @@ public void createTable(
   @Override
   public void alterSchema(String database, String tableName, Schema schema) {
     Table table = hiveMetaStore.getTable(database, tableName);
-    List<FieldSchema> columns = HiveSchemaConverter.convertSchema(schema);
+    List<FieldSchema> columns = HiveSchemaConverter.convertSchemaMaybeLogical(schema);
     removeFieldPartitionColumn(columns, table.getPartitionKeys());
     table.setFields(columns);
     hiveMetaStore.alterTable(table);
@@ -83,7 +83,7 @@ private Table constructParquetTable(
       throw new HiveMetaStoreException("Cannot find input/output format:", e);
     }
     // convert Connect schema schema to Hive columns
-    List<FieldSchema> columns = HiveSchemaConverter.convertSchema(schema);
+    List<FieldSchema> columns = HiveSchemaConverter.convertSchemaMaybeLogical(schema);
     removeFieldPartitionColumn(columns, partitioner.partitionFields());
     table.setFields(columns);
     table.setPartCols(partitioner.partitionFields());