[FLINK-36488] [TABLE SQL/API] Remove deprecated methods StreamTableEnvironment.toAppendStream from flink-table-api-java-bridge module

docs/content.zh/docs/dev/datastream/operators/process_function.md

@@ -248,9 +248,8 @@ if __name__ == '__main__':
             )
         """)
 
-    stream = t_env.to_append_stream(
-        t_env.from_path('my_source'),
-        Types.ROW([Types.SQL_TIMESTAMP(), Types.STRING(), Types.STRING()]))
+    stream = t_env.to_data_stream(
+        t_env.from_path('my_source'))
     watermarked_stream = stream.assign_timestamps_and_watermarks(
         WatermarkStrategy.for_monotonous_timestamps()
                          .with_timestamp_assigner(MyTimestampAssigner()))

docs/content.zh/docs/dev/python/datastream/intro_to_datastream_api.md

@@ -227,9 +227,8 @@ t_env.execute_sql("""
         )
     """)
 
-ds = t_env.to_append_stream(
-    t_env.from_path('my_source'),
-    Types.ROW([Types.INT(), Types.STRING()]))
+ds = t_env.to_data_stream(
+    t_env.from_path('my_source'))
 ```
 
 <span class="label label-info">Note</span> The StreamExecutionEnvironment `env` should be specified
@@ -266,7 +265,7 @@ It also supports to convert a `DataStream` to a `Table` and vice verse.
 table = t_env.from_data_stream(ds, 'a, b, c')
 
 # convert a Table to a DataStream
-ds = t_env.to_append_stream(table, Types.ROW([Types.INT(), Types.STRING()]))
+ds = t_env.to_data_stream(table)
 # or
 ds = t_env.to_retract_stream(table, Types.ROW([Types.INT(), Types.STRING()]))
 ```

docs/content.zh/docs/dev/python/datastream/operators/overview.md

@@ -108,9 +108,8 @@ def data_stream_api_demo():
             )
         """)
 
-    ds = t_env.to_append_stream(
-        t_env.from_path('my_source'),
-        Types.ROW([Types.INT(), Types.STRING()]))
+    ds = t_env.to_data_stream(
+        t_env.from_path('my_source'))
 
     def split(s):
         splits = s[1].split("|")

docs/content.zh/docs/dev/table/data_stream_api.md

@@ -3151,11 +3151,11 @@ Table table = tableEnv.fromValues(
     row("sarah", 32));
 
 // Convert the Table into an append DataStream of Row by specifying the class
-DataStream<Row> dsRow = tableEnv.toAppendStream(table, Row.class);
+DataStream<Row> dsRow = tableEnv.toDataStream(table, Row.class);
 
 // Convert the Table into an append DataStream of Tuple2<String, Integer> with TypeInformation
 TupleTypeInfo<Tuple2<String, Integer>> tupleType = new TupleTypeInfo<>(Types.STRING(), Types.INT());
-DataStream<Tuple2<String, Integer>> dsTuple = tableEnv.toAppendStream(table, tupleType);
+DataStream<Tuple2<String, Integer>> dsTuple = tableEnv.toDataStream(table, tupleType);
 
 // Convert the Table into a retract DataStream of Row.
 // A retract stream of type X is a DataStream<Tuple2<Boolean, X>>. 
@@ -3178,11 +3178,12 @@ val table: Table = tableEnv.fromValues(
     row("sarah", 32))
 
 // Convert the Table into an append DataStream of Row by specifying the class
-val dsRow: DataStream[Row] = tableEnv.toAppendStream[Row](table)
+val dsRow: DataStream[Row] = tableEnv.toDataStream(table)
 
 // Convert the Table into an append DataStream of (String, Integer) with TypeInformation
-val dsTuple: DataStream[(String, Int)] dsTuple = 
-  tableEnv.toAppendStream[(String, Int)](table)
+case class User(var name: String, var age: java.lang.Integer)
+val dsTuple: DataStream[User] dsTuple =
+  tableEnv.toDataStream(table, DataTypes.of(User.class))
 
 // Convert the Table into a retract DataStream of Row.
 // A retract stream of type X is a DataStream<Tuple2<Boolean, X>>. 

docs/content/docs/dev/datastream/operators/process_function.md

@@ -255,9 +255,8 @@ if __name__ == '__main__':
             )
         """)
 
-    stream = t_env.to_append_stream(
-        t_env.from_path('my_source'),
-        Types.ROW([Types.SQL_TIMESTAMP(), Types.STRING(), Types.STRING()]))
+    stream = t_env.to_data_stream(
+        t_env.from_path('my_source'))
     watermarked_stream = stream.assign_timestamps_and_watermarks(
         WatermarkStrategy.for_monotonous_timestamps()
                          .with_timestamp_assigner(MyTimestampAssigner()))

docs/content/docs/dev/python/datastream/intro_to_datastream_api.md

@@ -227,9 +227,8 @@ t_env.execute_sql("""
         )
     """)
 
-ds = t_env.to_append_stream(
-    t_env.from_path('my_source'),
-    Types.ROW([Types.INT(), Types.STRING()]))
+ds = t_env.to_data_stream(
+    t_env.from_path('my_source'))
 ```
 
 <span class="label label-info">Note</span> The StreamExecutionEnvironment `env` should be specified
@@ -266,7 +265,7 @@ It also supports to convert a `DataStream` to a `Table` and vice verse.
 table = t_env.from_data_stream(ds, 'a, b, c')
 
 # convert a Table to a DataStream
-ds = t_env.to_append_stream(table, Types.ROW([Types.INT(), Types.STRING()]))
+ds = t_env.to_data_stream(table)
 # or
 ds = t_env.to_retract_stream(table, Types.ROW([Types.INT(), Types.STRING()]))
 ```

docs/content/docs/dev/python/datastream/operators/overview.md

@@ -108,9 +108,8 @@ def data_stream_api_demo():
             )
         """)
 
-    ds = t_env.to_append_stream(
-        t_env.from_path('my_source'),
-        Types.ROW([Types.INT(), Types.STRING()]))
+    ds = t_env.to_data_stream(
+        t_env.from_path('my_source'))
 
     def split(s):
         splits = s[1].split("|")

docs/content/docs/dev/table/data_stream_api.md

@@ -3142,11 +3142,11 @@ Table table = tableEnv.fromValues(
     row("sarah", 32));
 
 // Convert the Table into an append DataStream of Row by specifying the class
-DataStream<Row> dsRow = tableEnv.toAppendStream(table, Row.class);
+DataStream<Row> dsRow = tableEnv.toDataStream(table, Row.class);
 
 // Convert the Table into an append DataStream of Tuple2<String, Integer> with TypeInformation
 TupleTypeInfo<Tuple2<String, Integer>> tupleType = new TupleTypeInfo<>(Types.STRING(), Types.INT());
-DataStream<Tuple2<String, Integer>> dsTuple = tableEnv.toAppendStream(table, tupleType);
+DataStream<Tuple2<String, Integer>> dsTuple = tableEnv.toDataStream(table, tupleType);
 
 // Convert the Table into a retract DataStream of Row.
 // A retract stream of type X is a DataStream<Tuple2<Boolean, X>>. 
@@ -3169,11 +3169,12 @@ val table: Table = tableEnv.fromValues(
     row("sarah", 32))
 
 // Convert the Table into an append DataStream of Row by specifying the class
-val dsRow: DataStream[Row] = tableEnv.toAppendStream[Row](table)
+val dsRow: DataStream[Row] = tableEnv.toDataStream(table)
 
 // Convert the Table into an append DataStream of (String, Integer) with TypeInformation
-val dsTuple: DataStream[(String, Int)] dsTuple = 
-  tableEnv.toAppendStream[(String, Int)](table)
+case class User(var name: String, var age: java.lang.Integer)
+val dsTuple: DataStream[User] dsTuple =
+  tableEnv.toDataStream(table, DataTypes.of(User.class))
 
 // Convert the Table into a retract DataStream of Row.
 // A retract stream of type X is a DataStream<Tuple2<Boolean, X>>. 
@@ -3194,10 +3195,10 @@ table = t_env.from_elements([("john", 35), ("sarah", 32)],
                              DataTypes.FIELD("age", DataTypes.INT())]))
 
 # Convert the Table into an append DataStream of Row by specifying the type information
-ds_row = t_env.to_append_stream(table, Types.ROW([Types.STRING(), Types.INT()]))
+ds_row = t_env.to_data_stream(table)
 
 # Convert the Table into an append DataStream of Tuple[str, int] with TypeInformation
-ds_tuple = t_env.to_append_stream(table, Types.TUPLE([Types.STRING(), Types.INT()]))
+ds_tuple = t_env.to_data_stream(table)
 
 # Convert the Table into a retract DataStream of Row by specifying the type information
 # A retract stream of type X is a DataStream of Tuple[bool, X]. 

flink-python/docs/reference/pyflink.table/table_environment.rst

@@ -256,7 +256,6 @@ StreamTableEnvironment
     StreamTableEnvironment.sql_query
     StreamTableEnvironment.to_data_stream
     StreamTableEnvironment.to_changelog_stream
-    StreamTableEnvironment.to_append_stream
     StreamTableEnvironment.to_retract_stream
     StreamTableEnvironment.unload_module
     StreamTableEnvironment.use_catalog

flink-python/pyflink/datastream/tests/test_stream_execution_environment.py

@@ -301,7 +301,7 @@ def plus_three(value):
 
         tab = t_env.from_data_stream(ds, col('a')) \
                    .select(add_three(col('a')))
-        t_env.to_append_stream(tab, Types.ROW([Types.LONG()])) \
+        t_env.to_data_stream(tab) \
              .map(lambda i: i[0]) \
              .add_sink(self.test_sink)
         env.execute("test add_python_file")

flink-python/pyflink/table/table_environment.py

@@ -1952,24 +1952,6 @@ def to_changelog_stream(self,
                     target_schema._j_schema,
                     changelog_mode._j_changelog_mode))
 
-    def to_append_stream(self, table: Table, type_info: TypeInformation) -> DataStream:
-        """
-        Converts the given Table into a DataStream of a specified type. The Table must only have
-        insert (append) changes. If the Table is also modified by update or delete changes, the
-        conversion will fail.
-
-        The fields of the Table are mapped to DataStream as follows: Row and Tuple types: Fields are
-        mapped by position, field types must match.
-
-        :param table: The Table to convert.
-        :param type_info: The TypeInformation that specifies the type of the DataStream.
-        :return: The converted DataStream.
-
-        .. versionadded:: 1.12.0
-        """
-        j_data_stream = self._j_tenv.toAppendStream(table._j_table, type_info.get_java_type_info())
-        return DataStream(j_data_stream=j_data_stream)
-
     def to_retract_stream(self, table: Table, type_info: TypeInformation) -> DataStream:
         """
         Converts the given Table into a DataStream of add and retract messages. The message will be

flink-python/pyflink/table/tests/test_table_environment_api.py

@@ -600,23 +600,6 @@ def test_from_and_to_changelog_stream_event_time(self):
         actual_results.sort()
         self.assertEqual(expected_results, actual_results)
 
-    def test_to_append_stream(self):
-        self.env.set_parallelism(1)
-        t_env = StreamTableEnvironment.create(
-            self.env,
-            environment_settings=EnvironmentSettings.in_streaming_mode())
-        table = t_env.from_elements([(1, "Hi", "Hello"), (2, "Hello", "Hi")], ["a", "b", "c"])
-        new_table = table.select(table.a + 1, table.b + 'flink', table.c)
-        ds = t_env.to_append_stream(table=new_table, type_info=Types.ROW([Types.LONG(),
-                                                                          Types.STRING(),
-                                                                          Types.STRING()]))
-        test_sink = DataStreamTestSinkFunction()
-        ds.add_sink(test_sink)
-        self.env.execute("test_to_append_stream")
-        result = test_sink.get_results(False)
-        expected = ['+I[2, Hiflink, Hello]', '+I[3, Helloflink, Hi]']
-        self.assertEqual(result, expected)
-
     def test_to_retract_stream(self):
         self.env.set_parallelism(1)
         t_env = StreamTableEnvironment.create(

flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/StreamTableEnvironment.java

@@ -763,61 +763,6 @@ DataStream<Row> toChangelogStream(
     @Deprecated
     <T> void createTemporaryView(String path, DataStream<T> dataStream, Expression... fields);
 
-    /**
-     * Converts the given {@link Table} into an append {@link DataStream} of a specified type.
-     *
-     * <p>The {@link Table} must only have insert (append) changes. If the {@link Table} is also
-     * modified by update or delete changes, the conversion will fail.
-     *
-     * <p>The fields of the {@link Table} are mapped to {@link DataStream} fields as follows:
-     *
-     * <ul>
-     *   <li>{@link Row} and {@link org.apache.flink.api.java.tuple.Tuple} types: Fields are mapped
-     *       by position, field types must match.
-     *   <li>POJO {@link DataStream} types: Fields are mapped by field name, field types must match.
-     * </ul>
-     *
-     * @param table The {@link Table} to convert.
-     * @param clazz The class of the type of the resulting {@link DataStream}.
-     * @param <T> The type of the resulting {@link DataStream}.
-     * @return The converted {@link DataStream}.
-     * @deprecated Use {@link #toDataStream(Table, Class)} instead. It integrates with the new type
-     *     system and supports all kinds of {@link DataTypes} that the table runtime can produce.
-     *     The semantics might be slightly different for raw and structured types. Use {@code
-     *     toDataStream(DataTypes.of(TypeInformation.of(Class)))} if {@link TypeInformation} should
-     *     be used as source of truth.
-     */
-    @Deprecated
-    <T> DataStream<T> toAppendStream(Table table, Class<T> clazz);
-
-    /**
-     * Converts the given {@link Table} into an append {@link DataStream} of a specified type.
-     *
-     * <p>The {@link Table} must only have insert (append) changes. If the {@link Table} is also
-     * modified by update or delete changes, the conversion will fail.
-     *
-     * <p>The fields of the {@link Table} are mapped to {@link DataStream} fields as follows:
-     *
-     * <ul>
-     *   <li>{@link Row} and {@link org.apache.flink.api.java.tuple.Tuple} types: Fields are mapped
-     *       by position, field types must match.
-     *   <li>POJO {@link DataStream} types: Fields are mapped by field name, field types must match.
-     * </ul>
-     *
-     * @param table The {@link Table} to convert.
-     * @param typeInfo The {@link TypeInformation} that specifies the type of the {@link
-     *     DataStream}.
-     * @param <T> The type of the resulting {@link DataStream}.
-     * @return The converted {@link DataStream}.
-     * @deprecated Use {@link #toDataStream(Table, Class)} instead. It integrates with the new type
-     *     system and supports all kinds of {@link DataTypes} that the table runtime can produce.
-     *     The semantics might be slightly different for raw and structured types. Use {@code
-     *     toDataStream(DataTypes.of(TypeInformation.of(Class)))} if {@link TypeInformation} should
-     *     be used as source of truth.
-     */
-    @Deprecated
-    <T> DataStream<T> toAppendStream(Table table, TypeInformation<T> typeInfo);
-
     /**
      * Converts the given {@link Table} into a {@link DataStream} of add and retract messages. The
      * message will be encoded as {@link Tuple2}. The first field is a {@link Boolean} flag, the

flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java

@@ -54,7 +54,6 @@
 import org.apache.flink.table.resource.ResourceManager;
 import org.apache.flink.table.types.AbstractDataType;
 import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.utils.TypeConversions;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.FlinkUserCodeClassLoaders;
 import org.apache.flink.util.MutableURLClassLoader;
@@ -316,22 +315,6 @@ public <T> void createTemporaryView(
         createTemporaryView(path, fromDataStream(dataStream, fields));
     }
 
-    @Override
-    public <T> DataStream<T> toAppendStream(Table table, Class<T> clazz) {
-        TypeInformation<T> typeInfo = extractTypeInformation(table, clazz);
-        return toAppendStream(table, typeInfo);
-    }
-
-    @Override
-    public <T> DataStream<T> toAppendStream(Table table, TypeInformation<T> typeInfo) {
-        OutputConversionModifyOperation modifyOperation =
-                new OutputConversionModifyOperation(
-                        table.getQueryOperation(),
-                        TypeConversions.fromLegacyInfoToDataType(typeInfo),
-                        OutputConversionModifyOperation.UpdateMode.APPEND);
-        return toStreamInternal(table, modifyOperation);
-    }
-
     @Override
     public <T> DataStream<Tuple2<Boolean, T>> toRetractStream(Table table, Class<T> clazz) {
         TypeInformation<T> typeInfo = extractTypeInformation(table, clazz);

flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/StreamTableEnvironment.scala

@@ -721,32 +721,6 @@ trait StreamTableEnvironment extends TableEnvironment {
   @deprecated
   def createTemporaryView[T](path: String, dataStream: DataStream[T], fields: Expression*): Unit
 
-  /**
-   * Converts the given [[Table]] into an append [[DataStream]] of a specified type.
-   *
-   * The [[Table]] must only have insert (append) changes. If the [[Table]] is also modified by
-   * update or delete changes, the conversion will fail.
-   *
-   * The fields of the [[Table]] are mapped to [[DataStream]] fields as follows:
-   *   - [[Row]] and Scala Tuple types: Fields are mapped by position, field types must match.
-   *   - POJO [[DataStream]] types: Fields are mapped by field name, field types must match.
-   *
-   * @param table
-   *   The [[Table]] to convert.
-   * @tparam T
-   *   The type of the resulting [[DataStream]].
-   * @return
-   *   The converted [[DataStream]].
-   * @deprecated
-   *   Use [[toDataStream(Table, Class)]] instead. It integrates with the new type system and
-   *   supports all kinds of [[DataTypes]] that the table runtime can produce. The semantics might
-   *   be slightly different for raw and structured types. Use
-   *   `toDataStream(DataTypes.of(Types.of[Class]))` if [[TypeInformation]] should be used as source
-   *   of truth.
-   */
-  @deprecated
-  def toAppendStream[T: TypeInformation](table: Table): DataStream[T]
-
   /**
    * Converts the given [[Table]] into a [[DataStream]] of add and retract messages. The message
    * will be encoded as [[Tuple2]]. The first field is a [[Boolean]] flag, the second field holds

flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/TableConversions.scala

@@ -189,40 +189,6 @@ class TableConversions(table: Table) {
   // Legacy before FLIP-136
   // ----------------------------------------------------------------------------------------------
 
-  /**
-   * Converts the given [[Table]] into an append [[DataStream]] of a specified type.
-   *
-   * The [[Table]] must only have insert (append) changes. If the [[Table]] is also modified by
-   * update or delete changes, the conversion will fail.
-   *
-   * The fields of the [[Table]] are mapped to [[DataStream]] fields as follows:
-   *   - [[org.apache.flink.types.Row]] and Scala Tuple types: Fields are mapped by position, field
-   *     types must match.
-   *   - POJO [[DataStream]] types: Fields are mapped by field name, field types must match.
-   *
-   * @tparam T
-   *   The type of the resulting [[DataStream]].
-   * @return
-   *   The converted [[DataStream]].
-   * @deprecated
-   *   Use [[toDataStream(Table, Class)]] instead. It integrates with the new type system and
-   *   supports all kinds of [[DataTypes]] that the table runtime can produce. The semantics might
-   *   be slightly different for raw and structured types. Use
-   *   `toDataStream(DataTypes.of(Types.of[Class]))` if [[TypeInformation]] should be used as source
-   *   of truth.
-   */
-  @deprecated
-  def toAppendStream[T: TypeInformation]: DataStream[T] = {
-    internalEnv match {
-      case tEnv: StreamTableEnvironment =>
-        tEnv.toAppendStream(table)
-      case _ =>
-        throw new ValidationException(
-          "Only tables that originate from Scala DataStreams " +
-            "can be converted to Scala DataStreams.")
-    }
-  }
-
   /**
    * Converts the [[Table]] to a [[DataStream]] of add and retract messages. The message will be
    * encoded as [[Tuple2]]. The first field is a [[Boolean]] flag, the second field holds the record