[HUDI-7173] Fix hudi-on-flink read issues involving schema evolution …

…and decimal types (#10247)

hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java

@@ -20,9 +20,9 @@
 
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
+import org.apache.hudi.table.format.cow.vector.HeapDecimalVector;
 import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector;
 import org.apache.hudi.table.format.cow.vector.HeapRowColumnVector;
-import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector;
 import org.apache.hudi.table.format.cow.vector.reader.ArrayColumnReader;
 import org.apache.hudi.table.format.cow.vector.reader.EmptyColumnReader;
 import org.apache.hudi.table.format.cow.vector.reader.FixedLenBytesColumnReader;
@@ -65,7 +65,6 @@
 import org.apache.flink.table.types.logical.MapType;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.table.types.logical.TimestampType;
-import org.apache.flink.table.types.logical.VarBinaryType;
 import org.apache.flink.util.Preconditions;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.parquet.ParquetRuntimeException;
@@ -234,17 +233,18 @@ private static ColumnVector createVectorFromConstant(
         }
         return lv;
       case DECIMAL:
-        DecimalType decimalType = (DecimalType) type;
-        int precision = decimalType.getPrecision();
-        int scale = decimalType.getScale();
-        DecimalData decimal = value == null
-            ? null
-            : Preconditions.checkNotNull(DecimalData.fromBigDecimal((BigDecimal) value, precision, scale));
-        ColumnVector internalVector = createVectorFromConstant(
-            new VarBinaryType(),
-            decimal == null ? null : decimal.toUnscaledBytes(),
-            batchSize);
-        return new ParquetDecimalVector(internalVector);
+        HeapDecimalVector decv = new HeapDecimalVector(batchSize);
+        if (value == null) {
+          decv.fillWithNulls();
+        } else {
+          DecimalType decimalType = (DecimalType) type;
+          int precision = decimalType.getPrecision();
+          int scale = decimalType.getScale();
+          DecimalData decimal = Preconditions.checkNotNull(
+              DecimalData.fromBigDecimal((BigDecimal) value, precision, scale));
+          decv.fill(decimal.toUnscaledBytes());
+        }
+        return decv;
       case FLOAT:
         HeapFloatVector fv = new HeapFloatVector(batchSize);
         if (value == null) {
@@ -513,7 +513,7 @@ private static WritableColumnVector createWritableColumnVector(
                 || typeName == PrimitiveType.PrimitiveTypeName.BINARY)
                 && primitiveType.getOriginalType() == OriginalType.DECIMAL,
             "Unexpected type: %s", typeName);
-        return new HeapBytesVector(batchSize);
+        return new HeapDecimalVector(batchSize);
       case ARRAY:
         ArrayType arrayType = (ArrayType) fieldType;
         return new HeapArrayVector(

hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapDecimalVector.java

@@ -0,0 +1,39 @@
+/*
+ * 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.table.format.cow.vector;
+
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.columnar.vector.DecimalColumnVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapBytesVector;
+
+/**
+ * This class represents a nullable heap map decimal vector.
+ */
+public class HeapDecimalVector extends HeapBytesVector implements DecimalColumnVector {
+
+  public HeapDecimalVector(int len) {
+    super(len);
+  }
+
+  @Override
+  public DecimalData getDecimal(int i, int precision, int scale) {
+    return DecimalData.fromUnscaledBytes(
+        this.getBytes(i).getBytes(), precision, scale);
+  }
+}