Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5132#discussion_r159757312
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
---
@@ -768,6 +768,39 @@ abstract class TableEnvironment(val config:
TableConfig) {
frameworkConfig
}
+ /**
+ * 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. Moreover, we can define proctime and rowtime
attributes at arbitrary
+ * positions using arbitrary names (except those that exist in the
result schema). This mode
+ * can be used for any input type, including POJOs.
+ *
+ * Reference input fields by position:
+ * Field references must refer to existing fields in the input type
(except for
+ * renaming with alias (as)). 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 no of fields
+ * references a field of the input type.
+ */
+ protected def isReferenceByPosition(t: TypeInformation[_], fields:
Array[Expression]): Boolean = {
+ if (t.isInstanceOf[PojoTypeInfo[_]]) {
+ return false
+ }
+
+ val inputNames = t match {
+ case ct: CompositeType[_] => ct.getFieldNames
+ case _ => return false // atomic types are references by name
--- End diff --
If atomic types are referenced by name, what's the name? Atomic types are
neither referenced by position or name. Instead we can reference the field
because there is only one field in the input.
Should we make this method only available for `CompositeType` by changing
the type of `t` to `CompositeType`?
---
