From c6533150afdbff49aefe63104429940aab4f5494 Mon Sep 17 00:00:00 2001 From: SuperSaiyan101 Date: Sat, 31 Oct 2026 11:10:05 +0530 Subject: [PATCH 1/2] Port ReadSpannerSchemaTest off deprecated DoFnTester --- .../io/gcp/spanner/ReadSpannerSchemaTest.java | 95 +++++++++++-------- 1 file changed, 57 insertions(+), 38 deletions(-) diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/ReadSpannerSchemaTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/ReadSpannerSchemaTest.java index 300c657c7f49..123ab1856ac9 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/ReadSpannerSchemaTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/ReadSpannerSchemaTest.java @@ -32,11 +32,12 @@ import com.google.cloud.spanner.Type; import java.util.Arrays; import java.util.List; +import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.transforms.Create; -import org.apache.beam.sdk.transforms.DoFnTester; +import org.apache.beam.sdk.transforms.ParDo; import org.apache.beam.sdk.transforms.View; -import org.apache.beam.sdk.transforms.windowing.GlobalWindow; +import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionView; import org.junit.Before; import org.junit.Rule; @@ -148,25 +149,35 @@ public void simple() throws Exception { .withServiceFactory(serviceFactory); PCollectionView dialectView = - pipeline.apply(Create.of(Dialect.GOOGLE_STANDARD_SQL)).apply(View.asSingleton()); - pipeline.run(); - DoFnTester tester = - DoFnTester.of(new ReadSpannerSchema(config, dialectView)); - tester.setSideInput(dialectView, GlobalWindow.INSTANCE, Dialect.GOOGLE_STANDARD_SQL); - - List schemas = tester.processBundle(Arrays.asList((Void) null)); - - assertEquals(1, schemas.size()); - - SpannerSchema schema = schemas.get(0); + pipeline + .apply("CreateDialect", Create.of(Dialect.GOOGLE_STANDARD_SQL)) + .apply(View.asSingleton()); + PCollection schemas = + pipeline + .apply("CreateVoid", Create.of((Void) null)) + .apply( + ParDo.of(new ReadSpannerSchema(config, dialectView)).withSideInputs(dialectView)); + + PAssert.that(schemas) + .satisfies( + iterable -> { + List schemaList = + org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists + .newArrayList(iterable); + assertEquals(1, schemaList.size()); + + SpannerSchema schema = schemaList.get(0); + assertEquals(1, schema.getTables().size()); + + SpannerSchema.Column column = SpannerSchema.Column.create("key", Type.int64()); + SpannerSchema.KeyPart keyPart = SpannerSchema.KeyPart.create("key", false); + + assertThat(schema.getColumns("test"), contains(column)); + assertThat(schema.getKeyParts("test"), contains(keyPart)); + return null; + }); - assertEquals(1, schema.getTables().size()); - - SpannerSchema.Column column = SpannerSchema.Column.create("key", Type.int64()); - SpannerSchema.KeyPart keyPart = SpannerSchema.KeyPart.create("key", false); - - assertThat(schema.getColumns("test"), contains(column)); - assertThat(schema.getKeyParts("test"), contains(keyPart)); + pipeline.run(); } @Test @@ -186,24 +197,32 @@ public void pgSimple() throws Exception { .withServiceFactory(serviceFactory); PCollectionView dialectView = - pipeline.apply(Create.of(Dialect.POSTGRESQL)).apply(View.asSingleton()); - pipeline.run(); - DoFnTester tester = - DoFnTester.of(new ReadSpannerSchema(config, dialectView)); - tester.setSideInput(dialectView, GlobalWindow.INSTANCE, Dialect.POSTGRESQL); - - List schemas = tester.processBundle(Arrays.asList((Void) null)); - - assertEquals(1, schemas.size()); - - SpannerSchema schema = schemas.get(0); + pipeline.apply("CreateDialect", Create.of(Dialect.POSTGRESQL)).apply(View.asSingleton()); + PCollection schemas = + pipeline + .apply("CreateVoid", Create.of((Void) null)) + .apply( + ParDo.of(new ReadSpannerSchema(config, dialectView)).withSideInputs(dialectView)); + + PAssert.that(schemas) + .satisfies( + iterable -> { + List schemaList = + org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists + .newArrayList(iterable); + assertEquals(1, schemaList.size()); + + SpannerSchema schema = schemaList.get(0); + assertEquals(1, schema.getTables().size()); + + SpannerSchema.Column column = SpannerSchema.Column.create("key", Type.int64()); + SpannerSchema.KeyPart keyPart = SpannerSchema.KeyPart.create("key", false); + + assertThat(schema.getColumns("test"), contains(column)); + assertThat(schema.getKeyParts("test"), contains(keyPart)); + return null; + }); - assertEquals(1, schema.getTables().size()); - - SpannerSchema.Column column = SpannerSchema.Column.create("key", Type.int64()); - SpannerSchema.KeyPart keyPart = SpannerSchema.KeyPart.create("key", false); - - assertThat(schema.getColumns("test"), contains(column)); - assertThat(schema.getKeyParts("test"), contains(keyPart)); + pipeline.run(); } } From 22365178e58dbae70aaf790ee1f876244594bfe3 Mon Sep 17 00:00:00 2001 From: SuperSaiyan101 Date: Sat, 31 Oct 2026 11:20:25 +0530 Subject: [PATCH 2/2] Address review feedback: import Lists from vendored Guava --- .../beam/sdk/io/gcp/spanner/ReadSpannerSchemaTest.java | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/ReadSpannerSchemaTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/ReadSpannerSchemaTest.java index 123ab1856ac9..4c351773eeb2 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/ReadSpannerSchemaTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/ReadSpannerSchemaTest.java @@ -39,6 +39,7 @@ import org.apache.beam.sdk.transforms.View; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionView; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -161,9 +162,7 @@ public void simple() throws Exception { PAssert.that(schemas) .satisfies( iterable -> { - List schemaList = - org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists - .newArrayList(iterable); + List schemaList = Lists.newArrayList(iterable); assertEquals(1, schemaList.size()); SpannerSchema schema = schemaList.get(0); @@ -207,9 +206,7 @@ public void pgSimple() throws Exception { PAssert.that(schemas) .satisfies( iterable -> { - List schemaList = - org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists - .newArrayList(iterable); + List schemaList = Lists.newArrayList(iterable); assertEquals(1, schemaList.size()); SpannerSchema schema = schemaList.get(0);