ByteKeyRange implements Serializable,
        HasDefaultTracker<
            ByteKeyRange, org.apache.beam.sdk.transforms.splittabledofn.ByteKeyRangeTracker> { public double estimateFractionForKey(ByteKey key) { checkNotNull(key, "key"); checkArgument(!key.isEmpty(), "Cannot compute fraction for an empty key"); checkArgument( key.compareTo(startKey) >= 0, "Expected key %s >= range start key %s", key, startKey); if (key.equals(endKey)) { return 1.0; } checkArgument(containsKey(key), "Cannot compute fraction for %s outside this %s", key, this); byte[] startBytes = startKey.getBytes(); byte[] endBytes = endKey.getBytes(); byte[] keyBytes = key.getBytes(); if (endKey.isEmpty()) { startBytes = addHeadByte(startBytes, (byte) 0); endBytes = addHeadByte(endBytes, (byte) 1); keyBytes = addHeadByte(keyBytes, (byte) 0); } int paddedKeyLength = Math.max(Math.max(startBytes.length, endBytes.length), keyBytes.length); BigInteger rangeStartInt = paddedPositiveInt(startBytes, paddedKeyLength); BigInteger rangeEndInt = paddedPositiveInt(endBytes, paddedKeyLength); BigInteger keyInt = paddedPositiveInt(keyBytes, paddedKeyLength); BigInteger range = rangeEndInt.subtract(rangeStartInt); if (range.equals(BigInteger.ZERO)) { LOG.warn( "Using 0.0 as the default fraction for this near-empty range {} where start and end keys" + " differ only by trailing zeros.", this); return 0.0; } BigInteger progressScaled = keyInt.subtract(rangeStartInt).shiftLeft(64); return progressScaled.divide(range).doubleValue() / Math.pow(2, 64); } private  ByteKeyRange(ByteKey startKey, ByteKey endKey); static ByteKeyRange of(ByteKey startKey, ByteKey endKey); ByteKey getStartKey(); ByteKey getEndKey(); Boolean containsKey(ByteKey key); Boolean overlaps(ByteKeyRange other); List<ByteKey> split(int numSplits); double estimateFractionForKey(ByteKey key); ByteKey interpolateKey(double fraction); ByteKeyRange withStartKey(ByteKey startKey); ByteKeyRange withEndKey(ByteKey endKey); @Override String toString(); @Override boolean equals(@Nullable Object o); @Override int hashCode(); @Override org.apache.beam.sdk.transforms.splittabledofn.ByteKeyRangeTracker newTracker(); static final ByteKeyRange ALL_KEYS; }
@Test public void testEstimateFractionForKey() { final double delta = 0.0000001; assertEquals(0.5, ByteKeyRange.ALL_KEYS.estimateFractionForKey(ByteKey.of(0x80)), delta); ByteKeyRange after0 = ByteKeyRange.of(ByteKey.of(0), ByteKey.EMPTY); assertEquals(0.5, after0.estimateFractionForKey(ByteKey.of(0x80)), delta); ByteKeyRange after00 = ByteKeyRange.of(ByteKey.of(0, 0), ByteKey.EMPTY); assertEquals(0.5, after00.estimateFractionForKey(ByteKey.of(0x80)), delta); ByteKeyRange upToFE = ByteKeyRange.of(ByteKey.EMPTY, ByteKey.of(0xfe)); assertEquals(0.5, upToFE.estimateFractionForKey(ByteKey.of(0x7f)), delta); assertEquals(0.25, ByteKeyRange.ALL_KEYS.estimateFractionForKey(ByteKey.of(0x40)), delta); ByteKeyRange upTo80 = ByteKeyRange.of(ByteKey.EMPTY, ByteKey.of(0x80)); assertEquals(0.50, upTo80.estimateFractionForKey(ByteKey.of(0x40)), delta); ByteKeyRange range30to50 = ByteKeyRange.of(ByteKey.of(0x30), ByteKey.of(0x50)); assertEquals(0.50, range30to50.estimateFractionForKey(ByteKey.of(0x40)), delta); ByteKeyRange range31to4f = ByteKeyRange.of(ByteKey.of(0x30, 0, 1), ByteKey.of(0x4f, 0xff, 0xff, 0, 0)); assertEquals(0.50, range31to4f.estimateFractionForKey(ByteKey.of(0x40)), delta); ByteKeyRange upTo47 = ByteKeyRange.of(ByteKey.EMPTY, ByteKey.of(47)); for (int i = 0; i <= 47; ++i) { assertEquals("i=" + i, i / 47.0, upTo47.estimateFractionForKey(ByteKey.of(i)), delta); } ByteKeyRange rangeFDECtoFDEC83 = ByteKeyRange.of(ByteKey.of(0xfd, 0xec), ByteKey.of(0xfd, 0xec, 83)); for (int i = 0; i <= 83; ++i) { assertEquals( "i=" + i, i / 83.0, rangeFDECtoFDEC83.estimateFractionForKey(ByteKey.of(0xfd, 0xec, i)), delta); } }