[HUDI-5829] Optimize conversion from json to row format when sanitizing field names

hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java

@@ -304,25 +304,12 @@ public Pair<Boolean, Object> convert(Object value, String name, Schema schema) {
 
       // Case 2: Input is a number or String number.
       LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
-      Pair<Boolean, BigDecimal> parseResult = parseObjectToBigDecimal(value);
+      Pair<Boolean, BigDecimal> parseResult = parseObjectToBigDecimal(value, schema);
       if (Boolean.FALSE.equals(parseResult.getLeft())) {
         return Pair.of(false, null);
       }
       BigDecimal bigDecimal = parseResult.getRight();
 
-      // As we don't do rounding, the validation will enforce the scale part and the integer part are all within the
-      // limit. As a result, if scale is 2 precision is 5, we only allow 3 digits for the integer.
-      // Allowed: 123.45, 123, 0.12
-      // Disallowed: 1234 (4 digit integer while the scale has already reserved 2 digit out of the 5 digit precision)
-      //             123456, 0.12345
-      if (bigDecimal.scale() > decimalType.getScale()
-          || (bigDecimal.precision() - bigDecimal.scale()) > (decimalType.getPrecision() - decimalType.getScale())) {
-        // Correspond to case
-        // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 5 as scale 2 without rounding.
-        // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 3 as scale 2 without rounding
-        return Pair.of(false, null);
-      }
-
       switch (schema.getType()) {
         case BYTES:
           // Convert to primitive Arvo type that logical type Decimal uses.

hudi-common/src/main/java/org/apache/hudi/avro/processors/DecimalLogicalTypeProcessor.java

@@ -49,21 +49,37 @@ protected static boolean isValidDecimalTypeConfig(Schema schema) {
    * @return Pair object, with left as boolean indicating if the parsing was successful and right as the
    * BigDecimal value.
    */
-  protected static Pair<Boolean, BigDecimal> parseObjectToBigDecimal(Object obj) {
+  protected static Pair<Boolean, BigDecimal> parseObjectToBigDecimal(Object obj, Schema schema) {
+    BigDecimal bigDecimal = null;
     if (obj instanceof Number) {
-      return Pair.of(true, BigDecimal.valueOf(((Number) obj).doubleValue()));
+      bigDecimal = BigDecimal.valueOf(((Number) obj).doubleValue());
     }
 
     // Case 2: Object is a number in String format.
     if (obj instanceof String) {
-      BigDecimal bigDecimal = null;
       try {
         bigDecimal = new BigDecimal(((String) obj));
       } catch (java.lang.NumberFormatException ignored) {
         /* ignore */
       }
-      return Pair.of(bigDecimal != null, bigDecimal);
     }
-    return Pair.of(false, null);
+
+    if (bigDecimal == null) {
+      return Pair.of(false, null);
+    }
+    // As we don't do rounding, the validation will enforce the scale part and the integer part are all within the
+    // limit. As a result, if scale is 2 precision is 5, we only allow 3 digits for the integer.
+    // Allowed: 123.45, 123, 0.12
+    // Disallowed: 1234 (4 digit integer while the scale has already reserved 2 digit out of the 5 digit precision)
+    //             123456, 0.12345
+    LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+    if (bigDecimal.scale() > decimalType.getScale()
+        || (bigDecimal.precision() - bigDecimal.scale()) > (decimalType.getPrecision() - decimalType.getScale())) {
+      // Correspond to case
+      // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 5 as scale 2 without rounding.
+      // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 3 as scale 2 without rounding
+      return Pair.of(false, null);
+    }
+    return Pair.of(true, bigDecimal);
   }
 }

hudi-utilities/src/main/java/org/apache/hudi/utilities/exception/HoodieJsonToRowConversionException.java

@@ -0,0 +1,32 @@
+/*
+ * 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.hudi.utilities.exception;
+
+import org.apache.hudi.exception.HoodieJsonConversionException;
+
+public class HoodieJsonToRowConversionException extends HoodieJsonConversionException {
+
+  public HoodieJsonToRowConversionException(String msg) {
+    super(msg);
+  }
+
+  public HoodieJsonToRowConversionException(String msg, Throwable t) {
+    super(msg, t);
+  }
+}