[core] Fix HashBucketAssigner load index too large error with refactor exception

paimon-common/src/main/java/org/apache/paimon/utils/Int2ShortHashMap.java

@@ -67,10 +67,18 @@ public void put(int key, short value) {
         }
 
         public Int2ShortHashMap build() {
-            Int2ShortHashMap map = new Int2ShortHashMap(keyList.size());
-            for (int i = 0; i < keyList.size(); i++) {
-                map.put(keyList.getInt(i), valueList.getShort(i));
+            Int2ShortHashMap map;
+            try {
+                map = new Int2ShortHashMap(keyList.size());
+                for (int i = 0; i < keyList.size(); i++) {
+                    map.put(keyList.getInt(i), valueList.getShort(i));
+                }
+            } catch (IllegalArgumentException e) {
+                throw new PaimonUtilsException(
+                        "capacity of Int2ShortOpenHashMap is too large, advise raise your parallelism in your Flink/Spark job",
+                        e);
             }
+
             return map;
         }
     }

paimon-common/src/main/java/org/apache/paimon/utils/PaimonUtilsException.java

@@ -0,0 +1,30 @@
+/*
+ * 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.utils;
+
+/**
+ * A {@code PaimonUtilsException} is thrown when occur exception during using with paimon utils,
+ * such as too large initial capacity when build Int2ShortHashMap and so on.
+ */
+public class PaimonUtilsException extends RuntimeException {
+
+    public PaimonUtilsException(String message, Throwable e) {
+        super(message, e);
+    }
+}

paimon-common/src/test/java/org/apache/paimon/utils/Int2ShortHashMapTest.java

@@ -25,6 +25,7 @@
 import java.util.Random;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 /** Test for {@link Int2ShortHashMap}. */
 public class Int2ShortHashMapTest {
@@ -53,4 +54,9 @@ public void testRandom() {
                     assertThat(map.get(k)).isEqualTo(v);
                 });
     }
+
+    @Test
+    public void testCapacity() {
+        assertThrows(IllegalArgumentException.class, () -> new Int2ShortHashMap(1073741824));
+    }
 }