ioctx: add ReaderAtSeekerAdapter for random access support

This commit adds support for adapting context-aware readers that implement
ReaderAtCtx and SeekerCtx into the standard io.ReaderAt and io.Seeker
interfaces. This is needed for integrating cloud storage readers with
libraries that require random access (e.g., Apache Parquet).

The adapter:
- Captures a context at construction time and uses it for all operations
- Validates that the underlying reader actually supports seeking
- Provides io.Reader, io.ReaderAt, io.Seeker, and io.Closer interfaces
- Enables efficient random access to cloud-stored files without buffering

Release note: None

Epic: CRDB-23802

Author: sravotto

pkg/BUILD.bazel

@@ -775,6 +775,7 @@ ALL_TESTS = [
     "//pkg/util/interval/generic:generic_test",
     "//pkg/util/interval:interval_test",
     "//pkg/util/intsets:intsets_test",
+    "//pkg/util/ioctx:ioctx_test",
     "//pkg/util/ipaddr:ipaddr_test",
     "//pkg/util/iterutil:iterutil_test",
     "//pkg/util/json/tokenizer:tokenizer_test",
@@ -2700,6 +2701,7 @@ GO_TARGETS = [
     "//pkg/util/intsets:intsets",
     "//pkg/util/intsets:intsets_test",
     "//pkg/util/ioctx:ioctx",
+    "//pkg/util/ioctx:ioctx_test",
     "//pkg/util/ipaddr:ipaddr",
     "//pkg/util/ipaddr:ipaddr_test",
     "//pkg/util/iterutil:iterutil",

pkg/util/ioctx/BUILD.bazel

@@ -1,4 +1,4 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "ioctx",
@@ -7,3 +7,14 @@ go_library(
     visibility = ["//visibility:public"],
     deps = ["@com_github_cockroachdb_errors//:errors"],
 )
+
+go_test(
+    name = "ioctx_test",
+    size = "small",
+    srcs = ["reader_test.go"],
+    embed = [":ioctx"],
+    deps = [
+        "@com_github_cockroachdb_errors//:errors",
+        "@com_github_stretchr_testify//require",
+    ],
+)

pkg/util/ioctx/reader.go

@@ -138,3 +138,93 @@ type nopCloser struct {
 
 // Close is part of the ReadClosedCtx interface.
 func (nopCloser) Close(ctx context.Context) error { return nil }
+
+// ReaderAtCtx is like io.ReaderAt, but the ReadAt() method takes a context.
+type ReaderAtCtx interface {
+	ReadAt(ctx context.Context, p []byte, off int64) (n int, err error)
+}
+
+// SeekerCtx is like io.Seeker, but the Seek() method takes a context.
+type SeekerCtx interface {
+	Seek(ctx context.Context, offset int64, whence int) (n int64, err error)
+}
+
+// ReaderAtSeekerCloser combines io.Reader, io.ReaderAt, io.Seeker, and io.Closer.
+// This is useful for formats like Parquet that require random access to files.
+type ReaderAtSeekerCloser interface {
+	io.Reader
+	io.ReaderAt
+	io.Seeker
+	io.Closer
+}
+
+// ReaderAtSeekerAdapter adapts a ReadCloserCtx that implements ReaderAtCtx
+// and SeekerCtx into a ReaderAtSeekerCloser. The context is captured at construction
+// time and used for all operations (Read, ReadAt, Seek, and Close).
+//
+// Returns an error if the underlying reader does not implement both ReaderAtCtx and
+// SeekerCtx, or if a test seek operation fails (indicating the reader claims to support
+// seeking but doesn't actually work).
+//
+// This is useful for integrating context-aware cloud storage readers with libraries
+// that require the standard io.ReaderAt and io.Seeker interfaces (e.g., Parquet).
+//
+// IMPORTANT: The returned adapter's lifetime is tied to the provided context. If the
+// context is canceled, all subsequent Read/ReadAt/Seek operations will fail. The caller
+// must ensure the context remains valid for the entire duration the adapter is in use.
+func ReaderAtSeekerAdapter(
+	ctx context.Context, readCloser ReadCloserCtx,
+) (ReaderAtSeekerCloser, error) {
+	readerAt, okReaderAt := readCloser.(ReaderAtCtx)
+	if !okReaderAt {
+		return nil, errors.Newf("reader does not implement ioctx.ReaderAtCtx (type %T)", readCloser)
+	}
+	seeker, okSeeker := readCloser.(SeekerCtx)
+	if !okSeeker {
+		return nil, errors.Newf("reader does not implement ioctx.SeekerCtx (type %T)", readCloser)
+	}
+
+	// Test if seeking actually works by attempting a no-op seek
+	// This catches cases where the reader implements the interface but fails at runtime
+	if _, err := seeker.Seek(ctx, 0, io.SeekCurrent); err != nil {
+		return nil, errors.Wrapf(err, "reader type %T claims to support seeking but test seek failed", readCloser)
+	}
+
+	return &readerAtSeekerAdapter{
+		ctx:        ctx,
+		readCloser: readCloser,
+		readerAt:   readerAt,
+		seeker:     seeker,
+	}, nil
+}
+
+type readerAtSeekerAdapter struct {
+	ctx        context.Context
+	readCloser ReadCloserCtx
+	readerAt   ReaderAtCtx
+	seeker     SeekerCtx
+}
+
+var _ ReaderAtSeekerCloser = &readerAtSeekerAdapter{}
+
+// Read implements io.Reader using the captured context.
+func (r *readerAtSeekerAdapter) Read(p []byte) (int, error) {
+	return r.readCloser.Read(r.ctx, p)
+}
+
+// ReadAt implements io.ReaderAt by delegating to the underlying ReaderAtCtx
+// using the captured context.
+func (r *readerAtSeekerAdapter) ReadAt(p []byte, off int64) (int, error) {
+	return r.readerAt.ReadAt(r.ctx, p, off)
+}
+
+// Seek implements io.Seeker by delegating to the underlying SeekerCtx
+// using the captured context.
+func (r *readerAtSeekerAdapter) Seek(offset int64, whence int) (int64, error) {
+	return r.seeker.Seek(r.ctx, offset, whence)
+}
+
+// Close implements io.Closer using the captured context.
+func (r *readerAtSeekerAdapter) Close() error {
+	return r.readCloser.Close(r.ctx)
+}

pkg/util/ioctx/reader_test.go

@@ -0,0 +1,262 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package ioctx
+
+import (
+	"context"
+	"io"
+	"testing"
+
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+// mockReadCloserCtx is a mock that implements ReadCloserCtx, ReaderAtCtx, and SeekerCtx
+type mockReadCloserCtx struct {
+	data      []byte
+	pos       int64
+	closed    bool
+	seekCalls int
+	readCalls int
+}
+
+func newMockReadCloserCtx(data []byte) *mockReadCloserCtx {
+	return &mockReadCloserCtx{
+		data: data,
+	}
+}
+
+func (m *mockReadCloserCtx) Read(ctx context.Context, p []byte) (int, error) {
+	m.readCalls++
+	if m.pos >= int64(len(m.data)) {
+		return 0, io.EOF
+	}
+	n := copy(p, m.data[m.pos:])
+	m.pos += int64(n)
+	return n, nil
+}
+
+func (m *mockReadCloserCtx) Close(ctx context.Context) error {
+	m.closed = true
+	return nil
+}
+
+func (m *mockReadCloserCtx) ReadAt(ctx context.Context, p []byte, off int64) (int, error) {
+	if off >= int64(len(m.data)) {
+		return 0, io.EOF
+	}
+	n := copy(p, m.data[off:])
+	if n < len(p) {
+		return n, io.EOF
+	}
+	return n, nil
+}
+
+func (m *mockReadCloserCtx) Seek(ctx context.Context, offset int64, whence int) (int64, error) {
+	m.seekCalls++
+	var newPos int64
+	switch whence {
+	case io.SeekStart:
+		newPos = offset
+	case io.SeekCurrent:
+		newPos = m.pos + offset
+	case io.SeekEnd:
+		newPos = int64(len(m.data)) + offset
+	default:
+		return 0, errors.Newf("invalid whence: %d", whence)
+	}
+
+	if newPos < 0 {
+		return 0, errors.New("negative position")
+	}
+
+	m.pos = newPos
+	return newPos, nil
+}
+
+// mockReadCloserCtxNoSeek implements ReadCloserCtx but NOT SeekerCtx
+type mockReadCloserCtxNoSeek struct {
+	data []byte
+	pos  int64
+}
+
+func (m *mockReadCloserCtxNoSeek) Read(ctx context.Context, p []byte) (int, error) {
+	if m.pos >= int64(len(m.data)) {
+		return 0, io.EOF
+	}
+	n := copy(p, m.data[m.pos:])
+	m.pos += int64(n)
+	return n, nil
+}
+
+func (m *mockReadCloserCtxNoSeek) Close(ctx context.Context) error {
+	return nil
+}
+
+func (m *mockReadCloserCtxNoSeek) ReadAt(ctx context.Context, p []byte, off int64) (int, error) {
+	if off >= int64(len(m.data)) {
+		return 0, io.EOF
+	}
+	n := copy(p, m.data[off:])
+	if n < len(p) {
+		return n, io.EOF
+	}
+	return n, nil
+}
+
+// mockReadCloserCtxBrokenSeek implements SeekerCtx but Seek always fails
+type mockReadCloserCtxBrokenSeek struct {
+	data []byte
+	pos  int64
+}
+
+func (m *mockReadCloserCtxBrokenSeek) Read(ctx context.Context, p []byte) (int, error) {
+	if m.pos >= int64(len(m.data)) {
+		return 0, io.EOF
+	}
+	n := copy(p, m.data[m.pos:])
+	m.pos += int64(n)
+	return n, nil
+}
+
+func (m *mockReadCloserCtxBrokenSeek) Close(ctx context.Context) error {
+	return nil
+}
+
+func (m *mockReadCloserCtxBrokenSeek) ReadAt(
+	ctx context.Context, p []byte, off int64,
+) (int, error) {
+	if off >= int64(len(m.data)) {
+		return 0, io.EOF
+	}
+	n := copy(p, m.data[off:])
+	if n < len(p) {
+		return n, io.EOF
+	}
+	return n, nil
+}
+
+func (m *mockReadCloserCtxBrokenSeek) Seek(
+	ctx context.Context, offset int64, whence int,
+) (int64, error) {
+	return 0, errors.New("seek operation failed")
+}
+
+func TestReaderAtSeekerAdapterBasicOperations(t *testing.T) {
+	ctx := context.Background()
+	testData := []byte("Hello, World! This is test data for seeking.")
+	mock := newMockReadCloserCtx(testData)
+
+	adapter, err := ReaderAtSeekerAdapter(ctx, mock)
+	require.NoError(t, err)
+	require.NotNil(t, adapter)
+
+	// Test that the initial seek test was called
+	require.Equal(t, 1, mock.seekCalls, "adapter should test seek during construction")
+
+	// Test Read
+	buf := make([]byte, 5)
+	n, err := adapter.Read(buf)
+	require.NoError(t, err)
+	require.Equal(t, 5, n)
+	require.Equal(t, []byte("Hello"), buf)
+
+	// Test ReadAt (should read at specific offset)
+	buf = make([]byte, 5)
+	n, err = adapter.ReadAt(buf, 7)
+	require.NoError(t, err)
+	require.Equal(t, 5, n)
+	require.Equal(t, []byte("World"), buf)
+
+	// Test Seek to start
+	pos, err := adapter.Seek(0, io.SeekStart)
+	require.NoError(t, err)
+	require.Equal(t, int64(0), pos)
+
+	// Test Seek current
+	pos, err = adapter.Seek(5, io.SeekCurrent)
+	require.NoError(t, err)
+	require.Equal(t, int64(5), pos)
+
+	// Test Seek from end
+	pos, err = adapter.Seek(-5, io.SeekEnd)
+	require.NoError(t, err)
+	require.Equal(t, int64(len(testData)-5), pos)
+
+	// Test Close
+	err = adapter.Close()
+	require.NoError(t, err)
+	require.True(t, mock.closed)
+}
+
+func TestReaderAtSeekerAdapterMissingReaderAtCtx(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a reader that doesn't implement ReaderAtCtx
+	reader := ReadCloserAdapter(io.NopCloser(nil))
+
+	adapter, err := ReaderAtSeekerAdapter(ctx, reader)
+	require.Error(t, err)
+	require.Nil(t, adapter)
+	require.Contains(t, err.Error(), "does not implement ioctx.ReaderAtCtx")
+}
+
+func TestReaderAtSeekerAdapterMissingSeekerCtx(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a reader that implements ReaderAtCtx but not SeekerCtx
+	reader := &mockReadCloserCtxNoSeek{
+		data: []byte("test data"),
+	}
+
+	adapter, err := ReaderAtSeekerAdapter(ctx, reader)
+	require.Error(t, err)
+	require.Nil(t, adapter)
+	require.Contains(t, err.Error(), "does not implement ioctx.SeekerCtx")
+}
+
+func TestReaderAtSeekerAdapterBrokenSeek(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a reader that implements SeekerCtx but seek fails
+	reader := &mockReadCloserCtxBrokenSeek{
+		data: []byte("test data"),
+	}
+
+	adapter, err := ReaderAtSeekerAdapter(ctx, reader)
+	require.Error(t, err)
+	require.Nil(t, adapter)
+	require.Contains(t, err.Error(), "test seek failed")
+}
+
+func TestReaderAtSeekerAdapterReadAtEOF(t *testing.T) {
+	ctx := context.Background()
+	testData := []byte("short")
+	mock := newMockReadCloserCtx(testData)
+
+	adapter, err := ReaderAtSeekerAdapter(ctx, mock)
+	require.NoError(t, err)
+
+	// Try to read past EOF
+	buf := make([]byte, 10)
+	n, err := adapter.ReadAt(buf, int64(len(testData)))
+	require.Equal(t, io.EOF, err)
+	require.Equal(t, 0, n)
+}
+
+func TestReaderAtSeekerAdapterInvalidSeek(t *testing.T) {
+	ctx := context.Background()
+	testData := []byte("test data")
+	mock := newMockReadCloserCtx(testData)
+
+	adapter, err := ReaderAtSeekerAdapter(ctx, mock)
+	require.NoError(t, err)
+
+	// Negative position
+	_, err = adapter.Seek(-10, io.SeekStart)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "negative position")
+}