diff --git a/CHANGES.txt b/CHANGES.txt
index fae5591649..d7a7a61e19 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 5.0.5
+ * Avoid purging deletions in RowFilter when reconciliation is required (CASSANDRA-20541)
  * Fixed multiple single-node SAI query bugs relating to static columns (CASSANDRA-20338)
  * Upgrade com.datastax.cassandra:cassandra-driver-core:3.11.5 to org.apache.cassandra:cassandra-driver-core:3.12.1 (CASSANDRA-17231)
 Merged from 4.0:
diff --git a/src/java/org/apache/cassandra/db/filter/RowFilter.java b/src/java/org/apache/cassandra/db/filter/RowFilter.java
index 9b83a56f23..f1b095920f 100644
--- a/src/java/org/apache/cassandra/db/filter/RowFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/RowFilter.java
@@ -257,7 +257,10 @@ public class RowFilter implements Iterable<RowFilter.Expression>
             @Override
             public Row applyToRow(Row row)
             {
-                Row purged = row.purge(DeletionPurger.PURGE_ALL, nowInSec, metadata.enforceStrictLiveness());
+                // If we purge deletions when reconciliation is required, we hide information replica filtering
+                // protection would require to filter rows that are no longer matches are the coordinator.
+                Row purged = needsReconciliation() ? row : row.purge(DeletionPurger.PURGE_ALL, nowInSec, metadata.enforceStrictLiveness());
+
                 if (purged == null)
                     return null;
 
diff --git a/test/distributed/org/apache/cassandra/distributed/test/TableLevelIncrementalBackupsTest.java b/test/distributed/org/apache/cassandra/distributed/test/TableLevelIncrementalBackupsTest.java
index 2bf6c5f350..833b9484d0 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/TableLevelIncrementalBackupsTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/TableLevelIncrementalBackupsTest.java
@@ -141,12 +141,6 @@ public class TableLevelIncrementalBackupsTest extends TestBaseImpl
             cluster.get(i).flush(keyspace);
     }
 
-    private void disableCompaction(Cluster cluster, String keyspace, String table)
-    {
-        for (int i = 1; i < cluster.size() + 1; i++)
-            cluster.get(i).nodetool("disableautocompaction", keyspace, table);
-    }
-
     private static  void assertBackupSSTablesCount(Cluster cluster, int expectedTablesCount, boolean enable, String ks, String... tableNames)
     {
         for (int i = 1; i < cluster.size() + 1; i++)
diff --git a/test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java b/test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java
index 35c59046ea..6f36408e9e 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java
@@ -212,4 +212,10 @@ public class TestBaseImpl extends DistributedTestBase
         // in real live repair is needed in this case, but in the test case it doesn't matter if the tables loose
         // anything, so ignoring repair to speed up the tests.
     }
+
+    protected static void disableCompaction(Cluster cluster, String keyspace, String table)
+    {
+        for (int i = 1; i < cluster.size() + 1; i++)
+            cluster.get(i).nodetool("disableautocompaction", keyspace, table);
+    }
 }
diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java
index 59cc1cc6a5..1892157712 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java
@@ -44,6 +44,42 @@ public class ReplicaFilteringWithStaticsTest extends TestBaseImpl
         CLUSTER = init(Cluster.build(3).withConfig(config -> config.set("hinted_handoff_enabled", false).with(GOSSIP).with(NETWORK)).start());
     }
 
+    @Test
+    public void testRowFilterDeletePurging()
+    {
+        testRowFilterDeletePurging(false);
+    }
+
+    @Test
+    public void testRowFilterDeletePurgingSAI()
+    {
+        testRowFilterDeletePurging(true);
+    }
+
+    public void testRowFilterDeletePurging(boolean sai)
+    {
+        String table = "row_filtering_delete_purging" + (sai ? "_sai" : "");
+
+        CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s." + table + " (pk0 double, ck0 boolean, s0 ascii static, v0 ascii, " +
+                                          "PRIMARY KEY (pk0, ck0)) WITH CLUSTERING ORDER BY (ck0 DESC) AND read_repair = 'NONE'"));
+        disableCompaction(CLUSTER, KEYSPACE, table);
+
+        if (sai)
+        {
+            CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s." + table + "(s0) USING 'sai'"));
+            SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
+        }
+
+        CLUSTER.get(3).executeInternal(withKeyspace("UPDATE %s." + table + " USING TIMESTAMP 1 SET s0='foo', v0='c' WHERE  pk0 = 2.9 AND  ck0 IN (false, true)"));
+        
+        // This delete must be resolved by RFP to eliminate the row with ck0 = true from node 3:
+        CLUSTER.get(1).executeInternal(withKeyspace("DELETE FROM %s." + table + " USING TIMESTAMP 2 WHERE  pk0 = 2.9 AND  ck0 = true"));
+        CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s." + table + " (pk0, ck0, s0, v0) VALUES (2.9, false, 'bar', 'xyz') USING TIMESTAMP 3"));
+
+        String select = withKeyspace("SELECT ck0 FROM %s." + table + " WHERE s0 = 'bar' ALLOW FILTERING");
+        assertRows(CLUSTER.coordinator(1).executeWithPaging(select, ALL, 100), row(false));
+    }
+
     @Test
     public void testStaticMatchWithPartitionDelete()
     {