NUAAwqy opened a new pull request, #150:
URL: https://github.com/apache/rocketmq-flink/pull/150
## What is the defect?
A row source whose table schema contains a `DATE`, `TIME` or `TIMESTAMP`
column delivers **no row at all**.
`RowDeserializationSchema` derives the value type of each field from its
conversion class
(`RowDeserializationSchema.java:113`):
```java
ByteSerializer.getTypeIndex(tableSchema.getFieldTypes()[index].getTypeClass())
```
A `DATE`, `TIME` or `TIMESTAMP` column is a `java.time.LocalDate`,
`java.time.LocalTime` or
`java.time.LocalDateTime`, and `ByteSerializer.getTypeIndex` maps those to
`V_LocalDate`, `V_LocalTime` and
`V_LocalDateTime` (`ByteSerializer.java:115-126`). The switch in
`StringSerializer.deserialize` had cases only for
the legacy `java.sql.Date` / `Time` / `Timestamp` values, which a
schema-driven source never produces, so every
temporal field fell through to `StringSerializer.java:142`:
```java
default:
throw new IllegalArgumentException();
```
That per-field exception is caught by `deserializeBytesMessage`
(`RowDeserializationSchema.java:232`) and routed to
the dirty-data strategy, which defaults to `SKIP`, so the row is dropped and
the job keeps running. The result is a
stream that connects, reports no error, and emits nothing.
The same `StringSerializer` backs the legacy
`RowKeyValueDeserializationSchema`, so both entry points are affected.
## What does this change?
Three cases are added to the switch, returning the internal representation a
`GenericRowData` expects:
| type | value | internal form |
|---|---|---|
| `V_LocalDate` | `java.time.LocalDate` | days since epoch (`int`) |
| `V_LocalTime` | `java.time.LocalTime` | milliseconds of the day (`int`) |
| `V_LocalDateTime` | `java.time.LocalDateTime` | `TimestampData` |
A timestamp written as a SQL literal (`2020-01-02 03:04:05.123`, the natural
CSV form) is normalised to
`2020-01-02T03:04:05.123` before parsing; ISO-8601 input is unchanged.
Leading/trailing blanks are trimmed.
`null` (and a configured null literal) still maps to `null`.
A value that cannot be parsed still throws, so the configured dirty-data
strategy keeps making the decision — this
change does not turn a malformed field into a row of garbage.
INTERVAL columns (`V_Duration`/`V_Period`) and `TIMESTAMP_LTZ` are
untouched: their textual form is ambiguous
(`TIMESTAMP_LTZ` is even `Unsupported` in `getTypeIndex`, since its
conversion class is `java.time.Instant`), so
they deserve their own change rather than a guess here.
## Tests
New `RowDeserializationSchemaTemporalTypeTest` (5 tests) builds a `STRING,
STRING, DATE, TIME, TIMESTAMP(3)` schema,
pushes a comma delimited body through
`RowDeserializationSchema.deserialize`, and asserts:
- the row is delivered and the `DATE` / `TIME` / `TIMESTAMP` values equal
the expected internal values;
- a two-line message delivers two rows (the row was dropped, not merely
mis-decoded);
- an unparsable date (`2020-13-45`) is still dropped by the dirty-data
strategy.
Against the pristine `StringSerializer`, the first four tests fail with
`Expected size: 1 but was: 0`; with the fix
all five pass. The whole module (`mvn -B -ntp -pl flink-connector-rocketmq
clean test`, JDK 11) is
`143 tests, 0 failures, 6 skipped`, and the CI command `mvn -B -ntp clean
package` succeeds; `spotless:check` is
clean on the JDK 11 leg.
No issue tracks this defect, so nothing is linked.
--
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]
