Repository: apache/arrow
File: cpp/src/arrow/dataset/file_parquet_test.cc (C++)

Code before:
      options->fragment_scan_options = fragment_scan_options;
      ScannerBuilder builder(ArithmeticDatasetFixture::schema(), fragment, options);

      ASSERT_OK(builder.UseThreads(true));
      ASSERT_OK(builder.BatchSize(10000));
      ASSERT_OK_AND_ASSIGN(scanner, builder.Finish());
    }

    ASSERT_OK_AND_ASSIGN(auto batch, scanner->Head(10000));
    [[maybe_unused]] auto fut = scanner->ScanBatchesUnorderedAsync();
    // Random ReadAsync calls, generate some futures to make the state machine
    // more complex.
    for (int yy = 0; yy < 16; yy++) {
      completes.emplace_back(buffer_reader->ReadAsync(::arrow::io::IOContext(), 0, 1001));
    }
    scanner = nullptr;
  }

  for (auto& f : completes) {
    f.Wait();
  }
}

TEST_F(TestParquetFileFormat, MultithreadedComputeRegression) {
  // GH-43694: Test similar situation as MultithreadedScanRegression but with
  // the exec context instead

  auto reader = MakeGeneratedRecordBatch(schema({field("utf8", utf8())}), 10000, 100);
  ASSERT_OK_AND_ASSIGN(auto buffer, ParquetFormatHelper::Write(reader.get()));

  std::vector<Future<>> completes;
  std::vector<std::shared_ptr<arrow::internal::ThreadPool>> pools;

  for (int idx = 0; idx < 2; ++idx) {
    auto buffer_reader = std::make_shared<DelayedBufferReader>(buffer);
    auto source = std::make_shared<FileSource>(buffer_reader, buffer->size());
    auto fragment = MakeFragment(*source);
    std::shared_ptr<Scanner> scanner;

    {
      auto options = std::make_shared<ScanOptions>();
      ASSERT_OK_AND_ASSIGN(auto thread_pool, arrow::internal::ThreadPool::Make(1));
      pools.emplace_back(thread_pool);
      options->cpu_executor = pools.back().get();
      auto fragment_scan_options = std::make_shared<ParquetFragmentScanOptions>();
      fragment_scan_options->arrow_reader_properties->set_pre_buffer(true);

      options->fragment_scan_options = fragment_scan_options;
      ScannerBuilder builder(ArithmeticDatasetFixture::schema(), fragment, options);

      ASSERT_OK(builder.UseThreads(true));

Review comment (suggestion):
```suggestion
  // customized CPU executor instead
```

Code after:
              "(((x >= 1) and (x <= 100)) or is_null(x, {nan_is_null=false}))");
  }
  {
    // Special case: when num_value is 0, it would return
    // "is_null".
    ::parquet::EncodedStatistics encoded_stats;
    encoded_stats.has_null_count = true;
    encoded_stats.null_count = 1;
    encoded_stats.all_null_value = true;
    auto stats = ::parquet::Statistics::Make(&descr, &encoded_stats, /*num_values=*/0);
    auto stat_expression =
        ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *stats);
    ASSERT_TRUE(stat_expression.has_value());
    EXPECT_EQ(stat_expression->ToString(), "is_null(x, {nan_is_null=false})");

    encoded_stats.has_null_count = false;
    encoded_stats.all_null_value = false;
    stats = ::parquet::Statistics::Make(&descr, &encoded_stats, /*num_values=*/0);
    stat_expression = ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *stats);
    ASSERT_TRUE(stat_expression.has_value());
    EXPECT_EQ(stat_expression->ToString(), "is_null(x, {nan_is_null=false})");
  }
}

TEST_F(TestParquetFileFormat, MultithreadedScanRegression) {
  // GH-38438: This test is similar to MultithreadedScan, but it try to use self
  // designed Executor and DelayedBufferReader to mock async execution to make
  // the state machine more complex.
  CustomizeScanOptionsWithThreadPool customize_io_context =
      [](ScanOptions& options, arrow::internal::ThreadPool* pool) {
        options.io_context = ::arrow::io::IOContext(::arrow::default_memory_pool(), pool);
      };
  TestMultithreadedRegression(customize_io_context);
}

TEST_F(TestParquetFileFormat, MultithreadedComputeRegression) {
  // GH-43694: Test similar situation as MultithreadedScanRegression but with
  // the customized CPU executor instead
  CustomizeScanOptionsWithThreadPool customize_cpu_executor =
      [](ScanOptions& options, arrow::internal::ThreadPool* pool) {
        options.cpu_executor = pool;
      };
  TestMultithreadedRegression(customize_cpu_executor);
}

}  // namespace dataset
}  // namespace arrow