englefly commented on code in PR #68282:
URL: https://github.com/apache/doris/pull/68282#discussion_r4118252533
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/TableStatsMeta.java:
##########
@@ -130,6 +140,94 @@ public TableStatsMeta(long rowCount, AnalysisInfo
analyzedJob, TableIf table) {
update(analyzedJob, table);
}
+ /**
+ * Create a record for a table which doesn't have one yet, in the state of
an empty table. The rows
+ * loaded into the table are accumulated by {@link
AnalysisManager#replayUpdateRowsRecord}, so a record
+ * has to exist before the first load, otherwise these rows can never be
turned into a row count.
+ */
+ 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;
+ this.indexesRowCount = buildEmptyIndexRowCount(table);
+ this.updatedRowsBase.set(0);
+ }
+
+ /**
+ * TRUNCATE TABLE removes all the data of the table. Reset this record
back to the state of an empty
+ * table instead of dropping it, so that the rows loaded after the
truncation can still be accumulated
+ * into {@link #updatedRows} and be reported as the row count of the table.
+ * <p>
+ * The transition runs under the monitor of this record, which {@link
#getRowCountWithDeltaRows} also
+ * takes, so a planner which reads the row count of the table without
holding its lock either sees the
+ * whole transition or none of it. The order the fields are published in
matters for the readers which
+ * don't take the monitor, for instance SHOW TABLE STATS: the baseline
first makes the delta empty while
+ * the collected row count is still the one of the removed data, so the
emptied row count is only
+ * published once no row of the removed data is counted as a delta row
anymore.
+ */
+ public synchronized void reset(OlapTable table) {
+ updatedRowsBase.set(updatedRows.get());
+ indexesRowCount = buildEmptyIndexRowCount(table);
+ updatedRows.set(0);
+ // None of the rows loaded from now on is included in the collected
row count. They are all delta rows.
+ updatedRowsBase.set(0);
+ rowCount = 0;
+ partitionUpdateRows.clear();
+ // Drop the column statistics baseline: the row count captured by the
previous analysis described
+ // the removed data, it must not cancel out the rows loaded after the
truncation.
+ colToColStatsMeta.clear();
+ // The statistics of the removed data is stale, let the analyzer
collect it again.
+ partitionChanged.set(true);
+ // The injected statistics described the removed data, it no longer
applies to this table.
+ userInjected = false;
+ // The emptied table has never been analyzed, and no analyze job
describes it any more.
+ updatedTime = 0;
+ lastAnalyzeTime = 0;
+ jobType = null;
+ }
+
+ private static ConcurrentMap<Long, Long> buildEmptyIndexRowCount(OlapTable
table) {
+ // TRUNCATE TABLE removed the data of every index of the table, so
every index whose row count is known
+ // to follow the row count of the base index is known to be empty. The
row count of an index which
+ // aggregates is unknown until the backends report it, so it is not
claimed to be 0 here.
+ 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()) {
Review Comment:
Confirmed and reproduced on the reviewed head, with your exact scenario.
Thank you for the precise numbers.
```
CREATE TABLE u (k INT, v INT) UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1;
TRUNCATE TABLE u;
INSERT INTO u VALUES (1, 10); -- transaction 1: one rowset, 1 row
INSERT INTO u VALUES (1, 20); -- transaction 2: another rowset, same
full key
SELECT count(*) FROM u; -- 1, the scan merges the key
EXPLAIN SELECT * FROM u; -- cardinality=2 <-- the physical row
count
SHOW TABLE STATS u; -- updated_rows=2, row_count=0
```
The delta is physical: `AnalysisManager.replayUpdateRowsRecord()`
accumulates `rowset_meta.num_rows()` per transaction
(`updatedRows.addAndGet(tableUpdateRows)`), so two transactions writing the
same full key contribute 2 to `updatedRows` while the scan's logical
cardinality is 1. The physical-delta semantics are pre-existing, but after a
truncation this delta becomes the only source of the row count, so the window
which used to report `-1` (and be clamped to 1) now reports 2, and
`computeDeltaRowCount()` feeds the same 2 into filter estimation.
Fix I intend (holding the code change for one round, see my reply on
`r4090300621`): classify the base index by key semantics as well, so a
merge-key base index does not get the physical delta for a row count that is a
logical one - either leave the fallback unknown or carry a logical delta - plus
a repeated-full-key case for a `UNIQUE` base and an `AGG` base.
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/TableStatsMeta.java:
##########
@@ -130,6 +140,94 @@ public TableStatsMeta(long rowCount, AnalysisInfo
analyzedJob, TableIf table) {
update(analyzedJob, table);
}
+ /**
+ * Create a record for a table which doesn't have one yet, in the state of
an empty table. The rows
+ * loaded into the table are accumulated by {@link
AnalysisManager#replayUpdateRowsRecord}, so a record
+ * has to exist before the first load, otherwise these rows can never be
turned into a row count.
+ */
+ 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;
+ this.indexesRowCount = buildEmptyIndexRowCount(table);
+ this.updatedRowsBase.set(0);
+ }
+
+ /**
+ * TRUNCATE TABLE removes all the data of the table. Reset this record
back to the state of an empty
+ * table instead of dropping it, so that the rows loaded after the
truncation can still be accumulated
+ * into {@link #updatedRows} and be reported as the row count of the table.
+ * <p>
+ * The transition runs under the monitor of this record, which {@link
#getRowCountWithDeltaRows} also
+ * takes, so a planner which reads the row count of the table without
holding its lock either sees the
+ * whole transition or none of it. The order the fields are published in
matters for the readers which
+ * don't take the monitor, for instance SHOW TABLE STATS: the baseline
first makes the delta empty while
+ * the collected row count is still the one of the removed data, so the
emptied row count is only
+ * published once no row of the removed data is counted as a delta row
anymore.
+ */
+ public synchronized void reset(OlapTable table) {
+ updatedRowsBase.set(updatedRows.get());
+ indexesRowCount = buildEmptyIndexRowCount(table);
+ updatedRows.set(0);
+ // None of the rows loaded from now on is included in the collected
row count. They are all delta rows.
+ updatedRowsBase.set(0);
+ rowCount = 0;
+ partitionUpdateRows.clear();
+ // Drop the column statistics baseline: the row count captured by the
previous analysis described
+ // the removed data, it must not cancel out the rows loaded after the
truncation.
+ colToColStatsMeta.clear();
+ // The statistics of the removed data is stale, let the analyzer
collect it again.
+ partitionChanged.set(true);
+ // The injected statistics described the removed data, it no longer
applies to this table.
+ userInjected = false;
+ // The emptied table has never been analyzed, and no analyze job
describes it any more.
+ updatedTime = 0;
+ lastAnalyzeTime = 0;
+ jobType = null;
+ }
+
+ private static ConcurrentMap<Long, Long> buildEmptyIndexRowCount(OlapTable
table) {
+ // TRUNCATE TABLE removed the data of every index of the table, so
every index whose row count is known
+ // to follow the row count of the base index is known to be empty. The
row count of an index which
+ // aggregates is unknown until the backends report it, so it is not
claimed to be 0 here.
+ 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)
{
Review Comment:
Accepted at code level; I have to be honest that I could not reproduce it on
my cluster yet.
What I verified in the code:
- `MaterializedIndexMeta.whereClause` is persisted and copied by
`SchemaChangeJobV2`
(`shadowIndexMeta.setWhereClause(originalIndexMeta.getWhereClause())`).
- The sink honours it: `OlapTableSink` and `Nereids`'
`BindSink.createSyncMvWhereClause()` skip the base index and indexes without a
where clause, so a filtered index only ever receives the rows which satisfy the
predicate.
- `MaterializedViewHandler` rejects a plain projection of a duplicate key
table as useless ("Rollup contain the columns of the base table in prefix order
for duplicate table is useless") *unless* there is a where clause, which is
exactly how a filtered `DUP_KEYS`/`NONE` index comes into existence.
So with 100 base rows of which one satisfies the predicate and the tablets
unreported, `keepsOneRowPerBaseRow()` returns true for that index, the reset
seeds it with 0 and both consumers report `0 + 100`, while the index holds one
row.
What I could not do here: `CREATE MATERIALIZED VIEW mv_f AS SELECT k AS mk,
v AS mv FROM t WHERE k > 98` on my local FE did not attach an index to the base
table (the statement went to the async path and the base table kept only its
own index), so I have no live numbers for this one. If you can point me at the
DDL (or the version/flag) you used for the filtered synchronous MV, I will
reproduce it exactly and record the numbers here.
Fix I intend: require `indexMeta.getWhereClause() == null` in the predicate,
and add a filtered projection MV case to the regression suite. Same as the
other two findings, I am holding the code change for this round until the
design of the per-index delta is settled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
