@Override
  public Connection connect(String url, Properties info) throws SQLException {
    // calciteConnection is initialized with an empty Beam schema,
    // we need to populate it with pipeline options, load table providers, etc
    return JdbcConnection.initialize((CalciteConnection) super.connect(url, info));
  }
@Test
  public void testSelectsFromExistingTable() throws Exception {
    TestTableProvider tableProvider = new TestTableProvider();
    Connection connection = JdbcDriver.connect(tableProvider, PipelineOptionsFactory.create());

    connection
        .createStatement()
        .executeUpdate("CREATE EXTERNAL TABLE person (id BIGINT, name VARCHAR) TYPE 'test'");

    tableProvider.addRows("person", row(1L, "aaa"), row(2L, "bbb"));

    ResultSet selectResult =
        connection.createStatement().executeQuery("SELECT id, name FROM person");

    List<Row> resultRows =
        readResultSet(selectResult).stream()
            .map(values -> values.stream().collect(toRow(BASIC_SCHEMA)))
            .collect(Collectors.toList());

    assertThat(resultRows, containsInAnyOrder(row(1L, "aaa"), row(2L, "bbb")));
  }