class StreamSerializer implements Serializable {

    public Object[] getRow(T input) {
        Preconditions.checkArgument(input.getClass() == schema.getTypeInfo().getTypeClass(),
            "Invalid input type: " + input + ", expected: " + schema.getTypeInfo());

        Object[] data;
        if (schema.isAtomicType()) {
            data = new Object[]{input};
        } else if (schema.isTupleType()) {
            Tuple tuple = (Tuple) input;
            data = new Object[schema.getFieldIndexes().length];
            for (int i = 0; i < schema.getFieldIndexes().length; i++) {
                data[i] = tuple.getField(schema.getFieldIndexes()[i]);
            }
        } else if (schema.isRowType()) {
            Row row = (Row) input;
            data = new Object[schema.getFieldIndexes().length];
            for (int i = 0; i < schema.getFieldIndexes().length; i++) {
                data[i] = row.getField(schema.getFieldIndexes()[i]);
            }
        } else if (schema.isPojoType() || schema.isCaseClassType()) {
            data = new Object[schema.getFieldIndexes().length];
            for (int i = 0; i < schema.getFieldNames().length; i++) {
                data[i] = getFieldValue(schema.getFieldNames()[i], input);
            }
        } else {
            throw new IllegalArgumentException("Failed to get field values from " + schema.getTypeInfo());
        }
        return data;
    }

    public  StreamSerializer(StreamSchema<T> schema);

}