[BEAM-2528] BeamSQL DDL :: CreateTable

[xumingming]

I started this PR as an initial attempt to implement the BeamSQL create table statement. The implementation might not be so mature, but I hope this could be a place we can discuss deeper about the create table. I will introduce this PR in the following 3 aspects:

• MetaStore
• TableProvider
• Grammar

MetaStore

Metastore is responsible for handling the CRUD of table during a session. e.g. create a table, query all tables, query a table by the specified name etc. When a table is created, the table meta info can be persisted by the metastore, but the default InMemoryMetaStore will only store the meta info in memory, so it will NOT be persisted, but user can implement the MetaStore interface to make a persistent implementation.

The table names in MetaStore need to be unique.

TableProvider

The tables in MetaStore can come from many different sources, the construction of a usable table is the responsibility of a TableProvider, TableProvider have the similar interface like MetaStore, but it only handles a specific type of table, e.g. TextTableProvider only handle text tables, while KafakaTableProvider only handle kafka tables.

In this PR, only TextTableProvider and KafakaTableProvider are implemented as example.

Grammar

The grammar for create a TEXT table is:

CREATE TABLE ORDERS(
   ID INT PRIMARY KEY COMMENT 'this is the primary key',
   NAME VARCHAR(127) COMMENT 'this is the name'
)
TYPE 'text'
COMMENT 'this is the table orders'
LOCATION 'text://home/admin/orders'
TBLPROPERTIES '{"format": "Excel"}'

LOCATION dictates where the data of the table is stored. The scheme of the LOCATION dictate the table type, e.g. in the above example, the table type is text, using the table type we can find the corresponding TextTableProvider using the ServiceLoader merchanism.

TBLPROPERTIES is used to specify some other properties of the table, in the above example, we specified the format of each line of text file: Excel(one variant of CSV format).

The grammar for create a KAFKA table is:

 CREATE TABLE ORDERS(
   ID INT PRIMARY KEY COMMENT 'this is the primary key',
   NAME VARCHAR(127) COMMENT 'this is the name'
 )
TYPE 'kafka'
 COMMENT 'this is the table orders'
 LOCATION 'kafka://localhost:2181/brokers?topic=test'
 TBLPROPERTIES '{"bootstrap.servers":"localhost:9092", "topics": ["topic1", "topic2"]}'

[xumingming]

R: @takidau
R: @xumingmin

[mingmxu]

Thanks @xumingming to bring back this topic, starting to read the code

[mingmxu]

overall it's great for DDL support. One minor opinion that, can we change LOCATION to TABLE_PROVIDER directly? It may be complex to fit every source to a location path.

[mingmxu]

retest this please

[xumingming]

I agree with you, use a dedicated syntax to denote the type of table will be more extensible, how about a keyword like type? It's more user-friendly.

[xumingming]

Rebased & added type keyword to denote the table type.

[mingmxu]

LGTM, merging