colexec: request file descriptors up front in external sorts/joins

[asubiotto]

Release justification: fix for high-severity bug in existing functionality. The
current methodology of requesting file descriptors one at a time could lead to
a potential deadlock when spilling many queries to disk at the same time.

These operators would previously call semaphore.Acquire as needed. This could
lead to a situation where concurrent sorts/joins could block since they could
potentially deadlock Acquiring a file descriptor and never release ones
already held.

Now, sorts and joins that spill to disk will acquire all necessary resources
before spilling to disk.

Release note: None (no release with this potential issue)

Fixes #45602

[cockroach-teamcity]

This change is 

[yuzefovich]

Reviewed 7 of 7 files at r1.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto)

---

pkg/sql/colexec/external_hash_joiner.go, line 266 at r1 (raw file):

		}
	}
	if maxNumberActivePartitions < 4 {

We should probably now have a hard limit on maxNumberActivePartitions dependent on VecMaxOpenFDsLimit (maybe 1/8 of it?).

---

pkg/sql/colexec/external_sort.go, line 196 at r1 (raw file):

	// In order to make progress when merging we have to merge at least two
	// partitions.
	if maxNumberPartitions < 2 {

ditto for maxNumberPartitions.

[asubiotto]

I'll wait for #45825 to go in before doing some more work on this one.

[bobvawter]

All committers have signed the CLA.

[asubiotto]

Rebased and updated to limit the number of file descriptors a sorter requests. I'm slightly confused with our current external hash join partition limits and minimums, which I left as a TODO to discuss.

[yuzefovich]

Reviewed 7 of 7 files at r2.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto)

---

pkg/sql/colexec/external_hash_joiner.go, line 266 at r1 (raw file):

Previously, yuzefovich wrote…

We should probably now have a hard limit on maxNumberActivePartitions dependent on VecMaxOpenFDsLimit (maybe 1/8 of it?).

Do you want to address this suggestion in this PR or in the follow-up?

---

pkg/sql/colexec/external_hash_joiner.go, line 104 at r2 (raw file):

	// fallback to sort and merge join. We'll be using the minimum necessary per
	// input + 4 (2 for each spilling queue that the merge joiner uses).
	// TODO(asubiotto): Is this right?

Yes, I believe the number is correct. I'd probably add a more verbose description of this computation, for example:

During the fallback, at minimum we can have the following:
- 2 FDs will be used to read from the partition of the external hash joiner that could not have been repartitioned, these are inputs to the externalSorters
- 2 * externalSorterMinPartitions FDs will be used by two externalSorters that are inputs into the merge joiner
- 2 FDs will be used by the merge joiner

---

pkg/sql/colexec/external_hash_joiner.go, line 217 at r2 (raw file):

		// needed) before it proceeds to actual join partitions.
		numForcedRepartitions int
		// delegateFDAcquisitions if true, means that a test wants to force the

nit: s/ if true,/, if true,/g.

---

pkg/sql/colexec/external_hash_joiner.go, line 249 at r2 (raw file):

// - delegateFDAcquisitions specifies whether the external hash joiner should
// let the partitioned disk queues acquire file descriptors instead of acquiring
// them up front in Next.

nit: I'd add that this should be true only in tests.

---

pkg/sql/colexec/external_hash_joiner.go, line 311 at r2 (raw file):

	externalSorterMaxNumberPartitions := (maxNumberActivePartitions - 4) / 2
	if externalSorterMaxNumberPartitions < externalSorterMinPartitions {
		externalSorterMaxNumberPartitions = externalSorterMinPartitions

I feel like we should panic in this case because we're upping maxNumberActivePartitions so that there is a least minimum number of FDs needed for each of the external sorters.

---

pkg/sql/colexec/external_hash_joiner.go, line 335 at r2 (raw file):

		execerror.VectorizedInternalPanic(err)
	}
	fmt.Println("max number of active partitions is", maxNumberActivePartitions)

Leftover logging.

---

pkg/sql/colexec/external_hash_joiner.go, line 702 at r2 (raw file):

	}
	if !hj.testingKnobs.delegateFDAcquisitions && hj.fdState.acquiredFDs > 0 {
		hj.fdState.fdSemaphore.Release(hj.fdState.acquiredFDs)

nit: for completeness, we probably should do hj.fdState.acquiredFDs = 0.

---

pkg/sql/colexec/external_hash_joiner_test.go, line 43 at r2 (raw file):

// the fallback to sort + merge join we will be using 6 (external sorter on
// each side will use 2 to read its own partitions and 1 more to write out).
// TODO(asubiotto): It would probably be good to discuss this again.

I think this comment is incorrect. I now believe that we need, in theory, the minimum should be 10. I think we should remove this constant and use externalHJMinPartitions in the tests.

---

pkg/sql/colexec/external_sort.go, line 137 at r2 (raw file):

	testingKnobs struct {
		// delegateFDAcquisitions if true, means that a test wants to force the

nit: s/ if/, if/g.

---

pkg/sql/colexec/external_sort.go, line 163 at r2 (raw file):

// - delegateFDAcquisitions specifies whether the external sorter should let
// the partitioned disk queue acquire file descriptors instead of acquiring
// them up front in Next.

ditto for the "tests only".

---

pkg/sql/colexec/external_sort.go, line 198 at r2 (raw file):

		// external sorter
		// TODO(asubiotto): this number should be tuned.
		fmt.Println("maxNumberPartitions", maxNumberPartitions)

Leftover logging.

---

pkg/sql/colexec/external_sort.go, line 199 at r2 (raw file):

		// TODO(asubiotto): this number should be tuned.
		fmt.Println("maxNumberPartitions", maxNumberPartitions)
		maxPartitionsAllowed := fdSemaphore.GetLimit() / 16

nit: we probably should now extract this computation into a function used by both sorter and joiner.

---

pkg/sql/colexec/external_sort.go, line 365 at r2 (raw file):

	}
	if !s.testingKnobs.delegateFDAcquisitions && s.fdState.fdSemaphore != nil && s.fdState.acquiredFDs > 0 {
		s.fdState.fdSemaphore.Release(s.fdState.acquiredFDs)

ditto to zero out .acquiredFDs.

---

pkg/sql/colexec/external_sort_test.go, line 152 at r2 (raw file):

		memMonitors []*mon.BytesMonitor
	)
	const maxNumberPartitions = 3

nit: these constants should probably be removed in favor of externalSorterMinPartitions.

[asubiotto]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto and @yuzefovich)

---

pkg/sql/colexec/external_hash_joiner.go, line 266 at r1 (raw file):

Previously, yuzefovich wrote…

Do you want to address this suggestion in this PR or in the follow-up?

fdSemaphore.GetLimit()/16 above is what sets the maximum. I think this is fine, no?

---

pkg/sql/colexec/external_hash_joiner.go, line 104 at r2 (raw file):

Previously, yuzefovich wrote…

Yes, I believe the number is correct. I'd probably add a more verbose description of this computation, for example:

During the fallback, at minimum we can have the following:
- 2 FDs will be used to read from the partition of the external hash joiner that could not have been repartitioned, these are inputs to the externalSorters
- 2 * externalSorterMinPartitions FDs will be used by two externalSorters that are inputs into the merge joiner
- 2 FDs will be used by the merge joiner

Done. It seems like the correct number, but I think the merge joiner will use 4 FDs since it uses 2 spilling queues, right?

---

pkg/sql/colexec/external_hash_joiner.go, line 311 at r2 (raw file):

Previously, yuzefovich wrote…

I feel like we should panic in this case because we're upping maxNumberActivePartitions so that there is a least minimum number of FDs needed for each of the external sorters.

Hmm, this doesn't go hand in hand with using the semaphore limit to set a maximum on the number of active partitions, since tests will set this limit low to ensure the hash joiner doesn't exceed the limit. This results in passing in a limit of 0 to the sorter with a nil fdSemaphore. I don't do a nil check there because we should never pass in a limit of 0 to a sorter if we account for its maximum number of partitions. In practice, this if statement only happens in tests, so I'm fine leaving as is if you are.

---

pkg/sql/colexec/external_hash_joiner.go, line 335 at r2 (raw file):

Previously, yuzefovich wrote…

Leftover logging.

Done.

---

pkg/sql/colexec/external_hash_joiner_test.go, line 43 at r2 (raw file):

Previously, yuzefovich wrote…

I think this comment is incorrect. I now believe that we need, in theory, the minimum should be 10. I think we should remove this constant and use externalHJMinPartitions in the tests.

Done.

---

pkg/sql/colexec/external_sort.go, line 196 at r1 (raw file):

Previously, yuzefovich wrote…

ditto for maxNumberPartitions.

Replied there.

---

pkg/sql/colexec/external_sort.go, line 198 at r2 (raw file):

Previously, yuzefovich wrote…

Leftover logging.

Done.

---

pkg/sql/colexec/external_sort.go, line 199 at r2 (raw file):

Previously, yuzefovich wrote…

nit: we probably should now extract this computation into a function used by both sorter and joiner.

I don't think it's worth it. If we tune this, the numbers will probably be different so we'll have to change this anyway.

---

pkg/sql/colexec/external_sort.go, line 365 at r2 (raw file):

Previously, yuzefovich wrote…

ditto to zero out .acquiredFDs.

Done.

[yuzefovich]

Reviewed 4 of 4 files at r3.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto)

---

pkg/sql/colexec/external_hash_joiner.go, line 266 at r1 (raw file):

Previously, asubiotto (Alfonso Subiotto Marqués) wrote…

fdSemaphore.GetLimit()/16 above is what sets the maximum. I think this is fine, no?

Oh yeah, I missed that.

---

pkg/sql/colexec/external_hash_joiner.go, line 104 at r2 (raw file):

Previously, asubiotto (Alfonso Subiotto Marqués) wrote…

Done. It seems like the correct number, but I think the merge joiner will use 4 FDs since it uses 2 spilling queues, right?

Merge joiner follows the pattern of "write all data, then read all data", so each queue should be using 1 FD at any point. (Probably merge joiner should use "close on new partition" strategy.) However, as I said in the comment above, we should not be forgetting about the FD to read from the partition that we are joining (but it is outside of merge joiner's knowledge that we use a FD there).

---

pkg/sql/colexec/external_hash_joiner.go, line 311 at r2 (raw file):

Previously, asubiotto (Alfonso Subiotto Marqués) wrote…

Hmm, this doesn't go hand in hand with using the semaphore limit to set a maximum on the number of active partitions, since tests will set this limit low to ensure the hash joiner doesn't exceed the limit. This results in passing in a limit of 0 to the sorter with a nil fdSemaphore. I don't do a nil check there because we should never pass in a limit of 0 to a sorter if we account for its maximum number of partitions. In practice, this if statement only happens in tests, so I'm fine leaving as is if you are.

SGTM. I'd add a comment here then that this if is in-place to handle artificially low FDs limits set by tests.

---

pkg/sql/colexec/external_hash_joiner.go, line 335 at r2 (raw file):

Previously, asubiotto (Alfonso Subiotto Marqués) wrote…

Done.

I still see it :)

---

pkg/sql/colexec/external_hash_joiner.go, line 109 at r3 (raw file):

	//   sort and partition this input. At this stage 2 + 2 *
	//   externalMinPartitions FDs are used.
	// - Once the inputs (the hash joiner partitions) are finished, both FDs will

I think it is possible that the hash joiner partitions' FDs are not released when the merge joiner kicks in. Consider the following scenario (looking only at one side):

• merge joiner requests the first batch from the external sorter (when it gets the batch, it will be written into the spilling queue) + 1 FD
• external sorter requests batches from the hash joiner partition and does its job. At some point, however, the external sorter needed to do recursive merging, and at that point it was using all of its available FDs. + externalSorterMinPartitions FDs
• the hash joiner partition is being read + 1 FD.

However, while typing out this comment, I realized that the merge joiner will start writing into its spilling queue only when it receives the first batch from the external sorter, and at that point, the hash joiner partition has been fully consumed, and the corresponding FD should have been released. I.e. we actually never use more than 1 + externalSorterMinPartitions number of FDs per side, so externalHJMinPartitions should be 2 + 2 * externalSorterMinPartitions.

---

pkg/sql/colexec/external_sort.go, line 163 at r2 (raw file):

Previously, yuzefovich wrote…

ditto for the "tests only".

Ping.

---

pkg/sql/colexec/external_sort.go, line 198 at r2 (raw file):

Previously, asubiotto (Alfonso Subiotto Marqués) wrote…

Done.

Still see it.

---

pkg/sql/colexec/external_sort.go, line 199 at r2 (raw file):

Previously, asubiotto (Alfonso Subiotto Marqués) wrote…

I don't think it's worth it. If we tune this, the numbers will probably be different so we'll have to change this anyway.

Ok.

---

pkg/sql/colexec/external_sort.go, line 216 at r3 (raw file):

		OneInputNode:       NewOneInputNode(inMemSorter),
		unlimitedAllocator: unlimitedAllocator,
		memoryLimit:        memoryLimit,

nit: external sorter doesn't need to store memoryLimit.

[asubiotto]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto and @yuzefovich)

---

pkg/sql/colexec/external_hash_joiner.go, line 104 at r2 (raw file):

Previously, yuzefovich wrote…

Merge joiner follows the pattern of "write all data, then read all data", so each queue should be using 1 FD at any point. (Probably merge joiner should use "close on new partition" strategy.) However, as I said in the comment above, we should not be forgetting about the FD to read from the partition that we are joining (but it is outside of merge joiner's knowledge that we use a FD there).

Just checking, is this comment void now after the However, while typing out this comment? I responded re the 1FD above, it uses only 1, but actually acquires 2 right now (since there is no enforcing of the fact that all writes happen before all reads in the spilling queue, we should probably fix that)

---

pkg/sql/colexec/external_hash_joiner.go, line 311 at r2 (raw file):

Previously, yuzefovich wrote…

SGTM. I'd add a comment here then that this if is in-place to handle artificially low FDs limits set by tests.

Done.

---

pkg/sql/colexec/external_hash_joiner.go, line 335 at r2 (raw file):

Previously, yuzefovich wrote…

I still see it :)

Oops, thanks for catching

---

pkg/sql/colexec/external_hash_joiner.go, line 109 at r3 (raw file):

Previously, yuzefovich wrote…

I think it is possible that the hash joiner partitions' FDs are not released when the merge joiner kicks in. Consider the following scenario (looking only at one side):

• merge joiner requests the first batch from the external sorter (when it gets the batch, it will be written into the spilling queue) + 1 FD
• external sorter requests batches from the hash joiner partition and does its job. At some point, however, the external sorter needed to do recursive merging, and at that point it was using all of its available FDs. + externalSorterMinPartitions FDs
• the hash joiner partition is being read + 1 FD.

However, while typing out this comment, I realized that the merge joiner will start writing into its spilling queue only when it receives the first batch from the external sorter, and at that point, the hash joiner partition has been fully consumed, and the corresponding FD should have been released. I.e. we actually never use more than 1 + externalSorterMinPartitions number of FDs per side, so externalHJMinPartitions should be 2 + 2 * externalSorterMinPartitions.

Right regarding that the FD is released. But note that each spilling queue, even though it only opens 1 FD, Acquires 2 of them (which is what needs to be reflected here), which is why it's 4 + 2 * externalSorterMinPartitions.

---

pkg/sql/colexec/external_sort.go, line 163 at r2 (raw file):

Previously, yuzefovich wrote…

Ping.

Done.

---

pkg/sql/colexec/external_sort.go, line 198 at r2 (raw file):

Previously, yuzefovich wrote…

Still see it.

I don't think it's there anymore.

[yuzefovich]

Reviewed 2 of 2 files at r4.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto)

---

pkg/sql/colexec/external_hash_joiner.go, line 104 at r2 (raw file):

Previously, asubiotto (Alfonso Subiotto Marqués) wrote…

Just checking, is this comment void now after the However, while typing out this comment? I responded re the 1FD above, it uses only 1, but actually acquires 2 right now (since there is no enforcing of the fact that all writes happen before all reads in the spilling queue, we should probably fix that)

Yeah, resolved.

---

pkg/sql/colexec/external_hash_joiner.go, line 109 at r3 (raw file):

Previously, asubiotto (Alfonso Subiotto Marqués) wrote…

Right regarding that the FD is released. But note that each spilling queue, even though it only opens 1 FD, Acquires 2 of them (which is what needs to be reflected here), which is why it's 4 + 2 * externalSorterMinPartitions.

Oh, I see, I forgot about that. I guess we could introduce non-default mode for spilling queue so that it would acquire only 1 FD. I'll leave it up to you to decide whether it is important to add a TODO for this or we can just ignore this.

---

pkg/sql/colexec/external_sort.go, line 198 at r4 (raw file):

	}
	// Each disk queue will use up to BufferSizeBytes of RAM, so we reduce the
	// memoryLimit of the partitions to sort in memory by those cache sizes.will give

nit: "sizes.will"

---

pkg/sql/colexec/external_sort.go, line 203 at r4 (raw file):

	batchMemSize := estimateBatchSizeBytes(inputTypes, coldata.BatchSize())
	// Reserve a certain amount of memory for the partition caches.
	memoryLimit -= int64((maxNumberPartitions * diskQueueCfg.BufferSizeBytes) + batchMemSize)

At this point this update no longer matters. We need to move the creation of inputPartitioningOperator after this update.

---

pkg/sql/colexec/external_sort.go, line 205 at r4 (raw file):

	memoryLimit -= int64((maxNumberPartitions * diskQueueCfg.BufferSizeBytes) + batchMemSize)
	if memoryLimit < 0 {
		memoryLimit = 0

I think that we need to have memory limit of at least 1 byte (otherwise inputPartitioningOperator will only return zero-length batches, always).

[asubiotto]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @yuzefovich)

---

pkg/sql/colexec/external_hash_joiner.go, line 109 at r3 (raw file):

Previously, yuzefovich wrote…

Oh, I see, I forgot about that. I guess we could introduce non-default mode for spilling queue so that it would acquire only 1 FD. I'll leave it up to you to decide whether it is important to add a TODO for this or we can just ignore this.

It was actually very easy to add (just check for whether the queue is rewindable or not) so added it.

---

pkg/sql/colexec/external_sort.go, line 203 at r4 (raw file):

Previously, yuzefovich wrote…

At this point this update no longer matters. We need to move the creation of inputPartitioningOperator after this update.

Done.

---

pkg/sql/colexec/external_sort.go, line 205 at r4 (raw file):

Previously, yuzefovich wrote…

I think that we need to have memory limit of at least 1 byte (otherwise inputPartitioningOperator will only return zero-length batches, always).

Done. This makes us hit what #46052 fixes a lot more easily.

[yuzefovich]

Reviewed 3 of 3 files at r5.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto)

---

pkg/sql/colexec/spilling_queue.go, line 210 at r5 (raw file):

func (q *spillingQueue) numFDsOpenAtAnyGivenTime() int {
	if q.rewindable {

Merge joiner uses rewindable queue only on the right side, yet it follows the pattern "write all, then read all" on both sides. We probably need a separate knob for this pattern after all. (Or, alternatively, we could be instantiating a rewindable queue in the merge joiner on both sides, but using rewind() method only on the right.)

[asubiotto]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @yuzefovich)

---

pkg/sql/colexec/spilling_queue.go, line 210 at r5 (raw file):

Previously, yuzefovich wrote…

Merge joiner uses rewindable queue only on the right side, yet it follows the pattern "write all, then read all" on both sides. We probably need a separate knob for this pattern after all. (Or, alternatively, we could be instantiating a rewindable queue in the merge joiner on both sides, but using rewind() method only on the right.)

The check is easy enough. If the underlying disk queue is not using a default cache mode then this pattern is enforced, so we can use that.

[yuzefovich]

Reviewed 1 of 1 files at r6.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @asubiotto)

---

pkg/sql/colexec/external_hash_joiner.go, line 320 at r6 (raw file):

	// the partitions that we need to join using sort + merge join strategy, and
	// all others are divided between the two inputs.
	externalSorterMaxNumberPartitions := (maxNumberActivePartitions - 4) / 2

s/4/2/g.

Maybe we should define a separate constant for this 2 so that we don't forget to update one of the places where this constant is used if the calculation changes for some reason?

---

pkg/sql/colexec/external_sort_test.go, line 89 at r6 (raw file):

						orderedVerifier,
						func(input []Operator) (Operator, error) {
							// A sorter should never exceed maxNumberPartitions+1, even during

nit: the comment needs updating.

---

pkg/sql/colexec/spilling_queue.go, line 210 at r5 (raw file):

Previously, asubiotto (Alfonso Subiotto Marqués) wrote…

The check is easy enough. If the underlying disk queue is not using a default cache mode then this pattern is enforced, so we can use that.

SGTM.

[asubiotto]

Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale) (waiting on @asubiotto and @yuzefovich)

---

pkg/sql/colexec/external_hash_joiner.go, line 320 at r6 (raw file):

Previously, yuzefovich wrote…

s/4/2/g.

Maybe we should define a separate constant for this 2 so that we don't forget to update one of the places where this constant is used if the calculation changes for some reason?

Done.

[asubiotto]

bors r=yuzefovich

[craig]

Build failed

• GitHub CI (Cockroach)

[asubiotto]

A release justification was needed in the PR description as well. Updated.

bors r=yuzefovich

[craig]

Build succeeded

• GitHub CI (Cockroach)