Added Labels In DetermineSchema

[DarshanSP19]

• While using more than two concat Dimension apache-beam is throwing below Error.

RuntimeError: A transform with label "DetermineSchema/_NestDim" already exists in the pipeline. To apply a transform with a specified label write pvalue | "label" >> transform

• Adding labels for every recursive call of PTransform in this PR.
• Find more discussion here Is it possible to specify more than one MergeDim to a recipe file pattern? #402.

[cisaacstern]

Thanks @DarshanSP19! Looking good.

It would be great to add a test before merging.

Could you share the code which caused the error you reported in #402? Or if you prefer not to share the full code, perhaps you could share a pseudo-code version of what you were running that prompted your bug report in #402.

Once I see the use case that this PR fixes, I can recommend how to add it to the tests. 🙏

[DarshanSP19]

Hey @cisaacstern I have tried with below snippets.

concat_dim = ConcatDim("time", date_range)
merge_dim = MergeDim("chunk", chunks)
pressure_levels = MergeDim("pressure", pressure_level)

def make_path(time: datetime.datetime, chunk: str, pressure: str) -> str:
    if chunk in pressure_level_chunk_data:
        return f"gs://gcp-public-data-arco-era5/raw/date-variable-pressure_level/{time.year:04d}/{time.month:02d}/{time.day:02d}/{chunk}/{pressure}.nc"
    else:
        return f"gs://gcp-public-data-arco-era5/raw/date-variable-single_level/{time.year:04d}/{time.month:02d}/{time.day:02d}/{chunk}/surface.nc"

pattern = FilePattern(make_path, merge_dim, concat_dim, pressure_levels,  file_type='netcdf3')

with beam.Pipeline(argv=pipeline_args) as p:
    pressure_level_schema = (
        p 
        | "Create" >> beam.Create(pattern.items())
        | "OpenURLWithFSSpec" >> OpenURLWithFSSpec()
        | "OpenDatasets" >> OpenWithXarray(file_type=pattern.file_type, copy_to_local=True, load=True)
        | "DetermineSchema" >> DetermineSchema(combine_dims=pattern.combine_dim_keys)
        | "PrepareZarrTarget" >> PrepareZarrTarget(known_args.output)
    )

Also the data is publicly available here.

[cisaacstern]

👋 @DarshanSP19 thanks for your patience on this! On further thought I think this is simple enough to defer testing to a separate PR. I've opened #624 to track that. Thanks for this fix!