Removed synchronous readahead from arrow-10183 PR

cpp/src/arrow/CMakeLists.txt

@@ -189,7 +189,6 @@ set(ARROW_SRCS
     util/future.cc
     util/int_util.cc
     util/io_util.cc
-    util/iterator.cc
     util/logging.cc
     util/key_value_metadata.cc
     util/memory.cc

cpp/src/arrow/csv/column_decoder.cc

@@ -84,7 +84,7 @@ class ConcreteColumnDecoder : public ColumnDecoder {
     auto chunk_index = next_chunk_++;
     WaitForChunkUnlocked(chunk_index);
     // Move Future to avoid keeping chunk alive
-    return std::move(chunks_[chunk_index]).result();
+    return chunks_[chunk_index].MoveResult();
   }
 
  protected:

cpp/src/arrow/csv/reader.cc

@@ -918,11 +918,11 @@ class AsyncThreadedTableReader
     ARROW_ASSIGN_OR_RAISE(auto istream_it,
                           io::MakeInputStreamIterator(input_, read_options_.block_size));
 
-    ARROW_ASSIGN_OR_RAISE(auto bg_it,
-                          MakeBackgroundGenerator(std::move(istream_it), thread_pool_));
+    ARROW_ASSIGN_OR_RAISE(auto bg_it, MakeBackgroundGenerator(std::move(istream_it)));
+    bg_it = TransferGenerator(std::move(bg_it), thread_pool_);
 
     int32_t block_queue_size = thread_pool_->GetCapacity();
-    auto rh_it = AddReadahead(bg_it, block_queue_size);
+    auto rh_it = AddReadahead(std::move(bg_it), block_queue_size);
     buffer_generator_ = CSVBufferIterator::MakeAsync(std::move(rh_it));
     return Status::OK();
   }
@@ -957,7 +957,7 @@ class AsyncThreadedTableReader
         return Status::OK();
       };
 
-      return VisitAsyncGenerator(block_generator, block_visitor)
+      return VisitAsyncGenerator(std::move(block_generator), block_visitor)
           .Then([self](...) -> Future<> {
             // By this point we've added all top level tasks so it is safe to call
             // FinishAsync

cpp/src/arrow/csv/reader_test.cc

@@ -62,6 +62,27 @@ void StressTableReader(TableReaderFactory reader_factory) {
   }
 }
 
+void StressInvalidTableReader(TableReaderFactory reader_factory) {
+  const int NTASKS = 100;
+  const int NROWS = 1000;
+  ASSERT_OK_AND_ASSIGN(auto table_buffer, MakeSampleCsvBuffer(NROWS, false));
+
+  std::vector<Future<std::shared_ptr<Table>>> task_futures(NTASKS);
+  for (int i = 0; i < NTASKS; i++) {
+    auto input = std::make_shared<io::BufferReader>(table_buffer);
+    ASSERT_OK_AND_ASSIGN(auto reader, reader_factory(input));
+    task_futures[i] = reader->ReadAsync();
+  }
+  auto combined_future = All(task_futures);
+  combined_future.Wait();
+
+  ASSERT_OK_AND_ASSIGN(std::vector<Result<std::shared_ptr<Table>>> results,
+                       combined_future.result());
+  for (auto&& result : results) {
+    ASSERT_RAISES(Invalid, result);
+  }
+}
+
 void TestNestedParallelism(std::shared_ptr<internal::ThreadPool> thread_pool,
                            TableReaderFactory reader_factory) {
   const int NROWS = 1000;
@@ -82,82 +103,77 @@ void TestNestedParallelism(std::shared_ptr<internal::ThreadPool> thread_pool,
   ASSERT_EQ(table->num_rows(), NROWS);
 }  // namespace csv
 
-TEST(SerialReaderTests, Stress) {
-  auto task_factory = [](std::shared_ptr<io::InputStream> input_stream) {
+TableReaderFactory MakeSerialFactory() {
+  return [](std::shared_ptr<io::InputStream> input_stream) {
+    auto read_options = ReadOptions::Defaults();
+    read_options.block_size = 1 << 10;
+    read_options.use_threads = false;
     return TableReader::Make(default_memory_pool(), io::AsyncContext(), input_stream,
-                             ReadOptions::Defaults(), ParseOptions::Defaults(),
+                             read_options, ParseOptions::Defaults(),
                              ConvertOptions::Defaults());
   };
-  StressTableReader(task_factory);
 }
 
+TEST(SerialReaderTests, Stress) { StressTableReader(MakeSerialFactory()); }
+TEST(SerialReaderTests, StressInvalid) { StressInvalidTableReader(MakeSerialFactory()); }
 TEST(SerialReaderTests, NestedParallelism) {
   ASSERT_OK_AND_ASSIGN(auto thread_pool, internal::ThreadPool::Make(1));
-  auto task_factory = [](std::shared_ptr<io::InputStream> input_stream) {
-    return TableReader::Make(default_memory_pool(), io::AsyncContext(), input_stream,
-                             ReadOptions::Defaults(), ParseOptions::Defaults(),
-                             ConvertOptions::Defaults());
-  };
-  TestNestedParallelism(thread_pool, task_factory);
+  TestNestedParallelism(thread_pool, MakeSerialFactory());
 }
 
-TEST(ThreadedReaderTests, Stress) {
-  ASSERT_OK_AND_ASSIGN(auto thread_pool, internal::ThreadPool::Make(1));
-  auto task_factory = [&thread_pool](std::shared_ptr<io::InputStream> input_stream)
-      -> Result<std::shared_ptr<TableReader>> {
+Result<TableReaderFactory> CreateThreadedFactory() {
+  ARROW_ASSIGN_OR_RAISE(auto thread_pool, internal::ThreadPool::Make(1));
+  return [thread_pool](std::shared_ptr<io::InputStream> input_stream)
+             -> Result<std::shared_ptr<TableReader>> {
     ReadOptions read_options = ReadOptions::Defaults();
     read_options.use_threads = true;
+    read_options.block_size = 1 << 10;
     read_options.legacy_blocking_reads = true;
     auto table_reader = TableReader::Make(
         default_memory_pool(), io::AsyncContext(thread_pool.get()), input_stream,
         read_options, ParseOptions::Defaults(), ConvertOptions::Defaults());
     return table_reader;
   };
-  StressTableReader(task_factory);
 }
 
-// Simulates deadlock that exists with ThreadedReaderTests
-// TEST(ThreadedReaderTests, NestedParallelism) {
-//   ASSERT_OK_AND_ASSIGN(auto thread_pool, internal::ThreadPool::Make(1));
-//   auto task_factory = [&thread_pool](std::shared_ptr<io::InputStream> input_stream)
-//       -> Result<std::shared_ptr<TableReader>> {
-//     ReadOptions read_options = ReadOptions::Defaults();
-//     read_options.use_threads = true;
-//     read_options.legacy_blocking_reads = true;
-//     auto table_reader = TableReader::Make(
-//         default_memory_pool(), io::AsyncContext(thread_pool.get()), input_stream,
-//         read_options, ParseOptions::Defaults(), ConvertOptions::Defaults());
-//     return table_reader;
-//   };
-//   TestNestedParallelism(thread_pool, task_factory);
-// }
+TEST(ThreadedReaderTests, Stress) {
+  ASSERT_OK_AND_ASSIGN(auto factory, CreateThreadedFactory());
+  StressTableReader(factory);
+}
+TEST(ThreadedReaderTests, StressInvalid) {
+  ASSERT_OK_AND_ASSIGN(auto factory, CreateThreadedFactory());
+  StressInvalidTableReader(factory);
+}
 
-TEST(AsyncReaderTests, Stress) {
-  ASSERT_OK_AND_ASSIGN(auto thread_pool, internal::ThreadPool::Make(1));
-  auto task_factory = [&thread_pool](std::shared_ptr<io::InputStream> input_stream)
-      -> Result<std::shared_ptr<TableReader>> {
+Result<TableReaderFactory> MakeAsyncFactory(
+    std::shared_ptr<internal::ThreadPool> thread_pool = nullptr) {
+  if (!thread_pool) {
+    ARROW_ASSIGN_OR_RAISE(thread_pool, internal::ThreadPool::Make(1));
+  }
+  return [thread_pool](std::shared_ptr<io::InputStream> input_stream)
+             -> Result<std::shared_ptr<TableReader>> {
     ReadOptions read_options = ReadOptions::Defaults();
     read_options.use_threads = true;
+    read_options.block_size = 1 << 10;
     auto table_reader = TableReader::Make(
         default_memory_pool(), io::AsyncContext(thread_pool.get()), input_stream,
         read_options, ParseOptions::Defaults(), ConvertOptions::Defaults());
     return table_reader;
   };
-  StressTableReader(task_factory);
 }
 
+TEST(AsyncReaderTests, Stress) {
+  ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory());
+  StressTableReader(table_factory);
+}
+TEST(AsyncReaderTests, StressInvalid) {
+  ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory());
+  StressInvalidTableReader(table_factory);
+}
 TEST(AsyncReaderTests, NestedParallelism) {
   ASSERT_OK_AND_ASSIGN(auto thread_pool, internal::ThreadPool::Make(1));
-  auto task_factory = [&thread_pool](std::shared_ptr<io::InputStream> input_stream)
-      -> Result<std::shared_ptr<TableReader>> {
-    ReadOptions read_options = ReadOptions::Defaults();
-    read_options.use_threads = true;
-    auto table_reader = TableReader::Make(
-        default_memory_pool(), io::AsyncContext(thread_pool.get()), input_stream,
-        read_options, ParseOptions::Defaults(), ConvertOptions::Defaults());
-    return table_reader;
-  };
-  TestNestedParallelism(thread_pool, task_factory);
+  ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory(thread_pool));
+  TestNestedParallelism(thread_pool, table_factory);
 }
 
 }  // namespace csv

cpp/src/arrow/csv/test_common.cc

@@ -91,14 +91,23 @@ static void WriteRow(std::ostream& writer, size_t row_index) {
   writer << GetCell(strptime_rows, row_index);
   writer << std::endl;
 }
+
+static void WriteInvalidRow(std::ostream& writer, size_t row_index) {
+  writer << "\"" << std::endl << "\"";
+  writer << std::endl;
+}
 }  // namespace
 
-Result<std::shared_ptr<Buffer>> MakeSampleCsvBuffer(size_t num_rows) {
+Result<std::shared_ptr<Buffer>> MakeSampleCsvBuffer(size_t num_rows, bool valid) {
   std::stringstream writer;
 
   WriteHeader(writer);
   for (size_t i = 0; i < num_rows; ++i) {
-    WriteRow(writer, i);
+    if (i == num_rows / 2 && !valid) {
+      WriteInvalidRow(writer, i);
+    } else {
+      WriteRow(writer, i);
+    }
   }
 
   auto table_str = writer.str();

cpp/src/arrow/csv/test_common.h

@@ -47,7 +47,7 @@ ARROW_TESTING_EXPORT
 void MakeColumnParser(std::vector<std::string> items, std::shared_ptr<BlockParser>* out);
 
 ARROW_TESTING_EXPORT
-Result<std::shared_ptr<Buffer>> MakeSampleCsvBuffer(size_t num_rows);
+Result<std::shared_ptr<Buffer>> MakeSampleCsvBuffer(size_t num_rows, bool valid = true);
 
 }  // namespace csv
 }  // namespace arrow

cpp/src/arrow/json/reader.cc

@@ -29,6 +29,7 @@
 #include "arrow/json/parser.h"
 #include "arrow/record_batch.h"
 #include "arrow/table.h"
+#include "arrow/util/async_iterator.h"
 #include "arrow/util/iterator.h"
 #include "arrow/util/logging.h"
 #include "arrow/util/string_view.h"

cpp/src/arrow/testing/gtest_util.h

@@ -144,18 +144,26 @@
     }                                                        \
   } while (false)
 
-#define ASSERT_FINISHES_OK(fut)                                               \
+#define ASSERT_FINISHES_OK(expr)                                              \
   do {                                                                        \
-    ASSERT_TRUE(fut.Wait(2));                                                 \
-    if (!fut.is_finished()) {                                                 \
+    auto&& _fut = (expr);                                                     \
+    ASSERT_TRUE(_fut.Wait(2));                                                \
+    if (!_fut.is_finished()) {                                                \
       FAIL() << "Future did not finish in a timely fashion";                  \
     }                                                                         \
-    auto _st = fut.status();                                                  \
+    auto _st = _fut.status();                                                 \
     if (!_st.ok()) {                                                          \
       FAIL() << "'" ARROW_STRINGIFY(expr) "' failed with " << _st.ToString(); \
     }                                                                         \
   } while (false)
 
+#define ASSERT_FINISHES_ERR(ENUM, expr) \
+  do {                                  \
+    auto&& fut = (expr);                \
+    ASSERT_FINISHES_IMPL(fut);          \
+    ASSERT_RAISES(ENUM, fut.status());  \
+  } while (false)
+
 #define ASSERT_FINISHES_OK_AND_ASSIGN_IMPL(lhs, rexpr, future_name) \
   auto future_name = (rexpr);                                       \
   ASSERT_FINISHES_IMPL(future_name);                                \