feat: Add ability to read and process complex/nested data types from a parquet file in Parquet Data Source: enums, repeated enums, message, repeated message, repeated primitives

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/complex/MessageHandler.java

@@ -1,5 +1,7 @@
 package io.odpf.dagger.common.serde.typehandler.complex;
 
+import com.google.protobuf.Descriptors;
+import io.odpf.dagger.common.serde.parquet.SimpleGroupValidation;
 import io.odpf.dagger.common.serde.typehandler.TypeHandler;
 import io.odpf.dagger.common.serde.typehandler.TypeHandlerFactory;
 import io.odpf.dagger.common.serde.typehandler.RowFactory;
@@ -24,6 +26,8 @@
 public class MessageHandler implements TypeHandler {
     private FieldDescriptor fieldDescriptor;
     private JsonRowSerializationSchema jsonRowSerializationSchema;
+    private DynamicMessage defaultMessageInstance;
+    private Descriptors.Descriptor fieldMessageDescriptor;
 
     /**
      * Instantiates a new Message proto handler.
@@ -32,6 +36,10 @@ public class MessageHandler implements TypeHandler {
      */
     public MessageHandler(FieldDescriptor fieldDescriptor) {
         this.fieldDescriptor = fieldDescriptor;
+        if (canHandle()) {
+            this.defaultMessageInstance = DynamicMessage.getDefaultInstance(fieldDescriptor.getMessageType());
+            this.fieldMessageDescriptor = fieldDescriptor.getMessageType();
+        }
     }
 
     @Override
@@ -74,7 +82,12 @@ public Object transformFromProto(Object field) {
 
     @Override
     public Object transformFromParquet(SimpleGroup simpleGroup) {
-        return null;
+        String fieldName = fieldDescriptor.getName();
+        if (simpleGroup != null && SimpleGroupValidation.checkFieldExistsAndIsInitialized(simpleGroup, fieldName)) {
+            SimpleGroup nestedGroup = (SimpleGroup) simpleGroup.getGroup(fieldName, 0);
+            return RowFactory.createRow(fieldMessageDescriptor, nestedGroup);
+        }
+        return RowFactory.createRow(defaultMessageInstance);
     }
 
     @Override

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/primitive/BooleanHandler.java

@@ -49,14 +49,28 @@ public Object parseSimpleGroup(SimpleGroup simpleGroup) {
     }
 
     @Override
-    public Object getArray(Object field) {
+    public Object parseRepeatedObjectField(Object field) {
         boolean[] inputValues = new boolean[0];
         if (field != null) {
             inputValues = Booleans.toArray((List<Boolean>) field);
         }
         return inputValues;
     }
 
+    @Override
+    public Object parseRepeatedSimpleGroupField(SimpleGroup simpleGroup) {
+        String fieldName = fieldDescriptor.getName();
+        if (simpleGroup != null && SimpleGroupValidation.checkFieldExistsAndIsInitialized(simpleGroup, fieldName)) {
+            int repetitionCount = simpleGroup.getFieldRepetitionCount(fieldName);
+            boolean[] booleanArray = new boolean[repetitionCount];
+            for (int i = 0; i < repetitionCount; i++) {
+                booleanArray[i] = simpleGroup.getBoolean(fieldName, i);
+            }
+            return booleanArray;
+        }
+        return new boolean[0];
+    }
+
     @Override
     public TypeInformation getTypeInformation() {
         return Types.BOOLEAN;

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/primitive/ByteStringHandler.java

@@ -51,14 +51,28 @@ public Object parseSimpleGroup(SimpleGroup simpleGroup) {
     }
 
     @Override
-    public Object getArray(Object field) {
+    public Object parseRepeatedObjectField(Object field) {
         List<ByteString> inputValues = new ArrayList<>();
         if (field != null) {
             inputValues = (List<ByteString>) field;
         }
         return inputValues.toArray(new ByteString[]{});
     }
 
+    @Override
+    public Object parseRepeatedSimpleGroupField(SimpleGroup simpleGroup) {
+        String fieldName = fieldDescriptor.getName();
+        ArrayList<ByteString> byteStringList = new ArrayList<>();
+        if (simpleGroup != null && SimpleGroupValidation.checkFieldExistsAndIsInitialized(simpleGroup, fieldName)) {
+            int repetitionCount = simpleGroup.getFieldRepetitionCount(fieldName);
+            for (int i = 0; i < repetitionCount; i++) {
+                byte[] byteArray = simpleGroup.getBinary(fieldName, i).getBytes();
+                byteStringList.add(ByteString.copyFrom(byteArray));
+            }
+        }
+        return byteStringList.toArray(new ByteString[]{});
+    }
+
     @Override
     public TypeInformation getTypeInformation() {
         return TypeInformation.of(ByteString.class);

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/primitive/DoubleHandler.java

@@ -49,14 +49,28 @@ public Object parseSimpleGroup(SimpleGroup simpleGroup) {
     }
 
     @Override
-    public Object getArray(Object field) {
+    public Object parseRepeatedObjectField(Object field) {
         double[] inputValues = new double[0];
         if (field != null) {
             inputValues = Doubles.toArray((List<Double>) field);
         }
         return inputValues;
     }
 
+    @Override
+    public Object parseRepeatedSimpleGroupField(SimpleGroup simpleGroup) {
+        String fieldName = fieldDescriptor.getName();
+        if (simpleGroup != null && SimpleGroupValidation.checkFieldExistsAndIsInitialized(simpleGroup, fieldName)) {
+            int repetitionCount = simpleGroup.getFieldRepetitionCount(fieldName);
+            double[] doubleArray = new double[repetitionCount];
+            for (int i = 0; i < repetitionCount; i++) {
+                doubleArray[i] = simpleGroup.getDouble(fieldName, i);
+            }
+            return doubleArray;
+        }
+        return new double[0];
+    }
+
     @Override
     public TypeInformation getTypeInformation() {
         return Types.DOUBLE;

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/primitive/FloatHandler.java

@@ -49,7 +49,7 @@ public Object parseSimpleGroup(SimpleGroup simpleGroup) {
     }
 
     @Override
-    public Object getArray(Object field) {
+    public Object parseRepeatedObjectField(Object field) {
 
         float[] inputValues = new float[0];
         if (field != null) {
@@ -58,6 +58,20 @@ public Object getArray(Object field) {
         return inputValues;
     }
 
+    @Override
+    public Object parseRepeatedSimpleGroupField(SimpleGroup simpleGroup) {
+        String fieldName = fieldDescriptor.getName();
+        if (simpleGroup != null && SimpleGroupValidation.checkFieldExistsAndIsInitialized(simpleGroup, fieldName)) {
+            int repetitionCount = simpleGroup.getFieldRepetitionCount(fieldName);
+            float[] floatArray = new float[repetitionCount];
+            for (int i = 0; i < repetitionCount; i++) {
+                floatArray[i] = simpleGroup.getFloat(fieldName, i);
+            }
+            return floatArray;
+        }
+        return new float[0];
+    }
+
     @Override
     public TypeInformation getTypeInformation() {
         return Types.FLOAT;

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/primitive/IntegerHandler.java

@@ -49,14 +49,28 @@ public Object parseSimpleGroup(SimpleGroup simpleGroup) {
     }
 
     @Override
-    public Object getArray(Object field) {
+    public Object parseRepeatedObjectField(Object field) {
         int[] inputValues = new int[0];
         if (field != null) {
             inputValues = Ints.toArray((List<Integer>) field);
         }
         return inputValues;
     }
 
+    @Override
+    public Object parseRepeatedSimpleGroupField(SimpleGroup simpleGroup) {
+        String fieldName = fieldDescriptor.getName();
+        if (simpleGroup != null && SimpleGroupValidation.checkFieldExistsAndIsInitialized(simpleGroup, fieldName)) {
+            int repetitionCount = simpleGroup.getFieldRepetitionCount(fieldName);
+            int[] intArray = new int[repetitionCount];
+            for (int i = 0; i < repetitionCount; i++) {
+                intArray[i] = simpleGroup.getInteger(fieldName, i);
+            }
+            return intArray;
+        }
+        return new int[0];
+    }
+
     @Override
     public TypeInformation getTypeInformation() {
         return Types.INT;

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/primitive/LongHandler.java

@@ -49,14 +49,27 @@ public Object parseSimpleGroup(SimpleGroup simpleGroup) {
     }
 
     @Override
-    public Object getArray(Object field) {
+    public Object parseRepeatedObjectField(Object field) {
         List<Long> inputValues = new ArrayList<>();
         if (field != null) {
             inputValues = (List<Long>) field;
         }
         return inputValues.toArray(new Long[]{});
     }
 
+    @Override
+    public Object parseRepeatedSimpleGroupField(SimpleGroup simpleGroup) {
+        String fieldName = fieldDescriptor.getName();
+        ArrayList<Long> deserializedValues = new ArrayList<>();
+        if (simpleGroup != null && SimpleGroupValidation.checkFieldExistsAndIsInitialized(simpleGroup, fieldName)) {
+            int repetitionCount = simpleGroup.getFieldRepetitionCount(fieldName);
+            for (int i = 0; i < repetitionCount; i++) {
+                deserializedValues.add(simpleGroup.getLong(fieldName, i));
+            }
+        }
+        return deserializedValues.toArray(new Long[]{});
+    }
+
     @Override
     public TypeInformation getTypeInformation() {
         return Types.LONG;

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/primitive/PrimitiveHandler.java

@@ -31,12 +31,21 @@ public interface PrimitiveHandler {
     Object parseSimpleGroup(SimpleGroup simpleGroup);
 
     /**
-     * Gets array.
+     * Parses a repeated object into an array of Java primitives.
      *
-     * @param field the field
-     * @return the array
+     * @param field the object array
+     * @return array of java primitive values as obtained from the repeated object
      */
-    Object getArray(Object field);
+    Object parseRepeatedObjectField(Object field);
+
+    /**
+     * Extracts the specific repeated primitive field from a SimpleGroup object and
+     * transforms it into an array of Java primitives.
+     *
+     * @param simpleGroup SimpleGroup object inside which the repeated primitive field resides
+     * @return array of java primitive values as obtained from the repeated primitive field
+     */
+    Object parseRepeatedSimpleGroupField(SimpleGroup simpleGroup);
 
     /**
      * Gets type information.

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/primitive/StringHandler.java

@@ -50,14 +50,28 @@ public Object parseSimpleGroup(SimpleGroup simpleGroup) {
     }
 
     @Override
-    public Object getArray(Object field) {
+    public Object parseRepeatedObjectField(Object field) {
         List<String> inputValues = new ArrayList<>();
         if (field != null) {
             inputValues = (List<String>) field;
         }
         return inputValues.toArray(new String[]{});
     }
 
+    @Override
+    public Object parseRepeatedSimpleGroupField(SimpleGroup simpleGroup) {
+        String fieldName = fieldDescriptor.getName();
+        if (simpleGroup != null && SimpleGroupValidation.checkFieldExistsAndIsInitialized(simpleGroup, fieldName)) {
+            int repetitionCount = simpleGroup.getFieldRepetitionCount(fieldName);
+            String[] stringArray = new String[repetitionCount];
+            for (int i = 0; i < repetitionCount; i++) {
+                stringArray[i] = simpleGroup.getString(fieldName, i);
+            }
+            return stringArray;
+        }
+        return new String[0];
+    }
+
     @Override
     public TypeInformation getTypeInformation() {
         return Types.STRING;

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/repeated/RepeatedEnumHandler.java

@@ -1,5 +1,6 @@
 package io.odpf.dagger.common.serde.typehandler.repeated;
 
+import io.odpf.dagger.common.serde.parquet.SimpleGroupValidation;
 import io.odpf.dagger.common.serde.typehandler.TypeHandler;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeinfo.Types;
@@ -54,7 +55,19 @@ public Object transformFromProto(Object field) {
 
     @Override
     public Object transformFromParquet(SimpleGroup simpleGroup) {
-        return null;
+        String defaultEnumValue = fieldDescriptor.getEnumType().findValueByNumber(0).getName();
+        List<String> deserializedEnumArray = new ArrayList<>();
+        String fieldName = fieldDescriptor.getName();
+        if (simpleGroup != null && SimpleGroupValidation.checkFieldExistsAndIsInitialized(simpleGroup, fieldName)) {
+            int repetitionCount = simpleGroup.getFieldRepetitionCount(fieldName);
+            for (int positionIndex = 0; positionIndex < repetitionCount; positionIndex++) {
+                String extractedValue = simpleGroup.getString(fieldName, positionIndex);
+                Descriptors.EnumValueDescriptor enumValueDescriptor = fieldDescriptor.getEnumType().findValueByName(extractedValue);
+                String enumValue = enumValueDescriptor == null ? defaultEnumValue : enumValueDescriptor.getName();
+                deserializedEnumArray.add(enumValue);
+            }
+        }
+        return deserializedEnumArray.toArray(new String[]{});
     }
 
     @Override

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/repeated/RepeatedMessageHandler.java

@@ -1,5 +1,7 @@
 package io.odpf.dagger.common.serde.typehandler.repeated;
 
+import com.google.protobuf.Descriptors;
+import io.odpf.dagger.common.serde.parquet.SimpleGroupValidation;
 import io.odpf.dagger.common.serde.typehandler.TypeHandler;
 import io.odpf.dagger.common.serde.typehandler.TypeHandlerFactory;
 import io.odpf.dagger.common.serde.typehandler.RowFactory;
@@ -28,6 +30,7 @@
 public class RepeatedMessageHandler implements TypeHandler {
     private JsonRowSerializationSchema jsonRowSerializationSchema;
     private FieldDescriptor fieldDescriptor;
+    private Descriptors.Descriptor fieldMessageDescriptor;
 
     /**
      * Instantiates a new Repeated message proto handler.
@@ -36,6 +39,9 @@ public class RepeatedMessageHandler implements TypeHandler {
      */
     public RepeatedMessageHandler(FieldDescriptor fieldDescriptor) {
         this.fieldDescriptor = fieldDescriptor;
+        if (canHandle()) {
+            this.fieldMessageDescriptor = fieldDescriptor.getMessageType();
+        }
     }
 
     @Override
@@ -90,7 +96,16 @@ public Object transformFromProto(Object field) {
 
     @Override
     public Object transformFromParquet(SimpleGroup simpleGroup) {
-        return null;
+        String fieldName = fieldDescriptor.getName();
+        ArrayList<Row> rowList = new ArrayList<>();
+        if (simpleGroup != null && SimpleGroupValidation.checkFieldExistsAndIsInitialized(simpleGroup, fieldName)) {
+            int repetitionCount = simpleGroup.getFieldRepetitionCount(fieldName);
+            for (int i = 0; i < repetitionCount; i++) {
+                SimpleGroup nestedGroup = (SimpleGroup) simpleGroup.getGroup(fieldName, i);
+                rowList.add(RowFactory.createRow(fieldMessageDescriptor, nestedGroup));
+            }
+        }
+        return rowList.toArray(new Row[]{});
     }
 
     @Override

dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/repeated/RepeatedPrimitiveHandler.java

@@ -66,12 +66,13 @@ public Object transformFromPostProcessor(Object field) {
     @Override
     public Object transformFromProto(Object field) {
         PrimitiveHandler primitiveHandler = PrimitiveHandlerFactory.getTypeHandler(fieldDescriptor);
-        return primitiveHandler.getArray(field);
+        return primitiveHandler.parseRepeatedObjectField(field);
     }
 
     @Override
     public Object transformFromParquet(SimpleGroup simpleGroup) {
-        return null;
+        PrimitiveHandler primitiveHandler = PrimitiveHandlerFactory.getTypeHandler(fieldDescriptor);
+        return primitiveHandler.parseRepeatedSimpleGroupField(simpleGroup);
     }
 
     @Override

dagger-common/src/test/java/io/odpf/dagger/common/serde/typehandler/PrimitiveTypeHandlerTest.java

@@ -94,7 +94,7 @@ public void shouldThrowInvalidDataTypeExceptionInCaseOfTypeMismatchForPostProces
     }
 
     @Test
-    public void shouldReturnSameValueForTransformFromKafka() {
+    public void shouldReturnSameValueForTransformFromProto() {
         Descriptors.Descriptor descriptor = TestBookingLogMessage.getDescriptor();
         Descriptors.FieldDescriptor stringFieldDescriptor = descriptor.findFieldByName("order_number");
         PrimitiveTypeHandler primitiveTypeHandler = new PrimitiveTypeHandler(stringFieldDescriptor);

dagger-common/src/test/java/io/odpf/dagger/common/serde/typehandler/complex/EnumHandlerTest.java

@@ -150,7 +150,7 @@ public void shouldReturnTypeInformation() {
     }
 
     @Test
-    public void shouldTransformValueForKafka() {
+    public void shouldTransformValueFromProto() {
         Descriptors.Descriptor descriptor = TestBookingLogMessage.getDescriptor();
         Descriptors.FieldDescriptor fieldDescriptor = descriptor.findFieldByName("status");
         EnumHandler enumHandler = new EnumHandler(fieldDescriptor);