Implement variable-precision time type

* Removes support for legacy semantics for time type.
* Changes the internal representation of time type to picoseconds since midnight. This is a backward
  incompatible change but since time type is broken in many ways today to the point of being almost
  unusable, this is not such a big problem.

presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/StandardColumnMappings.java

@@ -35,7 +35,6 @@
 import java.sql.Timestamp;
 import java.sql.Types;
 import java.time.Instant;
-import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.ZoneId;
@@ -59,6 +58,12 @@
 import static io.prestosql.spi.type.SmallintType.SMALLINT;
 import static io.prestosql.spi.type.TimeType.TIME;
 import static io.prestosql.spi.type.TimestampType.TIMESTAMP;
+import static io.prestosql.spi.type.Timestamps.NANOSECONDS_PER_DAY;
+import static io.prestosql.spi.type.Timestamps.NANOSECONDS_PER_MILLISECOND;
+import static io.prestosql.spi.type.Timestamps.PICOSECONDS_PER_DAY;
+import static io.prestosql.spi.type.Timestamps.PICOSECONDS_PER_MILLISECOND;
+import static io.prestosql.spi.type.Timestamps.PICOSECONDS_PER_NANOSECOND;
+import static io.prestosql.spi.type.Timestamps.roundDiv;
 import static io.prestosql.spi.type.TinyintType.TINYINT;
 import static io.prestosql.spi.type.VarbinaryType.VARBINARY;
 import static io.prestosql.spi.type.VarcharType.createUnboundedVarcharType;
@@ -271,26 +276,15 @@ public static LongWriteFunction dateWriteFunction()
      * {@link #timeColumnMapping} instead.
      */
     @Deprecated
-    public static ColumnMapping timeColumnMappingUsingSqlTime(ConnectorSession session)
+    public static ColumnMapping timeColumnMappingUsingSqlTime()
     {
-        if (session.isLegacyTimestamp()) {
-            ZoneId sessionZone = ZoneId.of(session.getTimeZoneKey().getId());
-            return ColumnMapping.longMapping(
-                    TIME,
-                    (resultSet, columnIndex) -> {
-                        Time time = resultSet.getTime(columnIndex);
-                        return toPrestoLegacyTimestamp(toLocalTime(time).atDate(LocalDate.ofEpochDay(0)), sessionZone);
-                    },
-                    timeWriteFunctionUsingSqlTime(session));
-        }
-
         return ColumnMapping.longMapping(
                 TIME,
                 (resultSet, columnIndex) -> {
                     Time time = resultSet.getTime(columnIndex);
-                    return NANOSECONDS.toMillis(toLocalTime(time).toNanoOfDay());
+                    return (toLocalTime(time).toNanoOfDay() * PICOSECONDS_PER_NANOSECOND) % PICOSECONDS_PER_DAY;
                 },
-                timeWriteFunctionUsingSqlTime(session));
+                timeWriteFunctionUsingSqlTime());
     }
 
     private static LocalTime toLocalTime(Time sqlTime)
@@ -307,13 +301,9 @@ private static LocalTime toLocalTime(Time sqlTime)
      * {@link #timeWriteFunction} instead.
      */
     @Deprecated
-    public static LongWriteFunction timeWriteFunctionUsingSqlTime(ConnectorSession session)
+    public static LongWriteFunction timeWriteFunctionUsingSqlTime()
     {
-        if (session.isLegacyTimestamp()) {
-            ZoneId sessionZone = ZoneId.of(session.getTimeZoneKey().getId());
-            return (statement, index, value) -> statement.setTime(index, toSqlTime(fromPrestoLegacyTimestamp(value, sessionZone).toLocalTime()));
-        }
-        return (statement, index, value) -> statement.setTime(index, toSqlTime(fromPrestoTimestamp(value).toLocalTime()));
+        return (statement, index, value) -> statement.setTime(index, toSqlTime(fromPrestoTime(value)));
     }
 
     private static Time toSqlTime(LocalTime localTime)
@@ -322,35 +312,36 @@ private static Time toSqlTime(LocalTime localTime)
         return new Time(Time.valueOf(localTime).getTime() + NANOSECONDS.toMillis(localTime.getNano()));
     }
 
-    public static ColumnMapping timeColumnMapping(ConnectorSession session)
+    public static ColumnMapping timeColumnMapping()
     {
-        if (session.isLegacyTimestamp()) {
-            ZoneId sessionZone = ZoneId.of(session.getTimeZoneKey().getId());
-            return ColumnMapping.longMapping(
-                    TIME,
-                    (resultSet, columnIndex) -> {
-                        LocalTime time = resultSet.getObject(columnIndex, LocalTime.class);
-                        return toPrestoLegacyTimestamp(time.atDate(LocalDate.ofEpochDay(0)), sessionZone);
-                    },
-                    timeWriteFunction(session));
-        }
+        return ColumnMapping.longMapping(
+                TIME,
+                (resultSet, columnIndex) -> {
+                    LocalTime time = resultSet.getObject(columnIndex, LocalTime.class);
+                    long nanos = time.toNanoOfDay();
+                    return (roundDiv(nanos, NANOSECONDS_PER_MILLISECOND) * PICOSECONDS_PER_MILLISECOND) % PICOSECONDS_PER_DAY;
+                },
+                timeWriteFunction());
+    }
 
+    /**
+     * Truncates the time value on read to millisecond precision.
+     */
+    public static ColumnMapping timeColumnMappingWithTruncation()
+    {
         return ColumnMapping.longMapping(
                 TIME,
                 (resultSet, columnIndex) -> {
                     LocalTime time = resultSet.getObject(columnIndex, LocalTime.class);
-                    return NANOSECONDS.toMillis(time.toNanoOfDay());
+                    return ((time.toNanoOfDay() / NANOSECONDS_PER_MILLISECOND) * PICOSECONDS_PER_MILLISECOND) % PICOSECONDS_PER_DAY;
                 },
-                timeWriteFunction(session));
+                timeWriteFunction(),
+                DISABLE_PUSHDOWN);
     }
 
-    public static LongWriteFunction timeWriteFunction(ConnectorSession session)
+    public static LongWriteFunction timeWriteFunction()
     {
-        if (session.isLegacyTimestamp()) {
-            ZoneId sessionZone = ZoneId.of(session.getTimeZoneKey().getId());
-            return (statement, index, value) -> statement.setObject(index, fromPrestoLegacyTimestamp(value, sessionZone).toLocalTime());
-        }
-        return (statement, index, value) -> statement.setObject(index, fromPrestoTimestamp(value).toLocalTime());
+        return (statement, index, value) -> statement.setObject(index, fromPrestoTime(value));
     }
 
     /**
@@ -454,6 +445,12 @@ public static LocalDateTime fromPrestoTimestamp(long value)
         return Instant.ofEpochMilli(value).atZone(UTC).toLocalDateTime();
     }
 
+    public static LocalTime fromPrestoTime(long value)
+    {
+        // value can round up to NANOSECONDS_PER_DAY, so we need to do % to keep it in the desired range
+        return LocalTime.ofNanoOfDay(roundDiv(value, PICOSECONDS_PER_NANOSECOND) % NANOSECONDS_PER_DAY);
+    }
+
     public static Optional<ColumnMapping> jdbcTypeToPrestoType(ConnectorSession session, JdbcTypeHandle type)
     {
         int columnSize = type.getColumnSize();
@@ -515,7 +512,7 @@ public static Optional<ColumnMapping> jdbcTypeToPrestoType(ConnectorSession sess
 
             case Types.TIME:
                 // TODO default to `timeColumnMapping`
-                return Optional.of(timeColumnMappingUsingSqlTime(session));
+                return Optional.of(timeColumnMappingUsingSqlTime());
 
             case Types.TIMESTAMP:
                 // TODO default to `timestampColumnMapping`

presto-base-jdbc/src/test/java/io/prestosql/plugin/jdbc/TestJdbcQueryBuilder.java

@@ -19,7 +19,6 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Multiset;
 import io.prestosql.spi.connector.ColumnHandle;
-import io.prestosql.spi.connector.ConnectorSession;
 import io.prestosql.spi.predicate.Domain;
 import io.prestosql.spi.predicate.Range;
 import io.prestosql.spi.predicate.SortedRangeSet;
@@ -339,9 +338,9 @@ public void testBuildSqlWithDateTime()
                         false),
                 columns.get(5), Domain.create(SortedRangeSet.copyOf(TIME,
                         ImmutableList.of(
-                                Range.range(TIME, toTimeRepresentation(SESSION, 6, 12, 23), false, toTimeRepresentation(SESSION, 8, 23, 37), true),
-                                Range.equal(TIME, toTimeRepresentation(SESSION, 2, 3, 4)),
-                                Range.equal(TIME, toTimeRepresentation(SESSION, 20, 23, 37)))),
+                                Range.range(TIME, toTimeRepresentation(6, 12, 23), false, toTimeRepresentation(8, 23, 37), true),
+                                Range.equal(TIME, toTimeRepresentation(2, 3, 4)),
+                                Range.equal(TIME, toTimeRepresentation(20, 23, 37)))),
                         false)));
 
         Connection connection = database.getConnection();
@@ -540,13 +539,10 @@ private static Time toTime(int hour, int minute, int second)
         return Time.valueOf(LocalTime.of(hour, minute, second));
     }
 
-    private static long toTimeRepresentation(ConnectorSession session, int hour, int minute, int second)
+    private static long toTimeRepresentation(int hour, int minute, int second)
     {
-        SqlTime time = sqlTimeOf(hour, minute, second, 0, session);
-        if (session.isLegacyTimestamp()) {
-            return time.getMillisUtc();
-        }
-        return time.getMillis();
+        SqlTime time = sqlTimeOf(hour, minute, second, 0);
+        return time.getPicos();
     }
 
     private static Multiset<List<Object>> read(ResultSet resultSet)

presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveIntegrationSmokeTest.java

@@ -1594,8 +1594,8 @@ private void testCreatePartitionedTableAs(Session session, HiveStorageFormat sto
     @Test
     public void testCreateTableWithUnsupportedType()
     {
-        assertQueryFails("CREATE TABLE test_create_table_with_unsupported_type(x time)", "Unsupported Hive type: time");
-        assertQueryFails("CREATE TABLE test_create_table_with_unsupported_type AS SELECT TIME '00:00:00' x", "Unsupported Hive type: time");
+        assertQueryFails("CREATE TABLE test_create_table_with_unsupported_type(x time)", "\\QUnsupported Hive type: time(3)\\E");
+        assertQueryFails("CREATE TABLE test_create_table_with_unsupported_type AS SELECT TIME '00:00:00' x", "\\QUnsupported Hive type: time(0)\\E");
     }
 
     @Test

presto-iceberg/src/main/java/io/prestosql/plugin/iceberg/IcebergPageSource.java

@@ -52,6 +52,7 @@
 import static io.prestosql.spi.type.TimeType.TIME;
 import static io.prestosql.spi.type.TimestampType.TIMESTAMP;
 import static io.prestosql.spi.type.TimestampWithTimeZoneType.TIMESTAMP_WITH_TIME_ZONE;
+import static io.prestosql.spi.type.Timestamps.PICOSECONDS_PER_MILLISECOND;
 import static io.prestosql.spi.type.Varchars.isVarcharType;
 import static java.lang.Double.parseDouble;
 import static java.lang.Float.floatToRawIntBits;
@@ -213,7 +214,7 @@ private static Object deserializePartitionValue(Type type, String valueString, S
                 return parseLong(valueString);
             }
             if (type.equals(TIME)) {
-                return parseLong(valueString);
+                return parseLong(valueString) * PICOSECONDS_PER_MILLISECOND;
             }
             if (type.equals(TIMESTAMP)) {
                 return MICROSECONDS.toMillis(parseLong(valueString));

presto-iceberg/src/main/java/io/prestosql/plugin/iceberg/PartitionTransforms.java

@@ -46,6 +46,7 @@
 import static io.prestosql.spi.type.TimeType.TIME;
 import static io.prestosql.spi.type.TimestampType.TIMESTAMP;
 import static io.prestosql.spi.type.TimestampWithTimeZoneType.TIMESTAMP_WITH_TIME_ZONE;
+import static io.prestosql.spi.type.Timestamps.PICOSECONDS_PER_MICROSECOND;
 import static io.prestosql.spi.type.VarbinaryType.VARBINARY;
 import static io.prestosql.spi.type.VarcharType.VARCHAR;
 import static io.prestosql.spi.type.Varchars.isVarcharType;
@@ -323,8 +324,8 @@ private static Block bucketDate(Block block, int count)
     private static Block bucketTime(Block block, int count)
     {
         return bucketBlock(block, count, position -> {
-            long value = TIME.getLong(block, position);
-            return bucketHash(MILLISECONDS.toMicros(value));
+            long picos = TIME.getLong(block, position);
+            return bucketHash(picos / PICOSECONDS_PER_MICROSECOND);
         });
     }
 

presto-jdbc/src/main/java/io/prestosql/jdbc/AbstractPrestoResultSet.java

@@ -52,6 +52,7 @@
 import java.sql.Types;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
+import java.time.ZonedDateTime;
 import java.util.Calendar;
 import java.util.GregorianCalendar;
 import java.util.HashMap;
@@ -83,6 +84,8 @@ abstract class AbstractPrestoResultSet
             "(?: (?<hour>\\d{1,2}):(?<minute>\\d{1,2})(?::(?<second>\\d{1,2})(?:\\.(?<fraction>\\d+))?)?)?" +
             "\\s*(?<timezone>.+)?");
 
+    public static final Pattern TIME_PATTERN = Pattern.compile("(?<hour>\\d{1,2}):(?<minute>\\d{1,2}):(?<second>\\d{1,2})(?:\\.(?<fraction>\\d+))?");
+
     private static final long[] POWERS_OF_TEN = {
             1L,
             10L,
@@ -296,9 +299,9 @@ private Time getTime(int columnIndex, DateTimeZone localTimeZone)
         }
 
         ColumnInfo columnInfo = columnInfo(columnIndex);
-        if (columnInfo.getColumnTypeName().equalsIgnoreCase("time")) {
+        if (columnInfo.getColumnTypeSignature().getRawType().equalsIgnoreCase("time")) {
             try {
-                return new Time(TIME_FORMATTER.withZone(localTimeZone).parseMillis(String.valueOf(value)));
+                return parseTime(String.valueOf(value), ZoneId.of(localTimeZone.getID()));
             }
             catch (IllegalArgumentException e) {
                 throw new SQLException("Invalid time from server: " + value, e);
@@ -1827,6 +1830,36 @@ private static Timestamp parseTimestamp(String value, Function<String, ZoneId> t
         return timestamp;
     }
 
+    private static Time parseTime(String value, ZoneId localTimeZone)
+    {
+        Matcher matcher = TIME_PATTERN.matcher(value);
+        if (!matcher.matches()) {
+            throw new IllegalArgumentException("Invalid time: " + value);
+        }
+
+        int hour = Integer.parseInt(matcher.group("hour"));
+        int minute = Integer.parseInt(matcher.group("minute"));
+        int second = matcher.group("second") == null ? 0 : Integer.parseInt(matcher.group("second"));
+
+        if (hour > 23 || minute > 59 || second > 59) {
+            throw new IllegalArgumentException("Invalid time: " + value);
+        }
+
+        int precision = 0;
+        String fraction = matcher.group("fraction");
+        long fractionValue = 0;
+        if (fraction != null) {
+            precision = fraction.length();
+            fractionValue = Long.parseLong(fraction);
+        }
+
+        long epochMilli = ZonedDateTime.of(1970, 1, 1, hour, minute, second, (int) rescale(fractionValue, precision, 9), localTimeZone)
+                .toInstant()
+                .toEpochMilli();
+
+        return new Time(epochMilli);
+    }
+
     private static long rescale(long value, int fromPrecision, int toPrecision)
     {
         if (value < 0) {

presto-jdbc/src/test/java/io/prestosql/jdbc/TestJdbcResultSet.java

@@ -176,7 +176,7 @@ public void testDate()
     public void testTime()
             throws Exception
     {
-        checkRepresentation("TIME '09:39:05'", Types.TIME, (rs, column) -> {
+        checkRepresentation("TIME '09:39:05.000'", Types.TIME, (rs, column) -> {
             assertEquals(rs.getObject(column), Time.valueOf(LocalTime.of(9, 39, 5)));
             assertEquals(rs.getObject(column, Time.class), Time.valueOf(LocalTime.of(9, 39, 5)));
             assertThrows(() -> rs.getDate(column));

presto-jdbc/src/test/java/io/prestosql/jdbc/TestPrestoDatabaseMetaData.java

@@ -762,7 +762,7 @@ public void testGetColumns()
                 assertColumnSpec(rs, Types.VARCHAR, (long) Integer.MAX_VALUE, null, null, (long) Integer.MAX_VALUE, createUnboundedVarcharType());
                 assertColumnSpec(rs, Types.CHAR, 345L, null, null, 345L, createCharType(345));
                 assertColumnSpec(rs, Types.VARBINARY, (long) Integer.MAX_VALUE, null, null, (long) Integer.MAX_VALUE, VarbinaryType.VARBINARY);
-                assertColumnSpec(rs, Types.TIME, 8L, null, null, null, TimeType.TIME);
+                assertColumnSpec(rs, Types.TIME, 12L, null, null, null, TimeType.TIME);
                 assertColumnSpec(rs, Types.TIME_WITH_TIMEZONE, 14L, null, null, null, TimeWithTimeZoneType.TIME_WITH_TIME_ZONE);
                 assertColumnSpec(rs, Types.TIMESTAMP, 25L, null, null, null, TimestampType.TIMESTAMP);
                 assertColumnSpec(rs, Types.TIMESTAMP, 31L, null, null, null, createTimestampType(9));

presto-main/src/main/java/io/prestosql/connector/system/jdbc/ColumnJdbcTable.java

@@ -40,6 +40,7 @@
 import io.prestosql.spi.type.ArrayType;
 import io.prestosql.spi.type.CharType;
 import io.prestosql.spi.type.DecimalType;
+import io.prestosql.spi.type.TimeType;
 import io.prestosql.spi.type.TimestampType;
 import io.prestosql.spi.type.TimestampWithTimeZoneType;
 import io.prestosql.spi.type.Type;
@@ -416,8 +417,12 @@ static Integer columnSize(Type type)
         if (type.equals(VARBINARY)) {
             return Integer.MAX_VALUE;
         }
-        if (type.equals(TIME)) {
-            return 8; // 00:00:00
+        if (type instanceof TimeType) {
+            // 8 characters for "HH:MM:SS"
+            // min(p, 1) for the fractional second period (i.e., no period if p == 0)
+            // p for the fractional digits
+            int precision = ((TimeType) type).getPrecision();
+            return 8 + min(precision, 1) + precision;
         }
         if (type.equals(TIME_WITH_TIME_ZONE)) {
             return 8 + 6; // 00:00:00+00:00

presto-main/src/main/java/io/prestosql/metadata/FunctionRegistry.java

@@ -143,6 +143,11 @@
 import io.prestosql.operator.scalar.VarbinaryFunctions;
 import io.prestosql.operator.scalar.WilsonInterval;
 import io.prestosql.operator.scalar.WordStemFunction;
+import io.prestosql.operator.scalar.time.LocalTimeFunction;
+import io.prestosql.operator.scalar.time.TimeFunctions;
+import io.prestosql.operator.scalar.time.TimeOperators;
+import io.prestosql.operator.scalar.time.TimeToTimestampCast;
+import io.prestosql.operator.scalar.time.TimeToTimestampWithTimeZoneCast;
 import io.prestosql.operator.scalar.timestamp.DateAdd;
 import io.prestosql.operator.scalar.timestamp.DateDiff;
 import io.prestosql.operator.scalar.timestamp.DateFormat;
@@ -165,7 +170,6 @@
 import io.prestosql.operator.scalar.timestamp.LocalTimestamp;
 import io.prestosql.operator.scalar.timestamp.SequenceIntervalDayToSecond;
 import io.prestosql.operator.scalar.timestamp.SequenceIntervalYearToMonth;
-import io.prestosql.operator.scalar.timestamp.TimeToTimestampCast;
 import io.prestosql.operator.scalar.timestamp.TimeWithTimezoneToTimestampCast;
 import io.prestosql.operator.scalar.timestamp.TimestampDistinctFromOperator;
 import io.prestosql.operator.scalar.timestamp.TimestampOperators;
@@ -184,7 +188,6 @@
 import io.prestosql.operator.scalar.timestamptz.AtTimeZoneWithOffset;
 import io.prestosql.operator.scalar.timestamptz.CurrentTimestamp;
 import io.prestosql.operator.scalar.timestamptz.DateToTimestampWithTimeZoneCast;
-import io.prestosql.operator.scalar.timestamptz.TimeToTimestampWithTimeZoneCast;
 import io.prestosql.operator.scalar.timestamptz.TimeWithTimeZoneToTimestampWithTimeZoneCast;
 import io.prestosql.operator.scalar.timestamptz.TimestampWithTimeZoneDistinctFromOperator;
 import io.prestosql.operator.scalar.timestamptz.TimestampWithTimeZoneOperators;
@@ -231,7 +234,6 @@
 import io.prestosql.type.QuantileDigestOperators;
 import io.prestosql.type.RealOperators;
 import io.prestosql.type.SmallintOperators;
-import io.prestosql.type.TimeOperators;
 import io.prestosql.type.TimeWithTimeZoneOperators;
 import io.prestosql.type.TinyintOperators;
 import io.prestosql.type.UnknownOperators;
@@ -502,8 +504,6 @@ public FunctionRegistry(Supplier<BlockEncodingSerde> blockEncodingSerdeSupplier,
                 .scalar(VarbinaryOperators.VarbinaryDistinctFromOperator.class)
                 .scalars(DateOperators.class)
                 .scalar(DateOperators.DateDistinctFromOperator.class)
-                .scalars(TimeOperators.class)
-                .scalar(TimeOperators.TimeDistinctFromOperator.class)
                 .scalars(IntervalDayTimeOperators.class)
                 .scalar(IntervalDayTimeOperators.IntervalDayTimeDistinctFromOperator.class)
                 .scalars(IntervalYearMonthOperators.class)
@@ -750,6 +750,12 @@ public FunctionRegistry(Supplier<BlockEncodingSerde> blockEncodingSerdeSupplier,
                 .scalar(TimeWithTimeZoneToTimestampWithTimeZoneCast.class)
                 .scalar(VarcharToTimestampWithTimeZoneCast.class);
 
+        // time without time zone functions and operators
+        builder.scalar(LocalTimeFunction.class)
+                .scalars(TimeOperators.class)
+                .scalars(TimeFunctions.class)
+                .scalar(TimeOperators.TimeDistinctFromOperator.class);
+
         switch (featuresConfig.getRegexLibrary()) {
             case JONI:
                 builder.scalars(JoniRegexpFunctions.class);