This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 61d51d9dcd3 branch-4.2: [fix](statistics) Create the table stats
record when a load finishes #69028 (#70139)
61d51d9dcd3 is described below
commit 61d51d9dcd3ad816633d3e8bcff653def4298af4
Author: morrySnow <[email protected]>
AuthorDate: Thu Nov 5 09:26:01 2026 +0800
branch-4.2: [fix](statistics) Create the table stats record when a load
finishes #69028 (#70139)
picked from master #69028
Co-authored-by: minghong <[email protected]>
---
.../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 +-
.../apache/doris/statistics/AnalysisManager.java | 251 +++++++++---
.../doris/statistics/StatisticsAutoCollector.java | 4 +-
.../apache/doris/statistics/StatisticsCleaner.java | 38 ++
.../statistics/StatisticsMetricCollector.java | 6 +-
.../apache/doris/statistics/TableStatsMeta.java | 150 ++++++-
.../apache/doris/statistics/UpdateRowsEvent.java | 11 +-
.../doris/alter/SchemaChangeJobV2StatsTest.java | 59 +++
.../doris/nereids/stats/StatsCalculatorTest.java | 45 +++
.../doris/statistics/AnalysisManagerTest.java | 436 +++++++++++++++++++++
.../doris/statistics/StatisticsCleanerTest.java | 83 ++++
.../doris/statistics/TableStatsMetaTest.java | 289 ++++++++++++++
.../nereids_p0/eager_agg/scalar_agg_pushdown.out | 8 +-
.../data/query_p0/runtimefilterV2/rfv2.out | 18 +-
.../tpcds_sf100/noStatsRfPrune/query24.out | 20 +-
.../eager_agg/scalar_agg_pushdown.groovy | 3 +
...lated_scalar_subquery_to_window_function.groovy | 4 +
.../tpcds_sf100/noStatsRfPrune/query24.groovy | 3 +
.../test_row_count_after_truncate.groovy | 131 +++++++
23 files changed, 1560 insertions(+), 95 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 65231191f25..57b24fae28f 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
@@ -55,7 +55,6 @@ import
org.apache.doris.nereids.rules.expression.check.CheckCast;
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.AnalysisManager;
import org.apache.doris.task.AgentBatchTask;
import org.apache.doris.task.AgentTask;
import org.apache.doris.task.AgentTaskExecutor;
@@ -69,6 +68,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;
@@ -714,18 +714,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 255f279be5a..79a91087c66 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
@@ -273,6 +273,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 9f8a133491e..b99382f82e0 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
@@ -523,6 +523,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()) {
@@ -532,7 +541,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 e497f6ba6c2..241b48d3437 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
@@ -155,6 +155,7 @@ import
org.apache.doris.statistics.StatisticsCache.OlapTableStatistics;
import org.apache.doris.statistics.TableStatsMeta;
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;
@@ -409,14 +410,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;
}
@@ -499,7 +507,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 f9bc380bb51..05ddf6fb663 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/AnalysisManager.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
index 1de58345dc1..3e6836b4400 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
@@ -39,6 +39,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;
@@ -737,8 +738,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.
@@ -1236,6 +1236,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 {
@@ -1258,17 +1305,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
@@ -1315,7 +1438,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());
}
@@ -1326,52 +1449,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();
@@ -1447,6 +1570,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/StatisticsAutoCollector.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
index 0dbe97f1ba1..3285ca783ee 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/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.iceberg.IcebergExternalTable;
-import org.apache.doris.persist.TableStatsDeletionLog;
import org.apache.doris.statistics.AnalysisInfo.AnalysisMethod;
import org.apache.doris.statistics.AnalysisInfo.JobType;
import org.apache.doris.statistics.AnalysisInfo.ScheduleType;
@@ -200,8 +199,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/StatisticsCleaner.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsCleaner.java
index 0ff2963683b..8482f75d78a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsCleaner.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsCleaner.java
@@ -31,6 +31,7 @@ import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.persist.TableStatsDeletionLog;
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;
@@ -137,6 +138,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) {
@@ -145,6 +153,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/main/java/org/apache/doris/statistics/StatisticsMetricCollector.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsMetricCollector.java
index 838ba9736ce..8263c4ec71f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsMetricCollector.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsMetricCollector.java
@@ -85,7 +85,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/TableStatsMeta.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/TableStatsMeta.java
index 0b77aba91e3..66e8958973d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/TableStatsMeta.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/TableStatsMeta.java
@@ -17,7 +17,10 @@
package org.apache.doris.statistics;
+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/UpdateRowsEvent.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/UpdateRowsEvent.java
index c9a0b9d820d..9f5f2615b23 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/UpdateRowsEvent.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/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/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..cf6058e6f79
--- /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.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 afa66c48159..9f3da28fb39 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.LogicalExcept;
@@ -384,6 +389,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/AnalysisManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java
index f2c1f313044..df66b1b8346 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java
@@ -23,9 +23,12 @@ 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.Table;
import org.apache.doris.catalog.TableIf;
+import org.apache.doris.catalog.TableProperty;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.Pair;
@@ -42,6 +45,8 @@ import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.plans.commands.AnalyzeTableCommand;
import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.persist.TableStatsDeletionLog;
import org.apache.doris.statistics.AnalysisInfo.AnalysisMethod;
import org.apache.doris.statistics.AnalysisInfo.AnalysisType;
import org.apache.doris.statistics.AnalysisInfo.JobType;
@@ -51,6 +56,7 @@ 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 mockit.Mock;
import mockit.MockUp;
import mockit.Mocked;
@@ -68,6 +74,12 @@ 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;
@@ -75,8 +87,12 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
// CHECKSTYLE OFF
public class AnalysisManagerTest {
@@ -579,4 +595,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/StatisticsCleanerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsCleanerTest.java
index 32524b0c132..2eb2c601923 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsCleanerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsCleanerTest.java
@@ -17,10 +17,16 @@
package org.apache.doris.statistics;
+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.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;
@@ -28,6 +34,9 @@ import org.mockito.ArgumentMatchers;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.util.ArrayList;
+import java.util.List;
+
class StatisticsCleanerTest {
@Test
@@ -50,4 +59,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/fe/fe-core/src/test/java/org/apache/doris/statistics/TableStatsMetaTest.java
b/fe/fe-core/src/test/java/org/apache/doris/statistics/TableStatsMetaTest.java
index 83e68e3090c..2a97dba45f6 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/statistics/TableStatsMetaTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/statistics/TableStatsMetaTest.java
@@ -17,20 +17,76 @@
package org.apache.doris.statistics;
+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 mockit.Mock;
import mockit.MockUp;
import mockit.Mocked;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
import java.util.HashSet;
import java.util.List;
+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(@Mocked OlapTable table) {
TableStatsMeta tableStatsMeta = new TableStatsMeta();
@@ -68,4 +124,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/regression-test/data/nereids_p0/eager_agg/scalar_agg_pushdown.out
b/regression-test/data/nereids_p0/eager_agg/scalar_agg_pushdown.out
index 8fa4341a049..975f6894552 100644
--- a/regression-test/data/nereids_p0/eager_agg/scalar_agg_pushdown.out
+++ b/regression-test/data/nereids_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/runtimefilterV2/rfv2.out
b/regression-test/data/query_p0/runtimefilterV2/rfv2.out
index 050c42ceaee..9659fc3df12 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 RFV2: RF0[a1->b1] RF1[a1->b1] RF2[a1->c1] RF3[a1->c1]
+----PhysicalIntersect RFV2: RF0[b1->a1] RF1[b1->a1] RF2[b1->c1] RF3[b1->c1]
------PhysicalDistribute[DistributionSpecHash]
---------PhysicalOlapScan[a]
+--------PhysicalOlapScan[b]
------PhysicalDistribute[DistributionSpecHash]
---------PhysicalOlapScan[b] RFV2: RF0 RF1
+--------PhysicalOlapScan[a] RFV2: RF0 RF1
------PhysicalDistribute[DistributionSpecHash]
--------PhysicalOlapScan[c] RFV2: RF2 RF3
-- !2 --
PhysicalResultSink
--PhysicalDistribute[DistributionSpecGather]
-----PhysicalIntersect RFV2: RF0[x->cast(b1 as BIGINT)] RF1[x->cast(b1 as
BIGINT)]
+----PhysicalIntersect RFV2: 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] RFV2: RF0 RF1
+----------PhysicalOlapScan[a] RFV2: RF0 RF1
-- !3 --
PhysicalResultSink
--PhysicalDistribute[DistributionSpecGather]
-----PhysicalIntersect RFV2: 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 RFV2: 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] RFV2: RF0 RF1
+----------PhysicalOlapScan[a] RFV2: RF0 RF1
------PhysicalDistribute[DistributionSpecHash]
--------PhysicalProject
----------PhysicalOlapScan[c] RFV2: 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 95e1d01aeae..fa0e822fa25 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 broadcast] hashCondition=((store.s_zip =
customer_address.ca_zip) and (store_sales.ss_store_sk = store.s_store_sk))
otherCondition=() build RFs:RF5 s_zip->[ca_zip];RF6 s_store_sk->[ss_store_sk]
+--------------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
(c_birth_country = upper(ca_country))))
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast]
hashCondition=((customer.c_current_addr_sk = customer_address.ca_address_sk))
otherCondition=(( not (c_birth_country = upper(ca_country)))) build RFs:RF4
ca_address_sk->[c_current_addr_sk]
+------------------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))
otherCondition=() build RFs:RF3 c_customer_sk->[ss_customer_sk]
+----------------------hashJoin[INNER_JOIN broadcast]
hashCondition=((store_sales.ss_item_sk = item.i_item_sk)) otherCondition=()
------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN broadcast]
hashCondition=((store_sales.ss_item_sk = item.i_item_sk)) otherCondition=()
+--------------------------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 colocated]
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[store_sales] apply RFs: RF3
RF6
+----------------------------------PhysicalOlapScan[store_sales] apply RFs: RF2
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[store_returns]
----------------------------PhysicalProject
-------------------------------PhysicalOlapScan[item]
+------------------------------filter((store.s_market_id = 8))
+--------------------------------PhysicalOlapScan[store]
------------------------PhysicalProject
---------------------------PhysicalOlapScan[customer] apply RFs: RF4
+--------------------------PhysicalOlapScan[item]
--------------------PhysicalProject
-----------------------PhysicalOlapScan[customer_address] apply RFs: RF5
+----------------------PhysicalOlapScan[customer]
----------------PhysicalProject
-------------------filter((store.s_market_id = 8))
---------------------PhysicalOlapScan[store]
+------------------PhysicalOlapScan[customer_address]
--PhysicalResultSink
----PhysicalQuickSort[MERGE_SORT]
------PhysicalDistribute[DistributionSpecGather]
diff --git
a/regression-test/suites/nereids_p0/eager_agg/scalar_agg_pushdown.groovy
b/regression-test/suites/nereids_p0/eager_agg/scalar_agg_pushdown.groovy
index c9dce6c678a..c9c8e9bdded 100644
--- a/regression-test/suites/nereids_p0/eager_agg/scalar_agg_pushdown.groovy
+++ b/regression-test/suites/nereids_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/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/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]
