data/README.md

LuminaMoon Data Engine v1.1.0

LuminaMoon Data (luminamoon.data) provides text cleaning, SIMD normalization, heuristic quality filtering, exact and fuzzy deduplication, and crash-resilient checkpointing for Python and Rust data pipelines, including PyTorch, Hugging Face datasets, Ray, and WebDataset.

---

1. Modular Architecture & Responsibilities

LuminaMoon Data focuses on in-memory text processing and deterministic state tracking, integrating smoothly with existing dataset readers:

flowchart LR
    subgraph External ["External Data Sources"]
        HF["Hugging Face datasets.load_dataset(streaming=True)"]
        Torch["PyTorch torch.utils.data.DataLoader"]
        Custom["Custom Python Generator / Iterables"]
    end

    subgraph Data_Engine ["LuminaMoon Data Engine (Rust Core)"]
        direction TB
        Cleaner["1. TextCleaner\n(AVX-512/AVX2/NEON SIMD CRLF & JSON path extraction)"]
        Quality["2. QualityFilter\n(Shannon entropy, repetition loop, symbol ratio)"]
        Dedup["3. StreamDeduplicator & MinHash\n(XXH3-64, SwissTable WAL, MinHash fuzzy LSH)"]
        Masker["4. PIIMasker\n(Email, IP, Bearer/API Key, Card pattern sanitizer)"]
        Tracker["5. StateTracker\n(Dual-Slot 128B Ping-Pong Slot A/B CRC32)"]
        Glue["6. Collate & Pipeline Glue\n(PyTorch collate_fn, ResumablePipeline)"]
    end

    HF --> Cleaner
    Torch --> Cleaner
    Custom --> Cleaner
    Cleaner --> Quality
    Quality --> Dedup
    Dedup --> Masker
    Masker --> Tracker
    Tracker --> Glue
    Glue --> Consumer["GPU Model Training Step"]

---

2. Component Breakdown

Module	Core Responsibility	Underlying Implementation
luminamoon.data.cleaner	Whitespace trimming, SIMD newline normalization (\r\n $\rightarrow$ \n), byte length boundaries, and JSON path extraction.	simd_cpu.rs, postprocess.rs (Rayon parallel multi-threading, AVX-512BW, AVX2, NEON)
luminamoon.data.quality	Shannon Character Entropy, loop repetition detector, and symbol-to-word ratio filtering.	quality.rs (Parallel Rayon heuristic quality filter)
luminamoon.data.dedup	Stateful 64-bit exact hashing and deduplication with write-ahead log (.dedup) persistence.	dedup.rs (xxh3_64 + IdentityBuildHasher SwissTable)
luminamoon.data.minhash	Fuzzy near-duplicate detection via MinHash shingling and Jaccard similarity estimation.	minhash.rs (Permuted XXH3-64 MinHash signatures)
luminamoon.data.masker	Pattern-based PII masking for emails, IPs, API keys/tokens, and payment cards.	masker.py (Compiled multi-pattern regex sanitizer)
luminamoon.data.checkpoint	Dual-slot atomic coordinate checkpointing (shard_id, row_offset, byte_offset, global_idx).	metadata.rs, coord.rs (128-byte Dual-Slot Ping-Pong Slot A/B CRC32)
luminamoon.data.string_tensor	Columnar string representation with CPU inlined StringView and GPU contiguous offset/payload buffers.	string_tensor.rs (16-byte Umbra/DuckDB StringView, SIMD prefix filter, contiguous u32 offsets + u8 payload)
luminamoon.data.adapters	PyTorch collate_fn adapter and batch ResumablePipeline.	adapters.py (Dynamic token padding & tensor generation)
luminamoon.data.serialization	Conversion of Python dictionaries to Parquet, JSONL, or Arrow IPC bytes.	serialize.rs (Arrow RecordBatch & Parquet ArrowWriter with ZSTD)

---

3. Quickstart & Usage

3.1. Text Cleaning with SIMD & Rayon

from luminamoon.data import TextCleaner

cleaner = TextCleaner(
    trim_whitespace=True,
    normalize_newlines=True,      # AVX-512 / AVX2 / NEON SIMD
    min_byte_len=10,
    max_byte_len=2048,
    truncate_over_max=True,
    json_extract_key="data.text",  # Extracts dotted path in Rust
)

cleaned = cleaner.clean_batch(["  sample 1\r\n", "sample 2 "])

3.2. Quality Filtering (Shannon Entropy & Repetition Detector)

from luminamoon.data import QualityFilter

qf = QualityFilter(
    min_entropy=2.5,             # Filters low-entropy noise
    max_repetition_ratio=0.35,   # Filters repetitive text loops
    max_symbol_ratio=0.30,       # Filters symbol-heavy dumps
)

if qf.is_high_quality("Sample prose text for language model training."):
    ...

3.3. Exact & Fuzzy Near-Deduplication

from luminamoon.data import StreamDeduplicator, MinHashDeduplicator

# Exact dedup with WAL
dedup = StreamDeduplicator(wal_path="./cache/dataset.dedup")
unique_batch = dedup.filter_batch(cleaned)
dedup.commit()

# Fuzzy near-dedup (> 85% similarity)
fuzzy = MinHashDeduplicator(threshold=0.85)
if fuzzy.is_unique("A document with slight punctuation differences."):
    ...

3.4. PII Sanitization

from luminamoon.data import PIIMasker

masker = PIIMasker(mask_emails=True, mask_ips=True, mask_api_keys=True)
safe_text = masker.mask("Contact support@lumi.ai with key Bearer sk-1234567890abcdef1234567890abcdef")
# Result: "Contact [EMAIL] with key [API_KEY]"

3.5. Dual-Layout StringTensor (CPU StringView & GPU Buffers)

from luminamoon.data import StringTensor

# 1. CPU Layout (Pattern 2 - Umbra / DuckDB / Arrow 2.0 StringView)
# Strings <= 12 bytes are stored inline in a 16-byte struct
st = StringTensor(["short", "this is a longer sample exceeding 12 bytes", "pref_sample"])
print(st.is_inlined_mask())  # [True, False, True]

# SIMD prefix and length filtering
filtered = st.filter_by_prefix("pref_")
hashes = st.hash_all()  # Multi-threaded XXH3-64 hashing in Rust

# 2. GPU Layout (Pattern 1 - Apache Arrow & TensorFlow Contiguous Offsets + Data Buffer)
offsets, payload = st.to_gpu_buffers()      # Raw uint32 offsets and uint8 payload
off_tensor, pay_tensor = st.to_torch_gpu()  # PyTorch tensors
off_arr, pay_arr = st.to_numpy_gpu()        # NumPy buffers
arrow_array = st.to_arrow()                 # Apache Arrow 2.0 StringArray
polars_series = st.to_polars()              # Polars Series

3.6. Fused Single-Pass Pipeline

from luminamoon.data import FusedPipeline, StreamDeduplicator

pipeline = FusedPipeline(
    trim_whitespace=True,
    normalize_newlines=True,
    min_chars=15,
    min_entropy=2.0,
    deduplicator=StreamDeduplicator(),
    commit_dedup=True,
)

# Process a batch in a single multi-threaded Rayon pass
string_tensor = pipeline(raw_records)

3.7. PyTorch Collate & Async Prefetch Pipeline

from torch.utils.data import DataLoader
from luminamoon.data import create_resumable_collate, ResumablePipeline, FusedPipeline

# 1. As PyTorch collate_fn
collate_fn = create_resumable_collate(
    fused_pipeline=FusedPipeline(),
    return_string_tensor=True,
    tokenizer=my_tokenizer,
    max_length=512,
)
loader = DataLoader(raw_dataset, batch_size=64, collate_fn=collate_fn)

# 2. Or as batch ResumablePipeline with background prefetcher
resumable_pipe = ResumablePipeline(
    source=raw_dataset,
    fused_pipeline=FusedPipeline(),
    batch_size=64,
    prefetch_batches=4,  # Background prefetch worker for processed batches
    return_string_tensor=True,
)
for batch in resumable_pipe:
    ...

---

4. Verification & Testing

Run Rust tests:

cargo test -p luminamoon-dataloader

Run Python modular test suite:

python -B tests/test_data_modular.py

---

5. Version Inspection

Query the current data engine module version:

import luminamoon.data as lmd

print(lmd.get_version())  # Active package release version string

Note: LuminaMoon Data operates on a single unified native SIMD engine. Package releases follow luminamoon.__version__.