@Override
    public StateEncoder<Tuple4<Boolean, String, String, String>> getStateEncoder() {
        return (in, serializationSchema) -> {
            boolean rowKind = getRowKind(in.getRowKind());

            GenericRowData row = (GenericRowData) in;
            String rowKey = row.getString(getKeyIndex()).toString();
            MapData fieldMap = row.getMap(getValueIndex());

            ArrayData keyArray = fieldMap.keyArray();
            ArrayData valueMap = fieldMap.valueArray();
            return IntStream
                    .range(0, fieldMap.size())
                    .boxed()
                    .map(i -> {
                        String fieldName = keyArray.getString(i).toString();
                        byte[] binary = valueMap.getBinary(i);
                        String fieldValue = StringData.fromBytes(binary).toString();
                        return Tuple4.of(rowKind, rowKey, fieldName, fieldValue);
                    })
                    .collect(Collectors.toList());
        };
    }
The RedisSchema for Hash data-type and Dynamic mode.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/schema/impl/HashDynamicSchema.java