diff --git a/cmds/statsd/Android.mk b/cmds/statsd/Android.mk
index bc1c51a..8e46714 100644
--- a/cmds/statsd/Android.mk
+++ b/cmds/statsd/Android.mk
@@ -210,7 +210,8 @@
     tests/e2e/DimensionInCondition_e2e_simple_cond_test.cpp \
     tests/e2e/Anomaly_count_e2e_test.cpp \
     tests/e2e/Anomaly_duration_sum_e2e_test.cpp \
-    tests/e2e/ConfigTtl_e2e_test.cpp
+    tests/e2e/ConfigTtl_e2e_test.cpp \
+    tests/e2e/PartialBucket_e2e_test.cpp
 
 LOCAL_STATIC_LIBRARIES := \
     $(statsd_common_static_libraries) \
diff --git a/cmds/statsd/src/StatsLogProcessor.cpp b/cmds/statsd/src/StatsLogProcessor.cpp
index de8dfe4..cf97e54 100644
--- a/cmds/statsd/src/StatsLogProcessor.cpp
+++ b/cmds/statsd/src/StatsLogProcessor.cpp
@@ -316,7 +316,7 @@
 
     // Fill in UidMap.
     uint64_t uidMapToken = proto->start(FIELD_TYPE_MESSAGE | FIELD_ID_UID_MAP);
-    mUidMap->appendUidMap(key, proto);
+    mUidMap->appendUidMap(dumpTimeStampNs, key, proto);
     proto->end(uidMapToken);
 
     // Fill in the timestamps.
diff --git a/cmds/statsd/src/StatsService.cpp b/cmds/statsd/src/StatsService.cpp
index adc2664..dac73ef 100644
--- a/cmds/statsd/src/StatsService.cpp
+++ b/cmds/statsd/src/StatsService.cpp
@@ -640,7 +640,7 @@
                                          "Only system uid can call informAllUidData");
     }
 
-    mUidMap->updateMap(uid, version, app);
+    mUidMap->updateMap(getElapsedRealtimeNs(), uid, version, app);
     VLOG("StatsService::informAllUidData succeeded");
 
     return Status::ok();
@@ -653,7 +653,7 @@
         return Status::fromExceptionCode(Status::EX_SECURITY,
                                          "Only system uid can call informOnePackage");
     }
-    mUidMap->updateApp(app, uid, version);
+    mUidMap->updateApp(getElapsedRealtimeNs(), app, uid, version);
     return Status::ok();
 }
 
@@ -664,7 +664,7 @@
         return Status::fromExceptionCode(Status::EX_SECURITY,
                                          "Only system uid can call informOnePackageRemoved");
     }
-    mUidMap->removeApp(app, uid);
+    mUidMap->removeApp(getElapsedRealtimeNs(), app, uid);
     mConfigManager->RemoveConfigs(uid);
     return Status::ok();
 }
diff --git a/cmds/statsd/src/StatsService.h b/cmds/statsd/src/StatsService.h
index 8d2fd33..a4552e1 100644
--- a/cmds/statsd/src/StatsService.h
+++ b/cmds/statsd/src/StatsService.h
@@ -262,6 +262,10 @@
     FRIEND_TEST(StatsServiceTest, TestAddConfig_simple);
     FRIEND_TEST(StatsServiceTest, TestAddConfig_empty);
     FRIEND_TEST(StatsServiceTest, TestAddConfig_invalid);
+    FRIEND_TEST(PartialBucketE2eTest, TestCountMetricNoSplitOnNewApp);
+    FRIEND_TEST(PartialBucketE2eTest, TestCountMetricSplitOnUpgrade);
+    FRIEND_TEST(PartialBucketE2eTest, TestCountMetricSplitOnRemoval);
+    FRIEND_TEST(PartialBucketE2eTest, TestCountMetricWithoutSplit);
 };
 
 }  // namespace statsd
diff --git a/cmds/statsd/src/guardrail/StatsdStats.cpp b/cmds/statsd/src/guardrail/StatsdStats.cpp
index 0c076e9..f501fd4 100644
--- a/cmds/statsd/src/guardrail/StatsdStats.cpp
+++ b/cmds/statsd/src/guardrail/StatsdStats.cpp
@@ -91,11 +91,10 @@
 const int FIELD_ID_ALERT_STATS_ID = 1;
 const int FIELD_ID_ALERT_STATS_COUNT = 2;
 
-const int FIELD_ID_UID_MAP_SNAPSHOTS = 1;
-const int FIELD_ID_UID_MAP_CHANGES = 2;
-const int FIELD_ID_UID_MAP_BYTES_USED = 3;
-const int FIELD_ID_UID_MAP_DROPPED_SNAPSHOTS = 4;
-const int FIELD_ID_UID_MAP_DROPPED_CHANGES = 5;
+const int FIELD_ID_UID_MAP_CHANGES = 1;
+const int FIELD_ID_UID_MAP_BYTES_USED = 2;
+const int FIELD_ID_UID_MAP_DROPPED_CHANGES = 3;
+const int FIELD_ID_UID_MAP_DELETED_APPS = 4;
 
 const std::map<int, std::pair<size_t, size_t>> StatsdStats::kAtomDimensionKeySizeLimitMap = {
         {android::util::CPU_TIME_PER_UID_FREQ, {6000, 10000}},
@@ -229,15 +228,14 @@
     it->second->dump_report_time_sec.push_back(timeSec);
 }
 
-void StatsdStats::noteUidMapDropped(int snapshots, int deltas) {
+void StatsdStats::noteUidMapDropped(int deltas) {
     lock_guard<std::mutex> lock(mLock);
-    mUidMapStats.dropped_snapshots += mUidMapStats.dropped_snapshots + snapshots;
     mUidMapStats.dropped_changes += mUidMapStats.dropped_changes + deltas;
 }
 
-void StatsdStats::setUidMapSnapshots(int snapshots) {
+void StatsdStats::noteUidMapAppDeletionDropped() {
     lock_guard<std::mutex> lock(mLock);
-    mUidMapStats.snapshots = snapshots;
+    mUidMapStats.deleted_apps++;
 }
 
 void StatsdStats::setUidMapChanges(int changes) {
@@ -478,11 +476,9 @@
         fprintf(out, "Subscriber alarm registrations: %d\n", mPeriodicAlarmRegisteredStats);
     }
 
-    fprintf(out,
-            "UID map stats: bytes=%d, snapshots=%d, changes=%d, snapshots lost=%d, changes "
-            "lost=%d\n",
-            mUidMapStats.bytes_used, mUidMapStats.snapshots, mUidMapStats.changes,
-            mUidMapStats.dropped_snapshots, mUidMapStats.dropped_changes);
+    fprintf(out, "UID map stats: bytes=%d, changes=%d, deleted=%d, changes lost=%d\n",
+            mUidMapStats.bytes_used, mUidMapStats.changes, mUidMapStats.deleted_apps,
+            mUidMapStats.dropped_changes);
 
     for (const auto& error : mLoggerErrors) {
         time_t error_time = error.first;
@@ -624,12 +620,10 @@
     }
 
     uint64_t uidMapToken = proto.start(FIELD_TYPE_MESSAGE | FIELD_ID_UIDMAP_STATS);
-    proto.write(FIELD_TYPE_INT32 | FIELD_ID_UID_MAP_SNAPSHOTS, mUidMapStats.snapshots);
     proto.write(FIELD_TYPE_INT32 | FIELD_ID_UID_MAP_CHANGES, mUidMapStats.changes);
     proto.write(FIELD_TYPE_INT32 | FIELD_ID_UID_MAP_BYTES_USED, mUidMapStats.bytes_used);
-    proto.write(FIELD_TYPE_INT32 | FIELD_ID_UID_MAP_DROPPED_SNAPSHOTS,
-                mUidMapStats.dropped_snapshots);
     proto.write(FIELD_TYPE_INT32 | FIELD_ID_UID_MAP_DROPPED_CHANGES, mUidMapStats.dropped_changes);
+    proto.write(FIELD_TYPE_INT32 | FIELD_ID_UID_MAP_DELETED_APPS, mUidMapStats.deleted_apps);
     proto.end(uidMapToken);
 
     for (const auto& error : mLoggerErrors) {
diff --git a/cmds/statsd/src/guardrail/StatsdStats.h b/cmds/statsd/src/guardrail/StatsdStats.h
index 00bef75..13e35dd 100644
--- a/cmds/statsd/src/guardrail/StatsdStats.h
+++ b/cmds/statsd/src/guardrail/StatsdStats.h
@@ -73,11 +73,10 @@
 };
 
 struct UidMapStats {
-    int32_t snapshots;
     int32_t changes;
     int32_t bytes_used;
-    int32_t dropped_snapshots;
     int32_t dropped_changes;
+    int32_t deleted_apps = 0;
 };
 
 // Keeps track of stats of statsd.
@@ -121,6 +120,9 @@
     // is critical for understanding the metrics.
     const static size_t kMaxBytesUsedUidMap = 50 * 1024;
 
+    // The number of deleted apps that are stored in the uid map.
+    const static int kMaxDeletedAppsInUidMap = 100;
+
     /* Minimum period between two broadcasts in nanoseconds. */
     static const int64_t kMinBroadcastPeriodNs = 60 * NS_PER_SEC;
 
@@ -244,14 +246,18 @@
     void noteRegisteredPeriodicAlarmChanged();
 
     /**
-     * Records the number of snapshot and delta entries that are being dropped from the uid map.
+     * Records the number of delta entries that are being dropped from the uid map.
      */
-    void noteUidMapDropped(int snapshots, int deltas);
+    void noteUidMapDropped(int deltas);
 
     /**
-     * Updates the number of snapshots currently stored in the uid map.
+     * Records that an app was deleted (from statsd's map).
      */
-    void setUidMapSnapshots(int snapshots);
+    void noteUidMapAppDeletionDropped();
+
+    /**
+     * Updates the number of changes currently stored in the uid map.
+     */
     void setUidMapChanges(int changes);
     void setCurrentUidMapMemory(int bytes);
 
diff --git a/cmds/statsd/src/metrics/MetricProducer.h b/cmds/statsd/src/metrics/MetricProducer.h
index 139a407..db5d32c 100644
--- a/cmds/statsd/src/metrics/MetricProducer.h
+++ b/cmds/statsd/src/metrics/MetricProducer.h
@@ -80,11 +80,13 @@
     };
 
     void notifyAppRemoved(const int64_t& eventTimeNs, const string& apk, const int uid) override{
-            // TODO: Implement me.
+        // Force buckets to split on removal also.
+        notifyAppUpgrade(eventTimeNs, apk, uid, 0);
     };
 
     void onUidMapReceived(const int64_t& eventTimeNs) override{
-            // TODO: Implement me.
+            // Purposefully don't flush partial buckets on a new snapshot.
+            // This occurs if a new user is added/removed or statsd crashes.
     };
 
     // Consume the parsed stats log entry that already matched the "what" of the metric.
diff --git a/cmds/statsd/src/metrics/metrics_manager_util.cpp b/cmds/statsd/src/metrics/metrics_manager_util.cpp
index b7c5795..566d34e 100644
--- a/cmds/statsd/src/metrics/metrics_manager_util.cpp
+++ b/cmds/statsd/src/metrics/metrics_manager_util.cpp
@@ -263,15 +263,14 @@
 }
 
 bool initMetrics(const ConfigKey& key, const StatsdConfig& config, const long timeBaseSec,
-                 const unordered_map<int64_t, int>& logTrackerMap,
+                 UidMap& uidMap, const unordered_map<int64_t, int>& logTrackerMap,
                  const unordered_map<int64_t, int>& conditionTrackerMap,
                  const vector<sp<LogMatchingTracker>>& allAtomMatchers,
                  vector<sp<ConditionTracker>>& allConditionTrackers,
                  vector<sp<MetricProducer>>& allMetricProducers,
                  unordered_map<int, std::vector<int>>& conditionToMetricMap,
                  unordered_map<int, std::vector<int>>& trackerToMetricMap,
-                 unordered_map<int64_t, int>& metricMap,
-                 std::set<int64_t> &noReportMetricIds) {
+                 unordered_map<int64_t, int>& metricMap, std::set<int64_t>& noReportMetricIds) {
     sp<ConditionWizard> wizard = new ConditionWizard(allConditionTrackers);
     const int allMetricsCount = config.count_metric_size() + config.duration_metric_size() +
                                 config.event_metric_size() + config.value_metric_size();
@@ -538,6 +537,9 @@
         }
         noReportMetricIds.insert(no_report_metric);
     }
+    for (auto it : allMetricProducers) {
+        uidMap.addListener(it);
+    }
     return true;
 }
 
@@ -642,12 +644,10 @@
     return true;
 }
 
-bool initStatsdConfig(const ConfigKey& key, const StatsdConfig& config,
-                      const UidMap& uidMap,
+bool initStatsdConfig(const ConfigKey& key, const StatsdConfig& config, UidMap& uidMap,
                       const sp<AlarmMonitor>& anomalyAlarmMonitor,
-                      const sp<AlarmMonitor>& periodicAlarmMonitor,
-                      const long timeBaseSec, const long currentTimeSec,
-                      set<int>& allTagIds,
+                      const sp<AlarmMonitor>& periodicAlarmMonitor, const long timeBaseSec,
+                      const long currentTimeSec, set<int>& allTagIds,
                       vector<sp<LogMatchingTracker>>& allAtomMatchers,
                       vector<sp<ConditionTracker>>& allConditionTrackers,
                       vector<sp<MetricProducer>>& allMetricProducers,
@@ -656,7 +656,7 @@
                       unordered_map<int, std::vector<int>>& conditionToMetricMap,
                       unordered_map<int, std::vector<int>>& trackerToMetricMap,
                       unordered_map<int, std::vector<int>>& trackerToConditionMap,
-                      std::set<int64_t> &noReportMetricIds) {
+                      std::set<int64_t>& noReportMetricIds) {
     unordered_map<int64_t, int> logTrackerMap;
     unordered_map<int64_t, int> conditionTrackerMap;
     unordered_map<int64_t, int> metricProducerMap;
@@ -673,9 +673,10 @@
         return false;
     }
 
-    if (!initMetrics(key, config, timeBaseSec, logTrackerMap, conditionTrackerMap, allAtomMatchers,
-                     allConditionTrackers, allMetricProducers, conditionToMetricMap,
-                     trackerToMetricMap, metricProducerMap, noReportMetricIds)) {
+    if (!initMetrics(key, config, timeBaseSec, uidMap, logTrackerMap, conditionTrackerMap,
+                     allAtomMatchers, allConditionTrackers, allMetricProducers,
+                     conditionToMetricMap, trackerToMetricMap, metricProducerMap,
+                     noReportMetricIds)) {
         ALOGE("initMetricProducers failed");
         return false;
     }
diff --git a/cmds/statsd/src/metrics/metrics_manager_util.h b/cmds/statsd/src/metrics/metrics_manager_util.h
index 3754ae0..0ebdcf9 100644
--- a/cmds/statsd/src/metrics/metrics_manager_util.h
+++ b/cmds/statsd/src/metrics/metrics_manager_util.h
@@ -81,7 +81,7 @@
 //                          the list of MetricProducer index
 // [trackerToMetricMap]: contains the mapping from log tracker to MetricProducer index.
 bool initMetrics(
-        const ConfigKey& key, const StatsdConfig& config, const long timeBaseSec,
+        const ConfigKey& key, const StatsdConfig& config, const long timeBaseSec, UidMap& uidMap,
         const std::unordered_map<int64_t, int>& logTrackerMap,
         const std::unordered_map<int64_t, int>& conditionTrackerMap,
         const std::unordered_map<int, std::vector<MetricConditionLink>>& eventConditionLinks,
@@ -90,16 +90,14 @@
         std::vector<sp<MetricProducer>>& allMetricProducers,
         std::unordered_map<int, std::vector<int>>& conditionToMetricMap,
         std::unordered_map<int, std::vector<int>>& trackerToMetricMap,
-        std::set<int64_t> &noReportMetricIds);
+        std::set<int64_t>& noReportMetricIds);
 
 // Initialize MetricsManager from StatsdConfig.
 // Parameters are the members of MetricsManager. See MetricsManager for declaration.
-bool initStatsdConfig(const ConfigKey& key, const StatsdConfig& config,
-                      const UidMap& uidMap,
+bool initStatsdConfig(const ConfigKey& key, const StatsdConfig& config, UidMap& uidMap,
                       const sp<AlarmMonitor>& anomalyAlarmMonitor,
-                      const sp<AlarmMonitor>& periodicAlarmMonitor,
-                      const long timeBaseSec, const long currentTimeSec,
-                      std::set<int>& allTagIds,
+                      const sp<AlarmMonitor>& periodicAlarmMonitor, const long timeBaseSec,
+                      const long currentTimeSec, std::set<int>& allTagIds,
                       std::vector<sp<LogMatchingTracker>>& allAtomMatchers,
                       std::vector<sp<ConditionTracker>>& allConditionTrackers,
                       std::vector<sp<MetricProducer>>& allMetricProducers,
@@ -108,7 +106,7 @@
                       std::unordered_map<int, std::vector<int>>& conditionToMetricMap,
                       std::unordered_map<int, std::vector<int>>& trackerToMetricMap,
                       std::unordered_map<int, std::vector<int>>& trackerToConditionMap,
-                      std::set<int64_t> &noReportMetricIds);
+                      std::set<int64_t>& noReportMetricIds);
 
 bool isStateTracker(const SimplePredicate& simplePredicate, std::vector<Matcher>* primaryKeys);
 
diff --git a/cmds/statsd/src/packages/UidMap.cpp b/cmds/statsd/src/packages/UidMap.cpp
index 8c8152d..b3425a4 100644
--- a/cmds/statsd/src/packages/UidMap.cpp
+++ b/cmds/statsd/src/packages/UidMap.cpp
@@ -45,6 +45,7 @@
 const int FIELD_ID_SNAPSHOT_PACKAGE_NAME = 1;
 const int FIELD_ID_SNAPSHOT_PACKAGE_VERSION = 2;
 const int FIELD_ID_SNAPSHOT_PACKAGE_UID = 3;
+const int FIELD_ID_SNAPSHOT_PACKAGE_DELETED = 4;
 const int FIELD_ID_SNAPSHOT_TIMESTAMP = 1;
 const int FIELD_ID_SNAPSHOT_PACKAGE_INFO = 2;
 const int FIELD_ID_SNAPSHOTS = 1;
@@ -53,7 +54,8 @@
 const int FIELD_ID_CHANGE_TIMESTAMP = 2;
 const int FIELD_ID_CHANGE_PACKAGE = 3;
 const int FIELD_ID_CHANGE_UID = 4;
-const int FIELD_ID_CHANGE_VERSION = 5;
+const int FIELD_ID_CHANGE_NEW_VERSION = 5;
+const int FIELD_ID_CHANGE_PREV_VERSION = 6;
 
 UidMap::UidMap() : mBytesUsed(0) {}
 
@@ -62,13 +64,8 @@
 bool UidMap::hasApp(int uid, const string& packageName) const {
     lock_guard<mutex> lock(mMutex);
 
-    auto range = mMap.equal_range(uid);
-    for (auto it = range.first; it != range.second; ++it) {
-        if (it->second.packageName == packageName) {
-            return true;
-        }
-    }
-    return false;
+    auto it = mMap.find(std::make_pair(uid, packageName));
+    return it != mMap.end() && !it->second.deleted;
 }
 
 string UidMap::normalizeAppName(const string& appName) const {
@@ -84,10 +81,10 @@
 
 std::set<string> UidMap::getAppNamesFromUidLocked(const int32_t& uid, bool returnNormalized) const {
     std::set<string> names;
-    auto range = mMap.equal_range(uid);
-    for (auto it = range.first; it != range.second; ++it) {
-        names.insert(returnNormalized ?
-            normalizeAppName(it->second.packageName) : it->second.packageName);
+    for (const auto& kv : mMap) {
+        if (kv.first.first == uid && !kv.second.deleted) {
+            names.insert(returnNormalized ? normalizeAppName(kv.first.second) : kv.first.second);
+        }
     }
     return names;
 }
@@ -95,18 +92,11 @@
 int64_t UidMap::getAppVersion(int uid, const string& packageName) const {
     lock_guard<mutex> lock(mMutex);
 
-    auto range = mMap.equal_range(uid);
-    for (auto it = range.first; it != range.second; ++it) {
-        if (it->second.packageName == packageName) {
-            return it->second.versionCode;
-        }
+    auto it = mMap.find(std::make_pair(uid, packageName));
+    if (it == mMap.end() || it->second.deleted) {
+        return 0;
     }
-    return 0;
-}
-
-void UidMap::updateMap(const vector<int32_t>& uid, const vector<int64_t>& versionCode,
-                       const vector<String16>& packageName) {
-    updateMap(getElapsedRealtimeNs(), uid, versionCode, packageName);
+    return it->second.versionCode;
 }
 
 void UidMap::updateMap(const int64_t& timestamp, const vector<int32_t>& uid,
@@ -115,20 +105,31 @@
     {
         lock_guard<mutex> lock(mMutex);  // Exclusively lock for updates.
 
-        mMap.clear();
-        vector<const SnapshotPackageInfo> infos;
-        for (size_t j = 0; j < uid.size(); j++) {
-            string package = string(String8(packageName[j]).string());
-            mMap.insert(make_pair(uid[j], AppData(package, versionCode[j])));
-            infos.emplace_back(package, versionCode[j], uid[j]);
+        std::unordered_map<std::pair<int, string>, AppData, PairHash> deletedApps;
+
+        // Copy all the deleted apps.
+        for (const auto& kv : mMap) {
+            if (kv.second.deleted) {
+                deletedApps[kv.first] = kv.second;
+            }
         }
 
-        mSnapshots.emplace_back(timestamp, infos);
+        mMap.clear();
+        for (size_t j = 0; j < uid.size(); j++) {
+            string package = string(String8(packageName[j]).string());
+            mMap[std::make_pair(uid[j], package)] = AppData(versionCode[j]);
+        }
 
-        mBytesUsed += mSnapshots.back().bytes;
+        for (const auto& kv : deletedApps) {
+            auto mMapIt = mMap.find(kv.first);
+            if (mMapIt != mMap.end()) {
+                // Insert this deleted app back into the current map.
+                mMap[kv.first] = kv.second;
+            }
+        }
+
         ensureBytesUsedBelowLimit();
         StatsdStats::getInstance().setCurrentUidMapMemory(mBytesUsed);
-        StatsdStats::getInstance().setUidMapSnapshots(mSnapshots.size());
         getListenerListCopyLocked(&broadcastList);
     }
     // To avoid invoking callback while holding the internal lock. we get a copy of the listener
@@ -144,38 +145,34 @@
     }
 }
 
-void UidMap::updateApp(const String16& app_16, const int32_t& uid, const int64_t& versionCode) {
-    updateApp(getElapsedRealtimeNs(), app_16, uid, versionCode);
-}
-
 void UidMap::updateApp(const int64_t& timestamp, const String16& app_16, const int32_t& uid,
                        const int64_t& versionCode) {
     vector<wp<PackageInfoListener>> broadcastList;
     string appName = string(String8(app_16).string());
     {
         lock_guard<mutex> lock(mMutex);
-
-        mChanges.emplace_back(false, timestamp, appName, uid, versionCode);
+        int32_t prevVersion = 0;
+        bool found = false;
+        auto it = mMap.find(std::make_pair(uid, appName));
+        if (it != mMap.end()) {
+            found = true;
+            prevVersion = it->second.versionCode;
+            it->second.versionCode = versionCode;
+            it->second.deleted = false;
+        }
+        if (!found) {
+            // Otherwise, we need to add an app at this uid.
+            mMap[std::make_pair(uid, appName)] = AppData(versionCode);
+        } 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.
+            getListenerListCopyLocked(&broadcastList);
+        }
+        mChanges.emplace_back(false, timestamp, appName, uid, versionCode, prevVersion);
         mBytesUsed += kBytesChangeRecord;
         ensureBytesUsedBelowLimit();
         StatsdStats::getInstance().setCurrentUidMapMemory(mBytesUsed);
         StatsdStats::getInstance().setUidMapChanges(mChanges.size());
-
-        auto range = mMap.equal_range(int(uid));
-        bool found = false;
-        for (auto it = range.first; it != range.second; ++it) {
-            // If we find the exact same app name and uid, update the app version directly.
-            if (it->second.packageName == appName) {
-                it->second.versionCode = versionCode;
-                found = true;
-                break;
-            }
-        }
-        if (!found) {
-            // Otherwise, we need to add an app at this uid.
-            mMap.insert(make_pair(uid, AppData(appName, versionCode)));
-        }
-        getListenerListCopyLocked(&broadcastList);
     }
 
     for (auto weakPtr : broadcastList) {
@@ -195,22 +192,14 @@
     }
     while (mBytesUsed > limit) {
         ALOGI("Bytes used %zu is above limit %zu, need to delete something", mBytesUsed, limit);
-        if (mSnapshots.size() > 0) {
-            mBytesUsed -= mSnapshots.front().bytes;
-            mSnapshots.pop_front();
-            StatsdStats::getInstance().noteUidMapDropped(1, 0);
-        } else if (mChanges.size() > 0) {
+        if (mChanges.size() > 0) {
             mBytesUsed -= kBytesChangeRecord;
             mChanges.pop_front();
-            StatsdStats::getInstance().noteUidMapDropped(0, 1);
+            StatsdStats::getInstance().noteUidMapDropped(1);
         }
     }
 }
 
-void UidMap::removeApp(const String16& app_16, const int32_t& uid) {
-    removeApp(getElapsedRealtimeNs(), app_16, uid);
-}
-
 void UidMap::getListenerListCopyLocked(vector<wp<PackageInfoListener>>* output) {
     for (auto weakIt = mSubscribers.begin(); weakIt != mSubscribers.end();) {
         auto strongPtr = weakIt->promote();
@@ -230,19 +219,26 @@
     {
         lock_guard<mutex> lock(mMutex);
 
-        mChanges.emplace_back(true, timestamp, app, uid, 0);
+        int32_t prevVersion = 0;
+        auto key = std::make_pair(uid, app);
+        auto it = mMap.find(key);
+        if (it != mMap.end() && !it->second.deleted) {
+            prevVersion = it->second.versionCode;
+            it->second.deleted = true;
+            mDeletedApps.push_back(key);
+        }
+        if (mDeletedApps.size() > StatsdStats::kMaxDeletedAppsInUidMap) {
+            // Delete the oldest one.
+            auto oldest = mDeletedApps.front();
+            mDeletedApps.pop_front();
+            mMap.erase(oldest);
+            StatsdStats::getInstance().noteUidMapAppDeletionDropped();
+        }
+        mChanges.emplace_back(true, timestamp, app, uid, 0, prevVersion);
         mBytesUsed += kBytesChangeRecord;
         ensureBytesUsedBelowLimit();
         StatsdStats::getInstance().setCurrentUidMapMemory(mBytesUsed);
         StatsdStats::getInstance().setUidMapChanges(mChanges.size());
-
-        auto range = mMap.equal_range(int(uid));
-        for (auto it = range.first; it != range.second; ++it) {
-            if (it->second.packageName == app) {
-                mMap.erase(it);
-                break;
-            }
-        }
         getListenerListCopyLocked(&broadcastList);
     }
 
@@ -290,22 +286,20 @@
 }
 
 void UidMap::clearOutput() {
-    mSnapshots.clear();
     mChanges.clear();
     // Also update the guardrail trackers.
     StatsdStats::getInstance().setUidMapChanges(0);
-    StatsdStats::getInstance().setUidMapSnapshots(1);
     mBytesUsed = 0;
     StatsdStats::getInstance().setCurrentUidMapMemory(mBytesUsed);
 }
 
 int64_t UidMap::getMinimumTimestampNs() {
     int64_t m = 0;
-    for (auto it : mLastUpdatePerConfigKey) {
+    for (const auto& kv : mLastUpdatePerConfigKey) {
         if (m == 0) {
-            m = it.second;
-        } else if (it.second < m) {
-            m = it.second;
+            m = kv.second;
+        } else if (kv.second < m) {
+            m = kv.second;
         }
     }
     return m;
@@ -315,10 +309,6 @@
     return mBytesUsed;
 }
 
-void UidMap::appendUidMap(const ConfigKey& key, ProtoOutputStream* proto) {
-    appendUidMap(getElapsedRealtimeNs(), key, proto);
-}
-
 void UidMap::appendUidMap(const int64_t& timestamp, const ConfigKey& key,
                           ProtoOutputStream* proto) {
     lock_guard<mutex> lock(mMutex);  // Lock for updates
@@ -332,34 +322,27 @@
                          (long long)record.timestampNs);
             proto->write(FIELD_TYPE_STRING | FIELD_ID_CHANGE_PACKAGE, record.package);
             proto->write(FIELD_TYPE_INT32 | FIELD_ID_CHANGE_UID, (int)record.uid);
-            proto->write(FIELD_TYPE_INT32 | FIELD_ID_CHANGE_VERSION, (int)record.version);
+            proto->write(FIELD_TYPE_INT32 | FIELD_ID_CHANGE_NEW_VERSION, (int)record.version);
+            proto->write(FIELD_TYPE_INT32 | FIELD_ID_CHANGE_PREV_VERSION, (int)record.prevVersion);
             proto->end(changesToken);
         }
     }
 
-    bool atLeastOneSnapshot = false;
-    unsigned int count = 0;
-    for (const SnapshotRecord& record : mSnapshots) {
-        // Ensure that we include at least the latest snapshot.
-        if ((count == mSnapshots.size() - 1 && !atLeastOneSnapshot) ||
-            record.timestampNs > mLastUpdatePerConfigKey[key]) {
-            uint64_t snapshotsToken =
-                    proto->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_SNAPSHOTS);
-            atLeastOneSnapshot = true;
-            count++;
-            proto->write(FIELD_TYPE_INT64 | FIELD_ID_SNAPSHOT_TIMESTAMP,
-                         (long long)record.timestampNs);
-            for (const SnapshotPackageInfo& info : record.infos) {
-                uint64_t token = proto->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED |
-                                              FIELD_ID_SNAPSHOT_PACKAGE_INFO);
-                proto->write(FIELD_TYPE_STRING | FIELD_ID_SNAPSHOT_PACKAGE_NAME, info.package);
-                proto->write(FIELD_TYPE_INT32 | FIELD_ID_SNAPSHOT_PACKAGE_VERSION, info.version);
-                proto->write(FIELD_TYPE_INT32 | FIELD_ID_SNAPSHOT_PACKAGE_UID, info.uid);
-                proto->end(token);
-            }
-            proto->end(snapshotsToken);
-        }
+    // Write snapshot from current uid map state.
+    uint64_t snapshotsToken =
+            proto->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_SNAPSHOTS);
+    proto->write(FIELD_TYPE_INT64 | FIELD_ID_SNAPSHOT_TIMESTAMP, (long long)timestamp);
+    for (const auto& kv : mMap) {
+        uint64_t token = proto->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED |
+                                      FIELD_ID_SNAPSHOT_PACKAGE_INFO);
+        proto->write(FIELD_TYPE_STRING | FIELD_ID_SNAPSHOT_PACKAGE_NAME, kv.first.second);
+        proto->write(FIELD_TYPE_INT32 | FIELD_ID_SNAPSHOT_PACKAGE_VERSION,
+                     (int)kv.second.versionCode);
+        proto->write(FIELD_TYPE_INT32 | FIELD_ID_SNAPSHOT_PACKAGE_UID, kv.first.first);
+        proto->write(FIELD_TYPE_BOOL | FIELD_ID_SNAPSHOT_PACKAGE_DELETED, kv.second.deleted);
+        proto->end(token);
     }
+    proto->end(snapshotsToken);
 
     int64_t prevMin = getMinimumTimestampNs();
     mLastUpdatePerConfigKey[key] = timestamp;
@@ -368,14 +351,6 @@
     if (newMin > prevMin) {  // Delete anything possible now that the minimum has
                              // moved forward.
         int64_t cutoff_nanos = newMin;
-        for (auto it_snapshots = mSnapshots.begin(); it_snapshots != mSnapshots.end();) {
-            if (it_snapshots->timestampNs < cutoff_nanos) {
-                mBytesUsed -= it_snapshots->bytes;
-                it_snapshots = mSnapshots.erase(it_snapshots);
-            } else {
-                ++it_snapshots;
-            }
-        }
         for (auto it_changes = mChanges.begin(); it_changes != mChanges.end();) {
             if (it_changes->timestampNs < cutoff_nanos) {
                 mBytesUsed -= kBytesChangeRecord;
@@ -384,53 +359,24 @@
                 ++it_changes;
             }
         }
-
-        if (mSnapshots.size() == 0) {
-            // Produce another snapshot. This results in extra data being uploaded but
-            // helps ensure we can re-construct the UID->app name, versionCode mapping
-            // in server.
-            vector<const SnapshotPackageInfo> infos;
-            for (const auto& it : mMap) {
-                infos.emplace_back(it.second.packageName, it.second.versionCode, it.first);
-            }
-
-            mSnapshots.emplace_back(timestamp, infos);
-            mBytesUsed += mSnapshots.back().bytes;
-        }
     }
     StatsdStats::getInstance().setCurrentUidMapMemory(mBytesUsed);
     StatsdStats::getInstance().setUidMapChanges(mChanges.size());
-    StatsdStats::getInstance().setUidMapSnapshots(mSnapshots.size());
 }
 
 void UidMap::printUidMap(FILE* out) const {
     lock_guard<mutex> lock(mMutex);
 
-    for (auto it : mMap) {
-        fprintf(out, "%s, v%" PRId64 " (%i)\n", it.second.packageName.c_str(),
-                it.second.versionCode, it.first);
+    for (const auto& kv : mMap) {
+        if (!kv.second.deleted) {
+            fprintf(out, "%s, v%" PRId64 " (%i)\n", kv.first.second.c_str(), kv.second.versionCode,
+                    kv.first.first);
+        }
     }
 }
 
 void UidMap::OnConfigUpdated(const ConfigKey& key) {
     mLastUpdatePerConfigKey[key] = -1;
-
-    // Ensure there is at least one snapshot available since this configuration also needs to know
-    // what all the uid's represent.
-    if (mSnapshots.size() == 0) {
-        sp<IStatsCompanionService> statsCompanion = nullptr;
-        // Get statscompanion service from service manager
-        const sp<IServiceManager> sm(defaultServiceManager());
-        if (sm != nullptr) {
-            const String16 name("statscompanion");
-            statsCompanion = interface_cast<IStatsCompanionService>(sm->checkService(name));
-            if (statsCompanion == nullptr) {
-                ALOGW("statscompanion service unavailable!");
-                return;
-            }
-            statsCompanion->triggerUidSnapshot();
-        }
-    }
 }
 
 void UidMap::OnConfigRemoved(const ConfigKey& key) {
@@ -441,9 +387,9 @@
     lock_guard<mutex> lock(mMutex);
 
     set<int32_t> results;
-    for (const auto& pair : mMap) {
-        if (pair.second.packageName == package) {
-            results.insert(pair.first);
+    for (const auto& kv : mMap) {
+        if (kv.first.second == package && !kv.second.deleted) {
+            results.insert(kv.first.first);
         }
     }
     return results;
diff --git a/cmds/statsd/src/packages/UidMap.h b/cmds/statsd/src/packages/UidMap.h
index a3632d2..7222e85 100644
--- a/cmds/statsd/src/packages/UidMap.h
+++ b/cmds/statsd/src/packages/UidMap.h
@@ -43,10 +43,13 @@
 namespace statsd {
 
 struct AppData {
-    const string packageName;
     int64_t versionCode;
+    bool deleted;
 
-    AppData(const string& a, const int64_t v) : packageName(a), versionCode(v){};
+    // Empty constructor needed for unordered map.
+    AppData() {
+    }
+    AppData(const int64_t v) : versionCode(v), deleted(false){};
 };
 
 // When calling appendUidMap, we retrieve all the ChangeRecords since the last
@@ -57,55 +60,21 @@
     const string package;
     const int32_t uid;
     const int32_t version;
+    const int32_t prevVersion;
 
     ChangeRecord(const bool isDeletion, const int64_t timestampNs, const string& package,
-                 const int32_t uid, const int32_t version)
+                 const int32_t uid, const int32_t version, const int32_t prevVersion)
         : deletion(isDeletion),
           timestampNs(timestampNs),
           package(package),
           uid(uid),
-          version(version) {
+          version(version),
+          prevVersion(prevVersion) {
     }
 };
 
 const unsigned int kBytesChangeRecord = sizeof(struct ChangeRecord);
 
-// Storing the int64 for a timestamp is expected to take 10 bytes (could take
-// less because of varint encoding).
-const unsigned int kBytesTimestampField = 10;
-
-struct SnapshotPackageInfo {
-    const string package;
-    const int32_t version;
-    const int32_t uid;
-    SnapshotPackageInfo(const string& package, const int32_t version, const int32_t uid)
-        : package(package), version(version), uid(uid) {
-    }
-};
-
-const unsigned int kBytesSnapshotInfo = sizeof(struct SnapshotPackageInfo);
-
-// When calling appendUidMap, we retrieve all the snapshots since the last
-// timestamp we called appendUidMap for this configuration key.
-struct SnapshotRecord {
-    const int64_t timestampNs;
-
-    // All the package info known.
-    vector<const SnapshotPackageInfo> infos;
-
-    // Tracks the number of bytes this snapshot consumes.
-    uint32_t bytes;
-
-    SnapshotRecord(const int64_t timestampNs, vector<const SnapshotPackageInfo>& infos)
-        : timestampNs(timestampNs), infos(infos) {
-        bytes = 0;
-        for (auto info : infos) {
-            bytes += info.package.size() + kBytesSnapshotInfo;
-        }
-        bytes += kBytesTimestampField;
-    }
-};
-
 // UidMap keeps track of what the corresponding app name (APK name) and version code for every uid
 // at any given moment. This map must be updated by StatsCompanionService.
 class UidMap : public virtual android::RefBase {
@@ -117,11 +86,12 @@
      * All three inputs must be the same size, and the jth element in each array refers to the same
      * tuple, ie. uid[j] corresponds to packageName[j] with versionCode[j].
      */
-    void updateMap(const vector<int32_t>& uid, const vector<int64_t>& versionCode,
-                   const vector<String16>& packageName);
+    void updateMap(const int64_t& timestamp, const vector<int32_t>& uid,
+                   const vector<int64_t>& versionCode, const vector<String16>& packageName);
 
-    void updateApp(const String16& packageName, const int32_t& uid, const int64_t& versionCode);
-    void removeApp(const String16& packageName, const int32_t& uid);
+    void updateApp(const int64_t& timestamp, const String16& packageName, const int32_t& uid,
+                   const int64_t& versionCode);
+    void removeApp(const int64_t& timestamp, const String16& packageName, const int32_t& uid);
 
     // Returns true if the given uid contains the specified app (eg. com.google.android.gms).
     bool hasApp(int uid, const string& packageName) const;
@@ -157,7 +127,8 @@
     // Gets all snapshots and changes that have occurred since the last output.
     // If every config key has received a change or snapshot record, then this
     // record is deleted.
-    void appendUidMap(const ConfigKey& key, util::ProtoOutputStream* proto);
+    void appendUidMap(const int64_t& timestamp, const ConfigKey& key,
+                      util::ProtoOutputStream* proto);
 
     // Forces the output to be cleared. We still generate a snapshot based on the current state.
     // This results in extra data uploaded but helps us reconstruct the uid mapping on the server
@@ -173,25 +144,20 @@
     std::set<string> getAppNamesFromUidLocked(const int32_t& uid, bool returnNormalized) const;
     string normalizeAppName(const string& appName) const;
 
-    void updateMap(const int64_t& timestamp, const vector<int32_t>& uid,
-                   const vector<int64_t>& versionCode, const vector<String16>& packageName);
-
-    void updateApp(const int64_t& timestamp, const String16& packageName, const int32_t& uid,
-                   const int64_t& versionCode);
-    void removeApp(const int64_t& timestamp, const String16& packageName, const int32_t& uid);
-
-    void appendUidMap(const int64_t& timestamp, const ConfigKey& key,
-                      util::ProtoOutputStream* proto);
-
     void getListenerListCopyLocked(std::vector<wp<PackageInfoListener>>* output);
 
     // TODO: Use shared_mutex for improved read-locking if a library can be found in Android.
     mutable mutex mMutex;
     mutable mutex mIsolatedMutex;
 
-    // Maps uid to application data. This must be multimap since there is a feature in Android for
-    // multiple apps to share the same uid.
-    std::unordered_multimap<int, AppData> mMap;
+    struct PairHash {
+        size_t operator()(std::pair<int, string> p) const noexcept {
+            std::hash<std::string> hash_fn;
+            return hash_fn(std::to_string(p.first) + p.second);
+        }
+    };
+    // Maps uid and package name to application data.
+    std::unordered_map<std::pair<int, string>, AppData, PairHash> mMap;
 
     // Maps isolated uid to the parent uid. Any metrics for an isolated uid will instead contribute
     // to the parent uid.
@@ -200,8 +166,8 @@
     // Record the changes that can be provided with the uploads.
     std::list<ChangeRecord> mChanges;
 
-    // Record the snapshots that can be provided with the uploads.
-    std::list<SnapshotRecord> mSnapshots;
+    // Store which uid and apps represent deleted ones.
+    std::list<std::pair<int, string>> mDeletedApps;
 
     // Metric producers that should be notified if there's an upgrade in any app.
     set<wp<PackageInfoListener>> mSubscribers;
@@ -228,6 +194,8 @@
 
     // Allows unit-test to access private methods.
     FRIEND_TEST(UidMapTest, TestClearingOutput);
+    FRIEND_TEST(UidMapTest, TestRemovedAppRetained);
+    FRIEND_TEST(UidMapTest, TestRemovedAppOverGuardrail);
     FRIEND_TEST(UidMapTest, TestOutputIncludesAtLeastOneSnapshot);
     FRIEND_TEST(UidMapTest, TestMemoryComputed);
     FRIEND_TEST(UidMapTest, TestMemoryGuardrail);
diff --git a/cmds/statsd/src/stats_log.proto b/cmds/statsd/src/stats_log.proto
index 4aa3c973..931c2d5 100644
--- a/cmds/statsd/src/stats_log.proto
+++ b/cmds/statsd/src/stats_log.proto
@@ -155,6 +155,8 @@
             optional int64 version = 2;
 
             optional int32 uid = 3;
+
+            optional bool deleted = 4;
         }
         optional int64 elapsed_timestamp_nanos = 1;
 
@@ -169,7 +171,8 @@
         optional string app = 3;
         optional int32 uid = 4;
 
-        optional int64 version = 5;
+        optional int64 new_version = 5;
+        optional int64 prev_version = 6;
     }
     repeated Change changes = 2;
 }
@@ -266,11 +269,10 @@
     repeated AtomStats atom_stats = 7;
 
     message UidMapStats {
-        optional int32 snapshots = 1;
-        optional int32 changes = 2;
-        optional int32 bytes_used = 3;
-        optional int32 dropped_snapshots = 4;
-        optional int32 dropped_changes = 5;
+        optional int32 changes = 1;
+        optional int32 bytes_used = 2;
+        optional int32 dropped_changes = 3;
+        optional int32 deleted_apps = 4;
     }
     optional UidMapStats uidmap_stats = 8;
 
diff --git a/cmds/statsd/tests/LogEntryMatcher_test.cpp b/cmds/statsd/tests/LogEntryMatcher_test.cpp
index b0da07b..4c6671d 100644
--- a/cmds/statsd/tests/LogEntryMatcher_test.cpp
+++ b/cmds/statsd/tests/LogEntryMatcher_test.cpp
@@ -146,11 +146,10 @@
     attributionMatcher->mutable_matches_tuple()->mutable_field_value_matcher(0)->set_eq_string("pkg0");
     EXPECT_FALSE(matchesSimple(uidMap, *simpleMatcher, event));
 
-    uidMap.updateMap({1111, 1111, 2222, 3333, 3333} /* uid list */,
-        {1, 1, 2, 1, 2} /* version list */,
-        {android::String16("pkg0"), android::String16("pkg1"),
-         android::String16("pkg1"), android::String16("Pkg2"),
-         android::String16("PkG3")} /* package name list */);
+    uidMap.updateMap(
+            1, {1111, 1111, 2222, 3333, 3333} /* uid list */, {1, 1, 2, 1, 2} /* version list */,
+            {android::String16("pkg0"), android::String16("pkg1"), android::String16("pkg1"),
+             android::String16("Pkg2"), android::String16("PkG3")} /* package name list */);
 
     EXPECT_TRUE(matchesSimple(uidMap, *simpleMatcher, event));
     attributionMatcher->mutable_matches_tuple()->mutable_field_value_matcher(0)
@@ -297,7 +296,7 @@
 TEST(AtomMatcherTest, TestNeqAnyStringMatcher) {
     UidMap uidMap;
     uidMap.updateMap(
-            {1111, 1111, 2222, 3333, 3333} /* uid list */, {1, 1, 2, 1, 2} /* version list */,
+            1, {1111, 1111, 2222, 3333, 3333} /* uid list */, {1, 1, 2, 1, 2} /* version list */,
             {android::String16("pkg0"), android::String16("pkg1"), android::String16("pkg1"),
              android::String16("Pkg2"), android::String16("PkG3")} /* package name list */);
 
@@ -372,7 +371,7 @@
 TEST(AtomMatcherTest, TestEqAnyStringMatcher) {
     UidMap uidMap;
     uidMap.updateMap(
-            {1111, 1111, 2222, 3333, 3333} /* uid list */, {1, 1, 2, 1, 2} /* version list */,
+            1, {1111, 1111, 2222, 3333, 3333} /* uid list */, {1, 1, 2, 1, 2} /* version list */,
             {android::String16("pkg0"), android::String16("pkg1"), android::String16("pkg1"),
              android::String16("Pkg2"), android::String16("PkG3")} /* package name list */);
 
diff --git a/cmds/statsd/tests/StatsLogProcessor_test.cpp b/cmds/statsd/tests/StatsLogProcessor_test.cpp
index e6fe3d8..09daf75 100644
--- a/cmds/statsd/tests/StatsLogProcessor_test.cpp
+++ b/cmds/statsd/tests/StatsLogProcessor_test.cpp
@@ -126,7 +126,7 @@
 TEST(StatsLogProcessorTest, TestUidMapHasSnapshot) {
     // Setup simple config key corresponding to empty config.
     sp<UidMap> m = new UidMap();
-    m->updateMap({1, 2}, {1, 2}, {String16("p1"), String16("p2")});
+    m->updateMap(1, {1, 2}, {1, 2}, {String16("p1"), String16("p2")});
     sp<AlarmMonitor> anomalyAlarmMonitor;
     sp<AlarmMonitor> subscriberAlarmMonitor;
     int broadcastCount = 0;
diff --git a/cmds/statsd/tests/UidMap_test.cpp b/cmds/statsd/tests/UidMap_test.cpp
index a9b67e0..2fab975 100644
--- a/cmds/statsd/tests/UidMap_test.cpp
+++ b/cmds/statsd/tests/UidMap_test.cpp
@@ -76,7 +76,7 @@
     apps.push_back(String16(kApp2.c_str()));
     versions.push_back(4);
     versions.push_back(5);
-    m.updateMap(uids, versions, apps);
+    m.updateMap(1, uids, versions, apps);
     EXPECT_TRUE(m.hasApp(1000, kApp1));
     EXPECT_TRUE(m.hasApp(1000, kApp2));
     EXPECT_FALSE(m.hasApp(1000, "not.app"));
@@ -102,7 +102,7 @@
     apps.push_back(String16(kApp2.c_str()));
     versions.push_back(4);
     versions.push_back(5);
-    m.updateMap(uids, versions, apps);
+    m.updateMap(1, uids, versions, apps);
 
     std::set<string> name_set = m.getAppNamesFromUid(1000, true /* returnNormalized */);
     EXPECT_EQ(name_set.size(), 2u);
@@ -110,7 +110,7 @@
     EXPECT_TRUE(name_set.find(kApp2) != name_set.end());
 
     // Update the app1 version.
-    m.updateApp(String16(kApp1.c_str()), 1000, 40);
+    m.updateApp(2, String16(kApp1.c_str()), 1000, 40);
     EXPECT_EQ(40, m.getAppVersion(1000, kApp1));
 
     name_set = m.getAppNamesFromUid(1000, true /* returnNormalized */);
@@ -118,7 +118,7 @@
     EXPECT_TRUE(name_set.find(kApp1) != name_set.end());
     EXPECT_TRUE(name_set.find(kApp2) != name_set.end());
 
-    m.removeApp(String16(kApp1.c_str()), 1000);
+    m.removeApp(3, String16(kApp1.c_str()), 1000);
     EXPECT_FALSE(m.hasApp(1000, kApp1));
     EXPECT_TRUE(m.hasApp(1000, kApp2));
     name_set = m.getAppNamesFromUid(1000, true /* returnNormalized */);
@@ -127,7 +127,7 @@
     EXPECT_TRUE(name_set.find(kApp2) != name_set.end());
 
     // Remove app2.
-    m.removeApp(String16(kApp2.c_str()), 1000);
+    m.removeApp(4, String16(kApp2.c_str()), 1000);
     EXPECT_FALSE(m.hasApp(1000, kApp1));
     EXPECT_FALSE(m.hasApp(1000, kApp2));
     name_set = m.getAppNamesFromUid(1000, true /* returnNormalized */);
@@ -136,14 +136,14 @@
 
 TEST(UidMapTest, TestUpdateApp) {
     UidMap m;
-    m.updateMap({1000, 1000}, {4, 5}, {String16(kApp1.c_str()), String16(kApp2.c_str())});
+    m.updateMap(1, {1000, 1000}, {4, 5}, {String16(kApp1.c_str()), String16(kApp2.c_str())});
     std::set<string> name_set = m.getAppNamesFromUid(1000, true /* returnNormalized */);
     EXPECT_EQ(name_set.size(), 2u);
     EXPECT_TRUE(name_set.find(kApp1) != name_set.end());
     EXPECT_TRUE(name_set.find(kApp2) != name_set.end());
 
     // Adds a new name for uid 1000.
-    m.updateApp(String16("NeW_aPP1_NAmE"), 1000, 40);
+    m.updateApp(2, String16("NeW_aPP1_NAmE"), 1000, 40);
     name_set = m.getAppNamesFromUid(1000, true /* returnNormalized */);
     EXPECT_EQ(name_set.size(), 3u);
     EXPECT_TRUE(name_set.find(kApp1) != name_set.end());
@@ -152,7 +152,7 @@
     EXPECT_TRUE(name_set.find("new_app1_name") != name_set.end());
 
     // This name is also reused by another uid 2000.
-    m.updateApp(String16("NeW_aPP1_NAmE"), 2000, 1);
+    m.updateApp(3, String16("NeW_aPP1_NAmE"), 2000, 1);
     name_set = m.getAppNamesFromUid(2000, true /* returnNormalized */);
     EXPECT_EQ(name_set.size(), 1u);
     EXPECT_TRUE(name_set.find("NeW_aPP1_NAmE") == name_set.end());
@@ -200,6 +200,66 @@
     EXPECT_EQ(1, results.snapshots_size());
 }
 
+TEST(UidMapTest, TestRemovedAppRetained) {
+    UidMap m;
+    // Initialize single config key.
+    ConfigKey config1(1, StringToId("config1"));
+    m.OnConfigUpdated(config1);
+    vector<int32_t> uids;
+    vector<int64_t> versions;
+    vector<String16> apps;
+    uids.push_back(1000);
+    apps.push_back(String16(kApp2.c_str()));
+    versions.push_back(5);
+    m.updateMap(1, uids, versions, apps);
+    m.removeApp(2, String16(kApp2.c_str()), 1000);
+
+    ProtoOutputStream proto;
+    m.appendUidMap(3, config1, &proto);
+
+    // Snapshot should still contain this item as deleted.
+    UidMapping results;
+    protoOutputStreamToUidMapping(&proto, &results);
+    EXPECT_EQ(1, results.snapshots(0).package_info_size());
+    EXPECT_EQ(true, results.snapshots(0).package_info(0).deleted());
+}
+
+TEST(UidMapTest, TestRemovedAppOverGuardrail) {
+    UidMap m;
+    // Initialize single config key.
+    ConfigKey config1(1, StringToId("config1"));
+    m.OnConfigUpdated(config1);
+    vector<int32_t> uids;
+    vector<int64_t> versions;
+    vector<String16> apps;
+    const int maxDeletedApps = StatsdStats::kMaxDeletedAppsInUidMap;
+    for (int j = 0; j < maxDeletedApps + 10; j++) {
+        uids.push_back(j);
+        apps.push_back(String16(kApp1.c_str()));
+        versions.push_back(j);
+    }
+    m.updateMap(1, uids, versions, apps);
+
+    // First, verify that we have the expected number of items.
+    UidMapping results;
+    ProtoOutputStream proto;
+    m.appendUidMap(3, config1, &proto);
+    protoOutputStreamToUidMapping(&proto, &results);
+    EXPECT_EQ(maxDeletedApps + 10, results.snapshots(0).package_info_size());
+
+    // Now remove all the apps.
+    m.updateMap(1, uids, versions, apps);
+    for (int j = 0; j < maxDeletedApps + 10; j++) {
+        m.removeApp(4, String16(kApp1.c_str()), j);
+    }
+
+    proto.clear();
+    m.appendUidMap(5, config1, &proto);
+    // Snapshot drops the first nine items.
+    protoOutputStreamToUidMapping(&proto, &results);
+    EXPECT_EQ(maxDeletedApps, results.snapshots(0).package_info_size());
+}
+
 TEST(UidMapTest, TestClearingOutput) {
     UidMap m;
 
@@ -218,7 +278,6 @@
     versions.push_back(4);
     versions.push_back(5);
     m.updateMap(1, uids, versions, apps);
-    EXPECT_EQ(1U, m.mSnapshots.size());
 
     ProtoOutputStream proto;
     m.appendUidMap(2, config1, &proto);
@@ -227,7 +286,6 @@
     EXPECT_EQ(1, results.snapshots_size());
 
     // We have to keep at least one snapshot in memory at all times.
-    EXPECT_EQ(1U, m.mSnapshots.size());
     proto.clear();
     m.appendUidMap(2, config1, &proto);
     protoOutputStreamToUidMapping(&proto, &results);
@@ -262,7 +320,6 @@
     EXPECT_EQ(1, results.snapshots_size());
     EXPECT_EQ(2, results.changes_size());
     // At this point both should be cleared.
-    EXPECT_EQ(1U, m.mSnapshots.size());
     EXPECT_EQ(0U, m.mChanges.size());
 }
 
@@ -280,11 +337,8 @@
     apps.push_back(String16(kApp1.c_str()));
     versions.push_back(1);
     m.updateMap(1, uids, versions, apps);
-    size_t snapshot_bytes = m.mBytesUsed;
-    EXPECT_TRUE(snapshot_bytes > startBytes);
 
     m.updateApp(3, String16(kApp1.c_str()), 1000, 40);
-    EXPECT_TRUE(m.mBytesUsed > snapshot_bytes);
 
     ProtoOutputStream proto;
     vector<uint8_t> bytes;
@@ -313,16 +367,13 @@
         versions.push_back(1);
     }
     m.updateMap(1, uids, versions, apps);
-    EXPECT_EQ(1U, m.mSnapshots.size());
 
     m.updateApp(3, String16("EXTREMELY_LONG_STRING_FOR_APP_TO_WASTE_MEMORY.0"), 1000, 2);
-    EXPECT_EQ(1U, m.mSnapshots.size());
     EXPECT_EQ(1U, m.mChanges.size());
 
     // Now force deletion by limiting the memory to hold one delta change.
     m.maxBytesOverride = 80; // Since the app string alone requires >45 characters.
     m.updateApp(5, String16("EXTREMELY_LONG_STRING_FOR_APP_TO_WASTE_MEMORY.0"), 1000, 4);
-    EXPECT_EQ(0U, m.mSnapshots.size());
     EXPECT_EQ(1U, m.mChanges.size());
 }
 #else
diff --git a/cmds/statsd/tests/e2e/Attribution_e2e_test.cpp b/cmds/statsd/tests/e2e/Attribution_e2e_test.cpp
index a04a6f9..a97bc41 100644
--- a/cmds/statsd/tests/e2e/Attribution_e2e_test.cpp
+++ b/cmds/statsd/tests/e2e/Attribution_e2e_test.cpp
@@ -18,6 +18,7 @@
 #include "src/stats_log_util.h"
 #include "tests/statsd_test_util.h"
 
+#include <iostream>
 #include <vector>
 
 namespace android {
@@ -66,14 +67,10 @@
     EXPECT_TRUE(processor->mMetricsManagers.begin()->second->isConfigValid());
 
     // Here it assumes that GMS core has two uids.
-    processor->getUidMap()->updateApp(
-        android::String16("com.android.gmscore"), 222 /* uid */, 1 /* version code*/);
-    processor->getUidMap()->updateApp(
-        android::String16("com.android.gmscore"), 444 /* uid */, 1 /* version code*/);
-    processor->getUidMap()->updateApp(
-        android::String16("app1"), 111 /* uid */, 2 /* version code*/);
-    processor->getUidMap()->updateApp(
-        android::String16("APP3"), 333 /* uid */, 2 /* version code*/);
+    processor->getUidMap()->updateMap(
+            1, {222, 444, 111, 333}, {1, 1, 2, 2},
+            {String16("com.android.gmscore"), String16("com.android.gmscore"), String16("app1"),
+             String16("APP3")});
 
     // GMS core node is in the middle.
     std::vector<AttributionNodeInternal> attributions1 = {CreateAttribution(111, "App1"),
@@ -212,14 +209,10 @@
     EXPECT_TRUE(processor->mMetricsManagers.begin()->second->isConfigValid());
 
     // Here it assumes that GMS core has two uids.
-    processor->getUidMap()->updateApp(
-        android::String16("com.android.gmscore"), 222 /* uid */, 1 /* version code*/);
-    processor->getUidMap()->updateApp(
-        android::String16("com.android.gmscore"), 444 /* uid */, 1 /* version code*/);
-    processor->getUidMap()->updateApp(
-        android::String16("app1"), 111 /* uid */, 2 /* version code*/);
-    processor->getUidMap()->updateApp(
-        android::String16("APP3"), 333 /* uid */, 2 /* version code*/);
+    processor->getUidMap()->updateMap(
+            1, {222, 444, 111, 333}, {1, 1, 2, 2},
+            {String16("com.android.gmscore"), String16("com.android.gmscore"), String16("app1"),
+             String16("APP3")});
 
     // GMS core node is in the middle.
     std::vector<AttributionNodeInternal> attributions1 = {CreateAttribution(111, "App1"),
diff --git a/cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp b/cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp
new file mode 100644
index 0000000..d4892ed
--- /dev/null
+++ b/cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp
@@ -0,0 +1,137 @@
+// Copyright (C) 2017 The Android Open Source Project
+//
+// Licensed 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 <gtest/gtest.h>
+
+#include "src/StatsLogProcessor.h"
+#include "src/StatsService.h"
+#include "src/stats_log_util.h"
+#include "tests/statsd_test_util.h"
+
+#include <vector>
+
+namespace android {
+namespace os {
+namespace statsd {
+
+#ifdef __ANDROID__
+
+const string kApp1 = "app1.sharing.1";
+const int kConfigKey = 789130123;  // Randomly chosen to avoid collisions with existing configs.
+
+void SendConfig(StatsService& service, const StatsdConfig& config) {
+    string str;
+    config.SerializeToString(&str);
+    std::vector<uint8_t> configAsVec(str.begin(), str.end());
+    bool success;
+    service.addConfiguration(kConfigKey, configAsVec, &success);
+}
+
+ConfigMetricsReport GetReports(StatsService& service) {
+    vector<uint8_t> output;
+    service.getData(kConfigKey, &output);
+    ConfigMetricsReportList reports;
+    reports.ParseFromArray(output.data(), output.size());
+    EXPECT_EQ(1, reports.reports_size());
+    return reports.reports(0);
+}
+
+StatsdConfig MakeConfig() {
+    StatsdConfig config;
+    config.add_allowed_log_source("AID_ROOT");  // LogEvent defaults to UID of root.
+
+    auto appCrashMatcher = CreateProcessCrashAtomMatcher();
+    *config.add_atom_matcher() = appCrashMatcher;
+    auto countMetric = config.add_count_metric();
+    countMetric->set_id(StringToId("AppCrashes"));
+    countMetric->set_what(appCrashMatcher.id());
+    countMetric->set_bucket(FIVE_MINUTES);
+    return config;
+}
+
+TEST(PartialBucketE2eTest, TestCountMetricWithoutSplit) {
+    StatsService service(nullptr);
+    SendConfig(service, MakeConfig());
+    const long start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
+                                                // initialized with.
+
+    service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 1).get());
+    service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 2).get());
+
+    ConfigMetricsReport report = GetReports(service);
+    // Expect no metrics since the bucket has not finished yet.
+    EXPECT_EQ(0, report.metrics_size());
+}
+
+TEST(PartialBucketE2eTest, TestCountMetricNoSplitOnNewApp) {
+    StatsService service(nullptr);
+    SendConfig(service, MakeConfig());
+    const long start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
+                                                // initialized with.
+
+    // Force the uidmap to update at timestamp 2.
+    service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 1).get());
+    // This is a new installation, so there shouldn't be a split (should be same as the without
+    // split case).
+    service.mUidMap->updateApp(start + 2, String16(kApp1.c_str()), 1, 2);
+    // Goes into the second bucket.
+    service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 3).get());
+
+    ConfigMetricsReport report = GetReports(service);
+    EXPECT_EQ(0, report.metrics_size());
+}
+
+TEST(PartialBucketE2eTest, TestCountMetricSplitOnUpgrade) {
+    StatsService service(nullptr);
+    SendConfig(service, MakeConfig());
+    const long start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
+                                                // initialized with.
+    service.mUidMap->updateMap(start, {1}, {1}, {String16(kApp1.c_str())});
+
+    // Force the uidmap to update at timestamp 2.
+    service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 1).get());
+    service.mUidMap->updateApp(start + 2, String16(kApp1.c_str()), 1, 2);
+    // Goes into the second bucket.
+    service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 3).get());
+
+    ConfigMetricsReport report = GetReports(service);
+    EXPECT_EQ(1, report.metrics_size());
+    EXPECT_EQ(1, report.metrics(0).count_metrics().data(0).bucket_info(0).count());
+}
+
+TEST(PartialBucketE2eTest, TestCountMetricSplitOnRemoval) {
+    StatsService service(nullptr);
+    SendConfig(service, MakeConfig());
+    const long start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
+                                                // initialized with.
+    service.mUidMap->updateMap(start, {1}, {1}, {String16(kApp1.c_str())});
+
+    // Force the uidmap to update at timestamp 2.
+    service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 1).get());
+    service.mUidMap->removeApp(start + 2, String16(kApp1.c_str()), 1);
+    // Goes into the second bucket.
+    service.mProcessor->OnLogEvent(CreateAppCrashEvent(100, start + 3).get());
+
+    ConfigMetricsReport report = GetReports(service);
+    EXPECT_EQ(1, report.metrics_size());
+    EXPECT_EQ(1, report.metrics(0).count_metrics().data(0).bucket_info(0).count());
+}
+
+#else
+GTEST_LOG_(INFO) << "This test does nothing.\n";
+#endif
+
+}  // namespace statsd
+}  // namespace os
+}  // namespace android
\ No newline at end of file
diff --git a/services/core/java/com/android/server/stats/StatsCompanionService.java b/services/core/java/com/android/server/stats/StatsCompanionService.java
index d4625e9..5f2ac4f 100644
--- a/services/core/java/com/android/server/stats/StatsCompanionService.java
+++ b/services/core/java/com/android/server/stats/StatsCompanionService.java
@@ -273,7 +273,8 @@
 
         // Add in all the apps for every user/profile.
         for (UserInfo profile : users) {
-            List<PackageInfo> pi = pm.getInstalledPackagesAsUser(0, profile.id);
+            List<PackageInfo> pi =
+                pm.getInstalledPackagesAsUser(PackageManager.MATCH_DISABLED_COMPONENTS, profile.id);
             for (int j = 0; j < pi.size(); j++) {
                 if (pi.get(j).applicationInfo != null) {
                     uids.add(pi.get(j).applicationInfo.uid);