presto-kinesis/src/main/java/io/prestosql/plugin/kinesis/KinesisRecordSetProvider.java/KinesisRecordSetProvider/getRecordSet
class KinesisRecordSetProvider:    @Override
    public RecordSet getRecordSet(
            ConnectorTransactionHandle transactionHandle,
            ConnectorSession session,
            ConnectorSplit split,
            List<? extends ColumnHandle> columns)
    {
        KinesisSplit kinesisSplit = (KinesisSplit) split;
        List<KinesisColumnHandle> kinesisColumns = columns.stream()
                .map(x -> (KinesisColumnHandle) x)
                .collect(ImmutableList.toImmutableList());

        ImmutableList.Builder<KinesisColumnHandle> handleBuilder = ImmutableList.builder();

        RowDecoder messageDecoder = decoderFactory.create(
                kinesisSplit.getMessageDataFormat(),
                new HashMap<>(),
                kinesisColumns.stream()
                        .filter(column -> !column.isInternal())
                        .collect(toImmutableSet()));

        for (ColumnHandle handle : columns) {
            KinesisColumnHandle columnHandle = (KinesisColumnHandle) handle;
            handleBuilder.add(columnHandle);
        }
        return new KinesisRecordSet(kinesisSplit, session, clientManager, handleBuilder.build(), messageDecoder, kinesisConfig);
    }