protected void validateDynamic(ResolvedSchema resolvedSchema) {
        validateRowKey(resolvedSchema);
        List<Column> columns = resolvedSchema.getColumns();
        Preconditions.checkState(columns.size() == 2,
                "The number of elements must be 2.");
        Column column = columns.get(1);
        DataType dataType = column.getDataType();
        Preconditions.checkState(dataType instanceof KeyValueDataType, "The second element's Type must be Map");
        Preconditions.checkState(((KeyValueDataType) dataType).getKeyDataType().getConversionClass() == String.class,
                "The key type of second element(Map) must be String");
        Preconditions.checkState(((KeyValueDataType) dataType).getValueDataType().getConversionClass() == String.class,
                "The value type of second element(Map) must be String");
    }
The base class of {@link RedisSchema}

@param <T> The type of Flink state.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/schema/impl/AbstractRedisSchema.java