hudi-agent commented on code in PR #19085:
URL: https://github.com/apache/hudi/pull/19085#discussion_r3489134271
##########
hudi-common/src/main/java/org/apache/hudi/common/util/collection/ExternalSpillableMap.java:
##########
@@ -315,6 +315,22 @@ public Stream<R> valueStream() {
return Stream.concat(inMemoryMap.values().stream(),
diskBasedMap.valueStream());
}
+ /**
+ * Returns a lazy stream over all keys in this map (in-memory + spilled).
+ *
+ * Unlike {@link #keySet()}, this method does NOT allocate a new {@link
java.util.HashSet} copying
+ * all keys. When the map has spilled ({@code diskBasedMap != null}), {@link
#keySet()} must
+ * materialise a full copy to combine both key sets into a single {@link
java.util.Set}. Callers
+ * that only need to iterate keys once (e.g. to seed a {@link
java.util.PriorityQueue}) should
+ * prefer this method to avoid the transient per-task heap spike (ENG-43078).
+ */
+ public Stream<T> keyStream() {
+ if (diskBasedMap == null) {
+ return inMemoryMap.keySet().stream();
+ }
+ return Stream.concat(inMemoryMap.keySet().stream(),
diskBasedMap.keySet().stream());
Review Comment:
🤖 When the map has spilled, `keyStream()` calls `diskBasedMap.keySet()`,
which this PR changes to eagerly return `new
HashSet<>(valueMetadataMap.keySet())` of all disk keys. So for a
heavily-spilled map (the MDT case) the transient HashSet is still materialized
here — does this fully realize the consumer-E savings, or would a truly
streaming keySet on `BitCaskDiskMap` be needed? The javadoc's claim that
keyStream "does NOT allocate a new HashSet copying all keys" seems only
partially true once disk keys dominate.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/collection/TestBitCaskDiskMap.java:
##########
@@ -275,4 +277,122 @@ private void verifyCleanup(BitCaskDiskMap<String,
HoodieRecord> records) {
records.close();
assertEquals(Objects.requireNonNull(basePathDir.list()).length, 0);
}
+
+ /**
+ * Re-insertion order: when a key is put twice, iteration order must place
it last (i.e. the
+ * second put's higher disk offset is at the LinkedHashMap tail), and reads
must return the
+ * latest value. Locks down the invariant that {@code BitCaskDiskMap.put}
removes-before-puts
+ * to keep {@code LinkedHashMap} insertion-order == disk-offset-order
(ENG-43078).
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testReInsertionPreservesOffsetOrder(boolean
isCompressionEnabled) throws IOException, URISyntaxException {
+ try (BitCaskDiskMap<String, HoodieRecord> records = new
BitCaskDiskMap<>(basePath, new DefaultSerializer<>(), isCompressionEnabled)) {
+ SchemaTestUtil testUtil = new SchemaTestUtil();
+ List<IndexedRecord> iRecords = testUtil.generateHoodieTestRecords(0, 5);
+ List<String> orderedKeys = new ArrayList<>();
+ for (IndexedRecord r : iRecords) {
+ String key = ((GenericRecord)
r).get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString();
+ String partitionPath = ((GenericRecord)
r).get(HoodieRecord.PARTITION_PATH_METADATA_FIELD).toString();
+ HoodieRecord value = new HoodieAvroRecord<>(new HoodieKey(key,
partitionPath), new HoodieAvroPayload(Option.of((GenericRecord) r)));
+ records.put(key, value);
+ orderedKeys.add(key);
+ }
+ // Re-insert the first key; it must move to the iteration tail.
+ String firstKey = orderedKeys.get(0);
+ List<IndexedRecord> updatedRec = testUtil.generateHoodieTestRecords(100,
1);
+ GenericRecord updatedGeneric = (GenericRecord) updatedRec.get(0);
+ HoodieRecord updatedValue = new HoodieAvroRecord<>(
+ new HoodieKey(firstKey,
updatedGeneric.get(HoodieRecord.PARTITION_PATH_METADATA_FIELD).toString()),
+ new HoodieAvroPayload(Option.of(updatedGeneric)));
+ records.put(firstKey, updatedValue);
+
+ // size must remain 5 (re-insertion replaces the existing entry).
+ List<HoodieRecord> iteratedFromEntrySet = new ArrayList<>();
Review Comment:
🤖 nit: `iteratedFromEntrySet` is populated by calling `records.iterator()`,
not `records.entrySet()` — could you rename it to something like
`iteratedRecords` or `iteratedValues` so a future reader isn't sent looking for
an `entrySet()` call?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/util/collection/BitCaskDiskMap.java:
##########
@@ -320,13 +427,30 @@ public Collection<R> values() {
@Override
public Stream<R> valueStream() {
final BufferedRandomAccessFile file = getRandomAccessFile();
- return valueMetadataMap.values().stream().sorted().map(valueMetaData ->
get(valueMetaData, file, valueSerializer, isCompressionEnabled));
+ // Snapshot under mapReadLock. LinkedHashMap insertion order equals disk
offset order
+ // (see put()), so no sort is needed — values are already in forward-read
order.
+ final List<ValueMetadata> snapshot;
+ mapReadLock.lock();
+ try {
+ snapshot = new ArrayList<>(valueMetadataMap.values());
+ } finally {
+ mapReadLock.unlock();
+ }
+ return snapshot.stream().sequential().map(valueMetaData -> (R)
get(valueMetaData, file, valueSerializer, isCompressionEnabled));
Review Comment:
🤖 nit: `.sequential()` is a no-op here — `ArrayList.stream()` already
returns a sequential stream. Could you drop it? It may leave a future reader
wondering whether parallelism was a concern and whether the stream might become
parallel.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/collection/TestBitCaskDiskMap.java:
##########
@@ -275,4 +277,122 @@ private void verifyCleanup(BitCaskDiskMap<String,
HoodieRecord> records) {
records.close();
assertEquals(Objects.requireNonNull(basePathDir.list()).length, 0);
}
+
+ /**
+ * Re-insertion order: when a key is put twice, iteration order must place
it last (i.e. the
+ * second put's higher disk offset is at the LinkedHashMap tail), and reads
must return the
+ * latest value. Locks down the invariant that {@code BitCaskDiskMap.put}
removes-before-puts
+ * to keep {@code LinkedHashMap} insertion-order == disk-offset-order
(ENG-43078).
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testReInsertionPreservesOffsetOrder(boolean
isCompressionEnabled) throws IOException, URISyntaxException {
+ try (BitCaskDiskMap<String, HoodieRecord> records = new
BitCaskDiskMap<>(basePath, new DefaultSerializer<>(), isCompressionEnabled)) {
+ SchemaTestUtil testUtil = new SchemaTestUtil();
+ List<IndexedRecord> iRecords = testUtil.generateHoodieTestRecords(0, 5);
+ List<String> orderedKeys = new ArrayList<>();
+ for (IndexedRecord r : iRecords) {
+ String key = ((GenericRecord)
r).get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString();
+ String partitionPath = ((GenericRecord)
r).get(HoodieRecord.PARTITION_PATH_METADATA_FIELD).toString();
+ HoodieRecord value = new HoodieAvroRecord<>(new HoodieKey(key,
partitionPath), new HoodieAvroPayload(Option.of((GenericRecord) r)));
+ records.put(key, value);
+ orderedKeys.add(key);
+ }
+ // Re-insert the first key; it must move to the iteration tail.
+ String firstKey = orderedKeys.get(0);
+ List<IndexedRecord> updatedRec = testUtil.generateHoodieTestRecords(100,
1);
+ GenericRecord updatedGeneric = (GenericRecord) updatedRec.get(0);
+ HoodieRecord updatedValue = new HoodieAvroRecord<>(
+ new HoodieKey(firstKey,
updatedGeneric.get(HoodieRecord.PARTITION_PATH_METADATA_FIELD).toString()),
+ new HoodieAvroPayload(Option.of(updatedGeneric)));
+ records.put(firstKey, updatedValue);
+
+ // size must remain 5 (re-insertion replaces the existing entry).
+ List<HoodieRecord> iteratedFromEntrySet = new ArrayList<>();
+ Iterator<HoodieRecord> iter = records.iterator();
+ while (iter.hasNext()) {
+ iteratedFromEntrySet.add(iter.next());
+ }
+ assertEquals(5, iteratedFromEntrySet.size());
+ // Use the stable HoodieRecord.getRecordKey() accessor rather than
dereffing payload avro
+ // (whose schema may not be metadata-augmented in this test setup).
+ assertEquals(firstKey,
+ iteratedFromEntrySet.get(4).getRecordKey(),
+ "Re-inserted key must appear last in iteration order");
+
+ // Read-back of the re-inserted key must return the latest value.
+ HoodieRecord readBack = records.get(firstKey);
+ assertNotNull(readBack);
+ }
+ }
+
+ /**
+ * Concurrent reader + writer: while a writer thread does put/remove under
{@code mapWriteLock},
+ * a reader thread doing {@code get}/{@code containsKey} under {@code
mapReadLock} must not throw
+ * {@link java.util.ConcurrentModificationException} and must observe values
eventually.
+ * Locks down the RW-lock semantics introduced in ENG-43078.
+ */
+ @Test
+ public void testConcurrentReadersAndWriter() throws Exception {
+ try (BitCaskDiskMap<String, HoodieRecord> records = new
BitCaskDiskMap<>(basePath, new DefaultSerializer<>(), false)) {
+ SchemaTestUtil testUtil = new SchemaTestUtil();
+ List<IndexedRecord> iRecords = testUtil.generateHoodieTestRecords(0,
200);
+ List<String> allKeys = new ArrayList<>();
+ List<HoodieRecord> allValues = new ArrayList<>();
+ for (IndexedRecord r : iRecords) {
+ String key = ((GenericRecord)
r).get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString();
+ String partitionPath = ((GenericRecord)
r).get(HoodieRecord.PARTITION_PATH_METADATA_FIELD).toString();
+ HoodieRecord value = new HoodieAvroRecord<>(new HoodieKey(key,
partitionPath), new HoodieAvroPayload(Option.of((GenericRecord) r)));
+ allKeys.add(key);
+ allValues.add(value);
+ }
+
+ java.util.concurrent.atomic.AtomicBoolean writerDone = new
java.util.concurrent.atomic.AtomicBoolean(false);
Review Comment:
🤖 nit: `AtomicBoolean` and `AtomicReference` are used with fully-qualified
names here — could you add imports at the top of the file instead?
Fully-qualified names mid-method add visual noise and make the lines quite hard
to scan.
<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]
