This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 966a917a561 [fix](row-binlog) Keep row binlog tablets on the binlog
compaction policy (#69240)
966a917a561 is described below
commit 966a917a5617114ba340a5b01f1b239f2d85b224
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Oct 19 07:15:36 2026 +0100
[fix](row-binlog) Keep row binlog tablets on the binlog compaction policy
(#69240)
### What problem does this PR solve?
Issue Number: None
Related PR: #68033, #65810
Problem Summary:
**In short.** In local (non-cloud) mode, altering a compaction setting
of a table with `binlog.format = 'ROW'` also reaches its hidden row
binlog tablets. A new `compaction_policy` moves them off the `binlog`
policy they must use. A `time_series_compaction_*` change on a
`time_series` table fails halfway. #68033 fixed the same thing for
cloud. This PR applies the same fix to the local path: the FE no longer
sends compaction settings to row binlog tablets. As a second guard, the
BE now refuses to change a row binlog tablet's policy.
**Background.**
- A table with `binlog.format = 'ROW'` has a hidden row binlog index.
Each data tablet has a row binlog tablet next to it, which stores the
change events.
- Row binlog tablets are created with the `binlog` compaction policy
(`BinlogCumulativeCompactionPolicy`). It is a tiered policy written for
change events; for example, its quick merge links the segments of
several rowsets into one `OVERLAPPING` rowset instead of rewriting them.
`CreateReplicaTask` sets it when it creates a row binlog tablet.
- `ALTER TABLE ... SET (<compaction property>)` sends the new value to
the tablets through `UpdateTabletMetaInfoTask`. #68033 made the cloud
handler (`CloudSchemaChangeHandler`) leave out row binlog tablets for
`compaction_policy`. The local handler
(`SchemaChangeHandler.updatePartitionProperties`) still uses
`getMaterializedIndices(VISIBLE, true)`, so the row binlog index is
included.
**The problem, and what it cost.** Both cases are reproduced on master
in local mode:
| Statement on a table with `binlog.format = 'ROW'` | What happens on
master |
|---|---|
| `ALTER TABLE t SET ("compaction_policy" = "time_series")` | Succeeds,
and every row binlog tablet meta now says `time_series`. From the next
compaction on, those tablets are compacted by the time series policy
instead of the binlog policy. |
| `ALTER TABLE t SET ("time_series_compaction_goal_size_mbytes" =
"2048")` on a `time_series` table | Fails: `Failed to update
partition[t]. tablet meta. Error: ...[INVALID_ARGUMENT]only time series
compaction policy support time series config]. This operation maybe
partial successfully, You should retry until success.` The BE rejects
time series settings for tablets that are not on the time series policy,
and the row binlog tablets are on `binlog`. The data tablets are already
updated by then. Retrying fails the same way. |
**How this PR fixes it.**
1. FE, `SchemaChangeHandler.updatePartitionProperties`: when the update
carries a compaction policy or a time series setting, the row binlog
index is left out. This is the same rule #68033 uses for cloud.
- `ALTER TABLE ... SET` changes one table property at a time
(`ModifyTablePropertiesOp.validate`), so such an update has nothing else
for the row binlog tablets.
- Every other tablet meta update (in memory, storage policy, binlog
config, `disable_auto_compaction`, ...) still reaches them, as before.
2. BE, `update_tablet_meta_callback`:
- It refuses to change the compaction policy of a row binlog tablet
(`can not change compaction policy of row binlog tablet <id>`).
- It no longer accepts `binlog` as a new policy for any tablet. #65810
added it to the accepted values, but no FE path sends it: the policy is
only set when a row binlog tablet is created.
- With these checks, an older FE, or a future caller, gets an error
instead of silently moving a row binlog tablet off its policy.
**Results.**
| Statement (local mode, table with `binlog.format = 'ROW'`) | Before |
After |
|---|---|---|
| `SET ("compaction_policy" = "time_series")` | Succeeds; row binlog
tablets switch to `time_series` | Succeeds; data tablets switch, row
binlog tablets stay on `binlog` |
| `SET ("time_series_compaction_goal_size_mbytes" = "2048")` on a
`time_series` table | Fails after updating the data tablets | Succeeds;
data tablets get 2048, row binlog tablets are untouched |
| The first statement from an FE without this fix, against a BE with it
| Succeeds; row binlog tablets switch to `time_series` | Fails with `can
not change compaction policy of row binlog tablet <id>`; row binlog
tablets stay on `binlog` |
---
be/src/agent/task_worker_pool.cpp | 13 ++-
.../apache/doris/alter/SchemaChangeHandler.java | 7 +-
.../doris/alter/SchemaChangeHandlerTest.java | 60 ++++++++++
.../test_row_binlog_alter_compaction_policy.out | 13 +++
.../test_row_binlog_alter_compaction_policy.groovy | 124 +++++++++++++++++++++
5 files changed, 212 insertions(+), 5 deletions(-)
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 96ad3390b30..2e584918976 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -73,7 +73,6 @@
#include "runtime/snapshot_loader.h"
#include "runtime/user_function_cache.h"
#include "service/backend_options.h"
-#include "storage/compaction/cumulative_compaction_binlog_policy.h"
#include "storage/compaction/cumulative_compaction_policy.h"
#include "storage/compaction/cumulative_compaction_time_series_policy.h"
#include "storage/data_dir.h"
@@ -910,12 +909,18 @@ void update_tablet_meta_callback(StorageEngine& engine,
const TAgentTaskRequest&
need_to_save = true;
}
if (tablet_meta_info.__isset.compaction_policy) {
+ // The binlog policy is set when a row binlog tablet is created
and never changes.
+ if (tablet->is_row_binlog_tablet()) {
+ status = Status::InvalidArgument(
+ "can not change compaction policy of row binlog tablet
{}",
+ tablet_meta_info.tablet_id);
+ continue;
+ }
if (tablet_meta_info.compaction_policy !=
CUMULATIVE_SIZE_BASED_POLICY &&
- tablet_meta_info.compaction_policy !=
CUMULATIVE_TIME_SERIES_POLICY &&
- tablet_meta_info.compaction_policy !=
CUMULATIVE_BINLOG_POLICY) {
+ tablet_meta_info.compaction_policy !=
CUMULATIVE_TIME_SERIES_POLICY) {
status = Status::InvalidArgument(
"invalid compaction policy, only support for
size_based or "
- "time_series or binlog");
+ "time_series");
continue;
}
tablet->tablet_meta()->set_compaction_policy(tablet_meta_info.compaction_policy);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java
index 11129f63e1b..b17fcdb172e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java
@@ -3101,9 +3101,14 @@ public class SchemaChangeHandler extends AlterHandler {
String partitionName = partition.getName();
// Use the captured partition identity, including temporary partitions
with the same name.
Map<Long, Set<Pair<Long, Integer>>> beIdToTabletIdWithHash =
Maps.newHashMap();
+ // Row binlog tablets keep the binlog compaction policy they are
created with, so a new compaction
+ // policy or time series setting must not reach them. ALTER sets one
table property at a time,
+ // so such an update has nothing else for them.
+ boolean includeRowBinlog = compactionPolicy == null
+ && (timeSeriesCompactionConfig == null ||
timeSeriesCompactionConfig.isEmpty());
olapTable.readLock();
try {
- for (MaterializedIndex index :
partition.getMaterializedIndices(IndexExtState.VISIBLE, true)) {
+ for (MaterializedIndex index :
partition.getMaterializedIndices(IndexExtState.VISIBLE, includeRowBinlog)) {
int schemaHash =
olapTable.getSchemaHashByIndexId(index.getId());
for (Tablet tablet : index.getTablets()) {
for (Replica replica : tablet.getReplicas()) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
index 4f0c2015663..15e1e648213 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
@@ -23,8 +23,11 @@ import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.Index;
import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.MaterializedIndex;
+import org.apache.doris.catalog.MaterializedIndex.IndexExtState;
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.catalog.ScalarType;
import org.apache.doris.catalog.Table;
@@ -32,6 +35,7 @@ import org.apache.doris.catalog.info.ColumnPosition;
import org.apache.doris.catalog.info.IndexType;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.FeConstants;
+import org.apache.doris.common.Pair;
import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.indexpolicy.DropIndexPolicyLog;
import org.apache.doris.indexpolicy.IndexPolicy;
@@ -43,6 +47,7 @@ import
org.apache.doris.nereids.trees.plans.commands.AlterTableCommand;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.task.UpdateTabletMetaInfoTask;
import org.apache.doris.utframe.TestWithFeService;
import com.google.common.collect.Lists;
@@ -52,12 +57,14 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
import org.mockito.Mockito;
import java.lang.reflect.Method;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntSupplier;
import java.util.stream.Collectors;
@@ -331,6 +338,59 @@ public class SchemaChangeHandlerTest extends
TestWithFeService {
Assertions.assertFalse(cols.contains(Column.generateBeforeColName(Column.SEQUENCE_COL)));
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testAlterCompactionSettingSkipsRowBinlogTablets() throws
Exception {
+ String tableName = "binlog_compaction_setting";
+ String create = "CREATE TABLE IF NOT EXISTS test." + tableName + " (\n"
+ + "k1 INT NOT NULL,\n"
+ + "v1 INT\n"
+ + ")\n"
+ + "DUPLICATE KEY(k1)\n"
+ + "DISTRIBUTED BY HASH(k1) BUCKETS 1\n"
+ +
"PROPERTIES('replication_num'='1','binlog.enable'='true','binlog.format'='ROW');";
+ createTableWithRowBinlog(create);
+
+ Database db =
Env.getCurrentInternalCatalog().getDbOrMetaException("test");
+ OlapTable tbl = (OlapTable) db.getTableOrMetaException(tableName,
Table.TableType.OLAP);
+ Set<Long> dataTabletIds = Sets.newHashSet();
+ Set<Long> rowBinlogTabletIds = Sets.newHashSet();
+ for (Partition partition : tbl.getPartitions()) {
+ for (MaterializedIndex index :
partition.getMaterializedIndices(IndexExtState.VISIBLE, true)) {
+ (index.isRowBinlog() ? rowBinlogTabletIds :
dataTabletIds).addAll(index.getTabletIdsInOrder());
+ }
+ }
+ Assertions.assertFalse(rowBinlogTabletIds.isEmpty());
+
+ // Build the update tasks without sending them, and record the tablets
they are for.
+ Set<Long> updatedTabletIds = Sets.newHashSet();
+ boolean originalRunningUnitTest = FeConstants.runningUnitTest;
+ FeConstants.runningUnitTest = true;
+ try (MockedConstruction<UpdateTabletMetaInfoTask> ignored =
Mockito.mockConstruction(
+ UpdateTabletMetaInfoTask.class, (task, context) -> {
+ for (Pair<Long, Integer> tablet : (Set<Pair<Long,
Integer>>) context.arguments().get(1)) {
+ updatedTabletIds.add(tablet.first);
+ }
+ })) {
+ alterTable("ALTER TABLE test." + tableName + " SET
('compaction_policy' = 'time_series')",
+ connectContext);
+ Assertions.assertEquals(dataTabletIds, updatedTabletIds);
+
+ updatedTabletIds.clear();
+ alterTable("ALTER TABLE test." + tableName + " SET
('time_series_compaction_goal_size_mbytes' = '2048')",
+ connectContext);
+ Assertions.assertEquals(dataTabletIds, updatedTabletIds);
+
+ // Other tablet meta still goes to the row binlog tablets.
+ updatedTabletIds.clear();
+ alterTable("ALTER TABLE test." + tableName + " SET
('disable_auto_compaction' = 'true')",
+ connectContext);
+ Assertions.assertEquals(Sets.union(dataTabletIds,
rowBinlogTabletIds), updatedTabletIds);
+ } finally {
+ FeConstants.runningUnitTest = originalRunningUnitTest;
+ }
+ }
+
@Test
public void testModifyTableLightSchemaChangeReplayNormalizesBfColumns()
throws Exception {
String tableName = "sc_replay_bf_case";
diff --git
a/regression-test/data/row_binlog_p0/test_row_binlog_alter_compaction_policy.out
b/regression-test/data/row_binlog_p0/test_row_binlog_alter_compaction_policy.out
new file mode 100644
index 00000000000..65b699664b2
--- /dev/null
+++
b/regression-test/data/row_binlog_p0/test_row_binlog_alter_compaction_policy.out
@@ -0,0 +1,13 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !alter_policy --
+0 1 10
+0 1 11
+0 2 20
+0 3 30
+
+-- !alter_time_series_setting --
+0 1 10
+0 1 11
+0 2 20
+0 3 30
+
diff --git
a/regression-test/suites/row_binlog_p0/test_row_binlog_alter_compaction_policy.groovy
b/regression-test/suites/row_binlog_p0/test_row_binlog_alter_compaction_policy.groovy
new file mode 100644
index 00000000000..8e174e68d47
--- /dev/null
+++
b/regression-test/suites/row_binlog_p0/test_row_binlog_alter_compaction_policy.groovy
@@ -0,0 +1,124 @@
+// 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.
+
+// Altering a compaction setting of a row binlog table must not reach its
hidden row binlog
+// tablets: they keep the binlog compaction policy they were created with.
+suite("test_row_binlog_alter_compaction_policy") {
+ if (isCloudMode()) {
+ return
+ }
+
+ def backendIdToIp = [:]
+ def backendIdToHttpPort = [:]
+ getBackendIpHttpPort(backendIdToIp, backendIdToHttpPort)
+
+ def getTabletMeta = { tablet ->
+ def meta = null
+ httpTest {
+ endpoint
"${backendIdToIp["${tablet.BackendId}"]}:${backendIdToHttpPort["${tablet.BackendId}"]}"
+ uri "/api/meta/header/${tablet.TabletId}"
+ op "get"
+ check { respCode, body ->
+ assertEquals(200, respCode as int)
+ meta = parseJson(body.toString())
+ }
+ }
+ return meta
+ }
+
+ // A manual compaction makes a tablet switch to the policy in its tablet
meta, so after it
+ // the tablet status shows the policy the tablet really compacts with.
+ def checkCompactionPolicy = { String tableName, String dataTabletPolicy ->
+ trigger_and_wait_compaction(tableName, "cumulative")
+ def rowBinlogTabletNum = 0
+ for (def tablet : sql_return_maparray("SHOW TABLETS FROM
${tableName}")) {
+ def expectedPolicy = dataTabletPolicy
+ if (tablet.IsRowBinlog == "true") {
+ expectedPolicy = "binlog"
+ rowBinlogTabletNum++
+ }
+ assertEquals(expectedPolicy,
getTabletMeta(tablet).compaction_policy)
+ def (code, out, err) =
be_show_tablet_status(backendIdToIp["${tablet.BackendId}"],
+ backendIdToHttpPort["${tablet.BackendId}"],
tablet.TabletId)
+ assertEquals(0, code)
+ assertEquals(expectedPolicy, parseJson(out.trim())["cumulative
policy type"])
+ }
+ assertTrue(rowBinlogTabletNum > 0)
+ }
+
+ // Case 1: change the compaction policy.
+ sql "DROP TABLE IF EXISTS test_row_binlog_alter_compaction_policy_dup
FORCE"
+ sql """
+ CREATE TABLE test_row_binlog_alter_compaction_policy_dup (
+ k1 INT,
+ v1 INT
+ )
+ DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW"
+ )
+ """
+ sql "INSERT INTO test_row_binlog_alter_compaction_policy_dup VALUES (1,
10), (2, 20)"
+ sql "INSERT INTO test_row_binlog_alter_compaction_policy_dup VALUES (1,
11), (3, 30)"
+
+ sql """ALTER TABLE test_row_binlog_alter_compaction_policy_dup SET
("compaction_policy" = "time_series")"""
+ checkCompactionPolicy("test_row_binlog_alter_compaction_policy_dup",
"time_series")
+ order_qt_alter_policy """
+ SELECT __DORIS_BINLOG_OP__ AS op, k1, v1
+ FROM binlog("table" = "test_row_binlog_alter_compaction_policy_dup")
+ """
+
+ // Case 2: change a time series setting. The row binlog tablets are not on
the time_series
+ // policy, so the setting must not be sent to them.
+ sql "DROP TABLE IF EXISTS test_row_binlog_alter_compaction_policy_ts FORCE"
+ sql """
+ CREATE TABLE test_row_binlog_alter_compaction_policy_ts (
+ k1 INT,
+ v1 INT
+ )
+ DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true",
+ "compaction_policy" = "time_series",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW"
+ )
+ """
+ sql "INSERT INTO test_row_binlog_alter_compaction_policy_ts VALUES (1,
10), (2, 20)"
+ sql "INSERT INTO test_row_binlog_alter_compaction_policy_ts VALUES (1,
11), (3, 30)"
+
+ sql """
+ ALTER TABLE test_row_binlog_alter_compaction_policy_ts
+ SET ("time_series_compaction_goal_size_mbytes" = "2048")
+ """
+ for (def tablet : sql_return_maparray("SHOW TABLETS FROM
test_row_binlog_alter_compaction_policy_ts")) {
+ if (tablet.IsRowBinlog != "true") {
+ assertEquals(2048L,
getTabletMeta(tablet).time_series_compaction_goal_size_mbytes as Long)
+ }
+ }
+ checkCompactionPolicy("test_row_binlog_alter_compaction_policy_ts",
"time_series")
+ order_qt_alter_time_series_setting """
+ SELECT __DORIS_BINLOG_OP__ AS op, k1, v1
+ FROM binlog("table" = "test_row_binlog_alter_compaction_policy_ts")
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
