%% Copyright (c) 2023 Peter Morgan %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. %% You may obtain a copy of the License at %% %% http://www.apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, software %% distributed under the License is distributed on an "AS IS" BASIS, %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% See the License for the specific language governing permissions and %% limitations under the License. %% @doc Replication protocol support for decoding of fields -module(msmp_binlog_field). -export([decode/3]). -import(scran_bytes, [take/1]). -import(scran_combinator, [map_result/2]). -import(scran_combinator, [rest/0]). -include_lib("kernel/include/logger.hrl"). decode(long, Unsigned, _Metadata) -> fun (Input) -> (fixed(4, Unsigned))(Input) end; decode(longlong = Type, Unsigned, Metadata) -> fun (Input) -> ?LOG_DEBUG(#{type => Type, unsigned => Unsigned, metadata => Metadata, input => Input}), (fixed(8, Unsigned))(Input) end; decode(tiny, Unsigned, _Metadata) -> fun (Input) -> (fixed(1, Unsigned))(Input) end; decode(short, Unsigned, _Metadata) -> fun (Input) -> (fixed(2, Unsigned))(Input) end; decode(int24, Unsigned, _) -> fun (Input) -> (fixed(3, Unsigned))(Input) end; decode(string, _, #{length := Length, field_type := enum}) -> fun (Input) -> (msmp_integer_fixed:decode(Length))(Input) end; decode(string, _, #{length := Length, field_type := string}) when Length < 256 -> fun (Input) -> (scran_bytes:length_encoded( msmp_integer_fixed:decode(1)))(Input) end; decode(string, _, #{field_type := string}) -> fun (Input) -> (scran_bytes:length_encoded( msmp_integer_fixed:decode(2)))(Input) end; decode(blob = Type, Unsigned, Length) -> fun (Input) -> ?LOG_DEBUG(#{type => Type, unsigned => Unsigned, length => Length, input => Input}), (scran_bytes:length_encoded( msmp_integer_fixed:decode(Length)))(Input) end; decode(varchar, _, Length) when is_integer(Length), Length < 256 -> fun (Input) -> (scran_bytes:length_encoded( msmp_integer_fixed:decode(1)))(Input) end; decode(varchar, _, Length) when is_integer(Length) -> fun (Input) -> (scran_bytes:length_encoded( msmp_integer_fixed:decode(2)))(Input) end; decode(date, Unsigned, _Metadata) -> fun (Input) -> (map_result( fixed(3, Unsigned), fun (Encoded) -> <> = <>, {Year, Month, Date} end))(Input) end; decode(time, Unsigned, _Metadata) -> fun (Input) -> (map_result( fixed(3, Unsigned), fun (Encoded) -> {Encoded div 10_000, (Encoded div 100) rem 100, Encoded rem 100} end))(Input) end; decode(time2, _, 0) -> fun (Input) -> (map_result( take(3), fun (<>) -> case Encoded - 16#80_00_00 of Negative when Negative < 0 -> <<_:2, H:10, M:6, S:6>> = <<(abs(Negative)):24>>, {-H, M, S}; Positive -> <<_:2, H:10, M:6, S:6>> = <>, {H, M, S} end end))(Input) end; decode(time2, _, Metadata) when Metadata == 5; Metadata == 6 -> fun (Input) -> (map_result( take(6), fun (<>) -> case Encoded - 16#80_00_00_00_00_00 of Negative when Negative < 0 -> <<_:2, H:10, M:6, S:6, Frac:24>> = <<(abs(Negative)):48>>, {-H, M, S + (Frac / 1_000_000)}; Positive -> <<_:2, H:10, M:6, S:6, Frac:24>> = <>, {H, M, S + (Frac / 1_000_000)} end end))(Input) end; decode(time2, _, Metadata) -> FracBytes = (Metadata + 1) div 2, {Delta, Divisor} = case Metadata of Metadata when Metadata =< 2 -> {16#1_00, 100}; Metadata when Metadata =< 4 -> {16#1_00_00, 10_000} end, fun (Input) -> (map_result( take(3 + FracBytes), fun (<>) -> case IntPart - 16#80_00_00 of Negative when Negative < 0 -> <<_:2, H:10, M:6, S:6>> = <<(abs(Negative)):24>>, {-H, M, S - 1 - ((FracPart - Delta) / Divisor)}; Positive -> <<_:2, H:10, M:6, S:6>> = <>, {H, M, S + (FracPart / Divisor)} end end))(Input) end; decode(timestamp2, _, 0) -> fun (Input) -> (map_result( take(4), fun (<>) -> erlang:convert_time_unit( Second, second, microsecond) end))(Input) end; decode(timestamp2, _, Decimals) when Decimals =< 2 -> fun (Input) -> (map_result( take(5), fun (<>) -> erlang:convert_time_unit( Second, second, microsecond) + (Micro * 10_000) end))(Input) end; decode(timestamp2, _, Decimals) when Decimals =< 4 -> fun (Input) -> (map_result( take(6), fun (<>) -> erlang:convert_time_unit( Second, second, microsecond) + (Micro * 100) end))(Input) end; decode(timestamp2, _, Decimals) when Decimals =< 6 -> fun (Input) -> (map_result( take(7), fun (<>) -> erlang:convert_time_unit( Second, second, microsecond) + Micro end))(Input) end; decode(year, _, _) -> fun (Input) -> (map_result( take(1), fun (<<0:8>>) -> 0; (<>) -> Year + 1_900 end))(Input) end; decode(datetime = Type, Unsigned, Metadata = 0) -> fun (Input) -> ?LOG_DEBUG(#{type => Type, unsigned => Unsigned, metadata => Metadata, input => Input}), (map_result( take(5), fun (<<_:1, YM:17, Day:5, Hour:5, Min:6, Sec:6>>) -> {{YM div 13, YM rem 13, Day}, {Hour, Min, Sec}} end))(Input) end; decode(datetime2 = Type, Unsigned, Metadata) when Metadata >= 5 -> fun (Input) -> ?LOG_DEBUG(#{type => Type, unsigned => Unsigned, metadata => Metadata, input => Input}), (map_result( take(8), fun (<<_:1, YM:17, Day:5, Hour:5, Min:6, Sec:6, Micro:24>>) -> datetime_to_system_microsecond( {{YM div 13, YM rem 13, Day}, {Hour, Min, Sec}}, Micro) %% {{YM div 13, YM rem 13, Day}, {Hour, Min, Sec + (Micro / 1_000_000)}} end))(Input) end; decode(datetime2 = Type, Unsigned, Metadata) when Metadata >= 3 -> fun (Input) -> ?LOG_DEBUG(#{type => Type, unsigned => Unsigned, metadata => Metadata, input => Input}), (map_result( take(7), fun (<<_:1, YM:17, Day:5, Hour:5, Min:6, Sec:6, Micro:16>>) -> datetime_to_system_microsecond( {{YM div 13, YM rem 13, Day}, {Hour, Min, Sec}}, Micro * 100) %% {{YM div 13, YM rem 13, Day}, {Hour, Min, Sec + (Micro * 100 / 1_000_000)}} end))(Input) end; decode(datetime2 = Type, Unsigned, Metadata) when Metadata >= 1 -> fun (Input) -> ?LOG_DEBUG(#{type => Type, unsigned => Unsigned, metadata => Metadata, input => Input}), (map_result( take(6), fun (<<_:1, YM:17, Day:5, Hour:5, Min:6, Sec:6, Micro:8>>) -> datetime_to_system_microsecond( {{YM div 13, YM rem 13, Day}, {Hour, Min, Sec}}, erlang:convert_time_unit( Micro * 10, millisecond, microsecond)) %% datetime_to_system_microsecond({{YM div 13, YM rem 13, Day}, {Hour, Min, Sec}}) %% + erlang:convert_time_unit( %% Micro * 10, %% millisecond, %% microsecond) %% + (Micro * 10_000 / 1_000_000) end))(Input) end; decode(datetime2 = Type, Unsigned, 0 = Metadata) -> fun (Input) -> ?LOG_DEBUG(#{type => Type, unsigned => Unsigned, metadata => Metadata, input => Input}), (map_result( take(5), fun (<<_:1, YM:17, Day:5, Hour:5, Min:6, Sec:6>>) -> datetime_to_system_microsecond({{YM div 13, YM rem 13, Day}, {Hour, Min, Sec}}) end))(Input) end; decode(float, _, _) -> fun (Input) -> (map_result( scran_number_le:f32(), scran_number:precision(6)))(Input) end; decode(double, _, _) -> fun (Input) -> (map_result( scran_number_le:f64(), scran_number:precision(15)))(Input) end; decode(newdecimal, _, #{precision := Precision, scale := Scale}) -> msmp_decimal:decode(Precision, Scale); decode(bit, _, 0) -> fun (Input) -> (fixed(1, false))(Input) end; decode(bit, _, Bits) -> fun (Input) -> (map_result( take((Bits + 7) div 8), fun (<<_:(8 - (Bits rem 8))/bits, Value:Bits/bits>>) -> Value end))(Input) end; decode(json, _, Metadata) -> msmp_jsonb:decode(Metadata); decode(Type, Unsigned, Metadata) -> fun (Input) -> ?LOG_WARNING(#{type => Type, unsigned => Unsigned, metadata => Metadata, input => Input}), (rest())(Input) end. datetime_to_system_microsecond(DateTime) -> ?FUNCTION_NAME(DateTime, 0). datetime_to_system_microsecond(DateTime, Microsecond) -> case erlang:convert_time_unit( calendar:datetime_to_gregorian_seconds(DateTime) - epoch(posix), second, microsecond) of SystemTime when SystemTime >= 0 -> SystemTime + Microsecond; SystemTime -> SystemTime - Microsecond end. epoch(posix) -> calendar:datetime_to_gregorian_seconds({{1970, 1, 1}, {0, 0, 0}}). fixed(Bytes, true) -> scran_number_le:u(Bytes * 8); fixed(Bytes, false) -> scran_number_le:i(Bytes * 8).