You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2019/09/27 23:00:33 UTC
[incubator-pinot] 01/01: Revert the binary-search change for
MapValue transform function
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch map_value
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit e26d0d20d9f0dfda86f9b61aedc3b254a33ff337
Author: Jackie (Xiaotian) Jiang <xa...@linkedin.com>
AuthorDate: Fri Sep 27 15:55:08 2019 -0700
Revert the binary-search change for MapValue transform function
Binary-search assumes ordered dictIds for key column, which will break for real-time case
Use linear-search for MapValue transform function
---
.../pinot/core/data/readers/AvroRecordReader.java | 2 +-
.../function/MapValueTransformFunction.java | 17 +++++-----
.../stream/AvroRecordToPinotRowGenerator.java | 2 +-
.../java/org/apache/pinot/core/util/AvroUtils.java | 36 +++++++++-------------
4 files changed, 26 insertions(+), 31 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/readers/AvroRecordReader.java b/pinot-core/src/main/java/org/apache/pinot/core/data/readers/AvroRecordReader.java
index 97d0718..bfa225f 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/readers/AvroRecordReader.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/readers/AvroRecordReader.java
@@ -82,7 +82,7 @@ public class AvroRecordReader implements RecordReader {
Object value = _reusableAvroRecord.get(fieldName);
// Allow default value for non-time columns
if (value != null || fieldSpec.getFieldType() != FieldSpec.FieldType.TIME) {
- AvroUtils.extractField(_schema, fieldSpec, _reusableAvroRecord, reuse);
+ AvroUtils.extractField(fieldSpec, _reusableAvroRecord, reuse);
}
}
return reuse;
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/MapValueTransformFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/MapValueTransformFunction.java
index 8e2e8fa..8b6f5c2 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/MapValueTransformFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/MapValueTransformFunction.java
@@ -19,7 +19,6 @@
package org.apache.pinot.core.operator.transform.function;
import com.google.common.base.Preconditions;
-import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.apache.pinot.core.common.DataSource;
@@ -103,13 +102,17 @@ public class MapValueTransformFunction extends BaseTransformFunction {
int length = projectionBlock.getNumDocs();
for (int i = 0; i < length; i++) {
int[] keyDictIds = keyDictIdsMV[i];
- int index = Arrays.binarySearch(keyDictIds, _keyDictId);
- if (index >= 0) {
- _dictIds[i] = valueDictIdsMV[i][index];
- } else {
- // Allow NULL_VALUE_INDEX for filter
- _dictIds[i] = Dictionary.NULL_VALUE_INDEX;
+
+ // Allow NULL_VALUE_INDEX for filter
+ int valueDictId = Dictionary.NULL_VALUE_INDEX;
+ int numValues = keyDictIds.length;
+ for (int j = 0; j < numValues; j++) {
+ if (keyDictIds[j] == _keyDictId) {
+ valueDictId = valueDictIdsMV[i][j];
+ break;
+ }
}
+ _dictIds[i] = valueDictId;
}
return _dictIds;
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/realtime/stream/AvroRecordToPinotRowGenerator.java b/pinot-core/src/main/java/org/apache/pinot/core/realtime/stream/AvroRecordToPinotRowGenerator.java
index 2e8de04..097b8d2 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/realtime/stream/AvroRecordToPinotRowGenerator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/realtime/stream/AvroRecordToPinotRowGenerator.java
@@ -44,7 +44,7 @@ public class AvroRecordToPinotRowGenerator {
for (FieldSpec fieldSpec : _schema.getAllFieldSpecs()) {
FieldSpec incomingFieldSpec =
fieldSpec.getFieldType() == FieldSpec.FieldType.TIME ? _incomingTimeFieldSpec : fieldSpec;
- AvroUtils.extractField(_schema, incomingFieldSpec, from, to);
+ AvroUtils.extractField(incomingFieldSpec, from, to);
}
return to;
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/util/AvroUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/util/AvroUtils.java
index 9610388..98466d5 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/util/AvroUtils.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/util/AvroUtils.java
@@ -24,7 +24,6 @@ import java.io.FileInputStream;
import java.io.IOException;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
@@ -42,7 +41,6 @@ import org.apache.pinot.common.data.Schema;
import org.apache.pinot.common.data.TimeFieldSpec;
import org.apache.pinot.core.data.GenericRow;
import org.apache.pinot.core.data.readers.RecordReaderUtils;
-import org.apache.pinot.core.data.recordtransformer.PinotDataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -305,36 +303,30 @@ public class AvroUtils {
}
@SuppressWarnings("unchecked")
- public static void extractField(Schema schema, FieldSpec fieldSpec, GenericRecord from, GenericRow to) {
+ public static void extractField(FieldSpec fieldSpec, GenericRecord from, GenericRow to) {
String fieldName = fieldSpec.getName();
- // Handle Map type
+ // Handle the Map type
if (fieldName.endsWith(MAP_KEY_COLUMN_SUFFIX)) {
String avroFieldName = fieldName.substring(0, fieldName.length() - MAP_KEY_COLUMN_SUFFIX.length());
Map map = (Map) from.get(avroFieldName);
- // Store sorted values in the key column
- PinotDataType keyType = PinotDataType.getPinotDataType(fieldSpec).getSingleValueType();
- TreeSet sortedKeys = new TreeSet();
- for (Object key : map.keySet()) {
- // Avro always store keys as string
- sortedKeys.add(keyType.convert(key.toString(), PinotDataType.STRING));
+ if (map != null) {
+ // Sort the keys so that the order is deterministic
+ TreeSet sortedKeys = new TreeSet(map.keySet());
+ to.putField(fieldName, RecordReaderUtils.convert(fieldSpec, sortedKeys));
+ return;
}
- to.putField(fieldName, RecordReaderUtils.convert(fieldSpec, sortedKeys));
} else if (fieldName.endsWith(MAP_VALUE_COLUMN_SUFFIX)) {
String avroFieldName = fieldName.substring(0, fieldName.length() - MAP_VALUE_COLUMN_SUFFIX.length());
Map map = (Map) from.get(avroFieldName);
- // Maintain the key-value pair order the same as the key column
- PinotDataType keyType =
- PinotDataType.getPinotDataType(schema.getFieldSpecFor(avroFieldName + MAP_KEY_COLUMN_SUFFIX))
- .getSingleValueType();
- TreeMap sortedMap = new TreeMap<>();
- for (Map.Entry entry : (Set<Map.Entry>) map.entrySet()) {
- // Avro always store keys as string
- sortedMap.put(keyType.convert(entry.getKey().toString(), PinotDataType.STRING), entry.getValue());
+ if (map != null) {
+ // Sort the keys so that the order is deterministic
+ TreeMap sortedMap = new TreeMap<>(map);
+ to.putField(fieldName, RecordReaderUtils.convert(fieldSpec, sortedMap.values()));
+ return;
}
- to.putField(fieldName, RecordReaderUtils.convert(fieldSpec, sortedMap.values()));
- } else {
- to.putField(fieldName, RecordReaderUtils.convert(fieldSpec, from.get(fieldName)));
}
+
+ to.putField(fieldName, RecordReaderUtils.convert(fieldSpec, from.get(fieldName)));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org