Copilot commented on code in PR #9259:
URL: https://github.com/apache/paimon/pull/9259#discussion_r3803905951
##########
paimon-python/pypaimon/data/variant_path.py:
##########
@@ -1880,6 +1940,168 @@ def _apply_edits(
return _materialize_value(results[0])
+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 the root layout."""
+ first_view = values.view(int(rows[0]))
+ header = first_view[0]
+ size, id_size, id_start, data_start, first_offsets, _ = (
+ _checked_object_layout(first_view, 0, len(first_view)))
+ ordered_offsets = sorted(first_offsets)
+ end_by_offset = dict(zip(ordered_offsets, ordered_offsets[1:]))
+ for index in range(size):
+ _checked_object_child_bounds(
+ first_view, 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(first_view, 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)
Review Comment:
The splice plan is derived exclusively from `rows[0]`. If `rows[0]` is a
valid-but-noncanonical layout (or an outlier), `_plan_root_insert_splice` can
return `None` and skip the fast path even when many other rows are eligible.
Consider selecting an exemplar from the first row that passes the cheap
predicates (e.g., scan forward until one passes header/size/id/offset checks)
so a single outlier doesn’t disable splicing for the whole batch.
##########
paimon-python/pypaimon/data/variant_path.py:
##########
@@ -368,18 +371,75 @@ def _validate_value_field_ids(value, pos, limit,
metadata_size):
end_by_offset = dict(zip(
ordered_offsets, ordered_offsets[1:]))
for slot in range(size):
- child_start, child_end = _checked_object_child_bounds(
- value, data_start, offsets, slot, end_by_offset)
- if (value[child_start] & 0x3) in (_OBJECT, _ARRAY):
- stack.append((child_start, child_end))
+ child_offset = offsets[slot]
+ stack.append((
+ data_start + child_offset,
+ data_start + end_by_offset[child_offset],
+ ))
elif basic_type == _ARRAY:
- size, data_start, offsets, _ = _checked_array_layout(
- value, current_pos, value_end)
+ size, data_start, offsets, value_end = _checked_array_layout(
+ value, current_pos, current_limit)
+ if structure_ranges is not None:
+ structure_ranges.append((current_pos, data_start))
for index in range(size):
stack.append((
data_start + offsets[index],
data_start + offsets[index + 1],
))
+ else:
+ value_end = current_pos + _checked_value_size(
+ value, current_pos, current_limit)
+ if structure_ranges is not None:
+ type_info = (value[current_pos] >> 2) & 0x3F
+ structure_end = current_pos + 1
+ if basic_type == _PRIMITIVE:
+ if type_info in (_BINARY, _LONG_STR):
+ structure_end += _U32_SIZE
+ elif type_info in (
+ _DECIMAL4, _DECIMAL8, _DECIMAL16):
+ structure_end = value_end
+ structure_ranges.append((current_pos, structure_end))
+ if value_end != current_limit:
+ _malformed("child size does not match container offsets")
+
+
+def _matching_value_structures(
+ value, source_data, row_starts, metadata_size):
+ """Match equal-length rows against one validated value structure."""
+ ranges = []
+ _validate_value_field_ids(
+ value, 0, len(value), metadata_size, ranges)
+ position_count = sum(end - start for start, end in ranges)
+ index_size = np.dtype(np.int64).itemsize
+ position_bytes = position_count * index_size
+ available_bytes = _STRUCTURE_MATCH_INDEX_BUDGET - position_bytes
+ if available_bytes < index_size:
+ return None
+ max_cells = available_bytes // index_size
+ positions = np.empty(position_count, dtype=np.int64)
+ cursor = 0
+ for start, end in ranges:
+ count = end - start
+ positions[cursor:cursor + count] = np.arange(
+ start, end, dtype=np.int64)
+ cursor += count
+
+ matches = np.ones(len(row_starts), dtype=bool)
+ expected = np.frombuffer(value, dtype=np.uint8)
+ width = max(1, min(position_count, max_cells))
+ row_chunk_size = min(len(matches), max(1, max_cells // width))
+ for row_start in range(0, len(matches), row_chunk_size):
+ row_end = min(row_start + row_chunk_size, len(matches))
+ batch_matches = matches[row_start:row_end]
+ batch_starts = row_starts[row_start:row_end]
+ for start in range(0, position_count, width):
+ offsets = positions[start:start + width]
+ batch_matches &= np.all(
+ source_data[batch_starts[:, None] + offsets]
+ == expected[offsets],
+ axis=1,
+ )
Review Comment:
Within `_matching_value_structures`, once `batch_matches` becomes all-False
for a chunk, continuing to compute additional `np.all(...)` comparisons for
further `offsets` slices is wasted work (and still allocates the advanced-index
result). Consider adding an early break when `not np.any(batch_matches)` inside
the inner loop to reduce unnecessary NumPy calls and allocations on mismatching
groups.
##########
paimon-python/pypaimon/data/variant_path.py:
##########
@@ -1880,6 +1940,168 @@ def _apply_edits(
return _materialize_value(results[0])
+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 the root layout."""
+ first_view = values.view(int(rows[0]))
+ header = first_view[0]
+ size, id_size, id_start, data_start, first_offsets, _ = (
+ _checked_object_layout(first_view, 0, len(first_view)))
+ ordered_offsets = sorted(first_offsets)
+ end_by_offset = dict(zip(ordered_offsets, ordered_offsets[1:]))
+ for index in range(size):
+ _checked_object_child_bounds(
+ first_view, 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(first_view, 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)
+
+ 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])
+ matches = _matching_value_structures(
+ values.view(int(rows[exemplar])), 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:
+ if (matching_structures is None
+ or not matching_structures[index]):
Review Comment:
`matching_structures` is always a boolean array when `source_metadata_size
is not None` (it’s initialized a few lines above), so the `matching_structures
is None` clause is dead code. Removing it simplifies the control flow and makes
the intended condition (`not matching_structures[index]`) clearer.
--
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]
