public static void validatePrimaryKey(TableSchema schema) {
        schema.getPrimaryKey()
                .ifPresent(
                        key -> {
                            List<LogicalTypeRoot> illegalTypes =
                                    key.getColumns().stream()
                                            .map(
                                                    fieldName -> {
                                                        LogicalType logicalType =
                                                                schema.getFieldDataType(fieldName)
                                                                        .get()
                                                                        .getLogicalType();
                                                        if (logicalType.is(
                                                                LogicalTypeRoot.DISTINCT_TYPE)) {
                                                            return ((DistinctType) logicalType)
                                                                    .getSourceType()
                                                                    .getTypeRoot();
                                                        } else {
                                                            return logicalType.getTypeRoot();
                                                        }
                                                    })
                                            .filter(ILLEGAL_PRIMARY_KEY_TYPES::contains)
                                            .collect(Collectors.toList());

                            if (!illegalTypes.isEmpty()) {
                                throw new ValidationException(
                                        String.format(
                                                "The table has a primary key on columns of illegal types: %s.\n"
                                                        + " Elasticsearch sink does not support primary keys on columns of types: %s.",
                                                illegalTypes, ILLEGAL_PRIMARY_KEY_TYPES));
                            }
                        });
    }
Checks that the table does not have primary key defined on illegal types. In Elasticsearch
the primary key is used to calculate the Elasticsearch document id, which is a string of up
to 512 bytes. It cannot have whitespaces. As of now it is calculated by concatenating the
fields. Certain types do not have a good string representation to be used in this scenario.
The illegal types are mostly {@link LogicalTypeFamily#COLLECTION} types and {@link
LogicalTypeRoot#RAW} type.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/ElasticsearchValidationUtils.java