[HUDI-7443] Fix decimal conversion with legacy bytes type

apache · Feb 26, 2024 · 9e6518c · 9e6518c
1 parent 5c444ff
commit 9e6518c
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 8 deletions.
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java
@@ -286,10 +286,14 @@ private Object serializePrimitive(PrimitiveObjectInspector fieldOI, Object struc
  throw new HoodieException("Unexpected Avro schema for Binary TypeInfo: " + schema.getType());
  }
  case DECIMAL:
- HiveDecimal dec = (HiveDecimal)fieldOI.getPrimitiveJavaObject(structFieldData);
- LogicalTypes.Decimal decimal = (LogicalTypes.Decimal)schema.getLogicalType();
+ HiveDecimal dec = (HiveDecimal) fieldOI.getPrimitiveJavaObject(structFieldData);
+ LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) schema.getLogicalType();
  BigDecimal bd = new BigDecimal(dec.toString()).setScale(decimal.getScale());
- return HoodieAvroUtils.DECIMAL_CONVERSION.toFixed(bd, schema, decimal);
+ if (schema.getType() == Schema.Type.BYTES) {
+ return HoodieAvroUtils.DECIMAL_CONVERSION.toBytes(bd, schema, decimal);
+ } else {
+ return HoodieAvroUtils.DECIMAL_CONVERSION.toFixed(bd, schema, decimal);
+ }
  case CHAR:
  HiveChar ch = (HiveChar)fieldOI.getPrimitiveJavaObject(structFieldData);
  return new Utf8(ch.getStrippedValue());

diff --git a/...hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java b/...hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
@@ -168,6 +168,9 @@ public static Writable avroToArrayWritable(Object value, Schema schema, boolean
  case STRING:
  return new Text(value.toString());
  case BYTES:
+ if (schema.getLogicalType() != null && schema.getLogicalType().getName().equals("decimal")) {
+ return toHiveDecimalWritable(((ByteBuffer) value).array(), schema);
+ }
  return new BytesWritable(((ByteBuffer) value).array());
  case INT:
  if (schema.getLogicalType() != null && schema.getLogicalType().getName().equals("date")) {
@@ -248,11 +251,7 @@ public static Writable avroToArrayWritable(Object value, Schema schema, boolean
  }
  case FIXED:
  if (schema.getLogicalType() != null && schema.getLogicalType().getName().equals("decimal")) {
- LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) LogicalTypes.fromSchema(schema);
- HiveDecimalWritable writable = new HiveDecimalWritable(((GenericFixed) value).bytes(),
- decimal.getScale());
- return HiveDecimalUtils.enforcePrecisionScale(writable,
- new DecimalTypeInfo(decimal.getPrecision(), decimal.getScale()));
+ return toHiveDecimalWritable(((GenericFixed) value).bytes(), schema);
  }
  return new BytesWritable(((GenericFixed) value).bytes());
  default:
@@ -319,4 +318,11 @@ private static Schema appendNullSchemaFields(Schema schema, List<String> newFiel
  }
  return appendFieldsToSchema(schema, newFields);
  }
+
+ private static HiveDecimalWritable toHiveDecimalWritable(byte[] bytes, Schema schema) {
+ LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) LogicalTypes.fromSchema(schema);
+ HiveDecimalWritable writable = new HiveDecimalWritable(bytes, decimal.getScale());
+ return HiveDecimalUtils.enforcePrecisionScale(writable,
+ new DecimalTypeInfo(decimal.getPrecision(), decimal.getScale()));
+ }
 }