Fix range intersection calculation in query.cpp (#2632)

#### Reference Issues/PRs
Monday refs:
- 9943908001
- 10031061308

#### What does this implement or fix?
- Fixes reading of indexes with old format (Monday ref: 9943908001)
  - Main fix: cpp/arcticdb/pipeline/query.cpp
- tests: cpp/arcticdb/pipeline/test/test_query.cpp and
python/tests/integration/arcticdb/version_store/test_num_storage_operations.py
- Fixes a bug where delete_range was not updating the index correctly
(Monday ref: 10031061308)
  - Main fix: cpp/arcticdb/version/version_core.cpp
- Tests:
https://github.com/man-group/ArcticDB/pull/2632/files#diff-7a61fb14e9865723e613017a346654e7c2694ffdb5eb13617e1665d1ad1ccba5R784
- Also tries to fix timeouts in macos builds

#### Any other comments?

#### Checklist

<details>
  <summary>
   Checklist for code changes...
  </summary>
 
- [ ] Have you updated the relevant docstrings, documentation and
copyright notice?
- [ ] Is this contribution tested against [all ArcticDB's
features](../docs/mkdocs/docs/technical/contributing.md)?
- [ ] Do all exceptions introduced raise appropriate [error
messages](https://docs.arcticdb.io/error_messages/)?
 - [ ] Are API changes highlighted in the PR description?
- [ ] Is the PR labelled as enhancement or bug so it appears in
autogenerated release notes?
</details>

<!--
Thanks for contributing a Pull Request to ArcticDB! Please ensure you
have taken a look at:
- ArcticDB's Code of Conduct:
https://github.com/man-group/ArcticDB/blob/master/CODE_OF_CONDUCT.md
- ArcticDB's Contribution Licensing:
https://github.com/man-group/ArcticDB/blob/master/docs/mkdocs/docs/technical/contributing.md#contribution-licensing
-->

Author: G-D-Petrov

.github/workflows/build_with_conda.yml

@@ -172,7 +172,7 @@ jobs:
             eval "$command"
           else
             cd python
-            python -m pytest --timeout=3600 -n logical --dist worksteal -v tests $ARCTICDB_PYTEST_ARGS
+            python -m pytest --timeout=3600 -v -n logical --dist worksteal tests $ARCTICDB_PYTEST_ARGS
           fi
         env:
           ARCTICDB_USING_CONDA: 1
@@ -281,7 +281,7 @@ jobs:
             eval "$command"
           else
             cd python
-            python -m pytest --timeout=3600 -n logical --dist worksteal tests $ARCTICDB_PYTEST_ARGS
+            python -m pytest --timeout=3600 -v -n logical --dist worksteal tests $ARCTICDB_PYTEST_ARGS
           fi
         env:
           ARCTICDB_USING_CONDA: 1

cpp/arcticdb/pipeline/index_utils.cpp

@@ -25,6 +25,7 @@ folly::Future<entity::AtomKey> write_index(
     for (const auto& slice_and_key : slice_and_keys) {
         writer.add(slice_and_key.key(), slice_and_key.slice_);
     }
+
     return writer.commit();
 }
 

cpp/arcticdb/pipeline/index_writer.hpp

@@ -97,12 +97,18 @@ class IndexWriter {
         );
 
         bool new_col_group = !current_col_.has_value() || *current_col_ < slice.col_range.first;
+        bool missing_row_val = !current_row_.has_value();
+        bool is_valid_col = *current_col_ == slice.col_range.first;
+        bool is_valid_row = *current_row_ < slice.row_range.first;
+        bool is_valid = (is_valid_col && is_valid_row);
         util::check_arg(
-                !current_row_.has_value() || new_col_group ||
-                        (*current_col_ == slice.col_range.first && *current_row_ < slice.row_range.first),
-                "expected increasing row group, last col range left value {}, arg {}",
+                missing_row_val || new_col_group || is_valid,
+                "expected increasing row group, last col range left value {}, col arg {}, row left value {}, row arg "
+                "{}",
                 current_col_.value_or(-1),
-                slice.col_range
+                slice.col_range,
+                current_row_.value_or(-1),
+                slice.row_range
         );
 
         add_unchecked(key, slice);

cpp/arcticdb/pipeline/query.cpp

@@ -32,7 +32,7 @@ template<typename ContainerType, typename IdxType>
 std::unique_ptr<util::BitSet> build_bitset_for_index(
         const ContainerType& container,
         IndexRange rg, // IndexRange is expected to be inclusive on both ends
-        bool dynamic_schema, bool column_groups, std::unique_ptr<util::BitSet>&& input
+        bool dynamic_schema, bool column_groups, bool is_read_operation, std::unique_ptr<util::BitSet>&& input
 ) {
     auto res = std::make_unique<util::BitSet>(static_cast<util::BitSetSizeType>(container.size()));
     if (container.empty())
@@ -53,7 +53,11 @@ std::unique_ptr<util::BitSet> build_bitset_for_index(
 
         // End index column is exclusive. We want to find the last position where `range_start` is < end_index at
         // position. This is equivalent to finding the first position where range_start + 1 >= end_index at position.
-        auto start_pos = std::lower_bound(end_index_col_begin, end_index_col_end, range_start + 1);
+        // If we are reading, we want to include the start index, in order to support backwards compatibility with older
+        // versions. The same fix should be done for updates, but that is not implemented yet and should be added with
+        // https://github.com/man-group/ArcticDB/issues/2655
+        const auto adjusted_range_start = is_read_operation ? range_start : range_start + 1;
+        auto start_pos = std::lower_bound(end_index_col_begin, end_index_col_end, adjusted_range_start);
 
         if (start_pos == end_idx_col.template end<IndexTagType>()) {
             ARCTICDB_DEBUG(log::version(), "Returning as start pos is at end");
@@ -92,7 +96,12 @@ std::unique_ptr<util::BitSet> build_bitset_for_index(
         const auto range_start = std::get<timestamp>(rg.start_);
         const auto range_end = std::get<timestamp>(rg.end_);
         for (auto i = 0u; i < container.size(); ++i) {
-            const auto intersects = range_intersects<RawType>(range_start, range_end, *start_idx_pos, *end_idx_pos - 1);
+            // If we are reading, we want to include the the end index, in order to support backwards compatibility with
+            // older versions. The same fix should be done for updates, but that is not implemented yet and should be
+            // added with https://github.com/man-group/ArcticDB/issues/2655
+            const auto adjusted_end_idx_pos = is_read_operation ? *end_idx_pos : *end_idx_pos - 1;
+            const auto intersects =
+                    range_intersects<RawType>(range_start, range_end, *start_idx_pos, adjusted_end_idx_pos);
             (*res)[i] = intersects;
             if (intersects)
                 ARCTICDB_DEBUG(log::version(), "range intersects at {}", i);
@@ -113,10 +122,11 @@ std::unique_ptr<util::BitSet> build_bitset_for_index(
 
 template std::unique_ptr<util::BitSet> build_bitset_for_index<
         IndexSegmentReader,
-        TimeseriesIndex>(const index::IndexSegmentReader&, IndexRange, bool, bool, std::unique_ptr<util::BitSet>&&);
+        TimeseriesIndex>(const index::IndexSegmentReader&, IndexRange, bool, bool, bool, std::unique_ptr<util::BitSet>&&);
 template std::unique_ptr<util::BitSet> build_bitset_for_index<
         IndexSegmentReader,
-        TableIndex>(const index::IndexSegmentReader&, IndexRange, bool, bool, std::unique_ptr<util::BitSet>&&);
+        TableIndex>(const index::IndexSegmentReader&, IndexRange, bool, bool, bool, std::unique_ptr<util::BitSet>&&);
 template std::unique_ptr<util::BitSet> build_bitset_for_index<
-        TestContainer, TimeseriesIndex>(const TestContainer&, IndexRange, bool, bool, std::unique_ptr<util::BitSet>&&);
+        TestContainer,
+        TimeseriesIndex>(const TestContainer&, IndexRange, bool, bool, bool, std::unique_ptr<util::BitSet>&&);
 } // namespace arcticdb::pipelines

cpp/arcticdb/pipeline/query.hpp

@@ -209,29 +209,29 @@ bool range_intersects(RawType a_start, RawType a_end, RawType b_start, RawType b
 
 template<typename ContainerType, typename IdxType>
 std::unique_ptr<util::BitSet> build_bitset_for_index(
-        const ContainerType& container, IndexRange rg, bool dynamic_schema, bool column_groups,
+        const ContainerType& container, IndexRange rg, bool dynamic_schema, bool column_groups, bool is_read_operation,
         std::unique_ptr<util::BitSet>&& input
 );
 
 template<typename ContainerType>
 inline FilterQuery<ContainerType> create_index_filter(
-        const IndexRange& range, bool dynamic_schema, bool column_groups
+        const IndexRange& range, bool dynamic_schema, bool column_groups, bool is_read_operation
 ) {
     static_assert(std::is_same_v<ContainerType, index::IndexSegmentReader>);
-    return [rg = range,
-            dynamic_schema,
-            column_groups](const ContainerType& container, std::unique_ptr<util::BitSet>&& input) mutable {
+    return [rg = range, dynamic_schema, column_groups, is_read_operation](
+                   const ContainerType& container, std::unique_ptr<util::BitSet>&& input
+           ) mutable {
         auto maybe_index_type = container.seg().template scalar_at<uint8_t>(0u, int(index::Fields::index_type));
         const auto index_type = IndexDescriptor::Type(maybe_index_type.value());
         switch (index_type) {
         case IndexDescriptorImpl::Type::TIMESTAMP: {
             return build_bitset_for_index<ContainerType, stream::TimeseriesIndex>(
-                    container, rg, dynamic_schema, column_groups, std::move(input)
+                    container, rg, dynamic_schema, column_groups, is_read_operation, std::move(input)
             );
         }
         case IndexDescriptorImpl::Type::STRING: {
             return build_bitset_for_index<ContainerType, stream::TableIndex>(
-                    container, rg, dynamic_schema, column_groups, std::move(input)
+                    container, rg, dynamic_schema, column_groups, is_read_operation, std::move(input)
             );
         }
         default:
@@ -252,7 +252,8 @@ inline void build_row_read_query_filters(
             },
             [&](const IndexRange& index_range) {
                 if (index_range.specified_) {
-                    queries.emplace_back(create_index_filter<ContainerType>(index_range, dynamic_schema, column_groups)
+                    queries.emplace_back(
+                            create_index_filter<ContainerType>(index_range, dynamic_schema, column_groups, true)
                     );
                 }
             },
@@ -335,25 +336,45 @@ inline std::vector<FilterQuery<ContainerType>> build_update_query_filters(
                         std::holds_alternative<stream::TimeseriesIndex>(index),
                         "Cannot partition by time when a rowcount-indexed frame was supplied"
                 );
-                queries.emplace_back(
-                        create_index_filter<ContainerType>(IndexRange{index_range}, dynamic_schema, column_groups)
-                );
+                queries.emplace_back(create_index_filter<ContainerType>(
+                        IndexRange{index_range}, dynamic_schema, column_groups, false
+                ));
             },
             [&](const auto&) {
                 util::variant_match(
                         index,
                         [&](const stream::TimeseriesIndex&) {
                             queries.emplace_back(create_index_filter<ContainerType>(
-                                    IndexRange{index_range}, dynamic_schema, column_groups
+                                    IndexRange{index_range}, dynamic_schema, column_groups, false
                             ));
                         },
-                        [&](const stream::RowCountIndex&) {
-                            RowRange row_range{
-                                    std::get<NumericId>(index_range.start_), std::get<NumericIndex>(index_range.end_)
-                            };
-                            queries.emplace_back(create_row_filter<ContainerType>(std::move(row_range)));
+                        [&](const IndexRange& index_range) {
+                            util::check(
+                                    std::holds_alternative<stream::TimeseriesIndex>(index),
+                                    "Cannot partition by time when a rowcount-indexed frame was supplied"
+                            );
+                            queries.emplace_back(create_index_filter<ContainerType>(
+                                    IndexRange{index_range}, dynamic_schema, column_groups, false
+                            ));
                         },
-                        [&](const auto&) {}
+                        [&](const auto&) {
+                            util::variant_match(
+                                    index,
+                                    [&](const stream::TimeseriesIndex&) {
+                                        queries.emplace_back(create_index_filter<ContainerType>(
+                                                IndexRange{index_range}, dynamic_schema, column_groups, false
+                                        ));
+                                    },
+                                    [&](const stream::RowCountIndex&) {
+                                        RowRange row_range{
+                                                std::get<NumericId>(index_range.start_),
+                                                std::get<NumericIndex>(index_range.end_)
+                                        };
+                                        queries.emplace_back(create_row_filter<ContainerType>(std::move(row_range)));
+                                    },
+                                    [&](const auto&) {}
+                            );
+                        }
                 );
             }
     );

cpp/arcticdb/pipeline/test/test_query.cpp

@@ -11,69 +11,89 @@
 
 #include <arcticdb/pipeline/test/test_container.hpp>
 
-TEST(BitsetForIndex, DynamicSchemaStrictlyBefore) {
+struct BitsetForIndex : public testing::TestWithParam<bool> {
+    bool is_read_operation() const { return GetParam(); }
+};
+
+TEST_P(BitsetForIndex, DynamicSchemaStrictlyBefore) {
     using namespace arcticdb;
     using namespace arcticdb::pipelines;
     TestContainer container;
     container.seg().set_range(3, 4);
     container.seg().set_range(5, 7);
     IndexRange rg(NumericIndex{0}, NumericIndex{2});
     auto bitset = build_bitset_for_index<TestContainer, TimeseriesIndex>(
-            container, rg, true, false, std::unique_ptr<util::BitSet>{}
+            container, rg, true, false, is_read_operation(), std::unique_ptr<util::BitSet>{}
     );
     ASSERT_EQ(bitset->count(), 0);
 }
 
-TEST(BitsetForIndex, DynamicSchemaStrictlyAfter) {
+TEST_P(BitsetForIndex, DynamicSchemaStrictlyAfter) {
     using namespace arcticdb;
     using namespace arcticdb::pipelines;
     TestContainer container;
     container.seg().set_range(0, 2);
     container.seg().set_range(3, 4);
     IndexRange rg(NumericIndex{5}, NumericIndex{7});
     auto bitset = build_bitset_for_index<TestContainer, TimeseriesIndex>(
-            container, rg, true, false, std::unique_ptr<util::BitSet>{}
+            container, rg, true, false, is_read_operation(), std::unique_ptr<util::BitSet>{}
     );
     ASSERT_EQ(bitset->count(), 0);
 }
 
-TEST(BitsetForIndex, DynamicSchemaMiddle) {
+TEST_P(BitsetForIndex, DynamicSchemaMiddle) {
     using namespace arcticdb;
     using namespace arcticdb::pipelines;
     TestContainer container;
     container.seg().set_range(0, 2);
     container.seg().set_range(5, 7);
     IndexRange rg(NumericIndex{3}, NumericIndex{4});
     auto bitset = build_bitset_for_index<TestContainer, TimeseriesIndex>(
-            container, rg, true, false, std::unique_ptr<util::BitSet>{}
+            container, rg, true, false, is_read_operation(), std::unique_ptr<util::BitSet>{}
     );
     ASSERT_EQ(bitset->count(), 0);
 }
 
-TEST(BitsetForIndex, DynamicSchemaOverlapBegin) {
+TEST_P(BitsetForIndex, DynamicSchemaOverlapBegin) {
     using namespace arcticdb;
     using namespace arcticdb::pipelines;
     TestContainer container;
     container.seg().set_range(2, 4);
     container.seg().set_range(5, 7);
     IndexRange rg(NumericIndex{1}, NumericIndex{3});
     auto bitset = build_bitset_for_index<TestContainer, TimeseriesIndex>(
-            container, rg, true, false, std::unique_ptr<util::BitSet>{}
+            container, rg, true, false, is_read_operation(), std::unique_ptr<util::BitSet>{}
     );
     ASSERT_EQ((*bitset)[0], true);
     ASSERT_EQ(bitset->count(), 1);
 }
 
-TEST(BitsetForIndex, DynamicSchemaOverlapEnd) {
+TEST_P(BitsetForIndex, DynamicSchemaOverlapEnd) {
     using namespace arcticdb;
     using namespace arcticdb::pipelines;
     TestContainer container;
     container.seg().set_range(2, 4);
     container.seg().set_range(5, 7);
     IndexRange rg(NumericIndex{6}, NumericIndex{8});
     auto bitset = build_bitset_for_index<TestContainer, TimeseriesIndex>(
-            container, rg, true, false, std::unique_ptr<util::BitSet>{}
+            container, rg, true, false, is_read_operation(), std::unique_ptr<util::BitSet>{}
     );
     ASSERT_EQ((*bitset)[1], true);
     ASSERT_EQ(bitset->count(), 1);
 }
+
+TEST_P(BitsetForIndex, DynamicSchemaMatchEndIndex) {
+    using namespace arcticdb;
+    using namespace arcticdb::pipelines;
+    TestContainer container;
+    container.seg().set_range(2, 4);
+    container.seg().set_range(5, 7);
+    IndexRange rg(NumericIndex{7}, NumericIndex{7});
+    auto bitset = build_bitset_for_index<TestContainer, TimeseriesIndex>(
+            container, rg, true, false, is_read_operation(), std::unique_ptr<util::BitSet>{}
+    );
+    ASSERT_EQ((*bitset)[1], is_read_operation());
+    ASSERT_EQ(bitset->count(), is_read_operation() ? 1 : 0);
+}
+
+INSTANTIATE_TEST_SUITE_P(BitsetForIndexTests, BitsetForIndex, testing::Values(true, false));

cpp/arcticdb/version/version_core.cpp

@@ -325,10 +325,12 @@ VersionedItem delete_range_impl(
     std::sort(std::begin(flattened_slice_and_keys), std::end(flattened_slice_and_keys));
     auto version_key_fut = util::variant_match(
             index,
-            [&index_segment_reader, &flattened_slice_and_keys, &stream_id, &update_info, &store](auto idx) {
+            [&index_segment_reader, &flattened_slice_and_keys, &stream_id, &update_info, &store, &row_count](auto idx) {
                 using IndexType = decltype(idx);
+                auto tsd = std::make_shared<TimeseriesDescriptor>(index_segment_reader.tsd().clone());
+                tsd->set_total_rows(row_count);
                 return pipelines::index::write_index<IndexType>(
-                        index_segment_reader.tsd(),
+                        *tsd,
                         std::move(flattened_slice_and_keys),
                         IndexPartialKey{stream_id, update_info.next_version_id_},
                         store

python/tests/conftest.py

@@ -1387,6 +1387,21 @@ def lmdb_version_store_static_and_dynamic(request) -> Generator[NativeVersionSto
     yield request.getfixturevalue(request.param)
 
 
+@pytest.fixture(
+    scope="function",
+    params=(
+        "lmdb_version_store_empty_types_v1",
+        "lmdb_version_store_empty_types_dynamic_schema_v1",
+    ),
+)
+def lmdb_version_store_static_and_dynamic_v1(request) -> Generator[NativeVersionStore, None, None]:
+    """
+    Designed to test the Native version store with API both static and dynamic schema
+    Uses only lmdb with encoding V1.
+    """
+    yield request.getfixturevalue(request.param)
+
+
 @pytest.fixture(
     scope="function",
     params=(

python/tests/integration/arcticdb/test_read_batch_more.py

@@ -147,7 +147,7 @@ def test_read_batch_2tables_7reads_different_slices(arctic_library):
     assert_frame_equal(df2_0_allfilters, batch[7].data)
 
 
-@pytest.mark.xfail(reason="ArcticDB#1970")
+@pytest.mark.skip(reason="ArcticDB#1970")
 @pytest.mark.storage
 def test_read_batch_query_with_and(arctic_library):
     """
@@ -175,6 +175,7 @@ def test_read_batch_query_with_and(arctic_library):
 
 
 @pytest.mark.storage
+@pytest.mark.skip(reason="ArcticDB#2004")
 def test_read_batch_metadata_on_different_version(arctic_library):
     """
     Here we test if read of metadata over several different states of DB with
@@ -388,7 +389,7 @@ def test_read_batch_multiple_wrong_things_at_once(arctic_library):
     assert_frame_equal_rebuild_index_first(df, batch[5].data)
 
 
-@pytest.mark.xfail(reason="ArcticDB#2004")
+@pytest.mark.skip(reason="ArcticDB#2004")
 @pytest.mark.storage
 def test_read_batch_query_and_columns_returned_order(arctic_library):
     """
@@ -413,7 +414,7 @@ def q(q):
     assert_frame_equal_rebuild_index_first(df_filtered, batch[0].data)
 
 
-@pytest.mark.xfail(reason="ArcticDB#2005")
+@pytest.mark.skip(reason="ArcticDB#2005")
 @pytest.mark.storage
 def test_read_batch_query_and_columns_wrong_column_names_passed(arctic_library):
     """

python/tests/integration/arcticdb/test_unicode_strings.py

@@ -105,7 +105,7 @@ def test_recursive_normalizers_blns(lmdb_version_store):
     assert_dicts_of_dfs_equal(dict, vit.data)
 
 
-@pytest.mark.xfail(reason="These do not roundtrip properly. Monday: 9256783357")
+@pytest.mark.skip(reason="These do not roundtrip properly. Monday: 9256783357")
 def test_recursive_normalizers_blns_in_keys(lmdb_version_store):
     lib = lmdb_version_store
     strings = read_strings()