lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

Returning empty Statues for an empty spark partition caused due to incorrect bin packing

Browse Source
This commit is contained in:
Nishith Agarwal
2018-11-26 23:17:32 -08:00
committed by vinoth chandar
parent 0015c9b00e
commit e83dde3b95
1 changed files with 10 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
hoodie-client/src/main/java/com/uber/hoodie/table/HoodieCopyOnWriteTable.java
View File

@@ -176,6 +176,11 @@ public class HoodieCopyOnWriteTable<T extends HoodieRecordPayload> extends Hoodi
public Iterator<List<WriteStatus>> handleUpdate(String commitTime, String fileId, public Iterator<List<WriteStatus>> handleUpdate(String commitTime, String fileId,
Iterator<HoodieRecord<T>> recordItr) throws IOException { Iterator<HoodieRecord<T>> recordItr) throws IOException {
// This is needed since sometimes some buckets are never picked in getPartition() and end up with 0 records
if (!recordItr.hasNext()) {
logger.info("Empty partition with fileId => " + fileId);
return Collections.singletonList((List<WriteStatus>) Collections.EMPTY_LIST).iterator();
}
// these are updates // these are updates
HoodieMergeHandle upsertHandle = getUpdateHandle(commitTime, fileId, recordItr); HoodieMergeHandle upsertHandle = getUpdateHandle(commitTime, fileId, recordItr);
return handleUpdateInternal(upsertHandle, commitTime, fileId); return handleUpdateInternal(upsertHandle, commitTime, fileId);
@@ -235,6 +240,11 @@ public class HoodieCopyOnWriteTable<T extends HoodieRecordPayload> extends Hoodi
public Iterator<List<WriteStatus>> handleInsert(String commitTime, public Iterator<List<WriteStatus>> handleInsert(String commitTime,
Iterator<HoodieRecord<T>> recordItr) throws Exception { Iterator<HoodieRecord<T>> recordItr) throws Exception {
// This is needed since sometimes some buckets are never picked in getPartition() and end up with 0 records
if (!recordItr.hasNext()) {
logger.info("Empty partition");
return Collections.singletonList((List<WriteStatus>) Collections.EMPTY_LIST).iterator();
}
return new CopyOnWriteLazyInsertIterable<>(recordItr, config, commitTime, this); return new CopyOnWriteLazyInsertIterable<>(recordItr, config, commitTime, this);
} }