Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 26 additions & 1 deletion src/orcapod/core/operators/merge_join.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,15 @@ def binary_static_process(
# Find colliding data columns
colliding_keys = set(left_data_keys) & set(right_data_keys)

# Snapshot Arrow types of colliding columns BEFORE the Polars round-trip.
# The round-trip may strip or alter extension metadata; we need the original
# element type to reconstruct the correct list extension type after merging.
colliding_col_types: dict[str, "pa.DataType"] = {
col: left_table.schema.field(col).type
for col in colliding_keys
if col in left_table.schema.names
}

# Capture nullable flags from input schemas BEFORE Polars conversion.
# Polars' join discards nullable info (defaults all to True); we derive
# the output schema from the inputs instead of from data null counts.
Expand Down Expand Up @@ -232,6 +241,11 @@ def binary_static_process(
)
joined = joined.drop(COMMON_JOIN_KEY)

# Use the left stream's type converter — not the default context — so that a
# MergeJoin over streams built with a non-default DataContext reconstructs the
# merged column's extension type from the correct registry.
tc = left_stream.data_context.type_converter

# Process colliding data columns: merge into sorted lists
for col in colliding_keys:
left_col_name = col
Expand Down Expand Up @@ -273,7 +287,18 @@ def binary_static_process(
joined = joined.drop(left_col_name)
joined = joined.drop(right_col_name)

merged_array = pa.array(merged_vals)
elem_arrow_type = colliding_col_types.get(col)
if elem_arrow_type is not None and isinstance(elem_arrow_type, pa.ExtensionType):
elem_python_type = tc.arrow_type_to_python_type(elem_arrow_type)
list_logical_type = tc.get_logical_type_for_python_type(list[elem_python_type])
if list_logical_type is not None:
list_ext_type = list_logical_type.get_arrow_extension_type()
storage_array = pa.array(merged_vals, type=list_ext_type.storage_type)
merged_array = pa.ExtensionArray.from_storage(list_ext_type, storage_array)
else:
merged_array = pa.array(merged_vals)
else:
merged_array = pa.array(merged_vals)
joined = joined.add_column(col_idx, left_col_name, merged_array)

if has_source:
Expand Down
88 changes: 84 additions & 4 deletions src/orcapod/hashing/visitors.py
Original file line number Diff line number Diff line change
Expand Up @@ -198,21 +198,101 @@ def visit_extension(
extension_type: "pa.ExtensionType",
storage_value: Any,
) -> tuple["pa.DataType", Any]:
"""Hash an extension type value to pa.large_binary(), or passthrough."""
"""Hash an extension type value to ``pa.large_binary()``, or passthrough.

For list-backed extension types (e.g. ``extension<list[orcapod.file]>``),
delegates to ``_visit_list_elements`` with a virtual
``large_list(elem_ext_type)`` so that each element is hashed identically
to the scalar ``visit_extension`` path. This covers ``list[T]``,
``set[T]``, and arbitrary nesting depth via recursion.

Three passthrough cases (extension type and storage value returned unchanged):
- ``storage_value`` is ``None``.
- The Python type could not be resolved (``typing.Any`` or not a plain ``type``).
- The element type has no registered semantic handler and is not a nested list/set.

Args:
extension_type: The Arrow extension type to process.
storage_value: The storage-level value (result of ``to_pylist()`` on the column).

Returns:
Tuple of ``(new_arrow_type, new_data)``. For hashable scalar types returns
``(pa.large_binary(), hash_bytes)``. For list/set-backed types returns
``(pa.large_list(...), [hash_bytes, ...])``. Passthroughs return the
original ``(extension_type, storage_value)``.
"""
if storage_value is None:
return extension_type, None

# Resolve extension type → Python type.
python_type = self._type_converter.arrow_type_to_python_type(extension_type)

# Detect list-backed extension types: extension<list[orcapod.file]>,
# extension<set[orcapod.file]>, etc. list[File] is a types.GenericAlias
# (not isinstance(..., type)), so the guard below would incorrectly skip it.
# We intercept here and hash each element, folding the outer extension name
# into the result (mirrors the scalar path) to prevent list[T]/set[T] collisions.
if (
typing.get_origin(python_type) in (list, set)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The outer wrapper's extension identity is dropped on this branch. The scalar path below (L281-289) deliberately folds extension_type.extension_name into the hash token, but this branch returns a bare large_list(large_binary) with no record of which wrapper produced it.

Verified: single-row extension<list[orcapod.file]> and extension<set[orcapod.file]> tables holding the same file both hash to 0000019c292aa9a0 on this branch; on main they differ (000001ea... vs 000001df...). Two structurally distinct logical types are now indistinguishable to hash_table, so a memoized record keyed on one can be served for the other.

This is a silent-wrong-result bug rather than a crash, and the fix is small (include extension_type.extension_name in the result the way the scalar path does). Worth doing before merge, since fixing it later is another cache-invalidating hash change.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. The result now includes the outer extension name the same way the scalar path does.

The list-backed path now returns (pa.large_binary(), combined_bytes) where combined_bytes = type_name.encode() + b"::" + b"\x00".join(element_hashes) and type_name = extension_type.extension_name.replace(".", ":"). For extension<list[orcapod.file]> this produces b"list[orcapod:file]::<h0>\x00<h1>". For extension<set[orcapod.file]> with the same files it produces b"set[orcapod:file]::<h0>\x00<h1>" — distinct because the outer wrapper name differs.

Added test_list_and_set_file_extension_produce_distinct_hashes to directly assert the two hashes differ.

and pa.types.is_large_list(extension_type.storage_type)
):
args = typing.get_args(python_type)
# Defensive guard: a well-formed list[T]/set[T] always has args, but if
# not, fall through to the isinstance(python_type, type) passthrough below.
if args:
elem_python_type = args[0]

# Type-driven hashability: unwrap list/set nesting to the innermost
# non-container type and check whether it has a semantic handler.
# Decision is made once per column (not per row) so empty and null
# inner lists are handled correctly without crashing.
inner = elem_python_type
while typing.get_origin(inner) in (list, set):
inner_args = typing.get_args(inner)
if not inner_args:
break
inner = inner_args[0]
hashable = (
isinstance(inner, type)
and self._python_hasher.type_handler_registry.has_handler(inner)
)
if not hashable:
# Innermost element type has no semantic handler — whole-column
# passthrough, identical to main-branch behaviour.
return extension_type, storage_value

# Hashable: delegate element-level hashing to _visit_list_elements.
# Using the converter's element arrow type (which may itself be an
# extension<list[...]>) ensures each element recurses back into
# visit_extension, producing large_binary() per element.
# We discard the returned list type (it may hold an extension type
# when data is empty) and derive the output type from the outer name.
elem_arrow_type = self._type_converter.python_type_to_arrow_type(
elem_python_type
)
virtual_list_type = pa.large_list(elem_arrow_type)
_, list_data = self._visit_list_elements(virtual_list_type, storage_value)

# Fold the outer extension name into the result, the same way the
# scalar path does. This ensures list[T] and set[T] with identical
# contents produce distinct hashes.
type_name = extension_type.extension_name.replace(".", ":")
combined = (
type_name.encode("utf-8")
+ b"::"
+ b"\x00".join(
elem if isinstance(elem, bytes) else b""
for elem in (list_data or [])
)
)
return pa.large_binary(), combined

# If the converter couldn't resolve to a concrete class, passthrough.
if python_type is typing.Any or not isinstance(python_type, type):
return extension_type, storage_value

# Only hash if a semantic hasher is registered for this Python type.
if not self._python_hasher.type_handler_registry.has_handler(
python_type
):
if not self._python_hasher.type_handler_registry.has_handler(python_type):
return extension_type, storage_value

# Convert storage value → Python object and hash it.
Expand Down
1 change: 1 addition & 0 deletions src/orcapod/logical_types/list_logical_type_factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,7 @@ def get_polars_extension_type(self) -> pl.BaseExtension:
polars_ext_class = make_polars_extension_type(
self._logical_type_name,
self._storage_type,
metadata=self._metadata_bytes.decode("utf-8"),
)
self._polars_ext = polars_ext_class()
return self._polars_ext
Expand Down
Loading
Loading