@Override
    public StateEncoder<Tuple4<Boolean, String, Long, Boolean>> getStateEncoder() {
        return (in, serializationSchema) -> {
            boolean rowKind = getRowKind(in.getRowKind());
            GenericRowData row = (GenericRowData) in;
            String rowKey = row.getString(getKeyIndex()).toString();

            checkState(row.getArity() % 2 == 1,
                    "The number of elements must be odd, and all even elements are field.");

            return IntStream
                    .range(1, row.getArity() / 2 + 1)
                    .boxed()
                    .map(i -> {
                        int keyPos = 2 * i - 1;
                        long offset = row.getLong(keyPos);
                        boolean aBoolean = row.getBoolean(keyPos + 1);
                        return Tuple4.of(rowKind, rowKey, offset, aBoolean);
                    })
                    .collect(Collectors.toList());

        };
    }
The RedisSchema for Bitmap data-type and StaticKvPair mode.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/schema/impl/BitmapStaticKvPairSchema.java