difin commented on code in PR #6707: URL: https://github.com/apache/hive/pull/6707#discussion_r3917322752
########## iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/stats/IcebergColStatsReader.java: ########## @@ -0,0 +1,369 @@ +/* + * 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.Collection; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.function.IntPredicate; +import java.util.function.Predicate; +import org.apache.commons.lang3.SerializationUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.iceberg.StatisticsFile; +import org.apache.iceberg.Table; +import org.apache.iceberg.io.DelegatingInputStream; +import org.apache.iceberg.io.IOUtil; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.SeekableInputStream; +import org.apache.iceberg.puffin.BlobMetadata; +import org.apache.iceberg.puffin.Puffin; +import org.apache.iceberg.puffin.PuffinReader; +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; + +/** + * Reads the column statistics {@link IcebergColStatsWriter} stores: table-level entries one blob + * per column, partition entries one framed blob per partition, and the aggregate a partition-level + * file also holds - fetching and decoding only the columns asked. + */ +public final class IcebergColStatsReader { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergColStatsReader.class); + + /** + * What a read is worth in bytes where the stream underneath is not one that states its own. + * Hadoop's streams state 4KiB and S3A takes that too; this stands higher because a partition + * entry runs to a few hundred bytes, which makes the gap between two a scan wants worth reading + * through rather than seeking over. + */ + private static final long DEFAULT_MIN_SEEK = 16 * 1024; + /** How much one read may hold where the stream does not state its own. Hadoop's own default. */ + private static final long DEFAULT_MAX_READ_SIZE = 1024 * 1024; + + private IcebergColStatsReader() { + } + + /** The stored statistics describing the whole table, with what the configuration asks for. */ + public static <T> List<T> read(Table table, long snapshotId, Collection<String> columns, + Configuration conf) { + return read(table, snapshotId, columns, fetchVectors(conf)); + } + + /** The same, told outright whether the vector is wanted. */ + public static <T> List<T> read(Table table, long snapshotId, Collection<String> columns, + boolean withSketch) { + StatisticsFile statsFile = IcebergStoredStats.findColStatsFile(table, snapshotId, false); + if (statsFile == null) { + LOG.warn("Column stats file not found for snapshot: {}", snapshotId); + return Lists.newArrayList(); + } + return read(table, statsFile, columns, withSketch); + } + + /** The same, out of a file the caller has already settled on. */ + public static <T> List<T> read(Table table, StatisticsFile statsFile, + Collection<String> columns, boolean withSketch) { + try { + return readOrThrow(table, statsFile, columns, withSketch); + } catch (Exception e) { + // serving no stats degrades the planner to estimates - never wrong + LOG.warn("Unable to read column stats: {}", e.getMessage()); + return Lists.newArrayList(); + } + } + + /** + * The strict variant for the merge path: an unreadable statistics file must not be mistaken for + * an absent one, or the increment would be persisted as the complete statistics. + */ + @SuppressWarnings("unchecked") + static <T> List<T> readOrThrow(Table table, StatisticsFile statsFile, + Collection<String> columns, boolean withSketch) + throws IOException { + Predicate<BlobMetadata> filter = columns != null ? blobsForColumns(table, columns) : null; + Map<Integer, ColumnStatisticsObj> entries = Maps.newLinkedHashMap(); + String statsPath = statsFile.path(); + try (PuffinReader reader = Puffin.read(table.io().newInputFile(statsPath)) + .withFileSize(statsFile.fileSizeInBytes()) + .withFooterSize(statsFile.fileFooterSizeInBytes()) + .build()) { + List<BlobMetadata> blobMetadata = reader.fileMetadata().blobs().stream() + .filter(IcebergColStatsReader::holdsColStats) + .filter(blob -> filter == null || filter.test(blob)) + .toList(); + LOG.info("Using column stats from: {}", statsPath); + + for (Pair<BlobMetadata, ByteBuffer> blob : reader.readAll(blobMetadata)) { + byte[] raw = ByteBuffers.toByteArray(blob.second()); + int fieldId = blob.first().inputFields().getFirst(); + entries.put(fieldId, decodeTableEntry(raw, blob.first().type(), withSketch)); + } + } + // a caller trims what it is given, so it is given a list of its own + return (List<T>) Lists.newArrayList(entries.values()); + } + + /** + * Whether the blob holds an entry this reads. A file may hold blobs of other kinds beside these + * - a sketch another engine wrote and keeps across its own writes - and one of those is nothing + * to decode: reading it as an entry would lose the whole file rather than the one blob. + */ + private static boolean holdsColStats(BlobMetadata blob) { + return IcebergColStatsWriter.HIVE_COL_STATS_BLOB_V1.equals(blob.type()) || + IcebergColStatsWriter.LEGACY_COL_STATS_BLOB.equals(blob.type()); + } + + /** An entry as the blob that names it was written: a Thrift struct, or a serialized Java object. */ + private static ColumnStatisticsObj decodeTableEntry(byte[] raw, String blobType, boolean withVectors) { + if (IcebergColStatsWriter.HIVE_COL_STATS_BLOB_V1.equals(blobType)) { + return IcebergColStatsCodec.decodeEntry(raw, withVectors); + } + // a table-level entry released before this holds a serialized Java object, and still reads - + // with the vector it was written with, which is what a merge of its distinct counts needs + ColumnStatisticsObj released = SerializationUtils.deserialize(raw); + return withVectors ? released : IcebergColStatsCodec.withoutVectors(released); + } + + /** + * Whether a read wants the vector a distinct count is merged from across partitions. It is the + * bulk of an entry, so a read that will not merge counts leaves it in the file rather than + * fetching and decoding it. A write never asks: the vector is computed for every column whatever + * anyone asked for, and a statistic stored without one can never be merged afterwards. + * A histogram answers to nothing here: one exists only where a statement was told to compute it, + * and a plan reads one wherever it finds it. + * + * <p>A read told not to fetch them is left to bound a distinct count from the entries themselves, + * as one of a native table is. + */ + private static boolean fetchVectors(Configuration conf) { + return MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.STATS_FETCH_BITVECTOR); + } + + /** The blobs naming any of the asked columns, by the name the table's schema gives the field now. */ + private static Predicate<BlobMetadata> blobsForColumns(Table table, Collection<String> columns) { + return metadata -> metadata.inputFields().stream() + .map(fieldId -> table.schema().findColumnName(fieldId)) + .anyMatch(columns::contains); + } + + /** + * What the file states of the table itself, where the ask covers every partition it describes + * and each still does. Null where it does not, or where the file holds no such entries - one + * written before they were kept, or by a gather that measured a partition subset. + */ + public static List<ColumnStatisticsObj> readAggr(Table table, StatisticsFile statsFile, + Set<String> asked, Predicate<String> upToDate, List<String> colNames, Configuration conf) { + return readAggr(table, statsFile, asked, upToDate, colNames, fetchVectors(conf)); + } + + private static List<ColumnStatisticsObj> readAggr(Table table, StatisticsFile statsFile, + Set<String> asked, Predicate<String> upToDate, List<String> colNames, boolean withSketch) { + Set<String> described = Sets.newHashSet(); + for (org.apache.iceberg.BlobMetadata blob : statsFile.blobMetadata()) { + String partName = blob.properties().get(IcebergColStatsWriter.PARTITION_FIELD); + if (partName != null) { + described.add(partName); + } + } + // nothing can be taken out of an aggregate or added to it: it answers only where the ask is + // exactly the partitions it holds, each still describing itself + if (described.isEmpty() || !described.equals(asked) || !described.stream().allMatch(upToDate)) { + return null; + } + Set<String> columns = Sets.newHashSet(colNames); + List<ColumnStatisticsObj> aggregated = + read(table, statsFile, columns, withSketch); + // a rename keeps the field, so a blob written before it still names the field under the old + // column name: it answers for the field asked about but not for the column, and is left out + aggregated.removeIf(statsObj -> !columns.contains(statsObj.getColName())); + return aggregated.size() == colNames.size() ? aggregated : null; + } + + /** + * The stored partition entries the given file holds for the partitions the filter admits, each + * trimmed to the asked columns; a null column set asks for all of them. + */ + public static Map<String, List<ColumnStatisticsObj>> readPart(Table table, StatisticsFile statsFile, + Predicate<String> partitionFilter, Set<String> columns, Configuration conf) { + return readPart(table, statsFile, partitionFilter, columns, fetchVectors(conf)); + } + + /** The same, told outright whether the vector is wanted. */ + public static Map<String, List<ColumnStatisticsObj>> readPart(Table table, StatisticsFile statsFile, + Predicate<String> partitionFilter, Set<String> columns, boolean withSketch) { + Map<String, List<ColumnStatisticsObj>> result = Maps.newLinkedHashMap(); + try (PuffinReader reader = Puffin.read(table.io().newInputFile(statsFile.path())) + .withFileSize(statsFile.fileSizeInBytes()) + .withFooterSize(statsFile.fileFooterSizeInBytes()) + .build()) { + List<BlobMetadata> blobs = reader.fileMetadata().blobs().stream() + .filter(metadata -> IcebergColStatsWriter.HIVE_PART_COL_STATS_BLOB_V1.equals(metadata.type()) && + metadata.properties().containsKey(IcebergColStatsWriter.PARTITION_FIELD)) + .filter(metadata -> { + String partName = metadata.properties().get(IcebergColStatsWriter.PARTITION_FIELD); + return partName != null && (partitionFilter == null || partitionFilter.test(partName)); + }) + .toList(); + LOG.info("Using column stats from: {}", statsFile.path()); + // one stream serves them all, seeking forward: the blobs of the partitions a scan reads sit + // near one another in the file, and one read spanning the gap between two of them costs less + // than the seek it saves. Reading each on its own is what makes a scan of many partitions + // expensive. + if (!blobs.isEmpty()) { + InputFile file = table.io().newInputFile(statsFile.path(), statsFile.fileSizeInBytes()); + try (SeekableInputStream in = file.newStream()) { + readRanges(in, blobs, columns, withSketch, result, fieldsOf(table, columns)); + } + } + } catch (Exception e) { + // serving no stats degrades the planner to estimates - never wrong + LOG.warn("Unable to read column stats: {}", e.getMessage()); + result.clear(); + } + return result; + } + + /** + * The blobs a scan reads, taken in as few reads as their places in the file allow. They are + * written one after another, so the ones a scan asks for are read in runs rather than one at a + * time; a run ends at a gap wider than the seek it would save, or at the bytes one read may hold. + */ + static void readRanges(SeekableInputStream in, List<BlobMetadata> blobs, Set<String> columns, + boolean withSketch, Map<String, List<ColumnStatisticsObj>> result, IntPredicate fields) + throws IOException { + List<BlobMetadata> ordered = blobs.stream() + .sorted(Comparator.comparingLong(BlobMetadata::offset)) + .toList(); + long minSeek = minSeek(in); + long maxReadSize = maxReadSize(in); + int cursor = 0; + while (cursor < ordered.size()) { + int first = cursor; + int last = cursor; + for (int next = cursor + 1; next < ordered.size(); next++) { + BlobMetadata held = ordered.get(last); + long gap = ordered.get(next).offset() - (held.offset() + held.length()); + long span = ordered.get(next).offset() + ordered.get(next).length() - ordered.get(first).offset(); + if (gap > minSeek || span > maxReadSize) { + break; + } + last = next; + } + long start = ordered.get(first).offset(); + int length = (int) (ordered.get(last).offset() + ordered.get(last).length() - start); Review Comment: `length` is computed as long, but stored as int. Will it be sufficient? -- 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]
