Page History

/v1/sessions

Verb: POST

Response code: 200 OK

Create a new session with the specific configuraion. In the release-1.16, we only supports to load the jar in the local file system. But we can extend this feature to load the jar in the remote filesystem, e.g. HDFS.

Request body

{

"session_name": "", # optional

"libs": [], # optional. 

"jars": [], # optional

"properties": { # optional, properties for current session

  "key": "value"

}

}

Response body

{

"session_handle": "", # if session is created successfully

}

/v1/sessions/:session_handle

Verb: DELETE

Response code: 200 OK

Close a session, release related resources including operations and properties

Request body

{}

Response body

{

"status": "CLOSED" # if cancel successfully

}

/v1/sessions/:session_handle

Verb: GET

Response code: 200 OK

Get the session config with the specified session handle.

Request body

{}

Response body

{

"properties": {

  "key": "value"

}

}

/v1/sessions/:session_handle/initialize_session

Verb: POST

Response code: 200 OK

Configures the session with the statement which could be SET/RESET/CREATE/ DROP/LOAD/UNLOAD/USE/ALTER/ADD JAR/REMOVE JAR. It can be used to initialize the Session:

• register Catalog 
• register user defined function
• create predefined table

Note: The statement must be a single command, otherwise the server will throw an exception.

Request body

{

"statement": "", # required

"execution_timeout": "" # execution time limit in milliseconds, optional

}

Response body

{}

/v1/sessions/:session_handle/operations/:operation_handle/status

Verb: GET

Response code: 200 OK

Get the status of a running job.

If the session is expired, the server will throw "session not found" exception.

If the job is finished, the server will throw "job not found" exception.

Request body

{}

Response body

{

"status": "" # refer to OperationStatus

}

/v1/sessions/:session_handle/operations/:operation_handle/cancel

Verb: PUT

Response code: 200 OK

Cancel the running operation and update the opeartion status.

Request body

{}

Response body

{

"status": "CANCELED" # if cancel successfully

}

/v1/sessions/:session_handle/operations/:operation_handle

Verb: DELETE

Response code: 200 OK

Remove the specified Operation.

If the user invokes closeOperation twice, the later invocation will get exception.

Request body

{}

Response body

{

"status": "CLOSED" # if close successfully

}

/v1/sessions/:session_handle/statements

Verb: POST

Response code: 200 OK

Execute a statement which could be all Flink supports SQL statement.

The SET xx=yy statement will override/update the TableConfig held by current session, and the RESET statement will reset all properties set by SET xx=yy statement.

The USE MODULE/CATALOG/DATABASE xx statement will update the default module/catalog/database in TableEnvironment held by current session.

The statement must be a single command, otherwise the server will throw an exception.

For BEGIN STATEMENT SET, it will open a buffer in the Session and allows the users to submit the insert statement into the Session later. When the Session receives the END statement, the Gateway will submit the buffered statements.

For ADD JAR/REMOVE JAR, if the jar is in the local environment, we will just add it into the class path or remove it from the class path. If the jar is the remote jar, we will create a session level directory and download the jar into the directory. When the session closes, it should also clean up all the resources in the session-level directory.

Request body

{

"statement": "", # required

"execution_timeout": "" # execution time limit in milliseconds, optional

}

Response body

{

"operation_handle": "",

"operation_type": "EXECUTE_STATEMNT",

"has_result": true/false # determine whether needs to fetch results later

}

/v1/sessions/:session_handle/operations/:operation_handle/result/:token

Verb: GET

Response code: 200 OK

Fetch a part of result for a flink job execution. If the result data is too large or the result is streaming, we can use this API to get a part of the result at a time. The initialized value of token is 0. The token in the next request must be the same as the token in the current request or must be equal to token (in the current request) + 1, otherwise the client will get an exception from the server. If multiple requests are executed with the same token, the result is the same. This design makes sure the client could get the result even if some errors occur in client. The client can get the next part of result using /v1/sessions/:session_id/jobs/:job_id/result/:{token+1} (which is the value of next_result_uri in the response data). If next_result_uri is empty, no more data is remaining in the server.

The server could drop the old data before current token. (The client successfully obtains those data)

We will introduce fetch_size or max_wait_time (to reach the fetch_size) for optimization in future.

The returned result has the same schema as the TableResult#getResolvedSchema.

Please refer to the Appendix about the transormation between the ResultSet and JSON.

Request body

{}

Response body

{

"result_type": "PAYLOAD",

"results": [ # currently, there is only one result now. If multiple queries is executed in a single job, there are many results.

{

"columns": [ # if the execution is successful

{

"name": "",

"type": {"type":  "BOOLEAN", "nullable": true}# string value of LogicalType

},

],

"data": [

["value", ], # a row data

]

},

],

"next_result_uri": /v1/sessions/:session_id/jobs/:job_id/result/:{token+1} # if not empty, uses this uri to fetch next part of result, else there is no more result.

"exception": {

      "root_cause": "....",

     "exception_stack": "..." 

}

}

/v1/sessions/:session_handle/complete_statement

Verb: GET

Response code: 200 OK

Complete the statements. For example, users input SELE in the terminal and press the tab, the terminal will use the API to complete the statement and return the SELECT.

Request body

{

"statement": "",

"postion":

}

Response body

{

"candidates": []

}

/v1/info

Verb: GET

Response code: 200 OK

Get meta data for this cluster

Request body

{}

Response body

{

"product_name": "Apache Flink",

"version": "1.16" # Flink version

}