Core: Only test if view exists when using SchemaVersion.V1 during tab…

…le rename (apache#9770)

core/src/main/java/org/apache/iceberg/jdbc/JdbcCatalog.java

@@ -39,8 +39,10 @@
 import java.util.stream.Stream;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.Schema;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.Transaction;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -313,6 +315,7 @@ public List<TableIdentifier> listTables(Namespace namespace) {
         JdbcUtil.namespaceToString(namespace));
   }
 
+  @SuppressWarnings("checkstyle:CyclomaticComplexity")
   @Override
   public void renameTable(TableIdentifier from, TableIdentifier to) {
     if (from.equals(to)) {
@@ -327,7 +330,7 @@ public void renameTable(TableIdentifier from, TableIdentifier to) {
       throw new NoSuchNamespaceException("Namespace does not exist: %s", to.namespace());
     }
 
-    if (viewExists(to)) {
+    if (schemaVersion == JdbcUtil.SchemaVersion.V1 && viewExists(to)) {
       throw new AlreadyExistsException("Cannot rename %s to %s. View already exists", from, to);
     }
 
@@ -812,4 +815,32 @@ private boolean deleteProperties(Namespace namespace, Set<String> properties) {
   protected Map<String, String> properties() {
     return catalogProperties == null ? ImmutableMap.of() : catalogProperties;
   }
+
+  @Override
+  public TableBuilder buildTable(TableIdentifier identifier, Schema schema) {
+    return new ViewAwareTableBuilder(identifier, schema);
+  }
+
+  /**
+   * The purpose of this class is to add view detection only when SchemaVersion.V1 schema is used
+   * when replacing a table.
+   */
+  protected class ViewAwareTableBuilder extends BaseMetastoreCatalogTableBuilder {
+
+    private final TableIdentifier identifier;
+
+    public ViewAwareTableBuilder(TableIdentifier identifier, Schema schema) {
+      super(identifier, schema);
+      this.identifier = identifier;
+    }
+
+    @Override
+    public Transaction replaceTransaction() {
+      if (schemaVersion == JdbcUtil.SchemaVersion.V1 && viewExists(identifier)) {
+        throw new AlreadyExistsException("View with same name already exists: %s", identifier);
+      }
+
+      return super.replaceTransaction();
+    }
+  }
 }

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java

@@ -140,7 +140,6 @@ private JdbcCatalog initCatalog(String catalogName, Map<String, String> props) {
 
     properties.put(JdbcCatalog.PROPERTY_PREFIX + "username", "user");
     properties.put(JdbcCatalog.PROPERTY_PREFIX + "password", "password");
-    properties.put(JdbcUtil.SCHEMA_VERSION_PROPERTY, JdbcUtil.SchemaVersion.V1.name());
     warehouseLocation = this.tableDir.toAbsolutePath().toString();
     properties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseLocation);
     properties.put("type", "jdbc");
@@ -154,7 +153,6 @@ public void testInitialize() {
     Map<String, String> properties = Maps.newHashMap();
     properties.put(CatalogProperties.WAREHOUSE_LOCATION, this.tableDir.toAbsolutePath().toString());
     properties.put(CatalogProperties.URI, "jdbc:sqlite:file::memory:?icebergDB");
-    properties.put(JdbcUtil.SCHEMA_VERSION_PROPERTY, JdbcUtil.SchemaVersion.V1.name());
     JdbcCatalog jdbcCatalog = new JdbcCatalog();
     jdbcCatalog.setConf(conf);
     jdbcCatalog.initialize("test_jdbc_catalog", properties);

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalogWithV1Schema.java

@@ -0,0 +1,66 @@
+/*
+ * 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.iceberg.jdbc;
+
+import java.util.Map;
+import java.util.UUID;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.catalog.CatalogTests;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+
+public class TestJdbcCatalogWithV1Schema extends CatalogTests<JdbcCatalog> {
+
+  private JdbcCatalog catalog;
+
+  @TempDir private java.nio.file.Path tableDir;
+
+  @Override
+  protected JdbcCatalog catalog() {
+    return catalog;
+  }
+
+  @Override
+  protected boolean supportsNamespaceProperties() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportsNestedNamespaces() {
+    return true;
+  }
+
+  @BeforeEach
+  public void setupCatalog() {
+    Map<String, String> properties = Maps.newHashMap();
+    properties.put(
+        CatalogProperties.URI,
+        "jdbc:sqlite:file::memory:?ic" + UUID.randomUUID().toString().replace("-", ""));
+    properties.put(JdbcCatalog.PROPERTY_PREFIX + "username", "user");
+    properties.put(JdbcCatalog.PROPERTY_PREFIX + "password", "password");
+    properties.put(CatalogProperties.WAREHOUSE_LOCATION, tableDir.toAbsolutePath().toString());
+    properties.put(JdbcUtil.SCHEMA_VERSION_PROPERTY, JdbcUtil.SchemaVersion.V1.name());
+
+    catalog = new JdbcCatalog();
+    catalog.setConf(new Configuration());
+    catalog.initialize("testCatalog", properties);
+  }
+}