[BugFix] Fix Delta Lake and Kudu un-partitioned materialized view query rewrite

fe/fe-core/src/main/java/com/starrocks/connector/partitiontraits/DeltaLakePartitionTraits.java

@@ -14,6 +14,7 @@
 
 package com.starrocks.connector.partitiontraits;
 
+import com.google.common.collect.Sets;
 import com.starrocks.catalog.BaseTableInfo;
 import com.starrocks.catalog.DeltaLakePartitionKey;
 import com.starrocks.catalog.MaterializedView;
@@ -37,7 +38,11 @@ public boolean isSupportPCTRefresh() {
     @Override
     public Set<String> getUpdatedPartitionNames(List<BaseTableInfo> baseTables,
                                                 MaterializedView.AsyncRefreshContext context) {
-        // TODO: implement
-        return null;
+        // TODO: Implement Delta Lake partition update tracking. Until then, return an empty set (meaning
+        //  "no updated partitions detected") rather than null. Returning null is interpreted as "update info
+        //  unknown", which forces the timeliness arbiter to treat the MV as stale (full refresh) and disables
+        //  query rewrite entirely in the default CHECKED consistency mode. Empty set keeps the MV eligible for
+        //  rewrite, matching the behavior of the other not-yet-implemented connectors (e.g. Hudi).
+        return Sets.newHashSet();
     }
 }

fe/fe-core/src/main/java/com/starrocks/connector/partitiontraits/KuduPartitionTraits.java

@@ -13,6 +13,7 @@
 // limitations under the License.
 package com.starrocks.connector.partitiontraits;
 
+import com.google.common.collect.Sets;
 import com.starrocks.catalog.BaseTableInfo;
 import com.starrocks.catalog.KuduPartitionKey;
 import com.starrocks.catalog.MaterializedView;
@@ -35,7 +36,11 @@ public PartitionKey createEmptyKey() {
     @Override
     public Set<String> getUpdatedPartitionNames(List<BaseTableInfo> baseTables,
                                                 MaterializedView.AsyncRefreshContext context) {
-        // TODO: implement
-        return null;
+        // TODO: Implement Kudu partition update tracking. Until then, return an empty set (meaning
+        //  "no updated partitions detected") rather than null. Returning null is interpreted as "update info
+        //  unknown", which forces the timeliness arbiter to treat the MV as stale (full refresh) and disables
+        //  query rewrite entirely in the default CHECKED consistency mode. Empty set keeps the MV eligible for
+        //  rewrite, matching the behavior of the other not-yet-implemented connectors (e.g. Hudi).
+        return Sets.newHashSet();
     }
 }

fe/fe-core/src/test/java/com/starrocks/sql/optimizer/rule/transformation/materialization/MvRewriteDeltaLakeTest.java

@@ -0,0 +1,93 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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.
+
+package com.starrocks.sql.optimizer.rule.transformation.materialization;
+
+import com.google.common.base.Joiner;
+import com.starrocks.catalog.BaseTableInfo;
+import com.starrocks.catalog.Database;
+import com.starrocks.catalog.DeltaLakeTable;
+import com.starrocks.catalog.MaterializedView;
+import com.starrocks.catalog.MvRefreshArbiter;
+import com.starrocks.catalog.MvUpdateInfo;
+import com.starrocks.catalog.Table;
+import com.starrocks.catalog.mv.MVTimelinessArbiter;
+import com.starrocks.qe.ConnectContext;
+import com.starrocks.server.GlobalStateMgr;
+import com.starrocks.server.MetadataMgr;
+import com.starrocks.sql.plan.ConnectorPlanTestBase;
+import mockit.Mock;
+import mockit.MockUp;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Optional;
+
+public class MvRewriteDeltaLakeTest extends MVTestBase {
+
+    @BeforeAll
+    public static void beforeClass() throws Exception {
+        MVTestBase.beforeClass();
+        ConnectorPlanTestBase.mockCatalog(connectContext, "deltalake_catalog");
+    }
+
+    /**
+     * Regression guard for the Delta Lake un-partitioned MV query rewrite bug (issue #11409).
+     *
+     * <p>{@code DeltaLakePartitionTraits#getUpdatedPartitionNames} used to return {@code null}. The MV timeliness
+     * arbiter interprets a null base-table update info as "unknown", returns {@link MvUpdateInfo.MvToRefreshType#FULL}
+     * and therefore treats the MV as stale, disabling query rewrite in the default CHECKED consistency mode.
+     *
+     * <p>After the fix it returns an empty set ("no updated partitions detected"), so the arbiter returns
+     * {@link MvUpdateInfo.MvToRefreshType#NO_REFRESH} and the MV stays eligible for rewrite. This test asserts that
+     * timeliness decision directly, which is exactly the signal that gates the rewrite (a full end-to-end EXPLAIN
+     * assertion is covered by the SQL test {@code test_mv_deltalake_rewrite}).
+     */
+    @Test
+    public void testDeltaLakeUnPartitionedMvIsNotStaleInCheckedMode() throws Exception {
+        // The mocked Delta Lake table has no snapshot, so stub a stable table identifier (used for base-table
+        // matching) to avoid an NPE and keep the query/MV base-table identifiers consistent.
+        new MockUp<DeltaLakeTable>() {
+            @Mock
+            public String getTableIdentifier() {
+                return Joiner.on(":").join("tbl", "mocked-delta-uuid");
+            }
+        };
+        new MockUp<MetadataMgr>() {
+            @Mock
+            public Optional<Table> getTableWithIdentifier(ConnectContext context, BaseTableInfo baseTableInfo) {
+                return Optional.of(new DeltaLakeTable());
+            }
+        };
+
+        String mvName = "mv_delta_checked";
+        // No query_rewrite_consistency property => defaults to CHECKED, the mode in which the bug manifested.
+        starRocksAssert.withMaterializedView("create materialized view " + mvName +
+                " refresh deferred manual " +
+                " as select col1, col2 from deltalake_catalog.deltalake_db.tbl");
+
+        Database testDb = GlobalStateMgr.getCurrentState().getLocalMetastore().getDb("test");
+        MaterializedView mv = (MaterializedView) GlobalStateMgr.getCurrentState().getLocalMetastore()
+                .getTable(testDb.getFullName(), mvName);
+
+        MvUpdateInfo info = MvRefreshArbiter.getMVTimelinessUpdateInfo(
+                mv, MVTimelinessArbiter.QueryRewriteParams.ofRefresh());
+        // Before the fix this was FULL (stale => no rewrite); after the fix it must be NO_REFRESH (rewriteable).
+        Assertions.assertEquals(MvUpdateInfo.MvToRefreshType.NO_REFRESH, info.getMVToRefreshType());
+        Assertions.assertTrue(info.isValidRewrite());
+
+        starRocksAssert.dropMaterializedView(mvName);
+    }
+}

fe/fe-core/src/test/java/com/starrocks/sql/optimizer/rule/transformation/materialization/MvRewriteKuduTest.java

@@ -0,0 +1,107 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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.
+
+package com.starrocks.sql.optimizer.rule.transformation.materialization;
+
+import com.google.common.collect.ImmutableList;
+import com.starrocks.catalog.BaseTableInfo;
+import com.starrocks.catalog.Column;
+import com.starrocks.catalog.Database;
+import com.starrocks.catalog.KuduTable;
+import com.starrocks.catalog.MaterializedView;
+import com.starrocks.catalog.MvRefreshArbiter;
+import com.starrocks.catalog.MvUpdateInfo;
+import com.starrocks.catalog.Table;
+import com.starrocks.catalog.mv.MVTimelinessArbiter;
+import com.starrocks.connector.kudu.KuduMetadata;
+import com.starrocks.qe.ConnectContext;
+import com.starrocks.server.GlobalStateMgr;
+import com.starrocks.server.MetadataMgr;
+import com.starrocks.sql.plan.ConnectorPlanTestBase;
+import com.starrocks.type.IntegerType;
+import com.starrocks.type.StringType;
+import mockit.Mock;
+import mockit.MockUp;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+public class MvRewriteKuduTest extends MVTestBase {
+
+    private static KuduTable kuduTable;
+
+    @BeforeAll
+    public static void beforeClass() throws Exception {
+        MVTestBase.beforeClass();
+        ConnectorPlanTestBase.mockCatalog(connectContext, ConnectorPlanTestBase.MOCK_KUDU_CATALOG_NAME);
+
+        List<Column> columns = ImmutableList.of(
+                new Column("col1", IntegerType.INT),
+                new Column("col2", StringType.STRING));
+        kuduTable = new KuduTable("localhost:7051", ConnectorPlanTestBase.MOCK_KUDU_CATALOG_NAME,
+                "kudu_db", "kudu_tbl", "kudu_tbl", columns, new ArrayList<>());
+
+        // The mocked kudu0 catalog is backed by a real KuduMetadata pointing at a fake master, so stub the
+        // metadata lookups that MV creation / rewrite need to a canned unpartitioned table.
+        new MockUp<KuduMetadata>() {
+            @Mock
+            public Table getTable(ConnectContext context, String dbName, String tblName) {
+                return kuduTable;
+            }
+
+            @Mock
+            public Database getDb(ConnectContext context, String dbName) {
+                return new Database(GlobalStateMgr.getCurrentState().getNextId(), dbName);
+            }
+        };
+        new MockUp<MetadataMgr>() {
+            @Mock
+            public Optional<Table> getTableWithIdentifier(ConnectContext context, BaseTableInfo baseTableInfo) {
+                return Optional.of(kuduTable);
+            }
+        };
+    }
+
+    /**
+     * Regression guard for the Kudu external-table MV query rewrite bug (same root cause as the Delta Lake case,
+     * issue #11409): {@code KuduPartitionTraits#getUpdatedPartitionNames} used to return {@code null}, so the MV
+     * timeliness arbiter returned {@link MvUpdateInfo.MvToRefreshType#FULL} and treated the MV as stale, disabling
+     * query rewrite in the default CHECKED consistency mode. After the fix it returns an empty set, so the arbiter
+     * returns {@link MvUpdateInfo.MvToRefreshType#NO_REFRESH} and the MV stays eligible for rewrite.
+     */
+    @Test
+    public void testKuduUnPartitionedMvIsNotStaleInCheckedMode() throws Exception {
+        String mvName = "mv_kudu_checked";
+        // No query_rewrite_consistency property => defaults to CHECKED, the mode in which the bug manifested.
+        starRocksAssert.withMaterializedView("create materialized view " + mvName +
+                " refresh deferred manual " +
+                " as select col1, col2 from kudu0.kudu_db.kudu_tbl");
+
+        Database testDb = GlobalStateMgr.getCurrentState().getLocalMetastore().getDb("test");
+        MaterializedView mv = (MaterializedView) GlobalStateMgr.getCurrentState().getLocalMetastore()
+                .getTable(testDb.getFullName(), mvName);
+
+        MvUpdateInfo info = MvRefreshArbiter.getMVTimelinessUpdateInfo(
+                mv, MVTimelinessArbiter.QueryRewriteParams.ofRefresh());
+        // Before the fix this was FULL (stale => no rewrite); after the fix it must be NO_REFRESH (rewriteable).
+        Assertions.assertEquals(MvUpdateInfo.MvToRefreshType.NO_REFRESH, info.getMVToRefreshType());
+        Assertions.assertTrue(info.isValidRewrite());
+
+        starRocksAssert.dropMaterializedView(mvName);
+    }
+}