From 3f3319d93e8dee739aa9db159cece49335aaa923 Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Fri, 15 May 2026 14:17:00 -0700
Subject: [PATCH] feat(gfql): add public declarative schema model

---
 CHANGELOG.md                                  |   1 +
 docs/source/api/gfql/index.rst                |   1 +
 docs/source/api/gfql/schema.rst               |   7 +
 docs/source/gfql/index.rst                    |   1 +
 docs/source/gfql/schema.rst                   | 109 ++++++
 graphistry/Plottable.py                       |   1 +
 graphistry/PlotterBase.py                     |   6 +
 graphistry/__init__.py                        |   7 +
 .../compute/gfql/frontends/cypher/binder.py   | 245 ++++++++++++-
 graphistry/compute/gfql_validate.py           |  53 ++-
 graphistry/pygraphistry.py                    |   8 +-
 graphistry/schema.py                          | 333 ++++++++++++++++++
 .../tests/compute/gfql/cypher/test_binder.py  |  17 +-
 .../tests/compute/gfql/test_public_schema.py  | 220 ++++++++++++
 14 files changed, 983 insertions(+), 26 deletions(-)
 create mode 100644 docs/source/api/gfql/schema.rst
 create mode 100644 docs/source/gfql/schema.rst
 create mode 100644 graphistry/schema.py
 create mode 100644 graphistry/tests/compute/gfql/test_public_schema.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5139638e33..27409e3f26 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **GFQL policy / Cypher compiler hooks (#1454)**: Added experimental exact-key `precompile` and `postcompile` policy hooks for local Cypher string-query compilation. `postcompile` reports success or failure using the existing policy `success`, `error`, and `error_type` fields plus a stable `CompileSummary` with scalar compiler metadata.
 
 ### Changed
+- **GFQL public schema declarations (#1337)**: Added stable `graphistry.schema` exports for `NodeType`, `EdgeType`, `GraphSchema`, and `EdgeTopology`, plus top-level `graphistry` re-exports. `NodeType` and `EdgeType` now accept Arrow-first `pyarrow.Schema` declarations, preserve dtype/nullability through GFQL `RowSchema`, and export back to Arrow with label/type columns via `to_arrow()`. `graphistry.bind(..., schema=schema)` / `g.bind(schema=schema)` now attach public schema declarations to plotters, and Cypher preflight validation consumes the adapted internal `GraphSchemaCatalog` for declared labels, properties, relationship types, and source/destination topology checks. `GraphSchema(strict=False)` now makes schema-bound `g.gfql_validate(...)` permissive by default while explicit call-level `strict=True` still forces strict validation.
 - **GFQL / Cypher pattern predicate existence semantics (#1449)**: Direct-Cypher `WHERE (pattern)` predicates now lower through correlated semi-apply markers instead of rewriting single positive predicates into appended `MATCH` clauses, preventing existence checks from multiplying result rows. Added pandas/cuDF coverage for the residual `expr-pattern1-10`, `expr-pattern1-13`, and `expr-pattern1-18` undirected pattern-predicate wrong-row cases.
 - **GFQL / Cypher reentry failfast scaffolding cleanup (#1421)**: Removed the obsolete `graphistry.compute.gfql.cypher.reentry.runtime` compatibility re-export shim after compile-time reentry ownership moved to `reentry.compiletime`, moved tests off the old private `gfql_unified._compiled_query_reentry_state` access path, and lifted the stale closed-#1256 aggregate failfast so chained reentry secondary-property carries now flow through downstream aggregating `WITH` stages with positive row assertions.
 - **GFQL / Cypher pre-strict binder compatibility guard deletion (#1420)**: Retired the legacy loose `FrontendBinder.bind(strict_name_resolution=False)` graph traversal path and unresolved-name fallbacks now that #1357 made strict binder semantics canonical. Cypher compile prepass and graph-constructor binding now pass `strict_name_resolution=True` explicitly, and binder tests now pin that the legacy false flag no longer admits unresolved `collect(...)`, single-alias list literal, or missing-schema inputs while preserving strict source-order traversal through `WITH → UNWIND → MATCH`.
diff --git a/docs/source/api/gfql/index.rst b/docs/source/api/gfql/index.rst
index 1eb11b75fd..9af8e4a05c 100644
--- a/docs/source/api/gfql/index.rst
+++ b/docs/source/api/gfql/index.rst
@@ -13,3 +13,4 @@ GFQL API Reference
    hop
    node
    predicates
+   schema
diff --git a/docs/source/api/gfql/schema.rst b/docs/source/api/gfql/schema.rst
new file mode 100644
index 0000000000..dc2c3ca78a
--- /dev/null
+++ b/docs/source/api/gfql/schema.rst
@@ -0,0 +1,7 @@
+GFQL Schema
+===========
+
+.. automodule:: graphistry.schema
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/docs/source/gfql/index.rst b/docs/source/gfql/index.rst
index ad0fa359af..8362feb6f7 100644
--- a/docs/source/gfql/index.rst
+++ b/docs/source/gfql/index.rst
@@ -63,6 +63,7 @@ See also:
    builtin_calls
    policy
    strict_mode
+   schema
    wire_protocol_examples
 
 .. toctree::
diff --git a/docs/source/gfql/schema.rst b/docs/source/gfql/schema.rst
new file mode 100644
index 0000000000..4578c64249
--- /dev/null
+++ b/docs/source/gfql/schema.rst
@@ -0,0 +1,109 @@
+Declarative Graph Schemas
+=========================
+
+GFQL accepts public schema declarations through the stable
+``graphistry.schema`` import path. Use this when application code owns a graph
+contract and wants Cypher preflight checks to fail before query execution.
+
+.. code-block:: python
+
+   import graphistry
+   import pyarrow as pa
+   from graphistry.schema import EdgeType, GraphSchema, NodeType
+
+   Person = NodeType(
+       "Person",
+       pa.schema([
+           pa.field("id", pa.int64(), nullable=False),
+           pa.field("name", pa.large_string()),
+       ]),
+   )
+   Company = NodeType(
+       "Company",
+       pa.schema([
+           pa.field("id", pa.int64(), nullable=False),
+           pa.field("name", pa.large_string()),
+       ]),
+   )
+   WorksAt = EdgeType(
+       "WORKS_AT",
+       source=Person,
+       destination=Company,
+       properties=pa.schema([pa.field("since", pa.int32(), nullable=False)]),
+   )
+
+   schema = GraphSchema(
+       node_types=[Person, Company],
+       edge_types=[WorksAt],
+       node_id_column="id",
+       edge_source_column="src",
+       edge_destination_column="dst",
+   )
+
+   g = graphistry.bind(
+       source="src",
+       destination="dst",
+       node="id",
+       schema=schema,
+   )
+
+   g.gfql_validate("MATCH (p:Person)-[:WORKS_AT]->(c:Company) RETURN p.name")
+
+Schema Objects
+--------------
+
+``NodeType(name, properties, labels=None)``
+  Declares a node contract. ``labels`` defaults to ``(name,)`` and maps to the
+  existing GFQL label-column convention ``label__<Label>``. ``properties``
+  accepts a ``pyarrow.Schema``, a GFQL ``RowSchema``, or a mapping shorthand
+  such as ``{"id": pa.int64(), "name": pa.large_string()}`` or
+  ``{"id": int, "name": str}``. Arrow schemas are the preferred declaration
+  path because they preserve dtype and nullability.
+
+``EdgeType(name, source, destination, properties=None)``
+  Declares an edge contract and topology. ``source`` and ``destination`` accept
+  ``NodeType`` objects, label strings, or label iterables. Edge properties use
+  the same Arrow-aligned schema inputs as node properties.
+
+``GraphSchema(node_types, edge_types, strict=True, ...)``
+  Groups node/edge contracts and adapts them to the internal
+  ``GraphSchemaCatalog`` used by binder/preflight validation. ``strict=False``
+  makes schema-bound ``g.gfql_validate(...)`` permissive by default; callers can
+  still override per call with ``g.gfql_validate(..., strict=True)``.
+
+``NodeType.to_arrow()`` and ``EdgeType.to_arrow()``
+  Export declarations as ``pyarrow.Schema`` objects through GFQL's row-schema
+  bridge. Label/type columns are included by default so exports line up with
+  the table columns used by binder/preflight validation.
+
+What Preflight Checks
+---------------------
+
+When a schema is bound to a graph, Cypher preflight checks validate:
+
+* node labels against declared node types,
+* node and edge property names against declared properties,
+* relationship types against declared edge types, and
+* relationship source/destination labels against declared topology when the
+  query provides enough label information.
+
+Invalid queries raise ``GFQLValidationError`` with structured context.
+
+Compatibility Notes
+-------------------
+
+The public import path is stable:
+
+.. code-block:: python
+
+   from graphistry.schema import NodeType, EdgeType, GraphSchema
+
+Top-level imports are also available:
+
+.. code-block:: python
+
+   from graphistry import NodeType, EdgeType, GraphSchema
+
+This lane exposes declaration, Arrow row-schema export, and binder/preflight
+integration. Inference from existing plottables remains a separate follow-on
+surface.
diff --git a/graphistry/Plottable.py b/graphistry/Plottable.py
index 1da9104e93..ed2b82d7a1 100644
--- a/graphistry/Plottable.py
+++ b/graphistry/Plottable.py
@@ -355,6 +355,7 @@ def bind(
         url: Optional[str] = None,
         nodes_file_id: Optional[str] = None,
         edges_file_id: Optional[str] = None,
+        schema: Optional[Any] = None,
     ) -> 'Plottable':
         ...
 
diff --git a/graphistry/PlotterBase.py b/graphistry/PlotterBase.py
index 5635eded80..0d4490b080 100644
--- a/graphistry/PlotterBase.py
+++ b/graphistry/PlotterBase.py
@@ -227,6 +227,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         self._point_y : Optional[str] = None
         self._point_longitude : Optional[str] = None
         self._point_latitude : Optional[str] = None
+        self._gfql_schema : Any = None
         # Settings
         self._height : int = 500
         self._render : RenderModesConcrete = resolve_render_mode(self, True)
@@ -1529,6 +1530,7 @@ def bind(self,
             url: Optional[str] = None,
             nodes_file_id: Optional[str] = None,
             edges_file_id: Optional[str] = None,
+            schema: Optional[Any] = None,
         ) -> Plottable:
         """Relate data attributes to graph structure and visual representation. To facilitate reuse and replayable notebooks, the binding call is chainable. Invocation does not effect the old binding: it instead returns a new Plotter instance with the new bindings added to the existing ones. Both the old and new bindings can then be used for different graphs.
 
@@ -1598,6 +1600,9 @@ def bind(self,
         :param edges_file_id: Remote edges file id
         :type edges_file_id: Optional[str]
 
+        :param schema: Optional public GFQL schema declaration from ``graphistry.schema``.
+        :type schema: Optional[Any]
+
         :returns: Plotter
         :rtype: Plotter
 
@@ -1677,6 +1682,7 @@ def bind(self,
         res._url = url or self._url
         res._nodes_file_id = nodes_file_id or self._nodes_file_id
         res._edges_file_id = edges_file_id or self._edges_file_id
+        res._gfql_schema = schema if schema is not None else self._gfql_schema
 
         # Invalidate dataset_id if we're changing encodings, not setting IDs
         encoding_params_changed = any([
diff --git a/graphistry/__init__.py b/graphistry/__init__.py
index 30c537155c..1a3370baf5 100644
--- a/graphistry/__init__.py
+++ b/graphistry/__init__.py
@@ -125,6 +125,13 @@
 
 from graphistry.Engine import Engine
 
+from graphistry.schema import (
+    EdgeTopology,
+    EdgeType,
+    GraphSchema,
+    NodeType,
+)
+
 from graphistry.privacy import (
     Mode, Privacy
 )
diff --git a/graphistry/compute/gfql/frontends/cypher/binder.py b/graphistry/compute/gfql/frontends/cypher/binder.py
index 3f8a5b0193..886aa57817 100644
--- a/graphistry/compute/gfql/frontends/cypher/binder.py
+++ b/graphistry/compute/gfql/frontends/cypher/binder.py
@@ -371,16 +371,18 @@ def _bind_match_clause(self, state: _BindState, clause: MatchClause) -> None:
                         )
                     continue
 
+                src_labels = _node_labels_at(pattern=pattern, idx=element_idx - 1)
+                dst_labels = _node_labels_at(pattern=pattern, idx=element_idx + 1)
                 self._validate_relationship_pattern_schema(
                     state=state,
                     alias=element.variable,
                     relationship_pattern=element,
                     stage=clause_kind,
                     location=f"{clause_kind}.patterns[{pattern_idx}].elements[{element_idx}]",
+                    source_labels=src_labels,
+                    destination_labels=dst_labels,
                 )
                 if element.variable is not None:
-                    src_labels = _node_labels_at(pattern=pattern, idx=element_idx - 1)
-                    dst_labels = _node_labels_at(pattern=pattern, idx=element_idx + 1)
                     changed_aliases.add(
                         self._bind_relationship_pattern(
                             state=state,
@@ -671,6 +673,7 @@ def _validate_node_pattern_schema(
         node_columns = state.catalog.node_columns
         if not node_columns:
             return
+        columns = _catalog_node_columns_for_labels(state.catalog, frozenset(node_pattern.labels)) or node_columns
         alias_label = alias or "<anonymous-node>"
         for label in node_pattern.labels:
             label_col = f"label__{label}"
@@ -683,14 +686,14 @@ def _validate_node_pattern_schema(
                     available_labels=_catalog_node_labels(state.catalog),
                 )
         for entry in node_pattern.properties:
-            if entry.key not in node_columns:
+            if entry.key not in columns:
                 raise _missing_property_in_schema_error(
                     alias=alias_label,
                     property_name=entry.key,
                     stage=stage,
                     field=f"{location}.properties",
                     entity_kind="node",
-                    available_columns=tuple(sorted(node_columns)),
+                    available_columns=tuple(sorted(columns)),
                 )
 
     def _validate_relationship_pattern_schema(
@@ -701,6 +704,8 @@ def _validate_relationship_pattern_schema(
         relationship_pattern: RelationshipPattern,
         stage: str,
         location: str,
+        source_labels: FrozenSet[str] = frozenset(),
+        destination_labels: FrozenSet[str] = frozenset(),
     ) -> None:
         if not _strict_schema_mode(state):
             return
@@ -708,17 +713,78 @@ def _validate_relationship_pattern_schema(
         if not edge_columns:
             return
         alias_label = alias or "<anonymous-edge>"
+        scoped_columns = _catalog_edge_columns_for_types(state.catalog, relationship_pattern.types)
+        columns = scoped_columns if scoped_columns is not None else edge_columns
+        available_types = _catalog_edge_types(state.catalog)
+        if available_types:
+            for rel_type in relationship_pattern.types:
+                if rel_type not in available_types:
+                    raise _missing_relationship_type_in_schema_error(
+                        alias=alias_label,
+                        relationship_type=rel_type,
+                        stage=stage,
+                        field=f"{location}.types",
+                        available_relationship_types=available_types,
+                    )
+        self._validate_relationship_topology_schema(
+            state=state,
+            alias=alias_label,
+            relationship_pattern=relationship_pattern,
+            stage=stage,
+            field=f"{location}.topology",
+            source_labels=source_labels,
+            destination_labels=destination_labels,
+        )
         for entry in relationship_pattern.properties:
-            if entry.key not in edge_columns:
+            if entry.key not in columns:
                 raise _missing_property_in_schema_error(
                     alias=alias_label,
                     property_name=entry.key,
                     stage=stage,
                     field=f"{location}.properties",
                     entity_kind="edge",
-                    available_columns=tuple(sorted(edge_columns)),
+                    available_columns=tuple(sorted(columns)),
                 )
 
+    def _validate_relationship_topology_schema(
+        self,
+        *,
+        state: _BindState,
+        alias: str,
+        relationship_pattern: RelationshipPattern,
+        stage: str,
+        field: str,
+        source_labels: FrozenSet[str],
+        destination_labels: FrozenSet[str],
+    ) -> None:
+        if not relationship_pattern.types:
+            return
+        topologies = _catalog_edge_topologies(state.catalog)
+        if not topologies:
+            return
+        if not source_labels and not destination_labels:
+            return
+        for rel_type in relationship_pattern.types:
+            candidates = tuple(t for t in topologies if t["relationship_type"] == rel_type)
+            if not candidates:
+                continue
+            if _relationship_topology_matches(
+                candidates=candidates,
+                direction=relationship_pattern.direction,
+                source_labels=source_labels,
+                destination_labels=destination_labels,
+            ):
+                return
+        raise _relationship_topology_error(
+            alias=alias,
+            relationship_types=relationship_pattern.types,
+            stage=stage,
+            field=field,
+            source_labels=tuple(sorted(source_labels)),
+            destination_labels=tuple(sorted(destination_labels)),
+            available_topologies=topologies,
+        )
+
     def _validate_where_clause_schema(self, *, state: _BindState, where: WhereClause, stage: str) -> None:
         if not _strict_schema_mode(state):
             return
@@ -741,12 +807,16 @@ def _validate_where_clause_schema(self, *, state: _BindState, where: WhereClause
                             location=f"{stage}.where.pattern[{idx}].elements[{pattern_idx}]",
                         )
                     else:
+                        source_labels = _node_labels_at(pattern=term.pattern, idx=pattern_idx - 1)
+                        destination_labels = _node_labels_at(pattern=term.pattern, idx=pattern_idx + 1)
                         self._validate_relationship_pattern_schema(
                             state=state,
                             alias=pattern_element.variable,
                             relationship_pattern=pattern_element,
                             stage=stage,
                             location=f"{stage}.where.pattern[{idx}].elements[{pattern_idx}]",
+                            source_labels=source_labels,
+                            destination_labels=destination_labels,
                         )
         if where.expr_tree is not None:
             self._validate_expression_property_refs(
@@ -864,10 +934,18 @@ def _validate_property_ref_schema(
                 return
             raise _unresolved_name_error(identifier=property_ref.alias, visible_scope=state.scope)
         if source.entity_kind == "node":
-            columns = state.catalog.node_columns
+            node_labels: FrozenSet[str] = frozenset()
+            if isinstance(source.logical_type, NodeRef):
+                node_labels = source.logical_type.labels
+            scoped_columns = _catalog_node_columns_for_labels(state.catalog, node_labels)
+            columns = scoped_columns if scoped_columns is not None else state.catalog.node_columns
             entity_kind: Literal["node", "edge"] = "node"
         elif source.entity_kind == "edge":
-            columns = state.catalog.edge_columns
+            edge_types: Tuple[str, ...] = tuple()
+            if isinstance(source.logical_type, EdgeRef) and source.logical_type.type is not None:
+                edge_types = (source.logical_type.type,)
+            scoped_columns = _catalog_edge_columns_for_types(state.catalog, edge_types)
+            columns = scoped_columns if scoped_columns is not None else state.catalog.edge_columns
             entity_kind = "edge"
         else:
             return
@@ -1720,8 +1798,7 @@ def _is_hidden_reentry_property(property_name: str) -> bool:
 
 
 def _strict_schema_mode(state: _BindState) -> bool:
-    _ = state
-    return True
+    return bool(state.catalog.metadata.get("strict", True))
 
 
 def _catalog_node_labels(catalog: GraphSchemaCatalog) -> Tuple[str, ...]:
@@ -1733,6 +1810,106 @@ def _catalog_node_labels(catalog: GraphSchemaCatalog) -> Tuple[str, ...]:
     return tuple(labels)
 
 
+def _catalog_edge_types(catalog: GraphSchemaCatalog) -> Tuple[str, ...]:
+    metadata_types = catalog.metadata.get("edge_types")
+    if isinstance(metadata_types, (list, tuple, set, frozenset)):
+        return tuple(sorted(str(value) for value in metadata_types))
+    types = sorted(
+        str(column).split("label__", 1)[1]
+        for column in catalog.edge_columns
+        if str(column).startswith("label__")
+    )
+    return tuple(types)
+
+
+def _catalog_edge_topologies(catalog: GraphSchemaCatalog) -> Tuple[Dict[str, object], ...]:
+    raw = catalog.metadata.get("edge_topologies")
+    if not isinstance(raw, (list, tuple)):
+        return tuple()
+    out: List[Dict[str, object]] = []
+    for item in raw:
+        if not isinstance(item, Mapping):
+            continue
+        relationship_type = item.get("relationship_type")
+        if relationship_type is None:
+            continue
+        source_labels = item.get("source_labels", ())
+        destination_labels = item.get("destination_labels", ())
+        out.append(
+            {
+                "relationship_type": str(relationship_type),
+                "source_labels": tuple(str(label) for label in source_labels),
+                "destination_labels": tuple(str(label) for label in destination_labels),
+            }
+        )
+    return tuple(out)
+
+
+def _catalog_node_columns_for_labels(catalog: GraphSchemaCatalog, labels: FrozenSet[str]) -> Optional[FrozenSet[str]]:
+    if not labels:
+        return None
+    raw = catalog.metadata.get("node_columns_by_label")
+    if not isinstance(raw, Mapping):
+        return None
+    matched: FrozenSet[str] = frozenset()
+    found = False
+    for label in labels:
+        columns = raw.get(label)
+        if isinstance(columns, (list, tuple, set, frozenset)):
+            matched = matched | frozenset(str(column) for column in columns)
+            found = True
+    if not found:
+        return None
+    return matched
+
+
+def _catalog_edge_columns_for_types(catalog: GraphSchemaCatalog, relationship_types: Tuple[str, ...]) -> Optional[FrozenSet[str]]:
+    if not relationship_types:
+        return None
+    raw = catalog.metadata.get("edge_columns_by_type")
+    if not isinstance(raw, Mapping):
+        return None
+    matched: FrozenSet[str] = frozenset()
+    found = False
+    for relationship_type in relationship_types:
+        columns = raw.get(relationship_type)
+        if isinstance(columns, (list, tuple, set, frozenset)):
+            matched = matched | frozenset(str(column) for column in columns)
+            found = True
+    if not found:
+        return None
+    return matched
+
+
+def _labels_match(required: Tuple[str, ...], actual: FrozenSet[str]) -> bool:
+    if not actual or not required:
+        return True
+    return not actual.isdisjoint(set(required))
+
+
+def _relationship_topology_matches(
+    *,
+    candidates: Tuple[Dict[str, object], ...],
+    direction: Literal["forward", "reverse", "undirected"],
+    source_labels: FrozenSet[str],
+    destination_labels: FrozenSet[str],
+) -> bool:
+    for candidate in candidates:
+        required_source = cast(Tuple[str, ...], candidate["source_labels"])
+        required_destination = cast(Tuple[str, ...], candidate["destination_labels"])
+        if direction == "reverse":
+            if _labels_match(required_source, destination_labels) and _labels_match(required_destination, source_labels):
+                return True
+        elif direction == "undirected":
+            forward_ok = _labels_match(required_source, source_labels) and _labels_match(required_destination, destination_labels)
+            reverse_ok = _labels_match(required_source, destination_labels) and _labels_match(required_destination, source_labels)
+            if forward_ok or reverse_ok:
+                return True
+        elif _labels_match(required_source, source_labels) and _labels_match(required_destination, destination_labels):
+            return True
+    return False
+
+
 def _missing_label_in_schema_error(
     *,
     alias: str,
@@ -1755,6 +1932,54 @@ def _missing_label_in_schema_error(
     )
 
 
+def _missing_relationship_type_in_schema_error(
+    *,
+    alias: str,
+    relationship_type: str,
+    stage: str,
+    field: str,
+    available_relationship_types: Tuple[str, ...],
+) -> GFQLValidationError:
+    return GFQLValidationError(
+        ErrorCode.E301,
+        "Cypher relationship type is missing from strict binder schema catalog",
+        field=field,
+        value=f"{alias}:{relationship_type}",
+        suggestion="Use relationship types that exist in the edge schema or extend the schema catalog.",
+        alias=alias,
+        relationship_type=relationship_type,
+        stage=stage,
+        available_relationship_types=available_relationship_types,
+        language="cypher",
+    )
+
+
+def _relationship_topology_error(
+    *,
+    alias: str,
+    relationship_types: Tuple[str, ...],
+    stage: str,
+    field: str,
+    source_labels: Tuple[str, ...],
+    destination_labels: Tuple[str, ...],
+    available_topologies: Tuple[Dict[str, object], ...],
+) -> GFQLValidationError:
+    return GFQLValidationError(
+        ErrorCode.E301,
+        "Cypher relationship topology is incompatible with strict binder schema catalog",
+        field=field,
+        value=f"{source_labels}-[{relationship_types}]->{destination_labels}",
+        suggestion="Use source/destination labels allowed by the edge schema topology or extend the schema catalog.",
+        alias=alias,
+        relationship_types=relationship_types,
+        source_labels=source_labels,
+        destination_labels=destination_labels,
+        stage=stage,
+        available_topologies=available_topologies,
+        language="cypher",
+    )
+
+
 def _missing_property_in_schema_error(
     *,
     alias: str,
diff --git a/graphistry/compute/gfql_validate.py b/graphistry/compute/gfql_validate.py
index 65d7096a73..bce8125b56 100644
--- a/graphistry/compute/gfql_validate.py
+++ b/graphistry/compute/gfql_validate.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from typing import Any, Dict, List, Literal, Mapping, NoReturn, Optional, Sequence, Tuple, Union, cast
+from dataclasses import replace
 
 from graphistry.Plottable import Plottable
 from graphistry.compute.ast import ASTLet, ASTObject, ASTNode, ASTEdge, ASTCall, ASTRef, from_json
@@ -78,7 +79,32 @@ def _raise_diagnostics(
     )
 
 
-def _build_schema_catalog(g: Plottable, *, strict: bool) -> GraphSchemaCatalog:
+def _build_schema_catalog(g: Plottable, *, strict: Optional[bool]) -> GraphSchemaCatalog:
+    bound_schema = getattr(g, "_gfql_schema", None)
+    if bound_schema is not None:
+        node_id_column = getattr(g, "_node", None)
+        edge_source_column = getattr(g, "_source", None)
+        edge_destination_column = getattr(g, "_destination", None)
+        if hasattr(bound_schema, "to_catalog") and callable(getattr(bound_schema, "to_catalog")):
+            return cast(Any, bound_schema).to_catalog(
+                node_id_column=node_id_column,
+                edge_source_column=edge_source_column,
+                edge_destination_column=edge_destination_column,
+                strict=strict,
+            )
+        if isinstance(bound_schema, GraphSchemaCatalog):
+            metadata = dict(bound_schema.metadata)
+            if strict is not None:
+                metadata["strict"] = bool(strict)
+            return replace(
+                bound_schema,
+                node_id_column=node_id_column or bound_schema.node_id_column,
+                edge_source_column=edge_source_column or bound_schema.edge_source_column,
+                edge_destination_column=edge_destination_column or bound_schema.edge_destination_column,
+                metadata=metadata,
+            )
+
+    strict_value = True if strict is None else bool(strict)
     node_columns: Tuple[str, ...] = tuple()
     edge_columns: Tuple[str, ...] = tuple()
     if getattr(g, "_nodes", None) is not None:
@@ -91,20 +117,35 @@ def _build_schema_catalog(g: Plottable, *, strict: bool) -> GraphSchemaCatalog:
         node_id_column=getattr(g, "_node", None),
         edge_source_column=getattr(g, "_source", None),
         edge_destination_column=getattr(g, "_destination", None),
-        metadata={"strict": strict},
+        metadata={"strict": strict_value},
     )
 
 
+def _resolve_strict_mode(g: Plottable, *, strict: Optional[bool]) -> bool:
+    if strict is not None:
+        return bool(strict)
+    bound_schema = getattr(g, "_gfql_schema", None)
+    if bound_schema is not None:
+        schema_strict = getattr(bound_schema, "strict", None)
+        if schema_strict is not None:
+            return bool(schema_strict)
+        metadata = getattr(bound_schema, "metadata", None)
+        if isinstance(metadata, Mapping) and "strict" in metadata:
+            return bool(metadata["strict"])
+    return True
+
+
 def _validate_cypher(
     g: Plottable,
     query: str,
     *,
     params: Optional[Mapping[str, Any]],
-    strict: bool,
+    strict: Optional[bool],
 ) -> Dict[str, Any]:
     parsed = parse_cypher(query)
-    if strict:
-        strict_ctx = PlanContext(catalog=_build_schema_catalog(g, strict=True))
+    strict_mode = _resolve_strict_mode(g, strict=strict)
+    if strict_mode:
+        strict_ctx = PlanContext(catalog=_build_schema_catalog(g, strict=strict))
         FrontendBinder().bind(parsed, strict_ctx, strict_name_resolution=True)
     compiled = compile_cypher_query(parsed, params=params)
     compiled_kind: Literal["query", "union", "graph"] = "query"
@@ -336,7 +377,7 @@ def gfql_validate(
     where: Optional[Sequence[WhereComparison]] = None,
     language: Optional[Literal["cypher", "gremlin"]] = None,
     params: Optional[Mapping[str, Any]] = None,
-    strict: bool = True,
+    strict: Optional[bool] = None,
     collect_all: bool = False,
     schema: bool = True,
 ) -> Dict[str, Any]:
diff --git a/graphistry/pygraphistry.py b/graphistry/pygraphistry.py
index 11e7ddbc4a..57782c032b 100644
--- a/graphistry/pygraphistry.py
+++ b/graphistry/pygraphistry.py
@@ -1888,6 +1888,7 @@ def bind(
         url: Optional[str] = None,
         nodes_file_id: Optional[str] = None,
         edges_file_id: Optional[str] = None,
+        schema: Optional[Any] = None,
     ) -> Plotter:
         """Create a base plotter.
 
@@ -1909,6 +1910,7 @@ def bind(
             source=source,
             destination=destination,
             node=node,
+            edge=edge,
             edge_title=edge_title,
             edge_label=edge_label,
             edge_color=edge_color,
@@ -1929,7 +1931,11 @@ def bind(
             point_y=point_y,
             point_longitude=point_longitude,
             point_latitude=point_latitude,
-            dataset_id=dataset_id
+            dataset_id=dataset_id,
+            url=url,
+            nodes_file_id=nodes_file_id,
+            edges_file_id=edges_file_id,
+            schema=schema,
         ))
 
     def from_dataset_id(self, dataset_id: str, api_token: Optional[str] = None) -> Plotter:
diff --git a/graphistry/schema.py b/graphistry/schema.py
new file mode 100644
index 0000000000..401e9a4712
--- /dev/null
+++ b/graphistry/schema.py
@@ -0,0 +1,333 @@
+"""Public declarative graph schema model for GFQL validation."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, FrozenSet, Iterable, Mapping, Optional, Tuple, Union
+
+from graphistry.compute.gfql.ir.compilation import GraphSchemaCatalog
+from graphistry.compute.gfql.ir.arrow_bridge import CoercionMode, from_arrow, to_arrow
+from graphistry.compute.gfql.ir.logical_plan import RowSchema
+from graphistry.compute.gfql.ir.types import EdgeRef, ListType, LogicalType, NodeRef, PathType, ScalarType
+
+
+NodeRefInput = Union["NodeType", str, Iterable[str]]
+PropertySchemaInput = Union[Mapping[str, Any], RowSchema, Any]
+
+
+def _is_arrow_schema(value: Any) -> bool:
+    try:
+        import pyarrow as pa
+    except ImportError:
+        return False
+    return isinstance(value, pa.Schema)
+
+
+def _is_arrow_field(value: Any) -> bool:
+    try:
+        import pyarrow as pa
+    except ImportError:
+        return False
+    return isinstance(value, pa.Field)
+
+
+def _is_arrow_data_type(value: Any) -> bool:
+    try:
+        import pyarrow as pa
+    except ImportError:
+        return False
+    return isinstance(value, pa.DataType)
+
+
+def _logical_type_from_annotation(value: Any) -> LogicalType:
+    if isinstance(value, (EdgeRef, ListType, NodeRef, PathType, ScalarType)):
+        return value
+
+    if value is str:
+        return ScalarType("string")
+    if value is int:
+        return ScalarType("int64")
+    if value is float:
+        return ScalarType("float64")
+    if value is bool:
+        return ScalarType("bool")
+    if value is bytes:
+        return ScalarType("binary")
+
+    if isinstance(value, str):
+        return ScalarType(value)
+
+    if _is_arrow_field(value):
+        import pyarrow as pa
+
+        row_schema, _ = from_arrow(pa.schema([value]), default_confidence="declared")
+        return row_schema.columns[str(value.name)]
+
+    if _is_arrow_data_type(value):
+        import pyarrow as pa
+
+        row_schema, _ = from_arrow(
+            pa.schema([pa.field("__property__", value)]),
+            default_confidence="declared",
+        )
+        return row_schema.columns["__property__"]
+
+    return ScalarType("unknown")
+
+
+def _normalize_properties(properties: Optional[PropertySchemaInput]) -> Dict[str, LogicalType]:
+    if properties is None:
+        return {}
+
+    if isinstance(properties, RowSchema):
+        return {str(name): logical_type for name, logical_type in properties.columns.items()}
+
+    if _is_arrow_schema(properties):
+        row_schema, _ = from_arrow(properties, default_confidence="declared")
+        return {str(name): logical_type for name, logical_type in row_schema.columns.items()}
+
+    if not isinstance(properties, Mapping):
+        raise TypeError(
+            "properties must be a mapping, pyarrow.Schema, or GFQL RowSchema; "
+            f"got {type(properties)!r}"
+        )
+
+    return {
+        str(name): _logical_type_from_annotation(value)
+        for name, value in dict(properties).items()
+    }
+
+
+def _label_columns(labels: Iterable[str]) -> Dict[str, LogicalType]:
+    return {f"label__{label}": ScalarType("bool") for label in labels}
+
+
+def _normalize_labels(name: str, labels: Optional[Iterable[str]]) -> FrozenSet[str]:
+    raw = (name,) if labels is None else tuple(labels)
+    return frozenset(str(label) for label in raw if str(label))
+
+
+def _labels_from_node_ref(value: NodeRefInput) -> FrozenSet[str]:
+    if isinstance(value, NodeType):
+        return value.labels
+    if isinstance(value, str):
+        return frozenset((value,))
+    return frozenset(str(label) for label in value if str(label))
+
+
+@dataclass(frozen=True)
+class NodeType:
+    """Declarative node contract for GFQL schema validation.
+
+    ``name`` is the stable user-facing type name. ``labels`` defaults to
+    ``(name,)`` and maps to GFQL's existing ``label__<Label>`` convention.
+    """
+
+    name: str
+    properties: Mapping[str, LogicalType] = field(default_factory=dict)
+    labels: FrozenSet[str] = field(default_factory=frozenset)
+
+    def __init__(
+        self,
+        name: str,
+        properties: Optional[PropertySchemaInput] = None,
+        labels: Optional[Iterable[str]] = None,
+    ) -> None:
+        object.__setattr__(self, "name", str(name))
+        object.__setattr__(self, "properties", _normalize_properties(properties))
+        object.__setattr__(self, "labels", _normalize_labels(str(name), labels))
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "name", str(self.name))
+        object.__setattr__(self, "properties", _normalize_properties(self.properties))
+        object.__setattr__(self, "labels", _normalize_labels(self.name, self.labels))
+
+    @property
+    def columns(self) -> FrozenSet[str]:
+        """Columns admitted for this node type, including label columns."""
+        property_columns = frozenset(str(name) for name in self.properties.keys())
+        label_columns = frozenset(f"label__{label}" for label in self.labels)
+        return property_columns | label_columns
+
+    def to_row_schema(self, *, include_labels: bool = True) -> RowSchema:
+        """Export this node contract as GFQL's Arrow-bridge row schema."""
+        columns = dict(self.properties)
+        if include_labels:
+            columns.update(_label_columns(self.labels))
+        return RowSchema(columns=columns)
+
+    def to_arrow(
+        self,
+        *,
+        include_labels: bool = True,
+        coercion: CoercionMode = "widen",
+    ) -> Any:
+        """Export this node contract as a ``pyarrow.Schema``."""
+        return to_arrow(self.to_row_schema(include_labels=include_labels), coercion=coercion)
+
+
+@dataclass(frozen=True)
+class EdgeTopology:
+    """Allowed source/destination labels for a relationship type."""
+
+    relationship_type: str
+    source_labels: FrozenSet[str]
+    destination_labels: FrozenSet[str]
+
+    def as_metadata(self) -> Dict[str, object]:
+        return {
+            "relationship_type": self.relationship_type,
+            "source_labels": tuple(sorted(self.source_labels)),
+            "destination_labels": tuple(sorted(self.destination_labels)),
+        }
+
+
+@dataclass(frozen=True)
+class EdgeType:
+    """Declarative edge contract with topology constraints."""
+
+    name: str
+    source: FrozenSet[str]
+    destination: FrozenSet[str]
+    properties: Mapping[str, LogicalType] = field(default_factory=dict)
+
+    def __init__(
+        self,
+        name: str,
+        source: NodeRefInput,
+        destination: NodeRefInput,
+        properties: Optional[PropertySchemaInput] = None,
+    ) -> None:
+        object.__setattr__(self, "name", str(name))
+        object.__setattr__(self, "properties", _normalize_properties(properties))
+        object.__setattr__(self, "source", _labels_from_node_ref(source))
+        object.__setattr__(self, "destination", _labels_from_node_ref(destination))
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "name", str(self.name))
+        object.__setattr__(self, "properties", _normalize_properties(self.properties))
+        object.__setattr__(self, "source", _labels_from_node_ref(self.source))
+        object.__setattr__(self, "destination", _labels_from_node_ref(self.destination))
+
+    @property
+    def columns(self) -> FrozenSet[str]:
+        """Columns admitted for this edge type, including relationship label."""
+        property_columns = frozenset(str(name) for name in self.properties.keys())
+        return property_columns | frozenset((f"label__{self.name}",))
+
+    @property
+    def topology(self) -> EdgeTopology:
+        return EdgeTopology(
+            relationship_type=self.name,
+            source_labels=self.source,
+            destination_labels=self.destination,
+        )
+
+    def to_row_schema(self, *, include_type_label: bool = True) -> RowSchema:
+        """Export this edge contract as GFQL's Arrow-bridge row schema."""
+        columns = dict(self.properties)
+        if include_type_label:
+            columns[f"label__{self.name}"] = ScalarType("bool")
+        return RowSchema(columns=columns)
+
+    def to_arrow(
+        self,
+        *,
+        include_type_label: bool = True,
+        coercion: CoercionMode = "widen",
+    ) -> Any:
+        """Export this edge contract as a ``pyarrow.Schema``."""
+        return to_arrow(self.to_row_schema(include_type_label=include_type_label), coercion=coercion)
+
+
+@dataclass(frozen=True)
+class GraphSchema:
+    """Public graph schema contract adaptable to GFQL's internal catalog."""
+
+    node_types: Tuple[NodeType, ...] = field(default_factory=tuple)
+    edge_types: Tuple[EdgeType, ...] = field(default_factory=tuple)
+    strict: bool = True
+    node_id_column: Optional[str] = None
+    edge_source_column: Optional[str] = None
+    edge_destination_column: Optional[str] = None
+
+    def __init__(
+        self,
+        node_types: Iterable[NodeType] = (),
+        edge_types: Iterable[EdgeType] = (),
+        *,
+        strict: bool = True,
+        node_id_column: Optional[str] = None,
+        edge_source_column: Optional[str] = None,
+        edge_destination_column: Optional[str] = None,
+    ) -> None:
+        object.__setattr__(self, "node_types", tuple(node_types))
+        object.__setattr__(self, "edge_types", tuple(edge_types))
+        object.__setattr__(self, "strict", bool(strict))
+        object.__setattr__(self, "node_id_column", node_id_column)
+        object.__setattr__(self, "edge_source_column", edge_source_column)
+        object.__setattr__(self, "edge_destination_column", edge_destination_column)
+
+    @property
+    def node_columns(self) -> FrozenSet[str]:
+        columns: FrozenSet[str] = frozenset()
+        for node_type in self.node_types:
+            columns = columns | node_type.columns
+        return columns
+
+    @property
+    def edge_columns(self) -> FrozenSet[str]:
+        columns: FrozenSet[str] = frozenset()
+        for edge_type in self.edge_types:
+            columns = columns | edge_type.columns
+        return columns
+
+    @property
+    def node_columns_by_label(self) -> Dict[str, Tuple[str, ...]]:
+        columns_by_label: Dict[str, FrozenSet[str]] = {}
+        for node_type in self.node_types:
+            for label in node_type.labels:
+                columns_by_label[label] = columns_by_label.get(label, frozenset()) | node_type.columns
+        return {label: tuple(sorted(columns)) for label, columns in sorted(columns_by_label.items())}
+
+    @property
+    def edge_columns_by_type(self) -> Dict[str, Tuple[str, ...]]:
+        return {edge_type.name: tuple(sorted(edge_type.columns)) for edge_type in self.edge_types}
+
+    def to_catalog(
+        self,
+        *,
+        node_id_column: Optional[str] = None,
+        edge_source_column: Optional[str] = None,
+        edge_destination_column: Optional[str] = None,
+        strict: Optional[bool] = None,
+    ) -> GraphSchemaCatalog:
+        """Adapt this public schema into the internal GFQL schema catalog."""
+        metadata = {
+            "schema_model": "graphistry.schema.GraphSchema",
+            "strict": self.strict if strict is None else bool(strict),
+            "node_types": tuple(node_type.name for node_type in self.node_types),
+            "edge_types": tuple(edge_type.name for edge_type in self.edge_types),
+            "edge_topologies": tuple(edge_type.topology.as_metadata() for edge_type in self.edge_types),
+            "node_columns_by_label": self.node_columns_by_label,
+            "edge_columns_by_type": self.edge_columns_by_type,
+            "node_row_schemas": {
+                node_type.name: node_type.to_row_schema(include_labels=False)
+                for node_type in self.node_types
+            },
+            "edge_row_schemas": {
+                edge_type.name: edge_type.to_row_schema(include_type_label=False)
+                for edge_type in self.edge_types
+            },
+        }
+        return GraphSchemaCatalog.from_schema_parts(
+            node_columns=self.node_columns,
+            edge_columns=self.edge_columns,
+            node_id_column=node_id_column or self.node_id_column,
+            edge_source_column=edge_source_column or self.edge_source_column,
+            edge_destination_column=edge_destination_column or self.edge_destination_column,
+            metadata=metadata,
+        )
+
+
+__all__ = ["EdgeTopology", "EdgeType", "GraphSchema", "NodeType"]
diff --git a/graphistry/tests/compute/gfql/cypher/test_binder.py b/graphistry/tests/compute/gfql/cypher/test_binder.py
index 7dee749938..c66d3647bd 100644
--- a/graphistry/tests/compute/gfql/cypher/test_binder.py
+++ b/graphistry/tests/compute/gfql/cypher/test_binder.py
@@ -726,20 +726,19 @@ def test_binder_strict_schema_accepts_admitted_label_and_property_shapes() -> No
     assert "nid" in bound.semantic_table.variables
 
 
-def test_binder_retired_loose_mode_rejects_missing_schema_fields() -> None:
+def test_binder_schema_metadata_can_disable_schema_field_checks() -> None:
     ctx = _strict_catalog_ctx(
         node_columns=["id"],
         edge_columns=["src", "dst"],
         strict=False,
     )
-    with pytest.raises(GFQLValidationError) as exc_info:
-        FrontendBinder().bind(
-            parse_cypher("MATCH (n:Person) RETURN n.unknown AS u"),
-            ctx,
-            strict_name_resolution=False,
-        )
-    assert exc_info.value.code == ErrorCode.E301
-    assert exc_info.value.context["field"] == "MATCH.patterns[0].elements[0].labels"
+    bound = FrontendBinder().bind(
+        parse_cypher("MATCH (n:Person) RETURN n.unknown AS u"),
+        ctx,
+        strict_name_resolution=False,
+    )
+
+    assert "u" in bound.semantic_table.variables
 
 
 def test_binder_schema_enforced_even_when_legacy_strict_name_flag_false() -> None:
diff --git a/graphistry/tests/compute/gfql/test_public_schema.py b/graphistry/tests/compute/gfql/test_public_schema.py
new file mode 100644
index 0000000000..c556683d14
--- /dev/null
+++ b/graphistry/tests/compute/gfql/test_public_schema.py
@@ -0,0 +1,220 @@
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+import graphistry
+from graphistry.compute.exceptions import ErrorCode, GFQLValidationError
+from graphistry.compute.gfql.ir.logical_plan import RowSchema
+from graphistry.compute.gfql.ir.types import ScalarType
+from graphistry.schema import EdgeType, GraphSchema, NodeType
+
+
+def _schema(*, strict: bool = True) -> GraphSchema:
+    person = NodeType("Person", {"age": int, "id": int, "name": str})
+    company = NodeType("Company", {"id": int, "name": str})
+    works_at = EdgeType("WORKS_AT", source=person, destination=company, properties={"since": int})
+    contracts = EdgeType("CONTRACTS", source=company, destination=person, properties={"fee": int})
+    return GraphSchema(
+        node_types=[person, company],
+        edge_types=[works_at, contracts],
+        strict=strict,
+        node_id_column="id",
+        edge_source_column="src",
+        edge_destination_column="dst",
+    )
+
+
+def _graph(schema: GraphSchema):
+    nodes = pd.DataFrame(
+        [
+            {"id": 1, "name": "alice", "label__Person": True, "label__Company": False},
+            {"id": 2, "name": "acme", "label__Person": False, "label__Company": True},
+        ]
+    )
+    edges = pd.DataFrame(
+        [
+            {"src": 1, "dst": 2, "since": 2020, "label__WORKS_AT": True},
+        ]
+    )
+    return graphistry.bind(source="src", destination="dst", node="id", schema=schema).edges(edges).nodes(nodes)
+
+
+def test_public_schema_imports_are_stable() -> None:
+    assert graphistry.NodeType is NodeType
+    assert graphistry.EdgeType is EdgeType
+    assert graphistry.GraphSchema is GraphSchema
+
+
+def test_graph_schema_adapts_to_internal_catalog() -> None:
+    catalog = _schema(strict=False).to_catalog()
+
+    assert catalog.node_columns == frozenset({"age", "id", "name", "label__Person", "label__Company"})
+    assert catalog.edge_columns == frozenset({"fee", "since", "label__CONTRACTS", "label__WORKS_AT"})
+    assert catalog.node_id == "id"
+    assert catalog.edge_source == "src"
+    assert catalog.edge_destination == "dst"
+    assert catalog.metadata["strict"] is False
+    assert catalog.metadata["edge_types"] == ("WORKS_AT", "CONTRACTS")
+    assert catalog.metadata["edge_topologies"] == (
+        {
+            "relationship_type": "WORKS_AT",
+            "source_labels": ("Person",),
+            "destination_labels": ("Company",),
+        },
+        {
+            "relationship_type": "CONTRACTS",
+            "source_labels": ("Company",),
+            "destination_labels": ("Person",),
+        },
+    )
+    assert catalog.metadata["node_columns_by_label"]["Person"] == ("age", "id", "label__Person", "name")
+    assert catalog.metadata["node_columns_by_label"]["Company"] == ("id", "label__Company", "name")
+    assert catalog.metadata["edge_columns_by_type"]["CONTRACTS"] == ("fee", "label__CONTRACTS")
+    assert catalog.metadata["edge_columns_by_type"]["WORKS_AT"] == ("label__WORKS_AT", "since")
+    assert catalog.metadata["node_row_schemas"]["Person"] == RowSchema(
+        columns={
+            "age": ScalarType("int64"),
+            "id": ScalarType("int64"),
+            "name": ScalarType("string"),
+        }
+    )
+    assert catalog.metadata["edge_row_schemas"]["WORKS_AT"] == RowSchema(
+        columns={"since": ScalarType("int64")}
+    )
+
+
+def test_public_schema_accepts_and_exports_arrow_schema() -> None:
+    pa = pytest.importorskip("pyarrow")
+
+    person = NodeType(
+        "Person",
+        pa.schema(
+            [
+                pa.field("id", pa.int64(), nullable=False),
+                pa.field("name", pa.large_string()),
+            ]
+        ),
+    )
+    works_at = EdgeType(
+        "WORKS_AT",
+        source=person,
+        destination="Company",
+        properties=pa.schema([pa.field("since", pa.int32(), nullable=False)]),
+    )
+
+    assert person.properties["id"] == ScalarType("int64", nullable=False)
+    node_arrow = person.to_arrow()
+    assert node_arrow.field("id").type == pa.int64()
+    assert node_arrow.field("id").nullable is False
+    assert node_arrow.field("name").type == pa.large_string()
+    assert node_arrow.field("label__Person").type == pa.bool_()
+
+    edge_arrow = works_at.to_arrow()
+    assert edge_arrow.field("since").type == pa.int32()
+    assert edge_arrow.field("since").nullable is False
+    assert edge_arrow.field("label__WORKS_AT").type == pa.bool_()
+
+
+def test_public_schema_accepts_arrow_mapping_values() -> None:
+    pa = pytest.importorskip("pyarrow")
+
+    person = NodeType(
+        "Person",
+        {
+            "id": pa.int64(),
+            "display_name": pa.field("name", pa.large_string(), nullable=False),
+        },
+    )
+
+    assert person.properties["id"] == ScalarType("int64")
+    assert person.properties["display_name"] == ScalarType("string", nullable=False)
+    exported = person.to_arrow(include_labels=False)
+    assert exported.field("id").type == pa.int64()
+    assert exported.field("display_name").nullable is False
+
+
+def test_bind_schema_is_chainable_and_used_by_preflight() -> None:
+    schema = _schema()
+    g = _graph(schema).bind(point_color="name")
+
+    assert g._gfql_schema is schema
+    report = g.gfql_validate("MATCH (p:Person)-[:WORKS_AT]->(c:Company) RETURN p.name AS name")
+    assert report["ok"] is True
+
+
+def test_schema_bound_preflight_rejects_missing_property() -> None:
+    g = _graph(_schema())
+
+    with pytest.raises(GFQLValidationError) as exc_info:
+        g.gfql_validate("MATCH (p:Person)-[:WORKS_AT]->(c:Company) RETURN c.age AS age")
+
+    err = exc_info.value
+    assert err.code == ErrorCode.E301
+    assert err.context["property"] == "age"
+    assert err.context["entity_kind"] == "node"
+
+
+def test_schema_bound_preflight_rejects_missing_relationship_type() -> None:
+    g = _graph(_schema())
+
+    with pytest.raises(GFQLValidationError) as exc_info:
+        g.gfql_validate("MATCH (p:Person)-[:KNOWS]->(c:Company) RETURN p.name AS name")
+
+    err = exc_info.value
+    assert err.code == ErrorCode.E301
+    assert err.context["relationship_type"] == "KNOWS"
+    assert err.context["available_relationship_types"] == ("CONTRACTS", "WORKS_AT")
+
+
+def test_schema_bound_preflight_rejects_edge_property_from_different_type() -> None:
+    g = _graph(_schema())
+
+    with pytest.raises(GFQLValidationError) as exc_info:
+        g.gfql_validate("MATCH (p:Person)-[:WORKS_AT {fee: 10}]->(c:Company) RETURN p.name AS name")
+
+    err = exc_info.value
+    assert err.code == ErrorCode.E301
+    assert err.context["property"] == "fee"
+    assert err.context["entity_kind"] == "edge"
+    assert err.context["available_columns"] == ("label__WORKS_AT", "since")
+
+
+def test_schema_bound_preflight_accepts_reverse_topology() -> None:
+    g = _graph(_schema())
+
+    report = g.gfql_validate("MATCH (c:Company)<-[:WORKS_AT]-(p:Person) RETURN p.name AS name")
+
+    assert report["ok"] is True
+
+
+def test_schema_bound_preflight_rejects_topology_mismatch() -> None:
+    g = _graph(_schema())
+
+    with pytest.raises(GFQLValidationError) as exc_info:
+        g.gfql_validate("MATCH (p:Person)-[:WORKS_AT]->(other:Person) RETURN p.name AS name")
+
+    err = exc_info.value
+    assert err.code == ErrorCode.E301
+    assert err.context["relationship_types"] == ("WORKS_AT",)
+    assert err.context["source_labels"] == ("Person",)
+    assert err.context["destination_labels"] == ("Person",)
+
+
+def test_schema_bound_preflight_can_be_called_permissively() -> None:
+    g = _graph(_schema(strict=False))
+
+    report = g.gfql_validate("MATCH (p:Unknown) RETURN p")
+
+    assert report["ok"] is True
+
+
+def test_schema_bound_preflight_explicit_strict_overrides_permissive_schema() -> None:
+    g = _graph(_schema(strict=False))
+
+    with pytest.raises(GFQLValidationError) as exc_info:
+        g.gfql_validate("MATCH (p:Unknown) RETURN p", strict=True)
+
+    err = exc_info.value
+    assert err.code == ErrorCode.E301
+    assert err.context["label"] == "Unknown"