amogh-jahagirdar commented on code in PR #18604:
URL: https://github.com/apache/iceberg/pull/18604#discussion_r4343106137
##########
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:
the important thing though is while this range is correct, we should be able
to correctly reconstruct the tuple afterwarsd. That would require some special
logic for these temporal types to "undo" what we're doing here to get the
actual value but I think so long as we are trying to account for this bug
that's required or we can say we no longer really care about this writer bug
but that feels kind of bad/inconsistent because our current evaluation code
already works around 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]
