| // 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.impala.catalog; |
| |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Comparator; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.atomic.AtomicLong; |
| |
| import javax.annotation.Nonnull; |
| |
| import org.apache.hadoop.fs.BlockLocation; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.impala.analysis.Expr; |
| import org.apache.impala.analysis.LiteralExpr; |
| import org.apache.impala.analysis.PartitionKeyValue; |
| import org.apache.impala.catalog.events.InFlightEvents; |
| import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventPropertyKey; |
| import org.apache.impala.common.FileSystemUtil; |
| import org.apache.impala.common.ImpalaException; |
| import org.apache.impala.common.Pair; |
| import org.apache.impala.common.Reference; |
| import org.apache.impala.compat.MetastoreShim; |
| import org.apache.impala.fb.FbCompression; |
| import org.apache.impala.fb.FbFileBlock; |
| import org.apache.impala.fb.FbFileDesc; |
| import org.apache.impala.thrift.CatalogObjectsConstants; |
| import org.apache.impala.thrift.TAccessLevel; |
| import org.apache.impala.thrift.TExpr; |
| import org.apache.impala.thrift.TExprNode; |
| import org.apache.impala.thrift.THdfsFileDesc; |
| import org.apache.impala.thrift.THdfsPartition; |
| import org.apache.impala.thrift.THdfsPartitionLocation; |
| import org.apache.impala.thrift.TNetworkAddress; |
| import org.apache.impala.thrift.TPartitionStats; |
| import org.apache.impala.util.HdfsCachingUtil; |
| import org.apache.impala.util.ListMap; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Function; |
| import com.google.common.base.Joiner; |
| import com.google.common.base.Objects; |
| import com.google.common.base.Preconditions; |
| import com.google.common.base.Predicate; |
| import com.google.common.base.Strings; |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.collect.ImmutableMap; |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| import com.google.flatbuffers.FlatBufferBuilder; |
| |
| /** |
| * Query-relevant information for one table partition. Partitions are comparable |
| * based on their partition-key values. The comparison orders partitions in ascending |
| * order with NULLs sorting last. The ordering is useful for displaying partitions |
| * in SHOW statements. |
| */ |
| public class HdfsPartition implements FeFsPartition, PrunablePartition { |
| /** |
| * Metadata for a single file in this partition. |
| */ |
| static public class FileDescriptor implements Comparable<FileDescriptor> { |
| // An invalid network address, which will always be treated as remote. |
| private final static TNetworkAddress REMOTE_NETWORK_ADDRESS = |
| new TNetworkAddress("remote*addr", 0); |
| |
| // Minimum block size in bytes allowed for synthetic file blocks (other than the last |
| // block, which may be shorter). |
| public final static long MIN_SYNTHETIC_BLOCK_SIZE = 1024 * 1024; |
| |
| // Internal representation of a file descriptor using a FlatBuffer. |
| private final FbFileDesc fbFileDescriptor_; |
| |
| private FileDescriptor(FbFileDesc fileDescData) { fbFileDescriptor_ = fileDescData; } |
| |
| public static FileDescriptor fromThrift(THdfsFileDesc desc) { |
| ByteBuffer bb = ByteBuffer.wrap(desc.getFile_desc_data()); |
| return new FileDescriptor(FbFileDesc.getRootAsFbFileDesc(bb)); |
| } |
| |
| /** |
| * Clone the descriptor, but change the replica indexes to reference the new host |
| * index 'dstIndex' instead of the original index 'origIndex'. |
| */ |
| public FileDescriptor cloneWithNewHostIndex(List<TNetworkAddress> origIndex, |
| ListMap<TNetworkAddress> dstIndex) { |
| // First clone the flatbuffer with no changes. |
| ByteBuffer oldBuf = fbFileDescriptor_.getByteBuffer(); |
| ByteBuffer newBuf = ByteBuffer.allocate(oldBuf.remaining()); |
| newBuf.put(oldBuf.array(), oldBuf.position(), oldBuf.remaining()); |
| newBuf.rewind(); |
| FbFileDesc cloned = FbFileDesc.getRootAsFbFileDesc(newBuf); |
| |
| // Now iterate over the blocks in the new flatbuffer and mutate the indexes. |
| FbFileBlock it = new FbFileBlock(); |
| for (int i = 0; i < cloned.fileBlocksLength(); i++) { |
| it = cloned.fileBlocks(it, i); |
| for (int j = 0; j < it.replicaHostIdxsLength(); j++) { |
| int origHostIdx = FileBlock.getReplicaHostIdx(it, j); |
| boolean isCached = FileBlock.isReplicaCached(it, j); |
| TNetworkAddress origHost = origIndex.get(origHostIdx); |
| int newHostIdx = dstIndex.getIndex(origHost); |
| it.mutateReplicaHostIdxs(j, FileBlock.makeReplicaIdx(isCached, newHostIdx)); |
| } |
| } |
| return new FileDescriptor(cloned); |
| } |
| |
| /** |
| * Creates the file descriptor of a file represented by 'fileStatus' with blocks |
| * stored in 'blockLocations'. 'fileSystem' is the filesystem where the |
| * file resides and 'hostIndex' stores the network addresses of the hosts that store |
| * blocks of the parent HdfsTable. 'isEc' indicates whether the file is erasure-coded. |
| * Populates 'numUnknownDiskIds' with the number of unknown disk ids. |
| * |
| * |
| */ |
| /** |
| * Creates a FileDescriptor with block locations. |
| * |
| * @param fileStatus the status returned from file listing |
| * @param relPath the path of the file relative to the partition directory |
| * @param blockLocations the block locations for the file |
| * @param hostIndex the host index to use for encoding the hosts |
| * @param isEc true if the file is known to be erasure-coded |
| * @param numUnknownDiskIds reference which will be set to the number of blocks |
| * for which no disk ID could be determined |
| */ |
| public static FileDescriptor create(FileStatus fileStatus, String relPath, |
| BlockLocation[] blockLocations, ListMap<TNetworkAddress> hostIndex, boolean isEc, |
| Reference<Long> numUnknownDiskIds) throws IOException { |
| FlatBufferBuilder fbb = new FlatBufferBuilder(1); |
| int[] fbFileBlockOffsets = new int[blockLocations.length]; |
| int blockIdx = 0; |
| for (BlockLocation loc: blockLocations) { |
| if (isEc) { |
| fbFileBlockOffsets[blockIdx++] = FileBlock.createFbFileBlock(fbb, |
| loc.getOffset(), loc.getLength(), |
| (short) hostIndex.getIndex(REMOTE_NETWORK_ADDRESS)); |
| } else { |
| fbFileBlockOffsets[blockIdx++] = |
| FileBlock.createFbFileBlock(fbb, loc, hostIndex, numUnknownDiskIds); |
| } |
| } |
| return new FileDescriptor(createFbFileDesc(fbb, fileStatus, relPath, |
| fbFileBlockOffsets, isEc)); |
| } |
| |
| /** |
| * Creates the file descriptor of a file represented by 'fileStatus' that |
| * resides in a filesystem that doesn't support the BlockLocation API (e.g. S3). |
| */ |
| public static FileDescriptor createWithNoBlocks(FileStatus fileStatus, |
| String relPath) { |
| FlatBufferBuilder fbb = new FlatBufferBuilder(1); |
| return new FileDescriptor(createFbFileDesc(fbb, fileStatus, relPath, null, false)); |
| } |
| |
| /** |
| * Serializes the metadata of a file descriptor represented by 'fileStatus' into a |
| * FlatBuffer using 'fbb' and returns the associated FbFileDesc object. |
| * 'fbFileBlockOffsets' are the offsets of the serialized block metadata of this file |
| * in the underlying buffer. Can be null if there are no blocks. |
| */ |
| private static FbFileDesc createFbFileDesc(FlatBufferBuilder fbb, |
| FileStatus fileStatus, String relPath, int[] fbFileBlockOffets, boolean isEc) { |
| int relPathOffset = fbb.createString(relPath); |
| // A negative block vector offset is used when no block offsets are specified. |
| int blockVectorOffset = -1; |
| if (fbFileBlockOffets != null) { |
| blockVectorOffset = FbFileDesc.createFileBlocksVector(fbb, fbFileBlockOffets); |
| } |
| FbFileDesc.startFbFileDesc(fbb); |
| // TODO(todd) rename to RelativePathin the FBS |
| FbFileDesc.addRelativePath(fbb, relPathOffset); |
| FbFileDesc.addLength(fbb, fileStatus.getLen()); |
| FbFileDesc.addLastModificationTime(fbb, fileStatus.getModificationTime()); |
| FbFileDesc.addIsEc(fbb, isEc); |
| HdfsCompression comp = HdfsCompression.fromFileName(fileStatus.getPath().getName()); |
| FbFileDesc.addCompression(fbb, comp.toFb()); |
| if (blockVectorOffset >= 0) FbFileDesc.addFileBlocks(fbb, blockVectorOffset); |
| fbb.finish(FbFileDesc.endFbFileDesc(fbb)); |
| // To eliminate memory fragmentation, copy the contents of the FlatBuffer to the |
| // smallest possible ByteBuffer. |
| ByteBuffer bb = fbb.dataBuffer().slice(); |
| ByteBuffer compressedBb = ByteBuffer.allocate(bb.capacity()); |
| compressedBb.put(bb); |
| return FbFileDesc.getRootAsFbFileDesc((ByteBuffer)compressedBb.flip()); |
| } |
| |
| public String getRelativePath() { return fbFileDescriptor_.relativePath(); } |
| public long getFileLength() { return fbFileDescriptor_.length(); } |
| |
| /** Compute the total length of files in fileDescs */ |
| public static long computeTotalFileLength(Collection<FileDescriptor> fileDescs) { |
| long totalLength = 0; |
| for (FileDescriptor fileDesc: fileDescs) { |
| totalLength += fileDesc.getFileLength(); |
| } |
| return totalLength; |
| } |
| |
| public HdfsCompression getFileCompression() { |
| return HdfsCompression.valueOf(FbCompression.name(fbFileDescriptor_.compression())); |
| } |
| |
| public long getModificationTime() { return fbFileDescriptor_.lastModificationTime(); } |
| public int getNumFileBlocks() { return fbFileDescriptor_.fileBlocksLength(); } |
| public boolean getIsEc() {return fbFileDescriptor_.isEc(); } |
| |
| public FbFileBlock getFbFileBlock(int idx) { |
| return fbFileDescriptor_.fileBlocks(idx); |
| } |
| |
| public THdfsFileDesc toThrift() { |
| THdfsFileDesc fd = new THdfsFileDesc(); |
| ByteBuffer bb = fbFileDescriptor_.getByteBuffer(); |
| fd.setFile_desc_data(bb); |
| return fd; |
| } |
| |
| @Override |
| public String toString() { |
| int numFileBlocks = getNumFileBlocks(); |
| List<String> blocks = Lists.newArrayListWithCapacity(numFileBlocks); |
| for (int i = 0; i < numFileBlocks; ++i) { |
| blocks.add(FileBlock.debugString(getFbFileBlock(i))); |
| } |
| return Objects.toStringHelper(this) |
| .add("RelativePath", getRelativePath()) |
| .add("Length", getFileLength()) |
| .add("Compression", getFileCompression()) |
| .add("ModificationTime", getModificationTime()) |
| .add("Blocks", Joiner.on(", ").join(blocks)).toString(); |
| } |
| |
| @Override |
| public int compareTo(FileDescriptor otherFd) { |
| return getRelativePath().compareTo(otherFd.getRelativePath()); |
| } |
| |
| /** |
| * Function to convert from a byte[] flatbuffer to the wrapper class. Note that |
| * this returns a shallow copy which continues to reflect any changes to the |
| * passed byte[]. |
| */ |
| public static final Function<byte[], FileDescriptor> FROM_BYTES = |
| new Function<byte[], FileDescriptor>() { |
| @Override |
| public FileDescriptor apply(byte[] input) { |
| ByteBuffer bb = ByteBuffer.wrap(input); |
| return new FileDescriptor(FbFileDesc.getRootAsFbFileDesc(bb)); |
| } |
| }; |
| |
| /** |
| * Function to convert from the wrapper class to a raw byte[]. Note that |
| * this returns a shallow copy and callers should not modify the returned array. |
| */ |
| public static final Function<FileDescriptor, byte[]> TO_BYTES = |
| new Function<FileDescriptor, byte[]>() { |
| @Override |
| public byte[] apply(FileDescriptor fd) { |
| ByteBuffer bb = fd.fbFileDescriptor_.getByteBuffer(); |
| byte[] arr = bb.array(); |
| assert bb.arrayOffset() == 0 && bb.remaining() == arr.length; |
| return arr; |
| } |
| }; |
| } |
| |
| /** |
| * Represents metadata of a single block replica. |
| */ |
| public static class BlockReplica { |
| private final boolean isCached_; |
| private final short hostIdx_; |
| |
| /** |
| * Creates a BlockReplica given a host ID/index and a flag specifying whether this |
| * replica is cahced. Host IDs are assigned when loading the block metadata in |
| * HdfsTable. |
| */ |
| public BlockReplica(short hostIdx, boolean isCached) { |
| hostIdx_ = hostIdx; |
| isCached_ = isCached; |
| } |
| |
| /** |
| * Parses the location (an ip address:port string) of the replica and returns a |
| * TNetworkAddress with this information, or null if parsing fails. |
| */ |
| public static TNetworkAddress parseLocation(String location) { |
| Preconditions.checkNotNull(location); |
| String[] ip_port = location.split(":"); |
| if (ip_port.length != 2) return null; |
| try { |
| return CatalogInterners.internNetworkAddress( |
| new TNetworkAddress(ip_port[0], Integer.parseInt(ip_port[1]))); |
| } catch (NumberFormatException e) { |
| return null; |
| } |
| } |
| |
| public boolean isCached() { return isCached_; } |
| public short getHostIdx() { return hostIdx_; } |
| } |
| |
| /** |
| * Static utility methods to serialize and access file block metadata from FlatBuffers. |
| */ |
| public static class FileBlock { |
| // Bit mask used to extract the replica host id and cache info of a file block. |
| // Use ~REPLICA_HOST_IDX_MASK to extract the cache info (stored in MSB). |
| private static short REPLICA_HOST_IDX_MASK = (1 << 15) - 1; |
| |
| /** |
| * Constructs an FbFileBlock object from the block location metadata |
| * 'loc'. Serializes the file block metadata into a FlatBuffer using 'fbb' and |
| * returns the offset in the underlying buffer where the encoded file block starts. |
| * 'hostIndex' stores the network addresses of the datanodes that store the files of |
| * the parent HdfsTable. Populates 'numUnknownDiskIds' with the number of unknown disk |
| * ids. |
| */ |
| public static int createFbFileBlock(FlatBufferBuilder fbb, BlockLocation loc, |
| ListMap<TNetworkAddress> hostIndex, Reference<Long> numUnknownDiskIds) |
| throws IOException { |
| Preconditions.checkNotNull(fbb); |
| Preconditions.checkNotNull(loc); |
| Preconditions.checkNotNull(hostIndex); |
| // replica host ids |
| FbFileBlock.startReplicaHostIdxsVector(fbb, loc.getNames().length); |
| Set<String> cachedHosts = Sets.newHashSet(loc.getCachedHosts()); |
| // Enumerate all replicas of the block, adding any unknown hosts |
| // to hostIndex. We pick the network address from getNames() and |
| // map it to the corresponding hostname from getHosts(). |
| for (int i = 0; i < loc.getNames().length; ++i) { |
| TNetworkAddress networkAddress = BlockReplica.parseLocation(loc.getNames()[i]); |
| short replicaIdx = (short) hostIndex.getIndex(networkAddress); |
| boolean isReplicaCached = cachedHosts.contains(loc.getHosts()[i]); |
| replicaIdx = makeReplicaIdx(isReplicaCached, replicaIdx); |
| fbb.addShort(replicaIdx); |
| } |
| int fbReplicaHostIdxOffset = fbb.endVector(); |
| short[] diskIds = createDiskIds(loc, numUnknownDiskIds); |
| Preconditions.checkState(diskIds.length == loc.getNames().length, |
| "Mismatch detected between number of diskIDs and block locations for block: " + |
| loc.toString()); |
| int fbDiskIdsOffset = FbFileBlock.createDiskIdsVector(fbb, diskIds); |
| FbFileBlock.startFbFileBlock(fbb); |
| FbFileBlock.addOffset(fbb, loc.getOffset()); |
| FbFileBlock.addLength(fbb, loc.getLength()); |
| FbFileBlock.addReplicaHostIdxs(fbb, fbReplicaHostIdxOffset); |
| FbFileBlock.addDiskIds(fbb, fbDiskIdsOffset); |
| return FbFileBlock.endFbFileBlock(fbb); |
| } |
| |
| private static short makeReplicaIdx(boolean isReplicaCached, int hostIdx) { |
| Preconditions.checkArgument((hostIdx & REPLICA_HOST_IDX_MASK) == hostIdx, |
| "invalid hostIdx: %s", hostIdx); |
| return isReplicaCached ? (short) (hostIdx | ~REPLICA_HOST_IDX_MASK) |
| : (short)hostIdx; |
| } |
| |
| /** |
| * Constructs an FbFileBlock object from the file block metadata that comprise block's |
| * 'offset', 'length' and replica index 'replicaIdx'. Serializes the file block |
| * metadata into a FlatBuffer using 'fbb' and returns the offset in the underlying |
| * buffer where the encoded file block starts. |
| */ |
| public static int createFbFileBlock(FlatBufferBuilder fbb, long offset, long length, |
| short replicaIdx) { |
| Preconditions.checkNotNull(fbb); |
| FbFileBlock.startReplicaHostIdxsVector(fbb, 1); |
| fbb.addShort(replicaIdx); |
| int fbReplicaHostIdxOffset = fbb.endVector(); |
| FbFileBlock.startFbFileBlock(fbb); |
| FbFileBlock.addOffset(fbb, offset); |
| FbFileBlock.addLength(fbb, length); |
| FbFileBlock.addReplicaHostIdxs(fbb, fbReplicaHostIdxOffset); |
| return FbFileBlock.endFbFileBlock(fbb); |
| } |
| |
| /** |
| * Creates the disk ids of a block from its BlockLocation 'location'. Returns the |
| * disk ids and populates 'numUnknownDiskIds' with the number of unknown disk ids. |
| */ |
| private static short[] createDiskIds(BlockLocation location, |
| Reference<Long> numUnknownDiskIds) throws IOException { |
| long unknownDiskIdCount = 0; |
| String[] storageIds = location.getStorageIds(); |
| String[] hosts = location.getHosts(); |
| if (storageIds.length != hosts.length) { |
| if (LOG.isTraceEnabled()) { |
| LOG.trace(String.format("Number of storage IDs and number of hosts for block " + |
| "%s mismatch (storageIDs:hosts) %d:%d. Skipping disk ID loading for this " + |
| "block.", location.toString(), storageIds.length, hosts.length)); |
| } |
| storageIds = new String[hosts.length]; |
| } |
| short[] diskIDs = new short[storageIds.length]; |
| for (int i = 0; i < storageIds.length; ++i) { |
| if (Strings.isNullOrEmpty(storageIds[i])) { |
| diskIDs[i] = (short) -1; |
| ++unknownDiskIdCount; |
| } else { |
| diskIDs[i] = DiskIdMapper.INSTANCE.getDiskId(hosts[i], storageIds[i]); |
| } |
| } |
| long count = numUnknownDiskIds.getRef() + unknownDiskIdCount; |
| numUnknownDiskIds.setRef(Long.valueOf(count)); |
| return diskIDs; |
| } |
| |
| public static long getOffset(FbFileBlock fbFileBlock) { return fbFileBlock.offset(); } |
| public static long getLength(FbFileBlock fbFileBlock) { return fbFileBlock.length(); } |
| // Returns true if there is at least one cached replica. |
| public static boolean hasCachedReplica(FbFileBlock fbFileBlock) { |
| boolean hasCachedReplica = false; |
| for (int i = 0; i < fbFileBlock.replicaHostIdxsLength(); ++i) { |
| hasCachedReplica |= isReplicaCached(fbFileBlock, i); |
| } |
| return hasCachedReplica; |
| } |
| |
| public static int getNumReplicaHosts(FbFileBlock fbFileBlock) { |
| return fbFileBlock.replicaHostIdxsLength(); |
| } |
| |
| public static int getReplicaHostIdx(FbFileBlock fbFileBlock, int pos) { |
| int idx = fbFileBlock.replicaHostIdxs(pos); |
| return idx & REPLICA_HOST_IDX_MASK; |
| } |
| |
| // Returns true if the block replica 'replicaIdx' is cached. |
| public static boolean isReplicaCached(FbFileBlock fbFileBlock, int replicaIdx) { |
| int idx = fbFileBlock.replicaHostIdxs(replicaIdx); |
| return (idx & ~REPLICA_HOST_IDX_MASK) != 0; |
| } |
| |
| /** |
| * Return the disk id of the block in BlockLocation.getNames()[hostIndex]; -1 if |
| * disk id is not supported. |
| */ |
| public static int getDiskId(FbFileBlock fbFileBlock, int hostIndex) { |
| if (fbFileBlock.diskIdsLength() == 0) return -1; |
| return fbFileBlock.diskIds(hostIndex); |
| } |
| |
| /** |
| * Returns a string representation of a FbFileBlock. |
| */ |
| public static String debugString(FbFileBlock fbFileBlock) { |
| int numReplicaHosts = getNumReplicaHosts(fbFileBlock); |
| List<Integer> diskIds = Lists.newArrayListWithCapacity(numReplicaHosts); |
| List<Integer> replicaHosts = Lists.newArrayListWithCapacity(numReplicaHosts); |
| List<Boolean> isBlockCached = Lists.newArrayListWithCapacity(numReplicaHosts); |
| for (int i = 0; i < numReplicaHosts; ++i) { |
| diskIds.add(getDiskId(fbFileBlock, i)); |
| replicaHosts.add(getReplicaHostIdx(fbFileBlock, i)); |
| isBlockCached.add(isReplicaCached(fbFileBlock, i)); |
| } |
| StringBuilder builder = new StringBuilder(); |
| return builder.append("Offset: " + getOffset(fbFileBlock)) |
| .append("Length: " + getLength(fbFileBlock)) |
| .append("IsCached: " + hasCachedReplica(fbFileBlock)) |
| .append("ReplicaHosts: " + Joiner.on(", ").join(replicaHosts)) |
| .append("DiskIds: " + Joiner.on(", ").join(diskIds)) |
| .append("Caching: " + Joiner.on(", ").join(isBlockCached)) |
| .toString(); |
| } |
| } |
| |
| // Struct-style class for caching all the information we need to reconstruct an |
| // HMS-compatible Partition object, for use in RPCs to the metastore. We do this rather |
| // than cache the Thrift partition object itself as the latter can be large - thanks |
| // mostly to the inclusion of the full FieldSchema list. This class is read-only - if |
| // any field can be mutated by Impala it should belong to HdfsPartition itself (see |
| // HdfsPartition.location_ for an example). |
| // |
| // TODO: Cache this descriptor in HdfsTable so that identical descriptors are shared |
| // between HdfsPartition instances. |
| // TODO: sdInputFormat and sdOutputFormat can be mutated by Impala when the file format |
| // of a partition changes; move these fields to HdfsPartition. |
| private static class CachedHmsPartitionDescriptor { |
| public String sdInputFormat; |
| public String sdOutputFormat; |
| public final boolean sdCompressed; |
| public final int sdNumBuckets; |
| public final org.apache.hadoop.hive.metastore.api.SerDeInfo sdSerdeInfo; |
| public final List<String> sdBucketCols; |
| public final List<org.apache.hadoop.hive.metastore.api.Order> sdSortCols; |
| public final Map<String, String> sdParameters; |
| public final int msCreateTime; |
| public final int msLastAccessTime; |
| |
| public CachedHmsPartitionDescriptor( |
| org.apache.hadoop.hive.metastore.api.Partition msPartition) { |
| org.apache.hadoop.hive.metastore.api.StorageDescriptor sd = null; |
| if (msPartition != null) { |
| sd = msPartition.getSd(); |
| CatalogInterners.internFieldsInPlace(sd); |
| msCreateTime = msPartition.getCreateTime(); |
| msLastAccessTime = msPartition.getLastAccessTime(); |
| } else { |
| msCreateTime = msLastAccessTime = 0; |
| } |
| if (sd != null) { |
| sdInputFormat = sd.getInputFormat(); |
| sdOutputFormat = sd.getOutputFormat(); |
| sdCompressed = sd.isCompressed(); |
| sdNumBuckets = sd.getNumBuckets(); |
| sdSerdeInfo = sd.getSerdeInfo(); |
| sdBucketCols = ImmutableList.copyOf(sd.getBucketCols()); |
| sdSortCols = ImmutableList.copyOf(sd.getSortCols()); |
| sdParameters = ImmutableMap.copyOf(CatalogInterners.internParameters( |
| sd.getParameters())); |
| } else { |
| sdInputFormat = ""; |
| sdOutputFormat = ""; |
| sdCompressed = false; |
| sdNumBuckets = 0; |
| sdSerdeInfo = null; |
| sdBucketCols = ImmutableList.of(); |
| sdSortCols = ImmutableList.of(); |
| sdParameters = ImmutableMap.of(); |
| } |
| } |
| } |
| |
| private final static Logger LOG = LoggerFactory.getLogger(HdfsPartition.class); |
| |
| // A predicate for checking if a given string is a key used for serializing |
| // TPartitionStats to HMS parameters. |
| private static Predicate<String> IS_INCREMENTAL_STATS_KEY = |
| new Predicate<String>() { |
| @Override |
| public boolean apply(String key) { |
| return key.startsWith(PartitionStatsUtil.INCREMENTAL_STATS_NUM_CHUNKS) |
| || key.startsWith(PartitionStatsUtil.INCREMENTAL_STATS_CHUNK_PREFIX); |
| } |
| }; |
| |
| private final HdfsTable table_; |
| private final List<LiteralExpr> partitionKeyValues_; |
| // estimated number of rows in partition; -1: unknown |
| private long numRows_ = -1; |
| private static AtomicLong partitionIdCounter_ = new AtomicLong(); |
| |
| // A unique ID for each partition, used to identify a partition in the thrift |
| // representation of a table. |
| private final long id_; |
| |
| /* |
| * Note: Although you can write multiple formats to a single partition (by changing |
| * the format before each write), Hive won't let you read that data and neither should |
| * we. We should therefore treat mixing formats inside one partition as user error. |
| * It's easy to add per-file metadata to FileDescriptor if this changes. |
| */ |
| private HdfsStorageDescriptor fileFormatDescriptor_; |
| |
| /** |
| * The file descriptors of this partition, encoded as flatbuffers. Storing the raw |
| * byte arrays here instead of the FileDescriptor object saves 100 bytes per file |
| * given the following overhead: |
| * - FileDescriptor object: |
| * - 16 byte object header |
| * - 8 byte reference to FbFileDesc |
| * - FbFileDesc superclass: |
| * - 16 byte object header |
| * - 56-byte ByteBuffer |
| * - 4-byte padding (objects are word-aligned) |
| */ |
| @Nonnull |
| private ImmutableList<byte[]> encodedFileDescriptors_; |
| private HdfsPartitionLocationCompressor.Location location_; |
| private boolean isDirty_; |
| // True if this partition is marked as cached. Does not necessarily mean the data is |
| // cached. |
| private boolean isMarkedCached_; |
| private final TAccessLevel accessLevel_; |
| |
| // (k,v) pairs of parameters for this partition, stored in the HMS. |
| private Map<String, String> hmsParameters_; |
| |
| // Binary representation of the TPartitionStats for this partition. Populated |
| // when the partition is loaded and updated using setPartitionStatsBytes(). |
| private byte[] partitionStats_; |
| |
| // True if partitionStats_ has intermediate_col_stats populated. |
| private boolean hasIncrementalStats_ ; |
| |
| // The last committed write ID which modified this partition. |
| // -1 means writeId_ is irrelevant(not supported). |
| private long writeId_ = -1L; |
| |
| private final InFlightEvents inFlightEvents_ = new InFlightEvents(20); |
| |
| private HdfsPartition(HdfsTable table, |
| org.apache.hadoop.hive.metastore.api.Partition msPartition, |
| List<LiteralExpr> partitionKeyValues, |
| HdfsStorageDescriptor fileFormatDescriptor, |
| List<HdfsPartition.FileDescriptor> fileDescriptors, long id, |
| HdfsPartitionLocationCompressor.Location location, TAccessLevel accessLevel) { |
| table_ = table; |
| if (msPartition == null) { |
| cachedMsPartitionDescriptor_ = null; |
| } else { |
| cachedMsPartitionDescriptor_ = new CachedHmsPartitionDescriptor(msPartition); |
| } |
| location_ = location; |
| partitionKeyValues_ = ImmutableList.copyOf(partitionKeyValues); |
| setFileDescriptors(fileDescriptors); |
| fileFormatDescriptor_ = fileFormatDescriptor; |
| id_ = id; |
| accessLevel_ = accessLevel; |
| if (msPartition != null && msPartition.getParameters() != null) { |
| isMarkedCached_ = HdfsCachingUtil.getCacheDirectiveId( |
| msPartition.getParameters()) != null; |
| hmsParameters_ = msPartition.getParameters(); |
| } else { |
| hmsParameters_ = new HashMap<>(); |
| } |
| addInflightVersionsFromParameters(); |
| extractAndCompressPartStats(); |
| // Intern parameters after removing the incremental stats |
| hmsParameters_ = CatalogInterners.internParameters(hmsParameters_); |
| if (MetastoreShim.getMajorVersion() > 2 && msPartition != null) { |
| writeId_ = MetastoreShim.getWriteIdFromMSPartition(msPartition); |
| } |
| } |
| |
| public HdfsPartition(HdfsTable table, |
| org.apache.hadoop.hive.metastore.api.Partition msPartition, |
| List<LiteralExpr> partitionKeyValues, |
| HdfsStorageDescriptor fileFormatDescriptor, |
| List<HdfsPartition.FileDescriptor> fileDescriptors, |
| TAccessLevel accessLevel) { |
| this(table, msPartition, partitionKeyValues, fileFormatDescriptor, fileDescriptors, |
| partitionIdCounter_.getAndIncrement(), |
| table.getPartitionLocationCompressor().new Location(msPartition != null |
| ? msPartition.getSd().getLocation() |
| : table.getLocation()), |
| accessLevel); |
| } |
| |
| @Override // FeFsPartition |
| public HdfsStorageDescriptor getInputFormatDescriptor() { |
| return fileFormatDescriptor_; |
| } |
| |
| @Override // FeFsPartition |
| public boolean isCacheable() { |
| return FileSystemUtil.isPathCacheable(new Path(getLocation())); |
| } |
| |
| @Override // FeFsPartition |
| public String getPartitionName() { |
| // TODO: Consider storing the PartitionKeyValue in HdfsPartition. It would simplify |
| // this code would be useful in other places, such as fromThrift(). |
| return FeCatalogUtils.getPartitionName(this); |
| } |
| |
| @Override |
| public List<String> getPartitionValuesAsStrings(boolean mapNullsToHiveKey) { |
| return FeCatalogUtils.getPartitionValuesAsStrings(this, mapNullsToHiveKey); |
| } |
| |
| @Override // FeFsPartition |
| public String getConjunctSql() { |
| // TODO: Remove this when the TODO elsewhere in this file to save and expose the |
| // list of TPartitionKeyValues has been resolved. |
| return FeCatalogUtils.getConjunctSqlForPartition(this); |
| } |
| |
| /** |
| * Returns a string of the form part_key1=value1/part_key2=value2... |
| */ |
| public String getValuesAsString() { |
| StringBuilder partDescription = new StringBuilder(); |
| for (int i = 0; i < getTable().getNumClusteringCols(); ++i) { |
| String columnName = getTable().getColumns().get(i).getName(); |
| String value = PartitionKeyValue.getPartitionKeyValueString( |
| getPartitionValues().get(i), |
| getTable().getNullPartitionKeyValue()); |
| partDescription.append(columnName + "=" + value); |
| if (i != getTable().getNumClusteringCols() - 1) partDescription.append("/"); |
| } |
| return partDescription.toString(); |
| } |
| |
| /** |
| * Returns the storage location (HDFS path) of this partition. Should only be called |
| * for partitioned tables. |
| */ |
| @Override |
| public String getLocation() { |
| return (location_ != null) ? location_.toString() : null; |
| } |
| |
| @Override |
| public THdfsPartitionLocation getLocationAsThrift() { |
| return location_ != null ? location_.toThrift() : null; |
| } |
| |
| @Override // FeFsPartition |
| public Path getLocationPath() { |
| Preconditions.checkNotNull(getLocation(), "HdfsPartition location is null"); |
| return new Path(getLocation()); |
| } |
| |
| @Override // FeFsPartition |
| public long getId() { return id_; } |
| |
| @Override // FeFsPartition |
| public HdfsTable getTable() { return table_; } |
| |
| @Override |
| public FileSystemUtil.FsType getFsType() { |
| Preconditions.checkNotNull(getLocationPath().toUri().getScheme(), |
| "Cannot get scheme from path " + getLocationPath()); |
| return FileSystemUtil.FsType.getFsType(getLocationPath().toUri().getScheme()); |
| } |
| |
| public void setNumRows(long numRows) { numRows_ = numRows; } |
| @Override // FeFsPartition |
| public long getNumRows() { return numRows_; } |
| @Override |
| public boolean isMarkedCached() { return isMarkedCached_; } |
| void markCached() { isMarkedCached_ = true; } |
| private CachedHmsPartitionDescriptor cachedMsPartitionDescriptor_; |
| |
| /** |
| * Updates the file format of this partition and sets the corresponding input/output |
| * format classes. |
| */ |
| public void setFileFormat(HdfsFileFormat fileFormat) { |
| fileFormatDescriptor_ = fileFormatDescriptor_.cloneWithChangedFileFormat( |
| fileFormat); |
| cachedMsPartitionDescriptor_.sdInputFormat = fileFormat.inputFormat(); |
| cachedMsPartitionDescriptor_.sdOutputFormat = fileFormat.outputFormat(); |
| cachedMsPartitionDescriptor_.sdSerdeInfo.setSerializationLib( |
| fileFormatDescriptor_.getFileFormat().serializationLib()); |
| } |
| |
| @Override // FeFsPartition |
| public HdfsFileFormat getFileFormat() { |
| return fileFormatDescriptor_.getFileFormat(); |
| } |
| |
| public void setLocation(String place) { |
| location_ = table_.getPartitionLocationCompressor().new Location(place); |
| } |
| |
| public org.apache.hadoop.hive.metastore.api.SerDeInfo getSerdeInfo() { |
| return cachedMsPartitionDescriptor_.sdSerdeInfo; |
| } |
| |
| @Override // FeFsPartition |
| public TPartitionStats getPartitionStats() { |
| return PartitionStatsUtil.getPartStatsOrWarn(this); |
| } |
| |
| @Override |
| public byte[] getPartitionStatsCompressed() { |
| return partitionStats_; |
| } |
| |
| public void setPartitionStatsBytes(byte[] partitionStats, boolean hasIncrStats) { |
| if (hasIncrStats) Preconditions.checkNotNull(partitionStats); |
| partitionStats_ = partitionStats; |
| hasIncrementalStats_ = hasIncrStats; |
| } |
| |
| /** |
| * Helper method that removes the partition stats from hmsParameters_, compresses them |
| * and updates partitionsStats_. |
| */ |
| private void extractAndCompressPartStats() { |
| try { |
| // Convert the stats stored in the hmsParams map to a deflate-compressed in-memory |
| // byte array format. After conversion, delete the entries in the hmsParams map |
| // as they are not needed anymore. |
| Reference<Boolean> hasIncrStats = new Reference<Boolean>(false); |
| byte[] partitionStats = |
| PartitionStatsUtil.partStatsBytesFromParameters(hmsParameters_, hasIncrStats); |
| setPartitionStatsBytes(partitionStats, hasIncrStats.getRef()); |
| } catch (ImpalaException e) { |
| LOG.warn(String.format("Failed to set partition stats for table %s partition %s", |
| getTable().getFullName(), getPartitionName()), e); |
| } finally { |
| // Delete the incremental stats entries. Cleared even on error conditions so that |
| // we do not persist the corrupt entries in the hmsParameters_ map when it is |
| // flushed to the HMS. |
| Maps.filterKeys(hmsParameters_, IS_INCREMENTAL_STATS_KEY).clear(); |
| } |
| } |
| |
| public void dropPartitionStats() { setPartitionStatsBytes(null, false); } |
| |
| @Override // FeFsPartition |
| public boolean hasIncrementalStats() { return hasIncrementalStats_; } |
| |
| @Override // FeFsPartition |
| public TAccessLevel getAccessLevel() { return accessLevel_; } |
| |
| @Override // FeFsPartition |
| public Map<String, String> getParameters() { |
| // Once the TPartitionStats in the parameters map are converted to a compressed |
| // format, the hmsParameters_ map should not contain any partition stats keys. |
| // Even though filterKeys() is O(n), we are not worried about the performance here |
| // since hmsParameters_ should be pretty small once the partition stats are removed. |
| Preconditions.checkState( |
| Maps.filterKeys(hmsParameters_, IS_INCREMENTAL_STATS_KEY).isEmpty()); |
| return hmsParameters_; |
| } |
| |
| public void putToParameters(String k, String v) { |
| Preconditions.checkArgument(!IS_INCREMENTAL_STATS_KEY.apply(k)); |
| hmsParameters_.put(k, v); |
| } |
| |
| public void putToParameters(Pair<String, String> kv) { |
| putToParameters(kv.first, kv.second); |
| } |
| |
| /** |
| * Gets the current list of versions for in-flight events for this partition |
| */ |
| public List<Long> getVersionsForInflightEvents() { |
| return inFlightEvents_.getAll(); |
| } |
| |
| /** |
| * Removes a given version from the in-flight events |
| * @param versionNumber version number to remove |
| * @return true if the versionNumber was removed, false if it didn't exist |
| */ |
| public boolean removeFromVersionsForInflightEvents(long versionNumber) { |
| return inFlightEvents_.remove(versionNumber); |
| } |
| |
| /** |
| * Adds a version number to the in-flight events of this partition |
| * @param versionNumber version number to add |
| */ |
| public void addToVersionsForInflightEvents(long versionNumber) { |
| if (!inFlightEvents_.add(versionNumber)) { |
| LOG.warn(String.format("Could not add %s version to the partition %s of table %s. " |
| + "This could cause unnecessary refresh of the partition when the event is" |
| + "received by the Events processor.", versionNumber, getPartitionName(), |
| getTable().getFullName())); |
| } |
| } |
| |
| /** |
| * Adds the version from the given Partition parameters. No-op if the parameters does |
| * not contain the <code>MetastoreEventPropertyKey.CATALOG_VERSION</code> |
| */ |
| private void addInflightVersionsFromParameters() { |
| Preconditions.checkNotNull(hmsParameters_); |
| Preconditions.checkState(inFlightEvents_.size() == 0); |
| if (!hmsParameters_.containsKey(MetastoreEventPropertyKey.CATALOG_VERSION.getKey())) { |
| return; |
| } |
| inFlightEvents_.add(Long.parseLong( |
| hmsParameters_.get(MetastoreEventPropertyKey.CATALOG_VERSION.getKey()))); |
| } |
| /** |
| * Marks this partition's metadata as "dirty" indicating that changes have been |
| * made and this partition's metadata should not be reused during the next |
| * incremental metadata refresh. |
| */ |
| public void markDirty() { isDirty_ = true; } |
| public boolean isDirty() { return isDirty_; } |
| |
| @Override // FeFsPartition |
| public List<LiteralExpr> getPartitionValues() { return partitionKeyValues_; } |
| @Override // FeFsPartition |
| public LiteralExpr getPartitionValue(int i) { return partitionKeyValues_.get(i); } |
| |
| @Override // FeFsPartition |
| public List<HdfsPartition.FileDescriptor> getFileDescriptors() { |
| // Return a lazily transformed list from our internal bytes storage. |
| return Lists.transform(encodedFileDescriptors_, FileDescriptor.FROM_BYTES); |
| } |
| |
| /** |
| * Returns a set of fully qualified file names in the partition. |
| */ |
| public Set<String> getFileNames() { |
| List<FileDescriptor> fdList = getFileDescriptors(); |
| Set<String> fileNames = new HashSet<>(fdList.size()); |
| // Fully qualified file names. |
| String location = getLocation(); |
| for (FileDescriptor fd : fdList) { |
| fileNames.add(location + Path.SEPARATOR + fd.getRelativePath()); |
| } |
| return fileNames; |
| } |
| |
| public void setFileDescriptors(List<FileDescriptor> descriptors) { |
| // Store an eagerly transformed-and-copied list so that we drop the memory usage |
| // of the flatbuffer wrapper. |
| encodedFileDescriptors_ = ImmutableList.copyOf(Lists.transform( |
| descriptors, FileDescriptor.TO_BYTES)); |
| } |
| |
| @Override // FeFsPartition |
| public int getNumFileDescriptors() { |
| return encodedFileDescriptors_.size(); |
| } |
| |
| @Override |
| public boolean hasFileDescriptors() { return !encodedFileDescriptors_.isEmpty(); } |
| |
| public CachedHmsPartitionDescriptor getCachedMsPartitionDescriptor() { |
| return cachedMsPartitionDescriptor_; |
| } |
| |
| /** |
| * Returns a Hive-compatible partition object that may be used in calls to the |
| * metastore. |
| */ |
| public org.apache.hadoop.hive.metastore.api.Partition toHmsPartition() { |
| if (cachedMsPartitionDescriptor_ == null) return null; |
| Preconditions.checkNotNull(table_.getNonPartitionFieldSchemas()); |
| // Update the serde library class based on the currently used file format. |
| org.apache.hadoop.hive.metastore.api.StorageDescriptor storageDescriptor = |
| new org.apache.hadoop.hive.metastore.api.StorageDescriptor( |
| // Make a shallow copy of the field schemas instead of passing a reference to |
| // the source list since it could potentially be modified once the current |
| // thread is out of table lock scope. |
| new ArrayList<>(table_.getNonPartitionFieldSchemas()), |
| getLocation(), |
| cachedMsPartitionDescriptor_.sdInputFormat, |
| cachedMsPartitionDescriptor_.sdOutputFormat, |
| cachedMsPartitionDescriptor_.sdCompressed, |
| cachedMsPartitionDescriptor_.sdNumBuckets, |
| cachedMsPartitionDescriptor_.sdSerdeInfo, |
| cachedMsPartitionDescriptor_.sdBucketCols, |
| cachedMsPartitionDescriptor_.sdSortCols, |
| cachedMsPartitionDescriptor_.sdParameters); |
| // Make a copy so that the callers do not need to delete the incremental stats |
| // strings from the hmsParams_ map later. |
| Map<String, String> hmsParams = Maps.newHashMap(getParameters()); |
| PartitionStatsUtil.partStatsToParams(this, hmsParams); |
| org.apache.hadoop.hive.metastore.api.Partition partition = |
| new org.apache.hadoop.hive.metastore.api.Partition( |
| getPartitionValuesAsStrings(true), getTable().getDb().getName(), |
| getTable().getName(), cachedMsPartitionDescriptor_.msCreateTime, |
| cachedMsPartitionDescriptor_.msLastAccessTime, storageDescriptor, |
| hmsParams); |
| return partition; |
| } |
| |
| public static HdfsPartition prototypePartition( |
| HdfsTable table, HdfsStorageDescriptor storageDescriptor) { |
| List<LiteralExpr> emptyExprList = new ArrayList<>(); |
| List<FileDescriptor> emptyFileDescriptorList = new ArrayList<>(); |
| return new HdfsPartition(table, null, emptyExprList, |
| storageDescriptor, emptyFileDescriptorList, |
| CatalogObjectsConstants.PROTOTYPE_PARTITION_ID, null, |
| TAccessLevel.READ_WRITE); |
| } |
| |
| @Override |
| public long getSize() { |
| long result = 0; |
| for (HdfsPartition.FileDescriptor fileDescriptor: getFileDescriptors()) { |
| result += fileDescriptor.getFileLength(); |
| } |
| return result; |
| } |
| |
| @Override |
| public String toString() { |
| return Objects.toStringHelper(this) |
| .add("fileDescriptors", getFileDescriptors()) |
| .toString(); |
| } |
| |
| public static HdfsPartition fromThrift(HdfsTable table, |
| long id, THdfsPartition thriftPartition) { |
| HdfsStorageDescriptor storageDesc = HdfsStorageDescriptor.fromThriftPartition( |
| thriftPartition, table.getName()); |
| |
| List<LiteralExpr> literalExpr = new ArrayList<>(); |
| if (id != CatalogObjectsConstants.PROTOTYPE_PARTITION_ID) { |
| List<Column> clusterCols = new ArrayList<>(); |
| for (int i = 0; i < table.getNumClusteringCols(); ++i) { |
| clusterCols.add(table.getColumns().get(i)); |
| } |
| |
| List<TExprNode> exprNodes = new ArrayList<>(); |
| for (TExpr expr: thriftPartition.getPartitionKeyExprs()) { |
| for (TExprNode node: expr.getNodes()) { |
| exprNodes.add(node); |
| } |
| } |
| Preconditions.checkState(clusterCols.size() == exprNodes.size(), |
| String.format("Number of partition columns (%d) does not match number " + |
| "of partition key expressions (%d)", |
| clusterCols.size(), exprNodes.size())); |
| |
| for (int i = 0; i < exprNodes.size(); ++i) { |
| literalExpr.add(LiteralExpr.fromThrift( |
| exprNodes.get(i), clusterCols.get(i).getType())); |
| } |
| } |
| |
| List<HdfsPartition.FileDescriptor> fileDescriptors = new ArrayList<>(); |
| if (thriftPartition.isSetFile_desc()) { |
| for (THdfsFileDesc desc: thriftPartition.getFile_desc()) { |
| fileDescriptors.add(HdfsPartition.FileDescriptor.fromThrift(desc)); |
| } |
| } |
| |
| TAccessLevel accessLevel = thriftPartition.isSetAccess_level() ? |
| thriftPartition.getAccess_level() : TAccessLevel.READ_WRITE; |
| HdfsPartitionLocationCompressor.Location location = thriftPartition.isSetLocation() |
| ? table.getPartitionLocationCompressor().new Location( |
| thriftPartition.getLocation()) |
| : null; |
| HdfsPartition partition = new HdfsPartition(table, null, literalExpr, storageDesc, |
| fileDescriptors, id, location, accessLevel); |
| if (thriftPartition.isSetStats()) { |
| partition.setNumRows(thriftPartition.getStats().getNum_rows()); |
| } |
| if (thriftPartition.isSetIs_marked_cached()) { |
| partition.isMarkedCached_ = thriftPartition.isIs_marked_cached(); |
| } |
| |
| if (thriftPartition.isSetHms_parameters()) { |
| partition.hmsParameters_ = CatalogInterners.internParameters( |
| thriftPartition.getHms_parameters()); |
| } else { |
| partition.hmsParameters_ = new HashMap<>(); |
| } |
| |
| partition.hasIncrementalStats_ = thriftPartition.has_incremental_stats; |
| if (thriftPartition.isSetPartition_stats()) { |
| partition.partitionStats_ = thriftPartition.getPartition_stats(); |
| } |
| |
| partition.writeId_ = thriftPartition.isSetWrite_id() ? |
| thriftPartition.getWrite_id() : -1L; |
| |
| return partition; |
| } |
| |
| /** |
| * Checks that this partition's metadata is well formed. This does not necessarily |
| * mean the partition is supported by Impala. |
| * Throws a CatalogException if there are any errors in the partition metadata. |
| */ |
| public void checkWellFormed() throws CatalogException { |
| try { |
| // Validate all the partition key/values to ensure you can convert them toThrift() |
| Expr.treesToThrift(getPartitionValues()); |
| } catch (Exception e) { |
| throw new CatalogException("Partition (" + getPartitionName() + |
| ") has invalid partition column values: ", e); |
| } |
| } |
| |
| /** |
| * Comparator to allow ordering of partitions by their partition-key values. |
| */ |
| public static final KeyValueComparator KV_COMPARATOR = new KeyValueComparator(); |
| public static class KeyValueComparator implements Comparator<FeFsPartition> { |
| @Override |
| public int compare(FeFsPartition o1, FeFsPartition o2) { |
| return comparePartitionKeyValues(o1.getPartitionValues(), o2.getPartitionValues()); |
| } |
| } |
| |
| @VisibleForTesting |
| public static int comparePartitionKeyValues(List<LiteralExpr> lhs, |
| List<LiteralExpr> rhs) { |
| int sizeDiff = lhs.size() - rhs.size(); |
| if (sizeDiff != 0) return sizeDiff; |
| for(int i = 0; i < lhs.size(); ++i) { |
| int cmp = lhs.get(i).compareTo(rhs.get(i)); |
| if (cmp != 0) return cmp; |
| } |
| return 0; |
| } |
| |
| @Override |
| public long getWriteId() { |
| return writeId_; |
| } |
| } |
