GH-32276: [C++][FlightRPC] Align RecordBatch buffers given to IPC

[EnricoMi]

Rationale for this change

Data retrieved via IPC is expected to provide memory-aligned arrays, but data retrieved via C++ Flight client is mis-aligned. Datafusion (Rust), which requires proper alignment, cannot handle such data: #43552.

What changes are included in this PR?

This aligns RecordBatch array buffers decoded by IPC if mis-aligned according to the data type byte width.
Implementation mirrors that of align_buffers in arrow-rs (apache/arrow-rs#4681).

Are these changes tested?

Configuration flag tested in unit test.
Manually end-to-end tested that memory alignment fixes issue with reproduction code provided in #43552.

Are there any user-facing changes?

Memory alignment is checked and fixed by default. This is configurable via IpcReadOptions.ensure_memory_alignment.

• GitHub Issue: [C++][FlightRPC] Flight generates misaligned buffers #32276

[EnricoMi]

@pitrou do you think this fix is viable?

[lidavidm]

Thanks for picking this up. This seems reasonable to me at a quick glance.

[lidavidm]

nit: put this include with the rest of the Arrow includes (and use quotes to be consistent)

[EnricoMi]

done

[lidavidm]

nit: you could iterate with for (auto& child : child_data) and avoid the explicit index?

[EnricoMi]

much better!

[lidavidm]

Where did this come from?

[EnricoMi]

Fixed.

[EnricoMi]

While attempting to write some unit tests I found there is util::EnsureAlignment:

arrow/cpp/src/arrow/util/align_util.cc

Lines 169 to 205 in e62fbaa

	Result<std::shared_ptr<ArrayData>> EnsureAlignment(std::shared_ptr<ArrayData> array_data,
	int64_t alignment,
	MemoryPool* memory_pool) {
	if (!CheckAlignment(*array_data, alignment)) {
	std::vector<std::shared_ptr<Buffer>> buffers = array_data->buffers;
	Type::type type_id = GetTypeForBuffers(*array_data);
	for (size_t i = 0; i < buffers.size(); ++i) {
	if (buffers[i]) {
	int64_t expected_alignment = alignment;
	if (alignment == kValueAlignment) {
	expected_alignment =
	RequiredValueAlignmentForBuffer(type_id, static_cast<int>(i));
	}
	ARROW_ASSIGN_OR_RAISE(
	buffers[i],
	EnsureAlignment(std::move(buffers[i]), expected_alignment, memory_pool));
	}
	}
	for (auto& it : array_data->child_data) {
	ARROW_ASSIGN_OR_RAISE(it, EnsureAlignment(std::move(it), alignment, memory_pool));
	}
	if (array_data->type->id() == Type::DICTIONARY) {
	ARROW_ASSIGN_OR_RAISE(
	array_data->dictionary,
	EnsureAlignment(std::move(array_data->dictionary), alignment, memory_pool));
	}
	auto new_array_data = ArrayData::Make(
	array_data->type, array_data->length, std::move(buffers), array_data->child_data,
	array_data->dictionary, array_data->GetNullCount(), array_data->offset);
	return new_array_data;
	} else {
	return array_data;
	}
	}

I will try to reuse that method rather than re-implementing it. There is also test infrastructure for misaligned array data.