class JsonReader implements DataReader<JsonReadOptions> {

  @Override
  public Table read(JsonReadOptions options) throws IOException {
    JsonNode jsonObj = mapper.readTree(options.source().createReader(null));
    if (options.path() != null) {
      jsonObj = jsonObj.at(options.path());
    }
    if (!jsonObj.isArray()) {
      throw new IllegalStateException(
          "Only reading a JSON array is currently supported. The array must hold an array or object for each row.");
    }
    if (jsonObj.size() == 0) {
      return Table.create(options.tableName());
    }

    JsonNode firstNode = jsonObj.get(0);
    if (firstNode.isArray()) {
      return convertArrayOfArrays(jsonObj, options);
    }
    return convertArrayOfObjects(jsonObj, options);
  }

}