[Bug]: ValueError: Invalid DisplayDataItem when using AsSingleton for side input (need better parameter validation before it hits this) #25860

import argparse
import json
import logging
from typing import Any, Dict, List

import apache_beam as beam
import apache_beam.io.fileio
from apache_beam.options.pipeline_options import PipelineOptions

def is_valid(json_file):
  """
  This function is used as a filter.
  
    Args:
        json_file: json file corresponding to one episode.

    Returns:
        True if the episode is valid, False otherwise.
  """
  with json_file.open() as f:
    first_line = f.readline()
    first_line_json = json.loads(first_line)
    return first_line_json['username'] == "username"


def run(
    input_raw_data: str,
    output_path: str,
    beam_args: List[str] = None,
) -> None:
    """Build and run the pipeline."""
    options = PipelineOptions(beam_args, save_main_session=True, streaming=False)

    with beam.Pipeline(options=options) as pipeline:
        filtered_files = (
            pipeline
            | 'Match file paths' >> beam.io.fileio.MatchFiles(input_raw_data)
            | 'Read file paths' >> beam.io.fileio.ReadMatches()
            | 'Filter correct episodes' >> beam.Filter(is_valid)
        )
        file_count = (
            filtered_files
            | 'Count files' >> beam.combiners.Count.Globally()
        )
        #removing the file_count_integer as a side input fixes the error.
        file_count_integer = beam.pvalue.AsSingleton(file_count)
        (   
            filtered_files
            | 'Read files' >> beam.Map(lambda x: x.read_utf8())
            | 'Write to files' >> beam.io.WriteToText(
                output_path,
                file_name_suffix='.json',
                num_shards=file_count_integer)
        )


if __name__ == "__main__":
    logging.getLogger().setLevel(logging.INFO)

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--output-path",
        required=True,
        help="Name of the folder to output the cleaned data files. ",
    )
    parser.add_argument(
        "--input-raw-data",
        required=True,
        help="Name of the folder containing the raw data files." ,
    )
    args, beam_args = parser.parse_known_args()

    run(
        input_raw_data=args.input_raw_data,
        output_path=args.output_path,
        beam_args=beam_args,
    )

python -m script.py --region europe-west1 --input gs://path-to-files/*.json --output gs://path-to-utputs/results/outputs --runner DataflowRunner --project google_project --temp_location gs://temp_bucket/tmp/