#[pg_test]
fn gql_unsupported_profile_corpus_has_stable_diagnostics() {
    let corpus: serde_json::Value = serde_json::from_str(include_str!(
        "../../../release/v1-gql-unsupported.json"
    ))
    .expect("unsupported GQL corpus must be valid JSON");
    let cases = corpus["cases"]
        .as_array()
        .expect("unsupported GQL corpus cases must be an array");

    for case in cases {
        let id = case["id"].as_str().expect("case id must be text");
        let query = case["query"].as_str().expect("case query must be text");
        let expected_state = case["sqlstate"]
            .as_str()
            .expect("case SQLSTATE must be text");
        let expected_message = case["message_contains"]
            .as_str()
            .expect("case message fragment must be text");
        let statement = format!("SELECT * FROM graph.gql({})", super::sql_literal(query));

        assert_eq!(
            sqlstate_for_error(&statement).as_deref(),
            Some(expected_state),
            "unexpected SQLSTATE for unsupported corpus case {id}"
        );
        let message = sql_error_message(&statement)
            .unwrap_or_else(|| panic!("missing error message for unsupported corpus case {id}"));
        assert!(
            message.contains(expected_message),
            "unexpected message for unsupported corpus case {id}: {message}"
        );
    }
}

#[pg_test]
fn gql_single_directed_match_matches_traverse_fixture() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (gql_count, source_id, target_id, source_table, traverse_count) = Spi::connect(|client| {
        let gql = client
            .select(
                "SELECT
                     count(*)::bigint,
                     min(row #>> '{u,_id,id}'),
                     min(row #>> '{v,_id,id}'),
                     min(row #>> '{u,_id,table}')
                 FROM graph.gql(
                     'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest) RETURN u, v'
                 )",
                None,
                &[],
            )
            .expect("gql query failed");
        let gql_row = gql.first();
        let traverse_count = client
            .select(
                "SELECT count(*)::bigint
                 FROM graph.traverse(
                     'graph_test_users_pgtest'::regclass,
                     'u1',
                     1,
                     edge_types := ARRAY['friend'],
                     hydrate := false
                 )
                 WHERE node_id = 'u2'",
                None,
                &[],
            )
            .expect("traverse query failed")
            .first()
            .get::<i64>(1)
            .expect("traverse count read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((
            gql_row
                .get::<i64>(1)
                .expect("gql count read failed")
                .unwrap_or_default(),
            gql_row
                .get::<String>(2)
                .expect("source id read failed")
                .unwrap_or_default(),
            gql_row
                .get::<String>(3)
                .expect("target id read failed")
                .unwrap_or_default(),
            gql_row
                .get::<String>(4)
                .expect("source table read failed")
                .unwrap_or_default(),
            traverse_count,
        ))
    })
    .expect("comparison query failed");

    assert_eq!(gql_count, traverse_count);
    assert_eq!(source_id, "u1");
    assert_eq!(target_id, "u2");
    assert_eq!(source_table, "graph_test_users_pgtest");
}

#[pg_test]
fn gql_statement_memory_limit_is_shared_across_read_phases() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name, age)
         SELECT 'bulk-' || i::text, repeat('x', 512), i
         FROM generate_series(3, 4000) AS i",
    )
    .expect("bulk GQL fixture insert failed");
    build_friendship_fixture_graph();
    Spi::run("SET LOCAL graph.query_memory_mb = 1").expect("set query memory limit failed");

    let statement = "SELECT * FROM graph.gql(
        'MATCH (u:graph_test_users_pgtest) RETURN u',
        NULL::jsonb,
        true
    )";
    assert_eq!(sqlstate_for_error(statement).as_deref(), Some("54000"));
    assert_eq!(
        sql_error_detail(statement).as_deref(),
        Some("pgGraph diagnostic: PG007")
    );

    Spi::run("SET LOCAL graph.query_memory_mb = 64").expect("restore query memory limit failed");
    let count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest) RETURN u',
            NULL::jsonb,
            false
        )",
    )
    .expect("post-rejection GQL query failed")
    .unwrap_or_default();
    assert_eq!(count, 4_000);
}

#[pg_test]
fn gql_relationship_expansion_uses_layered_manifest_snapshot() {
    build_persisted_mutable_friendship_graph();
    publish_friendship_segment_and_reload("f-layered-gql", "u2", "u1");

    let target = Spi::get_one::<String>(
        "SELECT row #>> '{v,_id,id}'
             FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                 RETURN u, v'
             )
             WHERE row #>> '{u,_id,id}' = 'u1'",
    )
    .expect("layered GQL relationship expansion failed")
    .unwrap_or_default();

    assert_eq!(target, "u2");
}

#[pg_test]
fn gql_explain_uses_registered_table_labels_after_catalog_read() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let explain = Spi::get_one::<String>(
        "SELECT graph.gql_explain(
            'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
             RETURN u.id AS source_id, v.id AS target_id'
         )",
    )
    .expect("gql explain failed")
    .unwrap_or_default();

    assert_eq!(
        explain,
        "Expand(source=u:graph_test_users_pgtest, rel=friend, hydration=edge_row, hops=1..1, target=v:graph_test_users_pgtest, return=[source_id, target_id])"
    );
}

#[pg_test]
fn direct_node_lookup_resolves_business_id_without_hydration_scan() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (node_id, node_name, has_node_idx, coordinate_only, missing_count) =
        Spi::connect(|client| {
            let row = client
                .select(
                    "SELECT
                        (SELECT node_id
                         FROM graph.get_node('default', 'graph_test_users_pgtest', 'u1')),
                        (SELECT node->>'name'
                         FROM graph.get_node('default', 'graph_test_users_pgtest', 'u1')),
                        (SELECT node_idx >= 0
                         FROM graph.get_node('default', 'graph_test_users_pgtest', 'u1')),
                        (SELECT node IS NULL
                         FROM graph.get_node(
                            'default',
                            'graph_test_users_pgtest',
                            'u1',
                            hydrate := false
                         )),
                        (SELECT count(*)
                         FROM graph.get_node('default', 'graph_test_users_pgtest', 'missing'))",
                    None,
                    &[],
                )
                .expect("direct lookup query failed")
                .first();
            Ok::<_, pgrx::spi::Error>((
                row.get::<String>(1)
                    .expect("node_id read failed")
                    .unwrap_or_default(),
                row.get::<String>(2)
                    .expect("node name read failed")
                    .unwrap_or_default(),
                row.get::<bool>(3)
                    .expect("node_idx flag read failed")
                    .unwrap_or(false),
                row.get::<bool>(4)
                    .expect("coordinate-only flag read failed")
                    .unwrap_or(false),
                row.get::<i64>(5)
                    .expect("missing count read failed")
                    .unwrap_or_default(),
            ))
        })
        .expect("direct lookup verification failed");

    assert_eq!(node_id, "u1");
    assert_eq!(node_name, "Alice");
    assert!(has_node_idx);
    assert!(coordinate_only);
    assert_eq!(missing_count, 0);
}

#[pg_test]
fn direct_identity_lazy_matches_eager_for_visible_hidden_and_absent_rows() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run(
        "DROP ROLE IF EXISTS graph_direct_lazy_reader;
         CREATE ROLE graph_direct_lazy_reader;
         ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_direct_lazy_policy
             ON public.graph_test_users_pgtest
             USING (id = current_setting('graph.test_visible_id', true));
         GRANT USAGE ON SCHEMA graph, public TO graph_direct_lazy_reader;
         GRANT USAGE ON SCHEMA public TO graph_direct_lazy_reader;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_direct_lazy_reader;
         SET graph.test_visible_id = 'u1'",
    )
    .expect("configure direct lazy RLS fixture failed");
    Spi::run("SET ROLE graph_direct_lazy_reader").expect("set direct lazy reader failed");

    let visible = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
    )
    .expect("visible direct lazy lookup failed")
    .unwrap_or_default();
    let hidden = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u2', hydrate := false)",
    )
    .expect("hidden direct lazy lookup failed")
    .unwrap_or_default();
    let absent = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'missing', hydrate := false)",
    )
    .expect("absent direct lazy lookup failed")
    .unwrap_or_default();
    let eager_visible = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.traverse(
            'graph_test_users_pgtest'::regclass, 'u1',
            max_depth := 1, hydrate := false)
         WHERE depth = 0",
    )
    .expect("eager visibility oracle lookup failed")
    .unwrap_or_default();
    let eager_hidden = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.traverse(
            'graph_test_users_pgtest'::regclass, 'u2',
            max_depth := 1, hydrate := false)
         WHERE depth = 0",
    )
    .expect("eager hidden visibility oracle lookup failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset direct lazy reader failed");

    assert_eq!((visible, hidden, absent), (1, 0, 0));
    assert_eq!((visible, hidden), (eager_visible, eager_hidden));
}

#[pg_test]
fn direct_identity_probe_observes_enclosing_statement_command_changes() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run(
        "CREATE TABLE public.graph_test_visibility_control_pgtest (
             visible_id text PRIMARY KEY
         );
         INSERT INTO public.graph_test_visibility_control_pgtest VALUES ('u1');
         ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_statement_snapshot_policy
             ON public.graph_test_users_pgtest
             USING (id IN (
                 SELECT visible_id
                 FROM public.graph_test_visibility_control_pgtest
             ));
         CREATE FUNCTION public.graph_statement_snapshot_source(candidate text)
             RETURNS bigint
             LANGUAGE sql VOLATILE
             AS $$
                 SELECT count(*)
                 FROM public.graph_test_users_pgtest
                 WHERE id = candidate
             $$;
         DROP ROLE IF EXISTS graph_statement_snapshot_reader;
         CREATE ROLE graph_statement_snapshot_reader;
         GRANT USAGE ON SCHEMA graph, public TO graph_statement_snapshot_reader;
         GRANT SELECT ON public.graph_test_users_pgtest,
             public.graph_test_visibility_control_pgtest
             TO graph_statement_snapshot_reader;
         GRANT UPDATE ON public.graph_test_visibility_control_pgtest
             TO graph_statement_snapshot_reader;
         GRANT EXECUTE ON FUNCTION public.graph_statement_snapshot_source(text)
             TO graph_statement_snapshot_reader;
         SET ROLE graph_statement_snapshot_reader",
    )
    .expect("configure statement-snapshot visibility fixture failed");

    // PostgreSQL makes the statement's own data-modifying CTE visible after
    // its command-counter advance, including to nested SPI policy evaluation.
    let (same_statement, source_oracle) = Spi::connect(|client| {
        let row = client
            .select(
        "WITH changed AS MATERIALIZED (
             UPDATE public.graph_test_visibility_control_pgtest
                SET visible_id = 'u2'
              RETURNING 1 AS marker
         )
         SELECT (SELECT count(*)
                   FROM graph.get_node(
                     'default', 'graph_test_users_pgtest',
                     CASE WHEN changed.marker = 1 THEN 'u1' END,
                     hydrate := false
                   )),
                public.graph_statement_snapshot_source(
                    CASE WHEN changed.marker = 1 THEN 'u1' END
                )
           FROM changed
          ",
                None,
                &[],
            )
            .expect("same-statement visibility command-change lookup failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<i64>(1)?.unwrap_or_default(),
            row.get::<i64>(2)?.unwrap_or_default(),
        ))
    })
    .expect("same-statement visibility command-change result failed");
    let stale = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
    )
    .expect("stale statement-snapshot lookup failed")
    .unwrap_or_default();
    let fresh = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u2', hydrate := false)",
    )
    .expect("fresh statement-snapshot lookup failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset statement-snapshot reader failed");

    assert_eq!(same_statement, source_oracle);
    assert_eq!((same_statement, stale, fresh), (0, 0, 1));
}

#[pg_test]
fn depth_zero_lazy_matches_eager_for_visible_hidden_and_absent_rows() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    create_error_sqlstate_helper();
    Spi::run(
        "DROP ROLE IF EXISTS graph_depth_zero_lazy_reader;
         CREATE ROLE graph_depth_zero_lazy_reader;
         ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_depth_zero_lazy_policy
             ON public.graph_test_users_pgtest USING (id = 'u1');
         GRANT USAGE ON SCHEMA graph, public TO graph_depth_zero_lazy_reader;
         GRANT USAGE ON SCHEMA public TO graph_depth_zero_lazy_reader;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_depth_zero_lazy_reader",
    )
    .expect("configure depth-zero lazy RLS fixture failed");
    Spi::run("SET ROLE graph_depth_zero_lazy_reader").expect("set depth-zero reader failed");

    let visible = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.traverse(
            'graph_test_users_pgtest'::regclass, 'u1', max_depth := 0, hydrate := false)",
    )
    .expect("visible depth-zero traversal failed")
    .unwrap_or_default();
    let hidden = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.traverse(
            'graph_test_users_pgtest'::regclass, 'u2', max_depth := 0, hydrate := false)",
    )
    .expect("hidden depth-zero traversal failed")
    .unwrap_or_default();
    let absent_sqlstate = Spi::get_one::<String>(
        "SELECT public.graph_test_sqlstate(
            $sql$SELECT * FROM graph.traverse(
                'graph_test_users_pgtest'::regclass, 'missing',
                max_depth := 0, hydrate := false)$sql$)",
    )
    .expect("capture absent depth-zero SQLSTATE failed");
    Spi::run("RESET ROLE").expect("reset depth-zero reader failed");

    assert_eq!((visible, hidden), (1, 0));
    assert_eq!(absent_sqlstate.as_deref(), Some("P0002"));
}

#[test]
fn hidden_and_absent_direct_identity_preserve_1_1_diagnostic_distinction() {
    // The 1.1 public contract keeps projection-absent seeds as PG010/P0002,
    // while a projected row hidden by RLS returns no topology row.
    assert_eq!(crate::safety::GraphError::NodeNotFound {
        table: "1".into(),
        pk: "missing".into(),
    }
    .sqlstate(), "P0002");
}

#[cfg(feature = "development")]
#[pg_test]
fn lazy_visibility_cancellation_clears_statement_state() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run(
        "ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_lazy_cancel_policy
             ON public.graph_test_users_pgtest USING (true);
         SELECT graph._test_arm_lazy_visibility_cancel()",
    )
    .expect("arm lazy visibility cancellation failed");

    let cancelled = std::sync::atomic::AtomicBool::new(false);
    pgrx::pg_sys::PgTryBuilder::new(std::panic::AssertUnwindSafe(|| {
        Spi::run(
            "SELECT * FROM graph.get_node(
                'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
        )
        .expect("armed lazy lookup unexpectedly returned an SPI error");
    }))
    .catch_when(
        pgrx::PgSqlErrorCode::ERRCODE_QUERY_CANCELED,
        |_| cancelled.store(true, std::sync::atomic::Ordering::Relaxed),
    )
    .execute();
    let state_empty = Spi::get_one::<bool>(
        "SELECT graph._test_visibility_resolution_state_empty()",
    )
    .expect("read lazy visibility state failed")
    .unwrap_or(false);
    let retry_count = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
    )
    .expect("retry direct lookup after cancellation failed")
    .unwrap_or_default();

    assert!(cancelled.load(std::sync::atomic::Ordering::Relaxed));
    assert!(state_empty);
    assert_eq!(retry_count, 1);
}

#[pg_test]
fn recursive_rls_policy_graph_lookup_fails_closed_and_cleans_up() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run(
        "DROP ROLE IF EXISTS graph_recursive_visibility_reader;
         CREATE ROLE graph_recursive_visibility_reader;
         ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE FUNCTION public.graph_recursive_visibility_policy()
             RETURNS boolean
             LANGUAGE sql STABLE
             AS $$
                 SELECT EXISTS (
                     SELECT 1 FROM graph.get_node(
                         'default', 'graph_test_users_pgtest', 'u1', hydrate := false)
                 )
             $$;
         CREATE POLICY graph_recursive_visibility_policy
             ON public.graph_test_users_pgtest
             USING (public.graph_recursive_visibility_policy());
         GRANT USAGE ON SCHEMA graph, public TO graph_recursive_visibility_reader;
         GRANT SELECT ON public.graph_test_users_pgtest
             TO graph_recursive_visibility_reader;
         GRANT EXECUTE ON FUNCTION public.graph_recursive_visibility_policy()
             TO graph_recursive_visibility_reader;
         SET ROLE graph_recursive_visibility_reader",
    )
    .expect("configure recursive RLS policy failed");

    let rejected = std::sync::atomic::AtomicBool::new(false);
    pgrx::pg_sys::PgTryBuilder::new(std::panic::AssertUnwindSafe(|| {
        Spi::run(
            "SELECT * FROM graph.get_node(
                'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
        )
        .expect("recursive visibility query unexpectedly returned an SPI error");
    }))
    .catch_when(
        pgrx::PgSqlErrorCode::ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE,
        |_| rejected.store(true, std::sync::atomic::Ordering::Relaxed),
    )
    .execute();
    Spi::run("RESET ROLE").expect("reset recursive visibility reader failed");
    let detail = sql_error_detail(
        "SET ROLE graph_recursive_visibility_reader;
         SELECT * FROM graph.get_node(
             'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
    );
    Spi::run(
        "RESET ROLE;
         DROP POLICY graph_recursive_visibility_policy
             ON public.graph_test_users_pgtest;
         CREATE POLICY graph_recursive_visibility_policy
             ON public.graph_test_users_pgtest USING (true)",
    )
    .expect("replace recursive visibility policy failed");
    #[cfg(feature = "development")]
    let state_empty = Spi::get_one::<bool>(
        "SELECT graph._test_visibility_resolution_state_empty()",
    )
    .expect("read recursive visibility state failed")
    .unwrap_or(false);
    let retry = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
    )
    .expect("retry after recursive policy failed")
    .unwrap_or_default();
    assert!(rejected.load(std::sync::atomic::Ordering::Relaxed));
    assert_eq!(detail.as_deref(), Some("pgGraph diagnostic: PG024"));
    #[cfg(feature = "development")]
    assert!(state_empty);
    assert_eq!(retry, 1);
}

#[pg_test]
fn direct_identity_lazy_uses_effective_security_definer_role() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run(
        "DROP ROLE IF EXISTS graph_lazy_wrapper_owner;
         CREATE ROLE graph_lazy_wrapper_owner;
         ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_lazy_wrapper_policy
             ON public.graph_test_users_pgtest USING (id = 'u1');
         GRANT USAGE ON SCHEMA graph, public TO graph_lazy_wrapper_owner;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_lazy_wrapper_owner;
         CREATE FUNCTION public.graph_lazy_wrapper_lookup(candidate text)
             RETURNS bigint
             LANGUAGE sql
             SECURITY DEFINER
             SET search_path = pg_catalog, pg_temp
             AS $$
                 SELECT count(*) FROM graph.get_node(
                     'default', 'graph_test_users_pgtest', candidate, hydrate := false)
             $$;
         ALTER FUNCTION public.graph_lazy_wrapper_lookup(text)
             OWNER TO graph_lazy_wrapper_owner",
    )
    .expect("configure security-definer lazy visibility fixture failed");

    let visible = Spi::get_one_with_args::<i64>(
        "SELECT public.graph_lazy_wrapper_lookup($1)",
        &["u1".into()],
    )
    .expect("visible wrapper lookup failed")
    .unwrap_or_default();
    let hidden = Spi::get_one_with_args::<i64>(
        "SELECT public.graph_lazy_wrapper_lookup($1)",
        &["u2".into()],
    )
    .expect("hidden wrapper lookup failed")
    .unwrap_or_default();

    assert_eq!((visible, hidden), (1, 0));
}

#[pg_test]
fn direct_identity_lazy_supports_composite_keys_without_text_casting_columns() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
             'graph_test_composite_pgtest'::regclass,
             id_columns := ARRAY['org_id', 'user_id'],
             columns := ARRAY['label']
         );
         SELECT * FROM graph.build();
         DROP ROLE IF EXISTS graph_composite_lazy_reader;
         CREATE ROLE graph_composite_lazy_reader;
         ALTER TABLE public.graph_test_composite_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_composite_lazy_policy
             ON public.graph_test_composite_pgtest
             USING (user_id = 'emp1');
         GRANT USAGE ON SCHEMA graph, public TO graph_composite_lazy_reader;
         GRANT SELECT ON public.graph_test_composite_pgtest TO graph_composite_lazy_reader;
         SET ROLE graph_composite_lazy_reader",
    )
    .expect("configure composite lazy visibility fixture failed");

    let visible = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_composite_pgtest', '[\"org1\", \"emp1\"]',
            hydrate := false)",
    )
    .expect("visible composite lookup failed")
    .unwrap_or_default();
    let hidden = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_composite_pgtest', '[\"org1\", \"emp2\"]',
            hydrate := false)",
    )
    .expect("hidden composite lookup failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset composite reader failed");

    assert_eq!((visible, hidden), (1, 0));
}

#[pg_test]
fn direct_identity_probe_plan_retains_primary_key_index_lookup() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    let table_oid = Spi::get_one::<pgrx::pg_sys::Oid>(
        "SELECT 'public.graph_test_users_pgtest'::regclass::oid",
    )
    .expect("read direct visibility table OID failed")
    .expect("direct visibility table OID was NULL")
    .to_u32();
    let (tables, _, _) = crate::catalog::read_catalog().expect("read graph catalog failed");
    let (payload, query) =
        crate::sql_visibility::_test_build_node_probe_query(table_oid, &tables, "u1")
            .expect("build production direct visibility probe failed");
    let explain = format!("EXPLAIN (FORMAT JSON, COSTS OFF) {query}");
    let plan = Spi::connect(|client| {
        client
            .select(&explain, None, &[payload.into()])?
            .first()
            .get::<pgrx::Json>(1)
    })
    .expect("explain production direct visibility probe failed")
    .expect("direct visibility EXPLAIN returned no plan");
    let rendered = plan.0.to_string();
    assert!(
        rendered.contains("Index Only Scan") || rendered.contains("Index Scan"),
        "direct visibility probe did not retain the primary-key index: {rendered}"
    );
}

#[pg_test]
fn direct_identity_no_rls_rejects_a_stale_projected_source_row() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run(
        "DELETE FROM public.graph_test_friendships_pgtest
          WHERE user_id = 'u1' OR friend_id = 'u1';
         DELETE FROM public.graph_test_users_pgtest WHERE id = 'u1'",
    )
        .expect("delete source row without syncing projection failed");
    let count = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
    )
    .expect("stale projected lookup failed")
    .unwrap_or_default();
    assert_eq!(count, 0);
}

#[pg_test]
fn direct_identity_legacy_bypass_still_checks_source_existence() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run(
        "DELETE FROM public.graph_test_friendships_pgtest
          WHERE user_id = 'u1' OR friend_id = 'u1';
         DELETE FROM public.graph_test_users_pgtest WHERE id = 'u1';
         SET LOCAL graph.rls_mode = 'legacy_bypass'",
    )
    .expect("configure stale legacy-bypass fixture failed");
    let count = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
    )
    .expect("legacy-bypass stale projected lookup failed")
    .unwrap_or_default();
    assert_eq!(count, 0);
}

#[pg_test]
fn direct_identity_probe_memory_limit_fails_before_policy_spi() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest(id, name, age)
         VALUES (repeat('k', 200000), 'large key', 1)",
    )
    .expect("insert large direct-identity key failed");
    build_friendship_fixture_graph();
    Spi::run(
        "ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_lazy_memory_policy
             ON public.graph_test_users_pgtest USING (true);
         SET LOCAL graph.query_memory_mb = 1",
    )
    .expect("configure lazy probe memory fixture failed");
    let sqlstate = sqlstate_for_error(
        "SELECT * FROM graph.get_node(
            'default', 'graph_test_users_pgtest', repeat('k', 200000), hydrate := false)",
    );
    assert_eq!(sqlstate.as_deref(), Some("54000"));
}

#[pg_test]
fn direct_identity_rls_fails_closed_for_guc_dependent_key_text() {
    reset_and_create_fixtures();
    create_error_sqlstate_helper();
    Spi::run(
        "DROP TABLE IF EXISTS public.graph_test_date_identity_pgtest CASCADE;
         CREATE TABLE public.graph_test_date_identity_pgtest (
             id date PRIMARY KEY,
             name text NOT NULL
         );
         SET LOCAL datestyle = 'SQL, MDY';
         INSERT INTO public.graph_test_date_identity_pgtest VALUES
             (DATE '2020-01-02', 'January'),
             (DATE '2020-02-01', 'February');
         SELECT graph.add_table(
             'graph_test_date_identity_pgtest'::regclass,
             id_column := 'id', columns := ARRAY['name']
         );
         SELECT * FROM graph.build();
         ALTER TABLE public.graph_test_date_identity_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_date_identity_policy
             ON public.graph_test_date_identity_pgtest
             USING (id = DATE '2020-02-01');
         DROP ROLE IF EXISTS graph_date_identity_reader;
         CREATE ROLE graph_date_identity_reader;
         GRANT USAGE ON SCHEMA graph, public TO graph_date_identity_reader;
         GRANT USAGE ON SCHEMA public TO graph_date_identity_reader;
         GRANT SELECT ON public.graph_test_date_identity_pgtest
             TO graph_date_identity_reader;
         SET LOCAL datestyle = 'ISO, DMY';
         SET ROLE graph_date_identity_reader",
    )
    .expect("configure GUC-dependent identity fixture failed");

    // This projected key meant January 2 at build time, but parsing it under
    // DMY would mean February 1 and could authorize the wrong row. The direct
    // resolver must not attempt that cast.
    let sqlstate = Spi::get_one::<String>(
        "SELECT public.graph_test_sqlstate(
            $sql$SELECT * FROM graph.get_node(
                'default', 'graph_test_date_identity_pgtest', '01/02/2020',
                hydrate := false)$sql$)",
    )
    .expect("capture unsupported date identity SQLSTATE failed");
    let eager_sqlstate = Spi::get_one::<String>(
        "SELECT public.graph_test_sqlstate(
            $sql$SELECT * FROM graph.traverse(
                'graph_test_date_identity_pgtest'::regclass, '01/02/2020',
                max_depth := 1, hydrate := false)$sql$)",
    )
    .expect("capture eager unsupported date identity SQLSTATE failed");
    Spi::run("RESET ROLE").expect("reset date identity reader failed");
    assert_eq!(sqlstate.as_deref(), Some("0A000"));
    assert_eq!(eager_sqlstate.as_deref(), Some("0A000"));
}

#[pg_test]
fn eager_relationship_rls_fails_closed_for_guc_dependent_identity_text() {
    reset_and_create_fixtures();
    create_error_sqlstate_helper();
    Spi::run(
        "DROP TABLE IF EXISTS public.graph_test_date_edges_pgtest CASCADE;
         CREATE TABLE public.graph_test_date_edges_pgtest (
             edge_date date PRIMARY KEY,
             source_id text NOT NULL REFERENCES public.graph_test_users_pgtest(id),
             target_id text NOT NULL REFERENCES public.graph_test_users_pgtest(id)
         );
         SET LOCAL datestyle = 'SQL, MDY';
         INSERT INTO public.graph_test_date_edges_pgtest VALUES
             (DATE '2020-01-02', 'u1', 'u2'),
             (DATE '2020-02-01', 'u2', 'u1');
         SELECT graph.add_table(
             'graph_test_users_pgtest'::regclass,
             id_column := 'id', columns := ARRAY['name', 'age']
         );
         SELECT graph.add_edge(
             'graph_test_date_edges_pgtest'::regclass,
             'source_id', 'graph_test_users_pgtest'::regclass,
             'target_id', 'dated'
         );
         SELECT * FROM graph.build();
         ALTER TABLE public.graph_test_date_edges_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_date_edge_policy
             ON public.graph_test_date_edges_pgtest
             USING (edge_date = DATE '2020-02-01');
         DROP ROLE IF EXISTS graph_date_edge_reader;
         CREATE ROLE graph_date_edge_reader;
         GRANT USAGE ON SCHEMA graph, public TO graph_date_edge_reader;
         GRANT SELECT ON public.graph_test_users_pgtest,
             public.graph_test_date_edges_pgtest TO graph_date_edge_reader;
         SET LOCAL datestyle = 'ISO, DMY';
         SET ROLE graph_date_edge_reader",
    )
    .expect("configure GUC-dependent relationship identity fixture failed");

    let sqlstate = Spi::get_one::<String>(
        "SELECT public.graph_test_sqlstate(
            $sql$SELECT * FROM graph.traverse(
                'graph_test_users_pgtest'::regclass, 'u1',
                max_depth := 1, hydrate := false)$sql$)",
    )
    .expect("capture unsupported relationship identity SQLSTATE failed");
    Spi::run("RESET ROLE").expect("reset date edge reader failed");

    assert_eq!(sqlstate.as_deref(), Some("0A000"));
}

#[pg_test]
fn eager_visibility_checks_outer_acl_before_unstable_identity_details() {
    reset_and_create_fixtures();
    create_error_sqlstate_helper();
    Spi::run(
        "DROP TABLE IF EXISTS public.graph_test_acl_date_identity_pgtest CASCADE;
         CREATE TABLE public.graph_test_acl_date_identity_pgtest (
             id date PRIMARY KEY,
             name text NOT NULL
         );
         INSERT INTO public.graph_test_acl_date_identity_pgtest
         VALUES (DATE '2020-01-02', 'hidden');
         SELECT graph.add_table(
             'graph_test_acl_date_identity_pgtest'::regclass,
             id_column := 'id', columns := ARRAY['name']
         );
         SELECT * FROM graph.build();
         ALTER TABLE public.graph_test_acl_date_identity_pgtest
             ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_acl_date_identity_policy
             ON public.graph_test_acl_date_identity_pgtest USING (true);
         DROP ROLE IF EXISTS graph_acl_date_identity_reader;
         CREATE ROLE graph_acl_date_identity_reader;
         GRANT USAGE ON SCHEMA graph, public TO graph_acl_date_identity_reader;
         SET ROLE graph_acl_date_identity_reader",
    )
    .expect("configure ACL-before-identity fixture failed");

    let sqlstate = Spi::get_one::<String>(
        "SELECT public.graph_test_sqlstate(
            $sql$SELECT * FROM graph.traverse(
                'graph_test_acl_date_identity_pgtest'::regclass,
                '2020-01-02', max_depth := 1, hydrate := false)$sql$)",
    )
    .expect("capture ACL-before-identity SQLSTATE failed");
    Spi::run("RESET ROLE").expect("reset ACL-before-identity reader failed");

    assert_eq!(sqlstate.as_deref(), Some("42501"));
}

#[pg_test]
fn direct_identity_lazy_observes_transaction_delta_and_savepoint_rollback() {
    reset_and_create_fixtures();
    Spi::run(
        "SET graph.mutable_enabled = on;
         SELECT graph.add_table(
             'graph_test_users_pgtest'::regclass,
             id_column := 'id', columns := ARRAY['name', 'age']
         );
         SELECT * FROM graph.build(mode := 'mutable_overlay');
         SELECT * FROM graph.gql(
             'CREATE (u:graph_test_users_pgtest {id: ''u3'', name: ''Cara'', age: 29}) RETURN u'
         );
         DO $block$
         BEGIN
             PERFORM * FROM graph.gql(
                 'CREATE (u:graph_test_users_pgtest {id: ''u4'', name: ''Drew'', age: 31}) RETURN u'
             );
             RAISE EXCEPTION 'rollback nested write';
         EXCEPTION WHEN others THEN
             NULL;
         END
         $block$;
         ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_lazy_tx_policy ON public.graph_test_users_pgtest
             USING (id IN ('u3', 'u4'));
         DROP ROLE IF EXISTS graph_lazy_tx_reader;
         CREATE ROLE graph_lazy_tx_reader;
         GRANT USAGE ON SCHEMA graph, public TO graph_lazy_tx_reader;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_lazy_tx_reader;
         SET ROLE graph_lazy_tx_reader",
    )
    .expect("configure transaction-local lazy identity fixture failed");

    let committed_in_transaction = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u3', hydrate := false)",
    )
    .expect("transaction-local direct identity failed")
    .unwrap_or_default();
    let rolled_back = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u4', hydrate := false)",
    )
    .expect("rolled-back direct identity lookup failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset transaction-local lazy reader failed");

    assert_eq!((committed_in_transaction, rolled_back), (1, 0));
}

#[cfg(feature = "development")]
#[pg_test]
fn lazy_visibility_policy_error_propagates_without_caching_or_stranding_state() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run(
        "ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE FUNCTION public.graph_lazy_error_policy()
             RETURNS boolean LANGUAGE plpgsql VOLATILE AS $$
             BEGIN
                 PERFORM 1 / 0;
                 RETURN true;
             END
             $$;
         CREATE POLICY graph_lazy_error_policy
             ON public.graph_test_users_pgtest
             USING (public.graph_lazy_error_policy());
         DROP ROLE IF EXISTS graph_lazy_error_reader;
         CREATE ROLE graph_lazy_error_reader;
         GRANT USAGE ON SCHEMA graph, public TO graph_lazy_error_reader;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_lazy_error_reader;
         GRANT EXECUTE ON FUNCTION public.graph_lazy_error_policy()
             TO graph_lazy_error_reader;
         SET ROLE graph_lazy_error_reader",
    )
    .expect("configure policy-error fixture failed");

    let rejected = std::sync::atomic::AtomicBool::new(false);
    pgrx::pg_sys::PgTryBuilder::new(std::panic::AssertUnwindSafe(|| {
        Spi::run(
            "SELECT * FROM graph.get_node(
                'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
        )
        .expect("errored policy unexpectedly returned an SPI error");
    }))
    .catch_when(
        pgrx::PgSqlErrorCode::ERRCODE_DIVISION_BY_ZERO,
        |_| rejected.store(true, std::sync::atomic::Ordering::Relaxed),
    )
    .execute();
    Spi::run(
        "RESET ROLE;
         DROP POLICY graph_lazy_error_policy ON public.graph_test_users_pgtest;
         CREATE POLICY graph_lazy_error_policy
             ON public.graph_test_users_pgtest USING (true);
         SET ROLE graph_lazy_error_reader",
    )
    .expect("replace errored policy failed");
    let state_empty = Spi::get_one::<bool>(
        "SELECT graph._test_visibility_resolution_state_empty()",
    )
    .expect("read post-error visibility state failed")
    .unwrap_or(false);
    let retry = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
    )
    .expect("retry after policy error failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset policy-error reader failed");

    assert!(rejected.load(std::sync::atomic::Ordering::Relaxed));
    assert!(state_empty);
    assert_eq!(retry, 1);
}

#[cfg(feature = "development")]
#[pg_test]
fn lazy_visibility_policy_cancellation_cleans_state_during_spi() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run(
        "ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE FUNCTION public.graph_lazy_cancel_policy()
             RETURNS boolean LANGUAGE plpgsql VOLATILE AS $$
             BEGIN
                 RAISE SQLSTATE '57014' USING MESSAGE = 'cancel inside RLS policy';
             END
             $$;
         CREATE POLICY graph_lazy_cancel_policy
             ON public.graph_test_users_pgtest
             USING (public.graph_lazy_cancel_policy());
         DROP ROLE IF EXISTS graph_lazy_timeout_reader;
         CREATE ROLE graph_lazy_timeout_reader;
         GRANT USAGE ON SCHEMA graph, public TO graph_lazy_timeout_reader;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_lazy_timeout_reader;
         SET ROLE graph_lazy_timeout_reader",
    )
    .expect("configure policy-timeout fixture failed");

    let cancelled = std::sync::atomic::AtomicBool::new(false);
    pgrx::pg_sys::PgTryBuilder::new(std::panic::AssertUnwindSafe(|| {
        Spi::run(
            "SELECT * FROM graph.get_node(
                'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
        )
        .expect("policy cancellation unexpectedly returned an SPI error");
    }))
    .catch_when(
        pgrx::PgSqlErrorCode::ERRCODE_QUERY_CANCELED,
        |_| cancelled.store(true, std::sync::atomic::Ordering::Relaxed),
    )
    .execute();
    Spi::run(
        "RESET ROLE;
         DROP POLICY graph_lazy_cancel_policy ON public.graph_test_users_pgtest;
         CREATE POLICY graph_lazy_cancel_policy
             ON public.graph_test_users_pgtest USING (true);
         SET ROLE graph_lazy_timeout_reader",
    )
    .expect("replace timed-out policy failed");
    let state_empty = Spi::get_one::<bool>(
        "SELECT graph._test_visibility_resolution_state_empty()",
    )
    .expect("read post-timeout visibility state failed")
    .unwrap_or(false);
    let retry = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.get_node(
            'default', 'graph_test_users_pgtest', 'u1', hydrate := false)",
    )
    .expect("retry after policy timeout failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset policy-timeout reader failed");

    assert!(cancelled.load(std::sync::atomic::Ordering::Relaxed));
    assert!(state_empty);
    assert_eq!(retry, 1);
}

#[pg_test]
fn direct_neighbor_lookup_expands_after_business_id_resolution() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (root_id, neighbor_id, depth, coordinate_only) = Spi::connect(|client| {
        let row = client
            .select(
                "SELECT root_id, node_id, depth, node IS NULL
                 FROM graph.get_neighbors(
                    'default',
                    'graph_test_users_pgtest',
                    'u1',
                    direction := 'out',
                    edge_types := ARRAY['friend'],
                    hydrate := false
                 )",
                None,
                &[],
            )
            .expect("direct neighbor query failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<String>(1)
                .expect("root id read failed")
                .unwrap_or_default(),
            row.get::<String>(2)
                .expect("neighbor id read failed")
                .unwrap_or_default(),
            row.get::<i32>(3).expect("depth read failed").unwrap_or_default(),
            row.get::<bool>(4)
                .expect("coordinate-only flag read failed")
                .unwrap_or(false),
        ))
    })
    .expect("direct neighbor verification failed");

    assert_eq!(root_id, "u1");
    assert_eq!(neighbor_id, "u2");
    assert_eq!(depth, 1);
    assert!(coordinate_only);
}

#[pg_test]
fn gql_identity_predicates_lower_to_node_lookup() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let explain = Spi::get_one::<String>(
        "SELECT graph.gql_explain(
            'MATCH (u:graph_test_users_pgtest) WHERE id(u) = ''u1'' RETURN u'
         )",
    )
    .expect("identity explain failed")
    .unwrap_or_default();
    let property_explain = Spi::get_one::<String>(
        "SELECT graph.gql_explain(
            'MATCH (u:graph_test_users_pgtest) WHERE u.id = ''u1'' RETURN u'
         )",
    )
    .expect("property identity explain failed")
    .unwrap_or_default();
    let node_id = Spi::get_one::<String>(
        "SELECT row #>> '{u,_id,id}'
         FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest) WHERE id(u) = $id RETURN u',
            params := '{\"id\":\"u1\"}'::jsonb,
            hydrate := false
         )",
    )
    .expect("identity gql execution failed")
    .unwrap_or_default();

    assert_eq!(
        explain,
        "NodeLookup(node=u:graph_test_users_pgtest, lookup=identity, return=[u])"
    );
    assert_eq!(
        property_explain,
        "NodeLookup(node=u:graph_test_users_pgtest, lookup=identity, return=[u])"
    );
    assert_eq!(node_id, "u1");
}

#[pg_test]
fn gql_binds_dynamic_edge_labels_from_registered_label_column() {
    reset_and_create_fixtures();
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest
         ADD COLUMN rel_type text NOT NULL DEFAULT 'colleague'",
    )
    .expect("add dynamic relationship label column failed");
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name, age)
         VALUES ('u3', 'Cara', 29);
         INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id, rel_type)
         VALUES ('f2', 'u1', 'u3', 'acquaintance')",
    )
    .expect("add second dynamic relationship spelling failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                from_table := 'graph_test_friendships_pgtest'::regclass,
                from_column := 'user_id',
                to_table := 'graph_test_users_pgtest'::regclass,
                to_column := 'friend_id',
                label := 'related_to',
                bidirectional := false,
                label_column := 'rel_type'
            )",
    )
    .expect("add dynamic friendship edge failed");
    Spi::run("SELECT * FROM graph.build()").expect("build dynamic relationship graph failed");

    let gql_target = Spi::get_one::<String>(
        "SELECT row #>> '{target,id}'
         FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest)-[:colleague]->(v:graph_test_users_pgtest)
             RETURN v AS target'
         )",
    )
    .expect("dynamic GQL relationship query failed")
    .expect("dynamic GQL relationship query returned no row");
    let cypher_target = Spi::get_one::<String>(
        "SELECT row #>> '{target,id}'
         FROM graph.cypher(
            'MATCH (u:graph_test_users_pgtest)-[:acquaintance]->(v:graph_test_users_pgtest)
             RETURN v AS target'
         )",
    )
    .expect("dynamic Cypher relationship query failed")
    .expect("dynamic Cypher relationship query returned no row");
    let absent_count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
         FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest)-[:not_loaded]->(v:graph_test_users_pgtest)
             RETURN v AS target'
         )",
    )
    .expect("absent dynamic relationship query failed")
    .unwrap_or_default();
    let wildcard_target = Spi::get_one::<String>(
        "SELECT COALESCE(
                    row #>> '{p,_path,nodes,1,_id,id}',
                    row #>> '{p,_path,nodes,1,id}'
                )
         FROM graph.gql(
            'MATCH p=(u:graph_test_users_pgtest)-[:acquaintance]->(v:graph_test_users_pgtest)
             RETURN p'
         )",
    )
    .expect("dynamic wildcard path query failed")
    .expect("dynamic wildcard path query returned no row");
    let absent_wildcard_count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
         FROM graph.gql(
            'MATCH p=(u:graph_test_users_pgtest)-[:not_loaded]->(v:graph_test_users_pgtest)
             RETURN p'
         )",
    )
    .expect("absent dynamic wildcard query failed")
    .unwrap_or_default();
    let join_count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
         FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest)-[:colleague]->(v:graph_test_users_pgtest),
                   (u)-[:acquaintance]->(w:graph_test_users_pgtest)
             RETURN v, w'
         )",
    )
    .expect("dynamic multi-pattern query failed")
    .unwrap_or_default();

    assert_eq!(gql_target, "u2");
    assert_eq!(cypher_target, "u3");
    assert_eq!(absent_count, 0);
    assert_eq!(wildcard_target, "u3");
    assert_eq!(absent_wildcard_count, 0);
    assert_eq!(join_count, 1);
}

#[pg_test]
fn gql_rejects_ambiguous_open_dynamic_relationship_mappings() {
    reset_and_create_fixtures();
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest
         ADD COLUMN rel_type text NOT NULL DEFAULT 'first_fallback';
         CREATE TABLE public.graph_test_other_dynamic_edges_pgtest (
            id text PRIMARY KEY,
            user_id text NOT NULL REFERENCES public.graph_test_users_pgtest(id),
            friend_id text NOT NULL REFERENCES public.graph_test_users_pgtest(id),
            rel_type text NOT NULL DEFAULT 'second_fallback'
         );
         SELECT graph.add_table(
            'graph_test_users_pgtest'::regclass,
            id_column := 'id', columns := ARRAY['name', 'age']);
         SELECT graph.add_edge(
            from_table := 'graph_test_friendships_pgtest'::regclass,
            from_column := 'user_id', to_table := 'graph_test_users_pgtest'::regclass,
            to_column := 'friend_id', label := 'first_fallback', bidirectional := false,
            label_column := 'rel_type');
         SELECT graph.add_edge(
            from_table := 'graph_test_other_dynamic_edges_pgtest'::regclass,
            from_column := 'user_id', to_table := 'graph_test_users_pgtest'::regclass,
            to_column := 'friend_id', label := 'second_fallback', bidirectional := false,
            label_column := 'rel_type');
         SELECT * FROM graph.build()",
    )
    .expect("build ambiguous dynamic mapping fixture failed");
    create_error_sqlstate_helper();

    let sqlstate = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sqlstate({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest)-[:runtime_type]->(v:graph_test_users_pgtest)
                 RETURN u, v',
                hydrate := false
             )"
        )
    ))
    .expect("capture ambiguous dynamic mapping SQLSTATE failed");

    assert_eq!(sqlstate.as_deref(), Some("22023"));
}

#[pg_test]
fn gql_dynamic_hidden_and_absent_types_have_eager_lazy_parity() {
    reset_and_create_fixtures();
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest
         ADD COLUMN rel_type text NOT NULL DEFAULT 'colleague';
         SELECT graph.add_table(
            'graph_test_users_pgtest'::regclass,
            id_column := 'id', columns := ARRAY['name', 'age']);
         SELECT graph.add_edge(
            from_table := 'graph_test_friendships_pgtest'::regclass,
            from_column := 'user_id', to_table := 'graph_test_users_pgtest'::regclass,
            to_column := 'friend_id', label := 'related_to', bidirectional := false,
            label_column := 'rel_type');
         SELECT * FROM graph.build();
         DROP ROLE IF EXISTS graph_gql_dynamic_hidden_reader;
         CREATE ROLE graph_gql_dynamic_hidden_reader;
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_dynamic_hidden_reader;
         GRANT SELECT ON public.graph_test_users_pgtest,
                         public.graph_test_friendships_pgtest
           TO graph_gql_dynamic_hidden_reader;
         ALTER TABLE public.graph_test_friendships_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_gql_dynamic_hidden_none
           ON public.graph_test_friendships_pgtest FOR SELECT
           TO graph_gql_dynamic_hidden_reader USING (false);
         SET ROLE graph_gql_dynamic_hidden_reader",
    )
    .expect("create dynamic hidden/absent fixture failed");

    let query_count = |rel_type: &str| {
        Spi::get_one::<i64>(&format!(
            "SELECT count(*)::bigint FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest)-[:{rel_type}]->(v:graph_test_users_pgtest)
                 WHERE id(u) = ''u1'' RETURN v',
                hydrate := false
             )"
        ))
        .expect("dynamic hidden/absent query failed")
        .unwrap_or_default()
    };
    let hidden_auto = query_count("colleague");
    let absent_auto = query_count("not_loaded");
    let optional_absent = || {
        Spi::get_one::<i64>(
            "SELECT count(*)::bigint FROM graph.gql(
                'OPTIONAL MATCH (u:graph_test_users_pgtest)-[:not_loaded]->(v:graph_test_users_pgtest)
                 WHERE id(u) = ''u1'' RETURN u, v',
                hydrate := false
             )",
        )
        .expect("optional absent dynamic query failed")
        .unwrap_or_default()
    };
    let optional_absent_auto = optional_absent();
    Spi::run("SELECT graph._test_set_visibility_strategy('eager')")
        .expect("force eager dynamic visibility failed");
    let hidden_eager = query_count("colleague");
    let absent_eager = query_count("not_loaded");
    let optional_absent_eager = optional_absent();
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')")
        .expect("restore dynamic visibility fixture failed");

    assert_eq!(
        (hidden_auto, absent_auto, hidden_eager, absent_eager),
        (0, 0, 0, 0)
    );
    assert_eq!((optional_absent_auto, optional_absent_eager), (2, 2));
}

#[pg_test]
fn gql_relationship_variables_hydrate_registered_edge_rows() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (edge_id, source_id, coordinate_only_has_id) = Spi::connect(|client| {
        let row = client
            .select(
                "SELECT
                    (SELECT row #>> '{r,id}'
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest) RETURN r'
                     )
                     LIMIT 1),
                    (SELECT row #>> '{r,user_id}'
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest) RETURN r'
                     )
                     LIMIT 1),
                    (SELECT row->'r' ? 'id'
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest) RETURN r',
                        hydrate := false
                     )
                     LIMIT 1)",
                None,
                &[],
            )
            .expect("relationship hydration query failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<String>(1)
                .expect("edge id read failed")
                .unwrap_or_default(),
            row.get::<String>(2)
                .expect("source edge column read failed")
                .unwrap_or_default(),
            row.get::<bool>(3)
                .expect("coordinate-only relationship flag read failed")
                .unwrap_or(true),
        ))
    })
    .expect("relationship hydration verification failed");

    assert_eq!(edge_id, "f1");
    assert_eq!(source_id, "u1");
    assert!(!coordinate_only_has_id);
}

#[pg_test]
fn gql_wildcard_planner_hostile_forms_have_stable_sqlstate() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    for statement in [
        "SELECT * FROM graph.gql(
            'MATCH p=()-[]->() RETURN DISTINCT p'
         )",
        "SELECT * FROM graph.gql(
            'MATCH p=()-[]->() RETURN p ORDER BY p'
         )",
        "SELECT * FROM graph.gql(
            'MATCH p=()-[]->() WITH p RETURN p'
         )",
        "SELECT * FROM graph.gql(
            'MATCH p=()-[]->() RETURN count(*) AS paths'
         )",
        "SELECT * FROM graph.gql(
            'OPTIONAL MATCH p=()-[]->() RETURN p'
         )",
    ] {
        assert_eq!(
            sqlstate_for_error(statement).as_deref(),
            Some("0A000"),
            "unexpected SQLSTATE for {statement}"
        );
    }
}

#[pg_test]
fn gql_node_only_optional_match_null_extends_missing_rows() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (row_count, node_is_null, property_is_null) = Spi::connect(|client| {
        let row = client
            .select(
                "SELECT count(*),
                        bool_and(row->'u' = 'null'::jsonb),
                        bool_and(row->'name' = 'null'::jsonb)
                 FROM graph.gql(
                    'OPTIONAL MATCH (u:graph_test_users_pgtest)
                     WHERE id(u) = ''missing''
                     RETURN u, u.name AS name'
                 )",
                None,
                &[],
            )
            .expect("optional node query failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<i64>(1)
                .expect("optional row count read failed")
                .unwrap_or_default(),
            row.get::<bool>(2)
                .expect("optional node null read failed")
                .unwrap_or(false),
            row.get::<bool>(3)
                .expect("optional property null read failed")
                .unwrap_or(false),
        ))
    })
    .expect("optional node verification failed");

    assert_eq!(row_count, 1);
    assert!(node_is_null);
    assert!(property_is_null);
}

#[pg_test]
fn gql_defaults_to_hydrated_nodes_and_projects_ids_explicitly() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (default_has_name, explicit_has_name, coordinate_has_name, scalar_id) =
        Spi::connect(|client| {
        let row = client
            .select(
                "SELECT
                    (SELECT row->'u' ? 'name'
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u1''}) RETURN u'
                     )
                     LIMIT 1),
                    (SELECT row->'u' ? 'name'
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u1''}) RETURN u',
                        hydrate := true
                     )
                     LIMIT 1),
                    (SELECT row->'u' ? 'name'
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u1''}) RETURN u',
                        hydrate := false
                     )
                     LIMIT 1),
                    (SELECT row #>> '{id}'
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u1''}) RETURN u.id AS id'
                     )
                     LIMIT 1)",
                None,
                &[],
            )
            .expect("gql hydrate behavior query failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<bool>(1)
                .expect("default hydrate flag read failed")
                .unwrap_or(false),
            row.get::<bool>(2)
                .expect("explicit hydrate flag read failed")
                .unwrap_or(false),
            row.get::<bool>(3)
                .expect("coordinate hydrate flag read failed")
                .unwrap_or(true),
            row.get::<String>(4)
                .expect("scalar id read failed")
                .unwrap_or_default(),
        ))
    })
    .expect("gql hydrate behavior verification failed");

    assert!(default_has_name);
    assert_eq!(explicit_has_name, default_has_name);
    assert!(!coordinate_has_name);
    assert_eq!(scalar_id, "u1");
}

#[pg_test]
fn gql_hydration_fails_closed_when_source_row_is_not_visible() {
    reset_and_create_fixtures();
    Spi::run("DROP ROLE IF EXISTS graph_gql_hydration_rls").expect("drop role failed");
    Spi::run("CREATE ROLE graph_gql_hydration_rls").expect("create role failed");
    Spi::run("DROP TABLE IF EXISTS public.graph_gql_hydration_rls_pgtest CASCADE")
        .expect("drop hydration rls table failed");
    Spi::run(
        "CREATE TABLE public.graph_gql_hydration_rls_pgtest (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL
            )",
    )
    .expect("create hydration rls table failed");
    Spi::run(
        "INSERT INTO public.graph_gql_hydration_rls_pgtest (id, name)
         VALUES ('u1', 'Hidden')",
    )
    .expect("insert hydration rls row failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_gql_hydration_rls_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add hydration rls table failed");
    Spi::run("SELECT * FROM graph.build()").expect("build hydration rls graph failed");
    Spi::run("ALTER TABLE public.graph_gql_hydration_rls_pgtest ENABLE ROW LEVEL SECURITY")
        .expect("enable hydration rls failed");
    Spi::run(
        "GRANT USAGE ON SCHEMA graph, public TO graph_gql_hydration_rls;
         GRANT SELECT ON public.graph_gql_hydration_rls_pgtest TO graph_gql_hydration_rls",
    )
    .expect("grant hydration rls role privileges failed");
    create_error_sqlstate_helper();

    Spi::run("SET ROLE graph_gql_hydration_rls").expect("set hydration rls role failed");
    let rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.gql(
            'MATCH (u:graph_gql_hydration_rls_pgtest) RETURN u',
            hydrate := true
         )",
    )
    .expect("hydration visibility count failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset hydration rls role failed");

    assert_eq!(rows, 0);
}

#[pg_test]
fn gql_coordinate_only_rows_fail_closed_when_source_row_is_not_visible() {
    reset_and_create_fixtures();
    Spi::run("DROP ROLE IF EXISTS graph_gql_coordinate_rls").expect("drop role failed");
    Spi::run("CREATE ROLE graph_gql_coordinate_rls").expect("create role failed");
    Spi::run("DROP TABLE IF EXISTS public.graph_gql_coordinate_rls_pgtest CASCADE")
        .expect("drop coordinate rls table failed");
    Spi::run(
        "CREATE TABLE public.graph_gql_coordinate_rls_pgtest (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL
            );
         INSERT INTO public.graph_gql_coordinate_rls_pgtest VALUES ('u1', 'Hidden');
         SELECT graph.add_table(
             'graph_gql_coordinate_rls_pgtest'::regclass,
             id_column := 'id',
             columns := ARRAY['name']
         );
         SELECT * FROM graph.build();
         ALTER TABLE public.graph_gql_coordinate_rls_pgtest ENABLE ROW LEVEL SECURITY;
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_coordinate_rls;
         GRANT SELECT ON public.graph_gql_coordinate_rls_pgtest TO graph_gql_coordinate_rls;",
    )
    .expect("create coordinate rls fixture failed");
    create_error_sqlstate_helper();

    Spi::run("SET ROLE graph_gql_coordinate_rls").expect("set coordinate rls role failed");
    let rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.gql(
            'MATCH (u:graph_gql_coordinate_rls_pgtest) RETURN u',
            hydrate := false
         )",
    )
    .expect("coordinate visibility count failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset coordinate rls role failed");

    assert_eq!(rows, 0);
}

#[pg_test]
fn gql_coordinate_only_relationships_fail_closed_when_edge_row_is_not_visible() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run("DROP ROLE IF EXISTS graph_gql_relationship_rls").expect("drop role failed");
    Spi::run("CREATE ROLE graph_gql_relationship_rls").expect("create role failed");
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest ENABLE ROW LEVEL SECURITY;
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_relationship_rls;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_gql_relationship_rls;
         GRANT SELECT ON public.graph_test_friendships_pgtest TO graph_gql_relationship_rls;",
    )
    .expect("create relationship rls fixture failed");
    create_error_sqlstate_helper();

    Spi::run("SET ROLE graph_gql_relationship_rls").expect("set relationship rls role failed");
    let coordinate_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                 RETURN u, r, v',
                hydrate := false
             )",
    )
    .expect("relationship coordinate count failed")
    .unwrap_or_default();
    let hydrated_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                 RETURN u, r, v',
                hydrate := true
             )",
    )
    .expect("relationship hydrated count failed")
    .unwrap_or_default();
    let aggregate_count = Spi::get_one::<i64>(
        "SELECT (row #>> '{relationship_count}')::bigint FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                 RETURN count(r) AS relationship_count',
                hydrate := false
             )",
    )
    .expect("relationship aggregate count failed")
    .unwrap_or_default();
    let exists = Spi::get_one::<bool>(
        "SELECT EXISTS (
                SELECT 1 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                     RETURN r',
                    hydrate := false
                )
             )",
    )
    .expect("relationship existence check failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset relationship rls role failed");

    assert_eq!((coordinate_rows, hydrated_rows, aggregate_count, exists), (0, 0, 0, false));
}

#[pg_test]
fn gql_mutable_overlay_relationships_fail_closed_under_rls() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name, age)
         VALUES ('u3', 'Carol', 43)",
    )
    .expect("insert distinct Any endpoint failed");
    Spi::run("SET graph.sync_mode = 'trigger'").expect("set trigger sync failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persisted build failed");
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.query_freshness = 'off'").expect("disable auto sync failed");
    Spi::run("SELECT graph.add_table('graph_test_users_pgtest'::regclass, 'id', ARRAY['name'])")
        .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
           'graph_test_friendships_pgtest'::regclass,
           'user_id',
           'graph_test_users_pgtest'::regclass,
           'friend_id',
           'friend',
           bidirectional := false
         )",
    )
    .expect("add friendship mapping failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f_overlay_hidden', 'u2', 'u1'),
                ('f_overlay_any_hidden', 'u3', 'u1'),
                ('f_overlay_visible', 'u1', 'u2')",
    )
    .expect("insert overlay relationship failed");
    let published = Spi::get_one::<i64>(
        "SELECT segments_published FROM graph.ingest_projection()",
    )
    .expect("ingest overlay relationship failed")
    .unwrap_or_default();
    assert!(published > 0, "overlay RLS fixture requires a durable segment");
    Spi::run("SET graph.auto_load = on").expect("enable auto-load failed");
    super::ENGINE.with(|engine| *engine.borrow_mut() = super::engine::Engine::new());
    let superuser_count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
           FROM graph.gql(
             'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN r',
             hydrate := false
           )",
    )
    .expect("load overlay relationship failed")
    .unwrap_or_default();
    assert_eq!(superuser_count, 1);

    Spi::run("DROP ROLE IF EXISTS graph_gql_overlay_rls").expect("drop overlay role failed");
    Spi::run("CREATE ROLE graph_gql_overlay_rls").expect("create overlay role failed");
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_gql_overlay_visible
           ON public.graph_test_friendships_pgtest FOR SELECT
           TO graph_gql_overlay_rls
           USING (id NOT IN ('f_overlay_hidden', 'f_overlay_any_hidden'));
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_overlay_rls;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_gql_overlay_rls;
         GRANT SELECT ON public.graph_test_friendships_pgtest TO graph_gql_overlay_rls;",
    )
    .expect("configure overlay RLS failed");
    create_error_sqlstate_helper();
    create_error_detail_helper();
    let row_count_queries = [
        "SELECT count(*)::bigint FROM graph.gql(
           'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN u, r, v',
           hydrate := false
         )",
        "SELECT count(*)::bigint FROM graph.gql(
           'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN u, r, v',
           hydrate := true
         )",
    ];
    Spi::run("SET ROLE graph_gql_overlay_rls").expect("set overlay role failed");
    let (visible_source_rows, hidden_source_rows, node_query_rows) = Spi::connect(|client| {
        let visible_source_rows = client
            .select(
                "SELECT count(*)::bigint
                   FROM public.graph_test_friendships_pgtest
                  WHERE id = 'f1'",
                None,
                &[],
            )?
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        let hidden_source_rows = client
            .select(
                "SELECT count(*)::bigint
                   FROM public.graph_test_friendships_pgtest
                  WHERE id = 'f_overlay_hidden'",
                None,
                &[],
            )?
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        let node_query_rows = client
            .select(
                "SELECT count(*)::bigint
                   FROM graph.gql(
                     'MATCH (u:graph_test_users_pgtest {id: ''u1''}) RETURN u',
                     hydrate := false
                   )",
                None,
                &[],
            )?
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((visible_source_rows, hidden_source_rows, node_query_rows))
    })
    .expect("restricted-role control queries failed");
    assert_eq!((visible_source_rows, hidden_source_rows, node_query_rows), (1, 0, 1));
    for query in row_count_queries {
        let rows = Spi::get_one::<i64>(query)
            .expect("overlay RLS row count failed")
            .unwrap_or_default();
        assert_eq!(rows, 0);
    }
    let aggregate_count = Spi::get_one::<i64>(
        "SELECT (row #>> '{relationship_count}')::bigint FROM graph.gql(
           'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN count(r) AS relationship_count',
           hydrate := false
         )",
    )
    .expect("overlay RLS aggregate count failed")
    .unwrap_or_default();
    let exists = Spi::get_one::<bool>(
        "SELECT EXISTS (
           SELECT 1 FROM graph.gql(
             'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN r',
             hydrate := false
           )
         )",
    )
    .expect("overlay RLS existence check failed")
    .unwrap_or_default();
    assert_eq!((aggregate_count, exists), (0, false));
    Spi::run("SELECT graph._test_set_visibility_strategy('eager')")
        .expect("force eager durable traversal failed");
    let eager_visible = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.traverse(
           'public.graph_test_users_pgtest'::regclass,
           'u1',
           1,
           edge_types := ARRAY['friend'],
           direction := 'out',
           strategy := 'bfs',
           hydrate := false
         )",
    )
    .expect("eager visible durable traversal failed")
    .unwrap_or_default();
    let eager_hidden_out = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.traverse(
           'public.graph_test_users_pgtest'::regclass,
           'u2',
           1,
           edge_types := ARRAY['friend'],
           direction := 'out',
           strategy := 'bfs',
           hydrate := false
         )",
    )
    .expect("eager hidden outbound durable traversal failed")
    .unwrap_or_default();
    let eager_hidden_in = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.traverse(
           'public.graph_test_users_pgtest'::regclass,
           'u1',
           1,
           edge_types := ARRAY['friend'],
           direction := 'in',
           strategy := 'bfs',
           hydrate := false
         )",
    )
    .expect("eager hidden inbound durable traversal failed")
    .unwrap_or_default();
    let eager_any = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.traverse(
           'public.graph_test_users_pgtest'::regclass,
           'u1',
           1,
           edge_types := ARRAY['friend'],
           direction := 'any',
           strategy := 'bfs',
           hydrate := false
         )",
    )
    .expect("eager Any durable traversal failed")
    .unwrap_or_default();
    Spi::run("SELECT graph._test_set_visibility_strategy('lazy')")
        .expect("force lazy durable traversal failed");
    let lazy_visible = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.traverse(
           'public.graph_test_users_pgtest'::regclass,
           'u1',
           1,
           edge_types := ARRAY['friend'],
           direction := 'out',
           strategy := 'bfs',
           hydrate := false
         )",
    )
    .expect("lazy visible durable traversal failed")
    .unwrap_or_default();
    let lazy_hidden_out = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.traverse(
           'public.graph_test_users_pgtest'::regclass,
           'u2',
           1,
           edge_types := ARRAY['friend'],
           direction := 'out',
           strategy := 'bfs',
           hydrate := false
         )",
    )
    .expect("lazy hidden outbound durable traversal failed")
    .unwrap_or_default();
    let lazy_hidden_in = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.traverse(
           'public.graph_test_users_pgtest'::regclass,
           'u1',
           1,
           edge_types := ARRAY['friend'],
           direction := 'in',
           strategy := 'bfs',
           hydrate := false
         )",
    )
    .expect("lazy hidden inbound durable traversal failed")
    .unwrap_or_default();
    let lazy_any = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.traverse(
           'public.graph_test_users_pgtest'::regclass,
           'u1',
           1,
           edge_types := ARRAY['friend'],
           direction := 'any',
           strategy := 'bfs',
           hydrate := false
         )",
    )
    .expect("lazy Any durable traversal failed")
    .unwrap_or_default();
    assert_eq!(
        (eager_visible, eager_hidden_out, eager_hidden_in, eager_any),
        (lazy_visible, lazy_hidden_out, lazy_hidden_in, lazy_any)
    );
    assert_eq!(
        (lazy_visible, lazy_hidden_out, lazy_hidden_in, lazy_any),
        (2, 1, 1, 2)
    );
    let lazy_metrics = Spi::get_one::<pgrx::JsonB>("SELECT graph._test_visibility_metrics()")
        .expect("read lazy durable traversal metrics failed")
        .expect("lazy durable traversal metrics missing");
    assert_eq!(
        lazy_metrics.0.get("strategy").and_then(|value| value.as_str()),
        Some("lazy")
    );
    Spi::run("SELECT graph._test_set_visibility_strategy('auto')")
        .expect("restore automatic visibility strategy failed");
    Spi::run(
        "RESET ROLE; SELECT graph._test_arm_missing_bfs_candidate_relationship_identity()",
    )
        .expect("arm missing durable relationship identity failed");
    Spi::run("SET ROLE graph_gql_overlay_rls").expect("restore overlay role failed");
    let missing_durable_statement = "SELECT count(*) FROM graph.traverse(
           'public.graph_test_users_pgtest'::regclass,
           'u2',
           1,
           edge_types := ARRAY['friend'],
           direction := 'out',
           strategy := 'bfs',
           hydrate := false
         )";
    let missing_durable_identity = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sqlstate({})",
        super::sql_literal(missing_durable_statement)
    ))
    .expect("missing durable relationship identity SQLSTATE capture failed");
    assert_eq!(missing_durable_identity.as_deref(), Some("55000"));
    Spi::run(
        "RESET ROLE; SELECT graph._test_arm_missing_bfs_candidate_relationship_identity()",
    )
    .expect("rearm missing durable relationship identity failed");
    Spi::run("SET ROLE graph_gql_overlay_rls").expect("restore overlay role for detail failed");
    let missing_durable_detail = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sql_error_detail({})",
        super::sql_literal(missing_durable_statement)
    ))
    .expect("missing durable relationship identity detail capture failed")
    .unwrap_or_default();
    assert!(missing_durable_detail.contains("PG023"));
    Spi::run("RESET ROLE").expect("reset overlay role failed");
    Spi::run("SET graph.auto_load = off").expect("restore auto-load failed");
    Spi::run("SET graph.persist_on_build = off").expect("restore persisted build failed");
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
    Spi::run("SET graph.sync_mode = 'manual'").expect("restore sync mode failed");
}

#[pg_test]
fn gql_join_relationships_fail_closed_when_edge_row_is_not_visible() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run("DROP ROLE IF EXISTS graph_gql_join_relationship_rls").expect("drop role failed");
    Spi::run("CREATE ROLE graph_gql_join_relationship_rls").expect("create role failed");
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest ENABLE ROW LEVEL SECURITY;
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_join_relationship_rls;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_gql_join_relationship_rls;
         GRANT SELECT ON public.graph_test_friendships_pgtest TO graph_gql_join_relationship_rls;",
    )
    .expect("create join relationship rls fixture failed");
    create_error_sqlstate_helper();

    Spi::run("SET ROLE graph_gql_join_relationship_rls").expect("set relationship rls role failed");
    let coordinate_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest),
                       (u)-[:friend]->(v)
                 RETURN u, r, v',
                hydrate := false
             )",
    )
    .expect("join relationship coordinate count failed")
    .unwrap_or_default();
    let hydrated_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest),
                       (u)-[:friend]->(v)
                 RETURN u, r, v',
                hydrate := true
             )",
    )
    .expect("join relationship hydrated count failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset relationship rls role failed");

    assert_eq!((coordinate_rows, hydrated_rows), (0, 0));
}

#[pg_test]
fn gql_wildcard_relationships_fail_closed_when_edge_row_is_not_visible() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run("DROP ROLE IF EXISTS graph_gql_wildcard_relationship_rls").expect("drop role failed");
    Spi::run("CREATE ROLE graph_gql_wildcard_relationship_rls").expect("create role failed");
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest ENABLE ROW LEVEL SECURITY;
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_wildcard_relationship_rls;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_gql_wildcard_relationship_rls;
         GRANT SELECT ON public.graph_test_friendships_pgtest TO graph_gql_wildcard_relationship_rls;",
    )
    .expect("create wildcard relationship rls fixture failed");
    create_error_sqlstate_helper();

    Spi::run("SET ROLE graph_gql_wildcard_relationship_rls")
        .expect("set relationship rls role failed");
    let coordinate_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.gql(
                'MATCH p=(u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                 RETURN p, r',
                hydrate := false
             )",
    )
    .expect("wildcard relationship coordinate count failed")
    .unwrap_or_default();
    let hydrated_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.gql(
                'MATCH p=(u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                 RETURN p, r',
                hydrate := true
             )",
    )
    .expect("wildcard relationship hydrated count failed")
    .unwrap_or_default();
    let relationship_list_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM graph.gql(
                'MATCH p=(u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                 RETURN relationships(p) AS relationships',
                hydrate := false
             )",
    )
    .expect("wildcard relationship-list count failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE").expect("reset relationship rls role failed");

    assert_eq!((coordinate_rows, hydrated_rows, relationship_list_rows), (0, 0, 0));
}

#[pg_test]
fn gql_join_and_wildcard_preflight_edge_acl_before_empty_results() {
    reset_and_create_fixtures();
    Spi::run(
        "DELETE FROM public.graph_test_friendships_pgtest;
         ALTER TABLE public.graph_test_friendships_pgtest
         ADD COLUMN rel_type text NOT NULL DEFAULT 'related_to';
         SELECT graph.add_table(
            'graph_test_users_pgtest'::regclass,
            id_column := 'id', columns := ARRAY['name', 'age']);
         SELECT graph.add_edge(
            from_table := 'graph_test_friendships_pgtest'::regclass,
            from_column := 'user_id', to_table := 'graph_test_users_pgtest'::regclass,
            to_column := 'friend_id', label := 'related_to', bidirectional := false,
            label_column := 'rel_type');
         SELECT * FROM graph.build()",
    )
    .expect("build empty dynamic-label graph failed");
    Spi::run("DROP ROLE IF EXISTS graph_gql_edge_acl_preflight")
        .expect("drop edge ACL role failed");
    Spi::run("CREATE ROLE graph_gql_edge_acl_preflight").expect("create edge ACL role failed");
    Spi::run(
        "GRANT USAGE ON SCHEMA graph, public TO graph_gql_edge_acl_preflight;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_gql_edge_acl_preflight;",
    )
    .expect("grant node-only ACL fixture failed");
    create_error_sqlstate_helper();

    Spi::run("SET ROLE graph_gql_edge_acl_preflight").expect("set edge ACL role failed");
    let join_sqlstate = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sqlstate({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest)-[:colleague]->(v:graph_test_users_pgtest),
                       (u)-[:colleague]->(v)
                 RETURN u, v',
                hydrate := false
             )"
        )
    ))
    .expect("join edge ACL SQLSTATE capture failed");
    let wildcard_sqlstate = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sqlstate({})",
        super::sql_literal(
            "SELECT EXISTS (
                SELECT 1 FROM graph.gql(
                    'MATCH p=(u:graph_test_users_pgtest)-[:colleague]->(v:graph_test_users_pgtest)
                     RETURN p',
                    hydrate := false
                )
             )"
        )
    ))
    .expect("wildcard edge ACL SQLSTATE capture failed");
    Spi::run("RESET ROLE").expect("reset edge ACL role failed");

    assert_eq!(join_sqlstate.as_deref(), Some("42501"));
    assert_eq!(wildcard_sqlstate.as_deref(), Some("42501"));
}

#[pg_test]
fn gql_with_projection_scope_aliases_and_shadows() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    create_error_capture_helper();

    let (source_name, shadowed_id, leak_denied) = Spi::connect(|client| {
        let source_name = client
            .select(
                "SELECT row #>> '{person_name}'
                 FROM graph.gql(
                     'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                      WITH u.name AS person_name, v AS u
                      RETURN person_name, u
                      ORDER BY person_name',
                     hydrate := true
                 )",
                None,
                &[],
            )
            .expect("with alias query failed")
            .first()
            .get::<String>(1)
            .expect("source name read failed")
            .unwrap_or_default();
        let shadowed_id = client
            .select(
                "SELECT row #>> '{u,_id,id}'
                 FROM graph.gql(
                     'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                      WITH u.name AS person_name, v AS u
                      RETURN person_name, u
                      ORDER BY person_name',
                     hydrate := true
                 )",
                None,
                &[],
            )
            .expect("with shadow query failed")
            .first()
            .get::<String>(1)
            .expect("shadowed id read failed")
            .unwrap_or_default();
        let leak_denied = client
            .select(
                "SELECT public.graph_test_sql_raises(
                     'SELECT * FROM graph.gql(
                        ''MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                          WITH v AS person
                          RETURN u''
                      )'
                 )",
                None,
                &[],
            )
            .expect("scope leak error capture failed")
            .first()
            .get::<bool>(1)
            .expect("scope leak bool read failed")
            .unwrap_or(false);
        Ok::<_, pgrx::spi::Error>((source_name, shadowed_id, leak_denied))
    })
    .expect("WITH scope query failed");

    assert_eq!(source_name, "Alice");
    assert_eq!(shadowed_id, "u2");
    assert!(leak_denied);
}

#[pg_test]
fn gql_optional_match_matches_left_outer_sql() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (gql_rows, gql_null_targets, sql_rows, sql_null_targets) = Spi::connect(|client| {
        let gql = client
            .select(
                "SELECT count(*)::bigint,
                        count(*) FILTER (WHERE row->'v' = 'null'::jsonb)::bigint
                 FROM graph.gql(
                     'OPTIONAL MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                      RETURN u, v',
                     hydrate := false
                 )",
                None,
                &[],
            )
            .expect("optional gql query failed")
            .first();
        let sql = client
            .select(
                "SELECT count(*)::bigint,
                        count(*) FILTER (WHERE v.id IS NULL)::bigint
                 FROM public.graph_test_users_pgtest u
                 LEFT JOIN public.graph_test_friendships_pgtest f ON f.user_id = u.id
                 LEFT JOIN public.graph_test_users_pgtest v ON v.id = f.friend_id",
                None,
                &[],
            )
            .expect("left outer sql query failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            gql.get::<i64>(1)
                .expect("gql row count read failed")
                .unwrap_or_default(),
            gql.get::<i64>(2)
                .expect("gql null count read failed")
                .unwrap_or_default(),
            sql.get::<i64>(1)
                .expect("sql row count read failed")
                .unwrap_or_default(),
            sql.get::<i64>(2)
                .expect("sql null count read failed")
                .unwrap_or_default(),
        ))
    })
    .expect("optional comparison failed");

    assert_eq!(gql_rows, sql_rows);
    assert_eq!(gql_null_targets, sql_null_targets);
    assert_eq!(gql_rows, 2);
    assert_eq!(gql_null_targets, 1);
}

#[pg_test]
fn gql_optional_match_filters_orders_and_paginates_null_rows() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (gql_rows, gql_null_names, gql_null_rels, sql_rows, sql_null_names, sql_null_rels) =
        Spi::connect(|client| {
            let gql = client
                .select(
                    "SELECT count(*)::bigint,
                            count(*) FILTER (WHERE row->'friend_name' = 'null'::jsonb)::bigint,
                            count(*) FILTER (WHERE row->'r' = 'null'::jsonb)::bigint
                     FROM graph.gql(
                         'OPTIONAL MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                          WHERE v.name = ''Alice''
                          RETURN u.id AS user_id, v.name AS friend_name, r
                          ORDER BY friend_name
                          SKIP 1 LIMIT 1',
                         hydrate := true
                     )",
                    None,
                    &[],
                )
                .expect("optional filtered gql query failed")
                .first();
            let sql = client
                .select(
                    "WITH optional_rows AS (
                         SELECT u.id AS user_id, opt.friend_name, opt.rel_id
                         FROM public.graph_test_users_pgtest u
                         LEFT JOIN (
                             SELECT f.user_id, f.id AS rel_id, v.name AS friend_name
                             FROM public.graph_test_friendships_pgtest f
                             JOIN public.graph_test_users_pgtest v
                               ON v.id = f.friend_id
                              AND v.name = 'Alice'
                         ) opt ON opt.user_id = u.id
                         ORDER BY opt.friend_name NULLS FIRST
                         OFFSET 1 LIMIT 1
                     )
                     SELECT count(*)::bigint,
                            count(*) FILTER (WHERE friend_name IS NULL)::bigint,
                            count(*) FILTER (WHERE rel_id IS NULL)::bigint
                     FROM optional_rows",
                    None,
                    &[],
                )
                .expect("optional filtered sql query failed")
                .first();
            Ok::<_, pgrx::spi::Error>((
                gql.get::<i64>(1)
                    .expect("gql row count read failed")
                    .unwrap_or_default(),
                gql.get::<i64>(2)
                    .expect("gql null name count read failed")
                    .unwrap_or_default(),
                gql.get::<i64>(3)
                    .expect("gql null relationship count read failed")
                    .unwrap_or_default(),
                sql.get::<i64>(1)
                    .expect("sql row count read failed")
                    .unwrap_or_default(),
                sql.get::<i64>(2)
                    .expect("sql null name count read failed")
                    .unwrap_or_default(),
                sql.get::<i64>(3)
                    .expect("sql null relationship count read failed")
                    .unwrap_or_default(),
            ))
        })
        .expect("optional filtered comparison failed");

    assert_eq!(gql_rows, sql_rows);
    assert_eq!(gql_null_names, sql_null_names);
    assert_eq!(gql_null_rels, sql_null_rels);
    assert_eq!((gql_rows, gql_null_names, gql_null_rels), (1, 1, 1));
}

#[pg_test]
fn gql_optional_match_hydrates_source_with_null_target_and_relationship() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (source_name, target_is_null, relationship_is_null) = Spi::connect(|client| {
        let row = client
            .select(
                "SELECT row #>> '{u,name}',
                        row->'v' = 'null'::jsonb,
                        row->'r' = 'null'::jsonb
                 FROM graph.gql(
                     'OPTIONAL MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                      RETURN u, r, v
                      ORDER BY u.name',
                     hydrate := true
                 )
                 WHERE row #>> '{u,_id,id}' = 'u2'",
                None,
                &[],
            )
            .expect("optional hydrated gql query failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<String>(1)
                .expect("source name read failed")
                .unwrap_or_default(),
            row.get::<bool>(2)
                .expect("target null read failed")
                .unwrap_or(false),
            row.get::<bool>(3)
                .expect("relationship null read failed")
                .unwrap_or(false),
        ))
    })
    .expect("optional hydrated check failed");

    assert_eq!(source_name, "Bob");
    assert!(target_is_null);
    assert!(relationship_is_null);
}

#[pg_test]
fn gql_preserves_parallel_source_relationship_rows() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f_parallel', 'u1', 'u2')",
    )
    .expect("insert parallel relationship failed");
    build_friendship_fixture_graph();

    let count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
           FROM graph.gql(
               'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                RETURN u, v',
               hydrate := false
           )
          WHERE row #>> '{u,_id,id}' = 'u1'
            AND row #>> '{v,_id,id}' = 'u2'",
    )
    .expect("parallel relationship query failed")
    .unwrap_or_default();
    assert_eq!(count, 2, "parallel source rows must remain distinct");

    let hydrated_edge_ids = Spi::get_one::<String>(
        "SELECT string_agg(row #>> '{r,id}', ',' ORDER BY row #>> '{r,id}')
           FROM graph.gql(
               'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                RETURN u, r, v',
               hydrate := true
           )
          WHERE row #>> '{u,_id,id}' = 'u1'
            AND row #>> '{v,_id,id}' = 'u2'",
    )
    .expect("parallel hydrated relationship query failed")
    .unwrap_or_default();
    assert_eq!(hydrated_edge_ids, "f1,f_parallel");
}

#[pg_test]
fn gql_ingested_parallel_relationship_identity_survives_reload() {
    reset_and_create_fixtures();
    Spi::run("SET graph.sync_mode = 'trigger'").expect("set trigger sync failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persisted build failed");
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.query_freshness = 'off'").expect("disable query auto-sync failed");
    Spi::run("SELECT graph.add_table('graph_test_users_pgtest'::regclass, 'id', ARRAY['name'])")
        .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
            'graph_test_friendships_pgtest'::regclass,
            'user_id',
            'graph_test_users_pgtest'::regclass,
            'friend_id',
            'friend',
            bidirectional := false
        )",
    )
    .expect("add friendship mapping failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable friendship graph failed");
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f_parallel', 'u1', 'u2')",
    )
    .expect("insert post-build parallel relationship failed");
    let published = Spi::get_one::<i64>(
        "SELECT segments_published FROM graph.ingest_projection()",
    )
    .expect("ingest parallel relationship failed")
    .unwrap_or_default();
    assert!(published > 0);
    Spi::run("SET graph.auto_load = on").expect("enable auto-load failed");
    super::ENGINE.with(|engine| *engine.borrow_mut() = super::engine::Engine::new());

    let (count, hydrated_ids) = Spi::connect(|client| {
        let count = client
            .select(
                "SELECT count(*)::bigint FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                     RETURN u, v',
                    hydrate := false
                 )
                 WHERE row #>> '{u,_id,id}' = 'u1'
                   AND row #>> '{v,_id,id}' = 'u2'",
                None,
                &[],
            )
            .expect("reloaded parallel coordinate query failed")
            .first()
            .get::<i64>(1)
            .expect("parallel count reads")
            .unwrap_or_default();
        let hydrated_ids = client
            .select(
                "SELECT string_agg(row #>> '{r,id}', ',' ORDER BY row #>> '{r,id}')
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest)
                     RETURN u, r, v',
                    hydrate := true
                 )
                 WHERE row #>> '{u,_id,id}' = 'u1'
                   AND row #>> '{v,_id,id}' = 'u2'",
                None,
                &[],
            )
            .expect("reloaded parallel hydration query failed")
            .first()
            .get::<String>(1)
            .expect("hydrated IDs read")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((count, hydrated_ids))
    })
    .expect("reloaded parallel results read");

    assert_eq!(count, 2);
    assert_eq!(hydrated_ids, "f1,f_parallel");
    Spi::run("SET graph.auto_load = off").expect("restore auto-load failed");
    Spi::run("SET graph.persist_on_build = off").expect("restore persisted build failed");
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
    Spi::run("SET graph.sync_mode = 'manual'").expect("restore sync mode failed");
}

#[pg_test]
fn standalone_relationship_sync_without_fk_uses_registered_endpoint_fallback() {
    reset_and_create_fixtures();
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest
           DROP CONSTRAINT graph_test_friendships_pgtest_user_id_fkey,
           DROP CONSTRAINT graph_test_friendships_pgtest_friend_id_fkey",
    )
    .expect("drop relationship endpoint constraints failed");
    Spi::run("SET graph.sync_mode = 'trigger'").expect("set trigger sync failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persisted build failed");
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.query_freshness = 'off'").expect("disable query auto-sync failed");
    Spi::run("SELECT graph.add_table('graph_test_users_pgtest'::regclass, 'id', ARRAY['name'])")
        .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
            'graph_test_friendships_pgtest'::regclass,
            'user_id',
            'graph_test_users_pgtest'::regclass,
            'friend_id',
            'friend',
            bidirectional := false
        )",
    )
    .expect("add relationship mapping failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f_reverse', 'u2', 'u1')",
    )
    .expect("insert relationship without FK failed");
    let published = Spi::get_one::<i64>(
        "SELECT segments_published FROM graph.ingest_projection()",
    )
    .expect("ingest relationship without FK failed")
    .unwrap_or_default();
    assert!(published > 0);
    Spi::run("SET graph.auto_load = on").expect("enable auto-load failed");
    super::ENGINE.with(|engine| *engine.borrow_mut() = super::engine::Engine::new());

    let count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
           FROM graph.traverse(
             'graph_test_users_pgtest'::regclass,
             'u2',
             1,
             edge_types := ARRAY['friend'],
             hydrate := false
           )
          WHERE node_id = 'u1'",
    )
    .expect("reloaded fallback traversal failed")
    .unwrap_or_default();
    assert_eq!(count, 1);
    Spi::run("SET graph.auto_load = off").expect("restore auto-load failed");
    Spi::run("SET graph.persist_on_build = off").expect("restore persisted build failed");
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
    Spi::run("SET graph.sync_mode = 'manual'").expect("restore sync mode failed");
}

#[pg_test]
fn partitioned_relationship_replay_uses_foreign_key_endpoint_root() {
    reset_and_create_fixtures();
    Spi::run("SET graph.sync_mode = 'trigger'").expect("set trigger sync failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persisted build failed");
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.query_freshness = 'off'").expect("disable query auto-sync failed");
    Spi::run(
        "CREATE TABLE public.graph_gql_replay_nodes_pgtest (
             id TEXT PRIMARY KEY,
             name TEXT NOT NULL
         ) PARTITION BY HASH (id);
         CREATE TABLE public.graph_gql_replay_nodes_p0_pgtest
           PARTITION OF public.graph_gql_replay_nodes_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 0);
         CREATE TABLE public.graph_gql_replay_nodes_p1_pgtest
           PARTITION OF public.graph_gql_replay_nodes_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 1);
         CREATE TABLE public.graph_gql_replay_decoy_pgtest (
             id TEXT PRIMARY KEY,
             name TEXT NOT NULL
         );
         CREATE TABLE public.graph_gql_replay_edges_pgtest (
             edge_id TEXT NOT NULL,
             bucket INTEGER NOT NULL,
             source_id TEXT NOT NULL REFERENCES public.graph_gql_replay_nodes_pgtest(id),
             target_id TEXT NOT NULL REFERENCES public.graph_gql_replay_nodes_pgtest(id),
             PRIMARY KEY (edge_id, bucket)
         ) PARTITION BY HASH (bucket);
         CREATE TABLE public.graph_gql_replay_edges_p0_pgtest
           PARTITION OF public.graph_gql_replay_edges_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 0);
         CREATE TABLE public.graph_gql_replay_edges_p1_pgtest
           PARTITION OF public.graph_gql_replay_edges_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 1);
         INSERT INTO public.graph_gql_replay_nodes_pgtest VALUES
           ('duplicate', 'Source'), ('target', 'Target');
         INSERT INTO public.graph_gql_replay_decoy_pgtest VALUES
           ('duplicate', 'Decoy');
         SELECT graph.add_table(
           'graph_gql_replay_nodes_pgtest'::regclass,
           id_column := 'id',
           columns := ARRAY['name']
         );
         SELECT graph.add_table(
           'graph_gql_replay_decoy_pgtest'::regclass,
           id_column := 'id',
           columns := ARRAY['name']
         );
         SELECT graph.add_edge(
           'graph_gql_replay_edges_pgtest'::regclass,
           'source_id',
           'graph_gql_replay_nodes_pgtest'::regclass,
           'target_id',
           'linked',
           bidirectional := false
         );
         SELECT * FROM graph.build(mode := 'mutable_overlay');",
    )
    .expect("create partitioned replay fixture failed");

    Spi::run(
        "INSERT INTO public.graph_gql_replay_edges_pgtest
           (edge_id, bucket, source_id, target_id)
         VALUES ('post-build', 1, 'duplicate', 'target')",
    )
    .expect("insert partitioned replay relationship failed");
    let published = Spi::get_one::<i64>(
        "SELECT segments_published FROM graph.ingest_projection()",
    )
    .expect("ingest partitioned replay relationship failed")
    .unwrap_or_default();
    assert!(published > 0);

    Spi::run("SET graph.auto_load = on").expect("enable auto-load failed");
    super::ENGINE.with(|engine| *engine.borrow_mut() = super::engine::Engine::new());
    let (main_reaches_target, decoy_reaches_target) = Spi::connect(|client| {
        let main = client
            .select(
                "SELECT count(*)::bigint
                   FROM graph.traverse(
                     'graph_gql_replay_nodes_pgtest'::regclass,
                     'duplicate',
                     1,
                     edge_types := ARRAY['linked'],
                     hydrate := false
                   )
                  WHERE node_id = 'target'",
                None,
                &[],
            )?
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        let decoy = client
            .select(
                "SELECT count(*)::bigint
                   FROM graph.traverse(
                     'graph_gql_replay_decoy_pgtest'::regclass,
                     'duplicate',
                     1,
                     edge_types := ARRAY['linked'],
                     hydrate := false
                   )
                  WHERE node_id = 'target'",
                None,
                &[],
            )?
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((main, decoy))
    })
    .expect("partitioned replay traversal verification failed");
    assert_eq!(main_reaches_target, 1);
    assert_eq!(decoy_reaches_target, 0);

    Spi::run("SET graph.auto_load = off").expect("restore auto-load failed");
    Spi::run("SET graph.persist_on_build = off").expect("restore persisted build failed");
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
    Spi::run("SET graph.sync_mode = 'manual'").expect("restore sync mode failed");
}

#[pg_test]
fn gql_aggregates_match_sql_grouping_and_numeric_results() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name, age)
         VALUES ('u3', 'Cara', 29), ('u4', 'Drew', 31)",
    )
    .expect("insert aggregate users failed");
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f2', 'u3', 'u2'), ('f3', 'u4', 'u1')",
    )
    .expect("insert aggregate friendships failed");
    build_friendship_fixture_graph();

    let (matches_sql, group_count, bob_names) = Spi::connect(|client| {
        let row = client
            .select(
                "WITH gql_rows AS (
                     SELECT jsonb_agg(row ORDER BY row->>'friend_name') AS rows,
                            count(*)::bigint AS group_count
                     FROM graph.gql(
                         'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                          RETURN v.name AS friend_name,
                                 count(u) AS users,
                                 sum(u.age) AS total_age,
                                 avg(u.age) AS avg_age,
                                 min(u.age) AS youngest,
                                 max(u.age) AS oldest,
                                 collect(u.name) AS names
                          ORDER BY friend_name',
                         hydrate := true
                     )
                 ),
                 sql_rows AS (
                     SELECT jsonb_agg(
                                jsonb_build_object(
                                    'friend_name', friend_name,
                                    'users', users,
                                    'total_age', total_age,
                                    'avg_age', avg_age,
                                    'youngest', youngest,
                                    'oldest', oldest,
                                    'names', names
                                )
                                ORDER BY friend_name
                            ) AS rows
                     FROM (
                         SELECT v.name AS friend_name,
                                count(u.id)::bigint AS users,
                                sum(u.age)::float8 AS total_age,
                                avg(u.age)::float8 AS avg_age,
                                min(u.age)::int AS youngest,
                                max(u.age)::int AS oldest,
                                jsonb_agg(to_jsonb(u.name) ORDER BY u.id) AS names
                         FROM public.graph_test_users_pgtest u
                         JOIN public.graph_test_friendships_pgtest f ON f.user_id = u.id
                         JOIN public.graph_test_users_pgtest v ON v.id = f.friend_id
                         GROUP BY v.name
                     ) expected
                 )
                 SELECT gql_rows.rows = sql_rows.rows,
                        gql_rows.group_count,
                        (
                            SELECT jsonb_agg(row->'names')->0
                            FROM graph.gql(
                                'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                                 RETURN v.name AS friend_name, collect(u.name) AS names
                                 ORDER BY friend_name',
                                hydrate := true
                            )
                            WHERE row->>'friend_name' = 'Bob'
                        )
                 FROM gql_rows, sql_rows",
                None,
                &[],
            )
            .expect("aggregate comparison query failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<bool>(1)
                .expect("aggregate equality failed")
                .unwrap_or(false),
            row.get::<i64>(2)
                .expect("aggregate group count failed")
                .unwrap_or_default(),
            row.get::<pgrx::JsonB>(3)
                .expect("aggregate Bob names failed")
                .unwrap(),
        ))
    })
    .expect("aggregate comparison failed");

    assert!(matches_sql);
    assert_eq!(group_count, 2);
    assert_eq!(bob_names.0, serde_json::json!(["Alice", "Cara"]));
}

#[pg_test]
fn gql_jsonb_property_paths_return_lists_maps_and_distinguish_missing_from_null() {
    reset_and_create_fixtures();
    Spi::run(
        "ALTER TABLE public.graph_test_users_pgtest
         ADD COLUMN profile jsonb NOT NULL DEFAULT '{}'::jsonb",
    )
    .expect("add profile column failed");
    Spi::run(
        "UPDATE public.graph_test_users_pgtest
         SET profile = CASE id
           WHEN 'u1' THEN '{\"plan\":\"pro\",\"tags\":[\"founder\",7],\"flags\":{\"beta\":true},\"explicit_null\":null}'::jsonb
           ELSE '{\"plan\":\"free\",\"tags\":[\"reader\"],\"flags\":{\"beta\":false}}'::jsonb
         END",
    )
    .expect("update profile json failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY[
                    'name',
                    'age',
                    'profile',
                    'profile.plan',
                    'profile.tags',
                    'profile.flags',
                    'profile.missing',
                    'profile.explicit_null'
                ]
            )",
    )
    .expect("add users table with profile failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := false
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build()").expect("build graph failed");

    let (tags, flags, missing, explicit_null_matches, missing_null_matches) =
        Spi::connect(|client| {
            let row = client
                .select(
                    "SELECT row->'tags',
                            row->'flags',
                            row->'missing',
                            (
                              SELECT count(*)::bigint
                              FROM graph.gql(
                                'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                                 WHERE u.profile.explicit_null IS NULL
                                 RETURN u.name AS name',
                                hydrate := true
                              )
                            ),
                            (
                              SELECT count(*)::bigint
                              FROM graph.gql(
                                'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                                 WHERE u.profile.missing IS NULL
                                 RETURN u.name AS name',
                                hydrate := true
                              )
                            )
                     FROM graph.gql(
                       'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                        WHERE u.profile.plan = ''pro''
                        RETURN u.profile.tags AS tags, u.profile.flags AS flags, u.profile.missing AS missing',
                       hydrate := true
                     )",
                    None,
                    &[],
                )
                .expect("jsonb gql query failed")
                .first();
            Ok::<_, pgrx::spi::Error>((
                row.get::<pgrx::JsonB>(1)
                    .expect("tags read failed")
                    .map(|json| json.0)
                    .unwrap_or(serde_json::Value::Null),
                row.get::<pgrx::JsonB>(2)
                    .expect("flags read failed")
                    .map(|json| json.0)
                    .unwrap_or(serde_json::Value::Null),
                row.get::<pgrx::JsonB>(3)
                    .expect("missing read failed")
                    .map(|json| json.0)
                    .unwrap_or(serde_json::Value::Null),
                row.get::<i64>(4)
                    .expect("explicit null count read failed")
                    .unwrap_or_default(),
                row.get::<i64>(5)
                    .expect("missing null count read failed")
                    .unwrap_or_default(),
            ))
        })
        .expect("jsonb property query failed");

    assert_eq!(tags, serde_json::json!(["founder", 7]));
    assert_eq!(flags, serde_json::json!({"beta": true}));
    assert!(missing.is_null());
    assert_eq!(explicit_null_matches, 1);
    assert_eq!(missing_null_matches, 0);
}

#[pg_test]
fn gql_rejects_jsonb_property_paths_on_non_jsonb_columns() {
    reset_and_create_fixtures();

    let denied = sql_raises(
        "SELECT graph.add_table(
             'graph_test_users_pgtest'::regclass,
             id_column := 'id',
             columns := ARRAY['name.first']
         )",
    );

    assert!(denied);
}

#[pg_test]
fn gql_distinct_matches_sql_distinct_counts() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name, age)
         VALUES ('u3', 'Cara', 29), ('u4', 'Drew', 31)",
    )
    .expect("insert distinct users failed");
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f2', 'u3', 'u2'), ('f3', 'u4', 'u2'), ('f4', 'u4', 'u1')",
    )
    .expect("insert distinct friendships failed");
    build_friendship_fixture_graph();

    let (
        return_distinct_matches,
        with_distinct_matches,
        aggregate_distinct_matches,
        collect_distinct_matches,
        optional_null_distinct_matches,
    ) =
        Spi::connect(|client| {
            let row = client
                .select(
                    "WITH sql_names AS (
                         SELECT jsonb_agg(name ORDER BY name) AS rows,
                                count(*)::bigint AS name_count
                         FROM (
                             SELECT DISTINCT v.name AS name
                             FROM public.graph_test_users_pgtest u
                             JOIN public.graph_test_friendships_pgtest f ON f.user_id = u.id
                             JOIN public.graph_test_users_pgtest v ON v.id = f.friend_id
                         ) distinct_names
                     ),
                     gql_names AS (
                         SELECT jsonb_agg(row->>'friend' ORDER BY row->>'friend') AS rows
                         FROM graph.gql(
                             'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                              RETURN DISTINCT v.name AS friend ORDER BY friend',
                             hydrate := true
                         )
                     ),
                     with_distinct AS (
                         SELECT (row->>'friends')::bigint AS friends
                         FROM graph.gql(
                             'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                              WITH DISTINCT v.name AS friend
                              RETURN count(*) AS friends',
                             hydrate := true
                         )
                     ),
                     aggregate_distinct AS (
                         SELECT row
                         FROM graph.gql(
                             'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                              RETURN count(DISTINCT v.name) AS friends,
                                     collect(DISTINCT v.name) AS names',
                             hydrate := true
                         )
                     ),
                     gql_collect_names AS (
                         SELECT jsonb_agg(value ORDER BY value) AS rows
                         FROM aggregate_distinct,
                              jsonb_array_elements_text(row->'names') AS value
                     ),
                     optional_gql AS (
                         SELECT row
                         FROM graph.gql(
                             'OPTIONAL MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                              RETURN count(DISTINCT v.name) AS named_friends,
                                     collect(DISTINCT v.name) AS names',
                             hydrate := true
                         )
                     ),
                     optional_sql AS (
                         SELECT count(DISTINCT v.name)::bigint AS named_friends,
                                false AS has_null
                         FROM public.graph_test_users_pgtest u
                         LEFT JOIN public.graph_test_friendships_pgtest f ON f.user_id = u.id
                         LEFT JOIN public.graph_test_users_pgtest v ON v.id = f.friend_id
                     ),
                     optional_sql_names AS (
                         SELECT jsonb_agg(name ORDER BY name) AS rows
                         FROM (
                             SELECT DISTINCT v.name AS name
                             FROM public.graph_test_users_pgtest u
                             LEFT JOIN public.graph_test_friendships_pgtest f ON f.user_id = u.id
                             LEFT JOIN public.graph_test_users_pgtest v ON v.id = f.friend_id
                             WHERE v.name IS NOT NULL
                         ) distinct_values
                     ),
                     optional_gql_names AS (
                         SELECT jsonb_agg(value ORDER BY value) AS rows
                         FROM optional_gql,
                              jsonb_array_elements(row->'names') AS value
                         WHERE value <> 'null'::jsonb
                     ),
                     optional_gql_null AS (
                         SELECT EXISTS (
                             SELECT 1
                             FROM optional_gql,
                                  jsonb_array_elements(row->'names') AS value
                             WHERE value = 'null'::jsonb
                         ) AS has_null
                     )
                     SELECT gql_names.rows = sql_names.rows,
                            (SELECT friends FROM with_distinct) = sql_names.name_count,
                            (SELECT (row->>'friends')::bigint FROM aggregate_distinct)
                                = sql_names.name_count,
                            (SELECT rows FROM gql_collect_names) = sql_names.rows,
                            (
                                SELECT (row->>'named_friends')::bigint = optional_sql.named_friends
                                       AND optional_gql_names.rows = optional_sql_names.rows
                                       AND optional_gql_null.has_null = optional_sql.has_null
                                FROM optional_gql, optional_sql, optional_gql_names,
                                     optional_sql_names, optional_gql_null
                            )
                     FROM gql_names, sql_names",
                    None,
                    &[],
                )
                .expect("distinct comparison query failed")
                .first();
            Ok::<_, pgrx::spi::Error>((
                row.get::<bool>(1)
                    .expect("return distinct equality failed")
                    .unwrap_or(false),
                row.get::<bool>(2)
                    .expect("with distinct count failed")
                    .unwrap_or(false),
                row.get::<bool>(3)
                    .expect("aggregate distinct count failed")
                    .unwrap_or(false),
                row.get::<bool>(4)
                    .expect("collect distinct names failed")
                    .unwrap_or(false),
                row.get::<bool>(5)
                    .expect("optional null distinct failed")
                    .unwrap_or(false),
            ))
        })
        .expect("distinct comparison failed");

    assert!(return_distinct_matches);
    assert!(with_distinct_matches);
    assert!(aggregate_distinct_matches);
    assert!(collect_distinct_matches);
    assert!(optional_null_distinct_matches);
}

#[pg_test]
fn gql_path_values_and_functions_have_stable_shape() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name, age)
         VALUES ('u3', 'Cara', 29)",
    )
    .expect("insert path user failed");
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f2', 'u2', 'u3')",
    )
    .expect("insert path friendship failed");
    build_friendship_fixture_graph();

    let (path_len, node_ids, relationship_count, raw_path_matches) = Spi::connect(|client| {
        let row = client
            .select(
                "WITH selected AS (
                     SELECT row
                     FROM graph.gql(
                         'MATCH (u:graph_test_users_pgtest)-[p:friend*2..2]->(v:graph_test_users_pgtest)
                          WHERE u.id = ''u1'' AND v.id = ''u3''
                          RETURN p,
                                 nodes(p) AS ns,
                                 relationships(p) AS rs,
                                 length(p) AS len',
                         hydrate := true
                     )
                     LIMIT 1
                 )
                 SELECT (row->>'len')::bigint,
                        (
                            SELECT jsonb_agg(node->>'id' ORDER BY ord)
                            FROM selected,
                                 jsonb_array_elements(row->'ns') WITH ORDINALITY AS n(node, ord)
                        ),
                        jsonb_array_length(row->'rs'),
                        row->'p'->'_path'->'nodes' = row->'ns'
                            AND row->'p'->'_path'->'relationships' = row->'rs'
                 FROM selected",
                None,
                &[],
            )
            .expect("path shape query failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<i64>(1)
                .expect("path length read failed")
                .unwrap_or_default(),
            row.get::<pgrx::JsonB>(2)
                .expect("path node ids read failed")
                .unwrap(),
            row.get::<i32>(3)
                .expect("relationship count read failed")
                .unwrap_or_default(),
            row.get::<bool>(4)
                .expect("raw path equality read failed")
                .unwrap_or(false),
        ))
    })
    .expect("path shape comparison failed");

    assert_eq!(path_len, 2);
    assert_eq!(node_ids.0, serde_json::json!(["u1", "u2", "u3"]));
    assert_eq!(relationship_count, 2);
    assert!(raw_path_matches);
}

#[pg_test]
fn gql_wildcard_path_values_and_functions_have_stable_shape() {
    reset_and_create_fixtures();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_companies_pgtest CASCADE")
        .expect("drop companies failed");
    Spi::run(
        "CREATE TABLE public.graph_test_companies_pgtest (
                id   TEXT PRIMARY KEY,
                name TEXT NOT NULL
            )",
    )
    .expect("create companies failed");
    Spi::run(
        "ALTER TABLE public.graph_test_users_pgtest
         ADD COLUMN company_id TEXT REFERENCES public.graph_test_companies_pgtest(id)",
    )
    .expect("add company reference failed");
    Spi::run(
        "INSERT INTO public.graph_test_companies_pgtest (id, name)
         VALUES ('c1', 'Acme')",
    )
    .expect("insert company failed");
    Spi::run("UPDATE public.graph_test_users_pgtest SET company_id = 'c1' WHERE id = 'u1'")
        .expect("set company failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_companies_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add companies table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := false
            )",
    )
    .expect("add friendship edge failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_users_pgtest'::regclass,
                'company_id',
                'graph_test_companies_pgtest'::regclass,
                'id',
                'works_at',
                bidirectional := false
            )",
    )
    .expect("add works_at edge failed");
    Spi::run("SELECT * FROM graph.build()").expect("build wildcard path graph failed");

    let (
        row_count,
        friend_rows,
        works_at_rows,
        named_works_at_rows,
        named_shape_matches,
        join_rows,
        join_shape_matches,
        join_predicate_matches,
        unhydrated_join_property_matches,
        wildcard_predicate_rows,
        wildcard_predicate_coordinate_only,
        wildcard_var_len_rows,
        wildcard_var_len_shape,
        wildcard_type_alternation_shape,
        shape_matches,
        coordinate_only_has_name,
    ) =
        Spi::connect(|client| {
            let row = client
                .select(
                    "WITH wildcard AS (
                         SELECT row
                         FROM graph.gql(
                             'MATCH p=()-[]->()
                              RETURN p,
                                     nodes(p) AS ns,
                                     relationships(p) AS rs,
                                     length(p) AS len',
                             hydrate := true
                         )
                     ),
                     coordinate_only AS (
                         SELECT row
                         FROM graph.gql('MATCH p=()-[]->() RETURN nodes(p) AS ns', hydrate := false)
                         LIMIT 1
                     ),
                     named_works_at AS (
                         SELECT row
                         FROM graph.gql(
                             'MATCH p=(s)-[r:works_at]->(e)
                              RETURN p, s, r, e',
                             hydrate := true
                         )
                     ),
                     joined_friend AS (
                         SELECT row
                         FROM graph.gql(
                             'MATCH (u:graph_test_users_pgtest)-[:friend]->(c:graph_test_users_pgtest),
                                    (v:graph_test_users_pgtest)-[:friend]->(c)
                              RETURN u.name AS source, v.name AS peer, c.name AS target',
                             hydrate := true
                         )
                     ),
                     joined_friend_predicate AS (
                         SELECT row
                         FROM graph.gql(
                             'MATCH (u:graph_test_users_pgtest)-[:friend]->(c:graph_test_users_pgtest),
                                    (v:graph_test_users_pgtest)-[:friend]->(c)
                              WHERE c.name = $target
                              RETURN u.name AS source, v.name AS peer, c.name AS target',
                             params := '{\"target\":\"Bob\"}'::jsonb,
                             hydrate := false
                         )
                     ),
                     joined_friend_unhydrated_property AS (
                         SELECT row
                         FROM graph.gql(
                             'MATCH (u:graph_test_users_pgtest)-[:friend]->(c:graph_test_users_pgtest),
                                    (v:graph_test_users_pgtest)-[:friend]->(c)
                              RETURN u.name AS source, c.name AS target',
                             hydrate := false
                         )
                     ),
                     wildcard_predicate AS (
                         SELECT row
                         FROM graph.gql(
                             'MATCH p=()-[:friend]->(e)
                              WHERE e.name = $target
                              RETURN e, length(p) AS len',
                             params := '{\"target\":\"Bob\"}'::jsonb,
                             hydrate := false
                         )
                     ),
                     wildcard_var_len AS (
                         SELECT row
                         FROM graph.gql(
                             'MATCH p=()-[*1..2]->()
                              RETURN p, length(p) AS len, relationships(p) AS rs',
                             hydrate := false
                         )
                     ),
                     wildcard_type_alternation AS (
                         SELECT row
                         FROM graph.gql(
                             'MATCH p=()-[:friend|works_at]->()
                              RETURN relationships(p) AS rs',
                             hydrate := false
                         )
                     )
                     SELECT count(*)::bigint,
                            count(*) FILTER (WHERE row->'rs'->0->>'_type' = 'friend')::bigint,
                            count(*) FILTER (WHERE row->'rs'->0->>'_type' = 'works_at')::bigint,
                            (SELECT count(*)::bigint FROM named_works_at),
                            (SELECT bool_and(
                                row->'s' ? '_id'
                                AND row->'e' ? '_id'
                                AND row->'r'->>'_type' = 'works_at'
                                AND row->'p'->'_path'->'relationships'->0 = row->'r'
                             ) FROM named_works_at),
                            (SELECT count(*)::bigint FROM joined_friend),
                            (SELECT bool_and(
                                row->>'source' = row->>'peer'
                                AND row->>'target' = 'Bob'
                             ) FROM joined_friend),
                            (SELECT bool_and(
                                row->>'source' = 'Alice'
                                AND row->>'peer' = 'Alice'
                                AND row->>'target' = 'Bob'
                             ) FROM joined_friend_predicate),
                            (SELECT bool_and(
                                row->>'source' = 'Alice'
                                AND row->>'target' = 'Bob'
                             ) FROM joined_friend_unhydrated_property),
                            (SELECT count(*)::bigint FROM wildcard_predicate),
                            (SELECT bool_and(
                                (row->>'len')::bigint = 1
                                AND row->'e' ? '_id'
                                AND NOT (row->'e' ? 'name')
                             ) FROM wildcard_predicate),
                            (SELECT count(*)::bigint FROM wildcard_var_len),
                            (SELECT bool_and(
                                (row->>'len')::bigint BETWEEN 1 AND 2
                                AND jsonb_array_length(row->'rs') = (row->>'len')::integer
                                AND row->'p'->'_path'->'relationships' = row->'rs'
                             ) FROM wildcard_var_len),
                            (SELECT count(*) = 2
                                AND count(*) FILTER (WHERE row->'rs'->0->>'_type' = 'friend') = 1
                                AND count(*) FILTER (WHERE row->'rs'->0->>'_type' = 'works_at') = 1
                             FROM wildcard_type_alternation),
                            bool_and(
                                (row->>'len')::bigint = 1
                                AND row->'p'->'_path'->'nodes' = row->'ns'
                                AND row->'p'->'_path'->'relationships' = row->'rs'
                                AND row->'rs'->0 ? '_start'
                                AND row->'rs'->0 ? '_end'
                            ),
                            (SELECT row->'ns'->0 ? 'name' FROM coordinate_only)
                     FROM wildcard",
                    None,
                    &[],
                )
                .expect("wildcard path shape query failed")
                .first();
            Ok::<_, pgrx::spi::Error>((
                row.get::<i64>(1)
                    .expect("wildcard row count read failed")
                    .unwrap_or_default(),
                row.get::<i64>(2)
                    .expect("friend row count read failed")
                    .unwrap_or_default(),
                row.get::<i64>(3)
                    .expect("works_at row count read failed")
                    .unwrap_or_default(),
                row.get::<i64>(4)
                    .expect("named works_at row count read failed")
                    .unwrap_or_default(),
                row.get::<bool>(5)
                    .expect("named works_at shape read failed")
                    .unwrap_or(false),
                row.get::<i64>(6)
                    .expect("join row count read failed")
                    .unwrap_or_default(),
                row.get::<bool>(7)
                    .expect("join shape read failed")
                    .unwrap_or(false),
                row.get::<bool>(8)
                    .expect("join predicate read failed")
                    .unwrap_or(false),
                row.get::<bool>(9)
                    .expect("unhydrated join property read failed")
                    .unwrap_or(false),
                row.get::<i64>(10)
                    .expect("wildcard predicate row count read failed")
                    .unwrap_or_default(),
                row.get::<bool>(11)
                    .expect("wildcard predicate shape read failed")
                    .unwrap_or(false),
                row.get::<i64>(12)
                    .expect("wildcard var-len row count read failed")
                    .unwrap_or_default(),
                row.get::<bool>(13)
                    .expect("wildcard var-len shape read failed")
                    .unwrap_or(false),
                row.get::<bool>(14)
                    .expect("wildcard type alternation shape read failed")
                    .unwrap_or(false),
                row.get::<bool>(15)
                    .expect("shape equality read failed")
                    .unwrap_or(false),
                row.get::<bool>(16)
                    .expect("coordinate-only shape read failed")
                    .unwrap_or(true),
            ))
        })
        .expect("wildcard path shape comparison failed");

    assert_eq!(
        row_count, 2,
        "unexpected wildcard path row count: friend={friend_rows}, works_at={works_at_rows}"
    );
    assert_eq!(friend_rows, 1, "unexpected friend path count");
    assert_eq!(works_at_rows, 1, "unexpected works_at path count");
    assert_eq!(named_works_at_rows, works_at_rows);
    assert!(named_shape_matches);
    assert_eq!(join_rows, 1, "unexpected multi-pattern join count");
    assert!(join_shape_matches);
    assert!(join_predicate_matches);
    assert!(unhydrated_join_property_matches);
    assert_eq!(
        wildcard_predicate_rows, 1,
        "unexpected wildcard predicate count"
    );
    assert!(wildcard_predicate_coordinate_only);
    assert!(
        wildcard_var_len_rows >= row_count,
        "unexpected wildcard var-len count: {wildcard_var_len_rows}"
    );
    assert!(wildcard_var_len_shape);
    assert!(wildcard_type_alternation_shape);
    assert!(shape_matches);
    assert!(!coordinate_only_has_name);

    let wildcard_var_len_named_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row->'s'->'_id'->>'id' = 'u1'
                    AND row->'e'->'_id'->>'id' = 'u2'
                    AND row->'r'->'_path'->'nodes'->0->'_id'->>'id' = 'u1'
                    AND row->'r'->'_path'->'nodes'->1->'_id'->>'id' = 'u2'
                    AND row->'r'->'_path'->'relationships'->0->>'_type' = 'friend'
                    AND (row->>'len')::integer = 1
                )
         FROM graph.gql(
             'MATCH p=(s:graph_test_users_pgtest)-[r:friend*1..1]->(e:graph_test_users_pgtest)
              RETURN s, r, e, length(r) AS len',
             hydrate := false
         )",
    )
    .expect("wildcard variable-length named segment query failed")
    .unwrap_or(false);

    assert!(wildcard_var_len_named_shape);

    let join_relationship_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row->'r'->>'_type' = 'friend'
                    AND row->'r'->'_start'->>'table' = 'graph_test_users_pgtest'
                    AND row->'r'->'_start'->>'id' = 'u1'
                    AND row->'r'->'_end'->>'table' = 'graph_test_users_pgtest'
                    AND row->'r'->'_end'->>'id' = 'u2'
                )
         FROM graph.gql(
             'MATCH (u:graph_test_users_pgtest)-[r:friend]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              RETURN r',
             hydrate := false
         )",
    )
    .expect("multi-pattern relationship projection query failed")
    .unwrap_or(false);

    assert!(join_relationship_shape);

    let join_path_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row->'path'->'_path'->'nodes'->0->'_id'->>'id' = 'u1'
                    AND row->'path'->'_path'->'nodes'->1->'_id'->>'id' = 'u2'
                    AND row->'path'->'_path'->'relationships'->0->>'_type' = 'friend'
                    AND row->'path'->'_path'->'relationships'->0->'_start'->>'id' = 'u1'
                    AND row->'path'->'_path'->'relationships'->0->'_end'->>'id' = 'u2'
                )
         FROM graph.gql(
             'MATCH p=(u:graph_test_users_pgtest)-[:friend]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              RETURN p AS path',
             hydrate := false
         )",
    )
    .expect("multi-pattern path projection query failed")
    .unwrap_or(false);

    assert!(join_path_shape);

    let join_path_function_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row->'ns'->0->'_id'->>'id' = 'u1'
                    AND row->'ns'->1->'_id'->>'id' = 'u2'
                    AND row->'rs'->0->>'_type' = 'friend'
                    AND (row->>'len')::integer = 1
                )
         FROM graph.gql(
             'MATCH p=(u:graph_test_users_pgtest)-[:friend]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              RETURN nodes(p) AS ns, relationships(p) AS rs, length(p) AS len',
             hydrate := false
         )",
    )
    .expect("multi-pattern path-function projection query failed")
    .unwrap_or(false);

    assert!(join_path_function_shape);

    let join_var_len_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row->'ns'->0->'_id'->>'id' = 'u1'
                    AND row->'ns'->1->'_id'->>'id' = 'u2'
                    AND row->'rs'->0->>'_type' = 'friend'
                    AND (row->>'len')::integer = 1
                )
         FROM graph.gql(
             'MATCH p=(u:graph_test_users_pgtest)-[:friend*1..1]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              RETURN nodes(p) AS ns, relationships(p) AS rs, length(p) AS len',
             hydrate := false
         )",
    )
    .expect("multi-pattern variable-length projection query failed")
    .unwrap_or(false);

    assert!(join_var_len_shape);

    let join_var_len_relationship_var_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row->'r'->'_path'->'nodes'->0->'_id'->>'id' = 'u1'
                    AND row->'r'->'_path'->'nodes'->1->'_id'->>'id' = 'u2'
                    AND row->'r'->'_path'->'relationships'->0->>'_type' = 'friend'
                    AND (row->>'len')::integer = 1
                )
         FROM graph.gql(
             'MATCH (u:graph_test_users_pgtest)-[r:friend*1..1]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              RETURN r, length(r) AS len',
             hydrate := false
         )",
    )
    .expect("multi-pattern variable-length relationship variable query failed")
    .unwrap_or(false);

    assert!(join_var_len_relationship_var_shape);

    let optional_join_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row->>'source' = 'Alice'
                    AND row->>'target' = 'Bob'
                    AND row->'peer' = 'null'::jsonb
                    AND row->'rel' = 'null'::jsonb
                    AND row->'path' = 'null'::jsonb
                    AND (row->>'hops')::integer = 0
                )
         FROM graph.gql(
             'OPTIONAL MATCH (u:graph_test_users_pgtest)-[:friend]->(c:graph_test_users_pgtest),
                    p=(v:graph_test_users_pgtest)-[r:friend]->(u)
              WHERE u.name = ''Alice''
              RETURN u.name AS source, c.name AS target, v.name AS peer,
                     r AS rel, p AS path, length(p) AS hops',
             hydrate := true
         )",
    )
    .expect("multi-pattern optional projection query failed")
    .unwrap_or(false);

    assert!(optional_join_shape);

    let join_aggregate_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row->>'target' = 'Bob'
                    AND (row->>'rows')::integer = 1
                    AND (row->>'total_age')::float8 = 37
                    AND (row->>'rels')::integer = 1
                    AND row #>> '{rel_values,0,_type}' = 'friend'
                    AND (row->>'paths')::integer = 1
                    AND row #>> '{path_values,0,_path,relationships,0,_type}' = 'friend'
                )
         FROM graph.gql(
             'MATCH p=(u:graph_test_users_pgtest)-[r:friend]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              RETURN c.name AS target, count(*) AS rows, sum(v.age) AS total_age,
                     count(DISTINCT r) AS rels, collect(r) AS rel_values,
                     count(DISTINCT p) AS paths, collect(p) AS path_values',
             hydrate := true
         )",
    )
    .expect("multi-pattern aggregate projection query failed")
    .unwrap_or(false);

    assert!(join_aggregate_shape);

    let join_with_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row->>'targets' = '1'
                )
         FROM graph.gql(
             'MATCH (u:graph_test_users_pgtest)-[:friend]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              WITH DISTINCT c.name AS target
              RETURN count(*) AS targets',
             hydrate := false
         )",
    )
    .expect("multi-pattern WITH projection query failed")
    .unwrap_or(false);

    assert!(join_with_shape);

    let join_relationship_path_with_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row #>> '{rel,_type}' = 'friend'
                    AND row #>> '{path,_path,relationships,0,_type}' = 'friend'
                )
         FROM graph.gql(
             'MATCH p=(u:graph_test_users_pgtest)-[r:friend]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              WITH r AS rel, p AS path, v.name AS peer
              RETURN rel, path
              ORDER BY peer
              LIMIT 1',
             hydrate := false
         )",
    )
    .expect("multi-pattern relationship/path WITH projection query failed")
    .unwrap_or(false);

    assert!(join_relationship_path_with_shape);

    let join_path_function_with_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    (row->>'hops')::integer = 1
                    AND row->'ns'->0->'_id'->>'id' = 'u1'
                    AND row->'rs'->0->>'_type' = 'friend'
                )
         FROM graph.gql(
             'MATCH p=(u:graph_test_users_pgtest)-[:friend]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              WITH length(p) AS hops, nodes(p) AS ns, relationships(p) AS rs
              RETURN hops, ns, rs
              ORDER BY hops',
             hydrate := false
         )",
    )
    .expect("multi-pattern path-function WITH projection query failed")
    .unwrap_or(false);

    assert!(join_path_function_with_shape);

    let join_aggregate_with_shape = Spi::get_one::<bool>(
        "SELECT bool_and(
                    row->>'target' = 'Bob'
                    AND (row->>'rows')::integer = 1
                    AND (row->>'rels')::integer = 1
                    AND row #>> '{paths,0,_path,relationships,0,_type}' = 'friend'
                )
         FROM graph.gql(
             'MATCH p=(u:graph_test_users_pgtest)-[r:friend]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              WITH DISTINCT c.name AS target, count(*) AS rows, count(DISTINCT r) AS rels,
                   collect(p) AS paths
              RETURN target, rows, rels, paths
              ORDER BY rows DESC, target',
             hydrate := false
         )",
    )
    .expect("multi-pattern aggregate WITH projection query failed")
    .unwrap_or(false);

    assert!(join_aggregate_with_shape);

    let join_post_aggregate_shape = Spi::get_one::<bool>(
        "SELECT bool_and((row->>'total')::numeric = 1)
         FROM graph.gql(
             'MATCH (u:graph_test_users_pgtest)-[:friend]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              WITH c.name AS target, count(*) AS rows
              WITH DISTINCT rows AS total
              RETURN total
              ORDER BY total',
             hydrate := false
         )",
    )
    .expect("multi-pattern post-aggregate WITH DISTINCT query failed")
    .unwrap_or(false);

    assert!(join_post_aggregate_shape);

    let join_post_aggregate_sum = Spi::get_one::<bool>(
        "SELECT bool_and((row->>'total')::numeric = 1)
         FROM graph.gql(
             'MATCH (u:graph_test_users_pgtest)-[:friend]->(c:graph_test_users_pgtest),
                    (v:graph_test_users_pgtest)-[:friend]->(c)
              WITH c.name AS target, count(*) AS rows
              RETURN sum(rows) AS total',
             hydrate := false
         )",
    )
    .expect("multi-pattern post-aggregate RETURN aggregate query failed")
    .unwrap_or(false);

    assert!(join_post_aggregate_sum);
}

#[pg_test]
fn gql_aggregates_return_empty_group_and_optional_null_counts() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (empty_count, empty_sum_is_null, empty_names, optional_rows, optional_matches) =
        Spi::connect(|client| {
            let empty = client
                .select(
                    "SELECT (row #>> '{rows}')::bigint,
                            row->'total_age' = 'null'::jsonb,
                            row->'names'
                     FROM graph.gql(
                         'MATCH (u:graph_test_users_pgtest)
                          WHERE u.name = ''Missing''
                          RETURN count(*) AS rows, sum(u.age) AS total_age, collect(u.name) AS names',
                         hydrate := true
                     )",
                    None,
                    &[],
                )
                .expect("empty aggregate gql query failed")
                .first();
            let optional = client
                .select(
                    "SELECT (row #>> '{source_rows}')::bigint,
                            (row #>> '{matched_targets}')::bigint
                     FROM graph.gql(
                         'OPTIONAL MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
                          WHERE v.name = ''Alice''
                          RETURN count(*) AS source_rows, count(v) AS matched_targets',
                         hydrate := true
                     )",
                    None,
                    &[],
                )
                .expect("optional aggregate gql query failed")
                .first();
            Ok::<_, pgrx::spi::Error>((
                empty.get::<i64>(1).expect("empty count failed").unwrap_or_default(),
                empty.get::<bool>(2).expect("empty sum failed").unwrap_or(false),
                empty.get::<pgrx::JsonB>(3).expect("empty names failed").unwrap(),
                optional
                    .get::<i64>(1)
                    .expect("optional rows failed")
                    .unwrap_or_default(),
                optional
                    .get::<i64>(2)
                    .expect("optional matches failed")
                    .unwrap_or_default(),
            ))
        })
        .expect("aggregate null comparison failed");

    assert_eq!(empty_count, 0);
    assert!(empty_sum_is_null);
    assert_eq!(empty_names.0, serde_json::json!([]));
    assert_eq!(optional_rows, 2);
    assert_eq!(optional_matches, 0);
}

#[pg_test]
fn gql_denies_without_select_on_bound_tables() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run("DROP ROLE IF EXISTS graph_gql_no_select").expect("drop role failed");
    Spi::run("CREATE ROLE graph_gql_no_select").expect("create role failed");
    Spi::run("GRANT USAGE ON SCHEMA graph TO graph_gql_no_select").expect("grant schema failed");
    Spi::run("REVOKE SELECT ON public.graph_test_users_pgtest FROM PUBLIC")
        .expect("revoke public select failed");
    create_error_capture_helper();
    Spi::run("SET ROLE graph_gql_no_select").expect("set role failed");
    let denied = Spi::get_one::<bool>(&format!(
        "SELECT public.graph_test_sql_raises({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest) RETURN u, v'
             )"
        )
    ))
    .expect("acl error capture query failed")
    .unwrap_or(false);
    let optional_denied = Spi::get_one::<bool>(&format!(
        "SELECT public.graph_test_sql_raises({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'OPTIONAL MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest) RETURN u, v'
             )"
        )
    ))
    .expect("optional acl error capture query failed")
    .unwrap_or(false);
    Spi::run("RESET ROLE").expect("reset role failed");

    assert!(denied);
    assert!(optional_denied);
}

#[pg_test]
fn gql_applies_session_tenant_scope_to_topology() {
    reset_and_create_fixtures();
    Spi::run("SET LOCAL graph.enforce_tenant_scope = on")
        .expect("enable tenant enforcement failed");
    Spi::run("SET LOCAL graph.tenant_setting = 'app.graph_gql_tenant'")
        .expect("set tenant GUC failed");
    Spi::run("DROP TABLE IF EXISTS public.graph_gql_tenant_pgtest CASCADE")
        .expect("drop tenant table failed");
    Spi::run(
        "CREATE TABLE public.graph_gql_tenant_pgtest (
                id TEXT PRIMARY KEY,
                tenant_id TEXT NOT NULL,
                name TEXT NOT NULL,
                parent_id TEXT REFERENCES public.graph_gql_tenant_pgtest(id)
            )",
    )
    .expect("create tenant table failed");
    Spi::run(
        "INSERT INTO public.graph_gql_tenant_pgtest (id, tenant_id, name, parent_id) VALUES
                ('a1', 'tenant-a', 'Root A', NULL),
                ('a2', 'tenant-a', 'Child A', 'a1'),
                ('b1', 'tenant-b', 'Root B', NULL),
                ('b2', 'tenant-b', 'Child B', 'b1')",
    )
    .expect("insert tenant rows failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_gql_tenant_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name'],
                tenant_column := 'tenant_id'
            )",
    )
    .expect("add tenant table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_gql_tenant_pgtest'::regclass,
                'parent_id',
                'graph_gql_tenant_pgtest'::regclass,
                'id',
                'parent',
                bidirectional := false
            )",
    )
    .expect("add tenant edge failed");
    Spi::run("SELECT * FROM graph.build()").expect("build tenant graph failed");

    Spi::run("SET LOCAL app.graph_gql_tenant = 'tenant-a'").expect("set tenant-a failed");
    let tenant_a_child = Spi::get_one::<String>(
        "SELECT row #>> '{child,_id,id}'
             FROM graph.gql(
                'MATCH (child:graph_gql_tenant_pgtest)-[:parent]->(parent:graph_gql_tenant_pgtest) RETURN child, parent'
             )",
    )
    .expect("tenant-a gql failed")
    .unwrap_or_default();

    Spi::run("SET LOCAL app.graph_gql_tenant = 'tenant-b'").expect("set tenant-b failed");
    let tenant_b_child = Spi::get_one::<String>(
        "SELECT row #>> '{child,_id,id}'
             FROM graph.gql(
                'MATCH (child:graph_gql_tenant_pgtest)-[:parent]->(parent:graph_gql_tenant_pgtest) RETURN child, parent'
             )",
    )
    .expect("tenant-b gql failed")
    .unwrap_or_default();

    Spi::run("RESET app.graph_gql_tenant").expect("reset tenant value failed");
    Spi::run("RESET graph.tenant_setting").expect("reset tenant setting failed");
    Spi::run("SET graph.enforce_tenant_scope = off").expect("disable tenant enforcement failed");

    assert_eq!(tenant_a_child, "a2");
    assert_eq!(tenant_b_child, "b2");
}

#[pg_test]
fn gql_reads_transaction_delta_edge_overlay() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable GQL writes failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("rebuild mutable friendship graph failed");

    Spi::run(
        "SELECT *
         FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest {id: ''u2''}),
                   (v:graph_test_users_pgtest {id: ''u1''})
             CREATE (u)-[r:friend {id: ''f2''}]->(v)
             RETURN r',
            hydrate := false
         )",
    )
    .expect("create transaction relationship failed");

    let reverse_count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
         FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest) RETURN u, v',
            hydrate := false
         )
         WHERE row #>> '{u,_id,id}' = 'u2'
           AND row #>> '{v,_id,id}' = 'u1'",
    )
    .expect("gql tx overlay read failed")
    .unwrap_or_default();

    assert_eq!(reverse_count, 1);
}

#[pg_test]
fn gql_hides_transaction_delta_edge_delete() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    Spi::run(
        "SELECT graph._test_record_tx_edge(
            'graph_test_users_pgtest'::regclass,
            'u1',
            'graph_test_users_pgtest'::regclass,
            'u2',
            'friend',
            'delete'
        )",
    )
    .expect("record tx edge delete failed");

    let base_edge_count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
         FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest) RETURN u, v',
            hydrate := false
         )
         WHERE row #>> '{u,_id,id}' = 'u1'
           AND row #>> '{v,_id,id}' = 'u2'",
    )
    .expect("gql tx overlay delete read failed")
    .unwrap_or_default();

    assert_eq!(base_edge_count, 0);
}

#[pg_test]
fn gql_create_node_requires_mutable_overlay_projection() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    create_error_capture_helper();

    let (denied, source_count) = Spi::connect(|client| {
        let denied = client
            .select(
                &format!(
                    "SELECT public.graph_test_sql_raises({})",
                    super::sql_literal(
                        "SELECT * FROM graph.gql(
                'CREATE (u:graph_test_users_pgtest {id: ''u3'', name: ''Cara'', age: 29}) RETURN u'
             )"
                    )
                ),
                None,
                &[],
            )
            .expect("create error capture query failed")
            .first()
            .get::<bool>(1)
            .expect("create error capture read failed")
            .unwrap_or(false);
        let source_count = client
            .select(
                "SELECT count(*)::bigint FROM public.graph_test_users_pgtest WHERE id = 'u3'",
                None,
                &[],
            )
            .expect("source count query failed")
            .first()
            .get::<i64>(1)
            .expect("source count read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((denied, source_count))
    })
    .expect("readonly create verification failed");

    assert!(denied);
    assert_eq!(source_count, 0);
}

#[pg_test]
fn gql_create_node_inserts_mapped_row_and_records_delta() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (created_id, created_name, source_count, tx_added_nodes, node_match_count) =
        Spi::connect(|client| {
        let created = client
            .select(
                "SELECT
                    row #>> '{u,_id,id}',
                    row #>> '{u,name}'
                 FROM graph.gql(
                    'CREATE (u:graph_test_users_pgtest {id: ''u3'', name: $name, age: 29}) RETURN u',
                    params := '{\"name\":\"Cara\"}'::jsonb
                 )",
                None,
                &[],
            )
            .expect("gql create failed")
            .first();
        let source_count = client
            .select(
                "SELECT count(*)::bigint
                 FROM public.graph_test_users_pgtest
                 WHERE id = 'u3' AND name = 'Cara' AND age = 29",
                None,
                &[],
            )
            .expect("source row count failed")
            .first()
            .get::<i64>(1)
            .expect("source row count read failed")
            .unwrap_or_default();
        let tx_added_nodes = client
            .select("SELECT tx_delta_added_nodes FROM graph.status()", None, &[])
            .expect("status query failed")
            .first()
            .get::<i32>(1)
            .expect("tx added node count read failed")
            .unwrap_or_default();
        let node_match_count = client
            .select(
                "SELECT count(*)::bigint
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u3''}) RETURN u',
                    hydrate := false
                 )",
                None,
                &[],
            )
            .expect("node scan query failed")
            .first()
            .get::<i64>(1)
            .expect("node scan count read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((
            created
                .get::<String>(1)
                .expect("created id read failed")
                .unwrap_or_default(),
            created
                .get::<String>(2)
                .expect("created name read failed")
                .unwrap_or_default(),
            source_count,
            tx_added_nodes,
            node_match_count,
        ))
        })
        .expect("create verification failed");

    assert_eq!(created_id, "u3");
    assert_eq!(created_name, "Cara");
    assert_eq!(source_count, 1);
    assert_eq!(tx_added_nodes, 1);
    assert_eq!(node_match_count, 1);
}

#[pg_test]
fn gql_create_respects_partition_routing_constraints_and_user_triggers() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "CREATE TABLE public.graph_gql_partitioned_nodes_pgtest (
             id TEXT PRIMARY KEY,
             name TEXT NOT NULL CHECK (name <> 'blocked')
         ) PARTITION BY HASH (id);
         CREATE TABLE public.graph_gql_partitioned_nodes_p0_pgtest
           PARTITION OF public.graph_gql_partitioned_nodes_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 0);
         CREATE TABLE public.graph_gql_partitioned_nodes_p1_pgtest
           PARTITION OF public.graph_gql_partitioned_nodes_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 1);
         CREATE FUNCTION public.graph_gql_partition_guard_pgtest()
         RETURNS trigger LANGUAGE plpgsql AS $$
         BEGIN
           IF NEW.name = 'triggered' THEN
             RAISE EXCEPTION 'partition trigger rejected row';
           END IF;
           RETURN NEW;
         END
         $$;
         CREATE TRIGGER graph_gql_partition_guard
           BEFORE INSERT ON public.graph_gql_partitioned_nodes_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_partition_guard_pgtest();
         SELECT graph.add_table(
           'graph_gql_partitioned_nodes_pgtest'::regclass,
           id_column := 'id',
           columns := ARRAY['name']
         );
         SELECT * FROM graph.build(mode := 'mutable_overlay');",
    )
    .expect("create partitioned write fixture failed");
    create_error_sqlstate_helper();

    for (name, expected_sqlstate) in [("blocked", "23514"), ("triggered", "P0001")] {
        let query = format!(
            "SELECT * FROM graph.gql(
               'CREATE (u:graph_gql_partitioned_nodes_pgtest {{id: ''{name}'', name: ''{name}''}}) RETURN u'
             )"
        );
        let sqlstate = Spi::get_one::<String>(&format!(
            "SELECT public.graph_test_sqlstate({})",
            super::sql_literal(&query)
        ))
        .expect("capture rejected partitioned write failed")
        .unwrap_or_default();
        assert_eq!(sqlstate, expected_sqlstate);
    }
    let rejected_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
           FROM public.graph_gql_partitioned_nodes_pgtest
          WHERE id IN ('blocked', 'triggered')",
    )
    .expect("rejected partition rows count failed")
    .unwrap_or_default();
    let dirty_after_rejections = Spi::get_one::<bool>(
        "SELECT tx_delta_dirty FROM graph.status()",
    )
    .expect("rejected overlay status failed")
    .unwrap_or(true);
    assert_eq!(rejected_rows, 0);
    assert!(!dirty_after_rejections);

    Spi::run(
        "SELECT * FROM graph.gql(
           'CREATE (u:graph_gql_partitioned_nodes_pgtest {id: ''accepted'', name: ''Accepted''}) RETURN u'
         )",
    )
    .expect("accepted partitioned write failed");
    let (partition_name, matched) = Spi::connect(|client| {
        let partition_name = client
            .select(
                "SELECT tableoid::regclass::text
                   FROM public.graph_gql_partitioned_nodes_pgtest
                  WHERE id = 'accepted'",
                None,
                &[],
            )?
            .first()
            .get::<String>(1)?
            .unwrap_or_default();
        let matched = client
            .select(
                "SELECT count(*)::bigint
                   FROM graph.gql(
                     'MATCH (u:graph_gql_partitioned_nodes_pgtest {id: ''accepted''}) RETURN u',
                     hydrate := false
                   )",
                None,
                &[],
            )?
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((partition_name, matched))
    })
    .expect("accepted partition write verification failed");
    assert!(partition_name.contains("graph_gql_partitioned_nodes_p"));
    assert_eq!(matched, 1);
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
}

#[pg_test]
fn gql_partitioned_node_mutations_preserve_source_and_filter_contracts() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
         "CREATE TABLE public.graph_gql_write_nodes_pgtest (
             id TEXT PRIMARY KEY,
             score INTEGER NOT NULL CHECK (score <= 100),
             note TEXT,
             required_note TEXT NOT NULL
         ) PARTITION BY HASH (id);
         CREATE TABLE public.graph_gql_write_nodes_p0_pgtest
           PARTITION OF public.graph_gql_write_nodes_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 0);
         CREATE TABLE public.graph_gql_write_nodes_p1_pgtest
           PARTITION OF public.graph_gql_write_nodes_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 1);
         CREATE TABLE public.graph_gql_write_edges_pgtest (
             id TEXT PRIMARY KEY,
             source_id TEXT NOT NULL REFERENCES public.graph_gql_write_nodes_pgtest(id),
             target_id TEXT NOT NULL REFERENCES public.graph_gql_write_nodes_pgtest(id)
         );
         INSERT INTO public.graph_gql_write_nodes_pgtest (id, score, note, required_note)
         VALUES ('n1', 1, 'keep', 'required'), ('n2', 2, 'protected', 'required');
         INSERT INTO public.graph_gql_write_edges_pgtest VALUES ('e1', 'n1', 'n2');
         CREATE FUNCTION public.graph_gql_write_node_guard_pgtest()
         RETURNS trigger LANGUAGE plpgsql AS $$
         BEGIN
           IF NEW.score = 13 THEN
             RAISE EXCEPTION 'node write trigger rejected row';
           ELSIF NEW.score IN (7, 8) THEN
             NEW.score := NEW.score * 2;
           END IF;
           IF TG_OP = 'UPDATE' AND OLD.id = 'n2'
              AND OLD.note IS NOT NULL AND NEW.note IS NULL THEN
             RAISE EXCEPTION 'node remove trigger rejected row';
           END IF;
           RETURN NEW;
         END
         $$;
         CREATE TRIGGER graph_gql_write_node_guard
           BEFORE INSERT OR UPDATE ON public.graph_gql_write_nodes_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_write_node_guard_pgtest();
         SELECT graph.add_table(
           'graph_gql_write_nodes_pgtest'::regclass,
           id_column := 'id',
           columns := ARRAY['score', 'note', 'required_note']
         );
         SELECT graph.add_edge(
           'graph_gql_write_edges_pgtest'::regclass,
           'source_id',
           'graph_gql_write_nodes_pgtest'::regclass,
           'target_id',
           'linked',
           bidirectional := false
         );
         SELECT graph.add_filter_column('graph_gql_write_nodes_pgtest'::regclass, 'score');
         SELECT * FROM graph.build(mode := 'mutable_overlay');",
    )
    .expect("create partitioned node write fixture failed");
    create_error_sqlstate_helper();

    for (query, expected_sqlstate) in [
        (
            "MATCH (u:graph_gql_write_nodes_pgtest {id: 'n2'}) SET u.score = 101 RETURN u",
            "23514",
        ),
        (
            "MATCH (u:graph_gql_write_nodes_pgtest {id: 'n2'}) SET u.score = 13 RETURN u",
            "P0001",
        ),
        (
            "MATCH (u:graph_gql_write_nodes_pgtest {id: 'n2'}) REMOVE u.note RETURN u",
            "P0001",
        ),
        (
            "MATCH (u:graph_gql_write_nodes_pgtest {id: 'n2'}) REMOVE u.required_note RETURN u",
            "23502",
        ),
        (
            "MERGE (u:graph_gql_write_nodes_pgtest {id: 'constraint-merge', score: 101, required_note: 'required'}) RETURN u",
            "23514",
        ),
        (
            "MERGE (u:graph_gql_write_nodes_pgtest {id: 'trigger-merge', score: 13, required_note: 'required'}) RETURN u",
            "P0001",
        ),
        (
            "MERGE (u:graph_gql_write_nodes_pgtest {id: 'n2'}) ON MATCH SET u.score = 101 RETURN u",
            "23514",
        ),
        (
            "MERGE (u:graph_gql_write_nodes_pgtest {id: 'n2'}) ON MATCH SET u.score = 13 RETURN u",
            "P0001",
        ),
    ] {
        let statement = format!("SELECT * FROM graph.gql({})", super::sql_literal(query));
        let sqlstate = Spi::get_one::<String>(&format!(
            "SELECT public.graph_test_sqlstate({})",
            super::sql_literal(&statement)
        ))
        .expect("capture rejected partitioned node write failed")
        .unwrap_or_default();
        assert_eq!(sqlstate, expected_sqlstate);

        let (source_score, source_note, required_note, rejected_rows, tx_dirty, matched_score) =
            Spi::connect(|client| {
                let source = client
                    .select(
                        "SELECT score, note, required_note
                           FROM public.graph_gql_write_nodes_pgtest
                          WHERE id = 'n2'",
                        None,
                        &[],
                    )?
                    .first();
                let rejected_rows = client
                    .select(
                        "SELECT count(*)::bigint
                           FROM public.graph_gql_write_nodes_pgtest
                          WHERE id IN ('constraint-merge', 'trigger-merge')",
                        None,
                        &[],
                    )?
                    .first()
                    .get::<i64>(1)?
                    .unwrap_or_default();
                let tx_dirty = client
                    .select("SELECT tx_delta_dirty FROM graph.status()", None, &[])?
                    .first()
                    .get::<bool>(1)?
                    .unwrap_or(true);
                let matched_score = client
                    .select(
                        "SELECT (row #>> '{u,score}')::int
                           FROM graph.gql(
                             'MATCH (u:graph_gql_write_nodes_pgtest {id: ''n2''}) RETURN u'
                           )",
                        None,
                        &[],
                    )?
                    .first()
                    .get::<i32>(1)?
                    .unwrap_or_default();
                Ok::<_, pgrx::spi::Error>((
                    source.get::<i32>(1)?.unwrap_or_default(),
                    source.get::<String>(2)?.unwrap_or_default(),
                    source.get::<String>(3)?.unwrap_or_default(),
                    rejected_rows,
                    tx_dirty,
                    matched_score,
                ))
            })
            .expect("verify rejected partitioned node write failed");
        assert_eq!(source_score, 2);
        assert_eq!(source_note, "protected");
        assert_eq!(required_note, "required");
        assert_eq!(rejected_rows, 0);
        assert!(!tx_dirty);
        assert_eq!(matched_score, 2);
    }

    let (set_score, source_score, filtered_neighbors) = Spi::connect(|client| {
        let set_score = client
            .select(
                "SELECT (row #>> '{score}')::int
                   FROM graph.gql(
                     'MATCH (u:graph_gql_write_nodes_pgtest {id: ''n2''}) SET u.score = 7 RETURN u.score AS score'
                   )",
                None,
                &[],
            )?
            .first()
            .get::<i32>(1)?
            .unwrap_or_default();
        let source_score = client
            .select(
                "SELECT score FROM public.graph_gql_write_nodes_pgtest WHERE id = 'n2'",
                None,
                &[],
            )?
            .first()
            .get::<i32>(1)?
            .unwrap_or_default();
        let filtered_neighbors = client
            .select(
                "SELECT count(*)::bigint
                   FROM graph.traverse(
                     'graph_gql_write_nodes_pgtest'::regclass,
                     'n1',
                     1,
                     filter := '{\"node\":{\"where\":{\"score\":{\"gt\":10}}}}'::jsonb,
                     hydrate := false
                   )
                  WHERE node_id = 'n2'",
                None,
                &[],
            )?
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((set_score, source_score, filtered_neighbors))
    })
    .expect("verify trigger-normalized SET failed");
    assert_eq!(set_score, 14);
    assert_eq!(source_score, 14);
    assert_eq!(filtered_neighbors, 1);

    let (matched_merge_score, filtered_neighbors, inserted_merge_score, merge_partition) =
        Spi::connect(|client| {
            let matched_merge_score = client
                .select(
                    "SELECT (row #>> '{score}')::int
                       FROM graph.gql(
                         'MERGE (u:graph_gql_write_nodes_pgtest {id: ''n2''})
                          ON MATCH SET u.score = 8
                          RETURN u.score AS score'
                       )",
                    None,
                    &[],
                )?
                .first()
                .get::<i32>(1)?
                .unwrap_or_default();
            let filtered_neighbors = client
                .select(
                    "SELECT count(*)::bigint
                       FROM graph.traverse(
                         'graph_gql_write_nodes_pgtest'::regclass,
                         'n1',
                         1,
                         filter := '{\"node\":{\"where\":{\"score\":{\"gt\":15}}}}'::jsonb,
                         hydrate := false
                       )
                      WHERE node_id = 'n2'",
                    None,
                    &[],
                )?
                .first()
                .get::<i64>(1)?
                .unwrap_or_default();
            let inserted_merge_score = client
            .select(
                "SELECT (row #>> '{score}')::int
                   FROM graph.gql(
                         'MERGE (u:graph_gql_write_nodes_pgtest {id: ''accepted-merge'', score: 8, required_note: ''required''}) RETURN u.score AS score'
                   )",
                None,
                &[],
            )?
            .first()
            .get::<i32>(1)?
            .unwrap_or_default();
            let merge_partition = client
            .select(
                "SELECT tableoid::regclass::text
                   FROM public.graph_gql_write_nodes_pgtest
                  WHERE id = 'accepted-merge' AND score = 16",
                None,
                &[],
            )?
            .first()
            .get::<String>(1)?
            .unwrap_or_default();
            Ok::<_, pgrx::spi::Error>((
                matched_merge_score,
                filtered_neighbors,
                inserted_merge_score,
                merge_partition,
            ))
        })
        .expect("verify trigger-normalized MERGE branches failed");
    assert_eq!(matched_merge_score, 16);
    assert_eq!(filtered_neighbors, 1);
    assert_eq!(inserted_merge_score, 16);
    assert!(merge_partition.contains("graph_gql_write_nodes_p"));
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
}

#[pg_test]
fn gql_partitioned_relationship_writes_preserve_composite_source_identity() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "CREATE TABLE public.graph_gql_write_rel_nodes_pgtest (
             id TEXT PRIMARY KEY
         );
         CREATE TABLE public.graph_gql_write_rels_pgtest (
             edge_id TEXT NOT NULL,
             revision INTEGER NOT NULL,
             source_id TEXT NOT NULL REFERENCES public.graph_gql_write_rel_nodes_pgtest(id),
             target_id TEXT NOT NULL REFERENCES public.graph_gql_write_rel_nodes_pgtest(id),
             note TEXT NOT NULL CHECK (note <> 'blocked'),
             PRIMARY KEY (edge_id, revision)
         ) PARTITION BY HASH (revision);
         CREATE TABLE public.graph_gql_write_rels_p0_pgtest
           PARTITION OF public.graph_gql_write_rels_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 0);
         CREATE TABLE public.graph_gql_write_rels_p1_pgtest
           PARTITION OF public.graph_gql_write_rels_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 1);
         INSERT INTO public.graph_gql_write_rel_nodes_pgtest VALUES ('n1'), ('n2'), ('n3');
         INSERT INTO public.graph_gql_write_rels_pgtest
         VALUES ('protected', 0, 'n2', 'n1', 'protected');
         CREATE FUNCTION public.graph_gql_write_rel_guard_pgtest()
         RETURNS trigger LANGUAGE plpgsql AS $$
         BEGIN
           IF TG_OP = 'INSERT' THEN
             IF NEW.note = 'reject' THEN
               RAISE EXCEPTION 'relationship insert trigger rejected row';
             ELSIF NEW.note = 'mutate' THEN
               NEW.note := 'normalized';
             END IF;
             RETURN NEW;
           END IF;
           IF OLD.note = 'protected' THEN
             RAISE EXCEPTION 'relationship delete trigger rejected row';
           END IF;
           RETURN OLD;
         END
         $$;
         CREATE TRIGGER graph_gql_write_rel_insert_guard
           BEFORE INSERT ON public.graph_gql_write_rels_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_write_rel_guard_pgtest();
         CREATE TRIGGER graph_gql_write_rel_delete_guard
           BEFORE DELETE ON public.graph_gql_write_rels_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_write_rel_guard_pgtest();
         SELECT graph.add_table('graph_gql_write_rel_nodes_pgtest'::regclass, 'id');
         SELECT graph.add_edge(
           'graph_gql_write_rels_pgtest'::regclass,
           'source_id',
           'graph_gql_write_rel_nodes_pgtest'::regclass,
           'target_id',
           'linked',
           bidirectional := false
         );
         SELECT * FROM graph.build(mode := 'mutable_overlay');",
    )
    .expect("create partitioned relationship write fixture failed");
    create_error_sqlstate_helper();
    create_error_message_helper();

    for (edge_id, revision, note, expected_sqlstate) in [
        ("constraint", 1, "blocked", "23514"),
        ("trigger", 2, "reject", "P0001"),
    ] {
        let query = format!(
            "SELECT * FROM graph.gql(
               'MATCH (a:graph_gql_write_rel_nodes_pgtest {{id: ''n1''}}),
                      (b:graph_gql_write_rel_nodes_pgtest {{id: ''n2''}})
                CREATE (a)-[r:linked {{edge_id: ''{edge_id}'', revision: {revision}, note: ''{note}''}}]->(b)
                RETURN r'
             )"
        );
        let sqlstate = Spi::get_one::<String>(&format!(
            "SELECT public.graph_test_sqlstate({})",
            super::sql_literal(&query)
        ))
        .expect("capture rejected relationship create failed")
        .unwrap_or_default();
        if sqlstate != expected_sqlstate {
            let message = Spi::get_one::<String>(&format!(
                "SELECT public.graph_test_sql_error_message({})",
                super::sql_literal(&query)
            ))
            .expect("capture rejected relationship create message failed")
            .unwrap_or_default();
            assert_eq!(sqlstate, expected_sqlstate, "{message}");
        }
        let rejected_rows = Spi::get_one::<i64>(&format!(
            "SELECT count(*)::bigint FROM public.graph_gql_write_rels_pgtest WHERE edge_id = {}",
            super::sql_literal(edge_id)
        ))
        .expect("rejected relationship source count failed")
        .unwrap_or_default();
        let (tx_dirty, projected_rows) = Spi::connect(|client| {
            let status = client
                .select("SELECT tx_delta_dirty FROM graph.status()", None, &[])?
                .first()
                .get::<bool>(1)?
                .unwrap_or(true);
            let projected = client
                .select(
                    "SELECT count(*)::bigint
                       FROM graph.gql(
                         'MATCH (a:graph_gql_write_rel_nodes_pgtest {id: ''n1''})-[r:linked]->(b:graph_gql_write_rel_nodes_pgtest {id: ''n2''}) RETURN r',
                         hydrate := false
                       )",
                    None,
                    &[],
                )?
                .first()
                .get::<i64>(1)?
                .unwrap_or_default();
            Ok::<_, pgrx::spi::Error>((status, projected))
        })
        .expect("verify rejected relationship create failed");
        assert_eq!(rejected_rows, 0);
        assert!(!tx_dirty);
        assert_eq!(projected_rows, 0);
    }

    let (returned_note, first_partition) = Spi::connect(|client| {
        let created = client
            .select(
                "SELECT row #>> '{r,note}'
                   FROM graph.gql(
                     'MATCH (a:graph_gql_write_rel_nodes_pgtest {id: ''n1''}),
                            (b:graph_gql_write_rel_nodes_pgtest {id: ''n2''})
                      CREATE (a)-[r:linked {edge_id: ''shared'', revision: 1, note: ''mutate''}]->(b)
                      RETURN r'
                   )",
                None,
                &[],
            )?
            .first()
            .get::<String>(1)?
            .unwrap_or_default();
        client.select(
            "SELECT * FROM graph.gql(
               'MATCH (a:graph_gql_write_rel_nodes_pgtest {id: ''n1''}),
                      (b:graph_gql_write_rel_nodes_pgtest {id: ''n3''})
                CREATE (a)-[r:linked {edge_id: ''shared'', revision: 2, note: ''second''}]->(b)
                RETURN r'
             )",
            None,
            &[],
        )?;
        let partition = client
            .select(
                "SELECT tableoid::regclass::text
                   FROM public.graph_gql_write_rels_pgtest
                  WHERE edge_id = 'shared' AND revision = 1 AND note = 'normalized'",
                None,
                &[],
            )?
            .first()
            .get::<String>(1)?
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((created, partition))
    })
    .expect("verify accepted relationship creates failed");
    assert_eq!(returned_note, "normalized");
    assert!(first_partition.contains("graph_gql_write_rels_p"));
    let shared_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
           FROM public.graph_gql_write_rels_pgtest
          WHERE edge_id = 'shared' AND revision IN (1, 2)",
    )
    .expect("composite relationship source count failed")
    .unwrap_or_default();
    assert_eq!(shared_rows, 2);

    let protected_delete =
        "SELECT * FROM graph.gql(
           'MATCH (a:graph_gql_write_rel_nodes_pgtest {id: ''n2''})-[r:linked]->(b:graph_gql_write_rel_nodes_pgtest {id: ''n1''}) DELETE r RETURN r'
         )";
    let sqlstate = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sqlstate({})",
        super::sql_literal(protected_delete)
    ))
    .expect("capture rejected relationship delete failed")
    .unwrap_or_default();
    assert_eq!(sqlstate, "P0001");
    let (protected_rows, deleted_edges) = Spi::connect(|client| {
        let protected_rows = client
            .select(
                "SELECT count(*)::bigint
                   FROM public.graph_gql_write_rels_pgtest
                  WHERE edge_id = 'protected' AND revision = 0",
                None,
                &[],
            )?
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        let deleted_edges = client
            .select("SELECT tx_delta_deleted_edges FROM graph.status()", None, &[])?
            .first()
            .get::<i32>(1)?
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((protected_rows, deleted_edges))
    })
    .expect("verify rejected relationship delete failed");
    assert_eq!(protected_rows, 1);
    assert_eq!(deleted_edges, 0);

    Spi::run(
        "SELECT * FROM graph.gql(
           'MATCH (a:graph_gql_write_rel_nodes_pgtest {id: ''n1''})-[r:linked]->(b:graph_gql_write_rel_nodes_pgtest {id: ''n2''})
            DELETE r RETURN r'
         )",
    )
    .expect("delete first accepted composite relationship failed");
    let remaining_revision = Spi::get_one::<i32>(
        "SELECT revision FROM public.graph_gql_write_rels_pgtest WHERE edge_id = 'shared'",
    )
    .expect("remaining composite relationship revision failed")
    .unwrap_or(-1);
    assert_eq!(remaining_revision, 2);
    Spi::run(
        "SELECT * FROM graph.gql(
           'MATCH (a:graph_gql_write_rel_nodes_pgtest {id: ''n1''})-[r:linked]->(b:graph_gql_write_rel_nodes_pgtest {id: ''n3''})
            DELETE r RETURN r'
         )",
    )
    .expect("delete second accepted composite relationship failed");
    let remaining_shared = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_gql_write_rels_pgtest WHERE edge_id = 'shared'",
    )
    .expect("remaining composite relationship count failed")
    .unwrap_or(-1);
    assert_eq!(remaining_shared, 0);
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
}

#[pg_test]
fn gql_create_relationship_inserts_edge_row_and_records_delta() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := false
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (
        returned_id,
        returned_type,
        returned_start,
        returned_end,
        source_rows,
        tx_added_edges,
        same_tx_gql_count,
        same_tx_hydrated_id,
    ) = Spi::connect(|client| {
        let created = client
            .select(
                "SELECT
                    row #>> '{r,id}',
                    row #>> '{r,_type}',
                    row #>> '{r,_start,id}',
                    row #>> '{r,_end,id}'
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''}), (v:graph_test_users_pgtest {id: ''u1''})
                     CREATE (u)-[r:friend {id: ''f2''}]->(v)
                     RETURN r'
                 )",
                None,
                &[],
            )
            .expect("gql relationship create failed")
            .first();
        let source_rows = client
            .select(
                "SELECT count(*)::bigint
                 FROM public.graph_test_friendships_pgtest
                 WHERE id = 'f2' AND user_id = 'u2' AND friend_id = 'u1'",
                None,
                &[],
            )
            .expect("source edge row count failed")
            .first()
            .get::<i64>(1)
            .expect("source edge row count read failed")
            .unwrap_or_default();
        let tx_added_edges = client
            .select("SELECT tx_delta_added_edges FROM graph.status()", None, &[])
            .expect("status query failed")
            .first()
            .get::<i32>(1)
            .expect("tx added edge count read failed")
            .unwrap_or_default();
        let same_tx_gql_count = client
            .select(
                "SELECT count(*)::bigint
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN r',
                    hydrate := false
                 )",
                None,
                &[],
            )
            .expect("same transaction relationship read failed")
            .first()
            .get::<i64>(1)
            .expect("same transaction relationship count read failed")
            .unwrap_or_default();
        let same_tx_hydrated_id = client
            .select(
                "SELECT row #>> '{r,id}'
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN r',
                    hydrate := true
                 )",
                None,
                &[],
            )
            .expect("same transaction hydrated relationship read failed")
            .first()
            .get::<String>(1)
            .expect("same transaction hydrated relationship id read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((
            created
                .get::<String>(1)
                .expect("returned relationship id read failed")
                .unwrap_or_default(),
            created
                .get::<String>(2)
                .expect("returned relationship type read failed")
                .unwrap_or_default(),
            created
                .get::<String>(3)
                .expect("returned start id read failed")
                .unwrap_or_default(),
            created
                .get::<String>(4)
                .expect("returned end id read failed")
                .unwrap_or_default(),
            source_rows,
            tx_added_edges,
            same_tx_gql_count,
            same_tx_hydrated_id,
        ))
    })
    .expect("relationship create verification failed");

    assert_eq!(returned_id, "f2");
    assert_eq!(returned_type, "friend");
    assert_eq!(returned_start, "u2");
    assert_eq!(returned_end, "u1");
    assert_eq!(source_rows, 1);
    assert_eq!(tx_added_edges, 1);
    assert_eq!(same_tx_gql_count, 1);
    assert_eq!(same_tx_hydrated_id, "f2");
}

#[pg_test]
fn gql_create_relationship_can_use_transaction_created_nodes() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := false
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (created_edge_rows, path_rows, tx_added_nodes, tx_added_edges) = Spi::connect(|client| {
        client
            .select(
                "SELECT *
                 FROM graph.gql(
                    'CREATE (u:graph_test_users_pgtest {id: ''u3'', name: ''Cara'', age: 29}) RETURN u',
                    hydrate := false
                 )",
                None,
                &[],
            )
            .expect("gql node create failed");
        client
            .select(
                "SELECT *
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u3''}), (v:graph_test_users_pgtest {id: ''u1''})
                     CREATE (u)-[r:friend {id: ''f3''}]->(v)
                     RETURN r',
                    hydrate := false
                 )",
                None,
                &[],
            )
            .expect("gql relationship create from tx node failed");
        let created_edge_rows = client
            .select(
                "SELECT count(*)::bigint
                 FROM public.graph_test_friendships_pgtest
                 WHERE id = 'f3' AND user_id = 'u3' AND friend_id = 'u1'",
                None,
                &[],
            )
            .expect("created edge source count failed")
            .first()
            .get::<i64>(1)
            .expect("created edge source count read failed")
            .unwrap_or_default();
        let path_rows = client
            .select(
                "SELECT count(*)::bigint
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u3''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN u, r, v',
                    hydrate := false
                 )",
                None,
                &[],
            )
            .expect("same transaction created path read failed")
            .first()
            .get::<i64>(1)
            .expect("same transaction created path count read failed")
            .unwrap_or_default();
        let status = client
            .select(
                "SELECT tx_delta_added_nodes, tx_delta_added_edges FROM graph.status()",
                None,
                &[],
            )
            .expect("status query failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            created_edge_rows,
            path_rows,
            status
                .get::<i32>(1)
                .expect("tx added nodes read failed")
                .unwrap_or_default(),
            status
                .get::<i32>(2)
                .expect("tx added edges read failed")
                .unwrap_or_default(),
        ))
    })
    .expect("transaction-created relationship verification failed");

    assert_eq!(created_edge_rows, 1);
    assert_eq!(path_rows, 1);
    assert_eq!(tx_added_nodes, 1);
    assert_eq!(tx_added_edges, 1);
}

#[pg_test]
fn gql_create_relationship_fills_registered_dynamic_label_column() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest
         ADD COLUMN rel_type text NOT NULL DEFAULT 'colleague'",
    )
    .expect("add dynamic relationship label column failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                from_table := 'graph_test_friendships_pgtest'::regclass,
                from_column := 'user_id',
                to_table := 'graph_test_users_pgtest'::regclass,
                to_column := 'friend_id',
                label := 'related_to',
                bidirectional := false,
                label_column := 'rel_type'
            )",
    )
    .expect("add dynamic friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable dynamic relationship graph failed");

    let (created_label, path_rows) = Spi::connect(|client| {
        client
            .select(
                "SELECT *
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''}), (v:graph_test_users_pgtest {id: ''u1''})
                     CREATE (u)-[r:colleague {id: ''f2''}]->(v)
                     RETURN r'
                 )",
                None,
                &[],
            )
            .expect("dynamic relationship create failed");
        let created_label = client
            .select(
                "SELECT rel_type
                 FROM public.graph_test_friendships_pgtest
                 WHERE id = 'f2' AND user_id = 'u2' AND friend_id = 'u1'",
                None,
                &[],
            )
            .expect("dynamic label source query failed")
            .first()
            .get::<String>(1)
            .expect("dynamic label read failed")
            .unwrap_or_default();
        let path_rows = client
            .select(
                "SELECT count(*)::bigint
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:colleague]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN r',
                    hydrate := false
                 )",
                None,
                &[],
            )
            .expect("dynamic label path read failed")
            .first()
            .get::<i64>(1)
            .expect("dynamic label path count read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((created_label, path_rows))
    })
    .expect("dynamic relationship create verification failed");

    assert_eq!(created_label, "colleague");
    assert_eq!(path_rows, 1);
}

#[pg_test]
fn gql_create_relationship_rejects_trigger_rewritten_graph_identity() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest
           ADD COLUMN rel_type TEXT NOT NULL DEFAULT 'friend';
         CREATE FUNCTION public.graph_gql_rewrite_edge_identity_pgtest()
         RETURNS trigger LANGUAGE plpgsql AS $$
         BEGIN
           IF NEW.id = 'rewrite-endpoint' THEN
             NEW.friend_id := 'u2';
           ELSIF NEW.id = 'rewrite-label' THEN
             NEW.rel_type := 'colleague';
           END IF;
           RETURN NEW;
         END
         $$;
         CREATE TRIGGER graph_gql_rewrite_edge_identity
           BEFORE INSERT ON public.graph_test_friendships_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_rewrite_edge_identity_pgtest();
         CREATE FUNCTION public.graph_gql_after_rewrite_edge_identity_pgtest()
         RETURNS trigger LANGUAGE plpgsql AS $$
         BEGIN
           IF NEW.id = 'after-endpoint' THEN
             UPDATE public.graph_test_friendships_pgtest
                SET friend_id = 'u2'
              WHERE id = NEW.id;
           ELSIF NEW.id = 'after-label' THEN
             UPDATE public.graph_test_friendships_pgtest
                SET rel_type = 'colleague'
              WHERE id = NEW.id;
           END IF;
           RETURN NULL;
         END
         $$;
         CREATE TRIGGER graph_gql_after_rewrite_edge_identity
           AFTER INSERT ON public.graph_test_friendships_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_after_rewrite_edge_identity_pgtest();
         SELECT graph.add_table(
           'graph_test_users_pgtest'::regclass,
           id_column := 'id',
           columns := ARRAY['name', 'age']
         );
         SELECT graph.add_edge(
           from_table := 'graph_test_friendships_pgtest'::regclass,
           from_column := 'user_id',
           to_table := 'graph_test_users_pgtest'::regclass,
           to_column := 'friend_id',
           label := 'related_to',
           bidirectional := false,
           label_column := 'rel_type'
         );
         SELECT * FROM graph.build(mode := 'mutable_overlay');",
    )
    .expect("create relationship identity trigger fixture failed");
    create_error_sqlstate_helper();

    for (edge_id, rel_type) in [
        ("rewrite-endpoint", "friend"),
        ("rewrite-label", "friend"),
        ("after-endpoint", "friend"),
        ("after-label", "friend"),
    ] {
        let query = format!(
            "SELECT * FROM graph.gql(
               'MATCH (u:graph_test_users_pgtest {{id: ''u2''}}), (v:graph_test_users_pgtest {{id: ''u1''}})
                CREATE (u)-[r:{rel_type} {{id: ''{edge_id}''}}]->(v)
                RETURN r'
             )"
        );
        let sqlstate = Spi::get_one::<String>(&format!(
            "SELECT public.graph_test_sqlstate({})",
            super::sql_literal(&query)
        ))
        .expect("capture rewritten relationship identity error failed")
        .unwrap_or_default();
        assert_eq!(sqlstate, "22000");
    }

    let (source_rows, tx_added_edges, overlay_dirty) = Spi::connect(|client| {
        let source_rows = client
            .select(
                "SELECT count(*)::bigint
                   FROM public.graph_test_friendships_pgtest
                  WHERE id IN (
                    'rewrite-endpoint', 'rewrite-label',
                    'after-endpoint', 'after-label'
                  )",
                None,
                &[],
            )?
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        let status = client
            .select(
                "SELECT tx_delta_added_edges, tx_delta_dirty FROM graph.status()",
                None,
                &[],
            )?
            .first();
        Ok::<_, pgrx::spi::Error>((
            source_rows,
            status.get::<i32>(1)?.unwrap_or_default(),
            status.get::<bool>(2)?.unwrap_or(true),
        ))
    })
    .expect("verify rewritten relationship rollback failed");
    assert_eq!(source_rows, 0);
    assert_eq!(tx_added_edges, 0);
    assert!(!overlay_dirty);
}

#[pg_test]
fn gql_create_relationship_preserves_bidirectional_registered_direction() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := true
            )",
    )
    .expect("add bidirectional friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (forward_start, forward_end, reverse_start, reverse_end) = Spi::connect(|client| {
        client
            .select(
                "SELECT *
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''}), (v:graph_test_users_pgtest {id: ''u1''})
                     CREATE (u)-[r:friend {id: ''f2''}]->(v)
                     RETURN r',
                    hydrate := false
                 )",
                None,
                &[],
            )
            .expect("bidirectional relationship create failed");
        let forward = client
            .select(
                "SELECT row #>> '{r,_start,id}', row #>> '{r,_end,id}'
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN r',
                    hydrate := false
                 )",
                None,
                &[],
            )
            .expect("forward created relationship read failed")
            .first();
        let reverse = client
            .select(
                "SELECT row #>> '{r,_start,id}', row #>> '{r,_end,id}'
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u2''}) RETURN r',
                    hydrate := false
                 )
                 WHERE row #>> '{r,_start,id}' = 'u2'
                   AND row #>> '{r,_end,id}' = 'u1'",
                None,
                &[],
            )
            .expect("reverse created relationship read failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            forward
                .get::<String>(1)
                .expect("forward start read failed")
                .unwrap_or_default(),
            forward
                .get::<String>(2)
                .expect("forward end read failed")
                .unwrap_or_default(),
            reverse
                .get::<String>(1)
                .expect("reverse start read failed")
                .unwrap_or_default(),
            reverse
                .get::<String>(2)
                .expect("reverse end read failed")
                .unwrap_or_default(),
        ))
    })
    .expect("bidirectional relationship create verification failed");

    assert_eq!((forward_start.as_str(), forward_end.as_str()), ("u2", "u1"));
    assert_eq!((reverse_start.as_str(), reverse_end.as_str()), ("u2", "u1"));
}

#[pg_test]
fn gql_create_relationship_requires_mutable_overlay_projection() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let readonly_sqlstate = sqlstate_for_error(
        "SELECT * FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest {id: ''u2''}), (v:graph_test_users_pgtest {id: ''u1''})
             CREATE (u)-[r:friend {id: ''f2''}]->(v)
             RETURN r'
         )",
    );
    let source_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest WHERE id = 'f2'",
    )
    .expect("source edge row count failed")
    .unwrap_or_default();

    assert_eq!(readonly_sqlstate.as_deref(), Some("0A000"));
    assert_eq!(source_rows, 0);
}

#[pg_test]
fn gql_traverses_transaction_created_node_entry_points() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := false
            )",
    )
    .expect("add friend edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (node_match_count, materialized_match_count, relationship_match_count, wildcard_path_count) =
        Spi::connect(|client| {
            client
                .select(
                    "SELECT *
                     FROM graph.gql(
                        'CREATE (u:graph_test_users_pgtest {id: ''u3'', name: ''Cara'', age: 29}) RETURN u',
                        hydrate := false
                     )",
                    None,
                    &[],
                )
                .expect("gql create failed");
            client
                .select(
                    "SELECT *
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u3''}), (v:graph_test_users_pgtest {id: ''u1''})
                         CREATE (u)-[r:friend {id: ''f3''}]->(v)
                         RETURN r',
                        hydrate := false
                     )",
                    None,
                    &[],
                )
                .expect("gql relationship create failed");
            let node_match_count = client
                .select(
                    "SELECT count(*)::bigint
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u3''}) RETURN u',
                        hydrate := false
                     )",
                    None,
                    &[],
                )
                .expect("node scan query failed")
                .first()
                .get::<i64>(1)
                .expect("node scan count read failed")
                .unwrap_or_default();
            let materialized_match_count = client
                .select(
                    "SELECT count(*)::bigint
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[:friend]->(v:graph_test_users_pgtest) RETURN u, v',
                        hydrate := false
                     )",
                    None,
                    &[],
                )
                .expect("materialized traversal query failed")
                .first()
                .get::<i64>(1)
                .expect("materialized traversal count read failed")
                .unwrap_or_default();
            let relationship_match_count = client
                .select(
                    "SELECT count(*)::bigint
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u3''})-[:friend]->(v:graph_test_users_pgtest {id: ''u1''}) RETURN u, v',
                        hydrate := false
                     )",
                    None,
                    &[],
                )
                .expect("transaction-created traversal query failed")
                .first()
                .get::<i64>(1)
                .expect("transaction-created traversal count read failed")
                .unwrap_or_default();
            let wildcard_path_count = client
                .select(
                    "SELECT count(*)::bigint
                     FROM graph.gql(
                        'MATCH p=(u:graph_test_users_pgtest)-[]->() RETURN p, u',
                        hydrate := false
                     )
                     WHERE row #>> '{u,_id,id}' = 'u3'",
                    None,
                    &[],
                )
                .expect("transaction-created wildcard traversal query failed")
                .first()
                .get::<i64>(1)
                .expect("transaction-created wildcard count read failed")
                .unwrap_or_default();
            Ok::<_, pgrx::spi::Error>((
                node_match_count,
                materialized_match_count,
                relationship_match_count,
                wildcard_path_count,
            ))
        })
        .expect("transaction-created traversal verification failed");

    assert_eq!(node_match_count, 1);
    assert_eq!(materialized_match_count, 1);
    assert_eq!(relationship_match_count, 1);
    assert_eq!(wildcard_path_count, 1);
}

#[pg_test]
fn gql_merge_node_inserts_then_updates_mapped_row() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (inserted_name, inserted_age, matched_name, matched_age, source_count, tx_added_nodes) =
        Spi::connect(|client| {
            let inserted = client
                .select(
                    "SELECT row #>> '{name}', (row #>> '{age}')::int
                     FROM graph.gql(
                        'MERGE (u:graph_test_users_pgtest {id: ''u3'', name: $name})
                         ON CREATE SET u.age = 31
                         ON MATCH SET u.name = ''Updated''
                         RETURN u.name AS name, u.age AS age',
                        params := '{\"name\":\"Cara\"}'::jsonb
                     )",
                    None,
                    &[],
                )
                .expect("gql merge insert failed")
                .first();
            let matched = client
                .select(
                    "SELECT row #>> '{name}', (row #>> '{age}')::int
                     FROM graph.gql(
                        'MERGE (u:graph_test_users_pgtest {id: ''u3'', name: ''Ignored''})
                         ON CREATE SET u.age = 99
                         ON MATCH SET u.name = $name
                         RETURN u.name AS name, u.age AS age',
                        params := '{\"name\":\"Caroline\"}'::jsonb
                     )",
                    None,
                    &[],
                )
                .expect("gql merge match failed")
                .first();
            let source_count = client
                .select(
                    "SELECT count(*)::bigint
                     FROM public.graph_test_users_pgtest
                     WHERE id = 'u3' AND name = 'Caroline' AND age = 31",
                    None,
                    &[],
                )
                .expect("source merge count failed")
                .first()
                .get::<i64>(1)
                .expect("source merge count read failed")
                .unwrap_or_default();
            let tx_added_nodes = client
                .select("SELECT tx_delta_added_nodes FROM graph.status()", None, &[])
                .expect("status query failed")
                .first()
                .get::<i32>(1)
                .expect("tx added node count read failed")
                .unwrap_or_default();
            Ok::<_, pgrx::spi::Error>((
                inserted
                    .get::<String>(1)
                    .expect("inserted merge name read failed")
                    .unwrap_or_default(),
                inserted
                    .get::<i32>(2)
                    .expect("inserted merge age read failed")
                    .unwrap_or_default(),
                matched
                    .get::<String>(1)
                    .expect("matched merge name read failed")
                    .unwrap_or_default(),
                matched
                    .get::<i32>(2)
                    .expect("matched merge age read failed")
                    .unwrap_or_default(),
                source_count,
                tx_added_nodes,
            ))
        })
        .expect("merge verification failed");

    assert_eq!(inserted_name, "Cara");
    assert_eq!(inserted_age, 31);
    assert_eq!(matched_name, "Caroline");
    assert_eq!(matched_age, 31);
    assert_eq!(source_count, 1);
    assert_eq!(tx_added_nodes, 1);
}

#[pg_test]
fn gql_merge_node_without_on_match_does_not_require_update_acl() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");
    Spi::run("DROP ROLE IF EXISTS graph_gql_merge_insert_only").expect("drop role failed");
    Spi::run("CREATE ROLE graph_gql_merge_insert_only").expect("create role failed");
    Spi::run(
        "GRANT USAGE ON SCHEMA graph, public TO graph_gql_merge_insert_only;
         GRANT SELECT, INSERT ON public.graph_test_users_pgtest TO graph_gql_merge_insert_only;
         REVOKE UPDATE ON public.graph_test_users_pgtest FROM graph_gql_merge_insert_only, PUBLIC",
    )
    .expect("grant merge ACL role privileges failed");
    create_error_sqlstate_helper();

    Spi::run("SET ROLE graph_gql_merge_insert_only").expect("set merge ACL role failed");
    let inserted_id = Spi::get_one::<String>(
        "SELECT row #>> '{u,_id,id}'
         FROM graph.gql(
            'MERGE (u:graph_test_users_pgtest {id: ''u3'', name: ''Cara''})
             ON CREATE SET u.age = 29
             RETURN u'
         )",
    )
    .expect("insert-only merge insert failed")
    .unwrap_or_default();
    let matched_id = Spi::get_one::<String>(
        "SELECT row #>> '{u,_id,id}'
         FROM graph.gql(
            'MERGE (u:graph_test_users_pgtest {id: ''u3'', name: ''Ignored''})
             RETURN u'
         )",
    )
    .expect("insert-only merge match failed")
    .unwrap_or_default();
    let denied_sqlstate = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sqlstate({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'MERGE (u:graph_test_users_pgtest {id: ''u3'', name: ''Ignored''})
                 ON MATCH SET u.name = ''Denied''
                 RETURN u'
             )"
        )
    ))
    .expect("merge ACL SQLSTATE capture failed");
    Spi::run("RESET ROLE").expect("reset role failed");

    assert_eq!(inserted_id, "u3");
    assert_eq!(matched_id, "u3");
    assert_eq!(denied_sqlstate.as_deref(), Some("42501"));
}

#[pg_test]
fn gql_merge_node_requires_identity_and_mutable_overlay() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let readonly_sqlstate = sqlstate_for_error(
        "SELECT * FROM graph.gql(
            'MERGE (u:graph_test_users_pgtest {id: ''u3'', name: ''Cara''}) RETURN u'
         )",
    );

    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");
    let missing_identity_sqlstate = sqlstate_for_error(
        "SELECT * FROM graph.gql(
            'MERGE (u:graph_test_users_pgtest {name: ''No Id''}) RETURN u'
         )",
    );

    assert_eq!(readonly_sqlstate.as_deref(), Some("0A000"));
    assert_eq!(missing_identity_sqlstate.as_deref(), Some("22000"));
}

#[pg_test]
fn gql_merge_node_does_not_evaluate_create_branch_on_match() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (matched_name, matched_age, source_count) = Spi::connect(|client| {
        let matched = client
            .select(
                "SELECT row #>> '{name}', (row #>> '{age}')::int
                 FROM graph.gql(
                    'MERGE (u:graph_test_users_pgtest {id: ''u1''})
                     ON CREATE SET u.age = $create_age
                     ON MATCH SET u.name = $match_name
                     RETURN u.name AS name, u.age AS age',
                    params := '{\"match_name\":\"Alice Matched\"}'::jsonb
                 )",
                None,
                &[],
            )
            .expect("gql merge match with missing create param failed")
            .first();
        let source_count = client
            .select(
                "SELECT count(*)::bigint
                 FROM public.graph_test_users_pgtest
                 WHERE id = 'u1' AND name = 'Alice Matched'",
                None,
                &[],
            )
            .expect("source merge branch count failed")
            .first()
            .get::<i64>(1)
            .expect("source merge branch count read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((
            matched
                .get::<String>(1)
                .expect("matched merge branch name read failed")
                .unwrap_or_default(),
            matched
                .get::<i32>(2)
                .expect("matched merge branch age read failed")
                .unwrap_or_default(),
            source_count,
        ))
    })
    .expect("merge branch verification failed");

    assert_eq!(matched_name, "Alice Matched");
    assert_eq!(matched_age, 37);
    assert_eq!(source_count, 1);
}

#[pg_test]
fn gql_merge_node_delta_limit_aborts_statement_before_source_insert() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.max_tx_delta_nodes = 0").expect("tighten node delta limit failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let sqlstate = sqlstate_for_error(
        "SELECT * FROM graph.gql(
            'MERGE (u:graph_test_users_pgtest {id: ''u3'', name: ''Cara'', age: 29}) RETURN u'
         )",
    );
    let source_count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_test_users_pgtest WHERE id = 'u3'",
    )
    .expect("source count failed")
    .unwrap_or_default();
    let tx_added_nodes = Spi::get_one::<i32>("SELECT tx_delta_added_nodes FROM graph.status()")
        .expect("status tx node count failed")
        .unwrap_or_default();

    Spi::run("RESET graph.max_tx_delta_nodes").expect("reset node delta limit failed");

    assert_eq!(sqlstate.as_deref(), Some("54000"));
    assert_eq!(source_count, 0);
    assert_eq!(tx_added_nodes, 0);
}

#[pg_test]
fn gql_create_node_delta_limit_aborts_statement_before_source_insert() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.max_tx_delta_nodes = 0").expect("tighten node delta limit failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let sqlstate = sqlstate_for_error(
        "SELECT * FROM graph.gql(
            'CREATE (u:graph_test_users_pgtest {id: ''u3'', name: ''Cara'', age: 29}) RETURN u'
         )",
    );
    let source_count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_test_users_pgtest WHERE id = 'u3'",
    )
    .expect("source count failed")
    .unwrap_or_default();
    let tx_added_nodes = Spi::get_one::<i32>("SELECT tx_delta_added_nodes FROM graph.status()")
        .expect("status tx node count failed")
        .unwrap_or_default();

    Spi::run("RESET graph.max_tx_delta_nodes").expect("reset node delta limit failed");

    assert_eq!(sqlstate.as_deref(), Some("54000"));
    assert_eq!(source_count, 0);
    assert_eq!(tx_added_nodes, 0);
}

#[pg_test]
fn transaction_local_delta_pressure_does_not_recommend_maintenance() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.compaction_threshold = 1").expect("tighten compaction threshold failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (tx_added_nodes, compaction_recommended, maintenance_recommended) =
        Spi::connect(|client| {
            client
                .select(
                    "SELECT * FROM graph.gql(
                        'CREATE (u:graph_test_users_pgtest {id: ''u3'', name: ''Cara'', age: 29}) RETURN u'
                     )",
                    None,
                    &[],
                )
                .expect("gql create failed");
            let status = client
                .select(
                    "SELECT tx_delta_added_nodes, compaction_recommended FROM graph.status()",
                    None,
                    &[],
                )
                .expect("status failed")
                .first();
            let health = client
                .select(
                    "SELECT maintenance_recommended FROM graph.sync_health()",
                    None,
                    &[],
                )
                .expect("sync health failed")
                .first();
            Ok::<_, pgrx::spi::Error>((
                status
                    .get::<i32>(1)
                    .expect("tx count read failed")
                    .unwrap_or_default(),
                status
                    .get::<bool>(2)
                    .expect("compaction read failed")
                    .unwrap_or(false),
                health
                    .get::<bool>(1)
                    .expect("maintenance read failed")
                    .unwrap_or(false),
            ))
        })
        .expect("tx pressure verification failed");

    Spi::run("RESET graph.compaction_threshold").expect("reset compaction threshold failed");

    assert_eq!(tx_added_nodes, 1);
    assert!(!compaction_recommended);
    assert!(!maintenance_recommended);
}

#[pg_test]
fn gql_create_node_applies_session_tenant_scope() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.enforce_tenant_scope = on").expect("enable tenant enforcement failed");
    Spi::run("SET graph.tenant_setting = 'app.graph_gql_create_tenant'")
        .expect("set tenant setting failed");
    Spi::run("DROP TABLE IF EXISTS public.graph_gql_create_tenant_pgtest CASCADE")
        .expect("drop tenant create table failed");
    Spi::run(
        "CREATE TABLE public.graph_gql_create_tenant_pgtest (
                id TEXT PRIMARY KEY,
                tenant_id TEXT NOT NULL,
                name TEXT NOT NULL
            )",
    )
    .expect("create tenant create table failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_gql_create_tenant_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name'],
                tenant_column := 'tenant_id'
            )",
    )
    .expect("add tenant create table failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable tenant graph failed");
    Spi::run("SET app.graph_gql_create_tenant = 'tenant-a'").expect("set tenant-a failed");

    let (created_id, created_tenant, source_tenant) = Spi::connect(|client| {
        let created = client
            .select(
                "SELECT
                    row #>> '{u,_id,id}',
                    row #>> '{u,tenant_id}'
                 FROM graph.gql(
                    'CREATE (u:graph_gql_create_tenant_pgtest {id: ''a3'', name: ''Leaf A''}) RETURN u'
                 )",
                None,
                &[],
            )
            .expect("tenant gql create failed")
            .first();
        let source_tenant = client
            .select(
                "SELECT tenant_id
                 FROM public.graph_gql_create_tenant_pgtest
                 WHERE id = 'a3'",
                None,
                &[],
            )
            .expect("source tenant query failed")
            .first()
            .get::<String>(1)
            .expect("source tenant read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((
            created
                .get::<String>(1)
                .expect("created id read failed")
                .unwrap_or_default(),
            created
                .get::<String>(2)
                .expect("created tenant read failed")
                .unwrap_or_default(),
            source_tenant,
        ))
    })
    .expect("tenant create verification failed");

    Spi::run("RESET app.graph_gql_create_tenant").expect("reset tenant-a failed");
    Spi::run("RESET graph.tenant_setting").expect("reset tenant setting failed");
    Spi::run("SET graph.enforce_tenant_scope = off").expect("disable tenant enforcement failed");

    assert_eq!(created_id, "a3");
    assert_eq!(created_tenant, "tenant-a");
    assert_eq!(source_tenant, "tenant-a");
}

#[pg_test]
fn gql_create_node_preserves_source_table_rls() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.enforce_tenant_scope = on").expect("enable tenant enforcement failed");
    Spi::run("SET graph.tenant_setting = 'app.graph_gql_rls_tenant'")
        .expect("set tenant setting failed");
    Spi::run("DROP ROLE IF EXISTS graph_gql_create_rls").expect("drop rls role failed");
    Spi::run("CREATE ROLE graph_gql_create_rls").expect("create rls role failed");
    Spi::run("DROP TABLE IF EXISTS public.graph_gql_create_rls_pgtest CASCADE")
        .expect("drop rls create table failed");
    Spi::run(
        "CREATE TABLE public.graph_gql_create_rls_pgtest (
                id TEXT PRIMARY KEY,
                tenant_id TEXT NOT NULL,
                name TEXT NOT NULL
            )",
    )
    .expect("create rls create table failed");
    Spi::run("ALTER TABLE public.graph_gql_create_rls_pgtest ENABLE ROW LEVEL SECURITY")
        .expect("enable source rls failed");
    Spi::run(
        "CREATE POLICY graph_gql_create_rls_insert
             ON public.graph_gql_create_rls_pgtest
             FOR INSERT
             WITH CHECK (tenant_id = 'tenant-a')",
    )
    .expect("create insert rls policy failed");
    Spi::run(
        "CREATE POLICY graph_gql_create_rls_select
             ON public.graph_gql_create_rls_pgtest
             FOR SELECT
             USING (tenant_id = 'tenant-a')",
    )
    .expect("create select rls policy failed");
    Spi::run(
        "GRANT USAGE ON SCHEMA graph TO graph_gql_create_rls;
         GRANT INSERT, SELECT ON public.graph_gql_create_rls_pgtest TO graph_gql_create_rls",
    )
    .expect("grant rls role privileges failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_gql_create_rls_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name'],
                tenant_column := 'tenant_id'
            )",
    )
    .expect("add rls create table failed");
    // This test's purpose is exercising RLS on the write path (GQL CREATE
    // routes through PostgreSQL DML, which enforces RLS WITH CHECK/USING),
    // not the topology-read boundary graph.build() gates by default.
    Spi::run("SET graph.allow_rls_tables = on").expect("acknowledge rls table build failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable rls graph failed");
    create_error_capture_helper();

    Spi::run("SET ROLE graph_gql_create_rls").expect("set rls role failed");
    Spi::run("SET app.graph_gql_rls_tenant = 'tenant-a'").expect("set tenant-a failed");
    let allowed_id = Spi::get_one::<String>(
        "SELECT row #>> '{u,_id,id}'
         FROM graph.gql(
            'CREATE (u:graph_gql_create_rls_pgtest {id: ''a4'', name: ''Allowed''}) RETURN u'
         )",
    )
    .expect("allowed rls create failed")
    .unwrap_or_default();
    Spi::run("SET app.graph_gql_rls_tenant = 'tenant-b'").expect("set tenant-b failed");
    let denied = Spi::get_one::<bool>(&format!(
        "SELECT public.graph_test_sql_raises({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'CREATE (u:graph_gql_create_rls_pgtest {id: ''b4'', name: ''Denied''}) RETURN u'
             )"
        )
    ))
    .expect("denied rls create capture failed")
    .unwrap_or(false);
    Spi::run("RESET ROLE").expect("reset rls role failed");

    let (allowed_count, denied_count) = Spi::connect(|client| {
        let allowed_count = client
            .select(
                "SELECT count(*)::bigint
                 FROM public.graph_gql_create_rls_pgtest
                 WHERE id = 'a4' AND tenant_id = 'tenant-a'",
                None,
                &[],
            )
            .expect("allowed source count failed")
            .first()
            .get::<i64>(1)
            .expect("allowed count read failed")
            .unwrap_or_default();
        let denied_count = client
            .select(
                "SELECT count(*)::bigint
                 FROM public.graph_gql_create_rls_pgtest
                 WHERE id = 'b4'",
                None,
                &[],
            )
            .expect("denied source count failed")
            .first()
            .get::<i64>(1)
            .expect("denied count read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((allowed_count, denied_count))
    })
    .expect("rls verification failed");

    Spi::run("RESET app.graph_gql_rls_tenant").expect("reset rls tenant failed");
    Spi::run("RESET graph.tenant_setting").expect("reset tenant setting failed");
    Spi::run("SET graph.enforce_tenant_scope = off").expect("disable tenant enforcement failed");
    Spi::run("RESET graph.allow_rls_tables").expect("reset allow_rls_tables failed");

    assert_eq!(allowed_id, "a4");
    assert!(denied);
    assert_eq!(allowed_count, 1);
    assert_eq!(denied_count, 0);
}

#[pg_test]
fn gql_set_property_updates_source_row_and_filter_index() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend'
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT graph.add_filter_column('graph_test_users_pgtest'::regclass, 'age')")
        .expect("add age filter failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (returned_age, source_age, filtered_neighbor_count) = Spi::connect(|client| {
        let updated = client
            .select(
                "SELECT row #>> '{age}'
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''}) SET u.age = $age RETURN u.age AS age',
                    params := '{\"age\":101}'::jsonb
                 )",
                None,
                &[],
            )
            .expect("gql set failed")
            .first()
            .get::<String>(1)
            .expect("updated age read failed")
            .unwrap_or_default();
        let source_age = client
            .select(
                "SELECT age FROM public.graph_test_users_pgtest WHERE id = 'u2'",
                None,
                &[],
            )
            .expect("source age query failed")
            .first()
            .get::<i32>(1)
            .expect("source age read failed")
            .unwrap_or_default();
        let filtered_neighbor_count = client
            .select(
                "SELECT count(*)::bigint
                 FROM graph.traverse(
                    'graph_test_users_pgtest'::regclass,
                    'u1',
                    1,
                    filter := '{\"node\":{\"where\":{\"age\":{\"gt\":100}}}}'::jsonb,
                    hydrate := false
                 )
                 WHERE node_id = 'u2'",
                None,
                &[],
            )
            .expect("filtered traverse failed")
            .first()
            .get::<i64>(1)
            .expect("filtered count read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((updated, source_age, filtered_neighbor_count))
    })
    .expect("set verification failed");

    assert_eq!(returned_age, "101");
    assert_eq!(source_age, 101);
    assert_eq!(filtered_neighbor_count, 1);
}

#[pg_test]
fn gql_node_writes_reject_trigger_rewritten_identity() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
         "CREATE TABLE public.graph_gql_identity_nodes_pgtest (
           id TEXT PRIMARY KEY,
           name TEXT NOT NULL,
           status TEXT,
           tenant_id TEXT NOT NULL DEFAULT 'tenant-a'
         );
         INSERT INTO public.graph_gql_identity_nodes_pgtest
           (id, name, status, tenant_id)
         VALUES ('set-node', 'Before set', 'active', 'tenant-a'),
                ('remove-node', 'Before remove', 'active', 'tenant-a'),
                ('merge-node', 'Before merge', 'active', 'tenant-a'),
                ('after-node', 'Before after trigger', 'active', 'tenant-a'),
                ('tenant-node', 'Before tenant trigger', 'active', 'tenant-a'),
                ('tenant-unscoped', 'Before unscoped tenant trigger', 'active', 'tenant-a'),
                ('after-remove', 'Before after remove', 'active', 'tenant-a'),
                ('after-merge', 'Before after merge', 'active', 'tenant-a'),
                ('property-node', 'Before property trigger', 'active', 'tenant-a');
         SELECT graph.add_table(
           'graph_gql_identity_nodes_pgtest'::regclass,
           id_column := 'id',
           columns := ARRAY['name', 'status'],
           tenant_column := 'tenant_id'
         );
         SELECT * FROM graph.build(mode := 'mutable_overlay');
         CREATE FUNCTION public.graph_gql_rewrite_node_identity_pgtest()
         RETURNS trigger LANGUAGE plpgsql AS $$
         BEGIN
           IF NEW.name IN ('After set', 'After merge')
              OR (NEW.id = 'remove-node' AND NEW.status IS NULL) THEN
             NEW.id := NEW.id || '-moved';
           END IF;
           RETURN NEW;
         END
         $$;
         CREATE TRIGGER graph_gql_rewrite_node_identity
           BEFORE UPDATE ON public.graph_gql_identity_nodes_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_rewrite_node_identity_pgtest();
         CREATE FUNCTION public.graph_gql_after_rewrite_node_identity_pgtest()
         RETURNS trigger LANGUAGE plpgsql AS $$
         BEGIN
           IF TG_OP = 'INSERT' AND NEW.id IN ('create-after', 'merge-after') THEN
             UPDATE public.graph_gql_identity_nodes_pgtest
                SET id = id || '-moved'
              WHERE id = NEW.id;
           ELSIF TG_OP = 'UPDATE'
              AND NEW.id = 'after-node'
              AND NEW.name = 'After statement set' THEN
             UPDATE public.graph_gql_identity_nodes_pgtest
                SET id = id || '-moved'
              WHERE id = NEW.id;
           ELSIF TG_OP = 'UPDATE'
              AND NEW.id = 'tenant-node'
              AND NEW.name = 'After tenant set'
              AND NEW.tenant_id = 'tenant-a' THEN
             UPDATE public.graph_gql_identity_nodes_pgtest
                SET tenant_id = 'tenant-b'
              WHERE id = NEW.id;
           ELSIF TG_OP = 'UPDATE'
              AND NEW.id = 'property-node'
              AND NEW.name = 'Trigger input' THEN
             UPDATE public.graph_gql_identity_nodes_pgtest
                SET name = 'Trigger output'
              WHERE id = NEW.id;
           ELSIF TG_OP = 'UPDATE'
              AND NEW.id = 'after-remove'
              AND NEW.status IS NULL THEN
             UPDATE public.graph_gql_identity_nodes_pgtest
                SET id = id || '-moved'
              WHERE id = NEW.id;
           ELSIF TG_OP = 'UPDATE'
              AND NEW.id = 'after-merge'
              AND NEW.name = 'After merge statement' THEN
             UPDATE public.graph_gql_identity_nodes_pgtest
                SET id = id || '-moved'
              WHERE id = NEW.id;
           ELSIF TG_OP = 'UPDATE'
              AND NEW.id = 'tenant-unscoped'
              AND NEW.name = 'After unscoped tenant set'
              AND NEW.tenant_id = 'tenant-a' THEN
             UPDATE public.graph_gql_identity_nodes_pgtest
                SET tenant_id = 'tenant-b'
              WHERE id = NEW.id;
           END IF;
           RETURN NULL;
         END
         $$;
         CREATE TRIGGER graph_gql_after_insert_node_identity
           AFTER INSERT ON public.graph_gql_identity_nodes_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_after_rewrite_node_identity_pgtest();
         CREATE TRIGGER graph_gql_after_update_node_identity
           AFTER UPDATE ON public.graph_gql_identity_nodes_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_after_rewrite_node_identity_pgtest();",
    )
    .expect("create node identity trigger fixture failed");
    Spi::run("SET graph.enforce_tenant_scope = on")
        .expect("enable tenant enforcement failed");
    Spi::run("SET graph.tenant_setting = 'app.graph_gql_identity_tenant'")
        .expect("set identity tenant setting failed");
    Spi::run("SET app.graph_gql_identity_tenant = 'tenant-a'")
        .expect("set identity tenant value failed");
    create_error_sqlstate_helper();
    create_error_message_helper();

    for query in [
        "MATCH (u:graph_gql_identity_nodes_pgtest {id: 'set-node'}) SET u.name = 'After set' RETURN u",
        "MATCH (u:graph_gql_identity_nodes_pgtest {id: 'remove-node'}) REMOVE u.status RETURN u",
        "MERGE (u:graph_gql_identity_nodes_pgtest {id: 'merge-node'}) ON MATCH SET u.name = 'After merge' RETURN u",
        "MATCH (u:graph_gql_identity_nodes_pgtest {id: 'after-node'}) SET u.name = 'After statement set' RETURN u",
        "MATCH (u:graph_gql_identity_nodes_pgtest {id: 'tenant-node'}) SET u.name = 'After tenant set' RETURN u",
        "MATCH (u:graph_gql_identity_nodes_pgtest {id: 'after-remove'}) REMOVE u.status RETURN u",
        "MERGE (u:graph_gql_identity_nodes_pgtest {id: 'after-merge'}) ON MATCH SET u.name = 'After merge statement' RETURN u",
        "CREATE (u:graph_gql_identity_nodes_pgtest {id: 'create-after', name: 'Created', status: 'active'}) RETURN u",
        "MERGE (u:graph_gql_identity_nodes_pgtest {id: 'merge-after', name: 'Merged', status: 'active'}) RETURN u",
    ] {
        let statement = format!(
            "SELECT * FROM graph.gql({})",
            super::sql_literal(query)
        );
        let sqlstate = Spi::get_one::<String>(&format!(
            "SELECT public.graph_test_sqlstate({})",
            super::sql_literal(&statement)
        ))
        .expect("capture rewritten node identity error failed")
        .unwrap_or_default();
        let message = Spi::get_one::<String>(&format!(
            "SELECT public.graph_test_sql_error_message({})",
            super::sql_literal(&statement)
        ))
        .expect("capture rewritten node identity message failed")
        .unwrap_or_default();
        assert_eq!(sqlstate, "22000", "query={query}; message={message}");
    }

    Spi::run(
        "RESET app.graph_gql_identity_tenant;
         SET graph.tenant_setting = '';
         SET graph.enforce_tenant_scope = off",
    )
    .expect("disable identity tenant scope failed");
    let unscoped_query =
        "MATCH (u:graph_gql_identity_nodes_pgtest {id: 'tenant-unscoped'}) SET u.name = 'After unscoped tenant set' RETURN u";
    let unscoped_statement = format!(
        "SELECT * FROM graph.gql({})",
        super::sql_literal(unscoped_query)
    );
    let unscoped_sqlstate = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sqlstate({})",
        super::sql_literal(&unscoped_statement)
    ))
    .expect("capture unscoped tenant identity error failed")
    .unwrap_or_default();
    let unscoped_message = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sql_error_message({})",
        super::sql_literal(&unscoped_statement)
    ))
    .expect("capture unscoped tenant identity message failed")
    .unwrap_or_default();
    assert_eq!(
        unscoped_sqlstate, "22000",
        "query={unscoped_query}; message={unscoped_message}"
    );

    let (returned_name, source_name) = Spi::connect(|client| {
        let returned_name = client
            .select(
                "SELECT row #>> '{name}'
                   FROM graph.gql(
                     'MATCH (u:graph_gql_identity_nodes_pgtest {id: ''property-node''})
                      SET u.name = ''Trigger input''
                      RETURN u.name AS name'
                   )",
                None,
                &[],
            )?
            .first()
            .get::<String>(1)?
            .unwrap_or_default();
        let source_name = client
            .select(
                "SELECT name
                   FROM public.graph_gql_identity_nodes_pgtest
                  WHERE id = 'property-node'",
                None,
                &[],
            )?
            .first()
            .get::<String>(1)?
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((returned_name, source_name))
    })
    .expect("verify post-trigger property result failed");
    assert_eq!(returned_name, "Trigger output");
    assert_eq!(source_name, "Trigger output");

    let unchanged_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
           FROM public.graph_gql_identity_nodes_pgtest
          WHERE tenant_id = 'tenant-a'
            AND ((id = 'set-node' AND name = 'Before set' AND status = 'active')
              OR (id = 'remove-node' AND name = 'Before remove' AND status = 'active')
              OR (id = 'merge-node' AND name = 'Before merge' AND status = 'active')
              OR (id = 'after-node' AND name = 'Before after trigger' AND status = 'active')
              OR (id = 'tenant-node' AND name = 'Before tenant trigger' AND status = 'active')
              OR (id = 'tenant-unscoped' AND name = 'Before unscoped tenant trigger' AND status = 'active')
              OR (id = 'after-remove' AND name = 'Before after remove' AND status = 'active')
              OR (id = 'after-merge' AND name = 'Before after merge' AND status = 'active'))",
    )
    .expect("verify trigger-rewritten node rollback failed")
    .unwrap_or_default();
    let moved_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
           FROM public.graph_gql_identity_nodes_pgtest
          WHERE id LIKE '%-moved'
             OR id IN ('create-after', 'merge-after')
             OR tenant_id <> 'tenant-a'",
    )
    .expect("count trigger-rewritten node identities failed")
    .unwrap_or_default();
    assert_eq!(unchanged_rows, 8);
    assert_eq!(moved_rows, 0);
}

#[pg_test]
fn gql_remove_typed_property_sets_source_column_null_idempotently() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("ALTER TABLE public.graph_test_users_pgtest ADD COLUMN status TEXT")
        .expect("add status column failed");
    Spi::run("UPDATE public.graph_test_users_pgtest SET status = 'active'")
        .expect("seed status failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age', 'status']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend'
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (first_is_null, second_is_null, source_is_null) = Spi::connect(|client| {
        let first = client
            .select(
                "SELECT row->'status' = 'null'::jsonb
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''}) REMOVE u.status RETURN u.status AS status'
                 )",
                None,
                &[],
            )
            .expect("first remove failed")
            .first()
            .get::<bool>(1)
            .expect("first null read failed")
            .unwrap_or(false);
        let second = client
            .select(
                "SELECT row->'status' = 'null'::jsonb
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''}) REMOVE u.status RETURN u.status AS status'
                 )",
                None,
                &[],
            )
            .expect("second remove failed")
            .first()
            .get::<bool>(1)
            .expect("second null read failed")
            .unwrap_or(false);
        let source = client
            .select(
                "SELECT status IS NULL FROM public.graph_test_users_pgtest WHERE id = 'u2'",
                None,
                &[],
            )
            .expect("source status query failed")
            .first()
            .get::<bool>(1)
            .expect("source status read failed")
            .unwrap_or(false);
        Ok::<_, pgrx::spi::Error>((first, second, source))
    })
    .expect("remove typed verification failed");

    assert!(first_is_null);
    assert!(second_is_null);
    assert!(source_is_null);
}

#[pg_test]
fn gql_remove_jsonb_property_path_drops_key_idempotently() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("ALTER TABLE public.graph_test_users_pgtest ADD COLUMN profile jsonb")
    .expect("add profile column failed");
    Spi::run(
        "UPDATE public.graph_test_users_pgtest
         SET profile = CASE id
           WHEN 'u1' THEN '{\"plan\":\"pro\",\"flags\":{\"beta\":true}}'::jsonb
           ELSE NULL
         END",
    )
    .expect("seed profile failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age', 'profile', 'profile.plan', 'profile.flags']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend'
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (first_plan_null, second_plan_null, profile, source_has_plan, null_root_preserved) =
        Spi::connect(|client| {
        let first = client
            .select(
                "SELECT row->'plan' = 'null'::jsonb
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u1''}) REMOVE u.profile.plan RETURN u.profile.plan AS plan'
                 )",
                None,
                &[],
            )
            .expect("first jsonb remove failed")
            .first()
            .get::<bool>(1)
            .expect("first plan null read failed")
            .unwrap_or(false);
        let second_row = client
            .select(
                "SELECT row->'plan' = 'null'::jsonb, row->'profile'
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u1''}) REMOVE u.profile.plan RETURN u.profile.plan AS plan, u.profile AS profile'
                 )",
                None,
                &[],
            )
            .expect("second jsonb remove failed")
            .first();
        let second = second_row
            .get::<bool>(1)
            .expect("second plan null read failed")
            .unwrap_or(false);
        let profile = second_row
            .get::<pgrx::JsonB>(2)
            .expect("profile read failed")
            .unwrap();
        let source_has_plan = client
            .select(
                "SELECT profile ? 'plan' FROM public.graph_test_users_pgtest WHERE id = 'u1'",
                None,
                &[],
            )
            .expect("source profile query failed")
            .first()
            .get::<bool>(1)
            .expect("source profile read failed")
            .unwrap_or(true);
        let null_root_preserved = client
            .select(
                "SELECT row->'profile' = 'null'::jsonb
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''}) REMOVE u.profile.plan RETURN u.profile AS profile'
                 )",
                None,
                &[],
            )
            .expect("null-root jsonb remove failed")
            .first()
            .get::<bool>(1)
            .expect("null-root profile read failed")
            .unwrap_or(false);
        Ok::<_, pgrx::spi::Error>((first, second, profile, source_has_plan, null_root_preserved))
    })
    .expect("remove jsonb verification failed");

    assert!(first_plan_null);
    assert!(second_plan_null);
    assert_eq!(profile.0, serde_json::json!({"flags": {"beta": true}}));
    assert!(!source_has_plan);
    assert!(null_root_preserved);
}

#[pg_test]
fn gql_remove_property_requires_mutable_overlay_projection() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    create_error_capture_helper();

    let denied = Spi::get_one::<bool>(&format!(
        "SELECT public.graph_test_sql_raises({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest {id: ''u2''}) REMOVE u.age RETURN u.age'
             )"
        )
    ))
    .expect("readonly remove capture failed")
    .unwrap_or(false);
    let source_age = Spi::get_one::<i32>(
        "SELECT age FROM public.graph_test_users_pgtest WHERE id = 'u2'",
    )
    .expect("source age query failed")
    .unwrap_or_default();

    assert!(denied);
    assert_eq!(source_age, 41);
}

#[pg_test]
fn gql_set_property_readonly_projection_reports_sqlstate() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    build_friendship_fixture_graph();

    let readonly_sqlstate = sqlstate_for_error(
        "SELECT * FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest {id: ''u2''}) SET u.age = 42 RETURN u.age'
             )",
    );

    assert_eq!(readonly_sqlstate.as_deref(), Some("0A000"));
}

#[pg_test]
fn gql_set_property_type_mismatch_does_not_mutate_source_row() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    build_friendship_fixture_graph();
    create_error_capture_helper();
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let type_denied = Spi::get_one::<bool>(&format!(
        "SELECT public.graph_test_sql_raises({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest {id: ''u2''}) SET u.age = ''not numeric'' RETURN u.age'
             )"
        )
    ))
    .expect("type mismatch capture failed")
    .unwrap_or(false);
    let source_age = Spi::get_one::<i32>(
        "SELECT age FROM public.graph_test_users_pgtest WHERE id = 'u2'",
    )
    .expect("source age query failed")
    .unwrap_or_default();

    assert!(type_denied);
    assert_eq!(source_age, 41);
}

#[pg_test]
fn gql_set_property_rejects_registered_tenant_column() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET LOCAL graph.enforce_tenant_scope = on")
        .expect("enable tenant enforcement failed");
    Spi::run("SET LOCAL graph.tenant_setting = 'app.graph_gql_set_tenant'")
        .expect("set tenant GUC failed");
    Spi::run("SET LOCAL app.graph_gql_set_tenant = 'tenant-a'").expect("set tenant failed");
    Spi::run("DROP TABLE IF EXISTS public.graph_gql_set_tenant_pgtest CASCADE")
        .expect("drop tenant set table failed");
    Spi::run(
        "CREATE TABLE public.graph_gql_set_tenant_pgtest (
                id TEXT PRIMARY KEY,
                tenant_id TEXT NOT NULL,
                name TEXT NOT NULL
            )",
    )
    .expect("create tenant set table failed");
    Spi::run(
        "INSERT INTO public.graph_gql_set_tenant_pgtest (id, tenant_id, name)
         VALUES ('a1', 'tenant-a', 'Ada')",
    )
    .expect("insert tenant set row failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_gql_set_tenant_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['tenant_id', 'name'],
                tenant_column := 'tenant_id'
            )",
    )
    .expect("add tenant set table failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable tenant graph failed");
    create_error_capture_helper();

    let denied = Spi::get_one::<bool>(&format!(
        "SELECT public.graph_test_sql_raises({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'MATCH (u:graph_gql_set_tenant_pgtest {id: ''a1''}) SET u.tenant_id = ''tenant-b'' RETURN u'
             )"
        )
    ))
    .expect("tenant column set capture failed")
    .unwrap_or(false);
    let tenant = Spi::get_one::<String>(
        "SELECT tenant_id FROM public.graph_gql_set_tenant_pgtest WHERE id = 'a1'",
    )
    .expect("tenant value query failed")
    .unwrap_or_default();

    Spi::run("RESET app.graph_gql_set_tenant").expect("reset tenant failed");
    Spi::run("RESET graph.tenant_setting").expect("reset tenant setting failed");
    Spi::run("SET graph.enforce_tenant_scope = off").expect("disable tenant enforcement failed");

    assert!(denied);
    assert_eq!(tenant, "tenant-a");
}

#[pg_test]
fn gql_delete_edge_removes_source_row_and_tombstones_neighbors() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := true
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (returned_source, returned_target, edge_rows, forward_count, reverse_count, user_count) =
        Spi::connect(|client| {
            let deleted = client
                .select(
                    "SELECT row #>> '{source}', row #>> '{target}'
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u2''}) DELETE r RETURN u.id AS source, v.id AS target'
                     )",
                    None,
                    &[],
                )
                .expect("gql delete failed")
                .first();
            let edge_rows = client
                .select(
                    "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest",
                    None,
                    &[],
                )
                .expect("edge row count failed")
                .first()
                .get::<i64>(1)
                .expect("edge count read failed")
                .unwrap_or_default();
            let forward_count = client
                .select(
                    "SELECT count(*)::bigint
                     FROM graph.traverse(
                        'graph_test_users_pgtest'::regclass,
                        'u1',
                        1,
                        edge_types := ARRAY['friend'],
                        hydrate := false
                     )
                     WHERE node_id = 'u2'",
                    None,
                    &[],
                )
                .expect("forward traverse failed")
                .first()
                .get::<i64>(1)
                .expect("forward count read failed")
                .unwrap_or_default();
            let reverse_count = client
                .select(
                    "SELECT count(*)::bigint
                     FROM graph.traverse(
                        'graph_test_users_pgtest'::regclass,
                        'u2',
                        1,
                        edge_types := ARRAY['friend'],
                        hydrate := false
                     )
                     WHERE node_id = 'u1'",
                    None,
                    &[],
                )
                .expect("reverse traverse failed")
                .first()
                .get::<i64>(1)
                .expect("reverse count read failed")
                .unwrap_or_default();
            let user_count = client
                .select(
                    "SELECT count(*)::bigint FROM public.graph_test_users_pgtest",
                    None,
                    &[],
                )
                .expect("user count failed")
                .first()
                .get::<i64>(1)
                .expect("user count read failed")
                .unwrap_or_default();
            Ok::<_, pgrx::spi::Error>((
                deleted
                    .get::<String>(1)
                    .expect("source read failed")
                    .unwrap_or_default(),
                deleted
                    .get::<String>(2)
                    .expect("target read failed")
                    .unwrap_or_default(),
                edge_rows,
                forward_count,
                reverse_count,
                user_count,
            ))
        })
        .expect("delete verification failed");

    assert_eq!(returned_source, "u1");
    assert_eq!(returned_target, "u2");
    assert_eq!(edge_rows, 0);
    assert_eq!(forward_count, 0);
    assert_eq!(reverse_count, 0);
    assert_eq!(user_count, 2);
}

#[pg_test]
fn synced_parallel_relationship_delete_preserves_sibling_through_reload() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f_parallel', 'u1', 'u2')",
    )
    .expect("insert parallel relationship failed");
    Spi::run("SET graph.sync_mode = 'trigger'").expect("set trigger sync failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persisted build failed");
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.query_freshness = 'off'").expect("disable auto sync failed");
    Spi::run("SELECT graph.add_table('graph_test_users_pgtest'::regclass, 'id', ARRAY['name'])")
        .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
            'graph_test_friendships_pgtest'::regclass,
            'user_id',
            'graph_test_users_pgtest'::regclass,
            'friend_id',
            'friend',
            bidirectional := false
        )",
    )
    .expect("add friendship mapping failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    Spi::run("DELETE FROM public.graph_test_friendships_pgtest WHERE id = 'f1'")
        .expect("delete one parallel relationship failed");

    let assert_only_sibling = || {
        let ids = Spi::get_one::<String>(
            "SELECT string_agg(row #>> '{r,id}', ',' ORDER BY row #>> '{r,id}')
               FROM graph.gql(
                 'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u2''}) RETURN r'
               )",
        )
        .expect("parallel sibling query failed")
        .unwrap_or_default();
        assert_eq!(ids, "f_parallel");
    };
    let published = Spi::get_one::<i64>(
        "SELECT segments_published FROM graph.ingest_projection()",
    )
    .expect("ingest identity delete failed")
    .unwrap_or_default();
    assert!(published > 0);
    Spi::run("SET graph.auto_load = on").expect("enable auto-load failed");
    super::ENGINE.with(|engine| *engine.borrow_mut() = super::engine::Engine::new());
    assert_only_sibling();

    let source_rows = Spi::get_one::<String>(
        "SELECT string_agg(id, ',' ORDER BY id)
           FROM public.graph_test_friendships_pgtest
          WHERE user_id = 'u1' AND friend_id = 'u2'",
    )
    .expect("source relationship rows read failed")
    .unwrap_or_default();
    assert_eq!(source_rows, "f_parallel");
    Spi::run("SET graph.auto_load = off").expect("restore auto-load failed");
    Spi::run("SET graph.persist_on_build = off").expect("restore persisted build failed");
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
    Spi::run("SET graph.sync_mode = 'manual'").expect("restore sync mode failed");
}

#[pg_test]
fn gql_delete_relationship_with_composite_source_key() {
    reset_and_create_fixtures();
    Spi::run(
        "CREATE TABLE public.graph_gql_composite_nodes_pgtest (
             id TEXT PRIMARY KEY
         );
         CREATE TABLE public.graph_gql_composite_edges_pgtest (
             edge_id TEXT NOT NULL,
             revision INTEGER NOT NULL,
             source_id TEXT NOT NULL REFERENCES public.graph_gql_composite_nodes_pgtest(id),
             target_id TEXT NOT NULL REFERENCES public.graph_gql_composite_nodes_pgtest(id),
             PRIMARY KEY (edge_id, revision)
         );
         INSERT INTO public.graph_gql_composite_nodes_pgtest VALUES ('n1'), ('n2');
         INSERT INTO public.graph_gql_composite_edges_pgtest
         VALUES ('edge', 1, 'n1', 'n2');",
    )
    .expect("create composite relationship fixture failed");
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table('graph_gql_composite_nodes_pgtest'::regclass, 'id');
         SELECT graph.add_edge(
           'graph_gql_composite_edges_pgtest'::regclass,
           'source_id',
           'graph_gql_composite_nodes_pgtest'::regclass,
           'target_id',
           'linked',
           bidirectional := false
         );
         SELECT * FROM graph.build(mode := 'mutable_overlay');",
    )
    .expect("build composite relationship graph failed");

    Spi::run(
        "SELECT * FROM graph.gql(
           'MATCH (a:graph_gql_composite_nodes_pgtest {id: ''n1''})-[r:linked]->(b:graph_gql_composite_nodes_pgtest {id: ''n2''}) DELETE r RETURN a.id AS source'
         )",
    )
    .expect("delete composite-key relationship failed");
    let rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_gql_composite_edges_pgtest",
    )
    .expect("composite relationship count failed")
    .unwrap_or(-1);
    assert_eq!(rows, 0);
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
}

#[pg_test]
fn gql_delete_rls_denial_keeps_source_row_and_projection() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SELECT graph.add_table('graph_test_users_pgtest'::regclass, 'id', ARRAY['name'])")
        .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
           'graph_test_friendships_pgtest'::regclass,
           'user_id',
           'graph_test_users_pgtest'::regclass,
           'friend_id',
           'friend',
           bidirectional := false
         )",
    )
    .expect("add relationship mapping failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");
    Spi::run("DROP ROLE IF EXISTS graph_gql_delete_rls").expect("drop delete role failed");
    Spi::run("CREATE ROLE graph_gql_delete_rls").expect("create delete role failed");
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_gql_delete_select
           ON public.graph_test_friendships_pgtest FOR SELECT
           TO graph_gql_delete_rls USING (true);
         CREATE POLICY graph_gql_delete_deny
           ON public.graph_test_friendships_pgtest FOR DELETE
           TO graph_gql_delete_rls USING (false);
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_delete_rls;
         GRANT SELECT ON public.graph_test_users_pgtest TO graph_gql_delete_rls;
         GRANT SELECT, DELETE ON public.graph_test_friendships_pgtest TO graph_gql_delete_rls;",
    )
    .expect("configure delete RLS fixture failed");
    create_error_capture_helper();

    Spi::run("SET ROLE graph_gql_delete_rls").expect("set delete role failed");
    let denied = Spi::get_one::<bool>(&format!(
        "SELECT public.graph_test_sql_raises({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
               'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u2''}) DELETE r RETURN u.id AS source'
             )"
        )
    ))
    .expect("capture denied delete failed")
    .unwrap_or(false);
    Spi::run("RESET ROLE").expect("reset delete role failed");

    let source_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest WHERE id = 'f1'",
    )
    .expect("source relationship count failed")
    .unwrap_or_default();
    let projected = Spi::get_one::<i64>(
        "SELECT count(*)::bigint
           FROM graph.gql(
             'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[:friend]->(v:graph_test_users_pgtest {id: ''u2''}) RETURN u, v'
           )",
    )
    .expect("projection relationship count failed")
    .unwrap_or_default();
    assert!(denied);
    assert_eq!(source_rows, 1);
    assert_eq!(projected, 1);
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
}

#[pg_test]
fn gql_delete_edge_write_recheck_rejects_stale_endpoint_predicate() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := true
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT graph.add_filter_column('graph_test_users_pgtest'::regclass, 'age')")
        .expect("add age filter column failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");
    Spi::run("UPDATE public.graph_test_users_pgtest SET age = 99 WHERE id = 'u2'")
        .expect("stale endpoint update failed");

    let sqlstate = sqlstate_for_error(
        "SELECT graph._test_recheck_delete_edge_predicate(
             'graph_test_users_pgtest'::regclass,
             'u1',
             'graph_test_users_pgtest'::regclass,
             'u2',
             'friend',
             'age',
             41
         )",
    );
    let edge_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest",
    )
    .expect("edge row count failed")
    .unwrap_or_default();

    assert_eq!(sqlstate.as_deref(), Some("22000"));
    assert_eq!(edge_rows, 1);
}

#[pg_test]
fn gql_wildcard_delete_edge_resolves_unique_mapped_row() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := true
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (returned_source, returned_target, edge_rows) = Spi::connect(|client| {
        let deleted = client
            .select(
                "SELECT row #>> '{source}', row #>> '{target}'
                 FROM graph.gql(
                    'MATCH (u)-[r]->(v)
                     WHERE u.id = ''u1'' AND v.id = ''u2''
                     DELETE r
                     RETURN u.id AS source, v.id AS target'
                 )",
                None,
                &[],
            )
            .expect("wildcard gql delete failed")
            .first();
        let edge_rows = client
            .select(
                "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest",
                None,
                &[],
            )
            .expect("edge row count failed")
            .first()
            .get::<i64>(1)
            .expect("edge count read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((
            deleted
                .get::<String>(1)
                .expect("source read failed")
                .unwrap_or_default(),
            deleted
                .get::<String>(2)
                .expect("target read failed")
                .unwrap_or_default(),
            edge_rows,
        ))
    })
    .expect("wildcard delete verification failed");

    assert_eq!(returned_source, "u1");
    assert_eq!(returned_target, "u2");
    assert_eq!(edge_rows, 0);
}

#[pg_test]
fn gql_delete_dynamic_edge_label_deletes_only_matching_label_row() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest
         ADD COLUMN rel_type text NOT NULL DEFAULT 'colleague'",
    )
    .expect("add dynamic relationship label column failed");
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id, rel_type)
         VALUES ('f2', 'u1', 'u2', 'mentor')",
    )
    .expect("insert second dynamic relationship failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                from_table := 'graph_test_friendships_pgtest'::regclass,
                from_column := 'user_id',
                to_table := 'graph_test_users_pgtest'::regclass,
                to_column := 'friend_id',
                label := 'related_to',
                bidirectional := false,
                label_column := 'rel_type'
            )",
    )
    .expect("add dynamic friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable dynamic relationship graph failed");

    let (returned_source, returned_target, colleague_rows, mentor_rows) =
        Spi::connect(|client| {
            let deleted = client
                .select(
                    "SELECT row #>> '{source}', row #>> '{target}'
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[r:colleague]->(v:graph_test_users_pgtest {id: ''u2''}) DELETE r RETURN u.id AS source, v.id AS target'
                     )",
                    None,
                    &[],
                )
                .expect("dynamic label delete failed")
                .first();
            let colleague_rows = client
                .select(
                    "SELECT count(*)::bigint
                     FROM public.graph_test_friendships_pgtest
                     WHERE rel_type = 'colleague'",
                    None,
                    &[],
                )
                .expect("colleague row count failed")
                .first()
                .get::<i64>(1)
                .expect("colleague count read failed")
                .unwrap_or_default();
            let mentor_rows = client
                .select(
                    "SELECT count(*)::bigint
                     FROM public.graph_test_friendships_pgtest
                     WHERE rel_type = 'mentor'",
                    None,
                    &[],
                )
                .expect("mentor row count failed")
                .first()
                .get::<i64>(1)
                .expect("mentor count read failed")
                .unwrap_or_default();
            Ok::<_, pgrx::spi::Error>((
                deleted
                    .get::<String>(1)
                    .expect("source read failed")
                    .unwrap_or_default(),
                deleted
                    .get::<String>(2)
                    .expect("target read failed")
                    .unwrap_or_default(),
                colleague_rows,
                mentor_rows,
            ))
        })
        .expect("dynamic label delete verification failed");

    assert_eq!(returned_source, "u1");
    assert_eq!(returned_target, "u2");
    assert_eq!(colleague_rows, 0);
    assert_eq!(mentor_rows, 1);
}

#[pg_test]
fn gql_delete_dynamic_edge_label_uses_registered_label_for_blank_values() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("ALTER TABLE public.graph_test_friendships_pgtest ADD COLUMN rel_type text")
        .expect("add nullable dynamic relationship label column failed");
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id, rel_type)
         VALUES ('f2', 'u2', 'u1', '')",
    )
    .expect("insert blank dynamic relationship failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                from_table := 'graph_test_friendships_pgtest'::regclass,
                from_column := 'user_id',
                to_table := 'graph_test_users_pgtest'::regclass,
                to_column := 'friend_id',
                label := 'related_to',
                bidirectional := false,
                label_column := 'rel_type'
            )",
    )
    .expect("add dynamic friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable dynamic relationship graph failed");

    let (returned_source, returned_target, remaining_edges) = Spi::connect(|client| {
        let deleted = client
            .select(
                "SELECT row #>> '{source}', row #>> '{target}'
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[r:related_to]->(v:graph_test_users_pgtest {id: ''u2''}) DELETE r RETURN u.id AS source, v.id AS target'
                 )",
                None,
                &[],
            )
            .expect("dynamic fallback label delete failed")
            .first();
        let remaining_edges = client
            .select(
                "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest",
                None,
                &[],
            )
            .expect("remaining edge count failed")
            .first()
            .get::<i64>(1)
            .expect("remaining edge count read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((
            deleted
                .get::<String>(1)
                .expect("source read failed")
                .unwrap_or_default(),
            deleted
                .get::<String>(2)
                .expect("target read failed")
                .unwrap_or_default(),
            remaining_edges,
        ))
    })
    .expect("dynamic fallback label delete verification failed");

    assert_eq!(returned_source, "u1");
    assert_eq!(returned_target, "u2");
    assert_eq!(remaining_edges, 1);
}

#[pg_test]
fn gql_delete_edge_delta_limit_aborts_before_source_delete() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.max_tx_delta_edges = 0").expect("tighten edge delta limit failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := true
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let sqlstate = sqlstate_for_error(
        "SELECT * FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u2''}) DELETE r RETURN u'
         )",
    );
    let edge_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest",
    )
    .expect("edge row count failed")
    .unwrap_or_default();
    let tx_deleted_edges = Spi::get_one::<i32>("SELECT tx_delta_deleted_edges FROM graph.status()")
        .expect("status tx edge count failed")
        .unwrap_or_default();

    Spi::run("RESET graph.max_tx_delta_edges").expect("reset edge delta limit failed");

    assert_eq!(sqlstate.as_deref(), Some("54000"));
    assert_eq!(edge_rows, 1);
    assert_eq!(tx_deleted_edges, 0);
}

#[pg_test]
fn gql_detach_delete_removes_incident_edges_before_node() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f2', 'u2', 'u1')",
    )
    .expect("insert reverse friendship failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend'
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (returned_name, edge_rows, deleted_user_rows, remaining_user_rows, gql_deleted_visible) =
        Spi::connect(|client| {
            let deleted = client
                .select(
                    "SELECT row #>> '{name}'
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u1''}) DETACH DELETE u RETURN u.name AS name'
                     )",
                    None,
                    &[],
                )
                .expect("gql detach delete failed")
                .first();
            let edge_rows = client
                .select(
                    "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest",
                    None,
                    &[],
                )
                .expect("edge row count failed")
                .first()
                .get::<i64>(1)
                .expect("edge count read failed")
                .unwrap_or_default();
            let deleted_user_rows = client
                .select(
                    "SELECT count(*)::bigint FROM public.graph_test_users_pgtest WHERE id = 'u1'",
                    None,
                    &[],
                )
                .expect("deleted user count failed")
                .first()
                .get::<i64>(1)
                .expect("deleted user read failed")
                .unwrap_or_default();
            let remaining_user_rows = client
                .select(
                    "SELECT count(*)::bigint FROM public.graph_test_users_pgtest WHERE id = 'u2'",
                    None,
                    &[],
                )
                .expect("remaining user count failed")
                .first()
                .get::<i64>(1)
                .expect("remaining user read failed")
                .unwrap_or_default();
            let gql_deleted_visible = client
                .select(
                    "SELECT count(*)::bigint
                     FROM graph.gql(
                        'MATCH (u:graph_test_users_pgtest {id: ''u1''}) RETURN u',
                        hydrate := false
                     )",
                    None,
                    &[],
                )
                .expect("gql deleted visibility check failed")
                .first()
                .get::<i64>(1)
                .expect("gql deleted visibility read failed")
                .unwrap_or_default();
            Ok::<_, pgrx::spi::Error>((
                deleted
                    .get::<String>(1)
                    .expect("deleted name read failed")
                    .unwrap_or_default(),
                edge_rows,
                deleted_user_rows,
                remaining_user_rows,
                gql_deleted_visible,
            ))
        })
        .expect("detach delete verification failed");
    let traverse_deleted_sqlstate = sqlstate_for_error(
        "SELECT *
         FROM graph.traverse(
            'graph_test_users_pgtest'::regclass,
            'u1',
            1,
            edge_types := ARRAY['friend'],
            hydrate := false
         )",
    )
    .unwrap_or_default();

    assert_eq!(returned_name, "Alice");
    assert_eq!(edge_rows, 0);
    assert_eq!(deleted_user_rows, 0);
    assert_eq!(remaining_user_rows, 1);
    assert_eq!(gql_deleted_visible, 0);
    assert_eq!(traverse_deleted_sqlstate, "P0002");
}

#[pg_test]
fn gql_detach_delete_delta_limit_rolls_back_source_rows() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.max_tx_delta_nodes = 0").expect("tighten node delta limit failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend'
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let sqlstate = sqlstate_for_error(
        "SELECT * FROM graph.gql(
            'MATCH (u:graph_test_users_pgtest {id: ''u1''}) DETACH DELETE u RETURN u'
         )",
    );
    let user_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_test_users_pgtest WHERE id = 'u1'",
    )
    .expect("user row count failed")
    .unwrap_or_default();
    let edge_rows = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest",
    )
    .expect("edge row count failed")
    .unwrap_or_default();

    Spi::run("RESET graph.max_tx_delta_nodes").expect("reset node delta limit failed");

    assert_eq!(sqlstate.as_deref(), Some("54000"));
    assert_eq!(user_rows, 1);
    assert_eq!(edge_rows, 1);
}

#[pg_test]
fn gql_partitioned_detach_delete_rolls_back_source_and_projection_on_trigger_error() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "CREATE TABLE public.graph_gql_detach_nodes_pgtest (
             id TEXT PRIMARY KEY,
             name TEXT NOT NULL
         ) PARTITION BY HASH (id);
         CREATE TABLE public.graph_gql_detach_nodes_p0_pgtest
           PARTITION OF public.graph_gql_detach_nodes_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 0);
         CREATE TABLE public.graph_gql_detach_nodes_p1_pgtest
           PARTITION OF public.graph_gql_detach_nodes_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 1);
         CREATE TABLE public.graph_gql_detach_edges_pgtest (
             edge_id TEXT NOT NULL,
             revision INTEGER NOT NULL,
             source_id TEXT NOT NULL REFERENCES public.graph_gql_detach_nodes_pgtest(id),
             target_id TEXT NOT NULL REFERENCES public.graph_gql_detach_nodes_pgtest(id),
             PRIMARY KEY (edge_id, revision)
         ) PARTITION BY HASH (revision);
         CREATE TABLE public.graph_gql_detach_edges_p0_pgtest
           PARTITION OF public.graph_gql_detach_edges_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 0);
         CREATE TABLE public.graph_gql_detach_edges_p1_pgtest
           PARTITION OF public.graph_gql_detach_edges_pgtest
           FOR VALUES WITH (MODULUS 2, REMAINDER 1);
         INSERT INTO public.graph_gql_detach_nodes_pgtest VALUES
           ('protected', 'Protected'), ('survivor', 'Survivor'), ('deletable', 'Deletable');
         INSERT INTO public.graph_gql_detach_edges_pgtest VALUES
           ('protected-edge', 1, 'protected', 'survivor'),
           ('deletable-edge', 2, 'deletable', 'survivor');
         CREATE FUNCTION public.graph_gql_detach_guard_pgtest()
         RETURNS trigger LANGUAGE plpgsql AS $$
         BEGIN
           IF OLD.id = 'protected' THEN
             RAISE EXCEPTION 'detach trigger rejected row';
           END IF;
           RETURN OLD;
         END
         $$;
         CREATE TRIGGER graph_gql_detach_guard
           BEFORE DELETE ON public.graph_gql_detach_nodes_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_detach_guard_pgtest();
         SELECT graph.add_table(
           'graph_gql_detach_nodes_pgtest'::regclass,
           id_column := 'id',
           columns := ARRAY['name']
         );
         SELECT graph.add_edge(
           'graph_gql_detach_edges_pgtest'::regclass,
           'source_id',
           'graph_gql_detach_nodes_pgtest'::regclass,
           'target_id',
           'linked',
           bidirectional := false
         );
         SELECT * FROM graph.build(mode := 'mutable_overlay');",
    )
    .expect("create partitioned detach fixture failed");
    create_error_sqlstate_helper();

    let rejected_detach =
        "SELECT * FROM graph.gql(
           'MATCH (u:graph_gql_detach_nodes_pgtest {id: ''protected''}) DETACH DELETE u RETURN u'
         )";
    let sqlstate = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sqlstate({})",
        super::sql_literal(rejected_detach)
    ))
    .expect("capture rejected partitioned detach failed")
    .unwrap_or_default();
    assert_eq!(sqlstate, "P0001");
    let (node_rows, edge_rows, projected_edge, tx_dirty, deleted_nodes, deleted_edges) =
        Spi::connect(|client| {
            let node_rows = client
                .select(
                    "SELECT count(*)::bigint FROM public.graph_gql_detach_nodes_pgtest",
                    None,
                    &[],
                )?
                .first()
                .get::<i64>(1)?
                .unwrap_or_default();
            let edge_rows = client
                .select(
                    "SELECT count(*)::bigint FROM public.graph_gql_detach_edges_pgtest",
                    None,
                    &[],
                )?
                .first()
                .get::<i64>(1)?
                .unwrap_or_default();
            let projected_edge = client
                .select(
                    "SELECT count(*)::bigint
                       FROM graph.traverse(
                         'graph_gql_detach_nodes_pgtest'::regclass,
                         'protected',
                         1,
                         edge_types := ARRAY['linked'],
                         hydrate := false
                       )
                      WHERE node_id = 'survivor'",
                    None,
                    &[],
                )?
                .first()
                .get::<i64>(1)?
                .unwrap_or_default();
            let status = client
                .select(
                    "SELECT tx_delta_dirty, tx_delta_deleted_nodes, tx_delta_deleted_edges
                       FROM graph.status()",
                    None,
                    &[],
                )?
                .first();
            Ok::<_, pgrx::spi::Error>((
                node_rows,
                edge_rows,
                projected_edge,
                status.get::<bool>(1)?.unwrap_or(true),
                status.get::<i32>(2)?.unwrap_or_default(),
                status.get::<i32>(3)?.unwrap_or_default(),
            ))
        })
        .expect("verify rejected partitioned detach failed");
    assert_eq!(node_rows, 3);
    assert_eq!(edge_rows, 2);
    assert_eq!(projected_edge, 1);
    assert!(!tx_dirty);
    assert_eq!(deleted_nodes, 0);
    assert_eq!(deleted_edges, 0);

    let (returned_id, remaining_node, remaining_edge, node_partition, edge_partition) =
        Spi::connect(|client| {
            let node_partition = client
                .select(
                    "SELECT tableoid::regclass::text
                       FROM public.graph_gql_detach_nodes_pgtest
                      WHERE id = 'deletable'",
                    None,
                    &[],
                )?
                .first()
                .get::<String>(1)?
                .unwrap_or_default();
            let edge_partition = client
                .select(
                    "SELECT tableoid::regclass::text
                       FROM public.graph_gql_detach_edges_pgtest
                      WHERE edge_id = 'deletable-edge'",
                    None,
                    &[],
                )?
                .first()
                .get::<String>(1)?
                .unwrap_or_default();
            let returned_id = client
                .select(
                    "SELECT row #>> '{id}'
                       FROM graph.gql(
                         'MATCH (u:graph_gql_detach_nodes_pgtest {id: ''deletable''}) DETACH DELETE u RETURN u.id AS id'
                       )",
                    None,
                    &[],
                )?
                .first()
                .get::<String>(1)?
                .unwrap_or_default();
            let remaining_node = client
                .select(
                    "SELECT count(*)::bigint
                       FROM public.graph_gql_detach_nodes_pgtest
                      WHERE id = 'deletable'",
                    None,
                    &[],
                )?
                .first()
                .get::<i64>(1)?
                .unwrap_or_default();
            let remaining_edge = client
                .select(
                    "SELECT count(*)::bigint
                       FROM public.graph_gql_detach_edges_pgtest
                      WHERE edge_id = 'deletable-edge'",
                    None,
                    &[],
                )?
                .first()
                .get::<i64>(1)?
                .unwrap_or_default();
            Ok::<_, pgrx::spi::Error>((
                returned_id,
                remaining_node,
                remaining_edge,
                node_partition,
                edge_partition,
            ))
        })
        .expect("verify accepted partitioned detach failed");
    assert_eq!(returned_id, "deletable");
    assert_eq!(remaining_node, 0);
    assert_eq!(remaining_edge, 0);
    assert!(node_partition.contains("graph_gql_detach_nodes_p"));
    assert!(edge_partition.contains("graph_gql_detach_edges_p"));
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
}

#[pg_test]
fn gql_delete_edge_handles_bidirectional_reverse_match() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := true
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");

    let (edge_rows, forward_count, reverse_count) = Spi::connect(|client| {
        client
            .select(
                "SELECT row
                 FROM graph.gql(
                    'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) DELETE r RETURN u, v'
                 )",
                None,
                &[],
            )
            .expect("reverse gql delete failed");
        let edge_rows = client
            .select(
                "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest",
                None,
                &[],
            )
            .expect("edge row count failed")
            .first()
            .get::<i64>(1)
            .expect("edge count read failed")
            .unwrap_or_default();
        let forward_count = client
            .select(
                "SELECT count(*)::bigint
                 FROM graph.traverse(
                    'graph_test_users_pgtest'::regclass,
                    'u1',
                    1,
                    edge_types := ARRAY['friend'],
                    hydrate := false
                 )
                 WHERE node_id = 'u2'",
                None,
                &[],
            )
            .expect("forward traverse failed")
            .first()
            .get::<i64>(1)
            .expect("forward count read failed")
            .unwrap_or_default();
        let reverse_count = client
            .select(
                "SELECT count(*)::bigint
                 FROM graph.traverse(
                    'graph_test_users_pgtest'::regclass,
                    'u2',
                    1,
                    edge_types := ARRAY['friend'],
                    hydrate := false
                 )
                 WHERE node_id = 'u1'",
                None,
                &[],
            )
            .expect("reverse traverse failed")
            .first()
            .get::<i64>(1)
            .expect("reverse count read failed")
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((edge_rows, forward_count, reverse_count))
    })
    .expect("reverse delete verification failed");

    assert_eq!(edge_rows, 0);
    assert_eq!(forward_count, 0);
    assert_eq!(reverse_count, 0);
}

#[pg_test]
fn gql_bidirectional_reverse_match_returns_registered_edge_direction() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := true
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build()").expect("build graph failed");

    let (
        rel_start,
        rel_end,
        path_node_0,
        path_node_1,
        path_rel_start,
        path_rel_end,
        relationships_start,
        relationships_end,
    ) = Spi::connect(|client| {
        let row = client
            .select(
                "SELECT COALESCE(row #>> '{r,_start,_id,id}', row #>> '{r,_start,id}'),
                        COALESCE(row #>> '{r,_end,_id,id}', row #>> '{r,_end,id}'),
                        COALESCE(row #>> '{p,_path,nodes,0,_id,id}', row #>> '{p,_path,nodes,0,id}'),
                        COALESCE(row #>> '{p,_path,nodes,1,_id,id}', row #>> '{p,_path,nodes,1,id}'),
                        COALESCE(row #>> '{p,_path,relationships,0,_start,_id,id}', row #>> '{p,_path,relationships,0,_start,id}'),
                        COALESCE(row #>> '{p,_path,relationships,0,_end,_id,id}', row #>> '{p,_path,relationships,0,_end,id}'),
                        COALESCE(row #>> '{rs,0,_start,_id,id}', row #>> '{rs,0,_start,id}'),
                        COALESCE(row #>> '{rs,0,_end,_id,id}', row #>> '{rs,0,_end,id}')
                 FROM graph.gql(
                    'MATCH p=()<-[r:friend]-()
                     RETURN r, p, relationships(p) AS rs',
                    hydrate := false
                 )
                 LIMIT 1",
                None,
                &[],
            )
            .expect("reverse bidirectional GQL query failed")
            .first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<String>(1)
                .expect("rel start read failed")
                .expect("rel start was NULL"),
            row.get::<String>(2)
                .expect("rel end read failed")
                .expect("rel end was NULL"),
            row.get::<String>(3)
                .expect("path node 0 read failed")
                .expect("path node 0 was NULL"),
            row.get::<String>(4)
                .expect("path node 1 read failed")
                .expect("path node 1 was NULL"),
            row.get::<String>(5)
                .expect("path rel start read failed")
                .expect("path rel start was NULL"),
            row.get::<String>(6)
                .expect("path rel end read failed")
                .expect("path rel end was NULL"),
            row.get::<String>(7)
                .expect("relationships start read failed")
                .expect("relationships start was NULL"),
            row.get::<String>(8)
                .expect("relationships end read failed")
                .expect("relationships end was NULL"),
        ))
    })
    .expect("reverse bidirectional GQL SPI failed");

    assert_eq!(rel_start, "u1");
    assert_eq!(rel_end, "u2");
    assert_eq!(path_node_0, "u1");
    assert_eq!(path_node_1, "u2");
    assert_eq!(path_rel_start, "u1");
    assert_eq!(path_rel_end, "u2");
    assert_eq!(relationships_start, "u1");
    assert_eq!(relationships_end, "u2");
}

#[pg_test]
fn gql_delete_edge_rejects_ambiguous_bidirectional_self_edge_rows() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f2', 'u2', 'u1')",
    )
    .expect("insert opposite friendship failed");
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'age']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend',
                bidirectional := true
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build mutable graph failed");
    create_error_capture_helper();

    let denied = Spi::get_one::<bool>(&format!(
        "SELECT public.graph_test_sql_raises({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest {id: ''u2''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u1''}) DELETE r RETURN u, v'
             )"
        )
    ))
    .expect("ambiguous delete capture failed")
    .unwrap_or(false);
    let edge_count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest",
    )
    .expect("edge count query failed")
    .unwrap_or_default();

    assert!(denied);
    assert_eq!(edge_count, 2);
}

#[pg_test]
fn gql_delete_edge_requires_mutable_overlay_projection() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    create_error_capture_helper();

    let denied = Spi::get_one::<bool>(&format!(
        "SELECT public.graph_test_sql_raises({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[r:friend]->(v:graph_test_users_pgtest {id: ''u2''}) DELETE r RETURN u, v'
             )"
        )
    ))
    .expect("readonly delete capture failed")
    .unwrap_or(false);
    let edge_count = Spi::get_one::<i64>(
        "SELECT count(*)::bigint FROM public.graph_test_friendships_pgtest",
    )
    .expect("edge count query failed")
    .unwrap_or_default();

    assert!(denied);
    assert_eq!(edge_count, 1);
}

#[pg_test]
fn gql_create_node_rejects_unregistered_label() {
    reset_and_create_fixtures();
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    build_friendship_fixture_graph();
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("rebuild mutable graph failed");
    create_error_capture_helper();

    let denied = Spi::get_one::<bool>(&format!(
        "SELECT public.graph_test_sql_raises({})",
        super::sql_literal(
            "SELECT * FROM graph.gql(
                'CREATE (m:graph_missing_pgtest {id: ''m1'', name: ''Missing''}) RETURN m'
             )"
        )
    ))
    .expect("unregistered label error capture query failed")
    .unwrap_or(false);

    assert!(denied);
}

#[cfg(feature = "development")]
fn build_p46_gql_rls_fixture(mutable: bool) {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name, age)
         VALUES ('u3', 'Hidden node', 43), ('u4', 'Hidden edge target', 47);
         INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f_hidden_node', 'u1', 'u3'),
                ('f_hidden_edge', 'u1', 'u4')",
    )
    .expect("create P4.6 topology fixture failed");
    if mutable {
        Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    }
    Spi::run(
        "SELECT graph.add_table(
             'graph_test_users_pgtest'::regclass,
             id_column := 'id',
             columns := ARRAY['name', 'age']
         );
         SELECT graph.add_edge(
             'graph_test_friendships_pgtest'::regclass,
             'user_id',
             'graph_test_users_pgtest'::regclass,
             'friend_id',
             'friend',
             bidirectional := false
         )",
    )
    .expect("register P4.6 fixture failed");
    if mutable {
        Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
            .expect("build mutable P4.6 fixture failed");
    } else {
        Spi::run("SELECT * FROM graph.build()")
            .expect("build read-only P4.6 fixture failed");
    }
    Spi::run(
        "DROP ROLE IF EXISTS graph_gql_p46_reader;
         CREATE ROLE graph_gql_p46_reader;
         ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         ALTER TABLE public.graph_test_friendships_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_gql_p46_nodes
           ON public.graph_test_users_pgtest
           TO graph_gql_p46_reader
           USING (id <> 'u3')
           WITH CHECK (id <> 'u3');
         CREATE POLICY graph_gql_p46_edges
           ON public.graph_test_friendships_pgtest
           FOR SELECT TO graph_gql_p46_reader
           USING (id <> 'f_hidden_edge');
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_p46_reader;
         GRANT SELECT, UPDATE ON public.graph_test_users_pgtest TO graph_gql_p46_reader;
         GRANT SELECT ON public.graph_test_friendships_pgtest TO graph_gql_p46_reader",
    )
    .expect("configure P4.6 real-role RLS fixture failed");
}

#[cfg(feature = "development")]
fn p46_graph_query_rows(surface: &str, query: &str) -> serde_json::Value {
    Spi::get_one::<pgrx::JsonB>(&format!(
        "SELECT COALESCE(jsonb_agg(row ORDER BY row::text), '[]'::jsonb)
           FROM graph.{surface}({}, hydrate := false)",
        super::sql_literal(query)
    ))
    .unwrap_or_else(|error| panic!("{surface} P4.6 query failed: {error}"))
    .expect("P4.6 query aggregate was NULL")
    .0
}

#[cfg(feature = "development")]
fn p46_visibility_metrics() -> serde_json::Value {
    Spi::get_one::<pgrx::JsonB>("SELECT graph._test_visibility_metrics()")
        .expect("read P4.6 visibility metrics failed")
        .expect("P4.6 visibility metrics were NULL")
        .0
}

#[cfg(feature = "development")]
fn p46_captured_sqlstate(statement: &str) -> Option<String> {
    Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sqlstate({})",
        super::sql_literal(statement)
    ))
    .expect("capture P4.6 SQLSTATE failed")
}

#[cfg(feature = "development")]
#[pg_test]
fn gql_identity_bounded_expansion_lazy_matches_eager_rls_optional_multipattern_order_and_caps() {
    build_p46_gql_rls_fixture(false);
    let queries = [
        "MATCH (u:graph_test_users_pgtest {id: 'u1'})-[r:friend]->(v:graph_test_users_pgtest)
         RETURN u.id AS source, r, v.id AS target ORDER BY target",
        "OPTIONAL MATCH (u:graph_test_users_pgtest {id: 'u1'})<-[r:friend]-(v:graph_test_users_pgtest)
         RETURN u.id AS source, r, v.id AS target ORDER BY target LIMIT 1",
        "MATCH (u:graph_test_users_pgtest)-[r:friend]->(v:graph_test_users_pgtest),
               (u)-[s:friend]->(w:graph_test_users_pgtest)
         WHERE u.id = 'u1'
         RETURN v.id AS target, w.id AS peer, r, s
         ORDER BY target, peer LIMIT 1",
    ];

    Spi::run(
        "SET ROLE graph_gql_p46_reader;
         SELECT graph._test_set_visibility_strategy('eager')",
    )
    .expect("force eager P4.6 GQL failed");
    let eager = queries
        .iter()
        .map(|query| p46_graph_query_rows("gql", query))
        .collect::<Vec<_>>();
    Spi::run("SELECT graph._test_set_visibility_strategy('lazy')")
        .expect("force lazy P4.6 GQL failed");
    let mut lazy = Vec::with_capacity(queries.len());
    let mut selected = Vec::with_capacity(queries.len());
    let mut read_recheck_calls = Vec::with_capacity(queries.len());
    for query in queries {
        lazy.push(p46_graph_query_rows("gql", query));
        let metrics = p46_visibility_metrics();
        selected.push(
            metrics["selected_strategy"]
                .as_str()
                .unwrap_or_default()
                .to_string(),
        );
        read_recheck_calls.push(
            metrics["gql_read_recheck_calls"]
                .as_u64()
                .unwrap_or_default(),
        );
    }
    let metrics = p46_visibility_metrics();
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')")
        .expect("restore P4.6 GQL strategy failed");

    assert_eq!(lazy, eager);
    assert_eq!(lazy[0].as_array().map(Vec::len), Some(1));
    assert_eq!(lazy[1].as_array().map(Vec::len), Some(1));
    assert_eq!(lazy[2].as_array().map(Vec::len), Some(1));
    assert_eq!(selected, ["lazy", "lazy", "eager"]);
    assert_eq!(read_recheck_calls[..2], [0, 0]);
    assert!(read_recheck_calls[2] > 0);
    assert!(metrics["spi_calls"].as_u64().unwrap_or_default() > 0);
}

#[cfg(feature = "development")]
#[pg_test]
fn gql_identity_bounded_write_match_lazy_uses_postgres_rls_and_matches_eager() {
    build_p46_gql_rls_fixture(true);
    create_error_sqlstate_helper();
    Spi::run(
        "SET ROLE graph_gql_p46_reader;
         SELECT graph._test_set_visibility_strategy('eager')",
    )
    .expect("force eager P4.6 write failed");
    let eager = p46_graph_query_rows(
        "gql",
        "MATCH (u:graph_test_users_pgtest {id: 'u1'}) SET u.age = 38 RETURN u.id AS id, u.age AS age",
    );
    Spi::run(
        "RESET ROLE;
         UPDATE public.graph_test_users_pgtest SET age = 37 WHERE id = 'u1';
         SET ROLE graph_gql_p46_reader;
         SELECT graph._test_set_visibility_strategy('lazy')",
    )
    .expect("prepare lazy P4.6 write failed");
    let lazy = p46_graph_query_rows(
        "cypher",
        "MATCH (u:graph_test_users_pgtest {id: 'u1'}) SET u.age = 38 RETURN u.id AS id, u.age AS age",
    );
    let metrics = p46_visibility_metrics();
    let hidden_statement = "SELECT * FROM graph.gql(
        'MATCH (u:graph_test_users_pgtest {id: ''u3''}) SET u.age = 99 RETURN u')";
    let hidden_state = Spi::get_one::<String>(&format!(
        "SELECT public.graph_test_sqlstate({})",
        super::sql_literal(hidden_statement)
    ))
    .expect("capture hidden P4.6 write failed");
    let hidden_age = Spi::get_one::<i32>(
        "RESET ROLE;
         SELECT age FROM public.graph_test_users_pgtest WHERE id = 'u3'",
    )
    .expect("read hidden P4.6 source row failed")
    .unwrap_or_default();
    Spi::run(
        "CREATE OR REPLACE FUNCTION public.graph_gql_p46_reject_age()
         RETURNS trigger LANGUAGE plpgsql AS $$
         BEGIN
           IF NEW.age = 39 THEN RAISE division_by_zero; END IF;
           RETURN NEW;
         END
         $$;
         CREATE TRIGGER graph_gql_p46_reject_age_trigger
           BEFORE UPDATE ON public.graph_test_users_pgtest
           FOR EACH ROW EXECUTE FUNCTION public.graph_gql_p46_reject_age();
         SET ROLE graph_gql_p46_reader",
    )
    .expect("configure P4.6 write trigger error failed");
    let trigger_statement = "SELECT * FROM graph.gql(
        'MATCH (u:graph_test_users_pgtest {id: ''u1''}) SET u.age = 39 RETURN u.id AS id')";
    let trigger_state = p46_captured_sqlstate(trigger_statement);
    let retry = p46_graph_query_rows(
        "gql",
        "MATCH (u:graph_test_users_pgtest {id: 'u1'}) SET u.age = 40 RETURN u.id AS id, u.age AS age",
    );
    Spi::run("SELECT graph._test_set_visibility_strategy('auto')")
        .expect("restore P4.6 write strategy failed");

    assert_eq!(lazy, eager);
    assert!(hidden_state.is_some(), "RLS-hidden write MATCH must not select a row");
    assert_eq!(hidden_age, 43);
    assert_eq!(trigger_state.as_deref(), Some("22012"));
    assert_eq!(retry.as_array().map(Vec::len), Some(1));
    assert_eq!(metrics["selected_strategy"].as_str(), Some("lazy"));
}

#[cfg(feature = "development")]
#[pg_test]
fn gql_identity_bounded_lazy_cancellation_policy_error_drop_state_then_retry() {
    build_p46_gql_rls_fixture(false);
    create_error_sqlstate_helper();
    let statement = "SELECT * FROM graph.gql(
        'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[:friend]->(v:graph_test_users_pgtest)
         RETURN v.id AS id')";
    Spi::run(
        "SET ROLE graph_gql_p46_reader;
         SELECT graph._test_set_visibility_strategy('lazy');
         SELECT graph._test_arm_lazy_visibility_cancel()",
    )
    .expect("arm P4.6 lazy cancellation failed");
    let cancelled = workflow_cancellation(statement);
    let empty_after_cancel =
        Spi::get_one::<bool>("SELECT graph._test_visibility_resolution_state_empty()")
            .expect("inspect P4.6 cancellation state failed")
            .unwrap_or(false);
    // If the red implementation never entered the lazy resolver, consume the
    // one-shot hook through an already-migrated API so it cannot leak to the
    // next PostgreSQL test in this backend.
    let _ = workflow_cancellation(
        "SELECT * FROM graph.traverse(
           'graph_test_users_pgtest'::regclass, 'u1', 1, hydrate := false)",
    );

    Spi::run(
        "RESET ROLE;
         CREATE OR REPLACE FUNCTION public.graph_gql_p46_policy_guard()
         RETURNS boolean LANGUAGE plpgsql AS $$
         BEGIN
           IF current_setting('graph.test_p46_policy_error', true) = 'on' THEN
             RAISE division_by_zero;
           END IF;
           RETURN true;
         END
         $$;
         DROP POLICY graph_gql_p46_nodes ON public.graph_test_users_pgtest;
         CREATE POLICY graph_gql_p46_nodes
           ON public.graph_test_users_pgtest
           TO graph_gql_p46_reader
           USING (id <> 'u3' AND public.graph_gql_p46_policy_guard());
         GRANT EXECUTE ON FUNCTION public.graph_gql_p46_policy_guard()
           TO graph_gql_p46_reader;
         SET graph.test_p46_policy_error = 'on';
         SET ROLE graph_gql_p46_reader",
    )
    .expect("configure P4.6 policy error failed");
    let policy_state = p46_captured_sqlstate(statement);
    let empty_after_error =
        Spi::get_one::<bool>("SELECT graph._test_visibility_resolution_state_empty()")
            .expect("inspect P4.6 policy error state failed")
            .unwrap_or(false);
    Spi::run("SET graph.test_p46_policy_error = 'off'")
        .expect("disable P4.6 policy error failed");
    let retry = p46_graph_query_rows(
        "gql",
        "MATCH (u:graph_test_users_pgtest {id: 'u1'})-[:friend]->(v:graph_test_users_pgtest)
         RETURN v.id AS id",
    );
    let metrics = p46_visibility_metrics();
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')")
        .expect("restore P4.6 cancellation strategy failed");

    assert!(cancelled);
    assert!(empty_after_cancel);
    assert_eq!(policy_state.as_deref(), Some("22012"));
    assert!(empty_after_error);
    assert_eq!(retry.as_array().map(Vec::len), Some(1));
    assert_eq!(metrics["selected_strategy"].as_str(), Some("lazy"));
}

#[cfg(feature = "development")]
#[pg_test]
fn gql_identity_bounded_acl_denial_precedes_null_missing_and_invalid_identity_values() {
    build_p46_gql_rls_fixture(false);
    create_error_sqlstate_helper();
    Spi::run(
        "DROP ROLE IF EXISTS graph_gql_p46_denied;
         CREATE ROLE graph_gql_p46_denied;
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_p46_denied;
         SET ROLE graph_gql_p46_denied;
         SELECT graph._test_set_visibility_strategy('lazy')",
    )
    .expect("configure denied P4.6 role failed");
    let statements = [
        "SELECT * FROM graph.gql(
           'MATCH (u:graph_test_users_pgtest {id: $id}) RETURN u.id AS id',
           '{\"id\":null}'::jsonb)",
        "SELECT * FROM graph.gql(
           'MATCH (u:graph_test_users_pgtest {id: $id}) RETURN u.id AS id',
           '{}'::jsonb)",
        "SELECT * FROM graph.gql(
           'MATCH (u:graph_test_users_pgtest {id: $id}) RETURN u.id AS id',
           '{\"id\":{\"bad\":true}}'::jsonb)",
    ];
    let states = statements
        .iter()
        .map(|statement| p46_captured_sqlstate(statement))
        .collect::<Vec<_>>();
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')")
        .expect("restore denied P4.6 role failed");
    assert_eq!(states, vec![Some("42501".into()); 3]);
}

#[cfg(feature = "development")]
#[pg_test]
fn gql_whole_source_scan_remains_eager_under_forced_lazy() {
    build_p46_gql_rls_fixture(false);
    Spi::run(
        "SET ROLE graph_gql_p46_reader;
         SELECT graph._test_set_visibility_strategy('lazy')",
    )
    .expect("force P4.6 whole-source strategy failed");
    let rows = p46_graph_query_rows(
        "gql",
        "MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest)
         RETURN u.id AS source, v.id AS target ORDER BY source, target",
    );
    let metrics = p46_visibility_metrics();
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')")
        .expect("restore P4.6 whole-source strategy failed");

    assert_eq!(rows.as_array().map(Vec::len), Some(1));
    assert_eq!(metrics["selected_strategy"].as_str(), Some("eager"));
    assert!(metrics["source_rows"].as_u64().unwrap_or_default() > 0);
}

#[cfg(feature = "development")]
#[pg_test]
fn gql_no_rls_identity_bounded_fast_path_has_zero_resolver_spi() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    create_error_sqlstate_helper();
    Spi::run("SELECT graph._test_set_visibility_strategy('auto')")
        .expect("set P4.6 no-RLS strategy failed");
    let rows = p46_graph_query_rows(
        "gql",
        "MATCH (u:graph_test_users_pgtest {id: 'u1'})-[:friend]->(v:graph_test_users_pgtest)
         RETURN v.id AS id",
    );
    let metrics = p46_visibility_metrics();
    Spi::run("DELETE FROM public.graph_test_friendships_pgtest WHERE id = 'f1'")
        .expect("delete no-RLS relationship source row failed");
    let stale_state = p46_captured_sqlstate(
        "SELECT * FROM graph.gql(
           'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[:friend]->(v:graph_test_users_pgtest)
            RETURN v.id AS id')",
    );

    assert_eq!(rows.as_array().map(Vec::len), Some(1));
    assert_eq!(stale_state.as_deref(), Some("22000"));
    assert_eq!(metrics["selected_strategy"].as_str(), Some("lazy"));
    assert_eq!(metrics["spi_calls"].as_u64(), Some(0));
    assert_eq!(metrics["requested_keys"].as_u64(), Some(0));
    assert_eq!(metrics["gql_read_recheck_calls"].as_u64(), Some(1));
    assert!(metrics["gql_read_recheck_rows"].as_u64().unwrap_or_default() > 0);
}

#[cfg(feature = "development")]
#[pg_test]
fn gql_identity_filtered_join_eager_fallback_preserves_result_row_cap() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name, age)
         SELECT 'cap-' || i::text, 'cap', i
           FROM generate_series(1, 101) AS i;
         INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         SELECT 'cap-edge-' || i::text, 'u1', 'cap-' || i::text
           FROM generate_series(1, 101) AS i",
    )
    .expect("create P4.6 row-cap fixture failed");
    build_friendship_fixture_graph();
    create_error_sqlstate_helper();
    Spi::run(
        "DROP ROLE IF EXISTS graph_gql_p46_cap_reader;
         CREATE ROLE graph_gql_p46_cap_reader;
         ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_gql_p46_cap_nodes
           ON public.graph_test_users_pgtest TO graph_gql_p46_cap_reader
           USING (true);
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_p46_cap_reader;
         GRANT SELECT ON public.graph_test_users_pgtest,
                         public.graph_test_friendships_pgtest
           TO graph_gql_p46_cap_reader;
         SET ROLE graph_gql_p46_cap_reader",
    )
    .expect("configure P4.6 row-cap role failed");
    let statement = "SELECT * FROM graph.gql(
        'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest),
               (u)-[:friend]->(w:graph_test_users_pgtest)
         WHERE u.id = ''u1''
         RETURN v.id AS target, w.id AS peer')";
    Spi::run("SELECT graph._test_set_visibility_strategy('eager')")
        .expect("force eager P4.6 row-cap strategy failed");
    let eager_state = p46_captured_sqlstate(statement);
    Spi::run("SELECT graph._test_set_visibility_strategy('lazy')")
        .expect("force lazy P4.6 row-cap strategy failed");
    let lazy_state = p46_captured_sqlstate(statement);
    let metrics = p46_visibility_metrics();
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')")
        .expect("restore P4.6 row-cap strategy failed");

    assert!(eager_state.is_some(), "eager GQL must enforce the result row cap");
    assert_eq!(lazy_state, eager_state);
    assert_eq!(metrics["selected_strategy"].as_str(), Some("eager"));
}

#[cfg(feature = "development")]
#[pg_test]
fn gql_identity_bounded_parallel_edge_after_distinct_node_cap_matches_eager_error() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name, age)
         SELECT 'p46-cap-' || lpad(i::text, 5, '0'), 'cap', i
           FROM generate_series(1, 10000) AS i;
         INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         SELECT 'p46-edge-' || lpad(i::text, 5, '0'), 'u1',
                'p46-cap-' || lpad(i::text, 5, '0')
           FROM generate_series(1, 10000) AS i;
         INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('p46-edge-zz-parallel', 'u1', 'p46-cap-10000')",
    )
    .expect("create P4.6 cap/parallel fixture failed");
    build_friendship_fixture_graph();
    create_error_sqlstate_helper();
    Spi::run(
        "DROP ROLE IF EXISTS graph_gql_p46_cap_parallel;
         CREATE ROLE graph_gql_p46_cap_parallel;
         ALTER TABLE public.graph_test_users_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_gql_p46_cap_parallel_nodes
           ON public.graph_test_users_pgtest TO graph_gql_p46_cap_parallel USING (true);
         ALTER TABLE public.graph_test_friendships_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_gql_p46_cap_parallel_edges
           ON public.graph_test_friendships_pgtest TO graph_gql_p46_cap_parallel USING (true);
         GRANT USAGE ON SCHEMA graph, public TO graph_gql_p46_cap_parallel;
         GRANT SELECT ON public.graph_test_users_pgtest,
                         public.graph_test_friendships_pgtest
           TO graph_gql_p46_cap_parallel;
         SET ROLE graph_gql_p46_cap_parallel",
    )
    .expect("configure P4.6 cap/parallel role failed");
    let statement = "SELECT * FROM graph.gql(
        'MATCH (u:graph_test_users_pgtest {id: ''u1''})-[:friend]->(v:graph_test_users_pgtest)
         RETURN v.id AS id')";
    Spi::run("SELECT graph._test_set_visibility_strategy('eager')")
        .expect("force eager P4.6 cap/parallel strategy failed");
    let eager_state = p46_captured_sqlstate(statement);
    Spi::run("SELECT graph._test_set_visibility_strategy('lazy')")
        .expect("force lazy P4.6 cap/parallel strategy failed");
    let lazy_state = p46_captured_sqlstate(statement);
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')")
        .expect("restore P4.6 cap/parallel strategy failed");
    assert!(eager_state.is_some());
    assert_eq!(lazy_state, eager_state);
}
