feat(source): `encode json` option `timestamptz.handling.mode` #16265

xiangjinwu · 2024-04-11T16:41:57Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Similar to sink, we allow specifying timestamptz number units:

micro
milli
guess_number_unit: This has been the default, but limits the range to [1973-03-03 09:46:40, 5138-11-16 09:46:40)
utc_string: This format is least ambiguous and can mostly be inferred correctly without the need to be specified.
utc_without_suffix: This allows the user to acknowledge naive timestamp is in UTC rather than local time.

Note the corresponding format has to be one of plain | upsert | debezium. The following formats do NOT accept this option as of now:

debezium_mongo
canal
maxwell

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added test labels as necessary. See details.
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

https://docs.risingwave.com/docs/current/supported-sources-and-formats/#debezium-json

When using format plain | upsert | debezium encode json (but not format debezium_mongo | canal | maxwell encode json), there is a new option timestamptz.handling.mode. There are 5 possible choices. See description above for the meaning of each one.

…earlier than 1974

xiangjinwu · 2024-04-11T17:27:41Z

src/connector/src/parser/plain_parser.rs

@@ -66,7 +67,7 @@ impl PlainParser {
 };

 let transaction_meta_builder = Some(AccessBuilderImpl::DebeziumJson(
- DebeziumJsonAccessBuilder::new()?,
+ DebeziumJsonAccessBuilder::new(TimestamptzHandling::GuessNumberUnit)?,


@StrikeW Why do we have DebeziumJsonAccessBuilder as part of PlainParser?

Shared cdc source use the plain parser, and to parse transaction metadata, we need a Debezium parser.

Given it is system metadata I believe [1973-03-03 09:46:40, 5138-11-16 09:46:40) is enough. Additionally, the structure of the metadata may be static rather than dynamic, so maybe we can derive(serde::Deserialize) rather than accessing a json Value. But this can be a separate issue.

xiangjinwu · 2024-04-11T17:31:08Z

src/stream/src/executor/backfill/cdc/cdc_backfill.rs

@@ -607,6 +607,7 @@ pub async fn transform_upstream(upstream: BoxedMessageStream, schema: &Schema) {
 key_encoding_config: None,
 encoding_config: EncodingProperties::Json(JsonProperties {
 use_schema_registry: false,
+ timestamptz_handling: None,


@StrikeW Is it possible timestamptz here is always one of micro/milli/utc_string? Being specific can read better than the legacy default guess_number_unit, but if you are unsure now, at least the current implementation is semantically equivalent as previous.

I think it is possible. But we need to implement a CustomConverter to enforce that. The default behavior is up to the precision of upstream data types.

I see. So this code path would remain limited by [1973-03-03 09:46:40, 5138-11-16 09:46:40).

tabVersion · 2024-04-11T17:40:33Z

The approach is LGTM since the guessing one can be wrong sometimes. Just wonder what we do if the schema contains multiple timestamptz fields and they have different units? 😇

xiangjinwu · 2024-04-12T02:31:37Z

The approach is LGTM since the guessing one can be wrong sometimes. Just wonder what we do if the schema contains multiple timestamptz fields and they have different units? 😇

When there are multiple timestamptz fields with different units, a per-column schema provided by kafka connect converters (json, json schema, avro, protobuf) would be necessary. The latter 3 options requires usage of schema registry, while the first json option would embed schema into every message, which is costly and could be disabled with schemas.enable=false. So the user has the following choice:

Use avro/protobuf/json schema with schema registry
Use json with schemas.enable=true, without schema registry but every message would be large
Use json with schemas.enable=false but all timestamptz fields are required to be the same units

See #16097 (comment) for a concrete example of json when schemas.enable is not set to false by user.

This PR supports the 3rd use case and we would support the former 2 kafka connect schema when needed.

e2e_test/source/basic/handling_mode.slt

StrikeW · 2024-04-12T06:30:22Z

src/connector/src/parser/plain_parser.rs

@@ -66,7 +67,7 @@ impl PlainParser {
 };

 let transaction_meta_builder = Some(AccessBuilderImpl::DebeziumJson(
- DebeziumJsonAccessBuilder::new()?,
+ DebeziumJsonAccessBuilder::new(TimestamptzHandling::GuessNumberUnit)?,


Shared cdc source use the plain parser, and to parse transaction metadata, we need a Debezium parser.

StrikeW · 2024-04-12T07:02:05Z

src/stream/src/executor/backfill/cdc/cdc_backfill.rs

@@ -607,6 +607,7 @@ pub async fn transform_upstream(upstream: BoxedMessageStream, schema: &Schema) {
 key_encoding_config: None,
 encoding_config: EncodingProperties::Json(JsonProperties {
 use_schema_registry: false,
+ timestamptz_handling: None,


I think it is possible. But we need to implement a CustomConverter to enforce that. The default behavior is up to the precision of upstream data types.

tabVersion

LGTM, do we have plan on publishing other params in JsonParseOptions

xiangjinwu · 2024-04-12T08:47:35Z

LGTM, do we have plan on publishing other params in JsonParseOptions

Yes. And unifying them with sink ones (similar yet with minor differences). But it does seem timestamptz control is the most popular one.

… (#16281) Co-authored-by: xiangjinwu <[email protected]>

xiangjinwu added 3 commits April 11, 2024 23:45

feat(source): FORMAT DEBEZIUM ENCODE JSON accept timestamptz units …

a961523

…earlier than 1974

plain and upsert (but not canal or maxwell)

8fab61b

e2e tests

ec5ae30

xiangjinwu added the need-cherry-pick-release-1.8 label Apr 11, 2024

github-actions bot added the type/feature label Apr 11, 2024

xiangjinwu requested review from StrikeW and tabVersion April 11, 2024 17:26

xiangjinwu commented Apr 11, 2024

View reviewed changes

StrikeW reviewed Apr 12, 2024

View reviewed changes

tabVersion approved these changes Apr 12, 2024

View reviewed changes

xiangjinwu added this pull request to the merge queue Apr 12, 2024

StrikeW approved these changes Apr 12, 2024

View reviewed changes

Merged via the queue into main with commit af175db Apr 12, 2024
37 of 38 checks passed

xiangjinwu deleted the feat-source-json-timestamptz-mode branch April 12, 2024 09:13

neverchanje mentioned this pull request Apr 12, 2024

Document: feat(source): encode json option timestamptz.handling.mode risingwavelabs/risingwave-docs#2064

Closed

github-actions bot pushed a commit that referenced this pull request Apr 12, 2024

feat(source): encode json option timestamptz.handling.mode (#16265)

b7e9e46

github-actions bot mentioned this pull request Apr 12, 2024

feat(source): encode json option timestamptz.handling.mode (#16265) #16281

Merged

github-merge-queue bot pushed a commit that referenced this pull request Apr 12, 2024

feat(source): encode json option timestamptz.handling.mode (#16265)…

2c18d64

… (#16281) Co-authored-by: xiangjinwu <[email protected]>

xiangjinwu mentioned this pull request Apr 16, 2024

bug: panic when UNIX epoch time out of i64 range #16094

Closed

ShanlanLi mentioned this pull request May 15, 2024

Add timestamptz.handling.mode to supported formats page risingwavelabs/risingwave-docs#2175

Merged

2 tasks

BugenZhao mentioned this pull request May 21, 2024

risingwave 1.9.1 risingwavelabs/homebrew-risingwave#38

Closed

1 task

fuyufjh mentioned this pull request May 28, 2024

feat(source): parse avro map into jsonb #16948

Merged

10 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(source): `encode json` option `timestamptz.handling.mode` #16265

feat(source): `encode json` option `timestamptz.handling.mode` #16265

xiangjinwu commented Apr 11, 2024 •

edited

Loading

xiangjinwu Apr 11, 2024

StrikeW Apr 12, 2024

xiangjinwu Apr 12, 2024

xiangjinwu Apr 11, 2024

StrikeW Apr 12, 2024

xiangjinwu Apr 12, 2024

tabVersion commented Apr 11, 2024

xiangjinwu commented Apr 12, 2024 •

edited

Loading

StrikeW Apr 12, 2024

StrikeW Apr 12, 2024

tabVersion left a comment

xiangjinwu commented Apr 12, 2024

feat(source): encode json option timestamptz.handling.mode #16265

feat(source): encode json option timestamptz.handling.mode #16265

Conversation

xiangjinwu commented Apr 11, 2024 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tabVersion commented Apr 11, 2024

xiangjinwu commented Apr 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tabVersion left a comment

Choose a reason for hiding this comment

xiangjinwu commented Apr 12, 2024

feat(source): `encode json` option `timestamptz.handling.mode` #16265

feat(source): `encode json` option `timestamptz.handling.mode` #16265

xiangjinwu commented Apr 11, 2024 •

edited

Loading

xiangjinwu commented Apr 12, 2024 •

edited

Loading