diff --git a/lib/identity_cache/cached/attribute.rb b/lib/identity_cache/cached/attribute.rb
index 9e222b49..3afaecd9 100644
--- a/lib/identity_cache/cached/attribute.rb
+++ b/lib/identity_cache/cached/attribute.rb
@@ -47,6 +47,20 @@ def expire_for_save(record)
         end
       end
 
+      def expire_for_values(values_hash)
+        key_values = key_fields.map { |name| values_hash.fetch(name) }
+        IdentityCache.cache.delete(cache_key_from_key_values(key_values))
+      end
+
+      def expire_for_update(old_values_hash, changes)
+        expire_for_values(old_values_hash, changes)
+
+        if key_fields.any? { |name| changes.key?(name) }
+          key_values = key_fields.map { |name| changes.fetch(name) { old_values_hash.fetch(name) } }
+          IdentityCache.cache.delete(cache_key_from_key_values(key_values))
+        end
+      end
+
       def cache_key(index_key)
         values_hash = IdentityCache.memcache_hash(unhashed_values_cache_key_string(index_key))
         "#{model.rails_cache_key_namespace}#{cache_key_prefix}#{values_hash}"
diff --git a/lib/identity_cache/configuration_dsl.rb b/lib/identity_cache/configuration_dsl.rb
index bb5e65fc..f7450800 100644
--- a/lib/identity_cache/configuration_dsl.rb
+++ b/lib/identity_cache/configuration_dsl.rb
@@ -129,6 +129,7 @@ def cache_attribute_by_alias(attribute_or_proc, alias_name:, by:, unique:)
         cached_attribute = klass.new(self, attribute_or_proc, alias_name, fields, unique)
         cached_attribute.build
         cache_indexes.push(cached_attribute)
+        @cache_indexed_columns = nil
       end
 
       def ensure_base_model
diff --git a/lib/identity_cache/parent_model_expiration.rb b/lib/identity_cache/parent_model_expiration.rb
index 25fbb956..e483165d 100644
--- a/lib/identity_cache/parent_model_expiration.rb
+++ b/lib/identity_cache/parent_model_expiration.rb
@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 module IdentityCache
-  module ParentModelExpiration # :nodoc:
+  # @api private
+  module ParentModelExpiration
     extend ActiveSupport::Concern
     include ArTransactionChanges
 
@@ -35,9 +36,27 @@ def lazy_hooks
       end
     end
 
+    module ClassMethods
+      def parent_expiration_entries
+        ParentModelExpiration.install_pending_parent_expiry_hooks(cached_model)
+        _parent_expiration_entries
+      end
+
+      def check_for_unsupported_parent_expiration_entries
+        return unless parent_expiration_entries.any?
+        msg = "Unsupported manual expiration of record embedded in parent associations:\n"
+        parent_expiration_entries.each do |association_name, cached_associations|
+          cached_associations.each do |parent_class, _only_on_foreign_key_change|
+            msg << "- #{parent_class}\##{association_name}"
+          end
+        end
+        raise msg
+      end
+    end
+
     included do
-      class_attribute(:parent_expiration_entries)
-      self.parent_expiration_entries = Hash.new { |hash, key| hash[key] = [] }
+      class_attribute(:_parent_expiration_entries)
+      self._parent_expiration_entries = Hash.new { |hash, key| hash[key] = [] }
     end
 
     def expire_parent_caches
@@ -49,7 +68,6 @@ def expire_parent_caches
     end
 
     def add_parents_to_cache_expiry_set(parents_to_expire)
-      ParentModelExpiration.install_pending_parent_expiry_hooks(cached_model)
       self.class.parent_expiration_entries.each do |association_name, cached_associations|
         parents_to_expire_on_changes(parents_to_expire, association_name, cached_associations)
       end
diff --git a/lib/identity_cache/query_api.rb b/lib/identity_cache/query_api.rb
index 8bd3734f..f349fe6b 100644
--- a/lib/identity_cache/query_api.rb
+++ b/lib/identity_cache/query_api.rb
@@ -9,6 +9,55 @@ def prefetch_associations(includes, records)
         Cached::Prefetcher.prefetch(self, includes, records)
       end
 
+      # Get only the columns whose values are needed to manually expire caches
+      # after updating or deleting rows without triggering after_commit callbacks.
+      #
+      # 1. Pass the returned columns into Active Record's `select` or `pluck` query
+      #    method on the scope that will be used to modify the database in order to
+      #    query original for these rows that will be modified.
+      # 2. Update or delete the rows
+      # 3. Use {expire_cache_for_update} or {expire_cache_for_delete} to expires the
+      #    caches, passing in the values from the query in step 1 as the indexed_values.
+      #
+      # @return [Array<Symbol>] the array of column names
+      def cache_indexed_columns
+        @cache_indexed_columns ||= begin
+          check_for_unsupported_parent_expiration_entries
+          columns = Set.new
+          columns << primary_key.to_sym if primary_cache_index_enabled
+          cache_indexes.each do |cached_attribute|
+            columns.merge(cached_attribute.key_fields)
+          end
+          columns.to_a.freeze
+        end
+      end
+
+      def expire_cache_for_update(old_indexed_values, changes)
+        if primary_cache_index_enabled
+          id = old_indexed_values.fetch(primary_key.to_sym)
+          expire_primary_key_cache_index(id)
+        end
+        cache_indexes.each do |cached_attribute|
+          cached_attribute.expire_for_update(old_indexed_values, changes)
+        end
+        check_for_unsupported_parent_expiration_entries
+      end
+
+      private def expire_cache_for_insert_or_delete(indexed_values)
+        if primary_cache_index_enabled
+          id = indexed_values.fetch(primary_key.to_sym)
+          expire_primary_key_cache_index(id)
+        end
+        cache_indexes.each do |cached_attribute|
+          cached_attribute.expire_for_values(indexed_values)
+        end
+        check_for_unsupported_parent_expiration_entries
+      end
+
+      alias_method :expire_cache_for_insert, :expire_cache_for_insert_or_delete
+
+      alias_method :expire_cache_for_delete, :expire_cache_for_insert_or_delete
+
       # @api private
       def cached_association(name) # :nodoc:
         cached_has_manys[name] || cached_has_ones[name] || cached_belongs_tos.fetch(name)