Metadata-Version: 2.4
Name: crxml
Version: 1.0.0
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Rust
Classifier: Topic :: Scientific/Engineering :: Information Analysis
Classifier: Topic :: Text Processing :: Markup :: XML
Requires-Dist: seoslug
Requires-Dist: crxml[pyarrow,pandas,polars] ; extra == 'all'
Requires-Dist: pytest>=7 ; extra == 'dev'
Requires-Dist: pytest-benchmark>=4 ; extra == 'dev'
Requires-Dist: openpyxl>=3.1 ; extra == 'dev'
Requires-Dist: crxml[all] ; extra == 'dev'
Requires-Dist: pandas>=1.5 ; extra == 'pandas'
Requires-Dist: polars>=0.20 ; extra == 'polars'
Requires-Dist: pyarrow>=15 ; extra == 'pyarrow'
Provides-Extra: all
Provides-Extra: dev
Provides-Extra: pandas
Provides-Extra: polars
Provides-Extra: pyarrow
License-File: LICENSE
Summary: Fast streaming parser for Crystal Reports XML, with Rust acceleration
Keywords: crystal-reports,xml,parser,pandas,etl
Author-email: Emiliano Gandini <emiliano.gandini@protonmail.com>
License: MIT
Requires-Python: >=3.10
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Homepage, https://github.com/emiliano-go/crxml
Project-URL: Issues, https://github.com/emiliano-go/crxml/issues
Project-URL: Repository, https://github.com/emiliano-go/crxml

<p align="center">
  <a href="https://www.python.org/downloads/">
    <img src="https://img.shields.io/badge/Python-3.10%2B-3776AB?logo=python&logoColor=white&style=for-the-badge" alt="Python">
  </a>
  <a href="LICENSE">
    <img src="https://img.shields.io/badge/License-MIT-10AC84?style=for-the-badge" alt="License">
  </a>
  <a href="https://pypi.org/project/crxml/">
    <img src="https://img.shields.io/pypi/v/crxml?logo=pypi&logoColor=white&style=for-the-badge" alt="PyPI">
  </a>
</p>

# crxml

Fast streaming parser for Crystal Reports XML exports.

```python
from crxml import CrystalXMLSource, to_dataframe

df = to_dataframe(CrystalXMLSource("report.xml", row_tag="Details"))
print(df.head())
```

## Installation

**Prerequisites:** Python 3.10 or later, and [Rust](https://rustup.rs).

```bash
pip install crxml
```

The columnar and parallel engines are included by default.  For performance profiling:

```bash
pip install -e . --config-settings=--features=profile  # see benchmarks/bench_profile.py
```

## About

crxml streams through Crystal Reports XML files row by row, never loading the full document into memory. It extracts field data from nested CR field elements and yields flat dictionaries. A built-in pipeline lets you rename, cast, filter, and drop fields with pipe operators.

The parallel engine parses 100 MB in about 0.2 seconds and 533 MB in 1.1 seconds (472 MB/s) on a laptop CPU. Row iteration runs at about 2.3 seconds (stream engine, 100 MB).

Fusable pipeline stages (rename, cast, drop, filter by predicate) are compiled into the columnar BuildPlan and run in Rust during parsing, avoiding the dict round-trip. Non-fusable stages (lambdas, custom predicates) apply after Arrow conversion.

This library is conceptually based on [carlosplanchon/xmlstreamer](https://github.com/carlosplanchon/xmlstreamer).

## Quick start

```python
from crxml import CrystalXMLSource, to_dataframe

source = CrystalXMLSource("report.xml", row_tag="Details")

# Row iteration
for row in source:
    print(row)

# DataFrame (routes to parallel engine automatically)
df = to_dataframe(source)
```

## CrystalXMLSource

The source object is the entry point for all parsing. It accepts these parameters:

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `source` | str or Path | required | Path to the Crystal Reports XML file. |
| `row_tag` | str | `"Row"` | XML element tag that delimits each record. For Crystal Reports exports this is often `"Row"` or `"Details"`. |
| `engine` | str | `"auto"` | Engine selection: `"auto"`, `"stream"`, `"columnar"`, or `"parallel"`. |
| `threads` | int | 0 | Number of chunks for parallel parsing (0 = CPU count). |
| `memory` | int or str | None | Memory budget for bounded parsing. Accepts bytes (int) or strings like `"8GB"`. |
| `field_mapping` | dict | None | Rename fields at parse time: `{"old_name": "new_name"}`. |
| `drop_fields` | list | None | Fields to omit from output. |
| `filter` | dict | None | Rust-side filter: `{"field": "Status", "op": "==", "value": "Active"}`. |
| `field_types` | dict | None | Coerce fields at parse time: `{"Score": "int64", "Amount": "float64"}`. |
| `dictionary_columns` | list | None | Fields to dictionary-encode during columnar parse. |
| `schema` | list | None | Ordered list of fields to include (others dropped). |
| `auto_dict` | bool | False | Automatically dictionary-encode string columns. |
| `use_mmap` | bool | True | Memory-map the file instead of reading into a buffer. |
| `prefault` | bool | varies | `True` → `MADV_WILLNEED` for speed, `False` → `MADV_SEQUENTIAL` for lower RSS. Default: True for columnar/parallel, False for bounded. |
| `batch_size` | int | 1024 | Rows fetched per Rust call during batched iteration. |

```python
from crxml import CrystalXMLSource

source = CrystalXMLSource(
    "report.xml",
    row_tag="Details",
    batch_size=2048,
    field_mapping={"f1": "invoice", "f2": "amount"},
    drop_fields=["tax_rate", "internal_id"],
    field_types={"amount": "float64", "quantity": "int64"},
    dictionary_columns=["product_code"],
    memory="4GB",
)
```

### File/string-like objects

The `source` parameter accepts a file path (string or `pathlib.Path`). File-like objects with a `.name` attribute pointing to an existing file are also supported.

## Engine selection

The parser has three backend engines, plus a bounded mode activated via the `memory=` parameter.  Auto mode selects the best engine per goal.

| Engine | Description |
|--------|-------------|
| `stream` | Row-by-row XML parsing. Produces dicts directly. Always available. |
| `columnar` | Single-threaded Arrow columnar parse. Zero-copy string columns. |
| `parallel` | Multi-threaded columnar parse (file split into chunks, workers parse in parallel). |

With any engine (except `stream`), passing `memory=` (e.g. `memory="500MB"`) switches to
**bounded mode**: the file is parsed in chunks that fit within the budget, keeping peak
RSS independent of file size.

### Auto routing

When `engine="auto"` (the default), the engine is resolved per call based on the goal and file size:

**For row iteration** (`for row in source`): always uses the stream engine. Batched parsing with GIL release (phase 1) handles the iteration efficiently.

**For table output** (`source.to_arrow()`, `source.to_dataframe()`): resolves in this priority order:
1. Parallel engine (file >= 8 MB, columnar feature available, within memory budget)
2. Columnar engine (columnar feature available, within memory budget)
3. Stream fallback (small files, or columnar not available)

```python
source = CrystalXMLSource("large_report.xml", engine="auto")

for row in source:         # always stream engine (dicts)
    pass

df = source.to_dataframe() # auto resolves to parallel (large file)
```

When the columnar or parallel engine is used for table output, the dict construction overhead is eliminated entirely. Strings are written directly into Arrow buffers. This gives a 3x speedup over the stream-based dict path.

### Explicit engine

You can also pick an engine explicitly:

```python
source = CrystalXMLSource("report.xml", engine="parallel", threads=8)
table = source.to_arrow()   # uses parallel engine

source = CrystalXMLSource("report.xml", engine="parallel", memory="500MB")
df = source.to_dataframe()  # bounded mode (chunked within 500 MB RSS)
```

## Row iteration

`CrystalXMLSource` is iterable. It yields dictionaries mapping field names to string values.

```python
source = CrystalXMLSource("report.xml")

for row in source:
    print(row["invoice"], row["amount"])
```

### How it works

For the stream engine (the iter goal always uses stream), iteration goes through a `_BatchIter` wrapper. Each call to `next()` fetches a batch of rows (configurable via `batch_size`) from Rust with the GIL released, then returns rows one by one from an internal buffer. This reduces Python/Rust boundary crossings from one-per-row to one-per-batch.

When the columnar or parallel engine is explicit and you iterate, the full Arrow table is parsed first and dicts are reconstructed from it. That path is slower and exists only for compatibility. Table-oriented callers should use `to_arrow()` or `to_dataframe()` directly.

### Batched iteration

The `_iter_batches` method yields lists of dicts instead of single rows:

```python
for batch in source._iter_batches(batch_size=4096):
    for row in batch:
        print(row)
```

This is used internally by pipeline fusion and the `collect` sink.

## Table output

These methods produce Arrow tables or DataFrames. All of them use the resolved table engine (columnar/parallel when available).

### to_arrow

Returns a `pyarrow.Table` of the parsed data. Zero-copy for columnar/parallel engines. For the stream engine, dicts are collected and converted to a table.

```python
table = source.to_arrow()
print(table.num_rows, table.column_names)
```

### to_dataframe / to_pandas

`to_dataframe()` is an alias for `to_pandas()`. Both return a pandas DataFrame.

```python
df = source.to_dataframe()                     # ArrowDtype columns (zero-copy)
df = source.to_pandas(dtype_backend="numpy")   # numpy-backed strings
```

By default (`dtype_backend="pyarrow"`), string columns use `pd.ArrowDtype` for zero-copy conversion from Arrow buffers. This requires pandas 1.5 or later. Pass `dtype_backend="numpy"` to materialize strings as Python str objects.

### to_polars

```python
import polars as pl
df = source.to_polars()  # zero-copy from Arrow
```

### to_parquet

```python
source.to_parquet("output.parquet")
# Forward kwargs to pyarrow.parquet.write_table
source.to_parquet("out.parquet", compression="zstd")
```

### schema

Returns the field names from the first row:

```python
fields = source.schema()  # ["invoice", "amount", ...]
```

## Pipeline stages

Stages transform the row stream. They are chained with the pipe operator `|`.

```python
from crxml.stages import RenameFields, CastTypes, DropFields, FilterRows

pipeline = (
    CrystalXMLSource("report.xml")
    | RenameFields({"f1": "invoice", "f2": "amount"})
    | CastTypes({"amount": float})
    | DropFields("tax_rate")
    | FilterRows(lambda r: r["amount"] > 100)
)

for row in pipeline:
    print(row)
```

### RenameFields

Renames dictionary keys.

```python
RenameFields({"old_name": "new_name", ...})
```

Fusable: yes. Compiled into the columnar BuildPlan field_mapping.

### CastTypes

Casts field values to target types.

```python
CastTypes({"amount": float, "quantity": int})
```

Fusable when the target type is one of: `int`, `float`, `bool`, `str`. Non-standard callables (lambdas, custom functions) force the stage to stay on the dict path.

```python
CastTypes({"amount": float})            # fusable
CastTypes({"amount": lambda v: v * 2})  # not fusable, stays on dict path
```

### DropFields

Removes fields from each row.

```python
DropFields(["tax_rate", "internal_id"])
```

Fusable: yes. Compiled into the BuildPlan drop_fields.

### FilterRows

Filters rows by a predicate. Supports three forms:

```python
# Lambda predicate (not fusable)
FilterRows(lambda r: r["amount"] > 100)

# Constant field comparison (fusable)
FilterRows(field="Status", op="==", value="Active")
FilterRows(field="Quantity", op="!=", value="0")

# Column-to-column comparison (fusable)
FilterRows(field_a="Price", op=">", field_b="Cost")
FilterRows(field_a="Name", op="eq", field_b="ExpectedName")
```

Fusable when using the keyword-argument form (constant comparison or column-to-column comparison). Lambda predicates always apply after Arrow conversion.

### Custom stages

Any callable that accepts and returns an iterable of dicts is a valid stage:

```python
def add_metadata(stream):
    for row in stream:
        row["source"] = "crystal_reports"
        yield row

pipeline = CrystalXMLSource("report.xml") | add_metadata
```

Custom stages are never fused. They always run on dicts after Arrow conversion.

## Pipeline fusion

When a pipeline contains fusible stages, the library tries to push them into the columnar BuildPlan so they execute in Rust during parsing. This avoids the cost of constructing dicts from the Arrow table and then immediately renaming/casting/dropping fields.

### How fusion works

1. `Pipeline.__iter__` calls `fused_iter(source, stages)`.
2. `_try_columnar_fusion` inspects each stage for a `_plan_kwargs` method.
3. Stages that return kwargs are merged into the columnar BuildPlan.
4. The plan is passed to `source._read_arrow(plan_overrides=...)`.
5. Non-fusable stages apply to the resulting dict stream.

### What fuses

| Stage | Fusable kwargs | Condition |
|-------|---------------|-----------|
| `RenameFields` | `field_mapping` | Always |
| `CastTypes` | `field_types` | Only for int, float, bool, str targets |
| `DropFields` | `drop_fields` | Always |
| `FilterRows` | `filter` | Only for keyword-argument form |

### What does not fuse

- Lambda predicates in `FilterRows`
- Custom stage functions
- `CastTypes` with non-standard callables
- Stages on an iterable that is not a `CrystalXMLSource` (no `_read_arrow` method)

### Dict-path fallback

When no stage can be fused, the library falls back to the dict path. Fusible stages are applied inline (one function call per row) using batched iteration (`_iter_batches`). Non-fusable stages wrap the stream as callables. This path still benefits from batched Rust parsing with GIL release.

```python
# All fusable: runs entirely in Rust (columnar plan)
pipeline = source | RenameFields({"a": "b"}) | DropFields(["x"])

# Mixed: columnar parse + lambda on dicts
pipeline = source | RenameFields({"a": "b"}) | FilterRows(lambda r: r["b"] > 0)

# No fusible: dict path with batched iteration
pipeline = source | FilterRows(lambda r: r["amount"] > 0)
```

### The _arrow_iter compatibility helper

When columnar or parallel engines produce row iterators, the `_arrow_iter` function reconstructs dicts from the Arrow table. This is a compatibility path for callers that request row iteration on a columnar-able source. Table-oriented callers (to_arrow, to_dataframe) bypass this entirely.

## Sinks

Sinks consume an iterable of dicts and produce a concrete result.

```python
from crxml import to_dataframe, to_csv, collect
```

### to_dataframe

Converts an iterable of dicts (source or pipeline) to a pandas DataFrame.

```python
df = to_dataframe(source)
df = to_dataframe(pipeline)
```

DataFrames use `pd.ArrowDtype` for zero-copy string columns (pandas 1.5+).

### to_csv

Writes rows to a CSV file.

```python
to_csv(pipeline, "output.csv", delimiter=",")
```

### collect

Collects all rows into a list.

```python
rows = collect(pipeline)
```

Uses batched iteration (`_iter_batches`) when available for efficiency.

## Parallel mode

Pipelines can be distributed across worker processes.

```python
pipeline = (
    CrystalXMLSource("report.xml")
    | RenameFields({"f1": "invoice"})
)

pipeline = pipeline.parallel(workers=4, batch_size=1000)
for row in pipeline:
    print(row)
```

This splits the file into chunks and processes each chunk in a subprocess. Stages must be picklable (standard stages are). Lambda predicates are not picklable and will raise an error.

## Feature flags

The `columnar` and `mmap` features are enabled by default.  One additional flag is available:

| Feature | Build command | What it enables |
|---------|--------------|-----------------|
| `profile` | `pip install -e . --config-settings=--features=profile` | Instant-based performance counters in the stream and parallel engines. Use `CrxmlReader.get_profile_data()`, `get_par_profile()`, or `benchmarks/bench_profile.py`. |

## Architecture overview

### Two parse paths

crxml has two fundamentally different parse paths:

**Stream path** (always available): The Rust `CrxmlReader` walks the XML with `quick-xml`, extracting field names and values. Batching (via `next_batch`) parses groups of rows with the GIL released, then builds Python dicts in one shot. This is the path used for row iteration.

**Columnar path** (requires `columnar` feature): The Rust code writes parsed field data directly into Arrow buffers. There are no intermediate Python dicts. The columnar engine also accepts a `BuildPlan` with field mapping, field types, drop list, dictionary columns, and filter predicates, allowing fusible pipeline stages to execute entirely in Rust.

### Goal-aware routing

The `CrystalXMLSource` routes calls based on their goal:

- `__iter__` and `_iter_batches` use the resolution for goal `"iter"` (always stream in auto mode).
- `to_arrow`, `to_dataframe`, `to_pandas`, `to_polars`, and `to_parquet` use the resolution for goal `"table"` (columnar or parallel when available).

This means iterating rows and building a DataFrame from the same source use different internal engines, each optimal for its goal.

Python layer manages the routing, caching, and dict conversion. The Rust layer handles XML parsing, field extraction, Arrow buffer construction, and parallel chunking.

## Benchmarks

Full details in [`docs/performance.md`](docs/performance.md).  All numbers from a single machine:

| Component | Detail |
|---|---|
| **CPU** | 13th Gen Intel Core i5-1335U (10 cores: 2 P + 8 E, 12 threads) |
| **RAM** | 15 GiB LPDDR5 |
| **OS** | Arch Linux, kernel 7.0.9 |
| **Build** | release, LTO, mimalloc, features `columnar` + `mmap` |

All runs warm-cache, best of 3.  Synthetic files are **directional only** (the 533 MB
real Crystal export is the ground truth).

### Input files

| File | Size | Rows | Fields | Origin |
|---|---|---|---|---|
| Synthetic | 100 MB | 90,384 | 10 | `benchmarks/benchmarks.py` |
| Real export | 533 MB | 465,136 | 11 | Crystal Reports |

### End-to-end `to_dataframe()` (user's actual goal)

| Engine | 100 MB (synthetic) | 533 MB (real) |
|---|---|---|
| **Stream** | 2.27 s / 44 MB/s / 40k r/s | 12.8 s / 42 MB/s / 36k r/s |
| **Parallel** (8 workers) | **213 ms** / 469 MB/s / 424k r/s | **1.13 s** / 472 MB/s / 412k r/s |
| **Parallel + auto-dict** | 312 ms | 2.0 s / 267 MB/s |

Parallel throughput improves with file size (split-scan amortises).  At 533 MB it
sustains **472 MB/s**, ~11× faster than the stream engine.

### Parallel-path breakdown (533 MB real)

| Phase | Time | % of wall |
|---|---|---|
| Split-scan (serial) | 257 ms | 23% |
| Off-GIL parse (8 threads) | 781 ms | 69% |
| On-GIL assembly | 25 ms | 2% |
| Coverage | - | 94% |

Parse is the ceiling (69%).  GIL assembly is a dead lever (2%).

### Pipeline fusion

Fusable stages (RenameFields, DropFields, CastTypes with standard types) are compiled
into the columnar BuildPlan and run in Rust during parsing.  Non-fusable stages
(lambdas, custom predicates) apply to dicts after Arrow conversion.  All-fusable
pipelines are ~3× faster than the equivalent dict-path pipeline.

### Memory (parallel engine, 533 MB)

| Metric | Value |
|---|---|
| Peak RSS | 534 MB (= file size; mmap) |
| Workload buffers | ~21 MB (columnar + Arrow) |
| Total allocations | 7,725 across 465k rows |

The mmap path maps the file into virtual memory; the OS pages it in on demand.  The
stream engine peaks at ~1.07 GB (accumulating Python dicts).

### Recommendations

| Goal | Engine | Notes |
|------|--------|-------|
| Row iteration (`for row in source`) | `"auto"` (stream) | Best for row-by-row processing |
| Arrow / DataFrame | `"auto"` (parallel) | ~11× over stream; mmap + off-GIL parse |
| Pipeline with fusable stages | `"auto"` (parallel) | Stages fused into BuildPlan |
| Minimise peak memory | `engine="parallel"` + `memory=` | RSS tracks budget, not file size |



## Documentation

Full documentation is available at the [project site](https://crxml.emiliano-go.com/), covering installation, usage, stages, custom stages, architecture, performance, FastAPI integration, and the Rust core.

## License

MIT

