src/RedshiftBase.php
private function insertOrUpdateTargetTable(
        string $stagingTempTableName,
        string $targetTableName,
        array $primaryKey,
        array $columns,
        bool $useTimestamp = true,
        array $convertEmptyValuesToNull = []
    ): void {
        $this->connection->beginTransaction();
        $nowFormatted = $this->getNowFormatted();

        $targetTableNameWithSchema = $this->nameWithSchemaEscaped($targetTableName);
        $stagingTableNameEscaped = $this->tableNameEscaped($stagingTempTableName);

        if (!empty($primaryKey)) {
            // Update target table
            $sql = "UPDATE " . $targetTableNameWithSchema . " SET ";

            $columnsSet = [];
            foreach ($columns as $columnName) {
                $columnsSet[] = sprintf(
                    "%s = %s.%s",
                    $this->quoteIdentifier($columnName),
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($columnName)
                );
            }

            $sql .= implode(', ', $columnsSet);
            if ($useTimestamp) {
                $sql .= ", _timestamp = '{$nowFormatted}' ";
            }
            $sql .= " FROM " . $stagingTableNameEscaped . " ";
            $sql .= " WHERE ";

            $pkWhereSql = [];
            foreach ($primaryKey as $pkColumn) {
                $pkWhereSql[] = sprintf(
                    "%s.%s = %s.%s",
                    $targetTableNameWithSchema,
                    $this->quoteIdentifier($pkColumn),
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($pkColumn)
                );
            }

            $sql .= implode(' AND ', $pkWhereSql) . " ";

            // update only changed rows - mysql TIMESTAMP ON UPDATE behaviour simulation
            $columnsComparsionSql = array_map(function ($columnName) use (
                $targetTableNameWithSchema,
                $stagingTableNameEscaped
            ) {
                return sprintf(
                    "COALESCE(CAST(%s.%s as varchar), '') != COALESCE(CAST(%s.%s as varchar), '')",
                    $targetTableNameWithSchema,
                    $this->quoteIdentifier($columnName),
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($columnName)
                );
            }, $columns);
            $sql .= " AND (" . implode(' OR ', $columnsComparsionSql) . ") ";

            Debugger::timer('updateTargetTable');
            $this->query($sql);
            $this->addTimer('updateTargetTable', Debugger::timer('updateTargetTable'));

            // Delete updated rows from staging table
            $sql = "DELETE FROM " . $stagingTableNameEscaped . " ";
            $sql .= "USING " . $targetTableNameWithSchema . " ";
            $sql .= "WHERE " . implode(' AND ', $pkWhereSql);

            Debugger::timer('deleteUpdatedRowsFromStaging');
            $this->query($sql);
            $this->addTimer('deleteUpdatedRowsFromStaging', Debugger::timer('deleteUpdatedRowsFromStaging'));

            // Dedup staging table
            Debugger::timer('dedupStaging');
            $this->dedup($stagingTempTableName, $columns, $primaryKey);
            $this->addTimer('dedupStaging', Debugger::timer('dedupStaging'));
        }

        // Insert from staging to target table
        $sql = "INSERT INTO " . $targetTableNameWithSchema . " (" . implode(', ', array_map(function ($column) {
            return $this->quoteIdentifier($column);
        }, $columns));

        $sql .= ($useTimestamp) ? ", _timestamp) " : ")";

        $columnsSetSql = [];

        foreach ($columns as $columnName) {
            if (in_array($columnName, $convertEmptyValuesToNull)) {
                $columnsSetSql[] = sprintf(
                    "CASE %s.%s::VARCHAR WHEN '' THEN NULL ELSE %s.%s END",
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($columnName),
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($columnName)
                );
            } else {
                $columnsSetSql[] = sprintf(
                    "%s.%s",
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($columnName)
                );
            }
        }
        $sql .= "SELECT " . implode(', ', $columnsSetSql);
        $sql .= ($useTimestamp) ? ", '{$nowFormatted}'" : "";
        $sql .= " FROM " . $stagingTableNameEscaped;
        Debugger::timer('insertIntoTargetFromStaging');
        $this->query($sql);
        $this->addTimer('insertIntoTargetFromStaging', Debugger::timer('insertIntoTargetFromStaging'));

        $this->connection->commit();
    }
private function insertOrUpdateTargetTable(
        string $stagingTempTableName,
        string $targetTableName,
        array $primaryKey,
        array $columns,
        bool $useTimestamp = true,
        array $convertEmptyValuesToNull = []
    ): void {
        $this->connection->beginTransaction();
        $nowFormatted = $this->getNowFormatted();

        $targetTableNameWithSchema = $this->nameWithSchemaEscaped($targetTableName);
        $stagingTableNameEscaped = $this->tableNameEscaped($stagingTempTableName);

        if (!empty($primaryKey)) {
            // Update target table
            $sql = "UPDATE " . $targetTableNameWithSchema . " SET ";

            $columnsSet = [];
            foreach ($columns as $columnName) {
                $columnsSet[] = sprintf(
                    "%s = %s.%s",
                    $this->quoteIdentifier($columnName),
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($columnName)
                );
            }

            $sql .= implode(', ', $columnsSet);
            if ($useTimestamp) {
                $sql .= ", _timestamp = '{$nowFormatted}' ";
            }
            $sql .= " FROM " . $stagingTableNameEscaped . " ";
            $sql .= " WHERE ";

            $pkWhereSql = [];
            foreach ($primaryKey as $pkColumn) {
                $pkWhereSql[] = sprintf(
                    "%s.%s = %s.%s",
                    $targetTableNameWithSchema,
                    $this->quoteIdentifier($pkColumn),
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($pkColumn)
                );
            }

            $sql .= implode(' AND ', $pkWhereSql) . " ";

            // update only changed rows - mysql TIMESTAMP ON UPDATE behaviour simulation
            $columnsComparsionSql = array_map(function ($columnName) use (
                $targetTableNameWithSchema,
                $stagingTableNameEscaped
            ) {
                return sprintf(
                    "COALESCE(CAST(%s.%s as varchar), '') != COALESCE(CAST(%s.%s as varchar), '')",
                    $targetTableNameWithSchema,
                    $this->quoteIdentifier($columnName),
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($columnName)
                );
            }, $columns);
            $sql .= " AND (" . implode(' OR ', $columnsComparsionSql) . ") ";

            Debugger::timer('updateTargetTable');
            $this->query($sql);
            $this->addTimer('updateTargetTable', Debugger::timer('updateTargetTable'));

            // Delete updated rows from staging table
            $sql = "DELETE FROM " . $stagingTableNameEscaped . " ";
            $sql .= "USING " . $targetTableNameWithSchema . " ";
            $sql .= "WHERE " . implode(' AND ', $pkWhereSql);

            Debugger::timer('deleteUpdatedRowsFromStaging');
            $this->query($sql);
            $this->addTimer('deleteUpdatedRowsFromStaging', Debugger::timer('deleteUpdatedRowsFromStaging'));

            // Dedup staging table
            Debugger::timer('dedupStaging');
            $this->dedup($stagingTempTableName, $columns, $primaryKey);
            $this->addTimer('dedupStaging', Debugger::timer('dedupStaging'));
        }

        // Insert from staging to target table
        $sql = "INSERT INTO " . $targetTableNameWithSchema . " (" . implode(', ', array_map(function ($column) {
            return $this->quoteIdentifier($column);
        }, $columns));

        $sql .= ($useTimestamp) ? ", _timestamp) " : ")";

        $columnsSetSql = [];

        foreach ($columns as $columnName) {
            if (in_array($columnName, $convertEmptyValuesToNull)) {
                $columnsSetSql[] = sprintf(
                    "CASE %s.%s::VARCHAR WHEN '' THEN NULL ELSE %s.%s END",
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($columnName),
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($columnName)
                );
            } else {
                $columnsSetSql[] = sprintf(
                    "%s.%s",
                    $stagingTableNameEscaped,
                    $this->quoteIdentifier($columnName)
                );
            }
        }
        $sql .= "SELECT " . implode(', ', $columnsSetSql);
        $sql .= ($useTimestamp) ? ", '{$nowFormatted}'" : "";
        $sql .= " FROM " . $stagingTableNameEscaped;
        Debugger::timer('insertIntoTargetFromStaging');
        $this->query($sql);
        $this->addTimer('insertIntoTargetFromStaging', Debugger::timer('insertIntoTargetFromStaging'));

        $this->connection->commit();
    }
private function insertOrUpdateTargetTable ( string $ stagingTempTableName , string $ targetTableName , array $ primaryKey , array $ columns , bool $ useTimestamp = true , array $ convertEmptyValuesToNull = [ ] ) : void { $ this -> connection -> beginTransaction ( ) ; $ nowFormatted = $ this -> getNowFormatted ( ) ; $ targetTableNameWithSchema = $ this -> nameWithSchemaEscaped ( $ targetTableName ) ; $ stagingTableNameEscaped = $ this -> tableNameEscaped ( $ stagingTempTableName ) ; if ( ! empty ( $ primaryKey ) ) { // Update target table $ sql = "UPDATE " . $ targetTableNameWithSchema . " SET " ; $ columnsSet = [ ] ; foreach ( $ columns as $ columnName ) { $ columnsSet [ ] = sprintf ( "%s = %s.%s" , $ this -> quoteIdentifier ( $ columnName ) , $ stagingTableNameEscaped , $ this -> quoteIdentifier ( $ columnName ) ) ; } $ sql .= implode ( ', ' , $ columnsSet ) ; if ( $ useTimestamp ) { $ sql .= ", _timestamp = '{$nowFormatted}' " ; } $ sql .= " FROM " . $ stagingTableNameEscaped . " " ; $ sql .= " WHERE " ; $ pkWhereSql = [ ] ; foreach ( $ primaryKey as $ pkColumn ) { $ pkWhereSql [ ] = sprintf ( "%s.%s = %s.%s" , $ targetTableNameWithSchema , $ this -> quoteIdentifier ( $ pkColumn ) , $ stagingTableNameEscaped , $ this -> quoteIdentifier ( $ pkColumn ) ) ; } $ sql .= implode ( ' AND ' , $ pkWhereSql ) . " " ; // update only changed rows - mysql TIMESTAMP ON UPDATE behaviour simulation $ columnsComparsionSql = array_map ( function ( $ columnName ) use ( $ targetTableNameWithSchema , $ stagingTableNameEscaped ) { return sprintf ( "COALESCE(CAST(%s.%s as varchar), '') != COALESCE(CAST(%s.%s as varchar), '')" , $ targetTableNameWithSchema , $ this -> quoteIdentifier ( $ columnName ) , $ stagingTableNameEscaped , $ this -> quoteIdentifier ( $ columnName ) ) ; } , $ columns ) ; $ sql .= " AND (" . implode ( ' OR ' , $ columnsComparsionSql ) . ") " ; Debugger :: timer ( 'updateTargetTable' ) ; $ this -> query ( $ sql ) ; $ this -> addTimer ( 'updateTargetTable' , Debugger :: timer ( 'updateTargetTable' ) ) ; // Delete updated rows from staging table $ sql = "DELETE FROM " . $ stagingTableNameEscaped . " " ; $ sql .= "USING " . $ targetTableNameWithSchema . " " ; $ sql .= "WHERE " . implode ( ' AND ' , $ pkWhereSql ) ; Debugger :: timer ( 'deleteUpdatedRowsFromStaging' ) ; $ this -> query ( $ sql ) ; $ this -> addTimer ( 'deleteUpdatedRowsFromStaging' , Debugger :: timer ( 'deleteUpdatedRowsFromStaging' ) ) ; // Dedup staging table Debugger :: timer ( 'dedupStaging' ) ; $ this -> dedup ( $ stagingTempTableName , $ columns , $ primaryKey ) ; $ this -> addTimer ( 'dedupStaging' , Debugger :: timer ( 'dedupStaging' ) ) ; } // Insert from staging to target table $ sql = "INSERT INTO " . $ targetTableNameWithSchema . " (" . implode ( ', ' , array_map ( function ( $ column ) { return $ this -> quoteIdentifier ( $ column ) ; } , $ columns ) ) ; $ sql .= ( $ useTimestamp ) ? ", _timestamp) " : ")" ; $ columnsSetSql = [ ] ; foreach ( $ columns as $ columnName ) { if ( in_array ( $ columnName , $ convertEmptyValuesToNull ) ) { $ columnsSetSql [ ] = sprintf ( "CASE %s.%s::VARCHAR WHEN '' THEN NULL ELSE %s.%s END" , $ stagingTableNameEscaped , $ this -> quoteIdentifier ( $ columnName ) , $ stagingTableNameEscaped , $ this -> quoteIdentifier ( $ columnName ) ) ; } else { $ columnsSetSql [ ] = sprintf ( "%s.%s" , $ stagingTableNameEscaped , $ this -> quoteIdentifier ( $ columnName ) ) ; } } $ sql .= "SELECT " . implode ( ', ' , $ columnsSetSql ) ; $ sql .= ( $ useTimestamp ) ? ", '{$nowFormatted}'" : "" ; $ sql .= " FROM " . $ stagingTableNameEscaped ; Debugger :: timer ( 'insertIntoTargetFromStaging' ) ; $ this -> query ( $ sql ) ; $ this -> addTimer ( 'insertIntoTargetFromStaging' , Debugger :: timer ( 'insertIntoTargetFromStaging' ) ) ; $ this -> connection -> commit ( ) ; }
Performs merge operation according to http://docs.aws.amazon.com/redshift/latest/dg/merge-specify-a-column-list.html
Performs merge operation according to http : // docs . aws . amazon . com / redshift / latest / dg / merge - specify - a - column - list . html