Aggarwal-Raghav commented on code in PR #6707: URL: https://github.com/apache/hive/pull/6707#discussion_r3902775830
########## iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/stats/IcebergColStatsCodec.java: ########## @@ -0,0 +1,236 @@ +/* + * 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.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.function.IntPredicate; +import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData; +import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.util.ByteBuffers; +import org.apache.thrift.TDeserializer; +import org.apache.thrift.TException; +import org.apache.thrift.TSerializer; +import org.apache.thrift.protocol.TCompactProtocol; + +/** + * How one column's statistics are written into a slice of a blob, and read back. + * + * <p>An entry is a Thrift struct, so it is written as one: the compact protocol states a field by + * its id, which makes a slice a fraction of the size and a read a fraction of the work of the Java + * serialization the first version of the blob used - that repeated the whole class descriptor in + * every slice of every partition. Reading it back stays: a file written before this is still read + * the way it was written. + * + * <p>Thrift is also what makes the blob safe to keep: a field the writer did not know is skipped + * and one it did not write defaults, so an entry written by another version reads rather than + * throwing, which is not true of a serialized Java class. + */ +final class IcebergColStatsCodec { + + private static final ThreadLocal<TSerializer> WRITERS = ThreadLocal.withInitial(() -> { + try { + return new TSerializer(new TCompactProtocol.Factory()); + } catch (TException e) { + throw new IllegalStateException("Cannot write column statistics", e); + } + }); + + private static final ThreadLocal<TDeserializer> READERS = ThreadLocal.withInitial(() -> { + try { + return new TDeserializer(new TCompactProtocol.Factory()); + } catch (TException e) { + throw new IllegalStateException("Cannot read column statistics", e); + } + }); + + /** + * What a sketch slice is named after the column it belongs to. No column name holds this, so a + * blob that carries no sketch simply has no slice named this way. + */ + + private IcebergColStatsCodec() { + } + + /** + * Takes the sketch out of the entry, so that what remains is the part every read needs. Returns + * what was taken, or null when the entry carried none. + */ + /** What a blob holding more than one entry is written as, so a read knows what it is reading. */ + static final int BLOB_VERSION = 1; + + /** + * Entries one after another, each behind the field it is for and its own length. The field is + * what lets a read take the columns it asked about and step over the rest, and it is written per + * entry rather than once for the blob because a merge carries partitions gathered separately - + * they need not hold the same columns, nor hold them in the same order. + */ + static byte[] encodeBlob(List<byte[]> parts, List<Integer> fieldIds) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + DataOutputStream data = new DataOutputStream(out); + data.writeInt(BLOB_VERSION); + data.writeInt(parts.size()); + for (int i = 0; i < parts.size(); i++) { + data.writeInt(fieldIds.get(i)); + data.writeInt(parts.get(i).length); + data.write(parts.get(i)); + } + data.flush(); + return out.toByteArray(); + } + + /** + * The entries the given places name, the rest skipped rather than copied. A scan of a wide table + * asks about a few of its columns, and an entry it does not want costs a read nothing beyond the + * length it steps over. + */ + static List<byte[]> decodeBlob(ByteBuffer buf, IntPredicate wanted) throws IOException { + DataInputStream data = new DataInputStream(new ByteArrayInputStream(ByteBuffers.toByteArray(buf))); Review Comment: ``` DataInputStream data = new DataInputStream(new ByteArrayInputStream(ByteBuffers.toByteArray(buf))); ``` is necessary? the ops like readInt etc can be done directly on ByteBuffer as well. -- 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]
