diff --git a/docs/configuration/http_filters/dynamodb_filter.rst b/docs/configuration/http_filters/dynamodb_filter.rst
index 99cbe54863b34..ab16d93cfe7de 100644
--- a/docs/configuration/http_filters/dynamodb_filter.rst
+++ b/docs/configuration/http_filters/dynamodb_filter.rst
@@ -54,6 +54,17 @@ in all operations from the batch.
     upstream_rq_total_xxx, Counter, Total number of requests on <table_name> table per response code (503/2xx/etc)
     upstream_rq_time_xxx, Timer, Time spent on <table_name> table per response code (400/3xx/etc)
 
+*Disclaimer: Please note that this is a pre-release Amazon DynamoDB feature that is not yet widely available.*
+Per partition and operation stats can be found in the *http.<stat_prefix>.dynamodb.table.<table_name>.*
+namespace. For batch operations, Envoy tracks per partition and operation stats only if it is the same
+table used in all operations.
+
+  .. csv-table::
+    :header: Name, Type, Description
+    :widths: 1, 1, 2
+
+    capacity.<operation_name>.__partition_id=<last_seven_characters_from_partition_id>, Counter, Total number of capacity for <operation_name> on <table_name> table for a given <partition_id>
+
 Additional detailed stats:
 
 * For 4xx responses and partial batch operation failures, the total number of failures for a given
diff --git a/docs/intro/arch_overview/dynamo.rst b/docs/intro/arch_overview/dynamo.rst
index ed072acdd52d8..d757fe5aa42df 100644
--- a/docs/intro/arch_overview/dynamo.rst
+++ b/docs/intro/arch_overview/dynamo.rst
@@ -6,7 +6,7 @@ DynamoDB
 Envoy supports an HTTP level DynamoDB sniffing filter with the following features:
 
 * DynamoDB API request/response parser.
-* DynamoDB per operation/per table statistics.
+* DynamoDB per operation/per table/per partition and operation statistics.
 * Failure type statistics for 4xx responses, parsed from response JSON,
   e.g., ProvisionedThroughputExceededException.
 * Batch operation partial failure statistics.
diff --git a/source/common/CMakeLists.txt b/source/common/CMakeLists.txt
index f79ab09cb1043..3bc7775129601 100644
--- a/source/common/CMakeLists.txt
+++ b/source/common/CMakeLists.txt
@@ -31,6 +31,7 @@ add_library(
   common/version.cc
   dynamo/dynamo_filter.cc
   dynamo/dynamo_request_parser.cc
+  dynamo/dynamo_utility.cc
   event/dispatcher_impl.cc
   event/event_impl_base.cc
   event/file_event_impl.cc
diff --git a/source/common/dynamo/dynamo_filter.cc b/source/common/dynamo/dynamo_filter.cc
index fa871d5aba104..5a9ad5f0a79ed 100644
--- a/source/common/dynamo/dynamo_filter.cc
+++ b/source/common/dynamo/dynamo_filter.cc
@@ -1,4 +1,5 @@
 #include "dynamo_filter.h"
+#include "dynamo_utility.h"
 
 #include "common/buffer/buffer_impl.h"
 #include "common/dynamo/dynamo_request_parser.h"
@@ -12,7 +13,7 @@ namespace Dynamo {
 Http::FilterHeadersStatus DynamoFilter::decodeHeaders(Http::HeaderMap& headers, bool) {
   if (enabled_) {
     start_decode_ = std::chrono::system_clock::now();
-    operation_ = Dynamo::RequestParser::parseOperation(headers);
+    operation_ = RequestParser::parseOperation(headers);
   }
 
   return Http::FilterHeadersStatus::Continue;
@@ -43,7 +44,8 @@ void DynamoFilter::onDecodeComplete(const Buffer::Instance& data) {
   std::string body = buildBody(decoder_callbacks_->decodingBuffer(), data);
   if (!body.empty()) {
     try {
-      table_descriptor_ = Dynamo::RequestParser::parseTable(operation_, body);
+      Json::StringLoader json_body(body);
+      table_descriptor_ = RequestParser::parseTable(operation_, json_body);
     } catch (const Json::Exception& jsonEx) {
       // Body parsing failed. This should not happen, just put a stat for that.
       stats_.counter(fmt::format("{}invalid_req_body", stat_prefix_)).inc();
@@ -52,19 +54,31 @@ void DynamoFilter::onDecodeComplete(const Buffer::Instance& data) {
 }
 
 void DynamoFilter::onEncodeComplete(const Buffer::Instance& data) {
-  if (response_headers_) {
-    uint64_t status = Http::Utility::getResponseStatus(*response_headers_);
+  if (!response_headers_) {
+    return;
+  }
 
-    chargeBasicStats(status);
+  uint64_t status = Http::Utility::getResponseStatus(*response_headers_);
+  chargeBasicStats(status);
 
-    if (Http::CodeUtility::is4xx(status)) {
-      chargeFailureSpecificStats(data);
-    }
-    // Batch Operations will always return status 200 for a partial or full success. Check
-    // unprocessed keys to determine partial success.
-    // http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Programming.Errors.html#Programming.Errors.BatchOperations
-    if (Dynamo::RequestParser::isBatchOperation(operation_)) {
-      chargeUnProcessedKeysStats(data);
+  std::string body = buildBody(encoder_callbacks_->encodingBuffer(), data);
+  if (!body.empty()) {
+    try {
+      Json::StringLoader json_body(body);
+      chargeTablePartitionIdStats(json_body);
+
+      if (Http::CodeUtility::is4xx(status)) {
+        chargeFailureSpecificStats(json_body);
+      }
+      // Batch Operations will always return status 200 for a partial or full success. Check
+      // unprocessed keys to determine partial success.
+      // http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Programming.Errors.html#Programming.Errors.BatchOperations
+      if (RequestParser::isBatchOperation(operation_)) {
+        chargeUnProcessedKeysStats(json_body);
+      }
+    } catch (const Json::Exception&) {
+      // Body parsing failed. This should not happen, just put a stat for that.
+      stats_.counter(fmt::format("{}invalid_resp_body", stat_prefix_)).inc();
     }
   }
 }
@@ -165,47 +179,43 @@ void DynamoFilter::chargeStatsPerEntity(const std::string& entity, const std::st
                               latency);
 }
 
-void DynamoFilter::chargeUnProcessedKeysStats(const Buffer::Instance& data) {
-  std::string body = buildBody(encoder_callbacks_->encodingBuffer(), data);
-  if (!body.empty()) {
-    try {
-      // The unprocessed keys block contains a list of tables and keys for that table that did not
-      // complete apart of the batch operation. Only the table names will be logged for errors.
-      std::vector<std::string> unprocessed_tables =
-          Dynamo::RequestParser::parseBatchUnProcessedKeys(body);
-      for (const std::string& unprocessed_table : unprocessed_tables) {
-        stats_.counter(fmt::format("{}error.{}.BatchFailureUnprocessedKeys", stat_prefix_,
-                                   unprocessed_table)).inc();
-      }
-    } catch (const Json::Exception& jsonEx) {
-      // Body parsing failed. This should not happen, just put a stat for that.
-      stats_.counter(fmt::format("{}invalid_resp_body", stat_prefix_)).inc();
-    }
+void DynamoFilter::chargeUnProcessedKeysStats(const Json::Object& json_body) {
+  // The unprocessed keys block contains a list of tables and keys for that table that did not
+  // complete apart of the batch operation. Only the table names will be logged for errors.
+  std::vector<std::string> unprocessed_tables = RequestParser::parseBatchUnProcessedKeys(json_body);
+  for (const std::string& unprocessed_table : unprocessed_tables) {
+    stats_.counter(fmt::format("{}error.{}.BatchFailureUnprocessedKeys", stat_prefix_,
+                               unprocessed_table)).inc();
   }
 }
 
-void DynamoFilter::chargeFailureSpecificStats(const Buffer::Instance& data) {
-  std::string body = buildBody(encoder_callbacks_->encodingBuffer(), data);
+void DynamoFilter::chargeFailureSpecificStats(const Json::Object& json_body) {
+  std::string error_type = RequestParser::parseErrorType(json_body);
 
-  if (!body.empty()) {
-    try {
-      std::string error_type = Dynamo::RequestParser::parseErrorType(body);
-
-      if (!error_type.empty()) {
-        if (table_descriptor_.table_name.empty()) {
-          stats_.counter(fmt::format("{}error.no_table.{}", stat_prefix_, error_type)).inc();
-        } else {
-          stats_.counter(fmt::format("{}error.{}.{}", stat_prefix_, table_descriptor_.table_name,
-                                     error_type)).inc();
-        }
-      }
-    } catch (const Json::Exception& jsonEx) {
-      // Body parsing failed. This should not happen, just put a stat for that.
-      stats_.counter(fmt::format("{}invalid_resp_body", stat_prefix_)).inc();
+  if (!error_type.empty()) {
+    if (table_descriptor_.table_name.empty()) {
+      stats_.counter(fmt::format("{}error.no_table.{}", stat_prefix_, error_type)).inc();
+    } else {
+      stats_.counter(fmt::format("{}error.{}.{}", stat_prefix_, table_descriptor_.table_name,
+                                 error_type)).inc();
     }
   } else {
     stats_.counter(fmt::format("{}empty_response_body", stat_prefix_)).inc();
   }
 }
 
+void DynamoFilter::chargeTablePartitionIdStats(const Json::Object& json_body) {
+  if (table_descriptor_.table_name.empty() || operation_.empty()) {
+    return;
+  }
+
+  std::vector<RequestParser::PartitionDescriptor> partitions =
+      RequestParser::parsePartitions(json_body);
+  for (const RequestParser::PartitionDescriptor& partition : partitions) {
+    std::string stats_string = Utility::buildPartitionStatString(
+        stat_prefix_, table_descriptor_.table_name, operation_, partition.partition_id_);
+    stats_.counter(stats_string).add(partition.capacity_);
+  }
+}
+
 } // Dynamo
diff --git a/source/common/dynamo/dynamo_filter.h b/source/common/dynamo/dynamo_filter.h
index 2decba7e01834..10ce104dedde5 100644
--- a/source/common/dynamo/dynamo_filter.h
+++ b/source/common/dynamo/dynamo_filter.h
@@ -6,6 +6,8 @@
 #include "envoy/runtime/runtime.h"
 #include "envoy/stats/stats.h"
 
+#include "common/json/json_loader.h"
+
 namespace Dynamo {
 
 /**
@@ -44,8 +46,9 @@ class DynamoFilter : public Http::StreamFilter {
   void chargeBasicStats(uint64_t status);
   void chargeStatsPerEntity(const std::string& entity, const std::string& entity_type,
                             uint64_t status);
-  void chargeFailureSpecificStats(const Buffer::Instance& data);
-  void chargeUnProcessedKeysStats(const Buffer::Instance& data);
+  void chargeFailureSpecificStats(const Json::Object& json_body);
+  void chargeUnProcessedKeysStats(const Json::Object& json_body);
+  void chargeTablePartitionIdStats(const Json::Object& json_body);
 
   Runtime::Loader& runtime_;
   std::string stat_prefix_;
diff --git a/source/common/dynamo/dynamo_request_parser.cc b/source/common/dynamo/dynamo_request_parser.cc
index 40b55f3c857de..5619a35383497 100644
--- a/source/common/dynamo/dynamo_request_parser.cc
+++ b/source/common/dynamo/dynamo_request_parser.cc
@@ -1,7 +1,6 @@
 #include "dynamo_request_parser.h"
 
 #include "common/common/utility.h"
-#include "common/json/json_loader.h"
 
 namespace Dynamo {
 
@@ -60,60 +59,51 @@ std::string RequestParser::parseOperation(const Http::HeaderMap& headerMap) {
 }
 
 RequestParser::TableDescriptor RequestParser::parseTable(const std::string& operation,
-                                                         const std::string& data) {
+                                                         const Json::Object& json_data) {
   TableDescriptor table{"", true};
 
   // Simple operations on a single table, have "TableName" explicitly specified.
   if (find(SINGLE_TABLE_OPERATIONS.begin(), SINGLE_TABLE_OPERATIONS.end(), operation) !=
       SINGLE_TABLE_OPERATIONS.end()) {
-    Json::StringLoader json(data);
-    if (json.hasObject("TableName")) {
-      table.table_name = json.getString("TableName");
-    }
+    table.table_name = json_data.getString("TableName", "");
   } else if (find(BATCH_OPERATIONS.begin(), BATCH_OPERATIONS.end(), operation) !=
              BATCH_OPERATIONS.end()) {
-    Json::StringLoader json(data);
-    if (json.hasObject("RequestItems")) {
-      Json::Object tables = json.getObject("RequestItems");
-      tables.iterate([&table](const std::string& key, const Json::Object&) {
-        if (table.table_name.empty()) {
-          table.table_name = key;
-        } else {
-          if (table.table_name != key) {
-            table.table_name = "";
-            table.is_single_table = false;
-            return false;
-          }
+    Json::Object tables = json_data.getObject("RequestItems", true);
+    tables.iterate([&table](const std::string& key, const Json::Object&) {
+      if (table.table_name.empty()) {
+        table.table_name = key;
+      } else {
+        if (table.table_name != key) {
+          table.table_name = "";
+          table.is_single_table = false;
+          return false;
         }
-
-        return true;
-      });
-    }
+      }
+      return true;
+    });
   }
 
   return table;
 }
-std::vector<std::string> RequestParser::parseBatchUnProcessedKeys(const std::string& data) {
+std::vector<std::string> RequestParser::parseBatchUnProcessedKeys(const Json::Object& json_data) {
   std::vector<std::string> unprocessed_tables;
-  Json::StringLoader json(data);
-  if (json.hasObject("UnprocessedKeys")) {
-    Json::Object tables = json.getObject("UnprocessedKeys");
-    tables.iterate([&unprocessed_tables](const std::string& key, const Json::Object&) {
-      unprocessed_tables.emplace_back(key);
-      return true;
-    });
-  }
+  Json::Object tables = json_data.getObject("UnprocessedKeys", true);
+  tables.iterate([&unprocessed_tables](const std::string& key, const Json::Object&) {
+    unprocessed_tables.emplace_back(key);
+    return true;
+  });
+
   return unprocessed_tables;
 }
-std::string RequestParser::parseErrorType(const std::string& data) {
-  Json::StringLoader json(data);
-
-  if (json.hasObject("__type")) {
-    std::string error_type = json.getString("__type");
-    for (const std::string& supported_error_type : SUPPORTED_ERROR_TYPES) {
-      if (StringUtil::endsWith(error_type, supported_error_type)) {
-        return supported_error_type;
-      }
+std::string RequestParser::parseErrorType(const Json::Object& json_data) {
+  std::string error_type = json_data.getString("__type", "");
+  if (error_type.empty()) {
+    return "";
+  }
+
+  for (const std::string& supported_error_type : SUPPORTED_ERROR_TYPES) {
+    if (StringUtil::endsWith(error_type, supported_error_type)) {
+      return supported_error_type;
     }
   }
 
@@ -125,4 +115,24 @@ bool RequestParser::isBatchOperation(const std::string& operation) {
          BATCH_OPERATIONS.end();
 }
 
+std::vector<RequestParser::PartitionDescriptor>
+RequestParser::parsePartitions(const Json::Object& json_data) {
+  std::vector<RequestParser::PartitionDescriptor> partition_descriptors;
+
+  Json::Object partitions =
+      json_data.getObject("ConsumedCapacity", true).getObject("Partitions", true);
+  partitions.iterate([&partition_descriptors, &partitions](const std::string& key,
+                                                           const Json::Object&) {
+    // For a given partition id, the amount of capacity used is returned in the body as a double.
+    // A stat will be created to track the capacity consumed for the operation, table and partition.
+    // Stats counter only increments by whole numbers, capacity is round up to the nearest integer
+    // to account for this.
+    uint64_t capacity_integer = static_cast<uint64_t>(std::ceil(partitions.getDouble(key, 0.0)));
+    partition_descriptors.emplace_back(key, capacity_integer);
+    return true;
+  });
+
+  return partition_descriptors;
+}
+
 } // Dynamo
\ No newline at end of file
diff --git a/source/common/dynamo/dynamo_request_parser.h b/source/common/dynamo/dynamo_request_parser.h
index 8d8fe9f4cadf7..8505f9fcae251 100644
--- a/source/common/dynamo/dynamo_request_parser.h
+++ b/source/common/dynamo/dynamo_request_parser.h
@@ -2,6 +2,8 @@
 
 #include "envoy/http/header_map.h"
 
+#include "common/json/json_loader.h"
+
 namespace Dynamo {
 
 /*
@@ -18,6 +20,13 @@ class RequestParser {
     bool is_single_table;
   };
 
+  struct PartitionDescriptor {
+    PartitionDescriptor(const std::string& partition, uint64_t capacity)
+        : partition_id_(partition), capacity_(capacity) {}
+    std::string partition_id_;
+    uint64_t capacity_;
+  };
+
   /**
    * Parse operation out of x-amz-target header.
    * @return empty string if operation cannot be parsed.
@@ -40,7 +49,7 @@ class RequestParser {
    *
    * @throw Json::Exception if data is not in valid Json format.
    */
-  static TableDescriptor parseTable(const std::string& operation, const std::string& data);
+  static TableDescriptor parseTable(const std::string& operation, const Json::Object& json_data);
 
   /**
    * Parse error details which might be provided for a given response code.
@@ -52,20 +61,29 @@ class RequestParser {
    *
    * @throw Json::Exception if data is not in valid Json format.
    */
-  static std::string parseErrorType(const std::string& data);
+  static std::string parseErrorType(const Json::Object& json_data);
 
   /**
    * Parse unprocessed keys for batch operation results.
    * @return empty set if there are no unprocessed keys or a set of table names that did not get
    * processed in the batch operation.
    */
-  static std::vector<std::string> parseBatchUnProcessedKeys(const std::string& data);
+  static std::vector<std::string> parseBatchUnProcessedKeys(const Json::Object& json_data);
 
   /**
    * @return true if the operation is in the set of supported BATCH_OPERATIONS
    */
   static bool isBatchOperation(const std::string& operation);
 
+  /**
+   * Parse the Partition ids and the consumed capacity from the body.
+   * @return empty set if there is no partition data or a set of partition data containing
+   * the partition id as a string and the capacity consumed as an integer.
+   *
+   * @throw Json::Exception if data is not in valid Json format.
+   */
+  static std::vector<PartitionDescriptor> parsePartitions(const Json::Object& json_data);
+
 private:
   static const Http::LowerCaseString X_AMZ_TARGET;
   static const std::vector<std::string> SINGLE_TABLE_OPERATIONS;
diff --git a/source/common/dynamo/dynamo_utility.cc b/source/common/dynamo/dynamo_utility.cc
new file mode 100644
index 0000000000000..aed628de938eb
--- /dev/null
+++ b/source/common/dynamo/dynamo_utility.cc
@@ -0,0 +1,27 @@
+#include "dynamo_utility.h"
+
+#include "common/stats/stats_impl.h"
+
+namespace Dynamo {
+
+std::string Utility::buildPartitionStatString(const std::string& stat_prefix,
+                                              const std::string& table_name,
+                                              const std::string& operation,
+                                              const std::string& partition_id) {
+  // Use the last 7 characters of the partition id.
+  std::string stats_partition_postfix =
+      fmt::format(".capacity.{}.__partition_id={}", operation,
+                  partition_id.substr(partition_id.size() - 7, partition_id.size()));
+
+  // Calculate how many characters are available for the table prefix.
+  size_t remaining_size = Stats::RawStatData::MAX_NAME_SIZE - stats_partition_postfix.size();
+
+  std::string stats_table_prefix = fmt::format("{}table.{}", stat_prefix, table_name);
+  // Truncate the table prefix if the current string is too large.
+  if (stats_table_prefix.size() > remaining_size) {
+    stats_table_prefix = stats_table_prefix.substr(0, remaining_size);
+  }
+  return fmt::format("{}{}", stats_table_prefix, stats_partition_postfix);
+}
+
+} // Dynamo
diff --git a/source/common/dynamo/dynamo_utility.h b/source/common/dynamo/dynamo_utility.h
new file mode 100644
index 0000000000000..b0c159cfb2c7a
--- /dev/null
+++ b/source/common/dynamo/dynamo_utility.h
@@ -0,0 +1,24 @@
+#pragma once
+
+namespace Dynamo {
+
+class Utility {
+public:
+  /*
+   * Creates the partition id stats string.
+   * The stats format is
+   * "<stat_prefix>table.<table_name>.capacity.<operation>.__partition_id=<partition_id>".
+   * Partition ids and dynamodb table names can be long. To satisfy the string length,
+   * we truncate in two ways:
+   * 1. We only take the last 7 characters of the partition id.
+   * 2. If the stats string with <table_name> is longer than the stats MAX_NAME_SIZE, we will
+   * truncate the table name to
+   * fit the size requirements.
+   */
+  static std::string buildPartitionStatString(const std::string& stat_prefix,
+                                              const std::string& table_name,
+                                              const std::string& operation,
+                                              const std::string& partition_id);
+};
+
+} // Dynamo
diff --git a/source/common/json/json_loader.cc b/source/common/json/json_loader.cc
index c5c153e1f25d1..cb7ae9fe8bb73 100644
--- a/source/common/json/json_loader.cc
+++ b/source/common/json/json_loader.cc
@@ -136,6 +136,23 @@ std::vector<std::string> Object::getStringArray(const std::string& name) const {
   return string_array;
 }
 
+double Object::getDouble(const std::string& name) const {
+  json_t* real = json_object_get(json_, name.c_str());
+  if (!real || !json_is_real(real)) {
+    throw Exception(fmt::format("key '{}' missing or not a double in '{}'", name, name_));
+  }
+
+  return json_real_value(real);
+}
+
+double Object::getDouble(const std::string& name, double default_value) const {
+  if (!json_object_get(json_, name.c_str())) {
+    return default_value;
+  } else {
+    return getDouble(name);
+  }
+}
+
 void Object::iterate(const ObjectCallback& callback) {
   const char* key;
   json_t* value;
diff --git a/source/common/json/json_loader.h b/source/common/json/json_loader.h
index aa10ad08b8ba7..95d435f2aa757 100644
--- a/source/common/json/json_loader.h
+++ b/source/common/json/json_loader.h
@@ -81,14 +81,14 @@ class Object {
 
   /**
    * Get a string value by name.
-   * @param name suppies the key name.
+   * @param name supplies the key name.
    * @return std::string the value.
    */
   std::string getString(const std::string& name) const;
 
   /**
    * Get a string value by name or return a default if name does not exist.
-   * @param name suppies the key name.
+   * @param name supplies the key name.
    * @param default_value supplies the value to return if name does not exist.
    * @return std::string the value.
    */
@@ -101,6 +101,21 @@ class Object {
    */
   std::vector<std::string> getStringArray(const std::string& name) const;
 
+  /**
+   * Get a double value by name.
+   * @param name supplies the key name.
+   * @return double the value.
+   */
+  double getDouble(const std::string& name) const;
+
+  /**
+   * Get a double value by name.
+   * @param name supplies the key name.
+   * @param default_value supplies the value to return if name does not exist.
+   * @return double the value.
+   */
+  double getDouble(const std::string& name, double default_value) const;
+
   /**
    * Iterate Object and call callback on key-value pairs
    */
diff --git a/source/common/stats/stats_impl.cc b/source/common/stats/stats_impl.cc
index ab2cae6e578ca..902b696ec7a60 100644
--- a/source/common/stats/stats_impl.cc
+++ b/source/common/stats/stats_impl.cc
@@ -16,7 +16,7 @@ RawStatData* HeapRawStatDataAllocator::alloc(const std::string& name) {
 
 void RawStatData::initialize(const std::string& name) {
   ASSERT(!initialized());
-  ASSERT(name.size() < MAX_NAME_SIZE);
+  ASSERT(name.size() <= MAX_NAME_SIZE);
   strncpy(name_, name.substr(0, MAX_NAME_SIZE).c_str(), MAX_NAME_SIZE + 1);
 }
 
diff --git a/source/precompiled/precompiled.h b/source/precompiled/precompiled.h
index 9e205bd853559..3a4457f78482a 100644
--- a/source/precompiled/precompiled.h
+++ b/source/precompiled/precompiled.h
@@ -2,6 +2,7 @@
 
 #include <arpa/inet.h>
 #include <chrono>
+#include <cmath>
 #include <condition_variable>
 #include <fstream>
 #include <iostream>
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index b543ed1376a10..e15cbd642d464 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -58,6 +58,7 @@ add_executable(envoy-test
   common/filter/tcp_proxy_test.cc
   common/dynamo/dynamo_filter_test.cc
   common/dynamo/dynamo_request_parser_test.cc
+  common/dynamo/dynamo_utility_test.cc
   common/json/json_loader_test.cc
   common/mongo/bson_impl_test.cc
   common/mongo/codec_impl_test.cc
diff --git a/test/common/dynamo/dynamo_filter_test.cc b/test/common/dynamo/dynamo_filter_test.cc
index 309b8f285d4a7..f22e0459bb366 100644
--- a/test/common/dynamo/dynamo_filter_test.cc
+++ b/test/common/dynamo/dynamo_filter_test.cc
@@ -410,4 +410,206 @@ TEST_F(DynamoFilterTest, operatorPresentRuntimeDisabled) {
   EXPECT_EQ(Http::FilterTrailersStatus::Continue, filter_->encodeTrailers(response_headers));
 }
 
+TEST_F(DynamoFilterTest, PartitionIdStats) {
+  setup(true);
+
+  Http::HeaderMapImpl request_headers{{"x-amz-target", "version.GetItem"}};
+  Buffer::OwnedImpl buffer;
+  std::string buffer_content = "{\"TableName\":\"locations\"";
+  buffer.add(buffer_content);
+  EXPECT_CALL(decoder_callbacks_, decodingBuffer()).WillRepeatedly(Return(&buffer));
+  Buffer::OwnedImpl data;
+  data.add("}", 1);
+
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_->decodeHeaders(request_headers, false));
+  EXPECT_EQ(Http::FilterDataStatus::StopIterationAndBuffer, filter_->decodeData(data, false));
+  EXPECT_EQ(Http::FilterDataStatus::Continue, filter_->decodeData(data, true));
+
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.operation.GetItem.upstream_rq_total_2xx"));
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.operation.GetItem.upstream_rq_total_200"));
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.operation.GetItem.upstream_rq_total"));
+
+  EXPECT_CALL(stats_,
+              deliverTimingToSinks("prefix.dynamodb.operation.GetItem.upstream_rq_time_2xx", _));
+  EXPECT_CALL(stats_,
+              deliverTimingToSinks("prefix.dynamodb.operation.GetItem.upstream_rq_time_200", _));
+  EXPECT_CALL(stats_,
+              deliverTimingToSinks("prefix.dynamodb.operation.GetItem.upstream_rq_time", _));
+
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.table.locations.upstream_rq_total_2xx"));
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.table.locations.upstream_rq_total_200"));
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.table.locations.upstream_rq_total"));
+
+  EXPECT_CALL(stats_,
+              deliverTimingToSinks("prefix.dynamodb.table.locations.upstream_rq_time_2xx", _));
+  EXPECT_CALL(stats_,
+              deliverTimingToSinks("prefix.dynamodb.table.locations.upstream_rq_time_200", _));
+  EXPECT_CALL(stats_, deliverTimingToSinks("prefix.dynamodb.table.locations.upstream_rq_time", _));
+
+  EXPECT_CALL(stats_,
+              counter("prefix.dynamodb.table.locations.capacity.GetItem.__partition_id=ition_1"))
+      .Times(1);
+  EXPECT_CALL(stats_,
+              counter("prefix.dynamodb.table.locations.capacity.GetItem.__partition_id=ition_2"))
+      .Times(1);
+
+  Http::HeaderMapImpl response_headers{{":status", "200"}};
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_->encodeHeaders(response_headers, false));
+
+  Buffer::OwnedImpl empty_data;
+  Buffer::OwnedImpl response_data;
+  std::string response_content = R"EOF(
+    {
+      "ConsumedCapacity": {
+        "Partitions": {
+          "partition_1" : 0.5,
+          "partition_2" : 3.0
+        }
+      }
+    }
+    )EOF";
+
+  response_data.add(response_content);
+
+  EXPECT_CALL(encoder_callbacks_, encodingBuffer()).Times(1).WillRepeatedly(Return(&response_data));
+  EXPECT_EQ(Http::FilterDataStatus::Continue, filter_->encodeData(empty_data, true));
+}
+
+TEST_F(DynamoFilterTest, NoPartitionIdStatsForMultipleTables) {
+  setup(true);
+
+  Http::HeaderMapImpl request_headers{{"x-amz-target", "version.BatchGetItem"}};
+  Buffer::OwnedImpl buffer;
+  std::string buffer_content = R"EOF(
+{
+  "RequestItems": {
+    "table_1": { "test1" : "something" },
+    "table_2": { "test2" : "something" }
+  }
+}
+)EOF";
+  buffer.add(buffer_content);
+  EXPECT_CALL(decoder_callbacks_, decodingBuffer()).WillRepeatedly(Return(&buffer));
+
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_->decodeHeaders(request_headers, false));
+  EXPECT_EQ(Http::FilterDataStatus::StopIterationAndBuffer, filter_->decodeData(buffer, false));
+  EXPECT_EQ(Http::FilterTrailersStatus::Continue, filter_->decodeTrailers(request_headers));
+
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.multiple_tables"));
+
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.operation.BatchGetItem.upstream_rq_total"));
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.operation.BatchGetItem.upstream_rq_total_2xx"));
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.operation.BatchGetItem.upstream_rq_total_200"));
+
+  EXPECT_CALL(stats_, deliverTimingToSinks(
+                          "prefix.dynamodb.operation.BatchGetItem.upstream_rq_time_2xx", _));
+  EXPECT_CALL(stats_, deliverTimingToSinks(
+                          "prefix.dynamodb.operation.BatchGetItem.upstream_rq_time_200", _));
+  EXPECT_CALL(stats_,
+              deliverTimingToSinks("prefix.dynamodb.operation.BatchGetItem.upstream_rq_time", _));
+
+  EXPECT_CALL(
+      stats_,
+      counter("prefix.dynamodb.table.locations.capacity.BatchGetItem.__partition_id=ition_1"))
+      .Times(0);
+  EXPECT_CALL(
+      stats_,
+      counter("prefix.dynamodb.table.locations.capacity.BatchGetItem.__partition_id=ition_2"))
+      .Times(0);
+
+  Http::HeaderMapImpl response_headers{{":status", "200"}};
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_->encodeHeaders(response_headers, false));
+
+  Buffer::OwnedImpl empty_data;
+  Buffer::OwnedImpl response_data;
+  std::string response_content = R"EOF(
+    {
+      "ConsumedCapacity": {
+        "Partitions": {
+          "partition_1" : 0.5,
+          "partition_2" : 3.0
+        }
+      }
+    }
+    )EOF";
+
+  response_data.add(response_content);
+
+  EXPECT_CALL(encoder_callbacks_, encodingBuffer()).Times(1).WillRepeatedly(Return(&response_data));
+  EXPECT_EQ(Http::FilterDataStatus::Continue, filter_->encodeData(empty_data, true));
+}
+
+TEST_F(DynamoFilterTest, PartitionIdStatsForSingleTableBatchOperation) {
+  setup(true);
+
+  Http::HeaderMapImpl request_headers{{"x-amz-target", "version.BatchGetItem"}};
+  Buffer::OwnedImpl buffer;
+  std::string buffer_content = R"EOF(
+{
+  "RequestItems": {
+    "locations": { "test1" : "something" }
+  }
+}
+)EOF";
+  buffer.add(buffer_content);
+  EXPECT_CALL(decoder_callbacks_, decodingBuffer()).WillRepeatedly(Return(&buffer));
+
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_->decodeHeaders(request_headers, false));
+  EXPECT_EQ(Http::FilterDataStatus::StopIterationAndBuffer, filter_->decodeData(buffer, false));
+  EXPECT_EQ(Http::FilterTrailersStatus::Continue, filter_->decodeTrailers(request_headers));
+
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.multiple_tables")).Times(0);
+
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.operation.BatchGetItem.upstream_rq_total"));
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.operation.BatchGetItem.upstream_rq_total_2xx"));
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.operation.BatchGetItem.upstream_rq_total_200"));
+
+  EXPECT_CALL(stats_, deliverTimingToSinks(
+                          "prefix.dynamodb.operation.BatchGetItem.upstream_rq_time_2xx", _));
+  EXPECT_CALL(stats_, deliverTimingToSinks(
+                          "prefix.dynamodb.operation.BatchGetItem.upstream_rq_time_200", _));
+  EXPECT_CALL(stats_,
+              deliverTimingToSinks("prefix.dynamodb.operation.BatchGetItem.upstream_rq_time", _));
+
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.table.locations.upstream_rq_total_2xx"));
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.table.locations.upstream_rq_total_200"));
+  EXPECT_CALL(stats_, counter("prefix.dynamodb.table.locations.upstream_rq_total"));
+
+  EXPECT_CALL(stats_,
+              deliverTimingToSinks("prefix.dynamodb.table.locations.upstream_rq_time_2xx", _));
+  EXPECT_CALL(stats_,
+              deliverTimingToSinks("prefix.dynamodb.table.locations.upstream_rq_time_200", _));
+  EXPECT_CALL(stats_, deliverTimingToSinks("prefix.dynamodb.table.locations.upstream_rq_time", _));
+
+  EXPECT_CALL(
+      stats_,
+      counter("prefix.dynamodb.table.locations.capacity.BatchGetItem.__partition_id=ition_1"))
+      .Times(1);
+  EXPECT_CALL(
+      stats_,
+      counter("prefix.dynamodb.table.locations.capacity.BatchGetItem.__partition_id=ition_2"))
+      .Times(1);
+
+  Http::HeaderMapImpl response_headers{{":status", "200"}};
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_->encodeHeaders(response_headers, false));
+
+  Buffer::OwnedImpl empty_data;
+  Buffer::OwnedImpl response_data;
+  std::string response_content = R"EOF(
+    {
+      "ConsumedCapacity": {
+        "Partitions": {
+          "partition_1" : 0.5,
+          "partition_2" : 3.0
+        }
+      }
+    }
+    )EOF";
+
+  response_data.add(response_content);
+
+  EXPECT_CALL(encoder_callbacks_, encodingBuffer()).Times(1).WillRepeatedly(Return(&response_data));
+  EXPECT_EQ(Http::FilterDataStatus::Continue, filter_->encodeData(empty_data, true));
+}
+
 } // Dynamo
diff --git a/test/common/dynamo/dynamo_request_parser_test.cc b/test/common/dynamo/dynamo_request_parser_test.cc
index 254b849b1e6c7..b7bfd75855553 100644
--- a/test/common/dynamo/dynamo_request_parser_test.cc
+++ b/test/common/dynamo/dynamo_request_parser_test.cc
@@ -35,7 +35,7 @@ TEST(DynamoRequestParser, parseTableNameSingleOperation) {
                                                        "PutItem", "UpdateItem", "DeleteItem"};
 
   {
-    std::string json = R"EOF(
+    std::string json_string = R"EOF(
     {
       "TableName": "Pets",
       "Key": {
@@ -44,72 +44,48 @@ TEST(DynamoRequestParser, parseTableNameSingleOperation) {
       }
     }
     )EOF";
+    Json::StringLoader json_data(json_string);
 
     // Supported operation
     for (const std::string& operation : supported_single_operations) {
-      EXPECT_EQ("Pets", RequestParser::parseTable(operation, json).table_name);
+      EXPECT_EQ("Pets", RequestParser::parseTable(operation, json_data).table_name);
     }
 
     // Not supported operation
-    EXPECT_EQ("", RequestParser::parseTable("NotSupportedOperation", json).table_name);
+    EXPECT_EQ("", RequestParser::parseTable("NotSupportedOperation", json_data).table_name);
   }
 
   {
-    std::string json = "{\"TableName\":\"Pets\"}";
-    EXPECT_EQ("Pets", RequestParser::parseTable("GetItem", json).table_name);
-  }
-
-  // Validate invalid json cases
-  {
-    std::string json = R"EOF(
-    {
-      "TableName": "Pets",
-      "Key": {
-        "AnimalType": "S": "Dog"},
-        "Name": {"S": "Fido"}
-      }
-    }
-    )EOF";
-
-    EXPECT_THROW(RequestParser::parseTable("GetItem", json), Json::Exception);
+    Json::StringLoader json_data("{\"TableName\":\"Pets\"}");
+    EXPECT_EQ("Pets", RequestParser::parseTable("GetItem", json_data).table_name);
   }
 }
 
 TEST(DynamoRequestParser, parseErrorType) {
-  { EXPECT_THROW(RequestParser::parseErrorType("{test"), Json::Exception); }
+  { EXPECT_THROW(RequestParser::parseErrorType(Json::StringLoader("{test")), Json::Exception); }
 
   {
     EXPECT_EQ("ResourceNotFoundException",
-              RequestParser::parseErrorType(
-                  "{\"__type\":\"com.amazonaws.dynamodb.v20120810#ResourceNotFoundException\"}"));
+              RequestParser::parseErrorType(Json::StringLoader(
+                  "{\"__type\":\"com.amazonaws.dynamodb.v20120810#ResourceNotFoundException\"}")));
   }
 
   {
     EXPECT_EQ("ResourceNotFoundException",
-              RequestParser::parseErrorType(
+              RequestParser::parseErrorType(Json::StringLoader(
                   "{\"__type\":\"com.amazonaws.dynamodb.v20120810#ResourceNotFoundException\","
-                  "\"message\":\"Requested resource not found: Table: tablename not found\"}"));
+                  "\"message\":\"Requested resource not found: Table: tablename not found\"}")));
   }
 
-  { EXPECT_EQ("", RequestParser::parseErrorType("{\"__type\":\"UnKnownError\"}")); }
-
   {
-    std::string json = R"EOF(
-    {
-      "not_important": "test",
-      "Key": {
-        "AnimalType": "S": "Dog"},
-        "Name": {"S": "Fido"}
-      }
-    }
-    )EOF";
-    EXPECT_THROW(RequestParser::parseErrorType(json), Json::Exception);
+    EXPECT_EQ("",
+              RequestParser::parseErrorType(Json::StringLoader("{\"__type\":\"UnKnownError\"}")));
   }
 }
 
 TEST(DynamoRequestParser, parseTableNameBatchOperation) {
   {
-    std::string json = R"EOF(
+    std::string json_string = R"EOF(
     {
       "RequestItems": {
         "table_1": { "test1" : "something" },
@@ -117,13 +93,15 @@ TEST(DynamoRequestParser, parseTableNameBatchOperation) {
       }
     }
     )EOF";
-    RequestParser::TableDescriptor table = RequestParser::parseTable("BatchGetItem", json);
+    Json::StringLoader json_data(json_string);
+
+    RequestParser::TableDescriptor table = RequestParser::parseTable("BatchGetItem", json_data);
     EXPECT_EQ("", table.table_name);
     EXPECT_FALSE(table.is_single_table);
   }
 
   {
-    std::string json = R"EOF(
+    std::string json_string = R"EOF(
     {
       "RequestItems": {
         "table_2": { "test1" : "something" },
@@ -131,13 +109,15 @@ TEST(DynamoRequestParser, parseTableNameBatchOperation) {
       }
     }
     )EOF";
-    RequestParser::TableDescriptor table = RequestParser::parseTable("BatchGetItem", json);
+    Json::StringLoader json_data(json_string);
+
+    RequestParser::TableDescriptor table = RequestParser::parseTable("BatchGetItem", json_data);
     EXPECT_EQ("table_2", table.table_name);
     EXPECT_TRUE(table.is_single_table);
   }
 
   {
-    std::string json = R"EOF(
+    std::string json_string = R"EOF(
     {
       "RequestItems": {
         "table_2": { "test1" : "something" },
@@ -146,13 +126,15 @@ TEST(DynamoRequestParser, parseTableNameBatchOperation) {
       }
     }
     )EOF";
-    RequestParser::TableDescriptor table = RequestParser::parseTable("BatchGetItem", json);
+    Json::StringLoader json_data(json_string);
+
+    RequestParser::TableDescriptor table = RequestParser::parseTable("BatchGetItem", json_data);
     EXPECT_EQ("", table.table_name);
     EXPECT_FALSE(table.is_single_table);
   }
 
   {
-    std::string json = R"EOF(
+    std::string json_string = R"EOF(
     {
       "RequestItems": {
         "table_2": { "test1" : "something" },
@@ -160,52 +142,60 @@ TEST(DynamoRequestParser, parseTableNameBatchOperation) {
       }
     }
     )EOF";
-    RequestParser::TableDescriptor table = RequestParser::parseTable("BatchWriteItem", json);
+    Json::StringLoader json_data(json_string);
+
+    RequestParser::TableDescriptor table = RequestParser::parseTable("BatchWriteItem", json_data);
     EXPECT_EQ("table_2", table.table_name);
     EXPECT_TRUE(table.is_single_table);
   }
 
   {
-    RequestParser::TableDescriptor table = RequestParser::parseTable("BatchWriteItem", "{}");
+    RequestParser::TableDescriptor table =
+        RequestParser::parseTable("BatchWriteItem", Json::StringLoader("{}"));
     EXPECT_EQ("", table.table_name);
     EXPECT_TRUE(table.is_single_table);
   }
 
   {
     RequestParser::TableDescriptor table =
-        RequestParser::parseTable("BatchWriteItem", "{\"RequestItems\":{}}");
+        RequestParser::parseTable("BatchWriteItem", Json::StringLoader("{\"RequestItems\":{}}"));
     EXPECT_EQ("", table.table_name);
     EXPECT_TRUE(table.is_single_table);
   }
 
   {
-    RequestParser::TableDescriptor table = RequestParser::parseTable("BatchGetItem", "{}");
+    RequestParser::TableDescriptor table =
+        RequestParser::parseTable("BatchGetItem", Json::StringLoader("{}"));
     EXPECT_EQ("", table.table_name);
     EXPECT_TRUE(table.is_single_table);
   }
 }
 TEST(DynamoRequestParser, parseBatchUnProcessedKeys) {
-  { EXPECT_THROW(RequestParser::parseBatchUnProcessedKeys("{test"), Json::Exception); }
+  {
+    EXPECT_THROW(RequestParser::parseBatchUnProcessedKeys(Json::StringLoader("{test")),
+                 Json::Exception);
+  }
 
   {
-    std::vector<std::string> unprocessed_tables = RequestParser::parseBatchUnProcessedKeys("{}");
+    std::vector<std::string> unprocessed_tables =
+        RequestParser::parseBatchUnProcessedKeys(Json::StringLoader("{}"));
     EXPECT_EQ(0u, unprocessed_tables.size());
   }
   {
     std::vector<std::string> unprocessed_tables =
-        RequestParser::parseBatchUnProcessedKeys("{\"UnprocessedKeys\":{}}");
+        RequestParser::parseBatchUnProcessedKeys(Json::StringLoader("{\"UnprocessedKeys\":{}}"));
     EXPECT_EQ(0u, unprocessed_tables.size());
   }
 
   {
-    std::vector<std::string> unprocessed_tables =
-        RequestParser::parseBatchUnProcessedKeys("{\"UnprocessedKeys\":{\"table_1\" :{}}}");
+    std::vector<std::string> unprocessed_tables = RequestParser::parseBatchUnProcessedKeys(
+        Json::StringLoader("{\"UnprocessedKeys\":{\"table_1\" :{}}}"));
     EXPECT_EQ("table_1", unprocessed_tables[0]);
     EXPECT_EQ(1u, unprocessed_tables.size());
   }
 
   {
-    std::string json = R"EOF(
+    std::string json_string = R"EOF(
     {
       "UnprocessedKeys": {
         "table_1": { "test1" : "something" },
@@ -213,7 +203,10 @@ TEST(DynamoRequestParser, parseBatchUnProcessedKeys) {
       }
     }
     )EOF";
-    std::vector<std::string> unprocessed_tables = RequestParser::parseBatchUnProcessedKeys(json);
+    Json::StringLoader json_data(json_string);
+
+    std::vector<std::string> unprocessed_tables =
+        RequestParser::parseBatchUnProcessedKeys(json_data);
     EXPECT_TRUE(find(unprocessed_tables.begin(), unprocessed_tables.end(), "table_1") !=
                 unprocessed_tables.end());
     EXPECT_TRUE(find(unprocessed_tables.begin(), unprocessed_tables.end(), "table_2") !=
@@ -221,4 +214,47 @@ TEST(DynamoRequestParser, parseBatchUnProcessedKeys) {
     EXPECT_EQ(2u, unprocessed_tables.size());
   }
 }
+
+TEST(DynamoRequestParser, parsePartitionIds) {
+  {
+    std::vector<RequestParser::PartitionDescriptor> partitions =
+        RequestParser::parsePartitions(Json::StringLoader("{}"));
+    EXPECT_EQ(0u, partitions.size());
+  }
+  {
+    std::vector<RequestParser::PartitionDescriptor> partitions =
+        RequestParser::parsePartitions(Json::StringLoader("{\"ConsumedCapacity\":{}}"));
+    EXPECT_EQ(0u, partitions.size());
+  }
+  {
+    std::vector<RequestParser::PartitionDescriptor> partitions = RequestParser::parsePartitions(
+        Json::StringLoader("{\"ConsumedCapacity\":{ \"Partitions\":{}}}"));
+    EXPECT_EQ(0u, partitions.size());
+  }
+  {
+    std::string json_string = R"EOF(
+    {
+      "ConsumedCapacity": {
+        "Partitions": {
+          "partition_1" : 0.5,
+          "partition_2" : 3.0
+        }
+      }
+    }
+    )EOF";
+    Json::StringLoader json_data(json_string);
+
+    std::vector<RequestParser::PartitionDescriptor> partitions =
+        RequestParser::parsePartitions(json_data);
+    for (const RequestParser::PartitionDescriptor& partition : partitions) {
+      if (partition.partition_id_ == "partition_1") {
+        EXPECT_EQ(1u, partition.capacity_);
+      } else {
+        EXPECT_EQ(3u, partition.capacity_);
+      }
+    }
+    EXPECT_EQ(2u, partitions.size());
+  }
+}
+
 } // Dynamo
\ No newline at end of file
diff --git a/test/common/dynamo/dynamo_utility_test.cc b/test/common/dynamo/dynamo_utility_test.cc
new file mode 100644
index 0000000000000..bea71125e5d02
--- /dev/null
+++ b/test/common/dynamo/dynamo_utility_test.cc
@@ -0,0 +1,53 @@
+#include "common/dynamo/dynamo_utility.h"
+#include "common/stats/stats_impl.h"
+
+using testing::_;
+
+namespace Dynamo {
+
+TEST(DynamoUtility, PartitionIdStatString) {
+  {
+    std::string stat_prefix = "stat.prefix.";
+    std::string table_name = "locations";
+    std::string operation = "GetItem";
+    std::string partition_id = "6235c781-1d0d-47a3-a4ea-eec04c5883ca";
+    std::string partition_stat_string =
+        Utility::buildPartitionStatString(stat_prefix, table_name, operation, partition_id);
+    std::string expected_stat_string =
+        "stat.prefix.table.locations.capacity.GetItem.__partition_id=c5883ca";
+    EXPECT_EQ(expected_stat_string, partition_stat_string);
+    EXPECT_TRUE(partition_stat_string.size() <= Stats::RawStatData::MAX_NAME_SIZE);
+  }
+
+  {
+    std::string stat_prefix = "http.egress_dynamodb_iad.dynamodb.";
+    std::string table_name = "locations-sandbox-partition-test-iad-mytest-really-long-name";
+    std::string operation = "GetItem";
+    std::string partition_id = "6235c781-1d0d-47a3-a4ea-eec04c5883ca";
+
+    std::string partition_stat_string =
+        Utility::buildPartitionStatString(stat_prefix, table_name, operation, partition_id);
+    std::string expected_stat_string = "http.egress_dynamodb_iad.dynamodb.table.locations-sandbox-"
+                                       "partition-test-iad-mytest-rea.capacity.GetItem.__partition_"
+                                       "id=c5883ca";
+    EXPECT_EQ(expected_stat_string, partition_stat_string);
+    EXPECT_TRUE(partition_stat_string.size() == Stats::RawStatData::MAX_NAME_SIZE);
+  }
+  {
+    std::string stat_prefix = "http.egress_dynamodb_iad.dynamodb.";
+    std::string table_name = "locations-sandbox-partition-test-iad-mytest-rea";
+    std::string operation = "GetItem";
+    std::string partition_id = "6235c781-1d0d-47a3-a4ea-eec04c5883ca";
+
+    std::string partition_stat_string =
+        Utility::buildPartitionStatString(stat_prefix, table_name, operation, partition_id);
+    std::string expected_stat_string = "http.egress_dynamodb_iad.dynamodb.table.locations-sandbox-"
+                                       "partition-test-iad-mytest-rea.capacity.GetItem.__partition_"
+                                       "id=c5883ca";
+
+    EXPECT_EQ(expected_stat_string, partition_stat_string);
+    EXPECT_TRUE(partition_stat_string.size() == Stats::RawStatData::MAX_NAME_SIZE);
+  }
+}
+
+} // Dynamo
diff --git a/test/common/json/json_loader_test.cc b/test/common/json/json_loader_test.cc
index f625801d724c9..2346bbe7d1260 100644
--- a/test/common/json/json_loader_test.cc
+++ b/test/common/json/json_loader_test.cc
@@ -129,4 +129,21 @@ TEST(JsonLoaderTest, Integer) {
   }
 }
 
+TEST(JsonLoaderTest, Double) {
+  {
+    StringLoader json("{\"value1\": 10.5, \"value2\": -12.3}");
+    EXPECT_EQ(10.5, json.getDouble("value1"));
+    EXPECT_EQ(-12.3, json.getDouble("value2"));
+  }
+  {
+    StringLoader json("{\"foo\": 13.22}");
+    EXPECT_EQ(13.22, json.getDouble("foo", 0));
+    EXPECT_EQ(0, json.getDouble("bar", 0));
+  }
+  {
+    StringLoader json("{\"foo\": \"bar\"}");
+    EXPECT_THROW(json.getDouble("foo"), Exception);
+  }
+}
+
 } // Json