xiangfu0 commented on code in PR #19073:
URL: https://github.com/apache/pinot/pull/19073#discussion_r3661177218
##########
pinot-core/src/main/java/org/apache/pinot/core/util/SegmentProcessorAvroUtils.java:
##########
@@ -51,43 +54,99 @@ public static GenericData.Record
convertGenericRowToAvroRecord(GenericRow generi
return convertGenericRowToAvroRecord(genericRow, reusableRecord,
genericRow.getFieldToValueMap().keySet());
}
- /// Convert a GenericRow to an avro GenericRecord
+ /// Convert a GenericRow to an avro GenericRecord.
+ ///
+ /// Values arrive in Pinot's internal (stored) representation and are
coordinated with the Avro field type produced
+ /// by `AvroSchemaUtil.toAvroSchema`: whatever a registered logical-type
[Conversion] can handle is left untouched
+ /// for the writer, and only the two cases Avro cannot resolve on its own
are fixed up here (see
+ /// [#convertValue(Schema, Object)]).
public static GenericData.Record convertGenericRowToAvroRecord(GenericRow
genericRow,
GenericData.Record reusableRecord, Set<String> fields) {
Schema avroSchema = reusableRecord.getSchema();
for (String field : fields) {
Object value = genericRow.getValue(field);
- if (value instanceof Object[]) {
- // Array elements are written as-is. For MV UUID
(array<string{logicalType:uuid}>) the elements are the raw
- // 16-byte values; the uuid Conversion registered on the writer's data
model (getAvroDataModel) renders each
- // element to its canonical string at write time.
- reusableRecord.put(field, Arrays.asList((Object[]) value));
- } else if (value instanceof byte[]) {
- // A byte[] bound for a plain BYTES field must be wrapped as
ByteBuffer (GenericDatumWriter requires it for the
- // bytes type). A byte[] bound for a UUID field
(string{logicalType:uuid}) is left raw so the uuid Conversion
- // registered on the writer's data model (getAvroDataModel) renders it
to a canonical string at write time.
- Schema.Field avroField = avroSchema.getField(field);
- if (avroField != null && avroField.schema().getType() ==
Schema.Type.BYTES) {
- reusableRecord.put(field, ByteBuffer.wrap((byte[]) value));
- } else {
- reusableRecord.put(field, value);
- }
- } else {
+ Schema.Field avroField = avroSchema.getField(field);
+ if (avroField == null) {
+ // Let Avro raise its own "Not a valid schema field" error for a
column missing from the Avro schema.
reusableRecord.put(field, value);
+ } else {
+ reusableRecord.put(avroField.pos(), convertValue(avroField.schema(),
value));
}
}
return reusableRecord;
}
- /// Shared Avro data model with [UuidConversion] registered. Populated once
at class initialization and never
- /// mutated afterward (effectively immutable), so it is safe to share across
writers.
+ /// Adapts a Pinot value to the representation the Avro writer expects for
the given field schema, recursing into
+ /// array elements for multi-value columns.
+ @Nullable
+ private static Object convertValue(Schema fieldSchema, @Nullable Object
value) {
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof Object[]) {
+ Object[] values = (Object[]) value;
+ Schema elementSchema =
+ fieldSchema.getType() == Schema.Type.ARRAY ?
fieldSchema.getElementType() : fieldSchema;
+ // Only BOOLEAN (stored int -> Boolean) and BYTES (byte[] -> ByteBuffer)
element schemas can require a
+ // per-element transform. Every other MV element type is written as-is —
INT/LONG/FLOAT/DOUBLE/STRING directly,
+ // and UUID (string element, raw byte[] rendered by the registered
Conversion) — so hand the writer a zero-copy
+ // view over the existing array; allocating and copying a fresh list per
row would be pure overhead on the
+ // segment-write hot path. (BIG_DECIMAL has a BYTES element schema and
so takes the copy path below, but its
+ // BigDecimal values still pass through convertSingleValue unchanged for
the registered Conversion.)
+ Schema.Type elementType = elementSchema.getType();
+ if (elementType != Schema.Type.BOOLEAN && elementType !=
Schema.Type.BYTES) {
+ return Arrays.asList(values);
+ }
+ List<Object> converted = new ArrayList<>(values.length);
Review Comment:
Addressed in dc7f174598: MV `bytes{logicalType:big-decimal}` now uses the
zero-copy array-view path; only BOOLEAN and plain BYTES convert per element. I
also added a regression test that mutates the backing array and verifies the
record retains that view.
##########
pinot-core/src/main/java/org/apache/pinot/core/util/SegmentProcessorAvroUtils.java:
##########
@@ -51,43 +54,99 @@ public static GenericData.Record
convertGenericRowToAvroRecord(GenericRow generi
return convertGenericRowToAvroRecord(genericRow, reusableRecord,
genericRow.getFieldToValueMap().keySet());
}
- /// Convert a GenericRow to an avro GenericRecord
+ /// Convert a GenericRow to an avro GenericRecord.
+ ///
+ /// Values arrive in Pinot's internal (stored) representation and are
coordinated with the Avro field type produced
+ /// by `AvroSchemaUtil.toAvroSchema`: whatever a registered logical-type
[Conversion] can handle is left untouched
+ /// for the writer, and only the two cases Avro cannot resolve on its own
are fixed up here (see
+ /// [#convertValue(Schema, Object)]).
public static GenericData.Record convertGenericRowToAvroRecord(GenericRow
genericRow,
GenericData.Record reusableRecord, Set<String> fields) {
Schema avroSchema = reusableRecord.getSchema();
for (String field : fields) {
Object value = genericRow.getValue(field);
- if (value instanceof Object[]) {
- // Array elements are written as-is. For MV UUID
(array<string{logicalType:uuid}>) the elements are the raw
- // 16-byte values; the uuid Conversion registered on the writer's data
model (getAvroDataModel) renders each
- // element to its canonical string at write time.
- reusableRecord.put(field, Arrays.asList((Object[]) value));
- } else if (value instanceof byte[]) {
- // A byte[] bound for a plain BYTES field must be wrapped as
ByteBuffer (GenericDatumWriter requires it for the
- // bytes type). A byte[] bound for a UUID field
(string{logicalType:uuid}) is left raw so the uuid Conversion
- // registered on the writer's data model (getAvroDataModel) renders it
to a canonical string at write time.
- Schema.Field avroField = avroSchema.getField(field);
- if (avroField != null && avroField.schema().getType() ==
Schema.Type.BYTES) {
- reusableRecord.put(field, ByteBuffer.wrap((byte[]) value));
- } else {
- reusableRecord.put(field, value);
- }
- } else {
+ Schema.Field avroField = avroSchema.getField(field);
+ if (avroField == null) {
+ // Let Avro raise its own "Not a valid schema field" error for a
column missing from the Avro schema.
reusableRecord.put(field, value);
+ } else {
+ reusableRecord.put(avroField.pos(), convertValue(avroField.schema(),
value));
}
}
return reusableRecord;
}
- /// Shared Avro data model with [UuidConversion] registered. Populated once
at class initialization and never
- /// mutated afterward (effectively immutable), so it is safe to share across
writers.
+ /// Adapts a Pinot value to the representation the Avro writer expects for
the given field schema, recursing into
+ /// array elements for multi-value columns.
+ @Nullable
+ private static Object convertValue(Schema fieldSchema, @Nullable Object
value) {
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof Object[]) {
+ Object[] values = (Object[]) value;
+ Schema elementSchema =
+ fieldSchema.getType() == Schema.Type.ARRAY ?
fieldSchema.getElementType() : fieldSchema;
+ // Only BOOLEAN (stored int -> Boolean) and BYTES (byte[] -> ByteBuffer)
element schemas can require a
+ // per-element transform. Every other MV element type is written as-is —
INT/LONG/FLOAT/DOUBLE/STRING directly,
+ // and UUID (string element, raw byte[] rendered by the registered
Conversion) — so hand the writer a zero-copy
+ // view over the existing array; allocating and copying a fresh list per
row would be pure overhead on the
+ // segment-write hot path. (BIG_DECIMAL has a BYTES element schema and
so takes the copy path below, but its
+ // BigDecimal values still pass through convertSingleValue unchanged for
the registered Conversion.)
+ Schema.Type elementType = elementSchema.getType();
+ if (elementType != Schema.Type.BOOLEAN && elementType !=
Schema.Type.BYTES) {
+ return Arrays.asList(values);
+ }
+ List<Object> converted = new ArrayList<>(values.length);
+ for (Object singleValue : values) {
+ converted.add(convertSingleValue(elementSchema, singleValue));
+ }
+ return converted;
+ }
+ return convertSingleValue(fieldSchema, value);
+ }
+
+ /// Adapts a single (non-array) Pinot value to what `GenericDatumWriter`
expects for `valueSchema`.
+ ///
+ /// Only two cases need fixing up; everything else is written as-is, either
because the Pinot representation already
+ /// *is* the Avro representation (`Integer` for `int`, `Long` for
`long{timestamp-millis}`, `String` for `string`)
+ /// or because a [Conversion] registered on [#getAvroDataModel] handles it
(`byte[]` for `string{uuid}`,
+ /// [java.math.BigDecimal] for `bytes{big-decimal}`).
+ @Nullable
+ private static Object convertSingleValue(Schema valueSchema, @Nullable
Object value) {
+ if (value == null) {
+ return null;
+ }
+ switch (valueSchema.getType()) {
+ case BOOLEAN:
+ // BOOLEAN is the one Pinot logical type with no Avro logical type to
carry a Conversion, so its stored int
+ // 0/1 has to be coerced here. Values that are already Boolean (e.g.
from a source that never went through
+ // Pinot's stored form) pass through.
+ return value instanceof Boolean ? value : ((Number) value).intValue()
!= 0;
Review Comment:
Yes for the production call paths: transformed and segment-read BOOLEAN
values are Integer, and plain BYTES values are byte[]. Simplified both to
direct casts in dc7f174598. BIG_DECIMAL also has Avro base type BYTES but
carries BigDecimal, so the conversion distinguishes the exact `big-decimal`
logical schema.
--
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]
