JdbcDriver extends Driver { @Override public Connection connect(String url, Properties info) throws SQLException { return JdbcConnection.initialize((CalciteConnection) super.connect(url, info)); }  @Override Connection connect(String url, Properties info); static JdbcConnection connect(TableProvider tableProvider, PipelineOptions options); static final JdbcDriver INSTANCE; static final String CONNECT_STRING_PREFIX; static final ObjectMapper OBJECT_MAPPER; }
@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"))); }
@Test public void testTimestampWithDefaultTimezone() throws Exception { TestTableProvider tableProvider = new TestTableProvider(); Connection connection = JdbcDriver.connect(tableProvider, PipelineOptionsFactory.create()); Schema schema = Schema.builder().addDateTimeField("ts").build(); connection .createStatement() .executeUpdate("CREATE EXTERNAL TABLE test (ts TIMESTAMP) TYPE 'test'"); ReadableInstant july1 = ISODateTimeFormat.dateTimeParser().parseDateTime("2018-07-01T01:02:03Z"); tableProvider.addRows("test", Row.withSchema(schema).addValue(july1).build()); ResultSet selectResult = connection.createStatement().executeQuery(String.format("SELECT ts FROM test")); selectResult.next(); Timestamp ts = selectResult.getTimestamp(1); assertThat( String.format( "Wrote %s to a table, but got back %s", ISODateTimeFormat.basicDateTime().print(july1), ISODateTimeFormat.basicDateTime().print(ts.getTime())), ts.getTime(), equalTo(july1.getMillis())); }
@Test @Ignore("https: public void testTimestampWithNonzeroTimezone() throws Exception { Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("Asia/Tokyo"), Locale.ROOT); TestTableProvider tableProvider = new TestTableProvider(); Connection connection = JdbcDriver.connect(tableProvider, PipelineOptionsFactory.create()); Schema schema = Schema.builder().addDateTimeField("ts").build(); connection .createStatement() .executeUpdate("CREATE EXTERNAL TABLE test (ts TIMESTAMP) TYPE 'test'"); ReadableInstant july1 = ISODateTimeFormat.dateTimeParser().parseDateTime("2018-07-01T01:02:03Z"); tableProvider.addRows("test", Row.withSchema(schema).addValue(july1).build()); ResultSet selectResult = connection.createStatement().executeQuery(String.format("SELECT ts FROM test")); selectResult.next(); Timestamp ts = selectResult.getTimestamp(1, cal); assertThat( String.format( "Wrote %s to a table, but got back %s", ISODateTimeFormat.basicDateTime().print(july1), ISODateTimeFormat.basicDateTime().print(ts.getTime())), ts.getTime(), equalTo(july1.getMillis())); }
@Test public void testTimestampWithZeroTimezone() throws Exception { Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("UTC"), Locale.ROOT); TestTableProvider tableProvider = new TestTableProvider(); Connection connection = JdbcDriver.connect(tableProvider, PipelineOptionsFactory.create()); Schema schema = Schema.builder().addDateTimeField("ts").build(); connection .createStatement() .executeUpdate("CREATE EXTERNAL TABLE test (ts TIMESTAMP) TYPE 'test'"); ReadableInstant july1 = ISODateTimeFormat.dateTimeParser().parseDateTime("2018-07-01T01:02:03Z"); tableProvider.addRows("test", Row.withSchema(schema).addValue(july1).build()); ResultSet selectResult = connection.createStatement().executeQuery(String.format("SELECT ts FROM test")); selectResult.next(); Timestamp ts = selectResult.getTimestamp(1, cal); assertThat( String.format( "Wrote %s to a table, but got back %s", ISODateTimeFormat.basicDateTime().print(july1), ISODateTimeFormat.basicDateTime().print(ts.getTime())), ts.getTime(), equalTo(july1.getMillis())); }
@Test public void testSelectsFromExistingComplexTable() throws Exception { TestTableProvider tableProvider = new TestTableProvider(); Connection connection = JdbcDriver.connect(tableProvider, PipelineOptionsFactory.create()); connection .createStatement() .executeUpdate( "CREATE EXTERNAL TABLE person ( \n" + "description VARCHAR, \n" + "nestedRow ROW< \n" + " id BIGINT, \n" + " name VARCHAR> \n" + ") \n" + "TYPE 'test'"); tableProvider.addRows( "person", row(COMPLEX_SCHEMA, "description1", row(1L, "aaa")), row(COMPLEX_SCHEMA, "description2", row(2L, "bbb"))); ResultSet selectResult = connection .createStatement() .executeQuery("SELECT person.nestedRow.id, person.nestedRow.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"))); }
@Test public void testInsertIntoCreatedTable() 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'"); connection .createStatement() .executeUpdate("CREATE EXTERNAL TABLE person_src (id BIGINT, name VARCHAR) TYPE 'test'"); tableProvider.addRows("person_src", row(1L, "aaa"), row(2L, "bbb")); connection.createStatement().execute("INSERT INTO person SELECT id, name FROM person_src"); ResultSet selectResult = connection.createStatement().executeQuery("SELECT id, name FROM person"); List<Row> resultRows = readResultSet(selectResult).stream() .map(resultValues -> resultValues.stream().collect(toRow(BASIC_SCHEMA))) .collect(Collectors.toList()); assertThat(resultRows, containsInAnyOrder(row(1L, "aaa"), row(2L, "bbb"))); }
@Test public void testInternalConnect_boundedTable() throws Exception { CalciteConnection connection = JdbcDriver.connect(BOUNDED_TABLE, PipelineOptionsFactory.create()); Statement statement = connection.createStatement(); ResultSet resultSet = statement.executeQuery("SELECT * FROM test"); assertTrue(resultSet.next()); assertEquals(1, resultSet.getInt("id")); assertEquals("first", resultSet.getString("name")); assertFalse(resultSet.next()); }
@Test public void testInternalConnect_bounded_limit() throws Exception { ReadOnlyTableProvider tableProvider = new ReadOnlyTableProvider( "test", ImmutableMap.of( "test", TestBoundedTable.of( Schema.FieldType.INT32, "id", Schema.FieldType.STRING, "name") .addRows(1, "first") .addRows(1, "second first") .addRows(2, "second"))); CalciteConnection connection = JdbcDriver.connect(tableProvider, PipelineOptionsFactory.create()); Statement statement = connection.createStatement(); ResultSet resultSet1 = statement.executeQuery("SELECT * FROM test LIMIT 5"); assertTrue(resultSet1.next()); assertTrue(resultSet1.next()); assertTrue(resultSet1.next()); assertFalse(resultSet1.next()); assertFalse(resultSet1.next()); ResultSet resultSet2 = statement.executeQuery("SELECT * FROM test LIMIT 1"); assertTrue(resultSet2.next()); assertFalse(resultSet2.next()); ResultSet resultSet3 = statement.executeQuery("SELECT * FROM test LIMIT 2"); assertTrue(resultSet3.next()); assertTrue(resultSet3.next()); assertFalse(resultSet3.next()); ResultSet resultSet4 = statement.executeQuery("SELECT * FROM test LIMIT 3"); assertTrue(resultSet4.next()); assertTrue(resultSet4.next()); assertTrue(resultSet4.next()); assertFalse(resultSet4.next()); }
@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()); }
@Test public void testInternalConnect_setDirectRunner() throws Exception { CalciteConnection connection = JdbcDriver.connect(BOUNDED_TABLE, PipelineOptionsFactory.create()); Statement statement = connection.createStatement(); assertEquals(0, statement.executeUpdate("SET runner = direct")); assertTrue(statement.execute("SELECT * FROM test")); }
@Test public void testInternalConnect_setBogusRunner() throws Exception { thrown.expectMessage("Unknown 'runner' specified 'bogus'"); CalciteConnection connection = JdbcDriver.connect(BOUNDED_TABLE, PipelineOptionsFactory.create()); Statement statement = connection.createStatement(); assertEquals(0, statement.executeUpdate("SET runner = bogus")); assertTrue(statement.execute("SELECT * FROM test")); }
@Test public void testInternalConnect_resetAll() throws Exception { CalciteConnection connection = JdbcDriver.connect(BOUNDED_TABLE, PipelineOptionsFactory.create()); Statement statement = connection.createStatement(); assertEquals(0, statement.executeUpdate("SET runner = bogus")); assertEquals(0, statement.executeUpdate("RESET ALL")); assertTrue(statement.execute("SELECT * FROM test")); }