apache / apache/datafusion

`ListingTable` should support scanning against versioned object store files

Open
#18,211 3 comments 1 reaction 0 assignees View on GitHub
enhancement
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### Is your feature request related to a problem or challenge?

Currently, when a plan is produced for a `ListingTable` provider it creates file groups based on the file at the time the plan was created.

I have a use case involving an S3 bucket with object versioning enabled, where the files are frequently overwritten. In this scenario, if the file is updated between planning and execution (or during execution) the calculated file ranges in the file group scan will change - resulting in query failures.

It would be ideal if the plan could create file groups that match against specific etags or file versions.

### Describe the solution you'd like

The [`ListingOptions`](https://docs.rs/datafusion/latest/datafusion/datasource/listing/struct.ListingOptions.html) struct for configuring a `ListingTable` could have a new property `object_versioning_type: Option`, where `ObjectVersionType` is:

```rust
enum ObjectVersionType {
ETag,
Version
}
```

This value could be set like `with_object_versioning_type()`. The default value is `None`.

This value is passed through to the `ListingTable` when it is built. Each enum variant corresponds to matching against a specific property from the `object_store` object meta where:

* `ETag`: stores the `e_tag` property, and performs a `get_opts` with a `if_match: Some(e_tag)`
* `Version`: stores the `version` property, and performs a `get_opts` with a `version: Some(version)`

Because this metadata must be encoded into the physical plan with the file groups (where the file scan ranges are stored), `DataSourceExec` must also be updated to support referencing specific versions.

Luckily, [`PartitionedFile`](https://docs.rs/datafusion/latest/datafusion/datasource/listing/struct.PartitionedFile.html) already stores the whole `ObjectMeta` object from `object_store`, so we only need to provide a configuration option to tell the `DataSourceExec` to respect the version/etag.

This results in `with_object_versioning_type()` being passed from `FileScanConfigBuilder` -> `FileScanConfig`.

The property can then be used from the `FileScanConfig` in respective file reading locations, like when constructing a [`ParquetOpener`](https://github.com/apache/datafusion/blob/main/datafusion/datasource-parquet/src/opener.rs#L60)

When the scan then takes place, if the versioning property is supplied it calls the respective `get_opts` object store functions instead of just `get_range` or `get_ranges`. For example, in `ParquetObjectReader::get_bytes`, `get_range` is replaced with `get_range_opts`.

`get_range_opts` doesn't currently exist though, so we must also create that! It should be relatively straightforward though, [as `get_range` just pre-sets the `GetOptions`](https://docs.rs/object_store/latest/src/object_store/lib.rs.html#648):

```rust
async fn get_range(&self, location: &Path, range: Range) -> Result {
let options = GetOptions {
range: Some(range.into()),
..Default::default()
};
self.get_opts(location, options).await?.bytes().await
}
```

While I'm at it, I would also like to update `GetOptions` to support a builder pattern, although this is just a refactor.

### Describe alternatives you've considered

Instead of storing an enum for the object versioning type, we could just support either the etag or version property. I thought it would be easy to support both though, and provide the flexibility for object stores that may or may not support either property.

Instead of creating `get_range_opts`, we could just use `get_opts` which is what `get_range` does anyway. I think a simple wrapper like `get_range_opts` would help with the experience of retrieving these files.

### Additional context

_No response_

Contributor guide

Open the contributing guide

Research direction

Start with ListingOptions and trace object_versioning_type through ListingTable, FileScanConfigBuilder, FileScanConfig, and DataSourceExec. Read ParquetOpener and ParquetObjectReader::get_bytes alongside object_store's get_opts and get_range implementations. Done means planned file groups can preserve ETag or version metadata, scans use the corresponding get options, get_range_opts exists, and GetOptions supports the proposed builder pattern.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
data-engineering
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
38/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.