edited ParquetSchemaExtractor to include some other LogicalTypes and …

…added YAML reader for partition fields user configuration

xtable-api/src/main/java/org/apache/xtable/model/schema/InternalType.java

@@ -52,11 +52,7 @@ public enum InternalType {
   DECIMAL,
   TIMESTAMP,
   TIMESTAMP_NTZ,
-  JSON,
-  BSON,
-  VARIANT,
-  GEOMETRY,
-  GEOGRAPHY;
+  JSON;
   private final String name;
 
   InternalType() {

xtable-core/src/main/java/org/apache/xtable/parquet/ParquetPartitionExtractor.java

@@ -31,12 +31,35 @@
 import org.apache.xtable.model.stat.Range;
 import org.apache.xtable.schema.SchemaFieldFinder;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
 public class ParquetPartitionExtractor {
   private static final ParquetPartitionExtractor INSTANCE = new ParquetPartitionExtractor();
 
   public static ParquetPartitionExtractor getInstance() {
     return INSTANCE;
   }
+  public static final ObjectMapper YAML_MAPPER = new ObjectMapper(new YAMLFactory());
+
+
+  public PartitionConfiguration getPartitionsFromUserConfiguration(String configPath) throws IOException {
+    PartitionConfiguration partitionConfiguration = new PartitionConfiguration();
+    try (InputStream inputStream = Files.newInputStream(Paths.get(configPath))) {
+      ObjectReader objectReader = YAML_MAPPER.readerForUpdating(partitionConfiguration);
+      objectReader.readValue(inputStream);
+      return partitionConfiguration;
+    }
+  }
+
 
   public List<InternalPartitionField> getInternalPartitionField(
       Set<String> partitionList, InternalSchema schema) {

xtable-core/src/main/java/org/apache/xtable/parquet/ParquetSchemaExtractror.java

@@ -34,6 +34,12 @@
 import org.apache.parquet.LogicalType;
 import org.apache.parquet.LogicalTypes;
 import org.apache.parquet.Schema;
+import org.apache.parquet.Schema.Type;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.format.NullType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+
+
 
 import org.apache.xtable.collectors.CustomCollectors;
 import org.apache.xtable.exception.SchemaExtractorException;
@@ -96,38 +102,49 @@ public InternalSchema toInternalSchema(Schema schema) {
    * @return a converted schema
    */
   private InternalSchema toInternalSchema(
-      Schema schema, String parentPath, Map<String, IdMapping> fieldNameToIdMapping) {
+      Type schema, String parentPath, Map<String, IdMapping> fieldNameToIdMapping) {
     // TODO - Does not handle recursion in parquet schema
     InternalType newDataType;
     Map<InternalSchema.MetadataKey, Object> metadata = new HashMap<>();
-    switch (schema.getType()) {
-      case INT:
-        LogicalType logicalType = schema.getLogicalType();
-        if (logicalType instanceof LogicalTypes.Date) {
+    switch (schema.getName()) {
+      case INT64:
+        LogicalType logicalType = schema.getLogicalTypeAnnotation();
+        if (logicalType instanceof LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) {
+          newDataType = InternalType.TIMESTAMP;
+        }
+        break;
+      case INT32:
+        LogicalType logicalType = schema.getLogicalTypeAnnotation();
+        if (logicalType instanceof LogicalTypeAnnotation.DateLogicalTypeAnnotation) {
           newDataType = InternalType.DATE;
-        } else {
-          newDataType = InternalType.INT;
         }
+        // is also a TIME
         break;
-      case STRING:
-        newDataType = InternalType.STRING;
+      case FLOAT:
+        LogicalType logicalType = schema.getLogicalTypeAnnotation();
+        if (logicalType instanceof LogicalTypeAnnotation.Float16LogicalTypeAnnotation) {
+          newDataType = InternalType.FLOAT;
+        }
         break;
-      case BOOLEAN:
-        newDataType = InternalType.BOOLEAN;
+      case BYTES:
+        logicalType = schema.getLogicalType();
         break;
+      case FIXED_LEN_BYTE_ARRAY:
+        logicalType=schema.getLogicalTypeAnnotation()
+        if (logicalType instanceof LogicalTypeAnnotation.UUIDLogicalTypeAnnotation) {
+          newDataType = InternalType.UUID;
+        }
+       //TODO how to add  INTERVAL ?
       case BYTE_ARRAY:
-        logicalType = schema.getLogicalType();
-        // TODO: any metadata to add ?
+        // includes string json,BSON, Variant,GEOMETRY, geography,
+        logicalType=schema.getLogicalTypeAnnotation()
+        if (logicalType instanceof LogicalTypeAnnotation.EnumLogicalTypeAnnotation) {
+          newDataType = InternalType.ENUM;
+        }
         if (logicalType == LogicalTypes.JSON) {
-          newDataType = InternalType.JSON;
-        } else if (logicalType instanceof LogicalTypes.BSON) {
-          newDataType = InternalType.BSON;
-        } else if (logicalType instanceof LogicalTypes.VARIANT) {
-          newDataType = InternalType.VARIANT;
-        } else if (logicalType instanceof LogicalTypes.GEOMETRY) {
-          newDataType = InternalType.GEOMETRY;
-        } else if (logicalType instanceof LogicalTypes.GEOGRAPHY) {
-          newDataType = InternalType.GEOGRAPHY;
+          newDataType = InternalType.BYTES;
+        } else if (logicalType instanceof LogicalTypeAnnotation.BsonLogicalTypeAnnotation) {
+          newDataType = InternalType.BYTES;
         }
         break;
       case FIXED:
@@ -157,12 +174,6 @@ private InternalSchema toInternalSchema(
           newDataType = InternalType.BYTES;
         }
         break;
-      case DOUBLE:
-        newDataType = InternalType.DOUBLE;
-        break;
-      case FLOAT:
-        newDataType = InternalType.FLOAT;
-        break;
       case LONG:
         logicalType = schema.getLogicalType();
         if (logicalType instanceof LogicalTypes.TimestampMillis) {
@@ -185,10 +196,6 @@ private InternalSchema toInternalSchema(
           newDataType = InternalType.LONG;
         }
         break;
-      case ENUM:
-        metadata.put(InternalSchema.MetadataKey.ENUM_VALUES, schema.getEnumSymbols());
-        newDataType = InternalType.ENUM;
-        break;
       case NULL:
         newDataType = InternalType.NULL;
         break;
@@ -473,8 +480,9 @@ private String buildCurrentPath(InternalField field, String parentPath) {
         .orElse(field.getName());
   }
 
-  private static Schema finalizeSchema(Schema targetSchema, InternalSchema inputSchema) {
+  private static Type finalizeSchema(Type targetSchema, InternalSchema inputSchema) {
     if (inputSchema.isNullable()) {
+      return targetSchema.union(Type("NULL","optional"))
       return Schema.createUnion(Schema.create(Schema.Type.NULL), targetSchema);
     }
     return targetSchema;

xtable-core/src/main/java/org/apache/xtable/parquet/ParquetTableExtractor.java

@@ -54,6 +54,7 @@ public InternalTable table(String tableName, Long version) {
     MessageType schema = parquetMetadataExtractor.getSchema(footer);
     InternalSchema schema = schemaExtractor.toInternalSchema(schema);
     Set<String> partitionKeys = initPartitionInfo().keySet();
+    // TODO getInternalPartitionField() to be replaced with the new YAML based partition fields
     List<InternalPartitionField> partitionFields =
         partitionExtractor.getInternalPartitionField(partitionKeys, schema);
     DataLayoutStrategy dataLayoutStrategy =

xtable-core/src/main/java/org/apache/xtable/parquet/PartitionConfiguration.java

@@ -0,0 +1,21 @@
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class PartitionConfiguration {
+    public String partition;
+
+
+    public PartitionConfiguration(String partition) {
+        this.partition = partition;
+    }
+
+    public PartitionConfiguration() {
+
+    }
+
+    public String getPartition() {
+        return this.partition;
+    }
+
+}