Repository: man-group/ArcticDB (language: python)

Issue:
Crash during resampling
### Describe the bug

2.1 million rows of fx intraday market data. Resample to 10s works. Resample to 1s crashes.

Orginal reporter: Tony Roberts from pyxll.

Possible OOM?

### Steps/Code to Reproduce

Download the Dec 2023 csv from here https://www.histdata.com/download-free-forex-historical-data/?/ascii/tick-data-quotes/eurusd/2023

(this is free sample data)

pandas data prep code:

```
file = "data/DAT_ASCII_EURUSD_T_202312.csv"
df2_raw = pd.read_csv(file, header=None)
df2 = df2_raw.drop(columns=3).rename(columns={0:'timestamp', 1:'bid', 2:'ask'})
df2['timestamp'] = pd.to_datetime(df2['timestamp'], format="%Y%m%d %H%M%S%f")
df2['mid'] = 0.5*(df2['bid'] + df2['ask'])
df2 = df2.set_index('timestamp')

lib.write("EURUSD", df2)
```

should produce a df with datetime index and bid, ask, mid columns with 2,102,540 rows and no missing data.

resample code:

```
def resampled_tick_data2(lib, symbol, start, end, freq, max_rows=1000):
    qb = adb.QueryBuilder()
    qb = qb.resample(freq, closed='right').agg({
        'high': ('mid', 'max'),
        'low': ('mid', 'min'),
        'open': ('mid', 'first'),
        'close': ('mid', 'last')
    })
    data = lib.read(symbol,
                    date_range=[start, end],
                    query_builder=qb)
    df = data.data.dropna()
    if max_rows is not None and len(df) > max_rows:
        raise RuntimeError("Number of rows is greater than max rows")
    return df

df = resampled_tick_data2(lib,
                          "EURUSD",
                          dt.datetime(2023,1,1),
                          dt.datetime(2023,12,31),
                          "1s",
                          max_rows=None)
```

### Expected Results

Either produce correct results or give a clear error message.

### OS, Python Version and ArcticDB Version

Py 3.10. arcticdb 4.5.0rc. WSL on Win11.

### Backend storage used

LMDB

### Additional Context

The same example runs ok with mimalloc, with freq='1s' and all freqs down to '10ms' which gives the same number of output rows as original data items.


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index 790789504f..4b079c0df5 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -784,7 +784,13 @@ std::shared_ptr<Column> ResampleClause<closed_boundary>::generate_output_index_c
     for (const auto& input_index_column: input_index_columns) {
         auto index_column_data = input_index_column->data();
         const auto cend = index_column_data.cend<IndexTDT>();
-        for (auto it = index_column_data.cbegin<IndexTDT>(); it != cend; ++it) {
+        auto it = index_column_data.cbegin<IndexTDT>();
+        // In case the passed date_range does not span the whole segment we need to skip the index values
+        // which are before the date range start.
+        while (it != cend && *it < date_range_->first) {
+            ++it;
+        }
+        for (;it != cend && *it <= date_range_->second; ++it) {
             if (ARCTICDB_LIKELY(current_bucket.contains(*it))) {
                 if (ARCTICDB_UNLIKELY(!current_bucket_added_to_index)) {
                     *output_index_column_it++ = label_boundary_ == ResampleBoundary::LEFT ? *std::prev(bucket_end_it) : *bucket_end_it;
diff --git a/cpp/arcticdb/processing/sorted_aggregation.cpp b/cpp/arcticdb/processing/sorted_aggregation.cpp
index 3cf4cb5662..57b6a84162 100644
--- a/cpp/arcticdb/processing/sorted_aggregation.cpp
+++ b/cpp/arcticdb/processing/sorted_aggregation.cpp
@@ -52,6 +52,7 @@ Column SortedAggregator<aggregation_operator, closed_boundary>::aggregate(const
                             input_agg_column->column_->type().data_type(),
                             [this,
                             &output_it,
+                            &output_end_it,
                             &bucket_aggregator,
                             &agg_column = *input_agg_column,
                             &input_index_column = input_index_columns.at(idx),
@@ -74,11 +75,15 @@ Column SortedAggregator<aggregation_operator, closed_boundary>::aggregate(const
                                     const auto index_cend = index_data.template cend<IndexTDT>();
                                     auto agg_data = agg_column.column_->data();
                                     auto agg_it = agg_data.template cbegin<typename input_type_info::TDT>();
+                                    bool bucket_has_values = false;
                                     for (auto index_it = index_data.template cbegin<IndexTDT>(); index_it != index_cend && !reached_end_of_buckets; ++index_it, ++agg_it) {
                                         if (ARCTICDB_LIKELY(current_bucket.contains(*index_it))) {
                                             push_to_aggregator<input_type_info::data_type>(bucket_aggregator, *agg_it, agg_column);
-                                        } else if (ARCTICDB_LIKELY(index_value_past_end_of_bucket(*index_it, *bucket_end_it))) {
-                                            *output_it++ = finalize_aggregator<output_type_info::data_type>(bucket_aggregator, string_pool);
+                                            bucket_has_values = true;
+                                        } else if (ARCTICDB_LIKELY(index_value_past_end_of_bucket(*index_it, *bucket_end_it)) && output_it != output_end_it) {
+                                            if (bucket_has_values) {
+                                                *output_it++ = finalize_aggregator<output_type_info::data_type>(bucket_aggregator, string_pool);
+                                            }
                                             // The following code is equivalent to:
                                             // if constexpr (closed_boundary == ResampleBoundary::LEFT) {
                                             //     bucket_end_it = std::upper_bound(bucket_end_it, bucket_boundaries_end, *index_it);
@@ -102,9 +107,11 @@ Column SortedAggregator<aggregation_operator, closed_boundary>::aggregate(const
                                                 }
                                             }
                                             if (ARCTICDB_LIKELY(!reached_end_of_buckets)) {
+                                                bucket_has_values = false;
                                                 current_bucket.set_boundaries(*bucket_start_it, *bucket_end_it);
                                                 if (ARCTICDB_LIKELY(current_bucket.contains(*index_it))) {
                                                     push_to_aggregator<input_type_info::data_type>(bucket_aggregator, *agg_it, agg_column);
+                                                    bucket_has_values = true;
                                                 }
                                             }
                                         }
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index 1f0497b869..05e7b7656e 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -596,7 +596,7 @@ void set_output_descriptors(
 
 std::shared_ptr<std::unordered_set<std::string>> columns_to_decode(const std::shared_ptr<PipelineContext>& pipeline_context) {
     std::shared_ptr<std::unordered_set<std::string>> res;
-    ARCTICDB_DEBUG(log::version(), "Creating columns list with {} bits set", pipeline_context->overall_column_bitset_->count());
+    ARCTICDB_DEBUG(log::version(), "Creating columns list with {} bits set", pipeline_context->overall_column_bitset_ ? pipeline_context->overall_column_bitset_->count() : -1);
     if(pipeline_context->overall_column_bitset_) {
         res = std::make_shared<std::unordered_set<std::string>>();
         auto en = pipeline_context->overall_column_bitset_->first();


Test patch:
diff --git a/python/tests/unit/arcticdb/version_store/test_resample.py b/python/tests/unit/arcticdb/version_store/test_resample.py
index ea49685215..2b12efc500 100644
--- a/python/tests/unit/arcticdb/version_store/test_resample.py
+++ b/python/tests/unit/arcticdb/version_store/test_resample.py
@@ -9,12 +9,34 @@
 import numpy as np
 import pandas as pd
 import pytest
+import datetime as dt
 
 from arcticdb import QueryBuilder
 from arcticdb.exceptions import ArcticDbNotYetImplemented, SchemaException
 from arcticdb.util.test import assert_frame_equal
 from arcticdb.util._versions import IS_PANDAS_TWO
 
+# Pandas recommended way to resample and exclude buckets with no index values, which is our behaviour
+# See https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.html#sparse-resampling
+def round(t, freq):
+    freq = pd.tseries.frequencies.to_offset(freq)
+    td = pd.Timedelta(freq)
+    return pd.Timestamp((t.value // td.value) * td.value)
+
+def generic_resample_test_with_empty_buckets(lib, sym, rule, aggregations, date_range=None):
+    # Pandas doesn't have a good date_range equivalent in resample, so just use read for that
+    expected = lib.read(sym, date_range=date_range).data
+    # Pandas 1.X needs None as the first argument to agg with named aggregators
+    expected = expected.groupby(partial(round, freq=rule)).agg(None, **aggregations)
+    expected = expected.reindex(columns=sorted(expected.columns))
+    
+    q = QueryBuilder()
+    q = q.resample(rule).agg(aggregations)
+    received = lib.read(sym, date_range=date_range, query_builder=q).data
+    received = received.reindex(columns=sorted(received.columns))
+
+    assert_frame_equal(expected, received, check_dtype=False)
+
 def generic_resample_test(lib, sym, rule, aggregations, date_range=None, closed=None, label=None):
     # Pandas doesn't have a good date_range equivalent in resample, so just use read for that
     expected = lib.read(sym, date_range=date_range).data
@@ -90,6 +112,75 @@ def test_resampling_duplicated_index_value_on_segment_boundary(lmdb_version_stor
     )
 
 
+class TestResamplingBucketInsideSegment:
+
+    def test_all_buckets_have_values(self, lmdb_version_store_v1):
+        lib = lmdb_version_store_v1
+        sym = "test_inner_buckets_are_empty"
+        start = dt.datetime(2023, 12, 7, 23, 59, 47, 500000);
+        idx = [start + i * pd.Timedelta('1s') for i in range(0, 8)]
+        df = pd.DataFrame({"mid": [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0]}, index=idx)
+        lib.write(sym, df)
+    
+        date_range = (dt.datetime(2023, 12, 7, 23, 59, 48), dt.datetime(2023, 12, 7, 23, 59, 52))
+        generic_resample_test_with_empty_buckets(lib, sym, 's', {'high': ('mid', 'max')}, date_range=date_range)
+
+    @pytest.mark.parametrize("closed", ("left", "right"))
+    def test_first_bucket_is_empy(self, lmdb_version_store_v1, closed):
+        lib = lmdb_version_store_v1
+        sym = "test_first_bucket_is_empy"
+        idx = pd.DatetimeIndex([
+            dt.datetime(2023, 12, 7, 23, 59, 48, 342000),
+            dt.datetime(2023, 12, 7, 23, 59, 49, 717000),
+            dt.datetime(2023, 12, 7, 23, 59, 49, 921000),
+            dt.datetime(2023, 12, 7, 23, 59, 50, 75000),
+            dt.datetime(2023, 12, 7, 23, 59, 50, 76000),
+            dt.datetime(2023, 12, 7, 23, 59, 55, 75000)
+        ])
+        df = pd.DataFrame({"mid": [1, 2, 3, 4, 5, 6]}, index=idx)
+        lib.write(sym, df)
+    
+        date_range = (dt.datetime(2023, 12, 7, 23, 59, 49), dt.datetime(2023, 12, 7, 23, 59, 50))
+        generic_resample_test(lib, sym, 's', {'high': ('mid', 'max')}, date_range=date_range, closed=closed)
+
+    @pytest.mark.parametrize("closed", ("left", "right"))
+    def test_last_bucket_is_empty(self, lmdb_version_store_v1, closed):
+        lib = lmdb_version_store_v1
+        sym = "test_last_bucket_is_empty"
+        idx = pd.DatetimeIndex([
+            dt.datetime(2023, 12, 7, 23, 59, 47, 342000),
+            dt.datetime(2023, 12, 7, 23, 59, 48, 342000),
+            dt.datetime(2023, 12, 7, 23, 59, 49, 717000),
+            dt.datetime(2023, 12, 7, 23, 59, 49, 921000),
+            dt.datetime(2023, 12, 7, 23, 59, 50, 75000),
+            dt.datetime(2023, 12, 7, 23, 59, 50, 76000),
+            dt.datetime(2023, 12, 7, 23, 59, 55, 75000)
+        ])
+        df = pd.DataFrame({"mid": [1, 2, 3, 4, 5, 6, 7]}, index=idx)
+        lib.write(sym, df)
+    
+        date_range = (dt.datetime(2023, 12, 7, 23, 59, 48), dt.datetime(2023, 12, 7, 23, 59, 49, 500000))
+        generic_resample_test(lib, sym, 's', {'high': ('mid', 'max')}, date_range=date_range, closed=closed)
+    
+    def test_inner_buckets_are_empty(self, lmdb_version_store_v1):
+        lib = lmdb_version_store_v1
+        sym = "test_inner_buckets_are_empty"
+        idx = pd.DatetimeIndex([
+            dt.datetime(2023, 12, 7, 23, 59, 48, 342000),
+            dt.datetime(2023, 12, 7, 23, 59, 49, 717000),
+            dt.datetime(2023, 12, 7, 23, 59, 49, 921000),
+            dt.datetime(2023, 12, 7, 23, 59, 52, 75000),
+            dt.datetime(2023, 12, 7, 23, 59, 53, 76000),
+            dt.datetime(2023, 12, 7, 23, 59, 55, 75000)
+        ])
+        df = pd.DataFrame({"mid": [1.0, 2.0, 3.0, 4.0, 5.0, 6.0]}, index=idx)
+        lib.write(sym, df)
+    
+        date_range = (dt.datetime(2023, 12, 7, 23, 59, 48), dt.datetime(2023, 12, 7, 23, 59, 55))
+        generic_resample_test_with_empty_buckets(lib, sym, 's', {'high': ('mid', 'max')}, date_range=date_range)
+        
+
+
 def test_resampling_timezones(lmdb_version_store_v1):
     lib = lmdb_version_store_v1
     sym = "test_resampling_timezones"
@@ -258,42 +349,20 @@ def test_resampling_empty_bucket_in_range(lmdb_version_store_v1):
     )
     lib.write(sym, df)
 
-    # Pandas recommended way to resample and exclude buckets with no index values, which is our behaviour
-    # See https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.html#sparse-resampling
-    def round(t, freq):
-        freq = pd.tseries.frequencies.to_offset(freq)
-        td = pd.Timedelta(freq)
-        return pd.Timestamp((t.value // td.value) * td.value)
-
-    expected = df.groupby(partial(round, freq="us")).agg(
-        {
-            "to_sum": "sum",
-            "to_mean": "mean",
-            "to_min": "min",
-            "to_max": "max",
-            "to_first": "first",
-            "to_last": "last",
-            "to_count": "count",
-        }
-    )
-    expected = expected.reindex(columns=sorted(expected.columns))
-    # expected["to_count"] = expected["to_count"].astype(np.uint64)
-
-    q = QueryBuilder()
-    q = q.resample("us").agg(
+    generic_resample_test_with_empty_buckets(
+        lib,
+        sym,
+        "us",
         {
-            "to_sum": "sum",
-            "to_mean": "mean",
-            "to_min": "min",
-            "to_max": "max",
-            "to_first": "first",
-            "to_last": "last",
-            "to_count": "count",
+            "to_sum": ("to_sum", "sum"),
+            "to_mean": ("to_mean", "mean"),
+            "to_min": ("to_min", "min"),
+            "to_max": ("to_max", "max"),
+            "to_first": ("to_first", "first"),
+            "to_last": ("to_last", "last"),
+            "to_count": ("to_count", "count"),
         }
     )
-    received = lib.read(sym, query_builder=q).data
-    received = received.reindex(columns=sorted(received.columns))
-    assert_frame_equal(expected, received, check_dtype=False)
 
 
 @pytest.mark.parametrize("use_date_range", (True, False))