Repository: duckdb/duckdb
File: src/execution/operator/scan/physical_positional_scan.cpp (C++)

Code before:
	}

	vector<unique_ptr<GlobalSourceState>> global_states;

	idx_t MaxThreads() override {
		return 1;
	}
};

class PositionalTableScanner {
public:
	PositionalTableScanner(ExecutionContext &context, PhysicalOperator &table_p, GlobalSourceState &gstate_p)
	    : table(table_p), global_state(gstate_p), source_offset(0), exhausted(false) {
		local_state = table.GetLocalSourceState(context, gstate_p);
		source.Initialize(Allocator::Get(context.client), table.types);
	}

	idx_t Refill(ExecutionContext &context) {
		if (source_offset >= source.size()) {
			if (!exhausted) {
				source.Reset();

				InterruptState interrupt_state;
				OperatorSourceInput source_input {global_state, *local_state, interrupt_state};
				auto source_result = table.GetData(context, source, source_input);
				if (source_result == SourceResultType::BLOCKED) {
					throw NotImplementedException(
					    "Unexpected interrupt from table Source in PositionalTableScanner refill");
				}
			}
			source_offset = 0;
		}

		const auto available = source.size() - source_offset;
		if (!available) {
			if (!exhausted) {
				source.Reset();
				for (idx_t i = 0; i < source.ColumnCount(); ++i) {
					auto &vec = source.data[i];
					vec.SetVectorType(VectorType::CONSTANT_VECTOR);
					ConstantVector::SetNull(vec, true);
				}
				exhausted = true;
			}
		}

		return available;
	}

	idx_t CopyData(ExecutionContext &context, DataChunk &output, const idx_t count, const idx_t col_offset) {
		if (!source_offset && (source.size() >= count || exhausted)) {

Review comment (bug):
I'm having trouble figuring out what could result in HAVE_MORE_OUTPUT with a chunk size of 0?
If it has *more* output, logically it should be creating *some* output already, no?
This looks like a bug in the downstream operator, not something we should accept

Code after:
	}

	vector<unique_ptr<GlobalSourceState>> global_states;

	idx_t MaxThreads() override {
		return 1;
	}
};

class PositionalTableScanner {
public:
	PositionalTableScanner(ExecutionContext &context, PhysicalOperator &table_p, GlobalSourceState &gstate_p)
	    : table(table_p), global_state(gstate_p), source_offset(0), exhausted(false) {
		local_state = table.GetLocalSourceState(context, gstate_p);
		source.Initialize(Allocator::Get(context.client), table.types);
	}

	idx_t Refill(ExecutionContext &context) {
		if (source_offset >= source.size()) {
			if (!exhausted) {
				source.Reset();

				InterruptState interrupt_state;
				OperatorSourceInput source_input {global_state, *local_state, interrupt_state};
				auto source_result = SourceResultType::HAVE_MORE_OUTPUT;
				while (source_result == SourceResultType::HAVE_MORE_OUTPUT && source.size() == 0) {
					// TODO: this could as well just be propagated further, but for now iterating it is
					source_result = table.GetData(context, source, source_input);
					if (source_result == SourceResultType::BLOCKED) {
						throw NotImplementedException(
						    "Unexpected interrupt from table Source in PositionalTableScanner refill");
					}
				}
			}
			source_offset = 0;
		}

		const auto available = source.size() - source_offset;
		if (!available) {
			if (!exhausted) {
				source.Reset();
				for (idx_t i = 0; i < source.ColumnCount(); ++i) {
					auto &vec = source.data[i];
					vec.SetVectorType(VectorType::CONSTANT_VECTOR);
					ConstantVector::SetNull(vec, true);
				}
				exhausted = true;
			}
		}

		return available;