Address comments

Author: voonhous

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java

@@ -69,7 +69,7 @@ private MessageType getWriteSchema(HoodieWriteConfig config, List<StoragePath> i
       // All files should have the same schema in this case
       try {
         ParquetUtils parquetUtils = new ParquetUtils();
-        MessageType fileSchema = parquetUtils.readSchema(table.getStorage(), inputFiles.get(0));
+        MessageType fileSchema = parquetUtils.readMessageType(table.getStorage(), inputFiles.get(0));
         log.info("Binary copy schema evolution disabled. Using schema from input file: " + inputFiles.get(0));
         return fileSchema;
       } catch (Exception e) {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java

@@ -263,7 +263,7 @@ public static <T> HoodieData<HoodieRecord> readSecondaryKeysFromFileSlices(Hoodi
       if (dataFilePath.isPresent()) {
         readerSchema = HoodieIOFactory.getIOFactory(metaClient.getStorage())
             .getFileFormatUtils(baseFileFormat)
-            .readHoodieSchema(metaClient.getStorage(), dataFilePath.get());
+            .readSchema(metaClient.getStorage(), dataFilePath.get());
       } else {
         readerSchema = tableSchema;
       }

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieBinaryCopyHandleSchemaEvolution.java

@@ -113,7 +113,7 @@ public void testSchemaEvolutionDisabled_UsesFileSchema() throws Exception {
     // Mock ParquetUtils to return file schema
     try (MockedConstruction<ParquetUtils> parquetUtilsConstruction = mockConstruction(ParquetUtils.class, 
         (mock, context) -> {
-          when(mock.readSchema(eq(storage), eq(inputFiles.get(0)))).thenReturn(fileSchema);
+          when(mock.readMessageType(eq(storage), eq(inputFiles.get(0)))).thenReturn(fileSchema);
         })) {
 
       // When: Creating HoodieBinaryCopyHandle (we can't instantiate directly due to complex dependencies,
@@ -201,7 +201,7 @@ public MessageType testGetWriteSchema(HoodieWriteConfig config, List<StoragePath
             throw new IOException("Simulated file read error");
           }
           ParquetUtils parquetUtils = new ParquetUtils();
-          MessageType fileSchema = parquetUtils.readSchema(table.getStorage(), inputFiles.get(0));
+          MessageType fileSchema = parquetUtils.readMessageType(table.getStorage(), inputFiles.get(0));
           return fileSchema;
         } catch (Exception e) {
           throw new HoodieIOException("Failed to read schema from input file when schema evolution is disabled: " + inputFiles.get(0),

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkParquetReader.java

@@ -189,7 +189,7 @@ public ClosableIterator<UnsafeRow> getUnsafeRowIterator(HoodieSchema requestedSc
 
   private MessageType getFileSchema() {
     if (fileSchemaOption.isEmpty()) {
-      MessageType messageType = ((ParquetUtils) parquetUtils).readSchema(storage, path);
+      MessageType messageType = ((ParquetUtils) parquetUtils).readMessageType(storage, path);
       fileSchemaOption = Option.of(messageType);
     }
     return fileSchemaOption.get();

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

@@ -659,7 +659,7 @@ public static Object convertValueForSpecificDataTypes(HoodieSchema fieldSchema,
    * @return HoodieSchema containing record key and partition path fields
    */
   public static HoodieSchema getRecordKeyPartitionPathSchema() {
-    List<HoodieSchemaField> toBeAddedFields = new ArrayList<>();
+    List<HoodieSchemaField> toBeAddedFields = new ArrayList<>(2);
 
     HoodieSchemaField recordKeyField =
         createNewSchemaField(HoodieRecord.RECORD_KEY_METADATA_FIELD, METADATA_FIELD_SCHEMA, "", JsonProperties.NULL_VALUE);

hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java

@@ -384,7 +384,7 @@ public Schema readSchemaFromLastCompaction(Option<HoodieInstant> lastCompactionC
             + lastCompactionCommit + ", could not get schema for table " + metaClient.getBasePath()));
     StoragePath path = new StoragePath(filePath);
     return HoodieIOFactory.getIOFactory(metaClient.getStorage())
-        .getFileFormatUtils(path).readHoodieSchema(metaClient.getStorage(), path).toAvroSchema();
+        .getFileFormatUtils(path).readSchema(metaClient.getStorage(), path).toAvroSchema();
   }
 
   private Schema readSchemaFromLogFile(StoragePath path) throws IOException {
@@ -568,7 +568,7 @@ private Schema fetchSchemaFromFiles(Stream<StoragePath> filePaths) {
           return readSchemaFromLogFile(filePath);
         } else {
           HoodieSchema hoodieSchema = HoodieIOFactory.getIOFactory(metaClient.getStorage())
-              .getFileFormatUtils(filePath).readHoodieSchema(metaClient.getStorage(), filePath);
+              .getFileFormatUtils(filePath).readSchema(metaClient.getStorage(), filePath);
           return hoodieSchema != null ? hoodieSchema.toAvroSchema() : null;
         }
       } catch (IOException e) {

hudi-common/src/main/java/org/apache/hudi/common/util/FileFormatUtils.java

@@ -288,7 +288,7 @@ protected HoodieSchema getKeyIteratorSchema(HoodieStorage storage, StoragePath f
           List<String> fields = new ArrayList<>();
           fields.addAll(keyGenerator.getRecordKeyFieldNames());
           fields.addAll(keyGenerator.getPartitionPathFields());
-          return HoodieSchemaUtils.projectSchema(readHoodieSchema(storage, filePath), fields);
+          return HoodieSchemaUtils.projectSchema(readSchema(storage, filePath), fields);
         })
         .orElse(partitionPath.isPresent() ? HoodieSchemaUtils.getRecordKeySchema() : HoodieSchemaUtils.getRecordKeyPartitionPathSchema());
   }
@@ -314,7 +314,7 @@ public abstract ClosableIterator<Pair<HoodieKey, Long>> fetchRecordKeysWithPosit
    * @param filePath the data file path.
    * @return the Avro schema of the data file.
    */
-  public abstract HoodieSchema readHoodieSchema(HoodieStorage storage, StoragePath filePath);
+  public abstract HoodieSchema readSchema(HoodieStorage storage, StoragePath filePath);
 
   /**
    * Reads column statistics stored in the metadata.

hudi-common/src/main/java/org/apache/hudi/common/util/HFileUtils.java

@@ -152,7 +152,7 @@ public ClosableIterator<Pair<HoodieKey, Long>> fetchRecordKeysWithPositions(Hood
   }
 
   @Override
-  public HoodieSchema readHoodieSchema(HoodieStorage storage, StoragePath filePath) {
+  public HoodieSchema readSchema(HoodieStorage storage, StoragePath filePath) {
     LOG.info("Reading schema from {}", filePath);
 
     try (HoodieFileReader fileReader =

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataParquetReader.java

@@ -116,7 +116,7 @@ public DataType getRowType() {
       // Some types in avro are not compatible with parquet.
       // Avro only supports representing Decimals as fixed byte array
       // and therefore if we convert to Avro directly we'll lose logical type-info.
-      MessageType messageType = parquetUtils.readSchema(storage, path);
+      MessageType messageType = parquetUtils.readMessageType(storage, path);
       RowType rowType = ParquetSchemaConverter.convertToRowType(messageType);
       fileRowType = DataTypes.of(rowType);
     }

hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/AvroOrcUtils.java

@@ -79,7 +79,7 @@ public class AvroOrcUtils {
    *
    * @param type        ORC schema of the value Object.
    * @param colVector   The column vector to store the value Object.
-   * @param schema  Avro schema of the value Object.
+   * @param schema      Schema of the value Object.
    *                    Only used to check logical types for timestamp unit conversion.
    * @param value       Object to be added to the column vector
    * @param vectorPos   The position in the vector where value will be stored at.
@@ -657,6 +657,15 @@ public static TypeDescription createOrcSchema(HoodieSchema schema) {
         }
       case TIMESTAMP:
         HoodieSchema.Timestamp timestampSchema = (HoodieSchema.Timestamp) schema;
+        // NOTE: Preserving old behavior from before HoodieSchema refactoring:
+        // - UTC-adjusted timestamps (TimestampMillis/Micros) are converted to ORC Timestamp
+        // - Local timestamps (LocalTimestampMillis/Micros) are converted to ORC Long
+        // This is because the old code did not handle LocalTimestamp logical types explicitly,
+        // causing them to fall through to the base type (LONG) conversion.
+        if (!timestampSchema.isUtcAdjusted()) {
+          // Local timestamp - treat as ORC Long (old behavior)
+          return TypeDescription.createLong();
+        }
         if (timestampSchema.getPrecision() == TimePrecision.MILLIS) {
           // The timestamp-millis logical type represents an instant on the global timeline, independent of a
           // particular time zone or calendar, with a precision of one millisecond.