[Temp] feature/branch off connector serde

drift-api/src/main/java/com/facebook/drift/annotations/ThriftField.java

@@ -45,6 +45,8 @@
 
     Recursiveness isRecursive() default Recursiveness.UNSPECIFIED;
 
+    Class serde() default Object.class;
+
     ThriftIdlAnnotation[] idlAnnotations() default {};
 
     enum Recursiveness

drift-codec/src/main/java/com/facebook/drift/codec/ThriftCodecManager.java

@@ -17,6 +17,7 @@
 
 import com.facebook.drift.codec.internal.EnumThriftCodec;
 import com.facebook.drift.codec.internal.ThriftCodecFactory;
+import com.facebook.drift.codec.internal.builtin.AnyCodec;
 import com.facebook.drift.codec.internal.builtin.BooleanArrayThriftCodec;
 import com.facebook.drift.codec.internal.builtin.BooleanThriftCodec;
 import com.facebook.drift.codec.internal.builtin.ByteBufferThriftCodec;
@@ -41,6 +42,7 @@
 import com.facebook.drift.codec.internal.builtin.VoidThriftCodec;
 import com.facebook.drift.codec.internal.coercion.CoercionThriftCodec;
 import com.facebook.drift.codec.internal.compiler.CompilerThriftCodecFactory;
+import com.facebook.drift.codec.metadata.Any;
 import com.facebook.drift.codec.metadata.ReflectionHelper;
 import com.facebook.drift.codec.metadata.ThriftCatalog;
 import com.facebook.drift.codec.metadata.ThriftType;
@@ -140,6 +142,9 @@ public ThriftCodec<?> load(ThriftType type)
 
                     switch (type.getProtocolType()) {
                         case STRUCT:
+                            if (type.getJavaType().equals(Any.class)) {
+                                return new AnyCodec();
+                            }
                             return factory.generateThriftTypeCodec(ThriftCodecManager.this, type.getStructMetadata());
                         case MAP:
                             return new MapThriftCodec<>(type, getElementCodec(type.getKeyTypeReference()), getElementCodec(type.getValueTypeReference()));

drift-codec/src/main/java/com/facebook/drift/codec/internal/ProtocolWriter.java

@@ -67,6 +67,18 @@ public <T> void writeField(String name, short id, ThriftCodec<T> codec, T value)
         protocol.writeFieldEnd();
     }
 
+    public void writeRawField(String name, short id, ThriftCodec codec, Object value)
+            throws Exception
+    {
+        if (codec.isNull(value)) {
+            return;
+        }
+
+        protocol.writeFieldBegin(new TField(name, codec.getType().getProtocolType().getType(), id));
+        codec.write(value, protocol);
+        protocol.writeFieldEnd();
+    }
+
     public void writeBinaryField(String name, short id, ByteBuffer buf)
             throws TException
     {

drift-codec/src/main/java/com/facebook/drift/codec/internal/builtin/AnyCodec.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright (C) 2012 ${project.organization.name}
+ *
+ * Licensed 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 com.facebook.drift.codec.internal.builtin;
+
+import com.facebook.drift.codec.ThriftCodec;
+import com.facebook.drift.codec.internal.ProtocolReader;
+import com.facebook.drift.codec.internal.ProtocolWriter;
+import com.facebook.drift.codec.metadata.Any;
+import com.facebook.drift.codec.metadata.ThriftType;
+import com.facebook.drift.protocol.TProtocolReader;
+import com.facebook.drift.protocol.TProtocolWriter;
+
+public class AnyCodec
+        implements ThriftCodec<Any>
+{
+    @Override
+    public ThriftType getType()
+    {
+        return ThriftType.ANY;
+    }
+
+    @Override
+    public Any read(TProtocolReader protocol)
+            throws Exception
+    {
+        String id = null;
+        byte[] byteArr = new byte[]{};
+        ProtocolReader protocolReader = new ProtocolReader(protocol);
+        protocolReader.readStructBegin();
+        while (protocolReader.nextField()) {
+            if (protocolReader.getFieldId() == 1) {
+                id = (String) protocolReader.readField(new StringThriftCodec());
+            }
+            /*else if (protocolReader.getFieldId() == 2) {
+                Class<? extends T> aClass = classResolver.apply(id);
+                ThriftCodec<?> codec = new ThriftCodecManager(new CompilerThriftCodecFactory(true)).getCodec(aClass);
+                byteArr = (byte[]) protocolReader.readField(codec);
+            }*/
+        }
+        protocolReader.readStructEnd();
+        return new Any(id, byteArr);
+    }
+
+    @Override
+    public void write(Any value, TProtocolWriter protocol)
+            throws Exception
+    {
+        ProtocolWriter writer = new ProtocolWriter(protocol);
+        writer.writeStructBegin("Any");
+        ThriftCodec<String> stringCodec = new StringThriftCodec();
+        String id = value.getId();
+        writer.writeField("id", (short) 1, stringCodec, id);
+        //Class<? extends T> aClass = classResolver.apply(id);
+        //ThriftCodec<?> codec = new ThriftCodecManager(new CompilerThriftCodecFactory(true)).getCodec(aClass);
+        //Object cast = aClass.cast(value.getBytes());
+        //writer.writeRawField("obj", (short) 2, codec, cast);
+        writer.writeStructEnd();
+    }
+}

drift-codec/src/main/java/com/facebook/drift/codec/internal/reflection/ReflectionThriftStructCodec.java

@@ -20,6 +20,8 @@
 import com.facebook.drift.codec.ThriftCodecManager;
 import com.facebook.drift.codec.internal.ProtocolReader;
 import com.facebook.drift.codec.internal.ProtocolWriter;
+import com.facebook.drift.codec.internal.builtin.AnyCodec;
+import com.facebook.drift.codec.metadata.Any;
 import com.facebook.drift.codec.metadata.FieldKind;
 import com.facebook.drift.codec.metadata.ThriftConstructorInjection;
 import com.facebook.drift.codec.metadata.ThriftFieldInjection;
@@ -117,7 +119,13 @@ public void write(T instance, TProtocolWriter protocol)
             if (fieldValue != null) {
                 @SuppressWarnings("unchecked")
                 ThriftCodec<Object> codec = (ThriftCodec<Object>) fields.get(fieldMetadata.getId());
-                writer.writeField(fieldMetadata.getName(), fieldMetadata.getId(), codec, fieldValue);
+                //Special handling
+                if (codec.getClass().equals(AnyCodec.class)) {
+                    writer.writeField("any", fieldMetadata.getId(), codec, new Any("id", new byte[]{}));
+                }
+                else {
+                    writer.writeField(fieldMetadata.getName(), fieldMetadata.getId(), codec, fieldValue);
+                }
             }
         }
         writer.writeStructEnd();
@@ -137,6 +145,10 @@ private T constructStruct(Map<Short, Object> data)
                 if (value == null) {
                     value = metadata.getField(parameter.getId()).getThriftType().getNullValue();
                 }
+                if (value instanceof Any) {
+                    //Special handling
+                    value = null;
+                }
                 parametersValues[parameter.getParameterIndex()] = value;
             }
 

drift-codec/src/main/java/com/facebook/drift/codec/metadata/AbstractThriftMetadataBuilder.java

@@ -505,6 +505,11 @@ protected final void normalizeThriftFields(ThriftCatalog catalog)
                 field.setRequiredness(requiredness);
             }
 
+            Class serde = extractFieldSerde(fieldId, fields);
+            for (FieldMetadata field : fields) {
+                field.setSerde(serde);
+            }
+
             // We need to do the isLegacyId check in two places. We've already done this
             // process for fields which had multiple `@ThriftField` annotations when we
             // assigned them all the same ID. It doesn't hurt to do it again. On the other
@@ -633,6 +638,25 @@ protected final boolean extractFieldIsRecursiveReference(short fieldId, Collecti
         return isRecursiveReferences.iterator().next();
     }
 
+    protected final Class extractFieldSerde(short fieldId, Collection<FieldMetadata> fields)
+    {
+        Set<Class> serdeClasses =
+                fields.stream()
+                        .map(FieldMetadata::getSerde)
+                        .filter(Objects::nonNull)
+                        .collect(toSet());
+
+        if (serdeClasses.size() > 1) {
+            metadataErrors.addError("Thrift class '%s' field '%s' has both isRecursiveReference=TRUE and isRecursiveReference=FALSE", structName, fieldId);
+        }
+        if (serdeClasses.isEmpty()) {
+            return Object.class;
+        }
+        else {
+            return serdeClasses.iterator().next();
+        }
+    }
+
     protected final boolean extractFieldIsLegacyId(short id, String fieldName, Collection<FieldMetadata> fields)
     {
         Set<Boolean> isLegacyIds = fields.stream()
@@ -711,6 +735,9 @@ protected final void verifyFieldType(short id, String name, Collection<FieldMeta
     {
         boolean isSupportedType = true;
         for (FieldMetadata field : fields) {
+            if (field.getSerde() != Object.class) {
+                break;
+            }
             if (!catalog.isSupportedStructFieldType(field.getJavaType())) {
                 metadataErrors.addError("Thrift class '%s' field '%s(%s)' type '%s' is not a supported Java type", structName, name, id, TypeToken.of(field.getJavaType()));
                 isSupportedType = false;

drift-codec/src/main/java/com/facebook/drift/codec/metadata/Any.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright (C) 2012 ${project.organization.name}
+ *
+ * Licensed 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 com.facebook.drift.codec.metadata;
+
+import com.facebook.drift.annotations.ThriftConstructor;
+import com.facebook.drift.annotations.ThriftField;
+import com.facebook.drift.annotations.ThriftStruct;
+
+import java.util.Arrays;
+
+@ThriftStruct
+public class Any
+{
+    private String id;
+    private byte[] bytes;
+
+    @ThriftConstructor
+    public Any(@ThriftField(1) String id, @ThriftField(2) byte[] bytes)
+    {
+        this.id = id;
+        this.bytes = Arrays.copyOf(bytes, bytes.length);
+    }
+
+    public String getId()
+    {
+        return id;
+    }
+
+    public byte[] getBytes()
+    {
+        //TODO : Remove copying. Temporarily for satisyfing spotbugs
+        return Arrays.copyOf(bytes, bytes.length);
+    }
+}

drift-codec/src/main/java/com/facebook/drift/codec/metadata/FieldMetadata.java

@@ -37,6 +37,8 @@ abstract class FieldMetadata
     private Boolean isRecursiveReference;
     private String name;
     private Requiredness requiredness;
+    //could be Class<? extends CustomSerdeInterface>
+    private Class serde;
     private Map<String, String> idlAnnotations;
     private final FieldKind type;
 
@@ -55,6 +57,9 @@ protected FieldMetadata(ThriftField annotation, FieldKind type)
                         name = annotation.name();
                     }
                     requiredness = requireNonNull(annotation.requiredness());
+                    if (annotation.serde() != Object.class) {
+                        serde = annotation.serde();
+                    }
 
                     ImmutableMap.Builder<String, String> annotationMapBuilder = ImmutableMap.builder();
                     for (ThriftIdlAnnotation idlAnnotation : annotation.idlAnnotations()) {
@@ -187,4 +192,14 @@ public void setIsRecursiveReference(Boolean isRecursiveReference)
     {
         this.isRecursiveReference = isRecursiveReference;
     }
+
+    public Class getSerde()
+    {
+        return serde;
+    }
+
+    public void setSerde(Class serde)
+    {
+        this.serde = serde;
+    }
 }

drift-codec/src/main/java/com/facebook/drift/codec/metadata/ThriftCatalog.java

@@ -55,6 +55,7 @@
 import static com.facebook.drift.codec.metadata.ReflectionHelper.getMapValueType;
 import static com.facebook.drift.codec.metadata.ReflectionHelper.getOptionalType;
 import static com.facebook.drift.codec.metadata.ThriftEnumMetadataBuilder.thriftEnumMetadata;
+import static com.facebook.drift.codec.metadata.ThriftType.ANY;
 import static com.facebook.drift.codec.metadata.ThriftType.BINARY;
 import static com.facebook.drift.codec.metadata.ThriftType.BOOL;
 import static com.facebook.drift.codec.metadata.ThriftType.BYTE;
@@ -321,6 +322,9 @@ private ThriftType buildThriftTypeInternal(Type javaType)
         if (void.class.isAssignableFrom(rawType) || Void.class.isAssignableFrom(rawType)) {
             return VOID;
         }
+        if (Any.class.isAssignableFrom(rawType)) {
+            return ANY;
+        }
         if (isStructType(rawType)) {
             ThriftStructMetadata structMetadata = getThriftStructMetadata(javaType);
             // Unions are covered because a union looks like a struct with a single field.
@@ -351,8 +355,16 @@ public ThriftTypeReference getFieldThriftTypeReference(FieldMetadata fieldMetada
                     "Field normalization should have set a non-null value for isRecursiveReference");
         }
 
+        Class serde = fieldMetadata.getSerde();
+
+        Type javaType = fieldMetadata.getJavaType();
+
+        if (!Object.class.equals(serde)) {
+            javaType = TypeToken.of(Any.class).getType();
+        }
+
         return getThriftTypeReference(
-                fieldMetadata.getJavaType(),
+                javaType,
                 isRecursive ? Recursiveness.FORCED : Recursiveness.NOT_ALLOWED);
     }
 

drift-codec/src/main/java/com/facebook/drift/codec/metadata/ThriftType.java

@@ -48,6 +48,7 @@ public class ThriftType
     public static final ThriftType STRING = new ThriftType(ThriftProtocolType.STRING, String.class);
     public static final ThriftType BINARY = new ThriftType(ThriftProtocolType.BINARY, ByteBuffer.class);
     public static final ThriftType VOID = new ThriftType(ThriftProtocolType.STRUCT, void.class);
+    public static final ThriftType ANY = new ThriftType(ThriftProtocolType.STRUCT, Any.class);
 
     public static final ThriftTypeReference BOOL_REF = new DefaultThriftTypeReference(BOOL);
     public static final ThriftTypeReference BYTE_REF = new DefaultThriftTypeReference(BYTE);