amogh-jahagirdar commented on code in PR #18604:
URL: https://github.com/apache/iceberg/pull/18604#discussion_r4342967893
##########
core/src/main/java/org/apache/iceberg/StatsUtil.java:
##########
@@ -325,4 +336,181 @@ private static boolean isScalar(Schema schema,
Map<Integer, Integer> parentIndex
return true;
}
+
+ /** Inclusive source column bounds for a partition value. Upper is null when
unbounded. */
+ record PartitionSourceBounds(Object lower, Object upper) {}
Review Comment:
I went with a plain bounds type here rather than returning FieldStats. The
adapter builds stats from the file's metric maps and owns the counts (it has
the record count), so building a FieldStats per partition field per file just
to pull the bounds back out would be wasted work on every manifest read.
##########
core/src/main/java/org/apache/iceberg/StatsUtil.java:
##########
@@ -325,4 +336,181 @@ private static boolean isScalar(Schema schema,
Map<Integer, Integer> parentIndex
return true;
}
+
+ /** Inclusive source column bounds for a partition value. Upper is null when
unbounded. */
+ record PartitionSourceBounds(Object lower, Object upper) {}
Review Comment:
I didn't put a tightness flag on these bounds since I think it makes more
sense for the adapter to set it. These bounds describe which source values a
partition value allows, while tightness is a statement about the actual values
in a file's stats, so it belongs with the adapter that builds those stats.
##########
core/src/main/java/org/apache/iceberg/StatsUtil.java:
##########
@@ -325,4 +336,181 @@ private static boolean isScalar(Schema schema,
Map<Integer, Integer> parentIndex
return true;
}
+
+ /** Inclusive source column bounds for a partition value. Upper is null when
unbounded. */
+ record PartitionSourceBounds(Object lower, Object upper) {}
+
+ /**
+ * Returns the source column bounds for a partition value, or null for a NaN
identity value.
+ *
+ * @throws IllegalArgumentException if the transform does not preserve order
or the value is null
+ */
+ static PartitionSourceBounds sourceBoundsFor(
+ Schema schema, PartitionField field, Object partitionValue) {
+ Preconditions.checkArgument(
+ field.transform().preservesOrder(),
Review Comment:
A null partition value means every source value is null for any transform
(including bucket), so I think the adapter should handle it up front, since it
has the record count and can just set the null counts. Returning null
lower/upper bounds here would make "all nulls" indistinguishable from "unknown
bounds". That leaves null as the return value only for a NaN identity value,
where there really are no bounds.
##########
core/src/main/java/org/apache/iceberg/StatsUtil.java:
##########
@@ -325,4 +336,181 @@ private static boolean isScalar(Schema schema,
Map<Integer, Integer> parentIndex
return true;
}
+
+ /** Inclusive source column bounds for a partition value. Upper is null when
unbounded. */
+ record PartitionSourceBounds(Object lower, Object upper) {}
+
+ /**
+ * Returns the source column bounds for a partition value, or null for a NaN
identity value.
+ *
+ * @throws IllegalArgumentException if the transform does not preserve order
or the value is null
+ */
+ static PartitionSourceBounds sourceBoundsFor(
+ Schema schema, PartitionField field, Object partitionValue) {
+ Preconditions.checkArgument(
+ field.transform().preservesOrder(),
+ "Cannot derive source column bounds from partition field that does not
preserve order: %s",
+ field);
+ Preconditions.checkArgument(
+ partitionValue != null, "Cannot derive source column bounds from null
partition value");
+ Type sourceType = schema.findType(field.sourceId());
+ Preconditions.checkArgument(
+ sourceType != null, "Cannot find source column for partition field:
%s", field);
+ return PartitionSpecVisitor.visit(
+ schema, field, new PartitionSourceBoundsVisitor(sourceType,
partitionValue));
+ }
+
+ private static class PartitionSourceBoundsVisitor
Review Comment:
I went with a PartitionSpecVisitor since each transform gets its own method,
which reads cleaner than a big switch over transform types that duplicates the
visitor's dispatch. The tradeoff is that visit() requires a Schema even though
we only need the source type, which I think is fine since the caller already
has it.
##########
core/src/main/java/org/apache/iceberg/StatsUtil.java:
##########
@@ -325,4 +336,181 @@ private static boolean isScalar(Schema schema,
Map<Integer, Integer> parentIndex
return true;
}
+
+ /** Inclusive source column bounds for a partition value. Upper is null when
unbounded. */
+ record PartitionSourceBounds(Object lower, Object upper) {}
+
+ /**
+ * Returns the source column bounds for a partition value, or null for a NaN
identity value.
+ *
+ * @throws IllegalArgumentException if the transform does not preserve order
or the value is null
+ */
+ static PartitionSourceBounds sourceBoundsFor(
+ Schema schema, PartitionField field, Object partitionValue) {
+ Preconditions.checkArgument(
+ field.transform().preservesOrder(),
+ "Cannot derive source column bounds from partition field that does not
preserve order: %s",
+ field);
+ Preconditions.checkArgument(
+ partitionValue != null, "Cannot derive source column bounds from null
partition value");
+ Type sourceType = schema.findType(field.sourceId());
+ Preconditions.checkArgument(
+ sourceType != null, "Cannot find source column for partition field:
%s", field);
+ return PartitionSpecVisitor.visit(
+ schema, field, new PartitionSourceBoundsVisitor(sourceType,
partitionValue));
+ }
+
+ private static class PartitionSourceBoundsVisitor
+ implements PartitionSpecVisitor<PartitionSourceBounds> {
+ private final Type sourceType;
+ private final Object value;
+
+ private PartitionSourceBoundsVisitor(Type sourceType, Object value) {
+ this.sourceType = sourceType;
+ this.value = value;
+ }
+
+ @Override
+ public PartitionSourceBounds identity(int fieldId, String sourceName, int
sourceId) {
+ // convert to the source type, which may have been promoted after the
value was written
+ Object bound =
+ switch (sourceType.typeId()) {
+ case LONG -> ((Number) value).longValue();
Review Comment:
The partition value can predate a type promotion, so identity widens int to
long and float to double, and truncate reads longs as Number for the same
reason. Decimal precision widening needs nothing since the scale doesn't
change. date to timestamp is only allowed for year/month/day, and those work
because we compute bounds from the current source type.
##########
core/src/main/java/org/apache/iceberg/StatsUtil.java:
##########
@@ -325,4 +336,181 @@ private static boolean isScalar(Schema schema,
Map<Integer, Integer> parentIndex
return true;
}
+
+ /** Inclusive source column bounds for a partition value. Upper is null when
unbounded. */
+ record PartitionSourceBounds(Object lower, Object upper) {}
+
+ /**
+ * Returns the source column bounds for a partition value, or null for a NaN
identity value.
+ *
+ * @throws IllegalArgumentException if the transform does not preserve order
or the value is null
+ */
+ static PartitionSourceBounds sourceBoundsFor(
+ Schema schema, PartitionField field, Object partitionValue) {
+ Preconditions.checkArgument(
+ field.transform().preservesOrder(),
+ "Cannot derive source column bounds from partition field that does not
preserve order: %s",
+ field);
+ Preconditions.checkArgument(
+ partitionValue != null, "Cannot derive source column bounds from null
partition value");
+ Type sourceType = schema.findType(field.sourceId());
+ Preconditions.checkArgument(
+ sourceType != null, "Cannot find source column for partition field:
%s", field);
+ return PartitionSpecVisitor.visit(
+ schema, field, new PartitionSourceBoundsVisitor(sourceType,
partitionValue));
+ }
+
+ private static class PartitionSourceBoundsVisitor
+ implements PartitionSpecVisitor<PartitionSourceBounds> {
+ private final Type sourceType;
+ private final Object value;
+
+ private PartitionSourceBoundsVisitor(Type sourceType, Object value) {
+ this.sourceType = sourceType;
+ this.value = value;
+ }
+
+ @Override
+ public PartitionSourceBounds identity(int fieldId, String sourceName, int
sourceId) {
+ // convert to the source type, which may have been promoted after the
value was written
+ Object bound =
+ switch (sourceType.typeId()) {
+ case LONG -> ((Number) value).longValue();
+ case DOUBLE -> ((Number) value).doubleValue();
+ default -> value;
+ };
+
+ if (bound instanceof ByteBuffer buffer) {
+ bound = ByteBuffers.copy(buffer);
+ }
+
+ // NaN cannot be used as a bound
+ if (bound instanceof Double d && d.isNaN() || bound instanceof Float f
&& f.isNaN()) {
+ return null;
+ }
+
+ return new PartitionSourceBounds(bound, bound);
+ }
+
+ @Override
+ public PartitionSourceBounds truncate(int fieldId, String sourceName, int
sourceId, int width) {
+ // source values are at most width - 1 larger than the partition value,
unscaled for decimals
+ return switch (sourceType.typeId()) {
+ case INTEGER -> {
+ int lower = (Integer) value;
+ yield new PartitionSourceBounds(
+ lower, (int) Math.min(Integer.MAX_VALUE, (long) lower + width -
1));
+ }
+ case LONG -> {
+ long lower = ((Number) value).longValue();
+ yield new PartitionSourceBounds(
+ lower, lower > Long.MAX_VALUE - (width - 1) ? Long.MAX_VALUE :
lower + width - 1);
+ }
+ case DECIMAL -> {
Review Comment:
Without this cap, for a truncate[7] partition value of 9999999.94 in a
decimal(9,2) column, sourceBoundsFor would return 10000000.00 as the upper
bound. That doesn't fit in a decimal(9,2), so it'd fail once we write it as a
v4 stat (bounds are stored as the column type, and the Parquet/Avro writers
reject decimals with too many digits). Capping at 9999999.99 keeps the bound a
valid decimal(9,2), which is the same thing we already do for int and long.
##########
core/src/test/java/org/apache/iceberg/TestStatsUtil.java:
##########
@@ -564,6 +571,273 @@ public void testContentStatsWriteSchemaOmitsListAndMap() {
assertThat(actual.fields()).isEmpty();
}
+ @Test
+ void identityBounds() {
+ assertThat(sourceBounds(Types.StringType.get(), Transforms.identity(),
"books"))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds("books", "books"));
+ }
+
+ @Test
+ void identityBoundsMatchSourceType() {
+ assertThat(sourceBounds(Types.LongType.get(), Transforms.identity(), 5))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds(5L, 5L));
+ assertThat(sourceBounds(Types.DoubleType.get(), Transforms.identity(),
1.5f))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds(1.5d, 1.5d));
+ }
+
+ @Test
+ void identityNaN() {
+ assertThat(sourceBounds(Types.FloatType.get(), Transforms.identity(),
Float.NaN)).isNull();
+ assertThat(sourceBounds(Types.DoubleType.get(), Transforms.identity(),
Double.NaN)).isNull();
+ }
+
+ private static Stream<Partition> truncatedNumbers() {
+ Type decimal = Types.DecimalType.of(9, 2);
+ return Stream.of(
+ partition(Types.IntegerType.get(), Transforms.truncate(10),
20).withExpectedBounds(20, 29),
+ partition(Types.IntegerType.get(), Transforms.truncate(10), -10)
+ .withExpectedBounds(-10, -1),
+ partition(Types.LongType.get(), Transforms.truncate(100), 12_300L)
+ .withExpectedBounds(12_300L, 12_399L),
+ partition(decimal, Transforms.truncate(50), new BigDecimal("12.00"))
+ .withExpectedBounds(new BigDecimal("12.00"), new
BigDecimal("12.49")),
+ partition(decimal, Transforms.truncate(50), new BigDecimal("-0.50"))
+ .withExpectedBounds(new BigDecimal("-0.50"), new
BigDecimal("-0.01")));
+ }
+
+ @ParameterizedTest
+ @MethodSource("truncatedNumbers")
+ void truncateNumberBounds(Partition partition) {
+ assertThat(sourceBounds(partition)).isEqualTo(partition.expectedBounds());
+ }
+
+ @Test
+ void truncateNumberNearMax() {
+ assertThat(sourceBounds(Types.IntegerType.get(), Transforms.truncate(10),
2_147_483_640))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds(2_147_483_640,
Integer.MAX_VALUE));
+ assertThat(sourceBounds(Types.LongType.get(), Transforms.truncate(100),
Long.MAX_VALUE - 7))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds(Long.MAX_VALUE - 7,
Long.MAX_VALUE));
+ assertThat(
+ sourceBounds(
+ Types.DecimalType.of(9, 2), Transforms.truncate(7), new
BigDecimal("9999999.94")))
+ .isEqualTo(
+ new StatsUtil.PartitionSourceBounds(
+ new BigDecimal("9999999.94"), new BigDecimal("9999999.99")));
+ }
+
+ @Test
+ void truncateStringBounds() {
+ assertThat(sourceBounds(Types.StringType.get(), Transforms.truncate(3),
"abc"))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds("abc", "abd"));
+ }
+
+ @Test
+ void truncateShortString() {
+ assertThat(sourceBounds(Types.StringType.get(), Transforms.truncate(3),
"ab"))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds("ab", "ab"));
+ }
+
+ @Test
+ void truncateShortStringWithSurrogatePair() {
+ String value = "ab" + new String(Character.toChars(0x1F4B0));
+
+ assertThat(sourceBounds(Types.StringType.get(), Transforms.truncate(4),
value))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds(value, value));
+ }
+
+ @Test
+ void truncateStringAtMaxCodePoint() {
+ String maxCodePoint = new
String(Character.toChars(Character.MAX_CODE_POINT));
+
+ assertThat(sourceBounds(Types.StringType.get(), Transforms.truncate(1),
maxCodePoint))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds(maxCodePoint, null));
+ }
+
+ @Test
+ void truncateBinaryBounds() {
+ ByteBuffer prefix = ByteBuffer.wrap(new byte[] {1, 2});
+
+ assertThat(sourceBounds(Types.BinaryType.get(), Transforms.truncate(2),
prefix))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds(prefix,
ByteBuffer.wrap(new byte[] {1, 3})));
+ }
+
+ @Test
+ void truncateShortBinary() {
+ ByteBuffer value = ByteBuffer.wrap(new byte[] {1});
+
+ assertThat(sourceBounds(Types.BinaryType.get(), Transforms.truncate(2),
value))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds(value, value));
+ }
+
+ @Test
+ void binaryBoundsDoNotSharePartitionValueBuffers() {
+ ByteBuffer value = ByteBuffer.wrap(new byte[] {1});
+ StatsUtil.PartitionSourceBounds identityBounds =
+ sourceBounds(Types.BinaryType.get(), Transforms.identity(), value);
+ StatsUtil.PartitionSourceBounds truncateBounds =
+ sourceBounds(Types.BinaryType.get(), Transforms.truncate(2), value);
+
+ value.put(0, (byte) 9);
+
+ ByteBuffer original = ByteBuffer.wrap(new byte[] {1});
+ assertThat(identityBounds).isEqualTo(new
StatsUtil.PartitionSourceBounds(original, original));
+ assertThat(truncateBounds).isEqualTo(new
StatsUtil.PartitionSourceBounds(original, original));
+ }
+
+ @Test
+ void truncateBinaryAtMaxBytes() {
+ ByteBuffer prefix = ByteBuffer.wrap(new byte[] {(byte) 0xFF, (byte) 0xFF});
+
+ assertThat(sourceBounds(Types.BinaryType.get(), Transforms.truncate(2),
prefix))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds(prefix, null));
+ }
+
+ private static Stream<Partition> timePartitionValues() {
+ Type date = Types.DateType.get();
+ Type timestamp = Types.TimestampType.withZone();
+ Type timestampNano = Types.TimestampNanoType.withZone();
+ return Stream.of(
+ partition(date, Transforms.year(), years("2023-01-01"))
+ .withExpectedBounds(days("2023-01-01"), days("2023-12-31")),
+ partition(date, Transforms.month(), months("2023-02-01"))
+ .withExpectedBounds(days("2023-02-01"), days("2023-02-28")),
+ partition(date, Transforms.day(), days("2023-02-10"))
+ .withExpectedBounds(days("2023-02-10"), days("2023-02-10")),
+ partition(timestamp, Transforms.year(), years("2023-01-01"))
+ .withExpectedBounds(
+ micros("2023-01-01T00:00:00Z"),
micros("2023-12-31T23:59:59.999999Z")),
+ partition(timestamp, Transforms.month(), months("2023-02-01"))
+ .withExpectedBounds(
+ micros("2023-02-01T00:00:00Z"),
micros("2023-02-28T23:59:59.999999Z")),
+ partition(timestamp, Transforms.day(), days("2023-02-10"))
+ .withExpectedBounds(
+ micros("2023-02-10T00:00:00Z"),
micros("2023-02-10T23:59:59.999999Z")),
+ partition(timestamp, Transforms.hour(), hours("2023-02-10T05:00:00Z"))
+ .withExpectedBounds(
+ micros("2023-02-10T05:00:00Z"),
micros("2023-02-10T05:59:59.999999Z")),
+ partition(timestampNano, Transforms.hour(),
hours("2023-02-10T05:00:00Z"))
+ .withExpectedBounds(
+ nanos("2023-02-10T05:00:00Z"),
nanos("2023-02-10T05:59:59.999999999Z")));
+ }
+
+ @ParameterizedTest
+ @MethodSource("timePartitionValues")
+ void timeBounds(Partition partition) {
+ assertThat(sourceBounds(partition)).isEqualTo(partition.expectedBounds());
+ }
+
+ private static Stream<Partition> timePartitionValuesAtOrBefore1970() {
+ Type timestamp = Types.TimestampType.withZone();
+ return Stream.of(
+ partition(timestamp, Transforms.year(), 0)
+ .withExpectedBounds(
+ micros("1969-01-01T00:00:00Z"),
micros("1970-12-31T23:59:59.999999Z")),
+ partition(timestamp, Transforms.hour(), -1)
+ .withExpectedBounds(
+ micros("1969-12-31T22:00:00Z"),
micros("1969-12-31T23:59:59.999999Z")),
+ partition(Types.DateType.get(), Transforms.month(), -1)
+ .withExpectedBounds(days("1969-11-01"), days("1969-12-31")),
+ partition(Types.DateType.get(), Transforms.day(), -1)
+ .withExpectedBounds(days("1969-12-31"), days("1969-12-31")));
+ }
+
+ @ParameterizedTest
+ @MethodSource("timePartitionValuesAtOrBefore1970")
+ void timeBoundsAtOrBefore1970(Partition partition) {
+ assertThat(sourceBounds(partition)).isEqualTo(partition.expectedBounds());
+ }
+
+ @Test
+ void dateBoundsNearMax() {
+ // the end of year 5881580 is past the largest date
+ assertThat(sourceBounds(Types.DateType.get(), Transforms.year(),
years("+5881580-01-01")))
+ .isEqualTo(new StatsUtil.PartitionSourceBounds(days("+5881580-01-01"),
Integer.MAX_VALUE));
+ }
+
+ @Test
+ void timestampNanoBoundsNearMax() {
+ // the end of 2262 is past the largest nanosecond timestamp
+ assertThat(
+ sourceBounds(
+ Types.TimestampNanoType.withZone(), Transforms.year(),
years("2262-01-01")))
+ .isEqualTo(
+ new StatsUtil.PartitionSourceBounds(nanos("2262-01-01T00:00:00Z"),
Long.MAX_VALUE));
+ }
+
+ @Test
+ void nullPartitionValueRejected() {
+ assertThatThrownBy(() -> sourceBounds(Types.TimestampType.withZone(),
Transforms.hour(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Cannot derive source column bounds from null partition
value");
+ }
+
+ @Test
+ void nonOrderPreservingTransformsRejected() {
+ assertThatThrownBy(() -> sourceBounds(Types.StringType.get(),
Transforms.bucket(16), 3))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Cannot derive source column bounds from partition field that does
not preserve order: 1000: part: bucket[16](1)");
+ assertThatThrownBy(() -> sourceBounds(Types.StringType.get(),
Transforms.alwaysNull(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Cannot derive source column bounds from partition field that does
not preserve order: 1000: part: void(1)");
+ }
+
+ private static StatsUtil.PartitionSourceBounds sourceBounds(
+ Type type, Transform<?, ?> transform, Object partitionValue) {
+ Schema schema = new Schema(Types.NestedField.optional(1, "col", type));
+ PartitionField field =
+ PartitionSpec.builderFor(schema).add(1, "part",
transform).build().fields().get(0);
+ return StatsUtil.sourceBoundsFor(schema, field, partitionValue);
+ }
+
+ private static StatsUtil.PartitionSourceBounds sourceBounds(Partition
partition) {
+ return sourceBounds(partition.type(), partition.transform(),
partition.value());
+ }
+
+ private static Partition partition(Type type, Transform<?, ?> transform,
Object value) {
+ return new Partition(type, transform, value, null, null);
+ }
+
+ private record Partition(
Review Comment:
I added this test structure to help with readability because otherwise each
case was a long list of positional arguments (type, transform, partition value,
lower, upper) and it was hard to tell which value was which. With the record
each case reads as the partition value and then its expected bounds, and the
dates and timestamps are written as ISO strings like days("2023-02-01") instead
of encoded integers like 637, so it's easy to check each case by eye.
##########
core/src/main/java/org/apache/iceberg/StatsUtil.java:
##########
@@ -325,4 +336,181 @@ private static boolean isScalar(Schema schema,
Map<Integer, Integer> parentIndex
return true;
}
+
+ /** Inclusive source column bounds for a partition value. Upper is null when
unbounded. */
+ record PartitionSourceBounds(Object lower, Object upper) {}
+
+ /**
+ * Returns the source column bounds for a partition value, or null for a NaN
identity value.
+ *
+ * @throws IllegalArgumentException if the transform does not preserve order
or the value is null
+ */
+ static PartitionSourceBounds sourceBoundsFor(
+ Schema schema, PartitionField field, Object partitionValue) {
+ Preconditions.checkArgument(
+ field.transform().preservesOrder(),
+ "Cannot derive source column bounds from partition field that does not
preserve order: %s",
+ field);
+ Preconditions.checkArgument(
+ partitionValue != null, "Cannot derive source column bounds from null
partition value");
+ Type sourceType = schema.findType(field.sourceId());
+ Preconditions.checkArgument(
+ sourceType != null, "Cannot find source column for partition field:
%s", field);
+ return PartitionSpecVisitor.visit(
+ schema, field, new PartitionSourceBoundsVisitor(sourceType,
partitionValue));
+ }
+
+ private static class PartitionSourceBoundsVisitor
+ implements PartitionSpecVisitor<PartitionSourceBounds> {
+ private final Type sourceType;
+ private final Object value;
+
+ private PartitionSourceBoundsVisitor(Type sourceType, Object value) {
+ this.sourceType = sourceType;
+ this.value = value;
+ }
+
+ @Override
+ public PartitionSourceBounds identity(int fieldId, String sourceName, int
sourceId) {
+ // convert to the source type, which may have been promoted after the
value was written
+ Object bound =
+ switch (sourceType.typeId()) {
+ case LONG -> ((Number) value).longValue();
+ case DOUBLE -> ((Number) value).doubleValue();
+ default -> value;
+ };
+
+ if (bound instanceof ByteBuffer buffer) {
+ bound = ByteBuffers.copy(buffer);
+ }
+
+ // NaN cannot be used as a bound
+ if (bound instanceof Double d && d.isNaN() || bound instanceof Float f
&& f.isNaN()) {
+ return null;
+ }
+
+ return new PartitionSourceBounds(bound, bound);
+ }
+
+ @Override
+ public PartitionSourceBounds truncate(int fieldId, String sourceName, int
sourceId, int width) {
+ // source values are at most width - 1 larger than the partition value,
unscaled for decimals
+ return switch (sourceType.typeId()) {
+ case INTEGER -> {
+ int lower = (Integer) value;
+ yield new PartitionSourceBounds(
+ lower, (int) Math.min(Integer.MAX_VALUE, (long) lower + width -
1));
+ }
+ case LONG -> {
+ long lower = ((Number) value).longValue();
+ yield new PartitionSourceBounds(
+ lower, lower > Long.MAX_VALUE - (width - 1) ? Long.MAX_VALUE :
lower + width - 1);
+ }
+ case DECIMAL -> {
+ BigDecimal lower = (BigDecimal) value;
+ int precision = ((Types.DecimalType) sourceType).precision();
+ BigInteger maxUnscaled =
BigInteger.TEN.pow(precision).subtract(BigInteger.ONE);
+ BigInteger unscaledUpper =
+ lower.unscaledValue().add(BigInteger.valueOf(width -
1)).min(maxUnscaled);
+ yield new PartitionSourceBounds(lower, new BigDecimal(unscaledUpper,
lower.scale()));
+ }
+ case STRING -> truncatedString(value.toString(), width);
+ case BINARY -> truncatedBinary((ByteBuffer) value, width);
+ default ->
+ throw new UnsupportedOperationException("Unsupported truncate
source: " + sourceType);
+ };
+ }
+
+ private static PartitionSourceBounds truncatedString(String prefix, int
width) {
+ if (prefix.codePointCount(0, prefix.length()) < width) {
+ // a prefix shorter than the width was not truncated, so every value
equals it
+ return new PartitionSourceBounds(prefix, prefix);
+ }
+
+ // add a character so truncateStringMax returns the next higher prefix
+ return new PartitionSourceBounds(
+ prefix, UnicodeUtil.truncateStringMax(prefix + "\u0000", width));
+ }
+
+ private static PartitionSourceBounds truncatedBinary(ByteBuffer value, int
width) {
+ ByteBuffer prefix = ByteBuffers.copy(value);
+ if (prefix.remaining() < width) {
+ // a prefix shorter than the width was not truncated, so every value
equals it
+ return new PartitionSourceBounds(prefix, prefix);
+ }
+
+ // add a byte so truncateBinaryMax returns the next higher prefix
+ ByteBuffer extended = ByteBuffer.allocate(prefix.remaining() + 1);
+ extended.put(prefix.duplicate()).put((byte) 0).flip();
+ return new PartitionSourceBounds(prefix,
BinaryUtil.truncateBinaryMax(extended, width));
+ }
+
+ @Override
+ public PartitionSourceBounds year(int fieldId, String sourceName, int
sourceId) {
+ return timeBoundsFor(ChronoUnit.YEARS);
+ }
+
+ @Override
+ public PartitionSourceBounds month(int fieldId, String sourceName, int
sourceId) {
+ return timeBoundsFor(ChronoUnit.MONTHS);
+ }
+
+ @Override
+ public PartitionSourceBounds day(int fieldId, String sourceName, int
sourceId) {
+ return timeBoundsFor(ChronoUnit.DAYS);
+ }
+
+ @Override
+ public PartitionSourceBounds hour(int fieldId, String sourceName, int
sourceId) {
+ return timeBoundsFor(ChronoUnit.HOURS);
+ }
+
+ private PartitionSourceBounds timeBoundsFor(ChronoUnit unit) {
+ long periods = (Integer) value;
+ // the lower bound starts one unit earlier for 0 or less since older
writers had a bug that
+ // rounded pre-1970 values toward zero, except for day transforms on
dates
+ long lowerPeriods = periods <= 0 ? periods - 1 : periods;
Review Comment:
Before #1981 (https://github.com/apache/iceberg/pull/1981, fixes #1680), the
date/time transforms rounded pre-1970 values toward zero, so a stored value can
be one unit too high. That's why the lower bound starts one unit earlier for
values <= 0 (not < 0, since all of 1969 was stored as year 0). Day on a date
was never affected, which is also why #1981's projection fix skips it.
--
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]
