diff --git a/core/trino-main/src/test/java/io/trino/operator/BenchmarkPagesIndexOrdering.java b/core/trino-main/src/test/java/io/trino/operator/BenchmarkPagesIndexOrdering.java
index 0c9b49407d68..af9faa0a89b0 100644
--- a/core/trino-main/src/test/java/io/trino/operator/BenchmarkPagesIndexOrdering.java
+++ b/core/trino-main/src/test/java/io/trino/operator/BenchmarkPagesIndexOrdering.java
@@ -13,10 +13,13 @@
  */
 package io.trino.operator;
 
+import io.airlift.slice.Slices;
 import io.trino.RowPagesBuilder;
 import io.trino.spi.Page;
 import io.trino.spi.connector.SortOrder;
+import io.trino.spi.type.BigintType;
 import io.trino.spi.type.Type;
+import io.trino.spi.type.VarcharType;
 import org.junit.jupiter.api.Test;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -39,6 +42,7 @@
 import static io.trino.jmh.Benchmarks.benchmark;
 import static io.trino.spi.connector.SortOrder.ASC_NULLS_FIRST;
 import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.VarcharType.VARCHAR;
 import static java.util.Collections.nCopies;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
@@ -49,8 +53,8 @@
 @OutputTimeUnit(MILLISECONDS)
 @BenchmarkMode(AverageTime)
 @Fork(2)
-@Warmup(iterations = 3)
-@Measurement(iterations = 10, time = 2, timeUnit = SECONDS)
+@Warmup(iterations = 10, time = 1, timeUnit = SECONDS)
+@Measurement(iterations = 10, time = 1, timeUnit = SECONDS)
 public class BenchmarkPagesIndexOrdering
 {
     private static final Random RANDOM = new Random(633969769);
@@ -62,6 +66,8 @@ public static class Context
     {
         @Param({"1", "10"})
         protected int numberOfChannels = 1;
+        @Param({"BIGINT", "VARCHAR"})
+        protected String typeName = "BIGINT";
 
         private List<Type> types;
         private List<Integer> sortChannels;
@@ -71,7 +77,8 @@ public static class Context
         @Setup
         public void setup()
         {
-            types = nCopies(numberOfChannels, BIGINT);
+            Type type = getType(typeName);
+            types = nCopies(numberOfChannels, type);
             sortChannels = IntStream.range(0, numberOfChannels).boxed().collect(toImmutableList());
             sortOrders = nCopies(numberOfChannels, ASC_NULLS_FIRST);
 
@@ -80,7 +87,11 @@ public void setup()
                 for (int row = 0; row < ROWS_PER_PAGE; row++) {
                     Object[] values = new Object[numberOfChannels];
                     for (int channel = 0; channel < numberOfChannels; channel++) {
-                        values[channel] = RANDOM.nextLong();
+                        values[channel] = switch (type) {
+                            case BigintType _ -> RANDOM.nextLong();
+                            case VarcharType _ -> Slices.utf8Slice(String.valueOf(RANDOM.nextLong()));
+                            default -> throw new IllegalArgumentException("Unsupported type: " + typeName);
+                        };
                     }
                     pagesBuilder.row(values);
                 }
@@ -88,6 +99,15 @@ public void setup()
             }
             pages = pagesBuilder.build();
         }
+
+        private static Type getType(String typeName)
+        {
+            return switch (typeName) {
+                case "BIGINT" -> BIGINT;
+                case "VARCHAR" -> VARCHAR;
+                default -> throw new IllegalArgumentException("Unsupported type: " + typeName);
+            };
+        }
     }
 
     @Benchmark
diff --git a/core/trino-spi/src/main/java/io/trino/spi/type/TypeOperators.java b/core/trino-spi/src/main/java/io/trino/spi/type/TypeOperators.java
index 53389d456d59..0a3895f59663 100644
--- a/core/trino-spi/src/main/java/io/trino/spi/type/TypeOperators.java
+++ b/core/trino-spi/src/main/java/io/trino/spi/type/TypeOperators.java
@@ -43,6 +43,7 @@
 import static io.trino.spi.function.InvocationConvention.InvocationArgumentConvention.FLAT;
 import static io.trino.spi.function.InvocationConvention.InvocationArgumentConvention.NEVER_NULL;
 import static io.trino.spi.function.InvocationConvention.InvocationArgumentConvention.NULL_FLAG;
+import static io.trino.spi.function.InvocationConvention.InvocationArgumentConvention.VALUE_BLOCK_POSITION_NOT_NULL;
 import static io.trino.spi.function.InvocationConvention.InvocationReturnConvention.BLOCK_BUILDER;
 import static io.trino.spi.function.InvocationConvention.InvocationReturnConvention.FAIL_ON_NULL;
 import static io.trino.spi.function.InvocationConvention.simpleConvention;
@@ -496,7 +497,12 @@ private OperatorMethodHandle generateOrderingOperator(OperatorConvention operato
             SortOrder sortOrder = operatorConvention.sortOrder().orElseThrow(() -> new IllegalArgumentException("Operator convention does not contain a sort order"));
             OperatorType comparisonType = operatorConvention.operatorType();
             if (operatorConvention.callingConvention().getArgumentConventions().equals(List.of(BLOCK_POSITION, BLOCK_POSITION))) {
-                OperatorConvention comparisonOperator = new OperatorConvention(operatorConvention.type(), comparisonType, Optional.empty(), simpleConvention(FAIL_ON_NULL, BLOCK_POSITION, BLOCK_POSITION));
+                OperatorConvention comparisonOperator = new OperatorConvention(
+                        operatorConvention.type(),
+                        comparisonType,
+                        Optional.empty(),
+                        // null positions are handled separately in adaptBlockPositionComparisonToOrdering and not used in the comparison
+                        simpleConvention(FAIL_ON_NULL, BLOCK_POSITION_NOT_NULL, BLOCK_POSITION_NOT_NULL));
                 MethodHandle comparisonInvoker = adaptOperator(comparisonOperator);
                 return adaptBlockPositionComparisonToOrdering(sortOrder, comparisonInvoker);
             }
@@ -538,7 +544,7 @@ private static int getScore(OperatorMethodHandle operatorMethodHandle)
             if (argument == NULL_FLAG || argument == FLAT) {
                 score += 100;
             }
-            else if (argument == BLOCK_POSITION) {
+            else if (argument == BLOCK_POSITION || argument == VALUE_BLOCK_POSITION_NOT_NULL) {
                 score += 1;
             }
         }