Repository: graphistry/pygraphistry

Issue:
[BUG] - GFQL
**Describe the bug**

GFQL not filtering on degrees


**To Reproduce**
Code, including data, than can be run without editing:
```python
g = graphistry.nodes(nodes).bind(node="Vertex")
                   .edges(edges).bind(source='Vertex 1', destination='Vertex 2')
         )

g_gfql = g_gqfl_1.get_degrees().nodes(lambda g: g._nodes.reset_index(drop=True)).edges(lambda g: g._edges.reset_index(drop=True))
          .chain([
              n({'degree': gt(1)}),
              e_undirected(),
              n({'degree': gt(1)})
          ])

```

**Expected behavior**
Should only return nodes with degree gt 1

**Actual behavior**
It is not filtering out based on degree.

**Graphistry GPU server environment**
 
AWS - 2.40.74

**Additional context**


Contact for datasample

Hints from issue comments:
@lmeyerov: Some minimal reproductions

### Preamble

```python
import graphistry
import pandas as pd
from graphistry import n, e_forward, e_reverse, e_undirected, is_in, gt
```

### 1.

```python
edf = pd.DataFrame({
    's': ['a1', 'b3', 'b3'],
    'd': ['b3', 'b3', 'c1']
})
g = graphistry.edges(edf, 's', 'd').materialize_nodes().get_degrees()
```

where

```
	id	degree_in	degree_out	degree
0	a1	0	1	1
1	b3	2	2	4
2	c1	1	0	1
```

Then unexpected output:

```python
g2 = (g.get_degrees()
  .chain([
      n({'degree': gt(1)}),
      e_undirected(),
      n({'degree': gt(1)})
  ])
)
```
=>

```python
	id	degree	degree_in	degree_out
0	b3	4	2	2
1	a1	1	0	1
```

We don't expect edge `(a1)->(b3)` as `(a1 {degree: 1})`

```
	s	d
0	a1	b3
1	b3	b3
```
@lmeyerov: @SinsBre published as 0.34.5 -- lmk if still happening

I added a test case on a minimal dataset with the same basic query structure, among others: https://github.com/graphistry/pygraphistry/blob/ecea21df3fff4bd4fadcf7f51add1c45ac0df3d7/graphistry/tests/compute/test_chain.py#L383

the bigger dataset may have more interesting shapes going on, so am curious

Solution patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6c8aef7f62..16414bf7b8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,18 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## [Development]
 
+## [0.34.5 - 2024-09-23]
+
+### Fixed
+
+* GFQL: Fix `chain()` regression around an incorrectly disabled check manifesting as https://github.com/graphistry/pygraphistry/issues/583
+* GFQL: Fix `chain()`, `hop()` traverse filtering logic for a multi-hop edge scenarios
+* GFQL: Fix `hop()` predicate handling in multihop scenarios
+
+### Infra
+
+* GFQL: Expand test suite around multihop edge predicates in `hop()` and `chain()`
+
 ## [0.34.4 - 2024-09-20]
 
 ### Added
diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py
index bbe18627c6..faea1d6a0e 100644
--- a/graphistry/compute/chain.py
+++ b/graphistry/compute/chain.py
@@ -1,3 +1,4 @@
+import logging
 from typing import Any, Dict, Union, cast, List, Tuple, TYPE_CHECKING
 import pandas as pd
 from graphistry.Engine import Engine, EngineAbstract, df_concat, resolve_engine
@@ -85,14 +86,19 @@ def combine_steps(g: Plottable, kind: str, steps: List[Tuple[ASTObject,Plottable
 
     concat = df_concat(engine)
 
+    logger.debug('-----------[ combine %s ---------------]', kind)
+
     # df[[id]]
     out_df = concat([
         getattr(g_step, df_fld)[[id]]
         for (_, g_step) in steps
     ]).drop_duplicates(subset=[id])
-    for (op, g_step) in steps:
-        logger.debug('adding nodes to concat: %s', g_step._nodes[[g_step._node]])
-        logger.debug('adding edges to concat: %s', g_step._edges[[g_step._source, g_step._destination]])
+    if logger.isEnabledFor(logging.DEBUG):
+        for (op, g_step) in steps:
+            if kind == 'edges':
+                logger.debug('adding edges to concat: %s', g_step._edges[[g_step._source, g_step._destination]])
+            else:
+                logger.debug('adding nodes to concat: %s', g_step._nodes[[g_step._node]])
 
     # df[[id, op_name1, ...]]
     logger.debug('combine_steps ops: %s', [op for (op, _) in steps])
@@ -107,7 +113,7 @@ def combine_steps(g: Plottable, kind: str, steps: List[Tuple[ASTObject,Plottable
             out_df[op._name] = out_df[op._name].fillna(False).astype(bool)
     out_df = out_df.merge(getattr(g, df_fld), on=id, how='left')
 
-    logger.debug('COMBINED[%s] >> %s', kind, out_df)
+    logger.debug('COMBINED[%s] >>\n%s', kind, out_df)
 
     return out_df
 
diff --git a/graphistry/compute/hop.py b/graphistry/compute/hop.py
index 57bd09bc83..e14da1aca7 100644
--- a/graphistry/compute/hop.py
+++ b/graphistry/compute/hop.py
@@ -51,8 +51,8 @@ def hop(self: Plottable,
     source_node_query: dataframe query to match nodes before hopping (including intermediate)
     destination_node_query: dataframe query to match nodes after hopping (including intermediate)
     edge_query: dataframe query to match edges before hopping (including intermediate)
-    return_as_wave_front: Only return the nodes/edges reached, ignoring past ones (primarily for internal use)
-    target_wave_front: Only consider these nodes for reachability, and for intermediate hops, also consider nodes (primarily for internal use by reverse pass)
+    return_as_wave_front: Exclude starting node(s) in return, returning only encountered nodes
+    target_wave_front: Only consider these nodes + self._nodes for reachability
     engine: 'auto', 'pandas', 'cudf' (GPU)
     """
 
@@ -141,7 +141,11 @@ def hop(self: Plottable,
     matches_edges = edges_indexed[[EDGE_ID]][:0]
 
     #richly-attributed subset for dest matching & return-enriching
-    base_target_nodes = target_wave_front if target_wave_front is not None else g2._nodes
+    if target_wave_front is None:
+        base_target_nodes = g2._nodes
+    else:
+        base_target_nodes = concat([target_wave_front, g2._nodes], ignore_index=True, sort=False).drop_duplicates(subset=[g2._node])
+    #TODO precompute src/dst match subset if multihop?
 
     if debugging_hop and logger.isEnabledFor(logging.DEBUG):
         logger.debug('~~~~~~~~~~ LOOP PRE ~~~~~~~~~~~')
@@ -162,6 +166,15 @@ def hop(self: Plottable,
             logger.debug('matches_nodes:\n%s', matches_nodes)
             logger.debug('matches_edges:\n%s', matches_edges)
             logger.debug('first_iter: %s', first_iter)
+            logger.debug('source_node_match: %s', source_node_match)
+            logger.debug('starting_nodes:\n%s', starting_nodes)
+            logger.debug('self._nodes:\n%s', self._nodes)
+            logger.debug('wave_front:\n%s', wave_front)
+            logger.debug('wave_front_base:\n%s',
+                starting_nodes
+                if first_iter else
+                wave_front.merge(self._nodes, on=g2._node, how='left'),
+            )
 
         if not to_fixed_point and hops_remaining is not None:
             if hops_remaining < 1:
@@ -171,12 +184,12 @@ def hop(self: Plottable,
         assert len(wave_front.columns) == 1, "just indexes"
         wave_front_iter : DataFrameT = query_if_not_none(
             source_node_query,
-                filter_by_dict(
-                    starting_nodes
-                    if first_iter else
-                    wave_front.merge(self._nodes, on=g2._node, how='left'),
-                    source_node_match
-                )
+            filter_by_dict(
+                starting_nodes
+                if first_iter else
+                wave_front.merge(self._nodes, on=g2._node, how='left'),
+                source_node_match
+            )
         )[[ g2._node ]]
         first_iter = False
 
@@ -195,11 +208,12 @@ def hop(self: Plottable,
                 [[g2._source, g2._destination, EDGE_ID]]
             )
             if target_wave_front is not None:
-                assert nodes is not None, "target_wave_front indicates nodes"
+                # target prev internal transitions (g._nodes) + starting point (target)
+                # final hop can only be to target
                 if hops_remaining:
                     intermediate_target_wave_front = concat([
                         target_wave_front[[g2._node]],
-                        nodes[[g2._node]]
+                        self._nodes[[g2._node]]
                         ], sort=False, ignore_index=True
                     ).drop_duplicates()
                 else:
@@ -243,13 +257,11 @@ def hop(self: Plottable,
                 logger.debug('--- direction in [reverse, undirected] ---')
                 logger.debug('hop_edges_reverse basic:\n%s', hop_edges_reverse)
 
-            #FIXME: What test case does this enable? Disabled to pass shortest path backwards pass steps
-            if False and target_wave_front is not None:
-                assert nodes is not None, "target_wave_front indicates nodes"
+            if target_wave_front is not None:
                 if hops_remaining:
                     intermediate_target_wave_front = concat([
                         target_wave_front[[g2._node]],
-                        nodes[[g2._node]]
+                        self._nodes[[g2._node]]
                         ], sort=False, ignore_index=True
                     ).drop_duplicates()
                 else:
@@ -265,6 +277,13 @@ def hop(self: Plottable,
             new_node_ids_reverse = hop_edges_reverse[[g2._source]].rename(columns={g2._source: g2._node}).drop_duplicates()
 
             if destination_node_query is not None or destination_node_match is not None:
+                if debugging_hop and logger.isEnabledFor(logging.DEBUG):
+                    logger.debug('--- destination predicate filtering ---')
+                    logger.debug('destination_node_query: %s', destination_node_query)
+                    logger.debug('destination_node_match: %s', destination_node_match)
+                    logger.debug('base_target_nodes:\n%s', base_target_nodes)
+                    logger.debug('new_node_ids_reverse:\n%s', new_node_ids_reverse)
+                    logger.debug('enriched nodes for filtering:\n%s', base_target_nodes.merge(new_node_ids_reverse, on=g2._node, how='inner'))
                 new_node_ids_reverse = query_if_not_none(
                     destination_node_query,
                     filter_by_dict(
@@ -277,6 +296,7 @@ def hop(self: Plottable,
                     on=g2._source
                 )
                 if debugging_hop and logger.isEnabledFor(logging.DEBUG):
+                    logger.debug('new_node_ids_reverse:\n%s', new_node_ids_reverse)
                     logger.debug('hop_edges_reverse filtered by destination predicates:\n%s', hop_edges_reverse)
             
             if debugging_hop and logger.isEnabledFor(logging.DEBUG):
@@ -305,10 +325,10 @@ def hop(self: Plottable,
             logger.debug('hop_edges_forward:\n%s', hop_edges_forward)
             logger.debug('hop_edges_reverse:\n%s', hop_edges_reverse)
 
-        # Finally include all initial root nodes matched against, now that edge triples satisfy all source/dest/edge predicates
-        # Only run first iteration b/c root nodes already accounted for in subsequent
-        # In wavefront mode, skip, as we only want to return reached nodes
-        if matches_nodes is None:
+        # When !return_as_wave_front, include starting nodes in returned matching node set
+        # (When return_as_wave_front, skip starting nodes, just include newly reached)
+        # Only need to do this in the first loop step
+        if matches_nodes is None:  # first iteration
             if return_as_wave_front:
                 matches_nodes = new_node_ids[:0]
             else:
@@ -326,7 +346,10 @@ def hop(self: Plottable,
                 logger.debug('~~~~~~~~~~ LOOP STEP MERGES 2 ~~~~~~~~~~~')
                 logger.debug('matches_edges:\n%s', matches_edges)
 
-        combined_node_ids = concat([matches_nodes, new_node_ids], ignore_index=True, sort=False).drop_duplicates()
+        if len(matches_nodes) > 0:
+            combined_node_ids = concat([matches_nodes, new_node_ids], ignore_index=True, sort=False).drop_duplicates()
+        else:
+            combined_node_ids = new_node_ids
 
         if len(combined_node_ids) == len(matches_nodes):
             #fixedpoint, exit early: future will come to same spot!
@@ -369,6 +392,10 @@ def hop(self: Plottable,
         if target_wave_front is not None:
             rich_nodes = concat([rich_nodes, target_wave_front], ignore_index=True, sort=False).drop_duplicates(subset=[g2._node])
         logger.debug('rich_nodes available for inner merge:\n%s', rich_nodes[[self._node]])
+        logger.debug('target_wave_front:\n%s', target_wave_front)
+        logger.debug('matches_nodes:\n%s', matches_nodes)
+        logger.debug('wave_front:\n%s', wave_front)
+        logger.debug('self._nodes:\n%s', self._nodes)
         final_nodes = rich_nodes.merge(
             matches_nodes if matches_nodes is not None else wave_front[:0],
             on=self._node,


Test patch:
diff --git a/graphistry/tests/compute/test_chain.py b/graphistry/tests/compute/test_chain.py
index ea3fb232e9..d38e0af1a2 100644
--- a/graphistry/tests/compute/test_chain.py
+++ b/graphistry/tests/compute/test_chain.py
@@ -1,13 +1,254 @@
 import os
 import pandas as pd
 from graphistry.compute.predicates.is_in import is_in
+from graphistry.compute.predicates.numeric import gt
 import pytest
 
-from graphistry.compute.ast import ASTNode, ASTEdge, n, e
+from graphistry.compute.ast import ASTNode, ASTEdge, n, e, e_undirected, e_forward
 from graphistry.compute.chain import Chain
 from graphistry.tests.test_compute import CGFull
 
 
+@pytest.fixture(scope='module')
+def g_long_forwards_chain():
+    """
+    a->b->c->d->e
+    """
+    return (CGFull()
+        .edges(pd.DataFrame({
+            's': ['a', 'b', 'c', 'd'],
+            'd': ['b', 'c', 'd', 'e'],
+            't': ['1', '2', '3', '4'],
+            'e': ['2', '3', '4', '5']}),
+            's', 'd')
+        .nodes(pd.DataFrame({
+            'v': ['a', 'b', 'c', 'd', 'e'],
+            'w': ['1', '2', '3', '4', '5']}),
+            'v'))
+
+@pytest.fixture(scope='module')
+def g_long_forwards_chain_dead_end():
+    """
+    a->b->c->d->e
+          c->x
+    """
+    return (CGFull()
+        .edges(pd.DataFrame({
+            's': ['a', 'b', 'c', 'd', 'c'],
+            'd': ['b', 'c', 'd', 'e', 'x']}),
+            's', 'd')
+        .nodes(pd.DataFrame({
+            'v': ['a', 'b', 'c', 'd', 'e', 'x']}),
+            'v'))
+
+@pytest.fixture(scope='module')
+def g_long_forwards_chain_loop():
+    """
+    a->b->c->d->e
+       c->x->c
+    """
+    return (CGFull()
+        .edges(pd.DataFrame({
+            's': ['a', 'b', 'c', 'd', 'c', 'x'],
+            'd': ['b', 'c', 'd', 'e', 'x', 'c']}),
+            's', 'd')
+        .nodes(pd.DataFrame({
+            'v': ['a', 'b', 'c', 'd', 'e', 'x']}),
+            'v'))
+
+class TestMultiHopChainForward():
+
+    def test_chain_short(self, g_long_forwards_chain):
+        g2 = g_long_forwards_chain.chain([n({'v': 'a'}), e_forward(hops=2), n({'v': 'd'})])
+        assert len(g2._nodes) == 0
+        assert len(g2._edges) == 0
+    
+    def test_chain_exact(self, g_long_forwards_chain):
+        g2 = g_long_forwards_chain.chain([n({'v': 'a'}), e_forward(hops=3), n({'v': 'd'})])
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).reset_index(drop=True).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+    def test_chain_long(self, g_long_forwards_chain):
+        g2 = g_long_forwards_chain.chain([n({'v': 'a'}), e_forward(hops=4), n({'v': 'd'})])
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).reset_index(drop=True).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+    def test_chain_fixedpoint(self, g_long_forwards_chain):
+        g2 = g_long_forwards_chain.chain([n({'v': 'a'}), e_forward(to_fixed_point=True), n({'v': 'd'})])
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).reset_index(drop=True).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+    def test_chain_predicates_ok_source(self, g_long_forwards_chain):
+        g2 = g_long_forwards_chain.chain([
+            n({'v': 'a'}),
+            e_forward(
+                source_node_match={'w': is_in(['1', '2', '3'])},
+                hops=3
+            ),
+            n({'v': 'd'})
+        ])
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).reset_index(drop=True).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+    def test_chain_predicates_ok_edge(self, g_long_forwards_chain):
+        g2 = g_long_forwards_chain.chain([
+            n({'v': 'a'}),
+            e_forward(
+                edge_match={
+                    't': is_in(['1', '2', '3']),
+                    'e': is_in(['2', '3', '4'])
+                },
+                hops=3
+            ),
+            n({'v': 'd'})
+        ])
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).reset_index(drop=True).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+    def test_chain_predicates_ok_destination(self, g_long_forwards_chain):
+        g2 = g_long_forwards_chain.chain([
+            n({'v': 'a'}),
+            e_forward(
+                destination_node_match={'w': is_in(['2', '3', '4'])},
+                hops=3
+            ),
+            n({'v': 'd'})
+        ])
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).reset_index(drop=True).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+    def test_chain_predicates_ok(self, g_long_forwards_chain):
+        g2 = g_long_forwards_chain.chain([
+            n({'v': 'a'}),
+            e_forward(
+                source_node_match={'w': is_in(['1', '2', '3'])},
+                edge_match={
+                    't': is_in(['1', '2', '3']),
+                    'e': is_in(['2', '3', '4'])
+                },
+                destination_node_match={'w': is_in(['2', '3', '4'])},
+                hops=3
+            ),
+            n({'v': 'd'})
+        ])
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).reset_index(drop=True).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+    def test_chain_predicates_source_fail(self, g_long_forwards_chain):
+        BAD = []
+        g2 = g_long_forwards_chain.chain([
+            n({'v': 'a'}),
+            e_forward(
+                source_node_match={'w': is_in(BAD)},
+                edge_match={
+                    't': is_in(['1', '2', '3']),
+                    'e': is_in(['2', '3', '4'])
+                },
+                destination_node_match={'w': is_in(['2', '3', '4'])},
+                hops=3
+            ),
+            n({'v': 'd'})
+        ])
+        assert len(g2._nodes) == 0
+        assert len(g2._edges) == 0
+
+    def test_chain_predicates_dest_fail(self, g_long_forwards_chain):
+        BAD = []
+        g2 = g_long_forwards_chain.chain([
+            n({'v': 'a'}),
+            e_forward(
+                source_node_match={'w': is_in(['1', '2', '3'])},
+                edge_match={
+                    't': is_in(['1', '2', '3']),
+                    'e': is_in(['2', '3', '4'])
+                },
+                destination_node_match={'w': is_in(BAD)},
+                hops=3
+            ),
+            n({'v': 'd'})
+        ])
+        assert len(g2._nodes) == 0
+        assert len(g2._edges) == 0
+
+    def test_chain_predicates_edge_fail(self, g_long_forwards_chain):
+        BAD = []
+        g2 = g_long_forwards_chain.chain([
+            n({'v': 'a'}),
+            e_forward(
+                source_node_match={'w': is_in(['1', '2', '3'])},
+                edge_match={
+                    't': is_in(BAD),
+                    'e': is_in(['2', '3', '4'])
+                },
+                destination_node_match={'w': is_in(['2', '3', '4'])},
+                hops=3
+            ),
+            n({'v': 'd'})
+        ])
+        assert len(g2._nodes) == 0
+        assert len(g2._edges) == 0
+
+
+class TestMultiHopDeadend():
+
+    def test_chain_fixedpoint(self, g_long_forwards_chain_dead_end: CGFull):
+        """
+        Same as chain; x should not be considered a hint
+        """
+        g2 = g_long_forwards_chain_dead_end.chain([n({'v': 'a'}), e_forward(to_fixed_point=True), n({'v': 'd'})])
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).reset_index(drop=True).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+
+class TestMultiHopLoop():
+
+    def test_chain_fixedpoint(self, g_long_forwards_chain_loop: CGFull):
+        """
+        Same as chain; + detour using x
+        """
+        g2 = g_long_forwards_chain_loop.chain([n({'v': 'a'}), e_forward(to_fixed_point=True), n({'v': 'd'})])
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c', 'd', 'x'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).reset_index(drop=True).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'},
+            {'s': 'c', 'd': 'x'},
+            {'s': 'x', 'd': 'c'}
+        ]
+
+
 def test_chain_serialization_mt():
     o = Chain([]).to_json()
     d = Chain.from_json(o)
@@ -138,3 +379,40 @@ def test_chain_pred_cudf():
     g_edges = g.chain([e({'src': is_in([0])})])
     assert isinstance(g_edges._edges, cudf.DataFrame)
     assert len(g_edges._edges) == 1
+
+def test_preds_more_pd():
+
+    # a->b3->c1
+    #    U
+    edf = pd.DataFrame({
+        's': ['a1', 'b3', 'b3'],
+        'd': ['b3', 'b3', 'c1']
+    })
+    g = CGFull().edges(edf, 's', 'd').materialize_nodes().get_degrees()
+
+    g2 = (g.get_degrees()
+        .chain([
+            n({'degree': gt(1)}),
+            e_undirected(),
+            n({'degree': gt(1)})
+        ])
+    )
+    assert set(g2._nodes[g2._node].tolist()) == set(['b3'])
+
+def test_preds_more_pd_2():
+
+    edf = pd.DataFrame({
+        's': ['a1', 'b2', 'c2'],
+        'd': ['b2', 'c2', 'd1']
+    })
+    g = CGFull().edges(edf, 's', 'd').materialize_nodes().get_degrees()
+
+    g2 = (g.get_degrees()
+        .chain([
+            n({'degree': gt(1)}),
+            e_undirected(),
+            n({'degree': gt(1)})
+        ])
+    )
+    assert len(g2._nodes) == 2
+    assert set(g2._nodes[g._node].tolist()) == set(['b2', 'c2'])
diff --git a/graphistry/tests/compute/test_hop.py b/graphistry/tests/compute/test_hop.py
index 2960c58325..232421c88d 100644
--- a/graphistry/tests/compute/test_hop.py
+++ b/graphistry/tests/compute/test_hop.py
@@ -7,6 +7,414 @@
 from graphistry.tests.test_compute import CGFull
 
 
+@pytest.fixture(scope='module')
+def g_long_forwards_chain() -> CGFull:
+    """
+    a->b->c->d->e
+    """
+    return (CGFull()
+        .edges(pd.DataFrame({
+            's': ['a', 'b', 'c', 'd'],
+            'd': ['b', 'c', 'd', 'e'],
+            't': ['1', '2', '3', '4'],
+            'e': ['2', '3', '4', '5']}),
+            's', 'd')
+        .nodes(pd.DataFrame({
+            'v': ['a', 'b', 'c', 'd', 'e'],
+            'w': ['1', '2', '3', '4', '5']}),
+            'v'))
+
+@pytest.fixture(scope='module')
+def n_a(g_long_forwards_chain: CGFull) -> pd.DataFrame:
+    return g_long_forwards_chain._nodes.query('v == "a"')
+
+
+@pytest.fixture(scope='module')
+def n_mt(g_long_forwards_chain: CGFull) -> pd.DataFrame:
+    return g_long_forwards_chain._nodes[:0]
+
+@pytest.fixture(scope='module')
+def n_d(g_long_forwards_chain: CGFull) -> pd.DataFrame:
+    return g_long_forwards_chain._nodes.query('v == "d"')
+
+
+class TestMultiHopForward():
+    """
+    Test multi-hop as used by chain, corresponding to chain multi-hop tests
+    """
+
+    def test_hop_short_forward(self, g_long_forwards_chain: CGFull, n_a):
+        g2 = g_long_forwards_chain.hop(
+            nodes=n_a,
+            hops=2,
+            to_fixed_point=False,
+            direction='forward',
+            return_as_wave_front=True
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['b', 'c'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'}
+        ]
+
+    def test_hop_short_back(self, g_long_forwards_chain: CGFull, n_mt, n_a):
+        g_reverse = g_long_forwards_chain.nodes(
+            g_long_forwards_chain._nodes[
+                g_long_forwards_chain._nodes['v'].isin(['b', 'c'])
+            ]).edges(
+                g_long_forwards_chain._edges.merge(
+                    pd.DataFrame({
+                        's': ['a', 'b'],
+                        'd': ['b', 'c']}),
+                    on=['s', 'd'],
+                    how='inner'
+                ))
+        g2 = g_reverse.hop(
+            nodes=n_mt,
+            hops=2,
+            to_fixed_point=False,
+            direction='reverse',
+            return_as_wave_front=True,
+            target_wave_front=n_a
+        )
+        assert len(g2._nodes) == 0
+        assert len(g2._edges) == 0
+
+    def test_hop_exact_forward(self, g_long_forwards_chain, n_a):
+
+        g2 = g_long_forwards_chain.hop(
+            nodes=n_a,
+            hops=3,
+            to_fixed_point=False,
+            direction='forward',
+            return_as_wave_front=True
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+    def test_hop_exact_back(self, g_long_forwards_chain: CGFull, n_d, n_a):
+        g_reverse = g_long_forwards_chain.nodes(
+            g_long_forwards_chain._nodes[
+                g_long_forwards_chain._nodes['v'].isin(['a', 'b', 'c', 'd'])
+            ]).edges(
+                g_long_forwards_chain._edges.merge(
+                    pd.DataFrame({
+                        's': ['a', 'b', 'c'],
+                        'd': ['b', 'c', 'd']}),
+                    on=['s', 'd'],
+                    how='inner'
+                ))
+        g2 = g_reverse.hop(
+            nodes=n_d,
+            hops=3,
+            to_fixed_point=False,
+            direction='reverse',
+            return_as_wave_front=True,
+            target_wave_front=n_a
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+
+    def test_hop_long_forward(self, g_long_forwards_chain, n_a):
+
+        g2 = g_long_forwards_chain.hop(
+            nodes=n_a,
+            hops=4,
+            to_fixed_point=False,
+            direction='forward',
+            return_as_wave_front=True
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['b', 'c', 'd', 'e'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'},
+            {'s': 'd', 'd': 'e'}
+        ]
+
+    def test_hop_long_back(self, g_long_forwards_chain: CGFull, n_d, n_a):
+        g_reverse = g_long_forwards_chain.nodes(
+            g_long_forwards_chain._nodes[
+                g_long_forwards_chain._nodes['v'].isin(['b', 'c', 'd', 'e'])
+            ]).edges(
+                g_long_forwards_chain._edges.merge(
+                    pd.DataFrame({
+                        's': ['a', 'b', 'c', 'd'],
+                        'd': ['b', 'c', 'd', 'e']}),
+                    on=['s', 'd'],
+                    how='inner'
+                ))
+        g2 = g_reverse.hop(
+            nodes=n_d,
+            hops=4,
+            to_fixed_point=False,
+            direction='reverse',
+            return_as_wave_front=True,
+            target_wave_front=n_a
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'}
+        ]
+
+    def test_hop_predicates_ok_source_forward(self, g_long_forwards_chain: CGFull, n_a):
+
+        g2 = g_long_forwards_chain.hop(
+            nodes=n_a,
+            hops=4,
+            to_fixed_point=False,
+            source_node_match={'w': is_in(['1', '2', '3'])},
+            direction='forward',
+            return_as_wave_front=True
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'},
+        ]
+
+    def test_hop_predicates_ok_source_back(self, g_long_forwards_chain: CGFull, n_a, n_d):
+
+        g_reverse = g_long_forwards_chain.nodes(
+            g_long_forwards_chain._nodes[
+                g_long_forwards_chain._nodes['v'].isin(['b', 'c', 'd'])
+            ]).edges(
+                g_long_forwards_chain._edges.merge(
+                    pd.DataFrame({
+                        's': ['a', 'b', 'c'],
+                        'd': ['b', 'c', 'd']}),
+                    on=['s', 'd'],
+                    how='inner'
+                ))
+
+        g2 = g_reverse.hop(
+            nodes=n_d,
+            hops=4,
+            to_fixed_point=False,
+            destination_node_match={'w': is_in(['1', '2', '3'])},
+            direction='reverse',
+            return_as_wave_front=True,
+            target_wave_front=n_a
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'},
+        ]
+
+    def test_hop_predicates_ok_edge_forward(self, g_long_forwards_chain: CGFull, n_a):
+
+        g2 = g_long_forwards_chain.hop(
+            nodes=n_a,
+            hops=4,
+            to_fixed_point=False,
+            edge_match={
+                't': is_in(['1', '2', '3']),
+                'e': is_in(['2', '3', '4'])
+            },
+            direction='forward',
+            return_as_wave_front=True
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'},
+        ]
+
+    def test_hop_predicates_ok_edge_back(self, g_long_forwards_chain: CGFull, n_a, n_d):
+
+        g_reverse = g_long_forwards_chain.nodes(
+            g_long_forwards_chain._nodes[
+                g_long_forwards_chain._nodes['v'].isin(['b', 'c', 'd'])
+            ]).edges(
+                g_long_forwards_chain._edges.merge(
+                    pd.DataFrame({
+                        's': ['a', 'b', 'c'],
+                        'd': ['b', 'c', 'd']}),
+                    on=['s', 'd'],
+                    how='inner'
+                ))
+
+        g2 = g_reverse.hop(
+            nodes=n_d,
+            hops=4,
+            to_fixed_point=False,
+            edge_match={
+                't': is_in(['1', '2', '3']),
+                'e': is_in(['2', '3', '4'])
+            },
+            direction='reverse',
+            return_as_wave_front=True,
+            target_wave_front=n_a
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'},
+        ]
+
+    def test_hop_predicates_ok_destination_forward(self, g_long_forwards_chain: CGFull, n_a):
+
+        g2 = g_long_forwards_chain.hop(
+            nodes=n_a,
+            hops=4,
+            to_fixed_point=False,
+            destination_node_match={'w': is_in(['2', '3', '4'])},
+            direction='forward',
+            return_as_wave_front=True
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'},
+        ]
+
+    def test_hop_predicates_ok_destination_back(self, g_long_forwards_chain: CGFull, n_a, n_d):
+
+        g_reverse = g_long_forwards_chain.nodes(
+            g_long_forwards_chain._nodes[
+                g_long_forwards_chain._nodes['v'].isin(['b', 'c', 'd'])
+            ]).edges(
+                g_long_forwards_chain._edges.merge(
+                    pd.DataFrame({
+                        's': ['a', 'b', 'c'],
+                        'd': ['b', 'c', 'd']}),
+                    on=['s', 'd'],
+                    how='inner'
+                ))
+
+        g2 = g_reverse.hop(
+            nodes=n_d,
+            hops=4,
+            to_fixed_point=False,
+            source_node_match={'w': is_in(['2', '3', '4'])},
+            direction='reverse',
+            return_as_wave_front=True,
+            target_wave_front=n_a
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'},
+        ]
+
+    def test_hop_predicates_ok_forward(self, g_long_forwards_chain: CGFull, n_a):
+
+        g2 = g_long_forwards_chain.hop(
+            nodes=n_a,
+            hops=4,
+            to_fixed_point=False,
+            source_node_match={'w': is_in(['1', '2', '3'])},
+            edge_match={
+                't': is_in(['1', '2', '3']),
+                'e': is_in(['2', '3', '4'])
+            },
+            destination_node_match={'w': is_in(['2', '3', '4'])},
+            direction='forward',
+            return_as_wave_front=True
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['b', 'c', 'd'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'},
+        ]
+
+    def test_hop_predicates_ok_back(self, g_long_forwards_chain: CGFull, n_a, n_d):
+
+        g_reverse = g_long_forwards_chain.nodes(
+            g_long_forwards_chain._nodes[
+                g_long_forwards_chain._nodes['v'].isin(['b', 'c', 'd'])
+            ]).edges(
+                g_long_forwards_chain._edges.merge(
+                    pd.DataFrame({
+                        's': ['a', 'b', 'c'],
+                        'd': ['b', 'c', 'd']}),
+                    on=['s', 'd'],
+                    how='inner'
+                ))
+
+        g2 = g_reverse.hop(
+            nodes=n_d,
+            hops=4,
+            to_fixed_point=False,
+            destination_node_match={'w': is_in(['1', '2', '3'])},
+            edge_match={
+                't': is_in(['1', '2', '3']),
+                'e': is_in(['2', '3', '4'])
+            },
+            source_node_match={'w': is_in(['2', '3', '4'])},
+            direction='reverse',
+            return_as_wave_front=True,
+            target_wave_front=n_a
+        )
+        assert set(g2._nodes['v'].tolist()) == set(['a', 'b', 'c'])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == [
+            {'s': 'a', 'd': 'b'},
+            {'s': 'b', 'd': 'c'},
+            {'s': 'c', 'd': 'd'},
+        ]
+
+    def test_hop_predicates_fail_source_forward(self, g_long_forwards_chain: CGFull, n_a):
+
+        g2 = g_long_forwards_chain.hop(
+            nodes=n_a,
+            hops=4,
+            to_fixed_point=False,
+            source_node_match={'w': is_in([])},
+            direction='forward',
+            return_as_wave_front=True
+        )
+        assert set(g2._nodes['v'].tolist()) == set([])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == []
+
+    def test_hop_predicates_fail_edge_forward(self, g_long_forwards_chain: CGFull, n_a):
+
+        g2 = g_long_forwards_chain.hop(
+            nodes=n_a,
+            hops=4,
+            to_fixed_point=False,
+            edge_match={
+                't': is_in([]),
+                'e': is_in([])
+            },
+            direction='forward',
+            return_as_wave_front=True
+        )
+        assert set(g2._nodes['v'].tolist()) == set([])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == []
+
+    def test_hop_predicates_fail_destination_forward(self, g_long_forwards_chain: CGFull, n_a):
+
+        g2 = g_long_forwards_chain.hop(
+            nodes=n_a,
+            hops=4,
+            to_fixed_point=False,
+            destination_node_match={'w': is_in([])},
+            direction='forward',
+            return_as_wave_front=True
+        )
+        assert set(g2._nodes['v'].tolist()) == set([])
+        assert g2._edges[['s', 'd']].sort_values(['s', 'd']).to_dict(orient='records') == []
+
+
 def test_hop_simple_cudf_pd():
     nodes_df = pd.DataFrame({'id': [0, 1, 2], 'label': ['a', 'b', 'c']})
     edges_df = pd.DataFrame({'src': [0, 1, 2], 'dst': [1, 2, 0]})
diff --git a/graphistry/tests/test_compute_chain.py b/graphistry/tests/test_compute_chain.py
index fb5d1da20d..a5c7c33087 100644
--- a/graphistry/tests/test_compute_chain.py
+++ b/graphistry/tests/test_compute_chain.py
@@ -192,9 +192,19 @@ def test_shortest_path(self):
         assert g3a._edges.shape == (1, 3)
         compare_graphs(g3a, g_out_nodes_1_hop, g_out_edges_1_hop)
 
+        #a->b
+        g3ba = g.chain([n({'n': 'a'}), e_forward(hops=1), n({'n': 'b'})])
+        assert g3ba._nodes.shape == (2, 2)
+        assert g3ba._edges.shape == (1, 3)
+
+        #a->b-c
+        g3baa = g.chain([n({'n': 'a'}), e_forward(hops=2)])
+        assert g3baa._nodes.shape == (3, 2)
+        assert g3baa._edges.shape == (2, 3)
+
         g3b = g.chain([n({'n': 'a'}), e_forward(hops=2), n({'n': 'c'})])
-        assert g3b._nodes.shape == (3, 2)
-        assert g3b._edges.shape == (2, 3)
+        assert g3b._nodes.shape == (3, 2), "nodes"
+        assert g3b._edges.shape == (2, 3), "edges"
         compare_graphs(g3b, g_out_nodes_2_hops, g_out_edges_2_hops)
 
         g3c = g.chain([n({'n': 'a'}), e_undirected(hops=2), n({'n': 'c'})])