diff --git a/CHANGES.txt b/CHANGES.txt
index d8ff60e08f..6d6c736e1e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -44,6 +44,7 @@
  * Add guardrail for ALTER TABLE ADD / DROP / REMOVE column operations (CASSANDRA-17495)
  * Rename DisableFlag class to EnableFlag on guardrails (CASSANDRA-17544)
 Merged from 4.1:
+ * upsert with adder support is not consistent with numbers and strings in LWT (CASSANDRA-17857)
  * Fix race and return after failing connections (CASSANDRA-17618)
  * Speculative execution threshold unit mismatch (CASSANDRA-17877)
  * Fix BulkLoader to load entireSSTableThrottle and entireSSTableInterDcThrottle (CASSANDRA-17677)
diff --git a/src/java/org/apache/cassandra/cql3/Constants.java b/src/java/org/apache/cassandra/cql3/Constants.java
index e8989ad24e..327b3f7bef 100644
--- a/src/java/org/apache/cassandra/cql3/Constants.java
+++ b/src/java/org/apache/cassandra/cql3/Constants.java
@@ -485,6 +485,8 @@ public abstract class Constants
                 @SuppressWarnings("unchecked") NumberType<Number> type = (NumberType<Number>) column.type;
                 ByteBuffer increment = t.bindAndGet(params.options);
                 ByteBuffer current = getCurrentCellBuffer(partitionKey, params);
+                if (current == null)
+                    return;
                 ByteBuffer newValue = type.add(type, current, type, increment);
                 params.addCell(column, newValue);
             }
@@ -492,17 +494,11 @@ public abstract class Constants
             {
                 ByteBuffer append = t.bindAndGet(params.options);
                 ByteBuffer current = getCurrentCellBuffer(partitionKey, params);
-                ByteBuffer newValue;
                 if (current == null)
-                {
-                    newValue = append;
-                }
-                else
-                {
-                    newValue = ByteBuffer.allocate(current.remaining() + append.remaining());
-                    FastByteOperations.copy(current, current.position(), newValue, newValue.position(), current.remaining());
-                    FastByteOperations.copy(append, append.position(), newValue, newValue.position() + current.remaining(), append.remaining());
-                }
+                    return;
+                ByteBuffer newValue = ByteBuffer.allocate(current.remaining() + append.remaining());
+                FastByteOperations.copy(current, current.position(), newValue, newValue.position(), current.remaining());
+                FastByteOperations.copy(append, append.position(), newValue, newValue.position() + current.remaining(), append.remaining());
                 params.addCell(column, newValue);
             }
         }
diff --git a/test/distributed/org/apache/cassandra/distributed/test/CASAddTest.java b/test/distributed/org/apache/cassandra/distributed/test/CASAddTest.java
index 59220cc24d..c50b9b526e 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/CASAddTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/CASAddTest.java
@@ -24,6 +24,9 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.utils.AssertionUtils;
+import org.assertj.core.api.Assertions;
 
 import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
 import static org.apache.cassandra.distributed.shared.AssertUtils.row;
@@ -62,6 +65,46 @@ public class CASAddTest extends TestBaseImpl
         }
     }
 
+    @Test
+    public void testAdditionNotExists() throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.create(3)))
+        {
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int PRIMARY KEY, a int, b text)");
+
+            // in this context partition/row not existing looks like column not existing, so to simplify the LWT required
+            // condition, add a row with null columns so can rely on IF EXISTS
+            cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk) VALUES (1)", ConsistencyLevel.QUORUM);
+
+            // n = n + value where n = null
+            cluster.coordinator(1).execute("UPDATE " + KEYSPACE + ".tbl SET a = a + 1, b = b + 'fail' WHERE pk = 1 IF EXISTS", ConsistencyLevel.QUORUM);
+            // the SET should all no-op due to null... so should no-op
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.SERIAL), row(1, null, null));
+
+            // this section is testing current limitations... if they start to fail due to the limitations going away... update this test to include those cases
+            Assertions.assertThatThrownBy(() -> cluster.coordinator(1).execute(batch(
+                      "INSERT INTO " + KEYSPACE + ".tbl (pk, a, b) VALUES (1, 0, '') IF NOT EXISTS",
+                      "UPDATE " + KEYSPACE + ".tbl SET a = a + 1, b = b + 'success' WHERE pk = 1 IF EXISTS"
+                      ), ConsistencyLevel.QUORUM))
+                      .is(AssertionUtils.is(InvalidRequestException.class))
+                      .hasMessage("Cannot mix IF EXISTS and IF NOT EXISTS conditions for the same row");
+            Assertions.assertThatThrownBy(() -> cluster.coordinator(1).execute(batch(
+                      "INSERT INTO " + KEYSPACE + ".tbl (pk, a, b) VALUES (1, 0, '') IF NOT EXISTS",
+
+                      "UPDATE " + KEYSPACE + ".tbl SET a = a + 1, b = b + 'success' WHERE pk = 1"
+                      ), ConsistencyLevel.QUORUM))
+                      .is(AssertionUtils.is(InvalidRequestException.class))
+                      .hasMessage("Invalid operation (a = a + 1) for non counter column a");
+
+            // since CAS doesn't allow the above cases, manually add the data to unblock...
+            cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, a, b) VALUES (1, 0, '')", ConsistencyLevel.QUORUM);
+
+            // have cas add defaults when missing
+            cluster.coordinator(1).execute("UPDATE " + KEYSPACE + ".tbl SET a = a + 1, b = b + 'success' WHERE pk = 1 IF EXISTS", ConsistencyLevel.QUORUM);
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.SERIAL), row(1, 1, "success"));
+        }
+    }
+
     @Test
     public void testConcat() throws Throwable
     {
diff --git a/test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java b/test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java
index 4755d70f8c..e97a08105f 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java
@@ -104,6 +104,16 @@ public class TestBaseImpl extends DistributedTestBase
         return TupleType.buildValue(bbs);
     }
 
+    public static String batch(String... queries)
+    {
+        StringBuilder sb = new StringBuilder();
+        sb.append("BEGIN UNLOGGED BATCH\n");
+        for (String q : queries)
+            sb.append(q).append(";\n");
+        sb.append("APPLY BATCH;");
+        return sb.toString();
+    }
+
     protected void bootstrapAndJoinNode(Cluster cluster)
     {
         IInstanceConfig config = cluster.newInstanceConfig();
diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/UpdateTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/UpdateTest.java
index 027955780b..59a0616106 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/operations/UpdateTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/operations/UpdateTest.java
@@ -28,6 +28,8 @@ import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.cql3.UntypedResultSet.Row;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.assertj.core.api.Assertions;
 
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertEquals;
@@ -657,4 +659,14 @@ public class UpdateTest extends CQLTester
         ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(currentTable());
         return cfs.metric.allMemtablesLiveDataSize.getValue() == 0;
     }
+
+    @Test
+    public void testAdderNonCounter()
+    {
+        createTable("CREATE TABLE %s (pk int PRIMARY KEY, a int, b text)");
+        Assertions.assertThatThrownBy(() -> execute("UPDATE %s SET a = a + 1, b = b + 'fail' WHERE pk = 1"))
+                  .isInstanceOf(InvalidRequestException.class)
+                  // if error ever includes "b" its safe to update this test
+                  .hasMessage("Invalid operation (a = a + 1) for non counter column a");
+    }
 }