Method execute_graph_algo

nodedb/src/data/executor/handlers/graph_algo.rs:24–73

(
        &self,
        task: &ExecutionTask,
        tid: u64,
        algorithm: &GraphAlgorithm,
        params: &AlgoParams,
    )

impl CoreLoop {
    pub(in crate::data::executor) fn execute_graph_algo(
        &self,
        task: &ExecutionTask,
        tid: u64,
        algorithm: &GraphAlgorithm,
        params: &AlgoParams,
    ) -> Response {
        debug!(
            core = self.core_id,
            tid,
            algorithm = algorithm.name(),
            collection = %params.collection,
            edge_label = ?params.edge_label,
            "graph algorithm dispatch"
        );

        if *algorithm == GraphAlgorithm::Sssp && params.source_node.is_none() {
            return self.response_error(
                task,
                ErrorCode::Internal {
                    detail: "SSSP requires FROM '<source_node>'".into(),
                },
            );
        }

        let scoped_csr = match build_csr_for_collection(
            &self.edge_store,
            tid,
            &params.collection,
            params.edge_label.as_deref(),
            None,
        ) {
            Ok(c) => c,
            Err(e) => return self.response_error(task, ErrorCode::from(e)),
        };

        if scoped_csr.node_count() == 0 {
            return match AlgoResultBatch::new(*algorithm).to_msgpack() {
                Ok(payload) => self.response_with_payload(task, payload),
                Err(e) => self.response_error(task, ErrorCode::from(e)),
            };
        }

        run_algorithm(&scoped_csr, algorithm, params, &self.graph_tuning)
            .and_then(|batch| batch.to_msgpack())
            .map_or_else(
                |e| self.response_error(task, ErrorCode::from(e)),
                |payload| self.response_with_payload(task, payload),
            )
    }
}

/// Build a `CsrIndex` containing only the edges for a specific `(tid, collection)`,