Revert improve readers by parallelizing I/O and compute operations. (#5471)

This reverts #5401 for parallelizing I/O and compute operations
(backported in #5451) and the follow-up PR #5446 (backported in #5458).

---
TYPE: IMPROVEMENT
DESC: Revert improve readers by parallelizing I/O and compute operations

Author: shaunrd0

test/src/unit-ReadCellSlabIter.cc

@@ -183,10 +183,7 @@ void set_result_tile_dim(
       std::nullopt,
       std::nullopt,
       std::nullopt);
-  ResultTile::TileData tile_data{
-      {nullptr, ThreadPool::SharedTask()},
-      {nullptr, ThreadPool::SharedTask()},
-      {nullptr, ThreadPool::SharedTask()}};
+  ResultTile::TileData tile_data{nullptr, nullptr, nullptr};
   result_tile.init_coord_tile(
       constants::format_version,
       array_schema,

test/src/unit-cppapi-consolidation-with-timestamps.cc

@@ -636,7 +636,7 @@ TEST_CASE_METHOD(
 
   // Will only allow to load two tiles out of 3.
   Config cfg;
-  cfg.set("sm.mem.total_budget", "50000");
+  cfg.set("sm.mem.total_budget", "30000");
   cfg.set("sm.mem.reader.sparse_global_order.ratio_coords", "0.15");
   ctx_ = Context(cfg);
 
@@ -685,7 +685,7 @@ TEST_CASE_METHOD(
 
   // Will only allow to load two tiles out of 3.
   Config cfg;
-  cfg.set("sm.mem.total_budget", "50000");
+  cfg.set("sm.mem.total_budget", "30000");
   cfg.set("sm.mem.reader.sparse_global_order.ratio_coords", "0.15");
   ctx_ = Context(cfg);
 

test/src/unit-result-tile.cc

@@ -213,10 +213,7 @@ TEST_CASE_METHOD(
         0,
         std::nullopt,
         std::nullopt);
-    ResultTile::TileData tile_data{
-        {nullptr, ThreadPool::SharedTask()},
-        {nullptr, ThreadPool::SharedTask()},
-        {nullptr, ThreadPool::SharedTask()}};
+    ResultTile::TileData tile_data{nullptr, nullptr, nullptr};
     rt.init_coord_tile(
         constants::format_version,
         array_schema,
@@ -233,10 +230,7 @@ TEST_CASE_METHOD(
       0,
       std::nullopt,
       std::nullopt);
-  ResultTile::TileData tile_data{
-      {nullptr, ThreadPool::SharedTask()},
-      {nullptr, ThreadPool::SharedTask()},
-      {nullptr, ThreadPool::SharedTask()}};
+  ResultTile::TileData tile_data{nullptr, nullptr, nullptr};
   rt.init_coord_tile(
       constants::format_version,
       array_schema,
@@ -332,10 +326,7 @@ TEST_CASE_METHOD(
         0,
         std::nullopt,
         std::nullopt);
-    ResultTile::TileData tile_data{
-        {nullptr, ThreadPool::SharedTask()},
-        {nullptr, ThreadPool::SharedTask()},
-        {nullptr, ThreadPool::SharedTask()}};
+    ResultTile::TileData tile_data{nullptr, nullptr, nullptr};
     rt.init_coord_tile(
         constants::format_version,
         array_schema,
@@ -352,10 +343,7 @@ TEST_CASE_METHOD(
       0,
       std::nullopt,
       std::nullopt);
-  ResultTile::TileData tile_data{
-      {nullptr, ThreadPool::SharedTask()},
-      {nullptr, ThreadPool::SharedTask()},
-      {nullptr, ThreadPool::SharedTask()}};
+  ResultTile::TileData tile_data{nullptr, nullptr, nullptr};
   rt.init_coord_tile(
       constants::format_version,
       array_schema,

test/src/unit-sparse-global-order-reader.cc

@@ -2224,10 +2224,9 @@ TEST_CASE_METHOD(
   }
 
   // FIXME: there is no per fragment budget anymore
-  // Two result tiles (2 * (2842 + 8)) = 5700 will be bigger than the per
-  // fragment budget (50000 * 0.11 / 2 fragments = 2750), so only one result
-  // tile will be loaded each time.
-  memory_.total_budget_ = "60000";
+  // Two result tile (2 * (~3000 + 8) will be bigger than the per fragment
+  // budget (1000).
+  memory_.total_budget_ = "35000";
   memory_.ratio_coords_ = "0.11";
   update_config();
 
@@ -2750,9 +2749,8 @@ TEST_CASE_METHOD(
   }
 
   // FIXME: there is no per fragment budget anymore
-  // Two result tiles (2 * (2842 + 8)) = 5700 will be bigger than the per
-  // fragment budget (40000 * 0.22 /2 frag = 4400), so only one will be loaded
-  // each time.
+  // Two result tile (2 * (~4000 + 8) will be bigger than the per fragment
+  // budget (1000).
   memory_.total_budget_ = "40000";
   memory_.ratio_coords_ = "0.22";
   update_config();

test/src/unit-sparse-unordered-with-dups-reader.cc

@@ -1064,12 +1064,9 @@ TEST_CASE_METHOD(
 
     if (one_frag) {
       CHECK(1 == loop_num->second);
+    } else {
+      CHECK(9 == loop_num->second);
     }
-    /**
-     * We can't do a similar check for multiple fragments as it is architecture
-     * dependent how many tiles fit in the memory budget. And thus also
-     * architecture dependent as to how many internal loops we have.
-     */
 
     // Try to read multiple frags without partial tile offset reading. Should
     // fail

tiledb/sm/filter/compression_filter.cc

@@ -636,7 +636,7 @@ Status CompressionFilter::decompress_var_string_coords(
   auto output_view = span<std::byte>(
       reinterpret_cast<std::byte*>(output_buffer->data()), uncompressed_size);
   auto offsets_view = span<uint64_t>(
-      offsets_tile->data_as_unsafe<offsets_t>(), uncompressed_offsets_size);
+      offsets_tile->data_as<offsets_t>(), uncompressed_offsets_size);
 
   if (compressor_ == Compressor::RLE) {
     uint8_t rle_len_bytesize, string_len_bytesize;

tiledb/sm/filter/filter_pipeline.cc

@@ -461,7 +461,7 @@ Status FilterPipeline::run_reverse(
     // If the pipeline is empty, just copy input to output.
     if (filters_.empty()) {
       void* output_chunk_buffer =
-          tile->data_as_unsafe<char>() + chunk_data.chunk_offsets_[i];
+          tile->data_as<char>() + chunk_data.chunk_offsets_[i];
       RETURN_NOT_OK(input_data.copy_to(output_chunk_buffer));
       continue;
     }
@@ -484,7 +484,7 @@ Status FilterPipeline::run_reverse(
       bool last_filter = filter_idx == 0;
       if (last_filter) {
         void* output_chunk_buffer =
-            tile->data_as_unsafe<char>() + chunk_data.chunk_offsets_[i];
+            tile->data_as<char>() + chunk_data.chunk_offsets_[i];
         RETURN_NOT_OK(output_data.set_fixed_allocation(
             output_chunk_buffer, chunk.unfiltered_data_size_));
         reader_stats->add_counter(

tiledb/sm/filter/test/filter_test_support.cc

@@ -203,8 +203,7 @@ Tile create_tile_for_unfiltering(
       tile->cell_size() * nelts,
       tile->filtered_buffer().data(),
       tile->filtered_buffer().size(),
-      tracker,
-      std::nullopt};
+      tracker};
 }
 
 void run_reverse(

tiledb/sm/filter/test/tile_data_generator.h

@@ -99,8 +99,7 @@ class TileDataGenerator {
         original_tile_size(),
         filtered_buffer.data(),
         filtered_buffer.size(),
-        memory_tracker,
-        std::nullopt);
+        memory_tracker);
   }
 
   /** Returns the size of the original unfiltered data. */

tiledb/sm/metadata/test/unit_metadata.cc

@@ -123,8 +123,7 @@ TEST_CASE(
       tile1->size(),
       tile1->filtered_buffer().data(),
       tile1->filtered_buffer().size(),
-      tracker,
-      ThreadPool::SharedTask());
+      tracker);
   memcpy(metadata_tiles[0]->data(), tile1->data(), tile1->size());
 
   metadata_tiles[1] = tdb::make_shared<Tile>(
@@ -136,8 +135,7 @@ TEST_CASE(
       tile2->size(),
       tile2->filtered_buffer().data(),
       tile2->filtered_buffer().size(),
-      tracker,
-      ThreadPool::SharedTask());
+      tracker);
   memcpy(metadata_tiles[1]->data(), tile2->data(), tile2->size());
 
   metadata_tiles[2] = tdb::make_shared<Tile>(
@@ -149,8 +147,7 @@ TEST_CASE(
       tile3->size(),
       tile3->filtered_buffer().data(),
       tile3->filtered_buffer().size(),
-      tracker,
-      ThreadPool::SharedTask());
+      tracker);
   memcpy(metadata_tiles[2]->data(), tile3->data(), tile3->size());
 
   meta = Metadata::deserialize(metadata_tiles);