This is an automated email from the ASF dual-hosted git repository.
morrySnow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 51dd80c69b2 [improvement](mtmv) Optimize MTMV partition lineage check
(#63899)
51dd80c69b2 is described below
commit 51dd80c69b2da1bcdb0829930790c657279327ce
Author: seawinde <[email protected]>
AuthorDate: Tue Aug 18 16:26:39 2026 +0800
[improvement](mtmv) Optimize MTMV partition lineage check (#63899)
### What problem does this PR solve?
Problem Summary:
Complex partitioned async MTMV creation can spend excessive FE CPU in
partition lineage analysis. The hot path repeatedly shuttles partition
and checked expressions through the full plan lineage replacer, so wide
`UNION ALL`, join, and aggregate plans multiply the same plan walks
during `CREATE MATERIALIZED VIEW` analysis.
Root cause: In
`PartitionIncrementMaintainer.PartitionIncrementChecker.checkPartition()`,
each partition candidate and checked expression calls
`ExpressionUtils.shuttleExpressionWithLineage()` separately. Each call
traverses the plan through `ExpressionLineageReplacer` and rebuilds
equivalent normalized expressions.
Change Summary:
| File | Change Description |
|------|-------------------|
| `PartitionIncrementMaintainer.java` | Batch lineage shuttle calls,
cache lineage-visible named expressions by plan identity, cache
normalized expressions, and reuse the normalization rewrite context
during one partition increment check. |
| `PartitionColumnTraceTest.java` | Add a CTE plus `UNION ALL` plus wide
aggregate lineage test to keep partition lineage behavior covered. |
| `test_mtmv_partition_lineage_performance.groovy` | Add a desensitized
static SQL performance regression case for the complex partitioned MTMV
shape. |
Design Rationale: The change keeps the existing
`ExpressionLineageReplacer` semantics and limits caching to a single
`PartitionIncrementCheckContext`. This avoids sharing mutable analysis
state across optimizer contexts while removing repeated full plan walks
for the same plan and expression set.
### Release note
Improve performance when creating complex partitioned async materialized
views.
---
.../mv/PartitionIncrementMaintainer.java | 156 +++--
.../exploration/mv/PartitionColumnTraceTest.java | 82 +++
.../test_mtmv_partition_lineage_performance.groovy | 663 +++++++++++++++++++++
3 files changed, 867 insertions(+), 34 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionIncrementMaintainer.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionIncrementMaintainer.java
index c35a06ed3b7..9f5b2c5f120 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionIncrementMaintainer.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionIncrementMaintainer.java
@@ -37,6 +37,7 @@ import
org.apache.doris.nereids.trees.expressions.WindowExpression;
import org.apache.doris.nereids.trees.expressions.functions.scalar.DateTrunc;
import org.apache.doris.nereids.trees.expressions.literal.Literal;
import
org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
import org.apache.doris.nereids.trees.plans.JoinType;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.algebra.SetOperation;
@@ -59,8 +60,8 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.trees.plans.visitor.ExpressionLineageReplacer;
import org.apache.doris.nereids.types.DataType;
-import org.apache.doris.nereids.util.ExpressionUtils;
import org.apache.doris.nereids.util.ImmutableEqualSet;
import com.google.common.collect.ImmutableList;
@@ -71,6 +72,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -146,6 +148,13 @@ public class PartitionIncrementMaintainer {
boolean allReachRelationCheck = true;
boolean allIsFromTablePartitionColumn = true;
for (PartitionIncrementCheckContext childContext :
childrenContextList) {
+ if (childContext.isFailFast()) {
+ context.addFailReason(String.format(
+ "union all child partition increment check failed,
fail reason is %s",
+ childContext.getFailReasons()));
+ context.setFailFast(true);
+ return null;
+ }
boolean childAnyIsFromTablePartitionColumn = false;
boolean childAnyReachRelationCheck = false;
for (RelatedTableColumnInfo tableColumnInfo :
childContext.getPartitionAndRefExpressionMap().values()) {
@@ -172,7 +181,7 @@ public class PartitionIncrementMaintainer {
context.collectFailedTableSet(union);
context.addFailReason("not union all output pass partition
increment check");
}
- return super.visit(union, context);
+ return null;
}
@Override
@@ -200,8 +209,7 @@ public class PartitionIncrementMaintainer {
Set<Set<Slot>> shuttledEqualSlotSet =
context.getShuttledEqualSlotSet();
for (Set<Slot> equalSlotSet : shuttledEqualSlotSet) {
if (equalSlotSet.contains(consumerSlot)) {
- Expression shuttledSlot =
ExpressionUtils.shuttleExpressionWithLineage(
- producerSlot, producerPlan);
+ Expression shuttledSlot =
context.shuttleExpressionWithLineage(producerSlot, producerPlan);
if (shuttledSlot instanceof Slot) {
equalSlotSet.add((Slot) shuttledSlot);
}
@@ -241,7 +249,7 @@ public class PartitionIncrementMaintainer {
continue;
}
Pair<Set<Slot>, Set<Slot>> partitionEqualSlotPair =
- calEqualSet((SlotReference) partitionSlotToCheck,
join);
+ calEqualSet((SlotReference) partitionSlotToCheck,
join, context);
if (!partitionEqualSlotPair.value().isEmpty()) {
context.getShuttledEqualSlotSet().add(partitionEqualSlotPair.value());
}
@@ -261,6 +269,9 @@ public class PartitionIncrementMaintainer {
// check join type and partition column side
Set<Slot> leftColumnSet = join.child(0).getOutputSet();
Set<NamedExpression> namedExpressions = new
HashSet<>(context.getPartitionAndRefExpressionMap().keySet());
+ boolean needVisitJoinChildren = false;
+ boolean needCollectInvalidRight = false;
+ boolean needCollectInvalidLeft = false;
for (NamedExpression partitionSlotToCheck : namedExpressions) {
if (!(partitionSlotToCheck instanceof SlotReference)) {
continue;
@@ -268,22 +279,31 @@ public class PartitionIncrementMaintainer {
boolean useLeft = leftColumnSet.contains(partitionSlotToCheck);
JoinType joinType = join.getJoinType();
if (joinType.isInnerJoin() || joinType.isCrossJoin()) {
- visit(join, context);
+ needVisitJoinChildren = true;
} else if ((joinType.isLeftJoin()
|| joinType.isLeftSemiJoin()
|| joinType.isLeftAntiJoin()) && useLeft) {
- context.collectInvalidTableSet(join.right());
- visit(join, context);
+ needCollectInvalidRight = true;
+ needVisitJoinChildren = true;
} else if ((joinType.isRightJoin()
|| joinType.isRightAntiJoin()
|| joinType.isRightSemiJoin()) && !useLeft) {
- context.collectInvalidTableSet(join.left());
- visit(join, context);
+ needCollectInvalidLeft = true;
+ needVisitJoinChildren = true;
} else {
context.addFailReason(String.format("partition column is
in un supported join null generate side, "
+ "current join type is %s, partitionSlot is %s",
joinType, partitionSlotToCheck));
}
}
+ if (needCollectInvalidRight) {
+ context.collectInvalidTableSet(join.right());
+ }
+ if (needCollectInvalidLeft) {
+ context.collectInvalidTableSet(join.left());
+ }
+ if (needVisitJoinChildren) {
+ visit(join, context);
+ }
return null;
}
@@ -586,31 +606,24 @@ public class PartitionIncrementMaintainer {
*/
private static boolean checkPartition(Collection<? extends Expression>
expressionsToCheck, Plan plan,
PartitionIncrementCheckContext context) {
- Set<Entry<NamedExpression, RelatedTableColumnInfo>>
partitionAndExprEntrySet
- = new
HashSet<>(context.getPartitionAndRefExpressionMap().entrySet());
+ List<Entry<NamedExpression, RelatedTableColumnInfo>>
partitionAndExprEntryList
+ = new
ArrayList<>(context.getPartitionAndRefExpressionMap().entrySet());
+ List<Expression> partitionExpressions = new
ArrayList<>(partitionAndExprEntryList.size());
+ for (Entry<NamedExpression, RelatedTableColumnInfo> entry :
partitionAndExprEntryList) {
+
partitionExpressions.add(entry.getValue().getPartitionExpression().orElse(entry.getKey()));
+ }
+ List<? extends Expression> partitionExpressionActualList =
+
context.shuttleAndNormalizeExpressionWithLineage(partitionExpressions,
context.getOriginalPlan());
+ List<? extends Expression> expressionsShuttledToCheck =
+
context.shuttleAndNormalizeExpressionWithLineage(expressionsToCheck,
context.getOriginalPlan());
boolean checked = false;
- for (Map.Entry<NamedExpression, RelatedTableColumnInfo>
partitionExpressionEntry
- : partitionAndExprEntrySet) {
- NamedExpression partitionNamedExpression =
partitionExpressionEntry.getKey();
+ for (int i = 0; i < partitionAndExprEntryList.size(); i++) {
+ Map.Entry<NamedExpression, RelatedTableColumnInfo>
partitionExpressionEntry =
+ partitionAndExprEntryList.get(i);
RelatedTableColumnInfo partitionTableColumnInfo =
partitionExpressionEntry.getValue();
- Optional<Expression> partitionExpressionOpt =
partitionTableColumnInfo.getPartitionExpression();
- Expression partitionExpressionActual = partitionExpressionOpt
- .map(expr ->
ExpressionUtils.shuttleExpressionWithLineage(expr,
- context.getOriginalPlan()))
- .orElseGet(() ->
ExpressionUtils.shuttleExpressionWithLineage(partitionNamedExpression,
- context.getOriginalPlan()));
- // merge date_trunc
- partitionExpressionActual = new
ExpressionNormalization().rewrite(partitionExpressionActual,
- new
ExpressionRewriteContext(context.getCascadesContext()));
+ Expression partitionExpressionActual =
partitionExpressionActualList.get(i);
OUTER_CHECK:
- for (Expression projectSlotToCheck : expressionsToCheck) {
- Expression expressionShuttledToCheck =
-
ExpressionUtils.shuttleExpressionWithLineage(projectSlotToCheck,
- context.getOriginalPlan());
- // merge date_trunc
- expressionShuttledToCheck = new
ExpressionNormalization().rewrite(expressionShuttledToCheck,
- new
ExpressionRewriteContext(context.getCascadesContext()));
-
+ for (Expression expressionShuttledToCheck :
expressionsShuttledToCheck) {
Set<SlotReference> expressionToCheckSlots =
expressionShuttledToCheck.collectToSet(SlotReference.class::isInstance);
Set<SlotReference> partitionColumnSlots =
@@ -728,6 +741,8 @@ public class PartitionIncrementMaintainer {
* The context used in IncrementChecker
*/
public static final class PartitionIncrementCheckContext {
+ private static final ExpressionNormalization EXPRESSION_NORMALIZATION
= new ExpressionNormalization();
+
// This is used to record partition slot, and the value of map is ref
date expression and bool value which
// identify it's original partition or not, the key of map is the
namedExpression to check
private final Map<NamedExpression, RelatedTableColumnInfo>
partitionAndRefExpressionMap
@@ -743,8 +758,15 @@ public class PartitionIncrementMaintainer {
private final Set<Set<Slot>> shuttledEqualSlotSet = new HashSet<>();
private final Map<CTEId, Plan> producerCteIdToPlanMap;
private final Plan originalPlan;
+ // Cache lineage-visible named expressions per plan identity to avoid
repeated full plan walks.
+ private final Map<Plan, List<NamedExpression>>
planLineageExpressionIndexes = new IdentityHashMap<>();
+ // Reuse the rewrite context because all normalization in this checker
shares the same CascadesContext.
+ private final ExpressionRewriteContext expressionRewriteContext;
private boolean failFast = false;
+ /**
+ * Construct partition increment check context.
+ */
public PartitionIncrementCheckContext(NamedExpression
mvPartitionColumn,
Expression mvPartitionExpression, Map<CTEId, Plan>
producerCteIdToPlanMap,
Plan originalPlan,
@@ -754,6 +776,7 @@ public class PartitionIncrementMaintainer {
this.cascadesContext = cascadesContext;
this.producerCteIdToPlanMap = producerCteIdToPlanMap;
this.originalPlan = originalPlan;
+ this.expressionRewriteContext = new
ExpressionRewriteContext(cascadesContext);
}
public Set<String> getFailReasons() {
@@ -803,6 +826,51 @@ public class PartitionIncrementMaintainer {
return originalPlan;
}
+ private Expression shuttleExpressionWithLineage(Expression expression,
Plan plan) {
+ return shuttleExpressionWithLineage(ImmutableList.of(expression),
plan).get(0);
+ }
+
+ private List<? extends Expression> shuttleExpressionWithLineage(List<?
extends Expression> expressions,
+ Plan plan) {
+ if (expressions.isEmpty()) {
+ return ImmutableList.of();
+ }
+ ExpressionLineageReplacer.ExpressionReplaceContext replaceContext =
+ new
ExpressionLineageReplacer.ExpressionReplaceContext(expressions);
+ for (NamedExpression namedExpression :
getLineageExpressionIndex(plan)) {
+ if
(!replaceContext.getUsedExprIdSet().contains(namedExpression.getExprId())) {
+ continue;
+ }
+
namedExpression.accept(ExpressionLineageReplacer.NamedExpressionCollector.INSTANCE,
replaceContext);
+ }
+ return replaceContext.getReplacedExpressions();
+ }
+
+ private List<? extends Expression>
shuttleAndNormalizeExpressionWithLineage(
+ Collection<? extends Expression> expressions, Plan plan) {
+ if (expressions.isEmpty()) {
+ return ImmutableList.of();
+ }
+ List<? extends Expression> shuttledExpressions =
+
shuttleExpressionWithLineage(ImmutableList.copyOf(expressions), plan);
+ List<Expression> normalizedExpressions = new
ArrayList<>(shuttledExpressions.size());
+ for (Expression expression : shuttledExpressions) {
+
normalizedExpressions.add(EXPRESSION_NORMALIZATION.rewrite(expression,
expressionRewriteContext));
+ }
+ return normalizedExpressions;
+ }
+
+ private List<NamedExpression> getLineageExpressionIndex(Plan plan) {
+ List<NamedExpression> lineageExpressionIndex =
planLineageExpressionIndexes.get(plan);
+ if (lineageExpressionIndex == null) {
+ List<NamedExpression> collectedIndex = new ArrayList<>();
+ plan.accept(LineageExpressionCollector.INSTANCE,
collectedIndex);
+ lineageExpressionIndex = collectedIndex;
+ planLineageExpressionIndexes.put(plan, lineageExpressionIndex);
+ }
+ return lineageExpressionIndex;
+ }
+
/**
* collect invalid table set to check self join
*/
@@ -832,6 +900,25 @@ public class PartitionIncrementMaintainer {
}
}
+ private static final class LineageExpressionCollector extends
DefaultPlanVisitor<Void, List<NamedExpression>> {
+ private static final LineageExpressionCollector INSTANCE = new
LineageExpressionCollector();
+
+ @Override
+ public Void visitGroupPlan(GroupPlan groupPlan, List<NamedExpression>
lineageExpressionIndex) {
+ return null;
+ }
+
+ @Override
+ public Void visit(Plan plan, List<NamedExpression>
lineageExpressionIndex) {
+ for (Expression expression : plan.getExpressions()) {
+ if (expression instanceof NamedExpression) {
+ lineageExpressionIndex.add((NamedExpression) expression);
+ }
+ }
+ return super.visit(plan, lineageExpressionIndex);
+ }
+ }
+
/**
* Add partitionEqualSlot to partitionAndRefExpressionToCheck if
partitionExpression use the partitionSlot
*/
@@ -876,7 +963,8 @@ public class PartitionIncrementMaintainer {
* the value equal set contain the slot itself
*/
private static Pair<Set<Slot>, Set<Slot>> calEqualSet(Slot slot,
- LogicalJoin<? extends Plan, ? extends Plan> join) {
+ LogicalJoin<? extends Plan, ? extends Plan> join,
+ PartitionIncrementCheckContext context) {
Set<Slot> partitionEqualSlotSet = new HashSet<>();
JoinType joinType = join.getJoinType();
if (joinType.isInnerJoin() || joinType.isSemiJoin()) {
@@ -889,7 +977,7 @@ public class PartitionIncrementMaintainer {
}
List<Expression> extendedPartitionEqualSlotSet = new
ArrayList<>(partitionEqualSlotSet);
extendedPartitionEqualSlotSet.add(slot);
- List<? extends Expression> shuttledEqualExpressions =
ExpressionUtils.shuttleExpressionWithLineage(
+ List<? extends Expression> shuttledEqualExpressions =
context.shuttleExpressionWithLineage(
extendedPartitionEqualSlotSet, join);
for (Expression shuttledEqualExpression : shuttledEqualExpressions) {
Set<Slot> objects = shuttledEqualExpression.collectToSet(expr ->
expr instanceof SlotReference);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionColumnTraceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionColumnTraceTest.java
index 53b98de4409..13b65661465 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionColumnTraceTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionColumnTraceTest.java
@@ -1107,6 +1107,28 @@ public class PartitionColumnTraceTest extends
TestWithFeService {
});
}
+ @Test
+ public void testUnionAllRejectsFailedChildContext() {
+ PlanChecker.from(connectContext)
+ .checkExplain("select L_SHIPDATE as part_date\n"
+ + "from lineitem\n"
+ + "cross join (\n"
+ + " select O_ORDERKEY from orders\n"
+ + " intersect\n"
+ + " select L_ORDERKEY from lineitem\n"
+ + ") unsupported_branch\n"
+ + "union all\n"
+ + "select O_ORDERDATE as part_date\n"
+ + "from orders",
+ nereidsPlanner -> {
+ Plan rewrittenPlan =
nereidsPlanner.getRewrittenPlan();
+ RelatedTableInfo relatedTableInfo =
+
MaterializedViewUtils.getRelatedTableInfos("part_date", null,
+ rewrittenPlan,
nereidsPlanner.getCascadesContext());
+ failWith(relatedTableInfo, "LogicalIntersect");
+ });
+ }
+
// test with cte
@Test
@@ -1157,6 +1179,66 @@ public class PartitionColumnTraceTest extends
TestWithFeService {
});
}
+ // CTE + union all + wide aggregate should keep partition lineage inside
each plan boundary.
+ @Test
+ public void testCteUnionAllWideAggregatePartitionLineage() {
+ PlanChecker.from(connectContext)
+ .checkExplain("with union_src as (\n"
+ + " select\n"
+ + " L_SHIPDATE as part_date,\n"
+ + " L_ORDERKEY as order_key,\n"
+ + " L_QUANTITY as metric1,\n"
+ + " L_EXTENDEDPRICE as metric2,\n"
+ + " L_DISCOUNT as metric3,\n"
+ + " L_TAX as metric4,\n"
+ + " L_RETURNFLAG as flag\n"
+ + " from lineitem\n"
+ + " union all\n"
+ + " select\n"
+ + " O_ORDERDATE as part_date,\n"
+ + " O_ORDERKEY as order_key,\n"
+ + " O_TOTALPRICE as metric1,\n"
+ + " O_TOTALPRICE as metric2,\n"
+ + " O_TOTALPRICE as metric3,\n"
+ + " O_TOTALPRICE as metric4,\n"
+ + " O_ORDERSTATUS as flag\n"
+ + " from orders\n"
+ + "), wide_project as (\n"
+ + " select\n"
+ + " date_trunc(part_date, 'day') as
part_day,\n"
+ + " part_date,\n"
+ + " order_key,\n"
+ + " metric1,\n"
+ + " metric2,\n"
+ + " metric3,\n"
+ + " metric4,\n"
+ + " flag\n"
+ + " from union_src\n"
+ + ")\n"
+ + "select\n"
+ + " part_day,\n"
+ + " flag,\n"
+ + " count(*) as cnt,\n"
+ + " sum(metric1) as sum_metric1,\n"
+ + " sum(metric2) as sum_metric2,\n"
+ + " sum(metric3) as sum_metric3,\n"
+ + " sum(metric4) as sum_metric4,\n"
+ + " max(order_key) as max_key,\n"
+ + " min(order_key) as min_key\n"
+ + "from wide_project\n"
+ + "group by part_day, flag",
+ nereidsPlanner -> {
+ Plan rewrittenPlan =
nereidsPlanner.getRewrittenPlan();
+ RelatedTableInfo relatedTableInfo =
+
MaterializedViewUtils.getRelatedTableInfos("part_day", null,
+ rewrittenPlan,
nereidsPlanner.getCascadesContext());
+ successWith(relatedTableInfo, ImmutableSet.of(
+ ImmutableList.of("lineitem",
"l_shipdate", "true", "true"),
+ ImmutableList.of("orders",
"o_orderdate", "true", "true")),
+ "day");
+ });
+ }
+
// test with union but not union all
@Test
diff --git
a/regression-test/suites/performance_p0/test_mtmv_partition_lineage_performance.groovy
b/regression-test/suites/performance_p0/test_mtmv_partition_lineage_performance.groovy
new file mode 100644
index 00000000000..7b780a36a4d
--- /dev/null
+++
b/regression-test/suites/performance_p0/test_mtmv_partition_lineage_performance.groovy
@@ -0,0 +1,663 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_mtmv_partition_lineage_performance", "mtmv") {
+ sql "SET enable_nereids_planner=true"
+ sql "SET enable_fallback_to_original_planner=false"
+ sql "SET enable_materialized_view_rewrite=false"
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS mtmv_partition_lineage_perf_mv"""
+ sql """DROP TABLE IF EXISTS mtmv_partition_lineage_perf_fact"""
+ sql """DROP TABLE IF EXISTS mtmv_partition_lineage_perf_header"""
+ sql """DROP TABLE IF EXISTS mtmv_partition_lineage_perf_detail"""
+ sql """DROP TABLE IF EXISTS mtmv_partition_lineage_perf_attr"""
+
+ sql """
+ CREATE TABLE mtmv_partition_lineage_perf_fact (
+ profile_id INT NOT NULL,
+ event_time DATETIME NOT NULL,
+ doc_id BIGINT NOT NULL,
+ line_id BIGINT NOT NULL,
+ dim_id INT NOT NULL,
+ entity_id INT NOT NULL,
+ attr_id INT NOT NULL,
+ doc_type INT NOT NULL,
+ status INT NOT NULL,
+ measure_a DECIMAL(18, 2) NOT NULL,
+ value_a DECIMAL(18, 2) NOT NULL,
+ value_b DECIMAL(18, 2) NOT NULL,
+ value_c DECIMAL(18, 2) NOT NULL,
+ value_d DECIMAL(18, 2) NOT NULL,
+ value_e DECIMAL(18, 2) NOT NULL
+ )
+ DUPLICATE KEY(profile_id, event_time, doc_id, line_id)
+ AUTO PARTITION BY RANGE (date_trunc(event_time, 'month')) ()
+ DISTRIBUTED BY HASH(doc_id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ CREATE TABLE mtmv_partition_lineage_perf_header (
+ profile_id INT NOT NULL,
+ event_time DATETIME NOT NULL,
+ doc_id BIGINT NOT NULL,
+ doc_type INT NOT NULL,
+ status INT NOT NULL,
+ flag_value INT NOT NULL,
+ source_id INT NOT NULL
+ )
+ DUPLICATE KEY(profile_id, event_time, doc_id)
+ AUTO PARTITION BY RANGE (date_trunc(event_time, 'month')) ()
+ DISTRIBUTED BY HASH(doc_id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ CREATE TABLE mtmv_partition_lineage_perf_detail (
+ profile_id INT NOT NULL,
+ event_time DATETIME NOT NULL,
+ doc_id BIGINT NOT NULL,
+ line_id BIGINT NOT NULL,
+ measure_b DECIMAL(18, 2) NOT NULL,
+ value_f DECIMAL(18, 2) NOT NULL,
+ factor_a DECIMAL(18, 2) NOT NULL,
+ factor_b DECIMAL(18, 2) NOT NULL
+ )
+ DUPLICATE KEY(profile_id, event_time, doc_id, line_id)
+ AUTO PARTITION BY RANGE (date_trunc(event_time, 'month')) ()
+ DISTRIBUTED BY HASH(doc_id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ CREATE TABLE mtmv_partition_lineage_perf_attr (
+ profile_id INT NOT NULL,
+ event_time DATETIME NOT NULL,
+ attr_id INT NOT NULL,
+ category_id INT NOT NULL,
+ active_flag INT NOT NULL
+ )
+ DUPLICATE KEY(profile_id, event_time, attr_id)
+ AUTO PARTITION BY RANGE (date_trunc(event_time, 'month')) ()
+ DISTRIBUTED BY HASH(attr_id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ INSERT INTO mtmv_partition_lineage_perf_fact VALUES
+ (1, '2024-01-10 00:00:00', 1001, 1, 10, 20, 30, 601, 20, 1.00, 100.00,
6.00, 2.00, 3.00, 4.00)
+ """
+ sql """
+ INSERT INTO mtmv_partition_lineage_perf_header VALUES
+ (1, '2024-01-10 00:00:00', 1001, 601, 20, 0, 1)
+ """
+ sql """
+ INSERT INTO mtmv_partition_lineage_perf_detail VALUES
+ (1, '2024-01-10 00:00:00', 1001, 1, 1.00, 60.00, 2.00, 3.00)
+ """
+ sql """
+ INSERT INTO mtmv_partition_lineage_perf_attr VALUES
+ (1, '2024-01-10 00:00:00', 30, 4, 1)
+ """
+
+ long createStartMs = System.currentTimeMillis()
+ sql """
+ CREATE MATERIALIZED VIEW mtmv_partition_lineage_perf_mv
+ BUILD DEFERRED REFRESH AUTO ON MANUAL
+ PARTITION BY (date_trunc(event_time, 'month'))
+ DISTRIBUTED BY RANDOM BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS
+ SELECT
+ profile_id,
+ event_time,
+ dim_id,
+ entity_id,
+ SUM(m01) AS m01,
+ SUM(m02) AS m02,
+ SUM(m03) AS m03,
+ SUM(m04) AS m04,
+ SUM(m05) AS m05,
+ SUM(m06) AS m06,
+ SUM(m07) AS m07,
+ SUM(m08) AS m08,
+ SUM(m09) AS m09,
+ SUM(m10) AS m10,
+ SUM(m11) AS m11,
+ SUM(m12) AS m12,
+ SUM(m13) AS m13,
+ SUM(m14) AS m14,
+ SUM(m15) AS m15,
+ SUM(m16) AS m16,
+ SUM(m17) AS m17,
+ SUM(m18) AS m18
+ FROM (
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 601 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (601, 701)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 602 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (602, 702)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 603 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (603, 703)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 604 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (604, 704)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 605 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (605, 705)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 606 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (606, 706)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 607 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (607, 707)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 608 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (608, 708)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 609 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (609, 709)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 610 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (610, 710)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 611 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (611, 711)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ UNION ALL
+ SELECT
+ f.profile_id,
+ f.event_time,
+ f.dim_id,
+ f.entity_id,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m01,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m02,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m03,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m04,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m05,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m06,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m07,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m08,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m09,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m10,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m11,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m12,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.measure_a ELSE 0 END) AS m13,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_a ELSE 0 END) AS m14,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_b ELSE 0 END) AS m15,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
d.value_f ELSE 0 END) AS m16,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_d ELSE 0 END) AS m17,
+ SUM(CASE WHEN h.doc_type = 612 AND a.active_flag = 1 THEN
f.value_e ELSE 0 END) AS m18
+ FROM mtmv_partition_lineage_perf_fact f
+ INNER JOIN mtmv_partition_lineage_perf_header h
+ ON f.profile_id = h.profile_id
+ AND f.event_time = h.event_time
+ AND f.doc_id = h.doc_id
+ INNER JOIN mtmv_partition_lineage_perf_detail d
+ ON f.profile_id = d.profile_id
+ AND f.event_time = d.event_time
+ AND f.doc_id = d.doc_id
+ AND f.line_id = d.line_id
+ INNER JOIN mtmv_partition_lineage_perf_attr a
+ ON f.profile_id = a.profile_id
+ AND f.event_time = a.event_time
+ AND f.attr_id = a.attr_id
+ WHERE h.status >= 20
+ AND h.doc_type IN (612, 712)
+ AND h.flag_value = 0
+ GROUP BY f.profile_id, f.event_time, f.dim_id, f.entity_id
+ ) union_src
+ GROUP BY profile_id, event_time, dim_id, entity_id
+ """
+
+ long createElapsedMs = System.currentTimeMillis() - createStartMs
+ logger.info("partition lineage MTMV create elapsed: ${createElapsedMs} ms")
+
+ def mvPartitions = sql """SHOW PARTITIONS FROM
mtmv_partition_lineage_perf_mv"""
+ logger.info("mtmv_partition_lineage_perf_mv partitions: " +
mvPartitions.toString())
+
assertTrue(mvPartitions.toString().contains("p_20240101000000_20240201000000")
+ || mvPartitions.toString().contains("p_20240101_20240201"))
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
