private static DynamicFormatter createFormatFunction(
            LogicalType indexFieldType,
            LogicalTypeRoot indexFieldLogicalTypeRoot,
            ZoneId localTimeZoneId) {
        switch (indexFieldLogicalTypeRoot) {
            case DATE:
                return (value, dateTimeFormatter) -> {
                    Integer indexField = (Integer) value;
                    return LocalDate.ofEpochDay(indexField).format(dateTimeFormatter);
                };
            case TIME_WITHOUT_TIME_ZONE:
                return (value, dateTimeFormatter) -> {
                    Integer indexField = (Integer) value;
                    return LocalTime.ofNanoOfDay(indexField * 1_000_000L).format(dateTimeFormatter);
                };
            case TIMESTAMP_WITHOUT_TIME_ZONE:
                return (value, dateTimeFormatter) -> {
                    TimestampData indexField = (TimestampData) value;
                    return indexField.toLocalDateTime().format(dateTimeFormatter);
                };
            case TIMESTAMP_WITH_TIME_ZONE:
                throw new UnsupportedOperationException(
                        "TIMESTAMP_WITH_TIME_ZONE is not supported yet");
            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
                return (value, dateTimeFormatter) -> {
                    TimestampData indexField = (TimestampData) value;
                    return indexField.toInstant().atZone(localTimeZoneId).format(dateTimeFormatter);
                };
            default:
                throw new TableException(
                        String.format(
                                "Unsupported type '%s' found in Elasticsearch dynamic index field, "
                                        + "time-related pattern only support types are: DATE,TIME,TIMESTAMP.",
                                indexFieldType));
        }
    }
Factory of {@link IndexGenerator}.

<p>Flink supports both static index and dynamic index.

<p>If you want to have a static index, this option value should be a plain string, e.g.
'myusers', all the records will be consistently written into "myusers" index.

<p>If you want to have a dynamic index, you can use '{field_name}' to reference a field value in
the record to dynamically generate a target index. You can also use
'{field_name|date_format_string}' to convert a field value of TIMESTAMP/DATE/TIME type into the
format specified by date_format_string. The date_format_string is compatible with {@link
java.text.SimpleDateFormat}. For example, if the option value is 'myusers_{log_ts|yyyy-MM-dd}',
then a record with log_ts field value 2020-03-27 12:25:55 will be written into
"myusers_2020-03-27" index.
inlong-sort/sort-flink/sort-flink-v1.18/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/IndexGeneratorFactory.java