def _get_max_partition_from_part_specs(part_specs, partition_key, filter_map):
        """
        
        """
        if not part_specs:
            return None

        # Assuming all specs have the same keys.
        if partition_key not in part_specs[0].keys():
            raise AirflowException("Provided partition_key {} "
                                   "is not in part_specs.".format(partition_key))
        if filter_map:
            is_subset = set(filter_map.keys()).issubset(set(part_specs[0].keys()))
        if filter_map and not is_subset:
            raise AirflowException("Keys in provided filter_map {} "
                                   "are not subset of part_spec keys: {}"
                                   .format(', '.join(filter_map.keys()),
                                           ', '.join(part_specs[0].keys())))

        candidates = [p_dict[partition_key] for p_dict in part_specs
                      if filter_map is None or
                      all(item in p_dict.items() for item in filter_map.items())]

        if not candidates:
            return None
        else:
            return max(candidates).encode('utf-8')