Page History

/**
 * Describes a table to connect. It is a same representation of SQL CREATE TABLE DDL.
 */
@PublicEvolving
public abstract class TableDescriptor {

	/**
	 * Specifies the table schema.
	 */
	public final TableDescriptor schema(Schema schema) {...}

	/**
	 * Specifies the partition keys of this table.
	 */
	public final TableDescriptor partitionedBy(String... fieldNamescolumnNames) {...}

	/**
	 * Extends some parts from the original regsitered table path.
	 */
	public final TableDescriptor like(String originalTablePath, LikeOption... likeOptions) {...}

	/**
	 * Extends some parts from the original table descriptor.
	 */
	public final TableDescriptor like(TableDescriptor originalTableDescriptor, LikeOption... likeOptions) {...}

	/**
	 * Specifies the connector options of this table, subclasses should override this method.
	 */
	protected abstract Map<String, String> connectorOptions();
}

public class Connector extends TableDescriptor {

	private final Map<String, String> options = new HashMap<>();

	public Connector(String identifier) {
		this.options.put(CONNECTOR.key(), identifier);
	}

	public Connector option(String key, String value) {
		options.put(key, value);
		return this;
	}

	protected Map<String, String> toConnectorOptions() {
		return new HashMap<>(options);
	}
}

public class Kafka extends TableDescriptor {

	private final Map<String, String> options = new HashMap<>();

	public Kafka() {
		this.options.put(CONNECTOR.key(), "kafka");
	}

	public Kafka version(String version) {
		this.options.put(CONNECTOR.key(), "kafka-" + version);
		return this;
	}

	public Kafka topic(String topic) {
		this.options.put("topic", topic);
		return this;
	}

	public Kafka format(FormatDescriptor formatDescriptor) {
		this.options.putAll(formatDescriptor.toFormatOptions());
		return this;
	}

   ...
}