private static IndexGenerator createRuntimeIndexGenerator(
			String index,
			String[] fieldNames,
			TypeInformation<?>[] typeInformatics,
			IndexHelper indexHelper) {
		final String dynamicIndexPatternStr = indexHelper.extractDynamicIndexPatternStr(index);
		final String indexPrefix = index.substring(0, index.indexOf(dynamicIndexPatternStr));
		final String indexSuffix = index.substring(indexPrefix.length() + dynamicIndexPatternStr.length());

		final boolean isDynamicIndexWithFormat = indexHelper.checkIsDynamicIndexWithFormat(index);
		final int indexFieldPos = indexHelper.extractIndexFieldPos(index, fieldNames, isDynamicIndexWithFormat);
		final TypeInformation<?> indexFieldType = typeInformatics[indexFieldPos];

		// validate index field type
		indexHelper.validateIndexFieldType(indexFieldType);

		// time extract dynamic index pattern
		if (isDynamicIndexWithFormat) {
			final String dateTimeFormat = indexHelper.extractDateFormat(index, indexFieldType);
			// DataTypes.SQL_TIMESTAMP
			if (indexFieldType.equals(Types.LOCAL_DATE_TIME)) {
				return new AbstractTimeIndexGenerator(index, dateTimeFormat) {
					@Override
					public String generate(Row row) {
						LocalDateTime indexField = (LocalDateTime) row.getField(indexFieldPos);
						String indexFieldValueStr = indexField.format(dateTimeFormatter);
						return indexPrefix.concat(indexFieldValueStr).concat(indexSuffix);
					}
				};
			}
			else if (indexFieldType.equals(Types.SQL_TIMESTAMP)) {
				return new AbstractTimeIndexGenerator(index, dateTimeFormat) {
					@Override
					public String generate(Row row) {
						Timestamp indexField = (Timestamp) row.getField(indexFieldPos);
						String indexFieldValueStr = indexField.toLocalDateTime().format(dateTimeFormatter);
						return indexPrefix.concat(indexFieldValueStr).concat(indexSuffix);
					}
				};
			}
			// DataTypes.SQL_DATE
			else if (indexFieldType.equals(Types.LOCAL_DATE)) {
				return new AbstractTimeIndexGenerator(index, dateTimeFormat) {
					@Override
					public String generate(Row row) {
						LocalDate indexField = (LocalDate) row.getField(indexFieldPos);
						String indexFieldValueStr = indexField.format(dateTimeFormatter);
						return indexPrefix.concat(indexFieldValueStr).concat(indexSuffix);
					}
				};
			} else if (indexFieldType.equals(Types.SQL_DATE)) {
				return new AbstractTimeIndexGenerator(index, dateTimeFormat) {
					@Override
					public String generate(Row row) {
						Date indexField = (Date) row.getField(indexFieldPos);
						String indexFieldValueStr = indexField.toLocalDate().format(dateTimeFormatter);
						return indexPrefix.concat(indexFieldValueStr).concat(indexSuffix);
					}
				};
			} // DataTypes.TIME
			else if (indexFieldType.equals(Types.LOCAL_TIME)) {
				return new AbstractTimeIndexGenerator(index, dateTimeFormat) {
					@Override
					public String generate(Row row) {
						LocalTime indexField = (LocalTime) row.getField(indexFieldPos);
						String indexFieldValueStr = indexField.format(dateTimeFormatter);
						return indexPrefix.concat(indexFieldValueStr).concat(indexSuffix);
					}
				};
			} else if (indexFieldType.equals(Types.SQL_TIME)) {
				return new AbstractTimeIndexGenerator(index, dateTimeFormat) {
					@Override
					public String generate(Row row) {
						Time indexField = (Time) row.getField(indexFieldPos);
						String indexFieldValueStr = indexField.toLocalTime().format(dateTimeFormatter);
						return indexPrefix.concat(indexFieldValueStr).concat(indexSuffix);
					}
				};
			} else {
				throw new TableException(String.format("Unsupported type '%s' found in Elasticsearch dynamic index field, " +
					"time-related pattern only support types are: DATE,TIME,TIMESTAMP.",
					TypeConversions.fromLegacyInfoToDataType(indexFieldType)));
			}
		}
		// general dynamic index pattern
		return new IndexGeneratorBase(index) {
			@Override
			public String generate(Row row) {
				Object indexField = row.getField(indexFieldPos);
				Preconditions.checkNotNull(indexField, "Index field's value is not allow null.");
				return indexPrefix.concat(indexField.toString()).concat(indexSuffix);
			}
		};
	}
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.
elasticsearch7/elasticsearch7-sink/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/index/IndexGeneratorFactory.java