Move Parquet related classes to the new package

pinot-common/pom.xml

@@ -313,11 +313,6 @@
       <groupId>org.glassfish.jersey.core</groupId>
       <artifactId>jersey-server</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.apache.parquet</groupId>
-      <artifactId>parquet-avro</artifactId>
-      <version>${parquet.version}</version>
-    </dependency>
   </dependencies>
   <profiles>
     <profile>

pinot-core/pom.xml

@@ -63,18 +63,6 @@
     <dependency>
       <groupId>org.apache.pinot</groupId>
       <artifactId>pinot-common</artifactId>
-      <version>${project.version}</version>
-      <type>test-jar</type>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.pinot</groupId>
-      <artifactId>pinot-common</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.testng</groupId>
-      <artifactId>testng</artifactId>
-      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>joda-time</groupId>
@@ -154,6 +142,10 @@
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-csv</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-math</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.clearspring.analytics</groupId>
       <artifactId>stream</artifactId>
@@ -199,24 +191,26 @@
 
     <!-- test -->
     <dependency>
-      <groupId>org.mockito</groupId>
-      <artifactId>mockito-core</artifactId>
+      <groupId>org.apache.pinot</groupId>
+      <artifactId>pinot-common</artifactId>
+      <version>${project.version}</version>
+      <type>test-jar</type>
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>nl.jqno.equalsverifier</groupId>
-      <artifactId>equalsverifier</artifactId>
+      <groupId>org.testng</groupId>
+      <artifactId>testng</artifactId>
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-client</artifactId>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
       <scope>test</scope>
     </dependency>
-
     <dependency>
-      <groupId>org.apache.commons</groupId>
-      <artifactId>commons-math</artifactId>
+      <groupId>nl.jqno.equalsverifier</groupId>
+      <artifactId>equalsverifier</artifactId>
+      <scope>test</scope>
     </dependency>
   </dependencies>
 </project>

pinot-core/src/main/java/org/apache/pinot/core/data/readers/AvroRecordReader.java

@@ -28,16 +28,12 @@
 import org.apache.pinot.core.data.GenericRow;
 import org.apache.pinot.core.indexsegment.generator.SegmentGeneratorConfig;
 import org.apache.pinot.core.util.AvroUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 
 /**
  * Record reader for AVRO file.
  */
 public class AvroRecordReader implements RecordReader {
-  private static final Logger LOGGER = LoggerFactory.getLogger(AvroRecordReader.class);
-
   private final File _dataFile;
   private final Schema _schema;
   private final List<FieldSpec> _fieldSpecs;

pinot-core/src/main/java/org/apache/pinot/core/data/readers/FileFormat.java

@@ -19,5 +19,5 @@
 package org.apache.pinot.core.data.readers;
 
 public enum FileFormat {
-  AVRO, GZIPPED_AVRO, CSV, JSON, PINOT, THRIFT, PARQUET, OTHER
+  AVRO, GZIPPED_AVRO, CSV, JSON, PINOT, THRIFT, OTHER
 }

pinot-core/src/main/java/org/apache/pinot/core/data/readers/RecordReader.java

@@ -36,7 +36,8 @@ public interface RecordReader extends Closeable {
   /**
    * Initializes the record reader when needed
    */
-  void init(SegmentGeneratorConfig segmentGeneratorConfig);
+  void init(SegmentGeneratorConfig segmentGeneratorConfig)
+      throws IOException;
 
   /**
    * Return <code>true</code> if more records remain to be read.

pinot-core/src/main/java/org/apache/pinot/core/data/readers/RecordReaderFactory.java

@@ -46,10 +46,10 @@ public static RecordReader getRecordReader(SegmentGeneratorConfig segmentGenerat
     // If this is set, this will override the file format
     if (recordReaderPath != null) {
       if (fileFormat != FileFormat.OTHER) {
-        // We currently have default file format set to AVRO inside segment generator config,
-        // do not want to break this behavior for clients.
-        LOGGER.warn("Using recordReaderPath {} to read segment, ignoring fileformat {}",
-            recordReaderPath, fileFormat.toString());
+        // NOTE: we currently have default file format set to AVRO inside segment generator config, do not want to break
+        // this behavior for clients.
+        LOGGER
+            .warn("Using class: {} to read segment, ignoring configured file format: {}", recordReaderPath, fileFormat);
       }
       RecordReader recordReader = (RecordReader) Class.forName(recordReaderPath).newInstance();
       recordReader.init(segmentGeneratorConfig);
@@ -70,8 +70,6 @@ public static RecordReader getRecordReader(SegmentGeneratorConfig segmentGenerat
       // NOTE: PinotSegmentRecordReader does not support time conversion (field spec must match)
       case PINOT:
         return new PinotSegmentRecordReader(dataFile, schema, segmentGeneratorConfig.getColumnSortOrder());
-      case PARQUET:
-        return new ParquetRecordReader(dataFile, schema);
       default:
         throw new UnsupportedOperationException("Unsupported input file format: " + fileFormat);
     }

pinot-core/src/main/java/org/apache/pinot/core/util/AvroUtils.java

@@ -22,7 +22,6 @@
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
-import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -32,15 +31,13 @@
 import org.apache.avro.Schema.Field;
 import org.apache.avro.SchemaBuilder;
 import org.apache.avro.file.DataFileStream;
-import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.pinot.common.data.DimensionFieldSpec;
 import org.apache.pinot.common.data.FieldSpec;
 import org.apache.pinot.common.data.MetricFieldSpec;
 import org.apache.pinot.common.data.Schema;
 import org.apache.pinot.common.data.TimeFieldSpec;
-import org.apache.pinot.core.data.GenericRow;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,7 +55,6 @@ private AvroUtils() {
    * @param fieldTypeMap Map from column to field type
    * @param timeUnit Time unit
    * @return Pinot schema
-   * @throws IOException
    */
   public static Schema getPinotSchemaFromAvroSchema(@Nonnull org.apache.avro.Schema avroSchema,
       @Nullable Map<String, FieldSpec.FieldType> fieldTypeMap, @Nullable TimeUnit timeUnit) {
@@ -103,7 +99,6 @@ public static Schema getPinotSchemaFromAvroSchema(@Nonnull org.apache.avro.Schem
    * @param fieldTypeMap Map from column to field type
    * @param timeUnit Time unit
    * @return Pinot schema
-   * @throws IOException
    */
   public static Schema getPinotSchemaFromAvroDataFile(@Nonnull File avroDataFile,
       @Nullable Map<String, FieldSpec.FieldType> fieldTypeMap, @Nullable TimeUnit timeUnit)
@@ -120,7 +115,6 @@ public static Schema getPinotSchemaFromAvroDataFile(@Nonnull File avroDataFile,
    *
    * @param avroDataFile Avro data file
    * @return Pinot schema
-   * @throws IOException
    */
   public static Schema getPinotSchemaFromAvroDataFile(@Nonnull File avroDataFile)
       throws IOException {
@@ -134,7 +128,6 @@ public static Schema getPinotSchemaFromAvroDataFile(@Nonnull File avroDataFile)
    * @param fieldTypeMap Map from column to field type
    * @param timeUnit Time unit
    * @return Pinot schema
-   * @throws IOException
    */
   public static Schema getPinotSchemaFromAvroSchemaFile(@Nonnull File avroSchemaFile,
       @Nullable Map<String, FieldSpec.FieldType> fieldTypeMap, @Nullable TimeUnit timeUnit)
@@ -203,16 +196,44 @@ public static org.apache.avro.Schema getAvroSchemaFromPinotSchema(Schema pinotSc
     return fieldAssembler.endRecord();
   }
 
+  /**
+   * Validates Pinot schema against the given Avro schema.
+   */
+  public static void validateSchema(Schema schema, org.apache.avro.Schema avroSchema) {
+    for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) {
+      String fieldName = fieldSpec.getName();
+      Field avroField = avroSchema.getField(fieldName);
+      if (avroField == null) {
+        LOGGER.warn("Pinot field: {} does not exist in Avro Schema", fieldName);
+      } else {
+        boolean isPinotFieldSingleValue = fieldSpec.isSingleValueField();
+        boolean isAvroFieldSingleValue = AvroUtils.isSingleValueField(avroField);
+        if (isPinotFieldSingleValue != isAvroFieldSingleValue) {
+          String errorMessage = "Pinot field: " + fieldName + " is " + (isPinotFieldSingleValue ? "Single" : "Multi")
+              + "-valued in Pinot schema but not in Avro schema";
+          LOGGER.error(errorMessage);
+          throw new IllegalStateException(errorMessage);
+        }
+
+        FieldSpec.DataType pinotFieldDataType = fieldSpec.getDataType();
+        FieldSpec.DataType avroFieldDataType = AvroUtils.extractFieldDataType(avroField);
+        if (pinotFieldDataType != avroFieldDataType) {
+          LOGGER.warn("Pinot field: {} of type: {} mismatches with corresponding field in Avro Schema of type: {}",
+              fieldName, pinotFieldDataType, avroFieldDataType);
+        }
+      }
+    }
+  }
+
   /**
    * Get the Avro file reader for the given file.
    */
   public static DataFileStream<GenericRecord> getAvroReader(File avroFile)
       throws IOException {
     if (avroFile.getName().endsWith(".gz")) {
-      return new DataFileStream<>(new GZIPInputStream(new FileInputStream(avroFile)),
-          new GenericDatumReader<GenericRecord>());
+      return new DataFileStream<>(new GZIPInputStream(new FileInputStream(avroFile)), new GenericDatumReader<>());
     } else {
-      return new DataFileStream<>(new FileInputStream(avroFile), new GenericDatumReader<GenericRecord>());
+      return new DataFileStream<>(new FileInputStream(avroFile), new GenericDatumReader<>());
     }
   }
 
@@ -275,33 +296,4 @@ private static org.apache.avro.Schema extractSupportedSchema(org.apache.avro.Sch
       return fieldSchema;
     }
   }
-
-  /**
-   * Valid table schema with avro schema
-   */
-  public static void validateSchema(Schema schema, org.apache.avro.Schema avroSchema) {
-    for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) {
-      String fieldName = fieldSpec.getName();
-      Field avroField = avroSchema.getField(fieldName);
-      if (avroField == null) {
-        LOGGER.warn("Pinot field: {} does not exist in Avro Schema", fieldName);
-      } else {
-        boolean isPinotFieldSingleValue = fieldSpec.isSingleValueField();
-        boolean isAvroFieldSingleValue = AvroUtils.isSingleValueField(avroField);
-        if (isPinotFieldSingleValue != isAvroFieldSingleValue) {
-          String errorMessage = "Pinot field: " + fieldName + " is " + (isPinotFieldSingleValue ? "Single" : "Multi")
-                  + "-valued in Pinot schema but not in Avro schema";
-          LOGGER.error(errorMessage);
-          throw new IllegalStateException(errorMessage);
-        }
-
-        FieldSpec.DataType pinotFieldDataType = fieldSpec.getDataType();
-        FieldSpec.DataType avroFieldDataType = AvroUtils.extractFieldDataType(avroField);
-        if (pinotFieldDataType != avroFieldDataType) {
-          LOGGER.warn("Pinot field: {} of type: {} mismatches with corresponding field in Avro Schema of type: {}",
-                  fieldName, pinotFieldDataType, avroFieldDataType);
-        }
-      }
-    }
-  }
 }

pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/SegmentCreationJob.java

@@ -104,12 +104,12 @@ public SegmentCreationJob(Properties properties) {
 
   @Override
   protected boolean isDataFile(String fileName) {
-    // Other files may have different extensions, eg: orc can have no extension
+    // For custom record reader, treat all files as data file
     if (_properties.getProperty(JobConfigConstants.RECORD_READER_PATH) != null) {
       return true;
     }
     return fileName.endsWith(".avro") || fileName.endsWith(".csv") || fileName.endsWith(".json") || fileName
-        .endsWith(".thrift") || fileName.endsWith(".parquet");
+        .endsWith(".thrift");
   }
 
   public void run()

pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mapper/SegmentCreationMapper.java

@@ -101,8 +101,7 @@ public void setup(Context context)
     if (readerConfigFile != null) {
       _readerConfigFile = new Path(readerConfigFile);
     }
-
-    _recordReaderPath = _jobConf.get(JobConfigConstants.RECORD_READER_PATH, null);
+    _recordReaderPath = _jobConf.get(JobConfigConstants.RECORD_READER_PATH);
 
     // Set up segment name generator
     String segmentNameGeneratorType =
@@ -207,8 +206,8 @@ protected void map(LongWritable key, Text value, Context context)
     segmentGeneratorConfig.setOutDir(_localSegmentDir.getPath());
     segmentGeneratorConfig.setSegmentNameGenerator(_segmentNameGenerator);
     segmentGeneratorConfig.setSequenceId(sequenceId);
-    segmentGeneratorConfig.setRecordReaderPath(_recordReaderPath);
     if (_recordReaderPath != null) {
+      segmentGeneratorConfig.setRecordReaderPath(_recordReaderPath);
       segmentGeneratorConfig.setFormat(FileFormat.OTHER);
     } else {
       FileFormat fileFormat = getFileFormat(inputFileName);
@@ -276,9 +275,6 @@ protected FileFormat getFileFormat(String fileName) {
     if (fileName.endsWith(".thrift")) {
       return FileFormat.THRIFT;
     }
-    if (fileName.endsWith(".parquet")) {
-      return FileFormat.PARQUET;
-    }
     throw new IllegalArgumentException("Unsupported file format: {}" + fileName);
   }
 

pinot-orc/pom.xml

@@ -74,4 +74,4 @@
       <scope>test</scope>
     </dependency>
   </dependencies>
-</project>
+</project>