Feedback addressed

Author: twalthr

docs/content.zh/docs/dev/table/functions/ptfs.md

@@ -737,6 +737,11 @@ State TTL is applied individually to each entry in a list or map, allowing for f
 elements.
 {{< /hint >}}
 
+The following example demonstrates how to declare and use a `MapView`. It assumes the PTF processes a table with the
+schema `(userId, eventId, ...)`, partitioned by `userId`, with a high cardinality of distinct `eventId` values. For this
+use case, it is generally recommended to partition the table by both `userId` and `eventId`. For example purposes, the
+large state is stored as a map state.
+
 {{< tabs "1837eeed-3d13-455c-8e2f-5e164da9f844" >}}
 {{< tab "Java" >}}
 ```java

docs/content/docs/dev/table/functions/ptfs.md

@@ -737,6 +737,11 @@ State TTL is applied individually to each entry in a list or map, allowing for f
 elements.
 {{< /hint >}}
 
+The following example demonstrates how to declare and use a `MapView`. It assumes the PTF processes a table with the
+schema `(userId, eventId, ...)`, partitioned by `userId`, with a high cardinality of distinct `eventId` values. For this
+use case, it is generally recommended to partition the table by both `userId` and `eventId`. For example purposes, the
+large state is stored as a map state.
+
 {{< tabs "1837eeed-3d13-455c-8e2f-5e164da9f844" >}}
 {{< tab "Java" >}}
 ```java

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/dataview/ListView.java

@@ -39,8 +39,12 @@
  *
  * <p>For aggregating functions, the view can be used as a field in the accumulator of an {@link
  * AggregateFunction} or {@link TableAggregateFunction} when large amounts of data are expected.
+ * Aggregate functions might be used at various locations (pre-aggregation, combiners, merging of
+ * window slides, etc.) for some of these locations the data view is not backed by state but {@link
+ * ArrayList}.
  *
- * <p>For process table functions, the view can be used as a top-level state entry.
+ * <p>For process table functions, the view can be used as a top-level state entry. Data views in
+ * PTFs are always backed by state.
  *
  * <p>Note: Elements of a {@link ListView} must not be null. For heap-based state backends, {@code
  * hashCode/equals} of the original (i.e. external) class are used. However, the serialization

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/dataview/MapView.java

@@ -26,6 +26,7 @@
 import org.apache.flink.table.functions.TableAggregateFunction;
 import org.apache.flink.table.types.DataType;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
@@ -40,8 +41,12 @@
  *
  * <p>For aggregating functions, the view can be used as a field in the accumulator of an {@link
  * AggregateFunction} or {@link TableAggregateFunction} when large amounts of data are expected.
+ * Aggregate functions might be used at various locations (pre-aggregation, combiners, merging of
+ * window slides, etc.) for some of these locations the data view is not backed by state but {@link
+ * ArrayList}.
  *
- * <p>For process table functions, the view can be used as a top-level state entry.
+ * <p>For process table functions, the view can be used as a top-level state entry. Data views in
+ * PTFs are always backed by state.
  *
  * <p>Note: Keys of a {@link MapView} must not be null. Nulls in values are supported. For
  * heap-based state backends, {@code hashCode/equals} of the original (i.e. external) class are
@@ -119,8 +124,9 @@ public void setMap(Map<K, V> map) {
     /**
      * Return the value for the specified key or {@code null} if the key is not in the map view.
      *
-     * @param key The lookup key.
-     * @return The value for the specified key.
+     * @param key The key whose associated value is to be returned
+     * @return The value to which the specified key is mapped, or {@code null} if this map contains
+     *     no mapping for the key
      * @throws Exception Thrown if the system cannot get data.
      */
     public V get(K key) throws Exception {

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala

@@ -316,7 +316,7 @@ object ProcessTableRunnerGenerator {
     val constructorCode = stateDataType.getConversionClass match {
       case rowType if rowType == classOf[Row] =>
         // This allows us to retrieve the converter term that has been generated
-        // in genToExternalConverter(). The converter is able to created named positions
+        // in genToExternalConverter(). The converter is able to create named positions
         // for row fields.
         val converterTerm = ctx.addReusableConverter(stateDataType)
         s"((${className[RowRowConverter]}) $converterTerm).createEmptyRow()"

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/dataview/DataViewUtils.java

@@ -77,7 +77,8 @@ public static boolean isDataView(LogicalType viewType, Class<? extends DataView>
     /** Checks that the given type and its children do not contain data views. */
     public static void checkForInvalidDataViews(LogicalType type) {
         if (hasNested(type, t -> isDataView(t, DataView.class))) {
-            throw new ValidationException("Data views are not supported at the declared location.");
+            throw new ValidationException(
+                    "Data views are not supported at the declared location. Given type: " + type);
         }
     }