text: #include "duckdb/execution/operator/helper/physical_streaming_sample.hpp"
  #include "duckdb/common/random_engine.hpp"
  #include "duckdb/common/to_string.hpp"
  
  namespace duckdb {
  
  PhysicalStreamingSample::PhysicalStreamingSample(vector<LogicalType> types, SampleMethod method, double percentage,
                                                   int64_t seed, idx_t estimated_cardinality)
      : PhysicalOperator(PhysicalOperatorType::STREAMING_SAMPLE, move(types), estimated_cardinality), method(method),
        percentage(percentage / 100), seed(seed) {
  }
  
  //===--------------------------------------------------------------------===//
  // Operator
  //===--------------------------------------------------------------------===//
  class StreamingSampleOperatorState : public OperatorState {
  public:
  	explicit StreamingSampleOperatorState(int64_t seed) : random(seed) {
  	}
  
  	RandomEngine random;
  };
  
  void PhysicalStreamingSample::SystemSample(DataChunk &input, DataChunk &result, OperatorState &state_p) const {
  	// system sampling: we throw one dice per chunk
  	auto &state = (StreamingSampleOperatorState &)state_p;
  	double rand = state.random.NextRandom();
  	if (rand <= percentage) {
  		// rand is smaller than sample_size: output chunk
  		result.Reference(input);
  	}
  }
  
  void PhysicalStreamingSample::BernoulliSample(DataChunk &input, DataChunk &result, OperatorState &state_p) const {
  	// bernoulli sampling: we throw one dice per tuple
  	// then slice the result chunk
  	auto &state = (StreamingSampleOperatorState &)state_p;
  	idx_t result_count = 0;
  	SelectionVector sel(STANDARD_VECTOR_SIZE);
  	for (idx_t i = 0; i < input.size(); i++) {
  		double rand = state.random.NextRandom();
  		if (rand <= percentage) {
  			sel.set_index(result_count++, i);
  		}
  	}
  	if (result_count > 0) {
  		result.Slice(input, sel, result_count);
  	}
  }
  
  unique_ptr<OperatorState> PhysicalStreamingSample::GetOperatorState(ClientContext &context) const {
  	return make_unique<StreamingSampleOperatorState>(seed);
  }
  
  OperatorResultType PhysicalStreamingSample::Execute(ExecutionContext &context, DataChunk &input, DataChunk &chunk,
                                                      GlobalOperatorState &gstate, OperatorState &state) const {
  	switch (method) {
  	case SampleMethod::BERNOULLI_SAMPLE:
  		BernoulliSample(input, chunk, state);
  		break;
  	case SampleMethod::SYSTEM_SAMPLE:
  		SystemSample(input, chunk, state);
  		break;
  	default:
  		throw InternalException("Unsupported sample method for streaming sample");
  	}
  	return OperatorResultType::NEED_MORE_INPUT;
  }
  
  string PhysicalStreamingSample::ParamsToString() const {
  	return SampleMethodToString(method) + ": " + to_string(100 * percentage) + "%";
  }
  
  } // namespace duckdb
  