Skip to content

Commit f957ba1

Browse files
committed
improve test suite performance
1 parent 2feabfa commit f957ba1

6 files changed

Lines changed: 48 additions & 32 deletions

File tree

.github/workflows/release.yaml

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -127,9 +127,6 @@ jobs:
127127
- name: Check out repository
128128
uses: actions/checkout@v4.1.7
129129

130-
- name: Install Python package dependencies
131-
run: python -m pip install --upgrade cython==3.1.2 wheel numpy setuptools_rust pyarrow
132-
133130
- name: Create source dist
134131
run: python setup.py sdist
135132

@@ -232,6 +229,4 @@ jobs:
232229
- run: mv -v dist-win-3.12/* dist/
233230

234231
- name: Publish distribution 📦 to PyPI
235-
uses: pypa/gh-action-pypi-publish@release/v1
236-
with:
237-
password: ${{ secrets.PYPI_API_TOKEN }}
232+
uses: pypa/gh-action-pypi-publish@release/v1

opteryx/__version__.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
__build__ = 1393
1+
__build__ = 1394
22

33
# Licensed under the Apache License, Version 2.0 (the "License");
44
# you may not use this file except in compliance with the License.

opteryx/operators/read_node.py

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -43,10 +43,10 @@ def struct_to_jsonb(table: pyarrow.Table) -> pyarrow.Table:
4343
# Check if the column is a STRUCT
4444
if pyarrow.types.is_struct(field.type):
4545
# Convert each row in the STRUCT column to a JSON string
46-
json_strings = [
47-
orjson.dumps(row.as_py()) if row.is_valid else None for row in table.column(i)
48-
]
49-
json_array = pyarrow.array(json_strings, type=pyarrow.binary())
46+
json_array = pyarrow.array(
47+
[None if row is None else orjson.dumps(row) for row in table.column(i).to_pylist()],
48+
type=pyarrow.binary(),
49+
)
5050

5151
# Drop the original STRUCT column
5252
table = table.drop_columns(field.name)
@@ -62,13 +62,13 @@ def struct_to_jsonb(table: pyarrow.Table) -> pyarrow.Table:
6262

6363
# Convert each list element
6464
converted_data = []
65-
for item in list_array:
65+
for item in list_array.to_pylist():
6666
if item is None:
6767
converted_data.append(None)
6868
else:
6969
# Each item is a list of structs
7070
converted_list = []
71-
for struct in item.as_py():
71+
for struct in item:
7272
if struct is None:
7373
converted_list.append(None)
7474
else:
@@ -98,28 +98,28 @@ def normalize_morsel(schema: RelationSchema, morsel: pyarrow.Table) -> pyarrow.T
9898
# rename columns for internal use
9999
target_column_names = []
100100
# columns in the data but not in the schema, droppable
101-
droppable_columns = []
101+
droppable_columns = set()
102102

103103
# Find which columns to drop and which columns we already have
104104
for i, column in enumerate(morsel.column_names):
105105
column_name = schema.find_column(column)
106106
if column_name is None:
107-
droppable_columns.append(i)
107+
droppable_columns.add(i)
108108
else:
109109
target_column_names.append(str(column_name))
110110

111111
# Remove from the end otherwise we'll remove the wrong columns after we've removed one
112-
droppable_columns.reverse()
113-
for droppable in droppable_columns:
114-
morsel = morsel.remove_column(droppable)
112+
if droppable_columns:
113+
keep_indices = [i for i in range(len(morsel.columns)) if i not in droppable_columns]
114+
morsel = morsel.select(keep_indices)
115115

116116
# remane columns to the internal names (identities)
117117
morsel = morsel.rename_columns(target_column_names)
118118

119119
# add columns we don't have, populate with nulls but try to get the correct type
120120
for column in schema.columns:
121121
if column.identity not in target_column_names:
122-
null_column = pyarrow.array([None] * morsel.num_rows, type=column.arrow_field.type)
122+
null_column = pyarrow.nulls(morsel.num_rows, type=column.arrow_field.type)
123123
field = pyarrow.field(name=column.identity, type=column.arrow_field.type)
124124
morsel = morsel.append_column(field, null_column)
125125

opteryx/planner/binder/binder_visitor.py

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
# Distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND.
55

66
import re
7+
from functools import lru_cache
78
from typing import List
89
from typing import Set
910
from typing import Tuple
@@ -216,6 +217,11 @@ def convert_using_to_on(
216217
return conditions[0]
217218

218219

220+
@lru_cache(maxsize=128)
221+
def node_type_to_method_name(node_type: str) -> str:
222+
return f"visit_{CAMEL_TO_SNAKE.sub('_', node_type).lower()}"
223+
224+
219225
class BinderVisitor:
220226
"""
221227
The BinderVisitor visits each node in the query plan and adds catalogue information
@@ -242,13 +248,13 @@ def visit_node(self, node: Node, context: BindingContext) -> Tuple[Node, Binding
242248
The node and context after binding.
243249
"""
244250
node_type = node.node_type.name # type:ignore
245-
visit_method_name = f"visit_{CAMEL_TO_SNAKE.sub('_', node_type).lower()}"
251+
visit_method_name = node_type_to_method_name(node_type)
246252
visit_method = getattr(self, visit_method_name, None)
247253
if visit_method is None:
248254
# DEBUG: print(f"BinderVisitor: No method found for {visit_method_name}")
249255
return node, context
250256

251-
return_node, return_context = visit_method(node.copy(), context.copy())
257+
return_node, return_context = visit_method(node, context)
252258

253259
# DEBUG: from opteryx.exceptions import InvalidInternalStateError
254260
# DEBUG:

opteryx/third_party/travers/graph.py

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -265,7 +265,12 @@ def outgoing_edges(self, source: str) -> List[Tuple]:
265265
Returns:
266266
Set of Tuples (Source, Target, Relationship)
267267
"""
268-
return [(source, t, r) for t, r in self._edges.get(source, tuple())]
268+
try:
269+
edges = self._edges[source]
270+
except KeyError:
271+
return []
272+
273+
return [(source, target, relation) for target, relation in edges]
269274

270275
def ingoing_edges(self, target) -> List[Tuple]:
271276
"""

opteryx/utils/file_decoders.py

Lines changed: 20 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -87,10 +87,10 @@ def convert_arrow_schema_to_orso_schema(
8787

8888
def get_decoder(dataset: str) -> Callable:
8989
"""helper routine to get the decoder for a given file"""
90-
ext = dataset.split(".")[-1].lower()
91-
if ext not in KNOWN_EXTENSIONS: # pragma: no cover
92-
raise UnsupportedFileTypeError(f"Unsupported file type - {ext}")
93-
file_decoder, file_type = KNOWN_EXTENSIONS[ext]
90+
ext = dataset.rpartition(".")[2].lower()
91+
file_decoder, file_type = KNOWN_EXTENSIONS.get(ext, (None, None))
92+
if file_type is None:
93+
raise UnsupportedFileTypeError(f"Unsupported file type: {ext}")
9494
if file_type != ExtentionType.DATA: # pragma: no cover
9595
return do_nothing
9696
return file_decoder
@@ -387,12 +387,12 @@ def jsonl_decoder(
387387
from opteryx.third_party.tktech import csimdjson as simdjson
388388

389389
if isinstance(buffer, memoryview):
390-
buffer = MemoryViewStream(buffer)
390+
# If it's a memoryview, we need to convert it to bytes
391+
buffer = buffer.tobytes()
391392
if not isinstance(buffer, bytes):
392393
buffer = buffer.read()
393394

394395
parser = simdjson.Parser()
395-
lines = buffer.split(b"\n")
396396

397397
# preallocate and reuse dicts
398398
rows = []
@@ -402,17 +402,27 @@ def jsonl_decoder(
402402
# If projection is specified, we only need to ensure we keep the projected keys
403403
keys_union = {c.value for c in projection}
404404

405-
for line in lines:
405+
start = 0
406+
end = len(buffer)
407+
408+
while start < end:
409+
newline = buffer.find(b"\n", start)
410+
if newline == -1:
411+
newline = end
412+
line = buffer[start:newline]
413+
start = newline + 1
414+
406415
if not line:
407416
continue
417+
408418
record = parser.parse(line)
409419

420+
# convert nested objects to string
421+
row = record.as_dict()
410422
# keep track of all keys for schema padding
411423
if not projection:
412-
keys_union.update(record.keys())
424+
keys_union.update(row.keys())
413425

414-
# convert nested objects to string
415-
row = record.as_dict()
416426
for key in keys_union:
417427
if isinstance(row.get(key), dict):
418428
row[key] = record[key].mini

0 commit comments

Comments
 (0)