Skip to content

[Feature](materialized-view) support count(1) on materialized view #28135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yiguolei merged 4 commits into from
Dec 8, 2023
Merged

[Feature](materialized-view) support count(1) on materialized view #28135

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
28cd4f3
support count(1) on materialized view
BiteTheDDDDt Dec 7, 2023
250a899
fix
BiteTheDDDDt Dec 8, 2023
a536d0e
fix
BiteTheDDDDt Dec 8, 2023
ed1b721
fix
BiteTheDDDDt Dec 8, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 5 additions & 25 deletions be/src/olap/schema_change.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -894,13 +894,6 @@ Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2&
for (auto item : request.materialized_view_params) {
AlterMaterializedViewParam mv_param;
mv_param.column_name = item.column_name;
/*
* origin_column_name is always be set now,
* but origin_column_name may be not set in some materialized view function. eg:count(1)
*/
if (item.__isset.origin_column_name) {
mv_param.origin_column_name = item.origin_column_name;
}

if (item.__isset.mv_expr) {
mv_param.expr = std::make_shared<TExpr>(item.mv_expr);
Expand Down Expand Up @@ -1195,9 +1188,7 @@ Status SchemaChangeHandler::_parse_request(const SchemaChangeParams& sc_params,
if (materialized_function_map.find(column_name_lower) != materialized_function_map.end()) {
auto mv_param = materialized_function_map.find(column_name_lower)->second;
column_mapping->expr = mv_param.expr;
int32_t column_index = base_tablet_schema->field_index(mv_param.origin_column_name);
column_mapping->ref_column = column_index;
if (column_index >= 0 || column_mapping->expr != nullptr) {
if (column_mapping->expr != nullptr) {
continue;
}
}
Expand Down Expand Up @@ -1253,7 +1244,7 @@ Status SchemaChangeHandler::_parse_request(const SchemaChangeParams& sc_params,
for (int i = 0, new_schema_size = new_tablet->num_key_columns(); i < new_schema_size; ++i) {
ColumnMapping* column_mapping = changer->get_mutable_column_mapping(i);

if (column_mapping->ref_column < 0) {
if (column_mapping->expr == nullptr) {
num_default_value++;
continue;
}
Expand Down Expand Up @@ -1302,22 +1293,11 @@ Status SchemaChangeHandler::_parse_request(const SchemaChangeParams& sc_params,

for (size_t i = 0; i < new_tablet->num_columns(); ++i) {
ColumnMapping* column_mapping = changer->get_mutable_column_mapping(i);
if (column_mapping->ref_column < 0) {
if (column_mapping->expr == nullptr) {
continue;
} else {
auto column_new = new_tablet_schema->column(i);
auto column_old = base_tablet_schema->column(column_mapping->ref_column);
if (column_new.type() != column_old.type() ||
column_new.precision() != column_old.precision() ||
column_new.frac() != column_old.frac() ||
column_new.length() != column_old.length() ||
column_new.is_bf_column() != column_old.is_bf_column() ||
column_new.has_bitmap_index() != column_old.has_bitmap_index() ||
new_tablet_schema->has_inverted_index(column_new) !=
base_tablet_schema->has_inverted_index(column_old)) {
*sc_directly = true;
return Status::OK();
}
*sc_directly = true;
return Status::OK();
}
}

Expand Down
21 changes: 16 additions & 5 deletions be/src/vec/data_types/data_type_decimal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -514,7 +514,9 @@ void convert_decimal_cols(
} else {
if (UNLIKELY(res > max_result.value || res < -max_result.value)) {
throw Exception(ErrorCode::ARITHMETIC_OVERFLOW_ERRROR,
"Arithmetic overflow");
"Arithmetic overflow, convert failed from {}, "
"expected data is [{}, {}]",
res, -max_result.value, max_result.value);
} else {
vec_to[i] = ToFieldType(res);
}
Expand All @@ -524,7 +526,9 @@ void convert_decimal_cols(
if constexpr (narrow_integral) {
if (UNLIKELY(res > max_result.value || res < -max_result.value)) {
throw Exception(ErrorCode::ARITHMETIC_OVERFLOW_ERRROR,
"Arithmetic overflow");
"Arithmetic overflow, convert failed from {}, "
"expected data is [{}, {}]",
res, -max_result.value, max_result.value);
}
}
vec_to[i] = ToFieldType(res);
Expand All @@ -540,7 +544,10 @@ void convert_decimal_cols(
if (UNLIKELY(vec_from[i].value > max_result.value ||
vec_from[i].value < -max_result.value)) {
throw Exception(ErrorCode::ARITHMETIC_OVERFLOW_ERRROR,
"Arithmetic overflow");
"Arithmetic overflow, convert failed from {}, "
"expected data is [{}, {}]",
vec_from[i].value, -max_result.value,
max_result.value);
}
}
vec_to[i] = ToFieldType(vec_from[i].value);
Expand All @@ -559,7 +566,9 @@ void convert_decimal_cols(
res = (vec_from[i].value + multiplier / 2) / multiplier;
if (UNLIKELY(res > max_result.value)) {
throw Exception(ErrorCode::ARITHMETIC_OVERFLOW_ERRROR,
"Arithmetic overflow");
"Arithmetic overflow, convert failed from {}, "
"expected data is [{}, {}]",
res, -max_result.value, max_result.value);
}
vec_to[i] = ToFieldType(res);
} else {
Expand All @@ -571,7 +580,9 @@ void convert_decimal_cols(
res = (vec_from[i].value - multiplier / 2) / multiplier;
if (UNLIKELY(res < -max_result.value)) {
throw Exception(ErrorCode::ARITHMETIC_OVERFLOW_ERRROR,
"Arithmetic overflow");
"Arithmetic overflow, convert failed from {}, "
"expected data is [{}, {}]",
res, -max_result.value, max_result.value);
}
vec_to[i] = ToFieldType(res);
} else {
Expand Down
15 changes: 7 additions & 8 deletions fe/fe-core/src/main/java/org/apache/doris/analysis/CreateMaterializedViewStmt.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -258,14 +258,6 @@ public void analyzeSelectClause(Analyzer analyzer) throws AnalysisException {
throw new AnalysisException("The materialized view only support the single column or function expr. "
+ "Error column: " + selectListItemExpr.toSql());
}
List<SlotRef> slots = new ArrayList<>();
selectListItemExpr.collect(SlotRef.class, slots);
if (!isReplay && slots.size() == 0) {
throw new AnalysisException(
"The materialized view contain constant expr is disallowed, expr: "
+ selectListItemExpr.toSql());
}


if (selectListItemExpr instanceof FunctionCallExpr
&& ((FunctionCallExpr) selectListItemExpr).isAggregateFunction()) {
Expand All @@ -278,6 +270,13 @@ public void analyzeSelectClause(Analyzer analyzer) throws AnalysisException {
// build mv column item
mvColumnItemList.add(buildMVColumnItem(analyzer, functionCallExpr));
} else {
List<SlotRef> slots = new ArrayList<>();
selectListItemExpr.collect(SlotRef.class, slots);
if (!isReplay && slots.size() == 0) {
throw new AnalysisException(
"The materialized view contain constant expr is disallowed, expr: "
+ selectListItemExpr.toSql());
}
if (meetAggregate) {
throw new AnalysisException("The aggregate column should be after the single column");
}
Expand Down
100 changes: 51 additions & 49 deletions .../java/org/apache/doris/nereids/rules/rewrite/mv/SelectMaterializedIndexWithAggregate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@
import org.apache.doris.nereids.trees.expressions.functions.scalar.HllHash;
import org.apache.doris.nereids.trees.expressions.functions.scalar.ToBitmap;
import org.apache.doris.nereids.trees.expressions.functions.scalar.ToBitmapWithCheck;
import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
import org.apache.doris.nereids.trees.plans.Plan;
Expand Down Expand Up @@ -1072,24 +1073,26 @@ private AggRewriteResult rewriteAgg(MaterializedIndex index,

// has rewritten agg functions
Map<Slot, Slot> slotMap = exprRewriteMap.slotMap;
if (!slotMap.isEmpty()) {
// Note that the slots in the rewritten agg functions shouldn't appear in filters or grouping expressions.
// For example: we have a duplicated-type table t(c1, c2) and a materialized index that has
// a bitmap_union column `mv_bitmap_union_c2` for the column c2.
// The query `select c1, count(distinct c2) from t where c2 > 0 group by c1` can't use the materialized
// index because we have a filter `c2 > 0` for the aggregated column c2.
Set<Slot> slotsToReplace = slotMap.keySet();
Set<String> indexConjuncts = PlanNode
// Note that the slots in the rewritten agg functions shouldn't appear in filters or grouping expressions.
// For example: we have a duplicated-type table t(c1, c2) and a materialized index that has
// a bitmap_union column `mv_bitmap_union_c2` for the column c2.
// The query `select c1, count(distinct c2) from t where c2 > 0 group by c1` can't use the materialized
// index because we have a filter `c2 > 0` for the aggregated column c2.
Set<Slot> slotsToReplace = slotMap.keySet();
Set<String> indexConjuncts;
try {
indexConjuncts = PlanNode
.splitAndCompoundPredicateToConjuncts(context.checkContext.getMeta().getWhereClause()).stream()
.map(e -> new NereidsParser().parseExpression(e.toSql()).toSql()).collect(Collectors.toSet());
if (isInputSlotsContainsNone(
predicates.stream().filter(e -> !indexConjuncts.contains(e.toSql())).collect(Collectors.toList()),
slotsToReplace) && isInputSlotsContainsNone(groupingExprs, slotsToReplace)) {
ImmutableSet<Slot> newRequiredSlots = requiredScanOutput.stream()
.map(slot -> (Slot) ExpressionUtils.replace(slot, slotMap))
.collect(ImmutableSet.toImmutableSet());
return new AggRewriteResult(index, true, newRequiredSlots, exprRewriteMap);
}
} catch (Exception e) {
return new AggRewriteResult(index, false, null, null);
}
if (isInputSlotsContainsNone(
predicates.stream().filter(e -> !indexConjuncts.contains(e.toSql())).collect(Collectors.toList()),
slotsToReplace) && isInputSlotsContainsNone(groupingExprs, slotsToReplace)) {
ImmutableSet<Slot> newRequiredSlots = requiredScanOutput.stream()
.map(slot -> (Slot) ExpressionUtils.replace(slot, slotMap)).collect(ImmutableSet.toImmutableSet());
return new AggRewriteResult(index, true, newRequiredSlots, exprRewriteMap);
}

return new AggRewriteResult(index, false, null, null);
Expand Down Expand Up @@ -1207,8 +1210,7 @@ public Expression visitCount(Count count, RewriteContext context) {

Expression expr = new ToBitmapWithCheck(castIfNeed(count.child(0), BigIntType.INSTANCE));
// count distinct a value column.
if (slotOpt.isPresent() && !context.checkContext.keyNameToColumn.containsKey(
normalizeName(expr.toSql()))) {
if (slotOpt.isPresent()) {
String bitmapUnionColumn = normalizeName(CreateMaterializedViewStmt.mvColumnBuilder(
AggregateType.BITMAP_UNION, CreateMaterializedViewStmt.mvColumnBuilder(expr.toSql())));

Expand All @@ -1229,33 +1231,37 @@ public Expression visitCount(Count count, RewriteContext context) {
return bitmapUnionCount;
}
}
} else if (!count.isDistinct() && count.arity() == 1) {
// count(col) -> sum(mva_SUM__CASE WHEN col IS NULL THEN 0 ELSE 1 END)
}

Expression child = null;
if (!count.isDistinct() && count.arity() == 1) {
// count(col) -> sum(mva_SUM__CASE WHEN col IS NULL THEN 0 ELSE 1 END)
Optional<Slot> slotOpt = ExpressionUtils.extractSlotOrCastOnSlot(count.child(0));
// count a value column.
if (slotOpt.isPresent() && !context.checkContext.keyNameToColumn.containsKey(
normalizeName(slotOpt.get().toSql()))) {
String countColumn = normalizeName(CreateMaterializedViewStmt
.mvColumnBuilder(AggregateType.SUM,
CreateMaterializedViewStmt.mvColumnBuilder(slotToCaseWhen(slotOpt.get()).toSql())));

Column mvColumn = context.checkContext.getColumn(countColumn);
// has bitmap_union_count column
if (mvColumn != null && context.checkContext.valueNameToColumn.containsValue(mvColumn)) {
Slot countSlot = context.checkContext.scan.getOutputByIndex(context.checkContext.index)
.stream()
.filter(s -> countColumn.equalsIgnoreCase(normalizeName(s.getName())))
.findFirst()
.orElseThrow(() -> new AnalysisException(
"cannot find count slot when select mv"));
if (slotOpt.isPresent()) {
child = slotOpt.get();
}
} else if (count.arity() == 0) {
// count(*) / count(1) -> sum(mva_SUM__CASE WHEN 1 IS NULL THEN 0 ELSE 1 END)
child = new TinyIntLiteral((byte) 1);
}

context.exprRewriteMap.slotMap.put(slotOpt.get(), countSlot);
context.exprRewriteMap.projectExprMap.put(slotOpt.get(), countSlot);
Sum sum = new Sum(countSlot);
context.exprRewriteMap.aggFuncMap.put(count, sum);
return sum;
if (child != null) {
String countColumn = normalizeName(CreateMaterializedViewStmt.mvColumnBuilder(AggregateType.SUM,
CreateMaterializedViewStmt.mvColumnBuilder(slotToCaseWhen(child).toSql())));

Column mvColumn = context.checkContext.getColumn(countColumn);
if (mvColumn != null && context.checkContext.valueNameToColumn.containsValue(mvColumn)) {
Slot countSlot = context.checkContext.scan.getOutputByIndex(context.checkContext.index).stream()
.filter(s -> countColumn.equalsIgnoreCase(normalizeName(s.getName()))).findFirst()
.orElseThrow(() -> new AnalysisException("cannot find count slot when select mv"));

if (child instanceof Slot) {
context.exprRewriteMap.slotMap.put((Slot) child, countSlot);
}
context.exprRewriteMap.projectExprMap.put(child, countSlot);
Sum sum = new Sum(countSlot);
context.exprRewriteMap.aggFuncMap.put(count, sum);
return sum;
}
}
return count;
Expand Down Expand Up @@ -1416,8 +1422,7 @@ public Expression visitNdv(Ndv ndv, RewriteContext context) {
}
Optional<Slot> slotOpt = ExpressionUtils.extractSlotOrCastOnSlot(ndv.child(0));
// ndv on a value column.
if (slotOpt.isPresent() && !context.checkContext.keyNameToColumn.containsKey(
normalizeName(slotOpt.get().toSql()))) {
if (slotOpt.isPresent()) {
Expression expr = castIfNeed(ndv.child(), VarcharType.SYSTEM_DEFAULT);
String hllUnionColumn = normalizeName(
CreateMaterializedViewStmt.mvColumnBuilder(AggregateType.HLL_UNION,
Expand Down Expand Up @@ -1450,8 +1455,7 @@ public Expression visitSum(Sum sum, RewriteContext context) {
return result;
}
Optional<Slot> slotOpt = ExpressionUtils.extractSlotOrCastOnSlot(sum.child(0));
if (!sum.isDistinct() && slotOpt.isPresent()
&& !context.checkContext.keyNameToColumn.containsKey(normalizeName(slotOpt.get().toSql()))) {
if (!sum.isDistinct() && slotOpt.isPresent()) {
Expression expr = castIfNeed(sum.child(), BigIntType.INSTANCE);
String sumColumn = normalizeName(CreateMaterializedViewStmt.mvColumnBuilder(AggregateType.SUM,
CreateMaterializedViewStmt.mvColumnBuilder(expr.toSql())));
Expand Down Expand Up @@ -1487,10 +1491,8 @@ public Expression visitAggregateFunction(AggregateFunction aggregateFunction, Re

Set<Slot> slots = aggregateFunction.collect(SlotReference.class::isInstance);
for (Slot slot : slots) {
if (!context.checkContext.keyNameToColumn.containsKey(normalizeName(slot.toSql()))) {
context.exprRewriteMap.slotMap.put(slot, aggStateSlot);
context.exprRewriteMap.projectExprMap.put(slot, aggStateSlot);
}
context.exprRewriteMap.slotMap.put(slot, aggStateSlot);
context.exprRewriteMap.projectExprMap.put(slot, aggStateSlot);
}

MergeCombinator mergeCombinator = new MergeCombinator(Arrays.asList(aggStateSlot), aggregateFunction);
Expand Down
1 change: 0 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/task/AlterReplicaTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,6 @@ public TAlterTabletReqV2 toThrift() {
List<SlotRef> slots = Lists.newArrayList();
entry.getValue().collect(SlotRef.class, slots);
TAlterMaterializedViewParam mvParam = new TAlterMaterializedViewParam(entry.getKey());
mvParam.setOriginColumnName(slots.get(0).getColumnName());
mvParam.setMvExpr(entry.getValue().treeToThrift());
req.addToMaterializedViewParams(mvParam);
}
Expand Down
Loading