Iceberg source schema

api/src/main/java/io/onetable/model/schema/OneType.java

@@ -35,7 +35,7 @@
 public enum OneType {
   RECORD,
   ENUM,
-  ARRAY,
+  LIST,
   MAP,
   UNION,
   FIXED,
@@ -62,7 +62,7 @@ public enum OneType {
           new HashSet<OneType>() {
             {
               add(RECORD);
-              add(ARRAY);
+              add(LIST);
               add(MAP);
               add(UNION);
             }

core/src/main/java/io/onetable/avro/AvroSchemaConverter.java

@@ -18,6 +18,8 @@
 
 package io.onetable.avro;
 
+import static io.onetable.schema.SchemaUtils.getFullyQualifiedPath;
+
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -207,7 +209,7 @@ private OneSchema toOneSchema(
                 .build();
         return OneSchema.builder()
             .name(schema.getName())
-            .dataType(OneType.ARRAY)
+            .dataType(OneType.LIST)
             .comment(schema.getDoc())
             .isNullable(schema.isNullable())
             .fields(Collections.singletonList(elementField))
@@ -288,13 +290,6 @@ private static Object getDefaultValue(Schema.Field avroField) {
         : avroField.defaultVal();
   }
 
-  private static String getFullyQualifiedPath(String path, String fieldName) {
-    if (path == null || path.isEmpty()) {
-      return fieldName;
-    }
-    return path + "." + fieldName;
-  }
-
   public Schema fromOneSchema(OneSchema oneSchema) {
     switch (oneSchema.getDataType()) {
       case RECORD:
@@ -361,7 +356,7 @@ public Schema fromOneSchema(OneSchema oneSchema) {
               LogicalTypes.localTimestampMillis().addToSchema(Schema.create(Schema.Type.LONG)),
               oneSchema);
         }
-      case ARRAY:
+      case LIST:
         OneSchema elementSchema =
             oneSchema.getFields().stream()
                 .filter(

core/src/main/java/io/onetable/delta/DeltaSchemaExtractor.java

@@ -18,6 +18,8 @@
 
 package io.onetable.delta;
 
+import static io.onetable.schema.SchemaUtils.getFullyQualifiedPath;
+
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -122,7 +124,7 @@ private DataType convertFieldType(OneField field) {
                 .orElseThrow(() -> new SchemaExtractorException("Invalid map schema"));
         return DataTypes.createMapType(
             convertFieldType(key), convertFieldType(value), value.getSchema().isNullable());
-      case ARRAY:
+      case LIST:
         OneField element =
             field.getSchema().getFields().stream()
                 .filter(
@@ -229,7 +231,7 @@ private OneSchema toOneSchema(
                 .parentPath(parentPath)
                 .schema(elementSchema)
                 .build();
-        type = OneType.ARRAY;
+        type = OneType.LIST;
         fields = Collections.singletonList(elementField);
         break;
       case "map":
@@ -273,11 +275,4 @@ private OneSchema toOneSchema(
         .fields(fields)
         .build();
   }
-
-  private static String getFullyQualifiedPath(String path, String fieldName) {
-    if (path == null || path.isEmpty()) {
-      return fieldName;
-    }
-    return path + "." + fieldName;
-  }
 }

core/src/main/java/io/onetable/iceberg/IcebergSchemaExtractor.java

@@ -18,9 +18,14 @@
 
 package io.onetable.iceberg;
 
+import static io.onetable.schema.SchemaUtils.getFullyQualifiedPath;
+
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -38,7 +43,6 @@
 import io.onetable.exception.SchemaExtractorException;
 import io.onetable.model.schema.OneField;
 import io.onetable.model.schema.OneSchema;
-import io.onetable.model.schema.OneSchema.OneSchemaBuilder;
 import io.onetable.model.schema.OneType;
 
 /**
@@ -93,36 +97,31 @@ public Schema toIceberg(OneSchema oneSchema) {
    * @return Internal representation of deserialized iceberg schema
    */
   public OneSchema fromIceberg(Schema iceSchema) {
-    OneSchemaBuilder irSchemaBuilder = OneSchema.builder();
-
-    List<OneField> irFields = new ArrayList<>();
-    for (Types.NestedField iceField : iceSchema.columns()) {
-      OneType irFieldType = fromIcebergType(iceField.type());
-      OneSchema irFieldSchema =
-          OneSchema.builder()
-              .dataType(irFieldType)
-              .isNullable(iceField.isOptional())
-              .metadata(new HashMap<>())
-              .build();
-      if (irFieldType == OneType.DECIMAL) {
-        Types.DecimalType decimalType = (Types.DecimalType) iceField.type();
-        irFieldSchema.getMetadata().put(OneSchema.MetadataKey.DECIMAL_SCALE, decimalType.scale());
-        irFieldSchema
-            .getMetadata()
-            .put(OneSchema.MetadataKey.DECIMAL_PRECISION, decimalType.precision());
-      }
-
-      OneField irField =
-          OneField.builder()
-              .name(iceField.name())
-              .fieldId(iceField.fieldId())
-              .schema(irFieldSchema)
-              .build();
-      irFields.add(irField);
-    }
-    irSchemaBuilder.fields(irFields);
+    return OneSchema.builder()
+        .dataType(OneType.RECORD)
+        .fields(fromIceberg(iceSchema.columns(), null))
+        .build();
+  }
 
-    return irSchemaBuilder.build();
+  private List<OneField> fromIceberg(List<Types.NestedField> iceFields, String parentPath) {
+    return iceFields.stream()
+        .map(
+            iceField -> {
+              Type type = iceField.type();
+              String doc = iceField.doc();
+              boolean isOptional = iceField.isOptional();
+              String fieldPath = getFullyQualifiedPath(parentPath, iceField.name());
+              OneSchema irFieldSchema = fromIcebergType(type, fieldPath, doc, isOptional);
+              return OneField.builder()
+                  .name(iceField.name())
+                  .fieldId(iceField.fieldId())
+                  .schema(irFieldSchema)
+                  .parentPath(parentPath)
+                  .defaultValue(
+                      iceField.isOptional() ? OneField.Constants.NULL_DEFAULT_VALUE : null)
+                  .build();
+            })
+        .collect(Collectors.toList());
   }
 
   static String convertFromOneTablePath(String path) {
@@ -163,7 +162,7 @@ private Type toIcebergType(OneField field, AtomicInteger fieldIdTracker) {
       case INT:
         return Types.IntegerType.get();
       case LONG:
-      case TIMESTAMP_NTZ:
+      case TIMESTAMP_NTZ: // TODO - revisit this
         return Types.LongType.get();
       case BYTES:
         return Types.BinaryType.get();
@@ -217,7 +216,7 @@ private Type toIcebergType(OneField field, AtomicInteger fieldIdTracker) {
               toIcebergType(key, fieldIdTracker),
               toIcebergType(value, fieldIdTracker));
         }
-      case ARRAY:
+      case LIST:
         OneField element =
             field.getSchema().getFields().stream()
                 .filter(
@@ -237,47 +236,124 @@ private Type toIcebergType(OneField field, AtomicInteger fieldIdTracker) {
     }
   }
 
-  /**
-   * Maps Iceberg type to internal type representation
-   *
-   * @param type Iceberg type
-   * @return Internal representation of Iceberg type
-   */
-  OneType fromIcebergType(Type type) {
-    switch (type.typeId()) {
-        // TODO ENUM type is not supported in Iceberg
+  private OneSchema fromIcebergType(
+      Type iceType, String fieldPath, String doc, boolean isOptional) {
+    OneType type;
+    List<OneField> fields = null;
+    Map<OneSchema.MetadataKey, Object> metadata = null;
+    switch (iceType.typeId()) {
       case STRING:
-        return OneType.STRING;
+        type = OneType.STRING;
+        break;
       case INTEGER:
-        return OneType.INT;
+        type = OneType.INT;
+        break;
       case LONG:
-        return OneType.LONG;
+        type = OneType.LONG;
+        break;
       case BINARY:
-        return OneType.BYTES;
-        // TODO
-        //      case FIXED:
-        //        return OneType.FIXED;
+        type = OneType.BYTES;
+        break;
       case BOOLEAN:
-        return OneType.BOOLEAN;
+        type = OneType.BOOLEAN;
+        break;
       case FLOAT:
-        return OneType.FLOAT;
+        type = OneType.FLOAT;
+        break;
       case DATE:
-        return OneType.DATE;
+        type = OneType.DATE;
+        break;
       case TIMESTAMP:
-        return OneType.TIMESTAMP;
+        Types.TimestampType timestampType = (Types.TimestampType) iceType;
+        type = timestampType.shouldAdjustToUTC() ? OneType.TIMESTAMP : OneType.TIMESTAMP_NTZ;
+        metadata =
+            Collections.singletonMap(
+                OneSchema.MetadataKey.TIMESTAMP_PRECISION, OneSchema.MetadataValue.MICROS);
+        break;
       case DOUBLE:
-        return OneType.DOUBLE;
+        type = OneType.DOUBLE;
+        break;
       case DECIMAL:
-        return OneType.DECIMAL;
-        // TODO
-        //      case STRUCT:
-        //        return OneType.RECORD;
-        //      case MAP:
-        //        return OneType.MAP;
-        //      case LIST:
-        //        return OneType.ARRAY;
+        Types.DecimalType decimalType = (Types.DecimalType) iceType;
+        metadata = new HashMap<>(2, 1.0f);
+        metadata.put(OneSchema.MetadataKey.DECIMAL_PRECISION, decimalType.precision());
+        metadata.put(OneSchema.MetadataKey.DECIMAL_SCALE, decimalType.scale());
+        type = OneType.DECIMAL;
+        break;
+      case FIXED:
+        type = OneType.FIXED;
+        Types.FixedType fixedType = (Types.FixedType) iceType;
+        metadata =
+            Collections.singletonMap(OneSchema.MetadataKey.FIXED_BYTES_SIZE, fixedType.length());
+        break;
+      case UUID:
+        type = OneType.FIXED;
+        metadata = Collections.singletonMap(OneSchema.MetadataKey.FIXED_BYTES_SIZE, 16);
+        break;
+      case STRUCT:
+        Types.StructType structType = (Types.StructType) iceType;
+        fields = fromIceberg(structType.fields(), fieldPath);
+        type = OneType.RECORD;
+        break;
+      case MAP:
+        Types.MapType mapType = (Types.MapType) iceType;
+        OneSchema keySchema =
+            fromIcebergType(
+                mapType.keyType(),
+                getFullyQualifiedPath(fieldPath, OneField.Constants.MAP_KEY_FIELD_NAME),
+                null,
+                false);
+        OneField keyField =
+            OneField.builder()
+                .name(OneField.Constants.MAP_KEY_FIELD_NAME)
+                .parentPath(fieldPath)
+                .schema(keySchema)
+                .fieldId(mapType.keyId())
+                .build();
+        OneSchema valueSchema =
+            fromIcebergType(
+                mapType.valueType(),
+                getFullyQualifiedPath(fieldPath, OneField.Constants.MAP_VALUE_FIELD_NAME),
+                null,
+                mapType.isValueOptional());
+        OneField valueField =
+            OneField.builder()
+                .name(OneField.Constants.MAP_VALUE_FIELD_NAME)
+                .parentPath(fieldPath)
+                .schema(valueSchema)
+                .fieldId(mapType.valueId())
+                .build();
+        type = OneType.MAP;
+        fields = Arrays.asList(keyField, valueField);
+        break;
+      case LIST:
+        Types.ListType listType = (Types.ListType) iceType;
+        OneSchema elementSchema =
+            fromIcebergType(
+                listType.elementType(),
+                getFullyQualifiedPath(fieldPath, OneField.Constants.ARRAY_ELEMENT_FIELD_NAME),
+                null,
+                listType.isElementOptional());
+        OneField elementField =
+            OneField.builder()
+                .name(OneField.Constants.ARRAY_ELEMENT_FIELD_NAME)
+                .parentPath(fieldPath)
+                .schema(elementSchema)
+                .fieldId(listType.elementId())
+                .build();
+        type = OneType.LIST;
+        fields = Collections.singletonList(elementField);
+        break;
       default:
-        throw new NotSupportedException("Unsupported type: " + type.typeId());
+        throw new NotSupportedException("Unsupported type: " + iceType.typeId());
     }
+    return OneSchema.builder()
+        .name(iceType.typeId().name().toLowerCase())
+        .dataType(type)
+        .comment(doc)
+        .isNullable(isOptional)
+        .metadata(metadata)
+        .fields(fields)
+        .build();
   }
 }

core/src/main/java/io/onetable/schema/SchemaUtils.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.onetable.schema;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class SchemaUtils {
+  public static String getFullyQualifiedPath(String path, String fieldName) {
+    if (path == null || path.isEmpty()) {
+      return fieldName;
+    }
+    return path + "." + fieldName;
+  }
+}