code: def test_generate_slices_sql(self):
      input_record_batches = [
          pa.RecordBatch.from_arrays(
              [
                  pa.array([[1], [2, 1], [3], [2, 1, 1], [3]]),
                  pa.array(
                      [
                          ["dog"],
                          ["cat"],
                          ["wolf"],
                          ["dog", "wolf"],
                          ["wolf"],
                      ]
                  ),
              ],
              ["a", "b"],
          ),
          pa.RecordBatch.from_arrays(
              [pa.array([[1]]), pa.array([["dog"]]), pa.array([[1]])],
              ["a", "b", "c"],
          ),
          pa.RecordBatch.from_arrays(
              [pa.array([[1]]), pa.array([["cat"]]), pa.array([[1]])],
              ["a", "b", "d"],
          ),
          pa.RecordBatch.from_arrays(
              [pa.array([[1]]), pa.array([["cat"]]), pa.array([[1]])],
              ["a", "b", "e"],
          ),
          pa.RecordBatch.from_arrays(
              [pa.array([[1]]), pa.array([["cat"]]), pa.array([[1]])],
              ["a", "b", "f"],
          ),
      ]
      record_batch_with_metadata = pa.RecordBatch.from_arrays(
          [pa.array([[1]]), pa.array([["cat"]])], ["a", "b"]
      )
      record_batch_with_metadata = pa.RecordBatch.from_arrays(
          arrays=record_batch_with_metadata.columns,
          schema=record_batch_with_metadata.schema.with_metadata(
              {b"foo": "bar"}
          ),
      )
      input_record_batches.append(record_batch_with_metadata)
      slice_sql = """
          SELECT
            STRUCT(a, b)
          FROM
            example.a, example.b
      """
  
      with beam.Pipeline() as pipeline:
          # pylint: disable=no-value-for-parameter
          result = (
              pipeline
              | "Create"
              >> beam.Create(input_record_batches, reshuffle=False)
              | "GenerateSlicesSql"
              >> beam.ParDo(
                  slicing_util.GenerateSlicesSqlDoFn(
                      slice_sqls=[slice_sql]
                  )
              )
          )
  
          # pylint: enable=no-value-for-parameter
  
          def check_result(got):
              try:
                  self.assertLen(got, 18)
                  expected_slice_keys = [
                      "a_1_b_dog",
                      "a_1_b_cat",
                      "a_2_b_cat",
                      "a_2_b_dog",
                      "a_1_b_wolf",
                      "a_2_b_wolf",
                      "a_3_b_wolf",
                      "a_1_b_dog",
                      "a_1_b_cat",
                      "a_1_b_cat",
                      "a_1_b_cat",
                      "a_1_b_cat",
                  ] + [constants.DEFAULT_SLICE_KEY] * 6
                  actual_slice_keys = [
                      slice_key for (slice_key, _) in got
                  ]
                  self.assertCountEqual(
                      expected_slice_keys, actual_slice_keys
                  )
  
              except AssertionError as err:
                  raise util.BeamAssertException(err)
  
          util.assert_that(result, check_result)
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import sys
  import unittest
  from absl.testing import absltest
  import apache_beam
  from apache_beam.testing import util
  import pyarrow
  from tensorflow_data_validation import constants
  from tensorflow_data_validation.utils import slicing_util
  from tfx_bsl.public.proto import slicing_spec_pb2
  from google.protobuf import text_format
label_desc: Use image node_id 13 for calling the SlicingUtilTest obj's underlying member method code with example usage: obj.test_generate_slices_sql() without return types