fix some issues

Signed-off-by: dragongu <andrewgu@vip.qq.com>

Author: dragongu

python/ray/data/_internal/execution/interfaces/physical_operator.py

@@ -38,6 +38,11 @@
 
 logger = logging.getLogger(__name__)
 
+# Timeout for waiting for metadata reference to become available (in seconds)
+METADATA_REF_WAIT_TIMEOUT_SECONDS = 0.1
+
+# Timeout for getting metadata from Ray object references (in seconds)
+METADATA_GET_TIMEOUT_SECONDS = 0.1
 
 # TODO(hchen): Ray Core should have a common interface for these two types.
 Waitable = Union[ray.ObjectRef, ObjectRefGenerator]
@@ -130,11 +135,11 @@ def on_data_ready(self, max_bytes_to_read: Optional[int]) -> int:
         bytes_read = 0
         while max_bytes_to_read is None or bytes_read < max_bytes_to_read:
             block_ref, meta_ref = self._pending_block_pair or (
-                ray.ObjectRef.nil(),
-                ray.ObjectRef.nil(),
+                None,
+                None,
             )
             try:
-                if block_ref.is_nil():
+                if block_ref is None:
                     block_ref = self._streaming_gen._next_sync(0)
                 if block_ref.is_nil():
                     # The generator currently doesn't have new output.
@@ -145,17 +150,24 @@ def on_data_ready(self, max_bytes_to_read: Optional[int]) -> int:
                 break
 
             try:
+                if meta_ref is None:
+                    meta_ref = self._streaming_gen._next_sync(
+                        METADATA_REF_WAIT_TIMEOUT_SECONDS
+                    )
                 if meta_ref.is_nil():
-                    meta_ref = self._streaming_gen._next_sync(1)
-                if meta_ref.is_nil():
-                    self._pending_block_pair = (block_ref, meta_ref)
+                    self._pending_block_pair = (block_ref, None)
                     break
                 meta_with_schema: "BlockMetadataWithSchema" = ray.get(
-                    meta_ref, timeout=1
+                    meta_ref, timeout=METADATA_GET_TIMEOUT_SECONDS
                 )
             except ray.exceptions.GetTimeoutError:
-                logger.warning(f"Get Meta timeout for (block_ref={block_ref.hex()})")
-                self._pending_block_pair = (block_ref, meta_ref)
+                logger.warning(
+                    f"Metadata object not ready for block_ref={block_ref.hex()[:8]}... "
+                    f"(operator={self.__class__.__name__}). "
+                    f"Metadata may still be computing or worker may have failed and object is being reconstructed. "
+                    f"Will retry in next iteration."
+                )
+                self._pending_block_pair = (block_ref, None)
                 break
             except StopIteration:
                 # The generator should always yield 2 values (block and metadata)