Add changes for presto mor queries (#1578)

Adds the neccessary changes to hudi for support of presto querying hudi
merge-on-read table's realtime view.

Co-authored-by: Brandon Scheller <bschelle@amazon.com>

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/UseRecordReaderFromInputFormat.java

@@ -0,0 +1,38 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License.  You may obtain a copy of the License at
+*
+*      http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.hudi.hadoop;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+* When annotated on a InputFormat, informs the query engines, that they should use the RecordReader provided by the input
+* format to execute the queries.
+*/
+@Inherited
+@Documented
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface UseRecordReaderFromInputFormat {
+
+}

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

@@ -47,6 +47,7 @@ import org.apache.hadoop.mapred.InputSplit;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.mapred.Reporter;
+import org.apache.hudi.hadoop.UseRecordReaderFromInputFormat;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
@@ -63,6 +64,7 @@ import java.util.stream.Stream;
 /**
  * Input Format, that provides a real-time view of data in a Hoodie table.
  */
+@UseRecordReaderFromInputFormat
 @UseFileSplitsFromInputFormat
 public class HoodieParquetRealtimeInputFormat extends HoodieParquetInputFormat implements Configurable {
 

hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestAnnotation.java

@@ -19,7 +19,7 @@
 package org.apache.hudi.hadoop;
 
 import org.junit.jupiter.api.Test;
-
+import org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat;
 import java.lang.annotation.Annotation;
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -27,7 +27,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 public class TestAnnotation {
 
   @Test
-  public void testAnnotation() {
+  public void testHoodieParquetInputFormatAnnotation() {
     assertTrue(HoodieParquetInputFormat.class.isAnnotationPresent(UseFileSplitsFromInputFormat.class));
     Annotation[] annotations = HoodieParquetInputFormat.class.getAnnotations();
     boolean found = false;
@@ -38,4 +38,23 @@ public class TestAnnotation {
     }
     assertTrue(found);
   }
+
+  @Test
+  public void testHoodieParquetRealtimeInputFormatAnnotations() {
+    assertTrue(HoodieParquetRealtimeInputFormat.class.isAnnotationPresent(UseFileSplitsFromInputFormat.class));
+    assertTrue(HoodieParquetRealtimeInputFormat.class.isAnnotationPresent(UseRecordReaderFromInputFormat.class));
+    Annotation[] annotations = HoodieParquetRealtimeInputFormat.class.getAnnotations();
+    boolean foundFileSplitsAnnotation = false;
+    boolean foundRecordReaderAnnotation = false;
+    for (Annotation annotation : annotations) {
+      if ("UseFileSplitsFromInputFormat".equals(annotation.annotationType().getSimpleName())) {
+        foundFileSplitsAnnotation = true;
+      }
+      if ("UseRecordReaderFromInputFormat".equals(annotation.annotationType().getSimpleName())) {
+        foundRecordReaderAnnotation = true;
+      }
+    }
+    assertTrue(foundFileSplitsAnnotation);
+    assertTrue(foundRecordReaderAnnotation);
+  }
 }

packaging/hudi-presto-bundle/pom.xml

@@ -68,6 +68,7 @@
                   <include>org.apache.hudi:hudi-hadoop-mr</include>
 
                   <include>org.apache.parquet:parquet-avro</include>
+                  <include>org.apache.avro:avro</include>
                   <include>com.esotericsoftware:kryo-shaded</include>
                   <include>org.objenesis:objenesis</include>
                   <include>com.esotericsoftware:minlog</include>
@@ -75,6 +76,10 @@
               </artifactSet>
               <relocations>
 
+                <relocation>
+                  <pattern>org.apache.avro.</pattern>
+                  <shadedPattern>org.apache.hudi.org.apache.avro.</shadedPattern>
+                </relocation>
                 <relocation>
                   <pattern>com.esotericsoftware.kryo.</pattern>
                   <shadedPattern>org.apache.hudi.com.esotericsoftware.kryo.</shadedPattern>
@@ -128,5 +133,18 @@
       <artifactId>hudi-hadoop-mr-bundle</artifactId>
       <version>${project.version}</version>
     </dependency>
+
+    <!-- Parquet -->
+    <dependency>
+      <groupId>org.apache.parquet</groupId>
+      <artifactId>parquet-avro</artifactId>
+      <scope>compile</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.avro</groupId>
+      <artifactId>avro</artifactId>
+      <scope>compile</scope>
+    </dependency>
   </dependencies>
 </project>