[WIP][DO NOT MERGE][CI Test Only] Remove hbase-server dependency, pull in HFile related classes, with deps resolution

hudi-aws/pom.xml

@@ -51,8 +51,6 @@
         <dependency>
             <groupId>org.apache.hadoop</groupId>
             <artifactId>hadoop-common</artifactId>
-            <classifier>tests</classifier>
-            <scope>test</scope>
             <exclusions>
                 <exclusion>
                     <groupId>org.mortbay.jetty</groupId>

hudi-client/hudi-client-common/pom.xml

@@ -120,7 +120,6 @@
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-hdfs</artifactId>
-      <classifier>tests</classifier>
       <scope>test</scope>
       <!-- Need these exclusions to make sure JavaSparkContext can be setup. https://issues.apache.org/jira/browse/SPARK-1693 -->
       <exclusions>
@@ -140,9 +139,28 @@
     </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-common</artifactId>
+      <artifactId>hadoop-hdfs</artifactId>
       <classifier>tests</classifier>
       <scope>test</scope>
+      <!-- Need these exclusions to make sure JavaSparkContext can be setup. https://issues.apache.org/jira/browse/SPARK-1693 -->
+      <exclusions>
+        <exclusion>
+          <groupId>org.mortbay.jetty</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>javax.servlet.jsp</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>javax.servlet</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-common</artifactId>
       <exclusions>
         <exclusion>
           <groupId>org.mortbay.jetty</groupId>

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

@@ -60,7 +60,7 @@
 import org.apache.hudi.table.action.compact.CompactionTriggerStrategy;
 import org.apache.hudi.table.action.compact.strategy.CompactionStrategy;
 
-import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.hudi.hbase.io.compress.Compression;
 import org.apache.hudi.table.storage.HoodieStorageLayout;
 import org.apache.orc.CompressionKind;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;

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

@@ -21,16 +21,15 @@
 import org.apache.hudi.common.bloom.BloomFilter;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HColumnDescriptor;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.io.compress.Compression;
-import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hudi.hbase.CellComparator;
+import org.apache.hudi.hbase.io.compress.Compression;
+import org.apache.hudi.hbase.io.hfile.CacheConfig;
 
 public class HoodieHFileConfig {
 
-  public static final KeyValue.KVComparator HFILE_COMPARATOR = new HoodieHBaseKVComparator();
+  public static final CellComparator HFILE_COMPARATOR = new HoodieHBaseKVComparator();
   public static final boolean PREFETCH_ON_OPEN = CacheConfig.DEFAULT_PREFETCH_ON_OPEN;
-  public static final boolean CACHE_DATA_IN_L1 = HColumnDescriptor.DEFAULT_CACHE_DATA_IN_L1;
+  public static final boolean CACHE_DATA_IN_L1 = false;// HColumnDescriptor.DEFAULT_CACHE_DATA_IN_L1;
   // This is private in CacheConfig so have been copied here.
   public static final boolean DROP_BEHIND_CACHE_COMPACTION = true;
 
@@ -42,12 +41,12 @@ public class HoodieHFileConfig {
   private final boolean dropBehindCacheCompaction;
   private final Configuration hadoopConf;
   private final BloomFilter bloomFilter;
-  private final KeyValue.KVComparator hfileComparator;
+  private final CellComparator hfileComparator;
   private final String keyFieldName;
 
   public HoodieHFileConfig(Configuration hadoopConf, Compression.Algorithm compressionAlgorithm, int blockSize,
                            long maxFileSize, String keyFieldName, boolean prefetchBlocksOnOpen, boolean cacheDataInL1,
-                           boolean dropBehindCacheCompaction, BloomFilter bloomFilter, KeyValue.KVComparator hfileComparator) {
+                           boolean dropBehindCacheCompaction, BloomFilter bloomFilter, CellComparator hfileComparator) {
     this.hadoopConf = hadoopConf;
     this.compressionAlgorithm = compressionAlgorithm;
     this.blockSize = blockSize;
@@ -96,7 +95,7 @@ public BloomFilter getBloomFilter() {
     return bloomFilter;
   }
 
-  public KeyValue.KVComparator getHfileComparator() {
+  public CellComparator getHFileComparator() {
     return hfileComparator;
   }
 

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

@@ -31,12 +31,11 @@
 import org.apache.avro.generic.IndexedRecord;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.HColumnDescriptor;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.io.hfile.CacheConfig;
-import org.apache.hadoop.hbase.io.hfile.HFile;
-import org.apache.hadoop.hbase.io.hfile.HFileContext;
-import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
+import org.apache.hudi.hbase.KeyValue;
+import org.apache.hudi.hbase.io.hfile.CacheConfig;
+import org.apache.hudi.hbase.io.hfile.HFile;
+import org.apache.hudi.hbase.io.hfile.HFileContext;
+import org.apache.hudi.hbase.io.hfile.HFileContextBuilder;
 import org.apache.hadoop.io.Writable;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.StringUtils;
@@ -56,6 +55,8 @@
  */
 public class HoodieHFileWriter<T extends HoodieRecordPayload, R extends IndexedRecord>
     implements HoodieFileWriter<R> {
+  // TODO(yihua): pulled from HColumnDescriptor
+  public static final String CACHE_DATA_IN_L1 = "CACHE_DATA_IN_L1";
   private static AtomicLong recordIndex = new AtomicLong(1);
 
   private final Path file;
@@ -95,16 +96,17 @@ public HoodieHFileWriter(String instantTime, Path file, HoodieHFileConfig hfileC
 
     HFileContext context = new HFileContextBuilder().withBlockSize(hfileConfig.getBlockSize())
         .withCompression(hfileConfig.getCompressionAlgorithm())
+        .withCellComparator(hfileConfig.getHFileComparator())
         .build();
 
     conf.set(CacheConfig.PREFETCH_BLOCKS_ON_OPEN_KEY, String.valueOf(hfileConfig.shouldPrefetchBlocksOnOpen()));
-    conf.set(HColumnDescriptor.CACHE_DATA_IN_L1, String.valueOf(hfileConfig.shouldCacheDataInL1()));
+    // HColumnDescriptor.CACHE_DATA_IN_L1
+    conf.set(CACHE_DATA_IN_L1, String.valueOf(hfileConfig.shouldCacheDataInL1()));
     conf.set(DROP_BEHIND_CACHE_COMPACTION_KEY, String.valueOf(hfileConfig.shouldDropBehindCacheCompaction()));
     CacheConfig cacheConfig = new CacheConfig(conf);
     this.writer = HFile.getWriterFactory(conf, cacheConfig)
         .withPath(this.fs, this.file)
         .withFileContext(context)
-        .withComparator(hfileConfig.getHfileComparator())
         .create();
 
     writer.appendFileInfo(HoodieHFileReader.KEY_SCHEMA.getBytes(), schema.toString().getBytes());

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

@@ -32,9 +32,9 @@
 import org.apache.avro.generic.IndexedRecord;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.io.compress.Compression;
-import org.apache.hadoop.hbase.io.hfile.CacheConfig;
-import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hudi.hbase.io.compress.Compression;
+import org.apache.hudi.hbase.io.hfile.CacheConfig;
+import org.apache.hudi.hbase.util.Pair;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.io.TempDir;

hudi-client/hudi-java-client/pom.xml

@@ -122,6 +122,26 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-hdfs</artifactId>
+            <scope>test</scope>
+            <!-- Need these exclusions to make sure JavaSparkContext can be setup. https://issues.apache.org/jira/browse/SPARK-1693 -->
+            <exclusions>
+                <exclusion>
+                    <groupId>org.mortbay.jetty</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>javax.servlet.jsp</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>javax.servlet</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>org.apache.hadoop</groupId>
             <artifactId>hadoop-hdfs</artifactId>

hudi-client/hudi-spark-client/pom.xml

@@ -65,6 +65,32 @@
       <artifactId>parquet-avro</artifactId>
     </dependency>
 
+    <!-- HBase -->
+    <dependency>
+      <groupId>org.apache.hbase</groupId>
+      <artifactId>hbase-client</artifactId>
+      <version>${hbase.version}</version>
+      <scope>compile</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>javax.servlet</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.codehaus.jackson</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.mortbay.jetty</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>tomcat</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+
     <!-- Hoodie - Test -->
     <dependency>
       <groupId>org.apache.hudi</groupId>

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java

@@ -95,8 +95,8 @@
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.io.hfile.CacheConfig;
-import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hudi.hbase.io.hfile.CacheConfig;
+import org.apache.hudi.hbase.util.Pair;
 import org.apache.hadoop.util.Time;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java

@@ -22,8 +22,8 @@
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.generic.IndexedRecord;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.io.hfile.CacheConfig;
-import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hudi.hbase.io.hfile.CacheConfig;
+import org.apache.hudi.hbase.util.Pair;
 import org.apache.hudi.avro.HoodieAvroUtils;
 import org.apache.hudi.avro.model.HoodieMetadataRecord;
 import org.apache.hudi.common.config.HoodieMetadataConfig;

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java

@@ -45,10 +45,10 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.io.hfile.CacheConfig;
-import org.apache.hadoop.hbase.io.hfile.HFile;
-import org.apache.hadoop.hbase.io.hfile.HFileScanner;
+import org.apache.hudi.hbase.Cell;
+import org.apache.hudi.hbase.io.hfile.CacheConfig;
+import org.apache.hudi.hbase.io.hfile.HFile;
+import org.apache.hudi.hbase.io.hfile.HFileScanner;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 import org.apache.spark.SparkConf;
@@ -241,9 +241,9 @@ public static Stream<GenericRecord> readHFile(JavaSparkContext jsc, String[] pat
     Schema schema = null;
     for (String path : paths) {
       try {
-        HFile.Reader reader = HFile.createReader(fs, new Path(path), cacheConfig, fs.getConf());
+        HFile.Reader reader = HFile.createReader(fs, new Path(path), cacheConfig, true, fs.getConf());
         if (schema == null) {
-          schema = new Schema.Parser().parse(new String(reader.loadFileInfo().get("schema".getBytes())));
+          schema = new Schema.Parser().parse(new String(reader.getHFileInfo().get("schema".getBytes())));
         }
         HFileScanner scanner = reader.getScanner(false, false);
         if (!scanner.seekTo()) {
@@ -252,7 +252,7 @@ public static Stream<GenericRecord> readHFile(JavaSparkContext jsc, String[] pat
         }
 
         do {
-          Cell c = scanner.getKeyValue();
+          Cell c = scanner.getCell();
           byte[] value = Arrays.copyOfRange(c.getValueArray(), c.getValueOffset(), c.getValueOffset() + c.getValueLength());
           valuesAsList.add(HoodieAvroUtils.bytesToAvro(value, schema));
         } while (scanner.next());

hudi-common/pom.xml

@@ -156,6 +156,12 @@
   </build>
 
   <dependencies>
+    <dependency>
+      <groupId>org.apache.hudi</groupId>
+      <artifactId>hudi-io</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
     <!-- Scala -->
     <dependency>
       <groupId>org.scala-lang</groupId>
@@ -293,7 +299,7 @@
       <groupId>org.apache.hbase</groupId>
       <artifactId>hbase-server</artifactId>
       <version>${hbase.version}</version>
-      <!-- Unfortunately, HFile is packaged ONLY under hbase-server --> 
+      <!-- Unfortunately, HFile is packaged ONLY under hbase-server -->
       <scope>compile</scope>
       <exclusions>
         <exclusion>

hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java

@@ -37,15 +37,16 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.io.hfile.CacheConfig;
-import org.apache.hadoop.hbase.io.hfile.HFile;
-import org.apache.hadoop.hbase.io.hfile.HFileContext;
-import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
-import org.apache.hadoop.hbase.io.hfile.HFileScanner;
-import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hudi.hbase.CellComparatorImpl;
+import org.apache.hudi.hbase.CellUtil;
+import org.apache.hudi.hbase.HConstants;
+import org.apache.hudi.hbase.KeyValue;
+import org.apache.hudi.hbase.io.hfile.CacheConfig;
+import org.apache.hudi.hbase.io.hfile.HFile;
+import org.apache.hudi.hbase.io.hfile.HFileContext;
+import org.apache.hudi.hbase.io.hfile.HFileContextBuilder;
+import org.apache.hudi.hbase.io.hfile.HFileScanner;
+import org.apache.hudi.hbase.util.Bytes;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
@@ -178,9 +179,7 @@ private static String getUserKeyFromCellKey(String cellKey) {
   private static HFile.Reader createReader(String hFilePath, Configuration conf, FileSystem fileSystem) {
     try {
       LOG.info("Opening HFile for reading :" + hFilePath);
-      HFile.Reader reader = HFile.createReader(fileSystem, new HFilePathForReader(hFilePath),
-          new CacheConfig(conf), conf);
-      return reader;
+      return HFile.createReader(fileSystem, new HFilePathForReader(hFilePath), new CacheConfig(conf), true, conf);
     } catch (IOException ioe) {
       throw new HoodieIOException(ioe.getMessage(), ioe);
     }
@@ -259,7 +258,7 @@ private void initIndexInfo() {
 
     private HoodieBootstrapIndexInfo fetchBootstrapIndexInfo() throws IOException {
       return TimelineMetadataUtils.deserializeAvroMetadata(
-          partitionIndexReader().loadFileInfo().get(INDEX_INFO_KEY),
+          partitionIndexReader().getHFileInfo().get(INDEX_INFO_KEY),
           HoodieBootstrapIndexInfo.class);
     }
 
@@ -306,7 +305,7 @@ private <T> List<T> getAllKeys(HFileScanner scanner, Function<String, T> convert
       try {
         boolean available = scanner.seekTo();
         while (available) {
-          keys.add(converter.apply(getUserKeyFromCellKey(CellUtil.getCellKeyAsString(scanner.getKeyValue()))));
+          keys.add(converter.apply(getUserKeyFromCellKey(CellUtil.getCellKeyAsString(scanner.getCell()))));
           available = scanner.next();
         }
       } catch (IOException ioe) {
@@ -528,13 +527,13 @@ public void close() {
     @Override
     public void begin() {
       try {
-        HFileContext meta = new HFileContextBuilder().build();
+        HFileContext meta = new HFileContextBuilder().withCellComparator(new HoodieKVComparator()).build();
         this.indexByPartitionWriter = HFile.getWriterFactory(metaClient.getHadoopConf(),
             new CacheConfig(metaClient.getHadoopConf())).withPath(metaClient.getFs(), indexByPartitionPath)
-            .withFileContext(meta).withComparator(new HoodieKVComparator()).create();
+            .withFileContext(meta).create();
         this.indexByFileIdWriter = HFile.getWriterFactory(metaClient.getHadoopConf(),
             new CacheConfig(metaClient.getHadoopConf())).withPath(metaClient.getFs(), indexByFileIdPath)
-            .withFileContext(meta).withComparator(new HoodieKVComparator()).create();
+            .withFileContext(meta).create();
       } catch (IOException ioe) {
         throw new HoodieIOException(ioe.getMessage(), ioe);
       }
@@ -581,6 +580,6 @@ public String getName() {
    * This class is explicitly used as Key Comparator to workaround hard coded
    * legacy format class names inside HBase. Otherwise we will face issues with shading.
    */
-  public static class HoodieKVComparator extends KeyValue.KVComparator {
+  public static class HoodieKVComparator extends CellComparatorImpl {
   }
 }

hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java

@@ -50,6 +50,7 @@
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Deque;
 import java.util.HashSet;
 import java.util.List;
@@ -424,6 +425,9 @@ private void processQueuedBlocksForInstant(Deque<HoodieLogBlock> logBlocks, int
           processDataBlock((HoodieAvroDataBlock) lastBlock, keys);
           break;
         case HFILE_DATA_BLOCK:
+          if (!keys.isPresent()) {
+            keys = Option.of(Collections.emptyList());
+          }
           processDataBlock((HoodieHFileDataBlock) lastBlock, keys);
           break;
         case DELETE_BLOCK: