[
https://issues.apache.org/jira/browse/FLINK-9852?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16546399#comment-16546399
]
ASF GitHub Bot commented on FLINK-9852:
---------------------------------------
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6343#discussion_r202977464
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogTable.scala
---
@@ -18,33 +18,299 @@
package org.apache.flink.table.catalog
+import org.apache.flink.table.descriptors.DescriptorProperties.toScala
+import
org.apache.flink.table.descriptors.StatisticsValidator.{STATISTICS_COLUMNS,
STATISTICS_ROW_COUNT, readColumnStats}
+import
org.apache.flink.table.descriptors.StreamTableDescriptorValidator.{UPDATE_MODE,
UPDATE_MODE_VALUE_APPEND, UPDATE_MODE_VALUE_RETRACT, UPDATE_MODE_VALUE_UPSERT}
import org.apache.flink.table.descriptors._
import org.apache.flink.table.plan.stats.TableStats
+import scala.collection.JavaConverters._
+
/**
- * Defines a table in an [[ExternalCatalog]].
+ * Defines a table in an [[ExternalCatalog]]. External catalog tables
describe table sources
+ * and/or sinks for both batch and stream environments.
+ *
+ * The catalog table takes descriptors which allow for declaring the
communication to external
+ * systems in an implementation-agnostic way. The classpath is scanned
for suitable table factories
+ * that match the desired configuration.
+ *
+ * Use the provided builder methods to configure the external catalog
table accordingly.
+ *
+ * The following example shows how to read from a connector using a JSON
format and
+ * declaring it as a table source:
*
- * @param connectorDesc describes the system to connect to
- * @param formatDesc describes the data format of a connector
- * @param schemaDesc describes the schema of the result table
- * @param statisticsDesc describes the estimated statistics of the result
table
- * @param metadataDesc describes additional metadata of a table
+ * {{{
+ * ExternalCatalogTable(
+ * new ExternalSystemXYZ()
+ * .version("0.11"))
+ * .withFormat(
+ * new Json()
+ * .jsonSchema("{...}")
+ * .failOnMissingField(false))
+ * .withSchema(
+ * new Schema()
+ * .field("user-name", "VARCHAR").from("u_name")
+ * .field("count", "DECIMAL")
+ * .asTableSource()
+ * }}}
+ *
+ * Note: For backwards-compatibility, the table is declared as a table
source for batch and
+ * streaming environment by default.
+ *
+ * See also [[org.apache.flink.table.factories.TableFactory]] for more
information about how
+ * to target suitable factories.
+ *
+ * @param connectorDescriptor describes the system to connect to
*/
-class ExternalCatalogTable(
- connectorDesc: ConnectorDescriptor,
- formatDesc: Option[FormatDescriptor],
- schemaDesc: Option[Schema],
- statisticsDesc: Option[Statistics],
- metadataDesc: Option[Metadata])
- extends TableSourceDescriptor {
-
- this.connectorDescriptor = Some(connectorDesc)
- this.formatDescriptor = formatDesc
- this.schemaDescriptor = schemaDesc
- this.statisticsDescriptor = statisticsDesc
- this.metaDescriptor = metadataDesc
-
- // expose statistics for external table source util
- override def getTableStats: Option[TableStats] = super.getTableStats
+class ExternalCatalogTable(val connectorDescriptor: ConnectorDescriptor)
+ extends TableDescriptor
--- End diff --
https://github.com/databricks/scala-style-guide#indent
> Expose descriptor-based sink creation in table environments
> -----------------------------------------------------------
>
> Key: FLINK-9852
> URL: https://issues.apache.org/jira/browse/FLINK-9852
> Project: Flink
> Issue Type: New Feature
> Components: Table API & SQL
> Reporter: Timo Walther
> Assignee: Timo Walther
> Priority: Major
> Labels: pull-request-available
>
> Currently, only a table source can be created using the unified table
> descriptors with {{tableEnv.from(...)}}. A similar approach should be
> supported for defining sinks or even both types at the same time.
> I suggest the following syntax:
> {code}
> tableEnv.connect(Kafka(...)).registerSource("name")
> tableEnv.connect(Kafka(...)).registerSink("name")
> tableEnv.connect(Kafka(...)).registerSourceAndSink("name")
> {code}
> A table could then access the registered source/sink.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
