JingsongLi commented on code in PR #424:
URL: https://github.com/apache/paimon-rust/pull/424#discussion_r3503795472
##########
crates/paimon/src/arrow/format/mosaic.rs:
##########
@@ -313,39 +318,86 @@ fn micros_to_millis_nanos(micros: i64) -> (i64, i32) {
)
}
-#[derive(Clone)]
-struct MemoryInputFile {
- data: Bytes,
+struct FileReadInputFile {
+ reader: Arc<dyn FileRead>,
+ handle: tokio::runtime::Handle,
}
-impl MemoryInputFile {
- fn new(data: Bytes) -> Self {
- Self { data }
+impl FileReadInputFile {
+ fn new(reader: Box<dyn FileRead>, handle: tokio::runtime::Handle) -> Self {
+ Self {
+ reader: Arc::from(reader),
+ handle,
+ }
}
}
-impl InputFile for MemoryInputFile {
+impl InputFile for FileReadInputFile {
fn read_at(&self, offset: u64, buf: &mut [u8]) -> io::Result<()> {
- let offset = usize::try_from(offset).map_err(|_| {
+ let len = u64::try_from(buf.len()).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidInput,
- "mosaic read offset exceeds usize",
+ "mosaic read length exceeds u64",
)
})?;
- let end = offset.checked_add(buf.len()).ok_or_else(|| {
+ let end = offset.checked_add(len).ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "mosaic read range
overflows")
})?;
- let src = self.data.get(offset..end).ok_or_else(|| {
- io::Error::new(
+ let bytes = block_on_file_read(&self.reader, &self.handle,
offset..end)?;
+ if bytes.len() != buf.len() {
+ return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
- "mosaic read range exceeds file size",
- )
- })?;
- buf.copy_from_slice(src);
+ format!(
+ "mosaic read expected {} bytes, got {}",
+ buf.len(),
+ bytes.len()
+ ),
+ ));
+ }
+ buf.copy_from_slice(&bytes);
Ok(())
}
}
+fn block_on_file_read(
+ reader: &Arc<dyn FileRead>,
+ handle: &tokio::runtime::Handle,
+ range: Range<u64>,
+) -> io::Result<Bytes> {
+ let do_read = || {
+ handle
+ .block_on(reader.read(range.clone()))
+ .map_err(|e| io::Error::other(e.to_string()))
+ };
+
+ match handle.runtime_flavor() {
+ tokio::runtime::RuntimeFlavor::MultiThread => {
+ let in_multi_thread_runtime =
+ tokio::runtime::Handle::try_current().is_ok_and(|handle| {
+ handle.runtime_flavor() ==
tokio::runtime::RuntimeFlavor::MultiThread
+ });
+ if in_multi_thread_runtime {
+ tokio::task::block_in_place(do_read)
+ } else {
+ do_read()
+ }
+ }
+ _ => {
Review Comment:
This current-thread runtime fallback can still hang for real `FileRead`
implementations. `read_batch_stream` is already blocked inside
`MosaicReader::new` / `read_at`, so the current-thread executor is not being
driven; the spawned thread then calls `Handle::block_on` on that same
current-thread runtime, which does not drive IO/timer drivers. The added tests
use an immediate in-memory `FileRead`, so they do not exercise
opendal/network/file futures that need the runtime. Please move the synchronous
Mosaic work onto a blocking thread and dispatch async reads back to the runtime
with a blocking channel/oneshot, or otherwise fail/require a multi-thread
runtime instead of trying to support current-thread this way.
--
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]
