Skip to content

HIVE-29695: ClassCastException in MapJoin when Parquet String is mapped to Date #6578

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

Open
Aggarwal-Raghav wants to merge 2 commits into
base: master
Choose a base branch
from
Open

HIVE-29695: ClassCastException in MapJoin when Parquet String is mapped to Date #6578

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7a3691a
HIVE-29695: ClassCastException in MapJoin when Parquet String is mapp…
Aggarwal-Raghav Jul 3, 2026
dea8c33
Add support for Parquet Sting-to-Date in vectorization flow
Aggarwal-Raghav Jul 8, 2026
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
14 changes: 14 additions & 0 deletions ql/src/java/org/apache/hadoop/hive/ql/io/parquet/convert/ETypeConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.ArrayList;
Expand All @@ -24,6 +25,7 @@
import java.util.TimeZone;

import com.google.common.base.MoreObjects;
import org.apache.hadoop.hive.common.type.Date;
import org.apache.hadoop.hive.common.type.HiveDecimal;
import org.apache.hadoop.hive.common.type.Timestamp;
import org.apache.hadoop.hive.conf.HiveConf;
Expand Down Expand Up @@ -525,6 +527,18 @@ protected HiveVarcharWritable convert(Binary binary) {
return new HiveVarcharWritable(binary.getBytes(), ((VarcharTypeInfo) hiveTypeInfo).getLength());
}
};
case DATE:
return new BinaryConverter<DateWritableV2>(type, parent, index) {
@Override
protected DateWritableV2 convert(Binary binary) {
String s = new String(binary.getBytes(), StandardCharsets.UTF_8);
try {
return new DateWritableV2(Date.valueOf(s));
} catch (IllegalArgumentException e) {
return null;
}
}
};
}
}
// STRING type
Expand Down
138 changes: 125 additions & 13 deletions ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetDataColumnReaderFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package org.apache.hadoop.hive.ql.io.parquet.vector;

import org.apache.hadoop.hive.common.type.Date;
import org.apache.hadoop.hive.common.type.HiveBaseChar;
import org.apache.hadoop.hive.common.type.HiveDecimal;
import org.apache.hadoop.hive.common.type.Timestamp;
Expand Down Expand Up @@ -1934,6 +1935,66 @@ private byte[] truncateIfNecesssary(byte[] bytes) {
}
}

/**
* Reads Parquet string bytes for Date column ("2024-07-09") and converts them to long (days since
* epoch)
*/
public static class TypesFromStringToDatePageReader extends TypesFromStringPageReader {
/**
* A cache to pre-calculate string-to-date conversions for dictionary encoded pages.
*
* <p>'parquet.dictionary.page.size' defaults to 1MB. If a column has too many unique values,
* Parquet automatically disables the dictionary (isDict = false) and falls back to the
* ValuesReader flow. Therefore, a 1MB dictionary holds at most ~100,000 date strings (10 bytes
* each). This guarantees our long[] cache will never exceed ~800 KB, and the boolean[] ~100 KB!
*/
private long[] dictDateCache;
private boolean[] dictDateCacheValid;

public TypesFromStringToDatePageReader(ValuesReader realReader, int length) {
super(realReader, length);
}

public TypesFromStringToDatePageReader(Dictionary dict, int length) {
super(dict, length);
int maxId = dict.getMaxId();
// IDs are zero-indexed
dictDateCache = new long[maxId + 1];
dictDateCacheValid = new boolean[maxId + 1];
for (int i = 0; i <= maxId; i++) {
dictDateCache[i] = parseDateString(dict.decodeToBinary(i).getBytesUnsafe());
dictDateCacheValid[i] = isValid;
}
}

@Override
public long readLong() {
return parseDateString(valuesReader.readBytes().getBytesUnsafe());
}

@Override
public long readLong(int id) {
isValid = dictDateCacheValid[id];
return dictDateCache[id];
}

private long parseDateString(byte[] bytes) {
if (bytes == null) {
isValid = false;
return 0;
}
try {
String s = new String(bytes, StandardCharsets.UTF_8);
long epochDay = Date.valueOf(s).toEpochDay();
isValid = true;
return epochDay;
} catch (Exception e) {
isValid = false;
return 0;
}
}
}

private static ParquetDataColumnReader getDataColumnReaderByTypeHelper(boolean isDictionary,
PrimitiveType parquetType,
TypeInfo hiveType,
Expand Down Expand Up @@ -2021,13 +2082,26 @@ private static ParquetDataColumnReader getDataColumnReaderByTypeHelper(boolean i
}
}

/**
* Finds the right reader for Parquet string and binary data (BINARY and FIXED_LEN_BYTE_ARRAY).
*
* <p>This method handles cases where the Parquet data type doesn't perfectly match the
* Hive data type (like reading a Parquet String as a Hive Date). This is incredibly important
* during Vectorized MapJoins, because if we don't convert the data into the exact format Hive
* expects, it can crash the query with a ClassCastException.
*
* @param isDict True if the Parquet page uses dictionary encoding
* @param parquetType The primitive type from the Parquet schema
* @param hiveType The expected Hive TypeInfo
* @param valuesReader The fallback values reader if dictionary is not used
* @param dictionary The dictionary reader
* @return A configured ParquetDataColumnReader
*/
private static ParquetDataColumnReader getConvertorFromBinary(boolean isDict,
PrimitiveType parquetType,
TypeInfo hiveType,
ValuesReader valuesReader,
Dictionary dictionary) {
LogicalTypeAnnotation logicalType = parquetType.getLogicalTypeAnnotation();

// max length for varchar and char cases
int length = getVarcharLength(hiveType);
TypeInfo realHiveType = (hiveType instanceof ListTypeInfo) ?
Expand All @@ -2042,12 +2116,20 @@ private static ParquetDataColumnReader getConvertorFromBinary(boolean isDict,
int hiveScale = (typeName.equalsIgnoreCase(serdeConstants.DECIMAL_TYPE_NAME)) ?
((DecimalTypeInfo) realHiveType).getScale() : 0;

LogicalTypeAnnotation logicalType = parquetType.getLogicalTypeAnnotation();

if (logicalType == null) {
return isDict ? new DefaultParquetDataColumnReader(dictionary, length) : new
DefaultParquetDataColumnReader(valuesReader, length);
// Check if we need to convert the string into a specific Hive type (like DATE)
ParquetDataColumnReader reader =
getReaderForString(realHiveType, isDict, dictionary, valuesReader, length);
return reader != null
? reader
: (isDict
? new DefaultParquetDataColumnReader(dictionary, length)
: new DefaultParquetDataColumnReader(valuesReader, length));
}

Optional<ParquetDataColumnReader> reader = parquetType.getLogicalTypeAnnotation()
Optional<ParquetDataColumnReader> reader = logicalType
.accept(new LogicalTypeAnnotationVisitor<>() {
@Override public Optional<ParquetDataColumnReader> visit(
DecimalLogicalTypeAnnotation logicalTypeAnnotation) {
Expand All @@ -2062,18 +2144,48 @@ private static ParquetDataColumnReader getConvertorFromBinary(boolean isDict,

@Override public Optional<ParquetDataColumnReader> visit(
StringLogicalTypeAnnotation logicalTypeAnnotation) {
return isDict ? Optional
.of(new TypesFromStringPageReader(dictionary, length)) : Optional
.of(new TypesFromStringPageReader(valuesReader, length));
// Only try to find a specialized reader (like for DATE) right when we actually need it.
// If we don't find one, we just fall back to reading it as a normal string.
ParquetDataColumnReader reader = getReaderForString(
realHiveType, isDict, dictionary, valuesReader, length);
return Optional.of(reader != null ? reader :
(isDict
? new TypesFromStringPageReader(dictionary, length)
: new TypesFromStringPageReader(valuesReader, length)));
}
});

if (reader.isPresent()) {
return reader.get();
}
// If the visitor above didn't find a reader (e.g. for unsupported logical types),
// default to a standard binary reader.
return reader.orElseGet(
() ->
isDict
? new DefaultParquetDataColumnReader(dictionary, length)
: new DefaultParquetDataColumnReader(valuesReader, length));
}

return isDict ? new DefaultParquetDataColumnReader(dictionary, length) : new
DefaultParquetDataColumnReader(valuesReader, length);
/**
* Returns a specialized reader if we need to convert Parquet string bytes into a specific Hive
* data type (like DATE).
*
* <p>If the type doesn't need special string conversion, it simply returns null. This makes it
* easy to add support for other data types in the future if they also need to be converted from
* strings during vectorized reads.
*/
private static ParquetDataColumnReader getReaderForString(TypeInfo hiveType,
boolean isDict, Dictionary dictionary, ValuesReader valuesReader, int length) {
if (!(hiveType instanceof PrimitiveTypeInfo)) {
return null;
}
return switch (((PrimitiveTypeInfo) hiveType).getPrimitiveCategory()) {
case DATE -> isDict
? new TypesFromStringToDatePageReader(dictionary, length)
: new TypesFromStringToDatePageReader(valuesReader, length);
case STRING, VARCHAR, CHAR -> isDict
? new TypesFromStringPageReader(dictionary, length)
: new TypesFromStringPageReader(valuesReader, length);
default -> null;
};
}

public static ParquetDataColumnReader getDataColumnReaderByTypeOnDictionary(
Expand Down
30 changes: 30 additions & 0 deletions ql/src/test/org/apache/hadoop/hive/ql/io/parquet/convert/TestETypeConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import static org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory.getPrimitiveTypeInfo;
import static org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory.stringTypeInfo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;

import java.nio.ByteBuffer;
Expand All @@ -33,6 +34,7 @@
import org.apache.hadoop.hive.ql.io.parquet.convert.ETypeConverter.BinaryConverter;
import org.apache.hadoop.hive.ql.io.parquet.timestamp.NanoTime;
import org.apache.hadoop.hive.ql.io.parquet.timestamp.NanoTimeUtils;
import org.apache.hadoop.hive.serde2.io.DateWritableV2;
import org.apache.hadoop.hive.serde2.io.HiveCharWritable;
import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
import org.apache.hadoop.hive.serde2.io.HiveVarcharWritable;
Expand Down Expand Up @@ -341,6 +343,34 @@ public void testGetTextConverterNoHiveTypeInfo() {
assertEquals(value, textWritable.toString());
}

@Test
public void testGetTextConverterForDate() {
PrimitiveType primitiveType =
Types.optional(PrimitiveTypeName.BINARY)
.as(LogicalTypeAnnotation.stringType())
.named("value");
String value = "2026-01-01";
DateWritableV2 dateWritable =
(DateWritableV2)
getWritableFromBinaryConverter(
TypeInfoFactory.dateTypeInfo, primitiveType, Binary.fromString(value));
assertEquals(org.apache.hadoop.hive.common.type.Date.valueOf(value), dateWritable.get());
}

@Test
public void testGetTextConverterForDateInvalid() {
PrimitiveType primitiveType =
Types.optional(PrimitiveTypeName.BINARY)
.as(LogicalTypeAnnotation.stringType())
.named("value");
String value = "invalid_date";
DateWritableV2 dateWritable =
(DateWritableV2)
getWritableFromBinaryConverter(
TypeInfoFactory.dateTypeInfo, primitiveType, Binary.fromString(value));
assertNull(dateWritable);
}

@Test
public void testGetIntConverterForTinyInt() throws Exception {
PrimitiveType primitiveType = Types.optional(PrimitiveTypeName.INT32)
Expand Down
36 changes: 36 additions & 0 deletions ql/src/test/queries/clientpositive/parquet_string_to_date_vectorization.q
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
CREATE TABLE repro_parquet_string (
id INT,
col1 STRING
) STORED AS PARQUET;

INSERT INTO repro_parquet_string VALUES (1, '2026-01-01');

CREATE EXTERNAL TABLE repro_parquet_date (
id INT,
col1 DATE
) STORED AS PARQUET
LOCATION '${hiveconf:hive.metastore.warehouse.dir}/repro_parquet_string';

@deniskuzZ deniskuzZ Jul 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

do you need custom location?

@Aggarwal-Raghav Aggarwal-Raghav Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

we can use ALTER also instead of creating a new table on top of previously created table repro_parquet_string. The goal was to pass String to DATE col


CREATE TABLE small_table (
id INT,
date_col DATE
);
INSERT INTO small_table VALUES (1, '2026-01-01');

SET hive.auto.convert.join=true;
SET hive.vectorized.execution.enabled=false;

@deniskuzZ deniskuzZ Jul 7, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

is that only reproduced in non-vectorized flow ?

  ┌─────────────────────────────┬──────────────────────────────────────────────────────────┐                                                                                                                     
  │            Path             │           Where string-source conversions live           │                                                                                                                     
  ├─────────────────────────────┼──────────────────────────────────────────────────────────┤                                                                                                                     
  │ Row-by-row (non-vectorized) │ ETypeConverter.ESTRING_CONVERTER                         │                                                                                                                     
  ├─────────────────────────────┼──────────────────────────────────────────────────────────┤                                                                                                                     
  │ Vectorized                  │ ParquetDataColumnReaderFactory.TypesFromStringPageReader │                                                                                                                     
  └─────────────────────────────┴──────────────────────────────────────────────────────────┘

@Aggarwal-Raghav Aggarwal-Raghav Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

yes, when enabled the above exception in comments is thrown.

@deniskuzZ deniskuzZ Jul 13, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

what about other types like timestamp?

@Aggarwal-Raghav Aggarwal-Raghav Jul 13, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

it will not work for now.

  1. Should we support it? Because I was surprised that no one has faced this existing JIRA till now for String -> Date usecase.
  2. Then there will be TimestampTZ etc types as well
    If we want to support it can we do that in separate JIRA ? (improvement type)

@deniskuzZ deniskuzZ Jul 13, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I was surprised that no one has faced this existing JIRA till now

probably because it only reproduces in non-vectorized flow. Wonder if we can extract common converters and not duplicate code?

@Aggarwal-Raghav Aggarwal-Raghav Jul 13, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I’ll see how to optimize the approach for other types. I’ll get back to you in 2 days.


SELECT /*+ MAPJOIN(small_table) */ a.col1
FROM repro_parquet_date a
JOIN small_table b ON a.id = b.id;

SET hive.vectorized.execution.enabled=true;

EXPLAIN VECTORIZATION DETAIL
SELECT /*+ MAPJOIN(small_table) */ a.col1
FROM repro_parquet_date a
JOIN small_table b ON a.id = b.id;

SELECT /*+ MAPJOIN(small_table) */ a.col1
FROM repro_parquet_date a
JOIN small_table b ON a.id = b.id;
Loading
Loading