To test a Flink Table UDF I wrote a while ago I created this code to test it:
(Full link: https://github.com/nielsbasjes/yauaa/blob/v6.0/udfs/flink-table/src/test/java/nl/basjes/parse/useragent/flink/table/TestTableFunction.java#L80 ) // The base execution environment StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment(); // The table environment StreamTableEnvironment tableEnv = StreamTableEnvironment.create(senv); // The demo input stream DataStream<Tuple3<String, String, String>> inputStream = getTestAgentStream(senv); tableEnv.createTemporaryView( "AgentStream", inputStream, $("useragent"), $("expectedDeviceClass"), $("expectedAgentNameVersionMajor")); This method (with the array of Expression s) was deprecated in 1.14 so I'm looking to refactor it to the new way of doing the same. Essentially create a streaming table with a few columns on which I then apply my function using SQL and then verify the result. Looking through the Flink codebase I expected this would do the trick tableEnv.createTemporaryView( "AgentStream", inputStream, Schema .newBuilder() .column("useragent", DataTypes.STRING()) .column("expectedDeviceClass", DataTypes.STRING()) .column("expectedAgentNameVersionMajor", DataTypes.STRING()) .build() ); I was wrong as this call to createTemporaryView fails with org.apache.flink.table.api.ValidationException: Unable to find a field named 'useragent' in the physical data type derived from the given type information for schema declaration. Make sure that the type information is not a generic raw type. Currently available fields are: [f0, f1, f2] I suspect I'm doing something wrong regarding the mentioned "generic raw type" and the way I'm trying to define the Schema. What I essentially am looking for is the correct way to give the 3 provided columns a new name and type. How do I do this correctly in the new API? -- Best regards / Met vriendelijke groeten, Niels Basjes
