Add table-based node lookup to GraphQL

datajunction-server/datajunction_server/alembic/versions/2026_09_30_0000-tbl0001idx_add_noderevision_table_index.py

@@ -0,0 +1,29 @@
+"""add noderevision table index
+
+Revision ID: tbl0001idx
+Revises: fm0001params
+Create Date: 2026-09-30 00:00:00.000000+00:00
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "tbl0001idx"
+down_revision = "fm0001params"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with op.batch_alter_table("noderevision", schema=None) as batch_op:
+        batch_op.create_index(
+            "ix_noderevision_table",
+            [sa.text('lower("table")'), sa.text("lower(schema_)"), "catalog_id"],
+            unique=False,
+        )
+
+
+def downgrade():
+    with op.batch_alter_table("noderevision", schema=None) as batch_op:
+        batch_op.drop_index("ix_noderevision_table")

datajunction-server/datajunction_server/api/graphql/main.py

@@ -35,6 +35,7 @@
     materialization_plan,
     measures_sql,
 )
+from datajunction_server.api.graphql.queries.tables import nodes_for_table
 from datajunction_server.api.graphql.queries.tags import (
     list_tag_types,
     list_tags,
@@ -273,6 +274,11 @@ class Query:
         resolver=log_resolver(upstream_nodes),
         description="Find upstream nodes (optionally, of a given type) from a given node.",
     )
+    nodes_for_table: list[Node] = strawberry.field(
+        resolver=log_resolver(nodes_for_table),
+        description="Find the nodes associated with a physical table: the source "
+        "nodes on that table and, unless disabled, everything downstream of them.",
+    )
 
     # Generate SQL queries
     measures_sql: list[GeneratedSQL] = strawberry.field(

datajunction-server/datajunction_server/api/graphql/queries/nodes.py

@@ -127,6 +127,17 @@ async def find_nodes(
             "Accepts dimension node names or dimension attributes",
         ),
     ] = None,
+    tables: Annotated[
+        list[str] | None,
+        strawberry.argument(
+            description="Filter to nodes backed by any of these physical tables. "
+            "Each entry may be fully qualified (catalog.schema.table), partially "
+            "qualified (schema.table) or a bare table name; matching is "
+            "case-insensitive. Only source nodes carry a table, so this returns "
+            "source nodes -- use downstreamNodes to reach the transforms and "
+            "metrics built on them.",
+        ),
+    ] = None,
     edited_by: Annotated[
         str | None,
         strawberry.argument(
@@ -234,6 +245,7 @@ async def find_nodes(
         node_types=node_types,
         tags=tags,
         dimensions=dimensions,
+        tables=tables,
         edited_by=edited_by,
         namespace=namespace,
         mode=mode,
@@ -295,6 +307,17 @@ async def find_nodes_paginated(
             "Accepts dimension node names or dimension attributes",
         ),
     ] = None,
+    tables: Annotated[
+        list[str] | None,
+        strawberry.argument(
+            description="Filter to nodes backed by any of these physical tables. "
+            "Each entry may be fully qualified (catalog.schema.table), partially "
+            "qualified (schema.table) or a bare table name; matching is "
+            "case-insensitive. Only source nodes carry a table, so this returns "
+            "source nodes -- use downstreamNodes to reach the transforms and "
+            "metrics built on them.",
+        ),
+    ] = None,
     edited_by: Annotated[
         str | None,
         strawberry.argument(
@@ -395,6 +418,7 @@ async def find_nodes_paginated(
         node_types=node_types,
         tags=tags,
         dimensions=dimensions,
+        tables=tables,
         edited_by=edited_by,
         namespace=namespace,
         limit=limit + 1,
@@ -424,6 +448,7 @@ async def find_nodes_paginated(
             node_types=node_types,
             tags=tags,
             dimensions=dimensions,
+            tables=tables,
             edited_by=edited_by,
             namespace=namespace,
             mode=mode,

datajunction-server/datajunction_server/api/graphql/queries/tables.py

@@ -0,0 +1,108 @@
+"""
+Physical-table lookup queries.
+"""
+
+from typing import Annotated
+
+import strawberry
+from strawberry.types import Info
+
+from datajunction_server.api.graphql.resolvers.nodes import (
+    find_nodes_by,
+    load_node_options,
+)
+from datajunction_server.api.graphql.utils import extract_fields, resolver_session
+from datajunction_server.database.node import Node
+from datajunction_server.models.node import NodeCursor
+from datajunction_server.models.node_type import NodeType
+from datajunction_server.sql.dag import get_downstream_nodes
+
+# ``find_by`` always applies a limit, so sources are read a page at a time.
+SOURCE_PAGE_SIZE = 500
+
+
+async def _all_sources(info: Info, table: str) -> list[Node]:
+    """
+    Every source node on ``table``, read page by page so none are dropped.
+    """
+    sources: list[Node] = []
+    after = None
+    while True:
+        # One extra row: ``after`` is inclusive, so it doubles as the next cursor.
+        page = await find_nodes_by(
+            info,
+            tables=[table],
+            node_types=[NodeType.SOURCE],
+            limit=SOURCE_PAGE_SIZE + 1,
+            after=after,
+        )
+        sources.extend(page[:SOURCE_PAGE_SIZE])
+        if len(page) <= SOURCE_PAGE_SIZE:
+            return sources
+        after = NodeCursor(
+            created_at=page[SOURCE_PAGE_SIZE].created_at,
+            id=page[SOURCE_PAGE_SIZE].id,
+        ).encode()
+
+
+async def nodes_for_table(
+    table: Annotated[
+        str,
+        strawberry.argument(
+            description="The physical table to look up. May be fully qualified "
+            "(catalog.schema.table), partially qualified (schema.table) or a bare "
+            "table name; matching is case-insensitive.",
+        ),
+    ],
+    node_types: Annotated[
+        list[NodeType] | None,
+        strawberry.argument(
+            description="Filter the returned nodes to these node types. Applies to "
+            "the whole result, the source nodes included.",
+        ),
+    ] = None,
+    include_downstream: Annotated[
+        bool,
+        strawberry.argument(
+            description="Also return everything transitively downstream of the "
+            "source nodes on this table (transforms, dimensions, metrics, cubes).",
+        ),
+    ] = True,
+    include_deactivated: Annotated[
+        bool,
+        strawberry.argument(
+            description="Whether to include deactivated downstream nodes.",
+        ),
+    ] = False,
+    *,
+    info: Info,
+) -> list[Node]:
+    """
+    Return the DJ nodes associated with a physical table.
+
+    A table only ever identifies source nodes -- ``table`` is a column on the
+    source node's revision -- so the transforms, dimensions, metrics and cubes
+    built on it are reached by walking the DAG down from those sources.
+    Results are deduplicated by node ID.
+    """
+    sources = await _all_sources(info, table)
+
+    found: dict[int, Node] = {source.id: source for source in sources}
+    if include_downstream and sources:
+        async with resolver_session(info) as session:
+            options = load_node_options(extract_fields(info))
+            for source in sources:
+                downstreams = await get_downstream_nodes(
+                    session,
+                    node_name=source.name,
+                    include_deactivated=include_deactivated,
+                    options=options,
+                )
+                for node in downstreams:
+                    found.setdefault(node.id, node)
+
+    nodes = list(found.values())
+    if node_types:
+        wanted = {NodeType(node_type) for node_type in node_types}
+        nodes = [node for node in nodes if NodeType(node.type) in wanted]
+    return nodes

datajunction-server/datajunction_server/api/graphql/resolvers/nodes.py

@@ -230,6 +230,7 @@ async def find_nodes_by(
     missing_description: bool = False,
     missing_owner: bool = False,
     dimensions: list[str] | None = None,
+    tables: list[str] | None = None,
     statuses: list[NodeStatus] | None = None,
     has_materialization: bool = False,
     orphaned_dimension: bool = False,
@@ -295,6 +296,7 @@ async def find_nodes_by(
             has_materialization=has_materialization,
             orphaned_dimension=orphaned_dimension,
             dimensions=dimensions,
+            tables=tables,
             search=search,
             custom_metadata_filters=custom_metadata_filters,
         )
@@ -328,6 +330,7 @@ async def count_nodes_by(
     missing_description: bool = False,
     missing_owner: bool = False,
     dimensions: list[str] | None = None,
+    tables: list[str] | None = None,
     statuses: list[NodeStatus] | None = None,
     has_materialization: bool = False,
     orphaned_dimension: bool = False,
@@ -362,6 +365,7 @@ async def count_nodes_by(
             missing_description=missing_description,
             missing_owner=missing_owner,
             dimensions=dimensions,
+            tables=tables,
             statuses=statuses,
             has_materialization=has_materialization,
             orphaned_dimension=orphaned_dimension,

datajunction-server/datajunction_server/api/graphql/schema.graphql

@@ -528,6 +528,11 @@ type Query {
     """
     dimensions: [String!] = null
 
+    """
+    Filter to nodes backed by any of these physical tables. Each entry may be fully qualified (catalog.schema.table), partially qualified (schema.table) or a bare table name; matching is case-insensitive. Only source nodes carry a table, so this returns source nodes -- use downstreamNodes to reach the transforms and metrics built on them.
+    """
+    tables: [String!] = null
+
     """Filter to nodes edited by this user"""
     editedBy: String = null
 
@@ -593,6 +598,11 @@ type Query {
     """
     dimensions: [String!] = null
 
+    """
+    Filter to nodes backed by any of these physical tables. Each entry may be fully qualified (catalog.schema.table), partially qualified (schema.table) or a bare table name; matching is case-insensitive. Only source nodes carry a table, so this returns source nodes -- use downstreamNodes to reach the transforms and metrics built on them.
+    """
+    tables: [String!] = null
+
     """Filter to nodes edited by this user"""
     editedBy: String = null
 
@@ -675,6 +685,29 @@ type Query {
     includeDeactivated: Boolean! = false
   ): [Node!]!
 
+  """
+  Find the nodes associated with a physical table: the source nodes on that table and, unless disabled, everything downstream of them.
+  """
+  nodesForTable(
+    """
+    The physical table to look up. May be fully qualified (catalog.schema.table), partially qualified (schema.table) or a bare table name; matching is case-insensitive.
+    """
+    table: String!
+
+    """
+    Filter the returned nodes to these node types. Applies to the whole result, the source nodes included.
+    """
+    nodeTypes: [NodeType!] = null
+
+    """
+    Also return everything transitively downstream of the source nodes on this table (transforms, dimensions, metrics, cubes).
+    """
+    includeDownstream: Boolean! = true
+
+    """Whether to include deactivated downstream nodes."""
+    includeDeactivated: Boolean! = false
+  ): [Node!]!
+
   """Get measures SQL for a list of metrics, dimensions, and filters."""
   measuresSql(
     cube: CubeDefinition!