Segmentation Fault when running mapping on chunked data #25

Preliminary checks

• I have checked that this issue has not already been reported.
• I have checked that this issue has not been fixed in the latest version.

Describe the bug

Segmentation Fault when running mapping on chunked data

To Reproduce

`"""Minimal reproduction: map_values segfaults on a chunked DataFrame.

TransformPlan 0.2.1 / polars 1.34.0 / Python 3.11.9

map_values crashes the interpreter with SIGSEGV when both of the following
hold for the frame it is applied to:

1. the target column consists of more than one chunk, and
2. the mapping has roughly 2000 entries or more.

Either condition alone is harmless. polars' own replace_strict handles the
same frame and mapping without trouble, so the fault is not in polars.

Chunked frames are easy to run into without noticing: any frame built by
concatenation - and, in our case, the frame produced by a longer TransformPlan -
keeps its chunks unless something calls rechunk().

The script runs each case in its own subprocess, because a segfault takes the
whole interpreter down.
"""
from future import annotations

import subprocess
import sys

import polars as pl
from transformplan import TransformPlan

N_KEYS = 2000
N_CHUNKS = 2

def make_frame(n_chunks: int) -> pl.DataFrame:
"""Return a frame whose single column has n_chunks chunks."""
return pl.concat([pl.DataFrame({"id": [i]}) for i in range(n_chunks)], rechunk=False)

def make_mapping(n_keys: int) -> dict[int, str]:
"""Return a mapping of n_keys entries."""
return {i: f"name {i}" for i in range(n_keys)}

def case(name: str) -> None:
"""Run one case in this process."""
mapping = make_mapping(N_KEYS if name != "small_mapping" else 100)
frame = make_frame(1 if name == "single_chunk" else N_CHUNKS)

if name == "rechunked":
    frame = frame.rechunk()

print(f"  chunks={frame['id'].n_chunks()}  rows={frame.height}  keys={len(mapping)}")

if name == "polars_replace_strict":
    frame.with_columns(pl.col("id").replace_strict(mapping, default=None))
else:
    TransformPlan().map_values("id", mapping).process(frame, validate=True)

print("  -> returned normally")

CASES = {
"chunked": "chunked frame, large mapping (expected to crash)",
"single_chunk": "single chunk, large mapping",
"small_mapping": "chunked frame, small mapping",
"rechunked": "chunked frame rechunked before the call",
"polars_replace_strict": "same frame and mapping, but polars directly",
}

def main() -> None:
"""Run every case in a subprocess and report how it ended."""
if len(sys.argv) > 1:
case(sys.argv[1])
return

print(f"polars {pl.__version__}, python {sys.version.split()[0]}\n")
for name, description in CASES.items():
    print(f"{name}: {description}")
    result = subprocess.run([sys.executable, __file__, name], capture_output=True, text=True)
    print(result.stdout.rstrip() or "  (no output)")
    verdict = "SEGFAULT" if result.returncode == -11 else f"exit {result.returncode}"
    print(f"  == {verdict}\n")

if name == "main":
main()
`

Expected behavior

No segmentation fault

Actual behavior

Segmentation fault

Additional context

No response