Hey fellow Flink devs, I have a use case with flink table API where there is a stream of JSONObject incoming (flattened one) and we are populating all values of that JSONObject into a single Row in a flink table. But the problem is with column names i.e field names. Here’s a sample code which works and am getting the output of the rows but I am not getting column names in the output as it’s a dynamic schema. (I am only getting f0 in the column name.) I want the column names too (which are keys of the JSONObject) as I want to use them to join with another table. Please help me out . Stuck on this for a while ! Appreciate the support in anticipation Flink version : 1.14.3

public class LookupFlinkApp implements Serializable {
  private static final String CONFIG_FILE = "/configuration.properties";
  private static ParameterTool applicationProperties;
  private static final StreamExecutionEnvironment STREAM_ENVIRONMENT =
      StreamExecutionEnvironment.getExecutionEnvironment();

private static final Logger logger = LoggerFactory.getLogger(LookupFlinkApp.class);
  static EnvironmentSettings settings = EnvironmentSettings
          .newInstance()
          .inStreamingMode()
          .build();
  static StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(STREAM_ENVIRONMENT, settings);

public static void main(String[] args) throws Exception {
    ParameterTool parameter;

try (InputStream in = LookupFlinkApp.class.getResourceAsStream(CONFIG_FILE)) {
      parameter = ParameterTool.fromPropertiesFile(in);
    }
    try (InputStream in =
                 LookupFlinkApp.class.getResourceAsStream(AppConstants.APPLICATION_PROPERTIES_YAML_FILE)) {
      applicationProperties = ParameterTool.fromPropertiesFile(in);
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
    Properties properties = parameter.getProperties();
    Properties appProperties = applicationProperties.getProperties();

KafkaSource<String> kafkaSource =
        KafkaSource.<String>builder()
            .setBootstrapServers(applicationProperties.get("input.bootstrap.servers"))
            .setTopics(applicationProperties.get("kafka.input.topic"))
            .setGroupId(applicationProperties.get("input.group.id"))
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .setStartingOffsets(OffsetsInitializer.latest())
            .build();

final SingleOutputStreamOperator<String> kafkaData =
        STREAM_ENVIRONMENT
            .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource")
            .setParallelism(1);

SingleOutputStreamOperator<FlowEntity> dataFromSource =
        kafkaData.process(new StringToFlowEntity()).setParallelism(1);

// Stream of JSONObjects
    DataStream<JSONObject> content = dataFromSource.map(new CustomMapFunction(properties));

// convert the content datastream into a table
    DataStream<Row> rows = content.flatMap(new BuildTableRowsFunc());

Table postingsTable = streamTableEnvironment.fromDataStream(rows);

postingsTable.printSchema();

Table result = postingsTable.select($("*"));

streamTableEnvironment.toDataStream(result, Row.class).map(new PrintFlinkTable(result));

STREAM_ENVIRONMENT.execute();
  }

public static class BuildTableRowsFunc implements FlatMapFunction<JSONObject, Row> {

@Override
    public void flatMap(JSONObject jsonObject, Collector<Row> collector) throws Exception {
      Set<String> keySet = jsonObject.keySet();
      Row row = new Row(keySet.size());
      int i = 0;
      for (String key : keySet) {
        row.setField(i++, jsonObject.get(key));
      }
      collector.collect(row);
    }
  }

public static class CustomMapFunction extends RichMapFunction<JSONObject, JSONObject> {
    private final Properties properties;

public CustomMapFunction(Properties properties) {
      this.properties = properties;
    }

@Override
    public JSONObject map(JSONObject jsonObject) throws Exception {
      jsonObject.put("tagId", "a1b2c3d4");
      return jsonObject;
    }
  }
}

Sample Output

| f0 |
| -- |
| +I[142, data.posting, dateTime, 2015-11-03T09:34:45, 201, 2023-04-30T12:48:15Z, 2016-11-30T20:59:59 application/json, Edm.Int64,_agency, a1b2c3d4e5]