[Bug]: [RunInference] WatchFilePattern may fail to correctly identify the most recent model #40638

What happened?

WatchFilePattern helps update a model in a RunInference pipeline on the fly. It monitors a location to find recently uploaded models and then has some logic to decide which model is the most recent one.

The logic is flawed for several reasons:

• _GetLatestFileByTimeStamp is a stateful DoFn. In Beam, state is partitioned per key, and right before calling _GetLatestFileByTimeStamp, we key each file name with its path:
  

  beam/sdks/python/apache_beam/ml/inference/utils.py

  Line 157 in 1fc615f

  | "AttachKey" >> beam.Map(lambda x: (x.path, x))

  . Because keys are different and state is partitioned per key, we don't have an overview of the entirety of available models. Each new file path gets a fresh TIME_STATE initialized to _START_TIME_STAMP, files are never compared against the latest timestamp of previously seen files, so an older file uploaded after _START_TIME_STAMP may still be emitted as the "latest" model.

• if more than one model is detected in the window during the poll, each with it own name, _ConvertIterToSingleton will emit a list of elements, this is well-illustrated by the existing test

  beam/sdks/python/apache_beam/ml/inference/utils_test.py

  Line 99 in 1fc615f

  assert_that(files_pc, equal_to(['', 'path3.py', 'path4.py']))

  . When RunInference consumes this via beam.pvalue.AsSingleton(...), it will crash since the value is not a singleton.

• Checking for timestamps might not be reliable for file systems that preserve the creation time when copying a file into a location. This is not a concern on GCS, since the timestamp is defined by object upload time, but if model is read from a local folder, reverting to an older model might not create desired results if some other model has a newer timestamp than the desired model to which we want to revert the processing. Though maybe a user would need to go out of their way to mess up the metadata, likely this is minor.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Prism Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner