@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 testInternalConnect_unbounded_limit() throws Exception {
    ReadOnlyTableProvider tableProvider =
        new ReadOnlyTableProvider(
            "test",
            ImmutableMap.of(
                "test",
                TestUnboundedTable.of(
                        Schema.FieldType.INT32, "order_id",
                        Schema.FieldType.INT32, "site_id",
                        Schema.FieldType.INT32, "price",
                        Schema.FieldType.DATETIME, "order_time")
                    .timestampColumnIndex(3)
                    .addRows(Duration.ZERO, 1, 1, 1, FIRST_DATE, 1, 2, 6, FIRST_DATE)));

    CalciteConnection connection =
        JdbcDriver.connect(tableProvider, PipelineOptionsFactory.create());
    Statement statement = connection.createStatement();

    ResultSet resultSet1 = statement.executeQuery("SELECT * FROM test LIMIT 1");
    assertTrue(resultSet1.next());
    assertFalse(resultSet1.next());

    ResultSet resultSet2 = statement.executeQuery("SELECT * FROM test LIMIT 2");
    assertTrue(resultSet2.next());
    assertTrue(resultSet2.next());
    assertFalse(resultSet2.next());
  }