From e4a5eea37a52429ccec55e3a3a1e0701ba5ff1ee Mon Sep 17 00:00:00 2001 From: Andrew Crites Date: Wed, 21 Jan 2026 20:31:30 +0000 Subject: [PATCH 1/9] Adds a new SchemaCoderPayload proto to use for portable SchemaCoders. --- .../model/pipeline/v1/beam_runner_api.proto | 4 ++++ .../beam/model/pipeline/v1/schema.proto | 22 ++++++++++++++++++- 2 files changed, 25 insertions(+), 1 deletion(-) 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 0bdc4f69aab6..67df8b9e8003 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 @@ -1073,6 +1073,10 @@ message StandardCoders { // Components: None ROW = 13 [(beam_urn) = "beam:coder:row:v1"]; + // Similar to ROW above, but for arbitrary types that can be converted + // to and from row objects, which can then be encoded with a schema. + SCHEMA = 18 [(beam_urn) = "beam:coder:schema:v1"]; + // Encodes a user key and a shard id which is an opaque byte string. // // The encoding for a sharded key consists of a shard id byte string and the 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 e05f6b565828..5250aaaa1a7a 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 @@ -48,7 +48,7 @@ message Field { // OPTIONAL. Human readable description of this field, such as the query that generated it. string description = 2; FieldType type = 3; - + 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, @@ -237,3 +237,23 @@ message MapTypeEntry { message LogicalTypeValue { FieldValue value = 1; } + +// Information needed to represent a Coder of type SCHEMA. +message SchemaCoderPayload { + // The schema to use for encoding corresponding Row types. + Schema schema = 1; + + // Function mapping from underlying object to Row type. + FunctionSpec to_row_fn = 2; + + // Function mapping from Row type to underlying object. + FunctionSpec from_row_fn = 3; + + // Any additional information SDKs need to encode/decode elements. + repeated AdditionalCoderInfo additional_coder_infos = 4; + + message AdditionalCoderInfo { + string urn = 1; + bytes payload = 2; + } +} From 7cb1a05cd578eb9ce76a7dfbac40c4962864070b Mon Sep 17 00:00:00 2001 From: Abdelrahman Ibrahim Date: Mon, 26 Jan 2026 18:19:27 +0200 Subject: [PATCH 2/9] disabled beam metrics report job (#37420) --- .github/workflows/beam_Metrics_Report.yml | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/.github/workflows/beam_Metrics_Report.yml b/.github/workflows/beam_Metrics_Report.yml index 70ed354958b8..f4e02edae6e0 100644 --- a/.github/workflows/beam_Metrics_Report.yml +++ b/.github/workflows/beam_Metrics_Report.yml @@ -17,10 +17,11 @@ name: Beam Metrics Report +# DISABLED: Workflow disabled on: - schedule: - - cron: '0 11 * * 2' - workflow_dispatch: + # schedule: + # - cron: '0 11 * * 2' + # workflow_dispatch: # This allows a subsequently queued workflow run to interrupt previous runs concurrency: From d090aab9773a2c77ed3fca2b8425fa70cbd8a67c Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Mon, 26 Jan 2026 12:23:36 -0500 Subject: [PATCH 3/9] Update BEAM_DEV_SDK_CONTAINER_TAG version (#37400) --- sdks/python/apache_beam/runners/dataflow/internal/names.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/runners/dataflow/internal/names.py b/sdks/python/apache_beam/runners/dataflow/internal/names.py index 7ffde6866fc7..8a0fd5d25e84 100644 --- a/sdks/python/apache_beam/runners/dataflow/internal/names.py +++ b/sdks/python/apache_beam/runners/dataflow/internal/names.py @@ -35,6 +35,6 @@ # Update this tag whenever there is a change that # requires changes to SDK harness container or SDK harness launcher. -BEAM_DEV_SDK_CONTAINER_TAG = 'beam-master-20260121' +BEAM_DEV_SDK_CONTAINER_TAG = 'beam-master-20260123' DATAFLOW_CONTAINER_IMAGE_REPOSITORY = 'gcr.io/cloud-dataflow/v1beta3' From f5af98bc167a1c502d35c704de63e65bd7fa50cd Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Mon, 26 Jan 2026 21:48:28 +0400 Subject: [PATCH 4/9] Fix website workflows (#37421) * Install yarn via corepack * Add corepack home dir --- website/Dockerfile | 22 +++++++--------------- 1 file changed, 7 insertions(+), 15 deletions(-) diff --git a/website/Dockerfile b/website/Dockerfile index 61ec7921703c..e09d06f22ead 100644 --- a/website/Dockerfile +++ b/website/Dockerfile @@ -51,25 +51,17 @@ RUN curl -sL https://deb.nodesource.com/setup_lts.x | bash - \ nodejs \ && apt-get autoremove -yqq --purge \ && apt-get clean \ - && rm -rf /var/lib/apt/lists/* + && rm -rf /var/lib/apt/lists/* RUN npm update -g npm - -RUN npm install postcss postcss-cli autoprefixer + +RUN npm install postcss postcss-cli autoprefixer # Install yarn -RUN set -eux; \ - apt-get update; \ - apt-get install -y --no-install-recommends curl gnupg ca-certificates; \ - mkdir -p /etc/apt/keyrings; \ - curl -fsSL https://dl.yarnpkg.com/debian/pubkey.gpg \ - | gpg --dearmor -o /etc/apt/keyrings/yarn.gpg; \ - echo "deb [signed-by=/etc/apt/keyrings/yarn.gpg] https://dl.yarnpkg.com/debian stable main" \ - > /etc/apt/sources.list.d/yarn.list; \ - apt-get update; \ - apt-get install -y --no-install-recommends yarn; \ - apt-get clean; \ - rm -rf /var/lib/apt/lists/* +ENV COREPACK_HOME=/usr/local/share/corepack +RUN mkdir -p "$COREPACK_HOME" \ + && corepack enable \ + && corepack prepare yarn@1.22.22 --activate # Install hugo extended version v0.117.0 RUN HUGOHOME="$(mktemp -d)" \ From 911ec6abf0f21040dde38dc41e1bed3f9ea9219c Mon Sep 17 00:00:00 2001 From: M Junaid Shaukat <154750865+junaiddshaukat@users.noreply.github.com> Date: Mon, 26 Jan 2026 23:28:12 +0500 Subject: [PATCH 5/9] Add filter() method to PCollection in TypeScript SDK (#37408) - Add filter() method following same pattern as map() and flatMap() - Add 5 test cases covering various filtering scenarios - Support optional context parameter for side inputs --- sdks/typescript/src/apache_beam/pvalue.ts | 49 +++++++++++++++++++++++ sdks/typescript/test/primitives_test.ts | 45 +++++++++++++++++++++ 2 files changed, 94 insertions(+) diff --git a/sdks/typescript/src/apache_beam/pvalue.ts b/sdks/typescript/src/apache_beam/pvalue.ts index 311ef13eabaa..8d1bdc51a8c8 100644 --- a/sdks/typescript/src/apache_beam/pvalue.ts +++ b/sdks/typescript/src/apache_beam/pvalue.ts @@ -163,6 +163,55 @@ export class PCollection { ); } + /** + * Returns a PCollection containing only elements that satisfy the given + * predicate function. + * + * This is analogous to JavaScript's `Array.filter()` method. + * + * Example usage: + * ``` + * const evens = pcoll.filter(x => x % 2 === 0); + * const positives = pcoll.filter(x => x > 0); + * ``` + * + * @param fn A predicate function that returns true for elements to keep, + * false for elements to filter out. The function receives the + * element and optionally a context object. + * @param context Optional context object to pass to the predicate function. + * @returns A new PCollection containing only the elements for which the + * predicate returned true. + */ + filter( + fn: + | (ContextT extends undefined ? (element: T) => boolean : never) + | ((element: T, context: ContextT) => boolean), + context: ContextT = undefined!, + ): PCollection { + if (extractContext(fn)) { + context = { ...extractContext(fn), ...context }; + } + return this.apply( + withName( + "filter(" + extractName(fn) + ")", + parDo( + { + process: function (element: T, context: ContextT) { + // Return the element wrapped in an array if predicate is true, + // otherwise return an empty array to filter it out. + const keep = + context === null || context === undefined + ? (fn as (element: T) => boolean)(element) + : fn(element, context); + return keep ? [element] : []; + }, + }, + context, + ), + ), + ); + } + root(): Root { return new Root(this.pipeline); } diff --git a/sdks/typescript/test/primitives_test.ts b/sdks/typescript/test/primitives_test.ts index 6cfb2d1bfa49..c5ab36f5de77 100644 --- a/sdks/typescript/test/primitives_test.ts +++ b/sdks/typescript/test/primitives_test.ts @@ -96,6 +96,51 @@ export function suite(runner: beam.Runner = directRunner()) { }); }); + it("runs a filter", async function () { + await runner.run((root) => { + root + .apply(beam.create([1, 2, 3, 4, 5, 6])) + .filter((x) => x % 2 === 0) + .apply(testing.assertDeepEqual([2, 4, 6])); + }); + }); + + it("runs a filter with predicate returning false for all", async function () { + await runner.run((root) => { + root + .apply(beam.create([1, 3, 5, 7])) + .filter((x) => x % 2 === 0) + .apply(testing.assertDeepEqual([])); + }); + }); + + it("runs a filter with predicate returning true for all", async function () { + await runner.run((root) => { + root + .apply(beam.create([2, 4, 6, 8])) + .filter((x) => x % 2 === 0) + .apply(testing.assertDeepEqual([2, 4, 6, 8])); + }); + }); + + it("runs a filter with context", async function () { + await runner.run((root) => { + root + .apply(beam.create([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])) + .filter((x: number, threshold: number) => x > threshold, 5) + .apply(testing.assertDeepEqual([6, 7, 8, 9, 10])); + }); + }); + + it("runs a filter on strings", async function () { + await runner.run((root) => { + root + .apply(beam.create(["apple", "banana", "apricot", "cherry"])) + .filter((s) => s.startsWith("a")) + .apply(testing.assertDeepEqual(["apple", "apricot"])); + }); + }); + it("runs a Splitter", async function () { await runner.run((root) => { const pcolls = root From adc04caf41216f015108149d0a6d5b9cd5811fe4 Mon Sep 17 00:00:00 2001 From: Abdelrahman Ibrahim Date: Mon, 26 Jan 2026 21:22:38 +0200 Subject: [PATCH 6/9] Fix gRPC failures in Python PreCommit tests (#37350) * Fixed failures in Python PreCommit tests * Remove explicit runner parameter * Removed Python 3.13 special case for bundle timeout --- sdks/python/apache_beam/yaml/yaml_testing.py | 1 + sdks/python/conftest.py | 5 ++--- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/sdks/python/apache_beam/yaml/yaml_testing.py b/sdks/python/apache_beam/yaml/yaml_testing.py index ead3ab9de319..10d23dd54868 100644 --- a/sdks/python/apache_beam/yaml/yaml_testing.py +++ b/sdks/python/apache_beam/yaml/yaml_testing.py @@ -46,6 +46,7 @@ def __init__(self, pipeline_spec, test_spec, options, fix_tests): self._test_spec = test_spec self._options = options self._fix_tests = fix_tests + self._fixes = None def runTest(self): self._fixes = run_test( diff --git a/sdks/python/conftest.py b/sdks/python/conftest.py index 683bd433e8a9..fe5d03fd66c3 100644 --- a/sdks/python/conftest.py +++ b/sdks/python/conftest.py @@ -54,10 +54,9 @@ def configure_beam_rpc_timeouts(): """ print("\n--- Applying Beam RPC timeout configuration ---") - # Set gRPC keepalive and timeout settings timeout_env_vars = { 'GRPC_ARG_KEEPALIVE_TIME_MS': '30000', - 'GRPC_ARG_KEEPALIVE_TIMEOUT_MS': '5000', + 'GRPC_ARG_KEEPALIVE_TIMEOUT_MS': '10000', 'GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA': '0', 'GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS': '1', 'GRPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS': '300000', @@ -66,7 +65,7 @@ def configure_beam_rpc_timeouts(): # Additional stability settings for DinD environment 'GRPC_ARG_MAX_RECONNECT_BACKOFF_MS': '120000', 'GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS': '1000', - 'GRPC_ARG_MAX_CONNECTION_IDLE_MS': '300000', + 'GRPC_ARG_MAX_CONNECTION_IDLE_MS': '600000', 'GRPC_ARG_MAX_CONNECTION_AGE_MS': '1800000', # Beam-specific retry and timeout settings From d0014ff38d2951d4989f61feff7921dfc2c317a2 Mon Sep 17 00:00:00 2001 From: Ben Feinstein <11245062+benfei@users.noreply.github.com> Date: Mon, 26 Jan 2026 21:31:06 +0200 Subject: [PATCH 7/9] Simplify encoding kw_only args and test init=False args. (#37231) * Simplify encoding kw_only args and test init=False args. 1. Replace the helper function for determining whether to use kw_only with a simpler to follow condition. 2. Add test coverage for the init=False case. * Fix pre-commit. --- sdks/python/apache_beam/coders/coder_impl.py | 15 +-------------- .../apache_beam/coders/coders_test_common.py | 14 +++++++++++++- 2 files changed, 14 insertions(+), 15 deletions(-) diff --git a/sdks/python/apache_beam/coders/coder_impl.py b/sdks/python/apache_beam/coders/coder_impl.py index 1e3bb2ece92a..3e0b5218b166 100644 --- a/sdks/python/apache_beam/coders/coder_impl.py +++ b/sdks/python/apache_beam/coders/coder_impl.py @@ -32,7 +32,6 @@ import decimal import enum -import functools import itertools import json import logging @@ -376,18 +375,6 @@ def _verify_dill_compat(): raise RuntimeError(base_error + f". Found dill version '{dill.__version__}") -dataclass_uses_kw_only: Callable[[Any], bool] -if dataclasses: - # Cache the result to avoid multiple checks for the same dataclass type. - @functools.cache - def dataclass_uses_kw_only(cls) -> bool: - return any( - field.init and field.kw_only for field in dataclasses.fields(cls)) - -else: - dataclass_uses_kw_only = lambda cls: False - - class FastPrimitivesCoderImpl(StreamCoderImpl): """For internal use only; no backwards-compatibility guarantees.""" def __init__( @@ -518,7 +505,7 @@ def encode_special_deterministic(self, value, stream): (value, type(value), self.requires_deterministic_step_label)) init_fields = [field for field in dataclasses.fields(value) if field.init] try: - if dataclass_uses_kw_only(type(value)): + if any(field.kw_only for field in init_fields): stream.write_byte(DATACLASS_KW_ONLY_TYPE) self.encode_type(type(value), stream) stream.write_var_int64(len(init_fields)) diff --git a/sdks/python/apache_beam/coders/coders_test_common.py b/sdks/python/apache_beam/coders/coders_test_common.py index 8f89ab9602c1..fcc5e6ac58bf 100644 --- a/sdks/python/apache_beam/coders/coders_test_common.py +++ b/sdks/python/apache_beam/coders/coders_test_common.py @@ -123,6 +123,15 @@ class UnFrozenDataClass: x: int y: int + @dataclasses.dataclass(frozen=True, kw_only=True) + class FrozenUnInitKwOnlyDataClass: + side: int + area: int = dataclasses.field(init=False) + + def __post_init__(self): + # Hack to update an attribute in a frozen dataclass. + object.__setattr__(self, 'area', self.side**2) + # These tests need to all be run in the same process due to the asserts # in tearDownClass. @@ -309,6 +318,8 @@ def test_deterministic_coder(self, compat_version): if dataclasses is not None: self.check_coder(deterministic_coder, FrozenDataClass(1, 2)) self.check_coder(deterministic_coder, FrozenKwOnlyDataClass(c=1, d=2)) + self.check_coder( + deterministic_coder, FrozenUnInitKwOnlyDataClass(side=11)) with self.assertRaises(TypeError): self.check_coder(deterministic_coder, UnFrozenDataClass(1, 2)) @@ -750,6 +761,7 @@ def test_cross_process_encoding_of_special_types_is_deterministic( from apache_beam.coders.coders_test_common import DefinesGetAndSetState from apache_beam.coders.coders_test_common import FrozenDataClass from apache_beam.coders.coders_test_common import FrozenKwOnlyDataClass + from apache_beam.coders.coders_test_common import FrozenUnInitKwOnlyDataClass from apache_beam.coders import proto2_coder_test_messages_pb2 as test_message @@ -786,7 +798,7 @@ def test_cross_process_encoding_of_special_types_is_deterministic( ("frozen_dataclass", FrozenDataClass(1, 2)), ("frozen_dataclass_list", [FrozenDataClass(1, 2), FrozenDataClass(3, 4)]), ("frozen_kwonly_dataclass", FrozenKwOnlyDataClass(c=1, d=2)), - ("frozen_kwonly_dataclass_list", [FrozenKwOnlyDataClass(c=1, d=2), FrozenKwOnlyDataClass(c=3, d=4)]), + ("frozen_kwonly_dataclass_list", [FrozenKwOnlyDataClass(c=1, d=2), FrozenUnInitKwOnlyDataClass(side=3)]), ]) compat_version = {'"'+ compat_version +'"' if compat_version else None} From 6a399dff51f2258f61a580085aadb80fcbe6204c Mon Sep 17 00:00:00 2001 From: chenxuesdu Date: Mon, 26 Jan 2026 13:41:05 -0800 Subject: [PATCH 8/9] Update ChangeStreamDao to query different TVF for postgresSQL based on (#36667) the change stream partition mode For MUTABLE_KEY_RANGE change stream, use read_proto_bytes_, else use read_json_ --- .../beam/sdk/io/gcp/spanner/SpannerIO.java | 64 +++- .../changestreams/dao/ChangeStreamDao.java | 20 +- .../spanner/changestreams/dao/DaoFactory.java | 8 +- .../SpannerIOReadChangeStreamTest.java | 292 ++++++++++++------ .../SpannerChangeStreamErrorTest.java | 33 ++ .../dao/ChangeStreamDaoTest.java | 97 ++++++ 6 files changed, 416 insertions(+), 98 deletions(-) create mode 100644 sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamDaoTest.java diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java index 450710112a1b..bbce5fad82f4 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java @@ -46,6 +46,8 @@ import com.google.cloud.spanner.Options; import com.google.cloud.spanner.Options.RpcPriority; import com.google.cloud.spanner.PartitionOptions; +import com.google.cloud.spanner.ReadOnlyTransaction; +import com.google.cloud.spanner.ResultSet; import com.google.cloud.spanner.Spanner; import com.google.cloud.spanner.SpannerException; import com.google.cloud.spanner.SpannerOptions; @@ -1998,6 +2000,11 @@ && getInclusiveStartAt().toSqlTimestamp().after(getInclusiveEndAt().toSqlTimesta final MapperFactory mapperFactory = new MapperFactory(changeStreamDatabaseDialect); final ChangeStreamMetrics metrics = new ChangeStreamMetrics(); final RpcPriority rpcPriority = MoreObjects.firstNonNull(getRpcPriority(), RpcPriority.HIGH); + final SpannerAccessor spannerAccessor = + SpannerAccessor.getOrCreate(changeStreamSpannerConfig); + final boolean isMutableChangeStream = + isMutableChangeStream( + spannerAccessor.getDatabaseClient(), changeStreamDatabaseDialect, changeStreamName); final DaoFactory daoFactory = new DaoFactory( changeStreamSpannerConfig, @@ -2007,7 +2014,8 @@ && getInclusiveStartAt().toSqlTimestamp().after(getInclusiveEndAt().toSqlTimesta rpcPriority, input.getPipeline().getOptions().getJobName(), changeStreamDatabaseDialect, - metadataDatabaseDialect); + metadataDatabaseDialect, + isMutableChangeStream); final ActionFactory actionFactory = new ActionFactory(); final Duration watermarkRefreshRate = @@ -2689,4 +2697,58 @@ static String resolveSpannerProjectId(SpannerConfig config) { ? SpannerOptions.getDefaultProjectId() : config.getProjectId().get(); } + + @VisibleForTesting + static boolean isMutableChangeStream( + DatabaseClient databaseClient, Dialect dialect, String changeStreamName) { + String fetchedPartitionMode = fetchPartitionMode(databaseClient, dialect, changeStreamName); + if (fetchedPartitionMode.isEmpty() + || fetchedPartitionMode.equalsIgnoreCase("IMMUTABLE_KEY_RANGE")) { + return false; + } + return true; + } + + private static String fetchPartitionMode( + DatabaseClient databaseClient, Dialect dialect, String changeStreamName) { + try (ReadOnlyTransaction tx = databaseClient.readOnlyTransaction()) { + Statement statement; + if (dialect == Dialect.POSTGRESQL) { + statement = + Statement.newBuilder( + "select option_value\n" + + "from information_schema.change_stream_options\n" + + "where change_stream_name = $1 and option_name = 'partition_mode'") + .bind("p1") + .to(changeStreamName) + .build(); + } else { + statement = + Statement.newBuilder( + "select option_value\n" + + "from information_schema.change_stream_options\n" + + "where change_stream_name = @changeStreamName and option_name = 'partition_mode'") + .bind("changeStreamName") + .to(changeStreamName) + .build(); + } + ResultSet resultSet = tx.executeQuery(statement); + while (resultSet.next()) { + String value = resultSet.getString(0); + if (value != null) { + return value; + } + } + return ""; + } catch (RuntimeException e) { + // Log the failure (with stack trace) but rethrow so the caller still observes + // the error. + LOG.warn( + "Failed to fetch partition_mode for change stream '{}', dialect={} - will propagate exception", + changeStreamName, + dialect, + e); + throw e; + } + } } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamDao.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamDao.java index 3ef9c13f4714..0f32fa46cde7 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamDao.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamDao.java @@ -31,12 +31,12 @@ * as a {@link ResultSet}, which can be consumed until the stream is finished. */ public class ChangeStreamDao { - private final String changeStreamName; private final DatabaseClient databaseClient; private final RpcPriority rpcPriority; private final String jobName; private final Dialect dialect; + private final boolean isMutableChangeStream; /** * Constructs a change stream dao. All the queries performed by this class will be for the given @@ -53,12 +53,14 @@ public class ChangeStreamDao { DatabaseClient databaseClient, RpcPriority rpcPriority, String jobName, - Dialect dialect) { + Dialect dialect, + boolean isMutableChangeStream) { this.changeStreamName = changeStreamName; this.databaseClient = databaseClient; this.rpcPriority = rpcPriority; this.jobName = jobName; this.dialect = dialect; + this.isMutableChangeStream = isMutableChangeStream; } /** @@ -91,8 +93,18 @@ public ChangeStreamResultSet changeStreamQuery( String query = ""; Statement statement; if (this.isPostgres()) { - query = - "SELECT * FROM \"spanner\".\"read_json_" + changeStreamName + "\"($1, $2, $3, $4, null)"; + // Ensure we have determined whether change stream uses mutable key range + if (this.isMutableChangeStream) { + query = + "SELECT * FROM \"spanner\".\"read_proto_bytes_" + + changeStreamName + + "\"($1, $2, $3, $4, null)"; + } else { + query = + "SELECT * FROM \"spanner\".\"read_json_" + + changeStreamName + + "\"($1, $2, $3, $4, null)"; + } statement = Statement.newBuilder(query) .bind("p1") diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/DaoFactory.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/DaoFactory.java index 787abad02e02..67b58bace70f 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/DaoFactory.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/DaoFactory.java @@ -49,6 +49,7 @@ public class DaoFactory implements Serializable { private final String jobName; private final Dialect spannerChangeStreamDatabaseDialect; private final Dialect metadataDatabaseDialect; + private final boolean isMutableChangeStream; /** * Constructs a {@link DaoFactory} with the configuration to be used for the underlying instances. @@ -68,7 +69,8 @@ public DaoFactory( RpcPriority rpcPriority, String jobName, Dialect spannerChangeStreamDatabaseDialect, - Dialect metadataDatabaseDialect) { + Dialect metadataDatabaseDialect, + boolean isMutableChangeStream) { if (metadataSpannerConfig.getInstanceId() == null) { throw new IllegalArgumentException("Metadata instance can not be null"); } @@ -83,6 +85,7 @@ public DaoFactory( this.jobName = jobName; this.spannerChangeStreamDatabaseDialect = spannerChangeStreamDatabaseDialect; this.metadataDatabaseDialect = metadataDatabaseDialect; + this.isMutableChangeStream = isMutableChangeStream; } /** @@ -143,7 +146,8 @@ public synchronized ChangeStreamDao getChangeStreamDao() { spannerAccessor.getDatabaseClient(), rpcPriority, jobName, - this.spannerChangeStreamDatabaseDialect); + this.spannerChangeStreamDatabaseDialect, + this.isMutableChangeStream); } return changeStreamDaoInstance; } diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOReadChangeStreamTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOReadChangeStreamTest.java index 5fd3548a3004..589d831e1a45 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOReadChangeStreamTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOReadChangeStreamTest.java @@ -19,10 +19,23 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import com.google.auth.Credentials; import com.google.cloud.Timestamp; +import com.google.cloud.spanner.DatabaseClient; +import com.google.cloud.spanner.Dialect; import com.google.cloud.spanner.Options.RpcPriority; +import com.google.cloud.spanner.ReadOnlyTransaction; +import com.google.cloud.spanner.ResultSet; +import com.google.cloud.spanner.Statement; +import java.util.Arrays; +import java.util.Collection; import org.apache.beam.sdk.extensions.gcp.auth.TestCredential; import org.apache.beam.sdk.extensions.gcp.options.GcpOptions; import org.apache.beam.sdk.io.gcp.spanner.changestreams.MetadataSpannerConfigFactory; @@ -30,10 +43,15 @@ import org.junit.Before; import org.junit.Rule; import org.junit.Test; +import org.junit.experimental.runners.Enclosed; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameter; +import org.junit.runners.Parameterized.Parameters; +import org.mockito.ArgumentCaptor; -@RunWith(JUnit4.class) +@RunWith(Enclosed.class) public class SpannerIOReadChangeStreamTest { private static final String TEST_PROJECT = "my-project"; @@ -44,100 +62,192 @@ public class SpannerIOReadChangeStreamTest { private static final String TEST_METADATA_TABLE = "my-metadata-table"; private static final String TEST_CHANGE_STREAM = "my-change-stream"; - @Rule public final transient TestPipeline testPipeline = TestPipeline.create(); - - private SpannerConfig spannerConfig; - private SpannerIO.ReadChangeStream readChangeStream; - - @Before - public void setUp() throws Exception { - spannerConfig = - SpannerConfig.create() - .withProjectId(TEST_PROJECT) - .withInstanceId(TEST_INSTANCE) - .withDatabaseId(TEST_DATABASE); - - Timestamp startTimestamp = Timestamp.now(); - Timestamp endTimestamp = - Timestamp.ofTimeSecondsAndNanos( - startTimestamp.getSeconds() + 10, startTimestamp.getNanos()); - readChangeStream = - SpannerIO.readChangeStream() - .withSpannerConfig(spannerConfig) - .withChangeStreamName(TEST_CHANGE_STREAM) - .withMetadataInstance(TEST_METADATA_INSTANCE) - .withMetadataDatabase(TEST_METADATA_DATABASE) - .withMetadataTable(TEST_METADATA_TABLE) - .withRpcPriority(RpcPriority.MEDIUM) - .withInclusiveStartAt(startTimestamp) - .withInclusiveEndAt(endTimestamp); - } + /** Basic configuration tests using standard JUnit4. */ + @RunWith(JUnit4.class) + public static class ConfigurationTests { + @Rule public final transient TestPipeline testPipeline = TestPipeline.create(); + private SpannerConfig spannerConfig; + private SpannerIO.ReadChangeStream readChangeStream; + + @Before + public void setUp() { + spannerConfig = + SpannerConfig.create() + .withProjectId(TEST_PROJECT) + .withInstanceId(TEST_INSTANCE) + .withDatabaseId(TEST_DATABASE); + + readChangeStream = + SpannerIO.readChangeStream() + .withSpannerConfig(spannerConfig) + .withChangeStreamName(TEST_CHANGE_STREAM) + .withMetadataInstance(TEST_METADATA_INSTANCE) + .withMetadataDatabase(TEST_METADATA_DATABASE) + .withMetadataTable(TEST_METADATA_TABLE) + .withRpcPriority(RpcPriority.MEDIUM) + .withInclusiveStartAt(Timestamp.now()); + } + + @Test + public void testSetPipelineCredential() { + TestCredential testCredential = new TestCredential(); + // Set the credential in the pipeline options. + testPipeline.getOptions().as(GcpOptions.class).setGcpCredential(testCredential); + SpannerConfig changeStreamSpannerConfig = readChangeStream.buildChangeStreamSpannerConfig(); + SpannerConfig metadataSpannerConfig = + MetadataSpannerConfigFactory.create( + changeStreamSpannerConfig, TEST_METADATA_INSTANCE, TEST_METADATA_DATABASE); + assertNull(changeStreamSpannerConfig.getCredentials()); + assertNull(metadataSpannerConfig.getCredentials()); + + SpannerConfig changeStreamSpannerConfigWithCredential = + SpannerIO.buildSpannerConfigWithCredential( + changeStreamSpannerConfig, testPipeline.getOptions()); + SpannerConfig metadataSpannerConfigWithCredential = + SpannerIO.buildSpannerConfigWithCredential( + metadataSpannerConfig, testPipeline.getOptions()); + assertEquals(testCredential, changeStreamSpannerConfigWithCredential.getCredentials().get()); + assertEquals(testCredential, metadataSpannerConfigWithCredential.getCredentials().get()); + } + + @Test + public void testSetSpannerConfigCredential() { + TestCredential testCredential = new TestCredential(); + // Set the credential in the SpannerConfig. + spannerConfig = spannerConfig.withCredentials(testCredential); + readChangeStream = readChangeStream.withSpannerConfig(spannerConfig); + SpannerConfig changeStreamSpannerConfig = readChangeStream.buildChangeStreamSpannerConfig(); + SpannerConfig metadataSpannerConfig = + MetadataSpannerConfigFactory.create( + changeStreamSpannerConfig, TEST_METADATA_INSTANCE, TEST_METADATA_DATABASE); + assertEquals(testCredential, changeStreamSpannerConfig.getCredentials().get()); + assertEquals(testCredential, metadataSpannerConfig.getCredentials().get()); - @Test - public void testSetPipelineCredential() { - TestCredential testCredential = new TestCredential(); - // Set the credential in the pipeline options. - testPipeline.getOptions().as(GcpOptions.class).setGcpCredential(testCredential); - SpannerConfig changeStreamSpannerConfig = readChangeStream.buildChangeStreamSpannerConfig(); - SpannerConfig metadataSpannerConfig = - MetadataSpannerConfigFactory.create( - changeStreamSpannerConfig, TEST_METADATA_INSTANCE, TEST_METADATA_DATABASE); - assertNull(changeStreamSpannerConfig.getCredentials()); - assertNull(metadataSpannerConfig.getCredentials()); - - SpannerConfig changeStreamSpannerConfigWithCredential = - SpannerIO.buildSpannerConfigWithCredential( - changeStreamSpannerConfig, testPipeline.getOptions()); - SpannerConfig metadataSpannerConfigWithCredential = - SpannerIO.buildSpannerConfigWithCredential( - metadataSpannerConfig, testPipeline.getOptions()); - assertEquals(testCredential, changeStreamSpannerConfigWithCredential.getCredentials().get()); - assertEquals(testCredential, metadataSpannerConfigWithCredential.getCredentials().get()); + SpannerConfig changeStreamSpannerConfigWithCredential = + SpannerIO.buildSpannerConfigWithCredential( + changeStreamSpannerConfig, testPipeline.getOptions()); + SpannerConfig metadataSpannerConfigWithCredential = + SpannerIO.buildSpannerConfigWithCredential( + metadataSpannerConfig, testPipeline.getOptions()); + assertEquals(testCredential, changeStreamSpannerConfigWithCredential.getCredentials().get()); + assertEquals(testCredential, metadataSpannerConfigWithCredential.getCredentials().get()); + } + + @Test + public void testWithDefaultCredential() { + // Get the default credential, without setting any credentials in the pipeline + // options or SpannerConfig. + Credentials defaultCredential = + testPipeline.getOptions().as(GcpOptions.class).getGcpCredential(); + SpannerConfig changeStreamSpannerConfig = readChangeStream.buildChangeStreamSpannerConfig(); + SpannerConfig metadataSpannerConfig = + MetadataSpannerConfigFactory.create( + changeStreamSpannerConfig, TEST_METADATA_INSTANCE, TEST_METADATA_DATABASE); + assertNull(changeStreamSpannerConfig.getCredentials()); + assertNull(metadataSpannerConfig.getCredentials()); + + SpannerConfig changeStreamSpannerConfigWithCredential = + SpannerIO.buildSpannerConfigWithCredential( + changeStreamSpannerConfig, testPipeline.getOptions()); + SpannerConfig metadataSpannerConfigWithCredential = + SpannerIO.buildSpannerConfigWithCredential( + metadataSpannerConfig, testPipeline.getOptions()); + assertEquals( + defaultCredential, changeStreamSpannerConfigWithCredential.getCredentials().get()); + assertEquals(defaultCredential, metadataSpannerConfigWithCredential.getCredentials().get()); + } } - @Test - public void testSetSpannerConfigCredential() { - TestCredential testCredential = new TestCredential(); - // Set the credential in the SpannerConfig. - spannerConfig = spannerConfig.withCredentials(testCredential); - readChangeStream = readChangeStream.withSpannerConfig(spannerConfig); - SpannerConfig changeStreamSpannerConfig = readChangeStream.buildChangeStreamSpannerConfig(); - SpannerConfig metadataSpannerConfig = - MetadataSpannerConfigFactory.create( - changeStreamSpannerConfig, TEST_METADATA_INSTANCE, TEST_METADATA_DATABASE); - assertEquals(testCredential, changeStreamSpannerConfig.getCredentials().get()); - assertEquals(testCredential, metadataSpannerConfig.getCredentials().get()); - - SpannerConfig changeStreamSpannerConfigWithCredential = - SpannerIO.buildSpannerConfigWithCredential( - changeStreamSpannerConfig, testPipeline.getOptions()); - SpannerConfig metadataSpannerConfigWithCredential = - SpannerIO.buildSpannerConfigWithCredential( - metadataSpannerConfig, testPipeline.getOptions()); - assertEquals(testCredential, changeStreamSpannerConfigWithCredential.getCredentials().get()); - assertEquals(testCredential, metadataSpannerConfigWithCredential.getCredentials().get()); + /** Parameterized tests for Dialect and Partition Mode combinations. */ + @RunWith(Parameterized.class) + public static class PartitionModeTests { + + @Parameters(name = "{index}: dialect={0}, mode={1}, expected={2}") + public static Collection data() { + return Arrays.asList( + new Object[][] { + {Dialect.GOOGLE_STANDARD_SQL, "MUTABLE_KEY_RANGE", true}, + {Dialect.GOOGLE_STANDARD_SQL, "IMMUTABLE_KEY_RANGE", false}, + {Dialect.GOOGLE_STANDARD_SQL, "", false}, // Empty string case + {Dialect.POSTGRESQL, "MUTABLE_KEY_RANGE", true}, + {Dialect.POSTGRESQL, "IMMUTABLE_KEY_RANGE", false}, + {Dialect.POSTGRESQL, "", false} + }); + } + + @Parameter(0) + public Dialect dialect; + + @Parameter(1) + public String partitionMode; + + @Parameter(2) + public boolean expected; + + @Test + public void testIsMutableChangeStream() { + DatabaseClient databaseClient = mock(DatabaseClient.class); + ReadOnlyTransaction transaction = mock(ReadOnlyTransaction.class); + ResultSet resultSet = mock(ResultSet.class); + + when(databaseClient.readOnlyTransaction()).thenReturn(transaction); + when(transaction.executeQuery(any(Statement.class))).thenReturn(resultSet); + + // Handle the different return values for the mock ResultSet + if (partitionMode.isEmpty()) { + // If the partition mode is empty (e.g., the option is not set in Spanner), + // simulate an empty result set by having next() return false immediately. + when(resultSet.next()).thenReturn(false); // No row returned + } else { + // If a partition mode exists, simulate a result set containing exactly one row. + // next() returns true for the first call (row exists) and false for the second (end of + // stream). + when(resultSet.next()).thenReturn(true).thenReturn(false); + // When getString(0) is called to retrieve the 'option_value' column from the row, + // return the specified partitionMode string (e.g., "MUTABLE_KEY_RANGE"). + when(resultSet.getString(0)).thenReturn(partitionMode); + } + + boolean actual = SpannerIO.isMutableChangeStream(databaseClient, dialect, TEST_CHANGE_STREAM); + assertEquals(expected, actual); + + // Verify SQL Syntax: Captures the statement to check for dialect-specific + // syntax + ArgumentCaptor statementCaptor = ArgumentCaptor.forClass(Statement.class); + verify(transaction).executeQuery(statementCaptor.capture()); + String sql = statementCaptor.getValue().getSql(); + + // Ensure the SQL uses the correct parameter placeholder syntax for the given dialect. + // Different dialects have different requirements for how parameters are bound in queries. + if (dialect == Dialect.POSTGRESQL) { + // PostgreSQL-dialect Spanner databases use positional parameters (e.g., $1, $2) + assertTrue("PostgreSQL SQL should use $1", sql.contains("$1")); + } else { + // Google Standard SQL-dialect Spanner databases use named parameters (e.g., + // @changeStreamName) + assertTrue("GoogleSQL SQL should use @changeStreamName", sql.contains("@changeStreamName")); + } + } } - @Test - public void testWithDefaultCredential() { - // Get the default credential, without setting any credentials in the pipeline options or - // SpannerConfig. - Credentials defaultCredential = - testPipeline.getOptions().as(GcpOptions.class).getGcpCredential(); - SpannerConfig changeStreamSpannerConfig = readChangeStream.buildChangeStreamSpannerConfig(); - SpannerConfig metadataSpannerConfig = - MetadataSpannerConfigFactory.create( - changeStreamSpannerConfig, TEST_METADATA_INSTANCE, TEST_METADATA_DATABASE); - assertNull(changeStreamSpannerConfig.getCredentials()); - assertNull(metadataSpannerConfig.getCredentials()); - - SpannerConfig changeStreamSpannerConfigWithCredential = - SpannerIO.buildSpannerConfigWithCredential( - changeStreamSpannerConfig, testPipeline.getOptions()); - SpannerConfig metadataSpannerConfigWithCredential = - SpannerIO.buildSpannerConfigWithCredential( - metadataSpannerConfig, testPipeline.getOptions()); - assertEquals(defaultCredential, changeStreamSpannerConfigWithCredential.getCredentials().get()); - assertEquals(defaultCredential, metadataSpannerConfigWithCredential.getCredentials().get()); + /** Tests for error handling and exceptions. */ + @RunWith(JUnit4.class) + public static class ErrorTests { + + @Test(expected = RuntimeException.class) + public void testIsMutableChangeStream_PropagatesException() { + DatabaseClient databaseClient = mock(DatabaseClient.class); + ReadOnlyTransaction transaction = mock(ReadOnlyTransaction.class); + + // Mock the transaction creation + when(databaseClient.readOnlyTransaction()).thenReturn(transaction); + + // Simulate a database failure when executing the query + when(transaction.executeQuery(any(Statement.class))) + .thenThrow(new RuntimeException("Database connection failed")); + + // The method should log the error and rethrow the exception + SpannerIO.isMutableChangeStream(databaseClient, Dialect.GOOGLE_STANDARD_SQL, "test-stream"); + } } } diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangeStreamErrorTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangeStreamErrorTest.java index 835ca0a0f5a8..c8435bcfdffd 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangeStreamErrorTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangeStreamErrorTest.java @@ -331,6 +331,7 @@ public void testInvalidRecordReceivedWithDefaultSettings() { Timestamp.ofTimeSecondsAndNanos(startTimestamp.getSeconds(), startTimestamp.getNanos() + 1); mockGetDialect(); + mockChangeStreamOptions(); mockTableExists(); mockGetWatermark(startTimestamp); ResultSet getPartitionResultSet = mockGetParentPartition(startTimestamp, endTimestamp); @@ -382,6 +383,38 @@ public void testInvalidRecordReceivedWithDefaultSettings() { } } + private void mockChangeStreamOptions() { + Statement changeStreamOptionsStatement = + Statement.newBuilder( + "select option_value\n" + + "from information_schema.change_stream_options\n" + + "where change_stream_name = @changeStreamName and option_name = 'partition_mode'") + .bind("changeStreamName") + .to(TEST_CHANGE_STREAM) + .build(); + ResultSetMetadata changeStreamOptionsResultSetMetadata = + ResultSetMetadata.newBuilder() + .setRowType( + StructType.newBuilder() + .addFields( + Field.newBuilder() + .setName("option_value") + .setType(Type.newBuilder().setCode(TypeCode.STRING).build()) + .build()) + .build()) + .build(); + ResultSet changeStreamOptionsResultSet = + ResultSet.newBuilder() + .addRows( + ListValue.newBuilder() + .addValues(Value.newBuilder().setStringValue("NEW_VALUES").build()) + .build()) + .setMetadata(changeStreamOptionsResultSetMetadata) + .build(); + mockSpannerService.putPartialStatementResult( + StatementResult.query(changeStreamOptionsStatement, changeStreamOptionsResultSet)); + } + private void mockInvalidChangeStreamRecordReceived(Timestamp now, Timestamp after3Seconds) { Statement changeStreamQueryStatement = Statement.newBuilder( diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamDaoTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamDaoTest.java new file mode 100644 index 000000000000..9bdaa7b9fa5a --- /dev/null +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamDaoTest.java @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.gcp.spanner.changestreams.dao; + +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.cloud.spanner.DatabaseClient; +import com.google.cloud.spanner.Dialect; +import com.google.cloud.spanner.Options.RpcPriority; +import com.google.cloud.spanner.ReadOnlyTransaction; +import com.google.cloud.spanner.ResultSet; +import com.google.cloud.spanner.Statement; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +public class ChangeStreamDaoTest { + private DatabaseClient databaseClient; + private RpcPriority rpcPriority; + private static final String CHANGE_STREAM_NAME = "testCS"; + + @Before + public void setUp() { + databaseClient = mock(DatabaseClient.class); + rpcPriority = mock(RpcPriority.class); + } + + // New tests for PostgreSQL branch to verify the chosen TVF in the generated + // SQL. + @Test + public void testChangeStreamQueryPostgresMutable() { + // Arrange: single-use transaction for the actual change stream query + ReadOnlyTransaction singleUseTx = mock(ReadOnlyTransaction.class); + when(databaseClient.singleUse()).thenReturn(singleUseTx); + + ChangeStreamDao changeStreamDao = + new ChangeStreamDao( + CHANGE_STREAM_NAME, databaseClient, rpcPriority, "testjob", Dialect.POSTGRESQL, true); + + // Act: call the method that constructs and executes the statement + changeStreamDao.changeStreamQuery(null, null, null, 0L); + + // Assert: capture the Statement passed to singleUse().executeQuery and verify + // SQL + ArgumentCaptor captor = ArgumentCaptor.forClass(Statement.class); + verify(singleUseTx).executeQuery(captor.capture(), any(), any()); + Statement captured = captor.getValue(); + String sql = captured.getSql(); // adjust if different accessor is used + assertTrue( + "Expected SQL to contain read_proto_bytes_", + sql.contains("read_proto_bytes_" + CHANGE_STREAM_NAME)); + } + + @Test + public void testChangeStreamQueryPostgresImmutable() { + // Arrange: single-use transaction for the actual change stream query + ReadOnlyTransaction singleUseTx = mock(ReadOnlyTransaction.class); + when(databaseClient.singleUse()).thenReturn(singleUseTx); + + ResultSet queryResult = mock(ResultSet.class); + when(singleUseTx.executeQuery(any(), any(), any())).thenReturn(queryResult); + + ChangeStreamDao changeStreamDao = + new ChangeStreamDao( + CHANGE_STREAM_NAME, databaseClient, rpcPriority, "testjob", Dialect.POSTGRESQL, false); + + // Act + changeStreamDao.changeStreamQuery(null, null, null, 0L); + + // Assert + ArgumentCaptor captor = ArgumentCaptor.forClass(Statement.class); + verify(singleUseTx).executeQuery(captor.capture(), any(), any()); + Statement captured = captor.getValue(); + String sql = captured.getSql(); + assertTrue( + "Expected SQL to contain read_json_", sql.contains("read_json_" + CHANGE_STREAM_NAME)); + } +} From f8013e6a6b30e68392f15be34ddbbfe7ee0ea773 Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Mon, 26 Jan 2026 16:42:42 -0500 Subject: [PATCH 9/9] Fix str(WindowedValueCoder) crash when underlying coder isn't KV coder (#37406) --- sdks/python/apache_beam/coders/coders.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/coders/coders.py b/sdks/python/apache_beam/coders/coders.py index 0f73197f5cb1..ddcb80630487 100644 --- a/sdks/python/apache_beam/coders/coders.py +++ b/sdks/python/apache_beam/coders/coders.py @@ -1575,7 +1575,7 @@ def value_coder(self): def __repr__(self): return ( f'WindowedValueCoder[window_coder={self.window_coder}, ' - f'value_coder={self.value_coder()}]') + f'wrapped_value_coder={self.wrapped_value_coder}]') def __eq__(self, other): return (