Iceberg.engine.hive.enabled Conf is not honouring for HIVE CATALOG #10286

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java

@@ -36,6 +36,7 @@
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hadoop.ConfigProperties;
 import org.apache.iceberg.hive.HiveSchemaUtil;
 import org.apache.iceberg.hive.HiveTableOperations;
 import org.apache.iceberg.io.FileIO;
@@ -151,7 +152,10 @@ public void rollbackCreateTable(org.apache.hadoop.hive.metastore.api.Table hmsTa
   @Override
   public void commitCreateTable(org.apache.hadoop.hive.metastore.api.Table hmsTable) {
     if (icebergTable == null) {
-      if (Catalogs.hiveCatalog(conf, catalogProperties)) {
+      // Check if the catalog is a Hive catalog and if the Hive engine is enabled in the
+      // configuration
+      if (Catalogs.hiveCatalog(conf, catalogProperties)
+          && Boolean.parseBoolean(conf.get(ConfigProperties.ENGINE_HIVE_ENABLED))) {
         catalogProperties.put(TableProperties.ENGINE_HIVE_ENABLED, true);
       }
 

mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerNoScan.java

@@ -25,6 +25,7 @@
 import static org.assertj.core.api.Assumptions.assumeThat;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -70,6 +71,8 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestTemplate;
+import org.junit.Assert;
+import org.junit.Assume;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -498,6 +501,47 @@ public void testCreateTableError() {
     }
   }
 
+  @TestTemplate
+  public void testEngineHiveDisable() throws TException, IOException {
+    Assume.assumeTrue(
+        "Only HiveCatalog attempts to load the Iceberg table prior to dropping it.",
+        testTableType == TestTables.TestTableType.HIVE_CATALOG);
+
+    shell.setHiveSessionValue("iceberg.engine.hive.enabled", "false");
+
+    TableIdentifier identifier = TableIdentifier.of("default", "customers");
+
+    shell.executeStatement(
+        "CREATE EXTERNAL TABLE customers "
+            + "STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler' "
+            + testTables.locationForCreateTableSQL(identifier)
+            + "TBLPROPERTIES ('"
+            + InputFormatConfig.TABLE_SCHEMA
+            + "'='"
+            + SchemaParser.toJson(HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA)
+            + "','"
+            + InputFormatConfig.CATALOG_NAME
+            + "'='"
+            + Catalogs.ICEBERG_DEFAULT_CATALOG_NAME
+            + "')");
+
+    List<Object[]> rows = shell.executeStatement("DESCRIBE formatted default.customers");
+    boolean containsExpectedString = false;
+
+    for (Object[] row : rows) {
+      containsExpectedString =
+          Arrays.stream(row)
+              .filter(item -> item instanceof String)
+              .anyMatch(
+                  item -> ((String) item).contains("org.apache.hadoop.mapred.FileInputFormat"));
+      if (containsExpectedString) {
+        break;
+      }
+    }
+
+    Assert.assertTrue("The row does not contain the expected string", containsExpectedString);
+  }
+
   @TestTemplate
   public void testCreateTableAboveExistingTable() throws IOException {
     // Create the Iceberg table