bug: wrong parsed timestamp from json #16097

MrCroxx · 2024-04-02T16:18:08Z

Describe the bug

When parsing json value to timestamp:

{
	"before": null,
	"after": {
	    "id": 13,
        "example_timestamp_tz": 63072000000000000
    },
	"source": {
		"table": "pathmetadata"
	},
	"op": "c",
	"ts_ms": 1707225270445
}

Value 63072000000000000 should be parsed as 1972-01-01 00:00:00. But RW parses it as 3968-09-03 02:00:00.000 +0200, which is 63072007200.

Error message/log

No response

To Reproduce

example json:

{
	"before": null,
	"after": {
	    "id": 13,
        "example_timestamp_tz": 63072000000000000
    },
	"source": {
		"table": "pathmetadata"
	},
	"op": "c",
	"ts_ms": 1707225270445
}

Expected behavior

Value 63072000000000000 should be parsed as 1972-01-01 00:00:00.

How did you deploy RisingWave?

No response

The version of RisingWave

No response

Additional context

No response

The text was updated successfully, but these errors were encountered:

tabVersion · 2024-04-02T16:57:57Z

seems the time unit inference gets this wrong, cc @xiangjinwu

risingwave/src/common/src/cast/mod.rs

Lines 60 to 70 in a285394

    
           pub fn i64_to_timestamptz(t: i64) -> Result<Timestamptz> { 
        
               const E11: u64 = 100_000_000_000; 
        
               const E14: u64 = 100_000_000_000_000; 
        
               const E17: u64 = 100_000_000_000_000_000; 
        
               match t.abs_diff(0) { 
        
                   0..E11 => Ok(Timestamptz::from_secs(t).unwrap()), // s 
        
                   E11..E14 => Ok(Timestamptz::from_millis(t).unwrap()), // ms 
        
                   E14..E17 => Ok(Timestamptz::from_micros(t)),      // us 
        
                   E17.. => Ok(Timestamptz::from_micros(t / 1000)),  // ns 
        
               } 
        
           }

xiangjinwu · 2024-04-03T04:23:43Z

To repeat myself one more time, this is by design and documented:

https://docs.risingwave.com/docs/current/sql-create-source/#debezium-json

Note that if you are ingesting data of type timestamp or timestamptz in RisingWave, the upstream value must be in the range of [1973-03-03 09:46:40, 5138-11-16 09:46:40] (UTC). The value may be parsed and ingested incorrectly without warning.

@MrCroxx Could you provide more context on this specific use case? How is the source created and how do we know if the unit is s / ms / us / ns?

Value 63072000000000000 should be parsed as 1970-01-01 00:00:00. But RW parses it as 3968-09-03 02:00:00.000 +0200, which is 63072007200.

This claim is problematic as well. How could a later date have a smaller value?

~~@MrCroxx Please correct the expected behavior before we can proceed.~~ (fixed. see comment below)

xiangjinwu · 2024-04-03T04:40:51Z

Sync'ed offline:

63072000000000000 in this case is expected to be 1972-01-01T00:00:00Z (in ns, not 1970) but is parsed as 3968-09-03T00:00:00Z (in us).

fuyufjh · 2024-04-03T06:26:23Z

seems the time unit inference gets this wrong, cc @xiangjinwu

risingwave/src/common/src/cast/mod.rs

Lines 60 to 70 in a285394

pub fn i64_to_timestamptz(t: i64) -> Result<Timestamptz> {

const E11: u64 = 100_000_000_000;

const E14: u64 = 100_000_000_000_000;

const E17: u64 = 100_000_000_000_000_000;

match t.abs_diff(0) {

0..E11 => Ok(Timestamptz::from_secs(t).unwrap()), // s

E11..E14 => Ok(Timestamptz::from_millis(t).unwrap()), // ms

E14..E17 => Ok(Timestamptz::from_micros(t)), // us

E17.. => Ok(Timestamptz::from_micros(t / 1000)), // ns

}

}

This solution looks acceptable, but is too general for some cases, for example

If it's using the direct CDC, we can ensure the format must be milliseconds (after fix(direct cdc): use time.precision.mode=connect #16119)
If users specify a parsing behavior manually via some WITH settings, we should follow that.

xiangjinwu · 2024-04-03T08:48:15Z

Update:

When consuming from debezium, the unit info is actually available as part of the schema:

If using avro, it is in the connect.name extension field. Right now we are ignoring this field, and there is an existing issue for it Add support for Avro schemas using some common Kafka Connect schema names/types #14703
If using json, there is also schema per message. It would appear as name of a type (not to be confused as name of a field). See the example below. We are also ignoring this name field in source right now. And only added partial support for it in the sink part recently kafka-cdc-sink: Missing name field in debezium json #13450
Having schema per json message is costly. So debezium does recommend avro or allow disabling schema in json. In the latter case unit info would be unavailable, and we would end up with the same problem here. There is also an existing sink issue requesting the ability to disable. sink message of debezium json without schema field #11539
- Not debezium but our upsert json sink does not include schema by default but it can be turned on.

We will focus on the second case here. Below is a message from debezium 2.4 MySQL connector:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": true,
            "field": "bid"
          },
          {
            "type": "int64",
            "optional": true,
            "name": "io.debezium.time.MicroTimestamp",
            "version": 1,
            "field": "at"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.bar.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": true,
            "field": "bid"
          },
          {
            "type": "int64",
            "optional": true,
            "name": "io.debezium.time.MicroTimestamp",
            "version": 1,
            "field": "at"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.bar.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "version"
          },
          {
            "type": "string",
            "optional": false,
            "field": "connector"
          },
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_ms"
          },
          {
            "type": "string",
            "optional": true,
            "name": "io.debezium.data.Enum",
            "version": 1,
            "parameters": {
              "allowed": "true,last,false,incremental"
            },
            "default": "false",
            "field": "snapshot"
          },
          {
            "type": "string",
            "optional": false,
            "field": "db"
          },
          {
            "type": "string",
            "optional": true,
            "field": "sequence"
          },
          {
            "type": "string",
            "optional": true,
            "field": "table"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "server_id"
          },
          {
            "type": "string",
            "optional": true,
            "field": "gtid"
          },
          {
            "type": "string",
            "optional": false,
            "field": "file"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "pos"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "row"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "thread"
          },
          {
            "type": "string",
            "optional": true,
            "field": "query"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.mysql.Source",
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "id"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "total_order"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "data_collection_order"
          }
        ],
        "optional": true,
        "name": "event.block",
        "version": 1,
        "field": "transaction"
      }
    ],
    "optional": false,
    "name": "dbserver1.inventory.bar.Envelope",
    "version": 1
  },
  "payload": {
    "before": null,
    "after": {
      "bid": 8,
      "at": 1712132700123457
    },
    "source": {
      "version": "2.4.2.Final",
      "connector": "mysql",
      "name": "dbserver1",
      "ts_ms": 1712132719000,
      "snapshot": "false",
      "db": "inventory",
      "sequence": null,
      "table": "bar",
      "server_id": 223344,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 3181,
      "row": 0,
      "thread": 8,
      "query": null
    },
    "op": "c",
    "ts_ms": 1712132719617,
    "transaction": null
  }
}

The relevant part is here:

      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": true,
            "field": "bid"
          },
          {
            "type": "int64",
            "optional": true,
            "name": "io.debezium.time.MicroTimestamp",
            "version": 1,
            "field": "at"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.bar.Value",
        "field": "after"
      },

and

    "after": {
      "bid": 8,
      "at": 1712132700123457
    },

Note the field after.at has "type": "int64" and "name": "io.debezium.time.MicroTimestamp"

The structure of this schema field is actually from Java class org.apache.kafka.connect.json.JsonConverter:
https://github.com/apache/kafka/blob/3.7.0/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java#L313

tabVersion · 2024-04-16T09:29:07Z

close this issue as completed

MrCroxx added the type/bug Something isn't working label Apr 2, 2024

MrCroxx assigned tabVersion Apr 2, 2024

github-actions bot added this to the release-1.8 milestone Apr 2, 2024

MrCroxx assigned wangrunji0408 and tabVersion and unassigned tabVersion and wangrunji0408 Apr 2, 2024

xiangjinwu assigned MrCroxx and unassigned tabVersion Apr 3, 2024

MrCroxx assigned tabVersion and unassigned MrCroxx Apr 3, 2024

xiangjinwu mentioned this issue Apr 3, 2024

feat: Allow adjusting the parsing behavior of Debezium JSON format #16120

Closed

xiangjinwu self-assigned this Apr 3, 2024

tabVersion modified the milestones: release-1.8, release-1.9 Apr 8, 2024

xiangjinwu mentioned this issue Apr 12, 2024

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

Merged

9 tasks

tabVersion closed this as completed Apr 16, 2024

StrikeW mentioned this issue May 20, 2024

fix(cdc): Revert "fix(direct cdc): use time.precision.mode=connect (#16119)" #16842

Merged

9 tasks

fuyufjh mentioned this issue May 21, 2024

feat(source): support kafka-connect schema guided debezium parsing #16852

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

bug: wrong parsed timestamp from json #16097

bug: wrong parsed timestamp from json #16097

MrCroxx commented Apr 2, 2024 •

edited

Loading

tabVersion commented Apr 2, 2024

xiangjinwu commented Apr 3, 2024 •

edited

Loading

xiangjinwu commented Apr 3, 2024

fuyufjh commented Apr 3, 2024 •

edited

Loading

xiangjinwu commented Apr 3, 2024 •

edited

Loading

tabVersion commented Apr 16, 2024

bug: wrong parsed timestamp from json #16097

bug: wrong parsed timestamp from json #16097

Comments

MrCroxx commented Apr 2, 2024 • edited Loading

Describe the bug

Error message/log

To Reproduce

Expected behavior

How did you deploy RisingWave?

The version of RisingWave

Additional context

tabVersion commented Apr 2, 2024

xiangjinwu commented Apr 3, 2024 • edited Loading

xiangjinwu commented Apr 3, 2024

fuyufjh commented Apr 3, 2024 • edited Loading

xiangjinwu commented Apr 3, 2024 • edited Loading

tabVersion commented Apr 16, 2024

MrCroxx commented Apr 2, 2024 •

edited

Loading

xiangjinwu commented Apr 3, 2024 •

edited

Loading

fuyufjh commented Apr 3, 2024 •

edited

Loading

xiangjinwu commented Apr 3, 2024 •

edited

Loading