Apache OpenDAL™ 内部实现：数据读取

随着 Apache OpenDAL™ 社区的不断发展，新的抽象在不断增加，为新的贡献者参与开发带来了不少负担，不少维护者都希望对 OpenDAL 的内部实现有更深入的了解。与此同时，OpenDAL 的核心设计已经很长时间没有大幅度的变化，为写一个内部实现系列提供了可能。我想现在是时候写一系列关于 OpenDAL 内部实现的文章，从维护者的角度来阐述 OpenDAL 如何设计，如何实现以及如何扩展。在 OpenDAL v0.40 即将发布之际，希望这系列文章能够更好的帮助社区理解过去，掌握现在，并确定未来。

第一篇文章会先聊聊 OpenDAL 最常使用的数据读取功能，我会从最外层的接口开始，然后按照 OpenDAL 的调用顺序来逐步展开。让我们开始吧！

整体框架

在开始介绍具体的 OpenDAL 接口之前，我们首先熟悉一下 OpenDAL 项目。

OpenDAL 是一个 Apache Incubator 项目，旨在帮助用户从各种存储服务中以统一的方式便捷高效访问数据。它的项目愿景是 “自由访问数据”：

Free from services: 任意服务都能通过原生接口自由访问
Free from implementations: 无论底层实现如何，都可以通过统一的方式调用
Free to integrate: 能够自由地与各种服务，语言集成
Free to zero cost: 用户不需要为用不到的功能付出开销

在这套理念的基础上，OpenDAL Rust Core 可以主要分成以下组成部分：

Operator: 对用户暴露的外层接口
Layers: 不同中间件的具体实现
Services: 不同服务的具体实现

所以从宏观的角度上来看，OpenDAL 的数据读取调用栈看起来会像是这样：

所有 Layers 和 Services 都实现了统一了 Accessor 接口，在进行 Operator 构建时会抹除所有的类型信息。对 Operator 来说，不管用户使用什么服务或者增加了多少中间件，所有的调用逻辑都是一致的。这一设计将 OpenDAL 的 API 拆分成了 Public API 和 Raw API 两层，其中 Public API 直接暴露给用户，提供便于使用的上层接口，而 Raw API 则是面向 OpenDAL 内部开发者提供，维护统一的内部接口，并提供一些便利的实现。

Operator

OpenDAL 的 Operator API 会尽可能遵循一致的调用范式，减少用户的学习和使用成本。以 read 为例，OpenDAL 提供了以下 API:

op.read(path): 将指定文件全部内容读出
op.reader(path): 创建一个 Reader 用来做流式读取
op.read_with(path).range(1..1024): 使用指定参数来读取文件内容，比如说 range
op.reader_with(path).range(1..1024): 使用指定参数来创建 Reader 做流式读取

不难看出 read 更像是一个语法糖，用来方便用户快速地进行文件读取而不需要考虑 AsyncRead 等各种 trait。而 reader 则给予了用户更多的灵活度，实现了 AsyncSeek，AsyncRead 等社区广泛使用的 trait，允许用户更灵活的读取数据。read_with 和 reader_with 则通过 Future Builder 系列函数，帮助用户以更自然的方式来指定各种参数。

Operator 内部的逻辑看起来会是这样：

它的主要工作是面向用户封装接口：

完成 OpRead 的构建
调用 Accessor 提供的 read 函数
将返回的值包裹为 Reader 并在 Reader 的基础上实现 AsyncSeek，AsyncRead 等接口

Layers

这里有一个隐藏的小秘密是 OpenDAL 会自动为 Service 套上一些 Layer 以实现一些内部逻辑，截止到本文完成的时候，OpenDAL 自动增加的 Layer 包括：

ErrorContextLayer: 为所有的 Operation 返回的 error 注入 context 信息，比如 scheme，path 等
CompleteLayer: 为服务补全必须的能力，比如说为 s3 增加 seek 支持
TypeEraseLayer: 实现类型擦除，将 Accessor 中的关联类型统一擦除，让用户使用时不需要携带泛型参数

这里的 ErrorContextLayer 和 TypeEraseLayer 都比较简单不再赘述，重点聊聊 CompleteLayer，它旨在以零开销的方式为 OpenDAL 返回的 Reader 增加 seek 或者 next 支持，让用户不需要再重复实现。OpenDAL 在早期版本中通过不同的函数调用来返回 Reader 和 SeekableReader，但是用户的实际反馈并不是很好，几乎所有用户都在使用 SeekableReader。因此后续 OpenDAL 在重构中将 seek 支持作为第一优先级加入了内部的 Read trait 中：

pub trait Read: Unpin + Send + Sync {
    /// Read bytes asynchronously.
    fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>>;

    /// Seek asynchronously.
    ///
    /// Returns `Unsupported` error if underlying reader doesn't support seek.
    fn poll_seek(&mut self, cx: &mut Context<'_>, pos: io::SeekFrom) -> Poll<Result<u64>>;

    /// Stream [`Bytes`] from underlying reader.
    ///
    /// Returns `Unsupported` error if underlying reader doesn't support stream.
    ///
    /// This API exists for avoiding bytes copying inside async runtime.
    /// Users can poll bytes from underlying reader and decide when to
    /// read/consume them.
    fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>>;
}

在 OpenDAL 中实现一个服务的读取能力就需要实现这个 trait，这是一个内部接口，不会直接暴露给用户，其中：

poll_read 是最基础的要求，所有服务都必须实现这一接口。
当服务原生支持 seek 时，可以实现 poll_seek，OpenDAL 会进行正确的 dispatch，比如说 local fs；
而当服务原生支持 next，即返回流式的 Bytes 时，可以实现 poll_next，比如说基于 HTTP 的服务，他们底层是一个 TCP Stream，hyper 会将其封装为一个 bytes stream。

通过 Read trait，OpenDAL 确保所有服务都能尽可能地暴露自己的原生支持能力，从而提供对不同服务都能实现高效的读取。

在此 trait 的基础上，OpenDAL 会根据各个服务支持的能力来进行补全：

seek/next 都支持：直接返回
不支持 next: 使用 StreamableReader 进行封装以模拟 next 支持
不支持 seek: 使用 ByRangeSeekableReader 进行封装以模拟 seek 支持
seek/next 均不支持：同时进行两种封装

ByRangeSeekableReader 主要利用了服务支持 range read 的能力，当用户进行 seek 的时候就 drop 当前 reader 并在指定的位置发起新的请求。

OpenDAL 通过 CompleteLayer 暴露出一个统一的 Reader 实现，用户不需要考虑底层服务是否支持 seek，OpenDAL 总是会选择最优的方式来发起请求。

Services

经过 Layers 的补全之后，就到调用 Service 具体实现的地方，这里分别以最常见的两类服务 fs 和 s3 来举例说明数据是如何读取的。

Service fs

tokio::fs::File 实现了 tokio::AsyncRead 和 tokio::AsyncSeek，通过使用 async_compat::Compat，我们将其转化为了 futures::AsyncRead 和 futures::AsyncSeek。在此基础上，我们提供了内置的函数 oio::into_read_from_file 将其转化为实现了 oio::Read 的类型，最终的类型名为：oio::FromFileReader<Compat<tokio::fs::File>>。

oio::into_read_from_file 实现中没有什么特别复杂的地方，read 和 seek 基本上都是在调用传入的 File 类型提供的函数。比较麻烦的地方是关于 seek 和 range 的正确处理：seek 到 range 右侧是允许的行为，此时不会报错，read 也只会返回空，但是 seek 到 range 左侧是非法行为，Reader 必须返回 InvalidInput 以便于上层正确处理。

有趣的历史：当初这块实现的时候有问题，还是在 fuzz 测试中发现的。

Services s3

S3 是一个基于 HTTP 的服务，opendal 提供了大量基于 HTTP 的封装以帮助开发者重用逻辑，只需要构建请求，并返回构造好的 Body 即可。OpenDAL Raw API 封装了一套基于 reqwest 的接口，HTTP GET 接口会返回一个 Response<IncomingAsyncBody>：

/// IncomingAsyncBody carries the content returned by remote servers.
pub struct IncomingAsyncBody {
    /// # TODO
    ///
    /// hyper returns `impl Stream<Item = crate::Result<Bytes>>` but we can't
    /// write the types in stable. So we will box here.
    ///
    /// After [TAIT](https://rust-lang.github.io/rfcs/2515-type_alias_impl_trait.html)
    /// has been stable, we can change `IncomingAsyncBody` into `IncomingAsyncBody<S>`.
    inner: oio::Streamer,
    size: Option<u64>,
    consumed: u64,
    chunk: Option<Bytes>,
}

这个 body 内部包含的 stream 是 reqwest 返回的 bytes stream，opendal 在此基础上实现了 content length 检查和 read 支持。

这里额外提一嘴关于 reqwest/hyper 的小坑：reqwets 和 hyper 并没有检查返回的 content length，所以一个非法的 server 可能会返回与预期的 content length 不符的数据量而非报错，进而导致数据的行为不符合预期。OpenDAL 在这里专门增加了检查，在数据不足时返回 ContentIncomplete，并在数据超出预期时返回 ContentTruncated，避免用户收到非法的数据。

总结

本文自顶向下介绍了 OpenDAL 如何实现数据读取：

Operator 负责对用户暴露易用的接口
Layers 负责对服务的能力进行补全
Services 负责不同服务的具体实现

在整个链路中 OpenDAL 都尽可能遵循零开销的原则，优先使用服务原生提供能力，其次再考虑通过其他的方法进行模拟，最后才会返回不支持的报错。通过这三层的设计，用户不需要了解底层服务的细节，也不需要接入不同服务的 SDK 就可以轻松地调用 op.read(path) 来访问任意存储服务中的数据。

这就是： How OpenDAL read data freely!