[FLINK-13313] create CatalogTableBuilder to support building CatalogTable from descriptors

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.9.0, 1.10.0
• Fix Version/s: 1.9.0
• Component/s: Table SQL / API
• Labels: pull-request-available
• Assignee: Bowen Li
• Reporter: Bowen Li
• Created: 17/Jul/19 22:20
• Updated: 01/Aug/19 10:57
• Resolved: 22/Jul/19 15:17

Description

Found a usability issue.

Previously, users can create an ExternalCatalogTable (deprecated) from descriptors via ExternalCatalogTableBuilder, and this helps smooth user experience of Flink Table API. E.g.

ExternalCatalogTable table = ExternalCatalogTableBuilder(
	new ExternalSystemXYZ()
		.version("0.11"))
		.withFormat(
			new Json()
				.jsonSchema("{...}")
				.failOnMissingField(false))
		。withSchema(
			new Schema()
				.field("user-name", "VARCHAR").from("u_name")
				.field("count", "DECIMAL")
		.supportsStreaming()
		.asTableSource()

oldCatalog.createTable("tble_name", table, false)

If we don't have a builder to connect new CatalogTable and descriptor, how a user creates CatalogTable would be like the following example, which is quite inconvenient given users have to know all the key names.

TableSchema schema = TableSchema.builder()
	.field("name", DataTypes.STRING())
	.field("age", DataTypes.INT())
	.build();

Map<String, String> properties = new HashMap<>();
properties.put(CatalogConfig.IS_GENERIC, String.valueOf(true));
properties.put("connector.type", "filesystem");
properties.put("connector.path", "/tmp");
properties.put("connector.property-version", "1");
properties.put("update-mode", "append");

properties.put("format.type", "csv");
properties.put("format.property-version", "1");
properties.put("format.fields.0.name", "name");
properties.put("format.fields.0.type", "STRING");
properties.put("format.fields.1.name", "age");
properties.put("format.fields.1.type", "INT");

ObjectPath path = new ObjectPath("mydb", "mytable");
CatalogTable table = new CatalogTableImpl(schema, properties, "csv table");

We need a similar new class CatalogTableBuilder for new Catalog APIs

cc tzulitai ykt836 xuefuz

Issue Links

• links to: GitHub Pull Request #9172