JingsongLi commented on code in PR #9253:
URL: https://github.com/apache/paimon/pull/9253#discussion_r3800913854
##########
paimon-python/pypaimon/data/variant_path.py:
##########
@@ -1473,3 +1637,369 @@ def variant_replace(
if not chunked:
return result_chunks[0]
return pa.chunked_array(result_chunks, type=data_type)
+
+
+def _build_object_value_ordered(fields):
+ """Build object value bytes keeping the given field order."""
+ if not fields:
+ return _build_object_value([])
+ size = len(fields)
+ data_size = sum(len(child) for _, child in fields)
+ large_size = size > _U8_MAX
+ size_bytes = _U32_SIZE if large_size else 1
+ max_id = max(field_id for field_id, _ in fields)
+ id_size = _get_int_size(max_id)
+ offset_size = _get_int_size(data_size) if data_size > 0 else 1
+ buf = bytearray()
+ buf.append(_object_header(large_size, id_size, offset_size))
+ buf += size.to_bytes(size_bytes, 'little')
+ for field_id, _ in fields:
+ buf += field_id.to_bytes(id_size, 'little')
+ offset = 0
+ for _, child in fields:
+ buf += offset.to_bytes(offset_size, 'little')
+ offset += len(child)
+ buf += offset.to_bytes(offset_size, 'little')
+ for _, child in fields:
+ buf += child
+ return bytes(buf)
+
+
+def _apply_edits(
+ value,
+ pos,
+ limit,
+ edits,
+ key_ids,
+ names_by_id,
+ source_metadata_size=None,
+):
+ """Apply edits and validate source ids before metadata extension."""
+ results = {}
+ next_token = 1
+ stack = [('visit', 0, pos, limit, edits)]
+ while stack:
+ action = stack.pop()
+ kind = action[0]
+ if kind == 'finish_object':
+ _, token, ids, children, inserts, child_tokens = action
+ for slot, child_token in child_tokens:
+ children[slot] = results.pop(child_token)
+ fields = list(zip(ids, children)) + inserts
+ if len({field_id for field_id, _ in fields}) != len(fields):
+ _malformed("duplicate object field id")
+ if inserts:
+ try:
+ fields.sort(
+ key=lambda field: names_by_id[
+ field[0]].encode('utf-8'))
+ except KeyError:
+ _malformed("object key is missing from metadata")
+ results[token] = _build_object_value_ordered(fields)
+ continue
+ if kind == 'finish_array':
+ _, token, children, child_tokens = action
+ for index, child_token in child_tokens:
+ children[index] = results.pop(child_token)
+ results[token] = _build_array_value(children)
Review Comment:
[P2] Avoid materializing the entire rebuilt child at every level. Both
finish branches copy the completed subtree into a new `bytes` value, so a deep
update is O(depth²) even though the traversal itself is iterative. On this
head, inserting into one valid 16,000-level, approximately 106 KB value took
about 4.1 seconds. Please accumulate fragments or precompute the final sizes
and write each output byte only once.
##########
paimon-common/src/main/java/org/apache/paimon/data/variant/GenericVariantUtil.java:
##########
@@ -145,6 +146,19 @@ public class GenericVariantUtil {
public static final int BINARY_SEARCH_THRESHOLD = 32;
+ static int compareUnsignedUtf8(String left, String right) {
+ byte[] leftBytes = left.getBytes(StandardCharsets.UTF_8);
+ byte[] rightBytes = right.getBytes(StandardCharsets.UTF_8);
Review Comment:
[P2] Avoid re-encoding both keys on every comparison. This helper runs
inside both the writer sort and every `getFieldByKey` binary-search probe.
Extracting all 32 fields from a 32-field object averages about 4.22 probes per
lookup, so the reader alone creates roughly 270 temporary byte arrays per row.
`BinaryString.compareTo` already implements unsigned UTF-8 byte ordering;
please encode/cache each `FieldEntry` key once, encode the lookup key once, and
compare it against the metadata byte slice without allocating per probe.
##########
paimon-python/pypaimon/data/variant_path.py:
##########
@@ -1473,3 +1637,369 @@ def variant_replace(
if not chunked:
return result_chunks[0]
return pa.chunked_array(result_chunks, type=data_type)
+
+
+def _build_object_value_ordered(fields):
+ """Build object value bytes keeping the given field order."""
+ if not fields:
+ return _build_object_value([])
+ size = len(fields)
+ data_size = sum(len(child) for _, child in fields)
+ large_size = size > _U8_MAX
+ size_bytes = _U32_SIZE if large_size else 1
+ max_id = max(field_id for field_id, _ in fields)
+ id_size = _get_int_size(max_id)
+ offset_size = _get_int_size(data_size) if data_size > 0 else 1
+ buf = bytearray()
+ buf.append(_object_header(large_size, id_size, offset_size))
+ buf += size.to_bytes(size_bytes, 'little')
+ for field_id, _ in fields:
+ buf += field_id.to_bytes(id_size, 'little')
+ offset = 0
+ for _, child in fields:
+ buf += offset.to_bytes(offset_size, 'little')
+ offset += len(child)
+ buf += offset.to_bytes(offset_size, 'little')
+ for _, child in fields:
+ buf += child
+ return bytes(buf)
+
+
+def _apply_edits(
+ value,
+ pos,
+ limit,
+ edits,
+ key_ids,
+ names_by_id,
+ source_metadata_size=None,
+):
+ """Apply edits and validate source ids before metadata extension."""
+ results = {}
+ next_token = 1
+ stack = [('visit', 0, pos, limit, edits)]
+ while stack:
+ action = stack.pop()
+ kind = action[0]
+ if kind == 'finish_object':
+ _, token, ids, children, inserts, child_tokens = action
+ for slot, child_token in child_tokens:
+ children[slot] = results.pop(child_token)
+ fields = list(zip(ids, children)) + inserts
+ if len({field_id for field_id, _ in fields}) != len(fields):
+ _malformed("duplicate object field id")
+ if inserts:
+ try:
+ fields.sort(
+ key=lambda field: names_by_id[
+ field[0]].encode('utf-8'))
+ except KeyError:
+ _malformed("object key is missing from metadata")
+ results[token] = _build_object_value_ordered(fields)
+ continue
+ if kind == 'finish_array':
+ _, token, children, child_tokens = action
+ for index, child_token in child_tokens:
+ children[index] = results.pop(child_token)
+ results[token] = _build_array_value(children)
+ continue
+
+ _, token, node_pos, node_limit, node_edits = action
+ value_end = node_pos + _checked_value_size(
+ value, node_pos, node_limit)
+ if value_end != node_limit:
+ _malformed("child size does not match container offsets")
+ inserts = []
+ descend = {}
+ replacement = None
+ for segments, op, key_id, payload in node_edits:
+ if op == 'replace' and not segments:
+ replacement = payload
+ break
+ if op == 'insert' and len(segments) == 1:
+ inserts.append((key_id, payload))
+ else:
+ descend.setdefault(segments[0], []).append(
+ (segments[1:], op, key_id, payload))
+ if replacement is not None:
+ results[token] = replacement
+ continue
+
+ basic_type = value[node_pos] & 0x3
+ child_actions = []
+ if basic_type == _OBJECT:
+ size, id_size, id_start, data_start, offsets, _ = (
+ _checked_object_layout(value, node_pos, value_end))
+ ids = [
+ _read_unsigned(value, id_start + i * id_size, id_size)
+ for i in range(size)
+ ]
+ if (source_metadata_size is not None
+ and any(field_id >= source_metadata_size
+ for field_id in ids)):
+ _malformed("object field id is missing from metadata")
+ ordered_offsets = sorted(offsets)
+ end_by_offset = dict(zip(
+ ordered_offsets, ordered_offsets[1:]))
+ slot_by_id = {
+ field_id: index for index, field_id in enumerate(ids)
+ }
+ edits_by_slot = {}
+ for (_, segment), child_edits in descend.items():
+ slot = slot_by_id[key_ids[segment]]
+ edits_by_slot[slot] = child_edits
+ children = []
+ child_tokens = []
+ for slot in range(size):
+ child_pos, child_end = _checked_object_child_bounds(
+ value, data_start, offsets, slot, end_by_offset)
+ child_edits = edits_by_slot.get(slot)
+ if child_edits is not None:
+ child_token = next_token
+ next_token += 1
+ children.append(None)
+ child_tokens.append((slot, child_token))
+ child_actions.append((
+ 'visit', child_token, child_pos, child_end,
+ child_edits,
+ ))
+ else:
+ if (source_metadata_size is not None
+ and (value[child_pos] & 0x3)
+ in (_OBJECT, _ARRAY)):
+ _validate_value_field_ids(
+ value, child_pos, child_end,
+ source_metadata_size)
+ children.append(bytes(value[child_pos:child_end]))
+ stack.append((
+ 'finish_object', token, ids, children, inserts,
+ child_tokens,
+ ))
+ elif basic_type == _ARRAY:
+ size, data_start, offsets, _ = _checked_array_layout(
+ value, node_pos, value_end)
+ edits_by_index = {
+ segment: child_edits
+ for (_, segment), child_edits in descend.items()
+ }
+ children = []
+ child_tokens = []
+ for index in range(size):
+ child_pos = data_start + offsets[index]
+ child_end = data_start + offsets[index + 1]
+ child_edits = edits_by_index.get(index)
+ if child_edits is not None:
+ child_token = next_token
+ next_token += 1
+ children.append(None)
+ child_tokens.append((index, child_token))
+ child_actions.append((
+ 'visit', child_token, child_pos, child_end,
+ child_edits,
+ ))
+ else:
+ if source_metadata_size is not None:
+ _validate_value_field_ids(
+ value, child_pos, child_end,
+ source_metadata_size)
+ children.append(bytes(value[child_pos:child_end]))
+ stack.append((
+ 'finish_array', token, children, child_tokens,
+ ))
+ else:
+ _malformed("path segment does not match the value type")
+ stack.extend(child_actions)
+ return results[0]
+
+
+def _set_chunk(chunk, values, parsed, global_row):
+ parsed_paths = [parsed_path for _, parsed_path, _ in parsed]
+ parent_paths = [parsed_path[:-1] for parsed_path in parsed_paths]
+ query_paths = tuple(parsed_paths) + tuple(parent_paths)
+ count = len(parsed)
+ metadata_column = chunk.field(1)
+ valid_rows = _valid_row_indices(chunk, values, metadata_column)
+ if not len(valid_rows):
+ return chunk
+ plans, slow_rows = _partition_path_plans(
+ values, metadata_column, valid_rows, query_paths)
+ slow_rows = set(int(row) for row in slow_rows)
+ metadata_values = _BinaryValues(metadata_column)
+ state = _PatchState(values)
+ rebuilt_rows = {}
+ rebuilt_metadata = {}
+ scalar_payloads = {}
+
+ def payload_for(index, provider, row):
+ if provider._array is not None:
+ return provider.encode(provider.scalar_at(global_row + row))
+ if index not in scalar_payloads:
+ scalar_payloads[index] = provider.encode(provider.scalar_at(0))
+ return scalar_payloads[index]
+
+ def rebuild_row(row, view, insert_set, key_ids, names_by_id,
+ new_metadata, source_metadata_size=None,
+ validated_positions=None):
+ edits = []
+ for index, (path, parsed_path, provider) in enumerate(parsed):
+ payload = payload_for(index, provider, row)
+ if index in insert_set:
+ edits.append((
+ parsed_path, 'insert',
+ key_ids[parsed_path[-1][1]], payload))
+ else:
+ if validated_positions is not None:
+ provider.validate_source(
+ view, validated_positions[index])
+ edits.append((parsed_path, 'replace', None, payload))
+ rebuilt = _apply_edits(
+ view, 0, len(view), edits, key_ids, names_by_id,
+ source_metadata_size)
+ if new_metadata is not None or rebuilt != view:
+ rebuilt_rows[row] = rebuilt
Review Comment:
[P2] Enforce the Java VARIANT size contract before storing this rebuilt row.
A variable-length replacement can make `rebuilt` exceed
`GenericVariantUtil.SIZE_LIMIT` (128 MiB), but this path still returns a valid
Arrow array; a Java reader later rejects it with
`VARIANT_CONSTRUCTOR_SIZE_LIMIT`. Please reject oversized rebuilt values and
`new_metadata` here, ideally through a shared Python constructor invariant, and
add boundary tests.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
