This is an automated email from the ASF dual-hosted git repository.
englefly 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 14ac5304b8e [fix](statistics) Create the table stats record when a
load finishes (#69028)
14ac5304b8e is described below
commit 14ac5304b8eaeefd2b69786d73605ec865d44d49
Author: minghong <[email protected]>
AuthorDate: Wed Oct 28 16:41:00 2026 +0800
[fix](statistics) Create the table stats record when a load finishes
(#69028)
An alternative implementation of the fix discussed in #68282, with a
much smaller change: instead of keeping
and resetting the statistics record across `TRUNCATE TABLE` (which also
had to order every record transition
with its journal entry), the record is recreated by the load itself, so
a truncation keeps removing it exactly
as it does today.
### What problem does this PR solve?
Issue Number: None (no upstream issue filed; the case is described
below)
Problem Summary:
Reproduction on a single bucket duplicate key table:
CREATE TABLE t (k INT, v INT) DUPLICATE KEY(k) DISTRIBUTED BY HASH(k)
BUCKETS 1;
INSERT INTO t VALUES (1,1),(2,2),(3,3),(4,4),(5,5);
TRUNCATE TABLE t;
INSERT INTO t VALUES (1,1),(2,2),(3,3);
EXPLAIN SELECT * FROM t; -- reports cardinality=1, although the table
holds 3 rows
TRUNCATE TABLE replaces the partitions and tablets of the table with new
ones and removes its statistics
record. The backends report the row count of the new tablets only after
the delay of
`tablet_stat_update_interval_second` (60s by default), and a partition
reports no row count at all until its
first load becomes visible (`DatabaseTransactionMgr` sets
`rowCountReported` to false for the first
transaction of a partition), so during that interval
`OlapTable.getRowCountForIndex(..., strict = true)`
returns -1 for the whole table. The only other source of a row count is
the statistics record of the table,
and the truncation had just removed it, so the planner clamps the table
to one row: joins are ordered and
distributed for a table that looks empty instead of for the rows that
were just loaded.
The load now recreates the record itself. While the loading transaction
becomes visible,
`AnalysisManager.replayUpdateRowsRecord()` creates the record of a table
which has none in the state of a
table whose row count has never been collected, and accumulates the rows
it reports into it, so the planner
reads them instead of -1. The record is created with a baseline of 0
(`updatedRowsBase`) and a collected row
count of 0 for the base index, so every row loaded afterwards is a delta
row and the reported row count is
the number of rows the table holds.
The creation happens in `updateUpdatedRows()`, which is the path the
master walks when the loading
transaction becomes visible
(`DatabaseTransactionMgr.updateCatalogAfterVisible`) and the path a
follower
walks while it replays the transaction state
(`replayUpsertTransactionState`), so every frontend ends up with
the record. It needs no edit log entry of its own and does not touch the
transaction state machine: the
record is a function of the transaction state the journal already
carries. The whole table `DROP STATS`, the
schema changes and the truncation keep removing the record as before,
and the next load recreates it. A
partition truncation is deliberately left out: the rows it reports are
the rows it removed, so it does not
create a record.
Two related fixes come with it, because the record is now the row count
source in more cases:
- the delta is only added to an index which keeps one row per base row
(`TableStatsMeta.keepsOneRowPerBaseRow`): an aggregating rollup, an
index which merges the rows of a unique
key table and a filtered materialized index have their own row count and
must not be reported with the
physical rows of the base index, and a row count which was never
collected is not combined with a delta;
- `StatisticsMetricCollector` decided "not analyzed" by the absence of a
record. A record now exists for a
loaded table which was never analyzed, so it decides by the analyze
timestamp instead.
### Known limitation
A record which is created for a table that already holds rows, for
instance right after a whole table
`DROP STATS` or after a schema change (the truncation case starts from
an empty table), only knows the rows
loaded after that point, so the table is estimated with them until the
backends report the tablets or an
analysis collects the row count. It is still closer than the unknown row
count it replaces: a merge key
table reports the physical row count in that window, which is an upper
bound of its logical row count, as it
does in the steady state today.
### Release note
None
### Check List (For Author)
- Test: new unit tests in `AnalysisManagerTest` (a load creates the
record and accumulates its rows, a load
which reports the rows by table id creates it as well, a replayed load
creates the same record, an existing
analyzed record is only accumulated, and an unresolvable table creates
none) and in `TableStatsMetaTest`
(the record of a table whose row count was never collected reports the
loaded rows; the baseline survives
dropping the column statistics; a legacy record derives it from the
column statistics; a stale analyze
snapshot and its baseline cancel out; a user injected row count is not
counted twice; the reader of the
snapshot never pairs two states; a row preserving rollup reports the
loaded rows while an aggregating one
and a filtered one do not). `AnalysisManagerTest` 21/21,
`TableStatsMetaTest` 12/12 (checkstyle not
skipped). Regression: the new suite `test_row_count_after_truncate` (the
truncate case, an aggregate rollup
and a projection rollup) and the existing `test_drop_stats_and_truncate`
both pass. Manual test on a local
FE and BE cluster with the reproduction above: the scan reports
cardinality=3 right after the load instead
of 1, a never analyzed table which receives 4 rows reports 4 instead of
1, and `SHOW TABLE STATS` shows the
accumulated rows with an empty analyze time.
- Behavior changed: Yes. A table whose statistics record was removed
reports the rows its loads accumulate
instead of an unknown row count (planned as one row) while the backends
have not reported the new tablets
yet; the "not analyzed" metric counts a table by its analyze timestamp.
- Does this need documentation: No
---
.../org/apache/doris/alter/SchemaChangeJobV2.java | 29 +-
.../apache/doris/catalog/CatalogRecycleBin.java | 9 +
.../transaction/CloudGlobalTransactionMgr.java | 11 +-
.../doris/nereids/stats/StatsCalculator.java | 20 +-
.../plans/commands/ShowTableStatsCommand.java | 27 +-
.../doris/statistics/analysis/AnalysisManager.java | 251 +++++++++---
.../analysis/StatisticsAutoCollector.java | 4 +-
.../analysis/StatisticsMetricCollector.java | 6 +-
.../doris/statistics/analysis/TableStatsMeta.java | 150 ++++++-
.../doris/statistics/analysis/UpdateRowsEvent.java | 11 +-
.../statistics/repository/StatisticsCleaner.java | 38 ++
.../doris/alter/SchemaChangeJobV2StatsTest.java | 59 +++
.../doris/nereids/stats/StatsCalculatorTest.java | 46 +++
.../statistics/analysis/AnalysisManagerTest.java | 436 +++++++++++++++++++++
.../statistics/analysis/TableStatsMetaTest.java | 288 ++++++++++++++
.../repository/StatisticsCleanerTest.java | 84 ++++
.../query_p0/eager_agg/scalar_agg_pushdown.out | 8 +-
.../data/query_p0/hint/multi_leading.out | 2 +-
.../data/query_p0/runtimefilterV2/rfv2.out | 18 +-
.../tpcds_sf100/noStatsRfPrune/query24.out | 28 +-
.../mtmv_p0/ivm/test_ivm_explain_refresh.groovy | 32 +-
...lated_scalar_subquery_to_window_function.groovy | 4 +
.../query_p0/eager_agg/scalar_agg_pushdown.groovy | 3 +
.../tpcds_sf100/noStatsRfPrune/query24.groovy | 3 +
.../test_row_count_after_truncate.groovy | 131 +++++++
25 files changed, 1594 insertions(+), 104 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java
b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java
index b88a394ef27..7d98c8b7916 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java
@@ -56,7 +56,6 @@ import org.apache.doris.nereids.trees.expressions.Cast;
import org.apache.doris.nereids.types.DataType;
import org.apache.doris.persist.gson.GsonPostProcessable;
import org.apache.doris.persist.gson.GsonUtils;
-import org.apache.doris.statistics.analysis.AnalysisManager;
import org.apache.doris.task.AgentBatchTask;
import org.apache.doris.task.AgentTask;
import org.apache.doris.task.AgentTaskExecutor;
@@ -70,6 +69,7 @@ import org.apache.doris.thrift.TTaskType;
import org.apache.doris.transaction.GlobalTransactionMgr;
import org.apache.doris.transaction.TransactionState;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.HashBasedTable;
@@ -728,18 +728,39 @@ public class SchemaChangeJobV2 extends AlterJobV2
implements GsonPostProcessable
Env.getCurrentEnv().getEditLog().logAlterJob(this);
this.showJobState = JobState.FINISHED;
pruneMeta();
+ // The record is removed with the journal entry of FINISHED, i.e.
still under the table write lock:
+ // a load which becomes visible in between updates (or recreates)
the record, and this removal would
+ // follow it on the master, while a follower replays FINISHED
(removing the record) before VISIBLE
+ // (recreating it with the rows of that load). The master would
then have no record while its
+ // followers kept one, and the row count of the table would differ
between them.
+ if (!FeConstants.runningUnitTest) {
+ removeTableStatsOnFinished(tbl);
+ }
} finally {
tbl.writeUnlock();
}
postProcessOriginIndex();
// Drop table column stats after schema change finished.
if (!FeConstants.runningUnitTest) {
- AnalysisManager manager = Env.getCurrentEnv().getAnalysisManager();
- manager.removeTableStats(tbl.getId());
- manager.dropStats(tbl, null);
+ Env.getCurrentEnv().getAnalysisManager().dropStats(tbl, null);
}
}
+ /**
+ * Remove the statistics record of the table when the schema change
finished.
+ *
+ * <p>It must be called with the table write lock held, in the same
critical section which journals FINISHED,
+ * so that a load cannot become visible in between: such a load updates or
recreates the record, and a
+ * follower replays the journal order "FINISHED (remove the record), then
VISIBLE (recreate it)", so the
+ * master and its followers would disagree about the row count of the
table.
+ */
+ @VisibleForTesting
+ void removeTableStatsOnFinished(OlapTable tbl) {
+ Preconditions.checkState(tbl.isWriteLockHeldByCurrentThread(),
+ "the statistics of table %s must be removed under its write
lock", tbl.getName());
+ Env.getCurrentEnv().getAnalysisManager().removeTableStats(tbl.getId());
+ }
+
private void onFinished(OlapTable tbl) {
// replace the origin index with shadow index, set index state as
NORMAL
for (Partition partition : tbl.getPartitions()) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java
index dcea067295c..cdeee0fb6ee 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java
@@ -275,6 +275,15 @@ public class CatalogRecycleBin extends MasterDaemon
implements Writable {
return idToRecycleTime.get(id);
}
+ /**
+ * The table of a recycle bin entry. It keeps the id of the dropped table
and is the instance which
+ * {@code RECOVER TABLE} registers again, so its lock is the lock of the
recovered table as well.
+ */
+ public Table getRecycledTable(long tableId) {
+ RecycleTableInfo tableInfo = idToTable.get(tableId);
+ return tableInfo == null ? null : tableInfo.getTable();
+ }
+
public void setRecycleTimeByIdForReplay(long id, Long recycleTime) {
idToRecycleTime.put(id, recycleTime);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
index ba5afbc2f6c..624d41df410 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
@@ -536,6 +536,15 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
long dbId = commitTxnResponse.getTxnInfo().getDbId();
long txnId = commitTxnResponse.getTxnInfo().getTxnId();
// 1. update rowCountfor AnalysisManager
+ // The partitions the rows were loaded into are handed over as well: a
report can arrive after those
+ // partitions are gone (a stream load queues its statistics report and
can return before the frontend
+ // received it, and a whole table TRUNCATE replaces the partitions
while it keeps the table id), and the
+ // rows of such a report are the rows which were removed.
+ Map<Long, List<Long>> reportedPartitions = Maps.newHashMap();
+ for (int idx = 0; idx < commitTxnResponse.getPartitionIdsCount();
++idx) {
+
reportedPartitions.computeIfAbsent(commitTxnResponse.getTableIds(idx), tableId
-> Lists.newArrayList())
+ .add(commitTxnResponse.getPartitionIds(idx));
+ }
Map<Long, Long> updatedRows = new HashMap<>();
for (TableStatsPB tableStats : commitTxnResponse.getTableStatsList()) {
if (tableStats.hasUpdatedRowCount()) {
@@ -545,7 +554,7 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
}
}
Env env = Env.getCurrentEnv();
- env.getAnalysisManager().updateUpdatedRows(updatedRows);
+ env.getAnalysisManager().updateUpdatedRows(updatedRows, dbId,
reportedPartitions);
// 2. update table and partition version
Map<Long, List<Long>> tablePartitionMap =
updateVersion(commitTxnResponse);
// 3. notify partition first load
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
index e1482c1bdcd..e957907fd62 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
@@ -150,6 +150,7 @@ import org.apache.doris.statistics.model.Statistics;
import org.apache.doris.statistics.model.StatisticsBuilder;
import org.apache.doris.statistics.util.StatisticsUtil;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
@@ -381,14 +382,21 @@ public class StatsCalculator extends
DefaultPlanVisitor<Statistics, Void> {
}
/**
- * returns the sum of deltaRowCount for all selected partitions or for the
table.
+ * The rows loaded into the base index which the collected column
statistics don't describe yet, i.e. the
+ * population {@link FilterEstimation#estimate} retries a zero estimate
with. It is deliberately not
+ * restricted to an index which keeps one row per base row: such an index
must not add the delta to its own
+ * row count ({@link TableStatsMeta#getRowCountWithDeltaRows} gates that),
but stale column statistics
+ * affect every index in the same way, and for an index which aggregates
the rows the delta is an upper
+ * bound of the rows it gained.
*/
- private long computeDeltaRowCount(OlapScan olapScan) {
+ @VisibleForTesting
+ long computeDeltaRowCount(OlapScan olapScan) {
+ OlapTable olapTable = olapScan.getTable();
AnalysisManager analysisManager =
Env.getCurrentEnv().getAnalysisManager();
- TableStatsMeta tableMeta =
analysisManager.findTableStatsStatus(olapScan.getTable().getId());
+ TableStatsMeta tableMeta =
analysisManager.findTableStatsStatus(olapTable.getId());
long deltaRowCount = 0;
if (tableMeta != null) {
- deltaRowCount =
tableMeta.getBaseIndexDeltaRowCount(olapScan.getTable());
+ deltaRowCount = tableMeta.getBaseIndexDeltaRowCount(olapTable);
}
return deltaRowCount;
}
@@ -471,7 +479,9 @@ public class StatsCalculator extends
DefaultPlanVisitor<Statistics, Void> {
rowCount =
olapTable.getRowCountForIndex(olapScan.getSelectedIndexId(), true);
if (rowCount == -1) {
if (tableMeta != null) {
- rowCount =
tableMeta.getRowCount(olapScan.getSelectedIndexId()) +
computeDeltaRowCount(olapScan);
+ // The collected row count and the rows loaded since it
was collected are read as one
+ // snapshot.
+ rowCount = tableMeta.getRowCountWithDeltaRows(olapTable,
olapScan.getSelectedIndexId());
}
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowTableStatsCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowTableStatsCommand.java
index 8cb4dc13711..7d265c57ca4 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowTableStatsCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowTableStatsCommand.java
@@ -242,28 +242,37 @@ public class ShowTableStatsCommand extends ShowCommand {
row.add(String.valueOf(tableStatistic.updatedRows));
row.add(String.valueOf(tableStatistic.queriedTimes.get()));
row.add(String.valueOf(tableStatistic.rowCount));
- LocalDateTime dateTime =
-
LocalDateTime.ofInstant(Instant.ofEpochMilli(tableStatistic.updatedTime),
- java.time.ZoneId.systemDefault());
- LocalDateTime lastAnalyzeTime =
-
LocalDateTime.ofInstant(Instant.ofEpochMilli(tableStatistic.lastAnalyzeTime),
- java.time.ZoneId.systemDefault());
- row.add(dateTime.format(formatter));
+ row.add(formatStatsTime(tableStatistic.updatedTime, formatter));
Set<Pair<String, String>> columnsSet = tableStatistic.analyzeColumns();
Set<Pair<String, String>> newColumnsSet = new HashSet<>();
for (Pair<String, String> pair : columnsSet) {
newColumnsSet.add(Pair.of(Util.getTempTableDisplayName(pair.first),
pair.second));
}
row.add(newColumnsSet.toString());
- row.add(tableStatistic.jobType.toString());
+ // A record which no analyze job created, for instance the record a
load creates for a table whose
+ // statistics were dropped, has no trigger to show.
+ row.add(tableStatistic.jobType == null ? "" :
tableStatistic.jobType.toString());
row.add(String.valueOf(tableStatistic.partitionChanged.get()));
row.add(String.valueOf(tableStatistic.userInjected));
row.add(table == null ? "N/A" :
String.valueOf(table.autoAnalyzeEnabled()));
- row.add(lastAnalyzeTime.format(formatter));
+ row.add(formatStatsTime(tableStatistic.lastAnalyzeTime, formatter));
result.add(row);
return new ShowResultSet(getMetaData(), result);
}
+ /**
+ * A stats record may exist for a table which has never been analyzed, for
instance the record a load
+ * creates for a table whose statistics were dropped. There is no
timestamp to display for those records.
+ */
+ private static String formatStatsTime(long timeInMs, DateTimeFormatter
formatter) {
+ if (timeInMs == 0) {
+ return "";
+ }
+ LocalDateTime dateTime =
LocalDateTime.ofInstant(Instant.ofEpochMilli(timeInMs),
+ java.time.ZoneId.systemDefault());
+ return dateTime.format(formatter);
+ }
+
/**
* constructPartitionResultSet
*/
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/AnalysisManager.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/AnalysisManager.java
index 914ac3afbbb..1fe368d480f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/AnalysisManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/AnalysisManager.java
@@ -40,6 +40,7 @@ import org.apache.doris.common.Pair;
import org.apache.doris.common.ThreadPoolManager;
import org.apache.doris.common.ThreadPoolManager.BlockedPolicy;
import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.MetaLockUtils;
import org.apache.doris.common.util.Util;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.InternalCatalog;
@@ -758,8 +759,7 @@ public class AnalysisManager implements Writable {
// Remove tableMetaStats if drop whole table stats.
if ((cols == null || cols.isEmpty()) && (!table.isPartitionedTable()
|| partitionNamesInfo == null
|| partitionNamesInfo.isStar() ||
partitionNamesInfo.getPartitionNames() == null)) {
- removeTableStats(tblId);
- Env.getCurrentEnv().getEditLog().logDeleteTableStats(new
TableStatsDeletionLog(tblId));
+ removeTableStatsAndLog(table);
}
invalidateLocalStats(catalogId, dbId, tblId, cols, tableStats,
partitionNamesInfo);
// Drop stats ddl is master only operation.
@@ -1276,6 +1276,53 @@ public class AnalysisManager implements Writable {
return idToTblStats.get(tblId);
}
+ /**
+ * The record of the table, created in the state of a table whose row
count has never been collected when
+ * it doesn't exist yet. A truncation, a whole table DROP STATS and a
schema change remove the record, so
+ * the rows which are loaded afterwards can only be turned into the row
count of the table while the
+ * record exists: the planner falls back to it exactly while the backends
have not reported the row count
+ * of the new tablets yet, which is the interval right after those
operations.
+ *
+ * <p>Loads run this from the transaction visibility of the load, both on
the master
+ * ({@code DatabaseTransactionMgr.updateCatalogAfterVisible}) and while a
follower replays the transaction
+ * state, so the record exists on every frontend without a journal entry
of its own.
+ *
+ * @return the existing or the created record, or null when the table
cannot be resolved any more
+ */
+ private TableStatsMeta findOrCreateTableStats(InternalCatalog catalog,
long dbId, long tableId) {
+ TableStatsMeta tableStats = idToTblStats.get(tableId);
+ if (tableStats != null) {
+ return tableStats;
+ }
+ Optional<Database> dbOption = catalog.getDb(dbId);
+ if (!dbOption.isPresent()) {
+ return null;
+ }
+ Optional<Table> tableOption = dbOption.get().getTable(tableId);
+ if (!tableOption.isPresent() || !(tableOption.get() instanceof
OlapTable)) {
+ return null;
+ }
+ return findOrCreateTableStats((OlapTable) tableOption.get());
+ }
+
+ private TableStatsMeta findOrCreateTableStats(OlapTable table) {
+ TableStatsMeta tableStats = idToTblStats.get(table.getId());
+ if (tableStats != null) {
+ return tableStats;
+ }
+ TableStatsMeta created = new TableStatsMeta(table);
+ // The insertion is published under the monitor {@link
#writeTableStats} holds while it writes the
+ // number of records and then the records themselves. A record which
is inserted in between would be
+ // written although the declared number excludes it, and the readers,
which trust that number, would
+ // take the extra record as the header of the next module of the
image. The creation stays outside the
+ // monitor: it only reads catalog metadata and must not be ordered
with the catalog locks which the
+ // image dump holds before it takes this monitor.
+ synchronized (idToTblStats) {
+ TableStatsMeta existing = idToTblStats.putIfAbsent(table.getId(),
created);
+ return existing == null ? created : existing;
+ }
+ }
+
// Invoke this when load transaction finished.
public void updateUpdatedRows(Map<Long, Map<Long, Long>> tabletRecords,
long dbId, long txnId) {
try {
@@ -1298,17 +1345,93 @@ public class AnalysisManager implements Writable {
}
// Invoke this for cloud version load.
- public void updateUpdatedRows(Map<Long, Long> updatedRows) {
+ public void updateUpdatedRows(Map<Long, Long> updatedRows, long dbId) {
+ updateUpdatedRows(updatedRows, dbId, Collections.emptyMap());
+ }
+
+ /**
+ * Apply and journal the rows a cloud load reported.
+ *
+ * @param reportedPartitions the partitions each table's rows were loaded
into, or an empty map when the
+ * caller has no partition information
+ */
+ public void updateUpdatedRows(Map<Long, Long> updatedRows, long dbId,
+ Map<Long, List<Long>> reportedPartitions) {
+ if (!Env.getCurrentEnv().isMaster() || Env.isCheckpointThread()) {
+ return;
+ }
+ // The record of a table which has none is created while this event is
applied, and the event is
+ // journaled afterwards. A whole table DROP STATS removes the record
and journals the deletion, so both
+ // mutations have to be applied in the order their journal entries
have: without the table lock the
+ // deletion can be applied (and journaled) between the application of
this event and its journal entry,
+ // and a follower which replays "remove, then add" keeps a record
which the master no longer has, which
+ // makes the row count of the table differ between the master and its
followers. The lock is the one
+ // DROP STATS takes and the one finishTransaction holds while a local
load recreates the record.
+ List<Table> tableList = getTablesOnIdOrder(dbId, updatedRows.keySet());
+ if (!MetaLockUtils.tryWriteLockTablesIfExist(tableList, 10,
TimeUnit.SECONDS)) {
+ LOG.warn("Failed to lock the tables {} of db {} to record the
updated rows {}, the row counts stay "
+ + "unknown until the backends report them",
updatedRows.keySet(), dbId, updatedRows);
+ return;
+ }
try {
- if (!Env.getCurrentEnv().isMaster() || Env.isCheckpointThread()) {
+ // A report can arrive after the partitions it counted are gone: a
stream load queues the report of
+ // the rows it loaded (`send_stats_to_fe_async`) and can return
before the frontend received it, and
+ // a whole table TRUNCATE replaces the partitions while it keeps
the id of the table. The rows of
+ // such a report are the rows which were removed, so they are
neither applied nor journaled: the
+ // record which the next load recreates would otherwise report a
table which holds 3 rows as if it
+ // held 103 rows, until the backends report the new tablets.
+ Map<Long, Long> recordedRows = new HashMap<>(updatedRows);
+ recordedRows.keySet().removeIf(tableId ->
!reportCountsCurrentPartitions(dbId, tableId,
+ reportedPartitions.getOrDefault(tableId,
Collections.emptyList())));
+ if (recordedRows.isEmpty()) {
return;
}
- UpdateRowsEvent updateRowsEvent = new UpdateRowsEvent(updatedRows);
+ UpdateRowsEvent updateRowsEvent = new UpdateRowsEvent(dbId,
recordedRows);
replayUpdateRowsRecord(updateRowsEvent);
logUpdateRowsRecord(updateRowsEvent);
} catch (Throwable t) {
LOG.warn("Failed to record update rows.", t);
+ } finally {
+ MetaLockUtils.writeUnlockTables(tableList);
+ }
+ }
+
+ /**
+ * Whether the partitions which a cloud row count report counted still
belong to the table.
+ *
+ * <p>A stream load queues the report of the rows it loaded and can return
before the frontend received it,
+ * so a report can arrive after the partitions it counted are gone: a
whole table TRUNCATE replaces the
+ * partitions of the table and keeps its id, so the rows of such a report
are the rows the truncation
+ * removed and must not be accumulated into the record which the next load
recreates.
+ */
+ @VisibleForTesting
+ static boolean reportCountsCurrentPartitions(OlapTable table,
Collection<Long> reportedPartitionIds) {
+ for (long partitionId : reportedPartitionIds) {
+ if (table.getPartition(partitionId) == null) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean reportCountsCurrentPartitions(long dbId, long
tableId,
+ Collection<Long> reportedPartitionIds) {
+ if (reportedPartitionIds.isEmpty()) {
+ return true;
}
+ return Env.getCurrentInternalCatalog().getDb(dbId)
+ .flatMap(db -> db.getTable(tableId))
+ .filter(table -> table instanceof OlapTable)
+ .map(table -> reportCountsCurrentPartitions((OlapTable) table,
reportedPartitionIds))
+ .orElse(true);
+ }
+
+ private List<Table> getTablesOnIdOrder(long dbId, Collection<Long>
tableIds) {
+ Optional<Database> db = Env.getCurrentInternalCatalog().getDb(dbId);
+ if (!db.isPresent()) {
+ return Lists.newArrayList();
+ }
+ return
db.get().getTablesOnIdOrderIfExist(Lists.newArrayList(tableIds));
}
// Set to true means new partition loaded data
@@ -1355,7 +1478,7 @@ public class AnalysisManager implements Writable {
InternalCatalog catalog = Env.getCurrentInternalCatalog();
if (event.getRecords() != null) {
for (Entry<Long, Long> record : event.getRecords().entrySet()) {
- TableStatsMeta statsStatus = idToTblStats.get(record.getKey());
+ TableStatsMeta statsStatus = findOrCreateTableStats(catalog,
event.getDbId(), record.getKey());
if (statsStatus != null) {
statsStatus.updatedRows.addAndGet(record.getValue());
}
@@ -1366,52 +1489,52 @@ public class AnalysisManager implements Writable {
// Record : TableId -> (TabletId -> update rows)
if (event.getTabletRecords() != null) {
for (Entry<Long, Map<Long, Long>> record :
event.getTabletRecords().entrySet()) {
- TableStatsMeta statsStatus = idToTblStats.get(record.getKey());
- if (statsStatus != null) {
- Optional<Database> dbOption =
catalog.getDb(event.getDbId());
- if (!dbOption.isPresent()) {
- LOG.warn("Database {} does not exist.",
event.getDbId());
- continue;
- }
- Database database = dbOption.get();
- Optional<Table> tableOption =
database.getTable(record.getKey());
- if (!tableOption.isPresent()) {
- LOG.warn("Table {} does not exist in DB {}.",
record.getKey(), event.getDbId());
- continue;
- }
- Table table = tableOption.get();
- if (!(table instanceof OlapTable)) {
- continue;
- }
- OlapTable olapTable = (OlapTable) table;
- short replicaNum =
olapTable.getTableProperty().getReplicaAllocation().getTotalReplicaNum();
- Map<Long, Long> tabletRows = record.getValue();
- if (tabletRows == null || tabletRows.isEmpty()) {
- LOG.info("Tablet row count map is empty");
- continue;
- }
- long rowsForAllReplica = 0;
- for (Entry<Long, Long> entry : tabletRows.entrySet()) {
- rowsForAllReplica += entry.getValue();
- if (LOG.isDebugEnabled()) {
- LOG.debug("Table id {}, tablet id {}, row count
{}",
- record.getKey(), entry.getKey(),
entry.getValue());
- }
- }
- long tableUpdateRows = rowsForAllReplica / replicaNum;
- LOG.info("Update rows for table {} is {}, replicaNum is
{}, "
- + "rows for all replica {}, tablets count {}",
- olapTable.getName(), tableUpdateRows, replicaNum,
rowsForAllReplica, tabletRows.size());
- statsStatus.updatedRows.addAndGet(tableUpdateRows);
- if (StatisticsUtil.enablePartitionAnalyze()) {
- updatePartitionRows(olapTable, tabletRows,
statsStatus, replicaNum);
+ Optional<Database> dbOption = catalog.getDb(event.getDbId());
+ if (!dbOption.isPresent()) {
+ LOG.warn("Database {} does not exist.", event.getDbId());
+ continue;
+ }
+ Database database = dbOption.get();
+ Optional<Table> tableOption =
database.getTable(record.getKey());
+ if (!tableOption.isPresent()) {
+ LOG.warn("Table {} does not exist in DB {}.",
record.getKey(), event.getDbId());
+ continue;
+ }
+ Table table = tableOption.get();
+ if (!(table instanceof OlapTable)) {
+ continue;
+ }
+ OlapTable olapTable = (OlapTable) table;
+ TableStatsMeta statsStatus = findOrCreateTableStats(olapTable);
+ short replicaNum =
olapTable.getTableProperty().getReplicaAllocation().getTotalReplicaNum();
+ Map<Long, Long> tabletRows = record.getValue();
+ if (tabletRows == null || tabletRows.isEmpty()) {
+ LOG.info("Tablet row count map is empty");
+ continue;
+ }
+ long rowsForAllReplica = 0;
+ for (Entry<Long, Long> entry : tabletRows.entrySet()) {
+ rowsForAllReplica += entry.getValue();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Table id {}, tablet id {}, row count {}",
+ record.getKey(), entry.getKey(),
entry.getValue());
}
}
+ long tableUpdateRows = rowsForAllReplica / replicaNum;
+ LOG.info("Update rows for table {} is {}, replicaNum is {}, "
+ + "rows for all replica {}, tablets count {}",
+ olapTable.getName(), tableUpdateRows, replicaNum,
rowsForAllReplica, tabletRows.size());
+ statsStatus.updatedRows.addAndGet(tableUpdateRows);
+ if (StatisticsUtil.enablePartitionAnalyze()) {
+ updatePartitionRows(olapTable, tabletRows, statsStatus,
replicaNum);
+ }
}
return;
}
- // Handle truncate table
+ // Handle truncate table. A partition truncation is not a load: the
rows it reports are the rows it
+ // removed, so it must not create a record for a table which has none,
and it can only be accumulated
+ // into a record which was created by a load or an analyze.
if (event.getPartitionToUpdateRows() != null && event.getTableId() >
0) {
Map<Long, Long> partRows = event.getPartitionToUpdateRows();
long totalRows = partRows.values().stream().mapToLong(rows ->
rows).sum();
@@ -1487,6 +1610,42 @@ public class AnalysisManager implements Writable {
}
}
+ /**
+ * Remove the statistics record of the table and journal the deletion,
under the table write lock.
+ *
+ * <p>A load recreates the record of a table which has none while its
transaction becomes visible: the
+ * transaction is journaled as {@code VISIBLE} and the record is created by
+ * {@code DatabaseTransactionMgr#updateCatalogAfterVisible} afterwards,
both under the table write lock
+ * which {@code finishTransaction} holds until that method returns. A
deletion which is applied in
+ * between is replayed by the followers after the creation (the journal
order is "transaction, then
+ * deletion") while the master applied it before the creation, so the
master keeps the record the load
+ * created and every follower, and every image built from a follower, ends
up without it. Removing and
+ * journaling the deletion under the same lock orders it either before the
transaction was journaled or
+ * after the record was created, which is the order the journal has as
well, so all frontends agree.
+ *
+ * <p>The lock is deliberately held across the journal write: the record
creation of a load which is
+ * already past its own journal write must not slip in between the removal
and its journal entry.
+ */
+ public void removeTableStatsAndLog(TableIf table) {
+ if (!(table instanceof OlapTable)) {
+ // Only a load into an OLAP table recreates the record of a table
which has none, there is no
+ // creation to be ordered with.
+ removeTableStatsAndLog(table.getId());
+ return;
+ }
+ table.writeLock();
+ try {
+ removeTableStatsAndLog(table.getId());
+ } finally {
+ table.writeUnlock();
+ }
+ }
+
+ private void removeTableStatsAndLog(long tableId) {
+ removeTableStats(tableId);
+ Env.getCurrentEnv().getEditLog().logDeleteTableStats(new
TableStatsDeletionLog(tableId));
+ }
+
public Set<Long> getIdToTblStatsKeys() {
return new HashSet<>(idToTblStats.keySet());
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/StatisticsAutoCollector.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/StatisticsAutoCollector.java
index 52ec6d2bd7e..86101fe588a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/StatisticsAutoCollector.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/StatisticsAutoCollector.java
@@ -27,7 +27,6 @@ import org.apache.doris.common.DdlException;
import org.apache.doris.common.Pair;
import org.apache.doris.common.util.MasterDaemon;
import org.apache.doris.datasource.plugin.PluginDrivenExternalTable;
-import org.apache.doris.persist.TableStatsDeletionLog;
import org.apache.doris.statistics.StatisticConstants;
import org.apache.doris.statistics.analysis.AnalysisInfo.AnalysisMethod;
import org.apache.doris.statistics.analysis.AnalysisInfo.JobType;
@@ -210,8 +209,7 @@ public class StatisticsAutoCollector extends MasterDaemon {
LOG.info("Table {} is empty, remove its old stats and skip auto
analyze it.", table.getName());
// Remove the table's old stats if exists.
if (tableStatsStatus != null &&
!tableStatsStatus.isColumnsStatsEmpty()) {
- manager.removeTableStats(table.getId());
- Env.getCurrentEnv().getEditLog().logDeleteTableStats(new
TableStatsDeletionLog(table.getId()));
+ manager.removeTableStatsAndLog(table);
manager.dropStats(table, null);
}
return false;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/StatisticsMetricCollector.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/StatisticsMetricCollector.java
index 23d3532ad91..8f6bdb9fc06 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/StatisticsMetricCollector.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/StatisticsMetricCollector.java
@@ -86,7 +86,11 @@ public class StatisticsMetricCollector extends MasterDaemon {
if (table.getRowCount() == 0) {
tmpEmptyTableCount += 1;
}
- if
(analysisManager.findTableStatsStatus(table.getId()) == null) {
+ // A record exists for a table which was never
analyzed as well, for instance the record
+ // a load creates for a table whose statistics were
dropped. Only a record with an
+ // analyze timestamp describes collected statistics.
+ TableStatsMeta tableStatsStatus =
analysisManager.findTableStatsStatus(table.getId());
+ if (tableStatsStatus == null ||
tableStatsStatus.updatedTime == 0) {
tmpNotAnalyzedTableCount += 1;
}
// Get all unhealthy columns.
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/TableStatsMeta.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/TableStatsMeta.java
index 99a5285834d..8c0bb2e1e02 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/TableStatsMeta.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/TableStatsMeta.java
@@ -17,7 +17,10 @@
package org.apache.doris.statistics.analysis;
+import org.apache.doris.catalog.AggregateType;
import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.MaterializedIndexMeta;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.Pair;
@@ -105,6 +108,14 @@ public class TableStatsMeta implements Writable,
GsonPostProcessable {
@SerializedName("irc")
private ConcurrentMap<Long, Long> indexesRowCount = new
ConcurrentHashMap<>();
+ // The value of updatedRows when indexesRowCount was collected, i.e. the
number of rows the collected row
+ // count already includes. The rows loaded after that point are the delta
row count of the table. It is
+ // kept here, and not derived from colToColStatsMeta, so that it survives
dropping the column statistics
+ // and so that a record which has never been analyzed has a baseline of
its own. -1 means no row count has
+ // ever been collected from the table, the baseline is then derived from
the column statistics.
+ @SerializedName("updatedRowsBase")
+ private final AtomicLong updatedRowsBase = new AtomicLong(-1);
+
@VisibleForTesting
public TableStatsMeta() {
ctlId = 0;
@@ -130,6 +141,68 @@ public class TableStatsMeta implements Writable,
GsonPostProcessable {
update(analyzedJob, table);
}
+ /**
+ * Create the record of a table which doesn't have one yet, in the state
of a table whose row count has
+ * never been collected. The rows loaded into the table are accumulated by
+ * {@link AnalysisManager#replayUpdateRowsRecord} while the record exists,
so a record which a truncation,
+ * a DROP STATS or a schema change removed has to be recreated by the next
load, otherwise these rows can
+ * never be turned into a row count: the planner sees them neither in the
collected row count nor in the
+ * row count the backends report for the new tablets during the reporting
interval.
+ */
+ public TableStatsMeta(OlapTable table) {
+ this.ctlId = table.getDatabase().getCatalog().getId();
+ this.ctlName = table.getDatabase().getCatalog().getName();
+ this.dbId = table.getDatabase().getId();
+ this.dbName = table.getDatabase().getFullName();
+ this.tblId = table.getId();
+ this.tblName = table.getName();
+ this.idxId = -1;
+ // Nothing has been collected for this table, and every index whose
row count follows the base index
+ // is known to be empty: the rows loaded from now on are all delta
rows.
+ this.indexesRowCount = buildEmptyIndexRowCount(table);
+ this.updatedRowsBase.set(0);
+ }
+
+ private static ConcurrentMap<Long, Long> buildEmptyIndexRowCount(OlapTable
table) {
+ ConcurrentMap<Long, Long> indexRowCount = new ConcurrentHashMap<>();
+ for (Long indexId : table.getIndexIdList()) {
+ if (keepsOneRowPerBaseRow(table, indexId)) {
+ indexRowCount.put(indexId, 0L);
+ }
+ }
+ return indexRowCount;
+ }
+
+ /**
+ * Whether the rows loaded into the base index are the rows of this index
as well. That holds for the base
+ * index itself and for an index which keeps one row per base row, i.e. a
duplicate key index whose columns
+ * are all plain. An index which aggregates, or which merges the rows of a
unique key table, has a smaller
+ * row count of its own, so the rows loaded into the base index must not
be added to it.
+ */
+ public static boolean keepsOneRowPerBaseRow(OlapTable table, long indexId)
{
+ if (indexId == table.getBaseIndexId()) {
+ return true;
+ }
+ MaterializedIndexMeta indexMeta = table.getIndexMetaByIndexId(indexId);
+ if (indexMeta == null || indexMeta.getKeysType() != KeysType.DUP_KEYS)
{
+ return false;
+ }
+ if (indexMeta.getWhereClause() != null) {
+ // A filtered materialized index only holds the rows which satisfy
its predicate, it does not keep
+ // one row per base row even though it inherits the key type of
the base table.
+ return false;
+ }
+ for (Column column : indexMeta.getSchema()) {
+ // A key column has no aggregation type at all, the value columns
of a duplicate key index are
+ // NONE. Neither of them merges rows, only an aggregating column
does.
+ AggregateType aggregationType = column.getAggregationType();
+ if (aggregationType != null && aggregationType !=
AggregateType.NONE) {
+ return false;
+ }
+ }
+ return true;
+ }
+
@Override
public void write(DataOutput out) throws IOException {
String json = GsonUtils.GSON.toJson(this);
@@ -161,7 +234,13 @@ public class TableStatsMeta implements Writable,
GsonPostProcessable {
return colToColStatsMeta.keySet();
}
- public void update(AnalysisInfo analyzedJob, TableIf tableIf) {
+ /**
+ * Apply the outcome of an analyze job to this record. The transition runs
under the monitor of this
+ * record, which {@link #getRowCountWithDeltaRows} also takes, so that a
planner reading the row count of
+ * the table without holding its lock doesn't pair the collected row count
of this analysis with the
+ * baseline of another one.
+ */
+ public synchronized void update(AnalysisInfo analyzedJob, TableIf tableIf)
{
updatedTime = analyzedJob.tblUpdateTime;
lastAnalyzeTime = analyzedJob.createTime;
if (analyzedJob.userInject) {
@@ -194,6 +273,21 @@ public class TableStatsMeta implements Writable,
GsonPostProcessable {
if (tableIf != null) {
if (tableIf instanceof OlapTable) {
OlapTable olapTable = (OlapTable) tableIf;
+ // The collected row counts published below already include
the rows which had been loaded when
+ // the job was built, remember how many they were, they are
not delta rows. The baseline may
+ // only advance together with the collected base index row
count, an analysis of another index
+ // (a materialized view) doesn't touch it.
+ // NOTICE: the baseline is published before the collected
counts. They are both read without
+ // the table lock by the planner, and a reader which pairs the
collected count of this analysis
+ // with the baseline of the previous one counts the rows
loaded since that previous analysis
+ // twice.
+ if
(analyzedJob.indexesRowCount.containsKey(olapTable.getBaseIndexId())) {
+ // A row count supplied by the user replaces the collected
one and describes the table as
+ // of now, so the rows loaded so far are part of it and
the baseline has to move to the
+ // current value. Keeping the older baseline would count
those rows again once DROP STATS
+ // clears userInjected.
+ updatedRowsBase.set(analyzedJob.userInject ?
updatedRows.get() : analyzedJob.updateRows);
+ }
indexesRowCount.putAll(analyzedJob.indexesRowCount);
clearStaleIndexRowCount(olapTable);
if (analyzedJob.jobColumns.containsAll(
@@ -254,8 +348,28 @@ public class TableStatsMeta implements Writable,
GsonPostProcessable {
indexesRowCount.put(indexId, rowCount);
}
+ // For unit test only. Simulate a record written before updatedRowsBase
was recorded in the table stats.
+ protected void clearUpdatedRowsBaseForTest() {
+ updatedRowsBase.set(-1);
+ }
+
+ // For unit test only.
+ protected void setUpdatedRowsBaseForTest(long updatedRowsBase) {
+ this.updatedRowsBase.set(updatedRowsBase);
+ }
+
public long getBaseIndexDeltaRowCount(OlapTable table) {
- if (colToColStatsMeta == null || colToColStatsMeta.isEmpty() ||
userInjected) {
+ if (userInjected) {
+ return 0;
+ }
+ long collectedRowCountBase = updatedRowsBase.get();
+ if (collectedRowCountBase >= 0) {
+ return updatedRows.get() - collectedRowCountBase;
+ }
+ // A record written before updatedRowsBase existed has no baseline of
its own. Derive it from the
+ // collected column statistics, which is where it used to live. Once
they are all dropped the baseline
+ // is unknown, so no row is reported as a delta row.
+ if (colToColStatsMeta == null || colToColStatsMeta.isEmpty()) {
return 0;
}
long maxUpdateRows = 0;
@@ -268,6 +382,38 @@ public class TableStatsMeta implements Writable,
GsonPostProcessable {
return updatedRows.get() - maxUpdateRows;
}
+ /**
+ * The row count of the index together with the rows loaded since it was
collected, i.e. the row count of
+ * the table. Both are read by the planner without the table lock, for
instance while it plans a direct
+ * scan of a materialized view, so they have to come from the same state
of this record: a collected row
+ * count paired with the baseline of another analysis would count rows
twice or miss them. The rows loaded
+ * while this call runs are not part of the snapshot, whichever state
reads them accumulates them in
+ * {@link #updatedRows} and reports them as delta rows.
+ */
+ public synchronized long getRowCountWithDeltaRows(OlapTable table, long
indexId) {
+ long rowCount = getRowCount(indexId);
+ if (!keepsOneRowPerBaseRow(table, indexId)) {
+ // The index aggregates, or it merges the rows of a unique key
table: it has its own, smaller row
+ // count, and the rows loaded into the base index would overstate
it.
+ return rowCount;
+ }
+ if (rowCount == TableIf.UNKNOWN_ROW_COUNT) {
+ // A row-preserving index which was added after the row counts
were collected has no entry of
+ // its own, for instance the rollup of an ALTER TABLE ADD ROLLUP:
the record describes a table
+ // without that index. It keeps one row per base row, so the
collected row count of the base
+ // index is its row count as well. Without this fallback the index
would be planned at one row
+ // until the backends report it for the first time, although the
base index has a usable count.
+ rowCount = getRowCount(table.getBaseIndexId());
+ }
+ if (rowCount == TableIf.UNKNOWN_ROW_COUNT) {
+ // Neither the index nor the base index has a collected row count.
The delta rows are relative to
+ // a collection which does not exist, so no row count can be
derived from them and the planner
+ // treats the unknown row count as no information.
+ return rowCount;
+ }
+ return rowCount + getBaseIndexDeltaRowCount(table);
+ }
+
public boolean isColumnsStatsEmpty() {
return colToColStatsMeta == null || colToColStatsMeta.isEmpty();
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/UpdateRowsEvent.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/UpdateRowsEvent.java
index dce8c277e69..fd103c5b2c4 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/UpdateRowsEvent.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/UpdateRowsEvent.java
@@ -46,9 +46,18 @@ public class UpdateRowsEvent implements Writable {
private final long tableId;
public UpdateRowsEvent(Map<Long, Long> records) {
+ this(-1, records);
+ }
+
+ /**
+ * For a load which reports the rows of a table by table id instead of by
tablet. The database id comes
+ * first because the tablet keyed constructor above already takes a map
followed by a long, and both
+ * would have the same erasure otherwise.
+ */
+ public UpdateRowsEvent(long dbId, Map<Long, Long> records) {
this.records = records;
this.tabletRecords = null;
- this.dbId = -1;
+ this.dbId = dbId;
this.partitionToUpdateRows = null;
this.tableId = -1;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/statistics/repository/StatisticsCleaner.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/repository/StatisticsCleaner.java
index 429d25e816f..237b400fa12 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/repository/StatisticsCleaner.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/repository/StatisticsCleaner.java
@@ -34,6 +34,7 @@ import org.apache.doris.statistics.analysis.AnalysisManager;
import org.apache.doris.statistics.analysis.TableStatsMeta;
import org.apache.doris.statistics.util.StatisticsUtil;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.text.StringSubstitutor;
@@ -140,6 +141,13 @@ public class StatisticsCleaner extends MasterDaemon {
}
LOG.info("Table {}.{}.{} with id {} not exist, remove its
table stats record.",
stats.ctlName, stats.dbName, stats.tblName, id);
+ TableIf recycledTable =
Env.getCurrentRecycleBin().getRecycledTable(id);
+ if (recycledTable instanceof OlapTable) {
+ removeRecycledTableStats((OlapTable) recycledTable,
analysisManager, internalCatalog);
+ continue;
+ }
+ // No table lock: the table is gone and cannot be recovered,
so no load can recreate the record
+ // of this id while the deletion is journaled.
analysisManager.removeTableStats(id);
Env.getCurrentEnv().getEditLog().logDeleteTableStats(new
TableStatsDeletionLog(id));
} catch (Exception e) {
@@ -148,6 +156,36 @@ public class StatisticsCleaner extends MasterDaemon {
}
}
+ /**
+ * Remove the record of a table which is in the recycle bin and journal
the deletion, under the table write
+ * lock.
+ *
+ * <p>`RECOVER TABLE` registers the recycled table again with its original
id, and the load which then
+ * recreates the record of that id is published under the table write lock
(`finishTransaction` holds it
+ * until `updateCatalogAfterVisible` returned). Removing and journaling
the deletion under the same lock
+ * orders it either before the recovery and the load, which journal the
creation after this deletion, or
+ * after them, in which case the record of that load is the one which is
removed: the master and its
+ * followers then apply the mutations in the order of the journal. Without
the lock the deletion could be
+ * journaled after a load recreated the record, and a follower would
replay "remove" after "create" while
+ * the master applied it before, which leaves the master with a record
that its followers removed.
+ */
+ @VisibleForTesting
+ void removeRecycledTableStats(OlapTable recycledTable, AnalysisManager
analysisManager,
+ InternalCatalog internalCatalog) {
+ recycledTable.writeLock();
+ try {
+ if (tableExistInInternalCatalog(internalCatalog,
recycledTable.getId())) {
+ // The table was recovered after its record was found
orphaned, keep its statistics.
+ return;
+ }
+ analysisManager.removeTableStats(recycledTable.getId());
+ Env.getCurrentEnv().getEditLog().logDeleteTableStats(
+ new TableStatsDeletionLog(recycledTable.getId()));
+ } finally {
+ recycledTable.writeUnlock();
+ }
+ }
+
private boolean tableExistInInternalCatalog(InternalCatalog
internalCatalog, long tableId) {
List<Long> dbIds = internalCatalog.getDbIds();
for (long dbId : dbIds) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeJobV2StatsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeJobV2StatsTest.java
new file mode 100644
index 00000000000..d5f94088f21
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeJobV2StatsTest.java
@@ -0,0 +1,59 @@
+// 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.
+
+package org.apache.doris.alter;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.statistics.analysis.AnalysisManager;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+public class SchemaChangeJobV2StatsTest {
+
+ /**
+ * The statistics record of a table is removed when the schema change
finished, together with the journal
+ * entry of FINISHED. A load which becomes visible in between updates or
recreates the record, and a follower
+ * replays "FINISHED, then VISIBLE", so the removal has to happen under
the table write lock which the load
+ * takes as well.
+ */
+ @Test
+ public void testRemoveTableStatsOnFinishedRequiresTheTableWriteLock() {
+ Env env = Mockito.mock(Env.class);
+ AnalysisManager analysisManager = Mockito.mock(AnalysisManager.class);
+ Mockito.when(env.getAnalysisManager()).thenReturn(analysisManager);
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Mockito.when(table.getId()).thenReturn(10001L);
+ Mockito.when(table.getName()).thenReturn("t");
+ SchemaChangeJobV2 job = new SchemaChangeJobV2("sql", 1L, 2L, 10001L,
"t", 10000L);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+
+
Mockito.when(table.isWriteLockHeldByCurrentThread()).thenReturn(false);
+ Assertions.assertThrows(IllegalStateException.class, () ->
job.removeTableStatsOnFinished(table));
+ Mockito.verify(analysisManager,
Mockito.never()).removeTableStats(Mockito.anyLong());
+
+
Mockito.when(table.isWriteLockHeldByCurrentThread()).thenReturn(true);
+ job.removeTableStatsOnFinished(table);
+ Mockito.verify(analysisManager).removeTableStats(10001L);
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/StatsCalculatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/StatsCalculatorTest.java
index f5bbc72e3db..90df4ecc4ae 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/StatsCalculatorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/StatsCalculatorTest.java
@@ -19,11 +19,15 @@ package org.apache.doris.nereids.stats;
import org.apache.doris.analysis.IntLiteral;
import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.MaterializedIndexMeta;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.common.Pair;
+import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.memo.Group;
import org.apache.doris.nereids.memo.GroupExpression;
@@ -41,6 +45,7 @@ import org.apache.doris.nereids.trees.plans.GroupPlan;
import org.apache.doris.nereids.trees.plans.JoinType;
import org.apache.doris.nereids.trees.plans.LimitPhase;
import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.algebra.OlapScan;
import org.apache.doris.nereids.trees.plans.algebra.SetOperation.Qualifier;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
@@ -58,6 +63,7 @@ import org.apache.doris.nereids.util.PlanConstructor;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.statistics.analysis.AnalysisManager;
+import org.apache.doris.statistics.analysis.TableStatsMeta;
import org.apache.doris.statistics.cache.StatisticsCache;
import org.apache.doris.statistics.model.ColumnStatistic;
import org.apache.doris.statistics.model.ColumnStatisticBuilder;
@@ -382,6 +388,46 @@ public class StatsCalculatorTest {
}
}
+ @Test
+ public void testDeltaRowCountIsReportedForNonRowPreservingIndex() {
+ long tableId = 100L;
+ long rollupIndexId = 200L;
+ OlapTable table = Mockito.mock(OlapTable.class);
+ DatabaseIf db = Mockito.mock(DatabaseIf.class);
+ CatalogIf catalog = Mockito.mock(CatalogIf.class);
+ Mockito.doReturn(db).when(table).getDatabase();
+ Mockito.doReturn(catalog).when(db).getCatalog();
+ Mockito.doReturn(tableId).when(table).getId();
+
Mockito.doReturn(Collections.<Long>emptyList()).when(table).getIndexIdList();
+ // The scanned index aggregates the rows of the base index, it does
not keep one row per base row.
+ MaterializedIndexMeta rollupMeta =
Mockito.mock(MaterializedIndexMeta.class);
+ Mockito.doReturn(KeysType.AGG_KEYS).when(rollupMeta).getKeysType();
+
Mockito.doReturn(rollupMeta).when(table).getIndexMetaByIndexId(rollupIndexId);
+
+ // The record a load creates: nothing has been collected yet, every
loaded row is a delta row.
+ TableStatsMeta tableStats = new TableStatsMeta(table);
+ tableStats.updatedRows.set(100);
+
+ AnalysisManager analysisManager = Mockito.mock(AnalysisManager.class);
+
Mockito.doReturn(tableStats).when(analysisManager).findTableStatsStatus(tableId);
+ Env env = Mockito.mock(Env.class);
+ Mockito.doReturn(analysisManager).when(env).getAnalysisManager();
+
+ OlapScan scan = Mockito.mock(OlapScan.class);
+ Mockito.doReturn(table).when(scan).getTable();
+ Mockito.doReturn(rollupIndexId).when(scan).getSelectedIndexId();
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ // These rows are not the row count of an index which does not
keep one row per base row, they must
+ // not be added to it (TableStatsMeta#getRowCountWithDeltaRows
gates that). They are still the
+ // population FilterEstimation retries a zero estimate with, for
every index: stale column
+ // statistics are not index specific.
+ Assertions.assertEquals(100L,
+ new StatsCalculator((CascadesContext)
null).computeDeltaRowCount(scan));
+ }
+ }
+
@Test
public void testLimit() {
List<String> qualifier = ImmutableList.of("test", "t");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/statistics/analysis/AnalysisManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/statistics/analysis/AnalysisManagerTest.java
index ec030d63e34..d206daa759d 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/statistics/analysis/AnalysisManagerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/statistics/analysis/AnalysisManagerTest.java
@@ -23,8 +23,11 @@ import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.ReplicaAllocation;
import org.apache.doris.catalog.TableIf;
+import org.apache.doris.catalog.TableProperty;
import org.apache.doris.catalog.info.TableNameInfo;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
@@ -40,6 +43,7 @@ import
org.apache.doris.nereids.trees.plans.commands.KillAnalyzeJobCommand;
import org.apache.doris.nereids.types.IntegerType;
import org.apache.doris.nereids.util.MoreFieldsThread;
import org.apache.doris.persist.EditLog;
+import org.apache.doris.persist.TableStatsDeletionLog;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.statistics.analysis.AnalysisInfo.AnalysisMethod;
import org.apache.doris.statistics.analysis.AnalysisInfo.AnalysisType;
@@ -50,11 +54,19 @@ import org.apache.doris.statistics.util.StatisticsUtil;
import org.apache.doris.thrift.TQueryColumn;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -64,7 +76,11 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
// CHECKSTYLE OFF
public class AnalysisManagerTest {
@@ -691,4 +707,424 @@ public class AnalysisManagerTest {
Mockito.when(command.getAnalyzeProperties()).thenReturn(analyzeProperties);
return command;
}
+
+ // A table of the given database which reports one replica for the rows a
load reports.
+ private OlapTable mockOlapTableInReplica(long tableId, String name,
Database db) {
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Mockito.when(table.getId()).thenReturn(tableId);
+ Mockito.when(table.getBaseIndexId()).thenReturn(10001L);
+
Mockito.when(table.getIndexIdList()).thenReturn(ImmutableList.of(10001L));
+ Mockito.when(table.getName()).thenReturn(name);
+ Mockito.when(table.getDatabase()).thenReturn(db);
+ TableProperty tableProperty = Mockito.mock(TableProperty.class);
+ ReplicaAllocation replicaAllocation =
Mockito.mock(ReplicaAllocation.class);
+
Mockito.when(tableProperty.getReplicaAllocation()).thenReturn(replicaAllocation);
+
Mockito.when(replicaAllocation.getTotalReplicaNum()).thenReturn((short) 1);
+ Mockito.when(table.getTableProperty()).thenReturn(tableProperty);
+ db.registerTable(table);
+ return table;
+ }
+
+ /**
+ * A load recreates the statistics record of a table whose record was
removed, e.g. by TRUNCATE TABLE, and
+ * accumulates the loaded rows into it. The record is created on the
transaction visibility of the load,
+ * which the master applies in {@code
DatabaseTransactionMgr.updateCatalogAfterVisible} and a follower
+ * applies while it replays the transaction state, so the row count is
reported on every frontend without
+ * an edit log entry of its own.
+ */
+ @Test
+ public void testLoadCreatesTheStatisticsRecordOfATableWithoutOne() throws
Exception {
+ AnalysisManager manager = Mockito.spy(new AnalysisManager());
+ InternalCatalog catalog = new InternalCatalog();
+ Database db = new Database(20001L, "test_load_creates_stats");
+ catalog.unprotectCreateDb(db);
+ OlapTable table = mockOlapTableInReplica(30001L, "t_load", db);
+
+ Map<Long, Map<Long, Long>> tabletRecords = Maps.newHashMap();
+ tabletRecords.put(30001L, ImmutableMap.of(1L, 3L, 2L, 2L));
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ // What the master does when the loading transaction becomes
visible.
+ manager.updateUpdatedRows(tabletRecords, db.getId(), 0L);
+
+ // The record exists with the rows of the load, and the planner
reads them as the row count of the
+ // table while the backends have not reported the new tablets yet.
+ TableStatsMeta record = manager.findTableStatsStatus(30001L);
+ Assertions.assertNotNull(record);
+ Assertions.assertEquals(5, record.updatedRows.get());
+ Assertions.assertEquals(5, record.getRowCountWithDeltaRows(table,
10001L));
+ Assertions.assertTrue(record.isColumnsStatsEmpty());
+
+ // A record which exists is accumulated, not replaced.
+ manager.updateUpdatedRows(tabletRecords, db.getId(), 0L);
+ Assertions.assertSame(record,
manager.findTableStatsStatus(30001L));
+ Assertions.assertEquals(10, record.updatedRows.get());
+ }
+ }
+
+ @Test
+ public void testReplayOfALoadCreatesTheSameStatisticsRecord() throws
Exception {
+ // A follower applies the row counts of a load while it replays the
transaction state, through the very
+ // same method. A table whose record the truncation removed gets the
same record there.
+ AnalysisManager manager = Mockito.spy(new AnalysisManager());
+ InternalCatalog catalog = new InternalCatalog();
+ Database db = new Database(20001L, "test_load_replay_creates_stats");
+ catalog.unprotectCreateDb(db);
+ OlapTable table = mockOlapTableInReplica(30002L, "t_replay", db);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ Map<Long, Map<Long, Long>> tabletRecords = Maps.newHashMap();
+ tabletRecords.put(30002L, ImmutableMap.of(1L, 4L));
+ manager.replayUpdateRowsRecord(new UpdateRowsEvent(tabletRecords,
db.getId()));
+ }
+
+ TableStatsMeta record = manager.findTableStatsStatus(30002L);
+ Assertions.assertNotNull(record);
+ Assertions.assertEquals(4, record.updatedRows.get());
+ Assertions.assertEquals(4, record.getRowCountWithDeltaRows(table,
10001L));
+ }
+
+ @Test
+ public void testLoadDoesNotTouchTheCollectedRowCountOfAnExistingRecord()
throws Exception {
+ AnalysisManager manager = Mockito.spy(new AnalysisManager());
+ InternalCatalog catalog = new InternalCatalog();
+ Database db = new Database(20001L, "test_load_keeps_stats");
+ catalog.unprotectCreateDb(db);
+ OlapTable table = mockOlapTableInReplica(30003L, "t_keeps", db);
+
+ // The record an analysis left: 100 collected rows, no delta.
+ TableStatsMeta analyzed = new TableStatsMeta(table);
+ analyzed.addIndexRowForTest(10001L, 100);
+ analyzed.rowCount = 100;
+ analyzed.updatedRows.set(100);
+ analyzed.setUpdatedRowsBaseForTest(100);
+ manager.replayUpdateTableStatsStatus(analyzed);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ Map<Long, Map<Long, Long>> tabletRecords = Maps.newHashMap();
+ tabletRecords.put(30003L, ImmutableMap.of(1L, 7L));
+ manager.updateUpdatedRows(tabletRecords, db.getId(), 0L);
+ }
+
+ TableStatsMeta record = manager.findTableStatsStatus(30003L);
+ Assertions.assertSame(analyzed, record);
+ Assertions.assertEquals(100, record.getRowCount(10001L));
+ Assertions.assertEquals(107, record.updatedRows.get());
+ Assertions.assertEquals(107, record.getRowCountWithDeltaRows(table,
10001L));
+ }
+
+ @Test
+ public void testALoadWhichReportsRowsByTableIdCreatesTheRecordAsWell()
throws Exception {
+ // A cloud commit reports the rows of a transaction by table id
instead of by tablet, and journals the
+ // event, so the follower replays it through this branch as well.
+ AnalysisManager manager = Mockito.spy(new AnalysisManager());
+ InternalCatalog catalog = new InternalCatalog();
+ Database db = new Database(20001L, "test_load_by_table_id");
+ catalog.unprotectCreateDb(db);
+ OlapTable table = mockOlapTableInReplica(30004L, "t_by_table", db);
+
+ Map<Long, Long> records = Maps.newHashMap();
+ records.put(30004L, 6L);
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ manager.replayUpdateRowsRecord(new UpdateRowsEvent(db.getId(),
records));
+ }
+
+ TableStatsMeta record = manager.findTableStatsStatus(30004L);
+ Assertions.assertNotNull(record);
+ Assertions.assertEquals(6, record.updatedRows.get());
+ Assertions.assertEquals(6, record.getRowCountWithDeltaRows(table,
10001L));
+ }
+
+ @Test
+ public void testLoadOfAnUnresolvableTableDoesNotCreateARecord() throws
Exception {
+ AnalysisManager manager = Mockito.spy(new AnalysisManager());
+ InternalCatalog catalog = new InternalCatalog();
+ Database db = new Database(20001L, "test_load_missing_table");
+ catalog.unprotectCreateDb(db);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ Map<Long, Map<Long, Long>> tabletRecords = Maps.newHashMap();
+ tabletRecords.put(39999L, ImmutableMap.of(1L, 1L));
+ manager.updateUpdatedRows(tabletRecords, db.getId(), 0L);
+ }
+ Assertions.assertNull(manager.findTableStatsStatus(39999L));
+ }
+
+ private OlapTable mockOlapTableForRemoval(long tableId, String name,
Database db, List<String> calls) {
+ OlapTable table = mockOlapTableInReplica(tableId, name, db);
+ // The lock is not taken by the mock, record the calls instead: the
deletion has to be journaled
+ // between them.
+ Mockito.doAnswer(invocation -> {
+ calls.add("writeLock");
+ return null;
+ }).when(table).writeLock();
+ Mockito.doAnswer(invocation -> {
+ calls.add("writeUnlock");
+ return null;
+ }).when(table).writeUnlock();
+ return table;
+ }
+
+ /**
+ * The record of a table which has none is recreated by the load that
makes a transaction visible, in
+ * {@code DatabaseTransactionMgr.updateCatalogAfterVisible}, which runs
under the table write lock until it
+ * returns. A deletion which is applied between the journaling of that
transaction and the recreation is
+ * replayed after the creation on every follower while the master applied
it before the creation, so the
+ * master would keep a record which no follower has. The deletion is
therefore applied and journaled under
+ * the same lock, and the journal entry is written before the lock is
released.
+ */
+ @Test
+ public void testRemoveTableStatsAndLogHoldsTheTableLock() {
+ AnalysisManager manager = new AnalysisManager();
+ InternalCatalog catalog = new InternalCatalog();
+ Database db = new Database(20001L, "test_remove_table_stats_lock");
+ catalog.unprotectCreateDb(db);
+ List<String> calls = new ArrayList<>();
+ OlapTable table = mockOlapTableForRemoval(30005L, "t_remove_stats",
db, calls);
+ Env env = Mockito.mock(Env.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ Mockito.doAnswer(invocation -> {
+ calls.add("logDeleteTableStats");
+ // The record is already removed, and the lock is still held.
+ Assertions.assertNull(manager.findTableStatsStatus(30005L));
+ return null;
+
}).when(editLog).logDeleteTableStats(Mockito.any(TableStatsDeletionLog.class));
+
+ manager.replayUpdateTableStatsStatus(new TableStatsMeta(table));
+ manager.removeTableStatsAndLog(table);
+
+ Assertions.assertEquals(ImmutableList.of("writeLock",
"logDeleteTableStats", "writeUnlock"), calls);
+ Assertions.assertNull(manager.findTableStatsStatus(30005L));
+ }
+ }
+
+ @Test
+ public void testRemoveTableStatsAndLogDoesNotLockANonOlapTable() {
+ AnalysisManager manager = new AnalysisManager();
+ Env env = Mockito.mock(Env.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+
+ TableIf table = Mockito.mock(TableIf.class);
+ Mockito.when(table.getId()).thenReturn(30006L);
+
+ manager.removeTableStatsAndLog(table);
+
+ // A load only recreates the record of an OLAP table, there is no
creation to be ordered with.
+ Mockito.verify(table, Mockito.never()).writeLock();
+
Mockito.verify(editLog).logDeleteTableStats(Mockito.any(TableStatsDeletionLog.class));
+ }
+ }
+ // A record whose serialization waits, so that a test can publish another
record while the image of the
+ // statistics map is written. The latches are transient: they are not part
of the image.
+ private static class BlockingTableStatsMeta extends TableStatsMeta {
+ private final transient CountDownLatch imageWriting = new
CountDownLatch(1);
+ private final transient CountDownLatch resumeImage = new
CountDownLatch(1);
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ imageWriting.countDown();
+ try {
+ resumeImage.await(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ super.write(out);
+ }
+ }
+
+ /**
+ * The image writes the number of records of the statistics map and then
the records, and the reader trusts
+ * that number. A record which a load publishes while the image is
written, for instance from the commit
+ * report of a cloud backend, must therefore not be inserted between the
two: it would be written although
+ * the declared number excludes it, and the reader would take the extra
record as the header of the next
+ * module of the image.
+ */
+ @Test
+ public void testRecordInsertedWhileTheImageIsWrittenWaitsForTheStatsMap()
throws Exception {
+ AnalysisManager manager = new AnalysisManager();
+ InternalCatalog catalog = new InternalCatalog();
+ Database db = new Database(20002L, "test_stats_map_lock");
+ catalog.unprotectCreateDb(db);
+ OlapTable table = mockOlapTableInReplica(30008L, "t_stats_map_lock",
db);
+ BlockingTableStatsMeta record = new BlockingTableStatsMeta();
+ manager.replayUpdateTableStatsStatus(record);
+
+ ByteArrayOutputStream image = new ByteArrayOutputStream();
+ Thread dump = new Thread(() -> {
+ try {
+ manager.write(new DataOutputStream(image));
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ });
+ dump.start();
+ Assertions.assertTrue(record.imageWriting.await(30, TimeUnit.SECONDS),
"the image is not being written");
+
+ // The load of another table publishes its record from its own thread,
exactly like the commit report
+ // of a cloud backend, which takes no table lock.
+ CountDownLatch loadReady = new CountDownLatch(1);
+ CountDownLatch insertNow = new CountDownLatch(1);
+ CountDownLatch inserted = new CountDownLatch(1);
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ Thread load = new Thread(() -> {
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ loadReady.countDown();
+ insertNow.await();
+ manager.replayUpdateRowsRecord(new UpdateRowsEvent(db.getId(),
+ ImmutableMap.of(table.getId(), 1L)));
+ inserted.countDown();
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+ load.start();
+ Assertions.assertTrue(loadReady.await(30, TimeUnit.SECONDS), "the load
is not ready");
+ insertNow.countDown();
+ // The insertion has to wait for the monitor which the image holds,
otherwise the record can be written
+ // although the number of records was already written.
+ boolean insertedWhileWriting = inserted.await(500,
TimeUnit.MILLISECONDS);
+
+ record.resumeImage.countDown();
+ dump.join(30000);
+ load.join(30000);
+
+ Assertions.assertFalse(insertedWhileWriting, "the record was inserted
while the image was written");
+ Assertions.assertNull(failure.get());
+ Assertions.assertFalse(dump.isAlive(), "the image was not written");
+ Assertions.assertFalse(load.isAlive());
+
+ DataInputStream in = new DataInputStream(new
ByteArrayInputStream(image.toByteArray()));
+ AnalysisManager loaded = AnalysisManager.readFields(in);
+ // The reader consumes exactly the records which the image declares.
+ Assertions.assertEquals(0, in.available(), "the image contains more
records than it declares");
+ Assertions.assertNotNull(loaded.findTableStatsStatus(0L));
+ Assertions.assertNull(loaded.findTableStatsStatus(table.getId()),
+ "the record which was created after the image was written is
in the image");
+ }
+ /**
+ * A cloud load reports the rows of its transaction from the backend, and
the report applies them and
+ * journals them afterwards. A whole table DROP STATS removes the record
and journals the deletion, so
+ * without a common lock the deletion can be applied between the
application of the report and its journal
+ * entry: a follower then replays "remove, then add" and keeps a record
which the master no longer has.
+ */
+ @Test
+ public void testCloudRowUpdateIsSerializedWithTheTableLock() throws
Exception {
+ AnalysisManager manager = new AnalysisManager();
+ InternalCatalog catalog = new InternalCatalog();
+ Database db = new Database(20003L, "test_cloud_row_update_lock");
+ catalog.unprotectCreateDb(db);
+ OlapTable table = mockOlapTableInReplica(30009L, "t_cloud_row_update",
db);
+ Env env = Mockito.mock(Env.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ Mockito.when(table.tryWriteLockIfExist(Mockito.anyLong(),
Mockito.any(TimeUnit.class)))
+ .thenAnswer(invocation ->
lock.writeLock().tryLock(invocation.getArgument(0),
+ invocation.getArgument(1)));
+ Mockito.doAnswer(invocation -> {
+ lock.writeLock().unlock();
+ return null;
+ }).when(table).writeUnlock();
+
+ CountDownLatch reportReady = new CountDownLatch(1);
+ CountDownLatch reported = new CountDownLatch(1);
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ Thread report = new Thread(() -> {
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ Mockito.when(env.isMaster()).thenReturn(true);
+ reportReady.countDown();
+ manager.updateUpdatedRows(ImmutableMap.of(table.getId(), 5L),
db.getId());
+ reported.countDown();
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+
+ // The deletion holds the table write lock while it removes the record
and journals it.
+ lock.writeLock().lock();
+ report.start();
+ Assertions.assertTrue(reportReady.await(30, TimeUnit.SECONDS), "the
report is not ready");
+ boolean appliedWhileTheDeletionRan = reported.await(500,
TimeUnit.MILLISECONDS);
+ lock.writeLock().unlock();
+ report.join(30000);
+
+ Assertions.assertFalse(appliedWhileTheDeletionRan, "the report was
applied while the table was locked");
+ Assertions.assertNull(failure.get());
+ Assertions.assertFalse(report.isAlive());
+ Mockito.verify(env, Mockito.atLeast(1)).isMaster();
+ TableStatsMeta record = manager.findTableStatsStatus(table.getId());
+ Assertions.assertNotNull(record);
+ Assertions.assertEquals(5, record.updatedRows.get());
+ Mockito.verify(editLog,
Mockito.times(1)).logUpdateRowsRecord(Mockito.any(UpdateRowsEvent.class));
+ }
+ /**
+ * A cloud stream load queues the report of the rows it loaded and can
return before the frontend received
+ * it. A whole table TRUNCATE replaces the partitions and keeps the table
id, so a report which arrives
+ * afterwards counts the rows the truncation removed: they must neither be
applied to the record which the
+ * next load recreates nor be journaled.
+ */
+ @Test
+ public void testReportOfAPartitionRemovedByTruncateIsNotApplied() throws
Exception {
+ AnalysisManager manager = new AnalysisManager();
+ InternalCatalog catalog = new InternalCatalog();
+ Database db = new Database(20004L, "test_stale_cloud_report");
+ catalog.unprotectCreateDb(db);
+ OlapTable table = mockOlapTableInReplica(30010L, "t_stale_report", db);
+ long replacedPartitionId = 40001L;
+ long partitionId = 40002L;
+
Mockito.when(table.getPartition(partitionId)).thenReturn(Mockito.mock(Partition.class));
+ Mockito.when(table.tryWriteLockIfExist(Mockito.anyLong(),
Mockito.any(TimeUnit.class))).thenReturn(true);
+ Env env = Mockito.mock(Env.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ Mockito.when(env.isMaster()).thenReturn(true);
+
+ // The report of the 100 rows which the truncation removed arrives
after the replacement partitions
+ // were created.
+ manager.updateUpdatedRows(ImmutableMap.of(table.getId(), 100L),
db.getId(),
+ ImmutableMap.of(table.getId(),
ImmutableList.of(replacedPartitionId)));
+ Assertions.assertNull(manager.findTableStatsStatus(table.getId()));
+ Mockito.verify(editLog,
Mockito.never()).logUpdateRowsRecord(Mockito.any(UpdateRowsEvent.class));
+
+ // The 3 rows which the next load reports belong to the current
partition and recreate the record.
+ manager.updateUpdatedRows(ImmutableMap.of(table.getId(), 3L),
db.getId(),
+ ImmutableMap.of(table.getId(),
ImmutableList.of(partitionId)));
+ TableStatsMeta record =
manager.findTableStatsStatus(table.getId());
+ Assertions.assertNotNull(record);
+ Assertions.assertEquals(3, record.updatedRows.get());
+
Mockito.verify(editLog).logUpdateRowsRecord(Mockito.any(UpdateRowsEvent.class));
+ }
+ }
+
+ @Test
+ public void testReportOfRemovedPartitionsIsRecognized() {
+ OlapTable table = Mockito.mock(OlapTable.class);
+
Mockito.when(table.getPartition(1L)).thenReturn(Mockito.mock(Partition.class));
+
+
Assertions.assertTrue(AnalysisManager.reportCountsCurrentPartitions(table,
ImmutableList.of()));
+
Assertions.assertTrue(AnalysisManager.reportCountsCurrentPartitions(table,
ImmutableList.of(1L)));
+
Assertions.assertFalse(AnalysisManager.reportCountsCurrentPartitions(table,
ImmutableList.of(2L)));
+ // A report which counts rows of a removed partition is stale as a
whole.
+
Assertions.assertFalse(AnalysisManager.reportCountsCurrentPartitions(table,
ImmutableList.of(1L, 2L)));
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/statistics/analysis/TableStatsMetaTest.java
b/fe/fe-core/src/test/java/org/apache/doris/statistics/analysis/TableStatsMetaTest.java
index 04208574f3d..a3ae026a679 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/statistics/analysis/TableStatsMetaTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/statistics/analysis/TableStatsMetaTest.java
@@ -17,17 +17,72 @@
package org.apache.doris.statistics.analysis;
+import org.apache.doris.analysis.Expr;
+import org.apache.doris.catalog.AggregateType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.MaterializedIndexMeta;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.common.Pair;
+import org.apache.doris.datasource.CatalogIf;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.util.HashSet;
+import java.util.concurrent.atomic.AtomicBoolean;
class TableStatsMetaTest {
+ private static final long BASE_INDEX_ID = 10001L;
+ private static final long ROLLUP_INDEX_ID = 10002L;
+ private static final String BASE_INDEX_NAME = "baseIndex";
+
+ private OlapTable mockOlapTable() {
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Mockito.doReturn(Lists.newArrayList(BASE_INDEX_ID,
ROLLUP_INDEX_ID)).when(table).getIndexIdList();
+ Mockito.doReturn(BASE_INDEX_ID).when(table).getBaseIndexId();
+
Mockito.doReturn(BASE_INDEX_NAME).when(table).getIndexNameById(BASE_INDEX_ID);
+ DatabaseIf db = Mockito.mock(DatabaseIf.class);
+ CatalogIf catalog = Mockito.mock(CatalogIf.class);
+ Mockito.doReturn(db).when(table).getDatabase();
+ Mockito.doReturn(catalog).when(db).getCatalog();
+ return table;
+ }
+
+ // 100 rows of the base index had been loaded when the job was built.
+ private AnalysisInfo analyzeJob() {
+ AnalysisInfo job = new AnalysisInfoBuilder().setRowCount(100)
+ .setJobColumns(Sets.newHashSet(Pair.of(BASE_INDEX_NAME,
"col1")))
+ .setUpdateRows(100)
+ .setTblUpdateTime(1000L)
+ .setJobType(AnalysisInfo.JobType.MANUAL)
+ .build();
+ job.addIndexRowCount(BASE_INDEX_ID, 100);
+ return job;
+ }
+
+ // The table has a rollup index besides the base index. The schema is the
one a real index has: a key
+ // column, which carries no aggregation type at all, and a value column
with the given one. A non null
+ // where clause makes it a filtered index.
+ private void mockRollupIndex(OlapTable table, long indexId, KeysType
keysType, AggregateType valueType,
+ Expr whereClause) {
+ MaterializedIndexMeta indexMeta =
Mockito.mock(MaterializedIndexMeta.class);
+ Mockito.doReturn(keysType).when(indexMeta).getKeysType();
+ Mockito.doReturn(whereClause).when(indexMeta).getWhereClause();
+ Column keyColumn = new Column("col1", PrimitiveType.INT);
+ keyColumn.setAggregationType(null, false);
+ Column valueColumn = new Column("col2", PrimitiveType.INT);
+ valueColumn.setAggregationType(valueType, false);
+ Mockito.doReturn(Lists.newArrayList(keyColumn,
valueColumn)).when(indexMeta).getSchema();
+ Mockito.doReturn(indexMeta).when(table).getIndexMetaByIndexId(indexId);
+ }
+
@Test
void update() {
OlapTable table = Mockito.mock(OlapTable.class);
@@ -58,4 +113,237 @@ class TableStatsMetaTest {
Assertions.assertEquals(-1, meta.getRowCount(3));
Assertions.assertEquals(-1, meta.getRowCount(4));
}
+
+ @Test
+ void testRecordCreatedByALoadReportsTheLoadedRows() {
+ OlapTable table = mockOlapTable();
+ // The record a load creates for a table whose statistics were
dropped, e.g. by TRUNCATE TABLE: no row
+ // count has ever been collected, so every loaded row is a delta row.
+ TableStatsMeta meta = new TableStatsMeta(table);
+ Assertions.assertTrue(meta.isColumnsStatsEmpty());
+ Assertions.assertEquals(0, meta.updatedTime);
+
+ meta.updatedRows.set(30);
+ Assertions.assertEquals(30, meta.getBaseIndexDeltaRowCount(table));
+ Assertions.assertEquals(0, meta.getRowCount(BASE_INDEX_ID));
+ // The planner reads this as the row count of the table while the
backends have not reported the new
+ // tablets yet.
+ Assertions.assertEquals(30, meta.getRowCountWithDeltaRows(table,
BASE_INDEX_ID));
+
+ // The rows of the next loads add up.
+ meta.updatedRows.addAndGet(12);
+ Assertions.assertEquals(42, meta.getRowCountWithDeltaRows(table,
BASE_INDEX_ID));
+ }
+
+ @Test
+ void testDeltaRowCountKeptAfterColumnStatsDropped() {
+ OlapTable table = mockOlapTable();
+ TableStatsMeta meta = new TableStatsMeta();
+ meta.updatedRows.set(100);
+ meta.update(analyzeJob(), table);
+ // 50 rows were loaded after the analysis.
+ meta.updatedRows.set(150);
+ Assertions.assertEquals(50, meta.getBaseIndexDeltaRowCount(table));
+
+ // The baseline is recorded in the table stats, dropping all the
column statistics of the table
+ // must not turn the 100 collected rows into delta rows again.
+ meta.removeColumn(BASE_INDEX_NAME, "col1");
+ Assertions.assertTrue(meta.isColumnsStatsEmpty());
+ Assertions.assertEquals(50, meta.getBaseIndexDeltaRowCount(table));
+ }
+
+ @Test
+ void testDeltaRowCountOfLegacyRecord() {
+ OlapTable table = mockOlapTable();
+ TableStatsMeta meta = new TableStatsMeta();
+ meta.updatedRows.set(150);
+ meta.update(analyzeJob(), table);
+ // A record written before the baseline was recorded in the table
stats has none of its own, it is
+ // derived from the collected column statistics.
+ meta.clearUpdatedRowsBaseForTest();
+ Assertions.assertEquals(50, meta.getBaseIndexDeltaRowCount(table));
+
+ // Once they are all dropped there is no baseline left, no row is
reported as a delta row.
+ meta.removeColumn(BASE_INDEX_NAME, "col1");
+ Assertions.assertTrue(meta.isColumnsStatsEmpty());
+ Assertions.assertEquals(0, meta.getBaseIndexDeltaRowCount(table));
+ }
+
+ @Test
+ void testDeltaRowCountBaselineNotAdvancedWithoutBaseIndexRowCount() {
+ OlapTable table = mockOlapTable();
+ TableStatsMeta meta = new TableStatsMeta();
+ meta.updatedRows.set(100);
+ meta.update(analyzeJob(), table);
+ meta.updatedRows.set(150);
+ Assertions.assertEquals(50, meta.getBaseIndexDeltaRowCount(table));
+
+ // An analyze job which collected the row count of another index only
(a materialized view) does not
+ // refresh the collected row count of the base index, so it must not
advance its baseline, otherwise
+ // the rows loaded in between are no longer reported as delta rows.
+ AnalysisInfo mvJob = new AnalysisInfoBuilder().setRowCount(100)
+ .setJobColumns(Sets.newHashSet(Pair.of("mvIndex", "col1")))
+ .setUpdateRows(150)
+ .build();
+ mvJob.addIndexRowCount(ROLLUP_INDEX_ID, 30);
+ meta.update(mvJob, table);
+ Assertions.assertEquals(50, meta.getBaseIndexDeltaRowCount(table));
+ }
+
+ @Test
+ void testDeltaRowCountOfStaleSnapshotDeliveredAfterTheRecordWasRecreated()
{
+ OlapTable table = mockOlapTable();
+ // The record of a table whose statistics were dropped, with 5 rows
loaded into the table since.
+ TableStatsMeta meta = new TableStatsMeta(table);
+ meta.updatedRows.set(5);
+
+ // An analyze job built before the statistics were dropped delivers
its snapshot afterwards: it still
+ // describes the 100 rows it collected and the baseline it carries is
the same 100 rows. The stale row
+ // count and the stale baseline cancel each other out, the table is
still estimated with its 5 rows.
+ meta.update(analyzeJob(), table);
+ Assertions.assertEquals(100, meta.getRowCount(BASE_INDEX_ID));
+ Assertions.assertEquals(-95, meta.getBaseIndexDeltaRowCount(table));
+ Assertions.assertEquals(5, meta.getRowCountWithDeltaRows(table,
BASE_INDEX_ID));
+ }
+
+ @Test
+ void testDeltaRowCountOfInjectedRowCountAfterDropStats() {
+ OlapTable table = mockOlapTable();
+ // The statistics of the table were dropped and 5 rows are loaded into
it.
+ TableStatsMeta meta = new TableStatsMeta(table);
+ meta.updatedRows.set(5);
+ // The user supplies the row count of the table as of now.
+ AnalysisInfo injected = new AnalysisInfoBuilder().setRowCount(5)
+ .setJobColumns(new HashSet<>()).setUserInject(true).build();
+ injected.addIndexRowCount(BASE_INDEX_ID, 5);
+ meta.update(injected, table);
+ Assertions.assertTrue(meta.userInjected);
+ // The row count supplied by the user is reported as long as it
applies.
+ Assertions.assertEquals(0, meta.getBaseIndexDeltaRowCount(table));
+ Assertions.assertEquals(5, meta.getRowCount(BASE_INDEX_ID));
+
+ // Dropping the statistics of a column clears userInjected without
clearing the row count. It and the
+ // baseline describe the same snapshot, so the loaded rows are not
counted twice.
+ meta.userInjected = false;
+ Assertions.assertEquals(0, meta.getBaseIndexDeltaRowCount(table));
+ Assertions.assertEquals(5, meta.getRowCountWithDeltaRows(table,
BASE_INDEX_ID));
+ }
+
+ @Test
+ void testRowCountSnapshotIsCoherent() throws Exception {
+ OlapTable table = mockOlapTable();
+ TableStatsMeta meta = new TableStatsMeta(table);
+ // An analysis collected the 100 rows the table had, then 50 more rows
were loaded.
+ meta.update(analyzeJob(), table);
+ meta.updatedRows.set(150);
+
+ AtomicBoolean incoherent = new AtomicBoolean(false);
+ AtomicBoolean stop = new AtomicBoolean(false);
+ Thread reader = new Thread(() -> {
+ while (!stop.get()) {
+ // Every state the writer goes through reports the 150 rows
the table holds.
+ if (meta.getRowCountWithDeltaRows(table, BASE_INDEX_ID) !=
150) {
+ incoherent.set(true);
+ }
+ }
+ });
+ reader.start();
+ // Only transitions which really happen are performed: an analysis
collecting the 100 rows the table
+ // had, while 150 rows are loaded. A reader which pairs the collected
row count of one state with the
+ // baseline of the other would compute 100 + (150 - 0) = 250.
+ for (int i = 0; i < 5000; i++) {
+ meta.update(analyzeJob(), table);
+ meta.updatedRows.set(150);
+ }
+ stop.set(true);
+ reader.join();
+ Assertions.assertFalse(incoherent.get(), "the reader saw a row count
of neither state");
+ }
+
+ @Test
+ void testRollupKeepingOneRowPerBaseRowReportsTheLoadedRows() {
+ OlapTable table = mockOlapTable();
+ mockRollupIndex(table, ROLLUP_INDEX_ID, KeysType.DUP_KEYS,
AggregateType.NONE, null);
+ TableStatsMeta meta = new TableStatsMeta(table);
+ meta.updatedRows.set(150);
+
+ Assertions.assertEquals(150, meta.getRowCountWithDeltaRows(table,
BASE_INDEX_ID));
+ Assertions.assertEquals(150, meta.getRowCountWithDeltaRows(table,
ROLLUP_INDEX_ID));
+ }
+
+ @Test
+ void testAggregatingRollupDoesNotReportTheLoadedRows() {
+ OlapTable table = mockOlapTable();
+ mockRollupIndex(table, ROLLUP_INDEX_ID, KeysType.AGG_KEYS,
AggregateType.SUM, null);
+ TableStatsMeta meta = new TableStatsMeta(table);
+ meta.updatedRows.set(150);
+
+ Assertions.assertEquals(150, meta.getRowCountWithDeltaRows(table,
BASE_INDEX_ID));
+ // The rollup aggregates the loaded rows, its row count is unknown
until the backends report it, and
+ // the rows of the base index must not be reported as its row count.
+ Assertions.assertEquals(-1, meta.getRowCountWithDeltaRows(table,
ROLLUP_INDEX_ID));
+ }
+
+ @Test
+ void testFilteredMaterializedIndexDoesNotReportTheLoadedRows() {
+ OlapTable table = mockOlapTable();
+ mockRollupIndex(table, ROLLUP_INDEX_ID, KeysType.DUP_KEYS,
AggregateType.NONE,
+ Mockito.mock(Expr.class));
+ TableStatsMeta meta = new TableStatsMeta(table);
+ meta.updatedRows.set(150);
+
+ Assertions.assertEquals(150, meta.getRowCountWithDeltaRows(table,
BASE_INDEX_ID));
+ // A filtered index only holds the rows which satisfy its predicate,
it must not be reported with the
+ // rows of the whole table.
+ Assertions.assertEquals(-1, meta.getRowCountWithDeltaRows(table,
ROLLUP_INDEX_ID));
+ }
+
+ /**
+ * ALTER TABLE ADD ROLLUP keeps the record which the analysis of the table
collected, so the new rollup
+ * has no entry of its own, and its tablets are not reported by the
backends until the next report cycle.
+ * The rollup keeps one row per base row, so it is planned with the
collected row count of the base index
+ * plus the rows loaded since then, instead of one row which the unknown
row count is clamped to.
+ */
+ @Test
+ void testRollupAddedAfterTheRowCountWasCollectedReportsTheBaseRows() {
+ OlapTable table = mockOlapTable();
+ TableStatsMeta meta = new TableStatsMeta();
+ // The analysis collected the 100 rows which had been loaded, and the
rollup does not exist yet.
+ meta.updatedRows.set(100);
+ meta.update(analyzeJob(), table);
+ // ALTER TABLE ADD ROLLUP r_dup(k1, v): the record is left as it is,
r_dup has no entry.
+ mockRollupIndex(table, ROLLUP_INDEX_ID, KeysType.DUP_KEYS,
AggregateType.NONE, null);
+ // 50 more rows are loaded before the backends report the tablets of
the new index.
+ meta.updatedRows.addAndGet(50);
+
+ Assertions.assertEquals(150, meta.getRowCountWithDeltaRows(table,
BASE_INDEX_ID));
+ Assertions.assertEquals(150, meta.getRowCountWithDeltaRows(table,
ROLLUP_INDEX_ID));
+ }
+
+ @Test
+ void
testAggregatingRollupAddedAfterTheRowCountWasCollectedReportsNoRowCount() {
+ OlapTable table = mockOlapTable();
+ TableStatsMeta meta = new TableStatsMeta();
+ meta.update(analyzeJob(), table);
+ mockRollupIndex(table, ROLLUP_INDEX_ID, KeysType.AGG_KEYS,
AggregateType.SUM, null);
+ meta.updatedRows.addAndGet(50);
+
+ // The rollup merges the rows of the base index, the collected row
count of the base index must not be
+ // reported as its own.
+ Assertions.assertEquals(-1, meta.getRowCountWithDeltaRows(table,
ROLLUP_INDEX_ID));
+ }
+
+ @Test
+ void testRollupWithoutCollectedRowCountOfTheBaseIndexStaysUnknown() {
+ OlapTable table = mockOlapTable();
+ mockRollupIndex(table, ROLLUP_INDEX_ID, KeysType.DUP_KEYS,
AggregateType.NONE, null);
+ // A record which has never collected a row count, e.g. one written
before updatedRowsBase existed.
+ TableStatsMeta meta = new TableStatsMeta();
+ meta.updatedRows.set(50);
+
+ // The loaded rows are a delta relative to a collection which does not
exist, neither the rollup nor
+ // the base index can be reported with a row count.
+ Assertions.assertEquals(-1, meta.getRowCountWithDeltaRows(table,
ROLLUP_INDEX_ID));
+ Assertions.assertEquals(-1, meta.getRowCountWithDeltaRows(table,
BASE_INDEX_ID));
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/statistics/repository/StatisticsCleanerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/statistics/repository/StatisticsCleanerTest.java
index 1823c09a5f3..5da3db1c0f6 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/statistics/repository/StatisticsCleanerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/statistics/repository/StatisticsCleanerTest.java
@@ -17,11 +17,18 @@
package org.apache.doris.statistics.repository;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.common.Config;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.persist.TableStatsDeletionLog;
import org.apache.doris.statistics.StatisticConstants;
+import org.apache.doris.statistics.analysis.AnalysisManager;
import org.apache.doris.statistics.repository.StatisticsCleaner.ExpiredStats;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -29,6 +36,9 @@ import org.mockito.ArgumentMatchers;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.util.ArrayList;
+import java.util.List;
+
class StatisticsCleanerTest {
@Test
@@ -51,4 +61,78 @@ class StatisticsCleanerTest {
Assertions.assertEquals(expiredStats,
StatisticConstants.FETCH_LIMIT);
}
}
+
+ /**
+ * The record of a table which is in the recycle bin is removed under the
table write lock, which the
+ * recovery of that table and the load that recreates the record take as
well. The deletion therefore
+ * cannot be journaled after such a load recreated the record, which would
leave the master with a record
+ * that its followers removed.
+ */
+ @Test
+ void testRecycledTableStatsAreRemovedUnderTheTableLock() {
+ StatisticsCleaner cleaner = new StatisticsCleaner();
+ AnalysisManager analysisManager = Mockito.mock(AnalysisManager.class);
+ InternalCatalog catalog = new InternalCatalog();
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Mockito.when(table.getId()).thenReturn(40001L);
+ Env env = Mockito.mock(Env.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ List<String> calls = new ArrayList<>();
+ Mockito.doAnswer(invocation -> {
+ calls.add("writeLock");
+ return null;
+ }).when(table).writeLock();
+ Mockito.doAnswer(invocation -> {
+ calls.add("removeTableStats");
+ return null;
+ }).when(analysisManager).removeTableStats(40001L);
+ Mockito.doAnswer(invocation -> {
+ calls.add("logDeleteTableStats");
+ return null;
+
}).when(editLog).logDeleteTableStats(Mockito.any(TableStatsDeletionLog.class));
+ Mockito.doAnswer(invocation -> {
+ calls.add("writeUnlock");
+ return null;
+ }).when(table).writeUnlock();
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+
+ cleaner.removeRecycledTableStats(table, analysisManager, catalog);
+
+ Assertions.assertEquals(ImmutableList.of("writeLock",
"removeTableStats", "logDeleteTableStats",
+ "writeUnlock"), calls);
+ }
+ }
+
+ /**
+ * A table which was recovered from the recycle bin after its record was
found orphaned keeps its
+ * statistics: the catalog is checked again under the table lock, so the
record is not removed just
+ * because it was orphaned when the cleaner looked at it.
+ */
+ @Test
+ void testTableRecoveredBeforeTheRemovalKeepsItsStats() {
+ StatisticsCleaner cleaner = new StatisticsCleaner();
+ AnalysisManager analysisManager = Mockito.mock(AnalysisManager.class);
+ InternalCatalog catalog = new InternalCatalog();
+ Database db = new Database(20001L, "test_recovered_table");
+ catalog.unprotectCreateDb(db);
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Mockito.when(table.getId()).thenReturn(40002L);
+ Mockito.when(table.getName()).thenReturn("t_recovered");
+ db.registerTable(table);
+ Env env = Mockito.mock(Env.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+
+ cleaner.removeRecycledTableStats(table, analysisManager, catalog);
+
+ Mockito.verify(analysisManager,
Mockito.never()).removeTableStats(Mockito.anyLong());
+ Mockito.verify(editLog,
Mockito.never()).logDeleteTableStats(Mockito.any(TableStatsDeletionLog.class));
+ }
+ }
}
diff --git a/regression-test/data/query_p0/eager_agg/scalar_agg_pushdown.out
b/regression-test/data/query_p0/eager_agg/scalar_agg_pushdown.out
index 8fa4341a049..975f6894552 100644
--- a/regression-test/data/query_p0/eager_agg/scalar_agg_pushdown.out
+++ b/regression-test/data/query_p0/eager_agg/scalar_agg_pushdown.out
@@ -14,8 +14,8 @@ PhysicalResultSink
--------PhysicalStorageLayerAggregate[t2]
Hint log:
-Used: leading(ta tb )
-UnUsed:
+Used:
+UnUsed: leading(ta broadcast tb)
SyntaxError:
-- !const_group_key_exe --
@@ -32,8 +32,8 @@ PhysicalResultSink
--------PhysicalStorageLayerAggregate[t2]
Hint log:
-Used: leading(ta broadcast tb )
-UnUsed:
+Used:
+UnUsed: leading(ta broadcast tb)
SyntaxError:
-- !pushdown_preserved_exe --
diff --git a/regression-test/data/query_p0/hint/multi_leading.out
b/regression-test/data/query_p0/hint/multi_leading.out
index e1446884fe2..742733456c0 100644
--- a/regression-test/data/query_p0/hint/multi_leading.out
+++ b/regression-test/data/query_p0/hint/multi_leading.out
@@ -302,7 +302,7 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
----PhysicalOlapScan[t1]
--PhysicalResultSink
----PhysicalDistribute[DistributionSpecGather]
-------NestedLoopJoin[INNER_JOIN](cast(sum(c11) as DOUBLE) > avg(t1.c11) / 20.0)
+------NestedLoopJoin[INNER_JOIN](cast(sum(c11) as DOUBLE) > (avg(c11) / 20.0))
--------hashAgg[GLOBAL]
----------PhysicalDistribute[DistributionSpecHash]
------------PhysicalCteConsumer ( cteId=CTEId#0 )
diff --git a/regression-test/data/query_p0/runtimefilterV2/rfv2.out
b/regression-test/data/query_p0/runtimefilterV2/rfv2.out
index b4eb3a96667..3cb34e5526f 100644
--- a/regression-test/data/query_p0/runtimefilterV2/rfv2.out
+++ b/regression-test/data/query_p0/runtimefilterV2/rfv2.out
@@ -2,35 +2,35 @@
-- !1 --
PhysicalResultSink
--PhysicalDistribute[DistributionSpecGather]
-----PhysicalIntersect build RFs:RF0 a1->b1 RF1 a1->b1 RF2 a1->c1 RF3 a1->c1
+----PhysicalIntersect build RFs:RF0 b1->a1 RF1 b1->a1 RF2 b1->c1 RF3 b1->c1
------PhysicalDistribute[DistributionSpecHash]
---------PhysicalOlapScan[a]
+--------PhysicalOlapScan[b]
------PhysicalDistribute[DistributionSpecHash]
---------PhysicalOlapScan[b] apply RFs: RF0 RF1
+--------PhysicalOlapScan[a] apply RFs: RF0 RF1
------PhysicalDistribute[DistributionSpecHash]
--------PhysicalOlapScan[c] apply RFs: RF2 RF3
-- !2 --
PhysicalResultSink
--PhysicalDistribute[DistributionSpecGather]
-----PhysicalIntersect build RFs:RF0 x->cast(b1 as BIGINT) RF1 x->cast(b1 as
BIGINT)
+----PhysicalIntersect build RFs:RF0 b1->(cast(a1 as BIGINT) + 1) RF1
b1->(cast(a1 as BIGINT) + 1)
------PhysicalDistribute[DistributionSpecHash]
--------PhysicalProject
-----------PhysicalOlapScan[a]
+----------PhysicalOlapScan[b]
------PhysicalDistribute[DistributionSpecHash]
--------PhysicalProject
-----------PhysicalOlapScan[b] apply RFs: RF0 RF1
+----------PhysicalOlapScan[a] apply RFs: RF0 RF1
-- !3 --
PhysicalResultSink
--PhysicalDistribute[DistributionSpecGather]
-----PhysicalIntersect build RFs:RF0 x->cast(b1 as BIGINT) RF1 x->cast(b1 as
BIGINT) RF2 x->cast(c1 as BIGINT) RF3 x->cast(c1 as BIGINT)
+----PhysicalIntersect build RFs:RF0 b1->(cast(a1 as BIGINT) + 1) RF1
b1->(cast(a1 as BIGINT) + 1) RF2 b1->cast(c1 as BIGINT) RF3 b1->cast(c1 as
BIGINT)
------PhysicalDistribute[DistributionSpecHash]
--------PhysicalProject
-----------PhysicalOlapScan[a]
+----------PhysicalOlapScan[b]
------PhysicalDistribute[DistributionSpecHash]
--------PhysicalProject
-----------PhysicalOlapScan[b] apply RFs: RF0 RF1
+----------PhysicalOlapScan[a] apply RFs: RF0 RF1
------PhysicalDistribute[DistributionSpecHash]
--------PhysicalProject
----------PhysicalOlapScan[c] apply RFs: RF2 RF3
diff --git
a/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query24.out
b/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query24.out
index c3708d1c485..125c9e8a136 100644
--- a/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query24.out
+++ b/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query24.out
@@ -7,28 +7,28 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
--------PhysicalDistribute[DistributionSpecHash]
----------hashAgg[LOCAL]
------------PhysicalProject
---------------hashJoin[INNER_JOIN bucketShuffle]
hashCondition=((store_sales.ss_item_sk = store_returns.sr_item_sk) and
(store_sales.ss_ticket_number = store_returns.sr_ticket_number))
otherCondition=()
+--------------hashJoin[INNER_JOIN broadcast]
hashCondition=((customer.c_current_addr_sk = customer_address.ca_address_sk)
and (store.s_zip = customer_address.ca_zip)) otherCondition=(( not
(customer.c_birth_country = upper(customer_address.ca_country))))
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast]
hashCondition=((store_sales.ss_item_sk = item.i_item_sk)) otherCondition=()
+------------------hashJoin[INNER_JOIN broadcast]
hashCondition=((store_sales.ss_customer_sk = customer.c_customer_sk))
otherCondition=()
--------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN broadcast]
hashCondition=((store_sales.ss_customer_sk = customer.c_customer_sk) and
(store_sales.ss_store_sk = store.s_store_sk)) otherCondition=() build RFs:RF2
c_customer_sk->ss_customer_sk;RF3 s_store_sk->ss_store_sk
+----------------------hashJoin[INNER_JOIN broadcast]
hashCondition=((store_sales.ss_item_sk = item.i_item_sk)) otherCondition=()
------------------------PhysicalProject
---------------------------PhysicalOlapScan[store_sales] apply RFs: RF2 RF3
-------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN shuffle]
hashCondition=((customer.c_current_addr_sk = customer_address.ca_address_sk))
otherCondition=(( not (customer.c_birth_country =
upper(customer_address.ca_country)))) build RFs:RF1
ca_address_sk->c_current_addr_sk
-----------------------------PhysicalProject
-------------------------------PhysicalOlapScan[customer] apply RFs: RF1
+--------------------------hashJoin[INNER_JOIN broadcast]
hashCondition=((store_sales.ss_store_sk = store.s_store_sk)) otherCondition=()
build RFs:RF2 s_store_sk->ss_store_sk
----------------------------PhysicalProject
-------------------------------hashJoin[INNER_JOIN broadcast]
hashCondition=((store.s_zip = customer_address.ca_zip)) otherCondition=() build
RFs:RF0 s_zip->ca_zip
+------------------------------hashJoin[INNER_JOIN bucketShuffle]
hashCondition=((store_sales.ss_item_sk = store_returns.sr_item_sk) and
(store_sales.ss_ticket_number = store_returns.sr_ticket_number))
otherCondition=()
--------------------------------PhysicalProject
-----------------------------------PhysicalOlapScan[customer_address] apply
RFs: RF0
+----------------------------------PhysicalOlapScan[store_sales] apply RFs: RF2
--------------------------------PhysicalProject
-----------------------------------filter((store.s_market_id = 8))
-------------------------------------PhysicalOlapScan[store]
+----------------------------------PhysicalOlapScan[store_returns]
+----------------------------PhysicalProject
+------------------------------filter((store.s_market_id = 8))
+--------------------------------PhysicalOlapScan[store]
+------------------------PhysicalProject
+--------------------------PhysicalOlapScan[item]
--------------------PhysicalProject
-----------------------PhysicalOlapScan[item]
+----------------------PhysicalOlapScan[customer]
----------------PhysicalProject
-------------------PhysicalOlapScan[store_returns]
+------------------PhysicalOlapScan[customer_address]
--PhysicalResultSink
----PhysicalQuickSort[MERGE_SORT]
------PhysicalDistribute[DistributionSpecGather]
diff --git a/regression-test/suites/mtmv_p0/ivm/test_ivm_explain_refresh.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_explain_refresh.groovy
index bf2136a6456..3a272189c79 100644
--- a/regression-test/suites/mtmv_p0/ivm/test_ivm_explain_refresh.groovy
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_explain_refresh.groovy
@@ -65,10 +65,12 @@ suite("test_ivm_explain_refresh") {
// CREATE-time snapshot of affectQueryResultInPlan session variables
(LogicalPlanBuilder:
// CreateMTMVInfo -> getAffectQueryResultInPlanVariables), not from the
session running the
// EXPLAIN. The variables that could pin this plan (join distribution,
runtime filters,
- // ignore_shape_nodes) carry no affectQueryResultInPlan flag, so `set`
cannot reach them —
- // verified empirically (regenerating with ignore_shape_nodes set produced
identical output).
- // The normalization in explainIvmPlanWithoutStreamId is therefore the
only reliable way to
- // keep these shape expectations stable across environments.
+ // ignore_shape_nodes, disable_join_reorder) carry no
affectQueryResultInPlan flag, so `set`
+ // cannot reach them — verified empirically (regenerating with
ignore_shape_nodes set
+ // produced identical output). The normalization in
explainIvmPlanWithoutStreamId and the
+ // statistics the two tables publish before the plans are explained (see
below) are what
+ // keep these shape expectations stable.
+
sql """drop materialized view if exists test_ivm_explain_refresh_mv;"""
sql """drop table if exists test_ivm_explain_refresh_t1;"""
sql """drop table if exists test_ivm_explain_refresh_t2;"""
@@ -126,6 +128,28 @@ suite("test_ivm_explain_refresh") {
sql "INSERT INTO test_ivm_explain_refresh_t1 VALUES (1, 11), (4, 40);"
+ // Pin the statistics of the two tables before the plans below are
explained. The shape
+ // of the refresh plan follows them: while a table has no statistics, it
is planned at
+ // the rows its load reported, and as soon as the Automatic Analyzer
collected them the
+ // outer join of the pending stream plan is mirrored. The analyzer runs at
some point
+ // after the load and cannot be reached from this session (see the note
above), so the
+ // tables publish their exact statistics instead: t1 holds the physical
rows (1,10),
+ // (2,20), (1,11) and (4,40), t2 holds (1,100) and (3,300).
+ multi_sql """
+ alter table test_ivm_explain_refresh_t1 modify column k1
+ set stats ('row_count'='4', 'ndv'='3', 'data_size'='12',
'avg_size'='3', 'max_size'='4',
+ 'num_nulls'='0', 'min_value'='1', 'max_value'='4');
+ alter table test_ivm_explain_refresh_t1 modify column v1
+ set stats ('row_count'='4', 'ndv'='4', 'data_size'='16',
'avg_size'='4', 'max_size'='4',
+ 'num_nulls'='0', 'min_value'='10', 'max_value'='40');
+ alter table test_ivm_explain_refresh_t2 modify column k1
+ set stats ('row_count'='2', 'ndv'='2', 'data_size'='8',
'avg_size'='4', 'max_size'='4',
+ 'num_nulls'='0', 'min_value'='1', 'max_value'='3');
+ alter table test_ivm_explain_refresh_t2 modify column v2
+ set stats ('row_count'='2', 'ndv'='2', 'data_size'='8',
'avg_size'='4', 'max_size'='4',
+ 'num_nulls'='0', 'min_value'='100', 'max_value'='300');
+ """
+
explainIvmPlanWithoutStreamId("all_streams_shape_plan", """
EXPLAIN SHAPE PLAN REFRESH MATERIALIZED VIEW
test_ivm_explain_refresh_mv INCREMENTAL WITH ALL STREAMS
""")
diff --git
a/regression-test/suites/nereids_rules_p0/subquery_to_window_function/correlated_scalar_subquery_to_window_function.groovy
b/regression-test/suites/nereids_rules_p0/subquery_to_window_function/correlated_scalar_subquery_to_window_function.groovy
index 08d155811a3..d71e91f8146 100644
---
a/regression-test/suites/nereids_rules_p0/subquery_to_window_function/correlated_scalar_subquery_to_window_function.groovy
+++
b/regression-test/suites/nereids_rules_p0/subquery_to_window_function/correlated_scalar_subquery_to_window_function.groovy
@@ -34,6 +34,10 @@ suite("correlated_scalar_subquery_to_window_function") {
set push_topn_to_agg = true;
set topn_opt_limit_threshold=1024;
"""
+ // The join order of the shape plans below must not follow the row counts
the
+ // planner estimates, otherwise a change of the collected table statistics
+ // rewrites the expectations without any rule of this suite changing.
+ sql "set disable_join_reorder=true"
sql "DROP TABLE IF EXISTS fact FORCE"
sql "DROP TABLE IF EXISTS dim FORCE"
diff --git
a/regression-test/suites/query_p0/eager_agg/scalar_agg_pushdown.groovy
b/regression-test/suites/query_p0/eager_agg/scalar_agg_pushdown.groovy
index c9dce6c678a..c9c8e9bdded 100644
--- a/regression-test/suites/query_p0/eager_agg/scalar_agg_pushdown.groovy
+++ b/regression-test/suites/query_p0/eager_agg/scalar_agg_pushdown.groovy
@@ -20,6 +20,9 @@ suite("scalar_agg_pushdown") {
sql "set disable_nereids_rules='SALT_JOIN';"
sql "set runtime_filter_mode=OFF;"
sql 'set ignore_shape_nodes="PhysicalProject, PhysicalDistribute";'
+ // Keep the join order of the shape plans below independent of the
estimated
+ // row counts, so a change of the collected table statistics cannot
rewrite them.
+ sql "set disable_join_reorder=true"
// Test: scalar aggregation (no GROUP BY) must not be pushed below a join,
// because scalar agg on empty input returns 1 NULL row (SQL standard),
diff --git
a/regression-test/suites/shape_check/tpcds_sf100/noStatsRfPrune/query24.groovy
b/regression-test/suites/shape_check/tpcds_sf100/noStatsRfPrune/query24.groovy
index 881ba1887ca..33dcb7c4731 100644
---
a/regression-test/suites/shape_check/tpcds_sf100/noStatsRfPrune/query24.groovy
+++
b/regression-test/suites/shape_check/tpcds_sf100/noStatsRfPrune/query24.groovy
@@ -32,6 +32,9 @@ sql 'set enable_runtime_filter_prune=true'
sql 'set parallel_pipeline_task_num=8'
sql 'set forbid_unknown_col_stats=false'
sql 'set enable_stats=false'
+ // Keep the join order independent of the estimated row counts, so that a
change of
+ // the collected table statistics cannot rewrite the shape expectation
below.
+ sql 'set disable_join_reorder=true'
sql "set runtime_filter_type=8"
sql 'set broadcast_row_count_limit = 30000000'
sql 'set enable_nereids_timeout = false'
diff --git
a/regression-test/suites/statistics/test_row_count_after_truncate.groovy
b/regression-test/suites/statistics/test_row_count_after_truncate.groovy
new file mode 100644
index 00000000000..22f0e07027b
--- /dev/null
+++ b/regression-test/suites/statistics/test_row_count_after_truncate.groovy
@@ -0,0 +1,131 @@
+// 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_row_count_after_truncate") {
+ sql """drop database if exists test_row_count_after_truncate"""
+ sql """create database test_row_count_after_truncate"""
+ sql """use test_row_count_after_truncate"""
+
+ sql """CREATE TABLE test_row_count_after_truncate_tbl (
+ k INT NOT NULL,
+ v INT NULL
+ ) ENGINE = OLAP
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ )
+ """
+ sql """INSERT INTO test_row_count_after_truncate_tbl VALUES (1, 1), (2,
2), (3, 3), (4, 4), (5, 5)"""
+
+ // TRUNCATE TABLE removes all the data and the statistics record of the
table. The backends report the row
+ // count of the new tablets with a delay and the rows loaded right after
the truncation are not visible in
+ // that report either, so the row count of the table can only come from
the statistics record, which the
+ // load itself recreates when it becomes visible.
+ sql """TRUNCATE TABLE test_row_count_after_truncate_tbl"""
+ sql """INSERT INTO test_row_count_after_truncate_tbl VALUES (1, 1), (2,
2), (3, 3)"""
+
+ // The scan must report the rows of the table instead of -1 (which is
planned as 1 row), i.e. the record
+ // the load recreated has accumulated the rows it loaded.
+ explain {
+ sql """SELECT * FROM test_row_count_after_truncate_tbl"""
+ contains "cardinality=3, "
+ }
+
+ // The same from the stats record itself. The backend reports the new
tablets with a delay, so the record
+ // is the only place the rows loaded so far are visible.
+ String updatedRows = ""
+ for (int i = 0; i < 30; i++) {
+ def stats = sql """SHOW TABLE STATS
test_row_count_after_truncate_tbl"""
+ updatedRows = stats[0][0]
+ if (updatedRows == "3") {
+ break
+ }
+ Thread.sleep(1000)
+ }
+ assertEquals("3", updatedRows)
+
+ // A scan which selects an aggregate rollup index must report the row
count of that index. The rows the
+ // load reports are only known for the base index, they must not be added
to an index which aggregates
+ // them, otherwise the rollup is estimated with the rows of the whole
table.
+ sql """DROP TABLE IF EXISTS test_row_count_after_truncate_aggr"""
+ sql """CREATE TABLE test_row_count_after_truncate_aggr (
+ k1 INT NOT NULL,
+ k2 INT NOT NULL,
+ v INT SUM NOT NULL
+ ) ENGINE = OLAP
+ AGGREGATE KEY(k1, k2)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ )
+ """
+ sql """ALTER TABLE test_row_count_after_truncate_aggr ADD ROLLUP r1 (k1,
v)"""
+ waitForSchemaChangeDone {
+ sql """SHOW ALTER TABLE ROLLUP WHERE
TableName='test_row_count_after_truncate_aggr' ORDER BY CreateTime DESC LIMIT
1"""
+ time 600
+ }
+ sql """INSERT INTO test_row_count_after_truncate_aggr SELECT 1, number, 1
FROM numbers("number" = "100")"""
+ sql """TRUNCATE TABLE test_row_count_after_truncate_aggr"""
+ sql """INSERT INTO test_row_count_after_truncate_aggr SELECT 1, number, 1
FROM numbers("number" = "100")"""
+
+ // 100 rows are loaded into the base index, the rollup aggregates them
into a single row. Whether the
+ // backends have reported the new tablets or not, the rollup scan reports
its own row count and the
+ // base index scan reports the 100 loaded rows.
+ explain {
+ sql """SELECT k1, sum(v) FROM test_row_count_after_truncate_aggr GROUP
BY k1"""
+ contains "(r1)"
+ contains "cardinality=1, "
+ }
+ explain {
+ sql """SELECT k2 FROM test_row_count_after_truncate_aggr"""
+ contains "cardinality=100, "
+ }
+
+ // A rollup of a duplicate key table which keeps one row per base row has
the same row count as the base
+ // index, so the rows the load reports are its rows as well. The rollup
drops a key column
+ // (k2), which is what makes it a rollup the analyzer accepts: a
projection whose columns are a prefix of
+ // the base table's columns is rejected as useless for a duplicate key
table.
+ sql """DROP TABLE IF EXISTS test_row_count_after_truncate_dup_rollup"""
+ sql """CREATE TABLE test_row_count_after_truncate_dup_rollup (
+ k1 INT NOT NULL,
+ k2 INT NOT NULL,
+ v INT NULL
+ ) ENGINE = OLAP
+ DUPLICATE KEY(k1, k2)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ )
+ """
+ sql """ALTER TABLE test_row_count_after_truncate_dup_rollup ADD ROLLUP
r_dup (k1, v)"""
+ waitForSchemaChangeDone {
+ sql """SHOW ALTER TABLE ROLLUP WHERE
TableName='test_row_count_after_truncate_dup_rollup' ORDER BY CreateTime DESC
LIMIT 1"""
+ time 600
+ }
+ sql """INSERT INTO test_row_count_after_truncate_dup_rollup VALUES (1, 1,
1), (2, 2, 2), (3, 3, 3)"""
+ sql """TRUNCATE TABLE test_row_count_after_truncate_dup_rollup"""
+ sql """INSERT INTO test_row_count_after_truncate_dup_rollup VALUES (1, 1,
1), (2, 2, 2), (3, 3, 3)"""
+
+ // The scan of the rollup reports the rows loaded into the table: it keeps
one row per base row, so
+ // neither its own -1 (the backends have not reported the new rollup
tablets yet) nor 1 is right.
+ explain {
+ sql """SELECT k1, v FROM test_row_count_after_truncate_dup_rollup
INDEX r_dup"""
+ contains "(r_dup)"
+ contains "cardinality=3, "
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
