# crxml Documentation > Full documentation for crxml: high-performance Crystal Reports XML to Arrow/DataFrame engine for Python > Source: https://crxml.emiliano-go.com > Pages: 22 ======================================================================== PAGE: https://crxml.emiliano-go.com/api/ ======================================================================== ## CrystalXMLSource ```python CrystalXMLSource(source: str | Path, *, row_tag: str = "Row") ``` | Param | Type | Default | Description | |-----------|--------------------|-----------|-------------------------------| | `source` | `str \| Path` | required | Path to CR XML file | | `row_tag` | `str` | `"Row"` | XML tag for each record row | **Returns:** iterable of `dict[str, str]` **Raises:** `FileNotFoundError`, `ValueError` (bad CR XML format) ## RenameFields ```python RenameFields(mapping: dict[str, str]) ``` | Param | Type | Description | |----------|------------------|------------------------------------| | `mapping`| `dict[str, str]` | Old key → new key mapping | **Fusable:** yes | **Picklable:** yes ## CastTypes ```python CastTypes(types: dict[str, type], errors: str = "raise") ``` | Param | Type | Default | Description | |---------|------------------|-----------|-------------------------------------| | `types` | `dict[str, type]`|, | Field name → target type | | `errors`| `str` | `"raise"` | One of `"raise"`, `"coerce"`, `"skip"` | **Fusable:** yes | **Picklable:** yes ## DropFields ```python DropFields(fields: list[str]) ``` | Param | Type | Description | |----------|---------------|---------------------------| | `fields` | `list[str]` | Keys to remove from rows | **Fusable:** yes | **Picklable:** yes ## FilterRows ```python FilterRows(predicate: Callable[[dict], bool]) ``` | Param | Type | Description | |------------|----------------------------|----------------------------------| | `predicate`| `Callable[[dict], bool]` | Return `True` to keep the row | **Fusable:** yes | **Picklable:** no (unless a module-level function) ## Pipeline ```python Pipeline(source: Iterable[dict], *stages: Stage) ``` Created implicitly via `|`. Not typically constructed directly. ### Methods | Method | Signature | Description | |-------------|-----------------------------------------|----------------------------| | `__or__` | `(self, stage) -> Pipeline` | Append a stage | | `__iter__` | `(self) -> Iterator[dict]` | Iterate rows | | `parallel` | `(self, workers=None, batch_size=1000)`| Return parallel variant | ## CrystalXMLSource.schema ```python source.schema() -> list[str] ``` Returns the field name keys from the first row. The source caches the first batch internally, so calling `schema()` before building a pipeline does not lose data. **Raises:** `StopIteration` if the source is empty. ```python from crxml import CrystalXMLSource src = CrystalXMLSource("report.xml") fields = src.schema() print(fields) # ['{Report.InvoiceNo}', '{Report.Amount}', ...] ``` ## to_dataframe ```python to_dataframe(pipeline: Pipeline, chunksize: int | None = None) -> pd.DataFrame ``` | Param | Type | Default | Description | |------------|--------------------|---------|--------------------------------| | `pipeline` | `Pipeline` |, | Pipeline to consume | | `chunksize`| `int \| None` | `None` | Incremental chunk size | ## to_csv ```python to_csv(pipeline: Pipeline, path: str | Path, encoding: str = "utf-8", delimiter: str = ",", fieldnames: list[str] | None = None) -> None ``` | Param | Type | Default | Description | |--------------|--------------------|---------|------------------------------------------------| | `pipeline` | `Pipeline` |, | Pipeline to consume | | `path` | `str \| Path` |, | Output CSV path | | `encoding` | `str` | `"utf-8"` | Output file encoding | | `delimiter` | `str` | `","` | Field separator | | `fieldnames` | `list[str] \| None`| `None` | Explicit header; defaults to first record's keys | The header comes from the first record unless `fieldnames` is given. CR exports are ragged: fields that appear later but are missing from the header are omitted, and a `UserWarning` names them (once per field). Pass a `fieldnames` union to include them. Fields missing from a record are written as empty strings. ## Exceptions Typed exceptions from the Rust core, importable from `crxml`: | Exception | Raised when | |--------------|--------------------------------------------------------------------| | `XmlError` | Input cannot be parsed (malformed XML, failed UTF-8 validation) | | `PlanError` | Invalid pushdown plan kwargs (unknown op, unknown field type) | | `MergeError` | Chunk merge conflict during multi-chunk/parallel/bounded parsing | ```python from crxml import CrystalXMLSource, XmlError try: CrystalXMLSource("export.xml").to_pandas() except XmlError as e: print(f"bad input: {e}") ``` ## collect ```python collect(pipeline: Pipeline) -> list[dict] ``` | Param | Type | Description | |------------|------------|--------------------------| | `pipeline` | `Pipeline` | Pipeline to materialize | ======================================================================== PAGE: https://crxml.emiliano-go.com/architecture/ ======================================================================== # Architecture ## Overview crxml is a fast XML-to-DataFrame pipeline that uses a Rust core for parsing and a Python layer for pipeline composition. The key architectural insight is **fusion**: each stage of a pipeline can be compiled down into the Rust rypipe engine, executed as a vectorized batch operation on Arrow arrays, or fused into a tight dict loop, depending on what the stages support. ### Fusion levels The three fusion levels are tried in priority order: 1. **Columnar fusion** (Layer A): compile stages into the Rust `ExecutionPlan`, eliminating row iteration entirely. Stages with `_plan_kwargs()` produce a merged kwargs dict passed to `read_to_columnar*`. Remaining stages run through the batchpipe chain over Arrow `RecordBatch` objects. 2. **Vectorized batch fusion**: arrow-fusable stages (rename, drop, declarative filter) compile to `Callable[[Batch], Batch]` functions clustered into a single-pass `FusedTransforms` operator. Row-local `.apply` stages cluster into `LambdaOp`. Trailing stateful generators wrap the dict stream. 3. **Dict fusion** (Layer B): a contiguous run of `.apply` stages is fused into one tight `for r in src: for fn in bound: ...` loop. These stack: columnar pushdown is tried first; if it succeeds, the remaining stages run through the batchpipe; any trailing stateful stages wrap the dict stream. If columnar fusion isn't possible (source lacks `_read_arrow`, or no stage produces `_plan_kwargs`), the system falls back to dict fusion only. ``` XML bytes ──► rypipe engine (via crxml wrapper) ──► Arrow Table ──► batchpipe chain ──► sink The engine has three modes: stream: row-by-row via CrxmlReader (GIL-released batching) columnar: single-threaded columnar parse with ExecutionPlan pushdown parallel: chunked + rayon parallel columnar parse ``` ## Data flow ``` XML file │ ├─► stream engine: CrxmlReader.next_batch(n) │ │ │ └─► list[dict[str,str]] ──► Pipeline stages ──► sink │ ├─► columnar engine: rypipe_core::TableBuilder via crxml wrapper │ │ │ ├─► simdutf8 validation (one SIMD pass) in the embedded CrystalXmlDecoder │ ├─► borrowed-slice quick-xml reader (zero-copy events) │ │ │ ├─► ColumnBuilder columns ──► finish_row (null-fill, filter) │ │ └─► TableBuilder::extend (merge across chunks for multi/parallel) │ │ │ └─► RecordBatch export / engines_to_record_batches │ │ │ └─► Arrow C Data Interface ──► pyarrow.Table │ │ │ ├─► batchpipe chain (build_chain) │ │ ├─► ArrowSource ──► FusedTransforms ──► LambdaOp │ │ ├─► iter_dicts() or collect_table() │ │ └─► trailing stateful stages wrap stream │ │ │ ├─► Compare filter: arrow::compute kernels (pure Rust) │ │ │ └─► sinks: to_dataframe / to_csv / collect / to_polars / to_parquet │ └─► parallel engine: CrystalXmlSplitter + rypipe_core::ParallelExecutor │ ├─► fast path (no auto_dict): per-chunk TableBuilders exported independently │ └─► engines_to_record_batches() → per-chunk RecordBatch → concat │ └─► merge path (auto_dict): TableBuilders merged → auto_dict_upgrade → export └─► TableBuilder::extend() → RecordBatch ``` ## Rust core (`crxml_core`) The crate at `src/crxml_core/` uses `mimalloc::MiMalloc` as the global allocator (profiling showed ~27% of CPU time in malloc/free during XML parsing). It now contains two layers: 1. **Streaming engine** (`CrxmlReader` / `RowParser`): Crystal Reports XML specific and stays in `crxml_core`. 2. **Columnar FFI wrappers**: thin Python-callable wrappers that delegate to the generic `rypipe` engine, consumed as a versioned crate from crates.io (`rypipe-core = "0.1"`). The format-agnostic engine pieces (`ExecutionPlan`, `ColumnBuilder`, parallel/bounded drivers, and Arrow export) live in the `rypipe-core` crate in the sibling `rypipe` workspace. The Crystal Reports XML decoder and splitter now live inside `crxml_core::xml` as a custom `rypipe-core` adapter. ### `lib.rs`: FFI boundary, stream engine, and columnar wrappers #### Global allocator ```rust #[global_allocator] static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc; ``` #### `CrxmlReader` (`#[pyclass]`): streaming XML parser **`RowParser`** is a pure-Rust struct that holds **no Python objects**. This is a load-bearing invariant: it allows `next_batch(n)` to release the GIL via `py.allow_threads()` (the `Ungil` bound on the closure requires no `Py<...>` references inside). Python-object state (the interned-key cache) lives on `CrxmlReader` beside this struct. ```rust struct RowParser { reader: Reader>, // quick-xml streaming reader (128 KB buffer) buf: Vec, // scratch for quick-xml events inner_buf: Vec, // scratch for child-element events row: Vec<(String, String)>, // per-row field-value pairs (cleared each row) row_tag: Vec, // e.g. b"Row" batch_vals: Vec<(String, String)>, // flat buffer: all rows concatenated batch_lens: Vec, // field count per row for slicing } ``` **Parse flow** (`read_one_row`): 1. Quick-xml event loop looking for `` (or custom `row_tag`). 2. For ``, enters a child event loop looking for ``, ``, `
`. 3. ``: extracts `FieldName` attribute → key, reads `` or `` text → value. 4. ``: extracts `Name` attribute → key, reads `` text → value. 5. `
`: captures `SectionNumber` attribute. 6. Collects into `row: Vec<(String, String)>`. **`read_batch_into(n)`**: calls `read_one_row` up to `n` times, extending `batch_vals` and `batch_lens`. Runs with the GIL released. **Dict construction** (GIL held, `#[pymethods]`): - `new_dict(py)`: creates a plain `PyDict`. The private CPython API `_PyDict_NewPresized` was benchmarked and removed; it delivered only 3.5% overall gain and used an `unsafe` call to a private, unstable symbol. - `cached_key(key)`: `FxHashMap>`: field names repeat every row; interned `PyString` objects are reused instead of allocating fresh `PyUnicode` per field per row. **`next_batch(n)`**: releases GIL → parses `n` rows into flat buffers → re-acquires GIL → walks `batch_vals`+`batch_lens` → builds `PyList[PyDict]`. #### Columnar FFI functions Four `#[pyfunction]` entry points are thin wrappers over the embedded XML adapter (`crxml_core::xml`) and `rypipe-core`: | Function | Parsing | Output | |----------|---------|--------| | `read_to_columnar` | Single-threaded via `crxml_core::xml::CrystalXmlDecoder` + `rypipe_core::TableBuilder` | One `pyarrow.Table` | | `read_to_columnar_multi` | Chunked via `crxml_core::xml::CrystalXmlSplitter`, sequential parse + `TableBuilder::extend` | One `pyarrow.Table` | | `read_to_columnar_par` | Chunked + `rayon` via `rypipe_core::ParallelExecutor` | Per-chunk batch concat (fast path) or merged table (auto_dict) | | `read_to_columnar_bounded` | Memory-bounded batches via `rypipe_core::BoundedExecutor` | Concatenated `pyarrow.Table` | All accept the same `ExecutionPlan` kwargs as before: `field_mapping`, `drop_fields`, `filter`, `field_types`, `dictionary_columns`, `use_mmap`, `schema`, `auto_dict`. The wrappers: 1. Build an `rypipe_core::ExecutionPlan` from Python kwargs (same semantics as the old `BuildPlan`). 2. Open the input via `rypipe_core::InputBuffer` (mmap when requested). 3. Drive `CrystalXmlDecoder` / `CrystalXmlSplitter` / `ParallelExecutor` / `BoundedExecutor`. 4. Finish the `TableBuilder` to an Arrow `RecordBatch`, apply any column-to-column `Compare` filter with `rypipe_core::arrow_export::apply_compare_filter`, and export via the Arrow C Data Interface. Compare filters are now evaluated in pure Rust with `arrow::compute` kernels; the previous `pyarrow.compute` call from inside Rust has been removed. ### Where the columnar engine lives now The files `src/crxml_core/src/columnar.rs` and `src/crxml_core/src/splitter.rs` have been removed. Their contents were extracted into the sibling `rypipe` workspace: - `rypipe-core::plan::ExecutionPlan`: the format-agnostic plan (renamed from `BuildPlan`). - `rypipe-core::columnar`: `StrColumn`, `ColumnBuilder`, and dictionary encoding. - `rypipe-core::engine::TableBuilder`: the per-chunk state machine (renamed from `ColumnarEngine`). - `rypipe-core::merge`: engine merging and fast parallel export. - `rypipe-core::arrow_export`: Arrow `RecordBatch` building and `Compare` filter evaluation via `arrow::compute` kernels. - `rypipe-core::decoder`: the `Splitter`, `RecordParser`, and `ColumnarSink` traits. - `rypipe-core::parallel` / `rypipe-core::bounded`: parallel and memory-bounded drivers. - `rypipe-core::input`: `InputBuffer` with optional mmap support. - `crxml_core::xml::decoder::CrystalXmlDecoder`: Crystal Reports XML row parser (embedded adapter). - `crxml_core::xml::splitter::CrystalXmlSplitter`: XML row-boundary splitting (embedded adapter). crxml still owns the Crystal-specific grammar as an embedded `rypipe-core` format adapter rather than inline engine code. Future formats can be added as additional `rypipe-core` adapters without touching crxml. ## Python source layer ### `__init__.py`: Lazy public API ```python __all__ = ["CrystalXMLSource", "Pipeline", "RenameFields", "CastTypes", "FilterRows", "DropFields", "to_dataframe", "to_csv", "collect"] _modules = {"CrystalXMLSource": ".source", "Pipeline": ".pipeline", ...} def __getattr__(name): if name in _modules: mod = importlib.import_module(_modules[name], __package__) return getattr(mod, name) raise AttributeError(...) ``` All public symbols are lazily imported via `__getattr__`. `import crxml` is instant; modules load only when their symbols are first accessed. ### `CrystalXMLSource` (in `source.py`) Wraps the Rust engines. Constructor parameters map one-to-one to `ExecutionPlan` fields plus engine selection: - `engine`: `"auto"` (default), `"stream"`, `"columnar"`, `"parallel"` - `threads`: multiplied by **4** to get `num_chunks`. The 4x multiplier exists because finer chunks give better load balancing; VTune showed 3-4x optimal on 24 cores (beyond 4x, rayon join/spin overhead dominates). - `memory`: optional string (`"8GB"`) or int bytes; enables bounded mode. - `use_mmap`: memory-map the file (Unix only). - `batch_size`: rows per Rust→Python batch call (default 1024). - `field_mapping`, `drop_fields`, `filter`, `field_types`, `dictionary_columns`, `schema`, `auto_dict`: map directly to `ExecutionPlan`. **Goal-aware engine dispatch** (`_resolve_engine(goal)`): | Goal | File size | Memory OK | Engine selected | |------|-----------|-----------|-----------------| | `"iter"` | any | any | `"stream"` (always) | | `"table"` | ≥ 8 MB | yes | `"parallel"` | | `"table"` | ≥ 8 MB | no | `"columnar"` (if available) → `"stream"` fallback | | `"table"` | < 8 MB | yes | `"columnar"` (if available) → `"stream"` fallback | The 8 MB threshold exists because parallel overhead (chunking + rayon + merge) doesn't pay off for small files. **`_build_plan_kwargs()`**: collects the source's config into a dict (`field_mapping`, `drop_fields`, `filter`, `field_types`, `dictionary_columns`, `schema`, `auto_dict`, `use_mmap`). This is the base dict that stage `plan_overrides` are merged into. **`_read_arrow(plan_overrides)`**: the core table-building method. 1. Resolves engine for `"table"` goal. 2. Merges `plan_overrides` into `_build_plan_kwargs()`. 3. Dispatches to Rust function: bounded → `read_to_columnar_bounded`; columnar → `read_to_columnar`; parallel → `read_to_columnar_par`; stream fallback → builds `pyarrow.Table` from Python dicts. 4. Caches the result (unless `plan_overrides` was provided, indicating a one-off fusion call). **Iteration modes**: - `__iter__`: stream → `_batch_iter(self._stream_iter())` calls `CrxmlReader.next_batch` in a loop, yielding dicts. Columnar/parallel → `_arrow_iter(self._read_arrow())` walks `Table.to_batches()`, yielding via `batch.to_pylist()`. - `_iter_batches`: stream → calls `reader.next_batch` directly (yields lists of dicts). Columnar/parallel → calls `to_arrow().to_batches()` then `.to_pylist()` per batch. **`_batch_iter(reader, batch_size)`**: wraps `CrxmlReader.next_batch` in a generator. One Rust call per batch with GIL released; `yield from batch` walks each batch list at C speed (no per-row Python `__next__`). **`_arrow_iter(table)`**: walks `table.to_batches()` and yields dicts via `batch.to_pylist()`. **`schema()`**: reads the first row via `next(iter(self), None)`, returns `list(first_row.keys())` or `[]` for empty files. The first batch is cached internally (via `_cached_arrow` for columnar, or the stream reader's internal state for stream) so schema inspection doesn't consume data. ### `Pipeline` (in `pipeline.py`) `Pipeline` is an immutable value object. ```python class Pipeline: __slots__ = ("_source", "_stages", "_batch_size", "_prefetch", "_workers") ``` **`__or__(stage)`**: creates a new `Pipeline` with the stage appended. The original is unchanged: ```python source | rename | cast → Pipeline(source, [rename, cast]) ``` **`__iter__()`**: decides execution strategy: 1. If `self._workers` is set → `parallel.parallel_iter()` (ProcessPoolExecutor). Stages are validated as picklable first. 2. Otherwise → `fusion.fused_iter(source, stages)`. **`_to_arrow()` shortcut**: - Returns a single `pyarrow.Table` if the whole pipeline can be executed as columnar fusion + batchpipe chain without trailing stages. - Returns `None` if: workers are set, source lacks `_read_arrow`, or trailing stateful stages remain. - This is the key fast path used by `to_dataframe()` and `collect()`; they check `_to_arrow()` first and skip the dict stream entirely. **`parallel(workers=None, batch_size=1000)`**: returns a new `Pipeline` with worker count set. Also carries forward the `_prefetch` flag (though prefetch is not toggled by the current public API). ### `fusion.py`: Fusion orchestrator **`plan_split(stages)`**: iterates stages calling `_plan_kwargs()` on each. Stages returning a dict contribute to `plan_overrides` (consumed); stages returning `None` or lacking the method go to `remaining`. ``` Input: [RenameFields, CastTypes, FilterRows(callable), DropFields] Output: plan_overrides = {field_mapping: ..., field_types: ..., drop_fields: ...} remaining = [FilterRows(callable)] ``` **`_try_columnar_fusion(source, stages)`**: 1. Guards: source must have `_read_arrow` and `_build_plan_kwargs`. 2. Calls `plan_split` → gets `plan_overrides` and `remaining`. 3. If no plan_overrides and all stages are remaining → returns `None` (no columnar benefit). 4. Calls `source._read_arrow(plan_overrides=plan_overrides or None)` → produces `pyarrow.Table` with Rust pushdown. 5. Calls `batchpipe.build_chain(table, remaining, batch_size)` → returns Volcano operator chain + trailing stages. 6. Returns `iter_dicts(op)` stream; trailing stages wrap the stream. **`fused_iter(source, stages)`**: the main execution entry point: 1. Tries `_try_columnar_fusion`: if it returns a non-None stream, done. 2. Falls back to dict fusion: - Scans from front for a contiguous run of fusable stages (has `callable(stage.apply)`). - Fused inner loop: `for r in src: for fn in bound: r = fn(r); if None: break; else: yield r`. - Non-fusable remaining stages wrap the fused generator. - If no fusable stages found: source bypasses the fused generator (avoids one generator frame per row). Stages wrap the source directly. **`is_fusable(stage)`**: `callable(stage.apply)`. ### `batchpipe.py`: Vectorized batch pipeline A pull-based (Volcano-style) operator chain over Arrow `RecordBatch` objects. **`Batch`**: the unit of flow. `namedtuple("Batch", "data, selection")` where `data` is a `RecordBatch` and `selection` is an optional `BooleanArray` mask. ```python class Batch: def compact(self): """Apply the selection and return a dense RecordBatch.""" if self.selection is None: return self.data return self.data.filter(self.selection) ``` **Operator hierarchy**: ``` Operator (abstract) open(), next_batch() -> Batch | None, close() ├── ArrowSource(table, batch_size) │ └── wraps pyarrow.Table, yields Batch objects ├── FusedTransforms(upstream, fns) │ └── applies list of batch-level functions (rename/drop/filter) └── LambdaOp(upstream, applies) └── row-level .apply fallback: compact → dict → apply → rebuild RecordBatch ``` **Selection masks**: filters produce boolean masks via `pyarrow.compute` (e.g., `pc.equal(rb.column("city"), "NYC")`). Masks are AND-ed into `Batch.selection`. Compaction (`Batch.compact()` → `RecordBatch.filter(selection)`) happens only at sinks or at `LambdaOp` boundaries, avoiding materialization of filtered-out rows until necessary. **Arrow-fusable stages** compiled by `_arrow_fusable(stage)`: | Stage | Compiles to | Implementation | |-------|------------|----------------| | `RenameFields` | `_fuse_rename(mapping)` | `RecordBatch.from_arrays(rb.columns, names=[mapping.get(n,n) for n in names])` | | `DropFields` | `_fuse_drop(fields)` | Keep columns by index, rebuild batch | | `FilterRows` (declarative) | `_fuse_filter_spec(spec)` | `pc.equal(column, value)` → AND into selection. Compare: `pc.greater(cola, colb)` etc. Null fill matches dict semantics | **Not arrow-fusable**: `CastTypes` (type coercion in Arrow is not a simple rename/drop/filter), `FilterRows` with callable predicate, lambda stages, generators. **`build_chain(table, stages, batch_size)`**: 1. Starts with `ArrowSource(table, batch_size)`. 2. Greedily clusters arrow-fusable stages into a single `FusedTransforms`. 3. Clusters consecutive row-level `.apply` stages into a single `LambdaOp` (only at boundaries where `_arrow_fusable` returns None and stage has `.apply`). 4. Stops at generic stream stages (no `.apply`, no arrow fusion). 5. Returns `(operator, trailing_stages)`. **Sinks**: - `iter_dicts(op)`: compact each batch, yield via `RecordBatch.to_pylist()`. - `collect_table(op)`: collect all compacted batches, return `pa.Table.from_batches(batches)`. ### `stages/`: The four built-in stages All four implement the same protocol: ```python class Stage: def apply(self, record: dict) -> dict | None: ... def __call__(self, stream): return map(self.apply, stream) def _plan_kwargs(self) -> dict | None: ... ``` | Stage | `apply()` behavior | `_plan_kwargs()` output | |-------|-------------------|------------------------| | `RenameFields(mapping)` | `{mapping.get(k,k): v for k,v in record.items()}` | `{"field_mapping": mapping}` | | `CastTypes(mapping)` | `record[field] = cast_fn(record[field])` in-place | `{"field_types": {name: type_str}}`. Maps `int→"int64"`, `float→"float64"`, `bool→"bool"`, `str→None` (skip). Returns `None` if any cast fn is not one of these | | `DropFields(fields)` | `{k:v for k,v in record.items() if k not in fields_set}` | `{"drop_fields": sorted(fields_set)}` | | `FilterRows(...)` | `record if predicate(record) else None` | `{"filter": spec}` for declarative; `None` for callable | **`FilterRows`** has three construction paths: 1. **Callable predicate**: `FilterRows(predicate=lambda r: ...)`: not columnar-pushdownable. 2. **Declarative constant**: `FilterRows(field="city", op="==", value="NYC")`: pushdownable as `FilterPredicate::Equal`/`NotEqual`. Uses `_ConstantPredicate` inner class. 3. **Declarative compare**: `FilterRows(field_a="age", op=">", field_b="threshold")`: pushdownable as `FilterPredicate::Compare` (post-reduce via pyarrow.compute). Uses `_ComparePredicate` inner class. **Filter semantics**: - Constant `==`: missing field returns unequal (dict `.get()` returns `None`). - Constant `!=`: missing field returns equal (`None != value` is true; the row is kept). - Compare: both columns must exist. Evaluated post-reduce via `pyarrow.compute`. ### `parallel.py`: Multi-process parallelism ``` Source ──► _prefetch_iter (bounded queue, maxsize=8) ──► ProcessPoolExecutor ──► ordered results ``` **`_prefetch_iter(source, batch_size, maxsize=8)`**: background `threading.Thread` reads the source, fills a `queue.Queue` with dict batches (size `batch_size`). Bounded at 8 batches to prevent unbounded memory. **`validate_stages_picklable(stages)`**: pickles each stage and raises `TypeError` at `.parallel()` call time (not in the worker). Catches lambdas and closures early. **`_worker_apply(batch, stages)`**: module-level function (required for pickling). Re-imports `fused_iter` from `.fusion` inside the worker process, runs `list(fused_iter(batch, stages))` and returns the result list. **`parallel_iter(source, stages, workers, batch_size)`**: 1. Wraps source in `_prefetch_iter`. 2. Creates `ProcessPoolExecutor(max_workers=workers)`. 3. **Double-buffered submission**: submits `workers * 2` futures initially. For each completed future, submits one new future. This keeps the executor saturated while bounding in-flight memory. 4. Yields results in submission order: `for idx in range(len(futures)): yield from futures[idx].result()`. ### `sinks.py`: Terminal operations **Shortcut hierarchy**: | Sink | Fast path | Fallback | |------|-----------|----------| | `to_dataframe` | `pipeline._to_arrow()` → single `pyarrow.Table` → `table.to_pandas()` (ArrowDtype, zero dicts) | `_iter_batches()` → chunked DataFrame → `pd.concat`; or `pd.DataFrame.from_records(iter(pipeline))` | | `collect` | `pipeline._to_arrow()` → `table.to_pylist()` | `_iter_batches()` → dicts; or `list(pipeline)` | | `to_csv` | Always streams row-by-row via `csv.DictWriter` (no intermediate list) | - | `to_dataframe(chunksize=N)` always uses batch-then-concat for memory control. `chunksize=None` triggers the single-table fast path. ## Fusion decision tree When `Pipeline.__iter__()` is called (not in worker mode): ``` fused_iter(source, stages) │ ├─ Has _read_arrow + _build_plan_kwargs? │ ├─ NO ──► skip to dict fusion │ └─ YES ──► plan_split(stages) │ │ │ ├─ plan_overrides empty AND len(remaining) == len(stages)? │ │ └─ YES ──► skip to dict fusion (no columnar benefit) │ │ │ └─ NO ──► Layer A: source._read_arrow(plan_overrides) │ │ │ └─ build_chain(table, remaining, batch_size) │ │ │ ├─ Returns (op, trailing) │ │ op = ArrowSource → FusedTransforms → LambdaOp │ │ trailing = [stateful stream stages] │ │ │ └─ stream = iter_dicts(op) │ for stage in trailing: stream = stage(stream) │ return stream │ └─ Dict fusion (Layer B): │ ├─ Scan front: contiguous .apply stages → fusables ├─ bound = [s.apply for s in fusables] │ ├─ If no bound: │ stream = source (or _iter_batches flat) │ for stage in remaining: stream = stage(stream) │ └─ If bound: def fused(): for r in source: for fn in bound: r = fn(r); if None: break else: yield r stream = fused() for stage in remaining: stream = stage(stream) return stream ``` When a sink is called: ``` to_dataframe(pipeline): ├─ has _to_arrow()? │ └─ YES → pipeline._to_arrow() │ ├─ returns Table? → table.to_pandas() [FASTEST] │ └─ returns None? → fallback ├─ has _iter_batches()? │ └─ YES → [pd.DataFrame.from_records(batch) for batch in pipeline._iter_batches()] └─ pd.DataFrame.from_records(iter(pipeline)) ``` ## Memory model - **Stream engine**: `RowParser` reuses `Vec` buffers across rows. Dicts are built in Python heap via `PyDict::new()`. The `FxHashMap` key cache lives for the reader's lifetime. - **Columnar engine**: `StrColumn` uses a flat byte arena + `i32` offsets; no per-cell `String` allocation. Numeric columns use `Vec>` (8 bytes + 1 validity per cell). Arrow arrays are built natively and exported via C Data Interface. - **mmap**: files are memory-mapped; advice is `MADV_WILLNEED` when `prefault` is set (parse-speed goal) and `MADV_SEQUENTIAL` otherwise (RSS-sensitive paths). The mapping is dropped synchronously after export; all data lives in owned Arrow buffers by then. - **Bounded mode**: `read_to_columnar_bounded` samples 64 KB → estimates `bytes_per_row` → splits into memory-sized chunks → parses/exports each chunk independently → chunk engine dropped → tables concatenated. - **Parallel mode RSS**: - Without auto_dict: each chunk's `TableBuilder` is exported and dropped before next is processed → peak RSS ≈ file size + overhead. - With auto_dict: all chunk engines held in memory before merge + dict upgrade → peak RSS can reach ~5x file size. ## Concurrency model | Component | Concurrency mechanism | GIL behavior | |-----------|----------------------|--------------| | Stream parser (`CrxmlReader`) | Single-threaded | Released during `read_batch_into` | | Columnar single (`read_to_columnar`) | Single-threaded | Released during parse, held for export | | Columnar multi (`read_to_columnar_multi`) | Sequential chunks | Released per chunk parse | | Columnar parallel (`read_to_columnar_par`) | `rayon::par_iter()` | Released for entire parallel parse (only GIL at start and end) | | Prefetch reader thread | `threading.Thread` | Held by reader for dict construction | | Parallel pipeline (`ProcessPoolExecutor`) | Separate processes | No GIL contention (separate interpreters) | | Batchpipe chain (FusedTransforms/LambdaOp) | Single-threaded (consumer) | Held (Arrow operations with GIL) | The expensive parts (XML parsing, string scanning) run with the GIL released in all paths. The columnar engine goes further: it never creates Python objects during parsing, so GIL release is more effective (no periodic Python GC interference). ## Key optimization summary | Optimization | Location | Impact | |-------------|----------|--------| | `mimalloc` global allocator | `lib.rs:22` | ~27% CPU reduction in malloc/free | | `PyDict::new` (no presize) | `src/crxml_core/src/lib.rs` | Removed private-CAPI hack; 3.5% gain not worth `unsafe` | | Key interning (`FxHashMap`) | `src/crxml_core/src/lib.rs` | Reuses `PyString` objects across rows | | SIMD UTF-8 validation | `rypipe_xml::decoder` | One SIMD pass per chunk (via `simdutf8`) | | Fast scanner (memchr-based) | `rypipe_xml::decoder` | Avoids quick-xml event loop overhead for standard CR XML | | `StrColumn` arena allocation | `rypipe_core::columnar` | No per-cell `String` allocation | | Deferred filter compaction | `batchpipe.py:31-48` | Only materializes alive rows at sinks/LambdaOp | | Columnar fusion (Layer A) | `fusion.py:23-44` | Entire pipeline compiled into Rust `ExecutionPlan` | | `_to_arrow()` shortcut | `pipeline.py:45-67` | Skips dict construction entirely for fast-path pipelines | | Arrow C Data Interface | `rypipe_core::arrow_export` | Zero-copy export from Rust Arrow to pyarrow | | Synchronous unmap after export | `rypipe_core::input` (`MmapInput`) | Releases file-backed pages before pandas conversion begins | | 4x chunk multiplier | `source.py:109` | Finer grains for rayon load balancing (VTune-optimized) | | Bounded memory batches | `rypipe_core::bounded` | Streams large files within configurable memory budget | | Fast parallel export (no merge) | `rypipe_core::merge::engines_to_record_batches` | Avoids per-chunk merge for non-auto-dict parallel parse | ## Key data types | Context | Type | Role | |---------|------|------| | Python stream | `dict[str, str]` | Single row (raw string values) | | Python columnar | `pyarrow.Table` | Full parsed dataset in Arrow format | | Python batchpipe | `Batch` (class with `__slots__`) | Unit of flow: `data` (`RecordBatch`) + optional boolean `selection` mask | | Python pipeline | `Pipeline` | Immutable composition of `source + stages` | | Python stage | `Callable[[Iterable[dict]], Iterable[dict]]` | Row transformation function | | Rust stream | `CrxmlReader` (PyClass) | Streaming XML parser | | Rust columnar | `rypipe_core::TableBuilder` | HashMap of `ColumnBuilder`s + plan + row count | | Rust columnar | `rypipe_core::ColumnBuilder` | String / Int64 / Float64 / Boolean / Dictionary variants | | Rust columnar | `rypipe_core::StrColumn` | Flat byte arena + `i32` offsets (Arrow layout) | | Rust columnar | `rypipe_core::ExecutionPlan` | Compilation target for stage pushdown | | Rust columnar | `rypipe_core::FilterPredicate` | Equal / NotEqual / Compare variants | | Rust splitter | `rypipe_xml::CrystalXmlSplitter` | Finds whole-row split points for parallel parsing | | Rust decoder | `rypipe_xml::CrystalXmlDecoder` | Emits field events from Crystal Reports XML | ======================================================================== PAGE: https://crxml.emiliano-go.com/changelog/ ======================================================================== # Changelog ## 1.2.0 (2026-08-23) ### Refactor - Extracted the columnar engine into the sibling `rypipe` workspace (`rypipe-core`, `rypipe-xml`, `rypipe-python`). - `rypipe-core` is now consumed from crates.io as a versioned dependency (`version = "0.1"`, `mmap` feature) instead of a path dependency: building crxml no longer requires a sibling rypipe checkout. - Embedded the Crystal Reports XML adapter (previously the separate `rypipe-xml` crate) directly in `crxml_core`. - Renamed the internal plan type from `BuildPlan` to `rypipe_core::ExecutionPlan`. - `Compare` filters now use `arrow::compute` kernels instead of `pyarrow.compute`. ### Removed - Deleted `src/crxml_core/src/columnar.rs` and `src/crxml_core/src/splitter.rs`; their logic lives in rypipe now. ### Kept - The streaming `CrxmlReader` remains in `crxml_core`. ### Packaging - sdist now ships `LICENSE` explicitly (PEP 639 license expression) so PyPI accepts the upload. - CI installs `rypipe` from PyPI for integration tests instead of cloning a sibling checkout. ### Testing - All existing tests pass. ## 1.0.0 (2026-07-06) ### Bug Fixes - **auto_dict plan lost in parallel merge**: `ColumnarEngine::new()` defaulted to `auto_dict: false`, making `auto_dict_upgrade()` a no-op. Fixed by using `ColumnarEngine::with_plan(est, plan)` to carry the build plan forward. - **Text field parsing in bounded path**: The parser was capturing whitespace-only text nodes as field values instead of looking for `` children. Fixed to correctly consume `TextValue` inner text. - **Stream engine column discovery**: Engine used first-row columns as schema; sparse columns appearing only in later rows caused crashes. Schema is now discovered across all rows. - **Publishing workflow missing `columnar` feature**: `maturin build --features mmap` risked overriding pyproject.toml's feature list and silently dropping `columnar` from the published wheel. CI now builds from pyproject.toml defaults. ### Features - **`prefault` parameter**: All engines accept `prefault: bool`. `True` = `MADV_WILLNEED` (speed), `False` = `MADV_SEQUENTIAL` (lower RSS). Defaults: True for columnar/parallel, False for bounded. - **Parallel engine profiling**: `get_par_profile()` returns nanosecond timing for split-scan, off-GIL parse, and on-GIL assembly phases (gated behind `profile` Cargo feature). - **Bounded mode RSS rewrite**: Mmap used only for initial split scan, then dropped. Chunks read via `File::seek`/`read_exact`. Peak RSS tracks the `memory=` budget, not the file size. - **`sort_columns()` on engine**: Ensures all batch engines produce identical column order for schema-match fast path in `concat_tables()`. ### Performance - **Splitter SIMD optimization**: `next_row_start()` searches for `= options["batch"]: Invoice.objects.bulk_create(batch) self.stdout.write(f"Imported {len(batch)} invoices") batch.clear() if batch: Invoice.objects.bulk_create(batch) self.stdout.write(f"Imported {len(batch)} invoices") ``` Run it: ```bash python manage.py import_report report.xml --batch 2000 ``` ## Upload + preview A simple admin-like view that accepts a file upload, parses it, and renders a preview table: ```python # yourapp/views.py from django.shortcuts import render from django.http import HttpResponseRedirect from django.urls import reverse from crxml import CrystalXMLSource from .forms import ReportUploadForm def preview_report(request): if request.method == "POST": form = ReportUploadForm(request.POST, request.FILES) if form.is_valid(): src = CrystalXMLSource(request.FILES["file"], row_tag="Details") rows = [row for row in src] return render(request, "preview.html", { "fields": list(rows[0].keys()) if rows else [], "rows": rows[:100], "total": len(rows), }) else: form = ReportUploadForm() return render(request, "upload.html", {"form": form}) ``` ## Periodic import with Celery ```python # yourapp/tasks.py from celery import shared_task from crxml import CrystalXMLSource, collect from .models import SalesRecord @shared_task def import_sales_report(path: str): src = CrystalXMLSource(path, row_tag="Details") records = [] for row in src: records.append(SalesRecord( product=row.get("{Report.Product}", ""), quantity=int(row.get("{Report.Qty}", 0)), price=float(row.get("{Report.Price}", 0)), )) SalesRecord.objects.bulk_create(records, ignore_conflicts=True) return len(records) ``` ## Admin action for file upload Add a Django admin action that accepts a file, parses it, and imports into a model: ```python # yourapp/admin.py from django.contrib import admin, messages from django import forms from django.shortcuts import render from crxml import CrystalXMLSource, collect from .models import Invoice class UploadXMLForm(forms.Form): file = forms.FileField() @admin.action(description="Import from Crystal Reports XML") def import_from_xml(modeladmin, request, queryset): if "file" not in request.FILES: if request.method == "POST": form = UploadXMLForm(request.POST, request.FILES) if form.is_valid(): uploaded = request.FILES["file"] rows = collect(CrystalXMLSource(uploaded.read(), row_tag="Details")) for row in rows: Invoice.objects.create( number=row.get("{Report.InvoiceNo}", ""), customer=row.get("{Report.Customer}", ""), amount=row.get("{Report.Amount", 0), ) modeladmin.message_user(request, f"Imported {len(rows)} invoices") return else: form = UploadXMLForm() return render(request, "admin/upload_xml.html", {"form": form}) return @admin.register(Invoice) class InvoiceAdmin(admin.ModelAdmin): actions = [import_from_xml] ``` ## Celery task with progress tracking Track import progress using the Celery task state: ```python # yourapp/tasks.py from celery import shared_task, current_task from crxml import CrystalXMLSource, RenameFields, CastTypes from django.db import transaction from .models import SalesRecord @shared_task(bind=True) def import_report(self, path: str): pipe = ( CrystalXMLSource(path, row_tag="Details") | RenameFields({ "{Report.Product}": "product", "{Report.Qty}": "quantity", "{Report.Price}": "price", }) | CastTypes({"quantity": int, "price": float}) ) batch = [] total = 0 for row in pipe: batch.append(SalesRecord( product=row["product"], quantity=row["quantity"], price=row["price"], )) if len(batch) >= 1000: with transaction.atomic(): SalesRecord.objects.bulk_create(batch, ignore_conflicts=True) total += len(batch) current_task.update_state( state="PROGRESS", meta={"current": total} ) batch.clear() if batch: with transaction.atomic(): SalesRecord.objects.bulk_create(batch, ignore_conflicts=True) total += len(batch) return {"imported": total} ``` Query the result from the view: ```python from celery.result import AsyncResult def task_status(request, task_id): result = AsyncResult(task_id) return JsonResponse({ "state": result.state, "progress": result.info.get("current", 0) if result.info else 0, }) ``` ## Thread safety Django's ORM is thread-safe. Each request or task gets its own `CrystalXMLSource` instance, so there is no shared state across requests. ======================================================================== PAGE: https://crxml.emiliano-go.com/fastapi/ ======================================================================== # FastAPI Integration A production pattern for accepting CR XML file uploads, parsing server-side, and returning structured data. ## Upload endpoint ```python from fastapi import FastAPI, UploadFile, File, HTTPException from tempfile import NamedTemporaryFile from crxml import CrystalXMLSource, collect import os app = FastAPI() MAX_FILE_SIZE = 500 * 1024 * 1024 # 500 MB @app.post("/parse-report") async def parse_report(file: UploadFile = File(...)): if file.size and file.size > MAX_FILE_SIZE: raise HTTPException(413, "File too large") ext = os.path.splitext(file.filename or "")[1].lower() if ext not in (".xml", ".rpt"): raise HTTPException(422, "Unsupported file type") with NamedTemporaryFile(delete=False, suffix=".xml") as tmp: content = await file.read() if len(content) > MAX_FILE_SIZE: os.unlink(tmp.name) raise HTTPException(413, "File too large") tmp.write(content) tmp_path = tmp.name try: rows = collect(CrystalXMLSource(tmp_path)) return {"rows": len(rows), "data": rows} except Exception as e: raise HTTPException(500, f"Parse failed: {e}") finally: os.unlink(tmp_path) ``` ## Large file background processing Offload parsing of large files to a background task and return a result ID: ```python from fastapi import BackgroundTasks from uuid import uuid4 from crxml import CrystalXMLSource, RenameFields, CastTypes, to_csv results: dict[str, str] = {} def process_large_file(tmp_path: str, result_id: str, output_path: str): pipe = ( CrystalXMLSource(tmp_path) | RenameFields({ "{Report.InvoiceNo}": "invoice", "{Report.Amount}": "amount", }) | CastTypes({"amount": float}) ) to_csv(pipe, output_path) results[result_id] = output_path @app.post("/parse-large") async def parse_large(file: UploadFile = File(...), background: BackgroundTasks = BackgroundTasks()): with NamedTemporaryFile(delete=False, suffix=".xml") as tmp: tmp.write(await file.read()) tmp_path = tmp.name result_id = str(uuid4()) output_path = f"/tmp/{result_id}.csv" background.add_task(process_large_file, tmp_path, result_id, output_path) return {"result_id": result_id, "status": "processing"} @app.get("/results/{result_id}") async def get_result(result_id: str): path = results.get(result_id) if path is None: raise HTTPException(404, "Result not found or still processing") return FileResponse(path, media_type="text/csv") ``` ## Dependency injection for pipelines Reusable pipeline factory via FastAPI dependencies: ```python from fastapi import Depends from crxml import CrystalXMLSource, RenameFields, CastTypes, collect DEFAULT_MAPPING = { "{Report.InvoiceNo}": "invoice", "{Report.Customer}": "customer", "{Report.Amount}": "amount", } def get_pipeline(tmp_path: str, mapping: dict[str, str] | None = None): src = CrystalXMLSource(tmp_path) if mapping: src = src | RenameFields(mapping) | CastTypes({"amount": float}) return src @app.post("/parse-with-mapping") async def parse_with_mapping(file: UploadFile = File(...)): with NamedTemporaryFile(delete=False, suffix=".xml") as tmp: tmp.write(await file.read()) tmp_path = tmp.name try: pipe = get_pipeline(tmp_path, DEFAULT_MAPPING) rows = collect(pipe) return {"rows": len(rows)} finally: os.unlink(tmp_path) ``` ## Streaming CSV response Stream CSV directly without buffering all rows: ```python from fastapi.responses import StreamingResponse import csv import io @app.post("/stream-csv") async def stream_csv(file: UploadFile = File(...)): with NamedTemporaryFile(delete=False, suffix=".xml") as tmp: tmp.write(await file.read()) tmp_path = tmp.name async def row_generator(): src = CrystalXMLSource(tmp_path) writer = None for row in src: if writer is None: output = io.StringIO() w = csv.DictWriter(output, fieldnames=list(row.keys())) w.writeheader() yield output.getvalue() output = io.StringIO() w = csv.DictWriter(output, fieldnames=list(row.keys())) w.writerow(row) yield output.getvalue() return StreamingResponse( row_generator(), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=report.csv"} ) ``` ## Streaming XLSX response ```python from fastapi.responses import StreamingResponse from openpyxl import Workbook from io import BytesIO @app.post("/to-xlsx") async def to_xlsx(file: UploadFile = File(...)): with NamedTemporaryFile(delete=False, suffix=".xml") as tmp: tmp.write(await file.read()) tmp_path = tmp.name try: rows = collect(CrystalXMLSource(tmp_path)) finally: os.unlink(tmp_path) wb = Workbook() ws = wb.active if rows: ws.append(list(rows[0].keys())) for row in rows: ws.append(list(row.values())) buf = BytesIO() wb.save(buf) buf.seek(0) return StreamingResponse(buf, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") ``` ## Error handling | Status | Condition | |--------|------------------------------------| | 413 | File exceeds size limit | | 422 | Unsupported file extension | | 500 | Parse failure (bad XML, CR format) | ## Thread safety FastAPI runs route handlers in thread pool workers by default. The Rust parser is `Send` but not `Sync`. Each request gets its own `CrystalXMLSource` instance, so there is no shared state. ======================================================================== PAGE: https://crxml.emiliano-go.com/flask/ ======================================================================== # Flask Integration Use crxml inside Flask views to upload, parse, and return Crystal Reports XML data as JSON, CSV, or XLSX. ## Upload + JSON response ```python from flask import Flask, request, jsonify from tempfile import NamedTemporaryFile from crxml import CrystalXMLSource, collect import os app = Flask(__name__) app.config["MAX_CONTENT_LENGTH"] = 500 * 1024 * 1024 # 500 MB @app.post("/parse-report") def parse_report(): if "file" not in request.files: return jsonify({"error": "No file provided"}), 400 file = request.files["file"] if not file.filename: return jsonify({"error": "Empty filename"}), 400 ext = os.path.splitext(file.filename or "")[1].lower() if ext not in (".xml", ".rpt"): return jsonify({"error": "Unsupported file type"}), 422 with NamedTemporaryFile(delete=False, suffix=".xml") as tmp: tmp.write(file.read()) tmp_path = tmp.name try: rows = collect(CrystalXMLSource(tmp_path)) return jsonify({"rows": len(rows), "data": rows[:100]}) except Exception as e: return jsonify({"error": str(e)}), 500 finally: os.unlink(tmp_path) ``` ## Streaming CSV download ```python from flask import Response import csv import io @app.post("/to-csv") def to_csv(): file = request.files["file"] with NamedTemporaryFile(delete=False, suffix=".xml") as tmp: tmp.write(file.read()) tmp_path = tmp.name def generate(): src = CrystalXMLSource(tmp_path) writer = None for row in src: if writer is None: output = io.StringIO() writer = csv.DictWriter(output, fieldnames=list(row.keys())) writer.writeheader() yield output.getvalue() output = io.StringIO() writer = csv.DictWriter(output, fieldnames=list(row.keys())) writer.writerow(row) yield output.getvalue() return Response(generate(), mimetype="text/csv", headers={ "Content-Disposition": "attachment; filename=report.csv" }) ``` ## XLSX download ```python from flask import send_file from openpyxl import Workbook from io import BytesIO @app.post("/to-xlsx") def to_xlsx(): file = request.files["file"] with NamedTemporaryFile(delete=False, suffix=".xml") as tmp: tmp.write(file.read()) tmp_path = tmp.name try: rows = collect(CrystalXMLSource(tmp_path)) finally: os.unlink(tmp_path) wb = Workbook() ws = wb.active if rows: ws.append(list(rows[0].keys())) for row in rows: ws.append(list(row.values())) buf = BytesIO() wb.save(buf) buf.seek(0) return send_file(buf, mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", as_attachment=True, download_name="report.xlsx") ``` ## Error handling | Status | Condition | |--------|------------------------------------| | 400 | No file or empty filename | | 413 | File exceeds size limit | | 422 | Unsupported file extension | | 500 | Parse failure (bad XML, CR format) | ## Config-based field mapping Define a mapping in app config so non-developers can adjust field names: ```python app.config["CRXML_FIELD_MAP"] = { "{Report.InvoiceNo}": "invoice", "{Report.Customer}": "customer", "{Report.Amount}": "amount", } @app.post("/parse-mapped") def parse_mapped(): file = request.files["file"] mapping = app.config["CRXML_FIELD_MAP"] with NamedTemporaryFile(delete=False, suffix=".xml") as tmp: tmp.write(file.read()) tmp_path = tmp.name try: src = CrystalXMLSource(tmp_path) pipe = src | RenameFields(mapping) rows = collect(pipe) return jsonify({"rows": len(rows), "data": rows[:100]}) finally: os.unlink(tmp_path) ``` ## Error handling middleware Register an error handler for crxml-specific errors: ```python from crxml import UnpicklableStageError @app.errorhandler(UnpicklableStageError) def handle_unpicklable(e): return jsonify({"error": "Stage not compatible with parallel mode"}), 400 @app.errorhandler(ValueError) def handle_value_error(e): return jsonify({"error": f"Parse error: {e}"}), 422 ``` ## Thread safety Flask's development server is single-threaded by default. In production with a WSGI server (gunicorn, waitress), each worker has its own `CrystalXMLSource` instance, no shared state. ======================================================================== PAGE: https://crxml.emiliano-go.com/getting-started/ ======================================================================== # Getting Started This guide walks through a complete round-trip: installation, first parse, schema inspection, a simple pipeline, and DataFrame conversion. ## Install ```bash pip install crxml ``` See [Installation](installation.md) for details on building from source and platform support. ## Your first source Create a small Crystal Reports XML file and point `CrystalXMLSource` at it: ```python from crxml import CrystalXMLSource src = CrystalXMLSource("report.xml") for row in src: print(row) ``` Each `row` is a `dict[str, str]`. The keys are field names from the CR XML (e.g. `{Report.FieldName}`) and the values are the raw text content. ## Inspect the schema Use `.schema()` to see the fields without consuming the stream: ```python src = CrystalXMLSource("report.xml") fields = src.schema() # list of field name strings ``` This is useful for building dynamic pipelines. ## Simple pipeline The `|` operator chains transformation stages. Nothing executes until you iterate or sink the result: ```python from crxml import CrystalXMLSource, RenameFields, CastTypes, DropFields pipe = ( CrystalXMLSource("report.xml") | RenameFields({ "{Report.InvoiceNo}": "invoice", "{Report.Customer}": "customer", "{Report.Amount}": "amount", }) | CastTypes({"amount": float}) | DropFields(["{Report.TaxRate}"]) ) for row in pipe: print(row["invoice"], row["amount"]) ``` ## Convert to DataFrame ```python from crxml import to_dataframe df = to_dataframe(pipe) ``` This collects all rows into a pandas DataFrame. For large files use `chunksize=` to build the DataFrame incrementally (see [Sinks](usage/sinks.md)). ## Next steps - [Usage guide](usage/basic.md), deeper topics: custom stages, parallel mode, branching - [Pipeline API](usage/pipeline.md), how `|` and lazy evaluation work - [Built-in stages](usage/stages.md), reference for all four stage types - [Performance](performance.md), benchmarks, memory model, bottlenecks ======================================================================== PAGE: https://crxml.emiliano-go.com/ ======================================================================== # crxml High-performance Crystal Reports XML to Arrow/DataFrame engine for Python. Parse, filter, rename, cast, and project Crystal Reports XML directly into columnar data, with Rust execution, parallel parsing, bounded-memory processing, and automatic query fusion. ## Features - Streaming: never loads the full file into memory - Fast: Rust parser via PyO3 + quick-xml - Pipeline API: compose transformations with `|` - Parallel mode: multi-core batch processing - Pandas-native: direct to DataFrame or CSV ## Quick Example ```python from crxml import CrystalXMLSource, RenameFields, CastTypes, to_dataframe pipe = ( CrystalXMLSource("report.xml") | RenameFields({"f1": "invoice", "f2": "amount"}) | CastTypes({"amount": float}) ) df = to_dataframe(pipe) ``` ## License MIT ======================================================================== PAGE: https://crxml.emiliano-go.com/installation/ ======================================================================== # Installation ## From PyPI ```bash pip install crxml ``` Pre-built wheels are published for Linux x86_64, macOS (arm64 + x86_64), and Windows x86_64. The wheel includes the compiled Rust extension so no Rust toolchain is required. ## Verify the Rust extension ```python from crxml._crxml_core import CrxmlReader print("Rust backend OK") ``` ## Building from source If no pre-built wheel matches your platform, pip will build from source. This requires the Rust toolchain. ### Prerequisites - Python ≥ 3.10 - [Rust](https://rustup.rs) (stable) The Rust core fetches its engine dependency (`rypipe-core`) from crates.io, so no other repositories are needed: `pip install .` and maturin work from a plain checkout of crxml alone. ### Build the Rust core ```bash pip install maturin maturin build --release # builds the crate in src/crxml_core/ ``` Or use `maturin develop` during development: ```bash maturin develop --release ``` ## Supported platforms | Platform | Wheel | |-------------|-------| | Linux x86_64 | ✅ | | macOS arm64 | ✅ | | macOS x86_64 | ✅ | | Windows x86_64 | ✅ | ======================================================================== PAGE: https://crxml.emiliano-go.com/performance/ ======================================================================== # Performance ## Environment All measurements recorded on a single development machine: | Component | Detail | |---|---| | **CPU** | 13th Gen Intel Core i5-1335U (10 cores: 2 P + 8 E, 12 threads) | | **L1d** | 352 KiB (10 instances) | | **L2** | 6.5 MiB (4 instances) | | **L3** | 12 MiB (1 instance) | | **RAM** | 15 GiB LPDDR5 (system-unknown speed) | | **OS** | Arch Linux, kernel 7.0.9-arch2-1 | | **Python** | 3.14.5 | | **pyarrow** | 24.0.0 | | **crxml** | 0.3.0 | | **Git SHA** | `bbc8a172` | | **Build** | release, LTO enabled, mimalloc allocator, feature `profile` | All runs are **warm-cache** (one warmup parse, collection, then measured). Each number is the best of 3 runs after variance stabilized. Parallel-path variance was ~8% on the 533 MB file, stream-path variance ~15%. ## Input files | File | Size | Rows | Fields/row | Origin | |---|---|---|---|---| | `test_10mb.xml` | 10 MB | 9,010 | 10 | Synthetic (`benchmarks/benchmarks.py`) | | `test_50mb.xml` | 50 MB | 45,328 | 10 | Synthetic | | `test_100mb.xml` | 100 MB | 90,384 | 10 | Synthetic | | `test_533mb.xml` | 533 MB | 465,136 | 11 | Real Crystal Reports export | Synthetic files use uniform rows, every field present on every row, and low cardinality (most fields repeat). These properties flatter parallel load balance and dictionary encoding, so synthetic numbers are **directional only**. The 533 MB real export is the ground truth for all reported conclusions. Synthetic files have 10 columns (including `FieldG`, absent from the real export) with cardinalities ranging from 1 (`Level`, `Section`, `Text20`) through 15 (`Field38`, `Field39`) to near-unique (`Field22`: 8,965 / 9,010). ### Field cardinality (real 533 MB file, 465,136 rows) Only 5 of 11 columns have high cardinality (≥1,000 distinct values); the other 6 are dictionary-encoding candidates. Not every column appears in every row (`Field72` and `Text21` are sparse); the rypipe engine discovers all distinct column names across all rows. | Column | Distinct values | |---|---| | `Level` | 1 | | `Section` | 1 | | `Text20` | 1 | | `Text21` | 1 | | `Field73` | 36 | | `Field72` | 8 | | `Field23` | 145 | | `Field38` | 1,528 | | `Field39` | 1,485 | | `Field61` | 1,406 | | `Field22` | 4,230 | ## Speed: end-to-end `to_dataframe()` (the user's actual goal) | Engine | 10 MB | 50 MB | 100 MB | 533 MB | |---|---|---|---|---| | **Stream** | 248 ms / 40 MB/s / 36k r/s | 1.35 s / 36 MB/s / 34k r/s | 2.27 s / 44 MB/s / 40k r/s | **12.8 s / 42 MB/s / 36k r/s** | | **Parallel** (8 workers) | 30 ms / 335 MB/s / 300k r/s | 124 ms / 402 MB/s / 365k r/s | 213 ms / 469 MB/s / 424k r/s | **1.13 s / 472 MB/s / 412k r/s** | | **Parallel + auto-dict** | 38 ms | 172 ms | 312 ms | **2.0 s / 267 MB/s** | Key observations: - **Parallel throughput improves with file size** (335 → 472 MB/s) because split-scan and worker startup are fixed costs that amortize. The 533 MB number (472 MB/s) is the asymptotic rate. - **Parallel is ~11× faster than stream** on the 533 MB real file. - **Auto-dict adds ~0.9 s of on-GIL overhead** at 533 MB (dictionary encoding happens after the GIL is reacquired). Use it only when downstream readers benefit from dictionary-encoded Arrow columns. ## Parallel-path breakdown (533 MB real file) | Phase | Time | % of wall | Notes | |---|---|---|---| | Split-scan (serial) | 257 ms | 23% | Two SIMD scans for ` RAM) | Pages evicted under pressure | Heap holds the full copy | | Cold-cache startup | Page-fault-driven (first touch) | Sequential read-ahead | On warm cache the RSS delta is near zero because the kernel's page cache already holds the file. The case for mmap is files near or exceeding physical RAM, where the OS can evict pages under memory pressure. The case for `fs::read` is cold-cache streaming, where the kernel's read-ahead is more predictable than page faults. ## Correctness Every performance measurement in this document is backed by a correctness cross-check: all three engines (stream, columnar, parallel) produce byte-identical field values against both the stream-oracle and against `xml.etree.ElementTree` on the synthetic corpus. The parallel engine's row-split boundaries are validated by the splitter test suite (18 tests covering prefix collision, CDATA/comment skipping, fallback, and random input). No engine cuts corners. ## The ceiling At 472 MB/s on a machine with ~30 GB/s of memory bandwidth, this parser is **CPU-bound**, not bandwidth-bound. The bottleneck is not moving bytes; it's tokenizing XML elements, unescaping entities, and copying field values. The breakdown says parse is 69% of wall time and the biggest sub-cost is the quick-xml event loop (tokenizing every ``, ``, ``, ``, `` child even when the field is dropped by the `ExecutionPlan`). The remaining high-leverage improvement is **skip-bytes-for-unwanted-fields**: detecting a dropped column name and memmem-skipping to `` or `` without tokenizing children. **Honest throughput ceiling for this codebase:** | CPU | Estimated ceiling (parallel) | |---|---| | i5-1335U (this machine) | ~500-550 MB/s | | Ryzen 7 5800X (desktop) | ~800 MB/s - 1.1 GB/s | 2 GB/s would require a genuinely different parse strategy (a hand-rolled scanner for the fixed Crystal Reports XML structure that skips quick-xml's generality), which is a separate project, not a tuning pass. ======================================================================== PAGE: https://crxml.emiliano-go.com/pipeline-fusion/ ======================================================================== # Pipeline Fusion Pipeline fusion compresses multiple transformation stages into a single execution pass, reducing Python overhead and memory allocations. ## Three levels of fusion crxml has three fusion mechanisms: | Level | Mechanism | When it applies | |-------|-----------|-----------------| | Dict-level fusion | `apply` + `__call__` protocol | Any pipeline with fusable stages | | Columnar fusion | `_plan_kwargs` into Rust `ExecutionPlan` | Source supports rypipe engine, stages export a plan | | Vectorized batch chain | Volcano-style pull on Arrow `RecordBatch` | After columnar fusion, remaining stages implement `_plan_kwargs` | A stage is **fusable** if it has both `apply(self, record) -> dict | None` and `__call__(self, stream)`. When a contiguous run of fusable stages exists at the front of the pipeline, they are fused into a single tight loop that avoids Python generator overhead. A stage supports **columnar fusion** if it implements `_plan_kwargs(self) -> dict | None`. When all stages in a pipeline are columnar-fusable, the entire pipeline compiles into the rypipe engine and no Python dicts are created until the final Arrow table is converted. ## Decision tree When you iterate a pipeline, `fusion.py` follows this logic: 1. **Try columnar fusion**: if the source has a `_read_arrow` method and stages export `_plan_kwargs`, the pipeline runs entirely in Rust. 2. **Vectorized batch chain**: if columnar fusion found pushdown stages but remaining stages exist, they run on Arrow `RecordBatch` objects via the batchpipe engine, keeping data in columnar format and avoiding per-row Python dict construction. 3. **Dict-level fusion**: if columnar fusion is not possible, the first contiguous run of fusable stages is fused into a single loop. 4. **Sequential**: remaining stages run as Python generators on the dict stream. ## When fusion applies | Pipeline | Fusion level | Performance | |----------|-------------|-------------| | `Source \| RenameFields \| CastTypes` | Columnar | Fastest (all Rust) | | `Source \| DropFields \| RenameFields` | Columnar | Fastest (all Rust) | | `Source \| FilterRows(field=..., op=..., value=...)` | Columnar | Fastest (all Rust) | | `Source \| custom_fusable_stage \| CastTypes` | Dict-level | Fast (no generator overhead) | | `Source \| generator_func \| CastTypes` | Sequential | Fusable stages after generator are NOT fused | | `Source \| CastTypes \| generator_func` | Columnar + dict tail | Columnar up to the generator, then dicts | For optimal performance, place fusable stages at the front of the pipeline: ```python # Good: CastTypes and DropFields fuse into columnar plan pipe = source | CastTypes({"amt": float}) | DropFields(["tmp"]) | custom_filter # Less good: custom_filter breaks the fusable chain pipe = source | custom_filter | CastTypes({"amt": float}) | DropFields(["tmp"]) ``` ## How columnar fusion works 1. The pipeline calls `_try_columnar_fusion(source, stages)`. 2. For each stage, `_plan_kwargs()` is called. If it returns a dict, the kwargs are merged into a single `plan_overrides` dict and the stage is skipped in the Python stage list. 3. `source._read_arrow(plan_overrides=plan_overrides)` is called. This runs the rypipe engine with the fused plan, producing a `pyarrow.Table` directly from the XML. 4. The Arrow table is wrapped in a row-by-row dict iterator. 5. Any remaining stages (those that did not provide `_plan_kwargs`) run on the dict stream. This means columnar fusion can be partial. A pipeline like: ```python source | RenameFields({...}) | CastTypes({...}) | custom_lambda ``` will execute `RenameFields` and `CastTypes` in Rust, produce an Arrow table, convert to dicts, then apply `custom_lambda` to each dict. No unnecessary Python object creation happens for the fused stages. ## How dict-level fusion works 1. The pipeline scans stages from the front until it finds a non-fusable stage (no `apply` method). 2. All fusable stages are combined into a single `fused()` generator: ```python def fused(): for record in source: r = record for fn in bound_stage_applies: r = fn(r) if r is None: break else: yield r ``` 3. Non-fusable stages wrap the fused generator. ## Verifying fusion Set logging to DEBUG to see fusion decisions: ```python import logging logging.basicConfig(level=logging.DEBUG) # Logs: "columnar fusion with overrides: ..." or "fused N stages" ``` Or check programmatically by inspecting the pipeline: ```python pipe = source | CastTypes({"x": float}) print(type(pipe._stages[0])) # ``` If the pipeline uses the rypipe engine, `source._read_arrow` is called internally and the Rust-side profile counters show the fused plan. ## Performance comparison Using a 100 MB file with a 4-stage pipeline: | Pipeline | Time | Speedup vs sequential | |----------|------|----------------------| | Sequential (no fusion) | 2.27s | 1x | | Dict-level fusion only | 1.89s | 1.2x | | Full columnar fusion | 0.69s | 3.3x | Columnar fusion is particularly effective because it skips the two most expensive operations in the stream path: HTML unescaping and Python dict construction. ======================================================================== PAGE: https://crxml.emiliano-go.com/prefect/ ======================================================================== # Prefect Integration Use crxml inside Prefect 2.x flows to parse Crystal Reports XML files as part of a larger data pipeline. ## Installation ```bash pip install crxml prefect ``` ## Basic flow ```python from pathlib import Path from prefect import flow, task from crxml import CrystalXMLSource, RenameFields, CastTypes, to_csv @task def parse_report(path: str) -> list[dict]: src = CrystalXMLSource(path) return list(src) @task def transform_rows(rows: list[dict]) -> list[dict]: pipe = ( rows | RenameFields({ "{Report.InvoiceNo}": "invoice", "{Report.Customer}": "customer", "{Report.Amount}": "amount", }) | CastTypes({"amount": float}) ) return list(pipe) @task def export_csv(rows: list[dict], output_path: str) -> None: to_csv(rows, output_path) @flow def crxml_pipeline(input_path: str, output_path: str = "output.csv"): rows = parse_report(input_path) cleaned = transform_rows(rows) export_csv(cleaned, output_path) if __name__ == "__main__": crxml_pipeline("report.xml") ``` ## Streaming flow with task runners For large files, use Prefect's `ThreadPoolTaskRunner` to stream rows without loading everything into memory: ```python from prefect import flow, task from prefect.task_runners import ThreadPoolTaskRunner from crxml import CrystalXMLSource, RenameFields, CastTypes, to_csv @task def stream_to_csv(input_path: str, output_path: str) -> None: pipe = ( CrystalXMLSource(input_path) | RenameFields({ "{Report.InvoiceNo}": "invoice", "{Report.Amount}": "amount", }) | CastTypes({"amount": float}) ) to_csv(pipe, output_path) @flow(task_runner=ThreadPoolTaskRunner()) def crxml_streaming_flow(input_path: str): stream_to_csv(input_path, "output.csv") ``` The file is streamed row by row. RSS stays constant regardless of file size. ## Parallel parsing with mapped tasks Use Prefect task mapping to parse multiple files in parallel: ```python from pathlib import Path from prefect import flow, task from crxml import CrystalXMLSource, collect @task def parse_single(path: str) -> dict: rows = collect(CrystalXMLSource(path)) return {"file": path, "rows": len(rows)} @flow def parse_directory(data_dir: str = "./data"): paths = [str(p) for p in Path(data_dir).glob("*.xml")] results = parse_single.map(paths) for r in results: print(f"{r['file']}: {r['rows']} rows") ``` ## Error handling and retries ```python from prefect import flow, task from crxml import CrystalXMLSource, collect @task(retries=2, retry_delay_seconds=5) def parse_with_retry(path: str) -> list[dict]: return collect(CrystalXMLSource(path)) @flow def resilient_parse(path: str): try: rows = parse_with_retry(path) print(f"Parsed {len(rows)} rows") except FileNotFoundError: print(f"File not found: {path}") except ValueError as e: print(f"Parse error: {e}") ``` ## Caching parsed results ```python from pathlib import Path from prefect import flow, task from crxml import CrystalXMLSource, collect @task(cache_policy=INPUTS) def parse_cached(path: str) -> list[dict]: return collect(CrystalXMLSource(path)) @flow def cached_parse_flow(path: str): rows = parse_cached(path) print(f"Parsed {len(rows)} rows") return rows ``` Prefect caches the task result keyed on the input path. Re-running with the same file skips parsing. ## Deployment notes - crxml's Rust parser releases the GIL during XML processing, so it does not block the asyncio event loop in Prefect's async task runner. - Each task creates its own `CrystalXMLSource` instance. There is no shared state between tasks. - For very large files (over 500 MB), the streaming approach is strongly recommended over `collect()` to avoid memory pressure. - Prefect's `ProcessPoolTaskRunner` is not needed because crxml has its own parallel mode via `Pipeline.parallel()`. ======================================================================== PAGE: https://crxml.emiliano-go.com/rust-core/ ======================================================================== # Rust Core The native accelerator is a PyO3 crate at `src/crxml_core/`. ## Crate structure ``` src/crxml_core/ ├── Cargo.toml └── src/ ├── lib.rs # CrxmlReader class + thin columnar FFI wrappers └── xml/ # Crystal Reports XML adapter for rypipe-core ├── decoder.rs # CrystalXmlDecoder implements RecordParser ├── splitter.rs # CrystalXmlSplitter implements Splitter ├── error.rs # adapter error type └── mod.rs ``` ### `lib.rs`: streaming engine and columnar wrappers `lib.rs` now contains two parts: 1. **`CrxmlReader`**: the streaming XML parser (remains in crxml). 2. **Thin columnar FFI wrappers**: `#[pyfunction]` entry points (`read_to_columnar`, `read_to_columnar_multi`, `read_to_columnar_par`, `read_to_columnar_bounded`) that build an `rypipe_core::ExecutionPlan` and delegate parsing to `rypipe-core` through the embedded Crystal Reports XML adapter in `src/xml/`. #### `CrxmlReader` A single Python-exposed class: ```rust #[pyclass] struct CrxmlReader { source: PathBuf, row_tag: String, buf: Vec, inner_buf: Vec, } ``` - `__iter__`, returns `self` - `__next__`, reads the next row as a `PyDict` The reader walks the XML stream, finds `` elements, and extracts field key/value pairs from nested `` and `` elements. ## Dependencies | Crate | Purpose | |--------------|--------------------------------| | `pyo3` | Python bindings | | `quick-xml` | Streaming XML reader | | `arrow` | Arrow C Data Interface export | | `mimalloc` | Fast allocator (replaces system malloc, ~27% CPU savings) | | `rypipe-core`| Generic columnar/parallel/bounded engine (from crates.io) | | `memchr` | Fast substring scans for the XML splitter | | `simdutf8` | SIMD UTF-8 validation for the XML decoder | | `thiserror` | Adapter error derives | The `rypipe-core` crate is a versioned dependency resolved from crates.io (see `src/crxml_core/Cargo.toml`): ```toml rypipe-core = { version = "0.1", features = ["mmap"] } ``` No sibling checkout is required to build crxml. To hack on the engine itself, clone [rypipe](https://github.com/emiliano-go/rypipe) separately and point the dependency at your checkout with a cargo `[patch]` entry. ## Building ```bash # Development build (editable) maturin develop --release # Production wheel maturin build --release ``` The `pyproject.toml` `[tool.maturin]` section controls the build: ```toml [tool.maturin] module-name = "crxml._crxml_core" manifest-path = "src/crxml_core/Cargo.toml" ``` ## Code style - Rust 2021 edition - `cargo fmt` for formatting - `cargo clippy`, no warnings allowed - Unsafe code is denied by default (`#![deny(unsafe_code)]`) ## Testing ```bash # Rust unit tests (streaming engine + XML adapter) PYO3_USE_ABI3_FORWARD_COMPATIBILITY=1 cargo test --manifest-path src/crxml_core/Cargo.toml --all-features # Python test suite pytest ``` ======================================================================== PAGE: https://crxml.emiliano-go.com/troubleshooting/ ======================================================================== # Troubleshooting ## Common errors ### FileNotFoundError ```python CrystalXMLSource("nonexistent.xml") # FileNotFoundError: File not found: nonexistent.xml ``` The path must be a local file path. Remote URLs and file-like objects are not supported. ### Empty result (zero rows) The most common cause is a wrong `row_tag`. CR XML files use different tag names for record rows. Inspect the file to find the correct tag: ```bash grep -o '<[A-Za-z][A-Za-z0-9]*' report.xml | sort | uniq -c | sort -rn | head -10 ``` The most frequent non-wrapper tag is usually the row tag. Pass it explicitly: ```python src = CrystalXMLSource("report.xml", row_tag="Detail") ``` ### ValueError on bad CR XML The parser raises `ValueError` if the XML is malformed or does not match the Crystal Reports schema. Check that: - The file is well-formed XML (validate with `xmllint`) - The file contains repeating row elements (not a single record) - The row tag contains `` children with `FieldName` attributes ### TypeError from CastTypes ```python CastTypes({"amount": float}, errors="raise") # raises on uncastable values ``` Use `errors="coerce"` to replace uncastable values with `None`, or `errors="skip"` to leave them unchanged. ### UnpicklableStageError Raised when calling `.parallel()` on a pipeline with non-picklable stages. Common causes: - Lambdas used as predicates in `FilterRows` - Local/nested functions used as stages - Custom class instances that cannot be pickled Fix: use module-level functions or built-in stages with the keyword-based `FilterRows(field=..., op=..., value=...)` API. ```python # Causes UnpicklableStageError: pipe | FilterRows(lambda r: r.get("x") == "y") # Works with .parallel(): pipe | FilterRows(field="x", op="==", value="y") ``` ### Cargo error: failed to load source for dependency `rypipe-core` ```text error: failed to get `rypipe-core` as a dependency of package `crxml-core` ``` The Rust core consumes the engine as a versioned crate from crates.io, so a build failure here almost always means the build environment has no network access to crates.io, or an offline mirror is missing the `0.1.x` release. Fix: allow egress to crates.io (or vendor `rypipe-core` with `cargo vendor`) and retry: ```bash cargo update -p rypipe-core pip install . ``` This only affects building from source; PyPI wheels bundle the compiled extension and need no Rust toolchain at all. Historical note: before 1.2.0 this dependency was a path reference into a sibling `../rypipe` checkout, which broke every build environment without that clone. ## FAQ ### How do I find the right row_tag? Open the XML in a text editor. Look for a repeating element that wraps each record. In standard CR XML this is `
`, but it varies by report. Common values: `Details`, `Detail`, `Row`, `Record`, `Item`, `Group`. ### Why are my field names like {Report.InvoiceNo}? Crystal Reports XML uses `{Report.FieldName}` as the `FieldName` attribute. This is the raw key from the XML. Use `RenameFields` to map them to shorter names: ```python CrystalXMLSource("report.xml") | RenameFields({ "{Report.InvoiceNo}": "invoice", "{Report.Customer}": "customer", }) ``` ### Parallel mode is slower than sequential Parallel mode adds overhead for batch serialization and IPC. It is recommended for files larger than 50 MB. For small files, sequential iteration is faster. If parallel is slower on a large file, check: - Are all stages picklable? (`validate_stages_picklable` from `crxml.parallel`) - Is the file on a fast SSD? (disk I/O can be the bottleneck) - Is `batch_size` tuned? (try 5000 to 20000) - Are you using the right number of workers? (defaults to CPU count) ### Pipeline fusion is not happening Check that your stages are compatible: - `RenameFields`, `CastTypes`, `DropFields`, `FilterRows` with keyword args all support columnar fusion. - Custom stages need `_plan_kwargs()` returning a dict. - Non-fusable stages (lambdas, generators) break the fused chain. Put them after fused stages to minimize the performance impact. ### The Rust extension won't build Ensure you have the Rust toolchain installed: ```bash rustup install stable ``` If building with the `columnar` feature, additional dependencies may be required (Arrow, Parquet). On Linux, install: ```bash sudo apt-get install libarrow-dev libparquet-dev ``` For other platforms, see the Apache Arrow C++ install guide. ### How do I process files larger than RAM? crxml streams the XML in constant memory. The RSS stays well below the file size (about 75 MB for a 100 MB file). Use `to_csv` or chunked `to_dataframe` to avoid loading all rows at once: ```python to_csv(pipe, "output.csv") ``` ======================================================================== PAGE: https://crxml.emiliano-go.com/usage/basic/ ======================================================================== # Basic Parsing ## Opening a file `CrystalXMLSource` accepts a file path (string or `pathlib.Path`): ```python from crxml import CrystalXMLSource # path string src = CrystalXMLSource("report.xml") # pathlib.Path from pathlib import Path src = CrystalXMLSource(Path("report.xml")) ``` ### Parameters | Param | Type | Default | Description | |----------|--------------------|-----------|-------------------------------| | `source` | `str \| Path` | required | Path to CR XML file | | `row_tag`| `str` | `"Row"` | XML tag for each record row | The `row_tag` parameter lets you target a different repeating element if your CR XML uses a non-standard tag name. ## Iteration `CrystalXMLSource` is iterable. Each row is a `dict[str, str]`: ```python for row in CrystalXMLSource("report.xml"): print(row["{Report.InvoiceNo}"], row["{Report.Amount}"]) ``` Keys are the `FieldName` attribute values from the CR XML (e.g. `{Report.InvoiceNo}`). Values are the raw text of the first `` or `` child element. ## Schema inspection Call `.schema()` to discover fields without consuming the stream: ```python src = CrystalXMLSource("report.xml") fields = src.schema() # list of field name strings ``` The source yields rows internally and caches them, so the first batch is not lost. `.schema()` is safe to call before building a pipeline. ## Memory model The parser streams the file in constant memory. The Rust backend reuses internal buffers across rows and never materializes the full document. RSS scales with file content (22 MB for 10 MB, 75 MB for 100 MB), staying well below file size. pandas is imported lazily: memory climbs only when `to_dataframe` is called. ## CR XML layout detection Crystal Reports XML stores field values in two patterns: - **Attribute style:** `123.45` - **Element style:** `{Report.Amount}123.45` - **Mixed:** some fields use attributes, others use child elements The parser detects both styles automatically, no configuration needed. ======================================================================== PAGE: https://crxml.emiliano-go.com/usage/branching/ ======================================================================== # Branching Pipelines are immutable. This makes it safe to reuse a base pipeline as the starting point for multiple branches. ## Example: split by region ```python from crxml import CrystalXMLSource, FilterRows, to_csv base = CrystalXMLSource("report.xml") north = base | FilterRows(lambda r: r.get("region") == "north") south = base | FilterRows(lambda r: r.get("region") == "south") to_csv(north, "north.csv") to_csv(south, "south.csv") ``` Each branch is independent. The source file is re-read for each branch, so disk I/O scales linearly with the number of branches. ## Example: different transformations per branch ```python from crxml import CrystalXMLSource, RenameFields, CastTypes, to_dataframe, to_csv base = CrystalXMLSource("report.xml") # Branch A: rename + CSV export branch_a = base | RenameFields({"f1": "invoice", "f2": "amount"}) to_csv(branch_a, "export.csv") # Branch B: cast + DataFrame branch_b = base | CastTypes({"amount": float}) df = to_dataframe(branch_b) ``` ## Performance note Each branch re-opens and re-parses the source file. For large files with many branches, consider: - Using a single pipeline with `FilterRows` for each output path - Pre-filtering with external tools like `grep` or `xsv` - Caching the parsed stream to a temporary file first ======================================================================== PAGE: https://crxml.emiliano-go.com/usage/custom-stages/ ======================================================================== # Custom Stages There are three styles for writing custom pipeline stages. ## Generator style A generator function that yields transformed rows: ```python def uppercase_names(stream): for row in stream: if "name" in row: row["name"] = row["name"].upper() yield row ``` Usage: ```python pipe = CrystalXMLSource("report.xml") | uppercase_names ``` ## Map style A function that returns a map iterator: ```python def strip_whitespace(stream): return map(lambda r: {k: v.strip() for k, v in r.items()}, stream) ``` ## Fusable protocol For optimal performance (especially in parallel mode), implement a class with both `apply` and `__call__`: ```python class MultiplyField: def __init__(self, field: str, factor: float): self.field = field self.factor = factor def apply(self, record: dict) -> dict | None: if self.field in record: try: record[self.field] = float(record[self.field]) * self.factor except (ValueError, TypeError): pass return record def __call__(self, stream): for row in stream: yield self.apply(row) ``` When a stage implements the `Fusable` protocol (has `apply` and `__call__`), the pipeline can fuse a contiguous run of fusable stages into a single tight loop, avoiding Python generator overhead. ## When to use each style | Style | Best for | Parallel? | |-------|----------|-----------| | Generator | Simplicity, complex logic | No (captures self/closures) | | Map | Simple transforms | No (lambda not picklable) | | Fusable | Performance, parallel mode | Yes | ## Picklability for parallel mode To use a custom stage with `.parallel()`, it must be picklable: - Top-level module functions only (no lambdas) - Classes with `__init__` storing simple data - No closures or local functions crxml validates picklability at pipeline construction time and raises `UnpicklableStageError` for incompatible stages. ## Columnar plan fusion For maximum performance, a stage can implement `_plan_kwargs(self) -> dict | None`. When this method returns a dict, the stage is compiled into the engine's `ExecutionPlan` and runs during XML parsing, before any Python dict is created. This bypasses Python entirely for that stage. Built-in stages that support columnar plan fusion: | Stage | `_plan_kwargs` effect | |-------|----------------------| | `RenameFields` | Adds `field_mapping` to the plan | | `CastTypes` | Adds `field_types` to the plan | | `DropFields` | Adds `drop_fields` to the plan | | `FilterRows` | Adds `filter` to the plan (constant/column predicates only) | Example of a custom stage that fuses into the columnar plan: ```python class DropFieldsIfEmpty: def __init__(self, fields: list[str]): self.fields = fields def apply(self, record: dict) -> dict | None: for f in self.fields: record.pop(f, None) return record def __call__(self, stream): for row in stream: yield self.apply(row) def _plan_kwargs(self) -> dict | None: return {"drop_fields": self.fields} ``` Notes: - `_plan_kwargs` is only called by `CrystalXMLSource` objects that support the columnar engine. - If `_plan_kwargs` returns `None`, the stage is treated as a regular fusable stage (dict-level fusion). - Non-fusable stages in the pipeline are always applied as Python generators on the dict stream after columnar fusion completes. ======================================================================== PAGE: https://crxml.emiliano-go.com/usage/ ======================================================================== # Usage The usage guides cover everything from basic file parsing to advanced topics like parallel execution and custom stage authoring.
- __[Basic Parsing](basic.md)__ Open files, iterate rows, inspect schema, memory model - __[Pipeline API](pipeline.md)__ The `|` operator, lazy evaluation, immutable composition - __[Built-in Stages](stages.md)__ Reference for RenameFields, CastTypes, DropFields, FilterRows - __[Custom Stages](custom-stages.md)__ Write your own pipeline stages - __[Parallel Execution](parallel.md)__ Multi-core batch processing - __[Sinks](sinks.md)__ DataFrame, CSV, list collection - __[Branching](branching.md)__ Reuse a base pipeline for multiple outputs - __[Pipeline Fusion](../pipeline-fusion.md)__ How stages compile into the Rust engine - __[Troubleshooting](../troubleshooting.md)__ Common errors, FAQs, debugging tips
======================================================================== PAGE: https://crxml.emiliano-go.com/usage/parallel/ ======================================================================== # Parallel Execution For large files, `.parallel()` splits work across multiple processes. ## Usage ```python from crxml import CrystalXMLSource, RenameFields, CastTypes, to_dataframe pipe = ( CrystalXMLSource("report.xml") | RenameFields({"f1": "name", "f2": "total"}) | CastTypes({"total": float}) ) df = to_dataframe(pipe.parallel(workers=4, batch_size=5000)) ``` ## Parameters | Param | Type | Default | Description | |--------------|-------|---------|----------------------------------------| | `workers` | `int` | `None` | Number of worker processes (CPU count) | | `batch_size` | `int` | `1000` | Rows per batch sent to workers | ## Requirements - All stages in the pipeline must be **fusable** (implement `apply` + `__call__`) - All stages must be **picklable** (no lambdas, no closures) - The source must be iterable multiple times (file re-opened per batch) ## How it works 1. A reader thread reads the source and splits rows into batches. 2. Batches are dispatched to a `ProcessPoolExecutor`. 3. Each worker runs the fused pipeline on its batch. 4. Results are returned in order via futures. ## When to use Parallel mode adds overhead for batch serialization and IPC. The heuristic: | File size | Recommended | |------------|-------------| | < 50 MB | Sequential | | 50 MB to 200 MB | Recommended | | > 200 MB | Parallel | ## Validation crxml validates all stages at pipeline construction: ```python from crxml import CrystalXMLSource, RenameFields, FilterRows pipe = CrystalXMLSource("report.xml") | RenameFields({"a": "b"}) # This works, both stages are Fusable and picklable pipe2 = CrystalXMLSource("report.xml") | FilterRows(lambda r: r["x"] > 1) pipe2.parallel() # raises UnpicklableStageError, lambda not picklable ``` ## Named function example ```python def above_threshold(row): return row if float(row.get("amount", 0)) > 100 else None pipe = CrystalXMLSource("report.xml") | above_threshold pipe.parallel(workers=2) # works, module-level function ``` ======================================================================== PAGE: https://crxml.emiliano-go.com/usage/pipeline/ ======================================================================== # Pipeline API The `|` operator composes transformation stages into a lazy pipeline. ## How it works `CrystalXMLSource | stage` returns a `Pipeline` object. Chaining multiple stages creates a composition; nothing executes until you iterate or sink the result. ```python from crxml import CrystalXMLSource, RenameFields, CastTypes pipe = ( CrystalXMLSource("report.xml") | RenameFields({"f1": "name", "f2": "total"}) | CastTypes({"total": float}) ) # No iteration has happened yet for row in pipe: # execution starts here print(row) ``` ## Immutable composition Pipelines are immutable. Every `|` produces a new `Pipeline` object without modifying the previous one: ```python base = CrystalXMLSource("report.xml") | RenameFields(mapping) # These are independent, each re-reads the source pipe_a = base | CastTypes({"amount": float}) pipe_b = base | DropFields("tax_rate") ``` ## Pipeline object Usually created implicitly via `|`. The `Pipeline` class is also importable: ```python from crxml import Pipeline, CrystalXMLSource, RenameFields pipe = Pipeline(CrystalXMLSource("report.xml"), RenameFields(mapping)) ``` ## Lazy evaluation Pipelines are fully lazy until one of: - `for row in pipeline:`, per-row iteration - `list(pipeline)`, collect all rows - `to_dataframe(pipeline)`, DataFrame sink - `to_csv(pipeline, path)`, CSV sink - `collect(pipeline)`, list sink ## Example: 3-stage pipeline ```python pipe = ( CrystalXMLSource("report.xml") | RenameFields({"vendor": "supplier", "price": "cost"}) | CastTypes({"cost": float}) | FilterRows(lambda r: r["cost"] > 100) ) ``` Each stage processes rows in sequence. A row dropped by `FilterRows` never reaches later stages. ======================================================================== PAGE: https://crxml.emiliano-go.com/usage/sinks/ ======================================================================== # Sinks Sinks terminate a pipeline and materialize the result. ## to_dataframe ```python to_dataframe(pipeline, chunksize: int | None = None) -> pd.DataFrame ``` Collects all rows into a pandas DataFrame. - `chunksize=None` (default): builds a list of dicts, then constructs the DataFrame. Simple but memory-intensive for large outputs. - `chunksize=N`: incrementally builds the DataFrame in chunks of N rows, then concatenates. Lower peak memory. ```python from crxml import CrystalXMLSource, RenameFields, to_dataframe pipe = CrystalXMLSource("report.xml") | RenameFields({"f1": "name"}) df = to_dataframe(pipe, chunksize=10000) ``` ## to_csv ```python to_csv(pipeline, path: str | Path, **csv_writer_kwargs) -> None ``` Streams rows directly to CSV. Supports all `csv.writer` kwargs via `**csv_writer_kwargs`: ```python from crxml import CrystalXMLSource, to_csv pipe = CrystalXMLSource("report.xml") to_csv(pipe, "output.csv", delimiter=";", quoting=1) ``` The CSV is written incrementally as rows are produced. No intermediate list. ## collect ```python collect(pipeline) -> list[dict] ``` Materializes the pipeline into a list of dicts. Useful for testing and debugging: ```python from crxml import CrystalXMLSource, collect rows = collect(CrystalXMLSource("report.xml")) assert len(rows) == 8306 ``` ## XLSX via openpyxl crxml does not have a built-in XLSX sink, but you can build one easily: ```python from openpyxl import Workbook from crxml import CrystalXMLSource, collect rows = collect(CrystalXMLSource("report.xml")) wb = Workbook() ws = wb.active if rows: ws.append(list(rows[0].keys())) for row in rows: ws.append(list(row.values())) wb.save("output.xlsx") ``` For large files, chunk the writes to avoid loading all rows into memory. ======================================================================== PAGE: https://crxml.emiliano-go.com/usage/stages/ ======================================================================== # Built-in Stages ## RenameFields ```python RenameFields(mapping: dict[str, str]) ``` Renames dict keys according to `mapping`. Unmapped keys pass through unchanged. **Example:** ```python # Input: {"{Report.Vendor}": "Acme", "{Report.Price}": "50.00"} # Output: {"supplier": "Acme", "cost": "50.00"} RenameFields({"{Report.Vendor}": "supplier", "{Report.Price}": "cost"}) ``` ## CastTypes ```python CastTypes(types: dict[str, type], errors: str = "raise") ``` Casts specified fields to the given types. **`errors` modes:** - `"raise"` (default), raises `TypeError` on conversion failure - `"coerce"`, replaces uncastable values with `None` - `"skip"`, leaves uncastable values as-is **Example:** ```python # Input: {"invoice": "INV-001", "qty": "3", "price": "19.99"} # Output: {"invoice": "INV-001", "qty": 3, "price": 19.99} CastTypes({"qty": int, "price": float}) ``` ## DropFields ```python DropFields(fields: list[str]) ``` Removes specified keys from each row. **Example:** ```python # Input: {"a": "1", "b": "2", "c": "3"} # Output: {"a": "1", "c": "3"} DropFields(["b"]) ``` ## FilterRows ```python FilterRows(predicate: Callable[[dict[str, Any]], bool]) ``` Keeps only rows where `predicate(row)` returns `True`. **Example:** ```python # Input: [{"amt": "10"}, {"amt": "200"}, {"amt": "50"}] # Output: [{"amt": "200"}] FilterRows(lambda r: float(r.get("amt", 0)) > 100) ``` ## Edge cases - **RenameFields:** If a mapping key does not exist in the row, it is silently ignored. Duplicate target names are not checked, the last mapping wins. - **CastTypes:** When `errors="coerce"`, the coerced value is `None`. The original key is always preserved in the output dict. - **DropFields:** Dropping a non-existent key is a no-op. - **FilterRows:** The predicate receives the row *after* all prior stages.