Peoplesoft to Salesforce: Student Service Indicator Sync

#   _____           _              _____        __      
#  |  __ \         (_)            |_   _|      / _|     
#  | |__) |___  ___ _ _ __   ___    | |  _ __ | |_ ___  
#  |  _  // _ \/ __| | '_ \ / _ \   | | | '_ \|  _/ _ \ 
#  | | \ \  __/ (__| | |_) |  __/  _| |_| | | | || (_) |
#  |_|  \_\___|\___|_| .__/ \___| |_____|_| |_|_| \___/ 
#                    | |                                
#                    |_|

# Project Name - Peoplesoft to Salesforce: Student Service Indicator Sync

# Recipe URL - https://app.lingk.io/a/10932/tf/19105

# Description - 11 - Peoplesoft to Salesforce: Student Service Indicator Sync
                # Select all service indicator from PeopleSoft SRVC_IND_DATA where INSTITUTION = '{{env.inst-code}}' and SRVC_IND_DTTM >= {{env.vars.serviceIndicatorLastUpdatedDate}}.
                # Reject non-matching records due to Contact & Term.
                # Collect info from all tables and transform fields.
                # Insert/Update record into SF by PeopleSoft_Service_Indicator_Unique_Id__c.
                # Insert service indicator requested fields with their SF id in Lingk environment file.

# Systems - Peoplesoft, Salesforce
# Connectors - HttpV2, Salesforce
# Data Flows - Single Direction
# Connection Type - API to API

# Data Models
## Peoplesoft Models Used:
#1. SRVC_IND_DATA

## Salesforce Objects Used:
#1. Contact
#2. hed__Term__c
#3. Service_Indicator__c

# Add Recipe notes / Change log information here!

#    _____                            _                 
#   / ____|                          | |                
#  | |     ___  _ __  _ __   ___  ___| |_ ___  _ __ ___ 
#  | |    / _ \| '_ \| '_ \ / _ \/ __| __/ _ \| '__/ __|
#  | |___| (_) | | | | | | |  __/ (__| || (_) | |  \__ \
#   \_____\___/|_| |_|_| |_|\___|\___|\__\___/|_|  |___/
#

# CONNECTORS specify what data will be pulled into the in-memory database during processing

connectors:

# Configure HTTP & Salesforce credentials in your Environment before running this recipe
  # HTTP Setup - https://help.lingk.io/en/articles/304-http-connector-setup-guide
  # Salesforce Setup - https://help.lingk.io/en/articles/267-salesforce-connector-setup-guide

###### Start - HTTP Reader connectors ######

## PeopleSoft Person Connector
  ## HTTP V2 connector in comparison with V1, handles pagination through the connector properties
  ## Authentication and URL values are provided via environment variables  
    # documentation - https://help.lingk.io/en/articles/202-custom-environment-variables

## PeopleSoft Service Indicator Connector
- 
    name: peopleSoftServiceIndicatorData
    type: httpV2Reader 
    properties: 
      httpMethod: POST
      headerJson: {
          "Authorization":"Basic {{env.vars.apiAccessToken}}",
          "Content-Type":"application/json"
        }
      acceptInvalidCerts: false
      url: '{{env.vars.midStateInstanceURL}}'
      paginationType: requestBody    
      body: 
            {
                "isDebugMode": false,
                "rowLimit": 1000,
                "pageNumber": 1,
                "records": [
                    {
                        "recordName": "SRVC_IND_DATA",
                        "sqlWhereClause": " A.INSTITUTION = '{{env.inst-code}}' and SRVC_IND_DTTM >= %DateTimeIn('{{env.vars.studentAccLastUpdateDate}}')  ",
                        "includeDescriptionsFor": [
                            "SRVC_IND_CD",
                            "SRVC_IND_REASON"
                        ]
                    },
                    {
                        "recordName": "SRVC_IND_CD_TBL",
                        "parentRecordName": "SRVC_IND_DATA"
                    },
                    {
                        "recordName": "SRVC_IMPACT_VW3",
                        "parentRecordName": "SRVC_IND_DATA"
                    },
                    {
                        "recordName": "SRVC_IN_RSN_TBL",
                        "parentRecordName": "SRVC_IND_DATA"
                    }
                    ,
                    {
                        "recordName": "DEPT_TBL",
                        "parentRecordName": "SRVC_IND_DATA",
                          "sqlWhereClause": " A.SETID = '{{env.inst-code}}' "
                    }
                ]
            }

schema:     
      fields:
        - name: 'AMOUNT'
          type: 'string'
        - name: 'COMM_COMMENTS'
          type: 'string'
        - name: 'CONTACT'
          type: 'string'
        - name: 'CONTACT_ID'
          type: 'string'
        - name: 'CURRENCY_CD'
          type: 'string'
        - name: 'DEPTID'
          type: 'string'
        - name: 'EMPLID'
          type: 'string'
        - name: 'INSTITUTION'
          type: 'string'
        - name: 'OPRID'
          type: 'string'
        - name: 'PLACED_METHOD'
          type: 'string'
        - name: 'PLACED_PERSON'
          type: 'string'
        - name: 'PLACED_PERSON_ID'
          type: 'string'
        - name: 'PLACED_PROCESS'
          type: 'string'
        - name: 'POSITION_NBR'
          type: 'string'
        - name: 'POS_SRVC_INDICATOR'
          type: 'string'
        - name: 'RELEASE_PROCESS'
          type: 'string'
        - name: 'SCC_SI_END_DT'
          type: 'string'
        - name: 'SCC_SI_END_TERM'
          type: 'string'
        - name: 'SEQ_3C'
          type: 'long'
        - name: 'SRVC_IMPACT_VW3'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'DESCR'
                        type: 'string'
                      - name: 'DESCRLONG'
                        type: 'string'
                      - name: 'EFFDT'
                        type: 'string'
                      - name: 'INSTITUTION'
                        type: 'string'
                      - name: 'SCC_IMPACT_DATE'
                        type: 'string'
                      - name: 'SCC_IMPACT_TERM'
                        type: 'string'
                      - name: 'SERVICE_IMPACT'
                        type: 'string'
                      - name: 'SRVC_IND_CD'
                        type: 'string'
                      - name: 'TERM_CATEGORY'
                        type: 'string'
              - name: 'objectName'
                type: 'string'
              - name: 'objectType'
                type: 'string'
        - name: 'SRVC_IN_RSN_TBL'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'DESCRSHORT'
                        type: 'string'
                      - name: 'DESCRLONG'
                        type: 'string'
                      - name: 'INSTITUTION'
                        type: 'string'
                      - name: 'SRVC_IND_CD'
                        type: 'string'
                      - name: 'SRVC_IND_REASON'
                        type: 'string'
                      - name: 'DEPTID'
                        type: 'string'
                      - name: 'EFFDT'
                        type: 'string'
        - name: 'DEPT_TBL'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'DEPTID'
                        type: 'string'
                      - name: 'DESCR'
                        type: 'string'
                      - name: 'SETID'
                        type: 'string'
                      - name: 'EFFDT'
                        type: 'string'
        - name: 'SRVC_IND_ACTIVE_DT'
          type: 'string'
        - name: 'SRVC_IND_ACT_TERM'
          type: 'string'
        - name: 'SRVC_IND_CD'
          type: 'string'
        - name: 'SRVC_IND_CD_TBL'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'DESCR'
                        type: 'string'
                      - name: 'DESCRSHORT'
                        type: 'string'
                      - name: 'DFLT_SRVC_IND_RSN'
                        type: 'string'
                      - name: 'EFFDT'
                        type: 'string'
                      - name: 'EFF_STATUS'
                        type: 'string'
                      - name: 'INSTITUTION'
                        type: 'string'
                      - name: 'POS_SRVC_INDICATOR'
                        type: 'string'
                      - name: 'SCC_DFLT_ACTDATE'
                        type: 'string'
                      - name: 'SCC_DFLT_ACTTERM'
                        type: 'string'
                      - name: 'SCC_HOLD_DISPLAY'
                        type: 'string'
                      - name: 'SCC_SI_ORG'
                        type: 'string'
                      - name: 'SCC_SI_PERS'
                        type: 'string'
                      - name: 'SRVC_IND_CD'
                        type: 'string'
                      - name: 'SRV_IND_DCSD_FLAG'
                        type: 'string'
              - name: 'objectName'
                type: 'string'
              - name: 'objectType'
                type: 'string'
        - name: 'SRVC_IND_DTTM'
          type: 'string'
        - name: 'SRVC_IND_REASON'
          type: 'string'
        - name: 'SRVC_IND_REFRNCE'
          type: 'string'
        - name: 'rowNumber'
          type: 'long'

###### End - HTTP Reader connectors ######

###### Start - Salesforce Reader connectors ######

# Salesforce Contact Reader
-
    name: sfContactReader
    type: salesforceReader
    delayedRead: true
    parameterizedBy: batchedStudentId
    properties:
      useBearerToken: true
      query: 
        SELECT 
          Id,
          University_Student_ID__c,
          Name   
        FROM Contact
        WHERE University_Student_ID__c IN (
            {% for element in var.output %}
            '{{element.id}}'{% if not loop.last %},{% endif %}
            {% endfor %}
          )

schema:
      fields:
        - name: 'Id'
          type: 'string'
        - name: 'University_Student_ID__c'
          type: 'string'
        - name: 'Name'
          type: 'string'

# Salesforce Terms Reader
-
    name: sfTermReader
    type: salesforceReader
    delayedRead: true
    properties:
      useBearerToken: true
      query: 
        SELECT 
          Id,
          Name,
          PeopleSoft_Term_ID__c,
          hed__Start_Date__c,
          hed__End_Date__c,
          hed__Type__c,
          hed__Account__c,
          Case_Safe_ID_Term__c 
        FROM hed__Term__c
    schema:
      fields:
        - name: 'Id'
          type: 'string'
        - name: 'Name'
          type: 'string'
        - name: 'PeopleSoft_Term_ID__c'
          type: 'string'
        - name: 'hed__Start_Date__c'
          type: 'string'
        - name: 'hed__End_Date__c'
          type: 'string'
        - name: 'hed__Type__c'
          type: 'string'
        - name: 'hed__Account__c'
          type: 'string'
        - name: 'Case_Safe_ID_Term__c'
          type: 'string'

# Salesforce Service Indicator Reader
-
    name: sfServiceIndicatorReader
    type: salesforceReader
    delayedRead: true
    parameterizedBy: batchedServiceIndicatorId
    properties:
      useBearerToken: true
      query: SELECT 
              Id, 
              Name,
              PeopleSoft_Service_Indicator_Unique_Id__c
             FROM Service_Indicator__c
             WHERE PeopleSoft_Service_Indicator_Unique_Id__c IN (
             {% for element in var.output %}
             '{{element.id}}'{% if not loop.last %},{% endif %}
             {% endfor %}
          ) 
    schema:
      fields:
        - name: 'Id'
          type: 'string'
        - name: 'Name'
          type: 'string' 
        - name: 'PeopleSoft_Service_Indicator_Unique_Id__c'
          type: 'string'

###### End - Salesforce Reader connectors ######

###### Start - Salesforce Writer connectors ######

# Salesforce Service Indicator Writer
-
    name: sfServiceIndicatorWriter
    type: salesforceBulkWriter
    onError: continue
    isSerial: true
    properties:
      onError: continue
      useBearerToken: true
      sfObject: Service_Indicator__c 
      batchSize: 50

###### End - Salesforce Writer connectors ######

###### Start - Local Writer connectors ######

# Local File Writer
-   
    name: finalMappedRecordFile
    type: localFileWriter
    format: csvWriter
    properties:
      fileName: "finalMappedStudentServiceIndicatorRecords.csv"

###### Start - Local Writer connectors ######

#   _______        _        
#  |__   __|      | |       
#     | | __ _ ___| | _____ 
#     | |/ _` / __| |/ / __|
#     | | (_| \__ \   <\__ \
#     |_|\__,_|___/_|\_\___/

# Batch task used for parameterized by statements/connectors
tasks:
-
    name: batchTask
    type: dataOperation 
    function: batch   
    parameters:
       inputBatchGroupBy: # required
         id: desc      
       inputBatchSize: 200 # inputBatchSize # required (to test)
       inputBatchFields: "id" # comma delimited for other fields #inputBatchFields #required
       batchedColumnAlias: output # outputColumnName - optional with a default of "output"
       # inputTable: # optional
       # result: #optional

#   ______                         _       
#  |  ____|                       | |      
#  | |__ ___  _ __ _ __ ___   __ _| |_ ___ 
#  |  __/ _ \| '__| '_ ` _ \ / _` | __/ __|
#  | | | (_) | |  | | | | | | (_| | |_\__ \
#  |_|  \___/|_|  |_| |_| |_|\__,_|\__|___/

writeFormats:
-   
    name: csvWriter
    type: delimited
    properties:
      quoteAllFields: true
      delimiter: ','
      header: true

#    _____ _        _                            _       
#   / ____| |      | |                          | |      
#  | (___ | |_ __ _| |_ ___ _ __ ___   ___ _ __ | |_ ___ 
#   \___ \| __/ _` | __/ _ \ '_ ` _ \ / _ \ '_ \| __/ __|
#   ____) | || (_| | ||  __/ | | | | |  __/ | | | |_\__ \
#  |_____/ \__\__,_|\__\___|_| |_| |_|\___|_| |_|\__|___/

# STATEMENTS specify how the data should be processed while in memory

statements:

#******************************************************************** D I S C L A I M E R ***********************************************************************************************
#                                                                                                                                                                                       * 
# Note that in an effort to keep recipes optimized for DPH (Data Processing Hours), print statements should be commented out after development has concluded for a recipe.              *
# For more information on DPH optimization, please visit the following help article - https://help.lingk.io/en/articles/212-minimizing-data-processing-hours-on-the-lingk-platform      *
#                                                                                                                                                                                       * 
#******************************************************************** D I S C L A I M E R ***********************************************************************************************

#- statement: print peopleSoftServiceIndicatorData
  - statement: (peopleSoftServiceIndicatorDataCount) => SELECT COUNT(*) FROM peopleSoftServiceIndicatorData
  #- statement: PRINT peopleSoftServiceIndicatorDataCount

# inline function - flatten the array to make it a basic table
  - statement: (inlineServiceIndicatorTable) => SELECT inline(SRVC_IND_CD_TBL.fields) FROM peopleSoftServiceIndicatorData
  #- statement: PRINT inlineServiceIndicatorTable

## Using the ROW_NUMBER function assigns a unique, sequential number to each row, starting with one,
   # according to the ordering of rows within the partition from the information pulled in the previous statement for the SRVC_IND_CD_TBL
  ## The PARTITION BY is dividing the rows into small partitions by EFFDT preference field
  - statement: |
      (distinctInlineServiceIndicatorTable) => 
          SELECT 
            *,
            ROW_NUMBER() OVER (PARTITION BY TRIM(SRVC_IND_CD), TRIM(INSTITUTION) ORDER BY TRIM(EFFDT) DESC) AS Row
          FROM inlineServiceIndicatorTable

#- statement: PRINT distinctInlineServiceIndicatorTable
  - statement: (distinctInlineServiceIndicatorTableCount) => SELECT COUNT(*) FROM distinctInlineServiceIndicatorTable
  #- statement: PRINT distinctInlineServiceIndicatorTableCount

# inline function - flatten the array to make it a basic table
  - statement: (inlineServiceImpact) => SELECT inline(SRVC_IMPACT_VW3.fields) FROM peopleSoftServiceIndicatorData
  #- statement: PRINT inlineServiceImpact

# inline function - flatten the array to make it a basic table
  - statement: (inlineServiceIndicatorReason) => SELECT inline(SRVC_IN_RSN_TBL.fields) FROM peopleSoftServiceIndicatorData

## Using the ROW_NUMBER function assigns a unique, sequential number to each row, starting with one,
   # according to the ordering of rows within the partition from the information pulled in the previous statement for the SRVC_IND_CD_TBL
  ## The PARTITION BY is dividing the rows into small partitions by EFFDT preference field 
  - statement: |
      (distinctServiceIndicatorReasonTable) => 
          SELECT 
            *,
            ROW_NUMBER() OVER (PARTITION BY TRIM(SRVC_IND_CD), TRIM(INSTITUTION) ORDER BY TRIM(EFFDT) DESC) AS Row
          FROM inlineServiceIndicatorReason

#- statement: PRINT distinctServiceIndicatorReasonTable
  - statement: (distinctServiceIndicatorReasonTableCount) => SELECT COUNT(*) FROM distinctServiceIndicatorReasonTable
  #- statement: PRINT distinctServiceIndicatorReasonTableCount

- statement: (inlineDepartmentInfo) => SELECT inline(DEPT_TBL.fields) FROM peopleSoftServiceIndicatorData
  - statement: (finalDepartmentInfo) => SELECT DISTINCT * FROM inlineDepartmentInfo
  #- statement: print finalDepartmentInfo

## Get Salesforce Student Info
  - statement: (distinctStudentId) => SELECT DISTINCT TRIM(EMPLID) `id` FROM peopleSoftServiceIndicatorData

# This statement executes a batch process task where the name, inputTable, outputTable and the results are the parameters needed 
  - statement: |
        execute task 
            --name batchTask
            --inputTable distinctStudentId
            --outputBatchTable batchedStudentId
            --result resultStatus

- statement: REFRESH sfContactReader
  - statement: (sfContactReaderCount) => SELECT COUNT(*) FROM sfContactReader
  #- statement: PRINT sfContactReaderCount

## Get Salesforce Terms
  - statement: REFRESH sfTermReader
  - statement: (sfTermReaderCount) => SELECT COUNT(*) FROM sfTermReader
  #- statement: PRINT sfTermReaderCount

## Reject non-matching records due to
  # Contact
  # The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table)
  - statement: (rejectedRecordByNonMatchingContact) => 
                    SELECT 
                      * 
                    FROM peopleSoftServiceIndicatorData si
                    LEFT ANTI JOIN sfContactReader sf ON IFNULL(TRIM(si.EMPLID), '') = IFNULL(TRIM(sf.University_Student_ID__c), '')

- statement: (rejectedRecordByNonMatchingContactCount) => SELECT COUNT(*) FROM rejectedRecordByNonMatchingContact
  #- statement: PRINT rejectedRecordByNonMatchingContactCount

- statement: 
      (serviceImpactGroup) =>
          SELECT
            IFNULL(TRIM(sid.EMPLID), '')                         `EMPLID`,
            IFNULL(TRIM(sid.INSTITUTION), '')                    `INSTITUTION`,
            IFNULL(TRIM(sid.SRVC_IND_CD), '')                    `SRVC_IND_CD`,
            collect_set(
                IFNULL(TRIM(imp.SERVICE_IMPACT), '')
            )                                                    `service_impact_list`
          FROM peopleSoftServiceIndicatorData sid
          INNER JOIN inlineServiceImpact imp ON TRIM(imp.INSTITUTION) = TRIM(sid.INSTITUTION)
            AND TRIM(imp.SRVC_IND_CD) = TRIM(sid.SRVC_IND_CD)
          GROUP BY TRIM(sid.EMPLID), TRIM(sid.INSTITUTION), TRIM(sid.SRVC_IND_CD)
  #- statement: PRINT serviceImpactGroup

## collect info from all tables required for Service Indicator.
  # The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table)
  - statement: (completeInfoForServiceIndicator) =>
                      SELECT
                        IFNULL(TRIM(si.EMPLID), '')                                  `EMPLID`,
                        IFNULL(TRIM(si.SRVC_IND_DTTM), '')                           `SRVC_IND_DTTM`,
                        IFNULL(TRIM(si.SRVC_IND_CD), '')                             `SRVC_IND_CD`,
                        IFNULL(TRIM(si.SRVC_IND_REASON), '')                         `SRVC_IND_REASON`,
                        IFNULL(TRIM(si.SRVC_IND_ACTIVE_DT), '')                      `SRVC_IND_ACTIVE_DT`,

IFNULL(TRIM(si.SRVC_IND_ACT_TERM), '')                       `SRVC_IND_ACT_TERM`,
                        IFNULL(TRIM(si.SCC_SI_END_TERM), '')                         `SCC_SI_END_TERM`,
                        IFNULL(TRIM(dpt.DESCR), '')                                  `DEPTID`,
                        replace(replace(replace(replace(replace(IFNULL(TRIM(si.COMM_COMMENTS), ''), '<p>', ''), '&nbsp;', ''), 
                                '<strong>', ''), '&amp;', ''), '&#39;', '') 
                                                                                     `COMM_COMMENTS`,
                        IFNULL(TRIM(si.CONTACT), '')                                 `CONTACT`,
                        IFNULL(TRIM(si.PLACED_PERSON), '')                           `PLACED_PERSON`,
                        IFNULL(TRIM(it.POS_SRVC_INDICATOR), '')                      `POS_SRVC_INDICATOR`,
                        concat_ws(',', imp.service_impact_list)                      `SERVICE_IMPACT`,
                        replace(replace(TRIM(concat_ws(' ', rs.DESCRSHORT ,rs.DESCRLONG)), '</p>', ''), '<p>', '')
                                                                                     `DESCR`

FROM peopleSoftServiceIndicatorData si
                      LEFT JOIN distinctInlineServiceIndicatorTable it ON TRIM(si.INSTITUTION) = TRIM(it.INSTITUTION) 
                        AND TRIM(si.SRVC_IND_CD) = TRIM(it.SRVC_IND_CD)
                        AND it.Row = 1
                      LEFT JOIN serviceImpactGroup imp ON TRIM(si.INSTITUTION) = TRIM(imp.INSTITUTION) 
                        AND TRIM(si.SRVC_IND_CD) = TRIM(imp.SRVC_IND_CD) AND TRIM(si.EMPLID) = TRIM(imp.EMPLID)
                      LEFT JOIN distinctServiceIndicatorReasonTable rs ON TRIM(si.INSTITUTION) = TRIM(rs.INSTITUTION) 
                        AND TRIM(si.SRVC_IND_CD) = TRIM(rs.SRVC_IND_CD)
                        AND rs.Row = 1
                      LEFT JOIN finalDepartmentInfo dpt ON IFNULL(UPPER(TRIM(si.DEPTID)), '') = IFNULL(UPPER(TRIM(dpt.DEPTID)), '')

# IFNULL(TRIM(si.SRVC_IND_END_DT), '')                         `SRVC_IND_END_DT`,

#- statement: PRINT completeInfoForServiceIndicator
  - statement: (completeInfoForServiceIndicatorCount) => SELECT COUNT(*) FROM completeInfoForServiceIndicator
  #- statement: PRINT completeInfoForServiceIndicatorCount

## Map Service Indicator info with SF fields
  # CASE function works as a dictionary and retrieve a specific value depending to the type field value pulled from the table as long as it is not null.
  # The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table)
  - statement: (mappedServiceIndicator) => 
                          SELECT
                            concat_ws('-', TRIM(si.EMPLID), TRIM(si.SRVC_IND_DTTM))                              `PeopleSoft_Service_Indicator_Unique_Id__c`,
                            sf.Id                                                                                `Contact__c`,
                            IFNULL(si.SRVC_IND_CD, '')                                                           `Service_Indicator_Code__c`,
                            IFNULL(si.SRVC_IND_REASON, '')                                                       `Service_Indicator_Reason__c`,
                            to_date(IFNULL(si.SRVC_IND_ACTIVE_DT, ''))                                           `Start_Date__c`,

t.Id                                                                                 `Start_Term__c`,
                            st.Id                                                                                `End_Term__c`,
                            IFNULL(si.DESCR, '')                                                                 `Description__c`,
                            IFNULL(si.DEPTID, '')                                                                `Department__c`,
                            CASE WHEN IFNULL(UPPER(si.POS_SRVC_INDICATOR), '') = 'N'
                              THEN 'Negative Service Indicator'
                             WHEN IFNULL(UPPER(si.POS_SRVC_INDICATOR), '') = 'Y'
                              THEN 'Positive Service Indicator'
                            ELSE ''
                            END                                                                                  `Effect__c`,
                            IFNULL(si.COMM_COMMENTS, '')                                                         `Comments__c`,
                            IFNULL(si.SERVICE_IMPACT, '')                                                        `Services_Impacted__c`,
                            IFNULL(si.CONTACT, '')                                                               `Contact_Person__c`,
                            IFNULL(si.PLACED_PERSON, '')                                                         `Placed_by__c`

FROM completeInfoForServiceIndicator si
                          INNER JOIN sfContactReader sf ON TRIM(si.EMPLID) = TRIM(sf.University_Student_ID__c)
                          LEFT JOIN sfTermReader t ON TRIM(si.SRVC_IND_ACT_TERM) = TRIM(t.PeopleSoft_Term_ID__c)
                          LEFT JOIN sfTermReader st ON TRIM(si.SCC_SI_END_TERM) = TRIM(st.PeopleSoft_Term_ID__c)
                          WHERE TRIM(si.EMPLID) != '' AND TRIM(si.SRVC_IND_DTTM) != ''

# to_date(IFNULL(si.SRVC_IND_END_DT, ''))                                              `End_Date__c`,

#- statement: PRINT mappedServiceIndicator
  - statement: (mappedServiceIndicatorCount) => SELECT COUNT(*) FROM mappedServiceIndicator
  #- statement: PRINT mappedServiceIndicatorCount

- statement: (distinctServiceIndicatorId) => SELECT DISTINCT PeopleSoft_Service_Indicator_Unique_Id__c `id` FROM mappedServiceIndicator

# This statement executes a batch process task where the name, inputTable, outputTable and the results are the parameters needed 
  - statement: |
        execute task 
            --name batchTask
            --inputTable distinctServiceIndicatorId
            --outputBatchTable batchedServiceIndicatorId
            --result resultStatus

- statement: REFRESH sfServiceIndicatorReader
  - statement: (sfServiceIndicatorReaderCount) => SELECT COUNT(*) FROM sfServiceIndicatorReader
  #- statement: PRINT sfServiceIndicatorReaderCount

- statement: (serviceIndicatorToUpdate) => SELECT 
                                              sf.Id,
                                              m.Contact__c,
                                              m.Service_Indicator_Code__c,
                                              m.Service_Indicator_Reason__c,
                                              m.Start_Date__c,

m.Start_Term__c,
                                              m.End_Term__c,
                                              m.Description__c,
                                              m.Department__c,
                                              m.Effect__c,
                                              m.Comments__c,
                                              m.Services_Impacted__c,
                                              m.Contact_Person__c,
                                              m.Placed_by__c
                                             FROM mappedServiceIndicator m
                                             INNER JOIN sfServiceIndicatorReader sf ON IFNULL(TRIM(m.PeopleSoft_Service_Indicator_Unique_Id__c), '') = IFNULL(TRIM(sf.PeopleSoft_Service_Indicator_Unique_Id__c), '')
  #- statement: PRINT serviceIndicatorToUpdate
  - statement: (serviceIndicatorToUpdateCount) => SELECT COUNT(*) FROM serviceIndicatorToUpdate
  #- statement: PRINT serviceIndicatorToUpdateCount

# m.End_Date__c,

# The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table)
  - statement: (serviceIndicatorToInsert) => SELECT 
                                              m.*
                                             FROM mappedServiceIndicator m
                                             LEFT ANTI JOIN sfServiceIndicatorReader sf ON IFNULL(TRIM(m.PeopleSoft_Service_Indicator_Unique_Id__c), '') = IFNULL(TRIM(sf.PeopleSoft_Service_Indicator_Unique_Id__c), '')

#- statement: PRINT serviceIndicatorToInsert
  - statement: (serviceIndicatorToInsertCount) => SELECT COUNT(*) FROM serviceIndicatorToInsert
  #- statement: PRINT serviceIndicatorToInsertCount

- statement: (updateResults, updateErrors) => UPDATE serviceIndicatorToUpdate INTO sfServiceIndicatorWriter
  #- statement: PRINT updateResults   
  #- statement: PRINT updateErrors

- statement: (insertResults, insertErrors) => INSERT serviceIndicatorToInsert INTO sfServiceIndicatorWriter
  #- statement: PRINT insertResults   
  #- statement: PRINT insertErrors

#################################### LOGS - Debugging purpuses ######################################

#- statement: PRINT peopleSoftServiceIndicatorDataCount
  #- statement: PRINT sfContactReaderCount
  #- statement: PRINT rejectedRecordByNonMatchingContactCount
  #- statement: PRINT sfTermReaderCount
  #- statement: PRINT completeInfoForServiceIndicatorCount
  #- statement: PRINT mappedServiceIndicatorCount
  #- statement: PRINT sfServiceIndicatorReaderCount
  #- statement: PRINT serviceIndicatorToUpdateCount
  #- statement: PRINT serviceIndicatorToInsertCount

###### Start - Insert Local file Contacts ######

# The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table)
  - statement: REFRESH sfServiceIndicatorReader
  - statement: |
      (existingServiceIndicatorInfo) =>
          SELECT
            sf.Id,
            m.*
          FROM mappedServiceIndicator m
          LEFT JOIN sfServiceIndicatorReader sf ON IFNULL(TRIM(m.PeopleSoft_Service_Indicator_Unique_Id__c), '') = IFNULL(TRIM(sf.PeopleSoft_Service_Indicator_Unique_Id__c), '')

- statement: INSERT existingServiceIndicatorInfo INTO finalMappedRecordFile

###### End - Insert Local file Contacts ######

###### Start - Update Dates env vars ######

## Update current date time in env variable 
  - statement: |
      (lastModifiedDateTime) => SELECT date_format(from_utc_timestamp(current_timestamp(), 'America/New_York'), "yyyy-MM-dd-HH.mm.ss.SSSSSS") AS `serviceIndicatorLastUpdatedDate`               
  - statement: UPDATE lastModifiedDateTime INTO env.vars

###### End - Update Dates env vars ######

# Add more statements to convert, join, aggregrate, transform, and integrate your data