wuchong commented on a change in pull request #12661:
URL: https://github.com/apache/flink/pull/12661#discussion_r441250789
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonOptions.java
##########
@@ -38,4 +44,40 @@
.defaultValue(false)
.withDescription("Optional flag to skip fields and rows
with parse errors instead of failing;\n"
+ "fields are set to null in case of
errors, false by default");
+
+ public static final ConfigOption<String> TIMESTAMP_FORMAT =
ConfigOptions
+ .key("timestamp-format.standard")
+ .stringType()
+ .defaultValue("SQL")
+ .withDescription("Optional flag to specify timestamp
format, SQL by default." +
+ " Option ISO-8601 will parse input timestamp in
\"yyyy-MM-ddTHH:mm:ss.s{precision}\" format and output timestamp in the same
format." +
+ " Option SQL will parse input timestamp in
\"yyyy-MM-dd HH:mm:ss.s{precision}\" format and output timestamp in the same
format.");
+
+ //
--------------------------------------------------------------------------------------------
+ // Option enumerations
+ //
--------------------------------------------------------------------------------------------
+
+ public static final String SQL = "SQL";
+ public static final String ISO_8601 = "ISO-8601";
+
+ public static final Set<String> TIMESTAMP_FORMAT_ENUM = new
HashSet<>(Arrays.asList(
+ SQL,
+ ISO_8601
+ ));
+
+ //
--------------------------------------------------------------------------------------------
+ // Utilities
+ //
--------------------------------------------------------------------------------------------
+
+ public static TimestampFormat getTimestampFormat(ReadableConfig config){
+ String timestampFormat = config.get(TIMESTAMP_FORMAT);
+ switch (timestampFormat){
+ case SQL:
+ return TimestampFormat.SQL;
+ case ISO_8601:
+ return TimestampFormat.ISO_8601;
+ }
+ throw new TableException(
Review comment:
Put this exception in the `default` branch of swith.
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonFormatFactory.java
##########
@@ -48,12 +49,9 @@
public static final String IDENTIFIER = "canal-json";
- public static final ConfigOption<Boolean> IGNORE_PARSE_ERRORS =
ConfigOptions
- .key("ignore-parse-errors")
- .booleanType()
- .defaultValue(false)
- .withDescription("Optional flag to skip fields and rows with
parse errors instead of failing, " +
- "fields are set to null in case of errors. Default is
false.");
+ public static final ConfigOption<Boolean> IGNORE_PARSE_ERRORS =
JsonOptions.IGNORE_PARSE_ERRORS;
+
+ public static final ConfigOption<String> TIMESTAMP_FORMAT =
JsonOptions.TIMESTAMP_FORMAT;
Review comment:
Add this into `optionalOptions()`.
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataSerializationSchema.java
##########
@@ -72,8 +74,12 @@
/** Reusable object node. */
private transient ObjectNode node;
- public JsonRowDataSerializationSchema(RowType rowType) {
+ /** Option for timestamp format. */
Review comment:
```suggestion
/** Timestamp format specification which is used to generate timestamp.
*/
```
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonFormatFactory.java
##########
@@ -57,12 +59,9 @@
"This option indicates the Debezium JSON data include
the schema in the message or not. " +
"Default is false.");
- public static final ConfigOption<Boolean> IGNORE_PARSE_ERRORS =
ConfigOptions
- .key("ignore-parse-errors")
- .booleanType()
- .defaultValue(false)
- .withDescription("Optional flag to skip fields and rows with
parse errors instead of failing, " +
- "fields are set to null in case of errors. Default is
false.");
+ public static final ConfigOption<Boolean> IGNORE_PARSE_ERRORS =
JsonOptions.IGNORE_PARSE_ERRORS;
+
+ public static final ConfigOption<String> TIMESTAMP_FORMAT =
JsonOptions.TIMESTAMP_FORMAT;
Review comment:
Add this into `optionalOptions()`.
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFileSystemFormatFactory.java
##########
@@ -79,13 +79,15 @@ public String factoryIdentifier() {
validateFormatOptions(options);
boolean failOnMissingField = options.get(FAIL_ON_MISSING_FIELD);
boolean ignoreParseErrors = options.get(IGNORE_PARSE_ERRORS);
+ TimestampFormat timestampOption =
JsonOptions.getTimestampFormat(options);
Review comment:
Add `TIMESTAMP_FORMAT` into `optionalOptions`.
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java
##########
@@ -93,11 +94,15 @@
/** Object mapper for parsing the JSON. */
private final ObjectMapper objectMapper = new ObjectMapper();
+ /** Options for timestamp format. */
+ private final TimestampFormat timestampFormat;
Review comment:
Add this into `equals` and `hashcode`.
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java
##########
@@ -93,11 +94,15 @@
/** Object mapper for parsing the JSON. */
private final ObjectMapper objectMapper = new ObjectMapper();
+ /** Options for timestamp format. */
Review comment:
```suggestion
/** Timestamp format specification which is used to parse timestamp. */
```
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java
##########
@@ -68,9 +69,10 @@
private final int fieldCount;
public CanalJsonDeserializationSchema(
- RowType rowType,
- TypeInformation<RowData> resultTypeInfo,
- boolean ignoreParseErrors) {
+ RowType rowType,
+ TypeInformation<RowData> resultTypeInfo,
+ boolean ignoreParseErrors,
+ TimestampFormat timestampFormatOption) {
Review comment:
Keep the original indent.
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java
##########
@@ -75,10 +76,11 @@
private final boolean ignoreParseErrors;
public DebeziumJsonDeserializationSchema(
- RowType rowType,
- TypeInformation<RowData> resultTypeInfo,
- boolean schemaInclude,
- boolean ignoreParseErrors) {
+ RowType rowType,
+ TypeInformation<RowData> resultTypeInfo,
+ boolean schemaInclude,
+ boolean ignoreParseErrors,
+ TimestampFormat timestampFormatOption) {
Review comment:
Keep the original indent.
##########
File path:
flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/debezium/DebeziumJsonFormatFactoryTest.java
##########
@@ -66,7 +67,8 @@ public void testSeDeSchema() {
ROW_TYPE,
new RowDataTypeInfo(ROW_TYPE),
true,
- true);
+ true,
+ TimestampFormat.ISO_8601);
Review comment:
Add the timestamp format option into `getAllOptions`.
```java
options.put("debezium-json.timestamp-format.standard", "ISO-8081");
```
##########
File path:
flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java
##########
@@ -91,6 +91,16 @@ public void testInvalidOptionForIgnoreParseErrors() {
testSchemaDeserializationSchema(tableOptions);
}
+ @Test
+ public void testInvalidOptionForTimestampFormat() {
Review comment:
Add an additional test for lowercase `iso-8081`.
##########
File path:
flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/canal/CanalJsonFormatFactoryTest.java
##########
@@ -65,7 +66,8 @@ public void testSeDeSchema() {
final CanalJsonDeserializationSchema expectedDeser = new
CanalJsonDeserializationSchema(
ROW_TYPE,
new RowDataTypeInfo(ROW_TYPE),
- true);
+ true,
+ TimestampFormat.ISO_8601);
Review comment:
Add the timestamp format option into `getAllOptions`.
```java
options.put("canal-json.timestamp-format.standard", "ISO-8081");
```
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataSerializationSchema.java
##########
@@ -72,8 +74,12 @@
/** Reusable object node. */
private transient ObjectNode node;
- public JsonRowDataSerializationSchema(RowType rowType) {
+ /** Option for timestamp format. */
+ private final TimestampFormat timestampFormat;
Review comment:
Add this into `equals` and `hashcode`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
