twalthr commented on a change in pull request #11692: [FLINK-16992][table-common] Add all ability interfaces for table sources and sinks URL: https://github.com/apache/flink/pull/11692#discussion_r410287174
########## File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsWatermarkPushDown.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.connector.source.abilities; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.connector.source.ScanTableSource; + +/** + * Enables to push down watermarks into a {@link ScanTableSource}. + * + * <p>The concept of watermarks defines when time operations based on an event time attribute will be + * triggered. A watermark tells operators that no elements with a timestamp older or equal to the watermark + * timestamp should arrive at the operator. Thus, watermarks are a trade-off between latency and completeness. + * + * <p>Given the following SQL: + * <pre>{@code + * CREATE TABLE t (i INT, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND) // `ts` becomes a time attribute + * }</pre> + * + * <p>In the above example, generated watermarks are lagging 5 seconds behind the highest seen timestamp. + * + * <p>By default, if this interface is not implemented, watermarks are generated in a subsequent operation + * after the source. + * + * <p>However, for correctness, it might be necessary to perform the watermark generation as early as + * possible in order to be close to the actual data generation within a source's data partition. + * + * <p>This interface provides a {@link WatermarkAssignerProvider} that needs to be applied to a runtime + * implementation. Most built-in Flink sources provide a way of setting the watermark assigner. + * + * <p>Note: In many cases, watermarks are generated from computed columns. If a source implements this + * interface, it is recommended to also implement {@link SupportsComputedColumnPushDown}. + */ +@PublicEvolving +public interface SupportsWatermarkPushDown { + + /** + * Provides actual runtime implementation for generating watermarks. + * + * <p>There exist different interfaces for runtime implementation which is why {@link WatermarkAssignerProvider} + * serves as the base interface. Concrete {@link WatermarkAssignerProvider} interfaces might be located + * in other Flink modules. + * + * <p>See {@code org.apache.flink.table.connector.source.SourceFunctionProvider} in {@code flink-table-api-java-bridge}. + */ + void applyWatermark(WatermarkAssignerProvider assigner); Review comment: Yes, you are right. They should be interfaces. So the planner can generate the old interface + the new interface. If the source doesn't support the interface, we need to fail. I also added a comment to the JavaDoc about this. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
