linliu-code commented on code in PR #667:
URL: https://github.com/apache/hudi-rs/pull/667#discussion_r3779922585
##########
crates/core/src/file_group/log_file/reader.rs:
##########
@@ -321,6 +491,64 @@ mod tests {
LogFileReader::new(hudi_configs, storage, file_name).await
}
+ /// A block whose recorded length disagrees with its trailing reverse
+ /// pointer is corrupt. Both sizes are written by the same writer, so a
+ /// mismatch means the span cannot be trusted.
+ #[tokio::test]
+ async fn test_corrupt_block_detected_when_trailing_length_disagrees() ->
Result<()> {
+ let (dir, file_name) = get_valid_log_avro_data();
+ let mut reader = create_log_file_reader(&dir, &file_name).await?;
+
+ let magic_pos = 0;
+ let real_length = {
+ reader.read_magic()?;
+ reader.read_block_length()?
+ };
+ assert!(
+ !reader.is_block_corrupted(magic_pos, real_length)?,
+ "a well-formed block must not be reported corrupt"
+ );
+ assert!(
+ reader.is_block_corrupted(magic_pos, real_length + 1)?,
+ "a length disagreeing with the trailing pointer must be reported
corrupt"
+ );
+ Ok(())
+ }
+
+ /// A length pointing past the end of the file is corrupt, and must be
+ /// decided by arithmetic rather than by trying to read there.
+ #[tokio::test]
+ async fn test_corrupt_block_detected_when_length_runs_past_eof() ->
Result<()> {
+ let (dir, file_name) = get_valid_log_avro_data();
+ let mut reader = create_log_file_reader(&dir, &file_name).await?;
+
+ assert!(
+ reader.is_block_corrupted(0, u64::MAX)?,
+ "overflow is corrupt"
+ );
+ assert!(
+ reader.is_block_corrupted(0, 1 << 40)?,
+ "a length past EOF is corrupt"
+ );
+ Ok(())
+ }
+
+ /// Recovery lands on the next magic marker, or the end of the file when
+ /// there is none — so one bad block costs its own span, not the rest.
+ #[tokio::test]
Review Comment:
Fixed, taking your second suggestion: renamed to
`test_scan_for_next_block_offset_stays_within_file_bounds`. The doc now says
outright that the fixture is a valid multi-block file so the scan lands on the
*next* block's magic rather than EOF, and that what's pinned is that either
answer stays inside the file — the old name claimed something this fixture
cannot produce.
##########
crates/core/src/file_group/log_file/reader.rs:
##########
@@ -99,32 +126,164 @@ impl<R: Read + Seek> LogFileReader<R> {
Ok(u64::from_be_bytes(size_buf))
}
- fn create_corrupted_block_if_needed(
- &mut self,
- _curent_pos: u64,
- _block_length: Option<u64>,
- ) -> Option<LogBlock> {
- // TODO: support creating corrupted block
- None
+ /// Window used when scanning for the next MAGIC after a corrupt block.
+ const BLOCK_SCAN_READ_BUFFER_SIZE: usize = 1024 * 1024;
+
+ /// Total length of the stream, restoring the original position.
+ fn stream_len(&mut self) -> Result<u64> {
+ let cur = self
+ .reader
+ .stream_position()
+ .map_err(CoreError::ReadLogFileError)?;
+ let end = self
+ .reader
+ .seek(SeekFrom::End(0))
+ .map_err(CoreError::ReadLogFileError)?;
+ self.reader
+ .seek(SeekFrom::Start(cur))
+ .map_err(CoreError::ReadLogFileError)?;
+ Ok(end)
}
- fn read_block_length_or_corrupted_block(
- &mut self,
- start_pos: u64,
- ) -> Result<(u64, Option<LogBlock>)> {
- match self.read_block_length() {
- Ok(length) => {
- if let Some(block) =
self.create_corrupted_block_if_needed(start_pos, Some(length))
- {
- Ok((0, Some(block)))
- } else {
- Ok((length, None))
- }
+ /// Whether the next bytes are a MAGIC marker, treating end-of-file as one.
+ fn next_is_magic_or_eof(&mut self) -> Result<bool> {
+ let mut magic = [0u8; 6];
+ match self.reader.read_exact(&mut magic) {
+ Ok(_) => Ok(magic == MAGIC),
+ Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => Ok(true),
+ Err(e) => Err(CoreError::ReadLogFileError(e)),
+ }
+ }
+
+ /// Whether the block starting at `magic_pos` is corrupt, given the
+ /// `block_length` already read from just after the magic.
+ ///
+ /// A well-formed block records its total size twice — once in the header
+ /// length field and once in a trailing reverse pointer — and is followed
by
+ /// either another block or the end of the file. Three checks, and every
+ /// offset is computed with checked arithmetic so a garbage length reports
+ /// corruption rather than panicking or allocating against it:
+ ///
+ /// 1. the trailing pointer has to lie inside the file
+ /// 2. the size it records has to agree with the header
+ /// 3. what follows the block has to be a MAGIC marker or the end
+ ///
+ /// The reader is left just after the length field either way, which is
+ /// where the caller expects to continue from.
+ fn is_block_corrupted(&mut self, magic_pos: u64, block_length: u64) ->
Result<bool> {
+ let after_length = magic_pos
+ .checked_add(MAGIC.len() as u64)
+ .and_then(|v| v.checked_add(8));
+ let Some(after_length) = after_length else {
+ return Ok(true);
+ };
+ // The trailing long sits 8 bytes before the block ends.
+ let trailing_pos = after_length
+ .checked_add(block_length)
+ .and_then(|v| v.checked_sub(8));
+ let Some(trailing_pos) = trailing_pos else {
+ return Ok(true);
+ };
+
+ let stream_len = self.stream_len()?;
+
+ if trailing_pos
+ .checked_add(8)
+ .map(|e| e > stream_len)
+ .unwrap_or(true)
+ {
+ self.reader
+ .seek(SeekFrom::Start(after_length))
+ .map_err(CoreError::ReadLogFileError)?;
+ return Ok(true);
+ }
+
+ self.reader
+ .seek(SeekFrom::Start(trailing_pos))
+ .map_err(CoreError::ReadLogFileError)?;
+ let mut buf = [0u8; 8];
+ self.reader
+ .read_exact(&mut buf)
+ .map_err(CoreError::ReadLogFileError)?;
+ let trailing = u64::from_be_bytes(buf);
+
+ // The trailing value counts the magic; the header length does not.
+ let corrupt = match trailing.checked_sub(MAGIC.len() as u64) {
+ Some(size_from_footer) => size_from_footer != block_length,
+ None => true,
+ };
+
+ let block_end = after_length
+ .checked_add(block_length)
+ .ok_or_else(|| CoreError::LogFormatError("Block length
overflow".to_string()))?;
+
+ let result = if corrupt {
+ true
+ } else {
+ self.reader
+ .seek(SeekFrom::Start(block_end))
+ .map_err(CoreError::ReadLogFileError)?;
+ !self.next_is_magic_or_eof()?
+ };
+
+ self.reader
Review Comment:
Fixed. The doc now reads "Offset of the next MAGIC strictly after the magic
at `from_pos`", and says why: the scan starts `MAGIC.len()` bytes in, so a
caller recovering from a bad block at `from_pos` would otherwise be handed it
straight back.
--
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]
