Repository: spotify/luigi
File: luigi/contrib/hive.py (Python)

Code before:
    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.decode('utf-8')


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])


def _validate_partition(partition):
    """
    If partition is set and it's size more than one and it's not ordered
    than we're unable to restore its path in warehouse definitely
    """
    if (
            partition
            and len(partition) > 1
            and not isinstance(partition, collections.OrderedDict)
    ):
        raise ValueError('Unable to restore table/partition location')


@six.add_metaclass(abc.ABCMeta)
class HiveClient(object):  # interface

    @abc.abstractmethod
    def table_location(self, table, database='default', partition=None):
        """
        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.

Review comment (suggestion):
`If partition is set and its size is more than one and not ordered, then we're unable to restore its path in the warehouse`

Code after:
    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.decode('utf-8')


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])


def _is_ordered_dict(dikt):
    if isinstance(dikt, collections.OrderedDict):
        return True

    if sys.version_info >= (3, 7):
        return isinstance(dikt, dict)

    return False


def _validate_partition(partition):
    """
    If partition is set and its size is more than one and not ordered,
    then we're unable to restore its path in the warehouse
    """
    if (
            partition
            and len(partition) > 1
            and not _is_ordered_dict(partition)
    ):
        raise ValueError('Unable to restore table/partition location')


@six.add_metaclass(abc.ABCMeta)
class HiveClient(object):  # interface

    @abc.abstractmethod