Aggarwal-Raghav commented on code in PR #6707: URL: https://github.com/apache/hive/pull/6707#discussion_r3902802749
########## iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/stats/IcebergColStatsWriter.java: ########## @@ -0,0 +1,633 @@ +/* + * 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.stats; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.function.BooleanSupplier; +import java.util.function.Predicate; +import java.util.stream.Collectors; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.Constants; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.conf.HiveConf.ConfVars; +import org.apache.hadoop.hive.metastore.api.ColumnStatistics; +import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj; +import org.apache.hadoop.hive.metastore.api.InvalidObjectException; +import org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils; +import org.apache.hadoop.hive.ql.Context.RewritePolicy; +import org.apache.hadoop.hive.ql.parse.ColumnStatsSemanticAnalyzer; +import org.apache.hadoop.hive.ql.plan.HiveOperation; +import org.apache.hadoop.hive.ql.session.SessionState; +import org.apache.hadoop.hive.ql.session.SessionStateUtil; +import org.apache.hadoop.hive.ql.txn.compactor.CompactorContext; +import org.apache.iceberg.GenericBlobMetadata; +import org.apache.iceberg.GenericStatisticsFile; +import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotSummary; +import org.apache.iceberg.StatisticsFile; +import org.apache.iceberg.Table; +import org.apache.iceberg.mr.hive.IcebergTableUtil; +import org.apache.iceberg.mr.hive.compaction.IcebergCompactionService; +import org.apache.iceberg.puffin.Blob; +import org.apache.iceberg.puffin.BlobMetadata; +import org.apache.iceberg.puffin.Puffin; +import org.apache.iceberg.puffin.PuffinCompressionCodec; +import org.apache.iceberg.puffin.PuffinReader; +import org.apache.iceberg.puffin.PuffinWriter; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.Iterators; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.ByteBuffers; +import org.apache.iceberg.util.Pair; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Writes the column statistics of one gather as the table's statistics file, per the policy the + * write's facts resolve to: replacing the stored file, merging into it by carrying what no write + * since has changed, or leaving it alone. The reading side is {@link IcebergColStatsReader}. + * + * At table level the file holds one blob per column. At partition level it holds one blob per + * partition, pulled and written one at a time so the whole of a large table's statistics is never + * held at once. A partition blob frames one slice per column behind a small header, so a read + * deserializes only the columns it was asked for: + * + * content := version, count, header length, count x (column name, stored length, raw length), + * slices + * slice := one column's ColumnStatisticsObj, Java-serialized and zstd-compressed on its own + * + * The blob itself is not compressed, and the header names every slice's position, so a reader can + * fetch the columns it was asked for and no others; the table-level blobs, one small one per + * column, stay compressed whole. + * + * The frame travels under its own blob type, so a file of the older layout reads as absent rather + * than wrong, and a version bump can change the frame without renaming the type. + */ +public final class IcebergColStatsWriter { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergColStatsWriter.class); + + /** + * The blob type of every entry this writer stores. A table-level entry holds one column bare; a + * partition entry carries the partition name as a property and frames its columns, and the frame + * is versioned apart from the name. + */ + /** + * What a blob of one partition's entries is named. It holds a frame of its own rather than a + * single entry, which is why it is named apart from a table-level one, and the frame states its + * own version, which is why the name does not. + */ + public static final String HIVE_PART_COL_STATS_BLOB_V1 = "hive-partition-column-statistics-v1"; + /** + * What a table-level entry is named now that it holds a Thrift struct and states its sketches + * apart. A reader that knows neither name reads it as absent, and one that knows both reads + * whichever it finds, so nothing has to be recomputed to move between them. + */ + public static final String HIVE_COL_STATS_BLOB_V1 = "hive-table-column-statistics-v1"; + /** What a table-level entry's sketches are named, which only a read that wants them fetches. */ + /** + * What a released writer named a table-level entry, holding a serialized Java object. Read, + * never written: the name is what a file already on disk carries, so it cannot be changed, and + * it stands apart from the names below, which are of a format of their own. + */ + public static final String LEGACY_COL_STATS_BLOB = ColumnStatisticsObj.class.getSimpleName(); + /** + * What a statistics file is called: the granularity it holds, the snapshot it describes, and + * something to tell two writes of one snapshot apart. Placed beside the table's metadata, where + * Iceberg puts the statistics it writes itself - it names those {snapshot}-{uuid}.stats and the + * partition statistics it computes partition-stats-{snapshot}-{uuid}. Ours say col, since what + * they hold is the statistics of a column rather than the counts of a partition, and end in the + * container rather than saying stats a second time. FileFormat.PUFFIN knows that ending. + */ + /** + * What the file holds, then the snapshot it describes. Named for the kind of statistics rather + * than the grain of them, since a partition-level file also carries the table-level entries + * folded from its partitions - and named apart from the statistics Iceberg keeps of its own, + * which sit in the same directory. + */ + private static final String STATS_FILE = "column-stats-%d-%s.puffin"; + + private IcebergColStatsWriter() { + } + + /** Everything written describes the snapshot it is written for, so a read asks only what happened after it. */ + public static boolean writeColStats(Table tbl, Snapshot snapshot, Iterator<ColumnStatistics> colStats, + Configuration conf) { + ColumnStatistics head = colStats.next(); + WritePolicy policy = WritePolicy.resolve(tbl, snapshot, head, conf); + if (policy == WritePolicy.SKIP) { + // storing nothing says the stored statistics still stand, and a caller that hears otherwise + // takes the mark of accuracy off the table - which leaves a good file unread + boolean accurate = IcebergTableUtil.colStatsAccurate(tbl, snapshot, conf); + LOG.info("Storing no column statistics for {} at snapshot {}: what was gathered describes" + + " {}, and what is stored {}", tbl.name(), snapshot.snapshotId(), + head.getStatsDesc().isIsTblLevel() ? "the whole table" : "a partition", + accurate ? "still stands" : "does not"); + return accurate; + } + LOG.info("Storing column statistics of {} at snapshot {}: {} what was gathered, which describes {}", + tbl.name(), snapshot.snapshotId(), policy, + head.getStatsDesc().isIsTblLevel() ? "the whole table" : "a partition"); + Iterator<ColumnStatistics> all = Iterators.concat(Iterators.singletonIterator(head), colStats); + try { + return head.getStatsDesc().isIsTblLevel() ? + writeTableColStats(tbl, snapshot, all, policy, conf) : + writePartitionColStats(tbl, snapshot, all, policy, conf); + } catch (IOException | InvalidObjectException e) { + // serving no stats degrades the planner to estimates - never wrong + LOG.warn("Unable to write column stats", e); + return false; + } + } + + private static boolean writeTableColStats(Table tbl, Snapshot snapshot, Iterator<ColumnStatistics> colStats, + WritePolicy policy, Configuration conf) throws IOException, InvalidObjectException { + // the table's statistics are one entry holding every column: nothing to stream + ColumnStatistics stats = colStats.next(); + if (policy == WritePolicy.MERGE) { + // A write commits a snapshot of its own, so what it completes sits on the one before it. An + // ANALYZE commits none, but replaces rather than merges, so it never asks. + Long parentId = snapshot.parentId(); + StatisticsFile statsOldSrc = parentId == null ? null : + IcebergTableUtil.getColStatsFile(tbl, parentId, false); + if (statsOldSrc == null) { + // a table-level increment has nothing to add itself to + return false; + } + List<ColumnStatisticsObj> statsOld = IcebergColStatsReader.readColStatsOrThrow( + tbl, statsOldSrc, null, true); + // drop columns the stored file does not describe: their stats cover only the inserted + // rows, and with nothing to merge into they would stand as stats for the whole table + Set<String> stored = statsOld.stream().map(ColumnStatisticsObj::getColName) + .collect(Collectors.toSet()); + stats.getStatsObj().removeIf(obj -> !stored.contains(obj.getColName())); + if (stats.getStatsObj().isEmpty()) { + return false; + } + MetaStoreServerUtils.mergeColStats(stats, new ColumnStatistics(null, statsOld)); + } + Schema schema = tbl.spec().schema(); + // a column dropped or renamed since the entry was stored resolves no field: its statistics + // leave with it + stats.getStatsObj().removeIf(obj -> schema.caseInsensitiveFindField(obj.getColName()) == null); + return commitColStatsFile(tbl, snapshot, conf, writer -> { + for (ColumnStatisticsObj obj : stats.getStatsObj()) { + // a column's statistics are one blob, sketches and all, as a partition's entries are one + // entry: what a read wants of them it settles once they are in hand. The vector stays + // whatever a read asks for, because a write needs it - an increment merges its own + // gather into what is stored, and only a vector lets the distinct counts be merged + writer.add(new Blob( + HIVE_COL_STATS_BLOB_V1, + List.of(schema.caseInsensitiveFindField(obj.getColName()).fieldId()), + snapshot.snapshotId(), snapshot.sequenceNumber(), + ByteBuffer.wrap(IcebergColStatsCodec.encodeEntry(obj)), + PuffinCompressionCodec.NONE, + // the count travels in the metadata, where a read takes it without opening the file + IcebergColStatsProperties.of(obj))); + } + }); + } + + private static boolean writePartitionColStats(Table tbl, Snapshot snapshot, Iterator<ColumnStatistics> colStats, + WritePolicy policy, Configuration conf) throws IOException, InvalidObjectException { + Schema schema = tbl.spec().schema(); + Set<String> written = Sets.newHashSet(); + // an ANALYZE commits no snapshot of its own: it writes to the snapshot it read, where the + // statistics already are, so the walk starts there rather than at the parent + StatisticsFile statsOldSrc = policy == WritePolicy.MERGE ? + IcebergTableUtil.findColStatsFile(tbl, snapshot.snapshotId(), true) : null; + Rollup rollup = new Rollup(); + return commitColStatsFile(tbl, snapshot, conf, writer -> { + boolean first = true; + while (colStats.hasNext()) { + ColumnStatistics stats = colStats.next(); + String partName = stats.getStatsDesc().getPartName(); + if (partName == null) { + // a group naming no partition describes none + continue; + } + // a column dropped or renamed since the entry was stored resolves no field: its + // statistics leave with it + stats.getStatsObj().removeIf(obj -> schema.caseInsensitiveFindField(obj.getColName()) == null); + List<Integer> fieldIds = stats.getStatsObj().stream() + .map(obj -> schema.caseInsensitiveFindField(obj.getColName()).fieldId()).toList(); + // only the first blob names them in the footer, which would otherwise repeat them once + // per partition; the blob itself names the field every entry is for. What it names is + // every column the file comes to hold, so that whether a column has statistics here can + // be answered from the table's metadata without opening anything + List<Integer> named = first ? mergedFieldIds(fieldIds, statsOldSrc, schema) : List.of(-1); + first = false; + writer.add(new Blob( + HIVE_PART_COL_STATS_BLOB_V1, named, + snapshot.snapshotId(), snapshot.sequenceNumber(), + encodePartitionBlob(stats.getStatsObj(), fieldIds), + PuffinCompressionCodec.NONE, + Map.of(IcebergTableUtil.PARTITION_FIELD, partName))); + written.add(partName); + rollup.add(stats.getStatsObj()); + } + if (policy == WritePolicy.MERGE) { + carryPartitionColStats(tbl, snapshot, writer, written, conf, statsOldSrc, rollup); + } + // the table's own entries, folded from every partition the file comes to hold + rollup.write(writer, snapshot, schema); + }); + } + + /** + * Carries forward, bytes for bytes, the stored entries of the partitions this write never + * measured, as long as no write since the stored file changed them. Carrying is the one place + * that can settle that without a reader paying for the walk, so the walk here is uncapped. + */ + private static void carryPartitionColStats(Table tbl, Snapshot snapshot, PuffinWriter writer, + Set<String> written, Configuration conf, StatisticsFile statsOldSrc, Rollup rollup) + throws IOException { + if (statsOldSrc == null) { + // a partition describes itself: with nothing stored there is nothing to carry, and what + // was computed stands on its own + return; + } + Predicate<String> stillHolds = IcebergTableUtil.upToDateColStats(tbl, snapshot, statsOldSrc, conf, false); + try (PuffinReader reader = Puffin.read(tbl.io().newInputFile(statsOldSrc.path())) + .withFileSize(statsOldSrc.fileSizeInBytes()) + .withFooterSize(statsOldSrc.fileFooterSizeInBytes()) + .build()) { + List<BlobMetadata> carried = reader.fileMetadata().blobs().stream() + .filter(metadata -> { + String partName = metadata.properties().get(IcebergTableUtil.PARTITION_FIELD); + return HIVE_PART_COL_STATS_BLOB_V1.equals(metadata.type()) && + partName != null && !written.contains(partName) && stillHolds.test(partName); + }) + .toList(); + for (Pair<BlobMetadata, ByteBuffer> blob : reader.readAll(carried)) { + byte[] carriedBytes = ByteBuffers.toByteArray(blob.second()); + writer.add(new Blob( + HIVE_PART_COL_STATS_BLOB_V1, List.of(-1), + snapshot.snapshotId(), snapshot.sequenceNumber(), + ByteBuffer.wrap(carriedBytes), + PuffinCompressionCodec.NONE, + Map.of(IcebergTableUtil.PARTITION_FIELD, + blob.first().properties().get(IcebergTableUtil.PARTITION_FIELD)))); + // the bytes travel untouched, but the table's entries are folded from every partition, + // and a carried one holds values no later gather will see again + try { + rollup.add(IcebergColStatsReader.decodePartitionBlob( + ByteBuffer.wrap(carriedBytes), null, true)); + } catch (InvalidObjectException e) { + throw new IOException(e); + } + } + } + } + + /** + * The entries of the whole table, folded from the partitions as they are written. A scan that + * asks about the table rather than about partitions of it is answered from these, so it reads + * one blob per column asked instead of one per partition it would otherwise have to merge. + * + * A partition contributes once, whether this gather measured it or carried it: nothing can be + * taken back out of a distinct count, so what is not folded here cannot be folded later. + */ + private static final class Rollup { + + private final Map<String, ColumnStatisticsObj> byColumn = Maps.newLinkedHashMap(); + /** How many partitions stated each column, so that a column short of any is not stated. */ + private final Map<String, Integer> statedBy = Maps.newHashMap(); + private int partitions; + + private void add(List<ColumnStatisticsObj> statsObjs) throws InvalidObjectException { + partitions++; + for (ColumnStatisticsObj statsObj : statsObjs) { + statedBy.merge(statsObj.getColName(), 1, Integer::sum); + ColumnStatisticsObj held = byColumn.get(statsObj.getColName()); + if (held == null) { + byColumn.put(statsObj.getColName(), statsObj.deepCopy()); + } else { + ColumnStatistics into = new ColumnStatistics(null, Lists.newArrayList(held)); + MetaStoreServerUtils.mergeColStats(into, new ColumnStatistics(null, List.of(statsObj))); + byColumn.put(statsObj.getColName(), into.getStatsObj().getFirst()); + } + } + } + + private void write(PuffinWriter writer, Snapshot snapshot, Schema schema) throws IOException { + for (ColumnStatisticsObj statsObj : byColumn.values()) { + Types.NestedField field = schema.caseInsensitiveFindField(statsObj.getColName()); + // a column any partition did not state is not the table's: a rename leaves the partitions + // this gather did not write naming the column it was, and what they hold is not this + if (field == null || statedBy.getOrDefault(statsObj.getColName(), 0) != partitions) { + continue; + } + writer.add(new Blob( + HIVE_COL_STATS_BLOB_V1, List.of(field.fieldId()), + snapshot.snapshotId(), snapshot.sequenceNumber(), + ByteBuffer.wrap(IcebergColStatsCodec.encodeEntry(statsObj)), + PuffinCompressionCodec.NONE, + IcebergColStatsProperties.of(statsObj))); + } + } + } + + @FunctionalInterface + private interface ColStatsBlobWriter { + void write(PuffinWriter writer) throws IOException, InvalidObjectException; + } + + /** Writes one statistics file through the given blobs and commits it for the snapshot. */ Review Comment: javadoc fix might be required. -- 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]
