HuangXingBo commented on a change in pull request #13371:
URL: https://github.com/apache/flink/pull/13371#discussion_r488448820
##########
File path: flink-python/pyflink/fn_execution/beam/beam_coder_impl_slow.py
##########
@@ -75,62 +82,106 @@ def decode_from_stream(self, in_stream, nested):
in_stream.read_var_int64()
yield self._decode_one_row_from_stream(in_stream, nested)
- def _decode_one_row_from_stream(self, in_stream: create_InputStream,
nested: bool) -> List:
- null_mask = self._read_null_mask(in_stream)
- return [None if null_mask[idx] else
self._field_coders[idx].decode_from_stream(
- in_stream, nested) for idx in range(0, self._field_count)]
+ def _decode_one_row_from_stream(
+ self, in_stream: create_InputStream, nested: bool) -> List:
+ mask = self._read_mask(in_stream)
+ # ignore the row kind value as it is unnecessary for stateless
operation
+ return [None if mask[idx + ROW_KIND_BIT_SIZE] else
+ self._field_coders[idx].decode_from_stream(
+ in_stream, nested) for idx in range(0, self._field_count)]
- def _write_null_mask(self, value, out_stream):
+ def _write_mask(self, value, out_stream, row_kind_value=0):
field_pos = 0
- null_byte_search_table = self.null_byte_search_table
+ mask_byte_search_table = self.mask_byte_search_table
remaining_bits_num = self._remaining_bits_num
- for _ in range(self._leading_complete_bytes_num):
+
+ # first byte contains the row kind bits
+ b = self.row_kind_byte_table[row_kind_value]
+ for i in range(0, 8 - ROW_KIND_BIT_SIZE):
+ if field_pos + i < len(value) and value[field_pos + i] is None:
+ b |= mask_byte_search_table[i + ROW_KIND_BIT_SIZE]
+ field_pos += 8 - ROW_KIND_BIT_SIZE
+ out_stream.write_byte(b)
+
+ for _ in range(1, self._leading_complete_bytes_num):
b = 0x00
for i in range(0, 8):
if value[field_pos + i] is None:
- b |= null_byte_search_table[i]
+ b |= mask_byte_search_table[i]
field_pos += 8
out_stream.write_byte(b)
- if remaining_bits_num:
+ if self._leading_complete_bytes_num >= 1 and remaining_bits_num:
b = 0x00
for i in range(remaining_bits_num):
if value[field_pos + i] is None:
- b |= null_byte_search_table[i]
+ b |= mask_byte_search_table[i]
out_stream.write_byte(b)
- def _read_null_mask(self, in_stream):
- null_mask = []
- null_mask_search_table = self.null_mask_search_table
+ def _read_mask(self, in_stream):
+ mask = []
+ mask_search_table = self.mask_search_table
remaining_bits_num = self._remaining_bits_num
for _ in range(self._leading_complete_bytes_num):
b = in_stream.read_byte()
- null_mask.extend(null_mask_search_table[b])
+ mask.extend(mask_search_table[b])
if remaining_bits_num:
b = in_stream.read_byte()
- null_mask.extend(null_mask_search_table[b][0:remaining_bits_num])
- return null_mask
+ mask.extend(mask_search_table[b][0:remaining_bits_num])
+ return mask
def __repr__(self):
return 'FlattenRowCoderImpl[%s]' % ', '.join(str(c) for c in
self._field_coders)
-class RowCoderImpl(FlattenRowCoderImpl):
+class FlattenRowCoderWithRowKindImpl(FlattenRowCoderImpl):
Review comment:
Why we need to introduce `FlattenRowCoderWithRowKindImpl `? What about
moving the override method `_decode_one_row_from_stream ` to `RowCoderImpl`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
