diff --git a/CHANGES.txt b/CHANGES.txt
index 6c627f596a..2fb55b032e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -171,6 +171,7 @@
  * Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
  * Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
 Merged from 5.0:
+ * 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)
  * Update netty to 4.1.119.Final and netty-tcnative to 2.0.70.Final (CASSANDRA-20314)
diff --git a/src/java/org/apache/cassandra/db/filter/RowFilter.java b/src/java/org/apache/cassandra/db/filter/RowFilter.java
index 4f6b7bd1d5..e843fe8379 100644
--- a/src/java/org/apache/cassandra/db/filter/RowFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/RowFilter.java
@@ -256,7 +256,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 f093410a62..d092757337 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java
@@ -240,4 +240,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..61f221c152 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java
@@ -27,6 +27,7 @@ import org.junit.Test;
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.api.ConsistencyLevel;
 import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.distributed.test.cql3.MultiNodeTableWalkWithoutReadRepairTest;
 
 import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL;
 import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
@@ -44,6 +45,45 @@ 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);
+    }
+
+    /**
+     * Originally discovered by {@link MultiNodeTableWalkWithoutReadRepairTest} with seed 6640281155419111674
+     */
+    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()
     {