58 lines
3 KiB
Diff
58 lines
3 KiB
Diff
# UNDF: UNDF-2026-000000763
|
||
# UNDF:
|
||
--- a/cpp/src/arrow/acero/asof_join_node.cc
|
||
+++ b/cpp/src/arrow/acero/asof_join_node.cc
|
||
@@ -483,6 +483,8 @@ class InputState : public util::SerialSequencingQueue::Processor {
|
||
InputState(size_t index, TolType tolerance, bool must_hash, bool may_rehash,
|
||
KeyHasher* key_hasher, AsofJoinNode* node, BackpressureHandler handler,
|
||
const std::shared_ptr<arrow::Schema>& schema,
|
||
const col_index_t time_col_index,
|
||
const std::vector<col_index_t>& key_col_index)
|
||
: sequencer_(util::SerialSequencingQueue::Make(this)),
|
||
queue_(std::move(handler)),
|
||
schema_(schema),
|
||
time_col_index_(time_col_index),
|
||
key_col_index_(key_col_index),
|
||
+ key_col_index_set_(key_col_index.begin(), key_col_index.end()),
|
||
time_type_id_(schema_->fields()[time_col_index_]->type()->id()),
|
||
key_type_id_(key_col_index.size()),
|
||
key_hasher_(key_hasher),
|
||
@@ -537,7 +539,9 @@ class InputState : public util::SerialSequencingQueue::Processor {
|
||
|
||
bool IsTimeOrKeyColumn(col_index_t i) const {
|
||
DCHECK_LT(i, schema_->num_fields());
|
||
- return (i == time_col_index_) || std_has(key_col_index_, i);
|
||
+ // key_col_index_set_ gives O(1) lookup; std_has(key_col_index_, i) was O(K)
|
||
+ // so InitSrcToDstMapping's loop over F fields was O(F×K). Now O(F).
|
||
+ return (i == time_col_index_) || (key_col_index_set_.count(i) > 0);
|
||
}
|
||
|
||
@@ -786,6 +790,8 @@ class InputState : public util::SerialSequencingQueue::Processor {
|
||
std::vector<col_index_t> key_col_index_;
|
||
+ // Shadow set of key_col_index_ for O(1) membership test (IsTimeOrKeyColumn)
|
||
+ std::unordered_set<col_index_t> key_col_index_set_;
|
||
// Type id of the time column
|
||
Type::type time_type_id_;
|
||
|
||
--- a/cpp/src/arrow/acero/asof_join_node.cc
|
||
+++ b/cpp/src/arrow/acero/asof_join_node.cc
|
||
@@ -1287,9 +1293,14 @@ static Result<std::shared_ptr<Schema>> MakeOutputSchema(
|
||
for (int i = 0; i < input_schema[j]->num_fields(); ++i) {
|
||
const auto field = input_schema[j]->field(i);
|
||
bool as_output; // true if the field appears as an output
|
||
if (i == on_field_ix) {
|
||
ARROW_RETURN_NOT_OK(is_valid_on_field(field));
|
||
as_output = (j == 0);
|
||
- } else if (std_has(by_field_ix, i)) {
|
||
+ } else if (by_field_ix_set.count(i) > 0) {
|
||
ARROW_RETURN_NOT_OK(is_valid_by_field(field));
|
||
as_output = (j == 0);
|
||
} else {
|
||
@@ -1255,6 +1258,10 @@ static Result<std::shared_ptr<Schema>> MakeOutputSchema(
|
||
for (size_t j = 0; j < input_schema.size(); ++j) {
|
||
const auto& on_field_ix = indices_of_on_key[j];
|
||
const auto& by_field_ix = indices_of_by_key[j];
|
||
+ // Build O(1) set for by-key membership; std_has(by_field_ix, i) was O(K) per field
|
||
+ // so the inner loop over F fields was O(F×K). Now O(F+K).
|
||
+ const std::unordered_set<col_index_t> by_field_ix_set(by_field_ix.begin(),
|
||
+ by_field_ix.end());
|