如何将JSON模式转换为Flink JavaScript语句？
例如，给定以下具有嵌套对象属性的JSON模式

{
  "type": "object",
  "properties": {
    "name": {
      "type": "string"
    },
    "id": {
      "type": "string"
    },
    "i": {
      "type": "integer"
    },
    "p": {
      "type": "object",
      "properties": {
        "name": {
          "type": "string"
        },
        "id": {
          "type": "string"
        },
        "i": {
          "type": "integer"
        },
        "p1": {
          "type": "object",
          "properties": {
            "name": {
              "type": "string"
            },
            "id": {
              "type": "string"
            },
            "i": {
              "type": "integer"
            }
          }
        }
      }
    }
  }
}

我希望将其转换为以下使用行类型的Flink SQL语句

CREATE TABLE test_nested (
    name STRING,
    id STRING,
    i INTEGER,
    p ROW(name STRING, id STRING, i INTEGER, p1 ROW(name STRING, id STRING, i INTEGER))
) WITH (
    'connector'='print'
);

我尝试遍历properties字段，但每当遇到object类型时，我都不知道该怎么办。我的猜测是我需要递归地做，但我不知道确切的步骤。

private static String traverse(JsonNode rootNode) {
    JsonNode properties = rootNode.get("properties");
    Iterator<Map.Entry<String, JsonNode>> fields = properties.fields();
    // columnNameType will contain the "columnName dataType", for example
    // "name STRING"
    // "id STRING"
    // "i INTEGER"
    // "p ROW(name STRING, id STRING, i INTEGER, p1 ROW(name STRING, id STRING, i INTEGER))"
    List<String> columnNameType = new ArrayList<>();
    while (fields.hasNext()) {
        Map.Entry<String, JsonNode> next = fields.next();
        String key = next.getKey();
        JsonNode value = next.getValue();
        String type = value.get("type").asText();
        if(type.equals("object")) {
            // recurse???
        } else {
            columnNameType.add(key+" "+type);
        }
    }
    // this should return a formatted string like
    /*
        name STRING,
        id STRING,
        i INTEGER,
        p ROW(name STRING, id STRING, i INTEGER, p1 ROW(name STRING, id STRING, i INTEGER))
    */
    return String.join(",\n",  columnNameType);
}