Introduction:
During global enterprise payroll processing cycles, syncing time management, corporate cross-references, and financial allowances between systems introduces complex data handling challenges. Integrating large datasets requires a middleware layer that manages data formatting transformations while operating within cloud memory limits.
A common bottleneck occurs when standard data loops run into string size boundaries while transforming data lines or calling target reporting modules.
This case study demonstrates how to construct a robust, biweekly scheduled processing pipeline within Oracle Integration Cloud (OIC). The solution combines data enrichment using the Snowflake Adapter, array manipulation via Read File in Segments, iterative BI Publisher (BIP) report processing, and PGP-encrypted bulk uploads through HCM Data Loader (HDL).
Technical Integration Architecture:
Rather than straining server boundaries, the data pipeline processes files using a segmented, decoupled, stream-based lifecycle:
1. SFTP Pull ──► Retrieve & Write to Stage Directory
│
▼
2. Snowflake Enrich ──► Dynamic Mapping Transformation
│
▼
3. Segment Chunking ──► Concatenate Rows into String
│
▼
4. Iterative BIP Call ──► Categorize Element / Abs
│
▼
5. HDL Load ──► PGP Encrypt, Compress & Monitor ID
Implementation Framework:
1. Inbound Processing & Snowflake Enrichment:
The scheduled orchestration begins by downloading the latest payroll raw file from a secure external SFTP directory server. If no files exist in the path, an inline Switch component catches the empty count (ItemCount = 0) and triggers a clean notifyNoFileFound exit routine, saving system resources.
When a payload is verified, OIC downloads the text file and writes it to a temporary local scratch directory via a Stage File Write action. The integration loop then immediately calls the endpoint getCompanyCodeCostCenterMapping using the Snowflake Adapter connection (MTWY Snowflake Connection). This function passes raw assignment parameters, returning validated corporate cost centers and company segment metadata strings straight into the OIC data object map.
2. Bypassing String Limits via Segment Chunking:
To prevent the application from hitting native string container memory thresholds when handling tens of thousands of employee rows, the workflow leverages an OIC Read File in Segments operation. This tool reads data in controlled, isolated arrays (e.g., chunks of 2,000 records per loop).
Inside the loop, the mapper concatenates variables into a delimited string. It structures individual column variables using a pipe separator (|) and joins row matrices together using standard comma flags (,).
The data transformation generates the following layout:
PERSON_NUM|ASG_NUM|DEPT_CD|COMPANY_CD,PERSON_NUM|ASG_NUM|DEPT_CD|COMPANY_CD
This structured block is mapped directly to consecutive collection fields (BIPParameter 1 through BIPParameter 8) to prepare the data for the next reporting step.
3. Hierarchical Regex Tokenizer & Tri-Directional Slicing:
OIC passes the generated parameter blocks to the Fusion public SOAP utility ExternalReportWSSService to execute a high-performance BI Publisher data model.
The custom SQL engine parses the comma-separated data stream on the fly. It utilizes a hierarchical regex tokenizer (REGEXP_SUBSTR and CONNECT BY level) to break the strings back into individual table rows. It also implements an analytical ranking function (ROW_NUMBER() OVER) to determine active assignment contexts before using a UNION ALL split to organize the data into either Element Entries or Absence Plans automatically:
/*
---------------------------------------------------------------------------------------------------------
-- Copyright(C) Oracle
-- All Rights Reserved
---------------------------------------------------------------------------------------------------------
-- Application : ORACLE ERP CLOUD
-- Report Name : ORA PY Kronos Time Usage Scheduled Time Off Data Model
-- Purpose :
-- CHANGE LOG
-- Date Author Version Description
---------------------------------------------------------------------------------------------------------
dd-mm-yyyy Satyanarayana Swamy Chinnamsetti 1.0 Initial Version
---------------------------------------------------------------------------------------------------------
*/
WITH input_value AS (
SELECT TRIM(REGEXP_SUBSTR(:p_input_value1, '[^,]+', 1, level)) AS param_set
FROM dual
WHERE TRIM(REGEXP_SUBSTR(:p_input_value1, '[^,]+', 1, level)) IS NOT NULL
OR TRIM(REGEXP_SUBSTR(:p_input_value1, '[^,]+', 1, level)) <> ''
CONNECT BY level <= REGEXP_COUNT(:p_input_value1, ',') + 1
UNION ALL
SELECT TRIM(REGEXP_SUBSTR(:p_input_value2, '[^,]+', 1, level)) AS param_set
FROM dual
WHERE TRIM(REGEXP_SUBSTR(:p_input_value2, '[^,]+', 1, level)) IS NOT NULL
OR TRIM(REGEXP_SUBSTR(:p_input_value2, '[^,]+', 1, level)) <> ''
CONNECT BY level <= REGEXP_COUNT(:p_input_value2, ',') + 1
UNION ALL
SELECT TRIM(REGEXP_SUBSTR(:p_input_value3, '[^,]+', 1, level)) AS param_set
FROM dual
WHERE TRIM(REGEXP_SUBSTR(:p_input_value3, '[^,]+', 1, level)) IS NOT NULL
OR TRIM(REGEXP_SUBSTR(:p_input_value3, '[^,]+', 1, level)) <> ''
CONNECT BY level <= REGEXP_COUNT(:p_input_value3, ',') + 1
UNION ALL
SELECT TRIM(REGEXP_SUBSTR(:p_input_value4, '[^,]+', 1, level)) AS param_set
FROM dual
WHERE TRIM(REGEXP_SUBSTR(:p_input_value4, '[^,]+', 1, level)) IS NOT NULL
OR TRIM(REGEXP_SUBSTR(:p_input_value4, '[^,]+', 1, level)) <> ''
CONNECT BY level <= REGEXP_COUNT(:p_input_value4, ',') + 1
UNION ALL
SELECT TRIM(REGEXP_SUBSTR(:p_input_value5, '[^,]+', 1, level)) AS param_set
FROM dual
WHERE TRIM(REGEXP_SUBSTR(:p_input_value5, '[^,]+', 1, level)) IS NOT NULL
OR TRIM(REGEXP_SUBSTR(:p_input_value5, '[^,]+', 1, level)) <> ''
CONNECT BY level <= REGEXP_COUNT(:p_input_value5, ',') + 1
UNION ALL
SELECT TRIM(REGEXP_SUBSTR(:p_input_value6, '[^,]+', 1, level)) AS param_set
FROM dual
WHERE TRIM(REGEXP_SUBSTR(:p_input_value6, '[^,]+', 1, level)) IS NOT NULL
OR TRIM(REGEXP_SUBSTR(:p_input_value6, '[^,]+', 1, level)) <> ''
CONNECT BY level <= REGEXP_COUNT(:p_input_value6, ',') + 1
UNION ALL
SELECT TRIM(REGEXP_SUBSTR(:p_input_value7, '[^,]+', 1, level)) AS param_set
FROM dual
WHERE TRIM(REGEXP_SUBSTR(:p_input_value7, '[^,]+', 1, level)) IS NOT NULL
OR TRIM(REGEXP_SUBSTR(:p_input_value7, '[^,]+', 1, level)) <> ''
CONNECT BY level <= REGEXP_COUNT(:p_input_value7, ',') + 1
UNION ALL
SELECT TRIM(REGEXP_SUBSTR(:p_input_value8, '[^,]+', 1, level)) AS param_set
FROM dual
WHERE TRIM(REGEXP_SUBSTR(:p_input_value8, '[^,]+', 1, level)) IS NOT NULL
OR TRIM(REGEXP_SUBSTR(:p_input_value8, '[^,]+', 1, level)) <> ''
CONNECT BY level <= REGEXP_COUNT(:p_input_value8, ',') + 1
), param_values AS (
SELECT REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 1, 'n', 1) AS person_number,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 2, 'n', 1) AS position_code,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 3, 'n', 1) AS job_code,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 4, 'n', 1) AS pay_code,
TO_DATE(REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 5, 'n', 1), 'YYYYMMDD') AS time_recorded_date,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 6, 'n', 1) AS hours,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 7, 'n', 1) AS rate,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 8, 'n', 1) AS shift,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 9, 'n', 1) AS gl_department,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 10, 'n', 1) AS gl_account,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 11, 'n', 1) AS gl_sub_account,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 12, 'n', 1) AS gl_company,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 13, 'n', 1) AS activity,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 14, 'n', 1) AS activity_account_category,
REGEXP_SUBSTR(param_set, '(.*?)(\||$)', 1, 15, 'n', 1) AS units
/* ,row_number() over (order by null) as rn */
FROM input_value
), element_entry_name AS (
SELECT DISTINCT
pv.pay_code,
pett_main.element_name AS element_name,
CASE
WHEN pett_main.element_name IS NOT NULL
THEN ''
ELSE 'Element Name not found'
END AS error_reason
FROM param_values pv
LEFT OUTER JOIN (
SELECT pett.element_name,
pett.description
FROM pay_element_types_f petf
INNER JOIN pay_ele_classifications pec
ON pec.classification_id = petf.secondary_classification_id
INNER JOIN pay_element_types_tl pett
ON petf.element_type_id = pett.element_type_id
WHERE pec.base_classification_name <> 'Standard Earnings Premium'
AND UPPER(pett.element_name) LIKE '%BC%'
) pett_main
ON pett_main.description = pv.pay_code
), premium_element_entry_name AS (
SELECT DISTINCT
pv.pay_code,
fuci.value AS premium_element_name
FROM ff_user_tables fut
INNER JOIN ff_user_rows_f furf
ON fut.user_table_id = furf.user_table_id
INNER JOIN ff_user_columns fuc
ON fut.user_table_id = fuc.user_table_id
INNER JOIN ff_user_columns_tl fuct
ON fuc.user_column_id = fuct.user_column_id
INNER JOIN ff_user_column_instances_f fuci
ON (
fuc.user_column_id = fuci.user_column_id AND furf.user_row_id = fuci.user_row_id
)
INNER JOIN param_values pv
ON furf.row_low_range_or_name = pv.pay_code
WHERE fut.base_user_table_name = 'EARNINGS_TABLE'
AND fuct.user_column_name = 'PREMIUM_ELEMENT'
), absence_plans AS (
SELECT DISTINCT
pv.pay_code,
fuci.value AS absence_plan_name
FROM ff_user_tables fut
INNER JOIN ff_user_rows_f furf
ON fut.user_table_id = furf.user_table_id
INNER JOIN ff_user_columns fuc
ON fut.user_table_id = fuc.user_table_id
INNER JOIN ff_user_columns_tl fuct
ON fuc.user_column_id = fuct.user_column_id
INNER JOIN ff_user_column_instances_f fuci
ON (
fuc.user_column_id = fuci.user_column_id AND furf.user_row_id = fuci.user_row_id
)
INNER JOIN param_values pv
ON furf.row_low_range_or_name = pv.pay_code
WHERE fut.base_user_table_name = 'EARNINGS_TABLE'
AND fuct.user_column_name = 'ABSENCE_PLAN'
), person AS (
SELECT DISTINCT
papf.person_id,
pv.person_number,
CASE WHEN papf.person_id IS NOT NULL THEN '' ELSE 'Person not found in Oracle' END AS error_reason
/* ,pv.rn */
FROM param_values pv
LEFT OUTER JOIN per_all_people_f papf
ON (
papf.person_number = pv.person_number
AND TRUNC(SYSDATE, 'DD') BETWEEN papf.effective_start_date AND papf.effective_end_date
)
), latest_assignment AS (
SELECT pe.person_number,
paam.assignment_number,
ROW_NUMBER() OVER (PARTITION BY paam.person_id ORDER BY paam.effective_start_date DESC) AS rn_assignment
FROM per_all_assignments_m paam
INNER JOIN hr_organization_units_f_tl houft
ON houft.organization_id = paam.business_unit_id
INNER JOIN person pe
ON paam.person_id = pe.person_id
INNER JOIN param_values pv
ON pe.person_number = pv.person_number
LEFT OUTER JOIN hr_all_positions_f hapf
ON paam.position_id = hapf.position_id
LEFT OUTER JOIN per_jobs_f pjf
ON paam.job_id = pjf.job_id
WHERE TRUNC(SYSDATE, 'DD') BETWEEN houft.effective_start_date AND houft.effective_end_date
AND paam.assignment_type = 'E'
AND houft.language = USERENV('LANG')
AND UPPER(houft.name) = UPPER('Oracle BU')
AND (
(
(
pv.position_code IS NOT NULL OR pv.position_code <> ''
)
AND
(
hapf.position_code = pv.position_code
)
)
OR (
(
pv.position_code IS NULL OR pv.position_code = ''
)
AND
(
pv.job_code IS NOT NULL OR pv.job_code <> ''
)
AND
(
pjf.job_code = pv.job_code
)
)
OR (
(
pv.position_code IS NULL OR pv.position_code = ''
)
AND
(
pv.job_code IS NULL OR pv.job_code = ''
)
AND
(
TRUNC(pv.time_recorded_date) BETWEEN paam.effective_start_date AND paam.effective_end_date AND paam.primary_flag = 'Y' AND paam.assignment_status_type = 'ACTIVE'
)
)
)
)
SELECT *
FROM (
SELECT pv.person_number, /* rn */
pv.position_code,
pv.job_code,
pv.pay_code,
TO_CHAR(pv.time_recorded_date, 'YYYY/MM/DD') AS time_recorded_date,
pv.hours,
pv.rate,
pv.shift,
pv.gl_department,
pv.gl_account,
pv.gl_sub_account,
pv.gl_company,
pv.activity,
pv.activity_account_category,
pv.units,
la.assignment_number,
een.element_name,
TO_CHAR(
1000 + ROW_NUMBER() OVER (PARTITION BY pv.person_number, pv.pay_code ORDER BY pv.person_number, pv.pay_code, pv.time_recorded_date) - 1
) AS create_entry_sequence,
'Element' AS hdl_type,
CASE
WHEN pe.error_reason IS NOT NULL AND een.error_reason IS NOT NULL
THEN pe.error_reason || ',' || een.error_reason
WHEN pe.error_reason IS NOT NULL
THEN pe.error_reason
WHEN een.error_reason IS NOT NULL
THEN een.error_reason
WHEN la.assignment_number IS NULL
THEN 'Assignment Number not found'
ELSE ''
END AS error_reason,
CASE WHEN pv.gl_company IS NOT NULL AND pv.gl_department IS NOT NULL THEN pv.gl_company || '.' || pv.gl_department ELSE '' END AS CONCAT
FROM param_values pv
INNER JOIN person pe
ON pe.person_number = pv.person_number
INNER JOIN element_entry_name een
ON pv.pay_code = een.pay_code
LEFT OUTER JOIN latest_assignment la
ON (
pv.person_number = la.person_number AND la.rn_assignment = 1
)
UNION ALL
SELECT pv.person_number, /* rn */
pv.position_code,
pv.job_code,
pv.pay_code,
TO_CHAR(pv.time_recorded_date, 'YYYY/MM/DD') AS time_recorded_date,
pv.hours,
pv.rate,
pv.shift,
pv.gl_department,
pv.gl_account,
pv.gl_sub_account,
pv.gl_company,
pv.activity,
pv.activity_account_category,
pv.units,
la.assignment_number,
peen.premium_element_name,
TO_CHAR(
1000 + ROW_NUMBER() OVER (PARTITION BY pv.person_number, pv.pay_code ORDER BY pv.person_number, pv.pay_code, pv.time_recorded_date) - 1
) AS create_entry_sequence,
'Element' AS hdl_type,
CASE WHEN pe.error_reason IS NOT NULL THEN pe.error_reason ELSE '' END AS error_reason,
CASE WHEN pv.gl_company IS NOT NULL AND pv.gl_department IS NOT NULL THEN pv.gl_company || '.' || pv.gl_department ELSE '' END AS CONCAT
FROM param_values pv
INNER JOIN person pe
ON pe.person_number = pv.person_number
INNER JOIN premium_element_entry_name peen
ON pv.pay_code = peen.pay_code
INNER JOIN latest_assignment la
ON (
pv.person_number = la.person_number AND la.rn_assignment = 1
)
UNION ALL
SELECT pv.person_number, /* rn */
pv.position_code,
pv.job_code,
pv.pay_code,
TO_CHAR(pv.time_recorded_date, 'YYYY/MM/DD') AS time_recorded_date,
pv.hours,
pv.rate,
pv.shift,
pv.gl_department,
pv.gl_account,
pv.gl_sub_account,
pv.gl_company,
pv.activity,
pv.activity_account_category,
pv.units,
la.assignment_number,
ap.absence_plan_name,
'' AS create_entry_sequence,
'Absence' AS hdl_type,
CASE WHEN pe.error_reason IS NOT NULL THEN pe.error_reason ELSE '' END AS error_reason,
CASE WHEN pv.gl_company IS NOT NULL AND pv.gl_department IS NOT NULL THEN pv.gl_company || '.' || pv.gl_department ELSE '' END AS CONCAT
FROM param_values pv
INNER JOIN person pe
ON pe.person_number = pv.person_number
INNER JOIN absence_plans ap
ON pv.pay_code = ap.pay_code
INNER JOIN latest_assignment la
ON (
pv.person_number = la.person_number AND la.rn_assignment = 1
)
)
ORDER BY TO_NUMBER(person_number)
An inline validation rule (reportBytes != DefaultContent) catches any blank data blocks early. This ensures that only valid, populated responses move forward to the file writing stage.
4: Formatting Native HDL Schema Files:
OIC decodes the Base64 binary text from the report response and reads the entries line-by-line. Based on the calculated hdl_type tag column value, the integration splits rows into two separate local stage files, structuring them to match the exact schema layouts expected by the HCM Data Loader engine:
File 1: ElementEntryWithCosting.dat:
This file handles payroll allocations and maps Snowflake dimensions directly to the entry costing segments (Segment1 through Segment3):
METADATA|ElementEntryWithCosting|SourceSystemOwner|SourceSystemId|ElementName|EffectiveStartDate|EffectiveEndDate|AssignmentNumber|EntryType|InputValueName1|ScreenEntryValue1|InputValueName2|ScreenEntryValue2|CreateEntrySequence|Segment1|Segment2|Segment3
MERGE|ElementEntryWithCosting|HRC_SQLLOADER|HRC_SQLLOADER_E112698_102|BC101 - Reg Pay|2025/08/02|4712/12/31|E112698|E|Hours|0.25|Rate|0|1000|6000|1001|6000
File 2: PersonAccrualDetail.dat:
METADATA|PersonAccrualDetail|PersonNumber|WorkTermsNumber|PlanName|AccrualType|Value|ProcdDate|AdjustmentReason
MERGE|PersonAccrualDetail|110684|ET110684|Extended Leave|ADJOTH|4|2025/07/26|CP
Once written, both files are compressed into a single zip archive. To maintain absolute data privacy compliance for sensitive employee financial data, the integration runs a secure command wrapper that encrypts the zip payload using a corporate PGP Public Key before passing the file to the HCM Cloud Adapter bulk load service.
4. Process Monitoring and Diagnostic Logging Flow:
The integration tracks execution health by capturing the unique Process ID returned by the background loading engine. A structured tracking matrix logs every phase of the lifecycle:
[OIC_TRACKING] [BIWEEKLY_RUN] Started payroll pipeline sequence.
[SFTP_STAGE] Successfully uncompressed source text stream.
[SNOWFLAKE] Enriched 4,500 record streams with matching company codes.
[SEGMENT_LOOP] Processing File Chunk 1 (Rows 1-2000). Total length within limits.
[BIP_INVOKE] Iterative Data Model call complete. Generated XML response elements.
[HDL_ENGINE] PGP Encryption complete. Submitted zip payload. Process Tracking ID: 890241.
[HDL_STATUS] Process ID 890241 returned: SUCCESS. Sending confirmation alerts.
[SFTP_ARCHIVE] Cleaned source folder pathing parameters. Pipeline shut down.
If an processing exception occurs, global fault-handling catch blocks intercept the failed payload instance, generate a detailed administrative alert, and safely archive the source file to prevent duplicate processing cycles.
5. Summary:
This integration has been developed to support the creation of Element Entries and Absence Entries as part of the payroll process. It operates on a biweekly schedule, automatically retrieving the latest encoded file from the SFTP server. Once the file is received, it is decoded and written to the OIC temporary directory using the Stage File action. The data is then processed and transformed, incorporating department and company code mappings sourced through the Snowflake adapter to ensure consistency and accuracy.
Following the transformation, the data is structured by separating columns with a pipe (“|”) and combining rows using commas. This formatting is performed within the “Read File in Segments” operation, enabling efficient processing and transmission to a BI Publisher report. Due to input size limitations, the data is divided into manageable segments, and the report is triggered iteratively until all records are processed. Based on the report output, records are categorized as either Element Entries or Absence Entries, and the corresponding DAT files are generated. These files are then compressed and encrypted using a PGP key before being submitted to Oracle HCM Cloud through the HCM Data Loader.
The outcome of the data load process is monitored using the generated Process ID, allowing appropriate success or failure notifications to be issued. Once processing is complete, the source file is archived from the SFTP directory. The integration also includes configurable flags to manage notifications and encryption requirements. Comprehensive error handling is implemented through both default and global fault handlers, and reusable integrations are leveraged to ensure consistent and efficient notification delivery and SFTP file movement.
Conclusion:
Combining BI Publisher, Oracle Integration Cloud, and HCM Data Loader into a single automated pipeline creates a highly scalable data integration framework. By extracting datasets in bulk using high-performance database models and processing data parsing entirely within local OIC memory maps, you protect environment stability, ensure clean data updates, and avoid system failures during business-critical integration windows.
Top comments (0)