From 0d0d948ee5b2f5a8322f827c926493912925a160 Mon Sep 17 00:00:00 2001 From: TsukiokaKogane Date: Sat, 11 Jul 2026 16:51:22 +0800 Subject: [PATCH] [fix](binlog) Report missing row binlog for INCR queries --- .../nereids/rules/analysis/BindRelation.java | 23 +++++------ .../rules/analysis/BindRelationTest.java | 11 +++++ .../test_incr_without_row_binlog.groovy | 41 +++++++++++++++++++ 3 files changed, 62 insertions(+), 13 deletions(-) create mode 100644 regression-test/suites/row_binlog_p0/test_incr_without_row_binlog.groovy diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java index 28515845d30a61..8b5a579a2a289a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java +++ b/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 partIds = getPartitionIds(table, unboundRelation, qualifier); List 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 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 keys = new HashSet(scanParams.getMapParams().keySet()); keys.remove(OlapScanNode.OLAP_INCREMENT_TYPE); diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindRelationTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindRelationTest.java index e877989f41e8a8..b74abb9c369d72 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindRelationTest.java +++ b/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(); diff --git a/regression-test/suites/row_binlog_p0/test_incr_without_row_binlog.groovy b/regression-test/suites/row_binlog_p0/test_incr_without_row_binlog.groovy new file mode 100644 index 00000000000000..f26141a048df64 --- /dev/null +++ b/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." + } +}