⚡ Process Sharding & Data Partitioning

Veloctra implements a 5-layer sharding and partitioning architecture designed to guarantee non-blocking high throughput, sub-second backpressure, and zero data loss.

Layer 1: Source Extraction Sharding
Cursor Slicing • Key Ranges • Watermarks
Relational SQL Sources
Postgres • MySQL • SQLite
  • Server-side streaming cursors without table locks
  • Incremental indexed watermark partitioning (id >= :start AND id < :end)
NoSQL & Object Stores
MongoDB • S3 • CSV / Zip
  • Token-aware range slices & uncompressed stream decompression
  • Zero-copy batch slice ingestion
Streaming Batches >> In-Flight PyArrow Buffers
Layer 2: In-Flight Chunk Sharding & MemoryGuard
Adaptive Chunking (10k → 50 → 1) • 75% RAM Ceiling
PyArrow RecordBatches
SIMD Vector Engine
  • High-efficiency zero-copy columnar buffers
  • Continuous evaluation of average byte-size per row
MemoryGuard Governor
Resource Monitoring
  • Massive blobs (≥5MB) auto-throttled to 1 row/chunk
  • >75% RAM triggers micro-sleep backpressure & GC window
Dispatch to Concurrency Workers
Layer 3: Pipeline Concurrency & Process Sharding
Async Workers • Multi-Job Concurrency
Worker 1 (Job A)
Async Stream Thread
  • Isolated state & metrics
Worker 2 (Job B)
Async Stream Thread
  • Non-blocking I/O loop
Worker N (Job N)
Async Stream Thread
  • Tenant-level isolation
Batch Execution & Error Slicing
Layer 4: Fault-Tolerant Sub-Chunk DLQ Sharding
Row-by-Row Isolation • Zero Data Loss
Valid Records Sub-Batch
Immediate Forwarding
  • 99.99% healthy rows proceed to sink without delay
  • Zero pipeline downtime on bad payload records
Corrupted Poison Pill Record
Isolated to DLQ Store
  • Single row sliced into Dead Letter Queue with error stack trace
  • Full audit trail & replay capability in Studio UI
Partitioned Storage Sinks
Layer 5: Destination Sink & File Partitioning
FilePartitioner (100k Rows / 100MB) • Bulk Upserts
FilePartitioner Sinks
part_00001.parquet • Snappy / ZSTD
  • Automatic rotation upon 100k rows or 100MB threshold
  • Direct streaming upload to AWS S3 & Google Cloud Storage
Database Bulk Engines
PostgreSQL • MongoDB Bulk Write
  • High-performance binary copy streaming
  • Unordered bulk write operations with connection pooling

1. Mathematical Guarantees: Gapless & Overlap-Free Process Sharding

A critical challenge in distributed ETL and streaming engines is ensuring that concurrent shards process all data without duplicates and without dropping records. Veloctra enforces strict mathematical partitioning invariants across all extractors and workers:

Zero-Overlap Invariant

Each shard $i$ is assigned a half-open interval $[Start_i, End_i)$ defined by the strict predicate:

WHERE id >= :start_id AND id < :end_id

Since $End_i = Start_{i+1}$, any boundary key $k = End_i$ is excluded from Shard $i$ and evaluated exclusively in Shard $i+1$, mathematically preventing duplicate processing.

Zero-Drop Completeness

The total keyspace domain $\mathcal{D} = [\text{MinID}, \text{MaxID}]$ is divided such that:

\bigcup_{i=1}^{N} [Start_i, End_i) \equiv [\text{MinID}, \text{MaxID}]

Because the intervals are contiguous with zero internal gaps, every record in the table belongs to exactly one shard interval.

2. Delta Processing & High-Watermark CDC Synchronization

Veloctra includes native Delta Processing to capture only new or modified rows since the last successful sync, avoiding wasteful full-table re-scans:

Delta Execution Lifecycle

  1. Watermark Discovery: When a pipeline starts, PipelineOrchestrator queries the StateStore for the latest committed watermark_value from previous runs of the same pipeline_id.
  2. Dynamic Query Rewriting: The extractor dynamically binds the watermark predicate to the SQL or NoSQL query: 
    -- Automatically injected by Veloctra SQLConnector
SELECT * FROM claims 
WHERE updated_at > '2026-01-03T10:00:00' 
ORDER BY updated_at ASC;
  3. Vectorised In-Flight High-Watermark Computation: As PyArrow columnar batches pass through the pipeline, the orchestrator tracks the maximum value in the watermark column: $\text{HighWatermark} = \max(\text{batch}[\text{watermark\_col}])$.
  4. Atomic Checkpoint Commit: The state store saves the new high-watermark alongside the chunk index, guaranteeing deterministic resumption upon any worker restart.

Sample Pipeline Configuration with Delta Processing

pipeline_id: pg_to_mongo_delta_claims
tenant_id: healthcare_enterprise

sources:
  - name: pg_claims_source
    type: database
    connection_string: "postgresql+asyncpg://user:secret@localhost:5432/claims_db"
    query: "SELECT id, beneficiary_id, amount, updated_at FROM raw_claims"
    chunk_size: 5000
    delta:
      enabled: true
      watermark_column: "updated_at"    # Supports timestamps, integer sequences, or offsets
      watermark_type: "timestamp"      # timestamp | integer | string
      initial_watermark: "1970-01-01T00:00:00Z"

destinations:
  - name: mongo_claims_sink
    type: nosql
    db_type: mongodb
    connection_string: "mongodb://localhost:27017"
    database: analytics_dw
    collection: claims
    upsert_key: "id"                   # Guarantees idempotent update-in-place for modified rows

3. Zero-Column Change Data Capture (CDC): Tables Without Timestamps

In legacy or third-party databases, tables frequently lack updated_at timestamps or auto-increment IDs, yet experience in-place updates and deletions. Veloctra solves this with the Universal Checksum-Diff CDC Engine:

Deterministic Row Checksums

The engine computes SHA-256 cryptographic hashes over non-key payload cells: $\mathcal{H}(r) = \text{SHA256}(\text{JSON}(r_{\text{non-keys}}))$.

In-Place Update & Delete Detection

Diffing against state snapshots detects:

  • INSERT: Key not in previous state
  • UPDATE: Key exists with modified checksum
  • DELETE: Key in state but absent in table

Replication Oplog Streaming

For MongoDB, PostgreSQL, and MySQL, native Change Streams subscribe to oplog/WAL logs with resume tokens.

4. Idempotent Merge Strategies (Exactly-Once Semantics)

Even in the event of unexpected network disconnects or process restarts, Veloctra destinations guarantee exactly-once data consistency via deterministic upsert engines:

Target Engine Merge Strategy Behavior Under Replay
PostgreSQL INSERT INTO ... ON CONFLICT (keys) DO UPDATE SET ... Updated records overwrite old rows; duplicate insertions are cleanly merged.
MongoDB ReplaceOne(filter={"upsert_key": val}, replacement, upsert=True) Document matching key is updated in place atomically; new keys are inserted.
SQLite INSERT OR REPLACE INTO table (cols) VALUES (...) Existing primary key row replaced atomically; zero duplicate accumulation.
Parquet / S3 Deterministic Partitions: part_{job_id}_{chunk_idx}.parquet Checkpointed chunk files overwritten atomically on retry; no file duplication.

4. Adaptive In-Flight Memory Sharding (MemoryGuard)

The MemoryGuard continuously monitors process RAM, OS CPU, and individual record sizes to ensure system stability:

Massive Records (≥ 5 MB/row)

Chunk size is immediately reduced to 1 record per chunk to prevent memory spikes.

Large Records (≥ 100 KB/row)

Chunk size is throttled down to 50 records per chunk.

Resource Ceiling (> 75% RAM/CPU)

Chunk sizes are halved and micro-sleep backpressure is applied, leaving ≥ 25% headroom for Python Garbage Collection.

5. Fault-Isolation Sub-Chunking (DLQ Routing)

If a vectorised PyArrow batch of 10,000 rows encounters corrupt data, the orchestrator sub-shards the batch row-by-row (batch.slice(i, 1)). The poison-pill record is routed to the MongoDB/SQLite Dead Letter Queue (dlq) with full stack trace, while all valid rows proceed immediately.

6. Intelligent Migration Sizing & KEDA Elastic Autoscaling

When starting large-scale migrations or processing high-volume delta backlogs, Veloctra features an integrated Migration Sizing Engine that interacts with KEDA (Kubernetes Event-Driven Autoscaling) to automatically scale worker pod counts up during heavy workloads and scale down to baseline when finished.

1. Automated Volume Discovery

Scans SQL table catalogs (pg_class.reltuples, SELECT COUNT(*)), MongoDB collection metadata (count_documents), and file headers to compute exact pending volume.

2. Mathematical Target Scaling

Calculates optimal replicas based on workload density: $$\text{TargetReplicas} = \operatorname{clamp}\left(\left\lceil \frac{\text{TotalPendingRows}}{\text{RowsPerWorker}} \right\rceil, \text{MinReplicas}, \text{MaxReplicas}\right)$$

3. Two-Tier Scale-Up & Scale-Down

MemoryGuard handles sub-millisecond RAM safety inside each pod, while KEDA horizontally provisions new pods (1 → 16) and smoothly drains them upon completion.

KEDA Prometheus Autoscaling Architecture

Prometheus Metric Type Description KEDA Scaler Target
veloctra_migration_workload_demand_replicas Gauge Target pod replica count recommended by Sizing Engine threshold: 1 (1 pod per demand unit)
veloctra_migration_pending_rows Gauge Total pending rows across executing migration pipelines threshold: 100000 (1 pod per 100k rows)
veloctra_migration_total_shards Gauge Total active shards available for worker allocation threshold: 1 (1 pod per shard in ScaledJob)
veloctra_migration_active_jobs Gauge Active pipeline execution count in cluster Cluster active monitoring

Production KEDA ScaledObject Configuration

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: veloctra-engine-autoscaler
  namespace: veloctra
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: veloctra-engine
  pollingInterval: 15       # Fast metric scrape interval
  cooldownPeriod: 300       # 5-minute stabilization cooldown
  minReplicaCount: 1        # Baseline pods (0 for serverless)
  maxReplicaCount: 16       # Elastic peak capacity
  triggers:
    - type: prometheus
      metadata:
        serverAddress: http://prometheus-k8s.monitoring.svc:9090
        metricName: veloctra_migration_workload_demand_replicas
        query: sum(veloctra_migration_workload_demand_replicas)
        threshold: "1"