/**
  * Creates a new table from the given descriptor.
  *
  * The table is created in the catalog defined by the given path.
  */
void createTable(String path, TableDescriptor descriptor);

/**
  * Creates a new temporary table from the given descriptor.
  *
  * Temporary objects can shadow permanent ones. If a permanent object in a given path exists,
  * it will be inaccessible in the current session. To make the permanent object available again
  * one can drop the corresponding temporary object.
  */
void createTemporaryTable(String path, TableDescriptor descriptor);

TableDescriptor {
  // Create a builder
  static TableDescriptorBuilder forConnector(String connector);
  
  Optional<Schema> getSchema();
  Map<String, String> getOptions();
  Optional<String> getComment();
  List<String> getPartitionKeys();
  Optional<TableLikeDescriptor> getLikeDescriptor();
}

TableDescriptorBuilder<SELF> {
  SELF schema(Schema schema);
  
  SELF comment(String comment);

  SELF option<T>(ConfigOption<T> configOption, T value);
  SELF option(String key, String value);

  SELF format(String format);
  SELF format(ConfigOption<?> formatOption, String format);  
  SELF format(FormatDescriptor formatDescriptor);
  SELF format(ConfigOption<?> formatOption, FormatDescriptor formatDescriptor);
  
  SELF partitionedBy(String... partitionKeys);
  
  SELF like(String tableName, LikeOption... likeOptions);
  
  TableDescriptor build();
}

TableLikeDescriptor {
  String getTableName();
  List<TableLikeOption> getLikeOptions();
}

FormatDescriptor {
  static FormatDescriptorBuilder forFormat(String format);
  
  String getFormat();
  Map<String, String> getOptions();
}

FormatDescriptorBuilder<SELF> {
  SELF option<T>(ConfigOption<T> configOption, T value);
  SELF option(String key, String value);
  FormatDescriptor build();
}

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

tEnv.createTable(
  "cat.db.MyTable",

  TableDescriptor.forConnector("kafka")
    .comment("This is a comment")
    .schema(Schema.newBuilder()
      .column("f0", DataTypes.BIGINT())
      .columnByExpression("f1", "2 * f0")
      .columnByMetadata("f3", DataTypes.STRING())
      .column("t", DataTypes.TIMESTAMP(3))
      .watermark("t", "t - INTERVAL '1' MINUTE")
      .primaryKey("f0")
      .build())  
    .partitionedBy("f0")
    .option(KafkaOptions.TOPIC, topic)
    .option("properties.bootstrap.servers", "…")
    .format("json")
    .build()
);

tEnv.createTemporaryTable(
  "MyTemporaryTable",

  TableDescriptor.forConnector("kafka")
    // …
    .like("cat.db.MyTable")
);

/**
  * Returns a {@link Table} backed by the given descriptor.
  */
Table from(TableDescriptor descriptor); 

/**
  * Declares that the pipeline defined by this table should be written to a table defined by the given descriptor.
  *
  * If no schema is defined in the descriptor, it will be inferred automatically.
  */
TableResult executeInsert(TableDescriptor descriptor);

/**
  * Adds the given table as a sink defined by the descriptor.
  */
StatementSet addInsert(TableDescriptor descriptor, Table table);

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")
            .scanStartupModeEarliest()
            .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");