diff --git a/graphistry/compute/__init__.py b/graphistry/compute/__init__.py index 726b1c0706..3cbb68ac16 100644 --- a/graphistry/compute/__init__.py +++ b/graphistry/compute/__init__.py @@ -1,7 +1,7 @@ from .ComputeMixin import ComputeMixin from .ast import ( n, e, e_forward, e_reverse, e_undirected, - let, remote, ref + let, remote, ref, call ) from .chain import Chain from .predicates.is_in import ( @@ -61,7 +61,7 @@ 'ComputeMixin', 'Chain', # AST nodes 'n', 'e', 'e_forward', 'e_reverse', 'e_undirected', - 'let', 'remote', 'ref', + 'let', 'remote', 'ref', 'call', # Predicates 'is_in', 'IsIn', 'duplicated', 'Duplicated', diff --git a/graphistry/compute/ast.py b/graphistry/compute/ast.py index 83a5bb62ae..5ffa53ff06 100644 --- a/graphistry/compute/ast.py +++ b/graphistry/compute/ast.py @@ -1,6 +1,6 @@ from abc import abstractmethod import logging -from typing import TYPE_CHECKING, Dict, List, Optional, Union, cast +from typing import Any, TYPE_CHECKING, Dict, List, Optional, Union, cast from typing_extensions import Literal if TYPE_CHECKING: @@ -738,9 +738,119 @@ def reverse(self) -> 'ASTRef': return ASTRef(self.ref, [op.reverse() for op in reversed(self.chain)]) +class ASTCall(ASTObject): + """Call a method on the current graph with validated parameters. + + Allows safe execution of Plottable methods through GFQL with parameter + validation and schema checking. + + Attributes: + function: Name of the method to call (must be in safelist) + params: Dictionary of parameters to pass to the method + """ + def __init__(self, function: str, params: Optional[Dict[str, Any]] = None): + """Initialize a Call operation. + + Args: + function: Name of the Plottable method to call + params: Optional dictionary of parameters for the method + """ + super().__init__() + self.function = function + self.params = params or {} + + def _validate_fields(self) -> None: + """Validate Call fields.""" + from graphistry.compute.exceptions import ErrorCode, GFQLTypeError + + if not isinstance(self.function, str): + raise GFQLTypeError( + ErrorCode.E201, + "function must be a string", + field="function", + value=type(self.function).__name__ + ) + + if len(self.function) == 0: + raise GFQLTypeError( + ErrorCode.E106, + "function name cannot be empty", + field="function", + value=self.function + ) + + if not isinstance(self.params, dict): + raise GFQLTypeError( + ErrorCode.E201, + "params must be a dictionary", + field="params", + value=type(self.params).__name__ + ) + + def to_json(self, validate=True) -> dict: + """Convert Call to JSON representation. + + Args: + validate: If True, validate before serialization + + Returns: + Dictionary with type, function, and params fields + """ + if validate: + self.validate() + return { + 'type': 'Call', + 'function': self.function, + 'params': self.params + } + + @classmethod + def from_json(cls, d: dict, validate: bool = True) -> 'ASTCall': + assert 'function' in d, "Call missing function" + out = cls( + function=d['function'], + params=d.get('params', {}) + ) + if validate: + out.validate() + return out + + def __call__(self, g: Plottable, prev_node_wavefront: Optional[DataFrameT], + target_wave_front: Optional[DataFrameT], engine: Engine) -> Plottable: + """Execute the method call on the graph. + + Args: + g: Graph to operate on + prev_node_wavefront: Previous node wavefront (unused) + target_wave_front: Target wavefront (unused) + engine: Execution engine (pandas/cudf) + + Returns: + New Plottable with method results + + Raises: + GFQLTypeError: If method not in safelist or parameters invalid + """ + # For chain_dag, we don't use wavefronts, just execute the call + from graphistry.compute.call_executor import execute_call + return execute_call(g, self.function, self.params, engine) + + def reverse(self) -> 'ASTCall': + """Reverse is not supported for Call operations. + + Most Plottable methods are not reversible as they perform + transformations that cannot be undone. + + Raises: + NotImplementedError: Always raised as calls cannot be reversed + """ + # Most method calls cannot be reversed + raise NotImplementedError(f"Method '{self.function}' cannot be reversed") + + ### -def from_json(o: JSONVal, validate: bool = True) -> Union[ASTNode, ASTEdge, ASTLet, ASTRemoteGraph, ASTRef]: +def from_json(o: JSONVal, validate: bool = True) -> Union[ASTNode, ASTEdge, ASTLet, ASTRemoteGraph, ASTRef, ASTCall]: from graphistry.compute.exceptions import ErrorCode, GFQLSyntaxError if not isinstance(o, dict): @@ -748,10 +858,10 @@ def from_json(o: JSONVal, validate: bool = True) -> Union[ASTNode, ASTEdge, ASTL if 'type' not in o: raise GFQLSyntaxError( - ErrorCode.E105, "AST JSON missing required 'type' field", suggestion="Add 'type' field: 'Node', 'Edge', 'QueryDAG', 'RemoteGraph', or 'Ref'" + ErrorCode.E105, "AST JSON missing required 'type' field", suggestion="Add 'type' field: 'Node', 'Edge', 'QueryDAG', 'RemoteGraph', 'Ref', or 'Call'" ) - out: Union[ASTNode, ASTEdge, ASTLet, ASTRemoteGraph, ASTRef] + out: Union[ASTNode, ASTEdge, ASTLet, ASTRemoteGraph, ASTRef, ASTCall] if o['type'] == 'Node': out = ASTNode.from_json(o, validate=validate) elif o['type'] == 'Edge': @@ -783,13 +893,15 @@ def from_json(o: JSONVal, validate: bool = True) -> Union[ASTNode, ASTEdge, ASTL out = ASTRemoteGraph.from_json(o, validate=validate) elif o['type'] == 'Ref': out = ASTRef.from_json(o, validate=validate) + elif o['type'] == 'Call': + out = ASTCall.from_json(o, validate=validate) else: raise GFQLSyntaxError( ErrorCode.E101, f"Unknown AST type: {o['type']}", field="type", value=o["type"], - suggestion="Use 'Node', 'Edge', 'Let', 'RemoteGraph', or 'Ref'", + suggestion="Use 'Node', 'Edge', 'Let', 'RemoteGraph', 'Ref', or 'Call'", ) return out @@ -800,3 +912,4 @@ def from_json(o: JSONVal, validate: bool = True) -> Union[ASTNode, ASTEdge, ASTL let = ASTLet # noqa: E305 remote = ASTRemoteGraph # noqa: E305 ref = ASTRef # noqa: E305 +call = ASTCall # noqa: E305 diff --git a/graphistry/compute/call_executor.py b/graphistry/compute/call_executor.py new file mode 100644 index 0000000000..67ac04e031 --- /dev/null +++ b/graphistry/compute/call_executor.py @@ -0,0 +1,83 @@ +"""Execute validated method calls on Plottable objects. + +This module provides the execution layer for GFQL Call operations after +parameter validation has been performed by the safelist module. +""" + +from typing import Dict, Any + +from graphistry.Plottable import Plottable +from graphistry.Engine import Engine +from graphistry.compute.call_safelist import validate_call_params +from graphistry.compute.exceptions import ErrorCode, GFQLTypeError + + +def execute_call(g: Plottable, function: str, params: Dict[str, Any], engine: Engine) -> Plottable: + """Execute a validated method call on a Plottable. + + Args: + g: The graph to call the method on + function: Name of the method to call + params: Parameters for the method (will be validated) + engine: Execution engine + + Returns: + Result of the method call (usually a new Plottable) + + Raises: + GFQLTypeError: If validation fails or method doesn't exist + AttributeError: If method doesn't exist on Plottable + """ + # Validate parameters against safelist + validated_params = validate_call_params(function, params) + + # Check if method exists on Plottable + if not hasattr(g, function): + raise AttributeError( + f"Plottable has no method '{function}'. " + f"This should not happen if safelist is properly configured." + ) + + # Get the method + method = getattr(g, function) + + # Special handling for methods that need the engine parameter + if function in ['materialize_nodes', 'hop']: + # These methods accept an engine parameter + if 'engine' not in validated_params: + # Add current engine if not specified + validated_params['engine'] = engine + + try: + # Execute the method with validated parameters + result = method(**validated_params) + + # Ensure result is a Plottable (most methods return self or new Plottable) + if not isinstance(result, Plottable): + raise GFQLTypeError( + ErrorCode.E201, + f"Method '{function}' returned non-Plottable result", + field="function", + value=f"{type(result).__name__}", + suggestion="Only methods that return Plottable objects are allowed" + ) + + return result + + except TypeError as e: + # Handle parameter mismatch errors + raise GFQLTypeError( + ErrorCode.E201, + f"Parameter error calling '{function}': {str(e)}", + field="params", + value=validated_params, + suggestion="Check parameter names and types" + ) from e + except Exception as e: + # Re-raise other exceptions with context + raise GFQLTypeError( + ErrorCode.E303, + f"Error executing '{function}': {str(e)}", + field="function", + value=function + ) from e diff --git a/graphistry/compute/call_safelist.py b/graphistry/compute/call_safelist.py new file mode 100644 index 0000000000..3b3ae3d304 --- /dev/null +++ b/graphistry/compute/call_safelist.py @@ -0,0 +1,519 @@ +"""Safelist of allowed methods for GFQL Call operations. + +This module defines which Plottable methods can be called through GFQL +and their parameter validation rules. +""" + +from typing import Dict, Any, Set, Optional, Union, Type +from graphistry.compute.exceptions import ErrorCode, GFQLTypeError + + +# Type validators +def is_string(v: Any) -> bool: + """Check if value is a string. + + Args: + v: Value to check + + Returns: + True if v is a string, False otherwise + """ + return isinstance(v, str) + + +def is_int(v: Any) -> bool: + """Check if value is an integer. + + Args: + v: Value to check + + Returns: + True if v is an integer, False otherwise + """ + return isinstance(v, int) + + +def is_bool(v: Any) -> bool: + """Check if value is a boolean. + + Args: + v: Value to check + + Returns: + True if v is a boolean, False otherwise + """ + return isinstance(v, bool) + + +def is_dict(v: Any) -> bool: + """Check if value is a dictionary. + + Args: + v: Value to check + + Returns: + True if v is a dictionary, False otherwise + """ + return isinstance(v, dict) + + +def is_string_or_none(v: Any) -> bool: + """Check if value is a string or None. + + Args: + v: Value to check + + Returns: + True if v is a string or None, False otherwise + """ + return v is None or isinstance(v, str) + + +def is_list_of_strings(v: Any) -> bool: + """Check if value is a list of strings. + + Args: + v: Value to check + + Returns: + True if v is a list containing only strings, False otherwise + """ + return isinstance(v, list) and all(isinstance(item, str) for item in v) + + +# Safelist configuration +# Each entry defines: +# - allowed_params: Set of parameter names that can be passed +# - required_params: Set of parameters that must be provided +# - param_validators: Dict of param_name -> validator function +# - description: Human-readable description of what the method does +# - schema_effects: Dict describing columns added/removed/required +# - adds_node_cols: List of columns added to nodes +# - adds_edge_cols: List of columns added to edges +# - requires_node_cols: List of node columns that must exist +# - requires_edge_cols: List of edge columns that must exist + +SAFELIST_V1: Dict[str, Dict[str, Any]] = { + 'get_degrees': { + 'allowed_params': {'col_in', 'col_out', 'col'}, + 'required_params': set(), + 'param_validators': { + 'col_in': is_string, + 'col_out': is_string, + 'col': is_string + }, + 'description': 'Calculate node degrees', + 'schema_effects': { + 'adds_node_cols': lambda p: [ + p.get('col', 'degree'), + p.get('col_in', 'degree_in'), + p.get('col_out', 'degree_out') + ], + 'adds_edge_cols': [], + 'requires_node_cols': [], + 'requires_edge_cols': [] + } + }, + + 'filter_nodes_by_dict': { + 'allowed_params': {'filter_dict'}, + 'required_params': {'filter_dict'}, + 'param_validators': { + 'filter_dict': is_dict + }, + 'description': 'Filter nodes by attribute values', + 'schema_effects': { + 'adds_node_cols': [], + 'adds_edge_cols': [], + 'requires_node_cols': lambda p: list(p.get('filter_dict', {}).keys()), + 'requires_edge_cols': [] + } + }, + + 'filter_edges_by_dict': { + 'allowed_params': {'filter_dict'}, + 'required_params': {'filter_dict'}, + 'param_validators': { + 'filter_dict': is_dict + }, + 'description': 'Filter edges by attribute values', + 'schema_effects': { + 'adds_node_cols': [], + 'adds_edge_cols': [], + 'requires_node_cols': [], + 'requires_edge_cols': lambda p: list(p.get('filter_dict', {}).keys()) + } + }, + + 'materialize_nodes': { + 'allowed_params': {'engine', 'reuse'}, + 'required_params': set(), + 'param_validators': { + 'engine': is_string, + 'reuse': is_bool + }, + 'description': 'Generate node table from edges', + 'schema_effects': { + 'adds_node_cols': ['node'], # Creates node column + 'adds_edge_cols': [], + 'requires_node_cols': [], + 'requires_edge_cols': [] + } + }, + + 'hop': { + 'allowed_params': { + 'nodes', 'hops', 'to_fixed_point', 'direction', + 'source_node_match', 'edge_match', 'destination_node_match', + 'source_node_query', 'edge_query', 'destination_node_query', + 'return_as_wave_front', 'target_wave_front', 'engine' + }, + 'required_params': set(), + 'param_validators': { + 'hops': is_int, + 'to_fixed_point': is_bool, + 'direction': lambda v: v in ['forward', 'reverse', 'undirected'], + 'source_node_match': is_dict, + 'edge_match': is_dict, + 'destination_node_match': is_dict, + 'source_node_query': is_string, + 'edge_query': is_string, + 'destination_node_query': is_string, + 'return_as_wave_front': is_bool, + 'engine': is_string + }, + 'description': 'Traverse graph by following edges', + 'schema_effects': { + 'adds_node_cols': [], + 'adds_edge_cols': [], + 'requires_node_cols': [], + 'requires_edge_cols': [] + } + }, + + # In/out degree methods + 'get_indegrees': { + 'allowed_params': {'col'}, + 'required_params': set(), + 'param_validators': { + 'col': is_string + }, + 'description': 'Calculate node in-degrees', + 'schema_effects': { + 'adds_node_cols': lambda p: [p.get('col', 'degree_in')], + 'adds_edge_cols': [], + 'requires_node_cols': [], + 'requires_edge_cols': [] + } + }, + + 'get_outdegrees': { + 'allowed_params': {'col'}, + 'required_params': set(), + 'param_validators': { + 'col': is_string + }, + 'description': 'Calculate node out-degrees', + 'schema_effects': { + 'adds_node_cols': lambda p: [p.get('col', 'degree_out')], + 'adds_edge_cols': [], + 'requires_node_cols': [], + 'requires_edge_cols': [] + } + }, + + # Graph algorithm operations + 'compute_cugraph': { + 'allowed_params': {'alg', 'out_col', 'params', 'kind', 'directed', 'G'}, + 'required_params': {'alg'}, + 'param_validators': { + 'alg': is_string, + 'out_col': is_string_or_none, + 'params': is_dict, + 'kind': is_string, + 'directed': is_bool, + 'G': lambda x: x is None # Allow None only + }, + 'description': 'Run cuGraph algorithms (pagerank, louvain, etc)', + 'schema_effects': { + 'adds_node_cols': lambda p: [p.get('out_col', p['alg'])], + 'adds_edge_cols': [], + 'requires_node_cols': [], + 'requires_edge_cols': [] + } + }, + + 'compute_igraph': { + 'allowed_params': {'alg', 'out_col', 'directed', 'use_vids', 'params'}, + 'required_params': {'alg'}, + 'param_validators': { + 'alg': is_string, + 'out_col': is_string_or_none, + 'directed': is_bool, + 'use_vids': is_bool, + 'params': is_dict + }, + 'description': 'Run igraph algorithms' + }, + + # Layout operations + 'layout_cugraph': { + 'allowed_params': {'layout', 'params', 'kind', 'directed', 'G', 'bind_position', 'x_out_col', 'y_out_col', 'play'}, + 'required_params': set(), + 'param_validators': { + 'layout': is_string, + 'params': is_dict, + 'kind': is_string, + 'directed': is_bool, + 'G': lambda x: x is None, + 'bind_position': is_bool, + 'x_out_col': is_string, + 'y_out_col': is_string, + 'play': is_int + }, + 'description': 'GPU-accelerated graph layouts' + }, + + 'layout_igraph': { + 'allowed_params': {'layout', 'directed', 'use_vids', 'bind_position', 'x_out_col', 'y_out_col', 'params', 'play'}, + 'required_params': {'layout'}, + 'param_validators': { + 'layout': is_string, + 'directed': is_bool, + 'use_vids': is_bool, + 'bind_position': is_bool, + 'x_out_col': is_string, + 'y_out_col': is_string, + 'params': is_dict, + 'play': is_int + }, + 'description': 'igraph-based layouts' + }, + + 'layout_graphviz': { + 'allowed_params': { + 'prog', 'args', 'directed', 'strict', 'graph_attr', + 'node_attr', 'edge_attr', 'x_out_col', 'y_out_col', 'bind_position' + }, + 'required_params': set(), + 'param_validators': { + 'prog': is_string, + 'args': is_string_or_none, + 'directed': is_bool, + 'strict': is_bool, + 'graph_attr': is_dict, + 'node_attr': is_dict, + 'edge_attr': is_dict, + 'x_out_col': is_string, + 'y_out_col': is_string, + 'bind_position': is_bool + }, + 'description': 'Graphviz layouts (dot, neato, etc)' + }, + + 'fa2_layout': { + 'allowed_params': {'fa2_params', 'circle_layout_params', 'partition_key', 'remove_self_edges', 'engine', 'featurize'}, + 'required_params': set(), + 'param_validators': { + 'fa2_params': is_dict, + 'circle_layout_params': is_dict, + 'partition_key': is_string_or_none, + 'remove_self_edges': is_bool, + 'engine': is_string, + 'featurize': is_dict + }, + 'description': 'ForceAtlas2 layout algorithm' + }, + + # Self-edge pruning + 'prune_self_edges': { + 'allowed_params': set(), + 'required_params': set(), + 'param_validators': {}, + 'description': 'Remove self-loops from graph' + }, + + # Graph transformations + 'collapse': { + 'allowed_params': {'node', 'attribute', 'column', 'self_edges', 'unwrap', 'verbose'}, + 'required_params': set(), + 'param_validators': { + 'node': is_string_or_none, + 'attribute': is_string_or_none, + 'column': is_string_or_none, + 'self_edges': is_bool, + 'unwrap': is_bool, + 'verbose': is_bool + }, + 'description': 'Collapse nodes by shared attribute values' + }, + + 'drop_nodes': { + 'allowed_params': {'nodes'}, + 'required_params': {'nodes'}, + 'param_validators': { + 'nodes': lambda v: isinstance(v, list) or is_dict(v) + }, + 'description': 'Remove specified nodes and their edges' + }, + + 'keep_nodes': { + 'allowed_params': {'nodes'}, + 'required_params': {'nodes'}, + 'param_validators': { + 'nodes': lambda v: isinstance(v, list) or is_dict(v) + }, + 'description': 'Keep only specified nodes and their edges' + }, + + # Topology analysis + 'get_topological_levels': { + 'allowed_params': {'level_col', 'allow_cycles', 'warn_cycles', 'remove_self_loops'}, + 'required_params': set(), + 'param_validators': { + 'level_col': is_string, + 'allow_cycles': is_bool, + 'warn_cycles': is_bool, + 'remove_self_loops': is_bool + }, + 'description': 'Compute topological levels for DAG analysis' + }, + + # Visual encoding methods + 'encode_point_color': { + 'allowed_params': {'column', 'palette', 'as_categorical', 'as_continuous', 'categorical_mapping', 'default_mapping'}, + 'required_params': {'column'}, + 'param_validators': { + 'column': is_string, + 'palette': lambda v: isinstance(v, list), + 'as_categorical': is_bool, + 'as_continuous': is_bool, + 'categorical_mapping': is_dict, + 'default_mapping': is_string_or_none + }, + 'description': 'Map node column values to colors' + }, + + 'encode_edge_color': { + 'allowed_params': {'column', 'palette', 'as_categorical', 'as_continuous', 'categorical_mapping', 'default_mapping'}, + 'required_params': {'column'}, + 'param_validators': { + 'column': is_string, + 'palette': lambda v: isinstance(v, list), + 'as_categorical': is_bool, + 'as_continuous': is_bool, + 'categorical_mapping': is_dict, + 'default_mapping': is_string_or_none + }, + 'description': 'Map edge column values to colors' + }, + + 'encode_point_size': { + 'allowed_params': {'column', 'categorical_mapping', 'default_mapping'}, + 'required_params': {'column'}, + 'param_validators': { + 'column': is_string, + 'categorical_mapping': is_dict, + 'default_mapping': lambda v: isinstance(v, (int, float)) + }, + 'description': 'Map node column values to sizes' + }, + + 'encode_point_icon': { + 'allowed_params': {'column', 'categorical_mapping', 'continuous_binning', 'default_mapping', 'as_text'}, + 'required_params': {'column'}, + 'param_validators': { + 'column': is_string, + 'categorical_mapping': is_dict, + 'continuous_binning': lambda v: isinstance(v, list), + 'default_mapping': is_string_or_none, + 'as_text': is_bool + }, + 'description': 'Map node column values to icons' + }, + + # Metadata methods + 'name': { + 'allowed_params': {'name'}, + 'required_params': {'name'}, + 'param_validators': { + 'name': is_string + }, + 'description': 'Set visualization name' + }, + + 'description': { + 'allowed_params': {'description'}, + 'required_params': {'description'}, + 'param_validators': { + 'description': is_string + }, + 'description': 'Set visualization description' + } +} + + +def validate_call_params(function: str, params: Dict[str, Any]) -> Dict[str, Any]: + """Validate parameters for a function call. + + Args: + function: Name of the function to call + params: Parameters to validate + + Returns: + Validated parameters (may be modified, e.g., defaults added) + + Raises: + GFQLTypeError: If validation fails + """ + # Check if function is in safelist + if function not in SAFELIST_V1: + raise GFQLTypeError( + ErrorCode.E303, + f"Function '{function}' is not in the safelist", + field="function", + value=function, + suggestion=f"Available functions: {', '.join(sorted(SAFELIST_V1.keys()))}" + ) + + config = SAFELIST_V1[function] + allowed_params = config['allowed_params'] + required_params = config['required_params'] + param_validators = config['param_validators'] + + # Check for required parameters + missing_required = required_params - set(params.keys()) + if missing_required: + raise GFQLTypeError( + ErrorCode.E105, + f"Missing required parameters for '{function}'", + field="params", + value=list(missing_required), + suggestion=f"Required parameters: {', '.join(sorted(missing_required))}" + ) + + # Check for unknown parameters + unknown_params = set(params.keys()) - allowed_params + if unknown_params: + raise GFQLTypeError( + ErrorCode.E303, + f"Unknown parameters for '{function}'", + field="params", + value=list(unknown_params), + suggestion=f"Allowed parameters: {', '.join(sorted(allowed_params))}" + ) + + # Validate parameter types + for param_name, param_value in params.items(): + if param_name in param_validators: + validator = param_validators[param_name] + if not validator(param_value): + raise GFQLTypeError( + ErrorCode.E201, + f"Invalid type for parameter '{param_name}' in '{function}'", + field=f"params.{param_name}", + value=f"{type(param_value).__name__}: {param_value}", + suggestion="Check the parameter type requirements" + ) + + return params diff --git a/graphistry/compute/chain_dag.py b/graphistry/compute/chain_dag.py index ef0911d4c2..26925747b1 100644 --- a/graphistry/compute/chain_dag.py +++ b/graphistry/compute/chain_dag.py @@ -3,7 +3,7 @@ from graphistry.Engine import Engine, EngineAbstract, resolve_engine from graphistry.Plottable import Plottable from graphistry.util import setup_logger -from .ast import ASTObject, ASTLet, ASTRef, ASTRemoteGraph, ASTNode, ASTEdge +from .ast import ASTObject, ASTLet, ASTRef, ASTRemoteGraph, ASTNode, ASTEdge, ASTCall from .execution_context import ExecutionContext logger = setup_logger(__name__) @@ -291,6 +291,10 @@ def execute_node(name: str, ast_obj: ASTObject, g: Plottable, output_type="all", # Get full graph (nodes and edges) engine=cast(Literal["pandas", "cudf"], engine.value) ) + elif isinstance(ast_obj, ASTCall): + # Execute method call with validation + from .call_executor import execute_call + result = execute_call(g, ast_obj.function, ast_obj.params, engine) else: # Other AST object types not yet implemented raise NotImplementedError(f"Execution of {type(ast_obj).__name__} not yet implemented") @@ -315,7 +319,7 @@ def chain_dag_impl(g: Plottable, dag: ASTLet, :param output: Name of binding to return (default: last executed) :returns: Result from specified or last executed node :rtype: Plottable - :raises TypeError: If dag is not an ASTQueryDAG + :raises TypeError: If dag is not an ASTLet :raises RuntimeError: If node execution fails :raises ValueError: If output binding not found """ diff --git a/graphistry/compute/validate/validate_schema.py b/graphistry/compute/validate/validate_schema.py index 6d9a174a17..228bc4c937 100644 --- a/graphistry/compute/validate/validate_schema.py +++ b/graphistry/compute/validate/validate_schema.py @@ -3,11 +3,10 @@ from typing import List, Optional, Union, TYPE_CHECKING, cast import pandas as pd from graphistry.Plottable import Plottable -from graphistry.compute.ast import ASTObject, ASTNode, ASTEdge, ASTLet, ASTRef, ASTRemoteGraph +from graphistry.compute.ast import ASTObject, ASTNode, ASTEdge, ASTLet, ASTRef, ASTRemoteGraph, ASTCall if TYPE_CHECKING: from graphistry.compute.chain import Chain - from graphistry.compute.exceptions import ErrorCode, GFQLSchemaError from graphistry.compute.predicates.ASTPredicate import ASTPredicate from graphistry.compute.predicates.numeric import NumericASTPredicate, Between @@ -64,6 +63,8 @@ def validate_chain_schema( op_errors = _validate_ref_op(op, g, collect_all) elif isinstance(op, ASTRemoteGraph): op_errors = _validate_remotegraph_op(op, collect_all) + elif isinstance(op, ASTCall): + op_errors = _validate_call_op(op, node_columns, edge_columns, collect_all) # Add operation index to all errors for e in op_errors: @@ -126,8 +127,6 @@ def _validate_querydag_op(op: ASTLet, g: Plottable, collect_all: bool) -> List[G if binding_errors: for error in binding_errors: error.context['dag_binding'] = binding_name - - if binding_errors: if collect_all: errors.extend(binding_errors) else: @@ -156,8 +155,6 @@ def _validate_ref_op(op: ASTRef, g: Plottable, collect_all: bool) -> List[GFQLSc if chain_errors: for error in chain_errors: error.context['ref'] = op.ref - - if chain_errors: if collect_all: errors.extend(chain_errors) else: @@ -305,6 +302,92 @@ def _validate_filter_dict( return errors +def _validate_call_op( + op: ASTCall, + node_columns: set, + edge_columns: set, + collect_all: bool = False +) -> List[GFQLSchemaError]: + """Validate Call operation schema requirements. + + Checks that all columns required by the called method exist in the graph. + Uses the schema_effects metadata from the safelist to determine requirements. + + Args: + op: ASTCall operation to validate + node_columns: Set of available node column names + edge_columns: Set of available edge column names + collect_all: If True, collect all errors. If False, raise on first error. + + Returns: + List of schema errors found (empty if valid) + + Raises: + GFQLSchemaError: If collect_all=False and validation fails + """ + errors: List[GFQLSchemaError] = [] + + # Import safelist to get schema effects + from graphistry.compute.call_safelist import SAFELIST_V1 + + # Check if method is in safelist + if op.function not in SAFELIST_V1: + # This should have been caught by parameter validation already + return errors + + method_info = SAFELIST_V1[op.function] + + # Check if method has schema effects defined + if 'schema_effects' not in method_info: + # Method doesn't define schema effects, so we can't validate + return errors + + schema_effects = method_info['schema_effects'] + + # Get required columns based on parameters + if 'requires_node_cols' in schema_effects: + if callable(schema_effects['requires_node_cols']): + required_node_cols = schema_effects['requires_node_cols'](op.params) + else: + required_node_cols = schema_effects['requires_node_cols'] + + for col in required_node_cols: + if col not in node_columns: + error = GFQLSchemaError( + ErrorCode.E301, + f'Call operation "{op.function}" requires node column "{col}" which does not exist', + field=f'{op.function}.{col}', + value=col, + suggestion=f'Available node columns: {", ".join(sorted(node_columns)[:10])}{"..." if len(node_columns) > 10 else ""}' + ) + if collect_all: + errors.append(error) + else: + raise error + + if 'requires_edge_cols' in schema_effects: + if callable(schema_effects['requires_edge_cols']): + required_edge_cols = schema_effects['requires_edge_cols'](op.params) + else: + required_edge_cols = schema_effects['requires_edge_cols'] + + for col in required_edge_cols: + if col not in edge_columns: + error = GFQLSchemaError( + ErrorCode.E301, + f'Call operation "{op.function}" requires edge column "{col}" which does not exist', + field=f'{op.function}.{col}', + value=col, + suggestion=f'Available edge columns: {", ".join(sorted(edge_columns)[:10])}{"..." if len(edge_columns) > 10 else ""}' + ) + if collect_all: + errors.append(error) + else: + raise error + + return errors + + # Add to Chain class def validate_schema(self: 'Chain', g: Plottable, collect_all: bool = False) -> Optional[List[GFQLSchemaError]]: """Validate this chain against a graph's schema without executing. diff --git a/graphistry/tests/compute/test_call_operations.py b/graphistry/tests/compute/test_call_operations.py new file mode 100644 index 0000000000..877f5b4edd --- /dev/null +++ b/graphistry/tests/compute/test_call_operations.py @@ -0,0 +1,394 @@ +"""Tests for GFQL Call operations.""" + +import pytest +import pandas as pd +from unittest.mock import Mock, patch, MagicMock + +from graphistry.tests.test_compute import CGFull +from graphistry.Engine import Engine, EngineAbstract +from graphistry.compute.ast import ASTCall, ASTLet, n +from graphistry.compute.chain_dag import chain_dag_impl +from graphistry.compute.call_safelist import validate_call_params +from graphistry.compute.call_executor import execute_call +from graphistry.compute.exceptions import ErrorCode, GFQLTypeError, GFQLSyntaxError + + +class TestCallSafelist: + """Test method safelist validation.""" + + def test_allowed_method(self): + """Test validation of allowed methods.""" + params = validate_call_params('get_degrees', { + 'col': 'degree' + }) + assert params == {'col': 'degree'} + + def test_unknown_method(self): + """Test rejection of unknown methods.""" + with pytest.raises(GFQLTypeError) as exc_info: + validate_call_params('unknown_method', {}) + assert exc_info.value.code == ErrorCode.E303 + assert 'not in the safelist' in str(exc_info.value) + + def test_required_params(self): + """Test validation of required parameters.""" + with pytest.raises(GFQLTypeError) as exc_info: + validate_call_params('filter_nodes_by_dict', {}) + assert exc_info.value.code == ErrorCode.E105 + assert 'Missing required parameters' in str(exc_info.value) + + def test_unknown_params(self): + """Test rejection of unknown parameters.""" + with pytest.raises(GFQLTypeError) as exc_info: + validate_call_params('get_degrees', { + 'col': 'degree', + 'unknown_param': 'value' + }) + assert exc_info.value.code == ErrorCode.E303 + assert 'Unknown parameters' in str(exc_info.value) + + def test_param_type_validation(self): + """Test parameter type validation.""" + # Valid types + params = validate_call_params('hop', { + 'hops': 2, + 'direction': 'forward', + 'to_fixed_point': True + }) + assert params['hops'] == 2 + + # Invalid type + with pytest.raises(GFQLTypeError) as exc_info: + validate_call_params('hop', { + 'hops': 'two' # Should be int + }) + assert exc_info.value.code == ErrorCode.E201 + assert 'Invalid type' in str(exc_info.value) + + def test_enum_validation(self): + """Test enum parameter validation.""" + # Valid enum value + params = validate_call_params('hop', { + 'direction': 'forward' + }) + assert params['direction'] == 'forward' + + # Invalid enum value + with pytest.raises(GFQLTypeError) as exc_info: + validate_call_params('hop', { + 'direction': 'sideways' + }) + assert exc_info.value.code == ErrorCode.E201 + + +class TestASTCall: + """Test ASTCall node validation and serialization.""" + + def test_basic_creation(self): + """Test creating a basic ASTCall.""" + call = ASTCall('get_degrees', {'col': 'degree'}) + assert call.function == 'get_degrees' + assert call.params == {'col': 'degree'} + + def test_empty_params(self): + """Test ASTCall with no parameters.""" + call = ASTCall('prune_self_edges') + assert call.function == 'prune_self_edges' + assert call.params == {} + + def test_validation(self): + """Test ASTCall field validation.""" + # Valid call + call = ASTCall('get_degrees', {'col': 'degree'}) + call.validate() # Should not raise + + # Invalid function type + with pytest.raises(GFQLTypeError) as exc_info: + call = ASTCall(123, {}) + call.validate() + assert exc_info.value.code == ErrorCode.E201 + assert 'function must be a string' in str(exc_info.value) + + # Invalid params type + with pytest.raises(GFQLTypeError) as exc_info: + call = ASTCall('get_degrees', 'not_a_dict') + call.validate() + assert exc_info.value.code == ErrorCode.E201 + assert 'params must be a dict' in str(exc_info.value) + + def test_to_json(self): + """Test ASTCall JSON serialization.""" + call = ASTCall('get_degrees', {'col': 'degree', 'engine': 'pandas'}) + json_data = call.to_json() + + assert json_data == { + 'type': 'Call', + 'function': 'get_degrees', + 'params': {'col': 'degree', 'engine': 'pandas'} + } + + def test_from_json(self): + """Test ASTCall JSON deserialization.""" + json_data = { + 'type': 'Call', + 'function': 'filter_nodes_by_dict', + 'params': {'filter_dict': {'type': 'user'}} + } + + call = ASTCall.from_json(json_data) + assert isinstance(call, ASTCall) + assert call.function == 'filter_nodes_by_dict' + assert call.params == {'filter_dict': {'type': 'user'}} + + def test_from_json_invalid(self): + """Test ASTCall from_json with invalid data.""" + # Missing function + with pytest.raises(AssertionError) as exc_info: + ASTCall.from_json({'type': 'Call'}) + assert 'Call missing function' in str(exc_info.value) + + # Wrong type - this would be caught earlier in the AST dispatch + # so we don't test it here + + +class TestCallExecution: + """Test call execution functionality.""" + + @pytest.fixture + def sample_graph(self): + """Create a sample graph for testing.""" + edges_df = pd.DataFrame({ + 'source': [0, 1, 2, 2], + 'target': [1, 2, 0, 3] + }) + nodes_df = pd.DataFrame({ + 'node': [0, 1, 2, 3], + 'type': ['user', 'user', 'admin', 'user'] + }) + + return CGFull()\ + .edges(edges_df)\ + .nodes(nodes_df)\ + .bind(source='source', destination='target', node='node') + + def test_execute_get_degrees(self, sample_graph): + """Test executing get_degrees method.""" + result = execute_call( + sample_graph, + 'get_degrees', + {'col': 'degree'}, + Engine.PANDAS + ) + + # Result should be a Plottable + assert hasattr(result, '_nodes') + assert 'degree' in result._nodes.columns + assert len(result._nodes) == 4 + + def test_execute_filter_nodes(self, sample_graph): + """Test executing filter_nodes_by_dict method.""" + result = execute_call( + sample_graph, + 'filter_nodes_by_dict', + {'filter_dict': {'type': 'user'}}, + Engine.PANDAS + ) + + # Should filter to only user nodes + assert len(result._nodes) == 3 + assert all(result._nodes['type'] == 'user') + + def test_execute_materialize_nodes(self): + """Test executing materialize_nodes method.""" + # Start with edges only + edges_df = pd.DataFrame({ + 'source': [0, 1, 2], + 'target': [1, 2, 0] + }) + g = CGFull().edges(edges_df).bind(source='source', destination='target') + + # No nodes initially + assert g._nodes is None + + # Execute materialize_nodes + result = execute_call( + g, + 'materialize_nodes', + {}, + Engine.PANDAS + ) + + # Should have nodes now + assert result._nodes is not None + assert len(result._nodes) == 3 + + def test_execute_with_validation_error(self, sample_graph): + """Test that validation errors are properly raised.""" + with pytest.raises(GFQLTypeError) as exc_info: + execute_call( + sample_graph, + 'hop', + {'hops': 'invalid'}, # Should be int + Engine.PANDAS + ) + assert exc_info.value.code == ErrorCode.E201 + + def test_execute_unknown_method(self, sample_graph): + """Test execution of unknown method.""" + with pytest.raises(GFQLTypeError) as exc_info: + execute_call( + sample_graph, + 'unknown_method', + {}, + Engine.PANDAS + ) + assert exc_info.value.code == ErrorCode.E303 + + +class TestCallInDAG: + """Test ASTCall execution within DAGs.""" + + @pytest.fixture + def sample_graph(self): + """Create a sample graph for testing.""" + edges_df = pd.DataFrame({ + 'source': [0, 1, 2, 2, 3], + 'target': [1, 2, 0, 3, 0], + 'weight': [1.0, 2.0, 1.5, 3.0, 0.5] + }) + nodes_df = pd.DataFrame({ + 'node': [0, 1, 2, 3], + 'type': ['user', 'user', 'admin', 'user'] + }) + + return CGFull()\ + .edges(edges_df)\ + .nodes(nodes_df)\ + .bind(source='source', destination='target', node='node') + + def test_call_in_dag(self, sample_graph): + """Test executing ASTCall within a DAG.""" + dag = ASTLet({ + 'filtered': n({'type': 'user'}), + 'with_degrees': ASTCall('get_degrees', {'col': 'degree'}) + }) + + result = chain_dag_impl(sample_graph, dag, EngineAbstract.PANDAS) + + # Should have degree column + assert 'degree' in result._nodes.columns + # Should still have all nodes (get_degrees doesn't filter) + assert len(result._nodes) == 4 + + def test_call_referencing_binding(self, sample_graph): + """Test ASTCall that operates on whole graph (not in chain).""" + from graphistry.compute.ast import ASTRef + + # Call operations work on the whole graph, not as part of chains + dag = ASTLet({ + 'users': n({'type': 'user'}), + 'with_degrees': ASTCall('get_degrees', {'col': 'degree'}) + }) + + result = chain_dag_impl(sample_graph, dag, EngineAbstract.PANDAS) + + # Should have degree column on all nodes + assert len(result._nodes) == 4 # All nodes + assert 'degree' in result._nodes.columns + + def test_multiple_calls(self, sample_graph): + """Test multiple call operations in sequence.""" + # First add degrees + dag1 = ASTLet({ + 'with_degrees': ASTCall('get_degrees', {'col': 'deg'}) + }) + result1 = chain_dag_impl(sample_graph, dag1, EngineAbstract.PANDAS) + assert 'deg' in result1._nodes.columns + + # Then filter - use the graph that has degrees + dag2 = ASTLet({ + 'filtered': ASTCall('filter_nodes_by_dict', {'filter_dict': {'deg': 2}}) + }) + result2 = chain_dag_impl(result1, dag2, EngineAbstract.PANDAS) + + # Should have nodes with degree 2 + assert len(result2._nodes) > 0 + assert all(result2._nodes['deg'] == 2) + + @patch('graphistry.compute.call_executor.getattr') + def test_call_execution_error(self, mock_getattr, sample_graph): + """Test handling of execution errors in calls.""" + # Make the method raise an error + mock_method = Mock(side_effect=RuntimeError("Method failed")) + mock_getattr.return_value = mock_method + + dag = ASTLet({ + 'failing': ASTCall('get_degrees', {}) + }) + + with pytest.raises(RuntimeError) as exc_info: + chain_dag_impl(sample_graph, dag, EngineAbstract.PANDAS) + assert "Failed to execute node 'failing'" in str(exc_info.value) + + +class TestGraphAlgorithmCalls: + """Test calls to graph algorithm methods.""" + + @pytest.fixture + def sample_graph(self): + """Create a sample graph for testing.""" + edges_df = pd.DataFrame({ + 'source': [0, 1, 2, 2], + 'target': [1, 2, 0, 3] + }) + + return CGFull()\ + .edges(edges_df)\ + .bind(source='source', destination='target') + + def test_compute_cugraph_params(self): + """Test compute_cugraph parameter validation.""" + # Valid params + params = validate_call_params('compute_cugraph', { + 'alg': 'pagerank', + 'out_col': 'pr_score', + 'directed': True + }) + assert params['alg'] == 'pagerank' + + # Missing required alg + with pytest.raises(GFQLTypeError) as exc_info: + validate_call_params('compute_cugraph', { + 'out_col': 'pr_score' + }) + assert exc_info.value.code == ErrorCode.E105 + + def test_compute_igraph_params(self): + """Test compute_igraph parameter validation.""" + params = validate_call_params('compute_igraph', { + 'alg': 'community_louvain', + 'directed': False + }) + assert params['alg'] == 'community_louvain' + + def test_layout_methods(self): + """Test layout method parameter validation.""" + # layout_cugraph + params = validate_call_params('layout_cugraph', { + 'layout': 'force_atlas2', + 'params': {'iterations': 100} + }) + assert params['layout'] == 'force_atlas2' + + # layout_igraph + params = validate_call_params('layout_igraph', { + 'layout': 'fruchterman_reingold', + 'directed': True + }) + assert params['layout'] == 'fruchterman_reingold' + + # fa2_layout + params = validate_call_params('fa2_layout', { + 'fa2_params': {'iterations': 500} + }) + assert params['fa2_params']['iterations'] == 500 diff --git a/graphistry/tests/compute/test_call_operations_gpu.py b/graphistry/tests/compute/test_call_operations_gpu.py new file mode 100644 index 0000000000..915571230f --- /dev/null +++ b/graphistry/tests/compute/test_call_operations_gpu.py @@ -0,0 +1,243 @@ +"""GPU tests for GFQL Call operations.""" + +import os +import pytest +import pandas as pd + +from graphistry.tests.test_compute import CGFull +from graphistry.Engine import Engine +from graphistry.compute.ast import ASTCall, ASTLet, n +from graphistry.compute.chain_dag import chain_dag_impl +from graphistry.compute.call_executor import execute_call +from graphistry.compute.validate.validate_schema import validate_chain_schema +from graphistry.compute.exceptions import ErrorCode, GFQLTypeError + + +# Skip all tests if TEST_CUDF not set +skip_gpu = pytest.mark.skipif( + not ("TEST_CUDF" in os.environ and os.environ["TEST_CUDF"] == "1"), + reason="cudf tests need TEST_CUDF=1" +) + + +class TestCallOperationsGPU: + """Test Call operations with GPU/cudf.""" + + @skip_gpu + def test_call_with_cudf_dataframes(self): + """Test that Call operations work with cudf DataFrames.""" + import cudf + + # Create cudf dataframes + edges_df = pd.DataFrame({ + 'source': [0, 1, 2, 2], + 'target': [1, 2, 0, 3], + 'weight': [1.0, 2.0, 3.0, 4.0] + }) + nodes_df = pd.DataFrame({ + 'node': [0, 1, 2, 3], + 'type': ['user', 'user', 'admin', 'user'] + }) + + edges_gdf = cudf.from_pandas(edges_df) + nodes_gdf = cudf.from_pandas(nodes_df) + + # Create graph with cudf data + g = CGFull()\ + .edges(edges_gdf)\ + .nodes(nodes_gdf)\ + .bind(source='source', destination='target', node='node') + + # Execute Call operation + result = execute_call(g, 'get_degrees', {'col': 'degree'}, Engine.CUDF) + + # Result should still have cudf nodes + assert hasattr(result._nodes, '__cuda_array_interface__') + assert 'degree' in result._nodes.columns + + @skip_gpu + def test_filter_with_cudf(self): + """Test filtering operations with cudf.""" + import cudf + + nodes_df = pd.DataFrame({ + 'node': [0, 1, 2, 3], + 'type': ['user', 'user', 'admin', 'user'], + 'score': [0.5, 0.8, 0.9, 0.3] + }) + nodes_gdf = cudf.from_pandas(nodes_df) + + g = CGFull().nodes(nodes_gdf).bind(node='node') + + # Filter nodes + result = execute_call( + g, + 'filter_nodes_by_dict', + {'filter_dict': {'type': 'user'}}, + Engine.CUDF + ) + + # Should still be cudf and filtered + assert hasattr(result._nodes, '__cuda_array_interface__') + assert len(result._nodes) == 3 + assert all(result._nodes['type'] == 'user') + + @skip_gpu + def test_compute_cugraph_call(self): + """Test compute_cugraph through Call operation.""" + import cudf + + # Create a simple graph + edges_df = pd.DataFrame({ + 'source': [0, 1, 2, 2, 3], + 'target': [1, 2, 0, 3, 0] + }) + edges_gdf = cudf.from_pandas(edges_df) + + g = CGFull().edges(edges_gdf).bind(source='source', destination='target') + + # Skip if cugraph not available + try: + import cugraph + except ImportError: + pytest.skip("cugraph not installed") + + # Call compute_cugraph for pagerank + result = execute_call( + g, + 'compute_cugraph', + {'alg': 'pagerank', 'out_col': 'pr_score'}, + Engine.CUDF + ) + + # Should have pagerank scores + assert 'pr_score' in result._nodes.columns + assert hasattr(result._nodes, '__cuda_array_interface__') + + @skip_gpu + def test_layout_cugraph_call(self): + """Test layout_cugraph through Call operation.""" + import cudf + + edges_df = pd.DataFrame({ + 'source': [0, 1, 2], + 'target': [1, 2, 0] + }) + edges_gdf = cudf.from_pandas(edges_df) + + g = CGFull().edges(edges_gdf).bind(source='source', destination='target') + + # Skip if cugraph not available + try: + import cugraph + except ImportError: + pytest.skip("cugraph not installed") + + # Call layout_cugraph + result = execute_call( + g, + 'layout_cugraph', + {'layout': 'force_atlas2'}, + Engine.CUDF + ) + + # Should have x,y coordinates + assert 'x' in result._nodes.columns + assert 'y' in result._nodes.columns + assert hasattr(result._nodes, '__cuda_array_interface__') + + @skip_gpu + def test_chain_dag_with_gpu_calls(self): + """Test DAG execution with Call operations on GPU.""" + import cudf + + edges_df = pd.DataFrame({ + 'source': [0, 1, 2, 2, 3], + 'target': [1, 2, 0, 3, 0], + 'weight': [1.0, 2.0, 1.5, 3.0, 0.5] + }) + nodes_df = pd.DataFrame({ + 'node': [0, 1, 2, 3], + 'type': ['user', 'user', 'admin', 'user'] + }) + + edges_gdf = cudf.from_pandas(edges_df) + nodes_gdf = cudf.from_pandas(nodes_df) + + g = CGFull()\ + .edges(edges_gdf)\ + .nodes(nodes_gdf)\ + .bind(source='source', destination='target', node='node') + + # Create DAG with Call operations + dag = ASTLet({ + 'filtered': n({'type': 'user'}), + 'with_degrees': ASTCall('get_degrees', {'col': 'degree'}) + }) + + result = chain_dag_impl(g, dag, Engine.CUDF) + + # Should have GPU data with degrees + assert hasattr(result._nodes, '__cuda_array_interface__') + assert 'degree' in result._nodes.columns + + @skip_gpu + def test_schema_validation_with_cudf(self): + """Test schema validation works with cudf DataFrames.""" + import cudf + + nodes_df = pd.DataFrame({ + 'node': [0, 1, 2], + 'type': ['A', 'B', 'C'] + }) + nodes_gdf = cudf.from_pandas(nodes_df) + + g = CGFull().nodes(nodes_gdf).bind(node='node') + + # Valid call - column exists + call = ASTCall('filter_nodes_by_dict', {'filter_dict': {'type': 'A'}}) + errors = validate_chain_schema(g, [call], collect_all=True) + assert len(errors) == 0 + + # Invalid call - column doesn't exist + call = ASTCall('filter_nodes_by_dict', {'filter_dict': {'missing': 'X'}}) + errors = validate_chain_schema(g, [call], collect_all=True) + assert len(errors) > 0 + assert any('missing' in str(e) for e in errors) + + @skip_gpu + def test_encode_with_gpu(self): + """Test visual encoding methods with GPU data.""" + import cudf + + nodes_df = pd.DataFrame({ + 'node': [0, 1, 2, 3], + 'category': ['A', 'B', 'A', 'C'], + 'score': [0.1, 0.5, 0.8, 0.3] + }) + nodes_gdf = cudf.from_pandas(nodes_df) + + g = CGFull().nodes(nodes_gdf).bind(node='node') + + # Test encode_point_color + result = execute_call( + g, + 'encode_point_color', + {'column': 'category'}, + Engine.CUDF + ) + + # Should still have GPU data + assert hasattr(result._nodes, '__cuda_array_interface__') + + # Test encode_point_size + result2 = execute_call( + result, + 'encode_point_size', + {'column': 'score'}, + Engine.CUDF + ) + + assert hasattr(result2._nodes, '__cuda_array_interface__') + # Should have size encoding set + assert result2._point_size == 'score' diff --git a/graphistry/tests/compute/test_call_schema_validation.py b/graphistry/tests/compute/test_call_schema_validation.py new file mode 100644 index 0000000000..3dd2d6f372 --- /dev/null +++ b/graphistry/tests/compute/test_call_schema_validation.py @@ -0,0 +1,126 @@ +"""Test schema validation for Call operations.""" + +import pytest +import pandas as pd +from graphistry.tests.test_compute import CGFull +from graphistry.compute.ast import ASTCall, n +from graphistry.compute.chain import Chain +from graphistry.compute.validate.validate_schema import validate_chain_schema +from graphistry.compute.exceptions import ErrorCode, GFQLSchemaError + + +class TestCallSchemaValidation: + """Test schema validation for Call operations.""" + + @pytest.fixture + def sample_graph(self): + """Create a sample graph for testing.""" + edges_df = pd.DataFrame({ + 'source': [0, 1, 2], + 'target': [1, 2, 0], + 'weight': [1.0, 2.0, 3.0] + }) + nodes_df = pd.DataFrame({ + 'node': [0, 1, 2], + 'type': ['user', 'user', 'admin'], + 'score': [0.5, 0.8, 0.9] + }) + + return CGFull()\ + .edges(edges_df)\ + .nodes(nodes_df)\ + .bind(source='source', destination='target', node='node') + + def test_filter_nodes_requires_columns(self, sample_graph): + """Test that filter_nodes_by_dict validates required columns.""" + # Valid: filtering by existing column + call = ASTCall('filter_nodes_by_dict', {'filter_dict': {'type': 'user'}}) + errors = validate_chain_schema(sample_graph, [call], collect_all=True) + assert len(errors) == 0 + + # Invalid: filtering by non-existent column + call = ASTCall('filter_nodes_by_dict', {'filter_dict': {'missing_col': 'value'}}) + with pytest.raises(GFQLSchemaError) as exc_info: + validate_chain_schema(sample_graph, [call], collect_all=False) + assert exc_info.value.code == ErrorCode.E301 + assert 'missing_col' in str(exc_info.value) + assert 'does not exist' in str(exc_info.value) + + def test_filter_edges_requires_columns(self, sample_graph): + """Test that filter_edges_by_dict validates required columns.""" + # Valid: filtering by existing edge column + call = ASTCall('filter_edges_by_dict', {'filter_dict': {'weight': 2.0}}) + errors = validate_chain_schema(sample_graph, [call], collect_all=True) + assert len(errors) == 0 + + # Invalid: filtering by non-existent edge column + call = ASTCall('filter_edges_by_dict', {'filter_dict': {'edge_type': 'friend'}}) + with pytest.raises(GFQLSchemaError) as exc_info: + validate_chain_schema(sample_graph, [call], collect_all=False) + assert exc_info.value.code == ErrorCode.E301 + assert 'edge_type' in str(exc_info.value) + + def test_encode_requires_columns(self, sample_graph): + """Test that encode methods validate required columns.""" + # Valid: encoding existing column + call = ASTCall('encode_point_color', {'column': 'type'}) + errors = validate_chain_schema(sample_graph, [call], collect_all=True) + assert len(errors) == 0 + + # Invalid: encoding non-existent column + call = ASTCall('encode_point_color', {'column': 'category'}) + errors = validate_chain_schema(sample_graph, [call], collect_all=True) + # Note: encode methods don't require columns to exist (they create bindings) + # so this should not produce errors + assert len(errors) == 0 + + def test_chain_with_multiple_calls(self, sample_graph): + """Test validation of chains with multiple Call operations.""" + chain = Chain([ + ASTCall('filter_nodes_by_dict', {'filter_dict': {'type': 'user'}}), + ASTCall('get_degrees', {'col': 'degree'}), + ASTCall('filter_nodes_by_dict', {'filter_dict': {'degree': 2}}) + ]) + + # The second filter expects 'degree' column which doesn't exist yet + # But schema validation is static and doesn't track added columns + errors = validate_chain_schema(sample_graph, chain, collect_all=True) + # Should have error for missing 'degree' column + assert any('degree' in str(e) for e in errors) + + def test_method_without_schema_effects(self, sample_graph): + """Test that methods without schema effects don't cause errors.""" + # materialize_nodes doesn't require any columns + call = ASTCall('materialize_nodes', {}) + errors = validate_chain_schema(sample_graph, [call], collect_all=True) + assert len(errors) == 0 + + def test_collect_all_mode(self, sample_graph): + """Test collect_all mode returns all errors.""" + chain = Chain([ + ASTCall('filter_nodes_by_dict', {'filter_dict': {'missing1': 'a', 'missing2': 'b'}}), + ASTCall('filter_edges_by_dict', {'filter_dict': {'missing3': 'c'}}) + ]) + + errors = validate_chain_schema(sample_graph, chain, collect_all=True) + # Should collect all 3 missing column errors + assert len(errors) >= 3 + missing_cols = {'missing1', 'missing2', 'missing3'} + error_cols = set() + for e in errors: + for col in missing_cols: + if col in str(e): + error_cols.add(col) + assert error_cols == missing_cols + + def test_operation_index_in_errors(self, sample_graph): + """Test that errors include operation index.""" + chain = Chain([ + n({'type': 'user'}), # op 0 + ASTCall('filter_nodes_by_dict', {'filter_dict': {'bad_col': 1}}) # op 1 + ]) + + errors = validate_chain_schema(sample_graph, chain, collect_all=True) + call_errors = [e for e in errors if 'bad_col' in str(e)] + assert len(call_errors) > 0 + assert call_errors[0].context['operation_index'] == 1