#pragma once #include #include #include #include #include #include #include #include #include #include #include namespace ProfileEvents { extern const Event SleepFunctionCalls; extern const Event SleepFunctionMicroseconds; } namespace DB { namespace ErrorCodes { extern const int ILLEGAL_TYPE_OF_ARGUMENT; extern const int TOO_SLOW; extern const int ILLEGAL_COLUMN; extern const int BAD_ARGUMENTS; } /** sleep(seconds) - the specified number of seconds sleeps each columns. */ enum class FunctionSleepVariant { PerBlock, PerRow }; template class FunctionSleep : public IFunction { private: UInt64 max_microseconds; public: static constexpr auto name = variant == FunctionSleepVariant::PerBlock ? "sleep" : "sleepEachRow"; static FunctionPtr create(ContextPtr context) { return std::make_shared>(context->getSettingsRef().function_sleep_max_microseconds_per_block); } FunctionSleep(UInt64 max_microseconds_) : max_microseconds(max_microseconds_) { } /// Get the name of the function. String getName() const override { return name; } /// Do not sleep during query analysis. bool isSuitableForConstantFolding() const override { return false; } size_t getNumberOfArguments() const override { return 1; } bool isSuitableForShortCircuitArgumentsExecution(const DataTypesWithConstInfo & /*arguments*/) const override { return true; } DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override { WhichDataType which(arguments[0]); if (!which.isFloat() && !which.isNativeUInt()) throw Exception(ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT, "Illegal type {} of argument of function {}, expected Float64", arguments[0]->getName(), getName()); return std::make_shared(); } ColumnPtr executeImplDryRun(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, size_t /*input_rows_count*/) const override { return execute(arguments, result_type, true); } ColumnPtr executeImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, size_t /*input_rows_count*/) const override { return execute(arguments, result_type, false); } ColumnPtr execute(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, bool dry_run) const { const IColumn * col = arguments[0].column.get(); if (!isColumnConst(*col)) throw Exception(ErrorCodes::ILLEGAL_COLUMN, "The argument of function {} must be constant.", getName()); Float64 seconds = applyVisitor(FieldVisitorConvertToNumber(), assert_cast(*col).getField()); if (seconds < 0 || !std::isfinite(seconds)) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot sleep infinite or negative amount of time (not implemented)"); size_t size = col->size(); /// We do not sleep if the columns is empty. if (size > 0) { /// When sleeping, the query cannot be cancelled. For ability to cancel query, we limit sleep time. if (max_microseconds && seconds * 1e6 > max_microseconds) throw Exception(ErrorCodes::TOO_SLOW, "The maximum sleep time is {} microseconds. Requested: {}", max_microseconds, seconds); if (!dry_run) { UInt64 count = (variant == FunctionSleepVariant::PerBlock ? 1 : size); UInt64 microseconds = static_cast(seconds * count * 1e6); if (max_microseconds && microseconds > max_microseconds) throw Exception(ErrorCodes::TOO_SLOW, "The maximum sleep time is {} microseconds. Requested: {} microseconds per block (of size {})", max_microseconds, microseconds, size); sleepForMicroseconds(microseconds); ProfileEvents::increment(ProfileEvents::SleepFunctionCalls, count); ProfileEvents::increment(ProfileEvents::SleepFunctionMicroseconds, microseconds); } } /// convertToFullColumn needed, because otherwise (constant expression case) function will not get called on each columns. return result_type->createColumnConst(size, 0u)->convertToFullColumnIfConst(); } }; }