Fix ValidateLimitWithPresortedInput for multiple sorted inputs

core/trino-main/src/main/java/io/trino/sql/planner/sanity/ValidateLimitWithPresortedInput.java

@@ -101,7 +101,7 @@ public Void visitLimit(LimitNode node, Void context)
                     .flatMap(property -> property.getColumns().stream())
                     .collect(toImmutableList());
 
-            if (sortedColumns.size() > node.getPreSortedInputs().size() ||
+            if (sortedColumns.size() < node.getPreSortedInputs().size() ||
                     !sortedColumns.subList(0, node.getPreSortedInputs().size()).equals(node.getPreSortedInputs())) {
                 throw new VerifyException(format("Expected Limit input to be sorted by: %s, but was %s", node.getPreSortedInputs(), sortedColumns));
             }

core/trino-main/src/test/java/io/trino/sql/planner/sanity/TestValidateLimitWithPresortedInput.java

@@ -120,6 +120,20 @@ public void testValidateSuccessful()
                                         p.symbol(COLUMN_NAME_A, VARCHAR), COLUMN_HANDLE_A,
                                         p.symbol(COLUMN_NAME_B, VARCHAR), COLUMN_HANDLE_B,
                                         p.symbol(COLUMN_NAME_C, VARCHAR), COLUMN_HANDLE_C))));
+
+        validatePlan(
+                p -> p.limit(
+                        10,
+                        ImmutableList.of(),
+                        true,
+                        ImmutableList.of(p.symbol(COLUMN_NAME_A, VARCHAR)),
+                        p.tableScan(
+                                MOCK_TABLE_HANDLE,
+                                ImmutableList.of(p.symbol(COLUMN_NAME_A, VARCHAR), p.symbol(COLUMN_NAME_B, VARCHAR), p.symbol(COLUMN_NAME_C, VARCHAR)),
+                                ImmutableMap.of(
+                                        p.symbol(COLUMN_NAME_A, VARCHAR), COLUMN_HANDLE_A,
+                                        p.symbol(COLUMN_NAME_B, VARCHAR), COLUMN_HANDLE_B,
+                                        p.symbol(COLUMN_NAME_C, VARCHAR), COLUMN_HANDLE_C))));
     }
 
     @Test(expectedExceptions = VerifyException.class, expectedExceptionsMessageRegExp = "Expected Limit input to be sorted by: \\[col_b], but was \\[col_a]")

testing/trino-tests/src/test/java/io/trino/tests/tpch/TestTpchConnectorTest.java

@@ -132,6 +132,31 @@ public void testAnalyze()
         assertQueryFails("ANALYZE orders WITH (foo = 'bar')", ".* does not support analyze property 'foo'.*");
     }
 
+    @Test
+    public void testPreSortedInput()
+    {
+        // TPCH connector produces pre-sorted data for orders and lineitem tables
+        assertExplain(
+                "EXPLAIN SELECT * FROM orders ORDER BY orderkey ASC NULLS FIRST LIMIT 10",
+                "\\QLimitPartial[10, input pre-sorted by (orderkey)]");
+        assertExplain(
+                "EXPLAIN SELECT * FROM lineitem ORDER BY orderkey ASC NULLS FIRST LIMIT 10",
+                "\\QLimitPartial[10, input pre-sorted by (orderkey)]");
+        assertExplain(
+                "EXPLAIN SELECT * FROM lineitem ORDER BY orderkey ASC NULLS FIRST, linenumber ASC NULLS FIRST LIMIT 10",
+                "\\QLimitPartial[10, input pre-sorted by (orderkey, linenumber)]");
+        assertExplain(
+                "EXPLAIN SELECT * FROM lineitem ORDER BY orderkey ASC NULLS FIRST, linenumber LIMIT 10",
+                "\\QTopNPartial[10 by (orderkey ASC NULLS FIRST, linenumber ASC NULLS LAST)]");
+        assertExplain(
+                "EXPLAIN SELECT * FROM lineitem ORDER BY orderkey ASC LIMIT 10",
+                "\\QTopNPartial[10 by (orderkey ASC NULLS LAST)]");
+
+        assertQuery(
+                "SELECT * FROM lineitem WHERE orderkey IS NOT NULL ORDER BY orderkey ASC NULLS FIRST LIMIT 10",
+                "SELECT * FROM lineitem ORDER BY orderkey ASC LIMIT 10");
+    }
+
     @Test
     @Override
     public void testShowTables()