englefly commented on code in PR #69028:
URL: https://github.com/apache/doris/pull/69028#discussion_r4355069153
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/AnalysisManager.java:
##########
@@ -1298,12 +1344,12 @@ public void updateUpdatedRows(Map<Long, Long>
partitionToUpdateRows, long dbId,
}
// Invoke this for cloud version load.
- public void updateUpdatedRows(Map<Long, Long> updatedRows) {
+ public void updateUpdatedRows(Map<Long, Long> updatedRows, long dbId) {
try {
if (!Env.getCurrentEnv().isMaster() || Env.isCheckpointThread()) {
return;
}
- UpdateRowsEvent updateRowsEvent = new UpdateRowsEvent(updatedRows);
+ UpdateRowsEvent updateRowsEvent = new UpdateRowsEvent(dbId,
updatedRows);
Review Comment:
Fixed in 50074783a42.
`AnalysisManager.updateUpdatedRows(Map, dbId)` — the cloud report path — now
holds the table write
lock of the reported tables while it applies **and** journals the rows, i.e.
the same lock which a
whole table `DROP STATS` takes to remove the record and journal the
deletion. The journal order is
then always the order in which the mutations are applied, so both
interleavings converge:
- report first: journal `[UPDATE_ROWS][DELETE]` -> master create(N) then
remove, replay create(N) then remove
- DROP first: journal `[DELETE][UPDATE_ROWS]` -> master remove then
create(N), replay remove then create(N)
The tables are locked in id order with the 10 second timeout
`finishTransaction` uses; when the lock
cannot be acquired the report is dropped with a warning and the row count
stays unknown until the
backends report it.
Test: `AnalysisManagerTest#testCloudRowUpdateIsSerializedWithTheTableLock`
holds the table write lock,
reports rows from another thread and asserts that neither the record nor the
journal entry appears
while the lock is held (and that both appear after the unlock). It fails
without the change with
"the report was applied while the table was locked".
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/AnalysisManager.java:
##########
@@ -1355,7 +1401,7 @@ public void replayUpdateRowsRecord(UpdateRowsEvent event)
{
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());
Review Comment:
Fixed in d1aca0b7213.
The partition generation of the report is now validated:
`CloudGlobalTransactionMgr.afterCommitTxnResp`
hands the `<table_id, partition_id>` pairs which `CommitTxnResponse` already
carries to
`AnalysisManager.updateUpdatedRows`, which drops a table's rows when any of
the partitions the report
counted no longer belongs to the table (a whole table `TRUNCATE` replaces
the partitions and keeps the
table id). The check runs inside the table write lock the report takes, so
the generation cannot change
between the check and the application, and a dropped report is neither
applied nor journaled — so the
record the next load recreates only counts the rows of the current
partitions.
Tests:
`AnalysisManagerTest#testReportOfAPartitionRemovedByTruncateIsNotApplied` (a
delayed report of
100 rows for the replaced partition is ignored; the reload's 3 rows recreate
the record with 3 rows,
not 103) and `AnalysisManagerTest#testReportOfRemovedPartitionsIsRecognized`.
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/analysis/AnalysisManager.java:
##########
@@ -1366,52 +1412,52 @@ public void replayUpdateRowsRecord(UpdateRowsEvent
event) {
// 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);
Review Comment:
Fixed in b01c9770473.
The removal moved into the same critical section which journals `FINISHED`,
i.e. before
`tbl.writeUnlock()`, so a load can no longer become visible in between:
either the load ran before the
schema change finished (the record it updated is removed by the job on both
the master and the replay)
or after (the record it recreates exists on both).
`removeTableStatsOnFinished` asserts that the table
write lock is held, so the call cannot be moved out of the critical section
again; dropping the
persisted statistics and invalidating the caches (`dropStats`) stays outside
the lock because it does
not touch the record.
Test:
`SchemaChangeJobV2StatsTest#testRemoveTableStatsOnFinishedRequiresTheTableWriteLock`
(the removal
is rejected without the table write lock and removes the record with it).
--
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]
