This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new a8b207e7815 Generate Avro schemas from Pinot logical data types
(#19071) (#19073)
a8b207e7815 is described below
commit a8b207e78156296f686eea3a312a9cb68d08ac6a
Author: Xiang Fu <[email protected]>
AuthorDate: Mon Jul 27 17:10:31 2026 -0700
Generate Avro schemas from Pinot logical data types (#19071) (#19073)
---
.../pinot/core/util/SegmentProcessorAvroUtils.java | 215 ++++++++++-------
.../core/util/SegmentProcessorAvroUtilsTest.java | 258 +++++++++++++++++++++
.../plugin/inputformat/avro/AvroSchemaUtil.java | 147 ++++++------
.../pinot/plugin/inputformat/avro/AvroUtils.java | 76 +-----
.../inputformat/avro/AvroSchemaUtilTest.java | 67 ++++--
.../plugin/inputformat/avro/AvroUtilsTest.java | 84 +++++++
.../filebased/FileBasedSegmentWriterTest.java | 46 ++++
.../converter/PinotSegmentConverterTest.java | 75 ++++++
8 files changed, 738 insertions(+), 230 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/util/SegmentProcessorAvroUtils.java
b/pinot-core/src/main/java/org/apache/pinot/core/util/SegmentProcessorAvroUtils.java
index 0c7f5b5e2a7..5eb58a5e3c3 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/util/SegmentProcessorAvroUtils.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/util/SegmentProcessorAvroUtils.java
@@ -19,12 +19,15 @@
package org.apache.pinot.core.util;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
import org.apache.avro.Conversion;
+import org.apache.avro.Conversions;
import org.apache.avro.LogicalType;
import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
@@ -51,43 +54,103 @@ public final class SegmentProcessorAvroUtils {
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 Integer -> Boolean) and plain BYTES (byte[] ->
ByteBuffer) element schemas require a
+ // per-element transform. Every other MV element type is written as-is —
including BIG_DECIMAL, whose logical
+ // BYTES schema is handled 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.
+ Schema.Type elementType = elementSchema.getType();
+ if (elementType != Schema.Type.BOOLEAN
+ && (elementType != Schema.Type.BYTES ||
isBigDecimalSchema(elementSchema))) {
+ 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. All production callers provide either a
transformed row or a segment-read row,
+ // so a BOOLEAN value is always in Pinot's stored Integer form.
+ return (Integer) value != 0;
+ case BYTES:
+ // A plain BYTES value is always byte[] and must be wrapped as
ByteBuffer (GenericDatumWriter requires it).
+ // BIG_DECIMAL's logical BYTES value must stay untouched for its
registered Conversion.
+ return isBigDecimalSchema(valueSchema) ? value :
ByteBuffer.wrap((byte[]) value);
+ default:
+ return value;
+ }
+ }
+
+ private static boolean isBigDecimalSchema(Schema schema) {
+ LogicalType logicalType = schema.getLogicalType();
+ return logicalType != null &&
LogicalTypes.bigDecimal().getName().equals(logicalType.getName());
+ }
+
+ /// Shared Avro data model with the logical-type conversions registered.
Populated once at class initialization and
+ /// never mutated afterward (effectively immutable), so it is safe to share
across writers.
private static final GenericData AVRO_DATA_MODEL = createAvroDataModel();
/// Returns the shared Avro data model that a `GenericDatumWriter` (or
`AvroParquetWriter`) must be constructed with
- /// to serialize UUID columns produced by [#convertGenericRowToAvroRecord]:
it registers [UuidConversion] so the
- /// internal 16-byte UUID form is rendered as the canonical string required
by `string{logicalType:uuid}` fields.
- /// The UUID column's field schema must be `string{logicalType:uuid}` — as
emitted by
- /// [#convertPinotSchemaToAvroSchema] and
`AvroUtils.getAvroSchemaFromPinotSchema` — for the conversion to apply.
+ /// to serialize the logical-type columns produced by
[#convertGenericRowToAvroRecord]:
+ /// - [UuidConversion] renders Pinot's internal 16-byte UUID form as the
canonical string required by
+ /// `string{logicalType:uuid}` fields.
+ /// - Avro's `BigDecimalConversion` encodes a [java.math.BigDecimal] —
unscaled value plus its own scale — into the
+ /// `bytes{logicalType:big-decimal}` fields Pinot emits for BIG_DECIMAL
columns.
+ ///
+ /// The column's field schema must be the one emitted by
[#convertPinotSchemaToAvroSchema] /
+ /// `AvroUtils.getAvroSchemaFromPinotSchema` for the conversions to apply.
BOOLEAN and TIMESTAMP need no entry
+ /// here: Avro has no `boolean` logical type (the coercion happens in
[#convertGenericRowToAvroRecord]), and a
+ /// `Long` already *is* the base representation of
`long{logicalType:timestamp-millis}`.
///
/// The returned instance is shared and must be treated as read-only: do not
call its mutators
/// (`addLogicalTypeConversion`, `setStringType`, ...), which are not
thread-safe against concurrent writer reads.
@@ -98,6 +161,7 @@ public final class SegmentProcessorAvroUtils {
private static GenericData createAvroDataModel() {
GenericData model = new GenericData();
model.addLogicalTypeConversion(new UuidConversion());
+ model.addLogicalTypeConversion(new Conversions.BigDecimalConversion());
return model;
}
@@ -133,74 +197,57 @@ public final class SegmentProcessorAvroUtils {
}
}
- /// 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));
+ default:
+ throw new UnsupportedOperationException("Unsupported data type: " +
dataType);
+ }
+ }
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/util/SegmentProcessorAvroUtilsTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/util/SegmentProcessorAvroUtilsTest.java
index 2f3012a2c66..b595ae2bfd2 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/util/SegmentProcessorAvroUtilsTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/util/SegmentProcessorAvroUtilsTest.java
@@ -19,8 +19,12 @@
package org.apache.pinot.core.util;
import java.io.File;
+import java.math.BigDecimal;
import java.nio.ByteBuffer;
+import java.nio.file.Files;
import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.avro.LogicalType;
import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder;
@@ -31,12 +35,20 @@ import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.commons.io.FileUtils;
+import org.apache.pinot.plugin.inputformat.avro.AvroRecordReader;
+import org.apache.pinot.plugin.inputformat.avro.AvroUtils;
+import org.apache.pinot.segment.local.utils.DataTypeTransformerUtils;
+import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.PinotDataType;
import org.apache.pinot.spi.utils.UuidUtils;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
@@ -67,6 +79,30 @@ public class SegmentProcessorAvroUtilsTest {
assertEquals(record.get("bytesCol"), ByteBuffer.wrap(rawBytes), "BYTES
byte[] must be wrapped as ByteBuffer");
}
+ /// A logical BYTES array such as `bytes{logicalType:big-decimal}` is
handled by its registered Conversion and must
+ /// not take the plain-BYTES per-element copy path.
+ @Test
+ public void
testConvertGenericRowToAvroRecordKeepsLogicalBytesArrayZeroCopy() {
+ Schema bigDecimalSchema =
LogicalTypes.bigDecimal().addToSchema(Schema.create(Schema.Type.BYTES));
+ Schema recordSchema = SchemaBuilder.record("record").fields()
+
.name("bigDecimalCol").type().array().items(bigDecimalSchema).noDefault()
+ .endRecord();
+
+ BigDecimal original = new BigDecimal("123.45");
+ BigDecimal replacement = new BigDecimal("678.90");
+ Object[] values = {original};
+ GenericRow row = new GenericRow();
+ row.putValue("bigDecimalCol", values);
+
+ GenericData.Record record = new GenericData.Record(recordSchema);
+ SegmentProcessorAvroUtils.convertGenericRowToAvroRecord(row, record);
+
+ List<?> converted = (List<?>) record.get("bigDecimalCol");
+ assertSame(converted.get(0), original);
+ values[0] = replacement;
+ assertSame(converted.get(0), replacement, "logical BYTES arrays must
retain the zero-copy array view");
+ }
+
/// End-to-end: a GenericDatumWriter built with getAvroDataModel()
serializes the raw 16-byte UUID values as their
/// canonical string (via the registered uuid Conversion) for both SV and
MV, and the on-disk value reads back as
/// that string with a vanilla reader. Without the registered Conversion the
write would fail (a ByteBuffer/byte[]
@@ -120,6 +156,168 @@ public class SegmentProcessorAvroUtilsTest {
}
}
+ /// The generated Avro schema must describe the *logical* Pinot type.
Switching on the stored type instead emitted
+ /// `int` for BOOLEAN, a bare `long` for TIMESTAMP, and rejected BIG_DECIMAL
outright.
+ @Test
+ public void testConvertPinotSchemaToAvroSchemaUsesLogicalTypes() {
+ Schema avroSchema =
SegmentProcessorAvroUtils.convertPinotSchemaToAvroSchema(allTypesPinotSchema());
+
+ assertFieldType(avroSchema, "intSv", Schema.Type.INT, null);
+ assertFieldType(avroSchema, "longSv", Schema.Type.LONG, null);
+ assertFieldType(avroSchema, "floatSv", Schema.Type.FLOAT, null);
+ assertFieldType(avroSchema, "doubleSv", Schema.Type.DOUBLE, null);
+ assertFieldType(avroSchema, "boolSv", Schema.Type.BOOLEAN, null);
+ assertFieldType(avroSchema, "tsSv", Schema.Type.LONG, "timestamp-millis");
+ assertFieldType(avroSchema, "bigDecimalSv", Schema.Type.BYTES,
"big-decimal");
+ assertFieldType(avroSchema, "stringSv", Schema.Type.STRING, null);
+ assertFieldType(avroSchema, "jsonSv", Schema.Type.STRING, null);
+ assertFieldType(avroSchema, "bytesSv", Schema.Type.BYTES, null);
+ assertFieldType(avroSchema, "uuidSv", Schema.Type.STRING, "uuid");
+
+ assertElementType(avroSchema, "intMv", Schema.Type.INT, null);
+ assertElementType(avroSchema, "boolMv", Schema.Type.BOOLEAN, null);
+ assertElementType(avroSchema, "tsMv", Schema.Type.LONG,
"timestamp-millis");
+ assertElementType(avroSchema, "bigDecimalMv", Schema.Type.BYTES,
"big-decimal");
+ assertElementType(avroSchema, "stringMv", Schema.Type.STRING, null);
+ assertElementType(avroSchema, "bytesMv", Schema.Type.BYTES, null);
+ assertElementType(avroSchema, "uuidMv", Schema.Type.STRING, "uuid");
+ }
+
+ /// This mapping is duplicated in `AvroUtils.getAvroSchemaFromPinotSchema`
(pinot-avro-base), because neither module
+ /// can depend on the other, yet both feed the same writers and the same
shared data model — `PinotSegmentToAvro
+ /// Converter` even pairs one class's schema with the other's data model.
Any divergence would silently produce
+ /// records the writer cannot serialize, so pin the two together here.
+ @Test
+ public void testConvertPinotSchemaToAvroSchemaMatchesAvroUtils() {
+ org.apache.pinot.spi.data.Schema pinotSchema = allTypesPinotSchema();
+
+ Schema avroSchema =
SegmentProcessorAvroUtils.convertPinotSchemaToAvroSchema(pinotSchema);
+ Schema avroUtilsSchema =
AvroUtils.getAvroSchemaFromPinotSchema(pinotSchema);
+
+ assertEquals(avroSchema.getFields().size(),
avroUtilsSchema.getFields().size());
+ for (Schema.Field field : avroSchema.getFields()) {
+ Schema.Field avroUtilsField = avroUtilsSchema.getField(field.name());
+ assertNotNull(avroUtilsField, "AvroUtils is missing field: " +
field.name());
+ assertEquals(field.schema(), avroUtilsField.schema(), "schema mismatch
for field: " + field.name());
+ }
+ }
+
+ /// Full ingest round trip for every supported type, SV and MV: Pinot's
internal (stored) values are written through
+ /// the shared data model, read back with the production `AvroRecordReader`,
run through the same data-type
+ /// transformation ingestion applies, and must land on the values we started
with.
+ @Test
+ public void testAllTypesRoundTripThroughAvroDataModel()
+ throws Exception {
+ org.apache.pinot.spi.data.Schema pinotSchema = allTypesPinotSchema();
+ Schema avroSchema =
SegmentProcessorAvroUtils.convertPinotSchemaToAvroSchema(pinotSchema);
+
+ byte[] uuidBytes =
UuidUtils.toBytes("12345678-1234-1234-1234-1234567890ab");
+ byte[] otherUuidBytes =
UuidUtils.toBytes("550e8400-e29b-41d4-a716-446655440000");
+ byte[] rawBytes = {1, 2, 3, 4};
+ // A scale the value's own magnitude does not imply, to prove
`big-decimal` carries the scale per value rather
+ // than pinning one for the column the way `decimal(precision, scale)`
would.
+ BigDecimal bigDecimal = new BigDecimal("123.45000");
+ BigDecimal hugeBigDecimal = new
BigDecimal("-9999999999999999999999.12345");
+
+ GenericRow row = new GenericRow();
+ row.putValue("intSv", 1);
+ row.putValue("longSv", 2L);
+ row.putValue("floatSv", 3.0f);
+ row.putValue("doubleSv", 4.0);
+ // BOOLEAN and TIMESTAMP arrive in Pinot's stored form: int 0/1 and epoch
millis.
+ row.putValue("boolSv", 1);
+ row.putValue("tsSv", 1609491661001L);
+ row.putValue("bigDecimalSv", bigDecimal);
+ row.putValue("stringSv", "5");
+ row.putValue("jsonSv", "{\"a\":1}");
+ row.putValue("bytesSv", rawBytes);
+ row.putValue("uuidSv", uuidBytes);
+ row.putValue("intMv", new Object[]{7, 8});
+ row.putValue("longMv", new Object[]{9L, 10L});
+ row.putValue("floatMv", new Object[]{11.0f, 12.0f});
+ row.putValue("doubleMv", new Object[]{13.0, 14.0});
+ row.putValue("boolMv", new Object[]{0, 1});
+ row.putValue("tsMv", new Object[]{1609491661001L, 0L});
+ row.putValue("bigDecimalMv", new Object[]{bigDecimal, hugeBigDecimal});
+ row.putValue("stringMv", new Object[]{"15", "16"});
+ row.putValue("bytesMv", new Object[]{rawBytes, new byte[]{9}});
+ row.putValue("uuidMv", new Object[]{uuidBytes, otherUuidBytes});
+
+ File tmpDir = Files.createTempDirectory("allTypesRoundTrip").toFile();
+ try {
+ File avroFile = new File(tmpDir, "data.avro");
+ GenericData.Record record =
+ SegmentProcessorAvroUtils.convertGenericRowToAvroRecord(row, new
GenericData.Record(avroSchema));
+ try (DataFileWriter<GenericData.Record> writer =
+ new DataFileWriter<>(new GenericDatumWriter<>(avroSchema,
SegmentProcessorAvroUtils.getAvroDataModel()))) {
+ writer.create(avroSchema, avroFile);
+ writer.append(record);
+ }
+
+ GenericRow readRow;
+ try (AvroRecordReader recordReader = new AvroRecordReader()) {
+ recordReader.init(avroFile, pinotSchema.getColumnNames(), null);
+ assertTrue(recordReader.hasNext());
+ readRow = recordReader.next();
+ }
+ // Apply the same data-type transformation ingestion runs, so the
comparison is against Pinot's stored form.
+ for (FieldSpec fieldSpec : pinotSchema.getAllFieldSpecs()) {
+ String column = fieldSpec.getName();
+ readRow.putValue(column,
DataTypeTransformerUtils.transformValue(column, readRow.getValue(column),
+ PinotDataType.getPinotDataTypeForIngestion(fieldSpec)));
+ }
+
+ assertEquals(readRow.getValue("intSv"), 1);
+ assertEquals(readRow.getValue("longSv"), 2L);
+ assertEquals(readRow.getValue("floatSv"), 3.0f);
+ assertEquals(readRow.getValue("doubleSv"), 4.0);
+ assertEquals(readRow.getValue("boolSv"), 1);
+ assertEquals(readRow.getValue("tsSv"), 1609491661001L);
+ assertEquals(readRow.getValue("bigDecimalSv"), bigDecimal);
+ assertEquals(((BigDecimal) readRow.getValue("bigDecimalSv")).scale(),
bigDecimal.scale(),
+ "big-decimal must preserve the value's own scale");
+ assertEquals(readRow.getValue("stringSv"), "5");
+ assertEquals(readRow.getValue("jsonSv"), "{\"a\":1}");
+ assertEquals((byte[]) readRow.getValue("bytesSv"), rawBytes);
+ assertEquals((byte[]) readRow.getValue("uuidSv"), uuidBytes);
+ assertEquals((Object[]) readRow.getValue("intMv"), new Object[]{7, 8});
+ assertEquals((Object[]) readRow.getValue("longMv"), new Object[]{9L,
10L});
+ assertEquals((Object[]) readRow.getValue("floatMv"), new Object[]{11.0f,
12.0f});
+ assertEquals((Object[]) readRow.getValue("doubleMv"), new Object[]{13.0,
14.0});
+ assertEquals((Object[]) readRow.getValue("boolMv"), new Object[]{0, 1});
+ assertEquals((Object[]) readRow.getValue("tsMv"), new
Object[]{1609491661001L, 0L});
+ assertEquals((Object[]) readRow.getValue("bigDecimalMv"), new
Object[]{bigDecimal, hugeBigDecimal});
+ assertEquals((Object[]) readRow.getValue("stringMv"), new Object[]{"15",
"16"});
+ // byte[] elements need element-wise comparison — Object[] equality
would compare them by identity.
+ assertBytesArrayEquals(readRow.getValue("bytesMv"), rawBytes, new
byte[]{9});
+ assertBytesArrayEquals(readRow.getValue("uuidMv"), uuidBytes,
otherUuidBytes);
+ } finally {
+ FileUtils.deleteDirectory(tmpDir);
+ }
+ }
+
+ /// BOOLEAN is the one logical type with no Avro logical type to carry a
`Conversion`, so the stored int 0/1 has to
+ /// be coerced to Boolean when building the record — both for single values
and for multi-value elements.
+ @Test
+ public void testConvertGenericRowToAvroRecordCoercesStoredBooleans() {
+ org.apache.pinot.spi.data.Schema pinotSchema = new
org.apache.pinot.spi.data.Schema.SchemaBuilder()
+ .setSchemaName("boolSchema")
+ .addSingleValueDimension("boolSv", DataType.BOOLEAN)
+ .addMultiValueDimension("boolMv", DataType.BOOLEAN)
+ .build();
+ Schema avroSchema =
SegmentProcessorAvroUtils.convertPinotSchemaToAvroSchema(pinotSchema);
+
+ GenericRow row = new GenericRow();
+ row.putValue("boolSv", 0);
+ row.putValue("boolMv", new Object[]{1, 0});
+
+ GenericData.Record record =
+ SegmentProcessorAvroUtils.convertGenericRowToAvroRecord(row, new
GenericData.Record(avroSchema));
+
+ assertEquals(record.get("boolSv"), Boolean.FALSE);
+ assertEquals(record.get("boolMv"), List.of(Boolean.TRUE, Boolean.FALSE));
+ }
+
/// convertPinotSchemaToAvroSchema must emit SV UUID as
string{logicalType:uuid} and MV UUID as
/// array<string{logicalType:uuid}>, which is what the uuid Conversion above
pairs with.
@Test
@@ -144,4 +342,64 @@ public class SegmentProcessorAvroUtilsTest {
assertEquals(LogicalTypes.fromSchemaIgnoreInvalid(mvElementSchema),
LogicalTypes.uuid(),
"MV UUID elements must carry the uuid logical type");
}
+
+ /// One column per supported data type, single-value and (where Pinot allows
it) multi-value.
+ private static org.apache.pinot.spi.data.Schema allTypesPinotSchema() {
+ return new org.apache.pinot.spi.data.Schema.SchemaBuilder()
+ .setSchemaName("allTypes")
+ .addSingleValueDimension("intSv", DataType.INT)
+ .addSingleValueDimension("longSv", DataType.LONG)
+ .addSingleValueDimension("floatSv", DataType.FLOAT)
+ .addSingleValueDimension("doubleSv", DataType.DOUBLE)
+ .addSingleValueDimension("boolSv", DataType.BOOLEAN)
+ .addSingleValueDimension("tsSv", DataType.TIMESTAMP)
+ .addSingleValueDimension("bigDecimalSv", DataType.BIG_DECIMAL)
+ .addSingleValueDimension("stringSv", DataType.STRING)
+ .addSingleValueDimension("jsonSv", DataType.JSON)
+ .addSingleValueDimension("bytesSv", DataType.BYTES)
+ .addSingleValueDimension("uuidSv", DataType.UUID)
+ .addMultiValueDimension("intMv", DataType.INT)
+ .addMultiValueDimension("longMv", DataType.LONG)
+ .addMultiValueDimension("floatMv", DataType.FLOAT)
+ .addMultiValueDimension("doubleMv", DataType.DOUBLE)
+ .addMultiValueDimension("boolMv", DataType.BOOLEAN)
+ .addMultiValueDimension("tsMv", DataType.TIMESTAMP)
+ .addMultiValueDimension("bigDecimalMv", DataType.BIG_DECIMAL)
+ .addMultiValueDimension("stringMv", DataType.STRING)
+ .addMultiValueDimension("bytesMv", DataType.BYTES)
+ .addMultiValueDimension("uuidMv", DataType.UUID)
+ .build();
+ }
+
+ private static void assertBytesArrayEquals(Object actual, byte[]...
expected) {
+ byte[][] actualArray = (byte[][]) actual;
+ assertEquals(actualArray.length, expected.length);
+ for (int i = 0; i < expected.length; i++) {
+ assertEquals(actualArray[i], expected[i], "mismatch at index " + i);
+ }
+ }
+
+ private static void assertFieldType(Schema avroSchema, String field,
Schema.Type expectedType,
+ @Nullable String expectedLogicalType) {
+ assertAvroType(avroSchema.getField(field).schema(), field, expectedType,
expectedLogicalType);
+ }
+
+ private static void assertElementType(Schema avroSchema, String field,
Schema.Type expectedType,
+ @Nullable String expectedLogicalType) {
+ Schema fieldSchema = avroSchema.getField(field).schema();
+ assertEquals(fieldSchema.getType(), Schema.Type.ARRAY, field + " must be
an array");
+ assertAvroType(fieldSchema.getElementType(), field, expectedType,
expectedLogicalType);
+ }
+
+ private static void assertAvroType(Schema schema, String field, Schema.Type
expectedType,
+ @Nullable String expectedLogicalType) {
+ assertEquals(schema.getType(), expectedType, "unexpected Avro type for " +
field);
+ LogicalType logicalType = LogicalTypes.fromSchemaIgnoreInvalid(schema);
+ if (expectedLogicalType == null) {
+ assertNull(logicalType, field + " must carry no logical type");
+ } else {
+ assertNotNull(logicalType, field + " must carry the " +
expectedLogicalType + " logical type");
+ assertEquals(logicalType.getName(), expectedLogicalType, "unexpected
logical type for " + field);
+ }
+ }
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtil.java
b/pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtil.java
index 099ccae9170..11bb628a57f 100644
---
a/pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtil.java
+++
b/pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtil.java
@@ -20,6 +20,7 @@ package org.apache.pinot.plugin.inputformat.avro;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.io.IOException;
import org.apache.avro.LogicalType;
import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
@@ -38,6 +39,70 @@ public class AvroSchemaUtil {
// in AvroRecordExtractor; this class only deals with schema-shape mapping.
private static final String UUID = "uuid";
+ /// Returns the Avro schema for a single value of the given Pinot
[DataType]. This is the canonical Pinot-to-Avro
+ /// type mapping; it is driven by the **original (logical)** data type
rather than the stored type, so logical types
+ /// stay self-describing in the generated Avro schema instead of collapsing
to their physical storage type.
+ ///
+ /// | Pinot type | Avro type | Value handed to the Avro
writer |
+ ///
|---------------|--------------------------|----------------------------------------------------------|
+ /// | INT | `int` | `Integer`
|
+ /// | LONG | `long` | `Long`
|
+ /// | FLOAT | `float` | `Float`
|
+ /// | DOUBLE | `double` | `Double`
|
+ /// | BOOLEAN | `boolean` | `Boolean` (Pinot stores
`int` 0/1 — see note below) |
+ /// | TIMESTAMP | `long{timestamp-millis}` | `Long` millis since epoch
(the logical type's base type) |
+ /// | BIG_DECIMAL | `bytes{big-decimal}` | [java.math.BigDecimal] via
Avro's `BigDecimalConversion` |
+ /// | STRING / JSON | `string` | `String`
|
+ /// | BYTES | `bytes` | [java.nio.ByteBuffer]
|
+ /// | UUID | `string{uuid}` | 16-byte `byte[]` via a
`uuid` `Conversion` |
+ ///
+ /// `timestamp-millis` needs no write-side conversion because `Long` *is*
that logical type's base representation.
+ /// `big-decimal` is used for BIG_DECIMAL rather than `decimal(precision,
scale)` because Pinot does not pin a
+ /// per-column precision/scale, and `big-decimal` encodes the scale with
every value. BOOLEAN is the one type with
+ /// no Avro logical type to hang a `Conversion` on, so writers must coerce
Pinot's stored `int` 0/1 to `Boolean`
+ /// themselves (see
`SegmentProcessorAvroUtils#convertGenericRowToAvroRecord` and `AvroWriter`).
+ ///
+ /// This mapping is intentionally one-way: [#valueOf(Schema)] does **not**
map `timestamp-millis` back to TIMESTAMP
+ /// or `big-decimal` back to BIG_DECIMAL, so that Pinot schema inference
from existing Avro data keeps its
+ /// long-standing behavior.
+ ///
+ /// Throws [UnsupportedOperationException] for types with no Avro
representation (STRUCT / MAP / OPEN_STRUCT /
+ /// LIST / UNKNOWN).
+ public 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));
+ default:
+ throw new UnsupportedOperationException("Unsupported data type: " +
dataType);
+ }
+ }
+
+ /// Returns the Avro schema for a whole Pinot column:
[#toAvroSchema(DataType)] for a single-value field, or an
+ /// array of it for a multi-value field.
+ public static Schema toAvroSchema(FieldSpec fieldSpec) {
+ Schema valueSchema = toAvroSchema(fieldSpec.getDataType());
+ return fieldSpec.isSingleValueField() ? valueSchema :
Schema.createArray(valueSchema);
+ }
+
/// Returns the Pinot data type for a bare Avro type. This does not honor
logical types (e.g. a `string` or `fixed`
/// carrying `logicalType:uuid` maps to STRING/BYTES, not UUID); prefer
[#valueOf(Schema)] when a full [Schema] is
/// available.
@@ -104,76 +169,26 @@ public class AvroSchemaUtil {
}
}
- /// Builds the Avro schema JSON for a single Pinot field. Used to generate
sample Avro data from a Pinot schema
- /// (see `AvroWriter`). Each field is emitted as a nullable union `["null",
<type>]`.
- ///
- /// The switch is driven by the original (logical) [DataType] rather than
the stored type, so logical types are
- /// represented faithfully instead of collapsing to their physical storage
type: BOOLEAN maps to Avro `boolean`,
- /// TIMESTAMP to a `timestamp-millis` long (not a plain `int`/`long`), and
UUID to a `uuid`-logical-type string
- /// (not raw `bytes`).
+ /// Builds the Avro schema JSON for a single Pinot field, as a nullable
union `["null", <type>]`. Used to generate
+ /// sample Avro data from a Pinot schema (see `AvroWriter`).
///
- /// This intentionally differs from the segment-processing converters
`AvroUtils.getAvroSchemaFromPinotSchema` and
- /// `SegmentProcessorAvroUtils.convertPinotSchemaToAvroSchema`, which switch
on the stored type because they
- /// serialize Pinot's physically-stored values (e.g. an int for BOOLEAN)
directly.
+ /// The `<type>` branch is [#toAvroSchema(DataType)] rendered as JSON, so
this shares the single logical-type
+ /// mapping documented there. The field is always emitted as the
single-value type even for a multi-value
+ /// [FieldSpec] — the data generator backing `AvroWriter` has never produced
Avro arrays.
public static ObjectNode toAvroSchemaJsonObject(FieldSpec fieldSpec) {
ObjectNode jsonSchema = JsonUtils.newObjectNode();
jsonSchema.put("name", fieldSpec.getName());
- DataType dataType = fieldSpec.getDataType();
- switch (dataType) {
- case INT:
- jsonSchema.set("type", convertStringsToJsonArray("null", "int"));
- return jsonSchema;
- case LONG:
- jsonSchema.set("type", convertStringsToJsonArray("null", "long"));
- return jsonSchema;
- case FLOAT:
- jsonSchema.set("type", convertStringsToJsonArray("null", "float"));
- return jsonSchema;
- case DOUBLE:
- jsonSchema.set("type", convertStringsToJsonArray("null", "double"));
- return jsonSchema;
- case BOOLEAN:
- jsonSchema.set("type", convertStringsToJsonArray("null", "boolean"));
- return jsonSchema;
- case TIMESTAMP:
- // TIMESTAMP is stored as LONG millis-since-epoch; annotate the long
branch with the timestamp-millis
- // logical type so the value stays a long but is self-describing as a
timestamp.
- ObjectNode timestampType = JsonUtils.newObjectNode();
- timestampType.put("type", "long");
- timestampType.put("logicalType", "timestamp-millis");
- ArrayNode timestampUnion = JsonUtils.newArrayNode();
- timestampUnion.add("null");
- timestampUnion.add(timestampType);
- jsonSchema.set("type", timestampUnion);
- return jsonSchema;
- case STRING:
- case JSON:
- jsonSchema.set("type", convertStringsToJsonArray("null", "string"));
- return jsonSchema;
- case UUID:
- // UUID is a logical type; represent it faithfully as an Avro string
annotated with logicalType "uuid" rather
- // than collapsing to raw bytes, so generated sample data round-trips
as canonical UUID strings.
- ObjectNode uuidType = JsonUtils.newObjectNode();
- uuidType.put("type", "string");
- uuidType.put("logicalType", "uuid");
- ArrayNode uuidUnion = JsonUtils.newArrayNode();
- uuidUnion.add("null");
- uuidUnion.add(uuidType);
- jsonSchema.set("type", uuidUnion);
- return jsonSchema;
- case BYTES:
- jsonSchema.set("type", convertStringsToJsonArray("null", "bytes"));
- return jsonSchema;
- default:
- throw new UnsupportedOperationException("Unsupported data type: " +
dataType);
- }
- }
-
- private static ArrayNode convertStringsToJsonArray(String... strings) {
- ArrayNode jsonArray = JsonUtils.newArrayNode();
- for (String string : strings) {
- jsonArray.add(string);
+ ArrayNode nullableUnion = JsonUtils.newArrayNode();
+ nullableUnion.add("null");
+ // Schema.toString() emits valid JSON: a bare name for primitives ("int"),
an object for logical types
+ // ({"type":"long","logicalType":"timestamp-millis"}).
+ try {
+
nullableUnion.add(JsonUtils.stringToJsonNode(toAvroSchema(fieldSpec.getDataType()).toString()));
+ } catch (IOException e) {
+ throw new IllegalStateException("Caught exception while parsing the Avro
schema generated for field: "
+ + fieldSpec.getName(), e);
}
- return jsonArray;
+ jsonSchema.set("type", nullableUnion);
+ return jsonSchema;
}
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroUtils.java
b/pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroUtils.java
index b01916df360..189657dbd5e 100644
---
a/pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroUtils.java
+++
b/pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroUtils.java
@@ -27,7 +27,6 @@ import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.zip.GZIPInputStream;
import javax.annotation.Nullable;
-import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema.Field;
import org.apache.avro.SchemaBuilder;
import org.apache.avro.file.DataFileStream;
@@ -151,73 +150,22 @@ public class AvroUtils {
}
}
- /**
- * Helper method to build Avro schema from Pinot schema.
- *
- * @param pinotSchema Pinot schema.
- * @return Avro schema.
- */
+ /// Builds an Avro schema from a Pinot schema, one non-nullable field per
[FieldSpec] in schema order.
+ ///
+ /// Field types come from [AvroSchemaUtil#toAvroSchema(FieldSpec)], which
maps 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. See that method
+ /// for the full mapping table and for the value representation each Avro
type expects.
+ ///
+ /// Rows are written into this schema by
`SegmentProcessorAvroUtils.convertGenericRowToAvroRecord` using the data
+ /// model returned by `SegmentProcessorAvroUtils.getAvroDataModel()`, which
registers the matching logical-type
+ /// conversions; the two must stay in sync.
public static org.apache.avro.Schema getAvroSchemaFromPinotSchema(Schema
pinotSchema) {
SchemaBuilder.FieldAssembler<org.apache.avro.Schema> fieldAssembler =
SchemaBuilder.record("record").fields();
-
for (FieldSpec fieldSpec : pinotSchema.getAllFieldSpecs()) {
- if (fieldSpec.getDataType() == DataType.UUID) {
- org.apache.avro.Schema uuidSchema =
LogicalTypes.uuid().addToSchema(org.apache.avro.Schema.create(
- org.apache.avro.Schema.Type.STRING));
- if (fieldSpec.isSingleValueField()) {
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type(uuidSchema).noDefault();
- } else {
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().array().items(uuidSchema).noDefault();
- }
- continue;
- }
- DataType storedType = fieldSpec.getDataType().getStoredType();
- if (fieldSpec.isSingleValueField()) {
- switch (storedType) {
- case INT:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().intType().noDefault();
- break;
- case LONG:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().longType().noDefault();
- break;
- case FLOAT:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().floatType().noDefault();
- break;
- case DOUBLE:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().doubleType().noDefault();
- break;
- case STRING:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().stringType().noDefault();
- break;
- case BYTES:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().bytesType().noDefault();
- break;
- default:
- throw new RuntimeException("Unsupported data type: " + storedType);
- }
- } else {
- switch (storedType) {
- case INT:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().array().items().intType().noDefault();
- break;
- case LONG:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().array().items().longType().noDefault();
- break;
- case FLOAT:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().array().items().floatType().noDefault();
- break;
- case DOUBLE:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().array().items().doubleType().noDefault();
- break;
- case STRING:
- fieldAssembler =
fieldAssembler.name(fieldSpec.getName()).type().array().items().stringType().noDefault();
- break;
- default:
- throw new RuntimeException("Unsupported data type: " + storedType);
- }
- }
+ fieldAssembler =
+
fieldAssembler.name(fieldSpec.getName()).type(AvroSchemaUtil.toAvroSchema(fieldSpec)).noDefault();
}
-
return fieldAssembler.endRecord();
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-avro-base/src/test/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtilTest.java
b/pinot-plugins/pinot-input-format/pinot-avro-base/src/test/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtilTest.java
index 436e3d8c68f..a899d4efb69 100644
---
a/pinot-plugins/pinot-input-format/pinot-avro-base/src/test/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtilTest.java
+++
b/pinot-plugins/pinot-input-format/pinot-avro-base/src/test/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtilTest.java
@@ -31,7 +31,12 @@ import static org.testng.Assert.assertThrows;
public class AvroSchemaUtilTest {
- /// The switch must be driven by the original (logical) data type, not the
stored type. Otherwise BOOLEAN collapses
+ /// Every Pinot data type that has an Avro representation.
+ private static final DataType[] SUPPORTED_DATA_TYPES =
+ {DataType.INT, DataType.LONG, DataType.FLOAT, DataType.DOUBLE,
DataType.BOOLEAN, DataType.TIMESTAMP,
+ DataType.BIG_DECIMAL, DataType.STRING, DataType.JSON,
DataType.BYTES, DataType.UUID};
+
+ /// The mapping must be driven by the original (logical) data type, not the
stored type. Otherwise BOOLEAN collapses
/// to "int" and TIMESTAMP to a plain "long", misrepresenting the column in
the generated Avro schema.
@Test
public void testToAvroSchemaJsonObjectUsesOriginalType() {
@@ -44,28 +49,50 @@ public class AvroSchemaUtilTest {
assertPrimitiveType(DataType.BYTES, "bytes");
// Logical types must not collapse to their stored INT/LONG type.
assertPrimitiveType(DataType.BOOLEAN, "boolean");
+ assertLogicalType(DataType.TIMESTAMP, "long", "timestamp-millis");
+ assertLogicalType(DataType.BIG_DECIMAL, "bytes", "big-decimal");
+ }
- JsonNode type = typeOf(DataType.TIMESTAMP);
- assertEquals(type.get(0).asText(), "null");
- JsonNode timestampBranch = type.get(1);
- assertEquals(timestampBranch.get("type").asText(), "long");
- assertEquals(timestampBranch.get("logicalType").asText(),
"timestamp-millis");
+ /// UUID is a logical type; a single-value column maps to an Avro string
carrying the "uuid" logical type.
+ @Test
+ public void testToAvroSchemaJsonObjectForUuid() {
+ assertLogicalType(DataType.UUID, "string", "uuid");
}
- /// Types with no Avro mapping (e.g. BIG_DECIMAL) must be rejected rather
than silently mishandled.
+ /// The Avro-schema and JSON forms are two views of one mapping, so they
must agree for every supported type.
@Test
- public void testToAvroSchemaJsonObjectRejectsUnsupportedType() {
- assertThrows(UnsupportedOperationException.class,
- () -> AvroSchemaUtil.toAvroSchemaJsonObject(new
DimensionFieldSpec("col", DataType.BIG_DECIMAL, true)));
+ public void testToAvroSchemaJsonObjectMatchesToAvroSchema() {
+ for (DataType dataType : SUPPORTED_DATA_TYPES) {
+ JsonNode type = typeOf(dataType);
+ assertEquals(type.get(1).toString(),
AvroSchemaUtil.toAvroSchema(dataType).toString(),
+ "mismatch for " + dataType);
+ }
}
- /// UUID is a logical type; a single-value column maps to an Avro string
carrying the "uuid" logical type.
+ /// Single-value columns map to the bare value schema; multi-value columns
to an array of it.
@Test
- public void testToAvroSchemaJsonObjectForUuid() {
- JsonNode type = typeOf(DataType.UUID);
- assertEquals(type.get(0).asText(), "null");
- assertEquals(type.get(1).get("type").asText(), "string");
- assertEquals(type.get(1).get("logicalType").asText(), "uuid");
+ public void testToAvroSchemaHonorsMultiValue() {
+ for (DataType dataType : SUPPORTED_DATA_TYPES) {
+ Schema valueSchema = AvroSchemaUtil.toAvroSchema(dataType);
+ assertEquals(AvroSchemaUtil.toAvroSchema(new DimensionFieldSpec("col",
dataType, true)), valueSchema,
+ "SV mismatch for " + dataType);
+ if (dataType == DataType.JSON) {
+ // JSON has no multi-value form in Pinot.
+ continue;
+ }
+ Schema mvSchema = AvroSchemaUtil.toAvroSchema(new
DimensionFieldSpec("col", dataType, false));
+ assertEquals(mvSchema.getType(), Schema.Type.ARRAY, "MV mismatch for " +
dataType);
+ assertEquals(mvSchema.getElementType(), valueSchema, "MV element
mismatch for " + dataType);
+ }
+ }
+
+ /// Types with no Avro representation must be rejected rather than silently
mishandled.
+ @Test
+ public void testToAvroSchemaRejectsUnsupportedType() {
+ for (DataType dataType : new DataType[]{DataType.MAP, DataType.STRUCT,
DataType.OPEN_STRUCT, DataType.LIST,
+ DataType.UNKNOWN}) {
+ assertThrows(UnsupportedOperationException.class, () ->
AvroSchemaUtil.toAvroSchema(dataType));
+ }
}
@Test
@@ -109,6 +136,14 @@ public class AvroSchemaUtilTest {
assertEquals(type.get(1).asText(), expectedAvroType);
}
+ private static void assertLogicalType(DataType dataType, String
expectedAvroType, String expectedLogicalType) {
+ JsonNode type = typeOf(dataType);
+ assertEquals(type.get(0).asText(), "null");
+ JsonNode branch = type.get(1);
+ assertEquals(branch.get("type").asText(), expectedAvroType);
+ assertEquals(branch.get("logicalType").asText(), expectedLogicalType);
+ }
+
private static JsonNode typeOf(DataType dataType) {
ObjectNode jsonSchema = AvroSchemaUtil.toAvroSchemaJsonObject(new
DimensionFieldSpec("col", dataType, true));
assertEquals(jsonSchema.get("name").asText(), "col");
diff --git
a/pinot-plugins/pinot-input-format/pinot-avro-base/src/test/java/org/apache/pinot/plugin/inputformat/avro/AvroUtilsTest.java
b/pinot-plugins/pinot-input-format/pinot-avro-base/src/test/java/org/apache/pinot/plugin/inputformat/avro/AvroUtilsTest.java
index 704de6ca58a..1250cc48120 100644
---
a/pinot-plugins/pinot-input-format/pinot-avro-base/src/test/java/org/apache/pinot/plugin/inputformat/avro/AvroUtilsTest.java
+++
b/pinot-plugins/pinot-input-format/pinot-avro-base/src/test/java/org/apache/pinot/plugin/inputformat/avro/AvroUtilsTest.java
@@ -22,6 +22,8 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.avro.LogicalType;
import org.apache.avro.LogicalTypes;
import org.apache.pinot.spi.config.table.ingestion.ComplexTypeConfig;
import org.apache.pinot.spi.data.FieldSpec;
@@ -32,6 +34,8 @@ import org.testng.annotations.Test;
import org.testng.collections.Lists;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
public class AvroUtilsTest {
@@ -160,6 +164,86 @@ public class AvroUtilsTest {
assertEquals(fieldSchema.getLogicalType().getName(), "uuid");
}
+ /// The generated Avro schema must describe the *logical* Pinot type.
Switching on the stored type instead emitted
+ /// `int` for BOOLEAN, a bare `long` for TIMESTAMP, and rejected BIG_DECIMAL
outright.
+ @Test
+ public void testGetAvroSchemaFromPinotSchemaUsesLogicalTypes() {
+ Schema pinotSchema = new Schema.SchemaBuilder()
+ .addSingleValueDimension("intCol", DataType.INT)
+ .addSingleValueDimension("longCol", DataType.LONG)
+ .addSingleValueDimension("floatCol", DataType.FLOAT)
+ .addSingleValueDimension("doubleCol", DataType.DOUBLE)
+ .addSingleValueDimension("boolCol", DataType.BOOLEAN)
+ .addSingleValueDimension("tsCol", DataType.TIMESTAMP)
+ .addSingleValueDimension("bigDecimalCol", DataType.BIG_DECIMAL)
+ .addSingleValueDimension("stringCol", DataType.STRING)
+ .addSingleValueDimension("jsonCol", DataType.JSON)
+ .addSingleValueDimension("bytesCol", DataType.BYTES)
+ .addSingleValueDimension("uuidCol", DataType.UUID)
+ .build();
+
+ org.apache.avro.Schema avroSchema =
AvroUtils.getAvroSchemaFromPinotSchema(pinotSchema);
+
+ assertFieldType(avroSchema, "intCol", org.apache.avro.Schema.Type.INT,
null);
+ assertFieldType(avroSchema, "longCol", org.apache.avro.Schema.Type.LONG,
null);
+ assertFieldType(avroSchema, "floatCol", org.apache.avro.Schema.Type.FLOAT,
null);
+ assertFieldType(avroSchema, "doubleCol",
org.apache.avro.Schema.Type.DOUBLE, null);
+ assertFieldType(avroSchema, "boolCol",
org.apache.avro.Schema.Type.BOOLEAN, null);
+ assertFieldType(avroSchema, "tsCol", org.apache.avro.Schema.Type.LONG,
"timestamp-millis");
+ assertFieldType(avroSchema, "bigDecimalCol",
org.apache.avro.Schema.Type.BYTES, "big-decimal");
+ assertFieldType(avroSchema, "stringCol",
org.apache.avro.Schema.Type.STRING, null);
+ assertFieldType(avroSchema, "jsonCol", org.apache.avro.Schema.Type.STRING,
null);
+ assertFieldType(avroSchema, "bytesCol", org.apache.avro.Schema.Type.BYTES,
null);
+ assertFieldType(avroSchema, "uuidCol", org.apache.avro.Schema.Type.STRING,
"uuid");
+ }
+
+ /// Multi-value columns become arrays of the same value schema. Before the
fix the MV switch only covered
+ /// INT/LONG/FLOAT/DOUBLE/STRING, so MV BYTES, BIG_DECIMAL, BOOLEAN and
TIMESTAMP all threw or lost their type.
+ @Test
+ public void testGetAvroSchemaFromPinotSchemaForMultiValueColumns() {
+ Schema pinotSchema = new Schema.SchemaBuilder()
+ .addMultiValueDimension("intCol", DataType.INT)
+ .addMultiValueDimension("boolCol", DataType.BOOLEAN)
+ .addMultiValueDimension("tsCol", DataType.TIMESTAMP)
+ .addMultiValueDimension("bigDecimalCol", DataType.BIG_DECIMAL)
+ .addMultiValueDimension("bytesCol", DataType.BYTES)
+ .addMultiValueDimension("uuidCol", DataType.UUID)
+ .build();
+
+ org.apache.avro.Schema avroSchema =
AvroUtils.getAvroSchemaFromPinotSchema(pinotSchema);
+
+ assertElementType(avroSchema, "intCol", org.apache.avro.Schema.Type.INT,
null);
+ assertElementType(avroSchema, "boolCol",
org.apache.avro.Schema.Type.BOOLEAN, null);
+ assertElementType(avroSchema, "tsCol", org.apache.avro.Schema.Type.LONG,
"timestamp-millis");
+ assertElementType(avroSchema, "bigDecimalCol",
org.apache.avro.Schema.Type.BYTES, "big-decimal");
+ assertElementType(avroSchema, "bytesCol",
org.apache.avro.Schema.Type.BYTES, null);
+ assertElementType(avroSchema, "uuidCol",
org.apache.avro.Schema.Type.STRING, "uuid");
+ }
+
+ private static void assertFieldType(org.apache.avro.Schema avroSchema,
String field,
+ org.apache.avro.Schema.Type expectedType, @Nullable String
expectedLogicalType) {
+ assertAvroType(avroSchema.getField(field).schema(), field, expectedType,
expectedLogicalType);
+ }
+
+ private static void assertElementType(org.apache.avro.Schema avroSchema,
String field,
+ org.apache.avro.Schema.Type expectedType, @Nullable String
expectedLogicalType) {
+ org.apache.avro.Schema fieldSchema = avroSchema.getField(field).schema();
+ assertEquals(fieldSchema.getType(), org.apache.avro.Schema.Type.ARRAY,
field + " must be an array");
+ assertAvroType(fieldSchema.getElementType(), field, expectedType,
expectedLogicalType);
+ }
+
+ private static void assertAvroType(org.apache.avro.Schema schema, String
field,
+ org.apache.avro.Schema.Type expectedType, @Nullable String
expectedLogicalType) {
+ assertEquals(schema.getType(), expectedType, "unexpected Avro type for " +
field);
+ LogicalType logicalType = LogicalTypes.fromSchemaIgnoreInvalid(schema);
+ if (expectedLogicalType == null) {
+ assertNull(logicalType, field + " must carry no logical type");
+ } else {
+ assertNotNull(logicalType, field + " must carry the " +
expectedLogicalType + " logical type");
+ assertEquals(logicalType.getName(), expectedLogicalType, "unexpected
logical type for " + field);
+ }
+ }
+
@Test
public void testGetPinotSchemaFromAvroSchemaWithUuidArray() {
org.apache.avro.Schema uuidSchema =
diff --git
a/pinot-plugins/pinot-segment-writer/pinot-segment-writer-file-based/src/test/java/org/apache/pinot/plugin/segmentwriter/filebased/FileBasedSegmentWriterTest.java
b/pinot-plugins/pinot-segment-writer/pinot-segment-writer-file-based/src/test/java/org/apache/pinot/plugin/segmentwriter/filebased/FileBasedSegmentWriterTest.java
index 57d6f62feeb..3f249fdd679 100644
---
a/pinot-plugins/pinot-segment-writer/pinot-segment-writer-file-based/src/test/java/org/apache/pinot/plugin/segmentwriter/filebased/FileBasedSegmentWriterTest.java
+++
b/pinot-plugins/pinot-segment-writer/pinot-segment-writer-file-based/src/test/java/org/apache/pinot/plugin/segmentwriter/filebased/FileBasedSegmentWriterTest.java
@@ -21,7 +21,9 @@ package org.apache.pinot.plugin.segmentwriter.filebased;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import java.io.File;
+import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
+import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -29,6 +31,7 @@ import java.util.List;
import java.util.Map;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.segment.local.segment.readers.PinotSegmentRecordReader;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
@@ -55,6 +58,10 @@ public class FileBasedSegmentWriterTest {
private static final String TABLE_NAME = "segmentWriter";
private static final String TIME_COLUMN_NAME = "aLong";
+ private static final long TIMESTAMP_VALUE = 1609491661001L;
+ // Beyond long precision and with a non-trivial scale, to prove the Avro
buffer's `big-decimal` logical type carries
+ // arbitrary precision and the value's own scale. Trailing zeros are avoided
because ingestion strips them.
+ private static final BigDecimal BIG_DECIMAL_VALUE = new
BigDecimal("-9999999999999999999999.12345");
private File _tmpDir;
private File _outputDir;
@@ -84,6 +91,8 @@ public class FileBasedSegmentWriterTest {
.addSingleValueDimension("anAdvancedMap_str",
FieldSpec.DataType.STRING)
.addSingleValueDimension("nullString", FieldSpec.DataType.STRING)
.addSingleValueDimension("aBoolean", FieldSpec.DataType.BOOLEAN)
+ .addSingleValueDimension("aTimestamp", FieldSpec.DataType.TIMESTAMP)
+ .addSingleValueDimension("aBigDecimal", FieldSpec.DataType.BIG_DECIMAL)
.addSingleValueDimension("aBytes", FieldSpec.DataType.BYTES)
.addMultiValueDimension("aStringList", FieldSpec.DataType.STRING)
.addMultiValueDimension("anIntList", FieldSpec.DataType.INT)
@@ -196,6 +205,41 @@ public class FileBasedSegmentWriterTest {
FileUtils.deleteQuietly(_outputDir);
}
+ /// The writer buffers rows in an Avro file and then builds the segment from
it, so every column has to survive a
+ /// full write/read round trip through the generated Avro schema and the
shared data model. This covers the logical
+ /// types that carry an Avro logical type (or, for BOOLEAN, a
stored-int-to-Boolean coercion) rather than being
+ /// written in their physical storage form.
+ @Test
+ public void testLogicalTypesRoundTripThroughAvroBuffer()
+ throws Exception {
+ FileUtils.deleteQuietly(_outputDir);
+ SegmentWriter segmentWriter = new FileBasedSegmentWriter();
+ segmentWriter.init(_tableConfig, _schema);
+ segmentWriter.collect(getGenericRow("record1", 1616238000000L));
+ segmentWriter.flush();
+ segmentWriter.close();
+
+ File segmentTar = new File(_outputDir,
"segmentWriter_1616238000000_1616238000000.tar.gz");
+ Assert.assertTrue(segmentTar.exists());
+ TarCompressionUtils.untar(segmentTar, _outputDir);
+ File segmentDir = new File(_outputDir,
"segmentWriter_1616238000000_1616238000000");
+
+ try (PinotSegmentRecordReader recordReader = new
PinotSegmentRecordReader(segmentDir)) {
+ Assert.assertTrue(recordReader.hasNext());
+ GenericRow row = recordReader.next();
+ // The segment stores Pinot's internal form: int 0/1 for BOOLEAN and
epoch millis for TIMESTAMP.
+ Assert.assertEquals(row.getValue("aBoolean"), 1);
+ Assert.assertEquals(row.getValue("aTimestamp"), TIMESTAMP_VALUE);
+ Assert.assertEquals(row.getValue("aBigDecimal"), BIG_DECIMAL_VALUE);
+ Assert.assertEquals(((BigDecimal) row.getValue("aBigDecimal")).scale(),
BIG_DECIMAL_VALUE.scale(),
+ "big-decimal must preserve the value's own scale");
+ Assert.assertEquals((byte[]) row.getValue("aBytes"),
"foo".getBytes(StandardCharsets.UTF_8));
+ Assert.assertEquals(row.getValue("aString"), "record1");
+ Assert.assertFalse(recordReader.hasNext());
+ }
+ FileUtils.deleteQuietly(_outputDir);
+ }
+
/**
* Tests flushing on empty collection
*/
@@ -381,6 +425,8 @@ public class FileBasedSegmentWriterTest {
row.putValue("aDouble", 10.5);
row.putValue("aFloat", 2.0);
row.putValue("aBoolean", true);
+ row.putValue("aTimestamp", new Timestamp(TIMESTAMP_VALUE));
+ row.putValue("aBigDecimal", BIG_DECIMAL_VALUE);
row.putValue("aBytes", "foo".getBytes(StandardCharsets.UTF_8));
List<String> stringList = new ArrayList<>();
stringList.add("a");
diff --git
a/pinot-tools/src/test/java/org/apache/pinot/tools/segment/converter/PinotSegmentConverterTest.java
b/pinot-tools/src/test/java/org/apache/pinot/tools/segment/converter/PinotSegmentConverterTest.java
index 666c06a4ba6..e6200c7e5b3 100644
---
a/pinot-tools/src/test/java/org/apache/pinot/tools/segment/converter/PinotSegmentConverterTest.java
+++
b/pinot-tools/src/test/java/org/apache/pinot/tools/segment/converter/PinotSegmentConverterTest.java
@@ -20,6 +20,8 @@ package org.apache.pinot.tools.segment.converter;
import java.io.File;
import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Timestamp;
import java.util.List;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.plugin.inputformat.avro.AvroRecordReader;
@@ -264,6 +266,79 @@ public class PinotSegmentConverterTest {
}
}
+ private static final String BOOLEAN_SV_COLUMN = "boolSVColumn";
+ private static final String TIMESTAMP_SV_COLUMN = "tsSVColumn";
+ private static final String BIG_DECIMAL_SV_COLUMN = "bigDecimalSVColumn";
+ private static final String BOOLEAN_MV_COLUMN = "boolMVColumn";
+ private static final String TIMESTAMP_MV_COLUMN = "tsMVColumn";
+ private static final long TIMESTAMP_VALUE = 1609491661001L;
+ // Beyond long precision and with a non-trivial scale, to prove the exported
Avro `big-decimal` carries arbitrary
+ // precision and the value's own scale. Trailing zeros are deliberately
avoided: ingestion's SpecialValueTransformer
+ // strips them, so a value like "123.45000" would already be "123.45" by the
time it reaches the segment.
+ private static final BigDecimal BIG_DECIMAL_VALUE = new
BigDecimal("-9999999999999999999999.12345");
+
+ /// Builds a segment with BOOLEAN, TIMESTAMP and BIG_DECIMAL columns and
converts it through both the Avro and
+ /// Parquet converters. These are exported using their Avro logical types
(`boolean`, `long{timestamp-millis}` and
+ /// `bytes{big-decimal}`) rather than their stored types, so this covers the
stored-int-to-Boolean coercion and the
+ /// big-decimal Conversion on both write paths — including Parquet's
distinct AvroParquetWriter.withDataModel path.
+ /// Before the fix, BOOLEAN exported as a bare `int`, TIMESTAMP as a bare
`long`, and BIG_DECIMAL threw.
+ @Test
+ public void testLogicalTypeConverters()
+ throws Exception {
+ Schema logicalSchema = new Schema.SchemaBuilder()
+ .addSingleValueDimension(BOOLEAN_SV_COLUMN, DataType.BOOLEAN)
+ .addSingleValueDimension(TIMESTAMP_SV_COLUMN, DataType.TIMESTAMP)
+ .addSingleValueDimension(BIG_DECIMAL_SV_COLUMN, DataType.BIG_DECIMAL)
+ .addMultiValueDimension(BOOLEAN_MV_COLUMN, DataType.BOOLEAN)
+ .addMultiValueDimension(TIMESTAMP_MV_COLUMN, DataType.TIMESTAMP)
+ .build();
+ TableConfig logicalTableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName("logicalTable").build();
+
+ GenericRow record = new GenericRow();
+ record.putValue(BOOLEAN_SV_COLUMN, true);
+ record.putValue(TIMESTAMP_SV_COLUMN, new Timestamp(TIMESTAMP_VALUE));
+ record.putValue(BIG_DECIMAL_SV_COLUMN, BIG_DECIMAL_VALUE);
+ record.putValue(BOOLEAN_MV_COLUMN, new Object[]{true, false});
+ record.putValue(TIMESTAMP_MV_COLUMN, new Object[]{new
Timestamp(TIMESTAMP_VALUE), new Timestamp(0L)});
+
+ SegmentGeneratorConfig config = new
SegmentGeneratorConfig(logicalTableConfig, logicalSchema);
+ config.setTableName("logicalTable");
+ config.setSegmentName("logicalSegment");
+ config.setOutDir(new File(TEMP_DIR, "logicalSegment").getPath());
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(List.of(record)));
+ driver.build();
+ String segmentDir = driver.getOutputDirectory().getPath();
+
+ File avroOut = new File(TEMP_DIR, "logicalSegment.avro");
+ new PinotSegmentToAvroConverter(segmentDir, avroOut.getPath()).convert();
+ try (AvroRecordReader reader = new AvroRecordReader()) {
+ reader.init(avroOut, logicalSchema.getFieldSpecMap().keySet(), null);
+ assertLogicalTypeRecord(reader.next());
+ assertFalse(reader.hasNext());
+ }
+
+ File parquetOut = new File(TEMP_DIR, "logicalSegment.parquet");
+ new PinotSegmentToParquetConverter(segmentDir,
parquetOut.getPath()).convert();
+ try (ParquetRecordReader reader = new ParquetRecordReader()) {
+ reader.init(parquetOut, logicalSchema.getFieldSpecMap().keySet(), null);
+ assertLogicalTypeRecord(reader.next());
+ assertFalse(reader.hasNext());
+ }
+ }
+
+ private static void assertLogicalTypeRecord(GenericRow record) {
+ // BOOLEAN reads back as a Boolean rather than the stored int, TIMESTAMP
as a Timestamp rather than a bare long.
+ assertEquals(record.getValue(BOOLEAN_SV_COLUMN), Boolean.TRUE);
+ assertEquals(record.getValue(TIMESTAMP_SV_COLUMN), new
Timestamp(TIMESTAMP_VALUE));
+ BigDecimal bigDecimal = (BigDecimal)
record.getValue(BIG_DECIMAL_SV_COLUMN);
+ assertEquals(bigDecimal, BIG_DECIMAL_VALUE);
+ assertEquals(bigDecimal.scale(), BIG_DECIMAL_VALUE.scale(), "big-decimal
must preserve the value's own scale");
+ assertEquals(record.getValue(BOOLEAN_MV_COLUMN), new
Object[]{Boolean.TRUE, Boolean.FALSE});
+ assertEquals(record.getValue(TIMESTAMP_MV_COLUMN),
+ new Object[]{new Timestamp(TIMESTAMP_VALUE), new Timestamp(0L)});
+ }
+
private static void assertUuidRecord(GenericRow record) {
// The reader may surface the uuid value as a String/UUID/byte[];
UuidUtils.toBytes normalizes all of them.
assertEquals(UuidUtils.toBytes(record.getValue(UUID_SV_COLUMN)),
UuidUtils.toBytes(UUID_SV_VALUE));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
