Method column_query

union/D3L/d3l/querying/query_engine.py:107–167

Perform column-level top-k nearest neighbour search over the configured LSH backends. Parameters ---------- column : pd.Series The column query as a Pandas Series. The series name will give the name queries. The series values will

(
        self,
        column: pd.Series,
        aggregator: Optional[callable] = None,
        k: Optional[int] = None,
    )

        return np.average(scores, axis=0, weights=ecdf_weights)

    def column_query(
        self,
        column: pd.Series,
        aggregator: Optional[callable] = None,
        k: Optional[int] = None,
    ) -> Iterable[Tuple[str, Iterable[float]]]:
        """
        Perform column-level top-k nearest neighbour search over the configured LSH backends.
        Parameters
        ----------
        column : pd.Series
            The column query as a Pandas Series.
            The series name will give the name queries.
            The series values will give the value queries.
        aggregator: Optional[callable] = None
            An aggregating function used to merge the results of all configured backends.
            If None then all scores are returned.
        k : Optional[int]
            Only the top-k neighbours will be retrieved from each backend.
            Then, these results are aggregated using the aggregator function and the results re-ranked to retrieve
            the top-k aggregated neighbours.
            If this is None all results are retrieved.

        Returns
        -------
        Iterable[Tuple[str, Iterable[float]]]
            A collection of (column id, aggregated score values) pairs.
            The scores are the values returned by the backends or one aggregated value if an aggregator is passed.
        """

        results = defaultdict(lambda: [0.0] * len(self.query_backends))
        query_name = str(column.name)
        query_value = column.values.tolist()

        for i, backend in enumerate(self.query_backends):
            if isinstance(backend, NameIndex):
                query_results = backend.query(query=query_name, k=k)
            else:
                query_results = backend.query(query=query_value, k=k)

            for rid, score in query_results:
                results[rid][i] = score

        if aggregator is None:
            # If not aggregation is used results are sorted by the mean of the scores.
            # Reverse sorting because the scores are similarities.
            results = sorted(
                results.items(),
                key=lambda items: sum(items[1]) / len(self.query_backends),
                reverse=True,
            )
        else:
            results = {rid: [aggregator(scores)] for rid, scores in results.items()}
            # Reverse sorting because the scores are similarities.
            results = sorted(
                results.items(), key=lambda items: items[1][0], reverse=True
            )