Skip to content

Write null byte when indexing numeric dimensions with Hadoop #7020

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
gianm merged 4 commits into from
Mar 12, 2019
Merged

Write null byte when indexing numeric dimensions with Hadoop #7020

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7ad1d91
write null byte in hadoop indexing for numeric dimensions
ferristseng Feb 6, 2019
2d15e7d
Add test case to check output serializing null numeric dimensions
ferristseng Feb 28, 2019
65673cb
Remove extra line
ferristseng Mar 7, 2019
68eaded
Add @Nullable annotations
ferristseng Mar 7, 2019
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
59 changes: 38 additions & 21 deletions indexing-hadoop/src/main/java/org/apache/druid/indexer/InputRowSerde.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

/**
*/
Expand All @@ -63,6 +64,34 @@ public class InputRowSerde
private static final IndexSerdeTypeHelper FLOAT_HELPER = new FloatIndexSerdeTypeHelper();
private static final IndexSerdeTypeHelper DOUBLE_HELPER = new DoubleIndexSerdeTypeHelper();

private static <T extends Number> void writeNullableNumeric(
T ret,
final ByteArrayDataOutput out,
final Supplier<T> getDefault,
final Consumer<T> write)
{
if (ret == null) {
ret = getDefault.get();
}

// Write the null byte only if the default numeric value is still null.
if (ret == null) {
out.writeByte(NullHandling.IS_NULL_BYTE);
return;
}

if (NullHandling.sqlCompatible()) {
out.writeByte(NullHandling.IS_NOT_NULL_BYTE);
}

write.accept(ret);
}

private static boolean isNullByteSet(final ByteArrayDataInput in)
{
return NullHandling.sqlCompatible() && in.readByte() == NullHandling.IS_NULL_BYTE;
}

public interface IndexSerdeTypeHelper<T>
{
ValueType getType();
Expand Down Expand Up @@ -175,22 +204,18 @@ public void serialize(ByteArrayDataOutput out, Object value)
exceptionToThrow = pe;
}

if (ret == null) {
// remove null -> zero conversion when https://github.com/apache/incubator-druid/pull/5278 series of patches is merged
// we'll also need to change the serialized encoding so that it can represent numeric nulls
ret = DimensionHandlerUtils.ZERO_LONG;
}
out.writeLong(ret);
writeNullableNumeric(ret, out, NullHandling::defaultLongValue, out::writeLong);

if (exceptionToThrow != null) {
throw exceptionToThrow;
}
}

@Override
@Nullable
public Long deserialize(ByteArrayDataInput in)
{
return in.readLong();
return isNullByteSet(in) ? null : in.readLong();
Copy link
Copy Markdown
Member

@asdf2014 asdf2014 Mar 6, 2019

Choose a reason for hiding this comment

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

Perhaps it would be better to use a functional programming style here.

return Optional.ofNullable(in)
               .filter(InputRowSerde::isNotNullByteSet)
               .map(ByteArrayDataInput::readLong)
               .get();

Copy link
Copy Markdown
Member

@clintropolis clintropolis Mar 6, 2019

Choose a reason for hiding this comment

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

Eh, I sort of prefer it the way it currently is, seems clearer to me, is there any reason it would be better other than preference?

Copy link
Copy Markdown
Member

@asdf2014 asdf2014 Mar 6, 2019

Choose a reason for hiding this comment

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

Alright, I prefer the functional style because it makes the code more readable. If we don't use Optional, then we need add the @Nullable annotation for this method. It's up to you. 😅

Copy link
Copy Markdown
Contributor Author

@ferristseng ferristseng Mar 7, 2019 •
edited
Loading

Choose a reason for hiding this comment

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

I think I prefer the non-functional style. Also, maybe I'm misunderstanding, but wouldn't the get() cause the code to throw if the null byte is set?

I'll add @Nullable annotations to these deserialize methods

Copy link
Copy Markdown
Member

@asdf2014 asdf2014 Mar 8, 2019

Choose a reason for hiding this comment

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

but wouldn't the get() cause the code to throw if the null byte is set?

@ferristseng If the null byte is set, then get will return a null value. What you describe should be the orElseThrow function. Thanks for your contribution.

}
}

Expand All @@ -214,22 +239,18 @@ public void serialize(ByteArrayDataOutput out, Object value)
exceptionToThrow = pe;
}

if (ret == null) {
// remove null -> zero conversion when https://github.com/apache/incubator-druid/pull/5278 series of patches is merged
// we'll also need to change the serialized encoding so that it can represent numeric nulls
ret = DimensionHandlerUtils.ZERO_FLOAT;
}
out.writeFloat(ret);
writeNullableNumeric(ret, out, NullHandling::defaultFloatValue, out::writeFloat);

if (exceptionToThrow != null) {
throw exceptionToThrow;
}
}

@Override
@Nullable
public Float deserialize(ByteArrayDataInput in)
{
return in.readFloat();
return isNullByteSet(in) ? null : in.readFloat();
Copy link
Copy Markdown
Member

@asdf2014 asdf2014 Mar 6, 2019

Choose a reason for hiding this comment

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

Same.

return Optional.ofNullable(in)
               .filter(InputRowSerde::isNotNullByteSet)
               .map(ByteArrayDataInput::readFloat)
               .get();

}
}

Expand All @@ -253,22 +274,18 @@ public void serialize(ByteArrayDataOutput out, Object value)
exceptionToThrow = pe;
}

if (ret == null) {
// remove null -> zero conversion when https://github.com/apache/incubator-druid/pull/5278 series of patches is merged
// we'll also need to change the serialized encoding so that it can represent numeric nulls
ret = DimensionHandlerUtils.ZERO_DOUBLE;
}
out.writeDouble(ret);
writeNullableNumeric(ret, out, NullHandling::defaultDoubleValue, out::writeDouble);

if (exceptionToThrow != null) {
throw exceptionToThrow;
}
}

@Override
@Nullable
public Double deserialize(ByteArrayDataInput in)
{
return in.readDouble();
return isNullByteSet(in) ? null : in.readDouble();
Copy link
Copy Markdown
Member

@asdf2014 asdf2014 Mar 6, 2019

Choose a reason for hiding this comment

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

Same.

return Optional.ofNullable(in)
               .filter(InputRowSerde::isNotNullByteSet)
               .map(ByteArrayDataInput::readDouble)
               .get();

}
}

Expand Down
55 changes: 55 additions & 0 deletions indexing-hadoop/src/test/java/org/apache/druid/indexer/InputRowSerdeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -254,4 +254,59 @@ public void testDimensionParseExceptions()
result.getParseExceptionMessages()
);
}

@Test
public void testDimensionNullOrDefaultForNumerics()
{
HashMap<String, Object> eventWithNulls = new HashMap<>();
eventWithNulls.put("d1", null);
eventWithNulls.put("d2", Arrays.asList("d2v1", "d2v2"));
eventWithNulls.put("d3", null);
eventWithNulls.put("d4", null);
eventWithNulls.put("d5", null);

InputRow in = new MapBasedInputRow(
timestamp,
dims,
eventWithNulls
);

DimensionsSpec dimensionsSpec = new DimensionsSpec(
Arrays.asList(
new StringDimensionSchema("d1"),
new StringDimensionSchema("d2"),
new LongDimensionSchema("d3"),
new FloatDimensionSchema("d4"),
new DoubleDimensionSchema("d5")
),
null,
null
);

byte[] result = InputRowSerde.toBytes(InputRowSerde.getTypeHelperMap(dimensionsSpec), in, new AggregatorFactory[0]).getSerializedRow();

if (NullHandling.replaceWithDefault()) {
long expected = 0;
expected += 9; // timestamp bytes + dims length
expected += 18; // dim_non_existing writes: 1 16 1 bytes
expected += 4; // d1: writes 1 2 1 bytes
expected += 14; // d2: writes 1 2 1 1 4 1 4 bytes
expected += 11; // d3: writes 1 2 8 bytes
expected += 7; // d4: writes 1 2 4 bytes
expected += 11; // d5: writes 1 2 8 bytes
expected += 1; // writes aggregator length

Assert.assertEquals(expected, result.length);
Assert.assertArrayEquals(new byte[] {0, 0, 0, 0, 0, 0, 0, 0}, Arrays.copyOfRange(result, 48, 56));
Assert.assertArrayEquals(new byte[] {0, 0, 0, 0}, Arrays.copyOfRange(result, 59, 63));
Assert.assertArrayEquals(new byte[] {0, 0, 0, 0, 0, 0, 0, 0}, Arrays.copyOfRange(result, 66, 74));
} else {
long expected = 9 + 18 + 4 + 14 + 4 + 4 + 4 + 1;

Assert.assertEquals(expected, result.length);
Assert.assertEquals(result[48], NullHandling.IS_NULL_BYTE);
Assert.assertEquals(result[52], NullHandling.IS_NULL_BYTE);
Assert.assertEquals(result[56], NullHandling.IS_NULL_BYTE);
}
}
}