You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lo...@apache.org on 2022/12/05 18:30:00 UTC
[beam] branch master updated: Run go fmt over full go directory with go 1.19 (#24525)
This is an automated email from the ASF dual-hosted git repository.
lostluck pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 33af390d3a4 Run go fmt over full go directory with go 1.19 (#24525)
33af390d3a4 is described below
commit 33af390d3a4571893b1865d4a816bd8e4aaa2e95
Author: Danny McCormick <da...@google.com>
AuthorDate: Mon Dec 5 13:29:53 2022 -0500
Run go fmt over full go directory with go 1.19 (#24525)
---
sdks/go/cmd/starcgen/starcgen.go | 15 +--
sdks/go/examples/cookbook/tornadoes/tornadoes.go | 4 +-
.../debugging_wordcount/debugging_wordcount.go | 8 +-
sdks/go/examples/fhirio/import/import.go | 33 ++---
.../fhirio/read_write_pubsub/read_write_pubsub.go | 37 ++---
sdks/go/examples/kafka/taxi.go | 62 +++++----
.../go/examples/large_wordcount/large_wordcount.go | 8 +-
.../minimal_wordcount/minimal_wordcount.go | 8 +-
sdks/go/examples/native_wordcap/wordcap.go | 7 +-
sdks/go/examples/streaming_wordcap/wordcap.go | 7 +-
sdks/go/examples/stringsplit/stringsplit.go | 6 +-
sdks/go/examples/xlang/bigquery/wordcount.go | 61 +++++----
sdks/go/examples/yatzy/yatzy.go | 2 +-
sdks/go/pkg/beam/core/funcx/fn.go | 6 +-
sdks/go/pkg/beam/core/funcx/output.go | 7 +-
sdks/go/pkg/beam/core/funcx/sideinput.go | 4 +-
sdks/go/pkg/beam/core/funcx/signature.go | 6 +-
sdks/go/pkg/beam/core/graph/bind.go | 18 +--
sdks/go/pkg/beam/core/graph/coder/registry.go | 8 +-
sdks/go/pkg/beam/core/graph/fn.go | 19 +--
.../beam/core/graph/window/trigger/trigger_test.go | 2 +-
sdks/go/pkg/beam/core/metrics/metrics.go | 7 +-
sdks/go/pkg/beam/core/runtime/exec/combine_test.go | 32 ++---
sdks/go/pkg/beam/core/runtime/exec/fn_test.go | 10 +-
sdks/go/pkg/beam/core/runtime/exec/sdf.go | 150 +++++++++++----------
.../beam/core/runtime/exec/sdf_invokers_test.go | 1 -
sdks/go/pkg/beam/core/runtime/xlangx/registry.go | 11 +-
sdks/go/pkg/beam/core/runtime/xlangx/resolve.go | 9 +-
sdks/go/pkg/beam/core/util/ioutilx/read.go | 1 +
sdks/go/pkg/beam/forward.go | 32 ++---
sdks/go/pkg/beam/gbk.go | 4 +-
sdks/go/pkg/beam/internal/errors/errors.go | 12 +-
sdks/go/pkg/beam/io/databaseio/mapper.go | 10 +-
sdks/go/pkg/beam/io/databaseio/util.go | 2 +-
sdks/go/pkg/beam/io/datastoreio/datastore.go | 1 -
sdks/go/pkg/beam/io/parquetio/parquetio.go | 38 +++---
sdks/go/pkg/beam/io/synthetic/source.go | 27 ++--
sdks/go/pkg/beam/io/synthetic/step.go | 6 +-
sdks/go/pkg/beam/io/xlang/bigqueryio/bigquery.go | 71 +++++-----
sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go | 32 ++---
sdks/go/pkg/beam/io/xlang/jdbcio/jdbc.go | 71 +++++-----
sdks/go/pkg/beam/io/xlang/kafkaio/kafka.go | 31 ++---
.../beam/model/fnexecution_v1/beam_fn_api.pb.go | 38 +++---
.../beam/model/jobmanagement_v1/beam_job_api.pb.go | 11 +-
.../beam/model/pipeline_v1/beam_runner_api.pb.go | 42 +++---
sdks/go/pkg/beam/pardo.go | 146 ++++++++++----------
sdks/go/pkg/beam/partition.go | 4 +-
.../beam/runners/dataflow/dataflowlib/messages.go | 18 +--
.../beam/runners/universal/runnerlib/compile.go | 8 +-
sdks/go/pkg/beam/schema.go | 10 +-
sdks/go/pkg/beam/testing/ptest/ptest.go | 1 -
sdks/go/pkg/beam/transforms/filter/filter.go | 16 +--
sdks/go/pkg/beam/transforms/stats/max.go | 5 +-
sdks/go/pkg/beam/transforms/stats/mean.go | 5 +-
sdks/go/pkg/beam/transforms/stats/min.go | 5 +-
sdks/go/pkg/beam/transforms/stats/sum.go | 5 +-
sdks/go/pkg/beam/transforms/top/top.go | 10 +-
.../pkg/beam/transforms/xlang/python/external.go | 31 +++--
sdks/go/pkg/beam/xlang.go | 61 +++++----
sdks/go/test/integration/expansions.go | 23 ++--
sdks/go/test/integration/flags.go | 9 +-
.../go/test/integration/io/xlang/jdbc/jdbc_test.go | 2 +-
62 files changed, 693 insertions(+), 643 deletions(-)
diff --git a/sdks/go/cmd/starcgen/starcgen.go b/sdks/go/cmd/starcgen/starcgen.go
index d6e6f1f10c5..9e9859b202b 100644
--- a/sdks/go/cmd/starcgen/starcgen.go
+++ b/sdks/go/cmd/starcgen/starcgen.go
@@ -21,15 +21,15 @@
// reflection based shims used by beam. Reflection is convenient for development,
// but is an unnecessary expense on pipeline performance.
//
-// Using This Tool
+// # Using This Tool
//
// This tool is intended for use with `go generate`. The recommended convention
// putting the types and functions used in a separate package from pipeline construction.
// Then, the tool can be used as follows:
//
-// //go:generate go install github.com/apache/beam/sdks/v2/go/cmd/starcgen
-// //go:generate starcgen --package=<mypackagename>
-// //go:generate go fmt
+// //go:generate go install github.com/apache/beam/sdks/v2/go/cmd/starcgen
+// //go:generate starcgen --package=<mypackagename>
+// //go:generate go fmt
//
// This will generate registrations and shim types for all types and functions
// in the package, in a file `<mypackagename>.shims.go`.
@@ -37,10 +37,9 @@
// Alternatively, it's possible to specify the specific input files and identifiers within
// the package for generation.
//
-// //go:generate go install github.com/apache/beam/sdks/v2/go/cmd/starcgen
-// //go:generate starcgen --package=<mypackagename> --inputs=foo.go --identifiers=myFn,myStructFn --output=custom.shims.go
-// //go:generate go fmt
-//
+// //go:generate go install github.com/apache/beam/sdks/v2/go/cmd/starcgen
+// //go:generate starcgen --package=<mypackagename> --inputs=foo.go --identifiers=myFn,myStructFn --output=custom.shims.go
+// //go:generate go fmt
package main
import (
diff --git a/sdks/go/examples/cookbook/tornadoes/tornadoes.go b/sdks/go/examples/cookbook/tornadoes/tornadoes.go
index 24b30f75a9f..1810f63e353 100644
--- a/sdks/go/examples/cookbook/tornadoes/tornadoes.go
+++ b/sdks/go/examples/cookbook/tornadoes/tornadoes.go
@@ -22,12 +22,12 @@
// Note: Before running this example, you must create a BigQuery dataset to
// contain your output table as described here:
//
-// https://cloud.google.com/bigquery/docs/tables#create-table
+// https://cloud.google.com/bigquery/docs/tables#create-table
//
// To execute this pipeline locally, specify the BigQuery table for the output
// with the form:
//
-// --output=YOUR_PROJECT_ID:DATASET_ID.TABLE_ID
+// --output=YOUR_PROJECT_ID:DATASET_ID.TABLE_ID
//
// The BigQuery input table defaults to clouddataflow-readonly:samples.weather_stations
// and can be overridden with {@code --input}.
diff --git a/sdks/go/examples/debugging_wordcount/debugging_wordcount.go b/sdks/go/examples/debugging_wordcount/debugging_wordcount.go
index 7dae41214aa..ee160862011 100644
--- a/sdks/go/examples/debugging_wordcount/debugging_wordcount.go
+++ b/sdks/go/examples/debugging_wordcount/debugging_wordcount.go
@@ -28,13 +28,13 @@
//
// New Concepts:
//
-// 1. Using the richer struct DoFn form and accessing optional arguments.
-// 2. Logging using the Beam log package, even in a distributed environment
-// 3. Testing your Pipeline via passert
+// 1. Using the richer struct DoFn form and accessing optional arguments.
+// 2. Logging using the Beam log package, even in a distributed environment
+// 3. Testing your Pipeline via passert
//
// To change the runner, specify:
//
-// --runner=YOUR_SELECTED_RUNNER
+// --runner=YOUR_SELECTED_RUNNER
//
// The input file defaults to a public data set containing the text of King
// Lear, by William Shakespeare. You can override it and choose your own input
diff --git a/sdks/go/examples/fhirio/import/import.go b/sdks/go/examples/fhirio/import/import.go
index ceb13eb5a40..aa3db0092ec 100644
--- a/sdks/go/examples/fhirio/import/import.go
+++ b/sdks/go/examples/fhirio/import/import.go
@@ -27,22 +27,23 @@
// to the usual flags for the Dataflow runner.
//
// An example command for executing this pipeline on GCP is as follows:
-// export PROJECT="$(gcloud config get-value project)"
-// export TEMP_LOCATION="gs://MY-BUCKET/temp"
-// export STAGING_LOCATION="gs://MY-BUCKET/staging"
-// export REGION="us-central1"
-// export SOURCE_GCS_LOCATION="gs://MY_BUCKET/path/to/resources/**"
-// export FHIR_STORE_PATH="MY_FHIR_STORE_PATH"
-// cd ./sdks/go
-// go run ./examples/fhirio/import/import.go \
-// --runner=dataflow \
-// --temp_location=$TEMP_LOCATION \
-// --staging_location=$STAGING_LOCATION \
-// --project=$PROJECT \
-// --region=$REGION \
-// --worker_harness_container_image=apache/beam_go_sdk:latest \
-// --sourceGcsLocation=$SOURCE_GCS_LOCATION \
-// --fhirStore=$FHIR_STORE_PATH
+//
+// export PROJECT="$(gcloud config get-value project)"
+// export TEMP_LOCATION="gs://MY-BUCKET/temp"
+// export STAGING_LOCATION="gs://MY-BUCKET/staging"
+// export REGION="us-central1"
+// export SOURCE_GCS_LOCATION="gs://MY_BUCKET/path/to/resources/**"
+// export FHIR_STORE_PATH="MY_FHIR_STORE_PATH"
+// cd ./sdks/go
+// go run ./examples/fhirio/import/import.go \
+// --runner=dataflow \
+// --temp_location=$TEMP_LOCATION \
+// --staging_location=$STAGING_LOCATION \
+// --project=$PROJECT \
+// --region=$REGION \
+// --worker_harness_container_image=apache/beam_go_sdk:latest \
+// --sourceGcsLocation=$SOURCE_GCS_LOCATION \
+// --fhirStore=$FHIR_STORE_PATH
package main
import (
diff --git a/sdks/go/examples/fhirio/read_write_pubsub/read_write_pubsub.go b/sdks/go/examples/fhirio/read_write_pubsub/read_write_pubsub.go
index fcf1108ac5a..4a2333b8c9b 100644
--- a/sdks/go/examples/fhirio/read_write_pubsub/read_write_pubsub.go
+++ b/sdks/go/examples/fhirio/read_write_pubsub/read_write_pubsub.go
@@ -31,24 +31,25 @@
// in addition to the usual flags for the Dataflow runner.
//
// An example command for executing this pipeline on GCP is as follows:
-// export PROJECT="$(gcloud config get-value project)"
-// export TEMP_LOCATION="gs://MY-BUCKET/temp"
-// export STAGING_LOCATION="gs://MY-BUCKET/staging"
-// export REGION="us-central1"
-// export SOURCE_GCS_LOCATION="gs://MY_BUCKET/path/to/resources/**"
-// export FHIR_STORE_PATH="MY_FHIR_STORE_PATH"
-// export PUBSUB_TOPIC="MY_FHIR_STORE_TOPIC"
-// cd ./sdks/go
-// go run ./examples/fhirio/read_write_pubsub/read_write_pubsub.go \
-// --runner=dataflow \
-// --temp_location=$TEMP_LOCATION \
-// --staging_location=$STAGING_LOCATION \
-// --project=$PROJECT \
-// --region=$REGION \
-// --worker_harness_container_image=apache/beam_go_sdk:latest \
-// --sourceGcsLocation=$SOURCE_GCS_LOCATION \
-// --fhirStore=$FHIR_STORE_PATH \
-// --pubsubTopic=$PUBSUB_TOPIC
+//
+// export PROJECT="$(gcloud config get-value project)"
+// export TEMP_LOCATION="gs://MY-BUCKET/temp"
+// export STAGING_LOCATION="gs://MY-BUCKET/staging"
+// export REGION="us-central1"
+// export SOURCE_GCS_LOCATION="gs://MY_BUCKET/path/to/resources/**"
+// export FHIR_STORE_PATH="MY_FHIR_STORE_PATH"
+// export PUBSUB_TOPIC="MY_FHIR_STORE_TOPIC"
+// cd ./sdks/go
+// go run ./examples/fhirio/read_write_pubsub/read_write_pubsub.go \
+// --runner=dataflow \
+// --temp_location=$TEMP_LOCATION \
+// --staging_location=$STAGING_LOCATION \
+// --project=$PROJECT \
+// --region=$REGION \
+// --worker_harness_container_image=apache/beam_go_sdk:latest \
+// --sourceGcsLocation=$SOURCE_GCS_LOCATION \
+// --fhirStore=$FHIR_STORE_PATH \
+// --pubsubTopic=$PUBSUB_TOPIC
package main
import (
diff --git a/sdks/go/examples/kafka/taxi.go b/sdks/go/examples/kafka/taxi.go
index 263280e07cb..6f4532dad93 100644
--- a/sdks/go/examples/kafka/taxi.go
+++ b/sdks/go/examples/kafka/taxi.go
@@ -26,7 +26,7 @@
// a flag, or if none is specified then the SDK will attempt to automatically
// start an appropriate expansion service.
//
-// Setting Up a Kafka Cluster
+// # Setting Up a Kafka Cluster
//
// Setting up a Kafka cluster is more involved than can be covered in this
// example. In order for this example to work, all that is necessary is a Kafka
@@ -34,7 +34,7 @@
// a flag. Some instructions for setting up a single node Kafka cluster in GCE
// can be found here: https://github.com/GoogleCloudPlatform/java-docs-samples/tree/master/dataflow/flex-templates/kafka_to_bigquery
//
-// Running an Expansion Server
+// # Running an Expansion Server
//
// If the automatic expansion service functionality is not available for your
// environment, or if you want improved performance, you will need to start a
@@ -47,48 +47,52 @@
// Numbered release: The expansion service jar is vendored as module
// org.apache.beam:beam-sdks-java-io-expansion-service in Maven Repository.
// This jar can be executed directly with the following command:
-// `java -jar <jar_name> <port_number>`
+//
+// `java -jar <jar_name> <port_number>`
+//
// Development env: This requires that the JAVA_HOME environment variable
// points to your JDK installation. From the root `beam/` directory of the
// Apache Beam repository, the jar can be built (or built and run) with the
// following commands:
-// Build: ./gradlew :sdks:java:io:expansion-service:build
-// Build and Run: ./gradlew :sdks:java:io:expansion-service:runExpansionService -PconstructionService.port=<port_num>
//
-// Running the Example on GCP
+// Build: ./gradlew :sdks:java:io:expansion-service:build
+// Build and Run: ./gradlew :sdks:java:io:expansion-service:runExpansionService -PconstructionService.port=<port_num>
+//
+// # Running the Example on GCP
//
// Running this pipeline requires providing an address for the Expansion Service
// and for the Kafka cluster's bootstrap servers as flags, in addition to the
// usual flags for pipelines.
//
// An example command for executing this pipeline on GCP is as follows:
-// export PROJECT="$(gcloud config get-value project)"
-// export TEMP_LOCATION="gs://MY-BUCKET/temp"
-// export REGION="us-central1"
-// export JOB_NAME="kafka-taxi-`date +%Y%m%d-%H%M%S`"
-// export BOOTSTRAP_SERVERS="123.45.67.89:1234"
-// export EXPANSION_ADDR="localhost:1234"
-// cd ./sdks/go
-// go run ./examples/kafka/taxi.go \
-// --runner=DataflowRunner \
-// --temp_location=$TEMP_LOCATION \
-// --staging_location=$STAGING_LOCATION \
-// --project=$PROJECT \
-// --region=$REGION \
-// --job_name="${JOB_NAME}" \
-// --bootstrap_servers=$BOOTSTRAP_SERVER \
-// --expansion_addr=$EXPANSION_ADDR
-//
-// Running the Example From a Git Clone
+//
+// export PROJECT="$(gcloud config get-value project)"
+// export TEMP_LOCATION="gs://MY-BUCKET/temp"
+// export REGION="us-central1"
+// export JOB_NAME="kafka-taxi-`date +%Y%m%d-%H%M%S`"
+// export BOOTSTRAP_SERVERS="123.45.67.89:1234"
+// export EXPANSION_ADDR="localhost:1234"
+// cd ./sdks/go
+// go run ./examples/kafka/taxi.go \
+// --runner=DataflowRunner \
+// --temp_location=$TEMP_LOCATION \
+// --staging_location=$STAGING_LOCATION \
+// --project=$PROJECT \
+// --region=$REGION \
+// --job_name="${JOB_NAME}" \
+// --bootstrap_servers=$BOOTSTRAP_SERVER \
+// --expansion_addr=$EXPANSION_ADDR
+//
+// # Running the Example From a Git Clone
//
// When running on a development environment, a custom container will likely
// need to be provided for the cross-language SDK. First this will require
// building and pushing the SDK container to container repository, such as
// Docker Hub.
//
-// export DOCKER_ROOT="Your Docker Repository Root"
-// ./gradlew :sdks:java:container:java8:docker -Pdocker-repository-root=$DOCKER_ROOT -Pdocker-tag=latest
-// docker push $DOCKER_ROOT/beam_java8_sdk:latest
+// export DOCKER_ROOT="Your Docker Repository Root"
+// ./gradlew :sdks:java:container:java8:docker -Pdocker-repository-root=$DOCKER_ROOT -Pdocker-tag=latest
+// docker push $DOCKER_ROOT/beam_java8_sdk:latest
//
// For runners in local mode, simply building the container using the default
// values for docker-repository-root and docker-tag will work to have it
@@ -98,8 +102,8 @@
// pipeline with the --sdk_harness_container_image_override flag for Java, or
// --environment_config flag for Go. For example:
//
-// --sdk_harness_container_image_override=".*java.*,${DOCKER_ROOT}/beam_java8_sdk:latest" \
-// --environment_config=${DOCKER_ROOT}/beam_go_sdk:latest
+// --sdk_harness_container_image_override=".*java.*,${DOCKER_ROOT}/beam_java8_sdk:latest" \
+// --environment_config=${DOCKER_ROOT}/beam_go_sdk:latest
package main
import (
diff --git a/sdks/go/examples/large_wordcount/large_wordcount.go b/sdks/go/examples/large_wordcount/large_wordcount.go
index 4f2585021a2..df04b19a383 100644
--- a/sdks/go/examples/large_wordcount/large_wordcount.go
+++ b/sdks/go/examples/large_wordcount/large_wordcount.go
@@ -30,16 +30,16 @@
//
// New Concepts:
//
-// 1. Using a SplittableDoFn transform to read the IOs.
-// 2. Using a Map Side Input to access values for specific keys.
-// 3. Testing your Pipeline via passert and metrics, using Go testing tools.
+// 1. Using a SplittableDoFn transform to read the IOs.
+// 2. Using a Map Side Input to access values for specific keys.
+// 3. Testing your Pipeline via passert and metrics, using Go testing tools.
//
// This example will not be enumerating concepts, but will document them as they
// appear. There may be repetition from previous examples.
//
// To change the runner, specify:
//
-// --runner=YOUR_SELECTED_RUNNER
+// --runner=YOUR_SELECTED_RUNNER
//
// The input file defaults to a public data set containing the text of King
// Lear, by William Shakespeare. You can override it and choose your own input
diff --git a/sdks/go/examples/minimal_wordcount/minimal_wordcount.go b/sdks/go/examples/minimal_wordcount/minimal_wordcount.go
index 5ba96d0a89b..75d8cce43b1 100644
--- a/sdks/go/examples/minimal_wordcount/minimal_wordcount.go
+++ b/sdks/go/examples/minimal_wordcount/minimal_wordcount.go
@@ -27,10 +27,10 @@
//
// Concepts:
//
-// 1. Reading data from text files
-// 2. Specifying 'inline' transforms
-// 3. Counting items in a PCollection
-// 4. Writing data to text files
+// 1. Reading data from text files
+// 2. Specifying 'inline' transforms
+// 3. Counting items in a PCollection
+// 4. Writing data to text files
//
// No arguments are required to run this pipeline. It will be executed with
// the direct runner. You can see the results in the output file named
diff --git a/sdks/go/examples/native_wordcap/wordcap.go b/sdks/go/examples/native_wordcap/wordcap.go
index 9fa2d0e55aa..ee3b1e4e12e 100644
--- a/sdks/go/examples/native_wordcap/wordcap.go
+++ b/sdks/go/examples/native_wordcap/wordcap.go
@@ -15,9 +15,10 @@
// native_wordcap is a toy streaming pipeline that uses PubSub. It
// does the following:
-// (1) create a topic and publish a few messages to it
-// (2) start a streaming pipeline that converts the messages to
-// upper case and logs the result.
+//
+// (1) create a topic and publish a few messages to it
+// (2) start a streaming pipeline that converts the messages to
+// upper case and logs the result.
//
// This is a portable PubSub option and does not need to be run on Dataflow.
package main
diff --git a/sdks/go/examples/streaming_wordcap/wordcap.go b/sdks/go/examples/streaming_wordcap/wordcap.go
index 7b470b62b56..ddd9eab4e5f 100644
--- a/sdks/go/examples/streaming_wordcap/wordcap.go
+++ b/sdks/go/examples/streaming_wordcap/wordcap.go
@@ -15,9 +15,10 @@
// streaming_wordcap is a toy streaming pipeline that uses PubSub. It
// does the following:
-// (1) create a topic and publish a few messages to it
-// (2) start a streaming pipeline that converts the messages to
-// upper case and logs the result.
+//
+// (1) create a topic and publish a few messages to it
+// (2) start a streaming pipeline that converts the messages to
+// upper case and logs the result.
//
// NOTE: it only runs on Dataflow and must be manually cancelled.
package main
diff --git a/sdks/go/examples/stringsplit/stringsplit.go b/sdks/go/examples/stringsplit/stringsplit.go
index 8b38c522120..c786c2e86bb 100644
--- a/sdks/go/examples/stringsplit/stringsplit.go
+++ b/sdks/go/examples/stringsplit/stringsplit.go
@@ -20,7 +20,8 @@
//
// 1. From a command line, navigate to the top-level beam/ directory and run
// the Flink job server:
-// ./gradlew :runners:flink:1.13:job-server:runShadow -Djob-host=localhost -Dflink-master=local
+//
+// ./gradlew :runners:flink:1.13:job-server:runShadow -Djob-host=localhost -Dflink-master=local
//
// 2. The job server is ready to receive jobs once it outputs a log like the
// following: `JobService started on localhost:8099`. Take note of the endpoint
@@ -29,7 +30,8 @@
// 3. While the job server is running in one command line window, create a
// second one in the same directory and run this example with the following
// command, using the endpoint you noted from step 2:
-// go run sdks/go/examples/stringsplit/stringsplit.go --runner=universal --endpoint=localhost:8099
+//
+// go run sdks/go/examples/stringsplit/stringsplit.go --runner=universal --endpoint=localhost:8099
//
// 4. Once the pipeline is complete, the job server can be closed with ctrl+C.
// To check the output of the pipeline, search the job server logs for the
diff --git a/sdks/go/examples/xlang/bigquery/wordcount.go b/sdks/go/examples/xlang/bigquery/wordcount.go
index 5a30423be88..b836efdca49 100644
--- a/sdks/go/examples/xlang/bigquery/wordcount.go
+++ b/sdks/go/examples/xlang/bigquery/wordcount.go
@@ -25,7 +25,7 @@
// service can be provided as a flag, or if none is specified then the SDK will attempt to
// automatically start an appropriate expansion service.
//
-// Running an Expansion Server
+// # Running an Expansion Server
//
// If the automatic expansion service functionality is not available for your environment, or if
// you want improved performance, you will need to start a persistent expansion service. These
@@ -37,44 +37,47 @@
// Numbered release: The expansion service jar is vendored as module
// org.apache.beam:beam-sdks-java-io-google-cloud-platform-expansion-service in Maven Repository.
// This jar can be executed directly with the following command:
-// `java -jar <jar_name> <port_number>`
+//
+// `java -jar <jar_name> <port_number>`
//
// Development env: This requires that the JAVA_HOME environment variable points to your JDK
// installation. From the root `beam/` directory of the Apache Beam repository, the jar can be
// built (or built and run) with the following commands:
-// ./gradlew :sdks:java:io:google-cloud-platform:expansion-service:build
-// ./gradlew :sdks:java:io:google-cloud-platform:expansion-service:runExpansionService -PconstructionService.port=<port_num>
//
-// Running the Example on GCP
+// ./gradlew :sdks:java:io:google-cloud-platform:expansion-service:build
+// ./gradlew :sdks:java:io:google-cloud-platform:expansion-service:runExpansionService -PconstructionService.port=<port_num>
+//
+// # Running the Example on GCP
//
// An example command for executing this pipeline on GCP is as follows:
-// export PROJECT="$(gcloud config get-value project)"
-// export TEMP_LOCATION="gs://MY-BUCKET/temp"
-// export REGION="us-central1"
-// export JOB_NAME="bigquery-wordcount-`date +%Y%m%d-%H%M%S`"
-// export OUTPUT_TABLE="123.45.67.89:1234"
-// export EXPANSION_ADDR="localhost:1234"
-// export OUTPUT_TABLE="project_id:dataset_id.table_id"
-// go run ./sdks/go/examples/kafka/types/types.go \
-// --runner=DataflowRunner \
-// --temp_location=$TEMP_LOCATION \
-// --staging_location=$STAGING_LOCATION \
-// --project=$PROJECT \
-// --region=$REGION \
-// --job_name="${JOB_NAME}" \
-// --bootstrap_servers=$BOOTSTRAP_SERVER \
-// --expansion_addr=$EXPANSION_ADDR \
-// --out_table=$OUTPUT_TABLE
-//
-// Running the Example From a Git Clone
+//
+// export PROJECT="$(gcloud config get-value project)"
+// export TEMP_LOCATION="gs://MY-BUCKET/temp"
+// export REGION="us-central1"
+// export JOB_NAME="bigquery-wordcount-`date +%Y%m%d-%H%M%S`"
+// export OUTPUT_TABLE="123.45.67.89:1234"
+// export EXPANSION_ADDR="localhost:1234"
+// export OUTPUT_TABLE="project_id:dataset_id.table_id"
+// go run ./sdks/go/examples/kafka/types/types.go \
+// --runner=DataflowRunner \
+// --temp_location=$TEMP_LOCATION \
+// --staging_location=$STAGING_LOCATION \
+// --project=$PROJECT \
+// --region=$REGION \
+// --job_name="${JOB_NAME}" \
+// --bootstrap_servers=$BOOTSTRAP_SERVER \
+// --expansion_addr=$EXPANSION_ADDR \
+// --out_table=$OUTPUT_TABLE
+//
+// # Running the Example From a Git Clone
//
// When running on a development environment, a custom container will likely need to be provided
// for the cross-language SDK. First this will require building and pushing the SDK container to
// container repository, such as Docker Hub.
//
-// export DOCKER_ROOT="Your Docker Repository Root"
-// ./gradlew :sdks:java:container:java8:docker -Pdocker-repository-root=$DOCKER_ROOT -Pdocker-tag=latest
-// docker push $DOCKER_ROOT/beam_java8_sdk:latest
+// export DOCKER_ROOT="Your Docker Repository Root"
+// ./gradlew :sdks:java:container:java8:docker -Pdocker-repository-root=$DOCKER_ROOT -Pdocker-tag=latest
+// docker push $DOCKER_ROOT/beam_java8_sdk:latest
//
// For runners in local mode, simply building the container using the default values for
// docker-repository-root and docker-tag will work to have it accessible locally.
@@ -83,8 +86,8 @@
// --sdk_harness_container_image_override flag for Java, or --environment_config flag for Go. For
// example:
//
-// --sdk_harness_container_image_override=".*java.*,${DOCKER_ROOT}/beam_java8_sdk:latest" \
-// --environment_config=${DOCKER_ROOT}/beam_go_sdk:latest
+// --sdk_harness_container_image_override=".*java.*,${DOCKER_ROOT}/beam_java8_sdk:latest" \
+// --environment_config=${DOCKER_ROOT}/beam_go_sdk:latest
package main
import (
diff --git a/sdks/go/examples/yatzy/yatzy.go b/sdks/go/examples/yatzy/yatzy.go
index 592ef7c3136..d2ff5b60594 100644
--- a/sdks/go/examples/yatzy/yatzy.go
+++ b/sdks/go/examples/yatzy/yatzy.go
@@ -62,7 +62,7 @@ func init() {
// is encoded in the shape of the pipeline, which will produce a single
// element of that value. The shape is as follows:
//
-// 0 -> \x.x+1 -> \x.x+1 -> (N times) -> \x.min(x, 6)
+// 0 -> \x.x+1 -> \x.x+1 -> (N times) -> \x.min(x, 6)
//
// The single output will be a number between 1 and 6.
func roll(ctx context.Context, s beam.Scope) beam.PCollection {
diff --git a/sdks/go/pkg/beam/core/funcx/fn.go b/sdks/go/pkg/beam/core/funcx/fn.go
index 89d92529f01..b579cb56d52 100644
--- a/sdks/go/pkg/beam/core/funcx/fn.go
+++ b/sdks/go/pkg/beam/core/funcx/fn.go
@@ -483,8 +483,10 @@ func SubReturns(list []ReturnParam, indices ...int) []ReturnParam {
// The order of present parameters and return values must be as follows:
// func(FnContext?, FnPane?, FnWindow?, FnEventTime?, FnWatermarkEstimator?, FnType?, FnBundleFinalization?, FnRTracker?, FnStateProvider?, (FnValue, SideInput*)?, FnEmit*) (RetEventTime?, RetOutput?, RetError?)
-// where ? indicates 0 or 1, and * indicates any number.
-// and a SideInput is one of FnValue or FnIter or FnReIter
+//
+// where ? indicates 0 or 1, and * indicates any number.
+// and a SideInput is one of FnValue or FnIter or FnReIter
+//
// Note: Fns with inputs must have at least one FnValue as the main input.
func validateOrder(u *Fn) error {
paramState := psStart
diff --git a/sdks/go/pkg/beam/core/funcx/output.go b/sdks/go/pkg/beam/core/funcx/output.go
index 03eebb696aa..24cdb99ead5 100644
--- a/sdks/go/pkg/beam/core/funcx/output.go
+++ b/sdks/go/pkg/beam/core/funcx/output.go
@@ -50,10 +50,9 @@ func IsEmitWithEventTime(t reflect.Type) bool {
// UnfoldEmit returns the parameter types, if an emitter. For example:
//
-// func (int) returns {int}
-// func (string, int) returns {string, int}
-// func (typex.EventTime, int) returns {typex.EventTime, int}
-//
+// func (int) returns {int}
+// func (string, int) returns {string, int}
+// func (typex.EventTime, int) returns {typex.EventTime, int}
func UnfoldEmit(t reflect.Type) ([]reflect.Type, bool) {
types, ok, _ := unfoldEmit(t)
return types, ok
diff --git a/sdks/go/pkg/beam/core/funcx/sideinput.go b/sdks/go/pkg/beam/core/funcx/sideinput.go
index ab289ca3eba..77c968fb10a 100644
--- a/sdks/go/pkg/beam/core/funcx/sideinput.go
+++ b/sdks/go/pkg/beam/core/funcx/sideinput.go
@@ -55,8 +55,8 @@ func IsMalformedIter(t reflect.Type) (bool, error) {
// UnfoldIter returns the parameter types, if a single sweep functional
// iterator. For example:
//
-// func (*int) bool returns {int}
-// func (*string, *int) bool returns {string, int}
+// func (*int) bool returns {int}
+// func (*string, *int) bool returns {string, int}
//
// EventTimes are not allowed in iterator types as per the Beam model
// (see https://github.com/apache/beam/issues/22404) for more
diff --git a/sdks/go/pkg/beam/core/funcx/signature.go b/sdks/go/pkg/beam/core/funcx/signature.go
index d880416f06f..03ee67e0406 100644
--- a/sdks/go/pkg/beam/core/funcx/signature.go
+++ b/sdks/go/pkg/beam/core/funcx/signature.go
@@ -29,7 +29,7 @@ import (
// function types in a group can differ in optional leading arguments and
// trailing returns only. For example, a signature can represent:
//
-// (context.Context?, int, string) -> (bool, error?)
+// (context.Context?, int, string) -> (bool, error?)
//
// where the context arguments and error return are optional. The int and
// string parameters as well as the bool return are mandatory.
@@ -102,8 +102,8 @@ func replace(list []reflect.Type, old, new reflect.Type) []reflect.Type {
// Satisfy returns nil iff the fn can satisfy the signature, respecting
// generics. For example, for
//
-// foo : (context.Context, X) -> bool
-// bar : (int) -> bool
+// foo : (context.Context, X) -> bool
+// bar : (int) -> bool
//
// both would satisfy a signature of (context.Context?, int) -> bool. Only
// "foo" would satisfy (context.Context, string) -> bool and only "bar" would
diff --git a/sdks/go/pkg/beam/core/graph/bind.go b/sdks/go/pkg/beam/core/graph/bind.go
index 1ef421c42c0..28dbc3f4f17 100644
--- a/sdks/go/pkg/beam/core/graph/bind.go
+++ b/sdks/go/pkg/beam/core/graph/bind.go
@@ -33,9 +33,11 @@ import (
//
// For example,
//
-// func (t EventTime, k typex.X, v int, emit func(string, typex.X))
+// func (t EventTime, k typex.X, v int, emit func(string, typex.X))
+//
// or
-// func (context.Context, k typex.X, v int) (string, typex.X, error)
+//
+// func (context.Context, k typex.X, v int) (string, typex.X, error)
//
// are UserFns that may take one or two incoming fulltypes: either KV<X,int>
// or X with a singleton side input of type int. For the purpose of the
@@ -45,18 +47,18 @@ import (
//
// If either was bound to the input type [KV<string,int>], bind would return:
//
-// inbound: [Main: KV<X,int>]
-// outbound: [KV<string,X>]
-// output: [KV<string,string>]
+// inbound: [Main: KV<X,int>]
+// outbound: [KV<string,X>]
+// output: [KV<string,string>]
//
// Note that it propagates the assignment of X to string in the output type.
//
// If either was instead bound to the input fulltypes [float, int], the
// result would be:
//
-// inbound: [Main: X, Singleton: int]
-// outbound: [KV<string,X>]
-// output: [KV<string, float>]
+// inbound: [Main: X, Singleton: int]
+// outbound: [KV<string,X>]
+// output: [KV<string, float>]
//
// Here, the inbound shape and output types are different from before.
func Bind(fn *funcx.Fn, typedefs map[string]reflect.Type, in ...typex.FullType) ([]typex.FullType, []InputKind, []typex.FullType, []typex.FullType, error) {
diff --git a/sdks/go/pkg/beam/core/graph/coder/registry.go b/sdks/go/pkg/beam/core/graph/coder/registry.go
index 7f59588fa40..5eb2abc5f61 100644
--- a/sdks/go/pkg/beam/core/graph/coder/registry.go
+++ b/sdks/go/pkg/beam/core/graph/coder/registry.go
@@ -33,10 +33,10 @@ var (
// Coders are encoder and decoder pairs, and operate around []bytes.
//
// The coder used for a given type follows this ordering:
-// 1. Coders for Known Beam types.
-// 2. Coders registered for specific types
-// 3. Coders registered for interfaces types
-// 4. Default coder (JSON)
+// 1. Coders for Known Beam types.
+// 2. Coders registered for specific types
+// 3. Coders registered for interfaces types
+// 4. Default coder (JSON)
//
// Types of kind Interface, are handled specially by the registry, so they may be iterated
// over to check if element types implement them.
diff --git a/sdks/go/pkg/beam/core/graph/fn.go b/sdks/go/pkg/beam/core/graph/fn.go
index 4350ff97117..4c7f57d1d70 100644
--- a/sdks/go/pkg/beam/core/graph/fn.go
+++ b/sdks/go/pkg/beam/core/graph/fn.go
@@ -57,7 +57,7 @@ func (f *Fn) Name() string {
// DynFn is a generator for dynamically-created functions:
//
-// gen: (name string, t reflect.Type, []byte) -> func : T
+// gen: (name string, t reflect.Type, []byte) -> func : T
//
// where the generated function, fn : T, is re-created at runtime. This concept
// allows serialization of dynamically-generated functions, which do not have a
@@ -415,7 +415,8 @@ func defaultConfig() *config {
// validation. Valid inputs are the package constants of type mainInputs.
//
// Example usage:
-// graph.NewDoFn(fn, graph.NumMainInputs(graph.MainKv))
+//
+// graph.NewDoFn(fn, graph.NumMainInputs(graph.MainKv))
func NumMainInputs(num mainInputs) func(*config) {
return func(cfg *config) {
cfg.numMainIn = num
@@ -427,8 +428,9 @@ func NumMainInputs(num mainInputs) func(*config) {
// validation.
//
// Example usage:
-// var col beam.PCollection
-// graph.NewDoFn(fn, graph.CoGBKMainInput(len(col.Type().Components())))
+//
+// var col beam.PCollection
+// graph.NewDoFn(fn, graph.CoGBKMainInput(len(col.Type().Components())))
func CoGBKMainInput(components int) func(*config) {
return func(cfg *config) {
cfg.numMainIn = mainInputs(components)
@@ -804,11 +806,12 @@ func validateSideInputsNumUnknown(processFnInputs []funcx.FnParam, method *funcx
// requirements for either case.
//
// For a Fn to be an SDF it must:
-// * Implement all the required (non-watermark related) SDF methods.
-// * Include an RTracker parameter in ProcessElement.
+// - Implement all the required (non-watermark related) SDF methods.
+// - Include an RTracker parameter in ProcessElement.
+//
// For a Fn to not be an SDF, it must:
-// * Implement none of the SDF methods.
-// * Not include an RTracker parameter in ProcessElement.
+// - Implement none of the SDF methods.
+// - Not include an RTracker parameter in ProcessElement.
func validateIsSdf(fn *Fn) (bool, error) {
// Store missing method names so we can output them to the user if validation fails.
var missing []string
diff --git a/sdks/go/pkg/beam/core/graph/window/trigger/trigger_test.go b/sdks/go/pkg/beam/core/graph/window/trigger/trigger_test.go
index 2974e47b51e..b1383d7f469 100644
--- a/sdks/go/pkg/beam/core/graph/window/trigger/trigger_test.go
+++ b/sdks/go/pkg/beam/core/graph/window/trigger/trigger_test.go
@@ -5,7 +5,7 @@
// (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
+// 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,
diff --git a/sdks/go/pkg/beam/core/metrics/metrics.go b/sdks/go/pkg/beam/core/metrics/metrics.go
index 6428d868f7d..78a4a852b2f 100644
--- a/sdks/go/pkg/beam/core/metrics/metrics.go
+++ b/sdks/go/pkg/beam/core/metrics/metrics.go
@@ -525,9 +525,10 @@ type SingleResult interface {
}
// Query allows metrics querying with filter. The filter takes the form of predicate function. Example:
-// qr = pr.Metrics().Query(func(mr beam.MetricResult) bool {
-// return sr.Namespace() == test.namespace
-// })
+//
+// qr = pr.Metrics().Query(func(mr beam.MetricResult) bool {
+// return sr.Namespace() == test.namespace
+// })
func (mr Results) Query(f func(SingleResult) bool) QueryResults {
counters := []CounterResult{}
distributions := []DistributionResult{}
diff --git a/sdks/go/pkg/beam/core/runtime/exec/combine_test.go b/sdks/go/pkg/beam/core/runtime/exec/combine_test.go
index 5797fbfa4c0..2457144ed51 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/combine_test.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/combine_test.go
@@ -455,14 +455,14 @@ func constructAndExecutePlan(t *testing.T, us []Unit) {
// mergeFn represents a combine that is just a binary merge, where
//
-// InputT == OutputT == AccumT == int
+// InputT == OutputT == AccumT == int
func mergeFn(a, b int) int {
return a + b
}
// nonBinaryMergeFn represents a combine with a context parameter and an error return, where
//
-// InputT == OutputT == AccumT == int
+// InputT == OutputT == AccumT == int
func nonBinaryMergeFn(ctx context.Context, a, b int) (int, error) {
return a + b, nil
}
@@ -470,8 +470,8 @@ func nonBinaryMergeFn(ctx context.Context, a, b int) (int, error) {
// MyCombine represents a combine with the same Input and Output type (int), but a
// distinct accumulator type (int64).
//
-// InputT == OutputT == int
-// AccumT == int64
+// InputT == OutputT == int
+// AccumT == int64
type MyCombine struct{}
func (*MyCombine) AddInput(a int64, v int) int64 {
@@ -488,9 +488,9 @@ func (*MyCombine) ExtractOutput(a int64) int {
// MyOtherCombine is the same as MyCombine, but has strings extracted as output.
//
-// InputT == int
-// AccumT == int64
-// OutputT == string
+// InputT == int
+// AccumT == int64
+// OutputT == string
type MyOtherCombine struct {
MyCombine // Embedding to re-use the exisitng AddInput and MergeAccumulators implementations
}
@@ -501,9 +501,9 @@ func (*MyOtherCombine) ExtractOutput(a int64) string {
// MyThirdCombine parses strings as Input, and doesn't specify an ExtractOutput
//
-// InputT == string
-// AccumT == int
-// OutputT == int
+// InputT == string
+// AccumT == int
+// OutputT == int
type MyThirdCombine struct{}
func (c *MyThirdCombine) AddInput(a int, s string) (int, error) {
@@ -520,9 +520,9 @@ func (*MyThirdCombine) MergeAccumulators(a, b int) int {
// MyContextCombine is the same as MyCombine, but requires a context parameter.
//
-// InputT == int
-// AccumT == int64
-// OutputT == string
+// InputT == int
+// AccumT == int64
+// OutputT == string
type MyContextCombine struct {
MyCombine // Embedding to re-use the exisitng AddInput implementations
}
@@ -533,9 +533,9 @@ func (*MyContextCombine) MergeAccumulators(_ context.Context, a, b int64) int64
// MyErrorCombine is the same as MyCombine, but may return an error.
//
-// InputT == int
-// AccumT == int64
-// OutputT == string
+// InputT == int
+// AccumT == int64
+// OutputT == string
type MyErrorCombine struct {
MyCombine // Embedding to re-use the exisitng AddInput implementations
}
diff --git a/sdks/go/pkg/beam/core/runtime/exec/fn_test.go b/sdks/go/pkg/beam/core/runtime/exec/fn_test.go
index 00733f6d0c0..21ec37dc98f 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/fn_test.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/fn_test.go
@@ -708,11 +708,11 @@ func (c *callerInt) Call1x1(a0 interface{}) interface{} {
// * Implicit or Explicit method Receiver
// * Pre-wrapped values and pre-allocated slices.
// * Invocations along the following ways
-// * Indirect via extracting from a reflect.Value.Interface()
-// * Reflect Package (reflect.Value.Call())
-// * Beam's reflectx.Func, and reflectx.FuncNxM interfaces
-// * Beam's default reflection based reflectx.Func shim
-// * A Type assertion specialized reflectx.Func shim
+// - Indirect via extracting from a reflect.Value.Interface()
+// - Reflect Package (reflect.Value.Call())
+// - Beam's reflectx.Func, and reflectx.FuncNxM interfaces
+// - Beam's default reflection based reflectx.Func shim
+// - A Type assertion specialized reflectx.Func shim
//
// The Implicit or Explicit method receiver difference exists because
// Go's reflect package treats the two cases different, and there are
diff --git a/sdks/go/pkg/beam/core/runtime/exec/sdf.go b/sdks/go/pkg/beam/core/runtime/exec/sdf.go
index 6c3fe5a12f7..f1afade98f8 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/sdf.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/sdf.go
@@ -80,15 +80,15 @@ func (n *PairWithRestriction) StartBundle(ctx context.Context, id string, data D
//
// Output Diagram:
//
-// *FullValue {
-// Elm: *FullValue (original input)
-// Elm2: *FullValue {
-// Elm: Restriction
-// Elm2: Watermark estimator state
-// }
-// Windows
-// Timestamps
-// }
+// *FullValue {
+// Elm: *FullValue (original input)
+// Elm2: *FullValue {
+// Elm: Restriction
+// Elm2: Watermark estimator state
+// }
+// Windows
+// Timestamps
+// }
func (n *PairWithRestriction) ProcessElement(ctx context.Context, elm *FullValue, values ...ReStream) error {
rest := n.inv.Invoke(elm)
output := FullValue{Elm: elm, Elm2: &FullValue{Elm: rest, Elm2: n.iwesInv.Invoke(rest, elm)}, Timestamp: elm.Timestamp, Windows: elm.Windows}
@@ -158,15 +158,15 @@ func (n *SplitAndSizeRestrictions) StartBundle(ctx context.Context, id string, d
//
// Input Diagram:
//
-// *FullValue {
-// Elm: *FullValue (original input)
-// Elm2: *FullValue {
-// Elm: Restriction
-// Elm2: Watermark estimator state
-// }
-// Windows
-// Timestamps
-// }
+// *FullValue {
+// Elm: *FullValue (original input)
+// Elm2: *FullValue {
+// Elm: Restriction
+// Elm2: Watermark estimator state
+// }
+// Windows
+// Timestamps
+// }
//
// ProcessElement splits the given restriction into one or more restrictions and
// then sizes each. The outputs are in the structure <<elem, <restriction, watermark estimator state>>, size>
@@ -175,18 +175,18 @@ func (n *SplitAndSizeRestrictions) StartBundle(ctx context.Context, id string, d
//
// Output Diagram:
//
-// *FullValue {
-// Elm: *FullValue {
-// Elm: *FullValue (original input)
-// Elm2: *FullValue {
-// Elm: Restriction
-// Elm2: Watermark estimator state
-// }
-// }
-// Elm2: float64 (size)
-// Windows
-// Timestamps
-// }
+// *FullValue {
+// Elm: *FullValue {
+// Elm: *FullValue (original input)
+// Elm2: *FullValue {
+// Elm: Restriction
+// Elm2: Watermark estimator state
+// }
+// }
+// Elm2: float64 (size)
+// Windows
+// Timestamps
+// }
func (n *SplitAndSizeRestrictions) ProcessElement(ctx context.Context, elm *FullValue, values ...ReStream) error {
rest := elm.Elm2.(*FullValue).Elm
ws := elm.Elm2.(*FullValue).Elm2
@@ -281,32 +281,34 @@ func (n *TruncateSizedRestriction) StartBundle(ctx context.Context, id string, d
// ProcessElement gets input elm as:
// Input Diagram:
-// *FullValue {
-// Elm: *FullValue {
-// Elm: *FullValue (original input)
-// Elm2: *FullValue {
-// Elm: Restriction
-// Elm2: Watermark estimator state
-// }
-// }
-// Elm2: float64 (size)
-// Windows
-// Timestamps
-// }
+//
+// *FullValue {
+// Elm: *FullValue {
+// Elm: *FullValue (original input)
+// Elm2: *FullValue {
+// Elm: Restriction
+// Elm2: Watermark estimator state
+// }
+// }
+// Elm2: float64 (size)
+// Windows
+// Timestamps
+// }
//
// Output Diagram:
-// *FullValue {
-// Elm: *FullValue {
-// Elm: *FullValue (original input)
-// Elm2: *FullValue {
-// Elm: Restriction
-// Elm2: Watermark estimator state
-// }
-// }
-// Elm2: float64 (size)
-// Windows
-// Timestamps
-// }
+//
+// *FullValue {
+// Elm: *FullValue {
+// Elm: *FullValue (original input)
+// Elm2: *FullValue {
+// Elm: Restriction
+// Elm2: Watermark estimator state
+// }
+// }
+// Elm2: float64 (size)
+// Windows
+// Timestamps
+// }
func (n *TruncateSizedRestriction) ProcessElement(ctx context.Context, elm *FullValue, values ...ReStream) error {
mainElm := elm.Elm.(*FullValue)
inp := mainElm.Elm
@@ -454,18 +456,18 @@ func (n *ProcessSizedElementsAndRestrictions) StartBundle(ctx context.Context, i
//
// Input Diagram:
//
-// *FullValue {
-// Elm: *FullValue {
-// Elm: *FullValue (KV input) or InputType (single-element input)
-// Elm2: *FullValue {
-// Elm: Restriction
-// Elm2: Watermark estimator state
-// }
-// }
-// Elm2: float64 (size)
-// Windows
-// Timestamps
-// }
+// *FullValue {
+// Elm: *FullValue {
+// Elm: *FullValue (KV input) or InputType (single-element input)
+// Elm2: *FullValue {
+// Elm: Restriction
+// Elm2: Watermark estimator state
+// }
+// }
+// Elm2: float64 (size)
+// Windows
+// Timestamps
+// }
//
// ProcessElement then creates a restriction tracker from the stored restriction
// and processes each element using the underlying ParDo and adding the
@@ -732,14 +734,14 @@ func (n *ProcessSizedElementsAndRestrictions) singleWindowSplit(f float64, pWeSt
// at all.
//
// Therefore, when such a split happens, the split result consists of:
-// 1. A primary containing the unsplit restriction and a subset of windows that
-// fall within the primary.
-// 2. (Optional) A second primary containing the primary half of a split
-// restriction and the single window it was split in.
-// 3. A residual containing the unsplit restriction and a subset of windows that
-// fall within the residual.
-// 4. (Optional) A second residual containing the residual half of a split
-// restriction and the window it was split in (same window as primary half).
+// 1. A primary containing the unsplit restriction and a subset of windows that
+// fall within the primary.
+// 2. (Optional) A second primary containing the primary half of a split
+// restriction and the single window it was split in.
+// 3. A residual containing the unsplit restriction and a subset of windows that
+// fall within the residual.
+// 4. (Optional) A second residual containing the residual half of a split
+// restriction and the window it was split in (same window as primary half).
//
// The current implementation does not split restrictions outside of the current
// RTracker (i.e. the current window). Otherwise, the split will occur at the
diff --git a/sdks/go/pkg/beam/core/runtime/exec/sdf_invokers_test.go b/sdks/go/pkg/beam/core/runtime/exec/sdf_invokers_test.go
index ba33113bcb9..e68dc71eb88 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/sdf_invokers_test.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/sdf_invokers_test.go
@@ -449,7 +449,6 @@ func TestInvokes(t *testing.T) {
// called with it. When using VetRestriction, the SDF methods it's used in
// should pass it as a pointer so the method can make changes to the restriction
// even if it doesn't output one directly (such as RestrictionSize).
-//
type VetRestriction struct {
// An identifier to differentiate restrictions on the same elements. When
// split, a suffix in the form of ".#" is appended to this ID.
diff --git a/sdks/go/pkg/beam/core/runtime/xlangx/registry.go b/sdks/go/pkg/beam/core/runtime/xlangx/registry.go
index 47f98023695..1957cc61bb9 100644
--- a/sdks/go/pkg/beam/core/runtime/xlangx/registry.go
+++ b/sdks/go/pkg/beam/core/runtime/xlangx/registry.go
@@ -36,8 +36,10 @@ var defaultReg = newRegistry()
// replace calls to a Beam ExpansionService.
//
// Then, expansion addresses of the forms
-// "<namespace>" or
-// "<namespace>:<configuration>"
+//
+// "<namespace>" or
+// "<namespace>:<configuration>"
+//
// can be used with beam.CrossLanguage. Any configuration after the separator is
// provided to the HandlerFunc on call for the handler func to use at it's leisure.
func RegisterHandler(namespace string, handler HandlerFunc) {
@@ -51,8 +53,9 @@ func RegisterHandler(namespace string, handler HandlerFunc) {
// or be a namespaced handler registered with RegisterHandler.
//
// When the expansion address is for a handler, it may take the forms
-// "<namespace>" or
-// "<namespace>:<configuration>"
+//
+// "<namespace>" or
+// "<namespace>:<configuration>"
func RegisterOverrideForUrn(urn, expansionAddr string) {
if err := defaultReg.RegisterOverrideForUrn(urn, expansionAddr); err != nil {
panic(err)
diff --git a/sdks/go/pkg/beam/core/runtime/xlangx/resolve.go b/sdks/go/pkg/beam/core/runtime/xlangx/resolve.go
index 32275bf8585..aabe6eface4 100644
--- a/sdks/go/pkg/beam/core/runtime/xlangx/resolve.go
+++ b/sdks/go/pkg/beam/core/runtime/xlangx/resolve.go
@@ -62,10 +62,11 @@ func defaultJoinFn(path, name string) string {
// transforms, but with some additional configuration to behavior. By default,
// this function performs the following steps for each cross-language transform
// in the list of edges:
-// 1. Retrieves a list of dependencies needed from the expansion service.
-// 2. Retrieves each dependency as an artifact and stages it to a default
-// local filepath.
-// 3. Adds the dependencies to the transform's stored environment proto.
+// 1. Retrieves a list of dependencies needed from the expansion service.
+// 2. Retrieves each dependency as an artifact and stages it to a default
+// local filepath.
+// 3. Adds the dependencies to the transform's stored environment proto.
+//
// The changes that can be configured are documented in ResolveConfig.
//
// This returns a map of "local path" to "sdk path". By default these are
diff --git a/sdks/go/pkg/beam/core/util/ioutilx/read.go b/sdks/go/pkg/beam/core/util/ioutilx/read.go
index c70cbfae2c8..dbb870dcefc 100644
--- a/sdks/go/pkg/beam/core/util/ioutilx/read.go
+++ b/sdks/go/pkg/beam/core/util/ioutilx/read.go
@@ -89,6 +89,7 @@ func ReadUnsafe(r io.Reader, b []byte) (int, error) {
// compiles down to zero instructions.
// USE CAREFULLY!
// This was copied from the runtime; see issues 23382 and 7921.
+//
//go:nosplit
func noescape(p unsafe.Pointer) unsafe.Pointer {
x := uintptr(p)
diff --git a/sdks/go/pkg/beam/forward.go b/sdks/go/pkg/beam/forward.go
index d855b08a412..ea437c77687 100644
--- a/sdks/go/pkg/beam/forward.go
+++ b/sdks/go/pkg/beam/forward.go
@@ -79,11 +79,11 @@ func RegisterFunction(fn interface{}) {
// RegisterDoFn will panic if the argument type is not a DoFn.
//
// Usage:
-// func init() {
-// beam.RegisterDoFn(FunctionalDoFn)
-// beam.RegisterDoFn(reflect.TypeOf((*StructuralDoFn)(nil)).Elem())
-// }
//
+// func init() {
+// beam.RegisterDoFn(FunctionalDoFn)
+// beam.RegisterDoFn(reflect.TypeOf((*StructuralDoFn)(nil)).Elem())
+// }
func RegisterDoFn(dofn interface{}) {
genx.RegisterDoFn(dofn)
}
@@ -100,10 +100,10 @@ func RegisterInit(hook func()) {
// Must be called prior to beam.Init(), preferably in an init() function.
//
// The coder used for a given type follows this ordering:
-// 1. Coders for Known Beam types.
-// 2. Coders registered for specific types
-// 3. Coders registered for interfaces types
-// 4. Default coder (JSON)
+// 1. Coders for Known Beam types.
+// 2. Coders registered for specific types
+// 3. Coders registered for interfaces types
+// 4. Default coder (JSON)
//
// Coders for interface types are iterated over to check if a type
// satisfies them, and the most recent one registered will be used.
@@ -115,17 +115,17 @@ func RegisterInit(hook func()) {
//
// Supported Encoder Signatures
//
-// func(T) []byte
-// func(reflect.Type, T) []byte
-// func(T) ([]byte, error)
-// func(reflect.Type, T) ([]byte, error)
+// func(T) []byte
+// func(reflect.Type, T) []byte
+// func(T) ([]byte, error)
+// func(reflect.Type, T) ([]byte, error)
//
// Supported Decoder Signatures
//
-// func([]byte) T
-// func(reflect.Type, []byte) T
-// func([]byte) (T, error)
-// func(reflect.Type, []byte) (T, error)
+// func([]byte) T
+// func(reflect.Type, []byte) T
+// func([]byte) (T, error)
+// func(reflect.Type, []byte) (T, error)
//
// where T is the matching user type.
func RegisterCoder(t reflect.Type, encoder, decoder interface{}) {
diff --git a/sdks/go/pkg/beam/gbk.go b/sdks/go/pkg/beam/gbk.go
index 2f41db9ff99..d2d5d70c55f 100644
--- a/sdks/go/pkg/beam/gbk.go
+++ b/sdks/go/pkg/beam/gbk.go
@@ -107,8 +107,8 @@ func TryCoGroupByKey(s Scope, cols ...PCollection) (PCollection, error) {
// while the output file is written with a smaller amount of
// parallelism, using the following pattern:
//
-// pc := bigHairyComputationNeedingParallelism(scope) // PCollection<string>
-// resharded := beam.Reshuffle(scope, pc) // PCollection<string>
+// pc := bigHairyComputationNeedingParallelism(scope) // PCollection<string>
+// resharded := beam.Reshuffle(scope, pc) // PCollection<string>
//
// Another use case is when one has a non-deterministic DoFn followed by one
// that performs externally-visible side effects. Inserting a Reshuffle
diff --git a/sdks/go/pkg/beam/internal/errors/errors.go b/sdks/go/pkg/beam/internal/errors/errors.go
index 3fc1bfbae3e..48b6e7c6ed5 100644
--- a/sdks/go/pkg/beam/internal/errors/errors.go
+++ b/sdks/go/pkg/beam/internal/errors/errors.go
@@ -125,12 +125,12 @@ func getTop(e error) string {
// The presence or lack of certain fields implicitly indicates some details
// about the error.
//
-// * If no cause is present it indicates that this instance is the original
-// error, and the message is assumed to be present.
-// * If both message and context are present, the context describes this error,
-// not the cause of this error.
-// * top is always propogated up from the cause. If it's empty that means that
-// it was never set on any error in the sequence.
+// - If no cause is present it indicates that this instance is the original
+// error, and the message is assumed to be present.
+// - If both message and context are present, the context describes this error,
+// not the cause of this error.
+// - top is always propogated up from the cause. If it's empty that means that
+// it was never set on any error in the sequence.
type beamError struct {
cause error // The error being wrapped. If nil then this is the first error.
context string // Adds additional context to this error and any following.
diff --git a/sdks/go/pkg/beam/io/databaseio/mapper.go b/sdks/go/pkg/beam/io/databaseio/mapper.go
index c327a07ff66..0e594a1854c 100644
--- a/sdks/go/pkg/beam/io/databaseio/mapper.go
+++ b/sdks/go/pkg/beam/io/databaseio/mapper.go
@@ -26,10 +26,10 @@ import (
"github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
)
-//rowMapper represents a record mapper
+// rowMapper represents a record mapper
type rowMapper func(value reflect.Value) ([]interface{}, error)
-//newQueryMapper creates a new record mapped
+// newQueryMapper creates a new record mapped
func newQueryMapper(columns []string, columnTypes []*sql.ColumnType, recordType reflect.Type) (rowMapper, error) {
val := reflect.New(recordType).Interface()
if _, isLoader := val.(MapLoader); isLoader {
@@ -40,7 +40,7 @@ func newQueryMapper(columns []string, columnTypes []*sql.ColumnType, recordType
return nil, errors.Errorf("unsupported type %s", recordType)
}
-//newQueryStructMapper creates a new record mapper for supplied struct type
+// newQueryStructMapper creates a new record mapper for supplied struct type
func newQueryStructMapper(columns []string, recordType reflect.Type) (rowMapper, error) {
mappedFieldIndex, err := mapFields(columns, recordType)
if err != nil {
@@ -57,7 +57,7 @@ func newQueryStructMapper(columns []string, recordType reflect.Type) (rowMapper,
return mapper, nil
}
-//newQueryStructMapper creates a new record mapper for supplied struct type
+// newQueryStructMapper creates a new record mapper for supplied struct type
func newQueryLoaderMapper(columns []string, columnTypes []*sql.ColumnType) (rowMapper, error) {
var record = make([]interface{}, len(columns))
var valueProviders = make([]func(index int, values []interface{}), len(columns))
@@ -112,7 +112,7 @@ func newQueryLoaderMapper(columns []string, columnTypes []*sql.ColumnType) (rowM
return mapper, nil
}
-//newQueryMapper creates a new record mapped
+// newQueryMapper creates a new record mapped
func newWriterRowMapper(columns []string, recordType reflect.Type) (rowMapper, error) {
mappedFieldIndex, err := mapFields(columns, recordType)
if err != nil {
diff --git a/sdks/go/pkg/beam/io/databaseio/util.go b/sdks/go/pkg/beam/io/databaseio/util.go
index 97d098cb073..b88429ee675 100644
--- a/sdks/go/pkg/beam/io/databaseio/util.go
+++ b/sdks/go/pkg/beam/io/databaseio/util.go
@@ -24,7 +24,7 @@ import (
"github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
)
-//mapFields maps column into field index in record type
+// mapFields maps column into field index in record type
func mapFields(columns []string, recordType reflect.Type) ([]int, error) {
var indexedFields = map[string]int{}
for i := 0; i < recordType.NumField(); i++ {
diff --git a/sdks/go/pkg/beam/io/datastoreio/datastore.go b/sdks/go/pkg/beam/io/datastoreio/datastore.go
index 05b43d0a831..f4391059aef 100644
--- a/sdks/go/pkg/beam/io/datastoreio/datastore.go
+++ b/sdks/go/pkg/beam/io/datastoreio/datastore.go
@@ -56,7 +56,6 @@ func init() {
// returns a PCollection<t>. You must also register your type with runtime.RegisterType which allows you to implement
// datastore.PropertyLoadSaver
//
-//
// Example:
// type Item struct {}
// itemKey = runtime.RegisterType(reflect.TypeOf((*Item)(nil)).Elem())
diff --git a/sdks/go/pkg/beam/io/parquetio/parquetio.go b/sdks/go/pkg/beam/io/parquetio/parquetio.go
index c088fc24b28..cad1423de0e 100644
--- a/sdks/go/pkg/beam/io/parquetio/parquetio.go
+++ b/sdks/go/pkg/beam/io/parquetio/parquetio.go
@@ -38,15 +38,16 @@ func init() {
// Read reads a set of files and returns lines as a PCollection<elem>
// based on type of a parquetStruct (struct with parquet tags).
// For example:
-// type Student struct {
-// Name string `parquet:"name=name, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
-// Age int32 `parquet:"name=age, type=INT32, encoding=PLAIN"`
-// Id int64 `parquet:"name=id, type=INT64"`
-// Weight float32 `parquet:"name=weight, type=FLOAT"`
-// Sex bool `parquet:"name=sex, type=BOOLEAN"`
-// Day int32 `parquet:"name=day, type=INT32, convertedtype=DATE"`
-// Ignored int32 //without parquet tag and won't write
-// }
+//
+// type Student struct {
+// Name string `parquet:"name=name, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
+// Age int32 `parquet:"name=age, type=INT32, encoding=PLAIN"`
+// Id int64 `parquet:"name=id, type=INT64"`
+// Weight float32 `parquet:"name=weight, type=FLOAT"`
+// Sex bool `parquet:"name=sex, type=BOOLEAN"`
+// Day int32 `parquet:"name=day, type=INT32, convertedtype=DATE"`
+// Ignored int32 //without parquet tag and won't write
+// }
func Read(s beam.Scope, glob string, t reflect.Type) beam.PCollection {
s = s.Scope("parquetio.Read")
filesystem.ValidateScheme(glob)
@@ -125,15 +126,16 @@ func (a *parquetReadFn) ProcessElement(ctx context.Context, filename string, emi
// Write writes a PCollection<parquetStruct> to .parquet file.
// Write expects a type t of struct with parquet tags
// For example:
-// type Student struct {
-// Name string `parquet:"name=name, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
-// Age int32 `parquet:"name=age, type=INT32, encoding=PLAIN"`
-// Id int64 `parquet:"name=id, type=INT64"`
-// Weight float32 `parquet:"name=weight, type=FLOAT"`
-// Sex bool `parquet:"name=sex, type=BOOLEAN"`
-// Day int32 `parquet:"name=day, type=INT32, convertedtype=DATE"`
-// Ignored int32 //without parquet tag and won't write
-// }
+//
+// type Student struct {
+// Name string `parquet:"name=name, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
+// Age int32 `parquet:"name=age, type=INT32, encoding=PLAIN"`
+// Id int64 `parquet:"name=id, type=INT64"`
+// Weight float32 `parquet:"name=weight, type=FLOAT"`
+// Sex bool `parquet:"name=sex, type=BOOLEAN"`
+// Day int32 `parquet:"name=day, type=INT32, convertedtype=DATE"`
+// Ignored int32 //without parquet tag and won't write
+// }
func Write(s beam.Scope, filename string, t reflect.Type, col beam.PCollection) {
s = s.Scope("parquetio.Write")
filesystem.ValidateScheme(filename)
diff --git a/sdks/go/pkg/beam/io/synthetic/source.go b/sdks/go/pkg/beam/io/synthetic/source.go
index d80a89c931c..ff55a174ff2 100644
--- a/sdks/go/pkg/beam/io/synthetic/source.go
+++ b/sdks/go/pkg/beam/io/synthetic/source.go
@@ -53,10 +53,10 @@ func init() {
// The recommended way to create SourceConfigs is via the SourceConfigBuilder.
// Usage example:
//
-// cfgs := beam.Create(s,
-// synthetic.DefaultSourceConfig().NumElements(1000).Build(),
-// synthetic.DefaultSourceConfig().NumElements(5000).InitialSplits(2).Build())
-// src := synthetic.Source(s, cfgs)
+// cfgs := beam.Create(s,
+// synthetic.DefaultSourceConfig().NumElements(1000).Build(),
+// synthetic.DefaultSourceConfig().NumElements(5000).InitialSplits(2).Build())
+// src := synthetic.Source(s, cfgs)
func Source(s beam.Scope, col beam.PCollection) beam.PCollection {
s = s.Scope("synthetic.Source")
@@ -73,8 +73,8 @@ func Source(s beam.Scope, col beam.PCollection) beam.PCollection {
// The recommended way to create SourceConfigs are via the SourceConfigBuilder.
// Usage example:
//
-// src := synthetic.SourceSingle(s,
-// synthetic.DefaultSourceConfig().NumElements(5000).InitialSplits(2).Build())
+// src := synthetic.SourceSingle(s,
+// synthetic.DefaultSourceConfig().NumElements(5000).InitialSplits(2).Build())
func SourceSingle(s beam.Scope, cfg SourceConfig) beam.PCollection {
s = s.Scope("synthetic.Source")
@@ -161,7 +161,7 @@ func (fn *sourceFn) ProcessElement(rt *sdf.LockRTracker, config SourceConfig, em
//
// Usage example:
//
-// cfg := synthetic.DefaultSourceConfig().NumElements(5000).InitialSplits(2).Build()
+// cfg := synthetic.DefaultSourceConfig().NumElements(5000).InitialSplits(2).Build()
type SourceConfigBuilder struct {
cfg SourceConfig
}
@@ -279,12 +279,13 @@ func (b *SourceConfigBuilder) Build() SourceConfig {
// contains unknown object keys.
//
// An example of valid JSON object:
-// {
-// "num_records": 5,
-// "key_size": 5,
-// "value_size": 5,
-// "num_hot_keys": 5,
-// }
+//
+// {
+// "num_records": 5,
+// "key_size": 5,
+// "value_size": 5,
+// "num_hot_keys": 5,
+// }
func (b *SourceConfigBuilder) BuildFromJSON(jsonData []byte) SourceConfig {
decoder := json.NewDecoder(bytes.NewReader(jsonData))
decoder.DisallowUnknownFields()
diff --git a/sdks/go/pkg/beam/io/synthetic/step.go b/sdks/go/pkg/beam/io/synthetic/step.go
index d654a2bbe34..b31e2aab358 100644
--- a/sdks/go/pkg/beam/io/synthetic/step.go
+++ b/sdks/go/pkg/beam/io/synthetic/step.go
@@ -44,8 +44,8 @@ func init() {
// The recommended way to create StepConfigs is via the StepConfigBuilder.
// Usage example:
//
-// cfg := synthetic.DefaultStepConfig().OutputPerInput(10).FilterRatio(0.5).Build()
-// step := synthetic.Step(s, cfg, input)
+// cfg := synthetic.DefaultStepConfig().OutputPerInput(10).FilterRatio(0.5).Build()
+// step := synthetic.Step(s, cfg, input)
func Step(s beam.Scope, cfg StepConfig, col beam.PCollection) beam.PCollection {
s = s.Scope("synthetic.Step")
if cfg.Splittable {
@@ -147,7 +147,7 @@ func (fn *sdfStepFn) ProcessElement(rt *sdf.LockRTracker, key, val []byte, emit
//
// Usage example:
//
-// cfg := synthetic.DefaultStepConfig().OutputPerInput(10).FilterRatio(0.5).Build()
+// cfg := synthetic.DefaultStepConfig().OutputPerInput(10).FilterRatio(0.5).Build()
type StepConfigBuilder struct {
cfg StepConfig
}
diff --git a/sdks/go/pkg/beam/io/xlang/bigqueryio/bigquery.go b/sdks/go/pkg/beam/io/xlang/bigqueryio/bigquery.go
index 6bfc8939106..ee34ac6de65 100644
--- a/sdks/go/pkg/beam/io/xlang/bigqueryio/bigquery.go
+++ b/sdks/go/pkg/beam/io/xlang/bigqueryio/bigquery.go
@@ -17,7 +17,7 @@
// (https://cloud.google.com/bigquery). These transforms only work on runners that support
// cross-language transforms.
//
-// Setup
+// # Setup
//
// Transforms specified here are cross-language transforms implemented in a
// different SDK (listed below). During pipeline construction, the Go SDK will
@@ -37,28 +37,30 @@
//
// Current supported SDKs, including expansion service modules and reference
// documentation:
-// * Java
-// - Vendored Module: beam-sdks-java-extensions-schemaio-expansion-service
-// - Run via Gradle: ./gradlew :sdks:java:extensions:schemaio-expansion-service:runExpansionService
-// - Reference Class: org.apache.beam.sdk.io.gcp.bigquery.BigQuerySchemaIOProvider and
-// org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO
//
-// Type Conversions
+// Java:
+// - Vendored Module: beam-sdks-java-extensions-schemaio-expansion-service
+// - Run via Gradle: ./gradlew :sdks:java:extensions:schemaio-expansion-service:runExpansionService
+// - Reference Class: org.apache.beam.sdk.io.gcp.bigquery.BigQuerySchemaIOProvider and
+// org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO
+//
+// # Type Conversions
//
// Elements are read from and written to BigQuery by first converting to a Beam schema Row type
// before converting to BigQuery compatible types. The following table lists all BigQuery types
// currently supported, and how they convert to Beam schema and Go types.
-// +----------------------------+------------------+-----------------+
-// | BigQuery Standard SQL Type | Beam Schema Type | Go Type |
-// +----------------------------+------------------+-----------------+
-// | BOOLEAN | BOOLEAN | bool |
-// | INT64 | INT64 | int64 |
-// | FLOAT64 | DOUBLE | float64 |
-// | BYTES | BYTES | []byte |
-// | STRING | STRING | string |
-// | ARRAY | ARRAY | Special: slice |
-// | STRUCT | ROW | Special: struct |
-// +----------------------------+------------------+-----------------+
+//
+// +----------------------------+------------------+-----------------+
+// | BigQuery Standard SQL Type | Beam Schema Type | Go Type |
+// +----------------------------+------------------+-----------------+
+// | BOOLEAN | BOOLEAN | bool |
+// | INT64 | INT64 | int64 |
+// | FLOAT64 | DOUBLE | float64 |
+// | BYTES | BYTES | []byte |
+// | STRING | STRING | string |
+// | ARRAY | ARRAY | Special: slice |
+// | STRUCT | ROW | Special: struct |
+// +----------------------------+------------------+-----------------+
//
// Array types are inferred from slice fields. For example, []int64 is equivalent to BigQuery's
// ARRAY<INT64>. Struct types are inferred from nested structs in Go.
@@ -69,11 +71,12 @@
// in SQL, as in the table above).
//
// Example of BigQuery fields with modes:
-// field1 *int64 // Nullable INT64
-// field2 int64 // Required INT64
-// field3 []int64 // Repeated INT64
//
-// Note On Documentation
+// field1 *int64 // Nullable INT64
+// field2 int64 // Required INT64
+// field3 []int64 // Repeated INT64
+//
+// # Note On Documentation
//
// This cross-language implementation relies on the behavior of external SDKs. In order to keep
// documentation up-to-date and consistent, BigQuery functionality will not be described in detail
@@ -131,12 +134,13 @@ type bigQueryConfig struct {
// within Read's function signature.
//
// Example:
-// expansionAddr := "localhost:1234"
-// table := "project_id:dataset_id.table_id"
-// outType := reflect.TypeOf((*Foo)(nil)).Elem()
-// pcol := bigqueryio.Read(s, outType,
-// bigqueryio.FromTable(table),
-// bigqueryio.ReadExpansionAddr(expansionAddr))
+//
+// expansionAddr := "localhost:1234"
+// table := "project_id:dataset_id.table_id"
+// outType := reflect.TypeOf((*Foo)(nil)).Elem()
+// pcol := bigqueryio.Read(s, outType,
+// bigqueryio.FromTable(table),
+// bigqueryio.ReadExpansionAddr(expansionAddr))
func Read(s beam.Scope, elmT reflect.Type, opts ...readOption) beam.PCollection {
s = s.Scope("bigqueryio.Read")
@@ -224,11 +228,12 @@ func WithQueryLocation(location string) readOption {
// function within Write's function signature.
//
// Example:
-// expansionAddr := "localhost:1234"
-// table := "project_id:dataset_id.table_id"
-// pcol := bigqueryio.Write(s, table, input,
-// bigqueryio.CreateDisposition(bigqueryio.CreateIfNeeded),
-// bigqueryio.WriteExpansionAddr(expansionAddr))
+//
+// expansionAddr := "localhost:1234"
+// table := "project_id:dataset_id.table_id"
+// pcol := bigqueryio.Write(s, table, input,
+// bigqueryio.CreateDisposition(bigqueryio.CreateIfNeeded),
+// bigqueryio.WriteExpansionAddr(expansionAddr))
func Write(s beam.Scope, table string, col beam.PCollection, opts ...writeOption) {
s = s.Scope("bigqueryio.Write")
diff --git a/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go b/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go
index 5c3a1c6faec..eb43fd74175 100644
--- a/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go
+++ b/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go
@@ -17,7 +17,7 @@
// (http://kafka.apache.org/). These transforms only work on runners that
// support cross-language transforms.
//
-// Setup
+// # Setup
//
// Transforms specified here are cross-language transforms implemented in a
// different SDK (listed below). During pipeline construction, the Go SDK will
@@ -33,12 +33,13 @@
// released to the Maven repository as modules. For development versions of
// Beam, it is recommended to build and run it from source using Gradle.
//
-// Current supported SDKs including expansion service modules
-// * Java
-// - Vendored Module: beam-sdks-java-io-debezium-expansion-service
-// - Run via Gradle: ./gradlew :sdks:java:io:debezium:expansion-service:shadowJar
-// java -jar <path-to-debezium-jar> <port>
-// - Reference Class: org.apache.beam.io.debezium.DebeziumIO
+// # Current supported SDKs including expansion service modules
+//
+// Java:
+// - Vendored Module: beam-sdks-java-io-debezium-expansion-service
+// - Run via Gradle: ./gradlew :sdks:java:io:debezium:expansion-service:shadowJar
+// java -jar <path-to-debezium-jar> <port>
+// - Reference Class: org.apache.beam.io.debezium.DebeziumIO
package debeziumio
import (
@@ -92,14 +93,15 @@ type readOption func(*debeziumConfig)
// persistent expansion service running.
//
// Example:
-// username := "debezium"
-// password := "dbz"
-// host := "localhost"
-// port := "5432"
-// connectorClass := debeziumIO.POSTGRESQL
-// maxrecords := 1
-// debeziumio.Read(s.Scope("Read from debezium"), expansionAddr, username, password, host, port, connectorClass,
-// reflectx.String, debeziumio.MaxRecord(maxrecords), debeziumio.ExpansionAddr("localhost:9000"))
+//
+// username := "debezium"
+// password := "dbz"
+// host := "localhost"
+// port := "5432"
+// connectorClass := debeziumIO.POSTGRESQL
+// maxrecords := 1
+// debeziumio.Read(s.Scope("Read from debezium"), expansionAddr, username, password, host, port, connectorClass,
+// reflectx.String, debeziumio.MaxRecord(maxrecords), debeziumio.ExpansionAddr("localhost:9000"))
func Read(s beam.Scope, username, password, host, port string, connectorClass DriverClassName, t reflect.Type, opts ...readOption) beam.PCollection {
rfds := readFromDebeziumSchema{
ConnectorClass: string(connectorClass),
diff --git a/sdks/go/pkg/beam/io/xlang/jdbcio/jdbc.go b/sdks/go/pkg/beam/io/xlang/jdbcio/jdbc.go
index a027e00c5db..5a968de2cf4 100644
--- a/sdks/go/pkg/beam/io/xlang/jdbcio/jdbc.go
+++ b/sdks/go/pkg/beam/io/xlang/jdbcio/jdbc.go
@@ -16,7 +16,7 @@
// Package jdbcio contains cross-language functionality for reading and writing data to JDBC.
// These transforms only work on runners that support cross-language transforms.
//
-// Setup
+// # Setup
//
// Transforms specified here are cross-language transforms implemented in a
// different SDK (listed below). During pipeline construction, the Go SDK will
@@ -34,11 +34,12 @@
//
// Current supported SDKs, including expansion service modules and reference
// documentation:
-// * Java
-// - Vendored Module: beam-sdks-java-extensions-schemaio-expansion-service
-// - Run via Gradle: ./gradlew :sdks:java:extensions:schemaio-expansion-service:build
-// java -jar <location_of_jar_file_generated_from_above> <port>
-// - Reference Class: org.apache.beam.sdk.io.jdbc.JdbcIO
+//
+// Java:
+// - Vendored Module: beam-sdks-java-extensions-schemaio-expansion-service
+// - Run via Gradle: ./gradlew :sdks:java:extensions:schemaio-expansion-service:build
+// java -jar <location_of_jar_file_generated_from_above> <port>
+// - Reference Class: org.apache.beam.sdk.io.jdbc.JdbcIO
package jdbcio
import (
@@ -126,15 +127,17 @@ func toRow(pl interface{}) []byte {
//
// The default write statement is: "INSERT INTO tableName(column1, ...) INTO VALUES(value1, ...)"
// Example:
-// tableName := "roles"
-// driverClassName := "org.postgresql.Driver"
-// username := "root"
-// password := "root123"
-// jdbcUrl := "jdbc:postgresql://localhost:5432/dbname"
-// jdbcio.Write(s, tableName, driverClassName, jdbcurl, username, password, jdbcio.ExpansionAddrWrite("localhost:9000"))
+//
+// tableName := "roles"
+// driverClassName := "org.postgresql.Driver"
+// username := "root"
+// password := "root123"
+// jdbcUrl := "jdbc:postgresql://localhost:5432/dbname"
+// jdbcio.Write(s, tableName, driverClassName, jdbcurl, username, password, jdbcio.ExpansionAddrWrite("localhost:9000"))
//
// With Classpath paramater:
-// jdbcio.Write(s, tableName, driverClassName, jdbcurl, username, password, jdbcio.ExpansionAddrWrite("localhost:9000"), jdbcio.WriteClasspaths([]string{"org.postgresql:postgresql:42.3.3"}))
+//
+// jdbcio.Write(s, tableName, driverClassName, jdbcurl, username, password, jdbcio.ExpansionAddrWrite("localhost:9000"), jdbcio.WriteClasspaths([]string{"org.postgresql:postgresql:42.3.3"}))
func Write(s beam.Scope, tableName, driverClassName, jdbcUrl, username, password string, col beam.PCollection, opts ...writeOption) {
s = s.Scope("jdbcio.Write")
@@ -219,11 +222,12 @@ func ExpansionAddrWrite(expansionAddr string) writeOption {
//
// The default write statement is: "INSERT INTO tableName(column1, ...) INTO VALUES(value1, ...)"
// Example:
-// tableName := "roles"
-// username := "root"
-// password := "root123"
-// jdbcUrl := "jdbc:postgresql://localhost:5432/dbname"
-// jdbcio.WriteToPostgres(s, tableName, jdbcurl, username, password, jdbcio.ExpansionAddrWrite("localhost:9000"))
+//
+// tableName := "roles"
+// username := "root"
+// password := "root123"
+// jdbcUrl := "jdbc:postgresql://localhost:5432/dbname"
+// jdbcio.WriteToPostgres(s, tableName, jdbcurl, username, password, jdbcio.ExpansionAddrWrite("localhost:9000"))
func WriteToPostgres(s beam.Scope, tableName, jdbcUrl, username, password string, col beam.PCollection, opts ...writeOption) {
driverClassName := "org.postgresql.Driver"
Write(s, tableName, driverClassName, jdbcUrl, username, password, col, opts...)
@@ -245,16 +249,18 @@ func WriteToPostgres(s beam.Scope, tableName, jdbcUrl, username, password string
// an optional parameter, call the function within Read's function signature.
//
// Example:
-// tableName := "roles"
-// driverClassName := "org.postgresql.Driver"
-// username := "root"
-// password := "root123"
-// jdbcUrl := "jdbc:postgresql://localhost:5432/dbname"
-// outT := reflect.TypeOf((*JdbcTestRow)(nil)).Elem()
-// jdbcio.Read(s, tableName, driverClassName, jdbcurl, username, password, outT, jdbcio.ExpansionAddrRead("localhost:9000"))
+//
+// tableName := "roles"
+// driverClassName := "org.postgresql.Driver"
+// username := "root"
+// password := "root123"
+// jdbcUrl := "jdbc:postgresql://localhost:5432/dbname"
+// outT := reflect.TypeOf((*JdbcTestRow)(nil)).Elem()
+// jdbcio.Read(s, tableName, driverClassName, jdbcurl, username, password, outT, jdbcio.ExpansionAddrRead("localhost:9000"))
//
// With Classpath parameter:
-// jdbcio.Read(s, tableName, driverClassName, jdbcurl, username, password, outT, jdbcio.ExpansionAddrRead("localhost:9000"), jdbcio.ReadClasspaths([]string{"org.postgresql:postgresql:42.3.3"})))
+//
+// jdbcio.Read(s, tableName, driverClassName, jdbcurl, username, password, outT, jdbcio.ExpansionAddrRead("localhost:9000"), jdbcio.ReadClasspaths([]string{"org.postgresql:postgresql:42.3.3"})))
func Read(s beam.Scope, tableName, driverClassName, jdbcUrl, username, password string, outT reflect.Type, opts ...readOption) beam.PCollection {
s = s.Scope("jdbcio.Read")
@@ -358,12 +364,13 @@ func ExpansionAddrRead(expansionAddr string) readOption {
// with custom postgres driver then use the conventional jdbcio.Read() transform.
//
// Example:
-// tableName := "roles"
-// username := "root"
-// password := "root123"
-// jdbcUrl := "jdbc:postgresql://localhost:5432/dbname"
-// outT := reflect.TypeOf((*JdbcTestRow)(nil)).Elem()
-// jdbcio.Read(s, tableName, jdbcurl, username, password, outT, jdbcio.ExpansionAddrRead("localhost:9000"))
+//
+// tableName := "roles"
+// username := "root"
+// password := "root123"
+// jdbcUrl := "jdbc:postgresql://localhost:5432/dbname"
+// outT := reflect.TypeOf((*JdbcTestRow)(nil)).Elem()
+// jdbcio.Read(s, tableName, jdbcurl, username, password, outT, jdbcio.ExpansionAddrRead("localhost:9000"))
func ReadFromPostgres(s beam.Scope, tableName, jdbcUrl, username, password string, outT reflect.Type, opts ...readOption) beam.PCollection {
driverClassName := "org.postgresql.Driver"
return Read(s, tableName, driverClassName, jdbcUrl, username, password, outT, opts...)
diff --git a/sdks/go/pkg/beam/io/xlang/kafkaio/kafka.go b/sdks/go/pkg/beam/io/xlang/kafkaio/kafka.go
index 2bec1d1d7b4..2373ec89c28 100644
--- a/sdks/go/pkg/beam/io/xlang/kafkaio/kafka.go
+++ b/sdks/go/pkg/beam/io/xlang/kafkaio/kafka.go
@@ -17,7 +17,7 @@
// (http://kafka.apache.org/). These transforms only work on runners that
// support cross-language transforms.
//
-// Setup
+// # Setup
//
// Transforms specified here are cross-language transforms implemented in a
// different SDK (listed below). During pipeline construction, the Go SDK will
@@ -35,10 +35,11 @@
//
// Current supported SDKs, including expansion service modules and reference
// documentation:
-// * Java
-// - Vendored Module: beam-sdks-java-io-expansion-service
-// - Run via Gradle: ./gradlew :sdks:java:io:expansion-service:runExpansionService
-// - Reference Class: org.apache.beam.sdk.io.kafka.KafkaIO
+//
+// Java:
+// - Vendored Module: beam-sdks-java-io-expansion-service
+// - Run via Gradle: ./gradlew :sdks:java:io:expansion-service:runExpansionService
+// - Reference Class: org.apache.beam.sdk.io.kafka.KafkaIO
package kafkaio
// TODO(https://github.com/apache/beam/issues/21000): Implement an API for specifying Kafka type serializers and
@@ -107,11 +108,11 @@ var autoStartupAddress string = xlangx.UseAutomatedJavaExpansionService(serviceG
//
// Example of Read with required and optional parameters:
//
-// expansionAddr := "localhost:1234"
-// bootstrapServer := "bootstrap-server:1234"
-// topic := "topic_name"
-// pcol := kafkaio.Read( s, expansionAddr, bootstrapServer, []string{topic},
-// kafkaio.MaxNumRecords(100), kafkaio.CommitOffsetInFinalize(true))
+// expansionAddr := "localhost:1234"
+// bootstrapServer := "bootstrap-server:1234"
+// topic := "topic_name"
+// pcol := kafkaio.Read( s, expansionAddr, bootstrapServer, []string{topic},
+// kafkaio.MaxNumRecords(100), kafkaio.CommitOffsetInFinalize(true))
func Read(s beam.Scope, addr string, servers string, topics []string, opts ...readOption) beam.PCollection {
s = s.Scope("kafkaio.Read")
@@ -252,11 +253,11 @@ type readPayload struct {
//
// Example of Write with required and optional parameters:
//
-// expansionAddr := "localhost:1234"
-// bootstrapServer := "bootstrap-server:1234"
-// topic := "topic_name"
-// pcol := kafkaio.Read(s, expansionAddr, bootstrapServer, topic,
-// kafkaio.ValueSerializer("foo.BarSerializer"))
+// expansionAddr := "localhost:1234"
+// bootstrapServer := "bootstrap-server:1234"
+// topic := "topic_name"
+// pcol := kafkaio.Read(s, expansionAddr, bootstrapServer, topic,
+// kafkaio.ValueSerializer("foo.BarSerializer"))
func Write(s beam.Scope, addr, servers, topic string, col beam.PCollection, opts ...writeOption) {
s = s.Scope("kafkaio.Write")
diff --git a/sdks/go/pkg/beam/model/fnexecution_v1/beam_fn_api.pb.go b/sdks/go/pkg/beam/model/fnexecution_v1/beam_fn_api.pb.go
index 5792509b826..de568f66f82 100644
--- a/sdks/go/pkg/beam/model/fnexecution_v1/beam_fn_api.pb.go
+++ b/sdks/go/pkg/beam/model/fnexecution_v1/beam_fn_api.pb.go
@@ -1552,14 +1552,14 @@ func (x *ProcessBundleSplitRequest) GetDesiredSplits() map[string]*ProcessBundle
// Represents a partition of the bundle: a "primary" and a "residual", with the
// following properties:
-// - The work in primary and residual doesn't overlap, and combined, adds up
-// to the work in the current bundle if the split hadn't happened.
-// - The current bundle, if it keeps executing, will have done exactly none of
-// the work under residual_roots and none of the elements at and beyond the
-// first_residual_element.
-// - The current bundle, if no further splits happen, will have done exactly
-// the work under primary_roots and all elements up to and including the
-// channel splits last_primary_element.
+// - The work in primary and residual doesn't overlap, and combined, adds up
+// to the work in the current bundle if the split hadn't happened.
+// - The current bundle, if it keeps executing, will have done exactly none of
+// the work under residual_roots and none of the elements at and beyond the
+// first_residual_element.
+// - The current bundle, if no further splits happen, will have done exactly
+// the work under primary_roots and all elements up to and including the
+// channel splits last_primary_element.
//
// This allows the SDK to relinquish ownership of and commit to not process some
// of the elements that it may have been sent (the residual) while retaining
@@ -3291,19 +3291,19 @@ func (x *ProcessBundleSplitRequest_DesiredSplit) GetEstimatedInputElements() int
// as some range in an underlying dataset).
//
// Note that for a split the following properties must hold:
-// - last_primary_element < first_residual_element
-// - primary roots and residual roots can only be specified if the
-// last_primary_element + 1 < first_residual_element
-// (typically there is one primary and residual root per element in the
-// range (last_primary_element, first_residual_element))
-// - primary roots and residual roots must represent a disjoint but full
-// coverage of work represented by the elements between last_primary_element
-// and first_residual_element
+// - last_primary_element < first_residual_element
+// - primary roots and residual roots can only be specified if the
+// last_primary_element + 1 < first_residual_element
+// (typically there is one primary and residual root per element in the
+// range (last_primary_element, first_residual_element))
+// - primary roots and residual roots must represent a disjoint but full
+// coverage of work represented by the elements between last_primary_element
+// and first_residual_element
//
// Note that subsequent splits of the same bundle must ensure that:
-// - the first_residual_element does not increase
-// - the first_residual_element does not decrease if there were residual
-// or primary roots returned in a prior split.
+// - the first_residual_element does not increase
+// - the first_residual_element does not decrease if there were residual
+// or primary roots returned in a prior split.
type ProcessBundleSplitResponse_ChannelSplit struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
diff --git a/sdks/go/pkg/beam/model/jobmanagement_v1/beam_job_api.pb.go b/sdks/go/pkg/beam/model/jobmanagement_v1/beam_job_api.pb.go
index b9fb775dc5d..5fbeb25579c 100644
--- a/sdks/go/pkg/beam/model/jobmanagement_v1/beam_job_api.pb.go
+++ b/sdks/go/pkg/beam/model/jobmanagement_v1/beam_job_api.pb.go
@@ -1159,11 +1159,12 @@ func (*JobMessagesResponse_StateResponse) isJobMessagesResponse_Response() {}
// Enumeration of all JobStates
//
// The state transition diagram is:
-// STOPPED -> STARTING -> RUNNING -> DONE
-// \> FAILED
-// \> CANCELLING -> CANCELLED
-// \> UPDATING -> UPDATED
-// \> DRAINING -> DRAINED
+//
+// STOPPED -> STARTING -> RUNNING -> DONE
+// \> FAILED
+// \> CANCELLING -> CANCELLED
+// \> UPDATING -> UPDATED
+// \> DRAINING -> DRAINED
//
// Transitions are optional such that a job may go from STOPPED to RUNNING
// without needing to pass through STARTING.
diff --git a/sdks/go/pkg/beam/model/pipeline_v1/beam_runner_api.pb.go b/sdks/go/pkg/beam/model/pipeline_v1/beam_runner_api.pb.go
index 04950a38b57..aca9f6bb81a 100644
--- a/sdks/go/pkg/beam/model/pipeline_v1/beam_runner_api.pb.go
+++ b/sdks/go/pkg/beam/model/pipeline_v1/beam_runner_api.pb.go
@@ -2011,10 +2011,10 @@ func (x *Components) GetEnvironments() map[string]*Environment {
// A Pipeline is a hierarchical graph of PTransforms, linked
// by PCollections. A typical graph may look like:
//
-// Impulse -> PCollection -> ParDo -> PCollection -> GroupByKey -> ...
-// \> PCollection -> ParDo -> ...
-// \> ParDo -> ...
-// Impulse -> PCollection -> ParDo -> PCollection -> ...
+// Impulse -> PCollection -> ParDo -> PCollection -> GroupByKey -> ...
+// \> PCollection -> ParDo -> ...
+// \> ParDo -> ...
+// Impulse -> PCollection -> ParDo -> PCollection -> ...
//
// This is represented by a number of by-reference maps to transforms,
// PCollections, SDK environments, coders, etc., for
@@ -5634,28 +5634,28 @@ func (*StandardRequirements) Descriptor() ([]byte, []int) {
//
// This structure is reused in two distinct, but compatible, ways:
//
-// 1. This can be a specification of the function over PCollections
-// that a PTransform computes.
-// 2. This can be a specification of a user-defined function, possibly
-// SDK-specific. (external to this message must be adequate context
-// to indicate the environment in which the UDF can be understood).
+// 1. This can be a specification of the function over PCollections
+// that a PTransform computes.
+// 2. This can be a specification of a user-defined function, possibly
+// SDK-specific. (external to this message must be adequate context
+// to indicate the environment in which the UDF can be understood).
//
// Though not explicit in this proto, there are two possibilities
// for the relationship of a runner to this specification that
// one should bear in mind:
//
-// 1. The runner understands the URN. For example, it might be
-// a well-known URN like "beam:transform:Top" or
-// "beam:window_fn:FixedWindows" with
-// an agreed-upon payload (e.g. a number or duration,
-// respectively).
-// 2. The runner does not understand the URN. It might be an
-// SDK specific URN such as "beam:dofn:javasdk:1.0"
-// that indicates to the SDK what the payload is,
-// such as a serialized Java DoFn from a particular
-// version of the Beam Java SDK. The payload will often
-// then be an opaque message such as bytes in a
-// language-specific serialization format.
+// 1. The runner understands the URN. For example, it might be
+// a well-known URN like "beam:transform:Top" or
+// "beam:window_fn:FixedWindows" with
+// an agreed-upon payload (e.g. a number or duration,
+// respectively).
+// 2. The runner does not understand the URN. It might be an
+// SDK specific URN such as "beam:dofn:javasdk:1.0"
+// that indicates to the SDK what the payload is,
+// such as a serialized Java DoFn from a particular
+// version of the Beam Java SDK. The payload will often
+// then be an opaque message such as bytes in a
+// language-specific serialization format.
type FunctionSpec struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
diff --git a/sdks/go/pkg/beam/pardo.go b/sdks/go/pkg/beam/pardo.go
index 67131010958..6f4c05e14f4 100644
--- a/sdks/go/pkg/beam/pardo.go
+++ b/sdks/go/pkg/beam/pardo.go
@@ -150,7 +150,7 @@ func ParDo0(s Scope, dofn interface{}, col PCollection, opts ...Option) {
// happens inside the "Mapper" or "Reducer" class of a MapReduce-style
// algorithm.
//
-// DoFns
+// # DoFns
//
// The function to use to process each element is specified by a DoFn, either as
// single function or as a struct with methods, notably ProcessElement. The
@@ -163,23 +163,23 @@ func ParDo0(s Scope, dofn interface{}, col PCollection, opts ...Option) {
// direct runner). For each bundle of input elements processing proceeds as
// follows:
//
-// * If a struct, a fresh instance of the argument DoFn is created on a
-// worker from json serialization, and the Setup method is called on this
-// instance, if present. A runner may reuse DoFn instances for multiple
-// bundles. A DoFn that has terminated abnormally (by returning an error)
-// will never be reused.
-// * The DoFn's StartBundle method, if provided, is called to initialize it.
-// * The DoFn's ProcessElement method is called on each of the input elements
-// in the bundle.
-// * The DoFn's FinishBundle method, if provided, is called to complete its
-// work. After FinishBundle is called, the framework will not again invoke
-// ProcessElement or FinishBundle until a new call to StartBundle has
-// occurred.
-// * If any of Setup, StartBundle, ProcessElement or FinishBundle methods
-// return an error, the Teardown method, if provided, will be called on the
-// DoFn instance.
-// * If a runner will no longer use a DoFn, the Teardown method, if provided,
-// will be called on the discarded instance.
+// - If a struct, a fresh instance of the argument DoFn is created on a
+// worker from json serialization, and the Setup method is called on this
+// instance, if present. A runner may reuse DoFn instances for multiple
+// bundles. A DoFn that has terminated abnormally (by returning an error)
+// will never be reused.
+// - The DoFn's StartBundle method, if provided, is called to initialize it.
+// - The DoFn's ProcessElement method is called on each of the input elements
+// in the bundle.
+// - The DoFn's FinishBundle method, if provided, is called to complete its
+// work. After FinishBundle is called, the framework will not again invoke
+// ProcessElement or FinishBundle until a new call to StartBundle has
+// occurred.
+// - If any of Setup, StartBundle, ProcessElement or FinishBundle methods
+// return an error, the Teardown method, if provided, will be called on the
+// DoFn instance.
+// - If a runner will no longer use a DoFn, the Teardown method, if provided,
+// will be called on the discarded instance.
//
// Each of the calls to any of the DoFn's processing methods can produce zero
// or more output elements. All of the of output elements from all of the DoFn
@@ -187,11 +187,10 @@ func ParDo0(s Scope, dofn interface{}, col PCollection, opts ...Option) {
//
// For example:
//
-// words := beam.ParDo(s, &Foo{...}, ...)
-// lengths := beam.ParDo(s, func (word string) int) {
-// return len(word)
-// }, words)
-//
+// words := beam.ParDo(s, &Foo{...}, ...)
+// lengths := beam.ParDo(s, func (word string) int) {
+// return len(word)
+// }, words)
//
// Each output element has the same timestamp and is in the same windows as its
// corresponding input element. The timestamp can be accessed and/or emitted by
@@ -199,7 +198,7 @@ func ParDo0(s Scope, dofn interface{}, col PCollection, opts ...Option) {
// used as the DoFn name. Function literals do not have stable names and should
// thus not be used in production code.
//
-// Side Inputs
+// # Side Inputs
//
// While a ParDo processes elements from a single "main input" PCollection, it
// can take additional "side input" PCollections. These SideInput along with
@@ -208,56 +207,55 @@ func ParDo0(s Scope, dofn interface{}, col PCollection, opts ...Option) {
// options, and their contents accessible to each of the DoFn operations. For
// example:
//
-// words := ...
-// cufoff := ... // Singleton PCollection<int>
-// smallWords := beam.ParDo(s, func (word string, cutoff int, emit func(string)) {
-// if len(word) < cutoff {
-// emit(word)
-// }
-// }, words, beam.SideInput{Input: cutoff})
+// words := ...
+// cufoff := ... // Singleton PCollection<int>
+// smallWords := beam.ParDo(s, func (word string, cutoff int, emit func(string)) {
+// if len(word) < cutoff {
+// emit(word)
+// }
+// }, words, beam.SideInput{Input: cutoff})
//
-// Additional Outputs
+// # Additional Outputs
//
// Optionally, a ParDo transform can produce zero or multiple output
// PCollections. Note the use of ParDo2 to specfic 2 outputs. For example:
//
-// words := ...
-// cufoff := ... // Singleton PCollection<int>
-// small, big := beam.ParDo2(s, func (word string, cutoff int, small, big func(string)) {
-// if len(word) < cutoff {
-// small(word)
-// } else {
-// big(word)
-// }
-// }, words, beam.SideInput{Input: cutoff})
-//
+// words := ...
+// cufoff := ... // Singleton PCollection<int>
+// small, big := beam.ParDo2(s, func (word string, cutoff int, small, big func(string)) {
+// if len(word) < cutoff {
+// small(word)
+// } else {
+// big(word)
+// }
+// }, words, beam.SideInput{Input: cutoff})
//
// By default, the Coders for the elements of each output PCollections is
// inferred from the concrete type.
//
-// No Global Shared State
+// # No Global Shared State
//
// There are three main ways to initialize the state of a DoFn instance
// processing a bundle:
//
-// * Define public instance variable state. This state will be automatically
-// JSON serialized and then deserialized in the DoFn instances created for
-// bundles. This method is good for state known when the original DoFn is
-// created in the main program, if it's not overly large. This is not
-// suitable for any state which must only be used for a single bundle, as
-// DoFn's may be used to process multiple bundles.
-//
-// * Compute the state as a singleton PCollection and pass it in as a side
-// input to the DoFn. This is good if the state needs to be computed by the
-// pipeline, or if the state is very large and so is best read from file(s)
-// rather than sent as part of the DoFn's serialized state.
-//
-// * Initialize the state in each DoFn instance, in a StartBundle method.
-// This is good if the initialization doesn't depend on any information
-// known only by the main program or computed by earlier pipeline
-// operations, but is the same for all instances of this DoFn for all
-// program executions, say setting up empty caches or initializing constant
-// data.
+// - Define public instance variable state. This state will be automatically
+// JSON serialized and then deserialized in the DoFn instances created for
+// bundles. This method is good for state known when the original DoFn is
+// created in the main program, if it's not overly large. This is not
+// suitable for any state which must only be used for a single bundle, as
+// DoFn's may be used to process multiple bundles.
+//
+// - Compute the state as a singleton PCollection and pass it in as a side
+// input to the DoFn. This is good if the state needs to be computed by the
+// pipeline, or if the state is very large and so is best read from file(s)
+// rather than sent as part of the DoFn's serialized state.
+//
+// - Initialize the state in each DoFn instance, in a StartBundle method.
+// This is good if the initialization doesn't depend on any information
+// known only by the main program or computed by earlier pipeline
+// operations, but is the same for all instances of this DoFn for all
+// program executions, say setting up empty caches or initializing constant
+// data.
//
// ParDo operations are intended to be able to run in parallel across multiple
// worker machines. This precludes easy sharing and updating mutable state
@@ -291,7 +289,7 @@ func ParDo0(s Scope, dofn interface{}, col PCollection, opts ...Option) {
// currently being processed. See the `RTracker` interface in core/sdf/sdf.go
// for more details.
//
-// Splitting
+// # Splitting
//
// Splitting means taking one restriction and splitting into two or more that
// cover the entire input space of the original one. In other words, processing
@@ -310,12 +308,12 @@ func ParDo0(s Scope, dofn interface{}, col PCollection, opts ...Option) {
// only occur on element boundaries, but for splittable DoFns this split
// can land within a restriction and will require splitting that restriction.
//
-// * Note: The Go SDK currently does not support dynamic splitting for SDFs,
-// only initial splitting. Only initially split restrictions can be
-// distributed by liquid sharding. Stragglers will not be split during
-// execution with dynamic splitting.
+// - Note: The Go SDK currently does not support dynamic splitting for SDFs,
+// only initial splitting. Only initially split restrictions can be
+// distributed by liquid sharding. Stragglers will not be split during
+// execution with dynamic splitting.
//
-// Splittable DoFn Methods
+// # Splittable DoFn Methods
//
// Making a splittable DoFn requires the following methods to be implemented on
// a DoFn in addition to the usual DoFn requirements. In the following
@@ -324,17 +322,17 @@ func ParDo0(s Scope, dofn interface{}, col PCollection, opts ...Option) {
// user-defined restriction, and can be any type as long as it is consistent
// throughout all the splittable DoFn methods:
//
-// * `CreateInitialRestriction(element) restriction`
+// - `CreateInitialRestriction(element) restriction`
// CreateInitialRestriction creates an initial restriction encompassing an
// entire element. The restriction created stays associated with the element
// it describes.
-// * `SplitRestriction(elem, restriction) []restriction`
+// - `SplitRestriction(elem, restriction) []restriction`
// SplitRestriction takes an element and its initial restriction, and
// optionally performs an initial split on it, returning a slice of all the
// split restrictions. If no splits are desired, the method returns a slice
// containing only the original restriction. This method will always be
// called on each newly created restriction before they are processed.
-// * `RestrictionSize(elem, restriction) float64`
+// - `RestrictionSize(elem, restriction) float64`
// RestrictionSize returns a cheap size estimation for a restriction. This
// size is an abstract non-negative scalar value that represents how much
// work a restriction takes compared to other restrictions in the same DoFn.
@@ -343,19 +341,19 @@ func ParDo0(s Scope, dofn interface{}, col PCollection, opts ...Option) {
// used by runners to estimate work for dynamic work rebalancing. Must be
// thread safe. Will be invoked concurrently during bundle processing due to
// runner initiated splitting and progress estimation.
-// * `CreateTracker(restriction) restrictionTracker`
+// - `CreateTracker(restriction) restrictionTracker`
// CreateTracker creates and returns a restriction tracker (a concrete type
// implementing the `sdf.RTracker` interface) given a restriction. The
// restriction tracker is used to track progress processing a restriction,
// and to allow for dynamic splits. This method is called on each
// restriction right before processing begins.
-// * `ProcessElement(sdf.RTracker, element, func emit(output))`
+// - `ProcessElement(sdf.RTracker, element, func emit(output))`
// For splittable DoFns, ProcessElement requires a restriction tracker
// before inputs, and generally requires emits to be used for outputs, since
// restrictions will generally produce multiple outputs. For more details
// on processing restrictions in a splittable DoFn, see `sdf.RTracker`.
//
-// Fault Tolerance
+// # Fault Tolerance
//
// In a distributed system, things can fail: machines can crash, machines can
// be unable to communicate across the network, etc. While individual failures
@@ -376,7 +374,7 @@ func ParDo0(s Scope, dofn interface{}, col PCollection, opts ...Option) {
// difficult to achieve, so it is advisable to strive to keep DoFns as pure
// functions as much as possible.
//
-// Optimization
+// # Optimization
//
// Beam runners may choose to apply optimizations to a pipeline before it is
// executed. A key optimization, fusion, relates to ParDo operations. If one
diff --git a/sdks/go/pkg/beam/partition.go b/sdks/go/pkg/beam/partition.go
index 8a6031e8970..a2cb1c78664 100644
--- a/sdks/go/pkg/beam/partition.go
+++ b/sdks/go/pkg/beam/partition.go
@@ -86,7 +86,7 @@ type partitionData struct {
// partitionFn is a Func with the following underlying type:
//
-// fn : (EventTime, T, emit_1, emit_2, ..., emit_N) -> error
+// fn : (EventTime, T, emit_1, emit_2, ..., emit_N) -> error
//
// where emit_i : (EventTime, T) -> () and N is given by the encoded
// partitionData value. For any input element, it invokes to the
@@ -124,7 +124,7 @@ func (f *partitionFn) Call(args []interface{}) []interface{} {
// partitionFnKV is a Func with the following underlying type:
//
-// fn : (EventTime, K, V, emit_1, emit_2, ..., emit_N) -> error
+// fn : (EventTime, K, V, emit_1, emit_2, ..., emit_N) -> error
//
// where emit_i : (EventTime, K, V) -> () and N is given by the encoded
// partitionData value. For any input element, it invokes to the
diff --git a/sdks/go/pkg/beam/runners/dataflow/dataflowlib/messages.go b/sdks/go/pkg/beam/runners/dataflow/dataflowlib/messages.go
index 0fef4be7641..86a5ae8f98b 100644
--- a/sdks/go/pkg/beam/runners/dataflow/dataflowlib/messages.go
+++ b/sdks/go/pkg/beam/runners/dataflow/dataflowlib/messages.go
@@ -46,20 +46,22 @@ type pipelineOptions struct {
// blobs. We manually add them here for convenient and safer use.
// userAgent models Job/Environment/UserAgent. Example value:
-// "userAgent": {
-// "name": "Apache Beam SDK for Python",
-// "version": "0.6.0.dev"
-// },
+//
+// "userAgent": {
+// "name": "Apache Beam SDK for Python",
+// "version": "0.6.0.dev"
+// },
type userAgent struct {
Name string `json:"name,omitempty"`
Version string `json:"version,omitempty"`
}
// version models Job/Environment/Version. Example value:
-// "version": {
-// "job_type": "PYTHON_BATCH",
-// "major": "5"
-// },
+//
+// "version": {
+// "job_type": "PYTHON_BATCH",
+// "major": "5"
+// },
type version struct {
JobType string `json:"job_type,omitempty"`
Major string `json:"major,omitempty"`
diff --git a/sdks/go/pkg/beam/runners/universal/runnerlib/compile.go b/sdks/go/pkg/beam/runners/universal/runnerlib/compile.go
index 219f44b1335..5bacad3a24c 100644
--- a/sdks/go/pkg/beam/runners/universal/runnerlib/compile.go
+++ b/sdks/go/pkg/beam/runners/universal/runnerlib/compile.go
@@ -55,10 +55,10 @@ func BuildTempWorkerBinary(ctx context.Context) (string, error) {
// BuildWorkerBinary creates a local worker binary for linux/amd64. It finds the filename
// by examining the call stack. We want the user entry (*), for example:
//
-// /Users/herohde/go/src/github.com/apache/beam/sdks/go/pkg/beam/runners/beamexec/main.go (skip: 2)
-// * /Users/herohde/go/src/github.com/apache/beam/sdks/go/examples/wordcount/wordcount.go (skip: 3)
-// /usr/local/go/src/runtime/proc.go (skip: 4) // not always present
-// /usr/local/go/src/runtime/asm_amd64.s (skip: 4 or 5)
+// /Users/herohde/go/src/github.com/apache/beam/sdks/go/pkg/beam/runners/beamexec/main.go (skip: 2)
+// * /Users/herohde/go/src/github.com/apache/beam/sdks/go/examples/wordcount/wordcount.go (skip: 3)
+// /usr/local/go/src/runtime/proc.go (skip: 4) // not always present
+// /usr/local/go/src/runtime/asm_amd64.s (skip: 4 or 5)
func BuildWorkerBinary(ctx context.Context, filename string) error {
program := ""
var isTest bool
diff --git a/sdks/go/pkg/beam/schema.go b/sdks/go/pkg/beam/schema.go
index 2a3134455d9..140956328c1 100644
--- a/sdks/go/pkg/beam/schema.go
+++ b/sdks/go/pkg/beam/schema.go
@@ -37,11 +37,11 @@ import (
//
// Providers have three tasks with respect to a given supported logical type:
//
-// * Producing schema representative types for their logical types.
-// * Producing schema encoders for values of that type, writing beam
-// schema encoded bytes for a value, matching the schema representative type.
-// * Producing schema decoders for values of that type, reading beam
-// schema encoded bytes, and producing a value of that type.
+// - Producing schema representative types for their logical types.
+// - Producing schema encoders for values of that type, writing beam
+// schema encoded bytes for a value, matching the schema representative type.
+// - Producing schema decoders for values of that type, reading beam
+// schema encoded bytes, and producing a value of that type.
//
// Representative Schema types must be structs with only exported fields.
//
diff --git a/sdks/go/pkg/beam/testing/ptest/ptest.go b/sdks/go/pkg/beam/testing/ptest/ptest.go
index 54e3a14acf4..b606b7ebf66 100644
--- a/sdks/go/pkg/beam/testing/ptest/ptest.go
+++ b/sdks/go/pkg/beam/testing/ptest/ptest.go
@@ -121,7 +121,6 @@ func RunAndValidate(t *testing.T, p *beam.Pipeline) beam.PipelineResult {
// func TestMain(m *testing.M) {
// ptest.Main(m)
// }
-//
func Main(m *testing.M) {
MainWithDefault(m, "direct")
}
diff --git a/sdks/go/pkg/beam/transforms/filter/filter.go b/sdks/go/pkg/beam/transforms/filter/filter.go
index 1b289bbe3dd..8966ad524cc 100644
--- a/sdks/go/pkg/beam/transforms/filter/filter.go
+++ b/sdks/go/pkg/beam/transforms/filter/filter.go
@@ -36,10 +36,10 @@ var (
// the filter function returns false. It returns a PCollection of the same type
// as the input. For example:
//
-// words := beam.Create(s, "a", "b", "long", "alsolong")
-// short := filter.Include(s, words, func(s string) bool {
-// return len(s) < 3
-// })
+// words := beam.Create(s, "a", "b", "long", "alsolong")
+// short := filter.Include(s, words, func(s string) bool {
+// return len(s) < 3
+// })
//
// Here, "short" will contain "a" and "b" at runtime.
func Include(s beam.Scope, col beam.PCollection, fn interface{}) beam.PCollection {
@@ -54,10 +54,10 @@ func Include(s beam.Scope, col beam.PCollection, fn interface{}) beam.PCollectio
// the filter function returns true. It returns a PCollection of the same type
// as the input. For example:
//
-// words := beam.Create(s, "a", "b", "long", "alsolong")
-// long := filter.Exclude(s, words, func(s string) bool {
-// return len(s) < 3
-// })
+// words := beam.Create(s, "a", "b", "long", "alsolong")
+// long := filter.Exclude(s, words, func(s string) bool {
+// return len(s) < 3
+// })
//
// Here, "long" will contain "long" and "alsolong" at runtime.
func Exclude(s beam.Scope, col beam.PCollection, fn interface{}) beam.PCollection {
diff --git a/sdks/go/pkg/beam/transforms/stats/max.go b/sdks/go/pkg/beam/transforms/stats/max.go
index faef087aff6..16508d7cd0c 100644
--- a/sdks/go/pkg/beam/transforms/stats/max.go
+++ b/sdks/go/pkg/beam/transforms/stats/max.go
@@ -28,9 +28,8 @@ import (
//
// For example:
//
-// col := beam.Create(s, 1, 11, 7, 5, 10)
-// max := stats.Max(s, col) // PCollection<int> with 11 as the only element.
-//
+// col := beam.Create(s, 1, 11, 7, 5, 10)
+// max := stats.Max(s, col) // PCollection<int> with 11 as the only element.
func Max(s beam.Scope, col beam.PCollection) beam.PCollection {
s = s.Scope("stats.Max")
return combine(s, findMaxFn, col)
diff --git a/sdks/go/pkg/beam/transforms/stats/mean.go b/sdks/go/pkg/beam/transforms/stats/mean.go
index 2c875e766d5..4324e31d5a8 100644
--- a/sdks/go/pkg/beam/transforms/stats/mean.go
+++ b/sdks/go/pkg/beam/transforms/stats/mean.go
@@ -28,9 +28,8 @@ import (
//
// For example:
//
-// col := beam.Create(s, 1, 11, 7, 5, 10)
-// mean := stats.Mean(s, col) // PCollection<float64> with 6.8 as the only element.
-//
+// col := beam.Create(s, 1, 11, 7, 5, 10)
+// mean := stats.Mean(s, col) // PCollection<float64> with 6.8 as the only element.
func Mean(s beam.Scope, col beam.PCollection) beam.PCollection {
s = s.Scope("stats.Mean")
diff --git a/sdks/go/pkg/beam/transforms/stats/min.go b/sdks/go/pkg/beam/transforms/stats/min.go
index 675c5df0048..a48b2275e5c 100644
--- a/sdks/go/pkg/beam/transforms/stats/min.go
+++ b/sdks/go/pkg/beam/transforms/stats/min.go
@@ -28,9 +28,8 @@ import (
//
// For example:
//
-// col := beam.Create(s, 1, 11, 7, 5, 10)
-// min := stats.Min(s, col) // PCollection<int> with 1 as the only element.
-//
+// col := beam.Create(s, 1, 11, 7, 5, 10)
+// min := stats.Min(s, col) // PCollection<int> with 1 as the only element.
func Min(s beam.Scope, col beam.PCollection) beam.PCollection {
s = s.Scope("stats.Min")
return combine(s, findMinFn, col)
diff --git a/sdks/go/pkg/beam/transforms/stats/sum.go b/sdks/go/pkg/beam/transforms/stats/sum.go
index a1f6ecf14c3..91797896f93 100644
--- a/sdks/go/pkg/beam/transforms/stats/sum.go
+++ b/sdks/go/pkg/beam/transforms/stats/sum.go
@@ -28,9 +28,8 @@ import (
//
// For example:
//
-// col := beam.Create(s, 1, 11, 7, 5, 10)
-// sum := stats.Sum(s, col) // PCollection<int> with 34 as the only element.
-//
+// col := beam.Create(s, 1, 11, 7, 5, 10)
+// sum := stats.Sum(s, col) // PCollection<int> with 34 as the only element.
func Sum(s beam.Scope, col beam.PCollection) beam.PCollection {
s = s.Scope("stats.Sum")
return combine(s, findSumFn, col)
diff --git a/sdks/go/pkg/beam/transforms/top/top.go b/sdks/go/pkg/beam/transforms/top/top.go
index 811f425f057..38fcc3bba3a 100644
--- a/sdks/go/pkg/beam/transforms/top/top.go
+++ b/sdks/go/pkg/beam/transforms/top/top.go
@@ -49,9 +49,8 @@ var (
//
// Example use:
//
-// col := beam.Create(s, 1, 11, 7, 5, 10)
-// top2 := stats.Largest(s, col, 2, less) // PCollection<[]int> with [11, 10] as the only element.
-//
+// col := beam.Create(s, 1, 11, 7, 5, 10)
+// top2 := stats.Largest(s, col, 2, less) // PCollection<[]int> with [11, 10] as the only element.
func Largest(s beam.Scope, col beam.PCollection, n int, less interface{}) beam.PCollection {
s = s.Scope(fmt.Sprintf("top.Largest(%v)", n))
@@ -79,9 +78,8 @@ func LargestPerKey(s beam.Scope, col beam.PCollection, n int, less interface{})
//
// Example use:
//
-// col := beam.Create(s, 1, 11, 7, 5, 10)
-// bottom2 := stats.Smallest(s, col, 2, less) // PCollection<[]int> with [1, 5] as the only element.
-//
+// col := beam.Create(s, 1, 11, 7, 5, 10)
+// bottom2 := stats.Smallest(s, col, 2, less) // PCollection<[]int> with [1, 5] as the only element.
func Smallest(s beam.Scope, col beam.PCollection, n int, less interface{}) beam.PCollection {
s = s.Scope(fmt.Sprintf("top.Smallest(%v)", n))
diff --git a/sdks/go/pkg/beam/transforms/xlang/python/external.go b/sdks/go/pkg/beam/transforms/xlang/python/external.go
index 3fd6edd37e3..2c7c529b143 100644
--- a/sdks/go/pkg/beam/transforms/xlang/python/external.go
+++ b/sdks/go/pkg/beam/transforms/xlang/python/external.go
@@ -46,25 +46,32 @@ func init() {
// that can be evaluated to Python callables in Python SDK.
//
// The snippet of Python code can be a valid Python expression such as
-// lambda x: x * x
-// str.upper
+//
+// lambda x: x * x
+// str.upper
+//
// a fully qualified name such as
-// math.sin
+//
+// math.sin
+//
// or a complete multi-line function or class definition such as
-// def foo(x):
-// ...
-// class Foo:
-// ...
+//
+// def foo(x):
+// ...
+// class Foo:
+// ...
//
// Any lines preceding the function definition are first evaluated to provide context in which to
// define the function which can be useful to declare imports or any other needed values, e.g.
-// import math
//
-// def helper(x):
-// return x * x
+// import math
+//
+// def helper(x):
+// return x * x
+//
+// def func(y):
+// return helper(y) + y
//
-// def func(y):
-// return helper(y) + y
// in which case `func` would get applied to each element.
type CallableSource string
diff --git a/sdks/go/pkg/beam/xlang.go b/sdks/go/pkg/beam/xlang.go
index 0a219660a6d..1fe26fc714a 100644
--- a/sdks/go/pkg/beam/xlang.go
+++ b/sdks/go/pkg/beam/xlang.go
@@ -26,7 +26,8 @@ import (
// beam.CrossLanguage.
//
// Example:
-// beam.CrossLanguage(s, urn, payload, addr, UnnamedInput(input), outputs)
+//
+// beam.CrossLanguage(s, urn, payload, addr, UnnamedInput(input), outputs)
func UnnamedInput(col PCollection) map[string]PCollection {
return map[string]PCollection{graph.UnnamedInputTag: col}
}
@@ -35,8 +36,9 @@ func UnnamedInput(col PCollection) map[string]PCollection {
// beam.CrossLanguage. The associated output can be accessed with beam.UnnamedOutputTag.
//
// Example:
-// resultMap := beam.CrossLanguage(s, urn, payload, addr, inputs, UnnamedOutput(output));
-// result := resultMap[beam.UnnamedOutputTag()]
+//
+// resultMap := beam.CrossLanguage(s, urn, payload, addr, inputs, UnnamedOutput(output));
+// result := resultMap[beam.UnnamedOutputTag()]
func UnnamedOutput(t FullType) map[string]FullType {
return map[string]FullType{graph.UnnamedOutputTag: t}
}
@@ -56,10 +58,11 @@ func UnnamedOutputTag() string {
// schemas, and pkg/beam/core/runtime/graphx/schema for details on schemas in the Go SDK.
//
// Example:
-// type stringPayload struct {
-// Data string
-// }
-// encodedPl := beam.CrossLanguagePayload(stringPayload{Data: "foo"})
+//
+// type stringPayload struct {
+// Data string
+// }
+// encodedPl := beam.CrossLanguagePayload(stringPayload{Data: "foo"})
func CrossLanguagePayload(pl interface{}) []byte {
bytes, err := xlangx.EncodeStructPayload(pl)
if err != nil {
@@ -81,7 +84,7 @@ func CrossLanguagePayload(pl interface{}) []byte {
// For more information on expansion services and other aspects of cross-language transforms in
// general, refer to the Beam programming guide: https://beam.apache.org/documentation/programming-guide/#multi-language-pipelines
//
-// Payload
+// # Payload
//
// Payloads are configuration data that some cross-language transforms require for expansion.
// Consult the documentation of the transform in the source SDK to find out what payload data it
@@ -91,7 +94,7 @@ func CrossLanguagePayload(pl interface{}) []byte {
// protobuf. The helper function beam.CrossLanguagePayload is the recommended way to easily encode
// a standard Go struct for use as a payload.
//
-// Inputs and Outputs
+// # Inputs and Outputs
//
// Like most transforms, any input PCollections must be provided. Unlike most transforms, output
// types must be provided because Go cannot infer output types from external transforms.
@@ -103,11 +106,11 @@ func CrossLanguagePayload(pl interface{}) []byte {
//
// An example of defining named inputs and outputs:
//
-// namedInputs := map[string]beam.PCollection{"pcol1": pcol1, "pcol2": pcol2}
-// namedOutputTypes := map[string]typex.FullType{
-// "main": typex.New(reflectx.String),
-// "side": typex.New(reflectx.Int64),
-// }
+// namedInputs := map[string]beam.PCollection{"pcol1": pcol1, "pcol2": pcol2}
+// namedOutputTypes := map[string]typex.FullType{
+// "main": typex.New(reflectx.String),
+// "side": typex.New(reflectx.Int64),
+// }
//
// CrossLanguage outputs a map of PCollections with associated names. These names will match those
// from provided named outputs. If the beam.UnnamedOutput method was used, the PCollection can be
@@ -115,28 +118,28 @@ func CrossLanguagePayload(pl interface{}) []byte {
//
// An example of retrieving named outputs from a call to CrossLanguage:
//
-// outputs := beam.CrossLanguage(...)
-// mainPcol := outputs["main"]
-// sidePcol := outputs["side"]
+// outputs := beam.CrossLanguage(...)
+// mainPcol := outputs["main"]
+// sidePcol := outputs["side"]
//
-// Example
+// # Example
//
// This example shows using CrossLanguage to execute the Prefix cross-language transform using an
// expansion service running on localhost:8099. Prefix requires a payload containing a prefix to
// prepend to every input string.
//
-// type prefixPayload struct {
-// Data string
-// }
-// encodedPl := beam.CrossLanguagePayload(prefixPayload{Data: "foo"})
-// urn := "beam:transforms:xlang:test:prefix"
-// expansionAddr := "localhost:8099"
-// outputType := beam.UnnamedOutput(typex.New(reflectx.String))
-// input := beam.UnnamedInput(inputPcol)
-// outs := beam.CrossLanguage(s, urn, encodedPl, expansionAddr, input, outputType)
-// outPcol := outputs[beam.UnnamedOutputTag()]
+// type prefixPayload struct {
+// Data string
+// }
+// encodedPl := beam.CrossLanguagePayload(prefixPayload{Data: "foo"})
+// urn := "beam:transforms:xlang:test:prefix"
+// expansionAddr := "localhost:8099"
+// outputType := beam.UnnamedOutput(typex.New(reflectx.String))
+// input := beam.UnnamedInput(inputPcol)
+// outs := beam.CrossLanguage(s, urn, encodedPl, expansionAddr, input, outputType)
+// outPcol := outputs[beam.UnnamedOutputTag()]
//
-// Alternative Expansion Handlers
+// # Alternative Expansion Handlers
//
// The xlangx.RegisterHandler function can be used to register alternative expansion
// handlers to a namespace, for use with this function. This allows for custom handling
diff --git a/sdks/go/test/integration/expansions.go b/sdks/go/test/integration/expansions.go
index 660dbfc5034..7e8c1164f50 100644
--- a/sdks/go/test/integration/expansions.go
+++ b/sdks/go/test/integration/expansions.go
@@ -28,7 +28,7 @@ import (
// on the --expansion_jar and --expansion_addr flags in this package. The main reason to use this
// instead of accessing the flags directly is to let it handle jar startup and shutdown.
//
-// Usage
+// # Usage
//
// Create an ExpansionServices object in TestMain with NewExpansionServices. Then use GetAddr for
// every expansion service needed for the test. Call Shutdown on it before finishing TestMain (or
@@ -39,16 +39,17 @@ import (
// recommended to only use ExpansionServices in TestMain to avoid this.
//
// Example:
-// flag.Parse()
-// beam.Init()
-// services := integration.NewExpansionServices()
-// defer func() { services.Shutdown() }()
-// addr, err := services.GetAddr("example")
-// if err != nil {
-// panic(err)
-// }
-// expansionAddr = addr // Save address to a package-level variable used by tests.
-// ptest.MainRet(m)
+//
+// flag.Parse()
+// beam.Init()
+// services := integration.NewExpansionServices()
+// defer func() { services.Shutdown() }()
+// addr, err := services.GetAddr("example")
+// if err != nil {
+// panic(err)
+// }
+// expansionAddr = addr // Save address to a package-level variable used by tests.
+// ptest.MainRet(m)
type ExpansionServices struct {
addrs map[string]string
jars map[string]string
diff --git a/sdks/go/test/integration/flags.go b/sdks/go/test/integration/flags.go
index 3a58826a436..04dc66f4739 100644
--- a/sdks/go/test/integration/flags.go
+++ b/sdks/go/test/integration/flags.go
@@ -104,10 +104,13 @@ func GetExpansionAddrs() map[string]string {
// of the flag.
//
// Example:
-// var myFlags stringSlice
-// flag.Var(&myFlags, "my_flag", "A list of flags")
+//
+// var myFlags stringSlice
+// flag.Var(&myFlags, "my_flag", "A list of flags")
+//
// With the example above, the slice can be set to contain ["foo", "bar"]:
-// cmd -my_flag foo -my_flag bar
+//
+// cmd -my_flag foo -my_flag bar
type stringSlice []string
// String implements the String method of flag.Value. This outputs the value
diff --git a/sdks/go/test/integration/io/xlang/jdbc/jdbc_test.go b/sdks/go/test/integration/io/xlang/jdbc/jdbc_test.go
index ef561eeb6c4..27021308a78 100644
--- a/sdks/go/test/integration/io/xlang/jdbc/jdbc_test.go
+++ b/sdks/go/test/integration/io/xlang/jdbc/jdbc_test.go
@@ -5,7 +5,7 @@
// (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
+// 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,