Data Loading & Processing

Relevant source files

The following files were used as context for generating this wiki page:

• src/flext_target_oracle/_utilities/loader.py
• src/flext_target_oracle/_utilities/client.py
• src/flext_target_oracle/_utilities/errors.py
• examples/03_production_setup.py
• tests/integration/test_oracle.py
• examples/01_basic_usage.py

Data Loading & Processing in flext-target-oracle is the core mechanism for ingesting Singer-formatted data streams and persisting them into an Oracle database. This system handles the translation of Singer SCHEMA and RECORD messages into Oracle DDL (Data Definition Language) and DML (Data Manipulation Language) operations, ensuring type safety and efficient bulk ingestion.

The architecture is built upon a separation of concerns between the FlextTargetOracle client, which orchestrates the message flow, and the FlextTargetOracleLoader, which executes database-specific logic via the flext-db-oracle API. This approach adheres to the "Railway-Oriented Programming" pattern for error handling and the "Source of Truth" pattern for database interactions.
Sources: src/flext_target_oracle/_utilities/loader.py:15-28, src/flext_target_oracle/_utilities/client.py:14-25, CONTRIBUTING.md:144-156

Architecture Overview

The data processing pipeline follows a synchronous flow where Singer messages are parsed, validated, and buffered before being flushed to the database in batches.

Data Flow Diagram

The following diagram illustrates how Singer messages move from the input stream to the Oracle database.

The diagram shows the logic routing within FlextTargetOracle and FlextTargetOracleLoader.
Sources: src/flext_target_oracle/_utilities/client.py:126-141, src/flext_target_oracle/_utilities/loader.py:504-541

Key Components

FlextTargetOracle (Client)

The primary interface for message routing. It manages the state of the ingestion process and coordinates between the incoming stream and the loader.

• process_singer_message(): Dispatches messages to specific handlers based on their type.
• _handle_schema(): Triggers table creation/verification.
• _handle_record(): Passes record data to the buffer.
  Sources: src/flext_target_oracle/_utilities/client.py:126-141, src/flext_target_oracle/_utilities/client.py:202-225

FlextTargetOracleLoader

Responsible for the actual transformation and loading of data.

• ensure_table_exists(): Uses flext-db-oracle to verify or create the target table structure.
• load_record(): Implements batching logic, adding records to an internal buffer until the batch_size threshold is met.
• _flush_batch(): The core DML execution method that transforms buffered records into Oracle-specific parameters and executes bulk inserts.
  Sources: src/flext_target_oracle/_utilities/loader.py:348-358, src/flext_target_oracle/_utilities/loader.py:504-515, src/flext_target_oracle/_utilities/loader.py:539-550

Schema Mapping & Table Management

The system dynamically creates Oracle tables based on Singer schema definitions. It utilizes a type mapping service to convert JSON types (string, integer, number, object, array) into Oracle-native types.

Column Generation Logic

1. Source Mapping: Maps Singer property names to Oracle column names (supports custom mapping overrides).
2. Type Translation: Resolves data types (e.g., Singer string with date-time format becomes Oracle TIMESTAMP).
3. Audit Columns: Automatically appends _SDC_EXTRACTED_AT and _SDC_LOADED_AT metadata columns.
4. Ordering: Supports alphabetical ordering or custom priority rules for primary keys, regular columns, and audit columns.

Feature	Description	File Path
JSON Storage	Stores nested objects/arrays as CLOB in a json_column_name if storage_mode is json or hybrid.	loader.py:141-150
Custom Indexes	Allows definition of unique or non-unique indexes via custom_indexes configuration.	loader.py:421-438
Truncate Mode	Optionally clears the table before loading if truncate_before_load is enabled.	loader.py:410-419

Sources: src/flext_target_oracle/_utilities/loader.py:101-140, src/flext_target_oracle/_utilities/loader.py:209-245

Record Processing Logic

Records are processed using an internal buffering mechanism to optimize database performance through bulk operations.

Sequence Diagram: Batch Processing

This diagram shows the internal batching logic before records reach the Oracle database.
Sources: src/flext_target_oracle/_utilities/loader.py:504-550, src/flext_target_oracle/_utilities/loader.py:588-605

Merge vs Append Mode

The loader supports different loading methods configured via sdc_mode or load_method:

• Append: Simply inserts new records.
• Merge: Before inserting the current batch, it executes a delete operation based on the primary key_properties to ensure existing records are updated (upsert behavior).
  Sources: src/flext_target_oracle/_utilities/loader.py:657-675, tests/integration/test_oracle.py:118-144

Error Handling

Data processing errors are managed through a specialized exception hierarchy that provides Oracle-specific context such as stream_name, record_count, and operation details.

class ProcessingError(Error):
    """Oracle processing error with Oracle-specific context."""
    def __init__(self, message, metadata=None, **kwargs):
        _, ctx = FlextTargetOracleExceptions._build_context(...)
        super().__init__(message=message)
        self.stream_name = ctx.get("stream_name")
        self.record_count = ctx.get("record_count")
        self.operation = str(ctx.get("operation"))

Sources: src/flext_target_oracle/_utilities/errors.py:118-136

Summary

Data Loading & Processing in the flext-target-oracle project provides a robust, batch-oriented pipeline for moving data from Singer sources to Oracle. By leveraging flext-db-oracle for SQL generation and execution, it maintains strict type safety and high performance through bulk operations, while offering extensive configuration for schema management, index creation, and conflict resolution (merge logic).