Jackie-Jiang commented on code in PR #20060:
URL: https://github.com/apache/pinot/pull/20060#discussion_r4392567101
##########
pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractor.java:
##########
@@ -66,8 +68,9 @@ public GenericRow extract(GenericRecord from, GenericRow to) {
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)
Review Comment:
[P1] Advance the Avro reader after INT96 rejection
With the Avro reader selected, `extractRawTimeValues=true`, and
`continueOnError=true`, an out-of-range INT96 value (such as year 1600) causes
ingestion to loop indefinitely. This conversion now throws
`IllegalArgumentException`, but `ParquetAvroRecordReader.next()` extracts
before advancing `_nextRecord`, so every subsequent call retries the same
rejected row. Both `RecordReaderSegmentCreationDataSource.gatherStats()` and
`SegmentMapper` catch this exception and continue; the latter's
consecutive-fetch-failure threshold does not apply to parse errors.
Make the Avro reader advance after extraction failures, and add a
file-backed rejected-row → valid-row → EOF regression test.
##########
pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractor.java:
##########
@@ -24,18 +24,20 @@
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].
+/// [ParquetUtils].
Review Comment:
[P3] Keep the Javadoc link on the preceding line
Merge `[ParquetUtils]` into the preceding line after `via`. The combined
line is 99 characters, within the 120-character limit:
```suggestion
/// `Timestamp` (or `Long` epoch nanos when `extractRawTimeValues` is
`true`) via [ParquetUtils].
```
##########
pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetAvroRecordExtractor.java:
##########
@@ -66,8 +68,9 @@ public GenericRow extract(GenericRecord from, GenericRow to) {
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);
Review Comment:
[P3] Put both ternary branches on separate lines
For a multiline ternary, put the `?` and `:` branches on separate
continuation lines:
```suggestion
return _extractRawTimeValues
? ParquetUtils.convertInt96ToEpochNanos(bytes)
: ParquetUtils.convertInt96ToTimestamp(bytes);
```
##########
pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractor.java:
##########
@@ -181,8 +185,9 @@ private Object extractValue(Group from, int fieldIndex,
Type fieldType, int inde
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);
Review Comment:
[P3] Apply the same ternary formatting to the native extractor
Move the `?` branch onto its own continuation line, aligned with `:`:
```suggestion
return _extractRawTimeValues
? ParquetUtils.convertInt96ToEpochNanos(bytes)
: ParquetUtils.convertInt96ToTimestamp(bytes);
```
--
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]
