Copilot commented on code in PR #18378:
URL: https://github.com/apache/pinot/pull/18378#discussion_r3165372708
##########
pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordExtractor.java:
##########
@@ -169,10 +212,37 @@ private Object extractValue(Group from, int fieldIndex,
Type fieldType, int inde
return null;
}
- public static long convertInt96ToLong(byte[] int96Bytes) {
+ /// Converts a 12-byte INT96 timestamp to [Timestamp]. Bytes 0..7 are nanos
within the day (long), bytes
+ /// 8..11 are the Julian day number (int). Sub-millisecond nanos are
preserved via [Timestamp#setNanos].
+ public static Timestamp convertInt96ToTimestamp(byte[] int96Bytes) {
ByteBuffer buf =
ByteBuffer.wrap(int96Bytes).order(ByteOrder.LITTLE_ENDIAN);
- return (buf.getInt(8) - JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH) *
DateTimeConstants.MILLIS_PER_DAY
- + buf.getLong(0) / NANOS_PER_MILLISECOND;
+ long days = buf.getInt(8) - JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH;
+ long nanosOfDay = buf.getLong(0);
+ long epochMillis = days * DateTimeConstants.MILLIS_PER_DAY + nanosOfDay /
NANOS_PER_MILLISECOND;
+ Timestamp ts = new Timestamp(epochMillis);
+ ts.setNanos((int) (nanosOfDay % 1_000_000_000L));
+ return ts;
+ }
+
+ /// Builds a [Timestamp] from `value` interpreted in `unit` (millis / micros
/ nanos). Sub-millisecond
+ /// precision is preserved via [Timestamp#setNanos].
+ private static Timestamp timestampFromLong(long value,
LogicalTypeAnnotation.TimeUnit unit) {
+ switch (unit) {
+ case MILLIS:
+ return new Timestamp(value);
+ case MICROS: {
+ Timestamp ts = new Timestamp(value / 1_000L);
+ ts.setNanos((int) Math.floorMod(value, 1_000_000L) * 1_000);
+ return ts;
+ }
+ case NANOS: {
+ Timestamp ts = new Timestamp(value / 1_000_000L);
+ ts.setNanos((int) Math.floorMod(value, 1_000_000_000L));
+ return ts;
+ }
+ default:
+ throw new IllegalArgumentException("Unsupported timestamp unit: " +
unit);
+ }
Review Comment:
timestampFromLong() mixes truncating division (e.g. value / 1_000L) with
Math.floorMod for the sub-second part. For negative (pre-epoch) timestamps this
produces an incorrect millis+nanos combination. Use Math.floorDiv to compute
the base seconds/millis consistently with floorMod, or derive seconds+nanos via
floorDiv/floorMod and then build the Timestamp.
##########
pinot-plugins/pinot-input-format/pinot-orc/src/main/java/org/apache/pinot/plugin/inputformat/orc/ORCRecordExtractor.java:
##########
@@ -0,0 +1,219 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.inputformat.orc;
+
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.TypeDescription;
+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 static java.nio.charset.StandardCharsets.UTF_8;
+
+
+/// Extracts a single ORC row into a [GenericRow]. Input is an
[ORCRecordExtractor.Record] handle wrapping
+/// a [VectorizedRowBatch] + schema + row index. Dispatch happens in
[#extractValue] (complex types —
+/// `LIST` / `MAP` / `STRUCT`) and [#extractSingleValue] (primitives); ORC
values never flow through
+/// `convertSingleValue`, so widening / `Temporal` handling is done locally
here.
+///
+/// **ORC schema category → Java output type:**
+/// - `BOOLEAN` → `Boolean`
+/// - `BYTE` / `SHORT` / `INT` → `Integer` (small ints widen to `Integer`)
+/// - `LONG` → `Long`
+/// - `FLOAT` → `Float`
+/// - `DOUBLE` → `Double`
+/// - `DECIMAL` → `BigDecimal`
+/// - `STRING` / `VARCHAR` / `CHAR` → `String`
+/// - `BINARY` → `byte[]`
+/// - `DATE` → [LocalDate] via [LocalDate#ofEpochDay] (TZ-independent,
calendar-date semantics)
+/// - `TIMESTAMP` / `TIMESTAMP_INSTANT` → [Timestamp] preserving full
sub-second nanos from [TimestampColumnVector]
+/// - `LIST<X>` → `Object[]` (null elements preserved; empty list surfaces as
empty `Object[]`)
+/// - `MAP<K, V>` → `Map<Object, Object>`
+/// - `STRUCT<...>` → `Map<String, Object>`
+/// - any nullable column with `isNull[rowId]` set → `null`
+public class ORCRecordExtractor extends
BaseRecordExtractor<ORCRecordExtractor.Record> {
+
+ /// One ORC row's worth of state. The reader allocates a single instance and
calls [#set] on each row
+ /// advance — no per-row allocation.
+ public static final class Record {
+ VectorizedRowBatch _batch;
+ TypeDescription _schema;
+ int _rowId;
+
+ public void set(VectorizedRowBatch batch, TypeDescription schema, int
rowId) {
+ _batch = batch;
+ _schema = schema;
+ _rowId = rowId;
+ }
+ }
+
+ private Set<String> _fields;
+ private boolean _extractAll;
+
+ @Override
+ public void init(@Nullable Set<String> fields, @Nullable
RecordExtractorConfig recordExtractorConfig) {
+ if (fields == null || fields.isEmpty()) {
+ _extractAll = true;
+ _fields = Set.of();
+ } else {
+ _extractAll = false;
+ _fields = Set.copyOf(fields);
+ }
+ }
+
+ @Override
+ public GenericRow extract(Record from, GenericRow to) {
+ List<String> fieldNames = from._schema.getFieldNames();
+ List<TypeDescription> fieldTypes = from._schema.getChildren();
+ int numFields = fieldNames.size();
+ for (int i = 0; i < numFields; i++) {
+ String fieldName = fieldNames.get(i);
+ if (_extractAll || _fields.contains(fieldName)) {
+ to.putValue(fieldName, extractValue(fieldName, from._batch.cols[i],
fieldTypes.get(i), from._rowId));
+ }
+ }
+ return to;
+ }
+
+ /// Extracts the value at `rowId` from `columnVector`, dispatching by
`fieldType.getCategory()`. Recurses
+ /// into [#extractValue] for nested complex types and falls through to
[#extractSingleValue] for primitives.
+ @Nullable
+ private static Object extractValue(String field, ColumnVector columnVector,
TypeDescription fieldType, int rowId) {
+ if (columnVector.isRepeating) {
+ rowId = 0;
+ }
+ if (!columnVector.noNulls && columnVector.isNull[rowId]) {
+ return null;
+ }
+ TypeDescription.Category category = fieldType.getCategory();
+ switch (category) {
+ case LIST: {
+ TypeDescription childType = fieldType.getChildren().get(0);
+ ListColumnVector listColumnVector = (ListColumnVector) columnVector;
+ int offset = (int) listColumnVector.offsets[rowId];
+ int length = (int) listColumnVector.lengths[rowId];
+ Object[] values = new Object[length];
+ for (int j = 0; j < length; j++) {
+ values[j] = extractValue(field, listColumnVector.child, childType,
offset + j);
+ }
+ return values;
+ }
+ case MAP: {
+ List<TypeDescription> children = fieldType.getChildren();
+ TypeDescription.Category keyCategory = children.get(0).getCategory();
+ TypeDescription valueType = children.get(1);
+ MapColumnVector mapColumnVector = (MapColumnVector) columnVector;
+ int offset = (int) mapColumnVector.offsets[rowId];
+ int length = (int) mapColumnVector.lengths[rowId];
+ Map<Object, Object> map = new HashMap<>();
+ for (int j = 0; j < length; j++) {
+ int childRowId = offset + j;
+ Object key = extractSingleValue(field, mapColumnVector.keys,
childRowId, keyCategory);
Review Comment:
ORC map key extraction calls extractSingleValue(mapColumnVector.keys,
childRowId, ...) but extractSingleValue does not handle
ColumnVector.isRepeating. If the keys vector is marked repeating, this will
read the wrong row index. Consider handling isRepeating in extractSingleValue
(like extractValue does) or adjusting rowId before reading map keys.
--
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]
