code: def testFlatten(self):
      with DeepCopyTest._MakeBeamPipeline() as p:
          create_1 = p | "Create1" >> beam.Create([(1, "a"), (2, "b")])
          create_2 = p | "Create2" >> beam.Create([(3, "c")])
          created = (create_1, create_2) | "Flatten1" >> beam.Flatten()
          grouped1 = (
              created
              | "PreGroup1"
              >> beam.Map(
                  lambda x: DeepCopyTest._CountingIdentityFn(
                      "PreGroup1", x
                  )
              )
              | "GBK1" >> beam.GroupByKey()
          )
          grouped2 = (
              p
              | beam.Create([(1, "a"), (2, "b"), (3, "c")])
              | "PreGroup2"
              >> beam.Map(
                  lambda x: DeepCopyTest._CountingIdentityFn(
                      "PreGroup2", x
                  )
              )
              | "GBK2" >> beam.GroupByKey()
          )
          modified1 = grouped1 | "Add1" >> beam.Map(
              DeepCopyTest._MakeAdd1CountingIdentityFn("Add1")
          )
          modified2 = grouped2 | "Add2" >> beam.Map(
              DeepCopyTest._MakeAdd1CountingIdentityFn("Add2")
          )
          flattened = (
              modified1,
              modified2,
          ) | "Flatten2" >> beam.Flatten()
          modified3 = flattened | "Add3" >> beam.Map(
              DeepCopyTest._MakeAdd1CountingIdentityFn("Add3")
          )
  
          copied = deep_copy.deep_copy(modified3)
  
          # Check that deep copy was performed.
          self.assertIsNot(
              copied.producer.inputs[0], modified3.producer.inputs[0]
          )
          self.assertIsNot(
              copied.producer.inputs[0].producer.inputs[0],
              modified3.producer.inputs[0].producer.inputs[0],
          )
          self.assertIsNot(
              copied.producer.inputs[0].producer.inputs[1],
              modified3.producer.inputs[0].producer.inputs[1],
          )
  
          # Check that copy stops at materialization boundary.
          self.assertIs(
              copied.producer.inputs[0]
              .producer.inputs[0]
              .producer.inputs[0],
              modified3.producer.inputs[0]
              .producer.inputs[0]
              .producer.inputs[0],
          )
          self.assertIs(
              copied.producer.inputs[0]
              .producer.inputs[1]
              .producer.inputs[0],
              modified3.producer.inputs[0]
              .producer.inputs[1]
              .producer.inputs[0],
          )
  
      # Check counts of processed items.
      self.assertEqual(DeepCopyTest._counts["PreGroup1"], 3)
      self.assertEqual(DeepCopyTest._counts["PreGroup2"], 3)
      self.assertEqual(DeepCopyTest._counts["Add1"], 6)
      self.assertEqual(DeepCopyTest._counts["Add2"], 6)
      self.assertEqual(DeepCopyTest._counts["Add3"], 12)
  
imports: import collections
  import apache_beam
  from apache_beam import pvalue
  from apache_beam.transforms import resources
  from tensorflow_transform.beam import deep_copy
  from tensorflow_transform.beam import test_helpers
  import unittest
label_desc: Use image node_id 8 for calling the DeepCopyTest obj's underlying member method code with example usage: obj.testFlatten() without return types