[fix](row binlog): reject flexible update on row binlog table

fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java

@@ -3216,6 +3216,7 @@ public boolean getEnableUniqueKeySkipBitmap() {
      * 2. Must have skip_bitmap column
      * 3. Must have light_schema_change enabled
      * 4. Cannot have variant columns
+     * 5. Cannot enable row binlog
      * @throws UserException if any constraint is not satisfied
      */
     public void validateForFlexiblePartialUpdate() throws UserException {
@@ -3234,6 +3235,9 @@ public void validateForFlexiblePartialUpdate() throws UserException {
         if (hasVariantColumns()) {
             throw new UserException("Flexible partial update can only support table without variant columns.");
         }
+        if (needRowBinlog()) {
+            throw new UserException("Flexible partial update does not support row binlog table.");
+        }
     }
 
     public boolean getEnableUniqueKeyMergeOnWrite() {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterTableCommand.java

@@ -295,6 +295,10 @@ private void rewriteAlterOpForOlapTable(ConnectContext ctx, OlapTable table) thr
                         throw new AnalysisException("Update flexible columns feature is only supported"
                                 + " on merge-on-write unique tables.");
                     }
+                    if (table.needRowBinlog()) {
+                        throw new AnalysisException("Update flexible columns feature does not support"
+                                + " row binlog table.");
+                    }
                     if (table.hasSkipBitmapColumn()) {
                         throw new AnalysisException("table " + table.getName()
                                 + " has enabled update flexible columns feature already.");

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java

@@ -723,6 +723,9 @@ public void validate(ConnectContext ctx) {
                     } catch (Exception e) {
                         throw new AnalysisException(e.getMessage(), e.getCause());
                     }
+                    if (isEnableSkipBitmapColumn && isEffectiveRowBinlogEnabled()) {
+                        throw new AnalysisException("Flexible partial update does not support row binlog table.");
+                    }
                 }
             }
 

fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableTest.java

@@ -88,6 +88,24 @@ private Replica mockReplica(Replica.ReplicaState state, long dataSize, long loca
         return replica;
     }
 
+    @Test
+    public void testFlexiblePartialUpdateRejectsRowBinlogTable() {
+        OlapTable table = Mockito.spy(new OlapTable());
+        Mockito.doReturn(true).when(table).getEnableUniqueKeyMergeOnWrite();
+        Mockito.doReturn(true).when(table).hasSkipBitmapColumn();
+        Mockito.doReturn(true).when(table).getEnableLightSchemaChange();
+        Mockito.doReturn(false).when(table).hasVariantColumns();
+        Mockito.doReturn(true).when(table).needRowBinlog();
+
+        try {
+            table.validateForFlexiblePartialUpdate();
+            Assert.fail("expected row binlog table to reject flexible partial update");
+        } catch (org.apache.doris.common.UserException e) {
+            Assert.assertTrue(e.getMessage().contains(
+                    "Flexible partial update does not support row binlog table."));
+        }
+    }
+
     @Test
     public void test() throws IOException {
 

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AlterTableCommandTest.java

@@ -19,9 +19,12 @@
 
 import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.info.TableNameInfo;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.UserException;
 import org.apache.doris.datasource.CatalogIf;
 import org.apache.doris.datasource.CatalogMgr;
 import org.apache.doris.datasource.mvcc.PluginDrivenMvccExternalTable;
@@ -34,12 +37,15 @@
 import org.apache.doris.nereids.trees.plans.commands.info.EnableFeatureOp;
 import org.apache.doris.nereids.trees.plans.commands.info.ModifyColumnOp;
 import org.apache.doris.nereids.trees.plans.commands.info.ReplacePartitionFieldOp;
+import org.apache.doris.qe.ConnectContext;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -129,6 +135,29 @@ void testMultiplePartitionFieldOps() {
         Assertions.assertTrue(sql.contains("ADD PARTITION KEY bucket(8, id)"));
     }
 
+    @Test
+    void testEnableFlexiblePartialUpdateRejectsRowBinlogTable() throws Exception {
+        List<AlterTableOp> ops = new ArrayList<>();
+        EnableFeatureOp op = new EnableFeatureOp("UPDATE_FLEXIBLE_COLUMNS");
+        op.validate(Mockito.mock(ConnectContext.class));
+        ops.add(op);
+        AlterTableCommand alterTableCommand = new AlterTableCommand(new TableNameInfo("db", "test"), ops);
+
+        OlapTable table = Mockito.mock(OlapTable.class);
+        Mockito.when(table.getKeysType()).thenReturn(KeysType.UNIQUE_KEYS);
+        Mockito.when(table.getEnableUniqueKeyMergeOnWrite()).thenReturn(true);
+        Mockito.when(table.needRowBinlog()).thenReturn(true);
+
+        Method method = AlterTableCommand.class.getDeclaredMethod(
+                "rewriteAlterOpForOlapTable", ConnectContext.class, OlapTable.class);
+        method.setAccessible(true);
+        InvocationTargetException exception = Assertions.assertThrows(InvocationTargetException.class,
+                () -> method.invoke(alterTableCommand, Mockito.mock(ConnectContext.class), table));
+        Assertions.assertInstanceOf(UserException.class, exception.getCause());
+        Assertions.assertTrue(exception.getCause().getMessage()
+                .contains("Update flexible columns feature does not support row binlog table."));
+    }
+
     @Test
     void testReplacePartitionFieldOp() {
         List<AlterTableOp> ops = new ArrayList<>();