Skip to content

[SPARK-21271][SQL] Ensure Unsafe.sizeInBytes is a multiple of 8 #18503

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

Closed
kiszk wants to merge 10 commits into from
Closed

[SPARK-21271][SQL] Ensure Unsafe.sizeInBytes is a multiple of 8 #18503

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
eb87c63
insert assert to ensure Unsafe.sizeInBytes is a multiple of 8
kiszk Jul 2, 2017
6cde32f
fix in ExternalSorter
kiszk Jul 2, 2017
ccc820f
fix in RowBasedKeyValueBatch
kiszk Jul 2, 2017
3bd72c9
fix size of a byte array in UnsafeRow since 4-byte area is used only …
kiszk Jul 5, 2017
8a7a948
round up size of key/value in UnsafeRow to a multiple of 8
kiszk Jul 5, 2017
7f5a269
add a test suite for storing to and recovering from a checkpoint
kiszk Jul 13, 2017
762f02a
address review comments
kiszk Jul 19, 2017
0159701
address review comments
kiszk Jul 25, 2017
54be80e
fix test failure by removing an assert for debugging
kiszk Jul 25, 2017
cc467de
address review comments
kiszk Jul 26, 2017
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
6 changes: 3 additions & 3 deletions ...main/java/org/apache/spark/sql/catalyst/expressions/FixedLengthRowBasedKeyValueBatch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ public UnsafeRow appendRow(Object kbase, long koff, int klen,

keyRowId = numRows;
keyRow.pointTo(base, recordOffset, klen);
valueRow.pointTo(base, recordOffset + klen, vlen + 4);

@cloud-fan cloud-fan Jul 5, 2017

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.

I'm wondering why we did this before. Was it a mistake?

@kiszk kiszk Jul 5, 2017 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I have the same question.
@sameeragarwal had similar question one year ago. However, no response from @ooq

@ooq ooq Jul 6, 2017

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.

I recall it being intentional.
See discussion here.

@kiszk kiszk Jul 7, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

@ooq thank you for pointing out interesting discussion.
This discussion seems to make sense for page management. The question of @cloud-fan and me is whether valueRow uses only vlen. I think that +4 is for page management.

valueRow.pointTo(base, recordOffset + klen, vlen);
numRows++;
return valueRow;
}
Expand Down Expand Up @@ -95,7 +95,7 @@ protected UnsafeRow getValueFromKey(int rowId) {
getKeyRow(rowId);
}
assert(rowId >= 0);
valueRow.pointTo(base, keyRow.getBaseOffset() + klen, vlen + 4);
valueRow.pointTo(base, keyRow.getBaseOffset() + klen, vlen);
return valueRow;
}

Expand Down Expand Up @@ -131,7 +131,7 @@ public boolean next() {
}

key.pointTo(base, offsetInPage, klen);
value.pointTo(base, offsetInPage + klen, vlen + 4);
value.pointTo(base, offsetInPage + klen, vlen);

offsetInPage += recordLength;
recordsInPage -= 1;
Expand Down
2 changes: 2 additions & 0 deletions sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeRow.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,7 @@ public UnsafeRow() {}
*/
public void pointTo(Object baseObject, long baseOffset, int sizeInBytes) {
assert numFields >= 0 : "numFields (" + numFields + ") should >= 0";
assert sizeInBytes % 8 == 0 : "sizeInBytes (" + sizeInBytes + ") should be a multiple of 8";

@cloud-fan cloud-fan Jul 19, 2017

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.

I think we only need the assertion here, in pointTo, and in setTotalSize. Other places are just checking length for existing unsafe rows, which is unnecessary.

@kiszk kiszk Jul 20, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Yes, done.

this.baseObject = baseObject;
this.baseOffset = baseOffset;
this.sizeInBytes = sizeInBytes;
Expand All @@ -183,6 +184,7 @@ public void pointTo(byte[] buf, int sizeInBytes) {
}

public void setTotalSize(int sizeInBytes) {
assert sizeInBytes % 8 == 0 : "sizeInBytes (" + sizeInBytes + ") should be a multiple of 8";
this.sizeInBytes = sizeInBytes;
}

Expand Down
6 changes: 3 additions & 3 deletions ...n/java/org/apache/spark/sql/catalyst/expressions/VariableLengthRowBasedKeyValueBatch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ public UnsafeRow appendRow(Object kbase, long koff, int klen,

keyRowId = numRows;
keyRow.pointTo(base, recordOffset + 8, klen);
valueRow.pointTo(base, recordOffset + 8 + klen, vlen + 4);
valueRow.pointTo(base, recordOffset + 8 + klen, vlen);
numRows++;
return valueRow;
}
Expand Down Expand Up @@ -102,7 +102,7 @@ public UnsafeRow getValueFromKey(int rowId) {
long offset = keyRow.getBaseOffset();
int klen = keyRow.getSizeInBytes();
int vlen = Platform.getInt(base, offset - 8) - klen - 4;
valueRow.pointTo(base, offset + klen, vlen + 4);
valueRow.pointTo(base, offset + klen, vlen);
return valueRow;
}

Expand Down Expand Up @@ -146,7 +146,7 @@ public boolean next() {
currentvlen = totalLength - currentklen;

key.pointTo(base, offsetInPage + 8, currentklen);
value.pointTo(base, offsetInPage + 8 + currentklen, currentvlen + 4);
value.pointTo(base, offsetInPage + 8 + currentklen, currentvlen);

offsetInPage += 8 + totalLength + 8;
recordsInPage -= 1;
Expand Down
7 changes: 4 additions & 3 deletions sql/catalyst/src/main/java/org/apache/spark/sql/execution/UnsafeExternalRowSorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,9 +208,10 @@ private static final class RowComparator extends RecordComparator {

@Override
public int compare(Object baseObj1, long baseOff1, Object baseObj2, long baseOff2) {
// TODO: Why are the sizes -1?
row1.pointTo(baseObj1, baseOff1, -1);
row2.pointTo(baseObj2, baseOff2, -1);
// Note that since ordering doesn't need the total length of the record, we just pass 0
// into the row.
row1.pointTo(baseObj1, baseOff1, 0);
row2.pointTo(baseObj2, baseOff2, 0);
return ordering.compare(row1, row2);
}
}
Expand Down
6 changes: 3 additions & 3 deletions sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeKVExternalSorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -238,10 +238,10 @@ private static final class KVComparator extends RecordComparator {

@Override
public int compare(Object baseObj1, long baseOff1, Object baseObj2, long baseOff2) {
// Note that since ordering doesn't need the total length of the record, we just pass -1
// Note that since ordering doesn't need the total length of the record, we just pass 0
// into the row.
row1.pointTo(baseObj1, baseOff1 + 4, -1);
row2.pointTo(baseObj2, baseOff2 + 4, -1);
row1.pointTo(baseObj1, baseOff1 + 4, 0);
row2.pointTo(baseObj2, baseOff2 + 4, 0);
return ordering.compare(row1, row2);
}
}
Expand Down
12 changes: 10 additions & 2 deletions ...n/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -363,7 +363,11 @@ private[state] class HDFSBackedStateStoreProvider extends StateStoreProvider wit
val valueRowBuffer = new Array[Byte](valueSize)
ByteStreams.readFully(input, valueRowBuffer, 0, valueSize)
val valueRow = new UnsafeRow(valueSchema.fields.length)
valueRow.pointTo(valueRowBuffer, valueSize)
// If valueSize in existing file is not multiple of 8, floor it to multiple of 8.
// This is work around for the following.

@cloud-fan cloud-fan Jul 26, 2017

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.

nit: This is a workaround for the following:

// Pre-Spark 2.3 mistakenly append 4 bytes to the value row in

@cloud-fan cloud-fan Jul 26, 2017

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.

nit: Prior to Spark 2.3, we mistakenly ...

// `FixedLengthRowBasedKeyValueBatch`, which gets persisted into the checkpoint data

@cloud-fan cloud-fan Jul 26, 2017

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.

and VariableLengthRowBasedKeyValueBatch, let's just say RowBasedKeyValueBatch

valueRow.pointTo(valueRowBuffer, (valueSize / 8) * 8)

@tdas tdas Jul 20, 2017 •
edited
Loading

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.

Nit: This isnt rounding. This essentially floor to the multiple of 8.
@cloud-fan is this safe to do with ANY row generated in earlier Spark 2.0 - 2.2? I want to be 100% sure.

@cloud-fan cloud-fan Jul 20, 2017

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.

yes, because the extra bytes that exceed the 8-bytes boundary in UnsafeRow will never be read, so it's safe to ignore them. And here we still respect the row length when reading the binary, so no data will be missed.

@kiszk can we add more comments to explain why this can happen? We should say that before Spark 2.3 we mistakenly append 4 bytes to the value row in aggregate buffer, which gets persisted into the checkpoint data and so on

@kiszk kiszk Jul 25, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Sure, added more comments.

@tdas tdas Jul 26, 2017

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.

@cloud-fan @kiszk Just to confirm again, are we absolutely sure that the issue is that there 4 extra bytes in checkpointed rows, and therefore this truncation is safe to do?

@cloud-fan cloud-fan Jul 26, 2017

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.

yes I'm absolutely sure, as we store aggregate buffers in checkpoint files.

BTW, I run this test without the workaround here, and it can still pass. Which means, after this PR, RowBasedKeyValueBatch is fixed, and the generated checkpoint files have corrected row length.

And I also run the test without the fix for RowBasedKeyValueBatch, and it fails. I think this proves that RowBasedKeyValueBatch is the reason why we have wrong row length in checkpoit files.

map.put(keyRow, valueRow)
}
}
Expand Down Expand Up @@ -427,7 +431,11 @@ private[state] class HDFSBackedStateStoreProvider extends StateStoreProvider wit
val valueRowBuffer = new Array[Byte](valueSize)
ByteStreams.readFully(input, valueRowBuffer, 0, valueSize)
val valueRow = new UnsafeRow(valueSchema.fields.length)
valueRow.pointTo(valueRowBuffer, valueSize)
// If valueSize in existing file is not multiple of 8, floor it to multiple of 8.
// This is work around for the following.
// Pre-Spark 2.3 mistakenly append 4 bytes to the value row in
// `FixedLengthRowBasedKeyValueBatch`, which gets persisted into the checkpoint data
valueRow.pointTo(valueRowBuffer, (valueSize / 8) * 8)
map.put(keyRow, valueRow)
}
}
Expand Down