public void validate(RedisCommand redisCommand, ResolvedSchema tableSchema) {
        LogicalType[] logicalTypes = tableSchema.getColumns().stream().filter(Column::isPhysical)
                .map(s -> s.getDataType().getLogicalType()).toArray(LogicalType[]::new);
        LogicalTypeRoot[] requiredLogicalTypes = schemaMap.get(redisCommand);
        for (int i = 0; i < requiredLogicalTypes.length; i++) {
            Preconditions.checkState(requiredLogicalTypes[i] == logicalTypes[i].getTypeRoot(),
                    "Table schema " + Arrays.deepToString(logicalTypes) + " is invalid. Table schema "
                            + Arrays.deepToString(requiredLogicalTypes) + " is required for command " + redisCommand
                                    .name());
        }
    }
Validate the schema

@param redisCommand The redis command
@param tableSchema The table schema
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/table/SchemaValidator.java