Airblader commented on a change in pull request #16598: URL: https://github.com/apache/flink/pull/16598#discussion_r700234951
########## File path: docs/content/docs/connectors/table/pubsub.md ########## @@ -0,0 +1,145 @@ +--- +title: PubSub +weight: 20 +type: docs +aliases: + - /dev/table/connectors/pubsub.html +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Google Cloud PubSub SQL Connector + +{{< label "Scan Source: Unbounded" >}} +{{< label "Sink: Streaming Append Mode" >}} + +This connector provides a Source that can read from Google Cloud PubSub. Review comment: ```suggestion This connector allows reading from Google Cloud PubSub. ``` ########## File path: docs/content/docs/connectors/table/pubsub.md ########## @@ -0,0 +1,145 @@ +--- +title: PubSub +weight: 20 +type: docs +aliases: + - /dev/table/connectors/pubsub.html +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Google Cloud PubSub SQL Connector + +{{< label "Scan Source: Unbounded" >}} +{{< label "Sink: Streaming Append Mode" >}} + +This connector provides a Source that can read from Google Cloud PubSub. + +Dependencies +------------ + +{{< sql_download_table "pubsub" >}} + +The PubSub connector is not currently part of the binary distribution. +See how to link with it for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}). + +How to create a PubSub table +---------------- + +The example below shows how to create a PubSub table: + +```sql +CREATE TABLE PubSubTable ( + `user_id` BIGINT, + `item_id` BIGINT, + `behavior` STRING +) WITH ( + 'connector' = 'pubsub', + 'format' = 'json', + 'projectName' = 'gcpProject', + 'subscription' = 'mySubscription' +) +``` + +Available Metadata +------------------ + +The following connector metadata can be accessed as metadata columns in a table definition. + +The `R/W` column defines whether a metadata field is readable (`R`) and/or writable (`W`). +Read-only columns must be declared `VIRTUAL` to exclude them during an `INSERT INTO` operation. + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Key</th> + <th class="text-center" style="width: 30%">Data Type</th> + <th class="text-center" style="width: 40%">Description</th> + <th class="text-center" style="width: 5%">R/W</th> + </tr> + </thead> + <tbody> + <tr> + <td><code>projectName</code></td> + <td><code>STRING NOT NULL</code></td> + <td>Name of the project in GCP contains the PubSub subscription.</td> + <td><code>R</code></td> + </tr> + <tr> + <td><code>subscription</code></td> + <td><code>STRING NOT NULL</code></td> + <td>Name of the GCP PubSub subscription.</td> + <td><code>R</code></td> + </tr> + </tbody> +</table> + + +Connector Options +---------------- + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Option</th> + <th class="text-center" style="width: 8%">Required</th> + <th class="text-center" style="width: 7%">Default</th> + <th class="text-center" style="width: 10%">Type</th> + <th class="text-center" style="width: 50%">Description</th> + </tr> + </thead> + <tbody> + <tr> + <td><h5>connector</h5></td> + <td>required</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>Specify what connector to use, for PubSub use <code>'pubsub'</code>.</td> + </tr> + <tr> + <td><h5>projectName</h5></td> + <td>required</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>Project name in GCP where the PubSub subscription/topic is defined.</td> + </tr> + <tr> + <td><h5>subscription</h5></td> + <td>required by source</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>PubSub subscriptiob name to read data from when the table is used as source.</td> Review comment: ```suggestion <td>Name of the PubSub subscription to read from.</td> ``` ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java ########## @@ -0,0 +1,27 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; + +/** + * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink + * SQL dialect and the Flink Table API. + */ +@PublicEvolving +public class PubSubConnectorConfigOptions { Review comment: Can we please rename this to `PubSubConnectorOptions`? We recently went through some changes and want to make this consistent. ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java ########## @@ -0,0 +1,67 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.factories.DeserializationFormatFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.types.DataType; + +import java.util.HashSet; +import java.util.Set; + +/** Factory for creating {@link PubsubDynamicSource}. */ +@Internal +public class PubSubDynamicTableFactory implements DynamicTableSourceFactory { + + @Override + public String factoryIdentifier() { + return PubSubConnectorConfigOptions.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + final Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PubSubConnectorConfigOptions.PROJECT_NAME); + options.add(PubSubConnectorConfigOptions.SUBSCRIPTION); + options.add(FactoryUtil.FORMAT); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return new HashSet<>(); + } + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + + final FactoryUtil.TableFactoryHelper helper = + FactoryUtil.createTableFactoryHelper(this, context); + + // discover a suitable decoding format + final DecodingFormat<DeserializationSchema<RowData>> decodingFormat = + helper.discoverDecodingFormat( + DeserializationFormatFactory.class, FactoryUtil.FORMAT); + + // validate all options Review comment: ```suggestion ``` ########## File path: docs/content/docs/connectors/table/pubsub.md ########## @@ -0,0 +1,145 @@ +--- +title: PubSub +weight: 20 +type: docs +aliases: + - /dev/table/connectors/pubsub.html +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Google Cloud PubSub SQL Connector + +{{< label "Scan Source: Unbounded" >}} +{{< label "Sink: Streaming Append Mode" >}} + +This connector provides a Source that can read from Google Cloud PubSub. + +Dependencies +------------ + +{{< sql_download_table "pubsub" >}} + +The PubSub connector is not currently part of the binary distribution. +See how to link with it for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}). + +How to create a PubSub table +---------------- + +The example below shows how to create a PubSub table: + +```sql +CREATE TABLE PubSubTable ( + `user_id` BIGINT, + `item_id` BIGINT, + `behavior` STRING +) WITH ( + 'connector' = 'pubsub', + 'format' = 'json', + 'projectName' = 'gcpProject', + 'subscription' = 'mySubscription' +) +``` + +Available Metadata +------------------ + +The following connector metadata can be accessed as metadata columns in a table definition. + +The `R/W` column defines whether a metadata field is readable (`R`) and/or writable (`W`). +Read-only columns must be declared `VIRTUAL` to exclude them during an `INSERT INTO` operation. + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Key</th> + <th class="text-center" style="width: 30%">Data Type</th> + <th class="text-center" style="width: 40%">Description</th> + <th class="text-center" style="width: 5%">R/W</th> + </tr> + </thead> + <tbody> + <tr> + <td><code>projectName</code></td> + <td><code>STRING NOT NULL</code></td> + <td>Name of the project in GCP contains the PubSub subscription.</td> + <td><code>R</code></td> + </tr> + <tr> + <td><code>subscription</code></td> + <td><code>STRING NOT NULL</code></td> + <td>Name of the GCP PubSub subscription.</td> + <td><code>R</code></td> + </tr> + </tbody> +</table> + + +Connector Options +---------------- + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Option</th> + <th class="text-center" style="width: 8%">Required</th> + <th class="text-center" style="width: 7%">Default</th> + <th class="text-center" style="width: 10%">Type</th> + <th class="text-center" style="width: 50%">Description</th> + </tr> + </thead> + <tbody> + <tr> + <td><h5>connector</h5></td> + <td>required</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>Specify what connector to use, for PubSub use <code>'pubsub'</code>.</td> + </tr> + <tr> + <td><h5>projectName</h5></td> + <td>required</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>Project name in GCP where the PubSub subscription/topic is defined.</td> + </tr> + <tr> + <td><h5>subscription</h5></td> + <td>required by source</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>PubSub subscriptiob name to read data from when the table is used as source.</td> + </tr> + <tr> + <td><h5>format</h5></td> + <td>required</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>The format used to deserialize and serialize the value part of PubSub messages. + Please refer to the <a href="{{< ref "docs/connectors/table/formats/overview" >}}">formats</a> page for + more details and more format options. + </td> + </tr> + + </tbody> +</table> + +Features +---------------- + N/A Review comment: This empty section doesn't seem useful. ```suggestion ``` ########## File path: docs/content/docs/connectors/table/pubsub.md ########## @@ -0,0 +1,145 @@ +--- +title: PubSub +weight: 20 +type: docs +aliases: + - /dev/table/connectors/pubsub.html +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Google Cloud PubSub SQL Connector + +{{< label "Scan Source: Unbounded" >}} +{{< label "Sink: Streaming Append Mode" >}} Review comment: What's the purpose of this label here given that this is a source-only connector? ########## File path: docs/content/docs/connectors/table/pubsub.md ########## @@ -0,0 +1,145 @@ +--- +title: PubSub +weight: 20 +type: docs +aliases: + - /dev/table/connectors/pubsub.html +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Google Cloud PubSub SQL Connector + +{{< label "Scan Source: Unbounded" >}} +{{< label "Sink: Streaming Append Mode" >}} + +This connector provides a Source that can read from Google Cloud PubSub. + +Dependencies +------------ + +{{< sql_download_table "pubsub" >}} + +The PubSub connector is not currently part of the binary distribution. Review comment: I think if we offer it, we should also distribute it? CC @twalthr ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java ########## @@ -0,0 +1,27 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; + +/** + * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink + * SQL dialect and the Flink Table API. + */ Review comment: ```suggestion /** Options for the PubSub connector. */ ``` ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java ########## @@ -0,0 +1,27 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; + +/** + * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink + * SQL dialect and the Flink Table API. + */ +@PublicEvolving +public class PubSubConnectorConfigOptions { + + public static final ConfigOption<String> PROJECT_NAME = + ConfigOptions.key("projectName") + .stringType() + .noDefaultValue() + .withDescription("Name of the PubSub project backing this table."); + + public static final ConfigOption<String> SUBSCRIPTION = + ConfigOptions.key("subscription") + .stringType() + .noDefaultValue() + .withDescription("Name of the PubSub subscription backing this table."); + + public static final String IDENTIFIER = "pubsub"; +} Review comment: ```suggestion private PubSubConnectorConfigOptions() {} } ``` ########## File path: docs/content/docs/connectors/table/pubsub.md ########## @@ -0,0 +1,145 @@ +--- +title: PubSub +weight: 20 +type: docs +aliases: + - /dev/table/connectors/pubsub.html +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Google Cloud PubSub SQL Connector + +{{< label "Scan Source: Unbounded" >}} +{{< label "Sink: Streaming Append Mode" >}} + +This connector provides a Source that can read from Google Cloud PubSub. + +Dependencies +------------ + +{{< sql_download_table "pubsub" >}} + +The PubSub connector is not currently part of the binary distribution. +See how to link with it for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}). + +How to create a PubSub table +---------------- + +The example below shows how to create a PubSub table: + +```sql +CREATE TABLE PubSubTable ( + `user_id` BIGINT, + `item_id` BIGINT, + `behavior` STRING +) WITH ( + 'connector' = 'pubsub', + 'format' = 'json', + 'projectName' = 'gcpProject', + 'subscription' = 'mySubscription' +) +``` + +Available Metadata +------------------ + +The following connector metadata can be accessed as metadata columns in a table definition. + +The `R/W` column defines whether a metadata field is readable (`R`) and/or writable (`W`). +Read-only columns must be declared `VIRTUAL` to exclude them during an `INSERT INTO` operation. + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Key</th> + <th class="text-center" style="width: 30%">Data Type</th> + <th class="text-center" style="width: 40%">Description</th> + <th class="text-center" style="width: 5%">R/W</th> + </tr> + </thead> + <tbody> + <tr> + <td><code>projectName</code></td> + <td><code>STRING NOT NULL</code></td> + <td>Name of the project in GCP contains the PubSub subscription.</td> + <td><code>R</code></td> + </tr> + <tr> + <td><code>subscription</code></td> + <td><code>STRING NOT NULL</code></td> + <td>Name of the GCP PubSub subscription.</td> + <td><code>R</code></td> + </tr> + </tbody> +</table> + + +Connector Options +---------------- + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Option</th> + <th class="text-center" style="width: 8%">Required</th> + <th class="text-center" style="width: 7%">Default</th> + <th class="text-center" style="width: 10%">Type</th> + <th class="text-center" style="width: 50%">Description</th> + </tr> + </thead> + <tbody> + <tr> + <td><h5>connector</h5></td> + <td>required</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>Specify what connector to use, for PubSub use <code>'pubsub'</code>.</td> + </tr> + <tr> + <td><h5>projectName</h5></td> + <td>required</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>Project name in GCP where the PubSub subscription/topic is defined.</td> Review comment: ```suggestion <td>Name of the GCP project that contains the subscription.</td> ``` ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java ########## @@ -0,0 +1,67 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.factories.DeserializationFormatFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.types.DataType; + +import java.util.HashSet; +import java.util.Set; + +/** Factory for creating {@link PubsubDynamicSource}. */ +@Internal +public class PubSubDynamicTableFactory implements DynamicTableSourceFactory { + + @Override + public String factoryIdentifier() { + return PubSubConnectorConfigOptions.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + final Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PubSubConnectorConfigOptions.PROJECT_NAME); + options.add(PubSubConnectorConfigOptions.SUBSCRIPTION); + options.add(FactoryUtil.FORMAT); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return new HashSet<>(); + } + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + + final FactoryUtil.TableFactoryHelper helper = + FactoryUtil.createTableFactoryHelper(this, context); + + // discover a suitable decoding format + final DecodingFormat<DeserializationSchema<RowData>> decodingFormat = + helper.discoverDecodingFormat( + DeserializationFormatFactory.class, FactoryUtil.FORMAT); + + // validate all options + helper.validate(); + + // get the validated options Review comment: ```suggestion ``` ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java ########## @@ -0,0 +1,27 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; + +/** + * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink + * SQL dialect and the Flink Table API. + */ +@PublicEvolving +public class PubSubConnectorConfigOptions { + + public static final ConfigOption<String> PROJECT_NAME = + ConfigOptions.key("projectName") + .stringType() + .noDefaultValue() + .withDescription("Name of the PubSub project backing this table."); + + public static final ConfigOption<String> SUBSCRIPTION = + ConfigOptions.key("subscription") + .stringType() + .noDefaultValue() + .withDescription("Name of the PubSub subscription backing this table."); + + public static final String IDENTIFIER = "pubsub"; Review comment: This isn't a config option. We currently have these reside in the factory class. I don't think that's ideal (since it is internal), but for now we should move this one to the factory as well, I think. ########## File path: docs/content/docs/connectors/table/pubsub.md ########## @@ -0,0 +1,145 @@ +--- +title: PubSub +weight: 20 +type: docs +aliases: + - /dev/table/connectors/pubsub.html +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Google Cloud PubSub SQL Connector + +{{< label "Scan Source: Unbounded" >}} +{{< label "Sink: Streaming Append Mode" >}} + +This connector provides a Source that can read from Google Cloud PubSub. + +Dependencies +------------ + +{{< sql_download_table "pubsub" >}} + +The PubSub connector is not currently part of the binary distribution. +See how to link with it for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}). + +How to create a PubSub table +---------------- + +The example below shows how to create a PubSub table: + +```sql +CREATE TABLE PubSubTable ( + `user_id` BIGINT, + `item_id` BIGINT, + `behavior` STRING +) WITH ( + 'connector' = 'pubsub', + 'format' = 'json', + 'projectName' = 'gcpProject', + 'subscription' = 'mySubscription' +) +``` + +Available Metadata +------------------ + +The following connector metadata can be accessed as metadata columns in a table definition. + +The `R/W` column defines whether a metadata field is readable (`R`) and/or writable (`W`). +Read-only columns must be declared `VIRTUAL` to exclude them during an `INSERT INTO` operation. + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Key</th> + <th class="text-center" style="width: 30%">Data Type</th> + <th class="text-center" style="width: 40%">Description</th> + <th class="text-center" style="width: 5%">R/W</th> + </tr> + </thead> + <tbody> + <tr> + <td><code>projectName</code></td> + <td><code>STRING NOT NULL</code></td> + <td>Name of the project in GCP contains the PubSub subscription.</td> + <td><code>R</code></td> + </tr> + <tr> + <td><code>subscription</code></td> + <td><code>STRING NOT NULL</code></td> + <td>Name of the GCP PubSub subscription.</td> + <td><code>R</code></td> + </tr> + </tbody> +</table> Review comment: The source does not implement any metadata, so this entire section should be removed. ```suggestion ``` ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java ########## @@ -0,0 +1,67 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.factories.DeserializationFormatFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.types.DataType; + +import java.util.HashSet; +import java.util.Set; + +/** Factory for creating {@link PubsubDynamicSource}. */ +@Internal +public class PubSubDynamicTableFactory implements DynamicTableSourceFactory { + + @Override + public String factoryIdentifier() { + return PubSubConnectorConfigOptions.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + final Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PubSubConnectorConfigOptions.PROJECT_NAME); + options.add(PubSubConnectorConfigOptions.SUBSCRIPTION); + options.add(FactoryUtil.FORMAT); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return new HashSet<>(); + } + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + + final FactoryUtil.TableFactoryHelper helper = + FactoryUtil.createTableFactoryHelper(this, context); + + // discover a suitable decoding format Review comment: All of these comments add no real value as the code reads clear enough. ```suggestion ``` ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubsubDynamicSource.java ########## @@ -0,0 +1,80 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; +import org.apache.flink.table.connector.source.SourceFunctionProvider; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.DataType; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; + +/** PubSub-backed {@link ScanTableSource}. */ +@Internal +public class PubsubDynamicSource implements ScanTableSource { + + private static Logger logger = LoggerFactory.getLogger(PubsubDynamicSource.class); + + /** Name of the PubSub project backing this table. */ + private final String project; + /** Name of the PubSub subscription backing this table. */ + private final String subscription; + /** Scan format for decoding records from PubSub. */ + private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat; + /** Data type that describes the final output of the source. */ + private final DataType producedDataType; + + public PubsubDynamicSource( + String project, + String subscription, + DecodingFormat<DeserializationSchema<RowData>> decodingFormat, + DataType producedDataType) { + + this.project = project; + this.subscription = subscription; + this.decodingFormat = decodingFormat; + this.producedDataType = producedDataType; + } + + @Override + public ChangelogMode getChangelogMode() { + return ChangelogMode.insertOnly(); + } + + @Override + public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) { + // create runtime classes that are shipped to the cluster + + final DeserializationSchema<RowData> deserializer = + decodingFormat.createRuntimeDecoder(runtimeProviderContext, producedDataType); + + try { + PubSubSource<RowData> source = + PubSubSource.newBuilder() + .withDeserializationSchema(deserializer) + .withProjectName(project) + .withSubscriptionName(subscription) Review comment: (We'll need to think about how to provide credentials (not through environment variables) as well, I think, but as a first version we can leave this as is.) ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java ########## @@ -0,0 +1,67 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.factories.DeserializationFormatFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.types.DataType; + +import java.util.HashSet; +import java.util.Set; + +/** Factory for creating {@link PubsubDynamicSource}. */ +@Internal +public class PubSubDynamicTableFactory implements DynamicTableSourceFactory { + + @Override + public String factoryIdentifier() { + return PubSubConnectorConfigOptions.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + final Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PubSubConnectorConfigOptions.PROJECT_NAME); + options.add(PubSubConnectorConfigOptions.SUBSCRIPTION); + options.add(FactoryUtil.FORMAT); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return new HashSet<>(); + } + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + + final FactoryUtil.TableFactoryHelper helper = + FactoryUtil.createTableFactoryHelper(this, context); + + // discover a suitable decoding format + final DecodingFormat<DeserializationSchema<RowData>> decodingFormat = + helper.discoverDecodingFormat( + DeserializationFormatFactory.class, FactoryUtil.FORMAT); + + // validate all options + helper.validate(); + + // get the validated options + final ReadableConfig options = helper.getOptions(); + final String project = options.get(PubSubConnectorConfigOptions.PROJECT_NAME); + final String topic = options.get(PubSubConnectorConfigOptions.SUBSCRIPTION); + + // derive the produced data type (excluding computed columns) from the catalog table + final DataType producedDataType = + context.getCatalogTable().getResolvedSchema().toPhysicalRowDataType(); + + // create and return dynamic table source Review comment: ```suggestion ``` ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java ########## @@ -0,0 +1,67 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.factories.DeserializationFormatFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.types.DataType; + +import java.util.HashSet; +import java.util.Set; + +/** Factory for creating {@link PubsubDynamicSource}. */ +@Internal +public class PubSubDynamicTableFactory implements DynamicTableSourceFactory { + + @Override + public String factoryIdentifier() { + return PubSubConnectorConfigOptions.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + final Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PubSubConnectorConfigOptions.PROJECT_NAME); + options.add(PubSubConnectorConfigOptions.SUBSCRIPTION); + options.add(FactoryUtil.FORMAT); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return new HashSet<>(); + } + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + + final FactoryUtil.TableFactoryHelper helper = + FactoryUtil.createTableFactoryHelper(this, context); + + // discover a suitable decoding format + final DecodingFormat<DeserializationSchema<RowData>> decodingFormat = + helper.discoverDecodingFormat( + DeserializationFormatFactory.class, FactoryUtil.FORMAT); + + // validate all options + helper.validate(); + + // get the validated options + final ReadableConfig options = helper.getOptions(); + final String project = options.get(PubSubConnectorConfigOptions.PROJECT_NAME); + final String topic = options.get(PubSubConnectorConfigOptions.SUBSCRIPTION); + + // derive the produced data type (excluding computed columns) from the catalog table Review comment: ```suggestion ``` ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java ########## @@ -0,0 +1,67 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.factories.DeserializationFormatFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.types.DataType; + +import java.util.HashSet; +import java.util.Set; + +/** Factory for creating {@link PubsubDynamicSource}. */ +@Internal +public class PubSubDynamicTableFactory implements DynamicTableSourceFactory { + + @Override + public String factoryIdentifier() { + return PubSubConnectorConfigOptions.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + final Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PubSubConnectorConfigOptions.PROJECT_NAME); + options.add(PubSubConnectorConfigOptions.SUBSCRIPTION); + options.add(FactoryUtil.FORMAT); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return new HashSet<>(); + } + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + + final FactoryUtil.TableFactoryHelper helper = + FactoryUtil.createTableFactoryHelper(this, context); + + // discover a suitable decoding format + final DecodingFormat<DeserializationSchema<RowData>> decodingFormat = + helper.discoverDecodingFormat( + DeserializationFormatFactory.class, FactoryUtil.FORMAT); + + // validate all options + helper.validate(); + + // get the validated options + final ReadableConfig options = helper.getOptions(); + final String project = options.get(PubSubConnectorConfigOptions.PROJECT_NAME); + final String topic = options.get(PubSubConnectorConfigOptions.SUBSCRIPTION); Review comment: Rename to `subscription`? ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubsubDynamicSource.java ########## @@ -0,0 +1,80 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; +import org.apache.flink.table.connector.source.SourceFunctionProvider; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.DataType; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; + +/** PubSub-backed {@link ScanTableSource}. */ +@Internal +public class PubsubDynamicSource implements ScanTableSource { + + private static Logger logger = LoggerFactory.getLogger(PubsubDynamicSource.class); + + /** Name of the PubSub project backing this table. */ + private final String project; + /** Name of the PubSub subscription backing this table. */ + private final String subscription; + /** Scan format for decoding records from PubSub. */ + private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat; + /** Data type that describes the final output of the source. */ + private final DataType producedDataType; + + public PubsubDynamicSource( + String project, + String subscription, + DecodingFormat<DeserializationSchema<RowData>> decodingFormat, + DataType producedDataType) { + + this.project = project; + this.subscription = subscription; + this.decodingFormat = decodingFormat; + this.producedDataType = producedDataType; + } + + @Override + public ChangelogMode getChangelogMode() { + return ChangelogMode.insertOnly(); + } + + @Override + public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) { + // create runtime classes that are shipped to the cluster + Review comment: ```suggestion ``` ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubsubDynamicSource.java ########## @@ -0,0 +1,80 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; +import org.apache.flink.table.connector.source.SourceFunctionProvider; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.DataType; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; + +/** PubSub-backed {@link ScanTableSource}. */ +@Internal +public class PubsubDynamicSource implements ScanTableSource { + + private static Logger logger = LoggerFactory.getLogger(PubsubDynamicSource.class); + + /** Name of the PubSub project backing this table. */ + private final String project; + /** Name of the PubSub subscription backing this table. */ + private final String subscription; + /** Scan format for decoding records from PubSub. */ + private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat; + /** Data type that describes the final output of the source. */ + private final DataType producedDataType; + + public PubsubDynamicSource( + String project, + String subscription, + DecodingFormat<DeserializationSchema<RowData>> decodingFormat, + DataType producedDataType) { + + this.project = project; + this.subscription = subscription; + this.decodingFormat = decodingFormat; + this.producedDataType = producedDataType; + } + + @Override + public ChangelogMode getChangelogMode() { + return ChangelogMode.insertOnly(); + } + + @Override + public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) { + // create runtime classes that are shipped to the cluster + + final DeserializationSchema<RowData> deserializer = + decodingFormat.createRuntimeDecoder(runtimeProviderContext, producedDataType); + + try { + PubSubSource<RowData> source = + PubSubSource.newBuilder() + .withDeserializationSchema(deserializer) + .withProjectName(project) + .withSubscriptionName(subscription) + .build(); + return SourceFunctionProvider.of(source, false); + } catch (IOException e) { + throw new RuntimeException("failed to create source", e); Review comment: ```suggestion throw new RuntimeException("Failed to create PubSub source.", e); ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
