sepuri sai krishna created SPARK-60320:
------------------------------------------
Summary: Fix wrong values from the vectorized DELTA_BYTE_ARRAY
reader for uncompressed pages
Key: SPARK-60320
URL: https://issues.apache.org/jira/browse/SPARK-60320
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.1.3, 4.2.0, 4.3.0, 5.0.0, 4.4.0
Reporter: sepuri sai krishna
The vectorized Parquet reader returns wrong values for {{STRING}}, {{BINARY}}
and {{DECIMAL}} columns that are encoded with {{DELTA_BYTE_ARRAY}} when the
column chunk is not compressed. There is no error, the strings and decimals are
just wrong. The non-vectorized reader returns the right values from the same
file.
On Spark 4.2.0 and 4.1.3 (official binaries), with default settings:
{code:sql}
SET spark.sql.parquet.compression.codec=uncompressed;
CREATE TABLE t USING parquet
OPTIONS ('parquet.writer.version' = 'PARQUET_2_0',
'parquet.enable.dictionary' = 'false')
AS SELECT id, CAST(id AS STRING) AS s, CAST(id AS DECIMAL(38, 2)) AS d FROM
range(5);
SELECT * FROM t ORDER BY id;
-- 0 \u0000 0.00
-- 1 \u0000 0.00
-- 2 \u0000 0.00
-- 3 \u0000 0.00
-- 4 0 1.00
SET spark.sql.parquet.enableVectorizedReader=false;
SELECT * FROM t ORDER BY id;
-- 0 0 0.00
-- 1 1 1.00
-- 2 2 2.00
-- 3 3 3.00
-- 4 4 4.00
{code}
When a later step reads such a value, the query can also fail with an unrelated
error, e.g. {{CAST_INVALID_INPUT}} on a join key or
{{NUMERIC_VALUE_OUT_OF_RANGE}} on a decimal.
{{VectorizedDeltaByteArrayReader.readValues}} and {{skipBinary}} copy each
suffix with
{code:java}
arrayData.appendBytes(suffixLength, suffix.array(), suffix.position());
{code}
which ignores {{suffix.arrayOffset()}}. That is only right while the page's
buffer starts at offset 0 of its array. With Hadoop vectored IO, which Parquet
enables by default since 1.16.0 (apache/parquet-java GH-3156), the column chunk
is read into a slice of a larger buffer, so the page of an uncompressed column
chunk has a non-zero array offset and the bytes are copied from the wrong
place. A compressed page is decompressed into a buffer of its own, which is why
only uncompressed files are affected. The other vectorized decoders, including
{{DELTA_LENGTH_BYTE_ARRAY}}, already add {{arrayOffset()}}.
* Affected by default: 4.1.x (Parquet 1.16.0), 4.2.x, 4.3 and branch-4.x
(Parquet 1.17.0), and master (Parquet 1.18.1). The 4.3.0 RC2 and 4.2.1 RC2 tags
have the same code and Parquet 1.17.0.
* 4.0.x (Parquet 1.15.2) is affected only with
{{parquet.hadoop.vectored.io.enabled=true}}, since vectored IO is off by
default there.
* Workarounds: {{spark.sql.parquet.enableVectorizedReader=false}}, or
{{parquet.hadoop.vectored.io.enabled=false}}.
Writing such a file takes the Parquet v2 writer and no compression. Spark's
defaults (v1 writer, snappy) do not do that, but other writers produce such
files too, and a v2 column whose dictionary falls back also uses
{{DELTA_BYTE_ARRAY}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
