Fix pre-sorted input validation for constants

SortingProperty is satisfied by ConstantProperty.
This fixes the validation to accept ConstantProperty
on a column as a sorted column.

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

@@ -17,6 +17,7 @@
 import io.trino.Session;
 import io.trino.execution.warnings.WarningCollector;
 import io.trino.metadata.Metadata;
+import io.trino.spi.connector.ConstantProperty;
 import io.trino.spi.connector.SortingProperty;
 import io.trino.spi.type.TypeOperators;
 import io.trino.sql.planner.Symbol;
@@ -97,7 +98,7 @@ public Void visitLimit(LimitNode node, Void context)
             // We do not use LocalProperties#match to fully verify sorting properties
             // as OrderingScheme of input is not tracked by LimitNode
             List<Symbol> sortedColumns = normalizeAndPrune(properties.getLocalProperties()).stream()
-                    .filter(property -> property instanceof SortingProperty)
+                    .filter(property -> property instanceof SortingProperty || property instanceof ConstantProperty)
                     .flatMap(property -> property.getColumns().stream())
                     .collect(toImmutableList());
 

core/trino-main/src/test/java/io/trino/sql/planner/optimizations/TestPartialTopNWithPresortedInput.java

@@ -31,11 +31,11 @@
 import io.trino.sql.parser.SqlParser;
 import io.trino.sql.planner.Plan;
 import io.trino.sql.planner.TypeAnalyzer;
-import io.trino.sql.planner.TypeProvider;
 import io.trino.sql.planner.assertions.BasePlanTest;
 import io.trino.sql.planner.assertions.PlanAssert;
 import io.trino.sql.planner.assertions.PlanMatchPattern;
 import io.trino.sql.planner.sanity.ValidateLimitWithPresortedInput;
+import io.trino.sql.tree.LongLiteral;
 import io.trino.testing.LocalQueryRunner;
 import org.intellij.lang.annotations.Language;
 import org.testng.annotations.Test;
@@ -55,6 +55,7 @@
 import static io.trino.sql.planner.assertions.PlanMatchPattern.sort;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.tableScan;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.topN;
+import static io.trino.sql.planner.assertions.PlanMatchPattern.values;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.window;
 import static io.trino.sql.planner.plan.ExchangeNode.Scope.LOCAL;
 import static io.trino.sql.planner.plan.ExchangeNode.Scope.REMOTE;
@@ -177,14 +178,31 @@ public void testWithSortedWindowFunction()
                                                                         tableScan("table_a", ImmutableMap.of("col_b", "col_b"))))))))));
     }
 
+    @Test
+    public void testWithConstantProperty()
+    {
+        assertPlanWithValidation(
+                "SELECT * FROM (VALUES (1), (1)) AS t (id) WHERE id = 1 ORDER BY 1 LIMIT 1",
+                output(
+                        topN(1, ImmutableList.of(sort("id", ASCENDING, LAST)), FINAL,
+                                exchange(LOCAL, GATHER, ImmutableList.of(),
+                                        limit(1, ImmutableList.of(), true, ImmutableList.of("id"),
+                                                anyTree(
+                                                        values(
+                                                                ImmutableList.of("id"),
+                                                                ImmutableList.of(
+                                                                        ImmutableList.of(new LongLiteral("1")),
+                                                                        ImmutableList.of(new LongLiteral("1"))))))))));
+    }
+
     private void assertPlanWithValidation(@Language("SQL") String sql, PlanMatchPattern pattern)
     {
         LocalQueryRunner queryRunner = getQueryRunner();
         queryRunner.inTransaction(queryRunner.getDefaultSession(), transactionSession -> {
             Plan actualPlan = queryRunner.createPlan(transactionSession, sql, OPTIMIZED_AND_VALIDATED, false, WarningCollector.NOOP);
             PlanAssert.assertPlan(transactionSession, queryRunner.getMetadata(), queryRunner.getStatsCalculator(), actualPlan, pattern);
             Metadata metadata = queryRunner.getMetadata();
-            new ValidateLimitWithPresortedInput().validate(actualPlan.getRoot(), transactionSession, metadata, queryRunner.getTypeOperators(), new TypeAnalyzer(new SqlParser(), metadata), TypeProvider.empty(), WarningCollector.NOOP);
+            new ValidateLimitWithPresortedInput().validate(actualPlan.getRoot(), transactionSession, metadata, queryRunner.getTypeOperators(), new TypeAnalyzer(new SqlParser(), metadata), actualPlan.getTypes(), WarningCollector.NOOP);
             return null;
         });
     }

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

@@ -36,6 +36,7 @@
 import io.trino.sql.planner.assertions.BasePlanTest;
 import io.trino.sql.planner.iterative.rule.test.PlanBuilder;
 import io.trino.sql.planner.plan.PlanNode;
+import io.trino.sql.tree.LongLiteral;
 import io.trino.testing.LocalQueryRunner;
 import io.trino.testing.TestingTransactionHandle;
 import org.testng.annotations.Test;
@@ -44,7 +45,9 @@
 import java.util.function.Function;
 
 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 io.trino.sql.planner.iterative.rule.test.PlanBuilder.expression;
 import static io.trino.testing.TestingSession.testSessionBuilder;
 
 public class TestValidateLimitWithPresortedInput
@@ -136,6 +139,24 @@ public void testValidateSuccessful()
                                         p.symbol(COLUMN_NAME_C, VARCHAR), COLUMN_HANDLE_C))));
     }
 
+    @Test
+    public void testValidateConstantProperty()
+    {
+        validatePlan(
+                p -> p.limit(
+                        10,
+                        ImmutableList.of(),
+                        true,
+                        ImmutableList.of(p.symbol("a", BIGINT)),
+                        p.filter(
+                                expression("a = 1"),
+                                p.values(
+                                        ImmutableList.of(p.symbol("a", BIGINT)),
+                                        ImmutableList.of(
+                                                ImmutableList.of(new LongLiteral("1")),
+                                                ImmutableList.of(new LongLiteral("1")))))));
+    }
+
     @Test(expectedExceptions = VerifyException.class, expectedExceptionsMessageRegExp = "Expected Limit input to be sorted by: \\[col_b], but was \\[col_a]")
     public void testValidateFailed()
     {