[Backport] Window Functions : Improve performance by comparing String…

…s in frame bytes without converting them (#17091) (#17311)

(cherry picked from commit b9a4c73)

Co-authored-by: Sree Charan Manamala <sree.manamala@imply.io>

benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlWindowFunctionsBenchmark.java

@@ -158,6 +158,12 @@ public int getNumMergeBuffers()
     {
       return 3;
     }
+
+    @Override
+    public int intermediateComputeSizeBytes()
+    {
+      return 200_000_000;
+    }
   };
 
   @Setup(Level.Trial)
@@ -334,7 +340,8 @@ public void tearDown() throws Exception
   public void querySql(String sql, Blackhole blackhole)
   {
     final Map<String, Object> context = ImmutableMap.of(
-        QueryContexts.MAX_SUBQUERY_BYTES_KEY, "auto"
+        QueryContexts.MAX_SUBQUERY_BYTES_KEY, "disabled",
+        QueryContexts.MAX_SUBQUERY_ROWS_KEY, -1
     );
     try (final DruidPlanner planner = plannerFactory.createPlannerForTesting(engine, sql, context)) {
       final PlannerResult plannerResult = planner.plan();
@@ -418,4 +425,15 @@ public void windowWithoutSorter(Blackhole blackhole)
                  + "GROUP BY dimUniform, dimSequential";
     querySql(sql, blackhole);
   }
+
+  @Benchmark
+  public void windowWithGroupbyTime(Blackhole blackhole)
+  {
+    String sql = "SELECT "
+                 + "SUM(dimSequentialHalfNull) + SUM(dimHyperUnique), "
+                 + "LAG(SUM(dimSequentialHalfNull + dimHyperUnique)) OVER (PARTITION BY dimUniform ORDER BY dimSequential) "
+                 + "FROM foo "
+                 + "GROUP BY __time, dimUniform, dimSequential";
+    querySql(sql, blackhole);
+  }
 }

processing/src/main/java/org/apache/druid/frame/read/columnar/StringFrameColumnReader.java

@@ -20,6 +20,7 @@
 package org.apache.druid.frame.read.columnar;
 
 import com.google.common.primitives.Ints;
+import org.apache.commons.lang.ObjectUtils;
 import org.apache.datasketches.memory.Memory;
 import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.error.DruidException;
@@ -507,6 +508,12 @@ protected Comparator<Object> getComparator()
       return Comparator.nullsFirst(Comparator.comparing(o -> ((String) o)));
     }
 
+    @Override
+    public int compareRows(int rowNum1, int rowNum2)
+    {
+      return ObjectUtils.compare(getStringUtf8(rowNum1), getStringUtf8(rowNum2));
+    }
+
     /**
      * Returns a ByteBuffer containing UTF-8 encoded string number {@code index}. The ByteBuffer is always newly
      * created, so it is OK to change its position, limit, etc. However, it may point to shared memory, so it is