From 8af3a0b206e7a4dd2c7da35eb8e38cd16e2102b9 Mon Sep 17 00:00:00 2001 From: gaoxin Date: Fri, 12 Jul 2024 11:44:48 +0800 Subject: [PATCH] [fix](hudi) return empty if there is no commit implemented --- .../hudi/source/EmptyIncrementalRelation.java | 76 +++++++++++++++++++ .../trees/plans/logical/LogicalHudiScan.java | 5 +- 2 files changed, 80 insertions(+), 1 deletion(-) create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/EmptyIncrementalRelation.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/EmptyIncrementalRelation.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/EmptyIncrementalRelation.java new file mode 100644 index 00000000000000..c483bc46cfcc77 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/EmptyIncrementalRelation.java @@ -0,0 +1,76 @@ +// 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.doris.datasource.hudi.source; + +import org.apache.doris.spi.Split; + +import org.apache.hudi.common.model.FileSlice; +import org.apache.hudi.exception.HoodieException; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +public class EmptyIncrementalRelation implements IncrementalRelation { + private static final String EMPTY_TS = "000"; + + private final Map optParams; + + public EmptyIncrementalRelation(Map optParams) { + this.optParams = optParams; + } + + @Override + public List collectFileSlices() throws HoodieException { + return Collections.emptyList(); + } + + @Override + public List collectSplits() throws HoodieException { + return Collections.emptyList(); + } + + @Override + public Map getHoodieParams() { + optParams.put("hoodie.datasource.read.incr.operation", "true"); + optParams.put("hoodie.datasource.read.begin.instanttime", EMPTY_TS); + optParams.put("hoodie.datasource.read.end.instanttime", EMPTY_TS); + optParams.put("hoodie.datasource.read.incr.includeStartTime", "true"); + return optParams; + } + + @Override + public boolean fallbackFullTableScan() { + return false; + } + + @Override + public boolean isIncludeStartTime() { + return true; + } + + @Override + public String getStartTs() { + return EMPTY_TS; + } + + @Override + public String getEndTs() { + return EMPTY_TS; + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalHudiScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalHudiScan.java index 8659ad3d9c3f63..deeca65efcbd1e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalHudiScan.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalHudiScan.java @@ -23,6 +23,7 @@ import org.apache.doris.datasource.hive.HMSExternalTable; import org.apache.doris.datasource.hive.HiveMetaStoreClientHelper; import org.apache.doris.datasource.hudi.source.COWIncrementalRelation; +import org.apache.doris.datasource.hudi.source.EmptyIncrementalRelation; import org.apache.doris.datasource.hudi.source.IncrementalRelation; import org.apache.doris.datasource.hudi.source.MORIncrementalRelation; import org.apache.doris.nereids.exceptions.AnalysisException; @@ -206,7 +207,9 @@ public LogicalHudiScan withScanParams(HMSExternalTable table, TableScanParams sc LOG.warn("Execute incremental read on RO table: {}", table.getFullQualifiers()); } } - if (isCowOrRoTable) { + if (hudiClient.getCommitsTimeline().filterCompletedInstants().countInstants() == 0) { + newIncrementalRelation = Optional.of(new EmptyIncrementalRelation(optParams)); + } else if (isCowOrRoTable) { newIncrementalRelation = Optional.of(new COWIncrementalRelation( optParams, HiveMetaStoreClientHelper.getConfiguration(table), hudiClient)); } else {