src/dirbs/importer/base_delta_importer.py
def staging_row_count(self):
        """Overrides AbstractImporter.staging_row_count.

        This is NOT a completely accurate staging_row_count for performance reasons. Since this is only used
        in the import size check, which can be disabled, it has been decided to make a tradeoff between performance
        and accuracy. This is only inaccurate when:

        1. There are duplicate rows with the same primary keys and the same action in the delta table. A different
           action for the same primary key will have already failed the delta_multiple_changes check.
        2. An add is already in the DB or a remove is not in the DB. This will open happen if
           self._perform_delta_adds_check or self._perform_delta_removes_check is False when these checks would have
           failed if enabled.
        """
        assert self._postprocessed and \
            'BaseDeltaImporter::staging_row_count called on un-postprocessed/non-delta table'

        self._logger.info('Counting the approximate number of entries in staging table based on delta...')
        futures_list = []
        filter_str = "change_type != \'update\'"
        with futures.ThreadPoolExecutor(max_workers=self._max_db_connections) as executor:
            if self._supports_imei_shards:
                for name, rstart, rend in partition_utils.physical_imei_shards(self._conn,
                                                                               tbl_name=self._staging_tbl_name):
                    futures_list.append(executor.submit(self._count_changes_by_type_single_shard, name,
                                                        filter_str=filter_str))
            else:
                futures_list.append(executor.submit(self._count_changes_by_type_single_shard,
                                                    self._staging_tbl_name,
                                                    filter_str=filter_str))

        # All futures should be done at this point as with block is exited above
        changes_by_type = defaultdict(int)
        for f in futures_list:
            result = f.result()
            for k, v in result.items():
                changes_by_type[k] += v

        old_row_count = self.row_count
        approx_add_count = sum(v for k, v in changes_by_type.items() if k == 'add')
        approx_remove_count = sum(v for k, v in changes_by_type.items() if k == 'remove')
        approx_net_adds = approx_add_count - approx_remove_count
        approx_new_count = old_row_count + approx_net_adds
        self._logger.info('Found {0} approximate entries in staging table based on delta '
                          '({1} in existing table, {2} adds, {3} removes)'.format(approx_new_count,
                                                                                  old_row_count, approx_add_count,
                                                                                  approx_remove_count))

        return old_row_count + approx_net_adds
Overrides AbstractImporter.staging_row_count.

        This is NOT a completely accurate staging_row_count for performance reasons. Since this is only used
        in the import size check, which can be disabled, it has been decided to make a tradeoff between performance
        and accuracy. This is only inaccurate when:

        1. There are duplicate rows with the same primary keys and the same action in the delta table. A different
           action for the same primary key will have already failed the delta_multiple_changes check.
        2. An add is already in the DB or a remove is not in the DB. This will open happen if
           self._perform_delta_adds_check or self._perform_delta_removes_check is False when these checks would have
           failed if enabled.
        


1. There are duplicate rows with the same primary keys and the same action in the delta table. A different
action for the same primary key will have already failed the delta_multiple_changes check.
2. An add is already in the DB or a remove is not in the DB. This will open happen if
self._perform_delta_adds_check or self._perform_delta_removes_check is False when these checks would have
failed if enabled.
1 . There are duplicate rows with the same primary keys and the same action in the delta table . A different action for the same primary key will have already failed the delta_multiple_changes check . 2 . An add is already in the DB or a remove is not in the DB . This will open happen if self . _perform_delta_adds_check or self . _perform_delta_removes_check is False when these checks would have failed if enabled .
def staging_row_count(self):
        assert self._postprocessed and \
            'BaseDeltaImporter::staging_row_count called on un-postprocessed/non-delta table'
        self._logger.info('Counting the approximate number of entries in staging table based on delta...')
        futures_list = []
        filter_str = "change_type != \'update\'"
        with futures.ThreadPoolExecutor(max_workers=self._max_db_connections) as executor:
            if self._supports_imei_shards:
                for name, rstart, rend in partition_utils.physical_imei_shards(self._conn,
                                                                               tbl_name=self._staging_tbl_name):
                    futures_list.append(executor.submit(self._count_changes_by_type_single_shard, name,
                                                        filter_str=filter_str))
            else:
                futures_list.append(executor.submit(self._count_changes_by_type_single_shard,
                                                    self._staging_tbl_name,
                                                    filter_str=filter_str))
        changes_by_type = defaultdict(int)
        for f in futures_list:
            result = f.result()
            for k, v in result.items():
                changes_by_type[k] += v
        old_row_count = self.row_count
        approx_add_count = sum(v for k, v in changes_by_type.items() if k == 'add')
        approx_remove_count = sum(v for k, v in changes_by_type.items() if k == 'remove')
        approx_net_adds = approx_add_count - approx_remove_count
        approx_new_count = old_row_count + approx_net_adds
        self._logger.info('Found {0} approximate entries in staging table based on delta '
                          '({1} in existing table, {2} adds, {3} removes)'.format(approx_new_count,
                                                                                  old_row_count, approx_add_count,
                                                                                  approx_remove_count))
        return old_row_count + approx_net_adds
def staging_row_count ( self ) : assert self . _postprocessed and 'BaseDeltaImporter::staging_row_count called on un-postprocessed/non-delta table' self . _logger . info ( 'Counting the approximate number of entries in staging table based on delta...' ) futures_list = [ ] filter_str = "change_type != \'update\'" with futures . ThreadPoolExecutor ( max_workers = self . _max_db_connections ) as executor : if self . _supports_imei_shards : for name , rstart , rend in partition_utils . physical_imei_shards ( self . _conn , tbl_name = self . _staging_tbl_name ) : futures_list . append ( executor . submit ( self . _count_changes_by_type_single_shard , name , filter_str = filter_str ) ) else : futures_list . append ( executor . submit ( self . _count_changes_by_type_single_shard , self . _staging_tbl_name , filter_str = filter_str ) ) changes_by_type = defaultdict ( int ) for f in futures_list : result = f . result ( ) for k , v in result . items ( ) : changes_by_type [ k ] += v old_row_count = self . row_count approx_add_count = sum ( v for k , v in changes_by_type . items ( ) if k == 'add' ) approx_remove_count = sum ( v for k , v in changes_by_type . items ( ) if k == 'remove' ) approx_net_adds = approx_add_count - approx_remove_count approx_new_count = old_row_count + approx_net_adds self . _logger . info ( 'Found {0} approximate entries in staging table based on delta ' '({1} in existing table, {2} adds, {3} removes)' . format ( approx_new_count , old_row_count , approx_add_count , approx_remove_count ) ) return old_row_count + approx_net_adds
Overrides AbstractImporter.staging_row_count.
Overrides AbstractImporter . staging_row_count .
"""Overrides AbstractImporter.staging_row_count.

        This is NOT a completely accurate staging_row_count for performance reasons. Since this is only used
        in the import size check, which can be disabled, it has been decided to make a tradeoff between performance
        and accuracy. This is only inaccurate when:

        1. There are duplicate rows with the same primary keys and the same action in the delta table. A different
           action for the same primary key will have already failed the delta_multiple_changes check.
        2. An add is already in the DB or a remove is not in the DB. This will open happen if
           self._perform_delta_adds_check or self._perform_delta_removes_check is False when these checks would have
           failed if enabled.
        """ # All futures should be done at this point as with block is exited above
self