mdreizin commented on issue #39176: URL: https://github.com/apache/beam/issues/39176#issuecomment-6621333020
@goutamadwant yes, that's the approach I'd like: optional, with a buffered fallback, and no breaking change to `OpenRead`. Thanks for offering, please go ahead with the API and filesystem tests. I'll follow up with Parquet reading on top. **Why** Today a Parquet reader on top of `fileio` has to call `file.Read(ctx)` and decode from an in-memory buffer, so every file is downloaded in full and held in memory before the first row comes out. [parquet-go/parquet-go](https://github.com/parquet-go/parquet-go) is built around `io.ReaderAt`, so with this API it can: - **Read only what it needs.** `parquet.OpenFile` reads the footer first, then fetches column chunks and pages by offset. Unprojected columns and skipped row groups are never downloaded. - **Overlap I/O with decoding.** With `ReadModeAsync`, pages are prefetched in the background while rows are decoded. - **Use bounded memory.** Roughly the read buffer plus pages in flight, instead of the whole file. - **Read in parallel.** `ReadAt` is safe for concurrent use, so row groups / column chunks can be fetched as independent range requests. - **Skip by metadata.** Row group stats and page indexes can be checked before fetching data. **Alignment with Java and Python** User-facing names follow Java, since Go's `fileio` already mirrors Java's `FileIO`, adapted to Go conventions the same way `FileMetadata` already does (`Size` / `LastModified` for Java's `sizeBytes` / `lastModifiedMillis`). So Java's `isReadSeekEfficient()` becomes the field `ReadSeekEfficient`, like `Secure` / `HttpOnly` on `http.Cookie`. The filesystem layer uses Go's existing optional-capability pattern (`Copier`, `Renamer`, `LastModifiedGetter`) instead of type-asserting the result of `OpenRead`. Java can return a seekable channel from `open()` because its GCS channel is lazy, but Go's GCS/S3 `OpenRead` start a full-object stream immediately. Making them lazy would change behavior for every existing reader, and any wrapper around the reader would silently drop the capability. | | Java | Python | Go (proposed) | |---|---|---|---| | Seekability flag | `MatchResult.Metadata.isReadSeekEfficient()`, default `false` | none | `fileio.FileMetadata.ReadSeekEfficient`, set by match when the filesystem implements `SeekableOpener` | | Per-file open | `FileIO.ReadableFile.openSeekable()`, fails if not seekable | `ReadableFile.open()` returns a file object with `seekable()` / `seek()` | `fileio.ReadableFile.OpenSeekable(ctx)`, returns `filesystem.ErrNotSeekable` if not seekable | | Filesystem | `FileSystems.open()` returns a `SeekableByteChannel` when supported | `FileSystems.open()` returns a seekable file object | optional `filesystem.SeekableOpener` | | Seek / position | `SeekableByteChannel.position()` | `seek()` / `tell()` | `io.NewSectionReader(r, 0, size)` over the `io.ReaderAt` | **API** ```go // package filesystem // ErrNotSeekable is returned when a file cannot be read with positional reads, // e.g. it is stored with a content encoding that breaks byte offsets. var ErrNotSeekable = errors.New("filesystem: file is not seekable") // ReaderAtCloser is a positional reader over a single file. // // ReadAt must be safe for concurrent use, must not depend on any shared offset, // and follows the io.ReaderAt contract (n < len(p) only with a non-nil error, // io.EOF at end of file). All reads observe the same version of the file. type ReaderAtCloser interface { io.ReaderAt io.Closer } // SeekableOpener is an optional capability, like Copier and Renamer. type SeekableOpener interface { // OpenSeekable opens filename for positional reads. It returns an error // wrapping ErrNotSeekable if this particular file does not support them. OpenSeekable(ctx context.Context, filename string) (ReaderAtCloser, error) } ``` ```go // package fileio type FileMetadata struct { Path string Size int64 LastModified time.Time // ReadSeekEfficient mirrors Java's MatchResult.Metadata.isReadSeekEfficient(). // MatchFiles / MatchAll set it when the filesystem implements SeekableOpener. ReadSeekEfficient bool } // OpenSeekable mirrors Java's FileIO.ReadableFile.openSeekable(). It returns an // error wrapping filesystem.ErrNotSeekable if ReadSeekEfficient is false, the // file is compressed, or the filesystem doesn't implement SeekableOpener. // The returned reader owns the underlying filesystem and closes it on Close. func (f ReadableFile) OpenSeekable(ctx context.Context) (filesystem.ReaderAtCloser, error) ``` Design notes: - **One capability, `OpenSeekable` decides per file.** `ReadSeekEfficient` only says the filesystem supports positional reads, so match needs no extra request per file and there's no second interface that can disagree with `OpenSeekable`. The per-file check happens at open: a GCS object with `Content-Encoding: gzip` is served with decompressive transcoding, so byte offsets would be wrong, and GCS/S3 return `ErrNotSeekable` for such files instead of wrong data. The difference from Java is that Java already reports `false` at match for those objects. Here the caller finds out at open and falls back, which is the same path either way. - **Lifecycle.** `ReadableFile.Open` currently does `defer fs.Close()` right after opening. A `ReaderAt` makes a request on every call, so `OpenSeekable` must hand ownership of the filesystem to the returned reader. - **Consistent reads.** Reads should be pinned to one object version, so a file overwritten mid-read fails instead of mixing bytes: GCS `ObjectHandle.Generation(attrs.Generation)`, S3 `IfMatch` with the ETag from the initial `HeadObject`. The same attrs call supplies the size and the encoding check. - **No `io.Seeker` requirement.** Nothing needs it today, and it would force reopen-on-seek logic into GCS/S3. `io.NewSectionReader` gives Java/Python-style seek/position on top for free. - **Stays thin.** Read coalescing and buffering belong to the caller (parquet-go already has `ReadBufferSize`), not the filesystem. Expected implementations: - local: `*os.File` - memfs: `bytes.Reader` plus a no-op `Close` - GCS: `NewRangeReader(off, len)` per `ReadAt`, generation-pinned; `ErrNotSeekable` when `Content-Encoding: gzip` - S3: `GetObject` with `Range` + `IfMatch` per `ReadAt`; `ErrNotSeekable` for the same encodings Java's `S3FileSystem` excludes Tests that would be great to have: - concurrent `ReadAt` from many goroutines - partial reads and `io.EOF` at and past end of file - `OpenSeekable` returning `ErrNotSeekable` (filesystem without the capability, compressed file, gzip content encoding) - the reader still working after `OpenSeekable` returns, i.e. the filesystem isn't closed early - `ReadSeekEfficient` set by `MatchFiles` / `MatchAll` only for filesystems implementing `SeekableOpener` **How Parquet reading would use it** ```go ra, err := file.OpenSeekable(ctx) switch { case errors.Is(err, filesystem.ErrNotSeekable): data, err := file.Read(ctx) // current behavior: buffer the whole file if err != nil { ... } return decode(parquet.NewReader(bytes.NewReader(data))) case err != nil: return err } defer ra.Close() f, err := parquet.OpenFile(ra, file.Metadata.Size, parquet.FileReadMode(parquet.ReadModeAsync), parquet.ReadBufferSize(4<<20), parquet.SkipBloomFilters(true), ) if err != nil { ... } return decode(parquet.NewReader(f)) ``` For small files a single GET can beat several ranged GETs, so the Parquet side will probably keep a size threshold for the buffered path. I'll benchmark buffered vs sync vs async (full scan and projected columns, a few file sizes) before landing it. Open question (fine as a follow-up): GCS `Size` and `LastModified` each fetch object attrs today. A single attrs-based metadata call would cut match to one request per file, and could also fill `ReadSeekEfficient` per file from the content encoding, which would match Java exactly. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
