Jackie-Jiang commented on a change in pull request #6847:
URL: https://github.com/apache/incubator-pinot/pull/6847#discussion_r621630795
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/segment/processing/collector/ConcatCollector.java
##########
@@ -18,50 +18,167 @@
*/
package org.apache.pinot.core.segment.processing.collector;
+import com.google.common.base.Preconditions;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.util.GenericRowSerDeUtils;
+import org.apache.pinot.segment.local.segment.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
/**
- * A Collector implementation for collecting and concatenating all incoming
rows
+ * A Collector implementation for collecting and concatenating all incoming
rows.
*/
public class ConcatCollector implements Collector {
- private final List<GenericRow> _collection = new ArrayList<>();
- private final GenericRowSorter _sorter;
+ private static final String RECORDS_FILE_NAME = "collector.records";
+
+ private final CollectorConfig _collectorConfig;
+ private final List<FieldSpec> _fieldSpecs = new ArrayList<>();
+ private final Comparator<GenericRow> _genericRowComparator;
+ private int _numDocs;
+
+ private File _workingDir;
+ private File _collectorRecordFile;
+ // TODO: Avoid using BufferedOutputStream, and use ByteBuffer directly.
+ // However, ByteBuffer has a limitation that the size cannot exceed 2G.
+ // There are no limits on the size of data inserted into the {@link
Collector}.
+ // Hence, would need to implement a hybrid approach or a trigger a flush
when size exceeds on Collector.
+ private BufferedOutputStream _collectorRecordOutputStream;
+ private List<Long> _collectorRecordOffsets;
+ private PinotDataBuffer _collectorRecordBuffer;
public ConcatCollector(CollectorConfig collectorConfig, Schema schema) {
+
+ _collectorConfig = collectorConfig;
+ for (FieldSpec spec : schema.getAllFieldSpecs()) {
+ if (!spec.isVirtualColumn()) {
+ _fieldSpecs.add(spec);
+ }
+ }
List<String> sortOrder = collectorConfig.getSortOrder();
if (CollectionUtils.isNotEmpty(sortOrder)) {
- _sorter = new GenericRowSorter(sortOrder, schema);
+ GenericRowSorter sorter = new GenericRowSorter(sortOrder, schema);
+ _genericRowComparator = sorter.getGenericRowComparator();
} else {
- _sorter = null;
+ _genericRowComparator = null;
}
+
+ initializeBuffer();
+ }
+
+ private void initializeBuffer() {
+ _workingDir =
+ new File(FileUtils.getTempDirectory(),
String.format("concat_collector_%d", System.currentTimeMillis()));
+ Preconditions.checkState(_workingDir.mkdirs(), "Failed to create dir: %s
for %s with config: %s",
+ _workingDir.getAbsolutePath(), ConcatCollector.class.getSimpleName(),
_collectorConfig);
+
+ _collectorRecordFile = new File(_workingDir, RECORDS_FILE_NAME);
+ Preconditions.checkState(!_collectorRecordFile.exists(),
+ "Collector record file: " + _collectorRecordFile + " already exists");
+ try {
+ _collectorRecordOutputStream = new BufferedOutputStream(new
FileOutputStream(_collectorRecordFile));
+ } catch (FileNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ _collectorRecordOffsets = new ArrayList<>();
+ _collectorRecordOffsets.add(0L);
+ _numDocs = 0;
}
@Override
- public void collect(GenericRow genericRow) {
- _collection.add(genericRow);
+ public void collect(GenericRow genericRow)
+ throws IOException {
+ byte[] genericRowBytes =
GenericRowSerDeUtils.serializeGenericRow(genericRow, _fieldSpecs);
+ _collectorRecordOutputStream.write(genericRowBytes);
+ _collectorRecordOffsets.add(_collectorRecordOffsets.get(_numDocs) +
genericRowBytes.length);
+ _numDocs++;
}
@Override
- public Iterator<GenericRow> iterator() {
- if (_sorter != null) {
- _sorter.sort(_collection);
+ public Iterator<GenericRow> iterator()
+ throws IOException {
+
+ _collectorRecordOutputStream.flush();
+
+ int[] sortedDocIds = new int[_numDocs];
Review comment:
Do not create `sortedDocIds` for unsorted case
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/segment/processing/collector/ConcatCollector.java
##########
@@ -18,50 +18,167 @@
*/
package org.apache.pinot.core.segment.processing.collector;
+import com.google.common.base.Preconditions;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.util.GenericRowSerDeUtils;
+import org.apache.pinot.segment.local.segment.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
/**
- * A Collector implementation for collecting and concatenating all incoming
rows
+ * A Collector implementation for collecting and concatenating all incoming
rows.
*/
public class ConcatCollector implements Collector {
- private final List<GenericRow> _collection = new ArrayList<>();
- private final GenericRowSorter _sorter;
+ private static final String RECORDS_FILE_NAME = "collector.records";
+
+ private final CollectorConfig _collectorConfig;
+ private final List<FieldSpec> _fieldSpecs = new ArrayList<>();
+ private final Comparator<GenericRow> _genericRowComparator;
+ private int _numDocs;
+
+ private File _workingDir;
Review comment:
`_workingDir` and `_collectorRecordFile` can be final? We don't need to
create a separate path each time we initialize the buffer.
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/util/GenericRowSerDeUtils.java
##########
@@ -0,0 +1,326 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.util;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import org.apache.pinot.segment.local.segment.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.readers.GenericRow;
+
+
+/**
+ * Utility methods for serde of {@link GenericRow}
+ * Deserialization assumes it is deserializing from a {@link PinotDataBuffer}
+ */
+public final class GenericRowSerDeUtils {
+
+ private GenericRowSerDeUtils() {
+
+ }
+
+ /**
+ * Serialize the given GenericRow
+ * @param genericRow GenericRow to serialize
+ * @param fieldSpecs the fields to serialize
+ * @return serialized bytes
+ */
+ public static byte[] serializeGenericRow(GenericRow genericRow,
List<FieldSpec> fieldSpecs) {
+ int numBytes = 0;
+
+ for (FieldSpec fieldSpec : fieldSpecs) {
+ Object value = genericRow.getValue(fieldSpec.getName());
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType().getStoredType()) {
+
+ case INT:
+ numBytes += Integer.BYTES;
+ break;
+ case LONG:
+ numBytes += Long.BYTES;
+ break;
+ case FLOAT:
+ numBytes += Float.BYTES;
+ break;
+ case DOUBLE:
+ numBytes += Double.BYTES;
+ break;
+ case STRING:
+ byte[] stringBytes = ((String)
value).getBytes(StandardCharsets.UTF_8);
+ numBytes += Integer.BYTES; // string length
+ numBytes += stringBytes.length;
+ break;
+ case BYTES:
+ numBytes += Integer.BYTES; // byte array length
+ numBytes += ((byte[]) value).length;
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ } else {
+ Object[] multiValue = (Object[]) value;
+ numBytes += Integer.BYTES; // array length
+
+ switch (fieldSpec.getDataType().getStoredType()) {
+ case INT:
+ numBytes += Integer.BYTES * multiValue.length;
+ break;
+ case LONG:
+ numBytes += Long.BYTES * multiValue.length;
+ break;
+ case FLOAT:
+ numBytes += Float.BYTES * multiValue.length;
+ break;
+ case DOUBLE:
+ numBytes += Double.BYTES * multiValue.length;
+ break;
+ case STRING:
+ for (Object element : multiValue) {
+ byte[] stringBytes = ((String)
element).getBytes(StandardCharsets.UTF_8);
+ numBytes += Integer.BYTES; // string length
+ numBytes += stringBytes.length;
+ }
+ break;
+ case BYTES:
+ for (Object element : multiValue) {
+ numBytes += Integer.BYTES; // byte array length
+ numBytes += ((byte[]) element).length;
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ }
+ }
+
+ byte[] genericRowBytes = new byte[numBytes];
+ ByteBuffer byteBuffer =
ByteBuffer.wrap(genericRowBytes).order(PinotDataBuffer.NATIVE_ORDER);
+
+ for (FieldSpec fieldSpec : fieldSpecs) {
+ Object value = genericRow.getValue(fieldSpec.getName());
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType()) {
Review comment:
Use stored-type here and remove BOOLEAN
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/util/GenericRowSerDeUtils.java
##########
@@ -0,0 +1,326 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.util;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import org.apache.pinot.segment.local.segment.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.readers.GenericRow;
+
+
+/**
+ * Utility methods for serde of {@link GenericRow}
+ * Deserialization assumes it is deserializing from a {@link PinotDataBuffer}
+ */
+public final class GenericRowSerDeUtils {
+
+ private GenericRowSerDeUtils() {
+
+ }
+
+ /**
+ * Serialize the given GenericRow
+ * @param genericRow GenericRow to serialize
+ * @param fieldSpecs the fields to serialize
+ * @return serialized bytes
+ */
+ public static byte[] serializeGenericRow(GenericRow genericRow,
List<FieldSpec> fieldSpecs) {
+ int numBytes = 0;
+
+ for (FieldSpec fieldSpec : fieldSpecs) {
+ Object value = genericRow.getValue(fieldSpec.getName());
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType().getStoredType()) {
+
+ case INT:
+ numBytes += Integer.BYTES;
+ break;
+ case LONG:
+ numBytes += Long.BYTES;
+ break;
+ case FLOAT:
+ numBytes += Float.BYTES;
+ break;
+ case DOUBLE:
+ numBytes += Double.BYTES;
+ break;
+ case STRING:
+ byte[] stringBytes = ((String)
value).getBytes(StandardCharsets.UTF_8);
+ numBytes += Integer.BYTES; // string length
+ numBytes += stringBytes.length;
+ break;
+ case BYTES:
+ numBytes += Integer.BYTES; // byte array length
+ numBytes += ((byte[]) value).length;
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ } else {
+ Object[] multiValue = (Object[]) value;
+ numBytes += Integer.BYTES; // array length
+
+ switch (fieldSpec.getDataType().getStoredType()) {
+ case INT:
+ numBytes += Integer.BYTES * multiValue.length;
+ break;
+ case LONG:
+ numBytes += Long.BYTES * multiValue.length;
+ break;
+ case FLOAT:
+ numBytes += Float.BYTES * multiValue.length;
+ break;
+ case DOUBLE:
+ numBytes += Double.BYTES * multiValue.length;
+ break;
+ case STRING:
+ for (Object element : multiValue) {
+ byte[] stringBytes = ((String)
element).getBytes(StandardCharsets.UTF_8);
+ numBytes += Integer.BYTES; // string length
+ numBytes += stringBytes.length;
+ }
+ break;
+ case BYTES:
+ for (Object element : multiValue) {
+ numBytes += Integer.BYTES; // byte array length
+ numBytes += ((byte[]) element).length;
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ }
+ }
+
+ byte[] genericRowBytes = new byte[numBytes];
+ ByteBuffer byteBuffer =
ByteBuffer.wrap(genericRowBytes).order(PinotDataBuffer.NATIVE_ORDER);
+
+ for (FieldSpec fieldSpec : fieldSpecs) {
+ Object value = genericRow.getValue(fieldSpec.getName());
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType()) {
+
+ case INT:
+ byteBuffer.putInt((int) value);
+ break;
+ case LONG:
+ byteBuffer.putLong((long) value);
+ break;
+ case FLOAT:
+ byteBuffer.putFloat((float) value);
+ break;
+ case DOUBLE:
+ byteBuffer.putDouble((double) value);
+ break;
+ case BOOLEAN:
+ case STRING:
+ byte[] stringBytes = ((String)
value).getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(stringBytes.length);
+ byteBuffer.put(stringBytes);
+ break;
+ case BYTES:
+ byte[] bytes = (byte[]) value;
+ byteBuffer.putInt(bytes.length);
+ byteBuffer.put(bytes);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ } else {
+ Object[] multiValue = (Object[]) value;
+ byteBuffer.putInt(multiValue.length);
+
+ switch (fieldSpec.getDataType()) {
+
+ case INT:
+ for (Object element : multiValue) {
+ byteBuffer.putInt((int) element);
+ }
+ break;
+ case LONG:
+ for (Object element : multiValue) {
+ byteBuffer.putLong((long) element);
+ }
+ break;
+ case FLOAT:
+ for (Object element : multiValue) {
+ byteBuffer.putFloat((float) element);
+ }
+ break;
+ case DOUBLE:
+ for (Object element : multiValue) {
+ byteBuffer.putDouble((double) element);
+ }
+ break;
+ case BOOLEAN:
+ case STRING:
+ for (Object element : multiValue) {
+ byte[] stringBytes = ((String)
element).getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(stringBytes.length);
+ byteBuffer.put(stringBytes);
+ }
+ break;
+ case BYTES:
+ for (Object element : multiValue) {
+ byte[] bytes = (byte[]) element;
+ byteBuffer.putInt(bytes.length);
+ byteBuffer.put(bytes);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ }
+ }
+ return genericRowBytes;
+ }
+
+ /**
+ * Deserializes bytes from the buffer to GenericRow
+ * @param dataBuffer the pinot data buffer
+ * @param offset offset to begin reading from
+ * @param fieldSpecs list of field specs to determine fields in
deserialization
+ * @param reuse GenericRow object for returning
+ * @return Deserialized GenericRow
+ */
+ public static GenericRow deserializeGenericRow(PinotDataBuffer dataBuffer,
long offset, List<FieldSpec> fieldSpecs,
+ GenericRow reuse) {
+ for (FieldSpec fieldSpec : fieldSpecs) {
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType().getStoredType()) {
+
+ case INT:
+ int intValue = dataBuffer.getInt(offset);
+ reuse.putValue(fieldSpec.getName(), intValue);
Review comment:
Put `fieldSpec.getName()` in a local variable
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/segment/processing/collector/ConcatCollector.java
##########
@@ -18,50 +18,167 @@
*/
package org.apache.pinot.core.segment.processing.collector;
+import com.google.common.base.Preconditions;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.util.GenericRowSerDeUtils;
+import org.apache.pinot.segment.local.segment.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
/**
- * A Collector implementation for collecting and concatenating all incoming
rows
+ * A Collector implementation for collecting and concatenating all incoming
rows.
*/
public class ConcatCollector implements Collector {
- private final List<GenericRow> _collection = new ArrayList<>();
- private final GenericRowSorter _sorter;
+ private static final String RECORDS_FILE_NAME = "collector.records";
+
+ private final CollectorConfig _collectorConfig;
+ private final List<FieldSpec> _fieldSpecs = new ArrayList<>();
+ private final Comparator<GenericRow> _genericRowComparator;
+ private int _numDocs;
+
+ private File _workingDir;
+ private File _collectorRecordFile;
+ // TODO: Avoid using BufferedOutputStream, and use ByteBuffer directly.
+ // However, ByteBuffer has a limitation that the size cannot exceed 2G.
+ // There are no limits on the size of data inserted into the {@link
Collector}.
+ // Hence, would need to implement a hybrid approach or a trigger a flush
when size exceeds on Collector.
+ private BufferedOutputStream _collectorRecordOutputStream;
+ private List<Long> _collectorRecordOffsets;
+ private PinotDataBuffer _collectorRecordBuffer;
public ConcatCollector(CollectorConfig collectorConfig, Schema schema) {
+
+ _collectorConfig = collectorConfig;
+ for (FieldSpec spec : schema.getAllFieldSpecs()) {
+ if (!spec.isVirtualColumn()) {
+ _fieldSpecs.add(spec);
+ }
+ }
List<String> sortOrder = collectorConfig.getSortOrder();
if (CollectionUtils.isNotEmpty(sortOrder)) {
- _sorter = new GenericRowSorter(sortOrder, schema);
+ GenericRowSorter sorter = new GenericRowSorter(sortOrder, schema);
+ _genericRowComparator = sorter.getGenericRowComparator();
} else {
- _sorter = null;
+ _genericRowComparator = null;
}
+
+ initializeBuffer();
+ }
+
+ private void initializeBuffer() {
+ _workingDir =
+ new File(FileUtils.getTempDirectory(),
String.format("concat_collector_%d", System.currentTimeMillis()));
+ Preconditions.checkState(_workingDir.mkdirs(), "Failed to create dir: %s
for %s with config: %s",
+ _workingDir.getAbsolutePath(), ConcatCollector.class.getSimpleName(),
_collectorConfig);
+
+ _collectorRecordFile = new File(_workingDir, RECORDS_FILE_NAME);
+ Preconditions.checkState(!_collectorRecordFile.exists(),
+ "Collector record file: " + _collectorRecordFile + " already exists");
+ try {
+ _collectorRecordOutputStream = new BufferedOutputStream(new
FileOutputStream(_collectorRecordFile));
+ } catch (FileNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ _collectorRecordOffsets = new ArrayList<>();
+ _collectorRecordOffsets.add(0L);
+ _numDocs = 0;
}
@Override
- public void collect(GenericRow genericRow) {
- _collection.add(genericRow);
+ public void collect(GenericRow genericRow)
+ throws IOException {
+ byte[] genericRowBytes =
GenericRowSerDeUtils.serializeGenericRow(genericRow, _fieldSpecs);
+ _collectorRecordOutputStream.write(genericRowBytes);
+ _collectorRecordOffsets.add(_collectorRecordOffsets.get(_numDocs) +
genericRowBytes.length);
+ _numDocs++;
}
@Override
- public Iterator<GenericRow> iterator() {
- if (_sorter != null) {
- _sorter.sort(_collection);
+ public Iterator<GenericRow> iterator()
+ throws IOException {
+
+ _collectorRecordOutputStream.flush();
+
+ int[] sortedDocIds = new int[_numDocs];
+ for (int i = 0; i < _numDocs; i++) {
+ sortedDocIds[i] = i;
}
- return _collection.iterator();
+
+ _collectorRecordBuffer = PinotDataBuffer
+ .mapFile(_collectorRecordFile, true, 0,
_collectorRecordOffsets.get(_numDocs), PinotDataBuffer.NATIVE_ORDER,
+ "ConcatCollector: generic row buffer");
+
+
+ // TODO: A lot of this code can be made common across Collectors, once
{@link RollupCollector} is also converted to off heap implementation
+ if (_genericRowComparator != null) {
+ it.unimi.dsi.fastutil.Arrays.quickSort(0, _numDocs, (i1, i2) -> {
Review comment:
Import `it.unimi.dsi.fastutil.Arrays`
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/util/GenericRowSerDeUtils.java
##########
@@ -0,0 +1,326 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.util;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import org.apache.pinot.segment.local.segment.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.readers.GenericRow;
+
+
+/**
+ * Utility methods for serde of {@link GenericRow}
+ * Deserialization assumes it is deserializing from a {@link PinotDataBuffer}
+ */
+public final class GenericRowSerDeUtils {
+
+ private GenericRowSerDeUtils() {
+
+ }
+
+ /**
+ * Serialize the given GenericRow
+ * @param genericRow GenericRow to serialize
+ * @param fieldSpecs the fields to serialize
+ * @return serialized bytes
+ */
+ public static byte[] serializeGenericRow(GenericRow genericRow,
List<FieldSpec> fieldSpecs) {
+ int numBytes = 0;
+
+ for (FieldSpec fieldSpec : fieldSpecs) {
+ Object value = genericRow.getValue(fieldSpec.getName());
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType().getStoredType()) {
+
+ case INT:
+ numBytes += Integer.BYTES;
+ break;
+ case LONG:
+ numBytes += Long.BYTES;
+ break;
+ case FLOAT:
+ numBytes += Float.BYTES;
+ break;
+ case DOUBLE:
+ numBytes += Double.BYTES;
+ break;
+ case STRING:
+ byte[] stringBytes = ((String)
value).getBytes(StandardCharsets.UTF_8);
+ numBytes += Integer.BYTES; // string length
+ numBytes += stringBytes.length;
+ break;
+ case BYTES:
+ numBytes += Integer.BYTES; // byte array length
+ numBytes += ((byte[]) value).length;
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ } else {
+ Object[] multiValue = (Object[]) value;
+ numBytes += Integer.BYTES; // array length
+
+ switch (fieldSpec.getDataType().getStoredType()) {
+ case INT:
+ numBytes += Integer.BYTES * multiValue.length;
+ break;
+ case LONG:
+ numBytes += Long.BYTES * multiValue.length;
+ break;
+ case FLOAT:
+ numBytes += Float.BYTES * multiValue.length;
+ break;
+ case DOUBLE:
+ numBytes += Double.BYTES * multiValue.length;
+ break;
+ case STRING:
+ for (Object element : multiValue) {
+ byte[] stringBytes = ((String)
element).getBytes(StandardCharsets.UTF_8);
+ numBytes += Integer.BYTES; // string length
+ numBytes += stringBytes.length;
+ }
+ break;
+ case BYTES:
+ for (Object element : multiValue) {
+ numBytes += Integer.BYTES; // byte array length
+ numBytes += ((byte[]) element).length;
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ }
+ }
+
+ byte[] genericRowBytes = new byte[numBytes];
+ ByteBuffer byteBuffer =
ByteBuffer.wrap(genericRowBytes).order(PinotDataBuffer.NATIVE_ORDER);
+
+ for (FieldSpec fieldSpec : fieldSpecs) {
+ Object value = genericRow.getValue(fieldSpec.getName());
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType()) {
+
+ case INT:
+ byteBuffer.putInt((int) value);
+ break;
+ case LONG:
+ byteBuffer.putLong((long) value);
+ break;
+ case FLOAT:
+ byteBuffer.putFloat((float) value);
+ break;
+ case DOUBLE:
+ byteBuffer.putDouble((double) value);
+ break;
+ case BOOLEAN:
+ case STRING:
+ byte[] stringBytes = ((String)
value).getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(stringBytes.length);
+ byteBuffer.put(stringBytes);
+ break;
+ case BYTES:
+ byte[] bytes = (byte[]) value;
+ byteBuffer.putInt(bytes.length);
+ byteBuffer.put(bytes);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ } else {
+ Object[] multiValue = (Object[]) value;
+ byteBuffer.putInt(multiValue.length);
+
+ switch (fieldSpec.getDataType()) {
+
+ case INT:
+ for (Object element : multiValue) {
+ byteBuffer.putInt((int) element);
+ }
+ break;
+ case LONG:
+ for (Object element : multiValue) {
+ byteBuffer.putLong((long) element);
+ }
+ break;
+ case FLOAT:
+ for (Object element : multiValue) {
+ byteBuffer.putFloat((float) element);
+ }
+ break;
+ case DOUBLE:
+ for (Object element : multiValue) {
+ byteBuffer.putDouble((double) element);
+ }
+ break;
+ case BOOLEAN:
+ case STRING:
+ for (Object element : multiValue) {
+ byte[] stringBytes = ((String)
element).getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(stringBytes.length);
+ byteBuffer.put(stringBytes);
+ }
+ break;
+ case BYTES:
+ for (Object element : multiValue) {
+ byte[] bytes = (byte[]) element;
+ byteBuffer.putInt(bytes.length);
+ byteBuffer.put(bytes);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ }
+ }
+ return genericRowBytes;
+ }
+
+ /**
+ * Deserializes bytes from the buffer to GenericRow
Review comment:
Add javadoc about expecting native order data buffer
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/util/GenericRowSerDeUtils.java
##########
@@ -0,0 +1,326 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.util;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import org.apache.pinot.segment.local.segment.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.readers.GenericRow;
+
+
+/**
+ * Utility methods for serde of {@link GenericRow}
+ * Deserialization assumes it is deserializing from a {@link PinotDataBuffer}
+ */
+public final class GenericRowSerDeUtils {
+
+ private GenericRowSerDeUtils() {
+
+ }
+
+ /**
+ * Serialize the given GenericRow
+ * @param genericRow GenericRow to serialize
+ * @param fieldSpecs the fields to serialize
+ * @return serialized bytes
+ */
+ public static byte[] serializeGenericRow(GenericRow genericRow,
List<FieldSpec> fieldSpecs) {
+ int numBytes = 0;
+
+ for (FieldSpec fieldSpec : fieldSpecs) {
+ Object value = genericRow.getValue(fieldSpec.getName());
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType().getStoredType()) {
+
+ case INT:
+ numBytes += Integer.BYTES;
+ break;
+ case LONG:
+ numBytes += Long.BYTES;
+ break;
+ case FLOAT:
+ numBytes += Float.BYTES;
+ break;
+ case DOUBLE:
+ numBytes += Double.BYTES;
+ break;
+ case STRING:
+ byte[] stringBytes = ((String)
value).getBytes(StandardCharsets.UTF_8);
+ numBytes += Integer.BYTES; // string length
+ numBytes += stringBytes.length;
+ break;
+ case BYTES:
+ numBytes += Integer.BYTES; // byte array length
+ numBytes += ((byte[]) value).length;
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ } else {
+ Object[] multiValue = (Object[]) value;
+ numBytes += Integer.BYTES; // array length
+
+ switch (fieldSpec.getDataType().getStoredType()) {
+ case INT:
+ numBytes += Integer.BYTES * multiValue.length;
+ break;
+ case LONG:
+ numBytes += Long.BYTES * multiValue.length;
+ break;
+ case FLOAT:
+ numBytes += Float.BYTES * multiValue.length;
+ break;
+ case DOUBLE:
+ numBytes += Double.BYTES * multiValue.length;
+ break;
+ case STRING:
+ for (Object element : multiValue) {
+ byte[] stringBytes = ((String)
element).getBytes(StandardCharsets.UTF_8);
+ numBytes += Integer.BYTES; // string length
+ numBytes += stringBytes.length;
+ }
+ break;
+ case BYTES:
+ for (Object element : multiValue) {
+ numBytes += Integer.BYTES; // byte array length
+ numBytes += ((byte[]) element).length;
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ }
+ }
+
+ byte[] genericRowBytes = new byte[numBytes];
+ ByteBuffer byteBuffer =
ByteBuffer.wrap(genericRowBytes).order(PinotDataBuffer.NATIVE_ORDER);
+
+ for (FieldSpec fieldSpec : fieldSpecs) {
+ Object value = genericRow.getValue(fieldSpec.getName());
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType()) {
+
+ case INT:
+ byteBuffer.putInt((int) value);
+ break;
+ case LONG:
+ byteBuffer.putLong((long) value);
+ break;
+ case FLOAT:
+ byteBuffer.putFloat((float) value);
+ break;
+ case DOUBLE:
+ byteBuffer.putDouble((double) value);
+ break;
+ case BOOLEAN:
+ case STRING:
+ byte[] stringBytes = ((String)
value).getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(stringBytes.length);
+ byteBuffer.put(stringBytes);
+ break;
+ case BYTES:
+ byte[] bytes = (byte[]) value;
+ byteBuffer.putInt(bytes.length);
+ byteBuffer.put(bytes);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ } else {
+ Object[] multiValue = (Object[]) value;
+ byteBuffer.putInt(multiValue.length);
+
+ switch (fieldSpec.getDataType()) {
+
+ case INT:
+ for (Object element : multiValue) {
+ byteBuffer.putInt((int) element);
+ }
+ break;
+ case LONG:
+ for (Object element : multiValue) {
+ byteBuffer.putLong((long) element);
+ }
+ break;
+ case FLOAT:
+ for (Object element : multiValue) {
+ byteBuffer.putFloat((float) element);
+ }
+ break;
+ case DOUBLE:
+ for (Object element : multiValue) {
+ byteBuffer.putDouble((double) element);
+ }
+ break;
+ case BOOLEAN:
+ case STRING:
+ for (Object element : multiValue) {
+ byte[] stringBytes = ((String)
element).getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(stringBytes.length);
+ byteBuffer.put(stringBytes);
+ }
+ break;
+ case BYTES:
+ for (Object element : multiValue) {
+ byte[] bytes = (byte[]) element;
+ byteBuffer.putInt(bytes.length);
+ byteBuffer.put(bytes);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ }
+ }
+ return genericRowBytes;
+ }
+
+ /**
+ * Deserializes bytes from the buffer to GenericRow
+ * @param dataBuffer the pinot data buffer
+ * @param offset offset to begin reading from
+ * @param fieldSpecs list of field specs to determine fields in
deserialization
+ * @param reuse GenericRow object for returning
+ * @return Deserialized GenericRow
+ */
+ public static GenericRow deserializeGenericRow(PinotDataBuffer dataBuffer,
long offset, List<FieldSpec> fieldSpecs,
+ GenericRow reuse) {
+ for (FieldSpec fieldSpec : fieldSpecs) {
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType().getStoredType()) {
+
+ case INT:
+ int intValue = dataBuffer.getInt(offset);
+ reuse.putValue(fieldSpec.getName(), intValue);
+ offset += Integer.BYTES;
+ break;
+ case LONG:
+ long longValue = dataBuffer.getLong(offset);
+ reuse.putValue(fieldSpec.getName(), longValue);
+ offset += Long.BYTES;
+ break;
+ case FLOAT:
+ float floatValue = dataBuffer.getFloat(offset);
+ reuse.putValue(fieldSpec.getName(), floatValue);
+ offset += Float.BYTES;
+ break;
+ case DOUBLE:
+ double doubleValue = dataBuffer.getDouble(offset);
+ reuse.putValue(fieldSpec.getName(), doubleValue);
+ offset += Double.BYTES;
+ break;
+ case STRING:
+ int stringSize = dataBuffer.getInt(offset);
+ offset += Integer.BYTES;
+ byte[] stringBytes = new byte[stringSize];
+ for (int j = 0; j < stringSize; j++) {
Review comment:
You can use `i` instead of `j` here. Same for other for statements
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/util/GenericRowSerDeUtils.java
##########
@@ -0,0 +1,326 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.util;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import org.apache.pinot.segment.local.segment.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.readers.GenericRow;
+
+
+/**
+ * Utility methods for serde of {@link GenericRow}
+ * Deserialization assumes it is deserializing from a {@link PinotDataBuffer}
+ */
+public final class GenericRowSerDeUtils {
+
+ private GenericRowSerDeUtils() {
+
+ }
+
+ /**
+ * Serialize the given GenericRow
+ * @param genericRow GenericRow to serialize
+ * @param fieldSpecs the fields to serialize
+ * @return serialized bytes
+ */
+ public static byte[] serializeGenericRow(GenericRow genericRow,
List<FieldSpec> fieldSpecs) {
Review comment:
Suggest merging the length calculation and the serialize so that we only
need one switch case, and no need to do UTF-8 encoding twice. Don't see much
value breaking them into 2 passes
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/segment/processing/collector/GenericRowSorter.java
##########
@@ -81,6 +81,10 @@ public GenericRowSorter(List<String> sortOrder, Schema
schema) {
};
}
+ public Comparator<GenericRow> getGenericRowComparator() {
Review comment:
(Optional) I would probably replace the `GenericRowSorter` class with
`GenericRowComparator` class
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/util/GenericRowSerDeUtils.java
##########
@@ -0,0 +1,326 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.util;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import org.apache.pinot.segment.local.segment.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.readers.GenericRow;
+
+
+/**
+ * Utility methods for serde of {@link GenericRow}
+ * Deserialization assumes it is deserializing from a {@link PinotDataBuffer}
+ */
+public final class GenericRowSerDeUtils {
+
+ private GenericRowSerDeUtils() {
+
+ }
+
+ /**
+ * Serialize the given GenericRow
Review comment:
Add javadoc about storing the data in native order
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/util/GenericRowSerDeUtils.java
##########
@@ -0,0 +1,326 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.util;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import org.apache.pinot.segment.local.segment.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.readers.GenericRow;
+
+
+/**
+ * Utility methods for serde of {@link GenericRow}
+ * Deserialization assumes it is deserializing from a {@link PinotDataBuffer}
+ */
+public final class GenericRowSerDeUtils {
+
+ private GenericRowSerDeUtils() {
+
+ }
+
+ /**
+ * Serialize the given GenericRow
+ * @param genericRow GenericRow to serialize
+ * @param fieldSpecs the fields to serialize
+ * @return serialized bytes
+ */
+ public static byte[] serializeGenericRow(GenericRow genericRow,
List<FieldSpec> fieldSpecs) {
+ int numBytes = 0;
+
+ for (FieldSpec fieldSpec : fieldSpecs) {
+ Object value = genericRow.getValue(fieldSpec.getName());
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType().getStoredType()) {
+
+ case INT:
+ numBytes += Integer.BYTES;
+ break;
+ case LONG:
+ numBytes += Long.BYTES;
+ break;
+ case FLOAT:
+ numBytes += Float.BYTES;
+ break;
+ case DOUBLE:
+ numBytes += Double.BYTES;
+ break;
+ case STRING:
+ byte[] stringBytes = ((String)
value).getBytes(StandardCharsets.UTF_8);
+ numBytes += Integer.BYTES; // string length
+ numBytes += stringBytes.length;
+ break;
+ case BYTES:
+ numBytes += Integer.BYTES; // byte array length
+ numBytes += ((byte[]) value).length;
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ } else {
+ Object[] multiValue = (Object[]) value;
+ numBytes += Integer.BYTES; // array length
+
+ switch (fieldSpec.getDataType().getStoredType()) {
+ case INT:
+ numBytes += Integer.BYTES * multiValue.length;
+ break;
+ case LONG:
+ numBytes += Long.BYTES * multiValue.length;
+ break;
+ case FLOAT:
+ numBytes += Float.BYTES * multiValue.length;
+ break;
+ case DOUBLE:
+ numBytes += Double.BYTES * multiValue.length;
+ break;
+ case STRING:
+ for (Object element : multiValue) {
+ byte[] stringBytes = ((String)
element).getBytes(StandardCharsets.UTF_8);
+ numBytes += Integer.BYTES; // string length
+ numBytes += stringBytes.length;
+ }
+ break;
+ case BYTES:
+ for (Object element : multiValue) {
+ numBytes += Integer.BYTES; // byte array length
+ numBytes += ((byte[]) element).length;
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ }
+ }
+
+ byte[] genericRowBytes = new byte[numBytes];
+ ByteBuffer byteBuffer =
ByteBuffer.wrap(genericRowBytes).order(PinotDataBuffer.NATIVE_ORDER);
+
+ for (FieldSpec fieldSpec : fieldSpecs) {
+ Object value = genericRow.getValue(fieldSpec.getName());
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType()) {
+
+ case INT:
+ byteBuffer.putInt((int) value);
+ break;
+ case LONG:
+ byteBuffer.putLong((long) value);
+ break;
+ case FLOAT:
+ byteBuffer.putFloat((float) value);
+ break;
+ case DOUBLE:
+ byteBuffer.putDouble((double) value);
+ break;
+ case BOOLEAN:
+ case STRING:
+ byte[] stringBytes = ((String)
value).getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(stringBytes.length);
+ byteBuffer.put(stringBytes);
+ break;
+ case BYTES:
+ byte[] bytes = (byte[]) value;
+ byteBuffer.putInt(bytes.length);
+ byteBuffer.put(bytes);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ } else {
+ Object[] multiValue = (Object[]) value;
+ byteBuffer.putInt(multiValue.length);
+
+ switch (fieldSpec.getDataType()) {
+
+ case INT:
+ for (Object element : multiValue) {
+ byteBuffer.putInt((int) element);
+ }
+ break;
+ case LONG:
+ for (Object element : multiValue) {
+ byteBuffer.putLong((long) element);
+ }
+ break;
+ case FLOAT:
+ for (Object element : multiValue) {
+ byteBuffer.putFloat((float) element);
+ }
+ break;
+ case DOUBLE:
+ for (Object element : multiValue) {
+ byteBuffer.putDouble((double) element);
+ }
+ break;
+ case BOOLEAN:
+ case STRING:
+ for (Object element : multiValue) {
+ byte[] stringBytes = ((String)
element).getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(stringBytes.length);
+ byteBuffer.put(stringBytes);
+ }
+ break;
+ case BYTES:
+ for (Object element : multiValue) {
+ byte[] bytes = (byte[]) element;
+ byteBuffer.putInt(bytes.length);
+ byteBuffer.put(bytes);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("DataType '%s' not supported",
fieldSpec.getDataType()));
+ }
+ }
+ }
+ return genericRowBytes;
+ }
+
+ /**
+ * Deserializes bytes from the buffer to GenericRow
+ * @param dataBuffer the pinot data buffer
+ * @param offset offset to begin reading from
+ * @param fieldSpecs list of field specs to determine fields in
deserialization
+ * @param reuse GenericRow object for returning
+ * @return Deserialized GenericRow
+ */
+ public static GenericRow deserializeGenericRow(PinotDataBuffer dataBuffer,
long offset, List<FieldSpec> fieldSpecs,
+ GenericRow reuse) {
+ for (FieldSpec fieldSpec : fieldSpecs) {
+
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType().getStoredType()) {
+
+ case INT:
+ int intValue = dataBuffer.getInt(offset);
+ reuse.putValue(fieldSpec.getName(), intValue);
+ offset += Integer.BYTES;
+ break;
+ case LONG:
+ long longValue = dataBuffer.getLong(offset);
+ reuse.putValue(fieldSpec.getName(), longValue);
+ offset += Long.BYTES;
+ break;
+ case FLOAT:
+ float floatValue = dataBuffer.getFloat(offset);
+ reuse.putValue(fieldSpec.getName(), floatValue);
+ offset += Float.BYTES;
+ break;
+ case DOUBLE:
+ double doubleValue = dataBuffer.getDouble(offset);
+ reuse.putValue(fieldSpec.getName(), doubleValue);
+ offset += Double.BYTES;
+ break;
+ case STRING:
+ int stringSize = dataBuffer.getInt(offset);
+ offset += Integer.BYTES;
+ byte[] stringBytes = new byte[stringSize];
+ for (int j = 0; j < stringSize; j++) {
+ stringBytes[j] = dataBuffer.getByte(offset);
Review comment:
Use `copyTo()` to read multiple bytes. Same for BYTES type and MV fields
--
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:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
