Copilot commented on code in PR #6924: URL: https://github.com/apache/hive/pull/6924#discussion_r4513295023
########## iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/RewritableDeletes.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.ForwardingMap; +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 extends ForwardingMap<String, DeleteFileSet> { + 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; + } + + @Override + protected Map<String, DeleteFileSet> delegate() { + return 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: `DagUtils.TEZ_TMP_DIR_KEY` being unset will cause a null/invalid Path and fail planning with a runtime exception. Add a guard with a clear error message or fall back to a safe scratch/temp directory (e.g., Hive scratch dir) when the key is missing, since this method is callable whenever storage handler JobConf config runs. ########## iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/RewritableDeletes.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.ForwardingMap; +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 extends ForwardingMap<String, DeleteFileSet> { + 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; + } + + @Override + protected Map<String, DeleteFileSet> delegate() { + return 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> getFromJob(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); + } + try { + // the writers keep the cached value itself: the LLAP object cache holds its values softly + return cache.retrieve(location, () -> readFile(location, conf)); + } catch (HiveException e) { + throw new RuntimeException(e); + } + } + + // a DV replaces every position delete of its data file + private static Map<String, DeleteFileSet> rewritableDeletes(TableScan scan) { + Map<String, DeleteFileSet> deletes = Maps.newHashMap(); + + try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) { + for (FileScanTask task : tasks) { + for (DeleteFile delete : task.deletes()) { + if (delete.content() == FileContent.POSITION_DELETES) { + deletes.computeIfAbsent(task.file().location(), ignored -> DeleteFileSet.create()).add(delete); + } + } + } + } catch (IOException e) { + throw new UncheckedIOException(String.format("Failed to close table scan: %s", scan), e); + } + return deletes; + } + + private static RewritableDeletes readFile(String location, Configuration conf) { + LOG.info("Reading rewritable deletes from {}", location); + Path path = new Path(location); + try (InputStream in = new GZIPInputStream(path.getFileSystem(conf).open(path))) { + return new RewritableDeletes(SerializationUtil.deserializeFromBytes(in.readAllBytes())); + } catch (IOException e) { + throw new UncheckedIOException(String.format("Failed to read rewritable deletes from %s", location), e); + } + } Review Comment: `InputStream#readAllBytes()` is Java 9+ and will break builds on Java 8. Use a Java-8-compatible utility to read the stream fully (or refactor `SerializationUtil` usage to deserialize from a stream if available). ########## iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java: ########## @@ -391,6 +394,22 @@ private static void configureOutputTableJobConf(TableDesc tableDesc, JobConf job } } + /** + * The DV writers of a merge-on-read DML merge the previous deletes of the data files they write. These are planned + * while the job is built. + */ + @Override + public void configureJobConf(FileSinkDesc sinkDesc, JobConf jobConf) { + TableDesc tableDesc = sinkDesc.getTableInfo(); + configureJobConf(tableDesc, jobConf); + Properties props = tableDesc.getProperties(); + if (sinkDesc.getWriteOperation() == Operation.DELETE && !sinkDesc.isCopyOnWrite() && + IcebergTableUtil.formatVersion(Maps.fromProperties(props)) >= 3) { + RewritableDeletes.addToJob(IcebergTableUtil.getTable(jobConf, props), tableDesc.getTableName(), Review Comment: The `tableName` identifier used to store rewritable-deletes location in the JobConf must match what writer tasks use to look it up. Here `tableDesc.getTableName()` may be unqualified (table only), while the writer side uses `hive_metastoreConstants.META_TABLE_NAME` (often db-qualified). Consider using a consistent fully-qualified identifier (e.g., `dbName.tableName`) on both sides to avoid silent cache misses and losing the intended optimization. ########## iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/WriterBuilder.java: ########## @@ -31,42 +29,29 @@ import java.util.function.Supplier; import java.util.function.UnaryOperator; import java.util.stream.Collectors; -import org.apache.commons.lang3.ObjectUtils; import org.apache.hadoop.hive.ql.Context.Operation; import org.apache.hadoop.hive.ql.security.authorization.HiveCustomStorageHandlerUtils; import org.apache.hadoop.hive.ql.session.SessionStateUtil; import org.apache.hadoop.mapred.TaskAttemptID; -import org.apache.iceberg.BatchScan; -import org.apache.iceberg.DeleteFile; -import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; -import org.apache.iceberg.FileScanTask; import org.apache.iceberg.MetadataColumns; -import org.apache.iceberg.ScanTask; -import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; import org.apache.iceberg.deletes.DeleteGranularity; -import org.apache.iceberg.expressions.Expression; -import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.OutputFileFactory; import org.apache.iceberg.mr.Catalogs; -import org.apache.iceberg.mr.InputFormatConfig; import org.apache.iceberg.mr.hive.IcebergTableUtil; -import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; -import org.apache.iceberg.util.ContentFileUtil; import org.apache.iceberg.util.DeleteFileSet; import org.apache.iceberg.util.PropertyUtil; -import org.apache.iceberg.util.SerializationUtil; import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT; import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT; import static org.apache.iceberg.TableProperties.DELETE_DEFAULT_FILE_FORMAT; public class WriterBuilder { private final Table table; - private final Supplier<Map<String, DeleteFileSet>> rewritableDeletes; + private Supplier<Map<String, DeleteFileSet>> rewritableDeletes; private final Context context; Review Comment: `rewritableDeletes` is no longer initialized in the constructor, so `build()` can NPE if any call path forgets to set it via `rewritableDeletes(...)`. Initialize it defensively (e.g., default to a supplier returning `Map.of()`), or enforce non-null by throwing an explicit error in `build()` when missing. ########## iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergDeletionVectors.java: ########## @@ -54,6 +85,51 @@ private List<Record> records(long from, long to) { return records; } + private Table createTable(String name, PartitionSpec spec) { + return testTables.createTable(shell, name, HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA, spec, fileFormat, + records(0, ROWS), formatVersion, ImmutableMap.of("write.parquet.row-group-size-bytes", "1024")); + } + + private Table createSingleFileTable(String name) { + Table table = createTable(name, PartitionSpec.unpartitioned()); + List<FileScanTask> files = planFiles(table); + Assert.assertEquals("one data file", 1, files.size()); + Assert.assertTrue("multiple row groups", files.get(0).file().splitOffsets().size() > 1); + return table; + } + + private static List<FileScanTask> planFiles(Table table) { + return planFiles(table.newScan()); + } + + private static List<FileScanTask> planFiles(TableScan scan) { + try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) { + return StreamSupport.stream(tasks.spliterator(), false).toList(); + } catch (IOException e) { + throw new RuntimeException(e); + } + } Review Comment: `Stream#toList()` is not available on Java 8/11. Replace with `collect(Collectors.toList())` to keep tests compiling under the project's supported Java level. ########## ql/src/java/org/apache/hadoop/hive/ql/parse/rewrite/sql/NonNativeAcidMultiInsertSqlGenerator.java: ########## @@ -94,4 +94,13 @@ public List<String> getSortKeys(Operation operation) { HiveUtils.unparseIdentifier(deletePrefix + fieldSchema.getName(), this.conf))) .collect(Collectors.toList()); } + + @Override + public List<String> getDistributeKeys(Operation operation) { + return targetTable.getStorageHandler().acidDistributeColumns(targetTable, operation) + .stream() + .map(fieldSchema -> qualify( + HiveUtils.unparseIdentifier(deletePrefix + fieldSchema.getName(), this.conf))) + .toList(); Review Comment: `Stream#toList()` requires newer Java than the rest of this module typically targets (many Hive branches still compile with Java 8/11). Use `collect(Collectors.toList())` (and add/import `Collectors` if needed) to avoid compilation failures. -- 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]
