Sink Operations & Batch Processing
Relevant source files
The following files were used as context for generating this wiki page:
Introduction
Sink Operations and Batch Processing within the flext-target-oracle-wms project facilitate the high-performance ingestion of Singer-compliant data into Oracle WMS Cloud. This system acts as the terminal point of a data pipeline, responsible for receiving stream records, transforming them into Oracle-compatible formats, and managing the persistence of data in optimized batches.
The architecture leverages a service-runtime pattern where high-level Singer sinks delegate record processing to a dedicated Oracle WMS runtime. This ensures that data is not only validated against Singer schemas but also normalized—such as uppercasing table and field names—to meet the specific requirements of the Oracle WMS environment.
Sources: src/flext_target_oracle_wms/api.py:33-46, src/flext_target_oracle_wms/_utilities/service_runtime.py:22-60, examples/README.md:99-115
Core Components & Architecture
The system is organized into a hierarchical structure involving service facades, adapter sinks, and specialized managers.
WMS Service Facade
The FlextTargetOracleWmsService serves as the primary orchestrator. It inherits from FlextMeltanoTargetServiceBase and overrides the create_sink method to instantiate specialized sinks for each data stream.
Sources: src/flext_target_oracle_wms/api.py:25-46
Runtime Target and Sink Adapters
The processing logic is split between a SingerSinkBase (the Sink) and the Target (the runtime).
- Singer Sink Adapter: A thin layer that captures incoming records and passes them to the internal WMS runtime.
- Runtime Target: Manages the lifecycle of the ingestion process, including initialization, record handling via the
StreamProcessor, and state management.
The diagram above illustrates the delegation from the service orchestrator down to the specialized management components.
Sources: src/flext_target_oracle_wms/_utilities/service_runtime.py:22-55, src/flext_target_oracle_wms/_utilities/client.py:97-124
Stream and Table Management
Before records can be processed, the system must register the stream metadata. The WMSTableManager handles the mapping between Singer stream names and Oracle database tables.
Key Logic:
- Normalization: Stream names are converted to uppercase to serve as table names (e.g., a stream "orders" becomes table "ORDERS").
- Catalog Registration: The
CatalogManagerstores in-memory metadata for each stream usingSingerCatalogEntrymodels. - Table Lookup: During record processing, the
StreamProcessorqueries theWMSTableManagerto identify the correct destination table.
| Component | Responsibility | Source File |
|---|---|---|
CatalogManager | Stores stream metadata and key properties. | client.py:29-57 |
WMSTableManager | Maps stream names to uppercase Oracle table names. | helpers.py:126-141 |
WMSSchemaMapper | Builds normalized catalog entries for table creation. | helpers.py:102-124 |
Sources: src/flext_target_oracle_wms/_utilities/client.py:29-57, src/flext_target_oracle_wms/_utilities/helpers.py:126-141, tests/unit/test_sinks.py:75-84
Data Transformation Pipeline
The WMSDataTransformer is responsible for converting incoming Singer records into a structure suitable for Oracle WMS. This involves type conversion and key normalization.
Transformation Steps:
- Type Resolution: Identifies the Singer type (e.g.,
integer,number,object) from the schema properties. - Value Conversion: The
WMSTypeConverterhandles scalars and complex types. Objects and arrays are serialized into JSON strings. - Key Uppercasing: All record keys are transformed to uppercase to match Oracle field conventions.
This sequence demonstrates the internal transformation loop during record ingestion.
Sources: src/flext_target_oracle_wms/_utilities/helpers.py:65-100, tests/unit/test_wms_patterns.py:84-93
Batch Processing and Performance
Batch processing is a primary feature for high-volume data ingestion. The system supports various performance-tuned configurations.
Performance Benchmarks
According to project documentation, batch processing provides significant throughput improvements over single-record processing:
- Single Record: ~100-500 records/second.
- Batch Mode (1K): ~2,000-5,000 records/second.
- Batch Mode (5K): ~8,000-15,000 records/second.
- Concurrent Streams: ~20,000+ records/second.
Batch Configuration Options
Users can tune the ingestion behavior through specific settings:
| Option | Type | Description |
|---|---|---|
batch_size | Integer | Number of records per batch (default varies). |
bulk_insert_mode | Boolean | Enables optimized bulk insertion patterns. |
parallel_processing | Boolean | Allows concurrent stream processing. |
memory_optimization | Boolean | Reduces memory footprint for large batches. |
Sources: examples/README.md:104-115, 148-155, src/flext_target_oracle_wms/_utilities/service_runtime.py:57-60
Implementation Example: Sink Record Processing
The following code demonstrates how the Sink adapter delegates a record to the runtime target for processing.
@override
def process_record(
self,
record: t.JsonMapping,
context: t.JsonMapping,
) -> None:
"""Process a single record through the Oracle WMS runtime."""
_ = context
result = self._runtime_target.handle_record_message(
m.Meltano.SingerRecordMessage.model_validate({
"type": "RECORD",
"stream": self.stream_name,
"record": u.normalize_to_json_mapping(record),
}),
)
if result.failure:
msg = result.error or "Oracle WMS runtime rejected the record"
raise RuntimeError(msg)
Sources: src/flext_target_oracle_wms/_utilities/service_runtime.py:40-55
Conclusion
Sink Operations and Batch Processing in the Oracle WMS Target provide a robust mechanism for enterprise-grade data loading. By separating the Singer protocol handling (Sinks) from the Oracle-specific transformation logic (Target Runtime), the system achieves high throughput while maintaining strict data integrity and normalization standards. The integration of batching and parallel processing allows the target to scale effectively for mission-critical Oracle WMS Cloud implementations.