You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2019/06/04 09:56:48 UTC
[flink] 01/03: [hotfix][table-api] Port TableSource specific
descriptors
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 99d4fd39f2b44630130d47b16ebcd69923f09e9a
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Fri May 24 12:55:48 2019 +0200
[hotfix][table-api] Port TableSource specific descriptors
---
.../flink/table/typeutils/FieldInfoUtils.java | 23 ++++--
.../table/sources/DefinedProctimeAttribute.java} | 38 ++++-----
.../table/sources/DefinedRowtimeAttributes.java} | 41 +++++-----
.../table/sources/RowtimeAttributeDescriptor.java | 77 ++++++++++++++++++
.../table/sources/definedTimeAttributes.scala | 95 ----------------------
5 files changed, 133 insertions(+), 141 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/typeutils/FieldInfoUtils.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/typeutils/FieldInfoUtils.java
index a0a3cc6..97c341c 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/typeutils/FieldInfoUtils.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/typeutils/FieldInfoUtils.java
@@ -49,7 +49,7 @@ import static java.lang.String.format;
import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.TIME_ATTRIBUTES;
/**
- * Utility classes for extracting names and indices of fields from different {@link TypeInformation}s.
+ * Utility methods for extracting names and indices of fields from different {@link TypeInformation}s.
*/
public class FieldInfoUtils {
@@ -234,8 +234,19 @@ public class FieldInfoUtils {
return fieldTypes;
}
- public static <T> TableSchema calculateTableSchema(
- TypeInformation<T> typeInfo,
+ /**
+ * Derives {@link TableSchema} out of a {@link TypeInformation}. It is complementary to other
+ * methods in this class. This also performs translation from time indicator markers such as
+ * {@link TimeIndicatorTypeInfo#ROWTIME_STREAM_MARKER} etc. to a corresponding
+ * {@link TimeIndicatorTypeInfo}.
+ *
+ * @param typeInfo input type info to calculate fields type infos from
+ * @param fieldIndexes indices within the typeInfo of the resulting Table schema
+ * @param fieldNames names of the fields of the resulting schema
+ * @return calculates resulting schema
+ */
+ public static TableSchema calculateTableSchema(
+ TypeInformation<?> typeInfo,
int[] fieldIndexes,
String[] fieldNames) {
@@ -292,6 +303,8 @@ public class FieldInfoUtils {
return new TableSchema(fieldNames, types);
}
+ /* Utility methods */
+
private static Optional<TypeInformation<?>> extractTimeMarkerType(int idx) {
switch (idx) {
case TimeIndicatorTypeInfo.ROWTIME_STREAM_MARKER:
@@ -306,10 +319,6 @@ public class FieldInfoUtils {
}
}
-
-
- /* Utility methods */
-
private static Set<FieldInfo> extractFieldInfoFromAtomicType(Expression[] exprs) {
boolean referenced = false;
FieldInfo fieldInfo = null;
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/StreamTableSourceTable.scala b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedProctimeAttribute.java
similarity index 51%
rename from flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/StreamTableSourceTable.scala
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedProctimeAttribute.java
index 38b7df8..803ca6a 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/StreamTableSourceTable.scala
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedProctimeAttribute.java
@@ -16,27 +16,27 @@
* limitations under the License.
*/
-package org.apache.flink.table.plan.schema
+package org.apache.flink.table.sources;
-import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
-import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.plan.stats.FlinkStatistic
-import org.apache.flink.table.sources.{StreamTableSource, TableSourceUtil}
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.table.api.TableSchema;
-class StreamTableSourceTable[T](
- tableSource: StreamTableSource[T],
- statistic: FlinkStatistic = FlinkStatistic.UNKNOWN)
- extends TableSourceTable[T](
- tableSource,
- statistic) {
+import javax.annotation.Nullable;
- TableSourceUtil.validateTableSource(tableSource)
+/**
+ * Extends a {@link TableSource} to specify a processing time attribute.
+ */
+@PublicEvolving
+public interface DefinedProctimeAttribute {
- def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
- TableSourceUtil.getRelDataType(
- tableSource,
- None,
- streaming = true,
- typeFactory.asInstanceOf[FlinkTypeFactory])
- }
+ /**
+ * Returns the name of a processing time attribute or null if no processing time attribute is
+ * present.
+ *
+ * <p>The referenced attribute must be present in the {@link TableSchema} of the {@link TableSource} and of
+ * type {@link Types#SQL_TIMESTAMP}.
+ */
+ @Nullable
+ String getProctimeAttribute();
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/BatchTableSourceTable.scala b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedRowtimeAttributes.java
similarity index 50%
rename from flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/BatchTableSourceTable.scala
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedRowtimeAttributes.java
index 14e8cb1..5238084 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/BatchTableSourceTable.scala
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedRowtimeAttributes.java
@@ -16,28 +16,29 @@
* limitations under the License.
*/
-package org.apache.flink.table.plan.schema
+package org.apache.flink.table.sources;
-import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
-import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.plan.stats.FlinkStatistic
-import org.apache.flink.table.sources.{BatchTableSource, TableSourceUtil}
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.table.api.TableSchema;
-class BatchTableSourceTable[T](
- tableSource: BatchTableSource[T],
- statistic: FlinkStatistic = FlinkStatistic.UNKNOWN)
- extends TableSourceTable[T](
- tableSource,
- statistic) {
+import java.util.List;
- TableSourceUtil.validateTableSource(tableSource)
+/**
+ * Extends a {@link TableSource} to specify rowtime attributes via a
+ * {@link RowtimeAttributeDescriptor}.
+ */
+@PublicEvolving
+public interface DefinedRowtimeAttributes {
- override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
- TableSourceUtil.getRelDataType(
- tableSource,
- None,
- streaming = false,
- typeFactory.asInstanceOf[FlinkTypeFactory])
- }
+ /**
+ * Returns a list of {@link RowtimeAttributeDescriptor} for all rowtime
+ * attributes of the table.
+ *
+ * <p>All referenced attributes must be present in the {@link TableSchema}
+ * of the {@link TableSource} and of type {@link Types#SQL_TIMESTAMP}.
+ *
+ * @return A list of {@link RowtimeAttributeDescriptor}.
+ */
+ List<RowtimeAttributeDescriptor> getRowtimeAttributeDescriptors();
}
-
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/RowtimeAttributeDescriptor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/RowtimeAttributeDescriptor.java
new file mode 100644
index 0000000..6851dd2
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/RowtimeAttributeDescriptor.java
@@ -0,0 +1,77 @@
+/*
+ * 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.sources;
+
+import org.apache.flink.table.sources.tsextractors.TimestampExtractor;
+import org.apache.flink.table.sources.wmstrategies.WatermarkStrategy;
+
+import java.util.Objects;
+
+/**
+ * Describes a rowtime attribute of a {@link TableSource}.
+ */
+public final class RowtimeAttributeDescriptor {
+
+ private final String attributeName;
+ private final TimestampExtractor timestampExtractor;
+ private final WatermarkStrategy watermarkStrategy;
+
+ public RowtimeAttributeDescriptor(
+ String attributeName,
+ TimestampExtractor timestampExtractor,
+ WatermarkStrategy watermarkStrategy) {
+ this.attributeName = attributeName;
+ this.timestampExtractor = timestampExtractor;
+ this.watermarkStrategy = watermarkStrategy;
+ }
+
+ /** Returns the name of the rowtime attribute. */
+ public String getAttributeName() {
+ return attributeName;
+ }
+
+ /** Returns the [[TimestampExtractor]] for the attribute. */
+ public TimestampExtractor getTimestampExtractor() {
+ return timestampExtractor;
+ }
+
+ /** Returns the [[WatermarkStrategy]] for the attribute. */
+ public WatermarkStrategy getWatermarkStrategy() {
+ return watermarkStrategy;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ RowtimeAttributeDescriptor that = (RowtimeAttributeDescriptor) o;
+ return Objects.equals(attributeName, that.attributeName) &&
+ Objects.equals(timestampExtractor, that.timestampExtractor) &&
+ Objects.equals(watermarkStrategy, that.watermarkStrategy);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(attributeName, timestampExtractor, watermarkStrategy);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/definedTimeAttributes.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/definedTimeAttributes.scala
deleted file mode 100644
index b144312..0000000
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/definedTimeAttributes.scala
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.sources
-
-import java.util
-import java.util.Objects
-import javax.annotation.Nullable
-
-import org.apache.flink.table.api.TableSchema
-import org.apache.flink.table.api.Types
-import org.apache.flink.table.sources.tsextractors.TimestampExtractor
-import org.apache.flink.table.sources.wmstrategies.WatermarkStrategy
-
-/**
- * Extends a [[TableSource]] to specify a processing time attribute.
- */
-trait DefinedProctimeAttribute {
-
- /**
- * Returns the name of a processing time attribute or null if no processing time attribute is
- * present.
- *
- * The referenced attribute must be present in the [[TableSchema]] of the [[TableSource]] and of
- * type [[Types.SQL_TIMESTAMP]].
- */
- @Nullable
- def getProctimeAttribute: String
-}
-
-/**
- * Extends a [[TableSource]] to specify rowtime attributes via a
- * [[RowtimeAttributeDescriptor]].
- */
-trait DefinedRowtimeAttributes {
-
- /**
- * Returns a list of [[RowtimeAttributeDescriptor]] for all rowtime attributes of the table.
- *
- * All referenced attributes must be present in the [[TableSchema]] of the [[TableSource]] and of
- * type [[Types.SQL_TIMESTAMP]].
- *
- * @return A list of [[RowtimeAttributeDescriptor]].
- */
- def getRowtimeAttributeDescriptors: util.List[RowtimeAttributeDescriptor]
-}
-
-/**
- * Describes a rowtime attribute of a [[TableSource]].
- *
- * @param attributeName The name of the rowtime attribute.
- * @param timestampExtractor The timestamp extractor to derive the values of the attribute.
- * @param watermarkStrategy The watermark strategy associated with the attribute.
- */
-class RowtimeAttributeDescriptor(
- val attributeName: String,
- val timestampExtractor: TimestampExtractor,
- val watermarkStrategy: WatermarkStrategy) {
-
- /** Returns the name of the rowtime attribute. */
- def getAttributeName: String = attributeName
-
- /** Returns the [[TimestampExtractor]] for the attribute. */
- def getTimestampExtractor: TimestampExtractor = timestampExtractor
-
- /** Returns the [[WatermarkStrategy]] for the attribute. */
- def getWatermarkStrategy: WatermarkStrategy = watermarkStrategy
-
- override def equals(other: Any): Boolean = other match {
- case that: RowtimeAttributeDescriptor =>
- Objects.equals(attributeName, that.attributeName) &&
- Objects.equals(timestampExtractor, that.timestampExtractor) &&
- Objects.equals(watermarkStrategy, that.watermarkStrategy)
- case _ => false
- }
-
- override def hashCode(): Int = {
- Objects.hash(attributeName, timestampExtractor, watermarkStrategy)
- }
-}