/** creates a temporary table from a descriptor. */
void createTemporaryTable(String tablePath, TableDescriptor tableDescriptor);

// register a table using specific descriptor
tEnv.createTemporaryTable(
   "MyTable",
   new Kafka()
      .version("0.11")
      .topic("user_logs")
      .property("bootstrap.servers", "localhost:9092")
      .property("group.id", "test-group")
      .startFromEarliest()
      .sinkPartitionerRoundRobin()
      .format(new Json().ignoreParseErrors(false))
      .schema(
         new Schema()
            .column("user_id", DataTypes.BIGINT())
            .column("user_name", DataTypes.STRING())
            .column("score", DataTypes.DECIMAL(10, 2))
            .column("log_ts", DataTypes.STRING())
            .column("part_field_0", DataTypes.STRING())
            .column("part_field_1", DataTypes.INT())
            .column("proc", proctime()) // define a processing-time attribute with column name "proc"
            .column("my_ts", toTimestamp($("log_ts"))  // computed column
            .watermarkFor("my_ts", $("my_ts").minus(lit(3).seconds()))  // defines watermark and rowtime attribute
            .primaryKey("user_id"))
      .partitionedBy("part_field_0", "part_field_1")  // Kafka doesn't support partitioned table yet, this is just an example for the API
);

// register a table using general purpose Connector descriptor, this would be helpful for custom source/sinks
tEnv.createTemporaryTable(
   "MyTable",
   new Connector("kafka-0.11")
      .option("topic", "user_logs")
      .option("properties.bootstrap.servers", "localhost:9092")
      .option("properties.group.id", "test-group")
      .option("scan.startup.mode", "earliest")
      .option("format", "json")
      .option("json.ignore-parse-errors", "true")
      .option("sink.partitioner", "round-robin")
      .schema(
         new Schema()
            .column("user_id", DataTypes.BIGINT())
            .column("user_name", DataTypes.STRING())
            .column("score", DataTypes.DECIMAL(10, 2))
            .column("log_ts", DataTypes.STRING())
            .column("part_field_0", DataTypes.STRING())
            .column("part_field_1", DataTypes.INT())
            .column("proc", proctime()) // define a processing-time attribute with column name "proc"
            .column("my_ts", toTimestamp($("log_ts"))  // computed column
            .watermarkFor("my_ts", $("my_ts").minus(lit(3).seconds())) // defines watermark and rowtime attribute
            .primaryKey("user_id"))
      .partitionedBy("part_field_0", "part_field_1") // Kafka doesn't support partitioned table yet, this is just an example for the API
);

tEnv.createTemporaryTable(
   "OrdersInKafka",
   new Kafka()
      .topic("user_logs")
      .property("bootstrap.servers", "localhost:9092")
      .property("group.id", "test-group")
      .format(new Json().ignoreParseErrors(false))
      .schema(
         new Schema()
            .column("user_id", DataTypes.BIGINT())
            .column("score", DataTypes.DECIMAL(10, 2))
            .column("log_ts", DataTypes.TIMESTAMP(3))
            .column("my_ts", toTimestamp($("log_ts"))
      )
);

tEnv.createTemporaryTable(
   "OrdersInFilesystem",
   new Connector("filesystem")
      .option("path", "path/to/whatever")
      .schema(
         new Schema()
            .watermarkFor("my_ts", $("my_ts").minus(lit(3).seconds())))
      .like("OrdersInKafka", LikeOption.EXCLUDING.ALL, LikeOption.INCLUDING.GENERATED)
);

tEnv.createTemporaryTable(
   "OrdersInFilesystem",
   new Connector("filesystem")
      .option("path", "path/to/whatever")
      .schema(
         new Schema()
            .column("user_id", DataTypes.BIGINT())
            .column("score", DataTypes.DECIMAL(10, 2))
            .column("log_ts", DataTypes.TIMESTAMP(3))
            .column("my_ts", toTimestamp($("log_ts"))
            .watermarkFor("my_ts", $("my_ts").minus(lit(3).seconds()))
      )
);

interface TableEnvironment {
  /** reads a table from the given descriptor */
  Table from(TableDescriptor tableDescriptor); 
  // we already have a "from(String)" method to get registered table from catalog
}

interface Table {
  /** Writes the Table to a sink that is specified by the given descriptor. */
  TableResult executeInsert(TableDescriptor tableDescriptor); 
  // we already have a "executeInsert(String)" method to write into a registered table in catalog
}

Schema schema = new Schema()
   .column("user_id", DataTypes.BIGINT())
   .column("score", DataTypes.DECIMAL(10, 2))
   .column("ts", DataTypes.TIMESTAMP(3));
Table myKafka = tEnv.from(
   new Kafka()
      .version("0.11")
      .topic("user_logs")
      .property("bootstrap.servers", "localhost:9092")
      .property("group.id", "test-group")
      .startFromEarliest()
      .sinkPartitionerRoundRobin()
      .format(new Json().ignoreParseErrors(false))
      .schema(schema)
);
// reading from kafka table and write into filesystem table
myKafka.executeInsert(
   new Connector("filesystem")
      .option("path", "/path/to/whatever")
      .option("format", "json")
      .schema(schema)
);

/**
 * 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... columnNames) {...}

	/**
	 * 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 interface LikeOption {
	enum INCLUDING implements LikeOption {
		ALL,
		CONSTRAINTS,
		GENERATED,
		OPTIONS,
		PARTITIONS,
		WATERMARKS
	}

	enum EXCLUDING implements LikeOption {
		ALL,
		CONSTRAINTS,
		GENERATED,
		OPTIONS,
		PARTITIONS,
		WATERMARKS
	}

	enum OVERWRITING implements LikeOption {
		GENERATED,
		OPTIONS,
		WATERMARKS
	}
}

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;
	}

   ...
}

/**
 * Describes a schema of a table.
 */
@PublicEvolving
public class Schema {

	/**
	 * Adds a column with the column name and the data type.
	 */
	public Schema column(String columnName, DataType columnType) {...}

	/**
	 * Adds a computed column with the column name and the column Expression.
	 */
	public Schema column(String columnName, Expression columnExpr) {...}


	/**
	 * Specifies the primary key constraint for a set of given columns.
	 */
	public Schema primaryKey(String... columnNames) {...}

	/**
	 * Specifies the watermark strategy for rowtime attribute.
	 */
	public SchemaWithWatermark watermarkFor(String rowtimeColumn, Expression watermarkExpr) {...}
}

tableEnv.connect(
        new Kafka() // can be replaced by new Connector("kafka-0.11")
            .version("0.11")
            .topic("myTopic")
            .property("bootstrap.servers", "localhost:9092")
            .property("group.id", "test-group")
            .startFromEarliest()
            .sinkPartitionerRoundRobin()
            .format(new Json().ignoreParseErrors(false))
    .schema(
        new Schema()
            .column("user_id", DataTypes.BIGINT())
            .column("user_name", DataTypes.STRING())
            .column("score", DataTypes.DECIMAL(10, 2))
            .column("log_ts", DataTypes.TIMESTAMP(3))
            .column("part_field_0", DataTypes.STRING())
            .column("part_field_1", DataTypes.INT())
            .column("proc", proctime())
            .column("my_ts", toTimestamp($("log_ts")) 
            .watermarkFor("my_ts", $("my_ts").minus(lit(3).seconds())))
            .primaryKey("user_id")
    .partitionedBy("part_field_0", "part_field_1")
    .createTemporaryTable("MyTable");