[DNM, stacked] fix(storage-format): fix multi-block previous-block offset and add byte-level HFile writer tests

hudi-io/hfile_format.md

@@ -197,7 +197,10 @@ Key:
 
 - **Key Content Size**: 2 byte, short, size of the key content.
 - **Key Content**: key content in byte array. In Hudi, we serialize the String into byte array using UTF-8.
-- **Other Information**: other information of the key, which is not used by Hudi.
+- **Other Information**: the remaining fields that complete the KeyValue key, so it parses as a
+  standard KeyValue key (and is readable by an HBase HFile reader): 1-byte column-family length
+  (`0`), 8-byte timestamp (`Long.MAX_VALUE`, the "latest" sentinel), and 1-byte key type (`Put` = 4).
+  Hudi's reader ignores these fields.
 
 Value:
 
@@ -255,8 +258,11 @@ Each block index entry, referencing one relevant Data or Meta Block, has the fol
 - **Block Offset**: 8 bytes, long, the start offset of a data or meta block in the file.
 - **Block Size on Disk**: 4 bytes, integer, the on-disk size of the block, so the block can be skipped based on the
   size.
-- **Key Length**: [variable-length encoded](https://en.wikipedia.org/wiki/Variable-length_quantity) number representing
-  the length of the "Key" part.
+- **Key Length**: the length of the "Key" part, encoded as a Hadoop `WritableUtils` variable-length integer (VInt). A
+  value in `[-112, 127]` is a single byte; otherwise the first byte encodes the number of following big-endian value
+  bytes and the sign. This is the encoding the HBase HFile block index uses and that the reader decodes. It is **not**
+  the Protobuf varint used for the trailer and file info length prefixes (see those sections): the two are byte-identical
+  for `0..127` but diverge at `>= 128`, so they are not interchangeable.
 
 Key:
 
@@ -286,7 +292,10 @@ For Data Index, the "Key Bytes" part has the following format (same as the key f
 
 - **Key Content Size**: 2 byte, short, size of the key content.
 - **Key Content**: key content in byte array. In Hudi, we encode the String into bytes using UTF-8.
-- **Other Information**: other information of the key, which is not used by Hudi.
+- **Other Information**: the remaining fields that complete the KeyValue key, so it parses as a
+  standard KeyValue key (and is readable by an HBase HFile reader): 1-byte column-family length
+  (`0`), 8-byte timestamp (`Long.MAX_VALUE`, the "latest" sentinel), and 1-byte key type (`Put` = 4).
+  Hudi's reader ignores these fields.
 
 For Meta Index, the "Key Bytes" part is the byte array of the key of the Meta Block.
 
@@ -308,7 +317,9 @@ The "Data" part of the File Info Block has the following format:
 ```
 
 - **PBUF Magic**: 4 bytes, magic bytes `PBUF` indicating the block is using Protobuf for serde.
-- **File Info**: a small key-value map of metadata serialized in Protobuf.
+- **File Info**: a small key-value map of metadata serialized in Protobuf, length-delimited (a Protobuf varint length
+  prefix followed by the `InfoProto` message, i.e. `writeDelimitedTo` / `parseDelimitedFrom`). This Protobuf varint is a
+  different encoding from the Hadoop `WritableUtils` VInt used for the block-index Key Length above.
 
 Here's the definition of the File Info proto `InfoProto`:
 
@@ -352,7 +363,10 @@ follows:
 ```
 
 - **Block Magic**: 8 bytes, a sequence of bytes indicating the Trailer, i.e., `TRABLK"$`.
-- **Trailer Content**: the metadata fields are serialized in Protobuf, defined as follows
+- **Trailer Content**: the metadata fields are serialized in Protobuf, length-delimited (a Protobuf varint length prefix
+  followed by the `TrailerProto` message, i.e. `writeDelimitedTo` / `parseDelimitedFrom`) right after the magic. As with
+  the file info block, this Protobuf varint differs from the Hadoop `WritableUtils` VInt used for the block-index Key
+  Length. The proto is defined as follows
 
 ```
 message TrailerProto {

hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileBlock.java

@@ -22,7 +22,6 @@
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.io.compress.CompressionCodec;
 
-import com.google.protobuf.CodedOutputStream;
 import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.Setter;
@@ -36,6 +35,7 @@
 
 import static org.apache.hudi.io.hfile.DataSize.MAGIC_LENGTH;
 import static org.apache.hudi.io.hfile.DataSize.SIZEOF_BYTE;
+import static org.apache.hudi.io.hfile.DataSize.SIZEOF_INT16;
 import static org.apache.hudi.io.hfile.DataSize.SIZEOF_INT32;
 import static org.apache.hudi.io.hfile.DataSize.SIZEOF_INT64;
 import static org.apache.hudi.io.util.IOUtils.readInt;
@@ -56,6 +56,12 @@ public abstract class HFileBlock {
   static final int CHECKSUM_SIZE = SIZEOF_INT32;
   private static final int DEFAULT_BYTES_PER_CHECKSUM = 16 * 1024;
   private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
+  // Hudi does not set a version timestamp on key-value pairs, so the latest timestamp is used.
+  private static final long LATEST_TIMESTAMP = Long.MAX_VALUE;
+  // Key type is constant Put (4) in Hudi.
+  private static final byte KEY_TYPE_PUT = (byte) 4;
+  // KeyValue key suffix beyond the row: column-family length (1) + timestamp (8) + type (1).
+  private static final int KEY_METADATA_SUFFIX_LENGTH = SIZEOF_BYTE + SIZEOF_INT64 + SIZEOF_BYTE;
 
   static class Header {
     // Format of header is:
@@ -319,16 +325,37 @@ private byte[] generateChecksumBytes(ChecksumType type, int numChecksumBytes) {
   }
 
   /**
-   * Returns the bytes of the variable length encoding for an integer.
-   * @param length       an integer, normally representing a length.
-   * @return             variable length encoding.
-   * @throws IOException upon error.
+   * Returns the serialized length of the KeyValue key for a row: the 2-byte row-length prefix, the
+   * row, and the 10-byte metadata suffix (column-family length, timestamp, key type).
+   *
+   * <p>The data block and the root index block both write the full KeyValue key (not just the row)
+   * so that a reader can parse and point-look-up either block: a point lookup compares index keys
+   * against data keys, so the two must use byte-identical key encoding.
+   *
+   * @param rowLength length of the row (key content) in bytes.
+   * @return the KeyValue key length.
+   */
+  protected static int keyValueKeyLength(int rowLength) {
+    return SIZEOF_INT16 + rowLength + KEY_METADATA_SUFFIX_LENGTH;
+  }
+
+  /**
+   * Writes the KeyValue key for a row:
+   * {@code [2-byte rowLen][row][1-byte cfLen=0][8-byte ts=LATEST][1-byte type=Put]}. See
+   * {@link #keyValueKeyLength(int)} for why the data and index blocks share this encoding.
+   *
+   * @param out       output stream to write to.
+   * @param row       buffer holding the row (key content) bytes.
+   * @param offset    start of the row within {@code row}; a key may be a view into a larger buffer.
+   * @param rowLength number of row bytes to write; passed explicitly because a key's backing array
+   *                  may be larger than its content length.
    */
-  static byte[] getVariableLengthEncodedBytes(int length) throws IOException {
-    ByteArrayOutputStream varintBuffer = new ByteArrayOutputStream();
-    CodedOutputStream varintOutput = CodedOutputStream.newInstance(varintBuffer);
-    varintOutput.writeUInt32NoTag(length);
-    varintOutput.flush();
-    return varintBuffer.toByteArray();
+  protected static void writeKey(DataOutputStream out, byte[] row, int offset, int rowLength)
+      throws IOException {
+    out.writeShort((short) rowLength);
+    out.write(row, offset, rowLength);
+    out.write(0);                     // column-family length
+    out.writeLong(LATEST_TIMESTAMP);  // timestamp
+    out.write(KEY_TYPE_PUT);          // key type
   }
 }

hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileDataBlock.java

@@ -28,9 +28,6 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import static org.apache.hudi.io.hfile.DataSize.SIZEOF_BYTE;
-import static org.apache.hudi.io.hfile.DataSize.SIZEOF_INT16;
-import static org.apache.hudi.io.hfile.DataSize.SIZEOF_INT64;
 import static org.apache.hudi.io.hfile.HFileReader.SEEK_TO_BEFORE_BLOCK_FIRST_KEY;
 import static org.apache.hudi.io.hfile.HFileReader.SEEK_TO_FOUND;
 import static org.apache.hudi.io.hfile.HFileReader.SEEK_TO_IN_RANGE;
@@ -40,18 +37,11 @@
  * Represents a {@link HFileBlockType#DATA} block.
  */
 public class HFileDataBlock extends HFileBlock {
-  private static final int KEY_LENGTH_LENGTH = SIZEOF_INT16;
-  private static final int COLUMN_FAMILY_LENGTH = SIZEOF_BYTE;
-  private static final int VERSION_TIMESTAMP_LENGTH = SIZEOF_INT64;
-  private static final int KEY_TYPE_LENGTH = SIZEOF_BYTE;
   // Hudi does not use HFile MVCC timestamp version so the version
   // is always 0, thus the byte length of the version is always 1.
   // This assumption is also validated when parsing {@link HFileInfo},
   // i.e., the maximum MVCC timestamp in a HFile must be 0.
   private static final long ZERO_TS_VERSION_BYTE_LENGTH = 1;
-  // Hudi does not set version timestamp for key value pairs,
-  // so the latest timestamp is used.
-  private static final long LATEST_TIMESTAMP = Long.MAX_VALUE;
 
   // End offset of content in the block, relative to the start of the start of the block
   protected final int uncompressedContentEndRelativeOffset;
@@ -216,24 +206,11 @@ protected ByteBuffer getUncompressedBlockDataToWrite() throws IOException {
         // Length of key + length of a short variable indicating length of key.
         // Note that 10 extra bytes are required by hbase reader.
         // That is: 1 byte for column family length, 8 bytes for timestamp, 1 bytes for key type.
-        dataOutputStream.writeInt(
-            kv.key.length + KEY_LENGTH_LENGTH + COLUMN_FAMILY_LENGTH + VERSION_TIMESTAMP_LENGTH + KEY_TYPE_LENGTH);
+        dataOutputStream.writeInt(keyValueKeyLength(kv.key.length));
         // Length of value.
         dataOutputStream.writeInt(kv.value.length);
-        // Key content length.
-        dataOutputStream.writeShort((short)kv.key.length);
         // Key.
-        dataOutputStream.write(kv.key);
-        // Column family length: constant 0.
-        dataOutputStream.write(0);
-        // Column qualifier: assume 0 bits.
-        // Timestamp: using the latest.
-        dataOutputStream.writeLong(LATEST_TIMESTAMP);
-        // Key type: constant Put (4) in Hudi.
-        // Minimum((byte) 0), Put((byte) 4), Delete((byte) 8),
-        // DeleteFamilyVersion((byte) 10), DeleteColumn((byte) 12),
-        // DeleteFamily((byte) 14), Maximum((byte) 255).
-        dataOutputStream.write(4);
+        writeKey(dataOutputStream, kv.key, 0, kv.key.length);
         // Value.
         dataOutputStream.write(kv.value);
         // MVCC.

hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileFileInfoBlock.java

@@ -23,6 +23,7 @@
 import org.apache.hudi.io.util.IOUtils;
 
 import com.google.protobuf.ByteString;
+import com.google.protobuf.CodedOutputStream;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -103,11 +104,31 @@ public ByteBuffer getUncompressedBlockDataToWrite() throws IOException {
     outputStream.write(PB_MAGIC);
     byte[] payload = builder.build().toByteArray();
     try {
-      outputStream.write(getVariableLengthEncodedBytes(payload.length));
+      outputStream.write(getProtobufVarIntBytes(payload.length));
     } catch (IOException e) {
       throw new RuntimeException("Failed to calculate File Info variable length");
     }
     outputStream.write(payload);
     return ByteBuffer.wrap(outputStream.toByteArray());
   }
+
+  /**
+   * Encodes an integer as a Protobuf varint (base-128, little-endian with MSB continuation bits,
+   * via {@link CodedOutputStream#writeUInt32NoTag}): the length framing Protobuf's
+   * {@code writeDelimitedTo} produces. It is used for the file info and trailer length prefixes,
+   * which the reader decodes with {@code parseDelimitedFrom}. This is NOT interchangeable with the
+   * Hadoop {@code WritableUtils} VInt used by the block index
+   * ({@link HFileIndexBlock#getVarIntBytes(int)}): the two diverge for values >= 128.
+   *
+   * @param value the integer value to encode.
+   * @return the Protobuf varint encoding.
+   * @throws IOException upon error.
+   */
+  static byte[] getProtobufVarIntBytes(int value) throws IOException {
+    ByteArrayOutputStream varIntBuffer = new ByteArrayOutputStream();
+    CodedOutputStream varIntOutput = CodedOutputStream.newInstance(varIntBuffer);
+    varIntOutput.writeUInt32NoTag(value);
+    varIntOutput.flush();
+    return varIntBuffer.toByteArray();
+  }
 }

hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileIndexBlock.java

@@ -52,4 +52,40 @@ public int getNumOfEntries() {
   public boolean isEmpty() {
     return entries.isEmpty();
   }
+
+  /**
+   * Encodes an integer using {@code WritableUtils} variable-length encoding (VInt): the encoding
+   * the HFile block index uses for the per-entry key length and that the reader decodes
+   * ({@link org.apache.hudi.io.util.IOUtils#readVarLong}). Values in [-112, 127] use a single
+   * byte; otherwise the first byte carries the number of following big-endian value bytes and the
+   * sign. This is NOT interchangeable with the Protobuf varint
+   * ({@link HFileFileInfoBlock#getProtobufVarIntBytes(int)}): the two diverge for values >= 128.
+   *
+   * @param value the integer value to encode.
+   * @return the encoded byte array.
+   */
+  static byte[] getVarIntBytes(int value) {
+    if (value >= -112 && value <= 127) {
+      return new byte[] {(byte) value};
+    }
+    long longValue = value;
+    int len = -112;
+    if (longValue < 0) {
+      longValue ^= -1L;
+      len = -120;
+    }
+    long tmp = longValue;
+    while (tmp != 0) {
+      tmp >>= 8;
+      len--;
+    }
+    int numBytes = (len < -120) ? -(len + 120) : -(len + 112);
+    byte[] result = new byte[1 + numBytes];
+    result[0] = (byte) len;
+    for (int idx = 0; idx < numBytes; idx++) {
+      int shiftBits = (numBytes - idx - 1) * 8;
+      result[1 + idx] = (byte) ((longValue >> shiftBits) & 0xFF);
+    }
+    return result;
+  }
 }

hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileMetaIndexBlock.java

@@ -24,8 +24,6 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 
-import static org.apache.hudi.io.util.IOUtils.writeVarInt;
-
 public class HFileMetaIndexBlock extends HFileIndexBlock {
 
   private HFileMetaIndexBlock(HFileContext context) {
@@ -43,12 +41,8 @@ public ByteBuffer getUncompressedBlockDataToWrite() throws IOException {
       for (BlockIndexEntry entry : entries) {
         outputStream.writeLong(entry.getOffset());
         outputStream.writeInt(entry.getSize());
-        // Key length.
-        // Use Hadoop WritableUtils VarInt encoding to match HBase's HFile format.
-        byte[] keyLength = writeVarInt(entry.getFirstKey().getLength());
-        outputStream.write(keyLength);
-        // Note that: NO two-bytes for encoding key length.
-        // Key.
+        outputStream.write(getVarIntBytes(entry.getFirstKey().getLength()));
+        // Unlike the data index, the meta key is stored as-is with no 2-byte row-length prefix.
         outputStream.write(entry.getFirstKey().getBytes());
       }
     }

hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileRootIndexBlock.java

@@ -29,13 +29,11 @@
 import java.util.List;
 import java.util.TreeMap;
 
-import static org.apache.hudi.io.hfile.DataSize.SIZEOF_INT16;
 import static org.apache.hudi.io.util.IOUtils.copy;
 import static org.apache.hudi.io.util.IOUtils.decodeVarLongSizeOnDisk;
 import static org.apache.hudi.io.util.IOUtils.readInt;
 import static org.apache.hudi.io.util.IOUtils.readLong;
 import static org.apache.hudi.io.util.IOUtils.readVarLong;
-import static org.apache.hudi.io.util.IOUtils.writeVarInt;
 
 /**
  * Represents a {@link HFileBlockType#ROOT_INDEX} block.
@@ -108,16 +106,10 @@ public ByteBuffer getUncompressedBlockDataToWrite() throws IOException {
       for (BlockIndexEntry entry : entries) {
         outputStream.writeLong(entry.getOffset());
         outputStream.writeInt(entry.getSize());
-
-        // Key length + 2 (SIZEOF_INT16 for the 2-byte row key length prefix).
-        // Use Hadoop WritableUtils VarInt encoding to match HBase's HFile format.
-        byte[] keyLength = writeVarInt(
-            entry.getFirstKey().getLength() + SIZEOF_INT16);
-        outputStream.write(keyLength);
-        // Key length.
-        outputStream.writeShort((short) entry.getFirstKey().getLength());
-        // Key.
-        outputStream.write(entry.getFirstKey().getBytes());
+        int kvKeyLength = keyValueKeyLength(entry.getFirstKey().getLength());
+        outputStream.write(getVarIntBytes(kvKeyLength));
+        Key firstKey = entry.getFirstKey();
+        writeKey(outputStream, firstKey.getBytes(), firstKey.getOffset(), firstKey.getLength());
       }
     }
 

hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileWriterImpl.java

@@ -31,8 +31,8 @@
 import java.util.Map;
 
 import static org.apache.hudi.io.hfile.DataSize.SIZEOF_INT16;
-import static org.apache.hudi.io.hfile.HFileBlock.getVariableLengthEncodedBytes;
 import static org.apache.hudi.io.hfile.HFileBlockType.TRAILER;
+import static org.apache.hudi.io.hfile.HFileFileInfoBlock.getProtobufVarIntBytes;
 import static org.apache.hudi.io.hfile.HFileInfo.KEY_VALUE_VERSION_WITH_MVCC_TS;
 import static org.apache.hudi.io.hfile.HFileInfo.LAST_KEY;
 import static org.apache.hudi.io.hfile.HFileInfo.MAX_MVCC_TS_KEY;
@@ -144,8 +144,8 @@ private void flushCurrentDataBlock() throws IOException {
     // 3. Create an index entry.
     rootIndexBlock.add(
         currentDataBlock.getFirstKey(), lastDataBlockOffset, blockBuffer.limit());
-    // 4. Create a new data block.
-    currentDataBlock = HFileDataBlock.createDataBlockToWrite(context, currentOffset);
+    // 4. Create a new data block whose previous-block offset points at the block just flushed.
+    currentDataBlock = HFileDataBlock.createDataBlockToWrite(context, lastDataBlockOffset);
   }
 
   // NOTE that: reader assumes that every meta info piece
@@ -200,7 +200,7 @@ private void writeTrailer() throws IOException {
     ByteBuffer trailer = ByteBuffer.allocate(TRAILER_SIZE);
     trailer.limit(TRAILER_SIZE);
     trailer.put(TRAILER.getMagic());
-    trailer.put(getVariableLengthEncodedBytes(trailerProto.getSerializedSize()));
+    trailer.put(getProtobufVarIntBytes(trailerProto.getSerializedSize()));
     trailer.put(trailerProto.toByteArray());
     // Force trailer to have fixed length.
     trailer.position(TRAILER_SIZE - 1);