ingore ct and tableReadersMap for serialization (#2488)

Author: VardhanThigle

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/reader/io/transform/AccumulatingTableReader.java

@@ -25,8 +25,6 @@
 import org.apache.beam.sdk.transforms.Flatten;
 import org.apache.beam.sdk.transforms.MapElements;
 import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.beam.sdk.transforms.SimpleFunction;
 import org.apache.beam.sdk.values.PBegin;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionList;
@@ -39,17 +37,25 @@
  */
 @AutoValue
 abstract class AccumulatingTableReader extends PTransform<PBegin, PCollectionTuple> {
-  public abstract ImmutableMap<SourceTableReference, PTransform<PBegin, PCollection<SourceRow>>>
-      tableTransforms();
+  // Avoid serializing tableTransform as it's needed only for expand.
+  private transient ImmutableMap<SourceTableReference, PTransform<PBegin, PCollection<SourceRow>>>
+      tableTransforms;
 
   public abstract TupleTag<SourceRow> sourceRowTag();
 
   public abstract TupleTag<SourceTableReference> sourceTableReferenceTag();
 
+  private AccumulatingTableReader setTableTransforms(
+      ImmutableMap<SourceTableReference, PTransform<PBegin, PCollection<SourceRow>>>
+          tableTransforms) {
+    this.tableTransforms = tableTransforms;
+    return this;
+  }
+
   @Override
   public PCollectionTuple expand(PBegin input) {
     ImmutableMap<SourceTableReference, PCollection<SourceRow>> tablePCollections =
-        this.tableTransforms().entrySet().stream()
+        this.tableTransforms.entrySet().stream()
             .collect(
                 ImmutableMap.toImmutableMap(
                     Entry::getKey,
@@ -93,7 +99,7 @@ static Builder builder(
   @AutoValue.Builder
   abstract static class Builder {
 
-    abstract ImmutableMap.Builder tableTransformsBuilder();
+    private ImmutableMap.Builder tableTransformsBuilder = new ImmutableMap.Builder();
 
     abstract Builder setSourceRowTag(TupleTag<SourceRow> sourceRowTag);
 
@@ -102,24 +108,14 @@ abstract static class Builder {
     Builder withTableReader(
         SourceTableReference sourceTableReference,
         PTransform<PBegin, PCollection<SourceRow>> tableReader) {
-      this.tableTransformsBuilder().put(sourceTableReference, tableReader);
+      this.tableTransformsBuilder.put(sourceTableReference, tableReader);
       return this;
     }
 
-    abstract AccumulatingTableReader build();
-  }
-
-  public class SourceTableReferenceWithCount extends SimpleFunction<Long, SourceTableReference>
-      implements SerializableFunction<Long, SourceTableReference> {
-    private SourceTableReference tableReference;
-
-    SourceTableReferenceWithCount(SourceTableReference tableReference) {
-      this.tableReference = tableReference;
-    }
+    abstract AccumulatingTableReader autoBuild();
 
-    @Override
-    public SourceTableReference apply(Long input) {
-      return this.tableReference.toBuilder().setRecordCount(input).build();
+    AccumulatingTableReader build() {
+      return autoBuild().setTableTransforms(this.tableTransformsBuilder.build());
     }
   }
 }

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/reader/io/transform/SourceTableReferenceWithCount.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright (C) 2025 Google LLC
+ *
+ * Licensed 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 com.google.cloud.teleport.v2.source.reader.io.transform;
+
+import com.google.cloud.teleport.v2.source.reader.io.schema.SourceTableReference;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.sdk.transforms.SimpleFunction;
+
+/** Generates SourceTableReference once the table read is complete. */
+public class SourceTableReferenceWithCount extends SimpleFunction<Long, SourceTableReference>
+    implements SerializableFunction<Long, SourceTableReference> {
+
+  private SourceTableReference tableReference;
+
+  SourceTableReferenceWithCount(SourceTableReference tableReference) {
+    this.tableReference = tableReference;
+  }
+
+  @Override
+  public SourceTableReference apply(Long input) {
+    return this.tableReference.toBuilder().setRecordCount(input).build();
+  }
+}

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/transformer/SourceRowToMutationDoFn.java

@@ -47,7 +47,7 @@ public abstract class SourceRowToMutationDoFn extends DoFn<SourceRow, RowContext
 
   private static final Logger LOG = LoggerFactory.getLogger(SourceRowToMutationDoFn.class);
 
-  private ISpannerMigrationTransformer sourceDbToSpannerTransformer;
+  private transient ISpannerMigrationTransformer sourceDbToSpannerTransformer;
 
   public void setSourceDbToSpannerTransformer(
       ISpannerMigrationTransformer sourceDbToSpannerTransformer) {