Add json line support for DocumentLoaderOperator

[jeff3071]

Why

DocumentLoaderOperator supports regular JSON, but it does not recognize JSON Lines (.jsonl) files.

What

• Add built-in .jsonl parsing without additional dependencies.
• Produce one document per non-empty line and assign consecutive item_index metadata.
• Support JSON Lines from file paths, directory discovery, and source_bytes with file_type=".jsonl".

Different with JSON file

For example, this regular JSON file produces two documents because its top-level array is expanded:

[
  {"body": "First"},
  {"body": "Second"}
]

In a JSON Lines file, each non-empty line produces exactly one document. The array on the first line remains one document instead of being expanded, so this file also produces two documents:

[{"body": "First"}, {"body": "Second"}]
{"body": "Third"}

Testing

Before

::group::Log message source details
/root/airflow/logs/dag_id=example_document_loader_jsonl/run_id=manual__2026-08-29T06:23:56.559631+00:00/task_id=load_events/attempt=1.log
::endgroup::
[2026-08-29T06:23:57.671540Z] INFO - ::group::Pre Execute
Task Identity ti_id=01a04c30-6aa2-7411-a073-f1ad1073cd42 dag_id=example_document_loader_jsonl task_id=load_events run_id=manual__2026-08-29T06:23:56.559631+00:00 try_number=1 map_index=-1
[2026-08-29T06:24:03.707696Z] INFO - DAG bundles loaded: dags-folder
[2026-08-29T06:24:03.708918Z] INFO - Filling up the DagBag from /files/dags/example_document_loader_jsonl.py
[2026-08-29T06:24:03.781146Z] INFO - Worker startup parse complete bundle_name=dags-folder  bundle_version=null  dag_file=example_document_loader_jsonl.py  bundle_prepare_ms=4  dag_file_parse_ms=72 
[2026-08-29T06:24:03.824981Z] INFO - ::endgroup::
[2026-08-29T06:24:03.825318Z] ERROR - Task failed with exceptionValueError: No parser registered for extension '.jsonl'. Supported extensions: .csv, .docx, .json, .md, .pdf, .txt. Set 'parser' explicitly to override auto-detection.

After

::group::Log message source details
/root/airflow/logs/dag_id=example_document_loader_jsonl/run_id=manual__2026-08-29T06:10:10.890844+00:00/task_id=load_events/attempt=1.log
::endgroup::
[2026-08-29T06:10:11.897592Z] INFO - ::group::Pre Execute
Task Identity ti_id=01a04c23-d167-7cbb-b0a1-888ca1cbeed1 dag_id=example_document_loader_jsonl task_id=load_events run_id=manual__2026-08-29T06:10:10.890844+00:00 try_number=1 map_index=-1
[2026-08-29T06:10:17.924766Z] INFO - DAG bundles loaded: dags-folder
[2026-08-29T06:10:17.926232Z] INFO - Filling up the DagBag from /files/dags/example_document_loader_jsonl.py
[2026-08-29T06:10:18.003334Z] INFO - Worker startup parse complete bundle_name=dags-folder  bundle_version=null  dag_file=example_document_loader_jsonl.py  bundle_prepare_ms=3  dag_file_parse_ms=77 
[2026-08-29T06:10:18.046544Z] INFO - ::endgroup::
[2026-08-29T06:10:18.047021Z] INFO - Parsed 3 documents from 1 file(s)
[2026-08-29T06:10:18.047139Z] INFO - ::group::Post Execute
[2026-08-29T06:10:18.047330Z] INFO - Pushing xcom ti=RuntimeTaskInstance(id=UUID('01a04c23-d167-7cbb-b0a1-888ca1cbeed1'), task_id='load_events', dag_id='example_document_loader_jsonl', run_id='manual__2026-08-29T06:10:10.890844+00:00', try_number=1, dag_version_id=UUID('01a04c22-161b-7e24-bb58-fbac6568e066'), map_index=-1, hostname='d5873b0b3ad5', context_carrier={'traceparent': '00-a8201c08aed315888a0215944bfaf514-c055bbb704c1e9ff-00'}, queue='default', task=<Task(DocumentLoaderOperator): load_events>, bundle_instance=LocalDagBundle(name=dags-folder), max_tries=0, start_date=datetime.datetime(2026, 8, 29, 6, 10, 17, 912797, tzinfo=datetime.timezone.utc), end_date=None, state=<TaskInstanceState.RUNNING: 'running'>, is_mapped=False, rendered_map_index=None, sentry_integration='') 
[2026-08-29T06:10:18.095744Z] INFO - ::endgroup::

---

Was generative AI tooling used to co-author this PR?

• Yes — Codex (GPT-5.6-sol)

Generated-by: Codex (GPT-5.6-sol) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[SameerMesiah97]

Could we add coverage for an invalid JSON line? Since each line is parsed separately, the raw JSONDecdeError will generally report line 1 rather than the line’s position in the JSONL file.

[jeff3071]

Great catch! Fixed in b6033b0. JSONDecodeError now reports the correct line number.