hudi-agent commented on code in PR #19200:
URL: https://github.com/apache/hudi/pull/19200#discussion_r3533010672
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieNativeLogFormatWriter.java:
##########
@@ -237,23 +234,12 @@ private Map<HeaderMetadataType, String>
addRecordPositionsIfRequired(
return header;
}
- Set<Long> positionSet = new HashSet<>(recordPositions.size());
- long previousPosition = Long.MIN_VALUE;
- for (Long position : recordPositions) {
- // RECORD_POSITIONS is encoded as a bitmap and decoded in ascending
order. Native log records
- // are read back in file write order, so only publish positions when
both orders match.
- if (!HoodieRecordLocation.isPositionValid(position) || position <=
previousPosition) {
- Map<HeaderMetadataType, String> updatedHeader = new HashMap<>(header);
-
updatedHeader.remove(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS);
- return updatedHeader;
- }
- previousPosition = position;
- positionSet.add(position);
- }
-
Map<HeaderMetadataType, String> updatedHeader = new HashMap<>(header);
- if (positionSet.size() == recordPositions.size()) {
- updatedHeader.put(HeaderMetadataType.RECORD_POSITIONS,
LogReaderUtils.encodePositions(positionSet));
+ Option<String> encodedRecordPositions =
LogReaderUtils.encodePositionsAsLongList(recordPositions);
Review Comment:
🤖 Thanks for the context on the bitmap count divergence — that clears up the
write-side rationale. The part that seems still open is the read side: native
blocks feed these positions into `PositionBasedFileGroupRecordBuffer` via
`getRecordPositionList()`, where they're used as map keys —
`records.put(recordPosition, ...)` in `processDeleteBlock` and
`processNextDataRecord(..., recordPosition)` for data. The count checks there
guard against record/position size mismatch, but not against two records in one
block sharing a position, which would silently overwrite the earlier entry
rather than throw. Since positions reference a single base-file instant, are
duplicates actually reachable on this path, or would a lightweight uniqueness
assert still be worth keeping to fail loud?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/LogReaderUtils.java:
##########
@@ -168,4 +193,66 @@ public static Roaring64NavigableMap
decodeRecordPositionsHeader(String content)
positionBitmap.deserializePortable(dis);
return positionBitmap;
}
+
+ /**
+ * Decodes native-log record positions written as an ordered unsigned
varlong list.
+ *
+ * @param content Base64-encoded bytes containing the list size followed by
varlong positions.
+ * @return record positions in file read order.
+ * @throws IOException upon I/O error or invalid payload shape.
+ */
+ public static List<Long> decodeRecordPositionsLongList(String content)
throws IOException {
+ ByteArrayInputStream bais = new
ByteArrayInputStream(Base64CodecUtil.decode(content));
+ DataInputStream dis = new DataInputStream(bais);
+ int size = dis.readInt();
+ if (size < 0) {
+ throw new IOException("Invalid negative record position count " + size);
+ }
+ List<Long> positions = new ArrayList<>(size);
Review Comment:
🤖 Since `size` is attacker/corruption-controlled here, `new
ArrayList<>(size)` pre-allocates before we've verified there are actually that
many bytes to read. A malformed payload (or a native block that somehow still
carries the old bitmap-encoded RECORD_POSITIONS from earlier TV10 dev writes)
would read the first 4 bytes as a large positive `size` and could OOM before
the per-position EOF check trips. Would it be worth bounding `size` by the
remaining bytes (each position is ≥ 1 byte, so `size` can't exceed
`bais.available()`) before allocating?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/LogReaderUtils.java:
##########
@@ -152,6 +155,28 @@ public static String encodePositions(Roaring64NavigableMap
positionBitmap) throw
return Base64CodecUtil.encode(baos.toByteArray());
}
+ /**
+ * Encodes the record positions as an integer count followed by unsigned
varlong positions,
+ * preserving the input order.
+ *
+ * @param positions record positions in file read order.
+ * @return Base64-encoded bytes containing the list size followed by varlong
positions,
+ * or {@link Option#empty()} if any position is invalid.
+ * @throws IOException upon I/O error.
+ */
+ public static Option<String> encodePositionsAsLongList(List<Long> positions)
throws IOException {
Review Comment:
🤖 nit: could you add "Record" here to match its inverse
`decodeRecordPositionsLongList`? The asymmetry (`encodePositionsAsLongList` vs
`decodeRecordPositionsLongList`) makes the pair harder to discover — a reader
searching for how to decode output from this method would naturally look for a
counterpart with a similar name.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieLogBlock.java:
##########
@@ -142,13 +144,33 @@ public boolean isCompactedLogBlock() {
* {@link Roaring64NavigableMap} bitmap.
* @throws IOException upon I/O error.
*/
- public Roaring64NavigableMap getRecordPositions() throws IOException {
+ private Roaring64NavigableMap getRecordPositions() throws IOException {
if (!logBlockHeader.containsKey(HeaderMetadataType.RECORD_POSITIONS)) {
return new Roaring64NavigableMap();
}
return
LogReaderUtils.decodeRecordPositionsHeader(logBlockHeader.get(HeaderMetadataType.RECORD_POSITIONS));
}
+ /**
+ * @return record positions aligned with the block record iterator order.
+ * @throws IOException upon I/O error.
+ */
+ public List<Long> getRecordPositionList() throws IOException {
+ List<Long> positions = new ArrayList<>();
+ getRecordPositions().iterator().forEachRemaining(positions::add);
+ return positions;
+ }
+
+ /**
+ * Decodes ordered record positions stored for native log blocks.
+ */
+ protected static List<Long>
decodeOrderedRecordPositionList(Map<HeaderMetadataType, String> logBlockHeader)
throws IOException {
+ if (!logBlockHeader.containsKey(HeaderMetadataType.RECORD_POSITIONS)) {
Review Comment:
🤖 nit: `decodeOrderedRecordPositionList` — "ordered" doesn't distinguish
this from the bitmap path (which also returns an ordered list, just sorted
ascending). Since this specifically decodes the native-log insert-order
encoding, something like `decodeNativeLogRecordPositionList` might make the
semantic contrast clearer to future subclass authors.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
