// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include <stdlib.h>
#include <algorithm>
#include <iostream>

#include <boost/bind.hpp>

#include "common/object-pool.h"
#include "testutil/gtest-util.h"
#include "util/container-util.h"
#include "util/periodic-counter-updater.h"
#include "util/runtime-profile-counters.h"
#include "util/thread.h"

#include "common/names.h"

DECLARE_int32(status_report_interval_ms);
DECLARE_int32(periodic_counter_update_period_ms);

namespace impala {

TEST(CountersTest, Basic) {
  ObjectPool pool;
  RuntimeProfile* profile_a = RuntimeProfile::Create(&pool, "ProfileA");
  RuntimeProfile* profile_a1 = RuntimeProfile::Create(&pool, "ProfileA1");
  RuntimeProfile* profile_a2 = RuntimeProfile::Create(&pool, "ProfileAb");

  TRuntimeProfileTree thrift_profile;

  profile_a->AddChild(profile_a1);
  profile_a->AddChild(profile_a2);

  // Test Empty
  profile_a->ToThrift(&thrift_profile);
  EXPECT_EQ(thrift_profile.nodes.size(), 3);
  thrift_profile.nodes.clear();

  RuntimeProfile::Counter* counter_a;
  RuntimeProfile::Counter* counter_b;
  RuntimeProfile::Counter* counter_merged;

  // Updating/setting counter
  counter_a = profile_a->AddCounter("A", TUnit::UNIT);
  EXPECT_TRUE(counter_a != NULL);
  counter_a->Add(10);
  counter_a->Add(-5);
  EXPECT_EQ(counter_a->value(), 5);
  counter_a->Set(1);
  EXPECT_EQ(counter_a->value(), 1);

  counter_b = profile_a2->AddCounter("B", TUnit::BYTES);
  EXPECT_TRUE(counter_b != NULL);

  // Update status to be included in ExecSummary
  TExecSummary exec_summary;
  TStatus status;
  status.__set_status_code(TErrorCode::CANCELLED);
  exec_summary.__set_status(status);
  profile_a->SetTExecSummary(exec_summary);

  // Serialize/deserialize to thrift
  profile_a->ToThrift(&thrift_profile);
  RuntimeProfile* from_thrift = RuntimeProfile::CreateFromThrift(&pool, thrift_profile);
  counter_merged = from_thrift->GetCounter("A");
  EXPECT_EQ(counter_merged->value(), 1);
  EXPECT_TRUE(from_thrift->GetCounter("Not there") ==  NULL);
  TExecSummary exec_summary_result;
  from_thrift->GetExecSummary(&exec_summary_result);
  EXPECT_EQ(exec_summary_result.status, status);

  // Serialize/deserialize to archive string
  string archive_str;
  EXPECT_OK(profile_a->SerializeToArchiveString(&archive_str));
  TRuntimeProfileTree deserialized_thrift_profile;
  EXPECT_OK(RuntimeProfile::DeserializeFromArchiveString(
      archive_str, &deserialized_thrift_profile));
  RuntimeProfile* deserialized_profile =
      RuntimeProfile::CreateFromThrift(&pool, deserialized_thrift_profile);
  counter_merged = deserialized_profile->GetCounter("A");
  EXPECT_EQ(counter_merged->value(), 1);
  EXPECT_TRUE(deserialized_profile->GetCounter("Not there") == nullptr);
  deserialized_profile->GetExecSummary(&exec_summary_result);
  EXPECT_EQ(exec_summary_result.status, status);

  // Averaged
  RuntimeProfile* averaged_profile = RuntimeProfile::Create(&pool, "Merged", true);
  averaged_profile->UpdateAverage(from_thrift);
  counter_merged = averaged_profile->GetCounter("A");
  EXPECT_EQ(counter_merged->value(), 1);

  // UpdateAverage again, there should be no change.
  averaged_profile->UpdateAverage(from_thrift);
  EXPECT_EQ(counter_merged->value(), 1);

  counter_a = profile_a2->AddCounter("A", TUnit::UNIT);
  counter_a->Set(3);
  averaged_profile->UpdateAverage(profile_a2);
  EXPECT_EQ(counter_merged->value(), 2);

  // Update
  RuntimeProfile* updated_profile = RuntimeProfile::Create(&pool, "Updated");
  updated_profile->Update(thrift_profile);
  RuntimeProfile::Counter* counter_updated = updated_profile->GetCounter("A");
  EXPECT_EQ(counter_updated->value(), 1);

  // Update 2 more times, counters should stay the same
  updated_profile->Update(thrift_profile);
  updated_profile->Update(thrift_profile);
  EXPECT_EQ(counter_updated->value(), 1);
}

void ValidateCounter(RuntimeProfile* profile, const string& name, int64_t value) {
  RuntimeProfile::Counter* counter = profile->GetCounter(name);
  EXPECT_TRUE(counter != NULL);
  EXPECT_EQ(counter->value(), value);
}

TEST(CountersTest, MergeAndUpdate) {
  // Create two trees.  Each tree has two children, one of which has the
  // same name in both trees.  Merging the two trees should result in 3
  // children, with the counters from the shared child aggregated.

  ObjectPool pool;
  RuntimeProfile* profile1 = RuntimeProfile::Create(&pool, "Parent1");
  RuntimeProfile* p1_child1 = RuntimeProfile::Create(&pool, "Child1");
  RuntimeProfile* p1_child2 = RuntimeProfile::Create(&pool, "Child2");
  profile1->AddChild(p1_child1);
  profile1->AddChild(p1_child2);

  RuntimeProfile* profile2 = RuntimeProfile::Create(&pool, "Parent2");
  RuntimeProfile* p2_child1 = RuntimeProfile::Create(&pool, "Child1");
  RuntimeProfile* p2_child3 = RuntimeProfile::Create(&pool, "Child3");
  profile2->AddChild(p2_child1);
  profile2->AddChild(p2_child3);

  // Create parent level counters
  RuntimeProfile::Counter* parent1_shared =
      profile1->AddCounter("Parent Shared", TUnit::UNIT);
  RuntimeProfile::Counter* parent2_shared =
      profile2->AddCounter("Parent Shared", TUnit::UNIT);
  RuntimeProfile::Counter* parent1_only =
      profile1->AddCounter("Parent 1 Only", TUnit::UNIT);
  RuntimeProfile::Counter* parent2_only =
      profile2->AddCounter("Parent 2 Only", TUnit::UNIT);
  parent1_shared->Add(1);
  parent2_shared->Add(3);
  parent1_only->Add(2);
  parent2_only->Add(5);

  // Create child level counters
  RuntimeProfile::Counter* p1_c1_shared =
    p1_child1->AddCounter("Child1 Shared", TUnit::UNIT);
  RuntimeProfile::Counter* p1_c1_only =
    p1_child1->AddCounter("Child1 Parent 1 Only", TUnit::UNIT);
  RuntimeProfile::Counter* p1_c2 =
    p1_child2->AddCounter("Child2", TUnit::UNIT);
  RuntimeProfile::Counter* p2_c1_shared =
    p2_child1->AddCounter("Child1 Shared", TUnit::UNIT);
  RuntimeProfile::Counter* p2_c1_only =
    p1_child1->AddCounter("Child1 Parent 2 Only", TUnit::UNIT);
  RuntimeProfile::Counter* p2_c3 =
    p2_child3->AddCounter("Child3", TUnit::UNIT);
  p1_c1_shared->Add(10);
  p1_c1_only->Add(50);
  p2_c1_shared->Add(20);
  p2_c1_only->Add(100);
  p2_c3->Add(30);
  p1_c2->Add(40);

  // Merge the two and validate
  TRuntimeProfileTree tprofile1;
  profile1->ToThrift(&tprofile1);
  RuntimeProfile* averaged_profile = RuntimeProfile::Create(&pool, "merged", true);
  averaged_profile->UpdateAverage(profile1);
  averaged_profile->UpdateAverage(profile2);
  EXPECT_EQ(5, averaged_profile->num_counters());
  ValidateCounter(averaged_profile, "Parent Shared", 2);
  ValidateCounter(averaged_profile, "Parent 1 Only", 2);
  ValidateCounter(averaged_profile, "Parent 2 Only", 5);

  vector<RuntimeProfile*> children;
  averaged_profile->GetChildren(&children);
  EXPECT_EQ(children.size(), 3);

  for (int i = 0; i < 3; ++i) {
    RuntimeProfile* profile = children[i];
    if (profile->name().compare("Child1") == 0) {
      EXPECT_EQ(5, profile->num_counters());
      ValidateCounter(profile, "Child1 Shared", 15);
      ValidateCounter(profile, "Child1 Parent 1 Only", 50);
      ValidateCounter(profile, "Child1 Parent 2 Only", 100);
    } else if (profile->name().compare("Child2") == 0) {
      EXPECT_EQ(3, profile->num_counters());
      ValidateCounter(profile, "Child2", 40);
    } else if (profile->name().compare("Child3") == 0) {
      EXPECT_EQ(3, profile->num_counters());
      ValidateCounter(profile, "Child3", 30);
    } else {
      FAIL();
    }
  }

  // make sure we can print
  stringstream dummy;
  averaged_profile->PrettyPrint(&dummy);

  // Update profile2 w/ profile1 and validate
  profile2->Update(tprofile1);
  EXPECT_EQ(5, profile2->num_counters());
  ValidateCounter(profile2, "Parent Shared", 1);
  ValidateCounter(profile2, "Parent 1 Only", 2);
  ValidateCounter(profile2, "Parent 2 Only", 5);

  profile2->GetChildren(&children);
  EXPECT_EQ(children.size(), 3);

  for (int i = 0; i < 3; ++i) {
    RuntimeProfile* profile = children[i];
    if (profile->name().compare("Child1") == 0) {
      EXPECT_EQ(5, profile->num_counters());
      ValidateCounter(profile, "Child1 Shared", 10);
      ValidateCounter(profile, "Child1 Parent 1 Only", 50);
      ValidateCounter(profile, "Child1 Parent 2 Only", 100);
    } else if (profile->name().compare("Child2") == 0) {
      EXPECT_EQ(3, profile->num_counters());
      ValidateCounter(profile, "Child2", 40);
    } else if (profile->name().compare("Child3") == 0) {
      EXPECT_EQ(3, profile->num_counters());
      ValidateCounter(profile, "Child3", 30);
    } else {
      FAIL();
    }
  }

  // make sure we can print
  profile2->PrettyPrint(&dummy);
}

// Regression test for IMPALA-6694 - child order isn't preserved if a child
// is prepended between updates.
TEST(CountersTest, MergeAndUpdateChildOrder) {
  ObjectPool pool;
  // Add Child2 first.
  RuntimeProfile* profile1 = RuntimeProfile::Create(&pool, "Parent");
  RuntimeProfile* p1_child2 = RuntimeProfile::Create(&pool, "Child2");
  profile1->AddChild(p1_child2);
  TRuntimeProfileTree tprofile1_v1, tprofile1_v2, tprofile1_v3;
  profile1->ToThrift(&tprofile1_v1);

  // Update averaged and deserialized profiles from the serialized profile.
  RuntimeProfile* averaged_profile = RuntimeProfile::Create(&pool, "merged", true);
  RuntimeProfile* deserialized_profile = RuntimeProfile::Create(&pool, "Parent");
  averaged_profile->UpdateAverage(profile1);
  deserialized_profile->Update(tprofile1_v1);

  std::vector<RuntimeProfile*> tmp_children;
  averaged_profile->GetChildren(&tmp_children);
  EXPECT_EQ(1, tmp_children.size());
  EXPECT_EQ("Child2", tmp_children[0]->name());
  deserialized_profile->GetChildren(&tmp_children);
  EXPECT_EQ(1, tmp_children.size());
  EXPECT_EQ("Child2", tmp_children[0]->name());

  // Prepend Child1 and update profiles.
  RuntimeProfile* p1_child1 = RuntimeProfile::Create(&pool, "Child1");
  profile1->PrependChild(p1_child1);
  profile1->ToThrift(&tprofile1_v2);
  averaged_profile->UpdateAverage(profile1);
  deserialized_profile->Update(tprofile1_v2);

  averaged_profile->GetChildren(&tmp_children);
  EXPECT_EQ(2, tmp_children.size());
  EXPECT_EQ("Child1", tmp_children[0]->name());
  EXPECT_EQ("Child2", tmp_children[1]->name());
  deserialized_profile->GetChildren(&tmp_children);
  EXPECT_EQ(2, tmp_children.size());
  EXPECT_EQ("Child1", tmp_children[0]->name());
  EXPECT_EQ("Child2", tmp_children[1]->name());

  // Test that changes in order of children is handled gracefully by preserving the
  // order from the previous update. Sorting puts the children in descending total time
  // order.
  p1_child1->total_time_counter()->Set(1);
  p1_child2->total_time_counter()->Set(2);
  profile1->SortChildrenByTotalTime();
  profile1->GetChildren(&tmp_children);
  EXPECT_EQ("Child2", tmp_children[0]->name());
  EXPECT_EQ("Child1", tmp_children[1]->name());
  profile1->ToThrift(&tprofile1_v3);
  averaged_profile->UpdateAverage(profile1);
  deserialized_profile->Update(tprofile1_v2);

  // The previous order of children that were already present is preserved.
  averaged_profile->GetChildren(&tmp_children);
  EXPECT_EQ(2, tmp_children.size());
  EXPECT_EQ("Child1", tmp_children[0]->name());
  EXPECT_EQ("Child2", tmp_children[1]->name());
  deserialized_profile->GetChildren(&tmp_children);
  EXPECT_EQ(2, tmp_children.size());
  EXPECT_EQ("Child1", tmp_children[0]->name());
  EXPECT_EQ("Child2", tmp_children[1]->name());

  // Make sure we can print the profiles.
  stringstream dummy;
  averaged_profile->PrettyPrint(&dummy);
  deserialized_profile->PrettyPrint(&dummy);
}

TEST(CountersTest, TotalTimeCounters) {
  ObjectPool pool;

  // Set up a three layer profile: parent -> child1 -> child2
  RuntimeProfile* parent = RuntimeProfile::Create(&pool, "Parent");
  RuntimeProfile* child1 = RuntimeProfile::Create(&pool, "Child1");
  RuntimeProfile* child2 = RuntimeProfile::Create(&pool, "Child2");
  child1->AddChild(child2);
  parent->AddChild(child1);

  // Part 1: Test accumulation of time up from child2 to child1 to parent
  // One millisecond passes in child2
  int64_t one_milli_ns = 1 * NANOS_PER_MICRO * MICROS_PER_MILLI;
  child2->total_time_counter()->Add(1 * NANOS_PER_MICRO * MICROS_PER_MILLI);
  parent->ComputeTimeInProfile();
  EXPECT_EQ(child2->total_time(), one_milli_ns);
  EXPECT_EQ(child2->local_time(), one_milli_ns);

  // Child1 is a parent of child2, so it is expected to contain at least as much time
  // as in child2. In this case, it is equal. However, none of the time is local.
  EXPECT_EQ(child1->total_time(), one_milli_ns);
  EXPECT_EQ(child1->local_time(), 0);

  // The parent is in the same situation as child1
  EXPECT_EQ(parent->total_time(), one_milli_ns);
  EXPECT_EQ(parent->local_time(), 0);

  // Time now accumulates up to child1
  child1->total_time_counter()->Add(child2->total_time());
  parent->ComputeTimeInProfile();

  // This doesn't change anything for anyone
  EXPECT_EQ(child2->total_time(), one_milli_ns);
  EXPECT_EQ(child2->local_time(), one_milli_ns);
  EXPECT_EQ(child1->total_time(), one_milli_ns);
  EXPECT_EQ(child1->local_time(), 0);
  EXPECT_EQ(parent->total_time(), one_milli_ns);
  EXPECT_EQ(parent->local_time(), 0);

  // Time now accumulates up to parent
  parent->total_time_counter()->Add(child1->total_time());
  parent->ComputeTimeInProfile();

  // This doesn't change anything for the parent
  EXPECT_EQ(child2->total_time(), one_milli_ns);
  EXPECT_EQ(child2->local_time(), one_milli_ns);
  EXPECT_EQ(child1->total_time(), one_milli_ns);
  EXPECT_EQ(child1->local_time(), 0);
  EXPECT_EQ(parent->total_time(), one_milli_ns);
  EXPECT_EQ(parent->local_time(), 0);

  // Part 2: Time accumulated in middle child
  // Add 1ms to the middle child
  child1->total_time_counter()->Add(one_milli_ns);
  parent->ComputeTimeInProfile();

  // Child2 did not change
  EXPECT_EQ(child2->total_time(), one_milli_ns);
  EXPECT_EQ(child2->local_time(), one_milli_ns);

  // Child1 has 1ms more of total time and local time
  EXPECT_EQ(child1->total_time(), 2 * one_milli_ns);
  EXPECT_EQ(child1->local_time(), one_milli_ns);

  // Parent has more total time, but no local time
  EXPECT_EQ(parent->total_time(), 2 * one_milli_ns);
  EXPECT_EQ(parent->local_time(), 0);

  // Accumulate the middle child up to the parent
  parent->total_time_counter()->Add(one_milli_ns);
  parent->ComputeTimeInProfile();

  // Doesn't change anything
  EXPECT_EQ(child2->total_time(), one_milli_ns);
  EXPECT_EQ(child2->local_time(), one_milli_ns);
  EXPECT_EQ(child1->total_time(), 2 * one_milli_ns);
  EXPECT_EQ(child1->local_time(), one_milli_ns);
  EXPECT_EQ(parent->total_time(), 2 * one_milli_ns);
  EXPECT_EQ(parent->local_time(), 0);

  // Part 3: Time accumulated at parent
  // Add 1ms to the parent
  parent->total_time_counter()->Add(one_milli_ns);
  parent->ComputeTimeInProfile();

  // Child1 and child2 don't change
  EXPECT_EQ(child2->total_time(), one_milli_ns);
  EXPECT_EQ(child2->local_time(), one_milli_ns);
  EXPECT_EQ(child1->total_time(), 2 * one_milli_ns);
  EXPECT_EQ(child1->local_time(), one_milli_ns);

  // Parent has 1ms more total time and local time
  EXPECT_EQ(parent->total_time(), 3 * one_milli_ns);
  EXPECT_EQ(parent->local_time(), one_milli_ns);
}

TEST(CountersTest, HighWaterMarkCounters) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
  RuntimeProfile::HighWaterMarkCounter* bytes_counter =
      profile->AddHighWaterMarkCounter("bytes", TUnit::BYTES);

  bytes_counter->Set(10);
  EXPECT_EQ(bytes_counter->current_value(), 10);
  EXPECT_EQ(bytes_counter->value(), 10);

  bytes_counter->Add(5);
  EXPECT_EQ(bytes_counter->current_value(), 15);
  EXPECT_EQ(bytes_counter->value(), 15);

  bytes_counter->Set(5);
  EXPECT_EQ(bytes_counter->current_value(), 5);
  EXPECT_EQ(bytes_counter->value(), 15);

  bytes_counter->Add(3);
  EXPECT_EQ(bytes_counter->current_value(), 8);
  EXPECT_EQ(bytes_counter->value(), 15);

  bool success = bytes_counter->TryAdd(20, 30);
  EXPECT_TRUE(success);
  EXPECT_EQ(bytes_counter->current_value(), 28);
  EXPECT_EQ(bytes_counter->value(), 28);

  success = bytes_counter->TryAdd(5, 30);
  EXPECT_FALSE(success);
  EXPECT_EQ(bytes_counter->current_value(), 28);
  EXPECT_EQ(bytes_counter->value(), 28);
}

TEST(CountersTest, SummaryStatsCounters) {
  ObjectPool pool;
  RuntimeProfile* profile1 = RuntimeProfile::Create(&pool, "Profile 1");
  RuntimeProfile::SummaryStatsCounter* summary_stats_counter_1 =
    profile1->AddSummaryStatsCounter("summary_stats", TUnit::UNIT);

  EXPECT_EQ(summary_stats_counter_1->value(), 0);
  EXPECT_EQ(summary_stats_counter_1->MinValue(), numeric_limits<int64_t>::max());
  EXPECT_EQ(summary_stats_counter_1->MaxValue(), numeric_limits<int64_t>::min());

  summary_stats_counter_1->UpdateCounter(10);
  EXPECT_EQ(summary_stats_counter_1->value(), 10);
  EXPECT_EQ(summary_stats_counter_1->MinValue(), 10);
  EXPECT_EQ(summary_stats_counter_1->MaxValue(), 10);

  // Check that the average stays the same when updating with the same number.
  summary_stats_counter_1->UpdateCounter(10);
  EXPECT_EQ(summary_stats_counter_1->value(), 10);
  EXPECT_EQ(summary_stats_counter_1->MinValue(), 10);
  EXPECT_EQ(summary_stats_counter_1->MaxValue(), 10);

  summary_stats_counter_1->UpdateCounter(40);
  EXPECT_EQ(summary_stats_counter_1->value(), 20);
  EXPECT_EQ(summary_stats_counter_1->MinValue(), 10);
  EXPECT_EQ(summary_stats_counter_1->MaxValue(), 40);

  // Verify an update with 0. This should still change the average as the number of
  // samples increase
  summary_stats_counter_1->UpdateCounter(0);
  EXPECT_EQ(summary_stats_counter_1->value(), 15);
  EXPECT_EQ(summary_stats_counter_1->MinValue(), 0);
  EXPECT_EQ(summary_stats_counter_1->MaxValue(), 40);

  // Verify a negative update..
  summary_stats_counter_1->UpdateCounter(-40);
  EXPECT_EQ(summary_stats_counter_1->value(), 4);
  EXPECT_EQ(summary_stats_counter_1->MinValue(), -40);
  EXPECT_EQ(summary_stats_counter_1->MaxValue(), 40);

  RuntimeProfile* profile2 = RuntimeProfile::Create(&pool, "Profile 2");
  RuntimeProfile::SummaryStatsCounter* summary_stats_counter_2 =
    profile2->AddSummaryStatsCounter("summary_stats", TUnit::UNIT);

  summary_stats_counter_2->UpdateCounter(100);
  EXPECT_EQ(summary_stats_counter_2->value(), 100);
  EXPECT_EQ(summary_stats_counter_2->MinValue(), 100);
  EXPECT_EQ(summary_stats_counter_2->MaxValue(), 100);

  TRuntimeProfileTree tprofile1;
  profile1->ToThrift(&tprofile1);

  // Merge profile1 and profile2 and check that profile2 is overwritten.
  profile2->Update(tprofile1);
  EXPECT_EQ(summary_stats_counter_2->value(), 4);
  EXPECT_EQ(summary_stats_counter_2->MinValue(), -40);
  EXPECT_EQ(summary_stats_counter_2->MaxValue(), 40);

}

TEST(CountersTest, DerivedCounters) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
  RuntimeProfile::Counter* bytes_counter =
      profile->AddCounter("bytes", TUnit::BYTES);
  RuntimeProfile::Counter* ticks_counter =
      profile->AddCounter("ticks", TUnit::TIME_NS);
  // set to 1 sec
  ticks_counter->Set(1000L * 1000L * 1000L);

  RuntimeProfile::DerivedCounter* throughput_counter =
      profile->AddDerivedCounter("throughput", TUnit::BYTES,
      bind<int64_t>(&RuntimeProfile::UnitsPerSecond, bytes_counter, ticks_counter));

  bytes_counter->Set(10);
  EXPECT_EQ(throughput_counter->value(), 10);
  bytes_counter->Set(20);
  EXPECT_EQ(throughput_counter->value(), 20);
  ticks_counter->Set(ticks_counter->value() / 2);
  EXPECT_EQ(throughput_counter->value(), 40);
}

TEST(CountersTest, AverageSetCounters) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
  RuntimeProfile::Counter* bytes_1_counter =
      profile->AddCounter("bytes 1", TUnit::BYTES);
  RuntimeProfile::Counter* bytes_2_counter =
      profile->AddCounter("bytes 2", TUnit::BYTES);

  bytes_1_counter->Set(10);
  RuntimeProfile::AveragedCounter bytes_avg(TUnit::BYTES);
  bytes_avg.UpdateCounter(bytes_1_counter);
  // Avg of 10L
  EXPECT_EQ(bytes_avg.value(), 10);
  bytes_1_counter->Set(20L);
  bytes_avg.UpdateCounter(bytes_1_counter);
  // Avg of 20L
  EXPECT_EQ(bytes_avg.value(), 20);
  bytes_2_counter->Set(40L);
  bytes_avg.UpdateCounter(bytes_2_counter);
  // Avg of 20L and 40L
  EXPECT_EQ(bytes_avg.value(), 30);
  bytes_2_counter->Set(30L);
  bytes_avg.UpdateCounter(bytes_2_counter);
  // Avg of 20L and 30L
  EXPECT_EQ(bytes_avg.value(), 25);

  RuntimeProfile::Counter* double_1_counter =
      profile->AddCounter("double 1", TUnit::DOUBLE_VALUE);
  RuntimeProfile::Counter* double_2_counter =
      profile->AddCounter("double 2", TUnit::DOUBLE_VALUE);
  double_1_counter->Set(1.0f);
  RuntimeProfile::AveragedCounter double_avg(TUnit::DOUBLE_VALUE);
  double_avg.UpdateCounter(double_1_counter);
  // Avg of 1.0f
  EXPECT_EQ(double_avg.double_value(), 1.0f);
  double_1_counter->Set(2.0f);
  double_avg.UpdateCounter(double_1_counter);
  // Avg of 2.0f
  EXPECT_EQ(double_avg.double_value(), 2.0f);
  double_2_counter->Set(4.0f);
  double_avg.UpdateCounter(double_2_counter);
  // Avg of 2.0f and 4.0f
  EXPECT_EQ(double_avg.double_value(), 3.0f);
  double_2_counter->Set(3.0f);
  double_avg.UpdateCounter(double_2_counter);
  // Avg of 2.0f and 3.0f
  EXPECT_EQ(double_avg.double_value(), 2.5f);
}

TEST(CountersTest, InfoStringTest) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
  EXPECT_TRUE(profile->GetInfoString("Key") == NULL);

  profile->AddInfoString("Key", "Value");
  const string* value = profile->GetInfoString("Key");
  EXPECT_TRUE(value != NULL);
  EXPECT_EQ(*value, "Value");

  // Convert it to thrift
  TRuntimeProfileTree tprofile;
  profile->ToThrift(&tprofile);

  // Convert it back
  RuntimeProfile* from_thrift = RuntimeProfile::CreateFromThrift(
      &pool, tprofile);
  value = from_thrift->GetInfoString("Key");
  EXPECT_TRUE(value != NULL);
  EXPECT_EQ(*value, "Value");

  // Test update.
  RuntimeProfile* update_dst_profile = RuntimeProfile::Create(&pool, "Profile2");
  update_dst_profile->Update(tprofile);
  value = update_dst_profile->GetInfoString("Key");
  EXPECT_TRUE(value != NULL);
  EXPECT_EQ(*value, "Value");

  // Update the original profile, convert it to thrift and update from the dst
  // profile
  profile->AddInfoString("Key", "NewValue");
  profile->AddInfoString("Foo", "Bar");
  EXPECT_EQ(*profile->GetInfoString("Key"), "NewValue");
  EXPECT_EQ(*profile->GetInfoString("Foo"), "Bar");
  profile->ToThrift(&tprofile);

  update_dst_profile->Update(tprofile);
  EXPECT_EQ(*update_dst_profile->GetInfoString("Key"), "NewValue");
  EXPECT_EQ(*update_dst_profile->GetInfoString("Foo"), "Bar");
}

TEST(CountersTest, RateCounters) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");

  RuntimeProfile::Counter* bytes_counter =
      profile->AddCounter("bytes", TUnit::BYTES);

  RuntimeProfile::Counter* rate_counter =
      profile->AddRateCounter("RateCounter", bytes_counter);
  EXPECT_TRUE(rate_counter->unit() == TUnit::BYTES_PER_SECOND);

  EXPECT_EQ(rate_counter->value(), 0);
  // set to 100MB.  Use bigger units to avoid truncating to 0 after divides.
  bytes_counter->Set(100L * 1024L * 1024L);

  // Wait one second.
  sleep(1);

  int64_t rate = rate_counter->value();

  // Stop the counter so it no longer gets updates
  profile->StopPeriodicCounters();

  // The rate counter is not perfectly accurate.  Currently updated at 500ms intervals,
  // we should have seen somewhere between 1 and 3 updates (33 - 200 MB/s)
  EXPECT_GT(rate, 66 * 1024 * 1024);
  EXPECT_LE(rate, 200 * 1024 * 1024);

  // Wait another second.  The counter has been removed. So the value should not be
  // changed (much).
  sleep(2);

  rate = rate_counter->value();
  EXPECT_GT(rate, 66 * 1024 * 1024);
  EXPECT_LE(rate, 200 * 1024 * 1024);
}

TEST(CountersTest, BucketCounters) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");

  RuntimeProfile::Counter* unit_counter =
      profile->AddCounter("unit", TUnit::UNIT);

  // Set the unit to 1 before sampling
  unit_counter->Set(1);

  // Create the bucket counters and start sampling
  vector<RuntimeProfile::Counter*>* buckets =
      profile->AddBucketingCounters(unit_counter, 2);

  // Wait two seconds.
  sleep(2);

  // Stop sampling
  profile->StopPeriodicCounters();

  // TODO: change the value to double
  // The value of buckets[0] should be zero and buckets[1] should be 1.
  double val0 = (*buckets)[0]->double_value();
  double val1 = (*buckets)[1]->double_value();
  EXPECT_EQ(0, val0);
  EXPECT_EQ(100, val1);

  // Wait another second.  The counter has been removed. So the value should not be
  // changed (much).
  sleep(2);
  EXPECT_EQ(val0, (*buckets)[0]->double_value());
  EXPECT_EQ(val1, (*buckets)[1]->double_value());
}

TEST(CountersTest, EventSequences) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
  RuntimeProfile::EventSequence* seq = profile->AddEventSequence("event sequence");
  seq->MarkEvent("aaaa");
  seq->MarkEvent("bbbb");
  seq->MarkEvent("cccc");

  vector<RuntimeProfile::EventSequence::Event> events;
  seq->GetEvents(&events);
  EXPECT_EQ(3, events.size());

  uint64_t last_timestamp = 0;
  string last_string = "";
  for (const RuntimeProfile::EventSequence::Event& ev: events) {
    EXPECT_TRUE(ev.second >= last_timestamp);
    last_timestamp = ev.second;
    EXPECT_TRUE(ev.first > last_string);
    last_string = ev.first;
  }

  TRuntimeProfileTree thrift_profile;
  profile->ToThrift(&thrift_profile);
  EXPECT_TRUE(thrift_profile.nodes[0].__isset.event_sequences);
  EXPECT_EQ(1, thrift_profile.nodes[0].event_sequences.size());

  RuntimeProfile* reconstructed_profile =
      RuntimeProfile::CreateFromThrift(&pool, thrift_profile);

  last_timestamp = 0;
  last_string = "";
  EXPECT_EQ(NULL, reconstructed_profile->GetEventSequence("doesn't exist"));
  seq = reconstructed_profile->GetEventSequence("event sequence");
  EXPECT_TRUE(seq != NULL);
  seq->GetEvents(&events);
  EXPECT_EQ(3, events.size());
  for (const RuntimeProfile::EventSequence::Event& ev: events) {
    EXPECT_TRUE(ev.second >= last_timestamp);
    last_timestamp = ev.second;
    EXPECT_TRUE(ev.first > last_string);
    last_string = ev.first;
  }
}

TEST(CountersTest, UpdateEmptyEventSequence) {
  // IMPALA-6824: This test makes sure that adding events to an empty event sequence does
  // not crash.
  ObjectPool pool;

  // Create the profile to send in the update and add some events.
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
  RuntimeProfile::EventSequence* seq = profile->AddEventSequence("event sequence");
  seq->Start();
  // Sleep for 10ms to make sure the events are logged at a time > 0.
  SleepForMs(10);
  seq->MarkEvent("aaaa");
  seq->MarkEvent("bbbb");

  vector<RuntimeProfile::EventSequence::Event> events;
  seq->GetEvents(&events);
  EXPECT_EQ(2, events.size());

  TRuntimeProfileTree thrift_profile;
  profile->ToThrift(&thrift_profile);

  // Create the profile that will be updated and add the empty event sequence to it.
  RuntimeProfile* updated_profile = RuntimeProfile::Create(&pool, "Updated Profile");
  seq = updated_profile->AddEventSequence("event sequence");
  updated_profile->Update(thrift_profile);

  // Verify that the events have been updated successfully.
  events.clear();
  seq->GetEvents(&events);
  EXPECT_EQ(2, events.size());
}

void ValidateSampler(const StreamingSampler<int, 10>& sampler, int expected_num,
    int expected_period, int expected_delta) {
  const int* samples = NULL;
  int num_samples;
  int period;

  samples = sampler.GetSamples(&num_samples, &period);
  EXPECT_TRUE(samples != NULL);
  EXPECT_EQ(num_samples, expected_num);
  EXPECT_EQ(period, expected_period);

  for (int i = 0; i < expected_num - 1; ++i) {
    EXPECT_EQ(samples[i] + expected_delta, samples[i + 1]) << i;
  }
}

TEST(CountersTest, StreamingSampler) {
  StreamingSampler<int, 10> sampler;

  int idx = 0;
  for (int i = 0; i < 3; ++i) {
    sampler.AddSample(idx++, 500);
  }
  ValidateSampler(sampler, 3, 500, 1);

  for (int i = 0; i < 3; ++i) {
    sampler.AddSample(idx++, 500);
  }
  ValidateSampler(sampler, 6, 500, 1);

  for (int i = 0; i < 3; ++i) {
    sampler.AddSample(idx++, 500);
  }
  ValidateSampler(sampler, 9, 500, 1);

  // Added enough to cause a collapse
  for (int i = 0; i < 3; ++i) {
    sampler.AddSample(idx++, 500);
  }
  // Added enough to cause a collapse
  ValidateSampler(sampler, 6, 1000, 2);

  for (int i = 0; i < 3; ++i) {
    sampler.AddSample(idx++, 500);
  }
  ValidateSampler(sampler, 7, 1000, 2);
}

// Test class to test ConcurrentStopWatch and RuntimeProfile::ConcurrentTimerCounter
// don't double count in multithread environment.
class TimerCounterTest {
 public:
  TimerCounterTest()
    : timercounter_(TUnit::TIME_NS) {}

  struct DummyWorker {
    thread* thread_handle;
    bool done;

    DummyWorker()
      : thread_handle(NULL), done(false) {}

    ~DummyWorker() {
      Stop();
    }

    void Stop() {
      if (!done && thread_handle != NULL) {
        done = true;
        thread_handle->join();
        delete thread_handle;
        thread_handle = NULL;
      }
    }
  };

  void Run(DummyWorker* worker) {
    SCOPED_CONCURRENT_STOP_WATCH(&csw_);
    SCOPED_CONCURRENT_COUNTER(&timercounter_);
    while (!worker->done) {
      SleepForMs(10);
      // Each test case should be no more than one second.
      // Consider test failed if timer is more than 3 seconds.
      if (csw_.TotalRunningTime() > 6000000000) {
        FAIL();
      }
    }
  }

  // Start certain number of worker threads. If interval is set, it will add some delay
  // between creating worker thread.
  void StartWorkers(int num, int interval) {
    workers_.reserve(num);
    for (int i = 0; i < num; ++i) {
      workers_.push_back(DummyWorker());
      DummyWorker& worker = workers_.back();
      worker.thread_handle = new thread(&TimerCounterTest::Run, this, &worker);
      SleepForMs(interval);
    }
  }

  // Stop specified thread by index. if index is -1, stop all threads
  void StopWorkers(int thread_index = -1) {
    if (thread_index >= 0) {
      workers_[thread_index].Stop();
    } else {
      for (int i = 0; i < workers_.size(); ++i) {
        workers_[i].Stop();
      }
    }
  }

  void Reset() {
    workers_.clear();
  }

  // Allow some timer inaccuracy (30ms) since thread join could take some time.
  static const int MAX_TIMER_ERROR_NS = 30000000;
  vector<DummyWorker> workers_;
  ConcurrentStopWatch csw_;
  RuntimeProfile::ConcurrentTimerCounter timercounter_;
};

void ValidateTimerValue(const TimerCounterTest& timer, int64_t start) {
  int64_t expected_value = MonotonicStopWatch::Now() - start;
  int64_t stopwatch_value = timer.csw_.TotalRunningTime();
  EXPECT_GE(stopwatch_value, expected_value - TimerCounterTest::MAX_TIMER_ERROR_NS);
  EXPECT_LE(stopwatch_value, expected_value + TimerCounterTest::MAX_TIMER_ERROR_NS);

  int64_t timer_value = timer.timercounter_.value();
  EXPECT_GE(timer_value, expected_value - TimerCounterTest::MAX_TIMER_ERROR_NS);
  EXPECT_LE(timer_value, expected_value + TimerCounterTest::MAX_TIMER_ERROR_NS);
}

void ValidateLapTime(TimerCounterTest* timer, int64_t expected_value) {
  int64_t stopwatch_value = timer->csw_.LapTime();
  EXPECT_GE(stopwatch_value, expected_value - TimerCounterTest::MAX_TIMER_ERROR_NS);
  EXPECT_LE(stopwatch_value, expected_value + TimerCounterTest::MAX_TIMER_ERROR_NS);

  int64_t timer_value = timer->timercounter_.LapTime();
  EXPECT_GE(timer_value, expected_value - TimerCounterTest::MAX_TIMER_ERROR_NS);
  EXPECT_LE(timer_value, expected_value + TimerCounterTest::MAX_TIMER_ERROR_NS);
}

TEST(TimerCounterTest, CountersTestOneThread) {
  TimerCounterTest tester;
  int64_t start = MonotonicStopWatch::Now();
  tester.StartWorkers(1, 0);
  SleepForMs(500);
  ValidateTimerValue(tester, start);
  tester.StopWorkers(-1);
  ValidateTimerValue(tester, start);
}

TEST(TimerCounterTest, CountersTestTwoThreads) {
  TimerCounterTest tester;
  int64_t start = MonotonicStopWatch::Now();
  tester.StartWorkers(2, 10);
  SleepForMs(500);
  ValidateTimerValue(tester, start);
  tester.StopWorkers(-1);
  ValidateTimerValue(tester, start);
}

TEST(TimerCounterTest, CountersTestRandom) {
  TimerCounterTest tester;
  int64_t start = MonotonicStopWatch::Now();
  ValidateTimerValue(tester, start);
  // First working period
  tester.StartWorkers(5, 10);
  ValidateTimerValue(tester, start);
  SleepForMs(400);
  tester.StopWorkers(2);
  ValidateTimerValue(tester, start);
  SleepForMs(100);
  tester.StopWorkers(4);
  ValidateTimerValue(tester, start);
  SleepForMs(600);
  tester.StopWorkers(-1);
  ValidateTimerValue(tester, start);
  tester.Reset();

  ValidateLapTime(&tester, MonotonicStopWatch::Now() - start);
  int64_t first_run_end = MonotonicStopWatch::Now();
  // Adding some idle time. concurrent stopwatch and timer should not count the idle time.
  SleepForMs(200);
  start += MonotonicStopWatch::Now() - first_run_end;

  // Second working period
  tester.StartWorkers(2, 0);
  // We just get lap time after first run finish. so at start of second run, expect lap time == 0
  ValidateLapTime(&tester, 0);
  int64_t lap_time_start = MonotonicStopWatch::Now();
  SleepForMs(200);
  ValidateTimerValue(tester, start);
  SleepForMs(200);
  tester.StopWorkers(-1);
  ValidateTimerValue(tester, start);
  ValidateLapTime(&tester, MonotonicStopWatch::Now() - lap_time_start);
}


TEST(TimeSeriesCounterTest, TestAddClearRace) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
  int i = 0;
  // Return and increment i
  auto f = [&i]() { return i++; };
  RuntimeProfile::TimeSeriesCounter* counter =
      profile->AddChunkedTimeSeriesCounter("Counter", TUnit::UNIT, f);
  // Sleep 1 second for some values to accumulate.
  sleep(1);
  int num_samples, period;
  counter->GetSamplesTest(&num_samples, &period);
  EXPECT_GT(num_samples, 0);

  // Wait for more values to show up
  sleep(1);

  // Stop the counters. The rest of the test assumes that no new values will be added.
  profile->StopPeriodicCounters();

  // Clear the counter
  profile->ClearChunkedTimeSeriesCounters();

  // Check that clearing multiple times doesn't affect valued that have not been
  // retrieved.
  profile->ClearChunkedTimeSeriesCounters();

  // Make sure that it still has values in it.
  counter->GetSamplesTest(&num_samples, &period);
  EXPECT_GT(num_samples, 0);

  // Clear it again
  profile->ClearChunkedTimeSeriesCounters();

  // Make sure the values are gone.
  counter->GetSamplesTest(&num_samples, &period);
  EXPECT_EQ(num_samples, 0);
}

/// Stops the periodic counter updater in 'profile' and then clears the samples in
/// 'counter'.
void StopAndClearCounter(RuntimeProfile* profile,
    RuntimeProfile::TimeSeriesCounter* counter) {
  // There's a race between adding the counter and calling StopPeriodicCounters so we
  // sleep here to make sure we exercise the code that handles the race.
  sleep(1);
  profile->StopPeriodicCounters();

  // Reset the counter state by reading and clearing its samples.
  int num_samples = 0;
  int result_period_unused = 0;
  counter->GetSamplesTest(&num_samples, &result_period_unused);
  ASSERT_GT(num_samples, 0);
  profile->ClearChunkedTimeSeriesCounters();
  // Ensure clean state.
  counter->GetSamplesTest(&num_samples, &result_period_unused);
  ASSERT_EQ(num_samples, 0);
}

/// Tests that ChunkedTimeSeriesCounters are bounded by a maximum size.
TEST(TimeSeriesCounterTest, TestMaximumSize) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");

  const int test_period = FLAGS_periodic_counter_update_period_ms;

  // Add a counter with a sample function that counts up, starting from 0.
  int value = 0;
  auto sample_fn = [&value]() { return value++; };
  RuntimeProfile::TimeSeriesCounter* counter =
      profile->AddChunkedTimeSeriesCounter("TestCounter", TUnit::UNIT, sample_fn);

  // Stop counter updates from interfering with the rest of the test.
  StopAndClearCounter(profile, counter);

  // Reset value after previous values have been retrieved.
  value = 0;

  int64_t max_size = 10 * FLAGS_status_report_interval_ms / test_period;
  for (int i = 0; i < 10 + max_size; ++i) counter->AddSample(test_period);

  int num_samples = 0;
  int result_period = 0;
  // Retrieve and validate samples.
  const int64_t* samples = counter->GetSamplesTest(&num_samples, &result_period);
  ASSERT_EQ(num_samples, max_size);
  // No resampling happens with ChunkedTimeSeriesCounter.
  ASSERT_EQ(result_period, test_period);

  // First 10 samples have been truncated
  ASSERT_EQ(samples[0], 10);
}

/// Test parameter class that helps to test time series resampling during profile pretty
/// printing with a varying number of test samples.
struct TimeSeriesTestParam {
  TimeSeriesTestParam(int num_samples, vector<const char*> expected)
    : num_samples(num_samples), expected(expected) {}
  int num_samples;
  vector<const char*> expected;

  // Used by gtest to print values of this struct
  friend std::ostream& operator<<(std::ostream& os, const TimeSeriesTestParam& p) {
    return os << "num_samples: " << p.num_samples << endl;
  }
};

class TimeSeriesCounterResampleTest : public testing::TestWithParam<TimeSeriesTestParam> {
};

/// Tests that pretty-printing a ChunkedTimeSeriesCounter limits the number or printed
/// samples to 64 or lower.
TEST_P(TimeSeriesCounterResampleTest, TestPrettyPrint) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");

  const TimeSeriesTestParam& param = GetParam();
  const int test_period = FLAGS_periodic_counter_update_period_ms;

  // Add a counter with a sample function that counts up, starting from 0.
  int value = 0;
  auto sample_fn = [&value]() { return value++; };
  // We increase the value of this flag to allow the counter to store enough samples.
  FLAGS_status_report_interval_ms = 50000;
  RuntimeProfile::TimeSeriesCounter* counter =
      profile->AddChunkedTimeSeriesCounter("TestCounter", TUnit::UNIT, sample_fn);

  // Stop counter updates from interfering with the rest of the test.
  StopAndClearCounter(profile, counter);

  // Reset value after previous values have been retrieved.
  value = 0;
  for (int i = 0; i < param.num_samples; ++i) counter->AddSample(test_period);

  int num_samples = 0;
  int result_period = 0;
  // Retrieve and validate samples.
  const int64_t* samples = counter->GetSamplesTest(&num_samples, &result_period);
  ASSERT_EQ(num_samples, param.num_samples);
  // No resampling happens with ChunkedTimeSeriesCounter.
  ASSERT_EQ(result_period, test_period);

  for (int i = 0; i < param.num_samples; ++i) ASSERT_EQ(samples[i], i);

  stringstream pretty;
  profile->PrettyPrint(&pretty);
  const string pretty_str = pretty.str();

  for (const char* e : param.expected) EXPECT_STR_CONTAINS(pretty_str, e);
}

INSTANTIATE_TEST_CASE_P(VariousNumbers, TimeSeriesCounterResampleTest,
    ::testing::Values(
    TimeSeriesTestParam(64, {"TestCounter (500.000ms): 0, 1, 2, 3", "61, 62, 63"}),

    TimeSeriesTestParam(65, {"TestCounter (1s000ms): 0, 2, 4, 6,",
    "60, 62, 64 (Showing 33 of 65 values from Thrift Profile)"}),

    TimeSeriesTestParam(80, {"TestCounter (1s000ms): 0, 2, 4, 6,",
    "74, 76, 78 (Showing 40 of 80 values from Thrift Profile)"}),

    TimeSeriesTestParam(127, {"TestCounter (1s000ms): 0, 2, 4, 6,",
    "122, 124, 126 (Showing 64 of 127 values from Thrift Profile)"}),

    TimeSeriesTestParam(128, {"TestCounter (1s000ms): 0, 2, 4, 6,",
    "122, 124, 126 (Showing 64 of 128 values from Thrift Profile)"}),

    TimeSeriesTestParam(129, {"TestCounter (1s500ms): 0, 3, 6, 9,",
    "120, 123, 126 (Showing 43 of 129 values from Thrift Profile)"})
    ));

// Tests that the __isset field for TRuntimeProfileNode.node_metadata is set correctly
// (IMPALA-8252).
TEST(ToThrift, NodeMetadataIsSetCorrectly) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
  TRuntimeProfileTree thrift_profile;
  profile->ToThrift(&thrift_profile);
  // Profile is empty, expect 0 nodes
  EXPECT_EQ(thrift_profile.nodes.size(), 1);
  EXPECT_FALSE(thrift_profile.nodes[0].__isset.node_metadata);

  // Set the plan node ID and make sure that the field is marked correctly
  profile->SetPlanNodeId(1);
  profile->ToThrift(&thrift_profile);
  EXPECT_TRUE(thrift_profile.nodes[0].__isset.node_metadata);
}

TEST(ToJson, RuntimeProfileToJsonTest) {
  ObjectPool pool;
  RuntimeProfile* profile_a = RuntimeProfile::Create(&pool, "ProfileA");
  RuntimeProfile* profile_a1 = RuntimeProfile::Create(&pool, "ProfileA1");
  RuntimeProfile* profile_ab = RuntimeProfile::Create(&pool, "ProfileAb");
  RuntimeProfile::Counter* counter_a;

  // Initialize for further validation
  profile_a->AddChild(profile_a1);
  profile_a->AddChild(profile_ab);
  profile_a->AddInfoString("Key", "Value");

  counter_a = profile_a->AddCounter("A", TUnit::UNIT);
  counter_a->Set(1);
  RuntimeProfile::HighWaterMarkCounter* high_water_counter =
      profile_a->AddHighWaterMarkCounter("high_water_counter", TUnit::BYTES);
  high_water_counter->Set(10);
  high_water_counter->Add(10);
  high_water_counter->Set(10);

  RuntimeProfile::SummaryStatsCounter* summary_stats_counter =
      profile_a->AddSummaryStatsCounter("summary_stats_counter", TUnit::TIME_NS);
  summary_stats_counter->UpdateCounter(10);
  summary_stats_counter->UpdateCounter(20);

  // Serialize to json
  rapidjson::Document doc(rapidjson::kObjectType);
  profile_a->ToJson(&doc);
  rapidjson::Value& content = doc["contents"];

  // Check profile correct
  EXPECT_EQ("ProfileA", content["profile_name"]);
  EXPECT_EQ("ProfileA1", content["child_profiles"][0]["profile_name"]);
  EXPECT_EQ("ProfileAb", content["child_profiles"][1]["profile_name"]);

  // Check Info String correct
  EXPECT_EQ(1, content["info_strings"].Size());
  EXPECT_EQ("Key", content["info_strings"][0]["key"]);
  EXPECT_EQ("Value", content["info_strings"][0]["value"]);

  // Check counter value matches
  EXPECT_EQ(2, content["counters"].Size());
  for (auto& itr : content["counters"].GetArray()) {
    // check normal Counter
    if (itr["counter_name"] == "A") {
      EXPECT_EQ(1, itr["value"].GetInt());
      EXPECT_EQ("UNIT", itr["unit"]);
      EXPECT_EQ("Counter", itr["kind"]);
    }// check HighWaterMarkCounter
    else if (itr["counter_name"] == "high_water_counter") {
      EXPECT_EQ(20, itr["value"].GetInt());
      EXPECT_EQ("BYTES", itr["unit"]);
      EXPECT_EQ("HighWaterMarkCounter", itr["kind"]);
    } else {
      DCHECK(false);
    }
  }

  // Check SummaryStatsCounter
  EXPECT_EQ(1, content["summary_stats_counters"].Size());
  for (auto& itr : content["summary_stats_counters"].GetArray()) {
    if (itr["counter_name"] == "summary_stats_counter") {
      EXPECT_EQ(10, itr["min"].GetInt());
      EXPECT_EQ(20, itr["max"].GetInt());
      EXPECT_EQ(15, itr["avg"].GetInt());
      EXPECT_EQ(2, itr["num_of_samples"].GetInt());
      EXPECT_EQ("TIME_NS", itr["unit"]);
      EXPECT_TRUE(!itr.HasMember("kind"));
    }
  }
}

// Test when some fields are not set. ToJson will not add them as a member
TEST(ToJson, EmptyTest) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");

  // Serialize to json
  rapidjson::Document doc(rapidjson::kObjectType);
  profile->ToJson(&doc);
  rapidjson::Value& content = doc["contents"];

  EXPECT_EQ("Profile", content["profile_name"]);
  EXPECT_TRUE(content.HasMember("num_children"));

  // Empty profile should not have following members
  EXPECT_TRUE(!content.HasMember("info_strings"));
  EXPECT_TRUE(!content.HasMember("event_sequences"));
  EXPECT_TRUE(!content.HasMember("counters"));
  EXPECT_TRUE(!content.HasMember("summary_stats_counters"));
  EXPECT_TRUE(!content.HasMember("time_series_counters"));
  EXPECT_TRUE(!content.HasMember("child_profiles"));

}

TEST(ToJson, EventSequenceToJsonTest) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
  RuntimeProfile::EventSequence* seq = profile->AddEventSequence("event sequence");
  seq->MarkEvent("aaaa");
  seq->MarkEvent("bbbb");
  seq->MarkEvent("cccc");

  // Serialize to json
  rapidjson::Document doc(rapidjson::kObjectType);
  rapidjson::Value event_sequence_json(rapidjson::kObjectType);
  seq->ToJson(doc, &event_sequence_json);

  EXPECT_EQ(0, event_sequence_json["offset"].GetInt());

  uint64_t last_timestamp = 0;
  string last_string = "";
  for (auto& itr : event_sequence_json["events"].GetArray()) {
    EXPECT_TRUE(itr["timestamp"].GetInt() >= last_timestamp);
    last_timestamp = itr["timestamp"].GetInt();
    string label = string(itr["label"].GetString());
    EXPECT_TRUE(label > last_string);
    last_string = label;
  }
}

TEST(ToJson, TimeSeriesCounterToJsonTest) {
  ObjectPool pool;
  RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");

  // 1. TimeSeriesCounter should be empty
  rapidjson::Document doc(rapidjson::kObjectType);
  profile->ToJson(&doc);
  EXPECT_TRUE(!doc["contents"].HasMember("time_series_counters"));

  // 2. Check Serialize to json
  const int test_period = FLAGS_periodic_counter_update_period_ms;

  // Add a counter with a sample function that counts up, starting from 0.
  int value = 0;
  auto sample_fn = [&value]() { return value++; };

  // We increase the value of this flag to allow the counter to store enough samples.
  FLAGS_status_report_interval_ms = 50000;
  RuntimeProfile::TimeSeriesCounter* counter =
      profile->AddChunkedTimeSeriesCounter("TimeSeriesCounter", TUnit::UNIT, sample_fn);
  RuntimeProfile::TimeSeriesCounter* counter2 =
      profile->AddSamplingTimeSeriesCounter("SamplingCounter", TUnit::UNIT, sample_fn);

  // Stop counter updates from interfering with the rest of the test.
  StopAndClearCounter(profile, counter);

  // Reset value after previous values have been retrieved.
  value = 0;
  for (int i = 0; i < 64; ++i) counter->AddSample(test_period);

  value = 0;
  for (int i = 0; i < 80; ++i) counter2->AddSample(test_period);

  profile->ToJson(&doc);
  EXPECT_STR_CONTAINS(
      doc["contents"]["time_series_counters"][1]["data"].GetString(), "0,1,2,3,4");

  EXPECT_STR_CONTAINS(
      doc["contents"]["time_series_counters"][1]["data"].GetString(), "60,61,62,63");

  EXPECT_STR_CONTAINS(
      doc["contents"]["time_series_counters"][0]["data"].GetString(), "0,2,4,6");

  EXPECT_STR_CONTAINS(
      doc["contents"]["time_series_counters"][0]["data"].GetString(), "72,74,76,78");
}

} // namespace impala