[CHORE][wal3] encapsulate fragment reading in FragmentConsumer trait

[rescrv]

Description of changes

Move fragment reading methods (read_raw_bytes, read_parquet, read_fragment)
into the FragmentConsumer trait, eliminating the need for LogReader to hold
a Storage reference. This improves encapsulation and simplifies the LogReader
interface.

• Rename FragmentPuller to S3FragmentPuller for naming consistency
• Add read_raw_bytes, read_parquet, read_fragment to FragmentConsumer trait
• Move checksum_parquet utility from reader.rs to interfaces/mod.rs
• Remove storage parameter from LogReader::new and LogReader::open
• Remove unused _writer_name parameters from make_log_reader helpers

Test plan

Pass locally + CI

Migration plan

N/A

Observability plan

N/A

Documentation Changes

N/A

Co-authored-by: AI

[github-actions]

Reviewer Checklist

Please leverage this checklist to ensure your code review is thorough before approving

Testing, Bugs, Errors, Logs, Documentation

• Can you think of any use case in which the code does not behave as intended? Have they been tested?
• Can you think of any inputs or external events that could break the code? Is user input validated and safe? Have they been tested?
• If appropriate, are there adequate property based tests?
• If appropriate, are there adequate unit tests?
• Should any logging, debugging, tracing information be added or removed?
• Are error messages user-friendly?
• Have all documentation changes needed been made?
• Have all non-obvious changes been commented?

System Compatibility

• Are there any potential impacts on other parts of the system or backward compatibility?
• Does this change intersect with any items on our roadmap, and if so, is there a plan for fitting them together?

Quality

• Is this code of a unexpectedly high quality (Readability, Modularity, Intuitiveness)

[propel-code-bot]

It also updates the downstream log-service and s3heap-service consumers to align with the new constructor signatures, adds coverage for relative and absolute parquet offsets, and refreshes the manifest utilities to follow the refactored call pattern.

Affected Areas

• rust/wal3/src/interfaces/mod.rs
• rust/wal3/src/interfaces/s3/fragment_puller.rs
• rust/wal3/src/interfaces/s3/mod.rs
• rust/wal3/src/reader.rs
• rust/log-service/src/lib.rs
• rust/s3heap-service/src/lib.rs
• rust/wal3/src/writer.rs
• rust/wal3/src/interfaces/s3/manifest_manager.rs

This summary was automatically generated by @propel-code-bot

[propel-code-bot]

[Logic] The LogPosition argument is ignored here as well. It should be passed to read_fragment to support relative offsets.

Context for Agents

The `LogPosition` argument is ignored here as well. It should be passed to `read_fragment` to support relative offsets.

File: rust/wal3/src/interfaces/s3/fragment_puller.rs
Line: 44

[rescrv]

No. I changed this elsewhere.

[propel-code-bot]

[Logic] The LogPosition argument is ignored here, effectively passing None to read_parquet. This prevents reading fragments with relative offsets (which require a base position). Consider passing Some(pos) to support both absolute and relative offset files.

Note: This change requires updating read_parquet in mod.rs to accept (Some(_), false) (see my other comment).

Context for Agents

The `LogPosition` argument is ignored here, effectively passing `None` to `read_parquet`. This prevents reading fragments with relative offsets (which require a base position). Consider passing `Some(pos)` to support both absolute and relative offset files.

Note: This change requires updating `read_parquet` in `mod.rs` to accept `(Some(_), false)` (see my other comment).

File: rust/wal3/src/interfaces/s3/fragment_puller.rs
Line: 40

[rescrv]

By design. Pass in None when you know it's absolute, pass in Some when you know it's relative.

[propel-code-bot]

[Logic] The current pattern matching treats (Some(_), false) as an error, which causes failures if a caller provides a position for a file with absolute offsets. To support S3FragmentPuller passing the position (needed for relative offsets), this case should be relaxed to return the records as-is.

Suggested change

	(Some(starting_log_position), true) => {
	for record in records.iter_mut() {
	record.0 = LogPosition::from_offset(
	starting_log_position
	.offset()
	.checked_add(record.0.offset())
	.ok_or(Error::Overflow(format!(
	"log position overflow: {} + {}",
	starting_log_position.offset(),
	record.0.offset()
	)))?,
	);
	}
	Ok((setsum, records, num_bytes))
	}
	(None, false) => Ok((setsum, records, num_bytes)),
	(Some(_), false) => Err(Error::internal(file!(), line!())),
	(None, true) => Err(Error::internal(file!(), line!())),
	}
	(Some(starting_log_position), true) => {
	for record in records.iter_mut() {
	record.0 = LogPosition::from_offset(
	starting_log_position
	.offset()
	.checked_add(record.0.offset())
	.ok_or(Error::Overflow(format!(
	"log position overflow: {} + {}",
	starting_log_position.offset(),
	record.0.offset()
	)))?,
	);
	}
	Ok((setsum, records, num_bytes))
	}
	(_, false) => Ok((setsum, records, num_bytes)),
	(None, true) => Err(Error::internal(file!(), line!())),
	}

Context for Agents

The current pattern matching treats `(Some(_), false)` as an error, which causes failures if a caller provides a position for a file with absolute offsets. To support `S3FragmentPuller` passing the position (needed for relative offsets), this case should be relaxed to return the records as-is.

```suggestion
        (Some(starting_log_position), true) => {
            for record in records.iter_mut() {
                record.0 = LogPosition::from_offset(
                    starting_log_position
                        .offset()
                        .checked_add(record.0.offset())
                        .ok_or(Error::Overflow(format!(
                            "log position overflow: {} + {}",
                            starting_log_position.offset(),
                            record.0.offset()
                        )))?,
                );
            }
            Ok((setsum, records, num_bytes))
        }
        (_, false) => Ok((setsum, records, num_bytes)),
        (None, true) => Err(Error::internal(file!(), line!())),
    }
```

File: rust/wal3/src/interfaces/s3/mod.rs
Line: 283

[sanketkedia]

nit: this type def seems to be in two places - here and in rust/log-service/src/lib.rs. Is there a shared util somewhere between s3heap and wal3?

[propel-code-bot]

[Logic] checksum_parquet already adds starting_log_position to the offsets when processing relative offset files (see offset_base calculation in interfaces/mod.rs).

Adding it again here results in double application of the offset (e.g., base + (base + offset)).

Since checksum_parquet handles the offset translation and overflow checking, this logic should be removed. Just return the records from checksum_parquet.

Context for Agents

`checksum_parquet` already adds `starting_log_position` to the offsets when processing relative offset files (see `offset_base` calculation in `interfaces/mod.rs`).

Adding it again here results in double application of the offset (e.g., `base + (base + offset)`).

Since `checksum_parquet` handles the offset translation and overflow checking, this logic should be removed. Just return the records from `checksum_parquet`.

File: rust/wal3/src/interfaces/s3/mod.rs
Line: 270

[rescrv]

I get this right with unit tests up the stack.