<commit_before>#!/usr/bin/env python3
# pylint: disable=duplicate-code

import copy
import singer

import tap_mysql.sync_strategies.common as common

LOGGER = singer.get_logger()

BOOKMARK_KEYS = {'last_pk_fetched', 'max_pk_values', 'version', 'initial_full_table_complete'}


def pks_are_auto_incrementing(connection, catalog_entry):
    database_name = common.get_database_name(catalog_entry)
    key_properties = common.get_key_properties(catalog_entry)

    if not key_properties:
        return False

    sql = """SELECT 1
               FROM information_schema.columns
              WHERE table_schema = '{}'
                AND table_name = '{}'
                AND column_name = '{}'

    """

    with connection.cursor() as cur:
        for pk in key_properties:
            cur.execute(sql.format(database_name,
                                   catalog_entry.table,
                                   pk))

            result = cur.fetchone()

            if not result:
                return False

    return True


def get_max_pk_values(connection, catalog_entry):
    database_name = common.get_database_name(catalog_entry)
    escaped_db = common.escape(database_name)
    escaped_table = common.escape(catalog_entry.table)

    key_properties = common.get_key_properties(catalog_entry)
    escaped_columns = [common.escape(c) for c in key_properties]

    sql = """SELECT {}
               FROM {}.{}
              ORDER BY {}
              LIMIT 1
    """

    select_column_clause = ", ".join(escaped_columns)
    order_column_clause = ", ".join([pk + " DESC" for pk in escaped_columns])

    with connection.cursor() as cur:
        cur.execute(sql.format(select_column_clause,
                               escaped_db,
                               escaped_table,
                               order_column_clause))
        result = cur.fetchone()

        return dict(zip(key_properties, result))


def generate_pk_clause(catalog_entry, state):
    key_properties = common.get_key_properties(catalog_entry)
    escaped_columns = [common.escape(c) for c in key_properties]

    where_clause = " AND ".join([pk + " > `{}`" for pk in escaped_columns])
    order_by_clause = ", ".join(['`{}`, ' for pk in escaped_columns])

    max_pk_values = singer.get_bookmark(state,
                                        catalog_entry.tap_stream_id,
                                        'max_pk_values')

    pk_comparisons = ["{} <= {}".format(common.escape(pk), max_pk_values[pk])
                      for pk in key_properties]

    last_pk_fetched = singer.get_bookmark(state,
                                          catalog_entry.tap_stream_id,
                                          'last_pk_fetched')

    if last_pk_fetched:
        last_pk_values = ""
    else:
        sql = " WHERE {} ORDER BY {} ASC".format(" AND ".join(pk_comparisons),
                                                 ", ".join(escaped_columns))
    return sql
<commit_msg>use last_pk_fetched for FULL_TABLE<commit_after>#!/usr/bin/env python3
# pylint: disable=duplicate-code

import copy
import singer

import tap_mysql.sync_strategies.common as common

LOGGER = singer.get_logger()

BOOKMARK_KEYS = {'last_pk_fetched', 'max_pk_values', 'version', 'initial_full_table_complete'}


def pks_are_auto_incrementing(connection, catalog_entry):
    database_name = common.get_database_name(catalog_entry)
    key_properties = common.get_key_properties(catalog_entry)

    if not key_properties:
        return False

    sql = """SELECT 1
               FROM information_schema.columns
              WHERE table_schema = '{}'
                AND table_name = '{}'
                AND column_name = '{}'

    """

    with connection.cursor() as cur:
        for pk in key_properties:
            cur.execute(sql.format(database_name,
                                   catalog_entry.table,
                                   pk))

            result = cur.fetchone()

            if not result:
                return False

    return True


def get_max_pk_values(connection, catalog_entry):
    database_name = common.get_database_name(catalog_entry)
    escaped_db = common.escape(database_name)
    escaped_table = common.escape(catalog_entry.table)

    key_properties = common.get_key_properties(catalog_entry)
    escaped_columns = [common.escape(c) for c in key_properties]

    sql = """SELECT {}
               FROM {}.{}
              ORDER BY {}
              LIMIT 1
    """

    select_column_clause = ", ".join(escaped_columns)
    order_column_clause = ", ".join([pk + " DESC" for pk in escaped_columns])

    with connection.cursor() as cur:
        cur.execute(sql.format(select_column_clause,
                               escaped_db,
                               escaped_table,
                               order_column_clause))
        result = cur.fetchone()

        return dict(zip(key_properties, result))


def generate_pk_clause(catalog_entry, state):
    key_properties = common.get_key_properties(catalog_entry)
    escaped_columns = [common.escape(c) for c in key_properties]

    where_clause = " AND ".join([pk + " > `{}`" for pk in escaped_columns])
    order_by_clause = ", ".join(['`{}`, ' for pk in escaped_columns])

    max_pk_values = singer.get_bookmark(state,
                                        catalog_entry.tap_stream_id,
                                        'max_pk_values')

    last_pk_fetched = singer.get_bookmark(state,
                                          catalog_entry.tap_stream_id,
                                          'last_pk_fetched')

    if last_pk_fetched:
        pk_comparisons = ["({} > {} AND {} <= {})".format(common.escape(pk),
                                                          last_pk_fetched[pk],
                                                          common.escape(pk),
                                                          max_pk_values[pk])
                          for pk in key_properties]
    else:
        pk_comparisons = ["{} <= {}".format(common.escape(pk), max_pk_values[pk])
                          for pk in key_properties]

    sql = " WHERE {} ORDER BY {} ASC".format(" AND ".join(pk_comparisons),
                                             ", ".join(escaped_columns))

    return sql

{'fraction_non_alphanumeric': 0.09429477020602219, 'fraction_numerical': 0.0009508716323296355, 'mean_word_length': 1.8389563652721548, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '237303', 'n_tokens_mistral': 1777, 'n_tokens_neox': 1709, 'n_words': 334}