Jackie-Jiang commented on code in PR #20060:
URL: https://github.com/apache/pinot/pull/20060#discussion_r4396162585
##########
pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetReaderLogicalTypesTest.java:
##########
@@ -177,6 +180,73 @@ public void testNestedUnsignedInt32TypeWithNativeReader()
}
}
+ @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 be represented as signed Long epoch nanos; the
next row is in range.
+ writer.write(groups.newGroup().append("value", new NanoTime(2305448,
0L)));
+ writer.write(groups.newGroup().append("value", new NanoTime(2440588,
7L)));
+ }
+ 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);
+ assertFalse(reader.hasNext(), "Extraction failures must not retry
the rejected record");
Review Comment:
Nonblocking cleanup: test rejection directly at EOF
This regression covers rejected → valid → EOF. Add a trailing rejected INT96
row and assert `hasNext()` becomes false immediately after its exception,
retaining the rewind coverage. That directly protects advancement when there is
no subsequent successful extraction.
##########
pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordReader.java:
##########
@@ -72,13 +72,16 @@ public boolean hasNext() {
@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);
Review Comment:
Pre-existing follow-up: classify unchecked fetch failures consistently
The prefetch catches only `IOException`, but parquet-hadoop 1.18.1 can
propagate unchecked `ParquetDecodingException` from
`InternalParquetRecordReader.nextKeyValue()` through `ParquetReader.read()`.
Those failures escape as parse errors in `SegmentMapper`, bypassing its
configured consecutive-fetch-failure threshold.
Wrap exceptions specifically around `_parquetReader.read()` as
`RecordFetchException`, consistently with the native reader, and add an
injected unchecked-read-failure test. The same narrow catch exists in the base
revision, so this is a separate follow-up and does not block approval of this
PR.
##########
pinot-plugins/pinot-input-format/pinot-parquet/src/test/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractorTest.java:
##########
@@ -274,6 +315,33 @@ public void testInt96ExtractedAsRawLongEpochNanosWhenRaw()
{
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);
+ long day = Math.floorDiv(instant.getEpochSecond(), 86_400L);
+ byte[] bytes = ByteBuffer.allocate(12).order(ByteOrder.LITTLE_ENDIAN)
+ .putLong(Math.floorMod(instant.getEpochSecond(), 86_400L) *
1_000_000_000L + instant.getNano())
+ .putInt(Math.toIntExact(day +
ParquetUtils.JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH)).array();
Review Comment:
Nonblocking cleanup: consolidate INT96 test encoding
The native and Avro tests duplicate this `Instant` → little-endian bytes
calculation, alongside older millisecond-based helpers. Add one
`ParquetTestUtils.int96Bytes(Instant)` helper using floor division, then reuse
it from both extractor tests. This removes duplicated date arithmetic and gives
negative dates the same canonical day/nanoseconds-of-day encoding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
