all draken vectors use SIMD to merge hashes

Author: joocer

examples/disk_reader_usage.py

@@ -21,7 +21,7 @@ def example_basic():
 # Example 2: Stream large files without cache pollution
 def example_streaming():
     """Process multiple large files efficiently."""
-    large_files = ["planets-gw0.duckdb", "planets-gw1.duckdb"]
+    large_files = ["tmp/planets-gw0.duckdb", "tmp/planets-gw1.duckdb"]
 
     for filename in large_files:
         # Read and evict from cache to save memory

opteryx/__version__.py

@@ -1,9 +1,9 @@
 # THIS FILE IS AUTOMATICALLY UPDATED DURING THE BUILD PROCESS
 # DO NOT EDIT THIS FILE DIRECTLY
 
-__build__ = 1824
+__build__ = 1829
 __author__ = "@joocer"
-__version__ = "0.26.2-beta.1824"
+__version__ = "0.26.2-beta.1829"
 
 # Store the version here so:
 # 1) we don't load dependencies by storing it in __init__.py

opteryx/draken/vectors/array_vector.pxd

@@ -1,6 +1,28 @@
+from libc.stdint cimport uint64_t
+
+from opteryx.draken.core.buffers cimport DrakenArrayBuffer
 from opteryx.draken.vectors.vector cimport Vector
 
+
 cdef class ArrayVector(Vector):
-    cdef object _arr  # Store the arrow array
+    cdef DrakenArrayBuffer* ptr
+    cdef object _child
+    cdef bint owns_offsets
+    cdef bint owns_null_bitmap
+    cdef object _arrow_parent
+    cdef object _arrow_offsets_buf
+    cdef object _arrow_null_buf
+    cdef object _arrow_child_array
+    cdef object _child_arrow_type
+    cdef bint _child_decode_utf8
+
+    cdef object _materialize_row(self, Py_ssize_t idx)
+    cdef void hash_into(
+        self,
+        uint64_t[::1] out_buf,
+        Py_ssize_t offset=*,
+        uint64_t mix_constant=*,
+    ) except *
+
 
 cdef ArrayVector from_arrow(object array)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "opteryx"
-version = "0.26.2-beta.1824"
+version = "0.26.2-beta.1829"
 description = "Query your data, where it lives"
 requires-python = '>=3.11'
 readme = {file = "README.md", content-type = "text/markdown"}

setup.py

@@ -691,9 +691,11 @@ def make_draken_extension(module_path, source_file, depends=None, language=None,
         list_ops_link_args.append("-pthread")
 
     for ext in extensions:
-        if ext.name == "opteryx.draken.vectors.int64_vector":
-            ext.sources.append("src/cpp/simd_hash.cpp")
-            break
+        if ext.name.startswith("opteryx.draken.vectors."):
+            if "src/cpp/simd_hash.cpp" not in ext.sources:
+                ext.sources.append("src/cpp/simd_hash.cpp")
+            ext.language = "c++"
+            ext.extra_compile_args = CPP_COMPILE_FLAGS
 
     extensions.append(
         Extension(

tests/__init__.py

@@ -45,6 +45,7 @@ def test_example():
         run_tests()
 """
 
+import datetime
 import os
 import platform
 from functools import wraps
@@ -615,12 +616,12 @@ def create_duck_db():  # pragma: no cover
 
     worker_id = os.environ.get('PYTEST_XDIST_WORKER', 'gw0')
 
-    if os.path.exists(f"planets-{worker_id}.duckdb"):
+    if os.path.exists(f"tmp/planets-{worker_id}.duckdb"):
         return
 
     import duckdb
 
-    conn = duckdb.connect(database=f"planets-{worker_id}.duckdb")
+    conn = duckdb.connect(database=f"tmp/planets-{worker_id}.duckdb")
     cur = conn.cursor()
     try:
         cur.execute(CREATE_DATABASE)
@@ -736,6 +737,48 @@ def cast_dataset(dataset):
     opteryx.register_store("iceberg", IcebergConnector, catalog=catalog)
 
     for dataset in ('planets', 'satellites', 'missions', 'astronauts'):
+        if dataset == 'planets':
+            from opteryx.virtual_datasets import planet_data
+
+            def snapshot_props(label, cutoff):
+                props = {"temporal_version": label}
+                if cutoff is not None:
+                    props["end_date"] = cutoff.isoformat()
+                else:
+                    props["end_date"] = "current"
+                return props
+
+            def load_planet_snapshot(cutoff):
+                table_data = planet_data.read() if cutoff is None else planet_data.read(end_date=cutoff)
+                return cast_dataset(table_data)
+
+            snapshots = [
+                ("pre_uranus", datetime.datetime(1781, 4, 25)),
+                ("pre_neptune", datetime.datetime(1846, 11, 12)),
+                ("pre_pluto", datetime.datetime(1930, 3, 12)),
+                ("modern", None),
+            ]
+
+            snapshot_label, cutoff = snapshots[0]
+            snapshot_data = load_planet_snapshot(cutoff)
+            table = catalog.create_table("iceberg.planets", schema=snapshot_data.schema)
+            table.append(snapshot_data, snapshot_properties=snapshot_props(snapshot_label, cutoff))
+
+            latest_snapshot = snapshot_data
+            for snapshot_label, cutoff in snapshots[1:]:
+                snapshot_data = load_planet_snapshot(cutoff)
+                table.overwrite(snapshot_data, snapshot_properties=snapshot_props(snapshot_label, cutoff))
+                latest_snapshot = snapshot_data
+
+            expected_rows = latest_snapshot.num_rows
+            del latest_snapshot
+            del snapshot_data
+
+            iceberged = opteryx.query("SELECT * FROM iceberg.planets")
+            assert iceberged.rowcount == expected_rows
+            del iceberged  # Free memory immediately
+            continue
+
         data = opteryx.query_to_arrow(f"SELECT * FROM ${dataset}")
         data = cast_dataset(data)
 
@@ -751,4 +794,4 @@ def cast_dataset(dataset):
         del iceberged  # Free memory immediately
 
 
-    return catalog
+    return catalog

tests/draken/vectors/test_jsonl_support.py

@@ -15,7 +15,7 @@
 import sys
 from pathlib import Path
 
-sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent))
 
 import pytest
 import pyarrow as pa

tests/draken/vectors/test_temporal_vectors.py

@@ -15,7 +15,7 @@
 import sys
 from pathlib import Path
 
-sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent))
 
 import pytest
 import pyarrow as pa
@@ -203,7 +203,7 @@ def test_array_nested_types():
     # List of strings
     string_list = pa.array([['a', 'b'], ['c'], None, ['d', 'e', 'f']], type=pa.list_(pa.string()))
     vec = Vector.from_arrow(string_list)
-    assert vec.to_pylist() == [['a', 'b'], ['c'], None, ['d', 'e', 'f']]
+    assert vec.to_pylist() == [[b'a', b'b'], [b'c'], None, [b'd', b'e', b'f']]
 
     # List of floats
     float_list = pa.array([[1.1, 2.2], [3.3], None, [4.4]], type=pa.list_(pa.float64()))

tests/fuzzing/test_sql_fuzzer_compare_engines.py

@@ -152,7 +152,7 @@ def test_sql_fuzzing_connector_comparisons(i):
     worker_id = os.environ.get('PYTEST_XDIST_WORKER', 'gw0')
 
     import duckdb
-    conn = duckdb.connect(database=f"planets-{worker_id}.duckdb")
+    conn = duckdb.connect(database=f"tmp/planets-{worker_id}.duckdb")
     # Use test iteration number as seed for reproducibility
     seed = i + hash(worker_id) % 1000
     random.seed(seed)

tests/fuzzing/test_sql_fuzzer_connectors.py

@@ -94,7 +94,7 @@ def set_up_connections():
     opteryx.register_store("iceberg", IcebergConnector, catalog=iceberg_catalog)
     #opteryx.register_store("cockroach", SqlConnector, remove_prefix=True, connection=COCKROACH_CONNECTION)
     #opteryx.register_store("datastax", CqlConnector, remove_prefix=True, cluster=cluster)
-    opteryx.register_store("duckdb", SqlConnector, remove_prefix=True, connection=f"duckdb:///planets-{worker_id}.duckdb")
+    opteryx.register_store("duckdb", SqlConnector, remove_prefix=True, connection=f"duckdb:///tmp/planets-{worker_id}.duckdb")
     #opteryx.register_store("mongo", MongoDbConnector, database=MONGO_DATABASE, connection=MONGO_CONNECTION, remove_prefix=True)
     opteryx.register_store("sqlite", SqlConnector, remove_prefix=True, connection="sqlite:///testdata/sqlite/database.db")