From fde7e205358cd71e67eb5f247b95cafc11f78f9c Mon Sep 17 00:00:00 2001 From: Andreas Motl Date: Sun, 2 Aug 2026 19:28:12 +0200 Subject: [PATCH 1/2] Filesystem: Add reader for ORC format --- docs/changelog.md | 1 + src/dlt_filesystem/source/adapter.py | 3 ++ src/dlt_filesystem/source/format/readers.py | 47 ++++++++++++++++++++ src/dlt_filesystem/source/format/registry.py | 1 + src/dlt_filesystem/testing/writer.py | 12 +++++ tests/dlt_filesystem/format/test_orc.py | 19 ++++++++ 6 files changed, 83 insertions(+) create mode 100644 tests/dlt_filesystem/format/test_orc.py diff --git a/docs/changelog.md b/docs/changelog.md index 5a271332c..9f4018877 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -8,6 +8,7 @@ - Filesystem: Fail ingestion when a concrete source path matches no file while keeping unmatched glob selections valid. Thanks, @hampsterx. - Filesystem: Added rsync source connector. Thanks, @oferchen. +- Filesystem: Added reader for ORC format ## 2026/07/27 v0.8.0 diff --git a/src/dlt_filesystem/source/adapter.py b/src/dlt_filesystem/source/adapter.py index 0914402d9..d2b3ea09d 100644 --- a/src/dlt_filesystem/source/adapter.py +++ b/src/dlt_filesystem/source/adapter.py @@ -34,6 +34,7 @@ read_jsonl, read_msgpack, read_ods, + read_orc, read_parquet, read_xml, read_yaml, @@ -78,6 +79,8 @@ def readers( filesystem_resource | dlt.transformer(name="read_msgpack", max_table_nesting=0)(read_msgpack), filesystem_resource + | dlt.transformer(name="read_orc", max_table_nesting=0)(read_orc), + filesystem_resource | dlt.transformer(name="read_cbor", max_table_nesting=0)(read_cbor), filesystem_resource | dlt.transformer(name="read_xml", max_table_nesting=0)(read_xml), diff --git a/src/dlt_filesystem/source/format/readers.py b/src/dlt_filesystem/source/format/readers.py index 4234a183b..12174e1fa 100644 --- a/src/dlt_filesystem/source/format/readers.py +++ b/src/dlt_filesystem/source/format/readers.py @@ -52,6 +52,24 @@ def _polars_csv_symbols() -> Dict[str, Any]: } +def _polars_orc_symbols() -> Dict[str, Any]: + """Symbols needed to resolve `polars.read_orc`'s type hints for casting reader hints.""" + + import fsspec + import pyarrow + from pandas import DataFrame + from pandas._typing import DtypeBackend, FilePath, ReadBuffer + + return { + "DataFrame": DataFrame, + "DtypeBackend": DtypeBackend, + "FilePath": FilePath, + "fsspec": fsspec, + "pyarrow": pyarrow, + "ReadBuffer": ReadBuffer, + } + + def _polars_spreadsheet_symbols() -> Dict[str, Any]: """Symbols needed to cast reader hint values for `polars.read_excel` and `polars.read_ods`.""" from typing import Sequence @@ -400,6 +418,31 @@ def read_spreadsheet( yield dlt.mark.with_table_name(rows, sheet_name) +def read_orc( + items: Iterator[FileItemDict], + **kwargs, +) -> Iterator[TDataItems]: + """Reader for ORC files.""" + + import pandas as pd + + reader = pd.read_orc + + kwargs = cast_kwargs_to_signature(reader, kwargs, symbols=_polars_orc_symbols()) + + for file_obj in items: + with file_obj.open() as f: + rec = reader(f, dtype_backend="pyarrow", **kwargs).to_records(index=False) + # Turn numpy recarray record into a Python dictionary. + # https://gist.github.com/rlabbe/d574eeac63fd126b2fcd1dc390cc3257 + # https://stackoverflow.com/a/67324508 + if rec.dtype is None or rec.dtype.names is None: + yield rec + return + result = {name: rec[name] for name in rec.dtype.names} + yield result + + def read_jsonl( items: Iterator[FileItemDict], chunksize: int = 1000 ) -> Iterator[TDataItems]: @@ -642,6 +685,10 @@ def read_bson(self) -> DltResource: def read_msgpack(self) -> DltResource: """MessagePack reader resource.""" + @copy_sig(read_orc) + def read_orc(self) -> DltResource: + """ORC reader resource (pyarrow).""" + @copy_sig(read_cbor) def read_cbor(self) -> DltResource: """CBOR reader resource.""" diff --git a/src/dlt_filesystem/source/format/registry.py b/src/dlt_filesystem/source/format/registry.py index 965802115..1cbaa0de6 100644 --- a/src/dlt_filesystem/source/format/registry.py +++ b/src/dlt_filesystem/source/format/registry.py @@ -6,6 +6,7 @@ "csv_headless": "read_csv_headless", "jsonl": "read_jsonl", "ods": "read_ods", + "orc": "read_orc", "parquet": "read_parquet", # bson is read-only: the file:// destination's WRITE_FORMATS is a separate tuple. "bson": "read_bson", diff --git a/src/dlt_filesystem/testing/writer.py b/src/dlt_filesystem/testing/writer.py index 935a34f0b..95023a75e 100644 --- a/src/dlt_filesystem/testing/writer.py +++ b/src/dlt_filesystem/testing/writer.py @@ -1,3 +1,9 @@ +import typing + +if typing.TYPE_CHECKING: + import pandas as pd + + def write_bson(path, docs): """Write BSON documents concatenated into a single file (on-disk mongodump form).""" import bson @@ -27,6 +33,12 @@ def write_msgpack(path, rows, **packb_kwargs): return path +def write_orc(path, df: "pd.DataFrame"): + """Write dataframe to ORC file.""" + df.to_orc(path) + return path + + def write_xml(path, text): """Write raw XML ``text`` to ``path`` as UTF-8 bytes.""" with open(path, "wb") as f: diff --git a/tests/dlt_filesystem/format/test_orc.py b/tests/dlt_filesystem/format/test_orc.py new file mode 100644 index 000000000..382962a9c --- /dev/null +++ b/tests/dlt_filesystem/format/test_orc.py @@ -0,0 +1,19 @@ +import pandas as pd + +from dlt_filesystem.source.fsspec.local import LocalFilesystemSource +from dlt_filesystem.testing.writer import write_orc + + +def _read_via_source(path): + """Read a local ORC file end-to-end through the shared filesystem reader.""" + return list(LocalFilesystemSource().dlt_source(f"file://{path}", "")) + + +# --- end-to-end reader (fsspec, no Docker) --- + + +def test_reads_single_top_level_object(tmp_path): + """A single top-level ORC map loads as one record.""" + data = pd.DataFrame.from_records([{"id": 1, "name": "alice"}]) + path = write_orc(tmp_path / "one.orc", data) + assert _read_via_source(path) == [{"id": 1, "name": "alice"}] From abf2c7c9e0eb9e612b7e2e35c935ae983c7a1d68 Mon Sep 17 00:00:00 2001 From: Andreas Motl Date: Mon, 3 Aug 2026 01:59:40 +0200 Subject: [PATCH 2/2] ORC: Implement suggestions by CodeRabbit --- src/dlt_filesystem/source/format/readers.py | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/src/dlt_filesystem/source/format/readers.py b/src/dlt_filesystem/source/format/readers.py index 12174e1fa..259d856b3 100644 --- a/src/dlt_filesystem/source/format/readers.py +++ b/src/dlt_filesystem/source/format/readers.py @@ -52,8 +52,8 @@ def _polars_csv_symbols() -> Dict[str, Any]: } -def _polars_orc_symbols() -> Dict[str, Any]: - """Symbols needed to resolve `polars.read_orc`'s type hints for casting reader hints.""" +def _pandas_orc_symbols() -> Dict[str, Any]: + """Symbols needed to resolve `pandas.read_orc`'s type hints for casting reader hints.""" import fsspec import pyarrow @@ -428,19 +428,13 @@ def read_orc( reader = pd.read_orc - kwargs = cast_kwargs_to_signature(reader, kwargs, symbols=_polars_orc_symbols()) + kwargs = cast_kwargs_to_signature(reader, kwargs, symbols=_pandas_orc_symbols()) + kwargs.setdefault("dtype_backend", "pyarrow") for file_obj in items: with file_obj.open() as f: - rec = reader(f, dtype_backend="pyarrow", **kwargs).to_records(index=False) - # Turn numpy recarray record into a Python dictionary. - # https://gist.github.com/rlabbe/d574eeac63fd126b2fcd1dc390cc3257 - # https://stackoverflow.com/a/67324508 - if rec.dtype is None or rec.dtype.names is None: - yield rec - return - result = {name: rec[name] for name in rec.dtype.names} - yield result + df = reader(f, **kwargs) + yield df.to_dict(orient="records") def read_jsonl(