Merge Arrow into Main for Release

.github/workflows/pre-commit.yaml

@@ -0,0 +1,39 @@
+name: pre-commit
+
+on:
+  pull_request:
+    branches:
+      - master
+      - staging
+    types:
+      - opened
+      - reopened
+      - ready_for_review
+      - synchronize
+
+env:
+  SKIP: pytest-check
+
+jobs:
+  pre-commit:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        token: ${{ secrets.GITHUB_TOKEN }}
+        fetch-depth: 0 # get full git history
+    - uses: actions/setup-python@v3
+      with:
+        cache: 'pip'
+    - name: Install pre-commit
+      run: |
+        pip install pre-commit
+    - name: Get changed files
+      id: changed-files
+      uses: tj-actions/changed-files@v21
+      with:
+        token: ${{ secrets.GITHUB_TOKEN }}
+    - name: Run pre-commit
+      uses: pre-commit/action@v2.0.3
+      with:
+        extra_args: --files ${{ steps.changed-files.outputs.all_changed_files }}

.github/workflows/release.yaml

@@ -14,13 +14,13 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        python-version: [3.7, 3.8, 3.9]
+        python-version: [3.7, 3.8, 3.9, '3.10']
         os: [ubuntu-latest, macos-latest, windows-latest]
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v3
     - name: Set up Python ${{ matrix.python-version }} ${{ matrix.os }}
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v4
       with:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies
@@ -39,7 +39,7 @@ jobs:
     if: startsWith(github.ref, 'refs/tags/')
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Create Release
         id: create_release
         uses: actions/create-release@v1
@@ -59,9 +59,9 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v3
     - name: Set up Python
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v4
       with:
         python-version: '3.8'
     - name: Install dependencies

.gitignore

@@ -118,3 +118,13 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# arrow parquet files
+*.parquet
+
+.idea
+.idea/misc.xml
+.idea/vcs.xml
+.idea/inspectionProfiles/profiles_settings.xml
+.idea/inspectionProfiles/Project_Default.xml
+/.idea/

.pre-commit-config.yaml

@@ -0,0 +1,35 @@
+repos:
+- repo: https://github.com/pre-commit/pre-commit-hooks
+  rev: v4.4.0
+  hooks:
+    - id: check-yaml
+    - id: end-of-file-fixer
+    - id: trailing-whitespace
+      exclude: ^(setup.cfg|btrdb/grpcinterface)
+- repo: https://github.com/psf/black
+  rev: 23.3.0
+  hooks:
+    - id: black-jupyter
+      args: [--line-length=88]
+      exclude: btrdb/grpcinterface/.*\.py
+- repo: https://github.com/pycqa/isort
+  rev: 5.11.5
+  hooks:
+    - id: isort
+      name: isort (python)
+      args: [--profile=black, --line-length=88]
+      exclude: btrdb/grpcinterface/.*\.py
+- repo: https://github.com/PyCQA/flake8
+  rev: 6.0.0
+  hooks:
+  - id: flake8
+    args: [--config=setup.cfg]
+    exclude: ^(btrdb/grpcinterface|tests|setup.py|btrdb4|docs|benchmarks)
+- repo: local
+  hooks:
+    - id: pytest-check
+      name: pytest-check
+      entry: pytest
+      language: system
+      pass_filenames: false
+      always_run: true

MANIFEST.in

@@ -20,4 +20,4 @@ global-exclude *.py[co]
 global-exclude .ipynb_checkpoints
 global-exclude .DS_Store
 global-exclude .env
-global-exclude .coverage.*
+global-exclude .coverage.*

benchmarks/benchmark_stream_inserts.py

@@ -0,0 +1,80 @@
+from time import perf_counter
+from typing import Dict, List, Tuple, Union
+
+import pyarrow
+
+import btrdb
+
+
+def time_stream_insert(
+    stream: btrdb.stream.Stream,
+    data: List[Tuple[int, float]],
+    merge_policy: str = "never",
+) -> Dict[str, Union[int, float, str]]:
+    """Insert raw data into a single stream, where data is a List of tuples of int64 timestamps and float64 values.
+
+    Parameters
+    ----------
+    stream : btrdb.stream.Stream, required
+        The stream to insert data into.
+    data : List[Tuple[int, float]], required
+        The data to insert into stream.
+    merge_policy : str, optional, default = 'never'
+        How should the platform handle duplicated data?
+        Valid policies:
+        `never`: the default, no points are merged
+        `equal`: points are deduplicated if the time and value are equal
+        `retain`: if two points have the same timestamp, the old one is kept
+        `replace`: if two points have the same timestamp, the new one is kept
+    """
+    prev_ver = stream.version()
+    tic = perf_counter()
+    new_ver = stream.insert(data, merge=merge_policy)
+    toc = perf_counter()
+    run_time = toc - tic
+    n_points = len(data)
+    result = {
+        "uuid": stream.uuid,
+        "previous_version": prev_ver,
+        "new_version": new_ver,
+        "points_to_insert": n_points,
+        "total_time_seconds": run_time,
+        "merge_policy": merge_policy,
+    }
+    return result
+
+
+def time_stream_arrow_insert(
+    stream: btrdb.stream.Stream, data: pyarrow.Table, merge_policy: str = "never"
+) -> Dict[str, Union[int, float, str]]:
+    """Insert raw data into a single stream, where data is a pyarrow Table of timestamps and float values.
+
+    Parameters
+    ----------
+    stream : btrdb.stream.Stream, required
+        The stream to insert data into.
+    data : pyarrow.Table, required
+        The table of data to insert into stream.
+    merge_policy : str, optional, default = 'never'
+        How should the platform handle duplicated data?
+        Valid policies:
+        `never`: the default, no points are merged
+        `equal`: points are deduplicated if the time and value are equal
+        `retain`: if two points have the same timestamp, the old one is kept
+        `replace`: if two points have the same timestamp, the new one is kept
+    """
+    prev_ver = stream.version()
+    tic = perf_counter()
+    new_ver = stream.arrow_insert(data, merge=merge_policy)
+    toc = perf_counter()
+    run_time = toc - tic
+    n_points = data.num_rows
+    result = {
+        "uuid": stream.uuid,
+        "previous_version": prev_ver,
+        "new_version": new_ver,
+        "points_to_insert": n_points,
+        "total_time_seconds": run_time,
+        "merge_policy": merge_policy,
+    }
+    return result