Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[fix](hudi) upgrade hudi to 0.15.0 #44267

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[fix](hudi) upgrade hudi to 0.15.0 #44267

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
758aa44
upgrade hudi to 0.15.0
suxiaogang223 Nov 19, 2024
5900c82
copy HoodieLocalEngineContext
suxiaogang223 Nov 19, 2024
a4efb00
fix checkstyle
suxiaogang223 Nov 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 27 additions & 40 deletions fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiLocalEngineContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,6 @@

package org.apache.doris.datasource.hudi.source;

import org.apache.doris.datasource.hive.HiveMetaStoreClientHelper;

import org.apache.hadoop.conf.Configuration;
import org.apache.hudi.common.config.SerializableConfiguration;
import org.apache.hudi.common.data.HoodieAccumulator;
import org.apache.hudi.common.data.HoodieAtomicLongAccumulator;
import org.apache.hudi.common.data.HoodieData;
Expand All @@ -39,7 +35,7 @@
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.exception.HoodieException;
import org.apache.hudi.storage.StorageConfiguration;

import java.util.Collections;
import java.util.Iterator;
Expand All @@ -50,18 +46,20 @@
import java.util.stream.Stream;

/**
* This file is copied from org.apache.hudi.common.engine.HoodieLocalEngineContext.
* This file is copied from
* org.apache.hudi.common.engine.HudiLocalEngineContext.
* Because we need set ugi in thread pool
* A java based engine context, use this implementation on the query engine integrations if needed.
* A java based engine context, use this implementation on the query engine
* integrations if needed.
*/
public final class HudiLocalEngineContext extends HoodieEngineContext {

public HudiLocalEngineContext(Configuration conf) {
public HudiLocalEngineContext(StorageConfiguration<?> conf) {
this(conf, new LocalTaskContextSupplier());
}

public HudiLocalEngineContext(Configuration conf, TaskContextSupplier taskContextSupplier) {
super(new SerializableConfiguration(conf), taskContextSupplier);
public HudiLocalEngineContext(StorageConfiguration<?> conf, TaskContextSupplier taskContextSupplier) {
super(conf, taskContextSupplier);
}

@Override
Expand All @@ -81,57 +79,47 @@ public <T> HoodieData<T> parallelize(List<T> data, int parallelism) {

@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
return data.stream().parallel().map(v1 -> {
try {
return HiveMetaStoreClientHelper.ugiDoAs(getHadoopConf().get(), () -> func.apply(v1));
} catch (Exception e) {
throw new HoodieException("Error occurs when executing map", e);
}
}).collect(Collectors.toList());
return data.stream().parallel().map(FunctionWrapper.throwingMapWrapper(func)).collect(Collectors.toList());
}

@Override
public <I, K, V> List<V> mapToPairAndReduceByKey(
List<I> data,
SerializablePairFunction<I, K, V> mapToPairFunc,
SerializableBiFunction<V, V, V> reduceFunc, int parallelism) {
List<I> data, SerializablePairFunction<I, K, V> mapToPairFunc, SerializableBiFunction<V, V, V> reduceFunc,
int parallelism) {
return data.stream().parallel().map(FunctionWrapper.throwingMapToPairWrapper(mapToPairFunc))
.collect(Collectors.groupingBy(p -> p.getKey())).values().stream()
.map(list ->
list.stream()
.map(e -> e.getValue())
.collect(Collectors.groupingBy(p -> p.getKey())).values().stream()
.map(list -> list.stream().map(e -> e.getValue())
.reduce(FunctionWrapper.throwingReduceWrapper(reduceFunc)).get())
.collect(Collectors.toList());
.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 FunctionWrapper.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(FunctionWrapper.throwingReduceWrapper(reduceFunc)).orElse(null)))
.filter(Objects::nonNull);
.collect(Collectors.groupingBy(Pair::getKey)).entrySet().stream()
.map(entry -> new ImmutablePair<>(entry.getKey(), entry.getValue().stream().map(
Pair::getValue).reduce(FunctionWrapper.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(FunctionWrapper.throwingReduceWrapper(reduceFunc)).orElse(null))
.filter(Objects::nonNull)
.collect(Collectors.toList());
.collect(Collectors.groupingBy(p -> p.getKey())).values().stream()
.map(list -> list.stream().map(e -> e.getValue())
.reduce(FunctionWrapper.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(FunctionWrapper.throwingFlatMapWrapper(func)).collect(Collectors.toList());
return data.stream().parallel().flatMap(FunctionWrapper.throwingFlatMapWrapper(func))
.collect(Collectors.toList());
}

@Override
Expand All @@ -142,8 +130,7 @@ public <I> void foreach(List<I> data, SerializableConsumer<I> consumer, int para
@Override
public <I, K, V> Map<K, V> mapToPair(List<I> data, SerializablePairFunction<I, K, V> func, Integer parallelism) {
return data.stream().map(FunctionWrapper.throwingMapToPairWrapper(func)).collect(
Collectors.toMap(Pair::getLeft, Pair::getRight, (oldVal, newVal) -> newVal)
);
Collectors.toMap(Pair::getLeft, Pair::getRight, (oldVal, newVal) -> newVal));
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion fe/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -317,7 +317,7 @@ under the License.
<avro.version>1.11.4</avro.version>
<arrow.version>17.0.0</arrow.version>
<!-- hudi -->
<hudi.version>0.14.1</hudi.version>
<hudi.version>0.15.0</hudi.version>
<presto.hadoop.version>2.7.4-11</presto.hadoop.version>
<presto.hive.version>3.0.0-8</presto.hive.version>
<!-- lakesoul -->
Expand Down
Loading