Fix DocumentLoaderOperator validation errors naming the wrong argument (#73053)

* Fix DocumentLoaderOperator validation errors naming the wrong argument

Two of the operator's argument checks reported something other than what
actually went wrong.

`file_type` is required with `source_bytes`, and whether it was supplied is
knowable without rendering, so it now raises in `__init__` alongside the
`source_path`/`source_bytes` pair rather than waiting for the task to run.

The two checks that remain in `execute()` are reached only when a supplied
template field rendered to None, but both reported provision failures. A user
who passed `source_path="{{ ... }}"` that rendered away was told to "Provide
exactly one of 'source_path' or 'source_bytes'" -- advice they had already
followed. Both messages now name the field that rendered to None.

The comment above those checks claimed provision "already happened in
__init__" for both fields, which was never true of `file_type`. #70503 makes
this file the reference other operators copy, so the comment now describes the
split it actually implements.

The rendered-to-None test drove the state by assigning the attribute directly,
which passes whether or not the field is still rendered. It now goes through
real templating under `render_template_as_native_obj`, and the template-fields
test renders instead of asserting tuple membership.

* Qualify the file_type parse-time claim for mapped tasks

expand() validates argument names only and defers construction to unmap(),
which runs on the worker, so the check is a run-time error for a mapped task.

Author: kaxil

providers/common/ai/src/airflow/providers/common/ai/operators/document_loader.py

@@ -75,9 +75,12 @@ class DocumentLoaderOperator(BaseOperator):
         ``ObjectStoragePath`` for cloud URIs (``aws_default``,
         ``google_cloud_default``, ...). Ignored for local paths.
     :param source_bytes: Raw file bytes, typically from XCom.
-    :param file_type: File extension hint when using ``source_bytes``
-        (e.g. ``".pdf"``). Also accepted with ``source_path`` to override
-        auto-detection.
+    :param file_type: File extension hint (e.g. ``".pdf"``). Required when
+        using ``source_bytes``, since bytes carry no extension to detect.
+        Omitting it is rejected when the operator is constructed -- Dag parse
+        time for a regular task, run time for a mapped one, since ``expand()``
+        validates argument names only and defers construction to ``unmap()``.
+        Optional with ``source_path``, where it overrides auto-detection.
     :param parser: Parsing backend selection. ``"auto"`` (default) picks the
         backend from the file extension.
     :param file_extensions: When ``source_path`` is a directory or glob,
@@ -140,6 +143,8 @@ def __init__(
             raise ValueError("Provide exactly one of 'source_path' or 'source_bytes', not both.")
         if source_path is None and source_bytes is None:
             raise ValueError("Provide exactly one of 'source_path' or 'source_bytes'.")
+        if source_bytes is not None and file_type is None:
+            raise ValueError("'file_type' is required when using 'source_bytes' (e.g. '.pdf').")
         self.source_path = source_path
         self.source_conn_id = source_conn_id
         self.source_bytes = source_bytes
@@ -152,15 +157,21 @@ def __init__(
         self.json_text_field = json_text_field
 
     def execute(self, context: Context) -> list[dict[str, Any]]:
-        # file_type and source_path can each be *supplied* (as non-None argument) yet still
-        # render to None. These aren't provision checks (that already happened in __init__);
-        # they guard the rendered value itself, since _parse_bytes/_resolve_files need a real
-        # value to work with. Checking this in __init__ would validate the unrendered template
-        # string instead of the value actually used here.
+        # Provision -- whether an argument was supplied at all -- is settled in __init__.
+        # Both guards below exist for a different reason: file_type and source_path are
+        # template fields, so a supplied argument can still arrive here as None once it has
+        # been rendered. __init__ cannot catch that, because it only ever sees the unrendered
+        # template string; _parse_bytes and _resolve_files need the rendered value to be real.
         if self.source_bytes is not None and self.file_type is None:
-            raise ValueError("'file_type' is required when using 'source_bytes' (e.g. '.pdf').")
+            raise ValueError(
+                "'file_type' was supplied but rendered to None. Check the template or the "
+                "upstream XCom value it resolves from."
+            )
         if self.source_bytes is None and self.source_path is None:
-            raise ValueError("Provide exactly one of 'source_path' or 'source_bytes'.")
+            raise ValueError(
+                "'source_path' was supplied but rendered to None. Check the template or the "
+                "upstream XCom value it resolves from."
+            )
 
         if self.source_bytes is not None:
             if TYPE_CHECKING:

providers/common/ai/tests/unit/common/ai/operators/test_document_loader.py

@@ -23,28 +23,26 @@
 import pytest
 
 from airflow.providers.common.ai.operators.document_loader import DocumentLoaderOperator
+from airflow.sdk import DAG
 
 
 class TestDocumentLoaderInit:
     def test_template_fields_render_source_path_and_metadata(self):
-        """
-        Behavioral check that the templated fields actually get rendered.
-        Replaces the previous tautological assertion that just round-tripped
-        the class attribute.
-        """
+        """The templated fields are substituted by render_template_fields, not merely listed."""
         op = DocumentLoaderOperator(
             task_id="test",
             source_path="/data/{{ ds }}/*.pdf",
-            file_type="{{ var.value.preferred_ext }}",
+            file_type="{{ params.preferred_ext }}",
             metadata_fields={"run_id": "{{ run_id }}"},
         )
-        # Make sure each one is in template_fields so render_template_fields
-        # would substitute them.
-        assert "source_path" in op.template_fields
-        assert "file_type" in op.template_fields
-        assert "file_extensions" in op.template_fields
-        assert "parser" in op.template_fields
-        assert "metadata_fields" in op.template_fields
+
+        op.render_template_fields(
+            context={"ds": "2026-01-01", "run_id": "manual__1", "params": {"preferred_ext": ".pdf"}}
+        )
+
+        assert op.source_path == "/data/2026-01-01/*.pdf"
+        assert op.file_type == ".pdf"
+        assert op.metadata_fields == {"run_id": "manual__1"}
         # source_bytes intentionally not templated -- Jinja stringifies bytes
         # to their repr, which would break binary parsing.
         assert "source_bytes" not in op.template_fields
@@ -58,23 +56,40 @@ def test_neither_source_raises(self):
         with pytest.raises(ValueError, match="Provide exactly one"):
             DocumentLoaderOperator(task_id="test")
 
-    def test_source_bytes_without_file_type_raises(self):
-        # file_type is a template field, so this check only fires at execute() time.
-        op = DocumentLoaderOperator(task_id="test", source_bytes=b"hello")
-        with pytest.raises(ValueError, match="file_type"):
-            op.execute(context={})
+    def test_source_bytes_without_file_type_raises_at_construction(self):
+        # Whether file_type was supplied at all is knowable without rendering, so it is a
+        # Dag-parse-time error like the source_path/source_bytes pair above it.
+        with pytest.raises(ValueError, match="'file_type' is required"):
+            DocumentLoaderOperator(task_id="test", source_bytes=b"hello")
+
+    def test_empty_bytes_without_file_type_raises_at_construction(self):
+        with pytest.raises(ValueError, match="'file_type' is required"):
+            DocumentLoaderOperator(task_id="test", source_bytes=b"")
 
-    def test_empty_bytes_without_file_type_raises(self):
-        op = DocumentLoaderOperator(task_id="test", source_bytes=b"")
-        with pytest.raises(ValueError, match="file_type"):
+    def test_source_path_rendering_to_none_raises(self):
+        """A supplied source_path that renders to None raises ValueError, not TypeError.
+
+        Driven through real templating rather than by assigning the attribute, so the test
+        fails if the field ever stops being rendered.
+        """
+        with DAG(dag_id="native", schedule=None, render_template_as_native_obj=True):
+            op = DocumentLoaderOperator(task_id="test", source_path="{{ none }}")
+
+        op.render_template_fields(context={})
+        assert op.source_path is None
+
+        with pytest.raises(ValueError, match="'source_path' was supplied but rendered to None"):
             op.execute(context={})
 
-    def test_source_path_none_after_render_raises(self):
-        # source_path can render to None even when supplied -- must raise ValueError,
-        # not a TypeError from _resolve_files.
-        op = DocumentLoaderOperator(task_id="test", source_path="{{ none }}")
-        op.source_path = None  # simulate the rendered value, bypassing real templating
-        with pytest.raises(ValueError, match="Provide exactly one"):
+    def test_file_type_rendering_to_none_raises(self):
+        """file_type passes the __init__ guard when supplied, then renders away to None."""
+        with DAG(dag_id="native", schedule=None, render_template_as_native_obj=True):
+            op = DocumentLoaderOperator(task_id="test", source_bytes=b"hello", file_type="{{ none }}")
+
+        op.render_template_fields(context={})
+        assert op.file_type is None
+
+        with pytest.raises(ValueError, match="'file_type' was supplied but rendered to None"):
             op.execute(context={})