This is an automated email from the ASF dual-hosted git repository.
voonhous 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 4985f93d6fb3 refactor(client): dedupe HoodieJavaEngineContext with
local context (#19981)
4985f93d6fb3 is described below
commit 4985f93d6fb38627f928ea9995625696a47a45c2
Author: voonhous <[email protected]>
AuthorDate: Fri Sep 18 13:56:55 2026 +0800
refactor(client): dedupe HoodieJavaEngineContext with local context (#19981)
* refactor(client): dedupe HoodieJavaEngineContext with local context
HoodieJavaEngineContext copied every method body of
HoodieLocalEngineContext. Make it extend HoodieLocalEngineContext
(no longer final) and keep only its constructors and the two real
differences: the system-property allowlist in getEngineProperties
and the Avro key generator in createKeyGenerator.
* refactor(common): share engine logic via BaseLocalEngineContext
Keep HoodieLocalEngineContext final. Move the shared in-JVM method
bodies into a new abstract BaseLocalEngineContext in hudi-common.
HoodieLocalEngineContext and HoodieJavaEngineContext both extend it,
so the Java context is no longer a HoodieLocalEngineContext.
---
.../client/common/HoodieJavaEngineContext.java | 170 +--------------------
...ineContext.java => BaseLocalEngineContext.java} | 18 +--
.../common/engine/HoodieLocalEngineContext.java | 168 +-------------------
3 files changed, 7 insertions(+), 349 deletions(-)
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/common/HoodieJavaEngineContext.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/common/HoodieJavaEngineContext.java
index a24d6fb4c25d..06b7566174b0 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/common/HoodieJavaEngineContext.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/common/HoodieJavaEngineContext.java
@@ -19,56 +19,20 @@
package org.apache.hudi.client.common;
import org.apache.hudi.common.config.TypedProperties;
-import org.apache.hudi.common.data.HoodieAccumulator;
-import org.apache.hudi.common.data.HoodieAtomicLongAccumulator;
-import org.apache.hudi.common.data.HoodieData;
-import org.apache.hudi.common.data.HoodieData.HoodieDataCacheKey;
-import org.apache.hudi.common.data.HoodieListData;
-import org.apache.hudi.common.data.HoodieListPairData;
-import org.apache.hudi.common.data.HoodiePairData;
-import org.apache.hudi.common.engine.AvroReaderContextFactory;
-import org.apache.hudi.common.engine.EngineProperty;
-import org.apache.hudi.common.engine.HoodieEngineContext;
-import org.apache.hudi.common.engine.ReaderContextFactory;
+import org.apache.hudi.common.engine.BaseLocalEngineContext;
import org.apache.hudi.common.engine.TaskContextSupplier;
-import org.apache.hudi.common.function.SerializableBiFunction;
-import org.apache.hudi.common.function.SerializableConsumer;
-import org.apache.hudi.common.function.SerializableFunction;
-import org.apache.hudi.common.function.SerializablePairFlatMapFunction;
-import org.apache.hudi.common.function.SerializablePairFunction;
-import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.common.util.Functions;
-import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.collection.ImmutablePair;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.keygen.KeyGenerator;
import org.apache.hudi.keygen.factory.HoodieAvroKeyGeneratorFactory;
import org.apache.hudi.storage.StorageConfiguration;
-import org.apache.avro.generic.IndexedRecord;
-
import java.io.IOException;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
import java.util.Map;
-import java.util.Objects;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import static java.util.stream.Collectors.toList;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingFlatMapToPairWrapper;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingFlatMapWrapper;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingForeachWrapper;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingMapToPairWrapper;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingMapWrapper;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingReduceWrapper;
/**
* A java engine implementation of HoodieEngineContext.
*/
-public class HoodieJavaEngineContext extends HoodieEngineContext {
+public class HoodieJavaEngineContext extends BaseLocalEngineContext {
public HoodieJavaEngineContext(StorageConfiguration<?> conf) {
this(conf, new JavaTaskContextSupplier());
@@ -78,121 +42,6 @@ public class HoodieJavaEngineContext extends
HoodieEngineContext {
super(conf, taskContextSupplier);
}
- @Override
- public HoodieAccumulator newAccumulator() {
- return HoodieAtomicLongAccumulator.create();
- }
-
- @Override
- public <T> HoodieData<T> emptyHoodieData() {
- return HoodieListData.eager(Collections.emptyList());
- }
-
- @Override
- public <K, V> HoodiePairData<K, V> emptyHoodiePairData() {
- return HoodieListPairData.eager(Collections.emptyList());
- }
-
- @Override
- public <T> HoodieData<T> parallelize(List<T> data, int parallelism) {
- return HoodieListData.eager(data);
- }
-
- @Override
- public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int
parallelism) {
- return
data.stream().parallel().map(throwingMapWrapper(func)).collect(toList());
- }
-
- @Override
- public <I, K, V> List<V> mapToPairAndReduceByKey(List<I> data,
SerializablePairFunction<I, K, V> mapToPairFunc, SerializableBiFunction<V, V,
V> reduceFunc, int parallelism) {
- return
data.stream().parallel().map(throwingMapToPairWrapper(mapToPairFunc))
- .collect(Collectors.groupingBy(p -> p.getKey())).values().stream()
- .map(list -> list.stream().map(e ->
e.getValue()).reduce(throwingReduceWrapper(reduceFunc)).get())
- .collect(Collectors.toList());
- }
-
- @Override
- public <I, K, V> Stream<ImmutablePair<K, V>>
mapPartitionsToPairAndReduceByKey(Stream<I> data,
SerializablePairFlatMapFunction<Iterator<I>, K, V> flatMapToPairFunc,
-
SerializableBiFunction<V, V, V> reduceFunc, int parallelism) {
- return
throwingFlatMapToPairWrapper(flatMapToPairFunc).apply(data.parallel().iterator())
- .collect(Collectors.groupingBy(Pair::getKey)).entrySet().stream()
- .map(entry -> new ImmutablePair<>(entry.getKey(),
entry.getValue().stream().map(
-
Pair::getValue).reduce(throwingReduceWrapper(reduceFunc)).orElse(null)))
- .filter(Objects::nonNull);
- }
-
- @Override
- public <I, K, V> List<V> reduceByKey(
- List<Pair<K, V>> data, SerializableBiFunction<V, V, V> reduceFunc, int
parallelism) {
- return data.stream().parallel()
- .collect(Collectors.groupingBy(p -> p.getKey())).values().stream()
- .map(list -> list.stream().map(e ->
e.getValue()).reduce(throwingReduceWrapper(reduceFunc)).orElse(null))
- .filter(Objects::nonNull)
- .collect(Collectors.toList());
- }
-
- @Override
- public <I, O> List<O> flatMap(List<I> data, SerializableFunction<I,
Stream<O>> func, int parallelism) {
- return
data.stream().parallel().flatMap(throwingFlatMapWrapper(func)).collect(toList());
- }
-
- @Override
- public <I> void foreach(List<I> data, SerializableConsumer<I> consumer, int
parallelism) {
- data.stream().forEach(throwingForeachWrapper(consumer));
- }
-
- @Override
- public <I, K, V> Map<K, V> mapToPair(List<I> data,
SerializablePairFunction<I, K, V> func, Integer parallelism) {
- return data.stream().map(throwingMapToPairWrapper(func)).collect(
- Collectors.toMap(Pair::getLeft, Pair::getRight, (oldVal, newVal) ->
newVal)
- );
- }
-
- @Override
- public void setProperty(EngineProperty key, String value) {
- // no operation for now
- }
-
- @Override
- public Option<String> getProperty(EngineProperty key) {
- return Option.empty();
- }
-
- @Override
- public void setJobStatus(String activeModule, String activityDescription) {
- // no operation for now
- }
-
- @Override
- public void clearJobStatus() {
- // no operation for now
- }
-
- @Override
- public void putCachedDataIds(HoodieDataCacheKey cacheKey, int... ids) {
- // no operation for now
- }
-
- @Override
- public List<Integer> getCachedDataIds(HoodieDataCacheKey cacheKey) {
- return Collections.emptyList();
- }
-
- @Override
- public List<Integer> removeCachedDataIds(HoodieDataCacheKey cacheKey) {
- return Collections.emptyList();
- }
-
- @Override
- public void cancelJob(String jobId) {
- // no operation for now
- }
-
- @Override
- public void cancelAllJobs() {
- // no operation for now
- }
-
// Allowlist of safe system properties to include in commit metadata. Avoid
wildcarding system
// properties since callers may pass credentials via -D flags (e.g.
-Ddb.password=...).
private static final String[] SAFE_SYSTEM_PROPERTIES = {
@@ -217,21 +66,6 @@ public class HoodieJavaEngineContext extends
HoodieEngineContext {
return info;
}
- @Override
- public <I, O> O aggregate(HoodieData<I> data, O zeroValue,
Functions.Function2<O, I, O> seqOp, Functions.Function2<O, O, O> combOp) {
- return data.collectAsList().stream().reduce(zeroValue, seqOp::apply,
combOp::apply);
- }
-
- @Override
- public ReaderContextFactory<IndexedRecord>
getReaderContextFactory(HoodieTableMetaClient metaClient) {
- return getEngineReaderContextFactory(metaClient);
- }
-
- @Override
- public ReaderContextFactory<IndexedRecord>
getEngineReaderContextFactory(HoodieTableMetaClient metaClient) {
- return new AvroReaderContextFactory(metaClient, new TypedProperties());
- }
-
@Override
public KeyGenerator createKeyGenerator(TypedProperties props) throws
IOException {
return HoodieAvroKeyGeneratorFactory.createKeyGenerator(props);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieLocalEngineContext.java
b/hudi-common/src/main/java/org/apache/hudi/common/engine/BaseLocalEngineContext.java
similarity index 91%
copy from
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieLocalEngineContext.java
copy to
hudi-common/src/main/java/org/apache/hudi/common/engine/BaseLocalEngineContext.java
index acae6245968e..735018e64503 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieLocalEngineContext.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/engine/BaseLocalEngineContext.java
@@ -36,12 +36,10 @@ import org.apache.hudi.common.util.Functions;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.ImmutablePair;
import org.apache.hudi.common.util.collection.Pair;
-import org.apache.hudi.keygen.KeyGenerator;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.avro.generic.IndexedRecord;
-import java.io.IOException;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -59,15 +57,12 @@ import static
org.apache.hudi.common.function.FunctionWrapper.throwingMapWrapper
import static
org.apache.hudi.common.function.FunctionWrapper.throwingReduceWrapper;
/**
- * A java based engine context, use this implementation on the query engine
integrations if needed.
+ * Base engine context that runs everything in the current JVM over in-memory
lists.
+ * Shared by {@link HoodieLocalEngineContext} and the Java client's engine
context.
*/
-public final class HoodieLocalEngineContext extends HoodieEngineContext {
+public abstract class BaseLocalEngineContext extends HoodieEngineContext {
- public HoodieLocalEngineContext(StorageConfiguration<?> conf) {
- this(conf, new LocalTaskContextSupplier());
- }
-
- public HoodieLocalEngineContext(StorageConfiguration<?> conf,
TaskContextSupplier taskContextSupplier) {
+ protected BaseLocalEngineContext(StorageConfiguration<?> conf,
TaskContextSupplier taskContextSupplier) {
super(conf, taskContextSupplier);
}
@@ -202,9 +197,4 @@ public final class HoodieLocalEngineContext extends
HoodieEngineContext {
public ReaderContextFactory<IndexedRecord>
getEngineReaderContextFactory(HoodieTableMetaClient metaClient) {
return new AvroReaderContextFactory(metaClient, new TypedProperties());
}
-
- @Override
- public KeyGenerator createKeyGenerator(TypedProperties props) throws
IOException {
- throw new UnsupportedOperationException("Not yet implemented");
- }
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieLocalEngineContext.java
b/hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieLocalEngineContext.java
index acae6245968e..dc6a8e9dbfcc 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieLocalEngineContext.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieLocalEngineContext.java
@@ -19,49 +19,15 @@
package org.apache.hudi.common.engine;
import org.apache.hudi.common.config.TypedProperties;
-import org.apache.hudi.common.data.HoodieAccumulator;
-import org.apache.hudi.common.data.HoodieAtomicLongAccumulator;
-import org.apache.hudi.common.data.HoodieData;
-import org.apache.hudi.common.data.HoodieData.HoodieDataCacheKey;
-import org.apache.hudi.common.data.HoodieListData;
-import org.apache.hudi.common.data.HoodieListPairData;
-import org.apache.hudi.common.data.HoodiePairData;
-import org.apache.hudi.common.function.SerializableBiFunction;
-import org.apache.hudi.common.function.SerializableConsumer;
-import org.apache.hudi.common.function.SerializableFunction;
-import org.apache.hudi.common.function.SerializablePairFlatMapFunction;
-import org.apache.hudi.common.function.SerializablePairFunction;
-import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.common.util.Functions;
-import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.collection.ImmutablePair;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.keygen.KeyGenerator;
import org.apache.hudi.storage.StorageConfiguration;
-import org.apache.avro.generic.IndexedRecord;
-
import java.io.IOException;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import static java.util.stream.Collectors.toList;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingFlatMapToPairWrapper;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingFlatMapWrapper;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingForeachWrapper;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingMapToPairWrapper;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingMapWrapper;
-import static
org.apache.hudi.common.function.FunctionWrapper.throwingReduceWrapper;
/**
* A java based engine context, use this implementation on the query engine
integrations if needed.
*/
-public final class HoodieLocalEngineContext extends HoodieEngineContext {
+public final class HoodieLocalEngineContext extends BaseLocalEngineContext {
public HoodieLocalEngineContext(StorageConfiguration<?> conf) {
this(conf, new LocalTaskContextSupplier());
@@ -71,138 +37,6 @@ public final class HoodieLocalEngineContext extends
HoodieEngineContext {
super(conf, taskContextSupplier);
}
- @Override
- public HoodieAccumulator newAccumulator() {
- return HoodieAtomicLongAccumulator.create();
- }
-
- @Override
- public <T> HoodieData<T> emptyHoodieData() {
- return HoodieListData.eager(Collections.emptyList());
- }
-
- @Override
- public <K, V> HoodiePairData<K, V> emptyHoodiePairData() {
- return HoodieListPairData.eager(Collections.emptyList());
- }
-
- @Override
- public <T> HoodieData<T> parallelize(List<T> data, int parallelism) {
- return HoodieListData.eager(data);
- }
-
- @Override
- public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int
parallelism) {
- return
data.stream().parallel().map(throwingMapWrapper(func)).collect(toList());
- }
-
- @Override
- public <I, K, V> List<V> mapToPairAndReduceByKey(List<I> data,
SerializablePairFunction<I, K, V> mapToPairFunc,
- SerializableBiFunction<V,
V, V> reduceFunc, int parallelism) {
- return
data.stream().parallel().map(throwingMapToPairWrapper(mapToPairFunc))
- .collect(Collectors.groupingBy(p -> p.getKey())).values().stream()
- .map(list -> list.stream().map(e ->
e.getValue()).reduce(throwingReduceWrapper(reduceFunc)).get())
- .collect(Collectors.toList());
- }
-
- @Override
- public <I, K, V> Stream<ImmutablePair<K, V>>
mapPartitionsToPairAndReduceByKey(
- Stream<I> data, SerializablePairFlatMapFunction<Iterator<I>, K, V>
flatMapToPairFunc,
- SerializableBiFunction<V, V, V> reduceFunc, int parallelism) {
- return
throwingFlatMapToPairWrapper(flatMapToPairFunc).apply(data.parallel().iterator())
- .collect(Collectors.groupingBy(Pair::getKey)).entrySet().stream()
- .map(entry -> new ImmutablePair<>(entry.getKey(),
entry.getValue().stream().map(
-
Pair::getValue).reduce(throwingReduceWrapper(reduceFunc)).orElse(null)))
- .filter(Objects::nonNull);
- }
-
- @Override
- public <I, K, V> List<V> reduceByKey(
- List<Pair<K, V>> data, SerializableBiFunction<V, V, V> reduceFunc, int
parallelism) {
- return data.stream().parallel()
- .collect(Collectors.groupingBy(p -> p.getKey())).values().stream()
- .map(list -> list.stream().map(e ->
e.getValue()).reduce(throwingReduceWrapper(reduceFunc)).orElse(null))
- .filter(Objects::nonNull)
- .collect(Collectors.toList());
- }
-
- @Override
- public <I, O> List<O> flatMap(List<I> data, SerializableFunction<I,
Stream<O>> func, int parallelism) {
- return
data.stream().parallel().flatMap(throwingFlatMapWrapper(func)).collect(toList());
- }
-
- @Override
- public <I> void foreach(List<I> data, SerializableConsumer<I> consumer, int
parallelism) {
- data.stream().forEach(throwingForeachWrapper(consumer));
- }
-
- @Override
- public <I, K, V> Map<K, V> mapToPair(List<I> data,
SerializablePairFunction<I, K, V> func, Integer parallelism) {
- return data.stream().map(throwingMapToPairWrapper(func)).collect(
- Collectors.toMap(Pair::getLeft, Pair::getRight, (oldVal, newVal) ->
newVal)
- );
- }
-
- @Override
- public void setProperty(EngineProperty key, String value) {
- // no operation for now
- }
-
- @Override
- public Option<String> getProperty(EngineProperty key) {
- return Option.empty();
- }
-
- @Override
- public void setJobStatus(String activeModule, String activityDescription) {
- // no operation for now
- }
-
- @Override
- public void clearJobStatus() {
- // no operation for now
- }
-
- @Override
- public void putCachedDataIds(HoodieDataCacheKey cacheKey, int... ids) {
- // no operation for now
- }
-
- @Override
- public List<Integer> getCachedDataIds(HoodieDataCacheKey cacheKey) {
- return Collections.emptyList();
- }
-
- @Override
- public List<Integer> removeCachedDataIds(HoodieDataCacheKey cacheKey) {
- return Collections.emptyList();
- }
-
- @Override
- public void cancelJob(String jobId) {
- // no operation for now
- }
-
- @Override
- public void cancelAllJobs() {
- // no operation for now
- }
-
- @Override
- public <I, O> O aggregate(HoodieData<I> data, O zeroValue,
Functions.Function2<O, I, O> seqOp, Functions.Function2<O, O, O> combOp) {
- return data.collectAsList().stream().reduce(zeroValue, seqOp::apply,
combOp::apply);
- }
-
- @Override
- public ReaderContextFactory<IndexedRecord>
getReaderContextFactory(HoodieTableMetaClient metaClient) {
- return getEngineReaderContextFactory(metaClient);
- }
-
- @Override
- public ReaderContextFactory<IndexedRecord>
getEngineReaderContextFactory(HoodieTableMetaClient metaClient) {
- return new AvroReaderContextFactory(metaClient, new TypedProperties());
- }
-
@Override
public KeyGenerator createKeyGenerator(TypedProperties props) throws
IOException {
throw new UnsupportedOperationException("Not yet implemented");
