Preserve preprocessing outputs in Grain pipelines

[maitry63]

This PR fixes, when a preprocessing function is executed inside a Grain data pipeline, its eager outputs should be returned without passing through convert_preprocessing_outputs().

Currently, @preprocessing_function always converts the output after the preprocessing function returns. This can convert eager Python/NumPy outputs unnecessarily and can change ragged Python outputs such as list[list[...]].

This change reuses Keras core's existing in_grain_data_pipeline() detection and skips output conversion when running inside a Grain data pipeline.

Outside Grain pipelines, the existing convert_preprocessing_outputs() behavior is unchanged.

Fixes: #2945

Added regression tests covering:

• Python ragged outputs remain list[list[...]] inside Grain pipelines.
• Outputs continue to be converted normally outside Grain pipelines.

Checklist

• I have added all the necessary unit tests for my change.
• I have verified that my change does not break existing code and works with all backends (TensorFlow, JAX, and PyTorch).
• My PR is based on the latest changes of the main branch (if unsure, rebase the code).
• I have followed the Keras Hub Model contribution guidelines in making these changes.
• I have followed the Keras Hub API design guidelines in making these changes.
• I have signed the Contributor License Agreement.

[gemini-code-assist]

Code Review

This pull request updates the preprocessing wrapper in tensor_utils.py to skip output conversion when running inside a Grain data pipeline, and adds corresponding unit tests. The review feedback suggests wrapping the import of in_grain_data_pipeline in a try-except block to prevent ImportError on older versions of Keras 3, ensuring backward compatibility.

[gemini-code-assist]

Importing in_grain_data_pipeline directly from keras.src.utils.backend_utils can cause an ImportError on older versions of Keras 3 where this function is not yet defined. To maintain backward compatibility and prevent import failures, we should wrap this import in a try-except block and provide a fallback implementation that returns False.

Suggested change

	from keras.src.utils.backend_utils import in_grain_data_pipeline
	try:
	from keras.src.utils.backend_utils import in_grain_data_pipeline
	except ImportError:
	def in_grain_data_pipeline():
	return False

References

1. Demand Robustness: Do not accept fragile code. If the proposed code is not robust enough or lacks proper error handling, explicitly tell the author why the current approach is brittle and what must be done to reinforce it. ^(link)

[maitry63]

With the latest 3.15.1 Keras version in_grain_data_pipeline is available. I verified the direct import against the released latest package, so I kept the direct import.

[divyashreepathihalli]

fix never fires on the eager/python path - convert_preprocessing_outputs_python needs the same guard

[divyashreepathihalli]

blanket skip regresses ragged/string tf outputs to tf.RaggedTensor- convert to NumPy/lists instead of skipping

[maitry63]

Hi @divyashreepathihalli , thank you for the review.
Added the in_grain_data_pipeline() check to convert_preprocessing_outputs_python so the fix now correctly fires on eager and Python-based execution paths.

Removed the blanket return x inside Grain pipelines.

[divyashreepathihalli]

input conversion still creates backend tensors in Grain workers

[maitry63]

Added the in_grain_data_pipeline() check to the top of convert_preprocessing_inputs to ensure input conversion correctly bypasses backend tensor creation inside Grain workers.

[divyashreepathihalli]

mocked detection can't catch either bug- add a real grain.MapDataset integration test

[maitry63]

Replaced the mocked patch test with a real grain.MapDataset integration test.