database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[fix](hudi) return empty if there is no commit implemented (#37703)

Browse Source 
bp: #37702
This commit is contained in:
Ashin Gau
2024-07-12 22:44:58 +08:00
committed by GitHub
parent f2556ba182
commit 019cd9b4ec
2 changed files with 80 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/EmptyIncrementalRelation.java Normal file
View File

@ -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<String, String> optParams;
public EmptyIncrementalRelation(Map<String, String> optParams) {
this.optParams = optParams;
}
@Override
public List<FileSlice> collectFileSlices() throws HoodieException {
return Collections.emptyList();
}
@Override
public List<Split> collectSplits() throws HoodieException {
return Collections.emptyList();
}
@Override
public Map<String, String> 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;
}
}

5
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalHudiScan.java
View File

@ -23,6 +23,7 @@ import org.apache.doris.datasource.ExternalTable;
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 class LogicalHudiScan extends LogicalFileScan {
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 {