Repository: duckdb/duckdb
File: src/execution/physical_operator.cpp (C++)

Code before:
		if (caching_supported) {
			state.can_cache_chunk = PhysicalOperator::SelectOperatorCachingMode(context);
		} else {
			state.can_cache_chunk = OperatorCachingMode::NONE;
		}
	}
	if (state.can_cache_chunk == OperatorCachingMode::NONE) {
		return child_result;
	}
	if (chunk.size() == 0) {
		// Nothing required to be done, this also means that BLOCKED is properly passed through

		// optimization: when FINISHED, empty cached_chunk already
		if (state.cached_chunk && child_result == OperatorResultType::FINISHED) {
			chunk.Move(*state.cached_chunk);
			state.cached_chunk.reset(); // Now remove the unique_ptr<DataChunk> already
		}
	} else if (chunk.size() < CACHE_THRESHOLD &&
	           (!state.cached_chunk || (state.cached_chunk_is_flat && state.cached_chunk->size() + chunk.size() <= STANDARD_VECTOR_SIZE))) {
		// we have filtered out a significant amount of tuples AND we have enough space in the cached chunk
		// add this chunk to the cache and continue

		if (!state.cached_chunk) {
			state.cached_chunk = make_uniq<DataChunk>();
			state.cached_chunk->Initialize(Allocator::Get(context.client), chunk.GetTypes());
			state.cached_chunk_is_flat = true;
		}

		state.cached_chunk->Append(chunk);

		if (state.cached_chunk->size() >= (STANDARD_VECTOR_SIZE - CACHE_THRESHOLD) ||
		    child_result == OperatorResultType::FINISHED) {
			// chunk cache full: return it
			chunk.Move(*state.cached_chunk);
			state.cached_chunk->Initialize(Allocator::Get(context.client), chunk.GetTypes());
			state.cached_chunk_is_flat = true;
			return child_result;
		} else {
			// chunk cache not full return empty result
			chunk.Reset();
		}

	} else if (state.cached_chunk &&
	           (state.can_cache_chunk == OperatorCachingMode::ORDERED || chunk.size() > state.cached_chunk->size() || (!state.cached_chunk_is_flat && chunk.size() < CACHE_THRESHOLD))) {
		// Either we are in ORDERED mode (so cached_chunk should be returned first)
		// or chunk is bigger than cached_chunk (so swapping them is beneficial)

		// In the case of ORDERED, this is needed to maintain order
		// in the other case, it's just an optimization, but can as well reuse the same codepath
		auto tmp = make_uniq<DataChunk>();
		tmp->Initialize(Allocator::Get(context.client), chunk.GetTypes());

Review comment (performance):
I think we the `Move` is in general not a great idea - we shouldn't be randomly appending to the input chunk imo, that seems potentially problematic. 

What would be better is if we, instead of moving the input chunk to the `cached_chunk`, would just return `HAVE_MORE_OUTPUT` (which would prevent a new input chunk from being requested), i.e.:

```sql
if (type == ORDERED && chunk.size() <= CACHE_THRESHOLD) {
   if (!cached_chunk.empty()) {
      // first return whatever we had cached, if any, and clear the cache
      result_chunk.Move(cached_chunk);
      cached_chunk.reset();
      return HAVE_MORE_OUTPUT;
   }
   // finished processing this chunk
   result_chunk.Reference(chunk);
   return NEED_MORE_INPUT;   
}
```

Code after:
                                                              const OperatorResultType child_result,
                                                              CachingOperatorState &state,
                                                              ClientContext &client_context) {
	if (state.can_cache_chunk == OperatorCachingMode::NONE) {
		return CachingPhysicalOperatorExecuteMode::RETURN_CHUNK;
	}
	const bool needs_continuation_chunk = (state.can_cache_chunk == OperatorCachingMode::PARTITIONED &&
	                                       child_result != OperatorResultType::HAVE_MORE_OUTPUT) ||
	                                      (child_result == OperatorResultType::FINISHED);
	const bool has_non_empty_cached_chunk = state.cached_chunk && state.cached_chunk->size() > 0;
	const bool has_space_for_chunk_in_cache =
	    !state.cached_chunk || (state.cached_chunk->size() + chunk.size() <= STANDARD_VECTOR_SIZE);

	if (has_non_empty_cached_chunk && needs_continuation_chunk) {
		if (chunk.size() == 0) {
			if (child_result == OperatorResultType::BLOCKED) {
				// First return cached, then empty chunk via continuation that will BLOCK
				return CachingPhysicalOperatorExecuteMode::RETURN_CACHED_THEN_CHUNK_VIA_CONTINUATION;
			}

			// Return cached, and the current result
			return CachingPhysicalOperatorExecuteMode::RETURN_CACHED;
		}
		if (chunk.size() <= CachingPhysicalOperator::CACHE_THRESHOLD && has_space_for_chunk_in_cache) {
			// chunk is small, both fit
			return CachingPhysicalOperatorExecuteMode::RETURN_CACHED_PLUS_CHUNK;
		}

		// First return cached, then chunk via continuation
		return CachingPhysicalOperatorExecuteMode::RETURN_CACHED_THEN_CHUNK_VIA_CONTINUATION;
	} else if (chunk.size() == 0) {
		// Nothing required to be done, this also means that BLOCKED is properly passed through
		// Note that this case works also for unordered cases, given no rows are there

		return CachingPhysicalOperatorExecuteMode::RETURN_CHUNK;
	} else if (chunk.size() <= CachingPhysicalOperator::CACHE_THRESHOLD && !needs_continuation_chunk) {
		// We have filtered out a significant amount of tuples

		if (!state.cached_chunk) {
			// Initialize cached_chunk
			state.cached_chunk = make_uniq<DataChunk>();
			state.cached_chunk->Initialize(Allocator::Get(client_context), chunk.GetTypes());
		}

		if (has_space_for_chunk_in_cache) {
			// We can just append, do and return empty chunk
			return CachingPhysicalOperatorExecuteMode::APPEND_CHUNK;
		}

		// Return what is now cached, and append chunk (via tmp)
		return CachingPhysicalOperatorExecuteMode::RETURN_CACHED_APPEND_CHUNK;