You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ul...@apache.org on 2024/08/22 01:05:31 UTC
(spark) branch master updated: [SPARK-48796][SQL] Optimze parquet binary getBytes with getBytesUnsafe to avoid copy cost
This is an automated email from the ASF dual-hosted git repository.
ulyssesyou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 72c1df39c049 [SPARK-48796][SQL] Optimze parquet binary getBytes with getBytesUnsafe to avoid copy cost
72c1df39c049 is described below
commit 72c1df39c04938f87edf5db89bb730e7bb220bc8
Author: xuyu <11...@vivo.com>
AuthorDate: Thu Aug 22 09:05:13 2024 +0800
[SPARK-48796][SQL] Optimze parquet binary getBytes with getBytesUnsafe to avoid copy cost
### What changes were proposed in this pull request?
Optimze parquet binary getBytes with getBytesUnsafe to avoid copy cost
### Why are the changes needed?
Binary##getBytes() need copy all the internal bytes and getBytesUnsafe not need with more quick performence.
```
/**
* Variant of getBytes() that avoids copying backing data structure by returning
* backing byte[] of the Binary. Do not modify backing byte[] unless you know what
* you are doing.
*
* return backing byte[] of correct size, with an offset of 0, if possible, else returns result of getBytes()
*/
public abstract byte[] getBytesUnsafe();
```
Add a benchmark test for it:
before the pr:
```
Java HotSpot(TM) 64-Bit Server VM 17.0.12+8-LTS-286 on Mac OS X 13.6.3
Intel(R) Core(TM) i7-1068NG7 CPU 2.30GHz
Parquet dictionary: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
Read binary dictionary 9137 9706 401 11.5 88.2 1.0X
```
after the pr:
```
Java HotSpot(TM) 64-Bit Server VM 17.0.12+8-LTS-286 on Mac OS X 13.6.3
Intel(R) Core(TM) i7-1068NG7 CPU 2.30GHz
Parquet dictionary: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
Read binary dictionary 9042 9502 370 10.9 87.1 1.0X
```
we found that after the pr speed had a improvemence.
<img width="833" alt="image" src="https://github.com/user-attachments/assets/d4343156-3bfd-4469-91b0-94979c901fd8">
This optizme also can refer from other patches which can improve parquet read pefermance, such as: https://github.com/apache/paimon/pull/3509
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Other test cover this pr.
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #47797 from xuzifu666/parquet_binary_optimze.
Lead-authored-by: xuyu <11...@vivo.com>
Co-authored-by: xuzifu666 <12...@qq.com>
Signed-off-by: youxiduo <yo...@corp.netease.com>
---
.../datasources/parquet/ParquetVectorUpdaterFactory.java | 10 +++++-----
.../sql/execution/datasources/parquet/ParquetFilters.scala | 8 ++++----
.../execution/datasources/parquet/ParquetRowConverter.scala | 6 +++---
3 files changed, 12 insertions(+), 12 deletions(-)
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
index b6065c24f2ec..70f806ba14f0 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
@@ -954,7 +954,7 @@ public class ParquetVectorUpdaterFactory {
WritableColumnVector dictionaryIds,
Dictionary dictionary) {
Binary v = dictionary.decodeToBinary(dictionaryIds.getDictId(offset));
- values.putByteArray(offset, v.getBytes());
+ values.putByteArray(offset, v.getBytesUnsafe());
}
}
@@ -1264,7 +1264,7 @@ public class ParquetVectorUpdaterFactory {
WritableColumnVector dictionaryIds,
Dictionary dictionary) {
Binary v = dictionary.decodeToBinary(dictionaryIds.getDictId(offset));
- values.putByteArray(offset, v.getBytes());
+ values.putByteArray(offset, v.getBytesUnsafe());
}
}
@@ -1498,7 +1498,7 @@ private static class BinaryToDecimalUpdater extends DecimalUpdater {
WritableColumnVector dictionaryIds,
Dictionary dictionary) {
BigInteger value =
- new BigInteger(dictionary.decodeToBinary(dictionaryIds.getDictId(offset)).getBytes());
+ new BigInteger(dictionary.decodeToBinary(dictionaryIds.getDictId(offset)).getBytesUnsafe());
BigDecimal decimal = new BigDecimal(value, parquetScale);
writeDecimal(offset, values, decimal);
}
@@ -1526,7 +1526,7 @@ private static class FixedLenByteArrayToDecimalUpdater extends DecimalUpdater {
int offset,
WritableColumnVector values,
VectorizedValuesReader valuesReader) {
- BigInteger value = new BigInteger(valuesReader.readBinary(arrayLen).getBytes());
+ BigInteger value = new BigInteger(valuesReader.readBinary(arrayLen).getBytesUnsafe());
BigDecimal decimal = new BigDecimal(value, this.parquetScale);
writeDecimal(offset, values, decimal);
}
@@ -1538,7 +1538,7 @@ private static class FixedLenByteArrayToDecimalUpdater extends DecimalUpdater {
WritableColumnVector dictionaryIds,
Dictionary dictionary) {
BigInteger value =
- new BigInteger(dictionary.decodeToBinary(dictionaryIds.getDictId(offset)).getBytes());
+ new BigInteger(dictionary.decodeToBinary(dictionaryIds.getDictId(offset)).getBytesUnsafe());
BigDecimal decimal = new BigDecimal(value, this.parquetScale);
writeDecimal(offset, values, decimal);
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
index 8c0b9aa0a427..3dc9ddf386f1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
@@ -806,8 +806,8 @@ class ParquetFilters(
}
override def keep(value: Binary): Boolean = {
- value != null && UTF8String.fromBytes(value.getBytes).startsWith(
- UTF8String.fromBytes(strToBinary.getBytes))
+ value != null && UTF8String.fromBytes(value.getBytesUnsafe).startsWith(
+ UTF8String.fromBytes(strToBinary.getBytesUnsafe))
}
}
)
@@ -822,7 +822,7 @@ class ParquetFilters(
override def canDrop(statistics: Statistics[Binary]): Boolean = false
override def inverseCanDrop(statistics: Statistics[Binary]): Boolean = false
override def keep(value: Binary): Boolean = {
- value != null && UTF8String.fromBytes(value.getBytes).endsWith(suffixStr)
+ value != null && UTF8String.fromBytes(value.getBytesUnsafe).endsWith(suffixStr)
}
}
)
@@ -837,7 +837,7 @@ class ParquetFilters(
override def canDrop(statistics: Statistics[Binary]): Boolean = false
override def inverseCanDrop(statistics: Statistics[Binary]): Boolean = false
override def keep(value: Binary): Boolean = {
- value != null && UTF8String.fromBytes(value.getBytes).contains(subStr)
+ value != null && UTF8String.fromBytes(value.getBytesUnsafe).contains(subStr)
}
}
)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
index 1bdff5234fd6..838eb30c38fb 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
@@ -95,7 +95,7 @@ private[parquet] class ParquetPrimitiveConverter(val updater: ParentContainerUpd
override def addLong(value: Long): Unit = updater.setLong(value)
override def addFloat(value: Float): Unit = updater.setFloat(value)
override def addDouble(value: Double): Unit = updater.setDouble(value)
- override def addBinary(value: Binary): Unit = updater.set(value.getBytes)
+ override def addBinary(value: Binary): Unit = updater.set(value.getBytesUnsafe)
}
/**
@@ -562,7 +562,7 @@ private[parquet] class ParquetRowConverter(
override def setDictionary(dictionary: Dictionary): Unit = {
this.expandedDictionary = Array.tabulate(dictionary.getMaxId + 1) { i =>
- UTF8String.fromBytes(dictionary.decodeToBinary(i).getBytes)
+ UTF8String.fromBytes(dictionary.decodeToBinary(i).getBytesUnsafe)
}
}
@@ -622,7 +622,7 @@ private[parquet] class ParquetRowConverter(
Decimal(unscaled, precision, scale)
} else {
// Otherwise, resorts to an unscaled `BigInteger` instead.
- Decimal(new BigDecimal(new BigInteger(value.getBytes), scale), precision, scale)
+ Decimal(new BigDecimal(new BigInteger(value.getBytesUnsafe), scale), precision, scale)
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org