[hive] Add catalog option to control alter table cascade

docs/generated/hive_catalog_configuration.html

@@ -26,6 +26,12 @@
         </tr>
     </thead>
     <tbody>
+        <tr>
+            <td><h5>alter-table-cascade</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Whether to cascade schema changes to Hive metastore partitions when altering table.</td>
+        </tr>
         <tr>
             <td><h5>client-pool-cache.eviction-interval-ms</h5></td>
             <td style="word-wrap: break-word;">300000</td>

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveAlterTableUtils.java

@@ -30,23 +30,33 @@
 public class HiveAlterTableUtils {
 
     public static void alterTable(
-            IMetaStoreClient client, Identifier identifier, Table table, boolean skipUpdateStats)
+            IMetaStoreClient client,
+            Identifier identifier,
+            Table table,
+            boolean skipUpdateStats,
+            boolean cascade)
             throws TException {
         try {
-            alterTableWithEnv(client, identifier, table, skipUpdateStats);
+            alterTableWithEnv(client, identifier, table, skipUpdateStats, cascade);
         } catch (NoClassDefFoundError | NoSuchMethodError e) {
             alterTableWithoutEnv(client, identifier, table);
         }
     }
 
     private static void alterTableWithEnv(
-            IMetaStoreClient client, Identifier identifier, Table table, boolean skipUpdateStats)
+            IMetaStoreClient client,
+            Identifier identifier,
+            Table table,
+            boolean skipUpdateStats,
+            boolean cascade)
             throws TException {
         boolean skipHiveUpdateStats =
                 Boolean.parseBoolean(table.getParameters().get(StatsSetupConst.DO_NOT_UPDATE_STATS))
                         || skipUpdateStats;
         EnvironmentContext environmentContext = new EnvironmentContext();
-        environmentContext.putToProperties(StatsSetupConst.CASCADE, "true");
+        if (cascade) {
+            environmentContext.putToProperties(StatsSetupConst.CASCADE, "true");
+        }
         environmentContext.putToProperties(
                 StatsSetupConst.DO_NOT_UPDATE_STATS, Boolean.toString(skipHiveUpdateStats));
         client.alter_table_with_environmentContext(

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java

@@ -114,6 +114,7 @@
 import static org.apache.paimon.catalog.CatalogUtils.listPartitionsFromFileSystem;
 import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
 import static org.apache.paimon.format.csv.CsvOptions.FIELD_DELIMITER;
+import static org.apache.paimon.hive.HiveCatalogOptions.ALTER_TABLE_CASCADE;
 import static org.apache.paimon.hive.HiveCatalogOptions.HADOOP_CONF_DIR;
 import static org.apache.paimon.hive.HiveCatalogOptions.HIVE_CONF_DIR;
 import static org.apache.paimon.hive.HiveCatalogOptions.HIVE_SKIP_UPDATE_STATS;
@@ -1298,7 +1299,11 @@ private void alterTableToHms(
                 .execute(
                         client ->
                                 HiveAlterTableUtils.alterTable(
-                                        client, identifier, table, skipUpdateStats));
+                                        client,
+                                        identifier,
+                                        table,
+                                        skipUpdateStats,
+                                        options.get(ALTER_TABLE_CASCADE)));
     }
 
     @Override

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogOptions.java

@@ -102,5 +102,12 @@ public final class HiveCatalogOptions {
                                                             + "E.g. specifying \"conf:a.b.c\" will add \"a.b.c\" to the key, and so that configurations with different default catalog wouldn't share the same client pool. Multiple conf elements can be specified."))
                                     .build());
 
+    public static final ConfigOption<Boolean> ALTER_TABLE_CASCADE =
+            ConfigOptions.key("alter-table-cascade")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Whether to cascade schema changes to Hive metastore partitions when altering table.");
+
     private HiveCatalogOptions() {}
 }

paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveAlterTableUtilsTest.java

@@ -0,0 +1,80 @@
+/*
+ * 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.paimon.hive;
+
+import org.apache.paimon.catalog.Identifier;
+
+import org.apache.hadoop.hive.common.StatsSetupConst;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.EnvironmentContext;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.HashMap;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link HiveAlterTableUtils}. */
+public class HiveAlterTableUtilsTest {
+
+    @Test
+    public void testAlterTableWithoutCascade() throws Exception {
+        IMetaStoreClient client = mock(IMetaStoreClient.class);
+        Identifier identifier = Identifier.create("db", "tbl");
+        Table table = new Table();
+        table.setParameters(new HashMap<>());
+
+        HiveAlterTableUtils.alterTable(client, identifier, table, true, false);
+
+        EnvironmentContext context = captureEnvironmentContext(client, table);
+        assertThat(context.getProperties())
+                .containsEntry(StatsSetupConst.DO_NOT_UPDATE_STATS, "true")
+                .doesNotContainKey(StatsSetupConst.CASCADE);
+    }
+
+    @Test
+    public void testAlterTableWithCascade() throws Exception {
+        IMetaStoreClient client = mock(IMetaStoreClient.class);
+        Identifier identifier = Identifier.create("db", "tbl");
+        Table table = new Table();
+        table.setParameters(new HashMap<>());
+
+        HiveAlterTableUtils.alterTable(client, identifier, table, true, true);
+
+        EnvironmentContext context = captureEnvironmentContext(client, table);
+        assertThat(context.getProperties())
+                .containsEntry(StatsSetupConst.DO_NOT_UPDATE_STATS, "true")
+                .containsEntry(StatsSetupConst.CASCADE, "true");
+    }
+
+    private EnvironmentContext captureEnvironmentContext(IMetaStoreClient client, Table table)
+            throws Exception {
+        ArgumentCaptor<EnvironmentContext> captor =
+                ArgumentCaptor.forClass(EnvironmentContext.class);
+        verify(client)
+                .alter_table_with_environmentContext(
+                        eq("db"), eq("tbl"), same(table), captor.capture());
+        return captor.getValue();
+    }
+}