[flink] 04/04: [FLINK-13287][table-api] Port StreamRecordTimestamp to api-java
This is an automated email from the ASF dual-hosted git repository. twalthr pushed a commit to branch release-1.9 in repository https://gitbox.apache.org/repos/asf/flink.git commit c43dc951bf52e2f76a9c17021cfdf1ca8bc54781 Author: JingsongLi AuthorDate: Wed Jul 17 20:25:04 2019 +0800 [FLINK-13287][table-api] Port StreamRecordTimestamp to api-java This closes #9129. --- .../tsextractors/StreamRecordTimestamp.java| 75 ++ .../tsextractors/StreamRecordTimestamp.scala | 67 --- 2 files changed, 75 insertions(+), 67 deletions(-) diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.java new file mode 100644 index 000..fdb4384 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.java @@ -0,0 +1,75 @@ +/* + * 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.tsextractors; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.table.descriptors.Rowtime; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ResolvedFieldReference; +import org.apache.flink.table.expressions.utils.ApiExpressionUtils; + +import java.util.HashMap; +import java.util.Map; + +import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.STREAM_RECORD_TIMESTAMP; + +/** + * Extracts the timestamp of a StreamRecord into a rowtime attribute. + * + * Note: This extractor only works for StreamTableSources. + */ +@PublicEvolving +public final class StreamRecordTimestamp extends TimestampExtractor { + + private static final long serialVersionUID = 1L; + + public static final StreamRecordTimestamp INSTANCE = new StreamRecordTimestamp(); + + @Override + public String[] getArgumentFields() { + return new String[0]; + } + + @Override + public void validateArgumentFields(TypeInformation[] argumentFieldTypes) { + } + + @Override + public Expression getExpression(ResolvedFieldReference[] fieldAccesses) { + return ApiExpressionUtils.unresolvedCall(STREAM_RECORD_TIMESTAMP); + } + + @Override + public Map toProperties() { + Map map = new HashMap<>(); + map.put(Rowtime.ROWTIME_TIMESTAMPS_TYPE, Rowtime.ROWTIME_TIMESTAMPS_TYPE_VALUE_FROM_SOURCE); + return map; + } + + @Override + public boolean equals(Object o) { + return this == o || o != null && getClass() == o.getClass(); + } + + @Override + public int hashCode() { + return this.getClass().hashCode(); + } +} diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.scala deleted file mode 100644 index 087b1a6..000 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.scala +++ /dev/null @@ -1,67 +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. -
[flink] 04/04: [FLINK-13287][table-api] Port StreamRecordTimestamp to api-java
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 commit 2e3aab3f6464ecae7f60a561ae7541296ce5681c Author: JingsongLi AuthorDate: Wed Jul 17 20:25:04 2019 +0800 [FLINK-13287][table-api] Port StreamRecordTimestamp to api-java This closes #9129. --- .../tsextractors/StreamRecordTimestamp.java| 75 ++ .../tsextractors/StreamRecordTimestamp.scala | 67 --- 2 files changed, 75 insertions(+), 67 deletions(-) diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.java new file mode 100644 index 000..fdb4384 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.java @@ -0,0 +1,75 @@ +/* + * 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.tsextractors; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.table.descriptors.Rowtime; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ResolvedFieldReference; +import org.apache.flink.table.expressions.utils.ApiExpressionUtils; + +import java.util.HashMap; +import java.util.Map; + +import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.STREAM_RECORD_TIMESTAMP; + +/** + * Extracts the timestamp of a StreamRecord into a rowtime attribute. + * + * Note: This extractor only works for StreamTableSources. + */ +@PublicEvolving +public final class StreamRecordTimestamp extends TimestampExtractor { + + private static final long serialVersionUID = 1L; + + public static final StreamRecordTimestamp INSTANCE = new StreamRecordTimestamp(); + + @Override + public String[] getArgumentFields() { + return new String[0]; + } + + @Override + public void validateArgumentFields(TypeInformation[] argumentFieldTypes) { + } + + @Override + public Expression getExpression(ResolvedFieldReference[] fieldAccesses) { + return ApiExpressionUtils.unresolvedCall(STREAM_RECORD_TIMESTAMP); + } + + @Override + public Map toProperties() { + Map map = new HashMap<>(); + map.put(Rowtime.ROWTIME_TIMESTAMPS_TYPE, Rowtime.ROWTIME_TIMESTAMPS_TYPE_VALUE_FROM_SOURCE); + return map; + } + + @Override + public boolean equals(Object o) { + return this == o || o != null && getClass() == o.getClass(); + } + + @Override + public int hashCode() { + return this.getClass().hashCode(); + } +} diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.scala deleted file mode 100644 index 087b1a6..000 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/tsextractors/StreamRecordTimestamp.scala +++ /dev/null @@ -1,67 +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. -