ENH: Lazy Cross-Join With Deferred Filtering

[ms7463]

Problem:
One of the only operations I ever find myself going out of my way to create a sql temp-table upload my frame and do the operation in sql is for filtered cross joins. For large datasets they can both take a long time and use large amounts of memory. Here is an example of such an operation

Current Solution:

dr = pd.date_range('1900-01-01', periods=10000)
df = pd.DataFrame({
    'start_date': pd.Series(dr), 
    'end_date': pd.Series(dr) + pd.Timedelta(days=30), 
    'value': range(len(dr)), 
    'id': np.random.choice(list('abcdefg'), len(dr))
})
dates = pd.Series(pd.date_range(df.start_date.min(), df.end_date.max())).to_frame('ts')

# takes ~8 sec (not particularly long in this example) and peak memory at ~10 GB
df.merge(dates, how='cross').loc[lambda dfx: (dfx.ts >= dfx.start_date) & (dfx.ts < dfx.end_date)]

This is how it would be done in SQL, which I don't have memory stats for, but is much faster

select * from table1
cross join table2
where ts >= start_date and ts <= end_date

This is a common problem people try to solve using pandas, as there are several instances on stack overflow of people asking how to do this (here are just a few I found with a quick search, there are many more):
https://stackoverflow.com/questions/43593554/merging-two-dataframes-based-on-a-date-between-two-other-dates-without-a-common/43594038
https://stackoverflow.com/questions/30627968/merge-pandas-dataframes-where-one-value-is-between-two-others
https://stackoverflow.com/questions/31328014/merging-dataframes-based-on-date-range

There are many different solutions ranging in complexity and performance. Often the answer people provide is to just do it in sql.

Proposed Solution: (not concrete/final proposal, just a starting point for discussion)
I think that one way this could be solved in a performant way with an easy to use interface is to provide a lazily evaluated cross join. Which is resolved to a pandas dataframe after a filtering operation (e.g. .loc, .query, etc.). Below I provide an example implementation, that I don't think is the complete or optimal solution, but rather just a quick example to get the point across.

import itertools as it

 def chunks(lst, n):
     for i in range(0, len(lst), n):
         yield lst[i:i + n]

class LazyMergeLoc:
    def __get__(self, instance, owner):
        self.chunk_pairs = instance.chunk_pairs
        return self

    def __getitem__(self, item):
        results = []
        for df1, df2 in self.chunk_pairs:
            result = df1.merge(df2, how='cross').loc[item]
            results.append(result)
        return pd.concat(results)

class LazyMerge:
    loc = LazyMergeLoc()

    def __init__(self, chunk_pairs):
        self.chunk_pairs = chunk_pairs

def lazy_cross_join(df1, df2, left_chunk_size, right_chunk_size):
    left_chunks = list(chunks(df, left_chunk_size))
    right_chunks = list(chunks(dates, right_chunk_size))
    chunk_pairs = it.product(left_chuknks, right_chunks)
    return LazyMerge(chunk_pairs)

pd.lazy_cross_join(df, dates, left_chunk_size=1000, right_chunk_size=1000).loc[lambda dfx: (dfx.ts >= dfx.start_date) & (dfx.ts < dfx.end_date)]

This solution provides the same result in the same time (and much faster too in many cases), and peak memory of only a couple hundred MB. ~50x less memory.

Again, I don't think this is necessarily the right implementation or interface. I'm sure there are even more performant ways to resolve the lazily defined join, not through chunks, perhaps some optimized cython code. But I do think this type of lazy merge would provide a large benefit to the library.

Breaking Implications:
None

Alternatives:
Many solutions provided in the linked (and other) SO posts. All of which are either (a) not performant, (b) complicated, (c) not generalizable, or (d) require leaving the pandas ecosystem.

[jreback]

this is a pretty common operation but easily and very performant to do: https://stackoverflow.com/questions/66963703/pandas-left-join-between-datetimes

rather than a cross join which if you read the original pull request is not necessarily the most performant rather is for convenience

[ms7463]

I gotta agree with the comment on that answer. "using IntervalIndex is brilliant!". Unfortunately, there are many cases where this answer falls short of a generalizable cross-join solution.

1. IntervalIndex will only handle non-overlapping indices. This means that if your start/endtime bounds have overlapping periods (which most cases do), this solution won't work.
   Edit: Slightly misspoke here. Meant to say IntervalIndex.get_indexer only handles non-overlapping periods. Looks like IntervalIndex.get_indexer_non_unique() exists, but implementing a similar solution with this starts getting into the "complicated" territory.

2. The IntervalIndex solution is specific only to a start <= t <= end filter. While I admit that is one of the more, if not most, common usecases, it would require a different solution if say you wanted to do a start <= t filter. Or any other potential filter, doesn't necessarily need to be a time-bounded join. It would be nice to handle all potential cases in a consistent manner.

Incidentally, the author of that solution goes on to propose a plain cross-join + filter as an alternative solution.

[ms7463]

@jreback: slightly misspoke in my above comment. Edited to clarify.

[ms7463]

@jreback any thoughts given my follow up comment? I still think that a performant lazy cross join would be a real benefit to the library. It would provide a performant way to solve a very common type of problem in a simple/intuitive way, that is directly analogous to how it would be solved in sql.

All that said if this suggestion doesn’t fall inline with scope of the pandas API then I will go ahead and close the ticket.

[jreback]

search for other issues which have a function provide for a conditional join - these are a reasonable api, which then could be solved by chunk merging

so would take a soln that does that - kind of your impl under the hood but uses the existing api (which a condition keyword)

[ms7463]

Cool, the condition keyword makes sense too, I’ll take a look soon. Btw I don’t think that chunked merging is necessarily the best solution. Could potentially be some cython based solution too, not sure.

[abubelinha]

@aa1371 I am just a beginner so forgive me if this is obvious but ... is there something missing in your code examples?
I just changed these lines:

• chunk_pairs = it.product(left_chuknks, right_chunks) <--- (small typo: left_chunks)
• pd.lazy_cross_join(...) <--- I removed pd. prefix for running the example
  I understand we should call it lazy_cross_join(...) until it is part of pandas, right?

Other than that, I copied-pasted everything but got a similar error message in both solutions:

OUTPUT: MergeError. No common columns to perform merge on

Current Solution: https://colab.research.google.com/drive/1J_nu9OWDn6vfaDTxZB0l9mMDSVq0dum8
Proposed Solution: https://colab.research.google.com/drive/12WGpqnGvMYLYUkT5Ki09u84ZGYqAr9tF

Perhaps Google Colab notebooks (pandas 1.1.5) are not suitable for running these examples?

[ms7463]

@abubelinha - this is because cross merge was introduced in v1.2. The reason that you're getting that error though, and not complaining about cross merge is because I guess that it does the check for join keys before it resolves the merge type. You can either upgrade to v1.2+ or you can replace this line:

result = df1.merge(df2, how='cross').loc[item]

with this:

result = df1.assign(__tmp=1).merge(df2.assign(__tmp=1)).drop('__tmp', axis=1).loc[item]

Since that will emulate a cross join.

[abubelinha]

Thanks a lot. I upgraded pandas to 1.2.0 in Google Colab, then restarted, re-run and finally merge(how='cross') worked.
Also, your replacement helped me to understand the cross join concept (1 always equals 1).

I can confirm the "current solution" exhausted notebook RAM when used with 10000 rows in dr (Your session crashed after using all available RAM) ... but your "proposed solution" finishes perfectly.