Skip to content

Core: exclude NaN from upper/lower bound of floating columns in Parquet/ORC #2464

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rdblue merged 3 commits into from
Jun 24, 2021
Merged

Core: exclude NaN from upper/lower bound of floating columns in Parquet/ORC #2464

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b83dea3
exclude NaN from upper/lower bound of floating columns
yyanyy Apr 8, 2021
c7ad319
update based on comments
yyanyy Jun 19, 2021
43faaab
address comments
yyanyy Jun 23, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 7 additions & 9 deletions core/src/main/java/org/apache/iceberg/FieldMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,25 +20,23 @@
package org.apache.iceberg;


import java.nio.ByteBuffer;

/**
* Iceberg internally tracked field level metrics.
*/
public class FieldMetrics {
public class FieldMetrics<T> {

@yyanyy yyanyy Apr 12, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Change in this class overlaps with what #1963 contains. For now this class is not directly used/created by Parquet/ORC, and will be for Avro it will, thus should be safe to change. Same goes for FloatFieldMetrics.

private final int id;
private final long valueCount;
private final long nullValueCount;
private final long nanValueCount;
private final ByteBuffer lowerBound;
private final ByteBuffer upperBound;
private final T lowerBound;
private final T upperBound;

public FieldMetrics(int id,
long valueCount,
long nullValueCount,
long nanValueCount,
ByteBuffer lowerBound,
ByteBuffer upperBound) {
T lowerBound,
T upperBound) {
this.id = id;
this.valueCount = valueCount;
this.nullValueCount = nullValueCount;
Expand Down Expand Up @@ -78,14 +76,14 @@ public long nanValueCount() {
/**
* Returns the lower bound value of this field.
*/
public ByteBuffer lowerBound() {
public T lowerBound() {
return lowerBound;
}

/**
* Returns the upper bound value of this field.
*/
public ByteBuffer upperBound() {
public T upperBound() {
return upperBound;
}
}
79 changes: 61 additions & 18 deletions core/src/main/java/org/apache/iceberg/FloatFieldMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,25 +19,17 @@

package org.apache.iceberg;

import java.nio.ByteBuffer;

/**
* Iceberg internally tracked field level metrics, used by Parquet and ORC writers only.
* <p>
* Parquet/ORC keeps track of most metrics in file statistics, and only NaN counter is actually tracked by writers.
* This wrapper ensures that metrics not being updated by those writers will not be incorrectly used, by throwing
* exceptions when they are accessed.
*/
public class FloatFieldMetrics extends FieldMetrics {

/**
* Constructor for creating a FieldMetrics with only NaN counter.
* @param id field id being tracked by the writer
* @param nanValueCount number of NaN values, will only be non-0 for double or float field.
*/
public FloatFieldMetrics(int id,
long nanValueCount) {
super(id, 0L, 0L, nanValueCount, null, null);
public class FloatFieldMetrics extends FieldMetrics<Number> {

@rdblue rdblue Jun 16, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is a public class, I think it would make more sense to split it into an implementation for float and an implementation for double and have each implementation correctly set the type that is tracked. That avoids needing to test the type of upper bound or lower bound because you know that both are going to be the same. That's an assumption that the code below makes, but it isn't necessarily the case.

@yyanyy yyanyy Jun 19, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds good! I was hesitant to do that since I think there are too many duplicated code, I guess I was trying too hard to eliminate duplications...


private FloatFieldMetrics(AbstractFloatFieldMetricsContext<?> context) {
super(context.id, 0L, 0L, context.nanValueCount, context.lowerBound, context.upperBound);
}

@Override
Expand All @@ -50,13 +42,64 @@ public long nullValueCount() {
throw new IllegalStateException("Shouldn't access this method, as this metric is tracked in file statistics. ");
}

@Override
public ByteBuffer lowerBound() {
throw new IllegalStateException("Shouldn't access this method, as this metric is tracked in file statistics. ");
public static class FloatFieldMetricsContext extends AbstractFloatFieldMetricsContext<Float> {
public FloatFieldMetricsContext(int id) {
super(id);
}

@Override
public void updateMetricsContext(Float value) {
if (Float.isNaN(value)) {
this.nanValueCount++;
} else {
if (lowerBound == null || Float.compare(value, lowerBound) < 0) {
this.lowerBound = value;
}
if (upperBound == null || Float.compare(value, upperBound) > 0) {
this.upperBound = value;
}
}
}
}

@Override
public ByteBuffer upperBound() {
throw new IllegalStateException("Shouldn't access this method, as this metric is tracked in file statistics. ");
public static class DoubleFieldMetricsContext extends AbstractFloatFieldMetricsContext<Double> {

@rdblue rdblue Jun 16, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This class is suspiciously similar to the builder pattern, so I would probably just convert it to be a builder for FieldMetrics instead:

  • Rename DoubleFieldMetricsContext to DoubleFieldMetrics.Builder and create it with builderFor(int id)
  • Rename updateMetricsContext to something shorter, like addValue
  • Rename buildMetrics to just build

public DoubleFieldMetricsContext(int id) {
super(id);
}

@Override
public void updateMetricsContext(Double value) {
if (Double.isNaN(value)) {
this.nanValueCount++;
} else {
if (lowerBound == null || Double.compare(value, lowerBound) < 0) {
this.lowerBound = value;
}
if (upperBound == null || Double.compare(value, upperBound) > 0) {
this.upperBound = value;
}
}
}
}

@SuppressWarnings("checkstyle:VisibilityModifier")
public abstract static class AbstractFloatFieldMetricsContext<T extends Number> {
private final int id;
protected long nanValueCount = 0;
protected T lowerBound = null;
protected T upperBound = null;

public AbstractFloatFieldMetricsContext(int id) {
this.id = id;
}

/**
* It is caller's responsibility to ensure input shouldn't be null
*/
public abstract void updateMetricsContext(T value);

public FloatFieldMetrics buildMetrics() {
return new FloatFieldMetrics(this);
}
}
}
45 changes: 12 additions & 33 deletions core/src/test/java/org/apache/iceberg/TestMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,7 +274,7 @@ public void testMetricsForNestedStructFields() throws IOException {
assertBounds(6, BinaryType.get(),
ByteBuffer.wrap("A".getBytes()), ByteBuffer.wrap("A".getBytes()), metrics);
assertCounts(7, 1L, 0L, 1L, metrics);
assertBounds(7, DoubleType.get(), Double.NaN, Double.NaN, metrics);
assertBounds(7, DoubleType.get(), null, null, metrics);
}

private Record buildNestedTestRecord() {
Expand Down Expand Up @@ -354,9 +354,9 @@ public void testMetricsForNaNColumns() throws IOException {
Assert.assertEquals(2L, (long) metrics.recordCount());
assertCounts(1, 2L, 0L, 2L, metrics);
assertCounts(2, 2L, 0L, 2L, metrics);
// below: current behavior; will be null once NaN is excluded from upper/lower bound
assertBounds(1, FloatType.get(), Float.NaN, Float.NaN, metrics);
assertBounds(2, DoubleType.get(), Double.NaN, Double.NaN, metrics);

assertBounds(1, FloatType.get(), null, null, metrics);
assertBounds(2, DoubleType.get(), null, null, metrics);
}

@Test
Expand All @@ -367,15 +367,8 @@ public void testColumnBoundsWithNaNValueAtFront() throws IOException {
assertCounts(1, 3L, 0L, 1L, metrics);
assertCounts(2, 3L, 0L, 1L, metrics);

// below: current behavior; will be non-NaN values once NaN is excluded from upper/lower bound. ORC and Parquet's
// behaviors differ due to their implementation of comparison being different.
if (fileFormat() == FileFormat.ORC) {
assertBounds(1, FloatType.get(), Float.NaN, Float.NaN, metrics);
assertBounds(2, DoubleType.get(), Double.NaN, Double.NaN, metrics);
} else {
assertBounds(1, FloatType.get(), 1.2F, Float.NaN, metrics);
assertBounds(2, DoubleType.get(), 3.4D, Double.NaN, metrics);
}
assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
}

@Test
Expand All @@ -386,15 +379,8 @@ public void testColumnBoundsWithNaNValueInMiddle() throws IOException {
assertCounts(1, 3L, 0L, 1L, metrics);
assertCounts(2, 3L, 0L, 1L, metrics);

// below: current behavior; will be non-NaN values once NaN is excluded from upper/lower bound. ORC and Parquet's
// behaviors differ due to their implementation of comparison being different.
if (fileFormat() == FileFormat.ORC) {
assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
} else {
assertBounds(1, FloatType.get(), 1.2F, Float.NaN, metrics);
assertBounds(2, DoubleType.get(), 3.4D, Double.NaN, metrics);
}
assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
}

@Test
Expand All @@ -405,15 +391,8 @@ public void testColumnBoundsWithNaNValueAtEnd() throws IOException {
assertCounts(1, 3L, 0L, 1L, metrics);
assertCounts(2, 3L, 0L, 1L, metrics);

// below: current behavior; will be non-NaN values once NaN is excluded from upper/lower bound. ORC and Parquet's
// behaviors differ due to their implementation of comparison being different.
if (fileFormat() == FileFormat.ORC) {
assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
} else {
assertBounds(1, FloatType.get(), 1.2F, Float.NaN, metrics);
assertBounds(2, DoubleType.get(), 3.4D, Double.NaN, metrics);
}
assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
}

@Test
Expand Down Expand Up @@ -506,7 +485,7 @@ public void testMetricsForNestedStructFieldsWithMultipleRowGroup() throws IOExce
assertBounds(6, BinaryType.get(),
ByteBuffer.wrap("A".getBytes()), ByteBuffer.wrap("A".getBytes()), metrics);
assertCounts(7, 201L, 0L, 201L, metrics);
assertBounds(7, DoubleType.get(), Double.NaN, Double.NaN, metrics);
assertBounds(7, DoubleType.get(), null, null, metrics);
}

@Test
Expand Down Expand Up @@ -567,7 +546,7 @@ public void testFullMetricsMode() throws IOException {
assertBounds(6, Types.BinaryType.get(),
ByteBuffer.wrap("A".getBytes()), ByteBuffer.wrap("A".getBytes()), metrics);
assertCounts(7, 1L, 0L, 1L, metrics);
assertBounds(7, Types.DoubleType.get(), Double.NaN, Double.NaN, metrics);
assertBounds(7, Types.DoubleType.get(), null, null, metrics);
}

@Test
Expand Down
24 changes: 8 additions & 16 deletions data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriters.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,12 +219,10 @@ public void nonNullWrite(int rowId, Long data, ColumnVector output) {
}

private static class FloatWriter implements OrcValueWriter<Float> {
private final int id;
private long nanCount;
private final FloatFieldMetrics.FloatFieldMetricsContext floatFieldMetricsContext;

private FloatWriter(int id) {
this.id = id;
this.nanCount = 0;
this.floatFieldMetricsContext = new FloatFieldMetrics.FloatFieldMetricsContext(id);
}

@Override
Expand All @@ -235,24 +233,20 @@ public Class<Float> getJavaClass() {
@Override
public void nonNullWrite(int rowId, Float data, ColumnVector output) {
((DoubleColumnVector) output).vector[rowId] = data;
if (Float.isNaN(data)) {
nanCount++;
}
floatFieldMetricsContext.updateMetricsContext(data);
}

@Override
public Stream<FieldMetrics> metrics() {
return Stream.of(new FloatFieldMetrics(id, nanCount));
return Stream.of(floatFieldMetricsContext.buildMetrics());
}
}

private static class DoubleWriter implements OrcValueWriter<Double> {
private final int id;
private long nanCount;
private final FloatFieldMetrics.DoubleFieldMetricsContext doubleFieldMetricsContext;

private DoubleWriter(Integer id) {
this.id = id;
this.nanCount = 0;
this.doubleFieldMetricsContext = new FloatFieldMetrics.DoubleFieldMetricsContext(id);
}

@Override
Expand All @@ -263,14 +257,12 @@ public Class<Double> getJavaClass() {
@Override
public void nonNullWrite(int rowId, Double data, ColumnVector output) {
((DoubleColumnVector) output).vector[rowId] = data;
if (Double.isNaN(data)) {
nanCount++;
}
doubleFieldMetricsContext.updateMetricsContext(data);
}

@Override
public Stream<FieldMetrics> metrics() {
return Stream.of(new FloatFieldMetrics(id, nanCount));
return Stream.of(doubleFieldMetricsContext.buildMetrics());
}
}

Expand Down
Loading