xiangfu0 commented on code in PR #19073:
URL: https://github.com/apache/pinot/pull/19073#discussion_r3661178385
##########
pinot-core/src/main/java/org/apache/pinot/core/util/SegmentProcessorAvroUtils.java:
##########
@@ -133,74 +193,57 @@ public byte[] fromCharSequence(CharSequence value, Schema
schema, LogicalType ty
}
}
- /// Converts a Pinot schema to an Avro schema
+ /// Converts a Pinot schema to an Avro schema, with the fields ordered by
column name.
+ ///
+ /// Field types are derived from the **original (logical)** Pinot data type,
so BOOLEAN becomes Avro `boolean`,
+ /// TIMESTAMP a `timestamp-millis` long, BIG_DECIMAL a `big-decimal` bytes
and UUID a `uuid` string, instead of all
+ /// four collapsing to their physical storage type. This must stay identical
to
+ /// `AvroSchemaUtil.toAvroSchema(FieldSpec)` in `pinot-avro-base` — the two
live in different modules (neither can
+ /// depend on the other) but feed the same writers, so
`SegmentProcessorAvroUtilsTest` pins them together. See that
+ /// method for the full mapping table and the value representation each Avro
type expects.
public static Schema
convertPinotSchemaToAvroSchema(org.apache.pinot.spi.data.Schema pinotSchema) {
SchemaBuilder.FieldAssembler<org.apache.avro.Schema> fieldAssembler =
SchemaBuilder.record("record").fields();
-
List<FieldSpec> orderedFieldSpecs = pinotSchema.getAllFieldSpecs().stream()
.sorted(Comparator.comparing(FieldSpec::getName))
.collect(Collectors.toList());
for (FieldSpec fieldSpec : orderedFieldSpecs) {
- String name = fieldSpec.getName();
- // Emit UUID columns as Avro string{logicalType:uuid} (matching
AvroUtils.getAvroSchemaFromPinotSchema)
- // so the runtime byte[] → canonical-string conversion in
convertGenericRowToAvroRecord lines up with
- // the field schema. Without this branch SV UUID would fall through to
BYTES (losing UUID semantics) and
- // MV UUID would throw at this point (MV switch below has no BYTES case).
- if (fieldSpec.getDataType() == DataType.UUID) {
- Schema uuidSchema =
LogicalTypes.uuid().addToSchema(Schema.create(Schema.Type.STRING));
- if (fieldSpec.isSingleValueField()) {
- fieldAssembler =
fieldAssembler.name(name).type(uuidSchema).noDefault();
- } else {
- fieldAssembler =
fieldAssembler.name(name).type().array().items(uuidSchema).noDefault();
- }
- continue;
- }
- DataType storedType = fieldSpec.getDataType().getStoredType();
- if (fieldSpec.isSingleValueField()) {
- switch (storedType) {
- case INT:
- fieldAssembler =
fieldAssembler.name(name).type().intType().noDefault();
- break;
- case LONG:
- fieldAssembler =
fieldAssembler.name(name).type().longType().noDefault();
- break;
- case FLOAT:
- fieldAssembler =
fieldAssembler.name(name).type().floatType().noDefault();
- break;
- case DOUBLE:
- fieldAssembler =
fieldAssembler.name(name).type().doubleType().noDefault();
- break;
- case STRING:
- fieldAssembler =
fieldAssembler.name(name).type().stringType().noDefault();
- break;
- case BYTES:
- fieldAssembler =
fieldAssembler.name(name).type().bytesType().noDefault();
- break;
- default:
- throw new RuntimeException("Unsupported data type: " + storedType);
- }
- } else {
- switch (storedType) {
- case INT:
- fieldAssembler =
fieldAssembler.name(name).type().array().items().intType().noDefault();
- break;
- case LONG:
- fieldAssembler =
fieldAssembler.name(name).type().array().items().longType().noDefault();
- break;
- case FLOAT:
- fieldAssembler =
fieldAssembler.name(name).type().array().items().floatType().noDefault();
- break;
- case DOUBLE:
- fieldAssembler =
fieldAssembler.name(name).type().array().items().doubleType().noDefault();
- break;
- case STRING:
- fieldAssembler =
fieldAssembler.name(name).type().array().items().stringType().noDefault();
- break;
- default:
- throw new RuntimeException("Unsupported data type: " + storedType);
- }
- }
+ fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type(toAvroSchema(fieldSpec)).noDefault();
}
return fieldAssembler.endRecord();
}
+
+ /// Returns the Avro schema for a whole Pinot column: the single-value type
from [#toAvroSchema(DataType)], or an
+ /// array of it for a multi-value column.
+ private static Schema toAvroSchema(FieldSpec fieldSpec) {
+ Schema valueSchema = toAvroSchema(fieldSpec.getDataType());
+ return fieldSpec.isSingleValueField() ? valueSchema :
Schema.createArray(valueSchema);
+ }
+
+ private static Schema toAvroSchema(DataType dataType) {
+ switch (dataType) {
+ case INT:
+ return Schema.create(Schema.Type.INT);
+ case LONG:
+ return Schema.create(Schema.Type.LONG);
+ case FLOAT:
+ return Schema.create(Schema.Type.FLOAT);
+ case DOUBLE:
+ return Schema.create(Schema.Type.DOUBLE);
+ case BOOLEAN:
+ return Schema.create(Schema.Type.BOOLEAN);
+ case TIMESTAMP:
+ return
LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG));
+ case BIG_DECIMAL:
+ return
LogicalTypes.bigDecimal().addToSchema(Schema.create(Schema.Type.BYTES));
+ case STRING:
+ case JSON:
+ return Schema.create(Schema.Type.STRING);
+ case BYTES:
+ return Schema.create(Schema.Type.BYTES);
+ case UUID:
+ return
LogicalTypes.uuid().addToSchema(Schema.create(Schema.Type.STRING));
Review Comment:
Kept `string{logicalType:uuid}` intentionally. Plain Avro BYTES would lose
UUID logical identity and change the output established by #18870; Avro UUID
supports string or fixed(16), not BYTES. fixed(16) would require a broader
named-schema, conversion, and compatibility change, so it is outside 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]
