<commit_before># 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.

import abc
import logging
import luigi
import luigi.hadoop
from luigi.target import FileSystemTarget, FileAlreadyExists
import os
import subprocess
import tempfile
from luigi.task import flatten

logger = logging.getLogger('luigi-interface')


class HiveCommandError(RuntimeError):
    def __init__(self, message, out=None, err=None):
        super(HiveCommandError, self).__init__(message, out, err)
        self.message = message
        self.out = out
        self.err = err


def load_hive_cmd():
    return luigi.configuration.get_config().get('hive', 'command', 'hive')


def get_hive_syntax():
    return luigi.configuration.get_config().get('hive', 'release', 'cdh4')


def run_hive(args, check_return_code=True):
    """Runs the `hive` from the command line, passing in the given args, and
       returning stdout.

       With the apache release of Hive, so of the table existence checks
       (which are done using DESCRIBE do not exit with a return code of 0
       so we need an option to ignore the return code and just return stdout for parsing
    """
    cmd = [load_hive_cmd()] + args
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, stderr = p.communicate()
    if check_return_code and p.returncode != 0:
        raise HiveCommandError("Hive command: {0} failed with error code: {1}".format(" ".join(cmd), p.returncode),
                               stdout, stderr)
    return stdout


def run_hive_cmd(hivecmd, check_return_code=True):
    """Runs the given hive query and returns stdout"""
    return run_hive(['-e', hivecmd], check_return_code)


def run_hive_script(script):
    """Runs the contents of the given script in hive and returns stdout"""
    if not os.path.isfile(script):
        raise RuntimeError("Hive script: {0} does not exist.".format(script))
    return run_hive(['-f', script])


class HiveClient(object):  # interface
    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod
    def table_location(self, table, database='default', partition={}):
        """
        Returns location of db.table (or db.table.partition). partition is a dict of partition key to
        value.
        """
        pass

    @abc.abstractmethod
    def table_schema(self, table, database='default'):
        """ Returns list of [(name, type)] for each column in database.table """
        pass

    @abc.abstractmethod
    def table_exists(self, table, database='default', partition={}):
        """
        Returns true iff db.table (or db.table.partition) exists. partition is a dict of partition key to
        value.
        """
        pass

    @abc.abstractmethod
    def partition_spec(self, partition):
        """ Turn a dict into a string partition specification """
        pass


class HiveCommandClient(HiveClient):
    """ Uses `hive` invocations to find information """
    def table_location(self, table, database='default', partition={}):
        cmd = "use {0}; describe formatted {1}".format(database, table)
        if partition:
            cmd += " PARTITION ({0})".format(self.partition_spec(partition))

        stdout = run_hive_cmd(cmd)

        for line in stdout.split("\n"):
            if "Location:" in line:
                return line.split("\t")[1]

    def table_exists(self, table, database='default', partition={}):
        if not partition:
            stdout = run_hive_cmd('use {0}; describe {1}'.format(database, table))

            return not "does not exist" in stdout
        else:
            stdout = run_hive_cmd("""use %s; show partitions %s partition
                                (%s)""" % (database, table, self.partition_spec(partition)))

            if stdout:
                return True
            else:
                return False

    def table_schema(self, table, database='default'):
        describe = run_hive_cmd("use {0}; describe {1}".format(database, table))
        if not describe or "does not exist" in describe:
            return None
        return [tuple([x.strip() for x in line.strip().split("\t")]) for line in describe.strip().split("\n")]

    def partition_spec(self, partition):
        """ Turns a dict into the a Hive partition specification string """
        return ','.join(["{0}='{1}'".format(k, v) for (k, v) in partition.items()])


class ApacheHiveCommandClient(HiveCommandClient):
    """
    A subclass for the HiveCommandClient to (in some cases) ignore the return code from
    the hive command so that we can just parse the output.
    """
    def table_exists(self, table, database='default', partition={}):
        if not partition:
            # Hive 0.11 returns 17 as the exit status if the table does not exist.
            # The actual message is: [Error 10001]: Table not found tablename
            # stdout is empty and an error message is returned on stderr.
            # This is why we can't check the return code on this command and
            # assume if stdout is empty that the table doesn't exist.
            stdout = run_hive_cmd('use {0}; describe {1}'.format(database, table), False)
            if stdout:
                return not "Table not found" in stdout
            else:
                # Hive returned a non-zero exit status and printed its output to stderr not stdout
                return False
        else:
            stdout = run_hive_cmd("""use %s; show partitions %s partition
                                (%s)""" % (database, table, self.partition_spec(partition)))

            if stdout:
                return True
            else:
                return False

    def table_schema(self, table, database='default'):
        describe = run_hive_cmd("use {0}; describe {1}".format(database, table), False)
        if not describe or "Table not found" in describe:
            return None
        return [tuple([x.strip() for x in line.strip().split("\t")]) for line in describe.strip().split("\n")]


class MetastoreClient(HiveClient):
    def table_location(self, table, database='default', partition={}):
        with HiveThriftContext() as client:
            if partition:
                partition_str = self.partition_spec(partition)
                thrift_table = client.get_partition_by_name(database, table, partition_str)
            else:
                thrift_table = client.get_table(database, table)
            return thrift_table.sd.location

    def table_exists(self, table, database='default', partition={}):
        with HiveThriftContext() as client:
            if not partition:
                return table in client.get_all_tables(database)
            else:
                partition_str = self.partition_spec(partition)
                # -1 is max_parts, the # of partition names to return (-1 = unlimited)
                return partition_str in client.get_partition_names(database, table, -1)<commit_msg>Fix the partition exists check so it matches the table exists code above.  Without this the check will always fail if the partition doesn't exist yet due to the underlying command returning a non-zero result<commit_after># 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.

import abc
import logging
import luigi
import luigi.hadoop
from luigi.target import FileSystemTarget, FileAlreadyExists
import os
import subprocess
import tempfile
from luigi.task import flatten

logger = logging.getLogger('luigi-interface')


class HiveCommandError(RuntimeError):
    def __init__(self, message, out=None, err=None):
        super(HiveCommandError, self).__init__(message, out, err)
        self.message = message
        self.out = out
        self.err = err


def load_hive_cmd():
    return luigi.configuration.get_config().get('hive', 'command', 'hive')


def get_hive_syntax():
    return luigi.configuration.get_config().get('hive', 'release', 'cdh4')


def run_hive(args, check_return_code=True):
    """Runs the `hive` from the command line, passing in the given args, and
       returning stdout.

       With the apache release of Hive, so of the table existence checks
       (which are done using DESCRIBE do not exit with a return code of 0
       so we need an option to ignore the return code and just return stdout for parsing
    """
    cmd = [load_hive_cmd()] + args
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, stderr = p.communicate()
    if check_return_code and p.returncode != 0:
        raise HiveCommandError("Hive command: {0} failed with error code: {1}".format(" ".join(cmd), p.returncode),
                               stdout, stderr)
    return stdout


def run_hive_cmd(hivecmd, check_return_code=True):
    """Runs the given hive query and returns stdout"""
    return run_hive(['-e', hivecmd], check_return_code)


def run_hive_script(script):
    """Runs the contents of the given script in hive and returns stdout"""
    if not os.path.isfile(script):
        raise RuntimeError("Hive script: {0} does not exist.".format(script))
    return run_hive(['-f', script])


class HiveClient(object):  # interface
    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod
    def table_location(self, table, database='default', partition={}):
        """
        Returns location of db.table (or db.table.partition). partition is a dict of partition key to
        value.
        """
        pass

    @abc.abstractmethod
    def table_schema(self, table, database='default'):
        """ Returns list of [(name, type)] for each column in database.table """
        pass

    @abc.abstractmethod
    def table_exists(self, table, database='default', partition={}):
        """
        Returns true iff db.table (or db.table.partition) exists. partition is a dict of partition key to
        value.
        """
        pass

    @abc.abstractmethod
    def partition_spec(self, partition):
        """ Turn a dict into a string partition specification """
        pass


class HiveCommandClient(HiveClient):
    """ Uses `hive` invocations to find information """
    def table_location(self, table, database='default', partition={}):
        cmd = "use {0}; describe formatted {1}".format(database, table)
        if partition:
            cmd += " PARTITION ({0})".format(self.partition_spec(partition))

        stdout = run_hive_cmd(cmd)

        for line in stdout.split("\n"):
            if "Location:" in line:
                return line.split("\t")[1]

    def table_exists(self, table, database='default', partition={}):
        if not partition:
            stdout = run_hive_cmd('use {0}; describe {1}'.format(database, table))

            return not "does not exist" in stdout
        else:
            stdout = run_hive_cmd("""use %s; show partitions %s partition
                                (%s)""" % (database, table, self.partition_spec(partition)))

            if stdout:
                return True
            else:
                return False

    def table_schema(self, table, database='default'):
        describe = run_hive_cmd("use {0}; describe {1}".format(database, table))
        if not describe or "does not exist" in describe:
            return None
        return [tuple([x.strip() for x in line.strip().split("\t")]) for line in describe.strip().split("\n")]

    def partition_spec(self, partition):
        """ Turns a dict into the a Hive partition specification string """
        return ','.join(["{0}='{1}'".format(k, v) for (k, v) in partition.items()])


class ApacheHiveCommandClient(HiveCommandClient):
    """
    A subclass for the HiveCommandClient to (in some cases) ignore the return code from
    the hive command so that we can just parse the output.
    """
    def table_exists(self, table, database='default', partition={}):
        if not partition:
            # Hive 0.11 returns 17 as the exit status if the table does not exist.
            # The actual message is: [Error 10001]: Table not found tablename
            # stdout is empty and an error message is returned on stderr.
            # This is why we can't check the return code on this command and
            # assume if stdout is empty that the table doesn't exist.
            stdout = run_hive_cmd('use {0}; describe {1}'.format(database, table), False)
            if stdout:
                return not "Table not found" in stdout
            else:
                # Hive returned a non-zero exit status and printed its output to stderr not stdout
                return False
        else:
            stdout = run_hive_cmd("""use %s; show partitions %s partition
                                (%s)""" % (database, table, self.partition_spec(partition)), False)

            if stdout:
                return True
            else:
                return False

    def table_schema(self, table, database='default'):
        describe = run_hive_cmd("use {0}; describe {1}".format(database, table), False)
        if not describe or "Table not found" in describe:
            return None
        return [tuple([x.strip() for x in line.strip().split("\t")]) for line in describe.strip().split("\n")]


class MetastoreClient(HiveClient):
    def table_location(self, table, database='default', partition={}):
        with HiveThriftContext() as client:
            if partition:
                partition_str = self.partition_spec(partition)
                thrift_table = client.get_partition_by_name(database, table, partition_str)
            else:
                thrift_table = client.get_table(database, table)
            return thrift_table.sd.location

    def table_exists(self, table, database='default', partition={}):
        with HiveThriftContext() as client:
            if not partition:
                return table in client.get_all_tables(database)
            else:
                partition_str = self.partition_spec(partition)
                # -1 is max_parts, the # of partition names to return (-1 = unlimited)
                return partition_str in client.get_partition_names(database, table, -1)
{'fraction_non_alphanumeric': 0.10053931686530394, 'fraction_numerical': 0.004927092349690392, 'mean_word_length': 2.6660971442518915, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1299275', 'n_tokens_mistral': 4173, 'n_tokens_neox': 4028, 'n_words': 1456}