Repository: SuperDuperDB/superduperdb
File: superduperdb/base/datalayer.py (Python)

Code before:
            ):
                continue
            G.add_edge(
                f'{download_content.__name__}()',
                f'{listener.model.identifier}.predict_in_db({listener.uuid})',
            )
            if not self.cdc.running:
                deps = self.listeners[identifier].dependencies
                for dep in deps:
                    upstream = self.load(uuid=dep)
                    assert isinstance(upstream, Component)
                    dep_node = (
                        f'{upstream.model.identifier}.predict_in_db({upstream.uuid})'
                    )
                    if dep_node in G.nodes:
                        G.add_edge(
                            dep_node,
                            f'{listener.model.identifier}.predict_in_db({listener.uuid})',
                        )
                    else:
                        raise Exception(
                            f'The dependent listener {upstream.uuid}'
                            ' does not exist.'
                            'Please make sure to add this listener'
                            f'`db.add(...)`'
                        )

        if s.CFG.cluster.vector_search.type == 'native':
            return G

        for identifier in self.show('vector_index'):
            # if a vector-searcher is not loaded, then skip
            # since s.CFG.cluster.vector_search_type == 'in_memory' implies the
            # program is standalone

            if (
                s.CFG.cluster.vector_search.uri is None
                and identifier not in self.fast_vector_searchers
            ):
                continue

            vi = self.vector_indices[identifier]
            if (
                vi.indexing_listener.select.table_or_collection.identifier
                != query.table_or_collection.identifier
            ):
                continue

            G.add_node(
                f'{identifier}.{copy_vectors.__name__}',
                FunctionJob(

Review comment (suggestion):
When multiple listeners are chained, there is no need to enforce relationship building in the absence of upstream tasks. Relationships should only be established when upstream and downstream tasks exist concurrently. Therefore, throwing exceptions is unnecessary.

Code after:
                    assert isinstance(upstream, Component)
                    dep_node = (
                        f'{upstream.model.identifier}.predict_in_db({upstream.uuid})'
                    )
                    if dep_node in G.nodes:
                        G.add_edge(
                            dep_node,
                            f'{listener.model.identifier}.predict_in_db({listener.uuid})',
                        )

        if s.CFG.cluster.vector_search.type == 'native':
            return G

        for identifier in self.show('vector_index'):
            # if a vector-searcher is not loaded, then skip
            # since s.CFG.cluster.vector_search_type == 'in_memory' implies the
            # program is standalone

            if (
                s.CFG.cluster.vector_search.uri is None
                and identifier not in self.fast_vector_searchers
            ):
                continue

            vi = self.vector_indices[identifier]
            if (
                vi.indexing_listener.select.table_or_collection.identifier
                != query.table_or_collection.identifier
            ):
                continue

            G.add_node(
                f'{identifier}.{copy_vectors.__name__}',
                FunctionJob(
                    callable=copy_vectors,
                    args=[],
                    kwargs={
                        'vector_index': identifier,
                        'ids': ids,
                        'query': vi.indexing_listener.select.encode(),
                    },
                ),
            )
            model = vi.indexing_listener.model.identifier
            uuid = vi.indexing_listener.uuid
            G.add_edge(
                f'{model}.predict_in_db({uuid})',
                f'{identifier}.{copy_vectors.__name__}',
            )

        return G