Skip to content

Commit 229f666

Browse files
alexey-milovidovalexey-milovidov
authored
Merge pull request #10611 from azat/optimize_skip_unused_shards-LowCardinality
Fix optimize_skip_unused_shards with LowCardinality
2 parents 4f1b115 + ea7038b commit 229f666

6 files changed

Lines changed: 44 additions & 50 deletions

‎src/Storages/Distributed/DistributedBlockOutputStream.cpp‎

Lines changed: 3 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,6 @@ namespace ErrorCodes
5757
{
5858
extern const int LOGICAL_ERROR;
5959
extern const int TIMEOUT_EXCEEDED;
60-
extern const int TYPE_MISMATCH;
6160
extern const int CANNOT_LINK;
6261
}
6362

@@ -457,34 +456,14 @@ void DistributedBlockOutputStream::writeSuffix()
457456
}
458457

459458

460-
IColumn::Selector DistributedBlockOutputStream::createSelector(const Block & source_block)
459+
IColumn::Selector DistributedBlockOutputStream::createSelector(const Block & source_block) const
461460
{
462461
Block current_block_with_sharding_key_expr = source_block;
463462
storage.getShardingKeyExpr()->execute(current_block_with_sharding_key_expr);
464463

465464
const auto & key_column = current_block_with_sharding_key_expr.getByName(storage.getShardingKeyColumnName());
466-
const auto & slot_to_shard = cluster->getSlotToShard();
467-
468-
// If key_column.type is DataTypeLowCardinality, do shard according to its dictionaryType
469-
#define CREATE_FOR_TYPE(TYPE) \
470-
if (typeid_cast<const DataType ## TYPE *>(key_column.type.get())) \
471-
return createBlockSelector<TYPE>(*key_column.column, slot_to_shard); \
472-
else if (auto * type_low_cardinality = typeid_cast<const DataTypeLowCardinality *>(key_column.type.get())) \
473-
if (typeid_cast<const DataType ## TYPE *>(type_low_cardinality->getDictionaryType().get())) \
474-
return createBlockSelector<TYPE>(*key_column.column->convertToFullColumnIfLowCardinality(), slot_to_shard);
475-
476-
CREATE_FOR_TYPE(UInt8)
477-
CREATE_FOR_TYPE(UInt16)
478-
CREATE_FOR_TYPE(UInt32)
479-
CREATE_FOR_TYPE(UInt64)
480-
CREATE_FOR_TYPE(Int8)
481-
CREATE_FOR_TYPE(Int16)
482-
CREATE_FOR_TYPE(Int32)
483-
CREATE_FOR_TYPE(Int64)
484-
485-
#undef CREATE_FOR_TYPE
486-
487-
throw Exception{"Sharding key expression does not evaluate to an integer type", ErrorCodes::TYPE_MISMATCH};
465+
466+
return storage.createSelector(cluster, key_column);
488467
}
489468

490469

‎src/Storages/Distributed/DistributedBlockOutputStream.h‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -46,9 +46,7 @@ class DistributedBlockOutputStream : public IBlockOutputStream
4646
void writeSuffix() override;
4747

4848
private:
49-
50-
IColumn::Selector createSelector(const Block & source_block);
51-
49+
IColumn::Selector createSelector(const Block & source_block) const;
5250

5351
void writeAsync(const Block & block);
5452

‎src/Storages/StorageDistributed.cpp‎

Lines changed: 26 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -163,29 +163,6 @@ UInt64 getMaximumFileNumber(const std::string & dir_path)
163163
return res;
164164
}
165165

166-
/// the same as DistributedBlockOutputStream::createSelector, should it be static?
167-
IColumn::Selector createSelector(const ClusterPtr cluster, const ColumnWithTypeAndName & result)
168-
{
169-
const auto & slot_to_shard = cluster->getSlotToShard();
170-
171-
#define CREATE_FOR_TYPE(TYPE) \
172-
if (typeid_cast<const DataType##TYPE *>(result.type.get())) \
173-
return createBlockSelector<TYPE>(*result.column, slot_to_shard);
174-
175-
CREATE_FOR_TYPE(UInt8)
176-
CREATE_FOR_TYPE(UInt16)
177-
CREATE_FOR_TYPE(UInt32)
178-
CREATE_FOR_TYPE(UInt64)
179-
CREATE_FOR_TYPE(Int8)
180-
CREATE_FOR_TYPE(Int16)
181-
CREATE_FOR_TYPE(Int32)
182-
CREATE_FOR_TYPE(Int64)
183-
184-
#undef CREATE_FOR_TYPE
185-
186-
throw Exception{"Sharding key expression does not evaluate to an integer type", ErrorCodes::TYPE_MISMATCH};
187-
}
188-
189166
std::string makeFormattedListOfShards(const ClusterPtr & cluster)
190167
{
191168
std::ostringstream os;
@@ -730,6 +707,32 @@ void StorageDistributed::ClusterNodeData::shutdownAndDropAllData() const
730707
directory_monitor->shutdownAndDropAllData();
731708
}
732709

710+
IColumn::Selector StorageDistributed::createSelector(const ClusterPtr cluster, const ColumnWithTypeAndName & result)
711+
{
712+
const auto & slot_to_shard = cluster->getSlotToShard();
713+
714+
// If result.type is DataTypeLowCardinality, do shard according to its dictionaryType
715+
#define CREATE_FOR_TYPE(TYPE) \
716+
if (typeid_cast<const DataType##TYPE *>(result.type.get())) \
717+
return createBlockSelector<TYPE>(*result.column, slot_to_shard); \
718+
else if (auto * type_low_cardinality = typeid_cast<const DataTypeLowCardinality *>(result.type.get())) \
719+
if (typeid_cast<const DataType ## TYPE *>(type_low_cardinality->getDictionaryType().get())) \
720+
return createBlockSelector<TYPE>(*result.column->convertToFullColumnIfLowCardinality(), slot_to_shard);
721+
722+
CREATE_FOR_TYPE(UInt8)
723+
CREATE_FOR_TYPE(UInt16)
724+
CREATE_FOR_TYPE(UInt32)
725+
CREATE_FOR_TYPE(UInt64)
726+
CREATE_FOR_TYPE(Int8)
727+
CREATE_FOR_TYPE(Int16)
728+
CREATE_FOR_TYPE(Int32)
729+
CREATE_FOR_TYPE(Int64)
730+
731+
#undef CREATE_FOR_TYPE
732+
733+
throw Exception{"Sharding key expression does not evaluate to an integer type", ErrorCodes::TYPE_MISMATCH};
734+
}
735+
733736
/// Returns a new cluster with fewer shards if constant folding for `sharding_key_expr` is possible
734737
/// using constraints from "PREWHERE" and "WHERE" conditions, otherwise returns `nullptr`
735738
ClusterPtr StorageDistributed::skipUnusedShards(ClusterPtr cluster, const ASTPtr & query_ptr, const Context & context) const

‎src/Storages/StorageDistributed.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,7 @@ class StorageDistributed final : public ext::shared_ptr_helper<StorageDistribute
112112

113113
ClusterPtr getCluster() const;
114114

115+
static IColumn::Selector createSelector(const ClusterPtr cluster, const ColumnWithTypeAndName & result);
115116
/// Apply the following settings:
116117
/// - optimize_skip_unused_shards
117118
/// - force_optimize_skip_unused_shards

‎tests/queries/0_stateless/01270_optimize_skip_unused_shards_low_cardinality.reference‎

Whitespace-only changes.

‎tests/queries/0_stateless/01270_optimize_skip_unused_shards_low_cardinality.sql‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
set optimize_skip_unused_shards=1;
2+
set force_optimize_skip_unused_shards=2;
3+
set allow_suspicious_low_cardinality_types=1;
4+
5+
drop table if exists data_01270;
6+
drop table if exists dist_01270;
7+
8+
create table data_01270 (key LowCardinality(Int)) Engine=Null();
9+
create table dist_01270 as data_01270 Engine=Distributed(test_cluster_two_shards, currentDatabase(), data_01270, key);
10+
select * from dist_01270 where key = 1;
11+
12+
drop table data_01270;
13+
drop table dist_01270;

0 commit comments

Comments
 (0)