xiangfu0 commented on code in PR #20060:
URL: https://github.com/apache/pinot/pull/20060#discussion_r4396248989
##########
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:
Consolidated in b79225748a. ParquetTestUtils.int96Bytes(Instant) now uses
floorDiv/floorMod for canonical day and nanosecond-of-day encoding; both
extractor tests, including the older millisecond helpers, reuse it. All 51
focused Maven cases and applicable formatting, license and Checkstyle checks
pass; independent code review found no issues.
##########
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:
Added in b79225748a: the file now ends with another rejected INT96 record,
and both native and forced-Avro paths assert hasNext() is false immediately
after that exception. Both passes retain rewind coverage. All 51 focused Maven
cases pass with zero skipped tests.
##########
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:
Tracked separately in https://github.com/apache/pinot/issues/20065 with the
injected-read reproduction and acceptance checks for fetch-error
classification, cause retention and bounded failure handling. Verified the same
narrow IOException catch exists at the base revision, so the follow-up
preserves your requested separation from this PR.
--
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]
