JingsongLi commented on code in PR #9259:
URL: https://github.com/apache/paimon/pull/9259#discussion_r3810289048
##########
paimon-python/pypaimon/data/variant_path.py:
##########
@@ -1976,8 +2228,75 @@ def rebuild_row(row, view, insert_set, key_ids,
names_by_id,
new_metadata, key_ids, names_by_id = _metadata_with_keys(
first_metadata, insert_keys)
insert_set = set(insert_indices)
+ splice_eligible = (
+ len(insert_indices) == 1
+ and len(parsed[insert_indices[0]][1]) == 1
+ and all(
+ index in insert_set
+ or parsed[index][2]._fixed_size is not None
+ for index in range(count))
+ )
+ if splice_eligible:
+ insert_index = insert_indices[0]
+ replace_indices = [
+ index for index in range(count) if index != insert_index
+ ]
+ if replace_indices:
+ _patch_planned_group(
+ (rows, row_starts, source_data,
+ [target_positions[index] for index in replace_indices],
+ [target_limits[index] for index in replace_indices]),
+ [parsed[index] for index in replace_indices],
+ len(chunk), global_row, state, group_slow, False)
+ slow_rows |= group_slow
+ if group_slow:
+ keep = np.fromiter(
+ (int(row) not in group_slow for row in rows),
+ bool, len(rows))
+ live_rows = rows[keep]
+ live_starts = row_starts[keep]
+ live_lengths = parent_limits[insert_index][keep]
+ else:
+ live_rows = rows
+ live_starts = row_starts
+ live_lengths = parent_limits[insert_index]
+ provider = parsed[insert_index][2]
+ if provider._array is None:
+ payloads = [
+ payload_for(insert_index, provider, 0)
+ ] * len(live_rows)
+ else:
Review Comment:
[P1] Avoid retaining the entire variable-width payload column here
For an array-backed string or binary insert, this list eagerly encodes every
live row. `_root_insert_splice` then keeps the list alive while accumulating
every rebuilt row, so peak memory grows by roughly one extra copy of the whole
inserted payload. I reproduced this with 50,000 rows x 1 KiB: max RSS increased
from 370.8 MB on the base to 440.9 MB on this head (+70.1 MB); an independent
100,000-row run showed +134.9 MB. Large production chunks can therefore OOM
even though the NumPy index matrix is bounded. Please stream or byte-budget the
encoding/splice work, and add a peak-memory regression test, instead of
materializing the entire payload list.
##########
paimon-python/pypaimon/data/variant_path.py:
##########
@@ -1880,6 +1944,192 @@ def _apply_edits(
return _materialize_value(results[0])
+def _root_insert_splice_layout(value, key_id, key_name, names_by_id):
+ """Return a root layout that can splice the new field."""
+ header = value[0]
+ size, id_size, id_start, data_start, first_offsets, _ = (
+ _checked_object_layout(value, 0, len(value)))
+ ordered_offsets = sorted(first_offsets)
+ end_by_offset = dict(zip(ordered_offsets, ordered_offsets[1:]))
+ for index in range(size):
+ _checked_object_child_bounds(
+ value, data_start, first_offsets, index, end_by_offset)
+ type_info = (header >> 2) & 0x3F
+ large_size = ((type_info >> 4) & 0x1) != 0
+ size_width = _U32_SIZE if large_size else 1
+ offset_size = (type_info & 0x3) + 1
+ offset_start = id_start + size * id_size
+ if not large_size and size + 1 > _U8_MAX:
Review Comment:
[P2] Reject globally ineligible layouts before child validation
`_root_insert_splice_layout` validates every child before checking whether
the added field can fit the current size and id widths. Because the planner
scans rows until it finds an eligible layout, a batch where every row is
impossible to splice (for example, a 255-field object or `key_id == 256` with
one-byte ids) pays a full validation scan for every row and then rebuilds every
row anyway. For 1,000 rows with 255 fields, I measured 0.705 s versus 0.488 s
with immediate fallback (+44.4%). Please move the cheap size/key-width
rejection before child validation, or detect batch-wide ineligibility once.
##########
paimon-python/pypaimon/data/variant_path.py:
##########
@@ -1880,6 +1944,192 @@ def _apply_edits(
return _materialize_value(results[0])
+def _root_insert_splice_layout(value, key_id, key_name, names_by_id):
+ """Return a root layout that can splice the new field."""
+ header = value[0]
+ size, id_size, id_start, data_start, first_offsets, _ = (
+ _checked_object_layout(value, 0, len(value)))
+ ordered_offsets = sorted(first_offsets)
+ end_by_offset = dict(zip(ordered_offsets, ordered_offsets[1:]))
+ for index in range(size):
+ _checked_object_child_bounds(
+ value, data_start, first_offsets, index, end_by_offset)
+ type_info = (header >> 2) & 0x3F
+ large_size = ((type_info >> 4) & 0x1) != 0
+ size_width = _U32_SIZE if large_size else 1
+ offset_size = (type_info & 0x3) + 1
+ offset_start = id_start + size * id_size
+ if not large_size and size + 1 > _U8_MAX:
+ return None
+ if key_id >= 1 << (8 * id_size):
+ return None
+ ids = [
+ _read_unsigned(value, id_start + i * id_size, id_size)
+ for i in range(size)
+ ]
+ names = [names_by_id.get(field_id) for field_id in ids]
+ if any(name is None for name in names) or names != sorted(names):
+ return None
+ slot = sum(name < key_name for name in names)
+ return (
+ header, size, size_width, id_size, id_start, data_start,
+ offset_size, offset_start, ids, slot,
+ )
+
+
+def _plan_root_insert_splice(
+ values, rows, row_starts, row_lengths, source_data,
+ key_id, key_name, names_by_id, payloads):
+ """Plan a splice and identify rows matching one root layout."""
+ layout = None
+ for row in rows:
+ layout = _root_insert_splice_layout(
+ values.view(int(row)), key_id, key_name, names_by_id)
+ if layout is not None:
+ break
+ if layout is None:
+ return None
+ (
+ header, size, size_width, id_size, id_start, data_start,
+ offset_size, offset_start, ids, slot,
+ ) = layout
+
+ widths = np.full(len(rows), size_width, dtype=np.int64)
+ ok = source_data[row_starts] == header
+ ok &= row_lengths >= data_start
+ safe_starts = np.where(ok, row_starts, 0)
+ ok &= _take_unsigned(source_data, safe_starts + 1, widths) == size
+ widths = np.full(len(rows), id_size, dtype=np.int64)
+ for index in range(size):
+ ok &= _take_unsigned(
+ source_data,
+ safe_starts + id_start + index * id_size,
+ widths,
+ ) == ids[index]
+ widths = np.full(len(rows), offset_size, dtype=np.int64)
+ sentinels = _take_unsigned(
+ source_data,
+ safe_starts + offset_start + size * offset_size,
+ widths,
+ )
+ ok &= data_start + sentinels == row_lengths
+ if size:
+ minimum = None
+ for index in range(size):
+ entry = _take_unsigned(
+ source_data,
+ safe_starts + offset_start + index * offset_size,
+ widths,
+ )
+ ok &= entry < sentinels
+ minimum = entry if minimum is None else np.minimum(
+ minimum, entry)
+ ok &= minimum == 0
+ payload_lengths = np.fromiter(
+ (len(payload) for payload in payloads), np.int64, len(payloads))
+ new_sentinels = sentinels + payload_lengths
+ ok &= new_sentinels < 1 << (8 * offset_size)
+
+ return (
+ header, size, size_width, id_size, id_start, data_start,
+ offset_size, offset_start, slot, sentinels, ok,
+ )
+
+
+def _root_insert_splice(
+ values, state, rows, row_starts, row_lengths, source_data,
+ key_id, key_name, names_by_id, payloads,
+ source_metadata_size, output_metadata_size):
+ """Splice one field into uniform root objects."""
+ plan = _plan_root_insert_splice(
+ values, rows, row_starts, row_lengths, source_data,
+ key_id, key_name, names_by_id, payloads)
+ if plan is None:
+ return None
+ (
+ header, size, size_width, id_size, id_start, data_start,
+ offset_size, offset_start, slot, sentinels, ok,
+ ) = plan
+ matching_structures = None
+ if source_metadata_size is not None:
+ matching_structures = np.zeros(len(rows), dtype=bool)
+ candidates = np.flatnonzero(ok)
+ if len(candidates):
+ lengths = row_lengths[candidates]
+ order = np.argsort(lengths, kind='stable')
+ candidates = candidates[order]
+ lengths = lengths[order]
+ boundaries = np.flatnonzero(lengths[1:] != lengths[:-1]) + 1
+ for group in np.split(candidates, boundaries):
+ exemplar = int(group[0])
+ value = values.view(int(rows[exemplar]))
+ if len(group) == 1:
+ _validate_value_field_ids(
+ value, 0, len(value), source_metadata_size)
+ matching_structures[exemplar] = True
+ continue
+ matches = _matching_value_structures(
+ value, source_data, row_starts[group],
+ source_metadata_size)
+ if matches is not None:
+ matching_structures[group[matches]] = True
+
+ if state.data is not None:
+ source_view = memoryview(state.data)
+ source_base = state.data_start
+ else:
+ source_view = values.data
+ source_base = 0
+ prefix = bytes([header]) + (size + 1).to_bytes(size_width, 'little')
+ id_bytes = key_id.to_bytes(id_size, 'little')
+ id_slot = id_start + slot * id_size
+ offset_slot = offset_start + slot * offset_size
+ sentinel_slot = offset_start + size * offset_size
+ rebuilt = {}
+ fallback_rows = []
+ for index, row in enumerate(rows):
+ row = int(row)
+ if not ok[index]:
+ fallback_rows.append(row)
+ continue
+ original = values.view(row)
+ if source_metadata_size is not None:
Review Comment:
[P2] Recursively validate nested values when metadata is reused
When the inserted key already exists in shared metadata,
`source_metadata_size` is `None`, so this path validates only the root layout
and immediate child spans. A simultaneous fixed-width nested replacement is
patched in place and bypasses `_apply_edits`, allowing malformed grandchildren
to survive. In a one-row repro, metadata knows `new` but the root omits it,
while `nested.bad` contains a one-byte NULL plus orphan bytes; replacing
`nested.target` and inserting `new` succeeds here, whereas forcing the legacy
rebuild raises `MALFORMED_VARIANT: child size does not match container
offsets`. This breaks fast/rebuild error equivalence and can emit a
still-malformed value. Please preserve the source metadata count separately and
run full or batched structure validation for every splice candidate, including
the metadata-reuse case.
--
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]
