This is an automated email from the ASF dual-hosted git repository.
swuferhong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 1dfa13fa7 [server] Add table-level config for standby replica with
ALTER TABLE support (#3257)
1dfa13fa7 is described below
commit 1dfa13fa78f75413e605caf713bbb4379a7ffccd
Author: yunhong <[email protected]>
AuthorDate: Wed Jun 3 18:14:04 2026 +0800
[server] Add table-level config for standby replica with ALTER TABLE
support (#3257)
---
.../fluss/client/admin/FlussAdminITCase.java | 3 +
.../fluss/client/table/FlussTableITCase.java | 1 +
.../org/apache/fluss/config/ConfigOptions.java | 11 ++
.../org/apache/fluss/config/FlussConfigUtils.java | 3 +-
.../java/org/apache/fluss/config/TableConfig.java | 8 +
.../org/apache/fluss/metadata/TableDescriptor.java | 12 ++
.../fluss/flink/catalog/FlinkCatalogITCase.java | 2 +
.../fluss/flink/utils/CatalogTableTestUtils.java | 3 +
.../coordinator/CoordinatorEventProcessor.java | 26 ++-
.../server/coordinator/CoordinatorService.java | 8 +-
.../statemachine/ReplicaLeaderElection.java | 22 +--
.../statemachine/TableBucketStateMachine.java | 28 +--
.../server/utils/TableDescriptorValidation.java | 9 +
.../coordinator/CoordinatorEventProcessorTest.java | 187 +++++++++++++++++++++
.../server/coordinator/TableManagerITCase.java | 3 +-
website/docs/engine-flink/options.md | 5 +-
16 files changed, 303 insertions(+), 28 deletions(-)
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
index 162f3c158..65cca8195 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
@@ -295,6 +295,7 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
options.put(
ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ options.put(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(),
"true");
assertThat(tableInfo.toTableDescriptor())
.isEqualTo(tableDescriptor.withProperties(options));
assertThat(schemaInfo2).isEqualTo(schemaInfo);
@@ -324,6 +325,7 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
options.put(
ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ options.put(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(),
"true");
assertThat(tableInfo.toTableDescriptor()).isEqualTo(expected.withProperties(options));
assertThat(schemaInfo2).isEqualTo(schemaInfo);
// assert created time
@@ -908,6 +910,7 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
options.put(
ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ options.put(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(),
"true");
assertThat(tableInfo.toTableDescriptor()).isEqualTo(expected.withProperties(options));
}
}
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java
b/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java
index 1d3865a08..03bcb8d96 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java
@@ -109,6 +109,7 @@ class FlussTableITCase extends ClientToServerITCaseBase {
.withDataLakeFormat(DataLakeFormat.PAIMON);
Map<String, String> options = new HashMap<>(expected.getProperties());
options.put(ConfigOptions.TABLE_KV_FORMAT_VERSION.key(), "2");
+ options.put(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(),
"true");
expected = expected.withProperties(options);
assertThat(tableInfo.toTableDescriptor()).isEqualTo(expected);
}
diff --git
a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index d0ede21a7..ec819f575 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -1503,6 +1503,17 @@ public class ConfigOptions {
+ "for optimization (encoded bytes can be
reused for bucket calculation). "
+ "Bucket key encoding always uses
datalake's encoder to align with datalake bucket calculation.");
+ public static final ConfigOption<Boolean> TABLE_KV_STANDBY_REPLICA_ENABLED
=
+ key("table.kv.standby-replica.enabled")
+ .booleanType()
+ .noDefaultValue()
+ .withDescription(
+ "Whether to enable standby replicas for primary
key tables. "
+ + "Standby replicas maintain recent KV
snapshots for fast leader promotion. "
+ + "Automatically set to true by the
coordinator during table creation for new PK tables. "
+ + "Tables created before this option was
introduced are treated as disabled. "
+ + "Can be dynamically enabled via ALTER
TABLE.");
+
public static final ConfigOption<Boolean> TABLE_AUTO_PARTITION_ENABLED =
key("table.auto-partition.enabled")
.booleanType()
diff --git
a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
index 7daa48c2b..17cfcb407 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
@@ -50,7 +50,8 @@ public class FlussConfigUtils {
ConfigOptions.TABLE_DATALAKE_FRESHNESS.key(),
ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS.key(),
ConfigOptions.TABLE_AUTO_PARTITION_NUM_RETENTION.key(),
- ConfigOptions.TABLE_STATISTICS_COLUMNS.key());
+ ConfigOptions.TABLE_STATISTICS_COLUMNS.key(),
+ ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key());
}
public static boolean isTableStorageConfig(String key) {
diff --git
a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java
b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java
index 13cb49e9e..fbf8c7726 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java
@@ -77,6 +77,14 @@ public class TableConfig {
return config.getOptional(ConfigOptions.TABLE_KV_FORMAT_VERSION);
}
+ /**
+ * Whether standby replicas are enabled for this primary key table.
Returns false for legacy
+ * tables that were created before this option was introduced.
+ */
+ public boolean isStandbyReplicaEnabled() {
+ return
config.getOptional(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED).orElse(false);
+ }
+
/** Gets the log TTL of the table. */
public long getLogTTLMs() {
return config.get(ConfigOptions.TABLE_LOG_TTL).toMillis();
diff --git
a/fluss-common/src/main/java/org/apache/fluss/metadata/TableDescriptor.java
b/fluss-common/src/main/java/org/apache/fluss/metadata/TableDescriptor.java
index 9f18dd401..fc19f6b7e 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metadata/TableDescriptor.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metadata/TableDescriptor.java
@@ -281,6 +281,18 @@ public final class TableDescriptor implements Serializable
{
return withProperties(newProperties);
}
+ /**
+ * Returns a new TableDescriptor instance that is a copy of this
TableDescriptor with a new
+ * standby replica enabled.
+ */
+ public TableDescriptor withStandbyReplicaEnabled(boolean
standbyReplicaEnabled) {
+ Map<String, String> newProperties = new HashMap<>(properties);
+ newProperties.put(
+ ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(),
+ String.valueOf(standbyReplicaEnabled));
+ return withProperties(newProperties);
+ }
+
/**
* Returns a new TableDescriptor instance that is a copy of this
TableDescriptor with a new
* bucket count.
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
index eebe58611..8991618fd 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
@@ -686,6 +686,7 @@ abstract class FlinkCatalogITCase {
expectedTableProperties.put("table.replication.factor", "1");
expectedTableProperties.put(
"table.kv.format-version",
String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ expectedTableProperties.put("table.kv.standby-replica.enabled",
"true");
assertThat(tableInfo.getProperties().toMap()).isEqualTo(expectedTableProperties);
Map<String, String> expectedCustomProperties = new HashMap<>();
@@ -1076,6 +1077,7 @@ abstract class FlinkCatalogITCase {
actualOptions.remove(ConfigOptions.BOOTSTRAP_SERVERS.key());
actualOptions.remove(ConfigOptions.TABLE_REPLICATION_FACTOR.key());
actualOptions.remove(ConfigOptions.TABLE_KV_FORMAT_VERSION.key());
+
actualOptions.remove(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key());
assertThat(actualOptions).isEqualTo(expectedOptions);
}
}
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/CatalogTableTestUtils.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/CatalogTableTestUtils.java
index 2c255a40e..975b515eb 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/CatalogTableTestUtils.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/CatalogTableTestUtils.java
@@ -27,6 +27,7 @@ import java.util.Map;
import static org.apache.fluss.config.ConfigOptions.BOOTSTRAP_SERVERS;
import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT;
import static org.apache.fluss.config.ConfigOptions.TABLE_KV_FORMAT_VERSION;
+import static
org.apache.fluss.config.ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED;
import static org.apache.fluss.config.ConfigOptions.TABLE_REPLICATION_FACTOR;
import static org.assertj.core.api.Assertions.assertThat;
@@ -142,6 +143,8 @@ public class CatalogTableTestUtils {
// Remove datalake format (auto-added when datalake is enabled in
Fluss cluster)
actualOptions.remove(TABLE_DATALAKE_FORMAT.key());
actualOptions.remove(TABLE_KV_FORMAT_VERSION.key());
+ // Remove standby replica (auto-added in Fluss cluster)
+ actualOptions.remove(TABLE_KV_STANDBY_REPLICA_ENABLED.key());
assertThat(actualOptions).isEqualTo(expectedOptions);
}
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index 96f39ef08..00ca807f0 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -814,7 +814,31 @@ public class CoordinatorEventProcessor implements
EventProcessor {
autoPartitionManager.updateAutoPartitionTables(newTableInfo);
}
- // more post-alter actions can be added here
+ // If standby replica config changed, trigger re-election for all
online buckets
+ // of this table to apply new standby replica assignment
+ boolean oldStandbyEnabled =
oldTableInfo.getTableConfig().isStandbyReplicaEnabled();
+ boolean newStandbyEnabled =
newTableInfo.getTableConfig().isStandbyReplicaEnabled();
+ if (oldStandbyEnabled != newStandbyEnabled) {
+ triggerReElectionForTable(newTableInfo.getTableId());
+ }
+ }
+
+ private void triggerReElectionForTable(long tableId) {
+ Set<TableBucket> onlineBuckets =
+ coordinatorContext.getAllBuckets().stream()
+ .filter(
+ tb ->
+ tb.getTableId() == tableId
+ &&
coordinatorContext.getBucketState(tb)
+ == OnlineBucket)
+ .collect(Collectors.toSet());
+ if (!onlineBuckets.isEmpty()) {
+ LOG.info(
+ "Triggering re-election for {} buckets of table {} due to
standby replica config change.",
+ onlineBuckets.size(),
+ tableId);
+ tableBucketStateMachine.handleStateChange(onlineBuckets,
OnlineBucket);
+ }
}
private void processCreatePartition(CreatePartitionEvent
createPartitionEvent) {
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
index e96e36f45..5a37aa9a6 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
@@ -655,7 +655,6 @@ public final class CoordinatorService extends
RpcServiceBase implements Coordina
newProperties.put(
ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
String.valueOf(CURRENT_KV_FORMAT_VERSION));
- newDescriptor = newDescriptor.withProperties(newProperties);
} else {
if (formatVersion > CURRENT_KV_FORMAT_VERSION) {
throw new InvalidConfigException(
@@ -665,6 +664,13 @@ public final class CoordinatorService extends
RpcServiceBase implements Coordina
formatVersion, CURRENT_KV_FORMAT_VERSION));
}
}
+
+ // Enable standby replica for new PK tables if not explicitly
configured
+ if
(!newProperties.containsKey(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key()))
{
+
newProperties.put(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "true");
+ }
+
+ newDescriptor = newDescriptor.withProperties(newProperties);
}
return newDescriptor;
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElection.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElection.java
index d63714423..d2e01eb8c 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElection.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElection.java
@@ -42,14 +42,14 @@ public abstract class ReplicaLeaderElection {
* @param assignments the assignments
* @param aliveReplicas the alive replicas
* @param leaderAndIsr the original leaderAndIsr
- * @param isPrimaryKeyTable whether this table bucket is primary key
table
+ * @param standbyReplicaEnabled whether standby replica is enabled for
this table bucket
* @return the election result
*/
public Optional<ElectionResult> leaderElection(
List<Integer> assignments,
List<Integer> aliveReplicas,
LeaderAndIsr leaderAndIsr,
- boolean isPrimaryKeyTable) {
+ boolean standbyReplicaEnabled) {
List<Integer> isr = leaderAndIsr.isr();
// First we will filter out the assignment list to only contain
the alive replicas and
// isr.
@@ -61,7 +61,7 @@ public abstract class ReplicaLeaderElection {
&& isr.contains(replica))
.collect(Collectors.toList());
return electLeader(
- availableReplicas, aliveReplicas, isr, leaderAndIsr,
isPrimaryKeyTable);
+ availableReplicas, aliveReplicas, isr, leaderAndIsr,
standbyReplicaEnabled);
}
}
@@ -74,7 +74,7 @@ public abstract class ReplicaLeaderElection {
* @param aliveReplicas the alive replicas
* @param leaderAndIsr the original leaderAndIsr
* @param shutdownTabletServers the shutdown tabletServers
- * @param isPrimaryKeyTable whether this table bucket is primary key
table
+ * @param standbyReplicaEnabled whether standby replica is enabled for
this table bucket
* @return the election result
*/
public Optional<ElectionResult> leaderElection(
@@ -82,7 +82,7 @@ public abstract class ReplicaLeaderElection {
List<Integer> aliveReplicas,
LeaderAndIsr leaderAndIsr,
Set<Integer> shutdownTabletServers,
- boolean isPrimaryKeyTable) {
+ boolean standbyReplicaEnabled) {
List<Integer> originIsr = leaderAndIsr.isr();
Set<Integer> isrSet = new HashSet<>(originIsr);
// Filter out available replicas: alive, in ISR, and not shutting
down.
@@ -108,7 +108,7 @@ public abstract class ReplicaLeaderElection {
new ArrayList<>(newAliveReplicaSet),
newIsr,
leaderAndIsr,
- isPrimaryKeyTable);
+ standbyReplicaEnabled);
}
}
@@ -121,7 +121,9 @@ public abstract class ReplicaLeaderElection {
}
public Optional<ElectionResult> leaderElection(
- List<Integer> liveReplicas, LeaderAndIsr leaderAndIsr, boolean
isPrimaryKeyTable) {
+ List<Integer> liveReplicas,
+ LeaderAndIsr leaderAndIsr,
+ boolean standbyReplicaEnabled) {
// For bucket reassignment, the first replica in newReplicas is
the target leader
// encoded by GoalOptimizerUtils. We must always honor this target
instead of using
// standby-promotion logic, otherwise the elected leader may
differ from the rebalance
@@ -141,7 +143,7 @@ public abstract class ReplicaLeaderElection {
// Always use the first available replica as leader to honor the
rebalance plan.
int newLeader = availableReplicas.get(0);
List<Integer> standbyReplicas =
- isPrimaryKeyTable
+ standbyReplicaEnabled
? findNewStandby(availableReplicas, newLeader)
: Collections.emptyList();
@@ -161,13 +163,13 @@ public abstract class ReplicaLeaderElection {
List<Integer> liveReplicas,
List<Integer> newIsr,
LeaderAndIsr leaderAndIsr,
- boolean isPrimaryKeyTable) {
+ boolean standbyReplicaEnabled) {
if (availableReplicas.isEmpty()) {
return Optional.empty();
}
// For log table, simply use the first available replica as leader.
- if (!isPrimaryKeyTable) {
+ if (!standbyReplicaEnabled) {
return Optional.of(
new ElectionResult(
liveReplicas,
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java
index 6a556ddea..2677b85c3 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java
@@ -20,6 +20,7 @@ package org.apache.fluss.server.coordinator.statemachine;
import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.server.coordinator.CoordinatorContext;
import org.apache.fluss.server.coordinator.CoordinatorRequestBatch;
import
org.apache.fluss.server.coordinator.statemachine.ReplicaLeaderElection.ControlledShutdownLeaderElection;
@@ -438,14 +439,15 @@ public class TableBucketStateMachine {
}
// For the case that the table bucket has been initialized, we use all
the live assigned
// servers as inSyncReplica set.
+ TableInfo tableInfo =
coordinatorContext.getTableInfoById(tableBucket.getTableId());
+ boolean standbyEnabled =
+ tableInfo.hasPrimaryKey() &&
tableInfo.getTableConfig().isStandbyReplicaEnabled();
Optional<ElectionResult> resultOpt =
initReplicaLeaderElection(
assignedServers,
liveServers,
coordinatorContext.getCoordinatorEpoch(),
- coordinatorContext
- .getTableInfoById(tableBucket.getTableId())
- .hasPrimaryKey());
+ standbyEnabled);
if (!resultOpt.isPresent()) {
LOG.error(
"The leader election for table bucket {} is empty.",
@@ -625,12 +627,14 @@ public class TableBucketStateMachine {
}
Optional<ElectionResult> resultOpt = Optional.empty();
- boolean isPkTable =
-
coordinatorContext.getTableInfoById(tableBucket.getTableId()).hasPrimaryKey();
+ TableInfo tableInfo =
coordinatorContext.getTableInfoById(tableBucket.getTableId());
+ boolean standbyReplicaEnabled =
+ tableInfo.hasPrimaryKey() &&
tableInfo.getTableConfig().isStandbyReplicaEnabled();
if (electionStrategy instanceof DefaultLeaderElection) {
resultOpt =
((DefaultLeaderElection) electionStrategy)
- .leaderElection(assignment, liveReplicas,
leaderAndIsr, isPkTable);
+ .leaderElection(
+ assignment, liveReplicas, leaderAndIsr,
standbyReplicaEnabled);
} else if (electionStrategy instanceof
ControlledShutdownLeaderElection) {
Set<Integer> shuttingDownTabletServers =
coordinatorContext.shuttingDownTabletServers();
resultOpt =
@@ -640,11 +644,11 @@ public class TableBucketStateMachine {
liveReplicas,
leaderAndIsr,
shuttingDownTabletServers,
- isPkTable);
+ standbyReplicaEnabled);
} else if (electionStrategy instanceof ReassignmentLeaderElection) {
resultOpt =
((ReassignmentLeaderElection) electionStrategy)
- .leaderElection(liveReplicas, leaderAndIsr,
isPkTable);
+ .leaderElection(liveReplicas, leaderAndIsr,
standbyReplicaEnabled);
}
if (!resultOpt.isPresent()) {
@@ -685,7 +689,7 @@ public class TableBucketStateMachine {
* @param assignments the assignments
* @param aliveReplicas the alive replicas
* @param coordinatorEpoch the coordinator epoch
- * @param isPrimaryKeyTable whether this table bucket is primary key table
+ * @param standbyReplicaEnabled whether standby replica is enabled for
this table bucket
* @return the election result
*/
@VisibleForTesting
@@ -693,7 +697,7 @@ public class TableBucketStateMachine {
List<Integer> assignments,
List<Integer> aliveReplicas,
int coordinatorEpoch,
- boolean isPrimaryKeyTable) {
+ boolean standbyReplicaEnabled) {
// First we will filter out the assignment list to only contain the
alive replicas.
List<Integer> availableReplicas =
assignments.stream().filter(aliveReplicas::contains).collect(Collectors.toList());
@@ -706,10 +710,10 @@ public class TableBucketStateMachine {
// Then we will use the first replica in assignment as the leader
replica.
int leader = availableReplicas.get(0);
- // If this table is primaryKey table, we will use the second replica
in assignment as the
+ // If standby replica is enabled, we will use the second replica in
assignment as the
// standby if exists.
List<Integer> standbyReplicas = new ArrayList<>();
- if (isPrimaryKeyTable) {
+ if (standbyReplicaEnabled) {
if (availableReplicas.size() > 1) {
standbyReplicas.add(availableReplicas.get(1));
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
index fcd5f1688..949afd8c6 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
@@ -170,6 +170,15 @@ public class TableDescriptorValidation {
.collect(Collectors.joining(", "))));
}
+ // Standby replica is only applicable to primary key tables
+ if
(tableKeysToChange.contains(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key())
+ && !currentTable.hasPrimaryKey()) {
+ throw new InvalidAlterTableException(
+ String.format(
+ "'%s' can only be altered on primary key tables.",
+
ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key()));
+ }
+
if (!currentConfig.getDataLakeFormat().isPresent()) {
List<String> datalakeKeys =
tableKeysToChange.stream()
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
index e1226156e..d6813abb0 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
@@ -24,6 +24,7 @@ import org.apache.fluss.cluster.rebalance.RebalanceStatus;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.FencedLeaderEpochException;
+import org.apache.fluss.exception.InvalidAlterTableException;
import org.apache.fluss.exception.InvalidCoordinatorException;
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.metadata.DatabaseDescriptor;
@@ -144,6 +145,7 @@ class CoordinatorEventProcessorTest {
.primaryKey("a")
.build())
.distributedBy(3, "a")
+
.property(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "true")
.build()
.withReplicationFactor(REPLICATION_FACTOR);
@@ -1009,6 +1011,191 @@ class CoordinatorEventProcessorTest {
});
}
+ @Test
+ void testAlterStandbyReplicaEnabled() throws Exception {
+ // make sure all request to gateway should be successful
+ initCoordinatorChannel();
+
+ // create a PK table with standby replica enabled (TEST_TABLE has it
enabled)
+ TablePath t1 = TablePath.of(defaultDatabase,
"test_alter_standby_replica");
+ int nBuckets = 1;
+ int replicationFactor = 3;
+ TableAssignment tableAssignment =
+ generateAssignment(
+ nBuckets,
+ replicationFactor,
+ new TabletServerInfo[] {
+ new TabletServerInfo(0, "rack0"),
+ new TabletServerInfo(1, "rack1"),
+ new TabletServerInfo(2, "rack2")
+ });
+ long tableId =
+ metadataManager.createTable(t1, remoteDataDir, TEST_TABLE,
tableAssignment, false);
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+
+ // wait for the bucket to become online with standby replica assigned
+ retryVerifyContext(
+ ctx -> {
+
assertThat(ctx.getBucketState(tableBucket)).isEqualTo(OnlineBucket);
+ LeaderAndIsr leaderAndIsr =
ctx.getBucketLeaderAndIsr(tableBucket).get();
+ // Standby should be assigned since the table is PK with
standby enabled
+ assertThat(leaderAndIsr.standbyReplicas()).hasSize(1);
+ });
+
+ // ALTER: disable standby replica
+ TablePropertyChanges.Builder disableBuilder =
TablePropertyChanges.builder();
+ disableBuilder.setTableProperty(
+ ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "false");
+ metadataManager.alterTableProperties(
+ t1, Collections.emptyList(), disableBuilder.build(), false,
null);
+
+ // verify standby replicas are removed after re-election
+ retryVerifyContext(
+ ctx -> {
+ TableInfo tableInfoInCtx = ctx.getTableInfoById(tableId);
+
assertThat(tableInfoInCtx.getTableConfig().isStandbyReplicaEnabled()).isFalse();
+ LeaderAndIsr leaderAndIsr =
ctx.getBucketLeaderAndIsr(tableBucket).get();
+ assertThat(leaderAndIsr.standbyReplicas()).isEmpty();
+ });
+
+ // Also verify from ZooKeeper that standby replicas are cleared
+ LeaderAndIsr zkLeaderAndIsr =
zookeeperClient.getLeaderAndIsr(tableBucket).get();
+ assertThat(zkLeaderAndIsr.standbyReplicas()).isEmpty();
+ }
+
+ @Test
+ void testAlterEnableStandbyReplicaForExistingTable() throws Exception {
+ // Simulate a legacy PK table created without standby replica config.
+ // After ALTER to enable standby replica, re-election should be
triggered
+ // and standby replicas should be assigned.
+ initCoordinatorChannel();
+
+ // Create PK table WITHOUT standby replica enabled (simulating legacy
table)
+ TablePath t1 = TablePath.of(defaultDatabase,
"test_enable_standby_existing");
+ TableDescriptor pkTableNoStandby =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .primaryKey("a")
+ .build())
+ .distributedBy(1, "a")
+ .build()
+ .withReplicationFactor(3);
+
+ int nBuckets = 1;
+ int replicationFactor = 3;
+ TableAssignment tableAssignment =
+ generateAssignment(
+ nBuckets,
+ replicationFactor,
+ new TabletServerInfo[] {
+ new TabletServerInfo(0, "rack0"),
+ new TabletServerInfo(1, "rack1"),
+ new TabletServerInfo(2, "rack2")
+ });
+ long tableId =
+ metadataManager.createTable(
+ t1, remoteDataDir, pkTableNoStandby, tableAssignment,
false);
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+
+ // Wait for the bucket to become online with NO standby (legacy
behavior)
+ retryVerifyContext(
+ ctx -> {
+
assertThat(ctx.getBucketState(tableBucket)).isEqualTo(OnlineBucket);
+ LeaderAndIsr leaderAndIsr =
ctx.getBucketLeaderAndIsr(tableBucket).get();
+ assertThat(leaderAndIsr.standbyReplicas()).isEmpty();
+ });
+
+ // Record the leader epoch before ALTER
+ int leaderEpochBefore =
zookeeperClient.getLeaderAndIsr(tableBucket).get().leaderEpoch();
+
+ // ALTER: enable standby replica
+ TablePropertyChanges.Builder enableBuilder =
TablePropertyChanges.builder();
+ enableBuilder.setTableProperty(
+ ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "true");
+ metadataManager.alterTableProperties(
+ t1, Collections.emptyList(), enableBuilder.build(), false,
null);
+
+ // Verify re-election happened: standby assigned and leaderEpoch
incremented
+ retryVerifyContext(
+ ctx -> {
+ TableInfo tableInfoInCtx = ctx.getTableInfoById(tableId);
+
assertThat(tableInfoInCtx.getTableConfig().isStandbyReplicaEnabled()).isTrue();
+ LeaderAndIsr leaderAndIsr =
ctx.getBucketLeaderAndIsr(tableBucket).get();
+ // Standby should now be assigned
+ assertThat(leaderAndIsr.standbyReplicas()).hasSize(1);
+ // Leader epoch should have incremented (proves
re-election)
+
assertThat(leaderAndIsr.leaderEpoch()).isGreaterThan(leaderEpochBefore);
+ });
+
+ // Also verify from ZooKeeper
+ LeaderAndIsr zkLeaderAndIsr =
zookeeperClient.getLeaderAndIsr(tableBucket).get();
+ assertThat(zkLeaderAndIsr.standbyReplicas()).hasSize(1);
+
assertThat(zkLeaderAndIsr.leaderEpoch()).isGreaterThan(leaderEpochBefore);
+ }
+
+ @Test
+ void testAlterStandbyReplicaEnabledForLogTable() throws Exception {
+ // Altering standby replica config on a log table (no PK) should throw
an error
+ initCoordinatorChannel();
+
+ TablePath t1 = TablePath.of(defaultDatabase,
"test_alter_standby_log_table");
+ TableDescriptor logTable =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.STRING())
+ .build())
+ .distributedBy(1)
+ .build()
+ .withReplicationFactor(3);
+
+ int nBuckets = 1;
+ int replicationFactor = 3;
+ TableAssignment tableAssignment =
+ generateAssignment(
+ nBuckets,
+ replicationFactor,
+ new TabletServerInfo[] {
+ new TabletServerInfo(0, "rack0"),
+ new TabletServerInfo(1, "rack1"),
+ new TabletServerInfo(2, "rack2")
+ });
+ metadataManager.createTable(t1, remoteDataDir, logTable,
tableAssignment, false);
+
+ // ALTER: try to enable standby replica on a log table should fail
+ TablePropertyChanges.Builder enableBuilder =
TablePropertyChanges.builder();
+ enableBuilder.setTableProperty(
+ ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "true");
+ assertThatThrownBy(
+ () ->
+ metadataManager.alterTableProperties(
+ t1,
+ Collections.emptyList(),
+ enableBuilder.build(),
+ false,
+ null))
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining("can only be altered on primary key
tables");
+
+ // ALTER: try to set to false on a log table should also fail
+ TablePropertyChanges.Builder disableBuilder =
TablePropertyChanges.builder();
+ disableBuilder.setTableProperty(
+ ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "false");
+ assertThatThrownBy(
+ () ->
+ metadataManager.alterTableProperties(
+ t1,
+ Collections.emptyList(),
+ disableBuilder.build(),
+ false,
+ null))
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining("can only be altered on primary key
tables");
+ }
+
@Test
void testDoBucketReassignment() throws Exception {
zookeeperClient.registerTabletServer(
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
index b2abac830..0428d4e10 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
@@ -497,7 +497,8 @@ class TableManagerITCase {
tableDescriptor = tableDescriptor.withProperties(properties);
assertThat(TableDescriptor.fromJsonBytes(tableMetadata.getTableJson()))
- .isEqualTo(tableDescriptor.withReplicationFactor(1));
+ .isEqualTo(
+
tableDescriptor.withReplicationFactor(1).withStandbyReplicaEnabled(true));
// now, check the table buckets metadata
assertThat(tableMetadata.getBucketMetadatasCount()).isEqualTo(expectBucketCount);
diff --git a/website/docs/engine-flink/options.md
b/website/docs/engine-flink/options.md
index af9f97a69..0b3d675dc 100644
--- a/website/docs/engine-flink/options.md
+++ b/website/docs/engine-flink/options.md
@@ -74,12 +74,13 @@ See more details about [ALTER TABLE ...
SET](engine-flink/ddl.md#set-properties)
| table.auto-partition.num-retention | Integer | 7
| The number of history partitions to retain for auto created
partitions in each check for auto partition. For example, if the current check
time is 2024-11-11, time-unit is DAY, and the value is configured as 3, then
the history partitions 20241108, 20241109, 20241110 will be retained. The
partitions earlier than 20241108 will be deleted. The default value is 7, which
means that 7 partitions will [...]
| table.auto-partition.time-zone | String | the system time zone
| The time zone for auto partitions, which is by default the same
as the system time zone.
[...]
| table.replication.factor | Integer | (None)
| The replication factor for the log of the new table. When it's
not set, Fluss will use the cluster's default replication factor configured by
default.replication.factor. It should be a positive number and not larger than
the number of tablet servers in the Fluss cluster. A value larger than the
number of tablet servers in Fluss cluster will result in an error when the new
table is created. [...]
-| table.statistics.columns | String | (None)
| Specifies which columns to collect statistics (min, max, null
count) for in log table batches. Use `*` to collect statistics for all
supported columns, or specify a comma-separated list of column names (e.g.,
`col1,col2`). Supported types: BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT,
FLOAT, DOUBLE, STRING, CHAR, DECIMAL, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ.
Unsupported types (BYTES, BINARY, ARRA [...]
+| table.statistics.columns | String | (None)
| Specifies which columns to collect statistics (min, max, null
count) for in log table batches. Use `*` to collect statistics for all
supported columns, or specify a comma-separated list of column names (e.g.,
`col1,col2`). Supported types: BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT,
FLOAT, DOUBLE, STRING, CHAR, DECIMAL, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ.
Unsupported types (BYTES, BINARY, ARRA [...]
| table.log.format | Enum | ARROW
| The format of the log records in log store. The default value is
`ARROW`. The supported formats are `ARROW`, `INDEXED` and `COMPACTED`.
[...]
| table.log.arrow.compression.type | Enum | ZSTD
| The compression type of the log records if the log format is set
to `ARROW`. The candidate compression type is `NONE`, `LZ4_FRAME`, `ZSTD`. The
default value is `ZSTD`.
[...]
| table.log.arrow.compression.zstd.level | Integer | 3
| The compression level of the log records if the log format is set
to `ARROW` and the compression type is set to `ZSTD`. The valid range is 1 to
22. The default value is 3.
[...]
| table.kv.format | Enum | COMPACTED
| The format of the kv records in kv store. The default value is
`COMPACTED`. The supported formats are `COMPACTED` and `INDEXED`.
[...]
| table.kv.format-version | Integer | (None)
| The version of the kv format. Automatically set by the
coordinator during table creation if not configured by users.<br></br>**Note:**
The datalake encoding and bucketing strategy mentioned below only takes effect
when `datalake.format` is configured at cluster level.<br></br>**Version
Behaviors:**<br></br>(1) **Version 1**: Tables created before
`table.kv.format-version` was introduced are treat [...]
+| table.kv.standby-replica.enabled | Boolean | (None)
| Whether to enable standby replicas for primary key tables.
Standby replicas maintain recent KV snapshots for fast leader promotion.
Automatically set to `true` by the coordinator during table creation for new PK
tables. Tables created before this option was introduced are treated as
disabled. Can be dynamically enabled via `ALTER TABLE SET
('table.kv.standby-replica.enabled' = 'true')`. [...]
| table.log.tiered.local-segments | Integer | 2
| The number of log segments to retain in local for each table when
log tiered storage is enabled. It must be greater that 0. The default is 2.
[...]
| table.datalake.enabled | Boolean | false
| Whether enable lakehouse storage for the table. Disabled by
default. When this option is set to ture and the datalake tiering service is
up, the table will be tiered and compacted into datalake format stored on
lakehouse storage.
[...]
| table.datalake.format | Enum | (None)
| The data lake format of the table specifies the tiered Lakehouse
storage format. Currently, supported formats are `paimon`, `iceberg`, and
`lance`. In the future, more kinds of data lake format will be supported, such
as DeltaLake or Hudi. Once the `table.datalake.format` property is configured,
Fluss adopts the key encoding and bucketing strategy used by the corresponding
data lake format. This [...]
@@ -91,7 +92,7 @@ See more details about [ALTER TABLE ...
SET](engine-flink/ddl.md#set-properties)
| table.delete.behavior | Enum | ALLOW
| Controls the behavior of delete operations on primary key tables.
Three modes are supported: `ALLOW` (default for default merge engine) - allows
normal delete operations; `IGNORE` - silently ignores delete requests without
errors; `DISABLE` - rejects delete requests and throws explicit errors. This
configuration provides system-level guarantees for some downstream pipelines
(e.g., Flink Delta Joi [...]
| table.changelog.image | Enum | FULL
| Defines the changelog image mode for primary key tables. This
configuration is inspired by similar settings in database systems like MySQL's
`binlog_row_image` and PostgreSQL's `replica identity`. Two modes are
supported: `FULL` (default) - produces both UPDATE_BEFORE and UPDATE_AFTER
records for update operations, capturing complete information about updates and
allowing tracking of previous val [...]
| table.auto-inc.batch-size | Long | 100000L
| The batch size of auto-increment IDs fetched from the distributed
counter each time. This value determines the length of the locally cached ID
segment. Default: 100000. A larger batch size may cause significant
auto-increment ID gaps, especially when unused cached ID segments are discarded
due to TabletServer restarts or abnormal terminations. Conversely, a smaller
batch size increases the freque [...]
-| table.statistics.columns | String | (None)
| Configures column-level statistics collection for the table. By
default this option is not set and no column statistics are collected. The
value `*` means collect statistics for all supported columns. A comma-separated
list of column names means collect statistics only for the specified columns
(recommended for minimal overhead). Supported types: BOOLEAN, TINYINT,
SMALLINT, INTEGER, BIGINT, FLOAT [...]
+| table.statistics.columns | String | (None)
| Configures column-level statistics collection for the table. By
default this option is not set and no column statistics are collected. The
value `*` means collect statistics for all supported columns. A comma-separated
list of column names means collect statistics only for the specified columns
(recommended for minimal overhead). Supported types: BOOLEAN, TINYINT,
SMALLINT, INTEGER, BIGINT, FLOAT [...]
## Read Options
