@Override
    public Table getTable(String dbName, String tblName) {
        JDBCTableName jdbcTable = new JDBCTableName(null, dbName, tblName);
        return tableInstanceCache.get(jdbcTable,
                k -> {
                    try (Connection connection = getConnection()) {
                        ResultSet columnSet = schemaResolver.getColumns(connection, dbName, tblName);
                        List<Column> fullSchema = schemaResolver.convertToSRTable(columnSet);
                        List<Column> partitionColumns = Lists.newArrayList();
                        if (schemaResolver.isSupportPartitionInformation()) {
                            partitionColumns = listPartitionColumns(dbName, tblName, fullSchema);
                        }
                        if (fullSchema.isEmpty()) {
                            return null;
                        }

                        Integer tableId = tableIdCache.getPersistentCache(jdbcTable,
                                j -> ConnectorTableId.CONNECTOR_ID_GENERATOR.getNextId().asInt());
                        return schemaResolver.getTable(tableId, tblName, fullSchema,
                                partitionColumns, dbName, catalogName, properties);
                    } catch (SQLException | DdlException e) {
                        LOG.warn("get table for JDBC catalog fail!", e);
                        return null;
                    }
                });
    }
@Test
    public void testColumnTypes() {
        JDBCMetadata jdbcMetadata = new JDBCMetadata(properties, "catalog", dataSource);
        Table table = jdbcMetadata.getTable("test", "tbl1");
        List<Column> columns = table.getColumns();
        Assert.assertEquals(columns.size(), columnResult.getRowCount());
        Assert.assertTrue(columns.get(0).getType().equals(ScalarType.createType(PrimitiveType.INT)));
        Assert.assertTrue(columns.get(1).getType().equals(ScalarType.createUnifiedDecimalType(10, 2)));
        Assert.assertTrue(columns.get(2).getType().equals(ScalarType.createCharType(10)));
        Assert.assertTrue(columns.get(3).getType().equals(ScalarType.createVarcharType(10)));
        Assert.assertTrue(columns.get(4).getType().equals(ScalarType.createType(PrimitiveType.SMALLINT)));
        Assert.assertTrue(columns.get(5).getType().equals(ScalarType.createType(PrimitiveType.INT)));
        Assert.assertTrue(columns.get(6).getType().equals(ScalarType.createType(PrimitiveType.BIGINT)));
        Assert.assertTrue(columns.get(7).getType().equals(ScalarType.createType(PrimitiveType.LARGEINT)));
        Assert.assertTrue(columns.get(8).getType().equals(ScalarType.createType(PrimitiveType.TINYINT)));
        Assert.assertTrue(columns.get(9).getType().equals(ScalarType.createType(PrimitiveType.SMALLINT)));
        Assert.assertTrue(columns.get(10).getType().equals(ScalarType.createType(PrimitiveType.INT)));
        Assert.assertTrue(columns.get(11).getType().equals(ScalarType.createType(PrimitiveType.BIGINT)));
        Assert.assertTrue(columns.get(12).getType().equals(ScalarType.createType(PrimitiveType.DATE)));
        Assert.assertTrue(columns.get(13).getType().equals(ScalarType.createType(PrimitiveType.TIME)));
        Assert.assertTrue(columns.get(14).getType().equals(ScalarType.createType(PrimitiveType.DATETIME)));
    }