You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by dk...@apache.org on 2023/09/13 14:22:39 UTC
[hive] branch master updated: HIVE-27309: Large number of partitions and small files causes OOM in query coordinator (Dmitriy Fingerman, reviewed by Butao Zhang, Denys Kuzmenko)
This is an automated email from the ASF dual-hosted git repository.
dkuzmenko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 88bc8269a64 HIVE-27309: Large number of partitions and small files causes OOM in query coordinator (Dmitriy Fingerman, reviewed by Butao Zhang, Denys Kuzmenko)
88bc8269a64 is described below
commit 88bc8269a64d31eee372bf3602933c75283c686b
Author: Dmitriy Fingerman <dm...@gmail.com>
AuthorDate: Wed Sep 13 10:22:33 2023 -0400
HIVE-27309: Large number of partitions and small files causes OOM in query coordinator (Dmitriy Fingerman, reviewed by Butao Zhang, Denys Kuzmenko)
Closes #4645
---
.../apache/iceberg/mr/mapreduce/IcebergInputFormat.java | 14 +++++++++-----
.../org/apache/iceberg/mr/mapreduce/IcebergSplit.java | 17 +----------------
2 files changed, 10 insertions(+), 21 deletions(-)
diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
index 7020d705cac..dc50a1e3401 100644
--- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
+++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
@@ -57,7 +57,6 @@ import org.apache.iceberg.Partitioning;
import org.apache.iceberg.Scan;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
-import org.apache.iceberg.SerializableTable;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
@@ -199,7 +198,12 @@ public class IcebergInputFormat<T> extends InputFormat<Void, T> {
Configuration conf = context.getConfiguration();
Table table = Optional
.ofNullable(HiveIcebergStorageHandler.table(conf, conf.get(InputFormatConfig.TABLE_IDENTIFIER)))
- .orElseGet(() -> Catalogs.loadTable(conf));
+ .orElseGet(() -> {
+ Table tbl = Catalogs.loadTable(conf);
+ conf.set(InputFormatConfig.TABLE_IDENTIFIER, tbl.name());
+ conf.set(InputFormatConfig.SERIALIZED_TABLE_PREFIX + tbl.name(), SerializationUtil.serializeToBase64(tbl));
+ return tbl;
+ });
List<InputSplit> splits = Lists.newArrayList();
boolean applyResidual = !conf.getBoolean(InputFormatConfig.SKIP_RESIDUAL_FILTERING, false);
@@ -215,14 +219,13 @@ public class IcebergInputFormat<T> extends InputFormat<Void, T> {
}
try (CloseableIterable<CombinedScanTask> tasksIterable = scan.planTasks()) {
- Table serializableTable = SerializableTable.copyOf(table);
tasksIterable.forEach(task -> {
if (applyResidual && (model == InputFormatConfig.InMemoryDataModel.HIVE ||
model == InputFormatConfig.InMemoryDataModel.PIG)) {
// TODO: We do not support residual evaluation for HIVE and PIG in memory data model yet
checkResiduals(task);
}
- splits.add(new IcebergSplit(serializableTable, conf, task));
+ splits.add(new IcebergSplit(conf, task));
});
} catch (IOException e) {
throw new UncheckedIOException(String.format("Failed to close table scan: %s", scan), e);
@@ -296,7 +299,8 @@ public class IcebergInputFormat<T> extends InputFormat<Void, T> {
CombinedScanTask task = ((IcebergSplit) split).task();
this.context = newContext;
this.conf = newContext.getConfiguration();
- this.table = ((IcebergSplit) split).table();
+ this.table = SerializationUtil.deserializeFromBase64(
+ conf.get(InputFormatConfig.SERIALIZED_TABLE_PREFIX + conf.get(InputFormatConfig.TABLE_IDENTIFIER)));
HiveIcebergStorageHandler.checkAndSetIoConfig(conf, table);
this.tasks = task.files().iterator();
this.nameMapping = table.properties().get(TableProperties.DEFAULT_NAME_MAPPING);
diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergSplit.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergSplit.java
index 8bc332eaa43..420bf9ea3b1 100644
--- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergSplit.java
+++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergSplit.java
@@ -26,7 +26,6 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.iceberg.CombinedScanTask;
import org.apache.iceberg.FileScanTask;
-import org.apache.iceberg.Table;
import org.apache.iceberg.hadoop.Util;
import org.apache.iceberg.mr.InputFormatConfig;
import org.apache.iceberg.util.SerializationUtil;
@@ -37,7 +36,6 @@ public class IcebergSplit extends InputSplit implements org.apache.hadoop.mapred
public static final String[] ANYWHERE = new String[]{"*"};
- private Table table;
private CombinedScanTask task;
private transient String[] locations;
@@ -47,8 +45,7 @@ public class IcebergSplit extends InputSplit implements org.apache.hadoop.mapred
public IcebergSplit() {
}
- IcebergSplit(Table table, Configuration conf, CombinedScanTask task) {
- this.table = table;
+ IcebergSplit(Configuration conf, CombinedScanTask task) {
this.task = task;
this.conf = conf;
}
@@ -83,10 +80,6 @@ public class IcebergSplit extends InputSplit implements org.apache.hadoop.mapred
@Override
public void write(DataOutput out) throws IOException {
- byte[] tableData = SerializationUtil.serializeToBytes(table);
- out.writeInt(tableData.length);
- out.write(tableData);
-
byte[] data = SerializationUtil.serializeToBytes(this.task);
out.writeInt(data.length);
out.write(data);
@@ -94,16 +87,8 @@ public class IcebergSplit extends InputSplit implements org.apache.hadoop.mapred
@Override
public void readFields(DataInput in) throws IOException {
- byte[] tableData = new byte[in.readInt()];
- in.readFully(tableData);
- this.table = SerializationUtil.deserializeFromBytes(tableData);
-
byte[] data = new byte[in.readInt()];
in.readFully(data);
this.task = SerializationUtil.deserializeFromBytes(data);
}
-
- public Table table() {
- return table;
- }
}