Enabling avroParquet to read Int96 as bytes

pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetUtils.java

@@ -27,6 +27,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.parquet.avro.AvroParquetReader;
 import org.apache.parquet.avro.AvroParquetWriter;
+import org.apache.parquet.avro.AvroReadSupport;
 import org.apache.parquet.avro.AvroSchemaConverter;
 import org.apache.parquet.format.converter.ParquetMetadataConverter;
 import org.apache.parquet.hadoop.ParquetFileReader;
@@ -100,6 +101,8 @@ public static Configuration getParquetHadoopConfiguration() {
     // in case that user's hadoop conf overwrite this item
     Configuration conf = new Configuration();
     conf.set("fs.defaultFS", DEFAULT_FS);
+    // To read Int96 as bytes.
+    conf.set(AvroReadSupport.READ_INT96_AS_FIXED, "true");
     conf.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
     return conf;
   }

pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetRecordReaderTest.java

@@ -125,19 +125,28 @@ public void testFileMetadataParsing()
   @Test
   public void testComparison()
       throws IOException {
-    testComparison(_dataFile, SAMPLE_RECORDS_SIZE);
-    testComparison(new File(getClass().getClassLoader().getResource("users.parquet").getFile()), 1);
-    testComparison(new File(getClass().getClassLoader().getResource("test-comparison.gz.parquet").getFile()), 363667);
-    testComparison(new File(getClass().getClassLoader().getResource("test-comparison.snappy.parquet").getFile()), 2870);
-    testComparison(new File(getClass().getClassLoader().getResource("baseballStats.snappy.parquet").getFile()), 97889);
-    testComparison(new File(getClass().getClassLoader().getResource("baseballStats.zstd.parquet").getFile()), 97889);
-    testComparison(new File(getClass().getClassLoader().getResource("githubEvents.snappy.parquet").getFile()), 10000);
-    testComparison(new File(getClass().getClassLoader().getResource("starbucksStores.snappy.parquet").getFile()), 6443);
-    testComparison(new File(getClass().getClassLoader().getResource("airlineStats.snappy.parquet").getFile()), 19492);
-    testComparison(new File(getClass().getClassLoader().getResource("githubActivities.gz.parquet").getFile()), 2000);
+    testComparison(_dataFile, SAMPLE_RECORDS_SIZE, false);
+    testComparison(new File(getClass().getClassLoader().getResource("users.parquet").getFile()), 1, false);
+    testComparison(new File(getClass().getClassLoader().getResource("test-comparison.gz.parquet").getFile()), 363667,
+        false);
+    testComparison(new File(getClass().getClassLoader().getResource("test-comparison.snappy.parquet").getFile()), 2870,
+        false);
+    testComparison(new File(getClass().getClassLoader().getResource("baseballStats.snappy.parquet").getFile()), 97889,
+        false);
+    testComparison(new File(getClass().getClassLoader().getResource("baseballStats.zstd.parquet").getFile()), 97889,
+        false);
+    testComparison(new File(getClass().getClassLoader().getResource("githubEvents.snappy.parquet").getFile()), 10000,
+        false);
+    testComparison(new File(getClass().getClassLoader().getResource("starbucksStores.snappy.parquet").getFile()), 6443,
+        false);
+    testComparison(new File(getClass().getClassLoader().getResource("airlineStats.snappy.parquet").getFile()), 19492,
+        false);
+    testComparison(new File(getClass().getClassLoader().getResource("githubActivities.gz.parquet").getFile()), 2000,
+        false);
+    testComparison(new File(getClass().getClassLoader().getResource("int96AvroParquet.parquet").getFile()), 1, true);
   }
 
-  private void testComparison(File dataFile, int totalRecords)
+  private void testComparison(File dataFile, int totalRecords, boolean skipIndividualRecordComparison)
       throws IOException {
     final ParquetRecordReader avroRecordReader = new ParquetRecordReader();
     ParquetRecordReaderConfig avroRecordReaderConfig = new ParquetRecordReaderConfig();
@@ -150,14 +159,14 @@ private void testComparison(File dataFile, int totalRecords)
     Assert.assertTrue(avroRecordReader.useAvroParquetRecordReader());
     Assert.assertFalse(nativeRecordReader.useAvroParquetRecordReader());
 
-    testComparison(avroRecordReader, nativeRecordReader, totalRecords);
+    testComparison(avroRecordReader, nativeRecordReader, totalRecords, skipIndividualRecordComparison);
     avroRecordReader.rewind();
     nativeRecordReader.rewind();
-    testComparison(avroRecordReader, nativeRecordReader, totalRecords);
+    testComparison(avroRecordReader, nativeRecordReader, totalRecords, skipIndividualRecordComparison);
   }
 
   private void testComparison(ParquetRecordReader avroRecordReader, ParquetRecordReader nativeRecordReader,
-      int totalRecords)
+      int totalRecords, boolean skipIndividualRecordComparison)
       throws IOException {
     GenericRow avroReuse = new GenericRow();
     GenericRow nativeReuse = new GenericRow();
@@ -166,8 +175,10 @@ private void testComparison(ParquetRecordReader avroRecordReader, ParquetRecordR
       Assert.assertTrue(nativeRecordReader.hasNext());
       final GenericRow avroReaderRow = avroRecordReader.next(avroReuse);
       final GenericRow nativeReaderRow = nativeRecordReader.next(nativeReuse);
-      Assert.assertEquals(nativeReaderRow.toString(), avroReaderRow.toString());
-      Assert.assertTrue(avroReaderRow.equals(nativeReaderRow));
+      if (!skipIndividualRecordComparison) {
+        Assert.assertEquals(nativeReaderRow.toString(), avroReaderRow.toString());
+        Assert.assertTrue(avroReaderRow.equals(nativeReaderRow));
+      }
       recordsRead++;
     }
     Assert.assertFalse(nativeRecordReader.hasNext());