JingsongLi commented on code in PR #9253:
URL: https://github.com/apache/paimon/pull/9253#discussion_r3800557311
##########
paimon-python/pypaimon/data/generic_variant.py:
##########
@@ -388,7 +388,9 @@ def append_timestamp_ntz(self, micros_since_epoch):
self._write_le(micros_since_epoch & 0xFFFFFFFFFFFFFFFF, 8)
def _finish_writing_object(self, start, fields):
- fields.sort(key=lambda f: f[0])
+ # Java binary-searches fields by UTF-16 order (String.compareTo), so
sort
+ # by key name as UTF-16 code units, not by Python code point.
+ fields.sort(key=lambda f: f[0].encode('utf-16-be'))
Review Comment:
[P1] Keep object fields in Parquet's UTF-8 byte order. The Variant encoding
requires field IDs to be ordered lexicographically using unsigned UTF-8 bytes
(https://github.com/apache/parquet-format/blob/master/VariantEncoding.md#object-field-id-order-and-uniqueness).
UTF-16 reverses U+E000 (`EE...` in UTF-8) and U+10000 (`F0...`), so a
compliant binary-search reader can miss U+E000 in a wide object. Matching Java
`String.compareTo` spreads the existing Java ordering bug into the Python
writer. Please retain UTF-8 ordering here and in `variant_set`, and make the
Java reader compatible with both legacy UTF-16-ordered data and spec-compliant
data before switching writers.
##########
docs/docs/pypaimon/python-api.mdx:
##########
@@ -1149,6 +1149,24 @@ string-keyed map types. Replacement supports scalar
types. Missing paths read
as NULL and remain unchanged unless `strict=True` is specified. Pass mappings
to process multiple paths in one pass.
+`variant_set` upserts paths: existing paths are replaced like
+`variant_replace`, and a missing final key is inserted when its parent path
+exists and is an OBJECT:
+
+```python
+updated_payload = variant_set(payload, {
+ '$.velocity.y': pc.negate(current),
+ '$.processed': pa.scalar(True, type=pa.bool_()),
+})
+```
+
+Values may be a `pa.Scalar` (broadcast to every row) or a `pa.Array` /
+`pa.ChunkedArray` with one value per row; Arrow NULL values are stored as
Review Comment:
[P2] This promise does not hold for Arrow's `null` type. `variant_set(...,
pa.scalar(None))`, `pa.nulls(n)`, and a null-typed `ChunkedArray` all fail with
`Unsupported exact VARIANT replacement type: null` because
`_supported_replacement_type` excludes `pa.null()`; only explicitly typed nulls
work. Either support `pa.null()` as a universal VARIANT NULL or narrow this
text to "typed Arrow NULL" and add coverage for the rejected
scalar/array/chunked cases.
##########
paimon-python/pypaimon/data/variant_path.py:
##########
@@ -1473,3 +1636,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-16-be'))
+ 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
+ if new_metadata is not None:
+ rebuilt_metadata[row] = new_metadata
+
+ for planned in plans:
+ rows, row_starts, source_data, positions, limits = planned
+ target_positions = positions[:count]
+ target_limits = limits[:count]
+ parent_positions = positions[count:]
+ insert_indices = []
+ for index, (path, parsed_path, provider) in enumerate(parsed):
+ if target_positions[index] is not None:
+ continue
+ parent_pos = parent_positions[index]
+ if parent_pos is None:
+ raise ValueError(
+ f"VARIANT parent path does not exist: {path}")
+ if not parsed_path or parsed_path[-1][0] != 'key':
+ raise ValueError(
+ "VARIANT array index insertion is not supported: "
+ + path)
+ parent_headers = source_data[row_starts + parent_pos]
+ if np.any((parent_headers & 0x3) != _OBJECT):
+ raise ValueError(
+ f"VARIANT parent path is not an object: {path}")
+ insert_indices.append(index)
+ if not insert_indices and all(
+ provider._fixed_size is not None
+ for _, _, provider in parsed):
+ _patch_planned_group(
Review Comment:
[P2] Validate every row before taking this fixed-size patch path.
`_vectorized_path_positions` fully validates only the exemplar row; peer rows
verify the selected slot but not uniqueness of the complete field-ID table.
With a valid first row `{a,b}` and a second row encoded with duplicate IDs
`{a,a}`, `variant_set(column, '$.a', pa.scalar(9.0))` succeeds and the
malformed second row silently decodes as `{'a': 2.0}`. Please validate each
peer ID table or route unvalidated rows through the checked slow path, and add
a multi-row regression test.
--
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]
