hudi-agent commented on code in PR #18849:
URL: https://github.com/apache/hudi/pull/18849#discussion_r3306349747
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java:
##########
@@ -205,6 +205,34 @@ private synchronized void persist(T data) {
}
}
+ /**
+ * Whether the source RDD/DataFrame should be persisted via {@link
#persist(Object)}.
+ *
+ * <p>Returns {@code false} when the writer's index type is a
metadata-table-backed record
+ * index ({@code RECORD_INDEX} or {@code PARTITIONED_RECORD_INDEX}), even if
+ * {@code hoodie.errortable.source.rdd.persist=true}. The RLI tagging path
already persists
+ * the records RDD inside {@code SparkMetadataTableRecordIndex#tagLocation},
and the
+ * downstream commit executor persists the tagged records again. Caching the
upstream
+ * Dataset on top of those two triples the on-heap serialized footprint and
contends with
+ * the MDT {@code HoodieAppendHandle} buffer, which has caused executor
OOMKills on
+ * RLI-indexed tables.
+ */
+ protected boolean isAllowSourcePersistRdd() {
+ if (!allowSourcePersistRdd) {
+ return false;
+ }
+ return !usesRecordIndex();
+ }
+
+ private boolean usesRecordIndex() {
+ String indexType = props.getString(HoodieIndexConfig.INDEX_TYPE.key(), "");
+ if (indexType.isEmpty()) {
+ return false;
+ }
+ return
HoodieIndex.IndexType.RECORD_INDEX.name().equalsIgnoreCase(indexType)
+ ||
HoodieIndex.IndexType.PARTITIONED_RECORD_INDEX.name().equalsIgnoreCase(indexType);
Review Comment:
🤖 The motivating OOM scenario applies equally to `RECORD_LEVEL_INDEX` and
`GLOBAL_RECORD_LEVEL_INDEX` (these are the current, non-deprecated MDT-backed
record indexes — `RECORD_INDEX` is marked `@Deprecated`). Could you include
both here so that users on the non-deprecated index types also get the OOM fix?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java:
##########
@@ -205,6 +205,34 @@ private synchronized void persist(T data) {
}
}
+ /**
+ * Whether the source RDD/DataFrame should be persisted via {@link
#persist(Object)}.
+ *
+ * <p>Returns {@code false} when the writer's index type is a
metadata-table-backed record
+ * index ({@code RECORD_INDEX} or {@code PARTITIONED_RECORD_INDEX}), even if
+ * {@code hoodie.errortable.source.rdd.persist=true}. The RLI tagging path
already persists
+ * the records RDD inside {@code SparkMetadataTableRecordIndex#tagLocation},
and the
+ * downstream commit executor persists the tagged records again. Caching the
upstream
+ * Dataset on top of those two triples the on-heap serialized footprint and
contends with
+ * the MDT {@code HoodieAppendHandle} buffer, which has caused executor
OOMKills on
+ * RLI-indexed tables.
+ */
+ protected boolean isAllowSourcePersistRdd() {
+ if (!allowSourcePersistRdd) {
+ return false;
+ }
+ return !usesRecordIndex();
+ }
+
+ private boolean usesRecordIndex() {
+ String indexType = props.getString(HoodieIndexConfig.INDEX_TYPE.key(), "");
+ if (indexType.isEmpty()) {
+ return false;
+ }
+ return
HoodieIndex.IndexType.RECORD_INDEX.name().equalsIgnoreCase(indexType)
+ ||
HoodieIndex.IndexType.PARTITIONED_RECORD_INDEX.name().equalsIgnoreCase(indexType);
+ }
Review Comment:
🤖 `HoodieIndex.IndexType.PARTITIONED_RECORD_INDEX` doesn't exist in the enum
— the values are `INMEMORY, BLOOM, GLOBAL_BLOOM, SIMPLE, GLOBAL_SIMPLE, BUCKET,
FLINK_STATE, RECORD_INDEX, GLOBAL_RECORD_LEVEL_INDEX, RECORD_LEVEL_INDEX`. This
won't compile (CI is red for that reason). Did you mean `RECORD_LEVEL_INDEX`
(and likely also `GLOBAL_RECORD_LEVEL_INDEX`)?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java:
##########
@@ -205,6 +205,34 @@ private synchronized void persist(T data) {
}
}
+ /**
+ * Whether the source RDD/DataFrame should be persisted via {@link
#persist(Object)}.
+ *
+ * <p>Returns {@code false} when the writer's index type is a
metadata-table-backed record
+ * index ({@code RECORD_INDEX} or {@code PARTITIONED_RECORD_INDEX}), even if
+ * {@code hoodie.errortable.source.rdd.persist=true}. The RLI tagging path
already persists
+ * the records RDD inside {@code SparkMetadataTableRecordIndex#tagLocation},
and the
+ * downstream commit executor persists the tagged records again. Caching the
upstream
+ * Dataset on top of those two triples the on-heap serialized footprint and
contends with
+ * the MDT {@code HoodieAppendHandle} buffer, which has caused executor
OOMKills on
+ * RLI-indexed tables.
+ */
+ protected boolean isAllowSourcePersistRdd() {
Review Comment:
🤖 nit: the guard-plus-return pattern could be collapsed into a single
`return allowSourcePersistRdd && !usesRecordIndex();`, which reads more
directly without losing any clarity.
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSource.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.utilities.sources;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieErrorTableConfig;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.utilities.schema.SchemaProvider;
+import org.apache.hudi.utilities.streamer.SourceProfileSupplier;
+import org.apache.hudi.utilities.streamer.StreamContext;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Unit tests for {@link Source#isAllowSourcePersistRdd()}.
+ *
+ * <p>The source-RDD persist controlled by
+ * {@code hoodie.errortable.source.rdd.persist} is bypassed when the writer is
+ * configured with a metadata-table-backed record index, since
+ * {@code SparkMetadataTableRecordIndex#tagLocation} already persists the
+ * records RDD at {@code MEMORY_AND_DISK_SER}. Double-caching has caused
+ * executor OOMKills on RLI-indexed tables.
+ */
+public class TestSource {
+
+ @Test
+ public void isAllowSourcePersistRddReturnsFalseWhenPersistFlagIsDisabled() {
+ TypedProperties props = new TypedProperties();
+
props.setProperty(HoodieErrorTableConfig.ERROR_TABLE_PERSIST_SOURCE_RDD.key(),
"false");
+ props.setProperty(HoodieIndexConfig.INDEX_TYPE.key(),
HoodieIndex.IndexType.BLOOM.name());
+
+ assertFalse(new NoopSource(props).isAllowSourcePersistRdd(),
+ "persist flag off must always disable source persist");
+ }
+
+ @Test
+ public void isAllowSourcePersistRddReturnsTrueWhenIndexTypeIsAbsent() {
+ TypedProperties props = new TypedProperties();
+
props.setProperty(HoodieErrorTableConfig.ERROR_TABLE_PERSIST_SOURCE_RDD.key(),
"true");
+
+ assertTrue(new NoopSource(props).isAllowSourcePersistRdd(),
+ "absent index type must not bypass persist (engine-level default
applies elsewhere)");
+ }
+
+ @ParameterizedTest
Review Comment:
🤖 `PARTITIONED_RECORD_INDEX` is not a value in `HoodieIndex.IndexType`, so
`@EnumSource(names = {..., "PARTITIONED_RECORD_INDEX"})` will fail at test
discovery / compile time. Please update to the correct enum names (likely
`RECORD_LEVEL_INDEX`, `GLOBAL_RECORD_LEVEL_INDEX`) and mirror the change in
`Source#usesRecordIndex`.
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java:
##########
@@ -205,6 +205,34 @@ private synchronized void persist(T data) {
}
}
+ /**
+ * Whether the source RDD/DataFrame should be persisted via {@link
#persist(Object)}.
+ *
+ * <p>Returns {@code false} when the writer's index type is a
metadata-table-backed record
+ * index ({@code RECORD_INDEX} or {@code PARTITIONED_RECORD_INDEX}), even if
+ * {@code hoodie.errortable.source.rdd.persist=true}. The RLI tagging path
already persists
+ * the records RDD inside {@code SparkMetadataTableRecordIndex#tagLocation},
and the
+ * downstream commit executor persists the tagged records again. Caching the
upstream
+ * Dataset on top of those two triples the on-heap serialized footprint and
contends with
+ * the MDT {@code HoodieAppendHandle} buffer, which has caused executor
OOMKills on
+ * RLI-indexed tables.
+ */
+ protected boolean isAllowSourcePersistRdd() {
+ if (!allowSourcePersistRdd) {
+ return false;
+ }
+ return !usesRecordIndex();
+ }
+
+ private boolean usesRecordIndex() {
+ String indexType = props.getString(HoodieIndexConfig.INDEX_TYPE.key(), "");
+ if (indexType.isEmpty()) {
Review Comment:
🤖 nit: the `isEmpty()` early-return is redundant — `equalsIgnoreCase`
already returns `false` for an empty string — and it implies an empty value is
a meaningful special case when it isn't. Could you drop those three lines and
let the `equalsIgnoreCase` comparisons carry the full logic?
<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]
