Support generating sequences in the Faker connector

docs/src/main/sphinx/connector/faker.md

@@ -111,6 +111,9 @@ The following table details all supported column properties.
 * - `allowed_values`
   - List of allowed values. Cannot be set together with the `min`, or `max`
     properties.
+* - `step`
+  - If set, generate sequential values with this step. For date and time columns
+    set this to a duration. Cannot be set for character-based type columns.
 :::
 
 ### Character types
@@ -175,7 +178,7 @@ Faker supports the following non-character types:
 - `UUID`
 
 You can not use generator expressions for non-character-based columns. To limit
-their data range, set the `min` and/or `max` column properties - see
+their data range, set the `min` and `max` column properties - see
 [](faker-usage).
 
 ### Unsupported types

plugin/trino-faker/pom.xml

@@ -40,6 +40,11 @@
             <artifactId>configuration</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>units</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>io.trino</groupId>
             <artifactId>trino-main</artifactId>
@@ -119,12 +124,6 @@
             <scope>runtime</scope>
         </dependency>
 
-        <dependency>
-            <groupId>io.airlift</groupId>
-            <artifactId>units</artifactId>
-            <scope>runtime</scope>
-        </dependency>
-
         <dependency>
             <groupId>io.trino</groupId>
             <artifactId>trino-client</artifactId>

plugin/trino-faker/src/main/java/io/trino/plugin/faker/ColumnInfo.java

@@ -27,6 +27,7 @@ public record ColumnInfo(FakerColumnHandle handle, ColumnMetadata metadata)
     public static final String MIN_PROPERTY = "min";
     public static final String MAX_PROPERTY = "max";
     public static final String ALLOWED_VALUES_PROPERTY = "allowed_values";
+    public static final String STEP_PROPERTY = "step";
 
     public ColumnInfo
     {

plugin/trino-faker/src/main/java/io/trino/plugin/faker/FakerColumnHandle.java

@@ -15,27 +15,37 @@
 package io.trino.plugin.faker;
 
 import com.google.common.collect.ImmutableList;
+import io.airlift.units.Duration;
 import io.trino.spi.TrinoException;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
 import io.trino.spi.predicate.Domain;
 import io.trino.spi.predicate.Range;
 import io.trino.spi.predicate.ValueSet;
 import io.trino.spi.type.CharType;
+import io.trino.spi.type.TimeType;
+import io.trino.spi.type.TimeWithTimeZoneType;
+import io.trino.spi.type.TimestampType;
+import io.trino.spi.type.TimestampWithTimeZoneType;
 import io.trino.spi.type.Type;
 import io.trino.spi.type.VarbinaryType;
 import io.trino.spi.type.VarcharType;
 
 import java.util.Collection;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
+import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static io.trino.plugin.faker.ColumnInfo.ALLOWED_VALUES_PROPERTY;
 import static io.trino.plugin.faker.ColumnInfo.GENERATOR_PROPERTY;
 import static io.trino.plugin.faker.ColumnInfo.MAX_PROPERTY;
 import static io.trino.plugin.faker.ColumnInfo.MIN_PROPERTY;
 import static io.trino.plugin.faker.ColumnInfo.NULL_PROBABILITY_PROPERTY;
+import static io.trino.plugin.faker.ColumnInfo.STEP_PROPERTY;
 import static io.trino.spi.StandardErrorCode.INVALID_COLUMN_PROPERTY;
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.DateType.DATE;
 import static java.util.Objects.requireNonNull;
 
 public record FakerColumnHandle(
@@ -44,13 +54,17 @@ public record FakerColumnHandle(
         Type type,
         double nullProbability,
         String generator,
-        Domain domain)
+        Domain domain,
+        ValueSet step)
         implements ColumnHandle
 {
     public FakerColumnHandle
     {
         requireNonNull(name, "name is null");
         requireNonNull(type, "type is null");
+        requireNonNull(domain, "domain is null");
+        requireNonNull(step, "step is null");
+        checkState(step.isNone() || step.isSingleValue(), "step must be a single value");
     }
 
     public static FakerColumnHandle of(int columnId, ColumnMetadata column, double defaultNullProbability)
@@ -63,22 +77,8 @@ public static FakerColumnHandle of(int columnId, ColumnMetadata column, double d
         if (generator != null && !isCharacterColumn(column)) {
             throw new TrinoException(INVALID_COLUMN_PROPERTY, "The `%s` property can only be set for CHAR, VARCHAR or VARBINARY columns".formatted(GENERATOR_PROPERTY));
         }
-        // only parse min, max, and options to validate literals - FakerColumnHandle needs to be serializable,
-        // and some internal Trino types are not (Int128, LongTimestamp, LongTimestampWithTimeZone), so they cannot be stored in the handle as native types
-        String min = (String) column.getProperties().get(MIN_PROPERTY);
-        try {
-            Literal.parse(min, column.getType());
-        }
-        catch (IllegalArgumentException e) {
-            throw new TrinoException(INVALID_COLUMN_PROPERTY, "The `%s` property must be a valid %s literal".formatted(MIN_PROPERTY, column.getType().getDisplayName()), e);
-        }
-        String max = (String) column.getProperties().get(MAX_PROPERTY);
-        try {
-            Literal.parse(max, column.getType());
-        }
-        catch (IllegalArgumentException e) {
-            throw new TrinoException(INVALID_COLUMN_PROPERTY, "The `%s` property must be a valid %s literal".formatted(MAX_PROPERTY, column.getType().getDisplayName()), e);
-        }
+        Object min = propertyValue(column, MIN_PROPERTY);
+        Object max = propertyValue(column, MAX_PROPERTY);
         Domain domain = Domain.all(column.getType());
         if (min != null || max != null) {
             if (isCharacterColumn(column)) {
@@ -108,27 +108,64 @@ public static FakerColumnHandle of(int columnId, ColumnMetadata column, double d
                 column.getType(),
                 nullProbability,
                 generator,
-                domain);
+                domain,
+                stepValue(column));
     }
 
     private static boolean isCharacterColumn(ColumnMetadata column)
     {
         return column.getType() instanceof CharType || column.getType() instanceof VarcharType || column.getType() instanceof VarbinaryType;
     }
 
-    private static Range range(Type type, String min, String max)
+    private static Object propertyValue(ColumnMetadata column, String property)
+    {
+        try {
+            return Literal.parse((String) column.getProperties().get(property), column.getType());
+        }
+        catch (IllegalArgumentException e) {
+            throw new TrinoException(INVALID_COLUMN_PROPERTY, "The `%s` property must be a valid %s literal".formatted(property, column.getType().getDisplayName()), e);
+        }
+    }
+
+    private static ValueSet stepValue(ColumnMetadata column)
+    {
+        Type type = column.getType();
+        String rawStep = (String) column.getProperties().get(STEP_PROPERTY);
+        if (rawStep == null) {
+            return ValueSet.none(type);
+        }
+        if (isCharacterColumn(column)) {
+            throw new TrinoException(INVALID_COLUMN_PROPERTY, "The `%s` property cannot be set for CHAR, VARCHAR or VARBINARY columns".formatted(STEP_PROPERTY));
+        }
+        if (DATE.equals(column.getType()) || type instanceof TimestampType || type instanceof TimestampWithTimeZoneType || type instanceof TimeType || type instanceof TimeWithTimeZoneType) {
+            try {
+                return ValueSet.of(BIGINT, Duration.valueOf(rawStep).roundTo(TimeUnit.NANOSECONDS));
+            }
+            catch (IllegalArgumentException e) {
+                throw new TrinoException(INVALID_COLUMN_PROPERTY, "The `%s` property for a %s column must be a valid duration literal".formatted(STEP_PROPERTY, column.getType().getDisplayName()), e);
+            }
+        }
+        try {
+            return ValueSet.of(type, Literal.parse(rawStep, type));
+        }
+        catch (IllegalArgumentException e) {
+            throw new TrinoException(INVALID_COLUMN_PROPERTY, "The `%s` property for a %s column must be a valid %s literal".formatted(STEP_PROPERTY, column.getType().getDisplayName(), type.getDisplayName()), e);
+        }
+    }
+
+    private static Range range(Type type, Object min, Object max)
     {
         requireNonNull(type, "type is null");
         if (min == null && max == null) {
             return Range.all(type);
         }
         if (max == null) {
-            return Range.greaterThanOrEqual(type, Literal.parse(min, type));
+            return Range.greaterThanOrEqual(type, min);
         }
         if (min == null) {
-            return Range.lessThanOrEqual(type, Literal.parse(max, type));
+            return Range.lessThanOrEqual(type, max);
         }
-        return Range.range(type, Literal.parse(min, type), true, Literal.parse(max, type), true);
+        return Range.range(type, min, true, max, true);
     }
 
     private static List<String> strings(Collection<?> values)

plugin/trino-faker/src/main/java/io/trino/plugin/faker/FakerConnector.java

@@ -39,6 +39,7 @@
 import static io.trino.plugin.faker.ColumnInfo.ALLOWED_VALUES_PROPERTY;
 import static io.trino.plugin.faker.ColumnInfo.MAX_PROPERTY;
 import static io.trino.plugin.faker.ColumnInfo.MIN_PROPERTY;
+import static io.trino.plugin.faker.ColumnInfo.STEP_PROPERTY;
 import static io.trino.spi.StandardErrorCode.INVALID_COLUMN_PROPERTY;
 import static io.trino.spi.StandardErrorCode.INVALID_SCHEMA_PROPERTY;
 import static io.trino.spi.StandardErrorCode.INVALID_TABLE_PROPERTY;
@@ -188,7 +189,12 @@ public List<PropertyMetadata<?>> getColumnProperties()
                         value -> ((List<?>) value).stream()
                                 .map(String.class::cast)
                                 .collect(toImmutableList()),
-                        value -> value));
+                        value -> value),
+                stringProperty(
+                        STEP_PROPERTY,
+                        "If set, generate sequential values with this step. For date and time columns set this to a duration",
+                        null,
+                        false));
     }
 
     private static void checkProperty(boolean expression, ErrorCodeSupplier errorCode, String errorMessage)

plugin/trino-faker/src/main/java/io/trino/plugin/faker/FakerMetadata.java

@@ -43,6 +43,8 @@
 import io.trino.spi.function.FunctionMetadata;
 import io.trino.spi.function.SchemaFunctionName;
 import io.trino.spi.predicate.Domain;
+import io.trino.spi.predicate.Range;
+import io.trino.spi.predicate.ValueSet;
 import io.trino.spi.security.TrinoPrincipal;
 import io.trino.spi.statistics.ComputedStatistics;
 import io.trino.spi.type.BigintType;
@@ -336,7 +338,8 @@ public synchronized FakerOutputTableHandle beginCreateTable(ConnectorSession ses
                         BigintType.BIGINT,
                         0,
                         "",
-                        Domain.all(BigintType.BIGINT)),
+                        Domain.create(ValueSet.ofRanges(Range.greaterThanOrEqual(BigintType.BIGINT, 0L)), false),
+                        ValueSet.of(BigintType.BIGINT, 1L)),
                 ColumnMetadata.builder()
                         .setName(ROW_ID_COLUMN_NAME)
                         .setType(BigintType.BIGINT)