WIP: narrow deep paths mutation

Author: nandikaK

config/defaults.py

@@ -78,6 +78,13 @@
 MUTPB_FV_SAMPLE_MAXN = 32  # max n of instantiations to sample from top k
 MUTPB_FV_QUERY_LIMIT = 256  # SPARQL query limit for the top k instantiations
 MUTPB_SP = 0.05  # prob to simplify pattern (warning: can restrict exploration)
+MUTPB_DN = 0.05  # prob to try a deep and narrow paths mutation
+MUTPB_DN_MIN_LEN = 2  # minimum length of the deep and narrow paths
+MUTPB_DN_MAX_LEN = 10  # absolute max of path length if not stopped by term_pb
+MUTPB_DN_TERM_PB = 0.3  # prob to terminate node expansion each step > min_len
+MUTPB_DN_FILTER_NODE_COUNT = 10
+MUTPB_DN_FILTER_EDGE_COUNT = 1
+MUTPB_DN_QUERY_LIMIT = 32
 
 # for import in helpers and __init__
 __all__ = [_v for _v in globals().keys() if _v.isupper()]

gp_learner.py

@@ -48,6 +48,7 @@
 from gp_query import predict_query
 from gp_query import query_time_hard_exceeded
 from gp_query import query_time_soft_exceeded
+from gp_query import variable_substitution_deep_narrow_mut_query
 from gp_query import variable_substitution_query
 from graph_pattern import canonicalize
 from graph_pattern import gen_random_var
@@ -653,6 +654,105 @@ def mutate_fix_var(
     return res
 
 
+def _mutate_deep_narrow_path_helper(
+            sparql,
+            timeout,
+            gtp_scores,
+            child,
+            edge_var,
+            node_var,
+            gtp_sample_n=config.MUTPB_FV_RGTP_SAMPLE_N,
+            limit_res=config.MUTPB_DN_QUERY_LIMIT,
+            sample_n=config.MUTPB_FV_SAMPLE_MAXN,
+):
+    assert isinstance(child, GraphPattern)
+    assert isinstance(gtp_scores, GTPScores)
+
+    # The further we get, the less gtps are remaining. Sampling too many (all)
+    # of them might hurt as common substitutions (> limit ones) which are dead
+    # ends could cover less common ones that could actually help
+    gtp_sample_n = min(gtp_sample_n, int(gtp_scores.remaining_gain))
+    gtp_sample_n = random.randint(1, gtp_sample_n)
+
+    ground_truth_pairs = gtp_scores.remaining_gain_sample_gtps(
+     n=gtp_sample_n)
+    t, substitution_counts = variable_substitution_deep_narrow_mut_query(
+     sparql, timeout, child, edge_var, node_var, ground_truth_pairs,
+     limit_res)
+    edge_count, node_sum_count = substitution_counts
+    if not node_sum_count:
+        # the current pattern is unfit, as we can't find anything fulfilling it
+        logger.debug("tried to fix a var %s without result:\n%s"
+                     "seems as if the pattern can't be fulfilled!",
+                     edge_var, child.to_sparql_select_query())
+        fixed = False
+        return child, fixed
+    mutate_fix_var_filter(node_sum_count)
+    mutate_fix_var_filter(edge_count)
+    if not node_sum_count:
+        # could have happened that we removed the only possible substitution
+        fixed = False
+        return child, fixed
+
+    prio = Counter()
+    for edge, node_sum in node_sum_count.items():
+        ec = edge_count[edge]
+        prio[edge] = ec / (node_sum / ec)  # ec / AVG degree
+    # randomly pick n of the substitutions with a prob ~ to their counts
+    edges, prios = zip(*prio.most_common())
+
+    substs = sample_from_list(edges, prios, sample_n)
+
+    logger.info(
+        'fixed variable %s in %sto:\n %s\n<%d out of:\n%s\n',
+        edge_var.n3(),
+        child,
+        '\n '.join([subst.n3() for subst in substs]),
+        sample_n,
+        '\n'.join([
+            ' %.3f: %s' % (c, v.n3()) for v, c in prio.most_common()]),
+    )
+    fixed = True
+    orig_child = child
+    children = [
+     GraphPattern(child, mapping={edge_var: subst})
+     for subst in substs
+     ]
+    children = [
+        c if fit_to_live(c) else orig_child
+        for c in children
+        ]
+    if children:
+        child = random.choice(list(children))
+    return child, fixed
+
+
+def mutate_deep_narrow_path(
+        child, sparql, timeout, gtp_scores,
+        min_len=config.MUTPB_DN_MIN_LEN,
+        max_len=config.MUTPB_DN_MAX_LEN,
+        term_pb=config.MUTPB_DN_TERM_PB,
+):
+    assert isinstance(child, GraphPattern)
+    nodes = list(child.nodes)
+    start_node = random.choice(nodes)
+    # target_nodes = set(nodes) - {start_node}
+    gp = child
+    hop = 0
+    while True:
+        if hop >= min_len and random.random() < term_pb:
+            break
+        if hop >= max_len:
+            break
+        hop += 1
+        new_triple, var_node, var_edge = _mutate_expand_node_helper(start_node)
+        gp += [new_triple]
+        gp, fixed = _mutate_deep_narrow_path_helper(
+            sparql, timeout, gtp_scores, gp, var_edge, var_node)
+        start_node = var_node
+    return gp
+
+
 def mutate_simplify_pattern(gp):
     if len(gp) < 2:
         return gp
@@ -757,6 +857,7 @@ def mutate(
         pb_dt=config.MUTPB_DT,
         pb_en=config.MUTPB_EN,
         pb_fv=config.MUTPB_FV,
+        pb_dn=config.MUTPB_DN,
         pb_id=config.MUTPB_ID,
         pb_iv=config.MUTPB_IV,
         pb_mv=config.MUTPB_MV,
@@ -796,15 +897,15 @@ def mutate(
     if random.random() < pb_sp:
         child = mutate_simplify_pattern(child)
 
+    if random.random() < pb_dn:
+        child = mutate_deep_narrow_path(child, sparql, timeout, gtp_scores)
+
     if random.random() < pb_fv:
         child = canonicalize(child)
         children = mutate_fix_var(sparql, timeout, gtp_scores, child)
     else:
         children = [child]
 
-
-    # TODO: deep & narrow paths mutation
-
     children = {
         c if fit_to_live(c) else orig_child
         for c in children

gp_query.py

@@ -32,6 +32,8 @@
 from graph_pattern import TARGET_VAR
 from graph_pattern import ASK_VAR
 from graph_pattern import COUNT_VAR
+from graph_pattern import NODE_VAR_SUM
+from graph_pattern import EDGE_VAR_COUNT
 from utils import exception_stack_catcher
 from utils import sparql_json_result_bindings_to_rdflib
 from utils import timer
@@ -279,7 +281,6 @@ def _combined_chunk_res(q_res, _vars, _ret_val_mapping):
     return chunk_res
 
 
-
 def count_query(sparql, timeout, graph_pattern, source=None,
                 **kwds):
     assert isinstance(graph_pattern, GraphPattern)
@@ -457,6 +458,68 @@ def _var_subst_res_update(res, update, **_):
     res += update
 
 
+def variable_substitution_deep_narrow_mut_query(
+        sparql, timeout, graph_pattern, edge_var, node_var,
+        source_target_pairs, limit_res, batch_size=config.BATCH_SIZE):
+    _vars, _values, _ret_val_mapping = _get_vars_values_mapping(
+        graph_pattern, source_target_pairs)
+    _edge_var_node_var_and_vars = (edge_var, node_var, _vars)
+    return _multi_query(
+        sparql, timeout, graph_pattern, source_target_pairs, batch_size,
+        _edge_var_node_var_and_vars, _values, _ret_val_mapping,
+        _var_subst_dnp_res_init, _var_subst_dnp_chunk_q,
+        _var_subst_dnp_chunk_result_ext,
+        _res_update=_var_subst_dnp_update,
+        limit=limit_res,
+        # non standard, passed via **kwds, see handling below
+    )
+
+
+# noinspection PyUnusedLocal
+def _var_subst_dnp_res_init(_, **kwds):
+    return Counter(), Counter()
+
+
+def _var_subst_dnp_chunk_q(gp, _edge_var_node_var_and_vars,
+                           values_chunk, limit):
+    edge_var, node_var, _vars = _edge_var_node_var_and_vars
+    return gp.to_find_edge_var_for_narrow_path_query(
+        edge_var=edge_var,
+        node_var=node_var,
+        vars_=_vars,
+        values={_vars: values_chunk},
+        limit_res=limit)
+
+
+# noinspection PyUnusedLocal
+def _var_subst_dnp_chunk_result_ext(
+        q_res, _edge_var_node_var_and_vars, _, **kwds):
+    edge_var, node_var, _vars = _edge_var_node_var_and_vars
+    chunk_edge_count, chunk_node_sum = Counter(), Counter()
+    res_rows_path = ['results', 'bindings']
+    bindings = sparql_json_result_bindings_to_rdflib(
+        get_path(q_res, res_rows_path, default=[])
+    )
+
+    for row in bindings:
+        row_res = get_path(row, [edge_var])
+        edge_count = int(get_path(row, [EDGE_VAR_COUNT], '0'))
+        chunk_edge_count[row_res] += edge_count
+        node_sum_count = int(get_path(row, [NODE_VAR_SUM], '0'))
+        chunk_node_sum[row_res] += node_sum_count
+    return chunk_edge_count, chunk_node_sum,
+
+
+def _var_subst_dnp_update(res, up, **_):
+    edge_count, node_sum_count = res
+    try:
+        chunk_edge_count, chunk_node_sum = up
+        edge_count.update(chunk_edge_count)
+        node_sum_count.update(chunk_node_sum)
+    except ValueError:
+        pass
+
+
 def generate_stps_from_gp(sparql, gp):
     """Generates a list of source target pairs from a given graph pattern.
 

graph_pattern.py

@@ -31,6 +31,7 @@
 import six
 
 from utils import URIShortener
+import config
 
 logger = logging.getLogger(__name__)
 
@@ -41,6 +42,8 @@
 TARGET_VAR = Variable('target')
 ASK_VAR = Variable('ask')
 COUNT_VAR = Variable('count')
+EDGE_VAR_COUNT = Variable('edge_var_count')
+NODE_VAR_SUM = Variable('node_var_sum')
 
 
 def gen_random_var():
@@ -714,6 +717,86 @@ def to_count_var_over_values_query(self, var, vars_, values, limit):
         res += 'LIMIT %d\n' % limit
         return self._sparql_prefix(res)
 
+    def to_find_edge_var_for_narrow_path_query(
+            self, edge_var, node_var, vars_, values, limit_res,
+            filter_node_count=config.MUTPB_DN_FILTER_NODE_COUNT,
+            filter_edge_count=config.MUTPB_DN_FILTER_EDGE_COUNT,
+    ):
+        """Counts possible substitutions for edge_var to get a narrow path
+
+        Meant to perform a query like this:
+        SELECT *
+        {
+          {
+            SELECT
+              ?edge_var
+              (COUNT(*) AS ?edge_var_count)
+              (MAX(?node_var_count) AS ?max_node_count)
+              (COUNT(*)/AVG(?node_var_count) as ?prio_var)
+            {
+              SELECT DISTINCT
+                ?source ?target ?edge_var (COUNT(?node_var) AS ?node_var_count)
+              {
+                VALUES (?source ?target) {
+                  (dbr:Adolescence dbr:Youth)
+                  (dbr:Adult dbr:Child)
+                  (dbr:Angel dbr:Heaven)
+                  (dbr:Arithmetic dbr:Mathematics)
+                }
+                ?node_var ?edge_var ?source .
+                ?source dbo:wikiPageWikiLink ?target .
+              }
+            }
+            GROUP BY ?edge_var
+            ORDER BY DESC(?edge_var_count)
+          }
+          FILTER(?max_node_count < 10 && ?edge_var_count > 1)
+        }
+        ORDER BY DESC(?prio_var)
+        LIMIT 32
+
+        :param edge_var: Edge variable to find substitution for.
+        :param node_var: Node variable to count.
+        :param vars_: List of vars to fix values for (e.g. ?source, ?target).
+        :param values: List of value lists for vars_.
+        :param filter_node_count: Filter on node count of edge variable.
+        :param filter_edge_count: Filter for edge count of triples.
+        :param limit_res : limit result size
+        :return: Query String.
+        """
+
+        res = 'SELECT * WHERE {\n'
+        res += ' {\n'\
+               '  SELECT %s (SUM (?node_var_count) AS %s) (COUNT(%s) AS %s) ' \
+               '(MAX(?node_var_count) AS ?max_node_count) WHERE {\n' % (
+                     edge_var.n3(),
+                     NODE_VAR_SUM.n3(),
+                     ' && '.join([v.n3() for v in vars_]),
+                     EDGE_VAR_COUNT.n3(), )
+        res += '    SELECT DISTINCT %s %s (COUNT(%s) AS ?node_var_count) ' \
+               'WHERE {\n   ' % (' '.join([v.n3() for v in vars_]),
+                                 edge_var.n3(), node_var.n3(), )
+        res += self._sparql_values_part(values)
+
+        # triples part
+        tres = []
+        for s, p, o in self:
+            tres.append('%s %s %s .' % (s.n3(), p.n3(), o.n3()))
+        indent = ' ' * 3
+        triples = indent + ('\n' + indent).join(tres) + '\n'
+        res += triples
+        res += '    }\n'\
+               '   }\n'
+        res += '   GROUP BY %s\n' % edge_var.n3()
+        res += '  }\n'
+        res += '  FILTER(?max_node_count < %d && %s > %d)\n' \
+               % (filter_node_count, EDGE_VAR_COUNT.n3(),
+                  filter_edge_count)
+        res += '}\n'
+        res += 'ORDER BY ASC(%s)\n' % NODE_VAR_SUM.n3()
+        res += 'LIMIT %d' % limit_res
+        return self._sparql_prefix(res)
+
     def to_dict(self):
         return {
             'fitness': self.fitness.values if self.fitness.valid else (),

tests/test_gp_learner_offline.py

@@ -13,6 +13,7 @@
 from gp_learner import mutate_increase_dist
 from gp_learner import mutate_merge_var
 from gp_learner import mutate_simplify_pattern
+from gp_learner import mutate_deep_narrow_path
 from graph_pattern import GraphPattern
 from graph_pattern import SOURCE_VAR
 from graph_pattern import TARGET_VAR
@@ -108,6 +109,35 @@ def test_mutate_merge_var():
         assert False, "merge never reached one of the cases: %s" % cases
 
 
+def test_mutate_deep_narrow_path():
+    p = Variable('p')
+    gp = GraphPattern([
+        (SOURCE_VAR, p, TARGET_VAR)
+    ])
+    child = mutate_deep_narrow_path(gp)
+    assert gp == child or len(child) > len(gp)
+    print(gp)
+    print(child)
+
+
+def test_to_find_edge_var_for_narrow_path_query():
+    node_var = Variable('node_variable')
+    edge_var = Variable('edge_variable')
+    gp = GraphPattern([
+        (node_var, edge_var, SOURCE_VAR),
+        (SOURCE_VAR, wikilink, TARGET_VAR)
+    ])
+    filter_node_count = 10
+    filter_edge_count = 1
+    limit_res = 32
+    vars_ = {SOURCE_VAR,TARGET_VAR}
+    res = GraphPattern.to_find_edge_var_for_narrow_path_query(gp, edge_var, node_var,
+                                           vars_, filter_node_count,
+                                           filter_edge_count, limit_res)
+    print(gp)
+    print(res)
+
+
 def test_simplify_pattern():
     gp = GraphPattern([(SOURCE_VAR, wikilink, TARGET_VAR)])
     res = mutate_simplify_pattern(gp)
@@ -270,3 +300,8 @@ def test_remaining_gain_sample_gtps():
 
 def test_gtp_scores():
     assert gtp_scores - gtp_scores == 0
+
+
+if __name__ == '__main__':
+    # test_mutate_deep_narrow_path()
+    test_to_find_edge_var_for_narrow_path_query()