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

Issue:
Fix resampling denormalization with multi-index
Minimal repro:
```
def test_resample_multiindex(lmdb_version_store_v1):
    lib = lmdb_version_store_v1
    sym = "test_resample_multiindex"
    df = pd.DataFrame(
        data={
            "a": [2, 3],
            "b": [4, 5],
        },
        index=pd.MultiIndex.from_arrays([[pd.Timestamp("2024-01-01"), pd.Timestamp("2024-01-02")], [0, 1]], names=["datetime", "sequence number"])
    )
    df.resample("D").agg({"a sum": ("a", "sum"), "b mean": ("b", "mean")})
    lib.write(sym, df)
    q = QueryBuilder()
    q = q.resample("D").agg({"a sum": ("a", "sum"), "b mean": ("b", "mean")})
    received_df = lib.read(sym, query_builder=q).data
    print(f"\n{received_df}")
# prints
                a sum
datetime             
2024-01-01 4.0      2
2024-01-02 5.0      3
```
i.e. the `b mean` column is in the place of the second-level of the index, and has lost its name


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index ceb508af69..1d54c71f44 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -173,7 +173,7 @@ AggregationClause::AggregationClause(const std::string& grouping_column,
         grouping_column_(grouping_column) {
     clause_info_.input_structure_ = ProcessingStructure::HASH_BUCKETED;
     clause_info_.can_combine_with_column_selection_ = false;
-    clause_info_.new_index_ = grouping_column_;
+    clause_info_.index_ = NewIndex(grouping_column_);
     clause_info_.input_columns_ = std::make_optional<std::unordered_set<std::string>>({grouping_column_});
     clause_info_.modifies_output_descriptor_ = true;
     str_ = "AGGREGATE {";
diff --git a/cpp/arcticdb/processing/clause.hpp b/cpp/arcticdb/processing/clause.hpp
index ff53585c61..901b622b79 100644
--- a/cpp/arcticdb/processing/clause.hpp
+++ b/cpp/arcticdb/processing/clause.hpp
@@ -346,6 +346,7 @@ struct ResampleClause {
         clause_info_.input_structure_ = ProcessingStructure::TIME_BUCKETED;
         clause_info_.can_combine_with_column_selection_ = false;
         clause_info_.modifies_output_descriptor_ = true;
+        clause_info_.index_ = KeepCurrentTopLevelIndex();
     }
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
diff --git a/cpp/arcticdb/processing/clause_utils.hpp b/cpp/arcticdb/processing/clause_utils.hpp
index ebac18ae3c..7bf93221d0 100644
--- a/cpp/arcticdb/processing/clause_utils.hpp
+++ b/cpp/arcticdb/processing/clause_utils.hpp
@@ -29,6 +29,10 @@ enum class ProcessingStructure {
     ALL
 };
 
+struct KeepCurrentIndex{};
+struct KeepCurrentTopLevelIndex{};
+using NewIndex = std::string;
+
 // Contains constant data about the clause identifiable at construction time
 struct ClauseInfo {
     // The arrangement of segments this clause needs in order for processing to be done correctly
@@ -40,8 +44,10 @@ struct ClauseInfo {
     // The names of the columns that are needed for this clause to make sense
     // Could either be on disk, or columns created by earlier clauses in the processing pipeline
     std::optional<std::unordered_set<std::string>> input_columns_{std::nullopt};
-    // The name of the index after this clause if it has been modified, std::nullopt otherwise
-    std::optional<std::string> new_index_{std::nullopt};
+    // KeepCurrentIndex if this clause does not modify the index in any way
+    // KeepCurrentTopLevelIndex if this clause requires multi-index levels>0 to be dropped, but otherwise does not modify it
+    // NewIndex if this clause has changed the index to a new (supplied) name
+    std::variant<KeepCurrentIndex, KeepCurrentTopLevelIndex, NewIndex> index_{KeepCurrentIndex()};
     // Whether this clause modifies the output descriptor
     bool modifies_output_descriptor_{false};
 };
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index 8aaa0cbd25..14d3fbb193 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -603,12 +603,38 @@ void set_output_descriptors(
         const std::shared_ptr<PipelineContext>& pipeline_context) {
     std::optional<std::string> index_column;
     for (auto clause = clauses.rbegin(); clause != clauses.rend(); ++clause) {
-        if (auto new_index = (*clause)->clause_info().new_index_; new_index.has_value()) {
-            index_column = new_index;
-            auto mutable_index = pipeline_context->norm_meta_->mutable_df()->mutable_common()->mutable_index();
-            mutable_index->set_name(*new_index);
-            mutable_index->clear_fake_name();
-            mutable_index->set_is_physically_stored(true);
+        bool should_break = util::variant_match(
+                (*clause)->clause_info().index_,
+                [](const KeepCurrentIndex&) { return false; },
+                [&](const KeepCurrentTopLevelIndex&) {
+                    if (pipeline_context->norm_meta_->mutable_df()->mutable_common()->has_multi_index()) {
+                        const auto& multi_index = pipeline_context->norm_meta_->mutable_df()->mutable_common()->multi_index();
+                        auto name = multi_index.name();
+                        auto tz = multi_index.tz();
+                        bool fake_name{false};
+                        for (auto pos: multi_index.fake_field_pos()) {
+                            if (pos == 0) {
+                                fake_name = true;
+                                break;
+                            }
+                        }
+                        auto mutable_index = pipeline_context->norm_meta_->mutable_df()->mutable_common()->mutable_index();
+                        mutable_index->set_tz(tz);
+                        mutable_index->set_is_physically_stored(true);
+                        mutable_index->set_name(name);
+                        mutable_index->set_fake_name(fake_name);
+                    }
+                    return true;
+                },
+                [&](const NewIndex& new_index) {
+                    index_column = new_index;
+                    auto mutable_index = pipeline_context->norm_meta_->mutable_df()->mutable_common()->mutable_index();
+                    mutable_index->set_name(new_index);
+                    mutable_index->clear_fake_name();
+                    mutable_index->set_is_physically_stored(true);
+                    return true;
+                });
+        if (should_break) {
             break;
         }
     }


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 1c910d13ca..965f679340 100644
--- a/python/tests/unit/arcticdb/version_store/test_resample.py
+++ b/python/tests/unit/arcticdb/version_store/test_resample.py
@@ -374,6 +374,36 @@ def test_resampling_empty_bucket_in_range(lmdb_version_store_v1):
     )
 
 
+@pytest.mark.parametrize("tz", (None, "Europe/London"))
+@pytest.mark.parametrize("named_levels", (True, False))
+def test_resample_multiindex(lmdb_version_store_v1, tz, named_levels):
+    lib = lmdb_version_store_v1
+    sym = "test_resample_multiindex"
+    multiindex = pd.MultiIndex.from_product([pd.date_range("2024-01-01", freq="h", periods=5, tz=tz), [0, 1], ["hello", "goodbye"]])
+    if named_levels:
+        multiindex.names = ["datetime", "sequence number", "another index level"]
+    df = pd.DataFrame(
+        data={
+            "to_sum": np.arange(len(multiindex)),
+            "to_mean": np.arange(len(multiindex)) * 10,
+        },
+        index=multiindex,
+    )
+    lib.write(sym, df)
+    freq = "2h"
+    aggs = {"to_sum": "sum", "to_mean": "mean"}
+
+    # Pandas doesn't support resampling multiindexed dataframes, but our behaviour is equivalent to there only being a
+    # top-level timeseries index
+    df.index = multiindex.droplevel(2).droplevel(1)
+    expected = df.resample(freq).agg(aggs)
+
+    q = QueryBuilder()
+    q = q.resample(freq).agg(aggs)
+    received = lib.read(sym, query_builder=q).data
+    assert_frame_equal(expected, received, check_dtype=False)
+
+
 @pytest.mark.parametrize("use_date_range", (True, False))
 @pytest.mark.parametrize("single_query", (True, False))
 def test_resampling_batch_read_query(lmdb_version_store_v1, use_date_range, single_query):