We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
✨ support pyarrow's IPC format
### Thanks for stopping by to let us know something could be better!

**Is your feature request related to a problem? Please describe.** _A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]_

**Describe the solution you'd like** _A clear and concise description of what you want to happen._

**Describe alternatives you've considered** _A clear and concise description of any alternative solutions or features you've considered._

**Additional context** _Add any other context or screenshots about the feature request here._


--- 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/utils/file_decoders.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 Decode files from a raw binary format to a PyArrow Table.
15 """
16 import io
17 from enum import Enum
18 from typing import Callable
19 from typing import Dict
20 from typing import List
21 from typing import Tuple
22 
23 import numpy
24 import pyarrow
25 from orso.tools import random_string
26 
27 from opteryx.exceptions import UnsupportedFileTypeError
28 from opteryx.utils.arrow import post_read_projector
29 
30 
31 class ExtentionType(str, Enum):
32     """labels for the file extentions"""
33 
34     DATA = "DATA"
35     CONTROL = "CONTROL"
36 
37 
38 def convert_arrow_schema_to_orso_schema(arrow_schema):
39     from orso.schema import FlatColumn
40     from orso.schema import RelationSchema
41 
42     return RelationSchema(
43         name="arrow",
44         columns=[FlatColumn.from_arrow(field) for field in arrow_schema],
45     )
46 
47 
48 def get_decoder(dataset: str) -> Callable:
49     """helper routine to get the decoder for a given file"""
50     ext = dataset.split(".")[-1].lower()
51     if ext not in KNOWN_EXTENSIONS:  # pragma: no cover
52         raise UnsupportedFileTypeError(f"Unsupported file type - {ext}")
53     file_decoder, file_type = KNOWN_EXTENSIONS[ext]
54     if file_type != ExtentionType.DATA:  # pragma: no cover
55         raise UnsupportedFileTypeError(f"File is not a data file - {ext}")
56     return file_decoder
57 
58 
59 def do_nothing(buffer, projection=None, just_schema: bool = False):  # pragma: no cover
60     """for when you need to look like you're doing something"""
61     return False
62 
63 
64 def filter_records(filter, table):
65     """
66     When we can't push predicates to the actual read, use this to filter records
67     just after the read.
68     """
69     # notes:
70     #   at this point we've not renamed any columns, this may affect some filters
71     from opteryx.managers.expression import NodeType
72     from opteryx.managers.expression import evaluate
73     from opteryx.models import Node
74 
75     if isinstance(filter, list) and filter:
76         filter_copy = [p for p in filter]
77         root = filter_copy.pop()
78 
79         if root.left.node_type == NodeType.IDENTIFIER:
80             root.left.schema_column.identity = root.left.source_column
81         if root.right.node_type == NodeType.IDENTIFIER:
82             root.right.schema_column.identity = root.right.source_column
83 
84         while filter_copy:
85             right = filter_copy.pop()
86             if right.left.node_type == NodeType.IDENTIFIER:
87                 right.left.schema_column.identity = right.left.source_column
88             if right.right.node_type == NodeType.IDENTIFIER:
89                 right.right.schema_column.identity = right.right.source_column
90             root = Node(
91                 NodeType.AND, left=root, right=right, schema_column=Node(identity=random_string())
92             )
93     else:
94         root = filter
95 
96     mask = evaluate(root, table)
97     return table.filter(mask)
98 
99 
100 def zstd_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):
101     """
102     Read zstandard compressed JSONL files
103     """
104     import zstandard
105 
106     stream = io.BytesIO(buffer)
107 
108     with zstandard.open(stream, "rb") as file:
109         return jsonl_decoder(
110             file, projection=projection, selection=selection, just_schema=just_schema
111         )
112 
113 
114 def parquet_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):
115     """
116     Read parquet formatted files.
117     """
118     from pyarrow import parquet
119 
120     from opteryx.connectors.capabilities import PredicatePushable
121 
122     # Convert the selection to DNF format if applicable
123     _select, selection = PredicatePushable.to_dnf(selection) if selection else (None, None)
124 
125     stream = io.BytesIO(buffer)
126 
127     if projection or just_schema:
128         # Open the parquet file only once
129         parquet_file = parquet.ParquetFile(stream)
130 
131         # Return just the schema if that's all that's needed
132         if just_schema:
133             return convert_arrow_schema_to_orso_schema(parquet_file.schema_arrow)
134 
135         # Special handling for projection of ["count_*"]
136         if projection == ["count_*"]:
137             return pyarrow.Table.from_pydict(
138                 {"_": numpy.full(parquet_file.metadata.num_rows, True, dtype=numpy.bool_)}
139             )
140 
141         # Projection processing
142         schema_columns_set = set(parquet_file.schema_arrow.names)
143         projection_names = {name for proj_col in projection for name in proj_col.all_names}
144         selected_columns = list(schema_columns_set.intersection(projection_names))
145 
146         # If no columns are selected, set to None
147         if not selected_columns:
148             selected_columns = None
149     else:
150         selected_columns = None
151 
152     # Read the parquet table with the optimized column list and selection filters
153     table = parquet.read_table(stream, columns=selected_columns, pre_buffer=False, filters=_select)
154     if selection:
155         table = filter_records(selection, table)
156     return table
157 
158 
159 def orc_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):
160     """
161     Read orc formatted files
162     """
163     import pyarrow.orc as orc
164 
165     stream = io.BytesIO(buffer)
166     orc_file = orc.ORCFile(stream)
167 
168     if just_schema:
169         orc_schema = orc_file.schema
170         return convert_arrow_schema_to_orso_schema(orc_schema)
171 
172     table = orc_file.read()
173     if selection:
174         table = filter_records(selection, table)
175     if projection:
176         table = post_read_projector(table, projection)
177     return table
178 
179 
180 def jsonl_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):
181     import pyarrow.json
182 
183     if isinstance(buffer, bytes):
184         stream = io.BytesIO(buffer)
185     else:
186         stream = buffer
187 
188     table = pyarrow.json.read_json(stream)
189     schema = table.schema
190     if just_schema:
191         return convert_arrow_schema_to_orso_schema(schema)
192 
193     if selection:
194         table = filter_records(selection, table)
195     if projection:
196         table = post_read_projector(table, projection)
197 
198     return table
199 
200 
201 def csv_decoder(
202     buffer, projection: List = None, selection=None, delimiter: str = ",", just_schema: bool = False
203 ):
204     import pyarrow.csv
205     from pyarrow.csv import ParseOptions
206 
207     stream = io.BytesIO(buffer)
208     parse_options = ParseOptions(delimiter=delimiter, newlines_in_values=True)
209     table = pyarrow.csv.read_csv(stream, parse_options=parse_options)
210     schema = table.schema
211     if just_schema:
212         return convert_arrow_schema_to_orso_schema(schema)
213 
214     if selection:
215         table = filter_records(selection, table)
216     if projection:
217         table = post_read_projector(table, projection)
218 
219     return table
220 
221 
222 def tsv_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):
223     return csv_decoder(
224         buffer=buffer,
225         projection=projection,
226         selection=selection,
227         delimiter="\t",
228         just_schema=just_schema,
229     )
230 
231 
232 def arrow_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):
233     import pyarrow.feather as pf
234 
235     stream = io.BytesIO(buffer)
236     table = pf.read_table(stream)
237     schema = table.schema
238     if just_schema:
239         return convert_arrow_schema_to_orso_schema(schema)
240 
241     if selection:
242         table = filter_records(selection, table)
243     if projection:
244         table = post_read_projector(table, projection)
245 
246     return table
247 
248 
249 def avro_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):
250     """
251     AVRO isn't well supported, it is converted between table types which is
252     really slow
253     """
254     try:
255         from avro.datafile import DataFileReader
256         from avro.io import DatumReader
257     except ImportError:  # pragma: no-cover
258         raise Exception("`avro` is missing, please install or include in your `requirements.txt`.")
259 
260     stream = io.BytesIO(buffer)
261     reader = DataFileReader(stream, DatumReader())
262 
263     table = pyarrow.Table.from_pylist(list(reader))
264     schema = table.schema
265     if just_schema:
266         return convert_arrow_schema_to_orso_schema(schema)
267 
268     if selection:
269         table = filter_records(selection, table)
270     if projection:
271         table = post_read_projector(table, projection)
272 
273     return table
274 
275 
276 # for types we know about, set up how we handle them
277 KNOWN_EXTENSIONS: Dict[str, Tuple[Callable, str]] = {
278     "avro": (avro_decoder, ExtentionType.DATA),
279     "complete": (do_nothing, ExtentionType.CONTROL),
280     "ignore": (do_nothing, ExtentionType.CONTROL),
281     "arrow": (arrow_decoder, ExtentionType.DATA),  # feather
282     "csv": (csv_decoder, ExtentionType.DATA),
283     "jsonl": (jsonl_decoder, ExtentionType.DATA),
284     "orc": (orc_decoder, ExtentionType.DATA),
285     "parquet": (parquet_decoder, ExtentionType.DATA),
286     "tsv": (tsv_decoder, ExtentionType.DATA),
287     "zstd": (zstd_decoder, ExtentionType.DATA),  # jsonl/zstd
288 }
289 
290 VALID_EXTENSIONS = set(f".{ext}" for ext in KNOWN_EXTENSIONS.keys())
291 TUPLE_OF_VALID_EXTENSIONS = tuple(VALID_EXTENSIONS)
292 DATA_EXTENSIONS = set(
293     f".{ext}" for ext, conf in KNOWN_EXTENSIONS.items() if conf[1] == ExtentionType.DATA
294 )
295
```

Path: `opteryx/__version__.py`
Content:
```
1 __build__ = 312
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 = 4
32 _status = VersionStatus.ALPHA
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__ = 312
+__build__ = 313
 
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
diff --git a/opteryx/utils/file_decoders.py b/opteryx/utils/file_decoders.py
--- a/opteryx/utils/file_decoders.py
+++ b/opteryx/utils/file_decoders.py
@@ -273,6 +273,32 @@
     return table
 
 
+def ipc_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):
+    from itertools import chain
+
+    from pyarrow import ipc
+
+    stream = io.BytesIO(buffer)
+    reader = ipc.open_stream(stream)
+
+    batch_one = next(reader, None)
+    if batch_one is None:
+        return None
+
+    schema = batch_one.schema
+
+    if just_schema:
+        return convert_arrow_schema_to_orso_schema(schema)
+
+    table = pyarrow.Table.from_batches([batch for batch in chain([batch_one], reader)])
+    if selection:
+        table = filter_records(selection, table)
+    if projection:
+        table = post_read_projector(table, projection)
+
+    return table
+
+
 # for types we know about, set up how we handle them
 KNOWN_EXTENSIONS: Dict[str, Tuple[Callable, str]] = {
     "avro": (avro_decoder, ExtentionType.DATA),
@@ -280,6 +306,7 @@
     "ignore": (do_nothing, ExtentionType.CONTROL),
     "arrow": (arrow_decoder, ExtentionType.DATA),  # feather
     "csv": (csv_decoder, ExtentionType.DATA),
+    "ipc": (ipc_decoder, ExtentionType.DATA),
     "jsonl": (jsonl_decoder, ExtentionType.DATA),
     "orc": (orc_decoder, ExtentionType.DATA),
     "parquet": (parquet_decoder, ExtentionType.DATA),

{"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__ = 312\n+__build__ = 313\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.\ndiff --git a/opteryx/utils/file_decoders.py b/opteryx/utils/file_decoders.py\n--- a/opteryx/utils/file_decoders.py\n+++ b/opteryx/utils/file_decoders.py\n@@ -273,6 +273,32 @@\n     return table\n \n \n+def ipc_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n+    from itertools import chain\n+\n+    from pyarrow import ipc\n+\n+    stream = io.BytesIO(buffer)\n+    reader = ipc.open_stream(stream)\n+\n+    batch_one = next(reader, None)\n+    if batch_one is None:\n+        return None\n+\n+    schema = batch_one.schema\n+\n+    if just_schema:\n+        return convert_arrow_schema_to_orso_schema(schema)\n+\n+    table = pyarrow.Table.from_batches([batch for batch in chain([batch_one], reader)])\n+    if selection:\n+        table = filter_records(selection, table)\n+    if projection:\n+        table = post_read_projector(table, projection)\n+\n+    return table\n+\n+\n # for types we know about, set up how we handle them\n KNOWN_EXTENSIONS: Dict[str, Tuple[Callable, str]] = {\n     \"avro\": (avro_decoder, ExtentionType.DATA),\n@@ -280,6 +306,7 @@\n     \"ignore\": (do_nothing, ExtentionType.CONTROL),\n     \"arrow\": (arrow_decoder, ExtentionType.DATA),  # feather\n     \"csv\": (csv_decoder, ExtentionType.DATA),\n+    \"ipc\": (ipc_decoder, ExtentionType.DATA),\n     \"jsonl\": (jsonl_decoder, ExtentionType.DATA),\n     \"orc\": (orc_decoder, ExtentionType.DATA),\n     \"parquet\": (parquet_decoder, ExtentionType.DATA),\n", "issue": "\u2728 support pyarrow's IPC format\n### Thanks for stopping by to let us know something could be better!\r\n\r\n**Is your feature request related to a problem? Please describe.** _A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]_\r\n\r\n**Describe the solution you'd like** _A clear and concise description of what you want to happen._\r\n\r\n**Describe alternatives you've considered** _A clear and concise description of any alternative solutions or features you've considered._\r\n\r\n**Additional context** _Add any other context or screenshots about the feature request here._\r\n\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\"\"\"\nDecode files from a raw binary format to a PyArrow Table.\n\"\"\"\nimport io\nfrom enum import Enum\nfrom typing import Callable\nfrom typing import Dict\nfrom typing import List\nfrom typing import Tuple\n\nimport numpy\nimport pyarrow\nfrom orso.tools import random_string\n\nfrom opteryx.exceptions import UnsupportedFileTypeError\nfrom opteryx.utils.arrow import post_read_projector\n\n\nclass ExtentionType(str, Enum):\n    \"\"\"labels for the file extentions\"\"\"\n\n    DATA = \"DATA\"\n    CONTROL = \"CONTROL\"\n\n\ndef convert_arrow_schema_to_orso_schema(arrow_schema):\n    from orso.schema import FlatColumn\n    from orso.schema import RelationSchema\n\n    return RelationSchema(\n        name=\"arrow\",\n        columns=[FlatColumn.from_arrow(field) for field in arrow_schema],\n    )\n\n\ndef get_decoder(dataset: str) -> Callable:\n    \"\"\"helper routine to get the decoder for a given file\"\"\"\n    ext = dataset.split(\".\")[-1].lower()\n    if ext not in KNOWN_EXTENSIONS:  # pragma: no cover\n        raise UnsupportedFileTypeError(f\"Unsupported file type - {ext}\")\n    file_decoder, file_type = KNOWN_EXTENSIONS[ext]\n    if file_type != ExtentionType.DATA:  # pragma: no cover\n        raise UnsupportedFileTypeError(f\"File is not a data file - {ext}\")\n    return file_decoder\n\n\ndef do_nothing(buffer, projection=None, just_schema: bool = False):  # pragma: no cover\n    \"\"\"for when you need to look like you're doing something\"\"\"\n    return False\n\n\ndef filter_records(filter, table):\n    \"\"\"\n    When we can't push predicates to the actual read, use this to filter records\n    just after the read.\n    \"\"\"\n    # notes:\n    #   at this point we've not renamed any columns, this may affect some filters\n    from opteryx.managers.expression import NodeType\n    from opteryx.managers.expression import evaluate\n    from opteryx.models import Node\n\n    if isinstance(filter, list) and filter:\n        filter_copy = [p for p in filter]\n        root = filter_copy.pop()\n\n        if root.left.node_type == NodeType.IDENTIFIER:\n            root.left.schema_column.identity = root.left.source_column\n        if root.right.node_type == NodeType.IDENTIFIER:\n            root.right.schema_column.identity = root.right.source_column\n\n        while filter_copy:\n            right = filter_copy.pop()\n            if right.left.node_type == NodeType.IDENTIFIER:\n                right.left.schema_column.identity = right.left.source_column\n            if right.right.node_type == NodeType.IDENTIFIER:\n                right.right.schema_column.identity = right.right.source_column\n            root = Node(\n                NodeType.AND, left=root, right=right, schema_column=Node(identity=random_string())\n            )\n    else:\n        root = filter\n\n    mask = evaluate(root, table)\n    return table.filter(mask)\n\n\ndef zstd_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    \"\"\"\n    Read zstandard compressed JSONL files\n    \"\"\"\n    import zstandard\n\n    stream = io.BytesIO(buffer)\n\n    with zstandard.open(stream, \"rb\") as file:\n        return jsonl_decoder(\n            file, projection=projection, selection=selection, just_schema=just_schema\n        )\n\n\ndef parquet_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    \"\"\"\n    Read parquet formatted files.\n    \"\"\"\n    from pyarrow import parquet\n\n    from opteryx.connectors.capabilities import PredicatePushable\n\n    # Convert the selection to DNF format if applicable\n    _select, selection = PredicatePushable.to_dnf(selection) if selection else (None, None)\n\n    stream = io.BytesIO(buffer)\n\n    if projection or just_schema:\n        # Open the parquet file only once\n        parquet_file = parquet.ParquetFile(stream)\n\n        # Return just the schema if that's all that's needed\n        if just_schema:\n            return convert_arrow_schema_to_orso_schema(parquet_file.schema_arrow)\n\n        # Special handling for projection of [\"count_*\"]\n        if projection == [\"count_*\"]:\n            return pyarrow.Table.from_pydict(\n                {\"_\": numpy.full(parquet_file.metadata.num_rows, True, dtype=numpy.bool_)}\n            )\n\n        # Projection processing\n        schema_columns_set = set(parquet_file.schema_arrow.names)\n        projection_names = {name for proj_col in projection for name in proj_col.all_names}\n        selected_columns = list(schema_columns_set.intersection(projection_names))\n\n        # If no columns are selected, set to None\n        if not selected_columns:\n            selected_columns = None\n    else:\n        selected_columns = None\n\n    # Read the parquet table with the optimized column list and selection filters\n    table = parquet.read_table(stream, columns=selected_columns, pre_buffer=False, filters=_select)\n    if selection:\n        table = filter_records(selection, table)\n    return table\n\n\ndef orc_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    \"\"\"\n    Read orc formatted files\n    \"\"\"\n    import pyarrow.orc as orc\n\n    stream = io.BytesIO(buffer)\n    orc_file = orc.ORCFile(stream)\n\n    if just_schema:\n        orc_schema = orc_file.schema\n        return convert_arrow_schema_to_orso_schema(orc_schema)\n\n    table = orc_file.read()\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n    return table\n\n\ndef jsonl_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    import pyarrow.json\n\n    if isinstance(buffer, bytes):\n        stream = io.BytesIO(buffer)\n    else:\n        stream = buffer\n\n    table = pyarrow.json.read_json(stream)\n    schema = table.schema\n    if just_schema:\n        return convert_arrow_schema_to_orso_schema(schema)\n\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n\n    return table\n\n\ndef csv_decoder(\n    buffer, projection: List = None, selection=None, delimiter: str = \",\", just_schema: bool = False\n):\n    import pyarrow.csv\n    from pyarrow.csv import ParseOptions\n\n    stream = io.BytesIO(buffer)\n    parse_options = ParseOptions(delimiter=delimiter, newlines_in_values=True)\n    table = pyarrow.csv.read_csv(stream, parse_options=parse_options)\n    schema = table.schema\n    if just_schema:\n        return convert_arrow_schema_to_orso_schema(schema)\n\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n\n    return table\n\n\ndef tsv_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    return csv_decoder(\n        buffer=buffer,\n        projection=projection,\n        selection=selection,\n        delimiter=\"\\t\",\n        just_schema=just_schema,\n    )\n\n\ndef arrow_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    import pyarrow.feather as pf\n\n    stream = io.BytesIO(buffer)\n    table = pf.read_table(stream)\n    schema = table.schema\n    if just_schema:\n        return convert_arrow_schema_to_orso_schema(schema)\n\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n\n    return table\n\n\ndef avro_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    \"\"\"\n    AVRO isn't well supported, it is converted between table types which is\n    really slow\n    \"\"\"\n    try:\n        from avro.datafile import DataFileReader\n        from avro.io import DatumReader\n    except ImportError:  # pragma: no-cover\n        raise Exception(\"`avro` is missing, please install or include in your `requirements.txt`.\")\n\n    stream = io.BytesIO(buffer)\n    reader = DataFileReader(stream, DatumReader())\n\n    table = pyarrow.Table.from_pylist(list(reader))\n    schema = table.schema\n    if just_schema:\n        return convert_arrow_schema_to_orso_schema(schema)\n\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n\n    return table\n\n\n# for types we know about, set up how we handle them\nKNOWN_EXTENSIONS: Dict[str, Tuple[Callable, str]] = {\n    \"avro\": (avro_decoder, ExtentionType.DATA),\n    \"complete\": (do_nothing, ExtentionType.CONTROL),\n    \"ignore\": (do_nothing, ExtentionType.CONTROL),\n    \"arrow\": (arrow_decoder, ExtentionType.DATA),  # feather\n    \"csv\": (csv_decoder, ExtentionType.DATA),\n    \"jsonl\": (jsonl_decoder, ExtentionType.DATA),\n    \"orc\": (orc_decoder, ExtentionType.DATA),\n    \"parquet\": (parquet_decoder, ExtentionType.DATA),\n    \"tsv\": (tsv_decoder, ExtentionType.DATA),\n    \"zstd\": (zstd_decoder, ExtentionType.DATA),  # jsonl/zstd\n}\n\nVALID_EXTENSIONS = set(f\".{ext}\" for ext in KNOWN_EXTENSIONS.keys())\nTUPLE_OF_VALID_EXTENSIONS = tuple(VALID_EXTENSIONS)\nDATA_EXTENSIONS = set(\n    f\".{ext}\" for ext, conf in KNOWN_EXTENSIONS.items() if conf[1] == ExtentionType.DATA\n)\n", "path": "opteryx/utils/file_decoders.py"}, {"content": "__build__ = 312\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 = 4\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"}], "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\"\"\"\nDecode files from a raw binary format to a PyArrow Table.\n\"\"\"\nimport io\nfrom enum import Enum\nfrom typing import Callable\nfrom typing import Dict\nfrom typing import List\nfrom typing import Tuple\n\nimport numpy\nimport pyarrow\nfrom orso.tools import random_string\n\nfrom opteryx.exceptions import UnsupportedFileTypeError\nfrom opteryx.utils.arrow import post_read_projector\n\n\nclass ExtentionType(str, Enum):\n    \"\"\"labels for the file extentions\"\"\"\n\n    DATA = \"DATA\"\n    CONTROL = \"CONTROL\"\n\n\ndef convert_arrow_schema_to_orso_schema(arrow_schema):\n    from orso.schema import FlatColumn\n    from orso.schema import RelationSchema\n\n    return RelationSchema(\n        name=\"arrow\",\n        columns=[FlatColumn.from_arrow(field) for field in arrow_schema],\n    )\n\n\ndef get_decoder(dataset: str) -> Callable:\n    \"\"\"helper routine to get the decoder for a given file\"\"\"\n    ext = dataset.split(\".\")[-1].lower()\n    if ext not in KNOWN_EXTENSIONS:  # pragma: no cover\n        raise UnsupportedFileTypeError(f\"Unsupported file type - {ext}\")\n    file_decoder, file_type = KNOWN_EXTENSIONS[ext]\n    if file_type != ExtentionType.DATA:  # pragma: no cover\n        raise UnsupportedFileTypeError(f\"File is not a data file - {ext}\")\n    return file_decoder\n\n\ndef do_nothing(buffer, projection=None, just_schema: bool = False):  # pragma: no cover\n    \"\"\"for when you need to look like you're doing something\"\"\"\n    return False\n\n\ndef filter_records(filter, table):\n    \"\"\"\n    When we can't push predicates to the actual read, use this to filter records\n    just after the read.\n    \"\"\"\n    # notes:\n    #   at this point we've not renamed any columns, this may affect some filters\n    from opteryx.managers.expression import NodeType\n    from opteryx.managers.expression import evaluate\n    from opteryx.models import Node\n\n    if isinstance(filter, list) and filter:\n        filter_copy = [p for p in filter]\n        root = filter_copy.pop()\n\n        if root.left.node_type == NodeType.IDENTIFIER:\n            root.left.schema_column.identity = root.left.source_column\n        if root.right.node_type == NodeType.IDENTIFIER:\n            root.right.schema_column.identity = root.right.source_column\n\n        while filter_copy:\n            right = filter_copy.pop()\n            if right.left.node_type == NodeType.IDENTIFIER:\n                right.left.schema_column.identity = right.left.source_column\n            if right.right.node_type == NodeType.IDENTIFIER:\n                right.right.schema_column.identity = right.right.source_column\n            root = Node(\n                NodeType.AND, left=root, right=right, schema_column=Node(identity=random_string())\n            )\n    else:\n        root = filter\n\n    mask = evaluate(root, table)\n    return table.filter(mask)\n\n\ndef zstd_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    \"\"\"\n    Read zstandard compressed JSONL files\n    \"\"\"\n    import zstandard\n\n    stream = io.BytesIO(buffer)\n\n    with zstandard.open(stream, \"rb\") as file:\n        return jsonl_decoder(\n            file, projection=projection, selection=selection, just_schema=just_schema\n        )\n\n\ndef parquet_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    \"\"\"\n    Read parquet formatted files.\n    \"\"\"\n    from pyarrow import parquet\n\n    from opteryx.connectors.capabilities import PredicatePushable\n\n    # Convert the selection to DNF format if applicable\n    _select, selection = PredicatePushable.to_dnf(selection) if selection else (None, None)\n\n    stream = io.BytesIO(buffer)\n\n    if projection or just_schema:\n        # Open the parquet file only once\n        parquet_file = parquet.ParquetFile(stream)\n\n        # Return just the schema if that's all that's needed\n        if just_schema:\n            return convert_arrow_schema_to_orso_schema(parquet_file.schema_arrow)\n\n        # Special handling for projection of [\"count_*\"]\n        if projection == [\"count_*\"]:\n            return pyarrow.Table.from_pydict(\n                {\"_\": numpy.full(parquet_file.metadata.num_rows, True, dtype=numpy.bool_)}\n            )\n\n        # Projection processing\n        schema_columns_set = set(parquet_file.schema_arrow.names)\n        projection_names = {name for proj_col in projection for name in proj_col.all_names}\n        selected_columns = list(schema_columns_set.intersection(projection_names))\n\n        # If no columns are selected, set to None\n        if not selected_columns:\n            selected_columns = None\n    else:\n        selected_columns = None\n\n    # Read the parquet table with the optimized column list and selection filters\n    table = parquet.read_table(stream, columns=selected_columns, pre_buffer=False, filters=_select)\n    if selection:\n        table = filter_records(selection, table)\n    return table\n\n\ndef orc_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    \"\"\"\n    Read orc formatted files\n    \"\"\"\n    import pyarrow.orc as orc\n\n    stream = io.BytesIO(buffer)\n    orc_file = orc.ORCFile(stream)\n\n    if just_schema:\n        orc_schema = orc_file.schema\n        return convert_arrow_schema_to_orso_schema(orc_schema)\n\n    table = orc_file.read()\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n    return table\n\n\ndef jsonl_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    import pyarrow.json\n\n    if isinstance(buffer, bytes):\n        stream = io.BytesIO(buffer)\n    else:\n        stream = buffer\n\n    table = pyarrow.json.read_json(stream)\n    schema = table.schema\n    if just_schema:\n        return convert_arrow_schema_to_orso_schema(schema)\n\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n\n    return table\n\n\ndef csv_decoder(\n    buffer, projection: List = None, selection=None, delimiter: str = \",\", just_schema: bool = False\n):\n    import pyarrow.csv\n    from pyarrow.csv import ParseOptions\n\n    stream = io.BytesIO(buffer)\n    parse_options = ParseOptions(delimiter=delimiter, newlines_in_values=True)\n    table = pyarrow.csv.read_csv(stream, parse_options=parse_options)\n    schema = table.schema\n    if just_schema:\n        return convert_arrow_schema_to_orso_schema(schema)\n\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n\n    return table\n\n\ndef tsv_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    return csv_decoder(\n        buffer=buffer,\n        projection=projection,\n        selection=selection,\n        delimiter=\"\\t\",\n        just_schema=just_schema,\n    )\n\n\ndef arrow_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    import pyarrow.feather as pf\n\n    stream = io.BytesIO(buffer)\n    table = pf.read_table(stream)\n    schema = table.schema\n    if just_schema:\n        return convert_arrow_schema_to_orso_schema(schema)\n\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n\n    return table\n\n\ndef avro_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    \"\"\"\n    AVRO isn't well supported, it is converted between table types which is\n    really slow\n    \"\"\"\n    try:\n        from avro.datafile import DataFileReader\n        from avro.io import DatumReader\n    except ImportError:  # pragma: no-cover\n        raise Exception(\"`avro` is missing, please install or include in your `requirements.txt`.\")\n\n    stream = io.BytesIO(buffer)\n    reader = DataFileReader(stream, DatumReader())\n\n    table = pyarrow.Table.from_pylist(list(reader))\n    schema = table.schema\n    if just_schema:\n        return convert_arrow_schema_to_orso_schema(schema)\n\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n\n    return table\n\n\ndef ipc_decoder(buffer, projection: List = None, selection=None, just_schema: bool = False):\n    from itertools import chain\n\n    from pyarrow import ipc\n\n    stream = io.BytesIO(buffer)\n    reader = ipc.open_stream(stream)\n\n    batch_one = next(reader, None)\n    if batch_one is None:\n        return None\n\n    schema = batch_one.schema\n\n    if just_schema:\n        return convert_arrow_schema_to_orso_schema(schema)\n\n    table = pyarrow.Table.from_batches([batch for batch in chain([batch_one], reader)])\n    if selection:\n        table = filter_records(selection, table)\n    if projection:\n        table = post_read_projector(table, projection)\n\n    return table\n\n\n# for types we know about, set up how we handle them\nKNOWN_EXTENSIONS: Dict[str, Tuple[Callable, str]] = {\n    \"avro\": (avro_decoder, ExtentionType.DATA),\n    \"complete\": (do_nothing, ExtentionType.CONTROL),\n    \"ignore\": (do_nothing, ExtentionType.CONTROL),\n    \"arrow\": (arrow_decoder, ExtentionType.DATA),  # feather\n    \"csv\": (csv_decoder, ExtentionType.DATA),\n    \"ipc\": (ipc_decoder, ExtentionType.DATA),\n    \"jsonl\": (jsonl_decoder, ExtentionType.DATA),\n    \"orc\": (orc_decoder, ExtentionType.DATA),\n    \"parquet\": (parquet_decoder, ExtentionType.DATA),\n    \"tsv\": (tsv_decoder, ExtentionType.DATA),\n    \"zstd\": (zstd_decoder, ExtentionType.DATA),  # jsonl/zstd\n}\n\nVALID_EXTENSIONS = set(f\".{ext}\" for ext in KNOWN_EXTENSIONS.keys())\nTUPLE_OF_VALID_EXTENSIONS = tuple(VALID_EXTENSIONS)\nDATA_EXTENSIONS = set(\n    f\".{ext}\" for ext, conf in KNOWN_EXTENSIONS.items() if conf[1] == ExtentionType.DATA\n)\n", "path": "opteryx/utils/file_decoders.py"}, {"content": "__build__ = 313\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 = 4\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"}]}