@Override
  public RecordReader<NullWritable, ArrayWritable> getRecordReader(final InputSplit split, final JobConf job,
                                                                   final Reporter reporter) throws IOException {
    HoodieRealtimeInputFormatUtils.addProjectionField(job, job.get(hive_metastoreConstants.META_TABLE_PARTITION_COLUMNS, "").split("/"));
    if (shouldUseFilegroupReader(job)) {
      try {
        if (!(split instanceof FileSplit) || !checkIfHudiTable(split, job)) {
          return super.getRecordReader(split, job, reporter);
        }
        if (supportAvroRead && HoodieColumnProjectionUtils.supportTimestamp(job)) {
          return new HoodieFileGroupReaderBasedRecordReader((s, j, r) -> {
            try {
              return new ParquetRecordReaderWrapper(new HoodieTimestampAwareParquetInputFormat(), s, j, r);
            } catch (InterruptedException e) {
              throw new RuntimeException(e);
            }
          }, split, job, reporter);
        } else {
          return new HoodieFileGroupReaderBasedRecordReader(super::getRecordReader, split, job, reporter);
        }
      } catch (final IOException e) {
        throw new RuntimeException("Cannot create a RecordReaderWrapper", e);
      }
    }
    // TODO enable automatic predicate pushdown after fixing issues
    // FileSplit fileSplit = (FileSplit) split;
    // HoodieTableMetadata metadata = getTableMetadata(fileSplit.getPath().getParent());
    // String tableName = metadata.getTableName();
    // String mode = HoodieHiveUtil.readMode(job, tableName);

    // if (HoodieHiveUtil.INCREMENTAL_SCAN_MODE.equals(mode)) {
    // FilterPredicate predicate = constructHoodiePredicate(job, tableName, split);
    // LOG.info("Setting parquet predicate push down as " + predicate);
    // ParquetInputFormat.setFilterPredicate(job, predicate);
    // clearOutExistingPredicate(job);
    // }
    if (split instanceof BootstrapBaseFileSplit) {
      return createBootstrappingRecordReader(split, job, reporter);
    }

    // adapt schema evolution
    new SchemaEvolutionContext(split, job).doEvolutionForParquetFormat();

    if (LOG.isDebugEnabled()) {
      LOG.debug("EMPLOYING DEFAULT RECORD READER - " + split);
    }

    return getRecordReaderInternal(split, job, reporter);
  }
@Test
  public void testHoodieParquetInputFormatReadTimeType() throws IOException {
    try {
      long testTimestampLong = System.currentTimeMillis();
      int testDate = 19116;// 2022-05-04

      Schema schema = SchemaTestUtil.getSchemaFromResource(getClass(), "/test_timetype.avsc");
      String commit = "20160628071126";
      HoodieTestUtils.init(HoodieTestUtils.getDefaultStorageConf(), basePath.toString(),
          HoodieTableType.COPY_ON_WRITE, HoodieFileFormat.PARQUET);
      java.nio.file.Path partitionPath = basePath.resolve(Paths.get("2016", "06", "28"));
      String fileId = FSUtils.makeBaseFileName(commit, "1-0-1", "fileid1",
          HoodieFileFormat.PARQUET.getFileExtension());
      try (AvroParquetWriter parquetWriter = new AvroParquetWriter(
          new Path(partitionPath.resolve(fileId).toString()), schema)) {
        GenericData.Record record = new GenericData.Record(schema);
        record.put("test_timestamp", testTimestampLong * 1000);
        record.put("test_long", testTimestampLong * 1000);
        record.put("test_date", testDate);
        record.put("_hoodie_commit_time", commit);
        record.put("_hoodie_commit_seqno", commit + 1);
        parquetWriter.write(record);
      }

      //this is not a hoodie table!!
      jobConf.set(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key(), "false");
      jobConf.set(IOConstants.COLUMNS, "test_timestamp,test_long,test_date,_hoodie_commit_time,_hoodie_commit_seqno");
      jobConf.set(IOConstants.COLUMNS_TYPES, "timestamp,bigint,date,string,string");
      jobConf.set(READ_COLUMN_NAMES_CONF_STR, "test_timestamp,test_long,test_date,_hoodie_commit_time,_hoodie_commit_seqno");
      InputFormatTestUtil.setupPartition(basePath, partitionPath);
      InputFormatTestUtil.commit(basePath, commit);
      FileInputFormat.setInputPaths(jobConf, partitionPath.toFile().getPath());

      InputSplit[] splits = inputFormat.getSplits(jobConf, 1);
      for (InputSplit split : splits) {
        RecordReader<NullWritable, ArrayWritable> recordReader = inputFormat
            .getRecordReader(split, jobConf, null);
        NullWritable key = recordReader.createKey();
        ArrayWritable writable = recordReader.createValue();
        while (recordReader.next(key, writable)) {
          // test timestamp
          if (HiveVersionInfo.getShortVersion().startsWith("3")) {
            LocalDateTime localDateTime = LocalDateTime.ofInstant(
                Instant.ofEpochMilli(testTimestampLong), ZoneOffset.UTC);
            assertEquals(Timestamp.valueOf(localDateTime).toString(), String.valueOf(writable.get()[0]));
          } else {
            Date date = new Date();
            date.setTime(testTimestampLong);
            Timestamp actualTime = ((TimestampWritable) writable.get()[0]).getTimestamp();
            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
            assertEquals(dateFormat.format(date), dateFormat.format(actualTime));
          }
          // test long
          assertEquals(testTimestampLong * 1000, ((LongWritable) writable.get()[1]).get());
          // test date
          assertEquals(LocalDate.ofEpochDay(testDate).toString(), String.valueOf(writable.get()[2]));
        }
      }
    } finally {
      jobConf.set(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key(), "true");
    }
  }