[FLINK-16036][table-api] Deprecate String based Expression DSL in TableEnvironments

This closes #11527

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

@@ -93,36 +93,90 @@ public interface BatchTableEnvironment extends TableEnvironment {
 	/**
 	 * Converts the given {@link DataSet} into a {@link Table} with specified field names.
 	 *
+	 * <p>There are two modes for mapping original fields to the fields of the {@link Table}:
+	 *
+	 * <p>1. Reference input fields by name:
+	 * All fields in the schema definition are referenced by name
+	 * (and possibly renamed using an alias (as). In this mode, fields can be reordered and
+	 * projected out. This mode can be used for any input type, including POJOs.
+	 *
 	 * <p>Example:
 	 *
 	 * <pre>
 	 * {@code
 	 *   DataSet<Tuple2<String, Long>> set = ...
-	 *   Table tab = tableEnv.fromDataSet(set, "a, b");
+	 *   // use the original 'f0' field and give a better name to the 'f1' field
+	 *   Table table = tableEnv.fromTable(set, "f0, f1 as name");
+	 * }
+	 * </pre>
+	 *
+	 * <p>2. Reference input fields by position:
+	 * In this mode, fields are simply renamed. This mode can only be
+	 * used if the input type has a defined field order (tuple, case class, Row) and none of
+	 * the {@code fields} references a field of the input type.
+	 *
+	 * <p>Example:
+	 *
+	 * <pre>
+	 * {@code
+	 *   DataSet<Tuple2<String, Long>> set = ...
+	 *   // renames the original fields as 'a' and 'b'
+	 *   Table table = tableEnv.fromDataSet(set, "a, b");
 	 * }
 	 * </pre>
 	 *
 	 * @param dataSet The {@link DataSet} to be converted.
-	 * @param fields The field names of the resulting {@link Table}.
+	 * @param fields The fields expressions to map original fields of the DataSet to the fields of the {@link Table}.
 	 * @param <T> The type of the {@link DataSet}.
 	 * @return The converted {@link Table}.
+	 * @deprecated use {@link #fromDataSet(DataSet, Expression...)}
 	 */
+	@Deprecated
 	<T> Table fromDataSet(DataSet<T> dataSet, String fields);
 
 	/**
 	 * Converts the given {@link DataSet} into a {@link Table} with specified field names.
 	 *
-	 * Example:
+	 * <p>There are two modes for mapping original fields to the fields of the {@link Table}:
+	 *
+	 * <p>1. Reference input fields by name:
+	 * All fields in the schema definition are referenced by name
+	 * (and possibly renamed using an alias (as). In this mode, fields can be reordered and
+	 * projected out. This mode can be used for any input type, including POJOs.
+	 *
+	 * <p>Example:
 	 *
 	 * <pre>
 	 * {@code
 	 *   DataSet<Tuple2<String, Long>> set = ...
-	 *   Table tab = tableEnv.fromDataSet(set, $("a"), $("b").as("name"), $("timestamp").rowtime());
+	 *   Table table = tableEnv.fromDataSet(
+	 *      set,
+	 *      $("f1"), // reorder and use the original field
+	 *      $("f0").as("name") // reorder and give the original field a better name
+	 *   );
+	 * }
+	 * </pre>
+	 *
+	 * <p>2. Reference input fields by position:
+	 * In this mode, fields are simply renamed. This mode can only be
+	 * used if the input type has a defined field order (tuple, case class, Row) and none of
+	 * the {@code fields} references a field of the input type.
+	 *
+	 * <p>Example:
+	 *
+	 * <pre>
+	 * {@code
+	 *   DataSet<Tuple2<String, Long>> set = ...
+	 *   Table table = tableEnv.fromDataSet(
+	 *      set,
+	 *      $("a"), // renames the first field to 'a'
+	 *      $("b") // renames the second field to 'b'
+	 *   );
 	 * }
 	 * </pre>
 	 *
 	 * @param dataSet The {@link DataSet} to be converted.
-	 * @param fields The field names of the resulting {@link Table}.
+	 * @param fields The fields expressions to map original fields of the DataSet to the fields of the {@link Table}.
 	 * @param <T> The type of the {@link DataSet}.
 	 * @return The converted {@link Table}.
 	 */
@@ -172,11 +226,34 @@ public interface BatchTableEnvironment extends TableEnvironment {
 	 * Creates a view from the given {@link DataSet} in a given path with specified field names.
 	 * Registered views can be referenced in SQL queries.
 	 *
+	 * <p>There are two modes for mapping original fields to the fields of the View:
+	 *
+	 * <p>1. Reference input fields by name:
+	 * All fields in the schema definition are referenced by name
+	 * (and possibly renamed using an alias (as). In this mode, fields can be reordered and
+	 * projected out. This mode can be used for any input type, including POJOs.
+	 *
+	 * <p>Example:
+	 *
+	 * <pre>
+	 * {@code
+	 *   DataSet<Tuple2<String, Long>> set = ...
+	 *   // use the original 'f0' field and give a better name to the 'f1' field
+	 *   tableEnv.registerDataSet("myTable", set, "f0, f1 as name");
+	 * }
+	 * </pre>
+	 *
+	 * <p>2. Reference input fields by position:
+	 * In this mode, fields are simply renamed. This mode can only be
+	 * used if the input type has a defined field order (tuple, case class, Row) and none of the {@code fields}
+	 * references a field of the input type.
+	 *
 	 * <p>Example:
 	 *
 	 * <pre>
 	 * {@code
 	 *   DataSet<Tuple2<String, Long>> set = ...
+	 *   // renames the original fields as 'a' and 'b'
 	 *   tableEnv.registerDataSet("myTable", set, "a, b");
 	 * }
 	 * </pre>
@@ -190,7 +267,7 @@ public interface BatchTableEnvironment extends TableEnvironment {
 	 *
 	 * @param name The name under which the {@link DataSet} is registered in the catalog.
 	 * @param dataSet The {@link DataSet} to register.
-	 * @param fields The field names of the registered view.
+	 * @param fields The fields expressions to map original fields of the DataSet to the fields of the View.
 	 * @param <T> The type of the {@link DataSet} to register.
 	 * @deprecated use {@link #createTemporaryView(String, DataSet, String)}
 	 */
@@ -201,11 +278,34 @@ public interface BatchTableEnvironment extends TableEnvironment {
 	 * Creates a view from the given {@link DataSet} in a given path with specified field names.
 	 * Registered views can be referenced in SQL queries.
 	 *
+	 * <p>There are two modes for mapping original fields to the fields of the View:
+	 *
+	 * <p>1. Reference input fields by name:
+	 * All fields in the schema definition are referenced by name
+	 * (and possibly renamed using an alias (as). In this mode, fields can be reordered and
+	 * projected out. This mode can be used for any input type, including POJOs.
+	 *
 	 * <p>Example:
 	 *
 	 * <pre>
 	 * {@code
 	 *   DataSet<Tuple2<String, Long>> set = ...
+	 *   // use the original 'f0' field and give a better name to the 'f1' field
+	 *   tableEnv.createTemporaryView("cat.db.myTable", set, "f0, f1 as name");
+	 * }
+	 * </pre>
+	 *
+	 * <p>2. Reference input fields by position:
+	 * In this mode, fields are simply renamed. This mode can only be
+	 * used if the input type has a defined field order (tuple, case class, Row) and none of the {@code fields}
+	 * references a field of the input type.
+	 *
+	 * <p>Example:
+	 *
+	 * <pre>
+	 * {@code
+	 *   DataSet<Tuple2<String, Long>> set = ...
+	 *   // renames the original fields as 'a' and 'b'
 	 *   tableEnv.createTemporaryView("cat.db.myTable", set, "a, b");
 	 * }
 	 * </pre>
@@ -217,21 +317,54 @@ public interface BatchTableEnvironment extends TableEnvironment {
 	 * @param path The path under which the view is created.
 	 *             See also the {@link TableEnvironment} class description for the format of the path.
 	 * @param dataSet The {@link DataSet} out of which to create the view.
-	 * @param fields The field names of the registered view.
+	 * @param fields The fields expressions to map original fields of the DataSet to the fields of the View.
 	 * @param <T> The type of the {@link DataSet}.
+	 * @deprecated use {@link #createTemporaryView(String, DataSet, Expression...)}
 	 */
+	@Deprecated
 	<T> void createTemporaryView(String path, DataSet<T> dataSet, String fields);
 
 	/**
 	 * Creates a view from the given {@link DataSet} in a given path with specified field names.
 	 * Registered views can be referenced in SQL queries.
 	 *
+	 * <p>There are two modes for mapping original fields to the fields of the View:
+	 *
+	 * <p>1. Reference input fields by name:
+	 * All fields in the schema definition are referenced by name
+	 * (and possibly renamed using an alias (as). In this mode, fields can be reordered and
+	 * projected out. This mode can be used for any input type, including POJOs.
+	 *
+	 * <p>Example:
+	 *
+	 * <pre>
+	 * {@code
+	 *   DataSet<Tuple2<String, Long>> set = ...
+	 *   tableEnv.createTemporaryView(
+	 *      "cat.db.myTable",
+	 *      set,
+	 *      $("f1"), // reorder and use the original field
+	 *      $("f0").as("name") // reorder and give the original field a better name
+	 *   );
+	 * }
+	 * </pre>
+	 *
+	 * <p>2. Reference input fields by position:
+	 * In this mode, fields are simply renamed. This mode can only be
+	 * used if the input type has a defined field order (tuple, case class, Row) and none of
+	 * the {@code fields} references a field of the input type.
+	 *
 	 * <p>Example:
 	 *
 	 * <pre>
 	 * {@code
 	 *   DataSet<Tuple2<String, Long>> set = ...
-	 *   tableEnv.createTemporaryView("cat.db.myTable", set, $("b").as("name"), $("timestamp").rowtime());
+	 *   tableEnv.createTemporaryView(
+	 *      "cat.db.myTable",
+	 *      set,
+	 *      $("a"), // renames the first field to 'a'
+	 *      $("b") // renames the second field to 'b'
+	 *   );
 	 * }
 	 * </pre>
 	 *
@@ -242,7 +375,7 @@ public interface BatchTableEnvironment extends TableEnvironment {
 	 * @param path The path under which the view is created.
 	 *             See also the {@link TableEnvironment} class description for the format of the path.
 	 * @param dataSet The {@link DataSet} out of which to create the view.
-	 * @param fields The field names of the registered view.
+	 * @param fields The fields expressions to map original fields of the DataSet to the fields of the View.
 	 * @param <T> The type of the {@link DataSet}.
 	 */
 	<T> void createTemporaryView(String path, DataSet<T> dataSet, Expression... fields);

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

@@ -195,36 +195,104 @@ public interface StreamTableEnvironment extends TableEnvironment {
 	/**
 	 * Converts the given {@link DataStream} into a {@link Table} with specified field names.
 	 *
+	 * <p>There are two modes for mapping original fields to the fields of the {@link Table}:
+	 *
+	 * <p>1. Reference input fields by name:
+	 * All fields in the schema definition are referenced by name
+	 * (and possibly renamed using an alias (as). Moreover, we can define proctime and rowtime
+	 * attributes at arbitrary positions using arbitrary names (except those that exist in the
+	 * result schema). In this mode, fields can be reordered and projected out. This mode can
+	 * be used for any input type, including POJOs.
+	 *
 	 * <p>Example:
 	 *
 	 * <pre>
 	 * {@code
 	 *   DataStream<Tuple2<String, Long>> stream = ...
-	 *   Table tab = tableEnv.fromDataStream(stream, "a, b");
+	 *   // reorder the fields, rename the original 'f0' field to 'name' and add event-time
+	 *   // attribute named 'rowtime'
+	 *   Table table = tableEnv.fromDataStream(stream, "f1, rowtime.rowtime, f0 as 'name'");
+	 * }
+	 * </pre>
+	 *
+	 * <p>2. Reference input fields by position:
+	 * In this mode, fields are simply renamed. Event-time attributes can
+	 * replace the field on their position in the input data (if it is of correct type) or be
+	 * appended at the end. Proctime attributes must be appended at the end. This mode can only be
+	 * used if the input type has a defined field order (tuple, case class, Row) and none of
+	 * the {@code fields} references a field of the input type.
+	 *
+	 * <p>Example:
+	 *
+	 * <pre>
+	 * {@code
+	 *   DataStream<Tuple2<String, Long>> stream = ...
+	 *   // rename the original fields to 'a' and 'b' and extract the internally attached timestamp into an event-time
+	 *   // attribute named 'rowtime'
+	 *   Table table = tableEnv.fromDataStream(stream, "a, b, rowtime.rowtime");
 	 * }
 	 * </pre>
 	 *
 	 * @param dataStream The {@link DataStream} to be converted.
-	 * @param fields The field names of the resulting {@link Table}.
+	 * @param fields The fields expressions to map original fields of the DataStream to the fields of the {@link Table}.
 	 * @param <T> The type of the {@link DataStream}.
 	 * @return The converted {@link Table}.
+	 * @deprecated use {@link #fromDataStream(DataStream, Expression...)}
 	 */
+	@Deprecated
 	<T> Table fromDataStream(DataStream<T> dataStream, String fields);
 
 	/**
 	 * Converts the given {@link DataStream} into a {@link Table} with specified field names.
 	 *
+	 * <p>There are two modes for mapping original fields to the fields of the {@link Table}:
+	 *
+	 * <p>1. Reference input fields by name:
+	 * All fields in the schema definition are referenced by name
+	 * (and possibly renamed using an alias (as). Moreover, we can define proctime and rowtime
+	 * attributes at arbitrary positions using arbitrary names (except those that exist in the
+	 * result schema). In this mode, fields can be reordered and projected out. This mode can
+	 * be used for any input type, including POJOs.
+	 *
 	 * <p>Example:
 	 *
 	 * <pre>
 	 * {@code
 	 *   DataStream<Tuple2<String, Long>> stream = ...
-	 *   Table tab = tableEnv.fromDataStream(stream, $("a"), $("b").as("name"), $("timestamp").rowtime());
+	 *   Table table = tableEnv.fromDataStream(
+	 *      stream,
+	 *      $("f1"), // reorder and use the original field
+	 *      $("rowtime").rowtime(), // extract the internally attached timestamp into an event-time
+	 *                              // attribute named 'rowtime'
+	 *      $("f0").as("name") // reorder and give the original field a better name
+	 *   );
+	 * }
+	 * </pre>
+	 *
+	 * <p>2. Reference input fields by position:
+	 * In this mode, fields are simply renamed. Event-time attributes can
+	 * replace the field on their position in the input data (if it is of correct type) or be
+	 * appended at the end. Proctime attributes must be appended at the end. This mode can only be
+	 * used if the input type has a defined field order (tuple, case class, Row) and none of
+	 * the {@code fields} references a field of the input type.
+	 *
+	 * <p>Example:
+	 *
+	 * <pre>
+	 * {@code
+	 *   DataStream<Tuple2<String, Long>> stream = ...
+	 *   Table table = tableEnv.fromDataStream(
+	 *      stream,
+	 *      $("a"), // rename the first field to 'a'
+	 *      $("b"), // rename the second field to 'b'
+	 *      $("rowtime").rowtime() // extract the internally attached timestamp into an event-time
+	 *                             // attribute named 'rowtime'
+	 *   );
 	 * }
 	 * </pre>
 	 *
 	 * @param dataStream The {@link DataStream} to be converted.
-	 * @param fields The field expressions of the resulting {@link Table}.
+	 * @param fields The fields expressions to map original fields of the DataStream to the fields of the {@code Table}.
 	 * @param <T> The type of the {@link DataStream}.
 	 * @return The converted {@link Table}.
 	 */
@@ -274,12 +342,41 @@ public interface StreamTableEnvironment extends TableEnvironment {
 	 * Creates a view from the given {@link DataStream} in a given path with specified field names.
 	 * Registered views can be referenced in SQL queries.
 	 *
+	 * <p>There are two modes for mapping original fields to the fields of the View:
+	 *
+	 * <p>1. Reference input fields by name:
+	 * All fields in the schema definition are referenced by name
+	 * (and possibly renamed using an alias (as). Moreover, we can define proctime and rowtime
+	 * attributes at arbitrary positions using arbitrary names (except those that exist in the
+	 * result schema). In this mode, fields can be reordered and projected out. This mode can
+	 * be used for any input type, including POJOs.
+	 *
+	 * <p>Example:
+	 *
+	 * <pre>
+	 * {@code
+	 *   DataStream<Tuple2<String, Long>> stream = ...
+	 *   // reorder the fields, rename the original 'f0' field to 'name' and add event-time
+	 *   // attribute named 'rowtime'
+	 *   tableEnv.registerDataStream("myTable", stream, "f1, rowtime.rowtime, f0 as 'name'");
+	 * }
+	 * </pre>
+	 *
+	 * <p>2. Reference input fields by position:
+	 * In this mode, fields are simply renamed. Event-time attributes can
+	 * replace the field on their position in the input data (if it is of correct type) or be
+	 * appended at the end. Proctime attributes must be appended at the end. This mode can only be
+	 * used if the input type has a defined field order (tuple, case class, Row) and none of
+	 * the {@code fields} references a field of the input type.
+	 *
 	 * <p>Example:
 	 *
 	 * <pre>
 	 * {@code
 	 *   DataStream<Tuple2<String, Long>> stream = ...
-	 *   tableEnv.registerDataStream("myTable", stream, "a, b")
+	 *   // rename the original fields to 'a' and 'b' and extract the internally attached timestamp into an event-time
+	 *   // attribute named 'rowtime'
+	 *   tableEnv.registerDataStream("myTable", stream, "a, b, rowtime.rowtime");
 	 * }
 	 * </pre>
 	 *
@@ -292,7 +389,7 @@ public interface StreamTableEnvironment extends TableEnvironment {
 	 *
 	 * @param name The name under which the {@link DataStream} is registered in the catalog.
 	 * @param dataStream The {@link DataStream} to register.
-	 * @param fields The field names of the registered view.
+	 * @param fields The fields expressions to map original fields of the DataStream to the fields of the View.
 	 * @param <T> The type of the {@link DataStream} to register.
 	 * @deprecated use {@link #createTemporaryView(String, DataStream, Expression...)}
 	 */
@@ -303,12 +400,41 @@ public interface StreamTableEnvironment extends TableEnvironment {
 	 * Creates a view from the given {@link DataStream} in a given path with specified field names.
 	 * Registered views can be referenced in SQL queries.
 	 *
+	 * <p>There are two modes for mapping original fields to the fields of the View:
+	 *
+	 * <p>1. Reference input fields by name:
+	 * All fields in the schema definition are referenced by name
+	 * (and possibly renamed using an alias (as). Moreover, we can define proctime and rowtime
+	 * attributes at arbitrary positions using arbitrary names (except those that exist in the
+	 * result schema). In this mode, fields can be reordered and projected out. This mode can
+	 * be used for any input type, including POJOs.
+	 *
 	 * <p>Example:
 	 *
 	 * <pre>
 	 * {@code
 	 *   DataStream<Tuple2<String, Long>> stream = ...
-	 *   tableEnv.createTemporaryView("cat.db.myTable", stream, "a, b")
+	 *   // reorder the fields, rename the original 'f0' field to 'name' and add event-time
+	 *   // attribute named 'rowtime'
+	 *   tableEnv.createTemporaryView("cat.db.myTable", stream, "f1, rowtime.rowtime, f0 as 'name'");
+	 * }
+	 * </pre>
+	 *
+	 * <p>2. Reference input fields by position:
+	 * In this mode, fields are simply renamed. Event-time attributes can
+	 * replace the field on their position in the input data (if it is of correct type) or be
+	 * appended at the end. Proctime attributes must be appended at the end. This mode can only be
+	 * used if the input type has a defined field order (tuple, case class, Row) and none of
+	 * the {@code fields} references a field of the input type.
+	 *
+	 * <p>Example:
+	 *
+	 * <pre>
+	 * {@code
+	 *   DataStream<Tuple2<String, Long>> stream = ...
+	 *   // rename the original fields to 'a' and 'b' and extract the internally attached timestamp into an event-time
+	 *   // attribute named 'rowtime'
+	 *   tableEnv.createTemporaryView("cat.db.myTable", stream, "a, b, rowtime.rowtime");
 	 * }
 	 * </pre>
 	 *
@@ -319,21 +445,61 @@ public interface StreamTableEnvironment extends TableEnvironment {
 	 * @param path The path under which the {@link DataStream} is created.
 	 *             See also the {@link TableEnvironment} class description for the format of the path.
 	 * @param dataStream The {@link DataStream} out of which to create the view.
-	 * @param fields The field names of the created view.
+	 * @param fields The fields expressions to map original fields of the DataStream to the fields of the View.
 	 * @param <T> The type of the {@link DataStream}.
+	 * @deprecated use {@link #createTemporaryView(String, DataStream, Expression...)}
 	 */
+	@Deprecated
 	<T> void createTemporaryView(String path, DataStream<T> dataStream, String fields);
 
 	/**
 	 * Creates a view from the given {@link DataStream} in a given path with specified field names.
 	 * Registered views can be referenced in SQL queries.
 	 *
+	 * <p>There are two modes for mapping original fields to the fields of the View:
+	 *
+	 * <p>1. Reference input fields by name:
+	 * All fields in the schema definition are referenced by name
+	 * (and possibly renamed using an alias (as). Moreover, we can define proctime and rowtime
+	 * attributes at arbitrary positions using arbitrary names (except those that exist in the
+	 * result schema). In this mode, fields can be reordered and projected out. This mode can
+	 * be used for any input type, including POJOs.
+	 *
+	 * <p>Example:
+	 *
+	 * <pre>
+	 * {@code
+	 *   DataStream<Tuple2<String, Long>> stream = ...
+	 *   tableEnv.createTemporaryView(
+	 *      "cat.db.myTable",
+	 *      stream,
+	 *      $("f1"), // reorder and use the original field
+	 *      $("rowtime").rowtime(), // extract the internally attached timestamp into an event-time
+	 *                              // attribute named 'rowtime'
+	 *      $("f0").as("name") // reorder and give the original field a better name
+	 *   );
+	 * }
+	 * </pre>
+	 *
+	 * <p>2. Reference input fields by position:
+	 * In this mode, fields are simply renamed. Event-time attributes can
+	 * replace the field on their position in the input data (if it is of correct type) or be
+	 * appended at the end. Proctime attributes must be appended at the end. This mode can only be
+	 * used if the input type has a defined field order (tuple, case class, Row) and none of
+	 * the {@code fields} references a field of the input type.
+	 *
 	 * <p>Example:
 	 *
 	 * <pre>
 	 * {@code
 	 *   DataStream<Tuple2<String, Long>> stream = ...
-	 *   tableEnv.createTemporaryView("cat.db.myTable", stream, $("a"), $("b").as("name"), $("timestamp").rowtime())
+	 *   tableEnv.createTemporaryView(
+	 *      "cat.db.myTable",
+	 *      stream,
+	 *      $("a"), // rename the first field to 'a'
+	 *      $("b"), // rename the second field to 'b'
+	 *      $("rowtime").rowtime() // adds an event-time attribute named 'rowtime'
+	 *   );
 	 * }
 	 * </pre>
 	 *
@@ -344,7 +510,7 @@ public interface StreamTableEnvironment extends TableEnvironment {
 	 * @param path The path under which the {@link DataStream} is created.
 	 *             See also the {@link TableEnvironment} class description for the format of the path.
 	 * @param dataStream The {@link DataStream} out of which to create the view.
-	 * @param fields The field expressions of the created view.
+	 * @param fields The fields expressions to map original fields of the DataStream to the fields of the View.
 	 * @param <T> The type of the {@link DataStream}.
 	 */
 	<T> void createTemporaryView(String path, DataStream<T> dataStream, Expression... fields);

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

@@ -80,15 +80,43 @@ trait BatchTableEnvironment extends TableEnvironment {
   /**
     * Converts the given [[DataSet]] into a [[Table]] with specified field names.
     *
+    * There are two modes for mapping original fields to the fields of the [[Table]]:
+    *
+    * 1. Reference input fields by name:
+    * All fields in the schema definition are referenced by name
+    * (and possibly renamed using an alias (as). In this mode, fields can be reordered and
+    * projected out. This mode can be used for any input type, including POJOs.
+    *
+    * Example:
+    *
+    * {{{
+    *   val set: DataSet[(String, Long)] = ...
+    *   val table: Table = tableEnv.fromDataSet(
+    *      set,
+    *      $"_2", // reorder and use the original field
+    *      $"_1" as "name" // reorder and give the original field a better name
+    *   )
+    * }}}
+    *
+    * 2. Reference input fields by position:
+    * In this mode, fields are simply renamed. This mode can only be
+    * used if the input type has a defined field order (tuple, case class, Row) and none of
+    * the `fields` references a field of the input type.
+    *
     * Example:
     *
     * {{{
     *   val set: DataSet[(String, Long)] = ...
-    *   val tab: Table = tableEnv.fromDataSet(set, 'a, 'b)
+    *   val table: Table = tableEnv.fromDataSet(
+    *      set,
+    *      $"a", // renames the first field to 'a'
+    *      $"b" // renames the second field to 'b'
+    *   )
     * }}}
     *
     * @param dataSet The [[DataSet]] to be converted.
-    * @param fields The field names of the resulting [[Table]].
+    * @param fields The fields expressions to map original fields of the DataSet to the fields of
+    *               the [[Table]].
     * @tparam T The type of the [[DataSet]].
     * @return The converted [[Table]].
     */
@@ -137,11 +165,40 @@ trait BatchTableEnvironment extends TableEnvironment {
     * Creates a view from the given [[DataSet]] in a given path with specified field names.
     * Registered views can be referenced in SQL queries.
     *
+    * There are two modes for mapping original fields to the fields of the View:
+    *
+    * 1. Reference input fields by name:
+    * All fields in the schema definition are referenced by name
+    * (and possibly renamed using an alias (as). In this mode, fields can be reordered and
+    * projected out. This mode can be used for any input type, including POJOs.
+    *
     * Example:
     *
     * {{{
     *   val set: DataSet[(String, Long)] = ...
-    *   tableEnv.registerDataSet("myTable", set, 'a, 'b)
+    *   tableEnv.registerDataSet(
+    *      "myTable",
+    *      set,
+    *      $"_2", // reorder and use the original field
+    *      $"_1" as "name" // reorder and give the original field a better name
+    *   );
+    * }}}
+    *
+    * 2. Reference input fields by position:
+    * In this mode, fields are simply renamed. This mode can only be
+    * used if the input type has a defined field order (tuple, case class, Row) and none of
+    * the `fields` references a field of the input type.
+    *
+    * Example:
+    *
+    * {{{
+    *   val set: DataSet[(String, Long)] = ...
+    *   tableEnv.registerDataSet(
+    *      "myTable",
+    *      set,
+    *      $"a", // renames the first field to 'a'
+    *      $"b" // renames the second field to 'b'
+    *   )
     * }}}
     *
     * The view is registered in the namespace of the current catalog and database. To register the
@@ -153,7 +210,8 @@ trait BatchTableEnvironment extends TableEnvironment {
     *
     * @param name The name under which the [[DataSet]] is registered in the catalog.
     * @param dataSet The [[DataSet]] to register.
-    * @param fields The field names of the registered table.
+    * @param fields The fields expressions to map original fields of the DataSet to the fields of
+    *               the View.
     * @tparam T The type of the [[DataSet]] to register.
     * @deprecated use [[createTemporaryView]]
     */
@@ -164,11 +222,40 @@ trait BatchTableEnvironment extends TableEnvironment {
     * Creates a view from the given [[DataSet]] in a given path with specified field names.
     * Registered views can be referenced in SQL queries.
     *
+    * There are two modes for mapping original fields to the fields of the View:
+    *
+    * 1. Reference input fields by name:
+    * All fields in the schema definition are referenced by name
+    * (and possibly renamed using an alias (as). In this mode, fields can be reordered and
+    * projected out. This mode can be used for any input type, including POJOs.
+    *
+    * Example:
+    *
+    * {{{
+    *   val set: DataSet[(String, Long)] = ...
+    *   tableEnv.createTemporaryView(
+    *      "cat.db.myTable",
+    *      set,
+    *      $"_2", // reorder and use the original field
+    *      $"_1" as "name" // reorder and give the original field a better name
+    *   )
+    * }}}
+    *
+    * 2. Reference input fields by position:
+    * In this mode, fields are simply renamed. This mode can only be
+    * used if the input type has a defined field order (tuple, case class, Row) and none of
+    * the `fields` references a field of the input type.
+    *
     * Example:
     *
     * {{{
     *   val set: DataSet[(String, Long)] = ...
-    *   tableEnv.createTemporaryView("cat.db.myTable", set, 'a, 'b)
+    *   tableEnv.createTemporaryView(
+    *      "cat.db.myTable",
+    *      set,
+    *      $"a", // renames the first field to 'a'
+    *      $"b" // renames the second field to 'b'
+    *   )
     * }}}
     *
     * Temporary objects can shadow permanent ones. If a permanent object in a given path exists,
@@ -179,7 +266,8 @@ trait BatchTableEnvironment extends TableEnvironment {
     *             See also the [[TableEnvironment]] class description for the format of the
     *             path.
     * @param dataSet The [[DataSet]] out of which to create the view.
-    * @param fields The field names of the created view.
+    * @param fields The fields expressions to map original fields of the DataSet to the fields of
+    *               the View.
     * @tparam T The type of the [[DataSet]].
     */
   def createTemporaryView[T](path: String, dataSet: DataSet[T], fields: Expression*): Unit

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

@@ -98,15 +98,50 @@ trait StreamTableEnvironment extends TableEnvironment {
   /**
     * Converts the given [[DataStream]] into a [[Table]] with specified field names.
     *
+    * There are two modes for mapping original fields to the fields of the [[Table]]:
+    *
+    * 1. Reference input fields by name:
+    * All fields in the schema definition are referenced by name
+    * (and possibly renamed using an alias (as). Moreover, we can define proctime and rowtime
+    * attributes at arbitrary positions using arbitrary names (except those that exist in the
+    * result schema). In this mode, fields can be reordered and projected out. This mode can
+    * be used for any input type, including POJOs.
+    *
+    * Example:
+    *
+    * {{{
+    *   val stream: DataStream[(String, Long)] = ...
+    *   val table: Table = tableEnv.fromDataStream(
+    *      stream,
+    *      $"_2", // reorder and use the original field
+    *      $"rowtime".rowtime, // extract the internally attached timestamp into an event-time
+    *                          // attribute named 'rowtime'
+    *      $"_1" as "name" // reorder and give the original field a better name
+    *   )
+    * }}}
+    *
+    * <p>2. Reference input fields by position:
+    * In this mode, fields are simply renamed. Event-time attributes can
+    * replace the field on their position in the input data (if it is of correct type) or be
+    * appended at the end. Proctime attributes must be appended at the end. This mode can only be
+    * used if the input type has a defined field order (tuple, case class, Row) and none of
+    * the `fields` references a field of the input type.
+    *
     * Example:
     *
     * {{{
     *   val stream: DataStream[(String, Long)] = ...
-    *   val tab: Table = tableEnv.fromDataStream(stream, 'a, 'b)
+    *   val table: Table = tableEnv.fromDataStream(
+    *      stream,
+    *      $"a", // rename the first field to 'a'
+    *      $"b" // rename the second field to 'b'
+    *      $"rowtime".rowtime // extract the internally attached timestamp into an event-time attribute named 'rowtime'
+    *   )
     * }}}
     *
     * @param dataStream The [[DataStream]] to be converted.
-    * @param fields The field names of the resulting [[Table]].
+    * @param fields The fields expressions to map original fields of the DataStream to the fields of
+    *               the [[Table]].
     * @tparam T The type of the [[DataStream]].
     * @return The converted [[Table]].
     */
@@ -156,11 +191,47 @@ trait StreamTableEnvironment extends TableEnvironment {
     * Creates a view from the given [[DataStream]] in a given path with specified field names.
     * Registered views can be referenced in SQL queries.
     *
+    * There are two modes for mapping original fields to the fields of the View:
+    *
+    * 1. Reference input fields by name:
+    * All fields in the schema definition are referenced by name
+    * (and possibly renamed using an alias (as). Moreover, we can define proctime and rowtime
+    * attributes at arbitrary positions using arbitrary names (except those that exist in the
+    * result schema). In this mode, fields can be reordered and projected out. This mode can
+    * be used for any input type, including POJOs.
+    *
     * Example:
     *
     * {{{
     *   val stream: DataStream[(String, Long)] = ...
-    *   tableEnv.registerDataStream("myTable", stream, 'a, 'b)
+    *   tableEnv.registerDataStream(
+    *      "myTable",
+    *      stream,
+    *      $"_2", // reorder and use the original field
+    *      $"rowtime".rowtime, // extract the internally attached timestamp into an event-time
+    *                          // attribute named 'rowtime'
+    *      $"_1" as "name" // reorder and give the original field a better name
+    *   )
+    * }}}
+    *
+    * 2. Reference input fields by position:
+    * In this mode, fields are simply renamed. Event-time attributes can
+    * replace the field on their position in the input data (if it is of correct type) or be
+    * appended at the end. Proctime attributes must be appended at the end. This mode can only be
+    * used if the input type has a defined field order (tuple, case class, Row) and none of
+    * the `fields` references a field of the input type.
+    *
+    * Example:
+    *
+    * {{{
+    *   val stream: DataStream[(String, Long)] = ...
+    *   tableEnv.registerDataStream(
+    *      "myTable",
+    *      stream,
+    *      $"a", // rename the first field to 'a'
+    *      $"b" // rename the second field to 'b'
+    *      $"rowtime".rowtime // adds an event-time attribute named 'rowtime'
+    *   )
     * }}}
     *
     * The view is registered in the namespace of the current catalog and database. To register the
@@ -172,7 +243,8 @@ trait StreamTableEnvironment extends TableEnvironment {
     *
     * @param name The name under which the [[DataStream]] is registered in the catalog.
     * @param dataStream The [[DataStream]] to register.
-    * @param fields The field names of the registered view.
+    * @param fields The fields expressions to map original fields of the DataStream to the fields of
+    *               the View.
     * @tparam T The type of the [[DataStream]] to register.
     * @deprecated use [[createTemporaryView]]
     */
@@ -183,11 +255,47 @@ trait StreamTableEnvironment extends TableEnvironment {
     * Creates a view from the given [[DataStream]] in a given path with specified field names.
     * Registered views can be referenced in SQL queries.
     *
+    * There are two modes for mapping original fields to the fields of the View:
+    *
+    * 1. Reference input fields by name:
+    * All fields in the schema definition are referenced by name
+    * (and possibly renamed using an alias (as). Moreover, we can define proctime and rowtime
+    * attributes at arbitrary positions using arbitrary names (except those that exist in the
+    * result schema). In this mode, fields can be reordered and projected out. This mode can
+    * be used for any input type, including POJOs.
+    *
+    * Example:
+    *
+    * {{{
+    *   val stream: DataStream[(String, Long)] = ...
+    *   tableEnv.createTemporaryView(
+    *      "cat.db.myTable",
+    *      stream,
+    *      $"_2", // reorder and use the original field
+    *      $"rowtime".rowtime, // extract the internally attached timestamp into an event-time
+    *                          // attribute named 'rowtime'
+    *      $"_1" as "name" // reorder and give the original field a better name
+    *   )
+    * }}}
+    *
+    * 2. Reference input fields by position:
+    * In this mode, fields are simply renamed. Event-time attributes can
+    * replace the field on their position in the input data (if it is of correct type) or be
+    * appended at the end. Proctime attributes must be appended at the end. This mode can only be
+    * used if the input type has a defined field order (tuple, case class, Row) and none of
+    * the `fields` references a field of the input type.
+    *
     * Example:
     *
     * {{{
     *   val stream: DataStream[(String, Long)] = ...
-    *   tableEnv.createTemporaryView("cat.db.myTable", stream, 'a, 'b)
+    *   tableEnv.createTemporaryView(
+    *      "cat.db.myTable",
+    *      stream,
+    *      $"a", // rename the first field to 'a'
+    *      $"b" // rename the second field to 'b'
+    *      $"rowtime".rowtime // adds an event-time attribute named 'rowtime'
+    *   )
     * }}}
     *
     * Temporary objects can shadow permanent ones. If a permanent object in a given path exists,
@@ -197,7 +305,8 @@ trait StreamTableEnvironment extends TableEnvironment {
     * @param path The path under which the [[DataStream]] is created.
     *             See also the [[TableEnvironment]] class description for the format of the path.
     * @param dataStream The [[DataStream]] out of which to create the view.
-    * @param fields The field names of the created view.
+    * @param fields The fields expressions to map original fields of the DataStream to the fields of
+    *               the View.
     * @tparam T The type of the [[DataStream]].
     */
   def createTemporaryView[T](path: String, dataStream: DataStream[T], fields: Expression*): Unit