Use cheaper operation to estimate json data byte size (airbytehq#13240)

* Simplify byte size estimation

* Format code

* Update comment

Author: tuliren

airbyte-commons/src/main/java/io/airbyte/commons/json/Jsons.java

@@ -129,6 +129,17 @@ public static byte[] toBytes(final JsonNode jsonNode) {
     return serialize(jsonNode).getBytes(Charsets.UTF_8);
   }
 
+  /**
+   * Use string length as an estimation for byte size, because all ASCII characters are one byte long
+   * in UTF-8, and ASCII characters cover most of the use cases. To be more precise, we can convert
+   * the string to byte[] and use the length of the byte[]. However, this conversion is expensive in
+   * memory consumption. Given that the byte size of the serialized JSON is already an estimation of
+   * the actual size of the JSON object, using a cheap operation seems an acceptable compromise.
+   */
+  public static int getEstimatedByteSize(final JsonNode jsonNode) {
+    return serialize(jsonNode).length();
+  }
+
   public static Set<String> keys(final JsonNode jsonNode) {
     if (jsonNode.isObject()) {
       return Jsons.object(jsonNode, new TypeReference<Map<String, Object>>() {}).keySet();

airbyte-commons/src/test/java/io/airbyte/commons/json/JsonsTest.java

@@ -246,6 +246,12 @@ void testGetStringOrNull() {
     assertNull(Jsons.getStringOrNull(json, "xyz"));
   }
 
+  @Test
+  void testGetEstimatedByteSize() {
+    final JsonNode json = Jsons.deserialize("{\"string_key\":\"abc\",\"array_key\":[\"item1\", \"item2\"]}");
+    assertEquals(Jsons.toBytes(json).length, Jsons.getEstimatedByteSize(json));
+  }
+
   private static class ToClass {
 
     @JsonProperty("str")

airbyte-workers/src/main/java/io/airbyte/workers/internal/AirbyteMessageTracker.java

@@ -111,10 +111,9 @@ private void handleSourceEmittedRecord(final AirbyteRecordMessage recordMessage)
     final long currentTotalCount = streamToTotalRecordsEmitted.getOrDefault(streamIndex, 0L);
     streamToTotalRecordsEmitted.put(streamIndex, currentTotalCount + 1);
 
-    // todo (cgardens) - pretty wasteful to do an extra serialization just to get size.
-    final int numBytes = Jsons.serialize(recordMessage.getData()).getBytes(Charsets.UTF_8).length;
+    final int estimatedNumBytes = Jsons.getEstimatedByteSize(recordMessage.getData());
     final long currentTotalStreamBytes = streamToTotalBytesEmitted.getOrDefault(streamIndex, 0L);
-    streamToTotalBytesEmitted.put(streamIndex, currentTotalStreamBytes + numBytes);
+    streamToTotalBytesEmitted.put(streamIndex, currentTotalStreamBytes + estimatedNumBytes);
   }
 
   /**

airbyte-workers/src/test/java/io/airbyte/workers/internal/AirbyteMessageTrackerTest.java

@@ -7,7 +7,6 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
-import com.google.common.base.Charsets;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.config.FailureReason;
 import io.airbyte.config.State;
@@ -53,7 +52,7 @@ public void testGetTotalRecordsStatesAndBytesEmitted() {
     messageTracker.acceptFromSource(s2);
 
     assertEquals(3, messageTracker.getTotalRecordsEmitted());
-    assertEquals(3 * Jsons.serialize(r1.getRecord().getData()).getBytes(Charsets.UTF_8).length, messageTracker.getTotalBytesEmitted());
+    assertEquals(3L * Jsons.getEstimatedByteSize(r1.getRecord().getData()), messageTracker.getTotalBytesEmitted());
     assertEquals(2, messageTracker.getTotalStateMessagesEmitted());
   }
 
@@ -112,9 +111,9 @@ public void testEmittedBytesByStream() {
     final AirbyteMessage r2 = AirbyteMessageUtils.createRecordMessage(STREAM_2, 2);
     final AirbyteMessage r3 = AirbyteMessageUtils.createRecordMessage(STREAM_3, 3);
 
-    final long r1Bytes = Jsons.serialize(r1.getRecord().getData()).getBytes(Charsets.UTF_8).length;
-    final long r2Bytes = Jsons.serialize(r2.getRecord().getData()).getBytes(Charsets.UTF_8).length;
-    final long r3Bytes = Jsons.serialize(r3.getRecord().getData()).getBytes(Charsets.UTF_8).length;
+    final long r1Bytes = Jsons.getEstimatedByteSize(r1.getRecord().getData());
+    final long r2Bytes = Jsons.getEstimatedByteSize(r2.getRecord().getData());
+    final long r3Bytes = Jsons.getEstimatedByteSize(r3.getRecord().getData());
 
     messageTracker.acceptFromSource(r1);
     messageTracker.acceptFromSource(r2);