public <T extends Tuple> DataSource<T> tupleType(Class<T> targetType) {
        Preconditions.checkNotNull(targetType, "The target type class must not be null.");
        if (!Tuple.class.isAssignableFrom(targetType)) {
            throw new IllegalArgumentException(
                    "The target type must be a subclass of " + Tuple.class.getName());
        }

        @SuppressWarnings("unchecked")
        TupleTypeInfo<T> typeInfo = (TupleTypeInfo<T>) TypeExtractor.createTypeInfo(targetType);
        CsvInputFormat<T> inputFormat =
                new TupleCsvInputFormat<T>(
                        path, this.lineDelimiter, this.fieldDelimiter, typeInfo, this.includedMask);

        Class<?>[] classes = new Class<?>[typeInfo.getArity()];
        for (int i = 0; i < typeInfo.getArity(); i++) {
            classes[i] = typeInfo.getTypeAt(i).getTypeClass();
        }

        configureInputFormat(inputFormat);
        return new DataSource<T>(
                executionContext, inputFormat, typeInfo, Utils.getCallLocationName());
    }
@Test
    void testSubClassWithPartialsInHierarchie() {
        CsvReader reader = getCsvReader();
        DataSource<FinalItem> sitems = reader.tupleType(FinalItem.class);
        TypeInformation<?> info = sitems.getType();

        assertThat(info.isTupleType()).isTrue();
        assertThat(info.getTypeClass()).isEqualTo(FinalItem.class);

        @SuppressWarnings("unchecked")
        TupleTypeInfo<SubItem> tinfo = (TupleTypeInfo<SubItem>) info;

        assertThat(tinfo.getTypeAt(0)).isEqualTo(BasicTypeInfo.INT_TYPE_INFO);
        assertThat(tinfo.getTypeAt(1)).isEqualTo(BasicTypeInfo.STRING_TYPE_INFO);
        assertThat(tinfo.getTypeAt(2)).isEqualTo(BasicTypeInfo.DOUBLE_TYPE_INFO);
        assertThat(tinfo.getTypeAt(3).getClass()).isEqualTo(ValueTypeInfo.class);
        assertThat(tinfo.getTypeAt(4).getClass()).isEqualTo(ValueTypeInfo.class);
        assertThat((tinfo.getTypeAt(3)).getTypeClass()).isEqualTo(StringValue.class);
        assertThat((tinfo.getTypeAt(4)).getTypeClass()).isEqualTo(LongValue.class);

        CsvInputFormat<?> inputFormat = (CsvInputFormat<?>) sitems.getInputFormat();
        assertThat(inputFormat.getFieldTypes())
                .containsExactly(
                        Integer.class,
                        String.class,
                        Double.class,
                        StringValue.class,
                        LongValue.class);
    }