We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
✨ improve CROSS JOIN UNNEST performance 
Usage logs show queries which use CROSS JOIN UNNEST are often amongst the slowest running queries.

Whilst CROSS JOIN and UNNEST are relatively slow operations, opportunities to improve the performance should be investigated.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `opteryx/operators/cross_join_node.py`
Content:
```
1 # Licensed under the Apache License, Version 2.0 (the "License");
2 # you may not use this file except in compliance with the License.
3 # You may obtain a copy of the License at
4 #
5 #     http://www.apache.org/licenses/LICENSE-2.0
6 #
7 # Unless required by applicable law or agreed to in writing, software
8 # distributed under the License is distributed on an "AS IS" BASIS,
9 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10 # See the License for the specific language governing permissions and
11 # limitations under the License.
12 
13 """
14 Cross Join Node
15 
16 This is a SQL Query Execution Plan Node.
17 
18 This performs a CROSS JOIN - CROSS JOIN is not natively supported by PyArrow so this is written
19 here rather than calling the join() functions
20 """
21 from typing import Generator
22 from typing import Tuple
23 
24 import numpy
25 import pyarrow
26 from orso.schema import FlatColumn
27 
28 from opteryx.managers.expression import NodeType
29 from opteryx.models import Node
30 from opteryx.models import QueryProperties
31 from opteryx.operators import BasePlanNode
32 
33 INTERNAL_BATCH_SIZE: int = 1000  # config
34 MAX_JOIN_SIZE: int = 500  # config
35 
36 
37 def _cross_join_unnest_column(
38     morsels: BasePlanNode, source: Node, target_column: FlatColumn
39 ) -> Generator[pyarrow.Table, None, None]:
40     """
41     Perform a cross join on an unnested column of pyarrow tables.
42 
43     Args:
44         morsels: An iterable of `pyarrow.Table` objects to be cross joined.
45         source: The source node indicating the column.
46         target_column: The column to be unnested.
47 
48     Returns:
49         A generator that yields the resulting `pyarrow.Table` objects.
50     """
51 
52     # Check if the source node type is an identifier, raise error otherwise
53     if source.node_type != NodeType.IDENTIFIER:
54         raise NotImplementedError("Can only CROSS JOIN UNNEST on a column")
55 
56     column_type = None
57     batch_size: int = INTERNAL_BATCH_SIZE
58 
59     # Loop through each morsel from the morsels execution
60     for left_morsel in morsels.execute():
61         # Break the morsel into batches to avoid memory issues
62         for left_block in left_morsel.to_batches(max_chunksize=batch_size):
63             # Fetch the data of the column to be unnested
64             column_data = left_block[source.schema_column.identity]
65 
66             # Set column_type if it hasn't been determined already
67             if column_type is None:
68                 column_type = column_data.type.value_type
69 
70             from opteryx.compiled import build_rows_indices_and_column
71 
72             indices, new_column_data = build_rows_indices_and_column(column_data)
73 
74             # If no new data was generated, skip to next iteration
75             if not new_column_data:
76                 continue
77 
78             new_block = left_morsel.take(indices)
79             new_block = new_block.append_column(target_column.identity, [new_column_data])
80             yield new_block
81 
82             if batch_size == INTERNAL_BATCH_SIZE:
83                 # we size the batches based on observations
84                 batch_size = int((INTERNAL_BATCH_SIZE / new_block.nbytes) * 8 * 1024 * 1024)
85 
86 
87 def _cross_join_unnest_literal(
88     morsels: BasePlanNode, source: Tuple, target_column: FlatColumn
89 ) -> Generator[pyarrow.Table, None, None]:
90     joined_list_size = len(source)
91 
92     # Loop through each morsel from the morsels execution
93     for left_morsel in morsels.execute():
94         # Break the morsel into batches to avoid memory issues
95         for left_block in left_morsel.to_batches(max_chunksize=INTERNAL_BATCH_SIZE):
96             left_block = pyarrow.Table.from_batches([left_block], schema=left_morsel.schema)
97             block_size = left_block.num_rows
98 
99             # Repeat each row in the table n times
100             repeated_indices = numpy.repeat(numpy.arange(block_size), joined_list_size)
101             appended_table = left_block.take(repeated_indices)
102 
103             # Tile the array to match the new number of rows
104             tiled_array = numpy.tile(source, block_size)
105 
106             # Convert tiled_array to PyArrow array and append it to the table
107             array_column = pyarrow.array(tiled_array)
108             appended_table = appended_table.append_column(target_column.identity, array_column)
109 
110             yield appended_table
111 
112 
113 def _cartesian_product(*arrays):
114     """
115     Cartesian product of arrays creates every combination of the elements in the arrays
116     """
117     array_count = len(arrays)
118     arr = numpy.empty([len(array) for array in arrays] + [array_count], dtype=numpy.int64)
119     for i, array in enumerate(numpy.ix_(*arrays)):
120         arr[..., i] = array
121     return numpy.hsplit(arr.reshape(-1, array_count), array_count)
122 
123 
124 def _cross_join(left, right):
125     """
126     A cross join is the cartesian product of two tables - this usually isn't very
127     useful, but it does allow you to the theta joins (non-equi joins)
128     """
129 
130     def _chunker(seq_1, seq_2, size):
131         """
132         Chunk two equal length interables into size sized chunks
133 
134         This returns a generator.
135         """
136         return (
137             (seq_1[pos : pos + size], seq_2[pos : pos + size]) for pos in range(0, len(seq_1), size)
138         )
139 
140     from opteryx.third_party.pyarrow_ops import align_tables
141 
142     for left_morsel in left.execute():
143         # Iterate through left table in chunks of size INTERNAL_BATCH_SIZE
144         for left_block in left_morsel.to_batches(max_chunksize=INTERNAL_BATCH_SIZE):
145             # Convert the chunk to a table to retain column names
146             left_block = pyarrow.Table.from_batches([left_block], schema=left_morsel.schema)
147 
148             # Create an array of row indices for each table
149             left_array = numpy.arange(left_block.num_rows, dtype=numpy.int64)
150             right_array = numpy.arange(right.num_rows, dtype=numpy.int64)
151 
152             # Calculate the cartesian product of the two arrays of row indices
153             left_align, right_align = _cartesian_product(left_array, right_array)
154 
155             # Further break down the result into manageable chunks of size MAX_JOIN_SIZE
156             for left_chunk, right_chunk in _chunker(left_align, right_align, MAX_JOIN_SIZE):
157                 # Align the tables using the specified chunks of row indices
158                 table = align_tables(left_block, right, left_chunk.flatten(), right_chunk.flatten())
159 
160                 # Yield the resulting table to the caller
161                 yield table
162 
163 
164 class CrossJoinNode(BasePlanNode):
165     """
166     Implements a SQL CROSS JOIN
167     """
168 
169     def __init__(self, properties: QueryProperties, **config):
170         super().__init__(properties=properties)
171 
172         self.source = config.get("column")
173 
174         # do we have unnest details?
175         self._unnest_column = config.get("unnest_column")
176         self._unnest_target = config.get("unnest_target")
177 
178         # handle variation in how the unnested column is represented
179         if self._unnest_column:
180             if self._unnest_column.node_type == NodeType.NESTED:
181                 self._unnest_column = self._unnest_column.centre
182             # if we have a literal that's not a tuple, wrap it
183             if self._unnest_column.node_type == NodeType.LITERAL and not isinstance(
184                 self._unnest_column.value, tuple
185             ):
186                 self._unnest_column.value = tuple([self._unnest_column.value])
187 
188     @property
189     def name(self):  # pragma: no cover
190         return "Cross Join"
191 
192     @property
193     def config(self):  # pragma: no cover
194         return ""
195 
196     def execute(self) -> Generator:
197         left_node = self._producers[0]  # type:ignore
198         right_node = self._producers[1]  # type:ignore
199         right_table = pyarrow.concat_tables(
200             right_node.execute(), promote_options="none"
201         )  # type:ignore
202 
203         if self._unnest_column is None:
204             yield from _cross_join(left_node, right_table)
205 
206         elif isinstance(self._unnest_column.value, tuple):
207             yield from _cross_join_unnest_literal(
208                 morsels=left_node,
209                 source=self._unnest_column.value,
210                 target_column=self._unnest_target,
211             )
212         else:
213             yield from _cross_join_unnest_column(
214                 morsels=left_node, source=self._unnest_column, target_column=self._unnest_target
215             )
216
```

Path: `opteryx/__version__.py`
Content:
```
1 __build__ = 271
2 
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 """
16 Store the version here so:
17 1) we don't load dependencies by storing it in __init__.py
18 2) we can import it in setup.py for the same reason
19 """
20 from enum import Enum  # isort: skip
21 
22 
23 class VersionStatus(Enum):
24     ALPHA = "alpha"
25     BETA = "beta"
26     RELEASE = "release"
27 
28 
29 _major = 0
30 _minor = 13
31 _revision = 2
32 _status = VersionStatus.RELEASE
33 
34 __author__ = "@joocer"
35 __version__ = f"{_major}.{_minor}.{_revision}" + (
36     f"-{_status.value}.{__build__}" if _status != VersionStatus.RELEASE else ""
37 )
38
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/opteryx/__version__.py b/opteryx/__version__.py
--- a/opteryx/__version__.py
+++ b/opteryx/__version__.py
@@ -1,4 +1,4 @@
-__build__ = 271
+__build__ = 276
 
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -27,9 +27,9 @@
 
 
 _major = 0
-_minor = 13
-_revision = 2
-_status = VersionStatus.RELEASE
+_minor = 14
+_revision = 0
+_status = VersionStatus.ALPHA
 
 __author__ = "@joocer"
 __version__ = f"{_major}.{_minor}.{_revision}" + (
diff --git a/opteryx/operators/cross_join_node.py b/opteryx/operators/cross_join_node.py
--- a/opteryx/operators/cross_join_node.py
+++ b/opteryx/operators/cross_join_node.py
@@ -48,6 +48,7 @@
     Returns:
         A generator that yields the resulting `pyarrow.Table` objects.
     """
+    from opteryx.compiled import build_rows_indices_and_column
 
     # Check if the source node type is an identifier, raise error otherwise
     if source.node_type != NodeType.IDENTIFIER:
@@ -62,20 +63,26 @@
         for left_block in left_morsel.to_batches(max_chunksize=batch_size):
             # Fetch the data of the column to be unnested
             column_data = left_block[source.schema_column.identity]
-
+            # we need the offsets before we drop the rows
+            valid_offsets = column_data.is_valid()
+            column_data = column_data.drop_null()
+            # if there's no valid records, continue to the next record
+            if len(column_data) == 0:
+                continue
+            # drop the rows here, wait until we know we need to
+            left_block = left_block.filter(valid_offsets)
             # Set column_type if it hasn't been determined already
             if column_type is None:
                 column_type = column_data.type.value_type
 
-            from opteryx.compiled import build_rows_indices_and_column
-
-            indices, new_column_data = build_rows_indices_and_column(column_data)
+            indices, new_column_data = build_rows_indices_and_column(column_data.to_numpy(False))
 
             # If no new data was generated, skip to next iteration
-            if not new_column_data:
+            if len(new_column_data) == 0:
                 continue
 
-            new_block = left_morsel.take(indices)
+            new_block = left_block.take(indices)
+            new_block = pyarrow.Table.from_batches([new_block], schema=left_morsel.schema)
             new_block = new_block.append_column(target_column.identity, [new_column_data])
             yield new_block

{"golden_diff": "diff --git a/opteryx/__version__.py b/opteryx/__version__.py\n--- a/opteryx/__version__.py\n+++ b/opteryx/__version__.py\n@@ -1,4 +1,4 @@\n-__build__ = 271\n+__build__ = 276\n \n # Licensed under the Apache License, Version 2.0 (the \"License\");\n # you may not use this file except in compliance with the License.\n@@ -27,9 +27,9 @@\n \n \n _major = 0\n-_minor = 13\n-_revision = 2\n-_status = VersionStatus.RELEASE\n+_minor = 14\n+_revision = 0\n+_status = VersionStatus.ALPHA\n \n __author__ = \"@joocer\"\n __version__ = f\"{_major}.{_minor}.{_revision}\" + (\ndiff --git a/opteryx/operators/cross_join_node.py b/opteryx/operators/cross_join_node.py\n--- a/opteryx/operators/cross_join_node.py\n+++ b/opteryx/operators/cross_join_node.py\n@@ -48,6 +48,7 @@\n     Returns:\n         A generator that yields the resulting `pyarrow.Table` objects.\n     \"\"\"\n+    from opteryx.compiled import build_rows_indices_and_column\n \n     # Check if the source node type is an identifier, raise error otherwise\n     if source.node_type != NodeType.IDENTIFIER:\n@@ -62,20 +63,26 @@\n         for left_block in left_morsel.to_batches(max_chunksize=batch_size):\n             # Fetch the data of the column to be unnested\n             column_data = left_block[source.schema_column.identity]\n-\n+            # we need the offsets before we drop the rows\n+            valid_offsets = column_data.is_valid()\n+            column_data = column_data.drop_null()\n+            # if there's no valid records, continue to the next record\n+            if len(column_data) == 0:\n+                continue\n+            # drop the rows here, wait until we know we need to\n+            left_block = left_block.filter(valid_offsets)\n             # Set column_type if it hasn't been determined already\n             if column_type is None:\n                 column_type = column_data.type.value_type\n \n-            from opteryx.compiled import build_rows_indices_and_column\n-\n-            indices, new_column_data = build_rows_indices_and_column(column_data)\n+            indices, new_column_data = build_rows_indices_and_column(column_data.to_numpy(False))\n \n             # If no new data was generated, skip to next iteration\n-            if not new_column_data:\n+            if len(new_column_data) == 0:\n                 continue\n \n-            new_block = left_morsel.take(indices)\n+            new_block = left_block.take(indices)\n+            new_block = pyarrow.Table.from_batches([new_block], schema=left_morsel.schema)\n             new_block = new_block.append_column(target_column.identity, [new_column_data])\n             yield new_block\n", "issue": "\u2728 improve CROSS JOIN UNNEST performance \nUsage logs show queries which use CROSS JOIN UNNEST are often amongst the slowest running queries.\n\nWhilst CROSS JOIN and UNNEST are relatively slow operations, opportunities to improve the performance should be investigated.\n", "before_files": [{"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nCross Join Node\n\nThis is a SQL Query Execution Plan Node.\n\nThis performs a CROSS JOIN - CROSS JOIN is not natively supported by PyArrow so this is written\nhere rather than calling the join() functions\n\"\"\"\nfrom typing import Generator\nfrom typing import Tuple\n\nimport numpy\nimport pyarrow\nfrom orso.schema import FlatColumn\n\nfrom opteryx.managers.expression import NodeType\nfrom opteryx.models import Node\nfrom opteryx.models import QueryProperties\nfrom opteryx.operators import BasePlanNode\n\nINTERNAL_BATCH_SIZE: int = 1000  # config\nMAX_JOIN_SIZE: int = 500  # config\n\n\ndef _cross_join_unnest_column(\n    morsels: BasePlanNode, source: Node, target_column: FlatColumn\n) -> Generator[pyarrow.Table, None, None]:\n    \"\"\"\n    Perform a cross join on an unnested column of pyarrow tables.\n\n    Args:\n        morsels: An iterable of `pyarrow.Table` objects to be cross joined.\n        source: The source node indicating the column.\n        target_column: The column to be unnested.\n\n    Returns:\n        A generator that yields the resulting `pyarrow.Table` objects.\n    \"\"\"\n\n    # Check if the source node type is an identifier, raise error otherwise\n    if source.node_type != NodeType.IDENTIFIER:\n        raise NotImplementedError(\"Can only CROSS JOIN UNNEST on a column\")\n\n    column_type = None\n    batch_size: int = INTERNAL_BATCH_SIZE\n\n    # Loop through each morsel from the morsels execution\n    for left_morsel in morsels.execute():\n        # Break the morsel into batches to avoid memory issues\n        for left_block in left_morsel.to_batches(max_chunksize=batch_size):\n            # Fetch the data of the column to be unnested\n            column_data = left_block[source.schema_column.identity]\n\n            # Set column_type if it hasn't been determined already\n            if column_type is None:\n                column_type = column_data.type.value_type\n\n            from opteryx.compiled import build_rows_indices_and_column\n\n            indices, new_column_data = build_rows_indices_and_column(column_data)\n\n            # If no new data was generated, skip to next iteration\n            if not new_column_data:\n                continue\n\n            new_block = left_morsel.take(indices)\n            new_block = new_block.append_column(target_column.identity, [new_column_data])\n            yield new_block\n\n            if batch_size == INTERNAL_BATCH_SIZE:\n                # we size the batches based on observations\n                batch_size = int((INTERNAL_BATCH_SIZE / new_block.nbytes) * 8 * 1024 * 1024)\n\n\ndef _cross_join_unnest_literal(\n    morsels: BasePlanNode, source: Tuple, target_column: FlatColumn\n) -> Generator[pyarrow.Table, None, None]:\n    joined_list_size = len(source)\n\n    # Loop through each morsel from the morsels execution\n    for left_morsel in morsels.execute():\n        # Break the morsel into batches to avoid memory issues\n        for left_block in left_morsel.to_batches(max_chunksize=INTERNAL_BATCH_SIZE):\n            left_block = pyarrow.Table.from_batches([left_block], schema=left_morsel.schema)\n            block_size = left_block.num_rows\n\n            # Repeat each row in the table n times\n            repeated_indices = numpy.repeat(numpy.arange(block_size), joined_list_size)\n            appended_table = left_block.take(repeated_indices)\n\n            # Tile the array to match the new number of rows\n            tiled_array = numpy.tile(source, block_size)\n\n            # Convert tiled_array to PyArrow array and append it to the table\n            array_column = pyarrow.array(tiled_array)\n            appended_table = appended_table.append_column(target_column.identity, array_column)\n\n            yield appended_table\n\n\ndef _cartesian_product(*arrays):\n    \"\"\"\n    Cartesian product of arrays creates every combination of the elements in the arrays\n    \"\"\"\n    array_count = len(arrays)\n    arr = numpy.empty([len(array) for array in arrays] + [array_count], dtype=numpy.int64)\n    for i, array in enumerate(numpy.ix_(*arrays)):\n        arr[..., i] = array\n    return numpy.hsplit(arr.reshape(-1, array_count), array_count)\n\n\ndef _cross_join(left, right):\n    \"\"\"\n    A cross join is the cartesian product of two tables - this usually isn't very\n    useful, but it does allow you to the theta joins (non-equi joins)\n    \"\"\"\n\n    def _chunker(seq_1, seq_2, size):\n        \"\"\"\n        Chunk two equal length interables into size sized chunks\n\n        This returns a generator.\n        \"\"\"\n        return (\n            (seq_1[pos : pos + size], seq_2[pos : pos + size]) for pos in range(0, len(seq_1), size)\n        )\n\n    from opteryx.third_party.pyarrow_ops import align_tables\n\n    for left_morsel in left.execute():\n        # Iterate through left table in chunks of size INTERNAL_BATCH_SIZE\n        for left_block in left_morsel.to_batches(max_chunksize=INTERNAL_BATCH_SIZE):\n            # Convert the chunk to a table to retain column names\n            left_block = pyarrow.Table.from_batches([left_block], schema=left_morsel.schema)\n\n            # Create an array of row indices for each table\n            left_array = numpy.arange(left_block.num_rows, dtype=numpy.int64)\n            right_array = numpy.arange(right.num_rows, dtype=numpy.int64)\n\n            # Calculate the cartesian product of the two arrays of row indices\n            left_align, right_align = _cartesian_product(left_array, right_array)\n\n            # Further break down the result into manageable chunks of size MAX_JOIN_SIZE\n            for left_chunk, right_chunk in _chunker(left_align, right_align, MAX_JOIN_SIZE):\n                # Align the tables using the specified chunks of row indices\n                table = align_tables(left_block, right, left_chunk.flatten(), right_chunk.flatten())\n\n                # Yield the resulting table to the caller\n                yield table\n\n\nclass CrossJoinNode(BasePlanNode):\n    \"\"\"\n    Implements a SQL CROSS JOIN\n    \"\"\"\n\n    def __init__(self, properties: QueryProperties, **config):\n        super().__init__(properties=properties)\n\n        self.source = config.get(\"column\")\n\n        # do we have unnest details?\n        self._unnest_column = config.get(\"unnest_column\")\n        self._unnest_target = config.get(\"unnest_target\")\n\n        # handle variation in how the unnested column is represented\n        if self._unnest_column:\n            if self._unnest_column.node_type == NodeType.NESTED:\n                self._unnest_column = self._unnest_column.centre\n            # if we have a literal that's not a tuple, wrap it\n            if self._unnest_column.node_type == NodeType.LITERAL and not isinstance(\n                self._unnest_column.value, tuple\n            ):\n                self._unnest_column.value = tuple([self._unnest_column.value])\n\n    @property\n    def name(self):  # pragma: no cover\n        return \"Cross Join\"\n\n    @property\n    def config(self):  # pragma: no cover\n        return \"\"\n\n    def execute(self) -> Generator:\n        left_node = self._producers[0]  # type:ignore\n        right_node = self._producers[1]  # type:ignore\n        right_table = pyarrow.concat_tables(\n            right_node.execute(), promote_options=\"none\"\n        )  # type:ignore\n\n        if self._unnest_column is None:\n            yield from _cross_join(left_node, right_table)\n\n        elif isinstance(self._unnest_column.value, tuple):\n            yield from _cross_join_unnest_literal(\n                morsels=left_node,\n                source=self._unnest_column.value,\n                target_column=self._unnest_target,\n            )\n        else:\n            yield from _cross_join_unnest_column(\n                morsels=left_node, source=self._unnest_column, target_column=self._unnest_target\n            )\n", "path": "opteryx/operators/cross_join_node.py"}, {"content": "__build__ = 271\n\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nStore the version here so:\n1) we don't load dependencies by storing it in __init__.py\n2) we can import it in setup.py for the same reason\n\"\"\"\nfrom enum import Enum  # isort: skip\n\n\nclass VersionStatus(Enum):\n    ALPHA = \"alpha\"\n    BETA = \"beta\"\n    RELEASE = \"release\"\n\n\n_major = 0\n_minor = 13\n_revision = 2\n_status = VersionStatus.RELEASE\n\n__author__ = \"@joocer\"\n__version__ = f\"{_major}.{_minor}.{_revision}\" + (\n    f\"-{_status.value}.{__build__}\" if _status != VersionStatus.RELEASE else \"\"\n)\n", "path": "opteryx/__version__.py"}], "after_files": [{"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nCross Join Node\n\nThis is a SQL Query Execution Plan Node.\n\nThis performs a CROSS JOIN - CROSS JOIN is not natively supported by PyArrow so this is written\nhere rather than calling the join() functions\n\"\"\"\nfrom typing import Generator\nfrom typing import Tuple\n\nimport numpy\nimport pyarrow\nfrom orso.schema import FlatColumn\n\nfrom opteryx.managers.expression import NodeType\nfrom opteryx.models import Node\nfrom opteryx.models import QueryProperties\nfrom opteryx.operators import BasePlanNode\n\nINTERNAL_BATCH_SIZE: int = 1000  # config\nMAX_JOIN_SIZE: int = 500  # config\n\n\ndef _cross_join_unnest_column(\n    morsels: BasePlanNode, source: Node, target_column: FlatColumn\n) -> Generator[pyarrow.Table, None, None]:\n    \"\"\"\n    Perform a cross join on an unnested column of pyarrow tables.\n\n    Args:\n        morsels: An iterable of `pyarrow.Table` objects to be cross joined.\n        source: The source node indicating the column.\n        target_column: The column to be unnested.\n\n    Returns:\n        A generator that yields the resulting `pyarrow.Table` objects.\n    \"\"\"\n    from opteryx.compiled import build_rows_indices_and_column\n\n    # Check if the source node type is an identifier, raise error otherwise\n    if source.node_type != NodeType.IDENTIFIER:\n        raise NotImplementedError(\"Can only CROSS JOIN UNNEST on a column\")\n\n    column_type = None\n    batch_size: int = INTERNAL_BATCH_SIZE\n\n    # Loop through each morsel from the morsels execution\n    for left_morsel in morsels.execute():\n        # Break the morsel into batches to avoid memory issues\n        for left_block in left_morsel.to_batches(max_chunksize=batch_size):\n            # Fetch the data of the column to be unnested\n            column_data = left_block[source.schema_column.identity]\n            # we need the offsets before we drop the rows\n            valid_offsets = column_data.is_valid()\n            column_data = column_data.drop_null()\n            # if there's no valid records, continue to the next record\n            if len(column_data) == 0:\n                continue\n            # drop the rows here, wait until we know we need to\n            left_block = left_block.filter(valid_offsets)\n            # Set column_type if it hasn't been determined already\n            if column_type is None:\n                column_type = column_data.type.value_type\n\n            indices, new_column_data = build_rows_indices_and_column(column_data.to_numpy(False))\n\n            # If no new data was generated, skip to next iteration\n            if len(new_column_data) == 0:\n                continue\n\n            new_block = left_block.take(indices)\n            new_block = pyarrow.Table.from_batches([new_block], schema=left_morsel.schema)\n            new_block = new_block.append_column(target_column.identity, [new_column_data])\n            yield new_block\n\n            if batch_size == INTERNAL_BATCH_SIZE:\n                # we size the batches based on observations\n                batch_size = int((INTERNAL_BATCH_SIZE / new_block.nbytes) * 8 * 1024 * 1024)\n\n\ndef _cross_join_unnest_literal(\n    morsels: BasePlanNode, source: Tuple, target_column: FlatColumn\n) -> Generator[pyarrow.Table, None, None]:\n    joined_list_size = len(source)\n\n    # Loop through each morsel from the morsels execution\n    for left_morsel in morsels.execute():\n        # Break the morsel into batches to avoid memory issues\n        for left_block in left_morsel.to_batches(max_chunksize=INTERNAL_BATCH_SIZE):\n            left_block = pyarrow.Table.from_batches([left_block], schema=left_morsel.schema)\n            block_size = left_block.num_rows\n\n            # Repeat each row in the table n times\n            repeated_indices = numpy.repeat(numpy.arange(block_size), joined_list_size)\n            appended_table = left_block.take(repeated_indices)\n\n            # Tile the array to match the new number of rows\n            tiled_array = numpy.tile(source, block_size)\n\n            # Convert tiled_array to PyArrow array and append it to the table\n            array_column = pyarrow.array(tiled_array)\n            appended_table = appended_table.append_column(target_column.identity, array_column)\n\n            yield appended_table\n\n\ndef _cartesian_product(*arrays):\n    \"\"\"\n    Cartesian product of arrays creates every combination of the elements in the arrays\n    \"\"\"\n    array_count = len(arrays)\n    arr = numpy.empty([len(array) for array in arrays] + [array_count], dtype=numpy.int64)\n    for i, array in enumerate(numpy.ix_(*arrays)):\n        arr[..., i] = array\n    return numpy.hsplit(arr.reshape(-1, array_count), array_count)\n\n\ndef _cross_join(left, right):\n    \"\"\"\n    A cross join is the cartesian product of two tables - this usually isn't very\n    useful, but it does allow you to the theta joins (non-equi joins)\n    \"\"\"\n\n    def _chunker(seq_1, seq_2, size):\n        \"\"\"\n        Chunk two equal length interables into size sized chunks\n\n        This returns a generator.\n        \"\"\"\n        return (\n            (seq_1[pos : pos + size], seq_2[pos : pos + size]) for pos in range(0, len(seq_1), size)\n        )\n\n    from opteryx.third_party.pyarrow_ops import align_tables\n\n    for left_morsel in left.execute():\n        # Iterate through left table in chunks of size INTERNAL_BATCH_SIZE\n        for left_block in left_morsel.to_batches(max_chunksize=INTERNAL_BATCH_SIZE):\n            # Convert the chunk to a table to retain column names\n            left_block = pyarrow.Table.from_batches([left_block], schema=left_morsel.schema)\n\n            # Create an array of row indices for each table\n            left_array = numpy.arange(left_block.num_rows, dtype=numpy.int64)\n            right_array = numpy.arange(right.num_rows, dtype=numpy.int64)\n\n            # Calculate the cartesian product of the two arrays of row indices\n            left_align, right_align = _cartesian_product(left_array, right_array)\n\n            # Further break down the result into manageable chunks of size MAX_JOIN_SIZE\n            for left_chunk, right_chunk in _chunker(left_align, right_align, MAX_JOIN_SIZE):\n                # Align the tables using the specified chunks of row indices\n                table = align_tables(left_block, right, left_chunk.flatten(), right_chunk.flatten())\n\n                # Yield the resulting table to the caller\n                yield table\n\n\nclass CrossJoinNode(BasePlanNode):\n    \"\"\"\n    Implements a SQL CROSS JOIN\n    \"\"\"\n\n    def __init__(self, properties: QueryProperties, **config):\n        super().__init__(properties=properties)\n\n        self.source = config.get(\"column\")\n\n        # do we have unnest details?\n        self._unnest_column = config.get(\"unnest_column\")\n        self._unnest_target = config.get(\"unnest_target\")\n\n        # handle variation in how the unnested column is represented\n        if self._unnest_column:\n            if self._unnest_column.node_type == NodeType.NESTED:\n                self._unnest_column = self._unnest_column.centre\n            # if we have a literal that's not a tuple, wrap it\n            if self._unnest_column.node_type == NodeType.LITERAL and not isinstance(\n                self._unnest_column.value, tuple\n            ):\n                self._unnest_column.value = tuple([self._unnest_column.value])\n\n    @property\n    def name(self):  # pragma: no cover\n        return \"Cross Join\"\n\n    @property\n    def config(self):  # pragma: no cover\n        return \"\"\n\n    def execute(self) -> Generator:\n        left_node = self._producers[0]  # type:ignore\n        right_node = self._producers[1]  # type:ignore\n        right_table = pyarrow.concat_tables(\n            right_node.execute(), promote_options=\"none\"\n        )  # type:ignore\n\n        if self._unnest_column is None:\n            yield from _cross_join(left_node, right_table)\n\n        elif isinstance(self._unnest_column.value, tuple):\n            yield from _cross_join_unnest_literal(\n                morsels=left_node,\n                source=self._unnest_column.value,\n                target_column=self._unnest_target,\n            )\n        else:\n            yield from _cross_join_unnest_column(\n                morsels=left_node, source=self._unnest_column, target_column=self._unnest_target\n            )\n", "path": "opteryx/operators/cross_join_node.py"}, {"content": "__build__ = 276\n\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nStore the version here so:\n1) we don't load dependencies by storing it in __init__.py\n2) we can import it in setup.py for the same reason\n\"\"\"\nfrom enum import Enum  # isort: skip\n\n\nclass VersionStatus(Enum):\n    ALPHA = \"alpha\"\n    BETA = \"beta\"\n    RELEASE = \"release\"\n\n\n_major = 0\n_minor = 14\n_revision = 0\n_status = VersionStatus.ALPHA\n\n__author__ = \"@joocer\"\n__version__ = f\"{_major}.{_minor}.{_revision}\" + (\n    f\"-{_status.value}.{__build__}\" if _status != VersionStatus.RELEASE else \"\"\n)\n", "path": "opteryx/__version__.py"}]}