diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 78e832ee1..43b421275 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -8,6 +8,17 @@ Changelog
 1.1
 ===
 
+1.1.9
+-----
+
+Fixed
+^^^^^
+- Fixed DELETE and UPDATE queries failing when filtering by related fields (foreign keys). Using a subquery pattern instead of JOIN for compatibility with MySQL and SQLite. (#283)
+
+Added
+^^^^^
+- Tests for model validators. (#2137)
+
 1.1.8
 -----
 
diff --git a/tests/test_queryset.py b/tests/test_queryset.py
index 8eaefb795..1e8cb31ae 100644
--- a/tests/test_queryset.py
+++ b/tests/test_queryset.py
@@ -519,6 +519,38 @@ async def test_delete_limit_order_by(db, intfields_data):
         await IntFields.get(intnum=97)
 
 
+@pytest.mark.asyncio
+async def test_delete_filter_with_foreign_key(db):
+    author = await Author.create(name="test")
+    await Book.create(name="book1", author=author, rating=5.0)
+    await Book.create(name="book2", author=author, rating=4.0)
+
+    author2 = await Author.create(name="test2")
+    await Book.create(name="book3", author=author2, rating=5.0)
+
+    # This is the failing query
+    await Book.filter(author__name="test").delete()
+
+    assert await Book.all().count() == 1
+
+
+@pytest.mark.asyncio
+async def test_update_filter_with_foreign_key(db):
+    author = await Author.create(name="test")
+    await Book.create(name="book1", author=author, rating=5.0)
+
+    author2 = await Author.create(name="test2")
+    await Book.create(name="book2", author=author2, rating=5.0)
+
+    await Book.filter(author__name="test").update(rating=1.0)
+
+    book = await Book.get(name="book1")
+    assert book.rating == 1.0
+
+    book2 = await Book.get(name="book2")
+    assert book2.rating == 5.0
+
+
 @pytest.mark.asyncio
 async def test_async_iter(db, intfields_data):
     counter = 0
diff --git a/tortoise/queryset.py b/tortoise/queryset.py
index fe7d4a3f7..a6cf92f1a 100644
--- a/tortoise/queryset.py
+++ b/tortoise/queryset.py
@@ -1355,6 +1355,26 @@ def _make_query(self) -> None:
             self.resolve_ordering(self.model, table, self._orderings, self._annotations)
 
         self.resolve_filters()
+        if self._joined_tables:
+            # If we have joins, we must use a subquery for update
+            # because standard UPDATE does not support JOINs on many DBs.
+            pk_column = self.model._meta.db_pk_column
+            subquery = self._db.query_class.from_(table).select(table[pk_column])
+            subquery._wheres = self.query._wheres
+            subquery._havings = self.query._havings
+            subquery._joins = self.query._joins
+            if hasattr(self.query, "_limit"):
+                subquery._limit = self.query._limit
+            if hasattr(self.query, "_orderbys"):
+                subquery._orderbys = self.query._orderbys
+
+            # To avoid MySQL Error 1093, we wrap the subquery in another SELECT
+            # To avoid MySQL Error 1235, the outer SELECT shouldn't have LIMIT
+            wrapper = self._db.query_class.from_(subquery.as_("_t")).select(Table("_t")[pk_column])
+
+            self.query = self._db.query_class.update(table)
+            self.query = self.query.where(table[pk_column].isin(wrapper))
+
         for key, value in self.update_kwargs.items():
             field_object = self.model._meta.fields_map.get(key)
             if not field_object:
@@ -1437,6 +1457,21 @@ def _make_query(self) -> None:
                 annotations=self._annotations,
             )
         self.resolve_filters()
+        if self._joined_tables:
+            # If we have joins, we must use a subquery for deletion
+            # because standard DELETE FROM does not support JOINs.
+            pk_column = self.model._meta.db_pk_column
+            subquery = self.query.select(self.model._meta.basetable[pk_column])
+
+            # To avoid MySQL Error 1093, we wrap the subquery in another SELECT
+            # To avoid MySQL Error 1235, the outer SELECT shouldn't have LIMIT
+            # We use the connection's query class directly to avoid carrying over
+            # the base table into the FROM clause.
+            wrapper = self._db.query_class.from_(subquery.as_("_t")).select(Table("_t")[pk_column])
+
+            self.query = copy(self.model._meta.basequery)
+            self.query = self.query.where(self.model._meta.basetable[pk_column].isin(wrapper))
+
         self.query._delete_from = True
         return