Add direct I/O support for WAL writes

db/c.cc

@@ -4299,6 +4299,16 @@ unsigned char rocksdb_options_get_use_direct_io_for_flush_and_compaction(
   return opt->rep.use_direct_io_for_flush_and_compaction;
 }
 
+void rocksdb_options_set_use_direct_io_for_wal(rocksdb_options_t* opt,
+                                               unsigned char v) {
+  opt->rep.use_direct_io_for_wal = v;
+}
+
+unsigned char rocksdb_options_get_use_direct_io_for_wal(
+    rocksdb_options_t* opt) {
+  return opt->rep.use_direct_io_for_wal;
+}
+
 void rocksdb_options_set_allow_mmap_reads(rocksdb_options_t* opt,
                                           unsigned char v) {
   opt->rep.allow_mmap_reads = v;

db/c_test.c

@@ -2650,6 +2650,9 @@ int main(int argc, char** argv) {
     CheckCondition(
         1 == rocksdb_options_get_use_direct_io_for_flush_and_compaction(o));
 
+    rocksdb_options_set_use_direct_io_for_wal(o, 1);
+    CheckCondition(1 == rocksdb_options_get_use_direct_io_for_wal(o));
+
     rocksdb_options_set_is_fd_close_on_exec(o, 1);
     CheckCondition(1 == rocksdb_options_get_is_fd_close_on_exec(o));
 
@@ -2857,6 +2860,7 @@ int main(int argc, char** argv) {
     CheckCondition(1 == rocksdb_options_get_use_direct_reads(copy));
     CheckCondition(
         1 == rocksdb_options_get_use_direct_io_for_flush_and_compaction(copy));
+    CheckCondition(1 == rocksdb_options_get_use_direct_io_for_wal(copy));
     CheckCondition(1 == rocksdb_options_get_is_fd_close_on_exec(copy));
     CheckCondition(18 == rocksdb_options_get_stats_dump_period_sec(copy));
     CheckCondition(5 == rocksdb_options_get_stats_persist_period_sec(copy));
@@ -3127,6 +3131,10 @@ int main(int argc, char** argv) {
     CheckCondition(
         1 == rocksdb_options_get_use_direct_io_for_flush_and_compaction(o));
 
+    rocksdb_options_set_use_direct_io_for_wal(copy, 0);
+    CheckCondition(0 == rocksdb_options_get_use_direct_io_for_wal(copy));
+    CheckCondition(1 == rocksdb_options_get_use_direct_io_for_wal(o));
+
     rocksdb_options_set_is_fd_close_on_exec(copy, 0);
     CheckCondition(0 == rocksdb_options_get_is_fd_close_on_exec(copy));
     CheckCondition(1 == rocksdb_options_get_is_fd_close_on_exec(o));

db/db_impl/db_impl_open.cc

@@ -2344,14 +2344,25 @@ IOStatus DBImpl::CreateWAL(const WriteOptions& write_options,
     lfile->SetWriteLifeTimeHint(opt_file_options.write_hint);
     lfile->SetPreallocationBlockSize(preallocate_block_size);
 
+    // For direct I/O WAL, pre-allocation is mandatory even if user specified 0
+    // Use a default block size of 1MB for direct I/O WAL
+    uint64_t direct_io_preallocation_block_size =
+        immutable_db_options_.wal_direct_io_preallocation_block_size;
+    if (immutable_db_options_.use_direct_io_for_wal &&
+        direct_io_preallocation_block_size == 0) {
+      direct_io_preallocation_block_size =
+          DBOptions{}.wal_direct_io_preallocation_block_size;
+    }
+
     const auto& listeners = immutable_db_options_.listeners;
     FileTypeSet tmp_set = immutable_db_options_.checksum_handoff_file_types;
     std::unique_ptr<WritableFileWriter> file_writer(new WritableFileWriter(
         std::move(lfile), log_fname, opt_file_options,
         immutable_db_options_.clock, io_tracer_, nullptr /* stats */,
         Histograms::HISTOGRAM_ENUM_MAX /* hist_type */, listeners, nullptr,
         tmp_set.Contains(FileType::kWalFile),
-        tmp_set.Contains(FileType::kWalFile)));
+        tmp_set.Contains(FileType::kWalFile),
+        direct_io_preallocation_block_size, immutable_db_options_.env));
     *new_log = new log::Writer(std::move(file_writer), log_file_num,
                                immutable_db_options_.recycle_log_file_num > 0,
                                immutable_db_options_.manual_wal_flush,