You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2026/07/22 18:36:09 UTC
(beam) branch master updated: docs: Document pipeline runner API protos (#39292)
This is an automated email from the ASF dual-hosted git repository.
damccorm 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 f1d9c3e122e docs: Document pipeline runner API protos (#39292)
f1d9c3e122e is described below
commit f1d9c3e122e210dc1a5dc09beaa2e2d3cc14197b
Author: Danny McCormick <da...@google.com>
AuthorDate: Wed Jul 22 14:36:02 2026 -0400
docs: Document pipeline runner API protos (#39292)
* Document pipeline runner API protos
Add documentation comments to previously undocumented messages and fields
in the Beam Runner API pipeline model protos:
beam_runner_api.proto: State specs (ReadModifyWrite, Bag, Combining, Map,
Multimap, Set, TimerFamily), environment payloads (Docker, Process,
External, AnyOf), artifact messages, triggers, RedistributePayload,
IsBounded, LabelledPayload, and more.
metrics.proto: MonitoringInfo, MonitoringInfoSpecs, labels, type URNs.
schema.proto: Schema, Field, FieldType, all type variants, Row, FieldValue.
external_transforms.proto: ManagedTransforms, Annotations.
endpoints.proto: AuthenticationSpec.
* Minor fixes
Co-authored-by: Danny McCormick <da...@google.com>
---
.../beam/model/pipeline/v1/beam_runner_api.proto | 94 +++++++++++++++++++---
.../apache/beam/model/pipeline/v1/endpoints.proto | 3 +
.../model/pipeline/v1/external_transforms.proto | 20 +++++
.../apache/beam/model/pipeline/v1/metrics.proto | 73 ++++++++++++++++-
.../org/apache/beam/model/pipeline/v1/schema.proto | 94 ++++++++++++++++++++++
5 files changed, 269 insertions(+), 15 deletions(-)
diff --git a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto
index 5824c9bf4b7..ceea6ee2c17 100644
--- a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto
+++ b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto
@@ -586,11 +586,17 @@ message StateSpec {
FunctionSpec protocol = 7;
}
+// Specification for a read-modify-write state cell, which holds a single value
+// per key and window. The value can be read, modified, and written back atomically.
message ReadModifyWriteStateSpec {
+ // (Required) The id of the Coder for the values stored in this state cell.
string coder_id = 1;
}
+// Specification for a bag state cell, which holds an unordered collection of elements
+// per key and window. Elements can be appended and read as an iterable.
message BagStateSpec {
+ // (Required) The id of the Coder for elements stored in this bag.
string element_coder_id = 1;
}
@@ -601,30 +607,57 @@ message OrderedListStateSpec {
string element_coder_id = 1;
}
+// Specification for a combining state cell, which incrementally combines values
+// using a CombineFn. Rather than storing all input values, it stores a single
+// accumulator that is updated as new values arrive.
message CombiningStateSpec {
+ // (Required) The id of the Coder for the accumulator.
string accumulator_coder_id = 1;
+ // (Required) The FunctionSpec of the CombineFn used to combine values.
FunctionSpec combine_fn = 2;
}
+// Specification for a map state cell, which holds a key-value mapping per key
+// and window. Keys and values can be read, written, and deleted individually.
+// Note: This is distinct from MultimapStateSpec in that MapState keys are unique.
+// In practice, Beam SDKs use Multimap state for both map and multimap semantics.
message MapStateSpec {
+ // (Required) The id of the Coder for map keys.
string key_coder_id = 1;
+ // (Required) The id of the Coder for map values.
string value_coder_id = 2;
}
+// Specification for a multimap state cell, which holds a key-to-multiple-values
+// mapping per key and window. A single key can be associated with multiple values,
+// analogous to a multimap.
message MultimapStateSpec {
+ // (Required) The id of the Coder for map keys.
string key_coder_id = 1;
+ // (Required) The id of the Coder for map values.
string value_coder_id = 2;
}
+// Specification for a set state cell, which holds an unordered collection of
+// unique elements per key and window. Elements can be added, removed, and checked
+// for existence.
message SetStateSpec {
+ // (Required) The id of the Coder for elements stored in this set.
string element_coder_id = 1;
}
+// Specification for a timer family, which groups timers that share the same
+// time domain and encoding. Timer families allow timers to be managed together,
+// including batch processing of expired timers.
message TimerFamilySpec {
+ // (Required) The time domain for this timer family (event time or processing time).
TimeDomain.Enum time_domain = 1;
+ // (Required) The id of the Coder for timers in this family.
string timer_family_coder_id = 2;
}
+// Indicates whether a PCollection is bounded (finite, known size) or unbounded (infinite, unknown size).
+// This affects how runners execute the pipeline and which features are available.
message IsBounded {
enum Enum {
UNSPECIFIED = 0;
@@ -811,7 +844,13 @@ message GroupIntoBatchesPayload {
int64 max_buffering_duration_millis = 2;
}
+// Payload for the RedistributeByKey and RedistributeArbitrarily composite transforms.
+// These transforms redistribute elements across workers, optionally allowing duplicates
+// to improve throughput at the cost of additional processing.
message RedistributePayload {
+ // (Optional) If true, the redistribution may produce duplicate elements.
+ // This can improve performance by avoiding costly deduplication, but requires
+ // downstream transforms to be tolerant of duplicates.
bool allow_duplicates = 1;
}
@@ -1330,8 +1369,9 @@ message Trigger {
message Default {
}
- // Ready whenever the requisite number of input elements have arrived
+ // Ready whenever the requisite number of input elements have arrived.
message ElementCount {
+ // (Required) The number of elements that must arrive before this trigger fires.
int32 element_count = 1;
}
@@ -1477,11 +1517,12 @@ message StandardArtifacts {
}
}
+// Payload for a file-based artifact, referencing a locally-accessible file on disk.
message ArtifactFilePayload {
- // a string for an artifact file path e.g. "/tmp/foo.jar"
+ // (Required) A file path for an artifact e.g. "/tmp/foo.jar"
string path = 1;
- // The hex-encoded sha256 checksum of the artifact.
+ // (Optional) The hex-encoded sha256 checksum of the artifact.
string sha256 = 2;
}
@@ -1498,11 +1539,12 @@ message EmbeddedFilePayload {
bytes data = 1;
}
+// Payload for a PyPI artifact, describing a Python package dependency.
message PyPIPayload {
- // Pypi compatible artifact id e.g. "apache-beam"
+ // (Required) A PyPI-compatible artifact id e.g. "apache-beam"
string artifact_id = 1;
- // Pypi compatible version string.
+ // (Required) A PyPI-compatible version string, e.g. "2.59.0"
string version = 2;
}
@@ -1515,29 +1557,40 @@ message MavenPayload {
string repository_url = 2;
}
+// Payload for a deferred artifact, where the interpretation of the data is
+// deferred to the creator. This allows custom artifact types not covered by
+// the standard artifact types.
message DeferredArtifactPayload {
- // A unique string identifier assigned by the creator of this payload. The creator may use this key to confirm
+ // (Required) A unique string identifier assigned by the creator of this payload. The creator may use this key to confirm
// whether they can parse the data.
string key = 1;
- // Data for deferred artifacts. Interpretation of bytes is delegated to the creator of this payload.
+ // (Required) Data for deferred artifacts. Interpretation of bytes is delegated to the creator of this payload.
bytes data = 2;
}
+// Payload for the staging_to role, used when an artifact needs to be staged
+// to a particular location. The staged name is relative to the staging directory.
message ArtifactStagingToRolePayload {
- // A generated staged name (relative path under staging directory).
+ // (Required) A generated staged name (relative path under staging directory).
string staged_name = 1;
}
+// Full descriptor of an artifact, including its type and role.
+// Used to describe dependencies of an Environment.
message ArtifactInformation {
- // A URN that describes the type of artifact
+ // (Required) A URN that describes the type of artifact (e.g. file, URL, embedded, PyPI, Maven).
+ // See StandardArtifacts.Types for well-known URNs.
string type_urn = 1;
+ // (Optional) The payload for the artifact type, as determined by type_urn.
bytes type_payload = 2;
- // A URN that describes the role of artifact
+ // (Required) A URN that describes the role of the artifact (e.g. staging_to, go_worker_binary).
+ // See StandardArtifacts.Roles for well-known URNs.
string role_urn = 3;
+ // (Optional) The payload for the artifact role, as determined by role_urn.
bytes role_payload = 4;
}
@@ -1598,24 +1651,41 @@ message StandardEnvironments {
}
}
-// The payload of a Docker image
+// Payload for a Docker container environment.
message DockerPayload {
+ // (Required) The Docker container image URL, e.g. "apache/beam_java_sdk:2.59.0"
string container_image = 1; // implicitly linux_amd64.
}
+// Payload for a native process environment, where the SDK harness is launched as
+// a subprocess directly on the worker machine (not inside a container).
message ProcessPayload {
+ // (Required) The operating system, e.g. "linux", "darwin".
string os = 1; // "linux", "darwin", ..
+ // (Required) The CPU architecture, e.g. "amd64", "arm64".
string arch = 2; // "amd64", ..
+ // (Required) The command to execute to start the SDK harness process.
string command = 3; // process to execute
+ // (Optional) Environment variables to set for the SDK harness process.
map<string, string> env = 4; // Environment variables
}
+// Payload for an external process environment, where the SDK harness is managed
+// by an external service rather than directly by the runner. The runner communicates
+// with the external worker pool via the provided gRPC endpoint.
message ExternalPayload {
+ // (Required) The gRPC endpoint serving the BeamFnExternalWorkerPool API.
ApiServiceDescriptor endpoint = 1; // Serving BeamFnExternalWorkerPool API.
+ // (Optional) Arbitrary extra parameters to pass to the external worker pool.
map<string, string> params = 2; // Arbitrary extra parameters to pass
}
+// Payload for an AnyOf environment, which presents a set of equivalent environments
+// that a runner may choose from. Each alternative is fully specified with its own
+// dependencies, capabilities, etc. This allows SDKs to express flexibility in
+// execution environment while preserving compatibility across runners.
message AnyOfEnvironmentPayload {
+ // (Required) The list of fully-specified environments the runner may choose from.
// Each is fully contained (with their own dependencies, capabilities, etc.)
repeated Environment environments = 1;
}
@@ -1852,6 +1922,8 @@ message StandardDisplayData {
}
}
+// A labelled display data payload, associating a human-readable label with a value.
+// Used as the payload for the StandardDisplayData.LABELLED URN.
message LabelledPayload {
// (Required) A human readable label for the value.
string label = 1;
diff --git a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/endpoints.proto b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/endpoints.proto
index 46fb3d5d3b1..202d8d6181b 100644
--- a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/endpoints.proto
+++ b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/endpoints.proto
@@ -39,6 +39,9 @@ message ApiServiceDescriptor {
AuthenticationSpec authentication = 2;
}
+// Describes the authentication mechanism for connecting to a Beam API endpoint.
+// Uses a URN-based approach to allow different authentication methods (e.g. OAuth2,
+// bearer tokens) to be specified in an extensible manner.
message AuthenticationSpec {
// (Required) A URN that describes the accompanying payload.
// For any URN that is not recognized (by whomever is inspecting
diff --git a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto
index c73986eed48..918455dbdbd 100644
--- a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto
+++ b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto
@@ -60,32 +60,49 @@ message ExpansionMethods {
}
// Defines the URNs for managed transforms.
+// Managed transforms are high-level, pre-configured transforms that map to
+// specific SchemaTransform implementations. They provide a way to
+// to use common I/O patterns without worrying about Beam version.
+// See https://beam.apache.org/documentation/io/managed-io/
message ManagedTransforms {
enum Urns {
+ // Reads from an Iceberg table.
ICEBERG_READ = 0 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:iceberg_read:v1"];
+ // Writes to an Iceberg table.
ICEBERG_WRITE = 1 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:iceberg_write:v1"];
+ // Reads from a Kafka topic.
KAFKA_READ = 2 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:kafka_read:v1"];
+ // Writes to a Kafka topic.
KAFKA_WRITE = 3 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:kafka_write:v1"];
+ // Reads from BigQuery using the Storage API.
BIGQUERY_READ = 4 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:bigquery_storage_read:v1"];
+ // Writes to BigQuery.
BIGQUERY_WRITE = 5 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:bigquery_write:v1"];
+ // Reads from Iceberg in CDC (Change Data Capture) mode.
ICEBERG_CDC_READ = 6 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:iceberg_cdc_read:v1"];
+ // Reads from a PostgreSQL database.
POSTGRES_READ = 7 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:postgres_read:v1"];
+ // Writes to a PostgreSQL database.
POSTGRES_WRITE = 8 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:postgres_write:v1"];
+ // Reads from a MySQL database.
MYSQL_READ = 9 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:mysql_read:v1"];
+ // Writes to a MySQL database.
MYSQL_WRITE = 10 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:mysql_write:v1"];
+ // Reads from a SQL Server database.
SQL_SERVER_READ = 11 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:sql_server_read:v1"];
+ // Writes to a SQL Server database.
SQL_SERVER_WRITE = 12 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:sql_server_write:v1"];
DELTA_LAKE_READ = 13 [(org.apache.beam.model.pipeline.v1.beam_urn) =
@@ -145,6 +162,9 @@ message BuilderMethod {
bytes payload = 3;
}
+// Defines annotation keys used to attach metadata to transforms in the
+// portable pipeline representation. These annotations allow SDKs to
+// recover configuration information that was used to construct a transform.
message Annotations {
enum Enum {
// The annotation key for the encoded configuration Row used to build a transform
diff --git a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/metrics.proto b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/metrics.proto
index fcce35394b9..49e0c7b3ea8 100644
--- a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/metrics.proto
+++ b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/metrics.proto
@@ -58,7 +58,9 @@ message MonitoringInfoSpec {
// The key name and value string of MonitoringInfo annotations.
message Annotation {
+ // The annotation key, e.g. "description", "units".
string key = 1;
+ // The annotation value as a string.
string value = 2;
}
@@ -198,7 +200,8 @@ message MonitoringInfoSpecs {
}]
}];
- // Represents a set of strings seen across bundles.
+ // Represents a bounded trie of strings seen across bundles. The trie has a
+ // maximum number of elements it will store before truncating.
USER_BOUNDED_TRIE = 22 [(monitoring_info_spec) = {
urn: "beam:metric:user:bounded_trie:v1",
type: "beam:metrics:bounded_trie:v1",
@@ -225,6 +228,7 @@ message MonitoringInfoSpecs {
// }]
// }];
+ // The total number of elements output to a PCollection by a PTransform.
ELEMENT_COUNT = 10 [(monitoring_info_spec) = {
urn: "beam:metric:element_count:v1",
type: "beam:metrics:sum_int64:v1",
@@ -235,6 +239,8 @@ message MonitoringInfoSpecs {
} ]
}];
+ // The distribution of byte sizes of a sampled set of elements in a PCollection.
+ // Sampling is used because serializing elements to compute byte size is CPU-intensive.
SAMPLED_BYTE_SIZE = 11 [(monitoring_info_spec) = {
urn: "beam:metric:sampled_byte_size:v1",
type: "beam:metrics:distribution_int64:v1",
@@ -248,6 +254,8 @@ message MonitoringInfoSpecs {
} ]
}];
+ // The total estimated execution time (in milliseconds) of the startBundle
+ // function in a ParDo transform.
START_BUNDLE_MSECS = 12 [(monitoring_info_spec) = {
urn: "beam:metric:pardo_execution_time:start_bundle_msecs:v1",
type: "beam:metrics:sum_int64:v1",
@@ -259,6 +267,8 @@ message MonitoringInfoSpecs {
} ]
}];
+ // The total estimated execution time (in milliseconds) of the processElement
+ // function in a ParDo transform.
PROCESS_BUNDLE_MSECS = 13 [(monitoring_info_spec) = {
urn: "beam:metric:pardo_execution_time:process_bundle_msecs:v1",
type: "beam:metrics:sum_int64:v1",
@@ -270,6 +280,8 @@ message MonitoringInfoSpecs {
} ]
}];
+ // The total estimated execution time (in milliseconds) of the finishBundle
+ // function in a ParDo transform.
FINISH_BUNDLE_MSECS = 14 [(monitoring_info_spec) = {
urn: "beam:metric:pardo_execution_time:finish_bundle_msecs:v1",
type: "beam:metrics:sum_int64:v1",
@@ -281,6 +293,7 @@ message MonitoringInfoSpecs {
} ]
}];
+ // The total estimated execution time (in milliseconds) of the entire PTransform.
TOTAL_MSECS = 15 [(monitoring_info_spec) = {
urn: "beam:metric:ptransform_execution_time:total_msecs:v1",
type: "beam:metrics:sum_int64:v1",
@@ -330,6 +343,8 @@ message MonitoringInfoSpecs {
}]
}];
+ // The total number of API requests made to an I/O service, grouped by
+ // service, method, resource, PTransform, and status.
API_REQUEST_COUNT = 19 [(monitoring_info_spec) = {
urn: "beam:metric:io:api_request_count:v1",
type: "beam:metrics:sum_int64:v1",
@@ -353,6 +368,8 @@ message MonitoringInfoSpecs {
]
}];
+ // A histogram of API request latencies (in milliseconds) made to I/O service
+ // APIs for batching reads or writes.
API_REQUEST_LATENCIES = 20 [(monitoring_info_spec) = {
urn: "beam:metric:io:api_request_latencies:v1",
type: "beam:metrics:histogram_int64:v1",
@@ -379,6 +396,8 @@ message MonitoringInfoSpecs {
]
}];
+ // Represents a user-defined histogram metric tracking the distribution of
+ // integer values across bundles.
USER_HISTOGRAM = 23 [(monitoring_info_spec) = {
urn: "beam:metric:user:histogram_int64:v1",
type: "beam:metrics:histogram_int64:v1",
@@ -400,13 +419,23 @@ message MonitoringInfoLabelProps {
// Enum extension to store MonitoringInfo related
// specifications, constants, etc.
+// This extension mechanism allows enum values in MonitoringInfoSpecs and
+// MonitoringInfoLabels to carry structured metadata (specs and label names)
+// directly on the enum value definition.
extend google.protobuf.EnumValueOptions {
+ // Properties for MonitoringInfoLabel enum values, specifying the label name
+ // to use in the MonitoringInfo labels map.
MonitoringInfoLabelProps label_props = 127337796; // From: commit 0x7970544.
- // Enum extension to store the MonitoringInfoSpecs.
+ // Enum extension to store the MonitoringInfoSpec.
MonitoringInfoSpec monitoring_info_spec = 207174266;
}
+// A monitoring info represents a single data point or aggregated metric reported
+// by the SDK or runner. It contains the URN identifying the metric, the type
+// encoding, the payload (encoded metric value), labels providing context (such
+// as which PTransform or PCollection the metric is associated with), and an
+// optional start_time for cumulative metrics.
message MonitoringInfo {
// (Required) Defines the semantic meaning of the metric or monitored state.
//
@@ -429,33 +458,62 @@ message MonitoringInfo {
// refer to them. For actively processed bundles, these should match the
// values within the ProcessBundleDescriptor. For job management APIs,
// these should match values within the original pipeline representation.
+ // Label referencing the PTransform that this metric is associated with.
TRANSFORM = 0 [(label_props) = { name: "PTRANSFORM" }];
+ // Label referencing the PCollection that this metric is associated with.
PCOLLECTION = 1 [(label_props) = { name: "PCOLLECTION" }];
+ // Label referencing the WindowingStrategy that this metric is associated with.
WINDOWING_STRATEGY = 2 [(label_props) = { name: "WINDOWING_STRATEGY" }];
+ // Label referencing the Coder used for encoding elements.
CODER = 3 [(label_props) = { name: "CODER" }];
+ // Label referencing the Environment that this metric is associated with.
ENVIRONMENT = 4 [(label_props) = { name: "ENVIRONMENT" }];
+ // Label for the namespace of a user metric, used to avoid name collisions
+ // between user-defined metrics.
NAMESPACE = 5 [(label_props) = { name: "NAMESPACE" }];
+ // Label for the name of a user metric within its namespace.
NAME = 6 [(label_props) = { name: "NAME" }];
+ // Label for the I/O service name (e.g. "bigquery", "pubsub").
SERVICE = 7 [(label_props) = { name: "SERVICE" }];
+ // Label for the I/O API method name (e.g. "read", "write").
METHOD = 8 [(label_props) = { name: "METHOD" }];
+ // Label for the I/O resource being accessed (e.g. a table or topic name).
RESOURCE = 9 [(label_props) = { name: "RESOURCE" }];
+ // Label for the HTTP status code of an I/O API request.
STATUS = 10 [(label_props) = { name: "STATUS" }];
+ // Label for the BigQuery project ID associated with this metric.
BIGQUERY_PROJECT_ID = 11 [(label_props) = { name: "BIGQUERY_PROJECT_ID" }];
+ // Label for the BigQuery dataset associated with this metric.
BIGQUERY_DATASET = 12 [(label_props) = { name: "BIGQUERY_DATASET" }];
+ // Label for the BigQuery table associated with this metric.
BIGQUERY_TABLE = 13 [(label_props) = { name: "BIGQUERY_TABLE" }];
+ // Label for the BigQuery view associated with this metric.
BIGQUERY_VIEW = 14 [(label_props) = { name: "BIGQUERY_VIEW" }];
+ // Label for a user-provided query name used to identify a BigQuery query.
BIGQUERY_QUERY_NAME = 15 [(label_props) = { name: "BIGQUERY_QUERY_NAME" }];
+ // Label for the GCS bucket associated with this metric.
GCS_BUCKET = 16 [(label_props) = { name: "GCS_BUCKET"}];
+ // Label for the GCP project ID associated with a GCS operation.
GCS_PROJECT_ID = 17 [(label_props) = { name: "GCS_PROJECT_ID"}];
+ // Label for the Datastore project associated with this metric.
DATASTORE_PROJECT = 18 [(label_props) = { name: "DATASTORE_PROJECT" }];
+ // Label for the Datastore namespace associated with this metric.
DATASTORE_NAMESPACE = 19 [(label_props) = { name: "DATASTORE_NAMESPACE" }];
+ // Label for the Bigtable project ID associated with this metric.
BIGTABLE_PROJECT_ID = 20 [(label_props) = { name: "BIGTABLE_PROJECT_ID"}];
+ // Label for the instance ID of a Bigtable or other service.
INSTANCE_ID = 21 [(label_props) = { name: "INSTANCE_ID"}];
+ // Label for the table ID of a Bigtable or other service.
TABLE_ID = 22 [(label_props) = { name: "TABLE_ID"}];
+ // Label for the Spanner project ID associated with this metric.
SPANNER_PROJECT_ID = 23 [(label_props) = { name: "SPANNER_PROJECT_ID" }];
+ // Label for the Spanner database ID associated with this metric.
SPANNER_DATABASE_ID = 24 [(label_props) = { name: "SPANNER_DATABASE_ID" }];
+ // Label for the Spanner table ID associated with this metric.
SPANNER_TABLE_ID = 25 [(label_props) = { name: "SPANNER_TABLE_ID" }];
+ // Label for the Spanner instance ID associated with this metric.
SPANNER_INSTANCE_ID = 26 [(label_props) = { name: "SPANNER_INSTANCE_ID" }];
+ // Label for a user-provided query name used to identify a Spanner query.
SPANNER_QUERY_NAME = 27 [(label_props) = { name: "SPANNER_QUERY_NAME" }];
// Label which if has a "true" value indicates that the metric is intended
// to be aggregated per-worker.
@@ -489,7 +547,10 @@ message MonitoringInfo {
google.protobuf.Timestamp start_time = 5;
}
-// A set of well known URNs that specify the encoding and aggregation method.
+// A set of well known URNs that specify the encoding and aggregation method
+// for metric payloads. Each URN defines how the payload bytes of a MonitoringInfo
+// should be decoded and how values from multiple bundles should be aggregated
+// together by the runner.
message MonitoringInfoTypeUrns {
enum Enum {
// Represents an integer counter where values are summed across bundles.
@@ -594,6 +655,9 @@ message MonitoringInfoTypeUrns {
BOTTOM_N_DOUBLE_TYPE = 9 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:metrics:bottom_n_double:v1"];
+ // Represents progress information for a PTransform, containing both completed
+ // and remaining work as separate values.
+ //
// Encoding: <iter><value1><value2>...<valueN></iter>
// - iter: beam:coder:iterable:v1
// - valueX: beam:coder:double:v1
@@ -614,7 +678,8 @@ message MonitoringInfoTypeUrns {
BOUNDED_TRIE_TYPE = 12 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:metrics:bounded_trie:v1"];
- // Represents a histogram.
+ // Represents a histogram of integer values. The payload is encoded as a
+ // HistogramValue protobuf message.
HISTOGRAM = 13 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:metrics:histogram_int64:v1"];
diff --git a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto
index 58d16b83749..d514be101be 100644
--- a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto
+++ b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto
@@ -37,6 +37,8 @@ message Schema {
repeated Field fields = 1;
// REQUIRED. An RFC 4122 UUID.
string id = 2;
+ // (Optional) Schema-level options that provide additional metadata
+ // or configuration for this schema.
repeated Option options = 3;
// Indicates that encoding positions have been overridden.
bool encoding_positions_set = 4;
@@ -47,8 +49,11 @@ message Field {
string name = 1;
// OPTIONAL. Human readable description of this field, such as the query that generated it.
string description = 2;
+ // (Required) The type of this field.
FieldType type = 3;
+ // (Optional) An integer identifier for this field. Used for schema evolution
+ // to track fields across schema changes even when field names or positions change.
int32 id = 4;
// OPTIONAL. The position of this field's data when encoded, e.g. with beam:coder:row:v1.
// Either no fields in a given row are have encoding position populated,
@@ -58,56 +63,104 @@ message Field {
// If this Field is part of a Schema where encoding_positions_set is True then encoding_position must be
// defined, otherwise this field is ignored.
int32 encoding_position = 5;
+ // (Optional) Field-level options that provide additional metadata
+ // or configuration for this field.
repeated Option options = 6;
}
+// Represents the type of a field in a Beam Schema. A field type can be one of
+// several categories: atomic types (primitives like INT32, STRING, etc.),
+// collection types (arrays, iterables, maps), row types (nested schemas),
+// or logical types (custom types with a URN-based representation).
message FieldType {
+ // Whether this field can hold null values.
bool nullable = 1;
oneof type_info {
+ // A primitive (atomic) type such as INT32, STRING, BYTES, etc.
AtomicType atomic_type = 2;
+ // An ordered array of elements of a uniform element type.
ArrayType array_type = 3;
+ // An unordered iterable of elements of a uniform element type.
IterableType iterable_type = 4;
+ // A map with typed keys and typed values.
MapType map_type = 5;
+ // A nested row with its own schema.
RowType row_type = 6;
+ // A user-defined logical type, identified by URN.
LogicalType logical_type = 7;
}
}
+// The set of primitive (atomic) types supported by Beam Schemas.
enum AtomicType {
+ // Unspecified type; used as the default/placeholder value.
UNSPECIFIED = 0;
+ // A single signed byte (8-bit integer).
BYTE = 1;
+ // A 16-bit signed integer.
INT16 = 2;
+ // A 32-bit signed integer.
INT32 = 3;
+ // A 64-bit signed integer.
INT64 = 4;
+ // A single-precision (32-bit) IEEE 754 floating-point number.
FLOAT = 5;
+ // A double-precision (64-bit) IEEE 754 floating-point number.
DOUBLE = 6;
+ // A UTF-8 encoded string.
STRING = 7;
+ // A boolean (true/false) value.
BOOLEAN = 8;
+ // A sequence of bytes.
BYTES = 9;
}
+// Represents an ordered, variable-length array type. All elements share
+// the same element type.
message ArrayType {
+ // The type of elements in this array.
FieldType element_type = 1;
}
+// Represents an unordered, variable-length iterable type. Similar to ArrayType
+// but without guaranteed ordering of elements.
message IterableType {
+ // The type of elements in this iterable.
FieldType element_type = 1;
}
+// Represents a map (key-value association) type. Keys and values may each
+// have their own distinct types.
message MapType {
+ // The type of the map keys.
FieldType key_type = 1;
+ // The type of the map values.
FieldType value_type = 2;
}
+// Represents a nested row type, which contains its own Schema. This enables
+// hierarchical/structured data within a single schema field.
message RowType {
+ // The schema describing the structure of nested rows.
Schema schema = 1;
}
+// Represents a user-defined logical type. Logical types extend the type system
+// by mapping a URN-identified type to an underlying representation type.
+// This allows SDKs to define custom types (e.g. Decimal, Date) while encoding
+// them using one of the built-in atomic or collection types.
message LogicalType {
+ // (Required) URN uniquely identifying this logical type.
string urn = 1;
+ // (Optional) Payload providing additional configuration for this logical type.
bytes payload = 2;
+ // (Required) The underlying FieldType used to encode/decode values of this
+ // logical type on the wire.
FieldType representation = 3;
+ // (Optional) The type of the argument passed when constructing an instance
+ // of this logical type (e.g. the length argument for FixedBytes).
FieldType argument_type = 4;
+ // (Optional) The actual argument value for this logical type instance.
FieldValue argument = 5;
}
@@ -184,6 +237,9 @@ message LogicalTypes {
}
}
+// Represents a schema option, which is a named, typed value attached to a
+// Schema or Field. Options provide additional metadata or configuration
+// that is not part of the core type system.
message Option {
// REQUIRED. Identifier for the option.
string name = 1;
@@ -191,57 +247,93 @@ message Option {
// Conventionally, options that don't require additional configuration should
// use a boolean type, with the value set to true.
FieldType type = 2;
+ // The typed value of this option.
FieldValue value = 3;
}
+// A concrete instance of a row, containing the values for each field as defined
+// by the associated Schema. The order of values matches the field order in the
+// Schema.
message Row {
+ // The values for each field in the row, in schema field order.
repeated FieldValue values = 1;
}
+// Represents the value of a single field within a Row. The oneof corresponds
+// to the FieldType categories: atomic, array, iterable, map, row, or logical.
message FieldValue {
// If none of these are set, value is considered null.
oneof field_value {
+ // The value for an atomic (primitive) field type.
AtomicTypeValue atomic_value = 1;
+ // The value for an array field type.
ArrayTypeValue array_value = 2;
+ // The value for an iterable field type.
IterableTypeValue iterable_value = 3;
+ // The value for a map field type.
MapTypeValue map_value = 4;
+ // The value for a nested row field type.
Row row_value = 5;
+ // The value for a logical type field.
LogicalTypeValue logical_type_value = 6;
}
}
+// The concrete value of an atomic (primitive) field. The oneof member must
+// correspond to the AtomicType of the field's FieldType.
message AtomicTypeValue {
oneof value {
+ // Value for AtomicType.BYTE.
int32 byte = 1;
+ // Value for AtomicType.INT16.
int32 int16 = 2;
+ // Value for AtomicType.INT32.
int32 int32 = 3;
+ // Value for AtomicType.INT64.
int64 int64 = 4;
+ // Value for AtomicType.FLOAT.
float float = 5;
+ // Value for AtomicType.DOUBLE.
double double = 6;
+ // Value for AtomicType.STRING.
string string = 7;
+ // Value for AtomicType.BOOLEAN.
bool boolean = 8;
+ // Value for AtomicType.BYTES.
bytes bytes = 9;
}
}
+// The concrete value of an array field, containing a list of element values.
message ArrayTypeValue {
+ // The element values in this array, in order.
repeated FieldValue element = 1;
}
+// The concrete value of an iterable field, containing a list of element values.
message IterableTypeValue {
+ // The element values in this iterable.
repeated FieldValue element = 1;
}
+// The concrete value of a map field, represented as a list of key-value entries.
message MapTypeValue {
+ // The key-value entries in this map.
repeated MapTypeEntry entries = 1;
}
+// A single key-value entry within a MapTypeValue.
message MapTypeEntry {
+ // The key of this map entry.
FieldValue key = 1;
+ // The value of this map entry.
FieldValue value = 2;
}
+// The concrete value of a logical type field. Contains the underlying
+// representation value that encodes the logical type's data.
message LogicalTypeValue {
+ // The underlying representation value for this logical type.
FieldValue value = 1;
}
@@ -260,7 +352,9 @@ message SchemaCoderPayload {
repeated AdditionalCoderInfo additional_coder_infos = 4;
message AdditionalCoderInfo {
+ // URN identifying the additional coder information.
string urn = 1;
+ // Payload for the additional coder information.
bytes payload = 2;
}
}