Copilot commented on code in PR #6924: URL: https://github.com/apache/hive/pull/6924#discussion_r4512335985
########## iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/RewritableDeletes.java: ########## @@ -0,0 +1,155 @@ +/* + * 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.iceberg.mr.hive.writer; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.util.Map; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.ql.exec.ObjectCache; +import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory; +import org.apache.hadoop.hive.ql.exec.tez.DagUtils; +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.mapred.JobConf; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableScan; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.mr.hive.IcebergTableUtil; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.util.DeleteFileSet; +import org.apache.iceberg.util.SerializationUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The previous deletes that the DV writers of a merge-on-read DML merge into the DVs they write: a DV replaces every + * position delete of its data file. They reach the writers in a temp file of the query that the job's configuration + * names, read once per LLAP daemon. + */ +public final class RewritableDeletes { + private static final Logger LOG = LoggerFactory.getLogger(RewritableDeletes.class); + + private static final String LOCATION_PREFIX = "iceberg.mr.rewritable.deletes."; + + // by data file location + private final Map<String, DeleteFileSet> deletes; + + private RewritableDeletes(Map<String, DeleteFileSet> deletes) { + this.deletes = deletes; + } + + /** + * Adds the rewritable deletes of a table to the job, if there are any. + * @param table The table the DML writes. + * @param tableName The name of the table in the job's configuration. + * @param ref The branch the DML writes. + * @param jobConf The job's configuration, with the temp dir of the query. + */ + public static void addToJob(Table table, String tableName, String ref, JobConf jobConf) { + String key = LOCATION_PREFIX + tableName; + if (jobConf.get(key) != null) { + return; + } + Snapshot snapshot = IcebergTableUtil.getTableSnapshot(table, ref); + if (snapshot == null) { + return; + } + TableScan scan = table.newScan().useSnapshot(snapshot.snapshotId()); + Map<String, DeleteFileSet> deletes = rewritableDeletes(scan); + if (deletes.isEmpty()) { + return; + } + Path path = new Path(jobConf.get(DagUtils.TEZ_TMP_DIR_KEY), tableName + "-rewritable-deletes"); Review Comment: The new hook is also reached by MapReduce through ExecDriver → MapredWork → BaseWork, but only TezTask initializes TEZ_TMP_DIR_KEY. With hive.execution.engine=mr, a V3 merge-on-read DELETE/UPDATE against a table with existing position deletes therefore reaches this constructor with a null parent and fails before job submission. Please initialize a query-owned scratch path in the MR submission path as well, or pass an engine-independent scratch directory into this hook. ########## iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/RewritableDeletes.java: ########## @@ -0,0 +1,155 @@ +/* + * 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.iceberg.mr.hive.writer; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.util.Map; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.ql.exec.ObjectCache; +import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory; +import org.apache.hadoop.hive.ql.exec.tez.DagUtils; +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.mapred.JobConf; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableScan; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.mr.hive.IcebergTableUtil; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.util.DeleteFileSet; +import org.apache.iceberg.util.SerializationUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The previous deletes that the DV writers of a merge-on-read DML merge into the DVs they write: a DV replaces every + * position delete of its data file. They reach the writers in a temp file of the query that the job's configuration + * names, read once per LLAP daemon. + */ +public final class RewritableDeletes { + private static final Logger LOG = LoggerFactory.getLogger(RewritableDeletes.class); + + private static final String LOCATION_PREFIX = "iceberg.mr.rewritable.deletes."; + + // by data file location + private final Map<String, DeleteFileSet> deletes; + + private RewritableDeletes(Map<String, DeleteFileSet> deletes) { + this.deletes = deletes; + } + + /** + * Adds the rewritable deletes of a table to the job, if there are any. + * @param table The table the DML writes. + * @param tableName The name of the table in the job's configuration. + * @param ref The branch the DML writes. + * @param jobConf The job's configuration, with the temp dir of the query. + */ + public static void addToJob(Table table, String tableName, String ref, JobConf jobConf) { + String key = LOCATION_PREFIX + tableName; + if (jobConf.get(key) != null) { + return; + } + Snapshot snapshot = IcebergTableUtil.getTableSnapshot(table, ref); + if (snapshot == null) { + return; + } + TableScan scan = table.newScan().useSnapshot(snapshot.snapshotId()); + Map<String, DeleteFileSet> deletes = rewritableDeletes(scan); + if (deletes.isEmpty()) { + return; + } + Path path = new Path(jobConf.get(DagUtils.TEZ_TMP_DIR_KEY), tableName + "-rewritable-deletes"); + try (OutputStream out = new GZIPOutputStream(path.getFileSystem(jobConf).create(path))) { + out.write(SerializationUtil.serializeToBytes(deletes)); + } catch (IOException e) { + throw new UncheckedIOException(String.format("Failed to write rewritable deletes to %s", path), e); + } + jobConf.set(key, path.toString()); + LOG.info("Added rewritable deletes of {} data files to the job: {}", deletes.size(), path); + } + + /** + * Returns the rewritable deletes of a table that the job carries. + * @param tableName The name of the table in the task's configuration. + * @param conf The task's configuration. + * @return The rewritable deletes by data file location. + */ + public static Map<String, DeleteFileSet> fromJob(String tableName, Configuration conf) { + String location = conf.get(LOCATION_PREFIX + tableName); + if (location == null) { + return Map.of(); + } + ObjectCache cache = + ObjectCacheFactory.getCache(conf, conf.get(HiveConf.ConfVars.HIVE_QUERY_ID.varname), false, true); + if (cache == null) { + return readFile(location, conf).deletes; + } + try { + return cache.retrieve(location, () -> readFile(location, conf)).deletes; Review Comment: `LlapObjectCache` uses soft references, but writers retain only the deletes map, not the cached wrapper. Under memory pressure the wrapper can be reclaimed while its map is still in use, so later writers deserialize duplicate maps into the same daemon. Cache the map itself for LLAP, while keeping the bounded-logging wrapper for the Tez container cache. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
