This is an automated email from the ASF dual-hosted git repository.
zanmato pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/main by this push:
new ea0b3c5226 GH-45877: [C++][Acero] Cleanup 64-bit temp states of Swiss
join by using 32-bit (#45878)
ea0b3c5226 is described below
commit ea0b3c522693b4afea82f492dc10f894fabbee58
Author: Rossi Sun <[email protected]>
AuthorDate: Mon Mar 24 13:16:47 2025 +0800
GH-45877: [C++][Acero] Cleanup 64-bit temp states of Swiss join by using
32-bit (#45878)
### Rationale for this change
See #45877.
### What changes are included in this PR?
1. Normalize all temp states/variables representing "row id"s to `uint32_t`.
2. Cleanup many `DCHECK`s (some added though) and `static_cast`s.
### Are these changes tested?
Existing tests should suffice.
### Are there any user-facing changes?
None.
* GitHub Issue: #45877
Authored-by: Rossi Sun <[email protected]>
Signed-off-by: Rossi Sun <[email protected]>
---
cpp/src/arrow/acero/swiss_join.cc | 113 ++++++++++++++++--------------
cpp/src/arrow/acero/swiss_join_internal.h | 22 +++---
2 files changed, 70 insertions(+), 65 deletions(-)
diff --git a/cpp/src/arrow/acero/swiss_join.cc
b/cpp/src/arrow/acero/swiss_join.cc
index b4d89df290..a32ea7d3ac 100644
--- a/cpp/src/arrow/acero/swiss_join.cc
+++ b/cpp/src/arrow/acero/swiss_join.cc
@@ -392,7 +392,7 @@ void RowArray::DecodeNulls(ResizableArrayData* output, int
output_start_row,
Status RowArrayMerge::PrepareForMerge(RowArray* target,
const std::vector<RowArray*>& sources,
- std::vector<int64_t>*
first_target_row_id,
+ std::vector<uint32_t>*
first_target_row_id,
MemoryPool* pool, int64_t
hardware_flags) {
ARROW_DCHECK(!sources.empty());
@@ -404,7 +404,7 @@ Status RowArrayMerge::PrepareForMerge(RowArray* target,
// Sum the number of rows from all input sources and calculate their total
// size.
//
- int64_t num_rows = 0;
+ uint32_t num_rows = 0;
int64_t num_bytes = 0;
if (first_target_row_id) {
first_target_row_id->resize(sources.size() + 1);
@@ -417,7 +417,9 @@ Status RowArrayMerge::PrepareForMerge(RowArray* target,
if (first_target_row_id) {
(*first_target_row_id)[i] = num_rows;
}
- num_rows += sources[i]->rows_.length();
+ DCHECK_LE(num_rows + sources[i]->rows_.length(),
+ std::numeric_limits<uint32_t>::max());
+ num_rows += static_cast<uint32_t>(sources[i]->rows_.length());
if (!metadata.is_fixed_length) {
num_bytes += sources[i]->rows_.offsets()[sources[i]->rows_.length()];
}
@@ -429,7 +431,7 @@ Status RowArrayMerge::PrepareForMerge(RowArray* target,
// Allocate target memory
//
target->rows_.Clean();
- RETURN_NOT_OK(target->rows_.AppendEmpty(static_cast<uint32_t>(num_rows),
num_bytes));
+ RETURN_NOT_OK(target->rows_.AppendEmpty(num_rows, num_bytes));
// In case of varying length rows,
// initialize the first row offset for each range of rows corresponding to a
@@ -440,7 +442,9 @@ Status RowArrayMerge::PrepareForMerge(RowArray* target,
num_bytes = 0;
for (size_t i = 0; i < sources.size(); ++i) {
target->rows_.mutable_offsets()[num_rows] = num_bytes;
- num_rows += sources[i]->rows_.length();
+ DCHECK_LE(num_rows + sources[i]->rows_.length(),
+ std::numeric_limits<uint32_t>::max());
+ num_rows += static_cast<uint32_t>(sources[i]->rows_.length());
num_bytes += sources[i]->rows_.offsets()[sources[i]->rows_.length()];
}
target->rows_.mutable_offsets()[num_rows] = num_bytes;
@@ -450,8 +454,8 @@ Status RowArrayMerge::PrepareForMerge(RowArray* target,
}
void RowArrayMerge::MergeSingle(RowArray* target, const RowArray& source,
- int64_t first_target_row_id,
- const int64_t* source_rows_permutation) {
+ uint32_t first_target_row_id,
+ const uint32_t* source_rows_permutation) {
// Source and target must:
// - be initialized
// - use the same row format
@@ -473,8 +477,8 @@ void RowArrayMerge::MergeSingle(RowArray* target, const
RowArray& source,
}
void RowArrayMerge::CopyFixedLength(RowTableImpl* target, const RowTableImpl&
source,
- int64_t first_target_row_id,
- const int64_t* source_rows_permutation) {
+ uint32_t first_target_row_id,
+ const uint32_t* source_rows_permutation) {
int64_t num_source_rows = source.length();
uint32_t fixed_length = target->metadata().fixed_length;
@@ -483,8 +487,7 @@ void RowArrayMerge::CopyFixedLength(RowTableImpl* target,
const RowTableImpl& so
// needed is memcpy.
//
if (!source_rows_permutation) {
- DCHECK_LE(first_target_row_id, std::numeric_limits<uint32_t>::max());
-
memcpy(target->mutable_fixed_length_rows(static_cast<uint32_t>(first_target_row_id)),
+ memcpy(target->mutable_fixed_length_rows(first_target_row_id),
source.fixed_length_rows(/*row_id=*/0), fixed_length *
num_source_rows);
} else {
// Row length must be a multiple of 64-bits due to enforced alignment.
@@ -493,15 +496,13 @@ void RowArrayMerge::CopyFixedLength(RowTableImpl* target,
const RowTableImpl& so
ARROW_DCHECK(fixed_length % sizeof(uint64_t) == 0);
int64_t num_words_per_row = fixed_length / sizeof(uint64_t);
- for (int64_t i = 0; i < num_source_rows; ++i) {
- int64_t source_row_id = source_rows_permutation[i];
- DCHECK_LE(source_row_id, std::numeric_limits<uint32_t>::max());
- const uint64_t* source_row_ptr = reinterpret_cast<const uint64_t*>(
- source.fixed_length_rows(static_cast<uint32_t>(source_row_id)));
- int64_t target_row_id = first_target_row_id + i;
- DCHECK_LE(target_row_id, std::numeric_limits<uint32_t>::max());
- uint64_t* target_row_ptr = reinterpret_cast<uint64_t*>(
-
target->mutable_fixed_length_rows(static_cast<uint32_t>(target_row_id)));
+ for (uint32_t i = 0; i < num_source_rows; ++i) {
+ uint32_t source_row_id = source_rows_permutation[i];
+ const uint64_t* source_row_ptr =
+ reinterpret_cast<const
uint64_t*>(source.fixed_length_rows(source_row_id));
+ uint32_t target_row_id = first_target_row_id + i;
+ uint64_t* target_row_ptr =
+
reinterpret_cast<uint64_t*>(target->mutable_fixed_length_rows(target_row_id));
for (int64_t word = 0; word < num_words_per_row; ++word) {
target_row_ptr[word] = source_row_ptr[word];
@@ -511,9 +512,9 @@ void RowArrayMerge::CopyFixedLength(RowTableImpl* target,
const RowTableImpl& so
}
void RowArrayMerge::CopyVaryingLength(RowTableImpl* target, const
RowTableImpl& source,
- int64_t first_target_row_id,
+ uint32_t first_target_row_id,
int64_t first_target_row_offset,
- const int64_t* source_rows_permutation) {
+ const uint32_t* source_rows_permutation)
{
int64_t num_source_rows = source.length();
RowTableImpl::offset_type* target_offsets = target->mutable_offsets();
const RowTableImpl::offset_type* source_offsets = source.offsets();
@@ -540,7 +541,7 @@ void RowArrayMerge::CopyVaryingLength(RowTableImpl* target,
const RowTableImpl&
uint64_t* target_row_ptr = reinterpret_cast<uint64_t*>(
target->mutable_var_length_rows() + target_row_offset);
for (int64_t i = 0; i < num_source_rows; ++i) {
- int64_t source_row_id = source_rows_permutation[i];
+ uint32_t source_row_id = source_rows_permutation[i];
const uint64_t* source_row_ptr = reinterpret_cast<const uint64_t*>(
source.var_length_rows() + source_offsets[source_row_id]);
int64_t length = source_offsets[source_row_id + 1] -
source_offsets[source_row_id];
@@ -564,22 +565,18 @@ void RowArrayMerge::CopyVaryingLength(RowTableImpl*
target, const RowTableImpl&
}
void RowArrayMerge::CopyNulls(RowTableImpl* target, const RowTableImpl& source,
- int64_t first_target_row_id,
- const int64_t* source_rows_permutation) {
+ uint32_t first_target_row_id,
+ const uint32_t* source_rows_permutation) {
int64_t num_source_rows = source.length();
int num_bytes_per_row = target->metadata().null_masks_bytes_per_row;
- DCHECK_LE(first_target_row_id, std::numeric_limits<uint32_t>::max());
- uint8_t* target_nulls =
- target->mutable_null_masks(static_cast<uint32_t>(first_target_row_id));
+ uint8_t* target_nulls = target->mutable_null_masks(first_target_row_id);
if (!source_rows_permutation) {
memcpy(target_nulls, source.null_masks(/*row_id=*/0),
num_bytes_per_row * num_source_rows);
} else {
for (uint32_t i = 0; i < num_source_rows; ++i) {
- int64_t source_row_id = source_rows_permutation[i];
- DCHECK_LE(source_row_id, std::numeric_limits<uint32_t>::max());
- const uint8_t* source_nulls =
- source.null_masks(static_cast<uint32_t>(source_row_id));
+ uint32_t source_row_id = source_rows_permutation[i];
+ const uint8_t* source_nulls = source.null_masks(source_row_id);
for (int64_t byte = 0; byte < num_bytes_per_row; ++byte) {
*target_nulls++ = *source_nulls++;
}
@@ -1301,10 +1298,12 @@ Status SwissTableForJoinBuild::PreparePrtnMerge() {
// Check if we have duplicate keys
//
- int64_t num_keys = partition_keys_first_row_id_[num_prtns_];
- int64_t num_rows = 0;
+ uint32_t num_keys = partition_keys_first_row_id_[num_prtns_];
+ uint32_t num_rows = 0;
for (int i = 0; i < num_prtns_; ++i) {
- num_rows += static_cast<int64_t>(prtn_states_[i].key_ids.size());
+ DCHECK_LE(num_rows + prtn_states_[i].key_ids.size(),
+ std::numeric_limits<uint32_t>::max());
+ num_rows += static_cast<uint32_t>(prtn_states_[i].key_ids.size());
}
bool no_duplicate_keys = reject_duplicate_keys_ || num_keys == num_rows;
@@ -1313,13 +1312,15 @@ Status SwissTableForJoinBuild::PreparePrtnMerge() {
target_->no_duplicate_keys_ = no_duplicate_keys;
if (!no_duplicate_keys) {
target_->row_offset_for_key_.resize(num_keys + 1);
- int64_t num_rows = 0;
+ uint32_t num_rows = 0;
for (int i = 0; i < num_prtns_; ++i) {
- int64_t first_key = partition_keys_first_row_id_[i];
- target_->row_offset_for_key_[first_key] =
static_cast<uint32_t>(num_rows);
- num_rows += static_cast<int64_t>(prtn_states_[i].key_ids.size());
+ uint32_t first_key = partition_keys_first_row_id_[i];
+ target_->row_offset_for_key_[first_key] = num_rows;
+ DCHECK_LE(num_rows + prtn_states_[i].key_ids.size(),
+ std::numeric_limits<uint32_t>::max());
+ num_rows += static_cast<uint32_t>(prtn_states_[i].key_ids.size());
}
- target_->row_offset_for_key_[num_keys] = static_cast<uint32_t>(num_rows);
+ target_->row_offset_for_key_[num_keys] = num_rows;
}
return Status::OK();
@@ -1327,6 +1328,10 @@ Status SwissTableForJoinBuild::PreparePrtnMerge() {
void SwissTableForJoinBuild::PrtnMerge(int prtn_id) {
PartitionState& prtn_state = prtn_states_[prtn_id];
+ // Emscripten has size_t as unsigned long thus warns about this comparison
being always
+ // true. We explicit cast to unsigned long long.
+ DCHECK_LE(static_cast<uint64_t>(prtn_state.key_ids.size()),
+ std::numeric_limits<uint32_t>::max() + 1ull);
// There are 4 data structures that require partition merging:
// 1. array of key rows
@@ -1346,10 +1351,10 @@ void SwissTableForJoinBuild::PrtnMerge(int prtn_id) {
//
SwissTableMerge::MergePartition(
target_->map_.swiss_table(), prtn_state.keys.swiss_table(), prtn_id,
log_num_prtns_,
- static_cast<uint32_t>(partition_keys_first_row_id_[prtn_id]),
- &prtn_state.overflow_key_ids, &prtn_state.overflow_hashes);
+ partition_keys_first_row_id_[prtn_id], &prtn_state.overflow_key_ids,
+ &prtn_state.overflow_hashes);
- std::vector<int64_t> source_payload_ids;
+ std::vector<uint32_t> source_payload_ids;
// 3. mapping from key id to first payload id
//
@@ -1360,11 +1365,11 @@ void SwissTableForJoinBuild::PrtnMerge(int prtn_id) {
// For convenience, we use an array in merged hash table mapping key ids to
// first payload ids to collect the counters.
//
- int64_t first_key = partition_keys_first_row_id_[prtn_id];
- int64_t num_keys = partition_keys_first_row_id_[prtn_id + 1] - first_key;
+ uint32_t first_key = partition_keys_first_row_id_[prtn_id];
+ uint32_t num_keys = partition_keys_first_row_id_[prtn_id + 1] - first_key;
uint32_t* counters = target_->row_offset_for_key_.data() + first_key;
uint32_t first_payload = counters[0];
- for (int64_t i = 0; i < num_keys; ++i) {
+ for (uint32_t i = 0; i < num_keys; ++i) {
counters[i] = 0;
}
for (size_t i = 0; i < prtn_state.key_ids.size(); ++i) {
@@ -1378,7 +1383,7 @@ void SwissTableForJoinBuild::PrtnMerge(int prtn_id) {
// Start by computing inclusive cumulative sum of counters.
//
uint32_t sum = 0;
- for (int64_t i = 0; i < num_keys; ++i) {
+ for (uint32_t i = 0; i < num_keys; ++i) {
sum += counters[i];
counters[i] = sum;
}
@@ -1388,14 +1393,14 @@ void SwissTableForJoinBuild::PrtnMerge(int prtn_id) {
// there at the beginning).
//
source_payload_ids.resize(prtn_state.key_ids.size());
- for (size_t i = 0; i < prtn_state.key_ids.size(); ++i) {
+ for (uint32_t i = 0; i < prtn_state.key_ids.size(); ++i) {
uint32_t key_id = prtn_state.key_ids[i];
- int64_t position = --counters[key_id];
- source_payload_ids[position] = static_cast<int64_t>(i);
+ uint32_t position = --counters[key_id];
+ source_payload_ids[position] = i;
}
// Add base payload id to all of the counters.
//
- for (int64_t i = 0; i < num_keys; ++i) {
+ for (uint32_t i = 0; i < num_keys; ++i) {
counters[i] += first_payload;
}
} else {
@@ -1403,7 +1408,7 @@ void SwissTableForJoinBuild::PrtnMerge(int prtn_id) {
// cumulative sum of counters and add the base payload id to all of them.
//
uint32_t sum = 0;
- for (int64_t i = 0; i < num_keys; ++i) {
+ for (uint32_t i = 0; i < num_keys; ++i) {
uint32_t sum_next = sum + counters[i];
counters[i] = sum + first_payload;
sum = sum_next;
@@ -1419,9 +1424,9 @@ void SwissTableForJoinBuild::PrtnMerge(int prtn_id) {
//
if (target_->no_duplicate_keys_) {
source_payload_ids.resize(prtn_state.key_ids.size());
- for (size_t i = 0; i < prtn_state.key_ids.size(); ++i) {
+ for (uint32_t i = 0; i < prtn_state.key_ids.size(); ++i) {
uint32_t key_id = prtn_state.key_ids[i];
- source_payload_ids[key_id] = static_cast<int64_t>(i);
+ source_payload_ids[key_id] = i;
}
}
// Merge partition payloads into target array using the permutation.
diff --git a/cpp/src/arrow/acero/swiss_join_internal.h
b/cpp/src/arrow/acero/swiss_join_internal.h
index 365f2917d8..ab7c5acaa9 100644
--- a/cpp/src/arrow/acero/swiss_join_internal.h
+++ b/cpp/src/arrow/acero/swiss_join_internal.h
@@ -261,7 +261,7 @@ class RowArrayMerge {
// caller can pass in nullptr to indicate that it is not needed.
//
static Status PrepareForMerge(RowArray* target, const
std::vector<RowArray*>& sources,
- std::vector<int64_t>* first_target_row_id,
+ std::vector<uint32_t>* first_target_row_id,
MemoryPool* pool, int64_t hardware_flags);
// Copy rows from source array to target array.
@@ -286,8 +286,8 @@ class RowArrayMerge {
// initializes boundary offsets for target row ranges for each source).
//
static void MergeSingle(RowArray* target, const RowArray& source,
- int64_t first_target_row_id,
- const int64_t* source_rows_permutation);
+ uint32_t first_target_row_id,
+ const uint32_t* source_rows_permutation);
private:
// Copy rows from source array to a region of the target array.
@@ -295,24 +295,24 @@ class RowArrayMerge {
// Null information needs to be handled separately.
//
static void CopyFixedLength(RowTableImpl* target, const RowTableImpl& source,
- int64_t first_target_row_id,
- const int64_t* source_rows_permutation);
+ uint32_t first_target_row_id,
+ const uint32_t* source_rows_permutation);
// Copy rows from source array to a region of the target array.
// This implementation is for varying length rows.
// Null information needs to be handled separately.
//
static void CopyVaryingLength(RowTableImpl* target, const RowTableImpl&
source,
- int64_t first_target_row_id,
+ uint32_t first_target_row_id,
int64_t first_target_row_offset,
- const int64_t* source_rows_permutation);
+ const uint32_t* source_rows_permutation);
// Copy null information from rows from source array to a region of the
target
// array.
//
static void CopyNulls(RowTableImpl* target, const RowTableImpl& source,
- int64_t first_target_row_id,
- const int64_t* source_rows_permutation);
+ uint32_t first_target_row_id,
+ const uint32_t* source_rows_permutation);
};
// Implements merging of multiple SwissTables into a single one, using
@@ -649,8 +649,8 @@ class SwissTableForJoinBuild {
std::vector<PartitionState> prtn_states_;
std::vector<ThreadState> thread_states_;
- std::vector<int64_t> partition_keys_first_row_id_;
- std::vector<int64_t> partition_payloads_first_row_id_;
+ std::vector<uint32_t> partition_keys_first_row_id_;
+ std::vector<uint32_t> partition_payloads_first_row_id_;
};
class JoinResultMaterialize {
