npawar commented on a change in pull request #5238: Evaluate schema transform
expressions during ingestion
URL: https://github.com/apache/incubator-pinot/pull/5238#discussion_r407768829
##########
File path:
pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/RecordReaderUtils.java
##########
@@ -67,167 +59,65 @@ public static InputStream getInputStream(File dataFile)
}
/**
- * Extracts all field specs from the given schema.
- * <p>For time field spec:
- * <ul>
- * <li>If incoming and outgoing time column name are the same, use
incoming time field spec</li>
- * <li>If incoming and outgoing time column name are different, put both
of them as time field spec</li>
- * <li>
- * We keep both incoming and outgoing time column to handle cases where
the input file contains time values that
- * are already converted
- * </li>
- * </ul>
+ * Converts the value to a multi-values value or a single values value
*/
- public static List<FieldSpec> extractFieldSpecs(Schema schema) {
- List<FieldSpec> fieldSpecs = new ArrayList<>();
- for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) {
- if (fieldSpec.getFieldType() == FieldSpec.FieldType.TIME) {
- TimeFieldSpec timeFieldSpec = (TimeFieldSpec) fieldSpec;
- fieldSpecs.add(new
TimeFieldSpec(timeFieldSpec.getIncomingGranularitySpec()));
- if
(!timeFieldSpec.getOutgoingTimeColumnName().equals(timeFieldSpec.getIncomingTimeColumnName()))
{
- fieldSpecs.add(new
TimeFieldSpec(timeFieldSpec.getOutgoingGranularitySpec()));
- }
- } else {
- fieldSpecs.add(fieldSpec);
- }
- }
- return fieldSpecs;
- }
+ public static @Nullable Object convert(@Nullable Object value) {
- /**
- * Converts the value based on the given field spec.
- */
- public static Object convert(FieldSpec fieldSpec, @Nullable Object value) {
- if (fieldSpec.isSingleValueField()) {
- return convertSingleValue(fieldSpec, value);
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof Collection) {
+ return convertMultiValue((Collection) value);
} else {
- return convertMultiValue(fieldSpec, (Collection) value);
+ return convertSingleValue(value);
}
}
/**
- * Converts the value to a single-valued value based on the given field spec.
+ * Converts the value to a single-valued value
*/
- public static Object convertSingleValue(FieldSpec fieldSpec, @Nullable
Object value) {
+ public static @Nullable Object convertSingleValue(@Nullable Object value) {
if (value == null) {
return null;
}
- DataType dataType = fieldSpec.getDataType();
- if (dataType == FieldSpec.DataType.BYTES) {
- // Avro ByteBuffer maps to byte[]
- if (value instanceof ByteBuffer) {
- ByteBuffer byteBufferValue = (ByteBuffer) value;
- // Use byteBufferValue.remaining() instead of
byteBufferValue.capacity() so that it still works when buffer is
- // over-sized
- byte[] bytesValue = new byte[byteBufferValue.remaining()];
- byteBufferValue.get(bytesValue);
- return bytesValue;
- } else {
- Preconditions
- .checkState(value instanceof byte[], "For BYTES data type, value
must be either ByteBuffer or byte[]");
- return value;
- }
- }
- if (value instanceof Number) {
- Number numberValue = (Number) value;
- switch (dataType) {
- case INT:
- return numberValue.intValue();
- case LONG:
- return numberValue.longValue();
- case FLOAT:
- return numberValue.floatValue();
- case DOUBLE:
- return numberValue.doubleValue();
- case STRING:
- return numberValue.toString();
- default:
- throw new IllegalStateException("Illegal data type: " + dataType);
- }
- }
- return convertSingleValue(fieldSpec, value.toString());
- }
+ if (value instanceof ByteBuffer) {
+ ByteBuffer byteBufferValue = (ByteBuffer) value;
- /**
- * Converts the string value to a single-valued value based on the given
field spec.
- */
- public static Object convertSingleValue(FieldSpec fieldSpec, @Nullable
String stringValue) {
- if (stringValue == null) {
- return null;
- }
- DataType dataType = fieldSpec.getDataType();
- // Treat empty string as null for data types other than STRING
- if (stringValue.isEmpty() && dataType != DataType.STRING) {
- return null;
+ // Use byteBufferValue.remaining() instead of byteBufferValue.capacity()
so that it still works when buffer is
+ // over-sized
+ byte[] bytesValue = new byte[byteBufferValue.remaining()];
+ byteBufferValue.get(bytesValue);
+ return bytesValue;
}
- switch (dataType) {
- case INT:
- return Integer.parseInt(stringValue);
- case LONG:
- return Long.parseLong(stringValue);
- case FLOAT:
- return Float.parseFloat(stringValue);
- case DOUBLE:
- return Double.parseDouble(stringValue);
- case STRING:
- return stringValue;
- case BYTES:
- return BytesUtils.toBytes(stringValue);
- default:
- throw new IllegalStateException("Illegal data type: " + dataType);
+ if (value instanceof Number) {
+ return value;
}
+ return value.toString();
Review comment:
I checked that this will be handled within DataTypeTransformer. There are
tests for this in the PinotDataTypeTest
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
