Repository: root-11/tablite (language: python)

Issue:
Join with constant memory footprint
@realratchet @cerv15 

Join has been optimized for speed using `numpy` and to maintain a constant memory footprint equal to 4 pages (4M values).

Occasionally (1 in 100) I can trigger the following error during multiprocessing:

```
/home/bjorn/github/tablite/tests/test_join.py::test_left_join_mp failed: def test_left_join_mp():
        Config.MULTIPROCESSING_MODE = Config.FORCE
>       do_left_join()

tests/test_join.py:52: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
tests/test_join.py:29: in do_left_join
    left_join.show()
tablite/base.py:1748: in show
    print(self.to_ascii(slice_=slice_, blanks=blanks, dtype=dtype))
tablite/base.py:1712: in to_ascii
    for name, values in self.display_dict(slice_=slice_, blanks=blanks, dtype=dtype).items():
tablite/base.py:1663: in display_dict
    data[name] = list(chain(iter(col), repeat(blanks, times=n - len(col))))[slc]
tablite/base.py:694: in __iter__
    data = page.get()
tablite/base.py:147: in get
    array = load_numpy(self.path)
tablite/utils.py:472: in load_numpy
    return np.load(path, allow_pickle=True, fix_imports=False)
../../venv/py310tablite/lib/python3.10/site-packages/numpy/lib/npyio.py:432: in load
    return format.read_array(fid, allow_pickle=allow_pickle,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

fp = <_io.BufferedReader name='/tmp/tablite-tmp/pid-30773/pages/3.npy'>
allow_pickle = True, pickle_kwargs = {'encoding': 'ASCII', 'fix_imports': False}

    def read_array(fp, allow_pickle=False, pickle_kwargs=None, *,
                   max_header_size=_MAX_HEADER_SIZE):
        """
        Read an array from an NPY file.
    
        ...
    
        """
        if allow_pickle:
            # Effectively ignore max_header_size, since `allow_pickle` indicates
            # that the input is fully trusted.
            max_header_size = 2**64
    
        version = read_magic(fp)
        _check_version(version)
        shape, fortran_order, dtype = _read_array_header(
                fp, version, max_header_size=max_header_size)
        if len(shape) == 0:
            count = 1
        else:
            count = numpy.multiply.reduce(shape, dtype=numpy.int64)
    
        # Now read the actual data.
        if dtype.hasobject:
            # The array contained Python objects. We need to unpickle the data.
            if not allow_pickle:
                raise ValueError("Object arrays cannot be loaded when "
                                 "allow_pickle=False")
            if pickle_kwargs is None:
                pickle_kwargs = {}
            try:
>               array = pickle.load(fp, **pickle_kwargs)
E               _pickle.UnpicklingError: unpickling stack underflow

../../venv/py310tablite/lib/python3.10/site-packages/numpy/lib/format.py:792: UnpicklingError

```

I believe this is not a join bug, but rather a multiprocessing problem. @realratchet ?




Discussion / hints:

Fix patch:
diff --git a/.gitignore b/.gitignore
index c4316375..9f60013d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,6 +3,7 @@
 *.code-workspace
 .vscode/
 /.env
+/mplite
 
 # Compiled files
 *.pyc
diff --git a/README.md b/README.md
index a7d6f31a..e60b92b0 100644
--- a/README.md
+++ b/README.md
@@ -82,11 +82,11 @@ Usage:  `>>> from tablite import Table`
 
 ## <a name="build & test"></a>Build & test
 
-```
 install nim >= 2.0.0
-chmod +x ./build_nim.sh
-run ./build_nim.sh
-```
+
+run: `chmod +x ./build_nim.sh`
+run: `./build_nim.sh`
+
 Should the default nim not be your desired taste, please use `nims` environment manager (`atlas`) and run `source nim-2.0.0/activate.sh` on UNIX or `nim-2.0.0/activate.bat` on windows.
 
 ```
diff --git a/tablite/_nimlite/funcs/column_selector/collectinfo.nim b/tablite/_nimlite/funcs/column_selector/collectinfo.nim
index 98ad61d4..50839398 100644
--- a/tablite/_nimlite/funcs/column_selector/collectinfo.nim
+++ b/tablite/_nimlite/funcs/column_selector/collectinfo.nim
@@ -112,7 +112,7 @@ proc collectColumnSelectInfo*(table: nimpy.PyObject, cols: nimpy.PyObject, dirPi
 
     var isCorrectType = initTable[string, bool]()
 
-    proc genpage(dirpid: string): ColSliceInfo {.inline.} = (dir_pid, tabliteBase().SimplePage.next_id(dir_pid).to(int))
+    proc genpage(dirpid: string): ColSliceInfo {.inline.} = (dir_pid, tabliteBase().SimplePage.next_id(dir_pid).to(string))
 
     discard pbar.update(5)
     discard pbar.display()
diff --git a/tablite/_nimlite/funcs/column_selector/infos.nim b/tablite/_nimlite/funcs/column_selector/infos.nim
index 8026a645..bc35180a 100644
--- a/tablite/_nimlite/funcs/column_selector/infos.nim
+++ b/tablite/_nimlite/funcs/column_selector/infos.nim
@@ -4,7 +4,7 @@ from std/sugar import collect
 from ../../pymodules import builtins
 from ../../pytypes import KindObjectND, str2ObjKind
 
-type ColSliceInfo* = (string, int)
+type ColSliceInfo* = (string, string)
 type ColInfo* = Table[string, ColSliceInfo]
 type DesiredColumnInfo* = object
     originalName*: string
diff --git a/tablite/_nimlite/funcs/column_selector/sliceconv.nim b/tablite/_nimlite/funcs/column_selector/sliceconv.nim
index ef08e311..b237f268 100644
--- a/tablite/_nimlite/funcs/column_selector/sliceconv.nim
+++ b/tablite/_nimlite/funcs/column_selector/sliceconv.nim
@@ -35,7 +35,7 @@ proc finalizeSlice(indices: var seq[int], columnNames: seq[string], infos: var T
 
 proc toColSliceInfo(path: Path): ColSliceInfo =
     let workdir = string path.parentDir.parentDir
-    let pid = parseInt(string path.extractFilename.changeFileExt(""))
+    let pid = string path.extractFilename.changeFileExt("")
 
     return (workdir, pid)
 
diff --git a/tablite/_nimlite/funcs/text_reader/csvparse.nim b/tablite/_nimlite/funcs/text_reader/csvparse.nim
index 18bb377b..a27a90b7 100644
--- a/tablite/_nimlite/funcs/text_reader/csvparse.nim
+++ b/tablite/_nimlite/funcs/text_reader/csvparse.nim
@@ -273,22 +273,15 @@ iterator parseCSV*(self: var ReaderObj, path: string, encoding: FileEncoding): (
     finally:
         fh.close()
 
-proc str2quoting*(quoting: string): Quoting =
+proc str2quoting*(quoting: string): Quoting {.inline.} =
     case quoting.toUpper():
-        of $QUOTE_MINIMAL:
-            return QUOTE_MINIMAL
-        of $QUOTE_ALL:
-            return QUOTE_ALL
-        of $QUOTE_NONNUMERIC:
-            return QUOTE_NONNUMERIC
-        of $QUOTE_NONE:
-            return QUOTE_NONE
-        of $QUOTE_STRINGS:
-            return QUOTE_STRINGS
-        of $QUOTE_NOTNULL:
-            return QUOTE_NOTNULL
-        else:
-            raise newException(Exception, "invalid quoting: " & quoting)
+    of $QUOTE_MINIMAL: return QUOTE_MINIMAL
+    of $QUOTE_ALL: return QUOTE_ALL
+    of $QUOTE_NONNUMERIC: return QUOTE_NONNUMERIC
+    of $QUOTE_NONE: return QUOTE_NONE
+    of $QUOTE_STRINGS: return QUOTE_STRINGS
+    of $QUOTE_NOTNULL: return QUOTE_NOTNULL
+    else: raise newException(Exception, "invalid quoting: " & quoting)
 
 proc findNewlinesNoQualifier*(fh: BaseEncodedFile): (seq[uint], uint) =
     var newline_offsets = newSeq[uint](1)
diff --git a/tablite/_nimlite/funcs/text_reader/text_reader.nim b/tablite/_nimlite/funcs/text_reader/text_reader.nim
index 23844e69..42a59833 100644
--- a/tablite/_nimlite/funcs/text_reader/text_reader.nim
+++ b/tablite/_nimlite/funcs/text_reader/text_reader.nim
@@ -67,7 +67,7 @@ proc textReaderTask*(task: TaskArgs): seq[nimpy.PyObject] =
                 for i in 0..<n_pages:
                     let path = Path(destinations[i])
                     let workdir = string path.parentDir.parentDir
-                    let id = parseInt(string path.extractFilename.changeFileExt(""))
+                    let id = string path.extractFilename.changeFileExt("")
                     let dtypes = pgTypes[i]
                     let len = pgLens[i]
 
diff --git a/tablite/_nimlite/numpy.nim b/tablite/_nimlite/numpy.nim
index c3e34494..03e6fc30 100644
--- a/tablite/_nimlite/numpy.nim
+++ b/tablite/_nimlite/numpy.nim
@@ -1017,7 +1017,7 @@ proc type2PyType(`type`: KindObjectND): nimpy.PyObject =
     of K_TIME: return pymodules.datetime().time
     of K_DATETIME: return pymodules.datetime().datetime
 
-proc newPyPage*(id: int, path: string, len: int, dtypes: Table[KindObjectND, int]): nimpy.PyObject =
+proc newPyPage*(id: string, path: string, len: int, dtypes: Table[KindObjectND, int]): nimpy.PyObject =
     let pyDtypes = pymodules.builtins().dict()
     
     for (dt, n) in dtypes.pairs:
diff --git a/tablite/base.py b/tablite/base.py
index 7c8bb1aa..291ca834 100644
--- a/tablite/base.py
+++ b/tablite/base.py
@@ -9,6 +9,7 @@
 import zipfile
 import numpy as np
 from pathlib import Path
+from typing import List, Union
 from tqdm import tqdm as _tqdm
 from collections import defaultdict, Counter
 from itertools import count, chain, product, repeat
@@ -69,7 +70,6 @@ class SimplePage(object):
     autocleanup = True
 
     def __init__(self, id, path, len, py_dtype) -> None:
-        self.id = id
         self.path = Path(path) / "pages" / f"{id}.npy"
         self.len = len
         self.dtype = py_dtype
@@ -96,7 +96,7 @@ def next_id(cls, path):
         path = Path(path)
 
         while True:
-            _id = next(cls.ids)
+            _id = f"{os.getpid()}-{next(cls.ids)}"
             _path = path / "pages" / f"{_id}.npy"
 
             if not _path.exists():
@@ -116,7 +116,7 @@ def __repr__(self) -> str:
             return f"{self.__class__.__name__}({self.path}, <{e}>)"
 
     def __hash__(self) -> int:
-        return hash(self.id)
+        return hash(self.path)
 
     def owns(self):
         parts = self.path.parts
@@ -201,7 +201,7 @@ def __init__(self, path, value=None) -> None:
         """Create Column
 
         Args:
-            path (Path): path of table.yml
+            path (Path): path of table.yml (defaults: Config.pid_dir)
             value (Iterable, optional): Data to store. Defaults to None.
         """
         self.path = path
@@ -246,9 +246,8 @@ def _paginate(values, page_size=None):
     def repaginate(self):
         """resizes pages to Config.PAGE_SIZE"""
         new_pages = []
-        start, end = 0, 0
-        for _ in range(0, len(self) + 1, Config.PAGE_SIZE):
-            start, end = end, end + Config.PAGE_SIZE
+        for start in range(0, len(self) + 1, Config.PAGE_SIZE):
+            end = start + Config.PAGE_SIZE
             array = self[slice(start, end, 1)]
 
             np_dtype, py_dtype = pytype_from_iterable(array.tolist())
@@ -359,7 +358,7 @@ def iter_by_page(self):
         start, end = 0, 0
         for page in self.pages:
             start, end = end, end + page.len
-            yield start, end, page.get()
+            yield start, end, page
 
     def __getitem__(self, item):  # USER FUNCTION.
         """gets numpy array.
@@ -633,7 +632,7 @@ def _del_by_int(self, key):  # PRIVATE FUNCTION
                 new_page = Page(self.path, new_data)
                 self.pages[index] = new_page
 
-    def _del_by_slice(self, key):
+    def _del_by_slice(self, key:slice) -> None:
         """handles the following case:
         ```
         del column[m:n:o]
@@ -666,7 +665,7 @@ def _del_by_slice(self, key):
         new_arrays = self._paginate(pruned)
         self.pages = head + [Page(self.path, arr) for arr in new_arrays] + tail
 
-    def get_by_indices(self, indices):
+    def get_by_indices(self, indices: Union[List[int], np.ndarray]) -> np.ndarray:
         """retrieves values from column given a set of indices.
 
         Args:
@@ -690,20 +689,21 @@ def get_by_indices(self, indices):
             indices.shape, dtype=object
         )  # placeholder for the indexed values.
 
-        for start, end, data in self.iter_by_page():
-            range_match = np.asarray(
-                ((indices >= start) & (indices < end)) | (indices == -1)
-            ).nonzero()[0]
+        for start, end, page in self.iter_by_page():
+            range_match = np.asarray(((indices >= start) & (indices < end)) | (indices == -1)).nonzero()[0]
             if len(range_match):
+                # only fetch the data if there's a range match!
+                data = page.get() 
                 sub_index = np.take(indices, range_match)
-                sub_index2 = np.where(sub_index == -1, -1, sub_index - start)
-                # diss: the line above is required to cover for cases where len(data) > (-1 - start)
-                #       as sub_index2 otherwise will raise index error
-                arr = np.take(data, sub_index2)
+                # sub_index2 otherwise will raise index error where len(data) > (-1 - start)
+                # so the clause below is required:
+                if len(data) > (-1 - start):
+                    sub_index = np.where(sub_index == -1, -1, sub_index - start)
+                arr = np.take(data, sub_index)
                 dtypes.add(arr.dtype)
                 np.put(values, range_match, arr)
 
-        if len(dtypes) == 1:  # simplify the datatype.
+        if len(dtypes) == 1:  # simplify the datatype
             dtype = next(iter(dtypes))
             values = np.array(values, dtype=dtype)
         return values
@@ -1064,7 +1064,13 @@ class Table(object):
     _ids = count()
     _add_row_slow_warning = False
 
-    def __init__(self, columns=None, headers=None, rows=None, _path=None) -> None:
+    def __init__(
+        self,
+        columns: [dict, None] = None,
+        headers: [list, None] = None,
+        rows: [list, None] = None,
+        _path: [Path, None] = None,
+    ) -> None:
         """creates Table
 
         Args:
@@ -1075,6 +1081,8 @@ def __init__(self, columns=None, headers=None, rows=None, _path=None) -> None:
                 headers (list of strings, optional): list of column names.
                 rows (list of tuples or lists, optional): values for columns
                 Example: t = Table(headers=["a", "b"], rows=[[1,3], [2,4]])
+
+            _path (pathlib.Path, optional): path to main process working directory.
         """
         if _path is None:
             if self._pid_dir is None:
@@ -1353,12 +1361,14 @@ def save(
         ----------------------------------------
         ```
         """
+        if isinstance(path, str):
+            path = Path(path)
         type_check(path, Path)
         if path.is_dir():
             raise TypeError(f"filename needed: {path}")
         if path.suffix != ".tpz":
-            path += ".tpz"
-
+            path = path.parent / (path.parts[-1] + ".tpz")
+        
         # create yaml document
         _page_counter = 0
         d = {}
diff --git a/tablite/config.py b/tablite/config.py
index 40264332..2b56d434 100644
--- a/tablite/config.py
+++ b/tablite/config.py
@@ -2,24 +2,26 @@
 import pathlib
 import tempfile
 import platform
+
 from dotenv import load_dotenv
 
 load_dotenv()
 
+
 class Config(object):
     """Config class for Tablite Tables.
 
     The default location for the storage is loaded as
-
+    ```
     Config.workdir = pathlib.Path(os.environ.get("TABLITE_TMPDIR", f"{tempfile.gettempdir()}/tablite-tmp"))
-
+    ```
     to overwrite, first import the config class, then set the new workdir.
     ```
     >>> from tablite import config
     >>> from pathlib import Path
     >>> config.workdir = Path("/this/new/location")
     ```
-    for every new table or record this path will be used.
+    the new path will now be used for every new table.
 
     PAGE_SIZE = 1_000_000 sets the page size limit.
 
@@ -34,7 +36,8 @@ class Config(object):
     when the number of fields (rows x columns) exceed this value,
     multiprocessing is used.
     """
-    USE_NIMPORTER = os.environ.get("USE_NIMPORTER", "true").lower() in ["1", "t", "true", "y", "yes"]
+
+    USE_NIMPORTER = os.environ.get("USE_NIMPORTER", "true").lower() in ["1","t","true","y","yes"]
     ALLOW_CSV_READER_FALLTHROUGH = os.environ.get("ALLOW_CSV_READER_FALLTHROUGH", "true").lower() in ["1", "t", "true", "y", "yes"]
 
     NIM_SUPPORTED_CONV_TYPES = ["Windows-1252", "ISO-8859-1"]
@@ -47,29 +50,23 @@ class Config(object):
     PAGE_SIZE = 1_000_000  # sets the page size limit.
     ENCODING = "UTF-8"  # sets the page encoding when using bytes
 
-    DISK_LIMIT = int(10e9)  # 10e9 (10Gb) on 100 Gb disk means raise at
-    # 90 Gb disk usage.
-    # if DISK_LIMIT <= 0, the check is turned off.
+    DISK_LIMIT = int(10e9)
+    """ 
+    10e9 (10Gb) on 100 Gb disk means raise at 90 Gb disk usage.
+    if DISK_LIMIT <= 0, the check is turned off.
+    """
 
     SINGLE_PROCESSING_LIMIT = 1_000_000
-    # when the number of fields (rows x columns)
-    # exceed this value, multiprocessing is used.
-
+    """
+    when the number of fields (rows x columns)
+    exceed this value, multiprocessing is used.
+    """
+    vpus = max(os.cpu_count() - 1, 1)
     AUTO = "auto"
     FALSE = "sp"
     FORCE = "mp"
     MULTIPROCESSING_MODE = AUTO
-    # Usage example (from import_utils in text_reader)
-    # if cpu_count < 2 or Config.MULTIPROCESSING_MODE == Config.FALSE:
-    #         for task in tasks:
-    #             task.execute()
-    #             pbar.update(dump_size)
-    #     else:
-    #         with TaskManager(cpu_count - 1) as tm:
-    #             errors = tm.execute(tasks, pbar=PatchTqdm())  # I expects a list of None's if everything is ok.
-    #             if any(errors):
-    #                 raise Exception("\n".join(e for e in errors if e))
-
+    
     TQDM_DISABLE = False  # set to True to disable tqdm
 
     @classmethod
diff --git a/tablite/core.py b/tablite/core.py
index 613af85c..bec97278 100644
--- a/tablite/core.py
+++ b/tablite/core.py
@@ -22,7 +22,7 @@
 from tablite import imputation
 from tablite import diff
 from tablite.config import Config
-from tablite.nimlite import column_select as _column_select
+from tablite.nimlite import column_select as _column_select, ColumnSelectorDict
 from mplite import TaskManager as _TaskManager
 
 logging.getLogger("lml").propagate = False
@@ -698,7 +698,7 @@ def merge(self, left, right, new, criteria):
         """
         return merge.where(self, criteria,left,right,new)
 
-    def column_select(self, cols, tqdm=_tqdm, TaskManager=_TaskManager):
+    def column_select(self, cols: list[ColumnSelectorDict], tqdm=_tqdm, TaskManager=_TaskManager):
         """
         type-casts columns from a given table to specified type(s)
 
diff --git a/tablite/joins.py b/tablite/joins.py
index b78a60e8..8f5b9725 100644
--- a/tablite/joins.py
+++ b/tablite/joins.py
@@ -1,18 +1,157 @@
-import math
+import psutil
 import numpy as np
+from typing import List, Union
 from itertools import product
+from pathlib import Path
 from tablite.config import Config
-from tablite.base import Table
 from tablite.reindex import reindex
 from tablite.merge import where
-from tablite.utils import sub_cls_check, unique_name
-from tablite.mp_utils import share_mem, map_task, select_processing_method
-from mplite import TaskManager, Task
-import psutil
-
+from tablite.base import Table, Column
+from tablite.utils import sub_cls_check, unique_name, type_check
+from tablite.mp_utils import select_processing_method
+from mplite import Task, TaskManager as _TaskManager
 from tqdm import tqdm as _tqdm
 
 
+def join(
+    T: Table,
+    other: Table,
+    left_keys: List[str],
+    right_keys: List[str],
+    left_columns: Union[List[str], None],
+    right_columns: Union[List[str], None],
+    kind: str = "inner",
+    merge_keys: bool = False,
+    tqdm=_tqdm,
+    pbar=None,
+):
+    """short-cut for all join functions.
+
+    Args:
+        T (Table): left table
+        other (Table): right table
+        left_keys (list): list of keys for the join from left table.
+        right_keys (list): list of keys for the join from right table.
+        left_columns (list): list of columns names to retain from left table.
+            If None, all are retained.
+        right_columns (list): list of columns names to retain from right table.
+            If None, all are retained.
+        kind (str, optional): 'inner', 'left', 'outer', 'cross'. Defaults to "inner".
+        tqdm (tqdm, optional): tqdm progress counter. Defaults to _tqdm.
+        pbar (tqdm.pbar, optional): tqdm.progressbar. Defaults to None.
+
+    Raises:
+        ValueError: if join type is unknown.
+
+    Returns:
+        Table: joined table.
+    
+    Example: "inner"
+    ```
+    SQL:   SELECT number, letter FROM numbers JOIN letters ON numbers.colour == letters.color
+    ```
+    Tablite: 
+    ```
+    >>> inner_join = numbers.inner_join(
+        letters, 
+        left_keys=['colour'], 
+        right_keys=['color'], 
+        left_columns=['number'], 
+        right_columns=['letter']
+    )
+    ```
+    
+    Example: "left" 
+    ```
+    SQL:   SELECT number, letter FROM numbers LEFT JOIN letters ON numbers.colour == letters.color
+    ```
+    Tablite: 
+    ```
+    >>> left_join = numbers.left_join(
+        letters, 
+        left_keys=['colour'], 
+        right_keys=['color'], 
+        left_columns=['number'], 
+        right_columns=['letter']
+    )
+    ```
+
+    Example: "outer"
+    ```
+    SQL:   SELECT number, letter FROM numbers OUTER JOIN letters ON numbers.colour == letters.color
+    ```
+
+    Tablite: 
+    ```
+    >>> outer_join = numbers.outer_join(
+        letters, 
+        left_keys=['colour'], 
+        right_keys=['color'], 
+        left_columns=['number'], 
+        right_columns=['letter']
+        )
+    ```
+
+    Example: "cross"
+
+    CROSS JOIN returns the Cartesian product of rows from tables in the join.
+    In other words, it will produce rows which combine each row from the first table
+    with each row from the second table
+    """
+    if left_columns is None:
+        left_columns = list(T.columns)
+    if right_columns is None:
+        right_columns = list(other.columns)
+    assert merge_keys in {True,False}
+
+    _jointype_check(T, other, left_keys, right_keys, left_columns, right_columns)
+
+    fields = len(T)*len(T.columns) + len(other)*len(other.columns)
+    m = select_processing_method(fields, sp=_sp_join, mp=_mp_join)
+
+    return m(kind, T,other,left_keys, right_keys, left_columns, right_columns, merge_keys=merge_keys,
+             tqdm=tqdm, pbar=pbar)
+
+# fmt:off
+def inner_join(T: Table, other: Table, left_keys: List[str], right_keys: List[str], 
+              left_columns: Union[List[str], None], right_columns: Union[List[str], None],
+              merge_keys: bool = False, tqdm=_tqdm, pbar=None):
+    return join(T, other, left_keys, right_keys, left_columns, right_columns, kind="inner", merge_keys=merge_keys, tqdm=tqdm,pbar=pbar)
+
+def left_join(T: Table, other: Table, left_keys: List[str], right_keys: List[str], 
+              left_columns: Union[List[str], None], right_columns: Union[List[str], None],
+              merge_keys: bool = False, tqdm=_tqdm, pbar=None):
+    return join(T, other, left_keys, right_keys, left_columns, right_columns, kind="left", merge_keys=merge_keys, tqdm=tqdm,pbar=pbar)
+
+def outer_join(T: Table, other: Table, left_keys: List[str], right_keys: List[str], 
+              left_columns: Union[List[str], None], right_columns: Union[List[str], None],
+              merge_keys: bool = False, tqdm=_tqdm, pbar=None):
+    return join(T, other, left_keys, right_keys, left_columns, right_columns, kind="outer", merge_keys=merge_keys, tqdm=tqdm,pbar=pbar)
+
+def cross_join(T: Table, other: Table, left_keys: List[str], right_keys: List[str], 
+              left_columns: Union[List[str], None], right_columns: Union[List[str], None],
+              merge_keys: bool = False, tqdm=_tqdm, pbar=None):
+    return join(T, other, left_keys, right_keys, left_columns, right_columns, kind="cross", merge_keys=merge_keys, tqdm=tqdm,pbar=pbar)
+# fmt: on
+
+
+def _vpus(tasks):
+    """private helper to determine how many VPUs there is memory for.
+
+    Args:
+        tasks (list): list of tasks
+
+    Returns:
+        integer: number of VPUs
+    """
+    if Config.MULTIPROCESSING_MODE == Config.FALSE:
+        raise TypeError("Config.MULTIPROCESSING_MODE == Config.FALSE")
+    else:
+        memory_per_join = 300e6
+        max_vpus = psutil.virtual_memory().free // memory_per_join
+        return int(min(Config.vpus, len(tasks), max_vpus))
+
+
 def _jointype_check(T, other, left_keys, right_keys, left_columns, right_columns):
     sub_cls_check(T, Table)
     sub_cls_check(other, Table)
@@ -23,9 +162,11 @@ def _jointype_check(T, other, left_keys, right_keys, left_columns, right_columns
         raise TypeError(f"Expected keys as list of strings, not {type(right_keys)}")
 
     if any(key not in T.columns for key in left_keys):
-        raise ValueError(f"left key(s) not found: {[k for k in left_keys if k not in T.columns]}")
+        e = f"left key(s) not found: {[k for k in left_keys if k not in T.columns]}"
+        raise ValueError(e)
     if any(key not in other.columns for key in right_keys):
-        raise ValueError(f"right key(s) not found: {[k for k in right_keys if k not in other.columns]}")
+        e = f"right key(s) not found: {[k for k in right_keys if k not in other.columns]}"
+        raise ValueError(e)
 
     if len(left_keys) != len(right_keys):
         raise ValueError(f"Keys do not have same length: \n{left_keys}, \n{right_keys}")
@@ -35,320 +176,638 @@ def _jointype_check(T, other, left_keys, right_keys, left_columns, right_columns
         if not set(Lcol.types()).intersection(set(Rcol.types())):
             left_types = tuple(t.__name__ for t in list(Lcol.types().keys()))
             right_types = tuple(t.__name__ for t in list(Rcol.types().keys()))
-            raise TypeError(f"Type mismatch: Left key '{L}' {left_types} will never match right keys {right_types}")
+            e = f"Type mismatch: Left key '{L}' {left_types} will never match right keys {right_types}"
+            raise TypeError(e)
 
     if not isinstance(left_columns, list) or not left_columns:
         raise TypeError("left_columns (list of strings) are required")
     if any(column not in T.columns for column in left_columns):
-        raise ValueError(f"Column not found: {[c for c in left_columns if c not in T.columns]}")
+        e = f"Column not found: {[c for c in left_columns if c not in T.columns]}"
+        raise ValueError(e)
 
     if not isinstance(right_columns, list) or not right_columns:
         raise TypeError("right_columns (list or strings) are required")
     if any(column not in other.columns for column in right_columns):
-        raise ValueError(f"Column not found: {[c for c in right_columns if c not in other.columns]}")
+        e = f"Column not found: {[c for c in right_columns if c not in other.columns]}"
+        raise ValueError(e)
     # Input is now guaranteed to be valid.
 
 
-def join(T, other, left_keys, right_keys, left_columns, right_columns, kind="inner", tqdm=_tqdm, pbar=None):
-    """short-cut for all join functions.
+# -------------------------
+# SINGLE PROCESSING SECTION
+# -------------------------
+
 
+def _sp_left_mapping(T, other, left_keys, right_keys, tqdm, pbar):
+    """
     Args:
         T (Table): left table
         other (Table): right table
         left_keys (list): list of keys for the join from left table.
         right_keys (list): list of keys for the join from right table.
-        left_columns (list): list of columns names to retain from left table. 
-            If None, all are retained.
-        right_columns (list): list of columns names to retain from right table. 
-            If None, all are retained.
-        kind (str, optional): 'inner', 'left', 'outer', 'cross'. Defaults to "inner".
-        tqdm (tqdm, optional): tqdm progress counter. Defaults to _tqdm.
-        pbar (tqdm.pbar, optional): tqdm.progressbar. Defaults to None.
+    
+    Returns: 
+        Table: joined table
+    """
+    left_index = T.index(*left_keys)
+    right_index = other.index(*right_keys)
+    _left, _right = [], []
+    for left_key, left_ixs in left_index.items():
+        right_ixs = right_index.get(left_key, (-1,))
+        for left_ix in left_ixs:
+            for right_ix in right_ixs:
+                _left.append(left_ix)
+                _right.append(right_ix)
 
-    Raises:
-        ValueError: if join type is unknown.
+    _left, _right = np.array(_left, dtype=int), np.array(_right, dtype=int)
+    return _left,_right
 
-    Returns:
-        Table: joined table.
-    """
-    kinds = {
-        "inner": T.inner_join,
-        "left": T.left_join,
-        "outer": T.outer_join,
-        "cross": T.cross_join,
-    }
-    if kind not in kinds:
-        raise ValueError(f"join type unknown: {kind}")
-    f = kinds.get(kind, None)
-    return f(other, left_keys, right_keys, left_columns, right_columns, tqdm=tqdm, pbar=pbar)
 
+def _sp_inner_mapping(T, other, left_keys, right_keys, tqdm, pbar):
+    """
+    Args:
+        T (Table): left table
+        other (Table): right table
+        left_keys (list): list of keys for the join from left table.
+        right_keys (list): list of keys for the join from right table.
+    
+    Returns: 
+        Table: joined table
 
-def _sp_join(T, other, LEFT, RIGHT, left_columns, right_columns, tqdm=_tqdm, pbar=None):
     """
-    private helper for single processing join
+    left_index = T.index(*left_keys)
+    right_index = other.index(*right_keys)
+    _left, _right = [], []
+    for left_key, left_ixs in left_index.items():
+        right_ixs = right_index.get(left_key, None)
+        if right_ixs is None:
+            continue
+        for left_ix in left_ixs:
+            for right_ix in right_ixs:
+                _left.append(left_ix)
+                _right.append(right_ix)
+
+    _left, _right = np.array(_left, dtype=int), np.array(_right, dtype=int)
+    return _left, _right
+
+
+def _sp_outer_mapping(T, other, left_keys, right_keys, tqdm, pbar):
     """
-    assert len(LEFT) == len(RIGHT)
-    if pbar is None:
-        total = len(left_columns) + len(right_columns)
-        pbar = tqdm(total=total, desc="join", disable=Config.TQDM_DISABLE)
+    Args:
+        T (Table): left table
+        other (Table): right table
+        left_keys (list): list of keys for the join from left table.
+        right_keys (list): list of keys for the join from right table.
+    
+    Returns: 
+        Table: joined table
 
-    result = reindex(T, LEFT, left_columns, tqdm=tqdm, pbar=pbar)
-    second = reindex(other, RIGHT, right_columns, tqdm=tqdm, pbar=pbar)
-    for name in right_columns:
-        revised_name = unique_name(name, result.columns)
-        result[revised_name] = second[name]
-    return result
+    """
+    left_index = T.index(*left_keys)
+    right_index = other.index(*right_keys)
+    _left, _right, _right_unused = [], [], set(right_index.keys())
+    for left_key, left_ixs in left_index.items():
+        right_ixs = right_index.get(left_key, (-1,))
+        for left_ix in left_ixs:
+            for right_ix in right_ixs:
+                _left.append(left_ix)
+                _right.append(right_ix)
+                _right_unused.discard(left_key)
 
+    for right_key in _right_unused:
+        for right_ix in right_index[right_key]:
+            _left.append(-1)
+            _right.append(right_ix)
 
-def _mp_join(T, other, LEFT, RIGHT, left_columns, right_columns, tqdm=_tqdm, pbar=None):
-    return _sp_join(T, other, LEFT, RIGHT, left_columns, right_columns, tqdm=tqdm, pbar=pbar)
+    _left, _right = np.array(_left, dtype=int), np.array(_right, dtype=int)
+    return _left, _right
 
-    assert len(LEFT) == len(RIGHT)
-    assert isinstance(LEFT, np.ndarray) and isinstance(RIGHT, np.ndarray)
 
-    result = type(T)()
-    cpus = max(psutil.cpu_count(logical=False), 2)
-    step_size = math.ceil(len(LEFT) / cpus)
+def _sp_cross_mapping(T, other, left_keys, right_keys, tqdm, pbar):
+    """
+    Args:
+        T (Table): left table
+        other (Table): right table
+        left_keys (list): list of keys for the join from left table.
+        right_keys (list): list of keys for the join from right table.
+    
+    Returns: 
+        Table: joined table
 
-    with TaskManager(cpu_count=cpus) as tm:  # keeps the CPU pool alive during the whole join.
-        for table, columns, side in ([T, left_columns, LEFT], [other, right_columns, RIGHT]):
-            for name in columns:
-                data = table[name][:]
-                # TODO         ^---- determine how much memory is free and then decide
-                # either to mmap the source or keep it in RAM.
+    """
+    _left, _right = zip(*product(range(len(T)), range(len(other))))
+    _left, _right = np.array(_left, dtype=int), np.array(_right, dtype=int)
+    return _left,_right
 
-                data, data_shm = share_mem(data, data.dtype)  # <-- this is source
-                index, index_shm = share_mem(side, np.int64)  # <-- this is index
-                # As all indices in `index` are positive, -1 is used as replacement for None.
-                destination, dest_shm = share_mem(np.empty(shape=(len(side),)), data.dtype)  # <--this is destination.
 
-                tasks = []
-                start, end = 0, step_size
-                for _ in range(cpus):
-                    tasks.append(Task(map_task, data_shm, index_shm, dest_shm, start, end))
-                    start, end = end, end + step_size
-                # All CPUS now work on the same column and memory footprint is predetermined.
-                results = tm.execute(tasks)
-                if any(i is not None for i in results):
-                    raise Exception("\n".join(filter(lambda x: x is not None, results)))
+_sp_mapping_methods = {
+    "inner": _sp_inner_mapping,
+    "left":  _sp_left_mapping, 
+    "outer": _sp_outer_mapping,
+    "cross": _sp_cross_mapping,
+}
 
-                # As the data and index no longer is needed, then can be closed.
-                index_shm.close()
-                data_shm.close()
+def _sp_join(kind, T, other, left_keys, right_keys, left_columns, right_columns, merge_keys=None, tqdm=_tqdm, pbar=None):
+    _mapping = _sp_mapping_methods.get(kind, None)
+    if _mapping is None:
+        raise ValueError(f"join type unknown: {kind}")
+    
+    _left,_right = _mapping(T, other, left_keys, right_keys, tqdm=tqdm, pbar=pbar)
+    assert len(_left) == len(_right)
 
-                # As all the tasks have been completed, the Column can handle the pagination at once.
-                name = unique_name(name, set(result.columns))
+    if pbar is None:
+        total = len(left_columns) + len(right_columns)
+        pbar = tqdm(total=total, desc="join", disable=Config.TQDM_DISABLE)
 
-                # deal with Nones, before storing.
-                nones = np.empty(shape=destination.shape, dtype=object)
-                result[name] = np.where(index == -1, nones, destination)
+    result = reindex(T, _left, left_columns, tqdm=tqdm, pbar=pbar)
+    second = reindex(other, _right, right_columns, tqdm=tqdm, pbar=pbar)
+    for name in right_columns:
+        revised_name = unique_name(name, result.columns)
+        result[revised_name] = second[name]
 
-                dest_shm.close()  # finally close dest.
+    if merge_keys is True:
+        result = _merge_keys(kind, T, result, _left,_right, left_keys, right_keys)
+    return result
 
+def _merge_keys(kind, T, result, left, right, left_keys, right_keys):
+    if kind in ["inner", "cross"]:
+        for right_name in right_keys:
+            right_name = unique_name(right_name, T.columns)
+            if right_name in result.columns:
+                del result[right_name]
+    else:
+        if kind == "outer":
+            boolean_map = (left != -1)
+        elif kind == "left":
+            boolean_map = (right == -1)
+        else:
+            raise TypeError(f"bad join type: {kind}")
+        
+        for left_name, right_name in zip(left_keys,right_keys):
+            right_name = unique_name(right_name, T.columns)
+            result = where(result, boolean_map, left_name, right_name, new=left_name)
     return result
 
 
-def left_join(T, other, left_keys, right_keys, left_columns=None, right_columns=None, merge_keys=None, tqdm=_tqdm, pbar=None):
+# -----------------------
+# MULTIPROCESSING SECTION
+# -----------------------
+
+
+def _mp_where(
+    T: Table,
+    mapping: Table,
+    field: str,
+    left: str,
+    right: str,
+    new: str,
+    start: int,
+    end: int,
+    path: Path,
+):
+    """takes from LEFT where criteria is True else RIGHT.
+
+    Args:
+        T (Table): Table to change.
+        mapping (Table): mapping table.
+        field (str): bool field name in mapping
+            if True take left column
+            else take right column
+        left (str): column name
+        right (str): column name
+        new (str): new name
+        start (int): start index
+        end (int): end index
+
+    :returns: None
     """
+    Constr = type(T)
+    criteria = mapping[field][start:end]
+    left_values = T[left][start:end]
+    right_values = T[right][start:end]
+    new_values = np.where(criteria, left_values, right_values)
+    return Constr({new: new_values}, _path=path)
+
+
+def _mp_left_mapping(
+    T: Table,
+    other: Table,
+    left_slice: slice,
+    right_slice: slice,
+    left_keys: List[str],
+    right_keys: List[str],
+    path: Path,
+):
+    """create mapping for left and right keys.
+
     Args:
         T (Table): left table
-        other (Table): right table
+        Other (Table): right table
+        left_slice (slice): slice of left table to perform mapping on.
+        right_slice (slice): slice of right table to perform mapping on.
         left_keys (list): list of keys for the join from left table.
         right_keys (list): list of keys for the join from right table.
-        left_columns (list): list of columns names to retain from left table. 
-            If None, all are retained.
-        right_columns (list): list of columns names to retain from right table. 
-            If None, all are retained.
-        kind (str, optional): 'inner', 'left', 'outer', 'cross'. Defaults to "inner".
-        tqdm (tqdm, optional): tqdm progress counter. Defaults to _tqdm.
-        pbar (tqdm.pbar, optional): tqdm.progressbar. Defaults to None.
-        merge_keys (boolean): merges keys to the left, so that cases where right key is None, a key exists.
-    
-    Returns: 
-        Table: joined table
+        path (pathlib.Path): directory of the main process
 
-    Example:
-    ```
-    SQL:   SELECT number, letter FROM numbers LEFT JOIN letters ON numbers.colour == letters.color
-    ```
-    Tablite: 
-    ```
-    >>> left_join = numbers.left_join(
-        letters, 
-        left_keys=['colour'], 
-        right_keys=['color'], 
-        left_columns=['number'], 
-        right_columns=['letter']
-    )
-    ```
+    Returns:
+        Table: table initiated in the main process' working directory
     """
-    if left_columns is None:
-        left_columns = list(T.columns)
-    if right_columns is None:
-        right_columns = list(other.columns)
-    assert merge_keys in {True,False,None}
-
-    _jointype_check(T, other, left_keys, right_keys, left_columns, right_columns)
+    left = T[left_keys + [left_slice]]
+    right = other[right_keys + [right_slice]]
+    left_index = left.index(*left_keys)
+    right_index = right.index(*right_keys)
+    del left, right
 
-    left_index = T.index(*left_keys)
-    right_index = other.index(*right_keys)
-    LEFT, RIGHT = [], []
+    _left, _right = [], []
     for left_key, left_ixs in left_index.items():
         right_ixs = right_index.get(left_key, (-1,))
         for left_ix in left_ixs:
             for right_ix in right_ixs:
-                LEFT.append(left_ix)
-                RIGHT.append(right_ix)
-
-    LEFT, RIGHT = np.array(LEFT), np.array(RIGHT)  # compress memory of python list to array.
-    f = select_processing_method(len(LEFT) * len(left_columns + right_columns), _sp_join, _mp_join)
-    result = f(T, other, LEFT, RIGHT, left_columns, right_columns, tqdm=tqdm, pbar=pbar)
-
-    if merge_keys is True:
-        boolean_map = (RIGHT == -1)
-        column_names = [n for n in T.columns]  # left side only.
-        for left_name,right_name in zip(left_keys,right_keys):
-            right_name = unique_name(right_name, T.columns)
-            column_names.append(right_name)
-            result = where(result, boolean_map, left_name,right_name,new=left_name)
-    return result
+                _left.append(left_ix)
+                _right.append(right_ix)
+
+    Constr = type(T)
+    mapping = Constr({"left": _left, "right": _right}, _path=path)
+    return mapping
+
+def _mp_inner_mapping(
+    T: Table,
+    other: Table,
+    left_slice: slice,
+    right_slice: slice,
+    left_keys: List[str],
+    right_keys: List[str],
+    path: Path,
+):
+    """create mapping for left and right keys.
 
+    Args:
+        T (Table): left table
+        Other (Table): right table
+        left_slice (slice): slice of left table to perform mapping on.
+        right_slice (slice): slice of right table to perform mapping on.
+        left_keys (list): list of keys for the join from left table.
+        right_keys (list): list of keys for the join from right table.
+        path (pathlib.Path): directory of the main process
 
-def inner_join(T, other, left_keys, right_keys, left_columns=None, right_columns=None, merge_keys=None, tqdm=_tqdm, pbar=None):
-    """
-    :param T: Table (left)
-    :param other: Table (right)
-    :param left_keys: list of keys for the join
-    :param right_keys: list of keys for the join
-    :param left_columns: list of left columns to retain, if None, all are retained.
-    :param right_columns: list of right columns to retain, if None, all are retained.
-    :param merge_keys: merges keys, so that only left key is present
-    :return: new Table
-    Example:
-    SQL:   SELECT number, letter FROM numbers JOIN letters ON numbers.colour == letters.color
-    Tablite: inner_join = numbers.inner_join(
-        letters, left_keys=['colour'], right_keys=['color'], left_columns=['number'], right_columns=['letter']
-        )
+    Returns:
+        Table: table initiated in the main process' working directory
     """
-    if left_columns is None:
-        left_columns = list(T.columns)
-    if right_columns is None:
-        right_columns = list(other.columns)
-    assert merge_keys in {True,False,None}
+    left = T[left_keys + [left_slice]]
+    right = other[right_keys + [right_slice]]
+    left_index = left.index(*left_keys)
+    right_index = right.index(*right_keys)
+    del left, right
 
-    _jointype_check(T, other, left_keys, right_keys, left_columns, right_columns)
-
-    left_index = T.index(*left_keys)
-    right_index = other.index(*right_keys)
-    LEFT, RIGHT = [], []
+    _left, _right = [], []
     for left_key, left_ixs in left_index.items():
         right_ixs = right_index.get(left_key, None)
         if right_ixs is None:
             continue
         for left_ix in left_ixs:
             for right_ix in right_ixs:
-                LEFT.append(left_ix)
-                RIGHT.append(right_ix)
+                _left.append(left_ix)
+                _right.append(right_ix)
 
-    LEFT, RIGHT = np.array(LEFT), np.array(RIGHT)
-    f = select_processing_method(len(LEFT) * len(left_columns + right_columns), _sp_join, _mp_join)
-    result = f(T, other, LEFT, RIGHT, left_columns, right_columns, tqdm=tqdm, pbar=pbar)
+    Constr = type(T)
+    mapping = Constr({"left": _left, "right": _right}, _path=path)
+    return mapping
 
-    if merge_keys:
-        for right_name in right_keys:
-            right_name = unique_name(right_name, T.columns)
-            if right_name in result.columns:
-                del result[right_name]
-    return result
 
+def _mp_outer_mapping(
+    T: Table,
+    other: Table,
+    left_slice: slice,
+    right_slice: slice,
+    left_keys: List[str],
+    right_keys: List[str],
+    path: Path,
+):
+    """create mapping for left and right keys.
 
-def outer_join(T, other, left_keys, right_keys, left_columns=None, right_columns=None, merge_keys=None, tqdm=_tqdm, pbar=None):
-    """
-    :param T: Table (left)
-    :param other: Table (right)
-    :param left_keys: list of keys for the join
-    :param right_keys: list of keys for the join
-    :param left_columns: list of left columns to retain, if None, all are retained.
-    :param right_columns: list of right columns to retain, if None, all are retained.
-    :param merge_keys: merges keys, so that cases where a key match is None, a key exists.
-    :return: new Table
-    Example:
-    SQL:   SELECT number, letter FROM numbers OUTER JOIN letters ON numbers.colour == letters.color
-    Tablite: outer_join = numbers.outer_join(
-        letters, left_keys=['colour'], right_keys=['color'], left_columns=['number'], right_columns=['letter']
-        )
-    """
-    if left_columns is None:
-        left_columns = list(T.columns)
-    if right_columns is None:
-        right_columns = list(other.columns)
-    assert merge_keys in {True,False,None}
+    Args:
+        T (Table): left table
+        Other (Table): right table
+        left_slice (slice): slice of left table to perform mapping on.
+        right_slice (slice): slice of right table to perform mapping on.
+        left_keys (list): list of keys for the join from left table.
+        right_keys (list): list of keys for the join from right table.
+        path (pathlib.Path): directory of the main process
 
-    _jointype_check(T, other, left_keys, right_keys, left_columns, right_columns)
+    Returns:
+        Table: table initiated in the main process' working directory
+    """
+    left = T[left_keys + [left_slice]]
+    right = other[right_keys + [right_slice]]
+    left_index = left.index(*left_keys)
+    right_index = right.index(*right_keys)
+    del left, right
 
-    left_index = T.index(*left_keys)
-    right_index = other.index(*right_keys)
-    LEFT, RIGHT, RIGHT_UNUSED = [], [], set(right_index.keys())
+    _left, _right, _right_unused = [], [], set(right_index.keys())
     for left_key, left_ixs in left_index.items():
         right_ixs = right_index.get(left_key, (-1,))
         for left_ix in left_ixs:
             for right_ix in right_ixs:
-                LEFT.append(left_ix)
-                RIGHT.append(right_ix)
-                RIGHT_UNUSED.discard(left_key)
+                _left.append(left_ix)
+                _right.append(right_ix)
+                _right_unused.discard(left_key)
 
-    for right_key in RIGHT_UNUSED:
+    for right_key in _right_unused:
         for right_ix in right_index[right_key]:
-            LEFT.append(-1)
-            RIGHT.append(right_ix)
+            _left.append(-1)
+            _right.append(right_ix)
 
-    LEFT, RIGHT = np.array(LEFT), np.array(RIGHT)
-    f = select_processing_method(len(LEFT) * len(left_columns + right_columns), _sp_join, _mp_join)
-    result = f(T, other, LEFT, RIGHT, left_columns, right_columns, tqdm=tqdm, pbar=pbar)
+    Constr = type(T)
+    mapping = Constr({"left": _left, "right": _right}, _path=path)
+    return mapping
 
-    if merge_keys is True:
-        boolean_map = (LEFT != -1)
-        column_names = [n for n in T.columns]
-        for left_name,right_name in zip(left_keys,right_keys):
-            right_name = unique_name(right_name, T.columns)
-            column_names.append(right_name)
-            result = where(result, boolean_map, left_name,right_name,new=left_name)
-    return result
 
+def _mp_cross_mapping(
+    T: Table,
+    other: Table,
+    left_slice: slice,
+    right_slice: slice,
+    left_keys: List[str],
+    right_keys: List[str],
+    path: Path,
+):
+    """create mapping for left and right keys.
 
-def cross_join(T, other, left_keys, right_keys, left_columns=None, right_columns=None, merge_keys=None, tqdm=_tqdm, pbar=None):
-    """
-    :param T: Table (left)
-    :param other: Table (right)
-    :param left_keys: list of keys for the join
-    :param right_keys: list of keys for the join
-    :param left_columns: list of left columns to retain, if None, all are retained.
-    :param right_columns: list of right columns to retain, if None, all are retained.
-    :param merge_keys: merges keys, so that only left key is present
-    :return: new Table
+    Args:
+        T (Table): left table
+        Other (Table): right table
+        left_slice (slice): slice of left table to perform mapping on.
+        right_slice (slice): slice of right table to perform mapping on.
+        left_keys (list): list of keys for the join from left table.
+        right_keys (list): list of keys for the join from right table.
+        path (pathlib.Path): directory of the main process
 
-    CROSS JOIN returns the Cartesian product of rows from tables in the join.
-    In other words, it will produce rows which combine each row from the first table
-    with each row from the second table
+    Returns:
+        Table: table initiated in the main process' working directory
+    """ 
+    Constr = type(T)
+    rr = np.arange(right_slice.start, min(right_slice.stop, len(other)))
+    tmp = Constr({"right": rr}, _path=path)
+    right = tmp["right"]
+    rr_shape = rr.shape
+    del rr
+
+    lr = np.arange(left_slice.start, min(left_slice.stop, len(T)))
+    mapping = Constr({"left": [], "right": []}, _path=path)
+    for a in lr:
+        mapping["right"].extend(right)  
+        # by using the right filepointer above, the page-id is 
+        # repeated, costing 0 Mb in storage.
+        mapping["left"].extend(np.full(rr_shape, a))
+    return mapping
+
+
+def _mp_reindex_page(
+    T: Table,
+    column_name: str,
+    mapping: Table,
+    index: str,
+    start: int,
+    end: int,
+    path: Path,
+):
+    """reindexes T on column_name using mapping between start and end.
+
+    Args:
+        T (Table): table to reindex
+        column_name (str): column to reindex
+        mapping (Table): table with mapping
+        index (str): column name of index in mapping
+        start (int): start of range
+        end (int): end of range
+        path (Path): directory of the main process
+
+    Returns:
+        Table: table initiated in the main process' working directory
     """
-    if left_columns is None:
-        left_columns = list(T.columns)
-    if right_columns is None:
-        right_columns = list(other.columns)
-    assert merge_keys in {True,False,None}
+    part = slice(start, end)
+    ix_arr = mapping[index][part]
+    if ix_arr[0]==start and ix_arr[-1] == end-1 and np.all(ix_arr == np.arange(start,end)):  
+        array = T[column_name][part]
+    else:
+        array = T[column_name].get_by_indices(ix_arr)
+        # in the array, the index of -1 will be wrong.
+        # so if there is any -1 in the indices, they will
+        # have to be replaced with Nones
+        mask = ix_arr == -1
+        if np.any(mask):
+            nones = np.full(ix_arr.shape, fill_value=None)
+            array = np.where(mask, nones, array)
 
-    _jointype_check(T, other, left_keys, right_keys, left_columns, right_columns)
 
-    LEFT, RIGHT = zip(*product(range(len(T)), range(len(other))))
+    Constr = type(T)
+    remapped_T = Constr({column_name: array}, _path=path)
 
-    LEFT, RIGHT = np.array(LEFT), np.array(RIGHT)
-    f = select_processing_method(len(LEFT), _sp_join, _mp_join)
-    result = f(T, other, LEFT, RIGHT, left_columns, right_columns, tqdm=tqdm, pbar=pbar)
+    return remapped_T
 
-    if merge_keys:
-        for right_name in right_keys:
-            right_name = unique_name(right_name, T.columns)
-            if right_name in result.columns:
-                del result[right_name]
+
+def _gets(task, *args):
+    """helper to get kwargs of a task
+
+    *Args:
+        names from kwargs to retrieve.
+
+    Returns:
+        tuple: tuple with kw-values in same order as args
+
+    Examples:
+
+    Verbose way:
+    ```
+    >>> col = task.kwarg.get("left")
+    >>> right = task.kwarg.get("right")
+    >>> end = task.kwarg.get("end")
+    ```
+    Compact way:
+    ```
+    >>> col, start, end = task.gets("left", "start", "end")
+    ```
+    """
+    result = tuple()
+    for arg in args:
+        result += (task.kwargs.get(arg),)
     return result
 
+
+_mp_mapping_methods = {
+    'left': _mp_left_mapping,
+    'inner': _mp_inner_mapping,
+    'outer': _mp_outer_mapping,
+    'cross': _mp_cross_mapping
+}
+
+
+def _mp_join(
+        kind: str,
+        T: Table,
+        other: Table,
+        left_keys: List[str],
+        right_keys: List[str],
+        left_columns: Union[List[str], None] = None,
+        right_columns: Union[List[str], None] = None,
+        merge_keys: bool = False,
+        tqdm=_tqdm,
+        pbar=None,
+        TaskManager=None,
+    ):
+
+    Constr = type(T)
+
+    if pbar is None:
+        _pbar_created_here = True
+        pbar = tqdm(total=5, desc="join", disable=Config.TQDM_DISABLE)
+    else:
+        _pbar_created_here = False
+    pbar.update(0)  # pbar start
+
+    class ProgressBar(object):
+        def update(self, n):
+            pbar.update(n / len(tasks))
+
+    # create left and right index
+    if TaskManager is None:
+        TaskManager = _TaskManager
+    
+    tasks = []
+
+    _pid_dir = Path(Config.workdir) / Config.pid
+
+    # step 1: create mapping tasks
+    _mapping = _mp_mapping_methods.get(kind)
+    if _mapping is None:
+        raise ValueError(f"join type unknown: {kind}")
+
+    step = Config.PAGE_SIZE
+    for left in range(0, len(T) + 1, step):
+        for right in range(0, len(other) + 1, step):
+            left_slice = slice(left, left + step, 1)
+            right_slice = slice(right, right + step, 1)
+            task = Task(
+                _mapping,
+                T=T,
+                other=other,
+                left_slice=left_slice,
+                right_slice=right_slice,
+                left_keys=left_keys,
+                right_keys=right_keys,
+                path=_pid_dir,
+            )
+            tasks.append(task)
+
+    with TaskManager(cpu_count=_vpus(tasks)) as tm:
+        results = tm.execute(tasks, pbar=ProgressBar())
+        errors = [s for s in results if isinstance(s, str)]
+
+        if len(errors) > 0:
+            raise Exception(errors[0])
+
+    # step 2: assemble mapping from tasks
+    mapping = Constr({"left": [], "right": []})
+    for result in results:
+        assert isinstance(result, Table)
+        mapping += result
+
+    pbar.update(1)
+
+    # step 3: initiate reindexing tasks
+    tasks = []
+    names = []  # will store (old name, new name) for derefences during assemble.
+    new_table = Constr()
+    n = len(mapping)
+    for name in left_columns:
+        new_table.add_column(name)
+
+        for start in range(0, n + 1, step):
+            names.append((name, name))
+            task = Task(
+                _mp_reindex_page,
+                T=T,
+                column_name=name,
+                mapping=mapping,
+                index="left",
+                start=start,
+                end=min(start + step, n),
+                path=_pid_dir,
+            )
+            tasks.append(task)
+
+    for name in right_columns:
+        new_name = unique_name(name, new_table.columns)
+        new_table.add_column(new_name)
+
+        for start in range(0, n + 1, step):
+            names.append((new_name, name))
+            task = Task(
+                _mp_reindex_page,
+                T=other,
+                column_name=name,
+                mapping=mapping,
+                index="right",
+                start=start,
+                end=min(start + step, n),
+                path=_pid_dir,
+            )
+            tasks.append(task)
+
+    with TaskManager(cpu_count=_vpus(tasks)) as tm:
+        results = tm.execute(tasks, pbar=ProgressBar())
+        errors = [s for s in results if isinstance(s, str)]
+
+        if len(errors) > 0:
+            raise Exception(errors[0])
+
+    # step 4: assemble the result
+    for task, result, (new, old) in zip(tasks, results, names):
+        arr = result[old]
+        new_table[new].extend(arr)
+
+    pbar.n = pbar.total - 1  # needed to overcome floating point error.
+    pbar.refresh()
+
+    # step 5: merge keys (if required)
+    if merge_keys is True:
+        mapping["boolean map"] = np.array(mapping[right] == -1, dtype=bool)
+        step = 1 / len(left_keys)
+        tasks = []
+        for left_name, right_name in zip(left_keys, right_keys):
+            right_name = unique_name(right_name, T.columns)
+            for start, end in Config.page_steps(len(mapping)):
+                task = Task(
+                    _mp_where,
+                    T=new_table,
+                    mapping=mapping,
+                    field="boolean map",
+                    left=left_name,
+                    right=right_name,
+                    new="bmap",
+                    start=start,
+                    end=end,
+                    path=_pid_dir,
+                )
+                tasks.append(task)
+
+        with TaskManager(cpu_count=_vpus(tasks)) as tm:
+            results = tm.execute(tasks, pbar=ProgressBar())
+            errors = [s for s in results if isinstance(s, str)]
+
+            if len(errors) > 0:
+                raise Exception(errors[0])
+
+        for task, result in zip(tasks, results):
+            col, start, end = _gets(task, "left", "start", "end")
+            new_table[col][start:end] = result["bmap"][:]
+
+        for name in right_keys:
+            del new_table[name]
+    else:
+        pbar.update(1)
+
+    if _pbar_created_here:
+        pbar.close()
+
+    return new_table
diff --git a/tablite/merge.py b/tablite/merge.py
index be66f695..e0fef272 100644
--- a/tablite/merge.py
+++ b/tablite/merge.py
@@ -5,7 +5,9 @@
 
 
 def where(T, criteria, left, right, new):
-    """ takes from LEFT where criteria is True else RIGHT.
+    """ takes from LEFT where criteria is True else RIGHT 
+    and creates a single new column.
+    
     :param: T: Table
     :param: criteria: np.array(bool): 
             if True take left column
@@ -23,21 +25,26 @@ def where(T, criteria, left, right, new):
     else:
         criteria = np.array(criteria, dtype='bool')
     
-    new_name = unique_name(new, list(T.columns))
-    T.add_column(new_name)
-    col = T[new_name]
+    new_uq = unique_name(new, list(T.columns))
+    T.add_column(new_uq)
+    col = T[new_uq]
     
     for start,end in Config.page_steps(len(criteria)):
-        indices = np.arange(start,end)
-        left_values = T[left].get_by_indices(indices)
-        right_values = T[right].get_by_indices(indices)
+        left_values = T[left][start:end]
+        right_values = T[right][start:end]
         new_values = np.where(criteria, left_values, right_values)
         col.extend(new_values)
 
-    del T[left]
-    del T[right]
-    if new != new_name and new not in T.columns:
-        T[new] = T[new_name]
-        del T[new_name]
+    if new == right:
+        T[right] = T[new_uq]  # keep column order
+        del T[new_uq]
+        del T[left]
+    elif new == left:
+        T[left] = T[new_uq]  # keep column order
+        del T[new_uq]
+        del T[right]
+    else:
+        T[new] = T[new_uq]
+        del T[left]
+        del T[right]
     return T
-
diff --git a/tablite/nimlite.py b/tablite/nimlite.py
index 51983996..4721f279 100644
--- a/tablite/nimlite.py
+++ b/tablite/nimlite.py
@@ -1,11 +1,14 @@
 import sys
 import psutil
 import platform
+import numpy as np
 from pathlib import Path
+from tablite.base import Table
 from tqdm import tqdm as _tqdm
 from tablite.base import Column
 from tablite.config import Config
 from mplite import Task, TaskManager
+from typing import Literal, Type, TypeVar, TypedDict, Union, List, Tuple
 
 if True:
     paths = sys.argv[:]
@@ -18,7 +21,20 @@
     sys.argv.clear()
     sys.argv.extend(paths)  # importing nim module messes with pythons launch arguments!!!
 
-def text_reader_task(*, pid, path, encoding, dialect, task, import_fields, guess_dtypes):
+
+K = TypeVar("K", bound=Table)
+ValidEncoders = Union[Literal["ENC_UTF8"], Literal["ENC_UTF16"], Literal["ENC_WIN1250"]]
+ValidQuoting = Union[Literal["QUOTE_MINIMAL"], Literal["QUOTE_ALL"], Literal["QUOTE_NONNUMERIC"], Literal["QUOTE_NONE"], Literal["QUOTE_STRINGS"], Literal["QUOTE_NOTNULL"]]
+ColumnSelectorDict = TypedDict(
+    "ColumnSelectorDict", {
+        "column": str,
+        "type": Union[Literal["int"], Literal["float"], Literal["bool"], Literal["str"], Literal["date"], Literal["time"], Literal["datetime"]],
+        "allow_empty": Union[bool, None],
+        "rename": Union[str, None]
+    }
+)
+
+def _text_reader_task(*, path, encoding, dialect, task, import_fields, guess_dtypes):
     return nl.text_reader_task(
         path=path,
         encoding=encoding,
@@ -40,18 +56,18 @@ def text_reader_task(*, pid, path, encoding, dialect, task, import_fields, guess
 
 
 def text_reader(
-    T,
-    pid, path,
-    encoding="ENC_UTF8",
+    T: Type[K],
+    pid: str, path: Union[str, Path],
+    encoding: ValidEncoders ="ENC_UTF8",
     *,
-    first_row_has_headers=True, header_row_index=0,
-    columns=None,
-    start=None, limit=None,
-    guess_datatypes=False,
-    newline='\n', delimiter=',', text_qualifier='"',
-    quoting, strip_leading_and_tailing_whitespace=True,
+    first_row_has_headers: bool=True, header_row_index: int=0,
+    columns: List[Union[str, None]]=None,
+    start: Union[str, None] = None, limit: Union[str, None]=None,
+    guess_datatypes: bool =False,
+    newline: str='\n', delimiter: str=',', text_qualifier: str='"',
+    quoting: ValidQuoting, strip_leading_and_tailing_whitespace: bool=True,
     tqdm=_tqdm
-):
+) -> K:
     assert isinstance(path, Path)
     assert isinstance(pid, Path)
 
@@ -87,14 +103,13 @@ def text_reader(
 
     tasks = [
         Task(
-            text_reader_task,
+            _text_reader_task,
             path=ti_path,
             encoding=ti_encoding,
             dialect=ti_dialect,
             task=t,
             guess_dtypes=ti_guess_dtypes,
-            import_fields=ti_import_fields,
-            pid=pid
+            import_fields=ti_import_fields
         ) for t in ti_tasks
     ]
 
@@ -147,11 +162,11 @@ def text_reader(
     return table
 
 
-def wrap(str_):
+def wrap(str_: str) -> str:
     return '"' + str_.replace('"', '\\"').replace("'", "\\'").replace("\n", "\\n").replace("\t", "\\t") + '"'
 
 
-def collect_cs_info(i: int, columns: dict, res_cols_pass: list, res_cols_fail: list):
+def _collect_cs_info(i: int, columns: dict, res_cols_pass: list, res_cols_fail: list):
     el = {
         k: column[i]
         for k, column in columns.items()
@@ -163,7 +178,7 @@ def collect_cs_info(i: int, columns: dict, res_cols_pass: list, res_cols_fail: l
     return el, col_pass, col_fail
 
 
-def column_select(table, cols, tqdm=_tqdm, TaskManager=TaskManager):
+def column_select(table: K, cols: list[ColumnSelectorDict], tqdm=_tqdm, TaskManager=TaskManager) -> Tuple[K, K]:
     with tqdm(total=100, desc="column select", bar_format='{desc}: {percentage:.1f}%|{bar}{r_bar}') as pbar:
         T = type(table)
         dir_pid = Config.workdir / Config.pid
@@ -187,7 +202,7 @@ def column_select(table, cols, tqdm=_tqdm, TaskManager=TaskManager):
             return (tbl_pass, tbl_fail)
 
         task_list_inp = (
-            collect_cs_info(i, columns, res_cols_pass, res_cols_fail)
+            _collect_cs_info(i, columns, res_cols_pass, res_cols_fail)
             for i in range(page_count)
         )
 
@@ -252,5 +267,5 @@ def extend_table(table, columns):
 
         return tbl_pass, tbl_fail
 
-def read_page(path):
+def read_page(path: Union[str, Path]) -> np.ndarray:
     return nl.read_page(str(path))
\ No newline at end of file
diff --git a/tablite/nimlite.pyi b/tablite/nimlite.pyi
deleted file mode 100644
index c6471778..00000000
--- a/tablite/nimlite.pyi
+++ /dev/null
@@ -1,37 +0,0 @@
-from typing import Literal, Type, TypeVar, TypedDict, Union
-from tqdm import tqdm as _tqdm
-from tablite import Table
-import numpy as np
-from pathlib import Path
-from mplite import TaskManager as _TaskManager
-
-K = TypeVar("K")
-ColumnSelectorDict = TypedDict(
-    "ColumnSelectorDict", {
-        "column": str,
-        "type": Union[Literal["int"], Literal["float"], Literal["bool"], Literal["str"], Literal["date"], Literal["time"], Literal["datetime"]],
-        "allow_empty": Union[bool, None],
-        "rename": Union[str, None]
-    }
-)
-
-def text_reader(
-        T: Type[Table],
-        pid: str, path: str,
-        encoding: Literal["ENC_UTF8"]|Literal["ENC_UTF16"]|Literal["ENC_WIN1250"] = "ENC_UTF8",
-        *,
-        first_row_has_headers: bool = True, header_row_index: int = 0,
-        columns: list[str]|None = None,
-        start: int|None = None, limit: int|None = None,
-        guess_datatypes: bool = False,
-        newline: str = '\n', delimiter: str = ',', text_qualifier: str = '"',
-        quoting: str, strip_leading_and_tailing_whitespace: bool = True,
-        tqdm=_tqdm
-    ) -> Table:
-    pass
-
-def column_select(table: K, cols: list[ColumnSelectorDict], tqdm=_tqdm, TaskManager=_TaskManager) -> tuple[K, K]:
-    pass
-
-def read_page(path: str | Path) -> np.ndarray:
-    pass
\ No newline at end of file
diff --git a/tablite/redux.py b/tablite/redux.py
index ec463030..5848c0fd 100644
--- a/tablite/redux.py
+++ b/tablite/redux.py
@@ -191,7 +191,8 @@ def i(x):
     mask = np.full((len(T),), True)
     for k, v in kwargs.items():
         col = T[k]
-        for start, end, data in col.iter_by_page():
+        for start, end, page in col.iter_by_page():
+            data = page.get()
             if callable(v):
                 vf = np.frompyfunc(v, 1, 1)
                 mask[start:end] = mask[start:end] & np.apply_along_axis(vf, 0, data)
@@ -211,7 +212,8 @@ def drop(T, *args):
     mask = np.full((len(T),), False)
     for name in T.columns:
         col = T[name]
-        for start, end, data in col.iter_by_page():
+        for start, end, page in col.iter_by_page():
+            data = page.get()
             for arg in args:
                 mask[start:end] = mask[start:end] | (data == arg)
 
@@ -231,7 +233,8 @@ def filter_any(T, **kwargs):
     mask = np.full((len(T),), False)
     for k, v in kwargs.items():
         col = T[k]
-        for start, end, data in col.iter_by_page():
+        for start, end, page in col.iter_by_page():
+            data = page.get()
             if callable(v):
                 vf = np.frompyfunc(v, 1, 1)
                 mask[start:end] = mask[start:end] | np.apply_along_axis(vf, 0, data)
diff --git a/tablite/version.py b/tablite/version.py
index 947a112f..86d7841b 100644
--- a/tablite/version.py
+++ b/tablite/version.py
@@ -1,3 +1,3 @@
-major, minor, patch = 2023, 9, 4
+major, minor, patch = 2023, 9, 5
 __version_info__ = (major, minor, patch)
 __version__ = ".".join(str(i) for i in __version_info__)


Test patch:
diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml
index 45db8c07..e1cc5fd5 100644
--- a/.github/workflows/python-test.yml
+++ b/.github/workflows/python-test.yml
@@ -16,7 +16,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, macos-latest, windows-latest]
-        python-version: ["3.8", "3.9", "3.10"]
+        python-version: ["3.9", "3.10", "3.11"]
 
     steps:
     - uses: actions/checkout@v3
diff --git a/tests/test_base.py b/tests/test_base.py
index 7995ea27..d35cd58e 100644
--- a/tests/test_base.py
+++ b/tests/test_base.py
@@ -185,7 +185,8 @@ def test_speed():
         block = t["A"][a:b]  # pure numpy.
         assert len(block) == b - a
     end = time.time()
-    print(f"numpy array: {end-start}")
+    numpy_array = end - start
+    print(f"numpy array: {numpy_array}")
 
     random.seed(42)
     start = time.time()
@@ -195,7 +196,10 @@ def test_speed():
         block = t["A"][a:b].tolist()  # python.
         assert len(block) == b - a
     end = time.time()
-    print(f"python list: {end-start}")
+    python_list = end - start
+    print(f"python list: {python_list}")
+
+    assert numpy_array < python_list, "something is wrong numpy should be faster."
 
 
 def test_immutability_of_pages():
@@ -312,7 +316,13 @@ def test_table_row_functions():
     t2 = Table()
     t2.add_column("A")
     t2.add_columns("B", "C")
-    t2.add_rows(**{"A": [i + max(A) for i in A], "B": [i + max(A) for i in A], "C": [i + max(C) for i in C]})
+    t2.add_rows(
+        **{
+            "A": [i + max(A) for i in A],
+            "B": [i + max(A) for i in A],
+            "C": [i + max(C) for i in C],
+        }
+    )
     t3 = t2.stack(t)
     assert len(t3) == len(t2) + len(t)
 
@@ -557,9 +567,11 @@ def test_get_by_indices():
 
     Config.PAGE_SIZE = old_cfg
 
+
 def fn_foo_table(tbl):
     return tbl
 
+
 def test_page_refcount():
     table = Table({"A": [0, 1, 2, 3], "B": [4, 5, 6, 7]})
 
diff --git a/tests/test_join.py b/tests/test_join.py
index 3213dc3f..55ecaa79 100644
--- a/tests/test_join.py
+++ b/tests/test_join.py
@@ -450,16 +450,17 @@ def test_join_with_key_merge():
     # | 5|None|None|   6|  16|
     # | 6|None|None|   7|  17|
     # +==+====+====+====+====+
+    assert list(c.columns.keys()) == ["A", "B", "C", "D"]
     assert c["A"] == [1,2,3,None,5,None,None]
     assert c['C'] == [1,2,3,None,None,6,7]
 
     d = c.copy().merge("A", "C", new="E", criteria=[v != None for v in c['A']])
-    assert "A" not in d.columns
-    assert "C" not in d.columns
+    assert list(d.columns.keys()) == list("BDE")
     assert d["E"] == [1,2,3,None,5,6,7]
 
 
     e = a.join(b,left_keys=['A'], right_keys=['C'], left_columns=['A', 'B'], right_columns=['C','D'], kind="outer", merge_keys=True)
+    assert list(e.columns.keys()) == list("ABD")
     assert e["A"] == [1,2,3,None,5,6,7]