tools/hive-bigquery/hive_to_bigquery/hive_component.py

# Copyright 2019 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Module to handle Hive related utilities like creating connection to Hive
database, executing a query, check whether the provided database & table
exists etc."""

import json
import logging
import os
import time
from dateutil.parser import parse
from uuid import uuid4

from pyhive import exc, hive
from thrift.transport import TTransport

from hive_to_bigquery import custom_exceptions
from hive_to_bigquery.utilities import calculate_time
from hive_to_bigquery.database_component import DatabaseComponent

logger = logging.getLogger('Hive2BigQuery')


class HiveComponent(DatabaseComponent):
    """Hive component to handle functions related to it.

    Has utilities which do Hive operations using the Hive connection, such as
    creating staging table, loading data into staging table, listing
    underlying files, getting information on data to migrate, migrating data
    to BigQuery, and checking for new data in the source Hive table etc.

    Attributes:
        host (str): Hive server host name.
        port (int): Port to be used.
        user (str): Hive user name.
        connection (pyhive.hive.Connection): Hive connection object.
    """
    def __init__(self, **kwargs):

        logger.debug("Initializing Hive Component")
        super(HiveComponent, self).__init__(**kwargs)

    def get_connection(self):
        """Gets connection to the Hive server.

        Returns:
            pyhive.hive.Connection: Hive Connection object.
        """

        logger.debug("Getting Hive Connection")
        try:
            connection = hive.connect(host=self.host,
                                      port=self.port,
                                      username=self.user)
            return connection
        except TTransport.TTransportException as error:
            logger.error("Failed to establish Hive connection")
            raise custom_exceptions.ConnectionError from error

    def get_cursor(self):
        """Gets the Hive cursor.

        Returns:
            pyhive.hive.Cursor: pyhive cursor object.
        """

        logger.debug("Getting Hive cursor")
        cursor = self.connection.cursor()
        return cursor

    def execute_query(self, query_cmds):
        """Executes Hive query and returns the results.

        Args:
            query_cmds (Union[List,str]): To be executed query/queries.

        Returns:
            List: Results of the query.
        """

        cursor = self.get_cursor()
        try:
            if isinstance(query_cmds, list):
                for query in query_cmds:
                    cursor.execute(query)
            else:
                cursor.execute(query_cmds)

            try:
                results = cursor.fetchall()
            except exc.ProgrammingError:
                results = []
            finally:
                return results

        except exc.OperationalError as error:
            logger.error("Hive Query {} execution failed".format(
                str(query_cmds)))
            raise custom_exceptions.HiveExecutionError from error

    def check_database_exists(self, database_name):
        """Checks whether the Hive database exists.

        Args:
            database_name (str): Hive database name.

        Returns:
            boolean : True, if database exists else False.
        """

        results = self.execute_query("SHOW DATABASES")
        for name in results:
            if database_name in name:
                return True
        return False

    def check_table_exists(self, database_name, table_name):
        """Checks whether the Hive table exists.

        Args:
            database_name (str): Hive database name.
            table_name (str): Hive table name.

        Returns:
            boolean : True, if table exists else False.
        """

        results = self.execute_query(
            "SHOW TABLES FROM {}".format(database_name))
        for name in results:
            if table_name in name:
                return True
        return False

    def get_table_location(self, database_name, table_name):
        """Returns the Hive table location.

        Args:
            database_name (str): Hive database name.
            table_name (str): Hive table name.

        Returns:
            str: Location of the Hive table.
        """

        queries = [
            "set hive.ddl.output.format=json",
            "desc extended {0}.{1}".format(database_name, table_name)
        ]
        results = self.execute_query(queries)
        location = json.loads(results[0][0])['tableInfo']['sd']['location']
        return location

    @staticmethod
    def list_hdfs_files(location):
        """Lists the underlying HDFS files with non-zero size.

        Args:
            location (str): Hive table location.

        Returns:
            List: List of the underlying data files.
        """

        file_name = "hdfs_files_{}.txt".format(uuid4())
        status_code = os.system("hdfs dfs -ls {0} > {1}".format(
            location, file_name))
        if status_code:
            logger.error("hdfs command execution failed")
            raise custom_exceptions.HDFSCommandError
        with open(file_name, "r") as file_content:
            content = file_content.readlines()
        os.remove(file_name)
        hdfs_files_list = []
        i = 0
        for i, line in enumerate(content):
            if line.startswith("Found "):
                break
        for j in range(i + 1, len(content)):
            size = content[j].split()[4]
            if size != '0':
                hdfs_files_list.append(content[j].split()[-1])

        return hdfs_files_list

    def list_partitions(self, database_name, table_name):
        """Gets information about the different partitions.

        Args:
            database_name (str): Hive database name.
            table_name (str): Hive table name.

        Returns:
            List: A list of dict elements containing information of every
                partition.
        """

        tracking_data = []
        queries = [
            "set hive.ddl.output.format=json",
            "SHOW PARTITIONS {0}.{1}".format(database_name, table_name)
        ]
        result_set = self.execute_query(queries)
        results = json.loads(result_set[0][0])['partitions']

        for item in results:
            # Form the WHERE clause by joining the partition column names and
            # their values
            clause = 'WHERE ' + ' AND '.join(
                partition['columnName'] + '=' + '"' +
                partition['columnValue'] + '"' for partition in item['values'])
            tracking_data.append({
                'table_name':
                'stage__{}__{}'.format(table_name.lower(),
                                       str(uuid4()).replace("-", "_")),
                'clause':
                clause
            })
        return tracking_data

    def get_hive_table_row_count(self, hive_table_model, clause=''):
        """Queries Hive table to get number of rows.

        Args:
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.
            clause (str): WHERE clause to filter the table on partitions, if any.

        Returns:
            int: Number of rows as an output from the query.
        """

        query = "SELECT COUNT(*) FROM {0}.{1} {2}".format(
            hive_table_model.db_name, hive_table_model.table_name, clause)
        results = self.execute_query(query)
        n_rows = results[0][0]
        return n_rows

    def get_info_on_data_to_migrate(self, hive_table_model):
        """Gets information on data to be migrated in case of first run of
        migration.

        Args:
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.

        Returns:
            List: A list of dict elements each containing information of the
                data that needs to be migrated.
        """

        if hive_table_model.is_partitioned is False:
            tracking_data = self.get_non_partition_table_info(hive_table_model)
        else:
            tracking_data = self.get_partition_table_info(hive_table_model)

        return tracking_data

    def get_non_partition_table_info(self, hive_table_model):
        """Gets information on data to be migrated in case of a non-partition
        table.

        Validates the incremental column (if any provided), queries the Hive
        table to get the minimum and maximum values of the column and sets
        the HiveTableModel attributes related to incremental column.

        Args:
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.

        Returns:
            List: A list of only one dict element containing the information
                of data to migrate.
        """

        tracking_data = list()

        if hive_table_model.inc_col is not None:
            logger.info("Validating given incremental column...")
            # If the provided incremental column is of timestamp/date type,
            # it cannot be validated by counting the number of rows.
            if hive_table_model.inc_col in hive_table_model.timestamp_type_col:
                logger.debug(
                    "Fetching minimum and maximum values of the timestamp "
                    "incremental column...")
                results = self.execute_query(
                    "SELECT MIN({0}),MAX({0}) FROM {1}.{2}".format(
                        hive_table_model.inc_col, hive_table_model.db_name,
                        hive_table_model.table_name))
                col_min, col_max = results[0]
                # Sets incremental attributes of hive_table_model.
                hive_table_model.inc_col_type = 'ts'

                logger.info(
                    "Incremental column {} found. Range - {} - {}".format(
                        hive_table_model.inc_col, col_min, col_max))

            # Validates the incremental column of int data type by comparing
            # the number of distinct values and number of rows.
            elif hive_table_model.inc_col in hive_table_model.int_type_col:

                logger.debug("Counting the total number of rows...")
                results = self.execute_query(
                    "SELECT COUNT(*) FROM {}.{}".format(
                        hive_table_model.db_name, hive_table_model.table_name))
                n_rows = results[0][0]
                logger.debug("Number of rows in the table: %d", n_rows)

                logger.debug(
                    "Fetching maximum value of the incremental column...")
                query = "SELECT COUNT(DISTINCT({0})),MIN({0}),MAX({0}) " \
                        "FROM {1}.{2}".format(
                            hive_table_model.inc_col, hive_table_model.db_name,
                            hive_table_model.table_name)
                results = self.execute_query(query)

                distinct_col_values, col_min, col_max = results[0]
                # Checks if number of distinct values matches the number of rows.
                if n_rows == distinct_col_values and (1 + col_max -
                                                      col_min == n_rows):
                    # Sets incremental attributes of hive_table_model.
                    hive_table_model.inc_col_type = 'int'

                    logger.info(
                        "Incremental column {} valid. Range - {} - {}".format(
                            hive_table_model.inc_col, col_min, col_max))
                else:
                    logger.error(
                        "Incremental column {0} not valid. Range - {1} - "
                        "{2}\nTry another incremental column or without "
                        "providing incremental column".format(
                            hive_table_model.inc_col, col_min, col_max))
                    raise custom_exceptions.IncrementalColumnError
            else:
                logger.error("Given incremental column is not present.")
                raise custom_exceptions.IncrementalColumnError

        if hive_table_model.is_inc_col_present:
            tracking_data.append({
                'table_name': hive_table_model.staging_table_name,
                'id': 1,
                'inc_col_min': col_min,
                'inc_col_max': col_max,
                'clause': ""
            })
        else:
            tracking_data.append({
                'table_name': hive_table_model.staging_table_name,
                'clause': ""
            })

        return tracking_data

    def get_partition_table_info(self, hive_table_model):
        """Gets information on data to be migrated in case of a partition table.

        Validates the incremental column (if any provided), queries the Hive
        table to get the minimum and maximum values of the column in every
        partition and sets the HiveTableModel attributes related to
        incremental column.

        Args:
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.

        Returns:
            List: A list of dict elements containing the information of data
            to migrate.
        """

        # Information about partitions.
        tracking_data = self.list_partitions(hive_table_model.db_name,
                                             hive_table_model.table_name)

        for item in tracking_data:
            # Id is set to 1 since the partition is migrated for the first time.
            item['id'] = 1
        if hive_table_model.inc_col is not None:
            logger.info("Validating given incremental column...")
            # If the provided incremental column is of timestamp/date type,
            # it cannot be validated by counting the number of rows.
            if hive_table_model.inc_col in hive_table_model.timestamp_type_col:
                logger.debug(
                    "Fetching minimum and maximum values of the timestamp "
                    "incremental column...")
                for item in tracking_data:
                    clause = item['clause']
                    results = self.execute_query(
                        "SELECT MIN({0}),MAX({0}) FROM {1}.{2} {3}".format(
                            hive_table_model.inc_col, hive_table_model.db_name,
                            hive_table_model.table_name, clause))
                    col_min, col_max = results[0]
                    item['inc_col_min'] = col_min
                    item['inc_col_max'] = col_max
                    # Setting incremental attributes of hive_table_model.
                    hive_table_model.inc_col_type = 'ts'
                    logger.info(
                        "Incremental column {} found in table {}. Range - {} "
                        "- {}".format(hive_table_model.inc_col, clause,
                                      col_min, col_max))
            # Validates the incremental column of int data type by comparing
            # the number of distinct values and number of rows.
            elif hive_table_model.inc_col in hive_table_model.int_type_col:
                # Sets incremental attributes of hive_table_model.
                hive_table_model.inc_col_type = 'int'
                n_rows = {}
                for data in tracking_data:
                    clause = data['clause']
                    logger.debug("Counting the number of rows %s ...", clause)
                    results = self.execute_query(
                        "SELECT COUNT(*) FROM {0}.{1} {2}".format(
                            hive_table_model.db_name,
                            hive_table_model.table_name, clause))
                    n_rows[clause] = results[0][0]
                    logger.debug("Number of rows in the table %s : %s", clause,
                                 n_rows[clause])
                for item in tracking_data:
                    clause = item['clause']
                    logger.debug(
                        "Fetching maximum value of the incremental column %s "
                        "...", clause)
                    query = "SELECT COUNT(DISTINCT({0})),MIN({0}),MAX({0}) " \
                            "FROM {1}.{2} {3}".format(
                                hive_table_model.inc_col,
                                hive_table_model.db_name,
                                hive_table_model.table_name, clause)
                    results = self.execute_query(query)
                    distinct_col_values, col_min, col_max = results[0]
                    # Checks if the number of distinct values matches the
                    # number of rows for every partition.
                    if n_rows[clause] == distinct_col_values and (
                            1 + col_max - col_min == n_rows[clause]):
                        # Sets incremental attributes of hive_table_model.
                        item['inc_col_min'] = col_min
                        item['inc_col_max'] = col_max
                        logger.debug(
                            "Incremental column %s found in table %s. Range - "
                            "%s - %s", hive_table_model.inc_col, clause,
                            col_min, col_max)

                    else:
                        logger.error(
                            "Incremental column {} not valid in partition {}. "
                            "Range - {} - {}\nTry another incremental column "
                            "or without providing incremental column".format(
                                hive_table_model.inc_col, clause, col_min,
                                col_max))
                        raise custom_exceptions.IncrementalColumnError
                if hive_table_model.is_inc_col_present:
                    logger.info("Incremental column {} found".format(
                        hive_table_model.inc_col))
            else:
                logger.error("Given incremental column is not present.")
                raise custom_exceptions.IncrementalColumnError
        return tracking_data

    def create_and_load_stage_table(self,
                                    hive_table_model,
                                    table_name,
                                    clause=''):
        """Creates Hive staging table and inserts data into it from the
        source table.

        Args:
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.
            table_name (str): Staging table name.
            clause (str): WHERE clause to filter the table (if any),
                and insert only the filtered data into the staging table.
        """

        logger.info("Staging for table " + table_name + "...")
        # Replaces TABLE_NAME_HERE place holder with staging table name.
        create_ddl_statement = hive_table_model.create_statement.replace(
            "TABLE_NAME_HERE", table_name)

        # Creates staging table.
        self.execute_query(create_ddl_statement)
        logger.debug("Table %s created in Hive. Inserting data...", table_name)
        start = time.time()

        # Inserts data into staging table.
        query = "INSERT OVERWRITE TABLE {} SELECT * FROM {}.{} {}".format(
            table_name, hive_table_model.db_name, hive_table_model.table_name,
            clause)
        logger.info(query)
        self.execute_query(query)

        end = time.time()
        time_hive_stage = calculate_time(start, end)
        logger.debug("Loaded data from %s into %s - Time taken - %s",
                     hive_table_model.table_name, table_name, time_hive_stage)

    def migrate_data(self, mysql_component, bq_component, gcs_component,
                     hive_table_model, bq_table_model, gcs_bucket_name,
                     table_data):
        """Invokes the function to migrate data based on whether the Hive
        table is partitioned.

        Args:
            mysql_component (:class:`MySQLComponent`): Instance of
                MySQLComponent to connect to MySQL.
            bq_component (:class:`BigQueryComponent`): Instance of
                BigQueryComponent to do BigQuery operations.
            gcs_component (:class:`GCSStorageComponent`): Instance of
                GCSStorageComponent to do GCS operations.
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.
            bq_table_model (:class:`BigQueryTableModel`): Wrapper to BigQuery
                table details.
            gcs_bucket_name (str): GCS bucket name.
            table_data (List): Information of data to migrate.
        """

        logger.debug("Populating tracking table..")

        if hive_table_model.is_partitioned is False:
            self.migrate_non_partition_table(mysql_component, bq_component,
                                             gcs_component, hive_table_model,
                                             bq_table_model, gcs_bucket_name,
                                             table_data)
        else:
            self.migrate_partition_table(mysql_component, bq_component,
                                         gcs_component, hive_table_model,
                                         bq_table_model, gcs_bucket_name,
                                         table_data)

    def migrate_non_partition_table(self, mysql_component, bq_component,
                                    gcs_component, hive_table_model,
                                    bq_table_model, gcs_bucket_name,
                                    table_data):
        """Migrates Hive data in case of a non-partitioned table.

        Invokes the function to create and load stage table, gets the staging
        table location, and lists down the underneath HDFS files. Updates the
        file paths in the tracking table and calls the function stage_to_gcs
        to copy files to GCS.

        Args:
            mysql_component (:class:`MySQLComponent`): Instance of
                MySQLComponent to connect to MySQL.
            bq_component (:class:`BigQueryComponent`): Instance of
                BigQueryComponent to do BigQuery operations.
            gcs_component (:class:`GCSStorageComponent`): Instance of
                GCSStorageComponent to do GCS operations.
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.
            bq_table_model (:class:`BigQueryTableModel`): Wrapper to BigQuery
                table details.
            gcs_bucket_name (str): GCS bucket name.
            table_data (List): Information of data to migrate.
        """

        table_name = table_data[0]['table_name']
        clause = table_data[0]['clause']
        insert_clause = clause
        if hive_table_model.is_inc_col_present:
            identifier = int(table_data[0]['id'])
            inc_col_min = table_data[0]['inc_col_min']
            inc_col_max = table_data[0]['inc_col_max']
            if identifier == 1:
                # Includes lower bound value in the stage table.
                insert_clause = "where {0}>='{1}' and {0}<='{2}'".format(
                    hive_table_model.inc_col, inc_col_min, inc_col_max)
            else:
                insert_clause = "where {0}>'{1}' and {0}<='{2}'".format(
                    hive_table_model.inc_col, inc_col_min, inc_col_max)

        # Creating staging table and loading data.
        if hive_table_model.is_table_type_supported is False:
            self.create_and_load_stage_table(hive_table_model, table_name,
                                             insert_clause)
            source_location = self.get_table_location("default", table_name)
        else:
            if hive_table_model.is_inc_col_present and \
                    hive_table_model.is_first_run is False:
                self.create_and_load_stage_table(hive_table_model, table_name,
                                                 insert_clause)
                source_location = self.get_table_location(
                    "default", table_name)
            else:
                source_location = self.get_table_location(
                    hive_table_model.db_name, hive_table_model.table_name)
        # Lists underlying HDFS files.
        hdfs_files_list = self.list_hdfs_files(source_location)
        logger.info("Updating file paths in the tracking table..")
        for file_path in hdfs_files_list:
            if hive_table_model.is_inc_col_present:
                query = "INSERT INTO {0} (id,table_name,inc_col_min," \
                        "inc_col_max,clause,file_path,gcs_copy_status," \
                        "bq_job_id,bq_job_retries,bq_job_status) VALUES({1}," \
                        "'{2}','{3}','{4}','{5}','{6}','TODO','TODO',0," \
                        "'TODO')".format(
                            hive_table_model.tracking_table_name, identifier,
                            table_name, inc_col_min, inc_col_max, clause,
                            file_path)
            else:
                query = "INSERT INTO {0} (table_name,clause,file_path," \
                        "gcs_copy_status,bq_job_id,bq_job_retries," \
                        "bq_job_status) VALUES('{1}','{2}','{3}','TODO'," \
                        "'TODO',0,'TODO')".format(
                            hive_table_model.tracking_table_name, table_name,
                            clause, file_path)
            # Commits information about the staging files.
            mysql_component.execute_transaction(query)
        # Copies files from HDFS to GCS.
        gcs_component.stage_to_gcs(mysql_component, bq_component,
                                   hive_table_model, bq_table_model,
                                   gcs_bucket_name)

    def migrate_partition_table(self, mysql_component, bq_component,
                                gcs_component, hive_table_model,
                                bq_table_model, gcs_bucket_name, table_data):
        """Migrates Hive data in case of a partition table.

        Invokes the function to create and load stage table, gets the staging
        table location, and lists down the underneath HDFS files. Updates the
        file paths in the tracking table and calls the function stage_to_gcs
        to copy files to GCS.

        Args:
            mysql_component (:class:`MySQLComponent`): Instance of
                MySQLComponent to connect to MySQL.
            bq_component (:class:`BigQueryComponent`): Instance of
                BigQueryComponent to do BigQuery operations.
            gcs_component (:class:`GCSStorageComponent`): Instance of
                GCSStorageComponent to do GCS operations.
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.
            bq_table_model (:class:`BigQueryTableModel`): Wrapper to BigQuery
                table details.
            gcs_bucket_name (str): GCS bucket name.
            table_data (List): Information of data to migrate.
        """

        for data in table_data:
            if hive_table_model.is_inc_col_present:
                insert_query = "INSERT INTO {0} (id,table_name,inc_col_min," \
                               "inc_col_max,clause,file_path) VALUES('{1}'," \
                               "'{2}','{3}','{4}','{5}','TODO')".format(
                                   hive_table_model.tracking_table_name,
                                   data['id'], data['table_name'],
                                   data['inc_col_min'], data['inc_col_max'],
                                   data['clause'])
                select_query = "SELECT id,table_name,inc_col_min,inc_col_max," \
                               "clause FROM {} WHERE file_path='TODO'".format(
                                   hive_table_model.tracking_table_name)
            else:
                insert_query = "INSERT INTO {0} (table_name,clause," \
                               "file_path)VALUES('{1}','{2}','TODO')".format(
                                   hive_table_model.tracking_table_name,
                                   data['table_name'], data['clause'])
                select_query = "SELECT table_name,clause FROM {} WHERE " \
                               "file_path='TODO'".format(
                                   hive_table_model.tracking_table_name)
            # Inserts a row in the tracking table for every partition.
            mysql_component.execute_transaction(insert_query)
            results = mysql_component.execute_query(select_query)

            for row in results:
                if hive_table_model.is_inc_col_present:
                    identifier, table_name, inc_col_min, inc_col_max, clause \
                        = row
                    if identifier == 1:
                        insert_clause = "{0} and {1}>='{2}' and " \
                                        "{1}<='{3}'".format(
                                            clause, hive_table_model.inc_col,
                                            inc_col_min, inc_col_max)
                    else:
                        insert_clause = "{0} and {1}>'{2}' and " \
                                        "{1}<='{3}'".format(
                                            clause, hive_table_model.inc_col,
                                            inc_col_min, inc_col_max)
                else:
                    table_name, clause = row
                    insert_clause = clause
                # Creates staging table and inserting data.
                self.create_and_load_stage_table(hive_table_model, table_name,
                                                 insert_clause)
                # Gets table location
                source_location = self.get_table_location(
                    "default", table_name)
                # Lists underlying HDFS files.
                hdfs_files_list = self.list_hdfs_files(source_location)

                logger.info("Updating file paths in the tracking table..")
                for file_path in hdfs_files_list:
                    if hive_table_model.is_inc_col_present:
                        query = "INSERT INTO {0} (id,table_name,inc_col_min," \
                                "inc_col_max,clause,file_path," \
                                "gcs_copy_status,bq_job_id,bq_job_retries," \
                                "bq_job_status) VALUES('{1}','{2}','{3}'," \
                                "'{4}','{5}','{6}','TODO','TODO',0," \
                                "'TODO')".format(
                                    hive_table_model.tracking_table_name,
                                    identifier, table_name, inc_col_min,
                                    inc_col_max, clause, file_path)
                    else:
                        query = "INSERT INTO {0} (table_name,clause," \
                                "file_path,gcs_copy_status,bq_job_id," \
                                "bq_job_retries,bq_job_status) VALUES('{1}'," \
                                "'{2}','{3}','TODO','TODO',0,'TODO')".format(
                                    hive_table_model.tracking_table_name,
                                    table_name, clause, file_path)
                    # Commits information about the staging files.
                    mysql_component.execute_transaction(query)

                query = "DELETE FROM {0} WHERE table_name='{1}' AND clause " \
                        "='{2}' AND file_path='TODO'".format(
                            hive_table_model.tracking_table_name,
                            table_name, clause)
                mysql_component.execute_transaction(query)
                # Copies files from HDFS to GCS.
                gcs_component.stage_to_gcs(mysql_component, bq_component,
                                           hive_table_model, bq_table_model,
                                           gcs_bucket_name)

    @staticmethod
    def compare_max_values(hive_table_model, old_max, new_max):
        """Compares the previously obtained maximum value with the newly
        obtained maximum value of incremental column.

        Args:
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.
            old_max (str): Maximum value of the incremental column from the
                tracking table.
            new_max (str): Maximum value of the incremental column from the
                Hive table.

        Returns:
            boolean: True if new_max value is greater than the old_max value,
            else False.
        """

        if hive_table_model.inc_col_type == "ts":

            try:
                old_max = parse(old_max)
                new_max = parse(new_max)

            except ValueError as error:
                logger.exception(error)
                logger.info("Failed to detect incremental column type")
                raise

            if new_max > old_max:
                return True

        else:
            # incremental column is of int type.
            if int(new_max) > int(old_max):
                return True

        return False

    def check_inc_data(self, mysql_component, bq_component, gcs_component,
                       hive_table_model, bq_table_model, gcs_bucket_name):
        """Invokes the functions to check for incremental data.

        Args:
            mysql_component (:class:`MySQLComponent`): Instance of
                MySQLComponent to connect to MySQL.
            bq_component (:class:`BigQueryComponent`): Instance of
                BigQueryComponent to do BigQuery operations.
            gcs_component (:class:`GCSStorageComponent`): Instance of
                GCSStorageComponent to do GCS operations.
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.
            bq_table_model (:class:`BigQueryTableModel`): Wrapper to BigQuery
                table details.
            gcs_bucket_name (str): GCS bucket name.

        Returns:
            List: A list of dict elements each containing information of the
                incremental data that needs to be migrated.
        """

        logger.info("Checking for any new data...")
        if hive_table_model.is_partitioned is False:
            tracking_data = self.check_inc_non_partition_table(
                mysql_component, bq_component, gcs_component, hive_table_model,
                bq_table_model, gcs_bucket_name)
        else:
            tracking_data = self.check_inc_partition_table(
                mysql_component, hive_table_model)

        return tracking_data

    def check_inc_non_partition_table(self, mysql_component, bq_component,
                                      gcs_component, hive_table_model,
                                      bq_table_model, gcs_bucket_name):
        """Checks for incremental data in case of a non-partitioned table.

        If there is an incremental column, the function queries the Hive
        table to get the maximum value and compares it with the maximum value
        from the tracking table. If there is no incremental column then
        Case A: If the data format is supported [Avro/ORC/Parquet], it looks
        for new files.
        Case B: If the data format is not supported, it cannot detect new data.

        Args:
            mysql_component (:class:`MySQLComponent`): Instance of
                MySQLComponent to connect to MySQL.
            bq_component (:class:`BigQueryComponent`): Instance of
                BigQueryComponent to do BigQuery operations.
            gcs_component (:class:`GCSStorageComponent`): Instance of
                GCSStorageComponent to do GCS operations.
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.
            bq_table_model (:class:`BigQueryTableModel`): Wrapper to BigQuery
                table details.
            gcs_bucket_name (str): GCS bucket name.

        Returns:
            List: A list of only one dict element, containing information of
                the incremental data.
        """

        tracking_data = []
        if hive_table_model.is_inc_col_present:
            # Fetches maximum value of the incremental column for each
            # partition from the tracking table and from Hive table and
            # compare these values to decide whether there is new data.
            results = mysql_component.execute_query(
                "SELECT MAX(id),MAX(inc_col_max) FROM {}".format(
                    hive_table_model.tracking_table_name))
            identifier, old_data_max = results[0]

            results = self.execute_query("SELECT MAX({0}) FROM {1}.{2}".format(
                hive_table_model.inc_col, hive_table_model.db_name,
                hive_table_model.table_name))
            new_data_max = results[0][0]

            new_data_exists = self.compare_max_values(hive_table_model,
                                                      old_data_max,
                                                      new_data_max)
            if new_data_exists:
                logger.info("New data found in source table")
                logger.debug(
                    "Previously incremental column %s maximum value "
                    "%s.Current maximum value %s", hive_table_model.inc_col,
                    old_data_max, new_data_max)
                tracking_data.append({
                    'table_name': hive_table_model.staging_table_name,
                    'id': identifier + 1,
                    'inc_col_min': old_data_max,
                    'inc_col_max': new_data_max,
                    'clause': ""
                })
            else:
                logger.info("No new data found")
        elif not hive_table_model.is_inc_col_present and \
                hive_table_model.is_table_type_supported is False:
            logger.info(
                "cannot check for new data in case of Non partitioned - No "
                "Incremental column - Text format table")
        elif not hive_table_model.is_inc_col_present and \
                hive_table_model.is_table_type_supported is True:
            # Lists HDFS files and compares them with tracking table and
            # migrates files which aren't present in the tracking table.
            results = mysql_component.execute_query(
                "SELECT file_path FROM {}".format(
                    hive_table_model.tracking_table_name))
            old_file_paths = [row[0] for row in results]
            new_file_paths = self.list_hdfs_files(
                self.get_table_location(hive_table_model.db_name,
                                        hive_table_model.table_name))

            new_data_exists = False
            for file_path in new_file_paths:
                if file_path not in old_file_paths:
                    # Updates the tracking table with new file paths.
                    new_data_exists = True
                    logger.debug("Found new data at file path %s", file_path)
                    query = "INSERT INTO {0} (table_name,file_path," \
                            "gcs_copy_status,bq_job_id,bq_job_retries," \
                            "bq_job_status) VALUES('{1}','{2}','TODO','TODO'," \
                            "0,'TODO')".format(
                                hive_table_model.tracking_table_name,
                                hive_table_model.table_name, file_path)
                    mysql_component.execute_transaction(query)
            # Copies the new files to GCS.
            if new_data_exists:
                logger.info("New files found in source table")
                gcs_component.stage_to_gcs(mysql_component, bq_component,
                                           hive_table_model, bq_table_model,
                                           gcs_bucket_name)
            else:
                logger.info("No new data found")

        return tracking_data

    def check_inc_partition_table(self, mysql_component, hive_table_model):
        """Checks for incremental data in case of a partition table.

        If there is no incremental column, this function queries the Hive
        table to get the list of new partitions, if any. If there is an
        incremental column, in addition to finding new partitions, this also
        gets the information of the incremental data in existing partitions.

        Args:
            mysql_component (:class:`MySQLComponent`): Instance of
                MySQLComponent to connect to MySQL.
            hive_table_model (:class:`HiveTableModel`): Wrapper to Hive table
                details.

        Returns:
            List: A list of only one dict element containing information of
                the incremental data.
        """

        tracking_data = []
        # Checks for any new partitions which haven't been recorded in
        # the tracking table and appends information to the list.
        logger.info("Checking for new partitions...")
        results = mysql_component.execute_query(
            "SELECT DISTINCT(clause) FROM {}".format(
                hive_table_model.tracking_table_name))
        old_partitions_list = [row[0] for row in results]

        partitions_list = self.list_partitions(hive_table_model.db_name,
                                               hive_table_model.table_name)
        present_partitions_list = [item['clause'] for item in partitions_list]

        new_partitions_list = list(
            set(present_partitions_list).difference(old_partitions_list))

        if not hive_table_model.is_inc_col_present:
            for clause in new_partitions_list:
                logger.info("Found new partition {}".format(clause))
                tracking_data.append({
                    'table_name': hive_table_model.staging_table_name,
                    'clause': clause
                })

        else:
            for clause in new_partitions_list:

                logger.info("Found new partition {}".format(clause))
                results = self.execute_query(
                    "SELECT MIN({0}),MAX({0}) FROM {1}.{2} {3}".format(
                        hive_table_model.inc_col, hive_table_model.db_name,
                        hive_table_model.table_name, clause))
                col_min, col_max = results[0]
                tracking_data.append({
                    "table_name": hive_table_model.staging_table_name,
                    "id": 1,
                    "inc_col_min": col_min,
                    "inc_col_max": col_max,
                    "clause": clause
                })

            logger.info("Checking for new data in existing partitions...")
            # Fetches maximum value of the incremental column for each
            # partition from the tracking table and from Hive table and
            # compare these values to decide whether there is new data.
            for clause in old_partitions_list:

                results = mysql_component.execute_query(
                    "SELECT MAX(id),MAX(inc_col_max) FROM {0} WHERE "
                    "clause='{1}'".format(hive_table_model.tracking_table_name,
                                          clause))
                identifier, old_data_max = results[0]
                logger.debug("Old maximum value %s - %s", clause, old_data_max)

                results = self.execute_query(
                    "SELECT MAX({0}) FROM {1}.{2} {3}".format(
                        hive_table_model.inc_col, hive_table_model.db_name,
                        hive_table_model.table_name, clause))
                new_data_max = results[0][0]
                logger.debug("New maximum value %s - %s", clause, new_data_max)

                new_data_exists = self.compare_max_values(
                    hive_table_model, old_data_max, new_data_max)
                # Appends information to the list if new data is found.
                if new_data_exists:
                    logger.info(
                        "New data found in partition - {}".format(clause))
                    tracking_data.append({
                        "table_name": hive_table_model.staging_table_name,
                        "id": identifier + 1,
                        "inc_col_min": old_data_max,
                        "inc_col_max": new_data_max,
                        "clause": clause
                    })
                else:
                    logger.info(
                        "No New data found in partition - {}".format(clause))

        return tracking_data