[fix](binlog) Report missing row binlog for INCR queries

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java

@@ -28,7 +28,6 @@
 import org.apache.doris.catalog.FunctionRegistry;
 import org.apache.doris.catalog.KeysType;
 import org.apache.doris.catalog.OlapTable;
-import org.apache.doris.catalog.OlapTableWrapper;
 import org.apache.doris.catalog.Partition;
 import org.apache.doris.catalog.RowBinlogTableWrapper;
 import org.apache.doris.catalog.SchemaTable;
@@ -242,9 +241,8 @@ private LogicalPlan makeOlapScan(TableIf table, UnboundRelation unboundRelation,
         LogicalOlapScan scan;
         List<Long> partIds = getPartitionIds(table, unboundRelation, qualifier);
         List<Long> tabletIds = unboundRelation.getTabletIds();
-        boolean isChangeRead = unboundRelation.getScanParams() != null
-                && unboundRelation.getScanParams().incrementalRead();
-        if (isChangeRead) {
+        StreamScanType changeScanType = checkChangeScanCondition((OlapTable) table, unboundRelation.getScanParams());
+        if (changeScanType != null) {
             table = new RowBinlogTableWrapper((OlapTable) table);
         } else if (unboundRelation.getScanParams() != null) {
             unboundRelation.getScanParams().validateOlapTable();
@@ -265,7 +263,7 @@ private LogicalPlan makeOlapScan(TableIf table, UnboundRelation unboundRelation,
                     throw new AnalysisException("Table " + olapTable.getName()
                         + " doesn't have materialized view " + indexName.get());
                 }
-                if (isChangeRead && olapTable.getBaseIndexId() != indexId) {
+                if (changeScanType != null && olapTable.getBaseIndexId() != indexId) {
                     throw new AnalysisException("Change read is not supported on non-base index " + indexName.get());
                 }
                 if (unboundRelation.getTableSnapshot().isPresent() && olapTable.getBaseIndexId() != indexId) {
@@ -291,15 +289,13 @@ private LogicalPlan makeOlapScan(TableIf table, UnboundRelation unboundRelation,
             // This tabletIds is set manually, so need to set specifiedTabletIds
             scan = scan.withManuallySpecifiedTabletIds(tabletIds);
         }
-        if (isChangeRead) {
+        if (changeScanType != null) {
             if (cascadesContext.getStatementContext().isHintForcePreAggOn()) {
                 throw new AnalysisException(
                         "PREAGGOPEN hint is not supported on @incr (change-read) scans.");
             }
-            StreamScanType scanType = checkChangeScanCondition(((OlapTableWrapper) table).getOriginTable(),
-                    unboundRelation.getScanParams());
-            return checkAndAddChangeScanFilter(scan, scanType, parseTimestampRange(unboundRelation.getScanParams()),
-                    false);
+            return checkAndAddChangeScanFilter(scan, changeScanType,
+                    parseTimestampRange(unboundRelation.getScanParams()), false);
         }
         // Time-travel (FOR VERSION/TIME AS OF): wrap the scan with a __DORIS_COMMIT_TSO_COL__
         // predicate (dup) or a base/binlog union (mow).
@@ -647,10 +643,11 @@ private Optional<LogicalPlan> handleMetaTable(TableIf table, UnboundRelation unb
 
     private StreamScanType checkChangeScanCondition(OlapTable olapTable, TableScanParams scanParams)
             throws AnalysisException {
+        if (scanParams == null || !scanParams.incrementalRead()) {
+            return null;
+        }
         if (!olapTable.needRowBinlog()) {
-            throw new AnalysisException("INCR query requires ROW binlog enabled on base table "
-                    + "(PROPERTIES('binlog.enable'='true','binlog.format'='ROW')). "
-                    + "Table " + olapTable.getQualifiedName() + " doesn't enable row binlog.");
+            throw new AnalysisException("INCR query requires ROW binlog enabled on base table.");
         }
         HashSet<String> keys = new HashSet(scanParams.getMapParams().keySet());
         keys.remove(OlapScanNode.OLAP_INCREMENT_TYPE);

fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindRelationTest.java

@@ -18,6 +18,7 @@
 package org.apache.doris.nereids.rules.analysis;
 
 import org.apache.doris.nereids.analyzer.UnboundRelation;
+import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.pattern.GeneratedPlanPatterns;
 import org.apache.doris.nereids.rules.RulePromise;
 import org.apache.doris.nereids.trees.expressions.Alias;
@@ -90,6 +91,16 @@ void bindByDbQualifier() {
                 ((LogicalOlapScan) plan).qualified());
     }
 
+    @Test
+    void rejectIncrementalReadWithoutRowBinlog() {
+        AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+                () -> PlanChecker.from(connectContext)
+                        .analyze("SELECT a, b, __DORIS_BINLOG_OP__ "
+                                + "FROM db1.t@incr(\"incrementType\" = \"DETAIL\")"));
+
+        Assertions.assertEquals("INCR query requires ROW binlog enabled on base table.", exception.getMessage());
+    }
+
     @Test
     void bindSchemaTable() {
         boolean originValue = connectContext.getSessionVariable().isFetchAllFeForSystemTable();

regression-test/suites/row_binlog_p0/test_incr_without_row_binlog.groovy

@@ -0,0 +1,41 @@
+// 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.
+
+suite("test_incr_without_row_binlog") {
+    if (isCloudMode()) {
+        return
+    }
+
+    sql "DROP TABLE IF EXISTS base_incr_no_binlog"
+    sql """
+        CREATE TABLE base_incr_no_binlog (
+            k1 INT,
+            v1 INT
+        )
+        DUPLICATE KEY(k1)
+        DISTRIBUTED BY HASH(k1) BUCKETS 1
+        PROPERTIES ("replication_num" = "1")
+    """
+
+    test {
+        sql """
+            SELECT k1, v1, __DORIS_BINLOG_OP__
+            FROM base_incr_no_binlog@incr("incrementType" = "DETAIL")
+        """
+        exception "errCode = 2, detailMessage = INCR query requires ROW binlog enabled on base table."
+    }
+}