diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/SerializablePairLongStringDeltaEncodedStagedSerde.java b/processing/src/main/java/org/apache/druid/query/aggregation/SerializablePairLongStringDeltaEncodedStagedSerde.java
index 2354bd085991..5850dc741343 100644
--- a/processing/src/main/java/org/apache/druid/query/aggregation/SerializablePairLongStringDeltaEncodedStagedSerde.java
+++ b/processing/src/main/java/org/apache/druid/query/aggregation/SerializablePairLongStringDeltaEncodedStagedSerde.java
@@ -36,6 +36,14 @@
  *
  * otherwise
  * Long:Integer:bytes
+ *
+ * The StringSize can be following:
+ * -1 : Denotes an empty string
+ * 0  : Denotes a null string
+ * >0 : Denotes a non-empty string
+ *
+ * Mapping of null and empty string is done weirdly to preserve backward compatibility when nulls were returned all the
+ * time, and there was no distinction between empty and null string
  */
 public class SerializablePairLongStringDeltaEncodedStagedSerde implements StagedSerde<SerializablePairLongString>
 {
@@ -76,6 +84,8 @@ public void store(ByteBuffer byteBuffer)
         }
 
         if (rhsString == null) {
+          byteBuffer.putInt(0);
+        } else if (rhsBytes.length == 0) {
           byteBuffer.putInt(-1);
         } else {
           byteBuffer.putInt(rhsBytes.length);
@@ -116,11 +126,13 @@ public SerializablePairLongString deserialize(ByteBuffer byteBuffer)
     int stringSize = readOnlyBuffer.getInt();
     String lastString = null;
 
-    if (stringSize >= 0) {
+    if (stringSize > 0) {
       byte[] stringBytes = new byte[stringSize];
 
       readOnlyBuffer.get(stringBytes, 0, stringSize);
       lastString = StringUtils.fromUtf8(stringBytes);
+    } else if (stringSize < 0) {
+      lastString = "";
     }
 
     return new SerializablePairLongString(lhs, lastString);
diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/SerializablePairLongStringSimpleStagedSerde.java b/processing/src/main/java/org/apache/druid/query/aggregation/SerializablePairLongStringSimpleStagedSerde.java
index 0a065fdc0c62..8a33c73379e9 100644
--- a/processing/src/main/java/org/apache/druid/query/aggregation/SerializablePairLongStringSimpleStagedSerde.java
+++ b/processing/src/main/java/org/apache/druid/query/aggregation/SerializablePairLongStringSimpleStagedSerde.java
@@ -34,6 +34,14 @@
  * <p>
  * or
  * Long:StringSize:StringData
+ *
+ * The StringSize can be following:
+ * -1 : Denotes an empty string
+ * 0  : Denotes a null string
+ * >0 : Denotes a non-empty string
+ *
+ * Mapping of null and empty string is done weirdly to preserve backward compatibility when nulls were returned all the
+ * time, and there was no distinction between empty and null string
  */
 public class SerializablePairLongStringSimpleStagedSerde implements StagedSerde<SerializablePairLongString>
 {
@@ -56,6 +64,8 @@ public void store(ByteBuffer byteBuffer)
 
         byteBuffer.putLong(value.lhs);
         if (rhsString == null) {
+          byteBuffer.putInt(0);
+        } else if (rhsBytes.length == 0) {
           byteBuffer.putInt(-1);
         } else {
           byteBuffer.putInt(rhsBytes.length);
@@ -87,11 +97,13 @@ public SerializablePairLongString deserialize(ByteBuffer byteBuffer)
     int stringSize = readOnlyBuffer.getInt();
     String lastString = null;
 
-    if (stringSize >= 0) {
+    if (stringSize > 0) {
       byte[] stringBytes = new byte[stringSize];
 
       readOnlyBuffer.get(stringBytes, 0, stringSize);
       lastString = StringUtils.fromUtf8(stringBytes);
+    } else if (stringSize < 0) {
+      lastString = "";
     }
 
     return new SerializablePairLongString(lhs, lastString);
diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/BackwordCompatibleSerializablePairLongStringDeltaEncodedStagedSerdeTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/BackwardCompatibleSerializablePairLongStringDeltaEncodedStagedSerdeTest.java
similarity index 96%
rename from processing/src/test/java/org/apache/druid/query/aggregation/BackwordCompatibleSerializablePairLongStringDeltaEncodedStagedSerdeTest.java
rename to processing/src/test/java/org/apache/druid/query/aggregation/BackwardCompatibleSerializablePairLongStringDeltaEncodedStagedSerdeTest.java
index 452bfff0f226..bc8cd8a1627e 100644
--- a/processing/src/test/java/org/apache/druid/query/aggregation/BackwordCompatibleSerializablePairLongStringDeltaEncodedStagedSerdeTest.java
+++ b/processing/src/test/java/org/apache/druid/query/aggregation/BackwardCompatibleSerializablePairLongStringDeltaEncodedStagedSerdeTest.java
@@ -33,7 +33,7 @@
 import java.nio.ByteOrder;
 import java.util.Random;
 
-public class BackwordCompatibleSerializablePairLongStringDeltaEncodedStagedSerdeTest
+public class BackwardCompatibleSerializablePairLongStringDeltaEncodedStagedSerdeTest
 {
   private static final OlderSerializablePairLongStringDeltaEncodedStagedSerde OLDER_INTEGER_SERDE =
       new OlderSerializablePairLongStringDeltaEncodedStagedSerde(0L, true);
@@ -71,11 +71,11 @@ public void testNullString()
 
     // Write using the older serde, read using the newer serde
     Assert.assertEquals(
-        new SerializablePairLongString(TIMESTAMP, ""),
+        new SerializablePairLongString(TIMESTAMP, null),
         readUsingSerde(writeUsingSerde(value, OLDER_INTEGER_SERDE), INTEGER_SERDE)
     );
     Assert.assertEquals(
-        new SerializablePairLongString(TIMESTAMP, ""),
+        new SerializablePairLongString(TIMESTAMP, null),
         readUsingSerde(writeUsingSerde(value, OLDER_LONG_SERDE), LONG_SERDE)
     );
 
@@ -104,11 +104,11 @@ public void testEmptyString()
 
     // Write using the older serde, read using the newer serde
     Assert.assertEquals(
-        new SerializablePairLongString(TIMESTAMP, ""),
+        new SerializablePairLongString(TIMESTAMP, null),
         readUsingSerde(writeUsingSerde(value, OLDER_INTEGER_SERDE), INTEGER_SERDE)
     );
     Assert.assertEquals(
-        new SerializablePairLongString(TIMESTAMP, ""),
+        new SerializablePairLongString(TIMESTAMP, null),
         readUsingSerde(writeUsingSerde(value, OLDER_LONG_SERDE), LONG_SERDE)
     );
 
diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/BackwordCompatibleSerializablePairLongStringSimpleStagedSerdeTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/BackwardCompatibleSerializablePairLongStringSimpleStagedSerdeTest.java
similarity index 98%
rename from processing/src/test/java/org/apache/druid/query/aggregation/BackwordCompatibleSerializablePairLongStringSimpleStagedSerdeTest.java
rename to processing/src/test/java/org/apache/druid/query/aggregation/BackwardCompatibleSerializablePairLongStringSimpleStagedSerdeTest.java
index a898c9929e97..db5e5490f653 100644
--- a/processing/src/test/java/org/apache/druid/query/aggregation/BackwordCompatibleSerializablePairLongStringSimpleStagedSerdeTest.java
+++ b/processing/src/test/java/org/apache/druid/query/aggregation/BackwardCompatibleSerializablePairLongStringSimpleStagedSerdeTest.java
@@ -32,7 +32,7 @@
 import java.nio.ByteOrder;
 import java.util.Random;
 
-public class BackwordCompatibleSerializablePairLongStringSimpleStagedSerdeTest
+public class BackwardCompatibleSerializablePairLongStringSimpleStagedSerdeTest
 {
   private static final OlderSerializablePairLongStringSimpleStagedSerde OLDER_SERDE =
       new OlderSerializablePairLongStringSimpleStagedSerde();
diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/SerializablePairLongStringDeltaEncodedStagedSerdeTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/SerializablePairLongStringDeltaEncodedStagedSerdeTest.java
index c22e89f4140e..d9966412d25f 100644
--- a/processing/src/test/java/org/apache/druid/query/aggregation/SerializablePairLongStringDeltaEncodedStagedSerdeTest.java
+++ b/processing/src/test/java/org/apache/druid/query/aggregation/SerializablePairLongStringDeltaEncodedStagedSerdeTest.java
@@ -84,7 +84,7 @@ public void testNullStringLong()
   @Test
   public void testEmptyStringLong()
   {
-    assertValueEquals(new SerializablePairLongString(100L, ""), 8, LONG_SERDE);
+    assertValueEquals(new SerializablePairLongString(100L, ""), 12, LONG_SERDE);
   }