OpenDAL如何實(shí)現(xiàn)數(shù)據(jù)讀取

隨著 OpenDAL 社區(qū)的不斷發(fā)展，新的抽象在不斷增加，為新的貢獻(xiàn)者參與開發(fā)帶來了不少負(fù)擔(dān)，不少維護(hù)者都希望對(duì) OpenDAL 的內(nèi)部實(shí)現(xiàn)有更深入的了解。與此同時(shí)，OpenDAL 的核心設(shè)計(jì)已經(jīng)很長(zhǎng)時(shí)間沒有大幅度的變化，為寫一個(gè)內(nèi)部實(shí)現(xiàn)系列提供了可能。我想現(xiàn)在是時(shí)候?qū)懸幌盗嘘P(guān)于 OpenDAL 內(nèi)部實(shí)現(xiàn)的文章，從維護(hù)者的角度來闡述 OpenDAL 如何設(shè)計(jì)，如何實(shí)現(xiàn)以及如何擴(kuò)展。在 OpenDAL v0.40 即將發(fā)布之際，希望這系列文章能夠更好的幫助社區(qū)理解過去，掌握現(xiàn)在，并確定未來。

第一篇文章會(huì)先聊聊 OpenDAL 最常使用的數(shù)據(jù)讀取功能，我會(huì)從最外層的接口開始，然后按照 OpenDAL 的調(diào)用順序來逐步展開。讓我們開始吧！

整體框架

在開始介紹具體的 OpenDAL 接口之前，我們首先熟悉一下 OpenDAL 項(xiàng)目。

OpenDAL[1]是一個(gè) Apache Incubator 項(xiàng)目，旨在幫助用戶從各種存儲(chǔ)服務(wù)中以統(tǒng)一的方式便捷高效訪問數(shù)據(jù)。它的項(xiàng)目愿景[2]是 “自由訪問數(shù)據(jù)”：

?Free from services: 任意服務(wù)都能通過原生接口自由訪問

?Free from implementations: 無(wú)論底層實(shí)現(xiàn)如何，都可以通過統(tǒng)一的方式調(diào)用

?Free to integrate: 能夠自由地與各種服務(wù)，語(yǔ)言集成

?Free to zero cost: 用戶不需要為用不到的功能付出開銷

在這套理念的基礎(chǔ)上，OpenDAL Rust Core 可以主要分成以下組成部分：

?Operator: 對(duì)用戶暴露的外層接口

?Layers: 不同中間件的具體實(shí)現(xiàn)

?Services: 不同服務(wù)的具體實(shí)現(xiàn)

所以從宏觀的角度上來看，OpenDAL 的數(shù)據(jù)讀取調(diào)用?？雌饋頃?huì)像是這樣：

所有 Layers 和 Services 都實(shí)現(xiàn)了統(tǒng)一了 Accessor 接口，在進(jìn)行 Operator 構(gòu)建時(shí)會(huì)抹除所有的類型信息。對(duì) Operator 來說，不管用戶使用什么服務(wù)或者增加了多少中間件，所有的調(diào)用邏輯都是一致的。這一設(shè)計(jì)將 OpenDAL 的 API 拆分成了 Public API 和 Raw API 兩層，其中 Public API 直接暴露給用戶，提供便于使用的上層接口，而 Raw API 則是面向 OpenDAL 內(nèi)部開發(fā)者提供，維護(hù)統(tǒng)一的內(nèi)部接口，并提供一些便利的實(shí)現(xiàn)。

Operator

OpenDAL 的 Operator API 會(huì)盡可能遵循一致的調(diào)用范式，減少用戶的學(xué)習(xí)和使用成本。以read為例，OpenDAL 提供了以下 API:

?op.read(path): 將指定文件全部?jī)?nèi)容讀出

?op.reader(path): 創(chuàng)建一個(gè) Reader 用來做流式讀取

?op.read_with(path).range(1..1024): 使用指定參數(shù)來讀取文件內(nèi)容，比如說 range

?op.reader_with(path).range(1..1024): 使用指定參數(shù)來創(chuàng)建 Reader 做流式讀取

不難看出read更像是一個(gè)語(yǔ)法糖，用來方便用戶快速地進(jìn)行文件讀取而不需要考慮AsyncRead等各種 trait。而reader則給予了用戶更多的靈活度，實(shí)現(xiàn)了AsyncSeek，AsyncRead等社區(qū)廣泛使用的 trait，允許用戶更靈活的讀取數(shù)據(jù)。read_with和reader_with則通過 Future Builder 系列函數(shù)，幫助用戶以更自然的方式來指定各種參數(shù)。

Operator 內(nèi)部的邏輯看起來會(huì)是這樣：

它的主要工作是面向用戶封裝接口：

?完成OpRead的構(gòu)建

?調(diào)用Accessor提供的read函數(shù)

?將返回的值包裹為Reader并在Reader的基礎(chǔ)上實(shí)現(xiàn)AsyncSeek，AsyncRead等接口

Layers

這里有一個(gè)隱藏的小秘密是 OpenDAL 會(huì)自動(dòng)為 Service 套上一些 Layer 以實(shí)現(xiàn)一些內(nèi)部邏輯，截止到本文完成的時(shí)候，OpenDAL 自動(dòng)增加的 Layer 包括：

?ErrorContextLayer: 為所有的 Operation 返回的 error 注入 context 信息，比如scheme，path等

?CompleteLayer: 為服務(wù)補(bǔ)全必須的能力，比如說為 s3 增加 seek 支持

?TypeEraseLayer: 實(shí)現(xiàn)類型擦除，將Accessor中的關(guān)聯(lián)類型統(tǒng)一擦除，讓用戶使用時(shí)不需要攜帶泛型參數(shù)

這里的ErrorContextLayer和TypeEraseLayer都比較簡(jiǎn)單不再贅述，重點(diǎn)聊聊CompleteLayer，它旨在以零開銷的方式為 OpenDAL 返回的Reader增加seek或者next支持，讓用戶不需要再重復(fù)實(shí)現(xiàn)。OpenDAL 在早期版本中通過不同的函數(shù)調(diào)用來返回Reader和SeekableReader，但是用戶的實(shí)際反饋并不是很好，幾乎所有用戶都在使用SeekableReader。因此后續(xù) OpenDAL 在重構(gòu)中將 seek 支持作為第一優(yōu)先級(jí)加入了內(nèi)部的Readtrait 中：

pubtraitRead:Unpin+Send+Sync{
///Readbytesasynchronously.
fnpoll_read(&mutself,cx:&mutContext<'_>,buf:&mut[u8])->Poll>;

///Seekasynchronously.
///
///Returns`Unsupported`errorifunderlyingreaderdoesn'tsupportseek.
fnpoll_seek(&mutself,cx:&mutContext<'_>,pos:io::SeekFrom)->Poll>;

///Stream[`Bytes`]fromunderlyingreader.
///
///Returns`Unsupported`errorifunderlyingreaderdoesn'tsupportstream.
///
///ThisAPIexistsforavoidingbytescopyinginsideasyncruntime.
///Userscanpollbytesfromunderlyingreaderanddecidewhento
///read/consumethem.
fnpoll_next(&mutself,cx:&mutContext<'_>)->Poll>>;
}

在 OpenDAL 中實(shí)現(xiàn)一個(gè)服務(wù)的讀取能力就需要實(shí)現(xiàn)這個(gè) trait，這是一個(gè)內(nèi)部接口，不會(huì)直接暴露給用戶，其中：

?poll_read是最基礎(chǔ)的要求，所有服務(wù)都必須實(shí)現(xiàn)這一接口。

?當(dāng)服務(wù)原生支持seek時(shí)，可以實(shí)現(xiàn)poll_seek，OpenDAL 會(huì)進(jìn)行正確的 dispatch，比如說 local fs；

?而當(dāng)服務(wù)原生支持next，即返回流式的 Bytes 時(shí)，可以實(shí)現(xiàn)poll_next，比如說基于 HTTP 的服務(wù)，他們底層是一個(gè) TCP Stream，hyper 會(huì)將其封裝為一個(gè) bytes stream。

通過Readtrait，OpenDAL 確保所有服務(wù)都能盡可能地暴露自己的原生支持能力，從而提供對(duì)不同服務(wù)都能實(shí)現(xiàn)高效的讀取。

在此 trait 的基礎(chǔ)上，OpenDAL 會(huì)根據(jù)各個(gè)服務(wù)支持的能力來進(jìn)行補(bǔ)全：

?seek/next 都支持：直接返回

?不支持 next: 使用StreamableReader進(jìn)行封裝以模擬 next 支持

?不支持 seek: 使用ByRangeSeekableReader進(jìn)行封裝以模擬 seek 支持

?seek/next 均不支持：同時(shí)進(jìn)行兩種封裝

ByRangeSeekableReader主要利用了服務(wù)支持 range read 的能力，當(dāng)用戶進(jìn)行 seek 的時(shí)候就 drop 當(dāng)前 reader 并在指定的位置發(fā)起新的請(qǐng)求。

OpenDAL 通過CompleteLayer暴露出一個(gè)統(tǒng)一的 Reader 實(shí)現(xiàn)，用戶不需要考慮底層服務(wù)是否支持 seek，OpenDAL 總是會(huì)選擇最優(yōu)的方式來發(fā)起請(qǐng)求。

Services

經(jīng)過 Layers 的補(bǔ)全之后，就到調(diào)用 Service 具體實(shí)現(xiàn)的地方，這里分別以最常見的兩類服務(wù)fs和s3來舉例說明數(shù)據(jù)是如何讀取的。

Service fs

tokio::File實(shí)現(xiàn)了tokio::AsyncRead和tokio::AsyncSeek，通過使用async_compat::Compat，我們將其轉(zhuǎn)化為了futures::AsyncRead和futures::AsyncSeek。在此基礎(chǔ)上，我們提供了內(nèi)置的函數(shù)oio::into_read_from_file將其轉(zhuǎn)化為實(shí)現(xiàn)了oio::Read的類型，最終的類型名為：oio::FromFileReader>。

oio::into_read_from_file實(shí)現(xiàn)中沒有什么特別復(fù)雜的地方，read 和 seek 基本上都是在調(diào)用傳入的 File 類型提供的函數(shù)。比較麻煩的地方是關(guān)于 seek 和 range 的正確處理：seek 到 range 右側(cè)是允許的行為，此時(shí)不會(huì)報(bào)錯(cuò)，read 也只會(huì)返回空，但是 seek 到 range 左側(cè)是非法行為，Reader 必須返回InvalidInput以便于上層正確處理。

有趣的歷史：當(dāng)初這塊實(shí)現(xiàn)的時(shí)候有問題，還是在 fuzz 測(cè)試中發(fā)現(xiàn)的。

Services s3

S3 是一個(gè)基于 HTTP 的服務(wù)，opendal 提供了大量基于 HTTP 的封裝以幫助開發(fā)者重用邏輯，只需要構(gòu)建請(qǐng)求，并返回構(gòu)造好的 Body 即可。OpenDAL Raw API 封裝了一套基于 reqwest 的接口，HTTP GET 接口會(huì)返回一個(gè)Response：

///IncomingAsyncBodycarriesthecontentreturnedbyremoteservers.
pubstructIncomingAsyncBody{
///#TODO
///
///hyperreturns`implStream>`butwecan't
///writethetypesinstable.Sowewillboxhere.
///
///After[TAIT](https://rust-lang.github.io/rfcs/2515-type_alias_impl_trait.html)
///hasbeenstable,wecanchange`IncomingAsyncBody`into`IncomingAsyncBody`.
inner:oio::Streamer,
size:Option,
consumed:u64,
chunk:Option,
}