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 testFieldTypes() {
        CsvReader reader = getCsvReader();
        DataSource<Item> items = reader.tupleType(Item.class);

        TypeInformation<?> info = items.getType();
        if (!info.isTupleType()) {
            fail("");
        } else {
            TupleTypeInfo<?> tinfo = (TupleTypeInfo<?>) 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)).isEqualTo(BasicTypeInfo.STRING_TYPE_INFO);
        }

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