diff --git a/CHANGELOG.md b/CHANGELOG.md
index c67227d993..90415ec291 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,7 @@
 * None.
 
 ### Enhancements
+* Added support for keypaths in `asFlow()` methods on objects and queries. This makes it possible to control which properties will trigger change events, including properties on objects below the default nested limit of 4. (Issue [#661](https://github.com/realm/realm-kotlin/issues/661))
 * Support for experimental K2-compilation with `kotlin.experimental.tryK2=true`. (Issue [#1483](https://github.com/realm/realm-kotlin/issues/1483))
 
 ### Fixed
@@ -25,7 +26,7 @@
 * Minimum R8: 8.0.34.
 
 ### Internal
-* None.
+* Updated to Realm Core 13.24.0, commit e593a5f19d0dc205db931ec5618a8c10c95cac90.
 
 
 ## 1.12.1-SNAPSHOT (YYYY-MM-DD)
diff --git a/packages/cinterop/src/commonMain/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt b/packages/cinterop/src/commonMain/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt
index 293190fe4d..b0e3447796 100644
--- a/packages/cinterop/src/commonMain/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt
+++ b/packages/cinterop/src/commonMain/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt
@@ -69,6 +69,7 @@ interface RealmCallbackTokenT : CapiT
 interface RealmNotificationTokenT : CapiT
 interface RealmChangesT : CapiT
 interface RealmSchedulerT : CapiT
+interface RealmKeyPathArrayT : CapiT
 
 // Public type aliases binding to internal verbose type safe type definitions. This should allow us
 // to easily change implementation details later on.
@@ -88,6 +89,7 @@ typealias RealmCallbackTokenPointer = NativePointer<RealmCallbackTokenT>
 typealias RealmNotificationTokenPointer = NativePointer<RealmNotificationTokenT>
 typealias RealmChangesPointer = NativePointer<RealmChangesT>
 typealias RealmSchedulerPointer = NativePointer<RealmSchedulerT>
+typealias RealmKeyPathArrayPointer = NativePointer<RealmKeyPathArrayT>
 
 // Sync types
 // Pure marker interfaces corresponding to the C-API realm_x_t struct types
@@ -435,24 +437,30 @@ expect object RealmInterop {
     ): RealmObjectPointer?
     fun realm_object_delete(obj: RealmObjectPointer)
 
+    fun realm_create_key_paths_array(realm: RealmPointer, clazz: ClassKey, keyPaths: List<String>): RealmKeyPathArrayPointer
     fun realm_object_add_notification_callback(
         obj: RealmObjectPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer
     fun realm_results_add_notification_callback(
         results: RealmResultsPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer
     fun realm_list_add_notification_callback(
         list: RealmListPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer
     fun realm_set_add_notification_callback(
         set: RealmSetPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer
     fun realm_dictionary_add_notification_callback(
         map: RealmMapPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer
     fun realm_object_changes_get_modified_properties(
diff --git a/packages/cinterop/src/jvm/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt b/packages/cinterop/src/jvm/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt
index 754b1dba70..53f18a3bed 100644
--- a/packages/cinterop/src/jvm/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt
+++ b/packages/cinterop/src/jvm/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt
@@ -41,6 +41,10 @@ import org.mongodb.kbson.ObjectId
 actual val INVALID_CLASS_KEY: ClassKey by lazy { ClassKey(realmc.getRLM_INVALID_CLASS_KEY()) }
 actual val INVALID_PROPERTY_KEY: PropertyKey by lazy { PropertyKey(realmc.getRLM_INVALID_PROPERTY_KEY()) }
 
+// The value to pass to JNI functions that accept longs as replacements for pointers and need
+// to represent null.
+const val NULL_POINTER_VALUE = 0L
+
 /**
  * JVM/Android interop implementation.
  *
@@ -813,14 +817,22 @@ actual object RealmInterop {
         return realmc.realm_dictionary_is_valid(dictionary.cptr())
     }
 
+    actual fun realm_create_key_paths_array(realm: RealmPointer, clazz: ClassKey, keyPaths: List<String>): RealmKeyPathArrayPointer {
+        val ptr = realmc.realm_create_key_path_array(realm.cptr(), clazz.key, keyPaths.size.toLong(), keyPaths.toTypedArray())
+        return LongPointerWrapper(ptr)
+    }
+
     actual fun realm_object_add_notification_callback(
         obj: RealmObjectPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
+
         return LongPointerWrapper(
             realmc.register_notification_cb(
                 obj.cptr(),
                 CollectionType.RLM_COLLECTION_TYPE_NONE.nativeValue,
+                keyPaths?.cptr() ?: NULL_POINTER_VALUE,
                 object : NotificationCallback {
                     override fun onChange(pointer: Long) {
                         callback.onChange(LongPointerWrapper(realmc.realm_clone(pointer), true))
@@ -833,11 +845,13 @@ actual object RealmInterop {
 
     actual fun realm_results_add_notification_callback(
         results: RealmResultsPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
         return LongPointerWrapper(
             realmc.register_results_notification_cb(
                 results.cptr(),
+                keyPaths?.cptr() ?: NULL_POINTER_VALUE,
                 object : NotificationCallback {
                     override fun onChange(pointer: Long) {
                         callback.onChange(LongPointerWrapper(realmc.realm_clone(pointer), true))
@@ -850,12 +864,14 @@ actual object RealmInterop {
 
     actual fun realm_list_add_notification_callback(
         list: RealmListPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
         return LongPointerWrapper(
             realmc.register_notification_cb(
                 list.cptr(),
                 CollectionType.RLM_COLLECTION_TYPE_LIST.nativeValue,
+                keyPaths?.cptr() ?: NULL_POINTER_VALUE,
                 object : NotificationCallback {
                     override fun onChange(pointer: Long) {
                         callback.onChange(LongPointerWrapper(realmc.realm_clone(pointer), true))
@@ -868,12 +884,14 @@ actual object RealmInterop {
 
     actual fun realm_set_add_notification_callback(
         set: RealmSetPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
         return LongPointerWrapper(
             realmc.register_notification_cb(
                 set.cptr(),
                 CollectionType.RLM_COLLECTION_TYPE_SET.nativeValue,
+                keyPaths?.cptr() ?: NULL_POINTER_VALUE,
                 object : NotificationCallback {
                     override fun onChange(pointer: Long) {
                         callback.onChange(LongPointerWrapper(realmc.realm_clone(pointer), true))
@@ -886,12 +904,14 @@ actual object RealmInterop {
 
     actual fun realm_dictionary_add_notification_callback(
         map: RealmMapPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
         return LongPointerWrapper(
             realmc.register_notification_cb(
                 map.cptr(),
                 CollectionType.RLM_COLLECTION_TYPE_DICTIONARY.nativeValue,
+                keyPaths?.cptr() ?: NULL_POINTER_VALUE,
                 object : NotificationCallback {
                     override fun onChange(pointer: Long) {
                         callback.onChange(LongPointerWrapper(realmc.realm_clone(pointer), true))
diff --git a/packages/cinterop/src/nativeDarwin/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt b/packages/cinterop/src/nativeDarwin/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt
index 213be632f4..914d9cb6fb 100644
--- a/packages/cinterop/src/nativeDarwin/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt
+++ b/packages/cinterop/src/nativeDarwin/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt
@@ -45,6 +45,7 @@ import kotlinx.cinterop.COpaquePointer
 import kotlinx.cinterop.CPointed
 import kotlinx.cinterop.CPointer
 import kotlinx.cinterop.CPointerVar
+import kotlinx.cinterop.CPointerVarOf
 import kotlinx.cinterop.CValue
 import kotlinx.cinterop.CVariable
 import kotlinx.cinterop.LongVar
@@ -70,6 +71,7 @@ import kotlinx.cinterop.readValue
 import kotlinx.cinterop.refTo
 import kotlinx.cinterop.set
 import kotlinx.cinterop.staticCFunction
+import kotlinx.cinterop.toCStringArray
 import kotlinx.cinterop.toKString
 import kotlinx.cinterop.useContents
 import kotlinx.cinterop.usePinned
@@ -1653,8 +1655,17 @@ actual object RealmInterop {
         checkedBooleanResult(realm_wrapper.realm_object_delete(obj.cptr()))
     }
 
+    actual fun realm_create_key_paths_array(realm: RealmPointer, clazz: ClassKey, keyPaths: List<String>): RealmKeyPathArrayPointer {
+        memScoped {
+            val userKeyPaths: CPointer<CPointerVarOf<CPointer<ByteVarOf<Byte>>>> = keyPaths.toCStringArray(this)
+            val keyPathPointer = realm_wrapper.realm_create_key_path_array(realm.cptr(), clazz.key.toUInt(), keyPaths.size.toULong(), userKeyPaths)
+            return CPointerWrapper(keyPathPointer)
+        }
+    }
+
     actual fun realm_object_add_notification_callback(
         obj: RealmObjectPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
         return CPointerWrapper(
@@ -1667,7 +1678,7 @@ actual object RealmInterop {
                         ?.dispose()
                         ?: error("Notification callback data should never be null")
                 },
-                null, // See https://github.com/realm/realm-kotlin/issues/661
+                keyPaths?.cptr(),
                 staticCFunction { userdata, change -> // Change callback
                     try {
                         userdata?.asStableRef<Callback<RealmChangesPointer>>()
@@ -1688,6 +1699,7 @@ actual object RealmInterop {
 
     actual fun realm_results_add_notification_callback(
         results: RealmResultsPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
         return CPointerWrapper(
@@ -1700,7 +1712,7 @@ actual object RealmInterop {
                         ?.dispose()
                         ?: error("Notification callback data should never be null")
                 },
-                null, // See https://github.com/realm/realm-kotlin/issues/661
+                keyPaths?.cptr(),
                 staticCFunction { userdata, change -> // Change callback
                     try {
                         userdata?.asStableRef<Callback<RealmChangesPointer>>()
@@ -1721,6 +1733,7 @@ actual object RealmInterop {
 
     actual fun realm_list_add_notification_callback(
         list: RealmListPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
         return CPointerWrapper(
@@ -1732,7 +1745,7 @@ actual object RealmInterop {
                     userdata?.asStableRef<Callback<RealmChangesPointer>>()?.dispose()
                         ?: error("Notification callback data should never be null")
                 },
-                null, // See https://github.com/realm/realm-kotlin/issues/661
+                keyPaths?.cptr(),
                 staticCFunction { userdata, change -> // Change callback
                     try {
                         userdata?.asStableRef<Callback<RealmChangesPointer>>()
@@ -1753,6 +1766,7 @@ actual object RealmInterop {
 
     actual fun realm_set_add_notification_callback(
         set: RealmSetPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
         return CPointerWrapper(
@@ -1765,7 +1779,7 @@ actual object RealmInterop {
                         ?.dispose()
                         ?: error("Notification callback data should never be null")
                 },
-                null, // See https://github.com/realm/realm-kotlin/issues/661
+                keyPaths?.cptr(),
                 staticCFunction { userdata, change -> // Change callback
                     try {
                         userdata?.asStableRef<Callback<RealmChangesPointer>>()
@@ -1786,6 +1800,7 @@ actual object RealmInterop {
 
     actual fun realm_dictionary_add_notification_callback(
         map: RealmMapPointer,
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
         return CPointerWrapper(
@@ -1798,7 +1813,7 @@ actual object RealmInterop {
                         ?.dispose()
                         ?: error("Notification callback data should never be null")
                 },
-                null, // See https://github.com/realm/realm-kotlin/issues/661
+                keyPaths?.cptr(),
                 staticCFunction { userdata, change -> // Change callback
                     try {
                         userdata?.asStableRef<Callback<RealmChangesPointer>>()
diff --git a/packages/external/core b/packages/external/core
index 7556b535aa..e593a5f19d 160000
--- a/packages/external/core
+++ b/packages/external/core
@@ -1 +1 @@
-Subproject commit 7556b535aa7b27d49c13444894f7e9db778b3203
+Subproject commit e593a5f19d0dc205db931ec5618a8c10c95cac90
diff --git a/packages/jni-swig-stub/realm.i b/packages/jni-swig-stub/realm.i
index 8bc6d3b223..fcaae087eb 100644
--- a/packages/jni-swig-stub/realm.i
+++ b/packages/jni-swig-stub/realm.i
@@ -301,7 +301,8 @@ return $jnicall;
                realm_flx_sync_mutable_subscription_set_t*, realm_flx_sync_subscription_desc_t*,
                realm_set_t*, realm_async_open_task_t*, realm_dictionary_t*,
                realm_sync_session_connection_state_notification_token_t*,
-               realm_dictionary_changes_t*, realm_scheduler_t*, realm_sync_socket_t* };
+               realm_dictionary_changes_t*, realm_scheduler_t*, realm_sync_socket_t*,
+               realm_key_path_array_t* };
 
 // For all functions returning a pointer or bool, check for null/false and throw an error if
 // realm_get_last_error returns true.
@@ -397,6 +398,66 @@ import static io.realm.kotlin.internal.interop.realm_errno_e.*;
 %include "enumtypeunsafe.swg"
 %javaconst(1);
 
+// Add support for String[] vs char** conversion
+// See https://www.swig.org/Doc4.0/Java.html#Java_converting_java_string_arrays
+// Begin --
+
+/* This tells SWIG to treat char ** as a special case when used as a parameter
+   in a function call */
+%typemap(in) char ** (jint size) {
+    int i = 0;
+    size = jenv->GetArrayLength($input);
+    $1 = (char **) malloc((size+1)*sizeof(char *));
+    /* make a copy of each string */
+    for (i = 0; i<size; i++) {
+        jstring j_string = (jstring)jenv->GetObjectArrayElement($input, i);
+        const char * c_string = jenv->GetStringUTFChars(j_string, 0);
+        $1[i] = (char*) malloc((strlen(c_string)+1)*sizeof(char));
+        strcpy($1[i], c_string);
+        jenv->ReleaseStringUTFChars(j_string, c_string);
+        jenv->DeleteLocalRef(j_string);
+    }
+    $1[i] = 0;
+}
+
+/* This cleans up the memory we malloc'd before the function call */
+%typemap(freearg) char ** {
+    int i;
+    for (i=0; i<size$argnum-1; i++) {
+        free($1[i]);
+    }
+    free($1);
+}
+
+/* This allows a C function to return a char ** as a Java String array */
+%typemap(out) char ** {
+    int i;
+    int len=0;
+    jstring temp_string;
+    const jclass clazz = jenv->FindClass("java/lang/String");
+
+    while ($1[len]) len++;
+    jresult = jenv->NewObjectArray(len, clazz, NULL);
+    /* exception checking omitted */
+    for (i=0; i<len; i++) {
+        temp_string = (*jenv)->NewStringUTF(*result++);
+        jenv->SetObjectArrayElement(jresult, i, temp_string);
+        jenv->DeleteLocalRef(temp_string);
+    }
+}
+
+/* These 3 typemaps tell SWIG what JNI and Java types to use */
+%typemap(jni) char ** "jobjectArray"
+%typemap(jtype) char ** "String[]"
+%typemap(jstype) char ** "String[]"
+
+/* These 2 typemaps handle the conversion of the jtype to jstype typemap type
+   and vice versa */
+%typemap(javain) char ** "$javainput"
+%typemap(javaout) char ** {
+    return $jnicall;
+}
+// -- End
 
 // FIXME OPTIMIZE Support getting/setting multiple attributes. Ignored for now due to incorrect
 //  type cast in Swig-generated wrapper for "const realm_property_key_t*" which is not cast
@@ -430,6 +491,12 @@ import static io.realm.kotlin.internal.interop.realm_errno_e.*;
 //  realm_convert_with_path.
 %ignore realm_convert_with_path;
 
+%ignore "realm_object_add_notification_callback";
+%ignore "realm_list_add_notification_callback";
+%ignore "realm_set_add_notification_callback";
+%ignore "realm_dictionary_add_notification_callback";
+%ignore "realm_results_add_notification_callback";
+
 // Swig doesn't understand __attribute__ so eliminate it
 #define __attribute__(x)
 
diff --git a/packages/jni-swig-stub/src/main/jni/realm_api_helpers.cpp b/packages/jni-swig-stub/src/main/jni/realm_api_helpers.cpp
index 03783aaf80..b047236a6f 100644
--- a/packages/jni-swig-stub/src/main/jni/realm_api_helpers.cpp
+++ b/packages/jni-swig-stub/src/main/jni/realm_api_helpers.cpp
@@ -143,7 +143,9 @@ bool migration_callback(void *userdata, realm_t *old_realm, realm_t *new_realm,
 // TODO OPTIMIZE Abstract pattern for all notification registrations for collections that receives
 //  changes as realm_collection_changes_t.
 realm_notification_token_t *
-register_results_notification_cb(realm_results_t *results, jobject callback) {
+register_results_notification_cb(realm_results_t *results,
+                                 int64_t key_path_array_ptr,
+                                 jobject callback) {
     auto jenv = get_env();
     static jclass notification_class = jenv->FindClass("io/realm/kotlin/internal/interop/NotificationCallback");
     static jmethodID on_change_method = jenv->GetMethodID(notification_class, "onChange", "(J)V");
@@ -155,7 +157,7 @@ register_results_notification_cb(realm_results_t *results, jobject callback) {
             [](void *userdata) {
                 get_env(true)->DeleteGlobalRef(static_cast<jobject>(userdata));
             },
-            NULL, // See https://github.com/realm/realm-kotlin/issues/661
+            reinterpret_cast<realm_key_path_array_t*>(key_path_array_ptr),
             // change callback
             [](void *userdata, const realm_collection_changes_t *changes) {
                 // TODO API-NOTIFICATION Consider catching errors and propagate to error callback
@@ -219,40 +221,45 @@ realm_on_dictionary_change_func_t get_on_dictionary_change() {
 }
 
 realm_notification_token_t *
-register_notification_cb(int64_t collection_ptr, realm_collection_type_e collection_type,
-                         jobject callback) {
+register_notification_cb(
+        int64_t collection_ptr,
+        realm_collection_type_e collection_type,
+        int64_t key_path_array_ptr,
+        jobject callback
+) {
     auto user_data = static_cast<jobject>(get_env()->NewGlobalRef(callback));
     auto user_data_free = [](void *userdata) {
         get_env(true)->DeleteGlobalRef(static_cast<jobject>(userdata));
     };
 
     switch (collection_type) {
-        case RLM_COLLECTION_TYPE_NONE: return realm_object_add_notification_callback(
+        case RLM_COLLECTION_TYPE_NONE:
+            return realm_object_add_notification_callback(
                     reinterpret_cast<realm_object_t*>(collection_ptr),
                     user_data, // Use the callback as user data
                     user_data_free,
-                    NULL, // See https://github.com/realm/realm-kotlin/issues/661
+                    (key_path_array_ptr == 0) ? NULL : reinterpret_cast<realm_key_path_array_t*>(key_path_array_ptr),
                     get_on_object_change()
             );
         case RLM_COLLECTION_TYPE_LIST: return realm_list_add_notification_callback(
                     reinterpret_cast<realm_list_t*>(collection_ptr),
                     user_data, // Use the callback as user data
                     user_data_free,
-                    NULL, // See https://github.com/realm/realm-kotlin/issues/661
+                    reinterpret_cast<realm_key_path_array_t*>(key_path_array_ptr),
                     get_on_collection_change()
             );
         case RLM_COLLECTION_TYPE_SET: return realm_set_add_notification_callback(
                     reinterpret_cast<realm_set_t*>(collection_ptr),
                     user_data, // Use the callback as user data
                     user_data_free,
-                    NULL, // See https://github.com/realm/realm-kotlin/issues/661
+                    reinterpret_cast<realm_key_path_array_t*>(key_path_array_ptr),
                     get_on_collection_change()
             );
         case RLM_COLLECTION_TYPE_DICTIONARY: return realm_dictionary_add_notification_callback(
                     reinterpret_cast<realm_dictionary_t*>(collection_ptr),
                     user_data, // Use the callback as user data
                     user_data_free,
-                    NULL, // See https://github.com/realm/realm-kotlin/issues/661
+                    reinterpret_cast<realm_key_path_array_t*>(key_path_array_ptr),
                     get_on_dictionary_change()
             );
     }
diff --git a/packages/jni-swig-stub/src/main/jni/realm_api_helpers.h b/packages/jni-swig-stub/src/main/jni/realm_api_helpers.h
index 022bca21a7..4f466beef2 100644
--- a/packages/jni-swig-stub/src/main/jni/realm_api_helpers.h
+++ b/packages/jni-swig-stub/src/main/jni/realm_api_helpers.h
@@ -35,11 +35,17 @@ migration_callback(void* userdata, realm_t* old_realm, realm_t* new_realm,
                    const realm_schema_t* schema);
 
 realm_notification_token_t*
-register_results_notification_cb(realm_results_t *results, jobject callback);
+register_results_notification_cb(
+        realm_results_t *results,
+        int64_t key_path_array_ptr,
+        jobject callback);
 
 realm_notification_token_t *
-register_notification_cb(int64_t collection_ptr, realm_collection_type_e collection_type,
-                         jobject callback);
+register_notification_cb(
+        int64_t collection_ptr,
+        realm_collection_type_e collection_type,
+        int64_t key_path_array_ptr,
+        jobject callback);
 
 realm_http_transport_t*
 realm_network_transport_new(jobject network_transport);
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/ext/BaseRealmObjectExt.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/ext/BaseRealmObjectExt.kt
index 544a7d909b..508d232790 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/ext/BaseRealmObjectExt.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/ext/BaseRealmObjectExt.kt
@@ -83,12 +83,22 @@ public fun BaseRealmObject.isValid(): Boolean = runIfManaged {
  * the elements in a timely manner the coroutine scope will be cancelled with a
  * [CancellationException].
  *
+ * @param keyPaths An optional list of properties that defines when a change to the object will
+ * result in a change being emitted. Nested properties can be defined using a dotted
+ * syntax, e.g. `parent.child.name`. Wildcards `*` can be be used to capture all properties at a
+ * given level, e.g. `child.*` or `*.*`. If no keypaths are provided, changes to all top-level
+ * properties and nested properties 4 levels down will trigger a change.
  * @return a flow representing changes to the object.
  * @throws UnsupportedOperationException if called on a live [RealmObject] or [EmbeddedRealmObject]
  * from a write transaction ([Realm.write]) or on a [DynamicRealmObject] inside a migration
  * ([AutomaticSchemaMigration.migrate]).
+ * @throws IllegalArgumentException if an invalid keypath is provided.
  */
-public fun <T : BaseRealmObject> T.asFlow(): Flow<ObjectChange<T>> = runIfManaged {
+public fun <T : BaseRealmObject> T.asFlow(keyPaths: List<String>? = null): Flow<ObjectChange<T>> = runIfManaged {
     checkNotificationsAvailable()
-    return owner.owner.registerObserver(this) as Flow<ObjectChange<T>>
+    val keyPathInfo = keyPaths?.let {
+        Pair(this.metadata.classKey, keyPaths)
+    }
+
+    return owner.owner.registerObserver(this, keyPathInfo) as Flow<ObjectChange<T>>
 } ?: throw IllegalStateException("Changes cannot be observed on unmanaged objects.")
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/BaseRealmImpl.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/BaseRealmImpl.kt
index 44c401ba5b..cb8b3372cc 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/BaseRealmImpl.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/BaseRealmImpl.kt
@@ -16,6 +16,7 @@
 package io.realm.kotlin.internal
 
 import io.realm.kotlin.BaseRealm
+import io.realm.kotlin.internal.interop.ClassKey
 import io.realm.kotlin.internal.interop.RealmInterop
 import io.realm.kotlin.notifications.internal.Callback
 import io.realm.kotlin.notifications.internal.Cancellable
@@ -64,7 +65,7 @@ public abstract class BaseRealmImpl internal constructor(
         return RealmInterop.realm_get_schema_version(realmReference.dbPointer)
     }
 
-    internal open fun <T : CoreNotifiable<T, C>, C> registerObserver(t: Observable<T, C>): Flow<C> {
+    internal open fun <T : CoreNotifiable<T, C>, C> registerObserver(t: Observable<T, C>, keyPaths: Pair<ClassKey, List<String>>?): Flow<C> {
         throw UnsupportedOperationException(OBSERVABLE_NOT_SUPPORTED_MESSAGE)
     }
 
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/Flowable.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/KeyPathFlowable.kt
similarity index 65%
rename from packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/Flowable.kt
rename to packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/KeyPathFlowable.kt
index 0bc81a0c15..ecf54903fb 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/Flowable.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/KeyPathFlowable.kt
@@ -21,7 +21,13 @@ import kotlinx.coroutines.flow.Flow
 /**
  * A __flowable__ is an internal entity that supports listening to changes on the type [T] as a
  * [Flow].
+ *
+ * This comes in two variants: One variant for flows on things like Objects, Results and Lists that
+ * allow you to pass in keypaths to restrict notifications to certain properties and one variant
+ * for flows on entities that doesn't support this, e.g. flows on query aggregates or the Realm as
+ * whole.
  */
-internal interface Flowable<T> {
-    fun asFlow(): Flow<T>
+
+internal interface KeyPathFlowable<T> {
+    fun asFlow(keyPaths: List<String>? = null): Flow<T>
 }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/Notifiable.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/Notifiable.kt
index 8ce71fd03c..f8eb5e1c51 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/Notifiable.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/Notifiable.kt
@@ -18,6 +18,7 @@ package io.realm.kotlin.internal
 import io.realm.kotlin.Versioned
 import io.realm.kotlin.internal.interop.Callback
 import io.realm.kotlin.internal.interop.RealmChangesPointer
+import io.realm.kotlin.internal.interop.RealmKeyPathArrayPointer
 import io.realm.kotlin.internal.interop.RealmNotificationTokenPointer
 import io.realm.kotlin.internal.util.Validation.sdkError
 import io.realm.kotlin.internal.util.trySendWithBufferOverflowCheck
@@ -126,10 +127,10 @@ public abstract class ChangeFlow<T, C>(private val producerScope: ProducerScope<
  * @param T the type of entity that is observed.
  * @param C the type of change events emitted for the T entity.
  */
-internal interface CoreNotifiable<T, C> : Notifiable<T, C>, Observable<T, C>, Versioned, Flowable<C>
+internal interface CoreNotifiable<T, C> : Notifiable<T, C>, Observable<T, C>, Versioned, KeyPathFlowable<C>
         where T : CoreNotifiable<T, C> {
     public fun thaw(liveRealm: RealmReference): T?
-    public fun registerForNotification(callback: Callback<RealmChangesPointer>): RealmNotificationTokenPointer
+    public fun registerForNotification(keyPaths: RealmKeyPathArrayPointer?, callback: Callback<RealmChangesPointer>): RealmNotificationTokenPointer
     public fun freeze(frozenRealm: RealmReference): T?
 
     // Default implementation as all Observables are just thawing themselves.
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/ObjectBoundRealmResults.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/ObjectBoundRealmResults.kt
index 62c847dee8..93d501ae9a 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/ObjectBoundRealmResults.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/ObjectBoundRealmResults.kt
@@ -33,7 +33,7 @@ import kotlinx.coroutines.flow.Flow
 internal class ObjectBoundRealmResults<E : BaseRealmObject>(
     val targetObject: RealmObjectReference<*>,
     val realmResults: RealmResults<E>,
-) : RealmResults<E> by realmResults, InternalDeleteable, Flowable<ResultsChange<E>> {
+) : RealmResults<E> by realmResults, InternalDeleteable {
 
     override val size: Int by realmResults::size
 
@@ -60,8 +60,8 @@ internal class ObjectBoundRealmResults<E : BaseRealmObject>(
      * values, if the object has not been deleted, if not it closes cancels the flow.
      */
 
-    override fun asFlow(): Flow<ResultsChange<E>> {
-        return realmResults.asFlow().bind(targetObject)
+    override fun asFlow(keyPaths: List<String>?): Flow<ResultsChange<E>> {
+        return realmResults.asFlow(keyPaths).bind(targetObject)
     }
 
     override fun delete() {
@@ -78,5 +78,7 @@ internal class ObjectBoundRealmResults<E : BaseRealmObject>(
  * Binds a flow to an object lifecycle. It allows flows on queries to complete once the object gets
  * deleted. It is used on sub-queries and backlinks.
  */
-internal fun <T> Flow<T>.bind(reference: RealmObjectReference<out BaseRealmObject>): Flow<T> =
+internal fun <T> Flow<T>.bind(
+    reference: RealmObjectReference<out BaseRealmObject>
+): Flow<T> =
     this.terminateWhen(reference.asFlow()) { it is DeletedObject }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmImpl.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmImpl.kt
index 9906b050df..adfef17780 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmImpl.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmImpl.kt
@@ -21,6 +21,7 @@ import io.realm.kotlin.MutableRealm
 import io.realm.kotlin.Realm
 import io.realm.kotlin.dynamic.DynamicRealm
 import io.realm.kotlin.internal.dynamic.DynamicRealmImpl
+import io.realm.kotlin.internal.interop.ClassKey
 import io.realm.kotlin.internal.interop.RealmInterop
 import io.realm.kotlin.internal.interop.SynchronizableObject
 import io.realm.kotlin.internal.platform.copyAssetFile
@@ -46,14 +47,13 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.sync.withLock
 import kotlin.reflect.KClass
 
 // TODO API-PUBLIC Document platform specific internals (RealmInitializer, etc.)
 // TODO Public due to being accessed from `SyncedRealmContext`
 public class RealmImpl private constructor(
     configuration: InternalConfiguration,
-) : BaseRealmImpl(configuration), Realm, InternalTypedRealm, Flowable<RealmChange<Realm>> {
+) : BaseRealmImpl(configuration), Realm, InternalTypedRealm {
 
     public val notificationScheduler: LiveRealmContext =
         configuration.notificationDispatcherFactory.createLiveRealmContext()
@@ -224,8 +224,8 @@ public class RealmImpl private constructor(
         )
     }
 
-    override fun <T : CoreNotifiable<T, C>, C> registerObserver(t: Observable<T, C>): Flow<C> {
-        return notifier.registerObserver(t)
+    override fun <T : CoreNotifiable<T, C>, C> registerObserver(t: Observable<T, C>, keyPaths: Pair<ClassKey, List<String>>?): Flow<C> {
+        return notifier.registerObserver(t, keyPaths)
     }
 
     /**
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmListInternal.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmListInternal.kt
index 2b4d12221b..8f1e60a0a7 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmListInternal.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmListInternal.kt
@@ -25,6 +25,7 @@ import io.realm.kotlin.internal.interop.RealmChangesPointer
 import io.realm.kotlin.internal.interop.RealmInterop
 import io.realm.kotlin.internal.interop.RealmInterop.realm_list_get
 import io.realm.kotlin.internal.interop.RealmInterop.realm_list_set_embedded
+import io.realm.kotlin.internal.interop.RealmKeyPathArrayPointer
 import io.realm.kotlin.internal.interop.RealmListPointer
 import io.realm.kotlin.internal.interop.RealmNotificationTokenPointer
 import io.realm.kotlin.internal.interop.RealmObjectInterop
@@ -32,6 +33,7 @@ import io.realm.kotlin.internal.interop.getterScope
 import io.realm.kotlin.internal.interop.inputScope
 import io.realm.kotlin.internal.query.ObjectBoundQuery
 import io.realm.kotlin.internal.query.ObjectQuery
+import io.realm.kotlin.internal.util.Validation
 import io.realm.kotlin.notifications.ListChange
 import io.realm.kotlin.notifications.internal.DeletedListImpl
 import io.realm.kotlin.notifications.internal.InitialListImpl
@@ -49,7 +51,7 @@ import kotlin.reflect.KClass
 internal class UnmanagedRealmList<E>(
     private val backingList: MutableList<E> = mutableListOf()
 ) : RealmList<E>, InternalDeleteable, MutableList<E> by backingList {
-    override fun asFlow(): Flow<ListChange<E>> =
+    override fun asFlow(keyPaths: List<String>?): Flow<ListChange<E>> =
         throw UnsupportedOperationException("Unmanaged lists cannot be observed.")
 
     override fun delete() {
@@ -113,9 +115,13 @@ internal class ManagedRealmList<E>(
         return operator.set(index, element)
     }
 
-    override fun asFlow(): Flow<ListChange<E>> {
+    override fun asFlow(keyPaths: List<String>?): Flow<ListChange<E>> {
         operator.realmReference.checkClosed()
-        return operator.realmReference.owner.registerObserver(this)
+        val keyPathInfo = keyPaths?.let {
+            Validation.isType<RealmObjectListOperator<*>>(operator, "Keypaths are only supported for lists of objects.")
+            Pair(operator.classKey, keyPaths)
+        }
+        return operator.realmReference.owner.registerObserver(this, keyPathInfo)
     }
 
     override fun freeze(frozenRealm: RealmReference): ManagedRealmList<E>? {
@@ -131,9 +137,10 @@ internal class ManagedRealmList<E>(
     }
 
     override fun registerForNotification(
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
-        return RealmInterop.realm_list_add_notification_callback(nativePointer, callback)
+        return RealmInterop.realm_list_add_notification_callback(nativePointer, keyPaths, callback)
     }
 
     override fun changeFlow(scope: ProducerScope<ListChange<E>>): ChangeFlow<ManagedRealmList<E>, ListChange<E>> =
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmMapInternal.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmMapInternal.kt
index 884f771f72..b6c8ff5b47 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmMapInternal.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmMapInternal.kt
@@ -31,6 +31,7 @@ import io.realm.kotlin.internal.interop.RealmInterop.realm_dictionary_get
 import io.realm.kotlin.internal.interop.RealmInterop.realm_dictionary_insert
 import io.realm.kotlin.internal.interop.RealmInterop.realm_dictionary_insert_embedded
 import io.realm.kotlin.internal.interop.RealmInterop.realm_results_get
+import io.realm.kotlin.internal.interop.RealmKeyPathArrayPointer
 import io.realm.kotlin.internal.interop.RealmMapPointer
 import io.realm.kotlin.internal.interop.RealmNotificationTokenPointer
 import io.realm.kotlin.internal.interop.RealmObjectInterop
@@ -39,6 +40,7 @@ import io.realm.kotlin.internal.interop.getterScope
 import io.realm.kotlin.internal.interop.inputScope
 import io.realm.kotlin.internal.query.ObjectBoundQuery
 import io.realm.kotlin.internal.query.ObjectQuery
+import io.realm.kotlin.internal.util.Validation
 import io.realm.kotlin.notifications.MapChange
 import io.realm.kotlin.notifications.MapChangeSet
 import io.realm.kotlin.notifications.internal.DeletedDictionaryImpl
@@ -61,7 +63,7 @@ internal abstract class ManagedRealmMap<K, V> constructor(
     internal val parent: RealmObjectReference<*>,
     internal val nativePointer: RealmMapPointer,
     val operator: MapOperator<K, V>
-) : AbstractMutableMap<K, V>(), RealmMap<K, V>, CoreNotifiable<ManagedRealmMap<K, V>, MapChange<K, V>>, Flowable<MapChange<K, V>> {
+) : AbstractMutableMap<K, V>(), RealmMap<K, V>, CoreNotifiable<ManagedRealmMap<K, V>, MapChange<K, V>> {
 
     private val keysPointer by lazy { RealmInterop.realm_dictionary_get_keys(nativePointer) }
     private val valuesPointer by lazy { RealmInterop.realm_dictionary_to_results(nativePointer) }
@@ -99,15 +101,20 @@ internal abstract class ManagedRealmMap<K, V> constructor(
 
     override fun remove(key: K): V? = operator.remove(key)
 
-    override fun asFlow(): Flow<MapChange<K, V>> {
+    override fun asFlow(keyPaths: List<String>?): Flow<MapChange<K, V>> {
         operator.realmReference.checkClosed()
-        return operator.realmReference.owner.registerObserver(this)
+        val keyPathInfo = keyPaths?.let {
+            Validation.isType<RealmObjectMapOperator<*, *>>(operator, "Keypaths are only supported for maps of objects.")
+            Pair(operator.classKey, keyPaths)
+        }
+        return operator.realmReference.owner.registerObserver(this, keyPathInfo)
     }
 
     override fun registerForNotification(
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer =
-        RealmInterop.realm_dictionary_add_notification_callback(nativePointer, callback)
+        RealmInterop.realm_dictionary_add_notification_callback(nativePointer, keyPaths, callback)
 
     internal fun isValid(): Boolean =
         !nativePointer.isReleased() && RealmInterop.realm_dictionary_is_valid(nativePointer)
@@ -618,7 +625,7 @@ internal class EmbeddedRealmObjectMapOperator<K, V : BaseRealmObject> constructo
 internal class UnmanagedRealmDictionary<V>(
     dictionary: Map<String, V> = mutableMapOf()
 ) : RealmDictionary<V>, MutableMap<String, V> by dictionary.toMutableMap() {
-    override fun asFlow(): Flow<MapChange<String, V>> =
+    override fun asFlow(keyPaths: List<String>?): Flow<MapChange<String, V>> =
         throw UnsupportedOperationException("Unmanaged dictionaries cannot be observed.")
 
     override fun toString(): String = entries.joinToString { (key, value) -> "[$key,$value]" }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmObjectReference.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmObjectReference.kt
index e282065338..ec59b2aea2 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmObjectReference.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmObjectReference.kt
@@ -20,6 +20,7 @@ import io.realm.kotlin.internal.interop.Callback
 import io.realm.kotlin.internal.interop.PropertyKey
 import io.realm.kotlin.internal.interop.RealmChangesPointer
 import io.realm.kotlin.internal.interop.RealmInterop
+import io.realm.kotlin.internal.interop.RealmKeyPathArrayPointer
 import io.realm.kotlin.internal.interop.RealmNotificationTokenPointer
 import io.realm.kotlin.internal.interop.RealmObjectInterop
 import io.realm.kotlin.internal.interop.RealmObjectPointer
@@ -103,10 +104,14 @@ public class RealmObjectReference<T : BaseRealmObject>(
             } as RealmObjectReference<T>?
     }
 
-    override fun registerForNotification(callback: Callback<RealmChangesPointer>): RealmNotificationTokenPointer {
+    override fun registerForNotification(
+        keyPaths: RealmKeyPathArrayPointer?,
+        callback: Callback<RealmChangesPointer>
+    ): RealmNotificationTokenPointer {
         // We should never get here unless it is a managed object as unmanaged doesn't support observing
         return RealmInterop.realm_object_add_notification_callback(
             this.objectPointer,
+            keyPaths,
             callback
         )
     }
@@ -124,8 +129,11 @@ public class RealmObjectReference<T : BaseRealmObject>(
         }.toTypedArray()
     }
 
-    override fun asFlow(): Flow<ObjectChange<T>> {
-        return this.owner.owner.registerObserver(this)
+    override fun asFlow(keyPaths: List<String>?): Flow<ObjectChange<T>> {
+        val keyPathInfo = keyPaths?.let {
+            Pair(metadata.classKey, it)
+        }
+        return this.owner.owner.registerObserver(this, keyPathInfo)
     }
 
     override fun delete() {
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmResultsImpl.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmResultsImpl.kt
index 87dab43f86..b5586d9ef2 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmResultsImpl.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmResultsImpl.kt
@@ -22,6 +22,7 @@ import io.realm.kotlin.internal.interop.ClassKey
 import io.realm.kotlin.internal.interop.RealmChangesPointer
 import io.realm.kotlin.internal.interop.RealmInterop
 import io.realm.kotlin.internal.interop.RealmInterop.realm_results_get
+import io.realm.kotlin.internal.interop.RealmKeyPathArrayPointer
 import io.realm.kotlin.internal.interop.RealmNotificationTokenPointer
 import io.realm.kotlin.internal.interop.RealmResultsPointer
 import io.realm.kotlin.internal.interop.getterScope
@@ -104,9 +105,12 @@ internal class RealmResultsImpl<E : BaseRealmObject> constructor(
         )
     }
 
-    override fun asFlow(): Flow<ResultsChange<E>> {
+    override fun asFlow(keyPaths: List<String>?): Flow<ResultsChange<E>> {
         realm.checkClosed()
-        return realm.owner.registerObserver(this)
+        val keyPathInfo = keyPaths?.let {
+            Pair(classKey, keyPaths)
+        }
+        return realm.owner.registerObserver(this, keyPathInfo)
     }
 
     override fun delete() {
@@ -134,8 +138,11 @@ internal class RealmResultsImpl<E : BaseRealmObject> constructor(
         return RealmResultsImpl(liveRealm, liveResultPtr, classKey, clazz, mediator)
     }
 
-    override fun registerForNotification(callback: Callback<RealmChangesPointer>): RealmNotificationTokenPointer {
-        return RealmInterop.realm_results_add_notification_callback(nativePointer, callback)
+    override fun registerForNotification(
+        keyPaths: RealmKeyPathArrayPointer?,
+        callback: Callback<RealmChangesPointer>
+    ): RealmNotificationTokenPointer {
+        return RealmInterop.realm_results_add_notification_callback(nativePointer, keyPaths, callback)
     }
 
     override fun changeFlow(scope: ProducerScope<ResultsChange<E>>): ChangeFlow<RealmResultsImpl<E>, ResultsChange<E>> =
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmSetInternal.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmSetInternal.kt
index c61098a768..57337a8973 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmSetInternal.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/RealmSetInternal.kt
@@ -24,6 +24,7 @@ import io.realm.kotlin.internal.interop.ClassKey
 import io.realm.kotlin.internal.interop.RealmChangesPointer
 import io.realm.kotlin.internal.interop.RealmInterop
 import io.realm.kotlin.internal.interop.RealmInterop.realm_set_get
+import io.realm.kotlin.internal.interop.RealmKeyPathArrayPointer
 import io.realm.kotlin.internal.interop.RealmNotificationTokenPointer
 import io.realm.kotlin.internal.interop.RealmObjectInterop
 import io.realm.kotlin.internal.interop.RealmSetPointer
@@ -32,6 +33,7 @@ import io.realm.kotlin.internal.interop.getterScope
 import io.realm.kotlin.internal.interop.inputScope
 import io.realm.kotlin.internal.query.ObjectBoundQuery
 import io.realm.kotlin.internal.query.ObjectQuery
+import io.realm.kotlin.internal.util.Validation
 import io.realm.kotlin.notifications.SetChange
 import io.realm.kotlin.notifications.internal.DeletedSetImpl
 import io.realm.kotlin.notifications.internal.InitialSetImpl
@@ -49,7 +51,7 @@ import kotlin.reflect.KClass
 internal class UnmanagedRealmSet<E>(
     private val backingSet: MutableSet<E> = mutableSetOf()
 ) : RealmSet<E>, InternalDeleteable, MutableSet<E> by backingSet {
-    override fun asFlow(): Flow<SetChange<E>> {
+    override fun asFlow(keyPaths: List<String>?): Flow<SetChange<E>> {
         throw UnsupportedOperationException("Unmanaged sets cannot be observed.")
     }
 
@@ -159,8 +161,12 @@ internal class ManagedRealmSet<E> constructor(
         }
     }
 
-    override fun asFlow(): Flow<SetChange<E>> {
-        return operator.realmReference.owner.registerObserver(this)
+    override fun asFlow(keyPaths: List<String>?): Flow<SetChange<E>> {
+        val keyPathInfo = keyPaths?.let {
+            Validation.isType<RealmObjectSetOperator<*>>(operator, "Keypaths are only supported for sets of objects.")
+            Pair(operator.classKey, keyPaths)
+        }
+        return operator.realmReference.owner.registerObserver(this, keyPathInfo)
     }
 
     override fun freeze(frozenRealm: RealmReference): ManagedRealmSet<E>? {
@@ -176,9 +182,10 @@ internal class ManagedRealmSet<E> constructor(
     }
 
     override fun registerForNotification(
+        keyPaths: RealmKeyPathArrayPointer?,
         callback: Callback<RealmChangesPointer>
     ): RealmNotificationTokenPointer {
-        return RealmInterop.realm_set_add_notification_callback(nativePointer, callback)
+        return RealmInterop.realm_set_add_notification_callback(nativePointer, keyPaths, callback)
     }
 
     override fun changeFlow(scope: ProducerScope<SetChange<E>>): ChangeFlow<ManagedRealmSet<E>, SetChange<E>> =
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/SuspendableNotifier.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/SuspendableNotifier.kt
index 260b254399..b3e68aaa7a 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/SuspendableNotifier.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/SuspendableNotifier.kt
@@ -2,8 +2,10 @@ package io.realm.kotlin.internal
 
 import io.realm.kotlin.VersionId
 import io.realm.kotlin.internal.interop.Callback
+import io.realm.kotlin.internal.interop.ClassKey
 import io.realm.kotlin.internal.interop.RealmChangesPointer
 import io.realm.kotlin.internal.interop.RealmInterop
+import io.realm.kotlin.internal.interop.RealmKeyPathArrayPointer
 import io.realm.kotlin.internal.platform.runBlocking
 import io.realm.kotlin.internal.schema.RealmSchemaImpl
 import io.realm.kotlin.internal.util.LiveRealmContext
@@ -93,7 +95,8 @@ internal class SuspendableNotifier(
         return _realmChanged.asSharedFlow()
     }
 
-    internal fun <T : CoreNotifiable<T, C>, C> registerObserver(flowable: Observable<T, C>): Flow<C> {
+    internal fun <T : CoreNotifiable<T, C>, C> registerObserver(flowable: Observable<T, C>, keyPaths: Pair<ClassKey, List<String>>?): Flow<C> {
+        val keypathsPtr: RealmKeyPathArrayPointer? = keyPaths?.let { RealmInterop.realm_create_key_paths_array(realm.owner.realmReference.dbPointer, keyPaths.first, keyPaths.second) }
         return callbackFlow {
             val token: AtomicRef<Cancellable> =
                 kotlinx.atomicfu.atomic(NO_OP_NOTIFICATION_TOKEN)
@@ -122,7 +125,7 @@ internal class SuspendableNotifier(
                                 changeFlow.emit(frozenObservable, change)
                             }
                         }
-                    token.value = NotificationToken(lifeRef.registerForNotification(interopCallback))
+                    token.value = NotificationToken(lifeRef.registerForNotification(keypathsPtr, interopCallback))
                 } else {
                     changeFlow.emit(null)
                 }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ObjectBoundQueries.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ObjectBoundQueries.kt
index 35da179879..f1db1ec0b1 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ObjectBoundQueries.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ObjectBoundQueries.kt
@@ -48,7 +48,9 @@ internal class ObjectBoundQuery<E : BaseRealmObject>(
         realmQuery.query(filter, *arguments)
     )
 
-    override fun asFlow(): Flow<ResultsChange<E>> = realmQuery.asFlow().bind(targetObject)
+    override fun asFlow(keyPaths: List<String>?): Flow<ResultsChange<E>> = realmQuery.asFlow(keyPaths).bind(
+        targetObject
+    )
 
     override fun sort(property: String, sortOrder: Sort): RealmQuery<E> = ObjectBoundQuery(
         targetObject,
@@ -107,7 +109,7 @@ internal class ObjectBoundRealmSingleQuery<E : BaseRealmObject>(
     val targetObject: RealmObjectReference<*>,
     val realmQuery: RealmSingleQuery<E>
 ) : RealmSingleQuery<E> by realmQuery {
-    override fun asFlow(): Flow<SingleQueryChange<E>> = realmQuery.asFlow().bind(targetObject)
+    override fun asFlow(keyPaths: List<String>?): Flow<SingleQueryChange<E>> = realmQuery.asFlow(keyPaths).bind(targetObject)
 }
 
 internal class ObjectBoundRealmScalarNullableQuery<E>(
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ObjectQuery.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ObjectQuery.kt
index 348f8c0f0b..ffe3dd6b62 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ObjectQuery.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ObjectQuery.kt
@@ -170,9 +170,10 @@ internal class ObjectQuery<E : BaseRealmObject> constructor(
     override fun notifiable(): Notifiable<RealmResultsImpl<E>, ResultsChange<E>> =
         QueryResultNotifiable(resultsPointer, classKey, clazz, mediator)
 
-    override fun asFlow(): Flow<ResultsChange<E>> {
+    override fun asFlow(keyPath: List<String>?): Flow<ResultsChange<E>> {
+        val keyPathInfo = keyPath?.let { Pair(classKey, it) }
         return realmReference.owner
-            .registerObserver(this)
+            .registerObserver(this, keyPathInfo)
     }
 
     override fun delete() {
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ScalarQuery.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ScalarQuery.kt
index 43e9964d93..bba1c4f58f 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ScalarQuery.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/ScalarQuery.kt
@@ -96,7 +96,7 @@ internal class CountQuery<E : BaseRealmObject> constructor(
     override fun asFlow(): Flow<Long> {
         realmReference.checkClosed()
         return realmReference.owner
-            .registerObserver(this)
+            .registerObserver(this, null)
             .map {
                 it.list.size.toLong()
             }.distinctUntilChanged()
@@ -148,7 +148,7 @@ internal class MinMaxQuery<E : BaseRealmObject, T : Any> constructor(
     override fun asFlow(): Flow<T?> {
         realmReference.checkClosed()
         return realmReference.owner
-            .registerObserver(this)
+            .registerObserver(this, null)
             .map {
                 val realmResults = it.list as RealmResultsImpl<*>
                 findFromResults(realmResults.nativePointer, realmResults.realm)
@@ -216,7 +216,7 @@ internal class SumQuery<E : BaseRealmObject, T : Any> constructor(
     override fun asFlow(): Flow<T> {
         realmReference.checkClosed()
         return realmReference.owner
-            .registerObserver(this)
+            .registerObserver(this, null)
             .map { findFromResults((it.list as RealmResultsImpl<*>).nativePointer) }
             .distinctUntilChanged()
     }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/SingleQuery.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/SingleQuery.kt
index 197068935d..3db3de2c10 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/SingleQuery.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/query/SingleQuery.kt
@@ -56,9 +56,12 @@ internal class SingleQuery<E : BaseRealmObject> constructor(
      * new head if any.
      * If there is an update, we ignore it, as the object flow would automatically emit the event.
      */
-    override fun asFlow(): Flow<SingleQueryChange<E>> {
+    override fun asFlow(keyPaths: List<String>?): Flow<SingleQueryChange<E>> {
         var oldHead: E? = null
-        return realmReference.owner.registerObserver(this)
+        val keyPathInfo = keyPaths?.let {
+            Pair(classKey, it)
+        }
+        return realmReference.owner.registerObserver(this, keyPathInfo)
             // Convert into flow of result head
             .map { resultChange: ResultsChange<E> -> resultChange.list.firstOrNull() }
             // Only react when head is changed
@@ -84,9 +87,9 @@ internal class SingleQuery<E : BaseRealmObject> constructor(
                 } else {
                     oldHead = newHead
                     if (!oldHeadDeleted) {
-                        newHead.asFlow()
+                        newHead.asFlow(keyPaths)
                     } else {
-                        newHead.asFlow().onStart { emit(DeletedObjectImpl()) }
+                        newHead.asFlow(keyPaths).onStart { emit(DeletedObjectImpl()) }
                     }
                 }
             }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/util/Validation.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/util/Validation.kt
index 1a76d2ee98..7172efdc20 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/util/Validation.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/util/Validation.kt
@@ -16,6 +16,9 @@
 
 package io.realm.kotlin.internal.util
 
+import kotlin.contracts.ExperimentalContracts
+import kotlin.contracts.contract
+
 /**
  * Collection of validation methods to ensure uniform input validation.
  */
@@ -30,6 +33,20 @@ public object Validation {
         return value
     }
 
+    /**
+     * Verifies that a given argument has a given type. If yes, it will be implicitly cast
+     * to that type, otherwise an IllegalArgumentException is thrown with the provided error message.
+     */
+    @OptIn(ExperimentalContracts::class)
+    public inline fun <reified T : Any?> isType(arg: Any?, errorMessage: String) {
+        contract {
+            returns() implies (arg is T)
+        }
+        if (arg !is T) {
+            throw IllegalArgumentException(errorMessage)
+        }
+    }
+
     public fun isEmptyString(str: String?): Boolean {
         return str == null || str.length == 0
     }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmElementQuery.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmElementQuery.kt
index b8aaf74de6..78326031ea 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmElementQuery.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmElementQuery.kt
@@ -60,7 +60,12 @@ public interface RealmElementQuery<T : BaseRealmObject> : Deleteable {
      *
      * **It is not allowed to call [asFlow] on queries generated from a [MutableRealm].**
      *
+     * @param keyPaths An optional list of properties that defines when a change to the object will
+     * result in a change being emitted. Nested properties can be defined using a dotted
+     * syntax, e.g. `parent.child.name`. If no keypaths are provided, changes to all top-level
+     * properties and nested properties 4 levels down will trigger a change.
      * @return a flow representing changes to the [RealmResults] resulting from running this query.
+     * @throws IllegalArgumentException if an invalid keypath is provided.
      */
-    public fun asFlow(): Flow<ResultsChange<T>>
+    public fun asFlow(keyPath: List<String>? = null): Flow<ResultsChange<T>>
 }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmResults.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmResults.kt
index f9d7d80732..f3de8dd200 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmResults.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmResults.kt
@@ -69,7 +69,15 @@ public interface RealmResults<T : BaseRealmObject> : List<T>, Deleteable, Versio
      * the elements in a timely manner the coroutine scope will be cancelled with a
      * [CancellationException].
      *
+     * @param keyPaths An optional list of model class properties that defines when a change to
+     * objects inside the RealmResults will result in a change being emitted. Nested properties can
+     * be defined using a dotted syntax, e.g. `parent.child.name`. Wildcards `*` can be be used
+     * to capture all properties at a given level, e.g. `child.*` or `*.*`. If no keypaths are
+     * provided, changes to all top-level properties and nested properties up to 4 levels down
+     * will trigger a change.
+     * @return a flow representing changes to the list.
+     * @throws IllegalArgumentException if an invalid keypath is provided.
      * @return a flow representing changes to the RealmResults.
      */
-    public fun asFlow(): Flow<ResultsChange<T>>
+    public fun asFlow(keyPaths: List<String>? = null): Flow<ResultsChange<T>>
 }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmSingleQuery.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmSingleQuery.kt
index 7ce8a28867..e73eab1804 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmSingleQuery.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/query/RealmSingleQuery.kt
@@ -72,8 +72,13 @@ public interface RealmSingleQuery<T : BaseRealmObject> : Deleteable {
      * the elements in a timely manner the coroutine scope will be cancelled with a
      * [CancellationException].
      *
+     * @param keyPaths An optional list of properties that defines when a change to the object will
+     * result in a change being emitted. Nested properties can be defined using a dotted
+     * syntex, e.g. `parent.child.name`. If no keypaths are provided, changes to all top-level
+     * properties and nested properties 4 levels down will trigger a change.
      * @return a flow representing changes to the [RealmObject] or [EmbeddedRealmObject] resulting from
      * running this query.
+     * @throws IllegalArgumentException if an invalid keypath is provided.
      */
-    public fun asFlow(): Flow<SingleQueryChange<T>>
+    public fun asFlow(keyPaths: List<String>? = null): Flow<SingleQueryChange<T>>
 }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmList.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmList.kt
index 5dcfca308a..0cc3fa4380 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmList.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmList.kt
@@ -72,9 +72,17 @@ public interface RealmList<E> : MutableList<E>, Deleteable {
      * the elements in a timely manner the coroutine scope will be cancelled with a
      * [CancellationException].
      *
+     * @param keyPaths An optional list of model class properties that defines when a change to
+     * objects inside the list will result in a change being emitted. Nested properties can be
+     * defined using a dotted syntax, e.g. `parent.child.name`. Wildcards `*` can be be used
+     * to capture all properties at a given level, e.g. `child.*` or `*.*`.If no keypaths are
+     * provided, changes to all top-level properties and nested properties up to 4 levels down
+     * will trigger a change. Keypaths are only supported for lists of objects.
      * @return a flow representing changes to the list.
+     * @throws IllegalArgumentException if an invalid keypath is provided or the RealmList does not
+     * contain realm objects.
      * @throws CancellationException if the stream produces changes faster than the consumer can
      * consume them and results in a buffer overflow.
      */
-    public fun asFlow(): Flow<ListChange<E>>
+    public fun asFlow(keyPaths: List<String>? = null): Flow<ListChange<E>>
 }
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmMap.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmMap.kt
index 2e45c4598b..7a03c805f9 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmMap.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmMap.kt
@@ -55,11 +55,20 @@ public interface RealmMap<K, V> : MutableMap<K, V> {
      * the elements in a timely manner the coroutine scope will be cancelled with a
      * [CancellationException].
      *
+     * @param keyPaths An optional list of model class properties that defines when a change to
+     * objects inside the map will result in a change being emitted. For maps, keypaths are
+     * evaluted based on the values of the map. This means that keypaths are only supported
+     * for maps containing realm objects. Nested properties can be defined using a dotted syntax,
+     * e.g. `parent.child.name`. Wildcards `*` can be be used to capture all properties at a given
+     * level, e.g. `child.*` or `*.*`. If no keypaths are provided, changes to all top-level
+     * properties and nested properties up to 4 levels down will trigger a change
      * @return a flow representing changes to the dictionary.
+     * @throws IllegalArgumentException if keypaths are invalid or the map does not contain realm
+     * objects.
      * @throws CancellationException if the stream produces changes faster than the consumer can
      * consume them and results in a buffer overflow.
      */
-    public fun asFlow(): Flow<MapChange<K, V>>
+    public fun asFlow(keyPaths: List<String>? = null): Flow<MapChange<K, V>>
 }
 
 /**
diff --git a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmSet.kt b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmSet.kt
index d7366f453e..6847974642 100644
--- a/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmSet.kt
+++ b/packages/library-base/src/commonMain/kotlin/io/realm/kotlin/types/RealmSet.kt
@@ -52,9 +52,17 @@ public interface RealmSet<E> : MutableSet<E>, Deleteable {
      * the elements in a timely manner the coroutine scope will be cancelled with a
      * [CancellationException].
      *
+     * @param keyPaths An optional list of model class properties that defines when a change to
+     * objects inside the set will result in a change being emitted. Nested properties can be
+     * defined using a dotted syntax, e.g. `parent.child.name`. Wildcards `*` can be be used
+     * to capture all properties at a given level, e.g. `child.*` or `*.*`. If no keypaths
+     * are provided, changes to all top-level properties and nested properties up to 4 levels down
+     * will trigger a change. Keypaths are only supported for sets containing realm objects.
      * @return a flow representing changes to the set.
+     * @throws IllegalArgumentException if an invalid keypath is provided or the set does not
+     * contain realm objects.
      * @throws CancellationException if the stream produces changes faster than the consumer can
      * consume them and results in a buffer overflow.
      */
-    public fun asFlow(): Flow<SetChange<E>>
+    public fun asFlow(keyPaths: List<String>? = null): Flow<SetChange<E>>
 }
diff --git a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/QueryTests.kt b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/QueryTests.kt
index 3f7267c8a0..a63f89d83b 100644
--- a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/QueryTests.kt
+++ b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/QueryTests.kt
@@ -2946,6 +2946,138 @@ class QueryTests {
         }
     }
 
+    @Test
+    fun asFlow_results_withKeyPath() {
+        val channel = Channel<ResultsChange<QuerySample>>(1)
+        runBlocking {
+            val observer = async {
+                realm.query<QuerySample>()
+                    .asFlow(listOf("stringField"))
+                    .collect { results ->
+                        assertNotNull(results)
+                        channel.send(results)
+                    }
+            }
+            channel.receiveOrFail().let { resultsChange ->
+                assertIs<InitialResults<*>>(resultsChange)
+                assertTrue(resultsChange.list.isEmpty())
+            }
+            val obj = realm.writeBlocking {
+                copyToRealm(QuerySample())
+            }
+            channel.receiveOrFail().let { resultsChange ->
+                assertIs<UpdatedResults<*>>(resultsChange)
+                assertEquals(1, resultsChange.list.size)
+            }
+            realm.writeBlocking {
+                // Should not trigger notification
+                findLatest(obj)!!.intField = 42
+            }
+            realm.writeBlocking {
+                // Should trigger notification
+                findLatest(obj)!!.stringField = "update"
+            }
+            channel.receiveOrFail().let { resultsChange ->
+                assertIs<UpdatedResults<*>>(resultsChange)
+                assertEquals(1, resultsChange.list.size)
+                assertEquals("update", resultsChange.list.first().stringField)
+            }
+            observer.cancel()
+            channel.close()
+        }
+    }
+
+    @Test
+    fun asFlow_objectBound_withKeyPath() {
+        val channel = Channel<SingleQueryChange<QuerySample>>(1)
+        runBlocking {
+            val observer = async {
+                realm.query<QuerySample>()
+                    .first()
+                    .asFlow(listOf("stringField"))
+                    .collect { change ->
+                        assertNotNull(change)
+                        channel.send(change)
+                    }
+            }
+            channel.receiveOrFail().let { objChange ->
+                assertIs<PendingObject<*>>(objChange)
+            }
+            val obj = realm.writeBlocking {
+                copyToRealm(QuerySample())
+            }
+            channel.receiveOrFail().let { objChange ->
+                assertIs<InitialObject<*>>(objChange)
+            }
+            realm.writeBlocking {
+                // Should not trigger notification
+                findLatest(obj)!!.intField = 42
+            }
+            realm.writeBlocking {
+                // Should trigger notification
+                findLatest(obj)!!.stringField = "update"
+            }
+            channel.receiveOrFail().let { objChange ->
+                assertIs<UpdatedObject<*>>(objChange)
+                assertEquals(1, objChange.changedFields.size)
+                assertEquals("stringField", objChange.changedFields.first())
+            }
+            observer.cancel()
+            channel.close()
+        }
+    }
+
+    // Smoke-test for wildcards.
+    @Test
+    fun keyPath_usingWildCards() = runBlocking<Unit> {
+        val channel = Channel<ResultsChange<QuerySample>>(1)
+        val observer = async {
+            realm.query<QuerySample>("stringField = 'parent'")
+                // Should match what the notifier is doing by default
+                .asFlow(listOf("*.*.*.*"))
+                .collect { results ->
+                    assertNotNull(results)
+                    channel.send(results)
+                }
+        }
+        channel.receiveOrFail().let { resultsChange ->
+            assertIs<InitialResults<*>>(resultsChange)
+            assertTrue(resultsChange.list.isEmpty())
+        }
+        val obj = realm.write {
+            copyToRealm(
+                QuerySample().apply {
+                    stringField = "parent"
+                    nullableRealmObject = QuerySample().apply {
+                        stringField = "child"
+                    }
+                }
+            )
+        }
+        channel.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<*>>(resultsChange)
+            assertEquals(1, resultsChange.list.size)
+        }
+        realm.write {
+            findLatest(obj)!!.intField = 42
+        }
+        channel.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<*>>(resultsChange)
+            assertEquals(1, resultsChange.list.size)
+            assertEquals(42, resultsChange.list.first().intField)
+        }
+        realm.write {
+            findLatest(obj)!!.nullableRealmObject!!.stringField = "update"
+        }
+        channel.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<*>>(resultsChange)
+            assertEquals(1, resultsChange.list.size)
+            assertEquals("update", resultsChange.list.first().nullableRealmObject!!.stringField)
+        }
+        observer.cancel()
+        channel.close()
+    }
+
     // ----------------
     // Coercion helpers
     // ----------------
diff --git a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/BacklinksNotificationsTests.kt b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/BacklinksNotificationsTests.kt
index e8d8d8a1c6..c32ca0f6eb 100644
--- a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/BacklinksNotificationsTests.kt
+++ b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/BacklinksNotificationsTests.kt
@@ -19,6 +19,7 @@ package io.realm.kotlin.test.common.notifications
 import io.realm.kotlin.Realm
 import io.realm.kotlin.RealmConfiguration
 import io.realm.kotlin.entities.Sample
+import io.realm.kotlin.ext.query
 import io.realm.kotlin.notifications.InitialResults
 import io.realm.kotlin.notifications.ResultsChange
 import io.realm.kotlin.notifications.UpdatedResults
@@ -39,6 +40,7 @@ import kotlin.test.BeforeTest
 import kotlin.test.Ignore
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertIs
 import kotlin.test.assertNotNull
 import kotlin.test.assertNull
@@ -408,4 +410,251 @@ class BacklinksNotificationsTests : RealmEntityNotificationTests {
             c.close()
         }
     }
+
+    @Test
+    override fun keyPath_topLevelProperty() = runBlocking<Unit> {
+        val c = Channel<ResultsChange<Sample>>(1)
+        realm.write {
+            copyToRealm(
+                Sample().apply {
+                    this.stringField = "parent"
+                    this.nullableObject = Sample().apply {
+                        this.stringField = "child"
+                    }
+                }
+            )
+        }
+        val result: RealmResults<Sample> = realm.query<Sample>("stringField = 'child'").find()
+        assertEquals(1, result.first().objectBacklinks.size)
+        val observer = async {
+            result.asFlow(listOf("objectBacklinks")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialResults<Sample>>(c.receiveOrFail())
+        realm.write {
+            val child = findLatest(result.first())!!
+            assertEquals("child", child.stringField)
+            copyToRealm(
+                Sample().apply {
+                    this.stringField = "newParent"
+                    this.nullableObject = child
+                }
+            )
+        }
+        c.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<Sample>>(resultsChange)
+            when (resultsChange) {
+                is UpdatedResults -> {
+                    assertEquals(0, resultsChange.insertions.size)
+                    assertEquals(1, resultsChange.changes.size)
+                    assertEquals(0, resultsChange.deletions.size)
+                    assertEquals(2, resultsChange.list.first().objectBacklinks.size)
+                }
+                else -> fail("Unexpected change: $resultsChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_nestedProperty() = runBlocking<Unit> {
+        val c = Channel<ResultsChange<Sample>>(1)
+        realm.write {
+            copyToRealm(
+                Sample().apply {
+                    this.stringField = "parent"
+                    this.nullableObject = Sample().apply {
+                        this.stringField = "child"
+                    }
+                }
+            )
+        }
+        val result: RealmResults<Sample> = realm.query<Sample>("stringField = 'child'").find()
+        assertEquals(1, result.first().objectBacklinks.size)
+        val observer = async {
+            result.asFlow(listOf("objectBacklinks.intField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialResults<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(result.first())!!.booleanField = false
+        }
+        realm.write {
+            findLatest(result.first())!!.objectBacklinks.first().intField = 1
+        }
+        c.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<Sample>>(resultsChange)
+            when (resultsChange) {
+                is UpdatedResults -> {
+                    assertEquals(0, resultsChange.insertions.size)
+                    assertEquals(1, resultsChange.changes.size)
+                    assertEquals(0, resultsChange.deletions.size)
+                    assertEquals(1, resultsChange.list.first().objectBacklinks.size)
+                    // This starts at 42, if the first write triggers a change event, it will
+                    // catch it here.
+                    assertEquals(resultsChange.list.first().objectBacklinks.first().intField, 1)
+                }
+                else -> fail("Unexpected change: $resultsChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_defaultDepth() = runBlocking<Unit> {
+        val c = Channel<ResultsChange<Sample>>(1)
+        realm.write {
+            copyToRealm(
+                Sample().apply {
+                    this.intField = 1
+                    this.stringField = "parent"
+                    this.nullableObject = Sample().apply {
+                        this.stringField = "child"
+                        this.nullableObject = Sample().apply {
+                            this.stringField = "child-child"
+                            this.nullableObject = Sample().apply {
+                                this.stringField = "child-child-child"
+                                this.nullableObject = Sample().apply {
+                                    this.stringField = "child-child-child-child"
+                                    this.nullableObject = Sample().apply {
+                                        this.stringField = "BottomChild"
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            )
+        }
+        val results = realm.query<Sample>("stringField = 'BottomChild'").find()
+        val observer = async {
+            // Default keypath
+            results.asFlow().collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialResults<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(results.first())!!
+                .objectBacklinks.first()
+                .objectBacklinks.first()
+                .objectBacklinks.first()
+                .objectBacklinks.first()
+                .objectBacklinks.first()
+                .stringField = "Bar"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(results.first())!!.intField = 1
+        }
+        c.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<Sample>>(resultsChange)
+            when (resultsChange) {
+                is ResultsChange<*> -> {
+                    // Default value is 42, so if this event is triggered by the first write
+                    // this assert will fail
+                    assertEquals(1, resultsChange.list.first().intField)
+                }
+                else -> fail("Unexpected change: $resultsChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_propertyBelowDefaultLimit() = runBlocking<Unit> {
+        val c = Channel<ResultsChange<Sample>>(1)
+        realm.write {
+            copyToRealm(
+                Sample().apply {
+                    this.intField = 1
+                    this.stringField = "parent"
+                    this.nullableObject = Sample().apply {
+                        this.stringField = "child"
+                        this.nullableObject = Sample().apply {
+                            this.stringField = "child-child"
+                            this.nullableObject = Sample().apply {
+                                this.stringField = "child-child-child"
+                                this.nullableObject = Sample().apply {
+                                    this.stringField = "child-child-child-child"
+                                    this.nullableObject = Sample().apply {
+                                        this.stringField = "BottomChild"
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            )
+        }
+        val results = realm.query<Sample>("stringField = 'BottomChild'").find()
+        val observer = async {
+            results.asFlow(listOf("objectBacklinks.objectBacklinks.objectBacklinks.objectBacklinks.objectBacklinks.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialResults<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(results.first())!!.booleanField = true
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(results.first())!!
+                .objectBacklinks.first()
+                .objectBacklinks.first()
+                .objectBacklinks.first()
+                .objectBacklinks.first()
+                .objectBacklinks.first()
+                .stringField = "Bar"
+        }
+        c.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<Sample>>(resultsChange)
+            when (resultsChange) {
+                is ResultsChange<*> -> {
+                    assertEquals(0, resultsChange.insertions.size)
+                    assertEquals(1, resultsChange.changes.size)
+                    assertEquals(0, resultsChange.deletions.size)
+                    assertEquals(1, resultsChange.list.first().objectBacklinks.size)
+                }
+                else -> fail("Unexpected change: $resultsChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_unknownTopLevelProperty() = runBlocking<Unit> {
+        val results = realm.query<Sample>()
+        assertFailsWith<IllegalArgumentException>() {
+            results.asFlow(listOf("foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_unknownNestedProperty() = runBlocking<Unit> {
+        val results = realm.query<Sample>()
+        assertFailsWith<IllegalArgumentException>() {
+            results.asFlow(listOf("objectBacklinks.foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_invalidNestedProperty() = runBlocking<Unit> {
+        val results = realm.query<Sample>()
+        assertFailsWith<IllegalArgumentException> {
+            results.asFlow(listOf("objectBacklinks.intField.foo"))
+        }
+        assertFailsWith<IllegalArgumentException> {
+            results.asFlow(listOf("objectBacklinks.intListField.foo"))
+        }
+    }
 }
diff --git a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmDictionaryNotificationsTests.kt b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmDictionaryNotificationsTests.kt
index 05cee44aad..15f65247ec 100644
--- a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmDictionaryNotificationsTests.kt
+++ b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmDictionaryNotificationsTests.kt
@@ -20,6 +20,7 @@ import io.realm.kotlin.Realm
 import io.realm.kotlin.RealmConfiguration
 import io.realm.kotlin.entities.dictionary.DictionaryEmbeddedLevel1
 import io.realm.kotlin.entities.dictionary.RealmDictionaryContainer
+import io.realm.kotlin.ext.realmDictionaryOf
 import io.realm.kotlin.notifications.DeletedMap
 import io.realm.kotlin.notifications.InitialMap
 import io.realm.kotlin.notifications.MapChange
@@ -29,6 +30,7 @@ import io.realm.kotlin.test.common.NULLABLE_DICTIONARY_OBJECT_VALUES
 import io.realm.kotlin.test.common.utils.RealmEntityNotificationTests
 import io.realm.kotlin.test.platform.PlatformUtils
 import io.realm.kotlin.test.util.receiveOrFail
+import io.realm.kotlin.types.RealmDictionary
 import kotlinx.coroutines.async
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.first
@@ -41,6 +43,7 @@ import kotlin.test.BeforeTest
 import kotlin.test.Ignore
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertIs
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
@@ -428,4 +431,274 @@ class RealmDictionaryNotificationsTests : RealmEntityNotificationTests {
             channel.close()
         }
     }
+
+    @Test
+    override fun keyPath_topLevelProperty() = runBlocking<Unit> {
+        val c = Channel<MapChange<String, RealmDictionaryContainer?>>(1)
+        val dict: RealmDictionary<RealmDictionaryContainer?> = realm.write {
+            copyToRealm(
+                RealmDictionaryContainer().apply {
+                    this.nullableObjectDictionaryField = realmDictionaryOf(
+                        "1" to RealmDictionaryContainer().apply { this.stringField = "dict-item-1" },
+                    )
+                }
+            )
+        }.nullableObjectDictionaryField
+        val observer = async {
+            dict.asFlow(listOf("stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialMap<String, RealmDictionaryContainer>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(dict.values.first()!!)!!.id = 42
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(dict.values.first()!!)!!.stringField = "Foo"
+        }
+        c.receiveOrFail().let { mapChange ->
+            assertIs<UpdatedMap<String, RealmDictionaryContainer?>>(mapChange)
+            when (mapChange) {
+                is UpdatedMap -> {
+                    assertEquals(1, mapChange.changes.size)
+                    assertEquals("1", mapChange.changes.first())
+                    // This starts as Realm, so if the first write triggers a change event, it will
+                    // catch it here.
+                    assertEquals("Foo", mapChange.map["1"]!!.stringField)
+                }
+                else -> fail("Unexpected change: $mapChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_nestedProperty() = runBlocking<Unit> {
+        val c = Channel<MapChange<String, RealmDictionaryContainer?>>(1)
+        val dict = realm.write {
+            copyToRealm(
+                RealmDictionaryContainer().apply {
+                    this.stringField = "parent"
+                    this.nullableObjectDictionaryField = realmDictionaryOf(
+                        "1" to RealmDictionaryContainer().apply {
+                            this.stringField = "child"
+                            this.nullableObjectDictionaryField = realmDictionaryOf(
+                                "1-inner" to RealmDictionaryContainer().apply { this.stringField = "list-item-1" }
+                            )
+                        }
+                    )
+                }
+            )
+        }.nullableObjectDictionaryField
+        assertEquals(1, dict.size)
+        val observer = async {
+            dict.asFlow(listOf("nullableObjectDictionaryField.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialMap<String, RealmDictionaryContainer?>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(dict.values.first()!!)!!.id = 1
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(dict.values.first()!!)!!.nullableObjectDictionaryField.values.first()!!.stringField = "Bar"
+        }
+        c.receiveOrFail().let { mapChange ->
+            assertIs<UpdatedMap<String, RealmDictionaryContainer>>(mapChange)
+            when (mapChange) {
+                is UpdatedMap -> {
+                    assertEquals(1, mapChange.changes.size)
+                }
+                else -> fail("Unexpected change: $mapChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_defaultDepth() = runBlocking<Unit> {
+        val c = Channel<MapChange<String, RealmDictionaryContainer?>>(1)
+        val dict = realm.write {
+            copyToRealm(
+                RealmDictionaryContainer().apply {
+                    this.id = 1
+                    this.stringField = "parent"
+                    this.nullableObjectDictionaryField = realmDictionaryOf(
+                        "parent" to RealmDictionaryContainer().apply {
+                            this.stringField = "child"
+                            this.nullableObjectDictionaryField = realmDictionaryOf(
+                                "child" to RealmDictionaryContainer().apply {
+                                    this.stringField = "child-child"
+                                    this.nullableObjectDictionaryField = realmDictionaryOf(
+                                        "child-child" to RealmDictionaryContainer().apply {
+                                            this.stringField = "child-child-child"
+                                            this.nullableObjectDictionaryField = realmDictionaryOf(
+                                                "child-child-child" to RealmDictionaryContainer().apply {
+                                                    this.stringField = "child-child-child-child"
+                                                    this.nullableObjectDictionaryField = realmDictionaryOf(
+                                                        "child-child-child-child" to RealmDictionaryContainer().apply {
+                                                            this.stringField = "child-child-child-child-child"
+                                                            this.nullableObjectDictionaryField = realmDictionaryOf(
+                                                                "child-child-child-child-child" to RealmDictionaryContainer().apply {
+                                                                    this.stringField = "BottomChild"
+                                                                }
+                                                            )
+                                                        }
+                                                    )
+                                                }
+                                            )
+                                        }
+                                    )
+                                }
+                            )
+                        }
+                    )
+                }
+            )
+        }.nullableObjectDictionaryField
+        val observer = async {
+            // Default keypath
+            dict.asFlow().collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialMap<String, RealmDictionaryContainer?>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            val obj = findLatest(dict.values.first()!!)!!
+                .nullableObjectDictionaryField.values.first()!!
+                .nullableObjectDictionaryField.values.first()!!
+                .nullableObjectDictionaryField.values.first()!!
+                .nullableObjectDictionaryField.values.first()!!
+                .nullableObjectDictionaryField.values.first()!!
+            obj.stringField = "Bar"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(dict.values.first()!!)!!.stringField = "Parent change"
+        }
+        c.receiveOrFail().let { mapChange ->
+            assertIs<UpdatedMap<String, RealmDictionaryContainer?>>(mapChange)
+            when (mapChange) {
+                is MapChange -> {
+                    // Core will only report something changed to the top-level property.
+                    assertEquals(1, mapChange.changes.size)
+                    // Default value is Realm, so if this event is triggered by the first write
+                    // this assert will fail
+                    assertEquals("Parent change", mapChange.map.values.first()!!.stringField)
+                }
+                else -> fail("Unexpected change: $mapChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_propertyBelowDefaultLimit() = runBlocking<Unit> {
+        val c = Channel<MapChange<String, RealmDictionaryContainer?>>(1)
+        val dict = realm.write {
+            copyToRealm(
+                RealmDictionaryContainer().apply {
+                    this.id = 1
+                    this.stringField = "parent"
+                    this.nullableObjectDictionaryField = realmDictionaryOf(
+                        "parent" to RealmDictionaryContainer().apply {
+                            this.stringField = "child"
+                            this.nullableObjectDictionaryField = realmDictionaryOf(
+                                "child" to RealmDictionaryContainer().apply {
+                                    this.stringField = "child-child"
+                                    this.nullableObjectDictionaryField = realmDictionaryOf(
+                                        "child-child" to RealmDictionaryContainer().apply {
+                                            this.stringField = "child-child-child"
+                                            this.nullableObjectDictionaryField = realmDictionaryOf(
+                                                "child-child-child" to RealmDictionaryContainer().apply {
+                                                    this.stringField = "child-child-child-child"
+                                                    this.nullableObjectDictionaryField = realmDictionaryOf(
+                                                        "child-child-child-child" to RealmDictionaryContainer().apply {
+                                                            this.stringField = "child-child-child-child-child"
+                                                            this.nullableObjectDictionaryField = realmDictionaryOf(
+                                                                "child-child-child-child-child" to RealmDictionaryContainer().apply {
+                                                                    this.stringField = "BottomChild"
+                                                                }
+                                                            )
+                                                        }
+                                                    )
+                                                }
+                                            )
+                                        }
+                                    )
+                                }
+                            )
+                        }
+                    )
+                }
+            )
+        }.nullableObjectDictionaryField
+        val observer = async {
+            dict.asFlow(listOf("nullableObjectDictionaryField.nullableObjectDictionaryField.nullableObjectDictionaryField.nullableObjectDictionaryField.nullableObjectDictionaryField.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialMap<String, RealmDictionaryContainer?>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(dict.values.first()!!)!!.stringField = "Parent change"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            val obj = findLatest(dict.values.first()!!)!!
+                .nullableObjectDictionaryField.values.first()!!
+                .nullableObjectDictionaryField.values.first()!!
+                .nullableObjectDictionaryField.values.first()!!
+                .nullableObjectDictionaryField.values.first()!!
+                .nullableObjectDictionaryField.values.first()!!
+            obj.stringField = "Bar"
+        }
+        c.receiveOrFail().let { mapChange ->
+            assertIs<UpdatedMap<String, RealmDictionaryContainer?>>(mapChange)
+            when (mapChange) {
+                is MapChange -> {
+                    // Core will only report something changed to the top-level property.
+                    assertEquals(1, mapChange.changes.size)
+                }
+                else -> fail("Unexpected change: $mapChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_unknownTopLevelProperty() = runBlocking<Unit> {
+        val dict = realm.write { copyToRealm(RealmDictionaryContainer()) }.nullableObjectDictionaryField
+        assertFailsWith<IllegalArgumentException>() {
+            dict.asFlow(listOf("foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_unknownNestedProperty() = runBlocking<Unit> {
+        val dict = realm.write { copyToRealm(RealmDictionaryContainer()) }.nullableObjectDictionaryField
+        assertFailsWith<IllegalArgumentException>() {
+            dict.asFlow(listOf("objectDictionaryField.foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_invalidNestedProperty() = runBlocking<Unit> {
+        val dict = realm.write { copyToRealm(RealmDictionaryContainer()) }.nullableObjectDictionaryField
+        assertFailsWith<IllegalArgumentException> {
+            dict.asFlow(listOf("intField.foo"))
+        }
+        assertFailsWith<IllegalArgumentException> {
+            dict.asFlow(listOf("objectDictionaryField.intListField.foo"))
+        }
+    }
 }
diff --git a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmListNotificationsTests.kt b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmListNotificationsTests.kt
index 1121d93221..8fcd21909d 100644
--- a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmListNotificationsTests.kt
+++ b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmListNotificationsTests.kt
@@ -18,8 +18,10 @@ package io.realm.kotlin.test.common.notifications
 
 import io.realm.kotlin.Realm
 import io.realm.kotlin.RealmConfiguration
+import io.realm.kotlin.entities.Sample
 import io.realm.kotlin.entities.list.RealmListContainer
 import io.realm.kotlin.entities.list.listTestSchema
+import io.realm.kotlin.ext.realmListOf
 import io.realm.kotlin.notifications.DeletedList
 import io.realm.kotlin.notifications.InitialList
 import io.realm.kotlin.notifications.ListChange
@@ -45,6 +47,7 @@ import kotlin.test.BeforeTest
 import kotlin.test.Ignore
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertIs
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
@@ -469,6 +472,267 @@ class RealmListNotificationsTests : RealmEntityNotificationTests {
         }
     }
 
+    @Test
+    override fun keyPath_topLevelProperty() = runBlocking<Unit> {
+        val c = Channel<ListChange<RealmListContainer>>(1)
+        val obj = realm.write {
+            copyToRealm(
+                RealmListContainer().apply {
+                    this.objectListField = realmListOf(
+                        RealmListContainer().apply { this.stringField = "list-item-1" },
+                        RealmListContainer().apply { this.stringField = "list-item-2" }
+                    )
+                }
+            )
+        }
+        val list: RealmList<RealmListContainer> = obj.objectListField
+        val observer = async {
+            list.asFlow(listOf("stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialList<RealmListContainer>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(list.first())!!.id = 42
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(list.first())!!.stringField = "Foo"
+        }
+        c.receiveOrFail().let { listChange ->
+            assertIs<UpdatedList<RealmListContainer>>(listChange)
+            when (listChange) {
+                is UpdatedList -> {
+                    assertEquals(1, listChange.changes.size)
+                    // This starts as Realm, so if the first write triggers a change event, it will
+                    // catch it here.
+                    assertEquals("Foo", listChange.list.first().stringField)
+                }
+                else -> fail("Unexpected change: $listChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_nestedProperty() = runBlocking<Unit> {
+        val c = Channel<ListChange<RealmListContainer>>(1)
+        val list = realm.write {
+            copyToRealm(
+                RealmListContainer().apply {
+                    this.stringField = "parent"
+                    this.objectListField = realmListOf(
+                        RealmListContainer().apply {
+                            this.stringField = "child"
+                            this.objectListField = realmListOf(
+                                RealmListContainer().apply { this.stringField = "list-item-1" }
+                            )
+                        }
+                    )
+                }
+            )
+        }.objectListField
+        assertEquals(1, list.size)
+        val observer = async {
+            list.asFlow(listOf("objectListField.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialList<RealmListContainer>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(list.first())!!.id = 1
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(list.first())!!.objectListField.first().stringField = "Bar"
+        }
+        c.receiveOrFail().let { listChange ->
+            assertIs<UpdatedList<RealmListContainer>>(listChange)
+            when (listChange) {
+                is UpdatedList -> {
+                    assertEquals(1, listChange.changes.size)
+                }
+                else -> fail("Unexpected change: $listChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_defaultDepth() = runBlocking<Unit> {
+        val c = Channel<ListChange<RealmListContainer>>(1)
+        val list = realm.write {
+            copyToRealm(
+                RealmListContainer().apply {
+                    this.id = 1
+                    this.stringField = "parent"
+                    this.objectListField = realmListOf(
+                        RealmListContainer().apply {
+                            this.stringField = "child"
+                            this.objectListField = realmListOf(
+                                RealmListContainer().apply {
+                                    this.stringField = "child-child"
+                                    this.objectListField = realmListOf(
+                                        RealmListContainer().apply {
+                                            this.stringField = "child-child-child"
+                                            this.objectListField = realmListOf(
+                                                RealmListContainer().apply {
+                                                    this.stringField = "child-child-child-child"
+                                                    this.objectListField = realmListOf(
+                                                        RealmListContainer().apply {
+                                                            this.stringField = "child-child-child-child-child"
+                                                            this.objectListField = realmListOf(
+                                                                RealmListContainer().apply {
+                                                                    this.stringField = "BottomChild"
+                                                                }
+                                                            )
+                                                        }
+                                                    )
+                                                }
+                                            )
+                                        }
+                                    )
+                                }
+                            )
+                        }
+                    )
+                }
+            )
+        }.objectListField
+        val observer = async {
+            // Default keypath
+            list.asFlow().collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialList<RealmListContainer>>(c.receiveOrFail())
+        realm.write {
+            // Update below the default limit should not trigger a notification
+            val obj = findLatest(list.first())!!.objectListField.first().objectListField.first().objectListField.first().objectListField.first().objectListField.first()
+            obj.stringField = "Bar"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(list.first())!!.id = 1
+        }
+        c.receiveOrFail().let { listChange ->
+            assertIs<UpdatedList<RealmListContainer>>(listChange)
+            when (listChange) {
+                is ListChange -> {
+                    // Core will only report something changed to the top-level property.
+                    assertEquals(1, listChange.changes.size)
+                    // Default value is -1, so if this event is triggered by the first write
+                    // this assert will fail
+                    assertEquals(1, listChange.list.first().id)
+                }
+                else -> fail("Unexpected change: $listChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_propertyBelowDefaultLimit() = runBlocking<Unit> {
+        val c = Channel<ListChange<RealmListContainer>>(1)
+        val list = realm.write {
+            copyToRealm(
+                RealmListContainer().apply {
+                    this.id = 1
+                    this.stringField = "parent"
+                    this.objectListField = realmListOf(
+                        RealmListContainer().apply {
+                            this.stringField = "child"
+                            this.objectListField = realmListOf(
+                                RealmListContainer().apply {
+                                    this.stringField = "child-child"
+                                    this.objectListField = realmListOf(
+                                        RealmListContainer().apply {
+                                            this.stringField = "child-child-child"
+                                            this.objectListField = realmListOf(
+                                                RealmListContainer().apply {
+                                                    this.stringField = "child-child-child-child"
+                                                    this.objectListField = realmListOf(
+                                                        RealmListContainer().apply {
+                                                            this.stringField = "child-child-child-child-child"
+                                                            this.objectListField = realmListOf(
+                                                                RealmListContainer().apply {
+                                                                    this.stringField = "BottomChild"
+                                                                }
+                                                            )
+                                                        }
+                                                    )
+                                                }
+                                            )
+                                        }
+                                    )
+                                }
+                            )
+                        }
+                    )
+                }
+            )
+        }.objectListField
+        val observer = async {
+            list.asFlow(listOf("objectListField.objectListField.objectListField.objectListField.objectListField.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialList<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(list.first())!!.stringField = "Parent change"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            val obj = findLatest(list.first())!!.objectListField.first().objectListField.first().objectListField.first().objectListField.first().objectListField.first()
+            obj.stringField = "Bar"
+        }
+        c.receiveOrFail().let { listChange ->
+            assertIs<UpdatedList<RealmListContainer>>(listChange)
+            when (listChange) {
+                is ListChange -> {
+                    // Core will only report something changed to the top-level property.
+                    assertEquals(1, listChange.changes.size)
+                }
+                else -> fail("Unexpected change: $listChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_unknownTopLevelProperty() = runBlocking<Unit> {
+        val list = realm.write { copyToRealm(RealmListContainer()) }.objectListField
+        assertFailsWith<IllegalArgumentException>() {
+            list.asFlow(listOf("foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_unknownNestedProperty() = runBlocking<Unit> {
+        val list = realm.write { copyToRealm(RealmListContainer()) }.objectListField
+        assertFailsWith<IllegalArgumentException>() {
+            list.asFlow(listOf("objectListField.foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_invalidNestedProperty() = runBlocking<Unit> {
+        val list = realm.write { copyToRealm(RealmListContainer()) }.objectListField
+        assertFailsWith<IllegalArgumentException> {
+            list.asFlow(listOf("intField.foo"))
+        }
+        assertFailsWith<IllegalArgumentException> {
+            list.asFlow(listOf("objectListField.intListField.foo"))
+        }
+    }
+
     fun RealmList<*>.removeRange(range: IntRange) {
         range.reversed().forEach { index -> removeAt(index) }
     }
diff --git a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmObjectNotificationsTests.kt b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmObjectNotificationsTests.kt
index 6a49bd0806..b02c273112 100644
--- a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmObjectNotificationsTests.kt
+++ b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmObjectNotificationsTests.kt
@@ -42,6 +42,7 @@ import kotlin.test.Ignore
 import kotlin.test.Test
 import kotlin.test.assertContains
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertIs
 import kotlin.test.assertNotNull
 import kotlin.test.assertNull
@@ -301,4 +302,217 @@ class RealmObjectNotificationsTests : RealmEntityNotificationTests {
             c.close()
         }
     }
+
+    @Test
+    override fun keyPath_topLevelProperty() = runBlocking<Unit> {
+        val c = Channel<ObjectChange<Sample>>(1)
+        val obj: Sample = realm.write { copyToRealm(Sample()) }
+        val observer = async {
+            obj.asFlow(listOf("stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialObject<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(obj)!!.intField = 42
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(obj)!!.stringField = "Bar"
+        }
+        c.receiveOrFail().let { objectChange ->
+            assertIs<UpdatedObject<Sample>>(objectChange)
+            when (objectChange) {
+                is UpdatedObject -> {
+                    assertEquals(1, objectChange.changedFields.size)
+                    assertEquals("stringField", objectChange.changedFields.first())
+                    assertEquals("Bar", objectChange.obj.stringField)
+                }
+                else -> fail("Unexpected change: $objectChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_nestedProperty() = runBlocking<Unit> {
+        val c = Channel<ObjectChange<Sample>>(1)
+        val obj: Sample = realm.write {
+            copyToRealm(
+                Sample().apply {
+                    this.stringField = "parent"
+                    this.nullableObject = Sample().apply {
+                        this.stringField = "child"
+                    }
+                }
+            )
+        }
+        val observer = async {
+            obj.asFlow(listOf("nullableObject.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialObject<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(obj)!!.stringField = "Parent change"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(obj)!!.nullableObject!!.stringField = "Bar"
+        }
+        c.receiveOrFail().let { objectChange ->
+            assertIs<UpdatedObject<Sample>>(objectChange)
+            when (objectChange) {
+                is UpdatedObject -> {
+                    // Core will only report something changed to the top-level property.
+                    assertEquals(1, objectChange.changedFields.size)
+                    assertEquals("nullableObject", objectChange.changedFields.first())
+                    assertEquals("Bar", objectChange.obj.nullableObject!!.stringField)
+                }
+                else -> fail("Unexpected change: $objectChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_defaultDepth() = runBlocking<Unit> {
+        val c = Channel<ObjectChange<Sample>>(1)
+        val obj: Sample = realm.write {
+            copyToRealm(
+                Sample().apply {
+                    this.stringField = "parent"
+                    this.nullableObject = Sample().apply {
+                        this.stringField = "child"
+                        this.nullableObject = Sample().apply {
+                            this.stringField = "child-child"
+                            this.nullableObject = Sample().apply {
+                                this.stringField = "child-child-child"
+                                this.nullableObject = Sample().apply {
+                                    this.stringField = "child-child-child-child"
+                                    this.nullableObject = Sample().apply {
+                                        this.stringField = "BottomChild"
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            )
+        }
+        val observer = async {
+            // Default keypath
+            obj.asFlow().collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialObject<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(obj)!!.nullableObject!!.nullableObject!!.nullableObject!!.nullableObject!!.nullableObject!!.intField = 1
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(obj)!!.stringField = "Parent change"
+        }
+        c.receiveOrFail().let { objectChange ->
+            assertIs<UpdatedObject<Sample>>(objectChange)
+            when (objectChange) {
+                is UpdatedObject -> {
+                    // Default value is Realm, so if this event is triggered by the first write
+                    // this assert will fail
+                    assertEquals("Parent change", objectChange.obj.stringField)
+                }
+                else -> fail("Unexpected change: $objectChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_propertyBelowDefaultLimit() = runBlocking<Unit> {
+        val c = Channel<ObjectChange<Sample>>(1)
+        val obj: Sample = realm.write {
+            copyToRealm(
+                Sample().apply {
+                    this.stringField = "parent"
+                    this.nullableObject = Sample().apply {
+                        this.stringField = "child"
+                        this.nullableObject = Sample().apply {
+                            this.stringField = "child-child"
+                            this.nullableObject = Sample().apply {
+                                this.stringField = "child-child-child"
+                                this.nullableObject = Sample().apply {
+                                    this.stringField = "child-child-child-child"
+                                    this.nullableObject = Sample().apply {
+                                        this.stringField = "BottomChild"
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            )
+        }
+        val observer = async {
+            obj.asFlow(listOf("nullableObject.nullableObject.nullableObject.nullableObject.nullableObject.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialObject<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(obj)!!.stringField = "Parent change"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(obj)!!.nullableObject!!.nullableObject!!.nullableObject!!.nullableObject!!.nullableObject!!.stringField = "Bar"
+        }
+        c.receiveOrFail().let { objectChange ->
+            assertIs<UpdatedObject<Sample>>(objectChange)
+            when (objectChange) {
+                is UpdatedObject -> {
+                    // Core will only report something changed to the top-level property.
+                    assertEquals(1, objectChange.changedFields.size)
+                    assertEquals("nullableObject", objectChange.changedFields.first())
+                    assertEquals("Bar", objectChange.obj.nullableObject!!.nullableObject!!.nullableObject!!.nullableObject!!.nullableObject!!.stringField)
+                }
+                else -> fail("Unexpected change: $objectChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_unknownTopLevelProperty() = runBlocking<Unit> {
+        val obj: Sample = realm.write { copyToRealm(Sample()) }
+        assertFailsWith<IllegalArgumentException>() {
+            obj.asFlow(listOf("foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_unknownNestedProperty() = runBlocking<Unit> {
+        val obj: Sample = realm.write { copyToRealm(Sample()) }
+        assertFailsWith<IllegalArgumentException>() {
+            obj.asFlow(listOf("nullableObject.foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_invalidNestedProperty() = runBlocking<Unit> {
+        val obj: Sample = realm.write { copyToRealm(Sample()) }
+        assertFailsWith<IllegalArgumentException> {
+            obj.asFlow(listOf("nullableObject.intField.foo"))
+        }
+        assertFailsWith<IllegalArgumentException> {
+            obj.asFlow(listOf("nullableObject.intListField.foo"))
+        }
+    }
 }
diff --git a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmResultsNotificationsTests.kt b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmResultsNotificationsTests.kt
index 4ed048123f..992491cbe6 100644
--- a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmResultsNotificationsTests.kt
+++ b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmResultsNotificationsTests.kt
@@ -22,6 +22,7 @@ import io.realm.kotlin.RealmConfiguration
 import io.realm.kotlin.entities.Sample
 import io.realm.kotlin.entities.list.RealmListContainer
 import io.realm.kotlin.entities.list.listTestSchema
+import io.realm.kotlin.ext.asFlow
 import io.realm.kotlin.ext.query
 import io.realm.kotlin.notifications.InitialResults
 import io.realm.kotlin.notifications.ListChangeSet.Range
@@ -32,6 +33,7 @@ import io.realm.kotlin.test.common.OBJECT_VALUES
 import io.realm.kotlin.test.common.OBJECT_VALUES2
 import io.realm.kotlin.test.common.OBJECT_VALUES3
 import io.realm.kotlin.test.common.utils.FlowableTests
+import io.realm.kotlin.test.common.utils.KeyPathFlowableTests
 import io.realm.kotlin.test.common.utils.assertIsChangeSet
 import io.realm.kotlin.test.platform.PlatformUtils
 import io.realm.kotlin.test.util.receiveOrFail
@@ -44,12 +46,13 @@ import kotlin.test.BeforeTest
 import kotlin.test.Ignore
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertIs
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
 import kotlin.test.fail
 
-class RealmResultsNotificationsTests : FlowableTests {
+class RealmResultsNotificationsTests : FlowableTests, KeyPathFlowableTests {
 
     lateinit var tmpDir: String
     lateinit var configuration: RealmConfiguration
@@ -387,4 +390,223 @@ class RealmResultsNotificationsTests : FlowableTests {
             c.close()
         }
     }
+
+    @Test
+    override fun keyPath_topLevelProperty() = runBlocking<Unit> {
+        val c = Channel<ResultsChange<Sample>>(1)
+        realm.write { copyToRealm(Sample()) }
+        val results = realm.query<Sample>().find()
+        val observer = async {
+            results.asFlow(listOf("stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialResults<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(results.first())!!.intField = 42
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(results.first())!!.stringField = "Bar"
+        }
+        c.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<Sample>>(resultsChange)
+            when (resultsChange) {
+                is UpdatedResults -> {
+                    assertEquals(1, resultsChange.changes.size)
+                }
+                else -> fail("Unexpected change: $resultsChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_nestedProperty() = runBlocking<Unit> {
+        val c = Channel<ResultsChange<Sample>>(1)
+        realm.write {
+            copyToRealm(
+                Sample().apply {
+                    this.stringField = "parent"
+                    this.nullableObject = Sample().apply {
+                        this.stringField = "child"
+                    }
+                }
+            )
+        }
+        val results = realm.query<Sample>("stringField != 'child'").find()
+        assertEquals(1, results.size)
+        val observer = async {
+            results.asFlow(listOf("nullableObject.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialResults<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(results.first())!!.stringField = "Parent change"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(results.first())!!.nullableObject!!.stringField = "Bar"
+        }
+        c.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<Sample>>(resultsChange)
+            when (resultsChange) {
+                is UpdatedResults -> {
+                    assertEquals(1, resultsChange.changes.size)
+                }
+                else -> fail("Unexpected change: $resultsChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    override fun keyPath_defaultDepth() = runBlocking<Unit> {
+        val c = Channel<ResultsChange<Sample>>(1)
+        realm.write {
+            copyToRealm(
+                Sample().apply {
+                    this.intField = 1
+                    this.stringField = "parent"
+                    this.nullableObject = Sample().apply {
+                        this.stringField = "child"
+                        this.nullableObject = Sample().apply {
+                            this.stringField = "child-child"
+                            this.nullableObject = Sample().apply {
+                                this.stringField = "child-child-child"
+                                this.nullableObject = Sample().apply {
+                                    this.stringField = "child-child-child-child"
+                                    this.nullableObject = Sample().apply {
+                                        this.stringField = "BottomChild"
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            )
+        }
+        val results = realm.query<Sample>("intField = 1").find()
+        assertEquals(1, results.size)
+        val observer = async {
+            // Default keypath
+            results.asFlow().collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialResults<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update below the default limit should not trigger a notification
+            findLatest(results.first())!!.nullableObject!!.nullableObject!!.nullableObject!!.nullableObject!!.nullableObject!!.stringField = "Bar"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(results.first())!!.intField = 1
+        }
+        c.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<Sample>>(resultsChange)
+            when (resultsChange) {
+                is ResultsChange<*> -> {
+                    // Core will only report something changed to the top-level property.
+                    assertEquals(1, resultsChange.changes.size)
+                    // Default value is 42, so if this event is triggered by the first write
+                    // this assert will fail
+                    assertEquals(1, resultsChange.list.first().intField)
+                }
+                else -> fail("Unexpected change: $resultsChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_propertyBelowDefaultLimit() = runBlocking<Unit> {
+        val c = Channel<ResultsChange<Sample>>(1)
+        realm.write {
+            copyToRealm(
+                Sample().apply {
+                    this.intField = 1
+                    this.stringField = "parent"
+                    this.nullableObject = Sample().apply {
+                        this.stringField = "child"
+                        this.nullableObject = Sample().apply {
+                            this.stringField = "child-child"
+                            this.nullableObject = Sample().apply {
+                                this.stringField = "child-child-child"
+                                this.nullableObject = Sample().apply {
+                                    this.stringField = "child-child-child-child"
+                                    this.nullableObject = Sample().apply {
+                                        this.stringField = "BottomChild"
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            )
+        }
+        val results = realm.query<Sample>("intField = 1").find()
+        assertEquals(1, results.size)
+        val observer = async {
+            results.asFlow(listOf("nullableObject.nullableObject.nullableObject.nullableObject.nullableObject.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialResults<Sample>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(results.first())!!.stringField = "Parent change"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(results.first())!!.nullableObject!!.nullableObject!!.nullableObject!!.nullableObject!!.nullableObject!!.stringField = "Bar"
+        }
+        c.receiveOrFail().let { resultsChange ->
+            assertIs<UpdatedResults<Sample>>(resultsChange)
+            when (resultsChange) {
+                is ResultsChange<*> -> {
+                    // Core will only report something changed to the top-level property.
+                    assertEquals(1, resultsChange.changes.size)
+                }
+                else -> fail("Unexpected change: $resultsChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_unknownTopLevelProperty() = runBlocking<Unit> {
+        realm.write { copyToRealm(Sample()) }
+        val results = realm.query<Sample>()
+        assertFailsWith<IllegalArgumentException>() {
+            results.asFlow(listOf("foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_unknownNestedProperty() = runBlocking<Unit> {
+        realm.write { copyToRealm(Sample()) }
+        val results = realm.query<Sample>()
+        assertFailsWith<IllegalArgumentException>() {
+            results.asFlow(listOf("nullableObject.foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_invalidNestedProperty() = runBlocking<Unit> {
+        realm.write { copyToRealm(Sample()) }
+        val results = realm.query<Sample>()
+        assertFailsWith<IllegalArgumentException> {
+            results.asFlow(listOf("nullableObject.intField.foo"))
+        }
+        assertFailsWith<IllegalArgumentException> {
+            results.asFlow(listOf("nullableObject.intListField.foo"))
+        }
+    }
 }
diff --git a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmSetNotificationsTests.kt b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmSetNotificationsTests.kt
index f4dd164bea..d2256eac9d 100644
--- a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmSetNotificationsTests.kt
+++ b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmSetNotificationsTests.kt
@@ -19,6 +19,7 @@ package io.realm.kotlin.test.common.notifications
 import io.realm.kotlin.Realm
 import io.realm.kotlin.RealmConfiguration
 import io.realm.kotlin.entities.set.RealmSetContainer
+import io.realm.kotlin.ext.realmSetOf
 import io.realm.kotlin.notifications.DeletedSet
 import io.realm.kotlin.notifications.InitialSet
 import io.realm.kotlin.notifications.SetChange
@@ -29,6 +30,7 @@ import io.realm.kotlin.test.common.SET_OBJECT_VALUES3
 import io.realm.kotlin.test.common.utils.RealmEntityNotificationTests
 import io.realm.kotlin.test.platform.PlatformUtils
 import io.realm.kotlin.test.util.receiveOrFail
+import io.realm.kotlin.types.RealmSet
 import kotlinx.coroutines.async
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.first
@@ -41,6 +43,7 @@ import kotlin.test.BeforeTest
 import kotlin.test.Ignore
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertIs
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
@@ -342,4 +345,278 @@ class RealmSetNotificationsTests : RealmEntityNotificationTests {
             channel.close()
         }
     }
+
+    @Test
+    override fun keyPath_topLevelProperty() = runBlocking<Unit> {
+        val c = Channel<SetChange<RealmSetContainer>>(1)
+        val set: RealmSet<RealmSetContainer> = realm.write {
+            copyToRealm(
+                RealmSetContainer().apply {
+                    this.objectSetField = realmSetOf(
+                        RealmSetContainer().apply { this.stringField = "set-item-1" },
+                        RealmSetContainer().apply { this.stringField = "set-item-2" }
+                    )
+                }
+            )
+        }.objectSetField
+        val observer = async {
+            set.asFlow(listOf("stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialSet<RealmSetContainer>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(set.first())!!.id = 42
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(set.first())!!.stringField = "Foo"
+        }
+        c.receiveOrFail().let { setChange ->
+            assertIs<UpdatedSet<RealmSetContainer>>(setChange)
+            when (setChange) {
+                is UpdatedSet -> {
+                    assertEquals(0, setChange.deletions)
+                    assertEquals(0, setChange.insertions)
+                    assertNotNull(setChange.set.firstOrNull { it.stringField == "Foo" })
+                }
+                else -> fail("Unexpected change: $setChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_nestedProperty() = runBlocking<Unit> {
+        val c = Channel<SetChange<RealmSetContainer>>(1)
+        val set = realm.write {
+            copyToRealm(
+                RealmSetContainer().apply {
+                    this.stringField = "parent"
+                    this.objectSetField = realmSetOf(
+                        RealmSetContainer().apply {
+                            this.stringField = "child"
+                            this.objectSetField = realmSetOf(
+                                RealmSetContainer().apply { this.stringField = "list-item-1" }
+                            )
+                        }
+                    )
+                }
+            )
+        }.objectSetField
+        assertEquals(1, set.size)
+        val observer = async {
+            set.asFlow(listOf("objectSetField.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialSet<RealmSetContainer>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(set.first())!!.id = 1
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            findLatest(set.first())!!.objectSetField.first().stringField = "Bar"
+        }
+        c.receiveOrFail().let { setChange ->
+            assertIs<UpdatedSet<RealmSetContainer>>(setChange)
+            when (setChange) {
+                is UpdatedSet -> {
+                    assertEquals(0, setChange.insertions)
+                    assertEquals(0, setChange.deletions)
+                    assertEquals("Bar", setChange.set.first().objectSetField.first().stringField)
+                }
+                else -> fail("Unexpected change: $setChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_defaultDepth() = runBlocking<Unit> {
+        val c = Channel<SetChange<RealmSetContainer>>(1)
+        val objectSet: RealmSet<RealmSetContainer> = realm.write {
+            copyToRealm(
+                RealmSetContainer().apply {
+                    this.id = 1
+                    this.stringField = "parent"
+                    this.objectSetField = realmSetOf(
+                        RealmSetContainer().apply {
+                            this.stringField = "child"
+                            this.objectSetField = realmSetOf(
+                                RealmSetContainer().apply {
+                                    this.stringField = "child-child"
+                                    this.objectSetField = realmSetOf(
+                                        RealmSetContainer().apply {
+                                            this.stringField = "child-child-child"
+                                            this.objectSetField = realmSetOf(
+                                                RealmSetContainer().apply {
+                                                    this.stringField = "child-child-child-child"
+                                                    this.objectSetField = realmSetOf(
+                                                        RealmSetContainer().apply {
+                                                            this.stringField = "child-child-child-child-child"
+                                                            this.objectSetField = realmSetOf(
+                                                                RealmSetContainer().apply {
+                                                                    this.stringField = "BottomChild"
+                                                                }
+                                                            )
+                                                        }
+                                                    )
+                                                }
+                                            )
+                                        }
+                                    )
+                                }
+                            )
+                        }
+                    )
+                }
+            )
+        }.objectSetField
+        val observer = async {
+            // Default keypath
+            objectSet.asFlow().collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialSet<RealmSetContainer>>(c.receiveOrFail())
+        realm.write {
+            // Update below the default limit should not trigger a notification
+            val obj = findLatest(objectSet.first())!!.objectSetField.first().objectSetField.first().objectSetField.first().objectSetField.first().objectSetField.first()
+            obj.stringField = "Bar"
+        }
+        realm.write {
+            findLatest(objectSet.first())!!.stringField = "Bar"
+        }
+        realm.write {
+        }
+        c.receiveOrFail().let { setChange ->
+            assertIs<UpdatedSet<RealmSetContainer>>(setChange)
+            when (setChange) {
+                is SetChange -> {
+                    // Core will only report something changed to the top-level property.
+                    assertEquals(0, setChange.insertions)
+                    assertEquals(0, setChange.deletions)
+                    // Default value is Realm, so if this event is triggered by the first write
+                    // this assert will fail
+                    assertEquals("Bar", setChange.set.first().stringField)
+                }
+                else -> fail("Unexpected change: $setChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_propertyBelowDefaultLimit() = runBlocking<Unit> {
+        val c = Channel<SetChange<RealmSetContainer>>(1)
+        val list = realm.write {
+            copyToRealm(
+                RealmSetContainer().apply {
+                    this.id = 1
+                    this.stringField = "parent"
+                    this.objectSetField = realmSetOf(
+                        RealmSetContainer().apply {
+                            this.stringField = "child"
+                            this.objectSetField = realmSetOf(
+                                RealmSetContainer().apply {
+                                    this.stringField = "child-child"
+                                    this.objectSetField = realmSetOf(
+                                        RealmSetContainer().apply {
+                                            this.stringField = "child-child-child"
+                                            this.objectSetField = realmSetOf(
+                                                RealmSetContainer().apply {
+                                                    this.stringField = "child-child-child-child"
+                                                    this.objectSetField = realmSetOf(
+                                                        RealmSetContainer().apply {
+                                                            this.stringField = "child-child-child-child-child"
+                                                            this.objectSetField = realmSetOf(
+                                                                RealmSetContainer().apply {
+                                                                    this.stringField = "BottomChild"
+                                                                }
+                                                            )
+                                                        }
+                                                    )
+                                                }
+                                            )
+                                        }
+                                    )
+                                }
+                            )
+                        }
+                    )
+                }
+            )
+        }.objectSetField
+        val observer = async {
+            list.asFlow(listOf("objectSetField.objectSetField.objectSetField.objectSetField.objectSetField.stringField")).collect {
+                c.trySend(it)
+            }
+        }
+        assertIs<InitialSet<RealmSetContainer>>(c.receiveOrFail())
+        realm.write {
+            // Update field that should not trigger a notification
+            findLatest(list.first())!!.stringField = "Parent change"
+        }
+        realm.write {
+            // Update field that should trigger a notification
+            val obj = findLatest(list.first())!!.objectSetField.first().objectSetField.first().objectSetField.first().objectSetField.first().objectSetField.first()
+            obj.stringField = "Bar"
+        }
+        c.receiveOrFail().let { setChange ->
+            assertIs<UpdatedSet<RealmSetContainer>>(setChange)
+            when (setChange) {
+                is SetChange -> {
+                    // Core will only report something changed to the top-level property.
+                    assertEquals(0, setChange.insertions)
+                    assertEquals(0, setChange.deletions)
+                    assertEquals(
+                        "Bar",
+                        setChange.set.first()
+                            .objectSetField.first()
+                            .objectSetField.first()
+                            .objectSetField.first()
+                            .objectSetField.first()
+                            .objectSetField.first()
+                            .stringField
+                    )
+                }
+                else -> fail("Unexpected change: $setChange")
+            }
+        }
+        observer.cancel()
+        c.close()
+    }
+
+    @Test
+    override fun keyPath_unknownTopLevelProperty() = runBlocking<Unit> {
+        val set = realm.write { copyToRealm(RealmSetContainer()) }.objectSetField
+        assertFailsWith<IllegalArgumentException>() {
+            set.asFlow(listOf("foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_unknownNestedProperty() = runBlocking<Unit> {
+        val set = realm.write { copyToRealm(RealmSetContainer()) }.objectSetField
+        assertFailsWith<IllegalArgumentException>() {
+            set.asFlow(listOf("objectSetField.foo"))
+        }
+    }
+
+    @Test
+    override fun keyPath_invalidNestedProperty() = runBlocking<Unit> {
+        val set = realm.write { copyToRealm(RealmSetContainer()) }.objectSetField
+        assertFailsWith<IllegalArgumentException> {
+            set.asFlow(listOf("id.foo"))
+        }
+        assertFailsWith<IllegalArgumentException> {
+            set.asFlow(listOf("objectSetField.intSetField.foo"))
+        }
+    }
 }
diff --git a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/utils/KeyPathFlowableTests.kt b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/utils/KeyPathFlowableTests.kt
new file mode 100644
index 0000000000..696f8601b9
--- /dev/null
+++ b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/utils/KeyPathFlowableTests.kt
@@ -0,0 +1,31 @@
+package io.realm.kotlin.test.common.utils
+
+import kotlin.test.Test
+
+/**
+ * All tests classes that tests classes exposing keypath notifications (RealmObject, RealmResults,
+ * RealmList, RealmSet, RealmMap) should implement this interface to be sure that we test common
+ * behaviour across those classes.
+ */
+interface KeyPathFlowableTests {
+    @Test
+    fun keyPath_topLevelProperty()
+
+    @Test
+    fun keyPath_nestedProperty()
+
+    @Test
+    fun keyPath_defaultDepth()
+
+    @Test
+    fun keyPath_propertyBelowDefaultLimit()
+
+    @Test
+    fun keyPath_unknownTopLevelProperty()
+
+    @Test
+    fun keyPath_unknownNestedProperty()
+
+    @Test
+    fun keyPath_invalidNestedProperty()
+}
diff --git a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/utils/RealmEntityNotificationTests.kt b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/utils/RealmEntityNotificationTests.kt
index 76a769d501..9f69e83046 100644
--- a/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/utils/RealmEntityNotificationTests.kt
+++ b/packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/utils/RealmEntityNotificationTests.kt
@@ -22,7 +22,7 @@ package io.realm.kotlin.test.common.utils
  * RealmResults) should implement this interface to be sure that we test common behaviour across
  * those classes.
  */
-interface RealmEntityNotificationTests : FlowableTests {
+interface RealmEntityNotificationTests : FlowableTests, KeyPathFlowableTests {
     // Verify that we get deletion events and close the Flow when the object being observed (or
     // containing object) is deleted.
     fun deleteEntity()