[FEA] Cache distinct_count in Column because of performance issues

[galipremsagar]

Is your feature request related to a problem? Please describe.
Proposal to cache distinct_count, from some of the profiles that are related to cudf.concat, I'm seeing a perf hit that we are taking every-time we would invoke Index.intersection: #9333. See the time taken to calculate Index.is_unique(calls Column.distinct_count()) and Index.unique(calls Frame.drop_duplicates ). :

In [1]: import cudf

In [2]: x = cudf.RangeIndex(start=0, stop=3000000).astype('str')

In [3]: x
Out[3]: StringIndex(['0' '1' '2' ... '2999997' '2999998' '2999999'], dtype='object')

In [4]: %timeit x.is_unique
38.8 ms ± 157 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

In [5]: %timeit x.unique()
42.9 ms ± 130 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

Describe the solution you'd like
The solution is to have a _distinct_count property in Column so that we can have this value cached.

[galipremsagar]

cc: @beckernick

[jrhemstad]

Why is the count of unique elements needed for concat?

[galipremsagar]

We have made some optimizations and improved correctness in concat when axis=1 i.e., appending along columns, for which we would have to perform intersection of index columns of all inputs.

In intersection we drop the duplicates if there are non unique elements:

def intersection(self, other, sort=False):
        """
        Form the intersection of two Index objects.

        This returns a new Index with elements common to the index and `other`.

        Parameters
        ----------
        other : Index or array-like
        sort : False or None, default False
            Whether to sort the resulting index.

            * False : do not sort the result.
            * None : sort the result, except when `self` and `other` are equal
              or when the values cannot be compared.

        Returns
        -------
        intersection : Index

        Examples
        --------
        >>> import cudf
        >>> import pandas as pd
        >>> idx1 = cudf.Index([1, 2, 3, 4])
        >>> idx2 = cudf.Index([3, 4, 5, 6])
        >>> idx1.intersection(idx2)
        Int64Index([3, 4], dtype='int64')

        MultiIndex case

        >>> idx1 = cudf.MultiIndex.from_pandas(
        ...    pd.MultiIndex.from_arrays(
        ...         [[1, 1, 3, 4], ["Red", "Blue", "Red", "Blue"]]
        ...    )
        ... )
        >>> idx2 = cudf.MultiIndex.from_pandas(
        ...    pd.MultiIndex.from_arrays(
        ...         [[1, 1, 2, 2], ["Red", "Blue", "Red", "Blue"]]
        ...    )
        ... )
        >>> idx1
        MultiIndex([(1,  'Red'),
                    (1, 'Blue'),
                    (3,  'Red'),
                    (4, 'Blue')],
                )
        >>> idx2
        MultiIndex([(1,  'Red'),
                    (1, 'Blue'),
                    (2,  'Red'),
                    (2, 'Blue')],
                )
        >>> idx1.intersection(idx2)
        MultiIndex([(1,  'Red'),
                    (1, 'Blue')],
                )
        >>> idx1.intersection(idx2, sort=False)
        MultiIndex([(1,  'Red'),
                    (1, 'Blue')],
                )
        """
        if not isinstance(other, BaseIndex):
            other = cudf.Index(other, name=self.name)

        if sort not in {None, False}:
            raise ValueError(
                f"The 'sort' keyword only takes the values of "
                f"None or False; {sort} was passed."
            )

        if self.equals(other):
            if self.has_duplicates:
                return self.unique()._get_reconciled_name_object(other)
            return self._get_reconciled_name_object(other)

        res_name = _get_result_name(self.name, other.name)

        if (self.is_boolean() and other.is_numeric()) or (
            self.is_numeric() and other.is_boolean()
        ):
            if isinstance(self, cudf.MultiIndex):
                return self[:0].rename(res_name)
            else:
                return cudf.Index([], name=res_name)

        if self.has_duplicates:
            lhs = self.unique()
        else:
            lhs = self
        if other.has_duplicates:
            rhs = other.unique()
        else:
            rhs = other
        result = lhs._intersection(rhs, sort=sort)
        result.name = res_name
        return result

[jrhemstad]

@galipremsagar isn't that operation you described just an inner join between the index columns?

At a high level, IIIUC, the goal here is to perform a "horiztonal" concat that first aligns along the index of both DFs.

You first need to ensure both indexes contain only unique elements, but then the implementation should just be a full outer join between the two indexes, yes? So where does this intersection part come into play?

[galipremsagar]

Apologies I wasn't clear on "why" intersection is being called.

So concat allows the user to control what kind of join is performed on the axis the concatenation is being performed on. In this case join can be inner, which then we would need to find intersection of all index of all DFs. There is also the default case where only outer joins are performed and thus intersection is not used in that case.

We have a Index.is_unique check present in concat much before we do the actual intersection, performing that will initialize distinct_count cache and hence avoid any additional runt-time occupancy by is_unique for all cases of intersection.