[C++] How to correctly accelerate Parquet read using the FileReader::RowGroup(i)->ReadTable interface

[renyd123]

Describe the usage question you have. Please include as many useful details as possible.

I encountered a performance issue when calling the FileReader::RowGroup(i)->ReadTable interface in a multi-threaded environment. For a RowGroup with 1 million rows, when the number of threads is 1, FileReader::RowGroup(i)->ReadTable takes around 40ms. However, as the number of threads increases, the execution time also increases gradually. When the number of threads reaches 16, the execution time reaches around 250ms.
This is reflected in the overall file parsing performance, where using 16 threads is only about 5 times faster than using a single thread.
Is this issue caused by locking or I/O in the interface? Are there any other methods for multi-threaded parsing?

Component(s)

C++

[mapleFU]

For a RowGroup with 1 million rows, when the number of threads is 1, FileReader::RowGroup(i)->ReadTable takes around 40ms. However, as the number of threads increases, the execution time also increases gradually. When the number of threads reaches 16, the execution time reaches around 250ms.

Are you read file on object store or Local FileSystem? I guess the task is a CPU-bound task, and may I ask the CPU count you're using?

[renyd123]

For a RowGroup with 1 million rows, when the number of threads is 1, FileReader::RowGroup(i)->ReadTable takes around 40ms. However, as the number of threads increases, the execution time also increases gradually. When the number of threads reaches 16, the execution time reaches around 250ms.

Are you read file on object store or Local FileSystem? I guess the task is a CPU-bound task, and may I ask the CPU count you're using?

Local FileSystem. The Parquet file is 2GB in size and contains almost 100 row groups. My CPU is an Intel Core i7-13700K, which has 16 cores and 24 threads.
When the number of threads increases, the execution time of this line of code also increases:
PARQUET_THROW_NOT_OK(reader->RowGroup(i)->ReadTable(&table));
(reader is parquet::arrow::FileReader* reader)

[mapleFU]

I've check the code here:

arrow/cpp/src/parquet/arrow/reader.cc

Lines 1247 to 1284 in 03a960d

	Future<std::shared_ptr<Table>> FileReaderImpl::DecodeRowGroups(
	std::shared_ptr<FileReaderImpl> self, const std::vector<int>& row_groups,
	const std::vector<int>& column_indices, ::arrow::internal::Executor* cpu_executor) {
	// `self` is used solely to keep `this` alive in an async context - but we use this
	// in a sync context too so use `this` over `self`
	std::vector<std::shared_ptr<ColumnReaderImpl>> readers;
	std::shared_ptr<::arrow::Schema> result_schema;
	RETURN_NOT_OK(GetFieldReaders(column_indices, row_groups, &readers, &result_schema));
	// OptionalParallelForAsync requires an executor
	if (!cpu_executor) cpu_executor = ::arrow::internal::GetCpuThreadPool();
	auto read_column = [row_groups, self, this](size_t i,
	std::shared_ptr<ColumnReaderImpl> reader)
	-> ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> {
	std::shared_ptr<::arrow::ChunkedArray> column;
	RETURN_NOT_OK(ReadColumn(static_cast<int>(i), row_groups, reader.get(), &column));
	return column;
	};
	auto make_table = [result_schema, row_groups, self,
	this](const ::arrow::ChunkedArrayVector& columns)
	-> ::arrow::Result<std::shared_ptr<Table>> {
	int64_t num_rows = 0;
	if (!columns.empty()) {
	num_rows = columns[0]->length();
	} else {
	for (int i : row_groups) {
	num_rows += parquet_reader()->metadata()->RowGroup(i)->num_rows();
	}
	}
	auto table = Table::Make(std::move(result_schema), columns, num_rows);
	RETURN_NOT_OK(table->Validate());
	return table;
	};
	return ::arrow::internal::OptionalParallelForAsync(reader_properties_.use_threads(),
	std::move(readers), read_column,
	cpu_executor)
	.Then(std::move(make_table));
	}

CPU Executor is used when use_thread enabled ( C++ disable use_thread by default ), and per-thread calls a ReadColumn to emit the column out, generally it couldn't get slower here. I guess maybe that's because contention on CPU thread or other reason 🤔, maybe I can draft some benchmark later this week

[mapleFU]

Besides, the concurrency is in column-level, which means in per rowgroup, it decode multiple leaf-column concurrently, how many columns do your file has?

[renyd123]

Besides, the concurrency is in column-level, which means in per rowgroup, it decode multiple leaf-column concurrently, how many columns do your file has?

only 4

[renyd123]

I've check the code here:

arrow/cpp/src/parquet/arrow/reader.cc

Lines 1247 to 1284 in 03a960d

	Future<std::shared_ptr<Table>> FileReaderImpl::DecodeRowGroups(
	std::shared_ptr<FileReaderImpl> self, const std::vector<int>& row_groups,
	const std::vector<int>& column_indices, ::arrow::internal::Executor* cpu_executor) {
	// `self` is used solely to keep `this` alive in an async context - but we use this
	// in a sync context too so use `this` over `self`
	std::vector<std::shared_ptr<ColumnReaderImpl>> readers;
	std::shared_ptr<::arrow::Schema> result_schema;
	RETURN_NOT_OK(GetFieldReaders(column_indices, row_groups, &readers, &result_schema));
	// OptionalParallelForAsync requires an executor
	if (!cpu_executor) cpu_executor = ::arrow::internal::GetCpuThreadPool();
	auto read_column = [row_groups, self, this](size_t i,
	std::shared_ptr<ColumnReaderImpl> reader)
	-> ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> {
	std::shared_ptr<::arrow::ChunkedArray> column;
	RETURN_NOT_OK(ReadColumn(static_cast<int>(i), row_groups, reader.get(), &column));
	return column;
	};
	auto make_table = [result_schema, row_groups, self,
	this](const ::arrow::ChunkedArrayVector& columns)
	-> ::arrow::Result<std::shared_ptr<Table>> {
	int64_t num_rows = 0;
	if (!columns.empty()) {
	num_rows = columns[0]->length();
	} else {
	for (int i : row_groups) {
	num_rows += parquet_reader()->metadata()->RowGroup(i)->num_rows();
	}
	}
	auto table = Table::Make(std::move(result_schema), columns, num_rows);
	RETURN_NOT_OK(table->Validate());
	return table;
	};
	return ::arrow::internal::OptionalParallelForAsync(reader_properties_.use_threads(),
	std::move(readers), read_column,
	cpu_executor)
	.Then(std::move(make_table));
	}

CPU Executor is used when enabled ( C++ disable use_thread by default ), and per-thread calls a to emit the column out, generally it couldn't get slower here. I guess maybe that's because contention on CPU thread or other reason 🤔, maybe I can draft some benchmark later this weekuse_thread``ReadColumn

I deliberately turned off 'use_thread' and instead manually called 'reader->RowGroup(i)->ReadTable(&table)' with multiple threads just like:
threads.push_back(std::thread([&reader, i]() { reader->RowGroup(i)->ReadTable(&table); }));

as the doc says:
More advanced users that also want to implement parallelism on top of each single Parquet files should do this on the RowGroup level. For this, they can call FileReader::RowGroup(i)->ReadTable to receive only the specified RowGroup as a table.

[mapleFU]

Aha. so a thread only handle one ReadTable call? Basically, if encryption is not enabled, and file doesn't on object-store, ReadTable is probably a cpu-bound task...I'll try to repro on my 3800x...

[renyd123]

Aha. so a thread only handle one ReadTable call? Basically, if encryption is not enabled, and file doesn't on object-store, ReadTable is probably a cpu-bound task...I'll try to repro on my 3800x...

yes, 16 threads handle 16 RowGroups in parallel. And columns are processed serially in each thread.

[github-actions]

This issue has been marked as stale because it has had no activity in the past 365 days. Please remove the stale label or comment below, or this issue will be closed in 14 days. If this usage question has evolved into a feature request or docs update, please remove the 'Type: usage' label and add the 'Type: enhancement' label instead.

[thisisnic]

Closing as stale usage question with no recent activity.