[C++][Acero] Incorrect results in inner join

[llama90]

Describe the bug, including details regarding any error messages, version, and platform.

Before explaining, if recreating previously submitted issues like this is considered a mistake, I apologize.

• Inner joins are incorrect #37729

Overview

In the previous issue, the user is discussing the occurrence of incorrect results when performing an inner join.

Although it was well-explained in the previous issue, to reiterate, the user has created this issue because they are getting incorrect results when performing an inner join between two tables, namely table_1 and table_2.

Each of these tables (table_1 and table_2) has columns col_1, col_2, col_3, and table_2 has an additional col_4. Upon my investigation, it appears that table_2.parquet has 7 more records, but for col_1, col_2, and col_3, it contains the same values as table_1.

The number of records in each table is 6282 and 6289, respectively.

So, when performing an inner join using col_1, col_2, and col_3 as the join keys, the result should be 6282, regardless of the order of the tables.

Reason

To start with the cause, there is an issue with the BloomFilter logic.

When testing in C++, if you set the BloomFilter option (disable_bloom_filter) to true, the join operation is performed without any issues.

// with bloom_filter (default)
HashJoinNodeOptions join_opts{
    JoinType::INNER, 
    {"col_1", "col_2", "col_3"}, 
    {"col_1", "col_2", "col_3"}, 
    literal(true), 
    "_l", 
    "_r"
};

// without bloom_filter
HashJoinNodeOptions join_opts{
    JoinType::INNER, 
    {"col_1", "col_2", "col_3"}, 
    {"col_1", "col_2", "col_3"}, 
    literal(true), 
    "_l", 
    "_r",
    true // disable_bloom_filter
};

Additionally, the findings from further investigation are as follows.

1. The number of matching records between the left and right tables must exceed 1024.
2. When the number of matching records is close to the number of records in the right table, errors occur.
3. The number of matching records does not appear to have any correlation with the number of records in the left table.

Up to this point, this is what I have gathered about the issue, and I am working hard to fix the bug.

However, I am a novice about Arrow, which is causing it to take longer than expected. Nonetheless, I will continue to make efforts to resolve the bug.

Any advice or insights from those who are more experienced would be greatly appreciated, and it would also be great if someone with expertise could tackle this issue first. So, I'm sharing this here.

Component(s)

C++

[ianmcook]

@llama90 thank you for investigating this and working to solve it.

There was a recent conversation on the Arrow developer mailing list that might be relevant (but I'm not sure):
https://lists.apache.org/thread/pq10zd4sr494hd09t93o3hb6zswn5wll
When batch sizes are too large, silent overflows can occur, producing incorrect hash join results.

Could this explain the incorrect results you are seeing?

[llama90]

@ianmcook Hello.

It seems like a different issue from a Swiss join. Even in the smallest case, when both table_1 and table_2 have 1025 identical records each, the result still shows 1024 join results. When I change the number of right records to 1537, I finally could get 1025 values.

Based on the information you mentioned, I became interested and conducted a test by increasing the number of records to 100,000.

• Left: 100,000
• Right: 100,000
• Match: 100,000
  • without BloomFilter: 100,000
  • with BloomFilter: 84,334

Furthermore, I tested increasing the record count to 10 million without BloomFilter, and I confirmed that the results were accurate in this scenario.

In this case, I was able to confirm that the results were coming out in batches of 32,768 each. Maybe it is a morsel unit.

• Left: 32,768
• Right: 32,768
• Match: 32,768
  • without BloomFilter: 32,768
  • with BloomFilter: 17,010

In my opinion, it seems clear that there might be an issue with the Bloom Filter. Cases where accurate results were obtained, whether using BloomFilter or not, were as follows:

• When the number of match records is 1024 or less, regardless of the number of records in the Left and Right tables.
• When the number of match records exceeds 1024, but the number of records in the Right table is greater than the number of match records.
  • For example, R - 5000, M - 5000, L - 10000 (In this case, reducing it to 9000 records results in 4904 as the output).

Below is an example of the tables.

table_1 (left)

• schema ({timestamp(TimeUnit::MICRO, "UTC"), large_utf8(), large_utf8()})
  • col_1: All values are 2023-09-14 21:43:18.678917
  • col_2: 1, 2, 3, 4, 5 ... N, -1, -1, -1, -1, -1, ... -1
  • col_3: All values are foo
• N is the number of matching records between table 1 and table 2

table_2 (right)

• schema ({timestamp(TimeUnit::MICRO, "UTC"), large_utf8(), large_utf8(), large_utf8()})
  • col_1: All values are 2023-09-14 21:43:18.678917
  • col_2: 1, 2, 3, 4, 5 ... M
  • col_3: All values are foo
  • col_4: All values are bar
• M is the number of records for table 2

Example

• The number of records for table 1: 5
• N: 3
• M: 4

table 1

col_1	col_2	col_3
2023-09-14 21:43:18.678917	1	foo
2023-09-14 21:43:18.678917	2	foo
2023-09-14 21:43:18.678917	3	foo
2023-09-14 21:43:18.678917	-1	foo
2023-09-14 21:43:18.678917	-1	foo

table 2

col_1	col_2	col_3	col_4
2023-09-14 21:43:18.678917	1	foo	bar
2023-09-14 21:43:18.678917	2	foo	bar
2023-09-14 21:43:18.678917	3	foo	bar
2023-09-14 21:43:18.678917	4	foo	bar

[benibus]

I haven't managed to find the exact issue yet (I'm not too familiar with this code in particular), but this section is fairly suspicious:

arrow/cpp/src/arrow/acero/hash_join_node.cc

Lines 1131 to 1146 in 3697bcd

	arrow::util::TempVectorHolder<uint32_t> hash_holder(
	stack, arrow::util::MiniBatch::kMiniBatchLength);
	uint32_t* hashes = hash_holder.mutable_data();
	for (int64_t i = 0; i < key_batch.length;
	i += arrow::util::MiniBatch::kMiniBatchLength) {
	int64_t length =
	std::min(static_cast<int64_t>(key_batch.length - i),
	static_cast<int64_t>(arrow::util::MiniBatch::kMiniBatchLength));
	std::vector<KeyColumnArray> temp_column_arrays;
	RETURN_NOT_OK(Hashing32::HashBatch(key_batch, hashes, temp_column_arrays,
	ctx_->cpu_info()->hardware_flags(), stack, i,
	length));
	RETURN_NOT_OK(build_.builder_->PushNextBatch(thread_index, length, hashes));
	}
	return Status::OK();

Note that this function is exclusive to the bloom filter path and the the value of kMiniBatchLength happens to be 1024...

[llama90]

It seems correct.

I've noticed that when modifying the kMiniBatchLength value arbitrarily to be larger than the number of match records, the results come out fine.

I suspect there might be an issue with handling the remainder when it exceeds the mini-batch size, so I'm looking into that part.

Thank you for checking!

[llama90]

It seems that the issue has been fixed.

I will clean up the code, write unit tests, and aim to submit a PR as soon as possible. Thanks to your review, I was able to reproduce the symptoms on a smaller scale for testing. I appreciate your review once again. @benibus