thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-24 16:42:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

First version

Browse Source
This commit is contained in:
kssenii 2021-05-30 13:57:30 +00:00
parent a6ef5778cc
commit d18609467b
61 changed files with 475 additions and 248 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/AggregateFunctions/AggregateFunctionAggThrow.cpp
View File

@ -84,13 +84,13 @@ public:
{
}
void serialize(ConstAggregateDataPtr, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
char c = 0;
buf.write(c);
}
void deserialize(AggregateDataPtr, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr /* place */, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
char c = 0;
buf.read(c);
@ -119,4 +119,3 @@ void registerAggregateFunctionAggThrow(AggregateFunctionFactory & factory)
}
}

4
src/AggregateFunctions/AggregateFunctionArgMinMax.h
View File

@ -79,13 +79,13 @@ public:
this->data(place).result.change(this->data(rhs).result, arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).result.write(buf, *serialization_res);
this->data(place).value.write(buf, *serialization_val);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
this->data(place).result.read(buf, *serialization_res, arena);
this->data(place).value.read(buf, *serialization_val, arena);

18
src/AggregateFunctions/AggregateFunctionArray.h
View File

@ -48,6 +48,16 @@ public:
return nested_func->getReturnType();
}
bool isVersioned() const override
{
return nested_func->isVersioned();
}
size_t getDefaultVersion() const override
{
return nested_func->getDefaultVersion();
}
void create(AggregateDataPtr __restrict place) const override
{
nested_func->create(place);
@ -110,14 +120,14 @@ public:
nested_func->merge(place, rhs, arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> version) const override
{
nested_func->serialize(place, buf);
nested_func->serialize(place, buf, version);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> version, Arena * arena) const override
{
nested_func->deserialize(place, buf, arena);
nested_func->deserialize(place, buf, version, arena);
}
void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena * arena) const override

4
src/AggregateFunctions/AggregateFunctionAvg.h
View File

@ -107,7 +107,7 @@ public:
this->data(place).denominator += this->data(rhs).denominator;
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
writeBinary(this->data(place).numerator, buf);
@ -117,7 +117,7 @@ public:
writeBinary(this->data(place).denominator, buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
readBinary(this->data(place).numerator, buf);

4
src/AggregateFunctions/AggregateFunctionBitwise.h
View File

@ -67,12 +67,12 @@ public:
this->data(place).update(this->data(rhs).value);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
writeBinary(this->data(place).value, buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
readBinary(this->data(place).value, buf);
}

4
src/AggregateFunctions/AggregateFunctionBoundingRatio.h
View File

@ -142,12 +142,12 @@ public:
data(place).merge(data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
data(place).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
data(place).deserialize(buf);
}

29
src/AggregateFunctions/AggregateFunctionCategoricalInformationValue.h
View File

@ -61,12 +61,7 @@ public:
return alignof(T);
}
void add(
AggregateDataPtr place,
const IColumn ** columns,
size_t row_num,
Arena *
) const override
void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena *) const override
{
auto y_col = static_cast<const ColumnUInt8 *>(columns[category_count]);
bool y = y_col->getData()[row_num];
@ -83,11 +78,7 @@ public:
reinterpret_cast<T *>(place)[category_count * 2 + size_t(y)] += 1;
}
void merge(
AggregateDataPtr place,
ConstAggregateDataPtr rhs,
Arena *
) const override
void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena *) const override
{
for (size_t i : ext::range(0, category_count + 1))
{
@ -96,19 +87,12 @@ public:
}
}
void serialize(
ConstAggregateDataPtr place,
WriteBuffer & buf
) const override
void serialize(ConstAggregateDataPtr place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
buf.write(place, sizeOfData());
}
void deserialize(
AggregateDataPtr place,
ReadBuffer & buf,
Arena *
) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
buf.read(place, sizeOfData());
}
@ -120,10 +104,7 @@ public:
);
}
void insertResultInto(
AggregateDataPtr place,
IColumn & to,
Arena *) const override
void insertResultInto(AggregateDataPtr place, IColumn & to, Arena *) const override
{
auto & col = static_cast<ColumnArray &>(to);
auto & data_col = static_cast<ColumnFloat64 &>(col.getData());

8
src/AggregateFunctions/AggregateFunctionCount.h
View File

@ -84,12 +84,12 @@ public:
data(place).count += data(rhs).count;
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
writeVarUInt(data(place).count, buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
readVarUInt(data(place).count, buf);
}
@ -141,12 +141,12 @@ public:
data(place).count += data(rhs).count;
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
writeVarUInt(data(place).count, buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
readVarUInt(data(place).count, buf);
}

4
src/AggregateFunctions/AggregateFunctionDeltaSum.h
View File

@ -102,7 +102,7 @@ public:
// Otherwise lhs either has data or is uninitialized, so we don't need to modify its values.
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
writeIntBinary(this->data(place).sum, buf);
writeIntBinary(this->data(place).first, buf);
@ -110,7 +110,7 @@ public:
writePODBinary<bool>(this->data(place).seen, buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
readIntBinary(this->data(place).sum, buf);
readIntBinary(this->data(place).first, buf);

4
src/AggregateFunctions/AggregateFunctionDeltaSumTimestamp.h
View File

@ -143,7 +143,7 @@ public:
}
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
writeIntBinary(this->data(place).sum, buf);
writeIntBinary(this->data(place).first, buf);
@ -153,7 +153,7 @@ public:
writePODBinary<bool>(this->data(place).seen, buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
readIntBinary(this->data(place).sum, buf);
readIntBinary(this->data(place).first, buf);

4
src/AggregateFunctions/AggregateFunctionDistinct.h
View File

@ -182,12 +182,12 @@ public:
this->data(place).merge(this->data(rhs), arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
this->data(place).deserialize(buf, arena);
}

4
src/AggregateFunctions/AggregateFunctionEntropy.h
View File

@ -125,12 +125,12 @@ public:
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(const_cast<AggregateDataPtr>(place)).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).deserialize(buf);
}

16
src/AggregateFunctions/AggregateFunctionForEach.h
View File

@ -129,6 +129,16 @@ public:
return std::make_shared<DataTypeArray>(nested_func->getReturnType());
}
bool isVersioned() const override
{
return nested_func->isVersioned();
}
size_t getDefaultVersion() const override
{
return nested_func->getDefaultVersion();
}
void destroy(AggregateDataPtr __restrict place) const noexcept override
{
AggregateFunctionForEachData & state = data(place);
@ -196,7 +206,7 @@ public:
}
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
const AggregateFunctionForEachData & state = data(place);
writeBinary(state.dynamic_array_size, buf);
@ -209,7 +219,7 @@ public:
}
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> version, Arena * arena) const override
{
AggregateFunctionForEachData & state = data(place);
@ -221,7 +231,7 @@ public:
char * nested_state = state.array_of_aggregate_datas;
for (size_t i = 0; i < new_size; ++i)
{
nested_func->deserialize(nested_state, buf, arena);
nested_func->deserialize(nested_state, buf, version, arena);
nested_state += nested_size_of_data;
}
}

8
src/AggregateFunctions/AggregateFunctionGroupArray.h
View File

@ -237,7 +237,7 @@ public:
// if constexpr (Trait::sampler == Sampler::DETERMINATOR)
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
const auto & value = this->data(place).value;
size_t size = value.size();
@ -256,7 +256,7 @@ public:
// if constexpr (Trait::sampler == Sampler::DETERMINATOR)
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
size_t size = 0;
readVarUInt(size, buf);
@ -550,7 +550,7 @@ public:
}
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
writeVarUInt(data(place).value.size(), buf);
@ -570,7 +570,7 @@ public:
// if constexpr (Trait::sampler == Sampler::DETERMINATOR)
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
UInt64 elems;
readVarUInt(elems, buf);

4
src/AggregateFunctions/AggregateFunctionGroupArrayInsertAt.h
View File

@ -144,7 +144,7 @@ public:
arr_lhs[i] = arr_rhs[i];
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
const Array & arr = data(place).value;
size_t size = arr.size();
@ -164,7 +164,7 @@ public:
}
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
size_t size = 0;
readVarUInt(size, buf);

4
src/AggregateFunctions/AggregateFunctionGroupArrayMoving.h
View File

@ -139,7 +139,7 @@ public:
cur_elems.sum += rhs_elems.sum;
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
const auto & value = this->data(place).value;
size_t size = value.size();
@ -147,7 +147,7 @@ public:
buf.write(reinterpret_cast<const char *>(value.data()), size * sizeof(value[0]));
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
size_t size = 0;
readVarUInt(size, buf);

8
src/AggregateFunctions/AggregateFunctionGroupBitmap.h
View File

@ -35,9 +35,9 @@ public:
this->data(place).rbs.merge(this->data(rhs).rbs);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override { this->data(place).rbs.write(buf); }
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override { this->data(place).rbs.write(buf); }
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override { this->data(place).rbs.read(buf); }
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override { this->data(place).rbs.read(buf); }
void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override
{
@ -95,9 +95,9 @@ public:
}
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override { this->data(place).rbs.write(buf); }
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override { this->data(place).rbs.write(buf); }
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override { this->data(place).rbs.read(buf); }
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override { this->data(place).rbs.read(buf); }
void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override
{

8
src/AggregateFunctions/AggregateFunctionGroupUniqArray.h
View File

@ -87,7 +87,7 @@ public:
}
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
auto & set = this->data(place).value;
size_t size = set.size();
@ -96,7 +96,7 @@ public:
writeIntBinary(elem, buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).value.read(buf);
}
@ -169,7 +169,7 @@ public:
return true;
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
auto & set = this->data(place).value;
writeVarUInt(set.size(), buf);
@ -180,7 +180,7 @@ public:
}
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
auto & set = this->data(place).value;
size_t size;

4
src/AggregateFunctions/AggregateFunctionHistogram.h
View File

@ -346,12 +346,12 @@ public:
this->data(place).merge(this->data(rhs), max_bins);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).read(buf, max_bins);
}

18
src/AggregateFunctions/AggregateFunctionIf.h
View File

@ -50,6 +50,16 @@ public:
return nested_func->getReturnType();
}
bool isVersioned() const override
{
return nested_func->isVersioned();
}
size_t getDefaultVersion() const override
{
return nested_func->getDefaultVersion();
}
void create(AggregateDataPtr __restrict place) const override
{
nested_func->create(place);
@ -124,14 +134,14 @@ public:
nested_func->mergeBatch(batch_size, places, place_offset, rhs, arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> version) const override
{
nested_func->serialize(place, buf);
nested_func->serialize(place, buf, version);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> version, Arena * arena) const override
{
nested_func->deserialize(place, buf, arena);
nested_func->deserialize(place, buf, version, arena);
}
void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena * arena) const override

4
src/AggregateFunctions/AggregateFunctionMLMethod.h
View File

@ -357,9 +357,9 @@ public:
void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena *) const override { this->data(place).merge(this->data(rhs)); }
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override { this->data(place).write(buf); }
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override { this->data(place).write(buf); }
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override { this->data(place).read(buf); }
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override { this->data(place).read(buf); }
void predictValues(
ConstAggregateDataPtr place,

4
src/AggregateFunctions/AggregateFunctionMannWhitney.h
View File

@ -215,12 +215,12 @@ public:
a.merge(b, arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
this->data(place).read(buf, arena);
}

4
src/AggregateFunctions/AggregateFunctionMaxIntersections.h
View File

@ -110,7 +110,7 @@ public:
cur_elems.value.insert(rhs_elems.value.begin(), rhs_elems.value.end(), arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
const auto & value = this->data(place).value;
size_t size = value.size();
@ -118,7 +118,7 @@ public:
buf.write(reinterpret_cast<const char *>(value.data()), size * sizeof(value[0]));
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
size_t size = 0;
readVarUInt(size, buf);

18
src/AggregateFunctions/AggregateFunctionMerge.h
View File

@ -50,6 +50,16 @@ public:
return nested_func->getReturnType();
}
bool isVersioned() const override
{
return nested_func->isVersioned();
}
size_t getDefaultVersion() const override
{
return nested_func->getDefaultVersion();
}
void create(AggregateDataPtr __restrict place) const override
{
nested_func->create(place);
@ -85,14 +95,14 @@ public:
nested_func->merge(place, rhs, arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> version) const override
{
nested_func->serialize(place, buf);
nested_func->serialize(place, buf, version);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> version, Arena * arena) const override
{
nested_func->deserialize(place, buf, arena);
nested_func->deserialize(place, buf, version, arena);
}
void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena * arena) const override

4
src/AggregateFunctions/AggregateFunctionMinMaxAny.h
View File

@ -735,12 +735,12 @@ public:
this->data(place).changeIfBetter(this->data(rhs), arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).write(buf, *serialization);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
this->data(place).read(buf, *serialization, arena);
}

4
src/AggregateFunctions/AggregateFunctionNothing.h
View File

@ -62,11 +62,11 @@ public:
{
}
void serialize(ConstAggregateDataPtr, WriteBuffer &) const override
void serialize(ConstAggregateDataPtr __restrict, WriteBuffer &, std::optional<size_t>) const override
{
}
void deserialize(AggregateDataPtr, ReadBuffer &, Arena *) const override
void deserialize(AggregateDataPtr, ReadBuffer &, std::optional<size_t>, Arena *) const override
{
}

8
src/AggregateFunctions/AggregateFunctionNull.h
View File

@ -130,16 +130,16 @@ public:
nested_function->merge(nestedPlace(place), nestedPlace(rhs), arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> version) const override
{
bool flag = getFlag(place);
if constexpr (serialize_flag)
writeBinary(flag, buf);
if (flag)
nested_function->serialize(nestedPlace(place), buf);
nested_function->serialize(nestedPlace(place), buf, version);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> version, Arena * arena) const override
{
bool flag = 1;
if constexpr (serialize_flag)
@ -147,7 +147,7 @@ public:
if (flag)
{
setFlag(place);
nested_function->deserialize(nestedPlace(place), buf, arena);
nested_function->deserialize(nestedPlace(place), buf, version, arena);
}
}

23
src/AggregateFunctions/AggregateFunctionOrFill.h
View File

@ -52,6 +52,16 @@ public:
return nested_function->getName() + "OrDefault";
}
bool isVersioned() const override
{
return nested_function->isVersioned();
}
size_t getDefaultVersion() const override
{
return nested_function->getDefaultVersion();
}
bool isState() const override
{
return nested_function->isState();
@ -209,21 +219,16 @@ public:
(places[i] + place_offset)[size_of_data] |= rhs[i][size_of_data];
}
void serialize(
ConstAggregateDataPtr place,
WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr place, WriteBuffer & buf, std::optional<size_t> version) const override
{
nested_function->serialize(place, buf);
nested_function->serialize(place, buf, version);
writeChar(place[size_of_data], buf);
}
void deserialize(
AggregateDataPtr place,
ReadBuffer & buf,
Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> version, Arena * arena) const override
{
nested_function->deserialize(place, buf, arena);
nested_function->deserialize(place, buf, version, arena);
readChar(place[size_of_data], buf);
}

4
src/AggregateFunctions/AggregateFunctionQuantile.h
View File

@ -131,13 +131,13 @@ public:
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
/// const_cast is required because some data structures apply finalizaton (like compactization) before serializing.
this->data(const_cast<AggregateDataPtr>(place)).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).deserialize(buf);
}

4
src/AggregateFunctions/AggregateFunctionRankCorrelation.h
View File

@ -82,12 +82,12 @@ public:
a.merge(b, arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
this->data(place).read(buf, arena);
}

29
src/AggregateFunctions/AggregateFunctionResample.h
View File

@ -134,11 +134,7 @@ public:
nested_function->destroy(place + i * size_of_data);
}
void add(
AggregateDataPtr place,
const IColumn ** columns,
size_t row_num,
Arena * arena) const override
void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena * arena) const override
{
Key key;
@ -155,30 +151,22 @@ public:
nested_function->add(place + pos * size_of_data, columns, row_num, arena);
}
void merge(
AggregateDataPtr place,
ConstAggregateDataPtr rhs,
Arena * arena) const override
void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena * arena) const override
{
for (size_t i = 0; i < total; ++i)
nested_function->merge(place + i * size_of_data, rhs + i * size_of_data, arena);
}
void serialize(
ConstAggregateDataPtr place,
WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr place, WriteBuffer & buf, std::optional<size_t> version) const override
{
for (size_t i = 0; i < total; ++i)
nested_function->serialize(place + i * size_of_data, buf);
nested_function->serialize(place + i * size_of_data, buf, version);
}
void deserialize(
AggregateDataPtr place,
ReadBuffer & buf,
Arena * arena) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> version, Arena * arena) const override
{
for (size_t i = 0; i < total; ++i)
nested_function->deserialize(place + i * size_of_data, buf, arena);
nested_function->deserialize(place + i * size_of_data, buf, version, arena);
}
DataTypePtr getReturnType() const override
@ -186,10 +174,7 @@ public:
return std::make_shared<DataTypeArray>(nested_function->getReturnType());
}
void insertResultInto(
AggregateDataPtr place,
IColumn & to,
Arena * arena) const override
void insertResultInto(AggregateDataPtr place, IColumn & to, Arena * arena) const override
{
auto & col = assert_cast<ColumnArray &>(to);
auto & col_offsets = assert_cast<ColumnArray::ColumnOffsets &>(col.getOffsetsColumn());

4
src/AggregateFunctions/AggregateFunctionRetention.h
View File

@ -115,12 +115,12 @@ public:
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).deserialize(buf);
}

4
src/AggregateFunctions/AggregateFunctionSegmentLengthSum.h
View File

@ -177,12 +177,12 @@ public:
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).deserialize(buf);
}

4
src/AggregateFunctions/AggregateFunctionSequenceMatch.h
View File

@ -169,12 +169,12 @@ public:
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).deserialize(buf);
}

15
src/AggregateFunctions/AggregateFunctionSimpleLinearRegression.h
View File

@ -125,26 +125,17 @@ public:
this->data(place).add(x, y);
}
void merge(
AggregateDataPtr place,
ConstAggregateDataPtr rhs, Arena *
) const override
void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena *) const override
{
this->data(place).merge(this->data(rhs));
}
void serialize(
ConstAggregateDataPtr place,
WriteBuffer & buf
) const override
void serialize(ConstAggregateDataPtr place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).serialize(buf);
}
void deserialize(
AggregateDataPtr place,
ReadBuffer & buf, Arena *
) const override
void deserialize(AggregateDataPtr place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).deserialize(buf);
}

28
src/AggregateFunctions/AggregateFunctionSimpleState.h
View File

@ -49,6 +49,16 @@ public:
return storage_type;
}
bool isVersioned() const override
{
return nested_func->isVersioned();
}
size_t getDefaultVersion() const override
{
return nested_func->getDefaultVersion();
}
void create(AggregateDataPtr __restrict place) const override { nested_func->create(place); }
void destroy(AggregateDataPtr __restrict place) const noexcept override { nested_func->destroy(place); }
@ -64,13 +74,19 @@ public:
nested_func->add(place, columns, row_num, arena);
}
void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena * arena) const override { nested_func->merge(place, rhs, arena); }
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override { nested_func->serialize(place, buf); }
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena * arena) const override
{
nested_func->deserialize(place, buf, arena);
nested_func->merge(place, rhs, arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> version) const override
{
nested_func->serialize(place, buf, version);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> version, Arena * arena) const override
{
nested_func->deserialize(place, buf, version, arena);
}
void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena * arena) const override

18
src/AggregateFunctions/AggregateFunctionState.h
View File

@ -35,6 +35,16 @@ public:
DataTypePtr getReturnType() const override;
bool isVersioned() const override
{
return nested_func->isVersioned();
}
size_t getDefaultVersion() const override
{
return nested_func->getDefaultVersion();
}
void create(AggregateDataPtr __restrict place) const override
{
nested_func->create(place);
@ -70,14 +80,14 @@ public:
nested_func->merge(place, rhs, arena);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> version) const override
{
nested_func->serialize(place, buf);
nested_func->serialize(place, buf, version);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> version, Arena * arena) const override
{
nested_func->deserialize(place, buf, arena);
nested_func->deserialize(place, buf, version, arena);
}
void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override

8
src/AggregateFunctions/AggregateFunctionStatistics.h
View File

@ -136,12 +136,12 @@ public:
this->data(place).mergeWith(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).deserialize(buf);
}
@ -390,12 +390,12 @@ public:
this->data(place).mergeWith(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).deserialize(buf);
}

4
src/AggregateFunctions/AggregateFunctionStatisticsSimple.h
View File

@ -148,12 +148,12 @@ public:
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).read(buf);
}

4
src/AggregateFunctions/AggregateFunctionSum.h
View File

@ -369,12 +369,12 @@ public:
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).read(buf);
}

60
src/AggregateFunctions/AggregateFunctionSumMap.h
View File

@ -17,6 +17,8 @@
#include <Common/assert_cast.h>
#include <AggregateFunctions/IAggregateFunction.h>
#include <map>
#include <common/logger_useful.h>
#include <Common/ClickHouseRevision.h>
namespace DB
@ -38,7 +40,7 @@ struct AggregateFunctionMapData
std::map<T, Array> merged_maps;
};
/** Aggregate function, that takes at least two arguments: keys and values, and as a result, builds a tuple of of at least 2 arrays -
/** Aggregate function, that takes at least two arguments: keys and values, and as a result, builds a tuple of at least 2 arrays -
* ordered keys and variable number of argument values aggregated by corresponding keys.
*
* sumMap function is the most useful when using SummingMergeTree to sum Nested columns, which name ends in "Map".
@ -85,6 +87,18 @@ public:
values_serializations.emplace_back(type->getDefaultSerialization());
}
bool isVersioned() const override { return true; }
size_t getDefaultVersion() const override { return 1; }
size_t getVersionFromRevision(size_t revision) const override
{
if (revision >= 54448)
return 1;
else
return 0;
}
DataTypePtr getReturnType() const override
{
DataTypes types;
@ -250,8 +264,12 @@ public:
}
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> version) const override
{
if (!version)
version = getDefaultVersion();
LOG_TRACE(&Poco::Logger::get("kssenii"), "version to serialize: {}, stack: {}", *version, StackTrace().toString());
const auto & merged_maps = this->data(place).merged_maps;
size_t size = merged_maps.size();
writeVarUInt(size, buf);
@ -260,12 +278,30 @@ public:
{
keys_serialization->serializeBinary(elem.first, buf);
for (size_t col = 0; col < values_types.size(); ++col)
values_serializations[col]->serializeBinary(elem.second[col], buf);
{
switch (*version)
{
case 0:
{
values_serializations[col]->serializeBinary(elem.second[col], buf);
break;
}
case 1:
{
SerializationNumber<Int64>().serializeBinary(elem.second[col], buf);
break;
}
}
}
}
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> version, Arena *) const override
{
if (!version)
version = getDefaultVersion();
LOG_TRACE(&Poco::Logger::get("kssenii"), "version to deserialize: {}, stack: {}", *version, StackTrace().toString());
auto & merged_maps = this->data(place).merged_maps;
size_t size = 0;
readVarUInt(size, buf);
@ -278,7 +314,21 @@ public:
Array values;
values.resize(values_types.size());
for (size_t col = 0; col < values_types.size(); ++col)
values_serializations[col]->deserializeBinary(values[col], buf);
{
switch (*version)
{
case 0:
{
values_serializations[col]->deserializeBinary(values[col], buf);
break;
}
case 1:
{
SerializationNumber<Int64>().deserializeBinary(values[col], buf);
break;
}
}
}
if constexpr (IsDecimalNumber<T>)
merged_maps[key.get<DecimalField<T>>()] = values;

4
src/AggregateFunctions/AggregateFunctionTTest.h
View File

@ -128,12 +128,12 @@ public:
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).read(buf);
}

8
src/AggregateFunctions/AggregateFunctionTopK.h
View File

@ -72,12 +72,12 @@ public:
set.merge(this->data(rhs).value);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).value.write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
auto & set = this->data(place).value;
set.resize(reserved);
@ -147,12 +147,12 @@ public:
return true;
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).value.write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena * arena) const override
{
auto & set = this->data(place).value;
set.clear();

8
src/AggregateFunctions/AggregateFunctionUniq.h
View File

@ -242,12 +242,12 @@ public:
this->data(place).set.merge(this->data(rhs).set);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).set.write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).set.read(buf);
}
@ -299,12 +299,12 @@ public:
this->data(place).set.merge(this->data(rhs).set);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).set.write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).set.read(buf);
}

8
src/AggregateFunctions/AggregateFunctionUniqCombined.h
View File

@ -162,12 +162,12 @@ public:
this->data(place).set.merge(this->data(rhs).set);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).set.write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).set.read(buf);
}
@ -226,12 +226,12 @@ public:
this->data(place).set.merge(this->data(rhs).set);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).set.write(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).set.read(buf);
}

9
src/AggregateFunctions/AggregateFunctionUniqUpTo.h
View File

@ -209,12 +209,12 @@ public:
this->data(place).merge(this->data(rhs), threshold);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).write(buf, threshold);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).read(buf, threshold);
}
@ -273,12 +273,12 @@ public:
this->data(place).merge(this->data(rhs), threshold);
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).write(buf, threshold);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).read(buf, threshold);
}
@ -295,4 +295,3 @@ public:
#if !defined(__clang__)
#pragma GCC diagnostic pop
#endif

4
src/AggregateFunctions/AggregateFunctionWindowFunnel.h
View File

@ -281,12 +281,12 @@ public:
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).deserialize(buf);
}

10
src/AggregateFunctions/IAggregateFunction.h
View File

@ -67,6 +67,12 @@ public:
throw Exception("Prediction is not supported for " + getName(), ErrorCodes::NOT_IMPLEMENTED);
}
virtual bool isVersioned() const { return false; }
virtual size_t getVersionFromRevision(size_t /* revision */) const { return 0; }
virtual size_t getDefaultVersion() const { return 0; }
virtual ~IAggregateFunction() = default;
/** Data manipulating functions. */
@ -99,10 +105,10 @@ public:
virtual void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena * arena) const = 0;
/// Serializes state (to transmit it over the network, for example).
virtual void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const = 0;
virtual void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> verison = std::nullopt) const = 0;
/// Deserializes state. This function is called only for empty (just created) states.
virtual void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena * arena) const = 0;
virtual void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> version = std::nullopt, Arena * arena = nullptr) const = 0;
/// Returns true if a function requires Arena to handle own states (see add(), merge(), deserialize()).
virtual bool allocatesMemoryInArena() const = 0;

4
src/Columns/ColumnAggregateFunction.cpp
View File

@ -514,7 +514,7 @@ void ColumnAggregateFunction::insert(const Field & x)
Arena & arena = createOrGetArena();
pushBackAndCreateState(data, arena, func.get());
ReadBufferFromString read_buffer(x.get<const AggregateFunctionStateData &>().data);
func->deserialize(data.back(), read_buffer, &arena);
func->deserialize(data.back(), read_buffer, std::nullopt, &arena);
}
void ColumnAggregateFunction::insertDefault()
@ -549,7 +549,7 @@ const char * ColumnAggregateFunction::deserializeAndInsertFromArena(const char *
* Probably this will not work under UBSan.
*/
ReadBufferFromMemory read_buffer(src_arena, std::numeric_limits<char *>::max() - src_arena - 1);
func->deserialize(data.back(), read_buffer, &dst_arena);
func->deserialize(data.back(), read_buffer, std::nullopt, &dst_arena);
return read_buffer.position();
}

3
src/Columns/ColumnAggregateFunction.h
View File

@ -94,8 +94,7 @@ private:
ColumnAggregateFunction(const AggregateFunctionPtr & func_);
ColumnAggregateFunction(const AggregateFunctionPtr & func_,
const ConstArenas & arenas_);
ColumnAggregateFunction(const AggregateFunctionPtr & func_, const ConstArenas & arenas_);
ColumnAggregateFunction(const ColumnAggregateFunction & src_);

46
src/DataTypes/DataTypeAggregateFunction.cpp
View File

@ -20,6 +20,7 @@
#include <Parsers/ASTFunction.h>
#include <Parsers/ASTLiteral.h>
#include <Parsers/ASTIdentifier.h>
#include <Common/ClickHouseRevision.h>
namespace DB
@ -35,10 +36,14 @@ namespace ErrorCodes
}
std::string DataTypeAggregateFunction::doGetName() const
String DataTypeAggregateFunction::doGetName() const
{
LOG_TRACE(&Poco::Logger::get("kssenii"), "doGetName stack trace:{}", StackTrace().toString());
WriteBufferFromOwnString stream;
stream << "AggregateFunction(" << function->getName();
stream << "AggregateFunction(";
if (version)
stream << version << ", ";
stream << function->getName();
if (!parameters.empty())
{
@ -61,10 +66,10 @@ std::string DataTypeAggregateFunction::doGetName() const
MutableColumnPtr DataTypeAggregateFunction::createColumn() const
{
/// FIXME: ColumnAggregateFunction also uses function->serialize methods
return ColumnAggregateFunction::create(function);
}
/// Create empty state
Field DataTypeAggregateFunction::getDefault() const
{
@ -79,7 +84,7 @@ Field DataTypeAggregateFunction::getDefault() const
try
{
WriteBufferFromString buffer_from_field(field.get<AggregateFunctionStateData &>().data);
function->serialize(place, buffer_from_field);
function->serialize(place, buffer_from_field, version);
}
catch (...)
{
@ -100,26 +105,46 @@ bool DataTypeAggregateFunction::equals(const IDataType & rhs) const
SerializationPtr DataTypeAggregateFunction::doGetDefaultSerialization() const
{
return std::make_shared<SerializationAggregateFunction>(function);
LOG_TRACE(&Poco::Logger::get("kssenii"), "get serializaton version: {}, name: {}, stack: {}", version, getName(), StackTrace().toString());
return std::make_shared<SerializationAggregateFunction>(function, version);
}
static DataTypePtr create(const ASTPtr & arguments)
{
LOG_TRACE(&Poco::Logger::get("kssenii"), "create data type: {}", StackTrace().toString());
String function_name;
AggregateFunctionPtr function;
DataTypes argument_types;
Array params_row;
std::optional<size_t> version;
if (!arguments || arguments->children.empty())
throw Exception("Data type AggregateFunction requires parameters: "
"name of aggregate function and list of data types for arguments", ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
if (const auto * parametric = arguments->children[0]->as<ASTFunction>())
ASTPtr data_type_ast = arguments->children[0];
size_t argument_types_start_idx = 1;
/* If aggregate function definition doesn't have version, it will have in AST children args [ASTFunction, types...] - in case
* it is parametric, or [ASTIdentifier, types...] - otherwise. If aggregate function has version in AST, then it will be:
* [ASTLitearl, ASTFunction (or ASTIdentifier), types].
*/
if (auto version_ast = arguments->children[0]->as<ASTLiteral>())
{
version = version_ast->value.safeGet<UInt64>();
data_type_ast = arguments->children[1];
argument_types_start_idx = 2;
}
if (const auto * parametric = data_type_ast->as<ASTFunction>())
{
if (parametric->parameters)
throw Exception("Unexpected level of parameters to aggregate function", ErrorCodes::SYNTAX_ERROR);
function_name = parametric->name;
LOG_TRACE(&Poco::Logger::get("kssenii"), "Paramtric function name: {}", function_name);
if (parametric->arguments)
{
@ -140,11 +165,11 @@ static DataTypePtr create(const ASTPtr & arguments)
}
}
}
else if (auto opt_name = tryGetIdentifierName(arguments->children[0]))
else if (auto opt_name = tryGetIdentifierName(data_type_ast))
{
function_name = *opt_name;
}
else if (arguments->children[0]->as<ASTLiteral>())
else if (data_type_ast->as<ASTLiteral>())
{
throw Exception("Aggregate function name for data type AggregateFunction must be passed as identifier (without quotes) or function",
ErrorCodes::BAD_ARGUMENTS);
@ -153,7 +178,7 @@ static DataTypePtr create(const ASTPtr & arguments)
throw Exception("Unexpected AST element passed as aggregate function name for data type AggregateFunction. Must be identifier or function.",
ErrorCodes::BAD_ARGUMENTS);
for (size_t i = 1; i < arguments->children.size(); ++i)
for (size_t i = argument_types_start_idx; i < arguments->children.size(); ++i)
argument_types.push_back(DataTypeFactory::instance().get(arguments->children[i]));
if (function_name.empty())
@ -161,7 +186,7 @@ static DataTypePtr create(const ASTPtr & arguments)
AggregateFunctionProperties properties;
function = AggregateFunctionFactory::instance().get(function_name, argument_types, params_row, properties);
return std::make_shared<DataTypeAggregateFunction>(function, argument_types, params_row);
return std::make_shared<DataTypeAggregateFunction>(function, argument_types, params_row, version);
}
void registerDataTypeAggregateFunction(DataTypeFactory & factory)
@ -169,5 +194,4 @@ void registerDataTypeAggregateFunction(DataTypeFactory & factory)
factory.registerDataType("AggregateFunction", create);
}
}

19
src/DataTypes/DataTypeAggregateFunction.h
View File

@ -3,6 +3,7 @@
#include <AggregateFunctions/IAggregateFunction.h>
#include <DataTypes/IDataType.h>
#include <common/logger_useful.h>
namespace DB
@ -17,19 +18,24 @@ private:
AggregateFunctionPtr function;
DataTypes argument_types;
Array parameters;
mutable size_t version;
public:
static constexpr bool is_parametric = true;
DataTypeAggregateFunction(const AggregateFunctionPtr & function_, const DataTypes & argument_types_, const Array & parameters_)
: function(function_), argument_types(argument_types_), parameters(parameters_)
DataTypeAggregateFunction(const AggregateFunctionPtr & function_, const DataTypes & argument_types_,
const Array & parameters_, std::optional<size_t> version_ = std::nullopt)
: function(function_)
, argument_types(argument_types_)
, parameters(parameters_)
, version(version_ ? *version_ : function_->getDefaultVersion())
{
}
std::string getFunctionName() const { return function->getName(); }
String getFunctionName() const { return function->getName(); }
AggregateFunctionPtr getFunction() const { return function; }
std::string doGetName() const override;
String doGetName() const override;
const char * getFamilyName() const override { return "AggregateFunction"; }
TypeIndex getTypeId() const override { return TypeIndex::AggregateFunction; }
@ -50,8 +56,9 @@ public:
bool shouldAlignRightInPrettyFormats() const override { return false; }
SerializationPtr doGetDefaultSerialization() const override;
bool isVersioned() const { return function->isVersioned(); }
void setVersion(size_t version_) const { version = version_; }
};
}

38
src/DataTypes/Serializations/SerializationAggregateFunction.cpp
View File

@ -37,7 +37,7 @@ void SerializationAggregateFunction::deserializeBinary(Field & field, ReadBuffer
void SerializationAggregateFunction::serializeBinary(const IColumn & column, size_t row_num, WriteBuffer & ostr) const
{
function->serialize(assert_cast<const ColumnAggregateFunction &>(column).getData()[row_num], ostr);
function->serialize(assert_cast<const ColumnAggregateFunction &>(column).getData()[row_num], ostr, version);
}
void SerializationAggregateFunction::deserializeBinary(IColumn & column, ReadBuffer & istr) const
@ -51,7 +51,7 @@ void SerializationAggregateFunction::deserializeBinary(IColumn & column, ReadBuf
function->create(place);
try
{
function->deserialize(place, istr, &arena);
function->deserialize(place, istr, version, &arena);
}
catch (...)
{
@ -74,7 +74,7 @@ void SerializationAggregateFunction::serializeBinaryBulk(const IColumn & column,
end = vec.end();
for (; it != end; ++it)
function->serialize(*it, ostr);
function->serialize(*it, ostr, version);
}
void SerializationAggregateFunction::deserializeBinaryBulk(IColumn & column, ReadBuffer & istr, size_t limit, double /*avg_value_size_hint*/) const
@ -100,7 +100,7 @@ void SerializationAggregateFunction::deserializeBinaryBulk(IColumn & column, Rea
try
{
function->deserialize(place, istr, &arena);
function->deserialize(place, istr, version, &arena);
}
catch (...)
{
@ -112,14 +112,14 @@ void SerializationAggregateFunction::deserializeBinaryBulk(IColumn & column, Rea
}
}
static String serializeToString(const AggregateFunctionPtr & function, const IColumn & column, size_t row_num)
static String serializeToString(const AggregateFunctionPtr & function, const IColumn & column, size_t row_num, size_t version)
{
WriteBufferFromOwnString buffer;
function->serialize(assert_cast<const ColumnAggregateFunction &>(column).getData()[row_num], buffer);
function->serialize(assert_cast<const ColumnAggregateFunction &>(column).getData()[row_num], buffer, version);
return buffer.str();
}
static void deserializeFromString(const AggregateFunctionPtr & function, IColumn & column, const String & s)
static void deserializeFromString(const AggregateFunctionPtr & function, IColumn & column, const String & s, size_t version)
{
ColumnAggregateFunction & column_concrete = assert_cast<ColumnAggregateFunction &>(column);
@ -132,7 +132,7 @@ static void deserializeFromString(const AggregateFunctionPtr & function, IColumn
try
{
ReadBufferFromString istr(s);
function->deserialize(place, istr, &arena);
function->deserialize(place, istr, version, &arena);
}
catch (...)
{
@ -145,13 +145,13 @@ static void deserializeFromString(const AggregateFunctionPtr & function, IColumn
void SerializationAggregateFunction::serializeText(const IColumn & column, size_t row_num, WriteBuffer & ostr, const FormatSettings &) const
{
writeString(serializeToString(function, column, row_num), ostr);
writeString(serializeToString(function, column, row_num, version), ostr);
}
void SerializationAggregateFunction::serializeTextEscaped(const IColumn & column, size_t row_num, WriteBuffer & ostr, const FormatSettings &) const
{
writeEscapedString(serializeToString(function, column, row_num), ostr);
writeEscapedString(serializeToString(function, column, row_num, version), ostr);
}
@ -159,13 +159,13 @@ void SerializationAggregateFunction::deserializeTextEscaped(IColumn & column, Re
{
String s;
readEscapedString(s, istr);
deserializeFromString(function, column, s);
deserializeFromString(function, column, s, version);
}
void SerializationAggregateFunction::serializeTextQuoted(const IColumn & column, size_t row_num, WriteBuffer & ostr, const FormatSettings &) const
{
writeQuotedString(serializeToString(function, column, row_num), ostr);
writeQuotedString(serializeToString(function, column, row_num, version), ostr);
}
@ -173,7 +173,7 @@ void SerializationAggregateFunction::deserializeTextQuoted(IColumn & column, Rea
{
String s;
readQuotedStringWithSQLStyle(s, istr);
deserializeFromString(function, column, s);
deserializeFromString(function, column, s, version);
}
@ -181,13 +181,13 @@ void SerializationAggregateFunction::deserializeWholeText(IColumn & column, Read
{
String s;
readStringUntilEOF(s, istr);
deserializeFromString(function, column, s);
deserializeFromString(function, column, s, version);
}
void SerializationAggregateFunction::serializeTextJSON(const IColumn & column, size_t row_num, WriteBuffer & ostr, const FormatSettings & settings) const
{
writeJSONString(serializeToString(function, column, row_num), ostr, settings);
writeJSONString(serializeToString(function, column, row_num, version), ostr, settings);
}
@ -195,19 +195,19 @@ void SerializationAggregateFunction::deserializeTextJSON(IColumn & column, ReadB
{
String s;
readJSONString(s, istr);
deserializeFromString(function, column, s);
deserializeFromString(function, column, s, version);
}
void SerializationAggregateFunction::serializeTextXML(const IColumn & column, size_t row_num, WriteBuffer & ostr, const FormatSettings &) const
{
writeXMLStringForTextElement(serializeToString(function, column, row_num), ostr);
writeXMLStringForTextElement(serializeToString(function, column, row_num, version), ostr);
}
void SerializationAggregateFunction::serializeTextCSV(const IColumn & column, size_t row_num, WriteBuffer & ostr, const FormatSettings &) const
{
writeCSV(serializeToString(function, column, row_num), ostr);
writeCSV(serializeToString(function, column, row_num, version), ostr);
}
@ -215,7 +215,7 @@ void SerializationAggregateFunction::deserializeTextCSV(IColumn & column, ReadBu
{
String s;
readCSV(s, istr, settings.csv);
deserializeFromString(function, column, s);
deserializeFromString(function, column, s, version);
}
}

3
src/DataTypes/Serializations/SerializationAggregateFunction.h
View File

@ -12,11 +12,12 @@ class SerializationAggregateFunction final : public ISerialization
{
private:
AggregateFunctionPtr function;
size_t version;
public:
static constexpr bool is_parametric = true;
SerializationAggregateFunction(const AggregateFunctionPtr & function_): function(function_) {}
SerializationAggregateFunction(const AggregateFunctionPtr & function_, size_t version_): function(function_), version(version_) {}
/// NOTE These two functions for serializing single values are incompatible with the functions below.
void serializeBinary(const Field & field, WriteBuffer & ostr) const override;

2
src/Formats/ProtobufSerializer.cpp
View File

@ -1652,7 +1652,7 @@ namespace
{
aggregate_function->create(data);
ReadBufferFromMemory buf(str.data(), str.length());
aggregate_function->deserialize(data, buf, &arena);
aggregate_function->deserialize(data, buf, std::nullopt, &arena);
return data;
}
catch (...)

4
src/Processors/Transforms/WindowTransform.cpp
View File

@ -1366,8 +1366,8 @@ struct WindowFunction
size_t alignOfData() const override { return 1; }
void add(AggregateDataPtr __restrict, const IColumn **, size_t, Arena *) const override { fail(); }
void merge(AggregateDataPtr __restrict, ConstAggregateDataPtr, Arena *) const override { fail(); }
void serialize(ConstAggregateDataPtr __restrict, WriteBuffer &) const override { fail(); }
void deserialize(AggregateDataPtr __restrict, ReadBuffer &, Arena *) const override { fail(); }
void serialize(ConstAggregateDataPtr __restrict, WriteBuffer &, std::optional<size_t>) const override { fail(); }
void deserialize(AggregateDataPtr __restrict, ReadBuffer &, std::optional<size_t>, Arena *) const override { fail(); }
void insertResultInto(AggregateDataPtr __restrict, IColumn &, Arena *) const override { fail(); }
};

15
src/Storages/MergeTree/IMergeTreeDataPart.cpp
View File

@ -19,6 +19,7 @@
#include <Compression/getCompressionCodecForFile.h>
#include <Parsers/queryToString.h>
#include <DataTypes/NestedUtils.h>
#include <DataTypes/DataTypeAggregateFunction.h>
namespace CurrentMetrics
@ -1026,10 +1027,24 @@ void IMergeTreeDataPart::loadColumns(bool require)
loaded_columns.writeText(*buf);
}
volume->getDisk()->moveFile(path + ".tmp", path);
LOG_TRACE(&Poco::Logger::get("kssenii"), "Loaded from metadata");
}
else
{
//LOG_TRACE(&Poco::Logger::get("kssenii"), "Loading columns stacktrace: {}", col.name, col.type->getName());
loaded_columns.readText(*volume->getDisk()->readFile(path));
LOG_TRACE(&Poco::Logger::get("kssenii"), "Loaded from disk");
for (auto & col : loaded_columns)
{
LOG_TRACE(&Poco::Logger::get("kssenii"), "Setting version for columns: {}, {}", col.name, col.type->getName());
if (auto agg = typeid_cast<const DataTypeAggregateFunction *>(col.type.get()))
agg->setVersion(0);
}
}
for (auto & col : loaded_columns)
{
LOG_TRACE(&Poco::Logger::get("kssenii"), "Loaded columns: {}, {}", col.name, col.type->getName());
}
setColumns(loaded_columns);

11
tests/integration/test_version_update/configs/log_conf.xml Normal file
View File

@ -0,0 +1,11 @@
<yandex>
<logger>
<level>trace</level>
<log>/var/log/clickhouse-server/log.log</log>
<errorlog>/var/log/clickhouse-server/log.err.log</errorlog>
<size>1000M</size>
<count>10</count>
<stderr>/var/log/clickhouse-server/stderr.log</stderr>
<stdout>/var/log/clickhouse-server/stdout.log</stdout>
</logger>
</yandex>

99
tests/integration/test_version_update/test.py
View File

@ -1,10 +1,33 @@
import pytest
from helpers.cluster import ClickHouseCluster
from helpers.test_tools import assert_eq_with_retry, exec_query_with_retry
cluster = ClickHouseCluster(__file__)
node1 = cluster.add_instance('node1', with_zookeeper=True, image='yandex/clickhouse-server', tag='21.2', with_installed_binary=True, stay_alive=True)
node1 = cluster.add_instance('node1', main_configs=["configs/log_conf.xml"])
node2 = cluster.add_instance('node2', main_configs=["configs/log_conf.xml"],
image='yandex/clickhouse-server',
tag='21.5', with_installed_binary=True, stay_alive=True)
node3 = cluster.add_instance('node3', with_zookeeper=True, image='yandex/clickhouse-server', tag='21.2', with_installed_binary=True, stay_alive=True)
def create_and_fill_table(node):
node.query("DROP TABLE IF EXISTS test_table;")
node.query("""
CREATE TABLE test_table
(
`col1` DateTime,
`col2` Int64,
`col3` AggregateFunction(sumMap, Tuple(Array(UInt8), Array(UInt8)))
)
ENGINE = AggregatingMergeTree() ORDER BY (col1, col2) """)
node.query(""" INSERT INTO test_table
SELECT toDateTime('2020-10-01 19:20:30'), 1,
sumMapState((arrayMap(i -> 1, range(300)), arrayMap(i -> 1, range(300))));""")
@pytest.fixture(scope="module")
@ -12,17 +35,71 @@ def start_cluster():
try:
cluster.start()
yield cluster
finally:
cluster.shutdown()
def test_modulo_partition_key_after_update(start_cluster):
node1.query("CREATE TABLE test (id Int64, v UInt64, value String) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/table1', '1', v) PARTITION BY id % 20 ORDER BY (id, v)")
node1.query("INSERT INTO test SELECT number, number, toString(number) FROM numbers(10)")
expected = node1.query("SELECT number, number, toString(number) FROM numbers(10)")
partition_data = node1.query("SELECT partition, name FROM system.parts WHERE table='test' ORDER BY partition")
assert(expected == node1.query("SELECT * FROM test ORDER BY id"))
node1.restart_with_latest_version(signal=9)
assert(expected == node1.query("SELECT * FROM test ORDER BY id"))
assert(partition_data == node1.query("SELECT partition, name FROM system.parts WHERE table='test' ORDER BY partition"))
def test_modulo_partition_key_issue_23508(start_cluster):
node3.query("CREATE TABLE test (id Int64, v UInt64, value String) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/table1', '1', v) PARTITION BY id % 20 ORDER BY (id, v)")
node3.query("INSERT INTO test SELECT number, number, toString(number) FROM numbers(10)")
expected = node3.query("SELECT number, number, toString(number) FROM numbers(10)")
partition_data = node3.query("SELECT partition, name FROM system.parts WHERE table='test' ORDER BY partition")
assert(expected == node3.query("SELECT * FROM test ORDER BY id"))
node3.restart_with_latest_version()
assert(expected == node3.query("SELECT * FROM test ORDER BY id"))
assert(partition_data == node3.query("SELECT partition, name FROM system.parts WHERE table='test' ORDER BY partition"))
def test_aggregate_function_versioning_issue_16587(start_cluster):
for node in [node1, node2]:
node.query("DROP TABLE IF EXISTS test_table;")
node.query("""
CREATE TABLE test_table (`col1` DateTime, `col2` Int64)
ENGINE = MergeTree() ORDER BY col1""")
node.query("insert into test_table select '2020-10-26 00:00:00', 70724110 from numbers(300)")
expected = "([1],[600])"
# Incorrect result on old server
result_on_old_version = node2.query("select sumMap(sm) from (select sumMap([1],[1]) as sm from remote('127.0.0.{1,2}', default.test_table) group by col1, col2);")
assert(result_on_old_version.strip() != expected)
# Correct result on new server
result_on_new_version = node1.query("select sumMap(sm) from (select sumMap([1],[1]) as sm from remote('127.0.0.{1,2}', default.test_table) group by col1, col2);")
assert(result_on_new_version.strip() == expected)
def test_aggregate_function_versioning_fetch_data_from_new_to_old_server(start_cluster):
for node in [node1, node2]:
create_and_fill_table(node)
expected = "([1],[300])"
new_server_data = node1.query("select finalizeAggregation(col3) from default.test_table;").strip()
assert(new_server_data == expected)
old_server_data = node2.query("select finalizeAggregation(col3) from default.test_table;").strip()
assert(old_server_data != expected)
data_from_old_to_new_server = node1.query("select finalizeAggregation(col3) from remote('node2', default.test_table);").strip()
assert(data_from_old_to_new_server == old_server_data)
def test_aggregate_function_versioning_server_upgrade(start_cluster):
for node in [node1, node2]:
create_and_fill_table(node)
expected = "([1],[300])"
new_server_data = node1.query("select finalizeAggregation(col3) from default.test_table;").strip()
old_server_data = node2.query("select finalizeAggregation(col3) from default.test_table;").strip()
assert(old_server_data != new_server_data)
node2.restart_with_latest_version()
upgraded_server_data = node2.query("select finalizeAggregation(col3) from default.test_table;").strip()
assert(upgraded_server_data == old_server_data)

1
tests/queries/0_stateless/01881_aggregate_functions_versioning.reference Normal file
View File

@ -0,0 +1 @@
CREATE TABLE default.test_table\n(\n `col1` DateTime,\n `col2` Int64,\n `col3` AggregateFunction(1, sumMap, Tuple(Array(UInt8), Array(UInt8)))\n)\nENGINE = AggregatingMergeTree\nORDER BY (col1, col2)\nSETTINGS index_granularity = 8192

10
tests/queries/0_stateless/01881_aggregate_functions_versioning.sql Normal file
View File

@ -0,0 +1,10 @@
DROP TABLE IF EXISTS test_table;
CREATE TABLE test_table
(
`col1` DateTime,
`col2` Int64,
`col3` AggregateFunction(sumMap, Tuple(Array(UInt8), Array(UInt8)))
)
ENGINE = AggregatingMergeTree() ORDER BY (col1, col2);
SHOW CREATE TABLE test_table;