ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-34212][SQL][FOLLOWUP] Parquet vectorized reader can read decimal fields with a larger precision

Browse source 
### What changes were proposed in this pull request?

This is a followup of https://github.com/apache/spark/pull/31357

#31357 added a very strong restriction to the vectorized parquet reader, that the spark data type must exactly match the physical parquet type, when reading decimal fields. This restriction is actually not necessary, as we can safely read parquet decimals with a larger precision. This PR releases this restriction a little bit.

### Why are the changes needed?

To not fail queries unnecessarily.

### Does this PR introduce _any_ user-facing change?

Yes, now users can read parquet decimals with mismatched `DecimalType` as long as the scale is the same and precision is larger.

### How was this patch tested?

updated test.

Closes #31443 from cloud-fan/improve.

Authored-by: Wenchen Fan <wenchen@databricks.com>
Signed-off-by: HyukjinKwon <gurwls223@apache.org>
This commit is contained in:
Wenchen Fan 2021-02-03 09:26:36 +09:00 committed by HyukjinKwon
parent 63866025d2
commit 00120ea537
2 changed files with 11 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
View file

@ -111,7 +111,9 @@ public class VectorizedColumnReader {
private boolean isDecimalTypeMatched(DataType dt) { private boolean isDecimalTypeMatched(DataType dt) {
DecimalType d = (DecimalType) dt; DecimalType d = (DecimalType) dt;
DecimalMetadata dm = descriptor.getPrimitiveType().getDecimalMetadata(); DecimalMetadata dm = descriptor.getPrimitiveType().getDecimalMetadata();
return dm != null && dm.getPrecision() == d.precision() && dm.getScale() == d.scale(); // It's OK if the required decimal precision is larger than or equal to the physical decimal
// precision in the Parquet metadata, as long as the decimal scale is the same.
return dm != null && dm.getPrecision() <= d.precision() && dm.getScale() == d.scale();
} }
private boolean canReadAsIntDecimal(DataType dt) { private boolean canReadAsIntDecimal(DataType dt) {

8
sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
View file

@ -3891,6 +3891,14 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
val df = sql("SELECT 1.0 a, CAST(1.23 AS DECIMAL(17, 2)) b, CAST(1.23 AS DECIMAL(36, 2)) c") val df = sql("SELECT 1.0 a, CAST(1.23 AS DECIMAL(17, 2)) b, CAST(1.23 AS DECIMAL(36, 2)) c")
df.write.parquet(path.toString) df.write.parquet(path.toString)
Seq(true, false).foreach { vectorizedReader =>
withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> vectorizedReader.toString) {
// We can read the decimal parquet field with a larger precision, if scale is the same.
val schema = "a DECIMAL(9, 1), b DECIMAL(18, 2), c DECIMAL(38, 2)"
checkAnswer(readParquet(schema, path), df)
}
}
withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") { withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
val schema1 = "a DECIMAL(3, 2), b DECIMAL(18, 3), c DECIMAL(37, 3)" val schema1 = "a DECIMAL(3, 2), b DECIMAL(18, 3), c DECIMAL(37, 3)"
checkAnswer(readParquet(schema1, path), df) checkAnswer(readParquet(schema1, path), df)