Convert DataSize for cpp worker

Author: shangm2

presto-native-execution/presto_cpp/main/PrestoTask.cpp

@@ -361,25 +361,21 @@ void updatePipelineStats(
     prestoOp.inputPositions = veloxOp.inputPositions;
     prestoOp.sumSquaredInputPositions =
         ((double)veloxOp.inputPositions) * veloxOp.inputPositions;
-    prestoOp.inputDataSize =
-        protocol::DataSize(veloxOp.inputBytes, protocol::DataUnit::BYTE);
+    prestoOp.inputDataSizeInBytes = veloxOp.inputBytes;
     prestoOp.rawInputPositions = veloxOp.rawInputPositions;
-    prestoOp.rawInputDataSize =
-        protocol::DataSize(veloxOp.rawInputBytes, protocol::DataUnit::BYTE);
+    prestoOp.rawInputDataSizeInBytes = veloxOp.rawInputBytes;
 
     // Report raw input statistics on the Project node following TableScan, if
     // exists.
     if (j == 1 && veloxOp.operatorType == "FilterProject" &&
         veloxPipelineStats.operatorStats[0].operatorType == "TableScan") {
       const auto& scanOp = veloxPipelineStats.operatorStats[0];
       prestoOp.rawInputPositions = scanOp.rawInputPositions;
-      prestoOp.rawInputDataSize =
-          protocol::DataSize(scanOp.rawInputBytes, protocol::DataUnit::BYTE);
+      prestoOp.rawInputDataSizeInBytes = scanOp.rawInputBytes;
     }
 
     prestoOp.outputPositions = veloxOp.outputPositions;
-    prestoOp.outputDataSize =
-        protocol::DataSize(veloxOp.outputBytes, protocol::DataUnit::BYTE);
+    prestoOp.outputDataSizeInBytes = veloxOp.outputBytes;
 
     setTiming(
         veloxOp.isBlockedTiming,
@@ -408,25 +404,14 @@ void updatePipelineStats(
     prestoOp.blockedWall = protocol::Duration(
         veloxOp.blockedWallNanos, protocol::TimeUnit::NANOSECONDS);
 
-    prestoOp.userMemoryReservation = protocol::DataSize(
-        veloxOp.memoryStats.userMemoryReservation, protocol::DataUnit::BYTE);
-    prestoOp.revocableMemoryReservation = protocol::DataSize(
-        veloxOp.memoryStats.revocableMemoryReservation,
-        protocol::DataUnit::BYTE);
-    prestoOp.systemMemoryReservation = protocol::DataSize(
-        veloxOp.memoryStats.systemMemoryReservation, protocol::DataUnit::BYTE);
-    prestoOp.peakUserMemoryReservation = protocol::DataSize(
-        veloxOp.memoryStats.peakUserMemoryReservation,
-        protocol::DataUnit::BYTE);
-    prestoOp.peakSystemMemoryReservation = protocol::DataSize(
-        veloxOp.memoryStats.peakSystemMemoryReservation,
-        protocol::DataUnit::BYTE);
-    prestoOp.peakTotalMemoryReservation = protocol::DataSize(
-        veloxOp.memoryStats.peakTotalMemoryReservation,
-        protocol::DataUnit::BYTE);
-
-    prestoOp.spilledDataSize =
-        protocol::DataSize(veloxOp.spilledBytes, protocol::DataUnit::BYTE);
+    prestoOp.userMemoryReservationInBytes = veloxOp.memoryStats.userMemoryReservation;
+    prestoOp.revocableMemoryReservationInBytes = veloxOp.memoryStats.revocableMemoryReservation;
+    prestoOp.systemMemoryReservationInBytes = veloxOp.memoryStats.systemMemoryReservation;
+    prestoOp.peakUserMemoryReservationInBytes = veloxOp.memoryStats.peakUserMemoryReservation;
+    prestoOp.peakSystemMemoryReservationInBytes = veloxOp.memoryStats.peakSystemMemoryReservation;
+    prestoOp.peakTotalMemoryReservationInBytes = veloxOp.memoryStats.peakTotalMemoryReservation;
+
+    prestoOp.spilledDataSizeInBytes = veloxOp.spilledBytes;
 
     if (veloxOp.operatorType == "HashBuild") {
       prestoOp.joinBuildKeyCount = veloxOp.inputPositions;

presto-native-execution/presto_cpp/main/tests/TaskManagerTest.cpp

@@ -101,7 +101,7 @@ int64_t sumOpSpillBytes(
       if (opStats.operatorType != opType) {
         continue;
       }
-      sum += opStats.spilledDataSize.getValue(protocol::DataUnit::BYTE);
+      sum += opStats.spilledDataSizeInBytes;
     }
   }
   return sum;