mz_mysql_util/decoding.rs
1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::fmt::Write;
11use std::str::FromStr;
12
13use mysql_async::{Row as MySqlRow, Value, from_value_opt};
14
15use mz_ore::cast::CastFrom;
16use mz_ore::error::ErrorExt;
17use mz_repr::adt::date::Date;
18use mz_repr::adt::jsonb::JsonbPacker;
19use mz_repr::adt::numeric::{NUMERIC_DATUM_MAX_PRECISION, Numeric, get_precision, get_scale};
20use mz_repr::adt::timestamp::CheckedTimestamp;
21use mz_repr::{Datum, Row, RowPacker, SqlScalarType};
22
23use crate::desc::MySqlColumnMeta;
24use crate::{MySqlColumnDesc, MySqlError, MySqlTableDesc};
25
26/// Canonical text for the MySQL zero-date sentinel ('0000-00-00 00:00:00').
27/// In binlog MYSQL_TYPE_DATETIME/MYSQL_TYPE_DATETIME2 encodings, sec=0 *cannot* represent unix
28/// epoch 0. The TIMESTAMP type's supported range starts at '1970-01-01 00:00:01' UTC
29/// (<https://dev.mysql.com/doc/refman/8.0/en/datetime.html>), so any sec=0 is
30/// unambiguously this sentinel.
31const MYSQL_ZERO_TIMESTAMP: &str = "0000-00-00 00:00:00";
32
33/// Maximum fractional-seconds precision MySQL accepts for DATETIME(p) and
34/// TIMESTAMP(p) — values are stored in microseconds, so 6 digits is the
35/// upper bound (<https://dev.mysql.com/doc/refman/8.0/en/fractional-seconds.html>).
36const MYSQL_MAX_FRACTIONAL_PRECISION: u32 = 6;
37
38/// Format the zero-date sentinel for a column with the given fractional
39/// precision (matches the Date arm's `{:0precision$}` behavior).
40fn mysql_zero_timestamp(precision: u32) -> String {
41 if precision > 0 {
42 format!(
43 "{}.{}",
44 MYSQL_ZERO_TIMESTAMP,
45 "0".repeat(usize::cast_from(precision))
46 )
47 } else {
48 MYSQL_ZERO_TIMESTAMP.to_string()
49 }
50}
51
52/// Format MySQL DATETIME/TIMESTAMP components as `YYYY-MM-DD HH:MM:SS[.ffff]`.
53/// `micros` is the raw microseconds (0..1_000_000); only the leading
54/// `precision` digits are kept, matching MySQL's DATETIME(p)/TIMESTAMP(p)
55/// display.
56fn format_mysql_timestamp(
57 y: u16,
58 m: u8,
59 d: u8,
60 hr: u8,
61 min: u8,
62 sec: u8,
63 micros: u32,
64 precision: u32,
65) -> String {
66 if precision == 0 {
67 return format!("{y:04}-{m:02}-{d:02} {hr:02}:{min:02}:{sec:02}");
68 }
69 // Clamp defensively: MySQL itself rejects precision > 6, but upstream
70 // metadata is untrusted and a larger value would make `pow()` below
71 // overflow its u32 exponent.
72 let p = precision.min(MYSQL_MAX_FRACTIONAL_PRECISION);
73 let scaled = micros / 10u32.pow(MYSQL_MAX_FRACTIONAL_PRECISION - p);
74 let width = usize::cast_from(p);
75 format!("{y:04}-{m:02}-{d:02} {hr:02}:{min:02}:{sec:02}.{scaled:0width$}")
76}
77
78pub fn pack_mysql_row(
79 row_container: &mut Row,
80 row: MySqlRow,
81 table_desc: &MySqlTableDesc,
82 gtid_set: Option<&str>,
83 binlog_full_metadata: bool,
84) -> Result<Row, MySqlError> {
85 let mut packer = row_container.packer();
86
87 // For each column in `table_desc` (in descriptor order), resolve its wire
88 // index. With binlog_full_metadata=true, columns are matched by name so a reordered upstream
89 // still decodes correctly; without binlog_full_metadata, rows have no column names and must be
90 // matched positionally. A `None` here means the upstream row is missing this column and is
91 // only tolerated for ignored columns, and for binlog_full_metadata = false, is only tolerated
92 // for ignored columns at the end of the table.
93 for (i, col_desc) in table_desc.columns.iter().enumerate() {
94 if col_desc.column_type.is_none() {
95 // This column is ignored, so don't decode it.
96 continue;
97 }
98 let wire_idx = if !binlog_full_metadata {
99 // No column name metadata, so we match by index.
100 (i < row.len()).then_some(i)
101 } else {
102 // This means the row from the binlog has column name included in the metadata,
103 // so we can match on that instead of position.
104 row.columns_ref()
105 .iter()
106 .position(|wc| wc.name_str() == col_desc.name.as_str())
107 };
108
109 let wire_idx = match wire_idx {
110 Some(idx) => idx,
111 None => {
112 // We could not find a column in the incoming row that matches this descriptor column.
113 // This is an error as the column is not ignored (ignored columns have already been skipped).
114 return Err(decode_error(
115 "extra column description",
116 col_desc,
117 table_desc,
118 gtid_set,
119 &row,
120 ));
121 }
122 };
123 let value = row
124 .as_ref(wire_idx)
125 .expect("wire_idx resolved from row")
126 .clone();
127 if let Err(err) = pack_val_as_datum(value, col_desc, &mut packer) {
128 return Err(decode_error(
129 &err.to_string(),
130 col_desc,
131 table_desc,
132 gtid_set,
133 &row,
134 ));
135 }
136 }
137
138 Ok(row_container.clone())
139}
140
141/// Build a `ValueDecodeError`, logging the schema, table, column, source
142/// gtid_set (if any), and a shape description of `row` at the same time.
143/// The shape string is only built here — pack_mysql_row's happy path does no
144/// per-row allocation beyond what decoding requires.
145fn decode_error(
146 err_msg: &str,
147 col_desc: &MySqlColumnDesc,
148 table_desc: &MySqlTableDesc,
149 gtid_set: Option<&str>,
150 row: &MySqlRow,
151) -> MySqlError {
152 let row_shape = describe_row_shape(row, table_desc);
153 tracing::warn!(
154 "mysql decode error for `{}`.`{}` column `{}`: {}; gtid_set={:?}; row_shape={}",
155 table_desc.schema_name,
156 table_desc.name,
157 col_desc.name,
158 err_msg,
159 gtid_set,
160 row_shape,
161 );
162 MySqlError::ValueDecodeError {
163 column_name: col_desc.name.clone(),
164 qualified_table_name: format!("{}.{}", table_desc.schema_name, table_desc.name),
165 error: err_msg.to_string(),
166 }
167}
168
169/// Describes the structural shape of a row without revealing any data values.
170/// Iterates every wire column. For each, emits the wire name, the binlog
171/// wire type, the character-set id (or `binary`), a classification relative
172/// to `table_desc` (`expected=<scalar>` for active columns, `ignored` for
173/// columns excluded from the source, `extra` for upstream columns with no
174/// descriptor entry), and a value disposition (`null` or `bytes(len=N)` /
175/// primitive kind). Intended for diagnostic logging on decode errors: MySQL
176/// serializes CHAR, VARCHAR, TEXT, JSON, BLOB, etc. all as `Value::Bytes`,
177/// so the wire type tag and the expected scalar type are what distinguish
178/// them.
179fn describe_row_shape(row: &MySqlRow, table_desc: &MySqlTableDesc) -> String {
180 // Binlogs without full row metadata use positional "@N" names, so we
181 // have to match by wire position rather than by name.
182 let fallback_names = row
183 .columns_ref()
184 .first()
185 .is_some_and(|col| col.name_ref().starts_with(b"@"));
186
187 let mut out = String::new();
188 out.push('[');
189 for (i, wire_col) in row.columns_ref().iter().enumerate() {
190 if i > 0 {
191 out.push_str(", ");
192 }
193 let wire_name = wire_col.name_str();
194 let cs = wire_col.character_set();
195 // 63 = binary collation (binary/blob columns).
196 let cs_str = if cs == 63 {
197 "binary".to_string()
198 } else {
199 format!("charset={cs}")
200 };
201 let wire_type = format!("{:?}", wire_col.column_type());
202
203 let matched_col = if fallback_names {
204 table_desc.columns.get(i)
205 } else {
206 table_desc
207 .columns
208 .iter()
209 .find(|c| c.name.as_str() == wire_name)
210 };
211 let match_info = match matched_col {
212 Some(col) => match &col.column_type {
213 Some(ct) => format!("expected={:?}", ct.scalar_type),
214 None => "ignored".to_string(),
215 },
216 None => "extra".to_string(),
217 };
218
219 let val_desc = match row.as_ref(i) {
220 None => "absent".to_string(),
221 Some(Value::NULL) => "null".to_string(),
222 Some(Value::Bytes(b)) => format!("bytes(len={})", b.len()),
223 Some(Value::Int(_)) => "int".to_string(),
224 Some(Value::UInt(_)) => "uint".to_string(),
225 Some(Value::Float(_)) => "float".to_string(),
226 Some(Value::Double(_)) => "double".to_string(),
227 Some(Value::Date(..)) => "date".to_string(),
228 Some(Value::Time(..)) => "time".to_string(),
229 };
230
231 let _ = write!(
232 out,
233 "{{name={wire_name}, wire={wire_type}, {cs_str}, {match_info}, val={val_desc}}}"
234 );
235 }
236 out.push(']');
237 out
238}
239
240// TODO(guswynn|roshan): This function has various `.to_string()` and `format!` calls that should
241// use a shared allocation if possible.
242fn pack_val_as_datum(
243 value: Value,
244 col_desc: &MySqlColumnDesc,
245 packer: &mut RowPacker,
246) -> Result<(), anyhow::Error> {
247 let column_type = match col_desc.column_type {
248 Some(ref column_type) => column_type,
249 None => anyhow::bail!("column type is not set for column: {}", col_desc.name),
250 };
251 match value {
252 Value::NULL => {
253 if column_type.nullable {
254 packer.push(Datum::Null);
255 } else {
256 Err(anyhow::anyhow!(
257 "received a null value in a non-null column".to_string(),
258 ))?
259 }
260 }
261 value => match &column_type.scalar_type {
262 SqlScalarType::Bool => packer.push(Datum::from(from_value_opt::<bool>(value)?)),
263 SqlScalarType::UInt16 => packer.push(Datum::from(from_value_opt::<u16>(value)?)),
264 SqlScalarType::Int16 => packer.push(Datum::from(from_value_opt::<i16>(value)?)),
265 SqlScalarType::UInt32 => packer.push(Datum::from(from_value_opt::<u32>(value)?)),
266 SqlScalarType::Int32 => packer.push(Datum::from(from_value_opt::<i32>(value)?)),
267 SqlScalarType::UInt64 => {
268 if let Some(MySqlColumnMeta::Bit(precision)) = &col_desc.meta {
269 let mut value = from_value_opt::<Vec<u8>>(value)?;
270
271 // Ensure we have the correct number of bytes.
272 let precision_bytes = (precision + 7) / 8;
273 if value.len() != usize::cast_from(precision_bytes) {
274 return Err(anyhow::anyhow!("'bit' column out of range!"));
275 }
276 // Be defensive and prune any bits that come over the wire and are
277 // greater than our precision.
278 let bit_index = precision % 8;
279 if bit_index != 0 {
280 let mask = !(u8::MAX << bit_index);
281 if value.len() > 0 {
282 value[0] &= mask;
283 }
284 }
285
286 // Based on experimentation the value coming across the wire is
287 // encoded in big-endian.
288 let mut buf = [0u8; 8];
289 buf[(8 - value.len())..].copy_from_slice(value.as_slice());
290 let value = u64::from_be_bytes(buf);
291 packer.push(Datum::from(value))
292 } else {
293 packer.push(Datum::from(from_value_opt::<u64>(value)?))
294 }
295 }
296 SqlScalarType::Int64 => packer.push(Datum::from(from_value_opt::<i64>(value)?)),
297 SqlScalarType::Float32 => packer.push(Datum::from(from_value_opt::<f32>(value)?)),
298 SqlScalarType::Float64 => packer.push(Datum::from(from_value_opt::<f64>(value)?)),
299 SqlScalarType::Char { length } => {
300 let val = from_value_opt::<String>(value)?;
301 check_char_length(length.map(|l| l.into_u32()), &val, col_desc)?;
302 packer.push(Datum::String(&val));
303 }
304 SqlScalarType::VarChar { max_length } => {
305 let val = from_value_opt::<String>(value)?;
306 check_char_length(max_length.map(|l| l.into_u32()), &val, col_desc)?;
307 packer.push(Datum::String(&val));
308 }
309 SqlScalarType::String => {
310 // Special case for string types, since this is the scalar type used for a column
311 // specified as a 'TEXT COLUMN'. In some cases we need to check the column
312 // metadata to know if the upstream value needs special handling
313 match &col_desc.meta {
314 Some(MySqlColumnMeta::Enum(e)) => {
315 match value {
316 Value::Bytes(data) => {
317 let data = std::str::from_utf8(&data)?;
318 packer.push(Datum::String(data));
319 }
320 Value::Int(val) => {
321 let enum_int = usize::try_from(val)?;
322
323 // If mysql strict mode is disabled when an invalid entry is inserted
324 // then the entry will be replicated as a 0. Outside the 1 indexed enum.
325 // https://dev.mysql.com/doc/refman/8.4/en/enum.html#enum-indexes
326 if enum_int == 0 {
327 packer.push(Datum::String(""));
328 } else {
329 // Enum types are provided as 1-indexed integers in the replication
330 // stream, so we need to find the string value from the enum meta
331 let enum_val = e.values.get(enum_int - 1).ok_or_else(|| {
332 anyhow::anyhow!(
333 "received invalid enum value: {} for column {}",
334 val,
335 col_desc.name
336 )
337 })?;
338
339 packer.push(Datum::String(enum_val));
340 }
341 }
342 _ => Err(anyhow::anyhow!(
343 "received unexpected value for enum type: {:?}",
344 value
345 ))?,
346 }
347 }
348 Some(MySqlColumnMeta::Json) => {
349 // JSON types in a query response are encoded as a string with whitespace,
350 // but when parsed from the binlog event by mysql-common they are provided
351 // as an encoded string sans-whitespace.
352 if let Value::Bytes(data) = value {
353 let json = serde_json::from_slice::<serde_json::Value>(&data)?;
354 packer.push(Datum::String(&json.to_string()));
355 } else {
356 Err(anyhow::anyhow!(
357 "received unexpected value for json type: {:?}",
358 value
359 ))?;
360 }
361 }
362 Some(MySqlColumnMeta::Year) => {
363 let mut val = from_value_opt::<u16>(value)?;
364 // mysql_common incorrectly handles MySQL YEAR type, which has a valid range
365 // of 1901-2155 (https://dev.mysql.com/doc/refman/8.0/en/year.html)
366 //
367 // We treat the value 1900 as the zero-value year - "0000"
368 // https://github.com/blackbeam/rust_mysql_common/blob/v0.35.5/src/binlog/value.rs#L124-L129
369 if val == 1900 {
370 val = 0;
371 }
372 packer.push(Datum::String(&format!("{val:04}")));
373 }
374 Some(MySqlColumnMeta::Date) => {
375 // Some MySQL dates are invalid in chrono/NaiveDate (e.g. 0000-00-00), so
376 // we need to handle them directly as strings
377 if let Value::Date(y, m, d, 0, 0, 0, 0) = value {
378 packer.push(Datum::String(&format!("{:04}-{:02}-{:02}", y, m, d)));
379 } else {
380 Err(anyhow::anyhow!(
381 "received unexpected value for date type: {:?}",
382 value
383 ))?;
384 }
385 }
386 Some(MySqlColumnMeta::Timestamp(precision)) => {
387 // Materialize treats DATETIME and TIMESTAMP as MySqlColumnMeta::Timestamp,
388 // but they have slightly different semantics as far as the range of dates
389 // they can represent.
390 // (see https://dev.mysql.com/doc/refman/8.0/en/date-and-time-types.html).
391 //
392 // Three mysql_common::Value variants exist, which are mapped to
393 // [`MySqlColumnMeta::Timestamp`]
394 // (see https://github.com/blackbeam/rust_mysql_common/blob/2e6f6696de03c91b9fd95a87356d081285290704/src/binlog/value.rs):
395 // Value::Date — MZ snapshot & binlog MYSQL_TYPE_DATETIME/MYSQL_TYPE_DATETIME2
396 // (value/mod.rs:443-445, binlog/value.rs:109-161)
397 // Value::Int — legacy binlog MYSQL_TYPE_TIMESTAMP, pre-5.6,
398 // 4-byte unix epoch (binlog/value.rs:87-90)
399 // Value::Bytes — binlog MYSQL_TYPE_TIMESTAMP2, 5.6+,
400 // "<sec>" or "<sec>.<usec>" (binlog/value.rs:145-154)
401 let str_timestamp = match value {
402 Value::Date(y, m, d, h, mm, s, ms) => {
403 format_mysql_timestamp(y, m, d, h, mm, s, ms, *precision)
404 }
405 // Pre-5.6 unix epoch, no fractional seconds.
406 // val == 0 is the zero-date sentinel, not epoch 0.
407 Value::Int(0) => mysql_zero_timestamp(*precision),
408 Value::Int(val) => chrono::DateTime::from_timestamp(val, 0)
409 .ok_or_else(|| {
410 anyhow::anyhow!("received invalid timestamp value: {}", val)
411 })?
412 .naive_utc()
413 .format("%Y-%m-%d %H:%M:%S")
414 .to_string(),
415 // 5.6+ epoch string; parse + reformat so all variants emit the
416 // same canonical YYYY-MM-DD HH:MM:SS[.ffff] text.
417 Value::Bytes(data) => {
418 let s = std::str::from_utf8(&data).map_err(|_| {
419 anyhow::anyhow!("received invalid timestamp value: {:?}", data)
420 })?;
421 // sec=0 (with or without fractional component) is the
422 // zero-date sentinel.
423 if s.split('.').next() == Some("0") {
424 mysql_zero_timestamp(*precision)
425 } else {
426 let dt = if s.contains('.') {
427 chrono::NaiveDateTime::parse_from_str(s, "%s%.6f")
428 } else {
429 chrono::NaiveDateTime::parse_from_str(s, "%s")
430 }
431 .map_err(|_| {
432 anyhow::anyhow!("received invalid timestamp value: {:?}", s)
433 })?;
434 use chrono::{Datelike, Timelike};
435 let y = u16::try_from(dt.year()).map_err(|_| {
436 anyhow::anyhow!(
437 "timestamp year out of range: {}",
438 dt.year()
439 )
440 })?;
441 format_mysql_timestamp(
442 y,
443 u8::try_from(dt.month())?,
444 u8::try_from(dt.day())?,
445 u8::try_from(dt.hour())?,
446 u8::try_from(dt.minute())?,
447 u8::try_from(dt.second())?,
448 dt.nanosecond() / 1000,
449 *precision,
450 )
451 }
452 }
453 _ => Err(anyhow::anyhow!(
454 "received unexpected value for timestamp type: {:?}",
455 value
456 ))?,
457 };
458 packer.push(Datum::String(&str_timestamp));
459 }
460 Some(MySqlColumnMeta::Bit(_)) => unreachable!("parsed as a u64"),
461 None => {
462 packer.push(Datum::String(&from_value_opt::<String>(value)?));
463 }
464 }
465 }
466 SqlScalarType::Jsonb => {
467 if let Value::Bytes(data) = value {
468 let packer = JsonbPacker::new(packer);
469 // TODO(guswynn): This still produces and extract allocation (in the
470 // `DeserializeSeed` impl used internally), which should be improved,
471 // for all users of the APIs in that module.
472 packer.pack_slice(&data).map_err(|e| {
473 anyhow::anyhow!(
474 "Failed to decode JSON: {}",
475 // See if we can output the string that failed to be converted to JSON.
476 match std::str::from_utf8(&data) {
477 Ok(str) => str.to_string(),
478 // Otherwise produce the nominally helpful error.
479 Err(_) => e.display_with_causes().to_string(),
480 }
481 )
482 })?;
483 } else {
484 Err(anyhow::anyhow!(
485 "received unexpected value for json type: {:?}",
486 value
487 ))?
488 }
489 }
490 SqlScalarType::Bytes => {
491 let data = from_value_opt::<Vec<u8>>(value)?;
492 packer.push(Datum::Bytes(&data));
493 }
494 SqlScalarType::Date => {
495 let date = Date::try_from(from_value_opt::<chrono::NaiveDate>(value)?)?;
496 packer.push(Datum::from(date));
497 }
498 SqlScalarType::Timestamp { precision: _ } => {
499 // Timestamps are encoded as different mysql_common::Value types depending on
500 // whether they are from a binlog event or a query, and depending on which
501 // mysql timestamp version is used. We handle those cases here
502 // https://github.com/blackbeam/rust_mysql_common/blob/v0.35.5/src/binlog/value.rs#L87-L155
503 // https://github.com/blackbeam/rust_mysql_common/blob/v0.35.5/src/value/mod.rs#L332
504 let chrono_timestamp = match value {
505 Value::Date(..) => from_value_opt::<chrono::NaiveDateTime>(value)?,
506 // old temporal format from before MySQL 5.6; didn't support fractional seconds
507 Value::Int(val) => chrono::DateTime::from_timestamp(val, 0)
508 .ok_or_else(|| {
509 anyhow::anyhow!("received invalid timestamp value: {}", val)
510 })?
511 .naive_utc(),
512 Value::Bytes(data) => {
513 let data = std::str::from_utf8(&data)?;
514 if data.contains('.') {
515 chrono::NaiveDateTime::parse_from_str(data, "%s%.6f")?
516 } else {
517 chrono::NaiveDateTime::parse_from_str(data, "%s")?
518 }
519 }
520 _ => Err(anyhow::anyhow!(
521 "received unexpected value for timestamp type: {:?}",
522 value
523 ))?,
524 };
525 packer.push(Datum::try_from(CheckedTimestamp::try_from(
526 chrono_timestamp,
527 )?)?);
528 }
529 SqlScalarType::Time => {
530 packer.push(Datum::from(from_value_opt::<chrono::NaiveTime>(value)?));
531 }
532 SqlScalarType::Numeric { max_scale } => {
533 // The wire-format of numeric types is a string when sent in a binary query
534 // response but is represented in a decimal binary format when sent in a binlog
535 // event. However the mysql-common crate abstracts this away and always returns
536 // a string. We parse the string into a numeric type here.
537 let val = from_value_opt::<String>(value)?;
538 let val = Numeric::from_str(&val)?;
539 if get_precision(&val) > NUMERIC_DATUM_MAX_PRECISION.into() {
540 Err(anyhow::anyhow!(
541 "received numeric value with precision {} for column {} which has a max precision of {}",
542 get_precision(&val),
543 col_desc.name,
544 NUMERIC_DATUM_MAX_PRECISION
545 ))?
546 }
547 if let Some(max_scale) = max_scale {
548 if get_scale(&val) > max_scale.into_u8().into() {
549 Err(anyhow::anyhow!(
550 "received numeric value with scale {} for column {} which has a max scale of {}",
551 get_scale(&val),
552 col_desc.name,
553 max_scale.into_u8()
554 ))?
555 }
556 }
557 packer.push(Datum::from(val));
558 }
559 // TODO(roshan): IMPLEMENT OTHER TYPES
560 data_type => Err(anyhow::anyhow!(
561 "received unexpected value for type: {:?}: {:?}",
562 data_type,
563 value
564 ))?,
565 },
566 }
567 Ok(())
568}
569
570fn check_char_length(
571 length: Option<u32>,
572 val: &str,
573 col_desc: &MySqlColumnDesc,
574) -> Result<(), anyhow::Error> {
575 if let Some(length) = length {
576 if let Some(_) = val.char_indices().nth(usize::cast_from(length)) {
577 Err(anyhow::anyhow!(
578 "received string value of length {} for column {} which has a max length of {}",
579 val.len(),
580 col_desc.name,
581 length
582 ))?
583 }
584 }
585 Ok(())
586}
587
588#[cfg(test)]
589mod tests {
590 //! Unit tests for the TEXT-COLUMNS decoding of MySQL TIMESTAMP values.
591 //!
592 //! These cover the regression where a MySQL TIMESTAMP column declared as
593 //! a TEXT COLUMN fails to decode when the wire value arrives as
594 //! `Value::Bytes("<unix-epoch>")` or `Value::Int(<unix-epoch>)` instead
595 //! of `Value::Date(..)`. The integration test in
596 //! `test/mysql-cdc/text-columns-timestamp.td` exercises this through
597 //! a real MySQL container but is non-deterministic: which `Value`
598 //! variant `mysql-async` produces depends on connection-state timing.
599 //! These unit tests pin each variant down directly.
600 //!
601 //! The wire-variant matrix exercised below is derived from mysql_common
602 //! v0.35.5:
603 //!
604 //! * Value::Int(epoch) — binlog MYSQL_TYPE_TIMESTAMP (pre-5.6):
605 //! https://github.com/blackbeam/rust_mysql_common/blob/v0.35.5/src/binlog/value.rs#L87-L90
606 //! * Value::Bytes("<sec>"/"<sec>.<usec>") — binlog MYSQL_TYPE_TIMESTAMP2 (5.6+):
607 //! https://github.com/blackbeam/rust_mysql_common/blob/v0.35.5/src/binlog/value.rs#L145-L154
608 //! * Value::Date(...) — binary query response + binlog DATETIME[2]:
609 //! https://github.com/blackbeam/rust_mysql_common/blob/v0.35.5/src/value/mod.rs#L443-L445
610 //! https://github.com/blackbeam/rust_mysql_common/blob/v0.35.5/src/binlog/value.rs#L109-L161
611 //!
612 //! MySQL semantics referenced by the zero-date and fractional-precision
613 //! cases:
614 //!
615 //! * Zero-date allowed when sql_mode disables NO_ZERO_DATE:
616 //! https://dev.mysql.com/doc/refman/8.0/en/sql-mode.html#sqlmode_no_zero_date
617 //! * TIMESTAMP(p) / DATETIME(p) fractional seconds:
618 //! https://dev.mysql.com/doc/refman/8.0/en/fractional-seconds.html
619 use super::*;
620 use mz_repr::{SqlColumnType, SqlScalarType};
621
622 fn timestamp_text_col(precision: u32) -> MySqlColumnDesc {
623 MySqlColumnDesc {
624 name: "created_at".to_string(),
625 column_type: Some(SqlColumnType {
626 scalar_type: SqlScalarType::String,
627 nullable: true,
628 }),
629 meta: Some(MySqlColumnMeta::Timestamp(precision)),
630 }
631 }
632
633 fn pack_one(value: Value, col: &MySqlColumnDesc) -> Result<String, anyhow::Error> {
634 let mut row = Row::default();
635 pack_val_as_datum(value, col, &mut row.packer())?;
636 Ok(row.unpack_first().unwrap_str().to_string())
637 }
638
639 #[mz_ore::test]
640 fn timestamp_value_date_no_precision() {
641 let col = timestamp_text_col(0);
642 let s = pack_one(Value::Date(2024, 4, 3, 10, 15, 13, 0), &col).unwrap();
643 assert_eq!(s, "2024-04-03 10:15:13");
644 }
645
646 #[mz_ore::test]
647 fn timestamp_value_date_with_precision() {
648 let col = timestamp_text_col(6);
649 let s = pack_one(Value::Date(2024, 4, 3, 10, 15, 13, 123456), &col).unwrap();
650 assert_eq!(s, "2024-04-03 10:15:13.123456");
651 }
652
653 #[mz_ore::test]
654 fn timestamp_value_date_zero_date() {
655 // The whole reason TEXT COLUMNS exists for TIMESTAMP: a
656 // zero-date arriving as Value::Date(0,..) should decode to the same
657 // "zero" timestamp value MySQL would display.
658 let col = timestamp_text_col(0);
659 let s = pack_one(Value::Date(0, 0, 0, 0, 0, 0, 0), &col).unwrap();
660 assert_eq!(s, "0000-00-00 00:00:00");
661 }
662
663 /// Regression: Value::Int (pre-5.6 legacy temporal format, unix
664 /// epoch seconds) was previously rejected with
665 /// `received unexpected value for timestamp type: Int(..)`.
666 #[mz_ore::test]
667 fn timestamp_value_int_epoch() {
668 let col = timestamp_text_col(0);
669 // 1743661234 == 2025-04-03 06:20:34 UTC
670 let s = pack_one(Value::Int(1_743_661_234), &col).unwrap();
671 assert_eq!(s, "2025-04-03 06:20:34");
672 }
673
674 /// sec=0 in the legacy TIMESTAMP encoding is the zero-date sentinel,
675 /// not unix epoch 0 — TIMESTAMP's range starts at '1970-01-01 00:00:01'
676 /// UTC so epoch 0 isn't a representable column value.
677 #[mz_ore::test]
678 fn timestamp_value_int_zero_is_sentinel() {
679 let col = timestamp_text_col(0);
680 let s = pack_one(Value::Int(0), &col).unwrap();
681 assert_eq!(s, "0000-00-00 00:00:00");
682 }
683
684 /// Out-of-range epochs must error rather than silently producing
685 /// a zero-timestamp — they aren't the MySQL zero-date marker, just
686 /// garbage chrono can't represent.
687 #[mz_ore::test]
688 fn timestamp_value_int_out_of_range_errors() {
689 let col = timestamp_text_col(0);
690 let err = pack_one(Value::Int(i64::MAX), &col).unwrap_err();
691 assert!(
692 err.to_string().contains("invalid timestamp value"),
693 "unexpected error message: {err}"
694 );
695 }
696
697 /// Regression: Value::Bytes carrying a unix-epoch string is the
698 /// wire variant that triggered the production failure
699 /// received unexpected value for timestamp type: Bytes("17436613..")
700 #[mz_ore::test]
701 fn timestamp_value_bytes_epoch() {
702 let col = timestamp_text_col(0);
703 let s = pack_one(Value::Bytes(b"1743661234".to_vec()), &col).unwrap();
704 assert_eq!(s, "2025-04-03 06:20:34");
705 }
706
707 /// sec=0 in the TIMESTAMP2 encoding is the zero-date sentinel; same
708 /// reasoning as `timestamp_value_int_zero_is_sentinel`.
709 #[mz_ore::test]
710 fn timestamp_value_bytes_zero_is_sentinel() {
711 let col = timestamp_text_col(0);
712 let s = pack_one(Value::Bytes(b"0".to_vec()), &col).unwrap();
713 assert_eq!(s, "0000-00-00 00:00:00");
714 }
715
716 /// Sentinel detection survives a fractional component ("0.NNNNNN"),
717 /// and the helper pads the output to the column's precision so that
718 /// snapshot and binlog paths produce identical text for the same
719 /// upstream row.
720 #[mz_ore::test]
721 fn timestamp_value_bytes_zero_with_fractional_is_sentinel() {
722 let col = timestamp_text_col(6);
723 let s = pack_one(Value::Bytes(b"0.000000".to_vec()), &col).unwrap();
724 assert_eq!(s, "0000-00-00 00:00:00.000000");
725 }
726
727 /// Fractional form of the TIMESTAMP2 binlog encoding —
728 /// "<sec>.<usec>" wrapped in Value::Bytes (binlog/value.rs:151-153).
729 /// Hits the `s.contains('.')` branch and the precision-aware
730 /// reformat.
731 #[mz_ore::test]
732 fn timestamp_value_bytes_epoch_fractional() {
733 let col = timestamp_text_col(6);
734 let s = pack_one(Value::Bytes(b"1743661234.123456".to_vec()), &col).unwrap();
735 assert_eq!(s, "2025-04-03 06:20:34.123456");
736 }
737
738 /// Bytes that aren't valid UTF-8 should produce a meaningful error,
739 /// not a panic.
740 #[mz_ore::test]
741 fn timestamp_value_bytes_invalid_utf8_errors() {
742 let col = timestamp_text_col(0);
743 // 0xC3 0x28 is an invalid 2-byte UTF-8 sequence.
744 let err = pack_one(Value::Bytes(vec![0xC3, 0x28]), &col).unwrap_err();
745 let msg = err.to_string();
746 assert!(
747 msg.contains("invalid timestamp value"),
748 "unexpected error message: {msg}"
749 );
750 }
751
752 /// Bytes that are valid UTF-8 but not parseable as a unix epoch
753 /// should produce the same structured error as invalid UTF-8 —
754 /// covers the chrono parse failure path that
755 /// `timestamp_value_bytes_invalid_utf8_errors` doesn't reach.
756 #[mz_ore::test]
757 fn timestamp_value_bytes_unparseable_errors() {
758 let col = timestamp_text_col(0);
759 // "2024-04-03 10:15:13" is not valid because Value::Bytes must contain seconds since
760 // epoch, Value::Bytes("<sec>"/"<sec>.<usec>")
761 for payload in [&b""[..], &b"not-an-epoch"[..], &b"2024-04-03 10:15:13"[..]] {
762 let err = pack_one(Value::Bytes(payload.to_vec()), &col).unwrap_err();
763 assert!(
764 err.to_string().contains("invalid timestamp value"),
765 "payload {payload:?}: unexpected error message: {err}"
766 );
767 }
768 }
769
770 /// Variants that have no defined mapping for a TIMESTAMP column
771 /// must still produce the existing structured decode error so the
772 /// source health surface can flag them.
773 #[mz_ore::test]
774 fn timestamp_value_unsupported_variant_errors() {
775 let col = timestamp_text_col(0);
776 let err = pack_one(Value::Float(1.0), &col).unwrap_err();
777 let msg = err.to_string();
778 assert!(
779 msg.contains("unexpected value for timestamp"),
780 "unexpected error message: {msg}"
781 );
782 }
783}