ENH: sped up inner/outer_join_indexer cython functions

Author: wesm

RELEASE.rst

@@ -21,6 +21,8 @@ This is a bug fix release
   - `read_csv` can read multiple columns into a `MultiIndex`. DataFrame's
     `to_csv` method will properly write out a `MultiIndex` which can be read
     back (GH #151, thanks to Skipper Seabold)
+  - Wrote fast time series merging / joining methods in Cython. Will be
+    integrated later into DataFrame.join and related functions
 
 **Bug fixes**
 

pandas/src/reindex.pyx

@@ -455,9 +455,34 @@ def inner_join_indexer(ndarray[int64_t] left, ndarray[int64_t] right):
     nleft = len(left)
     nright = len(right)
 
-    lindexer = np.empty(min(nleft, nright), dtype=np.int32)
-    rindexer = np.empty(min(nleft, nright), dtype=np.int32)
-    result = np.empty(min(nleft, nright), dtype=np.int64)
+    i = 0
+    j = 0
+    count = 0
+    while True:
+        if i == nleft:
+            break
+
+        val = left[i]
+
+        while j < nright and right[j] < val:
+            j += 1
+
+        if j == nright:
+            break
+
+        if val == right[j]:
+            count += 1
+            i += 1
+            j += 1
+        else:
+            while left[i] < right[j]:
+                i += 1
+
+    # do it again now that result size is known
+
+    lindexer = np.empty(count, dtype=np.int32)
+    rindexer = np.empty(count, dtype=np.int32)
+    result = np.empty(count, dtype=np.int64)
 
     i = 0
     j = 0
@@ -485,9 +510,7 @@ def inner_join_indexer(ndarray[int64_t] left, ndarray[int64_t] right):
             while left[i] < right[j]:
                 i += 1
 
-    return (result[:count].copy(),
-            lindexer[:count].copy(),
-            rindexer[:count].copy())
+    return result, lindexer, rindexer
 
 def _inner_join_count(ndarray[int64_t] left, ndarray[int64_t] right):
     pass
@@ -496,18 +519,48 @@ def _inner_join_count(ndarray[int64_t] left, ndarray[int64_t] right):
 @cython.boundscheck(False)
 def outer_join_indexer(ndarray[int64_t] left, ndarray[int64_t] right):
     cdef:
-        Py_ssize_t i, j, nright, nleft, tot, count
+        Py_ssize_t i, j, nright, nleft, count
         int64_t lval, rval
         ndarray[int32_t] lindexer, rindexer
         ndarray[int64_t] result
 
     nleft = len(left)
     nright = len(right)
-    tot = nleft + nright
 
-    lindexer = np.empty(tot, dtype=np.int32)
-    rindexer = np.empty(tot, dtype=np.int32)
-    result = np.empty(tot, dtype=np.int64)
+    i = 0
+    j = 0
+    count = 0
+    while True:
+        if i == nleft:
+            if j == nright:
+                # we are done
+                break
+            else:
+                while j < nright:
+                    j += 1
+                    count += 1
+                break
+        elif j == nright:
+            while i < nleft:
+                i += 1
+                count += 1
+            break
+        else:
+            if left[i] == right[j]:
+                i += 1
+                j += 1
+            elif left[i] < right[j]:
+                i += 1
+            else:
+                j += 1
+
+            count += 1
+
+    lindexer = np.empty(count, dtype=np.int32)
+    rindexer = np.empty(count, dtype=np.int32)
+    result = np.empty(count, dtype=np.int64)
+
+    # do it again, but populate the indexers / result
 
     i = 0
     j = 0
@@ -542,7 +595,7 @@ def outer_join_indexer(ndarray[int64_t] left, ndarray[int64_t] right):
                 result[count] = lval
                 i += 1
                 j += 1
-            elif left[i] < right[j]:
+            elif lval < rval:
                 lindexer[count] = i
                 rindexer[count] = -1
                 result[count] = lval
@@ -555,9 +608,7 @@ def outer_join_indexer(ndarray[int64_t] left, ndarray[int64_t] right):
 
             count += 1
 
-    return (result[:count].copy(),
-            lindexer[:count].copy(),
-            rindexer[:count].copy())
+    return result, lindexer, rindexer
 
 @cython.wraparound(False)
 @cython.boundscheck(False)

scripts/bench_join.R

@@ -1,6 +1,6 @@
 library(xts)
 
-iterations <- 100
+iterations <- 50
 
 ns = c(100, 1000, 10000, 100000, 1000000)
 kinds = c("outer", "left", "inner")
@@ -9,24 +9,30 @@ result = matrix(0, nrow=3, ncol=length(ns))
 n <- 100000
 pct.overlap <- 0.2
 
-k <- 5
+k <- 1
 
 for (ni in 1:length(ns)){
-  n <- ns[ni]
-  rng1 <- 1:n
-  offset <- as.integer(n * pct.overlap)
-  rng2 <- rng1 + offset
-  x <- xts(matrix(rnorm(n * k), nrow=n, ncol=k),
-           as.POSIXct(Sys.Date()) + rng1)
-  y <- xts(matrix(rnorm(n * k), nrow=n, ncol=k),
-           as.POSIXct(Sys.Date()) + rng2)
-  for (i in 1:3) {
-      kind = kinds[i]
-      timing <- system.time(for (j in 1:iterations) merge(x, y, join=kind),
-                            gcFirst=F)
-      timing <- as.list(timing)
-      result[i, ni] = (timing$elapsed / iterations) * 1000
-    }
+ n <- ns[ni]
+ rng1 <- 1:n
+ offset <- as.integer(n * pct.overlap)
+ rng2 <- rng1 + offset
+ x <- xts(matrix(rnorm(n * k), nrow=n, ncol=k),
+          as.POSIXct(Sys.Date()) + rng1)
+ y <- xts(matrix(rnorm(n * k), nrow=n, ncol=k),
+          as.POSIXct(Sys.Date()) + rng2)
+ timing <- numeric()
+ for (i in 1:3) {
+     kind = kinds[i]
+     for(j in 1:iterations) {
+       gc()  # just to be sure
+       timing[j] <- system.time(merge(x,y,join=kind))[3]
+     }
+     #timing <- system.time(for (j in 1:iterations) merge.xts(x, y, join=kind),
+     #                      gcFirst=F)
+     #timing <- as.list(timing)
+     result[i, ni] <- mean(timing) * 1000
+     #result[i, ni] = (timing$elapsed / iterations) * 1000
+   }
 }
 
 rownames(result) <- kinds

scripts/bench_join.py

@@ -4,7 +4,7 @@
 from copy import deepcopy
 import time
 
-n = 100000
+n = 1000000
 K = 1
 pct_overlap = 0.2
 
@@ -135,7 +135,7 @@ def bench_python(n=100000, pct_overlap=0.20):
     import gc
     ns = [2, 3, 4, 5, 6]
     iterations = 50
-    K = 1
+    K = 5
     pct_overlap = 0.2
     kinds = ['outer', 'left', 'inner']