Merge pull request #17 from dentiny/hjiang/pread

Avoid update file offset in PRead interface

Author: samansmink

extension/httpfs/httpfs.cpp

@@ -526,7 +526,6 @@ void HTTPFileSystem::Read(FileHandle &handle, void *buffer, int64_t nr_bytes, id
 			throw InternalException("Cached file not initialized properly");
 		}
 		memcpy(buffer, hfh.cached_file_handle->GetData() + location, nr_bytes);
-		hfh.file_offset = location + nr_bytes;
 		return;
 	}
 
@@ -537,22 +536,30 @@ void HTTPFileSystem::Read(FileHandle &handle, void *buffer, int64_t nr_bytes, id
 	bool skip_buffer = hfh.flags.DirectIO() || hfh.flags.RequireParallelAccess();
 	if (skip_buffer && to_read > 0) {
 		GetRangeRequest(hfh, hfh.path, {}, location, (char *)buffer, to_read);
+
+		// Update handle status within critical section for parallel access.
+		if (hfh.flags.RequireParallelAccess()) {
+			std::lock_guard<std::mutex> lck(hfh.mu);
+			hfh.buffer_available = 0;
+			hfh.buffer_idx = 0;
+			return;
+		}
+
 		hfh.buffer_available = 0;
 		hfh.buffer_idx = 0;
-		hfh.file_offset = location + nr_bytes;
 		return;
 	}
 
 	if (location >= hfh.buffer_start && location < hfh.buffer_end) {
-		hfh.file_offset = location;
 		hfh.buffer_idx = location - hfh.buffer_start;
 		hfh.buffer_available = (hfh.buffer_end - hfh.buffer_start) - hfh.buffer_idx;
 	} else {
 		// reset buffer
 		hfh.buffer_available = 0;
 		hfh.buffer_idx = 0;
-		hfh.file_offset = location;
 	}
+
+	idx_t start_offset = location;  // Start file offset to read from.
 	while (to_read > 0) {
 		auto buffer_read_len = MinValue<idx_t>(hfh.buffer_available, to_read);
 		if (buffer_read_len > 0) {
@@ -564,36 +571,37 @@ void HTTPFileSystem::Read(FileHandle &handle, void *buffer, int64_t nr_bytes, id
 
 			hfh.buffer_idx += buffer_read_len;
 			hfh.buffer_available -= buffer_read_len;
-			hfh.file_offset += buffer_read_len;
+			start_offset += buffer_read_len;
 		}
 
 		if (to_read > 0 && hfh.buffer_available == 0) {
-			auto new_buffer_available = MinValue<idx_t>(hfh.READ_BUFFER_LEN, hfh.length - hfh.file_offset);
+			auto new_buffer_available = MinValue<idx_t>(hfh.READ_BUFFER_LEN, hfh.length - start_offset);
 
 			// Bypass buffer if we read more than buffer size
 			if (to_read > new_buffer_available) {
 				GetRangeRequest(hfh, hfh.path, {}, location + buffer_offset, (char *)buffer + buffer_offset, to_read);
 				hfh.buffer_available = 0;
 				hfh.buffer_idx = 0;
-				hfh.file_offset += to_read;
+				start_offset += to_read;
 				break;
 			} else {
-				GetRangeRequest(hfh, hfh.path, {}, hfh.file_offset, (char *)hfh.read_buffer.get(),
+				GetRangeRequest(hfh, hfh.path, {}, start_offset, (char *)hfh.read_buffer.get(),
 				                new_buffer_available);
 				hfh.buffer_available = new_buffer_available;
 				hfh.buffer_idx = 0;
-				hfh.buffer_start = hfh.file_offset;
+				hfh.buffer_start = start_offset;
 				hfh.buffer_end = hfh.buffer_start + new_buffer_available;
 			}
 		}
 	}
 }
 
 int64_t HTTPFileSystem::Read(FileHandle &handle, void *buffer, int64_t nr_bytes) {
-	auto &hfh = (HTTPFileHandle &)handle;
+	auto &hfh = handle.Cast<HTTPFileHandle>();
 	idx_t max_read = hfh.length - hfh.file_offset;
 	nr_bytes = MinValue<idx_t>(max_read, nr_bytes);
 	Read(handle, buffer, nr_bytes, hfh.file_offset);
+	hfh.file_offset += nr_bytes;
 	return nr_bytes;
 }
 
@@ -602,7 +610,7 @@ void HTTPFileSystem::Write(FileHandle &handle, void *buffer, int64_t nr_bytes, i
 }
 
 int64_t HTTPFileSystem::Write(FileHandle &handle, void *buffer, int64_t nr_bytes) {
-	auto &hfh = (HTTPFileHandle &)handle;
+	auto &hfh = handle.Cast<HTTPFileHandle>();
 	Write(handle, buffer, nr_bytes, hfh.file_offset);
 	return nr_bytes;
 }

extension/httpfs/include/httpfs.hpp

@@ -8,6 +8,8 @@
 #include "duckdb/main/client_data.hpp"
 #include "http_metadata_cache.hpp"
 
+#include <mutex>
+
 namespace duckdb_httplib_openssl {
 struct Response;
 class Result;
@@ -112,6 +114,9 @@ class HTTPFileHandle : public FileHandle {
 	idx_t buffer_start;
 	idx_t buffer_end;
 
+	// Used when file handle created with parallel access flag specified.
+	std::mutex mu;
+
 	// Read buffer
 	duckdb::unique_ptr<data_t[]> read_buffer;
 	constexpr static idx_t READ_BUFFER_LEN = 1000000;