diff --git a/crates/bevy_ecs/src/schedule/condition.rs b/crates/bevy_ecs/src/schedule/condition.rs
index 939d04b19114e..5a93006a44464 100644
--- a/crates/bevy_ecs/src/schedule/condition.rs
+++ b/crates/bevy_ecs/src/schedule/condition.rs
@@ -1,6 +1,11 @@
+use std::any::TypeId;
 use std::borrow::Cow;
+use std::ops::Not;
 
+use crate::component::{self, ComponentId};
+use crate::query::Access;
 use crate::system::{CombinatorSystem, Combine, IntoSystem, ReadOnlySystem, System};
+use crate::world::World;
 
 pub type BoxedCondition = Box<dyn ReadOnlySystem<In = (), Out = bool>>;
 
@@ -131,13 +136,15 @@ mod sealed {
 }
 
 pub mod common_conditions {
-    use super::Condition;
+    use std::borrow::Cow;
+
+    use super::NotSystem;
     use crate::{
         change_detection::DetectChanges,
         event::{Event, EventReader},
         prelude::{Component, Query, With},
         schedule::{State, States},
-        system::{In, IntoPipeSystem, Res, Resource},
+        system::{IntoSystem, Res, Resource, System},
     };
 
     /// Generates a [`Condition`](super::Condition)-satisfying closure that returns `true`
@@ -915,12 +922,103 @@ pub mod common_conditions {
     /// app.run(&mut world);
     /// assert_eq!(world.resource::<Counter>().0, 0);
     /// ```
-    pub fn not<Marker, T>(condition: T) -> impl Condition<()>
+    pub fn not<Marker, TOut, T>(condition: T) -> NotSystem<T::System>
     where
-        T: Condition<Marker>,
+        TOut: std::ops::Not,
+        T: IntoSystem<(), TOut, Marker>,
     {
-        condition.pipe(|In(val): In<bool>| !val)
+        let condition = IntoSystem::into_system(condition);
+        let name = format!("!{}", condition.name());
+        NotSystem::<T::System> {
+            condition,
+            name: Cow::Owned(name),
+        }
+    }
+}
+
+/// Invokes [`Not`] with the output of another system.
+///
+/// See [`common_conditions::not`] for examples.
+#[derive(Clone)]
+pub struct NotSystem<T: System>
+where
+    T::Out: Not,
+{
+    condition: T,
+    name: Cow<'static, str>,
+}
+impl<T: System> System for NotSystem<T>
+where
+    T::Out: Not,
+{
+    type In = T::In;
+    type Out = <T::Out as Not>::Output;
+
+    fn name(&self) -> Cow<'static, str> {
+        self.name.clone()
+    }
+
+    fn type_id(&self) -> TypeId {
+        TypeId::of::<T>()
+    }
+
+    fn component_access(&self) -> &Access<ComponentId> {
+        self.condition.component_access()
+    }
+
+    fn archetype_component_access(&self) -> &Access<crate::archetype::ArchetypeComponentId> {
+        self.condition.archetype_component_access()
+    }
+
+    fn is_send(&self) -> bool {
+        self.condition.is_send()
+    }
+
+    fn is_exclusive(&self) -> bool {
+        self.condition.is_exclusive()
+    }
+
+    unsafe fn run_unsafe(&mut self, input: Self::In, world: &World) -> Self::Out {
+        // SAFETY: The inner condition system asserts its own safety.
+        !self.condition.run_unsafe(input, world)
+    }
+
+    fn run(&mut self, input: Self::In, world: &mut World) -> Self::Out {
+        !self.condition.run(input, world)
     }
+
+    fn apply_buffers(&mut self, world: &mut World) {
+        self.condition.apply_buffers(world);
+    }
+
+    fn initialize(&mut self, world: &mut World) {
+        self.condition.initialize(world);
+    }
+
+    fn update_archetype_component_access(&mut self, world: &World) {
+        self.condition.update_archetype_component_access(world);
+    }
+
+    fn check_change_tick(&mut self, change_tick: component::Tick) {
+        self.condition.check_change_tick(change_tick);
+    }
+
+    fn get_last_run(&self) -> component::Tick {
+        self.condition.get_last_run()
+    }
+
+    fn set_last_run(&mut self, last_run: component::Tick) {
+        self.condition.set_last_run(last_run);
+    }
+}
+
+// SAFETY: This trait is only implemented when the inner system is read-only.
+// The `Not` condition does not modify access, and thus cannot transform a read-only system into one that is not.
+unsafe impl<T> ReadOnlySystem for NotSystem<T>
+where
+    T: ReadOnlySystem,
+    T::Out: Not,
+{
 }
 
 /// Combines the outputs of two systems using the `&&` operator.
@@ -973,12 +1071,17 @@ where
 
 #[cfg(test)]
 mod tests {
-    use super::Condition;
+    use super::{common_conditions::*, Condition};
     use crate as bevy_ecs;
-    use crate::schedule::common_conditions::not;
-    use crate::schedule::IntoSystemConfig;
-    use crate::system::Local;
-    use crate::{change_detection::ResMut, schedule::Schedule, world::World};
+    use crate::{
+        change_detection::ResMut,
+        component::Component,
+        schedule::{
+            common_conditions::not, IntoSystemConfig, IntoSystemConfigs, Schedule, State, States,
+        },
+        system::Local,
+        world::World,
+    };
     use bevy_ecs_macros::Resource;
 
     #[derive(Resource, Default)]
@@ -1074,4 +1177,37 @@ mod tests {
         schedule.run(&mut world);
         assert_eq!(world.resource::<Counter>().0, 0);
     }
+
+    #[derive(States, PartialEq, Eq, Debug, Default, Hash, Clone)]
+    enum TestState {
+        #[default]
+        A,
+        B,
+    }
+
+    #[derive(Component)]
+    struct TestComponent;
+
+    fn test_system() {}
+
+    // Ensure distributive_run_if compiles with the common conditions.
+    #[test]
+    fn distributive_run_if_compiles() {
+        Schedule::default().add_systems(
+            (test_system, test_system)
+                .distributive_run_if(run_once())
+                .distributive_run_if(resource_exists::<State<TestState>>())
+                .distributive_run_if(resource_added::<State<TestState>>())
+                .distributive_run_if(resource_changed::<State<TestState>>())
+                .distributive_run_if(resource_exists_and_changed::<State<TestState>>())
+                .distributive_run_if(resource_changed_or_removed::<State<TestState>>())
+                .distributive_run_if(resource_removed::<State<TestState>>())
+                .distributive_run_if(state_exists::<TestState>())
+                .distributive_run_if(in_state(TestState::A))
+                .distributive_run_if(state_changed::<TestState>())
+                .distributive_run_if(on_event::<u8>())
+                .distributive_run_if(any_with_component::<TestComponent>())
+                .distributive_run_if(not(run_once())),
+        );
+    }
 }
diff --git a/crates/bevy_ecs/src/schedule/config.rs b/crates/bevy_ecs/src/schedule/config.rs
index 54eb9f863f5b7..ce0a45ae44b23 100644
--- a/crates/bevy_ecs/src/schedule/config.rs
+++ b/crates/bevy_ecs/src/schedule/config.rs
@@ -394,8 +394,8 @@ where
 
     /// Add a run condition to each contained system.
     ///
-    /// Each system will receive its own clone of the [`Condition`] and will only run
-    /// if the `Condition` is true.
+    /// The [`Condition`] must be [`Clone`]. Each system will receive its own clone
+    /// of the `Condition` and will only run if the `Condition` is true.
     ///
     /// Each individual condition will be evaluated at most once (per schedule run),
     /// right before the corresponding system prepares to run.
@@ -422,6 +422,9 @@ where
     /// Use [`run_if`](IntoSystemSetConfig::run_if) on a [`SystemSet`] if you want to make sure
     /// that either all or none of the systems are run, or you don't want to evaluate the run
     /// condition for each contained system separately.
+    ///
+    /// The [`Condition`] is cloned for each system.
+    /// Cloned instances of [`FunctionSystem`](crate::system::FunctionSystem) will be de-initialized.
     fn distributive_run_if<M>(self, condition: impl Condition<M> + Clone) -> SystemConfigs {
         self.into_configs().distributive_run_if(condition)
     }
diff --git a/crates/bevy_ecs/src/system/function_system.rs b/crates/bevy_ecs/src/system/function_system.rs
index 0e88ad0187af4..d8bf01392ff4a 100644
--- a/crates/bevy_ecs/src/system/function_system.rs
+++ b/crates/bevy_ecs/src/system/function_system.rs
@@ -367,6 +367,9 @@ pub struct In<In>(pub In);
 /// becomes the functions [`In`] tagged parameter or `()` if no such parameter exists.
 ///
 /// [`FunctionSystem`] must be `.initialized` before they can be run.
+///
+/// The [`Clone`] implementation for [`FunctionSystem`] returns a new instance which
+/// is NOT initialized. The cloned system must also be `.initialized` before it can be run.
 pub struct FunctionSystem<Marker, F>
 where
     F: SystemParamFunction<Marker>,
@@ -380,6 +383,23 @@ where
     marker: PhantomData<fn() -> Marker>,
 }
 
+// De-initializes the cloned system.
+impl<Marker, F> Clone for FunctionSystem<Marker, F>
+where
+    F: SystemParamFunction<Marker> + Clone,
+{
+    fn clone(&self) -> Self {
+        Self {
+            func: self.func.clone(),
+            param_state: None,
+            system_meta: SystemMeta::new::<F>(),
+            world_id: None,
+            archetype_generation: ArchetypeGeneration::initial(),
+            marker: PhantomData,
+        }
+    }
+}
+
 pub struct IsFunctionSystem;
 
 impl<Marker, F> IntoSystem<F::In, F::Out, (IsFunctionSystem, Marker)> for F