1 | #include <unistd.h> |
2 | #include <Functions/IFunctionImpl.h> |
3 | #include <Functions/FunctionHelpers.h> |
4 | #include <Columns/ColumnConst.h> |
5 | #include <DataTypes/DataTypesNumber.h> |
6 | #include <Common/FieldVisitors.h> |
7 | #include <Common/assert_cast.h> |
8 | #include <common/sleep.h> |
9 | #include <IO/WriteHelpers.h> |
10 | |
11 | |
12 | namespace DB |
13 | { |
14 | |
15 | namespace ErrorCodes |
16 | { |
17 | extern const int TOO_SLOW; |
18 | extern const int ILLEGAL_COLUMN; |
19 | extern const int BAD_ARGUMENTS; |
20 | } |
21 | |
22 | /** sleep(seconds) - the specified number of seconds sleeps each block. |
23 | */ |
24 | |
25 | enum class FunctionSleepVariant |
26 | { |
27 | PerBlock, |
28 | PerRow |
29 | }; |
30 | |
31 | template <FunctionSleepVariant variant> |
32 | class FunctionSleep : public IFunction |
33 | { |
34 | public: |
35 | static constexpr auto name = variant == FunctionSleepVariant::PerBlock ? "sleep" : "sleepEachRow" ; |
36 | static FunctionPtr create(const Context &) |
37 | { |
38 | return std::make_shared<FunctionSleep<variant>>(); |
39 | } |
40 | |
41 | /// Get the name of the function. |
42 | String getName() const override |
43 | { |
44 | return name; |
45 | } |
46 | |
47 | /// Do not sleep during query analysis. |
48 | bool isSuitableForConstantFolding() const override |
49 | { |
50 | return false; |
51 | } |
52 | |
53 | size_t getNumberOfArguments() const override |
54 | { |
55 | return 1; |
56 | } |
57 | |
58 | DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override |
59 | { |
60 | WhichDataType which(arguments[0]); |
61 | |
62 | if (!which.isFloat() |
63 | && !which.isNativeUInt()) |
64 | throw Exception("Illegal type " + arguments[0]->getName() + " of argument of function " + getName() + ", expected Float64" , |
65 | ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT); |
66 | |
67 | return std::make_shared<DataTypeUInt8>(); |
68 | } |
69 | |
70 | void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result, size_t /*input_rows_count*/) override |
71 | { |
72 | const IColumn * col = block.getByPosition(arguments[0]).column.get(); |
73 | |
74 | if (!isColumnConst(*col)) |
75 | throw Exception("The argument of function " + getName() + " must be constant." , ErrorCodes::ILLEGAL_COLUMN); |
76 | |
77 | Float64 seconds = applyVisitor(FieldVisitorConvertToNumber<Float64>(), assert_cast<const ColumnConst &>(*col).getField()); |
78 | |
79 | if (seconds < 0) |
80 | throw Exception("Cannot sleep negative amount of time (not implemented)" , ErrorCodes::BAD_ARGUMENTS); |
81 | |
82 | size_t size = col->size(); |
83 | |
84 | /// We do not sleep if the block is empty. |
85 | if (size > 0) |
86 | { |
87 | /// When sleeping, the query cannot be cancelled. For abitily to cancel query, we limit sleep time. |
88 | if (seconds > 3.0) /// The choice is arbitrary |
89 | throw Exception("The maximum sleep time is 3 seconds. Requested: " + toString(seconds), ErrorCodes::TOO_SLOW); |
90 | |
91 | UInt64 microseconds = seconds * (variant == FunctionSleepVariant::PerBlock ? 1 : size) * 1e6; |
92 | sleepForMicroseconds(microseconds); |
93 | } |
94 | |
95 | /// convertToFullColumn needed, because otherwise (constant expression case) function will not get called on each block. |
96 | block.getByPosition(result).column = block.getByPosition(result).type->createColumnConst(size, 0u)->convertToFullColumnIfConst(); |
97 | } |
98 | }; |
99 | |
100 | } |
101 | |