Just wanted to update that while i got my workaround done, i also tried the following suggested in a previous post that i failed to notice before. The following worked and results got correct.
ray.data.context.DatasetContext.get_current().execution_options.preserve_order = True