diff --git a/hudi-io/hfile_format.md b/hudi-io/hfile_format.md index 192c3d4313f87..3687c6fbc0070 100644 --- a/hudi-io/hfile_format.md +++ b/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 { diff --git a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileBlock.java b/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileBlock.java index 56f75466dc61c..ae99ef2f1cf7e 100644 --- a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileBlock.java +++ b/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). + * + *

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 } } diff --git a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileDataBlock.java b/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileDataBlock.java index cbf3e719f6a02..6664b54fa2a9e 100644 --- a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileDataBlock.java +++ b/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. diff --git a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileFileInfoBlock.java b/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileFileInfoBlock.java index 2fc1da4f74292..c625a73cb8ec4 100644 --- a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileFileInfoBlock.java +++ b/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(); + } } diff --git a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileIndexBlock.java b/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileIndexBlock.java index 72ddd6edf218c..e212c8488a00b 100644 --- a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileIndexBlock.java +++ b/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; + } } diff --git a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileMetaIndexBlock.java b/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileMetaIndexBlock.java index 72f720b9c4943..e461451c1195a 100644 --- a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileMetaIndexBlock.java +++ b/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()); } } diff --git a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileRootIndexBlock.java b/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileRootIndexBlock.java index 5b32a5df8925f..e9b8cb715bf06 100644 --- a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileRootIndexBlock.java +++ b/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()); } } diff --git a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileWriterImpl.java b/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileWriterImpl.java index 5620bba7c8114..348d1b06b6615 100644 --- a/hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileWriterImpl.java +++ b/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); diff --git a/hudi-io/src/main/java/org/apache/hudi/io/util/IOUtils.java b/hudi-io/src/main/java/org/apache/hudi/io/util/IOUtils.java index 274dfe6d8f907..3fd5930add469 100644 --- a/hudi-io/src/main/java/org/apache/hudi/io/util/IOUtils.java +++ b/hudi-io/src/main/java/org/apache/hudi/io/util/IOUtils.java @@ -150,42 +150,6 @@ public static boolean isNegativeVarLong(byte value) { return value < -120 || (value >= -112 && value < 0); } - /** - * Encodes an integer using Hadoop-compatible variable-length encoding - * (WritableUtils VarInt format) and returns the encoded bytes. - * - *

For values between -112 and 127 (inclusive), a single byte is used. - * For other values, the first byte indicates the number of following bytes - * and the sign, followed by the value in big-endian order. - * - * @param value the integer value to encode. - * @return the encoded byte array. - */ - public static byte[] writeVarInt(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; - } - /** * @param bytes input byte array. * @param offset offset to start reading. diff --git a/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileDataBlock.java b/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileDataBlock.java new file mode 100644 index 0000000000000..c040bca9a97e1 --- /dev/null +++ b/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileDataBlock.java @@ -0,0 +1,79 @@ +/* + * 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.io.hfile; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Validates the exact on-disk bytes the data block writer emits for each record. A data entry is a + * full HBase KeyValue: {@code [4-byte keyLen][4-byte valueLen][2-byte rowLen][row][1-byte cfLen=0] + * [8-byte ts=LATEST][1-byte type=Put][value][1-byte MVCC=0]}. An HBase reader relies on this exact + * framing, so the test asserts every field rather than a single opaque blob. + */ +class TestHFileDataBlock { + private static final long LATEST_TIMESTAMP = Long.MAX_VALUE; + private static final byte KEY_TYPE_PUT = (byte) 4; + // Column-family length (1) + timestamp (8) + key type (1) + the 2-byte row-length prefix. + private static final int KEY_SUFFIX_AND_PREFIX_LENGTH = 12; + + @Test + void writesFullHBaseKeyValuePerRecord() throws IOException { + HFileDataBlock block = + HFileDataBlock.createDataBlockToWrite(HFileContext.builder().build(), -1L); + // Two records of different key/value lengths to verify the framing repeats correctly. + block.add(utf8("key1"), utf8("value1")); + block.add(utf8("k22"), utf8("v2")); + + ByteBuffer buf = block.getUncompressedBlockDataToWrite(); + assertDataEntry(buf, "key1", "value1"); + assertDataEntry(buf, "k22", "v2"); + assertEquals(0, buf.remaining(), "unexpected trailing bytes after the last record"); + } + + private static void assertDataEntry(ByteBuffer buf, String key, String value) { + int rowLength = utf8(key).length; + int valueLength = utf8(value).length; + assertEquals(rowLength + KEY_SUFFIX_AND_PREFIX_LENGTH, buf.getInt(), "key length for " + key); + assertEquals(valueLength, buf.getInt(), "value length for " + key); + assertEquals((short) rowLength, buf.getShort(), "row length for " + key); + assertEquals(key, readString(buf, rowLength), "row for " + key); + assertEquals((byte) 0, buf.get(), "column-family length for " + key); + assertEquals(LATEST_TIMESTAMP, buf.getLong(), "timestamp for " + key); + assertEquals(KEY_TYPE_PUT, buf.get(), "key type for " + key); + assertEquals(value, readString(buf, valueLength), "value for " + key); + assertEquals((byte) 0, buf.get(), "MVCC version for " + key); + } + + private static byte[] utf8(String s) { + return s.getBytes(StandardCharsets.UTF_8); + } + + private static String readString(ByteBuffer buf, int length) { + byte[] bytes = new byte[length]; + buf.get(bytes); + return new String(bytes, StandardCharsets.UTF_8); + } +} diff --git a/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileReadCompatibility.java b/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileReadCompatibility.java index 8ca7ee7e5eabf..b4130a1afda29 100644 --- a/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileReadCompatibility.java +++ b/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileReadCompatibility.java @@ -22,6 +22,7 @@ import org.apache.hudi.io.ByteArraySeekableDataInputStream; import org.apache.hudi.io.ByteBufferBackedInputStream; import org.apache.hudi.io.SeekableDataInputStream; +import org.apache.hudi.io.compress.CompressionCodec; import org.apache.hudi.io.util.IOUtils; import org.apache.hadoop.conf.Configuration; @@ -29,19 +30,20 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellComparatorImpl; +import org.apache.hadoop.hbase.HConstants; 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.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.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.EnumSource; +import java.io.ByteArrayOutputStream; import java.io.DataOutputStream; import java.io.IOException; import java.net.URISyntaxException; @@ -57,6 +59,11 @@ import static org.apache.hudi.io.util.FileIOUtils.readAsByteArray; import static org.apache.hudi.io.util.IOUtils.readInt; import static org.apache.hudi.io.util.IOUtils.toBytes; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; class TestHFileReadCompatibility { // Test data - simple key-value pairs @@ -68,6 +75,11 @@ class TestHFileReadCompatibility { new TestRecord("row5", "value5") ); + // Small block size + many records => many data blocks => block-boundary keys land in the root + // index, which is exactly where the HBase point-lookup parsing happened. + private static final int MULTI_BLOCK_RECORDS = 2000; + private static final int SMALL_BLOCK_SIZE = 512; + @ParameterizedTest @CsvSource({ "/hfile/hudi-generated.hfile,/hfile/hbase-generated.hfile", @@ -78,8 +90,8 @@ void testHFileReadCompatibility(String hudiFilePath, String hbaseFilePath) throw org.apache.hadoop.hbase.io.hfile.HFile.Reader hbaseReader = createHBaseHFileReaderFromResource(hbaseFilePath)) { // Validate number of entries. - Assertions.assertEquals(5, hudiReader.getNumKeyValueEntries()); - Assertions.assertEquals(5, hbaseReader.getEntries()); + assertEquals(5, hudiReader.getNumKeyValueEntries()); + assertEquals(5, hbaseReader.getEntries()); // Validate data block content. hudiReader.seekTo(); HFileScanner scanner = hbaseReader.getScanner(true, true); @@ -89,60 +101,60 @@ void testHFileReadCompatibility(String hudiFilePath, String hbaseFilePath) throw org.apache.hudi.io.hfile.KeyValue keyValue = hudiReader.getKeyValue().get(); Cell cell = scanner.getCell(); // Ensure Hudi record is correct. - Assertions.assertEquals(TEST_RECORDS.get(i).key, keyValue.getKey().getContentInString()); + assertEquals(TEST_RECORDS.get(i).key, keyValue.getKey().getContentInString()); byte[] value = Arrays.copyOfRange( keyValue.getBytes(), keyValue.getValueOffset(), keyValue.getValueOffset() + keyValue.getValueLength()); - Assertions.assertArrayEquals(value, TEST_RECORDS.get(i).value.getBytes()); + assertArrayEquals(value, TEST_RECORDS.get(i).value.getBytes()); // Ensure Hbase record is correct. byte[] key = Arrays.copyOfRange( cell.getRowArray(), cell.getRowOffset(), cell.getRowOffset() + cell.getRowLength()); - Assertions.assertArrayEquals(TEST_RECORDS.get(i).key.getBytes(), key); + assertArrayEquals(TEST_RECORDS.get(i).key.getBytes(), key); value = Arrays.copyOfRange( cell.getValueArray(), cell.getValueOffset(), cell.getValueOffset() + cell.getValueLength()); - Assertions.assertArrayEquals(value, TEST_RECORDS.get(i).value.getBytes()); + assertArrayEquals(value, TEST_RECORDS.get(i).value.getBytes()); i++; } while (hudiReader.next() && scanner.next()); // Compare some meta information. // LAST KEY. - Assertions.assertTrue(hbaseReader.getHFileInfo().containsKey(HFileInfo.LAST_KEY.getBytes())); - Assertions.assertTrue(hudiReader.getMetaInfo(HFileInfo.LAST_KEY).isPresent()); + assertTrue(hbaseReader.getHFileInfo().containsKey(HFileInfo.LAST_KEY.getBytes())); + assertTrue(hudiReader.getMetaInfo(HFileInfo.LAST_KEY).isPresent()); // The last key value returned from hbase contains the extra fields, // e.g., column family, column qualifier, timestamp, key type, which is 10 more bytes. // Therefore, the last key value from hudi should be the prefix since hudi does not use these // extra fields. if (hudiReader.getMetaInfo(HFileInfo.LAST_KEY).get().length < hbaseReader.getHFileInfo().get(HFileInfo.LAST_KEY.getBytes()).length) { - Assertions.assertTrue(isPrefix( + assertTrue(isPrefix( hudiReader.getMetaInfo(HFileInfo.LAST_KEY).get(), hbaseReader.getHFileInfo().get(HFileInfo.LAST_KEY.getBytes()))); } else { - Assertions.assertTrue(isPrefix( + assertTrue(isPrefix( hbaseReader.getHFileInfo().get(HFileInfo.LAST_KEY.getBytes()), hudiReader.getMetaInfo(HFileInfo.LAST_KEY).get())); } // Average key length. - Assertions.assertTrue(hbaseReader.getHFileInfo().containsKey(HFileInfo.AVG_KEY_LEN.getBytes())); - Assertions.assertTrue(hudiReader.getMetaInfo(HFileInfo.AVG_KEY_LEN).isPresent()); + assertTrue(hbaseReader.getHFileInfo().containsKey(HFileInfo.AVG_KEY_LEN.getBytes())); + assertTrue(hudiReader.getMetaInfo(HFileInfo.AVG_KEY_LEN).isPresent()); // Average value length. - Assertions.assertTrue(hbaseReader.getHFileInfo().containsKey(HFileInfo.AVG_VALUE_LEN.getBytes())); - Assertions.assertTrue(hudiReader.getMetaInfo(HFileInfo.AVG_VALUE_LEN).isPresent()); - Assertions.assertTrue( + assertTrue(hbaseReader.getHFileInfo().containsKey(HFileInfo.AVG_VALUE_LEN.getBytes())); + assertTrue(hudiReader.getMetaInfo(HFileInfo.AVG_VALUE_LEN).isPresent()); + assertTrue( hbaseReader.getHFileInfo().getAvgValueLen() >= readInt(hudiReader.getMetaInfo(HFileInfo.AVG_VALUE_LEN).get(), 0)); // MVCC. - Assertions.assertTrue(hbaseReader.getHFileInfo().shouldIncludeMemStoreTS()); + assertTrue(hbaseReader.getHFileInfo().shouldIncludeMemStoreTS()); // Note that MemStoreTS is not set. - Assertions.assertFalse(hbaseReader.getHFileInfo().isDecodeMemstoreTS()); - Assertions.assertTrue(hudiReader.getMetaInfo(KEY_VALUE_VERSION).isPresent()); - Assertions.assertTrue(hudiReader.getMetaInfo(HFileInfo.MAX_MVCC_TS_KEY).isPresent()); - Assertions.assertEquals(0L, + assertFalse(hbaseReader.getHFileInfo().isDecodeMemstoreTS()); + assertTrue(hudiReader.getMetaInfo(KEY_VALUE_VERSION).isPresent()); + assertTrue(hudiReader.getMetaInfo(HFileInfo.MAX_MVCC_TS_KEY).isPresent()); + assertEquals(0L, IOUtils.readLong( hudiReader.getMetaInfo(HFileInfo.MAX_MVCC_TS_KEY).get(), 0)); } @@ -160,13 +172,13 @@ void testHbaseReaderSucceedsWhenKeyValueVersionIsSetTo1() throws IOException { // Create HBase HFile.Reader from the temporary file HFile.Reader reader = HFile.createReader(fs, new Path(tempFile.toString()), conf); byte[] keyValueVersion = reader.getHFileInfo().get(KEY_VALUE_VERSION.getBytes()); - Assertions.assertEquals(1, IOUtils.readInt(keyValueVersion, 0)); + assertEquals(1, IOUtils.readInt(keyValueVersion, 0)); // Values from trailer still works. - Assertions.assertEquals(5, reader.getEntries()); + assertEquals(5, reader.getEntries()); // Scanning the file succeeds. HFileScanner scanner = reader.getScanner(true, true); scanner.seekTo(); - Assertions.assertDoesNotThrow(() -> { + assertDoesNotThrow(() -> { int i = 0; do { Cell cell = scanner.getCell(); @@ -174,12 +186,145 @@ void testHbaseReaderSucceedsWhenKeyValueVersionIsSetTo1() throws IOException { cell.getRowArray(), cell.getRowOffset(), cell.getRowOffset() + cell.getRowLength()); - Assertions.assertArrayEquals(TEST_RECORDS.get(i).key.getBytes(), key); + assertArrayEquals(TEST_RECORDS.get(i).key.getBytes(), key); i++; } while (scanner.next()); }); } + /** + * Validates the block-index key encoding in the HFile: an HBase reader point-looks-up every key + * (including the block-boundary keys stored in the root index) in a native-written multi-block + * file and gets an exact match with the correct value. + */ + @ParameterizedTest + @EnumSource(value = CompressionCodec.class, names = {"NONE", "GZIP"}) + void hbaseReaderPointLooksUpEveryKeyInNativeMultiBlockFile(CompressionCodec codec) + throws IOException { + byte[] data = writeMultiBlockHudiHFile(MULTI_BLOCK_RECORDS, SMALL_BLOCK_SIZE, codec); + try (HFile.Reader reader = createHBaseHFileReader(data)) { + int blocks = reader.getTrailer().getDataIndexCount(); + assertTrue(blocks > 1, "expected a multi-block file; got " + blocks); + assertEquals(MULTI_BLOCK_RECORDS, reader.getEntries()); + HFileScanner scanner = reader.getScanner(true, true); + for (int i = 0; i < MULTI_BLOCK_RECORDS; i++) { + KeyValue probe = new KeyValue(Bytes.toBytes(key(i)), null, null, null); + assertEquals(0, scanner.seekTo(probe), "expected exact match for " + key(i)); + Cell cell = scanner.getCell(); + assertEquals(key(i), + Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())); + assertEquals(value(i), + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())); + } + } + } + + /** Validates the block-index key encoding parses as a {@code KeyValue} via {@code midKey()}. */ + @Test + void hbaseReaderMidKeyParsesNativeBlockIndexKey() throws IOException { + byte[] data = writeMultiBlockHudiHFile(MULTI_BLOCK_RECORDS, SMALL_BLOCK_SIZE, CompressionCodec.NONE); + try (HFile.Reader reader = createHBaseHFileReader(data)) { + assertTrue(reader.midKey().isPresent()); + } + } + + /** + * Byte comparison under both NONE and GZIP: given identical records, the native writer and the + * HBase writer produce cells that an HBase reader sees as byte-identical (key bytes and value + * bytes), establishing that the native writer emits HBase-format cells. + */ + @ParameterizedTest + @EnumSource(value = CompressionCodec.class, names = {"NONE", "GZIP"}) + void nativeAndHBaseWrittenCellsAreByteIdenticalUnderHBaseReader(CompressionCodec codec) + throws IOException { + byte[] nativeData = writeMultiBlockHudiHFile(MULTI_BLOCK_RECORDS, SMALL_BLOCK_SIZE, codec); + byte[] hbaseData = + writeMultiBlockHBaseHFile(MULTI_BLOCK_RECORDS, SMALL_BLOCK_SIZE, hbaseAlgo(codec)); + try (HFile.Reader nativeReader = createHBaseHFileReader(nativeData); + HFile.Reader hbaseReader = createHBaseHFileReader(hbaseData)) { + assertEquals(hbaseReader.getEntries(), nativeReader.getEntries()); + HFileScanner ns = nativeReader.getScanner(true, true); + HFileScanner hs = hbaseReader.getScanner(true, true); + assertTrue(ns.seekTo()); + assertTrue(hs.seekTo()); + int compared = 0; + boolean nativeHasNext; + boolean hbaseHasNext; + do { + Cell nativeCell = ns.getCell(); + Cell hbaseCell = hs.getCell(); + assertArrayEquals(keyBytes(nativeCell), keyBytes(hbaseCell), + "KeyValue key bytes differ at record " + compared); + assertArrayEquals(valueBytes(nativeCell), valueBytes(hbaseCell), + "value bytes differ at record " + compared); + compared++; + nativeHasNext = ns.next(); + hbaseHasNext = hs.next(); + } while (nativeHasNext && hbaseHasNext); + assertEquals(MULTI_BLOCK_RECORDS, compared); + assertFalse(nativeHasNext, "native scanner had extra cells"); + assertFalse(hbaseHasNext, "hbase scanner had extra cells"); + } + } + + /** + * Cross-reader equivalence: the same native-written multi-block file reads identically through + * the native hudi-io reader and the HBase reader (same rows and values, in order). + */ + @Test + void nativeWrittenFileReadsIdenticallyByBothReaders() throws IOException { + byte[] data = writeMultiBlockHudiHFile(MULTI_BLOCK_RECORDS, SMALL_BLOCK_SIZE, CompressionCodec.NONE); + try (HFileReader nativeReader = createHFileReader(data); + HFile.Reader hbaseReader = createHBaseHFileReader(data)) { + assertEquals(MULTI_BLOCK_RECORDS, hbaseReader.getEntries()); + nativeReader.seekTo(); + HFileScanner hbaseScanner = hbaseReader.getScanner(true, true); + assertTrue(hbaseScanner.seekTo()); + for (int i = 0; i < MULTI_BLOCK_RECORDS; i++) { + org.apache.hudi.io.hfile.KeyValue nativeKv = nativeReader.getKeyValue().get(); + Cell hbaseCell = hbaseScanner.getCell(); + assertEquals(key(i), nativeKv.getKey().getContentInString()); + assertEquals(key(i), + Bytes.toString(hbaseCell.getRowArray(), hbaseCell.getRowOffset(), hbaseCell.getRowLength())); + byte[] nativeValue = Arrays.copyOfRange(nativeKv.getBytes(), nativeKv.getValueOffset(), + nativeKv.getValueOffset() + nativeKv.getValueLength()); + assertArrayEquals(value(i).getBytes(StandardCharsets.UTF_8), nativeValue); + assertArrayEquals(nativeValue, valueBytes(hbaseCell)); + if (i < MULTI_BLOCK_RECORDS - 1) { + assertTrue(nativeReader.next()); + assertTrue(hbaseScanner.next()); + } + } + } + } + + /** + * The data block's previous-block-offset header must be correct: the HBase reader uses it in + * {@link HFileScanner#seekBefore} to step back to the prior block when the target lands on a + * block's first key. With a wrong offset (e.g. the block's own offset) the HBase reader re-reads + * the same block and throws {@code IllegalStateException} at every block boundary. On a + * native-written multi-block file, {@code seekBefore(key(i))} must return {@code key(i - 1)} for + * every record. The native reader is forward-only and never reads this field. + */ + @Test + void hbaseReaderSeekBeforeWorksOnNativeMultiBlockFile() throws IOException { + byte[] data = writeMultiBlockHudiHFile(MULTI_BLOCK_RECORDS, SMALL_BLOCK_SIZE, CompressionCodec.NONE); + try (HFile.Reader hbaseReader = createHBaseHFileReader(data)) { + assertTrue(hbaseReader.getTrailer().getDataIndexCount() > 1, + "precondition: the file must have multiple data blocks for seekBefore to cross boundaries"); + HFileScanner scanner = hbaseReader.getScanner(true, true); + for (int i = 1; i < MULTI_BLOCK_RECORDS; i++) { + KeyValue probe = new KeyValue(key(i).getBytes(StandardCharsets.UTF_8), + new byte[0], new byte[0], HConstants.LATEST_TIMESTAMP, new byte[0]); + assertTrue(scanner.seekBefore(probe), "seekBefore should find a key before " + key(i)); + Cell c = scanner.getCell(); + assertEquals(key(i - 1), + Bytes.toString(c.getRowArray(), c.getRowOffset(), c.getRowLength()), + "seekBefore(" + key(i) + ") must land on the immediately preceding key"); + } + } + } + static boolean isPrefix(byte[] prefix, byte[] array) { if (prefix.length > array.length) { return false; // can't be prefix if longer @@ -193,39 +338,25 @@ static boolean isPrefix(byte[] prefix, byte[] array) { } static HFileReader createHFileReaderFromResource(String fileName) throws IOException { - // Read HFile data from resources - byte[] hfileData = readHFileFromResources(fileName); - // Convert to ByteBuffer - ByteBuffer buffer = ByteBuffer.wrap(hfileData); - // Create SeekableDataInputStream + return createHFileReader(readHFileFromResources(fileName)); + } + + static HFileReader createHFileReader(byte[] hfileData) { SeekableDataInputStream inputStream = new ByteArraySeekableDataInputStream( - new ByteBufferBackedInputStream(buffer) - ); - // Create and return HFileReaderImpl + new ByteBufferBackedInputStream(ByteBuffer.wrap(hfileData))); return new HFileReaderImpl(inputStream, hfileData.length); } static HFile.Reader createHBaseHFileReaderFromResource(String fileName) throws IOException { - // Read HFile data from resources - byte[] hfileData = readHFileFromResources(fileName); - // Create a temporary file to write the HFile data + return createHBaseHFileReader(readHFileFromResources(fileName)); + } + + static HFile.Reader createHBaseHFileReader(byte[] hfileData) throws IOException { Path tempFile = new Path(Files.createTempFile("hbase_hfile_", ".hfile").toString()); - try { - // Write the byte array to temporary file - Files.write(Paths.get(tempFile.toString()), hfileData); - // Create Hadoop Configuration and FileSystem - Configuration conf = new Configuration(); - FileSystem fs = FileSystem.get(conf); - // Create HBase HFile.Reader from the temporary file - HFile.Reader reader = HFile.createReader(fs, new Path(tempFile.toString()), conf); - // Note: The temporary file will be cleaned up when the reader is closed - // or you can manually delete it after use - return reader; - } catch (IOException e) { - // Clean up temp file if creation fails - Files.deleteIfExists(Paths.get(tempFile.toString())); - throw e; - } + Files.write(Paths.get(tempFile.toString()), hfileData); + Configuration conf = new Configuration(); + FileSystem fs = FileSystem.get(conf); + return HFile.createReader(fs, tempFile, conf); } private static byte[] readHFileFromResources(String filename) throws IOException { @@ -250,7 +381,7 @@ private void writeHFileWithHbase(Path filePath) throws IOException { FileSystem fs = FileSystem.get(conf); // Create HFile context with appropriate settings - HFileContext context = new HFileContextBuilder() + org.apache.hadoop.hbase.io.hfile.HFileContext context = new HFileContextBuilder() .withBlockSize(64 * 1024) .withCompression(Compression.Algorithm.NONE) .withCellComparator(CellComparatorImpl.COMPARATOR) @@ -279,7 +410,7 @@ private void writeHFileWithHudi(Path filePath) throws IOException { } private void writeHFileWithHudi(Path filePath, int keyValueVersion) throws IOException { - org.apache.hudi.io.hfile.HFileContext context = org.apache.hudi.io.hfile.HFileContext.builder() + HFileContext context = HFileContext.builder() .blockSize(64 * 1024) .build(); try (DataOutputStream outputStream = new DataOutputStream( @@ -297,6 +428,67 @@ private void writeHFileWithHudi(Path filePath, int keyValueVersion) throws IOExc } } + private static String key(int i) { + return String.format("key%06d", i); + } + + private static String value(int i) { + return "value-" + i; + } + + private static Compression.Algorithm hbaseAlgo(CompressionCodec codec) { + return codec == CompressionCodec.GZIP ? Compression.Algorithm.GZ : Compression.Algorithm.NONE; + } + + private static byte[] writeMultiBlockHudiHFile(int numRecords, int blockSize, CompressionCodec codec) + throws IOException { + HFileContext context = HFileContext.builder() + .blockSize(blockSize).compressionCodec(codec).build(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (HFileWriter writer = new HFileWriterImpl(context, baos)) { + for (int i = 0; i < numRecords; i++) { + writer.append(key(i), value(i).getBytes(StandardCharsets.UTF_8)); + } + } + return baos.toByteArray(); + } + + // Same cells the native writer emits: family length 0, timestamp = LATEST, type = Put. + private static byte[] writeMultiBlockHBaseHFile(int numRecords, int blockSize, + Compression.Algorithm algo) throws IOException { + Configuration conf = new Configuration(); + FileSystem fs = FileSystem.getLocal(conf); + Path path = new Path(Files.createTempFile("hbase_write_", ".hfile").toString()); + org.apache.hadoop.hbase.io.hfile.HFileContext context = new HFileContextBuilder() + .withBlockSize(blockSize) + .withCompression(algo) + .withCellComparator(CellComparatorImpl.COMPARATOR) + .withIncludesMvcc(true) + .build(); + try (HFile.Writer writer = HFile.getWriterFactory(conf, new CacheConfig(conf)) + .withPath(fs, path).withFileContext(context).create()) { + for (int i = 0; i < numRecords; i++) { + writer.append(new KeyValue(Bytes.toBytes(key(i)), new byte[0], new byte[0], + HConstants.LATEST_TIMESTAMP, value(i).getBytes(StandardCharsets.UTF_8))); + } + } + return Files.readAllBytes(Paths.get(path.toString())); + } + + private static byte[] keyBytes(Cell c) { + KeyValue kv = new KeyValue(c.getRowArray(), c.getRowOffset(), c.getRowLength(), + c.getFamilyArray(), c.getFamilyOffset(), c.getFamilyLength(), + c.getQualifierArray(), c.getQualifierOffset(), c.getQualifierLength(), + c.getTimestamp(), KeyValue.Type.codeToType(c.getTypeByte()), + c.getValueArray(), c.getValueOffset(), c.getValueLength()); + return Arrays.copyOfRange(kv.getKey(), 0, kv.getKeyLength()); + } + + private static byte[] valueBytes(Cell c) { + return Arrays.copyOfRange(c.getValueArray(), c.getValueOffset(), + c.getValueOffset() + c.getValueLength()); + } + // Simple test record class private static class TestRecord { final String key; diff --git a/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileRootIndexBlock.java b/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileRootIndexBlock.java new file mode 100644 index 0000000000000..bf24801ea2580 --- /dev/null +++ b/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileRootIndexBlock.java @@ -0,0 +1,80 @@ +/* + * 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.io.hfile; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Validates the exact on-disk bytes the root index block writer emits for each entry. An entry is + * {@code [8-byte offset][4-byte size][varint keyLen][2-byte rowLen][row][1-byte cfLen=0] + * [8-byte ts=LATEST][1-byte type=Put]}: the block-index "first key" is a full HBase KeyValue key, + * byte-identical to the data block's key, so an HBase reader can point-look-up the block index. + */ +class TestHFileRootIndexBlock { + private static final long LATEST_TIMESTAMP = Long.MAX_VALUE; + private static final byte KEY_TYPE_PUT = (byte) 4; + // Column-family length (1) + timestamp (8) + key type (1) + the 2-byte row-length prefix. + private static final int KEY_SUFFIX_AND_PREFIX_LENGTH = 12; + + @Test + void writesFullHBaseKeyValueKeyPerEntry() throws IOException { + HFileRootIndexBlock block = + HFileRootIndexBlock.createRootIndexBlockToWrite(HFileContext.builder().build()); + // Two entries of different key lengths, offsets, and sizes. + block.add(utf8("key1"), 0L, 100); + block.add(utf8("key0002"), 100L, 250); + + ByteBuffer buf = block.getUncompressedBlockDataToWrite(); + assertIndexEntry(buf, "key1", 0L, 100); + assertIndexEntry(buf, "key0002", 100L, 250); + assertEquals(0, buf.remaining(), "unexpected trailing bytes after the last entry"); + } + + private static void assertIndexEntry(ByteBuffer buf, String key, long offset, int size) { + int rowLength = utf8(key).length; + assertEquals(offset, buf.getLong(), "offset for " + key); + assertEquals(size, buf.getInt(), "size for " + key); + // The key length is a Hadoop WritableUtils VarInt; it is a single byte here because + // rowLength + 12 < 128 for these keys (multi-byte VarInt is covered by long-key read tests). + assertEquals(rowLength + KEY_SUFFIX_AND_PREFIX_LENGTH, buf.get() & 0xff, + "varint key length for " + key); + assertEquals((short) rowLength, buf.getShort(), "row length for " + key); + assertEquals(key, readString(buf, rowLength), "row for " + key); + assertEquals((byte) 0, buf.get(), "column-family length for " + key); + assertEquals(LATEST_TIMESTAMP, buf.getLong(), "timestamp for " + key); + assertEquals(KEY_TYPE_PUT, buf.get(), "key type for " + key); + } + + private static byte[] utf8(String s) { + return s.getBytes(StandardCharsets.UTF_8); + } + + private static String readString(ByteBuffer buf, int length) { + byte[] bytes = new byte[length]; + buf.get(bytes); + return new String(bytes, StandardCharsets.UTF_8); + } +} diff --git a/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileWriter.java b/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileWriter.java index a4488351c7e24..23e8d70079611 100644 --- a/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileWriter.java +++ b/hudi-io/src/test/java/org/apache/hudi/io/hfile/TestHFileWriter.java @@ -22,11 +22,27 @@ import org.apache.hudi.io.ByteArraySeekableDataInputStream; import org.apache.hudi.io.ByteBufferBackedInputStream; import org.apache.hudi.io.SeekableDataInputStream; - -import lombok.extern.slf4j.Slf4j; +import org.apache.hudi.io.hfile.protobuf.generated.HFileProtos; +import org.apache.hudi.io.util.IOUtils; + +import com.google.protobuf.CodedOutputStream; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.hbase.CellComparatorImpl; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.io.compress.Compression; +import org.apache.hadoop.hbase.io.hfile.CacheConfig; +import org.apache.hadoop.hbase.io.hfile.HFile; +import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder; +import org.apache.hadoop.io.DataOutputBuffer; +import org.apache.hadoop.io.WritableUtils; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; @@ -36,25 +52,58 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.nio.file.StandardOpenOption; +import java.util.ArrayList; import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; import java.util.Random; import java.util.stream.Collectors; +import static org.apache.hudi.io.hfile.HFileBlock.HFILEBLOCK_HEADER_SIZE; import static org.apache.hudi.io.hfile.HFileBlockType.DATA; +import static org.apache.hudi.io.hfile.HFileBlockType.FILE_INFO; +import static org.apache.hudi.io.hfile.HFileBlockType.META; +import static org.apache.hudi.io.hfile.HFileBlockType.ROOT_INDEX; import static org.apache.hudi.io.hfile.HFileBlockType.TRAILER; import static org.apache.hudi.io.hfile.HFileInfo.AVG_KEY_LEN; import static org.apache.hudi.io.hfile.HFileInfo.AVG_VALUE_LEN; import static org.apache.hudi.io.hfile.HFileInfo.KEY_VALUE_VERSION; import static org.apache.hudi.io.hfile.HFileInfo.LAST_KEY; import static org.apache.hudi.io.hfile.HFileInfo.MAX_MVCC_TS_KEY; +import static org.apache.hudi.io.hfile.HFileTrailer.TRAILER_SIZE; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertTrue; -@Slf4j class TestHFileWriter { + private static final Logger LOG = LoggerFactory.getLogger(TestHFileWriter.class); private static final String TEST_FILE = "test.hfile"; private static final HFileContext CONTEXT = HFileContext.builder().build(); + // Golden bytes (NONE compression, fixed input) that lock the on-disk encoding of the data block + // and the root block-index block. Update intentionally ONLY after reviewing HBase-reader + // compatibility, since any change here is a storage-format change. + private static final String GOLDEN_DATA_REGION_HEX = + "44415441424c4b2a000000610000005dffffffffffffffff00000040000000007e000000100000000600046b65793100" + + "7fffffffffffffff0476616c75653100000000100000000600046b657932007fffffffffffffff0476616c7565320000" + + "0000100000000600046b657933007fffffffffffffff0476616c7565330000000000"; + private static final String GOLDEN_ROOT_INDEX_BLOCK_HEX = + "494458524f4f5432000000210000001dffffffffffffffff00000040000000003e000000000000000000000082100004" + + "6b657931007fffffffffffffff0400000000"; + // Golden for a single block holding keys of different lengths whose first key (116 chars) lands in + // the multi-byte vint range (keyLength 128). Pinned for both the native and HBase writers. + private static final String GOLDEN_VARLEN_DATA_REGION_HEX = + "44415441424c4b2a000000d1000000cdffffffffffffffff0000004000000000ee000000800000000200746161616161" + + "616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161" + + "616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161" + + "616161616161616161616161616161007fffffffffffffff04763000000000140000000200086262626262626262007f" + + "ffffffffffffff047631000000001800000002000c636363636363636363636363007fffffffffffffff047632000000" + + "0000"; + private static final String GOLDEN_VARLEN_ROOT_INDEX_BLOCK_HEX = + "494458524f4f5432000000920000008effffffffffffffff0000004000000000af0000000000000000000000f28f8000" + + "746161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161" + + "616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161" + + "616161616161616161616161616161616161616161007fffffffffffffff0400000000"; @AfterEach public void tearDown() throws IOException { @@ -71,7 +120,7 @@ void testOverflow() throws Exception { validateHFileStructure(); // 4. Validate consistency with HFileReader. validateConsistencyWithHFileReader(); - log.info("All validations passed!"); + LOG.info("All validations passed!"); } @Test @@ -182,56 +231,417 @@ void testUniqueKeyLocation() throws IOException { ); reader.next(); } + + // Each data block's previous-block-offset header (8 bytes at offset 16) must chain back to + // the prior block, and -1 for the first, so an HBase reader's seekBefore can step back across + // blocks. The writer once set it to the block's own offset; this guards that regression. + byte[] file = Files.readAllBytes(Paths.get(testFile)); + List dataIndex = parseIndexBlock( + file, (int) trailer.getLoadOnOpenDataOffset(), trailer.getDataIndexCount()); + assertTrue(dataIndex.size() > 1, "test setup: expected multiple data blocks"); + for (int i = 0; i < dataIndex.size(); i++) { + long prevOffset = readLongBE(file, (int) dataIndex.get(i).offset + 16); + long expected = i == 0 ? -1L : dataIndex.get(i - 1).offset; + assertEquals(expected, prevOffset, "data block " + i + " previous-block offset"); + } } } + /** + * Format lock for the native writer's on-disk bytes. Single-block inputs are checked against the + * HBase writer (proving the writers agree on the encoding): short keys (single-byte index + * keyLength) and the 116-char boundary (keyLength 128, where Protobuf and Hadoop VInt diverge) are + * also pinned to a golden; 500- and 2000-char first keys (3-byte keyLength) are checked against + * HBase without a golden. The golden cases also validate the trailer and file info bytes. A single + * block is deliberate for cross-writer checks: across blocks HBase stores shortened index separator + * keys, so only a single full-key entry is byte-comparable. A multi-block native file then covers + * the meta block and the rest of the section byte layout and round-trips. + */ @Test - void testLongKeys() throws IOException { - // Test that HFile blocks with long keys (>= 126 chars) can be written and read correctly. - // This verifies the fix for the varint encoding mismatch in the root index block. - HFileContext context = new HFileContext.Builder().blockSize(100).build(); - String testFile = TEST_FILE; - int numRecords = 10; - // Generate keys longer than 126 characters to trigger multi-byte Hadoop VarInt encoding - // in the root index block. The varint encodes (key_content_length + 2), so content >= 126 - // produces a value >= 128 which requires 2+ bytes in Hadoop VarInt format. - char[] chars = new char[200]; - Arrays.fill(chars, 'a'); - String longPrefix = new String(chars); - try (DataOutputStream outputStream = - new DataOutputStream(Files.newOutputStream(Paths.get(testFile))); - HFileWriter writer = new HFileWriterImpl(context, outputStream)) { - for (int i = 0; i < numRecords; i++) { - String key = longPrefix + String.format("%04d", i); - writer.append(key, String.format("value%04d", i).getBytes()); + void writerBlockBytesAreStableFormatLock() throws Exception { + byte[][] vals = {bytes("v0"), bytes("v1"), bytes("v2")}; + assertSingleBlockBytesMatchGoldenAndHBase( + new String[] {"key1", "key2", "key3"}, + new byte[][] {bytes("value1"), bytes("value2"), bytes("value3")}, + GOLDEN_DATA_REGION_HEX, GOLDEN_ROOT_INDEX_BLOCK_HEX); + assertSingleBlockBytesMatchGoldenAndHBase( + new String[] {rep('a', 116), rep('b', 8), rep('c', 12)}, vals, + GOLDEN_VARLEN_DATA_REGION_HEX, GOLDEN_VARLEN_ROOT_INDEX_BLOCK_HEX); + // Larger first keys (3-byte index keyLength) checked against the HBase writer without a golden, + // which for multi-KB key bytes would add no signal. + assertSingleBlockBytesMatchHBase(new String[] {rep('a', 500), rep('b', 8), rep('c', 12)}, vals, null, null); + assertSingleBlockBytesMatchHBase(new String[] {rep('a', 2000), rep('b', 8), rep('c', 12)}, vals, null, null); + assertAllSectionsValidOnMultiBlockFile(); + } + + @Test + void getVarIntBytesRoundTripsThroughReadVarLong() { + // HFileIndexBlock.getVarIntBytes output must decode back via the reader's IOUtils.readVarLong. + int[] testValues = {0, 1, 127, 128, 146, 200, 255, 256, 300, 1000, 32080, 65535, 100000, + 2034958, 632492350, Integer.MAX_VALUE}; + for (int value : testValues) { + byte[] encoded = HFileIndexBlock.getVarIntBytes(value); + int size = IOUtils.decodeVarLongSizeOnDisk(encoded, 0); + assertEquals(encoded.length, size, "Size mismatch for value " + value); + long decoded = IOUtils.readVarLong(encoded, 0, size); + assertEquals(value, decoded, "Round-trip mismatch for value " + value); + } + } + + @Test + void getVarIntBytesMatchesKnownHadoopVectors() { + // Cross-check HFileIndexBlock.getVarIntBytes against known Hadoop WritableUtils VInt encodings. + assertEquals(1, HFileIndexBlock.getVarIntBytes(0).length); + assertEquals(0, HFileIndexBlock.getVarIntBytes(0)[0]); + assertEquals(1, HFileIndexBlock.getVarIntBytes(98).length); + assertEquals(98, HFileIndexBlock.getVarIntBytes(98)[0]); + + // Value 208 requires 2 bytes. + byte[] enc208 = HFileIndexBlock.getVarIntBytes(208); + assertEquals(2, enc208.length); + assertEquals(208, IOUtils.readVarLong(enc208, 0)); + + // Value 32080 requires 3 bytes. + byte[] enc32080 = HFileIndexBlock.getVarIntBytes(32080); + assertEquals(3, enc32080.length); + assertEquals(32080, IOUtils.readVarLong(enc32080, 0)); + } + + /** + * Pins a single-block file's data block and root index block bytes to a golden the HBase writer + * must also reproduce, and validates the remaining sections of the native file (trailer, file + * info, and the empty meta index). The file has no meta block: a meta block would shift the data + * region and HBase serializes it differently, breaking the byte-parity (the meta block is covered + * by {@link #assertAllSectionsValidOnMultiBlockFile()}). + */ + private static void assertSingleBlockBytesMatchGoldenAndHBase( + String[] keys, byte[][] values, String dataGolden, String rootGolden) throws Exception { + assertSingleBlockBytesMatchHBase(keys, values, dataGolden, rootGolden); + byte[] nativeFile = Files.readAllBytes(Paths.get(TEST_FILE)); + HFileProtos.TrailerProto trailer = validateTrailerSection(nativeFile, keys.length, 0); + int loadOnOpen = (int) trailer.getLoadOnOpenDataOffset(); + int metaIndexOffset = loadOnOpen + HFILEBLOCK_HEADER_SIZE + readIntBE(nativeFile, loadOnOpen + 8); + assertBytesEqual("empty meta index magic", ROOT_INDEX.getMagic(), + Arrays.copyOfRange(nativeFile, metaIndexOffset, metaIndexOffset + 8)); + validateFileInfoSection( + nativeFile, metaIndexOffset, (int) trailer.getFileInfoOffset(), keys[keys.length - 1]); + } + + /** + * Asserts the native and HBase writers produce identical single-block data and root index bytes + * (optionally also matching a pinned golden), and the root-index keyLength is Hadoop VInt. Used + * without a golden for large keys, where a hardcoded golden of multi-KB key bytes adds no signal. + */ + private static void assertSingleBlockBytesMatchHBase( + String[] keys, byte[][] values, String dataGolden, String rootGolden) throws Exception { + writeNativeFile(TEST_FILE, HFileContext.builder().build(), keys, values, null, null); + byte[] nativeFile = Files.readAllBytes(Paths.get(TEST_FILE)); + String[] nativeBlocks = dataAndRootIndexBlockHex(nativeFile); + String[] hbaseBlocks = dataAndRootIndexBlockHex(writeHBaseFile(keys, values)); + assertEquals(hbaseBlocks[0], nativeBlocks[0], "native vs HBase data block bytes"); + assertEquals(hbaseBlocks[1], nativeBlocks[1], "native vs HBase root index bytes"); + if (dataGolden != null) { + LOG.info("DATA_REGION_HEX={}", nativeBlocks[0]); + LOG.info("ROOT_INDEX_BLOCK_HEX={}", nativeBlocks[1]); + assertEquals(dataGolden, nativeBlocks[0], "data block bytes changed (storage-format change)"); + assertEquals(rootGolden, nativeBlocks[1], "root index bytes changed (storage-format change)"); + } + IndexEntry first = + parseIndexBlock(nativeFile, indexOf(nativeFile, ROOT_INDEX.getMagic()), 1).get(0); + assertVarIntIsHadoopNotProtobuf("root index keyLength", first); + } + + /** Returns {@code [dataRegionHex, rootIndexBlockHex]} for an HFile's raw bytes. */ + private static String[] dataAndRootIndexBlockHex(byte[] data) { + int idxRootOffset = indexOf(data, HFileBlockType.ROOT_INDEX.getMagic()); + assertTrue(idxRootOffset > 0, "root index block not found"); + // Root index block: 33-byte v3 block header + onDiskSizeWithoutHeader payload (at header + 8). + int onDiskSizeWithoutHeader = readIntBE(data, idxRootOffset + 8); + return new String[] { + hex(Arrays.copyOfRange(data, 0, idxRootOffset)), + hex(Arrays.copyOfRange(data, idxRootOffset, idxRootOffset + 33 + onDiskSizeWithoutHeader)) + }; + } + + /** Writes the given records with the HBase HFile writer, matching the native writer's settings. */ + private static byte[] writeHBaseFile(String[] keys, byte[][] values) throws IOException { + Configuration conf = new Configuration(); + FileSystem fs = FileSystem.getLocal(conf); + org.apache.hadoop.fs.Path path = + new org.apache.hadoop.fs.Path(Files.createTempFile("hbase_fixed_", ".hfile").toString()); + org.apache.hadoop.hbase.io.hfile.HFileContext context = new HFileContextBuilder() + .withBlockSize(1024 * 1024) + .withCompression(Compression.Algorithm.NONE) + .withChecksumType(org.apache.hadoop.hbase.util.ChecksumType.NULL) + .withCellComparator(CellComparatorImpl.COMPARATOR) + .withIncludesMvcc(true) + .build(); + try (HFile.Writer writer = HFile.getWriterFactory(conf, new CacheConfig(conf)) + .withPath(fs, path).withFileContext(context).create()) { + for (int i = 0; i < keys.length; i++) { + writer.append(new org.apache.hadoop.hbase.KeyValue( + keys[i].getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], + HConstants.LATEST_TIMESTAMP, values[i])); } } + return Files.readAllBytes(Paths.get(path.toString())); + } - // Validate that all records can be read back correctly. - try (FileChannel channel = FileChannel.open(Paths.get(testFile), StandardOpenOption.READ)) { + /** + * Validates the raw bytes of every HFile section on a multi-block native file with one meta block: + * scanned data blocks, the non-scanned meta block, the load-on-open root data index / meta index / + * file info, and the trailer, then reads every record back. Index keyLengths are asserted Hadoop + * {@code WritableUtils} VInt; the file info and trailer length prefixes protobuf-delimited. Each + * section is checked in its own helper so callers stay under the per-test assertion limit. + */ + private static void assertAllSectionsValidOnMultiBlockFile() throws IOException { + int numRecords = 8; + String[] keys = new String[numRecords]; + byte[][] values = new byte[numRecords][]; + for (int i = 0; i < numRecords; i++) { + keys[i] = makeKey(200, i); + values[i] = bytes("v" + i); + } + String metaKey = makeKey(200, 9999); + byte[] metaValue = bytes("bloom-filter-payload-bytes"); + // Small block size forces one record per block, so the root index has multiple entries. + writeNativeFile( + TEST_FILE, new HFileContext.Builder().blockSize(100).build(), keys, values, metaKey, metaValue); + byte[] file = Files.readAllBytes(Paths.get(TEST_FILE)); + + HFileProtos.TrailerProto trailer = validateTrailerSection(file, numRecords, 1); + int loadOnOpen = (int) trailer.getLoadOnOpenDataOffset(); + validateRootDataIndexSection(file, loadOnOpen, trailer.getDataIndexCount(), keys[0]); + int metaIndexOffset = loadOnOpen + HFILEBLOCK_HEADER_SIZE + readIntBE(file, loadOnOpen + 8); + validateMetaIndexAndBlock(file, metaIndexOffset, metaKey, metaValue); + validateFileInfoSection(file, metaIndexOffset, (int) trailer.getFileInfoOffset(), keys[numRecords - 1]); + assertAllRecordsReadBack(numRecords, keys); + } + + /** Validates the trailer magic, protobuf framing, fields, and HFile version; returns the proto. */ + private static HFileProtos.TrailerProto validateTrailerSection( + byte[] file, int numRecords, int metaIndexCount) throws IOException { + int trailerStart = file.length - TRAILER_SIZE; + assertBytesEqual("trailer magic", + TRAILER.getMagic(), Arrays.copyOfRange(file, trailerStart, trailerStart + 8)); + HFileProtos.TrailerProto trailer = HFileProtos.TrailerProto.parseDelimitedFrom( + new ByteArrayInputStream(file, trailerStart + 8, TRAILER_SIZE - 8)); + assertProtobufDelimitedFraming("trailer", file, trailerStart + 8, trailer.getSerializedSize()); + assertEquals(numRecords, trailer.getEntryCount()); + assertEquals(metaIndexCount, trailer.getMetaIndexCount()); + assertEquals(1, trailer.getNumDataIndexLevels()); + assertEquals(2, trailer.getCompressionCodec(), "compression codec must be NONE (2)"); + assertEquals("org.apache.hudi.io.storage.HoodieHBaseKVComparator", + trailer.getComparatorClassName()); + assertEquals(0, trailer.getFirstDataBlockOffset(), "scanned section starts at offset 0"); + // HFile version: last 4 bytes hold the major version (3). + assertEquals(3, readIntBE(file, file.length - 4), "HFile major version"); + return trailer; + } + + /** Validates the scanned-section start and every root data index entry. */ + private static void validateRootDataIndexSection( + byte[] file, int loadOnOpen, int dataIndexCount, String firstKey) throws IOException { + assertBytesEqual("first data block magic", DATA.getMagic(), Arrays.copyOfRange(file, 0, 8)); + assertBytesEqual("root index magic", + ROOT_INDEX.getMagic(), Arrays.copyOfRange(file, loadOnOpen, loadOnOpen + 8)); + List dataIndex = parseIndexBlock(file, loadOnOpen, dataIndexCount); + assertEquals(dataIndexCount, dataIndex.size(), "root index entry count"); + for (IndexEntry e : dataIndex) { + // keyLength must be Hadoop WritableUtils VInt (never protobuf varint), and each entry must + // point at a real DATA block whose on-disk size matches the entry's size. + assertVarIntIsHadoopNotProtobuf("data index keyLength", e); + assertBytesEqual("data block magic at index offset", + DATA.getMagic(), Arrays.copyOfRange(file, (int) e.offset, (int) e.offset + 8)); + assertEquals(HFILEBLOCK_HEADER_SIZE + readIntBE(file, (int) e.offset + 8), e.size, + "data index entry size must equal the on-disk size of the referenced block"); + } + assertEquals(firstKey, keyContentString(dataIndex.get(0).key), "first index entry key content"); + } + + /** Validates the meta index entry and the meta block it points at. */ + private static void validateMetaIndexAndBlock( + byte[] file, int metaIndexOffset, String metaKey, byte[] metaValue) throws IOException { + assertBytesEqual("meta index magic", + ROOT_INDEX.getMagic(), Arrays.copyOfRange(file, metaIndexOffset, metaIndexOffset + 8)); + IndexEntry metaEntry = parseIndexBlock(file, metaIndexOffset, 1).get(0); + assertVarIntIsHadoopNotProtobuf("meta index keyLength", metaEntry); + assertBytesEqual("meta index key bytes", metaKey.getBytes(StandardCharsets.UTF_8), metaEntry.key); + assertBytesEqual("meta block magic", + META.getMagic(), Arrays.copyOfRange(file, (int) metaEntry.offset, (int) metaEntry.offset + 8)); + int metaPayload = (int) metaEntry.offset + HFILEBLOCK_HEADER_SIZE; + assertBytesEqual("meta block payload == meta value", + metaValue, Arrays.copyOfRange(file, metaPayload, metaPayload + metaValue.length)); + } + + /** Validates the file info block offset, magic, PBUF framing, and decoded entries. */ + private static void validateFileInfoSection( + byte[] file, int metaIndexOffset, int fileInfoOffset, String lastKey) throws IOException { + int metaIndexBlockSize = HFILEBLOCK_HEADER_SIZE + readIntBE(file, metaIndexOffset + 8); + assertEquals(metaIndexOffset + metaIndexBlockSize, fileInfoOffset, + "trailer fileInfoOffset must point right after the meta index block"); + assertBytesEqual("file info magic", + FILE_INFO.getMagic(), Arrays.copyOfRange(file, fileInfoOffset, fileInfoOffset + 8)); + int pbufStart = fileInfoOffset + HFILEBLOCK_HEADER_SIZE; + assertBytesEqual("file info PBUF magic", + "PBUF".getBytes(StandardCharsets.UTF_8), Arrays.copyOfRange(file, pbufStart, pbufStart + 4)); + HFileProtos.InfoProto info = HFileProtos.InfoProto.parseDelimitedFrom( + new ByteArrayInputStream(file, pbufStart + 4, file.length - (pbufStart + 4))); + // File info uses protobuf-delimited framing (correct here, unlike the index keyLength fields). + assertProtobufDelimitedFraming("file info", file, pbufStart + 4, info.getSerializedSize()); + Map infoMap = new LinkedHashMap<>(); + for (HFileProtos.BytesBytesPair pair : info.getMapEntryList()) { + infoMap.put(new String(pair.getFirst().toByteArray(), StandardCharsets.UTF_8), + pair.getSecond().toByteArray()); + } + for (UTF8StringKey required : new UTF8StringKey[] { + LAST_KEY, KEY_VALUE_VERSION, MAX_MVCC_TS_KEY, AVG_KEY_LEN, AVG_VALUE_LEN}) { + assertTrue(infoMap.containsKey(utf8(required)), "file info must contain " + utf8(required)); + } + assertEquals(lastKey, keyContentString(infoMap.get(utf8(LAST_KEY))), + "LASTKEY content must be the last appended key"); + } + + /** Reads every record back through the native reader. */ + private static void assertAllRecordsReadBack(int numRecords, String[] keys) throws IOException { + try (FileChannel channel = FileChannel.open(Paths.get(TEST_FILE), StandardOpenOption.READ)) { ByteBuffer buf = channel.map(FileChannel.MapMode.READ_ONLY, 0, channel.size()); - SeekableDataInputStream inputStream = - new ByteArraySeekableDataInputStream(new ByteBufferBackedInputStream(buf)); - HFileReaderImpl reader = new HFileReaderImpl(inputStream, channel.size()); + HFileReaderImpl reader = new HFileReaderImpl( + new ByteArraySeekableDataInputStream(new ByteBufferBackedInputStream(buf)), channel.size()); reader.initializeMetadata(); assertEquals(numRecords, reader.getNumKeyValueEntries()); reader.seekTo(); for (int i = 0; i < numRecords; i++) { - KeyValue kv = reader.getKeyValue().get(); - String expectedKey = longPrefix + String.format("%04d", i); - assertEquals(expectedKey, kv.getKey().getContentInString()); - assertArrayEquals( - String.format("value%04d", i).getBytes(), - Arrays.copyOfRange( - kv.getBytes(), - kv.getValueOffset(), - kv.getValueOffset() + kv.getValueLength()) - ); + assertEquals(keys[i], reader.getKeyValue().get().getKey().getContentInString()); reader.next(); } } } + /** Writes the given records (and an optional single meta block) with the native HFile writer. */ + private static void writeNativeFile(String testFile, HFileContext context, String[] keys, + byte[][] values, String metaKey, byte[] metaValue) + throws IOException { + try (DataOutputStream outputStream = + new DataOutputStream(Files.newOutputStream(Paths.get(testFile))); + HFileWriter writer = new HFileWriterImpl(context, outputStream)) { + for (int i = 0; i < keys.length; i++) { + writer.append(keys[i], values[i]); + } + if (metaKey != null) { + writer.appendMetaInfo(metaKey, metaValue); + } + } + } + + /** A parsed block index entry: block offset, on-disk size, raw keyLength vint bytes, and key. */ + private static final class IndexEntry { + long offset; + int size; + byte[] keyLengthVarIntBytes; + int keyLen; + byte[] key; + } + + /** Parses {@code numEntries} block index entries starting at the given index block offset. */ + private static List parseIndexBlock(byte[] file, int blockStart, int numEntries) { + List entries = new ArrayList<>(); + int p = blockStart + HFILEBLOCK_HEADER_SIZE; + for (int i = 0; i < numEntries; i++) { + IndexEntry e = new IndexEntry(); + e.offset = readLongBE(file, p); + e.size = readIntBE(file, p + 8); + int varIntSize = IOUtils.decodeVarLongSizeOnDisk(file, p + 12); + e.keyLen = (int) IOUtils.readVarLong(file, p + 12, varIntSize); + e.keyLengthVarIntBytes = Arrays.copyOfRange(file, p + 12, p + 12 + varIntSize); + e.key = Arrays.copyOfRange(file, p + 12 + varIntSize, p + 12 + varIntSize + e.keyLen); + entries.add(e); + p += 12 + varIntSize + e.keyLen; + } + return entries; + } + + /** + * Asserts that an index entry's keyLength bytes equal Hadoop {@code WritableUtils} VInt (the + * encoding the reader and HBase decode), and for values >= 128, that they are NOT protobuf + * varint (the pre-fix encoding that produced negative key lengths in HBase's reader). + */ + private static void assertVarIntIsHadoopNotProtobuf(String message, IndexEntry e) + throws IOException { + assertBytesEqual(message + " must be Hadoop WritableUtils VInt", + hadoopWritableUtilsVInt(e.keyLen), e.keyLengthVarIntBytes); + if (e.keyLen >= 128) { + assertNotEquals(hex(protobufVarInt(e.keyLen)), hex(e.keyLengthVarIntBytes), + message + " must NOT be protobuf varint for values >= 128 (the original bug)"); + } + } + + /** Asserts the on-disk length prefix at {@code offset} is a protobuf varint of {@code size}. */ + private static void assertProtobufDelimitedFraming(String section, byte[] file, int offset, + int serializedSize) throws IOException { + byte[] expectedPrefix = protobufVarInt(serializedSize); + assertBytesEqual(section + " must use protobuf-delimited framing", + expectedPrefix, Arrays.copyOfRange(file, offset, offset + expectedPrefix.length)); + } + + /** Reference Hadoop {@code WritableUtils} VInt encoder (the encoding HBase's reader expects). */ + private static byte[] hadoopWritableUtilsVInt(long value) throws IOException { + DataOutputBuffer out = new DataOutputBuffer(); + WritableUtils.writeVInt(out, (int) value); + return Arrays.copyOf(out.getData(), out.getLength()); + } + + /** Reference protobuf varint encoder (the pre-fix, HBase-incompatible encoding for the index). */ + private static byte[] protobufVarInt(int value) throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CodedOutputStream cos = CodedOutputStream.newInstance(baos); + cos.writeUInt32NoTag(value); + cos.flush(); + return baos.toByteArray(); + } + + /** Extracts the row (key content) bytes from a serialized HBase KeyValue key, as a String. */ + private static String keyContentString(byte[] keyValueKey) { + int rowLength = ((keyValueKey[0] & 0xff) << 8) | (keyValueKey[1] & 0xff); + return new String(keyValueKey, 2, rowLength, StandardCharsets.UTF_8); + } + + /** Builds a length-{@code len} key that sorts by {@code i} and is unique. */ + private static String makeKey(int len, int i) { + String suffix = String.format("%04d", i); + StringBuilder sb = new StringBuilder(len); + for (int k = 0; k < len - suffix.length(); k++) { + sb.append('a'); + } + return sb.append(suffix).toString(); + } + + private static String utf8(UTF8StringKey key) { + return new String(key.getBytes(), StandardCharsets.UTF_8); + } + + private static void assertBytesEqual(String message, byte[] expected, byte[] actual) { + assertEquals(hex(expected), hex(actual), message); + } + + private static String rep(char c, int n) { + char[] a = new char[n]; + Arrays.fill(a, c); + return new String(a); + } + + private static byte[] bytes(String s) { + return s.getBytes(StandardCharsets.UTF_8); + } + + private static long readLongBE(byte[] b, int off) { + long v = 0; + for (int i = 0; i < 8; i++) { + v = (v << 8) | (b[off + i] & 0xffL); + } + return v; + } + private static void writeTestFile() throws Exception { try ( DataOutputStream outputStream = @@ -246,7 +656,9 @@ private static void writeTestFile() throws Exception { private static void validateHFileSize() throws IOException { Path path = Paths.get(TEST_FILE); long actualSize = Files.size(path); - long expectedSize = 4537; + // Each root block-index entry carries the 10-byte HBase KeyValue suffix (column-family + // length + timestamp + key type). This file has one index entry, so the size grows by 10. + long expectedSize = 4547; assertEquals(expectedSize, actualSize); } @@ -339,6 +751,32 @@ private static void assertArrayEquals(byte[] expected, byte[] actual) { } } + private static int indexOf(byte[] haystack, byte[] needle) { + outer: + for (int i = 0; i + needle.length <= haystack.length; i++) { + for (int j = 0; j < needle.length; j++) { + if (haystack[i + j] != needle[j]) { + continue outer; + } + } + return i; + } + return -1; + } + + private static int readIntBE(byte[] b, int off) { + return ((b[off] & 0xff) << 24) | ((b[off + 1] & 0xff) << 16) + | ((b[off + 2] & 0xff) << 8) | (b[off + 3] & 0xff); + } + + private static String hex(byte[] b) { + StringBuilder sb = new StringBuilder(b.length * 2); + for (byte x : b) { + sb.append(Character.forDigit((x >> 4) & 0xf, 16)).append(Character.forDigit(x & 0xf, 16)); + } + return sb.toString(); + } + public static String generateRandomStringStream(int length) { String characters = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"; Random random = new Random(); diff --git a/hudi-io/src/test/java/org/apache/hudi/io/util/TestIOUtils.java b/hudi-io/src/test/java/org/apache/hudi/io/util/TestIOUtils.java index a100446428e1f..bc20d47a860b7 100644 --- a/hudi-io/src/test/java/org/apache/hudi/io/util/TestIOUtils.java +++ b/hudi-io/src/test/java/org/apache/hudi/io/util/TestIOUtils.java @@ -92,39 +92,6 @@ public void testDecodeVariableLengthNumber(byte[] bytes, long expectedNumber) th assertEquals(expectedNumber < 0, IOUtils.isNegativeVarLong(bytes[0])); } - @Test - public void testWriteVarIntRoundTrip() { - // Verify that writeVarInt produces bytes that readVarLong can decode correctly - int[] testValues = {0, 1, 127, 128, 146, 200, 255, 256, 300, 1000, 32080, 65535, 100000, - 2034958, 632492350, Integer.MAX_VALUE}; - for (int value : testValues) { - byte[] encoded = IOUtils.writeVarInt(value); - int size = IOUtils.decodeVarLongSizeOnDisk(encoded, 0); - assertEquals(encoded.length, size, "Size mismatch for value " + value); - long decoded = IOUtils.readVarLong(encoded, 0, size); - assertEquals(value, decoded, "Round-trip mismatch for value " + value); - } - } - - @Test - public void testWriteVarIntMatchesExistingTestVectors() { - // Cross-check writeVarInt against known Hadoop VarLong encoding from the existing test data - assertEquals(1, IOUtils.writeVarInt(0).length); - assertEquals(0, IOUtils.writeVarInt(0)[0]); - assertEquals(1, IOUtils.writeVarInt(98).length); - assertEquals(98, IOUtils.writeVarInt(98)[0]); - - // Value 208 requires 2 bytes - byte[] enc208 = IOUtils.writeVarInt(208); - assertEquals(2, enc208.length); - assertEquals(208, IOUtils.readVarLong(enc208, 0)); - - // Value 32080 requires 3 bytes - byte[] enc32080 = IOUtils.writeVarInt(32080); - assertEquals(3, enc32080.length); - assertEquals(32080, IOUtils.readVarLong(enc32080, 0)); - } - @Test public void testByteArrayCompareTo() { byte[] bytes1 = new byte[] {(byte) 0x9b, 0, 0x18, 0x65, 0x2e, (byte) 0xf3};