refactor: rename packages and files in dagger-common serde

dagger-common/src/main/java/io/odpf/dagger/common/serde/parquet/deserialization/SimpleGroupDeserializer.java

@@ -8,7 +8,7 @@
 import io.odpf.dagger.common.serde.DaggerDeserializer;
 import io.odpf.dagger.common.serde.parquet.SimpleGroupValidation;
 import io.odpf.dagger.common.serde.proto.deserialization.ProtoType;
-import io.odpf.dagger.common.serde.proto.protohandler.RowFactory;
+import io.odpf.dagger.common.serde.typehandler.RowFactory;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.types.Row;
 import org.apache.parquet.example.data.simple.SimpleGroup;

dagger-common/src/main/java/io/odpf/dagger/common/serde/proto/deserialization/ProtoDeserializer.java

@@ -11,7 +11,7 @@
 import io.odpf.dagger.common.core.StencilClientOrchestrator;
 import io.odpf.dagger.common.exceptions.DescriptorNotFoundException;
 import io.odpf.dagger.common.exceptions.serde.DaggerDeserializationException;
-import io.odpf.dagger.common.serde.proto.protohandler.RowFactory;
+import io.odpf.dagger.common.serde.typehandler.RowFactory;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

dagger-common/src/main/java/io/odpf/dagger/common/serde/proto/deserialization/ProtoType.java

@@ -8,7 +8,7 @@
 import io.odpf.dagger.common.core.StencilClientOrchestrator;
 import io.odpf.dagger.common.exceptions.DescriptorNotFoundException;
 import io.odpf.dagger.common.serde.DaggerInternalTypeInformation;
-import io.odpf.dagger.common.serde.proto.protohandler.TypeInformationFactory;
+import io.odpf.dagger.common.serde.typehandler.TypeInformationFactory;
 
 import java.io.Serializable;
 

dagger-common/src/main/java/io/odpf/dagger/common/serde/proto/serialization/ProtoSerializer.java

@@ -10,8 +10,8 @@
 import io.odpf.dagger.common.exceptions.DescriptorNotFoundException;
 import io.odpf.dagger.common.exceptions.serde.DaggerSerializationException;
 import io.odpf.dagger.common.exceptions.serde.InvalidColumnMappingException;
-import io.odpf.dagger.common.serde.proto.protohandler.ProtoHandler;
-import io.odpf.dagger.common.serde.proto.protohandler.ProtoHandlerFactory;
+import io.odpf.dagger.common.serde.typehandler.TypeHandler;
+import io.odpf.dagger.common.serde.typehandler.TypeHandlerFactory;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -137,10 +137,10 @@ private DynamicMessage.Builder populateBuilder(DynamicMessage.Builder builder, D
         if (fieldDescriptor == null) {
             return builder;
         }
-        ProtoHandler protoHandler = ProtoHandlerFactory.getProtoHandler(fieldDescriptor);
+        TypeHandler typeHandler = TypeHandlerFactory.getTypeHandler(fieldDescriptor);
         if (data != null) {
             try {
-                builder = protoHandler.transformToProtoBuilder(builder, data);
+                builder = typeHandler.transformToProtoBuilder(builder, data);
             } catch (IllegalArgumentException e) {
                 String protoType = fieldDescriptor.getType().toString();
                 if (fieldDescriptor.isRepeated()) {

dagger-common/src/main/java/io/odpf/dagger/common/serde/proto/protohandler/PrimitiveProtoHandler.java → dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/PrimitiveTypeHandler.java

@@ -1,26 +1,26 @@
-package io.odpf.dagger.common.serde.proto.protohandler;
+package io.odpf.dagger.common.serde.typehandler;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 
 import io.odpf.dagger.common.exceptions.serde.InvalidDataTypeException;
-import io.odpf.dagger.common.serde.proto.protohandler.typehandler.PrimitiveTypeHandlerFactory;
-import io.odpf.dagger.common.serde.proto.protohandler.typehandler.PrimitiveTypeHandler;
+import io.odpf.dagger.common.serde.typehandler.primitive.PrimitiveHandlerFactory;
+import io.odpf.dagger.common.serde.typehandler.primitive.PrimitiveHandler;
 import com.google.protobuf.Descriptors;
 import com.google.protobuf.DynamicMessage;
 import org.apache.parquet.example.data.simple.SimpleGroup;
 
 /**
  * The type Primitive proto handler.
  */
-public class PrimitiveProtoHandler implements ProtoHandler {
+public class PrimitiveTypeHandler implements TypeHandler {
     private Descriptors.FieldDescriptor fieldDescriptor;
 
     /**
      * Instantiates a new Primitive proto handler.
      *
      * @param fieldDescriptor the field descriptor
      */
-    public PrimitiveProtoHandler(Descriptors.FieldDescriptor fieldDescriptor) {
+    public PrimitiveTypeHandler(Descriptors.FieldDescriptor fieldDescriptor) {
         this.fieldDescriptor = fieldDescriptor;
     }
 
@@ -40,9 +40,9 @@ public Object transformFromPostProcessor(Object field) {
     }
 
     private Object transform(Object field) {
-        PrimitiveTypeHandler primitiveTypeHandler = PrimitiveTypeHandlerFactory.getTypeHandler(fieldDescriptor);
+        PrimitiveHandler primitiveHandler = PrimitiveHandlerFactory.getTypeHandler(fieldDescriptor);
         try {
-            return primitiveTypeHandler.parseObject(field);
+            return primitiveHandler.parseObject(field);
         } catch (NumberFormatException e) {
             String errMessage = String.format("type mismatch of field: %s, expecting %s type, actual type %s", fieldDescriptor.getName(), fieldDescriptor.getType(), field.getClass());
             throw new InvalidDataTypeException(errMessage);
@@ -56,8 +56,8 @@ public Object transformFromProto(Object field) {
 
     @Override
     public Object transformFromParquet(SimpleGroup simpleGroup) {
-        PrimitiveTypeHandler primitiveTypeHandler = PrimitiveTypeHandlerFactory.getTypeHandler(fieldDescriptor);
-        return primitiveTypeHandler.parseSimpleGroup(simpleGroup);
+        PrimitiveHandler primitiveHandler = PrimitiveHandlerFactory.getTypeHandler(fieldDescriptor);
+        return primitiveHandler.parseSimpleGroup(simpleGroup);
     }
 
     @Override
@@ -67,8 +67,8 @@ public Object transformToJson(Object field) {
 
     @Override
     public TypeInformation getTypeInformation() {
-        PrimitiveTypeHandler primitiveTypeHandler = PrimitiveTypeHandlerFactory.getTypeHandler(fieldDescriptor);
-        return primitiveTypeHandler.getTypeInformation();
+        PrimitiveHandler primitiveHandler = PrimitiveHandlerFactory.getTypeHandler(fieldDescriptor);
+        return primitiveHandler.getTypeInformation();
     }
 
 }

dagger-common/src/main/java/io/odpf/dagger/common/serde/proto/protohandler/RowFactory.java → dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/RowFactory.java

@@ -1,4 +1,4 @@
-package io.odpf.dagger.common.serde.proto.protohandler;
+package io.odpf.dagger.common.serde.typehandler;
 
 import org.apache.flink.types.Row;
 
@@ -28,9 +28,9 @@ public static Row createRow(Map<String, Object> inputMap, Descriptors.Descriptor
             return row;
         }
         for (FieldDescriptor fieldDescriptor : descriptorFields) {
-            ProtoHandler protoHandler = ProtoHandlerFactory.getProtoHandler(fieldDescriptor);
+            TypeHandler typeHandler = TypeHandlerFactory.getTypeHandler(fieldDescriptor);
             if (inputMap.get(fieldDescriptor.getName()) != null) {
-                row.setField(fieldDescriptor.getIndex(), protoHandler.transformFromPostProcessor(inputMap.get(fieldDescriptor.getName())));
+                row.setField(fieldDescriptor.getIndex(), typeHandler.transformFromPostProcessor(inputMap.get(fieldDescriptor.getName())));
             }
         }
         return row;
@@ -47,8 +47,8 @@ public static Row createRow(DynamicMessage proto, int extraColumns) {
         List<FieldDescriptor> descriptorFields = proto.getDescriptorForType().getFields();
         Row row = new Row(descriptorFields.size() + extraColumns);
         for (FieldDescriptor fieldDescriptor : descriptorFields) {
-            ProtoHandler protoHandler = ProtoHandlerFactory.getProtoHandler(fieldDescriptor);
-            row.setField(fieldDescriptor.getIndex(), protoHandler.transformFromProto(proto.getField(fieldDescriptor)));
+            TypeHandler typeHandler = TypeHandlerFactory.getTypeHandler(fieldDescriptor);
+            row.setField(fieldDescriptor.getIndex(), typeHandler.transformFromProto(proto.getField(fieldDescriptor)));
         }
         return row;
     }
@@ -57,8 +57,8 @@ public static Row createRow(Descriptors.Descriptor descriptor, SimpleGroup simpl
         List<FieldDescriptor> descriptorFields = descriptor.getFields();
         Row row = new Row(descriptorFields.size() + extraColumns);
         for (FieldDescriptor fieldDescriptor : descriptorFields) {
-            ProtoHandler protoHandler = ProtoHandlerFactory.getProtoHandler(fieldDescriptor);
-            row.setField(fieldDescriptor.getIndex(), protoHandler.transformFromParquet(simpleGroup));
+            TypeHandler typeHandler = TypeHandlerFactory.getTypeHandler(fieldDescriptor);
+            row.setField(fieldDescriptor.getIndex(), typeHandler.transformFromParquet(simpleGroup));
         }
         return row;
     }

dagger-common/src/main/java/io/odpf/dagger/common/serde/proto/protohandler/ProtoHandler.java → dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/TypeHandler.java

@@ -1,4 +1,4 @@
-package io.odpf.dagger.common.serde.proto.protohandler;
+package io.odpf.dagger.common.serde.typehandler;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 
@@ -8,7 +8,7 @@
 /**
  * The interface Proto handler.
  */
-public interface ProtoHandler {
+public interface TypeHandler {
     /**
      * Can handle boolean.
      *

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

@@ -0,0 +1,57 @@
+package io.odpf.dagger.common.serde.typehandler;
+
+import com.google.protobuf.Descriptors;
+import io.odpf.dagger.common.serde.typehandler.complex.EnumHandler;
+import io.odpf.dagger.common.serde.typehandler.complex.MapHandler;
+import io.odpf.dagger.common.serde.typehandler.complex.MessageHandler;
+import io.odpf.dagger.common.serde.typehandler.complex.StructMessageHandler;
+import io.odpf.dagger.common.serde.typehandler.complex.TimestampHandler;
+import io.odpf.dagger.common.serde.typehandler.repeated.RepeatedEnumHandler;
+import io.odpf.dagger.common.serde.typehandler.repeated.RepeatedMessageHandler;
+import io.odpf.dagger.common.serde.typehandler.repeated.RepeatedPrimitiveHandler;
+import io.odpf.dagger.common.serde.typehandler.repeated.RepeatedStructMessageHandler;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * The factory class for Type handler.
+ */
+public class TypeHandlerFactory {
+    private static Map<String, TypeHandler> typeHandlerMap = new ConcurrentHashMap<>();
+
+    /**
+     * Gets type handler.
+     *
+     * @param fieldDescriptor the field descriptor
+     * @return the type handler
+     */
+    public static TypeHandler getTypeHandler(final Descriptors.FieldDescriptor fieldDescriptor) {
+        return typeHandlerMap.computeIfAbsent(fieldDescriptor.getFullName(),
+                k -> getSpecificHandlers(fieldDescriptor).stream().filter(TypeHandler::canHandle)
+                        .findFirst().orElseGet(() -> new PrimitiveTypeHandler(fieldDescriptor)));
+    }
+
+    /**
+     * Clear type handler map.
+     */
+    protected static void clearTypeHandlerMap() {
+        typeHandlerMap.clear();
+    }
+
+    private static List<TypeHandler> getSpecificHandlers(Descriptors.FieldDescriptor fieldDescriptor) {
+        return Arrays.asList(
+                new MapHandler(fieldDescriptor),
+                new TimestampHandler(fieldDescriptor),
+                new EnumHandler(fieldDescriptor),
+                new StructMessageHandler(fieldDescriptor),
+                new RepeatedStructMessageHandler(fieldDescriptor),
+                new RepeatedPrimitiveHandler(fieldDescriptor),
+                new RepeatedMessageHandler(fieldDescriptor),
+                new RepeatedEnumHandler(fieldDescriptor),
+                new MessageHandler(fieldDescriptor)
+        );
+    }
+}

dagger-common/src/main/java/io/odpf/dagger/common/serde/proto/protohandler/TypeInformationFactory.java → dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/TypeInformationFactory.java

@@ -1,12 +1,11 @@
-package io.odpf.dagger.common.serde.proto.protohandler;
+package io.odpf.dagger.common.serde.typehandler;
 
+import com.google.protobuf.Descriptors;
+import io.odpf.dagger.common.exceptions.DescriptorNotFoundException;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.types.Row;
 
-import com.google.protobuf.Descriptors;
-import io.odpf.dagger.common.exceptions.DescriptorNotFoundException;
-
 /**
  * The factory class for Type information.
  */
@@ -29,7 +28,7 @@ public static TypeInformation<Row> getRowType(Descriptors.Descriptor descriptor)
         TypeInformation[] fieldTypes = descriptor
                 .getFields()
                 .stream()
-                .map(fieldDescriptor -> ProtoHandlerFactory.getProtoHandler(fieldDescriptor).getTypeInformation())
+                .map(fieldDescriptor -> TypeHandlerFactory.getTypeHandler(fieldDescriptor).getTypeInformation())
                 .toArray(TypeInformation[]::new);
         return Types.ROW_NAMED(fieldNames, fieldTypes);
     }

dagger-common/src/main/java/io/odpf/dagger/common/serde/proto/protohandler/EnumProtoHandler.java → dagger-common/src/main/java/io/odpf/dagger/common/serde/typehandler/complex/EnumHandler.java

@@ -1,6 +1,7 @@
-package io.odpf.dagger.common.serde.proto.protohandler;
+package io.odpf.dagger.common.serde.typehandler.complex;
 
 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;
 
@@ -12,15 +13,15 @@
 /**
  * The type Enum proto handler.
  */
-public class EnumProtoHandler implements ProtoHandler {
+public class EnumHandler implements TypeHandler {
     private Descriptors.FieldDescriptor fieldDescriptor;
 
     /**
      * Instantiates a new Enum proto handler.
      *
      * @param fieldDescriptor the field descriptor
      */
-    public EnumProtoHandler(Descriptors.FieldDescriptor fieldDescriptor) {
+    public EnumHandler(Descriptors.FieldDescriptor fieldDescriptor) {
         this.fieldDescriptor = fieldDescriptor;
     }