[VL] Refactor Velox metrics transport to use JSON payloads

backends-velox/src/main/java/org/apache/gluten/metrics/Metrics.java

@@ -16,209 +16,21 @@
  */
 package org.apache.gluten.metrics;
 
-import org.apache.gluten.exception.GlutenException;
-
 public class Metrics implements IMetrics {
-  public long[] inputRows;
-  public long[] inputVectors;
-  public long[] inputBytes;
-  public long[] rawInputRows;
-  public long[] rawInputBytes;
-  public long[] outputRows;
-  public long[] outputVectors;
-  public long[] outputBytes;
-  public long[] cpuCount;
-  public long[] wallNanos;
-  public long[] scanTime;
-  public long[] peakMemoryBytes;
-  public long[] numMemoryAllocations;
-  public long[] spilledInputBytes;
-  public long[] spilledBytes;
-  public long[] spilledRows;
-  public long[] spilledPartitions;
-  public long[] spilledFiles;
-  public long[] numDynamicFiltersProduced;
-  public long[] numDynamicFiltersAccepted;
-  public long[] numReplacedWithDynamicFilterRows;
-  public long[] numDynamicFilterInputRows;
-  public long[] flushRowCount;
-  public long[] abandonedPartialAggregationRows;
-  public long[] loadedToValueHook;
-  public long[] bloomFilterBlocksByteSize;
-  public long[] skippedSplits;
-  public long[] processedSplits;
-  public long[] skippedStrides;
-  public long[] processedStrides;
-  public long[] remainingFilterTime;
-  public long[] ioWaitTime;
-  public long[] storageReadBytes;
-  public long[] storageReads;
-  public long[] localReadBytes;
-  public long[] ramReadBytes;
-  public long[] preloadSplits;
-  public long[] pageLoadTime;
-  public long[] dataSourceAddSplitTime;
-  public long[] dataSourceReadTime;
-
-  public long[] physicalWrittenBytes;
-  public long[] writeIOTime;
-  public long[] numWrittenFiles;
-
-  public long[] loadLazyVectorTime;
-
-  public SingleMetric singleMetric = new SingleMetric();
+  public final String metricsJson;
+  public final int numMetrics;
+  public final SingleMetric singleMetric = new SingleMetric();
 
   public String taskStats;
 
   /** Create an instance for native metrics. */
-  public Metrics(
-      long[] inputRows,
-      long[] inputVectors,
-      long[] inputBytes,
-      long[] rawInputRows,
-      long[] rawInputBytes,
-      long[] outputRows,
-      long[] outputVectors,
-      long[] outputBytes,
-      long[] cpuCount,
-      long[] wallNanos,
-      long veloxToArrow,
-      long[] peakMemoryBytes,
-      long[] numMemoryAllocations,
-      long[] spilledInputBytes,
-      long[] spilledBytes,
-      long[] spilledRows,
-      long[] spilledPartitions,
-      long[] spilledFiles,
-      long[] numDynamicFiltersProduced,
-      long[] numDynamicFiltersAccepted,
-      long[] numReplacedWithDynamicFilterRows,
-      long[] numDynamicFilterInputRows,
-      long[] flushRowCount,
-      long[] abandonedPartialAggregationRows,
-      long[] loadedToValueHook,
-      long[] bloomFilterBlocksByteSize,
-      long[] scanTime,
-      long[] skippedSplits,
-      long[] processedSplits,
-      long[] skippedStrides,
-      long[] processedStrides,
-      long[] remainingFilterTime,
-      long[] ioWaitTime,
-      long[] storageReadBytes,
-      long[] storageReads,
-      long[] localReadBytes,
-      long[] ramReadBytes,
-      long[] preloadSplits,
-      long[] pageLoadTime,
-      long[] dataSourceAddSplitTime,
-      long[] dataSourceReadTime,
-      long[] physicalWrittenBytes,
-      long[] writeIOTime,
-      long[] numWrittenFiles,
-      long[] loadLazyVectorTime,
-      String taskStats) {
-    this.inputRows = inputRows;
-    this.inputVectors = inputVectors;
-    this.inputBytes = inputBytes;
-    this.rawInputRows = rawInputRows;
-    this.rawInputBytes = rawInputBytes;
-    this.outputRows = outputRows;
-    this.outputVectors = outputVectors;
-    this.outputBytes = outputBytes;
-    this.cpuCount = cpuCount;
-    this.wallNanos = wallNanos;
-    this.scanTime = scanTime;
+  public Metrics(String metricsJson, int numMetrics, long veloxToArrow, String taskStats) {
+    this.metricsJson = metricsJson;
+    this.numMetrics = numMetrics;
     this.singleMetric.veloxToArrow = veloxToArrow;
-    this.peakMemoryBytes = peakMemoryBytes;
-    this.numMemoryAllocations = numMemoryAllocations;
-    this.spilledInputBytes = spilledInputBytes;
-    this.spilledBytes = spilledBytes;
-    this.spilledRows = spilledRows;
-    this.spilledPartitions = spilledPartitions;
-    this.spilledFiles = spilledFiles;
-    this.numDynamicFiltersProduced = numDynamicFiltersProduced;
-    this.numDynamicFiltersAccepted = numDynamicFiltersAccepted;
-    this.numReplacedWithDynamicFilterRows = numReplacedWithDynamicFilterRows;
-    this.numDynamicFilterInputRows = numDynamicFilterInputRows;
-    this.flushRowCount = flushRowCount;
-    this.abandonedPartialAggregationRows = abandonedPartialAggregationRows;
-    this.loadedToValueHook = loadedToValueHook;
-    this.bloomFilterBlocksByteSize = bloomFilterBlocksByteSize;
-    this.skippedSplits = skippedSplits;
-    this.processedSplits = processedSplits;
-    this.skippedStrides = skippedStrides;
-    this.processedStrides = processedStrides;
-    this.remainingFilterTime = remainingFilterTime;
-    this.ioWaitTime = ioWaitTime;
-    this.storageReadBytes = storageReadBytes;
-    this.storageReads = storageReads;
-    this.localReadBytes = localReadBytes;
-    this.ramReadBytes = ramReadBytes;
-    this.preloadSplits = preloadSplits;
-    this.pageLoadTime = pageLoadTime;
-    this.dataSourceAddSplitTime = dataSourceAddSplitTime;
-    this.dataSourceReadTime = dataSourceReadTime;
-
-    this.physicalWrittenBytes = physicalWrittenBytes;
-    this.writeIOTime = writeIOTime;
-    this.numWrittenFiles = numWrittenFiles;
-    this.loadLazyVectorTime = loadLazyVectorTime;
     this.taskStats = taskStats;
   }
 
-  public OperatorMetrics getOperatorMetrics(int index) {
-    if (index >= inputRows.length) {
-      throw new GlutenException("Invalid index.");
-    }
-
-    return new OperatorMetrics(
-        inputRows[index],
-        inputVectors[index],
-        inputBytes[index],
-        rawInputRows[index],
-        rawInputBytes[index],
-        outputRows[index],
-        outputVectors[index],
-        outputBytes[index],
-        cpuCount[index],
-        wallNanos[index],
-        peakMemoryBytes[index],
-        numMemoryAllocations[index],
-        spilledInputBytes[index],
-        spilledBytes[index],
-        spilledRows[index],
-        spilledPartitions[index],
-        spilledFiles[index],
-        numDynamicFiltersProduced[index],
-        numDynamicFiltersAccepted[index],
-        numReplacedWithDynamicFilterRows[index],
-        numDynamicFilterInputRows[index],
-        flushRowCount[index],
-        abandonedPartialAggregationRows[index],
-        loadedToValueHook[index],
-        bloomFilterBlocksByteSize[index],
-        scanTime[index],
-        skippedSplits[index],
-        processedSplits[index],
-        skippedStrides[index],
-        processedStrides[index],
-        remainingFilterTime[index],
-        ioWaitTime[index],
-        storageReadBytes[index],
-        storageReads[index],
-        localReadBytes[index],
-        ramReadBytes[index],
-        preloadSplits[index],
-        pageLoadTime[index],
-        dataSourceAddSplitTime[index],
-        dataSourceReadTime[index],
-        physicalWrittenBytes[index],
-        writeIOTime[index],
-        numWrittenFiles[index],
-        loadLazyVectorTime[index]);
-  }
-
   public SingleMetric getSingleMetrics() {
     return singleMetric;
   }

backends-velox/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java

@@ -64,6 +64,9 @@ public class OperatorMetrics implements IOperatorMetrics {
 
   public long loadLazyVectorTime;
 
+  /** Create an empty instance for operator metrics. */
+  public OperatorMetrics() {}
+
   /** Create an instance for operator metrics. */
   public OperatorMetrics(
       long inputRows,