leonardBang commented on a change in pull request #18058:
URL: https://github.com/apache/flink/pull/18058#discussion_r772819208
##########
File path:
flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/table/IndexGeneratorFactory.java
##########
@@ -176,6 +207,8 @@ private static DynamicFormatter createFormatFunction(
private static final Pattern dynamicIndexPattern =
Pattern.compile("\\{[^\\{\\}]+\\}?");
private static final Pattern dynamicIndexTimeExtractPattern =
Pattern.compile(".*\\{.+\\|.*\\}.*");
+ private static final Pattern dynamicIndexProcTimeExtractPattern =
+ Pattern.compile(".*\\{now\\(\\)\\|.*\\}.*");
Review comment:
I think following function should also works, right?
`now()`
`NOW()`
`current_timestamp`
`CURRENT_TIMESTAMP`
##########
File path:
flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/table/IndexGeneratorFactory.java
##########
@@ -240,6 +273,11 @@ boolean checkIsDynamicIndexWithFormat(String index) {
return dynamicIndexTimeExtractPattern.matcher(index).matches();
}
+ /** Check generate a dynamic index is based on processing time. */
Review comment:
```suggestion
/** Check generate dynamic index is from system time or not. */
```
##########
File path:
flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/table/IndexGeneratorFactory.java
##########
@@ -240,6 +273,11 @@ boolean checkIsDynamicIndexWithFormat(String index) {
return dynamicIndexTimeExtractPattern.matcher(index).matches();
}
+ /** Check generate a dynamic index is based on processing time. */
+ boolean checkIsDynamicIndexWithProcTimeFormat(String index) {
Review comment:
Hi, @hackergin @alpreu @matriv , Thanks for your nice work.
I want to talk more about the diff between `proctime()` and `now()`, they
have different return type as well as the meaning in Flink SQL, the first one
is `TIMESTAMP_LTZ *PROCTIME*`， the later one is `TIMESTAMP_LTZ(3)`, I think the
PR description is improper with our current implementation. I suggest changing
the issue/PR description to that we supports a system time function(`now()` and
`current_timestamp`) in index pattern, this should be more clear and won't
mislead users. WDYT?
##########
File path:
flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/table/IndexGeneratorFactory.java
##########
@@ -63,30 +65,59 @@
private IndexGeneratorFactory() {}
public static IndexGenerator createIndexGenerator(
- String index, List<String> fieldNames, List<DataType> dataTypes) {
+ String index,
+ List<String> fieldNames,
+ List<DataType> dataTypes,
+ ZoneId localTimeZoneId) {
final IndexHelper indexHelper = new IndexHelper();
if (indexHelper.checkIsDynamicIndex(index)) {
return createRuntimeIndexGenerator(
index,
fieldNames.toArray(new String[0]),
dataTypes.toArray(new DataType[0]),
- indexHelper);
+ indexHelper,
+ localTimeZoneId);
} else {
return new StaticIndexGenerator(index);
}
}
+ public static IndexGenerator createIndexGenerator(
+ String index, List<String> fieldNames, List<DataType> dataTypes) {
+ return createIndexGenerator(index, fieldNames, dataTypes,
ZoneId.systemDefault());
+ }
+
interface DynamicFormatter extends Serializable {
String format(@Nonnull Object fieldValue, DateTimeFormatter formatter);
}
private static IndexGenerator createRuntimeIndexGenerator(
- String index, String[] fieldNames, DataType[] fieldTypes,
IndexHelper indexHelper) {
+ String index,
+ String[] fieldNames,
+ DataType[] fieldTypes,
+ IndexHelper indexHelper,
+ ZoneId localTimeZoneId) {
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 isDynamicIndexWithProcTimeFormat =
+ indexHelper.checkIsDynamicIndexWithProcTimeFormat(index);
+ if (isDynamicIndexWithProcTimeFormat) {
Review comment:
```suggestion
if (indexHelper.checkIsDynamicIndexWithProcTimeFormat(index)) {
```
##########
File path:
flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/table/IndexGeneratorFactory.java
##########
@@ -240,6 +273,11 @@ boolean checkIsDynamicIndexWithFormat(String index) {
return dynamicIndexTimeExtractPattern.matcher(index).matches();
}
+ /** Check generate a dynamic index is based on processing time. */
+ boolean checkIsDynamicIndexWithProcTimeFormat(String index) {
Review comment:
```suggestion
boolean checkIsDynamicIndexFromSystemTimeFormat(String index) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
