<commit_before>/*
 * Copyright (c) 2016. Teradata Inc.
 */

package com.thinkbiganalytics.ingest;


import com.thinkbiganalytics.util.PartitionBatch;
import com.thinkbiganalytics.util.PartitionSpec;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.Vector;

/**
 * Merge from a table into a target table. Dedupes and uses partition strategy of the target table
 */
public class TableMergeSupport implements Serializable {

    public static Logger logger = LoggerFactory.getLogger(TableRegisterSupport.class);

    private Connection conn;

    public TableMergeSupport(Connection conn) {
        Validate.notNull(conn);
        this.conn = conn;
    }

    protected TableMergeSupport() {
        // for unit testing
    }

    /**
     * Performs the doMerge and insert into the target table from the source table
     *
     * @param sourceTable      the source table
     * @param targetTable      the target table
     * @param partitionSpec    the partition specification
     * @param feedPartionValue the source processing partition value
     * @param shouldDedupe     whether to perform dedupe during merge
     */
    public List<PartitionBatch> doMerge(String sourceTable, String targetTable, PartitionSpec partitionSpec, String feedPartionValue, boolean shouldDedupe) {

        Validate.notEmpty(sourceTable);
        Validate.notEmpty(targetTable);
        Validate.notNull(partitionSpec);
        Validate.notNull(feedPartionValue);

        String[] selectFields = getSelectFields(sourceTable, targetTable, partitionSpec);
        if (partitionSpec.isNonPartitioned()) {
            String sql = generateDedupeNonPartitionQuery(selectFields, sourceTable, targetTable, feedPartionValue, shouldDedupe);
            doMerge(sql);
        } else {

            List<PartitionBatch> batches = createPartitionBatches(partitionSpec, sourceTable, feedPartionValue);
            if (batches.size() > 0) {
                logger.info("{} batches will be executed", batches.size());
                for (PartitionBatch batch : batches) {
                    String sql = generateDedupePartitionQuery(selectFields, partitionSpec, batch.getPartionValues(), sourceTable, targetTable, feedPartionValue, shouldDedupe);
                    doMerge(sql);
                }
                return batches;
            } else {
                logger.warn("No valid data found.");
            }
        }
        return null;
    }

    /**
     * Generates a query for inserting from a source table into the target table with no partitions
     *
     * @param selectFields the list of fields in the select clause of the source table
     * @param sourceTable  the source table
     * @param targetTable  the target table
     * @param shouldDedupe whether to deduplicate the results
     * @return the sql string
     */
    protected String generateDedupeNonPartitionQuery(String[] selectFields, String sourceTable, String targetTable, String feedPartitionValue, boolean shouldDedupe) {

        String selectSQL = StringUtils.join(selectFields, ",");

        StringBuffer sb = new StringBuffer();
        sb.append("insert overwrite table ").append(targetTable).append(" ");

        if (shouldDedupe) {
            sb.append(" select ").append(selectSQL).append(" from (");
        }

        sb.append(" select ").append(selectSQL)
            .append(" from ").append(sourceTable).append(" where processing_dttm='" + feedPartitionValue + "' ")
            .append(" union ")
            .append(" select ").append(selectSQL)
            .append(" from ").append(targetTable);

        if (shouldDedupe) {
            sb.append(") x group by ").append(selectSQL);
        }

        return sb.toString();
    }

    /**
     * Generates a query for inserting from a source table into the target table adhering to partitions
     *
     * @param selectFields    the list of fields in the select clause of the source table
     * @param spec            the partition specification or null if none
     * @param partitionValues the values containing the distinct partition data to process this iterator
     * @param sourceTable     the source table
     * @param targetTable     the target table
     * @param shouldDedupe    whether to deduplicate the results
     * @return the sql string
     */
    protected String generateDedupePartitionQuery(String[] selectFields, PartitionSpec spec, String[] partitionValues, String sourceTable, String targetTable, String feedPartitionValue, boolean
        shouldDedupe) {

        String selectSQL = StringUtils.join(selectFields, ",");
        String targetSqlWhereClause = spec.toTargetSQLWhere(partitionValues);
        String sourceSqlWhereClause = spec.toSourceSQLWhere(partitionValues);
        String partitionClause = spec.toPartitionSpec(partitionValues);

        StringBuffer sb = new StringBuffer();
        sb.append("insert overwrite table ").append(targetTable).append(" ")
            .append(partitionClause);

        if (shouldDedupe) {
            sb.append(" select ").append(selectSQL).append(" from (");
        }

        sb.append(" select ").append(selectSQL)
            .append(" from ").append(sourceTable).append(" ")
            .append(" where ")
            .append(" processing_dttm='" + feedPartitionValue + "' and ")
            .append(sourceSqlWhereClause)
            .append(" union ")
            .append(" select ").append(selectSQL)
            .append(" from ").append(targetTable).append(" ")
            .append(" where ")
            .append(targetSqlWhereClause);

        if (shouldDedupe) {
            sb.append(") x group by ").append(selectSQL);
        }

        return sb.toString();
    }

    /**
     * Generates a query for merging data from a source table into the target table adhering to partitions and using a last modify date and a primary key
     *
     * @param id              the primary key
     * @param lastModifyField the last modify date field used to determine
     * @param selectFields    the list of fields in the select clause of the source table
     * @param spec            the partition specification or null if none
     * @param partitionValues the values containing the distinct partition data to process this iterator
     * @param sourceTable     the source table
     * @param targetTable     the target table
     * @return the sql string
     */
    protected String generateDedupePartitionQueryPK(String id, String lastModifyField, String[] selectFields, PartitionSpec spec, String[] partitionValues, String sourceTable, String targetTable,
                                                    String feedPartitionValue) {
<commit_msg>PC-154<commit_after>/*
 * Copyright (c) 2016. Teradata Inc.
 */

package com.thinkbiganalytics.ingest;


import com.thinkbiganalytics.util.PartitionBatch;
import com.thinkbiganalytics.util.PartitionSpec;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.Vector;

/**
 * Merge from a table into a target table. Dedupes and uses partition strategy of the target table
 */
public class TableMergeSupport implements Serializable {

    public static Logger logger = LoggerFactory.getLogger(TableRegisterSupport.class);

    private Connection conn;

    public TableMergeSupport(Connection conn) {
        Validate.notNull(conn);
        this.conn = conn;
    }

    protected TableMergeSupport() {
        // for unit testing
    }

    /**
     * Performs the doMerge and insert into the target table from the source table
     *
     * @param sourceTable      the source table
     * @param targetTable      the target table
     * @param partitionSpec    the partition specification
     * @param feedPartionValue the source processing partition value
     * @param shouldDedupe     whether to perform dedupe during merge
     */
    public List<PartitionBatch> doMerge(String sourceTable, String targetTable, PartitionSpec partitionSpec, String feedPartionValue, boolean shouldDedupe) {

        Validate.notEmpty(sourceTable);
        Validate.notEmpty(targetTable);
        Validate.notNull(partitionSpec);
        Validate.notNull(feedPartionValue);

        String[] selectFields = getSelectFields(sourceTable, targetTable, partitionSpec);
        if (partitionSpec.isNonPartitioned()) {
            String sql = generateDedupeNonPartitionQuery(selectFields, sourceTable, targetTable, feedPartionValue, shouldDedupe);
            doMerge(sql);
        } else {

            List<PartitionBatch> batches = createPartitionBatches(partitionSpec, sourceTable, feedPartionValue);
            if (batches.size() > 0) {
                logger.info("{} batches will be executed", batches.size());
                for (PartitionBatch batch : batches) {
                    String sql = generateDedupePartitionQuery(selectFields, partitionSpec, batch.getPartionValues(), sourceTable, targetTable, feedPartionValue, shouldDedupe);
                    doMerge(sql);
                }
                return batches;
            } else {
                logger.warn("No valid data found.");
            }
        }
        return null;
    }

    /**
     * Generates a query for inserting from a source table into the target table with no partitions
     *
     * @param selectFields the list of fields in the select clause of the source table
     * @param sourceTable  the source table
     * @param targetTable  the target table
     * @param shouldDedupe whether to deduplicate the results
     * @return the sql string
     */
    protected String generateDedupeNonPartitionQuery(String[] selectFields, String sourceTable, String targetTable, String feedPartitionValue, boolean shouldDedupe) {

        String selectSQL = StringUtils.join(selectFields, ",");

        StringBuffer sb = new StringBuffer();
        sb.append("insert overwrite table ").append(targetTable).append(" ");

        if (shouldDedupe) {
            sb.append(" select ").append(selectSQL).append(" from (");
        }

        sb.append(" select ").append(selectSQL)
            .append(" from ").append(sourceTable).append(" where processing_dttm='" + feedPartitionValue + "' ")
            .append(" union all ")
            .append(" select ").append(selectSQL)
            .append(" from ").append(targetTable);

        if (shouldDedupe) {
            sb.append(") x group by ").append(selectSQL);
        }

        return sb.toString();
    }

    /**
     * Generates a query for inserting from a source table into the target table adhering to partitions
     *
     * @param selectFields    the list of fields in the select clause of the source table
     * @param spec            the partition specification or null if none
     * @param partitionValues the values containing the distinct partition data to process this iterator
     * @param sourceTable     the source table
     * @param targetTable     the target table
     * @param shouldDedupe    whether to deduplicate the results
     * @return the sql string
     */
    protected String generateDedupePartitionQuery(String[] selectFields, PartitionSpec spec, String[] partitionValues, String sourceTable, String targetTable, String feedPartitionValue, boolean
        shouldDedupe) {

        String selectSQL = StringUtils.join(selectFields, ",");
        String targetSqlWhereClause = spec.toTargetSQLWhere(partitionValues);
        String sourceSqlWhereClause = spec.toSourceSQLWhere(partitionValues);
        String partitionClause = spec.toPartitionSpec(partitionValues);

        StringBuffer sb = new StringBuffer();
        sb.append("insert overwrite table ").append(targetTable).append(" ")
            .append(partitionClause);

        if (shouldDedupe) {
            sb.append(" select ").append(selectSQL).append(" from (");
        }

        sb.append(" select ").append(selectSQL)
            .append(" from ").append(sourceTable).append(" ")
            .append(" where ")
            .append(" processing_dttm='" + feedPartitionValue + "' and ")
            .append(sourceSqlWhereClause)
            .append(" union all ")
            .append(" select ").append(selectSQL)
            .append(" from ").append(targetTable).append(" ")
            .append(" where ")
            .append(targetSqlWhereClause);

        if (shouldDedupe) {
            sb.append(") x group by ").append(selectSQL);
        }

        return sb.toString();
    }

    /**
     * Generates a query for merging data from a source table into the target table adhering to partitions and using a last modify date and a primary key
     *
     * @param id              the primary key
     * @param lastModifyField the last modify date field used to determine
     * @param selectFields    the list of fields in the select clause of the source table
     * @param spec            the partition specification or null if none
     * @param partitionValues the values containing the distinct partition data to process this iterator
     * @param sourceTable     the source table
     * @param targetTable     the target table
     * @return the sql string
     */
    protected String generateDedupePartitionQueryPK(String id, String lastModifyField, String[] selectFields, PartitionSpec spec, String[] partitionValues, String sourceTable, String targetTable,
                                                    String feedPartitionValue) {

{'fraction_non_alphanumeric': 0.08552111055167949, 'fraction_numerical': 0.001510465367186938, 'mean_word_length': 2.6998403406067055, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2899602', 'n_tokens_mistral': 3493, 'n_tokens_neox': 3396, 'n_words': 1106}