review(common): fold Avro schema cache into HoodieSchema.fromAvroSchema, drop AvroToHoodieSchemaCache

Author: voonhous

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/format/FlinkRecordContext.java

@@ -25,7 +25,6 @@
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieOperation;
 import org.apache.hudi.common.model.HoodieRecord;
-import org.apache.hudi.common.schema.AvroToHoodieSchemaCache;
 import org.apache.hudi.common.schema.HoodieSchema;
 import org.apache.hudi.common.schema.HoodieSchemaField;
 import org.apache.hudi.common.table.HoodieTableConfig;
@@ -126,7 +125,7 @@ public RowData getDeleteRow(String recordKey) {
   @Override
   public RowData convertAvroRecord(IndexedRecord avroRecord) {
     Schema recordSchema = avroRecord.getSchema();
-    AvroToRowDataConverters.AvroToRowDataConverter converter = RowDataQueryContexts.fromSchema(AvroToHoodieSchemaCache.intern(recordSchema), utcTimezone).getAvroToRowDataConverter();
+    AvroToRowDataConverters.AvroToRowDataConverter converter = RowDataQueryContexts.fromSchema(HoodieSchema.fromAvroSchema(recordSchema), utcTimezone).getAvroToRowDataConverter();
     RowData rowData = (RowData) converter.convert(avroRecord);
     Schema.Field operationField = recordSchema.getField(HoodieRecord.OPERATION_METADATA_FIELD);
     if (operationField != null) {

hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRecordContext.scala

@@ -21,7 +21,7 @@ package org.apache.hudi
 
 import org.apache.avro.generic.{GenericRecord, IndexedRecord}
 import org.apache.hudi.common.engine.RecordContext
-import org.apache.hudi.common.schema.{AvroToHoodieSchemaCache, HoodieSchema}
+import org.apache.hudi.common.schema.HoodieSchema
 import org.apache.hudi.common.table.HoodieTableConfig
 import org.apache.spark.sql.HoodieInternalRowUtils
 import org.apache.spark.sql.avro.{HoodieAvroDeserializer, HoodieAvroSerializer}
@@ -47,7 +47,7 @@ trait SparkFileFormatInternalRecordContext extends BaseSparkInternalRecordContex
    * @return An [[InternalRow]].
    */
   override def convertAvroRecord(avroRecord: IndexedRecord): InternalRow = {
-    val schema = AvroToHoodieSchemaCache.intern(avroRecord.getSchema)
+    val schema = HoodieSchema.fromAvroSchema(avroRecord.getSchema)
     val structType = HoodieInternalRowUtils.getCachedSchema(schema)
     val deserializer = deserializerMap.getOrElseUpdate(schema, {
       sparkAdapter.createAvroDeserializer(schema, structType)

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

@@ -24,7 +24,6 @@
 import org.apache.hudi.common.model.HoodieEmptyRecord;
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieRecord;
-import org.apache.hudi.common.schema.AvroToHoodieSchemaCache;
 import org.apache.hudi.common.schema.HoodieSchema;
 import org.apache.hudi.common.schema.HoodieSchemaField;
 import org.apache.hudi.common.table.HoodieTableConfig;
@@ -71,10 +70,7 @@ public AvroRecordContext() {
   public static Object getFieldValueFromIndexedRecord(
       IndexedRecord record,
       String fieldName) {
-    // Interning returns the canonical wrapper for this schema, whose lazily built field list and
-    // field map survive across calls, so the per-record cost is a cache hit instead of an
-    // O(schema width) wrapper rebuild.
-    HoodieSchema currentSchema = AvroToHoodieSchemaCache.intern(record.getSchema());
+    HoodieSchema currentSchema = HoodieSchema.fromAvroSchema(record.getSchema());
     IndexedRecord currentRecord = record;
     String[] path = fieldName.split("\\.");
     for (int i = 0; i < path.length; i++) {

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

@@ -19,7 +19,6 @@
 package org.apache.hudi.avro;
 
 import org.apache.hudi.common.model.HoodieRecord;
-import org.apache.hudi.common.schema.AvroToHoodieSchemaCache;
 import org.apache.hudi.common.schema.HoodieSchema;
 import org.apache.hudi.common.schema.HoodieSchemaUtils;
 import org.apache.hudi.common.util.DateTimeUtils;
@@ -835,7 +834,7 @@ public static Object[] getRecordColumnValues(HoodieRecord record,
                                                Schema schema,
                                                boolean consistentLogicalTimestampEnabled) {
     try {
-      GenericRecord genericRecord = (GenericRecord) (record.toIndexedRecord(AvroToHoodieSchemaCache.intern(schema), new Properties()).get()).getData();
+      GenericRecord genericRecord = (GenericRecord) (record.toIndexedRecord(HoodieSchema.fromAvroSchema(schema), new Properties()).get()).getData();
       List<Object> list = new ArrayList<>();
       for (String col : columns) {
         list.add(HoodieAvroUtils.getNestedFieldVal(genericRecord, col, true, consistentLogicalTimestampEnabled));

hudi-common/src/main/java/org/apache/hudi/common/HoodieJsonPayload.java

@@ -20,7 +20,7 @@
 
 import org.apache.hudi.avro.MercifulJsonConverter;
 import org.apache.hudi.common.model.HoodieRecordPayload;
-import org.apache.hudi.common.schema.AvroToHoodieSchemaCache;
+import org.apache.hudi.common.schema.HoodieSchema;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.io.util.FileIOUtils;
@@ -65,7 +65,7 @@ public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord oldRec, Sche
   @Override
   public Option<IndexedRecord> getInsertValue(Schema schema) throws IOException {
     MercifulJsonConverter jsonConverter = new MercifulJsonConverter();
-    return Option.of(jsonConverter.convert(getJsonData(), AvroToHoodieSchemaCache.intern(schema)));
+    return Option.of(jsonConverter.convert(getJsonData(), HoodieSchema.fromAvroSchema(schema)));
   }
 
   private String getJsonData() throws IOException {

hudi-common/src/main/java/org/apache/hudi/common/schema/AvroToHoodieSchemaCache.java

@@ -26,6 +26,8 @@
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.internal.schema.HoodieSchemaException;
 
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.LoadingCache;
 import lombok.Getter;
 import org.apache.avro.JsonProperties;
 import org.apache.avro.LogicalType;
@@ -350,16 +352,32 @@ private HoodieSchema(Schema avroSchema, List<HoodieSchemaField> fields) {
     this.fields = fields != null ? Collections.unmodifiableList(fields) : null;
   }
 
+  // Avro schemas are interned by identity (records of one file share the same Schema instance), so the
+  // per-record fromAvroSchema() call is a cache hit that reuses the canonical HoodieSchema and its lazily
+  // built field list / field map instead of rebuilding an O(schema width) wrapper. Misses convert and
+  // value-intern through HoodieSchemaCache so equal-but-distinct Avro schema instances still converge on
+  // one canonical HoodieSchema. Global cache for the JVM lifecycle; weakKeys lets dead schemas be GC'd.
+  private static final LoadingCache<Schema, HoodieSchema> AVRO_SCHEMA_CACHE =
+      Caffeine.newBuilder().weakKeys().maximumSize(1024)
+          .build(avroSchema -> HoodieSchemaCache.intern(convertFromAvroSchema(avroSchema)));
+
   /**
-   * Factory method to create HoodieSchema from an Avro schema.
+   * Factory method to create a {@link HoodieSchema} from an Avro schema.
+   *
+   * <p>The result is interned: passing the same Avro {@link Schema} instance (e.g. once per record)
+   * returns the canonical {@link HoodieSchema} rather than rebuilding a fresh wrapper each call.
    *
    * @param avroSchema the Avro schema to wrap
-   * @return new HoodieSchema instance
+   * @return canonical HoodieSchema instance, or {@code null} if {@code avroSchema} is null
    */
   public static HoodieSchema fromAvroSchema(Schema avroSchema) {
     if (avroSchema == null) {
       return null;
     }
+    return AVRO_SCHEMA_CACHE.get(avroSchema);
+  }
+
+  private static HoodieSchema convertFromAvroSchema(Schema avroSchema) {
     LogicalType logicalType = avroSchema.getLogicalType();
     if (logicalType != null) {
       if (logicalType instanceof LogicalTypes.Decimal) {

hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java

@@ -27,7 +27,6 @@
 import org.apache.hudi.common.model.HoodieAvroRecordMerger;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordMerger;
-import org.apache.hudi.common.schema.AvroToHoodieSchemaCache;
 import org.apache.hudi.common.schema.HoodieSchema;
 import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
 import org.apache.hudi.common.table.read.BufferedRecord;
@@ -205,8 +204,8 @@ private Option<HoodieAvroIndexedRecord> mergeRecord(HoodieRecord<?> newRecord, A
     // once presto on hudi have its own mor reader, we can remove the rewrite logical.
     GenericRecord genericRecord = HiveAvroSerializer.rewriteRecordIgnoreResultCheck(oldRecord, getLogScannerReaderSchema());
     RecordContext<IndexedRecord> recordContext = AvroRecordContext.getFieldAccessorInstance();
-    BufferedRecord record = BufferedRecords.fromEngineRecord(genericRecord, AvroToHoodieSchemaCache.intern(genericRecord.getSchema()), recordContext, orderingFields, newRecord.getRecordKey(), false);
-    BufferedRecord newBufferedRecord = BufferedRecords.fromHoodieRecord(newRecord, AvroToHoodieSchemaCache.intern(getLogScannerReaderSchema().toAvroSchema()),
+    BufferedRecord record = BufferedRecords.fromEngineRecord(genericRecord, HoodieSchema.fromAvroSchema(genericRecord.getSchema()), recordContext, orderingFields, newRecord.getRecordKey(), false);
+    BufferedRecord newBufferedRecord = BufferedRecords.fromHoodieRecord(newRecord, HoodieSchema.fromAvroSchema(getLogScannerReaderSchema().toAvroSchema()),
         recordContext, payloadProps, orderingFields, deleteContext);
     BufferedRecord mergeResult = merger.merge(record, newBufferedRecord, recordContext, payloadProps);
     if (mergeResult.isDelete()) {

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeCompactedRecordReader.java

@@ -22,7 +22,7 @@ import org.apache.hudi.HoodieSchemaConversionUtils.{convertHoodieSchemaToDataTyp
 import org.apache.hudi.SparkAdapterSupport.sparkAdapter
 import org.apache.hudi.avro.HoodieAvroUtils
 import org.apache.hudi.common.model.{DefaultHoodieRecordPayload, HoodiePayloadProps, HoodieRecord, HoodieRecordPayload, OverwriteWithLatestAvroPayload}
-import org.apache.hudi.common.schema.{AvroToHoodieSchemaCache, HoodieSchema, HoodieSchemaUtils}
+import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaUtils}
 import org.apache.hudi.common.util.{BinaryUtil, ConfigUtils, HoodieRecordUtils, Option => HOption, OrderingValues, StringUtils, ValidationUtils}
 import org.apache.hudi.common.util.ValidationUtils.checkState
 import org.apache.hudi.config.HoodieWriteConfig
@@ -116,7 +116,7 @@ class ExpressionPayload(@transient record: GenericRecord,
 
     // Get the Evaluator for each condition and update assignments.
     val updateConditionAndAssignments =
-      getEvaluator(updateConditionAndAssignmentsText.toString, AvroToHoodieSchemaCache.intern(inputRecord.asAvro.getSchema))
+      getEvaluator(updateConditionAndAssignmentsText.toString, HoodieSchema.fromAvroSchema(inputRecord.asAvro.getSchema))
 
     for ((conditionEvaluator, assignmentEvaluator) <- updateConditionAndAssignments
          if resultRecordOpt == null) {
@@ -145,7 +145,7 @@ class ExpressionPayload(@transient record: GenericRecord,
       // Process delete
       val deleteConditionText = properties.get(ExpressionPayload.PAYLOAD_DELETE_CONDITION)
       if (deleteConditionText != null) {
-        val (deleteConditionEvaluator, _) = getEvaluator(deleteConditionText.toString, AvroToHoodieSchemaCache.intern(inputRecord.asAvro.getSchema)).head
+        val (deleteConditionEvaluator, _) = getEvaluator(deleteConditionText.toString, HoodieSchema.fromAvroSchema(inputRecord.asAvro.getSchema)).head
         val deleteConditionEvalResult = deleteConditionEvaluator.apply(inputRecord.asRow)
           .get(0, BooleanType)
           .asInstanceOf[Boolean]
@@ -206,7 +206,7 @@ class ExpressionPayload(@transient record: GenericRecord,
    *       multiple times for different expression evaluation invocations
    */
   case class ConvertibleRecord(private val avro: GenericRecord) extends Logging {
-    private lazy val row: InternalRow = getAvroDeserializerFor(AvroToHoodieSchemaCache.intern(avro.getSchema)).deserialize(avro) match {
+    private lazy val row: InternalRow = getAvroDeserializerFor(HoodieSchema.fromAvroSchema(avro.getSchema)).deserialize(avro) match {
       case Some(row) => row.asInstanceOf[InternalRow]
       case None =>
         logError(s"Failed to deserialize Avro record `${avro.toString}` as Catalyst row")
@@ -231,7 +231,7 @@ class ExpressionPayload(@transient record: GenericRecord,
       properties.get(ExpressionPayload.PAYLOAD_INSERT_CONDITION_AND_ASSIGNMENTS).toString
     // Get the evaluator for each condition and insert assignment.
     val insertConditionAndAssignments =
-      ExpressionPayload.getEvaluator(insertConditionAndAssignmentsText, AvroToHoodieSchemaCache.intern(inputRecord.asAvro.getSchema))
+      ExpressionPayload.getEvaluator(insertConditionAndAssignmentsText, HoodieSchema.fromAvroSchema(inputRecord.asAvro.getSchema))
     var resultRecordOpt: HOption[IndexedRecord] = null
     for ((conditionEvaluator, assignmentEvaluator) <- insertConditionAndAssignments
          if resultRecordOpt == null) {
@@ -243,7 +243,7 @@ class ExpressionPayload(@transient record: GenericRecord,
       if (conditionEvalResult) {
         val writerSchema = getWriterSchema(properties, false)
         val resultingRow = assignmentEvaluator.apply(inputRecord.asRow)
-        val resultingAvroRecord = getAvroSerializerFor(AvroToHoodieSchemaCache.intern(writerSchema.getAvroSchema))
+        val resultingAvroRecord = getAvroSerializerFor(HoodieSchema.fromAvroSchema(writerSchema.getAvroSchema))
           .serialize(resultingRow)
           .asInstanceOf[GenericRecord]
 
@@ -315,7 +315,7 @@ class ExpressionPayload(@transient record: GenericRecord,
    */
   private def joinRecord(sourceRecord: IndexedRecord, targetRecord: IndexedRecord, props: Properties): GenericRecord = {
     val leftSchema = sourceRecord.getSchema
-    val joinSchema = getMergedSchema(AvroToHoodieSchemaCache.intern(leftSchema), AvroToHoodieSchemaCache.intern(targetRecord.getSchema))
+    val joinSchema = getMergedSchema(HoodieSchema.fromAvroSchema(leftSchema), HoodieSchema.fromAvroSchema(targetRecord.getSchema))
 
     // TODO rebase onto JoinRecord
     val values = new Array[AnyRef](joinSchema.getFields.size())