Skip to content

refactor : refactor loop inside _read_column_data method #418

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Aug 26, 2023
Merged

refactor : refactor loop inside _read_column_data method #418

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
10974a1
refactor : refactor loop inside _read_column_data method
mirageoasis Aug 14, 2023
a1f2b3f
Merge branch 'main' into refactor_read_column_data
mirageoasis Aug 26, 2023
ea12224
Merge branch 'main' into refactor_read_column_data
mirageoasis Aug 26, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
226 changes: 118 additions & 108 deletions pymysqlreplication/row_event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ def _read_column_data(self, cols_bitmap):
# See http://dev.mysql.com/doc/internals/en/rows-event.html
null_bitmap = self.packet.read((BitCount(cols_bitmap) + 7) / 8)

nullBitmapIndex = 0
null_bitmap_index = 0
nb_columns = len(self.columns)
for i in range(0, nb_columns):
column = self.columns[i]
Expand All @@ -112,117 +112,127 @@ def _read_column_data(self, cols_bitmap):
zerofill = self.table_map[self.table_id].columns[i].zerofill
fixed_binary_length = self.table_map[self.table_id].columns[i].fixed_binary_length

if BitGet(cols_bitmap, i) == 0:
values[name] = None
continue

if self._is_null(null_bitmap, nullBitmapIndex):
values[name] = None
elif column.type == FIELD_TYPE.TINY:
if unsigned:
values[name] = struct.unpack("<B", self.packet.read(1))[0]
if zerofill:
values[name] = format(values[name], '03d')
else:
values[name] = struct.unpack("<b", self.packet.read(1))[0]
elif column.type == FIELD_TYPE.SHORT:
if unsigned:
values[name] = struct.unpack("<H", self.packet.read(2))[0]
if zerofill:
values[name] = format(values[name], '05d')
else:
values[name] = struct.unpack("<h", self.packet.read(2))[0]
elif column.type == FIELD_TYPE.LONG:
if unsigned:
values[name] = struct.unpack("<I", self.packet.read(4))[0]
if zerofill:
values[name] = format(values[name], '010d')
else:
values[name] = struct.unpack("<i", self.packet.read(4))[0]
elif column.type == FIELD_TYPE.INT24:
if unsigned:
values[name] = self.packet.read_uint24()
if zerofill:
values[name] = format(values[name], '08d')
else:
values[name] = self.packet.read_int24()
elif column.type == FIELD_TYPE.FLOAT:
values[name] = struct.unpack("<f", self.packet.read(4))[0]
elif column.type == FIELD_TYPE.DOUBLE:
values[name] = struct.unpack("<d", self.packet.read(8))[0]
elif column.type == FIELD_TYPE.VARCHAR or \
column.type == FIELD_TYPE.STRING:
if column.max_length > 255:
values[name] = self.__read_string(2, column)
else:
values[name] = self.__read_string(1, column)

if fixed_binary_length and len(values[name]) < fixed_binary_length:
# Fixed-length binary fields are stored in the binlog
# without trailing zeros and must be padded with zeros up
# to the specified length at read time.
nr_pad = fixed_binary_length - len(values[name])
values[name] += b'\x00' * nr_pad

elif column.type == FIELD_TYPE.NEWDECIMAL:
values[name] = self.__read_new_decimal(column)
elif column.type == FIELD_TYPE.BLOB:
values[name] = self.__read_string(column.length_size, column)
elif column.type == FIELD_TYPE.DATETIME:
values[name] = self.__read_datetime()
elif column.type == FIELD_TYPE.TIME:
values[name] = self.__read_time()
elif column.type == FIELD_TYPE.DATE:
values[name] = self.__read_date()
elif column.type == FIELD_TYPE.TIMESTAMP:
values[name] = datetime.datetime.utcfromtimestamp(
self.packet.read_uint32())

# For new date format:
elif column.type == FIELD_TYPE.DATETIME2:
values[name] = self.__read_datetime2(column)
elif column.type == FIELD_TYPE.TIME2:
values[name] = self.__read_time2(column)
elif column.type == FIELD_TYPE.TIMESTAMP2:
values[name] = self.__add_fsp_to_time(
datetime.datetime.utcfromtimestamp(
self.packet.read_int_be_by_size(4)), column)
elif column.type == FIELD_TYPE.LONGLONG:
if unsigned:
values[name] = self.packet.read_uint64()
if zerofill:
values[name] = format(values[name], '020d')
else:
values[name] = self.packet.read_int64()
elif column.type == FIELD_TYPE.YEAR:
values[name] = self.packet.read_uint8() + 1900
elif column.type == FIELD_TYPE.ENUM:
values[name] = column.enum_values[
self.packet.read_uint_by_size(column.size)]
elif column.type == FIELD_TYPE.SET:
# We read set columns as a bitmap telling us which options
# are enabled
bit_mask = self.packet.read_uint_by_size(column.size)
values[name] = set(
val for idx, val in enumerate(column.set_values)
if bit_mask & 2 ** idx
) or None

elif column.type == FIELD_TYPE.BIT:
values[name] = self.__read_bit(column)
elif column.type == FIELD_TYPE.GEOMETRY:
values[name] = self.packet.read_length_coded_pascal_string(
column.length_size)
elif column.type == FIELD_TYPE.JSON:
values[name] = self.packet.read_binary_json(column.length_size)
else:
raise NotImplementedError("Unknown MySQL column type: %d" %
(column.type))
values[name] = self.__read_values_name(column, null_bitmap, null_bitmap_index,
cols_bitmap, unsigned, zerofill,
fixed_binary_length, i)

nullBitmapIndex += 1
if BitGet(cols_bitmap, i) != 0:
null_bitmap_index += 1

return values

def __read_values_name(self, column, null_bitmap, null_bitmap_index, cols_bitmap, unsigned, zerofill,
fixed_binary_length, i):

if BitGet(cols_bitmap, i) == 0:
return None

if self._is_null(null_bitmap, null_bitmap_index):
return None

if column.type == FIELD_TYPE.TINY:
if unsigned:
ret = struct.unpack("<B", self.packet.read(1))[0]
if zerofill:
ret = format(ret, '03d')
return ret
else:
return struct.unpack("<b", self.packet.read(1))[0]
elif column.type == FIELD_TYPE.SHORT:
if unsigned:
ret = struct.unpack("<H", self.packet.read(2))[0]
if zerofill:
ret = format(ret, '05d')
return ret
else:
return struct.unpack("<h", self.packet.read(2))[0]
elif column.type == FIELD_TYPE.LONG:
if unsigned:
ret = struct.unpack("<I", self.packet.read(4))[0]
if zerofill:
ret = format(ret, '010d')
return ret
else:
return struct.unpack("<i", self.packet.read(4))[0]
elif column.type == FIELD_TYPE.INT24:
if unsigned:
ret = self.packet.read_uint24()
if zerofill:
ret = format(ret, '08d')
return ret
else:
return self.packet.read_int24()
elif column.type == FIELD_TYPE.FLOAT:
return struct.unpack("<f", self.packet.read(4))[0]
elif column.type == FIELD_TYPE.DOUBLE:
return struct.unpack("<d", self.packet.read(8))[0]
elif column.type == FIELD_TYPE.VARCHAR or \
column.type == FIELD_TYPE.STRING:
ret = self.__read_string(2, column) if column.max_length > 255 else self.__read_string(1, column)

if fixed_binary_length and len(ret) < fixed_binary_length:
# Fixed-length binary fields are stored in the binlog
# without trailing zeros and must be padded with zeros up
# to the specified length at read time.
nr_pad = fixed_binary_length - len(ret)
ret += b'\x00' * nr_pad
return ret
elif column.type == FIELD_TYPE.NEWDECIMAL:
return self.__read_new_decimal(column)
elif column.type == FIELD_TYPE.BLOB:
return self.__read_string(column.length_size, column)
elif column.type == FIELD_TYPE.DATETIME:
return self.__read_datetime()
elif column.type == FIELD_TYPE.TIME:
return self.__read_time()
elif column.type == FIELD_TYPE.DATE:
return self.__read_date()
elif column.type == FIELD_TYPE.TIMESTAMP:
return datetime.datetime.fromtimestamp(
self.packet.read_uint32())

# For new date format:
elif column.type == FIELD_TYPE.DATETIME2:
return self.__read_datetime2(column)
elif column.type == FIELD_TYPE.TIME2:
return self.__read_time2(column)
elif column.type == FIELD_TYPE.TIMESTAMP2:
return self.__add_fsp_to_time(
datetime.datetime.fromtimestamp(
self.packet.read_int_be_by_size(4)), column)
elif column.type == FIELD_TYPE.LONGLONG:
if unsigned:
ret = self.packet.read_uint64()
if zerofill:
ret = format(ret, '020d')
return ret
else:
return self.packet.read_int64()
elif column.type == FIELD_TYPE.YEAR:
return self.packet.read_uint8() + 1900
elif column.type == FIELD_TYPE.ENUM:
return column.enum_values[
self.packet.read_uint_by_size(column.size)]
elif column.type == FIELD_TYPE.SET:
# We read set columns as a bitmap telling us which options
# are enabled
bit_mask = self.packet.read_uint_by_size(column.size)
return set(
val for idx, val in enumerate(column.set_values)
if bit_mask & 2 ** idx
) or None

elif column.type == FIELD_TYPE.BIT:
return self.__read_bit(column)
elif column.type == FIELD_TYPE.GEOMETRY:
return self.packet.read_length_coded_pascal_string(
column.length_size)
elif column.type == FIELD_TYPE.JSON:
return self.packet.read_binary_json(column.length_size)
else:
raise NotImplementedError("Unknown MySQL column type: %d" %
(column.type))

def __add_fsp_to_time(self, time, column):
"""Read and add the fractional part of time
For more details about new date format:
Expand Down