/*
 * 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 org.apache.flink.api.java.typeutils;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.CompositeType.FlatFieldDescriptor;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.types.Value;

import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.junit.Assert;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.HashSet;
import java.util.List;

/**
 * Pojo Type tests.
 *
 * <p>Pojo is a bean-style class with getters, setters and empty ctor OR a class with all fields
 * public (or for every private field, there has to be a public getter/setter) everything else is a
 * generic type (that can't be used for field selection)
 */
public class PojoTypeExtractionTest {

    /** Simple test type that implements the {@link Value} interface. */
    public static class MyValue implements Value {
        private static final long serialVersionUID = 8607223484689147046L;

        @Override
        public void write(DataOutputView out) throws IOException {}

        @Override
        public void read(DataInputView in) throws IOException {}
    }

    public static class HasDuplicateField extends WC {
        @SuppressWarnings("unused")
        private int count; // duplicate
    }

    @Test
    public void testDuplicateFieldException() {
        TypeInformation<?> ti = TypeExtractor.createTypeInfo(HasDuplicateField.class);
        Assert.assertTrue(ti instanceof GenericTypeInfo<?>);
    }

    // test with correct pojo types
    public static class WC { // is a pojo
        public ComplexNestedClass complex; // is a pojo
        private int count; // is a BasicType

        public WC() {}

        public int getCount() {
            return count;
        }

        public void setCount(int c) {
            this.count = c;
        }
    }

    public static class ComplexNestedClass { // pojo
        public static int ignoreStaticField;
        public transient int ignoreTransientField;
        public Date date; // generic type
        public Integer someNumberWithÜnicödeNäme; // BasicType
        public float someFloat; // BasicType
        public Tuple3<Long, Long, String> word; // Tuple Type with three basic types
        public Object nothing; // generic type
        public MyValue valueType; // writableType
        public List<String> collection;
    }

    // all public test
    public static class AllPublic extends ComplexNestedClass {
        public ArrayList<String> somethingFancy; // generic type
        public FancyCollectionSubtype<Integer> fancyIds; // generic type
        public String[] fancyArray; // generic type
    }

    public static class FancyCollectionSubtype<T> extends HashSet<T> {
        private static final long serialVersionUID = -3494469602638179921L;
    }

    public static class ParentSettingGenerics extends PojoWithGenerics<Integer, Long> {
        public String field3;
    }

    public static class PojoWithGenerics<T1, T2> {
        public int key;
        public T1 field1;
        public T2 field2;
    }

    public static class ComplexHierarchyTop extends ComplexHierarchy<Tuple1<String>> {}

    public static class ComplexHierarchy<T> extends PojoWithGenerics<FromTuple, T> {}

    // extends from Tuple and adds a field
    public static class FromTuple extends Tuple3<String, String, Long> {
        private static final long serialVersionUID = 1L;
        public int special;
    }

    public static class IncorrectPojo {
        private int isPrivate;

        public int getIsPrivate() {
            return isPrivate;
        }
        // setter is missing (intentional)
    }

    // correct pojo
    public static class BeanStylePojo {
        public String abc;
        private int field;

        public int getField() {
            return this.field;
        }

        public void setField(int f) {
            this.field = f;
        }
    }

    public static class WrongCtorPojo {
        public int a;

        public WrongCtorPojo(int a) {
            this.a = a;
        }
    }

    public static class PojoWithGenericFields {
        private Collection<String> users;
        private boolean favorited;

        public boolean isFavorited() {
            return favorited;
        }

        public void setFavorited(boolean favorited) {
            this.favorited = favorited;
        }

        public Collection<String> getUsers() {
            return users;
        }

        public void setUsers(Collection<String> users) {
            this.users = users;
        }
    }

    @Test
    public void testPojoWithGenericFields() {
        TypeInformation<?> typeForClass = TypeExtractor.createTypeInfo(PojoWithGenericFields.class);

        Assert.assertTrue(typeForClass instanceof PojoTypeInfo<?>);
    }

    // in this test, the location of the getters and setters is mixed across the type hierarchy.
    public static class TypedPojoGetterSetterCheck extends GenericPojoGetterSetterCheck<String> {
        public void setPackageProtected(String in) {
            this.packageProtected = in;
        }
    }

    public static class GenericPojoGetterSetterCheck<T> {
        T packageProtected;

        public T getPackageProtected() {
            return packageProtected;
        }
    }

    @Test
    public void testIncorrectPojos() {
        TypeInformation<?> typeForClass = TypeExtractor.createTypeInfo(IncorrectPojo.class);
        Assert.assertTrue(typeForClass instanceof GenericTypeInfo<?>);

        typeForClass = TypeExtractor.createTypeInfo(WrongCtorPojo.class);
        Assert.assertTrue(typeForClass instanceof GenericTypeInfo<?>);
    }

    @Test
    public void testCorrectPojos() {
        TypeInformation<?> typeForClass = TypeExtractor.createTypeInfo(BeanStylePojo.class);
        Assert.assertTrue(typeForClass instanceof PojoTypeInfo<?>);

        typeForClass = TypeExtractor.createTypeInfo(TypedPojoGetterSetterCheck.class);
        Assert.assertTrue(typeForClass instanceof PojoTypeInfo<?>);
    }

    @Test
    public void testPojoWC() {
        TypeInformation<?> typeForClass = TypeExtractor.createTypeInfo(WC.class);
        checkWCPojoAsserts(typeForClass);

        WC t = new WC();
        t.complex = new ComplexNestedClass();
        TypeInformation<?> typeForObject = TypeExtractor.getForObject(t);
        checkWCPojoAsserts(typeForObject);
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    private void checkWCPojoAsserts(TypeInformation<?> typeInfo) {
        Assert.assertFalse(typeInfo.isBasicType());
        Assert.assertFalse(typeInfo.isTupleType());
        Assert.assertEquals(10, typeInfo.getTotalFields());
        Assert.assertTrue(typeInfo instanceof PojoTypeInfo);
        PojoTypeInfo<?> pojoType = (PojoTypeInfo<?>) typeInfo;

        List<FlatFieldDescriptor> ffd = new ArrayList<FlatFieldDescriptor>();
        String[] fields = {
            "count",
            "complex.date",
            "complex.collection",
            "complex.nothing",
            "complex.someFloat",
            "complex.someNumberWithÜnicödeNäme",
            "complex.valueType",
            "complex.word.f0",
            "complex.word.f1",
            "complex.word.f2"
        };
        int[] positions = {9, 1, 0, 2, 3, 4, 5, 6, 7, 8};
        Assert.assertEquals(fields.length, positions.length);
        for (int i = 0; i < fields.length; i++) {
            pojoType.getFlatFields(fields[i], 0, ffd);
            Assert.assertEquals("Too many keys returned", 1, ffd.size());
            Assert.assertEquals(
                    "position of field " + fields[i] + " wrong",
                    positions[i],
                    ffd.get(0).getPosition());
            ffd.clear();
        }

        pojoType.getFlatFields("complex.word.*", 0, ffd);
        Assert.assertEquals(3, ffd.size());
        // check if it returns 5,6,7
        for (FlatFieldDescriptor ffdE : ffd) {
            final int pos = ffdE.getPosition();
            Assert.assertTrue(pos <= 8);
            Assert.assertTrue(6 <= pos);
            if (pos == 6) {
                Assert.assertEquals(Long.class, ffdE.getType().getTypeClass());
            }
            if (pos == 7) {
                Assert.assertEquals(Long.class, ffdE.getType().getTypeClass());
            }
            if (pos == 8) {
                Assert.assertEquals(String.class, ffdE.getType().getTypeClass());
            }
        }
        ffd.clear();

        // scala style full tuple selection for pojos
        pojoType.getFlatFields("complex.word._", 0, ffd);
        Assert.assertEquals(3, ffd.size());
        ffd.clear();

        pojoType.getFlatFields("complex.*", 0, ffd);
        Assert.assertEquals(9, ffd.size());
        // check if it returns 0-7
        for (FlatFieldDescriptor ffdE : ffd) {
            final int pos = ffdE.getPosition();
            Assert.assertTrue(ffdE.getPosition() <= 8);
            Assert.assertTrue(0 <= ffdE.getPosition());

            if (pos == 0) {
                Assert.assertEquals(List.class, ffdE.getType().getTypeClass());
            }
            if (pos == 1) {
                Assert.assertEquals(Date.class, ffdE.getType().getTypeClass());
            }
            if (pos == 2) {
                Assert.assertEquals(Object.class, ffdE.getType().getTypeClass());
            }
            if (pos == 3) {
                Assert.assertEquals(Float.class, ffdE.getType().getTypeClass());
            }
            if (pos == 4) {
                Assert.assertEquals(Integer.class, ffdE.getType().getTypeClass());
            }
            if (pos == 5) {
                Assert.assertEquals(MyValue.class, ffdE.getType().getTypeClass());
            }
            if (pos == 6) {
                Assert.assertEquals(Long.class, ffdE.getType().getTypeClass());
            }
            if (pos == 7) {
                Assert.assertEquals(Long.class, ffdE.getType().getTypeClass());
            }
            if (pos == 8) {
                Assert.assertEquals(String.class, ffdE.getType().getTypeClass());
            }
            if (pos == 9) {
                Assert.assertEquals(Integer.class, ffdE.getType().getTypeClass());
            }
        }
        ffd.clear();

        pojoType.getFlatFields("*", 0, ffd);
        Assert.assertEquals(10, ffd.size());
        // check if it returns 0-8
        for (FlatFieldDescriptor ffdE : ffd) {
            Assert.assertTrue(ffdE.getPosition() <= 9);
            Assert.assertTrue(0 <= ffdE.getPosition());
            if (ffdE.getPosition() == 9) {
                Assert.assertEquals(Integer.class, ffdE.getType().getTypeClass());
            }
        }
        ffd.clear();

        TypeInformation<?> typeComplexNested = pojoType.getTypeAt(0); // ComplexNestedClass complex
        Assert.assertTrue(typeComplexNested instanceof PojoTypeInfo);

        Assert.assertEquals(7, typeComplexNested.getArity());
        Assert.assertEquals(9, typeComplexNested.getTotalFields());
        PojoTypeInfo<?> pojoTypeComplexNested = (PojoTypeInfo<?>) typeComplexNested;

        boolean dateSeen = false,
                intSeen = false,
                floatSeen = false,
                tupleSeen = false,
                objectSeen = false,
                writableSeen = false,
                collectionSeen = false;
        for (int i = 0; i < pojoTypeComplexNested.getArity(); i++) {
            PojoField field = pojoTypeComplexNested.getPojoFieldAt(i);
            String name = field.getField().getName();
            if (name.equals("date")) {
                if (dateSeen) {
                    Assert.fail("already seen");
                }
                dateSeen = true;
                Assert.assertEquals(BasicTypeInfo.DATE_TYPE_INFO, field.getTypeInformation());
                Assert.assertEquals(Date.class, field.getTypeInformation().getTypeClass());
            } else if (name.equals("someNumberWithÜnicödeNäme")) {
                if (intSeen) {
                    Assert.fail("already seen");
                }
                intSeen = true;
                Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, field.getTypeInformation());
                Assert.assertEquals(Integer.class, field.getTypeInformation().getTypeClass());
            } else if (name.equals("someFloat")) {
                if (floatSeen) {
                    Assert.fail("already seen");
                }
                floatSeen = true;
                Assert.assertEquals(BasicTypeInfo.FLOAT_TYPE_INFO, field.getTypeInformation());
                Assert.assertEquals(Float.class, field.getTypeInformation().getTypeClass());
            } else if (name.equals("word")) {
                if (tupleSeen) {
                    Assert.fail("already seen");
                }
                tupleSeen = true;
                Assert.assertTrue(field.getTypeInformation() instanceof TupleTypeInfo<?>);
                Assert.assertEquals(Tuple3.class, field.getTypeInformation().getTypeClass());
                // do some more advanced checks on the tuple
                TupleTypeInfo<?> tupleTypeFromComplexNested =
                        (TupleTypeInfo<?>) field.getTypeInformation();
                Assert.assertEquals(
                        BasicTypeInfo.LONG_TYPE_INFO, tupleTypeFromComplexNested.getTypeAt(0));
                Assert.assertEquals(
                        BasicTypeInfo.LONG_TYPE_INFO, tupleTypeFromComplexNested.getTypeAt(1));
                Assert.assertEquals(
                        BasicTypeInfo.STRING_TYPE_INFO, tupleTypeFromComplexNested.getTypeAt(2));
            } else if (name.equals("nothing")) {
                if (objectSeen) {
                    Assert.fail("already seen");
                }
                objectSeen = true;
                Assert.assertEquals(
                        new GenericTypeInfo<Object>(Object.class), field.getTypeInformation());
                Assert.assertEquals(Object.class, field.getTypeInformation().getTypeClass());
            } else if (name.equals("valueType")) {
                if (writableSeen) {
                    Assert.fail("already seen");
                }
                writableSeen = true;
                Assert.assertEquals(new ValueTypeInfo<>(MyValue.class), field.getTypeInformation());
                Assert.assertEquals(MyValue.class, field.getTypeInformation().getTypeClass());
            } else if (name.equals("collection")) {
                if (collectionSeen) {
                    Assert.fail("already seen");
                }
                collectionSeen = true;
                Assert.assertEquals(new GenericTypeInfo(List.class), field.getTypeInformation());

            } else {
                Assert.fail("field " + field + " is not expected");
            }
        }
        Assert.assertTrue("Field was not present", dateSeen);
        Assert.assertTrue("Field was not present", intSeen);
        Assert.assertTrue("Field was not present", floatSeen);
        Assert.assertTrue("Field was not present", tupleSeen);
        Assert.assertTrue("Field was not present", objectSeen);
        Assert.assertTrue("Field was not present", writableSeen);
        Assert.assertTrue("Field was not present", collectionSeen);

        TypeInformation<?> typeAtOne = pojoType.getTypeAt(1); // int count
        Assert.assertTrue(typeAtOne instanceof BasicTypeInfo);

        Assert.assertEquals(typeInfo.getTypeClass(), WC.class);
        Assert.assertEquals(typeInfo.getArity(), 2);
    }

    @Test
    public void testPojoAllPublic() {
        TypeInformation<?> typeForClass = TypeExtractor.createTypeInfo(AllPublic.class);
        checkAllPublicAsserts(typeForClass);

        TypeInformation<?> typeForObject = TypeExtractor.getForObject(new AllPublic());
        checkAllPublicAsserts(typeForObject);
    }

    private void checkAllPublicAsserts(TypeInformation<?> typeInformation) {
        Assert.assertTrue(typeInformation instanceof PojoTypeInfo);
        Assert.assertEquals(10, typeInformation.getArity());
        Assert.assertEquals(12, typeInformation.getTotalFields());
        // check if the three additional fields are identified correctly
        boolean arrayListSeen = false, multisetSeen = false, strArraySeen = false;
        PojoTypeInfo<?> pojoTypeForClass = (PojoTypeInfo<?>) typeInformation;
        for (int i = 0; i < pojoTypeForClass.getArity(); i++) {
            PojoField field = pojoTypeForClass.getPojoFieldAt(i);
            String name = field.getField().getName();
            if (name.equals("somethingFancy")) {
                if (arrayListSeen) {
                    Assert.fail("already seen");
                }
                arrayListSeen = true;
                Assert.assertTrue(field.getTypeInformation() instanceof GenericTypeInfo);
                Assert.assertEquals(ArrayList.class, field.getTypeInformation().getTypeClass());
            } else if (name.equals("fancyIds")) {
                if (multisetSeen) {
                    Assert.fail("already seen");
                }
                multisetSeen = true;
                Assert.assertTrue(field.getTypeInformation() instanceof GenericTypeInfo);
                Assert.assertEquals(
                        FancyCollectionSubtype.class, field.getTypeInformation().getTypeClass());
            } else if (name.equals("fancyArray")) {
                if (strArraySeen) {
                    Assert.fail("already seen");
                }
                strArraySeen = true;
                Assert.assertEquals(
                        BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO, field.getTypeInformation());
                Assert.assertEquals(String[].class, field.getTypeInformation().getTypeClass());
            } else if (Arrays.asList(
                            "date",
                            "someNumberWithÜnicödeNäme",
                            "someFloat",
                            "word",
                            "nothing",
                            "valueType",
                            "collection")
                    .contains(name)) {
                // ignore these, they are inherited from the ComplexNestedClass
            } else {
                Assert.fail("field " + field + " is not expected");
            }
        }
        Assert.assertTrue("Field was not present", arrayListSeen);
        Assert.assertTrue("Field was not present", multisetSeen);
        Assert.assertTrue("Field was not present", strArraySeen);
    }

    @Test
    public void testPojoExtendingTuple() {
        TypeInformation<?> typeForClass = TypeExtractor.createTypeInfo(FromTuple.class);
        checkFromTuplePojo(typeForClass);

        FromTuple ft = new FromTuple();
        ft.f0 = "";
        ft.f1 = "";
        ft.f2 = 0L;
        TypeInformation<?> typeForObject = TypeExtractor.getForObject(ft);
        checkFromTuplePojo(typeForObject);
    }

    private void checkFromTuplePojo(TypeInformation<?> typeInformation) {
        Assert.assertTrue(typeInformation instanceof PojoTypeInfo<?>);
        Assert.assertEquals(4, typeInformation.getTotalFields());
        PojoTypeInfo<?> pojoTypeForClass = (PojoTypeInfo<?>) typeInformation;
        for (int i = 0; i < pojoTypeForClass.getArity(); i++) {
            PojoField field = pojoTypeForClass.getPojoFieldAt(i);
            String name = field.getField().getName();
            if (name.equals("special")) {
                Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, field.getTypeInformation());
            } else if (name.equals("f0") || name.equals("f1")) {
                Assert.assertEquals(BasicTypeInfo.STRING_TYPE_INFO, field.getTypeInformation());
            } else if (name.equals("f2")) {
                Assert.assertEquals(BasicTypeInfo.LONG_TYPE_INFO, field.getTypeInformation());
            } else {
                Assert.fail("unexpected field");
            }
        }
    }

    @Test
    public void testPojoWithGenerics() {
        TypeInformation<?> typeForClass = TypeExtractor.createTypeInfo(ParentSettingGenerics.class);
        Assert.assertTrue(typeForClass instanceof PojoTypeInfo<?>);
        PojoTypeInfo<?> pojoTypeForClass = (PojoTypeInfo<?>) typeForClass;
        for (int i = 0; i < pojoTypeForClass.getArity(); i++) {
            PojoField field = pojoTypeForClass.getPojoFieldAt(i);
            String name = field.getField().getName();
            if (name.equals("field1")) {
                Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, field.getTypeInformation());
            } else if (name.equals("field2")) {
                Assert.assertEquals(BasicTypeInfo.LONG_TYPE_INFO, field.getTypeInformation());
            } else if (name.equals("field3")) {
                Assert.assertEquals(BasicTypeInfo.STRING_TYPE_INFO, field.getTypeInformation());
            } else if (name.equals("key")) {
                Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, field.getTypeInformation());
            } else {
                Assert.fail("Unexpected field " + field);
            }
        }
    }

    /** Test if the TypeExtractor is accepting untyped generics, making them GenericTypes */
    @Test
    public void testPojoWithGenericsSomeFieldsGeneric() {
        TypeInformation<?> typeForClass = TypeExtractor.createTypeInfo(PojoWithGenerics.class);
        Assert.assertTrue(typeForClass instanceof PojoTypeInfo<?>);
        PojoTypeInfo<?> pojoTypeForClass = (PojoTypeInfo<?>) typeForClass;
        for (int i = 0; i < pojoTypeForClass.getArity(); i++) {
            PojoField field = pojoTypeForClass.getPojoFieldAt(i);
            String name = field.getField().getName();
            if (name.equals("field1")) {
                Assert.assertEquals(
                        new GenericTypeInfo<Object>(Object.class), field.getTypeInformation());
            } else if (name.equals("field2")) {
                Assert.assertEquals(
                        new GenericTypeInfo<Object>(Object.class), field.getTypeInformation());
            } else if (name.equals("key")) {
                Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, field.getTypeInformation());
            } else {
                Assert.fail("Unexpected field " + field);
            }
        }
    }

    @Test
    public void testPojoWithComplexHierarchy() {
        TypeInformation<?> typeForClass = TypeExtractor.createTypeInfo(ComplexHierarchyTop.class);
        Assert.assertTrue(typeForClass instanceof PojoTypeInfo<?>);
        PojoTypeInfo<?> pojoTypeForClass = (PojoTypeInfo<?>) typeForClass;
        for (int i = 0; i < pojoTypeForClass.getArity(); i++) {
            PojoField field = pojoTypeForClass.getPojoFieldAt(i);
            String name = field.getField().getName();
            if (name.equals("field1")) {
                Assert.assertTrue(
                        field.getTypeInformation()
                                instanceof PojoTypeInfo<?>); // From tuple is pojo (not tuple type!)
            } else if (name.equals("field2")) {
                Assert.assertTrue(field.getTypeInformation() instanceof TupleTypeInfo<?>);
                Assert.assertTrue(
                        ((TupleTypeInfo<?>) field.getTypeInformation())
                                .getTypeAt(0)
                                .equals(BasicTypeInfo.STRING_TYPE_INFO));
            } else if (name.equals("key")) {
                Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, field.getTypeInformation());
            } else {
                Assert.fail("Unexpected field " + field);
            }
        }
    }

    public static class MyMapper<T>
            implements MapFunction<PojoWithGenerics<Long, T>, PojoWithGenerics<T, T>> {
        private static final long serialVersionUID = 1L;

        @Override
        public PojoWithGenerics<T, T> map(PojoWithGenerics<Long, T> value) throws Exception {
            return null;
        }
    }

    @Test
    public void testGenericPojoTypeInference1() {
        MyMapper<String> function = new MyMapper<>();

        TypeInformation<?> ti =
                TypeExtractor.getMapReturnTypes(
                        function,
                        TypeInformation.of(new TypeHint<PojoWithGenerics<Long, String>>() {}));

        Assert.assertTrue(ti instanceof PojoTypeInfo<?>);
        PojoTypeInfo<?> pti = (PojoTypeInfo<?>) ti;
        for (int i = 0; i < pti.getArity(); i++) {
            PojoField field = pti.getPojoFieldAt(i);
            String name = field.getField().getName();
            if (name.equals("field1")) {
                Assert.assertEquals(BasicTypeInfo.STRING_TYPE_INFO, field.getTypeInformation());
            } else if (name.equals("field2")) {
                Assert.assertEquals(BasicTypeInfo.STRING_TYPE_INFO, field.getTypeInformation());
            } else if (name.equals("key")) {
                Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, field.getTypeInformation());
            } else {
                Assert.fail("Unexpected field " + field);
            }
        }
    }

    public static class PojoTuple<A, B, C> extends Tuple3<B, C, Long> {
        private static final long serialVersionUID = 1L;

        public A extraField;
    }

    public static class MyMapper2<D, E> implements MapFunction<Tuple2<E, D>, PojoTuple<E, D, D>> {
        private static final long serialVersionUID = 1L;

        @Override
        public PojoTuple<E, D, D> map(Tuple2<E, D> value) throws Exception {
            return null;
        }
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    @Test
    public void testGenericPojoTypeInference2() {
        MyMapper2<Boolean, Character> function = new MyMapper2<>();

        TypeInformation<?> ti =
                TypeExtractor.getMapReturnTypes(
                        function,
                        TypeInformation.of(new TypeHint<Tuple2<Character, Boolean>>() {}));

        Assert.assertTrue(ti instanceof PojoTypeInfo<?>);
        PojoTypeInfo<?> pti = (PojoTypeInfo<?>) ti;
        for (int i = 0; i < pti.getArity(); i++) {
            PojoField field = pti.getPojoFieldAt(i);
            String name = field.getField().getName();
            if (name.equals("extraField")) {
                Assert.assertEquals(BasicTypeInfo.CHAR_TYPE_INFO, field.getTypeInformation());
            } else if (name.equals("f0")) {
                Assert.assertEquals(BasicTypeInfo.BOOLEAN_TYPE_INFO, field.getTypeInformation());
            } else if (name.equals("f1")) {
                Assert.assertEquals(BasicTypeInfo.BOOLEAN_TYPE_INFO, field.getTypeInformation());
            } else if (name.equals("f2")) {
                Assert.assertEquals(BasicTypeInfo.LONG_TYPE_INFO, field.getTypeInformation());
            } else {
                Assert.fail("Unexpected field " + field);
            }
        }
    }

    public static class MyMapper3<D, E> implements MapFunction<PojoTuple<E, D, D>, Tuple2<E, D>> {
        private static final long serialVersionUID = 1L;

        @Override
        public Tuple2<E, D> map(PojoTuple<E, D, D> value) throws Exception {
            return null;
        }
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    @Test
    public void testGenericPojoTypeInference3() {
        MyMapper3<Boolean, Character> function = new MyMapper3<>();

        TypeInformation<?> ti =
                TypeExtractor.getMapReturnTypes(
                        function,
                        TypeInformation.of(
                                new TypeHint<PojoTuple<Character, Boolean, Boolean>>() {}));

        Assert.assertTrue(ti instanceof TupleTypeInfo<?>);
        TupleTypeInfo<?> tti = (TupleTypeInfo<?>) ti;
        Assert.assertEquals(BasicTypeInfo.CHAR_TYPE_INFO, tti.getTypeAt(0));
        Assert.assertEquals(BasicTypeInfo.BOOLEAN_TYPE_INFO, tti.getTypeAt(1));
    }

    public static class PojoWithParameterizedFields1<Z> {
        public Tuple2<Z, Z> field;
    }

    public static class MyMapper4<A> implements MapFunction<PojoWithParameterizedFields1<A>, A> {
        private static final long serialVersionUID = 1L;

        @Override
        public A map(PojoWithParameterizedFields1<A> value) throws Exception {
            return null;
        }
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    @Test
    public void testGenericPojoTypeInference4() {
        MyMapper4<Byte> function = new MyMapper4<>();

        TypeInformation<?> ti =
                TypeExtractor.getMapReturnTypes(
                        function,
                        TypeInformation.of(new TypeHint<PojoWithParameterizedFields1<Byte>>() {}));
        Assert.assertEquals(BasicTypeInfo.BYTE_TYPE_INFO, ti);
    }

    public static class PojoWithParameterizedFields2<Z> {
        public PojoWithGenerics<Z, Z> field;
    }

    public static class MyMapper5<A> implements MapFunction<PojoWithParameterizedFields2<A>, A> {
        private static final long serialVersionUID = 1L;

        @Override
        public A map(PojoWithParameterizedFields2<A> value) throws Exception {
            return null;
        }
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    @Test
    public void testGenericPojoTypeInference5() {
        MyMapper5<Byte> function = new MyMapper5<>();

        TypeInformation<?> ti =
                TypeExtractor.getMapReturnTypes(
                        function,
                        TypeInformation.of(new TypeHint<PojoWithParameterizedFields2<Byte>>() {}));
        Assert.assertEquals(BasicTypeInfo.BYTE_TYPE_INFO, ti);
    }

    public static class PojoWithParameterizedFields3<Z> {
        public Z[] field;
    }

    public static class MyMapper6<A> implements MapFunction<PojoWithParameterizedFields3<A>, A> {
        private static final long serialVersionUID = 1L;

        @Override
        public A map(PojoWithParameterizedFields3<A> value) throws Exception {
            return null;
        }
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    @Test
    public void testGenericPojoTypeInference6() {
        MyMapper6<Integer> function = new MyMapper6<>();

        TypeInformation<?> ti =
                TypeExtractor.getMapReturnTypes(
                        function,
                        TypeInformation.of(
                                new TypeHint<PojoWithParameterizedFields3<Integer>>() {}));
        Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, ti);
    }

    public static class MyMapper7<A> implements MapFunction<PojoWithParameterizedFields4<A>, A> {
        private static final long serialVersionUID = 1L;

        @Override
        public A map(PojoWithParameterizedFields4<A> value) throws Exception {
            return null;
        }
    }

    public static class PojoWithParameterizedFields4<Z> {
        public Tuple1<Z>[] field;
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    @Test
    public void testGenericPojoTypeInference7() {
        MyMapper7<Integer> function = new MyMapper7<>();

        TypeInformation<?> ti =
                TypeExtractor.getMapReturnTypes(
                        function,
                        TypeInformation.of(
                                new TypeHint<PojoWithParameterizedFields4<Integer>>() {}));

        Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, ti);
    }

    public static class RecursivePojo1 {
        public RecursivePojo1 field;
    }

    public static class RecursivePojo2 {
        public Tuple1<RecursivePojo2> field;
    }

    public static class RecursivePojo3 {
        public NestedPojo field;
    }

    public static class NestedPojo {
        public RecursivePojo3 field;
    }

    @Test
    public void testRecursivePojo1() {
        TypeInformation<?> ti = TypeExtractor.createTypeInfo(RecursivePojo1.class);
        Assert.assertTrue(ti instanceof PojoTypeInfo);
        Assert.assertEquals(
                GenericTypeInfo.class,
                ((PojoTypeInfo) ti).getPojoFieldAt(0).getTypeInformation().getClass());
    }

    @Test
    public void testRecursivePojo2() {
        TypeInformation<?> ti = TypeExtractor.createTypeInfo(RecursivePojo2.class);
        Assert.assertTrue(ti instanceof PojoTypeInfo);
        PojoField pf = ((PojoTypeInfo) ti).getPojoFieldAt(0);
        Assert.assertTrue(pf.getTypeInformation() instanceof TupleTypeInfo);
        Assert.assertEquals(
                GenericTypeInfo.class,
                ((TupleTypeInfo) pf.getTypeInformation()).getTypeAt(0).getClass());
    }

    @Test
    public void testRecursivePojo3() {
        TypeInformation<?> ti = TypeExtractor.createTypeInfo(RecursivePojo3.class);
        Assert.assertTrue(ti instanceof PojoTypeInfo);
        PojoField pf = ((PojoTypeInfo) ti).getPojoFieldAt(0);
        Assert.assertTrue(pf.getTypeInformation() instanceof PojoTypeInfo);
        Assert.assertEquals(
                GenericTypeInfo.class,
                ((PojoTypeInfo) pf.getTypeInformation())
                        .getPojoFieldAt(0)
                        .getTypeInformation()
                        .getClass());
    }

    public static class FooBarPojo {
        public int foo, bar;

        public FooBarPojo() {}
    }

    public static class DuplicateMapper
            implements MapFunction<FooBarPojo, Tuple2<FooBarPojo, FooBarPojo>> {
        @Override
        public Tuple2<FooBarPojo, FooBarPojo> map(FooBarPojo value) throws Exception {
            return null;
        }
    }

    @Test
    public void testDualUseOfPojo() {
        MapFunction<?, ?> function = new DuplicateMapper();
        TypeInformation<?> ti =
                TypeExtractor.getMapReturnTypes(
                        function, (TypeInformation) TypeExtractor.createTypeInfo(FooBarPojo.class));
        Assert.assertTrue(ti instanceof TupleTypeInfo);
        TupleTypeInfo<?> tti = ((TupleTypeInfo) ti);
        Assert.assertTrue(tti.getTypeAt(0) instanceof PojoTypeInfo);
        Assert.assertTrue(tti.getTypeAt(1) instanceof PojoTypeInfo);
    }

    public static class PojoWithRecursiveGenericField<K, V> {
        public PojoWithRecursiveGenericField<K, V> parent;

        public PojoWithRecursiveGenericField() {}
    }

    @Test
    public void testPojoWithRecursiveGenericField() {
        TypeInformation<?> ti = TypeExtractor.createTypeInfo(PojoWithRecursiveGenericField.class);
        Assert.assertTrue(ti instanceof PojoTypeInfo);
        Assert.assertEquals(
                GenericTypeInfo.class,
                ((PojoTypeInfo) ti).getPojoFieldAt(0).getTypeInformation().getClass());
    }

    public static class MutualPojoA {
        public MutualPojoB field;
    }

    public static class MutualPojoB {
        public MutualPojoA field;
    }

    @Test
    public void testPojosWithMutualRecursion() {
        TypeInformation<?> ti = TypeExtractor.createTypeInfo(MutualPojoB.class);
        Assert.assertTrue(ti instanceof PojoTypeInfo);
        TypeInformation<?> pti = ((PojoTypeInfo) ti).getPojoFieldAt(0).getTypeInformation();
        Assert.assertTrue(pti instanceof PojoTypeInfo);
        Assert.assertEquals(
                GenericTypeInfo.class,
                ((PojoTypeInfo) pti).getPojoFieldAt(0).getTypeInformation().getClass());
    }

    public static class Container<T> {
        public T field;
    }

    public static class MyType extends Container<Container<Object>> {}

    @Test
    public void testRecursivePojoWithTypeVariable() {
        TypeInformation<?> ti = TypeExtractor.createTypeInfo(MyType.class);
        Assert.assertTrue(ti instanceof PojoTypeInfo);
        TypeInformation<?> pti = ((PojoTypeInfo) ti).getPojoFieldAt(0).getTypeInformation();
        Assert.assertTrue(pti instanceof PojoTypeInfo);
        Assert.assertEquals(
                GenericTypeInfo.class,
                ((PojoTypeInfo) pti).getPojoFieldAt(0).getTypeInformation().getClass());
    }

    /** POJO generated using Lombok. */
    @Getter
    @Setter
    @NoArgsConstructor
    public static class TestLombok {
        private int age = 10;
        private String name;
    }

    @Test
    public void testLombokPojo() {
        TypeInformation<TestLombok> ti = TypeExtractor.getForClass(TestLombok.class);
        Assert.assertTrue(ti instanceof PojoTypeInfo);

        PojoTypeInfo<TestLombok> pti = (PojoTypeInfo<TestLombok>) ti;
        Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, pti.getTypeAt(0));
        Assert.assertEquals(BasicTypeInfo.STRING_TYPE_INFO, pti.getTypeAt(1));
    }
}
