delegated operator fixes (#4908)

fiftyone/factory/repos/delegated_operation.py

@@ -262,6 +262,8 @@ def update_run_state(
             else None
         )
 
+        needs_pipeline_update = False
+
         if run_state == ExecutionRunState.COMPLETED:
             update = {
                 "$set": {
@@ -272,10 +274,11 @@ def update_run_state(
                 }
             }
 
-        if outputs_schema:
-            update["$set"]["metadata.outputs_schema"] = {
-                "$ifNull": [outputs_schema, {}]
-            }
+            if outputs_schema:
+                update["$set"]["metadata.outputs_schema"] = (
+                    outputs_schema or {}
+                )
+                needs_pipeline_update = True
 
         elif run_state == ExecutionRunState.FAILED:
             update = {
@@ -325,9 +328,15 @@ def update_run_state(
         if required_state is not None:
             collection_filter["run_state"] = required_state
 
+        # Using pipeline update instead of a single update doc fixes a case
+        #   where `metadata` is null and so accessing the dotted field
+        #   `metadata.output_schema` creates the document instead of erroring.
+        if needs_pipeline_update:
+            update = [update]
+
         doc = self._collection.find_one_and_update(
             filter=collection_filter,
-            update=[update],
+            update=update,
             return_document=pymongo.ReturnDocument.AFTER,
         )
 

tests/unittests/operators/delegated_tests.py

@@ -11,6 +11,7 @@
 from unittest import mock
 from unittest.mock import patch
 
+import bson
 import pytest
 
 import fiftyone
@@ -484,6 +485,35 @@ def test_sets_progress(
         self.assertEqual(doc.status.label, "halfway there")
         self.assertIsNotNone(doc.status.updated_at)
 
+    def test_output_schema_null_metadata(
+        self, mock_get_operator, mock_operator_exists
+    ):
+        mock_outputs = MockOutputs()
+        doc = self.svc.queue_operation(
+            operator="@voxelfiftyone/operator/foo",
+            delegation_target="test_target",
+            context=ExecutionContext(request_params={"foo": "bar"}),
+        )
+
+        # Set metadata to null instead of being unset, to test that corner case
+        self.svc._repo._collection.find_one_and_update(
+            {"_id": bson.ObjectId(doc.id)}, {"$set": {"metadata": None}}
+        )
+
+        self.svc.set_completed(
+            doc.id,
+            result=ExecutionResult(outputs_schema=mock_outputs.to_json()),
+        )
+
+        doc = self.svc.get(doc_id=doc.id)
+        self.assertEqual(doc.run_state, ExecutionRunState.COMPLETED)
+        self.assertEqual(
+            doc.metadata,
+            {
+                "outputs_schema": mock_outputs.to_json(),
+            },
+        )
+
     @patch(
         "fiftyone.core.odm.utils.load_dataset",
     )