diff --git a/cmds/statsd/src/StatsService.h b/cmds/statsd/src/StatsService.h
index 774a3e9..b5957b5 100644
--- a/cmds/statsd/src/StatsService.h
+++ b/cmds/statsd/src/StatsService.h
@@ -272,6 +272,10 @@
     FRIEND_TEST(PartialBucketE2eTest, TestCountMetricSplitOnUpgrade);
     FRIEND_TEST(PartialBucketE2eTest, TestCountMetricSplitOnRemoval);
     FRIEND_TEST(PartialBucketE2eTest, TestCountMetricWithoutSplit);
+    FRIEND_TEST(PartialBucketE2eTest, TestValueMetricWithoutMinPartialBucket);
+    FRIEND_TEST(PartialBucketE2eTest, TestValueMetricWithMinPartialBucket);
+    FRIEND_TEST(PartialBucketE2eTest, TestGaugeMetricWithoutMinPartialBucket);
+    FRIEND_TEST(PartialBucketE2eTest, TestGaugeMetricWithMinPartialBucket);
 };
 
 }  // namespace statsd
diff --git a/cmds/statsd/src/metrics/GaugeMetricProducer.cpp b/cmds/statsd/src/metrics/GaugeMetricProducer.cpp
index 6886f7c..1270856 100644
--- a/cmds/statsd/src/metrics/GaugeMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/GaugeMetricProducer.cpp
@@ -47,6 +47,9 @@
 const int FIELD_ID_GAUGE_METRICS = 8;
 // for GaugeMetricDataWrapper
 const int FIELD_ID_DATA = 1;
+const int FIELD_ID_SKIPPED = 2;
+const int FIELD_ID_SKIPPED_START = 1;
+const int FIELD_ID_SKIPPED_END = 2;
 // for GaugeMetricData
 const int FIELD_ID_DIMENSION_IN_WHAT = 1;
 const int FIELD_ID_DIMENSION_IN_CONDITION = 2;
@@ -66,6 +69,7 @@
     : MetricProducer(metric.id(), key, timeBaseNs, conditionIndex, wizard),
       mStatsPullerManager(statsPullerManager),
       mPullTagId(pullTagId),
+      mMinBucketSizeNs(metric.min_bucket_size_nanos()),
       mDimensionSoftLimit(StatsdStats::kAtomDimensionKeySizeLimitMap.find(pullTagId) !=
                                           StatsdStats::kAtomDimensionKeySizeLimitMap.end()
                                   ? StatsdStats::kAtomDimensionKeySizeLimitMap.at(pullTagId).first
@@ -174,6 +178,15 @@
     protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_ID, (long long)mMetricId);
     uint64_t protoToken = protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_ID_GAUGE_METRICS);
 
+    for (const auto& pair : mSkippedBuckets) {
+        uint64_t wrapperToken =
+                protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_SKIPPED);
+        protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_SKIPPED_START, (long long)pair.first);
+        protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_SKIPPED_END, (long long)pair.second);
+        protoOutput->end(wrapperToken);
+    }
+    mSkippedBuckets.clear();
+
     for (const auto& pair : mPastBuckets) {
         const MetricDimensionKey& dimensionKey = pair.first;
 
@@ -440,12 +453,16 @@
         info.mBucketEndNs = fullBucketEndTimeNs;
     }
 
-    for (const auto& slice : *mCurrentSlicedBucket) {
-        info.mGaugeAtoms = slice.second;
-        auto& bucketList = mPastBuckets[slice.first];
-        bucketList.push_back(info);
-        VLOG("Gauge gauge metric %lld, dump key value: %s", (long long)mMetricId,
-             slice.first.toString().c_str());
+    if (info.mBucketEndNs - mCurrentBucketStartTimeNs >= mMinBucketSizeNs) {
+        for (const auto& slice : *mCurrentSlicedBucket) {
+            info.mGaugeAtoms = slice.second;
+            auto& bucketList = mPastBuckets[slice.first];
+            bucketList.push_back(info);
+            VLOG("Gauge gauge metric %lld, dump key value: %s", (long long)mMetricId,
+                 slice.first.toString().c_str());
+        }
+    } else {
+        mSkippedBuckets.emplace_back(info.mBucketStartNs, info.mBucketEndNs);
     }
 
     // If we have anomaly trackers, we need to update the partial bucket values.
diff --git a/cmds/statsd/src/metrics/GaugeMetricProducer.h b/cmds/statsd/src/metrics/GaugeMetricProducer.h
index 08765c2..71d5912 100644
--- a/cmds/statsd/src/metrics/GaugeMetricProducer.h
+++ b/cmds/statsd/src/metrics/GaugeMetricProducer.h
@@ -136,6 +136,11 @@
     // this slice (ie, for partial buckets, we use the last partial bucket in this full bucket).
     std::shared_ptr<DimToValMap> mCurrentSlicedBucketForAnomaly;
 
+    // Pairs of (elapsed start, elapsed end) denoting buckets that were skipped.
+    std::list<std::pair<int64_t, int64_t>> mSkippedBuckets;
+
+    const int64_t mMinBucketSizeNs;
+
     // Translate Atom based bucket to single numeric value bucket for anomaly and updates the map
     // for each slice with the latest value.
     void updateCurrentSlicedBucketForAnomaly();
diff --git a/cmds/statsd/src/metrics/ValueMetricProducer.cpp b/cmds/statsd/src/metrics/ValueMetricProducer.cpp
index 844c728..27fd78f 100644
--- a/cmds/statsd/src/metrics/ValueMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/ValueMetricProducer.cpp
@@ -50,6 +50,9 @@
 const int FIELD_ID_VALUE_METRICS = 7;
 // for ValueMetricDataWrapper
 const int FIELD_ID_DATA = 1;
+const int FIELD_ID_SKIPPED = 2;
+const int FIELD_ID_SKIPPED_START = 1;
+const int FIELD_ID_SKIPPED_END = 2;
 // for ValueMetricData
 const int FIELD_ID_DIMENSION_IN_WHAT = 1;
 const int FIELD_ID_DIMENSION_IN_CONDITION = 2;
@@ -69,6 +72,7 @@
       mValueField(metric.value_field()),
       mStatsPullerManager(statsPullerManager),
       mPullTagId(pullTagId),
+      mMinBucketSizeNs(metric.min_bucket_size_nanos()),
       mDimensionSoftLimit(StatsdStats::kAtomDimensionKeySizeLimitMap.find(pullTagId) !=
                                           StatsdStats::kAtomDimensionKeySizeLimitMap.end()
                                   ? StatsdStats::kAtomDimensionKeySizeLimitMap.at(pullTagId).first
@@ -156,12 +160,21 @@
     } else {
         flushIfNeededLocked(dumpTimeNs);
     }
-    if (mPastBuckets.empty()) {
+    if (mPastBuckets.empty() && mSkippedBuckets.empty()) {
         return;
     }
     protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_ID, (long long)mMetricId);
     uint64_t protoToken = protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_ID_VALUE_METRICS);
 
+    for (const auto& pair : mSkippedBuckets) {
+        uint64_t wrapperToken =
+                protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_SKIPPED);
+        protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_SKIPPED_START, (long long)pair.first);
+        protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_SKIPPED_END, (long long)pair.second);
+        protoOutput->end(wrapperToken);
+    }
+    mSkippedBuckets.clear();
+
     for (const auto& pair : mPastBuckets) {
         const MetricDimensionKey& dimensionKey = pair.first;
         VLOG("  dimension key %s", dimensionKey.toString().c_str());
@@ -391,18 +404,23 @@
         info.mBucketEndNs = fullBucketEndTimeNs;
     }
 
-    int tainted = 0;
-    for (const auto& slice : mCurrentSlicedBucket) {
-        tainted += slice.second.tainted;
-        tainted += slice.second.startUpdated;
-        if (slice.second.hasValue) {
-            info.mValue = slice.second.sum;
-            // it will auto create new vector of ValuebucketInfo if the key is not found.
-            auto& bucketList = mPastBuckets[slice.first];
-            bucketList.push_back(info);
+    if (info.mBucketEndNs - mCurrentBucketStartTimeNs >= mMinBucketSizeNs) {
+        // The current bucket is large enough to keep.
+        int tainted = 0;
+        for (const auto& slice : mCurrentSlicedBucket) {
+            tainted += slice.second.tainted;
+            tainted += slice.second.startUpdated;
+            if (slice.second.hasValue) {
+                info.mValue = slice.second.sum;
+                // it will auto create new vector of ValuebucketInfo if the key is not found.
+                auto& bucketList = mPastBuckets[slice.first];
+                bucketList.push_back(info);
+            }
         }
+        VLOG("%d tainted pairs in the bucket", tainted);
+    } else {
+        mSkippedBuckets.emplace_back(info.mBucketStartNs, info.mBucketEndNs);
     }
-    VLOG("%d tainted pairs in the bucket", tainted);
 
     if (eventTimeNs > fullBucketEndTimeNs) {  // If full bucket, send to anomaly tracker.
         // Accumulate partial buckets with current value and then send to anomaly tracker.
diff --git a/cmds/statsd/src/metrics/ValueMetricProducer.h b/cmds/statsd/src/metrics/ValueMetricProducer.h
index 9c5a56c..8df30d3 100644
--- a/cmds/statsd/src/metrics/ValueMetricProducer.h
+++ b/cmds/statsd/src/metrics/ValueMetricProducer.h
@@ -148,6 +148,11 @@
     // TODO: Add a lock to mPastBuckets.
     std::unordered_map<MetricDimensionKey, std::vector<ValueBucket>> mPastBuckets;
 
+    // Pairs of (elapsed start, elapsed end) denoting buckets that were skipped.
+    std::list<std::pair<int64_t, int64_t>> mSkippedBuckets;
+
+    const int64_t mMinBucketSizeNs;
+
     // Util function to check whether the specified dimension hits the guardrail.
     bool hitGuardRailLocked(const MetricDimensionKey& newKey);
 
diff --git a/cmds/statsd/src/packages/UidMap.cpp b/cmds/statsd/src/packages/UidMap.cpp
index b3425a4..8eb5327 100644
--- a/cmds/statsd/src/packages/UidMap.cpp
+++ b/cmds/statsd/src/packages/UidMap.cpp
@@ -166,6 +166,8 @@
         } else {
             // Only notify the listeners if this is an app upgrade. If this app is being installed
             // for the first time, then we don't notify the listeners.
+            // It's also OK to split again if we're forming a partial bucket after re-installing an
+            // app after deletion.
             getListenerListCopyLocked(&broadcastList);
         }
         mChanges.emplace_back(false, timestamp, appName, uid, versionCode, prevVersion);
diff --git a/cmds/statsd/src/stats_log.proto b/cmds/statsd/src/stats_log.proto
index eaa7bf1..e1b6386 100644
--- a/cmds/statsd/src/stats_log.proto
+++ b/cmds/statsd/src/stats_log.proto
@@ -121,6 +121,11 @@
 
   // Fields 2 and 3 are reserved.
 
+  message SkippedBuckets {
+      optional int64 start_elapsed_nanos = 1;
+      optional int64 end_elapsed_nanos = 2;
+  }
+
   message EventMetricDataWrapper {
     repeated EventMetricData data = 1;
   }
@@ -132,10 +137,12 @@
   }
   message ValueMetricDataWrapper {
     repeated ValueMetricData data = 1;
+    repeated SkippedBuckets skipped = 2;
   }
 
   message GaugeMetricDataWrapper {
     repeated GaugeMetricData data = 1;
+    repeated SkippedBuckets skipped = 2;
   }
 
   oneof data {
diff --git a/cmds/statsd/src/statsd_config.proto b/cmds/statsd/src/statsd_config.proto
index 870f92e..fd36560 100644
--- a/cmds/statsd/src/statsd_config.proto
+++ b/cmds/statsd/src/statsd_config.proto
@@ -236,6 +236,8 @@
     ALL_CONDITION_CHANGES = 2;
   }
   optional SamplingType sampling_type = 9 [default = RANDOM_ONE_SAMPLE] ;
+
+  optional int64 min_bucket_size_nanos = 10;
 }
 
 message ValueMetric {
@@ -259,6 +261,8 @@
     SUM = 1;
   }
   optional AggregationType aggregation_type = 8 [default = SUM];
+
+  optional int64 min_bucket_size_nanos = 10;
 }
 
 message Alert {
diff --git a/cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp b/cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp
index af07f6b..a34aaf0 100644
--- a/cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp
+++ b/cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp
@@ -14,6 +14,7 @@
 
 #include <gtest/gtest.h>
 
+#include <binder/IPCThreadState.h>
 #include "src/StatsLogProcessor.h"
 #include "src/StatsService.h"
 #include "src/stats_log_util.h"
@@ -39,9 +40,13 @@
     service.addConfiguration(kConfigKey, configAsVec, String16(kAndroid.c_str()));
 }
 
-ConfigMetricsReport GetReports(StatsService& service) {
+ConfigMetricsReport GetReports(sp<StatsLogProcessor> processor, int64_t timestamp,
+                               bool include_current = false) {
     vector<uint8_t> output;
-    service.getData(kConfigKey, String16(kAndroid.c_str()), &output);
+    IPCThreadState* ipc = IPCThreadState::self();
+    ConfigKey configKey(ipc->getCallingUid(), kConfigKey);
+    processor->onDumpReport(configKey, timestamp, include_current /* include_current_bucket*/,
+                            &output);
     ConfigMetricsReportList reports;
     reports.ParseFromArray(output.data(), output.size());
     EXPECT_EQ(1, reports.reports_size());
@@ -61,6 +66,47 @@
     return config;
 }
 
+StatsdConfig MakeValueMetricConfig(int64_t minTime) {
+    StatsdConfig config;
+    config.add_allowed_log_source("AID_ROOT");  // LogEvent defaults to UID of root.
+
+    auto temperatureAtomMatcher = CreateTemperatureAtomMatcher();
+    *config.add_atom_matcher() = temperatureAtomMatcher;
+    *config.add_atom_matcher() = CreateScreenTurnedOnAtomMatcher();
+    *config.add_atom_matcher() = CreateScreenTurnedOffAtomMatcher();
+
+    auto valueMetric = config.add_value_metric();
+    valueMetric->set_id(123456);
+    valueMetric->set_what(temperatureAtomMatcher.id());
+    *valueMetric->mutable_value_field() =
+            CreateDimensions(android::util::TEMPERATURE, {3 /* temperature degree field */});
+    *valueMetric->mutable_dimensions_in_what() =
+            CreateDimensions(android::util::TEMPERATURE, {2 /* sensor name field */});
+    valueMetric->set_bucket(FIVE_MINUTES);
+    valueMetric->set_min_bucket_size_nanos(minTime);
+    return config;
+}
+
+StatsdConfig MakeGaugeMetricConfig(int64_t minTime) {
+    StatsdConfig config;
+    config.add_allowed_log_source("AID_ROOT");  // LogEvent defaults to UID of root.
+
+    auto temperatureAtomMatcher = CreateTemperatureAtomMatcher();
+    *config.add_atom_matcher() = temperatureAtomMatcher;
+    *config.add_atom_matcher() = CreateScreenTurnedOnAtomMatcher();
+    *config.add_atom_matcher() = CreateScreenTurnedOffAtomMatcher();
+
+    auto gaugeMetric = config.add_gauge_metric();
+    gaugeMetric->set_id(123456);
+    gaugeMetric->set_what(temperatureAtomMatcher.id());
+    gaugeMetric->mutable_gauge_fields_filter()->set_include_all(true);
+    *gaugeMetric->mutable_dimensions_in_what() =
+            CreateDimensions(android::util::TEMPERATURE, {2 /* sensor name field */});
+    gaugeMetric->set_bucket(FIVE_MINUTES);
+    gaugeMetric->set_min_bucket_size_nanos(minTime);
+    return config;
+}
+
 TEST(PartialBucketE2eTest, TestCountMetricWithoutSplit) {
     StatsService service(nullptr);
     SendConfig(service, MakeConfig());
@@ -70,7 +116,7 @@
     service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 1).get());
     service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 2).get());
 
-    ConfigMetricsReport report = GetReports(service);
+    ConfigMetricsReport report = GetReports(service.mProcessor, start + 3);
     // Expect no metrics since the bucket has not finished yet.
     EXPECT_EQ(0, report.metrics_size());
 }
@@ -89,7 +135,7 @@
     // Goes into the second bucket.
     service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 3).get());
 
-    ConfigMetricsReport report = GetReports(service);
+    ConfigMetricsReport report = GetReports(service.mProcessor, start + 4);
     EXPECT_EQ(0, report.metrics_size());
 }
 
@@ -106,7 +152,7 @@
     // Goes into the second bucket.
     service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 3).get());
 
-    ConfigMetricsReport report = GetReports(service);
+    ConfigMetricsReport report = GetReports(service.mProcessor, start + 4);
     EXPECT_EQ(1, report.metrics_size());
     EXPECT_EQ(1, report.metrics(0).count_metrics().data(0).bucket_info(0).count());
 }
@@ -124,11 +170,85 @@
     // Goes into the second bucket.
     service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 3).get());
 
-    ConfigMetricsReport report = GetReports(service);
+    ConfigMetricsReport report = GetReports(service.mProcessor, start + 4);
     EXPECT_EQ(1, report.metrics_size());
     EXPECT_EQ(1, report.metrics(0).count_metrics().data(0).bucket_info(0).count());
 }
 
+TEST(PartialBucketE2eTest, TestValueMetricWithoutMinPartialBucket) {
+    StatsService service(nullptr);
+    // Partial buckets don't occur when app is first installed.
+    service.mUidMap->updateApp(1, String16(kApp1.c_str()), 1, 1);
+    SendConfig(service, MakeValueMetricConfig(0));
+    const long start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
+                                                // initialized with.
+
+    service.mProcessor->informPullAlarmFired(5 * 60 * NS_PER_SEC + start);
+    service.mUidMap->updateApp(5 * 60 * NS_PER_SEC + start + 2, String16(kApp1.c_str()), 1, 2);
+
+    ConfigMetricsReport report =
+            GetReports(service.mProcessor, 5 * 60 * NS_PER_SEC + start + 100, true);
+    EXPECT_EQ(1, report.metrics_size());
+    EXPECT_EQ(0, report.metrics(0).value_metrics().skipped_size());
+}
+
+TEST(PartialBucketE2eTest, TestValueMetricWithMinPartialBucket) {
+    StatsService service(nullptr);
+    // Partial buckets don't occur when app is first installed.
+    service.mUidMap->updateApp(1, String16(kApp1.c_str()), 1, 1);
+    SendConfig(service, MakeValueMetricConfig(60 * NS_PER_SEC /* One minute */));
+    const long start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
+                                                // initialized with.
+
+    const int64_t endSkipped = 5 * 60 * NS_PER_SEC + start + 2;
+    service.mProcessor->informPullAlarmFired(5 * 60 * NS_PER_SEC + start);
+    service.mUidMap->updateApp(endSkipped, String16(kApp1.c_str()), 1, 2);
+
+    ConfigMetricsReport report =
+            GetReports(service.mProcessor, 5 * 60 * NS_PER_SEC + start + 100 * NS_PER_SEC, true);
+    EXPECT_EQ(1, report.metrics_size());
+    EXPECT_EQ(1, report.metrics(0).value_metrics().skipped_size());
+    // Can't test the start time since it will be based on the actual time when the pulling occurs.
+    EXPECT_EQ(endSkipped, report.metrics(0).value_metrics().skipped(0).end_elapsed_nanos());
+}
+
+TEST(PartialBucketE2eTest, TestGaugeMetricWithoutMinPartialBucket) {
+    StatsService service(nullptr);
+    // Partial buckets don't occur when app is first installed.
+    service.mUidMap->updateApp(1, String16(kApp1.c_str()), 1, 1);
+    SendConfig(service, MakeGaugeMetricConfig(0));
+    const long start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
+                                                // initialized with.
+
+    service.mProcessor->informPullAlarmFired(5 * 60 * NS_PER_SEC + start);
+    service.mUidMap->updateApp(5 * 60 * NS_PER_SEC + start + 2, String16(kApp1.c_str()), 1, 2);
+
+    ConfigMetricsReport report =
+            GetReports(service.mProcessor, 5 * 60 * NS_PER_SEC + start + 100, true);
+    EXPECT_EQ(1, report.metrics_size());
+    EXPECT_EQ(0, report.metrics(0).gauge_metrics().skipped_size());
+}
+
+TEST(PartialBucketE2eTest, TestGaugeMetricWithMinPartialBucket) {
+    StatsService service(nullptr);
+    // Partial buckets don't occur when app is first installed.
+    service.mUidMap->updateApp(1, String16(kApp1.c_str()), 1, 1);
+    SendConfig(service, MakeGaugeMetricConfig(60 * NS_PER_SEC /* One minute */));
+    const long start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
+                                                // initialized with.
+
+    const int64_t endSkipped = 5 * 60 * NS_PER_SEC + start + 2;
+    service.mProcessor->informPullAlarmFired(5 * 60 * NS_PER_SEC + start);
+    service.mUidMap->updateApp(endSkipped, String16(kApp1.c_str()), 1, 2);
+
+    ConfigMetricsReport report =
+            GetReports(service.mProcessor, 5 * 60 * NS_PER_SEC + start + 100 * NS_PER_SEC, true);
+    EXPECT_EQ(1, report.metrics_size());
+    EXPECT_EQ(1, report.metrics(0).gauge_metrics().skipped_size());
+    // Can't test the start time since it will be based on the actual time when the pulling occurs.
+    EXPECT_EQ(endSkipped, report.metrics(0).gauge_metrics().skipped(0).end_elapsed_nanos());
+}
+
 #else
 GTEST_LOG_(INFO) << "This test does nothing.\n";
 #endif