This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 82c050203e3d refactor(flink): remove unused FlinkDeleteHelper (#19962)
82c050203e3d is described below
commit 82c050203e3dfd344452ab8d1b84354ba662fa2f
Author: voonhous <[email protected]>
AuthorDate: Wed Sep 16 10:45:30 2026 +0800
refactor(flink): remove unused FlinkDeleteHelper (#19962)
FlinkDeleteHelper has no main-source caller; it is a copy of
JavaDeleteHelper with an identical execute(). Delete it and its test,
and move the three execute tests into TestJavaDeleteHelper, where the
same code is still reached by the Java delete executors.
---
.../table/action/commit/FlinkDeleteHelper.java | 128 ---------------
.../table/action/commit/TestFlinkDeleteHelper.java | 174 ---------------------
.../table/action/commit/TestJavaDeleteHelper.java | 101 ++++++++++++
3 files changed, 101 insertions(+), 302 deletions(-)
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/FlinkDeleteHelper.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/FlinkDeleteHelper.java
deleted file mode 100644
index 733b79fa9987..000000000000
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/FlinkDeleteHelper.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * 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.table.action.commit;
-
-import org.apache.hudi.client.WriteStatus;
-import org.apache.hudi.common.data.HoodieListData;
-import org.apache.hudi.common.engine.HoodieEngineContext;
-import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
-import org.apache.hudi.common.model.HoodieAvroRecord;
-import org.apache.hudi.common.model.HoodieKey;
-import org.apache.hudi.common.model.HoodieRecord;
-import org.apache.hudi.common.util.collection.Pair;
-import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieUpsertException;
-import org.apache.hudi.table.HoodieTable;
-import org.apache.hudi.table.WorkloadProfile;
-import org.apache.hudi.table.WorkloadStat;
-import org.apache.hudi.table.action.HoodieWriteMetadata;
-
-import java.time.Duration;
-import java.time.Instant;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.stream.Collectors;
-
-/**
- * Flink delete helper.
- */
-@SuppressWarnings("checkstyle:LineLength")
-public class FlinkDeleteHelper<R> extends
- BaseDeleteHelper<EmptyHoodieRecordPayload,
List<HoodieRecord<EmptyHoodieRecordPayload>>, List<HoodieKey>,
List<WriteStatus>, R> {
-
- private FlinkDeleteHelper() {
- super(ignored -> -1);
- }
-
- private static class DeleteHelperHolder {
- private static final FlinkDeleteHelper FLINK_DELETE_HELPER = new
FlinkDeleteHelper();
- }
-
- public static FlinkDeleteHelper newInstance() {
- return DeleteHelperHolder.FLINK_DELETE_HELPER;
- }
-
- @Override
- public List<HoodieKey> deduplicateKeys(List<HoodieKey> keys,
HoodieTable<EmptyHoodieRecordPayload,
List<HoodieRecord<EmptyHoodieRecordPayload>>, List<HoodieKey>,
List<WriteStatus>> table, int parallelism) {
- boolean isIndexingGlobal = table.getIndex().isGlobal();
- if (isIndexingGlobal) {
- HashSet<String> recordKeys = new HashSet<>();
- List<HoodieKey> deduplicatedKeys = new LinkedList<>();
- keys.forEach(x -> {
- if (recordKeys.add(x.getRecordKey())) {
- deduplicatedKeys.add(x);
- }
- });
- return deduplicatedKeys;
- } else {
- LinkedHashSet<HoodieKey> set = new LinkedHashSet<>(keys);
- keys.clear();
- keys.addAll(set);
- return keys;
- }
- }
-
- @Override
- public HoodieWriteMetadata<List<WriteStatus>> execute(String instantTime,
- List<HoodieKey> keys,
- HoodieEngineContext
context,
- HoodieWriteConfig
config,
-
HoodieTable<EmptyHoodieRecordPayload,
List<HoodieRecord<EmptyHoodieRecordPayload>>, List<HoodieKey>,
List<WriteStatus>> table,
-
BaseCommitActionExecutor<EmptyHoodieRecordPayload,
List<HoodieRecord<EmptyHoodieRecordPayload>>, List<HoodieKey>,
List<WriteStatus>, R> deleteExecutor) {
- try {
- HoodieWriteMetadata<List<WriteStatus>> result = null;
- List<HoodieKey> dedupedKeys = keys;
- final int parallelism = config.getDeleteShuffleParallelism();
- if (config.shouldCombineBeforeDelete()) {
- // De-dupe/merge if needed
- dedupedKeys = deduplicateKeys(keys, table, parallelism);
- }
-
- List<HoodieRecord<EmptyHoodieRecordPayload>> dedupedRecords =
- dedupedKeys.stream().map(key -> new HoodieAvroRecord<>(key, new
EmptyHoodieRecordPayload())).collect(Collectors.toList());
- Instant beginTag = Instant.now();
- // perform index look up to get existing location of records
- List<HoodieRecord<EmptyHoodieRecordPayload>> taggedRecords =
table.getIndex().tagLocation(HoodieListData.eager(dedupedRecords), context,
table).collectAsList();
- Duration tagLocationDuration = Duration.between(beginTag, Instant.now());
-
- // filter out non existent keys/records
- List<HoodieRecord<EmptyHoodieRecordPayload>> taggedValidRecords =
taggedRecords.stream().filter(HoodieRecord::isCurrentLocationKnown).collect(Collectors.toList());
- if (!taggedValidRecords.isEmpty()) {
- result = deleteExecutor.execute(taggedValidRecords);
- result.setIndexLookupDuration(tagLocationDuration);
- } else {
- // if entire set of keys are non existent
- deleteExecutor.saveWorkloadProfileMetadataToInflight(new
WorkloadProfile(Pair.of(new HashMap<>(), new WorkloadStat())), instantTime);
- result = new HoodieWriteMetadata<>();
- result.setWriteStatuses(Collections.EMPTY_LIST);
- deleteExecutor.runPrecommitValidators(result);
- }
- return result;
- } catch (Throwable e) {
- if (e instanceof HoodieUpsertException) {
- throw (HoodieUpsertException) e;
- }
- throw new HoodieUpsertException("Failed to delete for commit time " +
instantTime, e);
- }
- }
-}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/action/commit/TestFlinkDeleteHelper.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/action/commit/TestFlinkDeleteHelper.java
deleted file mode 100644
index 3b9b6e8bc423..000000000000
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/action/commit/TestFlinkDeleteHelper.java
+++ /dev/null
@@ -1,174 +0,0 @@
-/*
- * 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.table.action.commit;
-
-import org.apache.hudi.client.WriteStatus;
-import org.apache.hudi.common.data.HoodieData;
-import org.apache.hudi.common.data.HoodieListData;
-import org.apache.hudi.common.engine.HoodieEngineContext;
-import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
-import org.apache.hudi.common.model.HoodieKey;
-import org.apache.hudi.common.model.HoodieRecord;
-import org.apache.hudi.common.model.HoodieRecordLocation;
-import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieUpsertException;
-import org.apache.hudi.index.HoodieIndex;
-import org.apache.hudi.table.HoodieTable;
-import org.apache.hudi.table.action.HoodieWriteMetadata;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.stream.Collectors;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotSame;
-import static org.junit.jupiter.api.Assertions.assertSame;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.reset;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-/** Tests for {@link FlinkDeleteHelper}. */
-@SuppressWarnings({"rawtypes", "unchecked"})
-class TestFlinkDeleteHelper {
-
- @Test
- void testDeduplicateKeysForGlobalAndPartitionedIndexes() {
- FlinkDeleteHelper helper = FlinkDeleteHelper.newInstance();
- assertSame(helper, FlinkDeleteHelper.newInstance());
-
- HoodieTable table = mock(HoodieTable.class);
- HoodieIndex index = mock(HoodieIndex.class);
- when(table.getIndex()).thenReturn(index);
- List<HoodieKey> keys = new ArrayList<>(Arrays.asList(
- new HoodieKey("id1", "p1"),
- new HoodieKey("id1", "p2"),
- new HoodieKey("id2", "p1"),
- new HoodieKey("id2", "p1")));
-
- when(index.isGlobal()).thenReturn(true);
- List<HoodieKey> globalResult = helper.deduplicateKeys(keys, table, 1);
- assertEquals(Arrays.asList("id1", "id2"), globalResult.stream()
- .map(HoodieKey::getRecordKey).collect(Collectors.toList()));
- assertEquals(Arrays.asList("p1", "p1"), globalResult.stream()
- .map(HoodieKey::getPartitionPath).collect(Collectors.toList()));
- assertNotSame(keys, globalResult);
-
- when(index.isGlobal()).thenReturn(false);
- List<HoodieKey> partitionedResult = helper.deduplicateKeys(keys, table, 1);
- assertSame(keys, partitionedResult);
- assertEquals(Arrays.asList(
- new HoodieKey("id1", "p1"),
- new HoodieKey("id1", "p2"),
- new HoodieKey("id2", "p1")), partitionedResult);
- }
-
- @Test
- void testExecuteTagsExistingRecordsAndDelegatesDelete() {
- HoodieTable table = mock(HoodieTable.class);
- HoodieIndex index = mock(HoodieIndex.class);
- HoodieEngineContext context = mock(HoodieEngineContext.class);
- BaseCommitActionExecutor executor = mock(BaseCommitActionExecutor.class);
- when(table.getIndex()).thenReturn(index);
- when(index.isGlobal()).thenReturn(false);
-
- when(index.tagLocation(any(HoodieData.class), eq(context),
eq(table))).thenAnswer(invocation -> {
- HoodieData<HoodieRecord<EmptyHoodieRecordPayload>> records =
invocation.getArgument(0);
- List<HoodieRecord<EmptyHoodieRecordPayload>> tagged =
records.collectAsList();
- tagged.get(0).setCurrentLocation(new HoodieRecordLocation("001",
"file-1"));
- return HoodieListData.eager(tagged);
- });
-
- HoodieWriteMetadata<List<WriteStatus>> expected = new
HoodieWriteMetadata<>();
- expected.setWriteStatuses(Collections.singletonList(new WriteStatus(false,
0.0)));
- when(executor.execute(any(List.class))).thenReturn(expected);
-
- HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
- .withPath("/tmp/flink-delete-helper")
- .combineDeleteInput(true)
- .build();
- List<HoodieKey> keys = new ArrayList<>(Arrays.asList(
- new HoodieKey("id1", "p1"), new HoodieKey("id1", "p1"), new
HoodieKey("missing", "p1")));
-
- HoodieWriteMetadata<List<WriteStatus>> result =
FlinkDeleteHelper.newInstance()
- .execute("002", keys, context, config, table, executor);
-
- assertSame(expected, result);
- assertTrue(result.getIndexLookupDuration().isPresent());
- verify(executor).execute(any(List.class));
- verify(executor, never()).saveWorkloadProfileMetadataToInflight(any(),
any());
- }
-
- @Test
- void testExecuteWithNoExistingRecordsCreatesEmptyMetadata() {
- HoodieTable table = mock(HoodieTable.class);
- HoodieIndex index = mock(HoodieIndex.class);
- HoodieEngineContext context = mock(HoodieEngineContext.class);
- BaseCommitActionExecutor executor = mock(BaseCommitActionExecutor.class);
- when(table.getIndex()).thenReturn(index);
- when(index.tagLocation(any(HoodieData.class), eq(context), eq(table)))
- .thenAnswer(invocation -> invocation.getArgument(0));
-
- HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
- .withPath("/tmp/flink-delete-helper")
- .combineDeleteInput(false)
- .build();
- HoodieWriteMetadata<List<WriteStatus>> result =
FlinkDeleteHelper.newInstance().execute(
- "003", Collections.singletonList(new HoodieKey("missing", "p1")),
- context, config, table, executor);
-
- assertTrue(result.getWriteStatuses().isEmpty());
- verify(executor).saveWorkloadProfileMetadataToInflight(any(), eq("003"));
- verify(executor).runPrecommitValidators(result);
- verify(executor, never()).execute(any(List.class));
- }
-
- @Test
- void testExecutePreservesOrWrapsFailures() {
- HoodieTable table = mock(HoodieTable.class);
- HoodieEngineContext context = mock(HoodieEngineContext.class);
- BaseCommitActionExecutor executor = mock(BaseCommitActionExecutor.class);
- HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
- .withPath("/tmp/flink-delete-helper")
- .build();
- List<HoodieKey> keys = Collections.singletonList(new HoodieKey("id1",
"p1"));
-
- HoodieUpsertException upsertException = new
HoodieUpsertException("expected");
- when(table.getIndex()).thenThrow(upsertException);
- HoodieUpsertException firstFailure =
assertThrows(HoodieUpsertException.class,
- () -> FlinkDeleteHelper.newInstance().execute("004", keys, context,
config, table, executor));
- assertSame(upsertException, firstFailure);
-
- reset(table);
- when(table.getIndex()).thenThrow(new IllegalStateException("boom"));
- HoodieUpsertException wrapped = assertThrows(HoodieUpsertException.class,
- () -> FlinkDeleteHelper.newInstance().execute("005", keys, context,
config, table, executor));
- assertTrue(wrapped.getCause() instanceof IllegalStateException);
- }
-}
diff --git
a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/table/action/commit/TestJavaDeleteHelper.java
b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/table/action/commit/TestJavaDeleteHelper.java
index fab7d0423163..5a3efa923263 100644
---
a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/table/action/commit/TestJavaDeleteHelper.java
+++
b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/table/action/commit/TestJavaDeleteHelper.java
@@ -18,21 +18,39 @@
package org.apache.hudi.table.action.commit;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.data.HoodieListData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieUpsertException;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/** Tests for {@link JavaDeleteHelper}. */
@@ -69,4 +87,87 @@ class TestJavaDeleteHelper {
new HoodieKey("id1", "p2"),
new HoodieKey("id2", "p1")), partitionedResult);
}
+
+ @Test
+ void testExecuteTagsExistingRecordsAndDelegatesDelete() {
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieIndex index = mock(HoodieIndex.class);
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ BaseCommitActionExecutor executor = mock(BaseCommitActionExecutor.class);
+ when(table.getIndex()).thenReturn(index);
+ when(index.isGlobal()).thenReturn(false);
+
+ when(index.tagLocation(any(HoodieData.class), eq(context),
eq(table))).thenAnswer(invocation -> {
+ HoodieData<HoodieRecord<EmptyHoodieRecordPayload>> records =
invocation.getArgument(0);
+ List<HoodieRecord<EmptyHoodieRecordPayload>> tagged =
records.collectAsList();
+ tagged.get(0).setCurrentLocation(new HoodieRecordLocation("001",
"file-1"));
+ return HoodieListData.eager(tagged);
+ });
+
+ HoodieWriteMetadata<List<WriteStatus>> expected = new
HoodieWriteMetadata<>();
+ expected.setWriteStatuses(Collections.singletonList(new WriteStatus(false,
0.0)));
+ when(executor.execute(any(List.class))).thenReturn(expected);
+
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/java-delete-helper")
+ .combineDeleteInput(true)
+ .build();
+ List<HoodieKey> keys = new ArrayList<>(Arrays.asList(
+ new HoodieKey("id1", "p1"), new HoodieKey("id1", "p1"), new
HoodieKey("missing", "p1")));
+
+ HoodieWriteMetadata<List<WriteStatus>> result =
JavaDeleteHelper.newInstance()
+ .execute("002", keys, context, config, table, executor);
+
+ assertSame(expected, result);
+ assertTrue(result.getIndexLookupDuration().isPresent());
+ verify(executor).execute(any(List.class));
+ verify(executor, never()).saveWorkloadProfileMetadataToInflight(any(),
any());
+ }
+
+ @Test
+ void testExecuteWithNoExistingRecordsCreatesEmptyMetadata() {
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieIndex index = mock(HoodieIndex.class);
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ BaseCommitActionExecutor executor = mock(BaseCommitActionExecutor.class);
+ when(table.getIndex()).thenReturn(index);
+ when(index.tagLocation(any(HoodieData.class), eq(context), eq(table)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/java-delete-helper")
+ .combineDeleteInput(false)
+ .build();
+ HoodieWriteMetadata<List<WriteStatus>> result =
JavaDeleteHelper.newInstance().execute(
+ "003", Collections.singletonList(new HoodieKey("missing", "p1")),
+ context, config, table, executor);
+
+ assertTrue(result.getWriteStatuses().isEmpty());
+ verify(executor).saveWorkloadProfileMetadataToInflight(any(), eq("003"));
+ verify(executor).runPrecommitValidators(result);
+ verify(executor, never()).execute(any(List.class));
+ }
+
+ @Test
+ void testExecutePreservesOrWrapsFailures() {
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ BaseCommitActionExecutor executor = mock(BaseCommitActionExecutor.class);
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/java-delete-helper")
+ .build();
+ List<HoodieKey> keys = Collections.singletonList(new HoodieKey("id1",
"p1"));
+
+ HoodieUpsertException upsertException = new
HoodieUpsertException("expected");
+ when(table.getIndex()).thenThrow(upsertException);
+ HoodieUpsertException firstFailure =
assertThrows(HoodieUpsertException.class,
+ () -> JavaDeleteHelper.newInstance().execute("004", keys, context,
config, table, executor));
+ assertSame(upsertException, firstFailure);
+
+ reset(table);
+ when(table.getIndex()).thenThrow(new IllegalStateException("boom"));
+ HoodieUpsertException wrapped = assertThrows(HoodieUpsertException.class,
+ () -> JavaDeleteHelper.newInstance().execute("005", keys, context,
config, table, executor));
+ assertTrue(wrapped.getCause() instanceof IllegalStateException);
+ }
}
