77// the Business Source License, use of this software will be governed
88// by the Apache License, Version 2.0.
99
10+ use std:: fmt:: Write ;
1011use std:: str:: FromStr ;
1112
12- use itertools:: { EitherOrBoth , Itertools } ;
1313use mysql_common:: value:: convert:: from_value_opt;
1414use mysql_common:: { Row as MySqlRow , Value } ;
1515
@@ -28,46 +28,177 @@ pub fn pack_mysql_row(
2828 row_container : & mut Row ,
2929 row : MySqlRow ,
3030 table_desc : & MySqlTableDesc ,
31+ gtid_set : Option < & str > ,
32+ binlog_full_metadata : bool ,
3133) -> Result < Row , MySqlError > {
3234 let mut packer = row_container. packer ( ) ;
33- let row_values = row. unwrap ( ) ;
3435
35- for values in table_desc. columns . iter ( ) . zip_longest ( row_values) {
36- let ( col_desc, value) = match values {
37- EitherOrBoth :: Both ( col_desc, value) => ( col_desc, value) ,
38- EitherOrBoth :: Left ( col_desc) => {
39- tracing:: error!(
40- "mysql: extra column description {col_desc:?} for table {}" ,
41- table_desc. name
42- ) ;
43- Err ( MySqlError :: ValueDecodeError {
44- column_name : col_desc. name . clone ( ) ,
45- qualified_table_name : format ! ( "{}.{}" , table_desc. schema_name, table_desc. name) ,
46- error : "extra column description" . to_string ( ) ,
47- } ) ?
48- }
49- EitherOrBoth :: Right ( _) => {
50- // If there are extra columns on the upstream table we can safely ignore them
51- break ;
52- }
36+ // If a column name begins with '@', then the binlog does not have full row metadata,
37+ // meaning that full column names are not available and we need to rely on the order
38+ // of the columns in the upstream table matching the order of the columns in the row.
39+ // This is a fallback for MySQL servers that do not have `binlog_row_metadata` set to
40+ // `FULL`. If the first column name does not begin with '@', then we can assume that
41+ // full metadata is available and we can match columns by name.
42+ let fallback_names = row
43+ . columns_ref ( )
44+ . first ( )
45+ . is_some_and ( |col| col. name_ref ( ) . starts_with ( b"@" ) ) ;
46+
47+ if binlog_full_metadata && fallback_names {
48+ // This should never happen, but if it does, it's a sign that something is very wrong with the MySQL server's binlog configuration. We want to error rather than silently producing incorrect results.
49+ return Err ( MySqlError :: ValueDecodeError {
50+ column_name : "<unknown>" . to_string ( ) ,
51+ qualified_table_name : format ! ( "{}.{}" , table_desc. schema_name, table_desc. name) ,
52+ error : "Table was created with binlog_row_metadata=FULL but binlog_row_metadata has since been set to a different value, meaning we cannot reliably decode the columns" . to_string ( ) ,
53+ } ) ;
54+ }
55+
56+ // For each column in `table_desc` (in descriptor order), resolve its wire
57+ // index. Non-fallback rows are matched by name so a reordered upstream
58+ // still decodes correctly; fallback rows have no names and are matched
59+ // positionally. A `None` here means the upstream row is missing this
60+ // column and is only tolerated for ignored columns.
61+ for ( i, col_desc) in table_desc. columns . iter ( ) . enumerate ( ) {
62+ let wire_idx = if !binlog_full_metadata {
63+ ( i < row. len ( ) ) . then_some ( i)
64+ } else {
65+ row. columns_ref ( )
66+ . iter ( )
67+ . position ( |wc| wc. name_str ( ) == col_desc. name . as_str ( ) )
5368 } ;
5469 if col_desc. column_type . is_none ( ) {
5570 // This column is ignored, so don't decode it.
5671 continue ;
5772 }
58- match pack_val_as_datum ( value, col_desc, & mut packer) {
59- Err ( err) => Err ( MySqlError :: ValueDecodeError {
60- column_name : col_desc. name . clone ( ) ,
61- qualified_table_name : format ! ( "{}.{}" , table_desc. schema_name, table_desc. name) ,
62- error : err. to_string ( ) ,
63- } ) ?,
64- Ok ( ( ) ) => ( ) ,
73+ let wire_idx = match wire_idx {
74+ Some ( idx) => idx,
75+ None => {
76+ return Err ( decode_error (
77+ "extra column description" ,
78+ col_desc,
79+ table_desc,
80+ gtid_set,
81+ & row,
82+ ) ) ;
83+ }
6584 } ;
85+ let value = row
86+ . as_ref ( wire_idx)
87+ . expect ( "wire_idx resolved from row" )
88+ . clone ( ) ;
89+ if let Err ( err) = pack_val_as_datum ( value, col_desc, & mut packer) {
90+ return Err ( decode_error (
91+ & err. to_string ( ) ,
92+ col_desc,
93+ table_desc,
94+ gtid_set,
95+ & row,
96+ ) ) ;
97+ }
6698 }
6799
68100 Ok ( row_container. clone ( ) )
69101}
70102
103+ /// Build a `ValueDecodeError`, logging the schema, table, column, source
104+ /// gtid_set (if any), and a shape description of `row` at the same time.
105+ /// The shape string is only built here — pack_mysql_row's happy path does no
106+ /// per-row allocation beyond what decoding requires.
107+ fn decode_error (
108+ err_msg : & str ,
109+ col_desc : & MySqlColumnDesc ,
110+ table_desc : & MySqlTableDesc ,
111+ gtid_set : Option < & str > ,
112+ row : & MySqlRow ,
113+ ) -> MySqlError {
114+ let row_shape = describe_row_shape ( row, table_desc) ;
115+ tracing:: warn!(
116+ "mysql decode error for `{}`.`{}` column `{}`: {}; gtid_set={:?}; row_shape={}" ,
117+ table_desc. schema_name,
118+ table_desc. name,
119+ col_desc. name,
120+ err_msg,
121+ gtid_set,
122+ row_shape,
123+ ) ;
124+ MySqlError :: ValueDecodeError {
125+ column_name : col_desc. name . clone ( ) ,
126+ qualified_table_name : format ! ( "{}.{}" , table_desc. schema_name, table_desc. name) ,
127+ error : err_msg. to_string ( ) ,
128+ }
129+ }
130+
131+ /// Describes the structural shape of a row without revealing any data values.
132+ /// Iterates every wire column. For each, emits the wire name, the binlog
133+ /// wire type, the character-set id (or `binary`), a classification relative
134+ /// to `table_desc` (`expected=<scalar>` for active columns, `ignored` for
135+ /// columns excluded from the source, `extra` for upstream columns with no
136+ /// descriptor entry), and a value disposition (`null` or `bytes(len=N)` /
137+ /// primitive kind). Intended for diagnostic logging on decode errors: MySQL
138+ /// serializes CHAR, VARCHAR, TEXT, JSON, BLOB, etc. all as `Value::Bytes`,
139+ /// so the wire type tag and the expected scalar type are what distinguish
140+ /// them.
141+ fn describe_row_shape ( row : & MySqlRow , table_desc : & MySqlTableDesc ) -> String {
142+ // Binlogs without full row metadata use positional "@N" names, so we
143+ // have to match by wire position rather than by name.
144+ let fallback_names = row
145+ . columns_ref ( )
146+ . first ( )
147+ . is_some_and ( |col| col. name_ref ( ) . starts_with ( b"@" ) ) ;
148+
149+ let mut out = String :: new ( ) ;
150+ out. push ( '[' ) ;
151+ for ( i, wire_col) in row. columns_ref ( ) . iter ( ) . enumerate ( ) {
152+ if i > 0 {
153+ out. push_str ( ", " ) ;
154+ }
155+ let wire_name = wire_col. name_str ( ) ;
156+ let cs = wire_col. character_set ( ) ;
157+ // 63 = binary collation (binary/blob columns).
158+ let cs_str = if cs == 63 {
159+ "binary" . to_string ( )
160+ } else {
161+ format ! ( "charset={cs}" )
162+ } ;
163+ let wire_type = format ! ( "{:?}" , wire_col. column_type( ) ) ;
164+
165+ let matched_col = if fallback_names {
166+ table_desc. columns . get ( i)
167+ } else {
168+ table_desc
169+ . columns
170+ . iter ( )
171+ . find ( |c| c. name . as_str ( ) == wire_name)
172+ } ;
173+ let match_info = match matched_col {
174+ Some ( col) => match & col. column_type {
175+ Some ( ct) => format ! ( "expected={:?}" , ct. scalar_type) ,
176+ None => "ignored" . to_string ( ) ,
177+ } ,
178+ None => "extra" . to_string ( ) ,
179+ } ;
180+
181+ let val_desc = match row. as_ref ( i) {
182+ None => "absent" . to_string ( ) ,
183+ Some ( Value :: NULL ) => "null" . to_string ( ) ,
184+ Some ( Value :: Bytes ( b) ) => format ! ( "bytes(len={})" , b. len( ) ) ,
185+ Some ( Value :: Int ( _) ) => "int" . to_string ( ) ,
186+ Some ( Value :: UInt ( _) ) => "uint" . to_string ( ) ,
187+ Some ( Value :: Float ( _) ) => "float" . to_string ( ) ,
188+ Some ( Value :: Double ( _) ) => "double" . to_string ( ) ,
189+ Some ( Value :: Date ( ..) ) => "date" . to_string ( ) ,
190+ Some ( Value :: Time ( ..) ) => "time" . to_string ( ) ,
191+ } ;
192+
193+ let _ = write ! (
194+ out,
195+ "{{name={wire_name}, wire={wire_type}, {cs_str}, {match_info}, val={val_desc}}}"
196+ ) ;
197+ }
198+ out. push ( ']' ) ;
199+ out
200+ }
201+
71202// TODO(guswynn|roshan): This function has various `.to_string()` and `format!` calls that should
72203// use a shared allocation if possible.
73204fn pack_val_as_datum (
0 commit comments