Don't lose mean offset from expanding window when persisting

Author: tveasey

lib/maths/CExpandingWindow.cc

@@ -25,6 +25,7 @@ namespace {
 const std::string BUCKET_LENGTH_INDEX_TAG("a");
 const std::string BUCKET_VALUES_TAG("b");
 const std::string START_TIME_TAG("c");
+const std::string MEAN_OFFSET_TAG("d");
 }
 
 CExpandingWindow::CExpandingWindow(core_t::TTime bucketLength,
@@ -45,6 +46,7 @@ bool CExpandingWindow::acceptRestoreTraverser(core::CStateRestoreTraverser& trav
         RESTORE_BUILT_IN(START_TIME_TAG, m_StartTime)
         RESTORE(BUCKET_VALUES_TAG,
                 core::CPersistUtils::restore(BUCKET_VALUES_TAG, m_BucketValues, traverser));
+        RESTORE(MEAN_OFFSET_TAG, m_MeanOffset.fromDelimited(traverser.value()))
     } while (traverser.next());
     return true;
 }
@@ -53,6 +55,7 @@ void CExpandingWindow::acceptPersistInserter(core::CStatePersistInserter& insert
     inserter.insertValue(BUCKET_LENGTH_INDEX_TAG, m_BucketLengthIndex);
     inserter.insertValue(START_TIME_TAG, m_StartTime);
     core::CPersistUtils::persist(BUCKET_VALUES_TAG, m_BucketValues, inserter);
+    inserter.insertValue(MEAN_OFFSET_TAG, m_MeanOffset.toDelimited());
 }
 
 core_t::TTime CExpandingWindow::startTime() const {
@@ -148,7 +151,8 @@ bool CExpandingWindow::needToCompress(core_t::TTime time) const {
 uint64_t CExpandingWindow::checksum(uint64_t seed) const {
     seed = CChecksum::calculate(seed, m_BucketLengthIndex);
     seed = CChecksum::calculate(seed, m_StartTime);
-    return CChecksum::calculate(seed, m_BucketValues);
+    seed = CChecksum::calculate(seed, m_BucketValues);
+    return CChecksum::calculate(seed, m_MeanOffset);
 }
 
 void CExpandingWindow::debugMemoryUsage(core::CMemoryUsage::TMemoryUsagePtr mem) const {

lib/maths/unittest/CExpandingWindowTest.cc

@@ -0,0 +1,90 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+#include "CExpandingWindowTest.h"
+
+#include <core/CContainerPrinter.h>
+#include <core/CLogger.h>
+#include <core/CRapidXmlStatePersistInserter.h>
+#include <core/CRapidXmlStateRestoreTraverser.h>
+#include <core/Constants.h>
+
+#include <maths/CBasicStatistics.h>
+#include <maths/CExpandingWindow.h>
+
+#include <test/CRandomNumbers.h>
+
+#include <boost/math/constants/constants.hpp>
+
+#include <vector>
+
+using namespace ml;
+
+namespace {
+using TDoubleVec = std::vector<double>;
+using TTimeVec = std::vector<core_t::TTime>;
+using TTimeCRng = core::CVectorRange<const TTimeVec>;
+using TFloatMeanAccumulator =
+    maths::CBasicStatistics::SSampleMean<maths::CFloatStorage>::TAccumulator;
+using TFloatMeanAccumulatorVec = std::vector<TFloatMeanAccumulator>;
+
+TTimeVec BUCKET_LENGTHS{300, 600, 1800, 3600};
+}
+
+void CExpandingWindowTest::testPersistence() {
+    // Test persist and restore is idempotent.
+
+    core_t::TTime bucketLength{300};
+    std::size_t size{336};
+    double decayRate{0.01};
+
+    test::CRandomNumbers rng;
+
+    maths::CExpandingWindow origWindow{bucketLength, TTimeCRng{BUCKET_LENGTHS, 0, 4},
+                                       size, decayRate};
+
+    TDoubleVec values;
+    rng.generateUniformSamples(0.0, 10.0, size, values);
+    for (core_t::TTime time = 0; time < static_cast<core_t::TTime>(size) * bucketLength;
+         time += bucketLength) {
+        double value{values[time / bucketLength]};
+        origWindow.add(time, value);
+    }
+
+    std::string origXml;
+    {
+        core::CRapidXmlStatePersistInserter inserter("root");
+        origWindow.acceptPersistInserter(inserter);
+        inserter.toXml(origXml);
+    }
+    LOG_TRACE(<< "Window XML = " << origXml);
+    LOG_DEBUG(<< "Window XML size = " << origXml.size());
+
+    // Restore the XML into a new window.
+    {
+        core::CRapidXmlParser parser;
+        CPPUNIT_ASSERT(parser.parseStringIgnoreCdata(origXml));
+        core::CRapidXmlStateRestoreTraverser traverser(parser);
+        maths::CExpandingWindow restoredWindow{
+            bucketLength, TTimeCRng{BUCKET_LENGTHS, 0, 4}, size, decayRate};
+        CPPUNIT_ASSERT_EQUAL(
+            true, traverser.traverseSubLevel(boost::bind(&maths::CExpandingWindow::acceptRestoreTraverser,
+                                                         &restoredWindow, _1)));
+
+        LOG_DEBUG(<< "orig checksum = " << origWindow.checksum()
+                  << ", new checksum = " << restoredWindow.checksum());
+        CPPUNIT_ASSERT_EQUAL(origWindow.checksum(), restoredWindow.checksum());
+    }
+}
+
+CppUnit::Test* CExpandingWindowTest::suite() {
+    CppUnit::TestSuite* suiteOfTests = new CppUnit::TestSuite("CExpandingWindowTest");
+
+    suiteOfTests->addTest(new CppUnit::TestCaller<CExpandingWindowTest>(
+        "CExpandingWindowTest::testPersistence", &CExpandingWindowTest::testPersistence));
+
+    return suiteOfTests;
+}

lib/maths/unittest/CExpandingWindowTest.h

@@ -0,0 +1,19 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+#ifndef INCLUDED_CExpandingWindowTest_h
+#define INCLUDED_CExpandingWindowTest_h
+
+#include <cppunit/extensions/HelperMacros.h>
+
+class CExpandingWindowTest : public CppUnit::TestFixture {
+public:
+    void testPersistence();
+
+    static CppUnit::Test* suite();
+};
+
+#endif // INCLUDED_CExpandingWindowTest_h

lib/maths/unittest/Main.cc

@@ -21,6 +21,7 @@
 #include "CDecayRateControllerTest.h"
 #include "CEntropySketchTest.h"
 #include "CEqualWithToleranceTest.h"
+#include "CExpandingWindowTest.h"
 #include "CForecastTest.h"
 #include "CGammaRateConjugateTest.h"
 #include "CGramSchmidtTest.h"
@@ -84,6 +85,7 @@
 int main(int argc, const char** argv) {
     ml::test::CTestRunner runner(argc, argv);
 
+    runner.addTest(CExpandingWindowTest::suite());
     runner.addTest(CAgglomerativeClustererTest::suite());
     runner.addTest(CAssignmentTest::suite());
     runner.addTest(CBasicStatisticsTest::suite());

lib/maths/unittest/Makefile

@@ -32,6 +32,7 @@ SRCS=\
 	CDecayRateControllerTest.cc \
 	CEntropySketchTest.cc \
 	CEqualWithToleranceTest.cc \
+	CExpandingWindowTest.cc \
 	CForecastTest.cc \
 	CGammaRateConjugateTest.cc \
 	CGramSchmidtTest.cc \