This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new eeebb520118 Preserve wide Parquet INT96 timestamps (#20060)
eeebb520118 is described below
commit eeebb5201181f6c534331d35599cf2d23da18f1f
Author: Xiang Fu <[email protected]>
AuthorDate: Tue Oct 27 20:33:34 2026 -0700
Preserve wide Parquet INT96 timestamps (#20060)
* Fix Parquet INT96 timestamp overflow
Reproduce: read an INT96 timestamp in year 1600 or 2300; it wraps through
signed epoch nanoseconds. Decode logical timestamps directly and reject raw
values outside the Long range.
* Reject Parquet UINT64 values outside the signed Long range
Reproduce: read UINT64 9223372036854775808 or 18446744073709551615 through
the native reader; both become negative Long values. Preserve representable
Long values and reject overflow at every nesting level.
* Clarify the Parquet INT96 overflow carry calculation
* Name the time units used for Parquet INT96 decoding
* Advance Parquet row position before extracting values
Reproduce: skip rejected UINT64 records, read the valid row, then check
hasNext at the end of the file. Count each successfully fetched physical row so
extraction errors cannot leave iteration stuck.
* Advance the Avro Parquet reader after rejected records
Reproduce: read an INT96 year-1600 row in raw mode and catch the conversion
error; the next call retries that row. Advance in finally and cover
rejected/valid/EOF plus rewind with a real Parquet file.
* Preserve legacy UINT64 signed-bit values
Reproduce: read a UINT64 hashed ID above Long.MAX_VALUE; rejection fails
previously valid ingestion. Preserve its signed Long bits and cover both
readers plus nested values, while retaining the INT96 fixes.
* Consolidate INT96 fixtures and cover rejected records at EOF
Share canonical Instant encoding across extractor tests and check trailing
rejection through EOF and rewind.
Correct the Avro and raw INT96 documentation; unchecked fetch
classification is tracked separately in #20065.
---
.../parquet/ParquetAvroRecordExtractor.java | 12 +--
.../parquet/ParquetAvroRecordReader.java | 20 ++---
.../parquet/ParquetNativeRecordExtractor.java | 9 ++-
.../parquet/ParquetNativeRecordReader.java | 3 +-
.../parquet/ParquetRecordReaderConfig.java | 1 +
.../plugin/inputformat/parquet/ParquetUtils.java | 42 +++++++++--
.../parquet/ParquetAvroRecordExtractorTest.java | 27 ++++---
.../parquet/ParquetNativeRecordExtractorTest.java | 85 ++++++++++++++++++----
.../parquet/ParquetReaderLogicalTypesTest.java | 73 +++++++++++++++++++
.../inputformat/parquet/ParquetTestUtils.java | 14 ++++
10 files changed, 233 insertions(+), 53 deletions(-)
diff --git
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractor.java
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractor.java
index be98b860388..7c2fe2a186d 100644
---
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractor.java
+++
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractor.java
@@ -24,18 +24,18 @@ import org.apache.avro.generic.GenericFixed;
import org.apache.avro.generic.GenericRecord;
import org.apache.pinot.plugin.inputformat.avro.AvroRecordExtractor;
import org.apache.pinot.spi.data.readers.GenericRow;
-import org.apache.pinot.spi.utils.TimestampUtils;
/// The type matrix is inherited from [AvroRecordExtractor], with two
overrides:
/// - INT96 timestamp (which parquet-avro surfaces as `fixed(12)` with `doc =
"INT96 represented as byte[12]"`) →
-/// `Timestamp` (or `Long` epoch nanos when `extractRawTimeValues` is
`true`) via
-/// [ParquetUtils#convertInt96ToEpochNanos].
+/// `Timestamp` (or `Long` epoch nanos when `extractRawTimeValues` is
`true`) via [ParquetUtils].
/// - Top-level unsigned 32-bit integer fields (`INTEGER(32, false)` / legacy
`UINT_32`), which parquet-avro surfaces
/// as a signed Avro `int` → `Long` via [Integer#toUnsignedLong]. The field
names come from the Parquet file schema
/// (see [ParquetUtils#getUnsignedInt32FieldNames]) since the Avro schema
carries no signedness. Unsigned values
/// nested in a struct, list or map are still read as signed ints by this
reader; [ParquetNativeRecordReader]
/// widens them at every level.
+///
+/// Unsigned 64-bit fields preserve their signed `Long` bit pattern, matching
[ParquetNativeRecordReader].
public class ParquetAvroRecordExtractor extends AvroRecordExtractor {
private static final int INT96_BYTE_SIZE = 12;
private static final String INT96_DOC = "INT96 represented as byte[12]";
@@ -66,8 +66,10 @@ public class ParquetAvroRecordExtractor extends
AvroRecordExtractor {
protected Object convertSingleValue(Schema schema, Object value) {
if (schema.getType() == Schema.Type.FIXED && schema.getFixedSize() ==
INT96_BYTE_SIZE
&& INT96_DOC.equals(schema.getDoc())) {
- long nanos = ParquetUtils.convertInt96ToEpochNanos(((GenericFixed)
value).bytes());
- return _extractRawTimeValues ? nanos :
TimestampUtils.fromNanosSinceEpoch(nanos);
+ byte[] bytes = ((GenericFixed) value).bytes();
+ return _extractRawTimeValues
+ ? ParquetUtils.convertInt96ToEpochNanos(bytes)
+ : ParquetUtils.convertInt96ToTimestamp(bytes);
}
return super.convertSingleValue(schema, value);
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordReader.java
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordReader.java
index 18e8796de9c..de5cfa6761c 100644
---
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordReader.java
+++
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordReader.java
@@ -32,9 +32,8 @@ import org.apache.pinot.spi.data.readers.RecordReader;
import org.apache.pinot.spi.data.readers.RecordReaderConfig;
import org.apache.pinot.spi.data.readers.RecordReaderUtils;
-/// Avro Record reader for Parquet file. This reader doesn't read parquet file
with incompatible Avro schemas,
-/// e.g. INT96, DECIMAL. Please use
[org.apache.pinot.plugin.inputformat.parquet.ParquetNativeRecordReader]
-/// instead.
+/// Reads Parquet records through Avro with logical-type conversion, including
INT96 timestamps and DECIMAL values.
+/// Parquet schemas that cannot be represented by Avro require
[ParquetNativeRecordReader].
///
/// For More info on Avro to Parquet schema conversion:
///
[https://javadoc.io/doc/org.apache.parquet/parquet-avro/latest/index.html][ref1]
@@ -72,13 +71,16 @@ public class ParquetAvroRecordReader implements
RecordReader {
@Override
public GenericRow next(GenericRow reuse)
throws IOException {
- // Data parsing: extract current record into GenericRow.
- _recordExtractor.extract(_nextRecord, reuse);
- // Record fetch: read next Parquet Avro record.
try {
- _nextRecord = _parquetReader.read();
- } catch (IOException e) {
- throw new RecordFetchException("Failed to read next Parquet Avro
record", e);
+ // Data parsing: extract current record into GenericRow.
+ _recordExtractor.extract(_nextRecord, reuse);
+ } finally {
+ // Advance even if extraction fails and the caller skips the rejected
record.
+ try {
+ _nextRecord = _parquetReader.read();
+ } catch (IOException e) {
+ throw new RecordFetchException("Failed to read next Parquet Avro
record", e);
+ }
}
return reuse;
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractor.java
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractor.java
index a96f3412b27..5e894355a14 100644
---
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractor.java
+++
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractor.java
@@ -34,7 +34,6 @@ import org.apache.parquet.schema.Type;
import org.apache.pinot.spi.data.readers.BaseRecordExtractor;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.data.readers.RecordExtractorConfig;
-import org.apache.pinot.spi.utils.TimestampUtils;
import org.apache.pinot.spi.utils.UuidUtils;
@@ -45,7 +44,7 @@ import org.apache.pinot.spi.utils.UuidUtils;
/// - `BOOLEAN` → `Boolean`
/// - `INT32` → `Integer`, or `Long` when annotated as unsigned 32-bit
(`INTEGER(32, false)` / legacy `UINT_32`),
/// since values above `Integer.MAX_VALUE` do not fit in a signed `int`
-/// - `INT64` → `Long`
+/// - `INT64` → `Long`, preserving the signed bit pattern for unsigned 64-bit
values
/// - `FLOAT` → `Float`
/// - `DOUBLE` → `Double`
/// - `BINARY` with `STRING` / `ENUM` annotation → `String`
@@ -181,8 +180,10 @@ public class ParquetNativeRecordExtractor extends
BaseRecordExtractor<Group> {
case INT96: {
// INT96 has no Parquet logical-type spec, but its physical encoding
(nanos-of-day + Julian day) is
// nanosecond-precision, so nanos is the natural raw unit.
- long nanos =
ParquetUtils.convertInt96ToEpochNanos(from.getInt96(fieldIndex,
index).getBytes());
- return _extractRawTimeValues ? nanos :
TimestampUtils.fromNanosSinceEpoch(nanos);
+ byte[] bytes = from.getInt96(fieldIndex, index).getBytes();
+ return _extractRawTimeValues
+ ? ParquetUtils.convertInt96ToEpochNanos(bytes)
+ : ParquetUtils.convertInt96ToTimestamp(bytes);
}
case FLOAT:
return from.getFloat(fieldIndex, index);
diff --git
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordReader.java
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordReader.java
index 744f6b0484f..24dfc2a019f 100644
---
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordReader.java
+++
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordReader.java
@@ -122,9 +122,10 @@ public class ParquetNativeRecordReader implements
RecordReader {
} catch (Exception e) {
throw new RecordFetchException("Failed to read next Parquet native
record", e);
}
+ // The physical row is consumed even if extraction fails and the caller
skips it.
+ _currentPageIdx++;
// Data parsing: extract into GenericRow.
_recordExtractor.extract(_nextRecord, reuse);
- _currentPageIdx++;
return reuse;
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetRecordReaderConfig.java
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetRecordReaderConfig.java
index 70f8d152046..fd61510c14b 100644
---
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetRecordReaderConfig.java
+++
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetRecordReaderConfig.java
@@ -28,6 +28,7 @@ import org.apache.pinot.spi.data.readers.RecordReaderConfig;
/// auto-detects via the file's `avro.schema` metadata (Avro reader if
present, native otherwise).
/// - `extractRawTimeValues` — opt out of TIMESTAMP / DATE / TIME conversion
at the extractor boundary,
/// surfacing the raw underlying integer in the column's declared unit
instead of the contract Java type.
+/// Raw INT96 values are signed `long` epoch nanoseconds; timestamps outside
that range are rejected.
/// DECIMAL and UUID always convert. See [ParquetAvroRecordExtractor] /
[ParquetNativeRecordExtractor] for
/// the per-type matrix.
public class ParquetRecordReaderConfig implements RecordReaderConfig {
diff --git
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetUtils.java
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetUtils.java
index 58cc84eca7a..bd28d39004c 100644
---
a/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetUtils.java
+++
b/pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetUtils.java
@@ -56,6 +56,8 @@ public class ParquetUtils {
/// Number of days from the Julian day epoch (4713-01-01 BC) to the Unix day
epoch (1970-01-01).
public static final long JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH = 2440588;
+ private static final long SECONDS_PER_DAY = TimeUnit.DAYS.toSeconds(1);
+ private static final long NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1);
private static final long NANOS_PER_DAY = TimeUnit.DAYS.toNanos(1);
/// Returns a ParquetReader with the given path.
@@ -159,14 +161,38 @@ public class ParquetUtils {
}
}
- /// Converts a 12-byte INT96 timestamp to `Long` epoch nanos. INT96 stores
the timestamp split as bytes
- /// 0..7 (nanos within the day, long, little-endian) and bytes 8..11 (Julian
day number, int, little-endian).
- /// Nanos is INT96's natural unit since this physical encoding carries
nanosecond precision; pair with
- /// [TimestampUtils#fromNanosSinceEpoch] for a [Timestamp].
+ /// Converts a 12-byte INT96 timestamp to [Timestamp] without narrowing it
to signed epoch nanoseconds.
+ /// INT96 stores bytes 0..7 as nanoseconds within the day and bytes 8..11 as
the Julian day, little-endian.
+ /// This also preserves instants outside the approximately 1677–2262 range
of signed epoch nanoseconds.
+ public static Timestamp convertInt96ToTimestamp(byte[] int96Bytes) {
+ ByteBuffer buffer =
ByteBuffer.wrap(int96Bytes).order(ByteOrder.LITTLE_ENDIAN);
+ long days = buffer.getInt(8) - JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH;
+ long nanosOfDay = buffer.getLong(0);
+ long epochSeconds = days * SECONDS_PER_DAY + Math.floorDiv(nanosOfDay,
NANOS_PER_SECOND);
+ Timestamp timestamp = new Timestamp(epochSeconds * 1_000L);
+ timestamp.setNanos((int) Math.floorMod(nanosOfDay, NANOS_PER_SECOND));
+ return timestamp;
+ }
+
+ /// Converts a 12-byte INT96 timestamp to signed `Long` epoch nanoseconds.
+ ///
+ /// @throws IllegalArgumentException if the exact value does not fit in a
signed `long`; logical extraction
+ /// should use [#convertInt96ToTimestamp(byte\[\])] instead
public static long convertInt96ToEpochNanos(byte[] int96Bytes) {
- ByteBuffer buf =
ByteBuffer.wrap(int96Bytes).order(ByteOrder.LITTLE_ENDIAN);
- long days = buf.getInt(8) - JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH;
- long nanosOfDay = buf.getLong(0);
- return days * NANOS_PER_DAY + nanosOfDay;
+ ByteBuffer buffer =
ByteBuffer.wrap(int96Bytes).order(ByteOrder.LITTLE_ENDIAN);
+ long days = buffer.getInt(8) - JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH;
+ long nanosOfDay = buffer.getLong(0);
+ // Evaluate the complete sum in 128 bits. At Long.MIN_VALUE, the day
product alone underflows, but adding
+ // the positive day fraction brings the sum back into range, so
multiplyExact followed by addExact is wrong.
+ long product = days * NANOS_PER_DAY;
+ long nanos = product + nanosOfDay;
+ // Unsigned low-word overflow produces a carry; nanosOfDay >> 63
sign-extends the addend into the high word.
+ long carry = Long.compareUnsigned(nanos, product) < 0 ? 1 : 0;
+ long high = Math.multiplyHigh(days, NANOS_PER_DAY) + (nanosOfDay >> 63) +
carry;
+ if (high != (nanos >> 63)) {
+ throw new IllegalArgumentException("INT96 timestamp " +
convertInt96ToTimestamp(int96Bytes).toInstant()
+ + " is outside the signed epoch-nanosecond interval and cannot be
returned as raw epoch nanoseconds");
+ }
+ return nanos;
}
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractorTest.java
b/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractorTest.java
index e4adca70a21..39018954223 100644
---
a/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractorTest.java
+++
b/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractorTest.java
@@ -18,9 +18,8 @@
*/
package org.apache.pinot.plugin.inputformat.parquet;
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
import java.sql.Timestamp;
+import java.time.Instant;
import java.util.List;
import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
@@ -30,7 +29,9 @@ import
org.apache.pinot.plugin.inputformat.avro.AvroRecordExtractorConfig;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.testng.annotations.Test;
+import static
org.apache.pinot.plugin.inputformat.parquet.ParquetTestUtils.int96Bytes;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.expectThrows;
/// Tests [ParquetAvroRecordExtractor]'s INT96 override — see its class
Javadoc. All other Avro behavior is
@@ -78,6 +79,18 @@ public class ParquetAvroRecordExtractorTest {
assertEquals(extract(record, null), new Timestamp(epochMillis));
}
+ @Test
+ public void testInt96BeyondEpochNanosRange() {
+ for (String text : new String[]{"1600-01-01T00:00:00.123456789Z",
"2300-01-01T00:00:00.123456789Z"}) {
+ Instant instant = Instant.parse(text);
+ byte[] bytes = int96Bytes(instant);
+ Schema fixed = Schema.createFixed("Int96", "INT96 represented as
byte[12]", null, 12);
+ GenericRecord record = singleField(fixed, new GenericData.Fixed(fixed,
bytes));
+ assertEquals(extract(record), Timestamp.from(instant), text);
+ expectThrows(IllegalArgumentException.class, () -> extract(record,
rawConfig()));
+ }
+ }
+
// === Helpers ===
private static Object extract(GenericRecord record,
AvroRecordExtractorConfig config) {
@@ -112,16 +125,8 @@ public class ParquetAvroRecordExtractorTest {
return Schema.createUnion(List.of(Schema.create(Schema.Type.NULL),
fixedSchema));
}
- /// Encode `epochMillis` as an INT96 byte[12]: bytes 0..7 = nanos-within-day
(long), bytes 8..11 = Julian
- /// day number (int).
private static GenericData.Fixed int96Fixed(long epochMillis) {
- long millisPerDay = 86_400_000L;
- long dayNumber = epochMillis / millisPerDay;
- long nanosWithinDay = (epochMillis - dayNumber * millisPerDay) *
1_000_000L;
- int julianDay = (int) (dayNumber +
ParquetUtils.JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH);
- byte[] int96 = ByteBuffer.allocate(12).order(ByteOrder.LITTLE_ENDIAN)
- .putLong(nanosWithinDay).putInt(julianDay).array();
Schema fixedSchema = Schema.createFixed("Int96", "INT96 represented as
byte[12]", null, 12);
- return new GenericData.Fixed(fixedSchema, int96);
+ return new GenericData.Fixed(fixedSchema,
int96Bytes(Instant.ofEpochMilli(epochMillis)));
}
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractorTest.java
b/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractorTest.java
index 0c5bcf7ae7c..0cb294e2c16 100644
---
a/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractorTest.java
+++
b/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractorTest.java
@@ -19,9 +19,8 @@
package org.apache.pinot.plugin.inputformat.parquet;
import java.math.BigDecimal;
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
import java.sql.Timestamp;
+import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalTime;
import java.util.Map;
@@ -40,8 +39,10 @@ import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.utils.UuidUtils;
import org.testng.annotations.Test;
+import static
org.apache.pinot.plugin.inputformat.parquet.ParquetTestUtils.int96Bytes;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.expectThrows;
/// Tests [ParquetNativeRecordExtractor] — see its class Javadoc for the
Parquet primitive + logical-type
@@ -70,11 +71,51 @@ public class ParquetNativeRecordExtractorTest {
assertEquals(extract(schema, g -> g.add(0, 1_588_469_340_000L)),
1_588_469_340_000L);
}
+ @Test
+ public void testUnsignedInt64PreservesSignedLongBits() {
+ for (String annotation : new String[]{"UINT_64", "INTEGER(64,false)"}) {
+ MessageType schema = MessageTypeParser.parseMessageType(
+ "message R { required int64 col (" + annotation + "); }");
+ for (long value : new long[]{0L, 1L, Long.MAX_VALUE, Long.MIN_VALUE,
-1L}) {
+ assertEquals(extract(schema, g -> g.add(0, value)), value, annotation);
+ assertEquals(extract(schema, g -> g.add(0, value), rawConfig()),
value, annotation);
+ }
+ }
+ MessageType signed = MessageTypeParser.parseMessageType("message R {
required int64 col; }");
+ assertEquals(extract(signed, g -> g.add(0, -1L)), -1L);
+ assertEquals(extract(signed, g -> g.add(0, Long.MIN_VALUE)),
Long.MIN_VALUE);
+ }
+
+ @Test
+ public void testNestedUnsignedInt64PreservesSignedLongBits() {
+ MessageType list = MessageTypeParser.parseMessageType("message R {
required group col (LIST) {"
+ + " repeated group list { required int64 element (UINT_64); } } }");
+ MessageType struct = MessageTypeParser.parseMessageType(
+ "message R { required group col { required int64 value (UINT_64); }
}");
+ MessageType map = MessageTypeParser.parseMessageType("message R { required
group col (MAP) {"
+ + " repeated group key_value { required int64 key (UINT_64); optional
int64 value (UINT_64); } } }");
+ for (long value : new long[]{Long.MIN_VALUE, -1L}) {
+ assertEquals(extract(list, g -> g.addGroup(0).addGroup(0).add(0,
value)), new Object[]{value});
+ assertEquals(extract(struct, g -> g.addGroup(0).add(0, value)),
Map.of("value", value));
+ assertEquals(extract(map, g -> {
+ Group entry = g.addGroup(0).addGroup(0);
+ entry.add(0, value);
+ entry.add(1, 0L);
+ }), Map.of(Long.toString(value), 0L));
+ assertEquals(extract(map, g -> {
+ Group entry = g.addGroup(0).addGroup(0);
+ entry.add(0, 0L);
+ entry.add(1, value);
+ }), Map.of("0", value));
+ }
+ }
+
@Test
public void testInt96ExtractedAsTimestamp() {
long epochMillis = 1_649_924_302_123L;
MessageType schema = MessageTypeParser.parseMessageType("message R {
required int96 col; }");
- assertEquals(extract(schema, g -> g.add(0,
Binary.fromConstantByteArray(int96Bytes(epochMillis)))),
+ assertEquals(extract(schema,
+ g -> g.add(0,
Binary.fromConstantByteArray(int96Bytes(Instant.ofEpochMilli(epochMillis))))),
new Timestamp(epochMillis));
}
@@ -270,10 +311,35 @@ public class ParquetNativeRecordExtractorTest {
// nanosecond precision, so nanos is the natural raw unit.
long epochMillis = 1_649_924_302_123L;
MessageType schema = MessageTypeParser.parseMessageType("message R {
required int96 col; }");
- assertEquals(extract(schema, g -> g.add(0,
Binary.fromConstantByteArray(int96Bytes(epochMillis))), rawConfig()),
+ assertEquals(extract(schema,
+ g -> g.add(0,
Binary.fromConstantByteArray(int96Bytes(Instant.ofEpochMilli(epochMillis)))),
rawConfig()),
epochMillis * 1_000_000L);
}
+ @Test
+ public void testInt96RangeAndNanosecondBoundaries() {
+ MessageType schema = MessageTypeParser.parseMessageType("message R {
required int96 col; }");
+ for (String text : new String[]{"1970-01-01T00:00:00Z",
"1969-12-31T23:59:59.999999999Z",
+ "1677-09-21T00:12:43.145224192Z", "2262-04-11T23:47:16.854775807Z",
+ "1677-09-21T00:12:43.145224191Z", "2262-04-11T23:47:16.854775808Z",
+ "1600-01-01T00:00:00.123456789Z", "2300-01-01T00:00:00.123456789Z"}) {
+ Instant instant = Instant.parse(text);
+ byte[] bytes = int96Bytes(instant);
+ assertEquals(extract(schema, g -> g.add(0,
Binary.fromConstantByteArray(bytes))), Timestamp.from(instant), text);
+ BigDecimal nanos =
BigDecimal.valueOf(instant.getEpochSecond()).movePointRight(9)
+ .add(BigDecimal.valueOf(instant.getNano()));
+ if (nanos.compareTo(BigDecimal.valueOf(Long.MIN_VALUE)) >= 0
+ && nanos.compareTo(BigDecimal.valueOf(Long.MAX_VALUE)) <= 0) {
+ assertEquals(extract(schema, g -> g.add(0,
Binary.fromConstantByteArray(bytes)), rawConfig()),
+ nanos.longValueExact(), text);
+ } else {
+ IllegalArgumentException exception =
expectThrows(IllegalArgumentException.class,
+ () -> extract(schema, g -> g.add(0,
Binary.fromConstantByteArray(bytes)), rawConfig()));
+ assertEquals(exception.getMessage().contains(instant.toString()),
true, exception.getMessage());
+ }
+ }
+ }
+
// === LIST — standard 3-level wrapper and legacy forms ===
@Test
@@ -449,15 +515,4 @@ public class ParquetNativeRecordExtractorTest {
config.setExtractRawTimeValues(true);
return config;
}
-
- /// Encodes `epochMillis` as an INT96 byte[12]: bytes 0..7 =
nanos-within-day (long, little-endian),
- /// bytes 8..11 = Julian day number (int, little-endian).
- private static byte[] int96Bytes(long epochMillis) {
- long millisPerDay = 86_400_000L;
- long dayNumber = epochMillis / millisPerDay;
- long nanosWithinDay = (epochMillis - dayNumber * millisPerDay) *
1_000_000L;
- int julianDay = (int) (dayNumber +
ParquetUtils.JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH);
- return ByteBuffer.allocate(12).order(ByteOrder.LITTLE_ENDIAN)
- .putLong(nanosWithinDay).putInt(julianDay).array();
- }
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetReaderLogicalTypesTest.java
b/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetReaderLogicalTypesTest.java
index 89b12bf5d4a..c7666a16f95 100644
---
a/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetReaderLogicalTypesTest.java
+++
b/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetReaderLogicalTypesTest.java
@@ -41,6 +41,9 @@ import org.apache.pinot.spi.data.readers.RecordReader;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
/// TODO: add test for other logical types
@@ -177,6 +180,76 @@ public class ParquetReaderLogicalTypesTest {
}
}
+ @Test
+ public void testReadersPreserveUnsignedInt64Bits()
+ throws IOException {
+ MessageType schema = MessageTypeParser.parseMessageType(
+ "message m { required int64 value (INTEGER(64, false)); }");
+ File dataFile = new File(FileUtils.getTempDirectory(), "unsigned-int64-" +
System.nanoTime() + ".parquet");
+ try {
+ try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new
Path(dataFile.getAbsolutePath()))
+ .withType(schema).build()) {
+ SimpleGroupFactory groups = new SimpleGroupFactory(schema);
+ for (long value : new long[]{-1L, 7L, Long.MIN_VALUE}) {
+ writer.write(groups.newGroup().append("value", value));
+ }
+ }
+ for (RecordReader reader : List.of(new ParquetNativeRecordReader(), new
ParquetAvroRecordReader())) {
+ try (reader) {
+ reader.init(dataFile, null, null);
+ for (long value : new long[]{-1L, 7L, Long.MIN_VALUE}) {
+ assertTrue(reader.hasNext());
+ assertEquals(reader.next(new GenericRow()).getValue("value"),
value);
+ }
+ assertFalse(reader.hasNext());
+ }
+ }
+ } finally {
+ FileUtils.deleteQuietly(dataFile);
+ }
+ }
+
+ @Test
+ public void testReadersAdvancePastRejectedInt96()
+ throws IOException {
+ MessageType schema = MessageTypeParser.parseMessageType("message m {
required int96 value; }");
+ File dataFile = new File(FileUtils.getTempDirectory(), "rejected-int96-" +
System.nanoTime() + ".parquet");
+ try {
+ try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new
Path(dataFile.getAbsolutePath()))
+ .withType(schema).build()) {
+ SimpleGroupFactory groups = new SimpleGroupFactory(schema);
+ // Year 1600 cannot fit signed Long epoch nanos; the valid middle row
is followed by another rejection.
+ writer.write(groups.newGroup().append("value", new NanoTime(2305448,
0L)));
+ writer.write(groups.newGroup().append("value", new NanoTime(2440588,
7L)));
+ writer.write(groups.newGroup().append("value", new NanoTime(2305448,
0L)));
+ }
+ for (boolean useAvro : new boolean[]{false, true}) {
+ ParquetRecordReaderConfig config = new ParquetRecordReaderConfig();
+ config.setUseParquetAvroRecordReader(useAvro);
+ config.setUseParquetNativeRecordReader(!useAvro);
+ config.setExtractRawTimeValues(true);
+ try (ParquetRecordReader reader = new ParquetRecordReader()) {
+ reader.init(dataFile, null, config);
+ assertEquals(reader.useAvroParquetRecordReader(), useAvro);
+ for (int iteration = 0; iteration < 2; iteration++) {
+ if (iteration > 0) {
+ reader.rewind();
+ }
+ assertTrue(reader.hasNext());
+ expectThrows(IllegalArgumentException.class, () -> reader.next(new
GenericRow()));
+ assertTrue(reader.hasNext());
+ assertEquals(reader.next(new GenericRow()).getValue("value"), 7L);
+ assertTrue(reader.hasNext());
+ expectThrows(IllegalArgumentException.class, () -> reader.next(new
GenericRow()));
+ assertFalse(reader.hasNext(), "Extraction failures must not retry
the rejected record");
+ }
+ }
+ }
+ } finally {
+ FileUtils.deleteQuietly(dataFile);
+ }
+ }
+
/// Write the given row object to a random parquet file and return the path
of the file
private String writeToFile(String schemaStr, Map<String, Object> rowObject)
throws IOException {
diff --git
a/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetTestUtils.java
b/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetTestUtils.java
index 97bb404c33f..36ac415edda 100644
---
a/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetTestUtils.java
+++
b/pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetTestUtils.java
@@ -19,6 +19,10 @@
package org.apache.pinot.plugin.inputformat.parquet;
import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.time.Instant;
+import java.util.concurrent.TimeUnit;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.fs.Path;
@@ -33,6 +37,16 @@ public class ParquetTestUtils {
private ParquetTestUtils() {
}
+ /// Encodes an instant as little-endian INT96 with a nonnegative
nanosecond-of-day and Julian day number.
+ static byte[] int96Bytes(Instant instant) {
+ long secondsPerDay = TimeUnit.DAYS.toSeconds(1);
+ long day = Math.floorDiv(instant.getEpochSecond(), secondsPerDay);
+ long nanosOfDay = Math.floorMod(instant.getEpochSecond(), secondsPerDay) *
TimeUnit.SECONDS.toNanos(1)
+ + instant.getNano();
+ return
ByteBuffer.allocate(12).order(ByteOrder.LITTLE_ENDIAN).putLong(nanosOfDay)
+ .putInt(Math.toIntExact(day +
ParquetUtils.JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH)).array();
+ }
+
/// Returns a ParquetWriter with the given path and schema.
public static ParquetWriter<GenericRecord> getParquetAvroWriter(Path path,
Schema schema)
throws IOException {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
