Fallback from O_DIRECT.

dbms/src/Common/ProfileEvents.cpp

@@ -36,8 +36,10 @@
     M(MarkCacheMisses, "") \
     M(CreatedReadBufferOrdinary, "") \
     M(CreatedReadBufferAIO, "") \
+    M(CreatedReadBufferAIOFailed, "") \
     M(CreatedWriteBufferOrdinary, "") \
     M(CreatedWriteBufferAIO, "") \
+    M(CreatedWriteBufferAIOFailed, "") \
     M(DiskReadElapsedMicroseconds, "Total time spent waiting for read syscall. This include reads from page cache.") \
     M(DiskWriteElapsedMicroseconds, "Total time spent waiting for write syscall. This include writes to page cache.") \
     M(NetworkReceiveElapsedMicroseconds, "") \

dbms/src/IO/createReadBufferFromFileBase.cpp

@@ -10,34 +10,38 @@ namespace ProfileEvents
 {
     extern const Event CreatedReadBufferOrdinary;
     extern const Event CreatedReadBufferAIO;
+    extern const Event CreatedReadBufferAIOFailed;
 }
 
 namespace DB
 {
-#if !defined(__linux__)
-namespace ErrorCodes
-{
-    extern const int NOT_IMPLEMENTED;
-}
-#endif
 
 std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(const std::string & filename_, size_t estimated_size,
         size_t aio_threshold, size_t buffer_size_, int flags_, char * existing_memory_, size_t alignment)
 {
-    if ((aio_threshold == 0) || (estimated_size < aio_threshold))
+#if defined(__linux__) || defined(__FreeBSD__)
+    if (aio_threshold && estimated_size >= aio_threshold)
     {
-        ProfileEvents::increment(ProfileEvents::CreatedReadBufferOrdinary);
-        return std::make_unique<ReadBufferFromFile>(filename_, buffer_size_, flags_, existing_memory_, alignment);
+        /// Attempt to open a file with O_DIRECT
+        try
+        {
+            auto res = std::make_unique<ReadBufferAIO>(filename_, buffer_size_, flags_, existing_memory_);
+            ProfileEvents::increment(ProfileEvents::CreatedReadBufferAIO);
+            return res;
+        }
+        catch (const ErrnoException &)
+        {
+            /// Fallback to cached IO if O_DIRECT is not supported.
+            ProfileEvents::increment(ProfileEvents::CreatedReadBufferAIOFailed);
+        }
     }
-    else
-    {
-#if defined(__linux__) || defined(__FreeBSD__)
-        ProfileEvents::increment(ProfileEvents::CreatedReadBufferAIO);
-        return std::make_unique<ReadBufferAIO>(filename_, buffer_size_, flags_, existing_memory_);
 #else
-        throw Exception("AIO is implemented only on Linux and FreeBSD", ErrorCodes::NOT_IMPLEMENTED);
+    (void)aio_threshold;
+    (void)estimated_size;
 #endif
-    }
+
+    ProfileEvents::increment(ProfileEvents::CreatedReadBufferOrdinary);
+    return std::make_unique<ReadBufferFromFile>(filename_, buffer_size_, flags_, existing_memory_, alignment);
 }
 
 }

dbms/src/IO/createWriteBufferFromFileBase.cpp

@@ -10,36 +10,39 @@ namespace ProfileEvents
 {
     extern const Event CreatedWriteBufferOrdinary;
     extern const Event CreatedWriteBufferAIO;
+    extern const Event CreatedWriteBufferAIOFailed;
 }
 
 namespace DB
 {
 
-#if !defined(__linux__)
-namespace ErrorCodes
-{
-    extern const int NOT_IMPLEMENTED;
-}
-#endif
-
 std::unique_ptr<WriteBufferFromFileBase> createWriteBufferFromFileBase(const std::string & filename_, size_t estimated_size,
         size_t aio_threshold, size_t buffer_size_, int flags_, mode_t mode, char * existing_memory_,
         size_t alignment)
 {
-    if ((aio_threshold == 0) || (estimated_size < aio_threshold))
+#if defined(__linux__) || defined(__FreeBSD__)
+    if (aio_threshold && estimated_size >= aio_threshold)
     {
-        ProfileEvents::increment(ProfileEvents::CreatedWriteBufferOrdinary);
-        return std::make_unique<WriteBufferFromFile>(filename_, buffer_size_, flags_, mode, existing_memory_, alignment);
+        /// Attempt to open a file with O_DIRECT
+        try
+        {
+            auto res = std::make_unique<WriteBufferAIO>(filename_, buffer_size_, flags_, mode, existing_memory_);
+            ProfileEvents::increment(ProfileEvents::CreatedWriteBufferAIO);
+            return res;
+        }
+        catch (const ErrnoException &)
+        {
+            /// Fallback to cached IO if O_DIRECT is not supported.
+            ProfileEvents::increment(ProfileEvents::CreatedWriteBufferAIOFailed);
+        }
     }
-    else
-    {
-#if defined(__linux__) || defined(__FreeBSD__)
-        ProfileEvents::increment(ProfileEvents::CreatedWriteBufferAIO);
-        return std::make_unique<WriteBufferAIO>(filename_, buffer_size_, flags_, mode, existing_memory_);
 #else
-        throw Exception("AIO is implemented only on Linux and FreeBSD", ErrorCodes::NOT_IMPLEMENTED);
+    (void)aio_threshold;
+    (void)estimated_size;
 #endif
-    }
+
+    ProfileEvents::increment(ProfileEvents::CreatedWriteBufferOrdinary);
+    return std::make_unique<WriteBufferFromFile>(filename_, buffer_size_, flags_, mode, existing_memory_, alignment);
 }
 
 }