Discussion thread | - | ||||||||
---|---|---|---|---|---|---|---|---|---|
Vote thread | - | ||||||||
JIRA | - | ||||||||
https://lists.apache.org/thread/d1owrg8zh77v0xygcpb93fxt0jpjdkb3 | |||||||||
Vote thread | https://lists.apache.org/thread/7jbmg22lnww31sbfdzztwrzgm6bkhjrj | ||||||||
JIRA |
| ||||||||
Release | 1.18.0 | Release | - |
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
Flink Jdbc Driver is used to submit queries to SqlGateway. There are old repositories The old repositories flink-sql-gateway[1] and flink-jdbc-driver[2] which support submitting query to the older version of flink cluster version. But for the latest flink version cluster there are many compatibility problems. Flink Sql Gateway has been a submodule in Flink, and the API has been greatly optimized. The old Flink JDBC Driver cannot connect to the new Gateway directly which will cause we users cannot upgrade our their Flink versionClusters. In this FLIP, Iwe'd like to introduce Flink Jdbc Driver module in Flink to connect the new Gateway, users can use Jdbc Driver to submit their queries and get results like a database in their applications.
...
draw.io Board Diagram | ||||||||||||||||||||
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
There are 6 7 main classes in Flink Jdbc Driver: FlinkDriver, FlinkDataSource, FlinkConnection, FlinkStatement, FlinkResultSet, FlinkDatabaseMetaData and FlinkResultSetMetaData which implement jdbc interface Driver, DataSource, Connection, Statement, ResultSet, DatabaseMetaData and ResultSetMetaData.
- FlinkDriver parses gateway address from url, and creates FlinkConnection
- FlinkDataSource manages connection pool for flink jdbc, it will create specific count of connections and choose one for client directly.
- FlinkConnection creates Executor according to gateway address. When the Connections is closed, it can close the connection with gateway by Executor
- FlinkStatement can get Executor from FlinkConnection, and submit sql query to it. After query is executed, FlinkStatement can get StatementResult from Executor, and create FlinkResultSet
- FlinkResultSet is an iterator, it gets results from StatementResult and return them to users
- FlinkDatabaseMetaData provides meta data of catalogs, databases and tables
- FlinkResultSetMetaData provides meta data of ResultSet such as columns
...
draw.io Board Diagram | ||||||||||||||||||||
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
User Case
Flink Jdbc Driver module will be packaged into an independent jar file such as flink-table-jdbc-driver-{version}.jar
, which will contains classes of flink jdbc driver and shaded flink classes such as data type. User only need to add jdbc dependency in pom.xml or add the jar in the classpath of external Jdbc Tools such as sqlite.
Code Block | ||
---|---|---|
| ||
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-jdbc-driver</artifactId>
<version>${flink.version}</version>
</dependency> |
More information about Flink Jdbc Driver
|
There're Session
s and Operation
s in SqlGateway
. SqlGateway
will open a Session
for each FlinkConnection
, and then do multiple Operation
s in one Session
. When users create a FlinkConnection
by FlinkDriver
with SqlGateway
, it will open an exist or a new Session
. Any time users want to issue SQL statements to the database, they require a FlinkStatement
instance from FlinkConnection
. Once users have a FlinkStatement
, they can use issue a query. This will return a FlinkResultSet
instance, which contains the entire result. Each operation such as the execution query(Flink job), fetching results in FlinkResultSet
will be an Operation
in the Session
of SqlGateway
.
User Case
Flink Jdbc Driver module will be packaged into an independent jar file such as flink-table-jdbc-driver-{version}.jar
, which will contains classes of flink jdbc driver and shaded flink classes such as data type. User only need to add jdbc dependency in pom.xml or add the jar in the classpath of external Jdbc Tools such as sqlite.
Code Block | ||
---|---|---|
| ||
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-jdbc-driver</artifactId>
<version>${flink.version}</version>
</dependency> |
More information about Flink Jdbc Driver
- Driver Name: org.apache.flink.table.jdbc.FlinkDriver
- Flink Connection URL: Users can use default catalog and database directly, or set them in url. Users can also set custom parameters in url to open the session in SqlGateway with key1=val1&key2=val2&...
- Driver Name: org.apache.flink.table.jdbc.FlinkDriver
- Flink Connection URL: Users can use default catalog and database directly, or set them in url. Users can also set custom parameters in url to open the session in SqlGateway with key1=val1&key2=val2&...
Use Flink Jdbc Driver in Java code
Code Block String driverName = "org.apache.flink.table.jdbc.FlinkDriver"; String url = "jdbclanguage java - jdbc:flink://{sql-gateway.host}:{sql-gateway.port}?key1=val1&key2=val2 jdbc:flink://{sql-gateway.host}:{sql-gateway.port}/{catalog name}?key1=val1&key2=val2
- jdbc:flink://{sql-gateway.host}:{sql-gateway.port}/{catalog name}/{database name}?key1=val1&key2=val2
- Currently SqlGateway does not support authentication,
user
andpassword
in connection are invalid. - jdbc:flink://{sql-gateway.host}:{sql-gateway.port}/{catalog
- name}?key1=val1&key2=val2
- jdbc:flink://{sql-gateway.host}:{sql-gateway.port}/{catalog name}/{database name}?key1=val1&key2=val2
- Currently SqlGateway does not support authentication,
user
andpassword
in connection are invalid. Use Flink Jdbc Driver in Java code
Code Block language java String driverName = "org.apache.flink.table.jdbc.FlinkDriver"; String url = "jdbc:flink://{sql-gateway.host}:{sql-gateway.port}/{catalog name}/{database name}?key1=val1&key2=val2"; Class.forName(driverName); try (Connection connection = DriverManager.getConnection(url)) { try (Statement statement = connection.createStatement"; Class.forName(driverName); try (Connection connection = DriverManager.getConnection(url)) { try (Statement statement = connection.createStatement()) { try (ResultSet resultSet = statement.execute("SELECT * FROM {Your Table}")) { while (resultSet.hasNext()) { try (ResultSet resultSet = statement.execute("SELECT * FROM {Your Table}")) { while (resultSet.hasNext()) { Do your work ... } } try (ResultSet resultSet } }
- User can also add the
flink-table-jdbc-driver-{version}.jar
to the classpath of external jdbc tools.
Data Types
We support to convert the following flink data type to sql data type in this FLIP, and more types can be supported as needed in the future.
= statement.execute("SELECT * FROM T1 JOIN T2 on T1.id=T2.id ...")) { while (resultSet.hasNext()) { Do your work ... } } } }
- User can also add the
flink-table-jdbc-driver-{version}.jar
to the classpath of external jdbc tools.
Data Types
The following basic flink data types are supported to convert to sql data type in this FLIP, and more types can be supported as needed in the future.
Flink Data Type | Java Sql Data Type | Java Data Type |
CharType/VarCharType | CHAR/VARCHAR | String |
BooleanType | BOOLEAN | Boolean |
TinyIntType | TINYINT | Byte |
SmallIntType | SMALLINT | Short |
IntType | INTEGER | Int |
BigIntType | BIGINT | Long |
FloatType | FLOAT | Float |
DoubleType | DOUBLE | Double |
DecimalType | DECIMAL |
Flink Data Type
Jdbc Sql Data Type
CharType/VarCharType
String
BooleanType
Boolean
TinyIntType
Byte
SmallIntType
Short
IntType
Int
BigIntType
Long
FloatType
Float
DoubleType
Double
BigDecimal | |
BinaryType/VarBinaryType |
BINARY/VARBINARY | byte[] |
DateType | DATE | Date |
TimeType | TIME | Time |
TimestampType | TIMESTAMP | Timestamp |
ArrayType
Array
Public Interface
There are many methods in Jdbc Driver, while this FLIP only implement the basic methods first and more methods will be implemented later when they are needed.
- Methods in
FlinkDriver
Code Block | ||
---|---|---|
| ||
/* Jdbc Driver for flink sql gateway */
public class FlinkDriver implements Driver {
/* Connect sql gateway with given url and open/create session with given priperties. */
@Override
public Connection connect(String url, Properties info) throws SQLException;
} |
- Methods in
FlinkConnection
ZonedTimestampType | TIMESTAMP_WITH_TIMEZONE | OffsetDateTime |
LocalZonedTimestampType | TIMESTAMP_WITH_LOCAL_TIMEZONE | Timestamp |
ArrayType | ARRAY | Array |
RowType | ROW(Not in java.sql.Types) | Row(Flink Row Data) |
MapType | MAP | Map<K, V> |
Currently TIMESTAMP_WITH_LOCAL_TIMEZONE is not exist in java.sql.Types, but it is supported by Flink
. Users can define a field as (f TIMESTAMP(p) WITH LOCAL TIME ZONE)
and set time zone through the connection parameters or dynamic parameters in console by table.local-time-zone
. After that, users can get Timestamp
which will be automatically converted from stored time data into specific Timestamp
according to the given time zone.
Java Sql Interfaces
There are many methods in Jdbc Driver, while this FLIP only implement the basic methods first and more methods will be implemented later when they are needed.
- Methods in
FlinkDriver
Code Block | ||
---|---|---|
| ||
/* Jdbc Driver for flink sql gateway. Only Batch Mode queries are supported. If you force to submit streaming queries, you may get unrecognized updates, deletions and other results in FlinkResultSet. */
public class FlinkDriver implements Driver {
/* Connect sql gateway with given url and open/create session with given priperties. */
@Override | ||
Code Block | ||
| ||
/* Connection to flink sql gateway for jdbc driver. */ public class FlinkConnection implements Connection { /* Create statement from connection. */ @Override public Statement createStatement() throws SQLException; /* Close session in sql gateway. */ @Override public void close() throws SQLException; /* Use given catalog to the session in sql gateway. */ @Override public voidConnection setCatalogconnect(String catalogurl, Properties info) throws SQLException; } |
- Methods in FlinkDataSource
Code Block | ||
---|---|---|
| ||
/* Jdbc DataSource manages connections for client, /* Get current catalog name from sessionwe can support more operations in it in the future. */ public @Overrideclass FlinkDataSource implements DataSource { /* publicThe Stringmax getCatalog()count throwsof SQLException; connections which the data source holds. */ private int maxActive; /* GetSet FlinkDatabaseMetaDatathe instanceurl for the current catalogof connection. */ @Overridepublic synchronized void public DatabaseMetaData getMetaData() throws SQLException; setUrl(String url); /* UseSet giventhe databasedriver toclass thename sessionfor inthe sql gatewaysource. */ @Override publicpublic synchronized void setSchemasetDriverClassName(String schemadriverClassName) throws; SQLException; /* Get current database name from session /* Set the max active connection for the source. */ @Override public synchronized publicvoid String getSchema() throws SQLException; } |
- Methods in
FlinkStatement
Code Block | ||
---|---|---|
| ||
/* Statement in flink jdbc driversetMaxActive(int maxActive); /* Get a connection from data source. */ @Override public class FlinkStatement implements Statementpublic { Connection getConnection() throws SQLException; @Override public Connection getConnection(String username, String password) throws SQLException; } |
- Methods in
FlinkConnection
Code Block | ||
---|---|---|
| ||
/* Connection to flink sql gateway for jdbc driver. */ public class FlinkConnection implements Connection { /* Submit sql to sql gateway and get result set. */ @Override public ResultSet executeQuery(String sql) throws SQLException; /* ExecuteCreate givenstatement update sql and return result countfrom connection. */ @Override public intStatement executeUpdatecreateStatement(String sql) throws SQLException; /* Cancel the running jobClose session in sql gateway. */ @Override public void close() throws SQLException; /* Use Returngiven truecatalog ifto the resultsession setin hassql more resultsgateway. */ @Override public booleanvoid getMoreResultssetCatalog(String catalog) throws SQLException; /* Get current resultcatalog setname infrom the statementsession. */ @Override public ResultSetString getResultSetgetCatalog() throws SQLException; } |
- Methods in
FlinkResultSet
:FlinkResultSet
only supports fetching data from iteratorStatementResult
, it supports getXXX methods and doesn't support deleting, updating or moving the cursor.
Code Block | ||
---|---|---|
| ||
/* ResultSet for flink jdbc driver. */ public class FlinkResultSet implements ResultSet { /* Return true if there are more resuts in result iterator /* Get FlinkDatabaseMetaData instance for the current catalog. */ @Override public booleanDatabaseMetaData nextgetMetaData() throws SQLException; /* Close Use given database to the session fetchin resultsql operationgateway. */ @Override public void closesetSchema(String schema) throws SQLException; /* Get differentcurrent valuesdatabase accordingname to data type and column indexfrom session. */ @Override public <V>String V getXXX(int columnIndexgetSchema() throws SQLException. ; } |
- Methods in
FlinkStatement
Code Block | ||
---|---|---|
| ||
/* Statement in flink jdbc driver. */ public class FlinkStatement implements Statement { /* Submit sql to sql gateway and get result set/* Get different values according to data type and column name. */ @Override public <V>ResultSet V getXXXexecuteQuery(String columnNamesql) throws SQLException. } |
- Methods in
FlinkDatabaseMetaData
:FlinkDatabaseMetaData
only supports TABLE and VIEW tables, getting information of catalog, database and tables.
Code Block | ||
---|---|---|
| ||
/* DatabaseMetaData in flink sql driver; /* Execute given update sql and return result count. */ public class @Override FlinkDatabaseMetaData implements DatabaseMetaData { public int /* Get the url of flink sql driverexecuteUpdate(String sql) throws SQLException; /* Cancel the running job in sql gateway. */ @Override public Stringvoid getURLclose() throws SQLException; /* Get catalog name list from session Return true if the result set has more results. */ @Override public ResultSetboolean getCatalogsgetMoreResults() throws SQLException; /* Get current databaseresult nameset listin fromthe sessionstatement. */ @Override public ResultSet getSchemasgetResultSet() throws SQLException; } |
- Methods in
FlinkResultSet
:FlinkResultSet
only supports fetching data from iteratorStatementResult
, it supports getXXX methods and doesn't support deleting, updating or moving the cursor. Compare withResultSet
, there isgetKind
method inFlinkResultSet
to get theRowKind
of current record.
Code Block | ||
---|---|---|
| ||
/* ResultSet for /* Get database name lins in given catalog from sessionflink jdbc driver. Only Batch Mode queries are supported. If you force to submit streaming queries, you may get unrecognized updates, deletions and other results. */ public @Overrideclass FlinkResultSet implements ResultSet { public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLException; /* Return Gettrue tableif namethere listare withmore givenresuts conditionin fromresult sessioniterator. */ @Override public ResultSetboolean getTablesnext(String) catalog, String schemaPattern, String tableNamePattern, String[] types) throws SQLException; throws SQLException; /* GetClose columnthe listfetch with given condition from sessionresult operation. */ @Override public ResultSetvoid getColumns(String catalog, String schemaPattern, String tableNamePattern, String columnNamePatternclose() throws SQLException; /* Get primarydifferent keyvalues listaccording forto givendata type tableand fromcolumn sessionindex. */ @Override public <V> ResultSetV getPrimaryKeysgetXXX(String catalog, String schema, String tableint columnIndex) throws SQLException; } |
- Methods in
FlinkResultSetMetaData
:FlinkResultSetMetaData
only supports getting column information according to column index or name.
Code Block | ||
---|---|---|
| ||
/* ResultSetMetaData in flink sql driver. */ public class FlinkResultSetMetaData implements ResultSetMetaData { /* Get column count in the result set. /* Get different values according to data type and column name. */ @Override public <V> intV getColumnCountgetXXX(String columnName) throws SQLException. } |
- Methods in
FlinkDatabaseMetaData
:FlinkDatabaseMetaData
only supports TABLE and VIEW tables, getting information of catalog, database and tables.
Code Block | ||
---|---|---|
| ||
/* DatabaseMetaData in flink sql driver; /* If the column may be null. */ public @Overrideclass FlinkDatabaseMetaData implements DatabaseMetaData { public int isNullable(int column) throws SQLException; /* Get display size for the column /* Get the url of flink sql driver. */ @Override public intString getColumnDisplaySizegetURL(int column) throws SQLException; /* Get columncatalog name accordinglist tofrom column indexsession. */ @Override public StringResultSet getColumnLabelgetCatalogs(int column) throws SQLException; public String getColumnName(int column) throws SQLException; /* Get precision for the column index/* Get database name list from session. */ @Override public intResultSet getPrecisiongetSchemas(int column) throws SQLException; /* Get database columnname typelins idin forgiven thecatalog columnfrom indexsession. */ @Override public intResultSet getColumnType(int columngetSchemas(String catalog, String schemaPattern) throws SQLException; /* Get columntable type name for the column indexlist with given condition from session. */ @Override public ResultSet getTables(String catalog, String schemaPattern, String getColumnTypeName(int column tableNamePattern, String[] types) throws SQLException; /* Get column typelist classwith namegiven forcondition thefrom column indexsession. */ @Override public public String getColumnClassName(int column) throws SQLException; } |
Unsupported Features
...
ResultSet getColumns(String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern) throws SQLException;
/* Get primary key list for given table from session. */
@Override
public ResultSet getPrimaryKeys(String catalog, String schema, String table) throws SQLException;
} |
- Methods in
FlinkResultSetMetaData
:FlinkResultSetMetaData
only supports getting column information according to column index or name.
Code Block | ||
---|---|---|
| ||
/* ResultSetMetaData in flink sql driver. */
public class FlinkResultSetMetaData implements ResultSetMetaData {
/* Get column count in the result set. */
@Override
public int getColumnCount() throws SQLException;
/* If the column may be null. */
@Override
public int isNullable(int column) throws SQLException;
/* Get display size for the column. */
@Override
public int getColumnDisplaySize(int column) throws SQLException;
/* Get column name according to column index. */
@Override
public String getColumnLabel(int column) throws SQLException;
public String getColumnName(int column) throws SQLException;
/* Get precision for the column index. */
@Override
public int getPrecision(int column) throws SQLException;
/* Get column type id for the column index. */
@Override
public int getColumnType(int column) throws SQLException;
/* Get column type name for the column index. */
@Override
public String getColumnTypeName(int column) throws SQLException;
/* Get column type class name for the column index. */
@Override
public String getColumnClassName(int column) throws SQLException;
} |
Unsupported Features
- Don't support transaction such as commit, rollback
- Don't support prepare statement, prepare call and etc operations
- Don't support management operations such as savepoint and etc
Exception Handling
When an error occurs, Flink Jdbc Driver mainly throws the following exceptions
SQLState Class | SQLState SubClass | Reason | Exception | Operations |
---|---|---|---|---|
22 | 000 to 02H according to different errors | Description of data conversion error | SQLDataException | Get data error from ResultSet in methods getXXX |
0A | 000 | Specific feature is not supported | SQLFeatureNotSupportedException | All unimplemented methods will throw this exception |
58 | 004 | The exception or error message from Gateway | SQLNonTransientException | Gateway throws an exception or returns an error message when executing the query |
08 | 006 | The session is not exist in Gateway and client need to create new connection to it | SQLNonTransientConnectionException | Gateway is restarted and the client need to create new connection |
We can continue to subdivide and throw different exceptions according to the error information returned by the Gateway in Flink Jdbc Driver in the future
...
[1] https://github.com/ververica/flink-sql-gateway
...