[fix](fe) Allow table access through Iceberg JDBC V0 catalogs (branch-4.2)

fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java

@@ -29,6 +29,8 @@
 import org.apache.doris.cluster.ClusterNamespace;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.FeConstants;
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.ThreadPoolManager;
@@ -1303,6 +1305,24 @@ public void dropTable(String dbName, String tableName, boolean isView, boolean i
             }
             throw new DdlException("Failed to get table: '" + tableName + "' in database: " + dbName);
         }
+        // The metadata API receives no kind/lifetime flags and may purge an ordinary table.
+        // External catalogs have no temporary tables or MTMVs; reject mismatches before dispatch.
+        if (mustTemporary) {
+            if (ifExists) {
+                return;
+            }
+            ErrorReport.reportDdlException(ErrorCode.ERR_UNKNOWN_TABLE, tableName, dbName);
+        }
+        if (isMtmv) {
+            ErrorReport.reportDdlException(ErrorCode.ERR_WRONG_OBJECT, dbName, tableName, "MTMV",
+                    "External catalogs do not support materialized views");
+        }
+        boolean actualIsView = dorisTable.isView();
+        if (isView != actualIsView) {
+            // IF EXISTS must not relax the type check for an existing object.
+            ErrorReport.reportDdlException(ErrorCode.ERR_WRONG_OBJECT, dbName, tableName, isView ? "VIEW" : "TABLE",
+                    "Use DROP " + (actualIsView ? "VIEW" : "TABLE") + " instead");
+        }
         try {
             metadataOps.dropTable(dorisTable, ifExists);
             DropInfo info = new DropInfo(getName(), db.getFullName(), dorisTable.getName(), db.getFullName());

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java

@@ -77,6 +77,7 @@
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.expressions.Literal;
 import org.apache.iceberg.expressions.Term;
+import org.apache.iceberg.jdbc.JdbcCatalog;
 import org.apache.iceberg.rest.RESTSessionCatalog;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.TypeUtil;
@@ -100,6 +101,7 @@
 import java.util.TreeSet;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -2061,6 +2063,19 @@ private boolean determineNestedNamespaceEnabled() {
         return userSessionCatalog != null && userSessionCatalog.isNestedNamespaceEnabled();
     }
 
+    private <T> T probeViews(ViewCatalog viewCatalog, Supplier<T> probe, T noViews) {
+        try {
+            return probe.get();
+        } catch (UnsupportedOperationException e) {
+            // JdbcCatalog implements ViewCatalog even for V0, which cannot store views. The SDK
+            // rejects these probes before SQL; do not require a V1 migration merely to read tables.
+            if (viewCatalog instanceof JdbcCatalog) {
+                return noViews;
+            }
+            throw e;
+        }
+    }
+
     private boolean isViewCatalogEnabled() {
         return viewCatalogEnabled;
     }
@@ -2141,8 +2156,8 @@ private boolean viewExistsInternal(String remoteDbName, String remoteViewName) {
     }
 
     private boolean viewExistsInternal(SessionContext ctx, String remoteDbName, String remoteViewName) {
-        return viewCatalog(ctx).map(activeCatalog ->
-                activeCatalog.viewExists(getTableIdentifier(remoteDbName, remoteViewName))).orElse(false);
+        return viewCatalog(ctx).map(activeCatalog -> probeViews(activeCatalog,
+                () -> activeCatalog.viewExists(getTableIdentifier(remoteDbName, remoteViewName)), false)).orElse(false);
     }
 
     private View loadViewInternal(String dbName, String tblName) {
@@ -2159,8 +2174,10 @@ private List<String> listViewNamesInternal(String dbName) {
     }
 
     private List<String> listViewNamesInternal(SessionContext ctx, String dbName) {
-        return viewCatalog(ctx).map(activeCatalog -> activeCatalog.listViews(getNamespace(dbName))
-                .stream().map(TableIdentifier::name).collect(Collectors.toList())).orElseGet(Collections::emptyList);
+        return viewCatalog(ctx).map(activeCatalog -> probeViews(activeCatalog,
+                () -> activeCatalog.listViews(getNamespace(dbName)).stream()
+                        .map(TableIdentifier::name).collect(Collectors.toList()), Collections.<String>emptyList()))
+                .orElseGet(Collections::emptyList);
     }
 
     private void performDropView(SessionContext ctx, String remoteDbName, String remoteViewName) throws DdlException {

fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalCatalogDropKindTest.java

@@ -0,0 +1,221 @@
+// 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.
+
+package org.apache.doris.datasource;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
+import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
+import org.apache.doris.persist.EditLog;
+
+import mockit.Mock;
+import mockit.MockUp;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.jdbc.JdbcCatalog;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+public class ExternalCatalogDropKindTest {
+    @Test
+    public void dropViewCannotDeleteV0Table() {
+        for (boolean ifExists : new boolean[] {false, true}) {
+            DropCatalog catalog = new DropCatalog(false);
+            Assertions.assertThrows(DdlException.class,
+                    () -> catalog.dropTable("db", "t", true, false, ifExists, false, false));
+            catalog.assertTableUnchanged();
+        }
+    }
+
+    @Test
+    public void dropMaterializedViewCannotDeleteV0Table() {
+        for (boolean ifExists : new boolean[] {false, true}) {
+            DropCatalog catalog = new DropCatalog(false);
+            Assertions.assertThrows(DdlException.class,
+                    () -> catalog.dropTable("db", "t", false, true, ifExists, false, false));
+            catalog.assertTableUnchanged();
+        }
+    }
+
+    @Test
+    public void dropTemporaryTableCannotDeleteV0Table() {
+        DropCatalog catalog = new DropCatalog(false);
+        Assertions.assertThrows(DdlException.class,
+                () -> catalog.dropTable("db", "t", false, false, false, true, false));
+        catalog.assertTableUnchanged();
+    }
+
+    @Test
+    public void dropTemporaryTableIfExistsDoesNotDeleteV0Table() throws Exception {
+        DropCatalog catalog = new DropCatalog(false);
+        catalog.dropTable("db", "t", false, false, true, true, false);
+        catalog.assertTableUnchanged();
+    }
+
+    @Test
+    public void dropTableCannotDeleteView() {
+        for (boolean ifExists : new boolean[] {false, true}) {
+            DropCatalog catalog = new DropCatalog(true);
+            Assertions.assertThrows(DdlException.class,
+                    () -> catalog.dropTable("db", "t", false, false, ifExists, false, false));
+            Assertions.assertEquals(0, catalog.sdk.viewDrops);
+            Assertions.assertTrue(catalog.sdk.present);
+            Mockito.verifyNoInteractions(catalog.editLog);
+        }
+    }
+
+    @Test
+    public void matchingV0TableDropStillWorks() throws Exception {
+        DropCatalog catalog = new DropCatalog(false);
+        catalog.dropTable("db", "t", false, false, false, false, false);
+        Assertions.assertFalse(catalog.sdk.present);
+        Assertions.assertEquals(1, catalog.sdk.tableDrops);
+        Assertions.assertTrue(catalog.sdk.purge);
+        Mockito.verify(catalog.editLog).logDropTable(Mockito.any());
+    }
+
+    @Test
+    public void matchingViewDropStillWorks() throws Exception {
+        DropCatalog catalog = new DropCatalog(true);
+        catalog.dropTable("db", "t", true, false, false, false, false);
+        Assertions.assertFalse(catalog.sdk.present);
+        Assertions.assertEquals(1, catalog.sdk.viewDrops);
+        Assertions.assertEquals(0, catalog.sdk.tableDrops);
+        Mockito.verify(catalog.editLog).logDropTable(Mockito.any());
+    }
+
+    @Test
+    public void missingObjectIfExistsStillDoesNotMutate() throws Exception {
+        for (boolean view : new boolean[] {false, true}) {
+            DropCatalog catalog = new DropCatalog(false);
+            Mockito.when(catalog.database.getTableNullable("t")).thenReturn(null);
+            catalog.dropTable("db", "t", view, false, true, false, false);
+            catalog.assertTableUnchanged();
+        }
+    }
+
+    private static class DropCatalog extends ExternalCatalog {
+        private final RecordingJdbcCatalog sdk;
+        private final ExternalDatabase<ExternalTable> database;
+        private final EditLog editLog = Mockito.mock(EditLog.class);
+
+        private DropCatalog(boolean view) {
+            super(1L, "drop_kind_test", InitCatalogLog.Type.TEST, "");
+            initialized = true;
+            catalogProperty = new CatalogProperty(null, Collections.emptyMap());
+            executionAuthenticator = new ExecutionAuthenticator() {
+            };
+            sdk = new RecordingJdbcCatalog(view);
+            metadataOps = new IcebergMetadataOps(this, sdk) {
+                @Override
+                public void afterDropTable(String dbName, String tableName) {
+                    // Cache retirement is outside this test's drop-kind and mutation boundary.
+                }
+            };
+            database = Mockito.mock(ExternalDatabase.class);
+            ExternalTable table = Mockito.mock(ExternalTable.class);
+            Mockito.when(database.getTableNullable("t")).thenReturn(table);
+            Mockito.when(database.getFullName()).thenReturn("db");
+            Mockito.when(table.getName()).thenReturn("t");
+            Mockito.when(table.getRemoteName()).thenReturn("t");
+            Mockito.when(table.getDbName()).thenReturn("db");
+            Mockito.when(table.getRemoteDbName()).thenReturn("db");
+            Mockito.when(table.isView()).thenReturn(view);
+            Env env = Mockito.mock(Env.class);
+            Mockito.when(env.getEditLog()).thenReturn(editLog);
+            new MockUp<Env>() {
+                @Mock
+                Env getCurrentEnv() {
+                    return env;
+                }
+            };
+        }
+
+        private void assertTableUnchanged() {
+            Assertions.assertTrue(sdk.present);
+            Assertions.assertEquals(Arrays.asList(1, 2), sdk.rows);
+            Assertions.assertEquals(0, sdk.tableDrops);
+            Assertions.assertEquals(0, sdk.viewDrops);
+            Mockito.verifyNoInteractions(editLog);
+        }
+
+        @Override
+        protected void initLocalObjectsImpl() {
+        }
+
+        @Override
+        public ExternalDatabase<ExternalTable> getDbNullable(String dbName) {
+            return database;
+        }
+
+        @Override
+        protected List<String> listTableNamesFromRemote(SessionContext ctx, String dbName) {
+            return Collections.singletonList("t");
+        }
+
+        @Override
+        public boolean tableExist(SessionContext ctx, String dbName, String tblName) {
+            return sdk.present;
+        }
+    }
+
+    // Retain the real SDK's V0 view guard while recording the destructive table operation it used to block.
+    private static class RecordingJdbcCatalog extends JdbcCatalog {
+        private final boolean view;
+        private final List<Integer> rows = new ArrayList<>(Arrays.asList(1, 2));
+        private boolean present = true;
+        private boolean purge;
+        private int tableDrops;
+        private int viewDrops;
+
+        private RecordingJdbcCatalog(boolean view) {
+            this.view = view;
+        }
+
+        @Override
+        public boolean viewExists(TableIdentifier identifier) {
+            return view ? present : super.viewExists(identifier);
+        }
+
+        @Override
+        public boolean tableExists(TableIdentifier identifier) {
+            return present;
+        }
+
+        @Override
+        public boolean dropTable(TableIdentifier identifier, boolean purge) {
+            this.purge = purge;
+            tableDrops++;
+            present = false;
+            rows.clear();
+            return true;
+        }
+
+        @Override
+        public boolean dropView(TableIdentifier identifier) {
+            viewDrops++;
+            present = false;
+            return true;
+        }
+    }
+}