This is an automated email from the ASF dual-hosted git repository. twalthr pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push: new 8f3f9f5 [FLINK-12253][table-common] Add a TIMESTAMP type 8f3f9f5 is described below commit 8f3f9f59648fb777c78485ee85d3c6156ecc3c70 Author: Timo Walther <twal...@apache.org> AuthorDate: Thu May 2 09:48:21 2019 +0200 [FLINK-12253][table-common] Add a TIMESTAMP type --- .../table/types/logical/LogicalTypeVisitor.java | 2 + ...{LogicalTypeVisitor.java => TimestampKind.java} | 42 +----- .../flink/table/types/logical/TimestampType.java | 163 +++++++++++++++++++++ .../apache/flink/table/types/LogicalTypesTest.java | 14 ++ 4 files changed, 187 insertions(+), 34 deletions(-) diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java index 07ae9d0..a65168e 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java @@ -57,5 +57,7 @@ public interface LogicalTypeVisitor<R> { R visit(TimeType timeType); + R visit(TimestampType timestampType); + R visit(LogicalType other); } diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/TimestampKind.java similarity index 53% copy from flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java copy to flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/TimestampKind.java index 07ae9d0..95bb09a 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/TimestampKind.java @@ -18,44 +18,18 @@ package org.apache.flink.table.types.logical; -import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.annotation.Internal; /** - * The visitor definition of {@link LogicalType}. The visitor transforms a logical type into - * instances of {@code R}. - * - * @param <R> result type + * Internal timestamp kind for attaching time attribute metadata to timestamps with or without a + * time zone. */ -@PublicEvolving -public interface LogicalTypeVisitor<R> { - - R visit(CharType charType); - - R visit(VarCharType varCharType); - - R visit(BooleanType booleanType); - - R visit(BinaryType binaryType); - - R visit(VarBinaryType varBinaryType); - - R visit(DecimalType decimalType); - - R visit(TinyIntType tinyIntType); - - R visit(SmallIntType smallIntType); - - R visit(IntType intType); - - R visit(BigIntType bigIntType); - - R visit(FloatType floatType); - - R visit(DoubleType doubleType); +@Internal +public enum TimestampKind { - R visit(DateType dateType); + REGULAR, - R visit(TimeType timeType); + ROWTIME, - R visit(LogicalType other); + PROCTIME } diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/TimestampType.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/TimestampType.java new file mode 100644 index 0000000..b2b2d64 --- /dev/null +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/TimestampType.java @@ -0,0 +1,163 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.types.logical; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.api.ValidationException; + +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.Set; + +/** + * Logical type of a timestamp WITHOUT time zone consisting of {@code year-month-day hour:minute:second[.fractional]} + * with up to nanosecond precision and values ranging from {@code 0000-01-01 00:00:00.000000000} to + * {@code 9999-12-31 23:59:59.999999999}. Compared to the SQL standard, leap seconds (23:59:60 and 23:59:61) + * are not supported as the semantics are closer to {@link java.time.LocalDateTime}. + * + * <p>The serialized string representation is {@code TIMESTAMP(p)} where {@code p} is the number of + * digits of fractional seconds (=precision). {@code p} must have a value between 0 and 9 (both inclusive). + * If no precision is specified, {@code p} is equal to 6. + * + * <p>A conversion from and to {@code long} is not supported as this would imply a time zone. However, + * this type is time zone free. For more {@link java.time.Instant}-like semantics use + * {@link LocalZonedTimestampType}. + * + * @see ZonedTimestampType + * @see LocalZonedTimestampType + */ +@PublicEvolving +public final class TimestampType extends LogicalType { + + private static final int MIN_PRECISION = 0; + + private static final int MAX_PRECISION = 9; + + private static final int DEFAULT_PRECISION = 6; + + private static final String FORMAT = "TIMESTAMP(%d)"; + + private static final Set<String> INPUT_OUTPUT_CONVERSION = conversionSet( + java.sql.Timestamp.class.getName(), + java.time.LocalDateTime.class.getName()); + + private static final Class<?> DEFAULT_CONVERSION = java.time.LocalDateTime.class; + + private final TimestampKind kind; + + private final int precision; + + /** + * Internal constructor that allows attaching additional metadata about time attribute + * properties. The additional metadata does not affect equality or serializability. + * + * <p>Use {@link #getKind()} for comparing this metadata. + */ + @Internal + public TimestampType(boolean isNullable, TimestampKind kind, int precision) { + super(isNullable, LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE); + if (precision < MIN_PRECISION || precision > MAX_PRECISION) { + throw new ValidationException( + String.format( + "Timestamp precision must be between %d and %d (both inclusive).", + MIN_PRECISION, + MAX_PRECISION)); + } + this.kind = kind; + this.precision = precision; + } + + public TimestampType(boolean isNullable, int precision) { + this(isNullable, TimestampKind.REGULAR, precision); + } + + public TimestampType(int precision) { + this(true, precision); + } + + public TimestampType() { + this(DEFAULT_PRECISION); + } + + @Internal + public TimestampKind getKind() { + return kind; + } + + public int getPrecision() { + return precision; + } + + @Override + public LogicalType copy(boolean isNullable) { + return new TimestampType(isNullable, kind, precision); + } + + @Override + public String asSerializableString() { + return withNullability(FORMAT, precision); + } + + @Override + public boolean supportsInputConversion(Class<?> clazz) { + return INPUT_OUTPUT_CONVERSION.contains(clazz.getName()); + } + + @Override + public boolean supportsOutputConversion(Class<?> clazz) { + return INPUT_OUTPUT_CONVERSION.contains(clazz.getName()); + } + + @Override + public Class<?> getDefaultConversion() { + return DEFAULT_CONVERSION; + } + + @Override + public List<LogicalType> getChildren() { + return Collections.emptyList(); + } + + @Override + public <R> R accept(LogicalTypeVisitor<R> visitor) { + return visitor.visit(this); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + if (!super.equals(o)) { + return false; + } + TimestampType that = (TimestampType) o; + return precision == that.precision; + } + + @Override + public int hashCode() { + return Objects.hash(super.hashCode(), precision); + } +} diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java index ef856e0..0411ada 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java @@ -30,6 +30,7 @@ import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.SmallIntType; import org.apache.flink.table.types.logical.TimeType; +import org.apache.flink.table.types.logical.TimestampType; import org.apache.flink.table.types.logical.TinyIntType; import org.apache.flink.table.types.logical.VarBinaryType; import org.apache.flink.table.types.logical.VarCharType; @@ -233,6 +234,19 @@ public class LogicalTypesTest { ); } + @Test + public void testTimestampType() { + testAll( + new TimestampType(9), + "TIMESTAMP(9)", + "TIMESTAMP(9)", + new Class[]{java.sql.Timestamp.class, java.time.LocalDateTime.class}, + new Class[]{java.time.LocalDateTime.class}, + new LogicalType[]{}, + new TimestampType(3) + ); + } + // -------------------------------------------------------------------------------------------- private static void testAll(