amogh-jahagirdar commented on code in PR #18604:
URL: https://github.com/apache/iceberg/pull/18604#discussion_r4403724152
##########
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);
+ }
Review Comment:
Good callout on the integer min wrap around. A partition value that
overflowed can never be a multiple of the truncation width, so we can detect it
consistently. As a result, we can create a range like [MIN, partitionValue -
2^32 + W - 1], i.e. undo the overflow to get the actual truncated value, then
add W - 1 to get the largest value that truncates to it. Later, for eq. delete
matching (adapting newer metadata to older on read), we can reconstruct the
tuple by applying truncate to the lower bound, which gives back the same
partition value.
##########
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);
+ }
Review Comment:
Switched to duplicate for the upper bound. I think the way to reason about
this is what defensive measures we need given how this is consumed. The v4
adapter only reads the bounds, and PartitionData.get already returns a fresh
copy, so I dropped the extra copy. Duplicate is a cheap guard against a future
adapter bug where reading one bound moves the other.
##########
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;
Review Comment:
NaN is a valid partition value (e.g. identity on a double column with a NaN
row), so throwing would break existing tables. Metrics never use NaN as a bound
either, and null just means the adapter falls back to the file's own stats.
Switched to NaNUtil.isNaN.
##########
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 left it untyped on purpose, we could change it but I think the usage would
end up a bit strange. With the current visitor, one PartitionSpecVisitor
returns bounds for every transform, and the type is only known at runtime from
the source column. So the visitor would have to return
PartitionSourceBounds<?>, and every caller would just see a wildcard anyway.
Let me know what you think.
--
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]
